security(scanner): detect stored SSRF webhook persistence

appguardrail_core/rules.py

@@ -21,16 +21,25 @@
         "OWASP A07:2021 - Identification and Authentication Failures",
         "CWE-798 - Use of Hard-coded Credentials",
     ),
+    "ssrf": (
+        "OWASP A10:2021 - Server-Side Request Forgery",
+        "CWE-918 - Server-Side Request Forgery",
+    ),
     "storage": ("OWASP A01:2021 - Broken Access Control",),
 }
 
+REFERENCE_CATEGORY_OVERRIDES = {
+    "CWE-918": "ssrf",
+}
+
 SAMM_BY_CATEGORY = {
     "authz": "Implementation / Secure Build",
     "dependency": "Implementation / Secure Build",
     "injection": "Implementation / Secure Build",
     "misconfig": "Operations / Environment Management",
     "payment": "Verification / Requirements-driven Testing",
     "secrets": "Operations / Environment Management",
+    "ssrf": "Implementation / Secure Build",
     "storage": "Implementation / Secure Build",
 }
 
@@ -59,6 +68,10 @@
         "Remove the secret from source, rotate it, and load future values from "
         "managed secret storage."
     ),
+    "ssrf": (
+        "Validate untrusted URLs before persistence, reject non-public destinations, "
+        "and revalidate redirects or pin the outbound destination before delivery."
+    ),
     "storage": "Enforce storage or database access controls with authenticated ownership policies.",
 }
 
@@ -101,6 +114,15 @@ def extract_public_references(message: str) -> tuple[str, ...]:
     )
 
 
+def _category_for_references(references: tuple[str, ...], fallback: str) -> str:
+    """Prefer an authoritative public taxonomy over a rule-id heuristic."""
+    for reference in references:
+        for prefix, category in REFERENCE_CATEGORY_OVERRIDES.items():
+            if reference.startswith(prefix):
+                return category
+    return fallback
+
+
 def build_rule_metadata(
     rule_id: str,
     severity: str,
@@ -110,8 +132,10 @@ def build_rule_metadata(
     source: str = "appguardrail-rule",
 ) -> RuleMetadata:
     """Build a stable metadata envelope for a scanner finding."""
+    public_references = extract_public_references(message)
+    category = _category_for_references(public_references, category)
     references = _merge_references(
-        extract_public_references(message),
+        public_references,
         CATEGORY_REFERENCE_DEFAULTS.get(category, ()),
     )
     return RuleMetadata(

scanner/cli/appguardrail.py

@@ -969,6 +969,7 @@ def finish_rule():
                 "message_lines": [],
                 "include_paths": [],
                 "exclude_paths": [],
+                "required_substrings": [],
                 "severity": "WARNING",
             }
             in_message = False
@@ -1001,6 +1002,12 @@ def finish_rule():
             current["languages"] = _parse_inline_list(raw_line.split(":", 1)[1])
             path_mode = None
             continue
+        if raw_line.startswith("    prefilter: "):
+            current["required_substrings"] = _parse_inline_list(
+                raw_line.split(":", 1)[1]
+            )
+            path_mode = None
+            continue
         if raw_line.startswith("      - pattern-regex: "):
             current["regexes"].append(
                 _unquote_rule_scalar(raw_line.split("pattern-regex:", 1)[1])
@@ -1038,6 +1045,9 @@ def _compile_yaml_regex_rule(rule):
                 "extensions": extensions,
                 "include_paths": rule.get("include_paths") or [],
                 "exclude_paths": rule.get("exclude_paths") or [],
+                "required_substrings": tuple(
+                    rule.get("required_substrings") or ()
+                ),
             }
         )
     return compiled_rules
@@ -2148,6 +2158,7 @@ def _get_applicable_rules(ext: str):
                 rule["pattern"].finditer,
                 tuple(rule.get("include_paths") or ()),
                 tuple(rule.get("exclude_paths") or ()),
+                tuple(rule.get("required_substrings") or ()),
             )
             for rule in SCAN_RULES
             if not rule["extensions"] or ext in rule["extensions"]
@@ -2962,7 +2973,12 @@ def _scan_file(
                 finditer,
                 include_paths,
                 exclude_paths,
+                required_substrings,
             ) in applicable_rules:
+                if required_substrings and not all(
+                    substring in content for substring in required_substrings
+                ):
+                    continue
                 if include_paths or exclude_paths:
                     if rel_path_for_filters is None:
                         rel_path_for_filters = _display_path(

scanner/rules/ssrf.yml

@@ -0,0 +1,14 @@
+rules:
+  - id: python-stored-ssrf-webhook-url
+    patterns:
+      - pattern-regex: '(?is)(?:\b(?P<unguarded_var>[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:\([^\)\n]*\)|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)(?:\.get\s*\(\s*["\x27]url["\x27]\s*\)|\[\s*["\x27]url["\x27]\s*\])(?:(?!\bset_webhook\b).){0,800}?(?m:^(?P<unguarded_indent>[ \t]*)if\s+(?:(?!\bnot\b)[^\n:]){0,400}_is_safe_url\s*\(\s*(?P=unguarded_var)\s*\)[^\n:]{0,100}:\s*(?:#[^\n]*)?\n(?:(?P=unguarded_indent)[ \t]+[^\n]*\n){0,20}?(?P=unguarded_indent)[ \t]+\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=unguarded_var)\s*\))(?:(?!\bset_webhook\b).){0,800}?(?m:^(?P=unguarded_indent)\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=unguarded_var)\s*\))|\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?:\([^\)\n]*\)|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)(?:\.get\s*\(\s*["\x27]url["\x27]\s*\)|\[\s*["\x27]url["\x27]\s*\])|\b(?P<webhook_url_var>[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:\([^\)\n]*\)|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*)(?:\.get\s*\(\s*["\x27]url["\x27]\s*\)|\[\s*["\x27]url["\x27]\s*\])(?!(?:(?!\bset_webhook\b).){0,800}(?m:^(?P<reject_indent>[ \t]*)if\s+(?:(?P=webhook_url_var)\s+is\s+not\s+None\s+and\s+not\s+_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)|(?:(?P=webhook_url_var)\s+and\s+)?not\s+_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)|(?P=webhook_url_var)\s+not\s+in\s*\(\s*None\s*,\s*["\x27]["\x27]\s*\)\s+and\s*\(\s*not\s+isinstance\s*\(\s*(?P=webhook_url_var)\s*,\s*str\s*\)\s+or\s+not\s+_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)\s*\))(?:(?!:).){0,100}:\s*(?:#[^\n]*)?\n(?:(?P=reject_indent)[ \t]+[^\n]*\n){0,20}?(?P=reject_indent)[ \t]+\b(?:return|raise)\b))(?!(?:(?!\bset_webhook\b).){0,800}(?m:^(?P<guard_indent>[ \t]*)if\s+(?:(?!\bnot\b)[^\n:]){0,400}_is_safe_url\s*\(\s*(?P=webhook_url_var)\s*\)[^\n:]{0,100}:\s*(?:#[^\n]*)?\n(?:(?P=guard_indent)[ \t]+[^\n]*\n){0,20}?(?P=guard_indent)[ \t]+\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=webhook_url_var)\s*\)))(?:(?!\bset_webhook\b).){0,800}?\bset_webhook\s*\(\s*[^,\n]+,\s*[^,\n]+,\s*(?P=webhook_url_var)\s*\))'
+    message: |
+      A webhook URL is persisted directly from request data without an explicit
+      validation boundary. Validate the URL with a fail-closed SSRF policy before
+      storing it, and revalidate or pin the destination before outbound delivery.
+      [CWE-918 - Server-Side Request Forgery]
+    severity: HIGH
+    languages: [python]
+    prefilter: [set_webhook]
+    cwe: [CWE-918]
+    owasp: [A10:2021]

tests/test_ssrf_rule_accessors.py

@@ -0,0 +1,77 @@
+"""Regression tests for equivalent stored-SSRF request accessors."""
+
+import pytest
+
+from scanner.cli.appguardrail import SCAN_RULES, _scan_file
+
+_RULE_ID = "python-stored-ssrf-webhook-url"
+
+
+def _rule():
+    """Return the packaged stored-SSRF rule under test."""
+    matches = [rule for rule in SCAN_RULES if rule["id"] == _RULE_ID]
+    assert len(matches) == 1
+    return matches[0]
+
+
+def _source(accessor, *, direct=False, validated=False):
+    """Build a direct or one-hop request URL persistence flow."""
+    sink = "set_" + "webhook"
+    if direct:
+        return "\n".join(
+            [
+                "def update_webhook(conn, org, body):",
+                f"    {sink}(conn, org, {accessor})",
+                "",
+            ]
+        )
+
+    lines = [
+        "def update_webhook(conn, org, body):",
+        f"    target = {accessor}",
+    ]
+    if validated:
+        lines.extend(
+            [
+                "    if not _is_safe_url(target):",
+                "        return",
+            ]
+        )
+    lines.extend([f"    {sink}(conn, org, target)", ""])
+    return "\n".join(lines)
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        _source('body["url"]', direct=True),
+        _source('request.json["url"]'),
+        _source('request.json.get("url")'),
+    ],
+)
+def test_packaged_rule_matches_equivalent_request_url_accessors(source):
+    """Detect direct, subscript, and attribute-based URL sources."""
+    assert _rule()["pattern"].search(source)
+
+
+def test_scan_file_emits_finding_for_subscript_variable_flow(tmp_path):
+    """Emit the stored-SSRF finding for a subscript one-hop flow."""
+    source_file = tmp_path / "webhook.py"
+    source_file.write_text(_source('body["url"]'), encoding="utf-8")
+
+    findings = [
+        finding
+        for finding in _scan_file(source_file, tmp_path)
+        if finding["rule_id"] == _RULE_ID
+    ]
+
+    assert len(findings) == 1
+    assert findings[0]["line"] == 2
+    assert findings[0]["category"] == "ssrf"
+
+
+def test_packaged_rule_ignores_validated_subscript_flow():
+    """Do not flag a subscript source protected by a fail-closed guard."""
+    assert not _rule()["pattern"].search(
+        _source('body["url"]', validated=True)
+    )