| 1 | """External URL validation for server-side HTTP downloads (SSRF protection). |
| 2 | |
| 3 | Standalone module — copy into other services without nanobot context. |
| 4 | Configure :func:`configure_download_policy` before calling :func:`validate_external_url`. |
| 5 | """ |
| 6 | |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | import ipaddress |
| 10 | import socket |
| 11 | from dataclasses import dataclass, field |
| 12 | from urllib.parse import urlparse |
| 13 | |
| 14 | _ALLOWED_SCHEMES = frozenset({"http", "https"}) |
| 15 | _DEFAULT_DOMAIN_SUFFIXES: tuple[str, ...] = () |
| 16 | _DEFAULT_ALLOWED_PORTS = frozenset({"", "80", "443"}) |
| 17 | _EXTRA_PRIVATE_NETWORKS = ( |
| 18 | ipaddress.ip_network("100.64.0.0/10"), |
| 19 | ipaddress.ip_network("198.18.0.0/15"), |
| 20 | ) |
| 21 | _DNS_TIMEOUT_S = 3.0 |
| 22 | |
| 23 | |
| 24 | @dataclass |
| 25 | class DownloadUrlPolicy: |
| 26 | """Configurable SSRF policy for outbound HTTP(S) downloads.""" |
| 27 | |
| 28 | allowed_domain_suffixes: tuple[str, ...] = _DEFAULT_DOMAIN_SUFFIXES |
| 29 | trusted_internal_domains: frozenset[str] = frozenset() |
| 30 | allowed_ports: frozenset[str] = _DEFAULT_ALLOWED_PORTS |
| 31 | reject_ip_literal_hosts: bool = True |
| 32 | reject_userinfo: bool = True |
| 33 | |
| 34 | |
| 35 | _policy = DownloadUrlPolicy() |
| 36 | |
| 37 | |
| 38 | def configure_download_policy(policy: DownloadUrlPolicy) -> None: |
| 39 | """Replace the active download URL policy (call once at process startup).""" |
| 40 | global _policy |
| 41 | _policy = policy |
| 42 | |
| 43 | |
| 44 | def get_download_policy() -> DownloadUrlPolicy: |
| 45 | return _policy |
| 46 | |
| 47 | |
| 48 | class UrlValidationError(ValueError): |
| 49 | """Raised when an external URL fails SSRF validation.""" |
| 50 | |
| 51 | |
| 52 | def validate_external_url(raw_url: str) -> str: |
| 53 | """Validate *raw_url* and return the normalized URL string. |
| 54 | |
| 55 | Steps (order matters): |
| 56 | 1. Parse URL |
| 57 | 2. Reject userinfo (``http://user@host`` bypass) |
| 58 | 3. Scheme whitelist (http/https only) |
| 59 | 4. Hostname required |
| 60 | 5. Reject IP-literal hosts (optional, default on) |
| 61 | 6. Domain suffix whitelist |
| 62 | 7. Port whitelist (80/443/default) |
| 63 | 8. DNS resolve + private/reserved IP block (skipped for trusted internal domains) |
| 64 | """ |
| 65 | policy = _policy |
| 66 | try: |
| 67 | parsed = urlparse(raw_url.strip()) |
| 68 | except Exception as exc: |
| 69 | raise UrlValidationError("URL格式无效") from exc |
| 70 | |
| 71 | if policy.reject_userinfo and parsed.username is not None: |
| 72 | raise UrlValidationError("URL不允许包含用户名密码") |
| 73 | |
| 74 | scheme = (parsed.scheme or "").lower() |
| 75 | if scheme not in _ALLOWED_SCHEMES: |
| 76 | raise UrlValidationError(f"不允许的协议: {parsed.scheme or '(none)'}") |
| 77 | |
| 78 | host = (parsed.hostname or "").lower() |
| 79 | if not host: |
| 80 | raise UrlValidationError("URL缺少主机名") |
| 81 | |
| 82 | if policy.reject_ip_literal_hosts: |
| 83 | try: |
| 84 | ipaddress.ip_address(host) |
| 85 | except ValueError: |
| 86 | pass |
| 87 | else: |
| 88 | raise UrlValidationError(f"不允许使用IP地址访问: {host}") |
| 89 | |
| 90 | if not _is_domain_allowed(host, policy.allowed_domain_suffixes): |
| 91 | raise UrlValidationError(f"域名不在白名单中: {host}") |
| 92 | |
| 93 | port = parsed.port |
| 94 | port_text = "" if port is None else str(port) |
| 95 | if port_text not in policy.allowed_ports: |
| 96 | raise UrlValidationError(f"不允许的端口: {port_text or '(default)'}") |
| 97 | |
| 98 | if host not in policy.trusted_internal_domains: |
| 99 | _assert_host_resolves_to_public_ip(host) |
| 100 | |
| 101 | return raw_url.strip() |
| 102 | |
| 103 | |
| 104 | def _is_domain_allowed(host: str, suffixes: tuple[str, ...]) -> bool: |
| 105 | for suffix in suffixes: |
| 106 | root = suffix[1:] if suffix.startswith(".") else suffix |
| 107 | if host == root or host.endswith(suffix): |
| 108 | return True |
| 109 | return False |
| 110 | |
| 111 | |
| 112 | def _assert_host_resolves_to_public_ip(host: str) -> None: |
| 113 | try: |
| 114 | socket.setdefaulttimeout(_DNS_TIMEOUT_S) |
| 115 | infos = socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM) |
| 116 | except socket.gaierror as exc: |
| 117 | raise UrlValidationError("DNS解析失败") from exc |
| 118 | finally: |
| 119 | socket.setdefaulttimeout(None) |
| 120 | |
| 121 | if not infos: |
| 122 | raise UrlValidationError("DNS解析失败") |
| 123 | |
| 124 | for info in infos: |
| 125 | ip_text = info[4][0] |
| 126 | try: |
| 127 | addr = ipaddress.ip_address(ip_text) |
| 128 | except ValueError: |
| 129 | continue |
| 130 | if _is_private_or_reserved(addr): |
| 131 | raise UrlValidationError(f"禁止访问内网地址: {host} -> {addr}") |
| 132 | |
| 133 | |
| 134 | def _is_private_or_reserved(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: |
| 135 | if addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_unspecified: |
| 136 | return True |
| 137 | if addr.is_reserved: |
| 138 | return True |
| 139 | for network in _EXTRA_PRIVATE_NETWORKS: |
| 140 | if addr in network: |
| 141 | return True |
| 142 | return False |
| 143 |