From 860278e8789b312001b0c498d77207c5cc00101a Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 24 Sep 2026 14:46:32 +0200 Subject: [PATCH 1/2] test: define a sink call made safe by one of its arguments Only a safe value given explicitly, by keyword or at its position, takes the model's kinds off the sink, directly or through a callee, and only those kinds; an absent argument, another value, a variable or unpacked arguments leave it. yaml.load with SafeLoader, CSafeLoader, the yaml.loader spelling or CBaseLoader is no insecure deserialization; with no loader, Loader, FullLoader or UnsafeLoader it still is. The plugin guide must document the model. --- tests/test_docs.py | 2 +- tests/test_safe_arguments.py | 114 +++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 1 deletion(-) create mode 100644 tests/test_safe_arguments.py diff --git a/tests/test_docs.py b/tests/test_docs.py index d604eac..6865c7f 100644 --- a/tests/test_docs.py +++ b/tests/test_docs.py @@ -78,7 +78,7 @@ def test_plugin_guide_documents_the_manifest_base_classes_and_models() -> None: for base in ("Plugin", "ModelPlugin", "ProjectPlugin", "SymbolCallDetector", "TaintDetector", "SecretDetector"): assert base in plugins.__all__ and f"`{base}`" in text for model in ("Source", "Sink", "Sanitizer", "EntryPoint", "TypedParameter", "NamedParameter", - "RouteRegistrar", "SuffixSink", "Validator", "AuthorizationGuard"): + "RouteRegistrar", "SuffixSink", "SafeArgument", "Validator", "AuthorizationGuard"): assert hasattr(taint, model) and re.search(rf"`{model}\b", text), model assert all(kind.name in text for kind in taint.TaintKind if kind.name not in {"NONE", "ALL"}) assert "--plugins" in text and "plugin.toml" in text diff --git a/tests/test_safe_arguments.py b/tests/test_safe_arguments.py new file mode 100644 index 0000000..329faa7 --- /dev/null +++ b/tests/test_safe_arguments.py @@ -0,0 +1,114 @@ +"""A sink call made safe by one of its arguments: ``yaml.load(data, Loader=SafeLoader)``. + +A ``SafeArgument`` model says that a call to a sink whose argument (by keyword, or at a +position) denotes one of some values is not a sink for some kinds. Only a value the +call gives explicitly makes it safe: an absent argument, another value, a variable or +unpacked arguments leave the sink as it is, so a doubt is always reported. +""" + +from __future__ import annotations + +import pytest + +from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager +from coretrace_python.findings import Finding +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 ( + ModelTable, + SafeArgument, + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + Source, + TaintAnalysis, + TaintKind, +) + +PLUGINS = engine.BUNDLED_PLUGINS +SAFE = SafeArgument(SymbolId("python.db.load"), "mode", ("python.db.SAFE",), position=1, kinds=TaintKind.CODE) + + +def models(*extra: Sink) -> ModelTable: + registry = SecurityModelRegistry() + registry.register( + Source(SymbolId("python.builtins.input"), "stdin"), + Sink(SymbolId("python.db.load"), TaintKind.CODE), + SAFE, + *extra, + ) + return registry.freeze() + + +def flows(source_text: str, table: ModelTable) -> list[tuple[str, TaintKind]]: + module = build_hir(SourceManager().add_source("safe.py", source_text)) + manager = AnalysisManager(module) + manager.register(*engine.ALL_ANALYSES) + manager.provide(SecurityModelAnalysis, table) + run = next(s for s in module.body if isinstance(s, nodes.Function) and s.name == "run") + return [(str(flow.sink.symbol), flow.kinds) for flow in manager.get(TaintAnalysis, run).flows] + + +# --------------------------------------------------------------------------- the model + + +def test_safe_arguments_are_models_of_the_table() -> None: + table = models() + + assert table.safe_argument(SymbolId("python.db.load")) == SAFE + assert table.safe_argument(SymbolId("python.os.system")) is None + + +@pytest.mark.parametrize( + "call, reported", + [ + ("db.load(input(), mode=db.SAFE)", False), + ("db.load(input(), db.SAFE)", False), + ("db.load(input())", True), + ("db.load(input(), mode=db.UNSAFE)", True), + ("db.load(input(), mode=choice)", True), + ("db.load(input(), **options)", True), + ], +) +def test_only_a_safe_value_given_explicitly_takes_the_kinds_off_the_sink(call: str, reported: bool) -> None: + found = flows(f"import db\n\ndef run(choice, options):\n {call}\n", models()) + + assert bool(found) is reported + + +def test_a_safe_argument_holds_through_a_callee() -> None: + source = "import db\n\ndef run():\n load(input())\n\ndef load(text):\n return db.load(text, db.SAFE)\n" + + assert flows(source, models()) == [] + + +def test_a_safe_argument_only_takes_off_its_own_kinds() -> None: + found = flows("import db\n\ndef run():\n db.load(input(), db.SAFE)\n", models(Sink(SymbolId("python.db.load"), TaintKind.COMMAND))) + + assert found == [("python.db.load", TaintKind.COMMAND)] + + +# --------------------------------------------------------------------------- yaml.load + + +def deserialization(loader: str | None) -> list[Finding]: + call = f"yaml.load(request.data{'' if loader is None else ', ' + loader})" + source = f"import yaml\nfrom flask import request\n\ndef read():\n return {call}\n" + findings = engine.check(SourceManager().add_source("app.py", source), [PLUGINS]) + return [f for f in findings if f.rule_id == "insecure-deserialization"] + + +@pytest.mark.parametrize( + "loader", + ["Loader=yaml.SafeLoader", "Loader=yaml.CSafeLoader", "yaml.loader.SafeLoader", "Loader=yaml.cyaml.CBaseLoader"], +) +def test_yaml_load_with_a_safe_loader_is_not_an_insecure_deserialization(loader: str) -> None: + assert deserialization(loader) == [] + + +@pytest.mark.parametrize("loader", [None, "yaml.Loader", "Loader=yaml.FullLoader", "Loader=yaml.UnsafeLoader"]) +def test_yaml_load_with_any_other_loader_still_is(loader: str | None) -> None: + assert len(deserialization(loader)) == 1 From 60d43f775a1d12a474f1bbba5ed98da25cd4de6e Mon Sep 17 00:00:00 2001 From: Hugo Date: Thu, 24 Sep 2026 14:46:32 +0200 Subject: [PATCH 2/2] feat: let a safe argument take kinds off a sink call yaml.load(data, Loader=yaml.SafeLoader) was reported as a critical insecure deserialization. A SafeArgument model names the argument, by keyword or position, and the values that make a call to the sink safe for some kinds; the taint engine applies it where it reports a sink, directly and through a callee's summary. The reading of a call's arguments moves to Arguments.given, shared with the advisory conditions. Shipped for yaml.load with SafeLoader, BaseLoader and their C versions under every spelling. Closes #121 --- CHANGELOG.md | 6 +++++ docs/plugins.md | 1 + .../models/python_stdlib/python_stdlib.py | 27 ++++++++++++++++++- .../dependency/correlation.py | 14 +--------- .../interprocedural/callgraph.py | 12 +++++++++ src/coretrace_python/taint/__init__.py | 2 ++ src/coretrace_python/taint/engine.py | 19 ++++++++++--- src/coretrace_python/taint/models.py | 24 +++++++++++++++++ 8 files changed, 88 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 601cb1d..9874ccd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Precision + +- A sink call can be made safe by one of its arguments: a `SafeArgument` model names the argument and the values that take kinds off the sink, only when the call gives one explicitly. `yaml.load` with `SafeLoader`, `BaseLoader` or their C versions, under every spelling, is no longer an insecure deserialization; an absent loader, another loader, a variable or unpacked arguments still are (#121). + ## 0.5.0 (2026-09-24) ### Dependencies diff --git a/docs/plugins.md b/docs/plugins.md index a6c7348..aa11065 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -148,6 +148,7 @@ class BottleModels(ModelPlugin): | `NamedParameter(pattern, label, kinds)` | A regular expression on parameter names; matching parameters carry `kinds`, as `password` carries `CREDENTIAL`. | | `RouteRegistrar(symbol, argument, label, kinds=ALL, keyword=None)` | A call registering a view at argument `argument` (or `keyword`), such as Django's `path`; the view receives `label` input. | | `SuffixSink(suffix, kinds, positions=())` | A sink matched by the end of the symbol, for methods of any class such as `objects.raw`. | +| `SafeArgument(symbol, argument, values, position=None, kinds=ALL)` | A call to the sink `symbol` whose `argument` (by keyword, or at `position`) denotes one of `values` is not a sink for `kinds`, such as `yaml.load` with `Loader=yaml.SafeLoader`. Values are written as the analyzer records arguments: a symbol by its canonical name, a constant as Python writes it (`True`). Only a value given explicitly makes the call safe. | | `Validator(symbol, kinds=ALL, argument=0)` | A callable whose truth proves its argument safe, such as `re.fullmatch`; a flow guarded by it is refuted. | | `AuthorizationGuard(symbol, label)` | A decorator or a condition restricting who reaches the code, such as `login_required`; a flow behind it is a hotspot. | diff --git a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py b/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py index e3d84c5..0aefa4f 100644 --- a/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py +++ b/src/coretrace_python/bundled/models/python_stdlib/python_stdlib.py @@ -6,7 +6,15 @@ from coretrace_python.plugins import ModelPlugin from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.taint import Model, Sanitizer, Sink, Source, TaintKind, Validator +from coretrace_python.taint import ( + Model, + SafeArgument, + Sanitizer, + Sink, + Source, + TaintKind, + Validator, +) _ENVIRONMENT_KINDS = TaintKind.ALL & ~(TaintKind.COMMAND | TaintKind.PATH) _PROCESS_OUTPUT_KINDS = TaintKind.ALL & ~TaintKind.PATH @@ -66,6 +74,23 @@ class PythonStdlibModels(ModelPlugin): Sink(_sym("dill.loads"), TaintKind.DESERIALIZATION), Sink(_sym("jsonpickle.decode"), TaintKind.DESERIALIZATION), Sink(_sym("yaml.load"), TaintKind.DESERIALIZATION), + # SafeLoader and BaseLoader build plain data only, in Python and in C. + SafeArgument( + _sym("yaml.load"), + "Loader", + ( + "python.yaml.SafeLoader", + "python.yaml.loader.SafeLoader", + "python.yaml.BaseLoader", + "python.yaml.loader.BaseLoader", + "python.yaml.CSafeLoader", + "python.yaml.cyaml.CSafeLoader", + "python.yaml.CBaseLoader", + "python.yaml.cyaml.CBaseLoader", + ), + position=1, + kinds=TaintKind.DESERIALIZATION, + ), Sink(_sym("yaml.unsafe_load"), TaintKind.DESERIALIZATION), Sink(_sym("yaml.full_load"), TaintKind.DESERIALIZATION), Sanitizer(_sym("os.path.basename"), TaintKind.PATH), diff --git a/src/coretrace_python/dependency/correlation.py b/src/coretrace_python/dependency/correlation.py index 6dff7d2..f1e98e5 100644 --- a/src/coretrace_python/dependency/correlation.py +++ b/src/coretrace_python/dependency/correlation.py @@ -69,7 +69,7 @@ def check_conditions(entry: AdvisoryEntryPoint | None, arguments: Arguments | No met: list[Condition] = [] pending: list[Condition] = [] for condition in entry.conditions: - given = _given(condition, arguments) if condition.checkable and arguments is not None else None + given = arguments.given(condition.argument, condition.position) if condition.checkable and arguments is not None else None if given is None: pending.append(condition) continue @@ -84,18 +84,6 @@ def check_conditions(entry: AdvisoryEntryPoint | None, arguments: Arguments | No return ConditionCheck(tuple(met), tuple(pending)) -def _given(condition: Condition, arguments: Arguments) -> tuple[bool, str | None] | None: - """Whether the call gives the condition's argument, and what it denotes: ``(True, - value)`` when given, ``(False, None)`` when surely absent, None when it cannot tell.""" - - for name, value in arguments.keywords: - if name == condition.argument: - return True, value - if condition.position is not None and condition.position < len(arguments.positional): - return True, arguments.positional[condition.position] - return None if arguments.unpacked else (False, None) - - def ruled_out(module: str, site: CallSite, check: ConditionCheck) -> str: """Which call a contradicted condition rules out, and why: ``app:12 python.yaml.load(Loader=python.yaml.SafeLoader)``.""" diff --git a/src/coretrace_python/interprocedural/callgraph.py b/src/coretrace_python/interprocedural/callgraph.py index 85c581b..a076d10 100644 --- a/src/coretrace_python/interprocedural/callgraph.py +++ b/src/coretrace_python/interprocedural/callgraph.py @@ -72,6 +72,18 @@ class Arguments: keywords: tuple[tuple[str, str | None], ...] = () unpacked: bool = False + def given(self, keyword: str | None, position: int | None = None) -> tuple[bool, str | None] | None: + """Whether the call gives an argument, by ``keyword`` or at ``position``, and what it + denotes: ``(True, value)`` when given, ``(False, None)`` when surely absent, None + when the call cannot tell because it unpacks arguments.""" + + for name, value in self.keywords: + if name == keyword: + return True, value + if position is not None and position < len(self.positional): + return True, self.positional[position] + return None if self.unpacked else (False, None) + @dataclass(frozen=True) class CallSite: diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index 769d2cc..b3c759d 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/src/coretrace_python/taint/__init__.py @@ -16,6 +16,7 @@ ModelTable, NamedParameter, RouteRegistrar, + SafeArgument, Sanitizer, SecurityModelAnalysis, SecurityModelRegistry, @@ -39,6 +40,7 @@ "RegisteredRoutes", "RouteRegistrar", "Routes", + "SafeArgument", "Sanitizer", "SecurityModelAnalysis", "SecurityModelRegistry", diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index c1efe21..5ae2238 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -396,9 +396,9 @@ def external( ) -> Taint: """Sinks, sanitizers and sources of a call to an external symbol.""" - sink = self.models.sink(symbol) + given = self.graph.arguments_at(self.name, call.location) + sink = self.guarded(self.models.sink(symbol), given) if sink is not None: - given = self.graph.arguments_at(self.name, call.location) for position, argument in enumerate(call.arguments): self.report(flows, sink, self.rendered(argument, state), argument, call, None, None, position, given) for argument in (*call.starred, *(value for _, value in call.keywords)): @@ -413,6 +413,19 @@ def external( return everything.join(Taint(source.kinds, frozenset({source}))) return everything + def guarded(self, sink: Sink | None, arguments: Arguments) -> Sink | None: + """The sink as this call makes it: an argument declared safe (``yaml.load`` with + ``Loader=SafeLoader``) takes off the kinds it covers, and no kind left, no sink.""" + + if sink is None: + return None + safe = self.models.safe_argument(sink.symbol) + given = arguments.given(safe.argument, safe.position) if safe is not None else None + if safe is None or given is None or not given[0] or given[1] not in safe.values: + return sink + kinds = sink.kinds & ~safe.kinds + return Sink(sink.symbol, kinds, sink.positions) if kinds else None + def rendered(self, argument: Value, state: Mapping[Key, Taint]) -> Taint: """The taint an argument brings to a sink. A container literal is serialised by every framework (``self.write({...})`` answers JSON), never rendered as markup, so @@ -481,7 +494,7 @@ def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: return taint, witness for reached in summary.external_calls: - sink = self.models.sink(reached.symbol) + sink = self.guarded(self.models.sink(reached.symbol), reached.arguments) if sink is None: continue positions: list[int | None] = [*range(len(reached.argument_dependencies)), None] diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index cf903e1..e40f109 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -149,6 +149,21 @@ def symbol(self) -> SymbolId: return SymbolId(f"python.suffix.{self.suffix.replace('.', '_')}") +@dataclass(frozen=True) +class SafeArgument: + """A call to the sink ``symbol`` whose ``argument`` (by keyword, or at ``position``) + denotes one of ``values`` is not a sink for ``kinds``: ``yaml.load`` with + ``Loader=SafeLoader``. Values are written as call sites record them: a symbol by its + canonical name, a constant as Python writes it. Only a value given explicitly makes + the call safe; an absent argument, a variable or unpacked arguments leave the sink.""" + + symbol: SymbolId + argument: str + values: tuple[str, ...] + position: int | None = None + kinds: TaintKind = TaintKind.ALL + + @dataclass(frozen=True) class Validator: """A callable whose truth proves its ``argument`` safe (refutation evidence, ยง24).""" @@ -178,6 +193,7 @@ class AuthorizationGuard: | NamedParameter | RouteRegistrar | SuffixSink + | SafeArgument ) @@ -193,6 +209,7 @@ class ModelTable: named_parameters: tuple[NamedParameter, ...] = () route_registrars: tuple[RouteRegistrar, ...] = () suffix_sinks: tuple[SuffixSink, ...] = () + safe_arguments: tuple[SafeArgument, ...] = () _by_symbol: dict[type[Model], dict[SymbolId, Model]] = field( init=False, repr=False, compare=False ) @@ -207,6 +224,7 @@ def __post_init__(self) -> None: Validator: {m.symbol: m for m in self.validators}, AuthorizationGuard: {m.symbol: m for m in self.authorizations}, RouteRegistrar: {m.symbol: m for m in self.route_registrars}, + SafeArgument: {m.symbol: m for m in self.safe_arguments}, } object.__setattr__(self, "_by_symbol", MappingProxyType(index)) @@ -272,12 +290,17 @@ def extended(self, *sinks: Sink) -> ModelTable: self.named_parameters, self.route_registrars, self.suffix_sinks, + self.safe_arguments, ) def sanitizer(self, symbol: SymbolId) -> Sanitizer | None: found = self._by_symbol[Sanitizer].get(symbol) return found if isinstance(found, Sanitizer) else None + def safe_argument(self, symbol: SymbolId) -> SafeArgument | None: + found = self._by_symbol[SafeArgument].get(symbol) + return found if isinstance(found, SafeArgument) else None + class SecurityModelRegistry: """Mutable collection point for the models plugins register.""" @@ -323,6 +346,7 @@ def freeze(self) -> ModelTable: named_parameters=tuple(m for m in models if isinstance(m, NamedParameter)), route_registrars=tuple(m for m in models if isinstance(m, RouteRegistrar)), suffix_sinks=tuple(m for m in models if isinstance(m, SuffixSink)), + safe_arguments=tuple(m for m in models if isinstance(m, SafeArgument)), )