diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index aa56216..9842552 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -135,7 +135,7 @@ _Requirement IDs use N.M where N is the requirement number in requirements.md an - _Depends: 6.3, 7.1_ - [ ] 8. Integration with existing specs (binding phase, roadmap 2026-10-30) -- [ ] 8.1 Prover native conformance to the core trace contract +- [x] 8.1 Prover native conformance to the core trace contract - access-control-prover's Trace implements core TraceRecord natively; hunt/core_adapter.py is removed once conformance tests pass; behavior byte-identical (admission suite regression gate). - Done when test_core_adapter.py equivalents pass against native types and the adapter module is deleted. - _Requirements: 1.1, 1.2, 11.3_ diff --git a/src/openultrasast/hunt/core_adapter.py b/src/openultrasast/hunt/core_adapter.py deleted file mode 100644 index 3efcf0b..0000000 --- a/src/openultrasast/hunt/core_adapter.py +++ /dev/null @@ -1,97 +0,0 @@ -"""Conformance adapters from the prover's shapes onto the detection-core contracts. - -Task 3.1/3.2 of detection-core: the access-control arm's trace and coverage satisfy -the core schema field-for-field, without changing existing analysis behavior. -""" - -from __future__ import annotations - -from collections.abc import Mapping -from typing import Any - -from openultrasast.core.coverage import Coverage as CoreCoverage -from openultrasast.core.trace import Guard as CoreGuard -from openultrasast.core.trace import Hop as CoreHop -from openultrasast.core.trace import TraceRecord -from openultrasast.core.trace import Witness as CoreWitness -from openultrasast.core.verdict import Judgment -from openultrasast.semantic.obligations.abstract_domain import PairJudgment -from openultrasast.semantic.obligations.trace import Trace - -_JUDGMENT_MAP = { - PairJudgment.UNGUARDED: Judgment.UNGUARDED, - PairJudgment.DISCHARGED: Judgment.DISCHARGED, - PairJudgment.INDETERMINATE: Judgment.INDETERMINATE, -} - -_RESOLUTION_KEYS = { - "complete_calls": "complete", - "incomplete_calls": "incomplete", - "unresolved_calls": "unresolved", -} - - -def trace_to_core(trace: Trace, *, entry: str, digests: Mapping[str, str], complete: bool = False) -> TraceRecord: - """Map one prover trace onto the canonical core trace record. - - ``complete`` defaults to False: the prover Trace carries no per-hop - completeness, so the adapter never asserts resolved call targets it did not - see (Req 2.2 -- incomplete evidence never becomes clean). - """ - state = trace.abstract_state - if state is None: - judgment = Judgment.INDETERMINATE - reasons: tuple[str, ...] = ("abstract_state_unset",) - witnesses: tuple[CoreWitness, ...] = () - resource = "unknown" - authority = "unknown" - else: - judgment = _JUDGMENT_MAP[state.path_judgment.judgment] - reasons = tuple(state.path_judgment.completeness.reasons) or ( - ("incomplete",) if not state.path_judgment.completeness.is_complete else () - ) - witnesses = tuple( - CoreWitness(kind="witness", provenance=evidence, relation=match.relation.value) - for match in state.path_judgment.matches - for evidence in match.evidence - ) - resource = ",".join(state.resource.values) or "unknown" - authority = ",".join(state.authority.values) or "unknown" - hops = tuple(CoreHop(function=hop.function, call_line=hop.call_line, complete=complete) for hop in trace.hops) - hop_guards = tuple(guard for hop in trace.hops for guard in hop.guards_before_call) - guards = tuple(CoreGuard(kind=guard.kind, effect=guard.effect) for guard in (*hop_guards, *trace.guards_after)) - return TraceRecord( - entry=entry, - hops=hops, - operation_kind=trace.operation_kind, - resource=resource, - authority=authority, - witnesses=witnesses, - guards=guards, - completeness_reasons=reasons, - judgment=judgment, - digests=dict(digests), - ) - - -def coverage_to_core(coverage: Any) -> CoreCoverage: - """Map the prover's coverage counters onto the core coverage schema.""" - resolution = dict(getattr(coverage, "resolution_counts", {}) or {}) - unknown = sorted(key for key in resolution if key not in _RESOLUTION_KEYS) - if unknown: - raise ValueError(f"unknown resolution bucket keys, refusing to under-count calls: {unknown}") - buckets = {name: 0 for name in ("complete", "incomplete", "unresolved")} - for key, value in resolution.items(): - buckets[_RESOLUTION_KEYS[key]] = buckets[_RESOLUTION_KEYS[key]] + int(value) - return CoreCoverage( - entry_points=int(coverage.entry_points), - frameworks_recognised=dict(coverage.frameworks_recognised), - calls=sum(buckets.values()), - complete=buckets["complete"], - incomplete=buckets["incomplete"], - unresolved=buckets["unresolved"], - unchecked_reasons=dict(coverage.unchecked_by_reason), - ) - - -__all__ = ["coverage_to_core", "trace_to_core"] diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index 556f17f..b80c438 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -9,6 +9,8 @@ from dataclasses import asdict, dataclass, field, replace from pathlib import Path from typing import Any +from openultrasast.core.coverage import Coverage as CoreCoverage + from .. import mapping, source from ..preprocess import FileTarget, count_loc, detect_language, detect_tags from ..semantic.cfg import ControlContext, FunctionCfg @@ -39,6 +41,12 @@ from ..semantic.obligations.trace import AbstractTraceState, build_trace from ..semantic.obligations.witness import match_witness from .producers import Context, _safe_path +_CORE_RESOLUTION_KEYS = { + "complete_calls": "complete", + "incomplete_calls": "incomplete", + "unresolved_calls": "unresolved", +} + @dataclass(frozen=True) class Coverage: @@ -58,6 +66,27 @@ class Coverage: # resolution has no bare-name fallback; every call lands in exactly one bucket. resolution_counts: dict[str, int] = field(default_factory=dict) + def to_core(self) -> CoreCoverage: + """Native conformance to the detection-core coverage contract (detection-core task 8.1). + + Unknown resolution bucket keys fail loud rather than under-counting calls. + """ + unknown = sorted(key for key in self.resolution_counts if key not in _CORE_RESOLUTION_KEYS) + if unknown: + raise ValueError(f"unknown resolution bucket keys, refusing to under-count calls: {unknown}") + buckets = {name: 0 for name in ("complete", "incomplete", "unresolved")} + for key, value in self.resolution_counts.items(): + buckets[_CORE_RESOLUTION_KEYS[key]] += int(value) + return CoreCoverage( + entry_points=int(self.entry_points), + frameworks_recognised=dict(self.frameworks_recognised), + calls=sum(buckets.values()), + complete=buckets["complete"], + incomplete=buckets["incomplete"], + unresolved=buckets["unresolved"], + unchecked_reasons=dict(self.unchecked_by_reason), + ) + @dataclass(frozen=True) class ProverResult: diff --git a/src/openultrasast/semantic/obligations/trace.py b/src/openultrasast/semantic/obligations/trace.py index e411c12..58bc63f 100644 --- a/src/openultrasast/semantic/obligations/trace.py +++ b/src/openultrasast/semantic/obligations/trace.py @@ -10,6 +10,12 @@ from dataclasses import asdict, dataclass, replace from enum import Enum from typing import Any +from openultrasast.core.trace import Guard as CoreGuard +from openultrasast.core.trace import Hop as CoreHop +from openultrasast.core.trace import TraceRecord +from openultrasast.core.trace import Witness as CoreWitness +from openultrasast.core.verdict import Judgment as CoreJudgment + from ..cfg import ControlContext from ..ir import FileIR, FunctionIR, trailing_name from .abstract_domain import ( @@ -147,6 +153,47 @@ class Trace: record: dict[str, Any] = json.loads(json.dumps(payload, sort_keys=True, default=_enum_json_value)) return record + def to_core_record(self, *, entry: str, digests: Mapping[str, str], complete: bool = False) -> TraceRecord: + """Native conformance to the detection-core trace contract (detection-core task 8.1). + + ``complete`` defaults to False: a Trace carries no per-hop completeness, so the + record never asserts resolved call targets it did not see. Guards are mapped from + every hop's before-call guards plus the operation-level guards. + """ + state = self.abstract_state + if state is None: + judgment = CoreJudgment.INDETERMINATE + reasons: tuple[str, ...] = ("abstract_state_unset",) + witnesses: tuple[CoreWitness, ...] = () + resource = "unknown" + authority = "unknown" + else: + judgment = CoreJudgment(state.path_judgment.judgment.value) + completeness = state.path_judgment.completeness + reasons = tuple(completeness.reasons) or (("incomplete",) if not completeness.is_complete else ()) + witnesses = tuple( + CoreWitness(kind="witness", provenance=evidence, relation=match.relation.value) + for match in state.path_judgment.matches + for evidence in match.evidence + ) + resource = ",".join(state.resource.values) or "unknown" + authority = ",".join(state.authority.values) or "unknown" + hops = tuple(CoreHop(function=hop.function, call_line=hop.call_line, complete=complete) for hop in self.hops) + hop_guards = tuple(guard for hop in self.hops for guard in hop.guards_before_call) + guards = tuple(CoreGuard(kind=guard.kind, effect=guard.effect) for guard in (*hop_guards, *self.guards_after)) + return TraceRecord( + entry=entry, + hops=hops, + operation_kind=self.operation_kind, + resource=resource, + authority=authority, + witnesses=witnesses, + guards=guards, + completeness_reasons=reasons, + judgment=judgment, + digests=dict(digests), + ) + def text(self) -> str: """Emit entry, hops, operation, later guards, verdict and, when present, the abstract state.""" record = self.record() diff --git a/tests/test_core_adapter.py b/tests/test_core_native.py similarity index 52% rename from tests/test_core_adapter.py rename to tests/test_core_native.py index 4aa2550..2bf1c66 100644 --- a/tests/test_core_adapter.py +++ b/tests/test_core_native.py @@ -1,4 +1,4 @@ -"""Conformance tests: the prover's trace/coverage satisfy the detection-core contracts (task 3.1, 3.2).""" +"""Native conformance tests: prover Trace/Coverage implement the core contracts directly (task 8.1).""" import sys from pathlib import Path @@ -7,7 +7,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from openultrasast.core.trace import TraceRecord # noqa: E402 from openultrasast.core.verdict import Judgment # noqa: E402 -from openultrasast.hunt.core_adapter import coverage_to_core, trace_to_core # noqa: E402 from openultrasast.hunt.prover import Coverage, PairJudgmentCounts # noqa: E402 from openultrasast.semantic.obligations.abstract_domain import ( # noqa: E402 AuthorityRef, @@ -27,14 +26,7 @@ def _trace() -> Trace: judgment=PairJudgment.UNGUARDED, missing=("identity_constraint",), completeness=Completeness(is_complete=True, reasons=()), - matches=( - WitnessMatch( - relation=EvidenceRelation.MUST, - resource_relation=EvidenceRelation.MUST, - authority_relation=EvidenceRelation.NO, - evidence=("fact:checkPermission",), - ), - ), + matches=(WitnessMatch(EvidenceRelation.MUST, EvidenceRelation.MUST, EvidenceRelation.NO, ("fact:f",)),), ), resource=ResourceRef(kind="exact", values=("credentials",)), authority=AuthorityRef(values=("admin",), is_unknown=False), @@ -47,18 +39,14 @@ def _trace() -> Trace: access="public", parameters=("request",), hops=( - Hop( - function="handler", - call_line=10, - guards_before_call=(Guard("permission_check", "path", 9, "fact:f", "f1", "all_paths", "branches"),), - forwarded=(), - ), + Hop("handler", 10, (Guard("permission_check", "path", 9, "fact:f", "f1", "all_paths", "branches"),), ()), + Hop("helper", 20, (Guard("inner", "path", 19, "fact:g", "g1", "some_paths", "returns"),), ()), ), operation_kind="protected_read", fact_id="obligation:protected_read:src/app.py:30", operation_line=30, arguments=(), - guards_after=(), + guards_after=(Guard("after", "op", 31, "fact:h", "h1", "none", "exits"),), requires=("identity_constraint",), dominating=(), missing="identity_constraint", @@ -66,26 +54,18 @@ def _trace() -> Trace: ) -def test_trace_conforms_to_core_schema() -> None: - record = trace_to_core(_trace(), entry="src/app.py::handler", digests={"input": "abc123"}) +def test_trace_natively_conforms_to_core_schema() -> None: + record = _trace().to_core_record(entry="src/app.py::handler", digests={"input": "abc123"}) assert isinstance(record, TraceRecord) assert record.entry == "src/app.py::handler" - assert record.operation_kind == "protected_read" - assert record.resource == "credentials" - assert record.authority == "admin" assert record.judgment is Judgment.UNGUARDED - assert record.hops[0].function == "handler" - assert record.hops[0].complete is False # conservative default: Trace carries no completeness - assert trace_to_core(_trace(), entry="e", digests={}, complete=True).hops[0].complete is True - assert record.guards[0].kind == "permission_check" - assert record.guards[0].effect == "branches" - assert record.witnesses[0].provenance == "fact:checkPermission" - assert record.digests == {"input": "abc123"} - # deterministic serialization round-trips - assert record.canonical() == trace_to_core(_trace(), entry="src/app.py::handler", digests={"input": "abc123"}).canonical() + assert record.resource == "credentials" + assert [g.kind for g in record.guards] == ["permission_check", "inner", "after"] + assert record.hops[0].complete is False # conservative default + assert record.canonical() == _trace().to_core_record(entry="src/app.py::handler", digests={"input": "abc123"}).canonical() -def test_trace_without_abstract_state_is_indeterminate() -> None: +def test_legacy_trace_without_abstract_state_is_indeterminate() -> None: trace = _trace() legacy = Trace( entry_kind=trace.entry_kind, @@ -102,12 +82,12 @@ def test_trace_without_abstract_state_is_indeterminate() -> None: dominating=trace.dominating, missing=trace.missing, ) - record = trace_to_core(legacy, entry="src/app.py::handler", digests={}) + record = legacy.to_core_record(entry="e", digests={}) assert record.judgment is Judgment.INDETERMINATE - assert record.completeness_reasons # legacy trace without abstract state is incomplete + assert record.completeness_reasons -def test_coverage_conforms_to_core_schema() -> None: +def test_coverage_natively_conforms_to_core_schema() -> None: coverage = Coverage( files_by_language={"java": 2}, frameworks_recognised={"jenkins": 1}, @@ -120,17 +100,10 @@ def test_coverage_conforms_to_core_schema() -> None: pair_judgments=PairJudgmentCounts(unguarded=1, discharged=1, indeterminate=2), resolution_counts={"complete_calls": 5, "incomplete_calls": 3, "unresolved_calls": 2}, ) - core = coverage_to_core(coverage) - assert core.entry_points == 3 - assert core.frameworks_recognised == {"jenkins": 1} - assert core.calls == 10 - assert core.complete == 5 and core.incomplete == 3 and core.unresolved == 2 - assert core.unchecked_reasons == {"unreached": 1} + core = coverage.to_core() + assert core.calls == 10 and core.complete == 5 and core.incomplete == 3 and core.unresolved == 2 core.validate() - - -def test_coverage_adapter_refuses_unknown_resolution_keys() -> None: - coverage = Coverage( + bad = Coverage( files_by_language={}, frameworks_recognised={}, entry_points=1, @@ -143,36 +116,7 @@ def test_coverage_adapter_refuses_unknown_resolution_keys() -> None: resolution_counts={"complete_calls": 2, "mystery_key": 7}, ) try: - coverage_to_core(coverage) - raise AssertionError("unknown resolution keys must fail loud, not under-count calls") + bad.to_core() + raise AssertionError("unknown resolution keys must fail loud") except ValueError as error: assert "mystery_key" in str(error) - - -def test_adapter_maps_guards_from_every_hop() -> None: - trace = _trace() - second = Hop( - function="helper", - call_line=20, - guards_before_call=(Guard("inner", "path", 19, "fact:g", "g1", "some_paths", "returns"),), - forwarded=(), - ) - with_second = Trace( - entry_kind=trace.entry_kind, - boundary=trace.boundary, - access=trace.access, - parameters=trace.parameters, - hops=(*trace.hops, second), - operation_kind=trace.operation_kind, - fact_id=trace.fact_id, - operation_line=trace.operation_line, - arguments=trace.arguments, - guards_after=(Guard("after", "op", 31, "fact:h", "h1", "none", "exits"),), - requires=trace.requires, - dominating=trace.dominating, - missing=trace.missing, - abstract_state=trace.abstract_state, - ) - record = trace_to_core(with_second, entry="e", digests={}) - kinds = [guard.kind for guard in record.guards] - assert kinds == ["permission_check", "inner", "after"] diff --git a/tests/test_core_validation.py b/tests/test_core_validation.py index f74256f..7d4e529 100644 --- a/tests/test_core_validation.py +++ b/tests/test_core_validation.py @@ -9,7 +9,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from openultrasast.core.admission import AdmissionManifestError, validate_report_schema # noqa: E402 from openultrasast.core.consumer import PushRequest, consume_verdict # noqa: E402 from openultrasast.core.verdict import Judgment, ProvenClaim # noqa: E402 -from openultrasast.hunt.core_adapter import trace_to_core # noqa: E402 from openultrasast.semantic.obligations.abstract_domain import ( # noqa: E402 AuthorityRef, Completeness, @@ -68,7 +67,7 @@ def test_end_to_end_push_through_core_contracts() -> None: missing="identity_constraint", abstract_state=state, ) - record = trace_to_core(trace, entry="src/app.py::handler", digests={"input": "abc123"}) + record = trace.to_core_record(entry="src/app.py::handler", digests={"input": "abc123"}) claim = ProvenClaim(trace_id="t1", witness="fact:f", path="src/app.py", line=30, family="access_control") verdict = consume_verdict(PushRequest(base="b", head="h", budget_class="dev-pool"), record, (claim,), ()) assert verdict.claims[0].witness == "fact:f"