mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(security): prevent Authorization header leakage across redirects (#947) (#1067)

Browse Source 
* 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 <kaifahmad087@gmail.com>
This commit is contained in:
Sameer Kadam
2026-08-17 18:55:38 +05:30
committed by GitHub
co-authored by KaifAhmad1
parent eedf1425ca
commit 04602a0e0e
11 changed files with 1561 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+10
View File
@@ -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
semantica/ingest/mcp_client.py
+26 -23
View File
@@ -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
semantica/ingest/public_api_ingestor.py
+32 -6
View File
@@ -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,
)
semantica/ingest/ssrf.py
+184 -43
View File
@@ -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
semantica/seed/seed_manager.py
+5 -2
View File
@@ -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}"
tests/ingest/conftest.py
+35
View File
@@ -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
tests/ingest/test_auth_header_redirect_security.py
+1063
View File
File diff suppressed because it is too large Load Diff
tests/ingest/test_cookbook_integration.py
+20 -20
View File
@@ -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):
"""
tests/ingest/test_public_api_ingestor.py
+78 -7
View File
@@ -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
tests/ingest/test_submodules.py
+54 -81
View File
@@ -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):
tests/test_seed_manager.py
+54
View File
@@ -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: