diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 8a26133a030f..c0e62b64fe09 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -13,6 +13,8 @@ Changelog the key loading APIs (including from X.509 certificates and certificate signing requests). Users should migrate to a more modern signature algorithm. +* Added :class:`~cryptography.x509.CustomExtensionType` which can be used to + parse and generate X.509 extensions using :mod:`cryptography.hazmat.asn1`. .. _v50-0-1: diff --git a/docs/x509/reference.rst b/docs/x509/reference.rst index 1dfc9975af8e..97a7f076b00b 100644 --- a/docs/x509/reference.rst +++ b/docs/x509/reference.rst @@ -2089,6 +2089,64 @@ X.509 Extensions A bytes string representing the extension's DER encoded value. +.. class:: CustomExtensionType(value) + :canonical: cryptography.x509.extensions.CustomExtensionType + + .. versionadded:: 51.0.0 + + A base class for defining extension types that are not built into + ``cryptography``. Subclasses must be parameterized with the ASN.1 type of + the extension's value, using the types supported by + :mod:`cryptography.hazmat.asn1`, and must define an ``oid`` class + attribute. + + Custom extension classes can be passed to + :meth:`Extensions.get_extension_for_class`, which parses the matching + extension's DER value into the custom class. Instances can also be passed + to the ``add_extension`` method of the certificate, CRL, CSR, and OCSP + builders, which serialize the value to DER with + :func:`~cryptography.hazmat.asn1.encode_der`. + + .. code-block:: python + + from cryptography import x509 + from cryptography.hazmat import asn1 + + @asn1.sequence + class Point: + x: int + y: int + + class PointExtension(x509.CustomExtensionType[Point]): + oid = x509.ObjectIdentifier("1.2.3.4") + + ext = cert.extensions.get_extension_for_class(PointExtension) + print(ext.value.value.x, ext.value.value.y) + + builder = builder.add_extension( + PointExtension(Point(x=1, y=2)), critical=False + ) + + :param value: The extension's value, an instance of the ASN.1 type the + class was parameterized with. + + .. attribute:: oid + + :type: :class:`ObjectIdentifier` + + Returns the OID associated with this extension, as defined by the + subclass. + + .. attribute:: value + + The parsed value of the extension. + + .. method:: public_bytes() + + :return bytes: + + A bytes string representing the extension's DER encoded value. + .. class:: KeyUsage(digital_signature, content_commitment, key_encipherment, data_encipherment, key_agreement, key_cert_sign, crl_sign, encipher_only, decipher_only) :canonical: cryptography.x509.extensions.KeyUsage diff --git a/src/cryptography/x509/__init__.py b/src/cryptography/x509/__init__.py index 6442b0096c1c..f7080b98a51a 100644 --- a/src/cryptography/x509/__init__.py +++ b/src/cryptography/x509/__init__.py @@ -40,6 +40,7 @@ CRLDistributionPoints, CRLNumber, CRLReason, + CustomExtensionType, DeltaCRLIndicator, DistributionPoint, DuplicateExtension, @@ -199,6 +200,7 @@ "CertificateRevocationListBuilder", "CertificateSigningRequest", "CertificateSigningRequestBuilder", + "CustomExtensionType", "DNSName", "DeltaCRLIndicator", "DirectoryName", diff --git a/src/cryptography/x509/extensions.py b/src/cryptography/x509/extensions.py index dc7728e23e67..f80d31b4f798 100644 --- a/src/cryptography/x509/extensions.py +++ b/src/cryptography/x509/extensions.py @@ -12,6 +12,8 @@ from collections.abc import Iterable, Iterator from cryptography import utils +from cryptography.hazmat.asn1.asn1 import decode_der as _decode_der +from cryptography.hazmat.asn1.asn1 import encode_der as _encode_der from cryptography.hazmat.bindings._rust import asn1 from cryptography.hazmat.bindings._rust import x509 as rust_x509 from cryptography.hazmat.primitives import _serialization, constant_time @@ -46,6 +48,10 @@ ExtensionTypeVar = typing.TypeVar( "ExtensionTypeVar", bound="ExtensionType", covariant=True ) +_ValueT = typing.TypeVar("_ValueT") +_CustomExtensionTypeVar = typing.TypeVar( + "_CustomExtensionTypeVar", bound="CustomExtensionType[typing.Any]" +) def _key_identifier_from_public_key( @@ -109,6 +115,77 @@ def public_bytes(self) -> bytes: ) +class CustomExtensionType(ExtensionType, typing.Generic[_ValueT]): + """ + Base class for user-defined extension types. The extension's value is + parsed and serialized with :mod:`cryptography.hazmat.asn1`. + + Subclasses must parameterize this class with the ASN.1 type of the + extension's value and define ``oid``:: + + class MyExtension(CustomExtensionType[MyValue]): + oid = ObjectIdentifier("1.2.3.4") + """ + + _asn1_type: typing.ClassVar[typing.Any] + + def __init_subclass__(cls, **kwargs: typing.Any) -> None: + super().__init_subclass__(**kwargs) + + if len(cls.__bases__) != 1: + raise TypeError( + "CustomExtensionType subclasses cannot use multiple " + "inheritance" + ) + + # Record the ASN.1 type this class was parameterized with. Subclasses + # of an already-parameterized class inherit it. + (base,) = cls.__dict__.get("__orig_bases__", cls.__bases__) + if typing.get_origin(base) is CustomExtensionType: + (asn1_type,) = typing.get_args(base) + if not isinstance(asn1_type, typing.TypeVar): + cls._asn1_type = asn1_type + + if not hasattr(cls, "_asn1_type"): + raise TypeError( + f"{cls.__name__} must subclass CustomExtensionType " + "parameterized with the ASN.1 type of the extension's value" + ) + if not isinstance(getattr(cls, "oid", None), ObjectIdentifier): + raise TypeError( + f"{cls.__name__} must define an 'oid' class attribute that " + "is an ObjectIdentifier" + ) + + def __init__(self, value: _ValueT) -> None: + self._value = value + + @property + def value(self) -> _ValueT: + return self._value + + @classmethod + def _from_der( + cls: type[_CustomExtensionTypeVar], data: bytes + ) -> _CustomExtensionTypeVar: + return cls(_decode_der(cls._asn1_type, data)) + + def __repr__(self) -> str: + return f"<{self.__class__.__name__}(value={self.value!r})>" + + def __eq__(self, other: object) -> bool: + if not isinstance(other, CustomExtensionType): + return NotImplemented + + return self.__class__ is other.__class__ and self.value == other.value + + def __hash__(self) -> int: + return hash((self.__class__, self.value)) + + def public_bytes(self) -> bytes: + return _encode_der(self.value) + + class Extensions: def __init__(self, extensions: Iterable[Extension[ExtensionType]]) -> None: self._extensions = list(extensions) @@ -135,6 +212,16 @@ def get_extension_for_class( for ext in self: if isinstance(ext.value, extclass): return ext + # Custom extension types are not known to the parser, so their + # extensions are present as `UnrecognizedExtension`. Parse the + # DER value with the custom type. + if ( + issubclass(extclass, CustomExtensionType) + and isinstance(ext.value, UnrecognizedExtension) + and ext.oid == extclass.oid + ): + value = extclass._from_der(ext.value.value) + return Extension(ext.oid, ext.critical, value) raise ExtensionNotFound( f"No {extclass} extension was found", extclass.oid diff --git a/src/rust/src/types.rs b/src/rust/src/types.rs index d0fff8be732f..24fc17943e59 100644 --- a/src/rust/src/types.rs +++ b/src/rust/src/types.rs @@ -136,6 +136,8 @@ pub static AUTHORITY_KEY_IDENTIFIER: LazyPyImport = LazyPyImport::new("cryptography.x509", &["AuthorityKeyIdentifier"]); pub static UNRECOGNIZED_EXTENSION: LazyPyImport = LazyPyImport::new("cryptography.x509", &["UnrecognizedExtension"]); +pub static CUSTOM_EXTENSION_TYPE: LazyPyImport = + LazyPyImport::new("cryptography.x509", &["CustomExtensionType"]); pub static EXTENSION: LazyPyImport = LazyPyImport::new("cryptography.x509", &["Extension"]); pub static EXTENSIONS: LazyPyImport = LazyPyImport::new("cryptography.x509", &["Extensions"]); pub static NAME: LazyPyImport = LazyPyImport::new("cryptography.x509", &["Name"]); diff --git a/src/rust/src/x509/common.rs b/src/rust/src/x509/common.rs index 332ec046e625..90fd77910d6d 100644 --- a/src/rust/src/x509/common.rs +++ b/src/rust/src/x509/common.rs @@ -460,11 +460,20 @@ pub(crate) fn encode_extensions< let oid = py_oid_to_oid(py_oid)?; let ext_val = py_ext.getattr(pyo3::intern!(py, "value"))?; - if ext_val.is_instance(&types::UNRECOGNIZED_EXTENSION.get(py)?)? { + if ext_val.is_instance(&types::UNRECOGNIZED_EXTENSION.get(py)?)? + || ext_val.is_instance(&types::CUSTOM_EXTENSION_TYPE.get(py)?)? + { + // Both of these carry their own DER encoding: as raw bytes for + // UnrecognizedExtension, and via the declarative ASN.1 encoder + // for CustomExtensionType subclasses. exts.push(Extension { extn_id: oid, critical: py_ext.getattr(pyo3::intern!(py, "critical"))?.extract()?, - extn_value: ka_bytes.add(ext_val.getattr(pyo3::intern!(py, "value"))?.extract()?), + extn_value: ka_bytes.add( + ext_val + .call_method0(pyo3::intern!(py, "public_bytes"))? + .extract()?, + ), }); continue; } diff --git a/tests/x509/test_x509.py b/tests/x509/test_x509.py index 76f1349fa8a7..365643b6762b 100644 --- a/tests/x509/test_x509.py +++ b/tests/x509/test_x509.py @@ -14,6 +14,7 @@ from cryptography import utils, x509 from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm +from cryptography.hazmat import asn1 from cryptography.hazmat.bindings._rust import test_support from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ( @@ -4901,6 +4902,45 @@ def test_unrecognized_extension( assert ext.value == unrecognized + def test_custom_extension(self, rsa_key_2048: rsa.RSAPrivateKey): + @asn1.sequence + class Point: + x: int + y: int + + class PointExtension(x509.CustomExtensionType[Point]): + oid = x509.ObjectIdentifier("1.2.3.4") + + private_key = rsa_key_2048 + + cert = ( + x509.CertificateBuilder() + .subject_name( + x509.Name([x509.NameAttribute(x509.OID_COUNTRY_NAME, "US")]) + ) + .issuer_name( + x509.Name([x509.NameAttribute(x509.OID_COUNTRY_NAME, "US")]) + ) + .not_valid_before(datetime.datetime(2002, 1, 1, 12, 1)) + .not_valid_after(datetime.datetime(2030, 12, 31, 8, 30)) + .public_key(private_key.public_key()) + .serial_number(123) + .add_extension(PointExtension(Point(x=1, y=2)), critical=True) + .sign(private_key, hashes.SHA256()) + ) + + ext = cert.extensions.get_extension_for_oid( + x509.ObjectIdentifier("1.2.3.4") + ) + assert ext.critical is True + assert isinstance(ext.value, x509.UnrecognizedExtension) + assert ext.value.value == b"\x30\x06\x02\x01\x01\x02\x01\x02" + + custom = cert.extensions.get_extension_for_class(PointExtension) + assert custom.critical is True + assert isinstance(custom.value, PointExtension) + assert (custom.value.value.x, custom.value.value.y) == (1, 2) + def test_sign_without_private_key(self, rsa_key_2048: rsa.RSAPrivateKey): subject_private_key = rsa_key_2048 diff --git a/tests/x509/test_x509_ext.py b/tests/x509/test_x509_ext.py index c759f5c85d0f..e1fcb26ec32f 100644 --- a/tests/x509/test_x509_ext.py +++ b/tests/x509/test_x509_ext.py @@ -12,6 +12,7 @@ import pytest from cryptography import utils, x509 +from cryptography.hazmat import asn1 from cryptography.hazmat._oid import _OID_NAMES from cryptography.hazmat.bindings._rust import x509 as rust_x509 from cryptography.hazmat.primitives import hashes @@ -285,6 +286,196 @@ def test_public_bytes(self): assert ext2.public_bytes() == b"\x03\x02\x01" +@asn1.sequence +class _Point: + x: int + y: int + + +class _PointExtension(x509.CustomExtensionType[_Point]): + oid = x509.ObjectIdentifier("1.2.3.4") + + +class _RawExtension(x509.CustomExtensionType[bytes]): + oid = x509.ObjectIdentifier("1.2.3.5") + + +_POINT_DER = b"\x30\x06\x02\x01\x01\x02\x01\x02" + + +class TestCustomExtensionType: + def test_value(self): + point = _Point(x=1, y=2) + ext = _PointExtension(point) + assert ext.oid == x509.ObjectIdentifier("1.2.3.4") + assert ext.value is point + + def test_public_bytes(self): + assert _PointExtension(_Point(x=1, y=2)).public_bytes() == _POINT_DER + assert _RawExtension(b"abc").public_bytes() == b"\x04\x03abc" + + def test_eq(self): + point = _Point(x=1, y=2) + assert _PointExtension(point) == _PointExtension(point) + assert _RawExtension(b"abc") == _RawExtension(b"abc") + + def test_ne(self): + class _OtherRawExtension(x509.CustomExtensionType[bytes]): + oid = x509.ObjectIdentifier("1.2.3.5") + + ext = _RawExtension(b"abc") + assert ext != _RawExtension(b"abd") + assert ext != _OtherRawExtension(b"abc") + assert ext != x509.UnrecognizedExtension( + x509.ObjectIdentifier("1.2.3.5"), b"\x04\x03abc" + ) + assert ext != object() + + def test_hash(self): + assert hash(_RawExtension(b"abc")) == hash(_RawExtension(b"abc")) + assert hash(_RawExtension(b"abc")) != hash(_RawExtension(b"abd")) + + def test_repr(self): + assert repr(_RawExtension(b"abc")) == "<_RawExtension(value=b'abc')>" + + def test_subclass_without_type_parameter(self): + with pytest.raises(TypeError, match="parameterized"): + + class _Bad(x509.CustomExtensionType): + oid = x509.ObjectIdentifier("1.2.3.4") + + def test_subclass_with_type_variable(self): + T = typing.TypeVar("T") + with pytest.raises(TypeError, match="parameterized"): + + class _Bad(x509.CustomExtensionType[T]): + oid = x509.ObjectIdentifier("1.2.3.4") + + def test_subclass_multiple_inheritance(self): + class _Mixin: + pass + + with pytest.raises(TypeError, match="multiple inheritance"): + + class _Bad(_Mixin, x509.CustomExtensionType[bytes]): + oid = x509.ObjectIdentifier("1.2.3.4") + + def test_subclass_without_oid(self): + with pytest.raises(TypeError, match="oid"): + + class _Bad(x509.CustomExtensionType[bytes]): + pass + + def test_subclass_of_subclass(self): + class _Sub(_PointExtension): + pass + + exts = x509.Extensions( + [ + x509.Extension( + x509.ObjectIdentifier("1.2.3.4"), + False, + x509.UnrecognizedExtension( + x509.ObjectIdentifier("1.2.3.4"), _POINT_DER + ), + ) + ] + ) + ext = exts.get_extension_for_class(_Sub) + assert isinstance(ext.value, _Sub) + assert (ext.value.value.x, ext.value.value.y) == (1, 2) + + def test_get_extension_for_class(self): + unrecognized = x509.UnrecognizedExtension( + x509.ObjectIdentifier("1.2.3.4"), _POINT_DER + ) + exts = x509.Extensions( + [ + x509.Extension( + x509.ObjectIdentifier("1.2.3.5"), + False, + x509.UnrecognizedExtension( + x509.ObjectIdentifier("1.2.3.5"), b"\x04\x03abc" + ), + ), + x509.Extension( + x509.ObjectIdentifier("1.2.3.4"), True, unrecognized + ), + ] + ) + + ext = exts.get_extension_for_class(_PointExtension) + assert ext.oid == x509.ObjectIdentifier("1.2.3.4") + assert ext.critical is True + assert isinstance(ext.value, _PointExtension) + assert (ext.value.value.x, ext.value.value.y) == (1, 2) + # The Extensions object itself is unchanged. + assert ( + exts.get_extension_for_oid(x509.ObjectIdentifier("1.2.3.4")).value + == unrecognized + ) + + raw = exts.get_extension_for_class(_RawExtension) + assert raw.oid == x509.ObjectIdentifier("1.2.3.5") + assert raw.critical is False + assert raw.value == _RawExtension(b"abc") + + def test_get_extension_for_class_existing_instance(self): + ext = x509.Extension( + x509.ObjectIdentifier("1.2.3.4"), + False, + _PointExtension(_Point(x=1, y=2)), + ) + exts = x509.Extensions([ext]) + assert exts.get_extension_for_class(_PointExtension) is ext + + def test_get_extension_for_class_not_found(self): + exts = x509.Extensions( + [ + x509.Extension( + x509.ObjectIdentifier("1.2.3.5"), + False, + x509.UnrecognizedExtension( + x509.ObjectIdentifier("1.2.3.5"), b"\x04\x03abc" + ), + ) + ] + ) + with pytest.raises(x509.ExtensionNotFound) as exc: + exts.get_extension_for_class(_PointExtension) + + assert exc.value.oid == x509.ObjectIdentifier("1.2.3.4") + + def test_get_extension_for_class_invalid_der(self): + exts = x509.Extensions( + [ + x509.Extension( + x509.ObjectIdentifier("1.2.3.4"), + False, + x509.UnrecognizedExtension( + x509.ObjectIdentifier("1.2.3.4"), b"\x04\x03abc" + ), + ) + ] + ) + with pytest.raises(ValueError): + exts.get_extension_for_class(_PointExtension) + + def test_get_extension_for_class_only_parses_unrecognized(self): + # A custom class whose OID is one we parse natively doesn't match + # the natively parsed extension. + class _RawSubjectKeyIdentifier(x509.CustomExtensionType[bytes]): + oid = ExtensionOID.SUBJECT_KEY_IDENTIFIER + + cert = _load_cert( + os.path.join("x509", "PKITS_data", "certs", "GoodCACert.crt"), + x509.load_der_x509_certificate, + ) + cert.extensions.get_extension_for_class(x509.SubjectKeyIdentifier) + with pytest.raises(x509.ExtensionNotFound): + cert.extensions.get_extension_for_class(_RawSubjectKeyIdentifier) + + class TestCertificateIssuer: def test_iter_names(self): ci = x509.CertificateIssuer(