sicceggetools/acme/certificate.py

   1 #!/usr/bin/python
   2
   3 import logging
   4 import os
   5 import os.path
   6
   7 from cryptography import x509
   8 from cryptography.hazmat.backends import default_backend
   9 from cryptography.hazmat.primitives import hashes
  10 from cryptography.hazmat.primitives import serialization
  11 from cryptography.x509.oid import NameOID
  12
  13 import OpenSSL
  14
  15 class Certificate:
  16     def __init__(self, servicetype, name, sans):
  17         self._name = name
  18         self._sans = sans
  19         self._servicetype = servicetype
  20         self._basename = os.path.join("certs", servicetype, name)
  21         if os.path.exists(os.path.join(self._basename, "key.pem")):
  22             self._from_private_key()
  23         elif os.path.exists(os.path.join(self._basename, "csr.pem")):
  24             self._from_csr()
  25         else:
  26             self._from_scratch()
  27
  28
  29     def _from_private_key(self):
  30         with open(os.path.join(self._basename, "key.pem"), "rb") as keyfd:
  31             private_key = serialization.load_pem_private_key(
  32                 keyfd.read(),
  33                 password=None,
  34                 backend=default_backend())
  35
  36             builder = x509.CertificateSigningRequestBuilder()
  37             builder = builder.subject_name(x509.Name([
  38                 x509.NameAttribute(NameOID.COMMON_NAME, self._name.decode()),
  39                 ]))
  40             builder = builder.add_extension(
  41                 x509.SubjectAlternativeName([x509.DNSName(x.decode()) for x in self._sans]),
  42                 critical=False)
  43
  44             request = builder.sign(private_key, hashes.SHA512(), default_backend())
  45             self._requeststring = request.public_bytes(serialization.Encoding.PEM)
  46
  47
  48     def _from_csr(self):
  49         if os.path.exists(os.path.join(self._basename, "csr.pem")):
  50             with open(os.path.join(self._basename, "csr.pem"), "rb") as csrfd:
  51                 self._requeststring = csrfd.read()
  52
  53
  54     def _from_scratch(self):
  55         raise NotImplementedError("Key generation is currently not implemented")
  56
  57
  58     def asString(self):
  59         return self._requeststring
  60
  61
  62     def save(self, certificate, chain):
  63         with open(os.path.join(self._basename, "cert.pem"), "wb") as certfd:
  64             certfd.write(certificate.body._dump(OpenSSL.crypto.FILETYPE_PEM))
  65             for cert in chain:
  66                 certfd.write(cert._dump(OpenSSL.crypto.FILETYPE_PEM))
  67