Nexus Agent
e833b923c8
接入 Playwright Worker、会话 WebSocket 与全局浏览器面板（固定于 App Bar 下）； 含验证码 stealth、设置项默认音与 URL 安全校验；附带 worker 部署与设计文档。 Co-authored-by: Cursor <cursoragent@cursor.com>
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""SSRF-safe URL validation (Worker copy — keep in sync with server/utils/browser_url_safe.py)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import socket
|
|
from urllib.parse import urlparse
|
|
|
|
BLOCKED_HOSTNAMES = frozenset({
|
|
"localhost",
|
|
"localhost.localdomain",
|
|
"metadata",
|
|
"metadata.google.internal",
|
|
"169.254.169.254",
|
|
})
|
|
|
|
|
|
def _blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
|
|
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast:
|
|
return True
|
|
if isinstance(ip, ipaddress.IPv4Address):
|
|
if ip in ipaddress.ip_network("0.0.0.0/8"):
|
|
return True
|
|
if ip in ipaddress.ip_network("100.64.0.0/10"):
|
|
return True
|
|
if isinstance(ip, ipaddress.IPv6Address):
|
|
if ip == ipaddress.IPv6Address("::1"):
|
|
return True
|
|
if ip in ipaddress.ip_network("fc00::/7"):
|
|
return True
|
|
if ip in ipaddress.ip_network("fe80::/10"):
|
|
return True
|
|
return False
|
|
|
|
|
|
def resolve_host_blocked(hostname: str) -> str | None:
|
|
host = hostname.lower().strip(".")
|
|
if not host:
|
|
return "Empty hostname"
|
|
if host in BLOCKED_HOSTNAMES or "metadata" in host:
|
|
return f"Blocked hostname: {hostname}"
|
|
try:
|
|
infos = socket.getaddrinfo(host, None, type=socket.SOCK_STREAM)
|
|
except socket.gaierror as exc:
|
|
return f"DNS resolution failed: {exc}"
|
|
if not infos:
|
|
return f"No addresses for hostname: {hostname}"
|
|
for info in infos:
|
|
ip_str = info[4][0]
|
|
try:
|
|
ip = ipaddress.ip_address(ip_str)
|
|
except ValueError:
|
|
return f"Invalid IP: {ip_str}"
|
|
if _blocked_ip(ip):
|
|
return f"Blocked IP: {ip_str}"
|
|
return None
|
|
|
|
|
|
def validate_navigation_url(url: str) -> tuple[str | None, str | None]:
|
|
trimmed = url.strip()
|
|
if not trimmed:
|
|
return None, "Empty URL"
|
|
try:
|
|
with_proto = trimmed if "://" in trimmed else f"https://{trimmed}"
|
|
parsed = urlparse(with_proto)
|
|
except Exception:
|
|
return None, "Invalid URL"
|
|
if parsed.scheme not in ("http", "https"):
|
|
return None, f"Unsupported scheme: {parsed.scheme}"
|
|
if parsed.username or parsed.password:
|
|
return None, "Credentials in URL are not allowed"
|
|
if not parsed.hostname:
|
|
return None, "Missing hostname"
|
|
host_err = resolve_host_blocked(parsed.hostname)
|
|
if host_err:
|
|
return None, host_err
|
|
return parsed.geturl(), None
|