fix: consume and report CryptographyDeprecationWarning, detect encrypted PKCS#8 keys

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]:

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"]

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)