From 1e09a2c4b80d0e0e702640b94710e50a1269e725 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 13 Jul 2026 18:39:06 +0000 Subject: [PATCH 1/4] feat(security): implement SEC13-01 mutual TLS validation Add MutualTlsCheck for north-south and east-west mTLS (or equivalent), with a shared probe, AWS wrapper (east-west provider-hidden by default), and my-isv DEMO_MODE wiring. Closes #309 Signed-off-by: Cursor Agent Co-authored-by: Alexandre Begnoche --- docs/test-plan.yaml | 2 + .../providers/aws/config/security.yaml | 27 +- .../aws/scripts/security/mutual_tls_test.py | 131 ++++++ .../providers/my-isv/config/security.yaml | 16 + .../providers/shared/mutual_tls_test.py | 434 ++++++++++++++++++ isvctl/configs/suites/README.md | 2 + isvctl/configs/suites/security.yaml | 7 + isvtest/src/isvtest/validations/__init__.py | 2 + isvtest/src/isvtest/validations/security.py | 51 ++ isvtest/tests/test_security.py | 89 ++++ scripts/tests/test_mutual_tls_probe.py | 169 +++++++ 11 files changed, 922 insertions(+), 8 deletions(-) create mode 100755 isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py create mode 100755 isvctl/configs/providers/shared/mutual_tls_test.py create mode 100644 scripts/tests/test_mutual_tls_probe.py diff --git a/docs/test-plan.yaml b/docs/test-plan.yaml index 8bc32cc88..2ba6749c1 100644 --- a/docs/test-plan.yaml +++ b/docs/test-plan.yaml @@ -3727,6 +3727,8 @@ domains: - summary: Verify mTLS (or equivalent) for all east-west and north-south traffic labels: - min_req + - network + - security priority: P1 milestone: M5 notes: "" diff --git a/isvctl/configs/providers/aws/config/security.yaml b/isvctl/configs/providers/aws/config/security.yaml index 99599e311..c14e6d066 100644 --- a/isvctl/configs/providers/aws/config/security.yaml +++ b/isvctl/configs/providers/aws/config/security.yaml @@ -31,13 +31,14 @@ # 3. BMC Protocol Security - Verify CNP10-01 BMC protocol posture # 4. BMC Bastion Access - Verify BMC reachable only via hardened bastion # 5. API Endpoint Isolation - Verify management APIs are not publicly exposed -# 6. Insecure Protocols - Verify SSLv3/TLS1.0/TLS1.1/plain HTTP disabled (SEC13-02) -# 7. MFA Enforcement - Verify admin interfaces (UI, CLI, API) require MFA -# 8. Cert Rotation Cycle - Verify cert rotation cycle / auto-renewal (SEC09-01) -# 9. KMS Options - Verify provider- and customer-managed keys (SEC09-02) -# 10. Centralized KMS - Verify encrypted resources resolve to KMS (SEC09-03) -# 11. Customer Managed Key - Verify customer-managed key encryption (SEC09-04) -# 12. Least Privilege - Verify scoped user/resource/network policies (SEC04-01/02) +# 6. Mutual TLS - Verify mTLS north-south/east-west (SEC13-01) +# 7. Insecure Protocols - Verify SSLv3/TLS1.0/TLS1.1/plain HTTP disabled (SEC13-02) +# 8. MFA Enforcement - Verify admin interfaces (UI, CLI, API) require MFA +# 9. Cert Rotation Cycle - Verify cert rotation cycle / auto-renewal (SEC09-01) +# 10. KMS Options - Verify provider- and customer-managed keys (SEC09-02) +# 11. Centralized KMS - Verify encrypted resources resolve to KMS (SEC09-03) +# 12. Customer Managed Key - Verify customer-managed key encryption (SEC09-04) +# 13. Least Privilege - Verify scoped user/resource/network policies (SEC04-01/02) # 13. Audit Logging - Verify management-event logging and retention (SEC08-01/02) # 14. SA Credential Auth - Verify IAM user + long-lived access key auth # 15. OIDC User Auth - Probe configured OIDC-protected platform endpoint (SEC01-01) @@ -105,7 +106,17 @@ commands: args: ["--region", "{{region}}"] timeout: 60 - # Test 6: Insecure Protocols (SEC13-02) + # Test 6a: Mutual TLS (SEC13-01) + # Probes EDGE_ENDPOINTS / EAST_WEST_ENDPOINTS with MTLS_* cert material. + # Without endpoints the script emits a structured skip. East-west is + # provider-hidden on AWS unless EAST_WEST_ENDPOINTS is set. + - name: mutual_tls_test + phase: test + command: "python3 ../scripts/security/mutual_tls_test.py" + args: ["--region", "{{region}}"] + timeout: 120 + + # Test 6b: Insecure Protocols (SEC13-02) # Provisions a temporary TLS-only ALB fixture, probes it for legacy # protocol refusal, then tears it down. When the orchestrator principal # lacks fixture permissions the script emits a structured skip. diff --git a/isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py b/isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py new file mode 100755 index 000000000..fd9d1c609 --- /dev/null +++ b/isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""AWS SEC13-01 mutual TLS probe wrapper. + +Delegates to the shared ``mutual_tls_test`` probe. Operators supply +``EDGE_ENDPOINTS`` / ``EAST_WEST_ENDPOINTS`` plus ``MTLS_*_PATH`` cert +material to exercise concrete endpoints. When east-west endpoints are +omitted, AWS marks that plane provider-hidden (intra-VPC service mTLS is +customer-owned mesh/sidecar). When no north-south endpoints are configured +either, the shared probe emits a structured skip. + +Usage: + python mutual_tls_test.py --region us-west-2 + EDGE_ENDPOINTS=edge.example.com:443 MTLS_CA_CERT_PATH=... \\ + MTLS_CLIENT_CERT_PATH=... MTLS_CLIENT_KEY_PATH=... \\ + python mutual_tls_test.py --region us-west-2 +""" + +from __future__ import annotations + +import argparse +import importlib.util +import json +import os +import sys +from pathlib import Path +from typing import Any + +DEMO_MODE = os.environ.get("ISVCTL_DEMO_MODE") == "1" + +EAST_WEST_HIDDEN_MESSAGE = ( + "east_west_mtls_enforced: AWS EC2/EKS tenants do not receive a " + "customer-provable east-west mTLS surface in region {region}; " + "intra-VPC service encryption is customer-owned (mesh/sidecar). " + "Set EAST_WEST_ENDPOINTS to probe platform-specific east-west endpoints when available." +) + + +def _load_shared_probe() -> Any: + """Load providers/shared/mutual_tls_test.py as a module.""" + shared_path = Path(__file__).resolve().parents[3] / "shared" / "mutual_tls_test.py" + spec = importlib.util.spec_from_file_location("shared_mutual_tls_probe", shared_path) + if spec is None or spec.loader is None: + msg = f"Cannot load shared probe at {shared_path}" + raise RuntimeError(msg) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def main() -> int: + """Run the shared mTLS probe with AWS-specific east-west defaults.""" + parser = argparse.ArgumentParser(description="AWS mutual TLS (SEC13-01) probe") + parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-west-2")) + parser.add_argument( + "--north-south-endpoints", + default=os.environ.get("EDGE_ENDPOINTS", ""), + help="Comma-separated host:port north-south endpoints (default: EDGE_ENDPOINTS)", + ) + parser.add_argument( + "--east-west-endpoints", + default=os.environ.get("EAST_WEST_ENDPOINTS", ""), + help="Comma-separated host:port east-west endpoints (default: EAST_WEST_ENDPOINTS)", + ) + parser.add_argument("--ca-cert", default=os.environ.get("MTLS_CA_CERT_PATH", "")) + parser.add_argument("--client-cert", default=os.environ.get("MTLS_CLIENT_CERT_PATH", "")) + parser.add_argument("--client-key", default=os.environ.get("MTLS_CLIENT_KEY_PATH", "")) + parser.add_argument("--timeout", default="5.0") + args = parser.parse_args() + + probe = _load_shared_probe() + + if DEMO_MODE: + result = probe._demo_result() + print(json.dumps(result, indent=2)) + return 0 + + try: + north_south = probe._parse_endpoints(args.north_south_endpoints) + east_west = probe._parse_endpoints(args.east_west_endpoints) + timeout = probe._parse_timeout(args.timeout) + ca_cert = probe._require_file(args.ca_cert, "CA cert") if args.ca_cert else None + client_cert = probe._require_file(args.client_cert, "Client cert") if args.client_cert else None + client_key = probe._require_file(args.client_key, "Client key") if args.client_key else None + except ValueError as exc: + result = { + "success": False, + "platform": "security", + "test_name": "mutual_tls", + "error": str(exc), + "error_type": "bad_input", + "tests": {name: {"passed": False, "error": str(exc)} for name in probe.REQUIRED_TESTS}, + } + print(json.dumps(result, indent=2)) + return 1 + + # Prefer operator-supplied east-west endpoints; otherwise mark provider-hidden + # so a north-south-only run can still satisfy the two-plane contract. + east_west_hidden = None if east_west else EAST_WEST_HIDDEN_MESSAGE.format(region=args.region) + + result = probe.run_mutual_tls_probe( + north_south_endpoints=north_south, + east_west_endpoints=east_west, + ca_cert=ca_cert, + client_cert=client_cert, + client_key=client_key, + timeout=timeout, + east_west_provider_hidden_message=east_west_hidden, + ) + print(json.dumps(result, indent=2)) + if result.get("skipped") is True: + return 0 + return 0 if result.get("success") is True else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/providers/my-isv/config/security.yaml b/isvctl/configs/providers/my-isv/config/security.yaml index dfcc3cff7..b330aeced 100644 --- a/isvctl/configs/providers/my-isv/config/security.yaml +++ b/isvctl/configs/providers/my-isv/config/security.yaml @@ -91,6 +91,17 @@ commands: args: ["--region", "{{region}}"] timeout: 60 + - name: mutual_tls_test + phase: test + command: "python ../../shared/mutual_tls_test.py" + args: + - "--north-south-endpoints={{mutual_tls_north_south_endpoints}}" + - "--east-west-endpoints={{mutual_tls_east_west_endpoints}}" + - "--ca-cert={{mutual_tls_ca_cert}}" + - "--client-cert={{mutual_tls_client_cert}}" + - "--client-key={{mutual_tls_client_key}}" + timeout: 60 + - name: insecure_protocols_test phase: test command: "python ../../shared/insecure_protocols_test.py" @@ -208,6 +219,11 @@ tests: settings: region: "my-isv-region-1" insecure_protocols_endpoints: "{{env.EDGE_ENDPOINTS | default('', true)}}" + mutual_tls_north_south_endpoints: "{{env.EDGE_ENDPOINTS | default('', true)}}" + mutual_tls_east_west_endpoints: "{{env.EAST_WEST_ENDPOINTS | default('', true)}}" + mutual_tls_ca_cert: "{{env.MTLS_CA_CERT_PATH | default('', true)}}" + mutual_tls_client_cert: "{{env.MTLS_CLIENT_CERT_PATH | default('', true)}}" + mutual_tls_client_key: "{{env.MTLS_CLIENT_KEY_PATH | default('', true)}}" tenant_id: "{{env.MY_ISV_TENANT_ID | default('demo-tenant', true)}}" reservation_id: "demo-capacity-reservation" reservation_count: 2 diff --git a/isvctl/configs/providers/shared/mutual_tls_test.py b/isvctl/configs/providers/shared/mutual_tls_test.py new file mode 100755 index 000000000..34b7400a0 --- /dev/null +++ b/isvctl/configs/providers/shared/mutual_tls_test.py @@ -0,0 +1,434 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Mutual TLS (or equivalent) probe for north-south and east-west traffic. + +SEC13-01: for each configured endpoint, prove that an anonymous TLS client is +rejected and that a client presenting the configured certificate is accepted. + +Emits the contract:: + + { + "success": bool, + "platform": "security", + "test_name": "mutual_tls", + "endpoints_tested": int, + "tests": { + "north_south_mtls_enforced": {"passed": bool, "message": str, "probes": [...]}, + "east_west_mtls_enforced": {"passed": bool, "message": str, "probes": [...]} + } + } + +When no endpoints are configured, emits a structured ``skipped`` payload. +``ISVCTL_DEMO_MODE=1`` short-circuits with dummy-success output. + +Usage: + python mutual_tls_test.py \\ + --north-south-endpoints edge.example.com:443 \\ + --east-west-endpoints mesh.internal:8443 \\ + --ca-cert /path/ca.pem --client-cert /path/client.pem --client-key /path/client.key +""" + +from __future__ import annotations + +import argparse +import json +import os +import socket +import ssl +import sys +from pathlib import Path +from typing import Any + +DEMO_MODE = os.environ.get("ISVCTL_DEMO_MODE") == "1" + +REQUIRED_TESTS: list[str] = [ + "north_south_mtls_enforced", + "east_west_mtls_enforced", +] + + +def _parse_endpoints(raw: str) -> list[tuple[str, int]]: + """Parse a comma-separated ``host:port`` list into (host, port) tuples.""" + endpoints: list[tuple[str, int]] = [] + for part in (raw or "").split(","): + item = part.strip() + if not item: + continue + if ":" not in item: + msg = f"Endpoint '{item}' must be host:port" + raise ValueError(msg) + host, port_str = item.rsplit(":", 1) + host = host.strip() + if not host: + msg = f"Endpoint '{item}' is missing a host" + raise ValueError(msg) + try: + port = int(port_str) + except ValueError as exc: + msg = f"Endpoint '{item}' has a non-integer port" + raise ValueError(msg) from exc + if not 1 <= port <= 65535: + msg = f"Endpoint '{item}' port out of range" + raise ValueError(msg) + endpoints.append((host, port)) + return endpoints + + +def _parse_timeout(raw: str) -> float: + """Parse a positive timeout in seconds.""" + try: + timeout = float(raw) + except ValueError as exc: + msg = f"Invalid timeout '{raw}'" + raise ValueError(msg) from exc + if timeout <= 0: + msg = "Timeout must be positive" + raise ValueError(msg) + return timeout + + +def _require_file(path: str, label: str) -> Path: + """Return a Path that exists and is a file.""" + file_path = Path(path) + if not file_path.is_file(): + msg = f"{label} not found: {path}" + raise ValueError(msg) + return file_path + + +def _ssl_context( + *, + ca_cert: Path | None, + client_cert: Path | None = None, + client_key: Path | None = None, +) -> ssl.SSLContext: + """Build an SSL context that optionally presents a client certificate.""" + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.check_hostname = False + context.verify_mode = ssl.CERT_REQUIRED if ca_cert is not None else ssl.CERT_NONE + if ca_cert is not None: + context.load_verify_locations(cafile=str(ca_cert)) + if client_cert is not None and client_key is not None: + context.load_cert_chain(certfile=str(client_cert), keyfile=str(client_key)) + return context + + +def _handshake( + host: str, + port: int, + context: ssl.SSLContext, + timeout: float, +) -> dict[str, Any]: + """Attempt a TLS handshake and return accepted/rejected classification.""" + result: dict[str, Any] = {"host": host, "port": port} + try: + with socket.create_connection((host, port), timeout=timeout) as raw: + raw.settimeout(timeout) + with context.wrap_socket(raw, server_hostname=host) as tls: + result["accepted"] = True + result["tls_version"] = tls.version() + return result + except ssl.SSLError as exc: + result.update(accepted=False, detail=f"SSLError: {exc}") + return result + except (ConnectionResetError, ConnectionRefusedError, BrokenPipeError, TimeoutError, OSError) as exc: + result.update(accepted=False, detail=f"{type(exc).__name__}: {exc}") + return result + + +def probe_mtls_endpoint( + host: str, + port: int, + *, + ca_cert: Path, + client_cert: Path, + client_key: Path, + timeout: float, + plane: str, +) -> dict[str, Any]: + """Probe one endpoint: anonymous client rejected, authenticated client accepted.""" + anonymous = _handshake(host, port, _ssl_context(ca_cert=ca_cert), timeout) + authenticated = _handshake( + host, + port, + _ssl_context(ca_cert=ca_cert, client_cert=client_cert, client_key=client_key), + timeout, + ) + anonymous_rejected = anonymous.get("accepted") is not True + authenticated_accepted = authenticated.get("accepted") is True + passed = anonymous_rejected and authenticated_accepted + return { + "host": host, + "port": port, + "plane": plane, + "anonymous_rejected": anonymous_rejected, + "authenticated_accepted": authenticated_accepted, + "passed": passed, + "detail": { + "anonymous": anonymous.get("detail") or ("accepted" if anonymous.get("accepted") else "rejected"), + "authenticated": authenticated.get("detail") + or ("accepted" if authenticated.get("accepted") else "rejected"), + "tls_version": authenticated.get("tls_version"), + }, + } + + +def _aggregate_plane( + endpoints: list[tuple[str, int]], + *, + plane: str, + ca_cert: Path, + client_cert: Path, + client_key: Path, + timeout: float, +) -> dict[str, Any]: + """Probe all endpoints for one traffic plane and aggregate pass/fail.""" + probes = [ + probe_mtls_endpoint( + host, + port, + ca_cert=ca_cert, + client_cert=client_cert, + client_key=client_key, + timeout=timeout, + plane=plane, + ) + for host, port in endpoints + ] + passed = all(probe.get("passed") is True for probe in probes) if probes else False + failures = [f"{probe['host']}:{probe['port']}" for probe in probes if probe.get("passed") is not True] + message = ( + f"mTLS enforced on {len(probes)} {plane} endpoint(s)" + if passed + else f"mTLS not enforced on: {', '.join(failures)}" + ) + return {"passed": passed, "message": message, "probes": probes} + + +def _provider_hidden_plane(plane: str, message: str) -> dict[str, Any]: + """Return a passing provider-hidden result for a non-probeable plane.""" + return { + "passed": True, + "provider_hidden": True, + "message": message, + "probes": [{"plane": plane, "provider_hidden": True}], + } + + +def _demo_result() -> dict[str, Any]: + """Return the demo-mode mutual TLS probe contract.""" + return { + "success": True, + "platform": "security", + "test_name": "mutual_tls", + "endpoints_tested": 2, + "tests": { + "north_south_mtls_enforced": { + "passed": True, + "message": "Demo: north-south mTLS enforced", + "probes": [{"plane": "north_south", "passed": True}], + }, + "east_west_mtls_enforced": { + "passed": True, + "message": "Demo: east-west mTLS enforced", + "probes": [{"plane": "east_west", "passed": True}], + }, + }, + } + + +def run_mutual_tls_probe( + *, + north_south_endpoints: list[tuple[str, int]], + east_west_endpoints: list[tuple[str, int]], + ca_cert: Path | None, + client_cert: Path | None, + client_key: Path | None, + timeout: float, + east_west_provider_hidden_message: str | None = None, +) -> dict[str, Any]: + """Build the SEC13-01 JSON contract for the given endpoint/cert inputs.""" + if not north_south_endpoints and not east_west_endpoints and not east_west_provider_hidden_message: + return { + "success": True, + "platform": "security", + "test_name": "mutual_tls", + "skipped": True, + "skip_reason": ( + "No SEC13-01 endpoints configured " + "(set EDGE_ENDPOINTS and/or EAST_WEST_ENDPOINTS with MTLS_CA_CERT_PATH, " + "MTLS_CLIENT_CERT_PATH, MTLS_CLIENT_KEY_PATH)" + ), + } + + if (north_south_endpoints or east_west_endpoints) and not (ca_cert and client_cert and client_key): + error = ( + "mTLS probe requires --ca-cert, --client-cert, and --client-key " + "(or MTLS_CA_CERT_PATH / MTLS_CLIENT_CERT_PATH / MTLS_CLIENT_KEY_PATH)" + ) + return { + "success": False, + "platform": "security", + "test_name": "mutual_tls", + "error": error, + "error_type": "bad_input", + "tests": {name: {"passed": False, "error": error} for name in REQUIRED_TESTS}, + } + + tests: dict[str, Any] = {} + endpoints_tested = 0 + + if north_south_endpoints: + assert ca_cert and client_cert and client_key + tests["north_south_mtls_enforced"] = _aggregate_plane( + north_south_endpoints, + plane="north_south", + ca_cert=ca_cert, + client_cert=client_cert, + client_key=client_key, + timeout=timeout, + ) + endpoints_tested += len(north_south_endpoints) + else: + tests["north_south_mtls_enforced"] = _provider_hidden_plane( + "north_south", + "north_south_mtls_enforced: no north-south endpoints configured for this run", + ) + + if east_west_endpoints: + assert ca_cert and client_cert and client_key + tests["east_west_mtls_enforced"] = _aggregate_plane( + east_west_endpoints, + plane="east_west", + ca_cert=ca_cert, + client_cert=client_cert, + client_key=client_key, + timeout=timeout, + ) + endpoints_tested += len(east_west_endpoints) + elif east_west_provider_hidden_message: + tests["east_west_mtls_enforced"] = _provider_hidden_plane( + "east_west", + east_west_provider_hidden_message, + ) + else: + tests["east_west_mtls_enforced"] = _provider_hidden_plane( + "east_west", + "east_west_mtls_enforced: no east-west endpoints configured for this run", + ) + + # Pure provider-hidden with nothing probed is not evidence — skip instead. + if endpoints_tested < 1: + return { + "success": True, + "platform": "security", + "test_name": "mutual_tls", + "skipped": True, + "skip_reason": ( + "No SEC13-01 endpoints configured " + "(set EDGE_ENDPOINTS and/or EAST_WEST_ENDPOINTS with MTLS_CA_CERT_PATH, " + "MTLS_CLIENT_CERT_PATH, MTLS_CLIENT_KEY_PATH)" + ), + } + + success = all(tests[name].get("passed") is True for name in REQUIRED_TESTS) + return { + "success": success, + "platform": "security", + "test_name": "mutual_tls", + "endpoints_tested": endpoints_tested, + "tests": tests, + } + + +def main() -> int: + """Probe configured endpoints for mTLS enforcement.""" + parser = argparse.ArgumentParser(description="Mutual TLS (SEC13-01) probe") + parser.add_argument( + "--north-south-endpoints", + default=os.environ.get("EDGE_ENDPOINTS", ""), + help="Comma-separated host:port list for north-south / edge endpoints", + ) + parser.add_argument( + "--east-west-endpoints", + default=os.environ.get("EAST_WEST_ENDPOINTS", ""), + help="Comma-separated host:port list for east-west / mesh endpoints", + ) + parser.add_argument( + "--ca-cert", + default=os.environ.get("MTLS_CA_CERT_PATH", ""), + help="Path to CA certificate used to verify the server", + ) + parser.add_argument( + "--client-cert", + default=os.environ.get("MTLS_CLIENT_CERT_PATH", ""), + help="Path to client certificate presented for mTLS", + ) + parser.add_argument( + "--client-key", + default=os.environ.get("MTLS_CLIENT_KEY_PATH", ""), + help="Path to client private key presented for mTLS", + ) + parser.add_argument("--timeout", default="5.0", help="Per-probe socket timeout in seconds") + parser.add_argument( + "--east-west-provider-hidden-message", + default="", + help="When set and no east-west endpoints are given, mark that plane provider-hidden", + ) + args = parser.parse_args() + + if DEMO_MODE: + result = _demo_result() + print(json.dumps(result, indent=2)) + return 0 + + try: + north_south = _parse_endpoints(args.north_south_endpoints) + east_west = _parse_endpoints(args.east_west_endpoints) + timeout = _parse_timeout(args.timeout) + ca_cert = _require_file(args.ca_cert, "CA cert") if args.ca_cert else None + client_cert = _require_file(args.client_cert, "Client cert") if args.client_cert else None + client_key = _require_file(args.client_key, "Client key") if args.client_key else None + except ValueError as exc: + result = { + "success": False, + "platform": "security", + "test_name": "mutual_tls", + "error": str(exc), + "error_type": "bad_input", + "tests": {name: {"passed": False, "error": str(exc)} for name in REQUIRED_TESTS}, + } + print(json.dumps(result, indent=2)) + return 1 + + result = run_mutual_tls_probe( + north_south_endpoints=north_south, + east_west_endpoints=east_west, + ca_cert=ca_cert, + client_cert=client_cert, + client_key=client_key, + timeout=timeout, + east_west_provider_hidden_message=args.east_west_provider_hidden_message or None, + ) + print(json.dumps(result, indent=2)) + if result.get("skipped") is True: + return 0 + return 0 if result.get("success") is True else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/suites/README.md b/isvctl/configs/suites/README.md index aadec5d52..75ba523f7 100644 --- a/isvctl/configs/suites/README.md +++ b/isvctl/configs/suites/README.md @@ -195,6 +195,8 @@ Validations use `sinfo`/`srun` directly: partitions, GPU allocation, job schedul | `bmc_protocol_security` | test | `providers/my-isv/scripts/security/bmc_protocol_security_test.py` | CNP10-01: IPMI disabled; Redfish over TLS with AAA | | `bmc_bastion_access` | test | `providers/my-isv/scripts/security/bmc_bastion_access_test.py` | SEC12-03: BMC reachable only through a hardened bastion | | `api_endpoint_isolation` | test | `providers/my-isv/scripts/security/api_endpoint_test.py` | API endpoints not publicly accessible | +| `mutual_tls_test` | test | `providers/shared/mutual_tls_test.py` | SEC13-01: mTLS (or equivalent) for north-south and east-west traffic | +| `insecure_protocols_test` | test | `providers/shared/insecure_protocols_test.py` | SEC13-02: insecure protocols (HTTP, SSLv3, TLSv1) disabled | | `mfa_enforcement` | test | `providers/my-isv/scripts/security/mfa_enforcement_test.py` | Administrative UI, CLI, and API access require MFA | | `cert_rotation_test` | test | `providers/my-isv/scripts/security/cert_rotation_test.py` | SEC09-01: TLS certificate rotation cycle or auto-renewal | | `kms_encryption_options_test` | test | `providers/my-isv/scripts/security/kms_encryption_options_test.py` | SEC09-02: Provider-managed and customer-managed KMS options | diff --git a/isvctl/configs/suites/security.yaml b/isvctl/configs/suites/security.yaml index b6e8c241c..e08fe66ad 100644 --- a/isvctl/configs/suites/security.yaml +++ b/isvctl/configs/suites/security.yaml @@ -85,6 +85,13 @@ tests: labels: ["min_req", "network", "security"] step: api_endpoint_isolation + mutual_tls: + checks: + MutualTlsCheck: + test_id: "SEC13-01" + labels: ["min_req", "network", "security"] + step: mutual_tls_test + insecure_protocols: checks: InsecureProtocolsCheck: diff --git a/isvtest/src/isvtest/validations/__init__.py b/isvtest/src/isvtest/validations/__init__.py index f3883f62f..81324934f 100644 --- a/isvtest/src/isvtest/validations/__init__.py +++ b/isvtest/src/isvtest/validations/__init__.py @@ -141,6 +141,7 @@ LeastPrivilegePolicyCheck, MfaEnforcedCheck, MinimalRoleEnforcementCheck, + MutualTlsCheck, OidcUserAuthCheck, ShortLivedCredentialsCheck, TenantIsolationCheck, @@ -201,6 +202,7 @@ "MemorySanitizationCheck", "MfaEnforcedCheck", "MinimalRoleEnforcementCheck", + "MutualTlsCheck", "NetworkConnectivityCheck", "NetworkProvisionedCheck", "NimHealthCheck", diff --git a/isvtest/src/isvtest/validations/security.py b/isvtest/src/isvtest/validations/security.py index e7fe993a9..2a23e5da3 100644 --- a/isvtest/src/isvtest/validations/security.py +++ b/isvtest/src/isvtest/validations/security.py @@ -230,6 +230,57 @@ def run(self) -> None: self.set_passed(f"Insecure protocols disabled ({endpoints_tested} endpoints tested)") +class MutualTlsCheck(BaseValidation): + """Validate mTLS (or equivalent) for north-south and east-west traffic (SEC13-01). + + Verifies that configured endpoints reject anonymous TLS clients and accept + authenticated client certificates on both traffic planes. Providers may mark + a plane ``provider_hidden`` when it is not customer-probeable (for example + AWS east-west mesh). + + Config: + step_output: The mutual_tls_test step output to check + + Step output: + endpoints_tested: Positive integer of endpoints actually probed + tests: dict with north_south_mtls_enforced, east_west_mtls_enforced + """ + + description: ClassVar[str] = "Check mTLS (or equivalent) for north-south and east-west traffic" + + def run(self) -> None: + """Validate required mTLS plane probe results from step output.""" + step_output = self.config.get("step_output", {}) + if step_output.get("skipped") is True: + pytest.skip(step_output.get("skip_reason") or "mTLS validation skipped (not configured)") + + if step_output.get("success") is False: + step_error = step_output.get("error") or step_output.get("message") + self.set_failed(step_error or "mTLS step failed: no error message provided") + return + + required = [ + "north_south_mtls_enforced", + "east_west_mtls_enforced", + ] + if not check_required_tests(self, required, "mTLS enforcement tests failed"): + return + + endpoints_tested = step_output.get("endpoints_tested") + if type(endpoints_tested) is not int or endpoints_tested < 1: + self.set_failed("mTLS output missing positive int 'endpoints_tested'") + return + + tests = step_output.get("tests", {}) + hidden = [ + name + for name in required + if isinstance(tests.get(name), dict) and tests[name].get("provider_hidden") is True + ] + hidden_note = f", provider_hidden={','.join(hidden)}" if hidden else "" + self.set_passed(f"mTLS enforced ({endpoints_tested} endpoints tested{hidden_note})") + + class BmcBastionAccessCheck(BaseValidation): """Validate BMC is only accessible via a hardened bastion. diff --git a/isvtest/tests/test_security.py b/isvtest/tests/test_security.py index a21fd400c..f8ad65203 100644 --- a/isvtest/tests/test_security.py +++ b/isvtest/tests/test_security.py @@ -32,6 +32,7 @@ InsecureProtocolsCheck, KmsEncryptionOptionCheck, MfaEnforcedCheck, + MutualTlsCheck, OidcUserAuthCheck, ShortLivedCredentialsCheck, TenantIsolationCheck, @@ -1046,3 +1047,91 @@ def test_missing_tests_fails(self) -> None: assert result["passed"] is False assert "tests" in result["error"].lower() + + +REQUIRED_MUTUAL_TLS_TESTS = [ + "north_south_mtls_enforced", + "east_west_mtls_enforced", +] + + +def _mutual_tls_config( + tests: dict[str, dict[str, Any]] | None = None, + *, + endpoints_tested: int = 2, +) -> dict[str, Any]: + """Build a MutualTlsCheck validation config.""" + default_tests = {name: {"passed": True, "message": f"{name} confirmed"} for name in REQUIRED_MUTUAL_TLS_TESTS} + return { + "step_output": { + "success": True, + "platform": "security", + "test_name": "mutual_tls", + "endpoints_tested": endpoints_tested, + "tests": tests if tests is not None else default_tests, + }, + } + + +class TestMutualTlsCheck: + """Tests for SEC13-01 mutual TLS validation.""" + + def test_all_required_tests_pass(self) -> None: + """Pass when both traffic-plane probes report mTLS enforced.""" + result = MutualTlsCheck(config=_mutual_tls_config()).execute() + + assert result["passed"] is True + assert "mTLS enforced (2 endpoints tested)" in result["output"] + + def test_failed_step_without_error_fails_with_default_message(self) -> None: + """Fail when the step reports success=False without an error message.""" + config = _mutual_tls_config() + config["step_output"]["success"] = False + result = MutualTlsCheck(config=config).execute() + + assert result["passed"] is False + assert "mTLS step failed: no error message provided" in result["error"] + + def test_failed_probe_reports_contract_key(self) -> None: + """Fail when one plane reports passed=False.""" + tests = {name: {"passed": True} for name in REQUIRED_MUTUAL_TLS_TESTS} + tests["north_south_mtls_enforced"] = { + "passed": False, + "error": "anonymous client accepted on edge.example.com:443", + } + result = MutualTlsCheck(config=_mutual_tls_config(tests)).execute() + + assert result["passed"] is False + assert "north_south_mtls_enforced" in result["error"] + + def test_provider_hidden_east_west_noted_in_pass_message(self) -> None: + """Pass notes provider-hidden planes when present.""" + tests = { + "north_south_mtls_enforced": {"passed": True}, + "east_west_mtls_enforced": { + "passed": True, + "provider_hidden": True, + "message": "no tenant east-west mTLS surface", + }, + } + result = MutualTlsCheck(config=_mutual_tls_config(tests, endpoints_tested=1)).execute() + + assert result["passed"] is True + assert "provider_hidden=east_west_mtls_enforced" in result["output"] + + def test_zero_endpoints_tested_fails(self) -> None: + """Fail when endpoints_tested is non-positive.""" + result = MutualTlsCheck(config=_mutual_tls_config(endpoints_tested=0)).execute() + + assert result["passed"] is False + assert "endpoints_tested" in result["error"] + + def test_skips_when_step_marks_skipped(self) -> None: + """pytest.skip when the step emits a structured skip.""" + step_output = { + "success": True, + "skipped": True, + "skip_reason": "No SEC13-01 endpoints configured", + } + with pytest.raises(pytest.skip.Exception, match="No SEC13-01 endpoints configured"): + MutualTlsCheck(config={"step_output": step_output}).execute() diff --git a/scripts/tests/test_mutual_tls_probe.py b/scripts/tests/test_mutual_tls_probe.py new file mode 100644 index 000000000..45434b540 --- /dev/null +++ b/scripts/tests/test_mutual_tls_probe.py @@ -0,0 +1,169 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for providers/shared/mutual_tls_test.py.""" + +from __future__ import annotations + +import importlib.util +import json +import ssl +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +_SCRIPT_PATH = ( + Path(__file__).resolve().parents[2] / "isvctl" / "configs" / "providers" / "shared" / "mutual_tls_test.py" +) +_spec = importlib.util.spec_from_file_location("mutual_tls_test", _SCRIPT_PATH) +assert _spec and _spec.loader +probe = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(probe) + + +def test_parse_endpoints_accepts_host_port_list() -> None: + """Comma-separated host:port strings parse into tuples.""" + assert probe._parse_endpoints("a.example:443,b.example:8443") == [ + ("a.example", 443), + ("b.example", 8443), + ] + + +def test_parse_endpoints_rejects_missing_port() -> None: + """Endpoints without a port raise ValueError.""" + try: + probe._parse_endpoints("edge.example") + except ValueError as exc: + assert "host:port" in str(exc) + else: + raise AssertionError("expected ValueError") + + +def test_demo_result_emits_both_planes() -> None: + """Demo mode contract includes both required planes.""" + result = probe._demo_result() + assert result["success"] is True + assert result["endpoints_tested"] == 2 + assert result["tests"]["north_south_mtls_enforced"]["passed"] is True + assert result["tests"]["east_west_mtls_enforced"]["passed"] is True + + +def test_run_skips_when_no_endpoints() -> None: + """Empty endpoint lists produce a structured skip.""" + result = probe.run_mutual_tls_probe( + north_south_endpoints=[], + east_west_endpoints=[], + ca_cert=None, + client_cert=None, + client_key=None, + timeout=1.0, + ) + assert result["skipped"] is True + assert "No SEC13-01 endpoints configured" in result["skip_reason"] + + +def test_run_fails_when_endpoints_without_certs() -> None: + """Endpoints without cert paths fail with bad_input.""" + result = probe.run_mutual_tls_probe( + north_south_endpoints=[("edge.example", 443)], + east_west_endpoints=[], + ca_cert=None, + client_cert=None, + client_key=None, + timeout=1.0, + east_west_provider_hidden_message="hidden", + ) + assert result["success"] is False + assert result["error_type"] == "bad_input" + + +def test_run_with_provider_hidden_east_west_and_probed_north_south(tmp_path: Path) -> None: + """North-south probe + east-west provider-hidden satisfies the contract.""" + ca = tmp_path / "ca.pem" + cert = tmp_path / "client.pem" + key = tmp_path / "client.key" + for path in (ca, cert, key): + path.write_text("placeholder\n", encoding="utf-8") + + def fake_probe( + host: str, + port: int, + *, + ca_cert: Path, + client_cert: Path, + client_key: Path, + timeout: float, + plane: str, + ) -> dict[str, Any]: + return { + "host": host, + "port": port, + "plane": plane, + "anonymous_rejected": True, + "authenticated_accepted": True, + "passed": True, + "detail": {}, + } + + with patch.object(probe, "probe_mtls_endpoint", side_effect=fake_probe): + result = probe.run_mutual_tls_probe( + north_south_endpoints=[("edge.example", 443)], + east_west_endpoints=[], + ca_cert=ca, + client_cert=cert, + client_key=key, + timeout=1.0, + east_west_provider_hidden_message="AWS east-west is provider-hidden", + ) + + assert result["success"] is True + assert result["endpoints_tested"] == 1 + assert result["tests"]["north_south_mtls_enforced"]["passed"] is True + assert result["tests"]["east_west_mtls_enforced"]["provider_hidden"] is True + + +def test_handshake_marks_ssl_error_as_rejected() -> None: + """SSL errors during wrap_socket classify as not accepted.""" + context = MagicMock() + raw = MagicMock() + raw.__enter__ = MagicMock(return_value=raw) + raw.__exit__ = MagicMock(return_value=None) + raw.settimeout = MagicMock() + + def raise_ssl(*_args: object, **_kwargs: object) -> None: + raise ssl.SSLError("certificate required") + + context.wrap_socket.side_effect = raise_ssl + + with patch.object(probe.socket, "create_connection", return_value=raw): + result = probe._handshake("edge.example", 443, context, 1.0) + + assert result["accepted"] is False + assert "SSLError" in result["detail"] + + +def test_main_demo_mode(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: + """CLI demo mode prints the demo contract and exits 0.""" + monkeypatch.setattr(probe, "DEMO_MODE", True) + monkeypatch.setattr( + "sys.argv", + ["mutual_tls_test.py"], + ) + assert probe.main() == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["test_name"] == "mutual_tls" + assert payload["success"] is True From 4bc02db3c34c0d0da984d0e3c46a0e91cdf9b1d9 Mon Sep 17 00:00:00 2001 From: Alexandre Begnoche Date: Mon, 13 Jul 2026 15:31:53 -0400 Subject: [PATCH 2/4] fix(security): avoid SEC11 cleanup failure on pending EC2 terminate boto3's InstanceTerminated waiter treats pending as terminal failure, so terminating a still-booting tenant fixture failed the step after probes passed. Wait for running on launch and poll terminate without that race. Signed-off-by: Alexandre Begnoche --- .../scripts/security/tenant_isolation_test.py | 54 ++++++++++++--- .../aws/test_aws_security_scripts.py | 67 +++++++++++++++++++ 2 files changed, 111 insertions(+), 10 deletions(-) diff --git a/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py index c7d245b9c..c3a229f0b 100644 --- a/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py +++ b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py @@ -261,6 +261,46 @@ def _launch_instance(ec2: Any, tenant: Tenant, ami_id: str) -> None: ) tenant.instance_id = response["Instances"][0]["InstanceId"] tenant.created["instance"] = True + # Wait until running before returning. Teardown may terminate immediately + # after probes; boto3's InstanceTerminated waiter treats ``pending`` as a + # terminal failure, so leaving the instance booting races cleanup. + ec2.get_waiter("instance_running").wait( + InstanceIds=[tenant.instance_id], + WaiterConfig={"Delay": 5, "MaxAttempts": 60}, + ) + + +def _wait_instance_terminated( + ec2: Any, + instance_id: str, + *, + delay: float = 5.0, + max_attempts: int = 60, +) -> None: + """Poll until ``instance_id`` is terminated (or already gone). + + Unlike boto3's ``instance_terminated`` waiter, treat ``pending`` as + in-progress. Terminating a still-booting instance briefly stays + ``pending`` before ``shutting-down``; the stock waiter fails that race + immediately and leaves ENIs holding the SG/subnet/VPC. + """ + for _ in range(max_attempts): + try: + response = ec2.describe_instances(InstanceIds=[instance_id]) + except ClientError as e: + if e.response.get("Error", {}).get("Code") == "InvalidInstanceID.NotFound": + return + raise + states = [ + instance.get("State", {}).get("Name") + for reservation in response.get("Reservations", []) + for instance in reservation.get("Instances", []) + ] + if not states or all(state == "terminated" for state in states): + return + time.sleep(delay) + msg = f"instance {instance_id} did not reach terminated within {delay * max_attempts:.0f}s" + raise TimeoutError(msg) def _create_volume(ec2: Any, tenant: Tenant) -> None: @@ -377,22 +417,16 @@ def _teardown_tenant( errors: list[str] = [] if tenant.created.get("instance") and tenant.instance_id: - # NB: catch WaiterError too -- the InstanceTerminated waiter treats - # "pending" as a terminal failure, which fires when setup raised - # before the instance reached running. Letting it propagate would - # mask the original error (and skip the rest of cleanup); the - # safety-net teardown step picks up any leftover instance. + # Catch wait failures too so a stuck terminate does not skip the rest + # of cleanup; the safety-net teardown step picks up leftovers. try: ec2.terminate_instances(InstanceIds=[tenant.instance_id]) except ClientError as e: errors.append(f"terminate instance {tenant.instance_id}: {e}") else: try: - ec2.get_waiter("instance_terminated").wait( - InstanceIds=[tenant.instance_id], - WaiterConfig={"Delay": 5, "MaxAttempts": 60}, - ) - except (ClientError, WaiterError) as e: + _wait_instance_terminated(ec2, tenant.instance_id) + except (ClientError, TimeoutError) as e: errors.append(f"wait terminated {tenant.instance_id}: {e}") if tenant.created.get("volume") and tenant.volume_id: diff --git a/isvctl/tests/providers/aws/test_aws_security_scripts.py b/isvctl/tests/providers/aws/test_aws_security_scripts.py index 6ef0f7712..274281606 100644 --- a/isvctl/tests/providers/aws/test_aws_security_scripts.py +++ b/isvctl/tests/providers/aws/test_aws_security_scripts.py @@ -3641,6 +3641,73 @@ def fake_teardown_tenant(*, tenant: Any, **_kwargs: Any) -> list[str]: assert cleanup_calls == [("isv-sec11-test-aaaa1111", {"vpc": True}, "vpc-partial")] +def test_wait_instance_terminated_tolerates_pending( + monkeypatch: pytest.MonkeyPatch, + tenant_isolation_module: ModuleType, +) -> None: + """Terminating a still-booting instance must not fail on ``pending``. + + boto3's InstanceTerminated waiter treats pending as terminal failure; + our poller must keep waiting until the instance reaches terminated. + """ + states = iter(["pending", "pending", "shutting-down", "terminated"]) + calls: list[str] = [] + + class FakeEc2: + def describe_instances(self, InstanceIds: list[str]) -> dict[str, Any]: + calls.append(InstanceIds[0]) + return {"Reservations": [{"Instances": [{"State": {"Name": next(states)}}]}]} + + monkeypatch.setattr(tenant_isolation_module.time, "sleep", lambda _seconds: None) + tenant_isolation_module._wait_instance_terminated(FakeEc2(), "i-pending123", delay=0.0, max_attempts=10) + assert calls == ["i-pending123", "i-pending123", "i-pending123", "i-pending123"] + + +def test_teardown_tenant_terminates_pending_instance_then_deletes_network( + monkeypatch: pytest.MonkeyPatch, + tenant_isolation_module: ModuleType, +) -> None: + """Cleanup must succeed when terminate is issued while the instance is pending.""" + states = iter(["pending", "shutting-down", "terminated"]) + deleted: list[tuple[str, str]] = [] + + class FakeEc2: + def terminate_instances(self, InstanceIds: list[str]) -> None: + deleted.append(("terminate", InstanceIds[0])) + + def describe_instances(self, InstanceIds: list[str]) -> dict[str, Any]: + return {"Reservations": [{"Instances": [{"State": {"Name": next(states)}}]}]} + + def delete_security_group(self, GroupId: str) -> None: + deleted.append(("sg", GroupId)) + + def delete_subnet(self, SubnetId: str) -> None: + deleted.append(("subnet", SubnetId)) + + def delete_vpc(self, VpcId: str) -> None: + deleted.append(("vpc", VpcId)) + + monkeypatch.setattr(tenant_isolation_module.time, "sleep", lambda _seconds: None) + tenant = _make_tenant(tenant_isolation_module, "cccc3333", "10.96.0.0/24") + tenant.created = {"instance": True, "sg": True, "subnet": True, "vpc": True} + + errors = tenant_isolation_module._teardown_tenant( + ec2=FakeEc2(), + iam=object(), + kms=object(), + s3=object(), + tenant=tenant, + ) + + assert errors == [] + assert deleted == [ + ("terminate", "i-cccc3333"), + ("sg", "sg-cccc3333"), + ("subnet", "subnet-cccc3333"), + ("vpc", "vpc-cccc3333"), + ] + + # --- teardown SEC11 sweep helpers ---------------------------------------- From 0f8b1bea0a24ed5fe001c917d4861c1a21d91da1 Mon Sep 17 00:00:00 2001 From: Alexandre Begnoche Date: Mon, 13 Jul 2026 16:05:07 -0400 Subject: [PATCH 3/4] refactor(security): simplify SEC13-01 probe and drop redundant AWS wrapper Wire aws/config/security.yaml straight to the shared mutual TLS probe via --east-west-provider-hidden-message instead of a 131-line wrapper that re-implemented the shared main() and reached into its private helpers. In the shared probe, collapse the duplicate skip payload into a single guard, extract _bad_input_result, loop over the two traffic planes instead of copy-pasting them, build the SSL contexts once per run instead of two per endpoint, and reduce the handshake except tuple to OSError. Also drop the instance_running launch waiter from the SEC11 tenant fixture: the pending-tolerant terminate poller already covers the race it guarded, and the waiter cost up to a minute per tenant of setup wall-clock on every run. Co-Authored-By: Claude Fable 5 Signed-off-by: Alexandre Begnoche --- .../providers/aws/config/security.yaml | 32 ++-- .../aws/scripts/security/mutual_tls_test.py | 131 ---------------- .../scripts/security/tenant_isolation_test.py | 14 +- .../providers/shared/mutual_tls_test.py | 146 +++++++----------- scripts/tests/test_mutual_tls_probe.py | 16 +- 5 files changed, 85 insertions(+), 254 deletions(-) delete mode 100755 isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py diff --git a/isvctl/configs/providers/aws/config/security.yaml b/isvctl/configs/providers/aws/config/security.yaml index c14e6d066..901fd5057 100644 --- a/isvctl/configs/providers/aws/config/security.yaml +++ b/isvctl/configs/providers/aws/config/security.yaml @@ -32,20 +32,20 @@ # 4. BMC Bastion Access - Verify BMC reachable only via hardened bastion # 5. API Endpoint Isolation - Verify management APIs are not publicly exposed # 6. Mutual TLS - Verify mTLS north-south/east-west (SEC13-01) -# 7. Insecure Protocols - Verify SSLv3/TLS1.0/TLS1.1/plain HTTP disabled (SEC13-02) +# 7. Insecure Protocols - Verify SSLv3/TLS1.0/TLS1.1/plain HTTP disabled (SEC13-02) # 8. MFA Enforcement - Verify admin interfaces (UI, CLI, API) require MFA # 9. Cert Rotation Cycle - Verify cert rotation cycle / auto-renewal (SEC09-01) # 10. KMS Options - Verify provider- and customer-managed keys (SEC09-02) # 11. Centralized KMS - Verify encrypted resources resolve to KMS (SEC09-03) # 12. Customer Managed Key - Verify customer-managed key encryption (SEC09-04) # 13. Least Privilege - Verify scoped user/resource/network policies (SEC04-01/02) -# 13. Audit Logging - Verify management-event logging and retention (SEC08-01/02) -# 14. SA Credential Auth - Verify IAM user + long-lived access key auth -# 15. OIDC User Auth - Probe configured OIDC-protected platform endpoint (SEC01-01) -# 16. Short-Lived Credentials - Verify STS issues node + workload creds with bounded TTL (SEC02-01) -# 17. Tenant Isolation - Verify hard tenant isolation across network, data, compute and storage (SEC11-01) -# 18. Capacity Grouping - Verify reservations are grouped and tenant-pinned (CAP04-01) -# 19. Topology Block - Verify topology block atomic allocation boundaries (CAP04-02) +# 14. Audit Logging - Verify management-event logging and retention (SEC08-01/02) +# 15. SA Credential Auth - Verify IAM user + long-lived access key auth +# 16. OIDC User Auth - Probe configured OIDC-protected platform endpoint (SEC01-01) +# 17. Short-Lived Credentials - Verify STS issues node + workload creds with bounded TTL (SEC02-01) +# 18. Tenant Isolation - Verify hard tenant isolation across network, data, compute and storage (SEC11-01) +# 19. Capacity Grouping - Verify reservations are grouped and tenant-pinned (CAP04-01) +# 20. Topology Block - Verify topology block atomic allocation boundaries (CAP04-02) # # Note: SEC12-* (BMC) tests cannot be fully validated on AWS because the # Nitro/hypervisor BMC plane is provider-hidden. The references inspect the @@ -108,12 +108,20 @@ commands: # Test 6a: Mutual TLS (SEC13-01) # Probes EDGE_ENDPOINTS / EAST_WEST_ENDPOINTS with MTLS_* cert material. - # Without endpoints the script emits a structured skip. East-west is - # provider-hidden on AWS unless EAST_WEST_ENDPOINTS is set. + # Without endpoints the shared script emits a structured skip. East-west + # is provider-hidden on AWS unless EAST_WEST_ENDPOINTS is set (intra-VPC + # service mTLS is customer-owned mesh/sidecar), expressed via the + # provider-hidden message below. - name: mutual_tls_test phase: test - command: "python3 ../scripts/security/mutual_tls_test.py" - args: ["--region", "{{region}}"] + command: "python3 ../../shared/mutual_tls_test.py" + args: + - >- + --east-west-provider-hidden-message=east_west_mtls_enforced: + AWS EC2/EKS tenants do not receive a customer-provable east-west + mTLS surface in region {{region}}; intra-VPC service encryption is + customer-owned (mesh/sidecar). Set EAST_WEST_ENDPOINTS to probe + platform-specific east-west endpoints when available. timeout: 120 # Test 6b: Insecure Protocols (SEC13-02) diff --git a/isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py b/isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py deleted file mode 100755 index fd9d1c609..000000000 --- a/isvctl/configs/providers/aws/scripts/security/mutual_tls_test.py +++ /dev/null @@ -1,131 +0,0 @@ -#!/usr/bin/env python3 -# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# SPDX-License-Identifier: Apache-2.0 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""AWS SEC13-01 mutual TLS probe wrapper. - -Delegates to the shared ``mutual_tls_test`` probe. Operators supply -``EDGE_ENDPOINTS`` / ``EAST_WEST_ENDPOINTS`` plus ``MTLS_*_PATH`` cert -material to exercise concrete endpoints. When east-west endpoints are -omitted, AWS marks that plane provider-hidden (intra-VPC service mTLS is -customer-owned mesh/sidecar). When no north-south endpoints are configured -either, the shared probe emits a structured skip. - -Usage: - python mutual_tls_test.py --region us-west-2 - EDGE_ENDPOINTS=edge.example.com:443 MTLS_CA_CERT_PATH=... \\ - MTLS_CLIENT_CERT_PATH=... MTLS_CLIENT_KEY_PATH=... \\ - python mutual_tls_test.py --region us-west-2 -""" - -from __future__ import annotations - -import argparse -import importlib.util -import json -import os -import sys -from pathlib import Path -from typing import Any - -DEMO_MODE = os.environ.get("ISVCTL_DEMO_MODE") == "1" - -EAST_WEST_HIDDEN_MESSAGE = ( - "east_west_mtls_enforced: AWS EC2/EKS tenants do not receive a " - "customer-provable east-west mTLS surface in region {region}; " - "intra-VPC service encryption is customer-owned (mesh/sidecar). " - "Set EAST_WEST_ENDPOINTS to probe platform-specific east-west endpoints when available." -) - - -def _load_shared_probe() -> Any: - """Load providers/shared/mutual_tls_test.py as a module.""" - shared_path = Path(__file__).resolve().parents[3] / "shared" / "mutual_tls_test.py" - spec = importlib.util.spec_from_file_location("shared_mutual_tls_probe", shared_path) - if spec is None or spec.loader is None: - msg = f"Cannot load shared probe at {shared_path}" - raise RuntimeError(msg) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module - - -def main() -> int: - """Run the shared mTLS probe with AWS-specific east-west defaults.""" - parser = argparse.ArgumentParser(description="AWS mutual TLS (SEC13-01) probe") - parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-west-2")) - parser.add_argument( - "--north-south-endpoints", - default=os.environ.get("EDGE_ENDPOINTS", ""), - help="Comma-separated host:port north-south endpoints (default: EDGE_ENDPOINTS)", - ) - parser.add_argument( - "--east-west-endpoints", - default=os.environ.get("EAST_WEST_ENDPOINTS", ""), - help="Comma-separated host:port east-west endpoints (default: EAST_WEST_ENDPOINTS)", - ) - parser.add_argument("--ca-cert", default=os.environ.get("MTLS_CA_CERT_PATH", "")) - parser.add_argument("--client-cert", default=os.environ.get("MTLS_CLIENT_CERT_PATH", "")) - parser.add_argument("--client-key", default=os.environ.get("MTLS_CLIENT_KEY_PATH", "")) - parser.add_argument("--timeout", default="5.0") - args = parser.parse_args() - - probe = _load_shared_probe() - - if DEMO_MODE: - result = probe._demo_result() - print(json.dumps(result, indent=2)) - return 0 - - try: - north_south = probe._parse_endpoints(args.north_south_endpoints) - east_west = probe._parse_endpoints(args.east_west_endpoints) - timeout = probe._parse_timeout(args.timeout) - ca_cert = probe._require_file(args.ca_cert, "CA cert") if args.ca_cert else None - client_cert = probe._require_file(args.client_cert, "Client cert") if args.client_cert else None - client_key = probe._require_file(args.client_key, "Client key") if args.client_key else None - except ValueError as exc: - result = { - "success": False, - "platform": "security", - "test_name": "mutual_tls", - "error": str(exc), - "error_type": "bad_input", - "tests": {name: {"passed": False, "error": str(exc)} for name in probe.REQUIRED_TESTS}, - } - print(json.dumps(result, indent=2)) - return 1 - - # Prefer operator-supplied east-west endpoints; otherwise mark provider-hidden - # so a north-south-only run can still satisfy the two-plane contract. - east_west_hidden = None if east_west else EAST_WEST_HIDDEN_MESSAGE.format(region=args.region) - - result = probe.run_mutual_tls_probe( - north_south_endpoints=north_south, - east_west_endpoints=east_west, - ca_cert=ca_cert, - client_cert=client_cert, - client_key=client_key, - timeout=timeout, - east_west_provider_hidden_message=east_west_hidden, - ) - print(json.dumps(result, indent=2)) - if result.get("skipped") is True: - return 0 - return 0 if result.get("success") is True else 1 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py index c3a229f0b..983bfd664 100644 --- a/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py +++ b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py @@ -261,13 +261,6 @@ def _launch_instance(ec2: Any, tenant: Tenant, ami_id: str) -> None: ) tenant.instance_id = response["Instances"][0]["InstanceId"] tenant.created["instance"] = True - # Wait until running before returning. Teardown may terminate immediately - # after probes; boto3's InstanceTerminated waiter treats ``pending`` as a - # terminal failure, so leaving the instance booting races cleanup. - ec2.get_waiter("instance_running").wait( - InstanceIds=[tenant.instance_id], - WaiterConfig={"Delay": 5, "MaxAttempts": 60}, - ) def _wait_instance_terminated( @@ -296,7 +289,8 @@ def _wait_instance_terminated( for reservation in response.get("Reservations", []) for instance in reservation.get("Instances", []) ] - if not states or all(state == "terminated" for state in states): + # all() on an empty list is True: no reservations means already gone. + if all(state == "terminated" for state in states): return time.sleep(delay) msg = f"instance {instance_id} did not reach terminated within {delay * max_attempts:.0f}s" @@ -897,8 +891,8 @@ def main() -> int: result["success"] = all(t.get("passed") for t in result["tests"].values()) except (ClientError, WaiterError, BotoCoreError) as exc: # Capture the FIRST error before cleanup runs, otherwise a downstream - # WaiterError from terminating a still-pending instance would mask - # the real cause (IAM limit, VPC limit, ResourceLimitExceeded, ...). + # cleanup error would mask the real cause (IAM limit, VPC limit, + # ResourceLimitExceeded, ...). error_type, error_msg = classify_aws_error(exc) result["error"] = f"[{error_type}] {error_msg}" result["success"] = False diff --git a/isvctl/configs/providers/shared/mutual_tls_test.py b/isvctl/configs/providers/shared/mutual_tls_test.py index 34b7400a0..e213ad5d6 100755 --- a/isvctl/configs/providers/shared/mutual_tls_test.py +++ b/isvctl/configs/providers/shared/mutual_tls_test.py @@ -134,7 +134,7 @@ def _handshake( timeout: float, ) -> dict[str, Any]: """Attempt a TLS handshake and return accepted/rejected classification.""" - result: dict[str, Any] = {"host": host, "port": port} + result: dict[str, Any] = {} try: with socket.create_connection((host, port), timeout=timeout) as raw: raw.settimeout(timeout) @@ -145,7 +145,7 @@ def _handshake( except ssl.SSLError as exc: result.update(accepted=False, detail=f"SSLError: {exc}") return result - except (ConnectionResetError, ConnectionRefusedError, BrokenPipeError, TimeoutError, OSError) as exc: + except OSError as exc: result.update(accepted=False, detail=f"{type(exc).__name__}: {exc}") return result @@ -154,20 +154,14 @@ def probe_mtls_endpoint( host: str, port: int, *, - ca_cert: Path, - client_cert: Path, - client_key: Path, + anonymous_context: ssl.SSLContext, + authenticated_context: ssl.SSLContext, timeout: float, plane: str, ) -> dict[str, Any]: """Probe one endpoint: anonymous client rejected, authenticated client accepted.""" - anonymous = _handshake(host, port, _ssl_context(ca_cert=ca_cert), timeout) - authenticated = _handshake( - host, - port, - _ssl_context(ca_cert=ca_cert, client_cert=client_cert, client_key=client_key), - timeout, - ) + anonymous = _handshake(host, port, anonymous_context, timeout) + authenticated = _handshake(host, port, authenticated_context, timeout) anonymous_rejected = anonymous.get("accepted") is not True authenticated_accepted = authenticated.get("accepted") is True passed = anonymous_rejected and authenticated_accepted @@ -191,9 +185,8 @@ def _aggregate_plane( endpoints: list[tuple[str, int]], *, plane: str, - ca_cert: Path, - client_cert: Path, - client_key: Path, + anonymous_context: ssl.SSLContext, + authenticated_context: ssl.SSLContext, timeout: float, ) -> dict[str, Any]: """Probe all endpoints for one traffic plane and aggregate pass/fail.""" @@ -201,9 +194,8 @@ def _aggregate_plane( probe_mtls_endpoint( host, port, - ca_cert=ca_cert, - client_cert=client_cert, - client_key=client_key, + anonymous_context=anonymous_context, + authenticated_context=authenticated_context, timeout=timeout, plane=plane, ) @@ -229,6 +221,18 @@ def _provider_hidden_plane(plane: str, message: str) -> dict[str, Any]: } +def _bad_input_result(error: str) -> dict[str, Any]: + """Return the failing bad_input contract with the error on every test.""" + return { + "success": False, + "platform": "security", + "test_name": "mutual_tls", + "error": error, + "error_type": "bad_input", + "tests": {name: {"passed": False, "error": error} for name in REQUIRED_TESTS}, + } + + def _demo_result() -> dict[str, Any]: """Return the demo-mode mutual TLS probe contract.""" return { @@ -262,7 +266,8 @@ def run_mutual_tls_probe( east_west_provider_hidden_message: str | None = None, ) -> dict[str, Any]: """Build the SEC13-01 JSON contract for the given endpoint/cert inputs.""" - if not north_south_endpoints and not east_west_endpoints and not east_west_provider_hidden_message: + # Pure provider-hidden with nothing probed is not evidence — skip instead. + if not north_south_endpoints and not east_west_endpoints: return { "success": True, "platform": "security", @@ -275,82 +280,49 @@ def run_mutual_tls_probe( ), } - if (north_south_endpoints or east_west_endpoints) and not (ca_cert and client_cert and client_key): - error = ( + if not (ca_cert and client_cert and client_key): + return _bad_input_result( "mTLS probe requires --ca-cert, --client-cert, and --client-key " "(or MTLS_CA_CERT_PATH / MTLS_CLIENT_CERT_PATH / MTLS_CLIENT_KEY_PATH)" ) - return { - "success": False, - "platform": "security", - "test_name": "mutual_tls", - "error": error, - "error_type": "bad_input", - "tests": {name: {"passed": False, "error": error} for name in REQUIRED_TESTS}, - } - tests: dict[str, Any] = {} - endpoints_tested = 0 + anonymous_context = _ssl_context(ca_cert=ca_cert) + authenticated_context = _ssl_context(ca_cert=ca_cert, client_cert=client_cert, client_key=client_key) - if north_south_endpoints: - assert ca_cert and client_cert and client_key - tests["north_south_mtls_enforced"] = _aggregate_plane( - north_south_endpoints, - plane="north_south", - ca_cert=ca_cert, - client_cert=client_cert, - client_key=client_key, - timeout=timeout, - ) - endpoints_tested += len(north_south_endpoints) - else: - tests["north_south_mtls_enforced"] = _provider_hidden_plane( + planes = [ + ( + "north_south_mtls_enforced", "north_south", + north_south_endpoints, "north_south_mtls_enforced: no north-south endpoints configured for this run", - ) - - if east_west_endpoints: - assert ca_cert and client_cert and client_key - tests["east_west_mtls_enforced"] = _aggregate_plane( - east_west_endpoints, - plane="east_west", - ca_cert=ca_cert, - client_cert=client_cert, - client_key=client_key, - timeout=timeout, - ) - endpoints_tested += len(east_west_endpoints) - elif east_west_provider_hidden_message: - tests["east_west_mtls_enforced"] = _provider_hidden_plane( - "east_west", - east_west_provider_hidden_message, - ) - else: - tests["east_west_mtls_enforced"] = _provider_hidden_plane( + ), + ( + "east_west_mtls_enforced", "east_west", - "east_west_mtls_enforced: no east-west endpoints configured for this run", - ) - - # Pure provider-hidden with nothing probed is not evidence — skip instead. - if endpoints_tested < 1: - return { - "success": True, - "platform": "security", - "test_name": "mutual_tls", - "skipped": True, - "skip_reason": ( - "No SEC13-01 endpoints configured " - "(set EDGE_ENDPOINTS and/or EAST_WEST_ENDPOINTS with MTLS_CA_CERT_PATH, " - "MTLS_CLIENT_CERT_PATH, MTLS_CLIENT_KEY_PATH)" - ), - } + east_west_endpoints, + east_west_provider_hidden_message + or "east_west_mtls_enforced: no east-west endpoints configured for this run", + ), + ] + tests: dict[str, Any] = {} + for name, plane, endpoints, hidden_message in planes: + if endpoints: + tests[name] = _aggregate_plane( + endpoints, + plane=plane, + anonymous_context=anonymous_context, + authenticated_context=authenticated_context, + timeout=timeout, + ) + else: + tests[name] = _provider_hidden_plane(plane, hidden_message) success = all(tests[name].get("passed") is True for name in REQUIRED_TESTS) return { "success": success, "platform": "security", "test_name": "mutual_tls", - "endpoints_tested": endpoints_tested, + "endpoints_tested": len(north_south_endpoints) + len(east_west_endpoints), "tests": tests, } @@ -404,15 +376,7 @@ def main() -> int: client_cert = _require_file(args.client_cert, "Client cert") if args.client_cert else None client_key = _require_file(args.client_key, "Client key") if args.client_key else None except ValueError as exc: - result = { - "success": False, - "platform": "security", - "test_name": "mutual_tls", - "error": str(exc), - "error_type": "bad_input", - "tests": {name: {"passed": False, "error": str(exc)} for name in REQUIRED_TESTS}, - } - print(json.dumps(result, indent=2)) + print(json.dumps(_bad_input_result(str(exc)), indent=2)) return 1 result = run_mutual_tls_probe( @@ -425,8 +389,6 @@ def main() -> int: east_west_provider_hidden_message=args.east_west_provider_hidden_message or None, ) print(json.dumps(result, indent=2)) - if result.get("skipped") is True: - return 0 return 0 if result.get("success") is True else 1 diff --git a/scripts/tests/test_mutual_tls_probe.py b/scripts/tests/test_mutual_tls_probe.py index 45434b540..5a65bd832 100644 --- a/scripts/tests/test_mutual_tls_probe.py +++ b/scripts/tests/test_mutual_tls_probe.py @@ -45,12 +45,8 @@ def test_parse_endpoints_accepts_host_port_list() -> None: def test_parse_endpoints_rejects_missing_port() -> None: """Endpoints without a port raise ValueError.""" - try: + with pytest.raises(ValueError, match="host:port"): probe._parse_endpoints("edge.example") - except ValueError as exc: - assert "host:port" in str(exc) - else: - raise AssertionError("expected ValueError") def test_demo_result_emits_both_planes() -> None: @@ -103,9 +99,8 @@ def fake_probe( host: str, port: int, *, - ca_cert: Path, - client_cert: Path, - client_key: Path, + anonymous_context: ssl.SSLContext, + authenticated_context: ssl.SSLContext, timeout: float, plane: str, ) -> dict[str, Any]: @@ -119,7 +114,10 @@ def fake_probe( "detail": {}, } - with patch.object(probe, "probe_mtls_endpoint", side_effect=fake_probe): + with ( + patch.object(probe, "_ssl_context", return_value=MagicMock()), + patch.object(probe, "probe_mtls_endpoint", side_effect=fake_probe), + ): result = probe.run_mutual_tls_probe( north_south_endpoints=[("edge.example", 443)], east_west_endpoints=[], From ee05570cd98671641d523e5189f35d6afc806295 Mon Sep 17 00:00:00 2001 From: Alexandre Begnoche Date: Mon, 13 Jul 2026 16:19:55 -0400 Subject: [PATCH 4/4] fix(security): address CodeRabbit findings on SEC13-01 probe Enable hostname verification with a CA, require TLS-layer rejection for anonymous probes, fail omitted planes unless east-west is explicitly provider-hidden, catch BotoCoreError during SEC11 terminate waits, and document the probe test helper. Signed-off-by: Alexandre Begnoche --- .../scripts/security/tenant_isolation_test.py | 4 +- .../providers/shared/mutual_tls_test.py | 27 +++-- scripts/tests/test_mutual_tls_probe.py | 102 +++++++++++++++++- 3 files changed, 123 insertions(+), 10 deletions(-) diff --git a/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py index 983bfd664..8a8211d1e 100644 --- a/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py +++ b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py @@ -415,12 +415,12 @@ def _teardown_tenant( # of cleanup; the safety-net teardown step picks up leftovers. try: ec2.terminate_instances(InstanceIds=[tenant.instance_id]) - except ClientError as e: + except (ClientError, BotoCoreError) as e: errors.append(f"terminate instance {tenant.instance_id}: {e}") else: try: _wait_instance_terminated(ec2, tenant.instance_id) - except (ClientError, TimeoutError) as e: + except (ClientError, BotoCoreError, TimeoutError) as e: errors.append(f"wait terminated {tenant.instance_id}: {e}") if tenant.created.get("volume") and tenant.volume_id: diff --git a/isvctl/configs/providers/shared/mutual_tls_test.py b/isvctl/configs/providers/shared/mutual_tls_test.py index e213ad5d6..fc449cd3a 100755 --- a/isvctl/configs/providers/shared/mutual_tls_test.py +++ b/isvctl/configs/providers/shared/mutual_tls_test.py @@ -118,10 +118,13 @@ def _ssl_context( ) -> ssl.SSLContext: """Build an SSL context that optionally presents a client certificate.""" context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - context.check_hostname = False - context.verify_mode = ssl.CERT_REQUIRED if ca_cert is not None else ssl.CERT_NONE if ca_cert is not None: + context.check_hostname = True + context.verify_mode = ssl.CERT_REQUIRED context.load_verify_locations(cafile=str(ca_cert)) + else: + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE if client_cert is not None and client_key is not None: context.load_cert_chain(certfile=str(client_cert), keyfile=str(client_key)) return context @@ -133,7 +136,12 @@ def _handshake( context: ssl.SSLContext, timeout: float, ) -> dict[str, Any]: - """Attempt a TLS handshake and return accepted/rejected classification.""" + """Attempt a TLS handshake and return accepted/rejected classification. + + ``rejection`` is ``tls`` for SSL-layer denials and ``transport`` for DNS, + connect, and timeout failures so callers do not treat unreachable hosts as + mTLS enforcement. + """ result: dict[str, Any] = {} try: with socket.create_connection((host, port), timeout=timeout) as raw: @@ -143,10 +151,10 @@ def _handshake( result["tls_version"] = tls.version() return result except ssl.SSLError as exc: - result.update(accepted=False, detail=f"SSLError: {exc}") + result.update(accepted=False, rejection="tls", detail=f"SSLError: {exc}") return result except OSError as exc: - result.update(accepted=False, detail=f"{type(exc).__name__}: {exc}") + result.update(accepted=False, rejection="transport", detail=f"{type(exc).__name__}: {exc}") return result @@ -162,7 +170,8 @@ def probe_mtls_endpoint( """Probe one endpoint: anonymous client rejected, authenticated client accepted.""" anonymous = _handshake(host, port, anonymous_context, timeout) authenticated = _handshake(host, port, authenticated_context, timeout) - anonymous_rejected = anonymous.get("accepted") is not True + # Only a TLS-layer rejection proves mTLS; connection refused / timeout is not evidence. + anonymous_rejected = anonymous.get("rejection") == "tls" authenticated_accepted = authenticated.get("accepted") is True passed = anonymous_rejected and authenticated_accepted return { @@ -314,8 +323,12 @@ def run_mutual_tls_probe( authenticated_context=authenticated_context, timeout=timeout, ) + elif name == "east_west_mtls_enforced" and east_west_provider_hidden_message: + tests[name] = _provider_hidden_plane(plane, east_west_provider_hidden_message) else: - tests[name] = _provider_hidden_plane(plane, hidden_message) + # Missing required plane without an explicit provider exception is a fail, + # not a fabricated pass. + tests[name] = {"passed": False, "message": hidden_message, "probes": []} success = all(tests[name].get("passed") is True for name in REQUIRED_TESTS) return { diff --git a/scripts/tests/test_mutual_tls_probe.py b/scripts/tests/test_mutual_tls_probe.py index 5a65bd832..5c7edf693 100644 --- a/scripts/tests/test_mutual_tls_probe.py +++ b/scripts/tests/test_mutual_tls_probe.py @@ -104,6 +104,7 @@ def fake_probe( timeout: float, plane: str, ) -> dict[str, Any]: + """Return a successful synthetic endpoint probe.""" return { "host": host, "port": port, @@ -135,7 +136,7 @@ def fake_probe( def test_handshake_marks_ssl_error_as_rejected() -> None: - """SSL errors during wrap_socket classify as not accepted.""" + """SSL errors during wrap_socket classify as TLS rejection.""" context = MagicMock() raw = MagicMock() raw.__enter__ = MagicMock(return_value=raw) @@ -151,9 +152,108 @@ def raise_ssl(*_args: object, **_kwargs: object) -> None: result = probe._handshake("edge.example", 443, context, 1.0) assert result["accepted"] is False + assert result["rejection"] == "tls" assert "SSLError" in result["detail"] +def test_handshake_marks_os_error_as_transport() -> None: + """DNS / connect / timeout failures classify as transport, not TLS rejection.""" + with patch.object(probe.socket, "create_connection", side_effect=ConnectionRefusedError("refused")): + result = probe._handshake("edge.example", 443, MagicMock(), 1.0) + + assert result["accepted"] is False + assert result["rejection"] == "transport" + + +def test_probe_does_not_treat_transport_failure_as_anonymous_rejection() -> None: + """Anonymous connect failure plus authenticated success must not pass.""" + anonymous_ctx = MagicMock() + authenticated_ctx = MagicMock() + + def fake_handshake( + host: str, + port: int, + context: ssl.SSLContext, + timeout: float, + ) -> dict[str, Any]: + if context is anonymous_ctx: + return {"accepted": False, "rejection": "transport", "detail": "ConnectionRefusedError"} + return {"accepted": True, "tls_version": "TLSv1.3"} + + with patch.object(probe, "_handshake", side_effect=fake_handshake): + result = probe.probe_mtls_endpoint( + "edge.example", + 443, + anonymous_context=anonymous_ctx, + authenticated_context=authenticated_ctx, + timeout=1.0, + plane="north_south", + ) + + assert result["anonymous_rejected"] is False + assert result["authenticated_accepted"] is True + assert result["passed"] is False + + +def test_run_fails_when_north_south_missing_without_provider_exception(tmp_path: Path) -> None: + """Only east-west endpoints without north-south must fail, not fabricate a pass.""" + ca = tmp_path / "ca.pem" + cert = tmp_path / "client.pem" + key = tmp_path / "client.key" + for path in (ca, cert, key): + path.write_text("placeholder\n", encoding="utf-8") + + def fake_probe( + host: str, + port: int, + *, + anonymous_context: ssl.SSLContext, + authenticated_context: ssl.SSLContext, + timeout: float, + plane: str, + ) -> dict[str, Any]: + """Return a successful synthetic endpoint probe.""" + return { + "host": host, + "port": port, + "plane": plane, + "anonymous_rejected": True, + "authenticated_accepted": True, + "passed": True, + "detail": {}, + } + + with ( + patch.object(probe, "_ssl_context", return_value=MagicMock()), + patch.object(probe, "probe_mtls_endpoint", side_effect=fake_probe), + ): + result = probe.run_mutual_tls_probe( + north_south_endpoints=[], + east_west_endpoints=[("mesh.internal", 8443)], + ca_cert=ca, + client_cert=cert, + client_key=key, + timeout=1.0, + ) + + assert result["success"] is False + assert result["tests"]["north_south_mtls_enforced"]["passed"] is False + assert "no north-south endpoints" in result["tests"]["north_south_mtls_enforced"]["message"] + assert result["tests"]["east_west_mtls_enforced"]["passed"] is True + + +def test_ssl_context_enables_hostname_check_with_ca(tmp_path: Path) -> None: + """CA-backed contexts verify the server hostname.""" + ca = tmp_path / "ca.pem" + # Minimal PEM so load_verify_locations can open a file; OpenSSL may still + # reject the contents, so mock load_verify_locations. + ca.write_text("placeholder\n", encoding="utf-8") + with patch.object(ssl.SSLContext, "load_verify_locations"): + context = probe._ssl_context(ca_cert=ca) + assert context.check_hostname is True + assert context.verify_mode == ssl.CERT_REQUIRED + + def test_main_demo_mode(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: """CLI demo mode prints the demo contract and exits 0.""" monkeypatch.setattr(probe, "DEMO_MODE", True)