Spaces:
Running
Running
| """资料持久媒体地址准入；不解析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 | |