Spaces:

VoiceOfML
/

Search

Running

Search / backend /upstream_http.py

fix(proxy): stop accepting the deleted Reader-Assets dataset as a Reader source

33b0f86 about 11 hours ago

11.9 kB

"""Validated upstream requests and bounded redirect reuse, independent of API state."""
import asyncio
import re
import time
from urllib.parse import unquote, urljoin, urlparse

import aiohttp

ALLOWED_DOWNLOAD_HOSTS = {"huggingface.co", "hf-mirror.com", "hf.co"}
ALLOWED_DOWNLOAD_HOST_SUFFIXES = (".hf.co", ".huggingface.co", ".xethub.hf.co")
# Redirects are a narrower trust boundary than user-supplied source URLs. Keep
# this exact-host list in sync with the signed storage targets we actually use;
# accepting every future *.hf.co host would let a redirect escape the dataset
# scope checked on the initial URL.
TRUSTED_REDIRECT_HOSTS = frozenset({
    "cdn-lfs.hf.co", "cdn-lfs.huggingface.co", "cdn-lfs-us-1.hf.co",
    "cdn-lfs-us-1.huggingface.co", "us.aws.cdn.hf.co",
    "cas-bridge.xethub.hf.co", "cas-server.xethub.hf.co", "transfer.xethub.hf.co",
})
MAX_REDIRECTS = 5
READER_ASSET_PATH_RE = re.compile(
    r"objects/[0-9a-f]{2}/[0-9a-f]{64}/(?:[0-9a-f]{16}/)?(?:page-manifest\.json|ocr-manifest\.json|pages/page-[0-9]{6}\.(?:webp|jxl)|ocr/(?:page-[0-9]{6}\.json\.gz|book-text\.json\.gz)|(?:[a-z0-9-]+/)?(?:linearized\.pdf|chapter-manifest\.json|document\.(?:pdf|epub|mobi|azw|azw3|fb2|docx|html|txt|md|webp|jpg|jpeg|png|gif|bmp|swf)|book\.epub|audio\.(?:mp3|wav|m4a|flac|mpga)|video\.(?:mp4|mov)|epub-chapters/(?:chapter-manifest\.json|chapters/chapter-[0-9]{4}\.xhtml|resources/[A-Za-z0-9._~%+\-/]+|epub-search-index\.json\.gz)))"
)
# Converted Reader assets are served from buckets; the former
# vomebook/Reader-Assets dataset is deleted and is not a Reader source.
VOICEOFML_READER_SOURCE_RE = re.compile(r"^/datasets/VoiceOfML/[A-Za-z0-9._-]+/(?:resolve|raw)/main/.+$")
UPSTREAM_REDIRECT_TTL_SECONDS = 300
UPSTREAM_REDIRECT_MAX_ENTRIES = 2000
# Statuses from a cached signed target that mean it expired or moved; the
# cached entry is evicted and the redirect chain is walked once from scratch.
STALE_REDIRECT_TARGET_STATUSES = frozenset({401, 403, 404, 410})
upstream_redirect_cache: dict[tuple[str, str], tuple[float, str]] = {}


def is_allowed_download_host(hostname: str | None) -> bool:
    return bool(hostname) and (
        hostname in ALLOWED_DOWNLOAD_HOSTS or hostname.endswith(ALLOWED_DOWNLOAD_HOST_SUFFIXES)
    )


def validate_download_url(url: str) -> str:
    parsed = urlparse(url)
    if (
        parsed.scheme != "https"
        or not is_allowed_download_host(parsed.hostname)
        or parsed.port not in (None, 443)
        or parsed.username is not None
        or parsed.password is not None
    ):
        raise ValueError(f"不允许的下载跳转地址: {parsed.hostname or 'unknown'}")
    return url


def normalize_download_url(url: str) -> str:
    parsed = urlparse(url)
    if parsed.hostname == "hf-mirror.com":
        return parsed._replace(netloc="huggingface.co").geturl()
    return url


def decode_url_path(path: str) -> str:
    decoded_path = path
    for _ in range(8):
        next_path = unquote(decoded_path)
        if next_path == decoded_path:
            return decoded_path
        decoded_path = next_path
    raise ValueError("不允许的下载来源路径")


def validate_path_safety(path: str) -> str:
    decoded_path = decode_url_path(path)
    if (
        "\\" in decoded_path
        or "\x00" in decoded_path
        or any(part in (".", "..") for part in decoded_path.split("/"))
    ):
        raise ValueError("不允许的下载来源路径")
    return decoded_path


def validate_voiceofml_source_url(url: str) -> str:
    url = normalize_download_url(url)
    validate_download_url(url)
    parsed = urlparse(url)
    decoded_path = validate_path_safety(parsed.path)
    if (
        parsed.hostname != "huggingface.co"
        or parsed.fragment
        or parsed.query
        or not decoded_path.startswith("/datasets/VoiceOfML/")
    ):
        raise ValueError("只允许 VoiceOfML 数据集文件")
    return url


def validate_reader_source_url(url: str) -> str:
    url = normalize_download_url(url)
    validate_download_url(url)
    parsed = urlparse(url)
    if parsed.hostname != "huggingface.co" or parsed.fragment or parsed.query:
        raise ValueError("不允许的阅读来源")
    decoded_path = validate_path_safety(parsed.path)
    if VOICEOFML_READER_SOURCE_RE.fullmatch(decoded_path):
        return url
    raise ValueError("不允许的阅读来源")


def validate_voiceofml_redirect_url(url: str) -> str:
    normalized = normalize_download_url(url)
    validate_download_url(normalized)
    if urlparse(normalized).hostname == "huggingface.co":
        parsed = urlparse(normalized)
        cache_source = reader_cache_source_path(parsed.path)
        if cache_source is not None and not parsed.fragment:
            # Hugging Face redirects existing files to commit-pinned cache
            # paths. Validate the mapped dataset source while preserving the
            # redirect URL and its cache metadata for the upstream request.
            validate_voiceofml_source_url("https://huggingface.co" + cache_source)
            return normalized
        return validate_voiceofml_source_url(normalized)
    if urlparse(normalized).hostname not in TRUSTED_REDIRECT_HOSTS:
        raise ValueError("上游重定向超出允许的存储范围")
    return normalized


def reader_cache_source_path(path: str) -> str | None:
    """Map HF's commit-pinned cache route back to its dataset source for validation."""
    match = re.fullmatch(
        r"/api/resolve-cache/datasets/([^/]+/[^/]+)/[0-9a-f]{40}/(.+)",
        validate_path_safety(path),
    )
    if match:
        return f"/datasets/{match[1]}/resolve/main/{match[2]}"
    return None


def validate_reader_redirect_url(url: str) -> str:
    normalized = normalize_download_url(url)
    validate_download_url(normalized)
    if urlparse(normalized).hostname == "huggingface.co":
        parsed = urlparse(normalized)
        cache_source = reader_cache_source_path(parsed.path)
        if cache_source is not None and not parsed.fragment:
            # Only redirects may use this route and its HF query metadata.
            # Apply the same dataset and path restrictions as the source.
            validate_reader_source_url("https://huggingface.co" + cache_source)
            return normalized
        return validate_reader_source_url(normalized)
    if urlparse(normalized).hostname not in TRUSTED_REDIRECT_HOSTS:
        raise ValueError("上游重定向超出允许的存储范围")
    return normalized


def validate_fixed_source_redirect_url(url: str, source_url: str) -> str:
    """Accept only the exact fixed source URL or a known signed-storage target."""
    validate_download_url(url)
    if url == source_url:
        return url
    parsed = urlparse(url)
    if parsed.hostname not in TRUSTED_REDIRECT_HOSTS or parsed.fragment:
        raise ValueError("上游重定向超出允许的存储范围")
    return url


def source_url_scope(url: str) -> str | None:
    parsed = urlparse(url)
    if parsed.hostname != "huggingface.co":
        return None
    decoded_path = reader_cache_source_path(parsed.path) or validate_path_safety(parsed.path)
    if decoded_path.startswith("/datasets/VoiceOfML/"):
        return "voiceofml"
    return None


def _redirect_cache_get(method: str, validated_url: str, url_validator, initial_scope: str | None) -> str | None:
    entry = upstream_redirect_cache.get((method, validated_url))
    if not entry:
        return None
    cached_at, target = entry
    if time.monotonic() - cached_at >= UPSTREAM_REDIRECT_TTL_SECONDS:
        upstream_redirect_cache.pop((method, validated_url), None)
        return None
    try:
        checked = url_validator(target)
    except ValueError:
        upstream_redirect_cache.pop((method, validated_url), None)
        return None
    next_scope = source_url_scope(checked)
    if initial_scope is not None and next_scope is not None and next_scope != initial_scope:
        upstream_redirect_cache.pop((method, validated_url), None)
        return None
    return checked


def _redirect_cache_put(method: str, validated_url: str, target: str) -> None:
    upstream_redirect_cache[(method, validated_url)] = (time.monotonic(), target)
    while len(upstream_redirect_cache) > UPSTREAM_REDIRECT_MAX_ENTRIES:
        oldest = min(upstream_redirect_cache, key=lambda key: upstream_redirect_cache[key][0])
        upstream_redirect_cache.pop(oldest, None)


async def _request(session, method, url, **options):
    request = getattr(session, "request", None)
    if request is not None:
        return await request(method, url, **options)
    request = getattr(session, method.lower(), None)
    if request is None:
        raise ValueError(f"上游会话不支持 {method} 请求")
    return await request(url, **options)


async def open_download_response(
    session,
    url: str,
    timeout: aiohttp.ClientTimeout,
    *,
    method: str = "GET",
    request_headers: dict[str, str] | None = None,
    url_validator=validate_download_url,
    redirect_validator=None,
    auto_decompress: bool | None = None,
    cache_redirects: bool = True,
):
    method = method.upper()
    if method not in ("GET", "HEAD"):
        raise ValueError("不支持的上游请求方法")
    validated_url = url_validator(url)
    redirect_validator = redirect_validator or url_validator
    initial_scope = source_url_scope(validated_url)
    headers = {"Accept-Encoding": "identity"}
    headers.update(request_headers or {})
    request_options = {} if auto_decompress is None else {"auto_decompress": auto_decompress}
    for attempt in range(2):
        current_url = validated_url
        cached = None
        if attempt == 0 and cache_redirects:
            cached = _redirect_cache_get(method, validated_url, redirect_validator, initial_scope)
            if cached is not None:
                current_url = cached
        walked = current_url != validated_url
        # Only a walk that started from a cached signed target may be retried;
        # a fresh walk's failure is returned or raised as-is.
        from_cache = cached is not None
        for _ in range(MAX_REDIRECTS + 1):
            try:
                response = await _request(
                    session, method, current_url, allow_redirects=False,
                    timeout=timeout, headers=headers, **request_options,
                )
            except (aiohttp.ClientError, asyncio.TimeoutError):
                if not from_cache:
                    raise
                upstream_redirect_cache.pop((method, validated_url), None)
                break
            if response.status not in (301, 302, 303, 307, 308):
                if response.status in STALE_REDIRECT_TARGET_STATUSES and from_cache:
                    upstream_redirect_cache.pop((method, validated_url), None)
                    response.release()
                    break
                # Reusing a signed target must not renew its fixed lifetime.
                if (cache_redirects and walked and current_url != cached
                        and response.status in (200, 206, 304)):
                    _redirect_cache_put(method, validated_url, current_url)
                return response
            location = response.headers.get("Location")
            response.release()
            if not location:
                raise ValueError("上游重定向缺少地址")
            next_url = redirect_validator(urljoin(current_url, location))
            next_scope = source_url_scope(next_url)
            if initial_scope is not None and next_scope is not None and next_scope != initial_scope:
                raise ValueError("上游重定向超出允许的数据集范围")
            current_url = next_url
            walked = True
        else:
            raise ValueError("上游重定向次数过多")

	"""Validated upstream requests and bounded redirect reuse, independent of API state."""
	import asyncio
	import re
	import time
	from urllib.parse import unquote, urljoin, urlparse

	import aiohttp

	ALLOWED_DOWNLOAD_HOSTS = {"huggingface.co", "hf-mirror.com", "hf.co"}
	ALLOWED_DOWNLOAD_HOST_SUFFIXES = (".hf.co", ".huggingface.co", ".xethub.hf.co")
	# Redirects are a narrower trust boundary than user-supplied source URLs. Keep
	# this exact-host list in sync with the signed storage targets we actually use;
	# accepting every future *.hf.co host would let a redirect escape the dataset
	# scope checked on the initial URL.
	TRUSTED_REDIRECT_HOSTS = frozenset({
	"cdn-lfs.hf.co", "cdn-lfs.huggingface.co", "cdn-lfs-us-1.hf.co",
	"cdn-lfs-us-1.huggingface.co", "us.aws.cdn.hf.co",
	"cas-bridge.xethub.hf.co", "cas-server.xethub.hf.co", "transfer.xethub.hf.co",
	})
	MAX_REDIRECTS = 5
	READER_ASSET_PATH_RE = re.compile(
	r"objects/[0-9a-f]{2}/[0-9a-f]{64}/(?:[0-9a-f]{16}/)?(?:page-manifest\.json\|ocr-manifest\.json\|pages/page-[0-9]{6}\.(?:webp\|jxl)\|ocr/(?:page-[0-9]{6}\.json\.gz\|book-text\.json\.gz)\|(?:[a-z0-9-]+/)?(?:linearized\.pdf\|chapter-manifest\.json\|document\.(?:pdf\|epub\|mobi\|azw\|azw3\|fb2\|docx\|html\|txt\|md\|webp\|jpg\|jpeg\|png\|gif\|bmp\|swf)\|book\.epub\|audio\.(?:mp3\|wav\|m4a\|flac\|mpga)\|video\.(?:mp4\|mov)\|epub-chapters/(?:chapter-manifest\.json\|chapters/chapter-[0-9]{4}\.xhtml\|resources/[A-Za-z0-9._~%+\-/]+\|epub-search-index\.json\.gz)))"
	)
	# Converted Reader assets are served from buckets; the former
	# vomebook/Reader-Assets dataset is deleted and is not a Reader source.
	VOICEOFML_READER_SOURCE_RE = re.compile(r"^/datasets/VoiceOfML/[A-Za-z0-9._-]+/(?:resolve\|raw)/main/.+$")
	UPSTREAM_REDIRECT_TTL_SECONDS = 300
	UPSTREAM_REDIRECT_MAX_ENTRIES = 2000
	# Statuses from a cached signed target that mean it expired or moved; the
	# cached entry is evicted and the redirect chain is walked once from scratch.
	STALE_REDIRECT_TARGET_STATUSES = frozenset({401, 403, 404, 410})
	upstream_redirect_cache: dict[tuple[str, str], tuple[float, str]] = {}


	def is_allowed_download_host(hostname: str \| None) -> bool:
	return bool(hostname) and (
	hostname in ALLOWED_DOWNLOAD_HOSTS or hostname.endswith(ALLOWED_DOWNLOAD_HOST_SUFFIXES)
	)


	def validate_download_url(url: str) -> str:
	parsed = urlparse(url)
	if (
	parsed.scheme != "https"
	or not is_allowed_download_host(parsed.hostname)
	or parsed.port not in (None, 443)
	or parsed.username is not None
	or parsed.password is not None
	):
	raise ValueError(f"不允许的下载跳转地址: {parsed.hostname or 'unknown'}")
	return url


	def normalize_download_url(url: str) -> str:
	parsed = urlparse(url)
	if parsed.hostname == "hf-mirror.com":
	return parsed._replace(netloc="huggingface.co").geturl()
	return url


	def decode_url_path(path: str) -> str:
	decoded_path = path
	for _ in range(8):
	next_path = unquote(decoded_path)
	if next_path == decoded_path:
	return decoded_path
	decoded_path = next_path
	raise ValueError("不允许的下载来源路径")


	def validate_path_safety(path: str) -> str:
	decoded_path = decode_url_path(path)
	if (
	"\\" in decoded_path
	or "\x00" in decoded_path
	or any(part in (".", "..") for part in decoded_path.split("/"))
	):
	raise ValueError("不允许的下载来源路径")
	return decoded_path


	def validate_voiceofml_source_url(url: str) -> str:
	url = normalize_download_url(url)
	validate_download_url(url)
	parsed = urlparse(url)
	decoded_path = validate_path_safety(parsed.path)
	if (
	parsed.hostname != "huggingface.co"
	or parsed.fragment
	or parsed.query
	or not decoded_path.startswith("/datasets/VoiceOfML/")
	):
	raise ValueError("只允许 VoiceOfML 数据集文件")
	return url


	def validate_reader_source_url(url: str) -> str:
	url = normalize_download_url(url)
	validate_download_url(url)
	parsed = urlparse(url)
	if parsed.hostname != "huggingface.co" or parsed.fragment or parsed.query:
	raise ValueError("不允许的阅读来源")
	decoded_path = validate_path_safety(parsed.path)
	if VOICEOFML_READER_SOURCE_RE.fullmatch(decoded_path):
	return url
	raise ValueError("不允许的阅读来源")


	def validate_voiceofml_redirect_url(url: str) -> str:
	normalized = normalize_download_url(url)
	validate_download_url(normalized)
	if urlparse(normalized).hostname == "huggingface.co":
	parsed = urlparse(normalized)
	cache_source = reader_cache_source_path(parsed.path)
	if cache_source is not None and not parsed.fragment:
	# Hugging Face redirects existing files to commit-pinned cache
	# paths. Validate the mapped dataset source while preserving the
	# redirect URL and its cache metadata for the upstream request.
	validate_voiceofml_source_url("https://huggingface.co" + cache_source)
	return normalized
	return validate_voiceofml_source_url(normalized)
	if urlparse(normalized).hostname not in TRUSTED_REDIRECT_HOSTS:
	raise ValueError("上游重定向超出允许的存储范围")
	return normalized


	def reader_cache_source_path(path: str) -> str \| None:
	"""Map HF's commit-pinned cache route back to its dataset source for validation."""
	match = re.fullmatch(
	r"/api/resolve-cache/datasets/([^/]+/[^/]+)/[0-9a-f]{40}/(.+)",
	validate_path_safety(path),
	)
	if match:
	return f"/datasets/{match[1]}/resolve/main/{match[2]}"
	return None


	def validate_reader_redirect_url(url: str) -> str:
	normalized = normalize_download_url(url)
	validate_download_url(normalized)
	if urlparse(normalized).hostname == "huggingface.co":
	parsed = urlparse(normalized)
	cache_source = reader_cache_source_path(parsed.path)
	if cache_source is not None and not parsed.fragment:
	# Only redirects may use this route and its HF query metadata.
	# Apply the same dataset and path restrictions as the source.
	validate_reader_source_url("https://huggingface.co" + cache_source)
	return normalized
	return validate_reader_source_url(normalized)
	if urlparse(normalized).hostname not in TRUSTED_REDIRECT_HOSTS:
	raise ValueError("上游重定向超出允许的存储范围")
	return normalized


	def validate_fixed_source_redirect_url(url: str, source_url: str) -> str:
	"""Accept only the exact fixed source URL or a known signed-storage target."""
	validate_download_url(url)
	if url == source_url:
	return url
	parsed = urlparse(url)
	if parsed.hostname not in TRUSTED_REDIRECT_HOSTS or parsed.fragment:
	raise ValueError("上游重定向超出允许的存储范围")
	return url


	def source_url_scope(url: str) -> str \| None:
	parsed = urlparse(url)
	if parsed.hostname != "huggingface.co":
	return None
	decoded_path = reader_cache_source_path(parsed.path) or validate_path_safety(parsed.path)
	if decoded_path.startswith("/datasets/VoiceOfML/"):
	return "voiceofml"
	return None


	def _redirect_cache_get(method: str, validated_url: str, url_validator, initial_scope: str \| None) -> str \| None:
	entry = upstream_redirect_cache.get((method, validated_url))
	if not entry:
	return None
	cached_at, target = entry
	if time.monotonic() - cached_at >= UPSTREAM_REDIRECT_TTL_SECONDS:
	upstream_redirect_cache.pop((method, validated_url), None)
	return None
	try:
	checked = url_validator(target)
	except ValueError:
	upstream_redirect_cache.pop((method, validated_url), None)
	return None
	next_scope = source_url_scope(checked)
	if initial_scope is not None and next_scope is not None and next_scope != initial_scope:
	upstream_redirect_cache.pop((method, validated_url), None)
	return None
	return checked


	def _redirect_cache_put(method: str, validated_url: str, target: str) -> None:
	upstream_redirect_cache[(method, validated_url)] = (time.monotonic(), target)
	while len(upstream_redirect_cache) > UPSTREAM_REDIRECT_MAX_ENTRIES:
	oldest = min(upstream_redirect_cache, key=lambda key: upstream_redirect_cache[key][0])
	upstream_redirect_cache.pop(oldest, None)


	async def _request(session, method, url, **options):
	request = getattr(session, "request", None)
	if request is not None:
	return await request(method, url, **options)
	request = getattr(session, method.lower(), None)
	if request is None:
	raise ValueError(f"上游会话不支持 {method} 请求")
	return await request(url, **options)


	async def open_download_response(
	session,
	url: str,
	timeout: aiohttp.ClientTimeout,
	*,
	method: str = "GET",
	request_headers: dict[str, str] \| None = None,
	url_validator=validate_download_url,
	redirect_validator=None,
	auto_decompress: bool \| None = None,
	cache_redirects: bool = True,
	):
	method = method.upper()
	if method not in ("GET", "HEAD"):
	raise ValueError("不支持的上游请求方法")
	validated_url = url_validator(url)
	redirect_validator = redirect_validator or url_validator
	initial_scope = source_url_scope(validated_url)
	headers = {"Accept-Encoding": "identity"}
	headers.update(request_headers or {})
	request_options = {} if auto_decompress is None else {"auto_decompress": auto_decompress}
	for attempt in range(2):
	current_url = validated_url
	cached = None
	if attempt == 0 and cache_redirects:
	cached = _redirect_cache_get(method, validated_url, redirect_validator, initial_scope)
	if cached is not None:
	current_url = cached
	walked = current_url != validated_url
	# Only a walk that started from a cached signed target may be retried;
	# a fresh walk's failure is returned or raised as-is.
	from_cache = cached is not None
	for _ in range(MAX_REDIRECTS + 1):
	try:
	response = await _request(
	session, method, current_url, allow_redirects=False,
	timeout=timeout, headers=headers, **request_options,
	)
	except (aiohttp.ClientError, asyncio.TimeoutError):
	if not from_cache:
	raise
	upstream_redirect_cache.pop((method, validated_url), None)
	break
	if response.status not in (301, 302, 303, 307, 308):
	if response.status in STALE_REDIRECT_TARGET_STATUSES and from_cache:
	upstream_redirect_cache.pop((method, validated_url), None)
	response.release()
	break
	# Reusing a signed target must not renew its fixed lifetime.
	if (cache_redirects and walked and current_url != cached
	and response.status in (200, 206, 304)):
	_redirect_cache_put(method, validated_url, current_url)
	return response
	location = response.headers.get("Location")
	response.release()
	if not location:
	raise ValueError("上游重定向缺少地址")
	next_url = redirect_validator(urljoin(current_url, location))
	next_scope = source_url_scope(next_url)
	if initial_scope is not None and next_scope is not None and next_scope != initial_scope:
	raise ValueError("上游重定向超出允许的数据集范围")
	current_url = next_url
	walked = True
	else:
	raise ValueError("上游重定向次数过多")