| 1 | Patch-Source: not yet submitted upstream; certauth has had no commit since
|
| 2 | 2019 and https://github.com/ikreymer/certauth/issues/23 (the same pyOpenSSL
|
| 3 | breakage) has been open and unanswered. To be filed at
|
| 4 | https://github.com/ikreymer/certauth/issues
|
| 5 |
|
| 6 | pyOpenSSL 26.4.0 removed X509Req, X509Extension, PKCS12, load_pkcs12 and
|
| 7 | X509.add_extensions, which leaves certauth completely non-functional -
|
| 8 | CertificateAuthority() cannot even be constructed. Replace pyopenssl with
|
| 9 | cryptography throughout.
|
| 10 |
|
| 11 | Notes on behaviour changes:
|
| 12 | - The CSR built in generate_host_cert() is dropped, it only carried the
|
| 13 | public key into a certificate signed by the CA.
|
| 14 | - Serial numbers now start at 1, cryptography rejects a zero serial number.
|
| 15 | - Host certificates get an authorityKeyIdentifier. Without it OpenSSL 3.5
|
| 16 | rejects them with "Missing Authority Key Identifier", which defeats the
|
| 17 | point of the package.
|
| 18 | - CertificateAuthority now hands out cryptography objects instead of
|
| 19 | OpenSSL.crypto ones. pyOpenSSL's SSL.Context.use_certificate() and
|
| 20 | use_privatekey() accept both, so wsgiprox keeps working.
|
| 21 |
|
| 22 | |
| 23 | |
| 24 | @@ -32,7 +32,7 @@
|
| 25 | 'certauth',
|
| 26 | ],
|
| 27 | install_requires=[
|
| 28 | - 'pyopenssl',
|
| 29 | + 'cryptography',
|
| 30 | 'tldextract',
|
| 31 | ],
|
| 32 | zip_safe=True,
|
| 33 |
|
| 34 | |
| 35 | |
| 36 | @@ -3,9 +3,14 @@
|
| 37 |
|
| 38 | from io import BytesIO
|
| 39 |
|
| 40 | -from OpenSSL import crypto
|
| 41 | -from OpenSSL.SSL import FILETYPE_PEM
|
| 42 | +from cryptography import x509
|
| 43 | +from cryptography.hazmat.primitives import hashes
|
| 44 | +from cryptography.hazmat.primitives import serialization
|
| 45 | +from cryptography.hazmat.primitives.asymmetric import rsa
|
| 46 | +from cryptography.hazmat.primitives.serialization import pkcs12
|
| 47 | +from cryptography.x509.oid import NameOID
|
| 48 |
|
| 49 | +import datetime
|
| 50 | import random
|
| 51 |
|
| 52 | import ipaddress
|
| 53 | @@ -33,6 +38,14 @@
|
| 54 |
|
| 55 |
|
| 56 | # =================================================================
|
| 57 | +def _hash_algorithm(hash_func):
|
| 58 | + try:
|
| 59 | + return getattr(hashes, hash_func.upper())()
|
| 60 | + except AttributeError:
|
| 61 | + raise ValueError('Unsupported hash function: ' + hash_func)
|
| 62 | +
|
| 63 | +
|
| 64 | +# =================================================================
|
| 65 | class CertificateAuthority(object):
|
| 66 | """
|
| 67 | Utility class for signing individual certificate
|
| 68 | @@ -191,10 +204,12 @@
|
| 69 | return res[2]
|
| 70 |
|
| 71 | def get_root_PKCS12(self):
|
| 72 | - p12 = crypto.PKCS12()
|
| 73 | - p12.set_certificate(self.ca_cert)
|
| 74 | - p12.set_privatekey(self.ca_key)
|
| 75 | - return p12.export()
|
| 76 | + return pkcs12.serialize_key_and_certificates(
|
| 77 | + name=None,
|
| 78 | + key=self.ca_key,
|
| 79 | + cert=self.ca_cert,
|
| 80 | + cas=None,
|
| 81 | + encryption_algorithm=serialization.NoEncryption())
|
| 82 |
|
| 83 | def get_root_pem(self):
|
| 84 | return self.ca_file_cache.get(ROOT_CA)
|
| 85 | @@ -203,40 +218,46 @@
|
| 86 | return self.ca_file_cache.ca_file
|
| 87 |
|
| 88 | def _make_cert(self, certname):
|
| 89 | - cert = crypto.X509()
|
| 90 | - cert.set_serial_number(random.randint(0, 2 ** 64 - 1))
|
| 91 | - cert.get_subject().CN = certname
|
| 92 | -
|
| 93 | - cert.set_version(2)
|
| 94 | - cert.gmtime_adj_notBefore(self.cert_not_before)
|
| 95 | - cert.gmtime_adj_notAfter(self.cert_not_after)
|
| 96 | - return cert
|
| 97 | + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, certname)])
|
| 98 | +
|
| 99 | + now = datetime.datetime.now(datetime.timezone.utc)
|
| 100 | +
|
| 101 | + builder = (x509.CertificateBuilder()
|
| 102 | + .serial_number(random.randint(1, 2 ** 64 - 1))
|
| 103 | + .subject_name(subject)
|
| 104 | + .not_valid_before(
|
| 105 | + now + datetime.timedelta(seconds=self.cert_not_before))
|
| 106 | + .not_valid_after(
|
| 107 | + now + datetime.timedelta(seconds=self.cert_not_after)))
|
| 108 | +
|
| 109 | + return builder, subject
|
| 110 |
|
| 111 | def generate_ca_root(self, ca_name, hash_func=DEF_HASH_FUNC):
|
| 112 | # Generate key
|
| 113 | - key = crypto.PKey()
|
| 114 | - key.generate_key(crypto.TYPE_RSA, 2048)
|
| 115 | + key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
| 116 |
|
| 117 | # Generate cert
|
| 118 | - cert = self._make_cert(ca_name)
|
| 119 | + builder, subject = self._make_cert(ca_name)
|
| 120 |
|
| 121 | - cert.set_issuer(cert.get_subject())
|
| 122 | - cert.set_pubkey(key)
|
| 123 | - cert.add_extensions([
|
| 124 | - crypto.X509Extension(b"basicConstraints",
|
| 125 | - True,
|
| 126 | - b"CA:TRUE, pathlen:0"),
|
| 127 | -
|
| 128 | - crypto.X509Extension(b"keyUsage",
|
| 129 | - True,
|
| 130 | - b"keyCertSign, cRLSign"),
|
| 131 | -
|
| 132 | - crypto.X509Extension(b"subjectKeyIdentifier",
|
| 133 | - False,
|
| 134 | - b"hash",
|
| 135 | - subject=cert),
|
| 136 | - ])
|
| 137 | - cert.sign(key, hash_func)
|
| 138 | + cert = (builder
|
| 139 | + .issuer_name(subject)
|
| 140 | + .public_key(key.public_key())
|
| 141 | + .add_extension(x509.BasicConstraints(ca=True, path_length=0),
|
| 142 | + critical=True)
|
| 143 | + .add_extension(x509.KeyUsage(digital_signature=False,
|
| 144 | + content_commitment=False,
|
| 145 | + key_encipherment=False,
|
| 146 | + data_encipherment=False,
|
| 147 | + key_agreement=False,
|
| 148 | + key_cert_sign=True,
|
| 149 | + crl_sign=True,
|
| 150 | + encipher_only=False,
|
| 151 | + decipher_only=False),
|
| 152 | + critical=True)
|
| 153 | + .add_extension(
|
| 154 | + x509.SubjectKeyIdentifier.from_public_key(key.public_key()),
|
| 155 | + critical=False)
|
| 156 | + .sign(key, _hash_algorithm(hash_func)))
|
| 157 |
|
| 158 | return cert, key
|
| 159 |
|
| 160 | @@ -245,51 +266,47 @@
|
| 161 | hash_func=DEF_HASH_FUNC,
|
| 162 | is_ip=False):
|
| 163 |
|
| 164 | - host = host.encode('utf-8')
|
| 165 | -
|
| 166 | # Generate key
|
| 167 | - key = crypto.PKey()
|
| 168 | - key.generate_key(crypto.TYPE_RSA, 2048)
|
| 169 | -
|
| 170 | - # Generate CSR
|
| 171 | - req = crypto.X509Req()
|
| 172 | - req.get_subject().CN = host
|
| 173 | - req.set_pubkey(key)
|
| 174 | - req.sign(key, hash_func)
|
| 175 | + key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
| 176 |
|
| 177 | # Generate Cert
|
| 178 | - cert = self._make_cert(host)
|
| 179 | + builder, dummy = self._make_cert(host)
|
| 180 |
|
| 181 | - cert.set_issuer(root_cert.get_subject())
|
| 182 | - cert.set_pubkey(req.get_pubkey())
|
| 183 | -
|
| 184 | - primary = b'DNS:' + host
|
| 185 | + alt_hosts = [x509.DNSName(host)]
|
| 186 |
|
| 187 | if wildcard:
|
| 188 | - alt_hosts = primary + b', DNS:*.' + host
|
| 189 | + alt_hosts.append(x509.DNSName('*.' + host))
|
| 190 |
|
| 191 | elif is_ip:
|
| 192 | - alt_hosts = b'IP:' + host + b', ' + primary
|
| 193 | + alt_hosts.insert(0, x509.IPAddress(ipaddress.ip_address(host)))
|
| 194 |
|
| 195 | - else:
|
| 196 | - alt_hosts = primary
|
| 197 | + root_ski = root_cert.extensions.get_extension_for_class(
|
| 198 | + x509.SubjectKeyIdentifier)
|
| 199 |
|
| 200 | - cert.add_extensions([
|
| 201 | - crypto.X509Extension(b'subjectAltName',
|
| 202 | - False,
|
| 203 | - alt_hosts)])
|
| 204 | + cert = (builder
|
| 205 | + .issuer_name(root_cert.subject)
|
| 206 | + .public_key(key.public_key())
|
| 207 | + .add_extension(x509.SubjectAlternativeName(alt_hosts),
|
| 208 | + critical=False)
|
| 209 | + .add_extension(
|
| 210 | + x509.AuthorityKeyIdentifier
|
| 211 | + .from_issuer_subject_key_identifier(root_ski.value),
|
| 212 | + critical=False)
|
| 213 | + .sign(root_key, _hash_algorithm(hash_func)))
|
| 214 |
|
| 215 | - cert.sign(root_key, hash_func)
|
| 216 | return cert, key
|
| 217 |
|
| 218 | def write_pem(self, buff, cert, key):
|
| 219 | - buff.write(crypto.dump_privatekey(FILETYPE_PEM, key))
|
| 220 | - buff.write(crypto.dump_certificate(FILETYPE_PEM, cert))
|
| 221 | + buff.write(key.private_bytes(
|
| 222 | + encoding=serialization.Encoding.PEM,
|
| 223 | + format=serialization.PrivateFormat.TraditionalOpenSSL,
|
| 224 | + encryption_algorithm=serialization.NoEncryption()))
|
| 225 | + buff.write(cert.public_bytes(serialization.Encoding.PEM))
|
| 226 |
|
| 227 | def read_pem(self, buff):
|
| 228 | - cert = crypto.load_certificate(FILETYPE_PEM, buff.read())
|
| 229 | - buff.seek(0)
|
| 230 | - key = crypto.load_privatekey(FILETYPE_PEM, buff.read())
|
| 231 | + data = buff.read()
|
| 232 | + cert = x509.load_pem_x509_certificate(data)
|
| 233 | + key = serialization.load_pem_private_key(data, password=None)
|
| 234 | return cert, key
|
| 235 |
|
| 236 |
|
| 237 |
|
| 238 | |
| 239 | |
| 240 | @@ -4,8 +4,8 @@
|
| 241 | from certauth.certauth import main, CertificateAuthority, FileCache, LRUCache, ROOT_CA
|
| 242 |
|
| 243 | import tempfile
|
| 244 | -from OpenSSL import crypto
|
| 245 | -from OpenSSL.SSL import FILETYPE_PEM
|
| 246 | +from cryptography import x509
|
| 247 | +from cryptography.hazmat.primitives.serialization import pkcs12
|
| 248 | import datetime
|
| 249 | import time
|
| 250 |
|
| 251 | @@ -39,13 +39,21 @@
|
| 252 | assert not os.path.isfile(TEST_CA_ROOT)
|
| 253 |
|
| 254 |
|
| 255 | -def verify_cert_san(cert, san_string):
|
| 256 | - assert cert.get_extension_count() == 1
|
| 257 | - assert str(cert.get_extension(0)) == san_string
|
| 258 | +def format_cert_san(cert):
|
| 259 | + san = cert.extensions.get_extension_for_class(
|
| 260 | + x509.SubjectAlternativeName).value
|
| 261 | +
|
| 262 | + return ', '.join(
|
| 263 | + 'IP Address:' + str(name.value) if isinstance(name, x509.IPAddress)
|
| 264 | + else 'DNS:' + name.value
|
| 265 | + for name in san)
|
| 266 | +
|
| 267 |
|
| 268 | - #if 'IP Address' in san_string:
|
| 269 | - # print(crypto.dump_certificate(crypto.FILETYPE_TEXT, cert).decode('utf-8'))
|
| 270 | - # assert False
|
| 271 | +def verify_cert_san(cert, san_string):
|
| 272 | + assert len(cert.extensions) == 2
|
| 273 | + assert cert.extensions.get_extension_for_class(
|
| 274 | + x509.AuthorityKeyIdentifier) is not None
|
| 275 | + assert format_cert_san(cert) == san_string
|
| 276 |
|
| 277 | def verify_san(ca, filename, san_string):
|
| 278 | assert os.path.isfile(filename)
|
| 279 | @@ -123,7 +131,7 @@
|
| 280 | filename = os.path.join(TEST_CA_DIR, '2001-0db8-85a3-0000-0000-8a2e-0370-7334.pem')
|
| 281 | assert cert_filename == filename
|
| 282 |
|
| 283 | - verify_san(ca, filename, 'IP Address:2001:DB8:85A3:0:0:8A2E:370:7334\n, DNS:2001:0db8:85a3:0000:0000:8a2e:0370:7334')
|
| 284 | + verify_san(ca, filename, 'IP Address:2001:db8:85a3::8a2e:370:7334, DNS:2001:0db8:85a3:0000:0000:8a2e:0370:7334')
|
| 285 |
|
| 286 | os.remove(filename)
|
| 287 |
|
| 288 | @@ -223,16 +231,13 @@
|
| 289 | buff_pk12 = ca.get_root_PKCS12()
|
| 290 | assert len(buff_pk12) > 0
|
| 291 |
|
| 292 | - cert = crypto.load_pkcs12(buff_pk12).get_certificate()
|
| 293 | + cert = pkcs12.load_key_and_certificates(buff_pk12, None)[1]
|
| 294 |
|
| 295 | expected_not_before = datetime.datetime.utcnow() - datetime.timedelta(seconds=60 * 60)
|
| 296 | expected_not_after = datetime.datetime.utcnow() + datetime.timedelta(seconds=60 * 60 * 24 * 3)
|
| 297 |
|
| 298 | -
|
| 299 | - actual_not_before = datetime.datetime.strptime(
|
| 300 | - cert.get_notBefore().decode('ascii'), '%Y%m%d%H%M%SZ')
|
| 301 | - actual_not_after = datetime.datetime.strptime(
|
| 302 | - cert.get_notAfter().decode('ascii'), '%Y%m%d%H%M%SZ')
|
| 303 | + actual_not_before = cert.not_valid_before_utc.replace(tzinfo=None)
|
| 304 | + actual_not_after = cert.not_valid_after_utc.replace(tzinfo=None)
|
| 305 |
|
| 306 | time.mktime(expected_not_before.utctimetuple())
|
| 307 | assert abs((time.mktime(actual_not_before.utctimetuple()) - time.mktime(expected_not_before.utctimetuple()))) < 10
|
| 308 | @@ -251,13 +256,15 @@
|
| 309 | # PEM stored in root_cert_dict
|
| 310 | assert root_cert_dict[ROOT_CA] == buff_pem
|
| 311 |
|
| 312 | - cert_pem = crypto.load_certificate(FILETYPE_PEM, buff_pem)
|
| 313 | + cert_pem = x509.load_pem_x509_certificate(buff_pem)
|
| 314 |
|
| 315 | # check PKCS12
|
| 316 | buff_pk12 = ca.get_root_PKCS12()
|
| 317 | assert len(buff_pk12) > 0
|
| 318 |
|
| 319 | - cert_pk12 = crypto.load_pkcs12(buff_pk12).get_certificate()
|
| 320 | + cert_pk12 = pkcs12.load_key_and_certificates(buff_pk12, None)[1]
|
| 321 | +
|
| 322 | + assert cert_pk12 == cert_pem
|
| 323 |
|
| 324 |
|
| 325 | def test_ca_lru_cache():
|