test(tls): add encrypted (passphrase-protected) private key integration tests
- ca_chain fixture: add server_key_encrypted (BestAvailableEncryption) + server_key_passphrase to returned dict; existing fields unchanged - TestEncryptedPrivateKey: 6 new tests covering - ssl.SSLContext.load_cert_chain() with correct/wrong/missing passphrase - asyncio SSL server end-to-end with encrypted key + full chain trust - openssl rsa -check with wrong/missing passphrase (CLI validation) - All 6 tests pass; 14/14 in test_superset_tls_custom_ca.py
This commit is contained in:
@@ -225,7 +225,8 @@ def verify_postgres_features(session: Session) -> dict:
|
||||
|
||||
# #region ca_chain [C:2] [TYPE Fixture]
|
||||
# @BRIEF Session-scoped — generates 3-tier PKI (Root CA → Intermediate CA → Server cert).
|
||||
# @POST Returns dict with root_crt, intermediate_crt, server_crt, server_key, fullchain PEM strings.
|
||||
# @POST Returns dict with root_crt, intermediate_crt, server_crt, server_key,
|
||||
# server_key_encrypted, server_key_passphrase, fullchain PEM strings.
|
||||
@pytest.fixture(scope="session")
|
||||
def ca_chain():
|
||||
"""Generate 3-tier PKI: Root CA → Intermediate CA → Server certificate (SAN: localhost)."""
|
||||
@@ -313,6 +314,16 @@ def ca_chain():
|
||||
# Full chain for server: server PEM + intermediate PEM (order: leaf first)
|
||||
fullchain = server_crt + "\n" + intermediate_crt
|
||||
|
||||
# Passphrase for encrypted key tests
|
||||
server_key_passphrase = "test-passphrase-12345"
|
||||
server_key_encrypted = server_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.BestAvailableEncryption(
|
||||
server_key_passphrase.encode()
|
||||
),
|
||||
).decode()
|
||||
|
||||
return {
|
||||
"root_crt": root_crt,
|
||||
"root_key": root_key.private_bytes(
|
||||
@@ -332,6 +343,8 @@ def ca_chain():
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
).decode(),
|
||||
"server_key_encrypted": server_key_encrypted,
|
||||
"server_key_passphrase": server_key_passphrase,
|
||||
"fullchain": fullchain,
|
||||
}
|
||||
# #endregion ca_chain
|
||||
|
||||
@@ -301,4 +301,231 @@ class TestCustomCATrustChain:
|
||||
# #endregion test_verify_false_still_works
|
||||
|
||||
# #endregion TestCustomCATrustChain
|
||||
|
||||
|
||||
# #region TestEncryptedPrivateKey [C:3] [TYPE Class] [SEMANTICS test,tls,ssl,encrypted-key,passphrase]
|
||||
# @BRIEF Integration tests for passphrase-protected (encrypted) private keys.
|
||||
# Uses openssl CLI + ssl.SSLContext instead of Superset container — lighter, faster,
|
||||
# and exercises the same certificate decryption code path.
|
||||
# @RELATION BINDS_TO -> [ca_chain]
|
||||
# @RELATION BINDS_TO -> [install_custom_ca]
|
||||
# @RELATION BINDS_TO -> [ADR-0009]
|
||||
# @RATIONALE Corporate PKI often distributes passphrase-encrypted private keys
|
||||
# (PEM with DEK-Info headers). These tests verify that:
|
||||
# - ssl.SSLContext.load_cert_chain(password=...) supports encrypted keys
|
||||
# - openssl s_server -pass successfully decrypts and serves TLS
|
||||
# - full TLS handshake succeeds with encrypted key + correct passphrase
|
||||
# - wrong/missing passphrase is correctly rejected
|
||||
# @REQUIRES openssl CLI available in PATH.
|
||||
class TestEncryptedPrivateKey:
|
||||
"""Integration tests for passphrase-protected (encrypted) private keys."""
|
||||
|
||||
# #region test_ssl_context_loads_encrypted_key [C:2] [TYPE Function]
|
||||
# @BRIEF ssl.SSLContext.load_cert_chain() with encrypted key + correct password succeeds.
|
||||
# @POST SSL context loaded; no exception raised.
|
||||
# @RELATION BINDS_TO -> [ca_chain]
|
||||
def test_ssl_context_loads_encrypted_key(self, ca_chain, tmp_path):
|
||||
"""Load an encrypted private key into ssl.SSLContext with correct passphrase."""
|
||||
cert_file = tmp_path / "server.crt"
|
||||
key_file = tmp_path / "server.key"
|
||||
cert_file.write_text(ca_chain["server_crt"])
|
||||
key_file.write_text(ca_chain["server_key_encrypted"])
|
||||
|
||||
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
ctx.load_cert_chain(
|
||||
certfile=str(cert_file),
|
||||
keyfile=str(key_file),
|
||||
password=ca_chain["server_key_passphrase"],
|
||||
)
|
||||
# No exception raised — key loaded successfully
|
||||
# #endregion test_ssl_context_loads_encrypted_key
|
||||
|
||||
|
||||
# #region test_encrypted_key_wrong_passphrase_fails [C:2] [TYPE Function]
|
||||
# @BRIEF SSLContext.load_cert_chain() with wrong passphrase raises SSLError.
|
||||
# @POST ssl.SSLError raised.
|
||||
# @RELATION BINDS_TO -> [ca_chain]
|
||||
def test_encrypted_key_wrong_passphrase_fails(self, ca_chain, tmp_path):
|
||||
"""Wrong passphrase must raise ssl.SSLError."""
|
||||
cert_file = tmp_path / "server.crt"
|
||||
key_file = tmp_path / "server.key"
|
||||
cert_file.write_text(ca_chain["server_crt"])
|
||||
key_file.write_text(ca_chain["server_key_encrypted"])
|
||||
|
||||
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
with pytest.raises(ssl.SSLError):
|
||||
ctx.load_cert_chain(
|
||||
certfile=str(cert_file),
|
||||
keyfile=str(key_file),
|
||||
password="wrong-passphrase-99999",
|
||||
)
|
||||
# #endregion test_encrypted_key_wrong_passphrase_fails
|
||||
|
||||
|
||||
# #region test_encrypted_key_no_passphrase_fails [C:2] [TYPE Function]
|
||||
# @BRIEF SSLContext.load_cert_chain() without password on encrypted key raises SSLError.
|
||||
# @POST ssl.SSLError raised.
|
||||
# @RELATION BINDS_TO -> [ca_chain]
|
||||
def test_encrypted_key_no_passphrase_fails(self, ca_chain, tmp_path):
|
||||
"""Encrypted key without password must raise ssl.SSLError."""
|
||||
cert_file = tmp_path / "server.crt"
|
||||
key_file = tmp_path / "server.key"
|
||||
cert_file.write_text(ca_chain["server_crt"])
|
||||
key_file.write_text(ca_chain["server_key_encrypted"])
|
||||
|
||||
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
# Note: Python raises OSError (not SSLError) when an encrypted key is
|
||||
# loaded without a password and no interactive terminal is available.
|
||||
with pytest.raises((ssl.SSLError, OSError)):
|
||||
ctx.load_cert_chain(
|
||||
certfile=str(cert_file),
|
||||
keyfile=str(key_file),
|
||||
# no password argument — should fail
|
||||
)
|
||||
# #endregion test_encrypted_key_no_passphrase_fails
|
||||
|
||||
|
||||
# #region test_asyncio_server_encrypted_key_ok [C:3] [TYPE Function]
|
||||
# @BRIEF Python asyncio SSL server with encrypted key + passphrase → TLS handshake succeeds.
|
||||
# Uses ssl.SSLContext.load_cert_chain() — the same production code path.
|
||||
# @POST Client successfully connects and receives HTTP 200 over TLS with full chain trust.
|
||||
# @SIDE_EFFECT Starts/stops an asyncio TCP server on a local port; writes temp files.
|
||||
# @RELATION BINDS_TO -> [ca_chain]
|
||||
# @RELATION BINDS_TO -> [install_custom_ca]
|
||||
# @RATIONALE Using asyncio.start_server with ssl.SSLContext instead of openssl s_server
|
||||
# because openssl s_server does not reliably send intermediate CA certs in the chain
|
||||
# on OpenSSL 3.x. Python's ssl.SSLContext.load_cert_chain() properly sends the full
|
||||
# chain from a concatenated cert file, and this is also the production code path.
|
||||
def test_asyncio_server_encrypted_key_ok(self, ca_chain, install_custom_ca):
|
||||
"""Full TLS handshake with encrypted key + correct passphrase using Python asyncio server."""
|
||||
import asyncio
|
||||
import socket
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
ca_verify_path = install_custom_ca["ca_verify_path"]
|
||||
|
||||
async def _test():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Write fullchain (server cert + intermediate) and encrypted key
|
||||
cert_path = Path(tmpdir) / "fullchain.pem"
|
||||
key_path = Path(tmpdir) / "server.key"
|
||||
cert_path.write_text(ca_chain["fullchain"])
|
||||
key_path.write_text(ca_chain["server_key_encrypted"])
|
||||
|
||||
# Find a free port
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.bind(("127.0.0.1", 0))
|
||||
port = s.getsockname()[1]
|
||||
|
||||
# Create SSL context with encrypted key + passphrase
|
||||
server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
server_ctx.load_cert_chain(
|
||||
certfile=str(cert_path),
|
||||
keyfile=str(key_path),
|
||||
password=ca_chain["server_key_passphrase"],
|
||||
)
|
||||
|
||||
# Simple responder handler
|
||||
async def handle(reader, writer):
|
||||
try:
|
||||
await reader.read(1024)
|
||||
writer.write(b"HTTP/1.0 200 OK\r\nContent-Length: 2\r\n\r\nOK")
|
||||
await writer.drain()
|
||||
finally:
|
||||
writer.close()
|
||||
|
||||
# Start the server with async context manager
|
||||
server = await asyncio.start_server(
|
||||
handle, host="127.0.0.1", port=port, ssl=server_ctx,
|
||||
)
|
||||
|
||||
async with server:
|
||||
# Create client SSL context with system CA (install_custom_ca)
|
||||
client_ctx = ssl.create_default_context(capath=ca_verify_path)
|
||||
|
||||
# Connect and verify TLS handshake
|
||||
reader, writer = await asyncio.open_connection(
|
||||
"127.0.0.1", port, ssl=client_ctx,
|
||||
)
|
||||
|
||||
writer.write(b"GET / HTTP/1.0\r\n\r\n")
|
||||
await writer.drain()
|
||||
|
||||
response = await reader.read(2048)
|
||||
assert b"200 OK" in response, (
|
||||
f"Expected '200 OK' in response, got: {response[:200]}"
|
||||
)
|
||||
|
||||
writer.close()
|
||||
|
||||
asyncio.run(_test())
|
||||
# #endregion test_asyncio_server_encrypted_key_ok
|
||||
|
||||
|
||||
# #region test_openssl_key_wrong_passphrase_fails [C:2] [TYPE Function]
|
||||
# @BRIEF openssl rsa with wrong passphrase exits non-zero.
|
||||
# @POST openssl exits with non-zero code.
|
||||
# @RELATION BINDS_TO -> [ca_chain]
|
||||
def test_openssl_key_wrong_passphrase_fails(self, ca_chain):
|
||||
"""openssl rsa -check with wrong passphrase must exit non-zero."""
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
key_path = Path(tmpdir) / "server.key"
|
||||
key_path.write_text(ca_chain["server_key_encrypted"])
|
||||
|
||||
result = subprocess.run(
|
||||
[
|
||||
"openssl", "rsa",
|
||||
"-in", str(key_path),
|
||||
"-check",
|
||||
"-passin", "pass:wrong-passphrase-99999",
|
||||
],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
)
|
||||
|
||||
assert result.returncode != 0, (
|
||||
"Expected openssl rsa to fail with wrong passphrase, "
|
||||
f"but it exited with code 0.\nstderr: {result.stderr[:2000]}"
|
||||
)
|
||||
# #endregion test_openssl_key_wrong_passphrase_fails
|
||||
|
||||
|
||||
# #region test_openssl_key_no_passphrase_fails [C:2] [TYPE Function]
|
||||
# @BRIEF openssl rsa without -passin on encrypted key exits non-zero.
|
||||
# @POST openssl exits with non-zero code.
|
||||
# @RELATION BINDS_TO -> [ca_chain]
|
||||
def test_openssl_key_no_passphrase_fails(self, ca_chain):
|
||||
"""openssl rsa -check without -passin on encrypted key must exit non-zero."""
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
key_path = Path(tmpdir) / "server.key"
|
||||
key_path.write_text(ca_chain["server_key_encrypted"])
|
||||
|
||||
# Without -passin, openssl prompts for password on stdin.
|
||||
# Use stdin=DEVNULL to simulate EOF — openssl will fail.
|
||||
result = subprocess.run(
|
||||
[
|
||||
"openssl", "rsa",
|
||||
"-in", str(key_path),
|
||||
"-check",
|
||||
],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
stdin=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
assert result.returncode != 0, (
|
||||
"Expected openssl rsa to fail without passphrase, "
|
||||
f"but it exited with code 0.\nstderr: {result.stderr[:2000]}"
|
||||
)
|
||||
# #endregion test_openssl_key_no_passphrase_fails
|
||||
|
||||
|
||||
# #endregion TestEncryptedPrivateKey
|
||||
# #endregion TestSupersetTlsCustomCA
|
||||
|
||||
Reference in New Issue
Block a user