feat: plaintext credentials

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

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),
+    )

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"

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),

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")
         ),

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")
     )

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"

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"

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."""
 

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",

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
 
 

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)),
         )
 
 

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",

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",