feat: framework models for Flask, FastAPI and SQLAlchemy

README.md

@@ -32,10 +32,12 @@ coretrace-python-analyzer --check app.py --plugins plugins/ --format sarif > rep
 ```
 
 `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be
-repeated. The repository ships a first set under `plugins/`: standard-library security
-models (`models/python_stdlib`), taint detectors for SQL injection, command injection,
-path traversal, SSRF and XSS (`security/`), and syntactic detectors for `eval`/`exec`
-and weak hashes (`syntax/`). Model plugins contribute sources, sinks and sanitizers;
+repeated. The repository ships a first set under `plugins/`: security models for the
+standard library, Flask, FastAPI and SQLAlchemy (`models/`), taint detectors for SQL
+injection, command injection, path traversal, SSRF and XSS (`security/`), and syntactic
+detectors for `eval`/`exec` and weak hashes (`syntax/`). Route handlers of Flask and
+FastAPI receive their parameters as HTTP input; `request.args` and its siblings are HTTP
+sources. Model plugins contribute sources, sinks and sanitizers;
 detectors consume the shared taint result, so adding a framework model makes every
 detector aware of it. Taint follows calls between functions of the same module through
 function summaries: a tainted argument passed to a helper that reaches a sink is reported
@@ -64,6 +66,9 @@ instructions and unassigned paths read an explicit `undefined` value.
 Unsupported syntax produces a source-located diagnostic and a non-zero exit status.
 
 Names are resolved to canonical symbols through lexical scopes, imports and builtins.
+Calling a symbol denotes that symbol, so `app = Flask(__name__)` makes `app.route` resolve
+to `python.flask.Flask.route` and `sqlite3.connect(p).cursor().execute` resolves to
+`python.sqlite3.connect.cursor.execute`; model plugins register those chains.
 Import aliases do not change an API's identity, so all of the following resolve to
 `python.os.system`:
 

plugins/models/fastapi/fastapi_models.py

@@ -0,0 +1,25 @@
+"""FastAPI security models: route handlers and response sinks."""
+
+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, Sink, TaintKind
+
+_METHODS = ("get", "post", "put", "delete", "patch", "options", "head", "api_route", "websocket")
+
+
+def _sym(path: str) -> SymbolId:
+    return SymbolId(f"python.{path}")
+
+
+class FastApiModels(ModelPlugin):
+    name: ClassVar[str] = "fastapi-models"
+    models: ClassVar[tuple[Model, ...]] = (
+        *(EntryPoint(_sym(f"fastapi.FastAPI.{method}"), "http") for method in _METHODS),
+        *(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),
+    )

plugins/models/fastapi/plugin.toml

@@ -0,0 +1,9 @@
+name = "fastapi-models"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = []
+provides = ["model.http-sources", "model.fastapi-routes"]
+
+[entrypoint]
+module = "fastapi_models"
+class = "FastApiModels"

plugins/models/flask/flask_models.py

@@ -0,0 +1,34 @@
+"""Flask security models: HTTP sources, route handlers, HTML sinks and escaping."""
+
+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, Sanitizer, Sink, Source, TaintKind
+
+_REQUEST_ATTRIBUTES = (
+    "args", "form", "values", "json", "data", "cookies", "headers", "files",
+    "get_json", "get_data", "url", "full_path", "path", "query_string", "stream",
+)
+
+
+def _sym(path: str) -> SymbolId:
+    return SymbolId(f"python.{path}")
+
+
+class FlaskModels(ModelPlugin):
+    name: ClassVar[str] = "flask-models"
+    models: ClassVar[tuple[Model, ...]] = (
+        *(Source(_sym(f"flask.request.{attribute}"), "http") for attribute in _REQUEST_ATTRIBUTES),
+        EntryPoint(_sym("flask.Flask.route"), "http"),
+        EntryPoint(_sym("flask.Blueprint.route"), "http"),
+        Sink(_sym("flask.render_template_string"), TaintKind.HTML),
+        Sink(_sym("flask.make_response"), TaintKind.HTML),
+        Sink(_sym("flask.Response"), TaintKind.HTML),
+        Sink(_sym("flask.Markup"), TaintKind.HTML),
+        Sink(_sym("flask.send_file"), TaintKind.PATH),
+        Sanitizer(_sym("flask.escape"), TaintKind.HTML),
+        Sanitizer(_sym("markupsafe.escape"), TaintKind.HTML),
+    )

plugins/models/flask/plugin.toml

@@ -0,0 +1,9 @@
+name = "flask-models"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = []
+provides = ["model.http-sources", "model.flask-routes"]
+
+[entrypoint]
+module = "flask_models"
+class = "FlaskModels"

plugins/models/python_stdlib/python_stdlib.py

@@ -35,6 +35,11 @@ class PythonStdlibModels(ModelPlugin):
         Sink(_sym("os.rmdir"), TaintKind.PATH),
         Sink(_sym("shutil.rmtree"), TaintKind.PATH),
         Sink(_sym("urllib.request.urlopen"), TaintKind.SSRF),
+        *(
+            Sink(_sym(f"sqlite3.connect{cursor}.{method}"), TaintKind.SQL)
+            for cursor in ("", ".cursor")
+            for method in ("execute", "executemany", "executescript")
+        ),
         Sanitizer(_sym("shlex.quote"), TaintKind.COMMAND),
         Sanitizer(_sym("html.escape"), TaintKind.HTML),
         Sanitizer(_sym("os.path.basename"), TaintKind.PATH),

plugins/models/sqlalchemy/plugin.toml

@@ -0,0 +1,9 @@
+name = "sqlalchemy-models"
+version = "1.0.0"
+plugin_api = ">=1,<2"
+requires = []
+provides = ["model.sql-sinks"]
+
+[entrypoint]
+module = "sqlalchemy_models"
+class = "SqlAlchemyModels"

plugins/models/sqlalchemy/sqlalchemy_models.py

@@ -0,0 +1,38 @@
+"""SQLAlchemy security models: raw SQL execution sinks.
+
+Symbols follow the engine's call-chain rule: the result of calling a symbol denotes
+that symbol, so ``create_engine(url).connect().execute`` is
+``python.sqlalchemy.create_engine.connect.execute``.
+"""
+
+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 Model, Sink, TaintKind
+
+_EXECUTORS = (
+    "sqlalchemy.create_engine",
+    "sqlalchemy.create_engine.connect",
+    "sqlalchemy.create_engine.begin",
+    "sqlalchemy.engine.create_engine",
+    "sqlalchemy.engine.create_engine.connect",
+    "sqlalchemy.orm.Session",
+    "sqlalchemy.orm.sessionmaker",
+    "sqlalchemy.orm.scoped_session",
+)
+
+
+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)
+        for executor in _EXECUTORS
+        for method in ("execute", "exec_driver_sql", "scalar", "scalars")
+    )

src/coretrace_python/interprocedural/callgraph.py

@@ -17,7 +17,7 @@
 from coretrace_python.cfg import CFGError
 from coretrace_python.hir import nodes
 from coretrace_python.ir.lowering import LoweringError, analyzable_functions, qualified_name
-from coretrace_python.ir.model import Call, FunctionIR, Global, Symbol, Value
+from coretrace_python.ir.model import Call, FunctionIR, GetAttr, Global, Symbol, Value, WithEnter
 from coretrace_python.ir.ssa import SSAAnalysis
 from coretrace_python.semantic.scopes import BindingKind, ScopeAnalysis, ScopeTable
 from coretrace_python.semantic.symbols import SymbolId
@@ -57,7 +57,9 @@ def __init__(
         definitions: Mapping[str, nodes.Function],
         sites: Mapping[str, tuple[CallSite, ...]],
         unsupported: frozenset[str],
+        symbols: Mapping[str, Mapping[Value, SymbolId]] | None = None,
     ) -> None:
+        self._symbols = {name: MappingProxyType(dict(found)) for name, found in (symbols or {}).items()}
         self.definitions: Mapping[str, nodes.Function] = MappingProxyType(dict(definitions))
         self.functions = tuple(definitions)
         self._names = {function.span: name for name, function in definitions.items()}
@@ -76,6 +78,11 @@ def __init__(
     def name_of(self, function: nodes.Function) -> str:
         return self._names[function.span]
 
+    def symbols(self, name: str) -> Mapping[Value, SymbolId]:
+        """Values of ``name`` that denote a symbol, including derived call-chain symbols."""
+
+        return self._symbols.get(name, MappingProxyType({}))
+
     def sites(self, caller: str) -> tuple[CallSite, ...]:
         return self._sites.get(caller, ())
 
@@ -91,26 +98,56 @@ def callers(self, name: str) -> frozenset[str]:
         return self._callers.get(name, frozenset())
 
 
+def derive_symbols(function: FunctionIR) -> dict[Value, SymbolId]:
+    """Symbols of values: ``Symbol`` results, attributes 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."""
+
+    symbols: dict[Value, SymbolId] = {}
+    changed = True
+    while changed:
+        changed = False
+        for block in function.blocks:
+            for instruction in block.instructions:
+                if instruction.result is None or instruction.result in symbols:
+                    continue
+                symbol: SymbolId | None = None
+                if isinstance(instruction, Symbol):
+                    symbol = instruction.symbol_id
+                elif isinstance(instruction, GetAttr) and instruction.object in symbols:
+                    symbol = symbols[instruction.object].attribute(instruction.attribute)
+                elif isinstance(instruction, Call | WithEnter):
+                    origin = (
+                        instruction.callee if isinstance(instruction, Call) else instruction.context
+                    )
+                    symbol = symbols.get(origin)
+                if symbol is not None:
+                    symbols[instruction.result] = symbol
+                    changed = True
+    return symbols
+
+
 def resolve_targets(
     function: FunctionIR, scopes: ScopeTable, known: frozenset[str]
-) -> dict[Value, Target]:
-    """Map every callee value of ``function`` to its target."""
+) -> tuple[dict[Value, Target], dict[Value, SymbolId]]:
+    """Map every callee value of ``function`` to its target, and every symbol value."""
 
     module = scopes.module_scope
-    targets: dict[Value, Target] = {}
+    symbols = derive_symbols(function)
+    targets: dict[Value, Target] = {
+        value: ExternalSymbol(symbol) for value, symbol in symbols.items()
+    }
     for block in function.blocks:
         for instruction in block.instructions:
-            if isinstance(instruction, Symbol):
-                targets[instruction.result] = ExternalSymbol(instruction.symbol_id)
-            elif isinstance(instruction, Global):
+            if isinstance(instruction, Global):
                 binding = module.bindings.get(instruction.name)
                 if (
                     binding is not None
                     and binding.kind is BindingKind.FUNCTION
                     and instruction.name in known
                 ):
                     targets[instruction.result] = KnownFunction(instruction.name)
-    return targets
+    return targets, symbols
 
 
 class CallGraphAnalysis(Analysis[CallGraph]):
@@ -128,6 +165,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
             name for name, function in definitions.items() if function in ctx.module.body
         )
         sites: dict[str, tuple[CallSite, ...]] = {}
+        symbols: dict[str, Mapping[Value, SymbolId]] = {}
         unsupported: set[str] = set()
         for name, function in definitions.items():
             try:
@@ -136,7 +174,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
                 unsupported.add(name)
                 sites[name] = ()
                 continue
-            targets = resolve_targets(ssa, scopes, known)
+            targets, symbols[name] = resolve_targets(ssa, scopes, known)
             found: list[CallSite] = []
             for block in ssa.blocks:
                 for instruction in block.instructions:
@@ -151,4 +189,4 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
                             )
                         )
             sites[name] = tuple(found)
-        return CallGraph(definitions, sites, frozenset(unsupported))
+        return CallGraph(definitions, sites, frozenset(unsupported), symbols)

src/coretrace_python/semantic/symbols.py

@@ -3,9 +3,12 @@
 from __future__ import annotations
 
 import builtins
+from collections.abc import Mapping
+from types import MappingProxyType
 from typing import ClassVar
 
 from coretrace_python.analysis import Analysis, AnalysisContext, AnyAnalysis
+from coretrace_python.hir import nodes
 from coretrace_python.semantic.identity import SymbolId
 from coretrace_python.semantic.imports import ImportAnalysis, ImportTable
 from coretrace_python.semantic.scopes import ResolutionKind, ScopeAnalysis, ScopeId, ScopeTable
@@ -14,22 +17,62 @@
 
 
 class SymbolTable:
-    """Resolve names to canonical symbols through scopes, imports and builtins."""
+    """Resolve names to canonical symbols through scopes, imports, builtins and the
+    module-level instances created by calling a symbol (``app = Flask(__name__)``)."""
 
-    def __init__(self, scopes: ScopeTable, imports: ImportTable) -> None:
+    def __init__(
+        self,
+        scopes: ScopeTable,
+        imports: ImportTable,
+        instances: Mapping[str, SymbolId] | None = None,
+    ) -> None:
         self._scopes = scopes
         self._imports = imports
+        self._instances = MappingProxyType(dict(instances or {}))
 
     def resolve(self, scope_id: ScopeId, name: str) -> SymbolId | None:
         resolution = self._scopes.resolve(scope_id, name)
         if resolution.kind is ResolutionKind.UNBOUND:
             return SymbolId(f"python.builtins.{name}") if name in BUILTIN_NAMES else None
         assert resolution.scope is not None
-        return self._imports.bindings(resolution.scope).get(name)
+        imported = self._imports.bindings(resolution.scope).get(name)
+        if imported is not None:
+            return imported
+        if resolution.scope == self._scopes.module_scope.id:
+            return self._instances.get(name)
+        return None
 
+    def resolve_expression(self, scope_id: ScopeId, node: nodes.Expression) -> SymbolId | None:
+        """The symbol of a name or attribute chain, or of the callee of a call."""
 
-def analyze_symbols(scopes: ScopeTable, imports: ImportTable) -> SymbolTable:
-    return SymbolTable(scopes, imports)
+        if isinstance(node, nodes.Name):
+            return self.resolve(scope_id, node.identifier)
+        if isinstance(node, nodes.Attribute):
+            parent = self.resolve_expression(scope_id, node.value)
+            return parent.attribute(node.name) if parent is not None else None
+        if isinstance(node, nodes.Call):
+            return self.resolve_expression(scope_id, node.callee)
+        return None
+
+
+def analyze_symbols(
+    scopes: ScopeTable, imports: ImportTable, module: nodes.Module | None = None
+) -> SymbolTable:
+    table = SymbolTable(scopes, imports)
+    if module is None:
+        return table
+    instances: dict[str, SymbolId] = {}
+    module_scope = scopes.module_scope.id
+    for statement in module.body:
+        if (
+            isinstance(statement, nodes.Assign)
+            and isinstance(statement.target, nodes.Name)
+            and isinstance(statement.value, nodes.Call)
+        ):
+            symbol = table.resolve_expression(module_scope, statement.value.callee)
+            if symbol is not None:
+                instances[statement.target.identifier] = symbol
+    return SymbolTable(scopes, imports, instances)
 
 
 class SymbolAnalysis(Analysis[SymbolTable]):
@@ -38,7 +81,7 @@ class SymbolAnalysis(Analysis[SymbolTable]):
 
     @classmethod
     def compute(cls, ctx: AnalysisContext) -> SymbolTable:
-        return analyze_symbols(ctx.get(ScopeAnalysis), ctx.get(ImportAnalysis))
+        return analyze_symbols(ctx.get(ScopeAnalysis), ctx.get(ImportAnalysis), ctx.module)
 
 
 __all__ = ["BUILTIN_NAMES", "SymbolAnalysis", "SymbolId", "SymbolTable", "analyze_symbols"]

src/coretrace_python/taint/__init__.py

@@ -8,6 +8,7 @@
     propagate_taint,
 )
 from coretrace_python.taint.models import (
+    EntryPoint,
     Model,
     ModelError,
     ModelTable,
@@ -20,6 +21,7 @@
 )
 
 __all__ = [
+    "EntryPoint",
     "Model",
     "ModelError",
     "ModelTable",