security(layout): SSRF guard on /layout/import-brand (reject internal hosts, no redirects)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
5ebb997978
commit
8535389cf1
@ -10,6 +10,7 @@ from sqlalchemy import text
|
|||||||
|
|
||||||
from .auth import require_token
|
from .auth import require_token
|
||||||
from .db import get_db
|
from .db import get_db
|
||||||
|
from .urlguard import safe_public_url
|
||||||
|
|
||||||
# Storefront customizer — per-store theme + header menu + which elements appear on cards /
|
# 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.
|
# 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"):
|
if not url.startswith("http"):
|
||||||
url = "https://" + url
|
url = "https://" + url
|
||||||
try:
|
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:
|
headers={"User-Agent": "RecordGod/0.1 brand-import"}) as c:
|
||||||
html = (await c.get(url)).text
|
html = (await c.get(url)).text
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
|||||||
16
app/urlguard.py
Normal file
16
app/urlguard.py
Normal file
@ -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
|
||||||
Loading…
Reference in New Issue
Block a user