Files
aios/active-runtime/app/governed_provider_auth_adapter.py
T

88 lines
2.9 KiB
Python

from __future__ import annotations
from dataclasses import is_dataclass, replace
from typing import Any, Iterable, Mapping
from trusted_provider_receipts import (
ProviderAuthenticationError,
ResolvedEvidence,
TrustedProviderReceiptResolver,
)
ALL_RELEASE_EVIDENCE_KINDS = ("tests", "canary", "gitea", "rollback", "secret_scan")
GATE_W_EVIDENCE_KINDS = ("tests", "gitea", "secret_scan")
def authenticate_evidence_request(
evidence_kind: str,
evidence_request: Mapping[str, Any],
*,
source_oid: str,
tree_oid: str,
resolver: TrustedProviderReceiptResolver | None,
) -> dict[str, Any]:
if resolver is None:
raise ProviderAuthenticationError("PROVIDER_AUTHENTICATION_REQUIRED")
resolved: ResolvedEvidence = resolver.resolve_evidence(
evidence_kind,
evidence_request,
source_oid=source_oid,
tree_oid=tree_oid,
)
return {
**dict(resolved.result),
"provider_receipt": dict(resolved.provider_receipt),
}
def authenticate_build_artifacts(
request: Any,
*,
resolver: TrustedProviderReceiptResolver | None,
required_kinds: Iterable[str] = ALL_RELEASE_EVIDENCE_KINDS,
) -> Any:
"""Authenticate receipt_ref-only evidence and return a replaced request.
The input is expected to be the existing GovernedBuildArtifacts dataclass.
The function deliberately relies only on field names already used by the
producer (`source_commit`, `tree_oid`, and the evidence kind fields), so it
can be integrated without changing the deployment/release-plan contract.
"""
if resolver is None:
raise ProviderAuthenticationError("PROVIDER_AUTHENTICATION_REQUIRED")
if not is_dataclass(request):
raise TypeError("request must be a dataclass instance")
source_oid = getattr(request, "source_commit")
tree_oid = getattr(request, "tree_oid")
replacements: dict[str, Any] = {}
for kind in tuple(required_kinds):
evidence_request = getattr(request, kind, None)
if evidence_request is None:
raise ProviderAuthenticationError(f"EVIDENCE_{kind.upper()}_REQUIRED")
replacements[kind] = authenticate_evidence_request(
kind,
evidence_request,
source_oid=source_oid,
tree_oid=tree_oid,
resolver=resolver,
)
return replace(request, **replacements)
def gate_w_authentication_status(
request: Any,
*,
resolver: TrustedProviderReceiptResolver | None,
) -> dict[str, str]:
"""Authenticate Gate-W evidence without pretending final closure exists."""
authenticate_build_artifacts(
request,
resolver=resolver,
required_kinds=GATE_W_EVIDENCE_KINDS,
)
return {
"producer_runtime": "PRODUCER_RUNTIME_REGISTRATION_PASS",
"provider_auth": "PROVIDER_AUTH_ENFORCEMENT_PASS",
"final_release_closure": "FINAL_RELEASE_CLOSURE_PENDING_GATE_R",
}