From 292d4b5557ead5fca6dcc33b95d99ed12e7b368a Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 1 Nov 2026 12:13:20 +0100 Subject: [PATCH] feat(detection-core): adapt authorization traces --- .kiro/specs/detection-core/tasks.md | 2 +- .../semantic/obligations/core_adapter.py | 214 ++++++++++++++++++ .../semantic/obligations/trace.py | 47 ---- tests/test_core_native.py | 7 +- tests/test_core_validation.py | 3 +- tests/test_obligation_core_adapter.py | 89 ++++++++ 6 files changed, 310 insertions(+), 52 deletions(-) create mode 100644 src/openultrasast/semantic/obligations/core_adapter.py create mode 100644 tests/test_obligation_core_adapter.py diff --git a/.kiro/specs/detection-core/tasks.md b/.kiro/specs/detection-core/tasks.md index d780b28..54012ae 100644 --- a/.kiro/specs/detection-core/tasks.md +++ b/.kiro/specs/detection-core/tasks.md @@ -302,7 +302,7 @@ - _Depends: 8.2, 8.9_ - [ ] 9. Integrate adjacent producers and benchmark-owned orchestration -- [ ] 9.1 (P) Translate the parked authorization trace conservatively into v2 evidence +- [x] 9.1 (P) Translate the parked authorization trace conservatively into v2 evidence - Move conversion out of the family trace domain and keep authorization vocabulary outside shared core modules. - Map missing per-hop knowledge to incomplete relations and indeterminate outcomes without behavior lifting. - Preserve existing parked-arm regression behavior and avoid any admission rerun or reactivation. diff --git a/src/openultrasast/semantic/obligations/core_adapter.py b/src/openultrasast/semantic/obligations/core_adapter.py new file mode 100644 index 0000000..692cffa --- /dev/null +++ b/src/openultrasast/semantic/obligations/core_adapter.py @@ -0,0 +1,214 @@ +"""Conservative authorization adapters for detection-core migration boundaries.""" + +from __future__ import annotations + +from collections.abc import Mapping + +from openultrasast.core import evidence, family, semantics +from openultrasast.core import provenance as core_provenance +from openultrasast.core.trace import Guard as LegacyGuard +from openultrasast.core.trace import Hop as LegacyHop +from openultrasast.core.trace import TraceRecord +from openultrasast.core.trace import Witness as LegacyWitness +from openultrasast.core.verdict import Judgment + +from .abstract_domain import EvidenceRelation, PairJudgment +from .trace import Trace + +_MISSING_CONSTRAINT = "missing_constraint_relation" +_AMBIGUOUS_CONSTRAINT = "ambiguous_constraint_relation" + + +def _location(trace: Trace, scope: evidence.AnalyzedScope) -> semantics.SourceLocation: + matches = tuple(region for region in scope.regions if region.start_line <= trace.operation_line <= region.end_line) + if len(matches) != 1: + raise ValueError("trace operation must resolve to exactly one analyzed source region") + return semantics.SourceLocation(matches[0].path, trace.operation_line, 0, trace.operation_kind) + + +def _product(stamp: core_provenance.ProvenanceStamp) -> semantics.EvidenceProvenance: + return semantics.EvidenceProvenance( + semantics.ProvenanceKind.PRODUCT_CODE, + stamp.change_input_digest, + None, + None, + None, + ) + + +def _constraint_state(trace: Trace) -> tuple[semantics.RelationStatus, str, str, evidence.NeutralOutcome]: + state = trace.abstract_state + if state is None or trace.hops: + return semantics.RelationStatus.INCOMPLETE, "unknown", _MISSING_CONSTRAINT, evidence.NeutralOutcome.INDETERMINATE + complete = state.path_judgment.completeness.is_complete + known_values = state.resource.kind != "unknown" and not state.authority.is_unknown + known_relations = state.resource_relation in (EvidenceRelation.MUST, EvidenceRelation.NO) and state.authority_relation in ( + EvidenceRelation.MUST, + EvidenceRelation.NO, + ) + if not (complete and known_values and known_relations): + return semantics.RelationStatus.INCOMPLETE, "unknown", _MISSING_CONSTRAINT, evidence.NeutralOutcome.INDETERMINATE + judgment = state.path_judgment.judgment + if judgment is PairJudgment.UNGUARDED: + return semantics.RelationStatus.COMPLETE, "unconstrained", "", evidence.NeutralOutcome.VIOLATION + if judgment is PairJudgment.DISCHARGED: + return semantics.RelationStatus.COMPLETE, "guarded", "", evidence.NeutralOutcome.SATISFIED + return semantics.RelationStatus.UNRESOLVED, "unknown", _AMBIGUOUS_CONSTRAINT, evidence.NeutralOutcome.INDETERMINATE + + +def _decision_rule(contract: family.FamilyContract, outcome: evidence.NeutralOutcome) -> family.DecisionRule: + matches = tuple(rule for rule in contract.decisions if rule.outcome is outcome) + if len(matches) != 1: + raise ValueError("authorization contract must declare exactly one rule for each neutral outcome") + return matches[0] + + +def _relations( + trace: Trace, + location: semantics.SourceLocation, + product: semantics.EvidenceProvenance, +) -> tuple[tuple[semantics.SemanticRelation, ...], str, evidence.NeutralOutcome]: + entry = semantics.SemanticNode("entry", "entry", location) + operation = semantics.SemanticNode("operation", "operation", location) + operation_relation = semantics.make_relation( + kind=semantics.RelationKind.OPERATION_EFFECT, + role="operation", + source=entry, + target=operation, + location=location, + status=semantics.RelationStatus.COMPLETE, + conditions=(), + reasons=(), + provenance=product, + ) + status, target_name, reason, outcome = _constraint_state(trace) + constraint_relation = semantics.make_relation( + kind=semantics.RelationKind.CONSTRAINT_EFFECT, + role="constraint", + source=operation, + target=semantics.SemanticNode(target_name, "constraint", location), + location=location, + status=status, + conditions=tuple(dict.fromkeys(trace.requires)), + reasons=() if not reason else (reason,), + provenance=product, + ) + return (operation_relation, constraint_relation), reason, outcome + + +def _packet( + trace: Trace, + contract: family.FamilyContract, + ecosystem: str, + scope: evidence.AnalyzedScope, + stamp: core_provenance.ProvenanceStamp, +) -> tuple[evidence.EvidencePacket, evidence.NeutralOutcome]: + location = _location(trace, scope) + product = _product(stamp) + transitions, reason, outcome = _relations(trace, location, product) + fields = ( + semantics.FieldValue("operation", semantics.FieldKind.TEXT, trace.operation_kind), + semantics.FieldValue("boundary", semantics.FieldKind.TEXT, trace.boundary), + ) + observation = semantics.make_observation( + role="security_object", + location=location, + fields=fields, + status=semantics.RelationStatus.COMPLETE, + provenance=product, + ) + completeness = () if not reason else (reason,) + return ( + evidence.EvidencePacket( + evidence.EVIDENCE_PACKET_SCHEMA, + contract.family_id, + family.family_contract_digest(contract), + family.derive_security_object_id(contract.security_object, fields), + ecosystem, + scope, + transitions, + (observation,), + fields, + completeness, + stamp, + ), + outcome, + ) + + +def to_evidence_record( + trace: Trace, + *, + contract: family.FamilyContract, + ecosystem: str, + scope: evidence.AnalyzedScope, + provenance: core_provenance.ProvenanceStamp, +) -> evidence.EvidenceRecord: + """Map one parked authorization trace without promoting missing hop knowledge.""" + if type(trace) is not Trace: + raise TypeError("authorization adapter requires an exact Trace") + family.validate_contract(contract) + packet, outcome = _packet(trace, contract, ecosystem, scope, provenance) + completeness = packet.completeness_reasons + rule = _decision_rule(contract, outcome) + decision = evidence.Decision( + rule.family_code, + outcome, + evidence.EvidenceBasis.DETERMINISTIC, + scope, + completeness, + evidence.derive_packet_digest(packet), + _deterministic_provenance(), + ) + record = evidence.EvidenceRecord(packet, decision, "") + family.validate_record(contract, record) + return record + + +def _deterministic_provenance() -> core_provenance.DecisionProvenance: + return core_provenance.DecisionProvenance(None, None, None, None, None, None) + + +def to_legacy_trace_record( + trace: Trace, + *, + entry: str, + digests: Mapping[str, str], + complete: bool = False, +) -> TraceRecord: + """Preserve the frozen v1 conversion while its writers remain parked.""" + state = trace.abstract_state + if state is None: + judgment = Judgment.INDETERMINATE + reasons: tuple[str, ...] = ("abstract_state_unset",) + witnesses: tuple[LegacyWitness, ...] = () + resource = authority = "unknown" + else: + judgment = Judgment(state.path_judgment.judgment.value) + completeness = state.path_judgment.completeness + reasons = tuple(completeness.reasons) or (("incomplete",) if not completeness.is_complete else ()) + witnesses = tuple( + LegacyWitness(kind="witness", provenance=item, relation=match.relation.value) + for match in state.path_judgment.matches + for item in match.evidence + ) + resource = ",".join(state.resource.values) or "unknown" + authority = ",".join(state.authority.values) or "unknown" + hops = tuple(LegacyHop(function=item.function, call_line=item.call_line, complete=complete) for item in trace.hops) + hop_guards = tuple(guard for hop in trace.hops for guard in hop.guards_before_call) + guards = tuple(LegacyGuard(kind=item.kind, effect=item.effect) for item 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), + ) + + +__all__ = ["to_evidence_record", "to_legacy_trace_record"] diff --git a/src/openultrasast/semantic/obligations/trace.py b/src/openultrasast/semantic/obligations/trace.py index 58bc63f..e411c12 100644 --- a/src/openultrasast/semantic/obligations/trace.py +++ b/src/openultrasast/semantic/obligations/trace.py @@ -10,12 +10,6 @@ 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 ( @@ -153,47 +147,6 @@ 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_native.py b/tests/test_core_native.py index 2bf1c66..206660f 100644 --- a/tests/test_core_native.py +++ b/tests/test_core_native.py @@ -17,6 +17,7 @@ from openultrasast.semantic.obligations.abstract_domain import ( # noqa: E402 ResourceRef, WitnessMatch, ) +from openultrasast.semantic.obligations.core_adapter import to_legacy_trace_record # noqa: E402 from openultrasast.semantic.obligations.trace import AbstractTraceState, Guard, Hop, Trace # noqa: E402 @@ -55,14 +56,14 @@ def _trace() -> Trace: def test_trace_natively_conforms_to_core_schema() -> None: - record = _trace().to_core_record(entry="src/app.py::handler", digests={"input": "abc123"}) + record = to_legacy_trace_record(_trace(), entry="src/app.py::handler", digests={"input": "abc123"}) assert isinstance(record, TraceRecord) assert record.entry == "src/app.py::handler" assert record.judgment is Judgment.UNGUARDED 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() + assert record.canonical() == to_legacy_trace_record(_trace(), entry="src/app.py::handler", digests={"input": "abc123"}).canonical() def test_legacy_trace_without_abstract_state_is_indeterminate() -> None: @@ -82,7 +83,7 @@ def test_legacy_trace_without_abstract_state_is_indeterminate() -> None: dominating=trace.dominating, missing=trace.missing, ) - record = legacy.to_core_record(entry="e", digests={}) + record = to_legacy_trace_record(legacy, entry="e", digests={}) assert record.judgment is Judgment.INDETERMINATE assert record.completeness_reasons diff --git a/tests/test_core_validation.py b/tests/test_core_validation.py index 7d4e529..2499844 100644 --- a/tests/test_core_validation.py +++ b/tests/test_core_validation.py @@ -18,6 +18,7 @@ from openultrasast.semantic.obligations.abstract_domain import ( # noqa: E402 ResourceRef, WitnessMatch, ) +from openultrasast.semantic.obligations.core_adapter import to_legacy_trace_record # noqa: E402 from openultrasast.semantic.obligations.trace import AbstractTraceState, Trace # noqa: E402 REPO = Path(__file__).resolve().parents[1] @@ -67,7 +68,7 @@ def test_end_to_end_push_through_core_contracts() -> None: missing="identity_constraint", abstract_state=state, ) - record = trace.to_core_record(entry="src/app.py::handler", digests={"input": "abc123"}) + record = to_legacy_trace_record(trace, 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" diff --git a/tests/test_obligation_core_adapter.py b/tests/test_obligation_core_adapter.py new file mode 100644 index 0000000..6542eba --- /dev/null +++ b/tests/test_obligation_core_adapter.py @@ -0,0 +1,89 @@ +"""Authorization traces cross into detection-core v2 only through the adapter.""" + +from __future__ import annotations + +import inspect +from dataclasses import replace + +from test_core_evidence import _scope, _stamp +from test_core_native import _trace +from test_core_reference_families import _contract, _load + +from openultrasast.core import evidence, family, semantics +from openultrasast.semantic.obligations import trace as trace_module +from openultrasast.semantic.obligations.abstract_domain import PairJudgment, ResourceRef +from openultrasast.semantic.obligations.core_adapter import to_evidence_record + + +def _adapt(trace: trace_module.Trace) -> evidence.EvidenceRecord: + contract = _contract(_load("obligation_family.json")) + stamp = replace(_stamp(), family_contract_digest=family.family_contract_digest(contract)) + return to_evidence_record( + trace, + contract=contract, + ecosystem="reference-obligation", + scope=_scope(), + provenance=stamp, + ) + + +def test_direct_complete_authorization_trace_maps_to_v2_violation() -> None: + trace = replace(_trace(), hops=(), operation_line=3) + + record = _adapt(trace) + + assert record.decision.outcome is evidence.NeutralOutcome.VIOLATION + assert record.decision.family_code == "constraint_exposed" + assert tuple(item.role for item in record.packet.transitions) == ("operation", "constraint") + assert all(item.status is semantics.RelationStatus.COMPLETE for item in record.packet.transitions) + + +def test_missing_per_hop_completeness_stays_incomplete_and_indeterminate() -> None: + trace = replace(_trace(), operation_line=3) + + record = _adapt(trace) + + constraint = next(item for item in record.packet.transitions if item.role == "constraint") + assert constraint.status is semantics.RelationStatus.INCOMPLETE + assert record.decision.outcome is evidence.NeutralOutcome.INDETERMINATE + assert record.decision.family_code == "constraint_unknown" + assert "missing_constraint_relation" in record.packet.completeness_reasons + + +def test_complete_discharged_trace_maps_to_v2_satisfaction() -> None: + trace = replace(_trace(), hops=(), operation_line=3) + assert trace.abstract_state is not None + state = replace( + trace.abstract_state, + path_judgment=replace(trace.abstract_state.path_judgment, judgment=PairJudgment.DISCHARGED, missing=()), + ) + + record = _adapt(replace(trace, abstract_state=state)) + + assert record.decision.outcome is evidence.NeutralOutcome.SATISFIED + assert record.decision.family_code == "constraint_enforced" + + +def test_unknown_resource_degrades_complete_legacy_judgment_to_indeterminate() -> None: + trace = replace(_trace(), hops=(), operation_line=3) + assert trace.abstract_state is not None + state = replace(trace.abstract_state, resource=ResourceRef(kind="unknown")) + + record = _adapt(replace(trace, abstract_state=state)) + + assert record.decision.outcome is evidence.NeutralOutcome.INDETERMINATE + assert record.packet.transitions[1].status is semantics.RelationStatus.INCOMPLETE + + +def test_unset_abstract_state_never_lifts_legacy_trace_to_a_clean_result() -> None: + record = _adapt(replace(_trace(), hops=(), operation_line=3, abstract_state=None)) + + assert record.decision.outcome is evidence.NeutralOutcome.INDETERMINATE + assert record.packet.transitions[1].status is semantics.RelationStatus.INCOMPLETE + + +def test_authorization_trace_domain_has_no_core_import_or_conversion_method() -> None: + source = inspect.getsource(trace_module) + + assert "openultrasast.core" not in source + assert not hasattr(trace_module.Trace, "to_core_record")