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/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 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/src/rasenmaeher_api/cert/cert_manager/private.py b/src/rasenmaeher_api/cert/cert_manager/private.py index eecf73cb..052b2f12 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,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) diff --git a/src/rasenmaeher_api/cert/usages.py b/src/rasenmaeher_api/cert/usages.py new file mode 100644 index 00000000..9693f306 --- /dev/null +++ b/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 diff --git a/src/rasenmaeher_api/db/enrollments.py b/src/rasenmaeher_api/db/enrollments.py index a07b02c4..f293e632 100644 --- a/src/rasenmaeher_api/db/enrollments.py +++ b/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) 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 086c3ffc..1532ed3c 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""" - 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() 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/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/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/src/rasenmaeher_api/web/api/utils/csr_utils.py b/src/rasenmaeher_api/web/api/utils/csr_utils.py index 836ff504..d9bdea30 100644 --- a/src/rasenmaeher_api/web/api/utils/csr_utils.py +++ b/src/rasenmaeher_api/web/api/utils/csr_utils.py @@ -3,8 +3,11 @@ import logging from cryptography import x509 +from cryptography.x509.oid import NameOID from libadvian.binpackers import ensure_utf8 +from ....cert.usages import disallowed_usages + LOGGER = logging.getLogger(__name__) @@ -12,14 +15,20 @@ 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}") - 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 - # crypto.X509Extension(b"keyUsage", True, b"digitalSignature,nonRepudiation,keyEncipherment"), - # crypto.X509Extension(b"extendedKeyUsage", True, b"clientAuth"), + 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 6ca1d64f..215ba1b0 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: @@ -228,6 +204,88 @@ 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(ValueError, 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_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 new file mode 100644 index 00000000..696eb6b6 --- /dev/null +++ b/tests/cert/test_usages.py @@ -0,0 +1,74 @@ +"""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""" + known = mint_csr("example", ext_key_usage=[ExtendedKeyUsageOID.SERVER_AUTH]) + assert csr_usages(known) == ["server auth"] + + +@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( + ("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/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 new file mode 100644 index 00000000..9441c1c5 --- /dev/null +++ b/tests/test_csr_utils.py @@ -0,0 +1,45 @@ +"""Test CSR subject verification""" + +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 +@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 + + +@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 + + +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 diff --git a/tests/test_db.py b/tests/test_db.py index 4676eb24..cf40211f 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: @@ -491,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) 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) diff --git a/tests/test_rasenmaeher_api.py b/tests/test_rasenmaeher_api.py index 62b6aa1c..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") @@ -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") 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" },