feat(detection): add SSRF (CWE-918) rule for Python (#86)

signetry_core/pipeline/findings/deterministic.py

@@ -271,10 +271,28 @@ def visit_Call(self, node: ast.Call) -> None:  # noqa: C901 - dispatch table by
                           0.85, "CWE-22")
 
         # --- SSRF: outbound request to a tainted URL ---
-        if target in ("requests.get", "requests.post", "requests.request", "requests.put",
-                      "requests.delete", "requests.head", "urllib.request.urlopen",
-                      "httpx.get", "httpx.post", "aiohttp.request") and args:
-            if self._is_tainted(args[0]):
+        ssrf_targets = (
+            "requests.get", "requests.post", "requests.put", "requests.patch",
+            "requests.delete", "requests.head", "requests.options", "requests.request",
+            "httpx.get", "httpx.post", "httpx.put", "httpx.patch",
+            "httpx.delete", "httpx.head", "httpx.options", "httpx.request",
+            "urllib.request.urlopen", "urllib.request.Request",
+            "aiohttp.request",
+        )
+        if target in ssrf_targets:
+            url_nodes: list[ast.AST] = []
+            for kw in node.keywords:
+                if kw.arg == "url":
+                    url_nodes.append(kw.value)
+            if target.endswith(".request"):
+                if len(args) >= 2:
+                    url_nodes.append(args[1])
+                elif len(args) == 1 and not url_nodes:
+                    url_nodes.append(args[0])
+            elif args:
+                url_nodes.append(args[0])
+
+            if any(self._is_tainted(u) for u in url_nodes):
                 self._add(node, "py.ssrf", "ssrf", Severity.HIGH,
                           "Server-side request to a user-controlled URL (SSRF)",
                           "An outbound HTTP request targets a user-controlled URL, allowing SSRF "

tests/test_findings_engine.py

@@ -263,6 +263,71 @@ def test_detects_ssrf():
     assert "ssrf" in _cats(src)
 
 
+def test_detects_ssrf_keyword_and_positional_variants():
+    # Keyword url= in requests.get
+    src_kw = (
+        "import requests, flask\n"
+        "def f():\n"
+        "    url = flask.request.args.get('u')\n"
+        "    return requests.get(url=url).text\n"
+    )
+    assert "ssrf" in _cats(src_kw)
+
+    # requests.request with method and url keyword
+    src_req_kw = (
+        "import requests, flask\n"
+        "def f():\n"
+        "    url = flask.request.args.get('u')\n"
+        "    return requests.request('GET', url=url).text\n"
+    )
+    assert "ssrf" in _cats(src_req_kw)
+
+    # requests.request with dynamic method and positional url
+    src_req_pos = (
+        "import requests, flask\n"
+        "def f(method_var):\n"
+        "    url = flask.request.args.get('u')\n"
+        "    return requests.request(method_var, url).text\n"
+    )
+    assert "ssrf" in _cats(src_req_pos)
+
+    # aiohttp.request with positional url
+    src_aiohttp = (
+        "import aiohttp, flask\n"
+        "async def f():\n"
+        "    target = flask.request.args.get('t')\n"
+        "    return await aiohttp.request('GET', target)\n"
+    )
+    assert "ssrf" in _cats(src_aiohttp)
+
+    # urllib.request.Request with tainted input
+    src_urllib_req = (
+        "import urllib.request, flask\n"
+        "def f():\n"
+        "    dest = flask.request.args.get('dest')\n"
+        "    req = urllib.request.Request(dest)\n"
+        "    return urllib.request.urlopen(req).read()\n"
+    )
+    assert "ssrf" in _cats(src_urllib_req)
+
+
+def test_no_false_positive_on_constant_url_with_tainted_kwargs():
+    src_clean_kwargs = (
+        "import requests, flask\n"
+        "def handler():\n"
+        "    user_q = flask.request.args.get('q')\n"
+        "    user_body = flask.request.json\n"
+        "    user_form = flask.request.form\n"
+        "    tok = flask.request.headers.get('Authorization')\n"
+        "    r1 = requests.get('https://api.example.com/search', params={'q': user_q})\n"
+        "    r2 = requests.post('https://api.example.com/v1/items', json=user_body)\n"
+        "    r3 = requests.post('https://api.example.com/submit', data=user_form)\n"
+        "    r4 = requests.get('https://api.example.com/me', headers={'Authorization': tok})\n"
+        "    return r1.text\n"
+    )
+    assert "ssrf" not in _cats(src_clean_kwargs)
+
+
 def test_detects_ssti():
     src = (
         "import flask\n"