fix(ingest): harden public API auth validation

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"
 

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 = "<!DOCTYPE html><html><body>403 Forbidden</body></html>"
+
+    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"