import ipaddress import re from urllib.parse import urlparse from fastapi import HTTPException from app.config import settings METADATA_IPS = frozenset({"169.254.169.254", "169.254.170.2"}) HOSTNAME_RE = re.compile( r"^(?=.{1,253}$)(?!-)[a-zA-Z0-9-]{1,63}(?\n\r]") def _parse_target(raw: str) -> tuple[str, str]: """Return (kind, normalized) where kind is ip|cidr|hostname|url.""" raw = raw.strip() if not raw: raise HTTPException(status_code=400, detail="Target is required") if SHELL_METACHAR_RE.search(raw): raise HTTPException(status_code=400, detail="Target contains invalid characters") if "://" in raw: parsed = urlparse(raw) if parsed.scheme not in ("http", "https"): raise HTTPException(status_code=400, detail="URL must be http or https") host = parsed.hostname if not host: raise HTTPException(status_code=400, detail="Invalid URL hostname") return "url", raw if "/" in raw and not raw.startswith("/"): try: net = ipaddress.ip_network(raw, strict=False) return "cidr", str(net) except ValueError: pass try: addr = ipaddress.ip_address(raw) return "ip", str(addr) except ValueError: pass host_part = raw.split(":")[0] if HOSTNAME_RE.match(host_part) or host_part == "localhost": return "hostname", raw raise HTTPException(status_code=400, detail="Invalid target: must be IP, CIDR, hostname, or URL") def _extract_ips(target: str, kind: str) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: if kind == "url": host = urlparse(target).hostname or "" try: return [ipaddress.ip_address(host)] except ValueError: return [] if kind == "cidr": net = ipaddress.ip_network(target, strict=False) if net.num_addresses == 1: return [ipaddress.ip_address(net.network_address)] return [] if kind in ("ip", "hostname"): try: return [ipaddress.ip_address(target.split(":")[0])] except ValueError: return [] return [] def _is_loopback_target(target: str, kind: str) -> bool: if kind == "hostname" and target.lower().startswith("localhost"): return True for ip in _extract_ips(target, kind): if ip.is_loopback: return True if kind == "url": host = (urlparse(target).hostname or "").lower() if host in ("localhost", "127.0.0.1", "::1"): return True return False def validate_scan_request(target: str, authorized: bool) -> str: if not authorized: raise HTTPException( status_code=400, detail="Authorization required: set authorized=true after confirming ownership", ) kind, normalized = _parse_target(target) if normalized in settings.blocked_targets_list or target in settings.blocked_targets_list: raise HTTPException(status_code=400, detail="Target is blocklisted") for blocked in METADATA_IPS: if blocked in target or normalized == blocked: raise HTTPException(status_code=400, detail="Cloud metadata endpoints are blocked") for blocked in settings.blocked_targets_list: if blocked in target: raise HTTPException(status_code=400, detail="Target is blocklisted") if _is_loopback_target(normalized, kind) and not settings.allow_loopback: raise HTTPException(status_code=400, detail="Loopback targets are blocked (set ALLOW_LOOPBACK=true to override)") for ip in _extract_ips(normalized, kind): if str(ip) in METADATA_IPS: raise HTTPException(status_code=400, detail="Cloud metadata endpoints are blocked") return normalized