feat: type what an object gives by attribute or item with a Members model

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## Unreleased
+
+### Detection
+
+- A module-level name bound to an attribute, an item or a call of another module-level name resolves in functions, as the same value does in a function body. `cursor = conn.cursor()` after `conn = sqlite3.connect(...)`, `db = client.shop` and `run = os.system` used to leave the calls made through them unknown, so a query built from request data and executed through a module-level cursor was not reported. Names resolve in statement order. The regression corpus is unchanged (#155).
+
+### Plugins
+
+- A `Members(symbol, dynamic=None, defined=(), typed=())` model says what the attributes and items of a class's instances give, for libraries whose objects give others by names the project chooses. A pymongo client gives the database it is dotted or indexed with, a database gives a collection. The names used to end up in the symbol (`MongoClient.shop.users.find`), or an item vanished into its container's symbol (`MongoClient.find`), so no model could list them. With the model, every path to a collection denotes the collection class, the one a parameter annotated `Collection` denotes: `client.shop.users`, `client["shop"]["users"]`, `client.get_database("shop").get_collection("users")`, a module-level `db = client.shop`, an attribute inherited from the client class. A sink on `Collection.find` covers them all. Symbol resolution now depends on these models, which the engine provides to every module and worker (#157).
+
 ## 0.11.0 (2026-09-25)
 
 ### Detection

docs/plugins.md

@@ -152,6 +152,7 @@ class BottleModels(ModelPlugin):
 | `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 for `kinds`, such as `re.fullmatch`. A flow guarded by it is refuted when it carries no other kind into the sink, and is a hotspot otherwise: a redirect validator does not make a shell command safe. `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. |
+| `Members(symbol, dynamic=None, defined=(), typed=())` | What the attributes and items of an instance of the class `symbol` give, for objects that give others by name. `typed` maps an attribute to the class of what it gives, called or not, since a call denotes its callee: pymongo's `get_database` gives a `Database`, as Flask-PyMongo's `db` property does. `dynamic`, when set, is the class of every item and of every attribute outside `defined` and `typed`: a pymongo client gives a `Database` by name (`client.shop`, `client["shop"]`), a database a `Collection`. Every path to a collection then denotes the collection class, so a sink on `Collection.find` covers them all. Without `dynamic`, an attribute keeps its path and an item its container's symbol, as for any class. |
 
 Two plugins may describe the same symbol. An identical model is registered once; two
 `Sink` models merge their kinds, positions and keywords, so a plugin can add `SQL` to a sink

src/coretrace_python/engine.py

@@ -109,7 +109,7 @@
 from coretrace_python.semantic import SEMANTIC_ANALYSES
 from coretrace_python.semantic.imports import ImportAnalysis, ImportResolutionError, ImportTable
 from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeError
-from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId
+from coretrace_python.semantic.symbols import MembersAnalysis, SymbolAnalysis, SymbolId
 from coretrace_python.source import SourceFile, SourceId, SourceManager, SourceSpan, decode_text
 from coretrace_python.taint import (
     EntryPoint,
@@ -257,6 +257,7 @@ def build_manager(
     manager = _register_all(module)
     table = (models or SecurityModelRegistry()).freeze()
     manager.provide(SecurityModelAnalysis, table)
+    manager.provide(MembersAnalysis, table.members_by_class())
     manager.provide(ClearingAnalysis, table.clearing())
     manager.provide(ProjectSummaries, SummaryIndex())
     manager.provide(RegisteredRoutes, _routes_of(manager))
@@ -311,6 +312,7 @@ def analyze_file(source: SourceFile, plugin_roots: Sequence[Path]) -> FileAnalys
     registry = load_plugins(plugin_roots, manager)
     table = plugin_models(loaded.plugin for loaded in registry)
     manager.provide(SecurityModelAnalysis, table)
+    manager.provide(MembersAnalysis, table.members_by_class())
     manager.provide(ClearingAnalysis, table.clearing())
     manager.provide(ProjectSummaries, SummaryIndex())
     manager.provide(RegisteredRoutes, _routes_of(manager))
@@ -402,8 +404,10 @@ def analyze_project(
     advisories, origins = _merge_advisories(_contributions(registry, plugins, root, file_advisories))
     affected = affected_symbols(dependencies, advisories)
     models = plugin_models(all_plugins, root).extended(*advisory_sinks(affected))
+    members = models.members_by_class()
     for manager in managers.values():
         manager.provide(SecurityModelAnalysis, models)
+        manager.provide(MembersAnalysis, members)
         manager.provide(DependencyAnalysis, dependencies)
 
     imports: dict[str, ImportTable] = {}
@@ -698,8 +702,10 @@ def _analyse_batch(batch: _Batch) -> dict[str, dict[str, Any]]:
     models = plugin_models(all_plugins, batch.root).extended(*advisory_sinks(affected))
     routes = _decode_routes(batch.routes)
     escaped = frozenset(batch.escaped)
+    members = models.members_by_class()
     for manager in managers.values():
         manager.provide(SecurityModelAnalysis, models)
+        manager.provide(MembersAnalysis, members)
         manager.provide(DependencyAnalysis, dependencies)
         manager.provide(RegisteredRoutes, routes)
         manager.provide(EscapedTemplates, escaped)

src/coretrace_python/interprocedural/callgraph.py

@@ -176,12 +176,12 @@ def callers(self, name: str) -> frozenset[str]:
 
 
 def derive_symbols(
-    function: FunctionIR, initial: Mapping[Value, SymbolId] | None = None
+    function: FunctionIR, table: SymbolTable, 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 derive nothing;
-    parameters only through ``initial``, their annotated classes."""
+    """Symbols of values: ``Symbol`` results, attributes and items of symbol values, as
+    ``table`` types them, results of calling a symbol (``sqlite3.connect(p)`` denotes
+    ``python.sqlite3.connect``) 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] = dict(initial or {})
     changed = True
@@ -195,11 +195,9 @@ def derive_symbols(
                 if isinstance(instruction, Symbol):
                     symbol = instruction.symbol_id
                 elif isinstance(instruction, GetAttr) and instruction.object in symbols:
-                    symbol = symbols[instruction.object].attribute(instruction.attribute)
+                    symbol = table.attribute(symbols[instruction.object], instruction.attribute)
                 elif isinstance(instruction, GetItem) and instruction.object in symbols:
-                    # An item of a symbol-denoted container (``request.files['f']``)
-                    # carries the container's symbol, so its methods resolve.
-                    symbol = symbols[instruction.object]
+                    symbol = table.item(symbols[instruction.object])
                 elif isinstance(instruction, Call | WithEnter):
                     origin = (
                         instruction.callee if isinstance(instruction, Call) else instruction.context
@@ -263,6 +261,8 @@ def resolve_targets(
     owner: str | None = None,
     typed: Mapping[Value, str] | None = None,
     bases: Mapping[str, SymbolId] | None = None,
+    *,
+    table: SymbolTable,
 ) -> tuple[dict[Value, Target], dict[Value, SymbolId]]:
     """Map every callee value of ``function`` to its target, and every symbol value.
     ``nested`` names the functions defined inside this one (``outer.inner``);
@@ -271,7 +271,8 @@ def resolve_targets(
     call to ``App.__init__``, ``app.run()`` and ``self.run()`` calls to ``App.run``.
     ``bases`` maps a module class to the symbol of its base: an attribute the class does
     not define is inherited, so ``self.get_argument`` in a ``RequestHandler`` subclass
-    denotes ``tornado.web.RequestHandler.get_argument``."""
+    denotes ``tornado.web.RequestHandler.get_argument``. ``table`` says what the
+    attributes and items of a class a ``Members`` model describes give."""
 
     module = scopes.module_scope
     classes = classes or {}
@@ -294,8 +295,8 @@ def resolve_targets(
                 class_name = instance_of[instruction.object]
                 base = (bases or {}).get(class_name)
                 if base is not None and instruction.attribute not in classes.get(class_name, ()):
-                    inherited[instruction.result] = base.attribute(instruction.attribute)
-    symbols = derive_symbols(function, inherited)
+                    inherited[instruction.result] = table.attribute(base, instruction.attribute)
+    symbols = derive_symbols(function, table, inherited)
     targets: dict[Value, Target] = {
         value: ExternalSymbol(symbol) for value, symbol in symbols.items()
     }
@@ -427,6 +428,7 @@ def compute(cls, ctx: AnalysisContext) -> CallGraph:
                 owner,
                 _typed_with_module_classes(function, ssa, classes),
                 bases,
+                table=table,
             )
             defs = {i.result: i for block in ssa.blocks for i in block.instructions if i.result is not None}
             found: list[CallSite] = []

src/coretrace_python/ir/lowering.py

@@ -141,7 +141,7 @@ def imported_symbol(self, node: nodes.Expression) -> SymbolId | None:
             return self.symbols.resolve(self.scope.id, node.identifier)
         if isinstance(node, nodes.Attribute):
             parent = self.imported_symbol(node.value)
-            return parent.attribute(node.name) if parent is not None else None
+            return self.symbols.attribute(parent, node.name) if parent is not None else None
         return None
 
     def expression(self, node: nodes.Expression) -> Value:

src/coretrace_python/semantic/__init__.py

@@ -2,8 +2,8 @@
 
 from coretrace_python.semantic.imports import ImportAnalysis
 from coretrace_python.semantic.scopes import ScopeAnalysis
-from coretrace_python.semantic.symbols import SymbolAnalysis
+from coretrace_python.semantic.symbols import MembersAnalysis, SymbolAnalysis
 
-SEMANTIC_ANALYSES = (ScopeAnalysis, ImportAnalysis, SymbolAnalysis)
+SEMANTIC_ANALYSES = (ScopeAnalysis, ImportAnalysis, MembersAnalysis, SymbolAnalysis)
 
-__all__ = ["SEMANTIC_ANALYSES", "ImportAnalysis", "ScopeAnalysis", "SymbolAnalysis"]
+__all__ = ["SEMANTIC_ANALYSES", "ImportAnalysis", "MembersAnalysis", "ScopeAnalysis", "SymbolAnalysis"]

src/coretrace_python/semantic/symbols.py

@@ -4,6 +4,7 @@
 
 import builtins
 from collections.abc import Mapping
+from dataclasses import dataclass
 from types import MappingProxyType
 from typing import ClassVar
 
@@ -16,19 +17,69 @@
 BUILTIN_NAMES = frozenset(name for name in dir(builtins) if not name.startswith("_"))
 
 
+@dataclass(frozen=True)
+class Members:
+    """What the attributes and items of an instance of ``symbol`` give, for a class whose
+    instances give other objects by name, as a pymongo client gives the database it is
+    dotted or indexed with (``client.shop``, ``client['shop']``).
+
+    ``typed`` maps an attribute to the class of what it gives, called or not, since a
+    call denotes its callee: ``get_database`` gives a database, as a ``db`` property does.
+    ``dynamic``, when set, is the class of every item and of every attribute outside
+    ``defined`` and ``typed``. Otherwise an attribute keeps its path and an item its
+    container's symbol, as for any other class."""
+
+    symbol: SymbolId
+    dynamic: SymbolId | None = None
+    defined: tuple[str, ...] = ()
+    typed: tuple[tuple[str, SymbolId], ...] = ()
+
+    def __post_init__(self) -> None:
+        # Sorted: the models' repr is part of the cache key, the same in every process.
+        object.__setattr__(self, "defined", tuple(sorted(set(self.defined))))
+
+    def attribute(self, name: str) -> SymbolId:
+        for attribute, gives in self.typed:
+            if attribute == name:
+                return gives
+        if self.dynamic is not None and name not in self.defined:
+            return self.dynamic
+        return self.symbol.attribute(name)
+
+    def item(self) -> SymbolId:
+        return self.symbol if self.dynamic is None else self.dynamic
+
+
 class SymbolTable:
     """Resolve names to canonical symbols through scopes, imports, builtins and the
-    module-level instances created by calling a symbol (``app = Flask(__name__)``)."""
+    module-level names bound to a value denoting a symbol (``app = Flask(__name__)``,
+    ``cursor = conn.cursor()``). ``members`` holds the ``Members`` models by class."""
 
     def __init__(
         self,
         scopes: ScopeTable,
         imports: ImportTable,
         instances: Mapping[str, SymbolId] | None = None,
+        members: Mapping[SymbolId, Members] | None = None,
     ) -> None:
         self._scopes = scopes
         self._imports = imports
         self._instances = MappingProxyType(dict(instances or {}))
+        self._members = MappingProxyType(dict(members or {}))
+
+    def attribute(self, symbol: SymbolId, name: str) -> SymbolId:
+        """The symbol of attribute ``name`` of what ``symbol`` denotes: its path, unless
+        a ``Members`` model describes the class."""
+
+        members = self._members.get(symbol)
+        return symbol.attribute(name) if members is None else members.attribute(name)
+
+    def item(self, symbol: SymbolId) -> SymbolId:
+        """The symbol of an item of what ``symbol`` denotes: the container's own, so its
+        methods resolve (``request.files['f'].save``), unless a ``Members`` model says."""
+
+        members = self._members.get(symbol)
+        return symbol if members is None else members.item()
 
     def resolve(self, scope_id: ScopeId, name: str) -> SymbolId | None:
         resolution = self._scopes.resolve(scope_id, name)
@@ -49,39 +100,78 @@ def resolve_expression(self, scope_id: ScopeId, node: nodes.Expression) -> Symbo
             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
+            return self.attribute(parent, 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
+    scopes: ScopeTable,
+    imports: ImportTable,
+    module: nodes.Module | None = None,
+    members: Mapping[SymbolId, Members] | None = None,
 ) -> SymbolTable:
-    table = SymbolTable(scopes, imports)
+    table = SymbolTable(scopes, imports, members=members)
     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 isinstance(statement, nodes.Assign) and isinstance(statement.target, nodes.Name):
+            symbol = _denoted(table, module_scope, statement.value)
             if symbol is not None:
                 instances[statement.target.identifier] = symbol
-    return SymbolTable(scopes, imports, instances)
+                # Later bindings resolve through this one, in statement order.
+                table = SymbolTable(scopes, imports, instances, members)
+    return table
+
+
+def _denoted(table: SymbolTable, scope_id: ScopeId, node: nodes.Expression) -> SymbolId | None:
+    """What a value denotes, as a function body derives it (``derive_symbols``): a
+    resolved name or an attribute of one, the result of calling one, or an item of one,
+    which carries its container's symbol (``os.environ['HOME']``) unless a ``Members``
+    model says what its items are."""
+
+    if isinstance(node, nodes.Subscript):
+        container = _denoted(table, scope_id, node.value)
+        return table.item(container) if container is not None else None
+    if isinstance(node, nodes.Attribute):
+        parent = _denoted(table, scope_id, node.value)
+        return table.attribute(parent, node.name) if parent is not None else None
+    if isinstance(node, nodes.Call):
+        return _denoted(table, scope_id, node.callee)
+    return table.resolve_expression(scope_id, node)
+
+
+class MembersAnalysis(Analysis[Mapping[SymbolId, Members]]):
+    """The ``Members`` models by class, provided by the engine from the security models;
+    none on its own, so every attribute keeps its path."""
+
+    name: ClassVar[str] = "semantic.members"
+
+    @classmethod
+    def compute(cls, ctx: AnalysisContext) -> Mapping[SymbolId, Members]:
+        return MappingProxyType({})
 
 
 class SymbolAnalysis(Analysis[SymbolTable]):
     name: ClassVar[str] = "semantic.symbols"
-    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({ScopeAnalysis, ImportAnalysis})
+    requires: ClassVar[frozenset[AnyAnalysis]] = frozenset({ScopeAnalysis, ImportAnalysis, MembersAnalysis})
 
     @classmethod
     def compute(cls, ctx: AnalysisContext) -> SymbolTable:
-        return analyze_symbols(ctx.get(ScopeAnalysis), ctx.get(ImportAnalysis), ctx.module)
-
-
-__all__ = ["BUILTIN_NAMES", "SymbolAnalysis", "SymbolId", "SymbolTable", "analyze_symbols"]
+        return analyze_symbols(
+            ctx.get(ScopeAnalysis), ctx.get(ImportAnalysis), ctx.module, ctx.get(MembersAnalysis)
+        )
+
+
+__all__ = [
+    "BUILTIN_NAMES",
+    "Members",
+    "MembersAnalysis",
+    "SymbolAnalysis",
+    "SymbolId",
+    "SymbolTable",
+    "analyze_symbols",
+]

src/coretrace_python/taint/__init__.py

@@ -1,5 +1,6 @@
 """Security models and the global multi-kind taint engine (architecture §16, §17)."""
 
+from coretrace_python.semantic.symbols import Members
 from coretrace_python.taint.engine import (
     EntryPointAnalysis,
     Taint,
@@ -36,6 +37,7 @@
     "EntryPoint",
     "EntryPointAnalysis",
     "EscapedTemplates",
+    "Members",
     "Model",
     "ModelError",
     "ModelTable",