a2a-source-edit: write agent.py

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: