Fix/csr verify

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "1.19.4+261027"
+current_version = "1.19.5+261028"
 parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)\\+(?P<build>\\d+)"
 serialize = ["{major}.{minor}.{patch}+{build}"]
 search = "{current_version}"

Dockerfile

@@ -133,7 +133,7 @@ RUN --mount=type=ssh apt-get update && apt-get install -y \
     && rm -rf /var/lib/apt/lists/* \
     && chmod a+x /docker-entrypoint.sh \
     && WHEELFILE=`echo /tmp/wheelhouse/rasenmaeher_api-*.whl` \
-    && pip3 install --find-links=/tmp/wheelhouse/ "$WHEELFILE"[all] \
+    && pip3 install --no-index --find-links=/tmp/wheelhouse/ "$WHEELFILE"[all] \
     && rm -rf /tmp/wheelhouse/ \
     # Do whatever else you need to
     && true

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "rasenmaeher_api"
-version = "1.19.4+261027"
+version = "1.19.5+261028"
 description = "python-rasenmaeher-api"
 authors = [
     { name = "Aciid", email = "703382+Aciid@users.noreply.github.com" },

src/rasenmaeher_api/__init__.py

@@ -1,3 +1,3 @@
 """python-rasenmaeher-api"""
 
-__version__ = "1.19.4+261027"  # NOTE Use `bump-my-version` to bump versions correctly
+__version__ = "1.19.5+261028"  # NOTE Use `bump-my-version` to bump versions correctly

src/rasenmaeher_api/cert/cert_manager/private.py

@@ -9,15 +9,15 @@
 import base64
 import logging
 from pathlib import Path
-from typing import Any, Literal, cast
+from typing import Any, cast
 
 import cryptography.x509
 from cloudcoil.errors import APIError, ResourceConflict, ResourceNotFound, WaitTimeout
 from cloudcoil.models.cert_manager.v1 import CertificateRequest
 from cryptography.hazmat.primitives import serialization
-from cryptography.x509.oid import ExtendedKeyUsageOID
 
 from ...rmsettings import RMSettings
+from ..usages import KeyUsages, csr_usages, disallowed_usages
 from .base import CertManagerError
 from .chain import lean_chain
 from .names import cr_name
@@ -49,88 +49,10 @@ def _csr_common_name(csr_pem: str) -> str | None:
     return None
 
 
-type KeyUsages = Literal[
-    "signing",
-    "digital signature",
-    "content commitment",
-    "key encipherment",
-    "key agreement",
-    "data encipherment",
-    "cert sign",
-    "crl sign",
-    "encipher only",
-    "decipher only",
-    "any",
-    "server auth",
-    "client auth",
-    "code signing",
-    "email protection",
-    "s/mime",
-    "ipsec end system",
-    "ipsec tunnel",
-    "ipsec user",
-    "timestamping",
-    "ocsp signing",
-    "microsoft sgc",
-    "netscape sgc",
-]
-
-_KEY_USAGE_NAMES: tuple[tuple[str, KeyUsages], ...] = (
-    ("digital_signature", "digital signature"),
-    ("content_commitment", "content commitment"),
-    ("key_encipherment", "key encipherment"),
-    ("data_encipherment", "data encipherment"),
-    ("key_agreement", "key agreement"),
-    ("key_cert_sign", "cert sign"),
-    ("crl_sign", "crl sign"),
-)
-
-_EXT_KEY_USAGE_NAMES: dict[cryptography.x509.ObjectIdentifier, KeyUsages] = {
-    ExtendedKeyUsageOID.SERVER_AUTH: "server auth",
-    ExtendedKeyUsageOID.CLIENT_AUTH: "client auth",
-    ExtendedKeyUsageOID.CODE_SIGNING: "code signing",
-    ExtendedKeyUsageOID.EMAIL_PROTECTION: "email protection",
-    ExtendedKeyUsageOID.TIME_STAMPING: "timestamping",
-    ExtendedKeyUsageOID.OCSP_SIGNING: "ocsp signing",
-}
-
-
-def _csr_usages(csr_pem: str) -> list[KeyUsages]:
-    """Extract cert-manager `spec.usages` values from a CSR's KeyUsage extensions.
-
-    cert-manager's admission webhook rejects requests whose declared ``usages``
-    don't match the KeyUsage/ExtendedKeyUsage extensions in the CSR. Derive the
-    list dynamically so we honor whatever the caller embedded.
-    """
-    csr = cryptography.x509.load_pem_x509_csr(csr_pem.encode("utf-8"))
-    out: list[KeyUsages] = []
-    try:
-        ku = csr.extensions.get_extension_for_class(cryptography.x509.KeyUsage).value
-        for attr, name in _KEY_USAGE_NAMES:
-            if getattr(ku, attr, False):
-                out.append(name)
-        if ku.key_agreement:
-            if getattr(ku, "encipher_only", False):
-                out.append("encipher only")
-            if getattr(ku, "decipher_only", False):
-                out.append("decipher only")
-    except cryptography.x509.ExtensionNotFound:
-        pass
-    try:
-        eku = csr.extensions.get_extension_for_class(cryptography.x509.ExtendedKeyUsage).value
-        for oid in eku:
-            eku_name = _EXT_KEY_USAGE_NAMES.get(oid)
-            if eku_name:
-                out.append(eku_name)
-    except cryptography.x509.ExtensionNotFound:
-        pass
-    return out
-
-
 async def _create_cr(name: str, namespace: str, csr_pem: str, csr_b64: str, settings: RMSettings) -> CertificateRequest:
     """Create the CertificateRequest and return it."""
     try:
-        usages: list[KeyUsages] = _csr_usages(csr_pem)
+        usages: list[KeyUsages] = csr_usages(csr_pem)
         if not usages:
             # Default for clients that ship a CSR without explicit KeyUsage extensions.
             usages = ["digital signature", "key encipherment", "client auth"]
@@ -195,6 +117,8 @@ async def sign_csr(csr: str, bundle: bool = True) -> str:
     """
     settings = RMSettings.singleton()
     cn = _csr_common_name(csr)
+    if extra := disallowed_usages(csr, cn):
+        raise ValueError(f"CSR for {cn} requests usages not allowed for it: {sorted(extra)}")
     name = cr_name(csr, cn)
     namespace = settings.cert_manager_namespace
     csr_b64 = _csr_pem_to_b64(csr)

src/rasenmaeher_api/cert/usages.py

@@ -0,0 +1,98 @@
+"""Key usages a CSR may request, shared by the signing backends and CSR verification"""
+
+from typing import Literal
+
+import cryptography.x509
+from cryptography.x509.oid import ExtendedKeyUsageOID
+
+from ..rmsettings import RMSettings
+
+type KeyUsages = Literal[
+    "signing",
+    "digital signature",
+    "content commitment",
+    "key encipherment",
+    "key agreement",
+    "data encipherment",
+    "cert sign",
+    "crl sign",
+    "encipher only",
+    "decipher only",
+    "any",
+    "server auth",
+    "client auth",
+    "code signing",
+    "email protection",
+    "s/mime",
+    "ipsec end system",
+    "ipsec tunnel",
+    "ipsec user",
+    "timestamping",
+    "ocsp signing",
+    "microsoft sgc",
+    "netscape sgc",
+]
+
+_KEY_USAGE_NAMES: tuple[tuple[str, KeyUsages], ...] = (
+    ("digital_signature", "digital signature"),
+    ("content_commitment", "content commitment"),
+    ("key_encipherment", "key encipherment"),
+    ("data_encipherment", "data encipherment"),
+    ("key_agreement", "key agreement"),
+    ("key_cert_sign", "cert sign"),
+    ("crl_sign", "crl sign"),
+)
+
+_EXT_KEY_USAGE_NAMES: dict[cryptography.x509.ObjectIdentifier, KeyUsages] = {
+    ExtendedKeyUsageOID.SERVER_AUTH: "server auth",
+    ExtendedKeyUsageOID.CLIENT_AUTH: "client auth",
+    ExtendedKeyUsageOID.CODE_SIGNING: "code signing",
+    ExtendedKeyUsageOID.EMAIL_PROTECTION: "email protection",
+    ExtendedKeyUsageOID.TIME_STAMPING: "timestamping",
+    ExtendedKeyUsageOID.OCSP_SIGNING: "ocsp signing",
+}
+
+_CLIENT_USAGES: frozenset[KeyUsages] = frozenset(
+    {"digital signature", "content commitment", "key encipherment", "client auth"}
+)
+
+
+def csr_usages(csr_pem: str) -> list[KeyUsages]:
+    """Extract cert-manager `spec.usages` values from a CSR's KeyUsage extensions.
+
+    cert-manager's admission webhook rejects requests whose declared ``usages``
+    don't match the KeyUsage/ExtendedKeyUsage extensions in the CSR. Derive the
+    list dynamically so we honor whatever the caller embedded.
+    """
+    csr = cryptography.x509.load_pem_x509_csr(csr_pem.encode("utf-8"))
+    out: list[KeyUsages] = []
+    try:
+        ku = csr.extensions.get_extension_for_class(cryptography.x509.KeyUsage).value
+        for attr, name in _KEY_USAGE_NAMES:
+            if getattr(ku, attr, False):
+                out.append(name)
+        if ku.key_agreement:
+            if getattr(ku, "encipher_only", False):
+                out.append("encipher only")
+            if getattr(ku, "decipher_only", False):
+                out.append("decipher only")
+    except cryptography.x509.ExtensionNotFound:
+        pass
+    try:
+        eku = csr.extensions.get_extension_for_class(cryptography.x509.ExtendedKeyUsage).value
+        for oid in eku:
+            eku_name = _EXT_KEY_USAGE_NAMES.get(oid)
+            if eku_name is None:
+                raise ValueError(f"Unsupported extended key usage {oid.dotted_string}")
+            out.append(eku_name)
+    except cryptography.x509.ExtensionNotFound:
+        pass
+    return out
+
+
+def disallowed_usages(csr_pem: str, cn: str | None) -> set[KeyUsages]:
+    """Usages the CSR requests beyond client usages, with server auth allowed only for service CNs"""
+    extra = set(csr_usages(csr_pem)) - _CLIENT_USAGES
+    if extra == {"server auth"} and cn is not None and RMSettings.singleton().is_reserved_cn(cn):
+        return set()
+    return extra

src/rasenmaeher_api/db/enrollments.py

@@ -281,10 +281,10 @@ async def create_for_callsign(
         csr: str | None = None,
     ) -> "Enrollment":
         """Create a new one with random code for the callsign"""
-        if callsign in RMSettings.singleton().valid_product_cns:
-            raise CallsignReserved("Using product CNs as callsigns is forbidden")
+        if RMSettings.singleton().is_reserved_cn(callsign):
+            raise CallsignReserved("Service CN used as callsign", detail="Callsign reserved")
         if csr and not verify_csr(csr, callsign):
-            raise CallsignReserved("CSR CN must match callsign")
+            raise CallsignReserved(detail="CSR subject or key usages not allowed for callsign")
         with EngineWrapper.get_session() as session:
             try:
                 await Enrollment.by_callsign(callsign)

src/rasenmaeher_api/db/errors.py

@@ -45,10 +45,10 @@ class EnrollmentError(Exception):
 class CallsignReserved(HTTPException, EnrollmentError, ValueError):
     """Callsign is already reserved"""
 
-    def __init__(self, *args: Sequence[Any]) -> None:
-        """make us also 403 HTTP error"""
+    def __init__(self, *args: Sequence[Any], detail: str = "Forbidden") -> None:
+        """make us also 403 HTTP error, only detail is shown to the client"""
         self.status_code = status.HTTP_403_FORBIDDEN
-        self.detail = "Forbidden"
+        self.detail = detail
         new_args = [status.HTTP_403_FORBIDDEN] + list(args)
         super(HTTPException, self).__init__(*new_args)
 

src/rasenmaeher_api/db/people.py

@@ -96,13 +96,11 @@ async def create_with_cert(
         cls, callsign: str, extra: dict[str, Any] | None = None, csrpem: str | None = None
     ) -> "Person":
         """Create the cert etc and save the person"""
-        if csrpem and not verify_csr(csrpem, callsign):
-            raise CallsignReserved("CSR CN must match callsign")
         cnf = RMSettings.singleton()
-        if callsign in cnf.valid_product_cns:
-            raise CallsignReserved("Using product CNs as callsigns is forbidden")
-        if callsign == cnf.mtls_client_cert_cn:
-            raise CallsignReserved("Using the rasenmaeher service CN as a callsign is forbidden")
+        if cnf.is_reserved_cn(callsign):
+            raise CallsignReserved("Service CN used as callsign", detail="Callsign reserved")
+        if csrpem and not verify_csr(csrpem, callsign):
+            raise CallsignReserved(detail="CSR subject or key usages not allowed for callsign")
         try:
             await Person.by_callsign(callsign)
             raise CallsignReserved()

src/rasenmaeher_api/rmsettings.py

@@ -183,5 +183,12 @@ def valid_product_cns(self) -> list[str]:
         self.load_manifest()
         return [product["certcn"] for product in self.kraftwerk_manifest_dict["products"].values()]
 
+    def is_reserved_cn(self, cn: str) -> bool:
+        """True for CNs that belong to services, never to users"""
+        normalized_cn = cn.lower()
+        return normalized_cn == self.mtls_client_cert_cn.lower() or normalized_cn in (
+            product_cn.lower() for product_cn in self.valid_product_cns
+        )
+
 
 switchme_to_singleton_call = RMSettings.singleton()

src/rasenmaeher_api/web/api/enrollment/views.py

@@ -10,7 +10,7 @@
 from multikeyjwt import config as jwtconfig
 
 from ....db import Enrollment, EnrollmentPool, Person
-from ....db.errors import NotFound
+from ....db.errors import CallsignReserved, NotFound
 from ....rmsettings import RMSettings
 from ..middleware.mtls import MTLSorJWT
 from ..middleware.user import ValidUser
@@ -717,7 +717,22 @@ async def post_enroll_invite_code(
     except NotFound:
         pass
 
-    enrollment = await obj.create_enrollment(callsign=callsign, csr=request_in.csr)
+    try:
+        enrollment = await obj.create_enrollment(callsign=callsign, csr=request_in.csr)
+    except CallsignReserved as exc:
+        LOGGER.audit(  # type: ignore[attr-defined]
+            "Enrollment via invitecode failed - callsign not allowed",
+            extra=build_audit_extra(
+                action="enrollment_invitecode",
+                outcome="failure",
+                target=callsign,
+                request=request,
+                error_code="CALLSIGN_RESERVED",
+                invitecode_id=str(obj.pk),
+            ),
+        )
+        LOGGER.error(f"{request.url} : {exc.detail}")
+        raise HTTPException(status_code=400, detail=exc.detail) from exc
 
     # Create JWT token for user
     claims = {"sub": callsign}

src/rasenmaeher_api/web/api/product/views.py

@@ -33,7 +33,10 @@
 async def csr_common(certs: CertificatesRequest) -> CertificatesResponse:
     """Common parts of CSR handling"""
     cachain = await get_ca()
-    certpem = (await sign_csr(certs.csr)).replace("\\n", "\n")
+    try:
+        certpem = (await sign_csr(certs.csr)).replace("\\n", "\n")
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
     bundlepem = await get_bundle(certpem)
 
     await record_issued_cert(certpem)
@@ -112,16 +115,7 @@ async def renew_csr(
     if ensure_str(req_cn) != ensure_str(peer_dn["CN"]):
         raise HTTPException(403, "Renewal must be for same name")
 
-    cachain = await get_ca()
-    certpem = (await sign_csr(certs.csr)).replace("\\n", "\n")
-    bundlepem = await get_bundle(certpem)
-
-    await record_issued_cert(certpem)
-
-    return CertificatesResponse(
-        ca=cachain,
-        certificate=bundlepem,
-    )
+    return await csr_common(certs)
 
 
 @router.post("/ready", dependencies=[Depends(MTLSHeader(auto_error=True))])