tests/test_caldav_test_connection_ssl.py
Rama dev · réplica de lectura del repositorio original.
"""Regression: CalDAV test_connection must trust the operator's CA bundle.
The pre-flight used httpx with trust_env=False, which ignored
SSL_CERT_FILE/REQUESTS_CA_BUNDLE. Self-signed CalDAV servers that the
real sync accepts (via caldav lib -> requests -> honors bundle) were
rejected by the test with CERTIFICATE_VERIFY_FAILED.
These tests exercise the *route handler* directly (via ASGI TestClient)
and capture the verify= kwarg passed to httpx.AsyncClient, ensuring the
route code — not a test-side duplicate — builds the SSL context correctly.
"""
import os
import ssl
import sys
from unittest.mock import MagicMock, patch
import httpx
import pytest
# No module-level sys.modules stubbing here: conftest pre-imports the real
# sqlalchemy/core.database, and stubbing extras (e.g. caldav) at collection
# time leaks MagicMocks into later tests in the same process — it made
# test_caldav_redirect_hardening's real DAVClient a mock that never sent
# the PROPFIND. The route's lazy imports are patched per-request instead.
def _fake_response(status_code=207, headers=None):
resp = MagicMock()
resp.status_code = status_code
resp.headers = headers or {}
return resp
@pytest.fixture()
def client():
from fastapi import FastAPI
from fastapi.testclient import TestClient
from routes.calendar_routes import setup_calendar_routes
with patch("routes.calendar_routes._require_user", return_value="test-owner"):
router = setup_calendar_routes()
app = FastAPI()
app.include_router(router)
return TestClient(app)
def _make_fake_async_client(captured):
"""Return a fake httpx.AsyncClient class that captures constructor kwargs."""
class FakeAsyncClient:
def __init__(self, **kwargs):
captured.update(kwargs)
async def __aenter__(self):
return self
async def __aexit__(self, *a):
pass
async def request(self, *a, **kw):
return _fake_response(207)
return FakeAsyncClient
def _post_test(client, captured, env=None):
"""POST /api/calendar/test with credentials in body so no DB lookup needed.
Patches httpx.AsyncClient at the real module level so the route's
``import httpx; httpx.AsyncClient(...)`` picks up the fake class.
Also stubs validate_caldav_url (lazy-imported from src.caldav_sync).
"""
fake_cls = _make_fake_async_client(captured)
# Stub the caldav_sync module so the lazy `from src.caldav_sync import validate_caldav_url`
# inside the route body resolves to a pass-through.
caldav_sync_stub = MagicMock()
caldav_sync_stub.validate_caldav_url = lambda u: u
ctx_managers = [
patch.object(httpx, "AsyncClient", fake_cls),
patch.dict(sys.modules, {"src.caldav_sync": caldav_sync_stub}),
patch("routes.calendar_routes._require_user", return_value="test-owner"),
]
if env is not None:
ctx_managers.append(patch.dict(os.environ, env))
# Enter all context managers
for cm in ctx_managers:
cm.__enter__()
try:
return client.post(
"/api/calendar/test",
json={"url": "https://cal.example.com", "username": "u", "password": "p"},
)
finally:
for cm in reversed(ctx_managers):
cm.__exit__(None, None, None)
# ---------------------------------------------------------------------------
# Route-level tests
# ---------------------------------------------------------------------------
def test_route_passes_ssl_context_with_correct_flags(client):
"""The route must pass an ssl.SSLContext to httpx.AsyncClient(verify=...)
with trust_env=False, follow_redirects=False, and VERIFY_X509_STRICT cleared."""
captured = {}
resp = _post_test(client, captured)
assert resp.status_code == 200
assert isinstance(captured.get("verify"), ssl.SSLContext), (
f"verify= should be an ssl.SSLContext, got {type(captured.get('verify'))}"
)
assert captured.get("trust_env") is False
assert captured.get("follow_redirects") is False
ctx = captured["verify"]
assert not (ctx.verify_flags & ssl.VERIFY_X509_STRICT), (
"VERIFY_X509_STRICT must be cleared for self-signed CA compat"
)
def test_route_ssl_cert_file_takes_precedence(client, tmp_path):
"""SSL_CERT_FILE is the exact bundle loaded when both variables are set."""
bundle_a = tmp_path / "ssl-cert-file.pem"
bundle_b = tmp_path / "requests-ca-bundle.pem"
bundle_a.write_text("ssl-cert-file", encoding="utf-8")
bundle_b.write_text("requests-ca-bundle", encoding="utf-8")
loaded = []
class FakeSSLContext:
def __init__(self):
self.verify_flags = ssl.VERIFY_X509_STRICT
def load_verify_locations(self, cafile=None, capath=None, cadata=None):
loaded.append(
{
"cafile": cafile,
"capath": capath,
"cadata": cadata,
}
)
ssl_context = FakeSSLContext()
captured = {}
env = {
"SSL_CERT_FILE": str(bundle_a),
"REQUESTS_CA_BUNDLE": str(bundle_b),
}
with patch.object(
ssl,
"create_default_context",
return_value=ssl_context,
):
resp = _post_test(client, captured, env=env)
assert resp.status_code == 200
assert resp.json() == {"ok": True}
assert loaded == [
{
"cafile": str(bundle_a),
"capath": None,
"cadata": None,
}
]
assert captured.get("verify") is ssl_context
assert captured.get("trust_env") is False
assert captured.get("follow_redirects") is False
assert not (
ssl_context.verify_flags & ssl.VERIFY_X509_STRICT
)
def test_route_missing_bundle_does_not_crash(client):
"""A nonexistent CA bundle path must not crash -- fall back to system CAs."""
captured = {}
resp = _post_test(client, captured, env={"SSL_CERT_FILE": "/nonexistent/ca-bundle.pem"})
assert resp.status_code == 200
ctx = captured["verify"]
assert isinstance(ctx, ssl.SSLContext)
assert not (ctx.verify_flags & ssl.VERIFY_X509_STRICT)
def test_route_empty_env_vars_use_system_defaults(client):
"""Empty SSL_CERT_FILE and REQUESTS_CA_BUNDLE should not crash."""
captured = {}
resp = _post_test(client, captured, env={"SSL_CERT_FILE": "", "REQUESTS_CA_BUNDLE": ""})
assert resp.status_code == 200
ctx = captured["verify"]
assert isinstance(ctx, ssl.SSLContext)
assert not (ctx.verify_flags & ssl.VERIFY_X509_STRICT)