From 4fbcef76463a3817d1b0ceeb5cb9082a7d8644ca Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 01:32:16 +0200 Subject: [PATCH 1/2] test: define proof and refutation behavior --- tests/test_detectors.py | 9 +- tests/test_refutation.py | 270 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 276 insertions(+), 3 deletions(-) create mode 100644 tests/test_refutation.py diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 31434df..3f9fb17 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -122,13 +122,16 @@ def test_taint_detector_reports_flows_of_its_kind_only() -> None: "source": "python.web.param", "source_label": "http", "sink": "python.os.system", + "verdict": "vulnerability", + "evidence": "no guard on the path to the sink", } -def test_taint_detector_declares_only_the_taint_analysis() -> None: +def test_taint_detector_declares_taint_and_refutation() -> None: + from coretrace_python.findings.refutation import RefutationAnalysis from coretrace_python.taint import TaintAnalysis - assert CommandFlows.requires == frozenset({TaintAnalysis}) + assert CommandFlows.requires == frozenset({TaintAnalysis, RefutationAnalysis}) def test_symbol_call_detector_reports_resolved_calls() -> None: @@ -172,7 +175,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "xss", } for rule in ("sql-injection", "command-injection", "path-traversal", "ssrf", "xss"): - assert by_name[rule].requires == ("taint.flows",) + assert by_name[rule].requires == ("taint.flows", "findings.refutation") assert by_name[rule].provides == (f"vulnerability.{rule}",) assert by_name["python-stdlib-models"].provides == ("model.python-stdlib",) assert by_name["weak-crypto"].provides == ("vulnerability.weak-crypto",) diff --git a/tests/test_refutation.py b/tests/test_refutation.py new file mode 100644 index 0000000..1c7f942 --- /dev/null +++ b/tests/test_refutation.py @@ -0,0 +1,270 @@ +"""Acceptance tests for the proof / refutation engine (``docs/architecture.md`` §24). + +After detection, every taint flow gets a ``Verdict``: ``REFUTED`` when a guard that +dominates the sink validates every tainted origin of the argument (string constraints +such as ``isdigit()``, membership in a constant allowlist, equality with a constant) or +when the sink is unreachable by constant propagation; ``HOTSPOT`` when a dominating guard +tests the value without proving it safe; ``VULNERABILITY`` otherwise. Detectors drop +refuted flows and lower the confidence of hotspots. + +Expected to remain red until ``coretrace_python.findings.refutation`` exists. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager +from coretrace_python.findings import Confidence +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import ( + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + Source, + TaintAnalysis, + TaintKind, +) + +try: + from coretrace_python.findings.refutation import ( + RefutationAnalysis, + Status, + Verdict, + Verdicts, + ) +except ImportError as error: # pragma: no cover - red until refutation lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_refutation() -> None: + if MISSING is not None: + pytest.fail(f"refutation engine is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +MODELS = ( + Source(SymbolId("python.builtins.input"), "stdin"), + Source(SymbolId("python.flask.request.args"), "http"), + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), + Sink(SymbolId("python.db.execute"), TaintKind.SQL), +) + +PRELUDE = "import os\nimport db\nfrom flask import request\n\n" + + +def verdicts(body: str) -> Verdicts: + module = build_hir(SourceManager().add_source("proof.py", PRELUDE + body)) + manager = AnalysisManager(module) + manager.register(*engine.ALL_ANALYSES) + registry = SecurityModelRegistry() + registry.register(*MODELS) + manager.provide(SecurityModelAnalysis, registry.freeze()) + function = next(s for s in module.body if isinstance(s, nodes.Function)) + return manager.get(RefutationAnalysis, function) + + +def only(result: Verdicts) -> Verdict: + (verdict,) = result.all() + return verdict + + +# --------------------------------------------------------------------------- the §24 example + + +def test_isdigit_guard_before_the_sink_refutes_the_flow() -> None: + verdict = only( + verdicts( + "def run():\n" + " value = request.args['id']\n" + " if not value.isdigit():\n" + " return\n" + " db.execute(value)\n" + ) + ) + + assert isinstance(verdict, Verdict) + assert verdict.status is Status.REFUTED + assert "isdigit" in verdict.evidence + assert verdict.flow.location.start_line == 9 + + +def test_guard_on_the_wrong_branch_does_not_refute() -> None: + verdict = only( + verdicts( + "def run():\n" + " value = request.args['id']\n" + " if value.isdigit():\n" + " return\n" + " db.execute(value)\n" + ) + ) + assert verdict.status is Status.VULNERABILITY + + +def test_unguarded_flow_is_a_vulnerability() -> None: + verdict = only(verdicts("def run():\n os.system(input())\n")) + assert verdict.status is Status.VULNERABILITY + assert verdict.evidence == "no guard on the path to the sink" + + +# --------------------------------------------------------------------------- guard shapes + + +def test_validator_on_the_positive_branch() -> None: + verdict = only( + verdicts( + "def run():\n value = input()\n if value.isalnum():\n os.system(value)\n" + ) + ) + assert verdict.status is Status.REFUTED + + +def test_constant_allowlist_membership_refutes() -> None: + positive = only( + verdicts("def run():\n cmd = input()\n if cmd in ('ls', 'pwd'):\n os.system(cmd)\n") + ) + negative = only( + verdicts("def run():\n cmd = input()\n if cmd not in ('ls', 'pwd'):\n return\n os.system(cmd)\n") + ) + + assert positive.status is Status.REFUTED + assert "allowlist" in positive.evidence + assert negative.status is Status.REFUTED + + +def test_equality_with_a_constant_refutes() -> None: + verdict = only(verdicts("def run():\n cmd = input()\n if cmd == 'ls':\n os.system(cmd)\n")) + assert verdict.status is Status.REFUTED + + +def test_non_validating_guards_make_a_hotspot() -> None: + truthy = only(verdicts("def run():\n cmd = input()\n if cmd:\n os.system(cmd)\n")) + length = only( + verdicts("def run():\n cmd = input()\n if len(cmd) < 10:\n os.system(cmd)\n") + ) + unknown = only( + verdicts("def run():\n cmd = input()\n if cmd in allowed():\n os.system(cmd)\n") + ) + + assert truthy.status is Status.HOTSPOT + assert length.status is Status.HOTSPOT + assert unknown.status is Status.HOTSPOT + assert "guard" in truthy.evidence + + +def test_guards_on_unrelated_values_are_ignored() -> None: + verdict = only( + verdicts("def run(flag):\n cmd = input()\n if flag.isdigit():\n os.system(cmd)\n") + ) + assert verdict.status is Status.VULNERABILITY + + +def test_validated_value_stays_validated_through_string_building() -> None: + verdict = only( + verdicts( + "def run():\n" + " name = request.args['name']\n" + " if not name.isdigit():\n" + " return\n" + " db.execute('SELECT * FROM t WHERE id = ' + name)\n" + ) + ) + assert verdict.status is Status.REFUTED + + +def test_every_tainted_origin_must_be_validated() -> None: + verdict = only( + verdicts( + "def run():\n" + " a = input()\n" + " b = input()\n" + " if not a.isdigit():\n" + " return\n" + " os.system(a + b)\n" + ) + ) + assert verdict.status is Status.VULNERABILITY + + +def test_boolean_combinations() -> None: + both = only( + verdicts("def run(other):\n cmd = input()\n if cmd.isdigit() and other:\n os.system(cmd)\n") + ) + either = only( + verdicts("def run(other):\n cmd = input()\n if cmd.isdigit() or other:\n os.system(cmd)\n") + ) + negated_or = only( + verdicts( + "def run(other):\n cmd = input()\n if not (cmd.isdigit() or other):\n return\n os.system(cmd)\n" + ) + ) + + assert both.status is Status.REFUTED + assert either.status is Status.HOTSPOT + assert negated_or.status is Status.HOTSPOT + + +def test_unreachable_sinks_are_refuted_by_constant_propagation() -> None: + verdict = only(verdicts("def run():\n cmd = input()\n if False:\n os.system(cmd)\n")) + assert verdict.status is Status.REFUTED + assert "unreachable" in verdict.evidence + + +def test_guard_inside_a_loop_still_dominates_the_sink() -> None: + verdict = only( + verdicts( + "def run(items):\n" + " for item in items:\n" + " cmd = input()\n" + " if cmd.isdigit():\n" + " os.system(cmd)\n" + ) + ) + assert verdict.status is Status.REFUTED + + +# --------------------------------------------------------------------------- wiring + + +def test_refutation_is_a_function_analysis_over_taint_and_dominance() -> None: + from coretrace_python.abstract import ConstantPropagation + from coretrace_python.cfg import DominanceAnalysis + + assert RefutationAnalysis.name == "findings.refutation" + assert {TaintAnalysis, DominanceAnalysis, ConstantPropagation} <= RefutationAnalysis.requires + result = verdicts("def run():\n os.system(input())\n os.system('ls')\n") + assert len(result.all()) == 1 + assert result.verdict(result.all()[0].flow) is result.all()[0] + + +def test_detectors_drop_refuted_flows_and_downgrade_hotspots() -> None: + findings = engine.check( + SourceManager().add_source( + "app.py", + "import os\n\n" + "def proven():\n os.system(input())\n\n" + "def checked():\n cmd = input()\n if cmd:\n os.system(cmd)\n\n" + "def safe():\n cmd = input()\n if cmd.isdigit():\n os.system(cmd)\n", + ), + [PLUGINS], + ) + + assert [(f.function, f.confidence) for f in findings] == [ + ("proven", Confidence.HIGH), + ("checked", Confidence.MEDIUM), + ] + assert findings[0].metadata["verdict"] == "vulnerability" + assert findings[1].metadata["verdict"] == "hotspot" + assert "guard" in findings[1].metadata["evidence"] From cad4c9741339f70c02dccf828404c718b971e243 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 01:32:16 +0200 Subject: [PATCH 2/2] feat: add the proof and refutation engine over dominating guards --- README.md | 6 +- .../security/command_injection/plugin.toml | 2 +- plugins/security/path_traversal/plugin.toml | 2 +- plugins/security/sql_injection/plugin.toml | 2 +- plugins/security/ssrf/plugin.toml | 2 +- plugins/security/xss/plugin.toml | 2 +- src/coretrace_python/engine.py | 2 + src/coretrace_python/findings/refutation.py | 292 ++++++++++++++++++ src/coretrace_python/plugins/detectors.py | 14 +- 9 files changed, 316 insertions(+), 8 deletions(-) create mode 100644 src/coretrace_python/findings/refutation.py diff --git a/README.md b/README.md index 2a794f3..7db42fe 100644 --- a/README.md +++ b/README.md @@ -41,7 +41,11 @@ sources. Model plugins contribute sources, sinks and sanitizers; detectors consume the shared taint result, so adding a framework model makes every detector aware of it. Taint follows calls between functions of the same module through function summaries: a tainted argument passed to a helper that reaches a sink is reported -at the call site, and a helper returning attacker-controlled data taints its result. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing +at the call site, and a helper returning attacker-controlled data taints its result. +Each flow is then judged: a dominating guard that proves the value safe (`isdigit()`, +membership in a constant allowlist, equality with a constant) refutes it and it is not +reported; a guard that only mentions the value makes it a hotspot with medium confidence; +the verdict and its evidence are in the finding metadata. `--format` is `text` (default), `json` or `sarif`. Exit status is 0 when nothing was found, 1 when findings were reported, and 2 on a usage or analysis error. ## Emit PyIR diff --git a/plugins/security/command_injection/plugin.toml b/plugins/security/command_injection/plugin.toml index e40b4d0..0c89205 100644 --- a/plugins/security/command_injection/plugin.toml +++ b/plugins/security/command_injection/plugin.toml @@ -1,7 +1,7 @@ name = "command-injection" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["taint.flows"] +requires = ["taint.flows", "findings.refutation"] provides = ["vulnerability.command-injection"] [entrypoint] diff --git a/plugins/security/path_traversal/plugin.toml b/plugins/security/path_traversal/plugin.toml index 4936023..8c51205 100644 --- a/plugins/security/path_traversal/plugin.toml +++ b/plugins/security/path_traversal/plugin.toml @@ -1,7 +1,7 @@ name = "path-traversal" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["taint.flows"] +requires = ["taint.flows", "findings.refutation"] provides = ["vulnerability.path-traversal"] [entrypoint] diff --git a/plugins/security/sql_injection/plugin.toml b/plugins/security/sql_injection/plugin.toml index c1a6587..07be710 100644 --- a/plugins/security/sql_injection/plugin.toml +++ b/plugins/security/sql_injection/plugin.toml @@ -1,7 +1,7 @@ name = "sql-injection" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["taint.flows"] +requires = ["taint.flows", "findings.refutation"] provides = ["vulnerability.sql-injection"] [entrypoint] diff --git a/plugins/security/ssrf/plugin.toml b/plugins/security/ssrf/plugin.toml index 084fcd3..72b7015 100644 --- a/plugins/security/ssrf/plugin.toml +++ b/plugins/security/ssrf/plugin.toml @@ -1,7 +1,7 @@ name = "ssrf" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["taint.flows"] +requires = ["taint.flows", "findings.refutation"] provides = ["vulnerability.ssrf"] [entrypoint] diff --git a/plugins/security/xss/plugin.toml b/plugins/security/xss/plugin.toml index bad14cf..3d3cf0a 100644 --- a/plugins/security/xss/plugin.toml +++ b/plugins/security/xss/plugin.toml @@ -1,7 +1,7 @@ name = "xss" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["taint.flows"] +requires = ["taint.flows", "findings.refutation"] provides = ["vulnerability.xss"] [entrypoint] diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 3b022d8..cd1fd1e 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -14,6 +14,7 @@ from coretrace_python.analysis import AnalysisManager, AnyAnalysis from coretrace_python.cfg import CFGAnalysis, CFGError, DominanceAnalysis, PostDominanceAnalysis from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.findings.refutation import RefutationAnalysis from coretrace_python.frontend import build_hir from coretrace_python.hir import nodes from coretrace_python.interprocedural import CallGraphAnalysis, SummaryAnalysis @@ -47,6 +48,7 @@ SummaryAnalysis, SecurityModelAnalysis, TaintAnalysis, + RefutationAnalysis, ) diff --git a/src/coretrace_python/findings/refutation.py b/src/coretrace_python/findings/refutation.py new file mode 100644 index 0000000..5deae47 --- /dev/null +++ b/src/coretrace_python/findings/refutation.py @@ -0,0 +1,292 @@ +"""Proof / refutation engine (architecture §24). + +Every taint flow gets a verdict. Walking the dominators of the sink block, each branch +whose one side alone reaches the sink fixes the truth of its condition there; the +condition is then interpreted: string validators (``isdigit()`` and friends), membership +in a constant allowlist and equality with a constant prove a value safe, ``and`` / ``or`` +and ``not`` combine as expected, and anything else that mentions the value is a guard +that does not prove it. A flow is refuted when every tainted origin of the argument is +proven safe or the sink is unreachable, a hotspot when an unproven guard mentions it, and +a vulnerability otherwise. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from enum import Enum +from types import MappingProxyType +from typing import ClassVar + +from coretrace_python.abstract import ConstantPropagation +from coretrace_python.analysis import AnalysisContext, AnyAnalysis, FunctionAnalysis +from coretrace_python.cfg import CFG, BlockId, CFGAnalysis, DominanceAnalysis, DominatorTree +from coretrace_python.hir import nodes +from coretrace_python.ir.model import ( + BoolOp, + Branch, + BuildList, + BuildTuple, + Call, + Compare, + Constant, + FunctionIR, + GetAttr, + Instruction, + UnaryOp, + Value, +) +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.taint import TaintAnalysis, TaintFacts, TaintFlow + +VALIDATORS = frozenset( + {"isdigit", "isnumeric", "isdecimal", "isalnum", "isalpha", "isidentifier", "isascii"} +) + + +class Status(Enum): + VULNERABILITY = "vulnerability" + HOTSPOT = "hotspot" + REFUTED = "refuted" + + +@dataclass(frozen=True) +class Verdict: + flow: TaintFlow + status: Status + evidence: str + + +class Verdicts: + def __init__(self, verdicts: tuple[Verdict, ...]) -> None: + self._all = verdicts + self._by_flow: Mapping[TaintFlow, Verdict] = MappingProxyType( + {verdict.flow: verdict for verdict in verdicts} + ) + + def all(self) -> tuple[Verdict, ...]: + return self._all + + def verdict(self, flow: TaintFlow) -> Verdict: + return self._by_flow[flow] + + +@dataclass(frozen=True) +class _Guard: + """A dominating branch condition with its truth at the sink.""" + + condition: Value + truth: bool + line: int + + +class _Judge: + def __init__( + self, function: FunctionIR, cfg: CFG, tree: DominatorTree, taint: TaintFacts + ) -> None: + self.function = function + self.cfg = cfg + self.tree = tree + self.taint = taint + self.blocks = {block.id: block for block in function.blocks} + self.defs: dict[Value, Instruction] = { + i.result: i for block in function.blocks for i in block.instructions if i.result + } + self._closures: dict[Value, frozenset[Value]] = {} + + # ------------------------------------------------------------------ dependencies + + def closure(self, value: Value) -> frozenset[Value]: + """Every value ``value`` transitively depends on, excluding itself.""" + + if value in self._closures: + return self._closures[value] + found: set[Value] = set() + pending = [value] + while pending: + definition = self.defs.get(pending.pop()) + if definition is None: + continue + for operand in definition.operands(): + if operand not in found: + found.add(operand) + pending.append(operand) + self._closures[value] = frozenset(found) + return self._closures[value] + + def origins(self, value: Value) -> frozenset[Value]: + """Tainted values the argument depends on whose own operands are untainted.""" + + candidates = self.closure(value) | {value} + return frozenset( + v + for v in candidates + if self.taint.taint(v) + and not any(self.taint.taint(o) for o in self.closure(v)) + ) + + # ------------------------------------------------------------------ guards + + def sink_block(self, flow: TaintFlow) -> BlockId | None: + for block in self.function.blocks: + for instruction in block.instructions: + if isinstance(instruction, Call) and instruction.location == flow.location: + return block.id + return None + + def reaches(self, start: BlockId, target: BlockId, avoiding: BlockId) -> bool: + seen: set[BlockId] = set() + pending = [start] + while pending: + block = pending.pop() + if block == target: + return True + if block in seen or block == avoiding: + continue + seen.add(block) + pending.extend(self.cfg.successors(block)) + return False + + def guards(self, sink: BlockId) -> list[_Guard]: + found: list[_Guard] = [] + dominator = self.tree.idom(sink) + while dominator is not None: + terminator = self.blocks[dominator].terminator + if isinstance(terminator, Branch): + then_reaches = self.reaches(terminator.then_block, sink, dominator) + else_reaches = self.reaches(terminator.else_block, sink, dominator) + if then_reaches != else_reaches: + found.append( + _Guard(terminator.condition, then_reaches, terminator.location.start_line) + ) + dominator = self.tree.idom(dominator) + return found + + def interpret( + self, condition: Value, truth: bool | None + ) -> tuple[dict[Value, str], set[Value]]: + """Values the condition proves safe (with the reason), and values it mentions + without proving anything. ``truth`` is ``None`` when it is not fixed at the sink. + + A recognised check evaluated the wrong way (``isdigit()`` known false) yields + nothing: it is neither a proof nor a reassuring guard.""" + + definition = self.defs.get(condition) + proven: dict[Value, str] = {} + mentioned: set[Value] = set() + if isinstance(definition, UnaryOp) and definition.operator == "not": + return self.interpret(definition.operand, None if truth is None else not truth) + if isinstance(definition, BoolOp): + fixed = truth is not None and (definition.operator == "and") == truth + for value in definition.values: + found, seen = self.interpret(value, truth if fixed else None) + proven.update(found) + mentioned |= seen + return proven, mentioned + recognised = self.recognise(definition) + if recognised is None: + return proven, set(self.closure(condition)) | {condition} + tested, reason, when_true = recognised + if truth is None: + mentioned = set(self.closure(condition)) | {condition} + elif truth == when_true: + proven[tested] = reason + return proven, mentioned + + def recognise(self, definition: Instruction | None) -> tuple[Value, str, bool] | None: + """``(validated value, reason, truth that validates)`` for known check shapes.""" + + if isinstance(definition, Call) and not definition.arguments: + callee = self.defs.get(definition.callee) + if isinstance(callee, GetAttr) and callee.attribute in VALIDATORS: + return callee.object, f"guarded by {callee.attribute}()", True + if isinstance(definition, Compare): + left, right = definition.left, definition.right + if definition.operator in ("in", "not_in") and self.is_constant_collection(right): + return ( + left, + "allowlisted by a membership check on constants", + definition.operator == "in", + ) + if definition.operator in ("eq", "not_eq"): + for tested, other in ((left, right), (right, left)): + if isinstance(self.defs.get(other), Constant): + return tested, "equals a constant", definition.operator == "eq" + return None + + def is_constant_collection(self, value: Value) -> bool: + definition = self.defs.get(value) + if isinstance(definition, Constant): + return isinstance(definition.value, str | bytes) + if isinstance(definition, BuildList | BuildTuple): + return all(isinstance(self.defs.get(e), Constant) for e in definition.elements) + return False + + # ------------------------------------------------------------------ verdicts + + def judge(self, flow: TaintFlow, reachable: bool) -> Verdict: + sink = self.sink_block(flow) + if sink is None or not reachable: + return Verdict(flow, Status.REFUTED, "sink unreachable by constant propagation") + origins = self.origins(flow.argument) + chains = { + origin: frozenset( + w for w in self.closure(flow.argument) | {flow.argument} + if w == origin or origin in self.closure(w) + ) + for origin in origins + } + proofs: dict[Value, str] = {} + mentions: dict[Value, int] = {} + for guard in self.guards(sink): + proven, mentioned = self.interpret(guard.condition, guard.truth) + for origin, chain in chains.items(): + for value, reason in proven.items(): + if value in chain: + proofs.setdefault(origin, reason) + if origin not in mentions and mentioned & chain: + mentions[origin] = guard.line + if origins and all(origin in proofs for origin in origins): + return Verdict(flow, Status.REFUTED, "; ".join(sorted(set(proofs.values())))) + unguarded = [origin for origin in origins if origin not in proofs and origin not in mentions] + if not origins or unguarded: + return Verdict(flow, Status.VULNERABILITY, "no guard on the path to the sink") + line = min(mentions.values()) + return Verdict( + flow, Status.HOTSPOT, f"guard at line {line} does not prove the value safe" + ) + + +def judge_flows( + function: FunctionIR, + cfg: CFG, + tree: DominatorTree, + taint: TaintFacts, + reachable: frozenset[BlockId], +) -> Verdicts: + judge = _Judge(function, cfg, tree, taint) + verdicts = [] + for flow in taint.flows: + sink = judge.sink_block(flow) + verdicts.append(judge.judge(flow, sink is not None and sink in reachable)) + return Verdicts(tuple(verdicts)) + + +class RefutationAnalysis(FunctionAnalysis[Verdicts]): + name: ClassVar[str] = "findings.refutation" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset( + {TaintAnalysis, DominanceAnalysis, ConstantPropagation, SSAAnalysis, CFGAnalysis} + ) + + @classmethod + def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> Verdicts: + ssa = ctx.get(SSAAnalysis, function) + constants = ctx.get(ConstantPropagation, function) + return judge_flows( + ssa, + ctx.get(CFGAnalysis, function), + ctx.get(DominanceAnalysis, function), + ctx.get(TaintAnalysis, function), + frozenset(b.id for b in ssa.blocks if constants.reachable(b.id)), + ) + diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py index 19917f1..e3b9b5f 100644 --- a/src/coretrace_python/plugins/detectors.py +++ b/src/coretrace_python/plugins/detectors.py @@ -13,6 +13,7 @@ from coretrace_python.analysis import AnyAnalysis from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.findings.refutation import RefutationAnalysis, Status from coretrace_python.ir.model import Call, Symbol from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.plugins.api import Plugin, PluginContext @@ -23,7 +24,7 @@ class TaintDetector(Plugin): """Report every taint flow carrying ``kind`` into a sink.""" - requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({TaintAnalysis}) + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({TaintAnalysis, RefutationAnalysis}) rule_id: ClassVar[str] kind: ClassVar[TaintKind] severity: ClassVar[Severity] @@ -32,25 +33,34 @@ class TaintDetector(Plugin): def analyze(self, ctx: PluginContext) -> Sequence[Finding]: findings: list[Finding] = [] for function in ctx.functions(): + verdicts = ctx.get(RefutationAnalysis, function) for flow in ctx.get(TaintAnalysis, function).flows: if not flow.kinds & self.kind: continue + verdict = verdicts.verdict(flow) + if verdict.status is Status.REFUTED: + continue message = f"{self.title}: {flow.source.label} input reaches {flow.sink.symbol}" metadata = { "source": str(flow.source.symbol), "source_label": flow.source.label, "sink": str(flow.sink.symbol), + "verdict": verdict.status.value, + "evidence": verdict.evidence, } if flow.through is not None and flow.sink_location is not None: message += f" through {flow.through}" metadata["through"] = flow.through metadata["sink_line"] = str(flow.sink_location.start_line) + confidence = ( + Confidence.HIGH if verdict.status is Status.VULNERABILITY else Confidence.MEDIUM + ) findings.append( Finding( rule_id=self.rule_id, message=message, severity=self.severity, - confidence=Confidence.HIGH, + confidence=confidence, span=flow.location, function=function.name, metadata=metadata,