wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 5 Packages Projects Releases Wiki Activity

feat(security): inventory public gateway certificates

Browse Source
This commit is contained in:
ogt
2026-07-14 14:58:12 +08:00
parent 1cc0737529
commit 248cf07ce1
4 changed files with 427 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
apps/api/tests/test_asset_scanner_job.py
View File

@@ -380,6 +380,9 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them(
async def fake_gitea_ci():
return [], []
async def fake_domain_tls():
return [], []
monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl)
monkeypatch.setattr(
asset_scanner_job,
@@ -406,6 +409,11 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them(
"_collect_gitea_ci_assets",
fake_gitea_ci,
)
monkeypatch.setattr(
asset_scanner_job,
"_collect_domain_tls_inventory_assets",
fake_domain_tls,
)
assets, relationships, prefixes, errors = asyncio.run(
asset_scanner_job._collect_runtime_assets()
@@ -528,6 +536,103 @@ def test_ingress_route_identity_is_stable_when_manifest_order_changes() -> None:
assert len(set(forward_keys.values())) == 2
def test_domain_tls_inventory_projects_fingerprints_without_raw_paths() -> 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": "app.example.test",
"control_tier": "C0",
"tls_certificate_path_present": True,
"certificate_owner_confirmation_required": False,
"acme_challenge_present": True,
"live_tls_probe_status": "not_executed",
"dns_resolution_status": "not_executed",
"certbot_renewal_status": "not_executed",
"owner_review_status": "repo_only_ready_for_owner_review",
"certificate_paths": [
"/etc/letsencrypt/live/example/fullchain.pem",
],
"certificate_key_paths": [
"/etc/letsencrypt/live/example/privkey.pem",
],
},
{
"domain": "api.example.test",
"control_tier": "C0",
"tls_certificate_path_present": True,
"certificate_owner_confirmation_required": True,
"acme_challenge_present": True,
"live_tls_probe_status": "not_executed",
"dns_resolution_status": "not_executed",
"certbot_renewal_status": "not_executed",
"owner_review_status": "repo_only_owner_confirmation_required",
"certificate_paths": [
"/etc/letsencrypt/live/example/fullchain.pem",
],
"certificate_key_paths": [
"/etc/letsencrypt/live/example/privkey.pem",
],
},
],
}
assets, relationships = asset_scanner_job._build_domain_tls_inventory_assets(
payload
)
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")
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
assert len(relationships) == 2
public_text = json.dumps({"assets": assets, "relationships": relationships})
assert "/etc/letsencrypt" not in public_text
assert "privkey.pem" not in public_text
def test_domain_tls_inventory_fails_closed_when_secret_boundary_is_not_false() -> (
None
):
payload = {
"schema_version": "domain_tls_certbot_inventory_v1",
"mode": "repo_only_from_nginx_source_of_truth",
"execution_boundaries": {
"secret_value_collected": True,
"live_tls_probe_executed": False,
"certbot_renew_executed": False,
"nginx_reload_executed": False,
"host_write_executed": False,
},
"managed_domains": [{"domain": "app.example.test"}],
}
try:
asset_scanner_job._build_domain_tls_inventory_assets(payload)
except RuntimeError as exc:
assert str(exc) == (
"domain_tls_inventory_boundary_invalid:secret_value_collected"
)
else:
raise AssertionError("domain TLS inventory must fail closed on secret boundary")
def test_k8s_metadata_assets_never_read_volume_secret_or_command_data() -> None:
pvc = asset_scanner_job._build_pvc_asset(
{
@@ -980,6 +1085,11 @@ services:
"_collect_gitea_ci_assets",
empty_collector,
)
monkeypatch.setattr(
asset_scanner_job,
"_collect_domain_tls_inventory_assets",
empty_collector,
)
assets, relationships, prefixes, errors = asyncio.run(
asset_scanner_job._collect_runtime_assets()

56
apps/api/tests/test_iwooos_security_asset_control_plane.py
View File

@@ -170,6 +170,8 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure
assert payload["supply_chain"]["raw_image_identity_returned"] is False
assert payload["supply_chain"]["registry_api_called"] is False
assert payload["supply_chain"]["github_api_used"] is False
assert payload["certificate_inventory"]["live_tls_health_proven"] is False
assert payload["certificate_inventory"]["private_key_material_read"] is False
assert any(
item["work_item_id"] == "AIA-P0-006-02" for item in payload["work_items"]
)
@@ -296,6 +298,59 @@ def test_unknown_coverage_and_compliance_create_p0_work_items() -> None:
} <= work_item_ids
def test_certificate_assets_remain_open_until_live_tls_probe_is_evidenced() -> 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=10,
),
],
coverage_rows=[],
relationship_row=_row(relationship_count=0, orphan_asset_count=10),
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"] == 10
assert payload["certificate_inventory"] == {
"certificate_asset_count": 10,
"live_tls_verified_count": 0,
"live_tls_unverified_count": 10,
"live_tls_health_proven": False,
"raw_certificate_path_returned": False,
"certificate_material_read": False,
"private_key_material_read": 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"] == 10
domains = {
domain["domain_id"]: domain for domain in payload["security_program_domains"]
}
assert domains["network_dns_tls"]["gap_code"] == (
"certificate_live_tls_probe_missing"
)
assert domains["network_dns_tls"]["status"] != "controlled"
def test_partial_source_failure_is_visible_without_erasing_core_inventory() -> None:
source_health = [
{"source_id": "asset_discovery", "status": "ready", "reason_code": None},
@@ -491,6 +546,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
def test_public_api_returns_live_aggregate(monkeypatch) -> None: