Add better support for Hardcoded Secret query

python/github/HardcodedSecretSinks.qll

@@ -44,7 +44,7 @@ class FlaskCredentialSink extends CredentialSink {
         // app = flask.Flask(__name__)
         // app.secret_key = VALUE
         node = Flask::FlaskApp::instance().getMember("secret_key") and
-        stmt = node.getAValueReachableFromSource().asExpr().getParentNode() and
+        stmt = node.getAValueReachingSink().asExpr().getParentNode() and
         this = DataFlow::exprNode(stmt.getValue())
       )
       or
@@ -62,20 +62,45 @@ class FlaskCredentialSink extends CredentialSink {
   }
 }
 
-// TODO: Django support
+class DjangoCredentialSink extends CredentialSink {
+  DjangoCredentialSink() {
+    // Check Django import is present
+    exists(API::moduleImport("django")) and
+    exists(AssignStmt stmt |
+      // Check is the SECRET_KEY is in the a settings.py file
+      // Removed "settings/develop.py"
+      stmt.getLocation().getFile().getBaseName() = ["settings.py", "settings/production.py"] and
+      (
+        stmt.getATarget().toString() = "SECRET_KEY" and
+        this.asExpr() = stmt.getValue()
+      )
+    )
+  }
+}
+
 // =========================
 // Databases
 // =========================
 class MySqlSink extends CredentialSink {
   MySqlSink() {
     this =
-      API::moduleImport("mysql.connector").getMember("connect").getACall().getArgByName("password")
+      API::moduleImport("mysql")
+          .getMember("connector")
+          .getMember("connect")
+          .getACall()
+          .getArgByName("password")
   }
 }
 
 class AsyncpgSink extends CredentialSink {
   AsyncpgSink() {
-    this = API::moduleImport("asyncpg").getMember("connect").getACall().getArgByName("password")
+    this = API::moduleImport("asyncpg").getMember("connect").getACall().getArgByName("password") or
+    this =
+      API::moduleImport("asyncpg")
+          .getMember("connection")
+          .getMember("Connection")
+          .getACall()
+          .getArgByName("password")
   }
 }
 
@@ -108,13 +133,15 @@ class AioredisSink extends CredentialSink {
           .getArgByName("password")
     or
     this =
-      API::moduleImport("aioredis.sentinel")
+      API::moduleImport("aioredis")
+          .getMember("sentinel")
           .getMember("create_sentinel")
           .getACall()
           .getArgByName("password")
     or
     this =
-      API::moduleImport("aioredis.sentinel")
+      API::moduleImport("aioredis")
+          .getMember("sentinel")
           .getMember("create_sentinel_pool")
           .getACall()
           .getArgByName("password")
@@ -128,7 +155,12 @@ class RequestsSink extends CredentialSink {
   RequestsSink() {
     // from requests.auth import HTTPBasicAuth
     // auth = HTTPBasicAuth('user', 'mysecretpassword')
-    this = API::moduleImport("requests.auth").getMember("HTTPBasicAuth").getACall().getArg(1)
+    this =
+      API::moduleImport("requests")
+          .getMember("auth")
+          .getMember("HTTPBasicAuth")
+          .getACall()
+          .getArg(1)
   }
 }
 
@@ -147,7 +179,7 @@ class PyOtpSink extends CredentialSink {
   PyOtpSink() {
     // import pyotp
     // totp = pyotp.TOTP('base32secret3232')
-    this = API::moduleImport("pyotp").getMember("TOTP").getACall().getArg(1)
+    this = API::moduleImport("pyotp").getMember("TOTP").getACall().getArg(0)
   }
 }
 

tests/python-tests/CWE-798/HardcodedFrameworkSecrets.expected

@@ -0,0 +1,21 @@
+| hardcoded_secrets.py:10:18:10:29 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:11:28:11:39 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:12:30:12:41 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:22:30:22:47 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:28:38:28:55 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:34:53:34:69 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:35:50:35:66 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:40:47:40:59 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:41:38:41:50 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:47:72:47:85 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:50:72:50:72 | ControlFlowNode for w | sinks |
+| hardcoded_secrets.py:56:19:56:36 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:59:20:59:20 | ControlFlowNode for p | sinks |
+| hardcoded_secrets.py:62:20:62:20 | ControlFlowNode for p | sinks |
+| hardcoded_secrets.py:70:23:70:40 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:71:27:71:48 | ControlFlowNode for Str | sinks |
+| hardcoded_secrets.py:72:23:72:42 | ControlFlowNode for Str | sinks |
+| settings.py:5:14:5:29 | ControlFlowNode for Str | sinks |
+| settings.py:7:14:7:51 | ControlFlowNode for Attribute() | sinks |
+| settings.py:9:14:9:41 | ControlFlowNode for Attribute() | sinks |
+| settings.py:13:14:13:26 | ControlFlowNode for RANDOM_STRING | sinks |

tests/python-tests/CWE-798/HardcodedFrameworkSecrets.ql

@@ -0,0 +1,6 @@
+
+import python
+import github.HardcodedSecretSinks
+
+from CredentialSink sinks
+select sinks, "sinks" 

tests/python-tests/CWE-798/hardcoded_secrets.py

@@ -0,0 +1,73 @@
+import os
+
+password = os.environ.get("SECRET_TOKEN")
+
+# Flask
+from flask import Flask
+
+app = Flask(__name__)
+
+app.secret_key = "SecretKey1"
+app.config["SECRET_KEY"] = "SecretKey2"
+app.config.update(SECRET_KEY="SecretKey3")
+
+
+# Django
+SECRET_KEY = "SuperSecretKey"  # False Positive, not a settings file
+
+
+# Requests
+from requests.auth import HTTPBasicAuth
+
+auth = HTTPBasicAuth("user", "mysecretpassword")
+
+
+# MySQL
+from mysql.connector import connect
+
+conn = connect(user="user", password="mysecretpassword")
+
+# Asyncpg
+from asyncpg import connect
+from asyncpg.connection import Connection
+
+asyncpg_conn1 = await connect(user="user", password="asyncpg_secret1")
+asyncpg_conn2 = Connection(user="user", password="asyncpg_secret2")
+
+# JWT
+import jwt
+
+jwt_encoded = jwt.encode({"some": "payload"}, "jwt_secret1", algorithm="HS256")
+jwt_decode = jwt.decode(jwt_encoded, "jwt_secret2", algorithm="HS256")
+
+
+# Redis
+import aioredis
+
+redis = await aioredis.create_redis_pool("redis://localhost", password="ReDiSsEcRet1")
+
+w = "ReDiSsEcRet2"
+redis = await aioredis.create_redis_pool("redis://localhost", password=w)
+
+
+# PyOtp
+import pyotp
+
+totp = pyotp.TOTP("base32secret3232")
+
+p = "base32secret3232"
+totp2 = pyotp.TOTP(p)
+
+p = os.environ.get("OPT_KEY")
+totp2 = pyotp.TOTP(p)
+
+
+# Bota3
+import boto3
+
+s3 = boto3.resource(
+    "s3",
+    aws_access_key_id="YOUR-ACCESSKEYID",
+    aws_secret_access_key="YOUR-SECRETACCESSKEY",
+    aws_session_token="YOUR-SESSION-TOKEN",
+)

tests/python-tests/CWE-798/options

@@ -0,0 +1 @@
+semmle-extractor-options: --max-import-depth=0

tests/python-tests/CWE-798/settings.py

@@ -0,0 +1,13 @@
+import os
+import django
+
+# const key
+SECRET_KEY = "SuperSecretKey"
+# const default key
+SECRET_KEY = os.environ.get("SECRET_KEY", "secret")
+# False Positive, key from env
+SECRET_KEY = os.environ.get("SECRET_KEY")
+
+
+RANDOM_STRING = "SuperRandomString"
+SECRET_KEY = RANDOM_STRING

tests/python-tests/codeql-pack.lock.yml

@@ -0,0 +1,12 @@
+---
+lockVersion: 1.0.0
+dependencies:
+  advanced-security/codeql-python:
+    version: 0.2.0
+  codeql/python-all:
+    version: 0.8.0
+  codeql/regex:
+    version: 0.0.7
+  codeql/tutorial:
+    version: 0.0.4
+compiled: false