fix: Cookie domain matching (#488)

Replace substring match by matching according to [RFC 6265,
§5.1.3](https://www.rfc-editor.org/rfc/rfc6265#section-5.1.3).
Add test.

# Issues:
Closes: #473

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: Pijukatel

impit-python/src/cookies.rs

@@ -123,7 +123,7 @@ impl CookieStore for PythonCookieJar {
                         .and_then(|attr| attr.extract::<bool>())
                         .unwrap_or_default();
 
-                    if !domain.is_empty() && !url.host_str().unwrap_or_default().contains(&domain) {
+                    if !domain_matches(url.host_str().unwrap_or_default(), &domain) {
                         return None;
                     }
                     if !url.path().starts_with(&path) {
@@ -175,6 +175,39 @@ impl CookieStore for PythonCookieJar {
     }
 }
 
+/// Checks whether a request `host` may receive a cookie scoped to `cookie_domain`,
+/// following the domain matching rules of
+/// [RFC 6265, §5.1.3](https://www.rfc-editor.org/rfc/rfc6265#section-5.1.3).
+///
+/// Leading dot on the cookie domain (e.g. `.example.com`) is ignored, as
+/// permitted by [RFC 6265, §4.1.2.3](https://www.rfc-editor.org/rfc/rfc6265#section-4.1.2.3).
+///
+/// An empty cookie domain imposes no host restriction and therefore matches any host.
+fn domain_matches(host: &str, cookie_domain: &str) -> bool {
+    let cookie_domain = cookie_domain.strip_prefix('.').unwrap_or(cookie_domain);
+
+    if cookie_domain.is_empty() {
+        return true;
+    }
+
+    // Host names are case-insensitive; normalise both sides instead of trusting the
+    // caller to pass a lowercased host.
+    let host = host.to_ascii_lowercase();
+    let cookie_domain = cookie_domain.to_ascii_lowercase();
+
+    // RFC 6265 §5.1.3: an IP-address host only matches an identical cookie domain.
+    if host.parse::<std::net::IpAddr>().is_ok() {
+        return host == cookie_domain;
+    }
+
+    // Exact match, or subdomain match where the cookie domain is a suffix of the host
+    // on a `.` label boundary (e.g. host `www.example.com`, cookie domain `example.com`).
+    host == cookie_domain
+        || host
+            .strip_suffix(cookie_domain.as_str())
+            .is_some_and(|prefix| prefix.ends_with('.'))
+}
+
 impl PythonCookieJar {
     pub fn new(py: Python<'_>, cookie_jar: Py<PyAny>) -> Self {
         let httpmodule = PyModule::import(py, "http.cookiejar").unwrap();

impit-python/test/no_client_test.py

@@ -2,6 +2,7 @@
 import socket
 import threading
 import time
+import urllib.parse
 from http.cookiejar import CookieJar
 
 import pytest
@@ -30,6 +31,31 @@ def thread_server(port_holder: list[int]) -> None:
     server.close()
 
 
+def cookie_echo_server(port_holder: list[int], captured: dict[str, str]) -> None:
+    """Serve a single HTTP/1.1 request on 127.0.0.1 and record its `Cookie` header."""
+    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    server.bind(('127.0.0.1', 0))
+    port_holder[0] = server.getsockname()[1]
+    server.listen(1)
+
+    conn, _ = server.accept()
+    request = b''
+    while b'\r\n\r\n' not in request:
+        chunk = conn.recv(4096)
+        if not chunk:
+            break
+        request += chunk
+
+    for line in request.decode('utf-8', errors='replace').split('\r\n'):
+        if line.lower().startswith('cookie:'):
+            captured['cookie'] = line.split(':', 1)[1].strip()
+
+    conn.send(b'HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok')
+    conn.close()
+    server.close()
+
+
 class TestBasicRequests:
     @pytest.mark.parametrize(
         ('protocol'),
@@ -174,6 +200,50 @@ def test_cookies_param_works(self) -> None:
         assert cookies.get('preset-cookie') == '123'
         assert cookies.get('set-by-server') == '321'
 
+    def test_cookie_domain_matching_does_not_leak_to_lookalike_hosts(self) -> None:
+        url = get_httpbin_url('/cookies')
+        host = urllib.parse.urlparse(url).hostname or ''
+        # The host must have at least two labels for this test to be meaningful.
+        assert '.' in host
+
+        first_label, _, parent_domain = host.partition('.')
+
+        cookies = Cookies()
+        # These domain-match the host and must be sent.
+        cookies.set('exact_match', 'yes', domain=host)  # host == domain
+        cookies.set('subdomain_ok', 'yes', domain=parent_domain)  # host is a subdomain of the cookie domain
+
+        # These only match the host as a substring (not on a label boundary) and must NOT
+        # be sent. With the old `.contains()` check both of these would leak.
+        cookies.set('substring_leak', 'no', domain=host[1:])  # e.g. host `httpbin.org` vs `ttpbin.org`
+        cookies.set('prefix_leak', 'no', domain=first_label)  # e.g. host `httpbin.org` vs `httpbin`
+
+        response = impit.get(url, cookies=cookies).json()
+
+        assert response['cookies'] == {'exact_match': 'yes', 'subdomain_ok': 'yes'}
+
+    def test_cookie_domain_matching_for_ip_hosts(self) -> None:
+        # RFC 6265 §5.1.3: subdomain/suffix matching applies to host names only, so an
+        # IP-address host must match the cookie domain exactly. Without that rule a
+        # cookie scoped to `0.0.1` would leak to host `127.0.0.1` (it is a label-boundary
+        # suffix of the IP). A local server is used so the request host is a real IP.
+        port_holder = [0]
+        captured: dict[str, str] = {}
+        thread = threading.Thread(target=cookie_echo_server, args=(port_holder, captured))
+        thread.start()
+        time.sleep(0.1)
+
+        cookies = Cookies()
+        cookies.set('exact_ip', 'yes', domain='127.0.0.1')  # exact IP match -> sent
+        cookies.set('ip_suffix_leak', 'no', domain='0.0.1')  # IP host matches exactly -> NOT sent
+
+        impit.get(f'http://127.0.0.1:{port_holder[0]}/', cookies=cookies, timeout=5)
+        thread.join()
+
+        sent = captured.get('cookie', '')
+        sent_names = {part.split('=')[0].strip() for part in sent.split(';') if '=' in part}
+        assert sent_names == {'exact_ip'}
+
     @pytest.mark.skip(reason='Flaky under the CI environment')
     def test_http3_works(self) -> None:
         response = impit.get('https://curl.se', force_http3=True)