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/rules/ssrf.yml

@@ -0,0 +1,13 @@
+rules:
+  - id: python-stored-ssrf-webhook-url
+    patterns:
+      - pattern-regex: '(?is)(?:\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\b(?:(?!:).){0,600}\bnot\s+_is_safe_url\s*\(\s*(?P=webhook_url_var)\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]
+    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)
+    )

tests/test_ssrf_rules.py

@@ -0,0 +1,235 @@
+"""Regression tests for stored SSRF detection in the packaged rule engine."""
+
+from scanner.cli.appguardrail import SCAN_RULES, _scan_file
+
+_RULE_ID = "python-stored-ssrf-webhook-url"
+
+
+def _rule():
+    """Return the single packaged stored-SSRF rule under test."""
+    matches = [rule for rule in SCAN_RULES if rule["id"] == _RULE_ID]
+    assert len(matches) == 1, f"expected one loaded rule for {_RULE_ID}"
+    return matches[0]
+
+
+def _vulnerable_source():
+    """Build the original direct request-to-persistence vulnerability."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body):",
+            f'    {sink}(conn, org, (body or {{}}).get("url"))',
+            "",
+        ]
+    )
+
+
+def _unvalidated_variable_source(variable="webhook_url"):
+    """Build an unvalidated local-variable flow into webhook persistence."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body):",
+            f'    {variable} = (body or {{}}).get("url")',
+            f"    {sink}(conn, org, {variable})",
+            "",
+        ]
+    )
+
+
+def _ignored_validation_result_source():
+    """Build a flow that calls the validator but discards its result."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body):",
+            '    target = (body or {}).get("url")',
+            "    _is_safe_url(target)",
+            f"    {sink}(conn, org, target)",
+            "",
+        ]
+    )
+
+
+def _non_enforcing_guard_source():
+    """Build a guard that logs invalid input without blocking persistence."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body):",
+            '    target = (body or {}).get("url")',
+            "    if not _is_safe_url(target):",
+            '        log.warning("unsafe webhook url")',
+            f"    {sink}(conn, org, target)",
+            "",
+        ]
+    )
+
+
+def _non_enforcing_guard_with_unrelated_return_source():
+    """Build a non-enforcing guard followed by an unrelated early return."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body, disabled):",
+            '    target = (body or {}).get("url")',
+            "    if not _is_safe_url(target):",
+            '        log.warning("unsafe webhook url")',
+            "    if disabled:",
+            "        return",
+            f"    {sink}(conn, org, target)",
+            "",
+        ]
+    )
+
+
+def _positive_guard_source():
+    """Build a safe flow whose persistence sink is inside a positive guard."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body):",
+            '    target = (body or {}).get("url")',
+            "    if _is_safe_url(target):",
+            f"        {sink}(conn, org, target)",
+            "",
+        ]
+    )
+
+
+def _safe_source():
+    """Build a safe flow that raises before persisting invalid input."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body):",
+            '    webhook_url = (body or {}).get("url")',
+            "    if webhook_url and not _is_safe_url(webhook_url):",
+            "        raise ValueError(\"unsafe webhook url\")",
+            f"    {sink}(conn, org, webhook_url)",
+            "",
+        ]
+    )
+
+
+def _production_guard_source():
+    """Build the multiline fail-closed guard used by the control plane."""
+    sink = "set_" + "webhook"
+    return "\n".join(
+        [
+            "def update_webhook(conn, org, body):",
+            '    webhook_url = body.get("url")',
+            '    if webhook_url not in (None, "") and (',
+            "        not isinstance(webhook_url, str)",
+            "        or not _is_safe_url(webhook_url)",
+            "    ):",
+            '        return {"error": "unsafe webhook url"}',
+            f"    {sink}(conn, org, webhook_url)",
+            "",
+        ]
+    )
+
+
+def _scan_rule_findings(tmp_path, source):
+    """Run the real file scanner and return only stored-SSRF findings."""
+    source_file = tmp_path / "webhook.py"
+    source_file.write_text(source, encoding="utf-8")
+    return [
+        finding
+        for finding in _scan_file(source_file, tmp_path)
+        if finding["rule_id"] == _RULE_ID
+    ]
+
+
+def test_packaged_rule_matches_direct_request_url_persistence():
+    """Detect direct request URL persistence with HIGH severity."""
+    rule = _rule()
+    assert rule["severity"] == "HIGH"
+    assert rule["pattern"].search(_vulnerable_source())
+
+
+def test_packaged_rule_matches_unvalidated_variable_persistence():
+    """Detect source-to-sink persistence through a local variable."""
+    assert _rule()["pattern"].search(_unvalidated_variable_source())
+
+
+def test_packaged_rule_does_not_depend_on_url_variable_name():
+    """Detect the flow even when the variable name omits the word URL."""
+    assert _rule()["pattern"].search(_unvalidated_variable_source("target"))
+
+
+def test_packaged_rule_does_not_treat_ignored_validator_result_as_safe():
+    """Detect a validator call whose boolean result is discarded."""
+    assert _rule()["pattern"].search(_ignored_validation_result_source())
+
+
+def test_packaged_rule_matches_non_enforcing_validation_guard():
+    """Detect a validation branch that logs but does not terminate."""
+    assert _rule()["pattern"].search(_non_enforcing_guard_source())
+
+
+def test_packaged_rule_matches_non_enforcing_guard_with_unrelated_return():
+    """Ignore unrelated returns when deciding whether validation enforces."""
+    assert _rule()["pattern"].search(
+        _non_enforcing_guard_with_unrelated_return_source()
+    )
+
+
+def test_packaged_rule_ignores_positive_guarded_persistence():
+    """Do not flag a sink that is reachable only after positive validation."""
+    assert not _rule()["pattern"].search(_positive_guard_source())
+
+
+def test_packaged_rule_ignores_fail_closed_guarded_persistence():
+    """Do not flag a flow that raises on invalid input before persistence."""
+    assert not _rule()["pattern"].search(_safe_source())
+
+
+def test_packaged_rule_ignores_production_fail_closed_guard():
+    """Do not self-flag the control plane's multiline rejection guard."""
+    assert not _rule()["pattern"].search(_production_guard_source())
+
+
+def test_scan_file_emits_stored_ssrf_finding(tmp_path):
+    """Emit a normalized SSRF finding through the production file scanner."""
+    matches = _scan_rule_findings(tmp_path, _vulnerable_source())
+
+    assert len(matches) == 1
+    finding = matches[0]
+    assert finding["severity"] == "HIGH"
+    assert finding["source"] == "appguardrail-rule"
+    assert finding["file"] == "webhook.py"
+    assert finding["line"] == 2
+    assert finding["category"] == "ssrf"
+    assert finding["cwe"] == ("CWE-918 - Server-Side Request Forgery",)
+    assert finding["owasp"] == ("OWASP A10:2021 - Server-Side Request Forgery",)
+    assert "destination" in finding["remediation"].lower()
+
+
+def test_scan_file_emits_stored_ssrf_finding_for_variable_flow(tmp_path):
+    """Emit a finding for an unvalidated local-variable persistence flow."""
+    matches = _scan_rule_findings(tmp_path, _unvalidated_variable_source())
+
+    assert len(matches) == 1
+    assert matches[0]["line"] == 2
+
+
+def test_scan_file_emits_finding_when_validator_result_is_ignored(tmp_path):
+    """Emit a finding when code ignores the validator's return value."""
+    matches = _scan_rule_findings(tmp_path, _ignored_validation_result_source())
+
+    assert len(matches) == 1
+    assert matches[0]["line"] == 2
+
+
+def test_scan_file_emits_finding_for_non_enforcing_guard(tmp_path):
+    """Emit a finding when an invalid branch fails to stop persistence."""
+    matches = _scan_rule_findings(tmp_path, _non_enforcing_guard_source())
+
+    assert len(matches) == 1
+    assert matches[0]["line"] == 2
+
+
+def test_scan_file_does_not_flag_validated_path(tmp_path):
+    """Suppress the finding for a verified fail-closed persistence path."""
+    assert not _scan_rule_findings(tmp_path, _safe_source())