booleanbeyond/jobfetch
0
1"""Safe async HTTP layer.2 3Everything the pipeline fetches goes through `SafeClient`. It enforces, per4scrape: a wall-clock deadline, a maximum number of requests, a maximum response5size, per-hop redirect validation, and optional robots.txt compliance.6 7The client is deliberately *not* a global singleton with unbounded state — one8`Budget` per scrape means a hostile or pathological site can't starve others.9"""10 11from __future__ import annotations12 13import asyncio14import logging15import os16import time17from dataclasses import dataclass, field18from typing import Optional19from urllib.parse import urljoin, urlsplit20from urllib.robotparser import RobotFileParser21 22import httpx23 24from .config import settings25from .security import UnsafeURLError, resolve_and_validate26 27log = logging.getLogger("jobfetch.net")28 29_PROXY_ENV = ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy")30 31 32def _proxy_in_use() -> bool:33 return any(os.getenv(k) for k in _PROXY_ENV)34 35 36class BudgetExceeded(RuntimeError):37 pass38 39 40class FetchError(RuntimeError):41 def __init__(self, message: str, *, status: int | None = None, url: str | None = None):42 super().__init__(message)43 self.status = status44 self.url = url45 46 47@dataclass48class Budget:49 """Per-scrape resource envelope."""50 51 deadline: float52 max_fetches: int = settings.max_fetches_per_scrape53 max_bytes: int = settings.max_response_bytes * 1254 fetches: int = 055 bytes_read: int = 056 truncated: bool = False57 58 @classmethod59 def start(cls, seconds: float | None = None) -> "Budget":60 return cls(deadline=time.monotonic() + (seconds or settings.total_budget_s))61 62 @property63 def remaining(self) -> float:64 return max(0.0, self.deadline - time.monotonic())65 66 def check(self) -> None:67 if self.remaining <= 0:68 raise BudgetExceeded("time budget exhausted")69 if self.fetches >= self.max_fetches:70 raise BudgetExceeded("request budget exhausted")71 if self.bytes_read >= self.max_bytes:72 raise BudgetExceeded("byte budget exhausted")73 74 def soft_ok(self) -> bool:75 try:76 self.check()77 return True78 except BudgetExceeded:79 self.truncated = True80 return False81 82 83@dataclass84class Response:85 url: str # final URL after redirects86 requested_url: str87 status: int88 headers: dict[str, str]89 content: bytes90 text: str91 elapsed_ms: int92 redirects: list[str] = field(default_factory=list)93 94 @property95 def content_type(self) -> str:96 return (self.headers.get("content-type") or "").split(";")[0].strip().lower()97 98 @property99 def is_html(self) -> bool:100 return "html" in self.content_type or self.content_type in ("", "text/plain")101 102 @property103 def is_json(self) -> bool:104 ct = self.content_type105 return "json" in ct or ct.endswith("+json")106 107 def json(self):108 import json as _json109 110 return _json.loads(self.text)111 112 113def _decode(content: bytes, headers: dict[str, str]) -> str:114 """Decode bytes to text without letting a bad charset blow up the request."""115 charset = None116 ct = headers.get("content-type", "")117 if "charset=" in ct:118 charset = ct.split("charset=", 1)[1].split(";")[0].strip().strip('"\'')119 for enc in filter(None, (charset, "utf-8", "cp1252", "latin-1")):120 try:121 return content.decode(enc)122 except (UnicodeDecodeError, LookupError):123 continue124 return content.decode("utf-8", errors="replace")125 126 127class SafeClient:128 """Async HTTP client with SSRF, size, count and time guards."""129 130 def __init__(self, budget: Budget, *, trace_id: str = "") -> None:131 self.budget = budget132 self.trace_id = trace_id133 self._sem = asyncio.Semaphore(settings.fetch_concurrency)134 self._robots: dict[str, Optional[RobotFileParser]] = {}135 self._robots_lock = asyncio.Lock()136 self._pin = settings.pin_dns and not _proxy_in_use()137 if settings.pin_dns and not self._pin:138 log.warning("DNS pinning disabled: an HTTP(S) proxy is configured")139 self._client = httpx.AsyncClient(140 follow_redirects=False, # we follow manually and re-validate each hop141 timeout=httpx.Timeout(142 settings.http_timeout_s,143 connect=settings.http_connect_timeout_s,144 ),145 headers={146 "User-Agent": settings.user_agent,147 "Accept-Language": "en-US,en;q=0.9",148 "Accept-Encoding": "gzip, deflate, br",149 },150 limits=httpx.Limits(max_connections=16, max_keepalive_connections=8),151 http2=False,152 )153 154 async def __aenter__(self) -> "SafeClient":155 return self156 157 async def __aexit__(self, *exc) -> None:158 await self.aclose()159 160 async def aclose(self) -> None:161 await self._client.aclose()162 163 # ---------------------------------------------------------------- robots164 165 async def _robots_for(self, origin: str) -> Optional[RobotFileParser]:166 if origin in self._robots:167 return self._robots[origin]168 async with self._robots_lock:169 if origin in self._robots:170 return self._robots[origin]171 parser: Optional[RobotFileParser] = None172 try:173 target = await resolve_and_validate(urljoin(origin, "/robots.txt"))174 resp = await self._raw_request("GET", target, timeout=settings.robots_timeout_s)175 if resp.status == 200 and len(resp.text) < 512_000:176 parser = RobotFileParser()177 parser.parse(resp.text.splitlines())178 except Exception as exc: # robots is best-effort, never fatal179 log.debug("robots fetch failed for %s: %s", origin, exc)180 parser = None181 self._robots[origin] = parser182 return parser183 184 async def robots_allows(self, url: str) -> bool:185 if not settings.respect_robots:186 return True187 parts = urlsplit(url)188 origin = f"{parts.scheme}://{parts.netloc}"189 parser = await self._robots_for(origin)190 if parser is None:191 return True # no robots.txt, or unreachable => allowed192 try:193 return parser.can_fetch(settings.user_agent, url)194 except Exception:195 return True196 197 # ---------------------------------------------------------------- fetch198 199 async def _raw_request(200 self,201 method: str,202 target,203 *,204 headers: dict[str, str] | None = None,205 json_body=None,206 content: bytes | None = None,207 timeout: float | None = None,208 ) -> Response:209 """Single hop, no redirect following, hard size cap."""210 req_headers = dict(headers or {})211 url = target.url212 extensions: dict = {}213 214 if self._pin:215 url = target.pinned_url216 req_headers["Host"] = (217 target.host if target.port in (80, 443) else f"{target.host}:{target.port}"218 )219 if target.scheme == "https":220 # Keeps SNI *and* certificate hostname verification on the real name.221 extensions["sni_hostname"] = target.host222 223 started = time.monotonic()224 remaining = self.budget.remaining225 if remaining <= 0:226 raise BudgetExceeded("time budget exhausted")227 eff_timeout = min(timeout or settings.http_timeout_s, max(1.0, remaining))228 229 buf = bytearray()230 try:231 req = self._client.build_request(232 method,233 url,234 headers=req_headers,235 json=json_body,236 content=content,237 timeout=httpx.Timeout(eff_timeout, connect=min(settings.http_connect_timeout_s, eff_timeout)),238 extensions=extensions or None,239 )240 resp = await self._client.send(req, stream=True)241 try:242 declared = resp.headers.get("content-length")243 if declared and declared.isdigit() and int(declared) > settings.max_response_bytes:244 raise FetchError(245 f"response too large ({declared} bytes)", status=resp.status_code, url=target.url246 )247 async for chunk in resp.aiter_bytes(65536):248 buf.extend(chunk)249 if len(buf) > settings.max_response_bytes:250 self.budget.truncated = True251 break252 finally:253 await resp.aclose()254 except httpx.TimeoutException as exc:255 raise FetchError(f"timeout fetching {target.host}", url=target.url) from exc256 except httpx.HTTPError as exc:257 raise FetchError(f"network error fetching {target.host}: {exc}", url=target.url) from exc258 259 data = bytes(buf)260 self.budget.fetches += 1261 self.budget.bytes_read += len(data)262 hdrs = {k.lower(): v for k, v in resp.headers.items()}263 return Response(264 url=target.url,265 requested_url=target.url,266 status=resp.status_code,267 headers=hdrs,268 content=data,269 text=_decode(data, hdrs),270 elapsed_ms=int((time.monotonic() - started) * 1000),271 )272 273 async def request(274 self,275 method: str,276 raw_url: str,277 *,278 headers: dict[str, str] | None = None,279 json_body=None,280 check_robots: bool = True,281 allow_status: tuple[int, ...] = (200,),282 max_redirects: int | None = None,283 ) -> Response:284 self.budget.check()285 limit = settings.max_redirects if max_redirects is None else max_redirects286 current = raw_url287 chain: list[str] = []288 289 async with self._sem:290 for hop in range(limit + 1):291 self.budget.check()292 target = await resolve_and_validate(current)293 294 if check_robots and not await self.robots_allows(target.url):295 raise FetchError(f"blocked by robots.txt: {target.url}", url=target.url)296 297 resp = await self._raw_request(298 method, target, headers=headers, json_body=json_body299 )300 resp.redirects = list(chain)301 302 if resp.status in (301, 302, 303, 307, 308):303 location = resp.headers.get("location")304 if not location:305 raise FetchError("redirect without Location", status=resp.status, url=current)306 nxt = urljoin(target.url, location)307 if nxt in chain or nxt == current:308 raise FetchError("redirect loop", url=current)309 chain.append(current)310 current = nxt311 # 303 and 302-on-POST degrade to GET, matching browser behaviour.312 if resp.status == 303 or (resp.status == 302 and method == "POST"):313 method, json_body = "GET", None314 continue315 316 if allow_status and resp.status not in allow_status:317 raise FetchError(318 f"unexpected status {resp.status}", status=resp.status, url=resp.url319 )320 resp.requested_url = raw_url321 return resp322 323 raise FetchError("too many redirects", url=raw_url)324 325 async def get(self, url: str, **kw) -> Response:326 return await self.request("GET", url, **kw)327 328 async def post_json(self, url: str, body, **kw) -> Response:329 kw.setdefault("headers", {})330 kw["headers"] = {**kw["headers"], "Content-Type": "application/json", "Accept": "application/json"}331 return await self.request("POST", url, json_body=body, **kw)332 333 async def try_get(self, url: str, **kw) -> Optional[Response]:334 """Fetch, returning None on any expected failure. Never raises for control flow."""335 if not self.budget.soft_ok():336 return None337 try:338 return await self.get(url, **kw)339 except (FetchError, UnsafeURLError, BudgetExceeded) as exc:340 log.debug("try_get(%s) -> %s", url, exc)341 return None342 except Exception as exc: # defensive: an extractor must never crash the run343 log.warning("try_get(%s) unexpected: %r", url, exc)344 return None345 346 async def try_post_json(self, url: str, body, **kw) -> Optional[Response]:347 if not self.budget.soft_ok():348 return None349 try:350 return await self.post_json(url, body, **kw)351 except (FetchError, UnsafeURLError, BudgetExceeded) as exc:352 log.debug("try_post_json(%s) -> %s", url, exc)353 return None354 except Exception as exc:355 log.warning("try_post_json(%s) unexpected: %r", url, exc)356 return None357