"""Small anon.legal client. Python 3.10+, standard library only. Keep returned keys private. Never put them in URLs or logs. Only GET requests are retried; create/delete failures may be ambiguous and must be reconciled by the caller. Redirects are refused so Authorization cannot follow another host. """ from __future__ import annotations import json import math import re import time from datetime import datetime, timezone from email.utils import parsedate_to_datetime from http.client import HTTPException from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener class ApiError(Exception): """API or transport failure. Exception text contains no keys or destination.""" def __init__(self, status: int | None, code: str, *, retry_after: float | None = None, may_have_succeeded: bool = False): self.status = status self.code = code self.retry_after = retry_after self.may_have_succeeded = may_have_succeeded super().__init__(f"anon.legal: {code}" + (f" (HTTP {status})" if status else "")) class _NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def _retry_after(value: str | None) -> float | None: if value is None: return None if re.fullmatch(r"\d+", value.strip()): try: return float(value) except (ValueError, OverflowError): # An enormous requested delay must not become an early fallback retry. return math.inf try: date = parsedate_to_datetime(value) if date.tzinfo is None: date = date.replace(tzinfo=timezone.utc) return max(0.0, (date - datetime.now(timezone.utc)).total_seconds()) except (ValueError, TypeError, OverflowError): return None class Client: def __init__(self, base_url: str = "https://api.anon.legal/v1", *, timeout: float = 10, retries: int = 2, max_retry_wait: float = 60): """timeout bounds each blocking socket operation, not the entire call. retries applies only to GET: transport errors and HTTP 429/503. A server Retry-After exceeding max_retry_wait is surfaced without an early retry. HTTP is accepted only for an explicitly configured loopback test server. """ parsed = urlsplit(base_url) if (parsed.username or parsed.password or parsed.query or parsed.fragment or not parsed.hostname or parsed.scheme not in ("http", "https") or (parsed.scheme == "http" and parsed.hostname not in ("localhost", "127.0.0.1", "::1"))): raise ValueError("Use an HTTPS API base URL without credentials or query parameters.") if not math.isfinite(timeout) or timeout <= 0: raise ValueError("timeout must be positive and finite.") if not isinstance(retries, int) or not 0 <= retries <= 5: raise ValueError("retries must be between 0 and 5.") if not math.isfinite(max_retry_wait) or not 0 <= max_retry_wait <= 300: raise ValueError("max_retry_wait must be between 0 and 300 seconds.") self.base_url = base_url.rstrip("/") self.timeout = timeout self.retries = retries self.max_retry_wait = max_retry_wait self._opener = build_opener(_NoRedirect()) def health(self) -> dict: return self._request("GET", "/health") def stats(self) -> dict: return self._request("GET", "/stats") def create(self, url: str, **settings) -> dict: """Creates once, without automatic retries. Save key from the result.""" return self._request("POST", "/links", {**settings, "url": url}) def inspect(self, link_id: str, delete_key: str) -> dict: """Does not open the destination or spend its remaining allowance.""" self._validate_key(link_id, delete_key) return self._request("GET", f"/links/{link_id}", key=delete_key) def delete(self, link_id: str, delete_key: str) -> None: """Deletes once. A 404 is reported as ApiError, including after a prior deletion.""" self._validate_key(link_id, delete_key) self._request("DELETE", f"/links/{link_id}", key=delete_key) @staticmethod def _validate_key(link_id: str, key: str): if not re.fullmatch(r"[A-Za-z0-9_-]{8,160}", link_id): raise ValueError("Invalid link ID.") if not re.fullmatch(r"[A-Za-z0-9_-]{43}", key): raise ValueError("Invalid key.") def _request(self, method: str, path: str, body=None, *, key: str | None = None): data = None if body is None else json.dumps(body, allow_nan=False).encode("utf-8") if data is not None and len(data) > 16384: raise ValueError("JSON body exceeds 16 KiB.") headers = {"Accept": "application/json", "User-Agent": "anon-legal-python/1.0"} if data is not None: headers["Content-Type"] = "application/json" if key: headers["Authorization"] = f"Bearer {key}" for attempt in range(self.retries + 1): failure = None request = Request(self.base_url + path, data=data, headers=headers, method=method) try: with self._opener.open(request, timeout=self.timeout) as result: raw = result.read(65537) if len(raw) > 65536: raise ApiError(result.status, "invalid_response", may_have_succeeded=method != "GET") if result.status == 204 and method == "DELETE": return None try: value = json.loads(raw) except (ValueError, UnicodeError): raise ApiError(result.status, "invalid_response", may_have_succeeded=method != "GET") from None if not isinstance(value, dict): raise ApiError(result.status, "invalid_response", may_have_succeeded=method != "GET") expected = 201 if method == "POST" else 200 if result.status != expected or (path == "/health" and value.get("ok") is not True): raise ApiError(result.status, "invalid_response", may_have_succeeded=method != "GET") return value except HTTPError as error: # Ignore server-provided messages: they may contain reflected credentials. codes = {400: "invalid_request", 401: "unauthorized", 404: "not_found", 405: "method_not_allowed", 429: "rate_limited", 503: "unavailable"} failure = ApiError(error.code, codes.get(error.code, "http_error"), retry_after=_retry_after(error.headers.get("Retry-After")), may_have_succeeded=method != "GET" and error.code >= 500) error.close() except (URLError, TimeoutError, OSError, HTTPException): failure = ApiError(None, "network_error", may_have_succeeded=method != "GET") if method != "GET" or attempt == self.retries or failure.status not in (None, 429, 503): raise failure from None delay = failure.retry_after if failure.retry_after is not None else 2 ** attempt if delay > self.max_retry_wait: raise failure from None time.sleep(delay) raise AssertionError("Unreachable")