Move _get_names_from_subject_and_extensions to acme's crypto_utils (#10111)

Make use of it in more places
This commit is contained in:
Alex Gaynor
2025-01-03 16:21:31 -08:00
committed by GitHub
parent a1fce6b398
commit 5dd898f56b
7 changed files with 97 additions and 56 deletions
+9
View File
@@ -24,6 +24,7 @@ from acme.client import ClientV2
CERT_SAN_PEM = test_util.load_vector('cert-san.pem') CERT_SAN_PEM = test_util.load_vector('cert-san.pem')
CSR_MIXED_PEM = test_util.load_vector('csr-mixed.pem') CSR_MIXED_PEM = test_util.load_vector('csr-mixed.pem')
CSR_NO_SANS_PEM = test_util.load_vector('csr-nosans.pem')
KEY = jose.JWKRSA.load(test_util.load_vector('rsa512_key.pem')) KEY = jose.JWKRSA.load(test_util.load_vector('rsa512_key.pem'))
DIRECTORY_V2 = messages.Directory({ DIRECTORY_V2 = messages.Directory({
@@ -97,6 +98,10 @@ class ClientV2Test(unittest.TestCase):
body=self.order, body=self.order,
uri='https://www.letsencrypt-demo.org/acme/acct/1/order/1', uri='https://www.letsencrypt-demo.org/acme/acct/1/order/1',
authorizations=[self.authzr, self.authzr2], csr_pem=CSR_MIXED_PEM) authorizations=[self.authzr, self.authzr2], csr_pem=CSR_MIXED_PEM)
self.orderr2 = messages.OrderResource(
body=self.order,
uri='https://www.letsencrypt-demo.org/acme/acct/1/order/1',
authorizations=[self.authzr, self.authzr2], csr_pem=CSR_NO_SANS_PEM)
def test_new_account(self): def test_new_account(self):
self.response.status_code = http_client.CREATED self.response.status_code = http_client.CREATED
@@ -158,6 +163,10 @@ class ClientV2Test(unittest.TestCase):
mock_post_as_get.side_effect = (authz_response, authz_response2) mock_post_as_get.side_effect = (authz_response, authz_response2)
assert self.client.new_order(CSR_MIXED_PEM) == self.orderr assert self.client.new_order(CSR_MIXED_PEM) == self.orderr
with mock.patch('acme.client.ClientV2._post_as_get') as mock_post_as_get:
mock_post_as_get.side_effect = (authz_response, authz_response2)
assert self.client.new_order(CSR_NO_SANS_PEM) == self.orderr2
def test_answer_challege(self): def test_answer_challege(self):
self.response.links['up'] = {'url': self.challr.authzr_uri} self.response.links['up'] = {'url': self.challr.authzr_uri}
self.response.json.return_value = self.challr.body.to_json() self.response.json.return_value = self.challr.body.to_json()
+14 -14
View File
@@ -1,16 +1,16 @@
-----BEGIN CERTIFICATE REQUEST----- -----BEGIN CERTIFICATE REQUEST-----
MIICdjCCAV4CAQIwADCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMXq MIICdjCCAV4CAQAwADCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBANoV
v1y8EIcCbaUIzCtOcLkLS0MJ35oS+6DmV5WB1A0cIk6YrjsHIsY2lwMm13BWIvmw T1pdvRUUBOqvm7M2ebLEHV7higUH7qAGUZEkfP6W4YriYVY+IHrH1svNPSa+oPTK
tY+Y6n0rr7eViNx5ZRGHpHEI/TL3Neb+VefTydL5CgvK3dd4ex2kSbTaed3fmpOx 7weDNmT11ehWnGyECIM9z2r2Hi9yVV0ycxh4hWQ4Nt8BAKZwCwaXpyWm7Gj6m2Ez
qMajEduwNcZPCcmoEXPkfrCP8w2vKQUkQ+JRPcdX1nTuzticeRP5B7YCmJsmxkEh pSN5Dd67g5YAQBrUUh1+RRbFi9c0Ls/6ZOExMvfg8kqt4c2sXCgH1IFnxvvOjBYo
Y0tzzZ+NIRDARoYNofefY86h3e5q66gtJxccNchmIM3YQahhg5n3Xoo8hGfM/TIc p7xh0x3L1Akyax0tw8qgQp/z5mkupmVDNJYPFmbzFPMNyDR61ed6QUTDg7P4UAuF
R7ncCBCLO6vtqo0QFva/NQODrgOmOsmgvqPkUWQFdZfWM8yIaU826dktx0CPB78t kejLLzFvz5YaO7vC+huaTuPhInAhpzqpr4yU97KIjos2/83Itu/Cv8U1RAeEeRTk
TudnJ1rBRvGsjHMsZikCAwEAAaAxMC8GCSqGSIb3DQEJDjEiMCAwHgYDVR0RBBcw h0WjUfltoem/5f8bIdsCAwEAAaAxMC8GCSqGSIb3DQEJDjEiMCAwHgYDVR0RBBcw
FYINYS5leGVtcGxlLmNvbYcEwAACbzANBgkqhkiG9w0BAQsFAAOCAQEAdGMcRCxq FYINYS5leGVtcGxlLmNvbYcEwAACbzANBgkqhkiG9w0BAQsFAAOCAQEAQ7n/hYen
1X09gn1TNdMt64XUv+wdJCKDaJ+AgyIJj7QvVw8H5k7dOnxS4I+a/yo4jE+LDl2/ 5INHlcslHPYCQ/BAbX6Ou+Y8hUu8puWNVpE2OM95L2C87jbWwTmCRnkFBwtyoNqo
AuHcBLFEI4ddewdJSMrTNZjuRYuOdr3KP7fL7MffICSBi45vw5EOXg0tnjJCEiKu j3DXVW2RYv8y/exq7V6Y5LtpHTgwfugINJ3XlcVzA4Vnf1xqOxv3kwejkq74RuXn
6gcJgbLSP5JMMd7Haf33Q/VWsmHofR3VwOMdrnakwAU3Ff5WTuXTNVhL1kT/uLFX xd5N28srgiFqb0e4tOAWVI8Tw27bgBqjoXl0QDFPZpctqUia5bcDJ9WzNSM7VaO1
yW1ru6BF4unwNqSR2UeulljpNfRBsiN4zJK11W6n9KT0NkBr9zY5WCM4sW7i8k9V CBNGHBRz+zL8sqoqJA4HV58tjcgzl+1RtGM+iUHxXpnH+aCNKWIUINrAzIm4Sm00
TeypWGo3jBKzYAGeuxZsB97U77jZ2lrGdBLZKfbcjnTeRVqCvCRrui4El7UGYFmj 93RJjhb1kdNR0BC7ikWVbAWaVviHdvATK/RfpmhWDqfEaNgBpvT91GnkhpzctSFD
7s6OJyWx5DSV8w== ro0yCUUXXrIr0w==
-----END CERTIFICATE REQUEST----- -----END CERTIFICATE REQUEST-----
+12 -7
View File
@@ -15,6 +15,8 @@ from typing import Set
from typing import Tuple from typing import Tuple
from typing import Union from typing import Union
from cryptography import x509
import josepy as jose import josepy as jose
import OpenSSL import OpenSSL
import requests import requests
@@ -121,18 +123,21 @@ class ClientV2:
:returns: The newly created order. :returns: The newly created order.
:rtype: OrderResource :rtype: OrderResource
""" """
csr = OpenSSL.crypto.load_certificate_request(OpenSSL.crypto.FILETYPE_PEM, csr_pem) csr = x509.load_pem_x509_csr(csr_pem)
# pylint: disable=protected-access dnsNames = crypto_util.get_names_from_subject_and_extensions(csr.subject, csr.extensions)
dnsNames = crypto_util._pyopenssl_cert_or_req_all_names(csr) try:
ipNames = crypto_util._pyopenssl_cert_or_req_san_ip(csr) san_ext = csr.extensions.get_extension_for_class(x509.SubjectAlternativeName)
# ipNames is now []string except x509.ExtensionNotFound:
ipNames = []
else:
ipNames = san_ext.value.get_values_for_type(x509.IPAddress)
identifiers = [] identifiers = []
for name in dnsNames: for name in dnsNames:
identifiers.append(messages.Identifier(typ=messages.IDENTIFIER_FQDN, identifiers.append(messages.Identifier(typ=messages.IDENTIFIER_FQDN,
value=name)) value=name))
for ips in ipNames: for ip in ipNames:
identifiers.append(messages.Identifier(typ=messages.IDENTIFIER_IP, identifiers.append(messages.Identifier(typ=messages.IDENTIFIER_IP,
value=ips)) value=str(ip)))
order = messages.NewOrder(identifiers=identifiers) order = messages.NewOrder(identifiers=identifiers)
response = self._post(self.directory['newOrder'], order) response = self._post(self.directory['newOrder'], order)
body = messages.Order.from_json(response.json()) body = messages.Order.from_json(response.json())
+27
View File
@@ -7,6 +7,7 @@ import logging
import os import os
import re import re
import socket import socket
import typing
from typing import Any from typing import Any
from typing import Callable from typing import Callable
from typing import List from typing import List
@@ -307,6 +308,32 @@ def make_csr(
return csr.public_bytes(serialization.Encoding.PEM) return csr.public_bytes(serialization.Encoding.PEM)
def get_names_from_subject_and_extensions(
subject: x509.Name, exts: x509.Extensions
) -> List[str]:
"""Gets all DNS SAN names as well as the first Common Name from subject.
"""
# We know these are always `str` because `bytes` is only possible for
# other OIDs.
cns = [
typing.cast(str, c.value)
for c in subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
]
try:
san_ext = exts.get_extension_for_class(x509.SubjectAlternativeName)
except x509.ExtensionNotFound:
dns_names = []
else:
dns_names = san_ext.value.get_values_for_type(x509.DNSName)
if not cns:
return dns_names
else:
# We only include the first CN, if there are multiple. This matches
# the behavior of the previously implementation using pyOpenSSL.
return [cns[0]] + [d for d in dns_names if d != cns[0]]
def _pyopenssl_cert_or_req_all_names(loaded_cert_or_req: Union[crypto.X509, crypto.X509Req] def _pyopenssl_cert_or_req_all_names(loaded_cert_or_req: Union[crypto.X509, crypto.X509Req]
) -> List[str]: ) -> List[str]:
# unlike its name this only outputs DNS names, other type of idents will ignored # unlike its name this only outputs DNS names, other type of idents will ignored
@@ -2,9 +2,11 @@
import sys import sys
from unittest import mock from unittest import mock
import OpenSSL
import pytest import pytest
from cryptography import x509
from cryptography.hazmat.primitives import serialization
from acme import challenges from acme import challenges
from acme import messages from acme import messages
from certbot import achallenges from certbot import achallenges
@@ -545,12 +547,10 @@ class NginxConfiguratorTest(util.NginxTest):
cert, key = self.config._get_snakeoil_paths() cert, key = self.config._get_snakeoil_paths()
assert os.path.exists(cert) assert os.path.exists(cert)
assert os.path.exists(key) assert os.path.exists(key)
with open(cert) as cert_file: with open(cert, "rb") as cert_file:
OpenSSL.crypto.load_certificate( x509.load_pem_x509_certificate(cert_file.read())
OpenSSL.crypto.FILETYPE_PEM, cert_file.read()) with open(key, "rb") as key_file:
with open(key) as key_file: serialization.load_pem_private_key(key_file.read(), password=None)
OpenSSL.crypto.load_privatekey(
OpenSSL.crypto.FILETYPE_PEM, key_file.read())
def test_redirect_enhance(self): def test_redirect_enhance(self):
# Test that we successfully add a redirect when there is # Test that we successfully add a redirect when there is
+7 -28
View File
@@ -8,7 +8,6 @@ import datetime
import hashlib import hashlib
import logging import logging
import re import re
import typing
import warnings import warnings
from typing import List from typing import List
from typing import Optional from typing import Optional
@@ -204,7 +203,7 @@ def import_csr_file(
except ValueError: except ValueError:
raise errors.Error("Failed to parse CSR file: {0}".format(csrfile)) raise errors.Error("Failed to parse CSR file: {0}".format(csrfile))
domains = _get_names_from_subject_and_extensions(csr.subject, csr.extensions) domains = acme_crypto_util.get_names_from_subject_and_extensions(csr.subject, csr.extensions)
# Internally we always use PEM, so re-encode as PEM before returning. # Internally we always use PEM, so re-encode as PEM before returning.
data_pem = csr.public_bytes(serialization.Encoding.PEM) data_pem = csr.public_bytes(serialization.Encoding.PEM)
return ( return (
@@ -214,30 +213,6 @@ def import_csr_file(
) )
def _get_names_from_subject_and_extensions(
subject: x509.Name, exts: x509.Extensions
) -> List[str]:
# We know these are always `str` because `bytes` is only possible for
# other OIDs.
cns = [
typing.cast(str, c.value)
for c in subject.get_attributes_for_oid(x509.NameOID.COMMON_NAME)
]
try:
san_ext = exts.get_extension_for_class(x509.SubjectAlternativeName)
except x509.ExtensionNotFound:
dns_names = []
else:
dns_names = san_ext.value.get_values_for_type(x509.DNSName)
if not cns:
return dns_names
else:
# We only include the first CN, if there are multiple. This matches
# the behavior of the previously implementation using pyOpenSSL.
return [cns[0]] + [d for d in dns_names if d != cns[0]]
def make_key(bits: int = 2048, key_type: str = "rsa", def make_key(bits: int = 2048, key_type: str = "rsa",
elliptic_curve: Optional[str] = None) -> bytes: elliptic_curve: Optional[str] = None) -> bytes:
"""Generate PEM encoded RSA|EC key. """Generate PEM encoded RSA|EC key.
@@ -483,7 +458,9 @@ def get_names_from_cert(
else: else:
assert typ == acme_crypto_util.Format.DER assert typ == acme_crypto_util.Format.DER
x509_cert = x509.load_der_x509_certificate(cert) x509_cert = x509.load_der_x509_certificate(cert)
return _get_names_from_subject_and_extensions(x509_cert.subject, x509_cert.extensions) return acme_crypto_util.get_names_from_subject_and_extensions(
x509_cert.subject, x509_cert.extensions
)
def get_names_from_req( def get_names_from_req(
@@ -503,7 +480,9 @@ def get_names_from_req(
else: else:
assert typ == acme_crypto_util.Format.DER assert typ == acme_crypto_util.Format.DER
x509_req = x509.load_der_x509_csr(csr) x509_req = x509.load_der_x509_csr(csr)
return _get_names_from_subject_and_extensions(x509_req.subject, x509_req.extensions) return acme_crypto_util.get_names_from_subject_and_extensions(
x509_req.subject, x509_req.extensions
)
def dump_pyopenssl_chain( def dump_pyopenssl_chain(
+21
View File
@@ -15,3 +15,24 @@ and for the CSR PEM (Certificate Signing Request):
and for the certificate: and for the certificate:
openssl req -new -out cert_X.pem -key rsaX_key.pem -subj '/CN=example.com' -x509 [-outform DER > cert_X.der] openssl req -new -out cert_X.pem -key rsaX_key.pem -subj '/CN=example.com' -x509 [-outform DER > cert_X.der]
`csr-mixed.pem` was generated with pyca/cryptography using the following snippet:
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
k = serialization.load_pem_private_key(
open("./acme/acme/_internal/tests/testdata/rsa2048_key.pem", "rb").read(), None
)
csr = (
x509.CertificateSigningRequestBuilder().add_extension(
x509.SubjectAlternativeName([x509.DNSName('a.exemple.com'), x509.IPAddress(ipaddress.ipaddr('192.0.2.111'))]),
critical=False
).subject_name(
x509.Name([])
).sign(
k, hashes.SHA256()
)
)
open("./acme/acme/_internal/tests/testdata/csr-mixed.pem", "wb").write(
csr.public_bytes(serialization.Encoding.PEM)
)