From 95d248139fe43b9ec3bcf213e334838f99f30df7 Mon Sep 17 00:00:00 2001 From: ai-anant Date: Sun, 2 Aug 2026 20:43:09 +0000 Subject: [PATCH] fix: consume and report CryptographyDeprecationWarning, detect encrypted PKCS#8 keys - Capture CryptographyDeprecationWarning raised while loading/serializing keys (e.g. DSA) and surface them in the analysis output instead of leaking raw Python warnings to stderr. - Recognize 'BEGIN ENCRYPTED PRIVATE KEY' (OpenSSL PKCS#8 encrypted) headers so such keys report type pkcs8 instead of unknown. - Add tests covering deprecation warning capture, encrypted PKCS#8 detection, and clean warnings for normal keys. Fixes #11 --- keychecker/core/key_analyzer.py | 94 +++++++++++++++++++++++---------- keychecker/utils/output.py | 8 +++ tests/test_key_analyzer.py | 74 +++++++++++++++++++++++++- 3 files changed, 148 insertions(+), 28 deletions(-) diff --git a/keychecker/core/key_analyzer.py b/keychecker/core/key_analyzer.py index cf25967..f7b39f7 100644 --- a/keychecker/core/key_analyzer.py +++ b/keychecker/core/key_analyzer.py @@ -5,9 +5,11 @@ import os import base64 import hashlib +import warnings from typing import Dict, Any, Optional, List from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa, dsa, ec, ed25519 +from cryptography.utils import CryptographyDeprecationWarning class SSHKeyAnalyzer: @@ -46,31 +48,53 @@ def analyze_key_file(self, key_path: str) -> Dict[str, Any]: private_key = None passphrase_protected = False - # Try different key loading methods - loaders = [ - # OpenSSH format (most common for modern SSH keys) - lambda data, pwd: serialization.load_ssh_private_key(data, password=pwd), - # PEM format (traditional format) - lambda data, pwd: serialization.load_pem_private_key(data, password=pwd), - # DER format (less common) - lambda data, pwd: serialization.load_der_private_key(data, password=pwd), - ] - - for loader in loaders: - try: - # Try without passphrase first - private_key = loader(key_data, None) - break - except TypeError: - # Key requires passphrase - passphrase_protected = True - continue - except ValueError: - # Wrong format, try next loader - continue - except Exception: - # Other error, try next loader - continue # nosec B112 + # Consume deprecation warnings emitted by the cryptography library + # (e.g. CryptographyDeprecationWarning when serializing DSA keys) so + # they can be reported in the analysis output instead of leaking to + # stderr as raw Python warnings. + deprecation_warnings: List[str] = [] + + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter("always") + + # Try different key loading methods + loaders = [ + # OpenSSH format (most common for modern SSH keys) + lambda data, pwd: serialization.load_ssh_private_key( + data, password=pwd + ), + # PEM format (traditional format) + lambda data, pwd: serialization.load_pem_private_key( + data, password=pwd + ), + # DER format (less common) + lambda data, pwd: serialization.load_der_private_key( + data, password=pwd + ), + ] + + for loader in loaders: + try: + # Try without passphrase first + private_key = loader(key_data, None) + break + except TypeError: + # Key requires passphrase + passphrase_protected = True + continue + except ValueError: + # Wrong format, try next loader + continue + except Exception: + # Other error, try next loader + continue # nosec B112 + + # Collect deprecation warnings raised while loading the key + deprecation_warnings.extend( + str(w.message) + for w in caught_warnings + if issubclass(w.category, CryptographyDeprecationWarning) + ) if private_key is None and passphrase_protected: # We can still analyze the key structure without decrypting @@ -117,8 +141,18 @@ def analyze_key_file(self, key_path: str) -> Dict[str, Any]: # Get key type and size key_info = self._get_key_info(private_key, public_key) - # Generate public key string - public_key_str = self._generate_public_key_string(public_key, key_info["type"]) + # Generate public key string (this can emit CryptographyDeprecationWarning + # for deprecated algorithms like DSA, so capture it) + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter("always") + public_key_str = self._generate_public_key_string( + public_key, key_info["type"] + ) + deprecation_warnings.extend( + str(w.message) + for w in caught_warnings + if issubclass(w.category, CryptographyDeprecationWarning) + ) # Extract comment from public key file if it exists comment = self._extract_comment(key_path, public_key_str) @@ -150,6 +184,7 @@ def analyze_key_file(self, key_path: str) -> Dict[str, Any]: }, "security": security_flags, "insights": insights, + "warnings": deprecation_warnings, } return result @@ -189,6 +224,10 @@ def _analyze_encrypted_key(self, key_data: bytes, key_path: str) -> Dict[str, An elif "BEGIN EC PRIVATE KEY" in key_data_str: key_type = "ecdsa" algorithm = "ecdsa-sha2-*" + elif "BEGIN ENCRYPTED PRIVATE KEY" in key_data_str: + # PKCS#8 encrypted format (OpenSSL's default for encrypted keys) + key_type = "pkcs8" + algorithm = "unknown" elif "BEGIN PRIVATE KEY" in key_data_str: # PKCS#8 format - could be any key type key_type = "pkcs8" @@ -212,6 +251,7 @@ def _analyze_encrypted_key(self, key_data: bytes, key_path: str) -> Dict[str, An }, "security": {"encrypted": True}, "insights": public_key_info.get("insights", {}), + "warnings": [], } def _try_extract_public_key_info(self, key_path: str) -> Dict[str, Any]: diff --git a/keychecker/utils/output.py b/keychecker/utils/output.py index c4827be..93d281e 100644 --- a/keychecker/utils/output.py +++ b/keychecker/utils/output.py @@ -112,6 +112,10 @@ def _format_human_readable( for warning in security["warnings"]: lines.append(f"⚠️ {warning}") + # Library deprecation warnings (e.g. CryptographyDeprecationWarning) + for warning in result.get("warnings", []): + lines.append(f"⚠️ {warning}") + # Insights if result.get("insights"): insights = result["insights"] @@ -359,6 +363,10 @@ def _format_key_analysis_human_readable(self, result: Dict[str, Any]) -> str: for warning in security["warnings"]: lines.append(f"⚠️ {warning}") + # Library deprecation warnings (e.g. CryptographyDeprecationWarning) + for warning in result.get("warnings", []): + lines.append(f"⚠️ {warning}") + # Insights if result.get("insights"): insights = result["insights"] diff --git a/tests/test_key_analyzer.py b/tests/test_key_analyzer.py index 5ebaa59..1c12c83 100644 --- a/tests/test_key_analyzer.py +++ b/tests/test_key_analyzer.py @@ -5,8 +5,9 @@ import pytest import tempfile import os +import warnings from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import rsa, ed25519 +from cryptography.hazmat.primitives.asymmetric import rsa, dsa, ed25519 from keychecker.core.key_analyzer import SSHKeyAnalyzer @@ -109,3 +110,74 @@ def test_extract_insights_from_comment(self): assert insights["local_user"] == "user" assert insights["host"] == "hostname" assert insights["original_comment"] == comment + + def test_analyze_dsa_key_captures_deprecation_warning(self): + """Test that CryptographyDeprecationWarning is consumed and reported.""" + # Generate a DSA key (deprecated algorithm in cryptography) + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + private_key = dsa.generate_private_key(key_size=1024) + + with tempfile.NamedTemporaryFile(mode="wb", delete=False) as f: + pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + f.write(pem) + key_path = f.name + + try: + # Analyze with warnings captured by the tool + from cryptography.utils import CryptographyDeprecationWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + result = self.analyzer.analyze_key_file(key_path) + + # The deprecation warning must not leak to the caller + assert not any( + issubclass(w.category, CryptographyDeprecationWarning) for w in caught + ) + # ...but must be reported inside the result + assert result["key"]["type"] == "dsa" + assert result["warnings"], "expected deprecation warning in result" + assert any("deprecated" in w.lower() for w in result["warnings"]) + + finally: + os.unlink(key_path) + + def test_encrypted_pkcs8_pem_detected(self): + """Test that OpenSSL PKCS#8 encrypted keys are detected as pkcs8.""" + key_data = ( + b"-----BEGIN ENCRYPTED PRIVATE KEY-----\n" + b"MIHpBgsqhkiG9w0BBQ0wLwYK\n" + b"-----END ENCRYPTED PRIVATE KEY-----\n" + ) + result = self.analyzer._analyze_encrypted_key(key_data, "/tmp/fake_key") + + assert result["key"]["type"] == "pkcs8" + assert result["key"]["passphrase"] is True + assert result["warnings"] == [] + + def test_rsa_key_has_empty_warnings(self): + """Test that normal keys have no captured warnings.""" + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + ) + + with tempfile.NamedTemporaryFile(mode="wb", delete=False) as f: + pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + f.write(pem) + key_path = f.name + + try: + result = self.analyzer.analyze_key_file(key_path) + assert result["warnings"] == [] + finally: + os.unlink(key_path)