wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
"""Network security utilities for partner channels — SSRF protection.
|
|
|
|
Stricter than ``deeptutor.services.mcp.network`` (which deliberately allows
|
|
RFC1918 because MCP servers are often LAN-hosted): channel media URLs come
|
|
from external IM platforms, so every private/internal range is blocked.
|
|
|
|
Ported from nanobot's ``security/network.py``, trimmed to what channels use.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from contextlib import suppress
|
|
import ipaddress
|
|
import socket
|
|
from urllib.parse import urlparse
|
|
|
|
_BLOCKED_NETWORKS = [
|
|
ipaddress.ip_network("0.0.0.0/8"),
|
|
ipaddress.ip_network("10.0.0.0/8"),
|
|
ipaddress.ip_network("100.64.0.0/10"), # carrier-grade NAT
|
|
ipaddress.ip_network("127.0.0.0/8"),
|
|
ipaddress.ip_network("169.254.0.0/16"), # link-local / cloud metadata
|
|
ipaddress.ip_network("172.16.0.0/12"),
|
|
ipaddress.ip_network("192.168.0.0/16"),
|
|
ipaddress.ip_network("::1/128"),
|
|
ipaddress.ip_network("fc00::/7"), # unique local
|
|
ipaddress.ip_network("fe80::/10"), # link-local v6
|
|
]
|
|
|
|
|
|
def _normalize_addr(
|
|
addr: ipaddress.IPv4Address | ipaddress.IPv6Address,
|
|
) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
|
|
"""Normalize IPv6-mapped IPv4 addresses (``::ffff:127.0.0.1``) to IPv4."""
|
|
if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped is not None:
|
|
return addr.ipv4_mapped
|
|
return addr
|
|
|
|
|
|
def _is_private(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
|
|
normalized = _normalize_addr(addr)
|
|
return any(normalized in net for net in _BLOCKED_NETWORKS)
|
|
|
|
|
|
def _is_allowed_loopback_target(
|
|
hostname: str,
|
|
addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address],
|
|
) -> bool:
|
|
if not addrs or not all(_normalize_addr(addr).is_loopback for addr in addrs):
|
|
return False
|
|
normalized = hostname.rstrip(".").lower()
|
|
if normalized == "localhost":
|
|
return True
|
|
with suppress(ValueError):
|
|
return ipaddress.ip_address(hostname).is_loopback
|
|
return False
|
|
|
|
|
|
def validate_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, str]:
|
|
"""Validate a URL is safe to fetch: scheme, hostname, and resolved IPs.
|
|
|
|
``allow_loopback`` is intentionally narrow: it only permits literal
|
|
loopback hosts (localhost, 127.0.0.0/8, ::1) when every resolved address
|
|
is loopback. It does not allow RFC1918, link-local, metadata, or public
|
|
DNS names that happen to resolve to loopback.
|
|
|
|
Returns (ok, error_message). When ok is True, error_message is empty.
|
|
"""
|
|
try:
|
|
p = urlparse(url)
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
if p.scheme not in ("http", "https"):
|
|
return False, f"Only http/https allowed, got '{p.scheme or 'none'}'"
|
|
if not p.netloc:
|
|
return False, "Missing domain"
|
|
|
|
hostname = p.hostname
|
|
if not hostname:
|
|
return False, "Missing hostname"
|
|
|
|
try:
|
|
infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
|
|
except socket.gaierror:
|
|
return False, f"Cannot resolve hostname: {hostname}"
|
|
|
|
addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
|
|
for info in infos:
|
|
try:
|
|
addr = ipaddress.ip_address(info[4][0])
|
|
except ValueError:
|
|
continue
|
|
addrs.append(addr)
|
|
if allow_loopback and _is_allowed_loopback_target(hostname, addrs):
|
|
return True, ""
|
|
for addr in addrs:
|
|
if _is_private(addr):
|
|
return False, f"Blocked: {hostname} resolves to private/internal address {addr}"
|
|
|
|
return True, ""
|