profile_media.py

"""资料持久媒体地址准入；不解析DNS、不下载内容。"""
import ipaddress
import re
from urllib.parse import unquote, urlsplit
from deployment_config import PUBLIC_API_BASE_URL

MAX_PROFILE_MEDIA_URL_BYTES = 16384
LOCAL_IMAGE_PREFIX = "/community_hub/image?url="
COMPLETE_IMAGE_SUFFIX = "&require_complete=1"
INVALID_MEDIA = "invalid_profile_media_url"


def _host(parts):
    host = parts.hostname
    if not host or "%" in host:
        raise ValueError(INVALID_MEDIA)
    host = host.removesuffix(".").lower()
    if not host:
        raise ValueError(INVALID_MEDIA)
    try:
        address = ipaddress.ip_address(host)
    except ValueError:
        ascii_host = host.encode("idna").decode("ascii")
        labels = ascii_host.split(".")
        if (len(ascii_host) > 253 or not all(re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", label) for label in labels)):
            raise ValueError(INVALID_MEDIA)
        return ascii_host, None
    return address.compressed, address


def normalize_profile_media(value):
    if value is None:
        return None
    if not isinstance(value, str):
        raise ValueError(INVALID_MEDIA)
    if value == "":
        return ""
    if value.strip() != value or any(ord(char) < 32 or ord(char) == 127 for char in value):
        raise ValueError(INVALID_MEDIA)
    if value[:5].lower() in {"data:", "blob:"}:
        return None
    if len(value.encode("utf-8")) > MAX_PROFILE_MEDIA_URL_BYTES or "\\" in value:
        raise ValueError(INVALID_MEDIA)
    source = value
    layers = 0
    while source.startswith(LOCAL_IMAGE_PREFIX):
        layers += 1
        if layers > 10:
            raise ValueError(INVALID_MEDIA)
        encoded = source[len(LOCAL_IMAGE_PREFIX):]
        if encoded.endswith(COMPLETE_IMAGE_SUFFIX):
            encoded = encoded[:-len(COMPLETE_IMAGE_SUFFIX)]
        if re.search(r"%(?![a-fA-F0-9]{2})", encoded):
            raise ValueError(INVALID_MEDIA)
        source = unquote(encoded, errors="strict")
    if source.strip() != source or "\\" in source or any(ord(char) < 32 or ord(char) == 127 for char in source):
        raise ValueError(INVALID_MEDIA)
    if not re.match(r"^https?://", source, re.I):
        raise ValueError(INVALID_MEDIA)
    try:
        parsed = urlsplit(source)
        if parsed.username is not None or parsed.password is not None or parsed.netloc.endswith(":"):
            raise ValueError(INVALID_MEDIA)
        if parsed.port is not None and not 1 <= parsed.port <= 65535:
            raise ValueError(INVALID_MEDIA)
        port = parsed.port or (443 if parsed.scheme == "https" else 80)
        host, address = _host(parsed)
        base = urlsplit(PUBLIC_API_BASE_URL)
        base_host, _ = _host(base)
        if host == base_host:
            base_port = base.port or (443 if base.scheme == "https" else 80)
            if parsed.scheme != base.scheme or port != base_port or parsed.path != base.path.rstrip("/") + "/api/image_proxy":
                raise ValueError(INVALID_MEDIA)
            return source
        if address:
            if (not address.is_global or address.is_multicast or (address.version == 6
                    and (address.ipv4_mapped is not None or int(address) >> 125 != 1))):
                raise ValueError(INVALID_MEDIA)
        elif (host == "localhost" or host.endswith((".localhost", ".local")) or "." not in host
                or re.fullmatch(r"(?:0x[0-9a-f]+|[0-9]+)(?:\.(?:0x[0-9a-f]+|[0-9]+))*", host, re.I)
                or re.fullmatch(r"(?:0x[0-9a-f]+|[0-9]+)", host.rsplit(".", 1)[-1], re.I)):
            raise ValueError(INVALID_MEDIA)
        return source
    except (ValueError, UnicodeError):
        raise ValueError(INVALID_MEDIA) from None