feat: recognise validators declared on project functions

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 - A sanitizer called inside a project function protects the flows through that function, for every caller. That covers a helper that escapes before answering, a view redirecting to `reverse(...)`, and a helper rendering a template the analyzer shows escaping. Function summaries keep, for each parameter, the taint kinds cleared on every path from it to an external call or to the return value. A path that skips the sanitizer, or a kind it does not clear, still reaches the sink. The cache format is bumped (#134).
 
+### Plugins
+
+- A `Validator` model may name a function of the project by its project symbol (`python.hc.accounts.views._allow_redirect`), and the refutation recognises it in its own module too, where a call to it has no imported symbol. A project can thus declare its own validation helpers, and a flow they guard is refuted (#134).
+
 ## 0.7.0 (2026-09-25)
 
 ### Precision

docs/plugins.md

@@ -150,7 +150,7 @@ class BottleModels(ModelPlugin):
 | `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. |
 | `TemplateRender(symbol, position=0, keyword="template_name")` | A call rendering, with autoescaping, the template named by its argument at `position` or `keyword`, such as `render_to_string`. What it returns carries no `HTML` when the name is written in the call and the project's template escapes everything it prints (see the usage guide). |
-| `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. |
+| `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. `symbol` may be a function of the project, by its project symbol (`python.hc.accounts.views._allow_redirect`), which then counts wherever it is called, in its own module too. |
 | `AuthorizationGuard(symbol, label)` | A decorator or a condition restricting who reaches the code, such as `login_required`; a flow behind it is a hotspot. |
 
 Two plugins may describe the same symbol. An identical model is registered once; two

src/coretrace_python/findings/refutation.py

@@ -29,7 +29,12 @@
 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.interprocedural import CallGraphAnalysis
+from coretrace_python.interprocedural import (
+    CallGraph,
+    CallGraphAnalysis,
+    KnownFunction,
+    project_symbol,
+)
 from coretrace_python.ir.model import (
     BoolOp,
     Branch,
@@ -478,6 +483,7 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> Verdicts:
         constants = ctx.get(ConstantPropagation, function)
         graph = ctx.get(CallGraphAnalysis)
         models = ctx.get(SecurityModelAnalysis)
+        name = graph.name_of(function)
         return judge_flows(
             ssa,
             ctx.get(CFGAnalysis, function),
@@ -486,7 +492,21 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> Verdicts:
             frozenset(b.id for b in ssa.blocks if constants.reachable(b.id)),
             ctx.get(RangeAnalysis, function),
             models,
-            graph.symbols(graph.name_of(function)),
+            _with_project_callees(ssa, graph, name, ctx.module.name),
             authorization_of(function, models, ctx.get(ScopeAnalysis), ctx.get(SymbolAnalysis)),
         )
 
+
+def _with_project_callees(function: FunctionIR, graph: CallGraph, name: str, module: str) -> dict[Value, SymbolId]:
+    """The symbols of ``name``'s values, with each call to a function of its module named
+    by its project symbol, so a model declared on it (``Validator``) applies there too."""
+
+    symbols = dict(graph.symbols(name))
+    for block in function.blocks:
+        for instruction in block.instructions:
+            if isinstance(instruction, Call):
+                target = graph.target_at(name, instruction.location)
+                if isinstance(target, KnownFunction):
+                    symbols.setdefault(instruction.callee, project_symbol(module, target.name))
+    return symbols
+

tests/test_project_validators.py

@@ -0,0 +1,83 @@
+"""A project's own function, declared as a ``Validator``, proves what it validates.
+
+healthchecks guards ``redirect(redirect_url)`` with ``_allow_redirect(redirect_url)``, a
+function of the same module that accepts relative URLs resolving to a known route. A
+``Validator`` model names such a function by its project symbol
+(``python.<module>.<function>``, as ``project_symbol`` builds it), and the refutation then
+recognises a call to it from its own module, as it does for an imported one. An
+undeclared function remains a guard that proves nothing.
+"""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from coretrace_python import engine
+from coretrace_python.findings.refutation import RefutationAnalysis, Status
+from coretrace_python.frontend import build_hir
+from coretrace_python.hir import nodes
+from coretrace_python.plugins import ModelPlugin
+from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.source import SourceManager
+from coretrace_python.taint import Model, SecurityModelRegistry, Sink, Source, TaintKind, Validator
+
+VIEW = (
+    "import os\n\n"
+    "def run():\n"
+    "    target = input()\n"
+    "    if allowed(target):\n"
+    "        os.system(target)\n\n"
+    "def allowed(value):\n"
+    "    return value in ('ls', 'pwd')\n"
+)
+
+
+def verdict(*models: Model) -> Status:
+    module = build_hir(SourceManager().add_source("views.py", VIEW))
+    registry = SecurityModelRegistry()
+    registry.register(
+        Source(SymbolId("python.builtins.input"), "stdin"), Sink(SymbolId("python.os.system"), TaintKind.COMMAND), *models
+    )
+    manager = engine.build_manager(module, registry)
+    run = next(s for s in module.body if isinstance(s, nodes.Function) and s.name == "run")
+    (found,) = manager.get(RefutationAnalysis, run).all()
+    return found.status
+
+
+def test_a_declared_project_validator_refutes_the_flow_it_guards() -> None:
+    assert verdict(Validator(SymbolId("python.views.allowed"))) is Status.REFUTED
+
+
+def test_an_undeclared_project_function_leaves_a_hotspot() -> None:
+    assert verdict() is Status.HOTSPOT
+    assert verdict(Validator(SymbolId("python.other.allowed"))) is Status.HOTSPOT
+
+
+class _AllowRedirect(ModelPlugin):
+    name: ClassVar[str] = "allow-redirect"
+    models: ClassVar[tuple[Model, ...]] = (Validator(SymbolId("python.hc.views._allow_redirect")),)
+
+
+def test_a_plugin_declares_the_validator_of_a_project(tmp_path) -> None:  # type: ignore[no-untyped-def]
+    source = (
+        "from django.http import HttpRequest\n"
+        "from django.shortcuts import redirect\n\n"
+        "def _allow_redirect(url):\n"
+        "    return url.startswith('/') and not url.startswith('//')\n\n"
+        "def login(request: HttpRequest):\n"
+        "    target = request.GET['next']\n"
+        "    if target and _allow_redirect(target):\n"
+        "        return redirect(target)\n"
+        "    return redirect('/')\n"
+    )
+    (tmp_path / "hc").mkdir()
+    (tmp_path / "hc" / "__init__.py").write_text("", encoding="utf-8")
+    (tmp_path / "hc" / "views.py").write_text(source, encoding="utf-8")
+
+    declared = engine.analyze_project(tmp_path, [engine.BUNDLED_PLUGINS], plugins=[_AllowRedirect()]).findings
+    undeclared = engine.analyze_project(tmp_path, [engine.BUNDLED_PLUGINS]).findings
+
+    assert [f.rule_id for f in declared if f.rule_id == "open-redirect"] == []
+    assert [(f.rule_id, f.metadata["verdict"]) for f in undeclared if f.rule_id == "open-redirect"] == [
+        ("open-redirect", "hotspot")
+    ]