Improve CSR processing and logging, make connection verification optional

nodeman/internal_ca.py

@@ -18,11 +18,9 @@
     CertificateInformation,
     PrivateKey,
     get_hash_algorithm_from_key,
-    verify_x509_csr_signature,
+    verify_x509_csr,
 )
 
-logger = logging.getLogger(__name__)
-
 
 class InternalCertificateAuthority(CertificateAuthorityClient):
     """Internal CA"""
@@ -55,6 +53,7 @@ def __init__(
         validity: timedelta | None = None,
         time_skew: timedelta | None = None,
     ):
+        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)
         self.issuer_ca_certificate = issuer_ca_certificate
         self.issuer_ca_private_key = issuer_ca_private_key
         self.root_ca_certificate = root_ca_certificate or issuer_ca_certificate
@@ -102,9 +101,9 @@ def ca_fingerprint(self) -> str:
     def sign_csr(self, csr: x509.CertificateSigningRequest, name: str) -> CertificateInformation:
         """Sign CSR with CA private key"""
 
-        logger.debug("Processing CSR from %s", name)
+        self.logger.debug("Processing CSR from %s", name)
 
-        verify_x509_csr_signature(csr=csr, name=name)
+        verify_x509_csr(csr=csr, name=name, validate_name=False)
 
         now = datetime.now(tz=timezone.utc)
 

nodeman/server.py

@@ -109,6 +109,7 @@ def get_step_client(settings: StepSettings) -> StepClient:
             ca_fingerprint=ca_fingerprint,
             provisioner_name=settings.provisioner_name,
             provisioner_jwk=provisioner_jwk,
+            ca_server_verify=settings.ca_server_verify,
         )
         logger.info("Connected to StepCA %s (%s)", res.ca_url, ca_fingerprint)
         return res

nodeman/settings.py

@@ -26,6 +26,7 @@ class MongoDB(BaseModel):
 
 class StepSettings(BaseModel):
     ca_url: AnyHttpUrl
+    ca_server_verify: bool = True
     ca_fingerprint: str | None = None
     ca_fingerprint_file: FilePath | None = None
     provisioner_name: str

nodeman/step.py

@@ -1,4 +1,5 @@
 import atexit
+import logging
 import os
 import tempfile
 import time
@@ -12,28 +13,44 @@
 from jwcrypto.jwt import JWT
 
 from .jose import jwk_to_alg
-from .x509 import CertificateAuthorityClient, CertificateInformation
+from .x509 import CertificateAuthorityClient, CertificateInformation, verify_x509_csr
 
 
 class StepClient(CertificateAuthorityClient):
-    def __init__(self, ca_url: str, ca_fingerprint: str, provisioner_name: str, provisioner_jwk: JWK):
+    def __init__(
+        self,
+        ca_url: str,
+        ca_fingerprint: str,
+        provisioner_name: str,
+        provisioner_jwk: JWK,
+        ca_server_verify: bool = True,
+    ):
+        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)
         self.ca_url = ca_url
         self.ca_fingerprint = ca_fingerprint
         self.provisioner_name = provisioner_name
         self.provisioner_jwk = provisioner_jwk
         self.ca_bundle_filename = self._get_root_ca_cert()
         self.token_ttl = 300
+        self.verify = self.ca_bundle_filename if ca_server_verify else False
 
     def sign_csr(self, csr: x509.CertificateSigningRequest, name: str) -> CertificateInformation:
+        """Sign CSR with Step CA"""
+
+        self.logger.debug("Processing CSR from %s", name)
+
+        verify_x509_csr(csr=csr, name=name, validate_name=True)
+
         csr_pem = csr.public_bytes(encoding=serialization.Encoding.PEM).decode()
         token = self._get_token(name)
         response = httpx.post(
             urljoin(self.ca_url, "1.0/sign"),
-            verify=self.ca_bundle_filename,
+            verify=self.verify,
             json={"csr": csr_pem, "ott": token},
         )
         response.raise_for_status()
         payload = response.json()
+
         return CertificateInformation(
             cert_chain=[x509.load_pem_x509_certificate(cert.encode()) for cert in payload["certChain"]],
             ca_cert=x509.load_pem_x509_certificate(payload["ca"].encode()),

nodeman/x509.py

@@ -85,7 +85,14 @@ class SubjectAlternativeNameMismatchError(CertificateSigningRequestException):
     pass
 
 
-def verify_x509_csr_data(csr: x509.CertificateSigningRequest, name: str) -> None:
+def verify_x509_csr(csr: x509.CertificateSigningRequest, name: str, validate_name: bool = True) -> None:
+    """Verify X.509 CSR"""
+
+    verify_x509_csr_signature(csr=csr, name=name)
+    verify_x509_csr_data(csr=csr, name=name, validate_name=validate_name)
+
+
+def verify_x509_csr_data(csr: x509.CertificateSigningRequest, name: str, validate_name: bool = True) -> None:
     """Verify X.509 CSR against name"""
 
     # ensure Subject is correct

tests/test_internal_ca.py

@@ -11,14 +11,7 @@
 from cryptography.x509.oid import NameOID
 
 from nodeman.internal_ca import InternalCertificateAuthority
-from nodeman.x509 import (
-    RSA_EXPONENT,
-    CertificateInformation,
-    PrivateKey,
-    generate_similar_key,
-    generate_x509_csr,
-    verify_x509_csr_data,
-)
+from nodeman.x509 import RSA_EXPONENT, CertificateInformation, PrivateKey, generate_similar_key, generate_x509_csr
 from tests.utils import generate_ca_certificate
 
 
@@ -50,8 +43,6 @@ def _test_internal_ca(ca_private_key: PrivateKey, verify: bool = True) -> None:
     name = "hostname.example.com"
     csr = generate_x509_csr(key=key, name=name)
 
-    verify_x509_csr_data(name=name, csr=csr)
-
     res = ca_client.sign_csr(csr, name)
 
     # Assert that the certificate chain is not empty

tests/test_step_ca.py

@@ -7,7 +7,7 @@
 
 from nodeman.settings import StepSettings
 from nodeman.step import StepClient
-from nodeman.x509 import generate_x509_csr, verify_x509_csr_data
+from nodeman.x509 import generate_x509_csr
 
 
 def test_step_ca() -> None:
@@ -46,7 +46,5 @@ def test_step_ca() -> None:
     key = ec.generate_private_key(ec.SECP256R1())
     csr = generate_x509_csr(key=key, name=name)
 
-    verify_x509_csr_data(name=name, csr=csr)
-
     res = ca_client.sign_csr(csr, name)
     print(res)