66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
"""Validated HTTP helpers for release tooling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
from typing import Mapping
|
|
|
|
|
|
DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class HttpFetchResponse:
|
|
status: int
|
|
headers: dict[str, str]
|
|
body: bytes
|
|
|
|
def text(self, encoding: str = "utf-8") -> str:
|
|
return self.body.decode(encoding)
|
|
|
|
|
|
def validate_http_url(value: str, *, label: str = "URL") -> str:
|
|
parsed = urllib.parse.urlparse(str(value).strip())
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError(f"{label} must be an absolute HTTP(S) URL.")
|
|
if parsed.username or parsed.password:
|
|
raise ValueError(f"{label} must not include embedded credentials.")
|
|
return urllib.parse.urlunparse(parsed)
|
|
|
|
|
|
def fetch_http(
|
|
url: str,
|
|
*,
|
|
timeout: float,
|
|
label: str = "URL",
|
|
method: str = "GET",
|
|
headers: Mapping[str, str] | None = None,
|
|
max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
|
|
) -> HttpFetchResponse:
|
|
if max_response_bytes <= 0:
|
|
raise ValueError("max_response_bytes must be positive.")
|
|
request = urllib.request.Request(
|
|
validate_http_url(url, label=label),
|
|
headers=dict(headers or {}),
|
|
method=method,
|
|
)
|
|
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - URL is validated by validate_http_url. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
|
|
raw_length = response.headers.get("Content-Length")
|
|
if raw_length is not None:
|
|
try:
|
|
declared_length = int(raw_length)
|
|
except (TypeError, ValueError):
|
|
declared_length = None
|
|
if declared_length is not None and declared_length > max_response_bytes:
|
|
raise ValueError(f"{label} response exceeds the configured size limit.")
|
|
body = response.read(max_response_bytes + 1)
|
|
if len(body) > max_response_bytes:
|
|
raise ValueError(f"{label} response exceeds the configured size limit.")
|
|
return HttpFetchResponse(
|
|
status=int(getattr(response, "status", 0)),
|
|
headers=dict(response.headers.items()),
|
|
body=body,
|
|
)
|