From 2317688cf59be313a816b622e55835bec29b7e86 Mon Sep 17 00:00:00 2001 From: Yunare Maia Date: Wed, 12 Aug 2026 23:09:38 +0000 Subject: [PATCH 1/3] feat(seed): add allow_private_ips opt-in for trusted internal API sources (Closes #943) SeedDataManager.load_from_api now delegates to the shared SSRF guard (semantica/ingest/ssrf.py, added in #906) instead of raw requests.get, gaining redirect validation and bounded DNS resolution for free. New config option allow_private_ips (parsed via the shared parse_bool helper) lets trusted internal deployments load from private APIs while the secure default (block private/loopback/link-local) is unchanged. Tests updated to mock request_with_ssrf_guard; new tests cover the block-by-default behavior and the opt-in flag reaching the guard. 19/19 green in test_seed_manager.py, 25/25 across both seed suites. Signed-off-by: Yunare Maia --- semantica/seed/seed_manager.py | 22 ++++++++++++++++++++-- tests/test_seed_manager.py | 32 ++++++++++++++++++++++++++++---- 2 files changed, 48 insertions(+), 6 deletions(-) diff --git a/semantica/seed/seed_manager.py b/semantica/seed/seed_manager.py index 0d3bb0461..c41f0ff6a 100644 --- a/semantica/seed/seed_manager.py +++ b/semantica/seed/seed_manager.py @@ -43,6 +43,7 @@ from ..utils.logging import get_logger from ..utils.progress_tracker import get_progress_tracker from ..utils.types import EntityDict, RelationshipDict +from ..ingest.ssrf import parse_bool, request_with_ssrf_guard @dataclass @@ -453,6 +454,11 @@ def load_from_api( 'entities', 'data', 'results', 'items' keys). Automatically adds entity_type, relationship_type, and source metadata if provided. + SSRF protection is enabled by default: URLs resolving to private, + loopback, or link-local addresses are rejected. For trusted internal + deployments, pass ``allow_private_ips=True`` in the manager config to + opt in (documented for internal use only). + Args: api_url: Base API URL endpoint: Optional API endpoint path (appended to api_url) @@ -491,8 +497,20 @@ def load_from_api( if api_key: request_headers["Authorization"] = f"Bearer {api_key}" - # Make API request - response = requests.get(full_url, headers=request_headers, timeout=30) + # SSRF guard: reject private/loopback/link-local targets by default. + # Trusted internal deployments can opt in via config + # (allow_private_ips=True) — see issue #943. + allow_private = parse_bool(self.config.get("allow_private_ips", False)) + + # Make API request (request_with_ssrf_guard validates the URL and + # every redirect before each hop) + response = request_with_ssrf_guard( + "GET", + full_url, + headers=request_headers, + timeout=30, + allow_private_ips=allow_private, + ) response.raise_for_status() # Parse response diff --git a/tests/test_seed_manager.py b/tests/test_seed_manager.py index 0f73b40d9..31f966f1d 100644 --- a/tests/test_seed_manager.py +++ b/tests/test_seed_manager.py @@ -147,11 +147,11 @@ def test_load_from_database_import_error(seed_manager): seed_manager.load_from_database("sqlite:///:memory:", query="SELECT 1") assert "Database ingestion module not available" in str(excinfo.value) -@patch("requests.get") -def test_load_from_api(mock_get, seed_manager): +@patch("semantica.seed.seed_manager.request_with_ssrf_guard") +def test_load_from_api(mock_guard, seed_manager): mock_response = MagicMock() mock_response.json.return_value = {"results": [{"id": 1, "name": "Alice"}]} - mock_get.return_value = mock_response + mock_guard.return_value = mock_response records = seed_manager.load_from_api( api_url="http://api.example.com", @@ -162,7 +162,31 @@ def test_load_from_api(mock_get, seed_manager): assert len(records) == 1 assert records[0]["id"] == 1 assert records[0]["entity_type"] == "User" - mock_get.assert_called_once() + mock_guard.assert_called_once() + +def test_load_from_api_blocks_private_by_default(seed_manager): + with pytest.raises(ProcessingError) as excinfo: + seed_manager.load_from_api(api_url="http://127.0.0.1:8000/secret") + assert "blocked" in str(excinfo.value).lower() or "not allowed" in str(excinfo.value).lower() + +@patch("semantica.seed.seed_manager.request_with_ssrf_guard") +def test_load_from_api_allows_private_when_configured(mock_guard, seed_manager): + mock_response = MagicMock() + mock_response.json.return_value = {"results": [{"id": 1, "name": "Alice"}]} + mock_guard.return_value = mock_response + + manager = SeedDataManager(config={"allow_private_ips": True}) + records = manager.load_from_api( + api_url="http://127.0.0.1:8000", + endpoint="users", + entity_type="User" + ) + + assert len(records) == 1 + mock_guard.assert_called_once() + # The opt-in flag must reach the guard + call_kwargs = mock_guard.call_args[1] + assert call_kwargs["allow_private_ips"] is True def test_load_source(seed_manager, temp_data_dir): json_file = temp_data_dir / "source.json" From 5b027364129bfc195693c6975fcef27c21c6d442 Mon Sep 17 00:00:00 2001 From: Yunare Maia Date: Thu, 13 Aug 2026 00:50:34 +0000 Subject: [PATCH 2/3] fix(ssrf): strip sensitive headers on cross-host redirects (Qodo finding) request_with_ssrf_guard reused the caller's headers on every redirect hop, so an Authorization bearer token from load_from_api could leak to a different redirect target host. Now strips Authorization and Proxy-Authorization when the redirect origin (netloc) changes, while keeping them for same-host hops (matching requests semantics). 2 new tests: cross-host redirect drops the credential; same-host keeps it. 37/37 green in test_ssrf_protection.py. load_from_api docstring now also documents cloud-metadata blocking and per-hop redirect validation. Signed-off-by: Yunare Maia --- semantica/ingest/ssrf.py | 12 ++++++ semantica/seed/seed_manager.py | 8 ++-- tests/ingest/test_ssrf_protection.py | 61 ++++++++++++++++++++++++++++ 3 files changed, 78 insertions(+), 3 deletions(-) diff --git a/semantica/ingest/ssrf.py b/semantica/ingest/ssrf.py index ade93a461..799e221b4 100644 --- a/semantica/ingest/ssrf.py +++ b/semantica/ingest/ssrf.py @@ -276,6 +276,18 @@ def request_with_ssrf_guard( 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 to a different origin (host) on + # redirects: reuse the caller's headers only while the origin is + # unchanged, mirroring requests' cross-host credential stripping. + current_origin = urlparse(current_url).netloc + next_origin = urlparse(next_url).netloc + if current_origin != next_origin: + kwargs = dict(kwargs) + headers = dict(kwargs.get("headers") or {}) + for sensitive in ("Authorization", "Proxy-Authorization"): + headers.pop(sensitive, None) + kwargs["headers"] = headers + # Match requests' historical method rewriting for 301/302/303. if ( response.status_code in _STRIP_BODY_ON_REDIRECT diff --git a/semantica/seed/seed_manager.py b/semantica/seed/seed_manager.py index c41f0ff6a..5051697d5 100644 --- a/semantica/seed/seed_manager.py +++ b/semantica/seed/seed_manager.py @@ -455,9 +455,11 @@ def load_from_api( entity_type, relationship_type, and source metadata if provided. SSRF protection is enabled by default: URLs resolving to private, - loopback, or link-local addresses are rejected. For trusted internal - deployments, pass ``allow_private_ips=True`` in the manager config to - opt in (documented for internal use only). + loopback, link-local (including cloud metadata endpoints such as + 169.254.169.254), or other blocked addresses are rejected, and every + redirect hop is re-validated before being followed. For trusted + internal deployments, pass ``allow_private_ips=True`` in the manager + config to opt in (documented for internal use only). Args: api_url: Base API URL diff --git a/tests/ingest/test_ssrf_protection.py b/tests/ingest/test_ssrf_protection.py index 2da845308..79779612e 100644 --- a/tests/ingest/test_ssrf_protection.py +++ b/tests/ingest/test_ssrf_protection.py @@ -182,6 +182,67 @@ def test_blocks_redirect_to_metadata_ip(self): session=session, ) + def test_strips_authorization_on_cross_host_redirect(self): + """Sensitive headers must not leak to a different redirect host.""" + redirect = MagicMock() + redirect.status_code = 302 + redirect.headers = {"Location": "https://other-host.example/final"} + redirect.close = MagicMock() + + final = MagicMock() + final.status_code = 200 + final.headers = {} + + session = MagicMock() + session.request.side_effect = [redirect, final] + + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(None, None, None, None, ("93.184.216.34", 0))], + ): + request_with_ssrf_guard( + "GET", + "https://example.com/start", + session=session, + headers={"Authorization": "Bearer secret-token"}, + ) + + assert session.request.call_count == 2 + second_call_headers = session.request.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second_call_headers + # The first hop still had the credential + first_call_headers = session.request.call_args_list[0].kwargs.get("headers", {}) + assert first_call_headers.get("Authorization") == "Bearer secret-token" + + def test_keeps_authorization_on_same_host_redirect(self): + """Same-host redirects keep the credential (requests semantics).""" + redirect = MagicMock() + redirect.status_code = 302 + redirect.headers = {"Location": "https://example.com/final"} + redirect.close = MagicMock() + + final = MagicMock() + final.status_code = 200 + final.headers = {} + + session = MagicMock() + session.request.side_effect = [redirect, final] + + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(None, None, None, None, ("93.184.216.34", 0))], + ): + request_with_ssrf_guard( + "GET", + "https://example.com/start", + session=session, + headers={"Authorization": "Bearer secret-token"}, + ) + + assert session.request.call_count == 2 + second_call_headers = session.request.call_args_list[1].kwargs.get("headers", {}) + assert second_call_headers.get("Authorization") == "Bearer secret-token" + def test_follows_safe_redirect(self): redirect = MagicMock() redirect.status_code = 302 From 524a204abad497968dd8b6d365bc0e109a36df98 Mon Sep 17 00:00:00 2001 From: Yunare Maia Date: Thu, 13 Aug 2026 12:21:52 +0000 Subject: [PATCH 3/3] fix(ssrf): strip credentials on https->http downgrade redirects (review feedback) _should_strip_auth now mirrors requests' should_strip_auth semantics: strip on hostname change, port change, or scheme downgrade; keep the credential only for the safe http->https upgrade on default ports. Previously only netloc was compared, so an https->http redirect on the same host replayed the Authorization header in cleartext. --- semantica/ingest/ssrf.py | 50 ++++++++++++++++++++--- tests/ingest/test_ssrf_protection.py | 61 ++++++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 6 deletions(-) diff --git a/semantica/ingest/ssrf.py b/semantica/ingest/ssrf.py index 799e221b4..083fbcca7 100644 --- a/semantica/ingest/ssrf.py +++ b/semantica/ingest/ssrf.py @@ -33,6 +33,46 @@ _REDIRECT_STATUS_CODES = frozenset({301, 302, 303, 307, 308}) _STRIP_BODY_ON_REDIRECT = frozenset({301, 302, 303}) +# Standard port per scheme (mirrors requests' DEFAULT_PORTS). +_DEFAULT_PORTS = {"http": 80, "https": 443} + + +def _should_strip_auth(old_url: str, new_url: str) -> bool: + """Decide whether credentials must not follow a redirect. + + Mirrors ``requests.utils.should_strip_auth``: credentials are stripped + when the hostname changes, when the port changes (outside default + ports), or on an https -> http downgrade on the same host. The single + exception is an http -> https upgrade on default ports, which requests + treats as safe to keep the credential for. + """ + old_parsed = urlparse(old_url) + new_parsed = urlparse(new_url) + + if old_parsed.hostname != new_parsed.hostname: + return True + + # Special case: allow http -> https redirect on standard ports. + if ( + old_parsed.scheme == "http" + and old_parsed.port in (80, None) + and new_parsed.scheme == "https" + and new_parsed.port in (443, None) + ): + return False + + changed_port = old_parsed.port != new_parsed.port + changed_scheme = old_parsed.scheme != new_parsed.scheme + default_port = (_DEFAULT_PORTS.get(old_parsed.scheme), None) + if ( + not changed_scheme + and old_parsed.port in default_port + and new_parsed.port in default_port + ): + return False + + return changed_port or changed_scheme + _dns_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None _dns_executor_lock = threading.Lock() @@ -276,12 +316,10 @@ def request_with_ssrf_guard( 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 to a different origin (host) on - # redirects: reuse the caller's headers only while the origin is - # unchanged, mirroring requests' cross-host credential stripping. - current_origin = urlparse(current_url).netloc - next_origin = urlparse(next_url).netloc - if current_origin != next_origin: + # 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"): diff --git a/tests/ingest/test_ssrf_protection.py b/tests/ingest/test_ssrf_protection.py index 79779612e..d7d3ab9bf 100644 --- a/tests/ingest/test_ssrf_protection.py +++ b/tests/ingest/test_ssrf_protection.py @@ -243,6 +243,67 @@ def test_keeps_authorization_on_same_host_redirect(self): second_call_headers = session.request.call_args_list[1].kwargs.get("headers", {}) assert second_call_headers.get("Authorization") == "Bearer secret-token" + def test_strips_authorization_on_scheme_downgrade(self): + """Credentials must not follow an https -> http downgrade on the same host.""" + redirect = MagicMock() + redirect.status_code = 302 + redirect.headers = {"Location": "http://example.com/final"} + redirect.close = MagicMock() + + final = MagicMock() + final.status_code = 200 + final.headers = {} + + session = MagicMock() + session.request.side_effect = [redirect, final] + + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(None, None, None, None, ("93.184.216.34", 0))], + ): + request_with_ssrf_guard( + "GET", + "https://example.com/start", + session=session, + headers={"Authorization": "Bearer secret-token"}, + ) + + assert session.request.call_count == 2 + second_call_headers = session.request.call_args_list[1].kwargs.get("headers", {}) + assert "Authorization" not in second_call_headers + # The first hop still had the credential + first_call_headers = session.request.call_args_list[0].kwargs.get("headers", {}) + assert first_call_headers.get("Authorization") == "Bearer secret-token" + + def test_keeps_authorization_on_scheme_upgrade(self): + """Credentials survive an http -> https upgrade on default ports (requests semantics).""" + redirect = MagicMock() + redirect.status_code = 302 + redirect.headers = {"Location": "https://example.com/final"} + redirect.close = MagicMock() + + final = MagicMock() + final.status_code = 200 + final.headers = {} + + session = MagicMock() + session.request.side_effect = [redirect, final] + + with patch( + "semantica.ingest.ssrf.socket.getaddrinfo", + return_value=[(None, None, None, None, ("93.184.216.34", 0))], + ): + request_with_ssrf_guard( + "GET", + "http://example.com/start", + session=session, + headers={"Authorization": "Bearer secret-token"}, + ) + + assert session.request.call_count == 2 + second_call_headers = session.request.call_args_list[1].kwargs.get("headers", {}) + assert second_call_headers.get("Authorization") == "Bearer secret-token" + def test_follows_safe_redirect(self): redirect = MagicMock() redirect.status_code = 302