From b71e80e709f3d63d703faf6c2a89aedb3996d101 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 14:03:31 +0200 Subject: [PATCH 1/2] test: define plaintext credential storage --- tests/test_detectors.py | 2 + tests/test_django_http_clients.py | 1 + tests/test_plaintext_credentials.py | 188 ++++++++++++++++++++++++++++ 3 files changed, 191 insertions(+) create mode 100644 tests/test_plaintext_credentials.py diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 8deac74..8f041c1 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -164,6 +164,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: assert set(by_name) == { "command-injection", "config-secrets", + "credential-models", "dangerous-eval", "dependency-policy", "django-models", @@ -174,6 +175,7 @@ def test_shipped_plugins_load_with_their_manifests() -> None: "http-client-models", "missing-timeout", "path-traversal", + "plaintext-credentials", "python-stdlib-models", "reachable-vulnerability", "sample-advisories", diff --git a/tests/test_django_http_clients.py b/tests/test_django_http_clients.py index cf483a5..a172499 100644 --- a/tests/test_django_http_clients.py +++ b/tests/test_django_http_clients.py @@ -294,6 +294,7 @@ def test_detectors_stay_generic() -> None: assert sorted(p.name for p in (PLUGINS / "security").iterdir() if p.is_dir()) == [ "command_injection", "path_traversal", + "plaintext_credentials", "sql_injection", "ssrf", "xss", diff --git a/tests/test_plaintext_credentials.py b/tests/test_plaintext_credentials.py new file mode 100644 index 0000000..efd771d --- /dev/null +++ b/tests/test_plaintext_credentials.py @@ -0,0 +1,188 @@ +"""Acceptance tests for plaintext credential storage, the last actionable point of the +``tests-project/SECURITY_AUDIT.md`` review (``uploadbox/db.py:24``). + +A ``CREDENTIAL`` taint kind, outside ``TaintKind.ALL`` so ordinary sources never carry it, +is sourced by a ``NamedParameter`` model (parameters named ``password`` and the like), +sinks into database writes and is cleared by password hashing functions. The detector +reports at medium confidence by construction: a name is a hint, not a proof. The audit's +own case stores through ``get_db().execute(...)``, a connection returned by a project +function, so calls on what a known function returns resolve through its summary. + +Expected to remain red until ``TaintKind.CREDENTIAL``, ``NamedParameter`` and the +``models/credentials`` and ``security/plaintext_credentials`` plugins exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.findings import Confidence, Finding, Severity +from coretrace_python.frontend import build_hir +from coretrace_python.plugins import discover_plugins +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import SecurityModelRegistry, Sink, TaintKind + +try: + from coretrace_python.taint import NamedParameter +except ImportError as error: # pragma: no cover - red until the model lands + MISSING: Exception | None = error +else: + MISSING = None + if not hasattr(TaintKind, "CREDENTIAL"): + MISSING = AttributeError("TaintKind has no CREDENTIAL") + + +@pytest.fixture(autouse=True) +def require_credentials() -> None: + if MISSING is not None: + pytest.fail(f"plaintext credential detection is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def check(text: str) -> tuple[Finding, ...]: + return engine.check(SourceManager().add_source("db.py", "import sqlite3\n\n" + text), [PLUGINS]) + + +def rules(findings: tuple[Finding, ...]) -> list[tuple[str, int]]: + return sorted((f.rule_id, f.span.start_line) for f in findings) + + +REGISTER = ( + "def register_user(email, password):\n" + " conn = sqlite3.connect('app.db')\n" + " conn.execute('insert into users(email, password) values(?, ?)', (email, password))\n" +) + + +# --------------------------------------------------------------------------- model + + +def test_credential_kind_is_outside_the_default_set() -> None: + assert TaintKind.CREDENTIAL & TaintKind.ALL == TaintKind.NONE + assert TaintKind.CREDENTIAL & TaintKind.SQL == TaintKind.NONE + + +def test_named_parameters_are_indexed_and_extended() -> None: + named = NamedParameter(r"(?i)^(password|passwd)$", "credential", TaintKind.CREDENTIAL) + registry = SecurityModelRegistry() + registry.register(named, Sink(SymbolId("python.db.write"), TaintKind.CREDENTIAL)) + table = registry.freeze() + + assert table.named_parameters == (named,) + assert named.matches("password") and named.matches("PASSWD") and not named.matches("password_hash") + assert table.extended(Sink(SymbolId("python.x"), TaintKind.ADVISORY)).named_parameters == (named,) + + +def test_named_parameters_are_credential_sources() -> None: + from coretrace_python.semantic.scopes import ScopeAnalysis + from coretrace_python.semantic.symbols import SymbolAnalysis + from coretrace_python.taint import SecurityModelAnalysis + from coretrace_python.taint.engine import parameter_sources + + module = build_hir(SourceManager().add_source("m.py", "def f(name, password):\n return password\n")) + registry = SecurityModelRegistry() + registry.register(NamedParameter(r"(?i)^password$", "credential", TaintKind.CREDENTIAL)) + manager = engine.build_manager(module, registry) + function = module.body[0] + assert isinstance(function, type(module.body[0])) + + sources = parameter_sources(function, module, manager.get(SecurityModelAnalysis), manager.get(ScopeAnalysis), manager.get(SymbolAnalysis)) # type: ignore[arg-type] + + assert list(sources) == [1] + assert sources[1].label == "credential" and sources[1].kinds == TaintKind.CREDENTIAL + + +# --------------------------------------------------------------------------- detector + + +def test_storing_a_password_parameter_unhashed_is_reported_at_medium_confidence() -> None: + (finding,) = check(REGISTER) + + assert finding.rule_id == "plaintext-credential-storage" + assert finding.span.start_line == 5 + assert finding.severity is Severity.HIGH + assert finding.confidence is Confidence.MEDIUM + assert finding.metadata["source_label"] == "credential" + + +@pytest.mark.parametrize( + "hashing", + [ + "from werkzeug.security import generate_password_hash\n", + "import hashlib\n", + "import bcrypt\n", + "from argon2 import PasswordHasher\n", + ], +) +def test_hashed_passwords_are_not_reported(hashing: str) -> None: + hashed = { + "from werkzeug.security import generate_password_hash\n": "generate_password_hash(password)", + "import hashlib\n": "hashlib.pbkdf2_hmac('sha256', password.encode(), b'salt', 100000)", + "import bcrypt\n": "bcrypt.hashpw(password.encode(), bcrypt.gensalt())", + "from argon2 import PasswordHasher\n": "PasswordHasher().hash(password)", + }[hashing] + findings = check( + hashing + "\ndef register_user(email, password):\n" + f" digest = {hashed}\n" + " conn = sqlite3.connect('app.db')\n" + " conn.execute('insert into users(email, password) values(?, ?)', (email, digest))\n" + ) + assert [f.rule_id for f in findings] == [] + + +def test_http_input_reaching_a_database_is_not_a_credential_finding() -> None: + findings = check( + "from flask import Flask, request\n\napp = Flask(__name__)\n\n" + "@app.route('/u')\n" + "def create():\n" + " conn = sqlite3.connect('app.db')\n" + " conn.execute(\"insert into users(name) values('\" + request.args['name'] + \"')\")\n" + ) + assert rules(findings) == [("sql-injection", 10)] + + +def test_calls_on_what_a_known_function_returns_resolve_through_its_summary() -> None: + findings = check( + "DATABASE = 'app.db'\n\n" + "def get_db():\n return sqlite3.connect(DATABASE)\n\n" + "def register_user(email, password):\n" + " cur = get_db().execute('insert into users(email, password) values(?, ?)', (email, password))\n" + " return cur\n" + ) + assert rules(findings) == [("plaintext-credential-storage", 9)] + + +def test_other_parameters_and_reads_stay_clean() -> None: + assert check("def find_user(email):\n conn = sqlite3.connect('app.db')\n conn.execute('select * from users where email=?', (email,))\n") == () + assert check("def login(email, password):\n conn = sqlite3.connect('app.db')\n return conn.execute('select 1 from users where email=?', (email,))\n") == () + + +def test_shipped_credential_plugins_load() -> None: + module = build_hir(SourceManager().add_source("empty.py", "")) + loaded = {p.manifest.name: p.manifest for p in discover_plugins(PLUGINS, engine.build_manager(module))} + + assert loaded["credential-models"].provides == ("model.credential-sources", "model.credential-sinks") + assert loaded["plaintext-credentials"].provides == ("vulnerability.plaintext-credential-storage",) + + +def test_getters_returning_several_symbols_pick_the_one_a_model_knows() -> None: + # uploadbox's ``get_db``: ``getattr(g, '_database', None)`` or a fresh connection. + findings = check( + "from flask import g\n\nDATABASE = 'app.db'\n\n" + "def get_db():\n" + " db = getattr(g, '_database', None)\n" + " if db is None:\n" + " db = g._database = sqlite3.connect(DATABASE)\n" + " return db\n\n" + "def register_user(email, password):\n" + " cur = get_db().execute('insert into users(email, password) values(?, ?)', (email, password))\n" + " return cur\n" + ) + assert rules(findings) == [("plaintext-credential-storage", 14)] From 28d7421e26cce17fa016d0c57cb8f1683c1e4d03 Mon Sep 17 00:00:00 2001 From: Hugo Date: Fri, 4 Sep 2026 14:03:31 +0200 Subject: [PATCH 2/2] feat: detect credentials reaching database queries unhashed --- README.md | 8 +- .../models/credentials/credential_models.py | 43 +++++++++++ plugins/models/credentials/plugin.toml | 9 +++ plugins/models/django/django_models.py | 12 +-- plugins/models/python_stdlib/python_stdlib.py | 2 +- .../models/sqlalchemy/sqlalchemy_models.py | 2 +- .../plaintext_credentials.py | 21 ++++++ .../plaintext_credentials/plugin.toml | 9 +++ src/coretrace_python/plugins/detectors.py | 7 +- src/coretrace_python/taint/__init__.py | 2 + src/coretrace_python/taint/engine.py | 75 +++++++++++++++---- src/coretrace_python/taint/models.py | 30 +++++++- 12 files changed, 195 insertions(+), 25 deletions(-) create mode 100644 plugins/models/credentials/credential_models.py create mode 100644 plugins/models/credentials/plugin.toml create mode 100644 plugins/security/plaintext_credentials/plaintext_credentials.py create mode 100644 plugins/security/plaintext_credentials/plugin.toml diff --git a/README.md b/README.md index 0c6e3d6..0833c5e 100644 --- a/README.md +++ b/README.md @@ -97,13 +97,17 @@ memory. `--plugins` is a directory searched recursively for `plugin.toml` manifests and may be repeated. The repository ships a first set under `plugins/`: security models for the standard library, Flask, FastAPI, Django, SQLAlchemy and the Requests and httpx clients (`models/`), taint detectors for SQL -injection, command injection, path traversal, SSRF and XSS (`security/`), and syntactic +injection, command injection, path traversal, SSRF, XSS and plaintext credential storage +(`security/`), and syntactic detectors for `eval`/`exec`, weak hashes, `app.run(debug=True)` and HTTP client calls without a timeout (`syntax/`). Route handlers of Flask and FastAPI receive their parameters as HTTP input, whether the application is created directly (`app = Flask(__name__)`) or by a project factory whose summary returns it (`app = create_app()`); `request.args` and its siblings are HTTP -sources. Django views are undecorated, so a parameter annotated with a request class +sources. A parameter named `password` or the like is a credential wherever it appears: stored +in a database without passing through a hashing function it is a +`plaintext-credential-storage` finding at medium confidence, since a name is a hint. +Django views are undecorated, so a parameter annotated with a request class (`request: HttpRequest`), a method of a class-based view and a view decorated by one of Django's or Django REST framework's view decorators receive HTTP input; a bare `request` parameter without any of these is not recognised, and URL configurations are diff --git a/plugins/models/credentials/credential_models.py b/plugins/models/credentials/credential_models.py new file mode 100644 index 0000000..7d7bffc --- /dev/null +++ b/plugins/models/credentials/credential_models.py @@ -0,0 +1,43 @@ +"""Credential models: password-named parameters are credentials, password hashing clears +them, and the database write sinks of the other model plugins also carry the +``CREDENTIAL`` kind, so a credential stored unhashed is a flow.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.plugins import ModelPlugin +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.taint import Model, NamedParameter, Sanitizer, TaintKind + +_HASHERS = ( + "hashlib.pbkdf2_hmac", + "hashlib.scrypt", + "hashlib.sha256", + "hashlib.sha512", + "hashlib.sha3_256", + "hashlib.blake2b", + "hashlib.new", + "bcrypt.hashpw", + "bcrypt.kdf", + "argon2.PasswordHasher.hash", + "argon2.low_level.hash_secret", + "werkzeug.security.generate_password_hash", + "passlib.hash.pbkdf2_sha256.hash", + "passlib.hash.bcrypt.hash", + "passlib.hash.argon2.hash", + "passlib.context.CryptContext.hash", + "django.contrib.auth.hashers.make_password", +) + + +def _sym(path: str) -> SymbolId: + return SymbolId(f"python.{path}") + + +class CredentialModels(ModelPlugin): + name: ClassVar[str] = "credential-models" + models: ClassVar[tuple[Model, ...]] = ( + NamedParameter(r"(?i)^(password|passwd|pwd|new_password|plain_password|secret)$", "credential", TaintKind.CREDENTIAL), + *(Sanitizer(_sym(hasher), TaintKind.CREDENTIAL) for hasher in _HASHERS), + ) diff --git a/plugins/models/credentials/plugin.toml b/plugins/models/credentials/plugin.toml new file mode 100644 index 0000000..8cd3ce3 --- /dev/null +++ b/plugins/models/credentials/plugin.toml @@ -0,0 +1,9 @@ +name = "credential-models" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = [] +provides = ["model.credential-sources", "model.credential-sinks"] + +[entrypoint] +module = "credential_models" +class = "CredentialModels" diff --git a/plugins/models/django/django_models.py b/plugins/models/django/django_models.py index 565049e..b073eae 100644 --- a/plugins/models/django/django_models.py +++ b/plugins/models/django/django_models.py @@ -86,12 +86,12 @@ class DjangoModels(ModelPlugin): *(TypedParameter(_sym(cls), "http") for cls in _REQUEST_CLASSES), *(EntryPoint(_sym(base), "http") for base in _VIEW_BASES), *(EntryPoint(_sym(decorator), "http") for decorator in _VIEW_DECORATORS), - Sink(_sym("django.db.connection.cursor.execute"), TaintKind.SQL), - Sink(_sym("django.db.connection.cursor.executemany"), TaintKind.SQL), - Sink(_sym("django.db.models.expressions.RawSQL"), TaintKind.SQL), - Sink(_sym("django.db.models.Manager.raw"), TaintKind.SQL), - Sink(_sym("django.db.models.QuerySet.raw"), TaintKind.SQL), - Sink(_sym("django.db.models.QuerySet.extra"), TaintKind.SQL), + Sink(_sym("django.db.connection.cursor.execute"), TaintKind.SQL | TaintKind.CREDENTIAL), + Sink(_sym("django.db.connection.cursor.executemany"), TaintKind.SQL | TaintKind.CREDENTIAL), + Sink(_sym("django.db.models.expressions.RawSQL"), TaintKind.SQL | TaintKind.CREDENTIAL), + Sink(_sym("django.db.models.Manager.raw"), TaintKind.SQL | TaintKind.CREDENTIAL), + Sink(_sym("django.db.models.QuerySet.raw"), TaintKind.SQL | TaintKind.CREDENTIAL), + Sink(_sym("django.db.models.QuerySet.extra"), TaintKind.SQL | TaintKind.CREDENTIAL), Sink(_sym("django.utils.safestring.mark_safe"), TaintKind.HTML), Sink(_sym("django.http.HttpResponse"), TaintKind.HTML), Sink(_sym("django.http.response.HttpResponse"), TaintKind.HTML), diff --git a/plugins/models/python_stdlib/python_stdlib.py b/plugins/models/python_stdlib/python_stdlib.py index 2352c0f..92f682e 100644 --- a/plugins/models/python_stdlib/python_stdlib.py +++ b/plugins/models/python_stdlib/python_stdlib.py @@ -40,7 +40,7 @@ class PythonStdlibModels(ModelPlugin): Sink(_sym("shutil.rmtree"), TaintKind.PATH), Sink(_sym("urllib.request.urlopen"), TaintKind.SSRF), *( - Sink(_sym(f"sqlite3.connect{cursor}.{method}"), TaintKind.SQL) + Sink(_sym(f"sqlite3.connect{cursor}.{method}"), TaintKind.SQL | TaintKind.CREDENTIAL) for cursor in ("", ".cursor") for method in ("execute", "executemany", "executescript") ), diff --git a/plugins/models/sqlalchemy/sqlalchemy_models.py b/plugins/models/sqlalchemy/sqlalchemy_models.py index 07a426a..2a3d8d0 100644 --- a/plugins/models/sqlalchemy/sqlalchemy_models.py +++ b/plugins/models/sqlalchemy/sqlalchemy_models.py @@ -32,7 +32,7 @@ def _sym(path: str) -> SymbolId: class SqlAlchemyModels(ModelPlugin): name: ClassVar[str] = "sqlalchemy-models" models: ClassVar[tuple[Model, ...]] = tuple( - Sink(_sym(f"{executor}.{method}"), TaintKind.SQL) + Sink(_sym(f"{executor}.{method}"), TaintKind.SQL | TaintKind.CREDENTIAL) for executor in _EXECUTORS for method in ("execute", "exec_driver_sql", "scalar", "scalars") ) diff --git a/plugins/security/plaintext_credentials/plaintext_credentials.py b/plugins/security/plaintext_credentials/plaintext_credentials.py new file mode 100644 index 0000000..885e2f0 --- /dev/null +++ b/plugins/security/plaintext_credentials/plaintext_credentials.py @@ -0,0 +1,21 @@ +"""Plaintext credentials in database queries: a password-named parameter reaching a +database statement without passing through a hashing function, whether the statement +stores it or compares it, since comparing in clear implies storing in clear. A name is +a hint, so medium confidence.""" + +from __future__ import annotations + +from typing import ClassVar + +from coretrace_python.findings import Confidence, Severity +from coretrace_python.plugins import TaintDetector +from coretrace_python.taint import TaintKind + + +class PlaintextCredentialsPlugin(TaintDetector): + name: ClassVar[str] = "plaintext-credentials" + rule_id: ClassVar[str] = "plaintext-credential-storage" + kind: ClassVar[TaintKind] = TaintKind.CREDENTIAL + severity: ClassVar[Severity] = Severity.HIGH + confidence: ClassVar[Confidence] = Confidence.MEDIUM + title: ClassVar[str] = "Plaintext credential in a database query" diff --git a/plugins/security/plaintext_credentials/plugin.toml b/plugins/security/plaintext_credentials/plugin.toml new file mode 100644 index 0000000..14945d9 --- /dev/null +++ b/plugins/security/plaintext_credentials/plugin.toml @@ -0,0 +1,9 @@ +name = "plaintext-credentials" +version = "1.0.0" +plugin_api = ">=1,<2" +requires = ["taint.flows", "findings.refutation"] +provides = ["vulnerability.plaintext-credential-storage"] + +[entrypoint] +module = "plaintext_credentials" +class = "PlaintextCredentialsPlugin" diff --git a/src/coretrace_python/plugins/detectors.py b/src/coretrace_python/plugins/detectors.py index e3b9b5f..987a17b 100644 --- a/src/coretrace_python/plugins/detectors.py +++ b/src/coretrace_python/plugins/detectors.py @@ -29,6 +29,7 @@ class TaintDetector(Plugin): kind: ClassVar[TaintKind] severity: ClassVar[Severity] title: ClassVar[str] + confidence: ClassVar[Confidence] = Confidence.HIGH def analyze(self, ctx: PluginContext) -> Sequence[Finding]: findings: list[Finding] = [] @@ -53,7 +54,7 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: metadata["through"] = flow.through metadata["sink_line"] = str(flow.sink_location.start_line) confidence = ( - Confidence.HIGH if verdict.status is Status.VULNERABILITY else Confidence.MEDIUM + self.confidence if verdict.status is Status.VULNERABILITY else _lower(self.confidence) ) findings.append( Finding( @@ -69,6 +70,10 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: return findings +def _lower(confidence: Confidence) -> Confidence: + return {Confidence.HIGH: Confidence.MEDIUM}.get(confidence, Confidence.LOW) + + class SymbolCallDetector(Plugin): """Report calls whose callee is one of ``symbols``, whatever name the file uses.""" diff --git a/src/coretrace_python/taint/__init__.py b/src/coretrace_python/taint/__init__.py index f5d2505..9970527 100644 --- a/src/coretrace_python/taint/__init__.py +++ b/src/coretrace_python/taint/__init__.py @@ -13,6 +13,7 @@ Model, ModelError, ModelTable, + NamedParameter, Sanitizer, SecurityModelAnalysis, SecurityModelRegistry, @@ -29,6 +30,7 @@ "Model", "ModelError", "ModelTable", + "NamedParameter", "Sanitizer", "SecurityModelAnalysis", "SecurityModelRegistry", diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index 494b381..076b189 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -279,24 +279,64 @@ def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> T if project is not None: through = target.symbol.canonical_name.removeprefix("python.") return self.known(project, through, arguments, keywords, everything, call, flows, state) - sink = self.models.sink(target.symbol) - if sink is not None: - for argument in call.argument_values(): - self.report(flows, sink, self.deep(argument, state), argument, call, None, None) - sanitizer = self.models.sanitizer(target.symbol) - if sanitizer is not None: - return everything.without(sanitizer.kinds) - # A method on a tainted object returns tainted data (``request.args.get``). - everything = everything.join(state.get(call.callee, Taint.none())) - source = self.models.source_covering(target.symbol) - if source is not None: - return everything.join(Taint(source.kinds, frozenset({source}))) - return everything + return self.external(target.symbol, everything, call, state, flows) if isinstance(target, KnownFunction): summary = self.summaries.summary(target.name) return self.known(summary, target.name, arguments, keywords, everything, call, flows, state) + returned = self.returned_symbol(call) + if returned is not None: + return self.external(returned, everything, call, state, flows) return everything.join(state.get(call.callee, Taint.none())) + def external( + self, symbol: SymbolId, everything: Taint, call: Call, state: Mapping[Key, Taint], flows: list[TaintFlow] + ) -> Taint: + """Sinks, sanitizers and sources of a call to an external symbol.""" + + sink = self.models.sink(symbol) + if sink is not None: + for argument in call.argument_values(): + self.report(flows, sink, self.deep(argument, state), argument, call, None, None) + sanitizer = self.models.sanitizer(symbol) + if sanitizer is not None: + return everything.without(sanitizer.kinds) + # A method on a tainted object returns tainted data (``request.args.get``). + everything = everything.join(state.get(call.callee, Taint.none())) + source = self.models.source_covering(symbol) + if source is not None: + return everything.join(Taint(source.kinds, frozenset({source}))) + return everything + + def returned_symbol(self, call: Call) -> SymbolId | None: + """``get_db().execute(...)``: the method of what a known function returns, when + its summary says the return value is one external symbol.""" + + callee = self.defs.get(call.callee) + if not isinstance(callee, GetAttr): + return None + origin = self.defs.get(callee.object) + if not isinstance(origin, Call): + return None + target = self.graph.target_at(self.name, origin.location) + summary: FunctionSummary | None = None + if isinstance(target, KnownFunction): + summary = self.summaries.summary(target.name) + elif isinstance(target, ExternalSymbol): + summary = self.project.summary(target.symbol) + if summary is None: + return None + # ``getattr(g, "_database", None) or sqlite3.connect(...)``: among the symbols the + # function may return, the one the models know about is the one that matters. + for returned in sorted(summary.return_externals, key=str): + candidate = returned.attribute(callee.attribute) + if ( + self.models.sink(candidate) is not None + or self.models.sanitizer(candidate) is not None + or self.models.source_covering(candidate) is not None + ): + return candidate + return None + def known( self, summary: FunctionSummary, @@ -488,6 +528,15 @@ def parameter_sources( typed = models.typed_parameter(symbol) if symbol is not None else None if typed is not None: sources[index] = Source(typed.symbol, typed.label, typed.kinds) + for index, parameter in enumerate(function.parameters): + if index in sources: + continue + for named in models.named_parameters: + if named.matches(parameter.name): + sources[index] = Source( + SymbolId(f"python.parameter.{parameter.name}"), named.label, named.kinds + ) + break return sources diff --git a/src/coretrace_python/taint/models.py b/src/coretrace_python/taint/models.py index 9ccbf31..35e8533 100644 --- a/src/coretrace_python/taint/models.py +++ b/src/coretrace_python/taint/models.py @@ -8,6 +8,8 @@ from __future__ import annotations +import hashlib +import re from dataclasses import dataclass, field from enum import Flag, auto from types import MappingProxyType @@ -27,6 +29,9 @@ class TaintKind(Flag): CODE = auto() ADVISORY = auto() ALL = SQL | COMMAND | HTML | PATH | SSRF | CODE | ADVISORY + # Outside ALL on purpose: only credential-named parameters carry it, so a database + # write reached by ordinary input is not a plaintext credential. + CREDENTIAL = auto() class ModelError(Exception): @@ -77,6 +82,24 @@ class TypedParameter: kinds: TaintKind = TaintKind.ALL +@dataclass(frozen=True) +class NamedParameter: + """Parameters whose name matches ``pattern`` carry ``kinds`` (``password`` is a + credential wherever it is a parameter).""" + + pattern: str + label: str + kinds: TaintKind + + @property + def symbol(self) -> SymbolId: + digest = hashlib.sha1(self.pattern.encode("utf-8")).hexdigest()[:12] + return SymbolId(f"python.parameter.p{digest}") + + def matches(self, name: str) -> bool: + return re.search(self.pattern, name) is not None + + @dataclass(frozen=True) class Validator: """A callable whose truth proves its ``argument`` safe (refutation evidence, ยง24).""" @@ -95,7 +118,9 @@ class AuthorizationGuard: label: str -Model = Source | Sink | Sanitizer | EntryPoint | TypedParameter | Validator | AuthorizationGuard +Model = ( + Source | Sink | Sanitizer | EntryPoint | TypedParameter | Validator | AuthorizationGuard | NamedParameter +) @dataclass(frozen=True) @@ -107,6 +132,7 @@ class ModelTable: typed_parameters: tuple[TypedParameter, ...] = () validators: tuple[Validator, ...] = () authorizations: tuple[AuthorizationGuard, ...] = () + named_parameters: tuple[NamedParameter, ...] = () _by_symbol: dict[type[Model], dict[SymbolId, Model]] = field( init=False, repr=False, compare=False ) @@ -175,6 +201,7 @@ def extended(self, *sinks: Sink) -> ModelTable: self.typed_parameters, self.validators, self.authorizations, + self.named_parameters, ) def sanitizer(self, symbol: SymbolId) -> Sanitizer | None: @@ -207,6 +234,7 @@ def freeze(self) -> ModelTable: typed_parameters=tuple(m for m in models if isinstance(m, TypedParameter)), validators=tuple(m for m in models if isinstance(m, Validator)), authorizations=tuple(m for m in models if isinstance(m, AuthorizationGuard)), + named_parameters=tuple(m for m in models if isinstance(m, NamedParameter)), )