"""Issue sanitized, release-bound target evidence from retained result files.""" from __future__ import annotations import base64 from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, Mapping, Sequence from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from .evidence import ( MAX_HASHED_FILE_BYTES, NEGATIVE_RESULTS, OPAQUE_ID_PATTERN, POSITIVE_RESULTS, PROOF_REQUIREMENTS, _active_authority_keys, _bounded_file_sha256, canonical_bytes, canonical_sha256, ) MAX_BOUNDARY_ARTIFACT_BYTES = MAX_HASHED_FILE_BYTES MAX_BOUNDARY_ARTIFACT_TOTAL_BYTES = 512 * 1024 * 1024 def issue_boundary_evidence( *, assessment: Mapping[str, Any], installed_evidence: Mapping[str, Any], proof_id: str, claims: Sequence[Mapping[str, Any]], authority_keyring: dict[str, Any], signing_keys: Mapping[str, Ed25519PrivateKey], expires_at: datetime, issued_at: datetime | None = None, expected_external_provider_subject: str | None = None, forbidden_authority_public_keys: frozenset[bytes] = frozenset(), release_origin_verified: bool = False, ) -> dict[str, Any]: """Build and sign one bounded proof after its release origin was verified. The public CLI is responsible for deriving ``release_origin_verified`` from the signed catalog, installed observation and installer receipt. Artifact paths are consumed here and never copied into the resulting receipt. """ if release_origin_verified is not True: raise ValueError( "boundary evidence requires independently verified installed release origin" ) _require_opaque_id(proof_id, label="proof ID") now = issued_at or datetime.now(tz=UTC) now = _aware_utc(now, label="issuance time") expiry = _aware_utc(expires_at, label="expiry time") if now >= expiry: raise ValueError("boundary evidence expiry must be after issuance") assessment_id = assessment.get("assessment_id") release = assessment.get("release") assessment_release = release.get("ref") if isinstance(release, Mapping) else None deployment = assessment.get("deployment_profile") deployment_subject = ( deployment.get("id") if isinstance(deployment, Mapping) else None ) _require_opaque_id(assessment_id, label="assessment ID") _require_opaque_id(assessment_release, label="assessment release") _require_opaque_id(deployment_subject, label="deployment subject") if expected_external_provider_subject is not None: _require_opaque_id( expected_external_provider_subject, label="external-provider subject", ) normalized_claims = _normalize_claims( claims=claims, deployment_subject=str(deployment_subject), expected_external_provider_subject=expected_external_provider_subject, ) active_at_issue, issue_findings = _active_authority_keys( authority_keyring=authority_keyring, verification_time=now, forbidden_public_keys=forbidden_authority_public_keys, ) active_until_expiry, expiry_findings = _active_authority_keys( authority_keyring=authority_keyring, verification_time=expiry - timedelta(microseconds=1), forbidden_public_keys=forbidden_authority_public_keys, ) if issue_findings or expiry_findings: raise ValueError( "proof-authority keyring is malformed, reused, or not valid for the full proof interval" ) if not signing_keys: raise ValueError( "at least one independently provisioned signing key is required" ) covered_scopes: set[str] = set() verified_signers: list[tuple[str, Ed25519PrivateKey]] = [] for key_id, private_key in sorted(signing_keys.items()): _require_opaque_id(key_id, label="signing key ID") if not isinstance(private_key, Ed25519PrivateKey): raise ValueError("boundary signing keys must be Ed25519") authority = active_at_issue.get(key_id) authority_at_expiry = active_until_expiry.get(key_id) public_bytes = private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) if ( authority is None or authority_at_expiry is None or authority[0] != public_bytes or authority_at_expiry[0] != public_bytes ): raise ValueError( f"signing key {key_id!r} is not independently authorized for the full proof interval" ) covered_scopes.update(authority[1]) verified_signers.append((key_id, private_key)) claim_scopes = {str(claim["scope"]) for claim in normalized_claims} uncovered_scopes = sorted(claim_scopes - covered_scopes) if uncovered_scopes: raise ValueError( "no supplied signing key is authorized for scopes: " + ", ".join(uncovered_scopes) ) proof: dict[str, Any] = { "$schema": "./capability-fit-boundary-evidence.schema.json", "schema_version": "0.1.0", "evidence_kind": "govoplan.capability-fit-boundary-proof", "proof_id": proof_id, "assessment_id": assessment_id, "assessment_release": assessment_release, "installed_evidence_sha256": canonical_sha256(installed_evidence), "issued_at": _iso_datetime(now), "expires_at": _iso_datetime(expiry), "claims": normalized_claims, } signature_payload = canonical_bytes(proof) proof["signatures"] = [ { "algorithm": "ed25519", "key_id": key_id, "value": base64.b64encode(private_key.sign(signature_payload)).decode( "ascii" ), } for key_id, private_key in verified_signers ] return proof def _normalize_claims( *, claims: Sequence[Mapping[str, Any]], deployment_subject: str, expected_external_provider_subject: str | None, ) -> list[dict[str, Any]]: if not 1 <= len(claims) <= len(PROOF_REQUIREMENTS): raise ValueError("boundary evidence requires between one and eight claims") normalized: list[dict[str, Any]] = [] seen_scopes: set[str] = set() total_bytes = 0 for claim in claims: scope = str(claim.get("scope") or "") if scope not in PROOF_REQUIREMENTS or scope in seen_scopes: raise ValueError("boundary claim scopes must be known and unique") seen_scopes.add(scope) result = str(claim.get("result") or "") if result not in {POSITIVE_RESULTS[scope], NEGATIVE_RESULTS[scope]}: raise ValueError(f"boundary claim result is invalid for {scope!r}") controls = _opaque_id_list( claim.get("control_ids"), label=f"{scope} control IDs", maximum=256, ) raw_artifacts = claim.get("artifacts") if not isinstance(raw_artifacts, Sequence) or isinstance( raw_artifacts, (str, bytes) ): raise ValueError(f"{scope} artifacts must be a non-empty list") if not 1 <= len(raw_artifacts) <= 256: raise ValueError(f"{scope} must reference between one and 256 artifacts") artifacts: list[dict[str, str]] = [] seen_artifacts: set[str] = set() for artifact in raw_artifacts: if not isinstance(artifact, Mapping): raise ValueError(f"{scope} contains a malformed artifact") artifact_id = artifact.get("artifact_id") _require_opaque_id(artifact_id, label=f"{scope} artifact ID") artifact_id = str(artifact_id) if artifact_id in seen_artifacts: raise ValueError(f"{scope} contains duplicate artifact IDs") seen_artifacts.add(artifact_id) raw_path = artifact.get("path") if not isinstance(raw_path, (str, Path)) or not str(raw_path): raise ValueError(f"{scope} artifact {artifact_id!r} has no file path") try: digest, size, changed = _bounded_file_sha256( Path(raw_path), max_bytes=MAX_BOUNDARY_ARTIFACT_BYTES ) except OSError as exc: raise ValueError( f"could not securely hash {scope} artifact {artifact_id!r}" ) from exc if changed: raise ValueError( f"{scope} artifact {artifact_id!r} changed or exceeded the size limit while hashing" ) total_bytes += size if total_bytes > MAX_BOUNDARY_ARTIFACT_TOTAL_BYTES: raise ValueError( "boundary evidence artifacts exceed the total size limit" ) artifacts.append({"artifact_id": artifact_id, "sha256": digest.hex()}) normalized.append( { "scope": scope, "result": result, "subject_id": ( expected_external_provider_subject if scope == "external_providers" else deployment_subject ), "control_ids": controls, "artifacts": artifacts, } ) if "external_providers" in seen_scopes and not expected_external_provider_subject: raise ValueError( "external-provider evidence requires its expected opaque subject ID" ) return sorted(normalized, key=lambda item: str(item["scope"])) def _opaque_id_list(value: object, *, label: str, maximum: int) -> list[str]: if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): raise ValueError(f"{label} must be a non-empty list") result = [str(item) for item in value] if not 1 <= len(result) <= maximum or len(result) != len(set(result)): raise ValueError(f"{label} must contain unique bounded IDs") for item in result: _require_opaque_id(item, label=label) return result def _require_opaque_id(value: object, *, label: str) -> None: if ( not isinstance(value, str) or len(value) > 160 or OPAQUE_ID_PATTERN.fullmatch(value) is None ): raise ValueError(f"{label} must be a bounded opaque ID") def _aware_utc(value: datetime, *, label: str) -> datetime: if value.tzinfo is None: raise ValueError(f"{label} must include a timezone") return value.astimezone(UTC) def _iso_datetime(value: datetime) -> str: return value.astimezone(UTC).isoformat().replace("+00:00", "Z")