diff --git a/CHANGELOG.md b/CHANGELOG.md index 92f04f2..0890758 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # Changelog +## Unreleased + +### Analysis + +- Two files with the same Python module name, `a/app.py` and `z/app.py` outside any package, no longer overwrite each other: both are discovered, analysed and covered whatever the order they are found in. The engine tells them apart by their path from the root as a module name, `a.app` and `z.app`, in the module graph, the summary index, the cache keys and what project plugins see; a unique name stays what it was. An import of such a name resolves to the file in the importer's own root only, and one no root resolves is not followed. Each colliding file carries an `ambiguous-module` note and is covered as `ambiguous`, since a model or a route naming `python.app.main` applies to every definition of that symbol, and so does an importer whose import stayed unresolved; a VEX statement over such a project stays `under_investigation` (#182). + +### Precision + +- A `TEMPLATES` assigned inside a function or a class no longer counts as Django settings: only a module-level assignment can activate the `request` context processor, and only a mention of the module's `TEMPLATES` elsewhere leaves that activation uncertain. A function-local `TEMPLATES`, called or not, used to enable the implicit request source, upgrading a template filter finding to exploitable (#184). +- A template block that renders `{{ block.super }}` keeps the parent block's content, so the filters the parent applies stay fed by the render context, with the parent template as the place of the call; a block without it still replaces the parent's whole block, and a chain of templates keeps exactly the parents each level renders. The parent's filters used to drop out of the exploitable flow as soon as a child defined the block (#185). +- A project tag library registering a filter of a built-in's name replaces the built-in in every Django template loading it before applying the filter, as Django's parser does: `{% load custom %}{{ bio|striptags }}` calls the project's `striptags` when `app/templatetags/custom.py` registers one, so it is no longer a reachable or exploitable call to `django.template.defaultfilters.striptags`; the project's function is ordinary code, analysed as any other. A load affects only the filters after it, a later load overrides an earlier one, `{% load striptags from custom %}` replaces only what it names, and Django's own libraries change nothing. The analyzer reads the registrations on `register`, a `django.template.Library` however it is imported, in decorator and call form, at module level or under `if`, `try` or a function. Where it cannot tell what a library registers — no such library file, one it cannot read or parse, a name not written as a constant, `register` passed elsewhere or filled through `filters=` or `register.filters`, or two apps giving the name and registering different filters — a filter of that name applied under the load is neither the built-in nor certainly replaced: no call is claimed, and the load counts among the templates the analyzer could not read (`app/templates/app/profile.html:1 loads 'custom', not read`), so an advisory naming the filter stays `under_investigation` in the VEX document. The settings' `OPTIONS["libraries"]` and `OPTIONS["builtins"]` are not read (#183). + +### Plugins + +- An `EntryPoint` naming a project function or class applies through a module-level alias of it: `main = actual` makes `python.app.handlers.main` name `actual`, so a deployment declaring `handlers.main` reaches the function doing the work. An alias is an assignment of one name to another at module level, followed transitively and in source order, in a conditional or a `try` as well; a name bound inside a function or class, or to a call, an import or anything but a name, is no alias, and the engine still infers nothing from a name. A model naming a method wins over one naming its class, whichever name it uses. + ## 0.18.0 (2026-09-28) ### Precision diff --git a/docs/plugins.md b/docs/plugins.md index 0da9842..8cc7791 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -53,7 +53,11 @@ resolved through imports and aliases, so `python.os.system` covers `os.system`, symbol, so a chain such as `sqlite3.connect(p).cursor().execute` is `python.sqlite3.connect.cursor.execute`, and `app = Flask(__name__)` gives `app.route` the symbol `python.flask.Flask.route`. Builtins are `python.builtins.`. Symbols are -`SymbolId` values from `coretrace_python.semantic.symbols`. +`SymbolId` values from `coretrace_python.semantic.symbols`. A function of the project is +`python.` followed by its module's import name and its qualified name, +`python.app.handlers.main`; two files with the same import name, `a/app.py` and +`z/app.py` outside any package, define the same symbols, so a model naming +`python.app.main` names the function of each. ## Kinds of plugin @@ -260,9 +264,11 @@ when the models make it one. `ctx.call_graph(module)` gives each function's call (`sites(function)`) and the symbols it reads, called or not (`reads(function)`: reading `request.form['name']` reads `python.flask.request.form`), each where it first reads it. `ctx.templates` gives the templates found under the project's `templates` directories: -their `names`, the filters they apply as `calls` (each a `FilterCall(symbol, span)`, the -function behind the filter and where the template applies it), and where they name a -template the analyzer cannot read (`unread`). +their `names`, the filters of Django's own libraries they apply as `calls` (each a +`FilterCall(symbol, span)`, the function behind the filter and where the template +applies it; a filter a project tag library the template loads replaces is no call), +and where they name a template the analyzer cannot read, or load a library it cannot +read while applying a filter of that library's name (`unread`). The shipped `vulnerable-dependency`, `reachable-vulnerability` and `dependency-policy` plugins are of this kind. diff --git a/docs/usage.md b/docs/usage.md index 35c8e90..1c658b9 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -65,7 +65,13 @@ coretrace-python-analyzer [--check | --emit-ir [--ssa]] [options] [path] ## What is analysed Given a directory, every `.py` file below it is a module of one project, named after -its package (`app/views.py` is `app.views`). Hidden directories, `node_modules`, +its package (`app/views.py` is `app.views`). Two files that would share a name, `a/app.py` +and `z/app.py` outside any package, are told apart by their path from the root, `a.app` +and `z.app`, and reported: each carries an `ambiguous-module` note, since a symbol +`python.app.main` may name a function of either and an import of `app` resolves only +from the importer's own directory, and each is covered as `ambiguous` rather than +analysed. An import of such a name that no file of the importer's own directory answers +is noted where it is written and not followed. Hidden directories, `node_modules`, `__pycache__`, `build`, `dist` and any virtual environment, recognised by its `pyvenv.cfg` whatever its name, are skipped. Dependency files at the root (`requirements*.txt`, `pyproject.toml`, `poetry.lock`, `uv.lock`) are read into a @@ -261,7 +267,9 @@ the directory of the checked file; a path outside it is printed as it is. } ``` -`root` is the directory the paths are relative to. +`root` is the directory the paths are relative to. A file's `status` is `analysed`, +`ambiguous` (analysed, but another file has its module name, see above), `syntax-error` +or `unreadable`; only `analysed` files count in `files_analysed`. A directory check also lists, in `tool.components`, what the result was produced with besides the engine and the sources: each plugin it loaded, by manifest name and version, @@ -372,6 +380,28 @@ Django. Most filters are functions of the same name in `django.template.defaultf are `_filter` there, and the filters of `i18n`, `l10n`, `tz` and `humanize` are in their library's module (`timezone` is `django.templatetags.tz.do_timezone`). +A project tag library, a module of a `templatetags` directory, that registers a filter +of that name replaces the built-in in every template loading it, as Django's parser +does: `{% load custom %}{{ bio|striptags }}` calls the project's function when +`custom.py` registers a `striptags` filter, which is no call to the built-in; the +project's function is ordinary code, whose own calls are analysed as any other. A +load affects only the filters applied after it in the file, a later load overrides an +earlier one, `{% load striptags from custom %}` replaces only the filters it names, and +Django's own libraries (`static`, `i18n`, `l10n`, `tz`, `cache`, `humanize`) change +nothing. The analyzer reads the registrations on the library's `register`, a +`django.template.Library` however it is imported: `@register.filter`, +`@register.filter("name")`, `@register.filter(name="name", is_safe=True)`, +`register.filter("name", function)` and `register.filter(function)`, at module level or +under `if`, `try` or a function. Where it cannot tell what a library registers, the +filter is neither the built-in nor certainly replaced: a library found in no +`templatetags` directory (one an installed package ships, or that the settings' +`OPTIONS["libraries"]` or `OPTIONS["builtins"]` name, which are not read), a file it +cannot read or parse, a name given by a variable or unpacked arguments, `register` +passed to another function, filled through `filters=` or `register.filters`, or bound +to anything but a `Library`, or two apps giving the same library name and registering +different filters. The analyzer then claims no call, and counts the template among those +it could not read (below) when a filter of the library's name is applied under the load. + Attacker input in the render context makes the filter exploitable at the render call: `render(request, "app/profile.html", {"bio": request.POST["bio"]})` with `{{ bio|striptags }}`. The analyzer links a context value to a filter only where it is @@ -386,7 +416,8 @@ certain of it: arguments of the filters before it, or the output of a `{% filter %}` block. A name a tag binds shadows the context, and `{% include ... only %}` passes none of it. - The filter renders: a template extending another renders only its blocks, which - replace the blocks of that name in what it extends. + replace the blocks of that name in what it extends, unless the block renders + `{{ block.super }}`, which keeps the parent's content and the filters it applies. - One file has the template's name; of two files of the same name, which one renders depends on the loaders. @@ -399,9 +430,12 @@ With the `django.template.context_processors.request` context processor, a templ rendered with the request reads it as `request`: `{{ request.GET.q|striptags }}` is exploitable at the render call too. The analyzer counts it only where all of these hold: -- The project's settings assign `TEMPLATES` a literal list, every `DjangoTemplates` - engine of it lists the processor in a literal `context_processors`, and no other code - of the project names `TEMPLATES`. A mere mention of the processor proves nothing. +- The project's settings assign `TEMPLATES` a literal list at module level, every + `DjangoTemplates` engine of it lists the processor in a literal `context_processors`, + and no other code of the project names the module's `TEMPLATES`, or a `TEMPLATES` it + does not bind, as a star import gives. A `TEMPLATES` bound in a function or a class + is another name: it configures nothing and changes nothing. + A mere mention of the processor proves nothing. - The render call passes the request: `render` and `TemplateResponse` always do, `render_to_string` when given `request`, `SimpleTemplateResponse` never. - The render's context is certain, absent or a dict literal as above, so it cannot hide @@ -502,10 +536,14 @@ A template the analyzer cannot read may apply any filter. The project names one `render`, `render_to_string`, `TemplateResponse`, an `{% include %}` or an `{% extends %}` names a template by an expression, or by a name found under no `templates` directory, including one an installed package ships; a template file that cannot be read counts -too. An advisory whose entry points include a template filter then stays -`under_investigation`, and the notes list where the project names each such template -(`app.views:12 names a template by an expression`). A template named elsewhere, such as -a class-based view's `template_name`, is not checked. +too, and so does a template loading a tag library the analyzer cannot read, or find, +or finds in several apps registering different filters, where it then applies a filter +of a built-in's name, since the library may replace it. An advisory whose entry points +include a template filter then stays `under_investigation`, and the notes list where +the project names each such template (`app.views:12 names a template by an +expression`, `app/templates/app/profile.html:1 loads 'crispy_forms_tags', not found`). +A template named elsewhere, such as a class-based view's `template_name`, is not +checked. `not_affected` needs the lock-file check because the analyzer reads the project's code, not the code of installed packages. A framework calling the vulnerable function on the diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index b877096..2a2bbd4 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -9,6 +9,7 @@ import concurrent.futures import hashlib import multiprocessing +from collections import Counter from collections.abc import Callable, Iterable, Mapping, Sequence from dataclasses import dataclass, field, replace from pathlib import Path @@ -88,6 +89,7 @@ TemplateCallsAnalysis, build_module_graph, discover_sources, + name_modules, project_symbol, ) from coretrace_python.ir import PYIR_SCHEMA_VERSION @@ -371,7 +373,11 @@ def analyze_project( strongly connected components of the module graph, imports first; with ``jobs`` above one the components of a wave are analysed in that many processes (§29). ``advisory_files`` add to the ``advisories.json`` at ``root``; ``policy_file`` - replaces the ``coretrace-policy.toml`` there.""" + replaces the ``coretrace-policy.toml`` there. + + Every module is known by the name ``name_modules`` gives its file: its import name, + or its path from the root when another file has that import name. Such files carry + an ``ambiguous-module`` note and are covered as ``ambiguous``.""" if jobs < 1: raise ValueError("jobs must be at least 1") @@ -398,18 +404,18 @@ def analyze_project( files: dict[str, SourceFile] = {} unreadable: list[tuple[Path, str]] = [] coverage: list[FileCoverage] = [] - discovered = discover_sources(root, sources, unreadable) + named = name_modules(root, discover_sources(root, sources, unreadable)) for path, reason in unreadable: findings.append(_note("syntax-error", f"{path}: {reason}", sources.add_source(str(path), ""), 1)) coverage.append(FileCoverage(str(path), "unreadable", 0, 0)) - for source in discovered: + for name, source in named.items(): try: - modules[source.module_name] = build_hir(source) + modules[name] = build_hir(source) except (ParseError, HIRBuildError) as error: findings.append(_note("syntax-error", str(error), source, 1)) coverage.append(FileCoverage(str(source.source_id), "syntax-error", 0, 0)) continue - files[source.module_name] = source + files[name] = source managers = {name: _register_all(module) for name, module in modules.items()} probe = next(iter(managers.values()), None) or _register_all(build_hir(sources.add_source("", ""))) @@ -439,6 +445,8 @@ def analyze_project( graph = build_module_graph( {name: files[name] for name in analysable}, {name: modules[name] for name in analysable}, imports ) + shared = _shared_names(graph) + findings.extend(_ambiguity_notes(graph, shared, root)) routes: dict[SymbolId, EntryPoint] = {} for name in sorted(analysable): for symbol, registered in _routes_of(analysable[name]).items(): @@ -447,7 +455,8 @@ def analyze_project( templates = project_templates(root) clearing = models.clearing(escaped) template_calls = models.template_calls( - templates.filters, request_processor(modules[name] for name in sorted(analysable)) + templates.filters, + request_processor((modules[name], analysable[name].get(ScopeAnalysis)) for name in sorted(analysable)), ) for manager in analysable.values(): manager.provide(RegisteredRoutes, routes) @@ -514,15 +523,16 @@ def analyze_project( if pool is not None: pool.shutdown() - index = _seed(results, graph, frozenset()) + index = _index(results, {name: name for name in results}) call_graphs: dict[str, CallGraph] = {} functions: dict[str, tuple[ModuleFunction, ...]] = {} for name in sorted(analysable): entry = results[name] findings.extend(entry.findings) unsupported = sum(1 for f in entry.findings if f.rule_id == "unsupported-syntax") + status = "ambiguous" if files[name].module_name in shared else "analysed" coverage.append( - FileCoverage(str(files[name].source_id), "analysed", len(entry.functions), len(entry.functions) - unsupported) + FileCoverage(str(files[name].source_id), status, len(entry.functions), len(entry.functions) - unsupported) ) functions[name] = entry.functions sites: dict[str, list[CallSite]] = {function.name: [] for function in entry.functions} @@ -644,24 +654,76 @@ def _advisory_paths(root: Path, advisory_files: Sequence[Path]) -> tuple[Path, . return (*([default] if default.is_file() else []), *advisory_files) +def _shared_names(graph: ModuleGraph) -> dict[str, list[str]]: + """The import names several analysable files have, each with those files' modules.""" + + by_name: dict[str, list[str]] = {} + for name in graph.modules: + by_name.setdefault(graph.source(name).module_name, []).append(name) + return {import_name: names for import_name, names in by_name.items() if len(names) > 1} + + +def _ambiguity_notes(graph: ModuleGraph, shared: Mapping[str, list[str]], root: Path) -> list[Finding]: + """One note on each file whose import name another file shares, since a symbol of + that name may denote a function of either, and one on each importer whose import of + such a name no root resolved, since calls through it are not followed.""" + + def located(name: str) -> str: + return _located(_path_of(graph.source(name)), root) + + notes: list[Finding] = [] + for import_name, names in sorted(shared.items()): + for name in names: + others = ", ".join(located(other) for other in names if other != name) + message = ( + f"{located(name)} is the module '{import_name}' like {others}: a symbol " + f"python.{import_name}.* may name a function of either, and an import of " + f"'{import_name}' resolves only from its own root" + ) + notes.append(_note("ambiguous-module", message, graph.source(name), 1)) + for name in graph.modules: + for import_name in sorted(graph.unresolved(name)): + candidates = " or ".join(located(other) for other in shared[import_name]) + message = f"'{import_name}' imported here is {candidates}: calls into it are not followed" + notes.append(_note("ambiguous-module", message, graph.source(name), 1)) + return notes + + def _seed(results: Mapping[str, CachedModule], graph: ModuleGraph, component: frozenset[str]) -> SummaryIndex: """The summaries a component starts from: those of every module it imports, - transitively, all final by the time its wave runs; the whole index for no component.""" - - if component: - wanted: set[str] = set() - pending = list(component) - while pending: - for imported in graph.imports(pending.pop()): - if imported in results and imported not in wanted: - wanted.add(imported) - pending.append(imported) - else: - wanted = set(results) + transitively, all final by the time its wave runs, each under the import name the + component's code refers to it by. Of several files with one import name, only the + one a member imports directly, resolved in its own root, goes under that name: the + others are reached through summaries that inline them already, and a name no + member's root resolved is followed into neither file.""" + + wanted: set[str] = set() + pending = list(component) + while pending: + for imported in graph.imports(pending.pop()): + if imported in results and imported not in wanted: + wanted.add(imported) + pending.append(imported) + direct = {imported for member in component for imported in graph.imports(member)} + by_import: dict[str, list[str]] = {} + for name in wanted: + by_import.setdefault(graph.source(name).module_name, []).append(name) + names: dict[str, str] = {} + for import_name, modules in by_import.items(): + chosen = modules if len(modules) == 1 else [name for name in modules if name in direct] + if len(chosen) == 1: + names[chosen[0]] = import_name + return _index(results, names) + + +def _index(results: Mapping[str, CachedModule], names: Mapping[str, str]) -> SummaryIndex: + """The summaries of the modules in ``names``, each function under the project symbol + of the module name given for its module.""" + return SummaryIndex( { - project_symbol(name, function): summary - for name in sorted(wanted) + project_symbol(module, function): summary + for name, module in sorted(names.items()) for function, summary in results[name].summaries.items() } ) @@ -674,8 +736,12 @@ def _analyse_managers( affected: Affected, ) -> dict[str, CachedModule]: """Analyse one component: iterate its summaries to a fixpoint over ``seed`` (§21), - then extract what the rest of the run needs from each module.""" + then extract what the rest of the run needs from each module. Each member registers + its functions under its import name, what the others' code refers to it by; a name + two members share, a cycle across roots, is registered by neither.""" + counts = Counter(manager.module.name for manager in managers.values()) + names = {name: manager.module.name for name, manager in managers.items() if counts[manager.module.name] == 1} index = seed for manager in managers.values(): manager.provide(ProjectSummaries, index) @@ -684,9 +750,9 @@ def _analyse_managers( { **seed.summaries, **{ - project_symbol(name, function): summary - for name, manager in managers.items() - for function, summary in _summaries_of(manager).items() + project_symbol(module, function): summary + for name, module in names.items() + for function, summary in _summaries_of(managers[name]).items() }, } ) @@ -701,8 +767,9 @@ def _analyse_managers( @dataclass(frozen=True) class _Batch: - """One component handed to a worker process: it rebuilds the configuration from the - project root and the plugin roots, so only paths and the imported summaries travel.""" + """One component handed to a worker process, its files under the engine's module + names: it rebuilds the configuration from the project root and the plugin roots, so + only paths and the imported summaries travel.""" root: Path plugin_roots: tuple[Path, ...] diff --git a/src/coretrace_python/findings/coverage.py b/src/coretrace_python/findings/coverage.py index 0d98027..54728a1 100644 --- a/src/coretrace_python/findings/coverage.py +++ b/src/coretrace_python/findings/coverage.py @@ -1,10 +1,12 @@ """Analysis coverage: which files and functions a run actually looked at. "No findings" only means something when the reader knows what was analysed. Each file -is ``analysed``, a ``syntax-error`` (the frontend rejected it) or ``unreadable`` (it could -not be decoded); analysed files count their functions and how many lowered. The -templates the project names but the run could not read are listed, each where the -project names it: they may call any template filter. +is ``analysed``, ``ambiguous`` (analysed, but another file has its Python module name, +so a symbol of that name may denote a function of either and an import of it resolves +only from its own root), a ``syntax-error`` (the frontend rejected it) or ``unreadable`` +(it could not be decoded); analysed and ambiguous files count their functions and how +many lowered. The templates the project names but the run could not read are listed, +each where the project names it: they may call any template filter. """ from __future__ import annotations diff --git a/src/coretrace_python/interprocedural/__init__.py b/src/coretrace_python/interprocedural/__init__.py index a770a8c..d1817ad 100644 --- a/src/coretrace_python/interprocedural/__init__.py +++ b/src/coretrace_python/interprocedural/__init__.py @@ -19,6 +19,7 @@ build_module_graph, discover_files, discover_sources, + name_modules, project_symbol, ) from coretrace_python.interprocedural.summaries import ( @@ -73,5 +74,6 @@ "cleared_by", "discover_files", "discover_sources", + "name_modules", "project_symbol", ] diff --git a/src/coretrace_python/interprocedural/modulegraph.py b/src/coretrace_python/interprocedural/modulegraph.py index 00dedfd..b540563 100644 --- a/src/coretrace_python/interprocedural/modulegraph.py +++ b/src/coretrace_python/interprocedural/modulegraph.py @@ -6,12 +6,18 @@ (``python.app.helpers.run``) whose summaries live in a ``SummaryIndex`` that the engine provides to every module's manager, so calls into other files are analysed through summaries rather than by retaining every module's PyIR. + +The identity of a module is its file. Two files may have the same import name +(``a/app.py`` and ``z/app.py`` outside any package are both ``app``, each from its own +directory): the engine then names them by their path from the root, ``a.app`` and +``z.app``, and an import of ``app`` resolves within the importer's own root only. """ from __future__ import annotations +from collections import Counter from collections.abc import Iterable, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from types import MappingProxyType @@ -67,6 +73,66 @@ def _is_environment(directory: Path, known: dict[Path, bool]) -> bool: return known[directory] +def name_modules(root: Path, sources: Iterable[SourceFile]) -> dict[str, SourceFile]: + """One name per file, in the order given: its import name when no other file has it; + otherwise its path from ``root`` as a dotted name (``a/app.py`` and ``z/app.py`` are + ``a.app`` and ``z.app``, a root-level ``manage.py`` stays ``manage``), or that path + itself when even the dotted names coincide (``x.y/app.py`` and ``x/y/app.py``). + Decided over the whole set, so the names do not depend on the order the files are + found in.""" + + found = tuple(sources) + counts = Counter(source.module_name for source in found) + shared = [source for source in found if counts[source.module_name] > 1] + paths = {source.source_id: _relative(root, source) for source in shared} + dotted = {source_id: _dotted(path) for source_id, path in paths.items()} + taken = ( + {name for name, count in counts.items() if count == 1} + | {name for name, count in Counter(dotted.values()).items() if count > 1} + | {path.as_posix() for path in paths.values()} + ) + named: dict[str, SourceFile] = {} + for source in found: + if counts[source.module_name] == 1: + named[source.module_name] = source + elif dotted[source.source_id] not in taken: + named[dotted[source.source_id]] = source + else: + named[paths[source.source_id].as_posix()] = source + return named + + +def _relative(root: Path, source: SourceFile) -> Path: + path = source.path if source.path is not None else Path(str(source.source_id)) + try: + return path.relative_to(root.resolve()) + except ValueError: + return path + + +def _dotted(path: Path) -> str: + """``a/app.py`` as ``a.app`` and ``pkg/__init__.py`` as ``pkg``; a root-level + ``__init__.py`` has no dotted name and keeps its path.""" + + parts = path.with_suffix("").parts + if parts and parts[-1] == "__init__": + parts = parts[:-1] + return ".".join(parts) or path.as_posix() + + +def _import_root(source: SourceFile) -> Path | None: + """The directory an import of ``source`` by its module name starts from: the parent + of its top-level package, or its own directory for a module outside any package, + where ``module_name_for`` stopped. ``None`` for a source not on disk.""" + + if source.path is None: + return None + directory = source.path.parent + while (directory / "__init__.py").is_file(): + directory = directory.parent + return directory + + def project_symbol(module_name: str, qualified_name: str) -> SymbolId: """The canonical symbol of a project function. A module whose name is not an identifier cannot be imported by that name, but its functions still need stable @@ -90,6 +156,7 @@ def _component(part: str) -> str: class ModuleGraph: _sources: Mapping[str, SourceFile] _imports: Mapping[str, frozenset[str]] + _unresolved: Mapping[str, frozenset[str]] = field(default_factory=dict) @property def modules(self) -> tuple[str, ...]: @@ -104,6 +171,12 @@ def imports(self, name: str) -> frozenset[str]: def importers(self, name: str) -> frozenset[str]: return frozenset(m for m, imported in self._imports.items() if name in imported) + def unresolved(self, name: str) -> frozenset[str]: + """The import names ``name`` imports that several project files have and not + exactly one in its own root: no edge records them.""" + + return self._unresolved.get(name, frozenset()) + def schedule(self) -> tuple[tuple[frozenset[str], ...], ...]: """Strongly connected components in waves: every component of a wave imports, outside itself, only components of earlier waves, so one wave can be analysed @@ -167,10 +240,17 @@ def build_module_graph( modules: Mapping[str, nodes.Module], imports: Mapping[str, ImportTable], ) -> ModuleGraph: - """Edges from each module to the project modules it imports, in any scope.""" - - names = frozenset(sources) + """Edges from each module to the project modules it imports, in any scope. An import + names a module by its import name; when several files have that name, it resolves to + the one in the importer's own root and, without exactly one there, stays unresolved: + no edge, and the name is recorded on the graph.""" + + by_name: dict[str, list[str]] = {} + for name, source in sources.items(): + by_name.setdefault(source.module_name, []).append(name) + roots = {name: _import_root(source) for name, source in sources.items()} edges: dict[str, frozenset[str]] = {} + unresolved: dict[str, frozenset[str]] = {} for name, module in modules.items(): candidates: set[str] = set() dotted_tops: set[str] = set() @@ -187,11 +267,24 @@ def build_module_graph( candidates.add(path) # The most specific project module each import names; ``app.helpers`` implies # the ``app`` package, which is not an edge worth recording. - found = { - next((m for m in _prefixes(candidate) if m in names), None) for candidate in candidates - } - edges[name] = frozenset(m for m in found if m is not None and m != name) - return ModuleGraph(MappingProxyType(dict(sources)), MappingProxyType(edges)) + found: set[str] = set() + missing: set[str] = set() + for candidate in candidates: + for prefix in _prefixes(candidate): + matching = by_name.get(prefix) + if matching is None: + continue + if len(matching) > 1: + matching = [m for m in matching if roots[m] == roots[name]] + if len(matching) == 1: + found.add(matching[0]) + else: + missing.add(prefix) + break + edges[name] = frozenset(found - {name}) + if missing: + unresolved[name] = frozenset(missing) + return ModuleGraph(MappingProxyType(dict(sources)), MappingProxyType(edges), MappingProxyType(unresolved)) def _prefixes(dotted: str) -> list[str]: @@ -217,5 +310,6 @@ def _statements(body: Iterable[nodes.Statement]) -> Iterable[nodes.Statement]: "SummaryIndex", "build_module_graph", "discover_sources", + "name_modules", "project_symbol", ] diff --git a/src/coretrace_python/semantic/scopes.py b/src/coretrace_python/semantic/scopes.py index 9f4a3b2..4120871 100644 --- a/src/coretrace_python/semantic/scopes.py +++ b/src/coretrace_python/semantic/scopes.py @@ -113,8 +113,8 @@ def scope(self, scope_id: ScopeId) -> Scope: def children(self, scope_id: ScopeId) -> tuple[Scope, ...]: return self._children[scope_id] - def scope_for(self, node: nodes.Function | nodes.Class | nodes.Comprehension) -> Scope: - """Return the scope introduced by a definition or comprehension node.""" + def scope_for(self, node: nodes.Function | nodes.Class | nodes.Lambda | nodes.Comprehension) -> Scope: + """Return the scope introduced by a definition, lambda or comprehension node.""" return self._scopes[self._spans[node.span]] diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 0430201..f33ea32 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -836,7 +836,9 @@ def function_entry_point( ) -> EntryPoint | None: """The entry point ``function`` is: one of its decorators or, for a method, one of its class's bases matching an entry-point model, a registration elsewhere (``routes``), or - a model naming the function, or its class, by its own project symbol.""" + a model naming the function, or its class, by its own project symbol or by a + module-level alias of it (``main = actual`` names ``actual`` as ``main``). A model + naming the method wins over one naming its class, whichever name it uses.""" owner = _owner(module, function) enclosing_function = _enclosing(module, function) @@ -844,12 +846,16 @@ def function_entry_point( # ``app = Flask(__name__)`` inside ``create_app``: routes defined there resolve. instances = {**(instances or {}), **local_instances(enclosing_function, scopes, symbols)} entry = entry_point_of(function, models, scopes, symbols, owner, instances, project) - if entry is None: - qualified = function.name if owner is None else f"{owner.name}.{function.name}" - entry = _declared(project_symbol(module.name, qualified), models, routes) - if entry is None and owner is not None: - entry = _declared(project_symbol(module.name, owner.name), models, routes) - return entry + if entry is not None: + return entry + defined = function if owner is None else owner + names = (defined.name, *_aliases(module, defined.name)) + qualified = names if owner is None else (*(f"{name}.{function.name}" for name in names), *names) + for name in qualified: + entry = _declared(project_symbol(module.name, name), models, routes) + if entry is not None: + return entry + return None def _declared(symbol: SymbolId, models: ModelTable, routes: Routes | None) -> EntryPoint | None: @@ -860,6 +866,43 @@ def _declared(symbol: SymbolId, models: ModelTable, routes: Routes | None) -> En return registered if registered is not None else models.entry_point(symbol) +def _aliases(module: nodes.Module, name: str) -> Iterable[str]: + """The module-level names bound to ``name`` by assigning one name to another, + transitively and in source order: ``handler = actual`` then ``main = handler`` make + both aliases of ``actual``. A name bound inside a function or class, or to anything + but a name (a call, an import), is no alias.""" + + aliases: dict[str, list[str]] = {} + for statement in _module_level(module.body): + if isinstance(statement, nodes.Assign) and isinstance(statement.target, nodes.Name) and isinstance(statement.value, nodes.Name): + aliases.setdefault(statement.value.identifier, []).append(statement.target.identifier) + found = {name} + pending = [name] + while pending: + for alias in aliases.get(pending.pop(0), ()): + if alias not in found: + found.add(alias) + pending.append(alias) + yield alias + + +def _module_level(body: Iterable[nodes.Statement]) -> Iterable[nodes.Statement]: + """The statements executed at module level: the body and what its compound + statements hold, not what a function or class defines.""" + + for statement in body: + if isinstance(statement, nodes.Function | nodes.Class): + continue + yield statement + for attribute in ("body", "orelse", "finalbody"): + nested = getattr(statement, attribute, None) + if isinstance(nested, tuple): + yield from _module_level(nested) + if isinstance(statement, nodes.Try): + for handler in statement.handlers: + yield from _module_level(handler.body) + + def _owner(module: nodes.Module, function: nodes.Function) -> nodes.Class | None: """The top-level class ``function`` is a method of, if any.""" diff --git a/src/coretrace_python/taint/templates.py b/src/coretrace_python/taint/templates.py index a41ed57..1072144 100644 --- a/src/coretrace_python/taint/templates.py +++ b/src/coretrace_python/taint/templates.py @@ -24,18 +24,30 @@ or not a Python call names it, since a class-based view renders its ``template_name`` inside Django. A template the engine cannot read may call any filter; the engine says where the project names one. + +A project tag library, a module of a ``templatetags`` directory, replaces a filter of +that name in every template loading it before applying the filter, as Django's parser +does: ``{% load custom %}{{ bio|striptags }}`` calls the project's ``striptags`` when +``custom.py`` registers one, a later load overrides an earlier one, and +``{% load striptags from custom %}`` selects. The engine reads what a library registers +on its ``register`` with the semantic layers, and where it cannot tell — no such library, +a file it cannot read or understand, several apps giving the name and disagreeing — the +filter is neither the built-in nor certainly replaced: no call is claimed, and the load +is listed among what the engine could not read. ``OPTIONS["builtins"]`` and +``OPTIONS["libraries"]`` of the settings are not read. """ from __future__ import annotations import re -from collections.abc import Iterable, Iterator, Mapping +from collections.abc import Container, Iterable, Iterator, Mapping from dataclasses import dataclass, field from pathlib import Path from types import MappingProxyType from typing import ClassVar from coretrace_python.analysis import Analysis, AnalysisContext +from coretrace_python.frontend import HIRBuildError, ParseError, build_hir from coretrace_python.hir import nodes from coretrace_python.hir.visitors import Node, children from coretrace_python.interprocedural import ( @@ -44,11 +56,21 @@ TemplateFilter, discover_files, ) -from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.source import SourceId, SourceSpan +from coretrace_python.semantic.imports import ImportResolutionError, analyze_imports +from coretrace_python.semantic.scopes import ( + BindingKind, + ResolutionKind, + ScopeError, + ScopeId, + ScopeTable, + analyze_scopes, +) +from coretrace_python.semantic.symbols import SymbolId, analyze_symbols +from coretrace_python.source import SourceId, SourceManager, SourceSpan from coretrace_python.taint.models import ModelTable TEMPLATES_DIRECTORY = "templates" +TEMPLATETAGS_DIRECTORY = "templatetags" _OUTPUT, _TAG, _URL = "\x00", "\x01", "/u" _SYNTAX = re.compile(r"{{(.*?)}}|{%(.*?)%}|{#.*?#}", re.DOTALL) @@ -102,6 +124,10 @@ # Every filter but ``safe`` and ``safeseq`` escapes its input, or returns it for # autoescaping to escape, when autoescaping is on. _FILTERS = frozenset(FILTER_FUNCTIONS) - {"safe", "safeseq"} +# The class a project tag library binds ``register`` to, by either of its paths, and +# its methods registering tags, which register no filter. +_LIBRARY_CLASSES = frozenset(SymbolId(f"python.django.template.{p}") for p in ("Library", "library.Library")) +_TAG_REGISTRARS = frozenset({"tag", "tag_function", "simple_tag", "simple_block_tag", "inclusion_tag"}) _URL_ATTRIBUTES = frozenset( { @@ -131,6 +157,8 @@ # the request context processor adds gives (``request.GET``). _VARIABLE = re.compile(r"([A-Za-z_]\w*)(?:\.(\w+))?") _LITERALS = frozenset({"True", "False", "None"}) +# The variable a block renders to keep the parent's content, as ``_chain`` reads it. +_BLOCK_SUPER = "block.super" REQUEST_PROCESSOR = "django.template.context_processors.request" _DJANGO_ENGINE = "django.template.backends.django.DjangoTemplates" @@ -314,10 +342,11 @@ class FilterCall: @dataclass(frozen=True) class ProjectTemplates: """The templates found under the project's ``templates`` directories: their - ``names``, as a render call names them; the filters they apply, as ``calls``; where - they name a template the engine cannot read, as ``unread``; and, by name, the filter - calls rendering each template makes with the context keys reaching them, as - ``filters``.""" + ``names``, as a render call names them; the filters of Django's own libraries they + apply, as ``calls``, unless a project library they load replaces the filter; where + they name a template the engine cannot read, or load a library it cannot read while + applying a filter of that name, as ``unread``; and, by name, the filter calls + rendering each template makes with the context keys reaching them, as ``filters``.""" names: frozenset[str] = frozenset() calls: tuple[FilterCall, ...] = () @@ -363,13 +392,44 @@ def maps(self, names: frozenset[str]) -> frozenset[str]: return frozenset(found) +@dataclass(frozen=True) +class _Load: + """A ``{% load %}`` of ``library`` at ``offset``, of the ``names`` it selects + (``{% load a b from lib %}``), None for the whole library.""" + + library: str + names: frozenset[str] | None + offset: int + + +@dataclass(frozen=True) +class _Library: + """What the project's tag library files of one name register: the filter names they + surely register, ``certain``, and those they may register, ``uncertain`` — None when + they may register anything, as a file the engine cannot read or understand may.""" + + certain: frozenset[str] = frozenset() + uncertain: frozenset[str] | None = None + + def overrides(self, name: str) -> bool | None: + """Whether loading the library replaces the filter ``name``; None when it may.""" + + if name in self.certain: + return True + return None if self.uncertain is None or name in self.uncertain else False + + @dataclass(frozen=True) class _Template: - """What one template file applies and references, and the ``blocks`` it defines.""" + """What one template file applies and references, the ``blocks`` it defines, and + among them the ``extended`` ones, which render ``{{ block.super }}``, and the loads + under which it applies a filter the engine could not resolve.""" applications: tuple[_Application, ...] references: tuple[_Reference, ...] blocks: frozenset[str] + extended: frozenset[str] = frozenset() + unresolved: tuple[_Load, ...] = () def rendered(self, blocks: tuple[str, ...], overridden: frozenset[str]) -> bool: """Whether what stands in ``blocks`` renders when a template extending this one @@ -379,15 +439,25 @@ def rendered(self, blocks: tuple[str, ...], overridden: frozenset[str]) -> bool: extends = any(reference.extends for reference in self.references) return not overridden.intersection(blocks) and (bool(blocks) or not extends) + @property + def replaced(self) -> frozenset[str]: + """The blocks this template replaces in what it extends: those it defines + without rendering ``{{ block.super }}``, which keeps the parent's content.""" + + return self.blocks - self.extended + def project_templates(root: Path) -> ProjectTemplates: """What the templates under ``root`` call, and the templates they include or extend that the engine cannot read: named by an expression, or by a name found under no - ``templates`` directory. A template file that cannot be read is unread too. A name + ``templates`` directory. A template file that cannot be read is unread too, and so + is a template loading a library the engine cannot read, or find, or finds in several + apps disagreeing, where it then applies a filter of that library's name. A name found in several files gives no ``filters``: which one renders depends on the loaders.""" found = {path: tuple(_names(path.relative_to(root).parts)) for path in discover_files(root)} + libraries = _libraries(found) templates = sorted(path for path, names in found.items() if names) files: dict[str, list[Path]] = {} for path in templates: @@ -401,13 +471,17 @@ def project_templates(root: Path) -> ProjectTemplates: if text is None: unread.append(f"{relative}, unreadable") continue - read[path] = _read_scopes(text, SourceId(str(path))) + read[path] = _read_scopes(text, SourceId(str(path)), libraries) for reference in read[path].references: where = f"{relative}:{_position(text, reference.offset)[0]}" if reference.name is None: unread.append(f"{where} names a template by an expression") elif reference.name not in files: unread.append(f"{where} names {reference.name!r}, not found") + for load in read[path].unresolved: + library = libraries.get(load.library) + reason = "not found" if library is None else "not read" if library.uncertain is None else "found in several files" + unread.append(f"{relative}:{_position(text, load.offset)[0]} loads {load.library!r}, {reason}") calls = tuple(FilterCall(a.symbol, a.span) for template in read.values() for a in template.applications) filters = { name: tuple( @@ -444,7 +518,7 @@ def _reaching( if reference.name is None: continue if reference.extends: - found = _reaching(reference.name, files, read, visiting | {name}, overridden | template.blocks) + found = _reaching(reference.name, files, read, visiting | {name}, overridden | template.replaced) elif template.rendered(reference.blocks, overridden): found = _reaching(reference.name, files, read, visiting | {name}) else: @@ -453,18 +527,23 @@ def _reaching( yield symbol, span, where, reference.maps(value), reference.maps(argument) -def _read_scopes(text: str, source: SourceId) -> _Template: +def _read_scopes(text: str, source: SourceId, libraries: Mapping[str, _Library]) -> _Template: """The filters a template applies and the templates it includes or extends, with the names free in it that reach each: through ``{% for %}``, ``{% with %}`` and the ``as`` of other tags, which bind a name until their block ends, the arguments of - earlier filters, and the output of ``{% filter %}`` blocks.""" + earlier filters, and the output of ``{% filter %}`` blocks. A filter is the one of + Django's own libraries unless a project library, among ``libraries``, loaded before + it in the file replaces it, as Django's parser resolves filters.""" applications: list[_Application] = [] references: list[_Reference] = [] opened: list[str] = [] defined: set[str] = set() + extended: set[str] = set() scopes: list[dict[str, frozenset[str]]] = [{}] blocks: list[tuple[list[tuple[str, int, frozenset[str]]], set[str]]] = [] + loaded: list[_Load] = [] + unresolved: dict[_Load, None] = {} def free(names: frozenset[str]) -> frozenset[str]: found: set[str] = set() @@ -474,14 +553,36 @@ def free(names: frozenset[str]) -> frozenset[str]: found |= {name} if bound is None else _through(bound, name) return frozenset(found) + def behind(name: str) -> SymbolId | _Load | None: + """The function behind the filter ``name`` applied here: the one of Django's own + libraries, unless the latest load so far of a library registering ``name`` + replaces it (None), or may (that load). Django's own libraries register no name + of another, so their loads change nothing.""" + + builtin = FILTER_FUNCTIONS.get(name) + if builtin is None: + return None + for load in reversed(loaded): + if (load.names is not None and name not in load.names) or load.library in _LIBRARIES: + continue + library = libraries.get(load.library) + verdict = library.overrides(name) if library is not None else None + if verdict is None: + return load + if verdict: + return None + return builtin + def apply(head: frozenset[str], chain: list[tuple[str, int, frozenset[str]]]) -> frozenset[str]: """Record the filters of ``chain`` applied to a value the free names ``head`` reach; the free names reaching the value of the whole expression.""" value = head for name, offset, argument in chain: - symbol = FILTER_FUNCTIONS.get(name) - if symbol is not None: + symbol = behind(name) + if isinstance(symbol, _Load): + unresolved[symbol] = None + elif symbol is not None: line, column = _position(text, offset) span = SourceSpan(source, line, column) applications.append(_Application(symbol, span, value, free(argument), tuple(opened))) @@ -491,6 +592,8 @@ def apply(head: frozenset[str], chain: list[tuple[str, int, frozenset[str]]]) -> for offset, expression, tag in _expressions(text): if not tag: head, chain = _chain(expression, offset) + if opened and _BLOCK_SUPER in head: + extended.add(opened[-1]) printed = apply(free(head), chain) for _, collected in blocks: collected |= printed @@ -543,10 +646,145 @@ def apply(head: frozenset[str], chain: list[tuple[str, int, frozenset[str]]]) -> else: context = {k: v for scope in scopes for k, v in scope.items()} references.append(_Reference(named, offset, bindings, "only" in words, context, tuple(opened))) + elif name == "load": + selected = frozenset(words[1 : words.index("from")]) if "from" in words else None + loaded.extend(_Load(library, selected, offset) for library in _loaded(words[1:])) elif "as" in words[1:-1]: target = words[words.index("as", 1) + 1] scopes[-1][target] = values.get(words[1], frozenset()) if name == "regroup" else frozenset() - return _Template(tuple(applications), tuple(references), frozenset(defined)) + return _Template( + tuple(applications), tuple(references), frozenset(defined), frozenset(extended), tuple(unresolved) + ) + + +def _libraries(paths: Iterable[Path]) -> dict[str, _Library]: + """The project's tag libraries among ``paths``, by the name ``{% load %}`` gives + them: the modules of the ``templatetags`` directories. Of several files of one name, + in different apps, only what every file registers is certain, and one the engine + cannot read leaves everything uncertain.""" + + readings: dict[str, list[frozenset[str] | None]] = {} + for path in paths: + if path.suffix == ".py" and path.parent.name == TEMPLATETAGS_DIRECTORY and path.stem != "__init__": + readings.setdefault(path.stem, []).append(_registered_filters(path)) + libraries: dict[str, _Library] = {} + for name, found in readings.items(): + read = [names for names in found if names is not None] + if len(read) < len(found): + libraries[name] = _Library() + else: + certain = read[0].intersection(*read[1:]) + libraries[name] = _Library(certain, read[0].union(*read[1:]) - certain) + return libraries + + +def _registered_filters(path: Path) -> frozenset[str] | None: + """The names of the filters the tag library module ``path`` registers, or None when + the engine cannot tell them all: the file cannot be read or parsed, ``register`` is + not bound to a ``Library`` at module level, a registration's name is not written as + a constant, or the library is filled or passed around in a way the engine does not + read. The ``Library`` is known by its symbol, whatever the import spelling, and a + registration under ``if``, ``try`` or a function counts.""" + + try: + module = build_hir(SourceManager().load_file(path)) + scopes = analyze_scopes(module) + symbols = analyze_symbols(scopes, analyze_imports(module, scopes), module) + except (OSError, UnicodeDecodeError, ParseError, HIRBuildError, ScopeError, ImportResolutionError): + return None + scope = scopes.module_scope.id + if symbols.resolve(scope, "register") not in _LIBRARY_CLASSES: + return None + functions = {name for name, bound in scopes.module_scope.bindings.items() if bound.kind is BindingKind.FUNCTION} + registered: set[str] = set() + + def record(name: str | None) -> bool: + if name is not None: + registered.add(name) + return name is not None + + def visit(node: Node) -> bool: + """Read the registrations under ``node``; False when one cannot be read.""" + + if isinstance(node, nodes.Function): + for decorator in node.decorators: + if _library_attribute(symbols.resolve_expression(scope, decorator)) == "filter": + named = _decorated(decorator, node.name) if isinstance(decorator, nodes.Call) else node.name + if not record(named): + return False + elif not visit(decorator): + return False + return all(map(visit, (*node.parameters, *node.body))) + if isinstance(node, nodes.Call) and isinstance(node.callee, nodes.Name | nodes.Attribute): + callee = symbols.resolve_expression(scope, node.callee) + if callee in _LIBRARY_CLASSES: + return not (node.arguments or node.keywords) # ``Library(filters=...)`` fills it unseen + if _library_attribute(callee) == "filter": + return record(_registered(node, functions)) and all(map(visit, (*node.arguments, *node.keywords))) + if isinstance(node, nodes.Name | nodes.Attribute): + symbol = symbols.resolve_expression(scope, node) + if _library_attribute(symbol) in _TAG_REGISTRARS: + return True + if symbol in _LIBRARY_CLASSES or _library_attribute(symbol) is not None: + return False # the library, or its filters, in hands the engine does not follow + if isinstance(node, nodes.Assign) and isinstance(node.target, nodes.Name): + return visit(node.value) + return all(map(visit, children(node))) + + return frozenset(registered) if all(map(visit, module.body)) else None + + +def _library_attribute(symbol: SymbolId | None) -> str | None: + """The attribute of a ``Library`` that ``symbol`` names, ``filter`` for + ``register.filter``; None for any other symbol.""" + + if symbol is None: + return None + owner, _, name = symbol.canonical_name.rpartition(".") + return name if "." in owner and SymbolId(owner) in _LIBRARY_CLASSES else None + + +def _decorated(call: nodes.Call, function: str) -> str | None: + """What ``@register.filter(...)`` registers ``function`` as: the constant ``name`` it + gives, or ``function`` itself when it gives only flags; None when the engine cannot + tell.""" + + given = _arguments(call, "name") + if given is None: + return None + return function if given[0] is None else _string(given[0]) + + +def _registered(call: nodes.Call, functions: Container[str]) -> str | None: + """What ``register.filter(...)`` registers when called as a statement: the constant + name of ``register.filter("x", fn)`` and ``register.filter(name="x", filter_func=fn)``, + the function's own name for ``register.filter(fn)`` when the module defines + ``fn``; None when the engine cannot tell.""" + + given = _arguments(call, "name", "filter_func") + if given is None: + return None + name, function = given + if function is None: + return name.identifier if isinstance(name, nodes.Name) and name.identifier in functions else None + return _string(name) + + +def _arguments(call: nodes.Call, *parameters: str) -> list[nodes.Expression | None] | None: + """The argument ``call`` gives each of ``parameters``, by position or keyword; None + when it unpacks any, which could give them too.""" + + if any(isinstance(a, nodes.Starred) for a in call.arguments) or any(k.name is None for k in call.keywords): + return None + keywords = {keyword.name: keyword.value for keyword in call.keywords} + return [ + call.arguments[position] if position < len(call.arguments) else keywords.get(name) + for position, name in enumerate(parameters) + ] + + +def _string(expression: nodes.Expression | None) -> str | None: + return expression.value if isinstance(expression, nodes.Constant) and isinstance(expression.value, str) else None def _chain(expression: str, offset: int) -> tuple[frozenset[str], list[tuple[str, int, frozenset[str]]]]: @@ -578,29 +816,41 @@ def _through(bound: frozenset[str], name: str) -> frozenset[str]: return frozenset(".".join(f"{b}.{attribute}".split(".")[:2]) if attribute else b for b in bound) -def request_processor(modules: Iterable[nodes.Module]) -> bool: +def request_processor(modules: Iterable[tuple[nodes.Module, ScopeTable]]) -> bool: """Whether every Django template engine the project's settings configure runs the - ``request`` context processor, for certain: each assignment to ``TEMPLATES`` is a - literal list of literal engines, at least one of them ``DjangoTemplates``, and each - of those lists the processor in a literal ``context_processors``; no other code - names ``TEMPLATES``, which could change it. A mere mention of the processor proves - nothing.""" + ``request`` context processor, for certain: each module-level assignment to + ``TEMPLATES`` is a literal list of literal engines, at least one of them + ``DjangoTemplates``, and each of those lists the processor in a literal + ``context_processors``; no other code names the module's ``TEMPLATES``, or a + ``TEMPLATES`` it does not bind (a star import), which could change it. A + ``TEMPLATES`` bound in a function or a class is another name, which configures + nothing and changes nothing. A mere mention of the processor proves nothing.""" assignments = mentions = 0 - for module in modules: - stack: list[Node] = list(module.body) - while stack: - node = stack.pop() - stack.extend(children(node)) + for module, scopes in modules: + for node, scope in _scoped(module, scopes): if isinstance(node, nodes.Name) and node.identifier == _ENGINES_SETTING: - mentions += 1 - elif isinstance(node, nodes.Assign) and _names_engines(node.target): + if scopes.resolve(scope, node.identifier).kind in (ResolutionKind.GLOBAL, ResolutionKind.UNBOUND): + mentions += 1 + elif isinstance(node, nodes.Assign) and _names_engines(node.target) and scope == scopes.module_scope.id: if not _runs_request_processor(node.value): return False assignments += 1 return assignments > 0 and mentions == assignments +def _scoped(module: nodes.Module, scopes: ScopeTable) -> Iterator[tuple[Node, ScopeId]]: + """Every node of ``module`` with the scope it stands in.""" + + stack: list[tuple[Node, ScopeId]] = [(node, scopes.module_scope.id) for node in module.body] + while stack: + node, scope = stack.pop() + yield node, scope + if isinstance(node, nodes.Function | nodes.Lambda | nodes.Class | nodes.Comprehension): + scope = scopes.scope_for(node).id + stack.extend((child, scope) for child in children(node)) + + def _names_engines(target: nodes.Expression) -> bool: return isinstance(target, nodes.Name) and target.identifier == _ENGINES_SETTING diff --git a/tests/regression/expected/DjangoGoat.json b/tests/regression/expected/DjangoGoat.json index 0f0a09a..3374f0a 100644 --- a/tests/regression/expected/DjangoGoat.json +++ b/tests/regression/expected/DjangoGoat.json @@ -1,11 +1,17 @@ { "coverage": { - "files": 29, - "files_analysed": 29, - "functions": 71, - "functions_analysed": 71 + "files": 31, + "files_analysed": 27, + "functions": 85, + "functions_analysed": 85 }, "findings": [ + { + "file": "authentication/__init__.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, { "file": "authentication/views.py", "line": 57, @@ -25,6 +31,24 @@ "rule": "hardcoded-credential", "function": "after_all" }, + { + "file": "features/steps/authentication.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, + { + "file": "features/steps/notes.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, + { + "file": "notes/__init__.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, { "file": "requirements_app.txt", "line": 2, diff --git a/tests/regression/expected/httpie.json b/tests/regression/expected/httpie.json index 7bbf6c1..097fc3d 100644 --- a/tests/regression/expected/httpie.json +++ b/tests/regression/expected/httpie.json @@ -1,9 +1,9 @@ { "coverage": { - "files": 132, - "files_analysed": 132, - "functions": 1305, - "functions_analysed": 1305 + "files": 133, + "files_analysed": 131, + "functions": 1307, + "functions_analysed": 1307 }, "findings": [ { @@ -19,6 +19,18 @@ "function": "", "verdict": "vulnerability" }, + { + "file": "docs/contributors/generate.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, + { + "file": "docs/installation/generate.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, { "file": "httpie/client.py", "line": 114, diff --git a/tests/regression/expected/pygoat.json b/tests/regression/expected/pygoat.json index 9a381e3..83c6394 100644 --- a/tests/regression/expected/pygoat.json +++ b/tests/regression/expected/pygoat.json @@ -1,9 +1,9 @@ { "coverage": { - "files": 78, - "files_analysed": 76, - "functions": 287, - "functions_analysed": 287 + "files": 80, + "files_analysed": 74, + "functions": 295, + "functions_analysed": 295 }, "findings": [ { @@ -50,12 +50,31 @@ "rule": "debug-enabled", "function": null }, + { + "file": "dockerized_labs/insec_des_lab/main.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, + { + "file": "dockerized_labs/insec_des_lab/main.py", + "line": 36, + "rule": "insecure-deserialization", + "function": "deserialize_data", + "verdict": "vulnerability" + }, { "file": "dockerized_labs/sensitive_data_exposure/dataexposure/views.py", "line": 42, "rule": "weak-random", "function": "generate_api_key" }, + { + "file": "dockerized_labs/sensitive_data_exposure/manage.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, { "file": "dockerized_labs/sensitive_data_exposure/sensitive_data_lab/settings.py", "line": 8, @@ -149,6 +168,12 @@ "rule": "syntax-error", "function": null }, + { + "file": "introduction/playground/A9/main.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, { "file": "introduction/views.py", "line": 162, @@ -321,6 +346,12 @@ "rule": "hardcoded-credential", "function": null }, + { + "file": "manage.py", + "line": 1, + "rule": "ambiguous-module", + "function": null + }, { "file": "pygoat/settings.py", "line": 25, diff --git a/tests/test_block_super.py b/tests/test_block_super.py new file mode 100644 index 0000000..b5a4e63 --- /dev/null +++ b/tests/test_block_super.py @@ -0,0 +1,155 @@ +"""``{{ block.super }}`` keeps the parent block's filters in the render context flow. + +A template extending another renders only its blocks, which replace the blocks of that +name in the parent. A block that renders ``{{ block.super }}`` extends the parent's +block instead of replacing it: Django executes the parent's content, so the filters it +applies stay fed by the render context, with the parent template as the place of the +call. A block without ``block.super`` still replaces the parent's whole block, other +overridden blocks stay excluded, and a chain of templates keeps exactly the parents each +level renders, without duplicate findings. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import Advisory, AdvisoryEntryPoint, dump_advisories +from coretrace_python.findings import Severity +from coretrace_python.semantic.symbols import SymbolId + +STRIPTAGS = "python.django.template.defaultfilters.striptags" +ADVISORY = Advisory( + "CVE-2099-5701", + "django", + "<4.2.17", + "striptags is quadratic in its input", + Severity.MEDIUM, + entry_points=(AdvisoryEntryPoint(SymbolId(STRIPTAGS), "a template filter"),), +) +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "django" }] + +[[package]] +name = "django" +version = "4.2.16" +source = { registry = "https://pypi.org/simple" } +""" +VIEW = ( + "from django.http import HttpRequest\n" + "from django.shortcuts import render\n\n" + "def profile(request: HttpRequest):\n" + " return render(request, 'app/profile.html', {'bio': request.POST['bio']})\n" +) +LINE = 5 +BASE = "app/templates/app/base.html" +MIDDLE = "app/templates/app/middle.html" +PAGE = "app/templates/app/profile.html" + + +def exploitable(root: Path, templates: dict[str, str]) -> list[tuple[int, str, str]]: + """The exploitable striptags calls the render makes: the render line, the template + holding the filter, and the line of the filter there.""" + + files = {"uv.lock": LOCK, "app/__init__.py": "", "app/views.py": VIEW, **templates} + for relative, text in files.items(): + (root / relative).parent.mkdir(parents=True, exist_ok=True) + (root / relative).write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + findings = engine.analyze_project(root, [engine.BUNDLED_PLUGINS]).findings + return sorted( + (f.span.start_line, f.metadata["through"], f.metadata["sink_line"]) + for f in findings + if f.rule_id == "exploitable-vulnerability" and f.metadata["advisory"] == ADVISORY.id + ) + + +def test_a_block_rendering_block_super_keeps_the_parent_flow(tmp_path: Path) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n", + PAGE: "{% extends 'app/base.html' %}\n{% block body %}{{ block.super }}{% endblock %}\n", + } + + assert exploitable(tmp_path, templates) == [(LINE, "app/base.html", "1")] + + +def test_a_block_may_add_its_own_filters_beside_block_super(tmp_path: Path) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n", + PAGE: "{% extends 'app/base.html' %}\n{% block body %}

{{ bio|striptags }}

\n{{ block.super }}{% endblock %}\n", + } + + assert exploitable(tmp_path, templates) == [(LINE, "app/base.html", "1"), (LINE, "app/profile.html", "2")] + + +def test_a_replacement_without_block_super_still_excludes_the_parent(tmp_path: Path) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n", + PAGE: "{% extends 'app/base.html' %}\n{% block body %}{{ bio }}{% endblock %}\n", + } + + assert exploitable(tmp_path, templates) == [] + + +def test_another_block_does_not_come_back_through_one_block_super(tmp_path: Path) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n{% block aside %}{{ bio|striptags }}{% endblock %}\n", + PAGE: ( + "{% extends 'app/base.html' %}\n" + "{% block body %}{{ block.super }}{% endblock %}\n" + "{% block aside %}{{ bio }}{% endblock %}\n" + ), + } + + assert exploitable(tmp_path, templates) == [(LINE, "app/base.html", "1")] + + +def test_block_super_at_each_level_keeps_the_whole_chain_once(tmp_path: Path) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n", + MIDDLE: "{% extends 'app/base.html' %}\n{% block body %}{{ block.super }}{{ bio|striptags }}{% endblock %}\n", + PAGE: "{% extends 'app/middle.html' %}\n{% block body %}{{ block.super }}{% endblock %}\n", + } + + assert exploitable(tmp_path, templates) == [(LINE, "app/base.html", "1"), (LINE, "app/middle.html", "2")] + + +def test_a_replacement_in_the_middle_cuts_the_chain_above_it(tmp_path: Path) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n", + MIDDLE: "{% extends 'app/base.html' %}\n{% block body %}{{ bio|striptags }}{% endblock %}\n", + PAGE: "{% extends 'app/middle.html' %}\n{% block body %}{{ block.super }}{% endblock %}\n", + } + + assert exploitable(tmp_path, templates) == [(LINE, "app/middle.html", "2")] + + +def test_a_replacement_at_the_end_cuts_the_whole_chain(tmp_path: Path) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n", + MIDDLE: "{% extends 'app/base.html' %}\n{% block body %}{{ block.super }}{{ bio|striptags }}{% endblock %}\n", + PAGE: "{% extends 'app/middle.html' %}\n{% block body %}{{ bio }}{% endblock %}\n", + } + + assert exploitable(tmp_path, templates) == [] + + +@pytest.mark.parametrize( + "child_block", + ["{{ block.super|lower }}", "{% if bio %}{{ block.super }}{% endif %}", "{{ block.super }}{{ block.super }}"], + ids=["filtered", "conditional", "twice"], +) +def test_block_super_counts_however_it_is_rendered(tmp_path: Path, child_block: str) -> None: + templates = { + BASE: "{% block body %}{{ bio|striptags }}{% endblock %}\n", + PAGE: "{% extends 'app/base.html' %}\n{% block body %}" + child_block + "{% endblock %}\n", + } + + assert exploitable(tmp_path, templates) == [(LINE, "app/base.html", "1")] diff --git a/tests/test_entry_point_aliases.py b/tests/test_entry_point_aliases.py new file mode 100644 index 0000000..88b38da --- /dev/null +++ b/tests/test_entry_point_aliases.py @@ -0,0 +1,160 @@ +"""A symbol naming an alias of a project function names that function. + +A deployment declaration names its handler as ``app.main``, and the module may define +``main`` as an alias, ``main = actual``, of the function that does the work. An +``EntryPoint`` naming ``python.app.main`` then makes ``actual`` the entry point, with the +inputs the model gives, as it would a function named ``main``. An alias is a module-level +assignment of one name to another, followed transitively; a class alias names the +class's methods the same way. A name bound to anything else, a call or an imported name, +is no alias: the model names nothing through it, and an integration reporting declared +handlers sees no entry point. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Finding +from coretrace_python.plugins import ProjectContext, ProjectPlugin + +MANIFEST = ( + 'name = "declared-aliases"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\nrequires = []\n' + 'provides = ["model.declared-aliases"]\n\n[entrypoint]\nmodule = "declared"\nclass = "Declared"\n' +) +PLUGIN = ''' +from typing import ClassVar + +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import EntryPoint, Model + + +class Declared(ModelPlugin): + name: ClassVar[str] = "declared-aliases" + models: ClassVar[tuple[Model, ...]] = ( + EntryPoint(SymbolId("python.app.handlers.main"), "event", inputs=(0,)), + EntryPoint(SymbolId("python.app.handlers.Worker"), "event"), + EntryPoint(SymbolId("python.app.handlers.Jobs.run"), "event", inputs=(0,)), + ) +''' +ACTUAL = "import os\n\ndef actual(event, context):\n os.system(event['cmd'])\n os.system(context.function_name)\n\n" + + +def plugins(tmp_path: Path) -> Path: + plugin = tmp_path / "plugins" / "declared" + plugin.mkdir(parents=True, exist_ok=True) + (plugin / "plugin.toml").write_text(MANIFEST, encoding="utf-8") + (plugin / "declared.py").write_text(PLUGIN, encoding="utf-8") + return tmp_path / "plugins" + + +def project(tmp_path: Path, files: dict[str, str]) -> Path: + root = tmp_path / "project" + for relative, text in {"app/__init__.py": "", **files}.items(): + (root / relative).parent.mkdir(parents=True, exist_ok=True) + (root / relative).write_text(text, encoding="utf-8") + return root + + +def commands(findings: tuple[Finding, ...]) -> list[tuple[int, str]]: + return sorted((f.span.start_line, f.metadata["source_label"]) for f in findings if f.rule_id == "command-injection") + + +def analyze(tmp_path: Path, handlers: str, **options: object) -> list[tuple[int, str]]: + root = project(tmp_path, {"app/handlers.py": handlers}) + return commands(engine.analyze_project(root, [engine.BUNDLED_PLUGINS, plugins(tmp_path)], **options).findings) # type: ignore[arg-type] + + +@pytest.mark.parametrize("jobs", [1, 2]) +def test_an_alias_of_a_function_names_that_function(tmp_path: Path, jobs: int) -> None: + assert analyze(tmp_path, ACTUAL + "main = actual\n", jobs=jobs) == [(4, "event")] + + +@pytest.mark.parametrize( + "aliases", + [ + "handler = actual\nmain = handler\n", + "main = handler = actual\n", + "if os.environ.get('DEBUG'):\n main = actual\nelse:\n main = actual\n", + "try:\n main = actual\nexcept NameError:\n pass\n", + ], + ids=["chained", "chained-in-one", "conditional", "guarded"], +) +def test_module_level_aliases_are_followed_transitively(tmp_path: Path, aliases: str) -> None: + assert analyze(tmp_path, ACTUAL + aliases) == [(4, "event")] + + +@pytest.mark.parametrize( + "aliases", + [ + "main = decorate(actual)\n", + "main = actual()\n", + "from app.other import main\n", + "def outer():\n main = actual\n", + "class Config:\n main = actual\n", + "main = 'actual'\n", + ], + ids=["a-call", "a-result", "an-import", "in-a-function", "in-a-class", "a-string"], +) +def test_a_name_bound_to_anything_else_is_no_alias(tmp_path: Path, aliases: str) -> None: + files = {"app/handlers.py": "def decorate(f):\n return f\n\n" + ACTUAL + aliases, "app/other.py": "main = 1\n"} + root = project(tmp_path, files) + + findings = engine.analyze_project(root, [engine.BUNDLED_PLUGINS, plugins(tmp_path)]).findings + + assert commands(findings) == [] + + +def test_an_alias_of_a_class_names_its_methods(tmp_path: Path) -> None: + handlers = ( + "import os\n\n" + "class Impl:\n" + " def start(self, event, context):\n" + " os.system(event['cmd'])\n" + " os.system(context.function_name)\n\n" + " def run(self, event, context):\n" + " os.system(event['cmd'])\n" + " os.system(context.function_name)\n\n" + "Worker = Impl\n" + "Jobs = Impl\n" + ) + + # ``Worker`` makes every parameter of every method input; ``Jobs.run``, naming the + # method, wins over it for ``run`` and makes only the event input. + assert analyze(tmp_path, handlers) == [(5, "event"), (6, "event"), (9, "event")] + + +def test_a_class_alias_alone_limits_the_inputs_it_declares(tmp_path: Path) -> None: + handlers = ( + "import os\n\n" + "class Impl:\n" + " def run(self, event, context):\n" + " os.system(event['cmd'])\n" + " os.system(context.function_name)\n\n" + "Jobs = Impl\n" + ) + + assert analyze(tmp_path, handlers) == [(5, "event")] + + +def test_a_project_plugin_sees_the_aliased_function_as_the_entry_point(tmp_path: Path) -> None: + seen: dict[str, str | None] = {} + + class Peek(ProjectPlugin): + name = "peek" + + def analyze_project(self, ctx: ProjectContext) -> tuple[Finding, ...]: + for module in ctx.modules: + for function in ctx.functions(module): + seen[f"{module}:{function.name}"] = function.entry_point + return () + + root = project(tmp_path, {"app/handlers.py": ACTUAL + "main = actual\n\ndef other(event, context):\n pass\n"}) + + engine.analyze_project(root, [engine.BUNDLED_PLUGINS, plugins(tmp_path)], plugins=[Peek()]) + + assert seen["app.handlers:actual"] == "event" + assert seen["app.handlers:other"] is None diff --git a/tests/test_module_identity.py b/tests/test_module_identity.py new file mode 100644 index 0000000..5431a2a --- /dev/null +++ b/tests/test_module_identity.py @@ -0,0 +1,384 @@ +"""Files sharing a Python module name stay distinct modules (#182). + +``a/app.py`` and ``z/app.py``, outside any package, are both the module ``app`` to +Python, each importable from its own directory. The engine keeps one entry per file: its +import name when no other file has it, otherwise its path from the root as a dotted name +(``a.app``, ``z.app``), so both files are discovered, analysed and covered whatever the +order they are found in. An import of a shared name resolves within the importer's own +root only; one no root resolves is reported and not followed. The colliding files carry +an ``ambiguous-module`` note and are covered as ``ambiguous``, since a model or a route +naming ``python.app.main`` applies to every definition of that symbol. +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +from coretrace_python import engine +from coretrace_python.cache import ProjectCache +from coretrace_python.dependency import Advisory, AdvisoryEntryPoint, dump_advisories, render_vex +from coretrace_python.findings import Finding, Severity +from coretrace_python.plugins import ProjectContext, ProjectPlugin +from coretrace_python.reporters import render_json +from coretrace_python.semantic.symbols import SymbolId + +NOW = datetime(2026, 9, 29, 12, 0, tzinfo=UTC) +STRIP_TAGS = "python.django.utils.html.strip_tags" +ADVISORY = Advisory( + "CVE-2099-5801", + "django", + "<4.2.17", + "strip_tags is quadratic in nested incomplete tags", + Severity.MEDIUM, + entry_points=(AdvisoryEntryPoint(SymbolId(STRIP_TAGS), "changed by the fix"),), +) +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "django" }] + +[[package]] +name = "django" +version = "4.2.16" +source = { registry = "https://pypi.org/simple" } +""" + +MANIFEST = ( + 'name = "declared-handlers"\nversion = "1.0.0"\nplugin_api = ">=1,<2"\nrequires = []\n' + 'provides = ["model.declared-handlers"]\n\n[entrypoint]\nmodule = "declared"\nclass = "Declared"\n' +) +PLUGIN = ''' +from typing import ClassVar + +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import EntryPoint, Model + + +class Declared(ModelPlugin): + name: ClassVar[str] = "declared-handlers" + models: ClassVar[tuple[Model, ...]] = (EntryPoint(SymbolId("python.app.main"), "event", inputs=(0,)),) +''' +# The issue's reproduction: ``import os`` makes ```` a second function of the +# vulnerable handler's file. +VULNERABLE = "import os\n\ndef main(event, context):\n os.system(event['cmd'])\n" +CLEAN = "def main(event, context):\n return 'ok'\n" +EXECUTE = "import os\n\ndef execute(command):\n os.system(command)\n" +RETURNS = "def execute(command):\n return command\n" +RUN = "from app import execute\n\ndef run():\n execute(input())\n" +NOTE = "ambiguous-module" + + +def plugins(tmp_path: Path) -> Path: + plugin = tmp_path / "plugins" / "declared" + plugin.mkdir(parents=True, exist_ok=True) + (plugin / "plugin.toml").write_text(MANIFEST, encoding="utf-8") + (plugin / "declared.py").write_text(PLUGIN, encoding="utf-8") + return tmp_path / "plugins" + + +def project(tmp_path: Path, files: dict[str, str]) -> Path: + root = tmp_path / "project" + for relative, text in files.items(): + (root / relative).parent.mkdir(parents=True, exist_ok=True) + (root / relative).write_text(text, encoding="utf-8") + return root + + +def check(tmp_path: Path, root: Path, **options: object) -> engine.ProjectAnalysis: + roots = [engine.BUNDLED_PLUGINS, plugins(tmp_path)] + return engine.analyze_project(root, roots, **options) # type: ignore[arg-type] + + +def analyze(tmp_path: Path, files: dict[str, str], **options: object) -> engine.ProjectAnalysis: + return check(tmp_path, project(tmp_path, files), **options) + + +def located(path: str, root: Path) -> str: + return Path(path).resolve().relative_to(root.resolve()).as_posix() + + +def rules(findings: tuple[Finding, ...], root: Path, rule_id: str) -> list[tuple[str, int]]: + return sorted((located(str(f.span.source_id), root), f.span.start_line) for f in findings if f.rule_id == rule_id) + + +def notes(findings: tuple[Finding, ...], root: Path) -> dict[str, list[str]]: + found: dict[str, list[str]] = {} + for finding in findings: + if finding.rule_id == NOTE: + found.setdefault(located(str(finding.span.source_id), root), []).append(finding.message) + return found + + +# --------------------------------------------------------------------------- discovery + + +def test_two_files_sharing_a_module_name_are_both_discovered_and_covered(tmp_path: Path) -> None: + root = project(tmp_path, {"a/app.py": VULNERABLE, "z/app.py": CLEAN}) + + analysis = check(tmp_path, root) + + assert analysis.graph.modules == ("a.app", "z.app") + assert analysis.graph.source("a.app").path == (root / "a" / "app.py").resolve() + assert analysis.graph.source("a.app").module_name == "app" + assert [(located(d.path, root), d.status, d.functions, d.analysed) for d in analysis.coverage.details] == [ + ("a/app.py", "ambiguous", 2, 2), + ("z/app.py", "ambiguous", 1, 1), + ] + assert analysis.coverage.summary() == "coverage: 0/2 files, 3/3 functions" + assert rules(analysis.findings, root, NOTE) == [("a/app.py", 1), ("z/app.py", 1)] + assert notes(analysis.findings, root) == { + "a/app.py": [ + ( + "a/app.py is the module 'app' like z/app.py: a symbol python.app.* may name a function of " + "either, and an import of 'app' resolves only from its own root" + ) + ], + "z/app.py": [ + ( + "z/app.py is the module 'app' like a/app.py: a symbol python.app.* may name a function of " + "either, and an import of 'app' resolves only from its own root" + ) + ], + } + + +@pytest.mark.parametrize("jobs", [1, 2]) +@pytest.mark.parametrize("vulnerable", ["a", "z"]) +def test_the_vulnerable_handler_keeps_its_finding_whatever_the_discovery_order( + tmp_path: Path, vulnerable: str, jobs: int +) -> None: + clean = "z" if vulnerable == "a" else "a" + root = project(tmp_path, {f"{vulnerable}/app.py": VULNERABLE, f"{clean}/app.py": CLEAN}) + + findings = check(tmp_path, root, jobs=jobs).findings + + commands = [f for f in findings if f.rule_id == "command-injection"] + assert [(located(str(f.span.source_id), root), f.span.start_line) for f in commands] == [ + (f"{vulnerable}/app.py", 4) + ] + assert commands[0].metadata["source_label"] == "event" + assert set(notes(findings, root)) == {"a/app.py", "z/app.py"} + + +def test_dotted_paths_that_coincide_fall_back_to_the_relative_path(tmp_path: Path) -> None: + root = project(tmp_path, {"x.y/app.py": VULNERABLE, "x/y/app.py": CLEAN}) + + analysis = check(tmp_path, root) + + assert analysis.graph.modules == ("x.y/app.py", "x/y/app.py") + assert analysis.graph.source("x.y/app.py").path == (root / "x.y" / "app.py").resolve() + assert [located(d.path, root) for d in analysis.coverage.details] == ["x.y/app.py", "x/y/app.py"] + assert rules(analysis.findings, root, "command-injection") == [("x.y/app.py", 4)] + + +# --------------------------------------------------------------------------- import resolution + + +def test_an_import_of_a_shared_name_resolves_within_the_importers_root(tmp_path: Path) -> None: + root = project(tmp_path, {"a/app.py": EXECUTE, "a/run.py": RUN, "z/app.py": RETURNS, "z/run.py": RUN}) + + analysis = check(tmp_path, root) + + assert analysis.graph.modules == ("a.app", "a.run", "z.app", "z.run") + assert analysis.graph.imports("a.run") == frozenset({"a.app"}) + assert analysis.graph.imports("z.run") == frozenset({"z.app"}) + assert analysis.graph.unresolved("a.run") == frozenset() + commands = [f for f in analysis.findings if f.rule_id == "command-injection"] + assert [(located(str(f.span.source_id), root), f.span.start_line) for f in commands] == [("a/run.py", 4)] + assert commands[0].metadata["through"] == "app.execute" + # The importers share the name ``run`` too; nothing they import stayed unresolved. + assert all("imported here" not in message for messages in notes(analysis.findings, root).values() for message in messages) + + +def test_an_import_no_root_resolves_is_reported_and_not_followed(tmp_path: Path) -> None: + root = project(tmp_path, {"a/app.py": EXECUTE, "z/app.py": RETURNS, "c/run.py": RUN}) + + analysis = check(tmp_path, root) + + assert analysis.graph.modules == ("a.app", "run", "z.app") + assert analysis.graph.imports("run") == frozenset() + assert analysis.graph.unresolved("run") == frozenset({"app"}) + assert rules(analysis.findings, root, "command-injection") == [] + assert notes(analysis.findings, root)["c/run.py"] == [ + "'app' imported here is a/app.py or z/app.py: calls into it are not followed" + ] + assert [(located(d.path, root), d.status) for d in analysis.coverage.details] == [ + ("a/app.py", "ambiguous"), + ("c/run.py", "analysed"), + ("z/app.py", "ambiguous"), + ] + + +THROUGH = "from app import execute\n\ndef through(command):\n execute(command)\n" +CALLER = ( + "from x import through as x_through\n" + "from y import through as y_through\n" + "from app import execute\n\n" + "def run():\n" + " execute(input())\n" + " x_through(input())\n" + " y_through(input())\n" +) + + +def test_a_component_reaching_both_files_follows_neither_through_an_unresolved_import(tmp_path: Path) -> None: + files = {"a/app.py": RETURNS, "a/x.py": THROUGH, "z/app.py": EXECUTE, "z/y.py": THROUGH, "c.py": CALLER} + root = project(tmp_path, files) + + analysis = check(tmp_path, root) + + assert analysis.graph.imports("c") == frozenset({"x", "y"}) + assert analysis.graph.unresolved("c") == frozenset({"app"}) + # ``y`` carries ``z/app.py``'s sink; the unresolved ``execute`` is followed into neither file. + assert rules(analysis.findings, root, "command-injection") == [("c.py", 8)] + assert notes(analysis.findings, root)["c.py"] == [ + "'app' imported here is a/app.py or z/app.py: calls into it are not followed" + ] + + +def test_a_component_reaching_both_files_follows_the_one_its_own_root_resolves(tmp_path: Path) -> None: + files = {"a/app.py": EXECUTE, "a/x.py": THROUGH, "a/c.py": CALLER, "z/app.py": RETURNS, "z/y.py": THROUGH} + root = project(tmp_path, files) + + analysis = check(tmp_path, root) + + assert analysis.graph.imports("c") == frozenset({"a.app", "x", "y"}) + assert analysis.graph.unresolved("c") == frozenset() + # ``execute`` is ``a/app.py``'s, whatever ``z/app.py`` reached through ``y`` defines. + assert rules(analysis.findings, root, "command-injection") == [("a/c.py", 6), ("a/c.py", 7)] + assert "a/c.py" not in notes(analysis.findings, root) + + +def test_a_unique_module_resolves_from_any_directory(tmp_path: Path) -> None: + root = project( + tmp_path, + {"app/__init__.py": "", "app/helpers.py": EXECUTE, "scripts/tool.py": "from app.helpers import execute\n"}, + ) + + analysis = check(tmp_path, root) + + assert analysis.graph.modules == ("app", "app.helpers", "tool") + assert analysis.graph.imports("tool") == frozenset({"app.helpers"}) + assert analysis.graph.unresolved("tool") == frozenset() + assert rules(analysis.findings, root, NOTE) == [] + + +def test_a_package_and_a_script_sharing_a_name_keep_their_own_importers(tmp_path: Path) -> None: + root = project( + tmp_path, + { + "authentication/__init__.py": "", + "authentication/views.py": "import authentication\n", + "features/steps/authentication.py": "", + "features/steps/notes.py": "import authentication\n", + }, + ) + + graph = check(tmp_path, root).graph + + assert graph.modules == ("authentication", "authentication.views", "features.steps.authentication", "notes") + assert graph.imports("authentication.views") == frozenset({"authentication"}) + assert graph.imports("notes") == frozenset({"features.steps.authentication"}) + assert graph.importers("authentication") == frozenset({"authentication.views"}) + assert graph.importers("features.steps.authentication") == frozenset({"notes"}) + + +# --------------------------------------------------------------------------- index, cache and plugins + + +def test_the_index_and_the_cache_keep_the_files_apart(tmp_path: Path) -> None: + root = project(tmp_path, {"a/app.py": VULNERABLE, "z/app.py": CLEAN}) + cache = ProjectCache(tmp_path / "cache") + + first = check(tmp_path, root, cache=cache) + + vulnerable = first.index.summary(SymbolId("python.a.app.main")) + clean = first.index.summary(SymbolId("python.z.app.main")) + assert vulnerable is not None and clean is not None and vulnerable != clean + assert first.index.summary(SymbolId("python.app.main")) is None + assert set(first.keys) == {"a.app", "z.app"} + assert first.keys["a.app"] != first.keys["z.app"] + assert first.reused == () + + second = check(tmp_path, root, cache=cache) + + assert set(second.reused) == {"a.app", "z.app"} + assert second.findings == first.findings + assert rules(second.findings, root, "command-injection") == [("a/app.py", 4)] + + (root / "z" / "app.py").unlink() + third = check(tmp_path, root, cache=cache) + + assert set(third.keys) == {"app"} + assert third.reused == () + assert rules(third.findings, root, "command-injection") == [("a/app.py", 4)] + assert rules(third.findings, root, NOTE) == [] + assert third.coverage.summary() == "coverage: 1/1 files, 2/2 functions" + + +def test_a_project_plugin_sees_both_files_with_their_entry_points(tmp_path: Path) -> None: + seen: dict[str, dict[str, str | None]] = {} + + class Peek(ProjectPlugin): + name = "peek" + + def analyze_project(self, ctx: ProjectContext) -> tuple[Finding, ...]: + for module in ctx.modules: + seen[module] = {f.name: f.entry_point for f in ctx.functions(module)} + return () + + root = project(tmp_path, {"a/app.py": VULNERABLE, "z/app.py": CLEAN}) + + check(tmp_path, root, plugins=[Peek()]) + + assert tuple(seen) == ("a.app", "z.app") + assert seen["a.app"]["main"] == "event" + assert seen["z.app"]["main"] == "event" + + +def test_ambiguous_files_keep_an_unreached_advisory_under_investigation(tmp_path: Path) -> None: + """A file another may be confused with is not fully analysed to the VEX document: + what the project does not reach through the analysed code is not established.""" + + def statement(files: dict[str, str]) -> dict[str, Any]: + root = project(tmp_path, {"uv.lock": LOCK, **files}) + (root / "advisories.json").write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + analysis = check(tmp_path, root) + evidence = (*analysis.findings, *analysis.suppressed, *analysis.accepted) + document = render_vex( + analysis.dependencies, analysis.advisories, evidence, analysis.coverage, root, "coretrace", "0.18.0", NOW + ) + return next(s for s in json.loads(document)["statements"] if s["vulnerability"]["name"] == ADVISORY.id) + + assert statement({"a/app.py": CLEAN, "z/other.py": CLEAN})["status"] == "not_affected" + + ambiguous = statement({"a/app.py": CLEAN, "z/app.py": CLEAN}) + + assert ambiguous["status"] == "under_investigation" + assert ambiguous["status_notes"] == ( + f"No analysed code reaches the entry points of {ADVISORY.id} ({STRIP_TAGS}), but a/app.py, z/app.py " + "could not be fully analysed." + ) + + +def test_the_json_report_carries_the_ambiguous_status(tmp_path: Path) -> None: + root = project(tmp_path, {"a/app.py": VULNERABLE, "z/app.py": CLEAN}) + analysis = check(tmp_path, root) + + document = json.loads(render_json(engine.report(analysis.findings, analysis.coverage, root))) + + assert document["coverage"]["files_analysed"] == 0 + assert [(d["path"], d["status"]) for d in document["coverage"]["details"]] == [ + ("a/app.py", "ambiguous"), + ("z/app.py", "ambiguous"), + ] + assert [f["rule_id"] for f in document["findings"]].count(NOTE) == 2 diff --git a/tests/test_settings_scope.py b/tests/test_settings_scope.py new file mode 100644 index 0000000..0a1ec5b --- /dev/null +++ b/tests/test_settings_scope.py @@ -0,0 +1,109 @@ +"""Only module-level settings configure Django. + +The request context processor is taken as active only from the settings a project +assigns at module level: a ``TEMPLATES`` assigned inside a function, called or not, +binds a local name and configures nothing, and so does a class attribute. A function +that reads or rebinds the module's ``TEMPLATES`` (through ``global``, or by mutating +it) may change the effective settings, so it leaves the activation uncertain, as any +other mention does. Uncertain activation leaves the filter reachable, neither +exploitable through the processor nor declared absent. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import Advisory, AdvisoryEntryPoint, dump_advisories +from coretrace_python.findings import Severity +from coretrace_python.semantic.symbols import SymbolId + +STRIPTAGS = "python.django.template.defaultfilters.striptags" +ADVISORY = Advisory( + "CVE-2099-5601", + "django", + "<4.2.17", + "striptags is quadratic in its input", + Severity.MEDIUM, + entry_points=(AdvisoryEntryPoint(SymbolId(STRIPTAGS), "a template filter"),), +) +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "django" }] + +[[package]] +name = "django" +version = "4.2.16" +source = { registry = "https://pypi.org/simple" } +""" +ENGINE = ( + "{'BACKEND': 'django.template.backends.django.DjangoTemplates', 'APP_DIRS': True, " + "'OPTIONS': {'context_processors': ['django.template.context_processors.request']}}" +) +VIEW = ( + "from django.http import HttpRequest\n" + "from django.shortcuts import render\n\n" + "def search(request: HttpRequest):\n" + " return render(request, 'app/search.html')\n" +) +PAGE = "app/templates/app/search.html" + + +def levels(root: Path, settings: str) -> list[str]: + files = { + "uv.lock": LOCK, + "app/__init__.py": "", + "app/settings.py": settings, + "app/views.py": VIEW, + PAGE: "{{ request.GET.q|striptags }}\n", + } + for relative, text in files.items(): + (root / relative).parent.mkdir(parents=True, exist_ok=True) + (root / relative).write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + findings = engine.analyze_project(root, [engine.BUNDLED_PLUGINS]).findings + reached = ("reachable-vulnerability", "exploitable-vulnerability") + return sorted(f.metadata["level"] for f in findings if f.rule_id in reached and f.metadata["advisory"] == ADVISORY.id) + + +def test_module_level_settings_activate_the_processor(tmp_path: Path) -> None: + assert levels(tmp_path, f"TEMPLATES = [{ENGINE}]\n") == ["exploitable", "reachable"] + + +@pytest.mark.parametrize( + "settings", + [ + f"def never_called():\n TEMPLATES = [{ENGINE}]\n", + f"def configure():\n TEMPLATES = [{ENGINE}]\n return TEMPLATES\n\nconfigure()\n", + f"class Settings:\n TEMPLATES = [{ENGINE}]\n", + f"def build():\n templates = [{ENGINE}]\n return templates\n", + ], + ids=["uncalled-function", "called-function", "class-attribute", "other-local-name"], +) +def test_settings_bound_below_the_module_do_not_configure_django(tmp_path: Path, settings: str) -> None: + assert levels(tmp_path, settings) == ["reachable"] + + +@pytest.mark.parametrize( + "settings", + [ + f"TEMPLATES = [{ENGINE}]\n\ndef tweak():\n TEMPLATES[0]['OPTIONS']['context_processors'].pop(0)\n", + f"TEMPLATES = [{ENGINE}]\n\ndef reset():\n global TEMPLATES\n TEMPLATES = []\n", + f"TEMPLATES = [{ENGINE}]\n\ndef read():\n return TEMPLATES\n", + ], + ids=["mutated-in-a-function", "rebound-through-global", "read-in-a-function"], +) +def test_a_function_touching_the_module_settings_leaves_activation_uncertain(tmp_path: Path, settings: str) -> None: + assert levels(tmp_path, settings) == ["reachable"] + + +def test_a_local_name_beside_module_settings_changes_nothing(tmp_path: Path) -> None: + settings = f"TEMPLATES = [{ENGINE}]\n\ndef helper():\n TEMPLATES = []\n return TEMPLATES\n" + + assert levels(tmp_path, settings) == ["exploitable", "reachable"] diff --git a/tests/test_template_filter_overrides.py b/tests/test_template_filter_overrides.py new file mode 100644 index 0000000..738d472 --- /dev/null +++ b/tests/test_template_filter_overrides.py @@ -0,0 +1,385 @@ +"""Project tag libraries overriding the filters of Django's own libraries (issue #183). + +A project library, ``app/templatetags/custom.py``, registering a filter named +``striptags`` replaces ``django.template.defaultfilters.striptags`` in every template +loading it before applying the filter: ``{% load custom %}{{ bio|striptags }}`` calls +the project's function, so it is no evidence of a call to the vulnerable built-in, at +the template's line or from the render's context. The engine reads what a library +registers with its own semantic layers, whatever the import spelling and the +registration form, and follows Django's parse-time rules: a load affects only the +filters after it, a later load overrides an earlier one, ``{% load a from lib %}`` +selects, and Django's own libraries change nothing. + +Where the engine cannot tell what a library registers, because no file gives the +library, it cannot read or understand the file, or several apps give the name and +disagree, the filter's identity is uncertain: it claims no call, and lists the load +among the templates the project names but it could not read, so an advisory naming +the filter stays ``under_investigation``. A library that registers no filter of the +built-in's name leaves the built-in what it was. +""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import Advisory, AdvisoryEntryPoint, dump_advisories, render_vex +from coretrace_python.findings import Severity +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import project_templates +from coretrace_python.taint.templates import escaped_templates + +NOW = datetime(2026, 9, 29, 12, 0, tzinfo=UTC) +STRIPTAGS = "python.django.template.defaultfilters.striptags" +STRIP_TAGS = "python.django.utils.html.strip_tags" +ADVISORY = Advisory( + "CVE-2099-5301", + "django", + "<4.2.17", + "strip_tags is quadratic in nested incomplete tags", + Severity.MEDIUM, + entry_points=( + AdvisoryEntryPoint(SymbolId(STRIP_TAGS), "changed by the fix"), + AdvisoryEntryPoint(SymbolId(STRIPTAGS), "the striptags filter calls strip_tags"), + ), +) +LOCK = """version = 1 + +[[package]] +name = "app" +version = "0.1.0" +source = { virtual = "." } +dependencies = [{ name = "django" }] + +[[package]] +name = "django" +version = "4.2.16" +source = { registry = "https://pypi.org/simple" } +""" +VIEW = ( + "from django.http import HttpRequest, HttpResponse\n" + "from django.shortcuts import render\n" + "from django.template.loader import render_to_string\n" + "from django.template.response import TemplateResponse\n\n" + "def profile(request: HttpRequest):\n" + " return {call}\n" +) +RENDER = "render(request, {name}, {{'bio': request.POST['bio']}})" +PAGE = "app/templates/app/profile.html" +LIBRARY = "app/templatetags/custom.py" +REGISTER = "from django import template\n\nregister = template.Library()\n\n" +REPLACE = '@register.filter(name="striptags")\ndef replace(value):\n return "constant"\n' +SHOUT = '@register.filter\ndef shout(value):\n return value.upper()\n' +LOADED = "{% load custom %}\n{{ bio|striptags }}\n" +NOT_FOUND = f"{PAGE}:1 loads 'crispy_forms_tags', not found" + + +def write(root: Path, files: dict[str, str]) -> None: + for relative, text in files.items(): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def check(root: Path, files: dict[str, str]) -> engine.ProjectAnalysis: + write(root, {"uv.lock": LOCK, "app/__init__.py": "", **files}) + (root / "advisories.json").write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + return engine.analyze_project(root, [engine.BUNDLED_PLUGINS]) + + +def project(template: str, name: str = "'app/profile.html'", **files: str) -> dict[str, str]: + return {"app/views.py": VIEW.format(call=RENDER.format(name=name)), PAGE: template, **files} + + +def library(body: str = REPLACE, name: str = "custom", app: str = "app") -> dict[str, str]: + """A tag library ``name`` of ``app`` registering what ``body`` registers.""" + + return {f"{app}/__init__.py": "", f"{app}/templatetags/__init__.py": "", f"{app}/templatetags/{name}.py": REGISTER + body} + + +def reached(root: Path, analysis: engine.ProjectAnalysis) -> list[tuple[str, int, str]]: + return [ + (Path(str(f.span.source_id)).relative_to(root).as_posix(), f.span.start_line, f.metadata["symbol"]) + for f in analysis.findings + if f.rule_id == "reachable-vulnerability" + ] + + +def exploitable(analysis: engine.ProjectAnalysis) -> list[tuple[int, str, str, str]]: + return [ + (f.span.start_line, f.metadata["symbol"], f.metadata["through"], f.metadata["sink_line"]) + for f in analysis.findings + if f.rule_id == "exploitable-vulnerability" + ] + + +def vex(root: Path, analysis: engine.ProjectAnalysis) -> dict[str, Any]: + evidence = (*analysis.findings, *analysis.suppressed, *analysis.accepted) + document = render_vex( + analysis.dependencies, analysis.advisories, evidence, analysis.coverage, root, "coretrace", "0.18.0", NOW + ) + return next(s for s in json.loads(document)["statements"] if s["vulnerability"]["name"] == ADVISORY.id) + + +# --------------------------------------------------------------------------- overrides + + +def test_a_project_filter_registered_under_a_built_in_name_replaces_the_built_in(tmp_path: Path) -> None: + analysis = check(tmp_path, project(LOADED, **library())) + + assert reached(tmp_path, analysis) == [] + assert exploitable(analysis) == [] + assert vex(tmp_path, analysis)["status"] == "not_affected" + assert analysis.coverage.unread_templates == () + + +@pytest.mark.parametrize( + "registration", + [ + REPLACE, + '@register.filter("striptags")\ndef replace(value):\n return "constant"\n', + '@register.filter\ndef striptags(value):\n return "constant"\n', + '@register.filter()\ndef striptags(value):\n return "constant"\n', + '@register.filter(is_safe=True)\ndef striptags(value):\n return "constant"\n', + '@register.filter("striptags", is_safe=True)\ndef replace(value):\n return "constant"\n', + '@register.filter(name="striptags", is_safe=True)\ndef replace(value):\n return "constant"\n', + ( + "from django.template.defaultfilters import stringfilter\n\n" + '@register.filter\n@stringfilter\ndef striptags(value):\n return "constant"\n' + ), + 'def replace(value):\n return "constant"\n\nregister.filter("striptags", replace)\n', + 'def replace(value):\n return "constant"\n\nregister.filter(name="striptags", filter_func=replace)\n', + 'def replace(value):\n return "constant"\n\nregister.filter("striptags", filter_func=replace)\n', + 'def striptags(value):\n return "constant"\n\nregister.filter(striptags)\n', + 'def striptags(value):\n return "constant"\n\nregister.filter(striptags, is_safe=True)\n', + 'register.filter("striptags", lambda value: "constant")\n', + 'if True:\n @register.filter(name="striptags")\n def replace(value):\n return "constant"\n', + ( + 'try:\n @register.filter(name="striptags")\n def replace(value):\n return "constant"\n' + "except ImportError:\n pass\n" + ), + 'def setup():\n @register.filter(name="striptags")\n def replace(value):\n return "constant"\n\nsetup()\n', + ], + ids=[ + "name-keyword", "name-positional", "bare", "empty-call", "flags-only", "name-and-flag", + "name-keyword-and-flag", "stringfilter", "call-positional", "call-keywords", "call-mixed", + "call-function", "call-function-and-flag", "call-lambda", "in-if", "in-try", "in-function", + ], +) +def test_every_registration_form_overrides(tmp_path: Path, registration: str) -> None: + analysis = check(tmp_path, project(LOADED, **library(registration))) + + assert reached(tmp_path, analysis) == [] + assert analysis.coverage.unread_templates == () + + +@pytest.mark.parametrize( + "binding", + [ + "from django import template\n\nregister = template.Library()\n", + "from django.template import Library\n\nregister = Library()\n", + "from django.template.library import Library\n\nregister = Library()\n", + "import django.template\n\nregister = django.template.Library()\n", + "from django import template as t\n\nregister = t.Library()\n", + "from django.template import Library as L\n\nregister = L()\n", + "from django.template import Library\n\nregister: Library = Library()\n", + ], + ids=["from-django", "from-template", "from-library", "import", "module-alias", "class-alias", "annotated"], +) +def test_every_import_spelling_of_the_library_class_is_recognised(tmp_path: Path, binding: str) -> None: + files = {**library(), LIBRARY: binding + "\n" + REPLACE} + + analysis = check(tmp_path, project(LOADED, **files)) + + assert reached(tmp_path, analysis) == [] + assert analysis.coverage.unread_templates == () + + +@pytest.mark.parametrize( + "template, files, line", + [ + (LOADED, library(SHOUT), 2), + ("{{ bio|striptags }}\n", library(), 1), + ("{% load humanize %}\n{{ bio|striptags }}\n", {}, 2), + ], + ids=["another-filter", "not-loaded", "django-library"], +) +def test_without_an_overriding_load_the_built_in_is_called( + tmp_path: Path, template: str, files: dict[str, str], line: int +) -> None: + analysis = check(tmp_path, project(template, **files)) + + assert reached(tmp_path, analysis) == [(PAGE, line, STRIPTAGS)] + assert analysis.coverage.unread_templates == () + + +def test_a_load_affects_only_the_filters_after_it(tmp_path: Path) -> None: + template = "{{ bio|striptags }}\n{% load custom %}\n{{ bio|striptags }}\n" + + analysis = check(tmp_path, project(template, **library())) + + assert reached(tmp_path, analysis) == [(PAGE, 1, STRIPTAGS)] + + +@pytest.mark.parametrize( + "loads, unread", + [ + ("{% load custom %}{% load humanize %}", ()), + ("{% load humanize %}{% load custom %}", ()), + ("{% load other %}{% load custom %}", ()), + ("{% load custom %}{% load other %}", ()), + ("{% load crispy_forms_tags %}{% load custom %}", ()), + ("{% load custom %}{% load crispy_forms_tags %}", (NOT_FOUND,)), + ("{% load crispy_forms_tags %}{% load other %}", (NOT_FOUND,)), + ], + ids=[ + "django-after", "django-before", "another-before", "another-after", "unknown-before", + "unknown-after", "unknown-then-another", + ], +) +def test_later_loads_override_earlier_ones(tmp_path: Path, loads: str, unread: tuple[str, ...]) -> None: + files = {**library(), **library(SHOUT, name="other")} + + analysis = check(tmp_path, project(loads + "\n{{ bio|striptags }}\n", **files)) + + assert reached(tmp_path, analysis) == [] + assert analysis.coverage.unread_templates == unread + + +@pytest.mark.parametrize( + "load, calls, unread", + [ + ("{% load striptags from custom %}", [], ()), + ("{% load shout striptags from custom %}", [], ()), + ("{% load shout from custom %}", [(PAGE, 2, STRIPTAGS)], ()), + ("{% load striptags from crispy_forms_tags %}", [], (NOT_FOUND,)), + ("{% load shout from crispy_forms_tags %}", [(PAGE, 2, STRIPTAGS)], ()), + ], + ids=["selected", "among-selected", "not-selected", "selected-unknown", "not-selected-unknown"], +) +def test_a_selective_load_overrides_only_the_filters_it_names( + tmp_path: Path, load: str, calls: list[tuple[str, int, str]], unread: tuple[str, ...] +) -> None: + analysis = check(tmp_path, project(load + "\n{{ bio|striptags }}\n", **library(REPLACE + "\n" + SHOUT))) + + assert reached(tmp_path, analysis) == calls + assert analysis.coverage.unread_templates == unread + + +# --------------------------------------------------------------------------- uncertainty + + +def test_an_unknown_library_leaves_the_filter_uncertain(tmp_path: Path) -> None: + analysis = check(tmp_path, project("{% load crispy_forms_tags %}\n{{ bio|striptags }}\n")) + + assert reached(tmp_path, analysis) == [] + assert exploitable(analysis) == [] + assert analysis.coverage.unread_templates == (NOT_FOUND,) + statement = vex(tmp_path, analysis) + assert statement["status"] == "under_investigation" + assert statement["status_notes"] == ( + f"No analysed code reaches the entry points of {ADVISORY.id} ({STRIPTAGS}, {STRIP_TAGS}), but a template " + f"can reach them, and the engine could not read every template the project names: {NOT_FOUND}." + ) + + +def test_an_unknown_library_whose_filters_are_no_built_in_adds_nothing(tmp_path: Path) -> None: + analysis = check(tmp_path, project("{% load crispy_forms_tags %}\n{{ form|crispy }}\n")) + + assert reached(tmp_path, analysis) == [] + assert analysis.coverage.unread_templates == () + assert vex(tmp_path, analysis)["status"] == "not_affected" + + +@pytest.mark.parametrize( + "text", + [ + "from django import template\n\nregister = template.Library(\n", + REGISTER.replace("\n\nregister", '\n\nNAME = "striptags"\nregister') + REPLACE.replace('"striptags"', "NAME"), + REGISTER + 'def replace(value):\n return "constant"\n\nregister.filters["striptags"] = replace\n', + ( + "from django import template\n\n" + 'def replace(value):\n return "constant"\n\n' + 'register = template.Library(filters={"striptags": replace})\n' + ), + REGISTER + "from app.setup import setup\n\nsetup(register)\n", + REGISTER + 'from app.setup import setup\n\n@setup(register)\ndef replace(value):\n return "constant"\n', + REGISTER.replace("register", "library") + REPLACE.replace("register", "library"), + REGISTER + 'add = register.filter\n\ndef replace(value):\n return "constant"\n\nadd("striptags", replace)\n', + REGISTER + 'ARGUMENTS = ("striptags",)\n\n@register.filter(*ARGUMENTS)\ndef replace(value):\n return "constant"\n', + REGISTER + 'def replace(value):\n return "constant"\n\nregister.filter(name="striptags")(replace)\n', + REGISTER + 'from django.template.defaultfilters import striptags\n\nregister.filter(striptags.__name__, striptags)\n', + ], + ids=[ + "syntax-error", "name-variable", "filters-item", "filters-argument", "passed-elsewhere", + "passed-to-a-decorator", "register-unbound", "filter-as-value", "starred", "chained-call", "attribute-name", + ], +) +def test_a_library_the_engine_cannot_read_leaves_the_filter_uncertain(tmp_path: Path, text: str) -> None: + files = {**library(), LIBRARY: text} + + analysis = check(tmp_path, project(LOADED, **files)) + + assert reached(tmp_path, analysis) == [] + assert analysis.coverage.unread_templates == (f"{PAGE}:1 loads 'custom', not read",) + + +@pytest.mark.parametrize( + "first, second, calls, unread", + [ + (REPLACE, REPLACE, [], ()), + (REPLACE, SHOUT, [], (f"{PAGE}:1 loads 'custom', found in several files",)), + (SHOUT, SHOUT, [(PAGE, 2, STRIPTAGS)], ()), + ], + ids=["both-override", "one-overrides", "neither-overrides"], +) +def test_two_apps_sharing_a_library_name_must_agree( + tmp_path: Path, first: str, second: str, calls: list[tuple[str, int, str]], unread: tuple[str, ...] +) -> None: + files = {**library(first), **library(second, app="other")} + + analysis = check(tmp_path, project(LOADED, **files)) + + assert reached(tmp_path, analysis) == calls + assert analysis.coverage.unread_templates == unread + + +# --------------------------------------------------------------------------- consequences + + +def test_an_override_calling_the_vulnerable_function_is_an_ordinary_call_at_its_line(tmp_path: Path) -> None: + body = ( + "from django.utils.html import strip_tags\n\n" + '@register.filter(name="striptags")\ndef replace(value):\n return strip_tags(value)\n' + ) + + analysis = check(tmp_path, project(LOADED, **library(body))) + + assert reached(tmp_path, analysis) == [(LIBRARY, 9, STRIP_TAGS)] + + +def test_an_override_removes_the_filter_from_the_template_calls(tmp_path: Path) -> None: + other = "app/templates/app/other.html" + views = project(LOADED)["app/views.py"] + "\ndef other(request: HttpRequest):\n return " + RENDER.format(name="'app/other.html'") + "\n" + files = {**project(LOADED, **library()), "app/views.py": views, other: "{{ bio|striptags }}\n"} + + analysis = check(tmp_path, files) + templates = project_templates(tmp_path) + + assert exploitable(analysis) == [(10, STRIPTAGS, "app/other.html", "1")] + assert [(Path(str(c.span.source_id)).relative_to(tmp_path).as_posix(), c.span.start_line) for c in templates.calls] == [ + (other, 1) + ] + assert templates.filters["app/profile.html"] == () + assert [(f.template, f.value) for f in templates.filters["app/other.html"]] == [("app/other.html", ("bio",))] + + +def test_a_template_loading_an_overriding_library_is_not_established_to_escape(tmp_path: Path) -> None: + write(tmp_path, {"app/__init__.py": "", PAGE: LOADED, **library()}) + + assert escaped_templates(tmp_path) == frozenset()