diff --git a/app/layout_routes.py b/app/layout_routes.py index 091beed..9898ae4 100644 --- a/app/layout_routes.py +++ b/app/layout_routes.py @@ -10,6 +10,7 @@ from sqlalchemy import text from .auth import require_token from .db import get_db +from .urlguard import safe_public_url # Storefront customizer — per-store theme + header menu + which elements appear on cards / # product pages, in what order. Lets RecordGod dress ANY shop without touching code. @@ -71,7 +72,12 @@ async def import_brand(body: UrlIn, ident=Depends(require_token)): if not url.startswith("http"): url = "https://" + url try: - async with httpx.AsyncClient(timeout=15, follow_redirects=True, + url = safe_public_url(url) + except ValueError as e: + raise HTTPException(400, f"refused url: {e}") + # ponytail: DNS-rebinding TOCTOU remains; follow_redirects=False closes the redirect bypass + try: + async with httpx.AsyncClient(timeout=15, follow_redirects=False, headers={"User-Agent": "RecordGod/0.1 brand-import"}) as c: html = (await c.get(url)).text except Exception as e: diff --git a/app/urlguard.py b/app/urlguard.py new file mode 100644 index 0000000..ae0218c --- /dev/null +++ b/app/urlguard.py @@ -0,0 +1,16 @@ +import ipaddress +import socket +from urllib.parse import urlparse + + +def safe_public_url(url): + """Resolve url's host and reject any that maps to a non-public IP (SSRF guard).""" + p = urlparse(url) + if p.scheme not in ("http", "https") or not p.hostname: + raise ValueError("url must be http(s) with a hostname") + for info in socket.getaddrinfo(p.hostname, None): + ip = ipaddress.ip_address(info[4][0]) + if (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved + or ip.is_multicast or ip.is_unspecified): + raise ValueError(f"host resolves to non-public ip {ip}") + return url