diff --git a/appguardrail_core/controlplane.py b/appguardrail_core/controlplane.py index 576b990f..df7f8947 100644 --- a/appguardrail_core/controlplane.py +++ b/appguardrail_core/controlplane.py @@ -137,6 +137,8 @@ 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.""" + if url is not None and not _is_safe_url(url): + raise ValueError("Invalid webhook URL") conn.execute("UPDATE orgs SET webhook_url = ? WHERE id = ?", (url or None, org_id)) conn.commit() diff --git a/tests/test_webhook_storage_ssrf_contract.py b/tests/test_webhook_storage_ssrf_contract.py new file mode 100644 index 00000000..82dcaa99 --- /dev/null +++ b/tests/test_webhook_storage_ssrf_contract.py @@ -0,0 +1,28 @@ +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