Spaces:
Runtime error
Runtime error
| """ | |
| HTTP client wrapper with sane defaults: retries, rate limiting, session | |
| persistence, CSRF token tracking. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| import time | |
| from typing import Any | |
| from urllib.parse import urljoin | |
| import requests | |
| from bs4 import BeautifulSoup, Tag | |
| from requests.adapters import HTTPAdapter | |
| from urllib3.util.retry import Retry | |
| from intelliscan.config import ( | |
| DEFAULT_TIMEOUT, | |
| DEFAULT_USER_AGENT, | |
| MAX_RETRIES, | |
| RATE_LIMIT_DELAY, | |
| ) | |
| log = logging.getLogger(__name__) | |
| class HttpClient: | |
| """A persistent HTTP session with retry, rate limit and CSRF token cache.""" | |
| def __init__( | |
| self, | |
| base_url: str, | |
| timeout: int = DEFAULT_TIMEOUT, | |
| user_agent: str = DEFAULT_USER_AGENT, | |
| verify_tls: bool = True, | |
| ) -> None: | |
| self.base_url = base_url.rstrip("/") | |
| self.timeout = timeout | |
| self.verify_tls = verify_tls | |
| self._last_token: str | None = None | |
| self.session = requests.Session() | |
| self.session.headers.update({"User-Agent": user_agent}) | |
| self.session.verify = verify_tls | |
| retry = Retry( | |
| total=MAX_RETRIES, | |
| backoff_factor=0.5, | |
| # 429 = Too Many Requests; 5xx = server errors | |
| status_forcelist=(429, 500, 502, 503, 504), | |
| allowed_methods=("GET", "POST", "HEAD"), | |
| respect_retry_after_header=True, | |
| ) | |
| adapter = HTTPAdapter(max_retries=retry, pool_connections=20, pool_maxsize=20) | |
| self.session.mount("http://", adapter) | |
| self.session.mount("https://", adapter) | |
| # ── core ────────────────────────────────────────────────────────────── | |
| def get(self, url: str, **kw: Any) -> requests.Response: | |
| return self._request("GET", url, **kw) | |
| def post(self, url: str, **kw: Any) -> requests.Response: | |
| return self._request("POST", url, **kw) | |
| def _request(self, method: str, url: str, **kw: Any) -> requests.Response: | |
| url = self._absolute(url) | |
| kw.setdefault("timeout", self.timeout) | |
| kw.setdefault("allow_redirects", True) | |
| time.sleep(RATE_LIMIT_DELAY) | |
| try: | |
| r = self.session.request(method, url, **kw) | |
| log.debug("%s %s -> %d (%d bytes)", method, url, r.status_code, len(r.content)) | |
| return r | |
| except requests.RequestException as e: | |
| log.warning("Request failed: %s %s -> %s", method, url, e) | |
| raise | |
| # ── helpers ─────────────────────────────────────────────────────────── | |
| def _absolute(self, url: str) -> str: | |
| if url.startswith(("http://", "https://")): | |
| return url.split("#")[0] # strip fragment | |
| return urljoin(self.base_url + "/", url.lstrip("/")).split("#")[0] | |
| def get_csrf_token(self, url: str, field: str = "user_token") -> str | None: | |
| """Fetch a page and extract a CSRF token from a hidden input.""" | |
| r = self.get(url) | |
| soup = BeautifulSoup(r.text, "html.parser") | |
| inp = soup.find("input", {"name": field}) | |
| if isinstance(inp, Tag) and inp.get("value"): | |
| self._last_token = str(inp["value"]) | |
| return self._last_token | |
| return None | |
| def last_csrf_token(self) -> str | None: | |
| return self._last_token | |
| def close(self) -> None: | |
| self.session.close() | |