diff --git a/.kiro/specs/access-control-prover/tasks.md b/.kiro/specs/access-control-prover/tasks.md index 05cf960..09dc9ca 100644 --- a/.kiro/specs/access-control-prover/tasks.md +++ b/.kiro/specs/access-control-prover/tasks.md @@ -201,7 +201,7 @@ its final resolution, dominance, fact-acquisition and measurement semantics with - _Requirements: 3.10, 4.1, 4.2, 4.3, 4.4, 4.5, 11.5_ - _Boundary: Obligation Checker_ - _Depends: 9.4, 10.4_ -- [ ] 10.6 Integrate three-way judgments into the prover +- [x] 10.6 Integrate three-way judgments into the prover - Emit hunt findings only for unguarded pairs and send no indeterminate pair to the proof arm. - Carry all judgments, completeness reasons, resources, authorities and stable identifiers in traces and coverage. - Record stage-specific exceptions without folding them into quiet coverage. diff --git a/src/openultrasast/hunt/prover.py b/src/openultrasast/hunt/prover.py index 44095ec..556f17f 100644 --- a/src/openultrasast/hunt/prover.py +++ b/src/openultrasast/hunt/prover.py @@ -15,16 +15,28 @@ from ..semantic.cfg import ControlContext, FunctionCfg from ..semantic.cst import parse_function_nodes from ..semantic.facts import load_facts from ..semantic.ir import FileIR, FunctionIR, trailing_name -from ..semantic.obligations.abstract_domain import PairJudgmentCounts +from ..semantic.obligations.abstract_domain import ( + Completeness, + EvidenceRelation, + PairJudgment, + PairJudgmentCounts, + PathJudgment, + count_pair_judgments, + join_alternatives, + parse_authority, + parse_resource, +) from ..semantic.obligations.cfg_dominance import CfgDominance from ..semantic.obligations.check import _on_caller_hop, _relevant, _route_path, check_obligations from ..semantic.obligations.enforcement import witnesses from ..semantic.obligations.facts import load_obligation_facts +from ..semantic.obligations.judgment import aggregate_judgments, judgment_for_path from ..semantic.obligations.operations import Discharge, Operation, find_discharges, find_operations, valid_witness from ..semantic.obligations.paths import compose_paths from ..semantic.obligations.policy import DEFAULT_POLICY_RELPATH, load_declared_policy from ..semantic.obligations.resolution import CallBindings, bind_calls -from ..semantic.obligations.trace import build_trace +from ..semantic.obligations.trace import AbstractTraceState, build_trace +from ..semantic.obligations.witness import match_witness from .producers import Context, _safe_path @@ -101,7 +113,120 @@ def _targets_from_bindings( return target_functions -def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = None, *, max_depth: int = 3) -> ProverResult: +def _path_judgment( + result: Any, + op: Operation, + graphs: CfgDominance, + facts: Any, +) -> PathJudgment: + """Judge one operation on one path with the strict three-way algebra (Req 4.1-4.3).""" + fact = next((f for f in facts.operations if f.id == op.fact_id), None) + pairs_by_kind = { + kind: tuple( + (match_witness(witness, op, fact), graphs.evidence(witness, op)) for witness in result.discharges if witness.kind == kind + ) + for kind in op.requires + } + return judgment_for_path(op, pairs_by_kind, Completeness(is_complete=True)) + + +def _abstract_state(judgment: PathJudgment, op: Operation) -> AbstractTraceState: + """Pair-level abstract state for a strict-mode trace (Req 9.2).""" + matches = judgment.matches + return AbstractTraceState( + path_judgment=judgment, + resource=parse_resource(op.resource), + authority=parse_authority(None), + resource_relation=_joined_relation(match.resource_relation for match in matches), + authority_relation=_joined_relation(match.authority_relation for match in matches), + ) + + +def _joined_relation(relations: Iterator[EvidenceRelation]) -> EvidenceRelation: + values = tuple(relations) + if not values: + return EvidenceRelation.NO + joined = values[0] + for value in values[1:]: + joined = join_alternatives(joined, value) + return joined + + +def _emit_unguarded( + findings: dict[str, dict[str, Any]], + op: Operation, + aggregate: PathJudgment, + record: Any, + trace: dict[str, Any], + entry_index: Mapping[str, Any], + facts: Any, + functions: Mapping[tuple[str, str], FunctionIR], +) -> None: + """Emit one finding per missing kind for an unguarded aggregate (Req 4.4).""" + if aggregate.judgment is not PairJudgment.UNGUARDED: + return + op_fact = next(fact for fact in facts.operations if fact.id == op.fact_id) + fn = functions[op.path, op.function] + calls = sorted( + { + trailing_name(call.name) + for call in fn.calls + if call.line == op.line and any(call.name == name or call.name.endswith("." + name) for name in op_fact.calls) + } + ) + entry_payload = asdict(entry_index[record.entry]) + hops = [list(hop) for hop in record.hops] + for missing_kind in aggregate.missing: + identity = f"obligation:{op.kind}:{op.path}:{op.line}:{missing_kind}" + findings.setdefault( + identity, + dict( + path=op.path, + line=op.line, + family="access_control", + function=op.function, + evidence=[ + f"hops: {' -> '.join(node for node, _ in record.hops)}", + f"operation {op.fact_id} ({op.kind}) at {op.path}:{op.line}; missing {missing_kind}", + "judgment: unguarded (strict three-way mode)", + ], + id=identity, + entry=entry_payload, + hops=hops, + operation=dict(id=op.fact_id, kind=op.kind, calls=calls, line=op.line), + missing=missing_kind, + label="unguarded_path", + trace=trace, + ), + ) + + +def _judgment_findings( + judgments_by_op: Mapping[Operation, list[tuple[Any, PathJudgment]]], + trace_by_op: Mapping[Operation, dict[str, Any]], + entry_index: Mapping[str, Any], + facts: Any, + functions: Mapping[tuple[str, str], FunctionIR], +) -> tuple[dict[str, dict[str, Any]], PairJudgmentCounts]: + """Aggregate per-path judgments per operation and emit unguarded findings only (Reqs 4.4-4.5, 11.5).""" + findings: dict[str, dict[str, Any]] = {} + aggregates: list[PathJudgment] = [] + for op, pairs in judgments_by_op.items(): + aggregate = aggregate_judgments(tuple(judgment for _, judgment in pairs)) + aggregates.append(aggregate) + record, _ = pairs[0] + _emit_unguarded(findings, op, aggregate, record, trace_by_op[op], entry_index, facts, functions) + return findings, count_pair_judgments(tuple(aggregate.judgment for aggregate in aggregates)) + + +def prove_change( + root: Path, + change: Mapping[str, Any], + ctx: Context | None = None, + *, + max_depth: int = 3, + judgment_mode: bool = False, +) -> ProverResult: """Check touched head functions and their direct operation-bearing callees, with no model.""" if max_depth < 0: raise ValueError("invalid path depth") @@ -275,6 +400,8 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No ): extra_witnesses.append(replace(witness, kind="path_guard")) findings: dict[str, dict[str, Any]] = {} + judgments_by_op: dict[Operation, list[tuple[Any, PathJudgment]]] = {} + trace_by_op: dict[Operation, dict[str, Any]] = {} traces: list[dict[str, Any]] = [] checked: set[Operation] = set() undischarged: set[Operation] = set() @@ -302,6 +429,30 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No pair_traces: dict[Operation, dict[str, Any]] = {} for op in operations.get((target_path, target_name), ()): checked.add(op) + if judgment_mode: + judgment = _path_judgment(result, op, graphs, facts) + judgments_by_op.setdefault(op, []).append((record, judgment)) + if judgment.judgment is PairJudgment.UNGUARDED: + undischarged.add(op) + trace_entry = entry_index[record.entry] + route = _route_path(op, ctx.texts[op.path]) or trace_entry.name + declared_access = policy.route_access(route) if policy is not None and route else None + trace = build_trace( + path=record, + entry=trace_entry, + access=declared_access, + operation=op, + irs=path_irs, + facts=facts, + witnesses=result.discharges, + dominance=dominance, + dominating=(), + missing=judgment.missing[0] if judgment.missing else None, + abstract_state=_abstract_state(judgment, op), + ).record() + traces.append(trace) + trace_by_op[op] = trace + continue # Silence from a public premise is not evidence that a check dominated the operation. dominated = { w.kind @@ -331,7 +482,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No ).record() traces.append(trace) pair_traces[op] = trace - for finding in result.findings: + for finding in result.findings if not judgment_mode else (): op = finding.operation identity = f"obligation:{op.kind}:{op.path}:{op.line}:{finding.missing}" op_fact = next(fact for fact in facts.operations if fact.id == op.fact_id) @@ -367,6 +518,10 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No ), ) degradations.extend(graphs.degradations) + pair_counts = PairJudgmentCounts() + if judgment_mode: + judgment_findings, pair_counts = _judgment_findings(judgments_by_op, trace_by_op, entry_index, facts, functions) + findings.update(judgment_findings) coverage = Coverage( dict(languages), dict(frameworks), @@ -377,6 +532,7 @@ def prove_change(root: Path, change: Mapping[str, Any], ctx: Context | None = No len(checked - undischarged), dict(reasons), resolution_counts=dict(bindings.counts), + pair_judgments=pair_counts, ) return ProverResult(tuple(findings.values()), coverage, tuple(degradations), tuple(traces)) diff --git a/src/openultrasast/semantic/obligations/witness.py b/src/openultrasast/semantic/obligations/witness.py index 133c057..e38b6bf 100644 --- a/src/openultrasast/semantic/obligations/witness.py +++ b/src/openultrasast/semantic/obligations/witness.py @@ -35,8 +35,9 @@ def match_witness( """Match one witness against one operation on kind, scope, resource and authority.""" if witness.kind not in operation.requires: return _match(EvidenceRelation.NO, EvidenceRelation.NO, EvidenceRelation.UNKNOWN, witness, operation, ()) - if _out_of_scope(witness, operation): - return _match(EvidenceRelation.NO, EvidenceRelation.NO, EvidenceRelation.UNKNOWN, witness, operation, ()) + scope_relation = _scope_relation(witness, operation) + if scope_relation is not EvidenceRelation.MUST: + return _match(scope_relation, EvidenceRelation.NO, EvidenceRelation.UNKNOWN, witness, operation, ()) resource_relation = _resource_relation(witness, operation, fact) authority_relation, implication_ids = _authority_relation(witness, required_authority, implications) relation = conjoin_dimensions( @@ -45,13 +46,15 @@ def match_witness( return _match(relation, resource_relation, authority_relation, witness, operation, implication_ids) -def _out_of_scope(witness: Discharge, operation: Operation) -> bool: +def _scope_relation(witness: Discharge, operation: Operation) -> EvidenceRelation: + """Whole-handler scopes on another handler are definitely unrelated; a statement-scope + witness in a caller of the operation's function is undecidable here (Req 2.7).""" same_function = (witness.path, witness.function) == (operation.path, operation.function) + if same_function or witness.scope == "hop": + return EvidenceRelation.MUST if witness.scope in _WHOLE_HANDLER_SCOPES: - return not same_function - if witness.scope == "hop": - return False - return not same_function + return EvidenceRelation.NO + return EvidenceRelation.UNKNOWN def _resource_relation(witness: Discharge, operation: Operation, fact: OperationFact | None) -> EvidenceRelation: diff --git a/tests/test_prover_judgment_mode.py b/tests/test_prover_judgment_mode.py new file mode 100644 index 0000000..db18a84 --- /dev/null +++ b/tests/test_prover_judgment_mode.py @@ -0,0 +1,135 @@ +"""End-to-end three-way judgment mode for ``prove_change`` (task 10.6). + +The fixtures mirror ``tests/test_hunt_prover.py``: a real git repository whose head +commit introduces one Stapler-shaped Java class. Judgment mode is opt-in; the +legacy boolean flow must stay byte-identical without the keyword (Req 4.6). +""" + +from __future__ import annotations + +import subprocess +from typing import Any + +from openultrasast.hunt.producers import Context +from openultrasast.hunt.prover import prove_change +from openultrasast.semantic.obligations.trace import parse_abstract_state + + +def repository(tmp_path: Any, java_text: str) -> tuple[Any, dict[str, Any]]: + """Build a one-commit repository whose head commit introduces ``java_text``.""" + work = tmp_path / "repo" + work.mkdir() + run = lambda *args: subprocess.run( # noqa: E731 - fixture-local shorthand + args, cwd=work, check=True, capture_output=True, text=True + ) + run("git", "init", "-q") + run("git", "config", "user.email", "fixture@example.com") + run("git", "config", "user.name", "fixture") + (work / "Fixture.java").write_text("class Plain {}\n", encoding="utf-8") + run("git", "add", "Fixture.java") + run("git", "commit", "-qm", "base") + head_lines = { + "Fixture.java": [ + index for index, line in enumerate(java_text.splitlines(), start=1) if line.strip() and "class Fixture" not in line + ] + } + (work / "Fixture.java").write_text(java_text, encoding="utf-8") + run("git", "add", "Fixture.java") + run("git", "commit", "-qm", "head") + head = run("git", "rev-parse", "HEAD").stdout.strip() + base = run("git", "rev-parse", "HEAD~1").stdout.strip() + change = dict( + id="change-1", + repo="fixture/repo", + kind="vulnerability", + base=base, + head=head, + files=[{"path": "Fixture.java", "head_lines": [[1, len(java_text.splitlines())]]}], + head_lines=head_lines, + ) + return work, change + + +UNGUARDED = """class Fixture extends Descriptor { + public Object doTestConnection() { + return CredentialsProvider.lookupCredentials(Thing.class); + } +} +""" + +GUARDED = """class Fixture extends Descriptor { + public Object doTestConnection() { + Jenkins.get().checkPermission(Jenkins.ADMINISTER); + return CredentialsProvider.lookupCredentials(Thing.class); + } +} +""" + +CROSS_FUNCTION = """class Fixture extends Descriptor { + public Object doTestConnection() { + Jenkins.get().checkPermission(Jenkins.ADMINISTER); + return this.helper(); + } + + private Object helper() { + return CredentialsProvider.lookupCredentials(Thing.class); + } +} +""" + + +def _prove(work: Any, change: dict[str, Any], **kwargs: Any) -> Any: + context = Context(root=work, change=change) + return prove_change(work, change, context, **kwargs) + + +def test_unguarded_fixture_flags_and_counts_unguarded(tmp_path: Any) -> None: + work, change = repository(tmp_path, UNGUARDED) + result = _prove(work, change, judgment_mode=True) + assert len(result.findings) == 3 + finding = result.findings[0] + assert finding["missing"] in {"path_guard", "ownership_check", "identity_constraint"} + assert finding["family"] == "access_control" + abstract = parse_abstract_state(finding["trace"]["abstract_state"]) + assert abstract.path_judgment.judgment.value == "unguarded" + assert abstract.path_judgment.missing == ("identity_constraint", "ownership_check", "path_guard") + assert result.coverage.pair_judgments.unguarded == 1 + + +def test_guarded_fixture_is_indeterminate_without_resource_binding(tmp_path: Any) -> None: + work, change = repository(tmp_path, GUARDED) + result = _prove(work, change, judgment_mode=True) + assert result.findings == () + assert result.coverage.pair_judgments.indeterminate == 1 + assert result.coverage.pair_judgments.discharged == 0 + abstract_states = [trace["abstract_state"] for trace in result.traces if "abstract_state" in trace] + assert len(abstract_states) == 1 + assert parse_abstract_state(abstract_states[0]).path_judgment.judgment.value == "indeterminate" + + +def test_cross_function_guard_is_indeterminate_in_strict_mode(tmp_path: Any) -> None: + work, change = repository(tmp_path, CROSS_FUNCTION) + result = _prove(work, change, judgment_mode=True) + assert result.findings == () + assert result.coverage.pair_judgments.indeterminate == 1 + assert result.coverage.pair_judgments.discharged == 0 + + +def test_legacy_mode_is_unchanged_without_the_keyword(tmp_path: Any) -> None: + work, change = repository(tmp_path, GUARDED) + result = _prove(work, change) + assert result.findings == () + assert all("abstract_state" not in trace for trace in result.traces) + assert result.coverage.pair_judgments.unguarded == 0 + assert result.coverage.pair_judgments.discharged == 0 + assert result.coverage.pair_judgments.indeterminate == 0 + + +def test_traces_carry_judgment_and_missing_kinds(tmp_path: Any) -> None: + work, change = repository(tmp_path, UNGUARDED) + result = _prove(work, change, judgment_mode=True) + assert result.traces + judgment_values = { + parse_abstract_state(trace["abstract_state"]).path_judgment.judgment.value for trace in result.traces if "abstract_state" in trace + } + assert judgment_values == {"unguarded"}