From 55a11bea1b742e38d4dc593dff251b4cd208f764 Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Tue, 27 Oct 2026 18:03:08 +0200 Subject: [PATCH 1/9] fix(csr): verify CN with strict equivalence instead of inclusion check --- .../web/api/utils/csr_utils.py | 4 ++- tests/test_csr_utils.py | 26 +++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) create mode 100644 tests/test_csr_utils.py diff --git a/src/rasenmaeher_api/web/api/utils/csr_utils.py b/src/rasenmaeher_api/web/api/utils/csr_utils.py index 836ff504..7d0b2497 100644 --- a/src/rasenmaeher_api/web/api/utils/csr_utils.py +++ b/src/rasenmaeher_api/web/api/utils/csr_utils.py @@ -3,6 +3,7 @@ import logging from cryptography import x509 +from cryptography.x509.oid import NameOID from libadvian.binpackers import ensure_utf8 LOGGER = logging.getLogger(__name__) @@ -16,7 +17,8 @@ def verify_csr(csrpem: str, callsign: str) -> bool: csr = x509.load_pem_x509_csr(ensure_utf8(csrpem)) dn = csr.subject.rfc4514_string() LOGGER.debug(f"DN={dn} callsign={callsign}") - if f"CN={callsign}" not in dn: + cns = csr.subject.get_attributes_for_oid(NameOID.COMMON_NAME) + if len(csr.subject) != 1 or len(cns) != 1 or cns[0].value != callsign: LOGGER.warning(f"Callsign does not match CSR subject. DN={dn} callsign={callsign}") return False # TODO: check that keyusages in the CSR are fine diff --git a/tests/test_csr_utils.py b/tests/test_csr_utils.py new file mode 100644 index 00000000..abb4829e --- /dev/null +++ b/tests/test_csr_utils.py @@ -0,0 +1,26 @@ +"""Test CSR subject verification""" + +from pathlib import Path + +import pytest +from libpvarki.mtlshelp.csr import async_create_client_csr, async_create_keypair + +from rasenmaeher_api.web.api.utils.csr_utils import verify_csr + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "subject,callsign,expected", + [ + pytest.param({"CN": "csr_test"}, "csr_test", True, id="exact-match"), + pytest.param({"CN": "csr_test2"}, "csr_test", False, id="callsign-prefix-of-cn"), + pytest.param({"CN": "csr_test", "O": "x"}, "csr_test", False, id="extra-subject-attribute"), + pytest.param({"CN": "a,b"}, "a,b", True, id="rfc4514-escaped-char"), + ], +) +async def test_verify_csr_subject(subject: dict[str, str], callsign: str, expected: bool, nice_tmpdir: str) -> None: + """Only a CN-only subject exactly matching the callsign passes""" + tempdir = Path(nice_tmpdir) + ckp = await async_create_keypair(tempdir / "user.key", tempdir / "user.pub") + csrpem = await async_create_client_csr(ckp, tempdir / "user.csr", subject) + assert verify_csr(csrpem, callsign) is expected From a4b5ba5d8af265436dab29797c7d6dd2842dc253 Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:06:14 +0200 Subject: [PATCH 2/9] fix: create single function for verifying CN is not reserved --- src/rasenmaeher_api/db/enrollments.py | 4 ++-- src/rasenmaeher_api/db/people.py | 8 +++----- src/rasenmaeher_api/rmsettings.py | 7 +++++++ tests/test_db.py | 20 ++++++++++++++++---- tests/test_rasenmaeher_api.py | 17 +++++++++++++++++ 5 files changed, 45 insertions(+), 11 deletions(-) diff --git a/src/rasenmaeher_api/db/enrollments.py b/src/rasenmaeher_api/db/enrollments.py index a07b02c4..93223550 100644 --- a/src/rasenmaeher_api/db/enrollments.py +++ b/src/rasenmaeher_api/db/enrollments.py @@ -281,8 +281,8 @@ 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("Using service CNs as callsigns is forbidden") if csr and not verify_csr(csr, callsign): raise CallsignReserved("CSR CN must match callsign") with EngineWrapper.get_session() as session: diff --git a/src/rasenmaeher_api/db/people.py b/src/rasenmaeher_api/db/people.py index 086c3ffc..75073db0 100644 --- a/src/rasenmaeher_api/db/people.py +++ b/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""" + cnf = RMSettings.singleton() + if cnf.is_reserved_cn(callsign): + raise CallsignReserved("Using service CNs as callsigns is forbidden") 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") try: await Person.by_callsign(callsign) raise CallsignReserved() diff --git a/src/rasenmaeher_api/rmsettings.py b/src/rasenmaeher_api/rmsettings.py index 00b4cdeb..2fd5e58a 100644 --- a/src/rasenmaeher_api/rmsettings.py +++ b/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() diff --git a/tests/test_db.py b/tests/test_db.py index 4676eb24..04ff5036 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -450,14 +450,26 @@ async def wait_for_pfxfile() -> None: assert pfxdata.cert +@pytest.mark.parametrize("callsign", ["fake.localmaeher.dev.pvarki.fi", "FAKE.localmaeher.dev.pvarki.fi"]) @pytest.mark.asyncio(loop_scope="session") -async def test_productcn_forbid(dbinit_func) -> None: - """Test that trying to create enrollment or person with callsign that matches a product CN fails""" +async def test_productcn_forbid(dbinit_func, callsign: str) -> None: + """Test that trying to create enrollment or person with callsign that matches a product CN fails, regardless of case""" _ = dbinit_func with pytest.raises(CallsignReserved): - await Person.create_with_cert("fake.localmaeher.dev.pvarki.fi") + await Person.create_with_cert(callsign) + with pytest.raises(CallsignReserved): + await Enrollment.create_for_callsign(callsign) + + +@pytest.mark.parametrize("callsign", ["rasenmaeher", "RASENMAEHER", "RasenMaeher"]) +@pytest.mark.asyncio(loop_scope="session") +async def test_servicecn_forbid(dbinit_func, callsign: str) -> None: + """Test that rasenmaeher service cn is forbidden as a callsign regardless of case""" + _ = dbinit_func + with pytest.raises(CallsignReserved): + await Person.create_with_cert(callsign) with pytest.raises(CallsignReserved): - await Enrollment.create_for_callsign("fake.localmaeher.dev.pvarki.fi") + await Enrollment.create_for_callsign(callsign) def _self_signed_pem(common_name: str) -> str: diff --git a/tests/test_rasenmaeher_api.py b/tests/test_rasenmaeher_api.py index 62b6aa1c..a91902bd 100644 --- a/tests/test_rasenmaeher_api.py +++ b/tests/test_rasenmaeher_api.py @@ -33,6 +33,23 @@ def test_settings() -> None: assert "fake.localmaeher.dev.pvarki.fi" in conf.valid_product_cns +@pytest.mark.parametrize( + ("cn", "expected"), + [ + ("rasenmaeher", True), + ("RASENMAEHER", True), + ("fake.localmaeher.dev.pvarki.fi", True), + ("Fake.Localmaeher.Dev.Pvarki.Fi", True), + ("kissa", False), + ("rasenmaeher2", False), + ], +) +def test_is_reserved_cn(cn: str, expected: bool) -> None: + """Test service CN detection is case-insensitive""" + conf = RMSettings.singleton() + assert conf.is_reserved_cn(cn) == expected + + # FIXME: Figure out WTF is asyncios problem @pytest.mark.skip(reason="asyncio/asyncpg is weird under pytest") @pytest.mark.asyncio(loop_scope="session") From aa65989e1e7e7bf31477c5c2ce3ae709d0ffacb0 Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:29:38 +0200 Subject: [PATCH 3/9] feat: verify usages in cert-manager CSR --- .../cert/cert_manager/private.py | 15 ++++ tests/cert/cert_manager/test_private.py | 70 +++++++++++++++++++ 2 files changed, 85 insertions(+) diff --git a/src/rasenmaeher_api/cert/cert_manager/private.py b/src/rasenmaeher_api/cert/cert_manager/private.py index eecf73cb..5f8b9da0 100644 --- a/src/rasenmaeher_api/cert/cert_manager/private.py +++ b/src/rasenmaeher_api/cert/cert_manager/private.py @@ -127,6 +127,20 @@ def _csr_usages(csr_pem: str) -> list[KeyUsages]: return out +_CLIENT_USAGES: frozenset[KeyUsages] = frozenset( + {"digital signature", "content commitment", "key encipherment", "client auth"} +) + + +def _check_usages(csr_pem: str, cn: str | None, settings: RMSettings) -> None: + """Allow client usages for everyone and server auth only for service CNs""" + extra = set(_csr_usages(csr_pem)) - _CLIENT_USAGES + if extra == {"server auth"} and cn is not None and settings.is_reserved_cn(cn): + return + if extra: + raise CertManagerError(f"CSR for {cn} requests usages not allowed for it: {sorted(extra)}") + + async def _create_cr(name: str, namespace: str, csr_pem: str, csr_b64: str, settings: RMSettings) -> CertificateRequest: """Create the CertificateRequest and return it.""" try: @@ -195,6 +209,7 @@ async def sign_csr(csr: str, bundle: bool = True) -> str: """ settings = RMSettings.singleton() cn = _csr_common_name(csr) + _check_usages(csr, cn, settings) name = cr_name(csr, cn) namespace = settings.cert_manager_namespace csr_b64 = _csr_pem_to_b64(csr) diff --git a/tests/cert/cert_manager/test_private.py b/tests/cert/cert_manager/test_private.py index 6ca1d64f..7001df2b 100644 --- a/tests/cert/cert_manager/test_private.py +++ b/tests/cert/cert_manager/test_private.py @@ -228,6 +228,76 @@ async def test_sign_csr_honours_csr_usages( assert created.spec.usages == ["digital signature", "client auth"] +@pytest.mark.parametrize( + ("cn", "ext_key_usage"), + [ + ("fake.localmaeher.dev.pvarki.fi", [ExtendedKeyUsageOID.SERVER_AUTH, ExtendedKeyUsageOID.CLIENT_AUTH]), + ("FAKE.localmaeher.dev.pvarki.fi", [ExtendedKeyUsageOID.SERVER_AUTH]), + ("rasenmaeher", [ExtendedKeyUsageOID.SERVER_AUTH]), + ], +) +@pytest.mark.asyncio(loop_scope="function") +async def test_sign_csr_allows_server_auth_for_service_cns( + fake_k8s: FakeK8s, + cm_settings: RMSettings, + mint_csr: CsrFactory, + cn: str, + ext_key_usage: list[cryptography.x509.ObjectIdentifier], +) -> None: + """Service CNs may request server auth""" + _ = cm_settings + await private.sign_csr(mint_csr(cn, ext_key_usage=ext_key_usage), bundle=False) + created = fake_k8s.created[0] + assert created.spec is not None + assert "server auth" in (created.spec.usages or []) + + +@pytest.mark.asyncio(loop_scope="function") +async def test_sign_csr_allows_libpvarki_client_csr( + fake_k8s: FakeK8s, cm_settings: RMSettings, mint_csr: CsrFactory, key_usage: KeyUsageFactory +) -> None: + """A user CN with the usages libpvarki's create_client_csr sets (as TAK sends) is signed""" + _ = cm_settings + csr = mint_csr( + "kissa", + key_usage(digital_signature=True, content_commitment=True, key_encipherment=True), + [ExtendedKeyUsageOID.CLIENT_AUTH], + ) + await private.sign_csr(csr, bundle=False) + created = fake_k8s.created[0] + assert created.spec is not None + assert created.spec.usages == ["digital signature", "content commitment", "key encipherment", "client auth"] + + +@pytest.mark.parametrize( + ("cn", "key_usage_flags", "ext_key_usage"), + [ + ("kissa", {}, [ExtendedKeyUsageOID.SERVER_AUTH]), + (None, {}, [ExtendedKeyUsageOID.SERVER_AUTH]), + ("fake.localmaeher.dev.pvarki.fi", {}, [ExtendedKeyUsageOID.CODE_SIGNING]), + ("fake.localmaeher.dev.pvarki.fi", {}, [ExtendedKeyUsageOID.OCSP_SIGNING]), + ("fake.localmaeher.dev.pvarki.fi", {"key_cert_sign": True}, None), + ("kissa", {"data_encipherment": True}, [ExtendedKeyUsageOID.CLIENT_AUTH]), + ], +) +@pytest.mark.asyncio(loop_scope="function") +async def test_sign_csr_rejects_disallowed_usages( + fake_k8s: FakeK8s, + cm_settings: RMSettings, + mint_csr: CsrFactory, + key_usage: KeyUsageFactory, + cn: str | None, + key_usage_flags: dict[str, bool], + ext_key_usage: list[cryptography.x509.ObjectIdentifier] | None, +) -> None: + """Usages outside the caller's allowed set are refused before any CR is created""" + _ = cm_settings + csr = mint_csr(cn, key_usage(**key_usage_flags) if key_usage_flags else None, ext_key_usage) + with pytest.raises(CertManagerError, match="not allowed"): + await private.sign_csr(csr, bundle=False) + assert not fake_k8s.created + + @pytest.mark.asyncio(loop_scope="function") async def test_sign_csr_conflict_polls_existing( fake_k8s: FakeK8s, cm_settings: RMSettings, csr_pem: str, monkeypatch: pytest.MonkeyPatch From 8b8e171a4f3473493ea8ce4a638bc0a7c7f284d0 Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:31:07 +0200 Subject: [PATCH 4/9] chore: bump version to 1.19.5+261028 --- .bumpversion.toml | 2 +- pyproject.toml | 2 +- src/rasenmaeher_api/__init__.py | 2 +- tests/test_rasenmaeher_api.py | 2 +- uv.lock | 2 +- 5 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.bumpversion.toml b/.bumpversion.toml index 9609497d..bff7649f 100644 --- a/.bumpversion.toml +++ b/.bumpversion.toml @@ -1,5 +1,5 @@ [tool.bumpversion] -current_version = "1.19.4+261027" +current_version = "1.19.5+261028" parse = "(?P\\d+)\\.(?P\\d+)\\.(?P\\d+)\\+(?P\\d+)" serialize = ["{major}.{minor}.{patch}+{build}"] search = "{current_version}" diff --git a/pyproject.toml b/pyproject.toml index e4e8dd7c..ff29a035 100644 --- a/pyproject.toml +++ b/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" }, diff --git a/src/rasenmaeher_api/__init__.py b/src/rasenmaeher_api/__init__.py index 0301e277..7b655f90 100644 --- a/src/rasenmaeher_api/__init__.py +++ b/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 diff --git a/tests/test_rasenmaeher_api.py b/tests/test_rasenmaeher_api.py index a91902bd..bbf1dfa9 100644 --- a/tests/test_rasenmaeher_api.py +++ b/tests/test_rasenmaeher_api.py @@ -15,7 +15,7 @@ def test_version() -> None: """Make sure version matches expected""" - assert __version__ == "1.19.4+261027" + assert __version__ == "1.19.5+261028" @pytest.mark.asyncio(loop_scope="session") diff --git a/uv.lock b/uv.lock index 7e461d3c..21fa0ae7 100644 --- a/uv.lock +++ b/uv.lock @@ -2150,7 +2150,7 @@ wheels = [ [[package]] name = "rasenmaeher-api" -version = "1.19.4+261027" +version = "1.19.5+261028" source = { editable = "." } dependencies = [ { name = "aiodns" }, From 82877264292b94aec3f28448b3672290bda1635a Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Wed, 28 Oct 2026 19:09:12 +0200 Subject: [PATCH 5/9] fix(certificates): move EKU logic to usages.py --- .../cert/cert_manager/private.py | 101 +----------------- src/rasenmaeher_api/cert/usages.py | 97 +++++++++++++++++ src/rasenmaeher_api/db/enrollments.py | 2 +- src/rasenmaeher_api/db/people.py | 2 +- .../web/api/utils/csr_utils.py | 10 +- tests/cert/cert_manager/test_private.py | 24 ----- tests/cert/test_usages.py | 66 ++++++++++++ tests/test_csr_utils.py | 11 +- 8 files changed, 186 insertions(+), 127 deletions(-) create mode 100644 src/rasenmaeher_api/cert/usages.py create mode 100644 tests/cert/test_usages.py diff --git a/src/rasenmaeher_api/cert/cert_manager/private.py b/src/rasenmaeher_api/cert/cert_manager/private.py index 5f8b9da0..71535090 100644 --- a/src/rasenmaeher_api/cert/cert_manager/private.py +++ b/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,102 +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 - - -_CLIENT_USAGES: frozenset[KeyUsages] = frozenset( - {"digital signature", "content commitment", "key encipherment", "client auth"} -) - - -def _check_usages(csr_pem: str, cn: str | None, settings: RMSettings) -> None: - """Allow client usages for everyone and server auth only for service CNs""" - extra = set(_csr_usages(csr_pem)) - _CLIENT_USAGES - if extra == {"server auth"} and cn is not None and settings.is_reserved_cn(cn): - return - if extra: - raise CertManagerError(f"CSR for {cn} requests usages not allowed for it: {sorted(extra)}") - - 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"] @@ -209,7 +117,8 @@ async def sign_csr(csr: str, bundle: bool = True) -> str: """ settings = RMSettings.singleton() cn = _csr_common_name(csr) - _check_usages(csr, cn, settings) + if extra := disallowed_usages(csr, cn): + raise CertManagerError(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) diff --git a/src/rasenmaeher_api/cert/usages.py b/src/rasenmaeher_api/cert/usages.py new file mode 100644 index 00000000..b6220c43 --- /dev/null +++ b/src/rasenmaeher_api/cert/usages.py @@ -0,0 +1,97 @@ +"""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: + 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 diff --git a/src/rasenmaeher_api/db/enrollments.py b/src/rasenmaeher_api/db/enrollments.py index 93223550..a349c503 100644 --- a/src/rasenmaeher_api/db/enrollments.py +++ b/src/rasenmaeher_api/db/enrollments.py @@ -284,7 +284,7 @@ async def create_for_callsign( if RMSettings.singleton().is_reserved_cn(callsign): raise CallsignReserved("Using service CNs as callsigns is forbidden") if csr and not verify_csr(csr, callsign): - raise CallsignReserved("CSR CN must match callsign") + raise CallsignReserved("CSR subject or key usages not allowed for callsign") with EngineWrapper.get_session() as session: try: await Enrollment.by_callsign(callsign) diff --git a/src/rasenmaeher_api/db/people.py b/src/rasenmaeher_api/db/people.py index 75073db0..a30b0de3 100644 --- a/src/rasenmaeher_api/db/people.py +++ b/src/rasenmaeher_api/db/people.py @@ -100,7 +100,7 @@ async def create_with_cert( if cnf.is_reserved_cn(callsign): raise CallsignReserved("Using service CNs as callsigns is forbidden") if csrpem and not verify_csr(csrpem, callsign): - raise CallsignReserved("CSR CN must match callsign") + raise CallsignReserved("CSR subject or key usages not allowed for callsign") try: await Person.by_callsign(callsign) raise CallsignReserved() diff --git a/src/rasenmaeher_api/web/api/utils/csr_utils.py b/src/rasenmaeher_api/web/api/utils/csr_utils.py index 7d0b2497..bacbb708 100644 --- a/src/rasenmaeher_api/web/api/utils/csr_utils.py +++ b/src/rasenmaeher_api/web/api/utils/csr_utils.py @@ -6,6 +6,8 @@ from cryptography.x509.oid import NameOID from libadvian.binpackers import ensure_utf8 +from ....cert.usages import disallowed_usages + LOGGER = logging.getLogger(__name__) @@ -13,7 +15,7 @@ def verify_csr(csrpem: str, callsign: str) -> bool: - """Verify CSR matches our rules for CN/DN for the given callsign""" + """Verify CSR matches our rules for CN/DN and key usages for the given callsign""" csr = x509.load_pem_x509_csr(ensure_utf8(csrpem)) dn = csr.subject.rfc4514_string() LOGGER.debug(f"DN={dn} callsign={callsign}") @@ -21,7 +23,7 @@ def verify_csr(csrpem: str, callsign: str) -> bool: if len(csr.subject) != 1 or len(cns) != 1 or cns[0].value != callsign: LOGGER.warning(f"Callsign does not match CSR subject. DN={dn} callsign={callsign}") return False - # TODO: check that keyusages in the CSR are fine - # crypto.X509Extension(b"keyUsage", True, b"digitalSignature,nonRepudiation,keyEncipherment"), - # crypto.X509Extension(b"extendedKeyUsage", True, b"clientAuth"), + if extra := disallowed_usages(csrpem, callsign): + LOGGER.warning(f"CSR requests usages not allowed for callsign. usages={sorted(extra)} callsign={callsign}") + return False return True diff --git a/tests/cert/cert_manager/test_private.py b/tests/cert/cert_manager/test_private.py index 7001df2b..34b3f69c 100644 --- a/tests/cert/cert_manager/test_private.py +++ b/tests/cert/cert_manager/test_private.py @@ -150,30 +150,6 @@ def test_common_name_absent_is_none(mint_csr: CsrFactory) -> None: assert private._csr_common_name(mint_csr(None)) is None -# _csr_usages -def test_usages_empty_without_extensions(csr_pem: str) -> None: - """A CSR with no KeyUsage or ExtendedKeyUsage extensions derives no usages""" - assert private._csr_usages(csr_pem) == [] - - -def test_key_usage_flags_mapped(mint_csr: CsrFactory, key_usage: KeyUsageFactory) -> None: - """KeyUsage bits map to their cert-manager names, dependent flags last""" - plain = mint_csr("example", key_usage(digital_signature=True, key_encipherment=True)) - assert private._csr_usages(plain) == ["digital signature", "key encipherment"] - agreement_only = mint_csr("example", key_usage(key_agreement=True)) - assert private._csr_usages(agreement_only) == ["key agreement"] - agreement = mint_csr("example", key_usage(key_agreement=True, encipher_only=True, decipher_only=True)) - assert private._csr_usages(agreement) == ["key agreement", "encipher only", "decipher only"] - - -def test_ext_key_usage_oids_mapped(mint_csr: CsrFactory) -> None: - """Known ExtendedKeyUsage OIDs map to their names, unknown ones are dropped""" - known = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.SERVER_AUTH]) - assert private._csr_usages(known) == ["server auth"] - unknown = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.KERBEROS_PKINIT_KDC]) - assert private._csr_usages(unknown) == [] - - # sign_csr @pytest.mark.asyncio(loop_scope="function") async def test_sign_csr_returns_bundle(fake_k8s: FakeK8s, cm_settings: RMSettings, csr_pem: str) -> None: diff --git a/tests/cert/test_usages.py b/tests/cert/test_usages.py new file mode 100644 index 00000000..791e2c9b --- /dev/null +++ b/tests/cert/test_usages.py @@ -0,0 +1,66 @@ +"""Test key usage extraction and the allowed usage rules""" + +import pytest +from cryptography.x509 import ObjectIdentifier +from cryptography.x509.oid import ExtendedKeyUsageOID + +from rasenmaeher_api.cert.usages import csr_usages, disallowed_usages +from tests.cert.helpers import CsrFactory, KeyUsageFactory + + +def test_usages_empty_without_extensions(csr_pem: str) -> None: + """A CSR with no KeyUsage or ExtendedKeyUsage extensions derives no usages""" + assert csr_usages(csr_pem) == [] + + +def test_key_usage_flags_mapped(mint_csr: CsrFactory, key_usage: KeyUsageFactory) -> None: + """KeyUsage bits map to their cert-manager names, dependent flags last""" + plain = mint_csr("example", key_usage(digital_signature=True, key_encipherment=True)) + assert csr_usages(plain) == ["digital signature", "key encipherment"] + agreement_only = mint_csr("example", key_usage(key_agreement=True)) + assert csr_usages(agreement_only) == ["key agreement"] + agreement = mint_csr("example", key_usage(key_agreement=True, encipher_only=True, decipher_only=True)) + assert csr_usages(agreement) == ["key agreement", "encipher only", "decipher only"] + + +def test_ext_key_usage_oids_mapped(mint_csr: CsrFactory) -> None: + """Known ExtendedKeyUsage OIDs map to their names, unknown ones are dropped""" + known = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.SERVER_AUTH]) + assert csr_usages(known) == ["server auth"] + unknown = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.KERBEROS_PKINIT_KDC]) + assert csr_usages(unknown) == [] + + +@pytest.mark.parametrize( + ("cn", "key_usage_flags", "ext_key_usage", "expected"), + [ + ("kissa", {}, None, set()), + ( + "kissa", + {"digital_signature": True, "content_commitment": True, "key_encipherment": True}, + [ExtendedKeyUsageOID.CLIENT_AUTH], + set(), + ), + ("fake.localmaeher.dev.pvarki.fi", {}, [ExtendedKeyUsageOID.SERVER_AUTH], set()), + ("kissa", {}, [ExtendedKeyUsageOID.SERVER_AUTH], {"server auth"}), + (None, {}, [ExtendedKeyUsageOID.SERVER_AUTH], {"server auth"}), + ( + "fake.localmaeher.dev.pvarki.fi", + {}, + [ExtendedKeyUsageOID.SERVER_AUTH, ExtendedKeyUsageOID.CODE_SIGNING], + {"server auth", "code signing"}, + ), + ("kissa", {"key_cert_sign": True}, None, {"cert sign"}), + ], +) +def test_disallowed_usages( + mint_csr: CsrFactory, + key_usage: KeyUsageFactory, + cn: str | None, + key_usage_flags: dict[str, bool], + ext_key_usage: list[ObjectIdentifier] | None, + expected: set[str], +) -> None: + """Client usages are open to all, server auth only to service CNs, everything else to nobody""" + csr = mint_csr(cn, key_usage(**key_usage_flags) if key_usage_flags else None, ext_key_usage) + assert disallowed_usages(csr, cn) == expected diff --git a/tests/test_csr_utils.py b/tests/test_csr_utils.py index abb4829e..3e21f395 100644 --- a/tests/test_csr_utils.py +++ b/tests/test_csr_utils.py @@ -3,7 +3,7 @@ from pathlib import Path import pytest -from libpvarki.mtlshelp.csr import async_create_client_csr, async_create_keypair +from libpvarki.mtlshelp.csr import async_create_client_csr, async_create_keypair, async_create_server_csr from rasenmaeher_api.web.api.utils.csr_utils import verify_csr @@ -24,3 +24,12 @@ async def test_verify_csr_subject(subject: dict[str, str], callsign: str, expect ckp = await async_create_keypair(tempdir / "user.key", tempdir / "user.pub") csrpem = await async_create_client_csr(ckp, tempdir / "user.csr", subject) assert verify_csr(csrpem, callsign) is expected + + +@pytest.mark.asyncio +async def test_verify_csr_rejects_server_csr(nice_tmpdir: str) -> None: + """A user callsign may not request server auth""" + tempdir = Path(nice_tmpdir) + ckp = await async_create_keypair(tempdir / "user.key", tempdir / "user.pub") + csrpem = await async_create_server_csr(ckp, tempdir / "user.csr", ["csr_test"]) + assert verify_csr(csrpem, "csr_test") is False From 5f87928726f929457b048692d7ecbefea6c4fc79 Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Wed, 28 Oct 2026 19:09:54 +0200 Subject: [PATCH 6/9] feat(certificates): reject unknown EKUs in CSRs --- src/rasenmaeher_api/cert/usages.py | 5 +++-- src/rasenmaeher_api/web/api/utils/csr_utils.py | 7 ++++++- tests/cert/cert_manager/test_private.py | 12 ++++++++++++ tests/cert/conftest.py | 7 ------- tests/cert/test_usages.py | 14 +++++++++++--- tests/conftest.py | 7 +++++++ tests/test_csr_utils.py | 10 ++++++++++ 7 files changed, 49 insertions(+), 13 deletions(-) diff --git a/src/rasenmaeher_api/cert/usages.py b/src/rasenmaeher_api/cert/usages.py index b6220c43..9693f306 100644 --- a/src/rasenmaeher_api/cert/usages.py +++ b/src/rasenmaeher_api/cert/usages.py @@ -82,8 +82,9 @@ def csr_usages(csr_pem: str) -> list[KeyUsages]: 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) + 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 diff --git a/src/rasenmaeher_api/web/api/utils/csr_utils.py b/src/rasenmaeher_api/web/api/utils/csr_utils.py index bacbb708..d9bdea30 100644 --- a/src/rasenmaeher_api/web/api/utils/csr_utils.py +++ b/src/rasenmaeher_api/web/api/utils/csr_utils.py @@ -23,7 +23,12 @@ def verify_csr(csrpem: str, callsign: str) -> bool: if len(csr.subject) != 1 or len(cns) != 1 or cns[0].value != callsign: LOGGER.warning(f"Callsign does not match CSR subject. DN={dn} callsign={callsign}") return False - if extra := disallowed_usages(csrpem, callsign): + try: + extra = disallowed_usages(csrpem, callsign) + except ValueError as exc: + LOGGER.warning(f"CSR usages could not be checked. error={exc} callsign={callsign}") + return False + if extra: LOGGER.warning(f"CSR requests usages not allowed for callsign. usages={sorted(extra)} callsign={callsign}") return False return True diff --git a/tests/cert/cert_manager/test_private.py b/tests/cert/cert_manager/test_private.py index 34b3f69c..a95ecee8 100644 --- a/tests/cert/cert_manager/test_private.py +++ b/tests/cert/cert_manager/test_private.py @@ -274,6 +274,18 @@ async def test_sign_csr_rejects_disallowed_usages( assert not fake_k8s.created +@pytest.mark.asyncio(loop_scope="function") +async def test_sign_csr_rejects_unknown_ext_key_usage( + fake_k8s: FakeK8s, cm_settings: RMSettings, mint_csr: CsrFactory +) -> None: + """An ExtendedKeyUsage OID without a name is refused before any CR is created""" + _ = cm_settings + csr = mint_csr("kissa", ext_key_usage=[ExtendedKeyUsageOID.CLIENT_AUTH, ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE]) + with pytest.raises(ValueError, match="Unsupported extended key usage"): + await private.sign_csr(csr, bundle=False) + assert not fake_k8s.created + + @pytest.mark.asyncio(loop_scope="function") async def test_sign_csr_conflict_polls_existing( fake_k8s: FakeK8s, cm_settings: RMSettings, csr_pem: str, monkeypatch: pytest.MonkeyPatch diff --git a/tests/cert/conftest.py b/tests/cert/conftest.py index a1b17329..cbd7126d 100644 --- a/tests/cert/conftest.py +++ b/tests/cert/conftest.py @@ -11,7 +11,6 @@ from .helpers import ( CA_PUBLIC_PATH, CertFactory, - CsrFactory, FakePKI, KeyUsageFactory, make_cert, @@ -74,12 +73,6 @@ def mint_cert() -> CertFactory: return make_cert -@pytest.fixture(scope="session") -def mint_csr() -> CsrFactory: - """Factory for CSRs with whatever subject and usage extensions a test needs""" - return make_csr - - @pytest.fixture(scope="session") def key_usage() -> KeyUsageFactory: """Helper for building KeyUsage extensions without spelling out all nine flags""" diff --git a/tests/cert/test_usages.py b/tests/cert/test_usages.py index 791e2c9b..696eb6b6 100644 --- a/tests/cert/test_usages.py +++ b/tests/cert/test_usages.py @@ -24,11 +24,19 @@ def test_key_usage_flags_mapped(mint_csr: CsrFactory, key_usage: KeyUsageFactory def test_ext_key_usage_oids_mapped(mint_csr: CsrFactory) -> None: - """Known ExtendedKeyUsage OIDs map to their names, unknown ones are dropped""" + """Known ExtendedKeyUsage OIDs map to their names""" known = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.SERVER_AUTH]) assert csr_usages(known) == ["server auth"] - unknown = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.KERBEROS_PKINIT_KDC]) - assert csr_usages(unknown) == [] + + +@pytest.mark.parametrize( + "unknown_oid", [ExtendedKeyUsageOID.KERBEROS_PKINIT_KDC, ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE] +) +def test_unknown_ext_key_usage_oid_raises(mint_csr: CsrFactory, unknown_oid: ObjectIdentifier) -> None: + """ExtendedKeyUsage OIDs without a name are refused, even next to allowed ones""" + csr = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.CLIENT_AUTH, unknown_oid]) + with pytest.raises(ValueError, match=unknown_oid.dotted_string): + csr_usages(csr) @pytest.mark.parametrize( diff --git a/tests/conftest.py b/tests/conftest.py index d423554b..95182bf6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,6 +26,7 @@ from rasenmaeher_api.productapihelpers import check_kraftwerk_manifest from rasenmaeher_api.rmsettings import switchme_to_singleton_call from rasenmaeher_api.testhelpers import create_test_users +from tests.cert.helpers import CsrFactory, make_csr # Register libadvian fixtures (nice_tmpdir, monkeysession, ...) as a pytest plugin # so they're available by name in tests without needing explicit imports. @@ -286,6 +287,12 @@ async def unauth_client_session(app_instance: FastAPI) -> AsyncGenerator[TestCli yield instance +@pytest.fixture(scope="session") +def mint_csr() -> CsrFactory: + """Factory for CSRs with whatever subject and usage extensions a test needs""" + return make_csr + + @pytest.fixture(scope="session") def issuer_cl() -> Issuer: """issuer using cl-key""" diff --git a/tests/test_csr_utils.py b/tests/test_csr_utils.py index 3e21f395..9441c1c5 100644 --- a/tests/test_csr_utils.py +++ b/tests/test_csr_utils.py @@ -3,9 +3,11 @@ from pathlib import Path import pytest +from cryptography.x509.oid import ExtendedKeyUsageOID from libpvarki.mtlshelp.csr import async_create_client_csr, async_create_keypair, async_create_server_csr from rasenmaeher_api.web.api.utils.csr_utils import verify_csr +from tests.cert.helpers import CsrFactory @pytest.mark.asyncio @@ -33,3 +35,11 @@ async def test_verify_csr_rejects_server_csr(nice_tmpdir: str) -> None: ckp = await async_create_keypair(tempdir / "user.key", tempdir / "user.pub") csrpem = await async_create_server_csr(ckp, tempdir / "user.csr", ["csr_test"]) assert verify_csr(csrpem, "csr_test") is False + + +def test_verify_csr_rejects_unknown_ext_key_usage(mint_csr: CsrFactory) -> None: + """An ExtendedKeyUsage OID without a name fails verification instead of raising""" + csrpem = mint_csr( + "csr_test", ext_key_usage=[ExtendedKeyUsageOID.CLIENT_AUTH, ExtendedKeyUsageOID.ANY_EXTENDED_KEY_USAGE] + ) + assert verify_csr(csrpem, "csr_test") is False From 4c1a98714b9c05679bc7df2d163821a54fe21a2a Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Wed, 28 Oct 2026 19:47:30 +0200 Subject: [PATCH 7/9] fix(build): add --no-index to use lockfile versions exactly --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 07daf52e..34d2aec7 100644 --- a/Dockerfile +++ b/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 From b3e42b8b5b8c21738e14aeed47f063fd87350363 Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:55:25 +0200 Subject: [PATCH 8/9] fix(enrollment): return 400 with reason for reserved callsigns on invite code --- src/rasenmaeher_api/db/enrollments.py | 4 ++-- src/rasenmaeher_api/db/errors.py | 6 +++--- src/rasenmaeher_api/db/people.py | 4 ++-- .../web/api/enrollment/views.py | 19 +++++++++++++++++-- tests/test_db.py | 6 ++++++ tests/test_enrollment.py | 8 ++++++++ 6 files changed, 38 insertions(+), 9 deletions(-) diff --git a/src/rasenmaeher_api/db/enrollments.py b/src/rasenmaeher_api/db/enrollments.py index a349c503..f293e632 100644 --- a/src/rasenmaeher_api/db/enrollments.py +++ b/src/rasenmaeher_api/db/enrollments.py @@ -282,9 +282,9 @@ async def create_for_callsign( ) -> "Enrollment": """Create a new one with random code for the callsign""" if RMSettings.singleton().is_reserved_cn(callsign): - raise CallsignReserved("Using service CNs as callsigns is forbidden") + raise CallsignReserved("Service CN used as callsign", detail="Callsign reserved") if csr and not verify_csr(csr, callsign): - raise CallsignReserved("CSR subject or key usages not allowed for 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) diff --git a/src/rasenmaeher_api/db/errors.py b/src/rasenmaeher_api/db/errors.py index 7643cd77..523f227f 100644 --- a/src/rasenmaeher_api/db/errors.py +++ b/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) diff --git a/src/rasenmaeher_api/db/people.py b/src/rasenmaeher_api/db/people.py index a30b0de3..1532ed3c 100644 --- a/src/rasenmaeher_api/db/people.py +++ b/src/rasenmaeher_api/db/people.py @@ -98,9 +98,9 @@ async def create_with_cert( """Create the cert etc and save the person""" cnf = RMSettings.singleton() if cnf.is_reserved_cn(callsign): - raise CallsignReserved("Using service CNs as callsigns is forbidden") + raise CallsignReserved("Service CN used as callsign", detail="Callsign reserved") if csrpem and not verify_csr(csrpem, callsign): - raise CallsignReserved("CSR subject or key usages not allowed for callsign") + raise CallsignReserved(detail="CSR subject or key usages not allowed for callsign") try: await Person.by_callsign(callsign) raise CallsignReserved() diff --git a/src/rasenmaeher_api/web/api/enrollment/views.py b/src/rasenmaeher_api/web/api/enrollment/views.py index 4117b4a2..64cf209e 100644 --- a/src/rasenmaeher_api/web/api/enrollment/views.py +++ b/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} diff --git a/tests/test_db.py b/tests/test_db.py index 04ff5036..cf40211f 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -503,3 +503,9 @@ async def test_record_issued_cert(dbinit_func) -> None: assert row is not None assert row.cn == common_name await record_issued_cert("not a pem at all") # must not raise + + +def test_callsignreserved_detail_is_explicit() -> None: + """Only the detail keyword reaches the client, positional messages stay internal""" + assert CallsignReserved("internal: db said no").detail == "Forbidden" + assert CallsignReserved("Service CN used as callsign", detail="Callsign reserved").detail == "Callsign reserved" diff --git a/tests/test_enrollment.py b/tests/test_enrollment.py index f9ae9df7..f171a2c9 100644 --- a/tests/test_enrollment.py +++ b/tests/test_enrollment.py @@ -595,6 +595,14 @@ async def test_enroll_with_invite_code( assert resp.status_code == 400 assert "taken" in resp_dict["detail"] + # ENROLL WITH INVITE CODE - SERVICE CN AS CALLSIGN + json_dict = {"invite_code": _inv_code, "callsign": "RasenMaeher"} + resp = await unauth_client_session.post("/api/v1/enrollment/invitecode/enroll", json=json_dict) + resp_dict = resp.json() + LOGGER.debug(resp_dict) + assert resp.status_code == 400 + assert resp_dict["detail"] == "Callsign reserved" + # ENROLL WITH INVITE CODE - CODE IS LOCKED json_dict = {"invite_code": _inv_code} resp = await tilauspalvelu_jwt_admin_client.put("/api/v1/enrollment/invitecode/deactivate", json=json_dict) From 5ca6f9c3c6fa8b732c2903b060ae85013cf40a5a Mon Sep 17 00:00:00 2001 From: Niilo Vuokila <59707802+niilovuo@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:59:57 +0200 Subject: [PATCH 9/9] fix(product): return 400 for CSRs rejected by usage policy --- .../cert/cert_manager/private.py | 2 +- src/rasenmaeher_api/web/api/product/views.py | 16 ++++-------- tests/cert/cert_manager/test_private.py | 2 +- tests/test_product_csr.py | 25 +++++++++++++++++++ 4 files changed, 32 insertions(+), 13 deletions(-) create mode 100644 tests/test_product_csr.py diff --git a/src/rasenmaeher_api/cert/cert_manager/private.py b/src/rasenmaeher_api/cert/cert_manager/private.py index 71535090..052b2f12 100644 --- a/src/rasenmaeher_api/cert/cert_manager/private.py +++ b/src/rasenmaeher_api/cert/cert_manager/private.py @@ -118,7 +118,7 @@ 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 CertManagerError(f"CSR for {cn} requests usages not allowed for it: {sorted(extra)}") + 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) diff --git a/src/rasenmaeher_api/web/api/product/views.py b/src/rasenmaeher_api/web/api/product/views.py index 9323d5fd..882ea923 100644 --- a/src/rasenmaeher_api/web/api/product/views.py +++ b/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))]) diff --git a/tests/cert/cert_manager/test_private.py b/tests/cert/cert_manager/test_private.py index a95ecee8..215ba1b0 100644 --- a/tests/cert/cert_manager/test_private.py +++ b/tests/cert/cert_manager/test_private.py @@ -269,7 +269,7 @@ async def test_sign_csr_rejects_disallowed_usages( """Usages outside the caller's allowed set are refused before any CR is created""" _ = cm_settings csr = mint_csr(cn, key_usage(**key_usage_flags) if key_usage_flags else None, ext_key_usage) - with pytest.raises(CertManagerError, match="not allowed"): + with pytest.raises(ValueError, match="not allowed"): await private.sign_csr(csr, bundle=False) assert not fake_k8s.created diff --git a/tests/test_product_csr.py b/tests/test_product_csr.py new file mode 100644 index 00000000..4b838f19 --- /dev/null +++ b/tests/test_product_csr.py @@ -0,0 +1,25 @@ +"""Test product CSR signing error mapping""" + +import pytest +from fastapi import HTTPException + +from rasenmaeher_api.web.api.product import views +from rasenmaeher_api.web.api.product.schema import CertificatesRequest + + +@pytest.mark.asyncio +async def test_csr_common_rejected_csr_is_400(monkeypatch: pytest.MonkeyPatch) -> None: + """A CSR the signer refuses is the caller's error, not a server error""" + + async def fake_get_ca() -> str: + return "ca" + + async def refuse(csr: str) -> str: + raise ValueError(f"CSR {csr} requests usages not allowed for it") + + monkeypatch.setattr(views, "get_ca", fake_get_ca) + monkeypatch.setattr(views, "sign_csr", refuse) + with pytest.raises(HTTPException) as excinfo: + await views.csr_common(CertificatesRequest(csr="x")) + assert excinfo.value.status_code == 400 + assert "not allowed" in str(excinfo.value.detail)