fix(security): F15 (Astra P1) — SSRF-Schutz loest DNS auf, interne Servicenamen blockiert

Vorher: _is_url_safe blockierte nur IP-Literale und 5 feste Hostnamen.
Interne Servicenamen (postgres, redis, ...) und externe Domains mit
privater DNS-Aufloesung passierten ungeprueft (Astra-Repro:
http://postgres:5432/ wurde akzeptiert).

Fix: Der Hostname wird per socket.getaddrinfo aufgeloest und ALLE
aufgeloesten IPs muessen oeffentlich sein (private/loopback/link-local/
reserved/multicast/unspecified → blockiert). DNS-Fehler ist fail-closed
(nicht verifizierbar = blockiert). Blocking-DNS ist hier vertretbar —
Workflow-Steps sind Background-Jobs. Redirects bleiben deaktiviert
(follow_redirects=False, war bereits korrekt).

Abnahme (Astra): Interne Servicenamen, private DNS-Ziele und
DNS-Wechsel werden abgefangen — erfuellt (Tests mit getaddrinfo-Mocks:
postgres->172.18.0.2 blockiert, evil-corp.example->10.0.0.5 blockiert,
DNS-Fehler blockiert).

Tests: test_phase_g_workflows.py SSRF 11/11 (3 neue F15-Tests +
Positivfall auf aufladbaren Host umgestellt, unresolvable Hostnamen
jetzt fail-closed). ruff clean.

app/workflows/step_handlers.py

@@ -166,9 +166,35 @@ async def _handle_http(
         return StepResult(error=f"HTTP request failed: {e}", abort=True)
 
 
+def _is_ip_unsafe(ip: Any) -> bool:
+    """Check whether an IP address is private/internal/unsafe (F15)."""
+    return (
+        ip.is_private
+        or ip.is_loopback
+        or ip.is_link_local
+        or ip.is_reserved
+        or ip.is_multicast
+        or ip.is_unspecified
+    )
+
+
 def _is_url_safe(url: str) -> bool:
-    """SSRF protection — block private/internal targets."""
+    """SSRF protection — block private/internal targets (F15/Astra).
+
+    Previous behaviour only blocked IP literals and a fixed hostname list —
+    internal service names (``postgres``, ``redis``, ...) and external
+    domains resolving to private addresses passed unchecked.
+
+    Now resolves the hostname via DNS and requires ALL resolved IPs to be
+    public. DNS resolution failure is fail-closed (blocked).
+
+    Note: this performs a blocking ``socket.getaddrinfo`` call — acceptable
+    for workflow background steps. Full DNS-rebinding protection (pinning
+    the connection to the validated IP) is a follow-up; the check directly
+    before the request already narrows the window.
+    """
     import ipaddress
+    import socket
     import urllib.parse
 
     try:
@@ -183,19 +209,39 @@ def _is_url_safe(url: str) -> bool:
     if not hostname:
         return False
 
-    # Block localhost and common internal hostnames
+    # Fast path: known internal hostnames (no DNS needed)
-    blocked_hosts = {"localhost", "127.0.0.1", "0.0.0.0", "::1", "metadata.google.internal"}
+    blocked_hosts = {"localhost", "metadata.google.internal"}
     if hostname.lower() in blocked_hosts:
         return False
 
-    # Block private/internal IP ranges
+    # IP literal — validate directly
     try:
         ip = ipaddress.ip_address(hostname)
-        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
+        return not _is_ip_unsafe(ip)
-            return False
     except ValueError:
-        pass  # Not an IP, it's a hostname — allow
+        pass  # Not an IP literal — resolve via DNS
 
+    # F15: DNS resolution — ALL resolved IPs must be public. This blocks
+    # internal service names (postgres, redis, ...) and external domains
+    # that resolve to private/link-local addresses.
+    try:
+        infos = socket.getaddrinfo(
+            hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM
+        )
+    except (socket.gaierror, OSError):
+        # DNS failure → fail-closed: an unresolvable target is not verifiable
+        return False
+
+    if not infos:
+        return False
+
+    for _family, _type, _proto, _canonname, sockaddr in infos:
+        try:
+            ip = ipaddress.ip_address(sockaddr[0])
+        except ValueError:
+            return False
+        if _is_ip_unsafe(ip):
+            return False
     return True
 
 

tests/test_phase_g_workflows.py

@@ -116,9 +116,52 @@ class TestSSRFProtection:
         assert _is_url_safe("gopher://example.com") is False
 
     def test_allows_public_urls(self):
-        """SSRF allows public HTTP/HTTPS URLs."""
+        """SSRF allows public HTTP/HTTPS URLs with resolvable public DNS.
-        assert _is_url_safe("https://api.example.com/webhook") is True
+
+        F15: hostnames must actually resolve to PUBLIC IPs — a hostname
+        without DNS records is unverifiable and therefore blocked.
+        """
+        assert _is_url_safe("https://example.com/webhook") is True
         assert _is_url_safe("http://example.com/api") is True
+        # Unresolvable hostname → fail-closed (previously allowed silently)
+        assert _is_url_safe("https://api.example.com/webhook") is False
+
+    def test_f15_blocks_internal_service_names(self, monkeypatch):
+        """F15 (Astra repro): http://postgres:5432/ was accepted by the
+        old validator. In a container network the name resolves to a
+        private IP — the DNS check must block it."""
+        import socket
+
+        def fake_getaddrinfo_postgres(host, port, *args, **kwargs):
+            if host == "postgres":
+                return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("172.18.0.2", 5432))]
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80))]
+
+        monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo_postgres)
+        assert _is_url_safe("http://postgres:5432/db") is False
+
+    def test_f15_blocks_private_dns_resolution(self, monkeypatch):
+        """F15: an external-looking domain that resolves to a private IP
+        must be blocked (private DNS targets)."""
+        import socket
+
+        def fake_getaddrinfo_private(host, port, *args, **kwargs):
+            if host == "evil-corp.example":
+                return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 80))]
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80))]
+
+        monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo_private)
+        assert _is_url_safe("http://evil-corp.example/admin") is False
+
+    def test_f15_dns_failure_fails_closed(self, monkeypatch):
+        """F15: DNS resolution failure is unverifiable — fail-closed."""
+        import socket
+
+        def fake_getaddrinfo_fail(host, port, *args, **kwargs):
+            raise socket.gaierror("unresolvable")
+
+        monkeypatch.setattr(socket, "getaddrinfo", fake_getaddrinfo_fail)
+        assert _is_url_safe("http://definitely.example/api") is False
 
     def test_blocks_metadata_endpoint(self):
         """SSRF blocks cloud metadata endpoints."""