From 04602a0e0e35d7b353d535c5303d757541901823 Mon Sep 17 00:00:00 2001 From: Sameer Kadam Date: Mon, 17 Aug 2026 18:55:38 +0530 Subject: [PATCH] fix(security): prevent Authorization header leakage across redirects (#947) (#1067) * fix(security): prevent auth header leakage across redirects * fix(security): harden redirect credential handling Address Copilot and Qodo review findings for #947. - Remove unused variables, imports, and unnecessary pass statements from tests. - Harden cross-origin redirect handling for per-request auth credentials. - Strip session-level auth handlers before cross-origin redirect hops. - Prevent session.auth from regenerating Authorization headers. - Disable trust_env during cross-origin hops to prevent .netrc credential injection. - Restore session auth and trust_env state reliably with try/finally. - Add regression coverage for auth=, session.auth, trust_env, and multi-hop redirects. - Preserve existing security behavior and same-origin authentication semantics. Validated with 189/189 security and affected tests passing. * fix(security): scope allow_private_ips to same-host redirects, fix error handling gaps Follow-up to review findings on #1067: - MCPClient hardcoded allow_private_ips=True for every redirect hop, not just its operator-configured host, so a compromised/malicious MCP server could 302 into private address space (e.g. cloud metadata) unchecked. request_with_ssrf_guard() gains allow_private_ips_on_redirect: a redirect target inherits the original host's private-IP trust only when it matches that host; MCPClient now pins it to False. - detect_public_api() only caught requests.exceptions.RequestException, but the SSRF guard raises ValidationError for blocked hosts/redirects, unlike its sibling ingest_public_api(). Now catches and re-raises it the same way. - detect_public_api()/ingest_public_api() forwarded session/allow_private_ips through **options into request_with_ssrf_guard(), which already passes both explicitly -- a caller supplying either would hit a duplicate-kwarg TypeError. Both are now popped from request_options first. New regression coverage for all three in tests/ingest/, plus a CHANGELOG entry under Unreleased/Security. --------- Co-authored-by: KaifAhmad1 --- CHANGELOG.md | 10 + semantica/ingest/mcp_client.py | 49 +- semantica/ingest/public_api_ingestor.py | 38 +- semantica/ingest/ssrf.py | 227 +++- semantica/seed/seed_manager.py | 7 +- tests/ingest/conftest.py | 35 + .../test_auth_header_redirect_security.py | 1063 +++++++++++++++++ tests/ingest/test_cookbook_integration.py | 40 +- tests/ingest/test_public_api_ingestor.py | 85 +- tests/ingest/test_submodules.py | 135 +-- tests/test_seed_manager.py | 54 + 11 files changed, 1561 insertions(+), 182 deletions(-) create mode 100644 tests/ingest/conftest.py create mode 100644 tests/ingest/test_auth_header_redirect_security.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f670cb22..0ebdc236 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -176,6 +176,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- **`Authorization`/`Proxy-Authorization` credentials could leak to a different origin across HTTP redirects, and several ingest paths bypassed the shared SSRF/redirect guard entirely** (#1067, closes #947) by @Sameer6305, reviewed by @KaifAhmad1 + - `request_with_ssrf_guard()` previously only stripped sensitive headers from per-request `kwargs["headers"]` on a cross-origin redirect; session-level `Authorization`/`Proxy-Authorization` headers, `session.auth`, and `session.trust_env` (`.netrc` lookup) could all still resurrect credentials on the hop to a foreign origin. All five credential sources are now stripped case-insensitively, kept stripped for the remainder of a multi-hop redirect chain (no resurrection even if a later hop returns to the original host), and unconditionally restored via `finally` — including on exceptions and redirect-limit errors + - `MCPClient._send_request_http()` and `PublicAPIIngestor.detect_public_api()`/`ingest_public_api()` called `httpx.post()`/`requests.post()`/`session.request()` directly, bypassing `request_with_ssrf_guard()` entirely. Both now route through the shared guard, including when `validate_no_auth=False` + - `SeedDataManager.load_from_api()` mutated the caller-supplied `headers` dict in place when adding an API-key `Authorization` header, silently leaking the key back into a dict the caller might reuse elsewhere. Now copies before modifying + - **Fixed during review** (@KaifAhmad1): `allow_private_ips=True` (used to let MCP servers run on localhost/internal networks) was applied to every redirect hop, not just the operator-configured host — a compromised or malicious MCP server could 302-redirect to an internal address (e.g. `169.254.169.254` cloud metadata) and the guard would follow it unchecked, defeating the SSRF protection this PR otherwise adds. Added `allow_private_ips_on_redirect` to `request_with_ssrf_guard()`: a redirect target inherits the original host's private-IP trust only when it matches that host; any other host falls back to strict validation. `MCPClient` now pins `allow_private_ips_on_redirect=False`, so only same-host redirects on a trusted MCP server keep working — a cross-host hop into private address space is blocked + - **Fixed during review** (@KaifAhmad1): `detect_public_api()` only caught `requests.exceptions.RequestException`, but `request_with_ssrf_guard()` raises `ValidationError` (a disjoint hierarchy) for SSRF-blocked hosts, blocked redirect targets, missing `Location`, or exceeded redirect limits — unlike its sibling `ingest_public_api()`, which already caught it. Callers (including `is_public_api()`) got an undocumented raw `ValidationError` instead of `ProcessingError`, and the error-logging call was skipped. Now catches `(ValidationError, ProcessingError)` and re-raises, matching the sibling method + - **Fixed during review** (@KaifAhmad1): `detect_public_api()`/`ingest_public_api()` forwarded `**options` into `request_with_ssrf_guard(..., session=self.session, allow_private_ips=self.allow_private_ips, **request_options)` without stripping `session`/`allow_private_ips` from `request_options` first — a caller passing either through the per-call `**options` (a plausible mistake, since `allow_private_ips` is also a documented constructor-level knob) got a raw `TypeError: got multiple values for keyword argument`. Both are now popped from `request_options` before the call + - New regression coverage added during review: `TestAllowPrivateIpsOnRedirect` (cross-host redirect into private space blocked, same-host redirect trust preserved, default behavior unchanged for existing callers that don't pass the new kwarg) and `TestMCPClientAuthRedirect::test_redirect_to_private_ip_is_blocked`/`test_same_host_redirect_on_private_mcp_server_is_not_blocked` in `tests/ingest/test_auth_header_redirect_security.py`; `test_detect_public_api_propagates_ssrf_validation_error` and duplicate-kwarg regression tests for both methods in `tests/ingest/test_public_api_ingestor.py` + - `pytest tests/ingest/test_auth_header_redirect_security.py tests/ingest/test_public_api_ingestor.py tests/test_seed_manager.py tests/ingest/test_submodules.py tests/ingest/test_cookbook_integration.py`: 111 passed + - **`FeedIngestor`/`FeedMonitor` (RSS/Atom feed ingestion) had no SSRF protection, allowing requests to internal/private network targets** (#928, closes #927) by @ZohaibHassan16 - `FeedIngestor.ingest_feed()`, `discover_feeds()` (link-tag fetch, common-path HEAD probe, and feed-validation GET), and `FeedMonitor.check_updates()` all called `requests.get()`/`requests.head()` directly with default redirect-following and no scheme allowlist or private/loopback/link-local IP validation — despite `semantica/ingest/ssrf.py`'s `request_with_ssrf_guard()` already existing and being used by `web_ingestor.py`/`api_ingestor.py`. `ingest_feed()`'s own URL check only verified `urlparse(url).scheme`/`.netloc` were non-empty, never that the scheme was http/https or that the resolved target IP was safe. Reachable via the public `ingest_feed()`/`ingest()` entry points with any caller-supplied feed URL - All 5 call sites now route through `request_with_ssrf_guard()`, which validates scheme (http/https only) and resolved IP before the request, and re-validates every redirect `Location` before following it — closing both the direct-IP and redirect-chain SSRF paths. Added an `allow_private_ips` config option to both `FeedIngestor` and `FeedMonitor`, consistent with the other ingestors diff --git a/semantica/ingest/mcp_client.py b/semantica/ingest/mcp_client.py index 2b33dfe1..302e3365 100644 --- a/semantica/ingest/mcp_client.py +++ b/semantica/ingest/mcp_client.py @@ -41,6 +41,7 @@ from typing import Any, Dict, List, Optional, Union from ..utils.exceptions import ProcessingError, ValidationError from ..utils.logging import get_logger +from .ssrf import request_with_ssrf_guard @dataclass @@ -341,36 +342,38 @@ class MCPClient: raise def _send_request_http(self, request: Dict[str, Any]) -> Optional[Dict[str, Any]]: - """Send request via HTTP.""" - try: - import httpx + """Send request via HTTP, with redirect-safe credential handling. - response = httpx.post( + Uses ``request_with_ssrf_guard`` so that: + + * ``Authorization`` / ``Proxy-Authorization`` headers are **not** + forwarded to a different origin if the MCP server issues a redirect + (issue #947). + * The redirect chain is bounded (default 10 hops). + + ``allow_private_ips=True`` is set because MCP servers are explicitly + configured by the operator and frequently run on localhost or an + internal network — the same trust model as ``allow_private_ips`` opt-in + in the other ingestors. That trust covers only ``self.url`` itself: + ``allow_private_ips_on_redirect=False`` keeps redirect targets held to + the normal public-address check, so a compromised or malicious MCP + server cannot use a redirect to route the client into private/ + internal address space (e.g. cloud metadata) that the operator never + configured. Scheme validation (http/https only) and the + auth-stripping logic remain active regardless of these flags. + """ + try: + response = request_with_ssrf_guard( + "POST", self.url, - json=request, headers=self.headers, + json=request, timeout=self.config.get("timeout", 30.0), + allow_private_ips=True, + allow_private_ips_on_redirect=False, ) response.raise_for_status() return response.json() - except (ImportError, OSError): - # Fallback to requests if httpx not available - try: - import requests - - response = requests.post( - self.url, - json=request, - headers=self.headers, - timeout=self.config.get("timeout", 30.0), - ) - response.raise_for_status() - return response.json() - except (ImportError, OSError): - raise ProcessingError( - "HTTP transport requires 'httpx' or 'requests' package. " - "Install with: pip install httpx or pip install requests" - ) except Exception as e: self.logger.error(f"Failed to send HTTP request: {e}") raise diff --git a/semantica/ingest/public_api_ingestor.py b/semantica/ingest/public_api_ingestor.py index afefbe27..2ea15b20 100644 --- a/semantica/ingest/public_api_ingestor.py +++ b/semantica/ingest/public_api_ingestor.py @@ -45,6 +45,7 @@ except ModuleNotFoundError: # pragma: no cover - fallback for minimal installs from ..utils.exceptions import ProcessingError, ValidationError from ..utils.logging import get_logger from .api_ingestor import APIData, RESTIngestor +from .ssrf import request_with_ssrf_guard AUTH_HEADER_NAMES = { "authorization", @@ -359,18 +360,31 @@ class PublicAPIIngestor(RESTIngestor): request_options = options.copy() timeout = request_options.pop("timeout", self.config.get("timeout", 30)) rate_limit_delay = request_options.pop("rate_limit_delay", None) + # session and allow_private_ips are always supplied explicitly below; + # drop any caller-provided copies so request_with_ssrf_guard() does + # not receive duplicate keyword arguments. + request_options.pop("session", None) + request_options.pop("allow_private_ips", None) request_headers = self._merged_headers(headers) try: self._wait_if_needed(rate_limit_delay=rate_limit_delay) - response = self.session.request( - method=method, - url=endpoint, + # Route through the SSRF guard so that: + # * redirects to private/loopback IPs are blocked, and + # * Authorization / Proxy-Authorization are stripped on + # cross-origin redirects (issue #947). + response = request_with_ssrf_guard( + method, + endpoint, + session=self.session, headers=request_headers, params=params, timeout=timeout, + allow_private_ips=self.allow_private_ips, **request_options, ) + except (ValidationError, ProcessingError): + raise except requests.exceptions.RequestException as exc: self.logger.error(f"Failed to detect public API {endpoint}: {exc}") raise ProcessingError(f"Failed to detect public API: {exc}") from exc @@ -440,18 +454,30 @@ class PublicAPIIngestor(RESTIngestor): request_options = options.copy() timeout = request_options.pop("timeout", self.config.get("timeout", 30)) + # session and allow_private_ips are always supplied explicitly below; + # drop any caller-provided copies so request_with_ssrf_guard() does + # not receive duplicate keyword arguments. + request_options.pop("session", None) + request_options.pop("allow_private_ips", None) request_headers = self._merged_headers(headers) try: self._wait_if_needed(rate_limit_delay=rate_limit_delay) - response = self.session.request( - method=method, - url=endpoint, + # Route through the SSRF guard so that: + # * redirects to private/loopback IPs are blocked, and + # * Authorization / Proxy-Authorization are stripped on + # cross-origin redirects even when validate_no_auth=False + # (issue #947). + response = request_with_ssrf_guard( + method, + endpoint, + session=self.session, headers=request_headers, params=params, data=data, json=json_data, timeout=timeout, + allow_private_ips=self.allow_private_ips, **request_options, ) diff --git a/semantica/ingest/ssrf.py b/semantica/ingest/ssrf.py index 083fbcca..488ae3cf 100644 --- a/semantica/ingest/ssrf.py +++ b/semantica/ingest/ssrf.py @@ -268,6 +268,7 @@ def request_with_ssrf_guard( *, session: Optional[requests.Session] = None, allow_private_ips: bool = False, + allow_private_ips_on_redirect: Optional[bool] = None, max_redirects: int = _DEFAULT_MAX_REDIRECTS, **kwargs: Any, ) -> requests.Response: @@ -277,10 +278,65 @@ def request_with_ssrf_guard( public URL to bounce into private/loopback/link-local space. This helper disables automatic redirects and re-validates each ``Location`` target before issuing the next hop. + + ``allow_private_ips`` trusts the caller's own *url* (e.g. an + operator-configured internal endpoint). That trust follows a redirect + only when the redirect target's host matches the original host (e.g. a + same-host path redirect on a private/localhost server); a redirect to a + *different* host is validated with ``allow_private_ips_on_redirect`` + instead, which defaults to ``allow_private_ips`` for backward + compatibility but can be pinned to ``False`` by callers that want to + trust only the original host and never extend private-IP eligibility to + any other host a redirect chain might reach — otherwise a private-IP- + eligible endpoint could be tricked into redirecting into arbitrary + internal address space (e.g. cloud metadata) the caller never + configured. + + Authorization / credential-header handling (issue #947) + -------------------------------------------------------- + Credentials are stripped from **all** sources that ``requests`` can use to + attach an ``Authorization`` header whenever a redirect changes origin: + + 1. ``kwargs["headers"]`` — per-request header dict (already handled). + 2. ``session.headers`` — session-level headers that ``requests`` merges + automatically; cleared for the hop and restored via ``finally``. + 3. ``kwargs["auth"]`` — per-request auth tuple/callable; removed from the + local ``kwargs`` copy when stripping is required. This copy never + escapes to the caller, so there is nothing to restore. + 4. ``session.auth`` — session-level auth handler that ``requests`` merges + via ``merge_setting(auth, self.auth)`` inside ``prepare_request``; + cleared for the hop and restored via ``finally``. + 5. ``session.trust_env`` — when ``True``, ``requests`` reads ``~/.netrc`` + for the *redirect target* host and calls ``prepare_auth()`` with those + credentials even after sources 3 and 4 are cleared; disabled for + cross-origin hops and restored via ``finally``. + + Leaving any one of these intact allows ``requests`` to re-attach + credentials on the hop to the foreign origin, defeating the header-level + strip. + + Session state that was removed is unconditionally restored in a ``finally`` + block so the session is left in its original state after this call returns, + regardless of how it exits (normal return, exception, redirect cap). The + loop is sequential and single-threaded within one call, so the mutation is + safe as long as the caller does not share the session across concurrent + threads (the standard Semantica pattern: one session per ingestor instance). + + Once credentials have been stripped for a cross-origin hop they are NOT + re-added for subsequent hops in the same chain, even if a later hop + happens to point back to the original host. This prevents credential + resurrection via crafted multi-hop redirect chains. """ kwargs = dict(kwargs) kwargs.pop("allow_redirects", None) + redirect_allow_private_ips = ( + allow_private_ips + if allow_private_ips_on_redirect is None + else allow_private_ips_on_redirect + ) + _original_host = (urlparse(url).hostname or "").lower() + validate_url_for_request(url, allow_private_ips=allow_private_ips) requester = session.request if session is not None else requests.request @@ -288,56 +344,141 @@ def request_with_ssrf_guard( current_method = method.upper() redirects_followed = 0 - while True: - response = requester( - current_method, - current_url, - allow_redirects=False, - **kwargs, - ) + # -- issue #947: snapshot every session-level credential source so we can + # restore them unconditionally when this call exits. + _SENSITIVE = ("Authorization", "Proxy-Authorization") + _session_auth_backup: dict = {} + _session_auth_handler_backup: Any = None # session.auth backup + _session_trust_env_backup: bool = True # session.trust_env backup - if response.status_code not in _REDIRECT_STATUS_CODES: - return response + if session is not None: + for _h in _SENSITIVE: + # requests stores session headers in a case-insensitive dict; + # .get() matches regardless of the casing used at insertion time. + _val = session.headers.get(_h) + if _val is not None: + _session_auth_backup[_h] = _val + # Snapshot session.auth (HTTPBasicAuth, tuple, callable, or None). + _session_auth_handler_backup = session.auth + # Snapshot session.trust_env (controls .netrc / env proxy lookup). + _session_trust_env_backup = session.trust_env - if redirects_followed >= max_redirects: - response.close() - raise ValidationError( - f"Exceeded maximum redirects ({max_redirects}) while " - f"fetching '{url}'" + # Track whether credentials have been stripped for this redirect chain. + # Once stripped they must not reappear on any subsequent hop. + _auth_stripped = False + + try: + while True: + response = requester( + current_method, + current_url, + allow_redirects=False, + **kwargs, ) - location = response.headers.get("Location") - if not location or not str(location).strip(): - response.close() - raise ValidationError( - f"Redirect from '{current_url}' is missing a Location header" + if response.status_code not in _REDIRECT_STATUS_CODES: + return response + + if redirects_followed >= max_redirects: + response.close() + raise ValidationError( + f"Exceeded maximum redirects ({max_redirects}) while " + f"fetching '{url}'" + ) + + location = response.headers.get("Location") + if not location or not str(location).strip(): + response.close() + raise ValidationError( + f"Redirect from '{current_url}' is missing a Location header" + ) + + next_url = urljoin(current_url, str(location).strip()) + next_host = (urlparse(next_url).hostname or "").lower() + # A redirect back to the original host inherits the caller's + # trust in that host (e.g. a same-host path redirect on a + # private/localhost MCP server). A redirect to a *different* + # host must not inherit that trust, even if the original host + # was private/internal — otherwise a compromised or malicious + # endpoint could redirect into arbitrary private address space + # (e.g. cloud metadata) the caller never configured. + hop_allow_private_ips = ( + allow_private_ips + if next_host and next_host == _original_host + else redirect_allow_private_ips ) + validate_url_for_request(next_url, allow_private_ips=hop_allow_private_ips) - next_url = urljoin(current_url, str(location).strip()) - validate_url_for_request(next_url, allow_private_ips=allow_private_ips) + # Do not leak sensitive headers or auth handlers to a different + # origin on redirects. All four credential sources are cleared: + # • kwargs["headers"] — per-request header dict + # • session.headers — session-level header dict + # • kwargs["auth"] — per-request auth tuple/callable + # • session.auth — session-level auth handler + # + # Once stripped (_auth_stripped=True), credentials stay absent for + # the remainder of the chain — even if a later hop targets the + # original host — to prevent credential resurrection. + if _auth_stripped or _should_strip_auth(current_url, next_url): + _auth_stripped = True - # Do not leak sensitive headers to a different origin on redirects: - # reuse the caller's headers only while host, port, and scheme keep - # the credential safe, mirroring requests' should_strip_auth. - if _should_strip_auth(current_url, next_url): - kwargs = dict(kwargs) - headers = dict(kwargs.get("headers") or {}) - for sensitive in ("Authorization", "Proxy-Authorization"): - headers.pop(sensitive, None) - kwargs["headers"] = headers + # 1. Strip from per-request kwargs headers. + kwargs = dict(kwargs) + headers = dict(kwargs.get("headers") or {}) + for sensitive in _SENSITIVE: + headers.pop(sensitive, None) + # Also remove any case variant the caller may have used + # (e.g. "authorization" or "AUTHORIZATION"). + for key in list(headers): + if key.lower() == sensitive.lower(): + del headers[key] + kwargs["headers"] = headers - # Match requests' historical method rewriting for 301/302/303. - if ( - response.status_code in _STRIP_BODY_ON_REDIRECT - and current_method not in {"GET", "HEAD"} - ): - current_method = "GET" - for key in ("data", "json", "files"): - kwargs.pop(key, None) + # 2. Strip per-request auth kwarg so requests cannot call + # prepare_auth() with the caller's credential on this hop. + kwargs.pop("auth", None) - # Params apply to the original request URL only; Location is authoritative. - kwargs.pop("params", None) + # 3. Strip session-level headers so requests cannot re-inject + # them when merging session + per-request headers for this hop. + if session is not None: + for sensitive in _SENSITIVE: + # CaseInsensitiveDict.pop(key, None) handles any casing. + session.headers.pop(sensitive, None) - response.close() - current_url = next_url - redirects_followed += 1 + # 4. Clear session.auth so prepare_request's merge_setting() + # cannot fall back to the session-level auth handler and + # reattach credentials on the foreign-origin hop. + session.auth = None + + # 5. Disable .netrc / environment-proxy credential lookup so + # requests cannot inject credentials from ~/.netrc for the + # redirect target host on this hop. + session.trust_env = False + + # Match requests' historical method rewriting for 301/302/303. + if ( + response.status_code in _STRIP_BODY_ON_REDIRECT + and current_method not in {"GET", "HEAD"} + ): + current_method = "GET" + for key in ("data", "json", "files"): + kwargs.pop(key, None) + + # Params apply to the original request URL only; Location is authoritative. + kwargs.pop("params", None) + + response.close() + current_url = next_url + redirects_followed += 1 + + finally: + # Unconditionally restore every session credential source we touched, + # so the session is in its original state after this call returns or raises. + if session is not None: + if _session_auth_backup: + for _h, _v in _session_auth_backup.items(): + session.headers[_h] = _v + # Restore session.auth to whatever it was before this call. + session.auth = _session_auth_handler_backup + # Restore session.trust_env (.netrc / env-proxy lookup flag). + session.trust_env = _session_trust_env_backup diff --git a/semantica/seed/seed_manager.py b/semantica/seed/seed_manager.py index 16f21ea1..6e52c382 100644 --- a/semantica/seed/seed_manager.py +++ b/semantica/seed/seed_manager.py @@ -501,8 +501,11 @@ class SeedDataManager: else: full_url = api_url - # Prepare headers - request_headers = headers or {} + # Prepare headers — copy the caller's dict so we never mutate it in-place. + # Without the copy, adding "Authorization" here would silently modify the + # caller's original dict and potentially leak the key to subsequent calls + # that reuse the same dict without expecting it to contain credentials. + request_headers = dict(headers) if headers else {} if api_key: request_headers["Authorization"] = f"Bearer {api_key}" diff --git a/tests/ingest/conftest.py b/tests/ingest/conftest.py new file mode 100644 index 00000000..98f913a5 --- /dev/null +++ b/tests/ingest/conftest.py @@ -0,0 +1,35 @@ +""" +Shared pytest fixtures for the ingest test suite. + +The ``mock_dns`` fixture is applied to *every* test in this directory +(``autouse=True``). It stubs out ``socket.getaddrinfo`` inside the SSRF +guard module so that unit tests that mock ``requests.Session.request`` do not +accidentally hit the network for DNS resolution — which would fail in offline +CI environments and cause intermittent timeouts. + +Tests that explicitly need to exercise DNS-related behaviour (e.g. checking +that a hostname resolving to a private IP is blocked) override this fixture +by patching ``semantica.ingest.ssrf.socket.getaddrinfo`` with their own +``side_effect`` *inside* the test body; that inner patch wins because +``unittest.mock.patch`` applies patches in innermost-last order. +""" +from __future__ import annotations + +import socket +from unittest.mock import patch + +import pytest + +_PUBLIC_IP = "93.184.216.34" # example.com — a safe, routable public address + + +@pytest.fixture(autouse=True) +def mock_dns(): + """Map every hostname to a safe public IP for the duration of each test.""" + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", (_PUBLIC_IP, 0)) + ], + ): + yield diff --git a/tests/ingest/test_auth_header_redirect_security.py b/tests/ingest/test_auth_header_redirect_security.py new file mode 100644 index 00000000..882a9d59 --- /dev/null +++ b/tests/ingest/test_auth_header_redirect_security.py @@ -0,0 +1,1063 @@ +"""Security regression tests for issue #947. + +Prevents Authorization / Proxy-Authorization headers from leaking across +cross-origin redirects in request_with_ssrf_guard, MCPClient, and +PublicAPIIngestor. + +Each test is focused on a single, specific security property so that a future +regression immediately pinpoints the broken invariant. +""" +from __future__ import annotations + +import socket +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from semantica.ingest.mcp_client import MCPClient +from semantica.ingest.public_api_ingestor import PublicAPIIngestor +from semantica.ingest.ssrf import request_with_ssrf_guard +from semantica.utils.exceptions import ValidationError + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_PUBLIC_IP = "93.184.216.34" # example.com — public, safe + + +def _public_getaddrinfo(host, *args, **kwargs): + """DNS stub that maps every hostname to a safe public IP.""" + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (_PUBLIC_IP, 0))] + + +def _make_session_with_auth(token: str = "Bearer secret") -> requests.Session: + """Return a real requests.Session with Authorization in session.headers.""" + sess = requests.Session() + sess.headers["Authorization"] = token + return sess + + +def _mock_redirect(location: str, status: int = 302) -> MagicMock: + r = MagicMock() + r.status_code = status + r.headers = {"Location": location} + r.close = MagicMock() + return r + + +def _mock_final(status: int = 200) -> MagicMock: + r = MagicMock() + r.status_code = status + r.headers = {} + r.close = MagicMock() + return r + + +# =========================================================================== +# Section 1 – request_with_ssrf_guard: session.headers stripping (#947) +# =========================================================================== + + +class TestSessionHeadersStripping: + """Authorization stored in session.headers must not reach a foreign origin.""" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_authorization_stripped_on_cross_origin_redirect(self, _): + """session.headers["Authorization"] must not appear in the hop to a new host.""" + sess = _make_session_with_auth() + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]) as mock_req: + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert mock_req.call_count == 2 + # The second call must not carry Authorization in kwargs["headers"]. + second_headers = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second_headers + # Also verify requests won't re-inject it via session (the guard must + # have cleared it from sess.headers before the second call). + assert "Authorization" not in sess.headers or sess.headers.get("Authorization") == "Bearer secret" + # Post-call restoration: session must be restored. + assert sess.headers.get("Authorization") == "Bearer secret" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_headers_cleared_before_second_hop_not_just_restored_after(self, _): + """Prove session.headers["Authorization"] is absent AT CALL TIME of the second hop. + + This test closes the gap where a mock-based test only checks kwargs["headers"] + but not whether session.headers was actually cleared before requests' internal + header-merge would re-inject the credential. + + Strategy: capture a snapshot of sess.headers at each call invocation so we + can assert it was empty during the second hop — not just after the guard returns. + """ + sess = _make_session_with_auth("Bearer proof-token") + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + snapshots: list = [] + + def capturing_side_effect(*args, **kwargs): + # Snapshot what session.headers contain at the exact moment of this call. + snapshots.append(dict(sess.headers)) + return [redirect, final][len(snapshots) - 1] + + with patch.object(sess, "request", side_effect=capturing_side_effect): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert len(snapshots) == 2 + + # Hop 1 (same origin, pre-redirect): Authorization PRESENT in session.headers. + assert snapshots[0].get("Authorization") == "Bearer proof-token", ( + "Authorization must be in session.headers for the first (same-origin) call" + ) + + # Hop 2 (cross-origin): Authorization ABSENT from session.headers. + # This is what prevents requests from re-injecting it via its header-merge step. + assert "Authorization" not in snapshots[1], ( + "Authorization must have been removed from session.headers BEFORE the " + "second (cross-origin) call — removing it only from kwargs is not enough " + "because requests.Session merges session.headers at call time." + ) + + # After the guard returns, session state is fully restored. + assert sess.headers.get("Authorization") == "Bearer proof-token", ( + "session.headers must be restored after the guard returns" + ) + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_authorization_preserved_on_same_origin_redirect(self, _): + """Same-origin redirect must keep Authorization in session.headers untouched.""" + sess = _make_session_with_auth() + redirect = _mock_redirect("https://example.com/page2") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]) as mock_req: + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert mock_req.call_count == 2 + # When no stripping occurred, kwargs["headers"] is unchanged from + # the caller (no headers kwarg was passed here, so it may be absent + # or empty — what matters is that the session header was NOT cleared). + assert sess.headers.get("Authorization") == "Bearer secret" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_credentials_restored_after_successful_request(self, _): + """Session headers must be restored after a redirect chain completes normally.""" + sess = _make_session_with_auth("Bearer my-token") + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert sess.headers.get("Authorization") == "Bearer my-token" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_credentials_restored_after_ssrf_exception(self, _): + """Session headers must be restored even when the guard raises ValidationError.""" + sess = _make_session_with_auth("Bearer my-token") + # Redirect to a loopback address — guard will raise. + redirect = _mock_redirect("http://127.0.0.1/secret") + + with patch.object(sess, "request", return_value=redirect): + with pytest.raises(ValidationError): + request_with_ssrf_guard( + "GET", "https://example.com/start", session=sess + ) + + assert sess.headers.get("Authorization") == "Bearer my-token" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_credentials_restored_after_max_redirects_exceeded(self, _): + """Session headers must be restored when the max-redirect cap is hit.""" + sess = _make_session_with_auth("Bearer loop-token") + hop = _mock_redirect("https://other.example/loop") + + # All hops redirect to the same foreign host → exceeds cap. + with patch.object(sess, "request", return_value=hop): + with pytest.raises(ValidationError, match="Exceeded maximum"): + request_with_ssrf_guard( + "GET", + "https://example.com/start", + session=sess, + max_redirects=2, + ) + + assert sess.headers.get("Authorization") == "Bearer loop-token" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_proxy_authorization_stripped_on_cross_origin_redirect(self, _): + """Proxy-Authorization must be stripped alongside Authorization.""" + sess = requests.Session() + sess.headers["Proxy-Authorization"] = "Basic cHJveHk6cGFzcw==" + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]) as mock_req: + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + second_headers = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Proxy-Authorization" not in second_headers + # Restored after call. + assert "Proxy-Authorization" in sess.headers + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_both_auth_headers_stripped_simultaneously(self, _): + """Both Authorization and Proxy-Authorization must be stripped together.""" + sess = requests.Session() + sess.headers["Authorization"] = "Bearer tok" + sess.headers["Proxy-Authorization"] = "Basic abc" + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]) as mock_req: + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + second_headers = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second_headers + assert "Proxy-Authorization" not in second_headers + # Restored after call. + assert sess.headers.get("Authorization") == "Bearer tok" + assert sess.headers.get("Proxy-Authorization") == "Basic abc" + + +# =========================================================================== +# Section 1b – request_with_ssrf_guard: auth= kwarg and session.auth stripping +# =========================================================================== + + +class TestAuthHandlerStripping: + """kwargs['auth'] and session.auth must not reach a foreign origin. + + requests uses two additional credential channels beyond header dicts: + • auth= kwarg → passed to PreparedRequest.prepare_auth() directly + • session.auth → merged by Session.prepare_request() via merge_setting() + and then calls prepare_auth() — so even if headers are + stripped, a live session.auth re-attaches Authorization. + + Both must be cleared on cross-origin redirect. + """ + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_kwargs_auth_stripped_on_cross_origin_redirect(self, _): + """auth= kwarg must not be forwarded to the second hop on a different host. + + Verifies that the second call to the underlying requester does NOT + receive an 'auth' kwarg, so requests cannot call prepare_auth() and + regenerate an Authorization header for the foreign origin. + """ + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + auth=("user", "secret-password"), + ) + + assert mock_req.call_count == 2 + + # First hop: auth= kwarg is present (same origin, no strip yet). + first_auth = mock_req.call_args_list[0].kwargs.get("auth") + assert first_auth == ("user", "secret-password"), ( + "auth= kwarg must be forwarded on the first (same-origin) hop" + ) + + # Second hop: auth= kwarg must be absent (cross-origin — stripped). + second_auth = mock_req.call_args_list[1].kwargs.get("auth") + assert second_auth is None, ( + "auth= kwarg must be removed before the cross-origin hop so " + "requests cannot call prepare_auth() and reattach Authorization" + ) + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_kwargs_auth_preserved_on_same_origin_redirect(self, _): + """auth= kwarg must survive a same-host redirect unchanged.""" + redirect = _mock_redirect("https://example.com/new-path") + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + auth=("user", "secret-password"), + ) + + assert mock_req.call_count == 2 + second_auth = mock_req.call_args_list[1].kwargs.get("auth") + assert second_auth == ("user", "secret-password"), ( + "auth= kwarg must be kept for same-origin redirects" + ) + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_auth_cleared_before_cross_origin_hop(self, _): + """session.auth must be None AT CALL TIME of the cross-origin hop. + + This test uses the same snapshot-at-invocation technique as the + session.headers equivalent: capture session.auth at the exact moment + each call is issued, so we can prove the handler was absent before + requests' merge_setting() could reattach it. + """ + sess = requests.Session() + sess.auth = ("user", "secret-password") + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + auth_snapshots: list = [] + + def capturing_side_effect(*args, **kwargs): + # Snapshot session.auth at the exact moment of this call. + auth_snapshots.append(sess.auth) + return [redirect, final][len(auth_snapshots) - 1] + + with patch.object(sess, "request", side_effect=capturing_side_effect): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert len(auth_snapshots) == 2 + + # Hop 1 (same origin): session.auth is PRESENT. + assert auth_snapshots[0] == ("user", "secret-password"), ( + "session.auth must be intact for the first (same-origin) call" + ) + + # Hop 2 (cross-origin): session.auth must be ABSENT (None). + assert auth_snapshots[1] is None, ( + "session.auth must have been cleared BEFORE the cross-origin call " + "so requests' merge_setting() cannot reattach the credential" + ) + + # After the guard returns, session.auth must be fully restored. + assert sess.auth == ("user", "secret-password"), ( + "session.auth must be restored after the guard returns" + ) + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_auth_preserved_on_same_origin_redirect(self, _): + """session.auth must not be touched for same-host redirects.""" + sess = requests.Session() + sess.auth = ("user", "secret-password") + redirect = _mock_redirect("https://example.com/page2") + final = _mock_final() + + auth_snapshots: list = [] + + def capturing_side_effect(*args, **kwargs): + auth_snapshots.append(sess.auth) + return [redirect, final][len(auth_snapshots) - 1] + + with patch.object(sess, "request", side_effect=capturing_side_effect): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert len(auth_snapshots) == 2 + # Both hops see session.auth intact. + assert auth_snapshots[0] == ("user", "secret-password") + assert auth_snapshots[1] == ("user", "secret-password") + assert sess.auth == ("user", "secret-password") + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_auth_restored_after_successful_request(self, _): + """session.auth must be restored to its original value after the call.""" + sess = requests.Session() + sess.auth = ("user", "secret-password") + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert sess.auth == ("user", "secret-password") + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_auth_restored_after_ssrf_exception(self, _): + """session.auth must be restored even when the guard raises.""" + sess = requests.Session() + sess.auth = ("user", "secret-password") + # Redirect to loopback — guard raises ValidationError. + redirect = _mock_redirect("http://127.0.0.1/secret") + + with patch.object(sess, "request", return_value=redirect): + with pytest.raises(ValidationError): + request_with_ssrf_guard( + "GET", "https://example.com/start", session=sess + ) + + assert sess.auth == ("user", "secret-password") + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_auth_none_by_default_remains_none(self, _): + """When session.auth is None (default), the finally block must not set it + to something unexpected — restoring None is a no-op, not a corruption.""" + sess = requests.Session() + assert sess.auth is None + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert sess.auth is None + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_kwargs_auth_does_not_reappear_in_multihop_chain(self, _): + """Once auth= is stripped at hop 2, it must not reappear at hop 3.""" + hop1 = _mock_redirect("https://other.example/step2") # cross-origin: strip + hop2 = _mock_redirect("https://other.example/final") # same host as hop1: stay stripped + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[hop1, hop2, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + auth=("user", "pass"), + ) + + assert mock_req.call_count == 3 + # Hop 1: auth present (same origin). + assert mock_req.call_args_list[0].kwargs.get("auth") == ("user", "pass") + # Hop 2: stripped. + assert mock_req.call_args_list[1].kwargs.get("auth") is None + # Hop 3: stays stripped — no resurrection. + assert mock_req.call_args_list[2].kwargs.get("auth") is None + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_trust_env_disabled_before_cross_origin_hop(self, _): + """session.trust_env must be False AT CALL TIME of the cross-origin hop. + + When trust_env=True, requests reads ~/.netrc for the redirect target host + and calls prepare_auth() with those credentials — even after session.auth + and kwargs['auth'] are cleared. Disabling trust_env before the hop closes + this bypass channel. + """ + sess = requests.Session() + sess.trust_env = True # explicit default + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + trust_env_snapshots: list = [] + + def capturing_side_effect(*args, **kwargs): + trust_env_snapshots.append(sess.trust_env) + return [redirect, final][len(trust_env_snapshots) - 1] + + with patch.object(sess, "request", side_effect=capturing_side_effect): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert len(trust_env_snapshots) == 2 + + # Hop 1 (same origin): trust_env is True (unchanged). + assert trust_env_snapshots[0] is True, ( + "trust_env must be unchanged for the first (same-origin) call" + ) + + # Hop 2 (cross-origin): trust_env must be False to block .netrc lookup. + assert trust_env_snapshots[1] is False, ( + "trust_env must be False BEFORE the cross-origin call to prevent " + "requests from looking up ~/.netrc credentials for the redirect target" + ) + + # After the guard returns, trust_env must be restored. + assert sess.trust_env is True, ( + "session.trust_env must be restored after the guard returns" + ) + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_trust_env_restored_after_exception(self, _): + """session.trust_env must be restored even when the guard raises.""" + sess = requests.Session() + sess.trust_env = True + redirect = _mock_redirect("http://127.0.0.1/secret") # will raise ValidationError + + with patch.object(sess, "request", return_value=redirect): + with pytest.raises(ValidationError): + request_with_ssrf_guard( + "GET", "https://example.com/start", session=sess + ) + + assert sess.trust_env is True + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_trust_env_false_stays_false_after_call(self, _): + """If trust_env was already False, it must stay False after the call.""" + sess = requests.Session() + sess.trust_env = False # caller explicitly disabled .netrc + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch.object(sess, "request", side_effect=[redirect, final]): + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + assert sess.trust_env is False # restored to the original False value + + +# =========================================================================== +# Section 2 – request_with_ssrf_guard: credential resurrection prevention +# =========================================================================== + + +class TestCredentialResurrection: + """Stripped credentials must not reappear for later hops in the same chain.""" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_credentials_do_not_reappear_after_cross_origin_hop(self, _): + """A subsequent same-origin-as-hop-2 redirect must not restore the credential.""" + # Chain: example.com → other.example (strip) → other.example/page2 (stay stripped) + hop1 = _mock_redirect("https://other.example/step2") + hop2 = _mock_redirect("https://other.example/final") # same host as hop1 target + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[hop1, hop2, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + headers={"Authorization": "Bearer secret"}, + ) + + assert mock_req.call_count == 3 + # Hop 1 (example.com): credential present + h1 = mock_req.call_args_list[0].kwargs.get("headers", {}) + assert h1.get("Authorization") == "Bearer secret" + # Hop 2 (other.example): stripped + h2 = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in h2 + # Hop 3 (still other.example): stays stripped — must NOT reappear + h3 = mock_req.call_args_list[2].kwargs.get("headers", {}) + assert "Authorization" not in h3 + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_auth_does_not_reappear_in_multihop_chain(self, _): + """session.headers auth stripped for hop 2 must stay absent for hop 3.""" + sess = _make_session_with_auth("Bearer multi") + hop1 = _mock_redirect("https://other.example/step2") # cross-origin: strip + hop2 = _mock_redirect("https://other.example/final") # same-as-hop1: stay stripped + final = _mock_final() + + with patch.object(sess, "request", side_effect=[hop1, hop2, final]) as mock_req: + request_with_ssrf_guard("GET", "https://example.com/start", session=sess) + + # After the call the session is restored. + assert sess.headers.get("Authorization") == "Bearer multi" + + h2 = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in h2 + h3 = mock_req.call_args_list[2].kwargs.get("headers", {}) + assert "Authorization" not in h3 + + +# =========================================================================== +# Section 3 – request_with_ssrf_guard: specific redirect-type coverage +# =========================================================================== + + +class TestRedirectTypesAndOriginChanges: + """Per-type and per-scenario auth-stripping rules.""" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_strips_on_307_cross_origin(self, _): + """307 Temporary Redirect to a different host must strip credentials.""" + redirect = MagicMock() + redirect.status_code = 307 + redirect.headers = {"Location": "https://other.example/final"} + redirect.close = MagicMock() + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + headers={"Authorization": "Bearer tok"}, + ) + + second = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_strips_on_308_cross_origin(self, _): + """308 Permanent Redirect to a different host must strip credentials.""" + redirect = MagicMock() + redirect.status_code = 308 + redirect.headers = {"Location": "https://other.example/final"} + redirect.close = MagicMock() + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + headers={"Authorization": "Bearer tok"}, + ) + + second = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_strips_on_port_change(self, _): + """Redirect that changes the port (non-default) must strip credentials.""" + redirect = _mock_redirect("https://example.com:8443/final") + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + headers={"Authorization": "Bearer tok"}, + ) + + second = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_strips_on_subdomain_change(self, _): + """Redirect from apex to subdomain (different hostname) must strip credentials.""" + redirect = _mock_redirect("https://api.example.com/final") + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + headers={"Authorization": "Bearer tok"}, + ) + + second = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_keeps_on_https_443_explicit_to_implicit(self, _): + """https://example.com:443 → https://example.com (same, just drop explicit port).""" + redirect = _mock_redirect("https://example.com/final") + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com:443/start", + headers={"Authorization": "Bearer tok"}, + ) + + second = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert second.get("Authorization") == "Bearer tok" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_case_insensitive_header_stripped(self, _): + """Lowercase/UPPERCASE variants of Authorization must also be stripped.""" + redirect = _mock_redirect("https://other.example/final") + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "https://example.com/start", + # Pass a lowercase variant to verify case-insensitive stripping. + headers={"authorization": "Bearer lower", "AUTHORIZATION": "Bearer upper"}, + ) + + second = mock_req.call_args_list[1].kwargs.get("headers", {}) + for key in second: + assert key.lower() != "authorization", ( + f"Authorization header variant {key!r} was not stripped" + ) + + +class TestAllowPrivateIpsOnRedirect: + """allow_private_ips must not extend to a redirect target on a different host.""" + + def test_cross_host_redirect_to_private_ip_is_blocked_when_pinned(self): + """allow_private_ips_on_redirect=False must block a cross-host hop into private space.""" + redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/") + + with patch( + "semantica.ingest.ssrf.requests.request", + return_value=redirect, + ): + with pytest.raises(ValidationError, match="blocked"): + request_with_ssrf_guard( + "GET", + "https://trusted.example.com/start", + allow_private_ips=True, + allow_private_ips_on_redirect=False, + ) + + def test_same_host_redirect_keeps_private_ip_trust_when_pinned(self): + """A same-host redirect must still inherit the original host's trust.""" + redirect = _mock_redirect("http://localhost:8000/v2") + final = _mock_final() + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + request_with_ssrf_guard( + "GET", + "http://localhost:8000/start", + allow_private_ips=True, + allow_private_ips_on_redirect=False, + ) + + assert mock_req.call_count == 2 + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_default_behavior_unchanged_without_the_new_kwarg(self, _): + """Existing callers that never pass allow_private_ips_on_redirect keep old behavior.""" + redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/") + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, _mock_final()], + ) as mock_req: + # allow_private_ips=True with no override: redirect target validation + # falls back to allow_private_ips, matching pre-fix behavior for the + # existing opt-in ingestors (web/feed/api/public-api/seed). + request_with_ssrf_guard( + "GET", + "https://trusted.example.com/start", + allow_private_ips=True, + ) + + assert mock_req.call_count == 2 + + +# =========================================================================== +# Section 4 – MCPClient: redirect auth-stripping (#947) +# =========================================================================== + + +class TestMCPClientAuthRedirect: + """MCPClient._send_request_http must not leak credentials on cross-origin redirect.""" + + def _mock_mcp_response(self, payload=None): + resp = MagicMock() + resp.status_code = 200 + resp.headers = {} + resp.raise_for_status = MagicMock() + resp.json.return_value = payload or { + "jsonrpc": "2.0", + "id": 1, + "result": {"serverInfo": {}, "capabilities": {}}, + } + return resp + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_cross_origin_redirect_strips_authorization(self, _): + """Authorization must not reach a different host after an MCP server redirect.""" + redirect = _mock_redirect("https://other.example/mcp") + redirect.status_code = 302 + final = self._mock_mcp_response() + + client = MCPClient( + url="https://mcp.example.com/mcp", + headers={"Authorization": "Bearer mcp-token"}, + ) + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + assert mock_req.call_count == 2 + second_headers = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second_headers + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_same_origin_redirect_preserves_authorization(self, _): + """Same-host redirect must keep Authorization intact.""" + redirect = _mock_redirect("https://mcp.example.com/mcp/v2") + redirect.status_code = 301 + final = self._mock_mcp_response() + + client = MCPClient( + url="https://mcp.example.com/mcp", + headers={"Authorization": "Bearer mcp-token"}, + ) + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + assert mock_req.call_count == 2 + second_headers = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert second_headers.get("Authorization") == "Bearer mcp-token" + + def test_localhost_mcp_server_is_not_blocked(self): + """localhost MCP endpoints must work (allow_private_ips=True).""" + final = self._mock_mcp_response() + client = MCPClient(url="http://localhost:8000/mcp") + + with patch( + "semantica.ingest.ssrf.requests.request", + return_value=final, + ) as mock_req: + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + mock_req.assert_called_once() + + def test_loopback_ip_mcp_server_is_not_blocked(self): + """127.0.0.1 MCP endpoints must work (allow_private_ips=True).""" + final = self._mock_mcp_response() + client = MCPClient(url="http://127.0.0.1:9000/mcp") + + with patch( + "semantica.ingest.ssrf.requests.request", + return_value=final, + ) as mock_req: + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + mock_req.assert_called_once() + + def test_same_host_redirect_on_private_mcp_server_is_not_blocked(self): + """A same-host redirect on a trusted private/localhost MCP server must still work.""" + redirect = _mock_redirect("http://localhost:8000/mcp/v2") + final = self._mock_mcp_response() + + client = MCPClient(url="http://localhost:8000/mcp") + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + assert mock_req.call_count == 2 + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_redirect_to_private_ip_is_blocked(self, _): + """A redirect from a public MCP server to a private/internal IP must be blocked. + + allow_private_ips=True trusts the operator-configured MCP host itself; + it must not let a compromised or malicious server redirect the client + into private address space (e.g. cloud metadata) via a cross-host hop. + """ + redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/") + + client = MCPClient(url="https://mcp.example.com/mcp") + + with patch( + "semantica.ingest.ssrf.requests.request", + return_value=redirect, + ): + with pytest.raises(ValidationError, match="blocked"): + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_scheme_downgrade_strips_authorization(self, _): + """https MCP server that redirects to http must strip the credential.""" + redirect = _mock_redirect("http://mcp.example.com/mcp") + redirect.status_code = 302 + final = self._mock_mcp_response() + + client = MCPClient( + url="https://mcp.example.com/mcp", + headers={"Authorization": "Bearer downgrade-test"}, + ) + + with patch( + "semantica.ingest.ssrf.requests.request", + side_effect=[redirect, final], + ) as mock_req: + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + second_headers = mock_req.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second_headers + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_max_redirect_cap_respected(self, _): + """Infinite redirect loop must raise ValidationError.""" + hop = _mock_redirect("https://mcp.example.com/mcp/loop") + + client = MCPClient(url="https://mcp.example.com/mcp") + + with patch( + "semantica.ingest.ssrf.requests.request", + return_value=hop, + ): + with pytest.raises((ValidationError, Exception), match="[Rr]edirect|[Ee]xceeded"): + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_allow_redirects_false_enforced(self, _): + """The guard must pass allow_redirects=False on every hop.""" + final = self._mock_mcp_response() + client = MCPClient( + url="https://mcp.example.com/mcp", + headers={"Authorization": "Bearer tok"}, + ) + + with patch( + "semantica.ingest.ssrf.requests.request", + return_value=final, + ) as mock_req: + client._send_request_http({"jsonrpc": "2.0", "method": "ping"}) + + assert mock_req.call_args.kwargs.get("allow_redirects") is False + + +# =========================================================================== +# Section 5 – PublicAPIIngestor: redirect auth-stripping (#947) +# =========================================================================== + + +def _mock_public_response(status: int = 200, json_payload=None) -> MagicMock: + resp = MagicMock() + resp.status_code = status + resp.headers = {"Content-Type": "application/json"} + resp.json.return_value = json_payload or [{"id": 1}] + resp.text = "" + if status >= 400: + resp.raise_for_status.side_effect = requests.exceptions.HTTPError( + f"{status} error" + ) + else: + resp.raise_for_status.return_value = None + resp.close = MagicMock() + return resp + + +class TestPublicAPIIngestorRedirectSecurity: + """PublicAPIIngestor must not leak credentials on redirect and must block SSRF.""" + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_redirect_to_private_ip_blocked_in_detect(self, _): + """detect_public_api() must reject a redirect that resolves to a private IP.""" + redirect = _mock_redirect("http://169.254.169.254/latest/meta-data/") + + with patch("requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.headers = {} + mock_session.request.return_value = redirect + mock_session.request.return_value.close = MagicMock() + + ingestor = PublicAPIIngestor(rate_limit_delay=0) + with pytest.raises(ValidationError, match="blocked"): + ingestor.detect_public_api("https://example.com/api") + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_redirect_to_private_ip_blocked_in_ingest(self, _): + """ingest_public_api() must reject a redirect that resolves to a private IP.""" + redirect = _mock_redirect("http://10.0.0.1/internal") + + with patch("requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.headers = {} + mock_session.request.return_value = redirect + mock_session.request.return_value.close = MagicMock() + + ingestor = PublicAPIIngestor(rate_limit_delay=0) + with pytest.raises(ValidationError, match="blocked"): + ingestor.ingest_public_api("https://example.com/api") + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_session_auth_not_leaked_on_cross_origin_redirect_ingest(self, _): + """Session-level auth header must not reach a foreign host via ingest_public_api.""" + redirect = _mock_redirect("https://other.example/api") + final = _mock_public_response(json_payload=[{"id": 1}]) + + # Simulate a session that somehow has Authorization (e.g. misconfiguration). + with patch("requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.headers = {"Authorization": "Bearer leaked"} + mock_session.request.side_effect = [redirect, final] + + ingestor = PublicAPIIngestor( + rate_limit_delay=0, validate_no_auth=False + ) + # Inject the auth-bearing session directly. + ingestor.session = mock_session + + ingestor.ingest_public_api("https://example.com/api") + + assert mock_session.request.call_count == 2 + second_headers = mock_session.request.call_args_list[1].kwargs.get( + "headers", {} + ) + assert "Authorization" not in second_headers + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_allow_redirects_false_enforced_in_detect(self, _): + """detect_public_api() must pass allow_redirects=False to the underlying call.""" + final = _mock_public_response() + + with patch("requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.headers = {} + mock_session.request.return_value = final + + ingestor = PublicAPIIngestor(rate_limit_delay=0) + ingestor.detect_public_api("https://example.com/api") + + assert mock_session.request.call_args.kwargs.get("allow_redirects") is False + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_allow_redirects_false_enforced_in_ingest(self, _): + """ingest_public_api() must pass allow_redirects=False to the underlying call.""" + final = _mock_public_response(json_payload=[{"id": 1}]) + + with patch("requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.headers = {} + mock_session.request.return_value = final + + ingestor = PublicAPIIngestor(rate_limit_delay=0) + ingestor.ingest_public_api("https://example.com/api") + + assert mock_session.request.call_args.kwargs.get("allow_redirects") is False + + @patch("semantica.ingest.ssrf.socket.getaddrinfo", side_effect=_public_getaddrinfo) + def test_validate_no_auth_false_does_not_bypass_redirect_stripping(self, _): + """Even with validate_no_auth=False the guard strips auth on cross-origin redirect.""" + redirect = _mock_redirect("https://other.example/api") + final = _mock_public_response(json_payload=[{"id": 1}]) + + with patch("requests.Session") as MockSession: + mock_session = MockSession.return_value + mock_session.headers = {} + mock_session.request.side_effect = [redirect, final] + + ingestor = PublicAPIIngestor( + rate_limit_delay=0, validate_no_auth=False + ) + ingestor.ingest_public_api( + "https://example.com/api", + headers={"Authorization": "Bearer should-be-stripped"}, + ) + + second_headers = mock_session.request.call_args_list[1].kwargs.get( + "headers", {} + ) + assert "Authorization" not in second_headers diff --git a/tests/ingest/test_cookbook_integration.py b/tests/ingest/test_cookbook_integration.py index c5120d65..ad470558 100644 --- a/tests/ingest/test_cookbook_integration.py +++ b/tests/ingest/test_cookbook_integration.py @@ -10,19 +10,20 @@ class TestCookbookIntegration: @pytest.fixture def mock_mcp_server(self): - # We need to patch both httpx and requests because MCPClient tries httpx first - with patch("httpx.post") as mock_httpx_post, \ - patch("requests.post") as mock_requests_post: - - def side_effect(url, json=None, **kwargs): + # MCPClient._send_request_http now routes through request_with_ssrf_guard, + # which calls requests.request (not httpx.post / requests.post directly). + # Patch at the point where the guard issues the actual HTTP call. + with patch("semantica.ingest.ssrf.requests.request") as mock_request: + + def side_effect(method, url, json=None, **kwargs): if not json: return MagicMock() - - method = json.get("method") + + rpc_method = json.get("method") response_mock = MagicMock() response_mock.status_code = 200 - - if method == "initialize": + + if rpc_method == "initialize": response_mock.json.return_value = { "jsonrpc": "2.0", "id": json.get("id"), @@ -32,7 +33,7 @@ class TestCookbookIntegration: "serverInfo": {"name": "test_server", "version": "1.0"} } } - elif method == "resources/list": + elif rpc_method == "resources/list": response_mock.json.return_value = { "jsonrpc": "2.0", "id": json.get("id"), @@ -44,7 +45,7 @@ class TestCookbookIntegration: ] } } - elif method == "tools/list": + elif rpc_method == "tools/list": response_mock.json.return_value = { "jsonrpc": "2.0", "id": json.get("id"), @@ -56,7 +57,7 @@ class TestCookbookIntegration: ] } } - elif method == "resources/read": + elif rpc_method == "resources/read": response_mock.json.return_value = { "jsonrpc": "2.0", "id": json.get("id"), @@ -66,13 +67,13 @@ class TestCookbookIntegration: ] } } - elif method == "tools/call": + elif rpc_method == "tools/call": tool_name = json.get("params", {}).get("name") content = [{"type": "text", "text": "Tool Output"}] - + if tool_name == "query_inventory": content = [{"type": "text", "text": '{"warehouse_id": "WH001", "level": 100}'}] - + response_mock.json.return_value = { "jsonrpc": "2.0", "id": json.get("id"), @@ -86,12 +87,11 @@ class TestCookbookIntegration: "id": json.get("id"), "result": {} } - + return response_mock - - mock_httpx_post.side_effect = side_effect - mock_requests_post.side_effect = side_effect - yield mock_httpx_post + + mock_request.side_effect = side_effect + yield mock_request def test_financial_data_integration(self, mock_mcp_server): """ diff --git a/tests/ingest/test_public_api_ingestor.py b/tests/ingest/test_public_api_ingestor.py index 61119427..920a99c8 100644 --- a/tests/ingest/test_public_api_ingestor.py +++ b/tests/ingest/test_public_api_ingestor.py @@ -198,15 +198,82 @@ def test_public_api_detection_reports_auth_required() -> None: headers={"WWW-Authenticate": "Bearer"}, ) - detection = PublicAPIIngestor(rate_limit_delay=0).detect_public_api( - "https://api.example.com/private" - ) + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(None, None, None, None, ("93.184.216.34", 0))], + ): + detection = PublicAPIIngestor(rate_limit_delay=0).detect_public_api( + "https://api.example.com/private" + ) assert detection.is_public is False assert detection.requires_auth is True assert detection.response_status == 401 +def test_detect_public_api_propagates_ssrf_validation_error() -> None: + """detect_public_api() must surface ValidationError, not swallow it. + + request_with_ssrf_guard() raises ValidationError (not + requests.exceptions.RequestException) for SSRF-blocked hosts, so + detect_public_api()'s error handling must catch it explicitly like its + sibling ingest_public_api() already does. + """ + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(None, None, None, None, ("127.0.0.1", 0))], + ): + with pytest.raises(ValidationError): + PublicAPIIngestor(rate_limit_delay=0).detect_public_api( + "https://blocked.example.com/data" + ) + + +def test_detect_public_api_rejects_duplicate_session_and_allow_private_ips_kwargs() -> None: + """Passing session/allow_private_ips through **options must not crash. + + Both are always supplied explicitly to request_with_ssrf_guard(); caller + copies must be dropped from **options rather than causing a + 'got multiple values for keyword argument' TypeError. + """ + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + headers={"Content-Type": "application/json"} + ) + + detection = PublicAPIIngestor(rate_limit_delay=0).detect_public_api( + "https://jsonplaceholder.typicode.com/posts", + allow_private_ips=True, + session=object(), + ) + + assert detection.is_public is True + + +def test_ingest_public_api_rejects_duplicate_session_and_allow_private_ips_kwargs() -> None: + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + mock_session.request.return_value = _mock_response( + json_payload=[{"id": 1}], + headers={"Content-Type": "application/json"}, + ) + + result = PublicAPIIngestor(rate_limit_delay=0).ingest_public_api( + "https://jsonplaceholder.typicode.com/posts", + allow_private_ips=True, + session=object(), + ) + + assert result.response_status == 200 + + def test_public_api_ingestor_rejects_authentication_inputs() -> None: with patch("requests.Session") as mock_session_class: mock_session = mock_session_class.return_value @@ -238,10 +305,14 @@ def test_public_api_ingestor_parses_string_boolean_config() -> None: config={"validate_no_auth": "false"}, rate_limit_delay=0, ) - result = ingestor.ingest_public_api( - "https://api.example.com/data", - headers={"Authorization": "Bearer token"}, - ) + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(None, None, None, None, ("93.184.216.34", 0))], + ): + result = ingestor.ingest_public_api( + "https://api.example.com/data", + headers={"Authorization": "Bearer token"}, + ) assert ingestor.validate_no_auth is False assert result.data == payload diff --git a/tests/ingest/test_submodules.py b/tests/ingest/test_submodules.py index e17cd330..cac9c092 100644 --- a/tests/ingest/test_submodules.py +++ b/tests/ingest/test_submodules.py @@ -195,89 +195,62 @@ class TestMCPIngestor: class TestMCPClient: def test_call_tool(self): - # Patch requests.post globally if requests is used, or httpx.post if httpx is used. - # The code tries importing httpx, then requests. - # We should patch both or ensure we catch the right one. - # Simpler to patch sys.modules to simulate httpx missing, then patch requests. - - with patch.dict(sys.modules, {'httpx': None}): - with patch("requests.post") as mock_post: - mock_response = MagicMock() - mock_response.status_code = 200 - - # Sequence of calls: - # 1. connect() calls _connect_http() -> calls _initialize() -> calls _send_request() - # _send_request() calls requests.post with method="initialize" - # 2. call_tool() calls _send_request() with method="tools/call" - - # Response for initialize - init_response = { - "jsonrpc": "2.0", - "result": {"serverInfo": {"name": "test", "version": "1.0"}}, - "id": 1 - } - - # Response for tool call - tool_response = { - "jsonrpc": "2.0", - "result": {"content": [{"type": "text", "text": "Tool Result"}]}, - "id": 2 - } - - mock_response.json.side_effect = [init_response, tool_response] - mock_post.return_value = mock_response - - client = MCPClient(url="http://localhost:8000") - client.connect() - - result = client.call_tool("my_tool", {"arg": "val"}) - - # result is the dict returned by tool call? - # call_tool returns dict? - # Check MCPClient.call_tool implementation - # It calls _send_request, which returns response.json(). - # But wait, call_tool might process the result. - # Let's check call_tool implementation in mcp_client.py (not read yet, but assumed). - # Wait, I read mcp_client.py but didn't check call_tool specifically. - # Assuming call_tool returns result part or whole response. - - # Actually, let's verify call_tool in mcp_client.py - pass + # MCPClient._send_request_http now routes through request_with_ssrf_guard, + # which calls requests.request (not requests.post) with allow_redirects=False. + # Patch the requests.request call inside ssrf.py. + with patch("semantica.ingest.ssrf.requests.request") as mock_request: + mock_response = MagicMock() + mock_response.status_code = 200 + + # Response for initialize + init_response = { + "jsonrpc": "2.0", + "result": {"serverInfo": {"name": "test", "version": "1.0"}}, + "id": 1, + } + + # Response for tool call + tool_response = { + "jsonrpc": "2.0", + "result": {"content": [{"type": "text", "text": "Tool Result"}]}, + "id": 2, + } + + mock_response.json.side_effect = [init_response, tool_response] + mock_request.return_value = mock_response + + client = MCPClient(url="http://localhost:8000") + client.connect() + + client.call_tool("my_tool", {"arg": "val"}) def test_call_tool_mock_check(self): - # Redoing the test with more specific mocking logic - with patch.dict(sys.modules, {'httpx': None}): - with patch("requests.post") as mock_post: - mock_response = MagicMock() - mock_response.status_code = 200 - - # initialize response - init_response = { - "jsonrpc": "2.0", - "result": {"serverInfo": {"name": "test", "version": "1.0"}}, - "id": 1 - } - - # tool call response - Assuming call_tool returns the 'result' part of JSON-RPC response - # If call_tool implementation wraps it, we need to know. - # Let's assume standard behavior for now. - tool_response = { - "jsonrpc": "2.0", - "result": {"content": [{"type": "text", "text": "Tool Result"}]}, - "id": 2 - } - - mock_response.json.side_effect = [init_response, tool_response] - mock_post.return_value = mock_response - - client = MCPClient(url="http://localhost:8000") - client.connect() - - result = client.call_tool("my_tool", {"arg": "val"}) - - # Verify result. - # If call_tool returns the 'result' dict from JSON-RPC: - assert result["content"] == [{"type": "text", "text": "Tool Result"}] + # Redo with the corrected patch target. + with patch("semantica.ingest.ssrf.requests.request") as mock_request: + mock_response = MagicMock() + mock_response.status_code = 200 + + init_response = { + "jsonrpc": "2.0", + "result": {"serverInfo": {"name": "test", "version": "1.0"}}, + "id": 1, + } + + tool_response = { + "jsonrpc": "2.0", + "result": {"content": [{"type": "text", "text": "Tool Result"}]}, + "id": 2, + } + + mock_response.json.side_effect = [init_response, tool_response] + mock_request.return_value = mock_response + + client = MCPClient(url="http://localhost:8000") + client.connect() + + result = client.call_tool("my_tool", {"arg": "val"}) + + assert result["content"] == [{"type": "text", "text": "Tool Result"}] class TestGDriveIngestor: def test_init_raises_if_no_google_libs(self): diff --git a/tests/test_seed_manager.py b/tests/test_seed_manager.py index c66149cc..95d490a1 100644 --- a/tests/test_seed_manager.py +++ b/tests/test_seed_manager.py @@ -209,6 +209,60 @@ def test_load_from_api_allows_private_when_configured(mock_guard, seed_manager): call_kwargs = mock_guard.call_args[1] assert call_kwargs["allow_private_ips"] is True + +@patch("semantica.seed.seed_manager.request_with_ssrf_guard") +def test_load_from_api_does_not_mutate_caller_headers_dict(mock_guard, seed_manager): + """Regression test for issue #947 audit: load_from_api must not mutate the + caller's headers dict in-place when api_key is provided. + + Before the fix, ``request_headers = headers or {}`` aliased the caller's dict. + Writing ``request_headers["Authorization"] = ...`` then silently modified the + caller's original dict, potentially leaking credentials to subsequent calls + that reused the same headers dict without expecting it to carry Authorization. + """ + mock_response = MagicMock() + mock_response.json.return_value = {"results": []} + mock_guard.return_value = mock_response + + # Caller owns this dict and expects it to be unchanged after the call. + original_headers = {"X-Custom-Header": "value"} + headers_before = dict(original_headers) # snapshot + + seed_manager.load_from_api( + api_url="http://api.example.com", + api_key="secret-key", + headers=original_headers, + ) + + # The caller's dict must be unchanged — Authorization must NOT have been added. + assert original_headers == headers_before, ( + "load_from_api must not mutate the caller's headers dict; " + f"expected {headers_before!r}, got {original_headers!r}" + ) + + # The guard must still have received Authorization (in its own copy). + call_kwargs = mock_guard.call_args[1] + guard_headers = call_kwargs.get("headers", {}) + assert guard_headers.get("Authorization") == "Bearer secret-key" + + +@patch("semantica.seed.seed_manager.request_with_ssrf_guard") +def test_load_from_api_does_not_mutate_empty_headers_dict(mock_guard, seed_manager): + """When headers=None, a fresh dict is created — no aliasing to a shared mutable default.""" + mock_response = MagicMock() + mock_response.json.return_value = {"results": []} + mock_guard.return_value = mock_response + + seed_manager.load_from_api( + api_url="http://api.example.com", + api_key="key", + headers=None, + ) + + call_kwargs = mock_guard.call_args[1] + guard_headers = call_kwargs.get("headers", {}) + assert guard_headers.get("Authorization") == "Bearer key" + def test_load_source(seed_manager, temp_data_dir): json_file = temp_data_dir / "source.json" with open(json_file, "w") as f: