fetch bundle asynchronously

ca/django_ca/acme/views.py

@@ -999,7 +999,8 @@ async def acme_request(self, slug: str) -> HttpResponse:  # type: ignore[overrid
         # self.prepared['cert'] = slug
         # self.prepared['csr'] = cert.csr
         # self.prepared['order'] = cert.order.slug
-        return HttpResponse(cert.cert.bundle_as_pem, content_type="application/pem-certificate-chain")
+        content = await cert.cert.aget_bundle_as_pem()
+        return HttpResponse(content, content_type="application/pem-certificate-chain")
 
 
 class AcmeAuthorizationView(AcmePostAsGetView):

ca/django_ca/models.py

@@ -327,6 +327,10 @@ def bundle_as_pem(self) -> str:
         #            means that an abstract "bundle" property here could not be correctly typed.
         return "".join(c.pub.pem for c in self.bundle)  # type:  ignore[attr-defined]
 
+    async def aget_bundle_as_pem(self) -> str:
+        bundle = await self.aget_bundle()  # type:  ignore[attr-defined]
+        return "".join(c.pub.pem for c in bundle)
+
     @property
     def jwk(self) -> Union[jose.jwk.JWKRSA, jose.jwk.JWKEC]:
         """Get a JOSE JWK public key for this certificate.
@@ -1108,6 +1112,18 @@ def bundle(self) -> list["CertificateAuthority"]:
             ca = ca.parent
         return bundle
 
+    async def aget_bundle(self) -> list["CertificateAuthority"]:
+        ca = self
+        bundle = [ca]
+        while ca.parent_id is not None:
+            if CertificateAuthority.parent.is_cached(ca):
+                ca = ca.parent  # type: ignore[assignment]  # checks above make sure it's not None
+            else:
+                ca = await CertificateAuthority.objects.select_related("parent").aget(pk=ca.parent_id)
+            bundle.append(ca)
+
+        return bundle
+
     @property
     def root(self) -> "CertificateAuthority":
         """Get the root CA for this CA."""
@@ -1172,6 +1188,16 @@ def bundle(self) -> list[X509CertMixin]:
         """The complete certificate bundle. This includes all CAs as well as the certificates itself."""
         return [typing.cast(X509CertMixin, self), *typing.cast(list[X509CertMixin], self.ca.bundle)]
 
+    async def aget_bundle(self) -> list[X509CertMixin]:
+        """The complete certificate bundle. This includes all CAs as well as the certificates itself."""
+        if Certificate.ca.is_cached(self):
+            ca = self.ca
+        else:
+            ca = await CertificateAuthority.objects.select_related("parent").aget(pk=self.ca_id)
+
+        ca_bundle = await ca.aget_bundle()
+        return [typing.cast(X509CertMixin, self), *typing.cast(list[X509CertMixin], ca_bundle)]
+
     @property
     def root(self) -> CertificateAuthority:
         """Get the root CA for this certificate."""

ca/django_ca/tests/models/base.py

@@ -17,7 +17,7 @@
 from django_ca.tests.base.constants import CERT_PEM_REGEX
 
 
-def assert_bundle(chain: list[X509CertMixin], cert: X509CertMixin) -> None:
+def assert_bundle(chain: list[X509CertMixin], bundle: str) -> None:
     """Assert that a bundle contains the expected certificates."""
     encoded_chain = [c.pub.pem.encode() for c in chain]
 
@@ -26,7 +26,6 @@ def assert_bundle(chain: list[X509CertMixin], cert: X509CertMixin) -> None:
     for member in encoded_chain:
         assert member.endswith(b"\n")
 
-    bundle = cert.bundle_as_pem
     assert isinstance(bundle, str)
     assert bundle.endswith("\n")
 

ca/django_ca/tests/models/test_certificate.py

@@ -16,6 +16,7 @@
 from datetime import datetime, timedelta, timezone as tz
 
 import josepy as jose
+from asgiref.sync import async_to_sync
 
 from cryptography import x509
 from cryptography.hazmat.primitives import hashes
@@ -24,6 +25,7 @@
 
 import pytest
 from _pytest.logging import LogCaptureFixture
+from pytest_django import DjangoAssertNumQueries
 from pytest_django.fixtures import SettingsWrapper
 
 from django_ca.constants import ReasonFlags
@@ -37,8 +39,25 @@ def test_bundle_as_pem(
     root: CertificateAuthority, root_cert: Certificate, child: CertificateAuthority, child_cert: Certificate
 ) -> None:
     """Test bundles of various CAs."""
-    assert_bundle([root_cert, root], root_cert)
-    assert_bundle([child_cert, child, root], child_cert)
+    assert_bundle([root_cert, root], root_cert.bundle_as_pem)
+    assert_bundle([child_cert, child, root], child_cert.bundle_as_pem)
+
+
+def test_aget_bundle_as_pem(
+    django_assert_num_queries: DjangoAssertNumQueries,
+    root: CertificateAuthority,
+    root_cert: Certificate,
+    child: CertificateAuthority,
+    child_cert: Certificate,
+) -> None:
+    """Test asynchronously getting the bundle."""
+    with django_assert_num_queries(0):
+        assert_bundle([root_cert, root], async_to_sync(root_cert.aget_bundle_as_pem)())
+        assert_bundle([child_cert, child, root], async_to_sync(child_cert.aget_bundle_as_pem)())
+
+    child_cert.refresh_from_db()
+    with django_assert_num_queries(1):
+        assert_bundle([child_cert, child, root], async_to_sync(child_cert.aget_bundle_as_pem)())
 
 
 def test_revocation() -> None:

ca/django_ca/tests/models/test_certificate_authority.py

@@ -22,6 +22,7 @@
 from typing import Any, NoReturn, Optional, Union, cast
 from unittest import mock
 
+from asgiref.sync import async_to_sync
 from pydantic import BaseModel
 
 from cryptography import x509
@@ -37,6 +38,7 @@
 
 import pytest
 from freezegun import freeze_time
+from pytest_django import DjangoAssertNumQueries
 from pytest_django.fixtures import SettingsWrapper
 
 from django_ca.conf import model_settings
@@ -102,8 +104,28 @@ def test_key_type(usable_cas: list[CertificateAuthority]) -> None:
 
 def test_bundle_as_pem(root: CertificateAuthority, child: CertificateAuthority) -> None:
     """Test bundles of various CAs."""
-    assert_bundle([root], root)
-    assert_bundle([child, root], child)
+    assert_bundle([root], root.bundle_as_pem)
+    assert_bundle([child, root], child.bundle_as_pem)
+
+
+def test_aget_bundle_as_pem_with_root(
+    django_assert_num_queries: DjangoAssertNumQueries, root: CertificateAuthority
+) -> None:
+    """Test Bundle for a root CA."""
+    with django_assert_num_queries(0):
+        assert_bundle([root], async_to_sync(root.aget_bundle_as_pem)())
+
+
+def test_aget_bundle_as_pem_with_child(
+    django_assert_num_queries: DjangoAssertNumQueries, root: CertificateAuthority, child: CertificateAuthority
+) -> None:
+    """Test Bundle for a child CA."""
+    with django_assert_num_queries(0):
+        assert_bundle([child, root], async_to_sync(child.aget_bundle_as_pem)())
+
+    child.refresh_from_db()
+    with django_assert_num_queries(1):
+        assert_bundle([child, root], async_to_sync(child.aget_bundle_as_pem)())
 
 
 def test_path_length(usable_ca: CertificateAuthority) -> None:

devscripts/validation/docker_compose.py

@@ -416,7 +416,7 @@ def get_postgres_version(path: Union[Path, str]) -> str:
     return parsed_data["services"]["db"]["image"].split(":")[1].split("-")[0]  # type: ignore[no-any-return]
 
 
-def test_update(release: str) -> int:
+def test_update(release: str) -> int:  # noqa: PLR0915
     """Validate updating with docker compose."""
     info("Validating docker compose update...")
     errors = 0