fix(webhook): enforce URL admission at storage boundary

appguardrail_core/controlplane.py

@@ -137,7 +137,12 @@ def _drift_fp(finding: dict[str, Any]) -> str:
 
 def set_webhook(conn: sqlite3.Connection, org_id: int, url: "str | None") -> None:
     """Set (or clear) the org's drift-alert webhook URL."""
-    conn.execute("UPDATE orgs SET webhook_url = ? WHERE id = ?", (url or None, org_id))
+    normalized_url = None if isinstance(url, str) and url == "" else url
+    if normalized_url is not None and not _is_safe_url(normalized_url):
+        raise ValueError("Invalid webhook URL")
+    conn.execute(
+        "UPDATE orgs SET webhook_url = ? WHERE id = ?", (normalized_url, org_id)
+    )
     conn.commit()
 
 
@@ -638,10 +643,11 @@ def do_POST(self):
                 if body is None or not isinstance(body, dict):
                     return self._json(400, {"error": "invalid JSON body"})
                 webhook_url = body.get("url")
-                if webhook_url is not None and not _is_safe_url(webhook_url):
+                try:
+                    set_webhook(conn, org, webhook_url)
+                except ValueError:
                     return self._json(400, {"error": "invalid webhook url"})
-                set_webhook(conn, org, webhook_url)
-                return self._json(200, {"webhook_url": webhook_url})
+                return self._json(200, {"webhook_url": None if isinstance(webhook_url, str) and webhook_url == "" else webhook_url})
 
             if path == "/api/v1/keys":
                 if not has_role(role, "owner"):

docs/TRACEABILITY.md

@@ -47,10 +47,12 @@ For stored webhook/callback SSRF, trace separately:
 
 Current protected-branch evidence keeps those controls distinct: PR #924 supplies the fail-closed webhook storage boundary, and PR #910 supplies the packaged `python-stored-ssrf-webhook-url` detector plus focused regression corpus. Neither control expands the detector beyond its declared source/sink and flow contract.
 
+PR #1107 is a Proposed repair of both boundaries, not protected-branch evidence. Moving validation into the local `set_webhook` persistence function exposed a detector false positive: the route's request-derived variable was reported even though the unique top-level sink rejected unsafe input before SQLite use. The regression contract suppresses that finding only when the Python AST proves the local sink is unique, is not rebound, rejects a directly derived destination unconditionally, and uses that validated value only after the guard. An unrelated conditional guard or later `set_webhook` rebinding remains positive. PR #1068 remains the prerequisite for the stronger unresolved-DNS fail-closed runtime and detector contract; persistence-time validation alone does not establish delivery-time DNS-rebinding resistance.
+
 ## Standards/research
 
 Existing repository docs/doctoring/security evidence remain the bibliography/source-of-truth for standards such as SARIF, CycloneDX, GitHub security interfaces, and applicable OWASP/CWE classes. Material new detector classes should add authoritative standard/CWE/OWASP references and APA 7 citations in doctoring where research/standards materially drive implementation.
 
 ## Change rule
 
-Every new issue-class detector or product security boundary should add/update a row and its concrete test/evidence path. Stale/queued/cancelled/rate-limited/predecessor checks cannot promote evidence maturity.
+Every new issue-class detector or product security boundary should add/update a row and its concrete test/evidence path. Stale/queued/cancelled/rate-limited/predecessor checks cannot promote evidence maturity.

scanner/cli/appguardrail.py

@@ -40,6 +40,7 @@
 """
 
 import argparse
+import ast
 import fnmatch
 import functools
 import importlib.resources as resources  # nosemgrep: python.lang.compatibility.python37.python37-compatibility-importlib2
@@ -2916,6 +2917,123 @@ def _run_codegraph_index(scan_path: Path):
     return _run_codegraph_command([codegraph, "status"], workdir, "status")
 
 
+def _is_fail_closed_url_guard(test, validated_names):
+    """Return whether ``test`` rejects an unsafe validated local unconditionally."""
+    terms = (
+        test.values
+        if isinstance(test, ast.BoolOp) and isinstance(test.op, ast.And)
+        else [test]
+    )
+    unsafe_guard = False
+    for term in terms:
+        if (
+            isinstance(term, ast.UnaryOp)
+            and isinstance(term.op, ast.Not)
+            and isinstance(term.operand, ast.Call)
+            and isinstance(term.operand.func, ast.Name)
+            and term.operand.func.id == "_is_safe_url"
+            and len(term.operand.args) == 1
+            and isinstance(term.operand.args[0], ast.Name)
+            and term.operand.args[0].id in validated_names
+        ):
+            unsafe_guard = True
+        elif not (
+            isinstance(term, ast.Compare)
+            and isinstance(term.left, ast.Name)
+            and term.left.id in validated_names
+            and len(term.ops) == 1
+            and isinstance(term.ops[0], ast.IsNot)
+            and len(term.comparators) == 1
+            and isinstance(term.comparators[0], ast.Constant)
+            and term.comparators[0].value is None
+        ):
+            return False
+    return unsafe_guard
+
+
+def _has_fail_closed_local_webhook_sink(content):
+    """Recognize a top-level ``set_webhook`` that validates before SQLite use."""
+    try:
+        module = ast.parse(content)
+    except (SyntaxError, ValueError):
+        return False
+
+    definitions = [
+        statement
+        for statement in module.body
+        if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and statement.name == "set_webhook"
+    ]
+    if len(definitions) != 1:
+        return False
+    function = definitions[0]
+    for later in module.body[module.body.index(function) + 1 :]:
+        if (
+            isinstance(later, ast.Assign)
+            and any(
+                isinstance(target, ast.Name) and target.id == "set_webhook"
+                for target in later.targets
+            )
+        ) or (
+            isinstance(later, (ast.AnnAssign, ast.AugAssign))
+            and isinstance(later.target, ast.Name)
+            and later.target.id == "set_webhook"
+        ):
+            return False
+
+    for function in definitions:
+        validated_names = {argument.arg for argument in function.args.args}
+        for index, statement in enumerate(function.body):
+            if (
+                index == 0
+                and isinstance(statement, ast.Expr)
+                and isinstance(statement.value, ast.Constant)
+                and isinstance(statement.value.value, str)
+            ):
+                continue
+            if (
+                isinstance(statement, ast.Assign)
+                and len(statement.targets) == 1
+                and isinstance(statement.targets[0], ast.Name)
+                and isinstance(statement.value, ast.IfExp)
+                and isinstance(statement.value.body, ast.Constant)
+                and statement.value.body.value is None
+                and isinstance(statement.value.orelse, ast.Name)
+                and statement.value.orelse.id in validated_names
+            ):
+                validated_names.add(statement.targets[0].id)
+                continue
+            if (
+                isinstance(statement, ast.If)
+                and statement.body
+                and isinstance(statement.body[0], (ast.Raise, ast.Return))
+                and _is_fail_closed_url_guard(statement.test, validated_names)
+            ):
+                guarded_name = next(
+                    node.args[0].id
+                    for node in ast.walk(statement.test)
+                    if isinstance(node, ast.Call)
+                    and isinstance(node.func, ast.Name)
+                    and node.func.id == "_is_safe_url"
+                    and node.args
+                    and isinstance(node.args[0], ast.Name)
+                )
+                return any(
+                    isinstance(node, ast.Call)
+                    and isinstance(node.func, ast.Attribute)
+                    and node.func.attr == "execute"
+                    and any(
+                        isinstance(value, ast.Name) and value.id == guarded_name
+                        for argument in node.args
+                        for value in ast.walk(argument)
+                    )
+                    for later in function.body[index + 1 :]
+                    for node in ast.walk(later)
+                )
+            return False
+    return False
+
+
 def _scan_file(
     file_path: Path,
     base_path: Path,
@@ -2988,6 +3106,11 @@ def _scan_file(
                         rel_path_for_filters, include_paths, exclude_paths
                     ):
                         continue
+                if (
+                    rule_id == "python-stored-ssrf-webhook-url"
+                    and _has_fail_closed_local_webhook_sink(content)
+                ):
+                    continue
                 # ⚡ Bolt: Progressive line counting for O(N) instead of O(N*M)
                 # finditer yields matches in order, allowing us to scan for newlines
                 # incrementally from the last known position rather than starting from 0.

tests/test_controlplane.py

@@ -260,6 +260,14 @@ def test_api_explicit_webhook_deletion(server):
     assert body["webhook_url"] is None
 
 
+def test_api_empty_string_webhook_deletion(server):
+    base, key = server
+    status, body = _req("POST", f"{base}/api/v1/webhook", key, {"url": ""})
+
+    assert status == 200
+    assert body["webhook_url"] is None
+
+
 def test_api_set_webhook_ssrf_protection(server):
     base, key = server
     # Invalid type

tests/test_ssrf_rules.py

@@ -177,6 +177,35 @@ def _none_aware_guard_source():
     )
 
 
+def _internally_validated_sink_source(*, conditional=False, rebound=False):
+    """Build a route whose local persistence function owns validation."""
+    condition = (
+        "disabled and not _is_safe_url(normalized_url)"
+        if conditional
+        else "normalized_url is not None and not _is_safe_url(normalized_url)"
+    )
+    lines = [
+            "def set_webhook(conn, org, url, disabled=False):",
+            '    """Persist a validated webhook destination."""',
+            '    normalized_url = None if isinstance(url, str) and url == "" else url',
+            f"    if {condition}:",
+            '        raise ValueError("unsafe webhook url")',
+            '    conn.execute("UPDATE orgs SET webhook_url = ?", (normalized_url,))',
+            "",
+        ]
+    if rebound:
+        lines.extend(["set_webhook = unsafe_set_webhook", ""])
+    lines.extend(
+        [
+            "def update_webhook(conn, org, body):",
+            '    webhook_url = body.get("url")',
+            "    set_webhook(conn, org, webhook_url)",
+            "",
+        ]
+    )
+    return "\n".join(lines)
+
+
 def _scan_rule_findings(tmp_path, source):
     """Run the real file scanner and return only stored-SSRF findings."""
     source_file = tmp_path / "webhook.py"
@@ -325,3 +354,26 @@ def test_scan_file_emits_finding_for_non_enforcing_guard(tmp_path):
 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())
+
+
+def test_scan_file_accepts_validation_owned_by_local_sink(tmp_path):
+    """Accept a top-level sink that rejects unsafe input before persistence."""
+    assert not _scan_rule_findings(tmp_path, _internally_validated_sink_source())
+
+
+def test_scan_file_rejects_conditionally_validated_local_sink(tmp_path):
+    """Do not trust sink validation gated by an unrelated condition."""
+    findings = _scan_rule_findings(
+        tmp_path, _internally_validated_sink_source(conditional=True)
+    )
+
+    assert len(findings) == 1
+
+
+def test_scan_file_rejects_rebound_local_sink(tmp_path):
+    """Do not transfer a definition's safety after the sink name is rebound."""
+    findings = _scan_rule_findings(
+        tmp_path, _internally_validated_sink_source(rebound=True)
+    )
+
+    assert len(findings) == 1

tests/test_webhook_storage_ssrf_contract.py

@@ -0,0 +1,53 @@
+import pytest
+
+from appguardrail_core.controlplane import connect, create_org, set_webhook
+
+
+def test_set_webhook_rejects_loopback_before_persistence() -> None:
+    conn = connect(":memory:")
+    org_id, _ = create_org(conn, "ssrf-contract")
+
+    with pytest.raises(ValueError, match="Invalid webhook URL"):
+        set_webhook(conn, org_id, "http://127.0.0.1:8080/internal")
+
+    row = conn.execute(
+        "SELECT webhook_url FROM orgs WHERE id = ?", (org_id,)
+    ).fetchone()
+    assert row["webhook_url"] is None
+
+
+def test_set_webhook_allows_explicit_clear_without_url_validation() -> None:
+    conn = connect(":memory:")
+    org_id, _ = create_org(conn, "clear-contract")
+
+    set_webhook(conn, org_id, None)
+
+    row = conn.execute(
+        "SELECT webhook_url FROM orgs WHERE id = ?", (org_id,)
+    ).fetchone()
+    assert row["webhook_url"] is None
+
+
+def test_set_webhook_treats_empty_string_as_explicit_clear() -> None:
+    conn = connect(":memory:")
+    org_id, _ = create_org(conn, "empty-clear-contract")
+    conn.execute(
+        "UPDATE orgs SET webhook_url = ? WHERE id = ?",
+        ("https://example.com/hook", org_id),
+    )
+    conn.commit()
+
+    set_webhook(conn, org_id, "")
+
+    row = conn.execute(
+        "SELECT webhook_url FROM orgs WHERE id = ?", (org_id,)
+    ).fetchone()
+    assert row["webhook_url"] is None
+
+
+def test_set_webhook_rejects_non_string_values() -> None:
+    conn = connect(":memory:")
+    org_id, _ = create_org(conn, "type-contract")
+
+    with pytest.raises(ValueError, match="Invalid webhook URL"):
+        set_webhook(conn, org_id, 1234)  # type: ignore[arg-type]