aios/tests/test_governed_provider_auth_adapter.py

from __future__ import annotations

import os
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

import pytest

from governed_provider_auth_adapter import (
    ALL_RELEASE_EVIDENCE_KINDS,
    authenticate_build_artifacts,
    gate_w_authentication_status,
)
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)


@dataclass(frozen=True)
class FakeBuild:
    source_commit: str
    tree_oid: str
    tests: dict
    canary: dict
    gitea: dict
    rollback: dict
    secret_scan: dict


def _policy(kind: str, root: Path) -> TrustedProviderPolicy:
    return TrustedProviderPolicy(
        evidence_kind=kind,
        provider_identity=f"aios.{kind}-provider",
        store_id=kind,
        store_root=root / kind,
        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,
    )


def _write_receipt(root: Path, kind: str, result: dict) -> str:
    store = root / kind
    store.mkdir(parents=True, exist_ok=True, mode=0o700)
    os.chmod(store, 0o700)
    ref = f"run-1/{kind}.json"
    path = store / ref
    path.parent.mkdir(parents=True, exist_ok=True)
    os.chmod(path.parent, 0o700)
    payload = {
        "schema": PROVIDER_RECEIPT_SCHEMA,
        "version": PROVIDER_RECEIPT_VERSION,
        "provider_identity": f"aios.{kind}-provider",
        "provider_operation": f"{kind}.run",
        "operation_run_id": f"{kind}-run-1",
        "result_reference": f"runs/{kind}-run-1/result.json",
        "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},
    }
    path.write_bytes(canonical_json_bytes(payload))
    os.chmod(path, 0o600)
    return ref


def _resolver(root: Path):
    policies = {kind: _policy(kind, root) for kind in ALL_RELEASE_EVIDENCE_KINDS}
    return TrustedProviderReceiptResolver(policies, now_fn=lambda: NOW)


def _request(root: Path) -> FakeBuild:
    refs = {}
    for kind in ALL_RELEASE_EVIDENCE_KINDS:
        result = {"passed": True, "kind": kind}
        if kind == "gitea":
            result = {"consistent": True, "kind": kind}
        elif kind == "rollback":
            result = {"ready": True, "kind": kind}
        refs[kind] = {"receipt_ref": _write_receipt(root, kind, result)}
    return FakeBuild(source_commit=SOURCE, tree_oid=TREE, **refs)


def test_full_release_authentication_replaces_receipt_refs_with_authenticated_results(tmp_path: Path):
    request = _request(tmp_path)
    authenticated = authenticate_build_artifacts(request, resolver=_resolver(tmp_path))
    assert authenticated.tests["passed"] is True
    assert authenticated.gitea["consistent"] is True
    assert authenticated.rollback["ready"] is True
    assert authenticated.tests["provider_receipt"]["provider_identity"] == "aios.tests-provider"
    assert "receipt_ref" not in authenticated.tests


def test_missing_resolver_fails_closed(tmp_path: Path):
    request = _request(tmp_path)
    with pytest.raises(ProviderAuthenticationError, match="PROVIDER_AUTHENTICATION_REQUIRED"):
        authenticate_build_artifacts(request, resolver=None)


def test_gate_w_authenticates_only_currently_real_provider_classes(tmp_path: Path):
    request = _request(tmp_path)
    # Remove canary/rollback stores entirely; Gate W must not pretend final
    # release closure exists, but it can validate tests/Gitea/secret-scan.
    for kind in ("canary", "rollback"):
        store = tmp_path / kind
        for child in sorted(store.rglob("*"), reverse=True):
            if child.is_file():
                child.unlink()
            elif child.is_dir():
                child.rmdir()
        store.rmdir()

    status = gate_w_authentication_status(request, resolver=_resolver(tmp_path))
    assert status == {
        "producer_runtime": "PRODUCER_RUNTIME_REGISTRATION_PASS",
        "provider_auth": "PROVIDER_AUTH_ENFORCEMENT_PASS",
        "final_release_closure": "FINAL_RELEASE_CLOSURE_PENDING_GATE_R",
    }


def test_full_release_authentication_requires_canary_and_rollback_receipts(tmp_path: Path):
    request = _request(tmp_path)
    (tmp_path / "canary" / "run-1" / "canary.json").unlink()
    with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_UNAVAILABLE"):
        authenticate_build_artifacts(request, resolver=_resolver(tmp_path))