feat(security): verify public tls asset health

This commit is contained in:
ogt
2026-07-14 15:31:15 +08:00
parent 248cf07ce1
commit 27f9e46f3b
7 changed files with 815 additions and 12 deletions

View File

@@ -596,7 +596,9 @@ def test_domain_tls_inventory_projects_fingerprints_without_raw_paths() -> None:
assert len(assets) == 3
assert {asset["asset_type"] for asset in assets} == {"website", "certificate"}
certificate = next(asset for asset in assets if asset["asset_type"] == "certificate")
certificate = next(
asset for asset in assets if asset["asset_type"] == "certificate"
)
assert certificate["metadata"]["raw_certificate_path_stored"] is False
assert certificate["metadata"]["certificate_material_read"] is False
assert certificate["metadata"]["private_key_material_read"] is False
@@ -607,6 +609,170 @@ def test_domain_tls_inventory_projects_fingerprints_without_raw_paths() -> None:
assert "privkey.pem" not in public_text
def test_public_tls_expiry_classification_is_conservative() -> None:
now_epoch = 1_700_000_000.0
assert asset_scanner_job._classify_public_tls_expiry(
now_epoch + (45 * 86400), now_epoch=now_epoch
) == ("healthy", 45, "tls_certificate_valid")
assert asset_scanner_job._classify_public_tls_expiry(
now_epoch + (20 * 86400), now_epoch=now_epoch
) == ("warning", 20, "tls_certificate_expiry_warning")
assert asset_scanner_job._classify_public_tls_expiry(
now_epoch + (3 * 86400), now_epoch=now_epoch
) == ("critical", 3, "tls_certificate_expiry_critical")
assert asset_scanner_job._classify_public_tls_expiry(
now_epoch - 1, now_epoch=now_epoch
) == ("critical", -1, "tls_certificate_expired")
def test_public_tls_address_policy_allows_only_global_addresses() -> None:
assert asset_scanner_job._validated_public_tls_addresses(
["1.1.1.1", "2606:4700:4700::1111", "1.1.1.1"]
) == ["1.1.1.1", "2606:4700:4700::1111"]
for rejected in (
"127.0.0.1",
"10.0.0.1",
"169.254.169.254",
"192.168.0.110",
"::1",
"fc00::1",
"not-an-address",
):
try:
asset_scanner_job._validated_public_tls_addresses([rejected])
except asset_scanner_job._PublicTlsAddressPolicyError:
pass
else:
raise AssertionError(f"non-public TLS address accepted: {rejected}")
def test_domain_tls_probe_aggregates_shared_certificate_without_false_green() -> (
None
):
payload = {
"schema_version": "domain_tls_certbot_inventory_v1",
"generated_at": "2026-07-14T00:00:00+08:00",
"mode": "repo_only_from_nginx_source_of_truth",
"execution_boundaries": {
"secret_value_collected": False,
"live_tls_probe_executed": False,
"certbot_renew_executed": False,
"nginx_reload_executed": False,
"host_write_executed": False,
},
"summary": {
"managed_domain_count": 2,
"unique_certificate_path_count": 1,
},
"managed_domains": [
{
"domain": domain,
"certificate_paths": ["/redacted/shared/fullchain.pem"],
}
for domain in ("app.example.test", "api.example.test")
],
}
assets, relationships = asset_scanner_job._build_domain_tls_inventory_assets(
payload
)
probe_results = {
"app.example.test": asset_scanner_job._public_tls_probe_result(
status="healthy",
evidenced=True,
healthy=True,
chain_verified=True,
hostname_verified=True,
reason_code="tls_certificate_valid",
expires_at="2026-09-01T00:00:00+00:00",
days_remaining=49,
),
"api.example.test": asset_scanner_job._public_tls_probe_result(
status="warning",
evidenced=True,
healthy=False,
chain_verified=True,
hostname_verified=True,
reason_code="tls_certificate_expiry_warning",
expires_at="2026-08-01T00:00:00+00:00",
days_remaining=18,
),
}
assets, _ = asset_scanner_job._apply_domain_tls_probe_results(
assets, relationships, probe_results
)
certificate = next(
asset for asset in assets if asset["asset_type"] == "certificate"
)
metadata = certificate["metadata"]
assert metadata["live_tls_probe_evidenced"] is True
assert metadata["live_tls_healthy"] is False
assert metadata["live_tls_probe_status"] == "warning"
assert metadata["tls_endpoint_count"] == 2
assert metadata["tls_evidenced_endpoint_count"] == 2
assert metadata["tls_healthy_endpoint_count"] == 1
assert metadata["tls_warning_endpoint_count"] == 1
assert metadata["tls_unavailable_endpoint_count"] == 0
assert metadata["tls_min_days_remaining"] == 18
assert metadata["certificate_material_read"] is False
assert metadata["private_key_material_read"] is False
def test_domain_tls_probe_keeps_shared_certificate_unverified_when_one_is_missing() -> (
None
):
payload = {
"schema_version": "domain_tls_certbot_inventory_v1",
"generated_at": "2026-07-14T00:00:00+08:00",
"mode": "repo_only_from_nginx_source_of_truth",
"execution_boundaries": {
"secret_value_collected": False,
"live_tls_probe_executed": False,
"certbot_renew_executed": False,
"nginx_reload_executed": False,
"host_write_executed": False,
},
"summary": {
"managed_domain_count": 2,
"unique_certificate_path_count": 1,
},
"managed_domains": [
{
"domain": domain,
"certificate_paths": ["/redacted/shared/fullchain.pem"],
}
for domain in ("app.example.test", "api.example.test")
],
}
assets, relationships = asset_scanner_job._build_domain_tls_inventory_assets(
payload
)
assets, _ = asset_scanner_job._apply_domain_tls_probe_results(
assets,
relationships,
{
"app.example.test": asset_scanner_job._public_tls_probe_result(
status="healthy",
evidenced=True,
healthy=True,
chain_verified=True,
hostname_verified=True,
reason_code="tls_certificate_valid",
days_remaining=49,
)
},
)
certificate = next(asset for asset in assets if asset["asset_type"] == "certificate")
assert certificate["metadata"]["live_tls_probe_evidenced"] is False
assert certificate["metadata"]["live_tls_healthy"] is False
assert certificate["metadata"]["tls_evidenced_endpoint_count"] == 1
assert certificate["metadata"]["tls_unavailable_endpoint_count"] == 1
def test_domain_tls_inventory_fails_closed_when_secret_boundary_is_not_false() -> (
None
):

View File

@@ -330,7 +330,10 @@ def test_certificate_assets_remain_open_until_live_tls_probe_is_evidenced() -> N
assert payload["certificate_inventory"] == {
"certificate_asset_count": 10,
"live_tls_verified_count": 0,
"live_tls_evidenced_count": 0,
"live_tls_healthy_count": 0,
"live_tls_unverified_count": 10,
"live_tls_unhealthy_count": 0,
"live_tls_health_proven": False,
"raw_certificate_path_returned": False,
"certificate_material_read": False,
@@ -351,6 +354,95 @@ def test_certificate_assets_remain_open_until_live_tls_probe_is_evidenced() -> N
assert domains["network_dns_tls"]["status"] != "controlled"
def test_unhealthy_live_tls_evidence_remains_a_p0_gap() -> None:
now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
payload = build_iwooos_security_asset_control_plane(
discovery_row=_row(
latest_status="success",
latest_scan_depth="full",
latest_success_ended_at=now - timedelta(minutes=1),
latest_total_assets=3,
),
inventory_rows=[
_row(
asset_type="certificate",
cnt=3,
unowned_count=3,
stale_count=0,
unverified_certificate_count=0,
unhealthy_certificate_count=1,
),
],
coverage_rows=[],
relationship_row=_row(relationship_count=0, orphan_asset_count=3),
compliance_rows=[],
automation_rows=[],
ai_trace_row=_row(trace_count=0, accepted_trace_count=0),
siem_row=_row(),
audit_row=_row(),
generated_at=now,
)
assert payload["summary"]["certificate_live_unverified_count"] == 0
assert payload["summary"]["certificate_live_unhealthy_count"] == 1
assert payload["certificate_inventory"]["live_tls_evidenced_count"] == 3
assert payload["certificate_inventory"]["live_tls_healthy_count"] == 2
assert payload["certificate_inventory"]["live_tls_unhealthy_count"] == 1
assert payload["certificate_inventory"]["live_tls_health_proven"] is False
work_item = next(
item
for item in payload["work_items"]
if item["work_item_id"] == "AIA-P0-006-02B"
)
assert work_item["gap_count"] == 1
domains = {
domain["domain_id"]: domain for domain in payload["security_program_domains"]
}
assert domains["network_dns_tls"]["gap_code"] == (
"certificate_live_tls_unhealthy"
)
def test_fully_evidenced_healthy_tls_inventory_proves_health() -> None:
now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
payload = build_iwooos_security_asset_control_plane(
discovery_row=_row(
latest_status="success",
latest_scan_depth="full",
latest_success_ended_at=now - timedelta(minutes=1),
latest_total_assets=10,
),
inventory_rows=[
_row(
asset_type="certificate",
cnt=10,
unowned_count=10,
stale_count=0,
unverified_certificate_count=0,
unhealthy_certificate_count=0,
),
],
coverage_rows=[],
relationship_row=_row(relationship_count=10, orphan_asset_count=0),
compliance_rows=[],
automation_rows=[],
ai_trace_row=_row(trace_count=0, accepted_trace_count=0),
siem_row=_row(),
audit_row=_row(),
generated_at=now,
)
assert payload["certificate_inventory"]["live_tls_evidenced_count"] == 10
assert payload["certificate_inventory"]["live_tls_healthy_count"] == 10
assert payload["certificate_inventory"]["live_tls_unverified_count"] == 0
assert payload["certificate_inventory"]["live_tls_unhealthy_count"] == 0
assert payload["certificate_inventory"]["live_tls_health_proven"] is True
assert all(
item["work_item_id"] != "AIA-P0-006-02B"
for item in payload["work_items"]
)
def test_partial_source_failure_is_visible_without_erasing_core_inventory() -> None:
source_health = [
{"source_id": "asset_discovery", "status": "ready", "reason_code": None},
@@ -547,6 +639,7 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
assert "forbidden_source_domain_detected" in inventory_query
assert "digest_pinned" in inventory_query
assert "live_tls_probe_evidenced" in inventory_query
assert "live_tls_healthy" in inventory_query
def test_public_api_returns_live_aggregate(monkeypatch) -> None: