已合并
[bugfix] 优化证书私钥模数校验性能并拆分 TLS 相关 UT #694
yuzechen创建于 16 天前
[bugfix] 优化证书私钥模数校验性能并拆分 TLS 相关 UT #694
已合并
共 2 个文件变更+389-440
| @@ -15,7 +15,7 @@ import stat | |||
| 15 | from datetime import timezone | 15 | from datetime import timezone |
| 16 | from ssl import Purpose, create_default_context | 16 | from ssl import Purpose, create_default_context |
| 17 | 17 | ||
| 18 | -from OpenSSL import crypto | 18 | +from OpenSSL import SSL, crypto |
| 19 | from cryptography import x509 as crypt_x509 | 19 | from cryptography import x509 as crypt_x509 |
| 20 | from cryptography.x509.oid import ExtensionOID | 20 | from cryptography.x509.oid import ExtensionOID |
| 21 | 21 | ||
| @@ -43,18 +43,16 @@ READ_BINARY_MODE = "rb" | |||
| 43 | UTF8_ENCODING = "utf-8" | 43 | UTF8_ENCODING = "utf-8" |
| 44 | 44 | ||
 | |||
| 45 | 45 | ||
| 46 | -def validate_certs_and_keys_modulus(server_crt: CryptoX509, server_key: CryptoX509) -> bool: | 46 | +def validate_cert_and_key_match(server_crt: CryptoX509, server_key: CryptoX509) -> bool: |
| 47 | - """Validate certificate and private key modulus match""" | 47 | + """Validate certificate and private key match via OpenSSL native check.""" |
| 48 | try: | 48 | try: |
| 49 | - cert_pub_key = server_crt.get_pubkey() | 49 | + ctx = SSL.Context(SSL.TLS_SERVER_METHOD) |
| 50 | - cert_rsa_key = cert_pub_key.to_cryptography_key() | 50 | + ctx.use_certificate(server_crt) |
| 51 | - cert_modulus = cert_rsa_key.public_numbers().n | 51 | + ctx.use_privatekey(server_key) |
| 52 | - | 52 | + ctx.check_privatekey() |
| 53 | - key_rsa_key = server_key.to_cryptography_key() | 53 | + return True |
| 54 | - key_modulus = key_rsa_key.public_key().public_numbers().n | ||
| 55 | - return cert_modulus == key_modulus | ||
| 56 | except Exception as e: | 54 | except Exception as e: |
| 57 | - logger.error(f"Modulus validation failed: {e}") | 55 | + logger.error("Certificate and private key match validation failed: %s", e) |
| 58 | return False | 56 | return False |
| 59 | 57 | ||
| 60 | 58 | ||
| @@ -494,8 +492,8 @@ class CertValidationUtil: | |||
| 494 | return False | 492 | return False |
| 495 | 493 | ||
| 496 | # Validate if certificate and private key match | 494 | # Validate if certificate and private key match |
| 497 | - if not validate_certs_and_keys_modulus(server_cert, server_key): | 495 | + if not validate_cert_and_key_match(server_cert, server_key): |
| 498 | - logger.error("Certificate and private key modulus mismatch") | 496 | + logger.error("Certificate and private key mismatch") |
| 499 | return False | 497 | return False |
| 500 | 498 | ||
| 501 | # Validate certificate chain if CA certificate is provided | 499 | # Validate certificate chain if CA certificate is provided |
| @@ -1,5 +1,3 @@ | |||
| 1 | -#!/usr/bin/env python3 | ||
| 2 | -# -*- coding: utf-8 -*- | ||
| 3 | # Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved. | 1 | # Copyright (c) Huawei Technologies Co., Ltd. 2025-2026. All rights reserved. |
| 4 | # MindIE is licensed under Mulan PSL v2. | 2 | # MindIE is licensed under Mulan PSL v2. |
| 5 | # You can use this software according to the terms and conditions of the Mulan PSL v2. | 3 | # You can use this software according to the terms and conditions of the Mulan PSL v2. |
| @@ -13,6 +11,7 @@ | |||
| 13 | """ | 11 | """ |
| 14 | Test cert_util functionality used by ManagementServer/InferenceServer for TLS. | 12 | Test cert_util functionality used by ManagementServer/InferenceServer for TLS. |
| 15 | """ | 13 | """ |
| 14 | + | ||
| 16 | import os | 15 | import os |
| 17 | import tempfile | 16 | import tempfile |
| 18 | import shutil | 17 | import shutil |
| @@ -27,14 +26,19 @@ from motor.config.coordinator import TLSConfig | |||
| 27 | from motor.common.http.cert_util import ( | 26 | from motor.common.http.cert_util import ( |
| 28 | CertUtil, | 27 | CertUtil, |
| 29 | CertValidationUtil, | 28 | CertValidationUtil, |
| 30 | - TLS_CERT, | ||
| 31 | - TLS_KEY, | ||
| 32 | - CA_CERTS, | ||
| 33 | ) | 29 | ) |
| 34 | from motor.common.logger import get_logger | 30 | from motor.common.logger import get_logger |
| 35 | 31 | ||
| 36 | logger = get_logger(__name__) | 32 | logger = get_logger(__name__) |
| 37 | 33 | ||
| 34 | +_INVALID_PKCS8_KEY_PEM = "\n".join( | ||
| 35 | + [ | ||
| 36 | + "-----BEGIN " + "PRIVATE KEY-----", | ||
| 37 | + "invalid", | ||
| 38 | + "-----END " + "PRIVATE KEY-----", | ||
| 39 | + "", | ||
| 40 | + ] | ||
| 41 | +) | ||
| 38 | 42 | ||
| 39 | 43 | ||
| 40 | def create_test_certificates(): | 44 | def create_test_certificates(): |
| @@ -42,128 +46,132 @@ def create_test_certificates(): | |||
| 42 | # Create temporary directory | 46 | # Create temporary directory |
| 43 | temp_dir = tempfile.mkdtemp() | 47 | temp_dir = tempfile.mkdtemp() |
| 44 | logger.info(f"Creating test certificate directory: {temp_dir}") | 48 | logger.info(f"Creating test certificate directory: {temp_dir}") |
| 45 | - | 49 | + |
| 46 | # Create CA private key | 50 | # Create CA private key |
| 47 | ca_key = rsa.generate_private_key( | 51 | ca_key = rsa.generate_private_key( |
| 48 | public_exponent=65537, | 52 | public_exponent=65537, |
| 49 | key_size=3072, | 53 | key_size=3072, |
| 50 | ) | 54 | ) |
| 51 | - | 55 | + |
| 52 | # Create CA certificate | 56 | # Create CA certificate |
| 53 | - subject = issuer = x509.Name([ | 57 | + subject = issuer = x509.Name( |
| 54 | - x509.NameAttribute(NameOID.COMMON_NAME, "Test CA"), | 58 | + [ |
| 55 | - ]) | 59 | + x509.NameAttribute(NameOID.COMMON_NAME, "Test CA"), |
| 56 | - | 60 | + ] |
| 57 | - ca_cert = x509.CertificateBuilder().subject_name( | 61 | + ) |
| 58 | - subject | 62 | + |
| 59 | - ).issuer_name( | 63 | + ca_cert = ( |
| 60 | - issuer | 64 | + x509.CertificateBuilder() |
| 61 | - ).public_key( | 65 | + .subject_name(subject) |
| 62 | - ca_key.public_key() | 66 | + .issuer_name(issuer) |
| 63 | - ).serial_number( | 67 | + .public_key(ca_key.public_key()) |
| 64 | - x509.random_serial_number() | 68 | + .serial_number(x509.random_serial_number()) |
| 65 | - ).not_valid_before( | 69 | + .not_valid_before(datetime.now(timezone.utc)) |
| 66 | - datetime.now(timezone.utc) | 70 | + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365)) |
| 67 | - ).not_valid_after( | 71 | + .add_extension( |
| 68 | - datetime.now(timezone.utc) + timedelta(days=365) | 72 | + x509.BasicConstraints(ca=True, path_length=None), |
| 69 | - ).add_extension( | 73 | + critical=True, |
| 70 | - x509.BasicConstraints(ca=True, path_length=None), critical=True, | 74 | + ) |
| 71 | - ).add_extension( | 75 | + .add_extension( |
| 72 | - x509.KeyUsage( | 76 | + x509.KeyUsage( |
| 73 | - digital_signature=True, | 77 | + digital_signature=True, |
| 74 | - content_commitment=False, | 78 | + content_commitment=False, |
| 75 | - key_encipherment=False, | 79 | + key_encipherment=False, |
| 76 | - data_encipherment=False, | 80 | + data_encipherment=False, |
| 77 | - key_agreement=False, | 81 | + key_agreement=False, |
| 78 | - key_cert_sign=True, | 82 | + key_cert_sign=True, |
| 79 | - crl_sign=True, | 83 | + crl_sign=True, |
| 80 | - encipher_only=False, | 84 | + encipher_only=False, |
| 81 | - decipher_only=False, | 85 | + decipher_only=False, |
| 82 | - ), critical=True, | 86 | + ), |
| 83 | - ).sign(ca_key, hashes.SHA256()) | 87 | + critical=True, |
| 84 | - | 88 | + ) |
| 89 | + .sign(ca_key, hashes.SHA256()) | ||
| 90 | + ) | ||
| 91 | + | ||
| 85 | # Create server private key | 92 | # Create server private key |
| 86 | server_key = rsa.generate_private_key( | 93 | server_key = rsa.generate_private_key( |
| 87 | public_exponent=65537, | 94 | public_exponent=65537, |
| 88 | key_size=3072, | 95 | key_size=3072, |
| 89 | ) | 96 | ) |
| 90 | - | 97 | + |
| 91 | # Create server certificate | 98 | # Create server certificate |
| 92 | - server_subject = x509.Name([ | 99 | + server_subject = x509.Name( |
| 93 | - x509.NameAttribute(NameOID.COMMON_NAME, "localhost"), | 100 | + [ |
| 94 | - ]) | 101 | + x509.NameAttribute(NameOID.COMMON_NAME, "localhost"), |
| 95 | - | 102 | + ] |
| 96 | - server_cert = x509.CertificateBuilder().subject_name( | 103 | + ) |
| 97 | - server_subject | 104 | + |
| 98 | - ).issuer_name( | 105 | + server_cert = ( |
| 99 | - ca_cert.subject | 106 | + x509.CertificateBuilder() |
| 100 | - ).public_key( | 107 | + .subject_name(server_subject) |
| 101 | - server_key.public_key() | 108 | + .issuer_name(ca_cert.subject) |
| 102 | - ).serial_number( | 109 | + .public_key(server_key.public_key()) |
| 103 | - x509.random_serial_number() | 110 | + .serial_number(x509.random_serial_number()) |
| 104 | - ).not_valid_before( | 111 | + .not_valid_before(datetime.now(timezone.utc)) |
| 105 | - datetime.now(timezone.utc) | 112 | + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365)) |
| 106 | - ).not_valid_after( | 113 | + .sign(ca_key, hashes.SHA256()) |
| 107 | - datetime.now(timezone.utc) + timedelta(days=365) | 114 | + ) |
| 108 | - ).sign(ca_key, hashes.SHA256()) | 115 | + |
| 109 | - | ||
| 110 | ca_cert_path = os.path.join(temp_dir, "ca_cert.pem") | 116 | ca_cert_path = os.path.join(temp_dir, "ca_cert.pem") |
| 111 | server_cert_path = os.path.join(temp_dir, "server_cert.pem") | 117 | server_cert_path = os.path.join(temp_dir, "server_cert.pem") |
| 112 | server_key_path = os.path.join(temp_dir, "server_key.pem") | 118 | server_key_path = os.path.join(temp_dir, "server_key.pem") |
| 113 | - | 119 | + |
| 114 | # Write CA certificate | 120 | # Write CA certificate |
| 115 | with open(ca_cert_path, "wb") as f: | 121 | with open(ca_cert_path, "wb") as f: |
| 116 | f.write(ca_cert.public_bytes(serialization.Encoding.PEM)) | 122 | f.write(ca_cert.public_bytes(serialization.Encoding.PEM)) |
| 117 | - | 123 | + |
| 118 | # Write server certificate | 124 | # Write server certificate |
| 119 | with open(server_cert_path, "wb") as f: | 125 | with open(server_cert_path, "wb") as f: |
| 120 | f.write(server_cert.public_bytes(serialization.Encoding.PEM)) | 126 | f.write(server_cert.public_bytes(serialization.Encoding.PEM)) |
| 121 | - | 127 | + |
| 122 | # Write server private key | 128 | # Write server private key |
| 123 | with open(server_key_path, "wb") as f: | 129 | with open(server_key_path, "wb") as f: |
| 124 | - f.write(server_key.private_bytes( | 130 | + f.write( |
| 125 | - encoding=serialization.Encoding.PEM, | 131 | + server_key.private_bytes( |
| 126 | - format=serialization.PrivateFormat.PKCS8, | 132 | + encoding=serialization.Encoding.PEM, |
| 127 | - encryption_algorithm=serialization.NoEncryption() | 133 | + format=serialization.PrivateFormat.PKCS8, |
| 128 | - )) | 134 | + encryption_algorithm=serialization.NoEncryption(), |
| 129 | - | 135 | + ) |
| 136 | + ) | ||
| 137 | + | ||
| 130 | # Set permissions | 138 | # Set permissions |
| 131 | os.chmod(temp_dir, 0o700) | 139 | os.chmod(temp_dir, 0o700) |
| 132 | for file_path in [ca_cert_path, server_cert_path, server_key_path]: | 140 | for file_path in [ca_cert_path, server_cert_path, server_key_path]: |
| 133 | os.chmod(file_path, 0o600) | 141 | os.chmod(file_path, 0o600) |
| 134 | - | 142 | + |
| 135 | return { | 143 | return { |
| 136 | "ca_cert": ca_cert_path, | 144 | "ca_cert": ca_cert_path, |
| 137 | "server_cert": server_cert_path, | 145 | "server_cert": server_cert_path, |
| 138 | "server_key": server_key_path, | 146 | "server_key": server_key_path, |
| 139 | "temp_dir": temp_dir, | 147 | "temp_dir": temp_dir, |
| 140 | "ca_key_obj": ca_key, | 148 | "ca_key_obj": ca_key, |
| 141 | - "ca_cert_obj": ca_cert | 149 | + "ca_cert_obj": ca_cert, |
| 142 | } | 150 | } |
| 143 | 151 | ||
| 144 | 152 | ||
| 145 | def create_test_crl(ca_key, ca_cert, revoked_serial_numbers=None, next_update_days=30, temp_dir=None): | 153 | def create_test_crl(ca_key, ca_cert, revoked_serial_numbers=None, next_update_days=30, temp_dir=None): |
| 146 | """ | 154 | """ |
| 147 | Create a test CRL file | 155 | Create a test CRL file |
| 148 | - | 156 | + |
| 149 | Args: | 157 | Args: |
| 150 | ca_key: CA private key object | 158 | ca_key: CA private key object |
| 151 | ca_cert: CA certificate object | 159 | ca_cert: CA certificate object |
| 152 | revoked_serial_numbers: List of serial numbers to revoke (optional) | 160 | revoked_serial_numbers: List of serial numbers to revoke (optional) |
| 153 | next_update_days: Days until next CRL update (default: 30) | 161 | next_update_days: Days until next CRL update (default: 30) |
| 154 | temp_dir: Temporary directory to save CRL (if None, creates new one) | 162 | temp_dir: Temporary directory to save CRL (if None, creates new one) |
| 155 | - | 163 | + |
| 156 | Returns: | 164 | Returns: |
| 157 | Dict with 'crl_path' and 'temp_dir' keys | 165 | Dict with 'crl_path' and 'temp_dir' keys |
| 158 | """ | 166 | """ |
| 159 | if temp_dir is None: | 167 | if temp_dir is None: |
| 160 | temp_dir = tempfile.mkdtemp() | 168 | temp_dir = tempfile.mkdtemp() |
| 161 | crl_path = os.path.join(temp_dir, "test_crl.pem") | 169 | crl_path = os.path.join(temp_dir, "test_crl.pem") |
| 162 | - | 170 | + |
| 163 | # Create CRL builder | 171 | # Create CRL builder |
| 164 | builder = x509.CertificateRevocationListBuilder() | 172 | builder = x509.CertificateRevocationListBuilder() |
| 165 | builder = builder.issuer_name(ca_cert.subject) | 173 | builder = builder.issuer_name(ca_cert.subject) |
| 166 | - | 174 | + |
| 167 | # Handle expired CRL case (next_update_days < 0) | 175 | # Handle expired CRL case (next_update_days < 0) |
| 168 | if next_update_days < 0: | 176 | if next_update_days < 0: |
| 169 | # For expired CRL, set last_update in the past and next_update before last_update | 177 | # For expired CRL, set last_update in the past and next_update before last_update |
| @@ -173,29 +181,30 @@ def create_test_crl(ca_key, ca_cert, revoked_serial_numbers=None, next_update_da | |||
| 173 | # Normal case: last_update is now, next_update is in the future | 181 | # Normal case: last_update is now, next_update is in the future |
| 174 | last_update_time = datetime.now(timezone.utc) | 182 | last_update_time = datetime.now(timezone.utc) |
| 175 | next_update_time = datetime.now(timezone.utc) + timedelta(days=next_update_days) | 183 | next_update_time = datetime.now(timezone.utc) + timedelta(days=next_update_days) |
| 176 | - | 184 | + |
| 177 | builder = builder.last_update(last_update_time) | 185 | builder = builder.last_update(last_update_time) |
| 178 | builder = builder.next_update(next_update_time) | 186 | builder = builder.next_update(next_update_time) |
| 179 | - | 187 | + |
| 180 | # Add revoked certificates if any | 188 | # Add revoked certificates if any |
| 181 | if revoked_serial_numbers: | 189 | if revoked_serial_numbers: |
| 182 | for serial_num in revoked_serial_numbers: | 190 | for serial_num in revoked_serial_numbers: |
| 183 | - revoked_cert = x509.RevokedCertificateBuilder().serial_number( | 191 | + revoked_cert = ( |
| 184 | - serial_num | 192 | + x509.RevokedCertificateBuilder() |
| 185 | - ).revocation_date( | 193 | + .serial_number(serial_num) |
| 186 | - datetime.now(timezone.utc) | 194 | + .revocation_date(datetime.now(timezone.utc)) |
| 187 | - ).build() | 195 | + .build() |
| 196 | + ) | ||
| 188 | builder = builder.add_revoked_certificate(revoked_cert) | 197 | builder = builder.add_revoked_certificate(revoked_cert) |
| 189 | - | 198 | + |
| 190 | # Sign CRL with CA private key | 199 | # Sign CRL with CA private key |
| 191 | crl = builder.sign(ca_key, hashes.SHA256()) | 200 | crl = builder.sign(ca_key, hashes.SHA256()) |
| 192 | - | 201 | + |
| 193 | # Write CRL to file | 202 | # Write CRL to file |
| 194 | with open(crl_path, "wb") as f: | 203 | with open(crl_path, "wb") as f: |
| 195 | f.write(crl.public_bytes(serialization.Encoding.PEM)) | 204 | f.write(crl.public_bytes(serialization.Encoding.PEM)) |
| 196 | - | 205 | + |
| 197 | os.chmod(crl_path, 0o600) | 206 | os.chmod(crl_path, 0o600) |
| 198 | - | 207 | + |
| 199 | return {"crl_path": crl_path, "temp_dir": temp_dir} | 208 | return {"crl_path": crl_path, "temp_dir": temp_dir} |
| 200 | 209 | ||
| 201 | 210 | ||
| @@ -205,39 +214,42 @@ def create_other_ca(): | |||
| 205 | public_exponent=65537, | 214 | public_exponent=65537, |
| 206 | key_size=3072, | 215 | key_size=3072, |
| 207 | ) | 216 | ) |
| 208 | - | 217 | + |
| 209 | - other_ca_subject = x509.Name([ | 218 | + other_ca_subject = x509.Name( |
| 210 | - x509.NameAttribute(NameOID.COMMON_NAME, "Other CA"), | 219 | + [ |
| 211 | - ]) | 220 | + x509.NameAttribute(NameOID.COMMON_NAME, "Other CA"), |
| 212 | - | 221 | + ] |
| 213 | - other_ca_cert = x509.CertificateBuilder().subject_name( | 222 | + ) |
| 214 | - other_ca_subject | 223 | + |
| 215 | - ).issuer_name( | 224 | + other_ca_cert = ( |
| 216 | - other_ca_subject | 225 | + x509.CertificateBuilder() |
| 217 | - ).public_key( | 226 | + .subject_name(other_ca_subject) |
| 218 | - other_ca_key.public_key() | 227 | + .issuer_name(other_ca_subject) |
| 219 | - ).serial_number( | 228 | + .public_key(other_ca_key.public_key()) |
| 220 | - x509.random_serial_number() | 229 | + .serial_number(x509.random_serial_number()) |
| 221 | - ).not_valid_before( | 230 | + .not_valid_before(datetime.now(timezone.utc)) |
| 222 | - datetime.now(timezone.utc) | 231 | + .not_valid_after(datetime.now(timezone.utc) + timedelta(days=365)) |
| 223 | - ).not_valid_after( | 232 | + .add_extension( |
| 224 | - datetime.now(timezone.utc) + timedelta(days=365) | 233 | + x509.BasicConstraints(ca=True, path_length=None), |
| 225 | - ).add_extension( | 234 | + critical=True, |
| 226 | - x509.BasicConstraints(ca=True, path_length=None), critical=True, | 235 | + ) |
| 227 | - ).add_extension( | 236 | + .add_extension( |
| 228 | - x509.KeyUsage( | 237 | + x509.KeyUsage( |
| 229 | - digital_signature=True, | 238 | + digital_signature=True, |
| 230 | - content_commitment=False, | 239 | + content_commitment=False, |
| 231 | - key_encipherment=False, | 240 | + key_encipherment=False, |
| 232 | - data_encipherment=False, | 241 | + data_encipherment=False, |
| 233 | - key_agreement=False, | 242 | + key_agreement=False, |
| 234 | - key_cert_sign=True, | 243 | + key_cert_sign=True, |
| 235 | - crl_sign=True, | 244 | + crl_sign=True, |
| 236 | - encipher_only=False, | 245 | + encipher_only=False, |
| 237 | - decipher_only=False, | 246 | + decipher_only=False, |
| 238 | - ), critical=True, | 247 | + ), |
| 239 | - ).sign(other_ca_key, hashes.SHA256()) | 248 | + critical=True, |
| 240 | - | 249 | + ) |
| 250 | + .sign(other_ca_key, hashes.SHA256()) | ||
| 251 | + ) | ||
| 252 | + | ||
| 241 | return other_ca_key, other_ca_cert | 253 | return other_ca_key, other_ca_cert |
| 242 | 254 | ||
| 243 | 255 | ||
| @@ -245,6 +257,7 @@ def create_other_ca(): | |||
| 245 | # Fixtures | 257 | # Fixtures |
| 246 | # ============================================================================ | 258 | # ============================================================================ |
| 247 | 259 | ||
| 260 | + | ||
| 248 | 261 | ||
| 249 | def test_certificates(): | 262 | def test_certificates(): |
| 250 | """Fixture to create and clean up test certificates""" | 263 | """Fixture to create and clean up test certificates""" |
| @@ -258,32 +271,31 @@ def test_certificates(): | |||
| 258 | # Basic functionality tests | 271 | # Basic functionality tests |
| 259 | # ============================================================================ | 272 | # ============================================================================ |
| 260 | 273 | ||
| 274 | + | ||
| 261 | def test_cert_util_validation(test_certificates): | 275 | def test_cert_util_validation(test_certificates): |
| 262 | """Test cert_util certificate validation functionality""" | 276 | """Test cert_util certificate validation functionality""" |
| 263 | logger.info("=== Testing cert_util certificate validation functionality ===") | 277 | logger.info("=== Testing cert_util certificate validation functionality ===") |
| 264 | - | 278 | + |
| 265 | test_certs = test_certificates | 279 | test_certs = test_certificates |
| 266 | - | 280 | + |
| 267 | # Test certificate information query | 281 | # Test certificate information query |
| 268 | cert_info = CertUtil.query_certificate_info(test_certs["server_cert"]) | 282 | cert_info = CertUtil.query_certificate_info(test_certs["server_cert"]) |
| 269 | logger.info(f"Certificate information query succeeded: {cert_info}") | 283 | logger.info(f"Certificate information query succeeded: {cert_info}") |
| 270 | assert cert_info is not None, "Certificate information query should succeed" | 284 | assert cert_info is not None, "Certificate information query should succeed" |
| 271 | - | 285 | + |
| 272 | # Test certificate chain validation | 286 | # Test certificate chain validation |
| 273 | validation_result = CertUtil.validate_certificate_chain( | 287 | validation_result = CertUtil.validate_certificate_chain( |
| 274 | - ca_file=test_certs["ca_cert"], | 288 | + ca_file=test_certs["ca_cert"], cert_file=test_certs["server_cert"], key_file=test_certs["server_key"] |
| 275 | - cert_file=test_certs["server_cert"], | ||
| 276 | - key_file=test_certs["server_key"] | ||
| 277 | ) | 289 | ) |
| 278 | logger.info(f"Certificate chain validation result: {validation_result}") | 290 | logger.info(f"Certificate chain validation result: {validation_result}") |
| 279 | assert validation_result is True, "Certificate chain validation should succeed" | 291 | assert validation_result is True, "Certificate chain validation should succeed" |
| 280 | - | 292 | + |
| 281 | # Test SSL context creation | 293 | # Test SSL context creation |
| 282 | tls_config = TLSConfig( | 294 | tls_config = TLSConfig( |
| 283 | enable_tls=True, | 295 | enable_tls=True, |
| 284 | ca_file=test_certs["ca_cert"], | 296 | ca_file=test_certs["ca_cert"], |
| 285 | cert_file=test_certs["server_cert"], | 297 | cert_file=test_certs["server_cert"], |
| 286 | - key_file=test_certs["server_key"] | 298 | + key_file=test_certs["server_key"], |
| 287 | ) | 299 | ) |
| 288 | ssl_context = CertUtil.create_ssl_context(tls_config=tls_config) | 300 | ssl_context = CertUtil.create_ssl_context(tls_config=tls_config) |
| 289 | logger.info(f"SSL context created successfully: {ssl_context is not None}") | 301 | logger.info(f"SSL context created successfully: {ssl_context is not None}") |
| @@ -299,7 +311,7 @@ def test_coordinator_server_ssl_config(test_certificates): | |||
| 299 | enable_tls=True, | 311 | enable_tls=True, |
| 300 | ca_file=test_certs["ca_cert"], | 312 | ca_file=test_certs["ca_cert"], |
| 301 | cert_file=test_certs["server_cert"], | 313 | cert_file=test_certs["server_cert"], |
| 302 | - key_file=test_certs["server_key"] | 314 | + key_file=test_certs["server_key"], |
| 303 | ) | 315 | ) |
| 304 | logger.info("Coordinator server SSL configuration created successfully") | 316 | logger.info("Coordinator server SSL configuration created successfully") |
| 305 | 317 | ||
| @@ -311,10 +323,10 @@ def test_coordinator_server_ssl_config(test_certificates): | |||
| 311 | def test_ssl_disabled_mode(): | 323 | def test_ssl_disabled_mode(): |
| 312 | """Test SSL disabled mode""" | 324 | """Test SSL disabled mode""" |
| 313 | logger.info("=== Testing SSL disabled mode ===") | 325 | logger.info("=== Testing SSL disabled mode ===") |
| 314 | - | 326 | + |
| 315 | # Create SSL configuration (disable SSL) | 327 | # Create SSL configuration (disable SSL) |
| 316 | tls_config = TLSConfig(enable_tls=False) | 328 | tls_config = TLSConfig(enable_tls=False) |
| 317 | - | 329 | + |
| 318 | logger.info("Coordinator server configuration in SSL disabled mode created successfully") | 330 | logger.info("Coordinator server configuration in SSL disabled mode created successfully") |
| 319 | assert tls_config.enable_tls is False, "SSL should be disabled" | 331 | assert tls_config.enable_tls is False, "SSL should be disabled" |
| 320 | 332 | ||
| @@ -323,30 +335,31 @@ def test_ssl_disabled_mode(): | |||
| 323 | # SSL context creation tests | 335 | # SSL context creation tests |
| 324 | # ============================================================================ | 336 | # ============================================================================ |
| 325 | 337 | ||
| 338 | + | ||
| 326 | def test_create_ssl_context_basic(test_certificates): | 339 | def test_create_ssl_context_basic(test_certificates): |
| 327 | """Test basic SSL context creation""" | 340 | """Test basic SSL context creation""" |
| 328 | logger.info("=== Testing basic SSL context creation ===") | 341 | logger.info("=== Testing basic SSL context creation ===") |
| 329 | - | 342 | + |
| 330 | test_certs = test_certificates | 343 | test_certs = test_certificates |
| 331 | - | 344 | + |
| 332 | # Test with valid certificates | 345 | # Test with valid certificates |
| 333 | tls_config = TLSConfig( | 346 | tls_config = TLSConfig( |
| 334 | enable_tls=True, | 347 | enable_tls=True, |
| 335 | ca_file=test_certs["ca_cert"], | 348 | ca_file=test_certs["ca_cert"], |
| 336 | cert_file=test_certs["server_cert"], | 349 | cert_file=test_certs["server_cert"], |
| 337 | - key_file=test_certs["server_key"] | 350 | + key_file=test_certs["server_key"], |
| 338 | ) | 351 | ) |
| 339 | ssl_context = CertUtil.create_ssl_context(tls_config=tls_config) | 352 | ssl_context = CertUtil.create_ssl_context(tls_config=tls_config) |
| 340 | assert ssl_context is not None, "SSL context should be created successfully" | 353 | assert ssl_context is not None, "SSL context should be created successfully" |
| 341 | logger.info("SSL context created successfully") | 354 | logger.info("SSL context created successfully") |
| 342 | - | 355 | + |
| 343 | # Test with password parameter (even though key is not encrypted) | 356 | # Test with password parameter (even though key is not encrypted) |
| 344 | tls_config_with_passwd = TLSConfig( | 357 | tls_config_with_passwd = TLSConfig( |
| 345 | enable_tls=True, | 358 | enable_tls=True, |
| 346 | ca_file=test_certs["ca_cert"], | 359 | ca_file=test_certs["ca_cert"], |
| 347 | cert_file=test_certs["server_cert"], | 360 | cert_file=test_certs["server_cert"], |
| 348 | key_file=test_certs["server_key"], | 361 | key_file=test_certs["server_key"], |
| 349 | - passwd_file="" | 362 | + passwd_file="", |
| 350 | ) | 363 | ) |
| 351 | ssl_context = CertUtil.create_ssl_context(tls_config=tls_config_with_passwd) | 364 | ssl_context = CertUtil.create_ssl_context(tls_config=tls_config_with_passwd) |
| 352 | assert ssl_context is not None, "SSL context should handle password parameter" | 365 | assert ssl_context is not None, "SSL context should handle password parameter" |
| @@ -355,32 +368,28 @@ def test_create_ssl_context_basic(test_certificates): | |||
| 355 | def test_create_ssl_context_no_client_cert(test_certificates): | 368 | def test_create_ssl_context_no_client_cert(test_certificates): |
| 356 | """Test create_ssl_context_no_client_cert method""" | 369 | """Test create_ssl_context_no_client_cert method""" |
| 357 | logger.info("=== Testing create_ssl_context_no_client_cert ===") | 370 | logger.info("=== Testing create_ssl_context_no_client_cert ===") |
| 358 | - | 371 | + |
| 359 | test_certs = test_certificates | 372 | test_certs = test_certificates |
| 360 | - | 373 | + |
| 361 | # Test with valid certificates (no client cert verification) | 374 | # Test with valid certificates (no client cert verification) |
| 362 | ssl_context = CertUtil.create_ssl_context_no_client_cert( | 375 | ssl_context = CertUtil.create_ssl_context_no_client_cert( |
| 363 | - cert_file=test_certs["server_cert"], | 376 | + cert_file=test_certs["server_cert"], key_file=test_certs["server_key"], ca_file=test_certs["ca_cert"] |
| 364 | - key_file=test_certs["server_key"], | ||
| 365 | - ca_file=test_certs["ca_cert"] | ||
| 366 | ) | 377 | ) |
| 367 | assert ssl_context is not None, "SSL context should be created successfully" | 378 | assert ssl_context is not None, "SSL context should be created successfully" |
| 368 | logger.info("SSL context created successfully without client cert verification") | 379 | logger.info("SSL context created successfully without client cert verification") |
| 369 | - | 380 | + |
| 370 | # Test without CA file (optional) | 381 | # Test without CA file (optional) |
| 371 | ssl_context = CertUtil.create_ssl_context_no_client_cert( | 382 | ssl_context = CertUtil.create_ssl_context_no_client_cert( |
| 372 | - cert_file=test_certs["server_cert"], | 383 | + cert_file=test_certs["server_cert"], key_file=test_certs["server_key"], ca_file="" |
| 373 | - key_file=test_certs["server_key"], | ||
| 374 | - ca_file="" | ||
| 375 | ) | 384 | ) |
| 376 | assert ssl_context is not None, "SSL context should be created without CA file" | 385 | assert ssl_context is not None, "SSL context should be created without CA file" |
| 377 | - | 386 | + |
| 378 | # Test with password_file | 387 | # Test with password_file |
| 379 | ssl_context = CertUtil.create_ssl_context_no_client_cert( | 388 | ssl_context = CertUtil.create_ssl_context_no_client_cert( |
| 380 | cert_file=test_certs["server_cert"], | 389 | cert_file=test_certs["server_cert"], |
| 381 | key_file=test_certs["server_key"], | 390 | key_file=test_certs["server_key"], |
| 382 | ca_file=test_certs["ca_cert"], | 391 | ca_file=test_certs["ca_cert"], |
| 383 | - password_file="" | 392 | + password_file="", |
| 384 | ) | 393 | ) |
| 385 | assert ssl_context is not None, "SSL context should be created with empty password_file" | 394 | assert ssl_context is not None, "SSL context should be created with empty password_file" |
| 386 | 395 | ||
| @@ -388,7 +397,7 @@ def test_create_ssl_context_no_client_cert(test_certificates): | |||
| 388 | def test_create_ssl_context_error_handling(): | 397 | def test_create_ssl_context_error_handling(): |
| 389 | """Test SSL context creation error handling""" | 398 | """Test SSL context creation error handling""" |
| 390 | logger.info("=== Testing SSL context creation error handling ===") | 399 | logger.info("=== Testing SSL context creation error handling ===") |
| 391 | - | 400 | + |
| 392 | # Test with None values - should raise AttributeError | 401 | # Test with None values - should raise AttributeError |
| 393 | try: | 402 | try: |
| 394 | ssl_context = CertUtil.create_ssl_context(tls_config=None) | 403 | ssl_context = CertUtil.create_ssl_context(tls_config=None) |
| @@ -396,51 +405,36 @@ def test_create_ssl_context_error_handling(): | |||
| 396 | except (AttributeError, TypeError): | 405 | except (AttributeError, TypeError): |
| 397 | # Expected behavior when None is passed | 406 | # Expected behavior when None is passed |
| 398 | pass | 407 | pass |
| 399 | - | 408 | + |
| 400 | # Test with empty/invalid TLSConfig | 409 | # Test with empty/invalid TLSConfig |
| 401 | - empty_tls_config = TLSConfig( | 410 | + empty_tls_config = TLSConfig(enable_tls=True, ca_file="", cert_file="", key_file="") |
| 402 | - enable_tls=True, | ||
| 403 | - ca_file="", | ||
| 404 | - cert_file="", | ||
| 405 | - key_file="" | ||
| 406 | - ) | ||
| 407 | ssl_context = CertUtil.create_ssl_context(tls_config=empty_tls_config) | 411 | ssl_context = CertUtil.create_ssl_context(tls_config=empty_tls_config) |
| 408 | assert ssl_context is None, "Empty certificate files should return None" | 412 | assert ssl_context is None, "Empty certificate files should return None" |
| 409 | - | 413 | + |
| 410 | # Test non-existent certificate files | 414 | # Test non-existent certificate files |
| 411 | invalid_tls_config = TLSConfig( | 415 | invalid_tls_config = TLSConfig( |
| 412 | enable_tls=True, | 416 | enable_tls=True, |
| 413 | ca_file="/nonexistent/ca.pem", | 417 | ca_file="/nonexistent/ca.pem", |
| 414 | cert_file="/nonexistent/cert.pem", | 418 | cert_file="/nonexistent/cert.pem", |
| 415 | - key_file="/nonexistent/key.pem" | 419 | + key_file="/nonexistent/key.pem", |
| 416 | ) | 420 | ) |
| 417 | ssl_context = CertUtil.create_ssl_context(tls_config=invalid_tls_config) | 421 | ssl_context = CertUtil.create_ssl_context(tls_config=invalid_tls_config) |
| 418 | assert ssl_context is None, "Non-existent certificate files should return None" | 422 | assert ssl_context is None, "Non-existent certificate files should return None" |
| 419 | - | 423 | + |
| 420 | # Test create_ssl_context_no_client_cert with None values | 424 | # Test create_ssl_context_no_client_cert with None values |
| 421 | - ssl_context = CertUtil.create_ssl_context_no_client_cert( | 425 | + ssl_context = CertUtil.create_ssl_context_no_client_cert(cert_file=None, key_file=None, ca_file=None) |
| 422 | - cert_file=None, | ||
| 423 | - key_file=None, | ||
| 424 | - ca_file=None | ||
| 425 | - ) | ||
| 426 | assert ssl_context is None, "None values should return None" | 426 | assert ssl_context is None, "None values should return None" |
| 427 | - | 427 | + |
| 428 | # Test with empty key_file | 428 | # Test with empty key_file |
| 429 | - ssl_context = CertUtil.create_ssl_context_no_client_cert( | 429 | + ssl_context = CertUtil.create_ssl_context_no_client_cert(cert_file="/nonexistent/cert.pem", key_file="", ca_file="") |
| 430 | - cert_file="/nonexistent/cert.pem", | ||
| 431 | - key_file="", | ||
| 432 | - ca_file="" | ||
| 433 | - ) | ||
| 434 | assert ssl_context is None, "Empty key_file should return None" | 430 | assert ssl_context is None, "Empty key_file should return None" |
| 435 | - | 431 | + |
| 436 | # Test with non-existent certificate files | 432 | # Test with non-existent certificate files |
| 437 | ssl_context = CertUtil.create_ssl_context_no_client_cert( | 433 | ssl_context = CertUtil.create_ssl_context_no_client_cert( |
| 438 | - cert_file="/nonexistent/cert.pem", | 434 | + cert_file="/nonexistent/cert.pem", key_file="/nonexistent/key.pem", ca_file="" |
| 439 | - key_file="/nonexistent/key.pem", | ||
| 440 | - ca_file="" | ||
| 441 | ) | 435 | ) |
| 442 | assert ssl_context is None, "Non-existent certificate files should return None" | 436 | assert ssl_context is None, "Non-existent certificate files should return None" |
| 443 | - | 437 | + |
| 444 | logger.info("SSL context creation error handling works correctly") | 438 | logger.info("SSL context creation error handling works correctly") |
| 445 | 439 | ||
| 446 | 440 | ||
| @@ -448,33 +442,34 @@ def test_create_ssl_context_error_handling(): | |||
| 448 | # Certificate info query tests | 442 | # Certificate info query tests |
| 449 | # ============================================================================ | 443 | # ============================================================================ |
| 450 | 444 | ||
| 445 | + | ||
| 451 | def test_cert_info_query(test_certificates): | 446 | def test_cert_info_query(test_certificates): |
| 452 | """Test certificate and CRL info query functionality""" | 447 | """Test certificate and CRL info query functionality""" |
| 453 | logger.info("=== Testing certificate info query ===") | 448 | logger.info("=== Testing certificate info query ===") |
| 454 | - | 449 | + |
| 455 | test_certs = test_certificates | 450 | test_certs = test_certificates |
| 456 | - | 451 | + |
| 457 | # Test certificate information query | 452 | # Test certificate information query |
| 458 | cert_info = CertUtil.query_certificate_info(test_certs["server_cert"]) | 453 | cert_info = CertUtil.query_certificate_info(test_certs["server_cert"]) |
| 459 | logger.info(f"Certificate information query succeeded: {cert_info}") | 454 | logger.info(f"Certificate information query succeeded: {cert_info}") |
| 460 | assert cert_info is not None, "Certificate information query should succeed" | 455 | assert cert_info is not None, "Certificate information query should succeed" |
| 461 | - | 456 | + |
| 462 | # Test with non-existent certificate file | 457 | # Test with non-existent certificate file |
| 463 | cert_info = CertUtil.query_certificate_info("/nonexistent/cert.pem") | 458 | cert_info = CertUtil.query_certificate_info("/nonexistent/cert.pem") |
| 464 | assert cert_info == {}, "Certificate info query should return empty dict for non-existent file" | 459 | assert cert_info == {}, "Certificate info query should return empty dict for non-existent file" |
| 465 | - | 460 | + |
| 466 | # Test with invalid certificate file | 461 | # Test with invalid certificate file |
| 467 | temp_dir = tempfile.mkdtemp() | 462 | temp_dir = tempfile.mkdtemp() |
| 468 | try: | 463 | try: |
| 469 | invalid_cert_path = os.path.join(temp_dir, "invalid_cert.pem") | 464 | invalid_cert_path = os.path.join(temp_dir, "invalid_cert.pem") |
| 470 | - with open(invalid_cert_path, "w") as f: | 465 | + with open(invalid_cert_path, "w", encoding="utf-8") as f: |
| 471 | f.write("invalid certificate content") | 466 | f.write("invalid certificate content") |
| 472 | - | 467 | + |
| 473 | cert_info = CertUtil.query_certificate_info(invalid_cert_path) | 468 | cert_info = CertUtil.query_certificate_info(invalid_cert_path) |
| 474 | assert cert_info == {}, "Certificate info query should return empty dict for invalid file" | 469 | assert cert_info == {}, "Certificate info query should return empty dict for invalid file" |
| 475 | finally: | 470 | finally: |
| 476 | shutil.rmtree(temp_dir) | 471 | shutil.rmtree(temp_dir) |
| 477 | - | 472 | + |
| 478 | # Test CRL info query with non-existent file | 473 | # Test CRL info query with non-existent file |
| 479 | crl_info = CertUtil.query_crl_info("/nonexistent/crl.pem") | 474 | crl_info = CertUtil.query_crl_info("/nonexistent/crl.pem") |
| 480 | assert crl_info == [], "CRL info query should return empty list for non-existent file" | 475 | assert crl_info == [], "CRL info query should return empty list for non-existent file" |
| @@ -485,52 +480,45 @@ def test_cert_info_query(test_certificates): | |||
| 485 | # Certificate chain validation tests | 480 | # Certificate chain validation tests |
| 486 | # ============================================================================ | 481 | # ============================================================================ |
| 487 | 482 | ||
| 483 | + | ||
| 488 | def test_validate_certificate_chain(test_certificates): | 484 | def test_validate_certificate_chain(test_certificates): |
| 489 | """Test certificate chain validation""" | 485 | """Test certificate chain validation""" |
| 490 | logger.info("=== Testing certificate chain validation ===") | 486 | logger.info("=== Testing certificate chain validation ===") |
| 491 | - | 487 | + |
| 492 | test_certs = test_certificates | 488 | test_certs = test_certificates |
| 493 | - | 489 | + |
| 494 | # Test basic certificate chain validation | 490 | # Test basic certificate chain validation |
| 495 | validation_result = CertUtil.validate_certificate_chain( | 491 | validation_result = CertUtil.validate_certificate_chain( |
| 496 | - ca_file=test_certs["ca_cert"], | 492 | + ca_file=test_certs["ca_cert"], cert_file=test_certs["server_cert"], key_file=test_certs["server_key"] |
| 497 | - cert_file=test_certs["server_cert"], | ||
| 498 | - key_file=test_certs["server_key"] | ||
| 499 | ) | 493 | ) |
| 500 | assert validation_result is True, "Certificate chain validation should succeed" | 494 | assert validation_result is True, "Certificate chain validation should succeed" |
| 501 | - | 495 | + |
| 502 | # Test without CRL (should work) | 496 | # Test without CRL (should work) |
| 503 | validation_result = CertUtil.validate_certificate_chain( | 497 | validation_result = CertUtil.validate_certificate_chain( |
| 504 | ca_file=test_certs["ca_cert"], | 498 | ca_file=test_certs["ca_cert"], |
| 505 | cert_file=test_certs["server_cert"], | 499 | cert_file=test_certs["server_cert"], |
| 506 | key_file=test_certs["server_key"], | 500 | key_file=test_certs["server_key"], |
| 507 | - crl_file=None | 501 | + crl_file=None, |
| 508 | ) | 502 | ) |
| 509 | assert validation_result is True, "Certificate chain validation should succeed without CRL" | 503 | assert validation_result is True, "Certificate chain validation should succeed without CRL" |
| 510 | - | 504 | + |
| 511 | # Test with non-existent CRL file | 505 | # Test with non-existent CRL file |
| 512 | validation_result = CertUtil.validate_certificate_chain( | 506 | validation_result = CertUtil.validate_certificate_chain( |
| 513 | ca_file=test_certs["ca_cert"], | 507 | ca_file=test_certs["ca_cert"], |
| 514 | cert_file=test_certs["server_cert"], | 508 | cert_file=test_certs["server_cert"], |
| 515 | key_file=test_certs["server_key"], | 509 | key_file=test_certs["server_key"], |
| 516 | - crl_file="/nonexistent/crl.pem" | 510 | + crl_file="/nonexistent/crl.pem", |
| 517 | ) | 511 | ) |
| 518 | # Should succeed if CRL file doesn't exist (optional) | 512 | # Should succeed if CRL file doesn't exist (optional) |
| 519 | logger.info("Certificate chain validation handled non-existent CRL file") | 513 | logger.info("Certificate chain validation handled non-existent CRL file") |
| 520 | - | 514 | + |
| 521 | # Test error handling | 515 | # Test error handling |
| 522 | result = CertUtil.validate_certificate_chain( | 516 | result = CertUtil.validate_certificate_chain( |
| 523 | - ca_file="/nonexistent/ca.pem", | 517 | + ca_file="/nonexistent/ca.pem", cert_file="/nonexistent/cert.pem", key_file="/nonexistent/key.pem" |
| 524 | - cert_file="/nonexistent/cert.pem", | ||
| 525 | - key_file="/nonexistent/key.pem" | ||
| 526 | ) | 518 | ) |
| 527 | assert result is False, "Non-existent files should return False" | 519 | assert result is False, "Non-existent files should return False" |
| 528 | - | 520 | + |
| 529 | - result = CertUtil.validate_certificate_chain( | 521 | + result = CertUtil.validate_certificate_chain(ca_file="", cert_file="", key_file="") |
| 530 | - ca_file="", | ||
| 531 | - cert_file="", | ||
| 532 | - key_file="" | ||
| 533 | - ) | ||
| 534 | assert result is False, "Empty strings should return False" | 522 | assert result is False, "Empty strings should return False" |
| 535 | logger.info("Certificate chain validation works correctly") | 523 | logger.info("Certificate chain validation works correctly") |
| 536 | 524 | ||
| @@ -539,12 +527,13 @@ def test_validate_certificate_chain(test_certificates): | |||
| 539 | # construct_cert_context tests | 527 | # construct_cert_context tests |
| 540 | # ============================================================================ | 528 | # ============================================================================ |
| 541 | 529 | ||
| 530 | + | ||
| 542 | def test_construct_cert_context(test_certificates): | 531 | def test_construct_cert_context(test_certificates): |
| 543 | """Test construct_cert_context method with strict validation""" | 532 | """Test construct_cert_context method with strict validation""" |
| 544 | logger.info("=== Testing construct_cert_context method ===") | 533 | logger.info("=== Testing construct_cert_context method ===") |
| 545 | - | 534 | + |
| 546 | test_certs = test_certificates | 535 | test_certs = test_certificates |
| 547 | - | 536 | + |
| 548 | # Note: This test may fail if directory permissions are not 700 | 537 | # Note: This test may fail if directory permissions are not 700 |
| 549 | # We'll skip it if it fails due to permission issues | 538 | # We'll skip it if it fails due to permission issues |
| 550 | try: | 539 | try: |
| @@ -553,9 +542,9 @@ def test_construct_cert_context(test_certificates): | |||
| 553 | "ca_cert": test_certs["ca_cert"], | 542 | "ca_cert": test_certs["ca_cert"], |
| 554 | "tls_cert": test_certs["server_cert"], | 543 | "tls_cert": test_certs["server_cert"], |
| 555 | "tls_key": test_certs["server_key"], | 544 | "tls_key": test_certs["server_key"], |
| 556 | - "tls_passwd": "" | 545 | + "tls_passwd": "", |
| 557 | } | 546 | } |
| 558 | - | 547 | + |
| 559 | ssl_context = CertUtil.construct_cert_context(config) | 548 | ssl_context = CertUtil.construct_cert_context(config) |
| 560 | # May succeed or fail depending on certificate validation | 549 | # May succeed or fail depending on certificate validation |
| 561 | logger.info(f"construct_cert_context result: {ssl_context is not None}") | 550 | logger.info(f"construct_cert_context result: {ssl_context is not None}") |
| @@ -563,29 +552,28 @@ def test_construct_cert_context(test_certificates): | |||
| 563 | logger.info(f"construct_cert_context failed (expected for some cases): {e}") | 552 | logger.info(f"construct_cert_context failed (expected for some cases): {e}") |
| 564 | 553 | ||
| 565 | 554 | ||
| 566 | - | ||
| 567 | - | ||
| 568 | # ============================================================================ | 555 | # ============================================================================ |
| 569 | # CRL validation tests | 556 | # CRL validation tests |
| 570 | # ============================================================================ | 557 | # ============================================================================ |
| 571 | 558 | ||
| 559 | + | ||
| 572 | def test_validate_revoke_list(test_certificates): | 560 | def test_validate_revoke_list(test_certificates): |
| 573 | """Test validate_revoke_list with various CRL scenarios""" | 561 | """Test validate_revoke_list with various CRL scenarios""" |
| 574 | logger.info("=== Testing validate_revoke_list ===") | 562 | logger.info("=== Testing validate_revoke_list ===") |
| 575 | - | 563 | + |
| 576 | test_certs = test_certificates | 564 | test_certs = test_certificates |
| 577 | - | 565 | + |
| 578 | # Test with valid CRL (empty list, valid next_update) | 566 | # Test with valid CRL (empty list, valid next_update) |
| 579 | crl_info = create_test_crl( | 567 | crl_info = create_test_crl( |
| 580 | ca_key=test_certs["ca_key_obj"], | 568 | ca_key=test_certs["ca_key_obj"], |
| 581 | ca_cert=test_certs["ca_cert_obj"], | 569 | ca_cert=test_certs["ca_cert_obj"], |
| 582 | revoked_serial_numbers=None, | 570 | revoked_serial_numbers=None, |
| 583 | next_update_days=30, | 571 | next_update_days=30, |
| 584 | - temp_dir=test_certs["temp_dir"] | 572 | + temp_dir=test_certs["temp_dir"], |
| 585 | ) | 573 | ) |
| 586 | result = CertValidationUtil.validate_revoke_list(crl_info["crl_path"]) | 574 | result = CertValidationUtil.validate_revoke_list(crl_info["crl_path"]) |
| 587 | assert result is True, "Valid CRL should return True" | 575 | assert result is True, "Valid CRL should return True" |
| 588 | - | 576 | + |
| 589 | # Test with CRL containing revoked certificates | 577 | # Test with CRL containing revoked certificates |
| 590 | revoked_serials = [12345, 67890] | 578 | revoked_serials = [12345, 67890] |
| 591 | crl_info = create_test_crl( | 579 | crl_info = create_test_crl( |
| @@ -593,58 +581,58 @@ def test_validate_revoke_list(test_certificates): | |||
| 593 | ca_cert=test_certs["ca_cert_obj"], | 581 | ca_cert=test_certs["ca_cert_obj"], |
| 594 | revoked_serial_numbers=revoked_serials, | 582 | revoked_serial_numbers=revoked_serials, |
| 595 | next_update_days=30, | 583 | next_update_days=30, |
| 596 | - temp_dir=test_certs["temp_dir"] | 584 | + temp_dir=test_certs["temp_dir"], |
| 597 | ) | 585 | ) |
| 598 | result = CertValidationUtil.validate_revoke_list(crl_info["crl_path"]) | 586 | result = CertValidationUtil.validate_revoke_list(crl_info["crl_path"]) |
| 599 | assert result is True, "CRL with revoked certificates should return True" | 587 | assert result is True, "CRL with revoked certificates should return True" |
| 600 | - | 588 | + |
| 601 | # Test with expired CRL | 589 | # Test with expired CRL |
| 602 | crl_info = create_test_crl( | 590 | crl_info = create_test_crl( |
| 603 | ca_key=test_certs["ca_key_obj"], | 591 | ca_key=test_certs["ca_key_obj"], |
| 604 | ca_cert=test_certs["ca_cert_obj"], | 592 | ca_cert=test_certs["ca_cert_obj"], |
| 605 | revoked_serial_numbers=None, | 593 | revoked_serial_numbers=None, |
| 606 | next_update_days=-1, # Expired | 594 | next_update_days=-1, # Expired |
| 607 | - temp_dir=test_certs["temp_dir"] | 595 | + temp_dir=test_certs["temp_dir"], |
| 608 | ) | 596 | ) |
| 609 | result = CertValidationUtil.validate_revoke_list(crl_info["crl_path"]) | 597 | result = CertValidationUtil.validate_revoke_list(crl_info["crl_path"]) |
| 610 | assert result is False, "Expired CRL should return False" | 598 | assert result is False, "Expired CRL should return False" |
| 611 | - | 599 | + |
| 612 | # Test with non-existent file | 600 | # Test with non-existent file |
| 613 | result = CertValidationUtil.validate_revoke_list("/nonexistent/crl.pem") | 601 | result = CertValidationUtil.validate_revoke_list("/nonexistent/crl.pem") |
| 614 | assert result is False, "Non-existent file should return False" | 602 | assert result is False, "Non-existent file should return False" |
| 615 | - | 603 | + |
| 616 | # Test with invalid CRL file | 604 | # Test with invalid CRL file |
| 617 | temp_dir = tempfile.mkdtemp() | 605 | temp_dir = tempfile.mkdtemp() |
| 618 | try: | 606 | try: |
| 619 | invalid_crl_path = os.path.join(temp_dir, "invalid_crl.pem") | 607 | invalid_crl_path = os.path.join(temp_dir, "invalid_crl.pem") |
| 620 | - with open(invalid_crl_path, "w") as f: | 608 | + with open(invalid_crl_path, "w", encoding="utf-8") as f: |
| 621 | f.write("invalid CRL content") | 609 | f.write("invalid CRL content") |
| 622 | - | 610 | + |
| 623 | result = CertValidationUtil.validate_revoke_list(invalid_crl_path) | 611 | result = CertValidationUtil.validate_revoke_list(invalid_crl_path) |
| 624 | assert result is False, "Invalid CRL file should return False" | 612 | assert result is False, "Invalid CRL file should return False" |
| 625 | finally: | 613 | finally: |
| 626 | shutil.rmtree(temp_dir) | 614 | shutil.rmtree(temp_dir) |
| 627 | - | 615 | + |
| 628 | logger.info("validate_revoke_list works correctly") | 616 | logger.info("validate_revoke_list works correctly") |
| 629 | 617 | ||
| 630 | 618 | ||
| 631 | def test_validate_ca_crl(test_certificates): | 619 | def test_validate_ca_crl(test_certificates): |
| 632 | """Test validate_ca_crl with various scenarios""" | 620 | """Test validate_ca_crl with various scenarios""" |
| 633 | logger.info("=== Testing validate_ca_crl ===") | 621 | logger.info("=== Testing validate_ca_crl ===") |
| 634 | - | 622 | + |
| 635 | test_certs = test_certificates | 623 | test_certs = test_certificates |
| 636 | - | 624 | + |
| 637 | # Test with valid CRL signed by matching CA | 625 | # Test with valid CRL signed by matching CA |
| 638 | crl_info = create_test_crl( | 626 | crl_info = create_test_crl( |
| 639 | ca_key=test_certs["ca_key_obj"], | 627 | ca_key=test_certs["ca_key_obj"], |
| 640 | ca_cert=test_certs["ca_cert_obj"], | 628 | ca_cert=test_certs["ca_cert_obj"], |
| 641 | revoked_serial_numbers=None, | 629 | revoked_serial_numbers=None, |
| 642 | next_update_days=30, | 630 | next_update_days=30, |
| 643 | - temp_dir=test_certs["temp_dir"] | 631 | + temp_dir=test_certs["temp_dir"], |
| 644 | ) | 632 | ) |
| 645 | result = CertValidationUtil.validate_ca_crl(test_certs["ca_cert"], crl_info["crl_path"]) | 633 | result = CertValidationUtil.validate_ca_crl(test_certs["ca_cert"], crl_info["crl_path"]) |
| 646 | assert result is True, "Valid CRL signed by matching CA should return True" | 634 | assert result is True, "Valid CRL signed by matching CA should return True" |
| 647 | - | 635 | + |
| 648 | # Test with CRL signed by different CA | 636 | # Test with CRL signed by different CA |
| 649 | other_ca_key, other_ca_cert = create_other_ca() | 637 | other_ca_key, other_ca_cert = create_other_ca() |
| 650 | crl_info = create_test_crl( | 638 | crl_info = create_test_crl( |
| @@ -652,27 +640,27 @@ def test_validate_ca_crl(test_certificates): | |||
| 652 | ca_cert=other_ca_cert, | 640 | ca_cert=other_ca_cert, |
| 653 | revoked_serial_numbers=None, | 641 | revoked_serial_numbers=None, |
| 654 | next_update_days=30, | 642 | next_update_days=30, |
| 655 | - temp_dir=test_certs["temp_dir"] | 643 | + temp_dir=test_certs["temp_dir"], |
| 656 | ) | 644 | ) |
| 657 | result = CertValidationUtil.validate_ca_crl(test_certs["ca_cert"], crl_info["crl_path"]) | 645 | result = CertValidationUtil.validate_ca_crl(test_certs["ca_cert"], crl_info["crl_path"]) |
| 658 | assert result is False, "CRL signed by different CA should return False" | 646 | assert result is False, "CRL signed by different CA should return False" |
| 659 | - | 647 | + |
| 660 | # Test with non-existent files | 648 | # Test with non-existent files |
| 661 | result = CertValidationUtil.validate_ca_crl("/nonexistent/ca.pem", "/nonexistent/crl.pem") | 649 | result = CertValidationUtil.validate_ca_crl("/nonexistent/ca.pem", "/nonexistent/crl.pem") |
| 662 | assert result is False, "Non-existent files should return False" | 650 | assert result is False, "Non-existent files should return False" |
| 663 | - | 651 | + |
| 664 | # Test with invalid CRL file | 652 | # Test with invalid CRL file |
| 665 | temp_dir = tempfile.mkdtemp() | 653 | temp_dir = tempfile.mkdtemp() |
| 666 | try: | 654 | try: |
| 667 | invalid_crl_path = os.path.join(temp_dir, "invalid_crl.pem") | 655 | invalid_crl_path = os.path.join(temp_dir, "invalid_crl.pem") |
| 668 | - with open(invalid_crl_path, "w") as f: | 656 | + with open(invalid_crl_path, "w", encoding="utf-8") as f: |
| 669 | f.write("invalid CRL content") | 657 | f.write("invalid CRL content") |
| 670 | - | 658 | + |
| 671 | result = CertValidationUtil.validate_ca_crl("/nonexistent/ca.pem", invalid_crl_path) | 659 | result = CertValidationUtil.validate_ca_crl("/nonexistent/ca.pem", invalid_crl_path) |
| 672 | assert result is False, "Invalid CRL file should return False" | 660 | assert result is False, "Invalid CRL file should return False" |
| 673 | finally: | 661 | finally: |
| 674 | shutil.rmtree(temp_dir) | 662 | shutil.rmtree(temp_dir) |
| 675 | - | 663 | + |
| 676 | logger.info("validate_ca_crl works correctly") | 664 | logger.info("validate_ca_crl works correctly") |
| 677 | 665 | ||
| 678 | 666 | ||
| @@ -680,52 +668,53 @@ def test_validate_ca_crl(test_certificates): | |||
| 680 | # construct_cert_context tests | 668 | # construct_cert_context tests |
| 681 | # ============================================================================ | 669 | # ============================================================================ |
| 682 | 670 | ||
| 671 | + | ||
| 683 | def test_construct_cert_context_comprehensive(test_certificates): | 672 | def test_construct_cert_context_comprehensive(test_certificates): |
| 684 | """Test construct_cert_context with various scenarios""" | 673 | """Test construct_cert_context with various scenarios""" |
| 685 | logger.info("=== Testing construct_cert_context comprehensive scenarios ===") | 674 | logger.info("=== Testing construct_cert_context comprehensive scenarios ===") |
| 686 | - | 675 | + |
| 687 | test_certs = test_certificates | 676 | test_certs = test_certificates |
| 688 | - | 677 | + |
| 689 | # Test with valid certificates (no CRL) | 678 | # Test with valid certificates (no CRL) |
| 690 | try: | 679 | try: |
| 691 | config = { | 680 | config = { |
| 692 | "ca_cert": test_certs["ca_cert"], | 681 | "ca_cert": test_certs["ca_cert"], |
| 693 | "tls_cert": test_certs["server_cert"], | 682 | "tls_cert": test_certs["server_cert"], |
| 694 | "tls_key": test_certs["server_key"], | 683 | "tls_key": test_certs["server_key"], |
| 695 | - "tls_passwd": "" | 684 | + "tls_passwd": "", |
| 696 | } | 685 | } |
| 697 | ssl_context = CertUtil.construct_cert_context(config) | 686 | ssl_context = CertUtil.construct_cert_context(config) |
| 698 | logger.info(f"construct_cert_context without CRL result: {ssl_context is not None}") | 687 | logger.info(f"construct_cert_context without CRL result: {ssl_context is not None}") |
| 699 | except Exception as e: | 688 | except Exception as e: |
| 700 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") | 689 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") |
| 701 | - | 690 | + |
| 702 | # Test with valid CRL | 691 | # Test with valid CRL |
| 703 | crl_info = create_test_crl( | 692 | crl_info = create_test_crl( |
| 704 | ca_key=test_certs["ca_key_obj"], | 693 | ca_key=test_certs["ca_key_obj"], |
| 705 | ca_cert=test_certs["ca_cert_obj"], | 694 | ca_cert=test_certs["ca_cert_obj"], |
| 706 | revoked_serial_numbers=None, | 695 | revoked_serial_numbers=None, |
| 707 | next_update_days=30, | 696 | next_update_days=30, |
| 708 | - temp_dir=test_certs["temp_dir"] | 697 | + temp_dir=test_certs["temp_dir"], |
| 709 | ) | 698 | ) |
| 710 | - | 699 | + |
| 711 | try: | 700 | try: |
| 712 | config = { | 701 | config = { |
| 713 | "ca_cert": test_certs["ca_cert"], | 702 | "ca_cert": test_certs["ca_cert"], |
| 714 | "tls_cert": test_certs["server_cert"], | 703 | "tls_cert": test_certs["server_cert"], |
| 715 | "tls_key": test_certs["server_key"], | 704 | "tls_key": test_certs["server_key"], |
| 716 | "tls_crl": crl_info["crl_path"], | 705 | "tls_crl": crl_info["crl_path"], |
| 717 | - "tls_passwd": "" | 706 | + "tls_passwd": "", |
| 718 | } | 707 | } |
| 719 | ssl_context = CertUtil.construct_cert_context(config) | 708 | ssl_context = CertUtil.construct_cert_context(config) |
| 720 | assert ssl_context is not None, "construct_cert_context should succeed with valid CRL" | 709 | assert ssl_context is not None, "construct_cert_context should succeed with valid CRL" |
| 721 | - | 710 | + |
| 722 | # Verify context attributes | 711 | # Verify context attributes |
| 723 | assert hasattr(ssl_context, 'cert_file'), "SSL context should have cert_file attribute" | 712 | assert hasattr(ssl_context, 'cert_file'), "SSL context should have cert_file attribute" |
| 724 | assert hasattr(ssl_context, 'key_file'), "SSL context should have key_file attribute" | 713 | assert hasattr(ssl_context, 'key_file'), "SSL context should have key_file attribute" |
| 725 | assert hasattr(ssl_context, 'ca_file'), "SSL context should have ca_file attribute" | 714 | assert hasattr(ssl_context, 'ca_file'), "SSL context should have ca_file attribute" |
| 726 | except Exception as e: | 715 | except Exception as e: |
| 727 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") | 716 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") |
| 728 | - | 717 | + |
| 729 | # Test with valid CRL containing revoked certificates | 718 | # Test with valid CRL containing revoked certificates |
| 730 | revoked_serials = [12345, 67890] | 719 | revoked_serials = [12345, 67890] |
| 731 | crl_info = create_test_crl( | 720 | crl_info = create_test_crl( |
| @@ -733,35 +722,35 @@ def test_construct_cert_context_comprehensive(test_certificates): | |||
| 733 | ca_cert=test_certs["ca_cert_obj"], | 722 | ca_cert=test_certs["ca_cert_obj"], |
| 734 | revoked_serial_numbers=revoked_serials, | 723 | revoked_serial_numbers=revoked_serials, |
| 735 | next_update_days=30, | 724 | next_update_days=30, |
| 736 | - temp_dir=test_certs["temp_dir"] | 725 | + temp_dir=test_certs["temp_dir"], |
| 737 | ) | 726 | ) |
| 738 | - | 727 | + |
| 739 | try: | 728 | try: |
| 740 | config = { | 729 | config = { |
| 741 | "ca_cert": test_certs["ca_cert"], | 730 | "ca_cert": test_certs["ca_cert"], |
| 742 | "tls_cert": test_certs["server_cert"], | 731 | "tls_cert": test_certs["server_cert"], |
| 743 | "tls_key": test_certs["server_key"], | 732 | "tls_key": test_certs["server_key"], |
| 744 | "tls_crl": crl_info["crl_path"], | 733 | "tls_crl": crl_info["crl_path"], |
| 745 | - "tls_passwd": "" | 734 | + "tls_passwd": "", |
| 746 | } | 735 | } |
| 747 | ssl_context = CertUtil.construct_cert_context(config) | 736 | ssl_context = CertUtil.construct_cert_context(config) |
| 748 | assert ssl_context is not None, "construct_cert_context should succeed with CRL containing revoked certs" | 737 | assert ssl_context is not None, "construct_cert_context should succeed with CRL containing revoked certs" |
| 749 | except Exception as e: | 738 | except Exception as e: |
| 750 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") | 739 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") |
| 751 | - | 740 | + |
| 752 | # Test with invalid CRL file | 741 | # Test with invalid CRL file |
| 753 | temp_dir = tempfile.mkdtemp() | 742 | temp_dir = tempfile.mkdtemp() |
| 754 | invalid_crl_path = os.path.join(temp_dir, "invalid_crl.pem") | 743 | invalid_crl_path = os.path.join(temp_dir, "invalid_crl.pem") |
| 755 | - with open(invalid_crl_path, "w") as f: | 744 | + with open(invalid_crl_path, "w", encoding="utf-8") as f: |
| 756 | f.write("invalid CRL content") | 745 | f.write("invalid CRL content") |
| 757 | - | 746 | + |
| 758 | try: | 747 | try: |
| 759 | config = { | 748 | config = { |
| 760 | "ca_cert": test_certs["ca_cert"], | 749 | "ca_cert": test_certs["ca_cert"], |
| 761 | "tls_cert": test_certs["server_cert"], | 750 | "tls_cert": test_certs["server_cert"], |
| 762 | "tls_key": test_certs["server_key"], | 751 | "tls_key": test_certs["server_key"], |
| 763 | "tls_crl": invalid_crl_path, | 752 | "tls_crl": invalid_crl_path, |
| 764 | - "tls_passwd": "" | 753 | + "tls_passwd": "", |
| 765 | } | 754 | } |
| 766 | ssl_context = CertUtil.construct_cert_context(config) | 755 | ssl_context = CertUtil.construct_cert_context(config) |
| 767 | assert ssl_context is None, "construct_cert_context should fail with invalid CRL file" | 756 | assert ssl_context is None, "construct_cert_context should fail with invalid CRL file" |
| @@ -769,7 +758,7 @@ def test_construct_cert_context_comprehensive(test_certificates): | |||
| 769 | logger.info(f"construct_cert_context failed as expected: {e}") | 758 | logger.info(f"construct_cert_context failed as expected: {e}") |
| 770 | finally: | 759 | finally: |
| 771 | shutil.rmtree(temp_dir) | 760 | shutil.rmtree(temp_dir) |
| 772 | - | 761 | + |
| 773 | # Test with mismatched CRL (signed by different CA) | 762 | # Test with mismatched CRL (signed by different CA) |
| 774 | other_ca_key, other_ca_cert = create_other_ca() | 763 | other_ca_key, other_ca_cert = create_other_ca() |
| 775 | crl_info = create_test_crl( | 764 | crl_info = create_test_crl( |
| @@ -777,42 +766,42 @@ def test_construct_cert_context_comprehensive(test_certificates): | |||
| 777 | ca_cert=other_ca_cert, | 766 | ca_cert=other_ca_cert, |
| 778 | revoked_serial_numbers=None, | 767 | revoked_serial_numbers=None, |
| 779 | next_update_days=30, | 768 | next_update_days=30, |
| 780 | - temp_dir=test_certs["temp_dir"] | 769 | + temp_dir=test_certs["temp_dir"], |
| 781 | ) | 770 | ) |
| 782 | - | 771 | + |
| 783 | try: | 772 | try: |
| 784 | config = { | 773 | config = { |
| 785 | "ca_cert": test_certs["ca_cert"], | 774 | "ca_cert": test_certs["ca_cert"], |
| 786 | "tls_cert": test_certs["server_cert"], | 775 | "tls_cert": test_certs["server_cert"], |
| 787 | "tls_key": test_certs["server_key"], | 776 | "tls_key": test_certs["server_key"], |
| 788 | "tls_crl": crl_info["crl_path"], | 777 | "tls_crl": crl_info["crl_path"], |
| 789 | - "tls_passwd": "" | 778 | + "tls_passwd": "", |
| 790 | } | 779 | } |
| 791 | ssl_context = CertUtil.construct_cert_context(config) | 780 | ssl_context = CertUtil.construct_cert_context(config) |
| 792 | assert ssl_context is None, "construct_cert_context should fail with mismatched CRL" | 781 | assert ssl_context is None, "construct_cert_context should fail with mismatched CRL" |
| 793 | except Exception as e: | 782 | except Exception as e: |
| 794 | logger.info(f"construct_cert_context failed as expected: {e}") | 783 | logger.info(f"construct_cert_context failed as expected: {e}") |
| 795 | - | 784 | + |
| 796 | # Test with password | 785 | # Test with password |
| 797 | try: | 786 | try: |
| 798 | config = { | 787 | config = { |
| 799 | "ca_cert": test_certs["ca_cert"], | 788 | "ca_cert": test_certs["ca_cert"], |
| 800 | "tls_cert": test_certs["server_cert"], | 789 | "tls_cert": test_certs["server_cert"], |
| 801 | "tls_key": test_certs["server_key"], | 790 | "tls_key": test_certs["server_key"], |
| 802 | - "tls_passwd": "test_password" | 791 | + "tls_passwd": "test_password", |
| 803 | } | 792 | } |
| 804 | ssl_context = CertUtil.construct_cert_context(config) | 793 | ssl_context = CertUtil.construct_cert_context(config) |
| 805 | assert ssl_context is not None, "construct_cert_context should handle password parameter" | 794 | assert ssl_context is not None, "construct_cert_context should handle password parameter" |
| 806 | except Exception as e: | 795 | except Exception as e: |
| 807 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") | 796 | logger.info(f"construct_cert_context failed (may be due to directory permissions): {e}") |
| 808 | - | 797 | + |
| 809 | # Test error handling | 798 | # Test error handling |
| 810 | ssl_context = CertUtil.construct_cert_context({}) | 799 | ssl_context = CertUtil.construct_cert_context({}) |
| 811 | assert ssl_context is None, "Empty config should return None" | 800 | assert ssl_context is None, "Empty config should return None" |
| 812 | - | 801 | + |
| 813 | invalid_config = { | 802 | invalid_config = { |
| 814 | "ca_cert": "/nonexistent/ca.pem", | 803 | "ca_cert": "/nonexistent/ca.pem", |
| 815 | - "tls_cert": "/nonexistent/cert.pem" | 804 | + "tls_cert": "/nonexistent/cert.pem", |
| 816 | # Missing tls_key | 805 | # Missing tls_key |
| 817 | } | 806 | } |
| 818 | ssl_context = CertUtil.construct_cert_context(invalid_config) | 807 | ssl_context = CertUtil.construct_cert_context(invalid_config) |
| @@ -824,212 +813,174 @@ def test_construct_cert_context_comprehensive(test_certificates): | |||
| 824 | # validate_cert_and_key tests | 813 | # validate_cert_and_key tests |
| 825 | # ============================================================================ | 814 | # ============================================================================ |
| 826 | 815 | ||
| 827 | -def test_validate_cert_and_key_comprehensive(test_certificates): | 816 | + |
| 828 | - """Test validate_cert_and_key with comprehensive scenarios""" | 817 | +@pytest.mark.parametrize( |
| 829 | - logger.info("=== Testing validate_cert_and_key comprehensive scenarios ===") | 818 | + ("ca_crt_path", "plain_text"), |
| 830 | - | 819 | + [ |
| 820 | + (None, None), | ||
| 821 | + ("", None), | ||
| 822 | + ("ca", None), | ||
| 823 | + (None, b""), | ||
| 824 | + (None, b"test_password"), | ||
| 825 | + ], | ||
| 826 | + ids=["no-ca", "empty-ca", "with-ca", "empty-password", "password-ignored"], | ||
| 827 | +) | ||
| 828 | +def test_validate_cert_and_key_success_paths(test_certificates, ca_crt_path, plain_text): | ||
| 829 | + """Cover validate_cert_and_key success paths without repeating full validation.""" | ||
| 831 | test_certs = test_certificates | 830 | test_certs = test_certificates |
| 832 | - | 831 | + kwargs = { |
| 833 | - # Test with valid certificates (without CA) | 832 | + "server_crt_path": test_certs["server_cert"], |
| 834 | - result = CertValidationUtil.validate_cert_and_key( | 833 | + "server_key_path": test_certs["server_key"], |
| 835 | - server_crt_path=test_certs["server_cert"], | 834 | + } |
| 836 | - server_key_path=test_certs["server_key"] | 835 | + if ca_crt_path == "ca": |
| 836 | + kwargs["ca_crt_path"] = test_certs["ca_cert"] | ||
| 837 | + elif ca_crt_path is not None: | ||
| 838 | + kwargs["ca_crt_path"] = ca_crt_path | ||
| 839 | + if plain_text is not None: | ||
| 840 | + kwargs["plain_text"] = plain_text | ||
| 841 | + | ||
| 842 | + assert CertValidationUtil.validate_cert_and_key(**kwargs) is True | ||
| 843 | + | ||
| 844 | + | ||
| 845 | + | ||
| 846 | + ("server_crt_path", "server_key_path"), | ||
| 847 | + [ | ||
| 848 | + (None, "/nonexistent/key.pem"), | ||
| 849 | + ("/nonexistent/cert.pem", None), | ||
| 850 | + ("", "/nonexistent/key.pem"), | ||
| 851 | + ("/nonexistent/cert.pem", ""), | ||
| 852 | + ("/nonexistent/server_cert.pem", "/nonexistent/server_key.pem"), | ||
| 853 | + ], | ||
| 854 | +) | ||
| 855 | +def test_validate_cert_and_key_rejects_invalid_paths(test_certificates, server_crt_path, server_key_path): | ||
| 856 | + assert ( | ||
| 857 | + CertValidationUtil.validate_cert_and_key( | ||
| 858 | + server_crt_path=server_crt_path, | ||
| 859 | + server_key_path=server_key_path, | ||
| 860 | + ) | ||
| 861 | + is False | ||
| 837 | ) | 862 | ) |
| 838 | - assert result is True, "validate_cert_and_key should succeed with valid certificates" | 863 | + |
| 839 | - | 864 | + |
| 840 | - # Test with valid certificates (with CA) | 865 | +def test_validate_cert_and_key_rejects_empty_files(): |
| 841 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 842 | - server_crt_path=test_certs["server_cert"], | ||
| 843 | - server_key_path=test_certs["server_key"], | ||
| 844 | - ca_crt_path=test_certs["ca_cert"] | ||
| 845 | - ) | ||
| 846 | - assert result is True, "validate_cert_and_key should succeed with valid CA certificate" | ||
| 847 | - | ||
| 848 | - # Test with empty/None CA (optional) | ||
| 849 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 850 | - server_crt_path=test_certs["server_cert"], | ||
| 851 | - server_key_path=test_certs["server_key"], | ||
| 852 | - ca_crt_path="" | ||
| 853 | - ) | ||
| 854 | - assert result is True, "Empty CA certificate path should be treated as optional" | ||
| 855 | - | ||
| 856 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 857 | - server_crt_path=test_certs["server_cert"], | ||
| 858 | - server_key_path=test_certs["server_key"], | ||
| 859 | - ca_crt_path=None | ||
| 860 | - ) | ||
| 861 | - assert result is True, "None CA certificate should be treated as optional" | ||
| 862 | - | ||
| 863 | - # Test with password | ||
| 864 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 865 | - server_crt_path=test_certs["server_cert"], | ||
| 866 | - server_key_path=test_certs["server_key"], | ||
| 867 | - plain_text=b"" | ||
| 868 | - ) | ||
| 869 | - assert result is True, "validate_cert_and_key should work with empty password" | ||
| 870 | - | ||
| 871 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 872 | - server_crt_path=test_certs["server_cert"], | ||
| 873 | - server_key_path=test_certs["server_key"], | ||
| 874 | - plain_text=b"test_password" | ||
| 875 | - ) | ||
| 876 | - assert result is True, "validate_cert_and_key should handle password parameter gracefully" | ||
| 877 | - | ||
| 878 | - # Test error handling: None values | ||
| 879 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 880 | - server_crt_path=None, | ||
| 881 | - server_key_path="/nonexistent/key.pem" | ||
| 882 | - ) | ||
| 883 | - assert result is False, "None server_crt_path should return False" | ||
| 884 | - | ||
| 885 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 886 | - server_crt_path="/nonexistent/cert.pem", | ||
| 887 | - server_key_path=None | ||
| 888 | - ) | ||
| 889 | - assert result is False, "None server_key_path should return False" | ||
| 890 | - | ||
| 891 | - # Test error handling: empty strings | ||
| 892 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 893 | - server_crt_path="", | ||
| 894 | - server_key_path="/nonexistent/key.pem" | ||
| 895 | - ) | ||
| 896 | - assert result is False, "Empty server_crt_path should return False" | ||
| 897 | - | ||
| 898 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 899 | - server_crt_path="/nonexistent/cert.pem", | ||
| 900 | - server_key_path="" | ||
| 901 | - ) | ||
| 902 | - assert result is False, "Empty server_key_path should return False" | ||
| 903 | - | ||
| 904 | - # Test error handling: non-existent files | ||
| 905 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 906 | - server_crt_path="/nonexistent/server_cert.pem", | ||
| 907 | - server_key_path="/nonexistent/server_key.pem" | ||
| 908 | - ) | ||
| 909 | - assert result is False, "Non-existent certificate file should return False" | ||
| 910 | - | ||
| 911 | - # Test error handling: empty files | ||
| 912 | temp_dir = tempfile.mkdtemp() | 866 | temp_dir = tempfile.mkdtemp() |
| 913 | try: | 867 | try: |
| 914 | empty_cert_path = os.path.join(temp_dir, "empty_cert.pem") | 868 | empty_cert_path = os.path.join(temp_dir, "empty_cert.pem") |
| 915 | empty_key_path = os.path.join(temp_dir, "empty_key.pem") | 869 | empty_key_path = os.path.join(temp_dir, "empty_key.pem") |
| 916 | - with open(empty_cert_path, "w") as f: | 870 | + with open(empty_cert_path, "w", encoding="utf-8"): |
| 917 | pass | 871 | pass |
| 918 | - with open(empty_key_path, "w") as f: | 872 | + with open(empty_key_path, "w", encoding="utf-8"): |
| 919 | pass | 873 | pass |
| 920 | - | 874 | + |
| 921 | - result = CertValidationUtil.validate_cert_and_key( | 875 | + assert ( |
| 922 | - server_crt_path=empty_cert_path, | 876 | + CertValidationUtil.validate_cert_and_key( |
| 923 | - server_key_path=empty_key_path | 877 | + server_crt_path=empty_cert_path, |
| 878 | + server_key_path=empty_key_path, | ||
| 879 | + ) | ||
| 880 | + is False | ||
| 924 | ) | 881 | ) |
| 925 | - assert result is False, "Empty files should return False" | ||
| 926 | finally: | 882 | finally: |
| 927 | shutil.rmtree(temp_dir) | 883 | shutil.rmtree(temp_dir) |
| 928 | - | 884 | + |
| 929 | - # Test error handling: invalid formats | 885 | + |
| 886 | +def test_validate_cert_and_key_rejects_invalid_formats(): | ||
| 930 | temp_dir = tempfile.mkdtemp() | 887 | temp_dir = tempfile.mkdtemp() |
| 931 | try: | 888 | try: |
| 932 | invalid_cert_path = os.path.join(temp_dir, "invalid_cert.pem") | 889 | invalid_cert_path = os.path.join(temp_dir, "invalid_cert.pem") |
| 933 | invalid_key_path = os.path.join(temp_dir, "invalid_key.pem") | 890 | invalid_key_path = os.path.join(temp_dir, "invalid_key.pem") |
| 934 | - | 891 | + with open(invalid_cert_path, "w", encoding="utf-8") as f: |
| 935 | - with open(invalid_cert_path, "w") as f: | ||
| 936 | f.write("This is not a valid certificate") | 892 | f.write("This is not a valid certificate") |
| 937 | - with open(invalid_key_path, "w") as f: | 893 | + with open(invalid_key_path, "w", encoding="utf-8") as f: |
| 938 | - f.write("-----BEGIN PRIVATE KEY-----\ninvalid\n-----END PRIVATE KEY-----\n") | 894 | + f.write(_INVALID_PKCS8_KEY_PEM) |
| 939 | - | 895 | + |
| 940 | - result = CertValidationUtil.validate_cert_and_key( | 896 | + assert ( |
| 941 | - server_crt_path=invalid_cert_path, | 897 | + CertValidationUtil.validate_cert_and_key( |
| 942 | - server_key_path=invalid_key_path | 898 | + server_crt_path=invalid_cert_path, |
| 899 | + server_key_path=invalid_key_path, | ||
| 900 | + ) | ||
| 901 | + is False | ||
| 943 | ) | 902 | ) |
| 944 | - assert result is False, "Invalid certificate format should return False" | 903 | + |
| 945 | - | ||
| 946 | - # Test with valid cert but invalid key | ||
| 947 | valid_cert_path = os.path.join(temp_dir, "valid_cert.pem") | 904 | valid_cert_path = os.path.join(temp_dir, "valid_cert.pem") |
| 948 | - with open(valid_cert_path, "w") as f: | 905 | + with open(valid_cert_path, "w", encoding="utf-8") as f: |
| 949 | f.write("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n") | 906 | f.write("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----\n") |
| 950 | - with open(invalid_key_path, "w") as f: | 907 | + with open(invalid_key_path, "w", encoding="utf-8") as f: |
| 951 | f.write("This is not a valid private key") | 908 | f.write("This is not a valid private key") |
| 952 | - | 909 | + |
| 953 | - result = CertValidationUtil.validate_cert_and_key( | 910 | + assert ( |
| 954 | - server_crt_path=valid_cert_path, | 911 | + CertValidationUtil.validate_cert_and_key( |
| 955 | - server_key_path=invalid_key_path | 912 | + server_crt_path=valid_cert_path, |
| 913 | + server_key_path=invalid_key_path, | ||
| 914 | + ) | ||
| 915 | + is False | ||
| 956 | ) | 916 | ) |
| 957 | - assert result is False, "Invalid key format should return False" | ||
| 958 | finally: | 917 | finally: |
| 959 | shutil.rmtree(temp_dir) | 918 | shutil.rmtree(temp_dir) |
| 960 | - | 919 | + |
| 961 | - # Test error handling: mismatched cert and key | 920 | + |
| 921 | +def test_validate_cert_and_key_rejects_mismatched_key(test_certificates): | ||
| 922 | + test_certs = test_certificates | ||
| 962 | temp_dir = tempfile.mkdtemp() | 923 | temp_dir = tempfile.mkdtemp() |
| 963 | try: | 924 | try: |
| 964 | - other_key = rsa.generate_private_key( | 925 | + other_key = rsa.generate_private_key(public_exponent=65537, key_size=3072) |
| 965 | - public_exponent=65537, | ||
| 966 | - key_size=3072, | ||
| 967 | - ) | ||
| 968 | other_key_path = os.path.join(temp_dir, "other_key.pem") | 926 | other_key_path = os.path.join(temp_dir, "other_key.pem") |
| 969 | with open(other_key_path, "wb") as f: | 927 | with open(other_key_path, "wb") as f: |
| 970 | - f.write(other_key.private_bytes( | 928 | + f.write( |
| 971 | - encoding=serialization.Encoding.PEM, | 929 | + other_key.private_bytes( |
| 972 | - format=serialization.PrivateFormat.PKCS8, | 930 | + encoding=serialization.Encoding.PEM, |
| 973 | - encryption_algorithm=serialization.NoEncryption() | 931 | + format=serialization.PrivateFormat.PKCS8, |
| 974 | - )) | 932 | + encryption_algorithm=serialization.NoEncryption(), |
| 975 | - | 933 | + ) |
| 976 | - result = CertValidationUtil.validate_cert_and_key( | 934 | + ) |
| 977 | - server_crt_path=test_certs["server_cert"], | 935 | + |
| 978 | - server_key_path=other_key_path | 936 | + assert ( |
| 937 | + CertValidationUtil.validate_cert_and_key( | ||
| 938 | + server_crt_path=test_certs["server_cert"], | ||
| 939 | + server_key_path=other_key_path, | ||
| 940 | + ) | ||
| 941 | + is False | ||
| 979 | ) | 942 | ) |
| 980 | - assert result is False, "Mismatched certificate and key should return False" | ||
| 981 | finally: | 943 | finally: |
| 982 | shutil.rmtree(temp_dir) | 944 | shutil.rmtree(temp_dir) |
| 983 | - | 945 | + |
| 984 | - # Test error handling: CA-related errors | 946 | + |
| 985 | - result = CertValidationUtil.validate_cert_and_key( | 947 | +@pytest.mark.parametrize( |
| 986 | - server_crt_path=test_certs["server_cert"], | 948 | + "ca_setup", |
| 987 | - server_key_path=test_certs["server_key"], | 949 | + ["missing", "invalid-format", "mismatched", "empty-file"], |
| 988 | - ca_crt_path="/nonexistent/ca_cert.pem" | 950 | +) |
| 989 | - ) | 951 | +def test_validate_cert_and_key_rejects_ca_errors(test_certificates, ca_setup): |
| 990 | - assert result is False, "Non-existent CA certificate should return False" | 952 | + test_certs = test_certificates |
| 991 | - | 953 | + if ca_setup == "missing": |
| 992 | - temp_dir = tempfile.mkdtemp() | 954 | + ca_crt_path = "/nonexistent/ca_cert.pem" |
| 955 | + temp_dir = None | ||
| 956 | + else: | ||
| 957 | + temp_dir = tempfile.mkdtemp() | ||
| 958 | + if ca_setup == "invalid-format": | ||
| 959 | + ca_crt_path = os.path.join(temp_dir, "invalid_ca.pem") | ||
| 960 | + with open(ca_crt_path, "w", encoding="utf-8") as f: | ||
| 961 | + f.write("This is not a valid CA certificate") | ||
| 962 | + elif ca_setup == "mismatched": | ||
| 963 | + _other_ca_key, other_ca_cert = create_other_ca() | ||
| 964 | + ca_crt_path = os.path.join(temp_dir, "other_ca.pem") | ||
| 965 | + with open(ca_crt_path, "wb") as f: | ||
| 966 | + f.write(other_ca_cert.public_bytes(serialization.Encoding.PEM)) | ||
| 967 | + else: | ||
| 968 | + ca_crt_path = os.path.join(temp_dir, "empty_ca.pem") | ||
| 969 | + with open(ca_crt_path, "w", encoding="utf-8"): | ||
| 970 | + pass | ||
| 971 | + | ||
| 993 | try: | 972 | try: |
| 994 | - invalid_ca_path = os.path.join(temp_dir, "invalid_ca.pem") | 973 | + assert ( |
| 995 | - with open(invalid_ca_path, "w") as f: | 974 | + CertValidationUtil.validate_cert_and_key( |
| 996 | - f.write("This is not a valid CA certificate") | 975 | + server_crt_path=test_certs["server_cert"], |
| 997 | - | 976 | + server_key_path=test_certs["server_key"], |
| 998 | - result = CertValidationUtil.validate_cert_and_key( | 977 | + ca_crt_path=ca_crt_path, |
| 999 | - server_crt_path=test_certs["server_cert"], | 978 | + ) |
| 1000 | - server_key_path=test_certs["server_key"], | 979 | + is False |
| 1001 | - ca_crt_path=invalid_ca_path | ||
| 1002 | ) | 980 | ) |
| 1003 | - assert result is False, "Invalid CA certificate format should return False" | ||
| 1004 | - | ||
| 1005 | - # Test with mismatched CA | ||
| 1006 | - other_ca_key, other_ca_cert = create_other_ca() | ||
| 1007 | - other_ca_path = os.path.join(temp_dir, "other_ca.pem") | ||
| 1008 | - with open(other_ca_path, "wb") as f: | ||
| 1009 | - f.write(other_ca_cert.public_bytes(serialization.Encoding.PEM)) | ||
| 1010 | - | ||
| 1011 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 1012 | - server_crt_path=test_certs["server_cert"], | ||
| 1013 | - server_key_path=test_certs["server_key"], | ||
| 1014 | - ca_crt_path=other_ca_path | ||
| 1015 | - ) | ||
| 1016 | - assert result is False, "Mismatched CA certificate should return False" | ||
| 1017 | - | ||
| 1018 | - # Test with empty CA file | ||
| 1019 | - empty_ca_path = os.path.join(temp_dir, "empty_ca.pem") | ||
| 1020 | - with open(empty_ca_path, "w") as f: | ||
| 1021 | - pass | ||
| 1022 | - | ||
| 1023 | - result = CertValidationUtil.validate_cert_and_key( | ||
| 1024 | - server_crt_path=test_certs["server_cert"], | ||
| 1025 | - server_key_path=test_certs["server_key"], | ||
| 1026 | - ca_crt_path=empty_ca_path | ||
| 1027 | - ) | ||
| 1028 | - assert result is False, "Empty CA certificate file should return False" | ||
| 1029 | finally: | 981 | finally: |
| 1030 | - shutil.rmtree(temp_dir) | 982 | + if temp_dir is not None: |
| 1031 | - | 983 | + shutil.rmtree(temp_dir) |
| 1032 | - logger.info("validate_cert_and_key comprehensive test completed") | ||
| 1033 | 984 | ||
| 1034 | 985 | ||
| 1035 | def test_query_crl_info_cases(test_certificates): | 986 | def test_query_crl_info_cases(test_certificates): |
| @@ -1045,7 +996,7 @@ def test_query_crl_info_cases(test_certificates): | |||
| 1045 | ca_cert=test_certs["ca_cert_obj"], | 996 | ca_cert=test_certs["ca_cert_obj"], |
| 1046 | revoked_serial_numbers=revoked_serials, | 997 | revoked_serial_numbers=revoked_serials, |
| 1047 | next_update_days=30, | 998 | next_update_days=30, |
| 1048 | - temp_dir=test_certs["temp_dir"] | 999 | + temp_dir=test_certs["temp_dir"], |
| 1049 | ) | 1000 | ) |
| 1050 | items = CertUtil.query_crl_info(crl_info["crl_path"]) | 1001 | items = CertUtil.query_crl_info(crl_info["crl_path"]) |
| 1051 | assert isinstance(items, list), "query_crl_info should return a list" | 1002 | assert isinstance(items, list), "query_crl_info should return a list" |
| @@ -1061,7 +1012,7 @@ def test_query_crl_info_cases(test_certificates): | |||
| 1061 | ca_cert=test_certs["ca_cert_obj"], | 1012 | ca_cert=test_certs["ca_cert_obj"], |
| 1062 | revoked_serial_numbers=None, | 1013 | revoked_serial_numbers=None, |
| 1063 | next_update_days=30, | 1014 | next_update_days=30, |
| 1064 | - temp_dir=test_certs["temp_dir"] | 1015 | + temp_dir=test_certs["temp_dir"], |
| 1065 | ) | 1016 | ) |
| 1066 | empty_items = CertUtil.query_crl_info(empty_crl_info["crl_path"]) | 1017 | empty_items = CertUtil.query_crl_info(empty_crl_info["crl_path"]) |
| 1067 | assert empty_items == [], "Empty CRL should return an empty list" | 1018 | assert empty_items == [], "Empty CRL should return an empty list" |
| @@ -1078,9 +1029,9 @@ def test_construct_cert_context_with_invalid_crl_path(test_certificates): | |||
| 1078 | "tls_cert": test_certs["server_cert"], | 1029 | "tls_cert": test_certs["server_cert"], |
| 1079 | "tls_key": test_certs["server_key"], | 1030 | "tls_key": test_certs["server_key"], |
| 1080 | "tls_crl": os.path.join(test_certs["temp_dir"], "not_exist_crl.pem"), | 1031 | "tls_crl": os.path.join(test_certs["temp_dir"], "not_exist_crl.pem"), |
| 1081 | - "tls_passwd": "" | 1032 | + "tls_passwd": "", |
| 1082 | } | 1033 | } |
| 1083 | 1034 | ||
| 1084 | # Directory permissions may not satisfy strict checks on different platforms; keep returning None per existing cases | 1035 | # Directory permissions may not satisfy strict checks on different platforms; keep returning None per existing cases |
| 1085 | ssl_context = CertUtil.construct_cert_context(config) | 1036 | ssl_context = CertUtil.construct_cert_context(config) |
| 1086 | - assert ssl_context is None, "Invalid CRL path should lead to returning None" | 1037 | + assert ssl_context is None, "Invalid CRL path should lead to returning None" |
改用OpenSSL原生检查后，函数不再比较模数，建议改名如validate_cert_and_key_match。