|
18 | 18 | from cryptography.hazmat.primitives.asymmetric import ed25519, padding, rsa
|
19 | 19 | from cryptography.hazmat.primitives.ciphers import algorithms
|
20 | 20 | from cryptography.hazmat.primitives.serialization import pkcs7
|
| 21 | +from cryptography.x509.verification import ( |
| 22 | + PolicyBuilder, |
| 23 | + Store, |
| 24 | + VerificationError, |
| 25 | +) |
21 | 26 | from tests.x509.test_x509 import _generate_ca_and_leaf
|
22 | 27 |
|
23 | 28 | from ...hazmat.primitives.fixtures_rsa import (
|
@@ -125,20 +130,129 @@ def test_load_pkcs7_empty_certificates(self):
|
125 | 130 |
|
126 | 131 | def _load_cert_key():
|
127 | 132 | key = load_vectors_from_file(
|
128 |
| - os.path.join("x509", "custom", "ca", "ca_key.pem"), |
| 133 | + os.path.join("pkcs7", "ca_key.pem"), |
129 | 134 | lambda pemfile: serialization.load_pem_private_key(
|
130 | 135 | pemfile.read(), None, unsafe_skip_rsa_key_validation=True
|
131 | 136 | ),
|
132 | 137 | mode="rb",
|
133 | 138 | )
|
134 | 139 | cert = load_vectors_from_file(
|
135 |
| - os.path.join("x509", "custom", "ca", "ca.pem"), |
| 140 | + os.path.join("pkcs7", "ca.pem"), |
136 | 141 | loader=lambda pemfile: x509.load_pem_x509_certificate(pemfile.read()),
|
137 | 142 | mode="rb",
|
138 | 143 | )
|
139 | 144 | return cert, key
|
140 | 145 |
|
141 | 146 |
|
| 147 | +class TestPKCS7VerifyCertificate: |
| 148 | + def test_verify_pkcs7_certificate(self): |
| 149 | + certificate, _ = _load_cert_key() |
| 150 | + ca_policy, ee_policy = pkcs7.pkcs7_x509_extension_policies() |
| 151 | + |
| 152 | + verifier = ( |
| 153 | + PolicyBuilder() |
| 154 | + .store(Store([certificate])) |
| 155 | + .extension_policies(ca_policy, ee_policy) |
| 156 | + .build_client_verifier() |
| 157 | + ) |
| 158 | + verifier.verify(certificate, []) |
| 159 | + |
| 160 | + @pytest.fixture(name="certificate_builder") |
| 161 | + def fixture_certificate_builder(self) -> x509.CertificateBuilder: |
| 162 | + certificate, private_key = _load_cert_key() |
| 163 | + return ( |
| 164 | + x509.CertificateBuilder() |
| 165 | + .serial_number(certificate.serial_number) |
| 166 | + .subject_name(certificate.subject) |
| 167 | + .issuer_name(certificate.issuer) |
| 168 | + .public_key(private_key.public_key()) |
| 169 | + .not_valid_before(certificate.not_valid_before) |
| 170 | + .not_valid_after(certificate.not_valid_after) |
| 171 | + ) |
| 172 | + |
| 173 | + def test_verify_pkcs7_certificate_wrong_bc(self, certificate_builder): |
| 174 | + certificate, private_key = _load_cert_key() |
| 175 | + |
| 176 | + # Add an invalid extension |
| 177 | + extension = x509.BasicConstraints(ca=True, path_length=None) |
| 178 | + certificate_builder = certificate_builder.add_extension( |
| 179 | + extension, True |
| 180 | + ) |
| 181 | + |
| 182 | + # Build the certificate |
| 183 | + pkcs7_certificate = certificate_builder.sign( |
| 184 | + private_key, certificate.signature_hash_algorithm, None |
| 185 | + ) |
| 186 | + |
| 187 | + # Verify the certificate |
| 188 | + self.verify_invalid_pkcs7_certificate(pkcs7_certificate) |
| 189 | + |
| 190 | + def test_verify_pkcs7_certificate_wrong_ku(self, certificate_builder): |
| 191 | + certificate, private_key = _load_cert_key() |
| 192 | + |
| 193 | + # Add an invalid extension |
| 194 | + extension = x509.KeyUsage( |
| 195 | + digital_signature=False, |
| 196 | + content_commitment=False, |
| 197 | + key_encipherment=True, |
| 198 | + data_encipherment=True, |
| 199 | + key_agreement=True, |
| 200 | + key_cert_sign=True, |
| 201 | + crl_sign=True, |
| 202 | + encipher_only=False, |
| 203 | + decipher_only=False, |
| 204 | + ) |
| 205 | + certificate_builder = certificate_builder.add_extension( |
| 206 | + extension, True |
| 207 | + ) |
| 208 | + |
| 209 | + # Build the certificate |
| 210 | + pkcs7_certificate = certificate_builder.sign( |
| 211 | + private_key, certificate.signature_hash_algorithm, None |
| 212 | + ) |
| 213 | + |
| 214 | + # Verify the certificate |
| 215 | + self.verify_invalid_pkcs7_certificate(pkcs7_certificate) |
| 216 | + |
| 217 | + def test_verify_pkcs7_certificate_wrong_eku(self, certificate_builder): |
| 218 | + certificate, private_key = _load_cert_key() |
| 219 | + |
| 220 | + # Add an invalid extension |
| 221 | + usages = [x509.ExtendedKeyUsageOID.CLIENT_AUTH] # type: ignore[attr-defined] |
| 222 | + extension = x509.ExtendedKeyUsage(usages) |
| 223 | + certificate_builder = certificate_builder.add_extension( |
| 224 | + extension, True |
| 225 | + ) |
| 226 | + |
| 227 | + # Add an invalid extension |
| 228 | + usages = [x509.ExtendedKeyUsageOID.CLIENT_AUTH] # type: ignore[attr-defined] |
| 229 | + extension = x509.ExtendedKeyUsage(usages) |
| 230 | + certificate_builder = certificate_builder.add_extension( |
| 231 | + extension, True |
| 232 | + ) |
| 233 | + |
| 234 | + # Build the certificate |
| 235 | + pkcs7_certificate = certificate_builder.sign( |
| 236 | + private_key, certificate.signature_hash_algorithm, None |
| 237 | + ) |
| 238 | + |
| 239 | + # Verify the certificate |
| 240 | + self.verify_invalid_pkcs7_certificate(pkcs7_certificate) |
| 241 | + |
| 242 | + @staticmethod |
| 243 | + def verify_invalid_pkcs7_certificate(certificate: x509.Certificate): |
| 244 | + ca_policy, ee_policy = pkcs7.pkcs7_x509_extension_policies() |
| 245 | + verifier = ( |
| 246 | + PolicyBuilder() |
| 247 | + .store(Store([certificate])) |
| 248 | + .extension_policies(ca_policy, ee_policy) |
| 249 | + .build_client_verifier() |
| 250 | + ) |
| 251 | + |
| 252 | + with pytest.raises(VerificationError): |
| 253 | + verifier.verify(certificate, []) |
| 254 | + |
| 255 | + |
142 | 256 | @pytest.mark.supported(
|
143 | 257 | only_if=lambda backend: backend.pkcs7_supported(),
|
144 | 258 | skip_message="Requires OpenSSL with PKCS7 support",
|
|
0 commit comments