From 2e4f98d4713409cbe79ad44aedf35efc2f75626c Mon Sep 17 00:00:00 2001 From: mtauha Date: Wed, 30 Sep 2026 20:39:13 +0500 Subject: [PATCH 1/3] feat: per-request PSX proxy passthrough via X-PSX-Proxy with SSRF hardening An optional X-PSX-Proxy header routes a request's PSX traffic, including the request token, through the caller's proxy using psxdata 1.2.0's proxy support. Off unless PSX_PROXY_PASSTHROUGH is enabled. Because the server connects to a caller-chosen address, the proxy must be http/socks5/socks5h with an explicit port (80, 443 or >= 1024), must resolve only to public addresses, and is pinned to the checked IP to defeat DNS rebinding. A 5s TCP check rejects dead proxies (502 proxy_unreachable). Proxied requests bypass the shared cache (X-Cache: BYPASS on /historical), are limited to 10/min per IP and 4 concurrent, and use a bounded client pool that also prunes psxdata's per-proxy token providers. Credentials are never logged or echoed. Bumps psxdata to psxdata[socks]==1.2.0 and the version to 0.4.0. --- CHANGELOG.md | 20 +++ README.md | 24 +++- api/main.py | 10 ++ api/proxy.py | 266 +++++++++++++++++++++++++++++++++++++++ api/routers/indices.py | 10 +- api/routers/market.py | 14 ++- api/routers/screener.py | 8 +- api/routers/sectors.py | 14 ++- api/routers/stocks.py | 49 +++++--- pyproject.toml | 4 +- tests/api/conftest.py | 9 ++ tests/api/test_proxy.py | 272 ++++++++++++++++++++++++++++++++++++++++ 12 files changed, 659 insertions(+), 41 deletions(-) create mode 100644 api/proxy.py create mode 100644 tests/api/test_proxy.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b30545a..3021e84 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,26 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm --- +## [0.4.0] — 2026-09-30 + +### Added + +- Per-request proxy passthrough. An optional `X-PSX-Proxy` header (`http://`, `socks5://` or `socks5h://`, with an explicit port and optional `user:pass@`) makes the API fetch that request's PSX data, including the request token, through the caller's proxy, using psxdata 1.2.0's proxy support. Off by default. Enable it with `PSX_PROXY_PASSTHROUGH=true`. +- Hardening against server-side request forgery (SSRF) for the proxy: + - the host must resolve only to public addresses, and the connection is pinned to the checked IP (defeats DNS rebinding); + - `https://` proxies and ports below 1024 (other than 80 and 443) are refused; + - a 5-second TCP check runs first (`502 proxy_unreachable` on failure); + - limits of 10 proxied requests per minute per IP and 4 concurrent server-wide; + - a bounded pool of proxied clients. +- Proxied requests never read or write the shared cache. `/historical` returns `X-Cache: BYPASS` for them. +- Proxy credentials are never logged or echoed in error messages. + +### Changed + +- Requires `psxdata[socks]==1.2.0` (was `psxdata==1.1.1`). + +--- + ## [0.3.0] — 2026-09-30 ### Added diff --git a/README.md b/README.md index 10eff01..64cbda5 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,7 @@ Every response wraps its payload in a consistent envelope. | 404 | `not_found` | Symbol or index does not exist | | 429 | `rate_limited` | Exceeded 60 requests/minute per IP | | 502 | `upstream_data_error` | Upstream PSX data failed validation | +| 502 | `proxy_unreachable` | The `X-PSX-Proxy` you sent did not accept a connection | | 503 | `psx_unavailable` | PSX website unreachable, or PSX is rate-limiting the API (sent with a `Retry-After` header) | | 500 | `internal_error` | Unexpected server error | @@ -139,13 +140,33 @@ Every `/historical` response says where it came from: | Header | Values | | ------ | ------ | -| `X-Cache` | `HIT` — fresh cached copy · `MISS` — fetched from PSX just now · `STALE` — PSX refused or was unreachable, so the last cached copy was served | +| `X-Cache` | `HIT` — fresh cached copy · `MISS` — fetched from PSX just now · `STALE` — PSX refused or was unreachable, so the last cached copy was served · `BYPASS` — fetched through your `X-PSX-Proxy`, cache not used | | `Age` | Seconds since the data was fetched from PSX (on `HIT` and `STALE`) | `meta.cached` is `true` for `HIT` and `STALE`. If PSX is rate-limiting and no cached copy exists, the API returns `503 psx_unavailable` with `Retry-After: 60`. --- +## Proxy Passthrough + +Send an `X-PSX-Proxy` header to have the API fetch that request's PSX data through your own proxy, using psxdata's [proxy support](https://psxdata.mintlify.app/sdk/guides/proxy). The request token is fetched through the same proxy. + +```bash +curl -H "X-PSX-Proxy: http://user:pass@proxy.example.com:8080" https://psxdata-api.fastapicloud.dev/stocks/ENGRO/quote +``` + +The server connects to an address you choose, so the proxy is checked before use: + +- **Only when enabled.** The server operator has to turn it on (`PSX_PROXY_PASSTHROUGH`). Otherwise the header is rejected with `400`. +- **Schemes:** `http://`, `socks5://` or `socks5h://`, with an explicit port (80, 443, or 1024–65535) and optional `user:pass@`. `https://` proxies are not accepted. +- **Public addresses only.** The proxy host must resolve only to public internet addresses. Loopback, private, link-local (including cloud metadata), CGNAT and multicast addresses are rejected. The connection is pinned to the checked IP, so the name cannot be re-pointed afterwards. +- **Quick reachability check.** A proxy that doesn't accept a TCP connection within 5 seconds returns `502 proxy_unreachable`. +- **No cache.** Proxied requests never read or write the API's cache. `/historical` answers with `X-Cache: BYPASS`. +- **Stricter limits.** 10 proxied requests per minute per IP and at most 4 in progress server-wide (`429` beyond that), on top of the normal limit. +- **Credentials are never logged or echoed back.** Send the proxy only in the header, never in the URL. + +--- + ## Configuration All settings are optional environment variables. @@ -154,6 +175,7 @@ All settings are optional environment variables. | -------- | ------- | ----------- | | `REDIS_URL` | unset | Redis-compatible server for the `/historical` cache, e.g. an [Aiven for Valkey](https://aiven.io/valkey) service URI (`rediss://default:@:`). Keeps the cache across restarts and instances. When unset, or when the server is unreachable, the API falls back to an in-memory cache and keeps working. | | `HISTORICAL_CACHE_MARKET_TTL` | `1800` | How long, in seconds, `/historical` data stays fresh during PSX trading hours. | +| `PSX_PROXY_PASSTHROUGH` | off | Set to `true` to honour the per-request `X-PSX-Proxy` header (see [Proxy Passthrough](#proxy-passthrough)). | ```bash docker run -p 8000:8000 -e REDIS_URL="rediss://default:@:" mtauha/psxdata-api diff --git a/api/main.py b/api/main.py index 3b4e6b0..e22a22d 100644 --- a/api/main.py +++ b/api/main.py @@ -20,6 +20,7 @@ from api.cache.factory import build_historical_service from api.cache.historical import COOLDOWN_MESSAGE, PSX_COOLDOWN_SECONDS from api.dependencies import limiter +from api.proxy import ProxyPassthrough, ProxyUnreachableError from api.routers import router_registry logger = logging.getLogger("api") @@ -30,6 +31,7 @@ async def lifespan(app: FastAPI): """Build the /historical cache on startup and close its backend on shutdown.""" service = build_historical_service(os.environ) app.state.historical_service = service + app.state.proxy_passthrough = ProxyPassthrough.from_env(os.environ) try: yield finally: @@ -92,6 +94,14 @@ async def psx_rate_limit_handler(request: Request, exc: PSXRateLimitError) -> JS ) +@app.exception_handler(ProxyUnreachableError) +async def proxy_unreachable_handler(request: Request, exc: ProxyUnreachableError) -> JSONResponse: + return JSONResponse( + status_code=502, + content={"error": {"status": 502, "code": "proxy_unreachable", "message": str(exc)}}, + ) + + @app.exception_handler(InvalidSymbolError) async def invalid_symbol_handler(request: Request, exc: InvalidSymbolError) -> JSONResponse: return JSONResponse( diff --git a/api/proxy.py b/api/proxy.py new file mode 100644 index 0000000..2587860 --- /dev/null +++ b/api/proxy.py @@ -0,0 +1,266 @@ +"""Per-request PSX proxy passthrough (``X-PSX-Proxy`` header) with SSRF hardening. + +A caller may send ``X-PSX-Proxy: `` to have this server fetch PSX data +through the caller's proxy. The server then opens a connection to an address +the caller chose, so every proxy is checked before use: + +- The feature is off unless ``PSX_PROXY_PASSTHROUGH`` is set to a true value. +- Only ``http``, ``socks5`` and ``socks5h`` proxies. ``https`` proxies are + refused: TLS to the proxy verifies its hostname, which rules out IP pinning. +- An explicit port is required: 80, 443, or 1024-65535. +- Every address the host resolves to must be public (globally routable), and + the connection is pinned to the checked IP, so DNS rebinding cannot point it + somewhere else between the check and the request. +- A short TCP connect check rejects dead proxies before any PSX request. +- Proxied requests never read or write a shared cache. +- Per-IP rate limit, a global concurrency cap, and a bounded pool of clients. + +Credentials in the proxy URL are never logged or echoed back. +""" +from __future__ import annotations + +import ipaddress +import os +import socket +import tempfile +import threading +from collections import OrderedDict +from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager +from typing import Any +from urllib.parse import urlsplit + +import psxdata +from fastapi import Header, HTTPException, Request +from limits import RateLimitItemPerMinute +from limits.storage import MemoryStorage +from limits.strategies import MovingWindowRateLimiter +from psxdata import PSXClient +from psxdata.proxy import normalize_proxy +from psxdata.scrapers import token as token_module +from slowapi.util import get_remote_address + +PROXY_HEADER = "X-PSX-Proxy" +ENABLE_ENV = "PSX_PROXY_PASSTHROUGH" +ALLOWED_SCHEMES = frozenset({"http", "socks5", "socks5h"}) +MAX_PROXY_URL_LENGTH = 2048 +CONNECT_TIMEOUT = 5.0 +MAX_CLIENTS = 32 +MAX_CONCURRENT = 4 +PER_IP_PER_MINUTE = 10 + +Resolver = Callable[[str, int], list[str]] +Connector = Callable[[str, int], None] + + +class ProxyUnreachableError(Exception): + """The caller's proxy passed validation but did not accept a TCP connection.""" + + +def _resolve(host: str, port: int) -> list[str]: + return [str(info[4][0]) for info in socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)] + + +def _connect(host: str, port: int) -> None: + socket.create_connection((host, port), timeout=CONNECT_TIMEOUT).close() + + +def _is_public(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + if not ip.is_global or ip.is_multicast: + return False + if isinstance(ip, ipaddress.IPv6Address): + # IPv4 addresses embedded in IPv6 must be public too + embedded = [ip.ipv4_mapped, ip.sixtofour, *(ip.teredo or ())] + return all(_is_public(e) for e in embedded if e is not None) + return True + + +def _reject(message: str) -> HTTPException: + return HTTPException(status_code=400, detail=f"{PROXY_HEADER}: {message}") + + +def pin_proxy(raw: str, resolver: Resolver = _resolve) -> str: + """Validate a caller-supplied proxy URL and return it pinned to a checked IP. + + Raises: + HTTPException: 400 for any rejected proxy. The message never includes + the URL, so credentials are not echoed back. + """ + raw = raw.strip() + if not raw or len(raw) > MAX_PROXY_URL_LENGTH: + raise _reject("must be a proxy URL of at most 2048 characters") + parts = urlsplit(raw) + scheme = parts.scheme.lower() + if scheme not in ALLOWED_SCHEMES: + raise _reject("scheme must be http://, socks5:// or socks5h://") + if parts.path not in ("", "/") or parts.query or parts.fragment: + raise _reject("must not contain a path, query or fragment") + try: + port = parts.port + except ValueError: + raise _reject("invalid port") from None + if port is None: + raise _reject("must include an explicit port") + if not (port in (80, 443) or 1024 <= port <= 65535): + raise _reject("port must be 80, 443, or between 1024 and 65535") + host = parts.hostname + if not host: + raise _reject("must include a host") + + try: + ips = {ipaddress.ip_address(addr.split("%", 1)[0]) for addr in resolver(host, port)} + except (OSError, UnicodeError, ValueError): + raise _reject("host could not be resolved") from None + if not ips or not all(_is_public(ip) for ip in ips): + raise _reject("host must resolve only to public internet addresses") + + ip = min(ips, key=lambda a: (a.version, int(a))) # prefer IPv4 + host_part = f"[{ip}]" if ip.version == 6 else str(ip) + userinfo = parts.netloc.rpartition("@")[0] + "@" if "@" in parts.netloc else "" + pinned = f"{scheme}://{userinfo}{host_part}:{port}" + try: + normalize_proxy(pinned) + except (ValueError, TypeError): + raise _reject("invalid proxy URL") from None + except ImportError: + raise _reject("SOCKS proxies are not available on this server") from None + return pinned + + +def _proxy_key(pinned: str) -> frozenset[tuple[str, str]]: + # Same key psxdata uses for its per-proxy token providers + return frozenset((normalize_proxy(pinned) or {}).items()) + + +class PsxSource: + """Where a request's PSX data comes from: the shared default client, or a proxied one. + + Proxied calls always pass ``cache=False``, so a caller's proxy can never + read from or write to the cache shared by all callers. + """ + + def __init__(self, client: PSXClient | None = None) -> None: + self._client = client + + @property + def proxied(self) -> bool: + return self._client is not None + + def fetch(self, name: str, *args: Any, **kwargs: Any) -> Any: + if self._client is None: + return getattr(psxdata, name)(*args, **kwargs) + return getattr(self._client, name)(*args, cache=False, **kwargs) + + +class ProxyPassthrough: + """Validates caller proxies and hands out bounded, pooled proxied clients.""" + + def __init__( + self, + enabled: bool, + *, + resolver: Resolver = _resolve, + connector: Connector = _connect, + max_clients: int = MAX_CLIENTS, + max_concurrent: int = MAX_CONCURRENT, + per_ip_per_minute: int = PER_IP_PER_MINUTE, + ) -> None: + self.enabled = enabled + self._resolver = resolver + self._connector = connector + self._max_clients = max_clients + self._clients: OrderedDict[str, PSXClient] = OrderedDict() + self._lock = threading.Lock() + self._slots = threading.BoundedSemaphore(max_concurrent) + self._rate_limiter = MovingWindowRateLimiter(MemoryStorage()) + self._rate = RateLimitItemPerMinute(per_ip_per_minute) + self._cache_dir: str | None = None + + @classmethod + def from_env(cls, env: Mapping[str, str]) -> ProxyPassthrough: + enabled = env.get(ENABLE_ENV, "").strip().lower() in {"1", "true", "yes", "on"} + return cls(enabled) + + @contextmanager + def acquire(self, request: Request, raw: str) -> Iterator[PsxSource]: + """Validate *raw*, reserve a concurrency slot, and yield a proxied source.""" + if not self.enabled: + raise HTTPException( + status_code=400, detail=f"{PROXY_HEADER} is not enabled on this server" + ) + # Rate-limit before DNS resolution so the check itself cannot be spammed + if not self._rate_limiter.hit(self._rate, "psx-proxy", get_remote_address(request)): + raise HTTPException(status_code=429, detail="Too many proxied requests") + pinned = pin_proxy(raw, self._resolver) + if not self._slots.acquire(blocking=False): + raise HTTPException( + status_code=429, detail="Too many proxied requests in progress; retry shortly" + ) + try: + parts = urlsplit(pinned) + try: + self._connector(parts.hostname or "", parts.port or 0) + except OSError: + raise ProxyUnreachableError( + f"{PROXY_HEADER}: proxy did not accept a connection" + ) from None + yield PsxSource(self._client_for(pinned)) + finally: + self._slots.release() + + def _client_for(self, pinned: str) -> PSXClient: + with self._lock: + client = self._clients.get(pinned) + if client is not None: + self._clients.move_to_end(pinned) + return client + if self._cache_dir is None: + # Never touched (proxied calls use cache=False), but kept apart from the shared one + self._cache_dir = tempfile.mkdtemp(prefix="psxdata-proxied-") + client = PSXClient(cache_dir=self._cache_dir, proxy=pinned) + self._clients[pinned] = client + if len(self._clients) > self._max_clients: + self._clients.popitem(last=False) + self._prune_token_providers() + return client + + def _prune_token_providers(self) -> None: + """Drop psxdata's per-proxy token providers for clients no longer pooled. + + psxdata keeps one provider per proxy for the life of the process; on a + server fed arbitrary proxies that would grow without bound. + """ + live = {_proxy_key(url) for url in self._clients} + with token_module._default_lock: + for key in [k for k in token_module._proxy_providers if k not in live]: + del token_module._proxy_providers[key] + + +def get_proxy_passthrough(request: Request) -> ProxyPassthrough: + """Return the app's ProxyPassthrough; build one from the environment if lifespan did not run.""" + passthrough: ProxyPassthrough | None = getattr(request.app.state, "proxy_passthrough", None) + if passthrough is None: + passthrough = ProxyPassthrough.from_env(os.environ) + request.app.state.proxy_passthrough = passthrough + return passthrough + + +def psx_source( + request: Request, + x_psx_proxy: str | None = Header( + default=None, + alias=PROXY_HEADER, + description=( + "Optional proxy for this request's PSX traffic: http://, socks5:// or socks5h://, " + "with an explicit port and optional user:pass@. Must resolve to a public address. " + "Proxied requests bypass the cache and have stricter limits. Only honoured when " + "the server enables proxy passthrough." + ), + ), +) -> Iterator[PsxSource]: + """FastAPI dependency: the PSX data source for this request.""" + if x_psx_proxy is None: + yield PsxSource() + return + with get_proxy_passthrough(request).acquire(request, x_psx_proxy) as source: + yield source diff --git a/api/routers/indices.py b/api/routers/indices.py index 2cc8330..dc5596c 100644 --- a/api/routers/indices.py +++ b/api/routers/indices.py @@ -4,12 +4,12 @@ from datetime import datetime, timezone import pandas as pd -import psxdata -from fastapi import APIRouter, HTTPException, Request +from fastapi import APIRouter, Depends, HTTPException, Request from psxdata.constants import INDEX_NAMES from psxdata.exceptions import PSXParseError from api.dependencies import limiter +from api.proxy import PsxSource, psx_source from api.schemas import ( IndexConstituentResponse, IndexConstituentRow, @@ -36,9 +36,11 @@ def list_indices(request: Request) -> StringListResponse: @router.get("/indices/{name}", response_model=IndexConstituentResponse) @limiter.limit("60/minute") -def get_index(request: Request, name: str) -> IndexConstituentResponse: +def get_index( + request: Request, name: str, psx: PsxSource = Depends(psx_source) +) -> IndexConstituentResponse: try: - df = psxdata.indices(name.upper()) + df = psx.fetch("indices", name.upper()) except PSXParseError: raise HTTPException(status_code=404, detail=f"Index {name.upper()} not found") diff --git a/api/routers/market.py b/api/routers/market.py index 730820a..d44cfbd 100644 --- a/api/routers/market.py +++ b/api/routers/market.py @@ -5,10 +5,10 @@ from typing import Any import pandas as pd -import psxdata -from fastapi import APIRouter, Request +from fastapi import APIRouter, Depends, Request from api.dependencies import limiter +from api.proxy import PsxSource, psx_source from api.schemas import MarketTablesResponse, MetaSingle router = APIRouter(tags=["market"]) @@ -36,8 +36,8 @@ def _serialize_tables(tables: dict[str, pd.DataFrame]) -> dict[str, list[dict[st @router.get("/debt-market", response_model=MarketTablesResponse) @limiter.limit("60/minute") -def get_debt_market(request: Request) -> MarketTablesResponse: - tables = psxdata.debt_market() +def get_debt_market(request: Request, psx: PsxSource = Depends(psx_source)) -> MarketTablesResponse: + tables = psx.fetch("debt_market") return MarketTablesResponse( data=_serialize_tables(tables), meta=MetaSingle(timestamp=_now_iso(), cached=False), @@ -46,8 +46,10 @@ def get_debt_market(request: Request) -> MarketTablesResponse: @router.get("/eligible-scrips", response_model=MarketTablesResponse) @limiter.limit("60/minute") -def get_eligible_scrips(request: Request) -> MarketTablesResponse: - tables = psxdata.eligible_scrips() +def get_eligible_scrips( + request: Request, psx: PsxSource = Depends(psx_source) +) -> MarketTablesResponse: + tables = psx.fetch("eligible_scrips") return MarketTablesResponse( data=_serialize_tables(tables), meta=MetaSingle(timestamp=_now_iso(), cached=False), diff --git a/api/routers/screener.py b/api/routers/screener.py index b12f896..3b13d84 100644 --- a/api/routers/screener.py +++ b/api/routers/screener.py @@ -4,10 +4,10 @@ from datetime import datetime, timezone import pandas as pd -import psxdata -from fastapi import APIRouter, Request +from fastapi import APIRouter, Depends, Request from api.dependencies import limiter +from api.proxy import PsxSource, psx_source from api.schemas import MetaList, ScreenerResponse, ScreenerRow router = APIRouter(tags=["screener"]) @@ -19,8 +19,8 @@ def _now_iso() -> str: @router.get("/screener", response_model=ScreenerResponse) @limiter.limit("60/minute") -def list_screener(request: Request) -> ScreenerResponse: - df = psxdata.screener() +def list_screener(request: Request, psx: PsxSource = Depends(psx_source)) -> ScreenerResponse: + df = psx.fetch("screener") rows: list[ScreenerRow] = [] if not df.empty: df = df.where(pd.notna(df), other=None) diff --git a/api/routers/sectors.py b/api/routers/sectors.py index 487493a..2b2cf3c 100644 --- a/api/routers/sectors.py +++ b/api/routers/sectors.py @@ -4,10 +4,10 @@ from datetime import datetime, timezone import pandas as pd -import psxdata -from fastapi import APIRouter, Request +from fastapi import APIRouter, Depends, Request from api.dependencies import limiter +from api.proxy import PsxSource, psx_source from api.schemas import ( MetaList, SectorRow, @@ -24,8 +24,8 @@ def _now_iso() -> str: @router.get("/sectors", response_model=SectorsResponse) @limiter.limit("60/minute") -def list_sectors(request: Request) -> SectorsResponse: - df = psxdata.sectors() +def list_sectors(request: Request, psx: PsxSource = Depends(psx_source)) -> SectorsResponse: + df = psx.fetch("sectors") rows: list[SectorRow] = [] if not df.empty: df = df.where(pd.notna(df), other=None) @@ -41,8 +41,10 @@ def list_sectors(request: Request) -> SectorsResponse: @router.get("/sectors/{name}/stocks", response_model=StringListResponse) @limiter.limit("60/minute") -def get_sector_stocks(request: Request, name: str) -> StringListResponse: - df = psxdata.symbols() +def get_sector_stocks( + request: Request, name: str, psx: PsxSource = Depends(psx_source) +) -> StringListResponse: + df = psx.fetch("symbols") tickers: list[str] = [] if not df.empty and "sector_name" in df.columns and "symbol" in df.columns: matched = df[df["sector_name"].str.upper() == name.upper()] diff --git a/api/routers/stocks.py b/api/routers/stocks.py index b277a42..73e3930 100644 --- a/api/routers/stocks.py +++ b/api/routers/stocks.py @@ -7,10 +7,11 @@ import pandas as pd import psxdata -from fastapi import APIRouter, HTTPException, Request, Response +from fastapi import APIRouter, Depends, HTTPException, Request, Response from api.cache.historical import CacheStatus from api.dependencies import get_cache, limiter +from api.proxy import PsxSource, psx_source from api.schemas import ( ErrorEnvelope, FundamentalsResponse, @@ -45,9 +46,10 @@ def _df_to_records(df: pd.DataFrame) -> list[dict]: 200: { "headers": { "X-Cache": { - "description": "HIT (fresh cache), MISS (fetched from PSX), or STALE " - "(PSX refused or was unreachable; last cached copy served)", - "schema": {"type": "string", "enum": ["HIT", "MISS", "STALE"]}, + "description": "HIT (fresh cache), MISS (fetched from PSX), STALE " + "(PSX refused or was unreachable; last cached copy served), or BYPASS " + "(fetched through the caller's X-PSX-Proxy; cache not used)", + "schema": {"type": "string", "enum": ["HIT", "MISS", "STALE", "BYPASS"]}, }, "Age": { "description": "Seconds since the data was fetched from PSX (HIT and STALE only)", @@ -89,8 +91,10 @@ def _fetch_full_history(symbol: str) -> list[dict]: @router.get("/stocks", response_model=StringListResponse) @limiter.limit("60/minute") -def list_stocks(request: Request, index: str | None = None) -> StringListResponse: - tickers = psxdata.tickers(index=index) +def list_stocks( + request: Request, index: str | None = None, psx: PsxSource = Depends(psx_source) +) -> StringListResponse: + tickers = psx.fetch("tickers", index=index) return StringListResponse( data=tickers, meta=MetaList(timestamp=_now_iso(), cached=False, count=len(tickers)), @@ -109,27 +113,34 @@ def get_historical( symbol: str, start: str | None = None, end: str | None = None, + psx: PsxSource = Depends(psx_source), ) -> HistoricalResponse: start_date = _parse_date("start", start) end_date = _parse_date("end", end) if start_date and end_date and start_date > end_date: raise HTTPException(status_code=422, detail="start must not be after end") - result = get_cache(request).get(symbol.upper(), _fetch_full_history) + if psx.proxied: + # A caller's proxy must never feed or read the shared cache + all_rows = _df_to_records(psx.fetch("stocks", symbol.upper())) + cached = False + response.headers["X-Cache"] = "BYPASS" + else: + result = get_cache(request).get(symbol.upper(), _fetch_full_history) + all_rows = result.rows + cached = result.status is not CacheStatus.MISS + response.headers["X-Cache"] = result.status.value + if cached: + age = (datetime.now(timezone.utc) - result.fetched_at).total_seconds() + response.headers["Age"] = str(max(0, int(age))) lo = start_date.isoformat() if start_date else None hi = end_date.isoformat() if end_date else None rows = [ OHLCVRow.model_validate(r) - for r in result.rows + for r in all_rows if (lo is None or r["date"] >= lo) and (hi is None or r["date"] <= hi) ] - - cached = result.status is not CacheStatus.MISS - response.headers["X-Cache"] = result.status.value - if cached: - age = (datetime.now(timezone.utc) - result.fetched_at).total_seconds() - response.headers["Age"] = str(max(0, int(age))) return HistoricalResponse( data=rows, meta=MetaList(timestamp=_now_iso(), cached=cached, count=len(rows)), @@ -138,8 +149,8 @@ def get_historical( @router.get("/stocks/{symbol}/quote", response_model=QuoteResponse) @limiter.limit("60/minute") -def get_quote(request: Request, symbol: str) -> QuoteResponse: - df = psxdata.quote(symbol.upper()) +def get_quote(request: Request, symbol: str, psx: PsxSource = Depends(psx_source)) -> QuoteResponse: + df = psx.fetch("quote", symbol.upper()) if df.empty: raise HTTPException(status_code=404, detail=f"{symbol.upper()} not found") row = _df_to_records(df)[0] @@ -152,8 +163,10 @@ def get_quote(request: Request, symbol: str) -> QuoteResponse: @router.get("/stocks/{symbol}/fundamentals", response_model=FundamentalsResponse) @limiter.limit("60/minute") -def get_fundamentals(request: Request, symbol: str) -> FundamentalsResponse: - df = psxdata.fundamentals(symbol=symbol.upper()) +def get_fundamentals( + request: Request, symbol: str, psx: PsxSource = Depends(psx_source) +) -> FundamentalsResponse: + df = psx.fetch("fundamentals", symbol=symbol.upper()) rows: list[FundamentalsRow] = [] if not df.empty: rows = [FundamentalsRow.model_validate(r) for r in _df_to_records(df)] diff --git a/pyproject.toml b/pyproject.toml index 88b4c37..e45dc90 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta" [project] name = "psxdata-api" -version = "0.3.0" +version = "0.4.0" description = "FastAPI REST service for Pakistan Stock Exchange (PSX) data" readme = "README.md" requires-python = ">=3.11" license = {file = "LICENSE.md"} dependencies = [ - "psxdata==1.1.1", + "psxdata[socks]==1.2.0", "fastapi[standard]>=0.100", "uvicorn[standard]>=0.23", "slowapi>=0.1", diff --git a/tests/api/conftest.py b/tests/api/conftest.py index 57a0c68..c159cc0 100644 --- a/tests/api/conftest.py +++ b/tests/api/conftest.py @@ -4,6 +4,15 @@ from api.cache.historical import HistoricalService from api.cache.store import MemoryLRU, RedisStore, TieredStore from api.main import app +from api.proxy import ProxyPassthrough + + +@pytest.fixture(autouse=True) +def proxy_passthrough_disabled() -> ProxyPassthrough: + """Start every test with proxy passthrough off and an empty client pool.""" + passthrough = ProxyPassthrough(False) + app.state.proxy_passthrough = passthrough + return passthrough @pytest.fixture(autouse=True) diff --git a/tests/api/test_proxy.py b/tests/api/test_proxy.py new file mode 100644 index 0000000..48bf22a --- /dev/null +++ b/tests/api/test_proxy.py @@ -0,0 +1,272 @@ +"""Tests for X-PSX-Proxy passthrough and its SSRF hardening — no network.""" +from __future__ import annotations + +from unittest.mock import patch + +import pandas as pd +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient +from psxdata import PSXClient +from psxdata.proxy import normalize_proxy +from psxdata.scrapers import token as token_module + +from api.main import app +from api.proxy import PROXY_HEADER, ProxyPassthrough, pin_proxy + +PUBLIC_IP = "93.184.216.34" +PROXY = "http://alice:s3cret@proxy.example.com:8080" +PINNED = f"http://alice:s3cret@{PUBLIC_IP}:8080" + + +def resolver_for(*addrs: str): + return lambda host, port: list(addrs) + + +def no_connect(host: str, port: int) -> None: + return None + + +@pytest.fixture +def client() -> TestClient: + return TestClient(app) + + +@pytest.fixture +def enabled() -> ProxyPassthrough: + passthrough = ProxyPassthrough(True, resolver=resolver_for(PUBLIC_IP), connector=no_connect) + app.state.proxy_passthrough = passthrough + return passthrough + + +def _quote_df() -> pd.DataFrame: + return pd.DataFrame([{"symbol": "ENGRO", "sector": "Fertilizer", "price": 300.0}]) + + +# --------------------------------------------------------------------------- +# pin_proxy — validation and SSRF checks +# --------------------------------------------------------------------------- + +class TestPinProxy: + def test_pins_hostname_to_resolved_public_ip(self): + assert pin_proxy(PROXY, resolver_for(PUBLIC_IP)) == PINNED + + def test_socks5h_without_credentials(self): + assert pin_proxy("socks5h://p.example.com:1080", resolver_for(PUBLIC_IP)) == ( + f"socks5h://{PUBLIC_IP}:1080" + ) + + def test_prefers_ipv4_and_brackets_ipv6(self): + v6 = "2606:4700:4700::1111" + assert pin_proxy("http://p:8080", resolver_for(v6, PUBLIC_IP)).endswith(f"{PUBLIC_IP}:8080") + assert pin_proxy("http://p:8080", resolver_for(v6)) == f"http://[{v6}]:8080" + + @pytest.mark.parametrize( + "address", + [ + "127.0.0.1", # loopback + "10.0.0.5", # private + "172.16.0.1", + "192.168.1.1", + "169.254.169.254", # cloud metadata / link-local + "100.64.0.1", # carrier-grade NAT + "0.0.0.0", + "224.0.0.1", # multicast + "::1", + "fd00::1", # unique local + "fe80::1%eth0", # link-local with scope id + "::ffff:10.0.0.1", # IPv4-mapped private + "2002:0a00:0001::1", # 6to4 wrapping 10.0.0.1 + ], + ) + def test_rejects_non_public_addresses(self, address): + with pytest.raises(HTTPException) as excinfo: + pin_proxy("http://p.example.com:8080", resolver_for(address)) + assert excinfo.value.status_code == 400 + + def test_rejects_when_any_resolved_address_is_private(self): + with pytest.raises(HTTPException): + pin_proxy("http://p.example.com:8080", resolver_for(PUBLIC_IP, "10.0.0.1")) + + def test_rejects_ip_literal_private_host(self): + def resolver(host, port): + return [host] + with pytest.raises(HTTPException): + pin_proxy("http://169.254.169.254:8080", resolver) + + @pytest.mark.parametrize( + "url", + [ + "https://p.example.com:8443", # https proxies cannot be IP-pinned + "ftp://p.example.com:2121", + "p.example.com:8080", + "http://p.example.com", # no explicit port + "http://p.example.com:22", # low port + "http://p.example.com:6379x", + "http://p.example.com:8080/path", + "http://p.example.com:8080?x=1", + "http://:8080", + "", + "http://p.example.com:8080" + "a" * 2048, + ], + ) + def test_rejects_malformed_or_disallowed_urls(self, url): + with pytest.raises(HTTPException) as excinfo: + pin_proxy(url, resolver_for(PUBLIC_IP)) + assert excinfo.value.status_code == 400 + + def test_unresolvable_host(self): + def resolver(host, port): + raise OSError("no such host") + with pytest.raises(HTTPException) as excinfo: + pin_proxy(PROXY, resolver) + assert "resolved" in excinfo.value.detail + + def test_rejection_never_echoes_credentials(self): + with pytest.raises(HTTPException) as excinfo: + pin_proxy("http://alice:s3cret@p.example.com:8080", resolver_for("10.0.0.1")) + assert "s3cret" not in excinfo.value.detail + assert "alice" not in excinfo.value.detail + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + +class TestEndpoints: + def test_no_header_uses_shared_default(self, client): + with patch("psxdata.quote", return_value=_quote_df()) as default_quote: + resp = client.get("/stocks/ENGRO/quote") + assert resp.status_code == 200 + default_quote.assert_called_once_with("ENGRO") + + def test_header_rejected_when_disabled(self, client): + with patch("psxdata.quote") as default_quote: + resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + assert resp.status_code == 400 + assert "not enabled" in resp.json()["error"]["message"] + default_quote.assert_not_called() + + def test_proxied_request_uses_pinned_proxy_and_no_cache(self, client, enabled): + with ( + patch.object(PSXClient, "quote", return_value=_quote_df()) as proxied_quote, + patch("psxdata.quote") as default_quote, + ): + resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + assert resp.status_code == 200 + default_quote.assert_not_called() + proxied_quote.assert_called_once_with("ENGRO", cache=False) + pooled = enabled._clients[PINNED] + assert pooled._historical._session.proxies == {"http": PINNED, "https": PINNED} + + def test_proxied_historical_bypasses_shared_cache(self, client, enabled): + df = pd.DataFrame({ + "date": [pd.Timestamp("2024-01-05")], + "open": [1.0], "high": [2.0], "low": [0.5], + "close": [1.5], "volume": [10], "is_anomaly": [False], + }) + with patch.object(PSXClient, "stocks", return_value=df) as proxied_stocks: + resp = client.get("/stocks/ENGRO/historical", headers={PROXY_HEADER: PROXY}) + assert resp.status_code == 200 + assert resp.headers["X-Cache"] == "BYPASS" + assert resp.json()["meta"]["cached"] is False + proxied_stocks.assert_called_once_with("ENGRO", cache=False) + # The shared cache was not populated: an unproxied call still misses + with patch("psxdata.stocks", return_value=df): + resp = client.get("/stocks/ENGRO/historical") + assert resp.headers["X-Cache"] == "MISS" + + def test_private_proxy_rejected_with_400(self, client): + app.state.proxy_passthrough = ProxyPassthrough( + True, resolver=resolver_for("169.254.169.254"), connector=no_connect + ) + with patch.object(PSXClient, "quote") as proxied_quote: + resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + assert resp.status_code == 400 + assert "s3cret" not in resp.text + proxied_quote.assert_not_called() + + def test_unreachable_proxy_returns_502(self, client): + def refuse(host, port): + raise ConnectionRefusedError + app.state.proxy_passthrough = ProxyPassthrough( + True, resolver=resolver_for(PUBLIC_IP), connector=refuse + ) + resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + assert resp.status_code == 502 + assert resp.json()["error"]["code"] == "proxy_unreachable" + assert "s3cret" not in resp.text + + def test_connect_check_targets_pinned_ip(self, client, enabled): + seen = [] + enabled._connector = lambda host, port: seen.append((host, port)) + with patch.object(PSXClient, "quote", return_value=_quote_df()): + client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + assert seen == [(PUBLIC_IP, 8080)] + + def test_per_ip_rate_limit(self, client): + app.state.proxy_passthrough = ProxyPassthrough( + True, resolver=resolver_for(PUBLIC_IP), connector=no_connect, per_ip_per_minute=2 + ) + with patch.object(PSXClient, "screener", return_value=pd.DataFrame()): + codes = [ + client.get("/screener", headers={PROXY_HEADER: PROXY}).status_code + for _ in range(3) + ] + assert codes == [200, 200, 429] + + def test_concurrency_cap(self, client): + passthrough = ProxyPassthrough( + True, resolver=resolver_for(PUBLIC_IP), connector=no_connect, max_concurrent=1 + ) + app.state.proxy_passthrough = passthrough + passthrough._slots.acquire() # simulate one proxied request in flight + try: + resp = client.get("/screener", headers={PROXY_HEADER: PROXY}) + finally: + passthrough._slots.release() + assert resp.status_code == 429 + + def test_slot_released_after_upstream_error(self, client): + from psxdata.exceptions import PSXConnectionError + + passthrough = ProxyPassthrough( + True, resolver=resolver_for(PUBLIC_IP), connector=no_connect, max_concurrent=1 + ) + app.state.proxy_passthrough = passthrough + with patch.object(PSXClient, "screener", side_effect=PSXConnectionError("down")): + assert client.get("/screener", headers={PROXY_HEADER: PROXY}).status_code == 503 + with patch.object(PSXClient, "screener", return_value=pd.DataFrame()): + assert client.get("/screener", headers={PROXY_HEADER: PROXY}).status_code == 200 + + +class TestPooling: + def test_client_pool_bounded_and_token_providers_pruned(self, monkeypatch): + monkeypatch.setattr(token_module, "_proxy_providers", {}) + passthrough = ProxyPassthrough(True, max_clients=1) + first = f"http://{PUBLIC_IP}:8080" + second = f"http://{PUBLIC_IP}:8081" + + passthrough._client_for(first) + token_module.get_default_provider(normalize_proxy(first)) + assert len(token_module._proxy_providers) == 1 + + passthrough._client_for(second) + assert list(passthrough._clients) == [second] + assert token_module._proxy_providers == {} + + def test_same_proxy_reuses_client(self): + passthrough = ProxyPassthrough(True) + url = f"http://{PUBLIC_IP}:8080" + assert passthrough._client_for(url) is passthrough._client_for(url) + + +class TestFromEnv: + @pytest.mark.parametrize("value,expected", [ + ("1", True), ("true", True), ("ON", True), ("", False), ("0", False), ("no", False), + ]) + def test_flag(self, value, expected): + assert ProxyPassthrough.from_env({"PSX_PROXY_PASSTHROUGH": value}).enabled is expected + + def test_default_off(self): + assert ProxyPassthrough.from_env({}).enabled is False From 15730441ff4bab655a63b2e8b251138b1b7e56cb Mon Sep 17 00:00:00 2001 From: mtauha Date: Wed, 30 Sep 2026 20:55:55 +0500 Subject: [PATCH 2/3] feat: serve and cache proxied requests through the shared caches The proxy only changes the egress to PSX, not the data: PSX is HTTPS-only with verified certificates, so a proxy cannot alter what it relays. Proxied requests now read from and write to the same /historical cache and SDK disk cache as unproxied ones, so caching keeps working when most traffic is proxied. Removes X-Cache: BYPASS. --- CHANGELOG.md | 2 +- README.md | 4 ++-- api/proxy.py | 22 ++++++++---------- api/routers/stocks.py | 38 ++++++++++++++---------------- tests/api/test_proxy.py | 51 ++++++++++++++++++++++++++++------------- 5 files changed, 65 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3021e84..e8b08a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - a 5-second TCP check runs first (`502 proxy_unreachable` on failure); - limits of 10 proxied requests per minute per IP and 4 concurrent server-wide; - a bounded pool of proxied clients. -- Proxied requests never read or write the shared cache. `/historical` returns `X-Cache: BYPASS` for them. +- Proxied requests use the same caches as all other requests. The proxy only changes the route to PSX: cached data is served without contacting PSX, and data fetched through a proxy is cached for everyone. PSX is HTTPS-only with certificate verification, so a proxy cannot alter what it relays. - Proxy credentials are never logged or echoed in error messages. ### Changed diff --git a/README.md b/README.md index 64cbda5..26b1a11 100644 --- a/README.md +++ b/README.md @@ -140,7 +140,7 @@ Every `/historical` response says where it came from: | Header | Values | | ------ | ------ | -| `X-Cache` | `HIT` — fresh cached copy · `MISS` — fetched from PSX just now · `STALE` — PSX refused or was unreachable, so the last cached copy was served · `BYPASS` — fetched through your `X-PSX-Proxy`, cache not used | +| `X-Cache` | `HIT` — fresh cached copy · `MISS` — fetched from PSX just now · `STALE` — PSX refused or was unreachable, so the last cached copy was served | | `Age` | Seconds since the data was fetched from PSX (on `HIT` and `STALE`) | `meta.cached` is `true` for `HIT` and `STALE`. If PSX is rate-limiting and no cached copy exists, the API returns `503 psx_unavailable` with `Retry-After: 60`. @@ -161,7 +161,7 @@ The server connects to an address you choose, so the proxy is checked before use - **Schemes:** `http://`, `socks5://` or `socks5h://`, with an explicit port (80, 443, or 1024–65535) and optional `user:pass@`. `https://` proxies are not accepted. - **Public addresses only.** The proxy host must resolve only to public internet addresses. Loopback, private, link-local (including cloud metadata), CGNAT and multicast addresses are rejected. The connection is pinned to the checked IP, so the name cannot be re-pointed afterwards. - **Quick reachability check.** A proxy that doesn't accept a TCP connection within 5 seconds returns `502 proxy_unreachable`. -- **No cache.** Proxied requests never read or write the API's cache. `/historical` answers with `X-Cache: BYPASS`. +- **Same cache as everyone else.** The proxy only changes the route to PSX, not the data. Proxied requests are served from the cache when it's fresh, and what they fetch is cached for all callers. PSX is HTTPS-only and certificates are verified, so a proxy tunnels encrypted traffic and cannot alter it. - **Stricter limits.** 10 proxied requests per minute per IP and at most 4 in progress server-wide (`429` beyond that), on top of the normal limit. - **Credentials are never logged or echoed back.** Send the proxy only in the header, never in the URL. diff --git a/api/proxy.py b/api/proxy.py index 2587860..4bcbc6a 100644 --- a/api/proxy.py +++ b/api/proxy.py @@ -12,9 +12,12 @@ the connection is pinned to the checked IP, so DNS rebinding cannot point it somewhere else between the check and the request. - A short TCP connect check rejects dead proxies before any PSX request. -- Proxied requests never read or write a shared cache. - Per-IP rate limit, a global concurrency cap, and a bounded pool of clients. +Proxied requests share the same caches as all other requests: the proxy only +changes the egress to PSX, not the data. PSX is HTTPS-only and certificates are +verified, so a proxy tunnels TLS end to end and cannot alter what it relays. + Credentials in the proxy URL are never logged or echoed back. """ from __future__ import annotations @@ -22,7 +25,6 @@ import ipaddress import os import socket -import tempfile import threading from collections import OrderedDict from collections.abc import Callable, Iterator, Mapping @@ -135,8 +137,7 @@ def _proxy_key(pinned: str) -> frozenset[tuple[str, str]]: class PsxSource: """Where a request's PSX data comes from: the shared default client, or a proxied one. - Proxied calls always pass ``cache=False``, so a caller's proxy can never - read from or write to the cache shared by all callers. + Both read and write the same caches; only the route to PSX differs. """ def __init__(self, client: PSXClient | None = None) -> None: @@ -149,7 +150,7 @@ def proxied(self) -> bool: def fetch(self, name: str, *args: Any, **kwargs: Any) -> Any: if self._client is None: return getattr(psxdata, name)(*args, **kwargs) - return getattr(self._client, name)(*args, cache=False, **kwargs) + return getattr(self._client, name)(*args, **kwargs) class ProxyPassthrough: @@ -174,7 +175,6 @@ def __init__( self._slots = threading.BoundedSemaphore(max_concurrent) self._rate_limiter = MovingWindowRateLimiter(MemoryStorage()) self._rate = RateLimitItemPerMinute(per_ip_per_minute) - self._cache_dir: str | None = None @classmethod def from_env(cls, env: Mapping[str, str]) -> ProxyPassthrough: @@ -214,10 +214,8 @@ def _client_for(self, pinned: str) -> PSXClient: if client is not None: self._clients.move_to_end(pinned) return client - if self._cache_dir is None: - # Never touched (proxied calls use cache=False), but kept apart from the shared one - self._cache_dir = tempfile.mkdtemp(prefix="psxdata-proxied-") - client = PSXClient(cache_dir=self._cache_dir, proxy=pinned) + # Default cache_dir: the same on-disk cache the module-level functions use + client = PSXClient(proxy=pinned) self._clients[pinned] = client if len(self._clients) > self._max_clients: self._clients.popitem(last=False) @@ -253,8 +251,8 @@ def psx_source( description=( "Optional proxy for this request's PSX traffic: http://, socks5:// or socks5h://, " "with an explicit port and optional user:pass@. Must resolve to a public address. " - "Proxied requests bypass the cache and have stricter limits. Only honoured when " - "the server enables proxy passthrough." + "Proxied requests share the normal cache but have stricter rate limits. Only " + "honoured when the server enables proxy passthrough." ), ), ) -> Iterator[PsxSource]: diff --git a/api/routers/stocks.py b/api/routers/stocks.py index 73e3930..9562122 100644 --- a/api/routers/stocks.py +++ b/api/routers/stocks.py @@ -2,11 +2,11 @@ from __future__ import annotations import re +from collections.abc import Callable from datetime import date, datetime, timezone from typing import Any import pandas as pd -import psxdata from fastapi import APIRouter, Depends, HTTPException, Request, Response from api.cache.historical import CacheStatus @@ -46,10 +46,9 @@ def _df_to_records(df: pd.DataFrame) -> list[dict]: 200: { "headers": { "X-Cache": { - "description": "HIT (fresh cache), MISS (fetched from PSX), STALE " - "(PSX refused or was unreachable; last cached copy served), or BYPASS " - "(fetched through the caller's X-PSX-Proxy; cache not used)", - "schema": {"type": "string", "enum": ["HIT", "MISS", "STALE", "BYPASS"]}, + "description": "HIT (fresh cache), MISS (fetched from PSX), or STALE " + "(PSX refused or was unreachable; last cached copy served)", + "schema": {"type": "string", "enum": ["HIT", "MISS", "STALE"]}, }, "Age": { "description": "Seconds since the data was fetched from PSX (HIT and STALE only)", @@ -85,8 +84,11 @@ def _parse_date(name: str, value: str | None) -> date | None: raise HTTPException(status_code=422, detail=f"{name} must be an ISO date (YYYY-MM-DD)") -def _fetch_full_history(symbol: str) -> list[dict]: - return _df_to_records(psxdata.stocks(symbol, cache=False)) +def _full_history_fetcher(psx: PsxSource) -> Callable[[str], list[dict]]: + """Fetch a symbol's full history from PSX, through the caller's proxy if one was given.""" + def fetch(symbol: str) -> list[dict]: + return _df_to_records(psx.fetch("stocks", symbol, cache=False)) + return fetch @router.get("/stocks", response_model=StringListResponse) @@ -120,27 +122,21 @@ def get_historical( if start_date and end_date and start_date > end_date: raise HTTPException(status_code=422, detail="start must not be after end") - if psx.proxied: - # A caller's proxy must never feed or read the shared cache - all_rows = _df_to_records(psx.fetch("stocks", symbol.upper())) - cached = False - response.headers["X-Cache"] = "BYPASS" - else: - result = get_cache(request).get(symbol.upper(), _fetch_full_history) - all_rows = result.rows - cached = result.status is not CacheStatus.MISS - response.headers["X-Cache"] = result.status.value - if cached: - age = (datetime.now(timezone.utc) - result.fetched_at).total_seconds() - response.headers["Age"] = str(max(0, int(age))) + result = get_cache(request).get(symbol.upper(), _full_history_fetcher(psx)) lo = start_date.isoformat() if start_date else None hi = end_date.isoformat() if end_date else None rows = [ OHLCVRow.model_validate(r) - for r in all_rows + for r in result.rows if (lo is None or r["date"] >= lo) and (hi is None or r["date"] <= hi) ] + + cached = result.status is not CacheStatus.MISS + response.headers["X-Cache"] = result.status.value + if cached: + age = (datetime.now(timezone.utc) - result.fetched_at).total_seconds() + response.headers["Age"] = str(max(0, int(age))) return HistoricalResponse( data=rows, meta=MetaList(timestamp=_now_iso(), cached=cached, count=len(rows)), diff --git a/tests/api/test_proxy.py b/tests/api/test_proxy.py index 48bf22a..d3f541b 100644 --- a/tests/api/test_proxy.py +++ b/tests/api/test_proxy.py @@ -39,6 +39,14 @@ def enabled() -> ProxyPassthrough: return passthrough +def _history_df() -> pd.DataFrame: + return pd.DataFrame({ + "date": [pd.Timestamp("2024-01-05")], + "open": [1.0], "high": [2.0], "low": [0.5], + "close": [1.5], "volume": [10], "is_anomaly": [False], + }) + + def _quote_df() -> pd.DataFrame: return pd.DataFrame([{"symbol": "ENGRO", "sector": "Fertilizer", "price": 300.0}]) @@ -147,7 +155,7 @@ def test_header_rejected_when_disabled(self, client): assert "not enabled" in resp.json()["error"]["message"] default_quote.assert_not_called() - def test_proxied_request_uses_pinned_proxy_and_no_cache(self, client, enabled): + def test_proxied_request_uses_pinned_proxy(self, client, enabled): with ( patch.object(PSXClient, "quote", return_value=_quote_df()) as proxied_quote, patch("psxdata.quote") as default_quote, @@ -155,26 +163,37 @@ def test_proxied_request_uses_pinned_proxy_and_no_cache(self, client, enabled): resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) assert resp.status_code == 200 default_quote.assert_not_called() - proxied_quote.assert_called_once_with("ENGRO", cache=False) + proxied_quote.assert_called_once_with("ENGRO") # SDK cache on, same as unproxied pooled = enabled._clients[PINNED] assert pooled._historical._session.proxies == {"http": PINNED, "https": PINNED} - def test_proxied_historical_bypasses_shared_cache(self, client, enabled): - df = pd.DataFrame({ - "date": [pd.Timestamp("2024-01-05")], - "open": [1.0], "high": [2.0], "low": [0.5], - "close": [1.5], "volume": [10], "is_anomaly": [False], - }) - with patch.object(PSXClient, "stocks", return_value=df) as proxied_stocks: + def test_proxied_historical_fills_shared_cache(self, client, enabled): + with ( + patch.object(PSXClient, "stocks", return_value=_history_df()) as proxied_stocks, + patch("psxdata.stocks") as default_stocks, + ): resp = client.get("/stocks/ENGRO/historical", headers={PROXY_HEADER: PROXY}) - assert resp.status_code == 200 - assert resp.headers["X-Cache"] == "BYPASS" - assert resp.json()["meta"]["cached"] is False - proxied_stocks.assert_called_once_with("ENGRO", cache=False) - # The shared cache was not populated: an unproxied call still misses - with patch("psxdata.stocks", return_value=df): + assert resp.status_code == 200 + assert resp.headers["X-Cache"] == "MISS" + proxied_stocks.assert_called_once_with("ENGRO", cache=False) + + # Data fetched through a proxy serves everyone: an unproxied call is a HIT resp = client.get("/stocks/ENGRO/historical") - assert resp.headers["X-Cache"] == "MISS" + assert resp.headers["X-Cache"] == "HIT" + assert resp.json()["data"][0]["close"] == 1.5 + default_stocks.assert_not_called() + + def test_proxied_historical_reads_shared_cache(self, client, enabled): + with patch("psxdata.stocks", return_value=_history_df()): + client.get("/stocks/ENGRO/historical") + with patch.object(PSXClient, "stocks") as proxied_stocks: + resp = client.get("/stocks/ENGRO/historical", headers={PROXY_HEADER: PROXY}) + assert resp.headers["X-Cache"] == "HIT" + proxied_stocks.assert_not_called() + + def test_proxied_clients_share_the_default_disk_cache(self, enabled): + proxied = enabled._client_for(PINNED) + assert proxied._cache._cache.directory == PSXClient()._cache._cache.directory def test_private_proxy_rejected_with_400(self, client): app.state.proxy_passthrough = ProxyPassthrough( From dfb55b082d19c2160e167706c846272fe07cf915 Mon Sep 17 00:00:00 2001 From: mtauha Date: Wed, 30 Sep 2026 21:04:55 +0500 Subject: [PATCH 3/3] feat: contact the caller's proxy only on a cache miss Proxied requests are answered from the shared caches first. A cache-only client (its scrapers raise CacheMiss instead of calling PSX) reads the SDK disk cache; /historical already calls its fetcher only on a miss. Only then are the rate limit, DNS pinning, concurrency slot and TCP check run and PSX fetched through the proxy. Checks that need no network (feature flag, URL syntax) still run on every request. Cache hits no longer count toward the proxied rate limit. --- CHANGELOG.md | 3 +- README.md | 6 +- api/proxy.py | 138 +++++++++++++++++----- tests/api/test_proxy.py | 251 +++++++++++++++++++++++++++------------- 4 files changed, 282 insertions(+), 116 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e8b08a4..3bd7de2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,8 +16,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - the host must resolve only to public addresses, and the connection is pinned to the checked IP (defeats DNS rebinding); - `https://` proxies and ports below 1024 (other than 80 and 443) are refused; - a 5-second TCP check runs first (`502 proxy_unreachable` on failure); - - limits of 10 proxied requests per minute per IP and 4 concurrent server-wide; + - limits of 10 proxied PSX fetches per minute per IP and 4 concurrent server-wide; - a bounded pool of proxied clients. +- The proxy is only used on a cache miss: cached data is served without resolving or contacting the caller's proxy, and cache hits don't count toward the proxy rate limit. A malformed header, or one sent while the feature is off, is still rejected on every request. - Proxied requests use the same caches as all other requests. The proxy only changes the route to PSX: cached data is served without contacting PSX, and data fetched through a proxy is cached for everyone. PSX is HTTPS-only with certificate verification, so a proxy cannot alter what it relays. - Proxy credentials are never logged or echoed in error messages. diff --git a/README.md b/README.md index 26b1a11..8001b13 100644 --- a/README.md +++ b/README.md @@ -155,14 +155,14 @@ Send an `X-PSX-Proxy` header to have the API fetch that request's PSX data throu curl -H "X-PSX-Proxy: http://user:pass@proxy.example.com:8080" https://psxdata-api.fastapicloud.dev/stocks/ENGRO/quote ``` -The server connects to an address you choose, so the proxy is checked before use: +The proxy is only used when PSX has to be contacted: if the data is already in the API's cache, it's served straight from there and your proxy is never touched. When PSX must be fetched, the server connects to an address you choose, so the proxy is checked first: - **Only when enabled.** The server operator has to turn it on (`PSX_PROXY_PASSTHROUGH`). Otherwise the header is rejected with `400`. - **Schemes:** `http://`, `socks5://` or `socks5h://`, with an explicit port (80, 443, or 1024–65535) and optional `user:pass@`. `https://` proxies are not accepted. - **Public addresses only.** The proxy host must resolve only to public internet addresses. Loopback, private, link-local (including cloud metadata), CGNAT and multicast addresses are rejected. The connection is pinned to the checked IP, so the name cannot be re-pointed afterwards. -- **Quick reachability check.** A proxy that doesn't accept a TCP connection within 5 seconds returns `502 proxy_unreachable`. +- **Quick reachability check.** On a cache miss, a proxy that doesn't accept a TCP connection within 5 seconds returns `502 proxy_unreachable`. - **Same cache as everyone else.** The proxy only changes the route to PSX, not the data. Proxied requests are served from the cache when it's fresh, and what they fetch is cached for all callers. PSX is HTTPS-only and certificates are verified, so a proxy tunnels encrypted traffic and cannot alter it. -- **Stricter limits.** 10 proxied requests per minute per IP and at most 4 in progress server-wide (`429` beyond that), on top of the normal limit. +- **Stricter limits on PSX fetches.** Requests that go through your proxy to PSX are limited to 10 per minute per IP, with at most 4 in progress server-wide (`429` beyond that). Requests answered from the cache don't count toward this, only toward the normal limit. - **Credentials are never logged or echoed back.** Send the proxy only in the header, never in the URL. --- diff --git a/api/proxy.py b/api/proxy.py index 4bcbc6a..78d5315 100644 --- a/api/proxy.py +++ b/api/proxy.py @@ -5,6 +5,9 @@ the caller chose, so every proxy is checked before use: - The feature is off unless ``PSX_PROXY_PASSTHROUGH`` is set to a true value. +- The proxy is only contacted when PSX must be: a request is first answered + from the shared caches, and the network checks below (rate limit, DNS + pinning, concurrency slot, TCP check) run only on a cache miss. - Only ``http``, ``socks5`` and ``socks5h`` proxies. ``https`` proxies are refused: TLS to the proxy verifies its hostname, which rules out IP pinning. - An explicit port is required: 80, 443, or 1024-65535. @@ -29,6 +32,7 @@ from collections import OrderedDict from collections.abc import Callable, Iterator, Mapping from contextlib import contextmanager +from dataclasses import dataclass from typing import Any from urllib.parse import urlsplit @@ -37,7 +41,8 @@ from limits import RateLimitItemPerMinute from limits.storage import MemoryStorage from limits.strategies import MovingWindowRateLimiter -from psxdata import PSXClient +from psxdata import BaseScraper, PSXClient +from psxdata.constants import CACHE_DIR from psxdata.proxy import normalize_proxy from psxdata.scrapers import token as token_module from slowapi.util import get_remote_address @@ -59,6 +64,20 @@ class ProxyUnreachableError(Exception): """The caller's proxy passed validation but did not accept a TCP connection.""" +class CacheMiss(Exception): + """Raised by a cache-only client when answering would need a PSX request.""" + + +@dataclass(frozen=True) +class ParsedProxy: + """A syntactically valid caller proxy, not yet resolved or contacted.""" + + scheme: str + userinfo: str # "user:pass@" or "" + host: str + port: int + + def _resolve(host: str, port: int) -> list[str]: return [str(info[4][0]) for info in socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)] @@ -81,12 +100,12 @@ def _reject(message: str) -> HTTPException: return HTTPException(status_code=400, detail=f"{PROXY_HEADER}: {message}") -def pin_proxy(raw: str, resolver: Resolver = _resolve) -> str: - """Validate a caller-supplied proxy URL and return it pinned to a checked IP. +def parse_proxy(raw: str) -> ParsedProxy: + """Check a caller-supplied proxy URL without any network access. Raises: - HTTPException: 400 for any rejected proxy. The message never includes - the URL, so credentials are not echoed back. + HTTPException: 400 for a malformed or disallowed URL. The message never + includes the URL, so credentials are not echoed back. """ raw = raw.strip() if not raw or len(raw) > MAX_PROXY_URL_LENGTH: @@ -108,9 +127,22 @@ def pin_proxy(raw: str, resolver: Resolver = _resolve) -> str: host = parts.hostname if not host: raise _reject("must include a host") + userinfo = parts.netloc.rpartition("@")[0] + "@" if "@" in parts.netloc else "" + return ParsedProxy(scheme, userinfo, host, port) + +def resolve_proxy(proxy: ParsedProxy, resolver: Resolver = _resolve) -> str: + """Resolve *proxy* and return its URL pinned to a checked public IP. + + Raises: + HTTPException: 400 if the host cannot be resolved or any of its + addresses is not public. + """ try: - ips = {ipaddress.ip_address(addr.split("%", 1)[0]) for addr in resolver(host, port)} + ips = { + ipaddress.ip_address(addr.split("%", 1)[0]) + for addr in resolver(proxy.host, proxy.port) + } except (OSError, UnicodeError, ValueError): raise _reject("host could not be resolved") from None if not ips or not all(_is_public(ip) for ip in ips): @@ -118,8 +150,7 @@ def pin_proxy(raw: str, resolver: Resolver = _resolve) -> str: ip = min(ips, key=lambda a: (a.version, int(a))) # prefer IPv4 host_part = f"[{ip}]" if ip.version == 6 else str(ip) - userinfo = parts.netloc.rpartition("@")[0] + "@" if "@" in parts.netloc else "" - pinned = f"{scheme}://{userinfo}{host_part}:{port}" + pinned = f"{proxy.scheme}://{proxy.userinfo}{host_part}:{proxy.port}" try: normalize_proxy(pinned) except (ValueError, TypeError): @@ -129,28 +160,60 @@ def pin_proxy(raw: str, resolver: Resolver = _resolve) -> str: return pinned +def pin_proxy(raw: str, resolver: Resolver = _resolve) -> str: + """Parse, resolve and pin *raw* in one step (see parse_proxy and resolve_proxy).""" + return resolve_proxy(parse_proxy(raw), resolver) + + def _proxy_key(pinned: str) -> frozenset[tuple[str, str]]: # Same key psxdata uses for its per-proxy token providers return frozenset((normalize_proxy(pinned) or {}).items()) +def _cache_only(client: PSXClient) -> PSXClient: + """Block every scraper of *client* from the network, so it can only answer from cache.""" + + def refuse(*args: Any, **kwargs: Any) -> Any: + raise CacheMiss + + for scraper in vars(client).values(): + if isinstance(scraper, BaseScraper): + scraper._request = refuse # type: ignore[method-assign] + return client + + class PsxSource: - """Where a request's PSX data comes from: the shared default client, or a proxied one. + """Where a request's PSX data comes from. - Both read and write the same caches; only the route to PSX differs. + Without a proxy: the module-level psxdata functions, as before. With one: + the shared caches first, and only on a miss a request through the caller's + proxy. Both read and write the same caches; only the route to PSX differs. """ - def __init__(self, client: PSXClient | None = None) -> None: - self._client = client + def __init__( + self, + passthrough: ProxyPassthrough | None = None, + request: Request | None = None, + proxy: ParsedProxy | None = None, + ) -> None: + self._passthrough = passthrough + self._request = request + self._proxy = proxy @property def proxied(self) -> bool: - return self._client is not None + return self._proxy is not None def fetch(self, name: str, *args: Any, **kwargs: Any) -> Any: - if self._client is None: + if self._passthrough is None or self._request is None or self._proxy is None: return getattr(psxdata, name)(*args, **kwargs) - return getattr(self._client, name)(*args, **kwargs) + if kwargs.get("cache", True): + try: + return getattr(self._passthrough.cache_only_client(), name)(*args, **kwargs) + except CacheMiss: + pass + with self._passthrough.acquire(self._request, self._proxy) as client: + return getattr(client, name)(*args, **kwargs) class ProxyPassthrough: @@ -165,8 +228,11 @@ def __init__( max_clients: int = MAX_CLIENTS, max_concurrent: int = MAX_CONCURRENT, per_ip_per_minute: int = PER_IP_PER_MINUTE, + cache_dir: str = CACHE_DIR, ) -> None: self.enabled = enabled + self._cache_dir = cache_dir + self._cache_only_client: PSXClient | None = None self._resolver = resolver self._connector = connector self._max_clients = max_clients @@ -181,17 +247,31 @@ def from_env(cls, env: Mapping[str, str]) -> ProxyPassthrough: enabled = env.get(ENABLE_ENV, "").strip().lower() in {"1", "true", "yes", "on"} return cls(enabled) - @contextmanager - def acquire(self, request: Request, raw: str) -> Iterator[PsxSource]: - """Validate *raw*, reserve a concurrency slot, and yield a proxied source.""" + def check(self, raw: str) -> ParsedProxy: + """Checks that need no network: the feature is enabled and *raw* is well formed.""" if not self.enabled: raise HTTPException( status_code=400, detail=f"{PROXY_HEADER} is not enabled on this server" ) + return parse_proxy(raw) + + def cache_only_client(self) -> PSXClient: + """A client over the shared SDK disk cache that raises CacheMiss instead of calling PSX.""" + with self._lock: + if self._cache_only_client is None: + self._cache_only_client = _cache_only(PSXClient(cache_dir=self._cache_dir)) + return self._cache_only_client + + @contextmanager + def acquire(self, request: Request, proxy: ParsedProxy) -> Iterator[PSXClient]: + """Resolve and pin *proxy*, reserve a concurrency slot, and yield a proxied client. + + Called only when PSX must actually be contacted. + """ # Rate-limit before DNS resolution so the check itself cannot be spammed if not self._rate_limiter.hit(self._rate, "psx-proxy", get_remote_address(request)): raise HTTPException(status_code=429, detail="Too many proxied requests") - pinned = pin_proxy(raw, self._resolver) + pinned = resolve_proxy(proxy, self._resolver) if not self._slots.acquire(blocking=False): raise HTTPException( status_code=429, detail="Too many proxied requests in progress; retry shortly" @@ -204,7 +284,7 @@ def acquire(self, request: Request, raw: str) -> Iterator[PsxSource]: raise ProxyUnreachableError( f"{PROXY_HEADER}: proxy did not accept a connection" ) from None - yield PsxSource(self._client_for(pinned)) + yield self._client_for(pinned) finally: self._slots.release() @@ -214,8 +294,8 @@ def _client_for(self, pinned: str) -> PSXClient: if client is not None: self._clients.move_to_end(pinned) return client - # Default cache_dir: the same on-disk cache the module-level functions use - client = PSXClient(proxy=pinned) + # Same on-disk cache as the module-level functions (CACHE_DIR by default) + client = PSXClient(cache_dir=self._cache_dir, proxy=pinned) self._clients[pinned] = client if len(self._clients) > self._max_clients: self._clients.popitem(last=False) @@ -251,14 +331,14 @@ def psx_source( description=( "Optional proxy for this request's PSX traffic: http://, socks5:// or socks5h://, " "with an explicit port and optional user:pass@. Must resolve to a public address. " - "Proxied requests share the normal cache but have stricter rate limits. Only " - "honoured when the server enables proxy passthrough." + "Cached data is served without contacting the proxy; it is used only when PSX " + "must be fetched, with stricter rate limits. Only honoured when the server " + "enables proxy passthrough." ), ), -) -> Iterator[PsxSource]: +) -> PsxSource: """FastAPI dependency: the PSX data source for this request.""" if x_psx_proxy is None: - yield PsxSource() - return - with get_proxy_passthrough(request).acquire(request, x_psx_proxy) as source: - yield source + return PsxSource() + passthrough = get_proxy_passthrough(request) + return PsxSource(passthrough, request, passthrough.check(x_psx_proxy)) diff --git a/tests/api/test_proxy.py b/tests/api/test_proxy.py index d3f541b..6c05262 100644 --- a/tests/api/test_proxy.py +++ b/tests/api/test_proxy.py @@ -1,15 +1,26 @@ -"""Tests for X-PSX-Proxy passthrough and its SSRF hardening — no network.""" +"""Tests for X-PSX-Proxy passthrough and its SSRF hardening — no network. + +PSX is faked at the scraper level: ``fake_psx`` replaces a scraper's +``fetch`` with one that first calls ``self._request`` (which raises CacheMiss +on the cache-only client, exactly like a real network call would) and then +records which proxy the fetching scraper was configured with. +""" from __future__ import annotations +from collections.abc import Iterator +from contextlib import contextmanager +from typing import Any from unittest.mock import patch import pandas as pd import pytest from fastapi import HTTPException from fastapi.testclient import TestClient -from psxdata import PSXClient +from psxdata import BaseScraper, PSXClient +from psxdata.exceptions import PSXConnectionError from psxdata.proxy import normalize_proxy from psxdata.scrapers import token as token_module +from psxdata.scrapers.screener import ScreenerScraper from api.main import app from api.proxy import PROXY_HEADER, ProxyPassthrough, pin_proxy @@ -17,6 +28,7 @@ PUBLIC_IP = "93.184.216.34" PROXY = "http://alice:s3cret@proxy.example.com:8080" PINNED = f"http://alice:s3cret@{PUBLIC_IP}:8080" +PINNED_PROXIES = {"http": PINNED, "https": PINNED} def resolver_for(*addrs: str): @@ -27,16 +39,61 @@ def no_connect(host: str, port: int) -> None: return None +class Counter: + """Connector that records every proxy contact.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, int]] = [] + + def __call__(self, host: str, port: int) -> None: + self.calls.append((host, port)) + + +def refuse_network(*args: Any) -> Any: + raise AssertionError("the proxy must not be contacted") + + +@contextmanager +def fake_psx( + scraper_cls: type, result: Any = None, error: Exception | None = None +) -> Iterator[list]: + """Fake PSX for *scraper_cls*; yields the proxies of every scraper that reached PSX.""" + reached: list[dict | None] = [] + + def fetch(self: BaseScraper, *args: Any, **kwargs: Any) -> Any: + self._request("GET", "https://dps.psx.com.pk/") # CacheMiss on the cache-only client + reached.append(self._proxies) + if error is not None: + raise error + return result + + with ( + patch.object(scraper_cls, "fetch", fetch), + patch.object(BaseScraper, "_request", return_value=None), + ): + yield reached + + @pytest.fixture def client() -> TestClient: return TestClient(app) @pytest.fixture -def enabled() -> ProxyPassthrough: - passthrough = ProxyPassthrough(True, resolver=resolver_for(PUBLIC_IP), connector=no_connect) - app.state.proxy_passthrough = passthrough - return passthrough +def make_passthrough(tmp_path): + """Build an enabled passthrough over an isolated SDK disk cache and install it.""" + def make(**kwargs: Any) -> ProxyPassthrough: + kwargs.setdefault("resolver", resolver_for(PUBLIC_IP)) + kwargs.setdefault("connector", no_connect) + passthrough = ProxyPassthrough(True, cache_dir=str(tmp_path / "cache"), **kwargs) + app.state.proxy_passthrough = passthrough + return passthrough + return make + + +@pytest.fixture +def enabled(make_passthrough) -> ProxyPassthrough: + return make_passthrough() def _history_df() -> pd.DataFrame: @@ -47,8 +104,12 @@ def _history_df() -> pd.DataFrame: }) -def _quote_df() -> pd.DataFrame: - return pd.DataFrame([{"symbol": "ENGRO", "sector": "Fertilizer", "price": 300.0}]) +def _screener_df() -> pd.DataFrame: + return pd.DataFrame([{"symbol": "ENGRO", "sector": 7.0, "price": 300.0}]) + + +def _proxied(client: TestClient, path: str, proxy: str = PROXY): + return client.get(path, headers={PROXY_HEADER: proxy}) # --------------------------------------------------------------------------- @@ -138,41 +199,81 @@ def test_rejection_never_echoes_credentials(self): # --------------------------------------------------------------------------- -# Endpoints +# Endpoints — upfront checks # --------------------------------------------------------------------------- -class TestEndpoints: +class TestUpfrontChecks: def test_no_header_uses_shared_default(self, client): - with patch("psxdata.quote", return_value=_quote_df()) as default_quote: - resp = client.get("/stocks/ENGRO/quote") + with patch("psxdata.screener", return_value=_screener_df()) as default_screener: + resp = client.get("/screener") assert resp.status_code == 200 - default_quote.assert_called_once_with("ENGRO") + default_screener.assert_called_once_with() def test_header_rejected_when_disabled(self, client): - with patch("psxdata.quote") as default_quote: - resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + with patch("psxdata.screener") as default_screener: + resp = _proxied(client, "/screener") assert resp.status_code == 400 assert "not enabled" in resp.json()["error"]["message"] - default_quote.assert_not_called() + default_screener.assert_not_called() - def test_proxied_request_uses_pinned_proxy(self, client, enabled): - with ( - patch.object(PSXClient, "quote", return_value=_quote_df()) as proxied_quote, - patch("psxdata.quote") as default_quote, - ): - resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + def test_malformed_proxy_rejected_even_when_cached(self, client, enabled): + with fake_psx(ScreenerScraper, _screener_df()): + _proxied(client, "/screener") # fill the cache + resp = _proxied(client, "/screener", proxy="ftp://alice:s3cret@p.example.com:2121") + assert resp.status_code == 400 + assert "s3cret" not in resp.text + + +# --------------------------------------------------------------------------- +# Endpoints — cache first, proxy only on a miss +# --------------------------------------------------------------------------- + +class TestCacheFirst: + def test_miss_fetches_through_pinned_proxy(self, client, make_passthrough): + connector = Counter() + make_passthrough(connector=connector) + with fake_psx(ScreenerScraper, _screener_df()) as reached: + resp = _proxied(client, "/screener") + assert resp.status_code == 200 + assert reached == [PINNED_PROXIES] + assert connector.calls == [(PUBLIC_IP, 8080)] + + def test_hit_is_served_without_contacting_proxy(self, client, make_passthrough): + connector = Counter() + make_passthrough(connector=connector) + with fake_psx(ScreenerScraper, _screener_df()) as reached: + first = _proxied(client, "/screener") + second = _proxied(client, "/screener") + assert first.json()["data"] == second.json()["data"] + assert len(reached) == 1 # PSX fetched once + assert len(connector.calls) == 1 # proxy contacted once + + def test_hit_on_data_cached_by_another_caller(self, client, make_passthrough, tmp_path): + make_passthrough(resolver=refuse_network, connector=refuse_network) + PSXClient(cache_dir=str(tmp_path / "cache"))._cache.set("screener_all", _screener_df()) + with fake_psx(ScreenerScraper, _screener_df()) as reached: + resp = _proxied(client, "/stocks/ENGRO/quote") assert resp.status_code == 200 - default_quote.assert_not_called() - proxied_quote.assert_called_once_with("ENGRO") # SDK cache on, same as unproxied - pooled = enabled._clients[PINNED] - assert pooled._historical._session.proxies == {"http": PINNED, "https": PINNED} + assert resp.json()["data"]["symbol"] == "ENGRO" + assert reached == [] + + def test_hits_do_not_count_against_proxy_rate_limit(self, client, make_passthrough): + make_passthrough(per_ip_per_minute=1) + with fake_psx(ScreenerScraper, _screener_df()): + codes = [_proxied(client, "/screener").status_code for _ in range(5)] + assert codes == [200] * 5 - def test_proxied_historical_fills_shared_cache(self, client, enabled): + def test_proxied_and_default_clients_share_disk_cache(self, enabled): + proxied = enabled._client_for(PINNED) + cache_only = enabled.cache_only_client() + assert proxied._cache._cache.directory == cache_only._cache._cache.directory + + def test_proxied_historical_miss_fills_shared_cache(self, client, enabled): with ( patch.object(PSXClient, "stocks", return_value=_history_df()) as proxied_stocks, patch("psxdata.stocks") as default_stocks, ): - resp = client.get("/stocks/ENGRO/historical", headers={PROXY_HEADER: PROXY}) + resp = _proxied(client, "/stocks/ENGRO/historical") assert resp.status_code == 200 assert resp.headers["X-Cache"] == "MISS" proxied_stocks.assert_called_once_with("ENGRO", cache=False) @@ -183,86 +284,70 @@ def test_proxied_historical_fills_shared_cache(self, client, enabled): assert resp.json()["data"][0]["close"] == 1.5 default_stocks.assert_not_called() - def test_proxied_historical_reads_shared_cache(self, client, enabled): + def test_proxied_historical_hit_never_contacts_proxy(self, client, make_passthrough): + make_passthrough(resolver=refuse_network, connector=refuse_network) with patch("psxdata.stocks", return_value=_history_df()): client.get("/stocks/ENGRO/historical") with patch.object(PSXClient, "stocks") as proxied_stocks: - resp = client.get("/stocks/ENGRO/historical", headers={PROXY_HEADER: PROXY}) + resp = _proxied(client, "/stocks/ENGRO/historical") assert resp.headers["X-Cache"] == "HIT" proxied_stocks.assert_not_called() - def test_proxied_clients_share_the_default_disk_cache(self, enabled): - proxied = enabled._client_for(PINNED) - assert proxied._cache._cache.directory == PSXClient()._cache._cache.directory - def test_private_proxy_rejected_with_400(self, client): - app.state.proxy_passthrough = ProxyPassthrough( - True, resolver=resolver_for("169.254.169.254"), connector=no_connect - ) - with patch.object(PSXClient, "quote") as proxied_quote: - resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) +# --------------------------------------------------------------------------- +# Endpoints — network checks on a miss +# --------------------------------------------------------------------------- + +class TestMissChecks: + # An empty screener is never cached, so every request below is a miss. + + def test_private_proxy_rejected_with_400(self, client, make_passthrough): + make_passthrough(resolver=resolver_for("169.254.169.254")) + with fake_psx(ScreenerScraper, pd.DataFrame()) as reached: + resp = _proxied(client, "/screener") assert resp.status_code == 400 assert "s3cret" not in resp.text - proxied_quote.assert_not_called() + assert reached == [] - def test_unreachable_proxy_returns_502(self, client): + def test_unreachable_proxy_returns_502(self, client, make_passthrough): def refuse(host, port): raise ConnectionRefusedError - app.state.proxy_passthrough = ProxyPassthrough( - True, resolver=resolver_for(PUBLIC_IP), connector=refuse - ) - resp = client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) + make_passthrough(connector=refuse) + with fake_psx(ScreenerScraper, pd.DataFrame()) as reached: + resp = _proxied(client, "/screener") assert resp.status_code == 502 assert resp.json()["error"]["code"] == "proxy_unreachable" assert "s3cret" not in resp.text + assert reached == [] - def test_connect_check_targets_pinned_ip(self, client, enabled): - seen = [] - enabled._connector = lambda host, port: seen.append((host, port)) - with patch.object(PSXClient, "quote", return_value=_quote_df()): - client.get("/stocks/ENGRO/quote", headers={PROXY_HEADER: PROXY}) - assert seen == [(PUBLIC_IP, 8080)] - - def test_per_ip_rate_limit(self, client): - app.state.proxy_passthrough = ProxyPassthrough( - True, resolver=resolver_for(PUBLIC_IP), connector=no_connect, per_ip_per_minute=2 - ) - with patch.object(PSXClient, "screener", return_value=pd.DataFrame()): - codes = [ - client.get("/screener", headers={PROXY_HEADER: PROXY}).status_code - for _ in range(3) - ] + def test_per_ip_rate_limit(self, client, make_passthrough): + make_passthrough(per_ip_per_minute=2) + with fake_psx(ScreenerScraper, pd.DataFrame()): + codes = [_proxied(client, "/screener").status_code for _ in range(3)] assert codes == [200, 200, 429] - def test_concurrency_cap(self, client): - passthrough = ProxyPassthrough( - True, resolver=resolver_for(PUBLIC_IP), connector=no_connect, max_concurrent=1 - ) - app.state.proxy_passthrough = passthrough - passthrough._slots.acquire() # simulate one proxied request in flight + def test_concurrency_cap(self, client, make_passthrough): + passthrough = make_passthrough(max_concurrent=1) + passthrough._slots.acquire() # simulate one proxied fetch in flight try: - resp = client.get("/screener", headers={PROXY_HEADER: PROXY}) + with fake_psx(ScreenerScraper, pd.DataFrame()): + resp = _proxied(client, "/screener") finally: passthrough._slots.release() assert resp.status_code == 429 - def test_slot_released_after_upstream_error(self, client): - from psxdata.exceptions import PSXConnectionError - - passthrough = ProxyPassthrough( - True, resolver=resolver_for(PUBLIC_IP), connector=no_connect, max_concurrent=1 - ) - app.state.proxy_passthrough = passthrough - with patch.object(PSXClient, "screener", side_effect=PSXConnectionError("down")): - assert client.get("/screener", headers={PROXY_HEADER: PROXY}).status_code == 503 - with patch.object(PSXClient, "screener", return_value=pd.DataFrame()): - assert client.get("/screener", headers={PROXY_HEADER: PROXY}).status_code == 200 + def test_slot_released_after_upstream_error(self, client, make_passthrough): + make_passthrough(max_concurrent=1) + with fake_psx(ScreenerScraper, error=PSXConnectionError("down")): + assert _proxied(client, "/screener").status_code == 503 + with fake_psx(ScreenerScraper, pd.DataFrame()): + assert _proxied(client, "/screener").status_code == 200 class TestPooling: - def test_client_pool_bounded_and_token_providers_pruned(self, monkeypatch): + def test_client_pool_bounded_and_token_providers_pruned(self, monkeypatch, tmp_path): monkeypatch.setattr(token_module, "_proxy_providers", {}) - passthrough = ProxyPassthrough(True, max_clients=1) + passthrough = ProxyPassthrough(True, max_clients=1, cache_dir=str(tmp_path)) first = f"http://{PUBLIC_IP}:8080" second = f"http://{PUBLIC_IP}:8081" @@ -274,8 +359,8 @@ def test_client_pool_bounded_and_token_providers_pruned(self, monkeypatch): assert list(passthrough._clients) == [second] assert token_module._proxy_providers == {} - def test_same_proxy_reuses_client(self): - passthrough = ProxyPassthrough(True) + def test_same_proxy_reuses_client(self, tmp_path): + passthrough = ProxyPassthrough(True, cache_dir=str(tmp_path)) url = f"http://{PUBLIC_IP}:8080" assert passthrough._client_for(url) is passthrough._client_for(url)