103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
"""Fail-closed runtime adoption gate for delegated evidence."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from typing import Any, NoReturn
|
|
|
|
from . import delegation_evidence
|
|
|
|
|
|
class AdoptionError(ValueError):
|
|
"""Raised when a delegation declaration cannot be safely adopted."""
|
|
|
|
|
|
_HEX40 = re.compile(r"^[0-9a-f]{40}$")
|
|
_SAFE_MESSAGE = "delegation adoption rejected"
|
|
|
|
|
|
def _reject() -> NoReturn:
|
|
raise AdoptionError(_SAFE_MESSAGE)
|
|
|
|
|
|
def _canonical_abs(value: Any, *, nullable: bool = False) -> str | None:
|
|
if nullable and value is None:
|
|
return None
|
|
if not isinstance(value, str) or not value or "\x00" in value:
|
|
_reject()
|
|
if not os.path.isabs(value) or value != os.path.normpath(value):
|
|
_reject()
|
|
return value
|
|
|
|
|
|
def _nonempty(value: Any) -> str:
|
|
if not isinstance(value, str) or not value.strip() or "\x00" in value:
|
|
_reject()
|
|
return value
|
|
|
|
|
|
def _source(value: Any) -> dict[str, str]:
|
|
if not isinstance(value, dict) or value.get("kind") not in ("git_tree", "captured_worktree"):
|
|
_reject()
|
|
field = "revision" if value["kind"] == "git_tree" else "base_revision"
|
|
if set(value) != {"kind", field} or not isinstance(value[field], str) or not _HEX40.fullmatch(value[field]):
|
|
_reject()
|
|
return value
|
|
|
|
|
|
def _declaration(value: Any) -> dict[str, Any]:
|
|
if not isinstance(value, dict) or not isinstance(value.get("mode"), str):
|
|
_reject()
|
|
if value["mode"] == "none":
|
|
if set(value) != {"mode"}:
|
|
_reject()
|
|
return value
|
|
if value["mode"] != "delegated" or set(value) != {
|
|
"mode", "bundle_root", "context_path", "result_path", "git_dir"
|
|
}:
|
|
_reject()
|
|
_canonical_abs(value["bundle_root"])
|
|
_canonical_abs(value["context_path"])
|
|
_canonical_abs(value["result_path"])
|
|
_canonical_abs(value["git_dir"], nullable=True)
|
|
return value
|
|
|
|
|
|
def require_adoption(
|
|
declaration: Any, *, expected_source: Any, expected_stage: Any
|
|
) -> dict[str, Any] | None:
|
|
"""Require independently expected provenance before consuming delegation evidence.
|
|
|
|
Lexical path checks are followed by delegation_evidence's descriptor-relative
|
|
O_NOFOLLOW reads (including ancestors). This is fresh verification, not an
|
|
atomic consume transaction or a sandbox against concurrent source mutation.
|
|
"""
|
|
try:
|
|
expected = _source(expected_source)
|
|
stage = _nonempty(expected_stage)
|
|
declared = _declaration(declaration)
|
|
if declared["mode"] == "none":
|
|
return None
|
|
|
|
context = delegation_evidence._json_file(declared["context_path"])
|
|
result = delegation_evidence._json_file(declared["result_path"])
|
|
if not isinstance(context, dict) or not isinstance(result, dict):
|
|
_reject()
|
|
if context.get("source") != expected or context.get("stage") != stage:
|
|
_reject()
|
|
obligations = context.get("obligations")
|
|
claims = result.get("claims")
|
|
if not isinstance(obligations, list) or not obligations or not isinstance(claims, list) or not claims:
|
|
_reject()
|
|
|
|
verified = delegation_evidence.verify_evidence(
|
|
declared["bundle_root"], context, result, declared["git_dir"]
|
|
)
|
|
if not isinstance(verified, dict) or verified.get("status") != "pass":
|
|
_reject()
|
|
return verified
|
|
except AdoptionError:
|
|
raise
|
|
except Exception:
|
|
_reject()
|