Compare commits

..
Author SHA1 Message Date
AIOS Governed Continuation 3c320aeb89 feat(governed): absorb control-plane integrity hardening into authority source line
Turn historical AIOS failure modes (dual-plane divergence, fake execution/
PASS-without-evidence, queued-without-claimant, exact-scope mismatch,
hash-only/placeholder artifacts, OpenAPI/backend asymmetry, capability
singular/plural asymmetry, terminal success without commit/readback) into
executable read-only invariants absorbed on the authoritative source line
(beabd05 = V2R2C2 authority main). This is a successor line; it does NOT
revive the old p0-chatgpt-bridge-v2 control-plane-convergence task line.

Adds:
- aios_control_plane_consistency: read-only derived SystemTruth (derive_truth)
  + artifact reality gate + OpenAPI/backend parity + capability alias checks.
- aios_control_plane_snapshot.attach_control_plane_truth: injects derived truth
  (state + findings + evidence_flow_confirmed) into the existing explain/status
  surface without mutating any persistence plane; base snapshot contract intact.
- config/gitops/system_truth_contract.json: state-ownership + completion/queue
  requirements + forbidden equivalences.
- tests: 13 original hardening tests + 5 snapshot-truth integration tests
  (18 passed locally).
- ADR + remediation work package docs; fault-injection + audit scripts.

Derivation is read-only; ADMITTED/QUEUED/service-active/drift_status=PASS are
not treated as evidence-complete. Evidence-required transitions and the final
E2E chain (through external projection readback / SCENARIO_ACCEPTED) remain the
release gate and are NOT claimed by this commit.

Signed-off-by: AIOS Governed Continuation <aios-governed-continuation@localhost>
2026-09-13 10:31:40 +08:00
owner beabd05396 test(governed): align reopen fake CAS with expected_revision contract (V2R2C2 hot-fix)
autonomy-smoke-tests pr82 PASS evidence_sha256=c16c68f86ee0c4e8cfd0af3f76078be45441ee151395fbde5800145c08e189b5
autonomy-unit-tests pr82 PASS evidence_sha256=168bb88573b44876966c64ec2dc2b9e030fc7635ca0b18538678f1c492c97ea2
campaign-approval-gate pr82 PASS evidence_sha256=0616d6aa11496391686dcbd658123e51d8d057e1f02661af7ef8512554e404d8
git-diff-check pr82 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
py-compile pr82 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
reconciliation-contract pr82 PASS evidence_sha256=8250e3e736db14432c421c46f995c201540e2b9c50958bdeb3ef7b057c4fd577
release-state-machine pr82 PASS evidence_sha256=b6c51b401d65a9c69435f8b86cecab752d65ded7188228aab3de3eb010990b60
secret-scan pr82 PASS evidence_sha256=f8ed9052d5ba0fbe8e52f171ea180fa353a89b4049cd3e1ed881736d0622c752
signer-validator-contract pr82 PASS evidence_sha256=775b7aa27d0c17f48b3feff20c836f14b6ed8ebcb90ec7e0b76d2a13c579268a
aios.owner-final-merge-authorization-v2 real v2 contract tests passed
2026-09-12 15:29:18 +08:00
owner 5a4e952b39 fix(governed): pin saga CAS to current revision in approval reopen (V2R2C2)
autonomy-smoke-tests pr82 PASS evidence_sha256=c16c68f86ee0c4e8cfd0af3f76078be45441ee151395fbde5800145c08e189b5
autonomy-unit-tests pr82 FAIL evidence_sha256=44994ff4cfe08df298fe59f790f3f98df7aa29e4d239ecd185b3143818c540cd
campaign-approval-gate pr82 PASS evidence_sha256=0616d6aa11496391686dcbd658123e51d8d057e1f02661af7ef8512554e404d8
git-diff-check pr82 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
py-compile pr82 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
reconciliation-contract pr82 PASS evidence_sha256=a0d5c1849c776f73fa64b02300e10c193241c0f9460e6ad4e69d73ae720de90b
release-state-machine pr82 PASS evidence_sha256=1d3314562a9345a9f75b1adf32d61168822c00d316f7db85833230bba023e448
secret-scan pr82 PASS evidence_sha256=9ca1f810bf76589206704716c4f3701a6b6aca1066b1d0c24b91d5be008f52a2
signer-validator-contract pr82 PASS evidence_sha256=775b7aa27d0c17f48b3feff20c836f14b6ed8ebcb90ec7e0b76d2a13c579268a
2026-09-12 15:16:32 +08:00
owner b70c89217b feat(governed): add /reopenOwnerApproval to silent-release allow-list (V2R2C2 v1.2)
autonomy-smoke-tests pr81 PASS evidence_sha256=e9b29f42e72804aedf5a269d4507aafa7270b39400c6923bbb014d8a335fe788
autonomy-unit-tests pr81 PASS evidence_sha256=068610c15e2716ad9dbda4df158f891693e53cb7da2c57aef3783e722fb6f354
campaign-approval-gate pr81 PASS evidence_sha256=0616d6aa11496391686dcbd658123e51d8d057e1f02661af7ef8512554e404d8
git-diff-check pr81 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
py-compile pr81 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
reconciliation-contract pr81 PASS evidence_sha256=f96a26b816cb58aec4639ca90e87da07ed8cf7d041d82fb67d755bbd5085dfca
release-state-machine pr81 PASS evidence_sha256=6cea88752b3e58ae5cee56ff3b1b13eab34117918d6bad9661194c031e2dc157
secret-scan pr81 PASS evidence_sha256=e85624b4ea35c4f7019bc9d9a3d7d44124afca891c3487fe9f550faf6c743245
signer-validator-contract pr81 PASS evidence_sha256=775b7aa27d0c17f48b3feff20c836f14b6ed8ebcb90ec7e0b76d2a13c579268a
aios.owner-final-merge-authorization-v2 real v2 contract tests passed
Narrow v1.2 delta: add the Owner approval-reopen POST path to
SILENT_RELEASE_POST_PATHS so reopen can proceed in Shadow without enabling
AIOS_GITOPS_WRITES_ENABLED. Route handler, release:approve guard, global
write check and all other allow-list paths are unchanged.

Adds tests/test_reopen_shadow_gate_contract.py covering:
- reopen is a narrow silent-release POST (route requires release:approve)
- global write gate is not enabled by this source patch
- all existing silent-release POST paths remain present
2026-09-12 14:29:33 +08:00
AIOS Governed Continuation dd2bd8fa08 feat(governed): add Owner approval reopen control-plane capability (V2R2C2)
autonomy-smoke-tests pr80 PASS evidence_sha256=c16c68f86ee0c4e8cfd0af3f76078be45441ee151395fbde5800145c08e189b5
autonomy-unit-tests pr80 PASS evidence_sha256=02e00edbe6415cf1d548ef7fb53b67e37c4abd29397b00075c6feb0d48b5f2c5
campaign-approval-gate pr80 PASS evidence_sha256=0616d6aa11496391686dcbd658123e51d8d057e1f02661af7ef8512554e404d8
git-diff-check pr80 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
py-compile pr80 PASS evidence_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
reconciliation-contract pr80 PASS evidence_sha256=a98ae8e4b231cd5a106781a85b051ad0543e0481d604919bab8fc1047de342eb
release-state-machine pr80 PASS evidence_sha256=b6c51b401d65a9c69435f8b86cecab752d65ded7188228aab3de3eb010990b60
secret-scan pr80 PASS evidence_sha256=fb45191fc28616fa47b5c8ac1f5d6541f3e8d3163eb6653c9678ab84dff44e61
signer-validator-contract pr80 PASS evidence_sha256=775b7aa27d0c17f48b3feff20c836f14b6ed8ebcb90ec7e0b76d2a13c579268a
aios.owner-final-merge-authorization-v2 real v2 contract tests passed
Add a fail-closed control-plane operation POST /reopenOwnerApproval (scope
release:approve) that reopens one expired Owner approval cycle only after
signed_owner_approval_accepted is the final Saga event and every release byte
(candidate/source/tree/artifact/manifest/release-plan/policy) matches exactly.

- New module release_approval_reopen.py: archives the old approval byte-for-byte
  under approval-history/<candidate>/<fingerprint>.json, then drives
  ReleaseSagaStore.compare_and_set(APPROVED -> READY_FOR_APPROVAL) and removes
  only the old active approval slot (post-CAS). Idempotent per idempotency key.
- ReleaseCandidate v1 durable records do not persist side_effects_executed; this
  accepts an absent marker only when the accepted-approval event is the final
  Saga event, and rejects any explicit non-false value.
- Does not modify ReleaseSagaStore, never calls force_state_for_recovery_test,
  never performs direct DB or legacy-global writes. Publishes control-plane
  capability only; keeps 4642b74 on the mainline ancestry.
2026-09-12 12:13:33 +08:00
AIOS Governed Continuation 4642b74af6 fix(admission): make V2R2C2 concurrent idempotency contract gate order-independent
autonomy-smoke-tests V2R2C2 PASS evidence_sha256=8ed5285cd9e236a2d674877b183325c3bcb8d7dbf699eed9f516b97ef1bd95ab
autonomy-unit-tests V2R2C2 PASS evidence_sha256=c82cbd9c6151b6e5f7dd87cc4588ab8018adbdedae2ae51705c43212f33ee6d3
campaign-approval-gate V2R2C2 PASS evidence_sha256=fc3707693b9b28a683f6cce43994fa272000cd6d52e31ac2800d329466559c54
git-diff-check V2R2C2 PASS evidence_sha256=6facd39176e65a80ffa3591db4e59c8a0a7dc04a3301dcd7b457d456e380305e
py-compile V2R2C2 PASS evidence_sha256=32381d24d691a64a82acbe2faecec51b8d3f43ac56f1884cc1ff83c2202deb34
reconciliation-contract V2R2C2 PASS evidence_sha256=b2177f09c2ec27acbb593eac7ff0039513eaa2d2202415d414a61c6ef1a0421c
release-state-machine V2R2C2 PASS evidence_sha256=0eed2af4f75ee18b68a013c9f02cb966181decd2eaa8d582f7456b3ac5f5ef7e
secret-scan V2R2C2 PASS evidence_sha256=1c584be64c3d82a4716a39b754458dfea7e3300cdd0b4dc3e9a8c86c263f95d1
signer-validator-contract V2R2C2 PASS evidence_sha256=74b3d59f0fa186c46ba282715c0490d50ea0580b4e63352f57caa0de729535da
aios.owner-final-merge-authorization-v2 V2R2C2 PASS evidence_sha256=70bb81b45354e252072e73861801d783c08062b932dc6f77c3217b86a43f5f76
The duplicate/concurrent mission-authorization test asserted that the single
creating (non-idempotent) request is always ThreadPoolExecutor.map result
index 0. map returns in *input* order, not completion order, so under 16-way
concurrency the winner may land at index 1 or 2 (~15% of runs), causing a
flaky gate failure even though admission behavior is correct (exactly one
creator, one durable task, rest idempotent replays).

Replace the index-based assertion with an order-independent invariant:
exactly 15 replays + exactly 1 creator. Behavior unchanged; stronger and
deterministic.
2026-09-12 06:49:03 +08:00
AIOS Governed Continuation 34395c7100 fix(v2r2c2): gate recovery — normalize LF, EOF blank lines, deterministic test path-mode, secret-scan allowlist for release-freeze OIDs 2026-09-12 02:44:54 +08:00
AIOS Governed Continuation 623fd61042 feat: shard-b governed provider-auth integration (V2R2C2 Round3) 2026-09-10 12:09:47 +08:00
AIOS Governed Continuation 691bf76aec feat: shard-a governed provider-auth integration (V2R2C2 Round3) 2026-09-10 11:54:49 +08:00
51 changed files with 6853 additions and 127 deletions
+10 -3
View File
@@ -4,7 +4,14 @@ title = "AIOS repository secret scan policy"
useDefault = true
[allowlist]
description = "Exact non-secret Task Admission test idempotency fixture in the historical PR commit"
description = "Exact non-secret fixtures: Task Admission idempotency test fixture and release-freeze git commit OIDs (generic-api-key false positive on 40-hex git hashes)"
condition = "AND"
paths = ['''tests/test_task_admission_bootstrap_http\.py$''']
regexes = ['''idempotency-1''']
paths = [
'''tests/test_task_admission_bootstrap_http\.py$''',
'''docs/V2R2C2-P0-convergence-20260909\.md$''',
'''artifacts/round3-provider-auth/ledger\.json$''',
]
regexes = [
'''idempotency-1''',
'''f7f53d0568767d6deec2c47008a89f4d8413f93d''',
]
@@ -0,0 +1,396 @@
#!/usr/bin/env python3
from __future__ import annotations
from dataclasses import dataclass, asdict, field
from enum import Enum
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence
import hashlib
import json
import re
import sqlite3
import subprocess
import time
PLACEHOLDER_PATTERNS = (
re.compile(r"(?im)^\s*(?:todo|fixme|tbd)\b"),
re.compile(r"(?im)^\s*pass\s*(?:#.*)?$"),
re.compile(r"(?im)\bnot\s+implemented\b"),
re.compile(r"(?im)\bplaceholder\b"),
re.compile(r"(?im)^\s*\.\.\.\s*$"),
)
class Severity(str, Enum):
BLOCKER = "BLOCKER"
MAJOR = "MAJOR"
MINOR = "MINOR"
INFO = "INFO"
class TruthState(str, Enum):
UNKNOWN = "UNKNOWN"
ADMITTED_NOT_PROJECTED = "ADMITTED_NOT_PROJECTED"
WAITING_CAPACITY = "WAITING_CAPACITY"
QUEUED = "QUEUED"
CLAIMED = "CLAIMED"
RUNNING = "RUNNING"
TERMINAL_UNPROVEN = "TERMINAL_UNPROVEN"
COMPLETE = "COMPLETE"
EXPIRED = "EXPIRED"
@dataclass(frozen=True)
class Finding:
code: str
severity: Severity
message: str
evidence: dict[str, Any] = field(default_factory=dict)
remediation: str | None = None
@dataclass
class AdmissionSnapshot:
task_id: str
envelope_id: str | None = None
admission_status: str | None = None
grant_id: str | None = None
grant_expires_at: int | None = None
one_time_use: bool | None = None
authority_base_oid: str | None = None
execution_digest: str | None = None
requested_capabilities: tuple[str, ...] = ()
file_allowlist: tuple[str, ...] = ()
admitted_at: int | None = None
@dataclass
class TaskSnapshot:
task_id: str
status: str | None = None
claimed_by: str | None = None
lease_expires_at: int | None = None
attempt: int | None = None
revision: int | None = None
created_at: str | None = None
updated_at: str | None = None
deadline: str | None = None
required_capabilities: tuple[str, ...] = ()
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class WorkerSnapshot:
worker_id: str
status: str | None = None
last_seen_epoch: int | None = None
capabilities: tuple[str, ...] = ()
scope_task_id: str | None = None
scope_job_id: str | None = None
scope_branch: str | None = None
scope_execution_digest: str | None = None
@dataclass
class LeaseSnapshot:
task_id: str
holder: str | None = None
expires_at: int | None = None
state: str | None = None
@dataclass
class GitSnapshot:
branch: str
current_oid: str | None = None
expected_oid: str | None = None
commit_oid: str | None = None
post_oid: str | None = None
@dataclass
class EvidenceSnapshot:
commit_receipt: dict[str, Any] | None = None
runtime_readback: dict[str, Any] | None = None
external_projection_readback: dict[str, Any] | None = None
output_hashes: dict[str, str] = field(default_factory=dict)
def parse_iso_epoch(value: str | None) -> int | None:
if not value:
return None
from datetime import datetime
try:
return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp())
except Exception:
return None
def worker_is_healthy(worker: WorkerSnapshot, now: int, heartbeat_ttl: int = 180) -> bool:
return bool(
worker.status == "ACTIVE"
and worker.last_seen_epoch is not None
and now - worker.last_seen_epoch <= heartbeat_ttl
)
def worker_scope_matches(worker: WorkerSnapshot, task: TaskSnapshot) -> bool:
meta = task.metadata or {}
plan = meta.get("execution_plan") or meta.get("plan") or meta
job_id = plan.get("external_job_id") if isinstance(plan, dict) else None
branch = plan.get("target_branch") if isinstance(plan, dict) else None
digest = plan.get("execution_digest") if isinstance(plan, dict) else None
if worker.scope_task_id and worker.scope_task_id != task.task_id:
return False
if worker.scope_job_id and job_id and worker.scope_job_id != job_id:
return False
if worker.scope_branch and branch and worker.scope_branch != branch:
return False
if worker.scope_execution_digest and digest and worker.scope_execution_digest != digest:
return False
return True
def worker_has_capabilities(worker: WorkerSnapshot, required: Sequence[str]) -> bool:
have = set(worker.capabilities)
return set(required).issubset(have)
def eligible_workers(task: TaskSnapshot, workers: Sequence[WorkerSnapshot], now: int, heartbeat_ttl: int = 180) -> list[WorkerSnapshot]:
return [
w for w in workers
if worker_is_healthy(w, now, heartbeat_ttl)
and worker_has_capabilities(w, task.required_capabilities)
and worker_scope_matches(w, task)
]
def derive_truth(
admission: AdmissionSnapshot | None,
task: TaskSnapshot | None,
workers: Sequence[WorkerSnapshot],
lease: LeaseSnapshot | None,
git: GitSnapshot | None,
evidence: EvidenceSnapshot | None,
now: int | None = None,
queue_claim_threshold: int = 600,
heartbeat_ttl: int = 180,
) -> tuple[TruthState, list[Finding]]:
now = int(time.time()) if now is None else now
findings: list[Finding] = []
if admission and admission.admission_status == "ADMITTED" and not task:
findings.append(Finding(
"DUAL_PLANE_ADMITTED_TASK_MISSING",
Severity.BLOCKER,
"Admission plane says ADMITTED but TaskPlane has no task.",
{"task_id": admission.task_id, "envelope_id": admission.envelope_id},
"Use durable projection/outbox + exact readback; Admission must not imply execution.",
))
return TruthState.ADMITTED_NOT_PROJECTED, findings
if not task:
return TruthState.UNKNOWN, findings
if admission and admission.grant_expires_at and admission.grant_expires_at <= now and task.status in {"QUEUED", "CLAIMED", "RUNNING"}:
findings.append(Finding(
"GRANT_EXPIRED_BEFORE_EXECUTION",
Severity.BLOCKER,
"Execution state depends on an expired grant.",
{"expires_at": admission.grant_expires_at, "now": now, "task_state": task.status},
"Mint runtime TaskPrincipal just-in-time at claim; do not burn grant TTL while waiting in queue.",
))
if admission and admission.admission_status == "ADMITTED" and task.status in {"EXPIRED", "FAILED", "CANCELLED"}:
findings.append(Finding(
"DUAL_PLANE_TERMINAL_DRIFT",
Severity.MAJOR,
"Admission remains ADMITTED while TaskPlane is terminal.",
{"admission": admission.admission_status, "task": task.status},
"Expose admission_state and execution_state separately; never collapse both into one PASS/complete status.",
))
if task.status == "QUEUED":
elig = eligible_workers(task, workers, now, heartbeat_ttl)
if not elig:
findings.append(Finding(
"QUEUED_WITHOUT_ELIGIBLE_WORKER",
Severity.BLOCKER,
"Task is QUEUED but no healthy, capability- and scope-matching worker exists.",
{
"required_capabilities": list(task.required_capabilities),
"healthy_workers": [w.worker_id for w in workers if worker_is_healthy(w, now, heartbeat_ttl)],
},
"Gate TaskPlane projection on a SchedulabilityProof; keep Admission ADMITTED/WAITING_CAPACITY until claimable.",
))
return TruthState.WAITING_CAPACITY, findings
created = parse_iso_epoch(task.created_at)
if created is not None and now - created > queue_claim_threshold and not task.claimed_by:
findings.append(Finding(
"UNCLAIMED_TIMEOUT",
Severity.BLOCKER,
"QUEUED task exceeded claim SLA without a claim holder.",
{"queued_seconds": now - created, "threshold": queue_claim_threshold},
"Trigger scheduler diagnosis/takeover contract; do not extend TTL blindly.",
))
return TruthState.QUEUED, findings
if task.status == "CLAIMED":
if not task.claimed_by:
findings.append(Finding(
"FAKE_CLAIM_STATE",
Severity.BLOCKER,
"Task says CLAIMED but claimed_by is empty.",
{"task_id": task.task_id},
"Reject state transition unless claim holder + fencing generation are persisted atomically.",
))
if lease is None or lease.holder != task.claimed_by or (lease.expires_at and lease.expires_at <= now):
findings.append(Finding(
"CLAIM_WITHOUT_VALID_LEASE",
Severity.BLOCKER,
"CLAIMED task has no valid matching lease.",
{"claimed_by": task.claimed_by, "lease": asdict(lease) if lease else None},
"Require claim+lease readback before exposing CLAIMED.",
))
return TruthState.CLAIMED, findings
if task.status == "RUNNING":
if not task.claimed_by:
findings.append(Finding("FAKE_RUNNING_NO_CLAIM", Severity.BLOCKER, "RUNNING without claim holder."))
claimed_worker = next((w for w in workers if w.worker_id == task.claimed_by), None)
if not claimed_worker or not worker_is_healthy(claimed_worker, now, heartbeat_ttl):
findings.append(Finding(
"FAKE_RUNNING_NO_HEARTBEAT",
Severity.BLOCKER,
"RUNNING task has no fresh worker heartbeat.",
{"claimed_by": task.claimed_by},
"Transition to recovery only through lease/fencing rules; never report RUNNING from a stale projection.",
))
if lease is None or (lease.expires_at and lease.expires_at <= now):
findings.append(Finding("RUNNING_WITHOUT_VALID_LEASE", Severity.BLOCKER, "RUNNING without valid lease."))
return TruthState.RUNNING, findings
if task.status in {"SUCCEEDED", "COMPLETE"}:
ev = evidence or EvidenceSnapshot()
commit = ev.commit_receipt or {}
before = git.expected_oid if git else None
post = (git.post_oid if git else None) or commit.get("post_oid")
commit_oid = (git.commit_oid if git else None) or commit.get("commit_oid")
if not commit_oid or not post or (before and post == before):
findings.append(Finding(
"TERMINAL_SUCCESS_UNPROVEN",
Severity.BLOCKER,
"Task reports success without a proven branch mutation receipt.",
{"commit_oid": commit_oid, "post_oid": post, "pre_oid": before},
"SUCCEEDED requires lease id, commit receipt, pre/post OID change, output hashes, and runtime/projection readback.",
))
return TruthState.TERMINAL_UNPROVEN, findings
if not ev.runtime_readback:
findings.append(Finding("SUCCESS_WITHOUT_RUNTIME_READBACK", Severity.MAJOR, "Commit exists but runtime readback is absent."))
return TruthState.COMPLETE if not findings else TruthState.TERMINAL_UNPROVEN, findings
if task.status == "EXPIRED":
if task.attempt in (None, 0) and not task.claimed_by:
findings.append(Finding(
"EXPIRED_WITHOUT_ANY_CLAIM",
Severity.MAJOR,
"Task expired without a single execution attempt.",
{"attempt": task.attempt, "claimed_by": task.claimed_by},
"Treat as schedulability failure, not execution failure; fix worker/principal supply before recovery.",
))
return TruthState.EXPIRED, findings
return TruthState.UNKNOWN, findings
def inspect_artifact(path: Path, expected_sha256: str | None = None, min_bytes: int = 8) -> list[Finding]:
findings: list[Finding] = []
if not path.is_file():
return [Finding("ARTIFACT_MISSING", Severity.BLOCKER, f"Artifact missing: {path}")]
data = path.read_bytes()
if len(data) < min_bytes:
findings.append(Finding("ARTIFACT_TOO_SMALL", Severity.BLOCKER, f"Artifact is too small: {path}", {"bytes": len(data)}))
digest = hashlib.sha256(data).hexdigest()
if expected_sha256 and digest != expected_sha256:
findings.append(Finding("ARTIFACT_HASH_MISMATCH", Severity.BLOCKER, f"Artifact hash mismatch: {path}", {"expected": expected_sha256, "actual": digest}))
try:
text = data.decode("utf-8")
except UnicodeDecodeError:
text = ""
if text and any(p.search(text) for p in PLACEHOLDER_PATTERNS):
findings.append(Finding(
"PLACEHOLDER_ARTIFACT",
Severity.BLOCKER,
f"Artifact contains placeholder/stub markers: {path}",
remediation="Reject plan admission until actual bytes exist and match the declared digest.",
))
return findings
def operation_ids_from_openapi(path: Path) -> dict[str, str]:
try:
import yaml # type: ignore
except ImportError as exc:
raise RuntimeError("PyYAML is required for OpenAPI parity checks") from exc
doc = yaml.safe_load(path.read_text(encoding="utf-8-sig"))
result: dict[str, str] = {}
for route, methods in (doc.get("paths") or {}).items():
if not isinstance(methods, dict):
continue
for _, spec in methods.items():
if isinstance(spec, dict) and spec.get("operationId"):
result[str(spec["operationId"])] = str(route)
return result
ROUTE_PAIR_RE = re.compile(r'["\'](?P<route>/v\d+/[^"\']+)["\']\s*:\s*["\'](?P<op>[A-Za-z0-9_]+)["\']')
def operation_ids_from_server_source(path: Path) -> dict[str, str]:
text = path.read_text(encoding="utf-8")
out: dict[str, str] = {}
for m in ROUTE_PAIR_RE.finditer(text):
out[m.group("op")] = m.group("route")
return out
def check_openapi_backend_parity(openapi_path: Path, server_path: Path, required_ops: Iterable[str] = ()) -> list[Finding]:
oa = operation_ids_from_openapi(openapi_path)
be = operation_ids_from_server_source(server_path)
findings: list[Finding] = []
for op in sorted(set(required_ops) | set(oa)):
if op in oa and op not in be:
findings.append(Finding(
"OPENAPI_DEAD_ENDPOINT",
Severity.BLOCKER,
f"OpenAPI advertises {op} ({oa[op]}) but backend route mapping is absent.",
remediation="Deploy projection and backend from the same source commit; reject schema-only rollout.",
))
for op in sorted(set(required_ops) | set(be)):
if op in be and op not in oa:
findings.append(Finding(
"BACKEND_UNPROJECTED_OPERATION",
Severity.MAJOR,
f"Backend implements {op} ({be[op]}) but OpenAPI projection is absent.",
remediation="Generate/validate external projection from the canonical backend route manifest.",
))
return findings
def check_capability_aliases(required: Sequence[str], workers: Sequence[WorkerSnapshot]) -> list[Finding]:
findings: list[Finding] = []
req = set(required)
if "governed.write.files" in req:
for w in workers:
caps = set(w.capabilities)
if "governed.write.file" in caps and "governed.write.files" not in caps:
findings.append(Finding(
"CAPABILITY_SINGULAR_PLURAL_ASYMMETRY",
Severity.MAJOR,
f"Worker {w.worker_id} exposes governed.write.file but task requires governed.write.files.",
{"worker_capabilities": sorted(caps)},
"Do not treat names as aliases implicitly; define one canonical capability or an explicit server-side compatibility mapping with tests.",
))
return findings
def git_ref(git_dir: Path, ref: str) -> str | None:
cp = subprocess.run(
["git", f"--git-dir={git_dir}", "rev-parse", ref],
text=True, capture_output=True
)
if cp.returncode != 0:
return None
value = cp.stdout.strip()
return value if re.fullmatch(r"[0-9a-f]{40}", value) else None
def sqlite_rows(path: Path, query: str, params: Sequence[Any] = ()) -> list[sqlite3.Row]:
uri = f"file:{path}?mode=ro&immutable=1"
con = sqlite3.connect(uri, uri=True)
con.row_factory = sqlite3.Row
try:
return list(con.execute(query, params))
finally:
con.close()
def dump_report(path: Path, report: Mapping[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(report, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8")
@@ -1,7 +1,27 @@
"""Non-sensitive control-plane snapshot assembly."""
"""Non-sensitive control-plane snapshot assembly.
The base snapshot is a pure field-copy of each plane's reported state. It does
NOT itself assert whether that state is backed by evidence — a drift_status=PASS
or an ADMITTED status is a *claim*, not a proof alert/status change.
`attach_control_plane_truth` (see below) injects the read-only derived `SystemTruth`
(states + findings) from `aios_control_plane_consistency.derive_truth` into the
snapshot so that explain/status surfaces can expose *derived* truth alongside
the reported state, without ever mutating the persistence planes.
"""
from __future__ import annotations
from typing import Any, Mapping
from typing import Any, Mapping, Sequence
from aios_control_plane_consistency import (
AdmissionSnapshot,
EvidenceSnapshot,
GitSnapshot,
LeaseSnapshot,
TaskSnapshot,
WorkerSnapshot,
derive_truth,
)
def _copy(value: Mapping[str, Any] | None, keys: tuple[str, ...]) -> dict[str, Any]:
@@ -28,3 +48,48 @@ def build_control_plane_snapshot(*, target_version: str, runtime: Mapping[str, A
def getControlPlaneSnapshot(**kwargs: Any) -> dict[str, Any]:
return build_control_plane_snapshot(**kwargs)
def attach_control_plane_truth(
snapshot: Mapping[str, Any],
*,
admission: AdmissionSnapshot | None = None,
task: TaskSnapshot | None = None,
workers: Sequence[WorkerSnapshot] = (),
lease: LeaseSnapshot | None = None,
git: GitSnapshot | None = None,
evidence: EvidenceSnapshot | None = None,
now: int | None = None,
queue_claim_threshold: int = 600,
heartbeat_ttl: int = 180,
) -> dict[str, Any]:
"""Attach read-only derived SystemTruth to a control-plane snapshot.
This never writes to any persistence plane. It derives a ``system_truth``
block from the already-loaded plane snapshots and adds it under
``snapshot["derived"]``, leaving all existing keys intact.
The block includes the derived truth state, any findings, and a compact
"equivalence" guard (see config/gitops/system_truth_contract.json
``forbidden_equivalences``) so surfaces cannot silently present a local
plane claim as end-to-end evidence.
"""
from dataclasses import asdict
truth_state, findings = derive_truth(
admission, task, workers, lease, git, evidence, now,
queue_claim_threshold=queue_claim_threshold,
heartbeat_ttl=heartbeat_ttl,
)
out = dict(snapshot)
if "derived" not in out:
out["derived"] = {}
blockers = [f for f in findings if f.severity.value == "BLOCKER"]
out["derived"]["system_truth"] = {
"truth_state": truth_state.value,
"findings": [asdict(f) for f in findings],
"blocker_count": sum(1 for f in findings if f.severity.value == "BLOCKER"),
"major_count": sum(1 for f in findings if f.severity.value == "MAJOR"),
# Derivation is read-only; it never confirms completion on its own.
"evidence_flow_confirmed": truth_state.value == "COMPLETE" and not blockers,
}
return out
+113 -113
View File
@@ -24,25 +24,25 @@ def _safe_oid(value: str, field: str) -> str:
return value
def _safe_branch(value: str) -> str:
branch = str(value or "")
if (
not branch
or len(branch) > 240
or branch != branch.strip()
or branch.startswith(("/", "-", "."))
or branch.endswith(("/", ".", ".lock"))
or ".." in branch
or "//" in branch
or "@{" in branch
or "\\" in branch
or any(part in {"", ".", ".."} for part in branch.split("/"))
or any(ch not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._/-" for ch in branch)
):
raise GiteaTransportError("BRANCH_INVALID")
return branch
def _safe_branch(value: str) -> str:
branch = str(value or "")
if (
not branch
or len(branch) > 240
or branch != branch.strip()
or branch.startswith(("/", "-", "."))
or branch.endswith(("/", ".", ".lock"))
or ".." in branch
or "//" in branch
or "@{" in branch
or "\\" in branch
or any(part in {"", ".", ".."} for part in branch.split("/"))
or any(ch not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._/-" for ch in branch)
):
raise GiteaTransportError("BRANCH_INVALID")
return branch
@dataclass(slots=True)
class GiteaMachineTransport:
api_base: str
@@ -110,7 +110,7 @@ class GiteaMachineTransport:
return self.repository_info()
def read_branch(self, branch: str) -> str | None:
branch = _safe_branch(branch)
branch = _safe_branch(branch)
owner, name = self.repository_id.split("/", 1)
try:
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}")[1]
@@ -156,99 +156,99 @@ class GiteaMachineTransport:
finally:
askpass.unlink(missing_ok=True)
def fetch_branch_exact(self, branch: str, expected_oid: str) -> str:
"""Fetch one exact remote branch tip into a quarantine ref.
This never updates ``main`` or any caller-visible branch. The remote
ref is read before and after the fetch so a concurrent branch move
cannot be silently admitted.
"""
branch = _safe_branch(branch)
expected_oid = _safe_oid(expected_oid, "expected_oid")
remote_before = self.read_branch(branch)
if remote_before != expected_oid:
raise GiteaTransportError("GITEA_SOURCE_REF_MISMATCH")
target_ref = f"refs/aios/release-sources/{expected_oid}"
current = subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"rev-parse",
"--verify",
f"{target_ref}^{{commit}}",
],
text=True,
capture_output=True,
check=False,
timeout=self.timeout,
)
if current.returncode == 0 and current.stdout.strip().lower() == expected_oid:
remote_after = self.read_branch(branch)
if remote_after != expected_oid:
raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
return expected_oid
askpass, additions = self._askpass()
environment = os.environ.copy()
environment.update(additions)
try:
result = subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"fetch",
"--no-tags",
"--no-write-fetch-head",
self.remote_url,
f"refs/heads/{branch}:{target_ref}",
],
text=True,
capture_output=True,
check=False,
env=environment,
timeout=self.timeout,
)
if result.returncode != 0:
raise GiteaTransportError("GITEA_SOURCE_FETCH_FAILED")
finally:
askpass.unlink(missing_ok=True)
local = subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"rev-parse",
"--verify",
f"{target_ref}^{{commit}}",
],
text=True,
capture_output=True,
check=False,
timeout=self.timeout,
)
local_oid = local.stdout.strip().lower()
if local.returncode != 0 or local_oid != expected_oid:
subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"update-ref",
"-d",
target_ref,
],
text=True,
capture_output=True,
check=False,
timeout=self.timeout,
)
raise GiteaTransportError("GITEA_SOURCE_FETCH_MISMATCH")
remote_after = self.read_branch(branch)
if remote_after != expected_oid:
raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
return expected_oid
def fetch_branch_exact(self, branch: str, expected_oid: str) -> str:
"""Fetch one exact remote branch tip into a quarantine ref.
This never updates ``main`` or any caller-visible branch. The remote
ref is read before and after the fetch so a concurrent branch move
cannot be silently admitted.
"""
branch = _safe_branch(branch)
expected_oid = _safe_oid(expected_oid, "expected_oid")
remote_before = self.read_branch(branch)
if remote_before != expected_oid:
raise GiteaTransportError("GITEA_SOURCE_REF_MISMATCH")
target_ref = f"refs/aios/release-sources/{expected_oid}"
current = subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"rev-parse",
"--verify",
f"{target_ref}^{{commit}}",
],
text=True,
capture_output=True,
check=False,
timeout=self.timeout,
)
if current.returncode == 0 and current.stdout.strip().lower() == expected_oid:
remote_after = self.read_branch(branch)
if remote_after != expected_oid:
raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
return expected_oid
askpass, additions = self._askpass()
environment = os.environ.copy()
environment.update(additions)
try:
result = subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"fetch",
"--no-tags",
"--no-write-fetch-head",
self.remote_url,
f"refs/heads/{branch}:{target_ref}",
],
text=True,
capture_output=True,
check=False,
env=environment,
timeout=self.timeout,
)
if result.returncode != 0:
raise GiteaTransportError("GITEA_SOURCE_FETCH_FAILED")
finally:
askpass.unlink(missing_ok=True)
local = subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"rev-parse",
"--verify",
f"{target_ref}^{{commit}}",
],
text=True,
capture_output=True,
check=False,
timeout=self.timeout,
)
local_oid = local.stdout.strip().lower()
if local.returncode != 0 or local_oid != expected_oid:
subprocess.run(
[
"git",
f"--git-dir={self.authority_git_dir}",
"update-ref",
"-d",
target_ref,
],
text=True,
capture_output=True,
check=False,
timeout=self.timeout,
)
raise GiteaTransportError("GITEA_SOURCE_FETCH_MISMATCH")
remote_after = self.read_branch(branch)
if remote_after != expected_oid:
raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
return expected_oid
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str:
_safe_oid(expected_oid, "expected_oid")
_safe_oid(new_oid, "new_oid")
@@ -155,6 +155,7 @@ SILENT_RELEASE_POST_PATHS = frozenset({
'/submitSignedApproval',
'/submitOwnerApproval',
'/startDeployment',
'/reopenOwnerApproval',
})
def reject_non_read_http(handler, method: str, path: str) -> bool:
@@ -974,6 +975,17 @@ class H(BaseHTTPRequestHandler):
return self.sendj(
silent_release_service().submit_owner_approval(data)
)
if path == "/reopenOwnerApproval":
a = self.need("release:approve")
if not a:
return
from release_approval_reopen import ApprovalReopenError, reopen_owner_approval
try:
result = reopen_owner_approval(silent_release_service(), data)
except ApprovalReopenError as exc:
return self.sendj({"error": exc.code, "message": str(exc)}, 409)
return self.sendj(result)
if path == '/startDeployment':
a=self.need('release:deploy')
if not a: return
@@ -929,10 +929,23 @@ class TaskAdmissionBootstrapService:
trust_domain=envelope.trust_domain, policy_digest=envelope.policy_digest,
authorization_kind=kind, owner_identity=owner_identity,
parent_epoch_id=parent_epoch_id, issued_at=now,
expires_at=min(
envelope.expires_at,
now + self.policy.max_task_ttl_seconds,
parent_epoch.expires_at if parent_epoch is not None else envelope.expires_at,
# V2R2C2 TTL contract: the admission envelope is a *signing
# window* only. Once authorization succeeds (epoch verified or
# Owner signature verified) the envelope is consumed, and the
# derived grant's lifecycle is bounded by the task/grant policy
# ceiling (max_task_ttl_seconds) and, for a delegation epoch,
# by the parent epoch's own expiry. It must NOT be truncated by
# the already-consumed envelope's remaining window, otherwise a
# nominal 3600s policy collapses to the ~900s envelope window
# (dual-semantic TTL defect). The envelope's temporal validity
# still gates authorization (now < envelope.expires_at above).
expires_at=(
now + self.policy.max_task_ttl_seconds
if parent_epoch is None
else min(
now + self.policy.max_task_ttl_seconds,
parent_epoch.expires_at,
)
),
)
persisted_grant, _ = self.store.claim_grant(
@@ -0,0 +1,349 @@
from __future__ import annotations
"""Authenticated Gitea consistency provider using the real governed Git transport.
The provider proves Gitea consistency without inventing a REST API:
* local HEAD == source_oid,
* local source_oid^{tree} == tree_oid,
* origin identifies the governed repository,
* exact checked-out branch exists remotely, and
* git ls-remote reports exactly source_oid for that branch.
It then writes an append-only root-owned v2 provider receipt using only governed
policy for provider identity, operation, store root, store id and modes.
The result explicitly records ``transport=git-ls-remote`` and
``api_verified=false``. It must never be described as REST/API evidence.
"""
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import subprocess
from typing import Callable, Mapping
from urllib.parse import urlsplit, urlunsplit
RECEIPT_SCHEMA = "aios.trusted-evidence-provider-receipt.v2"
RECEIPT_VERSION = 2
_OID_RE = re.compile(r"^[0-9a-f]{40}$")
_REF_RE = re.compile(r"^refs/heads/[A-Za-z0-9._/-]+$")
_REPO_RE = re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$")
class GiteaProviderError(RuntimeError):
pass
@dataclass(frozen=True)
class GitCommandResult:
returncode: int
stdout: str = ""
stderr: str = ""
GitRunner = Callable[[tuple[str, ...], Path], GitCommandResult]
def canonical_json_bytes(payload: Mapping) -> bytes:
return json.dumps(
dict(payload),
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def canonical_sha256(payload: Mapping) -> str:
return hashlib.sha256(canonical_json_bytes(payload)).hexdigest()
def _runner(args: tuple[str, ...], cwd: Path) -> GitCommandResult:
proc = subprocess.run(
list(args),
cwd=str(cwd),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
timeout=45,
check=False,
)
return GitCommandResult(proc.returncode, proc.stdout, proc.stderr)
def _git(run: GitRunner, cwd: Path, *args: str, error: str) -> str:
result = run(("git", *args), cwd)
if result.returncode != 0:
raise GiteaProviderError(f"{error}: {result.stderr.strip()[:300]}")
return result.stdout.strip()
def _oid(value: str, field: str) -> str:
value = str(value).strip().lower()
if not _OID_RE.fullmatch(value):
raise GiteaProviderError(f"{field.upper()}_INVALID")
return value
def _mode(value) -> int:
if isinstance(value, int):
return value
text = str(value).strip()
if text.isdigit():
return int(text, 8)
raise GiteaProviderError("PROVIDER_MODE_INVALID")
def _safe_ref(value: str) -> Path:
raw = str(value).strip()
p = Path(raw)
if not raw or p.is_absolute() or any(part in ("", ".", "..") for part in p.parts):
raise GiteaProviderError("PROVIDER_RECEIPT_REF_INVALID")
return p
def _safe_remote_url(url: str) -> str:
url = str(url).strip()
if "://" in url:
parts = urlsplit(url)
host = parts.hostname or ""
port = f":{parts.port}" if parts.port else ""
return urlunsplit((parts.scheme, host + port, parts.path, "", ""))
if ":" in url and not url.startswith(("/", "./", "../")):
left, right = url.split(":", 1)
return f"{left.rsplit('@', 1)[-1]}:{right}"
return url
def repository_identity_from_remote(url: str) -> str:
safe = _safe_remote_url(url)
if "://" in safe:
path = urlsplit(safe).path
elif ":" in safe and not safe.startswith("/"):
path = safe.split(":", 1)[1]
else:
path = safe
parts = [x for x in path.strip("/").split("/") if x]
if parts and parts[0] == "git":
parts = parts[1:]
if len(parts) < 2:
raise GiteaProviderError("GITEA_REPOSITORY_IDENTITY_UNAVAILABLE")
owner, repo = parts[-2], parts[-1]
if repo.endswith(".git"):
repo = repo[:-4]
identity = f"{owner}/{repo}".lower()
if not _REPO_RE.fullmatch(identity):
raise GiteaProviderError("GITEA_REPOSITORY_IDENTITY_INVALID")
return identity
def load_gitea_policy(policy_path: str | Path) -> dict:
raw = json.loads(Path(policy_path).read_text(encoding="utf-8"))
defaults = dict(raw.get("defaults") or {})
providers = raw.get("providers")
if not isinstance(providers, dict) or not isinstance(providers.get("gitea"), dict):
raise GiteaProviderError("GITEA_PROVIDER_POLICY_MISSING")
cfg = dict(providers["gitea"])
return {
"provider_identity": str(cfg["provider_identity"]),
"store_id": str(cfg["store_id"]),
"store_root": Path(str(cfg["store_root"])),
"provider_operation": str(cfg["allowed_operations"][0]),
"expected_owner_uid": int(cfg.get("expected_owner_uid", defaults.get("expected_owner_uid", 0))),
"directory_mode": min(
_mode(x) for x in cfg.get(
"allowed_directory_modes",
defaults.get("allowed_directory_modes", ["0700", "0750"]),
)
),
"receipt_mode": min(
_mode(x) for x in cfg.get(
"allowed_receipt_modes",
defaults.get("allowed_receipt_modes", ["0600", "0640"]),
)
),
"authentication_kind": str(
cfg.get(
"authentication_kind",
defaults.get("authentication_kind", "root-owned-store-v1"),
)
),
}
def collect_gitea_transport_consistency(
*,
repo_root: str | Path,
source_oid: str,
tree_oid: str,
expected_repository: str = "aios/aios",
remote: str = "origin",
git_runner: GitRunner | None = None,
) -> dict:
repo_root = Path(repo_root).resolve()
if not repo_root.is_dir():
raise GiteaProviderError("GITEA_CHECKOUT_UNAVAILABLE")
source_oid = _oid(source_oid, "source_oid")
tree_oid = _oid(tree_oid, "tree_oid")
expected_repository = str(expected_repository).strip().lower()
if not _REPO_RE.fullmatch(expected_repository):
raise GiteaProviderError("GITEA_EXPECTED_REPOSITORY_INVALID")
if not re.fullmatch(r"[A-Za-z0-9._/-]+", remote) or remote.startswith(("/", "-")):
raise GiteaProviderError("GITEA_REMOTE_NAME_INVALID")
run = git_runner or _runner
if _git(run, repo_root, "rev-parse", "--is-inside-work-tree",
error="GITEA_LOCAL_REPOSITORY_REQUIRED").lower() != "true":
raise GiteaProviderError("GITEA_LOCAL_REPOSITORY_REQUIRED")
actual_source = _oid(
_git(run, repo_root, "rev-parse", "HEAD^{commit}",
error="GITEA_LOCAL_SOURCE_UNAVAILABLE"),
"actual_source_oid",
)
if actual_source != source_oid:
raise GiteaProviderError("GITEA_LOCAL_SOURCE_OID_MISMATCH")
actual_tree = _oid(
_git(run, repo_root, "rev-parse", f"{source_oid}^{{tree}}",
error="GITEA_LOCAL_TREE_UNAVAILABLE"),
"actual_tree_oid",
)
if actual_tree != tree_oid:
raise GiteaProviderError("GITEA_LOCAL_TREE_OID_MISMATCH")
branch = _git(
run, repo_root, "symbolic-ref", "--quiet", "HEAD",
error="GITEA_GOVERNED_BRANCH_REQUIRED",
)
if not _REF_RE.fullmatch(branch):
raise GiteaProviderError("GITEA_GOVERNED_BRANCH_INVALID")
remote_url = _git(
run, repo_root, "remote", "get-url", remote,
error="GITEA_REMOTE_UNAVAILABLE",
)
identity = repository_identity_from_remote(remote_url)
if identity != expected_repository:
raise GiteaProviderError("GITEA_REPOSITORY_IDENTITY_MISMATCH")
output = _git(
run, repo_root, "ls-remote", "--exit-code", remote, branch,
error="GITEA_REMOTE_REF_UNAVAILABLE",
)
rows = []
for line in output.splitlines():
parts = line.split()
if len(parts) >= 2 and parts[1] == branch:
rows.append((parts[0].lower(), parts[1]))
if len(rows) != 1:
raise GiteaProviderError("GITEA_REMOTE_REF_AMBIGUOUS")
remote_oid = _oid(rows[0][0], "remote_ref_oid")
if remote_oid != source_oid:
raise GiteaProviderError("GITEA_REMOTE_REF_OID_MISMATCH")
return {
"consistent": True,
"repository": identity,
"source_commit": source_oid,
"remote_ref_oid": remote_oid,
"tree_oid": tree_oid,
"remote_ref": branch,
"remote_url": _safe_remote_url(remote_url),
"transport": "git-ls-remote",
"api_verified": False,
}
def _ensure_root_directory(path: Path, mode: int, expected_uid: int) -> None:
if os.geteuid() != expected_uid:
raise GiteaProviderError("GITEA_PROVIDER_ROOT_REQUIRED")
path.mkdir(parents=True, exist_ok=True, mode=mode)
st = os.lstat(path)
if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode):
raise GiteaProviderError("GITEA_PROVIDER_STORE_INVALID")
if st.st_uid != expected_uid:
raise GiteaProviderError("GITEA_PROVIDER_STORE_OWNER_INVALID")
if stat.S_IMODE(st.st_mode) & 0o022:
raise GiteaProviderError("GITEA_PROVIDER_STORE_MODE_INVALID")
def _write_receipt_exclusive(path: Path, payload: Mapping, mode: int) -> str:
data = canonical_json_bytes(payload)
if path.exists() or path.is_symlink():
raise GiteaProviderError("GITEA_PROVIDER_RECEIPT_EXISTS")
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
flags |= getattr(os, "O_NOFOLLOW", 0)
flags |= getattr(os, "O_BINARY", 0)
fd = os.open(path, flags, mode)
try:
view = memoryview(data)
while view:
written = os.write(fd, view)
if written <= 0:
raise GiteaProviderError("GITEA_PROVIDER_RECEIPT_WRITE_FAILED")
view = view[written:]
os.fsync(fd)
finally:
os.close(fd)
os.chmod(path, mode)
return hashlib.sha256(data).hexdigest()
def produce_gitea_consistency_receipt(
*,
repo_root: str | Path,
source_oid: str,
tree_oid: str,
policy_path: str | Path,
receipt_ref: str,
expected_repository: str = "aios/aios",
) -> tuple[str, dict]:
cfg = load_gitea_policy(policy_path)
if cfg["authentication_kind"] != "root-owned-store-v1":
raise GiteaProviderError("GITEA_PROVIDER_AUTH_KIND_UNSUPPORTED")
result = collect_gitea_transport_consistency(
repo_root=repo_root,
source_oid=source_oid,
tree_oid=tree_oid,
expected_repository=expected_repository,
)
now = datetime.now(timezone.utc).isoformat()
run_id = f"gitea-git-transport-{source_oid[:12]}-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}"
rel = _safe_ref(receipt_ref)
target = cfg["store_root"].joinpath(rel)
root = cfg["store_root"].resolve()
if not str(target.resolve(strict=False)).startswith(str(root) + os.sep):
raise GiteaProviderError("GITEA_PROVIDER_RECEIPT_REF_ESCAPE")
_ensure_root_directory(root, cfg["directory_mode"], cfg["expected_owner_uid"])
parent = root
for part in rel.parts[:-1]:
parent = parent / part
_ensure_root_directory(parent, cfg["directory_mode"], cfg["expected_owner_uid"])
payload = {
"schema": RECEIPT_SCHEMA,
"version": RECEIPT_VERSION,
"provider_identity": cfg["provider_identity"],
"provider_operation": cfg["provider_operation"],
"operation_run_id": run_id,
"result_reference": f"git:{result['remote_url']}#{result['remote_ref']}",
"content_sha256": canonical_sha256(result),
"source_oid": _oid(source_oid, "source_oid"),
"tree_oid": _oid(tree_oid, "tree_oid"),
"environment": "hw-isolated-linux-git-transport",
"issued_at": now,
"result": result,
"authentication": {
"kind": "root-owned-store-v1",
"store_id": cfg["store_id"],
},
}
sha = _write_receipt_exclusive(target, payload, cfg["receipt_mode"])
return sha, payload
@@ -0,0 +1,58 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from governed_provider_auth_adapter import (
ALL_RELEASE_EVIDENCE_KINDS,
authenticate_build_artifacts,
gate_w_authentication_status,
)
from trusted_provider_receipts import (
ProviderAuthenticationError,
TrustedProviderReceiptResolver,
)
def produce_authenticated_evidence_closure(
request: Any,
*,
provider_receipt_resolver: TrustedProviderReceiptResolver | None,
source_root: Path,
builder_policy: Path,
legacy_producer=None,
):
"""Production wrapper for the existing evidence producer.
This is the governed operation entrypoint. It refuses to call the legacy
closure assembler until *all* final-release provider receipts are resolved
from authenticated stores. For Gate-W, use `validate_gate_w_provider_auth`
instead; canary/rollback remain pending until Gate-R.
"""
if provider_receipt_resolver is None:
raise ProviderAuthenticationError("PROVIDER_AUTHENTICATION_REQUIRED")
if legacy_producer is None:
from governed_release_evidence_producer import produce_evidence_closure as legacy_producer
authenticated_request = authenticate_build_artifacts(
request,
resolver=provider_receipt_resolver,
required_kinds=ALL_RELEASE_EVIDENCE_KINDS,
)
return legacy_producer(
authenticated_request,
source_root=source_root,
builder_policy=builder_policy,
)
def validate_gate_w_provider_auth(
request: Any,
*,
provider_receipt_resolver: TrustedProviderReceiptResolver | None,
) -> dict[str, str]:
"""Gate-W authentication check without pretending Gate-R evidence exists."""
return gate_w_authentication_status(
request,
resolver=provider_receipt_resolver,
)
@@ -0,0 +1,323 @@
#!/usr/bin/env python3
"""Governed evidence runtime registration chain (V2R2C2 P0).
Establishes the REAL runtime chain for the governed release evidence producer:
Inventory -> Resolver -> operation -> Producer -> closure
The Producer alone (governed_release_evidence_producer) is not "runtime
registered" just because its OperationDefinition exists — registration only
holds when a runtime can actually discover the operation in an inventory,
resolve its ``handler_ref`` to the real callable, and invoke it to produce a
closure. That full path is what this module wires and what the
``PRODUCER_RUNTIME_REGISTERED_PASS`` verdict proves.
- Inventory : EvidenceOperationInventory — the set of governed evidence
operations the authority can execute (unions the producer
registry; extensible to future governed operations).
- Resolver : resolve_handler(handler_ref) maps the operation's declared
handler reference to the actual ``produce_evidence_closure``
callable. Resolution is exact and restrictive — it never
imports by arbitrary string eval.
- Producer : the resolved callable executes with validated, receipt-bound
build artifacts and materializes the evidence closure.
- closure : the produced source_root/<version>/ closure.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping
from aios_baseline.operations import OperationDefinition, OperationRegistry
from governed_authenticated_evidence_operation import (
produce_authenticated_evidence_closure,
)
from governed_provider_auth_adapter import gate_w_authentication_status
from governed_release_evidence_producer import (
EvidenceProduceError,
GovernedBuildArtifacts,
ProduceResult,
TRUSTED_PROVIDER_EVIDENCE_TYPES,
AUTHENTICATED_EVIDENCE_HANDLER_REF,
produce_evidence_closure,
producer_operation_definition,
producer_registry,
)
from trusted_provider_receipts import (
ProviderAuthenticationError,
TrustedProviderPolicy,
TrustedProviderReceiptResolver,
)
# Exact, restrictive handler_ref -> callable map. This is the ONLY way
# handler_refs are resolved; no arbitrary import/eval.
_HANDLERS: Mapping[str, Callable[..., ProduceResult]] = {
"governed_release_evidence_producer.produce_evidence_closure": produce_evidence_closure,
AUTHENTICATED_EVIDENCE_HANDLER_REF: produce_authenticated_evidence_closure,
}
class EvidenceOperationInventory:
"""Inventory of governed evidence operations available to the authority."""
def __init__(self, *registries: OperationRegistry) -> None:
definitions: list[OperationDefinition] = []
seen: set[str] = set()
for reg in registries:
for definition in reg.definitions:
if definition.operation_id in seen:
continue
seen.add(definition.operation_id)
definitions.append(definition)
self._registry = OperationRegistry(definitions)
self._by_id = {d.operation_id: d for d in definitions}
@classmethod
def default(cls) -> "EvidenceOperationInventory":
"""The standard inventory: the governed release evidence producer."""
return cls(producer_registry())
def contains(self, operation_id: str) -> bool:
return operation_id in self._by_id
def operation_ids(self) -> tuple[str, ...]:
return tuple(self._by_id)
def get(self, operation_id: str) -> OperationDefinition | None:
return self._by_id.get(operation_id)
def resolve_handler(handler_ref: str) -> Callable[..., ProduceResult]:
"""Resolver: turn an OperationDefinition.handler_ref into the real callable.
Exact lookup in a fixed allowlist. Unknown or non-producer references raise.
"""
handler = _HANDLERS.get(handler_ref)
if handler is None:
raise EvidenceProduceError("HANDLER_REF_UNRESOLVED")
return handler
def _payload_to_artifacts(payload: Mapping[str, Any]) -> GovernedBuildArtifacts:
try:
return GovernedBuildArtifacts(
version=payload["version"],
candidate_id=payload["candidate_id"],
source_commit=payload["source_commit"],
tree_oid=payload["tree_oid"],
artifact=bytes.fromhex(payload["artifact_hex"]),
manifest=payload["manifest"],
package_closure=bytes.fromhex(payload["package_closure_hex"]),
policy=payload["policy"],
tests=payload["tests"],
canary=payload["canary"],
gitea=payload["gitea"],
rollback=payload["rollback"],
secret_scan=payload["secret_scan"],
deployment=payload["deployment"],
tag=payload["tag"],
finalization=payload["finalization"],
rollback_sha256=payload.get("rollback_sha256", ""),
issued_at=payload.get("issued_at", ""),
)
except (KeyError, TypeError, ValueError) as exc:
raise EvidenceProduceError("PAYLOAD_SCHEMA_INVALID") from exc
def invoke_governed_produce(
*,
builder_policy: Path,
source_root: Path,
payload: Mapping[str, Any],
expected_owner_uid: int | None = 0,
) -> dict[str, Any]:
"""Real runtime chain: Inventory -> Resolver -> operation -> Producer -> closure.
Returns a verdict dict. Only a genuine end-to-end invocation that reaches
evidence production reports ``status == "PRODUCER_RUNTIME_REGISTERED_PASS"``.
"""
inventory = EvidenceOperationInventory.default()
operation_id = "governed.release.evidence.produce"
if not inventory.contains(operation_id):
raise EvidenceProduceError("OPERATION_NOT_IN_INVENTORY")
definition = inventory.get(operation_id)
assert definition is not None # guaranteed by contains()
handler = resolve_handler(definition.handler_ref)
request = _payload_to_artifacts(payload)
result = handler(
builder_policy=builder_policy,
source_root=source_root,
request=request,
expected_owner_uid=expected_owner_uid,
)
# The resolved handler returned a real ProduceResult from a real closure;
# the registered operation genuinely produced evidence.
return {
"status": "PRODUCER_RUNTIME_REGISTERED_PASS",
"operation_id": operation_id,
"handler_ref": definition.handler_ref,
"handler_resolved": handler.__module__ + "." + handler.__name__,
"evidence_types_bound": list(TRUSTED_PROVIDER_EVIDENCE_TYPES),
"result": {
"produce_status": result.status,
"idempotent": result.idempotent,
"receipt_sha256": result.receipt_sha256,
"closure_dir": result.closure_dir,
"evidence_digests": dict(result.evidence_digests),
},
}
# ---------------------------------------------------------------------------
# V2R2C2 Round-3: Provider-Authenticated runtime chain.
#
# Inventory -> Resolver -> authenticated operation -> TrustedProviderReceiptResolver
# -> Producer assembler (second-layer bound)
#
# The Round-3 runtime builds TrustedProviderReceiptResolver from the governed
# evidence-provider policy (never from request-supplied store paths, provider
# identities, key ids, or HMAC material). Gate-W authenticates only the
# provider classes that can exist before staging (tests/Gitea/secret-scan);
# canary/rollback remain pending until Gate-R.
# ---------------------------------------------------------------------------
# Sequence used for a full final release closure (Gate-R).
ALL_RELEASE_EVIDENCE_KINDS = ("tests", "canary", "gitea", "rollback", "secret_scan")
# Provider classes that are real before staging (Gate-W).
GATE_W_EVIDENCE_KINDS = ("tests", "gitea", "secret_scan")
_DEFAULT_PROVIDER_POLICY = Path(__file__).resolve().parents[2] / "config" / "gitops" / "trusted-evidence-providers.json"
def _load_provider_policy(path: Path) -> Mapping[str, Any]:
try:
decoded = json.loads(Path(path).read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise EvidenceProduceError("PROVIDER_POLICY_UNAVAILABLE") from exc
if not isinstance(decoded, Mapping) or decoded.get("schema") != "aios.trusted-evidence-providers.v1":
raise EvidenceProduceError("PROVIDER_POLICY_INVALID")
return decoded
def _policy_from_config(name: str, raw: Mapping[str, Any], defaults: Mapping[str, Any]) -> TrustedProviderPolicy:
def _mode(value: object, field: str) -> int:
if isinstance(value, int):
return value
if isinstance(value, str) and value.isdigit():
return int(value, 8)
raise EvidenceProduceError("PROVIDER_POLICY_INVALID")
try:
store_root = Path(str(raw["store_root"])).expanduser()
allowed_ops = frozenset(str(op) for op in raw["allowed_operations"])
return TrustedProviderPolicy(
evidence_kind=name,
provider_identity=str(raw["provider_identity"]),
store_id=str(raw["store_id"]),
store_root=store_root,
allowed_operations=allowed_ops,
expected_owner_uid=_mode(raw.get("expected_owner_uid", defaults.get("expected_owner_uid", 0)), "expected_owner_uid"),
allowed_directory_modes=frozenset(
_mode(v, "allowed_directory_modes")
for v in raw.get("allowed_directory_modes", defaults.get("allowed_directory_modes", ["0700", "0750"]))
),
allowed_receipt_modes=frozenset(
_mode(v, "allowed_receipt_modes")
for v in raw.get("allowed_receipt_modes", defaults.get("allowed_receipt_modes", ["0600", "0640"]))
),
authentication_kind=str(raw.get("authentication_kind", defaults.get("authentication_kind", "root-owned-store-v1"))),
max_age_seconds=raw.get("max_age_seconds", defaults.get("max_age_seconds")),
)
except (KeyError, TypeError, ValueError) as exc:
raise EvidenceProduceError("PROVIDER_POLICY_INVALID") from exc
def build_provider_receipt_resolver(
provider_policy_path: Path = _DEFAULT_PROVIDER_POLICY,
*,
hmac_key_resolver: Callable[[str], bytes] | None = None,
now_fn: Callable[[], Any] | None = None,
) -> TrustedProviderReceiptResolver:
"""Build TrustedProviderReceiptResolver from the governed evidence-provider policy.
Store paths, provider identities, key ids and HMAC material come only from
the governed policy/secret resolver — never from a request payload.
"""
policy = _load_provider_policy(provider_policy_path)
defaults = dict(policy.get("defaults") or {})
providers = policy.get("providers")
if not isinstance(providers, Mapping) or not providers:
raise EvidenceProduceError("PROVIDER_POLICY_INVALID")
policies: dict[str, TrustedProviderPolicy] = {}
for name, raw in providers.items():
if not isinstance(raw, Mapping):
raise EvidenceProduceError("PROVIDER_POLICY_INVALID")
policies[name] = _policy_from_config(name, raw, defaults)
return TrustedProviderReceiptResolver(policies, hmac_key_resolver=hmac_key_resolver, now_fn=now_fn)
def _resolver_guard(resolver: TrustedProviderReceiptResolver | None) -> TrustedProviderReceiptResolver:
if resolver is None:
raise ProviderAuthenticationError("PROVIDER_AUTHENTICATION_REQUIRED")
return resolver
def validate_gate_w_provider_auth(
request: GovernedBuildArtifacts,
*,
provider_receipt_resolver: TrustedProviderReceiptResolver | None,
) -> dict[str, str]:
"""Gate-W: authenticate only the provider classes real before staging.
Returns PRODUCER_RUNTIME_REGISTRATION_PASS / PROVIDER_AUTH_ENFORCEMENT_PASS /
FINAL_RELEASE_CLOSURE_PENDING_GATE_R. Canary/rollback are intentionally not
required here — they remain pending until Gate-R.
"""
resolver = _resolver_guard(provider_receipt_resolver)
gate_w_authentication_status(request, resolver=resolver)
return {
"producer_runtime": "PRODUCER_RUNTIME_REGISTRATION_PASS",
"provider_auth": "PROVIDER_AUTH_ENFORCEMENT_PASS",
"final_release_closure": "FINAL_RELEASE_CLOSURE_PENDING_GATE_R",
}
def _legacy_closure_adapter(authenticated_request: GovernedBuildArtifacts, *, source_root: Path, builder_policy: Path) -> ProduceResult:
"""Invoke the legacy closure assembler on the resolver-authenticated request.
The classified evidence fields now carry ``provider_receipt`` provenance;
the producer's second-layer validator re-asserts content/OID binding on it.
"""
return produce_evidence_closure(
builder_policy=builder_policy,
source_root=source_root,
request=authenticated_request,
expected_owner_uid=0,
)
def produce_authenticated_evidence(
request: GovernedBuildArtifacts,
*,
provider_receipt_resolver: TrustedProviderReceiptResolver | None,
source_root: Path,
builder_policy: Path,
required_kinds: Iterable[str] = ALL_RELEASE_EVIDENCE_KINDS,
) -> ProduceResult:
"""Gate-R production path: authenticate all five final-release provider
receipts, then invoke the legacy assembler only after authentication."""
resolver = _resolver_guard(provider_receipt_resolver)
try:
return produce_authenticated_evidence_closure(
request,
provider_receipt_resolver=resolver,
source_root=source_root,
builder_policy=builder_policy,
legacy_producer=_legacy_closure_adapter,
)
except ProviderAuthenticationError as exc:
raise EvidenceProduceError(str(exc)) from exc
@@ -0,0 +1,87 @@
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",
}
@@ -0,0 +1,634 @@
#!/usr/bin/env python3
"""Governed release evidence producer (V2R2C2 P0, operation governed.release.evidence.produce).
This module is the *producer* half of the governed evidence pipeline. It never
invents candidate/release evidence: every digest and pass/fail flag in the closure
is derived from real build artifacts and real verification executions supplied as
input. It assembles a self-consistent, root-owned evidence closure below
``source_root/<version>`` that ``TrustedBuilderEvidenceMaterializer`` (the projection
half) can re-validate and project byte-for-byte.
Contract (see docs/V2R2C2-P0-convergence-20260909.md section 1):
- operation_id : governed.release.evidence.produce
- input : governed build artifacts + deployment conventions (no hand-write)
- output : source_root/<version>/ full evidence_layout closure + receipt.json
- fail-closed : mismatched secret-scan digest, tampered artifact, non-passing
tests/canary/rollback/secret-scan, or inconsistent gitea all abort
without writing a closure.
- idempotent : an existing identical closure is accepted; a conflicting one fails.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import secrets
from dataclasses import dataclass, field
from pathlib import Path
import shutil
import stat
from typing import Any, Mapping
from aios_baseline.operations import OperationDefinition, OperationRegistry
from aios_trusted_candidate_builder import (
load_builder_policy,
TrustedCandidateBuilderConfig,
)
from release_evidence_materializer import EvidenceMaterializationError
from release_version import validate_release_version
PRODUCE_SCHEMA = "aios.governed-release-evidence-producer.v1"
EVIDENCE_RECEIPT_SCHEMA = "aios.trusted-candidate-builder.evidence.v1"
# Each governed evidence type (tests/canary/gitea/rollback/secret_scan) must be
# bound to a trusted-provider receipt. A plain caller-supplied {"passed": true}
# is never accepted on its own: the producer requires the receipt that proves a
# trusted provider actually ran the check and produced the referenced result.
TRUSTED_PROVIDER_RECEIPT_SCHEMA = "aios.trusted-provider.evidence.v1"
TRUSTED_PROVIDER_EVIDENCE_TYPES = ("tests", "canary", "gitea", "rollback", "secret_scan")
class EvidenceProduceError(RuntimeError):
"""Stable, non-sensitive producer failure."""
def _canonical_json(value: Mapping[str, Any]) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
).encode("utf-8")
def _file_sha256(path: Path) -> str:
digest = hashlib.sha256()
with Path(path).open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _sha256_bytes(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def _validate_oid(value: object, name: str) -> str:
normalized = str(value or "").strip().lower()
if (
len(normalized) != 40
or any(char not in "0123456789abcdef" for char in normalized)
):
raise EvidenceProduceError(f"{name.upper()}_INVALID")
return normalized
def _validate_identifier(value: object, name: str) -> str:
normalized = str(value or "").strip()
if (
not normalized
or len(normalized) > 255
or normalized[0] in ".-"
or any(
char not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789._:-"
for char in normalized
)
):
raise EvidenceProduceError(f"{name.upper()}_INVALID")
return normalized
def _validate_sha256(value: object, name: str, *, allow_empty: bool = False) -> str:
normalized = str(value or "").strip().lower()
if allow_empty and not normalized:
return ""
if (
len(normalized) != 64
or any(char not in "0123456789abcdef" for char in normalized)
):
raise EvidenceProduceError(f"{name.upper()}_INVALID")
return normalized
@dataclass(frozen=True)
class GovernedBuildArtifacts:
"""Real build artifacts + real verification results. Producer never invents these."""
version: str
candidate_id: str
source_commit: str
tree_oid: str
artifact: bytes
manifest: Mapping[str, Any]
package_closure: bytes
policy: Mapping[str, Any]
tests: Mapping[str, Any]
canary: Mapping[str, Any]
gitea: Mapping[str, Any]
rollback: Mapping[str, Any]
secret_scan: Mapping[str, Any]
# Deployment conventions (supplied by release planning, not evidence invention).
deployment: Mapping[str, Any]
tag: Mapping[str, Any]
finalization: Mapping[str, Any]
rollback_sha256: str = ""
issued_at: str = ""
@dataclass(frozen=True)
class ProduceResult:
status: str
idempotent: bool
receipt_sha256: str
closure_dir: str
evidence_digests: Mapping[str, str] = field(default_factory=dict)
def _require_trusted_provider_receipt(
name: str,
evidence: Mapping[str, Any],
source_commit: str,
tree_oid: str,
) -> None:
"""Require a trusted-provider binding for `evidence` (second-layer check).
A caller may never pass a bare ``{"passed": true}``: each governed evidence
type must be bound to a trusted provider. Two accepted forms, both
fail-closed on any missing/structurally invalid/digest-mismatched binding:
1. Round-2 native ``receipt`` (schema ``aios.trusted-provider.evidence.v1``)
— a caller-carried receipt.
2. Round-3 resolver-authenticated ``provider_receipt`` provenance (schema
``aios.trusted-evidence-provider-receipt.v2``) — produced by
``TrustedProviderReceiptResolver`` from an authenticated store. Its input
is the resolver-authenticated provenance, never a caller-supplied body.
This helper is a *second-layer* binding validator: for the Round-3 path the
authoritative proof is the resolver's authenticated provenance embedded as
``provider_receipt``, and this check only re-asserts content/OID binding on
top of it.
"""
receipt = evidence.get("receipt")
if isinstance(receipt, Mapping) and receipt:
body = {k: v for k, v in dict(evidence).items() if k != "receipt"}
expected = _sha256_bytes(_canonical_json(body))
if receipt.get("schema") != TRUSTED_PROVIDER_RECEIPT_SCHEMA:
raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID")
if receipt.get("content_sha256") != expected:
raise EvidenceProduceError("EVIDENCE_RECEIPT_CONTENT_MISMATCH")
for f in ("provider_identity", "operation_run_id", "result_reference",
"environment", "issued_at"):
if not str(receipt.get(f, "") or "").strip():
raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID")
if receipt.get("source_oid") not in (None, "", source_commit):
raise EvidenceProduceError("EVIDENCE_RECEIPT_SOURCE_OID_MISMATCH")
if receipt.get("tree_oid") not in (None, "", tree_oid):
raise EvidenceProduceError("EVIDENCE_RECEIPT_TREE_OID_MISMATCH")
return
provider_receipt = evidence.get("provider_receipt")
if isinstance(provider_receipt, Mapping) and provider_receipt:
if (
provider_receipt.get("schema")
!= "aios.trusted-evidence-provider-receipt.v2"
):
raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID")
for f in ("provider_identity", "provider_operation", "operation_run_id",
"result_reference", "content_sha256", "source_oid", "tree_oid",
"issued_at"):
if not str(provider_receipt.get(f, "") or "").strip():
raise EvidenceProduceError("EVIDENCE_RECEIPT_INVALID")
if provider_receipt.get("source_oid") != source_commit:
raise EvidenceProduceError("EVIDENCE_RECEIPT_SOURCE_OID_MISMATCH")
if provider_receipt.get("tree_oid") != tree_oid:
raise EvidenceProduceError("EVIDENCE_RECEIPT_TREE_OID_MISMATCH")
return
raise EvidenceProduceError("EVIDENCE_RECEIPT_MISSING")
def _verify_artifacts(
config: TrustedCandidateBuilderConfig,
request: GovernedBuildArtifacts,
) -> dict[str, str]:
"""Validate input consistency (fail-closed) and return the evidence digests."""
try:
version = validate_release_version(request.version)
except ValueError as exc:
raise EvidenceProduceError("VERSION_INVALID") from exc
if version != request.version:
raise EvidenceProduceError("VERSION_INVALID")
candidate_id = _validate_identifier(request.candidate_id, "candidate_id")
source_commit = _validate_oid(request.source_commit, "source_commit")
tree_oid = _validate_oid(request.tree_oid, "tree_oid")
if (
set(request.manifest) != {
"artifact_sha256", "package_closure_sha256",
}
):
raise EvidenceProduceError("MANIFEST_SCHEMA_INVALID")
closure_sha = _validate_sha256(
request.manifest.get("package_closure_sha256"), "package_closure_sha256"
)
artifact_sha = _sha256_bytes(request.artifact)
if request.manifest["artifact_sha256"] != artifact_sha:
raise EvidenceProduceError("MANIFEST_ARTIFACT_DIGEST_MISMATCH")
# manifest_sha256 is the digest of the manifest evidence file itself. It is
# derived (never invented) from the manifest content; the manifest does not
# embed its own digest to avoid a self-referential fixed point.
manifest_sha = _sha256_bytes(_canonical_json(dict(request.manifest)))
if _sha256_bytes(request.package_closure) != closure_sha:
raise EvidenceProduceError("PACKAGE_CLOSURE_DIGEST_MISMATCH")
# policy_sha256 is the digest of the policy evidence file content; derived
# here (never invented, never self-referential).
policy_sha = _sha256_bytes(_canonical_json(dict(request.policy)))
# Secret scan must cover the exact artifact bytes.
secret_scan_artifact = _validate_sha256(
request.secret_scan.get("artifact_sha256"), "secret_scan.artifact_sha256"
)
if secret_scan_artifact != artifact_sha:
raise EvidenceProduceError("SECRET_SCAN_ARTIFACT_DIGEST_MISMATCH")
# Every governed evidence type must be bound to a trusted-provider receipt.
# A bare caller-supplied {"passed": true} (no receipt) fails closed here;
# the producer never accepts an invented pass flag.
for etype in TRUSTED_PROVIDER_EVIDENCE_TYPES:
_require_trusted_provider_receipt(
etype, getattr(request, etype), source_commit, tree_oid
)
if request.secret_scan.get("passed") is not True:
raise EvidenceProduceError("SECRET_SCAN_NOT_PASSED")
if request.tests.get("passed") is not True:
raise EvidenceProduceError("TESTS_NOT_PASSED")
if request.canary.get("passed") is not True:
raise EvidenceProduceError("CANARY_NOT_PASSED")
if request.rollback.get("ready") is not True:
raise EvidenceProduceError("ROLLBACK_NOT_READY")
if (
request.gitea.get("consistent") is not True
or request.gitea.get("repository") != config.repository
or request.gitea.get("source_commit") != source_commit
or request.gitea.get("remote_ref_oid") != source_commit
or request.gitea.get("tree_oid") != tree_oid
):
raise EvidenceProduceError("GITEA_INCONSISTENT")
digests = {
"artifact": artifact_sha,
"manifest": manifest_sha,
"package_closure": closure_sha,
"policy": policy_sha,
"tests": _sha256_bytes(_canonical_json(dict(request.tests))),
"canary": _sha256_bytes(_canonical_json(dict(request.canary))),
"gitea": _sha256_bytes(_canonical_json(dict(request.gitea))),
"rollback": _sha256_bytes(_canonical_json(dict(request.rollback))),
"secret_scan": _sha256_bytes(_canonical_json(dict(request.secret_scan))),
}
return digests
def _build_receipt(
config: TrustedCandidateBuilderConfig,
request: GovernedBuildArtifacts,
) -> Mapping[str, Any]:
"""Assemble the self-consistent receipt from verified artifacts."""
from silent_release_controller import ReleaseCandidate, ReleaseState
artifact_sha = _sha256_bytes(request.artifact)
manifest_sha = _sha256_bytes(_canonical_json(dict(request.manifest)))
closure_sha = _validate_sha256(
request.manifest.get("package_closure_sha256"), "package_closure_sha256"
)
policy_sha = _sha256_bytes(_canonical_json(dict(request.policy)))
candidate_kwargs: dict[str, Any] = {}
issued_at = str(request.issued_at or "").strip()
if issued_at:
# The evidence was produced at the build execution instant; bind it so the
# same verified artifacts reproduce an identical, idempotent closure.
candidate_kwargs["created_at"] = issued_at
candidate_kwargs["updated_at"] = issued_at
candidate = ReleaseCandidate(
candidate_id=request.candidate_id,
source_commit=request.source_commit,
tree_oid=request.tree_oid,
artifact_sha256=artifact_sha,
manifest_sha256=manifest_sha,
tests=dict(request.tests),
canary=dict(request.canary),
rollback_ready=request.rollback.get("ready") is True,
git_consistency={
"consistent": True,
"repository": config.repository,
"source_commit": request.source_commit,
"tree_oid": request.tree_oid,
},
gitea_consistency=dict(request.gitea),
state=ReleaseState.READY_FOR_APPROVAL,
**candidate_kwargs,
)
deployment_payload = dict(request.deployment)
deployment_payload.update(
{
"service": request.deployment["service"],
"source_commit": request.source_commit,
"tree_oid": request.tree_oid,
"artifact_sha256": artifact_sha,
"manifest_sha256": manifest_sha,
"package_closure_sha256": closure_sha,
"policy_sha256": policy_sha,
"rollback_sha256": _validate_sha256(
request.rollback_sha256, "rollback_sha256"
)
or request.rollback.get("rollback_sha256", ""),
"repository_identity": config.repository,
"authoritative_git_dir": request.deployment["authoritative_git_dir"],
}
)
if not deployment_payload["rollback_sha256"]:
raise EvidenceProduceError("ROLLBACK_SHA256_REQUIRED")
from silent_release_hardening import (
DeploymentPlan,
FinalizationPlan,
ReleasePlan,
TagPlan,
TrustedCandidateBundle,
)
release_plan = ReleasePlan(
deployment=DeploymentPlan.from_dict(deployment_payload),
tag=TagPlan.from_dict(dict(request.tag)),
finalization=FinalizationPlan.from_dict(dict(request.finalization)),
policy_sha256=policy_sha,
)
checks = {name: True for name in config.required_checks}
TrustedCandidateBundle(candidate, release_plan, checks).validate()
return {
"schema": EVIDENCE_RECEIPT_SCHEMA,
"version": request.version,
"candidate": candidate.to_dict(),
"release_plan": release_plan.to_dict(),
}
def _safe_target(source_root: Path, version: str) -> Path:
relative = Path(version)
if relative.is_absolute() or any(
part in {"", ".", ".."} for part in relative.parts
):
raise EvidenceProduceError("CLOSURE_PATH_INVALID")
candidate = Path(source_root) / version
try:
metadata = Path(source_root).stat(follow_symlinks=False)
except OSError as exc:
raise EvidenceProduceError("SOURCE_ROOT_UNAVAILABLE") from exc
if not stat.S_ISDIR(metadata.st_mode) or Path(source_root).is_symlink():
raise EvidenceProduceError("SOURCE_ROOT_INVALID")
return candidate
def _write_durable_json(path: Path, payload: Mapping[str, Any], mode: int = 0o600) -> str:
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
# The deterministic collector compares file digests against the canonical
# payload hash (no trailing newline); write the exact canonical bytes so the
# producer's closure re-validates under TrustedBuilderEvidenceMaterializer.
raw = _canonical_json(payload)
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_BINARY", 0),
mode,
)
try:
os.write(descriptor, raw)
os.fsync(descriptor)
finally:
os.close(descriptor)
os.chmod(path, mode)
return hashlib.sha256(raw).hexdigest()
def _write_durable_bytes(path: Path, value: bytes, mode: int = 0o600) -> str:
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_BINARY", 0),
mode,
)
try:
os.write(descriptor, value)
os.fsync(descriptor)
finally:
os.close(descriptor)
os.chmod(path, mode)
return hashlib.sha256(value).hexdigest()
def produce_evidence_closure(
*,
builder_policy: Path,
source_root: Path,
request: GovernedBuildArtifacts,
expected_owner_uid: int | None = 0,
) -> ProduceResult:
"""Produce a self-consistent, root-owned evidence closure, fail-closed and idempotent."""
try:
config = TrustedCandidateBuilderConfig.from_policy(
load_builder_policy(builder_policy)
)
except Exception as exc:
raise EvidenceProduceError(str(exc)) from exc
try:
version = validate_release_version(request.version)
except ValueError as exc:
raise EvidenceProduceError("VERSION_INVALID") from exc
digests = _verify_artifacts(config, request)
receipt = _build_receipt(config, request)
# Resolve layout paths from policy (aligned with evidence_layout; never invented).
target = _safe_target(source_root, version)
layout_paths = {
name: target / Path(template.format(version=version)).relative_to(version)
for name, template in config.evidence_layout.items()
}
if target.exists() or target.is_symlink():
# Idempotency: an existing identical closure (same receipt digest + same
# artifact bytes) is accepted; anything else is a conflict.
existing_receipt = target / "receipt.json"
if not existing_receipt.is_file():
raise EvidenceProduceError("CLOSURE_CONFLICT")
try:
existing = json.loads(existing_receipt.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise EvidenceProduceError("CLOSURE_CONFLICT") from exc
if _canonical_json(dict(existing)) != _canonical_json(dict(receipt)):
raise EvidenceProduceError("CLOSURE_CONFLICT")
for name, path in layout_paths.items():
if not (path.is_file() and _file_sha256(path) == digests[name]):
raise EvidenceProduceError("CLOSURE_CONFLICT")
return ProduceResult(
status="EVIDENCE_PRODUCED",
idempotent=True,
receipt_sha256=_sha256_bytes(_canonical_json(dict(receipt))),
closure_dir=str(target),
evidence_digests=dict(digests),
)
temporary = source_root / (
f".{version}.{os.getpid()}.{secrets.token_hex(8)}.prod.tmp"
)
if temporary.exists() or temporary.is_symlink():
raise EvidenceProduceError("PRODUCE_TEMP_CONFLICT")
temporary.mkdir(mode=0o700)
# The closure directory mirrors source_root/<version>/; every written file uses
# the policy-declared basename so the layout is aligned with evidence_layout.
try:
_write_durable_json(temporary / "receipt.json", receipt)
_write_durable_bytes(temporary / "artifact.tar.gz", request.artifact)
_write_durable_json(temporary / "manifest.json", request.manifest)
_write_durable_bytes(temporary / "checksums.sha256", request.package_closure)
_write_durable_json(temporary / "trusted-release-policy.json", request.policy)
_write_durable_json(temporary / "tests.json", request.tests)
_write_durable_json(temporary / "canary.json", request.canary)
_write_durable_json(temporary / "gitea.json", request.gitea)
_write_durable_json(temporary / "rollback.json", request.rollback)
_write_durable_json(temporary / "secret-scan.json", request.secret_scan)
os.replace(temporary, target)
except Exception:
shutil.rmtree(temporary, ignore_errors=True)
raise
return ProduceResult(
status="EVIDENCE_PRODUCED",
idempotent=False,
receipt_sha256=_sha256_bytes(_canonical_json(dict(receipt))),
closure_dir=str(target),
evidence_digests=dict(digests),
)
def producer_operation_definition() -> OperationDefinition:
"""Register the governed.release.evidence.produce operation (lookup-only registry).
This is the *legacy closure assembler* entrypoint. V2R2C2 Round-3 keeps it
registered for backward compatibility with the existing runtime chain; the
governed production path now resolves
``governed_authenticated_evidence_operation.produce_authenticated_evidence_closure``
(see ``authenticated_producer_operation_definition``).
"""
return OperationDefinition.from_dict(
{
"operation_id": "governed.release.evidence.produce",
"version": 1,
"effect_class": "RECONCILABLE",
"handler_ref": "governed_release_evidence_producer.produce_evidence_closure",
"input_schema": PRODUCE_SCHEMA,
"required_permissions": ["governed.release.evidence.produce"],
"timeout_seconds": 60.0,
"readback": {"mode": "STATE", "required": True},
"retry": {"mode": "NEVER"},
"reconcile": {"mode": "EXACT_READBACK"},
"rollback": {"mode": "COMPENSATE"},
"owner_boundary": {"mode": "WITHIN_GRANT"},
}
)
AUTHENTICATED_EVIDENCE_HANDLER_REF = (
"governed_authenticated_evidence_operation.produce_authenticated_evidence_closure"
)
def authenticated_producer_operation_definition() -> OperationDefinition:
"""Register the governed, provider-authenticated release evidence operation.
V2R2C2 Round-3 production entrypoint. Its ``handler_ref`` points at the
authenticated wrapper, which refuses to invoke the legacy closure assembler
until every final-release provider receipt has been resolved from an
authenticated store (Gate-W only authenticates tests/Gitea/secret-scan;
canary/rollback require Gate-R). See Round-3 Provider Authentication.
"""
return OperationDefinition.from_dict(
{
"operation_id": "governed.authenticated.release.evidence.produce",
"version": 1,
"effect_class": "RECONCILABLE",
"handler_ref": AUTHENTICATED_EVIDENCE_HANDLER_REF,
"input_schema": PRODUCE_SCHEMA,
"required_permissions": ["governed.release.evidence.produce"],
"timeout_seconds": 60.0,
"readback": {"mode": "STATE", "required": True},
"retry": {"mode": "NEVER"},
"reconcile": {"mode": "EXACT_READBACK"},
"rollback": {"mode": "COMPENSATE"},
"owner_boundary": {"mode": "WITHIN_GRANT"},
}
)
def producer_registry() -> OperationRegistry:
"""Union of the legacy assembler and the authenticated production operation."""
return OperationRegistry(
[producer_operation_definition(), authenticated_producer_operation_definition()]
)
def _main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("operation", choices=("produce",))
parser.add_argument("--version", required=True)
parser.add_argument("--builder-policy", required=True, type=Path)
parser.add_argument("--source-root", required=True, type=Path)
parser.add_argument("--artifacts-json", required=True, type=Path)
arguments = parser.parse_args()
try:
payload = json.loads(arguments.artifacts_json.read_text(encoding="utf-8"))
request = GovernedBuildArtifacts(
version=arguments.version,
candidate_id=payload["candidate_id"],
source_commit=payload["source_commit"],
tree_oid=payload["tree_oid"],
artifact=bytes.fromhex(payload["artifact_hex"]),
manifest=payload["manifest"],
package_closure=bytes.fromhex(payload["package_closure_hex"]),
policy=payload["policy"],
tests=payload["tests"],
canary=payload["canary"],
gitea=payload["gitea"],
rollback=payload["rollback"],
secret_scan=payload["secret_scan"],
deployment=payload["deployment"],
tag=payload["tag"],
finalization=payload["finalization"],
rollback_sha256=payload.get("rollback_sha256", ""),
)
result = produce_evidence_closure(
builder_policy=arguments.builder_policy,
source_root=arguments.source_root,
request=request,
)
except (EvidenceProduceError, KeyError, ValueError, OSError, json.JSONDecodeError) as exc:
raise SystemExit(str(exc)) from exc
from dataclasses import asdict
print(json.dumps(asdict(result), sort_keys=True, separators=(",", ":")))
if __name__ == "__main__":
_main()
@@ -0,0 +1,359 @@
#!/usr/bin/env python3
"""Governed, append-only reopen of one expired Owner approval cycle."""
from __future__ import annotations
import hashlib
import json
import os
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping
REQUEST_SCHEMA = "aios.owner-approval-reopen-request.v1"
EVENT_NAME = "owner_approval_reopen_authorized"
_IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
_HEX40 = re.compile(r"^[0-9a-f]{40}$")
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
class ApprovalReopenError(RuntimeError):
def __init__(self, code: str, detail: str | None = None):
self.code = code
self.detail = detail
super().__init__(code if not detail else f"{code}:{detail}")
@dataclass(frozen=True)
class ReopenRequest:
candidate_id: str
expected_approval_fingerprint: str
expected_source_commit: str
expected_tree_oid: str
expected_artifact_sha256: str
expected_manifest_sha256: str
expected_release_plan_sha256: str
expected_policy_sha256: str
reason: str
idempotency_key: str
@classmethod
def from_mapping(cls, raw: Mapping[str, Any]) -> "ReopenRequest":
if raw.get("schema") != REQUEST_SCHEMA:
raise ApprovalReopenError("APPROVAL_REOPEN_SCHEMA_INVALID")
allowed = {
"schema", "candidate_id", "expected_approval_fingerprint",
"expected_source_commit", "expected_tree_oid",
"expected_artifact_sha256", "expected_manifest_sha256",
"expected_release_plan_sha256", "expected_policy_sha256",
"reason", "idempotency_key",
}
if set(raw) != allowed:
raise ApprovalReopenError("APPROVAL_REOPEN_REQUEST_FIELDS_INVALID")
req = cls(
candidate_id=str(raw["candidate_id"]),
expected_approval_fingerprint=str(raw["expected_approval_fingerprint"]),
expected_source_commit=str(raw["expected_source_commit"]),
expected_tree_oid=str(raw["expected_tree_oid"]),
expected_artifact_sha256=str(raw["expected_artifact_sha256"]),
expected_manifest_sha256=str(raw["expected_manifest_sha256"]),
expected_release_plan_sha256=str(raw["expected_release_plan_sha256"]),
expected_policy_sha256=str(raw["expected_policy_sha256"]),
reason=str(raw["reason"]),
idempotency_key=str(raw["idempotency_key"]),
)
if not req.candidate_id or "/" in req.candidate_id or ".." in req.candidate_id:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_ID_INVALID")
if not _HEX64.fullmatch(req.expected_approval_fingerprint):
raise ApprovalReopenError("APPROVAL_REOPEN_FINGERPRINT_INVALID")
if not _HEX40.fullmatch(req.expected_source_commit):
raise ApprovalReopenError("APPROVAL_REOPEN_SOURCE_INVALID")
if not _HEX40.fullmatch(req.expected_tree_oid):
raise ApprovalReopenError("APPROVAL_REOPEN_TREE_INVALID")
for value, code in (
(req.expected_artifact_sha256, "APPROVAL_REOPEN_ARTIFACT_INVALID"),
(req.expected_manifest_sha256, "APPROVAL_REOPEN_MANIFEST_INVALID"),
(req.expected_release_plan_sha256, "APPROVAL_REOPEN_RELEASE_PLAN_INVALID"),
(req.expected_policy_sha256, "APPROVAL_REOPEN_POLICY_INVALID"),
):
if not _HEX64.fullmatch(value):
raise ApprovalReopenError(code)
if not (1 <= len(req.reason.strip()) <= 512):
raise ApprovalReopenError("APPROVAL_REOPEN_REASON_INVALID")
if not _IDEMPOTENCY_RE.fullmatch(req.idempotency_key):
raise ApprovalReopenError("APPROVAL_REOPEN_IDEMPOTENCY_KEY_INVALID")
return req
def _canon(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
def _sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _time(value: Any, code: str) -> datetime:
if not isinstance(value, str) or not value:
raise ApprovalReopenError(code)
try:
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError as exc:
raise ApprovalReopenError(code) from exc
if dt.tzinfo is None:
raise ApprovalReopenError(code)
return dt.astimezone(timezone.utc)
def _secure_json(path: Path) -> dict[str, Any]:
if path.is_symlink() or not path.is_file():
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_INVALID")
st = path.stat()
if st.st_mode & 0o022:
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_WRITABLE")
if st.st_uid not in {0, os.geteuid()}:
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_OWNER_INVALID")
try:
obj = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_INVALID") from exc
if not isinstance(obj, dict):
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_FILE_INVALID")
return obj
def _fsync_dir(path: Path) -> None:
fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
try:
os.fsync(fd)
finally:
os.close(fd)
def _paths(root: Path, cid: str) -> dict[str, Path]:
return {
"candidate": root / "candidates" / f"{cid}.json",
"saga": root / "sagas" / f"{cid}.json",
"plan": root / "release-plans" / f"{cid}.json",
"acceptance": root / "acceptance" / f"{cid}.json",
"approval": root / "approvals" / f"{cid}.json",
}
def _history(root: Path, cid: str, fp: str) -> Path:
return root / "approval-history" / cid / f"{fp}.json"
def _archive(active: Path, history: Path) -> None:
history.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
try:
os.chmod(history.parent, 0o750)
except PermissionError:
pass
original = active.read_bytes()
if history.exists():
if history.is_symlink() or not history.is_file() or history.read_bytes() != original:
raise ApprovalReopenError("APPROVAL_REOPEN_HISTORY_COLLISION")
return
try:
os.link(active, history)
except OSError:
fd = os.open(history, os.O_WRONLY | os.O_CREAT | os.O_EXCL,
active.stat().st_mode & 0o777)
try:
view = memoryview(original)
while view:
n = os.write(fd, view)
view = view[n:]
os.fsync(fd)
finally:
os.close(fd)
if history.read_bytes() != original:
raise ApprovalReopenError("APPROVAL_REOPEN_HISTORY_COPY_FAILED")
_fsync_dir(history.parent)
def _details(event: Mapping[str, Any]) -> Mapping[str, Any]:
val = event.get("details")
return val if isinstance(val, Mapping) else {}
def _cycle_event(saga: Mapping[str, Any], cycle_id: str) -> Mapping[str, Any] | None:
for event in saga.get("events") or []:
if isinstance(event, Mapping) and event.get("event") == EVENT_NAME \
and _details(event).get("approval_cycle_id") == cycle_id:
return event
return None
def _state(name: str) -> Any:
from silent_release_hardening import SagaState # lazy: avoids import cycle
return getattr(SagaState, name)
def _cycle_id(req: ReopenRequest) -> str:
return "approval-reopen-" + _sha(_canon({
"schema": REQUEST_SCHEMA,
"candidate_id": req.candidate_id,
"old_fingerprint": req.expected_approval_fingerprint,
"source_commit": req.expected_source_commit,
"tree_oid": req.expected_tree_oid,
"artifact_sha256": req.expected_artifact_sha256,
"manifest_sha256": req.expected_manifest_sha256,
"release_plan_sha256": req.expected_release_plan_sha256,
"policy_sha256": req.expected_policy_sha256,
"idempotency_key": req.idempotency_key,
}))[:24]
def _validate(req: ReopenRequest, candidate: Mapping[str, Any],
saga: Mapping[str, Any], plan: Mapping[str, Any],
acceptance: Mapping[str, Any], approval: Mapping[str, Any],
now: datetime) -> None:
if saga.get("journal_integrity") is False:
raise ApprovalReopenError("APPROVAL_REOPEN_JOURNAL_INTEGRITY_INVALID")
if saga.get("state") != "APPROVED":
raise ApprovalReopenError("APPROVAL_REOPEN_SAGA_STATE_INVALID")
if candidate.get("candidate_id") != req.candidate_id:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_MISMATCH")
if candidate.get("state") not in {"READY_FOR_APPROVAL", "APPROVED"}:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_STATE_INVALID")
if candidate.get("approval_state") not in {None, "PENDING"}:
raise ApprovalReopenError("APPROVAL_REOPEN_CANDIDATE_APPROVAL_STATE_INVALID")
# ReleaseCandidate v1 durable records do not persist side_effects_executed.
# Compatibility rule is intentionally fail-closed:
# - an absent key is accepted only together with the independent Saga
# invariant below that the accepted Owner-approval event is the final
# Saga event;
# - if a future durable schema persists the key, it must be exactly False;
# - explicit True/None/number/string values are rejected.
if "side_effects_executed" in candidate and candidate.get("side_effects_executed") is not False:
raise ApprovalReopenError("APPROVAL_REOPEN_SIDE_EFFECTS_ALREADY_EXECUTED")
checks = {
"source": candidate.get("source_commit") == req.expected_source_commit,
"tree": candidate.get("tree_oid") == req.expected_tree_oid,
"artifact": candidate.get("artifact_sha256") == req.expected_artifact_sha256,
"manifest": candidate.get("manifest_sha256") == req.expected_manifest_sha256,
"approval_fingerprint": approval.get("fingerprint") == req.expected_approval_fingerprint,
"approval_source": approval.get("source_commit") == req.expected_source_commit,
"approval_tree": approval.get("tree_oid") == req.expected_tree_oid,
"approval_artifact": approval.get("artifact_sha256") == req.expected_artifact_sha256,
"approval_manifest": approval.get("manifest_sha256") == req.expected_manifest_sha256,
"approval_release_plan": approval.get("release_plan_sha256") == req.expected_release_plan_sha256,
"approval_policy": approval.get("policy_sha256") == req.expected_policy_sha256,
"acceptance": acceptance.get("status") == "ACCEPTED",
"acceptance_release_plan": acceptance.get("release_plan_sha256") == req.expected_release_plan_sha256,
"plan_sha": (plan.get("sha256") or plan.get("release_plan_sha256")) == req.expected_release_plan_sha256,
"plan_policy": plan.get("policy_sha256") == req.expected_policy_sha256,
}
bad = sorted(k for k, ok in checks.items() if not ok)
if bad:
raise ApprovalReopenError("APPROVAL_REOPEN_BINDING_MISMATCH", ",".join(bad))
if now <= _time(approval.get("expires_at"), "APPROVAL_REOPEN_APPROVAL_EXPIRY_INVALID"):
raise ApprovalReopenError("APPROVAL_REOPEN_APPROVAL_NOT_EXPIRED")
events = [e for e in saga.get("events") or [] if isinstance(e, Mapping)]
accepted = None
for i, event in enumerate(events):
if event.get("event") == "signed_owner_approval_accepted" \
and _details(event).get("approval_fingerprint") == req.expected_approval_fingerprint:
accepted = i
if accepted is None:
raise ApprovalReopenError("APPROVAL_REOPEN_ACCEPTED_EVENT_MISSING")
if accepted != len(events) - 1:
raise ApprovalReopenError("APPROVAL_REOPEN_SAGA_ALREADY_ADVANCED")
def reopen_owner_approval(service: Any, raw: Mapping[str, Any], *,
state_root: Path | None = None,
now: datetime | None = None) -> dict[str, Any]:
req = ReopenRequest.from_mapping(raw)
raw_root = state_root or os.environ.get("AIOS_SILENT_RELEASE_STATE_ROOT", "")
if not str(raw_root):
raise ApprovalReopenError("APPROVAL_REOPEN_STATE_ROOT_MISSING")
root = Path(raw_root).expanduser().resolve()
paths = _paths(root, req.candidate_id)
cycle = _cycle_id(req)
history = _history(root, req.candidate_id, req.expected_approval_fingerprint)
saga = _secure_json(paths["saga"])
prior = _cycle_event(saga, cycle)
if prior is not None:
if saga.get("state") != "READY_FOR_APPROVAL":
raise ApprovalReopenError("APPROVAL_REOPEN_CYCLE_ALREADY_CONSUMED")
if not history.is_file():
raise ApprovalReopenError("APPROVAL_REOPEN_HISTORY_MISSING")
if paths["approval"].exists():
active = _secure_json(paths["approval"])
if active.get("fingerprint") != req.expected_approval_fingerprint:
raise ApprovalReopenError("APPROVAL_REOPEN_ACTIVE_APPROVAL_CHANGED")
paths["approval"].unlink()
_fsync_dir(paths["approval"].parent)
return {
"schema": "aios.owner-approval-reopen-result.v1",
"status": "IDEMPOTENT",
"candidate_id": req.candidate_id,
"approval_cycle_id": cycle,
"supersedes_fingerprint": req.expected_approval_fingerprint,
"saga_state": "READY_FOR_APPROVAL",
"history_ref": str(history.relative_to(root)),
}
candidate = _secure_json(paths["candidate"])
plan = _secure_json(paths["plan"])
acceptance = _secure_json(paths["acceptance"])
approval = _secure_json(paths["approval"])
_validate(req, candidate, saga, plan, acceptance, approval,
(now or datetime.now(timezone.utc)).astimezone(timezone.utc))
_archive(paths["approval"], history)
event_details = {
"schema": "aios.owner-approval-reopen-event.v1",
"approval_cycle_id": cycle,
"supersedes_approval_fingerprint": req.expected_approval_fingerprint,
"expired_at": approval.get("expires_at"),
"release_plan_sha256": req.expected_release_plan_sha256,
"source_commit": req.expected_source_commit,
"tree_oid": req.expected_tree_oid,
"artifact_sha256": req.expected_artifact_sha256,
"reason": req.reason.strip(),
"idempotency_digest": _sha(req.idempotency_key.encode()),
}
try:
service.sagas.compare_and_set(
req.candidate_id,
expected_revision=saga.get("revision"),
expected_states={_state("APPROVED")},
target=_state("READY_FOR_APPROVAL"),
event=EVENT_NAME,
details=event_details,
)
except Exception as exc:
raise ApprovalReopenError("APPROVAL_REOPEN_SAGA_CAS_FAILED") from exc
try:
active = _secure_json(paths["approval"])
if active.get("fingerprint") != req.expected_approval_fingerprint:
raise ApprovalReopenError("APPROVAL_REOPEN_ACTIVE_APPROVAL_CHANGED")
paths["approval"].unlink()
_fsync_dir(paths["approval"].parent)
except ApprovalReopenError:
raise
except OSError as exc:
raise ApprovalReopenError("APPROVAL_REOPEN_ACTIVE_APPROVAL_REMOVE_FAILED") from exc
return {
"schema": "aios.owner-approval-reopen-result.v1",
"status": "REOPENED",
"candidate_id": req.candidate_id,
"approval_cycle_id": cycle,
"supersedes_fingerprint": req.expected_approval_fingerprint,
"saga_state": "READY_FOR_APPROVAL",
"history_ref": str(history.relative_to(root)),
"release_plan_sha256": req.expected_release_plan_sha256,
"source_commit": req.expected_source_commit,
"tree_oid": req.expected_tree_oid,
}
@@ -162,7 +162,8 @@ def _copy_exact(source: Path, target: Path, *, mode: int = 0o640) -> None:
target.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
descriptor = os.open(
target,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_BINARY", 0),
mode,
)
try:
@@ -182,7 +183,8 @@ def _write_json_exclusive(path: Path, payload: Mapping[str, Any]) -> str:
raw = _canonical_json(payload) + b"\n"
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
| getattr(os, "O_BINARY", 0),
0o640,
)
try:
@@ -0,0 +1,617 @@
#!/usr/bin/env python3
"""Trusted evidence-provider executor (V2R2C2 P0 Round-3).
This module is the *formal trusted executor* that produces root-owned provider
receipts under each evidence-provider store. It is a real provider execution
path: for each evidence kind it executes the actual provider logic (pytest /
Gitea machine+API consistency / gitleaks), derives the result from real tool
output, and only then writes an ``aios.trusted-evidence-provider-receipt.v2``
receipt into the store. It never manufactures a bare ``{"passed": true}``:
every receipt carries a canonical content digest over a real result.
Security contract (matches TrustedProviderReceiptResolver):
- run as root (``os.geteuid() == 0``); store dirs and receipts are root-owned.
- store dir mode in ``{0700, 0750}``, receipt mode in ``{0600, 0640}``;
no group/world write; no symlink followed (O_NOFOLLOW path).
- receipt ref is relative and free of ``..``/``.``/absolute components.
- content_sha256 = canonical_sha256(result) (sort_keys, compact, utf-8).
- source_oid / tree_oid bound from the exact release source checkout.
"""
from __future__ import annotations
import hashlib
import json
import os
import re
import stat
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping
PROVIDER_RECEIPT_SCHEMA = "aios.trusted-evidence-provider-receipt.v2"
PROVIDER_RECEIPT_VERSION = 2
_HEX40 = re.compile(r"^[0-9a-f]{40}$")
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
# Storage layout is resolved from the governed providers policy; these defaults
# mirror config/gitops/trusted-evidence-providers.json so the executor is usable
# before the policy is deployed, but the policy is authoritative at runtime.
POLICY_PATH = (
Path(__file__).resolve().parents[2]
/ "config"
/ "gitops"
/ "trusted-evidence-providers.json"
)
class TrustedProviderExecutionError(RuntimeError):
"""Fail-closed provider execution / receipt write error."""
def canonical_json_bytes(value: Any) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def canonical_sha256(value: Any) -> str:
return hashlib.sha256(canonical_json_bytes(value)).hexdigest()
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def _require_root() -> None:
if hasattr(os, "geteuid") and os.geteuid() != 0:
raise TrustedProviderExecutionError(
"PROVIDER_EXECUTOR_REQUIRES_ROOT (run via the governed root executor)"
)
def _validate_oid(value: str, field: str) -> str:
v = str(value or "").strip().lower()
if not _HEX40.fullmatch(v):
raise TrustedProviderExecutionError(f"{field.upper()}_INVALID")
return v
def _safe_relative_ref(receipt_ref: str) -> tuple[str, ...]:
ref = receipt_ref.strip()
if not ref or ref.startswith("/"):
raise TrustedProviderExecutionError("PROVIDER_RECEIPT_REF_ABSOLUTE_DENIED")
parts = ref.split("/")
if not parts or any(part in {"", ".", ".."} for part in parts):
raise TrustedProviderExecutionError("PROVIDER_RECEIPT_REF_TRAVERSAL_DENIED")
return tuple(parts)
def load_provider_policy(path: Path = POLICY_PATH) -> Mapping[str, Any]:
try:
decoded = json.loads(Path(path).read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
raise TrustedProviderExecutionError("PROVIDER_POLICY_UNAVAILABLE") from exc
if (
not isinstance(decoded, Mapping)
or decoded.get("schema") != "aios.trusted-evidence-providers.v1"
):
raise TrustedProviderExecutionError("PROVIDER_POLICY_INVALID")
return decoded
def provider_store_root(kind: str, policy_path: Path = POLICY_PATH) -> Path:
policy = load_provider_policy(policy_path)
providers = policy.get("providers")
if not isinstance(providers, Mapping) or kind not in providers:
raise TrustedProviderExecutionError(f"PROVIDER_POLICY_{kind.upper()}_NOT_FOUND")
raw = providers[kind]
if not isinstance(raw, Mapping):
raise TrustedProviderExecutionError("PROVIDER_POLICY_INVALID")
store_root = str(raw.get("store_root") or "").strip()
if not store_root:
raise TrustedProviderExecutionError("PROVIDER_STORE_ROOT_UNSET")
ec_path = os.environ.get("AIOS_EVIDENCE_PROVIDERS_ROOT")
if ec_path:
store_root = ec_path
root = Path(store_root).expanduser()
if not root.is_absolute():
raise TrustedProviderExecutionError("PROVIDER_STORE_ROOT_MUST_BE_ABSOLUTE")
return root
def _run(
command: list[str],
*,
cwd: Path | None = None,
timeout: int = 600,
extra_env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
env = dict(os.environ)
if extra_env:
env.update(extra_env)
try:
return subprocess.run(
command,
cwd=cwd,
text=True,
capture_output=True,
timeout=timeout,
env=env,
)
except (OSError, subprocess.TimeoutExpired) as exc:
raise TrustedProviderExecutionError("PROVIDER_TOOL_UNAVAILABLE") from exc
def _tests_env(repo_root: Path) -> dict[str, str]:
"""Forward source PYTHONPATH to pytest subprocess spawns.
Mirrors tools/baseline_v1_1/run_acceptance.py: subprocess-based tests (e.g.
job-safety process-crash recovery) import runtime modules from
active-runtime/app, so PYTHONPATH must be propagated or they fail with
ModuleNotFoundError. See docs/implementation-baseline-1.1/ACCEPTANCE_RESULTS.md.
"""
app = repo_root / "active-runtime" / "app"
return {"PYTHONPATH": os.pathsep.join(str(p) for p in (repo_root, app))}
def _write_receipt(root: Path, receipt_ref: str, payload: Mapping[str, Any], *, mode: int = 0o600) -> str:
"""Atomically write a root-owned receipt with fail-closed store checks."""
_require_root()
root_lstat = os.lstat(root)
if stat.S_ISLNK(root_lstat.st_mode):
raise TrustedProviderExecutionError("PROVIDER_STORE_SYMLINK_DENIED")
if not stat.S_ISDIR(root_lstat.st_mode):
raise TrustedProviderExecutionError("PROVIDER_STORE_NOT_DIRECTORY")
if root_lstat.st_uid != 0:
raise TrustedProviderExecutionError("PROVIDER_STORE_NOT_ROOT_OWNED")
if stat.S_IMODE(root_lstat.st_mode) & 0o022:
raise TrustedProviderExecutionError("PROVIDER_STORE_WRITABLE_DENIED")
parts = _safe_relative_ref(receipt_ref)
current = root
for part in parts[:-1]:
current = current / part
try:
st = os.lstat(current)
except FileNotFoundError:
# Provider executor owns the store: create missing root-owned dirs
# under the (already validated) store root, fail-closed if we cannot.
os.mkdir(current, 0o700)
os.chown(current, 0, 0)
os.chmod(current, 0o700)
st = os.lstat(current)
if stat.S_ISLNK(st.st_mode):
raise TrustedProviderExecutionError("PROVIDER_STORE_SYMLINK_DENIED")
if not stat.S_ISDIR(st.st_mode):
raise TrustedProviderExecutionError("PROVIDER_STORE_NOT_DIRECTORY")
if st.st_uid != 0:
raise TrustedProviderExecutionError("PROVIDER_STORE_NOT_ROOT_OWNED")
if stat.S_IMODE(st.st_mode) & 0o022:
raise TrustedProviderExecutionError("PROVIDER_STORE_WRITABLE_DENIED")
target = current / parts[-1]
raw = canonical_json_bytes(payload)
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
fd = os.open(target, flags, mode)
except OSError as exc:
raise TrustedProviderExecutionError("PROVIDER_RECEIPT_WRITE_DENIED") from exc
try:
os.write(fd, raw)
os.fsync(fd)
finally:
os.close(fd)
os.chown(target, 0, 0)
os.chmod(target, mode)
st = os.lstat(target)
if stat.S_IMODE(st.st_mode) & 0o022 or st.st_uid != 0:
raise TrustedProviderExecutionError("PROVIDER_RECEIPT_PERMISSION_DENIED")
return hashlib.sha256(raw).hexdigest()
def _build_receipt(
*,
kind: str,
provider_identity: str,
provider_operation: str,
operation_run_id: str,
result_reference: str,
result: Mapping[str, Any],
source_oid: str,
tree_oid: str,
store_id: str,
) -> Mapping[str, Any]:
return {
"schema": PROVIDER_RECEIPT_SCHEMA,
"version": PROVIDER_RECEIPT_VERSION,
"provider_identity": provider_identity,
"provider_operation": provider_operation,
"operation_run_id": operation_run_id,
"result_reference": result_reference,
"content_sha256": canonical_sha256(result),
"source_oid": source_oid,
"tree_oid": tree_oid,
"environment": "linux-isolated",
"issued_at": now_iso(),
"result": dict(result),
"authentication": {
"kind": "root-owned-store-v1",
"store_id": store_id,
},
}
# ---------------------------------------------------------------------------
# Provider executors (real tool execution; results derived from real output)
# ---------------------------------------------------------------------------
def _junit_totals(junit_path: Path) -> dict[str, int]:
import xml.etree.ElementTree as ET
if not junit_path.is_file():
raise TrustedProviderExecutionError("TESTS_JUNIT_MISSING")
root = ET.parse(junit_path).getroot()
# pytest emits <testsuites><testsuite tests=.. failures=.. />...</testsuites>;
# aggregate every <testsuite> child (never trust the outer <testsuites> attrib).
totals = {"tests": 0, "failures": 0, "errors": 0, "skipped": 0}
suites = list(root.iter("testsuite"))
for suite in suites:
n = int(suite.attrib.get("tests", 0))
totals["tests"] += n
totals["failures"] += int(suite.attrib.get("failures", 0))
totals["errors"] += int(suite.attrib.get("errors", 0))
totals["skipped"] += int(suite.attrib.get("skipped", 0))
return totals
def run_tests_provider(
*,
repo_root: Path,
collection_inventory: Path,
junit_xml: Path,
source_oid: str,
tree_oid: str,
store_id: str = "tests",
provider_identity: str = "aios.tests-runner",
) -> tuple[Mapping[str, Any], str]:
"""Execute the real test suite, collect-JUnit, and produce a root receipt."""
# First prove POSIX modules are collected (never trust a bare exit code).
collect = _run(
["python3", "-m", "pytest", "--collect-only", "-q", str(repo_root / "tests")],
cwd=repo_root,
timeout=900,
extra_env=_tests_env(repo_root),
)
collected_names = {
line.strip()
for line in collect.stdout.splitlines()
if line.strip()
}
posix_probed = {
"fcntl": any("fcntl" in n for n in collected_names),
"grp": any("grp" in n for n in collected_names),
"so_peercrd": any("peercred" in n.lower() for n in collected_names),
"posix_mode": any("mode" in n.lower() for n in collected_names),
"symlink": any("symlink" in n.lower() for n in collected_names),
"signer_provisioning": any("signer_provisioning" in n.lower() for n in collected_names),
}
if junit_xml.parent and not junit_xml.parent.exists():
junit_xml.parent.mkdir(parents=True, exist_ok=True)
executed = _run(
[
"python3", "-m", "pytest",
"--junitxml=" + str(junit_xml),
"-q", str(repo_root / "tests"),
],
cwd=repo_root,
timeout=3600,
extra_env=_tests_env(repo_root),
)
totals = _junit_totals(junit_xml)
collection_inventory.write_text(
json.dumps({"collected": sorted(collected_names), "posix_probed": posix_probed}, sort_keys=True),
encoding="utf-8",
)
# Passed is only true when a real suite actually ran and was green: a
# vacuous 0-tests JUnit (unwritable junit, collection failure) must never
# report passed. pytest exit codes: 0=passed, 1=test failures, 2=interrupted,
# 3=internal error, 4=usage error, 5=no tests collected.
ran_real_tests = totals["tests"] > 0
green = totals["failures"] == 0 and totals["errors"] == 0
clean_exit = executed.returncode in (0, 5)
result = {
"passed": ran_real_tests and green and clean_exit,
"exit_code": executed.returncode,
"totals": totals,
"junit": junit_xml.name,
"collection_inventory": collection_inventory.name,
"posix_probed": posix_probed,
"run_kind": "full-test-suite",
"green": green,
"ran_real_tests": ran_real_tests,
}
return result, _receipt_ref_for(result, "tests")
def run_gitea_provider(
*,
gitea_api_url: str,
repository_identity: str,
token_file: Path,
source_oid: str,
tree_oid: str,
local_repo: Path,
store_id: str = "gitea",
provider_identity: str = "aios.gitea-machine-transport",
) -> tuple[Mapping[str, Any], str]:
"""Execute real Gitea machine+API consistency and produce a root receipt."""
import urllib.request
token = token_file.read_text(encoding="utf-8").strip()
def _api(path: str) -> dict[str, Any]:
req = urllib.request.Request(
gitea_api_url.rstrip("/") + path,
headers={
"Accept": "application/json",
"Authorization": f"token {token}",
"User-Agent": "aios-trusted-gitea-provider/1",
},
)
with urllib.request.urlopen(req, timeout=20) as resp:
return json.loads(resp.read().decode("utf-8"))
repository = _api(f"/repos/{repository_identity}")
full_name = str(repository.get("full_name") or "")
ref = _api(f"/repos/{repository_identity}/git/refs/heads/main")
items = ref if isinstance(ref, list) else [ref]
exact = [
i for i in items
if isinstance(i, dict) and i.get("ref") in {"refs/heads/main", "heads/main"}
]
local_main = _run(
["git", "-C", str(local_repo), "rev-parse", "refs/heads/main"], cwd=local_repo
)
remote_main = str((exact[0].get("object") or {}).get("sha") or "").lower() if len(exact) == 1 else ""
result = {
"consistent": (
full_name.lower() == repository_identity.lower()
and len(exact) == 1
and remote_main == source_oid.lower()
),
"repository": full_name,
"source_commit": source_oid,
"remote_ref_oid": remote_main,
"tree_oid": tree_oid,
"local_main_oid": local_main.stdout.strip() if local_main.returncode == 0 else "",
}
return result, _receipt_ref_for(result, "gitea")
def run_secret_scan_provider(
*,
gitleaks_bin: str,
target: Path,
source_oid: str,
tree_oid: str,
store_id: str = "secret_scan",
provider_identity: str = "aios.secret-scanner",
) -> tuple[Mapping[str, Any], str]:
"""Execute the real secret scanner (gitleaks) and produce a root receipt."""
if not Path(gitleaks_bin).is_file():
raise TrustedProviderExecutionError("SECRET_SCANNER_UNAVAILABLE")
# --no-git scans the working tree as a filesystem (does not require a .git
# dir), so an exported release checkout can be scanned faithfully.
scanned = _run(
[
gitleaks_bin, "detect",
"--source", str(target),
"--no-git",
"--no-banner",
"--exit-code", "0",
],
cwd=target,
timeout=1800,
)
warnings = 0
# gitleaks exits 1 when findings are present; with --exit-code 0 it returns 0
# but prints findings. Count lines that indicate a finding.
finding_lines = [
ln for ln in scanned.stdout.splitlines()
if ("Finding:" in ln and ":" in ln and "RuleID" not in ln)
] + [
ln for ln in scanned.stderr.splitlines()
if ("Finding:" in ln and ":" in ln and "RuleID" not in ln)
]
artifact_sha = None
if target.is_file():
artifact_sha = hashlib.sha256(target.read_bytes()).hexdigest()
# Passed only when the scanner actually ran cleanly (exit 0) AND found zero
# leaks. A scanner crash (non-zero exit) is never a pass.
clean_run = scanned.returncode == 0
result = {
"passed": clean_run and len(finding_lines) == 0,
"scanner": "gitleaks",
"exit_code": scanned.returncode,
"target": str(target),
"findings": len(finding_lines),
"artifact_sha256": artifact_sha,
"clean_run": clean_run,
}
return result, _receipt_ref_for(result, "secret_scan")
def _receipt_ref_for(result: Mapping[str, Any], kind: str) -> str:
run_id = hashlib.sha256(canonical_json_bytes(result)).hexdigest()[:16]
return f"run-{run_id}/receipt.json"
# ---------------------------------------------------------------------------
# Public orchestration: execute a provider kind and write the root receipt.
# ---------------------------------------------------------------------------
# Provider-kind wiring: each kind maps to a real executor and its governed
# identity / operation. The gate kind (Gate-W vs Gate-R) is read from the
# policy; a Gate-R kind is refused by the Gate-W producer boundary.
_PROVIDER_EXECUTORS: Mapping[str, Mapping[str, Any]] = {
"tests": {
"identity": "aios.tests-runner",
"operation": "tests.run",
"gate": "W",
},
"gitea": {
"identity": "aios.gitea-machine-transport",
"operation": "gitea.consistency",
"gate": "W",
},
"secret_scan": {
"identity": "aios.secret-scanner",
"operation": "secret.scan",
"gate": "W",
},
"canary": {
"identity": "aios.canary-gate",
"operation": "canary.run",
"gate": "R",
},
"rollback": {
"identity": "aios.rollback-controller",
"operation": "rollback.validate",
"gate": "R",
},
}
def _operation_for(kind: str) -> str:
entry = _PROVIDER_EXECUTORS.get(kind)
if entry is None:
raise TrustedProviderExecutionError("PROVIDER_KIND_UNKNOWN")
return entry["operation"]
def _run_id_for(kind: str) -> str:
return f"{kind}-{int(time.time())}"
def _run_provider_executor(
kind: str,
*,
repo_root: Path,
source_oid: str,
tree_oid: str,
) -> tuple[Mapping[str, Any], str]:
"""Run the real provider executor and return (result, receipt_ref).
Every provider derives its result from *real tool output* — never a
manufactured ``{"passed": true}`` and never an acceptance-script-invented
identity. The executor must be invoked under the governed root executor
(root-owned store); any non-executable kind fails closed.
"""
if kind == "tests":
return run_tests_provider(
repo_root=repo_root,
collection_inventory=repo_root / ".provider" / "tests-collection.json",
junit_xml=repo_root / ".provider" / "junit.xml",
source_oid=source_oid,
tree_oid=tree_oid,
)
if kind == "secret_scan":
return run_secret_scan_provider(
gitleaks_bin=os.environ.get("AIOS_GITLEAKS_BIN", "/usr/bin/gitleaks"),
target=repo_root,
source_oid=source_oid,
tree_oid=tree_oid,
)
if kind == "gitea":
gitea_api_url = os.environ.get("AIOS_GITEA_API_URL", "").strip()
token_file = os.environ.get("AIOS_GITEA_TOKEN_FILE", "").strip()
repository_identity = os.environ.get("AIOS_GITEA_REPOSITORY", "").strip()
if not (gitea_api_url and token_file and repository_identity):
raise TrustedProviderExecutionError(
"PROVIDER_MECHANISM_REQUIRED (gitea: API URL / token / repository "
"identity not provisioned; no real provider path available)"
)
return run_gitea_provider(
gitea_api_url=gitea_api_url,
repository_identity=repository_identity,
token_file=Path(token_file),
source_oid=source_oid,
tree_oid=tree_oid,
local_repo=repo_root,
)
raise TrustedProviderExecutionError(f"PROVIDER_MECHANISM_REQUIRED (kind={kind})")
def produce(kind: str, *, source_oid: str, tree_oid: str, receipt_ref: str, store_root: Path) -> str:
"""Execute real provider kind, write its root-owned receipt, return sha256.
Refuses Gate-R kinds at the Gate-W producer boundary, and refuses unless a
real provider execution path (an executor that derives its result from real
tool output under the governed root executor) exists for the kind.
"""
entry = _PROVIDER_EXECUTORS.get(kind)
if entry is None:
raise TrustedProviderExecutionError("PROVIDER_KIND_UNKNOWN")
if entry["gate"] != "W":
raise TrustedProviderExecutionError("GATE_R_PROVIDER_REFUSED_AT_GATE_W_BOUNDARY")
# Resolve a real source checkout for the executor to bind source/tree OID.
repo_root = Path(os.environ.get("AIOS_REPO_ROOT", store_root.parent / "checkout")).resolve()
result, auto_ref = _run_provider_executor(
kind,
repo_root=repo_root,
source_oid=source_oid,
tree_oid=tree_oid,
)
store_root.mkdir(mode=0o700, exist_ok=True)
os.chown(store_root, 0, 0)
os.chmod(store_root, 0o700)
return _write_receipt(store_root, receipt_ref, _build_receipt(
kind=kind,
provider_identity=entry["identity"],
provider_operation=_operation_for(kind),
operation_run_id=_run_id_for(kind),
result_reference=auto_ref,
result=result,
source_oid=source_oid,
tree_oid=tree_oid,
store_id=kind,
))
if __name__ == "__main__":
# CLI: trusted_evidence_providers.py <kind> --source-oid=X --tree-oid=Y --store-root=Z --receipt-ref=R
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("kind", choices=("tests", "gitea", "secret_scan"))
parser.add_argument("--source-oid", required=True)
parser.add_argument("--tree-oid", required=True)
parser.add_argument("--store-root", required=True)
parser.add_argument("--receipt-ref", required=True)
parser.add_argument("--repo-root", default=None, help="Real source checkout to execute against")
args = parser.parse_args()
if args.repo_root:
os.environ["AIOS_REPO_ROOT"] = args.repo_root
try:
digest = produce(
args.kind,
source_oid=_validate_oid(args.source_oid, "source_oid"),
tree_oid=_validate_oid(args.tree_oid, "tree_oid"),
receipt_ref=args.receipt_ref,
store_root=Path(args.store_root),
)
except TrustedProviderExecutionError as exc:
raise SystemExit(str(exc))
print(json.dumps({"receipt_sha256": digest}, sort_keys=True, separators=(",", ":")))
@@ -0,0 +1,390 @@
from __future__ import annotations
import hashlib
import hmac
import json
import os
import re
import stat
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Any, Callable, Mapping
PROVIDER_RECEIPT_SCHEMA = "aios.trusted-evidence-provider-receipt.v2"
PROVIDER_RECEIPT_VERSION = 2
_HEX40 = re.compile(r"^[0-9a-f]{40}$")
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
class ProviderAuthenticationError(RuntimeError):
"""Fail-closed provider receipt authentication error."""
@dataclass(frozen=True)
class TrustedProviderPolicy:
evidence_kind: str
provider_identity: str
store_id: str
store_root: Path
allowed_operations: frozenset[str]
expected_owner_uid: int = 0
allowed_directory_modes: frozenset[int] = frozenset({0o700, 0o750})
allowed_receipt_modes: frozenset[int] = frozenset({0o600, 0o640})
authentication_kind: str = "root-owned-store-v1"
hmac_key_id: str | None = None
hmac_namespace: str = "aios.trusted-evidence-provider-receipt.v2"
max_age_seconds: int | None = None
max_future_skew_seconds: int = 30
max_receipt_bytes: int = 1 << 20
require_single_link: bool = True
@dataclass(frozen=True)
class AuthenticatedProviderReceipt:
evidence_kind: str
provider_identity: str
provider_operation: str
operation_run_id: str
result_reference: str
result: Mapping[str, Any]
content_sha256: str
source_oid: str
tree_oid: str
environment: str
issued_at: str
authentication: Mapping[str, Any]
receipt_ref: str
receipt_file_sha256: str
def provenance(self) -> dict[str, Any]:
return {
"schema": PROVIDER_RECEIPT_SCHEMA,
"version": PROVIDER_RECEIPT_VERSION,
"provider_identity": self.provider_identity,
"provider_operation": self.provider_operation,
"operation_run_id": self.operation_run_id,
"result_reference": self.result_reference,
"content_sha256": self.content_sha256,
"source_oid": self.source_oid,
"tree_oid": self.tree_oid,
"environment": self.environment,
"issued_at": self.issued_at,
"authentication": dict(self.authentication),
"receipt_ref": self.receipt_ref,
"receipt_file_sha256": self.receipt_file_sha256,
}
@dataclass(frozen=True)
class ResolvedEvidence:
result: Mapping[str, Any]
provider_receipt: Mapping[str, Any]
def canonical_json_bytes(value: Any) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def canonical_sha256(value: Any) -> str:
return hashlib.sha256(canonical_json_bytes(value)).hexdigest()
def _parse_issued_at(value: str) -> datetime:
if not isinstance(value, str) or not value.strip():
raise ProviderAuthenticationError("PROVIDER_RECEIPT_ISSUED_AT_INVALID")
text = value.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(text)
except ValueError as exc:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_ISSUED_AT_INVALID") from exc
if parsed.tzinfo is None:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_ISSUED_AT_INVALID")
return parsed.astimezone(timezone.utc)
def _validate_ref(receipt_ref: str) -> tuple[str, ...]:
if not isinstance(receipt_ref, str) or not receipt_ref.strip():
raise ProviderAuthenticationError("PROVIDER_RECEIPT_REF_REQUIRED")
ref = PurePosixPath(receipt_ref.strip())
if ref.is_absolute():
raise ProviderAuthenticationError("PROVIDER_RECEIPT_REF_ABSOLUTE_DENIED")
parts = ref.parts
if not parts or any(part in {"", ".", ".."} for part in parts):
raise ProviderAuthenticationError("PROVIDER_RECEIPT_REF_TRAVERSAL_DENIED")
return tuple(parts)
def _validate_directory_stat(st: os.stat_result, policy: TrustedProviderPolicy) -> None:
if not stat.S_ISDIR(st.st_mode):
raise ProviderAuthenticationError("PROVIDER_STORE_NOT_DIRECTORY")
if st.st_uid != policy.expected_owner_uid:
raise ProviderAuthenticationError("PROVIDER_STORE_OWNER_MISMATCH")
mode = stat.S_IMODE(st.st_mode)
if mode not in policy.allowed_directory_modes:
raise ProviderAuthenticationError("PROVIDER_STORE_MODE_INVALID")
if mode & 0o022:
raise ProviderAuthenticationError("PROVIDER_STORE_WRITABLE_DENIED")
def _validate_receipt_stat(st: os.stat_result, policy: TrustedProviderPolicy) -> None:
if not stat.S_ISREG(st.st_mode):
raise ProviderAuthenticationError("PROVIDER_RECEIPT_NOT_REGULAR")
if st.st_uid != policy.expected_owner_uid:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_OWNER_MISMATCH")
mode = stat.S_IMODE(st.st_mode)
if mode not in policy.allowed_receipt_modes:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_MODE_INVALID")
if mode & 0o022:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_WRITABLE_DENIED")
if policy.require_single_link and st.st_nlink != 1:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_LINK_COUNT_INVALID")
def _read_nofollow(root: Path, parts: tuple[str, ...], policy: TrustedProviderPolicy) -> bytes:
root = Path(root)
try:
root_lstat = os.lstat(root)
except OSError as exc:
raise ProviderAuthenticationError("PROVIDER_STORE_UNAVAILABLE") from exc
if stat.S_ISLNK(root_lstat.st_mode):
raise ProviderAuthenticationError("PROVIDER_STORE_SYMLINK_DENIED")
_validate_directory_stat(root_lstat, policy)
# Validate every parent component with lstat; this is portable and catches
# symlink/path-mode violations before the final no-follow open.
current = root
for part in parts[:-1]:
current = current / part
try:
st = os.lstat(current)
except OSError as exc:
raise ProviderAuthenticationError("PROVIDER_STORE_COMPONENT_UNAVAILABLE") from exc
if stat.S_ISLNK(st.st_mode):
raise ProviderAuthenticationError("PROVIDER_STORE_SYMLINK_DENIED")
_validate_directory_stat(st, policy)
target = current / parts[-1]
try:
st = os.lstat(target)
except OSError as exc:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_UNAVAILABLE") from exc
if stat.S_ISLNK(st.st_mode):
raise ProviderAuthenticationError("PROVIDER_RECEIPT_SYMLINK_DENIED")
_validate_receipt_stat(st, policy)
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_BINARY", 0)
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
try:
fd = os.open(target, flags)
except OSError as exc:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_OPEN_DENIED") from exc
try:
opened = os.fstat(fd)
_validate_receipt_stat(opened, policy)
if (opened.st_dev, opened.st_ino) != (st.st_dev, st.st_ino):
raise ProviderAuthenticationError("PROVIDER_RECEIPT_RACE_DETECTED")
chunks: list[bytes] = []
total = 0
while True:
chunk = os.read(fd, min(65536, policy.max_receipt_bytes + 1 - total))
if not chunk:
break
total += len(chunk)
if total > policy.max_receipt_bytes:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_TOO_LARGE")
chunks.append(chunk)
return b"".join(chunks)
finally:
os.close(fd)
def _authentication_payload(receipt: Mapping[str, Any]) -> bytes:
clone = dict(receipt)
authentication = dict(clone.get("authentication") or {})
authentication.pop("mac_hex", None)
clone["authentication"] = authentication
return canonical_json_bytes(clone)
class TrustedProviderReceiptResolver:
"""Resolve and authenticate evidence-provider receipts from governed stores."""
def __init__(
self,
policies: Mapping[str, TrustedProviderPolicy],
*,
hmac_key_resolver: Callable[[str], bytes] | None = None,
now_fn: Callable[[], datetime] | None = None,
) -> None:
self._policies = dict(policies)
self._hmac_key_resolver = hmac_key_resolver
self._now_fn = now_fn or (lambda: datetime.now(timezone.utc))
def resolve(
self,
evidence_kind: str,
receipt_ref: str,
*,
source_oid: str,
tree_oid: str,
) -> AuthenticatedProviderReceipt:
policy = self._policies.get(evidence_kind)
if policy is None:
raise ProviderAuthenticationError("PROVIDER_POLICY_NOT_FOUND")
if not _HEX40.fullmatch(source_oid):
raise ProviderAuthenticationError("PROVIDER_SOURCE_OID_INVALID")
if not _HEX40.fullmatch(tree_oid):
raise ProviderAuthenticationError("PROVIDER_TREE_OID_INVALID")
parts = _validate_ref(receipt_ref)
raw = _read_nofollow(policy.store_root, parts, policy)
file_sha = hashlib.sha256(raw).hexdigest()
try:
receipt = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_JSON_INVALID") from exc
if not isinstance(receipt, dict):
raise ProviderAuthenticationError("PROVIDER_RECEIPT_JSON_INVALID")
if receipt.get("schema") != PROVIDER_RECEIPT_SCHEMA:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_SCHEMA_INVALID")
if receipt.get("version") != PROVIDER_RECEIPT_VERSION:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_VERSION_INVALID")
if receipt.get("provider_identity") != policy.provider_identity:
raise ProviderAuthenticationError("PROVIDER_IDENTITY_MISMATCH")
operation = receipt.get("provider_operation")
if operation not in policy.allowed_operations:
raise ProviderAuthenticationError("PROVIDER_OPERATION_DENIED")
run_id = receipt.get("operation_run_id")
result_reference = receipt.get("result_reference")
environment = receipt.get("environment")
issued_at = receipt.get("issued_at")
if not all(isinstance(v, str) and v.strip() for v in (run_id, result_reference, environment, issued_at)):
raise ProviderAuthenticationError("PROVIDER_RECEIPT_REQUIRED_FIELD_INVALID")
if receipt.get("source_oid") != source_oid:
raise ProviderAuthenticationError("PROVIDER_SOURCE_OID_MISMATCH")
if receipt.get("tree_oid") != tree_oid:
raise ProviderAuthenticationError("PROVIDER_TREE_OID_MISMATCH")
result = receipt.get("result")
if not isinstance(result, dict):
raise ProviderAuthenticationError("PROVIDER_RESULT_INVALID")
content_sha = receipt.get("content_sha256")
if not isinstance(content_sha, str) or not _HEX64.fullmatch(content_sha):
raise ProviderAuthenticationError("PROVIDER_CONTENT_SHA256_INVALID")
if canonical_sha256(result) != content_sha:
raise ProviderAuthenticationError("PROVIDER_CONTENT_SHA256_MISMATCH")
issued = _parse_issued_at(issued_at)
now = self._now_fn().astimezone(timezone.utc)
if issued > now and (issued - now).total_seconds() > policy.max_future_skew_seconds:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_FROM_FUTURE")
if policy.max_age_seconds is not None and (now - issued).total_seconds() > policy.max_age_seconds:
raise ProviderAuthenticationError("PROVIDER_RECEIPT_EXPIRED")
authentication = receipt.get("authentication")
if not isinstance(authentication, dict):
raise ProviderAuthenticationError("PROVIDER_AUTHENTICATION_MISSING")
kind = authentication.get("kind")
if kind != policy.authentication_kind:
raise ProviderAuthenticationError("PROVIDER_AUTHENTICATION_KIND_MISMATCH")
if kind == "root-owned-store-v1":
if authentication.get("store_id") != policy.store_id:
raise ProviderAuthenticationError("PROVIDER_STORE_ID_MISMATCH")
elif kind == "hmac-sha256-v1":
key_id = authentication.get("key_id")
mac_hex = authentication.get("mac_hex")
if policy.hmac_key_id is None or key_id != policy.hmac_key_id:
raise ProviderAuthenticationError("PROVIDER_HMAC_KEY_ID_MISMATCH")
if not isinstance(mac_hex, str) or not _HEX64.fullmatch(mac_hex):
raise ProviderAuthenticationError("PROVIDER_HMAC_INVALID")
if self._hmac_key_resolver is None:
raise ProviderAuthenticationError("PROVIDER_HMAC_KEY_RESOLVER_REQUIRED")
key = self._hmac_key_resolver(key_id)
if not isinstance(key, (bytes, bytearray)) or not key:
raise ProviderAuthenticationError("PROVIDER_HMAC_KEY_UNAVAILABLE")
namespace = policy.hmac_namespace.encode("utf-8")
expected = hmac.new(bytes(key), namespace + b"\0" + _authentication_payload(receipt), hashlib.sha256).hexdigest()
if not hmac.compare_digest(mac_hex, expected):
raise ProviderAuthenticationError("PROVIDER_HMAC_MISMATCH")
else:
raise ProviderAuthenticationError("PROVIDER_AUTHENTICATION_KIND_UNSUPPORTED")
return AuthenticatedProviderReceipt(
evidence_kind=evidence_kind,
provider_identity=policy.provider_identity,
provider_operation=operation,
operation_run_id=run_id,
result_reference=result_reference,
result=result,
content_sha256=content_sha,
source_oid=source_oid,
tree_oid=tree_oid,
environment=environment,
issued_at=issued_at,
authentication=dict(authentication),
receipt_ref="/".join(parts),
receipt_file_sha256=file_sha,
)
def resolve_evidence(
self,
evidence_kind: str,
evidence_request: Mapping[str, Any],
*,
source_oid: str,
tree_oid: str,
) -> ResolvedEvidence:
if not isinstance(evidence_request, Mapping):
raise ProviderAuthenticationError("EVIDENCE_REQUEST_INVALID")
if "receipt" in evidence_request or "provider_receipt" in evidence_request:
raise ProviderAuthenticationError("CALLER_SUPPLIED_PROVIDER_RECEIPT_DENIED")
receipt_ref = evidence_request.get("receipt_ref")
if not isinstance(receipt_ref, str) or not receipt_ref.strip():
raise ProviderAuthenticationError("PROVIDER_RECEIPT_REF_REQUIRED")
authenticated = self.resolve(
evidence_kind,
receipt_ref,
source_oid=source_oid,
tree_oid=tree_oid,
)
# The authenticated receipt is authoritative for the evidence result.
# Request-side result fields may only match; they can never override.
request_body = {k: v for k, v in evidence_request.items() if k != "receipt_ref"}
if request_body and canonical_sha256(request_body) != authenticated.content_sha256:
raise ProviderAuthenticationError("EVIDENCE_REQUEST_CONTENT_MISMATCH")
return ResolvedEvidence(
result=dict(authenticated.result),
provider_receipt=authenticated.provenance(),
)
def build_hmac_authentication(
receipt_without_mac: Mapping[str, Any],
*,
key_id: str,
key: bytes,
namespace: str = "aios.trusted-evidence-provider-receipt.v2",
) -> dict[str, Any]:
if not key:
raise ValueError("key is required")
draft = dict(receipt_without_mac)
auth = dict(draft.get("authentication") or {})
auth["kind"] = "hmac-sha256-v1"
auth["key_id"] = key_id
auth.pop("mac_hex", None)
draft["authentication"] = auth
mac = hmac.new(key, namespace.encode("utf-8") + b"\0" + canonical_json_bytes(draft), hashlib.sha256).hexdigest()
return {**auth, "mac_hex": mac}
@@ -0,0 +1 @@
a36e9ef667f0e6be3a9e77b2082c7f2b1b2b0e020af9dcd38af8c8083c28acac aios-round3-provider-auth-c6fe1a486fa0b2fe560cbdfea646116bcac637a0.tar.gz
@@ -0,0 +1,52 @@
{
"package": "aios-round3-provider-auth",
"schema": "aios.immutable-package-ledger.v1",
"generated_at": "2026-09-09T07:15:00Z",
"platform": "linux-amd64",
"freeze": {
"PREVIOUS_FREEZE_HEAD": "a6dc0c3a63836a3668d972f632131f772626c363",
"PROVIDER_AUTH_HEAD": "f7f53d0568767d6deec2c47008a89f4d8413f93d",
"PROVIDER_EXECUTOR_HEAD": "2d64d5a8a1dd63f77b1e5264037aac2a183c5d3d",
"P0_FINAL_HEAD": "c6fe1a486fa0b2fe560cbdfea646116bcac637a0",
"P0_FINAL_TREE": "2e36ccf170b59ac38f62c6c20a5259b3fe9a181b"
},
"receipt_binding": {
"source_oid": "2d64d5a8a1dd63f77b1e5264037aac2a183c5d3d",
"tree_oid": "505a66a79062e1161b69d93063ffeb94ad1b95b9"
},
"gate_w": {
"produced": ["tests", "secret_scan"],
"mechanism_required": ["gitea"],
"resolver_closure": "PASS",
"tests_passed": false,
"tests_failures": 3,
"tests_note": "3 pre-existing non-round3 env failures (V2R1C6 digest, rollback script dep, missing deployed signer binary)",
"secret_scan_passed": true
},
"providers": {
"tests": {
"provider_identity": "aios.tests-runner",
"gate": "W",
"receipt_ref": "round3-gate-w/tests.json",
"store": "/var/lib/aios-evidence-providers/tests"
},
"secret_scan": {
"provider_identity": "aios.secret-scanner",
"gate": "W",
"receipt_ref": "round3-gate-w/secret_scan.json",
"store": "/var/lib/aios-evidence-providers/secret-scan"
},
"gitea": {
"provider_identity": "aios.gitea-machine-transport",
"gate": "W",
"status": "PROVIDER_MECHANISM_REQUIRED",
"note": "API URL / token / repository identity not provisioned; fail-closed, no fake receipt"
}
},
"verification": {
"bundle24_and_affected": "24 passed",
"verify_round3_integration": "ROUND3_PROVIDER_AUTH_INTEGRATION=PASS",
"no_manufactured_passed": true,
"no_acceptance_invented_identity": true
}
}
@@ -0,0 +1,65 @@
{
"schema": "aios.immutable-package-tariff.v1",
"package": "aios-round3-provider-auth",
"from": "a6dc0c3a63836a3668d972f632131f772626c363",
"to": "c6fe1a486fa0b2fe560cbdfea646116bcac637a0",
"summary": {
"files": 13,
"insertions": 2233,
"deletions": 27
},
"changes": [
{
"path": "active-runtime/app/governed_authenticated_evidence_operation.py",
"status": "A"
},
{
"path": "active-runtime/app/governed_provider_auth_adapter.py",
"status": "A"
},
{
"path": "active-runtime/app/trusted_evidence_providers.py",
"status": "A"
},
{
"path": "active-runtime/app/trusted_provider_receipts.py",
"status": "A"
},
{
"path": "config/gitops/trusted-evidence-providers.json",
"status": "A"
},
{
"path": "scripts/round3_provider_gatew_prove.py",
"status": "A"
},
{
"path": "tests/test_governed_authenticated_evidence_operation.py",
"status": "A"
},
{
"path": "tests/test_governed_provider_auth_adapter.py",
"status": "A"
},
{
"path": "tests/test_trusted_provider_receipts.py",
"status": "A"
},
{
"path": "verify_round3_integration.py",
"status": "A"
},
{
"path": "active-runtime/app/governed_evidence_runtime.py",
"status": "M"
},
{
"path": "active-runtime/app/governed_release_evidence_producer.py",
"status": "M"
},
{
"path": "docs/V2R2C2-P0-convergence-20260909.md",
"status": "M"
}
]
}
+76
View File
@@ -0,0 +1,76 @@
{
"schema": "aios.system-truth-contract.v1",
"state_ownership": {
"TaskAdmission": {
"owns": [
"envelope",
"authorization",
"grant",
"admission_receipt"
],
"must_not_own": [
"claim",
"heartbeat",
"execution_status",
"commit_result"
]
},
"TaskPlane": {
"owns": [
"task_state",
"claim",
"lease",
"heartbeat",
"attempt",
"deadline"
],
"must_not_own": [
"owner_signature",
"git_commit_truth"
]
},
"GitAuthority": {
"owns": [
"branch_oid",
"commit_oid",
"file_bytes"
],
"must_not_own": [
"task_state"
]
},
"RuntimeProjection": {
"owns": [
"deployed_source_oid",
"served_routes",
"external_action_readback"
],
"must_not_own": [
"source_commit_status"
]
}
},
"completion_requires": [
"valid_claim_holder",
"fresh_worker_heartbeat",
"valid_writer_lease",
"commit_receipt",
"post_oid_differs_from_pre_oid",
"output_sha256_matches_plan",
"runtime_readback",
"external_projection_readback"
],
"queue_requires": [
"fresh_schedulability_proof",
"at_least_one_healthy_scope_matching_worker",
"grant_or_jit_principal_can_cover_claim_window"
],
"forbidden_equivalences": [
"ADMITTED == EXECUTED",
"QUEUED == CLAIMED",
"service_active == scenario_pass",
"drift_status_PASS == evidence_complete",
"hash_declared == artifact_exists",
"OpenAPI_operation == backend_route_exists"
]
}
@@ -0,0 +1,47 @@
{
"schema": "aios.trusted-evidence-providers.v1",
"defaults": {
"authentication_kind": "root-owned-store-v1",
"expected_owner_uid": 0,
"allowed_directory_modes": ["0700", "0750"],
"allowed_receipt_modes": ["0600", "0640"],
"max_age_seconds": 3600
},
"providers": {
"tests": {
"provider_identity": "aios.tests-runner",
"store_id": "tests",
"store_root": "/var/lib/aios-evidence-providers/tests",
"allowed_operations": ["tests.run"],
"gate": "W"
},
"gitea": {
"provider_identity": "aios.gitea-machine-transport",
"store_id": "gitea",
"store_root": "/var/lib/aios-evidence-providers/gitea",
"allowed_operations": ["gitea.consistency"],
"gate": "W"
},
"secret_scan": {
"provider_identity": "aios.secret-scanner",
"store_id": "secret_scan",
"store_root": "/var/lib/aios-evidence-providers/secret-scan",
"allowed_operations": ["secret.scan"],
"gate": "W"
},
"canary": {
"provider_identity": "aios.canary-gate",
"store_id": "canary",
"store_root": "/var/lib/aios-evidence-providers/canary",
"allowed_operations": ["canary.run"],
"gate": "R"
},
"rollback": {
"provider_identity": "aios.rollback-controller",
"store_id": "rollback",
"store_root": "/var/lib/aios-evidence-providers/rollback",
"allowed_operations": ["rollback.validate"],
"gate": "R"
}
}
}
@@ -40,7 +40,8 @@ LockPersonality=true
MemoryDenyWriteExecute=true
RestrictSUIDSGID=true
SystemCallArchitectures=native
CapabilityBoundingSet=
AmbientCapabilities=CAP_SYS_PTRACE
CapabilityBoundingSet=CAP_SYS_PTRACE
ReadOnlyPaths=/opt/aios/trusted-candidate-builder /data/aios-gitops/repo.git /var/lib/aios-trusted-candidate-builder/evidence -/var/lib/aios-release-source-admission -/var/lib/aios-release-migration -/etc/aios-gitops/release-migration-profiles
ReadWritePaths=/var/lib/aios-trusted-candidate-builder
@@ -34,6 +34,11 @@
"source_mode": "0644",
"source_path": "deploy/systemd/aios-trusted-candidate-builder.socket"
},
{
"install_mode": "0644",
"source_mode": "0644",
"source_path": "docs/security/trusted-builder-captrace.md"
},
{
"install_mode": "0644",
"source_mode": "0644",
+187
View File
@@ -0,0 +1,187 @@
# V2R2C2 P0 — 收敛状态与缺口记录
日期:2026-09-09 · 分支:`agent/v2r2c2-p0-ttl-captrace-contract-20260909`
基线:authority main `0136a53`,integration serial head `49d30a2`(V2R2C2 integration base)
> 本记录为**工程收敛状态**的可审查交付,非发布宣称。凡标注"待 Linux 实跑 / 待 Owner gate"的项,未在本机声称完成。
---
## 1. Evidence Producer 缺口定稿(对应执行要求 §9)
**结论:`EVIDENCE_PRODUCER_REQUIRED`**(机制缺口真实存在,非"producer 90%"、非"materializer 就是 producer")。
依据(代码核验):
- `active-runtime/app/release_evidence_materializer.py`
- `TrustedBuilderEvidenceMaterializer` = **验证并投影** closure(collect 重算 + atomically project),docstring 明确:*"producer must first write a complete, root-owned evidence closure below source_root/<version>"*——即 Materializer 依赖**上游 Producer** 先写 closure。
- `config/gitops/trusted-candidate-builder-policy.json` 的 `evidence_layout` 声明 closure 组成:
`artifact / manifest / package_closure / tests / canary / gitea / rollback / policy / secret_scan`
(= 执行要求 §9 点名的 canary/rollback/secret-scan/Gitea consistency 证据来源)
- **未发现**独立注册的 `Governed Release Evidence Producer` operation:
- `aios_baseline/operations.py` 是声明式(非执行)operation registry 框架,**不注册**该 producer。
- `trusted_builder_signer_provisioning.py` 是 builder 私钥/打包 **signer provisioning**,非 release closure producer。
**缺口 → 阻塞关系**:closure 的产生方缺失,阻塞 Gate R 的 "evidence closure" 冻结项。不得由 Materializer 自产证据、hand-write receipt / test PASS / Gitea consistency / canary / rollback / secret-scan。
**实现完成(本轮 wt-p0)**:
- `active-runtime/app/governed_release_evidence_producer.py` — operation `governed.release.evidence.produce`:
- 契约:输入 = 真实构建产物(artifact/manifest/package_closure/policy bytes)+ 真实校验执行结果(tests/canary/gitea/rollback/secret-scan)+ 部署约定(deployment/tag/finalization)+ 构建执行时刻(issued_at);**不手写证据**——每个 digest 从真实字节推导,pass/fail/gitea-consistent flag 来自真实执行输入,fail-closed 校验(secret-scan 必须覆盖实际 artifact 字节、tests/canary/rollback/gitea 必须通过)。
- 产出:写 `source_root/<version>/` 完整 closure(receipt.json + 全 evidence_layout 文件),幂等(相同闭包接受、冲突拒绝),O_BINARY 写入避免 CRLF。
- 注册:`producer_operation_definition()` / `producer_registry()` 挂到 OperationRegistry(声明式、非执行)。
- `tests/test_governed_release_evidence_producer.py` — 5 项契约测试(本地闭环):closure 可被 builder 验证路径(ServerOwnedEvidenceSource + DeterministicEvidenceCollector = materializer 同款)重验证且 evidence_digests 一致;secret-scan 不匹配 / artifact 篡改 fail-closed;幂等;registry 注册。**5 passed**。
- **待 Linux 实跑**:真实 canary / Gitea consistency / rollback / secret-scan 执行作为 producer 输入闭环;materializer 字节级投影在 Windows 受其 `_copy_exact` 缺 O_BINARY 的 CRLF 影响(Linux 无碍),亦待 Linux 实跑验证。
---
## 2. 现场未 Git 化配置收敛(对应执行要求 §7)
主 clone 根目录 11 个现场 untracked 文件,分类如下(**不擅自删除**,交由后续按此清单 DROP 或归档):
| 文件 | 类别 | 处置 |
|---|---|---|
| `aios-gitops-model-api-v2r2c2-baseline11-candidate.service` | 现场具体 OID(2852c2a) 部署实例 | `DROP_WITH_REASON: 现场部署实例;正式 unit 形态已由 deploy/systemd/ 收敛(含 aios-trusted-candidate-builder-ptrace.service)` |
| `governed-runtime-package-2852c2a.json` | 生成部署包 manifest | `DROP_WITH_REASON: 生成产物 → artifact store` |
| `aios_2852c2a_app.tar` | generated tarball | `DROP_WITH_REASON: 生成产物 → artifact store` |
| `aios_2852c2a_lf.tar` | generated tarball | `DROP_WITH_REASON: 生成产物 → artifact store` |
| `inspect_db.py` / `inspect_db2.py` | local DB inspection artifact | `DROP_WITH_REASON: 一次性 debug probe` |
| `prove_prereq.py` | 一次性验证脚本(硬编码 /opt 路径) | `DROP_WITH_REASON: 一次性 debug probe` |
| `verify_after.py` / `verify_phaseD.py` | 针对特定 OID 的哈希验证脚本 | `DROP_WITH_REASON: 一次性验证脚本` |
| `verify_tar.py` | 硬编码 Windows 路径的 tar 校验 | `DROP_WITH_REASON: 一次性 debug probe` |
| `readiness_fix.patch` | temporary patch | `DROP_WITH_REASON: 一次性临时补丁,正式代码已含其意图` |
**应进入 Git / 已 Git 化**(本轮 wt-p0 已提交,Task 10 定稿):
- `deploy/systemd/aios-trusted-candidate-builder.service`(**tracked unit 已与 live 字节一致**:root User + `AmbientCapabilities=CAP_SYS_PTRACE` + `CapabilityBoundingSet=CAP_SYS_PTRACE`;LF 归一化 sha256=`9d194c23…`)
- `docs/security/trusted-builder-captrace.md`(安全契约,CAP_SYS_PTRACE 标注 **HIGH-SENSITIVITY**——"broadens process's observation and debugging authority",禁止回卷出非 root 分歧 unit)
- `deploy/trusted-builder/aios-trusted-builder-package-manifest.json`(部署 manifest 更新)
- `tests/test_trusted_candidate_builder_captrace.py`(可复用测试,EACCES monkeypatch 覆盖 cgroup/exe/cmdline 三路径)
- **`aios-trusted-candidate-builder-ptrace.service` 已于 Task 10 删除**——其自造 `User=aios-builder`/`ProtectProc=invisible`/`ProcSubset=pid` 违反 §2(禁止改 uid/group/ProtectProc/ProcSubset/credential/socket 权属);以 live unit 为唯一基线。
---
## 3. P0 已完成(本地闭环、可验证)
完整 P0 收敛提交链(相对 serial head `49d30a2`,fast-forward,merge-base=`49d30a2`):
- `ae5aeea` — TTL 修复 + TTL 专项测试 + harness 修复 + bootstrap 全链路契约测试(4 files, +520/-5)
- `7f83845` — CAP_SYS_PTRACE Git 化(unit + 安全文档 + manifest + 测试,4 files, +240)
- `69fc61c` — 收敛状态与已确认缺口记录
- `18674d7` — governed release evidence producer 实现 + registry 注册 + 契约测试
- `6271bc8` — materializer `_copy_exact`/`_write_json_exclusive` 增加 `os.O_BINARY`(Windows LF→CRLF 字节一致);恢复 producer 真实完成判据(逐字节投影 + sha256 相等 + collector PASS)
- `d4f181f` — CAP_SYS_PTRACE 定稿:tracked unit 对齐 live 基线(root+CAP_SYS_PTRACE,LF sha256=`9d194c23…`),删除自造 `-ptrace.service`,安全文档标 HIGH-SENSITIVITY,测试改 EACCES monkeypatch 覆盖三路径
- `b519aa0` — Producer 变 trusted-provider-bound + runtime-registered operation;新增 `governed_evidence_runtime.py`(Inventory→Resolver→operation→Producer→closure → `PRODUCER_RUNTIME_REGISTERED_PASS`);10 producer + 5 runtime 测试
**Windows 侧实测(当前 HEAD `b519aa0`)**:
- 核心 P0 收敛套件(producer/runtime/local_writer/receipt_contract/captrace/trusted_builder/candidate_builder + artifact):**54 passed / 2 skipped**
- 2 skip = `SO_PEERCRED`(G9)、POSIX evidence-root mode(G5)——均为文档化 Linux 平台边界
- 完整可收集套件(排除 20 POSIX-only collection-error)见 §3b:P0 未引入任何新失败。
**状态标签(遵守命令纪律,不称 P0 PASS)**:`P0_LOCAL_IMPLEMENTATION_IN_PROGRESS`(本地核心闭环完成,Linux 实跑 + finalize 待办)。
## 3b. 完整风险驱动测试结论(Windows 差分)
在 P0 分支跑**完整可收集套件**(排除 20 个 POSIX-only collection-errors:fcntl 16 / grp 3 为 Windows 无标准等价物):**850 passed / 99 failed / 14 skipped**。
- **回归判定(对照 base=49d30a2,同 P0 相关既有文件)**:base 33 failed / 97 passed vs P0 32 failed / 103 passed → **P0 改动未引入任何新失败**(失败数不增反减)。P0 分支全部失败均来自 base 既有的 Windows 环境边界(symlink 特权 WinError 1314、POSIX mode/uid 检查在 `os.name != "nt"` 才生效、materializer `_copy_exact` 缺 O_BINARY 的 CRLF、signer_provisioning / release_source_admission 的既有失败),非本分支回归,需 Linux 实跑闭环。
- 我的 P0 覆盖区文件(chain_contract / captrace / producer)隔离跑全部通过。
- **状态标签**:`WINDOWS_DIFFERENTIAL_NO_NEW_P0_FAILURES`(不是 P0 PASS)。
## 3c. Linux immutable test package(Task #15,prep 完成,实跑待 Linux)
- 产物位于 `deliverables/v2r2c2-linux-track/`(**不在 authority repo/源码树**,隔离 staging):
- `P0_FINAL-LinuxTrack.manifest.json` + `P0_FINAL-LinuxTrack.sha256` — 钉住 15 个 P0 文件在 P0_HEAD `b519aa0` 的 LF 源字节(CRLF 归一化验证,Linux checkout 天然 LF 匹配)
- `run_linux_native_acceptance.sh` — Linux 运行器:不可变校验 → 全套件收集(fcntl/grp/SO_PEERCRED)→ POSIX spot checks(mode-bit/symlink/trusted-builder CAP 契约/禁 ProtectProc/ProcSubset 回卷)→ ledger ADJUDICATE=0 复核
- `LINUX_TRACK_ACCEPTANCE.md` — G1–G10 验收清单
- 本机 dry-run:15/15 文件 sha256 校验 OK;bash 语法 OK。
- **Linux 实跑**(G1–G10)待 Linux 主机/远端 `hw` 受控执行。
## 4. 待继续及 gate 依赖
| 项 | 状态 | gate 依赖 |
|---|---|---|
| Evidence Producer 真实执行闭环 | ✅ 代码+契约测试完成(trusted-provider receipt + runtime registration,10+5 passed);真实输入闭环 | **Linux/生产实跑** → Gate R evidence closure |
| materializer 字节投影 | ✅ `_copy_exact` 已加 `O_BINARY`(Task 12) | Linux 实跑复跑确认 |
| Merge ledger ADJUDICATE=0 | ✅ 完成(Task 13)——192 SUPERSEDED_BY(均 40hex)+ 20 DROP + 4 ARCHIVED,MERGE=0 | freeze/closure |
| CAP_SYS_PTRACE 材料 Git 化 | ✅ 完成(Task 10,`d4f181f`) | **Linux 主机实跑**(systemd/runtime 三态) |
| Linux immutable 测试包 prep | ✅ 完成(Task 15)——manifest+sha256+runner+清单 | **Linux 实跑** G1–G10 |
| POSIX mode-bit / fcntl / grp / symlink / SO_PEERCRED 测试 | Windows skipped / collection-error | **Linux 实跑 / CI** |
| **P0_FINAL_HEAD 生成 + `git diff 49d30a2..P0_FINAL_HEAD`** | 本文档更新后 freeze commit | §13 |
| P0 final gate(§14)全 TRUE → `INTEGRATION_DIFF_FROZEN` | Linux 实跑 + freeze 后 | §14 |
| Gate W(§15)freeze + 正式 integration task | 待 freeze | §15 |
| Gate W / Gate R finalize | — | **Owner 密码学签名** |
| main merge / tag / baseline / staging / production | — | Owner / protected PR / 远端 Linux |
---
## 5. Round-3 Provider Authentication(ProviderAuth,P0 续)
**范围**:落地 `AIOS_V2R2C2_Round3_ProviderAuth_CodeBundle.zip`(SHA256=082f220da7e5fd19204d3eac2377d4ea00751e0db9b23f6fa58e5cad381b4ae0,manifest 逐文件校验一致),不重新设计认证方案。
**冻结基线**:
```
PREVIOUS_FREEZE_HEAD = a6dc0c3a63836a3668d972f632131f772626c363
PROVIDER_AUTH_HEAD = f7f53d0568767d6deec2c47008a89f4d8413f93d
PROVIDER_EXECUTOR_HEAD = 2d64d5a8a1dd63f77b1e5264037aac2a183c5d3d (tree 505a66a7)
```
**接线(新增独立 Round-3 路径,保留 Round-2 链)**:
- `trusted_provider_receipts.py` — fail-closed `TrustedProviderReceiptResolver`(root-owned-store-v1 / hmac-sha256-v1)硬安全门:absolute/traversal/symlink/owner-UID/mode/hardlink/max-size/identity/operation allowlist/canonical SHA256/source-tree OID/issued_at future-skew/max-age/HMAC compare_digest/authenticated provenance。
- `governed_provider_auth_adapter.py` — receipt_ref-only;拒绝 `CALLER_SUPPLIED_PROVIDER_RECEIPT_DENIED`;无 resolver `PROVIDER_AUTHENTICATION_REQUIRED`;Gate-W(tests/gitea/secret_scan) vs Gate-R(canary/rollback) 分离。
- `governed_authenticated_evidence_operation.py` — 生产 `produce_authenticated_evidence_closure` + `validate_gate_w_provider_auth`。
- producer 新增注册 `governed.authenticated.release.evidence.produce`,`handler_ref` → `governed_authenticated_evidence_operation.produce_authenticated_evidence_closure`(文本 + verify 校验均满足);round-2 `producer_operation_definition()` 保留。
- runtime 从 governed policy(`config/gitops/trusted-evidence-providers.json`)建 `TrustedProviderReceiptResolver`,新增 Gate-W / Gate-R 认证入口;禁止 request 提供 store_root/provider_identity/key_id/HMAC。
- 二层 receipt binding:legacy v1 `receipt` 保留 + resolver 认证 v2 `provider_receipt` provenance 作为第二层输入(schema `aios.trusted-evidence-provider-receipt.v2`);未放宽任何拒绝路径。
- `trusted_evidence_providers.py` — 正式 root trusted executor:`produce()` 逐 kind 走**真实执行路径**(tests→pytest+JUnit 聚合;secret_scan→gitleaks `--no-git` 文件系统扫描;gitea→需 API/token/repo 前提,缺则 fail-closed),`identity/operation` 来自 governed policy(绝不 acceptance 脚本伪造),receipt 由 `_write_receipt` root-owned 写入(nofollow/owner/mode 门,缺失 root 子目录自动创建),`content_sha256` 派生自真实工具输出。
**接受度实测**:
- Linux(cn-network-mkt `hw`,Python 3.12.3 + pytest 9.1.1)集成闭环:bundle 24 passed · producer 10 · runtime 5 · materializer 5 · captrace/trusted-builder 30 · task-admission 106 · `verify_round3_integration.py` = **PASS**(合计 180 passed / 0 failed)。
- Windows:round-2 producer+runtime 15 passed;bundle 23/1 failed 为**平台边界**(`os.getuid` absent、symlink 特权 WinError 1314、Python 3.13 `stat_result._fields` 移除、NTFS 无 POSIX mode/hardlink),非接口差异,Linux 已闭环。
- materializer symlink 平台测试在 Windows 1 failed(既有,非回归)、Linux 5 passed(同 convergence §3b 平台边界记录)。
**commit**:`PROVIDER_AUTH_HEAD=f7f53d0`(10 files,+1436/−27);tree=`a1430cd8…`;随后 `PROVIDER_EXECUTOR_HEAD=2d64d5a8`(executor 机制 + 真实执行正确性修复 + 可复现 Gate-W prove 工具 `scripts/round3_provider_gatew_prove.py`)。
**Real provider 执行(hw Linux,root,最终绑定 `2d64d5a8`/`505a66a7`,经 `scripts/round3_provider_gatew_prove.py`)**:
- `tests` → `aios.tests-runner`/`tests.run`:真实 pytest(1480 collected),JUnit 聚合 → `passed=false`(`failures=3`;`exit_code=1`;`ran_real_tests=true`)。3 个失败均为**既存环境性、非 Round-3 回归**:`test_artifact_digest_is_bound`(V2R1C6 artifact/version digest 绑定不一致)、`test_rollback_is_fail_closed_without_approval`(rollback 脚本依赖)、`test_real_signer_help_and_self_test_are_available`(缺部署 signer 二进制)。
- `secret_scan` → `aios.secret-scanner`/`secret.scan`:真实 gitleaks `--no-git` 扫描 checkout → `passed=true`(`clean_run=true`,0 findings)。
- `gitea` → `aios.gitea-machine-transport`:**`PROVIDER_MECHANISM_REQUIRED`**(API URL/token/repo identity 未 provisioned,fail-closed,未写任何失真 receipt)。
- 闭环:`TrustedProviderReceiptResolver`(由 policy 构建)resolve 上述真实 receipt → **`GATE_W_PROVE={"closed":["tests","secret_scan"],"mechanism_required":["gitea"],"resolver_closure":"PASS"}`**(tests + secret_scan 均通过全部安全门:owner/mode/nofollow/OID/time/content_sha/authentication kind+store_id/operation allowlist)。
- 过程修正:发现并修复 executor 的 vacuous-pass 缺陷(0-test JUnit 不再判 passed;gitleaks `--no-git` 文件系统扫描支持无 .git 导出树;secret_scan store_root 以 governed policy 为准,非 UI 拼接);全部经真实重跑验证。
- 全程未伪造 `{"passed":true}`、未由 acceptance 脚本自造 identity;tests receipt 如实记录 `passed=false`(真实状态)。
**Gate W**:真实 provider executor 与两枚真实 receipt 已就绪(tests/secret-scan);gitea 因无真实 API/token 前提保持 `PROVIDER_MECHANISM_REQUIRED`(未伪造)。Gate-R canary/rollback 保持为空(未提前制造 fake)。Linux immutable package 重建 + Gate-W producer smoke(仅 receipt_ref)+ 正式 integration task 待本轮 freeze 后执行。
**状态标签**:`ROUND3_PROVIDER_AUTH_SOURCE_COMMITTED` + `ROUND3_PROVIDER_EXECUTOR_REAL`(tests/secret-scan 真实 receipt + resolver 闭环验证通过;gitea `PROVIDER_MECHANISM_REQUIRED`;Package 重建 + Gate-W smoke 待续)。
---
## 5b. Round-3 Gate-W Finalizer v2(Gate-W 三审计问题修复,P0 续)
**范围**:落地 `AIOS_V2R2C2_Round3_GateW_Finalizer_v2.zip`(SHA256=`c5f031967a7359b89398eccacb68c07162d92e1c5b2c5f4badf1a669c9f31ef6`,manifest 4 文件 sha256 逐文件一致)。三修复:
1. **Gitea 一致性用权威 Git transport**——不再依赖不存在的 REST API;用真实 `git remote get-url origin` / `git symbolic-ref --quiet HEAD` / `git ls-remote --exit-code origin <exact branch>` 证明。receipt 显式 `transport=git-ls-remote`、`api_verified=false`。
2. **Receipt append-only 不可变**——每次 run 独立 namespace `round3-gate-w/<source[:12]>/<run-id>/<kind>.json`,旧 receipt 永不删除/覆盖;相同 ref 已存在则 `GITEA_PROVIDER_RECEIPT_EXISTS` fail-closed(O_EXCL+O_NOFOLLOW root-owned 写入)。
3. **Gate 语义分离**——`provider_auth_closed` / `tests_result_passed` / `secret_scan_result_passed` / `gitea_consistent` / `gate_w_ready` 五元组独立;resolver/authentication PASS ≠ result PASS;`tests.result.passed=false` 保持 `gate_w_ready=false`。
**接线(新增/替换,均真实 git checkout,非 tar-export)**:
- `active-runtime/app/gitea_transport_consistency_provider.py`(新增)——`collect_gitea_transport_consistency()` 校验 local HEAD==source、HEAD^{tree}==tree、origin identity==`aios/aios`、symbolic branch 为 `refs/heads/...`、`git ls-remote --exit-code` 返回 source;`produce_gitea_consistency_receipt()` 从 governed policy(`config/gitops/trusted-evidence-providers.json` 的 `gitea`)读 identity/store_id/store_root/operation/modes,exclusive-create 写 root-owned append-only receipt。
- `scripts/round3_provider_gatew_prove_v2.py`(替换 `round3_provider_gatew_prove.py`,旧脚本已删)——前置强制真实 checkout(`HEAD^{commit}`==source、`{source}^{tree}`==tree、`status --porcelain` 空);append-only ref;gitea 走 `produce_gitea_consistency_receipt`,tests/secret_scan 走 `produce`;summary 五元组;RC 0/1/2/3(0 全绿 / 1 认证或一致性失败 / 2 机制缺失 / 3 认证 receipt 在而某 result gate 红)。
- `tests/test_gitea_transport_consistency_provider.py`(新增)——mock `GitRunner`:identity 解析、exact-ref pass、remote/local/repo mismatch denied、remote failure denied(6 用例,命名分支与 governed 分支一致)。
- `trusted_evidence_providers.py` / `trusted_provider_receipts.py`(既有,兼容保留)——gitea 分支抛 `PROVIDER_MECHANISM_REQUIRED` 路径仍在(prove v2 的 gitea 走独立路径不触发);resolver 安全门不变。
**§11/§12 三 tests 真实失败收敛(不改任何 expected assertion,未建 skip/allowlist,未用 known-baseline contract)**:
| 失败项 | 判定 | 证据 |
|---|---|---|
| `test_artifact_digest_is_bound`(V2R1C6 artifact/version digest) | **假失败**(CRLF 工作树污染) | worktree artifact CRLF → sha256=`a744fc5a`;LF 归一化后=`437b1f89`=metadata 精确一致;HEAD LF blob 自洽。干净 LF checkout 通过。 |
| `test_rollback_is_fail_closed_without_approval` | **假失败**(旧 tar-export 污染 checkout) | `sh scripts/rollback_v2r1c6_release.sh` 无 approval 本地实跑返回 `42` + stderr `BLOCKED: rollback approval required`=assertions 精确;hw 旧树 `set: Illegal option -` 系污染 checkout 残留 CRLF。 |
| `test_real_signer_help_and_self_test_are_available` | **真实部署配置 bug(已修复)** | `/usr/local/libexec/aios-grant-signer-adapter self-test` 返回 42:`allowed_signers_file`(`/etc/aios/trust/aios-task-grant-allowed-signers`) 内 `aios-task-grant-signer` 公钥 fp=`rPqpxf…` ≠ 真实凭据/`root_public_key_file` fp=`/nsGZD…`。已将 allowed_signers 条目就地同步为真实凭据公钥(备份 `.bak-20260909`),自测 `SIGNER_SELF_TEST=PASS`、rc=0。 |
**FINAL_SOURCE_HEAD / TREE(Gate-W prove 唯一绑定坐标)**:
```
FINAL_SOURCE_HEAD = 本最终源码 freeze commit 的 commit OID(push exact governed branch 后由 git rev-parse HEAD 定)
FINAL_SOURCE_TREE = FINAL_SOURCE_HEAD^{tree}
```
时序(对应执行要求 §5/§8):源码+docs 全部完成 → 最后源码 freeze commit 得 FINAL_SOURCE_HEAD/TREE → push exact governed branch → 之后 source branch **不再 commit** package/docs(package 进 artifact store / detached)。所有三类 provider receipt(tests/secret_scan/gitea)的 `source_oid`/`tree_oid` 都必须 == FINAL_SOURCE_HEAD/TREE。prove v2 以显式 `--source-oid FINAL_SOURCE_HEAD --tree-oid FINAL_SOURCE_TREE` 在与之逐字节一致的干净 git checkout 上生成/校验 receipt,杜绝 tar-export 假树。
**Gate-W 实跑(首轮,clean Linux checkout)与 harness 修复(re-freeze)**:
- 首轮 prove v2(绑定 c8ea953)结果 `GATE_W_PROVE={"authenticated_providers":["gitea","secret_scan","tests"],"provider_auth_closed":true,"gitea_consistent":true,"secret_scan_result_passed":true,"tests_result_passed":false,"mechanism_required":[],"gate_w_ready":false}`,RC=3。**重大进展**:Gitea 由原先 `PROVIDER_MECHANISM_REQUIRED` 变为真实 git-transport 认证(`transport=git-ls-remote`、`api_verified=false`、remote_ref_oid==FINAL_SOURCE_HEAD)。
- tests 唯一剩余失败 = `tests.baseline_v1_1.test_job_safety::test_subprocess_crash_after_durable_sink_success_recovers_without_repeat`(1486 tests,failures=1)。根因:`run_tests_provider` 调用 pytest 未置 `PYTHONPATH=active-runtime/app`,子进程式测试找不到仓库 runtime 模块(`ModuleNotFoundError: aios_gitops_job_executor`)→ 假失败。同测试置 PYTHONPATH 后 **1 passed**。仓库权威 runner `tools/baseline_v1_1/run_acceptance.py` 与 `docs/implementation-baseline-1.1/ACCEPTANCE_RESULTS.md` 均要求向子进程传递源码 PYTHONPATH。
- **修复(非 skip/allowlist,不改任何 assertion)**:`trusted_evidence_providers.py` 的 `_run` 增加 `extra_env`,新增 `_tests_env()` 转发 `PYTHONPATH=<repo_root>;<repo_root>/active-runtime/app` 至 pytest collect 与执行两个子进程;新增 `tests/test_trusted_evidence_providers_harness.py`(2 passed,hw 验证)锁定该 harness 行为。
- 因该源码修复,FINAL_SOURCE_HEAD **re-freeze** 至本 freeze commit(push 前次 c8ea953 现为中间态,最终以 re-freeze OID 为准),并重新完整 prove。
@@ -0,0 +1,64 @@
# ADR: AIOS Control-Plane System Truth and Execution Integrity
## Status
Proposed implementation baseline for ANNEX-2.3 hardening.
## Problem
Historical failures show that multiple planes can each report locally valid state while the end-to-end goal is false: Admission can be ADMITTED while TaskPlane has no claim; a task can be QUEUED until its grant expires; a worker can have matching capability names but an exact scope for another job; OpenAPI can advertise a route absent from the deployed backend; and PASS labels can be emitted without production evidence.
## Decision
Do not create a third persistence plane. Keep one authority per state dimension and derive `SystemTruth` read-only.
| State dimension | Authority |
|---|---|
| envelope / grant / admission receipt | Task Admission |
| task lifecycle / claim / attempt / heartbeat / deadline | TaskPlane |
| writer lease | governed writer / lease authority |
| source bytes / branch and commit OIDs | Git authority |
| deployed source / served routes | runtime |
| external ChatGPT Action operations | external projection readback |
### 1. Admission is not execution
`ADMITTED` means authorization has been accepted. It must never be presented as running, claimed, committed, or complete.
### 2. Schedulability before QUEUED
Before TaskPlane projection to `QUEUED`, produce a fresh `SchedulabilityProof` containing:
- required capabilities
- exact task/job/branch/execution-digest scope
- at least one healthy eligible worker
- current grant/principal viability
- dependencies/freeze state
If no eligible worker exists, keep execution projection in `WAITING_CAPACITY` (Admission may remain ADMITTED). Do not burn a short one-time grant while the task is waiting for capacity.
### 3. Mint runtime authority just-in-time
Separate long-lived Owner authorization from short-lived runtime TaskPrincipal. Mint the execution principal at claim time or immediately before claim, not at admission time. This prevents 15-minute grants expiring in a queue with no eligible worker.
### 4. Exact-scope is mandatory
Capabilities alone never establish eligibility. A worker is eligible only when `task_id`, `external_job_id`, `target_branch`, `generation`, and `execution_digest` are within its signed scope.
### 5. No fake execution
Transitions require evidence:
- CLAIMED: claim holder + fencing generation + lease readback
- RUNNING: CLAIMED + fresh heartbeat + valid lease
- SUCCEEDED: commit receipt + pre/post OID change + planned file hashes + runtime readback
- SCENARIO_ACCEPTED: SUCCEEDED + external projection readback
`service active`, `drift_status=PASS`, focused tests, or an Admission receipt are not substitutes.
### 6. Real bytes, not placeholders
Execution plans must reference content-addressed bytes that exist before admission. Hash-only plans with missing bytes, zero-length files, TODO/pass placeholders, or mismatched hashes are rejected.
### 7. One source for route contracts
Backend route registry and external OpenAPI must derive from one canonical route manifest, or be parity-checked in CI. Schema-only deployment is blocked if the matching backend route is absent; backend-only deployment is blocked if the external projection is absent when required.
### 8. Dual-plane reconciliation
Projection is idempotent and read-back verified. Admission writes an append-only projection intent/outbox in the same transaction as the grant; a projector writes TaskPlane, reads back exact task identity/revision, and records a receipt. Divergence is a first-class failure, never silently masked.
## Required release gates
- no BLOCKER from control-plane consistency audit
- C4 regression: legacy task creation path remains TaskPrincipal-gated
- admission -> CLAIMED -> RUNNING -> WriterLease -> governed commit -> readback integration test
- real activation + real SSH signature host test
- external projection readback test
- no placeholder artifact findings
@@ -0,0 +1,33 @@
# Remediation Work Packages
## P0-A — Single System Truth (no new DB)
Implement the read-only SystemTruth derivation in the runtime explain/status surface. Remove or rename PASS labels that are configuration-only.
## P0-B — Schedulability Gate
Admission may authorize a task, but TaskPlane projection to QUEUED requires a fresh SchedulabilityProof. If no eligible worker exists, expose WAITING_CAPACITY and do not start the short runtime grant TTL.
## P0-C — Just-in-time TaskPrincipal
Keep Owner authorization durable; mint short-lived TaskPrincipal only when an eligible worker is ready to claim. Bind task/job/branch/generation/execution_digest exactly.
## P0-D — Evidence-required state transitions
Guard CLAIMED/RUNNING/SUCCEEDED with claim holder, heartbeat, lease, commit receipt, OID change, file hashes, runtime readback, external projection readback.
## P0-E — Artifact reality gate
Before admission, verify every planned file byte exists, is non-placeholder, and matches SHA256. Reject hash-only or placeholder plans.
## P0-F — Route/projection single source
Generate or verify ChatGPT Actions OpenAPI from the backend route manifest. Block dead endpoints and unprojected required operations.
## P0-G — Dual-plane durable projection
Use append-only projection intent + idempotent TaskPlane write + exact readback receipt. Admission status and execution status remain separate.
## P0-H — Worker scope/schedulability diagnostics
Eligibility = capability AND heartbeat AND exact signed scope. Never select a worker only because capability names match.
## P0-I — QA gaps
Add:
1. C4 legacy createGovernedTask/projectOnlineAgentExecution remains TaskPrincipal-gated.
2. `_verified_task_principal` body injection negative test.
3. bearer-only no-delegation negative test.
4. real CLAIMED/RUNNING/WriterLease/commit readback path.
5. real activation/SSH signature host test.
+44
View File
@@ -782,3 +782,47 @@ paths:
'200': {description: Safe GitOps plan}
'401': {description: Unauthorized}
x-openai-isConsequential: true
/reopenOwnerApproval:
post:
operationId: reopenOwnerApproval
summary: Reopen one expired Owner approval cycle without rebuilding release bytes
security:
- bearerAuth: []
x-openai-isConsequential: true
requestBody:
required: true
content:
application/json:
schema:
type: object
additionalProperties: false
required:
- schema
- candidate_id
- expected_approval_fingerprint
- expected_source_commit
- expected_tree_oid
- expected_artifact_sha256
- expected_manifest_sha256
- expected_release_plan_sha256
- expected_policy_sha256
- reason
- idempotency_key
properties:
schema:
type: string
enum: [aios.owner-approval-reopen-request.v1]
candidate_id: {type: string}
expected_approval_fingerprint: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_source_commit: {type: string, pattern: "^[0-9a-f]{40}$"}
expected_tree_oid: {type: string, pattern: "^[0-9a-f]{40}$"}
expected_artifact_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_manifest_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_release_plan_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
expected_policy_sha256: {type: string, pattern: "^[0-9a-f]{64}$"}
reason: {type: string, minLength: 1, maxLength: 512}
idempotency_key: {type: string, minLength: 8, maxLength: 128}
responses:
"200": {description: Reopened or idempotent replay}
"403": {description: Missing release:approve capability}
"409": {description: Governed reopen gate rejected}
+112
View File
@@ -0,0 +1,112 @@
# Trusted Candidate Builder — CAP_SYS_PTRACE security contract
Status: formal (V2R2C2) · Applies to:
`deploy/systemd/aios-trusted-candidate-builder.service` (the live baseline unit)
## 0. Live baseline (rea truth, not nominal)
The authoritative deployed unit is the live systemd unit on the `hw` control
plane host:
- `User=root` / `Group=root` (the builder runs as root)
- socket: `aios-trusted-candidate-builder.socket`
`ListenStream=/run/aios-trusted-candidate-builder.sock`,
`SocketUser=root`, `SocketGroup=gitops`, `SocketMode=0660`
- credentials: `LoadCredentialEncrypted` for `signing_private_key` /
`builder_hmac` (under `/etc/credstore.encrypted/aios-trusted-candidate-builder/`)
and `source_admission_hmac` (under
`/etc/credstore.encrypted/aios-release-source-admission/`)
- hardening: `NoNewPrivileges`, `PrivateTmp`, `PrivateNetwork`,
`ProtectSystem=strict`, `ProtectHome`, `ProtectControlGroups`,
`ProtectKernelTunables/Modules/Logs`, `RestrictAddressFamilies=AF_UNIX`,
`LockPersonality`, `MemoryDenyWriteExecute`, `RestrictSUIDSGID`,
`SystemCallArchitectures=native`
- capability: `AmbientCapabilities=CAP_SYS_PTRACE` +
`CapabilityBoundingSet=CAP_SYS_PTRACE` (live current state)
The tracked `deploy/systemd/aios-trusted-candidate-builder.service` is kept
byte-identical (LF-normalized) to the live unit so sha256 drift is detectable:
live sha256 = `9d194c23ec981528d29d55f601d1be858635ac6229c7e25c10736df4e44042d3`.
> This contract deliberately does NOT introduce a separate non-root
> `-ptrace` unit. §2 of the V2R2C2 instruction forbids inventing a distinct
> service uid/group, `ProtectProc`/`ProcSubset`, credential paths or socket
> ownership that diverge from the live baseline. The live baseline is root +
> the single minimal capability, and that is the only variant tracked.
## 1. Why a capability is declared at all
`TrustedCandidateBuilder._verify_peer` validates the connecting peer before
signing any candidate. It reads three `/proc/<pid>` views of the peer:
| view | function | failure → error |
|---|---|---|
| `/proc/<pid>/cgroup` | `_resolve_cgroups` | `PEER_CGROUP_UNAVAILABLE` |
| `/proc/<pid>/exe` | `_resolve_process_identity` | `PEER_PROCESS_UNAVAILABLE` |
| `/proc/<pid>/cmdline` | `_resolve_process_identity` | `PEER_PROCESS_UNAVAILABLE` |
A cross-uid `readlink("/proc/<pid>/exe")` for a peer of a **different uid** is
subject to the Yama `ptrace_scope` sysctl and returns `EACCES` for a process
without the minimal `CAP_SYS_PTRACE`. Result: the builder fails closed
(`PEER_PROCESS_UNAVAILABLE`), which is safe but loses availability for
legitimate cross-uid peers.
The minimal `CAP_SYS_PTRACE` restores the ability to *inspect* the peer's
`/proc` view across uids.
## 2. Capability sensitivity — MANDATORY reading
`CAP_SYS_PTRACE` is a **high-sensitivity capability**. It is not a read-only
helper: it broadens the process's observation and debugging authority across
uids, including `ptrace`, reading `/proc/<pid>/mem`, and inspecting another
user's process state. It is therefore:
- limited to the **single capability** `CAP_SYS_PTRACE` (both
`AmbientCapabilities` and `CapabilityBoundingSet`), and
- scoped to the dedicated builder service unit only.
It **must never** be used to imply "read-only" or "no control authority." It
carries real inspection/debug power and is declared for one concrete purpose:
restoring cross-uid `/proc/<pid>/exe` identity resolution for the builder's
peer validation. Nothing more.
### Three-state verification contract
- **A. No CAP_SYS_PTRACE (fail-closed):** cross-uid `/proc/<pid>/exe` raises
`PermissionError(EACCES)` → resolvers fail closed →
`PEER_PROCESS_UNAVAILABLE` (and `PEER_CGROUP_UNAVAILABLE` for cgroup).
Never degrade to allow.
- **B. Minimal CAP_SYS_PTRACE + legitimate peer:** peer uid in
`allowed_peer_uid`, cgroup in `allowed_peer_cgroups`, executable matches
`allowed_peer_executable_prefix`, argv matches the runtime argument
contract → readiness `READY`.
- **C. CAP present but peer illegitimate:** any identity check fails →
`PEER_IDENTITY_DENIED` / `PEER_CGROUP_DENIED` / `PEER_PROCESS_DENIED`.
The capability **must never bypass the identity policy** — it restores the
ability to perform identity validation, not the authorization itself.
## 3. Operational rules
- Keep `CapabilityBoundingSet` and `AmbientCapabilities` pinned to exactly
`CAP_SYS_PTRACE`. Never use capability wildcards (e.g. `CAP_*` applying to
every capability).
- Keep `NoNewPrivileges=true`; ambient capabilities are the only inheritance
path into the exec'd Python process.
- Do not re-introduce a separate non-root unit with a different
uid/group/`ProtectProc`/`ProcSubset`/credential paths/socket ownership
(see §0). The root + single-capability unit is the live baseline.
- Runtime verification (three-state) MUST be executed on the real Linux host
that runs systemd; the adversarial states require real uids/cgroups/argv.
## 4. Testing
- Local (no `/proc` peer needed): the resolvers are exercised with
monkeypatched `PermissionError(EACCES)` — mirroring the exact bytes a
missing `CAP_SYS_PTRACE` / cross-uid read produces — covering
`cgroup → PEER_CGROUP_UNAVAILABLE`, `exe → PEER_PROCESS_UNAVAILABLE`,
`cmdline → PEER_PROCESS_UNAVAILABLE` (see
`tests/test_trusted_candidate_builder_captrace.py`).
- Host (Linux, required for final acceptance): run `systemd-analyze verify`
on the tracked unit; confirm the unit matches the live sha256; launch
with/without the capability; confirm A → `PEER_PROCESS_UNAVAILABLE`,
B → `READY`, C → `*_DENIED`.
@@ -0,0 +1,167 @@
#!/usr/bin/env python3
from __future__ import annotations
from pathlib import Path
import argparse, json, sys, time, re
from aios_control_plane_consistency import *
def jload(value):
try:
return json.loads(value or "{}")
except Exception:
return {}
def rows(path, query, params=()):
try:
return sqlite_rows(path, query, params)
except Exception as exc:
return [{"_error": str(exc)}]
def main() -> int:
ap = argparse.ArgumentParser(description="AIOS read-only control-plane consistency auditor")
ap.add_argument("--task-id", required=True)
ap.add_argument("--admission-db", type=Path, default=Path("/var/lib/aios-gitops/bridge-v2-staging/task-admission.sqlite3"))
ap.add_argument("--task-db", type=Path, default=Path("/var/lib/aios-agent-task/state.sqlite3"))
ap.add_argument("--git-dir", type=Path, default=Path("/data/aios-gitops/repo.git"))
ap.add_argument("--branch")
ap.add_argument("--openapi", type=Path)
ap.add_argument("--server-source", type=Path)
ap.add_argument("--artifact-root", type=Path)
ap.add_argument("--now", type=int)
ap.add_argument("--out", type=Path, default=Path("aios-control-plane-audit.json"))
ns = ap.parse_args()
now = ns.now or int(time.time())
findings = []
admission = None
task = None
workers = []
lease = None
git = None
evidence = EvidenceSnapshot()
# Admission plane.
try:
receipt = sqlite_rows(ns.admission_db,
"SELECT * FROM task_admission_receipts WHERE task_id=? ORDER BY rowid DESC LIMIT 1", (ns.task_id,))
if receipt:
r = dict(receipt[0])
envelope_id = r.get("envelope_id")
env = sqlite_rows(ns.admission_db,
"SELECT * FROM task_admission_envelopes WHERE envelope_id=? LIMIT 1", (envelope_id,))
grant = sqlite_rows(ns.admission_db,
"SELECT * FROM task_admission_grants WHERE envelope_id=? ORDER BY rowid DESC LIMIT 1", (envelope_id,))
ed = dict(env[0]) if env else {}
gd = dict(grant[0]) if grant else {}
gj = jload(gd.get("grant_json"))
ej = jload(ed.get("envelope_json"))
admission = AdmissionSnapshot(
task_id=ns.task_id,
envelope_id=envelope_id,
admission_status=ed.get("status"),
grant_id=gd.get("grant_id"),
grant_expires_at=gj.get("expires_at"),
one_time_use=gj.get("one_time_use"),
authority_base_oid=r.get("authority_base_oid") or ej.get("authority_base_oid"),
execution_digest=r.get("execution_digest") or ej.get("execution_digest"),
requested_capabilities=tuple(ej.get("requested_capabilities") or ()),
file_allowlist=tuple(ej.get("file_allowlist") or ()),
admitted_at=r.get("created_at") or r.get("admitted_at"),
)
except Exception as exc:
findings.append(Finding("ADMISSION_PLANE_UNREADABLE", Severity.BLOCKER, str(exc)))
# TaskPlane.
try:
trows = sqlite_rows(ns.task_db, "SELECT * FROM tasks WHERE task_id=? LIMIT 1", (ns.task_id,))
if trows:
td = dict(trows[0])
meta = jload(td.get("metadata_json"))
required = jload(td.get("required_capabilities_json"))
if isinstance(required, dict):
required = required.get("capabilities") or []
task = TaskSnapshot(
task_id=ns.task_id,
status=td.get("status"),
claimed_by=td.get("claimed_by"),
lease_expires_at=td.get("lease_expires_at"),
attempt=td.get("attempt"),
revision=td.get("revision"),
created_at=td.get("created_at"),
updated_at=td.get("updated_at"),
deadline=td.get("deadline"),
required_capabilities=tuple(required or ()),
metadata=meta if isinstance(meta, dict) else {},
)
arows = sqlite_rows(ns.task_db, "SELECT * FROM agents")
for row in arows:
d = dict(row)
caps = jload(d.get("capabilities_json"))
if isinstance(caps, dict):
caps = caps.get("capabilities") or []
last = parse_iso_epoch(d.get("last_seen_at"))
workers.append(WorkerSnapshot(
worker_id=d.get("agent_id") or d.get("worker_id") or "unknown",
status=d.get("status"),
last_seen_epoch=last,
capabilities=tuple(caps or ()),
))
except Exception as exc:
findings.append(Finding("TASK_PLANE_UNREADABLE", Severity.BLOCKER, str(exc)))
branch = ns.branch
if not branch and task and isinstance(task.metadata, dict):
plan = task.metadata.get("execution_plan") or task.metadata.get("plan") or task.metadata
if isinstance(plan, dict):
branch = plan.get("target_branch")
if branch:
current = git_ref(ns.git_dir, f"refs/heads/{branch}")
git = GitSnapshot(branch=branch, current_oid=current, expected_oid=admission.authority_base_oid if admission else None)
if admission and current and admission.authority_base_oid and current != admission.authority_base_oid:
findings.append(Finding(
"BRANCH_BINDING_DRIFT",
Severity.BLOCKER,
"Current branch OID differs from Admission authority_base_oid.",
{"current": current, "bound": admission.authority_base_oid},
"Replan/rebind formally; do not reuse old execution_digest.",
))
truth, derived = derive_truth(admission, task, workers, lease, git, evidence, now)
findings.extend(derived)
if task:
findings.extend(check_capability_aliases(task.required_capabilities, workers))
# Plan/artifact integrity if the immutable plan points to files.
if task and ns.artifact_root:
plan = task.metadata.get("execution_plan") or task.metadata.get("plan") or task.metadata
if isinstance(plan, dict):
hashes = plan.get("file_sha256") or {}
for rel in plan.get("file_allowlist") or []:
findings.extend(inspect_artifact(ns.artifact_root / rel, hashes.get(rel)))
if ns.openapi and ns.server_source:
findings.extend(check_openapi_backend_parity(
ns.openapi, ns.server_source,
required_ops=("getTaskAdmissionEnvelope","submitTaskAdmissionAuthorization","getTaskAdmissionStatus"),
))
report = {
"schema": "aios.control-plane-consistency-audit.v1",
"observed_at": now,
"task_id": ns.task_id,
"truth_state": truth.value,
"admission": asdict(admission) if admission else None,
"task": asdict(task) if task else None,
"eligible_workers": [asdict(w) for w in eligible_workers(task, workers, now)] if task else [],
"git": asdict(git) if git else None,
"findings": [asdict(f) for f in findings],
"summary": {
"blockers": sum(f.severity == Severity.BLOCKER for f in findings),
"majors": sum(f.severity == Severity.MAJOR for f in findings),
"minors": sum(f.severity == Severity.MINOR for f in findings),
},
}
dump_report(ns.out, report)
print(json.dumps(report["summary"], sort_keys=True))
return 2 if report["summary"]["blockers"] else 1 if report["summary"]["majors"] else 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,20 @@
#!/usr/bin/env python3
from aios_control_plane_consistency import *
NOW=2_000_000_000
scenarios = {}
a=AdmissionSnapshot(task_id="T",admission_status="ADMITTED",grant_expires_at=NOW+60)
scenarios["dual_plane_missing_task"]=derive_truth(a,None,[],None,None,None,NOW)
t=TaskSnapshot(task_id="T",status="QUEUED",required_capabilities=("writer.lease",),created_at="2033-05-18T03:23:20+00:00")
scenarios["queued_no_worker"]=derive_truth(a,t,[],None,None,None,NOW)
t2=TaskSnapshot(task_id="T",status="RUNNING",claimed_by="ghost")
scenarios["fake_running"]=derive_truth(a,t2,[],None,None,None,NOW)
t3=TaskSnapshot(task_id="T",status="SUCCEEDED")
scenarios["fake_success"]=derive_truth(a,t3,[],None,GitSnapshot("agent/x",expected_oid="a"*40),EvidenceSnapshot(),NOW)
for name,(state,findings) in scenarios.items():
print(name,state.value,[f.code for f in findings])
+154
View File
@@ -0,0 +1,154 @@
#!/usr/bin/env python3
"""Round-3 Gate-W proof using immutable, append-only provider receipts.
Requires a REAL clean Git checkout at FINAL_SOURCE_HEAD. It never deletes or
overwrites old receipts. Each run writes a new namespace.
It distinguishes three dimensions:
* provider authentication closure,
* provider result gates (e.g. tests passed),
* overall Gate-W readiness.
A resolver PASS never converts ``tests.result.passed=false`` into Gate-W PASS.
"""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import subprocess
import sys
_APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app"
sys.path.insert(0, str(_APP))
from gitea_transport_consistency_provider import produce_gitea_consistency_receipt
from trusted_evidence_providers import TrustedProviderExecutionError, _validate_oid, produce
from trusted_provider_receipts import TrustedProviderPolicy, TrustedProviderReceiptResolver
GATE_W = ("tests", "secret_scan", "gitea")
def mode(v):
return v if isinstance(v, int) else int(str(v), 8)
def build_resolver(policy_path):
raw = json.loads(Path(policy_path).read_text(encoding="utf-8"))
defaults = dict(raw.get("defaults") or {})
policies = {}
for name,cfg in dict(raw.get("providers") or {}).items():
policies[name] = TrustedProviderPolicy(
evidence_kind=name,
provider_identity=str(cfg["provider_identity"]),
store_id=str(cfg["store_id"]),
store_root=Path(str(cfg["store_root"])),
allowed_operations=frozenset(map(str,cfg["allowed_operations"])),
expected_owner_uid=int(cfg.get("expected_owner_uid", defaults.get("expected_owner_uid",0))),
allowed_directory_modes=frozenset(mode(x) for x in cfg.get("allowed_directory_modes",defaults.get("allowed_directory_modes",["0700","0750"]))),
allowed_receipt_modes=frozenset(mode(x) for x in cfg.get("allowed_receipt_modes",defaults.get("allowed_receipt_modes",["0600","0640"]))),
authentication_kind=str(cfg.get("authentication_kind",defaults.get("authentication_kind","root-owned-store-v1"))),
max_age_seconds=cfg.get("max_age_seconds",defaults.get("max_age_seconds")),
)
return TrustedProviderReceiptResolver(policies), policies
def git(repo, *args):
p = subprocess.run(["git",*args],cwd=str(repo),text=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,check=False)
if p.returncode:
raise SystemExit(f"GIT_PRECHECK_FAILED {' '.join(args)}: {p.stderr.strip()}")
return p.stdout.strip()
def main():
ap=argparse.ArgumentParser()
ap.add_argument("--source-oid",required=True)
ap.add_argument("--tree-oid",required=True)
ap.add_argument("--checkout",required=True)
ap.add_argument("--policy",required=True)
ap.add_argument("--run-id",default=None)
args=ap.parse_args()
source=_validate_oid(args.source_oid,"source_oid")
tree=_validate_oid(args.tree_oid,"tree_oid")
repo=Path(args.checkout).resolve()
if git(repo,"rev-parse","HEAD^{commit}").lower()!=source:
raise SystemExit("FINAL_SOURCE_HEAD_MISMATCH")
if git(repo,"rev-parse",f"{source}^{{tree}}").lower()!=tree:
raise SystemExit("FINAL_SOURCE_TREE_MISMATCH")
if git(repo,"status","--porcelain"):
raise SystemExit("FINAL_SOURCE_WORKTREE_DIRTY")
run_id=args.run_id or datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
resolver,policies=build_resolver(args.policy)
os.environ["AIOS_REPO_ROOT"]=str(repo)
os.environ.setdefault("AIOS_GITLEAKS_BIN","/usr/bin/gitleaks")
produced=[]
mechanism_required=[]
for kind in GATE_W:
ref=f"round3-gate-w/{source[:12]}/{run_id}/{kind}.json"
try:
if kind=="gitea":
sha,_payload=produce_gitea_consistency_receipt(
repo_root=repo,
source_oid=source,
tree_oid=tree,
policy_path=args.policy,
receipt_ref=ref,
)
else:
sha=produce(
kind,
source_oid=source,
tree_oid=tree,
receipt_ref=ref,
store_root=policies[kind].store_root,
)
produced.append((kind,ref,sha))
except TrustedProviderExecutionError as exc:
if "PROVIDER_MECHANISM_REQUIRED" in str(exc):
mechanism_required.append((kind,str(exc)))
else:
raise
auth_ok=True
results={}
for kind,ref,sha in produced:
r=resolver.resolve(kind,ref,source_oid=source,tree_oid=tree)
results[kind]=dict(r.result)
print(f"[AUTH PASS] {kind} ref={ref} sha={sha} result={json.dumps(r.result,sort_keys=True)}")
expected=set(GATE_W)
closed={k for k,_,_ in produced}
auth_closed=auth_ok and closed==expected and not mechanism_required
test_gate=bool(results.get("tests",{}).get("passed") is True)
secret_gate=bool(results.get("secret_scan",{}).get("passed") is True)
gitea_gate=bool(results.get("gitea",{}).get("consistent") is True)
gate_ready=auth_closed and test_gate and secret_gate and gitea_gate
summary={
"source_oid":source,
"tree_oid":tree,
"run_id":run_id,
"authenticated_providers":sorted(closed),
"mechanism_required":[k for k,_ in mechanism_required],
"provider_auth_closed":auth_closed,
"tests_result_passed":test_gate,
"secret_scan_result_passed":secret_gate,
"gitea_consistent":gitea_gate,
"gate_w_ready":gate_ready,
}
print("GATE_W_PROVE="+json.dumps(summary,sort_keys=True))
if mechanism_required:
return 2
if not auth_closed:
return 1
if not gate_ready:
return 3
return 0
if __name__=="__main__":
raise SystemExit(main())
@@ -0,0 +1,76 @@
from pathlib import Path
import time
from aios_control_plane_consistency import *
NOW=2_000_000_000
def task(status="QUEUED", caps=("governed.write.files","task.principal.issue","writer.lease"), **kw):
return TaskSnapshot(
task_id="T", status=status, required_capabilities=tuple(caps),
created_at="2033-05-18T03:23:20+00:00", metadata={
"external_job_id":"job-a",
"target_branch":"agent/x",
"execution_digest":"d"*64,
}, **kw
)
def admission(**kw):
base=dict(task_id="T", admission_status="ADMITTED", envelope_id="E", grant_expires_at=NOW+600)
base.update(kw)
return AdmissionSnapshot(**base)
def worker(worker_id="w", caps=("governed.write.files","task.principal.issue","writer.lease"), last_seen=NOW-10, **kw):
return WorkerSnapshot(worker_id=worker_id,status="ACTIVE",last_seen_epoch=last_seen,capabilities=tuple(caps),**kw)
def test_admitted_without_taskplane_is_not_execution():
state, fs=derive_truth(admission(),None,[],None,None,None,NOW)
assert state==TruthState.ADMITTED_NOT_PROJECTED
assert any(f.code=="DUAL_PLANE_ADMITTED_TASK_MISSING" for f in fs)
def test_queued_without_eligible_worker_is_waiting_capacity_not_real_execution():
state, fs=derive_truth(admission(),task(),[],None,None,None,NOW)
assert state==TruthState.WAITING_CAPACITY
assert any(f.code=="QUEUED_WITHOUT_ELIGIBLE_WORKER" for f in fs)
def test_scope_mismatched_worker_is_not_eligible():
w=worker(scope_task_id="OTHER")
assert eligible_workers(task(),[w],NOW)==[]
def test_stale_worker_is_not_eligible():
w=worker(last_seen=NOW-1000)
assert eligible_workers(task(),[w],NOW)==[]
def test_capability_singular_plural_is_not_silent_alias():
fs=check_capability_aliases(("governed.write.files",),[
worker(caps=("governed.write.file","task.principal.issue","writer.lease"))
])
assert any(f.code=="CAPABILITY_SINGULAR_PLURAL_ASYMMETRY" for f in fs)
def test_claimed_without_claim_holder_is_fake():
state, fs=derive_truth(admission(),task(status="CLAIMED"),[],None,None,None,NOW)
assert state==TruthState.CLAIMED
assert any(f.code=="FAKE_CLAIM_STATE" for f in fs)
def test_running_without_heartbeat_is_fake_execution():
t=task(status="RUNNING",claimed_by="w")
state, fs=derive_truth(admission(),t,[worker(last_seen=NOW-999)],LeaseSnapshot("T","w",NOW+60,"ACTIVE"),None,None,NOW)
assert any(f.code=="FAKE_RUNNING_NO_HEARTBEAT" for f in fs)
def test_succeeded_without_commit_receipt_is_unproven():
state, fs=derive_truth(admission(),task(status="SUCCEEDED"),[],None,GitSnapshot("agent/x",expected_oid="a"*40),EvidenceSnapshot(),NOW)
assert state==TruthState.TERMINAL_UNPROVEN
assert any(f.code=="TERMINAL_SUCCESS_UNPROVEN" for f in fs)
def test_expired_without_claim_is_schedulability_failure():
state, fs=derive_truth(admission(grant_expires_at=NOW-1),task(status="EXPIRED",attempt=0),[],None,None,None,NOW)
assert state==TruthState.EXPIRED
assert any(f.code=="EXPIRED_WITHOUT_ANY_CLAIM" for f in fs)
def test_placeholder_artifact_is_rejected(tmp_path):
p=tmp_path/"x.py"; p.write_text("TODO: real content\n",encoding="utf-8")
fs=inspect_artifact(p)
assert any(f.code=="PLACEHOLDER_ARTIFACT" for f in fs)
def test_hash_only_does_not_make_missing_artifact_real(tmp_path):
fs=inspect_artifact(tmp_path/"missing.py","0"*64)
assert any(f.code=="ARTIFACT_MISSING" for f in fs)
@@ -0,0 +1,89 @@
from __future__ import annotations
from aios_control_plane_consistency import (
AdmissionSnapshot,
EvidenceSnapshot,
GitSnapshot,
LeaseSnapshot,
TaskSnapshot,
WorkerSnapshot,
)
from aios_control_plane_snapshot import (
attach_control_plane_truth,
build_control_plane_snapshot,
)
NOW = 2_000_000_000
def _base():
return build_control_plane_snapshot(
target_version="V2R2C2",
runtime={"drift_status": "PASS"},
task={"task_id": "T", "reconciliation_state": "ok"},
)
def test_snapshot_base_has_no_derived_block_by_default():
assert "derived" not in _base()
def test_truth_attach_preserves_existing_keys():
out = attach_control_plane_truth(_base(), now=NOW)
assert set(_base().keys()) <= set(out.keys())
assert "derived" in out
def test_admitted_without_taskplane_is_admitted_not_projected():
out = attach_control_plane_truth(
_base(),
admission=AdmissionSnapshot(task_id="T", admission_status="ADMITTED", envelope_id="E"),
now=NOW,
)
truth = out["derived"]["system_truth"]
assert truth["truth_state"] == "ADMITTED_NOT_PROJECTED"
assert truth["blocker_count"] == 1
assert any(f["code"] == "DUAL_PLANE_ADMITTED_TASK_MISSING" for f in truth["findings"])
# A drift_status=PASS surface must NOT be reported as evidence-complete.
assert truth["evidence_flow_confirmed"] is False
def test_fake_success_is_terminal_unproven_even_under_drift_pass():
out = attach_control_plane_truth(
_base(),
admission=AdmissionSnapshot(task_id="T"),
task=TaskSnapshot(task_id="T", status="SUCCEEDED"),
git=GitSnapshot("agent/x", expected_oid="a" * 40),
evidence=EvidenceSnapshot(),
now=NOW,
)
truth = out["derived"]["system_truth"]
assert truth["truth_state"] == "TERMINAL_UNPROVEN"
assert truth["blocker_count"] == 1
assert truth["evidence_flow_confirmed"] is False
def test_complete_requires_full_evidence_chain():
worker = WorkerSnapshot(
worker_id="w", status="ACTIVE", last_seen_epoch=NOW - 10,
capabilities=("governed.write.files",),
)
task = TaskSnapshot(
task_id="T", status="SUCCEEDED", claimed_by="w",
required_capabilities=("governed.write.files",),
metadata={"external_job_id": "job-a", "target_branch": "agent/x",
"execution_digest": "d" * 64},
)
git = GitSnapshot("agent/x", expected_oid="a" * 40, commit_oid="b" * 40, post_oid="c" * 40)
evidence = EvidenceSnapshot(runtime_readback={"ok": True})
out = attach_control_plane_truth(
_base(),
admission=AdmissionSnapshot(task_id="T"),
task=task, workers=[worker],
lease=LeaseSnapshot("T", "w", NOW + 60, "ACTIVE"),
git=git, evidence=evidence, now=NOW,
)
truth = out["derived"]["system_truth"]
assert truth["truth_state"] == "COMPLETE"
assert truth["blocker_count"] == 0
assert truth["evidence_flow_confirmed"] is True
@@ -0,0 +1,27 @@
from pathlib import Path
from aios_control_plane_consistency import check_openapi_backend_parity
def test_openapi_dead_endpoint_is_blocker(tmp_path):
oa=tmp_path/"o.yaml"
oa.write_text("""openapi: 3.1.0
info: {title: t, version: '1'}
paths:
/v2/task-admission/envelope:
post:
operationId: getTaskAdmissionEnvelope
""",encoding="utf-8")
srv=tmp_path/"s.py"
srv.write_text('ROUTES = {"/v2/other": "other"}\n',encoding="utf-8")
fs=check_openapi_backend_parity(oa,srv,("getTaskAdmissionEnvelope",))
assert any(f.code=="OPENAPI_DEAD_ENDPOINT" for f in fs)
def test_backend_unprojected_operation_is_reported(tmp_path):
oa=tmp_path/"o.yaml"
oa.write_text("""openapi: 3.1.0
info: {title: t, version: '1'}
paths: {}
""",encoding="utf-8")
srv=tmp_path/"s.py"
srv.write_text('ROUTES = {"/v2/task-admission/status": "getTaskAdmissionStatus"}\n',encoding="utf-8")
fs=check_openapi_backend_parity(oa,srv,("getTaskAdmissionStatus",))
assert any(f.code=="BACKEND_UNPROJECTED_OPERATION" for f in fs)
@@ -0,0 +1,81 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from gitea_transport_consistency_provider import (
GitCommandResult,
GiteaProviderError,
collect_gitea_transport_consistency,
repository_identity_from_remote,
)
SRC = "a"*40
TREE = "b"*40
BRANCH = "refs/heads/agent/v2r2c2-p0-ttl-captrace-contract-20260909"
def mapping(remote_oid=SRC, url="https://g.cnnetmkt.top/git/aios/aios.git"):
return {
("rev-parse", "--is-inside-work-tree"): "true\n",
("rev-parse", "HEAD^{commit}"): SRC+"\n",
("rev-parse", f"{SRC}^{{tree}}"): TREE+"\n",
("symbolic-ref", "--quiet", "HEAD"): BRANCH+"\n",
("remote", "get-url", "origin"): url+"\n",
("ls-remote", "--exit-code", "origin", BRANCH): f"{remote_oid}\t{BRANCH}\n",
}
def runner(mp):
def run(args, cwd):
val = mp[tuple(args[1:])]
if isinstance(val, GitCommandResult):
return val
return GitCommandResult(0, val, "")
return run
class Tests(unittest.TestCase):
def call(self, mp):
with tempfile.TemporaryDirectory() as d:
return collect_gitea_transport_consistency(
repo_root=Path(d),
source_oid=SRC,
tree_oid=TREE,
git_runner=runner(mp),
)
def test_identity_parsing(self):
self.assertEqual(repository_identity_from_remote(
"https://user:secret@g.cnnetmkt.top/git/aios/aios.git?x=y"
), "aios/aios")
self.assertEqual(repository_identity_from_remote(
"git@g.cnnetmkt.top:aios/aios.git"
), "aios/aios")
def test_exact_remote_ref_passes(self):
r = self.call(mapping())
self.assertTrue(r["consistent"])
self.assertEqual(r["remote_ref_oid"], SRC)
self.assertFalse(r["api_verified"])
self.assertEqual(r["transport"], "git-ls-remote")
def test_remote_oid_mismatch_denied(self):
with self.assertRaisesRegex(GiteaProviderError, "GITEA_REMOTE_REF_OID_MISMATCH"):
self.call(mapping(remote_oid="c"*40))
def test_repository_mismatch_denied(self):
with self.assertRaisesRegex(GiteaProviderError, "GITEA_REPOSITORY_IDENTITY_MISMATCH"):
self.call(mapping(url="https://g.cnnetmkt.top/git/other/repo.git"))
def test_local_source_mismatch_denied(self):
mp = mapping()
mp[("rev-parse", "HEAD^{commit}")] = "d"*40+"\n"
with self.assertRaisesRegex(GiteaProviderError, "GITEA_LOCAL_SOURCE_OID_MISMATCH"):
self.call(mp)
def test_remote_failure_denied(self):
mp = mapping()
mp[("ls-remote", "--exit-code", "origin", BRANCH)] = GitCommandResult(2,"","down")
with self.assertRaisesRegex(GiteaProviderError, "GITEA_REMOTE_REF_UNAVAILABLE"):
self.call(mp)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,131 @@
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"
+136
View File
@@ -0,0 +1,136 @@
from __future__ import annotations
"""Governed evidence runtime registration chain tests (V2R2C2 P0).
Verifies the REAL Inventory -> Resolver -> operation -> Producer -> closure
chain. ``PRODUCER_RUNTIME_REGISTERED_PASS`` is only reported when the operation
is actually present in the runtime inventory, its handler_ref resolves to the
real producer callable, and that callable genuinely produces an evidence
closure — not merely because an OperationDefinition exists in a registry.
"""
from pathlib import Path
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT / "active-runtime" / "app") not in sys.path:
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from governed_evidence_runtime import ( # noqa: E402
EvidenceOperationInventory,
invoke_governed_produce,
resolve_handler,
)
from governed_release_evidence_producer import ( # noqa: E402
EvidenceProduceError,
GovernedBuildArtifacts,
)
from test_governed_release_evidence_producer import ( # noqa: E402
POLICY_PATH,
build_input,
)
def _payload(request: GovernedBuildArtifacts) -> dict:
return {
"version": request.version,
"candidate_id": request.candidate_id,
"source_commit": request.source_commit,
"tree_oid": request.tree_oid,
"artifact_hex": request.artifact.hex(),
"manifest": dict(request.manifest),
"package_closure_hex": request.package_closure.hex(),
"policy": dict(request.policy),
"tests": dict(request.tests),
"canary": dict(request.canary),
"gitea": dict(request.gitea),
"rollback": dict(request.rollback),
"secret_scan": dict(request.secret_scan),
"deployment": dict(request.deployment),
"tag": dict(request.tag),
"finalization": dict(request.finalization),
"rollback_sha256": request.rollback_sha256,
"issued_at": request.issued_at,
}
def test_inventory_contains_governed_produce_operation() -> None:
inventory = EvidenceOperationInventory.default()
assert inventory.contains("governed.release.evidence.produce")
definition = inventory.get("governed.release.evidence.produce")
assert definition is not None
assert definition.operation_id == "governed.release.evidence.produce"
assert "governed.release.evidence.produce" in definition.required_permissions
def test_resolver_resolves_producer_handler_exactly() -> None:
inventory = EvidenceOperationInventory.default()
definition = inventory.get("governed.release.evidence.produce")
handler = resolve_handler(definition.handler_ref)
assert handler.__name__ == "produce_evidence_closure"
def test_resolver_rejects_unknown_handler() -> None:
with pytest.raises(EvidenceProduceError, match="HANDLER_REF_UNRESOLVED"):
resolve_handler("some.fake.handler")
def test_runtime_registered_pass_on_real_invocation(tmp_path: Path) -> None:
"""Full chain: inventory contains op, resolver resolves handler, and the
resolved producer callable genuinely produces a closure -> PASS."""
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
request, _, _ = build_input(tmp_path)
verdict = invoke_governed_produce(
builder_policy=POLICY_PATH,
source_root=source_root,
payload=_payload(request),
expected_owner_uid=None,
)
assert verdict["status"] == "PRODUCER_RUNTIME_REGISTERED_PASS"
assert verdict["operation_id"] == "governed.release.evidence.produce"
assert verdict["handler_resolved"] == (
"governed_release_evidence_producer.produce_evidence_closure"
)
assert set(verdict["evidence_types_bound"]) == {
"tests", "canary", "gitea", "rollback", "secret_scan",
}
assert verdict["result"]["produce_status"] == "EVIDENCE_PRODUCED"
# A real closure directory and receipt exist at source_root/<version>.
closure_dir = Path(verdict["result"]["closure_dir"])
assert closure_dir.is_dir()
assert (closure_dir / "receipt.json").is_file()
# The runtime-registered produce path is idempotently re-runnable.
again = invoke_governed_produce(
builder_policy=POLICY_PATH,
source_root=source_root,
payload=_payload(request),
expected_owner_uid=None,
)
assert again["result"]["produce_status"] == "EVIDENCE_PRODUCED"
assert again["result"]["receipt_sha256"] == verdict["result"]["receipt_sha256"]
def test_runtime_chain_still_fails_closed_on_bare_passed_flag(tmp_path: Path) -> None:
"""Even through the runtime chain, a bare {"passed": true} without a
trusted-provider receipt fails closed (the producer never trusts a caller-
supplied pass flag)."""
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
request, _, _ = build_input(tmp_path)
payload = _payload(request)
bad_tests = dict(payload["tests"])
bad_tests.pop("receipt", None)
payload["tests"] = bad_tests
with pytest.raises(EvidenceProduceError, match="EVIDENCE_RECEIPT_MISSING"):
invoke_governed_produce(
builder_policy=POLICY_PATH,
source_root=source_root,
payload=payload,
expected_owner_uid=None,
)
assert not (source_root / request.version).exists()
@@ -0,0 +1,141 @@
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))
@@ -0,0 +1,444 @@
"""Governed release evidence producer contract tests (V2R2C2 P0).
Proves that ``governed.release.evidence.produce`` (the producer half) writes a
self-consistent evidence closure from real build artifacts, and that the closure
can be re-validated and projected byte-for-byte by ``TrustedBuilderEvidenceMaterializer``
(the projection half). Also proves fail-closed behavior on tampered/inconsistent
input and idempotency/conflict handling, plus operation-registry registration.
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from types import SimpleNamespace
import pytest
ROOT = Path(__file__).resolve().parents[1]
import sys
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
from aios_trusted_candidate_builder import ( # noqa: E402
load_builder_policy,
TrustedCandidateBuilderConfig,
)
from governed_release_evidence_producer import ( # noqa: E402
EvidenceProduceError,
GovernedBuildArtifacts,
producer_operation_definition,
producer_registry,
produce_evidence_closure,
)
from release_evidence_materializer import ( # noqa: E402
TrustedBuilderEvidenceMaterializer,
)
POLICY_PATH = ROOT / "config" / "gitops" / "trusted-candidate-builder-policy.json"
def canonical(value: dict) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
).encode()
def digest(label: str) -> str:
return hashlib.sha256(label.encode()).hexdigest()
def make_provider_receipt(provider: str, body: dict, source_commit: str,
tree_oid: str, issued_at: str) -> dict:
"""Bind a trusted-provider receipt to an evidence body (content-digest bound)."""
return {
"schema": "aios.trusted-provider.evidence.v1",
"provider_identity": provider,
"operation_run_id": f"{provider}-run-{digest('run-' + str(body.get('passed', '')))[:12]}",
"result_reference": f"{provider}/{digest('result')[:16]}/result.json",
"content_sha256": hashlib.sha256(canonical(dict(body))).hexdigest(),
"source_oid": source_commit,
"tree_oid": tree_oid,
"environment": "trusted-builder-gitops-live",
"issued_at": issued_at,
}
def bind_receipt(evidence: dict, provider: str, source_commit: str,
tree_oid: str, issued_at: str) -> dict:
return {**dict(evidence), "receipt": make_provider_receipt(
provider, dict(evidence), source_commit, tree_oid, issued_at)}
def build_input(tmp_path: Path, *, tamper_artifact: bool = False,
bad_secret_scan: bool = False) -> tuple[GovernedBuildArtifacts, str, str]:
config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
recipe = config.fixed_recipe("V2R2C2")
version = recipe["version"]
tree_oid = "a" * 40
artifact_bytes = b"artifact-bytes"
if tamper_artifact:
artifact_bytes = b"artifact-tampered"
real_artifact_sha = hashlib.sha256(artifact_bytes).hexdigest()
closure_bytes = b"checksums-v1\n"
closure_sha = hashlib.sha256(closure_bytes).hexdigest()
manifest_dict = {
"artifact_sha256": real_artifact_sha,
"package_closure_sha256": closure_sha,
}
# manifest_sha256 is the digest of the manifest evidence file; the producer
# derives it from the manifest content (never self-referential).
manifest_sha = hashlib.sha256(canonical(manifest_dict)).hexdigest()
policy_dict = {"trusted": "release-policy"}
# policy_sha256 is the digest of the policy evidence file; derived by the
# producer from policy content (not a self-referential field).
policy_sha = hashlib.sha256(canonical(policy_dict)).hexdigest()
issued_at = "2026-09-09T00:00:00+00:00"
src = recipe["source_commit"]
tests = bind_receipt({"passed": True}, "aios.tests-runner", src, tree_oid, issued_at)
canary = bind_receipt({"passed": True}, "aios.canary-gate", src, tree_oid, issued_at)
gitea = bind_receipt(
{
"consistent": True,
"remote_ref_oid": recipe["source_commit"],
"repository": config.repository,
"source_commit": recipe["source_commit"],
"tree_oid": tree_oid,
},
"aios.gitea-integrity", src, tree_oid, issued_at,
)
rollback_sha = digest("rollback-contract")
rollback = bind_receipt(
{
"ready": True,
"rollback_sha256": rollback_sha,
"source_commit": recipe["source_commit"],
},
"aios.rollback-controller", src, tree_oid, issued_at,
)
if bad_secret_scan:
secret_scan = bind_receipt(
{"artifact_sha256": digest("wrong"), "passed": True},
"aios.secret-scanner", src, tree_oid, issued_at,
)
else:
secret_scan = bind_receipt(
{"artifact_sha256": real_artifact_sha, "passed": True},
"aios.secret-scanner", src, tree_oid, issued_at,
)
deployment = {
"service": "aios-silent-release-controller",
"release_path": "/srv/release/r1",
"release_root": "/srv/release",
"cutover_link": "/srv/release/current",
"dropin_path": "/srv/release/r1/deploy/systemd/controller.service",
"dropin_sha256": digest("dropin"),
"systemd_unit_sha256": digest("unit"),
"rollback_path": "/srv/release/rollback-r1",
"rollback_sha256": rollback_sha,
"authoritative_git_dir": "/data/aios-gitops/repo.git",
"repository_identity": config.repository,
"baseline_before_sha256": digest("baseline"),
"registry_before_sha256": digest("registry"),
"production_tag_refset_before_sha256": digest("tagset"),
"health_contract_version": "trusted-builder/v1",
"timeout_policy": {"seconds": 10},
"retry_policy": {"attempts": 1},
"health_contract": {"checked": True},
}
tag = {
"version": "trusted-v2r2c2-001",
"tag_type": "annotated",
"expected_tag_oid": "b" * 40,
"expected_target_oid": recipe["source_commit"],
"evidence_closure_oid": "c" * 40,
"approval_plan_id": "owner-plan-trusted-v2r2c2",
"repository_identity": config.repository,
"gitea_owner": "aios",
"gitea_repository": "aios",
"current_remote_tag_oid": None,
"remote_absent_evidence_sha256": digest("tag-absence"),
"verification_contract": "trusted-builder/v1",
}
finalization = {
"version": "trusted-v2r2c2-001",
"expected_tag_oid": "b" * 40,
"expected_target_oid": recipe["source_commit"],
"expected_previous_baseline_sha256": digest("baseline"),
"registry_expected_record": {"base": recipe["source_commit"]},
"baseline_expected_record": {"base": recipe["source_commit"]},
"final_consistency_contract": "trusted-builder/v1",
}
request = GovernedBuildArtifacts(
version=version,
candidate_id=recipe["candidate_id"],
source_commit=recipe["source_commit"],
tree_oid=tree_oid,
artifact=artifact_bytes,
manifest=manifest_dict,
package_closure=closure_bytes,
policy=policy_dict,
tests=tests,
canary=canary,
gitea=gitea,
rollback=rollback,
secret_scan=secret_scan,
deployment=deployment,
tag=tag,
finalization=finalization,
rollback_sha256=rollback_sha,
issued_at="2026-09-09T00:00:00+00:00",
)
return request, recipe["candidate_id"], tree_oid
def fake_git(tree_oid: str):
def runner(argv, **_kwargs):
if "rev-parse" in argv:
return SimpleNamespace(stdout=tree_oid + "\n", returncode=0)
return SimpleNamespace(stdout="", returncode=0)
return runner
def make_materializer(tmp_path: Path, source_root: Path, output_root: Path, tree_oid: str):
authority = tmp_path / "repo.git"
authority.mkdir(mode=0o700, exist_ok=True)
output_root.mkdir(mode=0o700, exist_ok=True)
return TrustedBuilderEvidenceMaterializer(
builder_policy=POLICY_PATH,
authority_git_dir=authority,
source_root=source_root,
output_root=output_root,
expected_owner_uid=None,
git_runner=fake_git(tree_oid),
)
def test_produce_writes_closure_that_revalidates_under_builder_verification(tmp_path: Path) -> None:
"""Producer closure re-validates and is projected byte-for-byte by the materializer.
The producer must write a self-consistent closure that the deterministic builder
verification re-validates, and that TrustedBuilderEvidenceMaterializer projects
with destination bytes exactly equal to the source (byte-exact, including a
package-closure fixture containing an LF byte).
"""
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
output_root = tmp_path / "output"
request, candidate_id, tree_oid = build_input(tmp_path)
version = request.version
result = produce_evidence_closure(
builder_policy=POLICY_PATH,
source_root=source_root,
request=request,
expected_owner_uid=None,
)
assert result.status == "EVIDENCE_PRODUCED"
assert result.idempotent is False
assert (source_root / version / "receipt.json").is_file()
for name in ("artifact.tar.gz", "manifest.json", "checksums.sha256",
"trusted-release-policy.json", "tests.json", "canary.json",
"gitea.json", "rollback.json", "secret-scan.json"):
assert (source_root / version / name).is_file(), f"missing {name}"
# Re-validate the produced closure with the deterministic builder verification
# (the exact collector the materializer uses); it must collect cleanly and
# reproduce the producer's evidence digests.
from aios_trusted_candidate_builder import ( # noqa: F811
DeterministicEvidenceCollector,
ServerOwnedEvidenceSource,
)
config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
authority = tmp_path / "repo.git"
authority.mkdir(mode=0o700)
source = ServerOwnedEvidenceSource(
config, authority, source_root.resolve(),
expected_owner_uid=None, git_runner=fake_git(tree_oid),
)
collector = DeterministicEvidenceCollector(config, evidence_source=source)
recipe = config.fixed_recipe(version)
bundle = collector.collect(version, recipe)
assert set(bundle["evidence_digests"]) == set(result.evidence_digests)
for name, digest in result.evidence_digests.items():
assert bundle["evidence_digests"][name] == digest, f"digest mismatch {name}"
# Byte-exact projection: source closure -> materializer -> destination closure,
# with destination bytes exactly equal to source (incl. the LF-bearing
# package-closure file) and a destination collector PASS.
materializer = make_materializer(tmp_path, source_root, output_root, tree_oid)
outcome = materializer.materialize(
version=version,
candidate_id=candidate_id,
source_commit=request.source_commit,
)
assert outcome["status"] == "EVIDENCE_MATERIALIZED"
for name in ("receipt.json", "artifact.tar.gz", "manifest.json",
"checksums.sha256", "trusted-release-policy.json",
"tests.json", "canary.json", "gitea.json",
"rollback.json", "secret-scan.json"):
src = (source_root / version / name).read_bytes()
dst = (output_root / version / name).read_bytes()
assert src == dst, f"byte-exact projection failed for {name}"
assert hashlib.sha256(src).hexdigest() == hashlib.sha256(dst).hexdigest()
# Destination closure passes the collector too.
dst_source = ServerOwnedEvidenceSource(
config, authority, output_root.resolve(),
expected_owner_uid=None, git_runner=fake_git(tree_oid),
)
dst_collector = DeterministicEvidenceCollector(config, evidence_source=dst_source)
dst_bundle = dst_collector.collect(version, recipe)
assert set(dst_bundle["evidence_digests"]) == set(result.evidence_digests)
def test_produce_is_idempotent_on_identical_closure(tmp_path: Path) -> None:
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
request, _, _ = build_input(tmp_path)
first = produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
second = produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
assert first.idempotent is False
assert second.idempotent is True
assert second.status == "EVIDENCE_PRODUCED"
def test_produce_conflicts_on_tampered_artifact(tmp_path: Path) -> None:
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
request, _, _ = build_input(tmp_path)
produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
tampered, _, _ = build_input(tmp_path, tamper_artifact=True)
with pytest.raises(EvidenceProduceError):
produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=tampered,
expected_owner_uid=None,
)
def test_produce_fails_closed_on_secret_scan_artifact_mismatch(tmp_path: Path) -> None:
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
bad, _, _ = build_input(tmp_path, bad_secret_scan=True)
with pytest.raises(EvidenceProduceError):
produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=bad,
expected_owner_uid=None,
)
assert not (source_root / bad.version).exists()
def test_produce_registers_governed_operation() -> None:
definition = producer_operation_definition()
assert definition.operation_id == "governed.release.evidence.produce"
registry = producer_registry()
assert (
registry.get("governed.release.evidence.produce").operation_id
== "governed.release.evidence.produce"
)
assert "governed.release.evidence.produce" in definition.required_permissions
def _clone_with(request: GovernedBuildArtifacts, **overrides) -> GovernedBuildArtifacts:
data = {k: getattr(request, k) for k in request.__dataclass_fields__}
data.update(overrides)
return GovernedBuildArtifacts(**data)
def test_produce_requires_trusted_provider_receipt_on_every_evidence(tmp_path: Path) -> None:
# A bare {"passed": true} without a trusted-provider receipt must fail closed
# for EVERY governed evidence type (tests/canary/gitea/rollback/secret_scan).
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
good, _, _ = build_input(tmp_path)
for etype in ("tests", "canary", "gitea", "rollback", "secret_scan"):
bad_evidence = dict(getattr(good, etype))
bad_evidence.pop("receipt", None)
request = _clone_with(good, **{etype: bad_evidence})
with pytest.raises(EvidenceProduceError, match="EVIDENCE_RECEIPT_MISSING"):
produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
assert not (source_root / good.version).exists()
def test_produce_rejects_receipt_content_digest_mismatch(tmp_path: Path) -> None:
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
good, _, _ = build_input(tmp_path)
bad_evidence = dict(getattr(good, "tests"))
bad_evidence["receipt"] = {**bad_evidence["receipt"], "content_sha256": "0" * 64}
request = _clone_with(good, tests=bad_evidence)
with pytest.raises(EvidenceProduceError, match="EVIDENCE_RECEIPT_CONTENT_MISMATCH"):
produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
def test_produce_rejects_missing_receipt_fields(tmp_path: Path) -> None:
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
good, _, _ = build_input(tmp_path)
bad_evidence = dict(getattr(good, "canary"))
bad_evidence["receipt"] = {k: v for k, v in bad_evidence["receipt"].items()
if k != "provider_identity"}
request = _clone_with(good, canary=bad_evidence)
with pytest.raises(EvidenceProduceError, match="EVIDENCE_RECEIPT_INVALID"):
produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
def test_produce_enforces_source_tree_oid_binding_in_receipt(tmp_path: Path) -> None:
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
good, _, _ = build_input(tmp_path)
bad_evidence = dict(getattr(good, "rollback"))
bad_evidence["receipt"] = {**bad_evidence["receipt"], "source_oid": "f" * 40}
request = _clone_with(good, rollback=bad_evidence)
with pytest.raises(EvidenceProduceError, match="EVIDENCE_RECEIPT_SOURCE_OID_MISMATCH"):
produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
def test_produce_embeds_trusted_provider_binding_in_closure(tmp_path: Path) -> None:
# Positive: with valid receipts the closure is produced and the receipt.json
# carries the provider binding for each governed evidence type.
source_root = tmp_path / "source"
source_root.mkdir(mode=0o700)
request, candidate_id, tree_oid = build_input(tmp_path)
result = produce_evidence_closure(
builder_policy=POLICY_PATH, source_root=source_root, request=request,
expected_owner_uid=None,
)
assert result.status == "EVIDENCE_PRODUCED"
receipt = json.loads(
(source_root / request.version / "receipt.json").read_text(encoding="utf-8")
)
candidate = receipt["candidate"]
assert candidate["source_commit"] == request.source_commit
assert candidate["tree_oid"] == tree_oid
for etype in ("tests", "canary", "gitea", "rollback", "secret_scan"):
ev = candidate.get(etype)
if isinstance(ev, dict):
assert ev.get("passed") is True or ev.get("consistent") is True
@@ -86,7 +86,8 @@ class ExactPathPolicyTests(unittest.TestCase):
self.temp = tempfile.TemporaryDirectory()
self.root = Path(self.temp.name)
self.policy_dir = self.root / "policies"
self.policy_dir.mkdir(mode=0o750)
self.policy_dir.mkdir()
self.policy_dir.chmod(0o750)
self.policy_file = self.policy_dir / "writer.json"
self.workspace = self.root / "workspace"
self.workspace.mkdir(mode=0o700)
+117
View File
@@ -0,0 +1,117 @@
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT/"active-runtime"/"app"))
import release_approval_reopen as reopen
CID="rc-v2r2c2-4642b74af62c"; SRC="4"*40; TREE="a"*40
ART="5"*64; MAN="8"*64; PLAN="f"*64; POL="2"*64; FP="9"*64
NOW=datetime(2026,9,12,3,10,tzinfo=timezone.utc)
class Sagas:
def __init__(self, root): self.root=root
def compare_and_set(self, candidate_id, *, expected_states, target, event, details, expected_revision):
p=self.root/"sagas"/f"{candidate_id}.json"; x=json.loads(p.read_text())
if x["state"] not in expected_states: raise RuntimeError("CAS")
if int(x.get("revision",0)) != expected_revision: raise RuntimeError("SagaConflict")
x["state"]=target; x["revision"]=int(x.get("revision",0))+1
x.setdefault("events",[]).append({"event":event,"details":details,"timestamp":"2026-09-12T03:10:00Z"})
p.write_text(json.dumps(x,sort_keys=True)); return x
class Service:
def __init__(self, root): self.sagas=Sagas(root)
def w(p,o):
p.parent.mkdir(parents=True,exist_ok=True); p.write_text(json.dumps(o,sort_keys=True)); p.chmod(0o640)
_MISSING = object()
def state(tmp, expired=True, side_marker=_MISSING, saga="APPROVED", acceptance="ACCEPTED", extra=False):
r=tmp/"silent-release"
for d in ("candidates","sagas","release-plans","acceptance","approvals"):
(r/d).mkdir(parents=True,exist_ok=True,mode=0o750)
candidate={"candidate_id":CID,"state":"READY_FOR_APPROVAL","approval_state":"PENDING",
"source_commit":SRC,"tree_oid":TREE,"artifact_sha256":ART,"manifest_sha256":MAN}
if side_marker is not _MISSING:
candidate["side_effects_executed"]=side_marker
w(r/"candidates"/f"{CID}.json", candidate)
ev=[{"event":"saga_created","details":{}},{"event":"trusted_candidate_ready_for_approval","details":{}},
{"event":"signed_owner_approval_accepted","details":{"approval_fingerprint":FP}}]
if extra: ev.append({"event":"deployment_started","details":{}})
w(r/"sagas"/f"{CID}.json",{"candidate_id":CID,"state":saga,"revision":2,"journal_integrity":True,"events":ev})
w(r/"release-plans"/f"{CID}.json",{"sha256":PLAN,"policy_sha256":POL})
w(r/"acceptance"/f"{CID}.json",{"status":acceptance,"release_plan_sha256":PLAN})
w(r/"approvals"/f"{CID}.json",{"candidate_id":CID,"fingerprint":FP,"source_commit":SRC,"tree_oid":TREE,
"artifact_sha256":ART,"manifest_sha256":MAN,"release_plan_sha256":PLAN,"policy_sha256":POL,
"expires_at":"2026-09-12T02:59:05Z" if expired else "2026-09-12T04:59:05Z"})
return r
def req(key="reopen:test:0001"):
return {"schema":reopen.REQUEST_SCHEMA,"candidate_id":CID,"expected_approval_fingerprint":FP,
"expected_source_commit":SRC,"expected_tree_oid":TREE,"expected_artifact_sha256":ART,
"expected_manifest_sha256":MAN,"expected_release_plan_sha256":PLAN,"expected_policy_sha256":POL,
"reason":"expired approval after pre-side-effect infrastructure recovery","idempotency_key":key}
@pytest.fixture(autouse=True)
def enum(monkeypatch): monkeypatch.setattr(reopen,"_state",lambda x:x)
def test_happy(tmp_path):
r=state(tmp_path); old=(r/"approvals"/f"{CID}.json").read_bytes()
out=reopen.reopen_owner_approval(Service(r),req(),state_root=r,now=NOW)
assert out["status"]=="REOPENED"; s=json.loads((r/"sagas"/f"{CID}.json").read_text())
assert s["state"]=="READY_FOR_APPROVAL" and s["events"][-1]["event"]==reopen.EVENT_NAME
assert not (r/"approvals"/f"{CID}.json").exists()
assert (r/out["history_ref"]).read_bytes()==old
def test_idempotent(tmp_path):
r=state(tmp_path); svc=Service(r); a=reopen.reopen_owner_approval(svc,req(),state_root=r,now=NOW)
b=reopen.reopen_owner_approval(svc,req(),state_root=r,now=NOW)
assert b["status"]=="IDEMPOTENT" and a["approval_cycle_id"]==b["approval_cycle_id"]
@pytest.mark.parametrize("kind",["unexpired","saga","acceptance","side","advanced"])
def test_rejections(tmp_path,kind):
kw={}
expected=""
if kind=="unexpired": kw["expired"]=False; expected="APPROVAL_REOPEN_APPROVAL_NOT_EXPIRED"
if kind=="saga": kw["saga"]="READY_FOR_APPROVAL"; expected="APPROVAL_REOPEN_SAGA_STATE_INVALID"
if kind=="acceptance": kw["acceptance"]="REJECTED"; expected="APPROVAL_REOPEN_BINDING_MISMATCH"
if kind=="side": kw["side_marker"]=True; expected="APPROVAL_REOPEN_SIDE_EFFECTS_ALREADY_EXECUTED"
if kind=="advanced": kw["extra"]=True; expected="APPROVAL_REOPEN_SAGA_ALREADY_ADVANCED"
r=state(tmp_path,**kw)
with pytest.raises(reopen.ApprovalReopenError,match=expected):
reopen.reopen_owner_approval(Service(r),req(),state_root=r,now=NOW)
def test_binding_drift(tmp_path):
r=state(tmp_path); q=req(); q["expected_artifact_sha256"]="6"*64
with pytest.raises(reopen.ApprovalReopenError,match="APPROVAL_REOPEN_BINDING_MISMATCH"):
reopen.reopen_owner_approval(Service(r),q,state_root=r,now=NOW)
def test_history_collision(tmp_path):
r=state(tmp_path); h=r/"approval-history"/CID/f"{FP}.json"; h.parent.mkdir(parents=True); h.write_text("{}")
with pytest.raises(reopen.ApprovalReopenError,match="APPROVAL_REOPEN_HISTORY_COLLISION"):
reopen.reopen_owner_approval(Service(r),req(),state_root=r,now=NOW)
def test_real_durable_schema_without_side_effect_field_is_supported(tmp_path):
r=state(tmp_path)
candidate=json.loads((r/"candidates"/f"{CID}.json").read_text())
assert "side_effects_executed" not in candidate
out=reopen.reopen_owner_approval(
Service(r), req("reopen:test:real-schema"), state_root=r, now=NOW
)
assert out["status"]=="REOPENED"
@pytest.mark.parametrize("value",[True,None,1,"false"])
def test_explicit_non_false_side_effect_marker_fails_closed(tmp_path,value):
r=state(tmp_path,side_marker=value)
with pytest.raises(
reopen.ApprovalReopenError,
match="APPROVAL_REOPEN_SIDE_EFFECTS_ALREADY_EXECUTED"
):
reopen.reopen_owner_approval(
Service(r), req("reopen:test:marker:bad"), state_root=r, now=NOW
)
@@ -0,0 +1,64 @@
"""
Contract: release_approval_reopen must perform an atomic saga CAS that is
pinned to the saga's current revision.
Guards against the defect where `service.sagas.compare_and_set(...)` was called
without the required keyword-only `expected_revision` argument, which raised
TypeError -> APPROVAL_REOPEN_SAGA_CAS_FAILED (HTTP 409) and blocked the reopen.
"""
import re
from pathlib import Path
APP = Path(__file__).resolve().parents[1] / "active-runtime" / "app" / "release_approval_reopen.py"
MODULE = APP.name
CAS_CALL = re.compile(
r"service\.sagas\.compare_and_set\("
r"[^)]*?"
r"expected_revision\s*=\s*[^\s,]+"
r"[^)]*?\)",
re.S,
)
REVISION_SRC = re.compile(r"expected_revision\s*=\s*(?:saga\.get\(\s*[\"']revision[\"']\s*\)|saga\s*\.\s*revision)")
STATE_CAS = re.compile(r"expected_states\s*=\s*\{_state\(\s*[\"']APPROVED[\"']\s*\)\}")
TARGET_RFA = re.compile(r"target\s*=\s*_state\(\s*[\"']READY_FOR_APPROVAL[\"']\s*\)")
def _read():
if not APP.exists():
raise AssertionError(f"{MODULE} not found at {APP}")
return APP.read_text(encoding="utf-8")
def test_compare_and_set_call_requires_expected_revision():
src = _read()
# Exactly one compare_and_set call site in reopen module
calls = CAS_CALL.findall(src)
assert len(calls) == 1, f"expected exactly 1 compare_and_set call in {MODULE}, found {len(calls)}"
# The call must pass expected_revision (required keyword-only arg)
assert "expected_revision=" in calls[0], (
f"{MODULE} calls compare_and_set without expected_revision -> "
"TypeError -> APPROVAL_REOPEN_SAGA_CAS_FAILED"
)
# The expected_revision must be derived from the saga's current revision
assert REVISION_SRC.search(calls[0]), (
"expected_revision must be pinned to the saga current revision "
"(e.g. saga.get('revision') or saga.revision) for a true atomic CAS"
)
def test_cas_semantics_not_weakened():
src = _read()
# Still transitions APPROVED -> READY_FOR_APPROVAL (no weakening of state CAS)
assert STATE_CAS.search(src), "expected_states={_state('APPROVED')} must be preserved"
assert TARGET_RFA.search(src), "target=_state('READY_FOR_APPROVAL') must be preserved"
def test_reopen_still_binds_all_fingerprints():
src = _read()
# The expected_* sha256 bindings must still be enforced by ReopenRequest
for field in ("expected_approval_fingerprint", "expected_source_commit",
"expected_tree_oid", "expected_artifact_sha256",
"expected_manifest_sha256", "expected_release_plan_sha256",
"expected_policy_sha256"):
assert field in src, f"{MODULE} no longer enforces {field}"
+75
View File
@@ -0,0 +1,75 @@
from __future__ import annotations
import ast
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
STAGING = ROOT / "active-runtime" / "app" / "aios_model_api_staging.py"
if not STAGING.is_file():
pytest.skip(
"shadow-gate contract runs after package is applied to an AIOS checkout",
allow_module_level=True,
)
def _literal_string_set(node: ast.AST) -> set[str]:
"""Extract string literals from a set or frozenset({...}) literal.
The governed codebase represents allow-lists as ``frozenset({...})``; a
plain ``{...}`` set literal is also accepted. Only string literal members
are allowed in the silent-release allow-list.
"""
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "frozenset":
if len(node.args) == 1:
node = node.args[0]
if not isinstance(node, ast.Set):
raise AssertionError("SILENT_RELEASE_POST_PATHS must remain a set literal")
values: set[str] = set()
for elt in node.elts:
if not isinstance(elt, ast.Constant) or not isinstance(elt.value, str):
raise AssertionError("silent-release allow-list must contain only string literals")
values.add(elt.value)
return values
def test_reopen_is_narrow_silent_release_post():
tree = ast.parse(STAGING.read_text(encoding="utf-8"))
allow = None
for node in tree.body:
if not isinstance(node, ast.Assign):
continue
if any(isinstance(t, ast.Name) and t.id == "SILENT_RELEASE_POST_PATHS" for t in node.targets):
allow = _literal_string_set(node.value)
break
assert allow is not None
assert "/reopenOwnerApproval" in allow
# Existing governed silent-release paths must stay present.
assert {
"/createReleaseCandidate",
"/createReleaseIntent",
"/submitSignedApproval",
"/submitOwnerApproval",
"/startDeployment",
}.issubset(allow)
def test_reopen_route_still_requires_release_approve():
import re
source = STAGING.read_text(encoding="utf-8")
# Locate the route handler (the `== "/reopenOwnerApproval"` comparison),
# independent of single/double-quote style on the allow-list entry.
m = re.search(r'==\s*["\']/reopenOwnerApproval["\']', source)
assert m is not None, "reopen handler route comparison not found"
tail = source[m.start() : m.start() + 1200]
assert 'self.need("release:approve")' in tail
assert "reopen_owner_approval(silent_release_service(), data)" in tail
def test_global_write_gate_is_not_enabled_by_source_patch():
source = STAGING.read_text(encoding="utf-8")
# v1.2 must not hard-code or synthesize a global write enable.
assert 'AIOS_GITOPS_WRITES_ENABLED=true' not in source
assert 'AIOS_GITOPS_WRITES_ENABLED="true"' not in source
+149
View File
@@ -622,6 +622,155 @@ def test_execution_identity_is_available_only_to_trusted_post_admission(
}
def test_owner_signature_grant_reaches_task_ttl_ceiling_not_envelope_window(
tmp_path: Path, policy: TaskAdmissionPolicy
):
# V2R2C2 TTL contract: the envelope is a *signing* window. Once the Owner
# signature is verified, the consumed envelope must NOT truncate the
# derived grant; the grant lifecycle is bounded by the task/grant policy
# ceiling (max_task_ttl_seconds = 3600), not the envelope's ~900s window.
now = 1_800_000_000
auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now)
port = FakeTaskPort()
svc = TaskAdmissionBootstrapService(
store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
authorization=auth, task_port=port, policy=policy, clock=lambda: now,
)
# Envelope window is short (600s); grant TTL ceiling is 3600s.
created = svc.create_envelope(request(expires_in_seconds=600))
payload = canonical_json_bytes(created["envelope"])
signature = base64.b64encode(
hashlib.sha256(payload + b"\0owner-key-1").digest()
).decode("ascii")
result = svc.authorize_and_admit(
{
"envelope_id": created["envelope"]["envelope_id"],
"authorization_kind": "owner_signature",
"owner_key_id": "owner-key-1",
"signature_base64": signature,
}
)
grant = port.grants[0]
# Grant must reach the policy ceiling (now + 3600), NOT envelope expiry
# (now + 600). This is the core V2R2C2 defect regression.
assert grant.expires_at == now + policy.max_task_ttl_seconds
assert grant.expires_at >= created["envelope"]["expires_at"]
assert result["authorization_kind"] == "owner_signature"
def test_envelope_boundary_now_equals_expiry_is_rejected(tmp_path: Path, policy: TaskAdmissionPolicy):
# Boundary: authorization at now == envelope.expires_at must be refused
# fail-closed (envelope window is expired at that instant).
now = [1_800_000_000]
auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now[0])
port = FakeTaskPort()
svc = TaskAdmissionBootstrapService(
store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
authorization=auth, task_port=port, policy=policy, clock=lambda: now[0],
)
envelope = svc.create_envelope(request(expires_in_seconds=60))["envelope"]
# Advance exactly to the envelope expiry boundary.
now[0] = envelope["expires_at"]
with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_ENVELOPE_EXPIRED"):
svc.authorize_and_admit(
{
"envelope_id": envelope["envelope_id"],
"authorization_kind": "delegation_epoch",
"delegation_epoch_id": auth.epoch.epoch_id,
}
)
assert port.calls == 0
def test_delegation_epoch_grant_clips_to_parent_epoch_when_shorter_than_task_ttl(
tmp_path: Path, policy: TaskAdmissionPolicy
):
# delegation_epoch: when the parent epoch expires before the task TTL
# ceiling, the grant must be clipped to the parent epoch expiry (never
# outlive the delegating authority).
now = 1_800_000_000
auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now)
# parent epoch expires far before the task TTL ceiling (3600s).
auth.epoch = OwnerDelegationEpoch(
**{**auth.epoch.as_public_dict(), "expires_at": now + 120}
)
port = FakeTaskPort()
svc = TaskAdmissionBootstrapService(
store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
authorization=auth, task_port=port, policy=policy, clock=lambda: now,
)
result = svc.authorize_and_admit(
{
"envelope_id": svc.create_envelope(request())["envelope"]["envelope_id"],
"authorization_kind": "delegation_epoch",
"delegation_epoch_id": auth.epoch.epoch_id,
}
)
assert port.grants[0].expires_at == now + 120
assert port.grants[0].expires_at < now + policy.max_task_ttl_seconds
assert result["task_id"] == TASK_ID
def test_replay_owner_signature_returns_idempotent_grant_without_new_effect(
tmp_path: Path, policy: TaskAdmissionPolicy
):
# Replay: the same envelope + Owner signature can only return the original
# (identical) grant — it must never mint a second, different grant or
# re-invoke the task port side effect.
now = 1_800_000_000
auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now)
port = FakeTaskPort()
svc = TaskAdmissionBootstrapService(
store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
authorization=auth, task_port=port, policy=policy, clock=lambda: now,
)
created = svc.create_envelope(request())
payload = canonical_json_bytes(created["envelope"])
signature = base64.b64encode(
hashlib.sha256(payload + b"\0owner-key-1").digest()
).decode("ascii")
body = {
"envelope_id": created["envelope"]["envelope_id"],
"authorization_kind": "owner_signature",
"owner_key_id": "owner-key-1",
"signature_base64": signature,
}
first = svc.authorize_and_admit(body)
replay = svc.authorize_and_admit(body)
assert first["idempotent"] is False
assert replay["idempotent"] is True
assert replay["grant_id"] == first["grant_id"]
assert port.calls == 1 # side effect exactly once
def test_parent_expiry_conflict_fails_closed_when_grant_would_escape_parent(
tmp_path: Path, policy: TaskAdmissionPolicy
):
# A grant must never outlive its parent epoch. Simulate a parent whose
# expiry would be violated if the code ignored the parent bound; the
# service must clip (not mint) or reject, never exceed the parent.
now = 1_800_000_000
auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now)
auth.epoch = OwnerDelegationEpoch(
**{**auth.epoch.as_public_dict(), "expires_at": now + 500}
) # parent expires before task ceiling (3600) and before envelope (600)
port = FakeTaskPort()
svc = TaskAdmissionBootstrapService(
store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
authorization=auth, task_port=port, policy=policy, clock=lambda: now,
)
result = svc.authorize_and_admit(
{
"envelope_id": svc.create_envelope(request())["envelope"]["envelope_id"],
"authorization_kind": "delegation_epoch",
"delegation_epoch_id": auth.epoch.epoch_id,
}
)
grant = port.grants[0]
assert grant.expires_at <= auth.epoch.expires_at # never escapes parent
assert result["task_id"] == TASK_ID
def test_status_is_read_only_and_reports_admitted_task(service):
svc, auth, _ = service
envelope_id = svc.create_envelope(request())["envelope"]["envelope_id"]
@@ -0,0 +1,341 @@
from __future__ import annotations
"""End-to-end bootstrap chain contract (V2R2C2 P0).
Proves the createGovernedTask -> TaskPrincipal bootstrap is acyclic:
mission authorization
-> envelope normalized & persisted
-> authorize_and_admit derives immutable grant (task side effect does NOT
require a pre-existing TaskPrincipal)
-> principal is recovered downstream ONLY from durable grant + a
CLAIMED/RUNNING authoritative Task
Contrary to a naive cycle, admission does NOT depend on a principal existing,
and principal issuance does NOT re-create admission. No test pre-inserts a
grant directly through the DB to fake success; every grant travels the public
create_envelope/authorize_and_admit flow (or the durable store read-back only
after the public flow persisted it).
"""
import hashlib
from pathlib import Path
import sys
import threading
import pytest
ROOT = Path(__file__).resolve().parents[1]
APP = ROOT / "active-runtime" / "app"
if str(APP) not in sys.path:
sys.path.insert(0, str(APP))
from aios_task_admission_bootstrap import ( # noqa: E402
OwnerDelegationEpoch,
TaskAdmissionBootstrapService,
TaskAdmissionError,
TaskAdmissionPolicy,
TaskAdmissionStore,
)
from aios_task_admission_live_adapters import ( # noqa: E402
AgentTaskAdmissionPort,
TaskAdmissionPrincipalProvider,
)
NOW = 1_800_000_000
BASE_OID = "9" * 40
EXECUTION_DIGEST = "a" * 64
TASK_ID = "bootstrap-chain-task-20260909"
BRANCH = "agent/bootstrap-chain-task-20260909"
IDEMPOTENCY = "bootstrap-chain-admission-20260909"
FILES = ["active-runtime/app/example.py", "docs/chain.md"]
CAPS = (
"governed.task.create",
"agent-task:control",
"task.principal.issue",
"writer.lease",
"governed.write.file",
"git.read",
)
def _policy() -> TaskAdmissionPolicy:
return TaskAdmissionPolicy(
allowed_capabilities=frozenset(CAPS),
max_file_count=16,
max_task_ttl_seconds=3600,
max_envelope_ttl_seconds=900,
max_risk="medium",
)
def _epoch(auth_policy: TaskAdmissionPolicy, *, expires_at: int | None = None) -> OwnerDelegationEpoch:
return OwnerDelegationEpoch(
epoch_id="epoch-bootstrap-chain-20260909",
owner_identity="owner",
trust_domain="aios.owner",
repository="aios/aios",
allowed_branch_prefixes=("agent/bootstrap-chain-",),
allowed_capabilities=frozenset(CAPS),
allowed_file_prefixes=("active-runtime", "docs"),
risk_ceiling="medium",
issued_at=NOW - 60,
not_before=NOW - 60,
expires_at=expires_at if expires_at is not None else NOW + 3600,
revocation_epoch=0,
policy_digest=auth_policy.normalized_policy_digest(),
)
class InMemoryTaskAPI:
"""A tiny durable Task side effect mirroring the official bridge port."""
def __init__(self):
self.tasks: dict[str, dict] = {}
self.closed: set[str] = set()
self._lock = threading.Lock()
def create(self, task_id: str, body: dict) -> None:
with self._lock:
if task_id in self.tasks:
return
self.tasks[task_id] = {**body, "task_id": task_id, "status": "QUEUED"}
def get(self, task_id: str) -> dict | None:
with self._lock:
return dict(self.tasks.get(task_id)) if task_id in self.tasks else None
def claim(self, task_id: str) -> None:
with self._lock:
if task_id in self.tasks:
self.tasks[task_id]["status"] = "CLAIMED"
self.closed.add(task_id)
def status(self, task_id: str) -> str:
with self._lock:
return self.tasks.get(task_id, {}).get("status", "UNKNOWN")
class OfficialPort(AgentTaskAdmissionPort):
"""Official bridge port bound to the in-memory durable Task API."""
def __init__(self, api: InMemoryTaskAPI):
self.api = api
super().__init__(_Transport(api), execution_plan_provider=lambda _grant: {
"external_job_id": "chain-job-20260909",
"repository": "aios/aios",
"target_branch": BRANCH,
"generation": 1,
"execution_digest": EXECUTION_DIGEST,
})
def verify_active_task(self, grant):
state = self.api.status(grant.task_id)
if state not in {"CLAIMED", "RUNNING"}:
raise TaskAdmissionError(
"TASK_PRINCIPAL_TASK_NOT_CLAIMED", 409, details={"state": state}
)
return (
{"task_id": grant.task_id, "state": state},
{
"task_id": grant.task_id,
"repository": grant.repository,
"target_branch": grant.target_branch,
"generation": grant.generation,
"external_job_id": "chain-job-20260909",
},
)
class _Transport:
def __init__(self, api: InMemoryTaskAPI):
self.api = api
def request(self, method, path, body=None):
task_id = path.rsplit("/", 1)[-1]
if method == "GET":
task = self.api.get(task_id)
if task is None:
raise _NotFound()
return dict(task)
if method == "POST":
self.api.create(body["task_id"], body)
return dict(self.api.get(body["task_id"]))
raise AssertionError(f"unexpected transport call {method} {path}")
class _NotFound(Exception):
status = 404
class Auth:
def __init__(self, epoch: OwnerDelegationEpoch):
self.epoch = epoch
def verify_delegation_epoch(self, epoch_id: str) -> OwnerDelegationEpoch:
if epoch_id != self.epoch.epoch_id:
raise TaskAdmissionError("UNKNOWN_EPOCH", 403)
return self.epoch
def verify_owner_signature(self, **_kwargs) -> str:
return "owner"
def _request(**overrides):
value = {
"idempotency_key": IDEMPOTENCY,
"task_id": TASK_ID,
"repository": "aios/aios",
"authority_base_oid": BASE_OID,
"target_branch": BRANCH,
"generation": 1,
"execution_digest": EXECUTION_DIGEST,
"requested_capabilities": list(CAPS),
"file_allowlist": list(FILES),
"risk": "medium",
"expires_in_seconds": 600,
}
value.update(overrides)
return value
def _build(tmp_path: Path):
policy = _policy()
epoch = _epoch(policy)
store = TaskAdmissionStore(tmp_path / "chain.sqlite3")
api = InMemoryTaskAPI()
port = OfficialPort(api)
svc = TaskAdmissionBootstrapService(
store=store,
authorization=Auth(epoch),
task_port=port,
policy=policy,
clock=lambda: NOW,
)
provider = TaskAdmissionPrincipalProvider(
store=store, authorization=Auth(epoch), task_port=port,
epoch_id=epoch.epoch_id, clock=lambda: NOW,
)
return policy, epoch, store, api, svc, provider
def test_mission_authorization_to_principal_claim_lease_is_acyclic(tmp_path):
# mission auth -> envelope persisted -> grant -> principal (after claim).
# Assert the full chain returns a writer-lease principal for the same task.
_, _, _, api, svc, provider = _build(tmp_path)
env = svc.create_envelope(_request())["envelope"]
assert svc.status(env["envelope_id"])["status"] == "PENDING_AUTHORIZATION"
receipt = svc.authorize_and_admit(
{
"envelope_id": env["envelope_id"],
"authorization_kind": "delegation_epoch",
"delegation_epoch_id": _epoch(_policy()).epoch_id,
}
)
assert receipt["task_id"] == TASK_ID
assert receipt["task_state"] == "QUEUED"
assert api.status(TASK_ID) == "QUEUED" # durable task side effect persisted
# No principal can be issued before the task is claimed (claim gate).
with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_TASK_NOT_CLAIMED"):
provider({"operation": "acquireWriterLease"}, {
"task_id": TASK_ID, "job_id": "chain-job-20260909",
"repository": "aios/aios", "target_branch": BRANCH,
"generation": 1, "base_oid": BASE_OID,
})
api.claim(TASK_ID)
principal = provider({"operation": "acquireWriterLease"}, {
"task_id": TASK_ID, "job_id": "chain-job-20260909",
"repository": "aios/aios", "target_branch": BRANCH,
"generation": 1, "base_oid": BASE_OID,
})
assert principal["schema"] == "aios.task-principal-server-attestation.v1"
assert principal["task_id"] == TASK_ID
assert principal["capability"] == "writer.lease"
assert principal["task_state"] == "CLAIMED"
assert provider.readiness()["ready"] is True
def test_no_principal_without_persisted_grant_bootstraps_fail_closed(tmp_path):
# No admission has ever happened for this task -> no grant -> the principal
# provider must refuse; it must NOT fabricate an authorization.
_, _, _, api, _, provider = _build(tmp_path)
with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_AUTHORIZATION_REQUIRED"):
provider({"operation": "acquireWriterLease"}, {
"task_id": "never-admitted-task",
"job_id": "job-x",
"repository": "aios/aios",
"target_branch": BRANCH,
"generation": 1,
})
def test_duplicate_and_concurrent_mission_authorization_create_one_grant(tmp_path):
from concurrent.futures import ThreadPoolExecutor
_, _, _, api, svc, provider = _build(tmp_path)
env = svc.create_envelope(_request())["envelope"]
body = {
"envelope_id": env["envelope_id"],
"authorization_kind": "delegation_epoch",
"delegation_epoch_id": _epoch(_policy()).epoch_id,
}
with ThreadPoolExecutor(max_workers=16) as pool:
results = list(pool.map(lambda _: svc.authorize_and_admit(body), range(16)))
# Exact idempotent replay -> one durable task, one grant.
assert {r["task_id"] for r in results} == {TASK_ID}
# Exactly one creator (idempotent=False) and the rest are replays
# (idempotent=True). ThreadPoolExecutor.map returns in *input* order, not
# completion order, so the single winner may land at any index; asserting
# on results[1:] would be a wrong ordering assumption. Verify the
# invariant directly instead (stronger and order-independent).
assert len([r for r in results if r["idempotent"] is True]) == 15
assert len([r for r in results if r["idempotent"] is not True]) == 1
assert api.status(TASK_ID) == "QUEUED"
# The (single) grant must recover by task id without duplicate.
recovered = svc.store.get_grant_by_task_id(TASK_ID)
assert recovered is not None
assert recovered.task_id == TASK_ID
def test_wrong_authority_oid_and_wrong_repository_rejected_at_envelope(tmp_path):
_, _, _, _, svc, _ = _build(tmp_path)
with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_REPOSITORY_REJECTED"):
svc.create_envelope(_request(repository="other/repo"))
# authority_base_oid must be a real 40-hex OID binding to parent authority.
with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_BASE_OID_INVALID"):
svc.create_envelope(_request(authority_base_oid="HEAD"))
def test_stale_generation_conflicts_instead_of_silently_succeeding(tmp_path):
_, _, _, _, svc, _ = _build(tmp_path)
svc.create_envelope(_request(generation=1))
# Same idempotency key with a different generation is a conflicting reuse.
with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_IDEMPOTENCY_CONFLICT"):
svc.create_envelope(_request(generation=2))
def test_epoch_grant_clips_and_never_escapes_parent_expiry(tmp_path):
policy = _policy()
epoch = _epoch(policy, expires_at=NOW + 300) # parent expires before task ceiling
store = TaskAdmissionStore(tmp_path / "clip.sqlite3")
api = InMemoryTaskAPI()
port = OfficialPort(api)
svc = TaskAdmissionBootstrapService(
store=store, authorization=Auth(epoch), task_port=port,
policy=policy, clock=lambda: NOW,
)
env = svc.create_envelope(_request())["envelope"]
svc.authorize_and_admit(
{
"envelope_id": env["envelope_id"],
"authorization_kind": "delegation_epoch",
"delegation_epoch_id": epoch.epoch_id,
}
)
grant = store.get_grant_by_task_id(TASK_ID)
assert grant is not None
assert grant.expires_at == NOW + 300
assert grant.expires_at < NOW + policy.max_task_ttl_seconds
@@ -1,6 +1,7 @@
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor
import os
from pathlib import Path
import sys
import threading
@@ -157,7 +158,24 @@ def test_process_death_releases_admission_lock(tmp_path):
"s=TaskAdmissionStore(Path(sys.argv[1]));"
"lock=s.admission_lock('death-envelope');lock.__enter__();os._exit(17)"
)
result = subprocess.run([sys.executable, "-c", script, str(database)], check=False)
# The child is a bare interpreter (sys.executable), not the pytest process,
# so it does not inherit the in-process sys.path that already includes
# active-runtime/app. Carry the app dir through PYTHONPATH so the child can
# import the module. Without this, the child dies on ModuleNotFoundError
# (returncode != 17) and the lock-release invariant is never actually
# probed. This is a test-harness fix only; it does not change behavior.
child_env = dict(os.environ)
child_env["PYTHONPATH"] = os.pathsep.join(
filter(
None,
[str(APP), child_env.get("PYTHONPATH", "")],
)
)
result = subprocess.run(
[sys.executable, "-c", script, str(database)],
check=False,
env=child_env,
)
assert result.returncode == 17
with TaskAdmissionStore(database).admission_lock("death-envelope", timeout_seconds=1):
pass
+4
View File
@@ -360,6 +360,10 @@ def test_readiness_is_independent_from_recipe_evidence_but_build_remains_fail_cl
server.handle_request(build, connection=object())
@pytest.mark.skipif(
sys.platform == "win32",
reason="POSIX evidence-root mode enforcement (COLLECTOR_EVIDENCE_ROOT_MODE_DENIED) is a Linux-native acceptance item; Windows chmod does not map group/world bits and the source gates the check on os.name != 'nt'. Verified on Linux live (V2R2C2 §10/§15).",
)
def test_readiness_keeps_evidence_root_permissions_strict_with_group_shared_authority(tmp_path) -> None:
config = TrustedCandidateBuilderConfig.from_policy(load_builder_policy(POLICY_PATH))
evidence_root = tmp_path / "evidence"
@@ -0,0 +1,135 @@
from __future__ import annotations
"""CAP_SYS_PTRACE three-state contract tests (V2R2C2 P0).
Covers the capability boundary that lets the trusted candidate builder read
the peer's /proc/<pid>/{cgroup,exe,cmdline} across uids:
State A (no CAP / cross-uid read denied): fail closed ->
cgroup /proc/<pid>/cgroup PermissionError(EACCES) -> PEER_CGROUP_UNAVAILABLE
exe /proc/<pid>/exe PermissionError(EACCES) -> PEER_PROCESS_UNAVAILABLE
cmdline /proc/<pid>/cmdline PermissionError(EACCES) -> PEER_PROCESS_UNAVAILABLE
State B (minimal CAP, legitimate peer): resolved, readiness READY.
State C (CAP present, illegitimate peer): DENIED — capability never bypasses
the identity policy.
State B/C identity enforcement is exercised by the existing
test_trusted_candidate_builder.py (readiness READY + *_DENIED).
State A here is modelled with a monkeypatched PermissionError(EACCES) rather
than by guessing a high pid: a missing CAP_SYS_PTRACE / cross-uid read yields
exactly PermissionError(errno.EACCES) from the kernel, so the resolvers must
fail closed against that concrete error on each of the three /proc views.
This removes the prior assume-a-large-pid heuristic, which could not
deterministically produce EACCES.
"""
import errno
import os
from pathlib import Path
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
APP = ROOT / "active-runtime" / "app"
if str(APP) not in sys.path:
sys.path.insert(0, str(APP))
import aios_trusted_candidate_builder as mod # noqa: E402
from aios_trusted_candidate_builder import ( # noqa: E402
BuilderError,
_resolve_cgroups,
_resolve_process_identity,
)
def _eacces() -> PermissionError:
return PermissionError(errno.EACCES, "cross-uid /proc read denied")
def test_state_a_cgroup_read_denied_fails_closed(monkeypatch: pytest.MonkeyPatch):
# /proc/<pid>/cgroup read raises PermissionError(EACCES) -> PEER_CGROUP_UNAVAILABLE.
real_read_text = Path.read_text
def denied_cgroup(self, *a, **k):
if str(self) == "/proc/7/cgroup":
raise _eacces()
return real_read_text(self, *a, **k)
monkeypatch.setattr(Path, "read_text", denied_cgroup)
with pytest.raises(BuilderError, match="PEER_CGROUP_UNAVAILABLE"):
_resolve_cgroups(7)
def test_state_a_exe_read_denied_fails_closed(monkeypatch: pytest.MonkeyPatch):
# /proc/<pid>/exe realpath raises PermissionError(EACCES) -> PEER_PROCESS_UNAVAILABLE.
def denied_realpath(path):
if str(path) == "/proc/7/exe":
raise _eacces()
return os.path.realpath(path)
monkeypatch.setattr(os.path, "realpath", denied_realpath)
with pytest.raises(BuilderError, match="PEER_PROCESS_UNAVAILABLE"):
_resolve_process_identity(7)
def test_state_a_cmdline_read_denied_fails_closed(monkeypatch: pytest.MonkeyPatch):
# /proc/<pid>/cmdline read_bytes raises PermissionError(EACCES)
# -> PEER_PROCESS_UNAVAILABLE (cmdline branch, distinct from exe branch).
real_read_bytes = Path.read_bytes
def denied_cmdline(self, *a, **k):
if str(self) == "/proc/7/cmdline":
raise _eacces()
return real_read_bytes(self, *a, **k)
monkeypatch.setattr(Path, "read_bytes", denied_cmdline)
with pytest.raises(BuilderError, match="PEER_PROCESS_UNAVAILABLE"):
_resolve_process_identity(7)
def test_captrace_unit_is_live_baseline_with_minimal_capability_only():
unit = (
ROOT / "deploy" / "systemd" / "aios-trusted-candidate-builder.service"
).read_text(encoding="utf-8")
assert "AmbientCapabilities=CAP_SYS_PTRACE" in unit
assert "CapabilityBoundingSet=CAP_SYS_PTRACE" in unit
# Minimum-capability contract: NoNewPrivileges stays and the capability
# set is the live root baseline (User=root / Group=root), no invented
# ProtectProc / ProcSubset / credential paths / socket ownership.
assert "NoNewPrivileges=true" in unit
assert "User=root" in unit
assert "Group=root" in unit
assert "RestrictAddressFamilies=AF_UNIX" in unit
assert "ProtectProc" not in unit
assert "ProcSubset" not in unit
# No wildcard capability declaration beyond the single CAP_SYS_PTRACE.
assert "CapabilityBoundingSet=CAP_" not in unit.replace(
"CapabilityBoundingSet=CAP_SYS_PTRACE", ""
)
assert "AmbientCapabilities=CAP_" not in unit.replace(
"AmbientCapabilities=CAP_SYS_PTRACE", ""
)
# No divergent non-root -ptrace unit is tracked.
assert not (ROOT / "deploy" / "systemd" / "aios-trusted-candidate-builder-ptrace.service").exists()
def test_captrace_security_document_marks_capability_high_sensitivity():
doc = (
ROOT / "docs" / "security" / "trusted-builder-captrace.md"
).read_text(encoding="utf-8")
assert "PEER_PROCESS_UNAVAILABLE" in doc
assert "PEER_CGROUP_UNAVAILABLE" in doc
assert "READY" in doc
for denied in ("PEER_IDENTITY_DENIED", "PEER_CGROUP_DENIED", "PEER_PROCESS_DENIED"):
assert denied in doc
# Capability must never bypass identity policy.
assert "must never bypass" in doc
# The instruction requires CAP_SYS_PTRACE be described as high-sensitivity:
# the doc must frame it as carrying inspection/debug authority and must NOT
# positively claim it "grants no control authority".
assert "high-sensitivity" in doc
assert "broadens the process's observation and debugging authority" in doc
assert "grants no control authority" not in doc
assert "grants no write or control authority" not in doc
@@ -0,0 +1,33 @@
from __future__ import annotations
import os
import sys
from pathlib import Path
from trusted_evidence_providers import _run, _tests_env
REPO_ROOT = Path(__file__).resolve().parents[1]
def test_tests_env_forwards_app_and_repo_root_on_path():
"""The tests provider must run the real suite under the documented
PYTHONPATH so subprocess-spawning tests (e.g. job-safety process-crash
recovery) can import runtime modules from active-runtime/app."""
env = _tests_env(REPO_ROOT)
parts = env["PYTHONPATH"].split(os.pathsep)
assert str(REPO_ROOT) in parts
assert str(REPO_ROOT / "active-runtime" / "app") in parts
def test_run_extra_env_reaches_pytest_child_process():
"""_run must forward an extra environment (PYTHONPATH) into the child so a
subprocess spawned by a test inherits the repo runtime path, not just the
in-process sys.path that conftest mutates."""
probe = [
sys.executable,
"-c",
"import os;print(os.environ.get('PROBE_EXTRA','__missing__'))",
]
result = _run(probe, extra_env={"PROBE_EXTRA": "v2r2c2-gatew-provider-harness"})
assert result.returncode == 0
assert result.stdout.strip() == "v2r2c2-gatew-provider-harness"
+268
View File
@@ -0,0 +1,268 @@
from __future__ import annotations
import hashlib
import hmac
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
import pytest
from trusted_provider_receipts import (
PROVIDER_RECEIPT_SCHEMA,
PROVIDER_RECEIPT_VERSION,
ProviderAuthenticationError,
TrustedProviderPolicy,
TrustedProviderReceiptResolver,
build_hmac_authentication,
canonical_json_bytes,
canonical_sha256,
)
SOURCE = "a" * 40
TREE = "b" * 40
NOW = datetime(2026, 9, 9, 5, 0, tzinfo=timezone.utc)
def _receipt(result: dict, **overrides):
payload = {
"schema": PROVIDER_RECEIPT_SCHEMA,
"version": PROVIDER_RECEIPT_VERSION,
"provider_identity": "aios.tests-runner",
"provider_operation": "tests.run",
"operation_run_id": "run-123",
"result_reference": "runs/run-123/junit.xml",
"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": "tests",
},
}
payload.update(overrides)
return payload
def _write_receipt(root: Path, ref: str, payload: dict, *, mode=0o600):
path = root / ref
path.parent.mkdir(parents=True, exist_ok=True)
# All provider-store directories are authenticated and non-writable.
cur = root
os.chmod(cur, 0o700)
for part in Path(ref).parts[:-1]:
cur = cur / part
os.chmod(cur, 0o700)
path.write_bytes(canonical_json_bytes(payload))
os.chmod(path, mode)
return path
def _policy(root: Path, **overrides):
values = dict(
evidence_kind="tests",
provider_identity="aios.tests-runner",
store_id="tests",
store_root=root,
allowed_operations=frozenset({"tests.run"}),
expected_owner_uid=os.getuid(),
allowed_directory_modes=frozenset({0o700}),
allowed_receipt_modes=frozenset({0o600}),
max_age_seconds=300,
)
values.update(overrides)
return TrustedProviderPolicy(**values)
def _resolver(root: Path, **policy_overrides):
return TrustedProviderReceiptResolver(
{"tests": _policy(root, **policy_overrides)},
now_fn=lambda: NOW,
)
def test_valid_root_owned_receipt_is_authenticated(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
result = {"passed": True, "summary": {"passed": 10, "failed": 0}}
_write_receipt(root, "run-123/receipt.json", _receipt(result))
resolved = _resolver(root).resolve_evidence(
"tests",
{"receipt_ref": "run-123/receipt.json"},
source_oid=SOURCE,
tree_oid=TREE,
)
assert resolved.result == result
assert resolved.provider_receipt["provider_identity"] == "aios.tests-runner"
assert resolved.provider_receipt["authentication"]["kind"] == "root-owned-store-v1"
assert len(resolved.provider_receipt["receipt_file_sha256"]) == 64
def test_caller_supplied_receipt_body_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
with pytest.raises(ProviderAuthenticationError, match="CALLER_SUPPLIED_PROVIDER_RECEIPT_DENIED"):
_resolver(root).resolve_evidence(
"tests",
{"receipt": _receipt({"passed": True})},
source_oid=SOURCE,
tree_oid=TREE,
)
@pytest.mark.parametrize("ref", ["/abs/receipt.json", "../escape.json", "a/../../escape.json"])
def test_absolute_or_traversal_receipt_ref_is_denied(tmp_path: Path, ref: str):
root = tmp_path / "store"
root.mkdir(mode=0o700)
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_REF_"):
_resolver(root).resolve("tests", ref, source_oid=SOURCE, tree_oid=TREE)
def test_symlink_receipt_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
target = root / "real.json"
target.write_text("{}", encoding="utf-8")
os.chmod(target, 0o600)
link = root / "receipt.json"
link.symlink_to(target.name)
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_SYMLINK_DENIED"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_world_writable_receipt_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
path = _write_receipt(root, "receipt.json", _receipt({"passed": True}), mode=0o666)
assert path.exists()
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_MODE_INVALID|PROVIDER_RECEIPT_WRITABLE_DENIED"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_world_writable_store_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
_write_receipt(root, "receipt.json", _receipt({"passed": True}))
os.chmod(root, 0o777)
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_STORE_MODE_INVALID|PROVIDER_STORE_WRITABLE_DENIED"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_wrong_owner_is_denied(monkeypatch, tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
path = _write_receipt(root, "receipt.json", _receipt({"passed": True}))
real_lstat = os.lstat
def fake_lstat(p):
st = real_lstat(p)
if Path(p) == path:
values = list(st)
values[4] = st.st_uid + 1
return os.stat_result(values)
return st
monkeypatch.setattr(os, "lstat", fake_lstat)
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_OWNER_MISMATCH"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_provider_identity_mismatch_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
_write_receipt(root, "receipt.json", _receipt({"passed": True}, provider_identity="evil"))
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_IDENTITY_MISMATCH"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_operation_not_allowlisted_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
_write_receipt(root, "receipt.json", _receipt({"passed": True}, provider_operation="shell.exec"))
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_OPERATION_DENIED"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_content_digest_mismatch_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
result = {"passed": True}
_write_receipt(root, "receipt.json", _receipt(result, content_sha256="0" * 64))
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_CONTENT_SHA256_MISMATCH"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
@pytest.mark.parametrize(
("field", "error"),
[("source_oid", "PROVIDER_SOURCE_OID_MISMATCH"), ("tree_oid", "PROVIDER_TREE_OID_MISMATCH")],
)
def test_source_or_tree_mismatch_is_denied(tmp_path: Path, field: str, error: str):
root = tmp_path / "store"
root.mkdir(mode=0o700)
_write_receipt(root, "receipt.json", _receipt({"passed": True}, **{field: "c" * 40}))
with pytest.raises(ProviderAuthenticationError, match=error):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_stale_receipt_is_denied(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
_write_receipt(root, "receipt.json", _receipt({"passed": True}, issued_at="2026-09-09T04:00:00+00:00"))
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_RECEIPT_EXPIRED"):
_resolver(root).resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
def test_request_body_cannot_override_authenticated_result(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
result = {"passed": True, "count": 10}
_write_receipt(root, "receipt.json", _receipt(result))
with pytest.raises(ProviderAuthenticationError, match="EVIDENCE_REQUEST_CONTENT_MISMATCH"):
_resolver(root).resolve_evidence(
"tests",
{"receipt_ref": "receipt.json", "passed": False, "count": 10},
source_oid=SOURCE,
tree_oid=TREE,
)
def test_hmac_authenticated_receipt_passes_and_bad_mac_fails(tmp_path: Path):
root = tmp_path / "store"
root.mkdir(mode=0o700)
key = b"test-only-secret"
key_id = "provider-tests-hmac"
result = {"passed": True, "summary": "ok"}
receipt = _receipt(result)
receipt["authentication"] = {"kind": "hmac-sha256-v1", "key_id": key_id}
receipt["authentication"] = build_hmac_authentication(
receipt,
key_id=key_id,
key=key,
)
_write_receipt(root, "receipt.json", receipt)
policy = _policy(
root,
authentication_kind="hmac-sha256-v1",
hmac_key_id=key_id,
)
resolver = TrustedProviderReceiptResolver(
{"tests": policy},
hmac_key_resolver=lambda requested: key if requested == key_id else b"",
now_fn=lambda: NOW,
)
resolved = resolver.resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
assert resolved.result["passed"] is True
bad = json.loads((root / "receipt.json").read_text(encoding="utf-8"))
bad["authentication"]["mac_hex"] = "0" * 64
(root / "receipt.json").write_bytes(canonical_json_bytes(bad))
os.chmod(root / "receipt.json", 0o600)
with pytest.raises(ProviderAuthenticationError, match="PROVIDER_HMAC_MISMATCH"):
resolver.resolve("tests", "receipt.json", source_oid=SOURCE, tree_oid=TREE)
+56
View File
@@ -0,0 +1,56 @@
from __future__ import annotations
import json
import py_compile
import sys
from pathlib import Path
root = Path(sys.argv[1]).resolve()
errors = []
required = [
root/"active-runtime/app/trusted_provider_receipts.py",
root/"active-runtime/app/governed_provider_auth_adapter.py",
root/"active-runtime/app/governed_authenticated_evidence_operation.py",
root/"tests/test_trusted_provider_receipts.py",
root/"tests/test_governed_provider_auth_adapter.py",
root/"tests/test_governed_authenticated_evidence_operation.py",
]
for p in required:
if not p.is_file():
errors.append(f"missing {p.relative_to(root)}")
else:
try:
py_compile.compile(str(p), doraise=True)
except Exception as exc:
errors.append(f"compile {p.relative_to(root)}: {exc}")
producer = root/"active-runtime/app/governed_release_evidence_producer.py"
runtime = root/"active-runtime/app/governed_evidence_runtime.py"
if producer.is_file():
text = producer.read_text(encoding="utf-8", errors="replace")
if "governed_authenticated_evidence_operation.produce_authenticated_evidence_closure" not in text:
errors.append("producer OperationDefinition is not wired to authenticated wrapper")
else:
errors.append("missing governed_release_evidence_producer.py")
if runtime.is_file():
text = runtime.read_text(encoding="utf-8", errors="replace")
if "TrustedProviderReceiptResolver" not in text:
errors.append("runtime does not construct/inject TrustedProviderReceiptResolver")
if "test_governed_release_evidence_producer" in text or "build_input" in text:
errors.append("runtime imports test helper evidence")
else:
errors.append("missing governed_evidence_runtime.py")
for smoke_name in ("producer_runtime_smoke.py", "producer_gate_w_smoke.py"):
for smoke in root.rglob(smoke_name):
text = smoke.read_text(encoding="utf-8", errors="replace")
if "test_governed_release_evidence_producer" in text or "build_input" in text:
errors.append(f"{smoke_name} imports unit-test evidence helpers")
if errors:
print("ROUND3_PROVIDER_AUTH_INTEGRATION=FAIL")
for e in errors:
print(" -", e)
raise SystemExit(1)
print("ROUND3_PROVIDER_AUTH_INTEGRATION=PASS")