132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from governed_authenticated_evidence_operation import (
|
|
produce_authenticated_evidence_closure,
|
|
validate_gate_w_provider_auth,
|
|
)
|
|
from trusted_provider_receipts import (
|
|
PROVIDER_RECEIPT_SCHEMA,
|
|
PROVIDER_RECEIPT_VERSION,
|
|
ProviderAuthenticationError,
|
|
TrustedProviderPolicy,
|
|
TrustedProviderReceiptResolver,
|
|
canonical_json_bytes,
|
|
canonical_sha256,
|
|
)
|
|
|
|
SOURCE = "a" * 40
|
|
TREE = "b" * 40
|
|
NOW = datetime(2026, 9, 9, 5, 0, tzinfo=timezone.utc)
|
|
KINDS = ("tests", "canary", "gitea", "rollback", "secret_scan")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Req:
|
|
source_commit: str
|
|
tree_oid: str
|
|
tests: dict
|
|
canary: dict
|
|
gitea: dict
|
|
rollback: dict
|
|
secret_scan: dict
|
|
|
|
|
|
def _result(kind):
|
|
if kind == "gitea":
|
|
return {"consistent": True}
|
|
if kind == "rollback":
|
|
return {"ready": True}
|
|
return {"passed": True}
|
|
|
|
|
|
def _setup(root: Path):
|
|
policies = {}
|
|
refs = {}
|
|
for kind in KINDS:
|
|
store = root / kind
|
|
store.mkdir(mode=0o700)
|
|
os.chmod(store, 0o700)
|
|
result = _result(kind)
|
|
receipt = {
|
|
"schema": PROVIDER_RECEIPT_SCHEMA,
|
|
"version": PROVIDER_RECEIPT_VERSION,
|
|
"provider_identity": f"aios.{kind}",
|
|
"provider_operation": f"{kind}.run",
|
|
"operation_run_id": f"{kind}-1",
|
|
"result_reference": f"{kind}/1/result",
|
|
"content_sha256": canonical_sha256(result),
|
|
"source_oid": SOURCE,
|
|
"tree_oid": TREE,
|
|
"environment": "linux-isolated",
|
|
"issued_at": "2026-09-09T04:59:00+00:00",
|
|
"result": result,
|
|
"authentication": {"kind": "root-owned-store-v1", "store_id": kind},
|
|
}
|
|
ref = "receipt.json"
|
|
(store / ref).write_bytes(canonical_json_bytes(receipt))
|
|
os.chmod(store / ref, 0o600)
|
|
refs[kind] = {"receipt_ref": ref}
|
|
policies[kind] = TrustedProviderPolicy(
|
|
evidence_kind=kind,
|
|
provider_identity=f"aios.{kind}",
|
|
store_id=kind,
|
|
store_root=store,
|
|
allowed_operations=frozenset({f"{kind}.run"}),
|
|
expected_owner_uid=os.getuid(),
|
|
allowed_directory_modes=frozenset({0o700}),
|
|
allowed_receipt_modes=frozenset({0o600}),
|
|
max_age_seconds=300,
|
|
)
|
|
req = Req(SOURCE, TREE, **refs)
|
|
resolver = TrustedProviderReceiptResolver(policies, now_fn=lambda: NOW)
|
|
return req, resolver
|
|
|
|
|
|
def test_wrapper_requires_resolver(tmp_path):
|
|
req, _ = _setup(tmp_path)
|
|
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_AUTHENTICATION_REQUIRED"):
|
|
produce_authenticated_evidence_closure(
|
|
req, provider_receipt_resolver=None,
|
|
source_root=tmp_path/"src", builder_policy=tmp_path/"policy.json",
|
|
legacy_producer=lambda *a, **k: None,
|
|
)
|
|
|
|
|
|
def test_wrapper_authenticates_then_calls_legacy(tmp_path):
|
|
req, resolver = _setup(tmp_path)
|
|
seen = {}
|
|
|
|
def legacy(authenticated, *, source_root, builder_policy):
|
|
seen["request"] = authenticated
|
|
return {"status": "ok", "source_root": str(source_root), "policy": str(builder_policy)}
|
|
|
|
out = produce_authenticated_evidence_closure(
|
|
req,
|
|
provider_receipt_resolver=resolver,
|
|
source_root=tmp_path/"src",
|
|
builder_policy=tmp_path/"policy.json",
|
|
legacy_producer=legacy,
|
|
)
|
|
assert out["status"] == "ok"
|
|
assert seen["request"].tests["passed"] is True
|
|
assert "provider_receipt" in seen["request"].tests
|
|
assert "receipt_ref" not in seen["request"].tests
|
|
|
|
|
|
def test_gate_w_status_does_not_claim_final_closure(tmp_path):
|
|
req, resolver = _setup(tmp_path)
|
|
# Gate-W is intentionally independent of real canary/rollback receipts.
|
|
(tmp_path/"canary"/"receipt.json").unlink()
|
|
(tmp_path/"rollback"/"receipt.json").unlink()
|
|
status = validate_gate_w_provider_auth(req, provider_receipt_resolver=resolver)
|
|
assert status["producer_runtime"] == "PRODUCER_RUNTIME_REGISTRATION_PASS"
|
|
assert status["provider_auth"] == "PROVIDER_AUTH_ENFORCEMENT_PASS"
|
|
assert status["final_release_closure"] == "FINAL_RELEASE_CLOSURE_PENDING_GATE_R"
|