feat: let a safe argument take kinds off a sink call

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

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. |
 

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),

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)``."""

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:

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",

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]

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)),
         )
 
 

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

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