Refactor CSR signing to include explicit role parameter

Updated the `sign_csr` method to accept an explicit `role` parameter, replacing the previous heuristic using `is_server`. This improves clarity and flexibility, ensuring correct certificate attributes based on the provided role ("server" or "client"). Also updated related invocations and adjusted the builder logic accordingly.

Author: YeisonCardona

chaski/ca.py

@@ -91,8 +91,8 @@ async def sign_csr(
             f"{self.name}: Starting the signing process for CSR data for node '{node_id}'"
         )
         # Sign the provided Certificate Signing Request (CSR) using the Certificate Authority (CA)
-        signed_csr_client = self.ca.sign_csr(csr_data_client)
-        signed_csr_server = self.ca.sign_csr(csr_data_server)
+        signed_csr_client = self.ca.sign_csr(csr_data_client, role="client")
+        signed_csr_server = self.ca.sign_csr(csr_data_server, role="server")
 
         # Prepare the dictionary containing the signed certificates for client and server,
         # and the path to the CA certificate. This dictionary will be returned as the result

chaski/utils/certificate_authority.py

@@ -2,6 +2,7 @@
 import ssl
 import datetime
 from ipaddress import ip_address
+from typing import Literal, Optional
 
 # Importing cryptography modules for X509 certificates, private key generation,
 # and serialization, including RSA key generation and PEM encoding without encryption.
@@ -241,7 +242,9 @@ def ca_certificate_path(self, path: str) -> None:
         """
         self.ca_cert_path_ = path
 
-    def sign_csr(self, csr_data: bytes, is_server: bool | None = None) -> bytes:
+    def sign_csr(
+        self, csr_data: bytes, role: Optional[Literal["server", "client"]] = None
+    ) -> bytes:
         """
         Sign a Certificate Signing Request (CSR) with the Certificate Authority (CA) key.
 
@@ -252,9 +255,9 @@ def sign_csr(self, csr_data: bytes, is_server: bool | None = None) -> bytes:
         ----------
         csr_data : bytes
             The certificate signing request data in PEM format.
-        is_server : bool, optional
-            If True, indicates this is a server certificate. If False, indicates this is a client certificate.
-            If None, will be inferred from the Common Name (CN) in the CSR by checking if it contains "server".
+        role : Optional[Literal["server", "client"]], optional
+            The role of the certificate, either "server" or "client". If not provided,
+            it will be inferred from the Common Name in the CSR.
 
         Returns
         -------
@@ -282,10 +285,15 @@ def sign_csr(self, csr_data: bytes, is_server: bool | None = None) -> bytes:
         # Load client CSR
         csr = x509.load_pem_x509_csr(csr_data)
 
-        # infer by CN if not provided
-        if is_server is None:
-            cn = csr.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
-            is_server = "server" in cn.lower()
+        # Heuristic if role not explicitly provided
+        if role is None:
+            try:
+                cn = csr.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[
+                    0
+                ].value.lower()
+                role = "server" if "server" in cn else "client"
+            except Exception:
+                role = "client"
 
         ku = x509.KeyUsage(
             digital_signature=True,
@@ -300,12 +308,12 @@ def sign_csr(self, csr_data: bytes, is_server: bool | None = None) -> bytes:
         )
         eku = x509.ExtendedKeyUsage(
             [ExtendedKeyUsageOID.SERVER_AUTH]
-            if is_server
+            if role == "server"
             else [ExtendedKeyUsageOID.CLIENT_AUTH]
         )
 
         # Generate client certificate
-        certificate = (
+        builder = (
             x509.CertificateBuilder()
             # Set the subject name for the certificate to the subject specified in the CSR (Certificate Signing Request)
             .subject_name(csr.subject)
@@ -332,16 +340,6 @@ def sign_csr(self, csr_data: bytes, is_server: bool | None = None) -> bytes:
             # Add the subject alternative name extension with the IP address
             .add_extension(ku, critical=True)
             .add_extension(eku, critical=False)
-            .add_extension(
-                (
-                    x509.SubjectAlternativeName(
-                        [x509.IPAddress(ip_address(self.ip_address))]
-                    )
-                    if is_server
-                    else x509.SubjectAlternativeName([])
-                ),  # no SAN-IP en client por defecto
-                critical=False,
-            )
             .add_extension(
                 x509.SubjectKeyIdentifier.from_public_key(csr.public_key()),
                 critical=False,
@@ -350,10 +348,19 @@ def sign_csr(self, csr_data: bytes, is_server: bool | None = None) -> bytes:
                 x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()),
                 critical=False,
             )
-            # Sign the certificate using the CA's private key with SHA256 as the hashing algorithm
-            .sign(ca_key, hashes.SHA256())
         )
 
+        if role == "server":
+            builder = builder.add_extension(
+                x509.SubjectAlternativeName(
+                    [x509.IPAddress(ip_address(self.ip_address))]
+                ),
+                critical=False,
+            )
+
+        # Sign the certificate using the CA's private key with SHA256 as the hashing algorithm
+        certificate = builder.sign(ca_key, hashes.SHA256())
+
         return certificate.public_bytes(serialization.Encoding.PEM)
 
     def _key_and_csr(self, name="client") -> tuple:
@@ -683,7 +690,7 @@ def get_context(self) -> tuple[ssl.SSLContext, ssl.SSLContext]:
 
         return ssl_context_client, ssl_context_server
 
-    def sign_and_store(self, name: str) -> str:
+    def sign_and_store(self, name: Literal["server", "client"]) -> str:
         """
         Sign a Certificate Signing Request (CSR) and store the resulting certificate.
 
@@ -709,6 +716,7 @@ def sign_and_store(self, name: str) -> str:
         """
         csr_path = self.certificate_paths[name]
         cert_path = self.certificate_signed_paths[name]
-        cert_pem = self.sign_csr(self.load_certificate(csr_path))
+        cert_pem = self.sign_csr(self.load_certificate(csr_path), role=name)
         self.write_certificate(cert_path, cert_pem)
+        os.chmod(cert_path, 0o644)
         return cert_path

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "chaski-confluent"
-version = "0.1a6"
+version = "0.1a8"
 description = "Chaski Confluent"
 readme = "README.md"
 requires-python = ">=3.10"

test/test_certificate_authority.py

@@ -271,10 +271,18 @@ async def test_sign(self, certificate_authority: CertificateAuthority) -> None:
         )
 
         with open(ca.certificate_signed_paths["client"], "wb") as file:
-            file.write(ca.sign_csr(ca.load_certificate(ca.certificate_paths["client"])))
+            file.write(
+                ca.sign_csr(
+                    ca.load_certificate(ca.certificate_paths["client"]), role="client"
+                )
+            )
 
         with open(ca.certificate_signed_paths["server"], "wb") as file:
-            file.write(ca.sign_csr(ca.load_certificate(ca.certificate_paths["server"])))
+            file.write(
+                ca.sign_csr(
+                    ca.load_certificate(ca.certificate_paths["server"]), role="server"
+                )
+            )
 
         ca_cert: Certificate = x509.load_pem_x509_certificate(
             ca.load_certificate(ca.ca_certificate_path), default_backend()