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..901fd5057 100644 --- a/isvctl/configs/providers/aws/config/security.yaml +++ b/isvctl/configs/providers/aws/config/security.yaml @@ -31,20 +31,21 @@ # 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) -# 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) +# 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) +# 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 @@ -105,7 +106,25 @@ 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 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 ../../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) # 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/tenant_isolation_test.py b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py index c7d245b9c..8a8211d1e 100644 --- a/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py +++ b/isvctl/configs/providers/aws/scripts/security/tenant_isolation_test.py @@ -263,6 +263,40 @@ def _launch_instance(ec2: Any, tenant: Tenant, ami_id: str) -> None: tenant.created["instance"] = True +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", []) + ] + # 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" + raise TimeoutError(msg) + + def _create_volume(ec2: Any, tenant: Tenant) -> None: """Create a 1 GiB gp3 EBS volume in the tenant's AZ.""" response = ec2.create_volume( @@ -377,22 +411,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: + except (ClientError, BotoCoreError) 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, BotoCoreError, TimeoutError) as e: errors.append(f"wait terminated {tenant.instance_id}: {e}") if tenant.created.get("volume") and tenant.volume_id: @@ -863,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/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..fc449cd3a --- /dev/null +++ b/isvctl/configs/providers/shared/mutual_tls_test.py @@ -0,0 +1,409 @@ +#!/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) + 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 + + +def _handshake( + host: str, + port: int, + context: ssl.SSLContext, + timeout: float, +) -> dict[str, Any]: + """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: + 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, rejection="tls", detail=f"SSLError: {exc}") + return result + except OSError as exc: + result.update(accepted=False, rejection="transport", detail=f"{type(exc).__name__}: {exc}") + return result + + +def probe_mtls_endpoint( + host: str, + port: int, + *, + 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, anonymous_context, timeout) + authenticated = _handshake(host, port, authenticated_context, timeout) + # 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 { + "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, + anonymous_context: ssl.SSLContext, + authenticated_context: ssl.SSLContext, + timeout: float, +) -> dict[str, Any]: + """Probe all endpoints for one traffic plane and aggregate pass/fail.""" + probes = [ + probe_mtls_endpoint( + host, + port, + anonymous_context=anonymous_context, + authenticated_context=authenticated_context, + 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 _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 { + "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.""" + # 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", + "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 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)" + ) + + anonymous_context = _ssl_context(ca_cert=ca_cert) + authenticated_context = _ssl_context(ca_cert=ca_cert, client_cert=client_cert, client_key=client_key) + + planes = [ + ( + "north_south_mtls_enforced", + "north_south", + north_south_endpoints, + "north_south_mtls_enforced: no north-south endpoints configured for this run", + ), + ( + "east_west_mtls_enforced", + "east_west", + 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, + ) + elif name == "east_west_mtls_enforced" and east_west_provider_hidden_message: + tests[name] = _provider_hidden_plane(plane, east_west_provider_hidden_message) + else: + # 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 { + "success": success, + "platform": "security", + "test_name": "mutual_tls", + "endpoints_tested": len(north_south_endpoints) + len(east_west_endpoints), + "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: + print(json.dumps(_bad_input_result(str(exc)), 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)) + 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/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 ---------------------------------------- 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..5c7edf693 --- /dev/null +++ b/scripts/tests/test_mutual_tls_probe.py @@ -0,0 +1,267 @@ +# 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.""" + with pytest.raises(ValueError, match="host:port"): + probe._parse_endpoints("edge.example") + + +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, + *, + 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=[("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 TLS rejection.""" + 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 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) + 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