feat: FastAPI and CLI precision

README.md

@@ -96,15 +96,19 @@ memory.
 
 `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be
 repeated. The repository ships a first set under `plugins/`: security models for the
-standard library, Flask, FastAPI, Django, SQLAlchemy and the Requests and httpx clients (`models/`), taint detectors for SQL
-injection, command injection, path traversal, SSRF, XSS and plaintext credential storage
-(`security/`), and syntactic
+standard library, Flask, FastAPI, Django, SQLAlchemy, the Requests and httpx clients, and the
+click and Typer command-line frameworks (`models/`), taint detectors for SQL
+injection, command injection, path traversal, SSRF, XSS, insecure deserialization, open
+redirects and plaintext credential storage (`security/`), and syntactic
 detectors for `eval`/`exec`, weak hashes, `app.run(debug=True)` and HTTP client calls
 without a timeout (`syntax/`). Route handlers of Flask and
 FastAPI receive their parameters as HTTP input, whether the application is created
 directly (`app = Flask(__name__)`) or by a project factory whose summary returns it
 (`app = create_app()`); `request.args` and its siblings are HTTP
-sources. A parameter named `password` or the like is a credential wherever it appears: stored
+sources; the parameters of a click or Typer command are `argv` input. SQL sinks read the
+statement only, so a tainted value in the parameter tuple of a parameterised query is not an
+injection, and an annotated parameter denotes its class, so `db: Session = Depends(get_db)`
+gives `db.execute` its sink. A parameter named `password` or the like is a credential wherever it appears: stored
 in a database without passing through a hashing function it is a
 `plaintext-credential-storage` finding at medium confidence, since a name is a hint.
 Django views are undecorated, so a parameter annotated with a request class

plugins/models/cli/cli_models.py

@@ -0,0 +1,35 @@
+"""Command-line entry points: the parameters of click and Typer commands are argv input."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from coretrace_python.plugins import ModelPlugin
+from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.taint import EntryPoint, Model, TaintKind
+
+# A command-line tool is expected to open the paths it is given: argv input carries every
+# kind but PATH.
+ARGV_KINDS = TaintKind.ALL & ~TaintKind.PATH
+
+_COMMANDS = (
+    "click.command",
+    "click.group",
+    "click.command.command",
+    "click.group.command",
+    "click.group.group",
+    "click.Group.command",
+    "typer.Typer.command",
+    "typer.Typer.callback",
+)
+
+
+def _sym(path: str) -> SymbolId:
+    return SymbolId(f"python.{path}")
+
+
+class CliModels(ModelPlugin):
+    name: ClassVar[str] = "cli-models"
+    models: ClassVar[tuple[Model, ...]] = tuple(
+        EntryPoint(_sym(command), "argv", ARGV_KINDS) for command in _COMMANDS
+    )

plugins/models/cli/plugin.toml

@@ -0,0 +1,9 @@
+name = "cli-models"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = []
+provides = ["model.cli-sources"]
+
+[entrypoint]
+module = "cli_models"
+class = "CliModels"

plugins/models/django/django_models.py

@@ -86,8 +86,8 @@ class DjangoModels(ModelPlugin):
         *(TypedParameter(_sym(cls), "http") for cls in _REQUEST_CLASSES),
         *(EntryPoint(_sym(base), "http") for base in _VIEW_BASES),
         *(EntryPoint(_sym(decorator), "http") for decorator in _VIEW_DECORATORS),
-        Sink(_sym("django.db.connection.cursor.execute"), TaintKind.SQL | TaintKind.CREDENTIAL),
-        Sink(_sym("django.db.connection.cursor.executemany"), TaintKind.SQL | TaintKind.CREDENTIAL),
+        Sink(_sym("django.db.connection.cursor.execute"), TaintKind.SQL | TaintKind.CREDENTIAL, ((TaintKind.SQL, (0,)),)),
+        Sink(_sym("django.db.connection.cursor.executemany"), TaintKind.SQL | TaintKind.CREDENTIAL, ((TaintKind.SQL, (0,)),)),
         Sink(_sym("django.db.models.expressions.RawSQL"), TaintKind.SQL | TaintKind.CREDENTIAL),
         Sink(_sym("django.db.models.Manager.raw"), TaintKind.SQL | TaintKind.CREDENTIAL),
         Sink(_sym("django.db.models.QuerySet.raw"), TaintKind.SQL | TaintKind.CREDENTIAL),
@@ -97,6 +97,9 @@ class DjangoModels(ModelPlugin):
         Sink(_sym("django.http.response.HttpResponse"), TaintKind.HTML),
         Sink(_sym("django.template.Template"), TaintKind.HTML),
         Sink(_sym("django.http.FileResponse"), TaintKind.PATH),
+        Sink(_sym("django.shortcuts.redirect"), TaintKind.REDIRECT),
+        Sink(_sym("django.http.HttpResponseRedirect"), TaintKind.REDIRECT),
+        Sink(_sym("django.http.HttpResponsePermanentRedirect"), TaintKind.REDIRECT),
         Sanitizer(_sym("django.utils.html.escape"), TaintKind.HTML),
         Sanitizer(_sym("django.utils.html.conditional_escape"), TaintKind.HTML),
         *(AuthorizationGuard(_sym(decorator), label) for decorator, label in _AUTHORIZATION_DECORATORS),

plugins/models/fastapi/fastapi_models.py

@@ -22,4 +22,6 @@ class FastApiModels(ModelPlugin):
         *(EntryPoint(_sym(f"fastapi.APIRouter.{method}"), "http") for method in _METHODS),
         Sink(_sym("fastapi.responses.HTMLResponse"), TaintKind.HTML),
         Sink(_sym("fastapi.responses.FileResponse"), TaintKind.PATH),
+        Sink(_sym("fastapi.responses.RedirectResponse"), TaintKind.REDIRECT),
+        Sink(_sym("starlette.responses.RedirectResponse"), TaintKind.REDIRECT),
     )

plugins/models/flask/flask_models.py

@@ -37,6 +37,8 @@ class FlaskModels(ModelPlugin):
         Sink(_sym("flask.Response"), TaintKind.HTML),
         Sink(_sym("flask.Markup"), TaintKind.HTML),
         Sink(_sym("flask.send_file"), TaintKind.PATH),
+        Sink(_sym("flask.redirect"), TaintKind.REDIRECT),
+        Sink(_sym("werkzeug.utils.redirect"), TaintKind.REDIRECT),
         Sink(_sym("flask.request.files.save"), TaintKind.PATH),
         Sanitizer(_sym("werkzeug.utils.secure_filename"), TaintKind.PATH),
         Sanitizer(_sym("flask.escape"), TaintKind.HTML),

plugins/models/python_stdlib/python_stdlib.py

@@ -18,7 +18,8 @@ class PythonStdlibModels(ModelPlugin):
     models: ClassVar[tuple[Model, ...]] = (
         Source(_sym("builtins.input"), "stdin"),
         Source(_sym("sys.stdin"), "stdin"),
-        Source(_sym("sys.argv"), "argv"),
+        # A command-line tool is expected to open the paths it is given.
+        Source(_sym("sys.argv"), "argv", TaintKind.ALL & ~TaintKind.PATH),
         Source(_sym("os.environ"), "environment"),
         Source(_sym("subprocess.run.stdout"), "process-output"),
         Source(_sym("subprocess.check_output"), "process-output"),
@@ -40,12 +41,27 @@ class PythonStdlibModels(ModelPlugin):
         Sink(_sym("shutil.rmtree"), TaintKind.PATH),
         Sink(_sym("urllib.request.urlopen"), TaintKind.SSRF),
         *(
-            Sink(_sym(f"sqlite3.connect{cursor}.{method}"), TaintKind.SQL | TaintKind.CREDENTIAL)
+            Sink(
+                _sym(f"sqlite3.connect{cursor}.{method}"),
+                TaintKind.SQL | TaintKind.CREDENTIAL,
+                ((TaintKind.SQL, (0,)),),
+            )
             for cursor in ("", ".cursor")
             for method in ("execute", "executemany", "executescript")
         ),
         Sanitizer(_sym("shlex.quote"), TaintKind.COMMAND),
         Sanitizer(_sym("html.escape"), TaintKind.HTML),
+        Sink(_sym("pickle.loads"), TaintKind.DESERIALIZATION),
+        Sink(_sym("pickle.load"), TaintKind.DESERIALIZATION),
+        Sink(_sym("pickle.Unpickler"), TaintKind.DESERIALIZATION),
+        Sink(_sym("marshal.loads"), TaintKind.DESERIALIZATION),
+        Sink(_sym("marshal.load"), TaintKind.DESERIALIZATION),
+        Sink(_sym("shelve.open"), TaintKind.DESERIALIZATION),
+        Sink(_sym("dill.loads"), TaintKind.DESERIALIZATION),
+        Sink(_sym("jsonpickle.decode"), TaintKind.DESERIALIZATION),
+        Sink(_sym("yaml.load"), TaintKind.DESERIALIZATION),
+        Sink(_sym("yaml.unsafe_load"), TaintKind.DESERIALIZATION),
+        Sink(_sym("yaml.full_load"), TaintKind.DESERIALIZATION),
         Sanitizer(_sym("os.path.basename"), TaintKind.PATH),
         Validator(_sym("re.fullmatch"), argument=1),
         Validator(_sym("re.compile.fullmatch")),

plugins/models/sqlalchemy/sqlalchemy_models.py

@@ -11,7 +11,7 @@
 
 from coretrace_python.plugins import ModelPlugin
 from coretrace_python.semantic.symbols import SymbolId
-from coretrace_python.taint import Model, Sink, TaintKind
+from coretrace_python.taint import Model, Sanitizer, Sink, TaintKind
 
 _EXECUTORS = (
     "sqlalchemy.create_engine",
@@ -25,14 +25,23 @@
 )
 
 
+_TEXT_MODULES = ("sqlalchemy", "sqlalchemy.sql", "sqlalchemy.sql.expression")
+
+
 def _sym(path: str) -> SymbolId:
     return SymbolId(f"python.{path}")
 
 
 class SqlAlchemyModels(ModelPlugin):
     name: ClassVar[str] = "sqlalchemy-models"
-    models: ClassVar[tuple[Model, ...]] = tuple(
-        Sink(_sym(f"{executor}.{method}"), TaintKind.SQL | TaintKind.CREDENTIAL)
-        for executor in _EXECUTORS
-        for method in ("execute", "exec_driver_sql", "scalar", "scalars")
+    models: ClassVar[tuple[Model, ...]] = (
+        *(
+            Sink(_sym(f"{executor}.{method}"), TaintKind.SQL | TaintKind.CREDENTIAL, ((TaintKind.SQL, (0,)),))
+            for executor in _EXECUTORS
+            for method in ("execute", "exec_driver_sql", "scalar", "scalars")
+        ),
+        # ``text()`` is where SQL text is committed: a sink, and its result carries no
+        # further SQL taint so the ``execute`` that follows does not report it twice.
+        *(Sink(_sym(f"{module}.text"), TaintKind.SQL) for module in _TEXT_MODULES),
+        *(Sanitizer(_sym(f"{module}.text"), TaintKind.SQL) for module in _TEXT_MODULES),
     )

plugins/security/insecure_deserialization/insecure_deserialization.py

@@ -0,0 +1,17 @@
+"""Insecure deserialization: attacker-controlled input reaching a insecure deserialization sink."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from coretrace_python.findings import Severity
+from coretrace_python.plugins import TaintDetector
+from coretrace_python.taint import TaintKind
+
+
+class InsecureDeserializationPlugin(TaintDetector):
+    name: ClassVar[str] = "insecure-deserialization"
+    rule_id: ClassVar[str] = "insecure-deserialization"
+    kind: ClassVar[TaintKind] = TaintKind.DESERIALIZATION
+    severity: ClassVar[Severity] = Severity.CRITICAL
+    title: ClassVar[str] = "Insecure deserialization"

plugins/security/insecure_deserialization/plugin.toml

@@ -0,0 +1,9 @@
+name = "insecure-deserialization"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = ["taint.flows", "findings.refutation"]
+provides = ["vulnerability.insecure-deserialization"]
+
+[entrypoint]
+module = "insecure_deserialization"
+class = "InsecureDeserializationPlugin"

plugins/security/open_redirect/open_redirect.py

@@ -0,0 +1,17 @@
+"""Open redirect: attacker-controlled input reaching a open redirect sink."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from coretrace_python.findings import Severity
+from coretrace_python.plugins import TaintDetector
+from coretrace_python.taint import TaintKind
+
+
+class OpenRedirectPlugin(TaintDetector):
+    name: ClassVar[str] = "open-redirect"
+    rule_id: ClassVar[str] = "open-redirect"
+    kind: ClassVar[TaintKind] = TaintKind.REDIRECT
+    severity: ClassVar[Severity] = Severity.MEDIUM
+    title: ClassVar[str] = "Open redirect"

plugins/security/open_redirect/plugin.toml

@@ -0,0 +1,9 @@
+name = "open-redirect"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = ["taint.flows", "findings.refutation"]
+provides = ["vulnerability.open-redirect"]
+
+[entrypoint]
+module = "open_redirect"
+class = "OpenRedirectPlugin"

src/coretrace_python/interprocedural/callgraph.py

@@ -29,7 +29,7 @@
 )
 from coretrace_python.ir.ssa import SSAAnalysis
 from coretrace_python.semantic.scopes import BindingKind, ScopeAnalysis, ScopeTable
-from coretrace_python.semantic.symbols import SymbolId
+from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, SymbolTable
 from coretrace_python.source import SourceSpan
 
 
@@ -108,13 +108,15 @@ def callers(self, name: str) -> frozenset[str]:
         return self._callers.get(name, frozenset())
 
 
-def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]:
+def derive_symbols(
+    function: FunctionIR, initial: Mapping[Value, SymbolId] | None = None
+) -> dict[Value, SymbolId]:
     """Symbols of values: ``Symbol`` results, attributes and items of symbol values,
     results of calling a symbol (``sqlite3.connect(p)`` denotes ``python.sqlite3.connect``)
-    and the values a ``with`` on such a result binds. Known functions and parameters
-    derive nothing."""
+    and the values a ``with`` on such a result binds. Known functions derive nothing;
+    parameters only through ``initial``, their annotated classes."""
 
-    symbols: dict[Value, SymbolId] = {}
+    symbols: dict[Value, SymbolId] = dict(initial or {})
     changed = True
     while changed:
         changed = False
@@ -143,12 +145,15 @@ def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]:
 
 
 def resolve_targets(
-    function: FunctionIR, scopes: ScopeTable, known: frozenset[str]
+    function: FunctionIR,
+    scopes: ScopeTable,
+    known: frozenset[str],
+    parameters: Mapping[Value, SymbolId] | None = None,
 ) -> tuple[dict[Value, Target], dict[Value, SymbolId]]:
     """Map every callee value of ``function`` to its target, and every symbol value."""
 
     module = scopes.module_scope
-    symbols = derive_symbols(function)
+    symbols = derive_symbols(function, parameters)
     targets: dict[Value, Target] = {
         value: ExternalSymbol(symbol) for value, symbol in symbols.items()
     }
@@ -165,13 +170,31 @@ def resolve_targets(
     return targets, symbols
 
 
+def _annotated(
+    function: nodes.Function, ssa: FunctionIR, scopes: ScopeTable, table: SymbolTable
+) -> dict[Value, SymbolId]:
+    """Parameters annotated with a resolvable class denote that class (``db: Session``)."""
+
+    scope = scopes.scope_for(function)
+    enclosing = scope.parent if scope.parent is not None else scope.id
+    found: dict[Value, SymbolId] = {}
+    for value, parameter in zip(ssa.parameters, function.parameters, strict=True):
+        if parameter.annotation is None:
+            continue
+        symbol = table.resolve_expression(enclosing, parameter.annotation)
+        if symbol is not None:
+            found[value] = symbol
+    return found
+
+
 class CallGraphAnalysis(Analysis[CallGraph]):
     name: ClassVar[str] = "interprocedural.callgraph"
-    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, ScopeAnalysis})
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({SSAAnalysis, ScopeAnalysis, SymbolAnalysis})
 
     @classmethod
     def compute(cls, ctx: AnalysisContext) -> CallGraph:
         scopes = ctx.get(ScopeAnalysis)
+        table = ctx.get(SymbolAnalysis)
         definitions: dict[str, nodes.Function] = {}
         for function in analyzable_functions(ctx.module):
             # A property and its setter, or a redefinition, share a qualified name; each
@@ -195,7 +218,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
                 unsupported.add(name)
                 sites[name] = ()
                 continue
-            targets, symbols[name] = resolve_targets(ssa, scopes, known)
+            targets, symbols[name] = resolve_targets(ssa, scopes, known, _annotated(function, ssa, scopes, table))
             found: list[CallSite] = []
             for block in ssa.blocks:
                 for instruction in block.instructions: