mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-11-13 17:58:18 +00:00
feat(detection-core): isolate legacy v1 ownership
This commit is contained in:
Marius Ciepluch
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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"]
|
||||
@@ -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"]
|
||||
|
||||
@@ -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"]
|
||||
|
||||
Reference in New Issue
Block a user