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

feat(security): inventory runtime supply chain

Browse Source
This commit is contained in:
ogt
2026-07-14 14:34:38 +08:00
parent d4bc3cb453
commit ca8771a210
4 changed files with 560 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

219
apps/api/src/jobs/asset_scanner_job.py
View File

@@ -211,6 +211,7 @@ async def scan_once(
"ai_catalog",
"model_registry",
"gitea_bundle_backup",
"gitea_ci",
),
scan_depth: str = "shallow",
) -> str:
@@ -516,6 +517,108 @@ def _build_workload_component_assets(
return assets, relationships
def _build_workload_package_assets(
item: dict[str, Any],
kind: str,
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
"""Inventory deployed OCI image references without pulling image content."""
meta = item.get("metadata", {}) or {}
labels = meta.get("labels", {}) or {}
namespace = meta.get("namespace") or "default"
workload_name = meta.get("name") or "unknown"
pod_spec = (
((item.get("spec", {}) or {}).get("template", {}) or {}).get("spec", {})
or {}
)
owner = (
_normalize_owner_team(labels.get("system"))
or _owner_from_namespace(namespace)
or "platform_security"
)
workload_key = f"k8s/{kind.lower()}/{namespace}/{workload_name}"
assets: list[dict[str, Any]] = []
relationships: list[dict[str, str]] = []
for container_kind, containers in (
("container", pod_spec.get("containers", []) or []),
("init_container", pod_spec.get("initContainers", []) or []),
):
for container in containers:
if not isinstance(container, dict):
continue
container_name = _normalize_owner_team(container.get("name"))
image_ref = str(container.get("image") or "").strip()
if (
not container_name
or not image_ref
or len(image_ref) > 2048
or any(ord(character) < 32 for character in image_ref)
):
continue
image_lower = image_ref.lower()
forbidden_domain = next(
(
domain
for domain in (
"github.com",
"api.github.com",
"raw.githubusercontent.com",
"codeload.github.com",
"ghcr.io",
)
if image_lower == domain
or image_lower.startswith(f"{domain}/")
),
None,
)
digest = image_ref.rsplit("@", 1)[1] if "@" in image_ref else None
last_segment = image_ref.rsplit("/", 1)[-1].split("@", 1)[0]
tag = last_segment.rsplit(":", 1)[1] if ":" in last_segment else None
package_key = (
f"k8s/package/{kind.lower()}/{namespace}/"
f"{workload_name}/{container_kind}/{container_name}"
)
assets.append(
{
"asset_key": package_key,
"asset_type": "package",
"host": None,
"namespace": namespace,
"name": f"{workload_name}/{container_name}",
"metadata": {
"kind": "DeployedOCIImage",
"workload_asset_key": workload_key,
"container_kind": container_kind,
"container_name": container_name,
"image_ref": image_ref,
"image_tag": tag,
"image_digest": digest,
"digest_pinned": bool(
digest and digest.lower().startswith("sha256:")
),
"forbidden_source_domain_detected": forbidden_domain,
"image_content_pulled_by_scanner": False,
"registry_api_called_by_scanner": False,
"secret_value_read": False,
},
"tags": [
"source:k8s_workload_spec",
"ecosystem:oci",
f"owner:{owner}",
],
}
)
relationships.append(
{
"from_key": workload_key,
"to_key": package_key,
"relationship_type": "depends_on",
}
)
return assets, relationships
def _build_service_asset(item: dict[str, Any]) -> dict[str, Any]:
"""Service → asset_inventory row (asset_type='k8s_resource')."""
meta = item.get("metadata", {}) or {}
@@ -1080,9 +1183,15 @@ async def _collect_runtime_assets() -> (
component_assets, component_relationships = (
_build_workload_component_assets(item, kind)
)
package_assets, package_relationships = (
_build_workload_package_assets(item, kind)
)
assets.extend(component_assets)
assets.extend(package_assets)
relationships.extend(component_relationships)
relationships.extend(package_relationships)
reconciled_prefixes.add(f"k8s/{kind.lower()}/")
reconciled_prefixes.add(f"k8s/package/{kind.lower()}/")
for component_type in (
"frontend",
"backend",
@@ -1254,6 +1363,17 @@ async def _collect_runtime_assets() -> (
logger.warning("collect_gitea_bundle_failed", error=str(e))
collector_errors.append("gitea_bundle_backup")
# 13. Gitea workflow identities joined with live CI receipt-chain status.
try:
ci_assets, ci_relationships = await _collect_gitea_ci_assets()
assets.extend(ci_assets)
relationships.extend(ci_relationships)
if ci_assets:
reconciled_prefixes.add("gitea/workflow/")
except Exception as e:
logger.warning("collect_gitea_ci_failed", error=str(e))
collector_errors.append("gitea_ci")
return (
list({asset["asset_key"]: asset for asset in assets}.values()),
relationships,
@@ -1383,6 +1503,105 @@ async def _collect_gitea_bundle_assets() -> (
return _build_gitea_bundle_assets(payload)
def _build_gitea_ci_assets(
workflow_health: dict[str, Any],
receipt_chain: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
"""Project Gitea workflow identities with live receipt-chain status."""
if receipt_chain.get("source_control_authority") != "gitea":
raise RuntimeError("ci_source_control_authority_not_gitea")
boundaries = receipt_chain.get("operation_boundaries") or {}
if boundaries.get("github_api_allowed") is not False:
raise RuntimeError("ci_github_freeze_not_evidenced")
if workflow_health.get("schema_version") != "gitea_workflow_runner_health_v1":
raise RuntimeError("ci_workflow_health_schema_invalid")
workflow_rows = workflow_health.get("workflow_records") or []
if not isinstance(workflow_rows, list) or not workflow_rows:
raise RuntimeError("ci_workflow_records_missing")
summary = receipt_chain.get("summary") or {}
chain_status = str(receipt_chain.get("status") or "unknown")
chain_ready = receipt_chain.get("ready") is True
assets: list[dict[str, Any]] = []
relationships: list[dict[str, str]] = []
seen: set[str] = set()
for row in workflow_rows:
if not isinstance(row, dict):
continue
workflow_id = _normalize_owner_team(row.get("workflow_id"))
if not workflow_id or workflow_id in seen:
continue
seen.add(workflow_id)
pipeline_key = f"gitea/workflow/{workflow_id}"
assets.append(
{
"asset_key": pipeline_key,
"asset_type": "ci_pipeline",
"host": None,
"namespace": "wooo/awoooi",
"name": workflow_id,
"metadata": {
"kind": "GiteaWorkflow",
"display_name": str(row.get("display_name") or workflow_id),
"file_ref": str(row.get("file_ref") or ""),
"status": str(row.get("status") or "unknown"),
"risk_level": str(row.get("risk_level") or "unknown"),
"triggers": sorted(str(value) for value in row.get("triggers") or []),
"runner_labels": sorted(
str(value) for value in row.get("runner_labels") or []
),
"runner_evidence_status": str(
row.get("runner_evidence_status") or "unknown"
),
"job_count": int(row.get("job_count") or 0),
"notify_bridge_calls": int(row.get("notify_bridge_calls") or 0),
"receipt_chain_status": chain_status,
"receipt_chain_ready": chain_ready,
"workflow_notify_bridge_count": int(
summary.get("workflow_notify_bridge_count") or 0
),
"workflow_source_content_read": False,
"workflow_triggered_by_scanner": False,
"gitea_api_write_performed": False,
"github_api_used": False,
"secret_value_read": False,
},
"tags": [
"source:gitea_workflow_health",
"source:gitea_cicd_receipt_chain",
"owner:awoooi",
],
}
)
relationships.append(
{
"from_key": pipeline_key,
"to_key": "gitea/repo/wooo/awoooi",
"relationship_type": "depends_on",
}
)
if not assets:
raise RuntimeError("ci_workflow_identities_invalid")
return assets, relationships
async def _collect_gitea_ci_assets() -> (
tuple[list[dict[str, Any]], list[dict[str, str]]]
):
from src.services.gitea_cicd_alert_receipt_chain_readback import (
load_latest_gitea_cicd_alert_receipt_chain_readback,
)
from src.services.gitea_workflow_runner_health import (
load_latest_gitea_workflow_runner_health,
)
workflow_health, receipt_chain = await asyncio.gather(
asyncio.to_thread(load_latest_gitea_workflow_runner_health),
load_latest_gitea_cicd_alert_receipt_chain_readback(project_id=_PROJECT_ID),
)
return _build_gitea_ci_assets(workflow_health, receipt_chain)
def _is_valid_ipv4(s: str) -> bool:
"""嚴格 IPv4 判斷: 4 段 + 每段 0-255 整數.

116
apps/api/src/services/iwooos_security_asset_control_plane.py
View File

@@ -364,6 +364,10 @@ def _build_security_program_domains(
ai_stage_percent: int,
ai_evidence_count: int,
audit_evidence_count: int,
package_digest_pinned_percent: int,
github_supply_chain_compliant_percent: int,
forbidden_github_supply_chain_asset_count: int,
package_digest_unpinned_count: int,
) -> list[dict[str, Any]]:
scope_by_id = {row["scope_id"]: row for row in asset_scopes}
function_by_id = {row["function_id"]: row for row in control_functions}
@@ -387,6 +391,13 @@ def _build_security_program_domains(
def average(*values: int) -> int:
return round(sum(values) / len(values)) if values else 0
if forbidden_github_supply_chain_asset_count > 0:
supply_chain_gap_code = "forbidden_github_supply_chain_source_detected"
elif package_digest_unpinned_count > 0:
supply_chain_gap_code = "runtime_image_digest_pin_missing"
else:
supply_chain_gap_code = "sbom_signature_dependency_evidence_missing"
domain_values = (
(
"governance_risk",
@@ -473,9 +484,11 @@ def _build_security_program_domains(
average(
scope_percent("supply_chain"),
compliance_percent("cve_scan"),
package_digest_pinned_percent,
github_supply_chain_compliant_percent,
),
scope_assets("supply_chain") + compliance_total("cve_scan"),
"sbom_signature_dependency_evidence_missing",
supply_chain_gap_code,
),
(
"telemetry_detection",
@@ -654,6 +667,14 @@ def build_iwooos_security_asset_control_plane(
stale_total = sum(
int(_value(row, "stale_count", 0) or 0) for row in inventory_source
)
forbidden_github_supply_chain_asset_count = sum(
int(_value(row, "forbidden_github_source_count", 0) or 0)
for row in inventory_source
)
package_digest_unpinned_count = sum(
int(_value(row, "unpinned_package_count", 0) or 0)
for row in inventory_source
)
present_types = {
asset_type for asset_type in _ASSET_TYPES if by_type.get(asset_type, 0) > 0
}
@@ -849,6 +870,22 @@ def build_iwooos_security_asset_control_plane(
+ audit["mcp_tool_audit_count_24h"]
+ audit["automation_operation_count_24h"]
),
package_digest_pinned_percent=_percent(
max(0, by_type.get("package", 0) - package_digest_unpinned_count),
by_type.get("package", 0),
),
github_supply_chain_compliant_percent=_percent(
max(
0,
by_type.get("package", 0)
- forbidden_github_supply_chain_asset_count,
),
by_type.get("package", 0),
),
forbidden_github_supply_chain_asset_count=(
forbidden_github_supply_chain_asset_count
),
package_digest_unpinned_count=package_digest_unpinned_count,
)
work_items: list[dict[str, Any]] = []
@@ -886,6 +923,24 @@ def build_iwooos_security_asset_control_plane(
"依 scope connector 接入網站、資料、供應鏈、安全工具與 AI 資產。",
)
)
supply_chain_policy_gap_count = (
forbidden_github_supply_chain_asset_count
+ package_digest_unpinned_count
)
if supply_chain_policy_gap_count > 0:
work_items.append(
_work_item(
"AIA-P0-006-02A",
"P0",
"supply_chain_policy",
"封閉 runtime image 供應鏈",
supply_chain_policy_gap_count,
(
"以內部 registry 或 repo-owned immutable artifact 替換禁用來源，"
"補 digest pin、canary rollout 與獨立 verifier；不得連線 GitHub。"
),
)
)
if coverage_total == 0 or coverage_unknown > 0:
work_items.append(
_work_item(
@@ -975,6 +1030,7 @@ def build_iwooos_security_asset_control_plane(
sum(domain["controlled_percent"] for domain in security_program_domains)
/ len(security_program_domains)
)
package_inventory_evidenced = by_type.get("package", 0) > 0
strict_runtime_closure_percent = 100 if same_run_closed_loop_proven else 0
completion_dimensions = [
{
@@ -1034,6 +1090,10 @@ def build_iwooos_security_asset_control_plane(
"stale_asset_count": stale_total,
"orphan_asset_count": orphan_asset_count,
"relationship_count": relationship_count,
"package_digest_unpinned_count": package_digest_unpinned_count,
"forbidden_github_supply_chain_asset_count": (
forbidden_github_supply_chain_asset_count
),
"automation_coverage_green_percent": _percent(
coverage_green, coverage_total
),
@@ -1092,6 +1152,30 @@ def build_iwooos_security_asset_control_plane(
"same_run_closed_loop_count": 0,
},
"security_audit": audit,
"supply_chain": {
"package_asset_count": by_type.get("package", 0),
"package_inventory_evidenced": package_inventory_evidenced,
"package_digest_pinned_count": max(
0,
by_type.get("package", 0) - package_digest_unpinned_count,
),
"package_digest_unpinned_count": package_digest_unpinned_count,
"forbidden_github_supply_chain_asset_count": (
forbidden_github_supply_chain_asset_count
),
"github_freeze_compliant": (
package_inventory_evidenced
and forbidden_github_supply_chain_asset_count == 0
),
"runtime_image_policy_compliant": (
package_inventory_evidenced
and forbidden_github_supply_chain_asset_count == 0
and package_digest_unpinned_count == 0
),
"raw_image_identity_returned": False,
"registry_api_called": False,
"github_api_used": False,
},
"source_health": source_health_rows,
"work_items": work_items,
"boundaries": {
@@ -1131,6 +1215,8 @@ def build_unavailable_iwooos_security_asset_control_plane(
"stale_asset_count": 0,
"orphan_asset_count": 0,
"relationship_count": 0,
"package_digest_unpinned_count": 0,
"forbidden_github_supply_chain_asset_count": 0,
"automation_coverage_green_percent": 0,
"automation_coverage_unknown_count": 0,
"compliance_violation_count": 0,
@@ -1282,6 +1368,18 @@ def build_unavailable_iwooos_security_asset_control_plane(
"accepted_ai_trace_count_24h": 0,
"immutable_external_audit_store_evidenced": False,
},
"supply_chain": {
"package_asset_count": 0,
"package_inventory_evidenced": False,
"package_digest_pinned_count": 0,
"package_digest_unpinned_count": 0,
"forbidden_github_supply_chain_asset_count": 0,
"github_freeze_compliant": False,
"runtime_image_policy_compliant": False,
"raw_image_identity_returned": False,
"registry_api_called": False,
"github_api_used": False,
},
"source_health": [
{
"source_id": source_id,
@@ -1489,7 +1587,21 @@ class IwoooSSecurityAssetControlPlaneService:
) AS unowned_count,
count(*) FILTER (
WHERE last_seen_at < NOW() - INTERVAL '2 hours'
) AS stale_count
) AS stale_count,
count(*) FILTER (
WHERE asset_type = 'package'
AND COALESCE(
metadata->>'forbidden_source_domain_detected',
''
) <> ''
) AS forbidden_github_source_count,
count(*) FILTER (
WHERE asset_type = 'package'
AND COALESCE(
metadata->>'digest_pinned',
'false'
) <> 'true'
) AS unpinned_package_count
FROM asset_inventory
WHERE lifecycle_state = 'active'
GROUP BY asset_type

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

@@ -377,6 +377,9 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them(
async def fake_gitea_bundle():
return [], []
async def fake_gitea_ci():
return [], []
monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl)
monkeypatch.setattr(
asset_scanner_job,
@@ -398,6 +401,11 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them(
"_collect_gitea_bundle_assets",
fake_gitea_bundle,
)
monkeypatch.setattr(
asset_scanner_job,
"_collect_gitea_ci_assets",
fake_gitea_ci,
)
assets, relationships, prefixes, errors = asyncio.run(
asset_scanner_job._collect_runtime_assets()
@@ -654,6 +662,103 @@ services:
assert len(registry_relationships) == len(registry_assets)
def test_workload_package_projection_never_pulls_images_or_uses_github() -> None:
assets, relationships = asset_scanner_job._build_workload_package_assets(
{
"metadata": {
"namespace": "awoooi-prod",
"name": "awoooi-api",
"labels": {"system": "awoooi"},
},
"spec": {
"template": {
"spec": {
"containers": [
{
"name": "api",
"image": "192.168.0.110:5000/library/api:abc123",
}
],
"initContainers": [
{
"name": "blocked-source",
"image": "ghcr.io/example/tool@sha256:abcd",
}
],
}
}
},
},
"Deployment",
)
assert len(assets) == 2
assert all(asset["asset_type"] == "package" for asset in assets)
assert assets[0]["metadata"]["image_content_pulled_by_scanner"] is False
assert assets[0]["metadata"]["registry_api_called_by_scanner"] is False
assert assets[0]["metadata"]["forbidden_source_domain_detected"] is None
assert assets[1]["metadata"]["digest_pinned"] is True
assert assets[1]["metadata"]["forbidden_source_domain_detected"] == "ghcr.io"
assert len(relationships) == 2
def test_gitea_ci_projection_requires_gitea_authority_and_github_freeze() -> None:
assets, relationships = asset_scanner_job._build_gitea_ci_assets(
{
"schema_version": "gitea_workflow_runner_health_v1",
"workflow_records": [
{
"workflow_id": "cd",
"display_name": "CD",
"file_ref": ".gitea/workflows/cd.yaml",
"status": "manifest_mapped",
"risk_level": "high",
"triggers": ["push", "workflow_dispatch"],
"runner_labels": ["awoooi-non110-ubuntu"],
"runner_evidence_status": "non110_runner_mapped",
"job_count": 2,
"notify_bridge_calls": 1,
}
],
},
{
"source_control_authority": "gitea",
"status": "gitea_cicd_alert_receipt_chain_ready",
"ready": True,
"operation_boundaries": {"github_api_allowed": False},
"summary": {"workflow_notify_bridge_count": 6},
},
)
assert [asset["asset_type"] for asset in assets] == ["ci_pipeline"]
assert assets[0]["metadata"]["receipt_chain_ready"] is True
assert assets[0]["metadata"]["workflow_triggered_by_scanner"] is False
assert assets[0]["metadata"]["github_api_used"] is False
assert relationships == [
{
"from_key": "gitea/workflow/cd",
"to_key": "gitea/repo/wooo/awoooi",
"relationship_type": "depends_on",
}
]
try:
asset_scanner_job._build_gitea_ci_assets(
{
"schema_version": "gitea_workflow_runner_health_v1",
"workflow_records": [{"workflow_id": "cd"}],
},
{
"source_control_authority": "gitea",
"operation_boundaries": {"github_api_allowed": True},
},
)
except RuntimeError as exc:
assert str(exc) == "ci_github_freeze_not_evidenced"
else:
raise AssertionError("CI projection must fail closed without GitHub freeze")
def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefixes(
monkeypatch,
) -> None:
@@ -696,7 +801,18 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix
"system": "awoooi",
},
},
"spec": {},
"spec": {
"template": {
"spec": {
"containers": [
{
"name": "web",
"image": "192.168.0.110:5000/library/web:test",
}
]
}
}
},
"status": {},
},
{
@@ -708,7 +824,18 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix
"system": "awoooi",
},
},
"spec": {},
"spec": {
"template": {
"spec": {
"containers": [
{
"name": "api",
"image": "192.168.0.110:5000/library/api:test",
}
]
}
}
},
"status": {},
},
]
@@ -724,7 +851,18 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix
"app.kubernetes.io/component": "log-collector"
},
},
"spec": {},
"spec": {
"template": {
"spec": {
"containers": [
{
"name": "otel-collector",
"image": "otel/collector:0.96.0",
}
]
}
}
},
"status": {},
}
]
@@ -837,6 +975,11 @@ services:
"_collect_gitea_bundle_assets",
empty_collector,
)
monkeypatch.setattr(
asset_scanner_job,
"_collect_gitea_ci_assets",
empty_collector,
)
assets, relationships, prefixes, errors = asyncio.run(
asset_scanner_job._collect_runtime_assets()
@@ -858,6 +1001,7 @@ services:
"dashboard",
"cache",
"message_queue",
"package",
} <= {asset["asset_type"] for asset in assets}
assert {
"k8s/secret-ref/",
@@ -873,6 +1017,8 @@ services:
"service-registry/dashboard/",
"service-registry/cache/",
"service-registry/message_queue/",
"k8s/package/deployment/",
"k8s/package/daemonset/",
} <= set(prefixes)
assert any(
relationship["to_key"] == "k8s/secret-ref/prod/database-ref"

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

@@ -162,6 +162,14 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure
assert domain_by_id["siem_soc_threat_intel"]["controlled_percent"] == 50
assert len(payload["siem"]["pipeline"]) == 8
assert payload["security_audit"]["mcp_tool_audit_count_24h"] == 11
assert payload["summary"]["package_digest_unpinned_count"] == 0
assert payload["summary"]["forbidden_github_supply_chain_asset_count"] == 0
assert payload["supply_chain"]["package_inventory_evidenced"] is False
assert payload["supply_chain"]["github_freeze_compliant"] is False
assert payload["supply_chain"]["runtime_image_policy_compliant"] is False
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 any(
item["work_item_id"] == "AIA-P0-006-02" for item in payload["work_items"]
)
@@ -177,6 +185,71 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure
assert 'secret_value_collection_allowed": true' not in public_text.lower()
def test_runtime_image_policy_gaps_are_visible_and_never_false_green() -> 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_started_at=now - timedelta(minutes=10),
latest_ended_at=now - timedelta(minutes=9),
latest_success_ended_at=now - timedelta(minutes=9),
latest_total_assets=31,
),
inventory_rows=[
_row(
asset_type="package",
cnt=31,
unowned_count=0,
stale_count=0,
forbidden_github_source_count=3,
unpinned_package_count=31,
),
],
coverage_rows=[],
relationship_row=_row(relationship_count=31, 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["summary"]["package_digest_unpinned_count"] == 31
assert payload["summary"]["forbidden_github_supply_chain_asset_count"] == 3
assert payload["supply_chain"] == {
"package_asset_count": 31,
"package_inventory_evidenced": True,
"package_digest_pinned_count": 0,
"package_digest_unpinned_count": 31,
"forbidden_github_supply_chain_asset_count": 3,
"github_freeze_compliant": False,
"runtime_image_policy_compliant": False,
"raw_image_identity_returned": False,
"registry_api_called": False,
"github_api_used": False,
}
work_item = next(
item
for item in payload["work_items"]
if item["work_item_id"] == "AIA-P0-006-02A"
)
assert work_item["gap_count"] == 34
assert work_item["status"] == "open"
domains = {
domain["domain_id"]: domain for domain in payload["security_program_domains"]
}
assert domains["software_supply_chain"]["gap_code"] == (
"forbidden_github_supply_chain_source_detected"
)
assert domains["software_supply_chain"]["status"] != "controlled"
public_text = json.dumps(payload, ensure_ascii=False).lower()
assert "ghcr.io/" not in public_text
assert "raw.githubusercontent.com/" not in public_text
def test_unknown_coverage_and_compliance_create_p0_work_items() -> None:
now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
payload = build_iwooos_security_asset_control_plane(
@@ -413,6 +486,11 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
)
assert "ORDER BY snapshot_id DESC" in compliance_query
assert "LIMIT 100000" in compliance_query
inventory_query = next(
statement for statement in statements if "FROM asset_inventory" in statement
)
assert "forbidden_source_domain_detected" in inventory_query
assert "digest_pinned" in inventory_query
def test_public_api_returns_live_aggregate(monkeypatch) -> None: