mirror of
https://github.com/apache/httpd.git
synced 2025-08-06 11:06:17 +00:00
af10058840
- When the server starts, it looks for new, staged certificates to
activate. If the staged set of files in 'md/staging/<domain>' is messed
up, this could prevent further renewals to happen. Now, when the staging
set is present, but could not be activated due to an error, purge the
whole directory. [icing]
- Fix certificate retrieval on ACME renewal to not require a 'Location:'
header returned by the ACME CA. This was the way it was done in ACME
before it became an IETF standard. Let's Encrypt still supports this,
but other CAs do not. [icing]
- Restore compatibility with OpenSSL < 1.1. [ylavic]
git-svn-id: https://svn.apache.org/repos/asf/httpd/httpd/trunk@1920747 13f79535-47bb-0310-9956-ffa450edef68
514 lines
20 KiB
Python
514 lines
20 KiB
Python
import os
|
|
import re
|
|
from datetime import timedelta, datetime
|
|
from typing import List, Any, Optional
|
|
|
|
from cryptography import x509
|
|
from cryptography.hazmat.backends import default_backend
|
|
from cryptography.hazmat.primitives import hashes
|
|
from cryptography.hazmat.primitives.asymmetric import ec, rsa
|
|
from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
|
|
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
|
|
from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, NoEncryption, load_pem_private_key
|
|
from cryptography.x509 import ExtendedKeyUsageOID, NameOID
|
|
|
|
|
|
EC_SUPPORTED = {}
|
|
EC_SUPPORTED.update([(curve.name.upper(), curve) for curve in [
|
|
ec.SECP192R1,
|
|
ec.SECP224R1,
|
|
ec.SECP256R1,
|
|
ec.SECP384R1,
|
|
]])
|
|
|
|
|
|
def _private_key(key_type):
|
|
if isinstance(key_type, str):
|
|
key_type = key_type.upper()
|
|
m = re.match(r'^(RSA)?(\d+)$', key_type)
|
|
if m:
|
|
key_type = int(m.group(2))
|
|
|
|
if isinstance(key_type, int):
|
|
return rsa.generate_private_key(
|
|
public_exponent=65537,
|
|
key_size=key_type,
|
|
backend=default_backend()
|
|
)
|
|
if not isinstance(key_type, ec.EllipticCurve) and key_type in EC_SUPPORTED:
|
|
key_type = EC_SUPPORTED[key_type]
|
|
return ec.generate_private_key(
|
|
curve=key_type,
|
|
backend=default_backend()
|
|
)
|
|
|
|
|
|
class CertificateSpec:
|
|
|
|
def __init__(self, name: str = None, domains: List[str] = None,
|
|
email: str = None,
|
|
key_type: str = None, single_file: bool = False,
|
|
valid_from: timedelta = timedelta(days=-1),
|
|
valid_to: timedelta = timedelta(days=89),
|
|
client: bool = False,
|
|
sub_specs: List['CertificateSpec'] = None):
|
|
self._name = name
|
|
self.domains = domains
|
|
self.client = client
|
|
self.email = email
|
|
self.key_type = key_type
|
|
self.single_file = single_file
|
|
self.valid_from = valid_from
|
|
self.valid_to = valid_to
|
|
self.sub_specs = sub_specs
|
|
|
|
@property
|
|
def name(self) -> Optional[str]:
|
|
if self._name:
|
|
return self._name
|
|
elif self.domains:
|
|
return self.domains[0]
|
|
return None
|
|
|
|
@property
|
|
def type(self) -> Optional[str]:
|
|
if self.domains and len(self.domains):
|
|
return "server"
|
|
elif self.client:
|
|
return "client"
|
|
elif self.name:
|
|
return "ca"
|
|
return None
|
|
|
|
|
|
class Credentials:
|
|
|
|
def __init__(self, name: str, cert: Any, pkey: Any, issuer: 'Credentials' = None):
|
|
self._name = name
|
|
self._cert = cert
|
|
self._pkey = pkey
|
|
self._issuer = issuer
|
|
self._cert_file = None
|
|
self._pkey_file = None
|
|
self._store = None
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@property
|
|
def subject(self) -> x509.Name:
|
|
return self._cert.subject
|
|
|
|
@property
|
|
def key_type(self):
|
|
if isinstance(self._pkey, RSAPrivateKey):
|
|
return f"rsa{self._pkey.key_size}"
|
|
elif isinstance(self._pkey, EllipticCurvePrivateKey):
|
|
return f"{self._pkey.curve.name}"
|
|
else:
|
|
raise Exception(f"unknown key type: {self._pkey}")
|
|
|
|
@property
|
|
def private_key(self) -> Any:
|
|
return self._pkey
|
|
|
|
@property
|
|
def certificate(self) -> Any:
|
|
return self._cert
|
|
|
|
@property
|
|
def cert_pem(self) -> bytes:
|
|
return self._cert.public_bytes(Encoding.PEM)
|
|
|
|
@property
|
|
def pkey_pem(self) -> bytes:
|
|
return self._pkey.private_bytes(
|
|
Encoding.PEM,
|
|
PrivateFormat.TraditionalOpenSSL if self.key_type.startswith('rsa') else PrivateFormat.PKCS8,
|
|
NoEncryption())
|
|
|
|
@property
|
|
def issuer(self) -> Optional['Credentials']:
|
|
return self._issuer
|
|
|
|
def set_store(self, store: 'CertStore'):
|
|
self._store = store
|
|
|
|
def set_files(self, cert_file: str, pkey_file: str = None):
|
|
self._cert_file = cert_file
|
|
self._pkey_file = pkey_file
|
|
|
|
@property
|
|
def cert_file(self) -> str:
|
|
return self._cert_file
|
|
|
|
@property
|
|
def pkey_file(self) -> Optional[str]:
|
|
return self._pkey_file
|
|
|
|
def get_first(self, name) -> Optional['Credentials']:
|
|
creds = self._store.get_credentials_for_name(name) if self._store else []
|
|
return creds[0] if len(creds) else None
|
|
|
|
def get_credentials_for_name(self, name) -> List['Credentials']:
|
|
return self._store.get_credentials_for_name(name) if self._store else []
|
|
|
|
def issue_certs(self, specs: List[CertificateSpec],
|
|
chain: List['Credentials'] = None) -> List['Credentials']:
|
|
return [self.issue_cert(spec=spec, chain=chain) for spec in specs]
|
|
|
|
def issue_cert(self, spec: CertificateSpec, chain: List['Credentials'] = None) -> 'Credentials':
|
|
key_type = spec.key_type if spec.key_type else self.key_type
|
|
creds = None
|
|
if self._store:
|
|
creds = self._store.load_credentials(
|
|
name=spec.name, key_type=key_type, single_file=spec.single_file, issuer=self)
|
|
if creds is None:
|
|
creds = HttpdTestCA.create_credentials(spec=spec, issuer=self, key_type=key_type,
|
|
valid_from=spec.valid_from, valid_to=spec.valid_to)
|
|
if self._store:
|
|
self._store.save(creds, single_file=spec.single_file)
|
|
if spec.type == "ca":
|
|
self._store.save_chain(creds, "ca", with_root=True)
|
|
|
|
if spec.sub_specs:
|
|
if self._store:
|
|
sub_store = CertStore(fpath=os.path.join(self._store.path, creds.name))
|
|
creds.set_store(sub_store)
|
|
subchain = chain.copy() if chain else []
|
|
subchain.append(self)
|
|
creds.issue_certs(spec.sub_specs, chain=subchain)
|
|
return creds
|
|
|
|
def save_cert_pem(self, fpath):
|
|
with open(fpath, "wb") as fd:
|
|
fd.write(self.cert_pem)
|
|
|
|
def save_pkey_pem(self, fpath):
|
|
with open(fpath, "wb") as fd:
|
|
fd.write(self.pkey_pem)
|
|
|
|
|
|
class CertStore:
|
|
|
|
def __init__(self, fpath: str):
|
|
self._store_dir = fpath
|
|
if not os.path.exists(self._store_dir):
|
|
os.makedirs(self._store_dir)
|
|
self._creds_by_name = {}
|
|
|
|
@property
|
|
def path(self) -> str:
|
|
return self._store_dir
|
|
|
|
def save(self, creds: Credentials, name: str = None,
|
|
chain: List[Credentials] = None,
|
|
single_file: bool = False) -> None:
|
|
name = name if name is not None else creds.name
|
|
cert_file = self.get_cert_file(name=name, key_type=creds.key_type)
|
|
pkey_file = self.get_pkey_file(name=name, key_type=creds.key_type)
|
|
if single_file:
|
|
pkey_file = None
|
|
with open(cert_file, "wb") as fd:
|
|
fd.write(creds.cert_pem)
|
|
if chain:
|
|
for c in chain:
|
|
fd.write(c.cert_pem)
|
|
if pkey_file is None:
|
|
fd.write(creds.pkey_pem)
|
|
if pkey_file is not None:
|
|
with open(pkey_file, "wb") as fd:
|
|
fd.write(creds.pkey_pem)
|
|
creds.set_files(cert_file, pkey_file)
|
|
self._add_credentials(name, creds)
|
|
|
|
def save_chain(self, creds: Credentials, infix: str, with_root=False):
|
|
name = creds.name
|
|
chain = [creds]
|
|
while creds.issuer is not None:
|
|
creds = creds.issuer
|
|
chain.append(creds)
|
|
if not with_root and len(chain) > 1:
|
|
chain = chain[:-1]
|
|
chain_file = os.path.join(self._store_dir, f'{name}-{infix}.pem')
|
|
with open(chain_file, "wb") as fd:
|
|
for c in chain:
|
|
fd.write(c.cert_pem)
|
|
|
|
def _add_credentials(self, name: str, creds: Credentials):
|
|
if name not in self._creds_by_name:
|
|
self._creds_by_name[name] = []
|
|
self._creds_by_name[name].append(creds)
|
|
|
|
def get_credentials_for_name(self, name) -> List[Credentials]:
|
|
return self._creds_by_name[name] if name in self._creds_by_name else []
|
|
|
|
def get_cert_file(self, name: str, key_type=None) -> str:
|
|
key_infix = ".{0}".format(key_type) if key_type is not None else ""
|
|
return os.path.join(self._store_dir, f'{name}{key_infix}.cert.pem')
|
|
|
|
def get_pkey_file(self, name: str, key_type=None) -> str:
|
|
key_infix = ".{0}".format(key_type) if key_type is not None else ""
|
|
return os.path.join(self._store_dir, f'{name}{key_infix}.pkey.pem')
|
|
|
|
def load_pem_cert(self, fpath: str) -> x509.Certificate:
|
|
with open(fpath) as fd:
|
|
return x509.load_pem_x509_certificate("".join(fd.readlines()).encode())
|
|
|
|
def load_pem_pkey(self, fpath: str):
|
|
with open(fpath) as fd:
|
|
return load_pem_private_key("".join(fd.readlines()).encode(), password=None)
|
|
|
|
def load_credentials(self, name: str, key_type=None, single_file: bool = False, issuer: Credentials = None):
|
|
cert_file = self.get_cert_file(name=name, key_type=key_type)
|
|
pkey_file = cert_file if single_file else self.get_pkey_file(name=name, key_type=key_type)
|
|
if os.path.isfile(cert_file) and os.path.isfile(pkey_file):
|
|
cert = self.load_pem_cert(cert_file)
|
|
pkey = self.load_pem_pkey(pkey_file)
|
|
creds = Credentials(name=name, cert=cert, pkey=pkey, issuer=issuer)
|
|
creds.set_store(self)
|
|
creds.set_files(cert_file, pkey_file)
|
|
self._add_credentials(name, creds)
|
|
return creds
|
|
return None
|
|
|
|
|
|
class HttpdTestCA:
|
|
|
|
@classmethod
|
|
def create_root(cls, name: str, store_dir: str, key_type: str = "rsa2048") -> Credentials:
|
|
store = CertStore(fpath=store_dir)
|
|
creds = store.load_credentials(name="ca", key_type=key_type, issuer=None)
|
|
if creds is None:
|
|
creds = HttpdTestCA._make_ca_credentials(name=name, key_type=key_type)
|
|
store.save(creds, name="ca")
|
|
creds.set_store(store)
|
|
return creds
|
|
|
|
@staticmethod
|
|
def create_credentials(spec: CertificateSpec, issuer: Credentials, key_type: Any,
|
|
valid_from: timedelta = timedelta(days=-1),
|
|
valid_to: timedelta = timedelta(days=89),
|
|
serial: Optional[int] = None,
|
|
) -> Credentials:
|
|
"""Create a certificate signed by this CA for the given domains.
|
|
:returns: the certificate and private key PEM file paths
|
|
"""
|
|
if spec.domains and len(spec.domains):
|
|
creds = HttpdTestCA._make_server_credentials(name=spec.name, domains=spec.domains,
|
|
issuer=issuer, valid_from=valid_from,
|
|
valid_to=valid_to, key_type=key_type,
|
|
serial=serial)
|
|
elif spec.client:
|
|
creds = HttpdTestCA._make_client_credentials(name=spec.name, issuer=issuer,
|
|
email=spec.email, valid_from=valid_from,
|
|
valid_to=valid_to, key_type=key_type,
|
|
serial=serial)
|
|
elif spec.name:
|
|
creds = HttpdTestCA._make_ca_credentials(name=spec.name, issuer=issuer,
|
|
valid_from=valid_from, valid_to=valid_to,
|
|
key_type=key_type,
|
|
serial=serial)
|
|
else:
|
|
raise Exception(f"unrecognized certificate specification: {spec}")
|
|
return creds
|
|
|
|
@staticmethod
|
|
def _make_x509_name(org_name: str = None, common_name: str = None, parent: x509.Name = None) -> x509.Name:
|
|
name_pieces = []
|
|
if org_name:
|
|
oid = NameOID.ORGANIZATIONAL_UNIT_NAME if parent else NameOID.ORGANIZATION_NAME
|
|
name_pieces.append(x509.NameAttribute(oid, org_name))
|
|
elif common_name:
|
|
name_pieces.append(x509.NameAttribute(NameOID.COMMON_NAME, common_name))
|
|
if parent:
|
|
name_pieces.extend([rdn for rdn in parent])
|
|
return x509.Name(name_pieces)
|
|
|
|
@staticmethod
|
|
def _make_csr(
|
|
subject: x509.Name,
|
|
pkey: Any,
|
|
issuer_subject: Optional[Credentials],
|
|
valid_from_delta: timedelta = None,
|
|
valid_until_delta: timedelta = None,
|
|
serial: Optional[int] = None
|
|
):
|
|
pubkey = pkey.public_key()
|
|
issuer_subject = issuer_subject if issuer_subject is not None else subject
|
|
|
|
valid_from = datetime.now()
|
|
if valid_until_delta is not None:
|
|
valid_from += valid_from_delta
|
|
valid_until = datetime.now()
|
|
if valid_until_delta is not None:
|
|
valid_until += valid_until_delta
|
|
if serial is None:
|
|
serial = x509.random_serial_number()
|
|
return (
|
|
x509.CertificateBuilder()
|
|
.subject_name(subject)
|
|
.issuer_name(issuer_subject)
|
|
.public_key(pubkey)
|
|
.not_valid_before(valid_from)
|
|
.not_valid_after(valid_until)
|
|
.serial_number(serial)
|
|
.add_extension(
|
|
x509.SubjectKeyIdentifier.from_public_key(pubkey),
|
|
critical=False,
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def _add_ca_usages(csr: Any) -> Any:
|
|
return csr.add_extension(
|
|
x509.BasicConstraints(ca=True, path_length=9),
|
|
critical=True,
|
|
).add_extension(
|
|
x509.KeyUsage(
|
|
digital_signature=True,
|
|
content_commitment=False,
|
|
key_encipherment=False,
|
|
data_encipherment=False,
|
|
key_agreement=False,
|
|
key_cert_sign=True,
|
|
crl_sign=True,
|
|
encipher_only=False,
|
|
decipher_only=False),
|
|
critical=True
|
|
).add_extension(
|
|
x509.ExtendedKeyUsage([
|
|
ExtendedKeyUsageOID.CLIENT_AUTH,
|
|
ExtendedKeyUsageOID.SERVER_AUTH,
|
|
ExtendedKeyUsageOID.CODE_SIGNING,
|
|
]),
|
|
critical=True
|
|
)
|
|
|
|
@staticmethod
|
|
def _add_leaf_usages(csr: Any, domains: List[str], issuer: Credentials) -> Any:
|
|
csr = csr.add_extension(
|
|
x509.BasicConstraints(ca=False, path_length=None),
|
|
critical=True,
|
|
)
|
|
if issuer is not None:
|
|
csr = csr.add_extension(
|
|
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(
|
|
issuer.certificate.extensions.get_extension_for_class(
|
|
x509.SubjectKeyIdentifier).value),
|
|
critical=False
|
|
)
|
|
csr = csr.add_extension(
|
|
x509.SubjectAlternativeName([x509.DNSName(domain) for domain in domains]),
|
|
critical=True,
|
|
)
|
|
csr = csr.add_extension(
|
|
x509.ExtendedKeyUsage([
|
|
ExtendedKeyUsageOID.SERVER_AUTH,
|
|
]),
|
|
critical=True
|
|
)
|
|
return csr
|
|
|
|
@staticmethod
|
|
def _add_client_usages(csr: Any, issuer: Credentials, rfc82name: str = None) -> Any:
|
|
cert = csr.add_extension(
|
|
x509.BasicConstraints(ca=False, path_length=None),
|
|
critical=True,
|
|
).add_extension(
|
|
x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(
|
|
issuer.certificate.extensions.get_extension_for_class(
|
|
x509.SubjectKeyIdentifier).value),
|
|
critical=False
|
|
)
|
|
if rfc82name:
|
|
cert.add_extension(
|
|
x509.SubjectAlternativeName([x509.RFC822Name(rfc82name)]),
|
|
critical=True,
|
|
)
|
|
cert.add_extension(
|
|
x509.ExtendedKeyUsage([
|
|
ExtendedKeyUsageOID.CLIENT_AUTH,
|
|
]),
|
|
critical=True
|
|
)
|
|
return cert
|
|
|
|
@staticmethod
|
|
def _make_ca_credentials(name, key_type: Any,
|
|
issuer: Credentials = None,
|
|
valid_from: timedelta = timedelta(days=-1),
|
|
valid_to: timedelta = timedelta(days=89),
|
|
serial: Optional[int] = None,
|
|
) -> Credentials:
|
|
pkey = _private_key(key_type=key_type)
|
|
if issuer is not None:
|
|
issuer_subject = issuer.certificate.subject
|
|
issuer_key = issuer.private_key
|
|
else:
|
|
issuer_subject = None
|
|
issuer_key = pkey
|
|
subject = HttpdTestCA._make_x509_name(org_name=name, parent=issuer.subject if issuer else None)
|
|
csr = HttpdTestCA._make_csr(subject=subject,
|
|
issuer_subject=issuer_subject, pkey=pkey,
|
|
valid_from_delta=valid_from, valid_until_delta=valid_to,
|
|
serial=serial)
|
|
csr = HttpdTestCA._add_ca_usages(csr)
|
|
cert = csr.sign(private_key=issuer_key,
|
|
algorithm=hashes.SHA256(),
|
|
backend=default_backend())
|
|
return Credentials(name=name, cert=cert, pkey=pkey, issuer=issuer)
|
|
|
|
@staticmethod
|
|
def _make_server_credentials(name: str, domains: List[str], issuer: Credentials,
|
|
key_type: Any,
|
|
valid_from: timedelta = timedelta(days=-1),
|
|
valid_to: timedelta = timedelta(days=89),
|
|
serial: Optional[int] = None,
|
|
) -> Credentials:
|
|
name = name
|
|
pkey = _private_key(key_type=key_type)
|
|
if issuer is not None:
|
|
issuer_subject = issuer.certificate.subject
|
|
issuer_key = issuer.private_key
|
|
else:
|
|
issuer_subject = None
|
|
issuer_key = pkey
|
|
subject = HttpdTestCA._make_x509_name(common_name=name, parent=issuer_subject)
|
|
csr = HttpdTestCA._make_csr(subject=subject,
|
|
issuer_subject=issuer_subject, pkey=pkey,
|
|
valid_from_delta=valid_from, valid_until_delta=valid_to,
|
|
serial=serial)
|
|
csr = HttpdTestCA._add_leaf_usages(csr, domains=domains, issuer=issuer)
|
|
cert = csr.sign(private_key=issuer_key,
|
|
algorithm=hashes.SHA256(),
|
|
backend=default_backend())
|
|
return Credentials(name=name, cert=cert, pkey=pkey, issuer=issuer)
|
|
|
|
@staticmethod
|
|
def _make_client_credentials(name: str,
|
|
issuer: Credentials, email: Optional[str],
|
|
key_type: Any,
|
|
valid_from: timedelta = timedelta(days=-1),
|
|
valid_to: timedelta = timedelta(days=89),
|
|
serial: Optional[int] = None,
|
|
) -> Credentials:
|
|
pkey = _private_key(key_type=key_type)
|
|
if issuer is not None:
|
|
issuer_subject = issuer.certificate.subject
|
|
issuer_key = issuer.private_key
|
|
else:
|
|
issuer_subject = None
|
|
issuer_key = pkey
|
|
subject = HttpdTestCA._make_x509_name(common_name=name, parent=issuer_subject)
|
|
csr = HttpdTestCA._make_csr(subject=subject,
|
|
issuer_subject=issuer_subject, pkey=pkey,
|
|
valid_from_delta=valid_from, valid_until_delta=valid_to,
|
|
serial=serial)
|
|
csr = HttpdTestCA._add_client_usages(csr, issuer=issuer, rfc82name=email)
|
|
cert = csr.sign(private_key=issuer_key,
|
|
algorithm=hashes.SHA256(),
|
|
backend=default_backend())
|
|
return Credentials(name=name, cert=cert, pkey=pkey, issuer=issuer)
|