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) 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"}