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.
This commit is contained in:
@@ -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
|
||||
blocked_hosts = {"localhost", "127.0.0.1", "0.0.0.0", "::1", "metadata.google.internal"}
|
||||
# Fast path: known internal hostnames (no DNS needed)
|
||||
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 False
|
||||
return not _is_ip_unsafe(ip)
|
||||
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
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user