diff --git a/agent.py b/agent.py index b2b053f..c9160dd 100644 --- a/agent.py +++ b/agent.py @@ -40,6 +40,35 @@ READ_TIMEOUT_SECONDS = 5.0 TOTAL_TIMEOUT_SECONDS = 12.0 USER_AGENT = "LaunchCheckStudio/0.1 (+https://a2acloud.io)" DB_CONNECT_OPTIONS = "-c statement_timeout=5000 -c lock_timeout=2000 -c idle_in_transaction_session_timeout=5000" +ALLOWED_PORTS_BY_SCHEME = {"http": {80}, "https": {443}} +DISALLOWED_IP_NETWORKS = tuple( + ipaddress.ip_network(value) + for value in ( + "0.0.0.0/8", + "10.0.0.0/8", + "100.64.0.0/10", + "127.0.0.0/8", + "169.254.0.0/16", + "172.16.0.0/12", + "192.0.0.0/24", + "192.0.2.0/24", + "192.168.0.0/16", + "198.18.0.0/15", + "198.51.100.0/24", + "203.0.113.0/24", + "224.0.0.0/4", + "240.0.0.0/4", + "::/128", + "::1/128", + "::ffff:0:0/96", + "64:ff9b::/96", + "100::/64", + "2001:db8::/32", + "fc00::/7", + "fe80::/10", + "ff00::/8", + ) +) class LaunchCheckStudioV1Config(BaseModel): @@ -145,7 +174,7 @@ class LaunchCheckStudioV1(A2AAgent[LaunchCheckStudioV1Config, PlatformUserAuth]) "LaunchCheck audits public HTTP(S) launch pages with bounded SSRF-safe " "network checks and user-scoped Postgres persistence." ) - version = "0.1.1" + version = "0.1.2" config_model = LaunchCheckStudioV1Config auth_model = PlatformUserAuth @@ -305,7 +334,8 @@ async def _validate_public_url(raw_url: str) -> dict[str, Any]: if len(value) > 2048: return {"ok": False, "code": "invalid_url", "message": "URL is too long."} parsed = urlparse(value) - if parsed.scheme.lower() not in {"http", "https"}: + scheme = parsed.scheme.lower() + if scheme not in {"http", "https"}: return {"ok": False, "code": "invalid_scheme", "message": "Only http and https URLs are supported."} if parsed.username or parsed.password or "@" in parsed.netloc: return {"ok": False, "code": "credentials_not_allowed", "message": "URLs with embedded credentials are not allowed."} @@ -318,8 +348,12 @@ async def _validate_public_url(raw_url: str) -> dict[str, Any]: port = parsed.port except ValueError: return {"ok": False, "code": "invalid_url", "message": "URL port is invalid."} - if port is not None and not (1 <= port <= 65535): - return {"ok": False, "code": "invalid_url", "message": "URL port is invalid."} + if port is not None and port not in ALLOWED_PORTS_BY_SCHEME[scheme]: + return { + "ok": False, + "code": "disallowed_port", + "message": "Only default public web ports are supported: http/80 and https/443.", + } try: host = str(ipaddress.ip_address(host.strip("[]"))) except ValueError: @@ -327,10 +361,8 @@ async def _validate_public_url(raw_url: str) -> dict[str, Any]: host = host.encode("idna").decode("ascii").lower() except UnicodeError: return {"ok": False, "code": "invalid_url", "message": "URL hostname is invalid."} - addresses = await _resolve_public_addresses( - host, - port or (443 if parsed.scheme == "https" else 80), - ) + effective_port = port or (443 if scheme == "https" else 80) + addresses = await _resolve_public_addresses(host, effective_port) if addresses is None: return {"ok": False, "code": "disallowed_private_target", "message": "Target resolves to a non-public network address."} if not addresses: @@ -338,12 +370,12 @@ async def _validate_public_url(raw_url: str) -> dict[str, Any]: netloc = f"[{host}]" if ":" in host else host if port: netloc = f"{netloc}:{port}" - normalized = urlunparse((parsed.scheme.lower(), netloc, parsed.path or "", "", parsed.query, "")) + normalized = urlunparse((scheme, netloc, parsed.path or "", "", parsed.query, "")) return { "ok": True, "url": normalized, "hostname": host, - "port": port or (443 if parsed.scheme == "https" else 80), + "port": effective_port, "addresses": addresses, } @@ -377,16 +409,7 @@ async def _resolve_public_addresses(host: str, port: int) -> tuple[str, ...] | N def _ip_disallowed(ip: ipaddress._BaseAddress) -> bool: - metadata = ipaddress.ip_address("169.254.169.254") - return bool( - ip.is_private - or ip.is_loopback - or ip.is_link_local - or ip.is_multicast - or ip.is_reserved - or ip.is_unspecified - or ip == metadata - ) + return (not ip.is_global) or any(ip in network for network in DISALLOWED_IP_NETWORKS) async def _fetch_bounded(url: str, max_redirects: int, max_bytes: int) -> FetchResult: