validate host value, sanetize user agent string

Author: mtmail

opencage/geocoder.py

@@ -18,6 +18,36 @@
 DEFAULT_DOMAIN = 'api.opencagedata.com'
 
 
+def _validate_domain(domain):
+    """Validate that the API domain is an allowed hostname.
+
+    Only subdomains of opencagedata.com, localhost, and 0.0.0.0 are
+    permitted. An optional port suffix (e.g. ``localhost:8080``) is allowed.
+
+    Args:
+        domain: Hostname string, optionally with a port.
+
+    Returns:
+        The validated domain string.
+
+    Raises:
+        ValueError: If the domain is not in the allow-list.
+    """
+    # Strip optional port
+    host = domain.rsplit(':', 1)[0] if ':' in domain else domain
+
+    if host in ('localhost', '0.0.0.0'):
+        return domain
+
+    if host.endswith('.opencagedata.com'):
+        return domain
+
+    raise ValueError(
+        f"Invalid API domain '{domain}'. "
+        f"Must be a subdomain of opencagedata.com, localhost, or 0.0.0.0."
+    )
+
+
 def backoff_max_time():
     """Return the maximum backoff time in seconds for retrying API requests.
 
@@ -146,6 +176,7 @@ def __init__(
 
         if protocol and protocol not in ('http', 'https'):
             protocol = 'https'
+        _validate_domain(domain)
         self.url = protocol + '://' + domain + '/geocode/v1/json'
 
         # https://docs.aiohttp.org/en/stable/client_advanced.html#ssl-control-for-tcp-sockets
@@ -353,7 +384,8 @@ def _opencage_headers(self, client):
 
         comment = ''
         if self.user_agent_comment:
-            comment = f" ({self.user_agent_comment})"
+            clean = self.user_agent_comment.replace('\r', '').replace('\n', '')
+            comment = f" ({clean})"
 
         return {
             'User-Agent': f"opencage-python/{__version__} Python/{py_version} {client}/{client_version}{comment}"

test/fixtures/cli/.DS_Store

@@ -11,8 +11,11 @@
 
 @pytest.mark.asyncio
 async def test_sslerror():
-    bad_domain = 'wrong.host.badssl.com'
-    async with OpenCageGeocode('6d0e711d72d74daeb2b0bfd2a5cdfdba', domain=bad_domain) as geocoder:
+    # Use a bad certificate (from badssl.com) against the real OpenCage domain
+    # to trigger an SSL error without needing a non-allowed domain
+    sslcontext = ssl.create_default_context(cafile='test/fixtures/badssl-com-chain.pem')
+
+    async with OpenCageGeocode('6d0e711d72d74daeb2b0bfd2a5cdfdba', sslcontext=sslcontext) as geocoder:
         with pytest.raises(SSLError) as excinfo:
             await geocoder.geocode_async("something")
         assert str(excinfo.value).startswith('SSL Certificate error')

test/test_error_ssl.py

@@ -21,5 +21,18 @@ def test_api_key_env_var():
 
 def test_custom_domain():
     """Test that custom domain can be set"""
-    geocoder = OpenCageGeocode('abcde', domain='example.com')
-    assert geocoder.url == 'https://example.com/geocode/v1/json'
+    geocoder = OpenCageGeocode('abcde', domain='api2.opencagedata.com')
+    assert geocoder.url == 'https://api2.opencagedata.com/geocode/v1/json'
+
+
+def test_custom_domain_localhost():
+    """Test that localhost domain can be set"""
+    geocoder = OpenCageGeocode('abcde', domain='localhost:8080')
+    assert geocoder.url == 'https://localhost:8080/geocode/v1/json'
+
+
+def test_custom_domain_invalid():
+    """Test that invalid domains are rejected"""
+    import pytest
+    with pytest.raises(ValueError, match="Invalid API domain"):
+        OpenCageGeocode('abcde', domain='www.example.com')

test/test_geocoder_args.py

@@ -33,3 +33,25 @@ def test_sync():
     user_agent = request.headers['User-Agent']
 
     assert user_agent_format.match(user_agent) is not None
+
+
+@responses.activate
+def test_user_agent_comment_crlf_stripped():
+    """Test that CR/LF characters are stripped from user_agent_comment to prevent header injection."""
+    geocoder_crlf = OpenCageGeocode('abcde', user_agent_comment="bad\r\nInjected-Header: value")
+
+    responses.add(
+        responses.GET,
+        geocoder_crlf.url,
+        body=Path('test/fixtures/uk_postcode.json').read_text(encoding="utf-8"),
+        status=200
+    )
+
+    geocoder_crlf.geocode("EC1M 5RF")
+
+    request = responses.calls[-1].request
+    user_agent = request.headers['User-Agent']
+
+    assert '\r' not in user_agent
+    assert '\n' not in user_agent
+    assert 'Injected-Header' in user_agent  # still present, just on the same line