diff --git a/backend/tests/integration/conftest.py b/backend/tests/integration/conftest.py index 6c260a66..83227247 100644 --- a/backend/tests/integration/conftest.py +++ b/backend/tests/integration/conftest.py @@ -37,6 +37,8 @@ def pytest_collection_modifyitems(config, items): # applies to ALL items, not just integration tests. if "/integration/" in item.nodeid: item.add_marker(skip_integration) + + import requests from sqlalchemy import create_engine, event, text from sqlalchemy.orm import Session, sessionmaker @@ -101,6 +103,8 @@ def postgres_container(): dbname="test_translate", ) as pg: yield pg + + # #endregion postgres_container @@ -120,6 +124,8 @@ def pg_engine(postgres_container): # Cleanup Base.metadata.drop_all(bind=engine) engine.dispose() + + # #endregion pg_engine @@ -151,6 +157,8 @@ def db_session(pg_engine): session.close() transaction.rollback() connection.close() + + # #endregion db_session @@ -181,6 +189,8 @@ def mock_config_manager(): ) config_manager.get_config.return_value = MagicMock() return config_manager + + # #endregion mock_config_manager @@ -202,24 +212,21 @@ def verify_postgres_features(session: Session) -> dict: # Check FK constraints try: - result = session.execute(text( - "SELECT COUNT(*) FROM information_schema.table_constraints " - "WHERE constraint_type = 'FOREIGN KEY'" - )).scalar() + result = session.execute(text("SELECT COUNT(*) FROM information_schema.table_constraints WHERE constraint_type = 'FOREIGN KEY'")).scalar() features["foreign_keys"] = result > 0 except Exception: features["foreign_keys"] = False # Check indexes try: - result = session.execute(text( - "SELECT COUNT(*) FROM pg_indexes WHERE schemaname = 'public'" - )).scalar() + result = session.execute(text("SELECT COUNT(*) FROM pg_indexes WHERE schemaname = 'public'")).scalar() features["indexes"] = result > 0 except Exception: features["indexes"] = False return features + + # #endregion verify_postgres_features @@ -245,12 +252,20 @@ def ca_chain(): def _gen_cert_pem(subject_name, issuer_name, subject_key, issuer_key, is_ca=False, san_dns=None, san_ip=None): now = datetime.datetime.now(datetime.timezone.utc) builder = x509.CertificateBuilder() - builder = builder.subject_name(x509.Name([ - x509.NameAttribute(NameOID.COMMON_NAME, subject_name), - ])) - builder = builder.issuer_name(x509.Name([ - x509.NameAttribute(NameOID.COMMON_NAME, issuer_name), - ])) + builder = builder.subject_name( + x509.Name( + [ + x509.NameAttribute(NameOID.COMMON_NAME, subject_name), + ] + ) + ) + builder = builder.issuer_name( + x509.Name( + [ + x509.NameAttribute(NameOID.COMMON_NAME, issuer_name), + ] + ) + ) builder = builder.not_valid_before(now - datetime.timedelta(hours=1)) builder = builder.not_valid_after(now + datetime.timedelta(days=365)) builder = builder.serial_number(x509.random_serial_number()) @@ -258,20 +273,28 @@ def ca_chain(): if is_ca: builder = builder.add_extension( - x509.BasicConstraints(ca=True, path_length=None), critical=True, + x509.BasicConstraints(ca=True, path_length=None), + critical=True, ) builder = builder.add_extension( x509.KeyUsage( - key_cert_sign=True, crl_sign=True, - digital_signature=False, content_commitment=False, - key_encipherment=False, data_encipherment=False, - key_agreement=False, encipher_only=False, decipher_only=False, - ), critical=True, + key_cert_sign=True, + crl_sign=True, + digital_signature=False, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, ) else: # Server cert builder = builder.add_extension( - x509.BasicConstraints(ca=False, path_length=None), critical=True, + x509.BasicConstraints(ca=False, path_length=None), + critical=True, ) sans: list[x509.GeneralName] = [] if san_dns: @@ -279,7 +302,8 @@ def ca_chain(): if san_ip: sans.append(x509.IPAddress(san_ip)) builder = builder.add_extension( - x509.SubjectAlternativeName(sans), critical=False, + x509.SubjectAlternativeName(sans), + critical=False, ) # Subject Key Identifier — required for OpenSSL 3.x capath chain building @@ -302,16 +326,24 @@ def ca_chain(): intermediate_key = _gen_key() intermediate_crt = _gen_cert_pem( - "Test Intermediate CA", "Test Root CA", - intermediate_key, root_key, is_ca=True, + "Test Intermediate CA", + "Test Root CA", + intermediate_key, + root_key, + is_ca=True, ) server_key = _gen_key() import ipaddress + server_crt = _gen_cert_pem( - "localhost", "Test Intermediate CA", - server_key, intermediate_key, is_ca=False, - san_dns="localhost", san_ip=ipaddress.IPv4Address("127.0.0.1"), + "localhost", + "Test Intermediate CA", + server_key, + intermediate_key, + is_ca=False, + san_dns="localhost", + san_ip=ipaddress.IPv4Address("127.0.0.1"), ) # Full chain for server: server PEM + intermediate PEM (order: leaf first) @@ -322,9 +354,7 @@ def ca_chain(): server_key_encrypted = server_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, - encryption_algorithm=serialization.BestAvailableEncryption( - server_key_passphrase.encode() - ), + encryption_algorithm=serialization.BestAvailableEncryption(server_key_passphrase.encode()), ).decode() return { @@ -350,6 +380,8 @@ def ca_chain(): "server_key_passphrase": server_key_passphrase, "fullchain": fullchain, } + + # #endregion ca_chain @@ -386,10 +418,10 @@ def superset_db_url(postgres_container): pg_container.reload() pg_ip = pg_container.attrs["NetworkSettings"]["Networks"]["bridge"]["IPAddress"] pg_port = 5432 # internal port, not exposed port - superset_url = ( - f"postgresql+psycopg2://test:test@{pg_ip}:{pg_port}/superset_meta" - ) + superset_url = f"postgresql+psycopg2://test:test@{pg_ip}:{pg_port}/superset_meta" return superset_url + + # #endregion superset_db_url @@ -398,6 +430,8 @@ def superset_db_url(postgres_container): @pytest.fixture(scope="session") def superset_secret_key(): return "test-superset-secret-key-for-integration-tests-xyz" + + # #endregion superset_secret_key @@ -406,6 +440,8 @@ def superset_secret_key(): @pytest.fixture(scope="session") def superset_admin_password(): return "admin123" + + # #endregion superset_admin_password @@ -428,7 +464,7 @@ def superset_admin_password(): @pytest.fixture(scope="session") def superset_container(request, superset_db_url, superset_secret_key, superset_admin_password): """Start an Apache Superset instance for integration testing. - + Parametrize with indirect=True, {"tls": True} to enable HTTPS with custom CA chain. """ from testcontainers.core.container import DockerContainer @@ -437,36 +473,36 @@ def superset_container(request, superset_db_url, superset_secret_key, superset_a superset_image = "apache/superset:4.1.2" # Check if TLS was requested via parametrization - tls_config = getattr(request, 'param', None) - use_tls = tls_config is not None and tls_config.get('tls', False) + tls_config = getattr(request, "param", None) + use_tls = tls_config is not None and tls_config.get("tls", False) if use_tls: - ca_chain = request.getfixturevalue('ca_chain') + ca_chain = request.getfixturevalue("ca_chain") # Base64-encode certs to avoid shell escaping issues in heredoc fullchain_b64 = base64.b64encode(ca_chain["fullchain"].encode()).decode() - key_b64 = base64.b64encode(ca_chain["server_key"].encode()).decode() - _write_certs = ( - f"echo '{fullchain_b64}' | base64 -d > /tmp/server.crt && " - f"echo '{key_b64}' | base64 -d > /tmp/server.key" - ) + + if tls_config.get("encrypted_key", False): + key_pem = ca_chain["server_key_encrypted"] + key_pass = ca_chain["server_key_passphrase"] + key_b64 = base64.b64encode(key_pem.encode()).decode() + _write_certs = f"echo '{fullchain_b64}' | base64 -d > /tmp/server.crt && echo '{key_b64}' | base64 -d > /tmp/server.key" + _decrypt_key = f"export SSL_KEY_PASSPHRASE='{key_pass}' && openssl rsa -in /tmp/server.key -passin env:SSL_KEY_PASSPHRASE -out /tmp/server.key 2>/dev/null; if [ $? -ne 0 ]; then echo 'ERROR: Failed to decrypt private key with SSL_KEY_PASSPHRASE'; exit 1; fi" + else: + key_b64 = base64.b64encode(ca_chain["server_key"].encode()).decode() + _write_certs = f"echo '{fullchain_b64}' | base64 -d > /tmp/server.crt && echo '{key_b64}' | base64 -d > /tmp/server.key" + _decrypt_key = "" + _protocol = "https" - _superset_run = ( - f"superset run -h 0.0.0.0 -p 8088 " - f"--cert /tmp/server.crt --key /tmp/server.key --with-threads" - ) + _superset_run = f"superset run -h 0.0.0.0 -p 8088 --cert /tmp/server.crt --key /tmp/server.key --with-threads" else: _write_certs = "" + _decrypt_key = "" _protocol = "http" _superset_run = "superset run -h 0.0.0.0 -p 8088 --with-threads" # Superset ignores SQLALCHEMY_DATABASE_URI env var — we must create # superset_config.py and point SUPERSET_CONFIG_PATH to it. - _bootstrap_config = ( - "cat > /tmp/superset_config.py << \"PYEOF\"\n" - "import os\n" - "SQLALCHEMY_DATABASE_URI = os.environ.get(\"SUPERSET_DB_URI\", \"sqlite://\")\n" - "PYEOF\n" - ) + _bootstrap_config = 'cat > /tmp/superset_config.py << "PYEOF"\nimport os\nSQLALCHEMY_DATABASE_URI = os.environ.get("SUPERSET_DB_URI", "sqlite://")\nPYEOF\n' _export_config = "export SUPERSET_CONFIG_PATH=/tmp/superset_config.py" # ── Step 1: Init container ── @@ -476,18 +512,7 @@ def superset_container(request, superset_db_url, superset_secret_key, superset_a init.with_env("SUPERSET_LOAD_EXAMPLES", "no") init.with_env("FLASK_APP", "superset") # Note: init container does NOT need TLS certs — only runs db upgrade + init - init.with_command( - f"sh -c '" - f"python -m pip install psycopg2-binary -q && " - f"{_bootstrap_config}" - f"{_export_config} && " - f"superset db upgrade && " - f"superset fab create-admin " - f" --username admin --firstname Admin --lastname User " - f" --email admin@test.local --password {superset_admin_password} && " - f"superset init" - f"'" - ) + init.with_command(f"sh -c 'python -m pip install psycopg2-binary -q && {_bootstrap_config}{_export_config} && superset db upgrade && superset fab create-admin --username admin --firstname Admin --lastname User --email admin@test.local --password {superset_admin_password} && superset init'") try: init.start() @@ -497,9 +522,7 @@ def superset_container(request, superset_db_url, superset_secret_key, superset_a if exit_code != 0: logs = init.get_logs() init.stop() - raise RuntimeError( - f"Superset init container failed (exit {exit_code}):\n{logs}" - ) + raise RuntimeError(f"Superset init container failed (exit {exit_code}):\n{logs}") finally: init.stop() @@ -510,136 +533,17 @@ def superset_container(request, superset_db_url, superset_secret_key, superset_a superset.with_env("SUPERSET_LOAD_EXAMPLES", "no") superset.with_env("FLASK_APP", "superset") - # Build web server command: optionally write certs, then start superset + # Build web server command: optionally write certs, decrypt key, then start superset if use_tls and _write_certs: - web_cmd = ( - f"sh -c '" - f"python -m pip install psycopg2-binary -q && " - f"{_write_certs} && " - f"{_bootstrap_config}" - f"{_export_config} && " - f"{_superset_run}'" - ) - else: - web_cmd = ( - f"sh -c '" - f"python -m pip install psycopg2-binary -q && " - f"{_bootstrap_config}" - f"{_export_config} && " - f"{_superset_run}'" - ) - - superset.with_command(web_cmd) - superset.with_exposed_ports(8088) - - superset.start() - - # Store protocol on the container object for superset_url fixture - superset._ss_protocol = _protocol - - # Wait for the /health endpoint to respond - host = superset.get_container_host_ip() - port = superset.get_exposed_port(8088) - health_url = f"{_protocol}://{host}:{port}/health" - - deadline = time.time() + 90 - last_error = None - - while time.time() < deadline: - try: - # verify=False in TLS mode because the custom CA hasn't been installed yet - resp = requests.get(health_url, timeout=3, verify=False) - if resp.status_code == 200: - break - except requests.RequestException as e: - last_error = e - time.sleep(2) - else: - logs = superset.get_logs() - superset.stop() - raise TimeoutError( - f"Superset did not become healthy within 90s. " - f"Last error: {last_error}\nContainer logs:\n{logs}" - ) - - yield superset - - superset.stop() -# #endregion superset_container - - -# #region superset_url [C:1] [TYPE Fixture] -# @BRIEF Function-scoped — base URL of the running Superset container (auto-detects http/https). -@pytest.fixture -def superset_url(superset_container): - host = superset_container.get_container_host_ip() - port = superset_container.get_exposed_port(8088) - protocol = getattr(superset_container, '_ss_protocol', 'http') - return f"{protocol}://{host}:{port}" -# #endregion superset_url - - -# #region superset_admin_headers [C:2] [TYPE Fixture] -# @BRIEF Function-scoped — authenticated session headers for Superset admin API calls. -@pytest.fixture -def superset_admin_headers(superset_url, superset_admin_password): - """Log in as admin and return headers with CSRF token and session cookie.""" - session = requests.Session() - - # Step 1: Get CSRF token from login page - login_url = f"{superset_url}/login/" - resp = session.get(login_url, timeout=10) - resp.raise_for_status() - - csrf_token = None - if "csrf_token" in resp.text: - import re - match = re.search(r'csrf_token"\s*:\s*"([^"]+)"', resp.text) - if match: - csrf_token = match.group(1) - - # Step 2: Submit login form - resp = session.post( - login_url, - data={ - "username": "admin", - "password": superset_admin_password, - "csrf_token": csrf_token or "", - }, - headers={"Referer": login_url}, - timeout=10, - ) - resp.raise_for_status() - - return session -# #endregion superset_admin_headers - - -# ── JWT-based Superset Fixtures ──────────────────────────────────── -# ADR-0012: после установки psycopg2 + superset_config.py + Docker bridge IP, -# JWT-авторизация (/api/v1/security/login) работает в тестовом контейнере. - - -# #region superset_jwt_headers [C:2] [TYPE Fixture] -# @BRIEF Function-scoped — JWT Bearer token headers for Superset REST API. -# @POST Returns dict with Authorization: Bearer . -@pytest.fixture -def superset_jwt_headers(superset_url, superset_admin_password): - """Obtain JWT access token from the running Superset container.""" - resp = requests.post( - f"{superset_url}/api/v1/security/login", - json={ - "username": "admin", - "password": superset_admin_password, - "provider": "db", - }, - timeout=10, - ) + _decrypt_prefix = f"{_decrypt_key} && " if _decrypt_key else "" + web_cmd = f"sh -c 'python -m pip install psycopg2-binary -q && {_write_certs} && {_decrypt_prefix}{_bootstrap_config}{_export_config} && {_superset_run}'" resp.raise_for_status() data = resp.json() access_token = data["access_token"] assert access_token, "No access_token in JWT response" return {"Authorization": f"Bearer {access_token}"} + + # #endregion superset_jwt_headers @@ -659,6 +563,8 @@ def superset_env(superset_url, superset_admin_password): verify_ssl=False, timeout=30, ) + + # #endregion superset_env @@ -672,6 +578,8 @@ async def superset_client(superset_env): client = SupersetClient(superset_env) await client.authenticate() return client + + # #endregion superset_client @@ -687,7 +595,7 @@ async def superset_client(superset_env): @pytest.fixture def install_custom_ca(ca_chain): """Install the custom Root CA into the system trust store. - + Strategy 1 (prod-like): write .crt → update-ca-certificates (requires root). Strategy 2 (fallback): SSL_CERT_DIR with hash symlinks. """ @@ -708,7 +616,9 @@ def install_custom_ca(ca_chain): ca_path.write_text(ca_pem) subprocess.run( ["update-ca-certificates", "--fresh"], - check=True, timeout=30, capture_output=True, + check=True, + timeout=30, + capture_output=True, ) ca_verify_path = "/etc/ssl/certs" used_update_ca = True @@ -721,7 +631,10 @@ def install_custom_ca(ca_chain): # Create hash symlink for capath result = subprocess.run( ["openssl", "x509", "-hash", "-noout"], - input=ca_pem, capture_output=True, text=True, timeout=10, + input=ca_pem, + capture_output=True, + text=True, + timeout=10, ) cert_hash = result.stdout.strip() symlink_path = cert_dir / f"{cert_hash}.0" @@ -747,7 +660,8 @@ def install_custom_ca(ca_chain): try: subprocess.run( ["update-ca-certificates", "--fresh"], - timeout=30, capture_output=True, + timeout=30, + capture_output=True, ) except Exception: pass @@ -756,7 +670,10 @@ def install_custom_ca(ca_chain): del os.environ["SSL_CERT_DIR"] if cert_dir and cert_dir.exists(): import shutil + shutil.rmtree(str(cert_dir), ignore_errors=True) + + # #endregion install_custom_ca diff --git a/backend/tests/integration/test_superset_tls_custom_ca.py b/backend/tests/integration/test_superset_tls_custom_ca.py index 271a13c9..d8b73cfa 100644 --- a/backend/tests/integration/test_superset_tls_custom_ca.py +++ b/backend/tests/integration/test_superset_tls_custom_ca.py @@ -48,28 +48,29 @@ class TestCustomCATrustChain: result = subprocess.run( [ - "openssl", "s_client", - "-connect", f"{host}:{port}", - "-servername", host, - "-CApath", ca_path, + "openssl", + "s_client", + "-connect", + f"{host}:{port}", + "-servername", + host, + "-CApath", + ca_path, "-verify_return_error", ], - input="", capture_output=True, text=True, timeout=15, + input="", + capture_output=True, + text=True, + timeout=15, ) # Parse Verify return code from output match = re.search(r"Verify return code:\s*(\d+)", result.stderr + result.stdout) - assert match is not None, ( - f"No Verify return code in openssl output.\n" - f"stdout: {result.stdout[:2000]}\nstderr: {result.stderr[:2000]}" - ) + assert match is not None, f"No Verify return code in openssl output.\nstdout: {result.stdout[:2000]}\nstderr: {result.stderr[:2000]}" code = int(match.group(1)) - assert code == 0, ( - f"Expected verify code 0, got {code}.\n" - f"Full output:\n{result.stderr[:2000]}" - ) - # #endregion test_openssl_verify_capath_ok + assert code == 0, f"Expected verify code 0, got {code}.\nFull output:\n{result.stderr[:2000]}" + # #endregion test_openssl_verify_capath_ok # #region test_openssl_verify_cafile_certifi_fails [C:2] [TYPE Function] # @BRIEF openssl s_client -CAfile with certifi's bundle fails. @@ -81,34 +82,35 @@ class TestCustomCATrustChain: This proves that our custom CA is NOT trusted by certifi, which is what httpx used before our patches. After patches, we use system CA store.""" import certifi + parsed = urlparse(superset_url) host = parsed.hostname port = parsed.port or 443 result = subprocess.run( [ - "openssl", "s_client", - "-connect", f"{host}:{port}", - "-servername", host, - "-CAfile", certifi.where(), + "openssl", + "s_client", + "-connect", + f"{host}:{port}", + "-servername", + host, + "-CAfile", + certifi.where(), "-verify_return_error", ], - input="", capture_output=True, text=True, timeout=15, + input="", + capture_output=True, + text=True, + timeout=15, ) match = re.search(r"Verify return code:\s*(\d+)", result.stderr + result.stdout) - assert match is not None, ( - f"No Verify return code in openssl output.\n" - f"stdout: {result.stdout[:2000]}\nstderr: {result.stderr[:2000]}" - ) + assert match is not None, f"No Verify return code in openssl output.\nstdout: {result.stdout[:2000]}\nstderr: {result.stderr[:2000]}" code = int(match.group(1)) - assert code != 0, ( - "Expected non-zero verify code with certifi CA bundle, got 0.\n" - f"This means certifi somehow trusts our generated CA, which should not happen.\n" - f"Full output:\n{result.stderr[:2000]}" - ) - # #endregion test_openssl_verify_cafile_certifi_fails + assert code != 0, f"Expected non-zero verify code with certifi CA bundle, got 0.\nThis means certifi somehow trusts our generated CA, which should not happen.\nFull output:\n{result.stderr[:2000]}" + # #endregion test_openssl_verify_cafile_certifi_fails # ── httpx SSL context tests ────────────────────────────────────── @@ -131,8 +133,8 @@ class TestCustomCATrustChain: assert resp.text.strip() == "OK" asyncio.run(_test()) - # #endregion test_httpx_capath_works + # #endregion test_httpx_capath_works # #region test_httpx_certifi_fails [C:2] [TYPE Function] # @BRIEF httpx with certifi's CA bundle must fail. @@ -157,8 +159,8 @@ class TestCustomCATrustChain: await client.get(f"{superset_url}/health") asyncio.run(_test()) - # #endregion test_httpx_certifi_fails + # #endregion test_httpx_certifi_fails # ── superset-tools client tests ──────────────────────────────────────── @@ -200,8 +202,8 @@ class TestCustomCATrustChain: await client.aclose() asyncio.run(_test()) - # #endregion test_asyncapiclient_verify_true + # #endregion test_asyncapiclient_verify_true # #region test_superset_client_full_auth [C:2] [TYPE Function] # @BRIEF Full SupersetClient auth + API call over TLS with verify_ssl=True. @@ -233,9 +235,7 @@ class TestCustomCATrustChain: assert "csrf_token" in tokens, "No csrf_token" # Fetch dashboards list via the authenticated client - result = await client.get_dashboards( - query={"page": 0, "page_size": 10} - ) + result = await client.get_dashboards(query={"page": 0, "page_size": 10}) # get_dashboards returns (count, items) tuple assert isinstance(result, (dict, tuple)), f"Expected dict or tuple, got {type(result)}" @@ -247,8 +247,8 @@ class TestCustomCATrustChain: await client.aclose() asyncio.run(_test()) - # #endregion test_superset_client_full_auth + # #endregion test_superset_client_full_auth # ── Regression tests ───────────────────────────────────────────── @@ -274,12 +274,9 @@ class TestCustomCATrustChain: with open(certifi.where(), "rb") as f: bundle_data = f.read() - assert our_fingerprint not in bundle_data.hex(), ( - "Our custom Root CA was found in certifi bundle — " - "this should not happen since the CA was just generated at test time." - ) - # #endregion test_certifi_bundle_notrust + assert our_fingerprint not in bundle_data.hex(), "Our custom Root CA was found in certifi bundle — this should not happen since the CA was just generated at test time." + # #endregion test_certifi_bundle_notrust # #region test_verify_false_still_works [C:1] [TYPE Function] # @BRIEF verify_ssl=False must still work (regression test). @@ -298,11 +295,78 @@ class TestCustomCATrustChain: assert resp.text.strip() == "OK" asyncio.run(_test()) + # #endregion test_verify_false_still_works + # #endregion TestCustomCATrustChain +# #region TestEncryptedPrivateKeySuperset [C:3] [TYPE Class] [SEMANTICS test,tls,ssl,encrypted-key,passphrase,superset] +# @BRIEF Integration test for Superset container with passphrase-protected +# private key, decrypted via SSL_KEY_PASSPHRASE env var. +# Validates the production flow: encrypted key → openssl decrypt → TLS handshake. +# @RELATION BINDS_TO -> [superset_container] +# @RELATION BINDS_TO -> [ca_chain] +# @RELATION BINDS_TO -> [install_custom_ca] +# @RELATION BINDS_TO -> [ADR-0009] +# @RATIONALE Corporate PKI often distributes passphrase-protected keys. +# superset-tools entrypoint decrypts these with SSL_KEY_PASSPHRASE. +# This test validates the full pipeline: encrypted key in container → +# openssl decrypt via env var → Flask/Werkzeug TLS server → client trust. +@pytest.mark.parametrize("superset_container", [{"tls": True, "encrypted_key": True}], indirect=True) +class TestEncryptedPrivateKeySuperset: + """Superset container with encrypted private key + SSL_KEY_PASSPHRASE.""" + + def test_superset_health_over_tls_with_encrypted_key(self, superset_url, install_custom_ca): + """Health endpoint must respond 200 OK over HTTPS with encrypted key.""" + import asyncio + import httpx + + ca_verify_path = install_custom_ca["ca_verify_path"] + ctx = ssl.create_default_context(capath=ca_verify_path) + + async def _test(): + async with httpx.AsyncClient(verify=ctx, timeout=10) as client: + resp = await client.get(f"{superset_url}/health") + assert resp.status_code == 200 + assert resp.text.strip() == "OK" + + asyncio.run(_test()) + + def test_openssl_capath_encrypted_key(self, superset_url, install_custom_ca): + """openssl s_client -CApath must succeed with encrypted key container.""" + ca_verify_path = install_custom_ca["ca_verify_path"] + parsed = urlparse(superset_url) + host = parsed.hostname + port = parsed.port or 443 + + result = subprocess.run( + [ + "openssl", + "s_client", + "-connect", + f"{host}:{port}", + "-servername", + host, + "-CApath", + ca_verify_path, + "-verify_return_error", + ], + input="", + capture_output=True, + text=True, + timeout=15, + ) + + match = re.search(r"Verify return code:\s*(\d+)", result.stderr + result.stdout) + assert match is not None, "No Verify return code in openssl output" + assert int(match.group(1)) == 0, f"Expected verify code 0 with encrypted key container, got {match.group(1)}" + + +# #endregion TestEncryptedPrivateKeySuperset + + # #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, @@ -338,8 +402,8 @@ class TestEncryptedPrivateKey: password=ca_chain["server_key_passphrase"], ) # No exception raised — key loaded successfully - # #endregion test_ssl_context_loads_encrypted_key + # #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. @@ -359,8 +423,8 @@ class TestEncryptedPrivateKey: keyfile=str(key_file), password="wrong-passphrase-99999", ) - # #endregion test_encrypted_key_wrong_passphrase_fails + # #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. @@ -382,8 +446,8 @@ class TestEncryptedPrivateKey: keyfile=str(key_file), # no password argument — should fail ) - # #endregion test_encrypted_key_no_passphrase_fails + # #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. @@ -437,7 +501,10 @@ class TestEncryptedPrivateKey: # Start the server with async context manager server = await asyncio.start_server( - handle, host="127.0.0.1", port=port, ssl=server_ctx, + handle, + host="127.0.0.1", + port=port, + ssl=server_ctx, ) async with server: @@ -446,22 +513,22 @@ class TestEncryptedPrivateKey: # Connect and verify TLS handshake reader, writer = await asyncio.open_connection( - "127.0.0.1", port, ssl=client_ctx, + "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]}" - ) + 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 + # #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. @@ -479,21 +546,23 @@ class TestEncryptedPrivateKey: result = subprocess.run( [ - "openssl", "rsa", - "-in", str(key_path), + "openssl", + "rsa", + "-in", + str(key_path), "-check", - "-passin", "pass:wrong-passphrase-99999", + "-passin", + "pass:wrong-passphrase-99999", ], - capture_output=True, text=True, timeout=10, + 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]}" - ) + assert result.returncode != 0, f"Expected openssl rsa to fail with wrong passphrase, 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. @@ -512,18 +581,20 @@ class TestEncryptedPrivateKey: # Use stdin=DEVNULL to simulate EOF — openssl will fail. result = subprocess.run( [ - "openssl", "rsa", - "-in", str(key_path), + "openssl", + "rsa", + "-in", + str(key_path), "-check", ], - capture_output=True, text=True, timeout=10, + 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]}" - ) + assert result.returncode != 0, f"Expected openssl rsa to fail without passphrase, but it exited with code 0.\nstderr: {result.stderr[:2000]}" + # #endregion test_openssl_key_no_passphrase_fails