feat(cases): verify browser case source against owner receipts
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
This commit is contained in:
parent
994d11c869
commit
c31d21ba7d
5 changed files with 497 additions and 1 deletions
|
|
@ -50,3 +50,48 @@ This ordinary command refuses:1.2 is acknowledged, not implemented. For source-o
|
|||
The Python API is load_type(default_root, exact_reference, version_roots={"1.2": explicit_root}). Each key must be an exact major.minor and each root a Path. CLI --version-root VERSION=PATH may repeat for DIFFERENT versions. A matching explicit version selects exactly one root; a missing/wrong/ambiguous source refuses without trying the default or another revision. Versions without a mapping continue using --types. No directory discovery, highest/latest selection, status promotion or automatic owner-adapter adoption occurs. Duplicate mappings (even identical), duplicate YAML fields, mismatched source ID/version, malformed/traversal references and selected symlinks refuse. Direct case validation also requires the case reference to match the supplied type ID/version.
|
||||
|
||||
An explicit root is source input, not an owner acknowledgment or grant. Callers retain exact source hashes/type/case/render in their own reviewed receipts. Existing two-argument load_type callers, including ops-mason, retain their default behavior until their owner deliberately adopts the optional input. Adapter/global provenance/native first-reference gates remain separate in NK-WP-0048-T06 and the existing ORGREF driver.
|
||||
|
||||
## Check existing owner evidence compatibility
|
||||
|
||||
The normal validator CLI now accepts `check-evidence` using the same exact type
|
||||
selection, declarations, validation and deterministic render as `render`:
|
||||
|
||||
```sh
|
||||
python3 tools/security-case-types/security_case_validator.py check-evidence \
|
||||
security-cases/forgejo-netkingdom-oidc.yaml --types security-case-types \
|
||||
--structure-request /path/to/mason-structure-request.json \
|
||||
--warden-journal /path/to/existing/journal.sqlite
|
||||
```
|
||||
|
||||
Normal admission still refuses the acknowledged 1.1 and 1.2 types. Explicit
|
||||
`--allow-unimplemented` permits source preparation, with the existing
|
||||
`--version-root 1.2=security-case-type-revisions/1.2` for that version. No owner
|
||||
protocol is automatically adopted, no journal is created, and no status changes.
|
||||
|
||||
This is the cross-owner **compatibility preflight**, not native acceptance. It
|
||||
consumes the existing `ops-mason.case-structure-request.v1` and reads latest rows
|
||||
from the existing `warden.case.step-journal.v1` SQLite journal in read-only mode.
|
||||
Mason source/request hashes, complete current structural inventory and structural
|
||||
projection must match the fresh render. Warden case/type and type-declared machine
|
||||
descriptor pins, owners, bounded histories, created-object scope and successful
|
||||
attempt/revocation fingerprints must match. Unknown or duplicate fields, source
|
||||
substitution and missing completed-attempt fingerprints refuse. Running, uncertain,
|
||||
failed or rolled-back outcomes retain their exact meanings; an earlier success
|
||||
never overrides the latest journal row. Generated-value paths require distinct
|
||||
parent-session and broker-grant fingerprint fields.
|
||||
|
||||
Exit 0 means these supplied artifacts are source-compatible; it never means the
|
||||
case is complete. No artifact is required solely to render missing gates. Output
|
||||
always has `native_complete: false`, `execution.permitted: false`, and explicit
|
||||
unresolved native gates. A Mason Request has no native Observation provenance;
|
||||
a Warden journal retains fingerprints rather than portable signed/native proof.
|
||||
This command deliberately offers no trust callback, native-success flag, dispatch
|
||||
or way to promote self-authored JSON into native evidence. Actual independent
|
||||
owner evidence admission, status-hash binding, accepted machine descriptors,
|
||||
delivery, human sign-in/consumer denial, revocation and created-only rollback
|
||||
remain NK-WP-0048-T06 owner integration work. It does not mark any type implemented.
|
||||
|
||||
Integration tests invoke actual sibling Mason `prepare` and Warden `Journal.save`
|
||||
with metadata fixtures only. They need those owner source checkouts, and existing
|
||||
type suites need the public KeyCape request file and provider declarations beside
|
||||
the checkout. `NET_KINGDOM_ROOT` selects the source root for the 1.2 package tests.
|
||||
|
|
|
|||
200
tools/security-case-types/security_case_evidence.py
Normal file
200
tools/security-case-types/security_case_evidence.py
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
"""Read-only compatibility checks for existing Mason and Warden evidence seams.
|
||||
|
||||
This checks real owner artifacts against a fresh NetKingdom render. These source
|
||||
protocols carry no portable authenticated native evidence. A matching request or
|
||||
journal can NEVER establish materialized/live/implemented or authorize dispatch.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sqlite3
|
||||
|
||||
import yaml
|
||||
|
||||
LIMIT = 262144
|
||||
DIGEST = re.compile(r"sha256:[0-9a-f]{64}\Z")
|
||||
ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,191}\Z")
|
||||
|
||||
|
||||
class EvidenceRefused(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def sha(source):
|
||||
return "sha256:" + hashlib.sha256(source).hexdigest()
|
||||
|
||||
|
||||
def read(path):
|
||||
path = Path(path)
|
||||
if path.is_symlink() or not path.is_file() or path.stat().st_size > LIMIT:
|
||||
raise EvidenceRefused("unsafe or oversized metadata source")
|
||||
return path.read_bytes()
|
||||
|
||||
|
||||
def _unique(pairs):
|
||||
result = {}
|
||||
for key, value in pairs:
|
||||
if key in result:
|
||||
raise EvidenceRefused("duplicate metadata field")
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
|
||||
def decode(source):
|
||||
if type(source) is not bytes or len(source) > LIMIT:
|
||||
raise EvidenceRefused("oversized metadata input")
|
||||
try:
|
||||
return json.loads(source, object_pairs_hook=_unique)
|
||||
except (ValueError, UnicodeError):
|
||||
raise EvidenceRefused("invalid metadata JSON") from None
|
||||
|
||||
|
||||
def check_structure(request, case_source, type_source, rendered):
|
||||
"""Consume ops-mason.case-structure-request.v1, not a new native flag."""
|
||||
fields = {"schema", "case", "case_digest", "type_ref", "type_digest",
|
||||
"render_digest", "plan_digest", "bundle_digest", "structural_digest",
|
||||
"reviewed_render_digest", "run_source_changed", "objects"}
|
||||
if type(request) is not dict or set(request) != fields:
|
||||
raise EvidenceRefused("unknown Mason structural request fields")
|
||||
if request["schema"] != "ops-mason.case-structure-request.v1":
|
||||
raise EvidenceRefused("unknown Mason structural request protocol")
|
||||
for field in fields & {"case_digest", "type_digest", "render_digest", "plan_digest",
|
||||
"bundle_digest", "structural_digest", "reviewed_render_digest"}:
|
||||
if type(request[field]) is not str or not DIGEST.fullmatch(request[field]):
|
||||
raise EvidenceRefused("malformed structural source pin")
|
||||
if (request["case"], request["case_digest"], request["type_ref"],
|
||||
request["type_digest"], request["render_digest"]) != (
|
||||
rendered["case"], sha(case_source), rendered["type"], sha(type_source), rendered["digest"]):
|
||||
raise EvidenceRefused("Mason request has stale case/type/render pins")
|
||||
if type(request["run_source_changed"]) is not bool or request["run_source_changed"] != (
|
||||
request["reviewed_render_digest"] != rendered["digest"]):
|
||||
raise EvidenceRefused("Mason run drift flag disagrees with pins")
|
||||
owners = {"openbao_policy": ("railiance-platform", "native-structure"),
|
||||
"openbao_role": ("railiance-platform", "native-structure"),
|
||||
"kv_path": ("railiance-platform", "layout-only"),
|
||||
"catalog_entry": ("ops-warden", "pointer-source")}
|
||||
expected = []
|
||||
for item in rendered["materialization"]:
|
||||
if item["kind"] not in owners:
|
||||
raise EvidenceRefused("unsupported structural owner kind")
|
||||
owner, proof = owners[item["kind"]]
|
||||
expected.append({"id": item["id"], "kind": item["kind"], "target": item["template"],
|
||||
"expected_digest": sha(json.dumps(item, sort_keys=True, separators=(",", ":")).encode()),
|
||||
"evidence_owner": owner, "proof": proof})
|
||||
if request["objects"] != expected:
|
||||
raise EvidenceRefused("Mason structural inventory differs from exact render")
|
||||
case = yaml.safe_load(case_source)
|
||||
acc = case["spec"].get("acceptance") or {}
|
||||
acceptance = ({"record": acc["record"], "at": acc.get("at"),
|
||||
"by": [item.get("role") for item in acc.get("by", [])]} if acc.get("record") else None)
|
||||
structural = {"case": rendered["case"], "type": rendered["type"],
|
||||
"acceptance": acceptance, "materialization": rendered["materialization"]}
|
||||
if request["structural_digest"] != sha(json.dumps(structural, sort_keys=True, separators=(",", ":")).encode()):
|
||||
raise EvidenceRefused("Mason structural projection pin differs from current source")
|
||||
# A request has neither Observation provenance nor StatusBinding authority.
|
||||
return {"request_digest": sha(json.dumps(request, sort_keys=True, separators=(",", ":")).encode()),
|
||||
"objects": len(expected), "run_source_changed": request["run_source_changed"],
|
||||
"native_materialization_verified": False}
|
||||
|
||||
|
||||
def read_journal(path):
|
||||
"""Read bounded Warden rows through SQLite read-only mode; never create a DB."""
|
||||
path = Path(path)
|
||||
if path.is_symlink() or not path.is_file() or path.stat().st_size > 16 * 1024 * 1024:
|
||||
raise EvidenceRefused("unsafe or oversized Warden journal")
|
||||
try:
|
||||
connection = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)
|
||||
try:
|
||||
connection.execute("PRAGMA query_only=ON")
|
||||
connection.execute("PRAGMA trusted_schema=OFF")
|
||||
rows = connection.execute(
|
||||
"SELECT seq,binding,payload FROM receipts ORDER BY seq LIMIT 1025").fetchall()
|
||||
finally:
|
||||
connection.close()
|
||||
if len(rows) > 1024:
|
||||
raise EvidenceRefused("Warden journal inventory exceeds limit")
|
||||
latest = {}
|
||||
for seq, binding, payload in rows:
|
||||
if type(seq) is not int or type(binding) is not str or not ID.fullmatch(binding):
|
||||
raise EvidenceRefused("invalid Warden journal identity")
|
||||
if type(payload) is not str:
|
||||
raise EvidenceRefused("invalid Warden journal payload")
|
||||
latest[binding] = decode(payload.encode())
|
||||
return tuple(latest.values())
|
||||
except sqlite3.Error:
|
||||
raise EvidenceRefused("invalid Warden journal database") from None
|
||||
|
||||
|
||||
def check_steps(rows, case_source, type_source, rendered):
|
||||
"""Validate latest durable Warden states; they remain local evidence only."""
|
||||
expected = {item["id"]: item for item in rendered["follow_through"]}
|
||||
fields = {"step_id", "owner", "pins", "allowed_created", "retry_safe", "receipts",
|
||||
"journal_schema", "evidence_digest", "session_revoked", "session_revocation_digest",
|
||||
"grant_revocation_digest", "attempt_id", "operation"}
|
||||
type_spec = yaml.safe_load(type_source)["spec"]
|
||||
descriptors = type_spec.get("execution_descriptors", {})
|
||||
seen, result = set(), []
|
||||
for row in rows:
|
||||
if type(row) is not dict or set(row) != fields or row["journal_schema"] != "warden.case.step-journal.v1":
|
||||
raise EvidenceRefused("unknown Warden journal receipt fields")
|
||||
step = row["step_id"]
|
||||
if type(step) is not str or step not in expected or step in seen or row["owner"] != expected[step]["owner"]:
|
||||
raise EvidenceRefused("unknown, duplicate or wrong-owner Warden step")
|
||||
seen.add(step)
|
||||
pins = row["pins"]
|
||||
if type(pins) is not dict or set(pins) != {"case", "case_type", "adapter", "catalog", "acceptance", "run_sheet"}:
|
||||
raise EvidenceRefused("unknown Warden source pins")
|
||||
if any(type(value) is not str or not DIGEST.fullmatch(value) for value in pins.values()):
|
||||
raise EvidenceRefused("malformed Warden source pin")
|
||||
if (pins["case"], pins["case_type"]) != (sha(case_source), sha(type_source)):
|
||||
raise EvidenceRefused("Warden receipt has stale case/type pins")
|
||||
if descriptors.get(step) != pins["adapter"]:
|
||||
raise EvidenceRefused("Warden receipt has no exact type-pinned machine descriptor")
|
||||
history = row["receipts"]
|
||||
if type(history) is not list or not 1 <= len(history) <= 128:
|
||||
raise EvidenceRefused("invalid Warden receipt history")
|
||||
outcomes = {"running", "verified-success", "verified-no-change", "failed-before-change",
|
||||
"failed-after-change", "uncertain", "rollback-running", "verified-rolled-back", "rollback-uncertain"}
|
||||
allowed = row["allowed_created"]
|
||||
if type(allowed) is not list or len(allowed) > 32 or len(set(allowed)) != len(allowed) or any(
|
||||
type(value) is not str or not ID.fullmatch(value) for value in allowed):
|
||||
raise EvidenceRefused("invalid Warden created-object scope")
|
||||
if type(row["retry_safe"]) is not bool:
|
||||
raise EvidenceRefused("invalid Warden retry policy")
|
||||
for number, receipt in enumerate(history, 1):
|
||||
if type(receipt) is not dict or set(receipt) != {"attempt", "outcome", "created"} or (
|
||||
type(receipt["attempt"]) is not int or receipt["attempt"] != number or
|
||||
receipt["outcome"] not in outcomes or type(receipt["created"]) is not list or
|
||||
any(type(v) is not str or v not in allowed for v in receipt["created"]) or
|
||||
len(set(receipt["created"])) != len(receipt["created"])):
|
||||
raise EvidenceRefused("invalid Warden receipt state")
|
||||
last = history[-1]["outcome"]
|
||||
if last in {"verified-success", "verified-no-change", "verified-rolled-back"}:
|
||||
proofs = [row["evidence_digest"], row["session_revocation_digest"]]
|
||||
if expected[step]["value_handling"] == "generated_in_custody":
|
||||
proofs.append(row["grant_revocation_digest"])
|
||||
if row["session_revoked"] is not True or any(
|
||||
type(value) is not str or not DIGEST.fullmatch(value) for value in proofs):
|
||||
raise EvidenceRefused("Warden completed state lacks revocation fingerprints")
|
||||
if type(row["attempt_id"]) is not str or not ID.fullmatch(row["attempt_id"]) or row["operation"] not in {"run", "rollback", "reconcile"}:
|
||||
raise EvidenceRefused("Warden completed state lacks exact attempt identity")
|
||||
result.append({"step": step, "outcome": last, "native_authenticated": False})
|
||||
return result, sorted(set(expected) - seen)
|
||||
|
||||
|
||||
def audit(case_source, type_source, rendered, *, structure=None, journal_rows=()):
|
||||
structural = check_structure(structure, case_source, type_source, rendered) if structure is not None else None
|
||||
steps, missing = check_steps(journal_rows, case_source, type_source, rendered)
|
||||
type_spec = yaml.safe_load(type_source)["spec"]
|
||||
return {"schema": "netkingdom.case-evidence-compatibility.v1",
|
||||
"case_digest": sha(case_source), "type_digest": sha(type_source), "render_digest": rendered["digest"],
|
||||
"structure": structural, "steps": steps, "missing_follow_through": missing,
|
||||
"required_verification": [item["id"] for item in rendered["verification"]],
|
||||
"missing_machine_descriptors": [item["id"] for item in rendered["follow_through"]
|
||||
if item["id"] not in type_spec.get("execution_descriptors", {})],
|
||||
"unresolved_native_gates": ["authenticated-owner-evidence-admission", "accepted-to-materialized-status-binding",
|
||||
"human-sign-in", "consumer-group-denial", "parent-session-and-broker-grant-revocation", "created-only-rollback"],
|
||||
"native_complete": False, "execution": {"permitted": False}}
|
||||
|
|
@ -630,6 +630,15 @@ def main(argv: list[str] | None = None) -> int:
|
|||
rd.add_argument("--version-root", action="append", default=[], metavar="VERSION=PATH",
|
||||
help="select an exact version root; no search or fallback")
|
||||
rd.add_argument("--allow-unimplemented", action="store_true")
|
||||
ev = sub.add_parser("check-evidence", help="check existing Mason/Warden source artifacts; never native acceptance")
|
||||
ev.add_argument("case")
|
||||
ev.add_argument("declarations", nargs="*")
|
||||
ev.add_argument("--repos-root", default=str(TOOL_DIR.parents[2]))
|
||||
ev.add_argument("--types", required=True)
|
||||
ev.add_argument("--version-root", action="append", default=[], metavar="VERSION=PATH")
|
||||
ev.add_argument("--allow-unimplemented", action="store_true")
|
||||
ev.add_argument("--structure-request", help="ops-mason structure-request JSON")
|
||||
ev.add_argument("--warden-journal", help="existing Warden journal.sqlite; opened read-only")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.cmd == "validate-type":
|
||||
|
|
@ -644,7 +653,7 @@ def main(argv: list[str] | None = None) -> int:
|
|||
selection_issues, version_roots = parse_version_roots(args.version_root)
|
||||
if has_errors(selection_issues):
|
||||
return _report(selection_issues)
|
||||
issues, type_data, _ = load_type(Path(args.types), (case.get("spec") or {}).get("type", ""),
|
||||
issues, type_data, selected_type_dir = load_type(Path(args.types), (case.get("spec") or {}).get("type", ""),
|
||||
version_roots=version_roots)
|
||||
if type_data is None:
|
||||
return _report(issues)
|
||||
|
|
@ -663,6 +672,23 @@ def main(argv: list[str] | None = None) -> int:
|
|||
issues = load_issues + issues
|
||||
if rendered is None:
|
||||
return _report(issues)
|
||||
if args.cmd == "check-evidence":
|
||||
import security_case_evidence as evidence
|
||||
try:
|
||||
source = evidence.read(args.case)
|
||||
type_source = evidence.read(selected_type_dir / "type.yaml")
|
||||
# Refuse source changes between ordinary validation/render and audit.
|
||||
if yaml.safe_load(source) != case or yaml.safe_load(type_source) != type_data:
|
||||
raise evidence.EvidenceRefused("case/type source changed during validation")
|
||||
structure = evidence.decode(evidence.read(args.structure_request)) if args.structure_request else None
|
||||
rows = evidence.read_journal(args.warden_journal) if args.warden_journal else ()
|
||||
report = evidence.audit(source, type_source, rendered, structure=structure, journal_rows=rows)
|
||||
except (evidence.EvidenceRefused, OSError, ValueError, TypeError, KeyError):
|
||||
print("ERROR evidence: incompatible or unsafe owner metadata (diagnostics withheld)", file=sys.stderr)
|
||||
return 1
|
||||
json.dump(report, sys.stdout, indent=2, sort_keys=True)
|
||||
print()
|
||||
return 0
|
||||
json.dump(rendered, sys.stdout, indent=2, sort_keys=True)
|
||||
print()
|
||||
return _report(issues)
|
||||
|
|
|
|||
200
tools/security-case-types/tests/test_case_evidence.py
Normal file
200
tools/security-case-types/tests/test_case_evidence.py
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
"""Real owner source API compatibility, with synthetic metadata only; no native evidence."""
|
||||
import json
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
TOOL = Path(__file__).resolve().parents[1]
|
||||
REPO = TOOL.parents[1]
|
||||
OWNERS = Path("/home/worsch")
|
||||
sys.path.insert(0, str(TOOL))
|
||||
import security_case_evidence as e
|
||||
import security_case_validator as v
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def source():
|
||||
case_path = REPO / "security-cases/forgejo-netkingdom-oidc.yaml"
|
||||
raw = case_path.read_bytes()
|
||||
case = yaml.safe_load(raw)
|
||||
issues, doc, folder = v.load_type(REPO / "security-case-types", case["spec"]["type"])
|
||||
paths, issues = v.discover_declarations(doc, OWNERS)
|
||||
declarations, issues = v.composer.load_declarations(paths)
|
||||
issues, rendered = v.render_case(case, doc, declarations, require_implemented=False)
|
||||
assert rendered is not None and not v.has_errors(issues)
|
||||
return raw, (folder / "type.yaml").read_bytes(), rendered
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mason(tmp_path, source):
|
||||
sys.path.insert(0, str(OWNERS / "ops-mason/src"))
|
||||
from ops_mason import case_structure
|
||||
root = OWNERS / "ops-mason"
|
||||
bundle = tmp_path / "bundle"
|
||||
shutil.copytree(root / "builds/case-forgejo-netkingdom-oidc", bundle)
|
||||
# Existing Mason source tests identify this historic bundle's wrong consumer.
|
||||
rel = "credential-change-requests/CCR-DRAFT-FORGEJO-NETKINGDOM-OIDC-FORGEJO.yaml"
|
||||
ccr = yaml.safe_load((bundle / rel).read_bytes())
|
||||
ccr["target"]["workload"] = "forgejo"
|
||||
(bundle / rel).write_text(yaml.safe_dump(ccr, sort_keys=False, width=110))
|
||||
manifest = json.loads((bundle / "manifest.json").read_bytes())
|
||||
manifest["files"][rel] = e.sha((bundle / rel).read_bytes()).split(":")[1]
|
||||
(bundle / "manifest.json").write_text(json.dumps(manifest))
|
||||
request = case_structure.prepare(REPO / "security-cases/forgejo-netkingdom-oidc.yaml",
|
||||
root / "plans/case-forgejo-netkingdom-oidc.md", bundle,
|
||||
nk_root=OWNERS / "net-kingdom", allow_unimplemented=True)
|
||||
return json.loads(json.dumps(request.as_dict()))
|
||||
|
||||
|
||||
def test_real_mason_request_recomputes_all_source_and_projection_pins(source, mason):
|
||||
result = e.audit(*source, structure=mason)
|
||||
assert result["structure"]["objects"] == 5
|
||||
assert result["native_complete"] is False
|
||||
assert result["structure"]["native_materialization_verified"] is False
|
||||
assert "mint-secret" in result["missing_follow_through"]
|
||||
assert "register-client" in result["missing_machine_descriptors"]
|
||||
assert "consumer-enforces-groups" in result["required_verification"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field", ["case_digest", "type_digest", "render_digest", "structural_digest"])
|
||||
def test_changed_source_or_projection_refused(source, mason, field):
|
||||
mason[field] = "sha256:" + "0" * 64
|
||||
with pytest.raises(e.EvidenceRefused):
|
||||
e.audit(*source, structure=mason)
|
||||
|
||||
|
||||
def test_rehashed_structural_inventory_cannot_add_role_or_mint(source, mason):
|
||||
mason["objects"][0]["target"] += "-sibling"
|
||||
with pytest.raises(e.EvidenceRefused):
|
||||
e.audit(*source, structure=mason)
|
||||
|
||||
|
||||
def test_structure_unknown_native_success_flag_is_refused(source, mason):
|
||||
mason["native_complete"] = True
|
||||
with pytest.raises(e.EvidenceRefused):
|
||||
e.audit(*source, structure=mason)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def warden(tmp_path, source):
|
||||
sys.path.insert(0, str(OWNERS / "ops-warden/src"))
|
||||
from warden.case_run import Invocation, Journal, OwnerResult
|
||||
from warden.case_receipt_state import Outcome, Pins, Receipt, StepState
|
||||
raw, type_raw, rendered = source
|
||||
# Test-only migration: current acknowledged real type has no descriptors.
|
||||
doc = yaml.safe_load(type_raw)
|
||||
descriptor_digest = e.sha(b"fixture-descriptor")
|
||||
doc["spec"]["execution_descriptors"] = {"mint-secret": descriptor_digest}
|
||||
type_raw = yaml.safe_dump(doc).encode()
|
||||
pins = Pins(e.sha(raw), e.sha(type_raw), descriptor_digest, *([e.sha(b"fixture")] * 3))
|
||||
state = StepState("mint-secret", "railiance-platform", pins, (), False,
|
||||
(Receipt(1, Outcome.RUNNING), Receipt(2, Outcome.SUCCESS)))
|
||||
invocation = Invocation("mint-secret", "railiance-platform", "forgejo", "human", "fixture-human",
|
||||
"keycape-client-secret-generate", pins, "attempt-fixture", "run")
|
||||
receipt = OwnerResult(Outcome.SUCCESS, "fixture-evidence", (), True,
|
||||
"fixture-parent-revocation", "fixture-grant-revocation", "attempt-fixture", "run")
|
||||
journal = Journal(tmp_path / "private-journal")
|
||||
journal.save("mint-secret:forgejo", state, receipt, invocation)
|
||||
return (raw, type_raw, rendered), journal
|
||||
|
||||
|
||||
def test_actual_warden_journal_save_is_read_without_mutation_and_not_native_proof(warden):
|
||||
source, journal = warden
|
||||
path = journal.directory / "journal.sqlite"
|
||||
before = path.read_bytes()
|
||||
rows = e.read_journal(path)
|
||||
result = e.audit(*source, journal_rows=rows)
|
||||
assert path.read_bytes() == before
|
||||
assert result["steps"] == [{"step": "mint-secret", "outcome": "verified-success", "native_authenticated": False}]
|
||||
assert result["native_complete"] is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("field", ["session_revocation_digest", "grant_revocation_digest", "evidence_digest"])
|
||||
def test_missing_revocation_or_owner_fingerprint_refused(warden, field):
|
||||
source, journal = warden
|
||||
row = e.read_journal(journal.directory / "journal.sqlite")[0]
|
||||
row[field] = None
|
||||
with pytest.raises(e.EvidenceRefused):
|
||||
e.audit(*source, journal_rows=(row,))
|
||||
|
||||
|
||||
def test_unresolved_attempt_not_promoted_to_success(warden):
|
||||
source, journal = warden
|
||||
row = e.read_journal(journal.directory / "journal.sqlite")[0]
|
||||
row["receipts"][-1]["outcome"] = "uncertain"
|
||||
assert e.audit(*source, journal_rows=(row,))["steps"][0]["outcome"] == "uncertain"
|
||||
|
||||
|
||||
def test_no_descriptor_in_actual_acknowledged_type_refuses_journal_success(source, warden):
|
||||
_, journal = warden
|
||||
row = e.read_journal(journal.directory / "journal.sqlite")[0]
|
||||
row["pins"]["case_type"] = e.sha(source[1])
|
||||
with pytest.raises(e.EvidenceRefused, match="machine descriptor"):
|
||||
e.audit(*source, journal_rows=(row,))
|
||||
|
||||
|
||||
def test_missing_or_symlink_journal_never_created(tmp_path):
|
||||
path = tmp_path / "missing.sqlite"
|
||||
with pytest.raises(e.EvidenceRefused):
|
||||
e.read_journal(path)
|
||||
assert not path.exists()
|
||||
target = tmp_path / "target.sqlite"
|
||||
target.write_bytes(b"not sqlite")
|
||||
path.symlink_to(target)
|
||||
with pytest.raises(e.EvidenceRefused):
|
||||
e.read_journal(path)
|
||||
|
||||
|
||||
def test_duplicate_and_oversized_json_refused():
|
||||
for source in (b'{"schema":1,"schema":2}', b"x" * (e.LIMIT + 1)):
|
||||
with pytest.raises(e.EvidenceRefused):
|
||||
e.decode(source)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("version", ["1.1", "1.2"])
|
||||
def test_existing_cli_exact_source_selection_and_normal_admission_refusal(version):
|
||||
case = REPO / ("security-cases/forgejo-netkingdom-oidc.yaml" if version == "1.1" else
|
||||
"security-case-type-revisions/1.2/oidc-browser-relying-party/examples/forgejo.yaml")
|
||||
argv = [sys.executable, str(TOOL / "security_case_validator.py"), "check-evidence", str(case),
|
||||
"--types", str(REPO / "security-case-types"), "--repos-root", str(OWNERS),
|
||||
"--version-root", "1.2=" + str(REPO / "security-case-type-revisions/1.2")]
|
||||
refused = subprocess.run(argv, capture_output=True, text=True)
|
||||
assert refused.returncode == 1 and "only implemented types" in refused.stderr
|
||||
result = subprocess.run(argv + ["--allow-unimplemented"], capture_output=True, text=True)
|
||||
assert result.returncode == 0, result.stderr
|
||||
report = json.loads(result.stdout)
|
||||
assert report["native_complete"] is False and report["execution"]["permitted"] is False
|
||||
assert ("deliver-consumer-secret" in report["missing_follow_through"]) == (version == "1.2")
|
||||
|
||||
|
||||
def test_cli_consumes_actual_mason_request_and_withholds_rejected_payload(tmp_path, mason):
|
||||
request = tmp_path / "request.json"
|
||||
request.write_text(json.dumps(mason))
|
||||
args = [sys.executable, str(TOOL / "security_case_validator.py"), "check-evidence",
|
||||
str(REPO / "security-cases/forgejo-netkingdom-oidc.yaml"), "--types", str(REPO / "security-case-types"),
|
||||
"--repos-root", str(OWNERS), "--allow-unimplemented", "--structure-request", str(request)]
|
||||
result = subprocess.run(args, capture_output=True, text=True)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert json.loads(result.stdout)["structure"]["objects"] == 5
|
||||
mason["unexpected"] = "sensitive-fixture-diagnostic"
|
||||
request.write_text(json.dumps(mason))
|
||||
result = subprocess.run(args, capture_output=True, text=True)
|
||||
assert result.returncode == 1 and "sensitive-fixture-diagnostic" not in result.stderr + result.stdout
|
||||
|
||||
|
||||
def test_actual_journal_latest_uncertainty_does_not_replay_prior_success(warden):
|
||||
source, journal = warden
|
||||
from warden.case_receipt_state import Pins, Receipt, StepState, Outcome
|
||||
row = e.read_journal(journal.directory / "journal.sqlite")[0]
|
||||
history = tuple(Receipt(item["attempt"], Outcome(item["outcome"]), tuple(item["created"]))
|
||||
for item in row["receipts"])
|
||||
state = StepState(row["step_id"], row["owner"], Pins(**row["pins"]), (), False,
|
||||
history + (Receipt(3, Outcome.UNCERTAIN),))
|
||||
journal.save("mint-secret:forgejo", state)
|
||||
rows = e.read_journal(journal.directory / "journal.sqlite")
|
||||
assert len(rows) == 1
|
||||
assert e.audit(*source, journal_rows=rows)["steps"][0]["outcome"] == "uncertain"
|
||||
|
|
@ -559,3 +559,28 @@ docs/evidence/2026-10-11-explicit-type-version-selection/source-integration-vali
|
|||
FormalNetownerreview/reference-source integration and all existing native adapter,
|
||||
provenance/broker/materialization/attendedproof gates remain; T06 stayswait, no
|
||||
new residual or firstcase/typeimplementation/native grant is inferred.
|
||||
|
||||
|
||||
2026-10-16 — isolated source preparation for T06 (canonical integration pending):
|
||||
Added `check-evidence` to the existing validator CLI. It reuses exact version
|
||||
selection and ordinary type/case/render admission, checks actual Mason structural
|
||||
request source/inventory/projection pins and reads actual Warden journal rows
|
||||
read-only. Successful rows require exact type-declared descriptors and attempt/
|
||||
revocation fingerprints. Existing acknowledged types without machine descriptors
|
||||
remain plan-only. The compatibility result never claims authenticated native
|
||||
completion: Request and journal fingerprints are not portable native evidence.
|
||||
Owner authentication/status binding and actual native gates remain unresolved;
|
||||
T06/T07 statuses and both type/case lifecycles remain unchanged. No dispatch,
|
||||
credential access or authority expansion is introduced. See tool README for the
|
||||
existing owner protocol boundaries and source integration tests.
|
||||
|
||||
Concrete next Warden owner integration: current `Executor.admit` refuses every
|
||||
case parameter whose sensitivity is not public/operational, so browser
|
||||
`redirect_uris`, `allowed_scopes`, `client_type`, `mfa_required` (and 1.2
|
||||
`consumer_namespace`) fail after otherwise prepared admission. The sensitive
|
||||
classification is a security-change review level, not secret content. The
|
||||
Warden owner must validate through NetKingdom's existing exact-version case/type
|
||||
validator, then project only descriptor-declared safe target metadata; do not
|
||||
remove the credential/value refusal or copy complete parameters into journals.
|
||||
Actual owner descriptor acknowledgment and independent evidence admission remain
|
||||
prerequisites, with normal admission still refusing unimplemented types.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue