Merge pull request #7 from subspace/certificate-issues

Certificate management -> CertificateManager

auto_identity/__init__.py

@@ -16,11 +16,11 @@
     pem_to_public_key,
     load_public_key,
     save_key)
-from .certificate_management import create_csr, issue_certificate, self_issue_certificate, get_subject_common_name
+from .certificate_manager import CertificateManager
 from .registry import Registry
 from .utils import der_encode_signature_algorithm_oid
 
-__version__ = '0.1.2'
+__version__ = '0.1.3'
 
 __all__ = [
     "generate_rsa_key_pair",
@@ -32,10 +32,7 @@
     "pem_to_private_key",
     "pem_to_public_key",
     "key_to_pem",
-    "create_csr",
-    "issue_certificate",
-    "self_issue_certificate",
-    "get_subject_common_name",
+    "CertificateManager",
     "der_encode_signature_algorithm_oid",
     "Registry",
     "Keypair",

auto_identity/certificate_manager.py

@@ -0,0 +1,159 @@
+from datetime import datetime, timedelta, timezone
+from cryptography import x509
+from cryptography.x509.oid import NameOID
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import ed25519, rsa
+from .key_management import do_public_keys_match
+
+
+class CertificateManager:
+    """
+    Certificate management class.
+    """
+
+    def __init__(self,  certificate=None, private_key=None):
+        """
+        Initializes a certificate issuer.
+
+        Args:
+            certificate(Certificate): Certificate.
+            private_key(PrivateKey): Private key.
+        """
+        self.private_key = private_key
+        self.certificate = certificate
+
+    @staticmethod
+    def _to_common_name(subject_name):
+        """
+        Converts a subject name to a common name.
+
+        Args:
+            subject_name(str): Subject name for the certificate(common name).
+
+        Returns:
+            str: Common name.
+        """
+        return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, subject_name)])
+
+    def _prepare_signing_params(self):
+        """
+        Prepares the signing parameters based on the key type.
+
+        Returns:
+            dict: Signing parameters.
+        """
+        private_key = self.private_key
+        if isinstance(private_key, ed25519.Ed25519PrivateKey):
+            return {"private_key": private_key, "algorithm": None}
+        if isinstance(private_key, rsa.RSAPrivateKey):
+            return {"private_key": private_key, "algorithm": hashes.SHA256()}
+
+        raise ValueError("Unsupported key type for signing.")
+
+    def create_csr(self, subject_name):
+        """
+        Creates a Certificate Signing Request(CSR).
+
+        Args:
+            subject_name(str): Subject name for the CSR(common name).
+
+        Returns:
+            CertificateSigningRequest: Created X.509 CertificateSigningRequest.
+        """
+        signing_params = self._prepare_signing_params()
+        csr = x509.CertificateSigningRequestBuilder().subject_name(
+            x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, subject_name)])
+        ).sign(**signing_params)
+
+        return csr
+
+    def issue_certificate(self, csr, validity_period_days=365):
+        """
+        Issues a certificate for Certificate Signing Request(CSR).
+
+        Args:
+            csr(CertificateSigningRequest): Certificate Signing Request.
+            issuer_certificate(Certificate): Issuer certificate.
+            issuer_private_key(PrivateKey): Private key to sign the certificate with .
+            validity_period_days(int, optional): Number of days the certificate is valid. Defaults to 365.
+
+        Returns:
+            Certificate: Created X.509 certificate.
+        """
+        if self.certificate is None:
+            raise ValueError("Certificate is not set.")
+        if self.private_key is None:
+            raise ValueError("Private key is not set.")
+
+        issuer_certificate = self.certificate
+
+        # Verify that the public key derived from the private key matches the one in the issuer's certificate
+        if not do_public_keys_match(issuer_certificate.public_key(), self.private_key.public_key()):
+            raise ValueError(
+                "Issuer certificate public key does not match the private key used for signing.")
+
+        signing_params = self._prepare_signing_params()
+
+        certificate = x509.CertificateBuilder().subject_name(
+            csr.subject,
+        ).issuer_name(
+            issuer_certificate.subject,
+        ).public_key(
+            csr.public_key(),
+        ).serial_number(
+            x509.random_serial_number(),
+        ).not_valid_before(
+            datetime.now(timezone.utc),
+        ).not_valid_after(
+            datetime.now(timezone.utc) + timedelta(days=validity_period_days),
+        ).sign(**signing_params)
+
+        return certificate
+
+    def self_issue_certificate(self, subject_name: str, validity_period_days=365):
+        """
+        Issues a self-signed certificate for the identity.
+
+        Args:
+            subject_name(str): Subject name for the certificate(common name).
+            private_key(PrivateKey): Private key to sign the certificate with .
+            validity_period_days(int, optional): Number of days the certificate is valid. Defaults to 365.
+
+        Returns:
+            Certificate: Created X.509 certificate.
+        """
+        if self.private_key is None:
+            raise ValueError("Private key is not set.")
+
+        signing_params = self._prepare_signing_params()
+        common_name = self._to_common_name(subject_name)
+
+        certificate = x509.CertificateBuilder().subject_name(
+            common_name,
+        ).issuer_name(
+            common_name,
+        ).public_key(
+            self.private_key.public_key(),
+        ).serial_number(
+            x509.random_serial_number(),
+        ).not_valid_before(
+            datetime.now(timezone.utc),
+        ).not_valid_after(
+            datetime.now(timezone.utc) + timedelta(days=validity_period_days),
+        ).sign(**signing_params)
+
+        self.certificate = certificate
+        return certificate
+
+    @staticmethod
+    def get_subject_common_name(certificate: x509.Certificate):
+        """
+        Retrieves the common name from the subject of the certificate.
+
+        Args:
+            certificate(x509.Certificate): Certificate to retrieve the common name from .
+
+        Returns:
+            str: Common name of the certificate.
+        """
+        return certificate.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value

auto_identity/key_management.py

@@ -41,6 +41,9 @@ def key_to_pem(key, password: str = None) -> str:
     Args:
         key: The key to convert (private or public).
         password (str): The password used to encrypt the key.
+
+    Returns:
+            str: The PEM string representation of the key.
     """
     if hasattr(key, 'private_bytes'):
         encoding = serialization.Encoding.PEM
@@ -79,7 +82,13 @@ def pem_to_private_key(pem_data: str, password: str = None):
     Args:
         pem_data (str): The PEM string to convert.
         password (str): The password used to encrypt the key.
+
+    Returns:
+        The private key.
     """
+    # Ensure pem_data is a bytes object
+    if isinstance(pem_data, str):
+        pem_data = pem_data.encode()
 
     private_key = serialization.load_pem_private_key(
         pem_data,
@@ -111,6 +120,9 @@ def pem_to_public_key(pem_data: str):
 
     Args:
         pem_data (str): The PEM string to convert.
+
+    Returns:
+        The public key.
     """
     public_key = serialization.load_pem_public_key(
         pem_data,
@@ -160,3 +172,27 @@ def key_to_hex(key) -> str:
         raise ValueError("Unsupported key type")
 
     return key_data.hex()
+
+
+def do_public_keys_match(public_key_1, public_key_2):
+    """
+    Checks if two public keys match.
+
+    Args:
+        public_key_1: The first public key.
+        public_key_2: The second public key.
+
+    Returns:
+            bool: True if the keys match, False otherwise.
+    """
+    # Serialize public keys to DER format for comparison
+    public_key_1_der = public_key_1.public_bytes(
+        encoding=serialization.Encoding.DER,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo
+    )
+
+    public_key_2_der = public_key_2.public_bytes(
+        encoding=serialization.Encoding.DER,
+        format=serialization.PublicFormat.SubjectPublicKeyInfo
+    )
+    return public_key_1_der == public_key_2_der

requirements.txt

@@ -2,4 +2,5 @@ cryptography==42.0.5
 substrate-interface==1.7.7
 pyasn1==0.6.0
 python-dotenv==1.0.1
-wheel==0.43.0
+wheel==0.43.0
+pytest==8.1.1

setup.py

@@ -2,7 +2,7 @@
 
 setup(
     name='auto-sdk',
-    version='0.1.2',
+    version='0.1.3',
     author='Autonomys',
     author_email='[email protected]',
     url='https://github.com/subspace/auto-kit',