"""Tests for the S3 presigned-download path in ``services/makerworld.py``. MakerWorld hands back an AWS presigned URL for the 3MF, and we fetch that one with ``urllib.request`` rather than httpx — httpx re-encodes the query string and invalidates the S3 signature. That choice silently changed the trust store: urllib verifies against the OS CA store, httpx against the bundled ``certifi`` bundle. On Windows the two disagree and the download dies with ``CERTIFICATE_VERIFY_FAILED: unable to get local issuer certificate`` (#2562). These tests pin the fix (the opener carries a certifi-backed TLS context) and the two properties the fix must not break: the no-redirect SSRF guard, and the URL reaching the transport byte-for-byte. """ from __future__ import annotations import ssl from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch import certifi import pytest from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.x509.oid import NameOID from backend.app.services import makerworld as mw # A presigned URL in the shape Bambu Cloud actually mints: the signature is # computed over these exact query-string bytes, so any re-encoding breaks it. S3_URL = ( "https://s3.us-west-2.amazonaws.com/bbl-prod/models/benchy.3mf" "?X-Amz-Algorithm=AWS4-HMAC-SHA256" "&X-Amz-Credential=AKIA%2F20260714%2Fus-west-2%2Fs3%2Faws4_request" "&X-Amz-Date=20260714T070000Z&X-Amz-Expires=300" "&X-Amz-Signature=abc123&X-Amz-SignedHeaders=host" ) def _write_test_ca(path) -> str: """Write a throwaway self-signed CA to ``path`` and return its CN. Lets a test assert the opener's TLS context was loaded from *certifi's* bundle specifically, rather than from the OS store or any other source: we point ``certifi.where()`` at this file and then check the context trusts exactly this one cert. """ key = ec.generate_private_key(ec.SECP256R1()) common_name = "Bambuddy Test Root CA" subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)]) now = datetime.now(timezone.utc) cert = ( x509.CertificateBuilder() .subject_name(subject) .issuer_name(subject) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(now - timedelta(days=1)) .not_valid_after(now + timedelta(days=3650)) .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) .sign(key, hashes.SHA256()) ) path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) return common_name class _FakeResponse: """Stand-in for the ``http.client.HTTPResponse`` urllib hands back.""" def __init__(self, body: bytes, status: int = 200): self.status = status self._body = body self._offset = 0 def __enter__(self): return self def __exit__(self, *exc): return False def read(self, size: int) -> bytes: chunk = self._body[self._offset : self._offset + size] self._offset += len(chunk) return chunk class _OpenerCapture: """Captures the handlers ``build_opener`` was called with, and the Request the resulting opener was asked to open.""" def __init__(self, response: _FakeResponse | None = None, raises: BaseException | None = None): self.handlers: tuple = () self.request = None self._response = response or _FakeResponse(b"3MF") self._raises = raises def build_opener(self, *handlers): self.handlers = handlers opener = MagicMock() opener.open = self._open return opener def _open(self, request, timeout=None): self.request = request if self._raises is not None: raise self._raises return self._response def https_handler(self): for handler in self.handlers: if isinstance(handler, mw_https_handler_type()): return handler return None def mw_https_handler_type(): from urllib.request import HTTPSHandler return HTTPSHandler def _patched_opener(capture: _OpenerCapture): """``_download_s3_urllib`` imports ``build_opener`` from ``urllib.request`` at call time, so patching the module attribute is enough.""" return patch("urllib.request.build_opener", side_effect=capture.build_opener) class TestS3TrustStore: """The regression under test: urllib must not fall back to the OS CA store.""" @pytest.mark.asyncio async def test_opener_gets_an_https_handler(self): """Without an explicit HTTPSHandler, urllib builds its own from the OS trust store — which is exactly what fails on Windows (#2562).""" capture = _OpenerCapture() with _patched_opener(capture): await mw._download_s3_urllib(S3_URL, "benchy.3mf") handler = capture.https_handler() assert handler is not None, "opener was built without an HTTPSHandler — falls back to the OS trust store" assert isinstance(handler._context, ssl.SSLContext) @pytest.mark.asyncio async def test_context_is_loaded_from_certifi(self, tmp_path, monkeypatch): """Point certifi at a bundle holding one throwaway root; the opener's context must trust exactly that root and nothing else. Proves the CAs come from certifi rather than the system store.""" ca_pem = tmp_path / "test-cacert.pem" common_name = _write_test_ca(ca_pem) monkeypatch.setattr(mw.certifi, "where", lambda: str(ca_pem)) capture = _OpenerCapture() with _patched_opener(capture): await mw._download_s3_urllib(S3_URL, "benchy.3mf") loaded = capture.https_handler()._context.get_ca_certs() assert len(loaded) == 1, f"expected only the certifi bundle's cert, got {len(loaded)}" subject_values = [value for rdn in loaded[0]["subject"] for _, value in rdn] assert common_name in subject_values @pytest.mark.asyncio async def test_context_verifies_and_checks_hostname(self): """certifi swaps the CA source, not the verification policy — a context with verification off would 'fix' #2562 by disabling TLS security.""" capture = _OpenerCapture() with _patched_opener(capture): await mw._download_s3_urllib(S3_URL, "benchy.3mf") context = capture.https_handler()._context assert context.verify_mode == ssl.CERT_REQUIRED assert context.check_hostname is True def test_real_context_trusts_the_certifi_bundle(self): """Sanity-check the un-mocked helper against the shipped bundle: it must load a real-world number of roots, not an empty set.""" context = mw._s3_ssl_context() assert len(context.get_ca_certs()) == len(ssl.create_default_context(cafile=certifi.where()).get_ca_certs()) assert len(context.get_ca_certs()) > 50 class TestS3DownloadUnchanged: """Properties the TLS fix must not regress.""" @pytest.mark.asyncio async def test_redirects_are_still_refused(self): """The host allowlist is only enforced on the initial URL, so following a 302 off S3 would bypass it. The no-redirect handler must survive.""" capture = _OpenerCapture() with _patched_opener(capture): await mw._download_s3_urllib(S3_URL, "benchy.3mf") from urllib.request import HTTPRedirectHandler # build_opener takes handler classes *or* instances; the redirect # blocker is passed as a class, so normalise before probing it. blockers = [] for handler in capture.handlers: instance = handler() if isinstance(handler, type) else handler if isinstance(instance, HTTPRedirectHandler): if instance.redirect_request(None, None, None, None, None) is None: blockers.append(instance) assert blockers, "no redirect-blocking handler passed to build_opener" @pytest.mark.asyncio async def test_url_reaches_the_transport_verbatim(self): """The whole reason this path uses urllib: S3 signs the exact query-string bytes. Any normalisation yields SignatureDoesNotMatch.""" capture = _OpenerCapture() with _patched_opener(capture): await mw._download_s3_urllib(S3_URL, "benchy.3mf") assert capture.request.full_url == S3_URL @pytest.mark.asyncio async def test_returns_body_and_filename(self): capture = _OpenerCapture(response=_FakeResponse(b"PK\x03\x04payload")) with _patched_opener(capture): data, filename = await mw._download_s3_urllib(S3_URL, "benchy.3mf") assert data == b"PK\x03\x04payload" assert filename == "benchy.3mf" @pytest.mark.asyncio async def test_non_200_raises_unavailable(self): capture = _OpenerCapture(response=_FakeResponse(b"", status=403)) with _patched_opener(capture), pytest.raises(mw.MakerWorldUnavailableError, match="HTTP 403"): await mw._download_s3_urllib(S3_URL, "benchy.3mf") @pytest.mark.asyncio async def test_size_cap_enforced(self, monkeypatch): monkeypatch.setattr(mw, "_MAX_3MF_BYTES", 1024) capture = _OpenerCapture(response=_FakeResponse(b"x" * 4096)) with _patched_opener(capture), pytest.raises(mw.MakerWorldUnavailableError, match="exceeds"): await mw._download_s3_urllib(S3_URL, "benchy.3mf") @pytest.mark.asyncio async def test_tls_failure_still_surfaces_as_s3_download_failed(self): """If verification fails for a genuine reason (expired cert, MITM proxy), the user must still get the actionable wrapped error — the fix removes the spurious failures, it doesn't swallow the real ones.""" verify_error = ssl.SSLCertVerificationError("certificate verify failed: unable to get local issuer certificate") capture = _OpenerCapture(raises=verify_error) with _patched_opener(capture), pytest.raises(mw.MakerWorldUnavailableError, match="S3 download failed"): await mw._download_s3_urllib(S3_URL, "benchy.3mf")