🛡️ Sentinel: [HIGH] 저장된 SSRF 취약점 방지 및 URL 검증 간 예외(AttributeError) 패치

.jules/sentinel.md

@@ -122,3 +122,8 @@
 **Vulnerability:** DOM XSS via unescaped `severity` string interpolated into `innerHTML` in `scanner/dashboard/index.html`.
 **Learning:** Even enum-like or seemingly safe meta-fields like `severity` can contain malicious payloads if sourced from user input (findings file) and directly injected into innerHTML.
 **Prevention:** Always use the `esc()` sanitizer for any dynamically rendered property from `findings.json`, regardless of expected schema types.
+
+## 2025-02-28 - Stored SSRF and Unhandled Parsing Exceptions Guardrail
+**Vulnerability:** The `/api/v1/webhook` POST endpoint in `appguardrail_core/controlplane.py` failed to validate the `url` property when accepting it into the database, leading to Stored SSRF risks. In addition, the core SSRF validation logic (`_is_safe_url`) in both the CLI and control-plane did not verify the input type (e.g. `isinstance(url, str)`). Passing non-string types (like integers) resulted in unhandled `AttributeError` exceptions inside `urllib.parse.urlparse`, which led to API 500 crashes on malicious JSON payloads.
+**Learning:** Network endpoints must explicitly validate the data type of user-provided configurations prior to execution or storage. Furthermore, webhooks configured by users should always be checked for SSRF when saved, as trusting them later assumes input has already been safely validated, bypassing downstream network guardrails.
+**Prevention:** Apply `_is_safe_url` checks directly upon ingestion (e.g., in `/api/v1/webhook`) and enforce type checks `if not isinstance(url, str): return False` prior to using library parsing functions like `urlparse`. Always return gracefully failing responses (like `400 Bad Request`) for unsafe URLs instead of allowing unhandled 500 server errors.

appguardrail_core/controlplane.py

@@ -221,6 +221,9 @@ def _is_safe_url(url: str) -> bool:
     import urllib.parse
     import socket
 
+    if not isinstance(url, str):
+        return False
+
     try:
         parsed = urllib.parse.urlparse(
             url
@@ -612,7 +615,10 @@ def _body(self):
                 # Negative reads until EOF; oversized bodies exhaust memory.
                 return None
             try:
-                return json.loads(self.rfile.read(length) or b"{}")
+                raw_body = self.rfile.read(length)
+                if not raw_body:
+                    return None
+                return json.loads(raw_body)
             except (ValueError, TypeError):
                 return None
 
@@ -629,10 +635,13 @@ def do_POST(self):
                 if not has_role(role, "owner"):
                     return self._json(403, {"error": "owner role required"})
                 body = self._body()
-                if body is None:
+                if body is None or not isinstance(body, dict):
                     return self._json(400, {"error": "invalid JSON body"})
-                set_webhook(conn, org, (body or {}).get("url"))
-                return self._json(200, {"webhook_url": (body or {}).get("url")})
+                webhook_url = body.get("url")
+                if webhook_url is not None and not _is_safe_url(webhook_url):
+                    return self._json(400, {"error": "invalid webhook url"})
+                set_webhook(conn, org, webhook_url)
+                return self._json(200, {"webhook_url": webhook_url})
 
             if path == "/api/v1/keys":
                 if not has_role(role, "owner"):

scanner/cli/appguardrail.py

@@ -1623,6 +1623,9 @@ def _is_safe_url(url: str) -> bool:
     import socket
     import urllib.parse
 
+    if not isinstance(url, str):
+        return False
+
     try:
         parsed = urllib.parse.urlparse(
             url

tests/test_controlplane.py

@@ -231,6 +231,49 @@ def test_api_set_webhook(server):
     assert status == 200 and body["webhook_url"] == "http://hook.example/y"
 
 
+def test_api_empty_webhook_body_rejected(server):
+    import http.client
+    from urllib.parse import urlparse as _u
+
+    base, key = server
+    _req("POST", f"{base}/api/v1/webhook", key, {"url": "http://hook.example/y"})
+
+    parsed = _u(base)
+    conn = http.client.HTTPConnection(parsed.hostname, parsed.port, timeout=5)
+    conn.putrequest("POST", "/api/v1/webhook")
+    conn.putheader("Authorization", f"Bearer {key}")
+    conn.putheader("Content-Type", "application/json")
+    conn.putheader("Content-Length", "0")
+    conn.endheaders()
+    response = conn.getresponse()
+
+    assert response.status == 400
+    assert json.loads(response.read()) == {"error": "invalid JSON body"}
+    conn.close()
+
+
+def test_api_explicit_webhook_deletion(server):
+    base, key = server
+    status, body = _req("POST", f"{base}/api/v1/webhook", key, {"url": None})
+
+    assert status == 200
+    assert body["webhook_url"] is None
+
+
+def test_api_set_webhook_ssrf_protection(server):
+    base, key = server
+    # Invalid type
+    with pytest.raises(urllib.error.HTTPError) as exc:
+        _req("POST", f"{base}/api/v1/webhook", key, {"url": 1234})
+    assert exc.value.code == 400
+    assert json.loads(exc.value.read())["error"] == "invalid webhook url"
+    # Localhost SSRF attempt
+    with pytest.raises(urllib.error.HTTPError) as exc:
+        _req("POST", f"{base}/api/v1/webhook", key, {"url": "http://127.0.0.1/hook"})
+    assert exc.value.code == 400
+    assert json.loads(exc.value.read())["error"] == "invalid webhook url"
+
+
 def test_roles_and_key_scoping():
     conn = connect(":memory:")
     oid, owner_key = create_org(conn, "Acme")

tests/test_ssrf_protection.py

@@ -10,6 +10,11 @@ def test_is_safe_url_public_domains():
     assert _is_safe_url("http://google.com/")
     assert _is_safe_url("https://github.com/")
 
+def test_is_safe_url_invalid_types():
+    assert not _is_safe_url(None)
+    assert not _is_safe_url(123)
+    assert not _is_safe_url(True)
+
 
 def test_is_safe_url_ipv4_localhost():
     assert not _is_safe_url("http://127.0.0.1/")