diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index cabae5a..cbdbea8 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -354,7 +354,7 @@ - _Depends: 8.9, 9.5_ - [ ] 10. Complete the single-writer v2 migration -- [ ] 10.1 Isolate every live v1 contract and writer behind the closed compatibility namespace +- [x] 10.1 Isolate every live v1 contract and writer behind the closed compatibility namespace - Move all authorization-shaped v1 values into one compatibility contract owner and all live v1 serialization into its sole writer facade. - Limit the compatibility namespace to the canonical primitive needed for byte stability and prevent every v2 module from importing it. - Preserve the frozen v1 wire representation while eliminating independent contract or serializer implementations elsewhere. diff --git a/src/openultrasast/core/_legacy_factpack.py b/src/openultrasast/core/_legacy_factpack.py index d06ad1c..6179801 100644 --- a/src/openultrasast/core/_legacy_factpack.py +++ b/src/openultrasast/core/_legacy_factpack.py @@ -1,59 +1,19 @@ -"""Compatibility facade for the pre-v2 file-backed fact-pack API.""" +"""Temporary explicit import shim for the pre-v2 fact-pack contract.""" -from __future__ import annotations +from openultrasast.core.legacy_v1.contracts import ( + FactPackError, + LegacyFactPack, + admit_fact_pack, + admit_pack_file, + load_pack_file, + verify_fact_pack, +) -import hashlib -import json -from collections.abc import Mapping, Sequence -from dataclasses import dataclass -from pathlib import Path -from typing import Any - - -class FactPackError(ValueError): - """A legacy fact pack is unreviewed or inconsistent.""" - - -@dataclass(frozen=True) -class LegacyFactPack: - family: str - facts: tuple[dict[str, Any], ...] - review_digest: str - fact_set_digest: str - - -def _digest(payload: object) -> str: - encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() - return hashlib.sha256(encoded).hexdigest()[:12] - - -def admit_fact_pack(family: str, facts: Sequence[Mapping[str, Any]], *, review_digest: str) -> LegacyFactPack: - if not review_digest: - raise FactPackError("unreviewed fact pack refused: no review digest") - normalized = tuple(dict(fact) for fact in facts) - return LegacyFactPack(family, normalized, review_digest, _digest([family, normalized])) - - -def verify_fact_pack(pack: LegacyFactPack, facts: Sequence[Mapping[str, Any]]) -> None: - normalized = tuple(dict(fact) for fact in facts) - if _digest([pack.family, normalized]) != pack.fact_set_digest: - raise FactPackError(f"fact content does not match the stamped digest of pack {pack.family!r}") - - -def load_pack_file(path: str | Path) -> LegacyFactPack: - payload = json.loads(Path(path).read_text(encoding="utf-8")) - review = payload.get("review", {}) - note = review.get("note", "") - if not note or review.get("digest") != hashlib.sha256(note.encode()).hexdigest(): - raise FactPackError(f"pack {path} lacks a verifiable review attestation") - facts = payload.get("facts") - if not isinstance(facts, list) or not all(isinstance(fact, dict) for fact in facts): - raise FactPackError(f"pack {path}: facts must be a list of objects") - return admit_fact_pack(str(payload.get("family", "")), facts, review_digest=review["digest"]) - - -def admit_pack_file(path: str | Path, *, review_digest: str) -> LegacyFactPack: - pack = load_pack_file(path) - if review_digest != pack.review_digest: - raise FactPackError(f"pack {path}: supplied review digest does not match the embedded attestation") - return pack +__all__ = [ + "FactPackError", + "LegacyFactPack", + "admit_fact_pack", + "admit_pack_file", + "load_pack_file", + "verify_fact_pack", +] diff --git a/src/openultrasast/core/admission.py b/src/openultrasast/core/admission.py index f82e397..09cacb8 100644 --- a/src/openultrasast/core/admission.py +++ b/src/openultrasast/core/admission.py @@ -1,137 +1,13 @@ -"""Admission contract for the detection core (Req 7, 10, 12). +"""Temporary explicit import shim for the pre-v2 admission contract.""" -Family-agnostic generalization of the prover's consume-only manifest validation: -only manifests produced and reviewed by the benchmark workflow are consumed, never -produced, mutated or re-labeled here. Refusal order is fixed and deterministic. -Reports must carry the claimed families so numbers are never read across families -the arm does not cover. -""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - -from openultrasast.tools.fact_init import FORBIDDEN_LABEL_MARKERS - - -class AdmissionManifestError(ValueError): - """Raised when an admission manifest is refused; the message names the reason.""" - - -@dataclass(frozen=True) -class AdmissionManifest: - """A validated, immutable admission manifest with its claimed families.""" - - produced_by: str - reviewed: bool - label_isolated: bool - spent: bool - population_role: str - analyzer_revision: str - fact_set_digest: str - acquisition_record_digest: str - configuration_digest: str - population_digest: str - labels: tuple[str, ...] - families: tuple[str, ...] = () - - -_REQUIRED_FIELDS = ( - "produced_by", - "reviewed", - "label_isolated", - "spent", - "population_role", - "analyzer_revision", - "fact_set_digest", - "acquisition_record_digest", - "configuration_digest", - "population_digest", - "labels", +from openultrasast.core.legacy_v1.contracts import ( + REPORT_SCHEMA_KEYS, + AdmissionManifest, + AdmissionManifestError, + validate_admission_manifest, + validate_report_schema, ) -_BENCHMARK_PRODUCER = "valid-detection-benchmark" - -# The fixed report schema every admission report must carry (Req 7.3, 10.2, 12.1). -REPORT_SCHEMA_KEYS = frozenset( - { - "manifest", - "families", - "pairs", - "pair_correctness", - "fix_recall", - "verified_location_recall", - "flagged_ordinary_per_100", - "precision", - "indeterminate_rate", - "coverage", - "runtime", - "instrument_failures", - "failed_gates", - "decision", - } -) - - -def validate_admission_manifest(payload: dict[str, object], *, expected_acquisition_digest: str | None = None) -> AdmissionManifest: - """Validate an externally produced admission manifest and return it, or refuse. - - Refusal reasons, in check order: malformed, producer, unreviewed, - label-isolated, label-exposure, acquisition-output-changed, spent. - """ - if not all(field in payload for field in _REQUIRED_FIELDS): - raise AdmissionManifestError("malformed: missing required manifest fields") - if payload["produced_by"] != _BENCHMARK_PRODUCER: - raise AdmissionManifestError(f"producer: only {_BENCHMARK_PRODUCER} manifests are consumed") - if payload["reviewed"] is not True: - raise AdmissionManifestError("unreviewed: manifest lacks external review") - if payload["label_isolated"] is not True: - raise AdmissionManifestError("label-isolated: isolation attestation missing") - raw_labels = payload["labels"] - if not isinstance(raw_labels, (tuple, list)) or any(marker in str(label) for label in raw_labels for marker in FORBIDDEN_LABEL_MARKERS): - raise AdmissionManifestError("label-exposure: manifest carries admission labels") - if expected_acquisition_digest is not None and payload["acquisition_record_digest"] != expected_acquisition_digest: - raise AdmissionManifestError("acquisition-output-changed: digest mismatch") - if payload["spent"] is True: - raise AdmissionManifestError("spent: population already consumed") - raw_families = payload.get("families", ()) - if not isinstance(raw_families, (tuple, list)): - raise AdmissionManifestError("malformed: families must be a sequence") - return AdmissionManifest( - produced_by=str(payload["produced_by"]), - reviewed=True, - label_isolated=True, - spent=False, - population_role=str(payload["population_role"]), - analyzer_revision=str(payload["analyzer_revision"]), - fact_set_digest=str(payload["fact_set_digest"]), - acquisition_record_digest=str(payload["acquisition_record_digest"]), - configuration_digest=str(payload["configuration_digest"]), - population_digest=str(payload["population_digest"]), - labels=tuple(str(label) for label in raw_labels), - families=tuple(str(family) for family in raw_families), - ) - - -def validate_report_schema(report: dict[str, Any]) -> None: - """Refuse an admission report missing any fixed-schema key (Req 12.1).""" - missing = sorted(REPORT_SCHEMA_KEYS - set(report)) - if missing: - raise AdmissionManifestError(f"report schema: missing keys {missing}") - for rate_key in ( - "pair_correctness", - "fix_recall", - "verified_location_recall", - "flagged_ordinary_per_100", - "precision", - "indeterminate_rate", - ): - rate = report[rate_key] - if not {"n", "count", "rate", "wilson95"} <= set(rate): - raise AdmissionManifestError(f"report schema: {rate_key} lacks n/count/rate/wilson95") - - __all__ = [ "REPORT_SCHEMA_KEYS", "AdmissionManifest", diff --git a/src/openultrasast/core/consumer.py b/src/openultrasast/core/consumer.py index c0784d8..cc7996c 100644 --- a/src/openultrasast/core/consumer.py +++ b/src/openultrasast/core/consumer.py @@ -1,55 +1,5 @@ -"""Consumer contracts for the detection core (Req 9, 11). - -The pre-push transaction consumes traces and verdicts through this interface within -its (measured) budget; the proof layer attaches demonstrated evidence to a verdict. -Every claim must carry a witness; every silence must carry a coverage reason. -""" - -from __future__ import annotations - -from collections.abc import Sequence -from dataclasses import dataclass - -from openultrasast.core.trace import TraceRecord -from openultrasast.core.verdict import BoundedSilence, ProvenClaim - - -@dataclass(frozen=True) -class PushRequest: - """One push transaction handed to the core.""" - - base: str - head: str - budget_class: str - - -@dataclass(frozen=True) -class PushVerdict: - """What the push consumer receives: proven claims plus bounded silence.""" - - claims: tuple[ProvenClaim, ...] - silence: tuple[BoundedSilence, ...] - digests: dict[str, str] - - -def consume_verdict( - request: PushRequest, - record: TraceRecord, - claims: Sequence[ProvenClaim], - silence: Sequence[BoundedSilence], -) -> PushVerdict: - """Validate and wrap one trace's verdict for the push consumer (Req 5, 11.1). - - A claim without a witness or a silence without a reason is refused: silence is - never a clean bill of health. - """ - for claim in claims: - if not claim.witness or not claim.path or claim.line <= 0: - raise ValueError(f"proven claim lacks witness or exact location: {claim}") - for entry in silence: - if not entry.reason: - raise ValueError(f"bounded silence lacks a coverage reason: {entry}") - return PushVerdict(claims=tuple(claims), silence=tuple(silence), digests=dict(record.digests)) +"""Temporary explicit import shim for the pre-v2 consumer contract.""" +from openultrasast.core.legacy_v1.contracts import PushRequest, PushVerdict, consume_verdict __all__ = ["PushRequest", "PushVerdict", "consume_verdict"] diff --git a/src/openultrasast/core/legacy_v1/__init__.py b/src/openultrasast/core/legacy_v1/__init__.py new file mode 100644 index 0000000..58ec318 --- /dev/null +++ b/src/openultrasast/core/legacy_v1/__init__.py @@ -0,0 +1,51 @@ +"""Explicit temporary exports for the closed detection-core v1 namespace.""" + +from openultrasast.core.legacy_v1.contracts import ( + REPORT_SCHEMA_KEYS, + AdmissionManifest, + AdmissionManifestError, + BoundedSilence, + FactPackError, + Guard, + Hop, + Judgment, + LegacyFactPack, + ProvenClaim, + PushRequest, + PushVerdict, + TraceRecord, + Witness, + admit_fact_pack, + admit_pack_file, + as_clean, + consume_verdict, + load_pack_file, + validate_admission_manifest, + validate_report_schema, + verify_fact_pack, +) + +__all__ = [ + "REPORT_SCHEMA_KEYS", + "AdmissionManifest", + "AdmissionManifestError", + "BoundedSilence", + "FactPackError", + "Guard", + "Hop", + "Judgment", + "LegacyFactPack", + "ProvenClaim", + "PushRequest", + "PushVerdict", + "TraceRecord", + "Witness", + "admit_fact_pack", + "admit_pack_file", + "as_clean", + "consume_verdict", + "load_pack_file", + "validate_admission_manifest", + "validate_report_schema", + "verify_fact_pack", +] diff --git a/src/openultrasast/core/legacy_v1/contracts.py b/src/openultrasast/core/legacy_v1/contracts.py new file mode 100644 index 0000000..c390bd1 --- /dev/null +++ b/src/openultrasast/core/legacy_v1/contracts.py @@ -0,0 +1,305 @@ +"""Closed temporary owner of detection-core v1 contracts and behavior.""" + +from __future__ import annotations + +import enum +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + + +class AdmissionManifestError(ValueError): + """A legacy admission manifest was refused.""" + + +class FactPackError(ValueError): + """A legacy fact pack is unreviewed or inconsistent.""" + + +class Judgment(enum.Enum): + UNGUARDED = "unguarded" + DISCHARGED = "discharged" + INDETERMINATE = "indeterminate" + + +@dataclass(frozen=True) +class ProvenClaim: + trace_id: str + witness: str + path: str + line: int + family: str + + +@dataclass(frozen=True) +class BoundedSilence: + path: str + reason: str + coverage: dict[str, Any] + + +@dataclass(frozen=True) +class Hop: + function: str + call_line: int | None + complete: bool + + +@dataclass(frozen=True) +class Witness: + kind: str + provenance: str + relation: str + + +@dataclass(frozen=True) +class Guard: + kind: str + effect: str + + +@dataclass(frozen=True) +class TraceRecord: + entry: str + hops: tuple[Hop, ...] + operation_kind: str + resource: str + authority: str + witnesses: tuple[Witness, ...] = () + guards: tuple[Guard, ...] = () + completeness_reasons: tuple[str, ...] = () + judgment: Judgment = Judgment.INDETERMINATE + digests: dict[str, str] = field(default_factory=dict) + + def record(self) -> dict[str, Any]: + from openultrasast.core.legacy_v1 import writer + + return writer.trace_record(self) + + def canonical(self) -> bytes: + from openultrasast.core.legacy_v1 import writer + + return writer.trace_bytes(self) + + +@dataclass(frozen=True) +class PushRequest: + base: str + head: str + budget_class: str + + +@dataclass(frozen=True) +class PushVerdict: + claims: tuple[ProvenClaim, ...] + silence: tuple[BoundedSilence, ...] + digests: dict[str, str] + + +@dataclass(frozen=True) +class AdmissionManifest: + produced_by: str + reviewed: bool + label_isolated: bool + spent: bool + population_role: str + analyzer_revision: str + fact_set_digest: str + acquisition_record_digest: str + configuration_digest: str + population_digest: str + labels: tuple[str, ...] + families: tuple[str, ...] = () + + +@dataclass(frozen=True) +class LegacyFactPack: + family: str + facts: tuple[dict[str, Any], ...] + review_digest: str + fact_set_digest: str + + +REPORT_SCHEMA_KEYS = frozenset( + { + "manifest", + "families", + "pairs", + "pair_correctness", + "fix_recall", + "verified_location_recall", + "flagged_ordinary_per_100", + "precision", + "indeterminate_rate", + "coverage", + "runtime", + "instrument_failures", + "failed_gates", + "decision", + } +) +_REQUIRED_MANIFEST_FIELDS = ( + "produced_by", + "reviewed", + "label_isolated", + "spent", + "population_role", + "analyzer_revision", + "fact_set_digest", + "acquisition_record_digest", + "configuration_digest", + "population_digest", + "labels", +) +_BENCHMARK_PRODUCER = "valid-detection-benchmark" +_FORBIDDEN_LABEL_MARKERS = ("vulnerability", "fix-location", "ordinary-change") + + +def as_clean(judgment: Judgment) -> Judgment: + if judgment is Judgment.INDETERMINATE: + raise ValueError("indeterminate never converts to a clean judgment") + return judgment + + +def consume_verdict( + request: PushRequest, + record: TraceRecord, + claims: Sequence[ProvenClaim], + silence: Sequence[BoundedSilence], +) -> PushVerdict: + del request + for claim in claims: + if not claim.witness or not claim.path or claim.line <= 0: + raise ValueError(f"proven claim lacks witness or exact location: {claim}") + for entry in silence: + if not entry.reason: + raise ValueError(f"bounded silence lacks a coverage reason: {entry}") + return PushVerdict(tuple(claims), tuple(silence), dict(record.digests)) + + +def validate_admission_manifest( + payload: dict[str, object], + *, + expected_acquisition_digest: str | None = None, +) -> AdmissionManifest: + if not all(field in payload for field in _REQUIRED_MANIFEST_FIELDS): + raise AdmissionManifestError("malformed: missing required manifest fields") + if payload["produced_by"] != _BENCHMARK_PRODUCER: + raise AdmissionManifestError(f"producer: only {_BENCHMARK_PRODUCER} manifests are consumed") + if payload["reviewed"] is not True: + raise AdmissionManifestError("unreviewed: manifest lacks external review") + if payload["label_isolated"] is not True: + raise AdmissionManifestError("label-isolated: isolation attestation missing") + raw_labels = payload["labels"] + if not isinstance(raw_labels, (tuple, list)): + raise AdmissionManifestError("label-exposure: manifest carries admission labels") + if any(marker in str(label) for label in raw_labels for marker in _FORBIDDEN_LABEL_MARKERS): + raise AdmissionManifestError("label-exposure: manifest carries admission labels") + if expected_acquisition_digest is not None and payload["acquisition_record_digest"] != expected_acquisition_digest: + raise AdmissionManifestError("acquisition-output-changed: digest mismatch") + if payload["spent"] is True: + raise AdmissionManifestError("spent: population already consumed") + raw_families = payload.get("families", ()) + if not isinstance(raw_families, (tuple, list)): + raise AdmissionManifestError("malformed: families must be a sequence") + return _manifest(payload, raw_labels, raw_families) + + +def _manifest(payload: dict[str, object], labels: Sequence[object], families: Sequence[object]) -> AdmissionManifest: + return AdmissionManifest( + produced_by=str(payload["produced_by"]), + reviewed=True, + label_isolated=True, + spent=False, + population_role=str(payload["population_role"]), + analyzer_revision=str(payload["analyzer_revision"]), + fact_set_digest=str(payload["fact_set_digest"]), + acquisition_record_digest=str(payload["acquisition_record_digest"]), + configuration_digest=str(payload["configuration_digest"]), + population_digest=str(payload["population_digest"]), + labels=tuple(str(label) for label in labels), + families=tuple(str(family) for family in families), + ) + + +def validate_report_schema(report: dict[str, Any]) -> None: + missing = sorted(REPORT_SCHEMA_KEYS - set(report)) + if missing: + raise AdmissionManifestError(f"report schema: missing keys {missing}") + for rate_key in ( + "pair_correctness", + "fix_recall", + "verified_location_recall", + "flagged_ordinary_per_100", + "precision", + "indeterminate_rate", + ): + if not {"n", "count", "rate", "wilson95"} <= set(report[rate_key]): + raise AdmissionManifestError(f"report schema: {rate_key} lacks n/count/rate/wilson95") + + +def admit_fact_pack(family: str, facts: Sequence[Mapping[str, Any]], *, review_digest: str) -> LegacyFactPack: + from openultrasast.core.legacy_v1 import writer + + if not review_digest: + raise FactPackError("unreviewed fact pack refused: no review digest") + normalized = tuple(dict(fact) for fact in facts) + return LegacyFactPack(family, normalized, review_digest, writer.fact_pack_digest(family, normalized)) + + +def verify_fact_pack(pack: LegacyFactPack, facts: Sequence[Mapping[str, Any]]) -> None: + from openultrasast.core.legacy_v1 import writer + + normalized = tuple(dict(fact) for fact in facts) + if writer.fact_pack_digest(pack.family, normalized) != pack.fact_set_digest: + raise FactPackError(f"fact content does not match the stamped digest of pack {pack.family!r}") + + +def load_pack_file(path: str | Path) -> LegacyFactPack: + from openultrasast.core.legacy_v1 import writer + + payload = writer.read_json(Path(path)) + review = payload.get("review", {}) + if not isinstance(review, dict): + raise FactPackError(f"pack {path} lacks a verifiable review attestation") + note = review.get("note", "") + if not isinstance(note, str) or not note or review.get("digest") != writer.text_digest(note): + raise FactPackError(f"pack {path} lacks a verifiable review attestation") + facts = payload.get("facts") + if not isinstance(facts, list) or not all(isinstance(fact, dict) for fact in facts): + raise FactPackError(f"pack {path}: facts must be a list of objects") + return admit_fact_pack(str(payload.get("family", "")), facts, review_digest=str(review["digest"])) + + +def admit_pack_file(path: str | Path, *, review_digest: str) -> LegacyFactPack: + pack = load_pack_file(path) + if review_digest != pack.review_digest: + raise FactPackError(f"pack {path}: supplied review digest does not match the embedded attestation") + return pack + + +__all__ = [ + "REPORT_SCHEMA_KEYS", + "AdmissionManifest", + "AdmissionManifestError", + "BoundedSilence", + "FactPackError", + "Guard", + "Hop", + "Judgment", + "LegacyFactPack", + "ProvenClaim", + "PushRequest", + "PushVerdict", + "TraceRecord", + "Witness", + "admit_fact_pack", + "admit_pack_file", + "as_clean", + "consume_verdict", + "load_pack_file", + "validate_admission_manifest", + "validate_report_schema", + "verify_fact_pack", +] diff --git a/src/openultrasast/core/legacy_v1/writer.py b/src/openultrasast/core/legacy_v1/writer.py new file mode 100644 index 0000000..b836478 --- /dev/null +++ b/src/openultrasast/core/legacy_v1/writer.py @@ -0,0 +1,41 @@ +"""The sole temporary detection-core v1 JSON serialization capability.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from dataclasses import asdict +from pathlib import Path +from typing import Any, cast + +from openultrasast.core.legacy_v1 import contracts + + +def trace_record(value: contracts.TraceRecord) -> dict[str, Any]: + payload: dict[str, Any] = asdict(value) + payload["judgment"] = value.judgment.value + return cast(dict[str, Any], json.loads(json.dumps(payload, sort_keys=True))) + + +def trace_bytes(value: contracts.TraceRecord) -> bytes: + return json.dumps(trace_record(value), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + + +def fact_pack_digest(family: str, facts: Sequence[Mapping[str, Any]]) -> str: + encoded = json.dumps([family, tuple(dict(fact) for fact in facts)], sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(encoded).hexdigest()[:12] + + +def text_digest(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def read_json(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise contracts.FactPackError(f"pack {path}: root must be an object") + return value + + +__all__ = ["fact_pack_digest", "read_json", "text_digest", "trace_bytes", "trace_record"] diff --git a/src/openultrasast/core/trace.py b/src/openultrasast/core/trace.py index b00d7d9..07ed8db 100644 --- a/src/openultrasast/core/trace.py +++ b/src/openultrasast/core/trace.py @@ -1,74 +1,5 @@ -"""Canonical trace schema for the detection core (Req 1). +"""Temporary explicit import shim for the pre-v2 trace contract.""" -One family-agnostic trace record per analyzed change, serialized deterministically: -identical inputs produce byte-identical canonical output. The access-control -prover's trace is the first implementation of this contract. -""" +from openultrasast.core.legacy_v1.contracts import Guard, Hop, TraceRecord, Witness -from __future__ import annotations - -import json -from dataclasses import asdict, dataclass, field -from typing import Any - -from openultrasast.core.verdict import Judgment - - -def _canonical_bytes(payload: dict[str, Any]) -> bytes: - return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() - - -@dataclass(frozen=True) -class Hop: - """One ordered hop with its call-target completeness.""" - - function: str - call_line: int | None - complete: bool - - -@dataclass(frozen=True) -class Witness: - """Evidence for or against a guard, with provenance and match relation.""" - - kind: str - provenance: str - relation: str - - -@dataclass(frozen=True) -class Guard: - """A guard effect observed on a path (kind plus effect).""" - - kind: str - effect: str - - -@dataclass(frozen=True) -class TraceRecord: - """The canonical security trace: entry, hops, operation, witnesses, guards, judgment, digests.""" - - entry: str - hops: tuple[Hop, ...] - operation_kind: str - resource: str - authority: str - witnesses: tuple[Witness, ...] = () - guards: tuple[Guard, ...] = () - completeness_reasons: tuple[str, ...] = () - judgment: Judgment = Judgment.INDETERMINATE - digests: dict[str, str] = field(default_factory=dict) - - def record(self) -> dict[str, Any]: - """JSON-ready dict with sorted keys; enums encode as their value strings.""" - payload: dict[str, Any] = asdict(self) - payload["judgment"] = self.judgment.value - record: dict[str, Any] = json.loads(json.dumps(payload, sort_keys=True)) - return record - - def canonical(self) -> bytes: - """Deterministic byte serialization (Req 1.2).""" - return _canonical_bytes(self.record()) - - -__all__ = ["Hop", "TraceRecord", "Witness"] +__all__ = ["Guard", "Hop", "TraceRecord", "Witness"] diff --git a/src/openultrasast/core/verdict.py b/src/openultrasast/core/verdict.py index 3f68de6..30f3e27 100644 --- a/src/openultrasast/core/verdict.py +++ b/src/openultrasast/core/verdict.py @@ -1,51 +1,5 @@ -"""Verdict vocabulary and verdict contract for the detection core (Req 2, 5, 9). - -Every analyzed operation is judged as exactly one of unguarded, discharged, or -indeterminate. A verdict is either a proven claim (witness attached, exact -location) or a bounded silence (coverage reason attached). Indeterminate never -converts to clean, and models may only judge residuals zero-shot over a trace. -""" - -from __future__ import annotations - -import enum -from dataclasses import dataclass -from typing import Any - - -class Judgment(enum.Enum): - """The three-way judgment vocabulary; value strings are the wire format.""" - - UNGUARDED = "unguarded" - DISCHARGED = "discharged" - INDETERMINATE = "indeterminate" - - -@dataclass(frozen=True) -class ProvenClaim: - """A defect claim carrying its witness and exact location (Req 5.1).""" - - trace_id: str - witness: str - path: str - line: int - family: str - - -@dataclass(frozen=True) -class BoundedSilence: - """An undecided site with its coverage reason; never a clean result (Req 5.2).""" - - path: str - reason: str - coverage: dict[str, Any] - - -def as_clean(judgment: Judgment) -> Judgment: - """Return a clean judgment, refusing to convert indeterminate (Req 2.1).""" - if judgment is Judgment.INDETERMINATE: - raise ValueError("indeterminate never converts to a clean judgment") - return judgment +"""Temporary explicit import shim for the pre-v2 verdict contract.""" +from openultrasast.core.legacy_v1.contracts import BoundedSilence, Judgment, ProvenClaim, as_clean __all__ = ["BoundedSilence", "Judgment", "ProvenClaim", "as_clean"] diff --git a/tests/core_migration_inventory.py b/tests/core_migration_inventory.py index cf13ed9..1937a4e 100644 --- a/tests/core_migration_inventory.py +++ b/tests/core_migration_inventory.py @@ -197,6 +197,11 @@ def load_migration_inventory(path: str | Path) -> LegacyMigrationInventory: def _defined_symbols(tree: ast.Module) -> set[str]: symbols = {""} + for node in tree.body: + if isinstance(node, ast.ImportFrom): + symbols.update(alias.asname or alias.name for alias in node.names) + elif isinstance(node, ast.Import): + symbols.update(alias.asname or alias.name.split(".", 1)[0] for alias in node.names) def visit(nodes: list[ast.stmt], parents: tuple[str, ...] = ()) -> None: for node in nodes: @@ -271,6 +276,9 @@ def _capability_index(tree: ast.Module) -> dict[str, set[str]]: constants = {node.value for node in ast.walk(tree) if isinstance(node, ast.Constant) and isinstance(node.value, str)} names = {node.id for node in ast.walk(tree) if isinstance(node, ast.Name)} names.update(node.attr for node in ast.walk(tree) if isinstance(node, ast.Attribute)) + names.update( + alias.asname or alias.name for node in ast.walk(tree) if isinstance(node, (ast.Import, ast.ImportFrom)) for alias in node.names + ) return { "call": {_call_leaf(node) for node in ast.walk(tree) if isinstance(node, ast.Call)}, "constant": constants, @@ -293,6 +301,16 @@ def _missing_anchors(entry: LegacyInventoryEntry, tree: ast.Module) -> tuple[str return tuple(missing) +def _explicit_legacy_shim(tree: ast.Module, symbol: str) -> bool: + root_symbol = symbol.split(".", 1)[0] + return any( + isinstance(node, ast.ImportFrom) + and (node.module or "").startswith("openultrasast.core.legacy_v1") + and (symbol == "" or any((alias.asname or alias.name) == root_symbol for alias in node.names)) + for node in tree.body + ) + + def validate_inventory_sources(inventory: LegacyMigrationInventory, *, root: Path) -> None: """Prove every registered migration target is readable, nonzero, and present.""" for entry in inventory.entries: @@ -307,10 +325,11 @@ def validate_inventory_sources(inventory: LegacyMigrationInventory, *, root: Pat tree = ast.parse(content, filename=entry.path) except (SyntaxError, UnicodeDecodeError) as exc: raise LegacyInventoryError(f"inventory source is not parseable: {entry.path}") from exc - if entry.symbol not in _defined_symbols(tree): + is_shim = _explicit_legacy_shim(tree, entry.symbol) + if entry.symbol not in _defined_symbols(tree) and not is_shim: raise LegacyInventoryError(f"inventory target is absent: {entry.path}:{entry.symbol}") missing = _missing_anchors(entry, tree) - if missing: + if missing and not is_shim: raise LegacyInventoryError(f"inventory capability is absent: {entry.path}:{missing}") diff --git a/tests/test_core_migration_namespace.py b/tests/test_core_migration_namespace.py new file mode 100644 index 0000000..22cbfb6 --- /dev/null +++ b/tests/test_core_migration_namespace.py @@ -0,0 +1,111 @@ +"""Task 10.1: isolate the live v1 contract and serialization owners.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] +V1_FIXTURES = REPO / "tests/fixtures/detection_core/v1" +PURE_COMPATIBILITY_MODULES = ( + "admission.py", + "consumer.py", + "trace.py", + "verdict.py", + "_legacy_factpack.py", +) + + +def _trace() -> object: + from openultrasast.core.legacy_v1 import contracts + + return contracts.TraceRecord( + entry="src/app.py::handler", + hops=(contracts.Hop(function="handler", call_line=10, complete=True),), + operation_kind="protected_read", + resource="credentials", + authority="admin", + witnesses=(contracts.Witness(kind="guard", provenance="fact:checkPermission", relation="all_paths"),), + guards=(contracts.Guard(kind="permission_check", effect="branches"),), + completeness_reasons=(), + judgment=contracts.Judgment.UNGUARDED, + digests={"input": "abc123"}, + ) + + +def test_legacy_contract_values_have_one_compatibility_owner() -> None: + from openultrasast.core.legacy_v1 import contracts + + names = { + "AdmissionManifest", + "AdmissionManifestError", + "BoundedSilence", + "FactPackError", + "Guard", + "Hop", + "Judgment", + "LegacyFactPack", + "ProvenClaim", + "PushRequest", + "PushVerdict", + "TraceRecord", + "Witness", + } + assert {getattr(contracts, name).__module__ for name in names} == {"openultrasast.core.legacy_v1.contracts"} + + +def test_pure_compatibility_modules_define_no_contract_or_serializer() -> None: + for name in PURE_COMPATIBILITY_MODULES: + path = REPO / "src/openultrasast/core" / name + content = path.read_bytes() + assert content, f"ownership instrument must read nonzero bytes from {name}" + tree = ast.parse(content, filename=path.as_posix()) + assert not [node.name for node in tree.body if isinstance(node, ast.ClassDef)] + assert not any(isinstance(node, (ast.Import, ast.ImportFrom)) and "json" in ast.unparse(node) for node in tree.body) + assert any(isinstance(node, ast.ImportFrom) and node.module == "openultrasast.core.legacy_v1.contracts" for node in tree.body) + + +def test_writer_is_the_only_compatibility_json_serializer() -> None: + namespace = REPO / "src/openultrasast/core/legacy_v1" + paths = sorted(namespace.glob("*.py")) + assert paths and all(path.read_bytes() for path in paths) + json_importers = [] + for path in paths: + tree = ast.parse(path.read_bytes(), filename=path.as_posix()) + if any(isinstance(node, (ast.Import, ast.ImportFrom)) and "json" in ast.unparse(node) for node in tree.body): + json_importers.append(path.name) + assert json_importers == ["writer.py"] + + +def test_namespace_and_v2_dependency_directions_are_closed() -> None: + namespace = REPO / "src/openultrasast/core/legacy_v1" + for path in namespace.glob("*.py"): + tree = ast.parse(path.read_bytes(), filename=path.as_posix()) + imports = {node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module} + assert all(module.startswith("openultrasast.core.legacy_v1") for module in imports if module.startswith("openultrasast")) + + compatibility = {*PURE_COMPATIBILITY_MODULES, "legacy.py", "__init__.py"} + for path in (REPO / "src/openultrasast/core").glob("*.py"): + if path.name in compatibility: + continue + content = path.read_bytes() + assert content, f"dependency instrument must read nonzero bytes from {path.name}" + tree = ast.parse(content, filename=path.as_posix()) + imports = {node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom) and node.module} + assert not any(module.startswith("openultrasast.core.legacy_v1") for module in imports) + + +def test_legacy_trace_write_remains_byte_identical() -> None: + expected = (V1_FIXTURES / "trace.json").read_bytes() + assert len(expected) == 401 + value = _trace() + assert value.canonical() == expected.rstrip(b"\n") # type: ignore[attr-defined] + + +def test_historical_decoder_stays_independent_from_compatibility_namespace() -> None: + path = REPO / "src/openultrasast/core/legacy.py" + content = path.read_bytes() + assert content + tree = ast.parse(content, filename=path.as_posix()) + imports = {node.module for node in ast.walk(tree) if isinstance(node, ast.ImportFrom)} + assert "openultrasast.core.legacy_v1" not in imports diff --git a/tests/test_core_migration_provenance.py b/tests/test_core_migration_provenance.py index d48277b..501daed 100644 --- a/tests/test_core_migration_provenance.py +++ b/tests/test_core_migration_provenance.py @@ -170,11 +170,22 @@ def test_independent_reader_audit_matches_split_inventory_capabilities() -> None inventory = load_migration_inventory(INVENTORY) reader_entries = tuple(entry for entry in inventory.entries if entry.dependency_kind.value == "reader") registered = {(entry.path, anchor) for entry in reader_entries for anchor in entry.capability_anchors} + current = set(report.capabilities) + open_registered = { + (entry.path, anchor) + for entry in reader_entries + if entry.disposition is MigrationDisposition.OPEN + for anchor in entry.capability_anchors + } assert report.files_read > 0 and report.bytes_read > 0 assert ("tests/test_core_probes.py", "top_level:ProbeResult") in report.capabilities assert all(len(entry.capability_anchors) == 1 for entry in reader_entries) - assert registered == set(report.capabilities) + assert current <= registered, "the frozen inventory may never grow to admit a new reader" + for path, _anchor in open_registered - current: + source = (REPO / path).read_bytes() + assert source, f"migration shim instrument must read nonzero bytes from {path}" + assert b"openultrasast.core.legacy_v1" in source, f"open reader disappeared without an explicit compatibility shim: {path}" def test_inventory_targets_require_exact_capability_anchors(tmp_path: Path) -> None: