From 327e5f3d4a77ba12e9b0754e91ae29abb7c0e9fb Mon Sep 17 00:00:00 2001 From: Hugo Date: Sat, 26 Sep 2026 04:08:56 +0200 Subject: [PATCH 1/2] test: specify when the URL of an SSRF flow downgrades or refutes it The ssrf rule now also reads how the URL of each flow is built, so its manifest requires the SSA form and the module's strings. --- tests/test_detectors.py | 5 +- tests/test_ssrf_destinations.py | 154 ++++++++++++++++++++++++++++++++ 2 files changed, 158 insertions(+), 1 deletion(-) create mode 100644 tests/test_ssrf_destinations.py diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 74b0b8d..6e6c65d 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -201,8 +201,11 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "xss", } for rule in ("sql-injection", "command-injection", "code-injection", "nosql-injection", "path-traversal", "ssrf", "xss"): - assert by_name[rule].requires == ("taint.flows", "findings.refutation") assert by_name[rule].provides == (f"vulnerability.{rule}",) + for rule in ("sql-injection", "command-injection", "code-injection", "nosql-injection", "path-traversal", "xss"): + assert by_name[rule].requires == ("taint.flows", "findings.refutation") + # ``ssrf`` also reads how the URL of each flow is built. + assert by_name["ssrf"].requires == ("taint.flows", "findings.refutation", "ir.ssa", "abstract.module_strings") assert by_name["python-stdlib-models"].provides == ("model.python-stdlib",) assert by_name["weak-crypto"].provides == ("vulnerability.weak-crypto",) diff --git a/tests/test_ssrf_destinations.py b/tests/test_ssrf_destinations.py new file mode 100644 index 0000000..016b165 --- /dev/null +++ b/tests/test_ssrf_destinations.py @@ -0,0 +1,154 @@ +"""An SSRF flow is judged on how its URL is built, with ``+`` and f-strings only. + +A fixed host is a limited proof. When constant text fixes ``scheme://host`` and ends the +authority with ``/``, ``?`` or ``#`` before any input, the input cannot choose the host; +it still chooses the path, which may reach a sensitive resource of that host, and a +redirect may lead elsewhere. The flow is a hotspot. It is refuted only when the constant +text fixes the path too, the input reaching the query or the fragment, and the call +disables redirects (``allow_redirects=False``, ``follow_redirects=False``): the +destination is then fixed. + +Every other URL stays a vulnerability: a base of unknown value (an import, a call +result), which may leave the authority open (``BASE_URL = "https:/"``); a constant that +leaves it open (no ``/`` after the host, a variable port, no scheme); ``urljoin``; a URL +built in a helper function. A module-level name is known text only when the module binds +it once, to a string literal, and never rebinds it. The advisory correlation keeps its +own verdict. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import Advisory, AdvisoryEntryPoint, dump_advisories +from coretrace_python.findings import Finding, Severity +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager + +PLUGINS = [engine.BUNDLED_PLUGINS] + +APP = ( + "import httpx\nimport requests\nfrom urllib.parse import urljoin\nfrom flask import Flask, request\n\n" + "app = Flask(__name__)\n" + "API = 'https://api.example.com/'\n\n" + "@app.route('/users/')\n" + "def fetch(uid):\n" + " return {call}\n" +) + + +def check(text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("app.py", text), PLUGINS) + + +def ssrf(text: str) -> list[tuple[int, str]]: + return [(f.span.start_line, f.metadata["verdict"]) for f in check(text) if f.rule_id == "ssrf"] + + +@pytest.mark.parametrize( + "call, verdict", + [ + ("requests.get('https://api.example.com/users/' + uid, timeout=5)", "hotspot"), + ("requests.get(f'https://api.example.com/users/{uid}', timeout=5)", "hotspot"), + ("requests.get(API + 'users/' + uid, timeout=5)", "hotspot"), + # ``path='@evil.com'`` makes ``https://api.example.com@evil.com``: evil.com is the host. + ("requests.get('https://api.example.com' + request.args['path'], timeout=5)", "vulnerability"), + # An absolute or scheme-relative ``path`` replaces the host. + ("requests.get(urljoin('https://api.example.com/', request.args['path']), timeout=5)", "vulnerability"), + ("requests.get(request.args['url'], timeout=5)", "vulnerability"), + ], +) +def test_constant_text_fixing_the_host_makes_the_flow_a_hotspot(call: str, verdict: str) -> None: + assert ssrf(APP.format(call=call)) == [(11, verdict)] + + +@pytest.mark.parametrize( + "prelude, call", + [ + # An imported base may be anything: ``"https:/"`` would let ``uid`` choose the host. + ("from settings import BASE_URL\n", "requests.get(f'{BASE_URL}/users/{uid}', timeout=5)"), + ("BASE = 'https:/'\n", "requests.get(f'{BASE}/{uid}', timeout=5)"), + ("BASE = get_base()\n", "requests.get(BASE + '/users/' + uid, timeout=5)"), + ("\n", "requests.get('https://api.example.com:' + uid + '/users/', timeout=5)"), + ("\n", "requests.get('//api.example.com/users/' + uid, timeout=5)"), + ("\n", "requests.get('https://api.example.com/users/%s' % uid, timeout=5)"), + ], +) +def test_an_unknown_or_incomplete_base_stays_a_vulnerability(prelude: str, call: str) -> None: + text = APP.format(call=call).replace("API = 'https://api.example.com/'\n", prelude) + + assert ssrf(text) == [(11, "vulnerability")] + + +@pytest.mark.parametrize( + "rebinding", + [ + "API = 'https://b.example.com/'\n", + "def configure():\n global API\n API = input()\n", + "for API in ('https://a.example.com/', 'https://b.example.com/'):\n pass\n", + "def API():\n return 'https://a.example.com/'\n", + ], +) +def test_a_module_string_bound_more_than_once_is_unknown(rebinding: str) -> None: + text = APP.format(call="requests.get(API + 'users/' + uid, timeout=5)") + "\n" + rebinding + + assert ssrf(text) == [(11, "vulnerability")] + + +@pytest.mark.parametrize( + "call, verdict", + [ + ("requests.get('https://api.example.com/search?q=' + request.args['q'], allow_redirects=False, timeout=5)", None), + ("httpx.get(f'https://api.example.com/search?q={uid}', follow_redirects=False, timeout=5)", None), + ("requests.get('https://api.example.com/search#' + uid, allow_redirects=False, timeout=5)", None), + # Redirects followed: the fixed destination may send the request elsewhere. + ("requests.get('https://api.example.com/search?q=' + request.args['q'], timeout=5)", "hotspot"), + # Redirects disabled, but the input chooses the path. + ("requests.get('https://api.example.com/users/' + uid, allow_redirects=False, timeout=5)", "hotspot"), + # Unpacked options may enable redirects again. + ("requests.get('https://api.example.com/search?q=' + uid, **{'allow_redirects': True})", "hotspot"), + ], +) +def test_only_a_fixed_destination_without_redirects_is_refuted(call: str, verdict: str | None) -> None: + assert ssrf(APP.format(call=call)) == ([(11, verdict)] if verdict else []) + + +def test_the_evidence_names_the_fixed_text_and_what_the_input_still_chooses() -> None: + (finding,) = [ + f for f in check(APP.format(call="requests.get('https://api.example.com/users/' + uid, timeout=5)")) + if f.rule_id == "ssrf" + ] + + assert "'https://api.example.com/'" in finding.metadata["evidence"] + assert "path" in finding.metadata["evidence"] and "redirect" in finding.metadata["evidence"] + + +def test_a_url_built_in_a_helper_function_stays_a_vulnerability() -> None: + text = APP.format(call="get(uid)") + "\ndef get(path):\n return requests.get('https://api.example.com/' + path, timeout=5)\n" + + assert ssrf(text) == [(11, "vulnerability")] + + +def test_the_advisory_correlation_keeps_its_own_verdict(tmp_path: Path) -> None: + advisory = Advisory( + "CVE-2099-0801", + "requests", + "<3", + "get sends credentials to the host of a crafted URL", + Severity.HIGH, + entry_points=(AdvisoryEntryPoint(SymbolId("python.requests.get"), "commit abc1234"),), + modules=("requests",), + ) + (tmp_path / "requirements.txt").write_text("requests==2.31.0\nflask==3.0.0\n", encoding="utf-8") + (tmp_path / "advisories.json").write_text(dump_advisories((advisory,)), encoding="utf-8") + (tmp_path / "app.py").write_text( + APP.format(call="requests.get('https://api.example.com/users/' + uid, timeout=5)"), encoding="utf-8" + ) + + findings = engine.analyze_project(tmp_path, PLUGINS).findings + + verdicts = {f.rule_id: f.metadata["verdict"] for f in findings if f.rule_id in ("ssrf", "exploitable-vulnerability")} + assert verdicts == {"ssrf": "hotspot", "exploitable-vulnerability": "vulnerability"} From 2a7df8671f5ad061579587ed04d9317004a23e83 Mon Sep 17 00:00:00 2001 From: Hugo Date: Sat, 26 Sep 2026 04:08:56 +0200 Subject: [PATCH 2/2] feat: judge SSRF flows on the destination their URL's constant text fixes A fixed host makes a flow a hotspot; a fixed host and path with redirects disabled refutes it. Only + and f-strings are read, over string literals and module-level strings bound once. The rule refines its own verdict through TaintDetector.judge, so the advisory correlation keeps the refutation's. --- CHANGELOG.md | 10 ++ docs/plugins.md | 6 ++ docs/usage.md | 2 +- src/coretrace_python/abstract/strings.py | 98 +++++++++++++++++++ .../bundled/security/ssrf/plugin.toml | 2 +- .../bundled/security/ssrf/ssrf.py | 53 +++++++++- src/coretrace_python/engine.py | 2 + src/coretrace_python/plugins/detectors.py | 15 ++- 8 files changed, 181 insertions(+), 7 deletions(-) create mode 100644 src/coretrace_python/abstract/strings.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 289387e..5d704f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## Unreleased + +### Precision + +- `ssrf` judges a flow on how `+` and f-strings build its URL. A fixed host is a limited proof: when constant text fixes `scheme://host` and ends the authority with `/`, `?` or `#` before any input, as in `"https://api.example.com/users/" + id`, the flow is a hotspot, because the input still chooses the path and a redirect may lead elsewhere. It is refuted only when the constant text fixes the path too, the input reaching the query or fragment only, and the call disables redirects (`allow_redirects=False`, `follow_redirects=False`). A module-level name counts as known text when the module binds it once to a string literal and never rebinds it. Everything else stays a vulnerability, including a base of unknown value such as an imported `BASE_URL`, which may leave the authority open. The advisory correlation keeps its own verdict. The regression corpus is unchanged: its six `ssrf` findings build their URL from values of unknown content (#165). + +### Plugins + +- `TaintDetector.judge(ctx, function, flow, verdict)` lets a rule refine the refutation's verdict on a flow for its own findings, without changing the verdict other consumers of the flow read (#165). + ## 0.13.0 (2026-09-26) ### Precision diff --git a/docs/plugins.md b/docs/plugins.md index b5038c8..c1d453a 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -115,6 +115,12 @@ ordinary input reaching a database write is not a plaintext credential; `LOG`, f values that must not be written to a log as they are; and `PII`, for personal data that must not leave the application through a log, a third party or plain storage. +A rule may know more about its own sinks than the refutation does: `judge(ctx, function, +flow, verdict)` returns the rule's verdict on a flow, given the refutation's, and by +default returns it unchanged. `ssrf` reads there how the URL is built. The refutation's +verdict, which other consumers of the flow such as the advisory correlation read, does +not change; a rule reading more analyses declares them in its `requires`. + ### Security models: `ModelPlugin` Declares sources, sinks and sanitizers for a library or a framework. A model plugin diff --git a/docs/usage.md b/docs/usage.md index b0c2079..1d8ffe3 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -114,7 +114,7 @@ variable is not treated that way yet. | `sql-injection` | A database statement (`cursor.execute`, SQLAlchemy `text`, Django `raw`, …). Parameters of a parameterised query are not statements. | | `nosql-injection` | A NoSQL query, such as a MongoDB filter, given attacker input that may hold a structure: a JSON body giving `{"$ne": null}` matches any value. The input that may hold a structure is JSON, the raw bodies a `json.loads` decodes, files, WebSocket messages and standard input. Text cannot hold an operator: form fields, query parameters, headers, cookies, URL parameters, the command line, the environment, and an entry-point parameter annotated with scalars (`str`, `int`, `Optional[str]`, `Annotated[str, Query()]`, `list[str]`, …), which FastAPI validates. A Django, REST framework, aiohttp or tornado view reads a request object: its text attributes (`GET`, `POST`, `headers`, `query_params`, `query`, `match_info`, `arguments`, …) and the URL parameters after it are text, its body, files and `data` are not. Limits: a helper function the request is passed to reads it as a whole; the keys of a dict are not followed, so a string concatenated into `$where` is not reported. The analyzer ships no NoSQL sink: model plugins declare them. | | `path-traversal` | A file system path (`open`, `send_file`, `os.remove`, …). | -| `ssrf` | The URL of an HTTP client request (Requests, httpx, `urllib`), by position or as `url=`; the body, headers and query parameters are not the destination. | +| `ssrf` | The URL of an HTTP client request (Requests, httpx, `urllib`), by position or as `url=`; the body, headers and query parameters are not the destination. How `+` and f-strings build the URL can prove part of its destination. When constant text fixes `scheme://host` and ends the authority with `/`, `?` or `#` before any input (`"https://api.example.com/users/" + id`, or a module-level string bound once and never rebound), the flow is a hotspot: the input still chooses the path, which may reach a sensitive resource, and a redirect may lead elsewhere. With the path fixed too, the input reaching the query or fragment only, and the call disabling redirects (`allow_redirects=False`, `follow_redirects=False`), the destination is fixed and the flow is refuted. Anything unproven stays a vulnerability: a base of unknown value (imported, a call result), a constant leaving the authority open (`"https:/"`, no `/` after the host, a variable port), `urljoin`, `%`, `format`, a URL built in a helper function. | | `xss` | An HTTP response body built without escaping. A Django template rendered by `render_to_string` escapes, where the analyzer can establish it (below). | | `insecure-deserialization` | `pickle`, `marshal`, `dill`, `jsonpickle`, `shelve`, and the YAML loaders (`yaml.load`, `load_all`, `full_load`, `unsafe_load` and their several-document versions); `yaml.load` and `load_all` given a safe loader (`SafeLoader`, `BaseLoader`) are not. | | `open-redirect` | An HTTP redirect target. | diff --git a/src/coretrace_python/abstract/strings.py b/src/coretrace_python/abstract/strings.py new file mode 100644 index 0000000..ccf2714 --- /dev/null +++ b/src/coretrace_python/abstract/strings.py @@ -0,0 +1,98 @@ +"""The constant text a string starts with (architecture §18): what ``+`` and f-strings +build from string literals and from module-level names bound once to one. Anything +else, a parameter, a call result, an import, ``%`` or ``format``, is text of unknown +value.""" + +from __future__ import annotations + +from collections.abc import Mapping +from types import MappingProxyType +from typing import ClassVar + +from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis +from coretrace_python.cfg import CFGError +from coretrace_python.hir import nodes +from coretrace_python.ir.lowering import ( + MODULE_BODY, + LoweringError, + PyIRAnalysis, + analyzable_functions, +) +from coretrace_python.ir.model import ( + BinaryOp, + BuildString, + Constant, + ForNext, + Global, + Instruction, + SetGlobal, + StoreLocal, + Value, +) + + +def leading_text(value: Value, defs: Mapping[Value, Instruction], strings: Mapping[str, str]) -> tuple[str, bool]: + """The constant text ``value`` starts with, and whether that text is all of it, given + the definitions of the function's values and the module's ``strings``.""" + + made = defs.get(value) + if isinstance(made, Constant): + return (made.value, True) if isinstance(made.value, str) else ("", False) + if isinstance(made, Global): + known = strings.get(made.name) + return ("", False) if known is None else (known, True) + if isinstance(made, BinaryOp) and made.operator == "add": + left, whole = leading_text(made.left, defs, strings) + if not whole: + return left, False + right, whole = leading_text(made.right, defs, strings) + return left + right, whole + if isinstance(made, BuildString): + text = "" + for part in made.parts: + piece, whole = leading_text(part, defs, strings) + text += piece + if not whole: + return text, False + return text, True + return "", False + + +class ModuleStringsAnalysis(Analysis[Mapping[str, str]]): + """The module-level names bound once, to a string literal, and never rebound: by no + other statement of the module body, no function assigning them as ``global``, no + definition of that name. A module that does not lower entirely has none.""" + + name: ClassVar[str] = "abstract.module_strings" + requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({PyIRAnalysis}) + + @classmethod + def compute(cls, ctx: AnalysisContext) -> Mapping[str, str]: + try: + functions = [ctx.get(PyIRAnalysis, function) for function in analyzable_functions(ctx.module)] + except (LoweringError, CFGError): + return MappingProxyType({}) + body = next((function for function in functions if function.name == MODULE_BODY), None) + if body is None: + return MappingProxyType({}) + defs = {i.result: i for block in body.blocks for i in block.instructions if i.result is not None} + stored: dict[str, list[Value | None]] = {} + for block in body.blocks: + for instruction in block.instructions: + if isinstance(instruction, StoreLocal): + stored.setdefault(instruction.name, []).append(instruction.value) + if isinstance(block.terminator, ForNext): + # A ``for`` binds its target to each item. + stored.setdefault(block.terminator.target, []).append(None) + rebound = { + i.name for function in functions for block in function.blocks for i in block.instructions + if isinstance(i, SetGlobal) + } + rebound |= {s.name for s in ctx.module.body if isinstance(s, nodes.Function | nodes.Class)} + found: dict[str, str] = {} + for name, values in stored.items(): + value = values[0] if len(values) == 1 and name not in rebound else None + made = defs.get(value) if value is not None else None + if isinstance(made, Constant) and isinstance(made.value, str): + found[name] = made.value + return MappingProxyType(found) diff --git a/src/coretrace_python/bundled/security/ssrf/plugin.toml b/src/coretrace_python/bundled/security/ssrf/plugin.toml index 72b7015..b962c88 100644 --- a/src/coretrace_python/bundled/security/ssrf/plugin.toml +++ b/src/coretrace_python/bundled/security/ssrf/plugin.toml @@ -1,7 +1,7 @@ name = "ssrf" version = "1.0.0" plugin_api = ">=1,<2" -requires = ["taint.flows", "findings.refutation"] +requires = ["taint.flows", "findings.refutation", "ir.ssa", "abstract.module_strings"] provides = ["vulnerability.ssrf"] [entrypoint] diff --git a/src/coretrace_python/bundled/security/ssrf/ssrf.py b/src/coretrace_python/bundled/security/ssrf/ssrf.py index 61b0788..bc57564 100644 --- a/src/coretrace_python/bundled/security/ssrf/ssrf.py +++ b/src/coretrace_python/bundled/security/ssrf/ssrf.py @@ -1,12 +1,37 @@ -"""Server-side request forgery: attacker-controlled input reaching a SSRF sink.""" +"""Server-side request forgery: attacker-controlled input reaching a SSRF sink. + +How the URL is built, with ``+`` and f-strings, can prove part of its destination. Constant +text fixing ``scheme://host`` and ending the authority with ``/``, ``?`` or ``#`` before +any input keeps the input from choosing the host; the input still chooses the path, which +may reach a sensitive resource of that host, and a redirect may lead elsewhere, so the +flow is a hotspot. With the path fixed too, the input reaching the query or fragment +only, and redirects disabled in the call, the destination is fixed and the flow is +refuted. Anything unproven, a base of unknown value or a URL built in a helper function, +stays a vulnerability. +""" from __future__ import annotations +import re from typing import ClassVar +from coretrace_python.abstract.strings import ModuleStringsAnalysis, leading_text +from coretrace_python.analysis import AnyAnalysis from coretrace_python.findings import Severity -from coretrace_python.plugins import TaintDetector -from coretrace_python.taint import TaintKind +from coretrace_python.findings.refutation import Status, Verdict +from coretrace_python.hir import nodes +from coretrace_python.ir.ssa import SSAAnalysis +from coretrace_python.plugins import PluginContext, TaintDetector +from coretrace_python.taint import TaintFlow, TaintKind + +# ``scheme://`` and an authority ended by ``/``, ``?`` or ``#``. A backslash, which some +# URL parsers take for a slash, does not end it here. +_FIXED_HOST = re.compile(r"[A-Za-z][A-Za-z0-9+.-]*://[^/?#\\]+[/?#]") +# The same with the path fixed too: what follows is query or fragment. +_FIXED_PATH = re.compile(r"[A-Za-z][A-Za-z0-9+.-]*://[^/?#\\]+(?:/[^?#]*)?[?#]") +# A client honours the one of these keywords it knows and rejects the other with a +# ``TypeError``: either way, the call follows no redirect. +_NO_REDIRECTS = ("allow_redirects", "follow_redirects") class SsrfPlugin(TaintDetector): @@ -15,3 +40,25 @@ class SsrfPlugin(TaintDetector): kind: ClassVar[TaintKind] = TaintKind.SSRF severity: ClassVar[Severity] = Severity.HIGH title: ClassVar[str] = "Server-side request forgery" + requires: ClassVar[frozenset[AnyAnalysis]] = TaintDetector.requires | {SSAAnalysis, ModuleStringsAnalysis} + + def judge(self, ctx: PluginContext, function: nodes.Function, flow: TaintFlow, verdict: Verdict) -> Verdict: + if flow.through is not None: + return verdict # the URL is built in the callee + ssa = ctx.get(SSAAnalysis, function) + defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None} + text, _ = leading_text(flow.argument, defs, ctx.get(ModuleStringsAnalysis)) + host = _FIXED_HOST.match(text) + if host is None: + return verdict + arguments = flow.sink_arguments + no_redirects = arguments is not None and any(arguments.given(name) == (True, "False") for name in _NO_REDIRECTS) + if _FIXED_PATH.match(text) is not None: + if no_redirects: + return Verdict(flow, Status.REFUTED, f"the destination is fixed by {text!r} and redirects are disabled") + left = "a redirect may lead elsewhere" + else: + left = "the input chooses the path" + ("" if no_redirects else ", and a redirect may lead elsewhere") + if verdict.status is not Status.VULNERABILITY: + return verdict + return Verdict(flow, Status.HOTSPOT, f"the host is fixed by {host.group(0)!r}, but {left}") diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 416d881..bcc3148 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -17,6 +17,7 @@ from coretrace_python import __version__ from coretrace_python.abstract import ConstantPropagation, HeapAnalysis, RangeAnalysis +from coretrace_python.abstract.strings import ModuleStringsAnalysis from coretrace_python.analysis import ( AnalysisContext, AnalysisManager, @@ -143,6 +144,7 @@ DefUseAnalysis, ConstantPropagation, RangeAnalysis, + ModuleStringsAnalysis, HeapAnalysis, CallGraphAnalysis, SummaryAnalysis, diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py index 846183a..aa42088 100644 --- a/src/coretrace_python/plugins/detectors.py +++ b/src/coretrace_python/plugins/detectors.py @@ -13,13 +13,14 @@ 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.findings.refutation import RefutationAnalysis, Status, Verdict +from coretrace_python.hir import nodes from coretrace_python.interprocedural import CallGraphAnalysis from coretrace_python.ir.model import Call, Instruction, Symbol, Value from coretrace_python.ir.ssa import SSAAnalysis from coretrace_python.plugins.api import Plugin, PluginContext from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.taint import TaintAnalysis, TaintKind +from coretrace_python.taint import TaintAnalysis, TaintFlow, TaintKind class TaintDetector(Plugin): @@ -40,6 +41,9 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: if not flow.kinds & self.kind: continue verdict = verdicts.verdict(flow) + if verdict.status is Status.REFUTED: + continue + verdict = self.judge(ctx, function, flow, verdict) if verdict.status is Status.REFUTED: continue message = f"{self.title}: {flow.source.label} input reaches {flow.sink.symbol}" @@ -70,6 +74,13 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: ) return findings + def judge(self, ctx: PluginContext, function: nodes.Function, flow: TaintFlow, verdict: Verdict) -> Verdict: + """This rule's verdict on ``flow``, given the refutation's: the same, unless the rule + knows more about its own sinks. The refutation's verdict, which other consumers + of the flow read, does not change.""" + + return verdict + def _lower(confidence: Confidence) -> Confidence: return {Confidence.HIGH: Confidence.MEDIUM}.get(confidence, Confidence.LOW)