| 1 | """Network security utilities — SSRF protection and internal URL detection.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import ipaddress |
| 6 | import re |
| 7 | import socket |
| 8 | from urllib.parse import urlparse |
| 9 | |
| 10 | _BLOCKED_NETWORKS = [ |
| 11 | ipaddress.ip_network("0.0.0.0/8"), |
| 12 | ipaddress.ip_network("10.0.0.0/8"), |
| 13 | ipaddress.ip_network("100.64.0.0/10"), # carrier-grade NAT |
| 14 | ipaddress.ip_network("127.0.0.0/8"), |
| 15 | ipaddress.ip_network("169.254.0.0/16"), # link-local / cloud metadata |
| 16 | ipaddress.ip_network("172.16.0.0/12"), |
| 17 | ipaddress.ip_network("192.168.0.0/16"), |
| 18 | ipaddress.ip_network("::1/128"), |
| 19 | ipaddress.ip_network("fc00::/7"), # unique local |
| 20 | ipaddress.ip_network("fe80::/10"), # link-local v6 |
| 21 | ] |
| 22 | |
| 23 | _URL_RE = re.compile(r"https?://[^\s\"'`;|<>]+", re.IGNORECASE) |
| 24 | |
| 25 | _allowed_networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = [] |
| 26 | |
| 27 | |
| 28 | def configure_ssrf_whitelist(cidrs: list[str]) -> None: |
| 29 | """Allow specific CIDR ranges to bypass SSRF blocking (e.g. Tailscale's 100.64.0.0/10).""" |
| 30 | global _allowed_networks |
| 31 | nets = [] |
| 32 | for cidr in cidrs: |
| 33 | try: |
| 34 | nets.append(ipaddress.ip_network(cidr, strict=False)) |
| 35 | except ValueError: |
| 36 | pass |
| 37 | _allowed_networks = nets |
| 38 | |
| 39 | |
| 40 | def _is_private(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: |
| 41 | if _allowed_networks and any(addr in net for net in _allowed_networks): |
| 42 | return False |
| 43 | return any(addr in net for net in _BLOCKED_NETWORKS) |
| 44 | |
| 45 | |
| 46 | def validate_url_target(url: str) -> tuple[bool, str]: |
| 47 | """Validate a URL is safe to fetch: scheme, hostname, and resolved IPs. |
| 48 | |
| 49 | Returns (ok, error_message). When ok is True, error_message is empty. |
| 50 | """ |
| 51 | try: |
| 52 | p = urlparse(url) |
| 53 | except Exception as e: |
| 54 | return False, str(e) |
| 55 | |
| 56 | if p.scheme not in ("http", "https"): |
| 57 | return False, f"Only http/https allowed, got '{p.scheme or 'none'}'" |
| 58 | if not p.netloc: |
| 59 | return False, "Missing domain" |
| 60 | |
| 61 | hostname = p.hostname |
| 62 | if not hostname: |
| 63 | return False, "Missing hostname" |
| 64 | |
| 65 | try: |
| 66 | infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) |
| 67 | except socket.gaierror: |
| 68 | return False, f"Cannot resolve hostname: {hostname}" |
| 69 | |
| 70 | for info in infos: |
| 71 | try: |
| 72 | addr = ipaddress.ip_address(info[4][0]) |
| 73 | except ValueError: |
| 74 | continue |
| 75 | if _is_private(addr): |
| 76 | return False, f"Blocked: {hostname} resolves to private/internal address {addr}" |
| 77 | |
| 78 | return True, "" |
| 79 | |
| 80 | |
| 81 | def validate_resolved_url(url: str) -> tuple[bool, str]: |
| 82 | """Validate an already-fetched URL (e.g. after redirect). Only checks the IP, skips DNS.""" |
| 83 | try: |
| 84 | p = urlparse(url) |
| 85 | except Exception: |
| 86 | return True, "" |
| 87 | |
| 88 | hostname = p.hostname |
| 89 | if not hostname: |
| 90 | return True, "" |
| 91 | |
| 92 | try: |
| 93 | addr = ipaddress.ip_address(hostname) |
| 94 | if _is_private(addr): |
| 95 | return False, f"Redirect target is a private address: {addr}" |
| 96 | except ValueError: |
| 97 | # hostname is a domain name, resolve it |
| 98 | try: |
| 99 | infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) |
| 100 | except socket.gaierror: |
| 101 | return True, "" |
| 102 | for info in infos: |
| 103 | try: |
| 104 | addr = ipaddress.ip_address(info[4][0]) |
| 105 | except ValueError: |
| 106 | continue |
| 107 | if _is_private(addr): |
| 108 | return False, f"Redirect target {hostname} resolves to private address {addr}" |
| 109 | |
| 110 | return True, "" |
| 111 | |
| 112 | |
| 113 | def contains_internal_url(command: str) -> bool: |
| 114 | """Return True if the command string contains a URL targeting an internal/private address.""" |
| 115 | for m in _URL_RE.finditer(command): |
| 116 | url = m.group(0) |
| 117 | ok, _ = validate_url_target(url) |
| 118 | if not ok: |
| 119 | return True |
| 120 | return False |
| 121 |