From b535839003996ffefdad94f13e2752fd99cd1ca9 Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Wed, 10 Jun 2026 12:41:44 +0530 Subject: [PATCH] fix(ingest): harden public API auth validation --- semantica/ingest/public_api_ingestor.py | 16 +++-- tests/ingest/test_public_api_ingestor.py | 87 ++++++++++++++++++++++++ 2 files changed, 99 insertions(+), 4 deletions(-) diff --git a/semantica/ingest/public_api_ingestor.py b/semantica/ingest/public_api_ingestor.py index d1b16429..0b467053 100644 --- a/semantica/ingest/public_api_ingestor.py +++ b/semantica/ingest/public_api_ingestor.py @@ -25,7 +25,7 @@ import time from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional, Tuple -from urllib.parse import urlparse +from urllib.parse import parse_qs, urlparse import requests from lxml import etree as lxml_etree @@ -342,7 +342,7 @@ class PublicAPIIngestor(RESTIngestor): """ self._validate_endpoint(endpoint) auth_indicators = self._auth_indicators( - headers=headers, params=params, options=options + headers=headers, params=params, options=options, endpoint=endpoint ) if auth_indicators: return PublicAPIDetection( @@ -429,7 +429,7 @@ class PublicAPIIngestor(RESTIngestor): APIData: Normalized public API response and metadata """ self._validate_endpoint(endpoint) - self._validate_no_auth_request(headers=headers, params=params, options=options) + self._validate_no_auth_request(headers=headers, params=params, options=options, endpoint=endpoint) tracking_id = self.progress_tracker.start_tracking( file=endpoint, @@ -606,6 +606,7 @@ class PublicAPIIngestor(RESTIngestor): headers: Optional[Dict[str, str]] = None, params: Optional[Dict[str, Any]] = None, options: Optional[Dict[str, Any]] = None, + endpoint: Optional[str] = None, ) -> List[str]: indicators: List[str] = [] merged_headers = self._merged_headers(headers) @@ -620,6 +621,11 @@ class PublicAPIIngestor(RESTIngestor): if options and options.get("auth") is not None: indicators.append("request:auth") + if endpoint: + for param_name in parse_qs(urlparse(endpoint).query): + if param_name.lower() in AUTH_PARAM_NAMES: + indicators.append(f"url_param:{param_name}") + return indicators def _validate_no_auth_request( @@ -627,6 +633,7 @@ class PublicAPIIngestor(RESTIngestor): headers: Optional[Dict[str, str]] = None, params: Optional[Dict[str, Any]] = None, options: Optional[Dict[str, Any]] = None, + endpoint: Optional[str] = None, ) -> None: if not self.validate_no_auth: return @@ -635,6 +642,7 @@ class PublicAPIIngestor(RESTIngestor): headers=headers, params=params, options=options, + endpoint=endpoint, ) if auth_indicators: indicators = ", ".join(auth_indicators) @@ -725,7 +733,7 @@ class PublicAPIIngestor(RESTIngestor): return "xml" if text.startswith(("{", "[")): return "json" - if text.startswith("<"): + if text.startswith("<") and "text/html" not in content_type: return "xml" return "text" diff --git a/tests/ingest/test_public_api_ingestor.py b/tests/ingest/test_public_api_ingestor.py index 60e976b5..61119427 100644 --- a/tests/ingest/test_public_api_ingestor.py +++ b/tests/ingest/test_public_api_ingestor.py @@ -275,3 +275,90 @@ def test_public_api_convenience_methods_and_registry_dispatch() -> None: assert unified["data"].data == payload assert "endpoint" in methods["public_api"] assert "detect" in methods["public_api"] + + +@pytest.mark.parametrize("url", [ + "https://api.example.com/data?api_key=SECRET", + "https://api.example.com/data?token=abc123", + "https://api.example.com/data?access_token=xyz&other=1", +]) +def test_public_api_ingestor_rejects_auth_credential_in_url(url: str) -> None: + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + + with pytest.raises(ValidationError, match="no-auth endpoints"): + PublicAPIIngestor(rate_limit_delay=0).ingest_public_api(url) + + mock_session.request.assert_not_called() + + +def test_public_api_detection_returns_not_public_for_url_with_auth_param() -> None: + # detect_public_api does not raise; it returns a detection result with is_public=False + with patch("requests.Session") as mock_session_class: + mock_session = mock_session_class.return_value + mock_session.headers = {} + + detection = PublicAPIIngestor(rate_limit_delay=0).detect_public_api( + "https://api.example.com/data?api_key=SECRET" + ) + + assert detection.is_public is False + assert detection.requires_auth is True + assert "url_param:api_key" in detection.metadata.get("auth_indicators", []) + mock_session.request.assert_not_called() + + +def test_require_public_false_allows_successful_response() -> None: + payload = [{"id": 1}] + + 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=payload, + text='[{"id": 1}]', + headers={"Content-Type": "application/json"}, + ) + + result = PublicAPIIngestor(rate_limit_delay=0).ingest_public_api( + "https://example.com/api/data", + require_public=False, + ) + + assert result.data == payload + assert result.metadata["requires_auth"] is False + + +def test_require_public_false_with_401_still_raises_via_raise_for_status() -> 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(status_code=401) + + with pytest.raises(ProcessingError): + PublicAPIIngestor(rate_limit_delay=0).ingest_public_api( + "https://example.com/api/data", + require_public=False, + ) + + +def test_html_response_is_not_misclassified_as_xml() -> None: + html = "403 Forbidden" + + 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( + text=html, + headers={"Content-Type": "text/html; charset=utf-8"}, + ) + + result = PublicAPIIngestor( + rate_limit_delay=0, validate_no_auth=False + ).ingest_public_api( + "https://example.com/api/data", + require_public=False, + ) + + assert result.metadata["response_format"] == "text"