diff --git a/tests/test_caldav_redirect_hardening.py b/tests/test_caldav_redirect_hardening.py index cd82974aa..6a0f76345 100644 --- a/tests/test_caldav_redirect_hardening.py +++ b/tests/test_caldav_redirect_hardening.py @@ -13,6 +13,7 @@ import socketserver import threading import pytest +import requests from src import caldav_sync, caldav_writeback @@ -26,12 +27,13 @@ def test_build_dav_client_disables_redirects(): def test_dav_client_does_not_follow_redirect_to_internal_host(): """End-to-end through the real DAVClient: a 302 toward an internal host - must NOT be followed. Without the fix the sink is contacted (SSRF); with it - the redirect is refused and the sink is never reached.""" + must NOT be followed. Exercise the session limit separately because some + caldav versions independently refuse cross-origin PROPFIND redirects.""" pytest.importorskip("caldav") sink_hits: list[str] = [] public_methods: list[str] = [] + same_origin_hits: list[str] = [] class _Internal(http.server.BaseHTTPRequestHandler): # Stand-in for an internal service the attacker redirects toward. @@ -48,9 +50,15 @@ def test_dav_client_does_not_follow_redirect_to_internal_host(): class _Public(http.server.BaseHTTPRequestHandler): # The "validated" public CalDAV server that redirects everything inward. def do_GET(self): # noqa: N802 + if self.path == "/landing": + same_origin_hits.append(self.path) + self.send_response(200) + self.end_headers() + return public_methods.append(self.command) self.send_response(302) - self.send_header("Location", f"http://127.0.0.1:{internal_port}/leak") + location = "/landing" if self.path == "/same" else f"http://127.0.0.1:{internal_port}/leak" + self.send_header("Location", location) self.end_headers() do_PROPFIND = do_GET @@ -62,8 +70,10 @@ def test_dav_client_does_not_follow_redirect_to_internal_host(): internal_port = internal.server_address[1] public = socketserver.TCPServer(("127.0.0.1", 0), _Public) public_port = public.server_address[1] - threading.Thread(target=internal.serve_forever, daemon=True).start() - threading.Thread(target=public.serve_forever, daemon=True).start() + internal_thread = threading.Thread(target=internal.serve_forever, daemon=True) + public_thread = threading.Thread(target=public.serve_forever, daemon=True) + internal_thread.start() + public_thread.start() try: public_url = f"http://127.0.0.1:{public_port}/dav" client = caldav_sync._build_dav_client(public_url, "u", "p") @@ -82,17 +92,35 @@ def test_dav_client_does_not_follow_redirect_to_internal_host(): assert public_methods in (["PROPFIND"], ["PROPFIND", "GET"]), public_methods assert sink_hits == [], "redirect toward an internal host must not be followed" - # Positive control: demonstrate that the sink would see the redirect - # if the session's redirect protection were removed. + # GET redirects are handled by the HTTP session even in caldav 3.7. + # This makes the cross-port SSRF assertion sensitive to our own limit. + with pytest.raises(Exception, match="Exceeded 0 redirects"): + client.request(public_url, "GET", "") + assert sink_hits == [], "GET redirect toward the internal sink must be blocked" + + # requests can reach the cross-port sink when redirects are enabled, + # even if DAVClient independently rejects cross-origin PROPFIND redirects. + with requests.Session() as session: + response = session.get(public_url, timeout=5) + assert response.status_code == 207 + assert sink_hits == ["/leak"], "the cross-port redirect must reach the sink" + + # GET follows redirects at the HTTP session layer across these caldav + # versions. Check the application's limit on a same-origin redirect too. + same_url = f"http://127.0.0.1:{public_port}/same" + with pytest.raises(Exception, match="Exceeded 0 redirects"): + client.request(same_url, "GET", "") + assert same_origin_hits == [] client.session.max_redirects = 30 - try: - client.request(public_url, "PROPFIND", "") - except Exception: - pass # The sink's response is intentionally not a valid DAV body. - assert sink_hits == ["/leak"], "the test must detect a followed redirect" + client.request(same_url, "GET", "") + assert same_origin_hits == ["/landing"] finally: internal.shutdown() public.shutdown() + internal_thread.join() + public_thread.join() + internal.server_close() + public.server_close() def test_sync_and_writeback_construct_clients_through_the_helper():