diff --git a/docs/test-plan.yaml b/docs/test-plan.yaml index 8bc32cc88..800c5b69d 100644 --- a/docs/test-plan.yaml +++ b/docs/test-plan.yaml @@ -179,6 +179,8 @@ domains: notes: "" github_issues: - "#248" + labels: + - vm - name: Identity and Asset Mgmt (IAM) capabilities: - description: Manages users, service identities, roles, and permissions across tenants and services. Forms the security foundation of the platform. diff --git a/isvctl/configs/providers/aws/config/vm.yaml b/isvctl/configs/providers/aws/config/vm.yaml index f278df720..8079c5c68 100644 --- a/isvctl/configs/providers/aws/config/vm.yaml +++ b/isvctl/configs/providers/aws/config/vm.yaml @@ -24,14 +24,16 @@ # 3. verify_tags - boto3: Verifies user-defined tags (test) # 4. serial_console - boto3: Retrieves serial console output (test) # 5. console_rbac - boto3: Simulates EC2 serial console IAM access (test) -# 6. stop_instance - boto3: Stops EC2, verifies stopped state (test) -# 7. start_instance - boto3: Starts stopped EC2, verifies recovery (test) -# 8. reboot_instance - boto3: Reboots EC2, validates recovery (test) -# 9. describe_instance - boto3: Describes current state, passes SSH info (test) +# 6. component_key_access - boto3: AUTH03 specified-key SOL access (test) +# 7. virtual_device_hardening - guest/hypervisor device hardening (test) +# 8. stop_instance - boto3: Stops EC2, verifies stopped state (test) +# 9. start_instance - boto3: Starts stopped EC2, verifies recovery (test) +# 10. reboot_instance - boto3: Reboots EC2, validates recovery (test) +# 11. describe_instance - boto3: Describes current state, passes SSH info (test) # (runs after reboot so IP is current; validation anchor) -# 10. deploy_nim - paramiko: Deploys NIM container via SSH (test) -# 11. teardown_nim - paramiko: Stops NIM container via SSH (teardown) -# 12. teardown - boto3: Terminates EC2 (teardown) +# 12. deploy_nim - paramiko: Deploys NIM container via SSH (test) +# 13. teardown_nim - paramiko: Stops NIM container via SSH (teardown) +# 14. teardown - boto3: Terminates EC2 (teardown) # # Usage: # uv run isvctl test run -f isvctl/configs/providers/aws/config/vm.yaml @@ -105,6 +107,21 @@ commands: - "{{region}}" timeout: 120 + # AWS-specific: Specified key access to SOL (AUTH03); network devices provider-hidden + - name: component_key_access + phase: test + command: "python3 ../scripts/vm/component_key_access.py" + args: + - "--instance-id" + - "{{steps.launch_instance.instance_id}}" + - "--key-file" + - "{{steps.launch_instance.key_file}}" + - "--key-name" + - "{{steps.launch_instance.key_name}}" + - "--region" + - "{{region}}" + timeout: 120 + # AWS-specific: Virtual device hardening evidence - name: virtual_device_hardening phase: test diff --git a/isvctl/configs/providers/aws/scripts/common/serial_console.py b/isvctl/configs/providers/aws/scripts/common/serial_console.py index d7b7e9a0b..0056849d1 100644 --- a/isvctl/configs/providers/aws/scripts/common/serial_console.py +++ b/isvctl/configs/providers/aws/scripts/common/serial_console.py @@ -27,6 +27,7 @@ from botocore.exceptions import ClientError RETENTION_EVIDENCE_LIMITATION = "EC2 GetConsoleOutput does not prove one-month log retention" +SERIAL_ACCESS_DISABLED_SKIP_REASON = "EC2 serial console access is disabled for this account or region" SERIAL_CONSOLE_RETENTION_DAYS_REQUIRED = 30 diff --git a/isvctl/configs/providers/aws/scripts/iam/test_credentials.py b/isvctl/configs/providers/aws/scripts/iam/test_credentials.py index aefd02931..9f1ac0771 100755 --- a/isvctl/configs/providers/aws/scripts/iam/test_credentials.py +++ b/isvctl/configs/providers/aws/scripts/iam/test_credentials.py @@ -28,8 +28,8 @@ "account_id": "123456789", "arn": "arn:aws:iam::...", "tests": { - "sts_identity": {"passed": true}, - "iam_access": {"passed": true} + "identity": {"passed": true}, + "access": {"passed": true} } } """ @@ -69,9 +69,9 @@ def test_sts_identity(session: boto3.Session, result: dict) -> bool: result["account_id"] = identity["Account"] result["arn"] = identity["Arn"] result["user_id"] = identity["UserId"] - result["tests"]["sts_identity"] = {"passed": True} + result["tests"]["identity"] = {"passed": True} if attempt > 0: - result["tests"]["sts_identity"]["retries"] = attempt + result["tests"]["identity"]["retries"] = attempt return True except ClientError as e: last_error = e @@ -88,7 +88,7 @@ def test_sts_identity(session: boto3.Session, result: dict) -> bool: error_type, error_msg = classify_aws_error(last_error) result["error_type"] = error_type result["error"] = error_msg - result["tests"]["sts_identity"] = {"passed": False, "error": error_msg} + result["tests"]["identity"] = {"passed": False, "error": error_msg} return False @@ -117,19 +117,19 @@ def main() -> int: print(json.dumps(result, indent=2)) return 1 - # Test IAM access + # Test authorized resource access (IAM GetUser, or AccessDenied proving authz) try: iam = session.client("iam") iam.get_user() - result["tests"]["iam_access"] = {"passed": True} + result["tests"]["access"] = {"passed": True} except ClientError as e: if e.response["Error"]["Code"] == "AccessDenied": # Access denied means credentials work but limited permissions - result["tests"]["iam_access"] = {"passed": True, "note": "Access denied (expected)"} + result["tests"]["access"] = {"passed": True, "note": "Access denied (expected)"} else: - result["tests"]["iam_access"] = {"passed": False, "error": str(e)} + result["tests"]["access"] = {"passed": False, "error": str(e)} - result["success"] = result["tests"]["sts_identity"]["passed"] + result["success"] = result["tests"]["identity"]["passed"] and result["tests"]["access"]["passed"] print(json.dumps(result, indent=2)) return 0 if result["success"] else 1 diff --git a/isvctl/configs/providers/aws/scripts/vm/component_key_access.py b/isvctl/configs/providers/aws/scripts/vm/component_key_access.py new file mode 100755 index 000000000..b159438e8 --- /dev/null +++ b/isvctl/configs/providers/aws/scripts/vm/component_key_access.py @@ -0,0 +1,156 @@ +#!/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. + +"""Prove a specified SSH key can access SOL (serial console) on AWS EC2. + +AUTH03-01: after AUTH02 launches an instance with a requested key, push that +key's public material via EC2 Instance Connect serial-console authorization. +Tenant-visible network-device SSH is not offered on AWS, so that probe is +marked provider-hidden. + +Usage: + python component_key_access.py --instance-id i-xxx --key-file /tmp/key.pem \\ + --key-name isv-test-key --region us-west-2 +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +import boto3 +from botocore.exceptions import ClientError +from common.errors import handle_aws_errors +from common.serial_console import SERIAL_ACCESS_DISABLED_SKIP_REASON, check_serial_access +from paramiko import ECDSAKey, Ed25519Key, RSAKey +from paramiko.ssh_exception import SSHException + + +def _load_openssh_public_key(key_file: str) -> str: + """Derive an OpenSSH public-key line from a private key PEM file.""" + path = Path(key_file) + if not path.is_file(): + msg = f"Key file not found: {key_file}" + raise FileNotFoundError(msg) + + loaders = (RSAKey, ECDSAKey, Ed25519Key) + last_error: Exception | None = None + for loader in loaders: + try: + private_key = loader.from_private_key_file(str(path)) + return f"{private_key.get_name()} {private_key.get_base64()}" + except (SSHException, OSError, ValueError) as exc: + last_error = exc + + msg = f"Unable to load private key from {key_file}: {last_error}" + raise ValueError(msg) + + +def _probe_sol_access(ec2: Any, eic: Any, instance_id: str, ssh_public_key: str) -> dict[str, Any]: + """Authorize the specified public key for EC2 serial-console SSH.""" + serial_access = check_serial_access(ec2) + if serial_access.get("error"): + return {"passed": False, "error": serial_access["error"], "probes": ["serial_console_access"]} + if serial_access.get("enabled") is not True: + return { + "passed": False, + "skipped": True, + "skip_reason": SERIAL_ACCESS_DISABLED_SKIP_REASON, + "probes": ["serial_console_access"], + } + + try: + response = eic.send_serial_console_ssh_public_key( + InstanceId=instance_id, + SSHPublicKey=ssh_public_key, + SerialPort=0, + ) + except ClientError as exc: + return {"passed": False, "error": str(exc), "probes": ["send_serial_console_ssh_public_key"]} + + return { + "passed": bool(response.get("Success")), + "message": "Authorized specified key for EC2 serial console SSH", + "probes": ["serial_console_access", "send_serial_console_ssh_public_key"], + } + + +def _probe_network_device_access() -> dict[str, Any]: + """AWS has no tenant-visible network-device key path; mark provider-hidden.""" + return { + "passed": True, + "provider_hidden": True, + "message": "AWS does not expose tenant-visible network-device SSH for key-based access", + "probes": ["network_device_ssh"], + } + + +@handle_aws_errors +def main() -> int: + """Run AUTH03 key-based component access probes and emit JSON.""" + parser = argparse.ArgumentParser(description="Prove specified key access to SOL / network devices") + parser.add_argument("--instance-id", required=True, help="EC2 instance ID") + parser.add_argument("--key-file", required=True, help="Path to the instance private key PEM") + parser.add_argument("--key-name", required=True, help="Key pair name requested at launch (AUTH02)") + parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-west-2")) + args = parser.parse_args() + + result: dict[str, Any] = { + "success": False, + "platform": "vm", + "test_name": "component_key_access", + "instance_id": args.instance_id, + "key_name": args.key_name, + "tests": {}, + } + + try: + ssh_public_key = _load_openssh_public_key(args.key_file) + except (OSError, ValueError) as exc: + result["error"] = str(exc) + result["tests"]["sol_access"] = {"passed": False, "error": str(exc)} + print(json.dumps(result, indent=2)) + return 1 + + ec2 = boto3.client("ec2", region_name=args.region) + eic = boto3.client("ec2-instance-connect", region_name=args.region) + + sol = _probe_sol_access(ec2, eic, args.instance_id, ssh_public_key) + if sol.get("skipped") is True: + result["success"] = True + result["skipped"] = True + result["skip_reason"] = sol["skip_reason"] + print(json.dumps(result, indent=2)) + return 0 + + result["tests"]["sol_access"] = sol + result["tests"]["network_device_access"] = _probe_network_device_access() + result["success"] = all(test["passed"] for test in result["tests"].values()) + if not result["success"] and sol.get("error"): + result["error"] = sol["error"] + + print(json.dumps(result, indent=2)) + return 0 if result["success"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/providers/aws/scripts/vm/console_rbac.py b/isvctl/configs/providers/aws/scripts/vm/console_rbac.py index 4c9f8a1ae..a8299b952 100644 --- a/isvctl/configs/providers/aws/scripts/vm/console_rbac.py +++ b/isvctl/configs/providers/aws/scripts/vm/console_rbac.py @@ -41,13 +41,12 @@ import botocore.session from botocore.exceptions import ClientError from common.errors import handle_aws_errors -from common.serial_console import check_serial_access +from common.serial_console import SERIAL_ACCESS_DISABLED_SKIP_REASON, check_serial_access CONSOLE_ACTION = "ec2-instance-connect:SendSerialConsoleSSHPublicKey" TEST_USER_PREFIX = "isv-console-rbac-test-" ALLOW_POLICY_NAME = "IsvConsoleRbacScopedAccess" OWNER_TAG = {"Key": "CreatedBy", "Value": "isvtest"} -SERIAL_ACCESS_DISABLED_SKIP_REASON = "EC2 serial console access is disabled for this account or region" REQUIRED_TESTS = ( "serial_console_access_enabled", "denied_principal_cannot_access_console", diff --git a/isvctl/configs/providers/my-isv/config/vm.yaml b/isvctl/configs/providers/my-isv/config/vm.yaml index 3ac8cde85..bf85f470e 100644 --- a/isvctl/configs/providers/my-isv/config/vm.yaml +++ b/isvctl/configs/providers/my-isv/config/vm.yaml @@ -25,7 +25,7 @@ # (InstanceStateCheck, InstanceCreatedCheck, InstanceListCheck, # InstanceTagCheck, InstanceStopCheck, InstanceStartCheck, # StableIdentifierCheck, InstanceRebootCheck, SerialConsoleCheck, -# ConsoleRbacCheck, StepSuccessCheck). SSH-based host checks (ConnectivityCheck, +# ConsoleRbacCheck, ComponentKeyAccessCheck, StepSuccessCheck). SSH-based host checks (ConnectivityCheck, # GpuCheck, DriverCheck, CpuInfoCheck, NIM checks, etc.) are excluded # via `exclude.labels: [ssh]` because a dummy-success stub cannot # spin up a real host to SSH into. The deploy_nim / teardown_nim @@ -45,10 +45,12 @@ # 6. reboot_instance - Reboot it # 7. serial_console - Confirm serial console access # 8. console_rbac - Verify console access is RBAC-scoped -# 9. deploy_nim - SKIPPED (needs real SSH) +# 9. component_key_access - AUTH03 specified-key SOL / network device access +# 10. virtual_device_hardening - Confirm unused virtio devices are absent +# 11. deploy_nim - SKIPPED (needs real SSH) # TEARDOWN -# 10. teardown_nim - SKIPPED (needs real SSH) -# 11. teardown - Terminate the VM +# 12. teardown_nim - SKIPPED (needs real SSH) +# 13. teardown - Terminate the VM # # Usage: # uv run isvctl test run -f isvctl/configs/providers/my-isv/config/vm.yaml @@ -154,6 +156,20 @@ commands: - "{{region}}" timeout: 60 + - name: component_key_access + phase: test + command: "python ../scripts/vm/component_key_access.py" + args: + - "--instance-id" + - "{{steps.launch_instance.instance_id}}" + - "--key-file" + - "{{steps.launch_instance.key_file}}" + - "--key-name" + - "{{steps.launch_instance.key_name}}" + - "--region" + - "{{region}}" + timeout: 60 + - name: virtual_device_hardening phase: test command: "python ../scripts/vm/virtual_device_hardening.py" diff --git a/isvctl/configs/providers/my-isv/scripts/README.md b/isvctl/configs/providers/my-isv/scripts/README.md index 569b78282..3d7b89dba 100644 --- a/isvctl/configs/providers/my-isv/scripts/README.md +++ b/isvctl/configs/providers/my-isv/scripts/README.md @@ -29,7 +29,7 @@ template, then fill in the TODOs. |--------|---------|----------|---------------|---------------| | `iam/` | 3 | [`suites/iam.yaml`](../../../suites/iam.yaml) | [`config/iam.yaml`](../config/iam.yaml) | [`providers/aws/scripts/iam/`](../../aws/scripts/iam/) | | `control-plane/` | 10 | [`suites/control-plane.yaml`](../../../suites/control-plane.yaml) | [`config/control-plane.yaml`](../config/control-plane.yaml) | [`providers/aws/scripts/control-plane/`](../../aws/scripts/control-plane/) | -| `vm/` | 9 | [`suites/vm.yaml`](../../../suites/vm.yaml) | [`config/vm.yaml`](../config/vm.yaml) | [`providers/aws/scripts/vm/`](../../aws/scripts/vm/) | +| `vm/` | 10 | [`suites/vm.yaml`](../../../suites/vm.yaml) | [`config/vm.yaml`](../config/vm.yaml) | [`providers/aws/scripts/vm/`](../../aws/scripts/vm/) | | `bare_metal/` | 12 | [`suites/bare_metal.yaml`](../../../suites/bare_metal.yaml) | [`config/bare_metal.yaml`](../config/bare_metal.yaml) | [`providers/aws/scripts/bare_metal/`](../../aws/scripts/bare_metal/) | | `storage/` | 17 | [`suites/storage.yaml`](../../../suites/storage.yaml) | [`config/storage.yaml`](../config/storage.yaml) | [`providers/aws/scripts/storage/`](../../aws/scripts/storage/) | | `network/` | 18 | [`suites/network.yaml`](../../../suites/network.yaml) | [`config/network.yaml`](../config/network.yaml) | [`providers/aws/scripts/network/`](../../aws/scripts/network/) | diff --git a/isvctl/configs/providers/my-isv/scripts/vm/component_key_access.py b/isvctl/configs/providers/my-isv/scripts/vm/component_key_access.py new file mode 100755 index 000000000..c838703ac --- /dev/null +++ b/isvctl/configs/providers/my-isv/scripts/vm/component_key_access.py @@ -0,0 +1,125 @@ +#!/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. + +"""Component key access (SOL / network devices) - TEMPLATE. + +AUTH03-01: prove the specified key from launch can access serial-over-LAN +and network devices where the platform exposes them. + +Required JSON output fields: + { + "success": true, + "platform": "vm", + "test_name": "component_key_access", + "instance_id": "", + "key_name": "", + "tests": { + "sol_access": {"passed": true}, + "network_device_access": {"passed": true} + } + } + +When a component class is not customer-visible, mark that subtest +``provider_hidden: true`` with ``passed: true`` instead of failing. + +Usage: + python component_key_access.py --instance-id --key-file \\ + --key-name --region + +Reference implementation: ../../../aws/scripts/vm/component_key_access.py +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from typing import Any + +# ISVCTL_DEMO_MODE=1 enables demo-success output (used by `make demo-test`). +DEMO_MODE = os.environ.get("ISVCTL_DEMO_MODE") == "1" + + +def _demo_result(instance_id: str, key_name: str) -> dict[str, Any]: + """Return a passing demo payload for the AUTH03 contract.""" + return { + "success": True, + "platform": "vm", + "test_name": "component_key_access", + "instance_id": instance_id, + "key_name": key_name, + "tests": { + "sol_access": { + "passed": True, + "message": "Demo SOL access via specified key", + "probes": ["serial_console_ssh"], + }, + "network_device_access": { + "passed": True, + "message": "Demo network-device access via specified key", + "probes": ["network_device_ssh"], + }, + }, + } + + +def main() -> int: + """Probe key-based SOL / network-device access (template) and emit JSON.""" + parser = argparse.ArgumentParser(description="Component key access (template)") + parser.add_argument("--instance-id", required=True, help="Instance identifier") + parser.add_argument("--key-file", required=True, help="Path to the instance private key") + parser.add_argument("--key-name", required=True, help="Key name requested at launch") + parser.add_argument("--region", default="my-isv-region-1", help="Region") + args = parser.parse_args() + + result: dict[str, Any] = { + "success": False, + "platform": "vm", + "test_name": "component_key_access", + "instance_id": args.instance_id, + "key_name": args.key_name, + "tests": {}, + } + + # ╔══════════════════════════════════════════════════════════════════╗ + # ║ TODO: Replace this block with your platform's key-access proof ║ + # ║ ║ + # ║ Example (pseudocode): ║ + # ║ sol = platform.authorize_sol_key( ║ + # ║ instance_id=args.instance_id, ║ + # ║ key_file=args.key_file, ║ + # ║ ) ║ + # ║ result["tests"]["sol_access"] = {"passed": sol.ok} ║ + # ║ net = platform.probe_network_device_key(args.key_file) ║ + # ║ result["tests"]["network_device_access"] = { ║ + # ║ "passed": True, ║ + # ║ "provider_hidden": not net.available, ║ + # ║ } ║ + # ║ result["success"] = True ║ + # ╚══════════════════════════════════════════════════════════════════╝ + + if DEMO_MODE: + result = _demo_result(args.instance_id, args.key_name) + else: + result["error"] = "Not implemented - replace with your platform's component key access logic" + + print(json.dumps(result, indent=2)) + return 0 if result["success"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/suites/README.md b/isvctl/configs/suites/README.md index aadec5d52..23ce1632d 100644 --- a/isvctl/configs/suites/README.md +++ b/isvctl/configs/suites/README.md @@ -31,7 +31,7 @@ For the domain / script-count / AWS-reference overview see the | Step | Phase | Script | Key JSON Fields | |------|-------|--------|-----------------| | `create_user` | setup | `providers/my-isv/scripts/iam/create_user.py` | `username`, `user_id`, `access_key_id`, `secret_access_key` | -| `test_credentials` | test | `providers/my-isv/scripts/iam/test_credentials.py` | `account_id`, `tests.identity.passed`, `tests.access.passed` | +| `test_credentials` | test | `providers/my-isv/scripts/iam/test_credentials.py` | `account_id`, `tests.identity.passed`, `tests.access.passed` (`IamCredentialAccessCheck` / IAM03-01) | | `teardown` | teardown | `providers/my-isv/scripts/iam/delete_user.py` | `resources_deleted`, `message` | ### Network (`network.yaml`) @@ -81,10 +81,11 @@ For the domain / script-count / AWS-reference overview see the | Step | Phase | Script | Key JSON Fields | |------|-------|--------|-----------------| -| `launch_instance` | setup | `providers/my-isv/scripts/vm/launch_instance.py` | `instance_id`, `public_ip`, `key_file`, `vpc_id` | +| `launch_instance` | setup | `providers/my-isv/scripts/vm/launch_instance.py` | `instance_id`, `public_ip`, `key_file`, `vpc_id`, `requested_key_name`, `key_name` | | `list_instances` | test | `providers/my-isv/scripts/vm/list_instances.py` | `instances`, `total_count` | | `verify_tags` | test | `providers/my-isv/scripts/vm/describe_tags.py` | `instance_id`, `tags`, `tag_count` | | `serial_console` | test | `providers/my-isv/scripts/vm/serial_console.py` | `console_available`, `serial_access_enabled` | +| `component_key_access` | test | `providers/my-isv/scripts/vm/component_key_access.py` | `key_name`; for non-skipped results also `tests.sol_access.passed`, `tests.network_device_access.passed` (`ComponentKeyAccessCheck` / AUTH03-01; AWS may emit top-level `skipped` when serial console access is disabled) | | `stop_instance` | test | `providers/my-isv/scripts/vm/stop_instance.py` | `instance_id`, `state`, `stop_initiated` | | `start_instance` | test | `providers/my-isv/scripts/vm/start_instance.py` | `instance_id`, `state`, `public_ip`, `ssh_ready` | | `reboot_instance` | test | `providers/my-isv/scripts/vm/reboot_instance.py` | `reboot_initiated`, `ssh_ready`, `uptime_seconds` | diff --git a/isvctl/configs/suites/iam.yaml b/isvctl/configs/suites/iam.yaml index 028f626f8..4ac3be7e1 100644 --- a/isvctl/configs/suites/iam.yaml +++ b/isvctl/configs/suites/iam.yaml @@ -62,10 +62,9 @@ tests: credentials: step: test_credentials checks: - FieldExistsCheck: + IamCredentialAccessCheck: test_id: "IAM03-01" labels: ["iam"] - field: "account_id" StepSuccessCheck: test_id: "N/A" labels: ["iam"] diff --git a/isvctl/configs/suites/vm.yaml b/isvctl/configs/suites/vm.yaml index af0f2efb2..92f11e6aa 100644 --- a/isvctl/configs/suites/vm.yaml +++ b/isvctl/configs/suites/vm.yaml @@ -49,7 +49,8 @@ tests: # These are PROVIDER-AGNOSTIC. They only check JSON field names/values. # # Layout mirrors bare_metal.yaml: - # launch checks -> list -> tags -> serial_console -> console_rbac -> cloud_init + # launch checks -> list -> tags -> serial_console -> console_rbac + # -> component_key_access (AUTH03) -> cloud_init # -> describe_instance (SSH / GPU / host anchor, post-reboot) # -> lifecycle (stop / start / reboot) # -> NIM -> teardown @@ -69,6 +70,13 @@ tests: test_id: "AUTH02-01" labels: ["vm"] + component_key_access: + step: component_key_access + checks: + ComponentKeyAccessCheck: + test_id: "AUTH03-01" + labels: ["vm"] + instance_created: step: launch_instance checks: diff --git a/isvctl/tests/test_component_key_access_contract.py b/isvctl/tests/test_component_key_access_contract.py new file mode 100644 index 000000000..dedba8992 --- /dev/null +++ b/isvctl/tests/test_component_key_access_contract.py @@ -0,0 +1,132 @@ +# 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. + +"""Contract tests for AUTH03 component key access scripts.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import pytest +from botocore.exceptions import ClientError +from paramiko import RSAKey + +from .conftest import load_vm_script + +ISVCTL_ROOT = Path(__file__).resolve().parents[1] +MY_ISV_VM_SCRIPTS = ISVCTL_ROOT / "configs" / "providers" / "my-isv" / "scripts" / "vm" + + +def test_load_openssh_public_key_from_rsa_pem(tmp_path: Path) -> None: + """Private key PEM derives an OpenSSH public key line.""" + module = load_vm_script("component_key_access.py") + key_path = tmp_path / "isv-auth03-key.pem" + RSAKey.generate(2048).write_private_key_file(str(key_path)) + + public_key = module._load_openssh_public_key(str(key_path)) + + assert public_key.startswith("ssh-rsa ") + + +def test_probe_sol_access_skips_when_serial_disabled(monkeypatch: pytest.MonkeyPatch) -> None: + """Disabled serial console yields a skipped SOL probe.""" + module = load_vm_script("component_key_access.py") + monkeypatch.setattr(module, "check_serial_access", lambda _ec2: {"enabled": False}) + + result = module._probe_sol_access(MagicMock(), MagicMock(), "i-abc", "ssh-rsa AAAAB3") + + assert result["passed"] is False + assert result["skipped"] is True + + +def test_probe_sol_access_authorizes_key(monkeypatch: pytest.MonkeyPatch) -> None: + """Enabled serial console + Instance Connect Success proves SOL key access.""" + module = load_vm_script("component_key_access.py") + monkeypatch.setattr(module, "check_serial_access", lambda _ec2: {"enabled": True}) + eic = MagicMock() + eic.send_serial_console_ssh_public_key.return_value = {"Success": True, "RequestId": "req-1"} + + result = module._probe_sol_access(MagicMock(), eic, "i-abc", "ssh-rsa AAAAB3") + + assert result["passed"] is True + eic.send_serial_console_ssh_public_key.assert_called_once_with( + InstanceId="i-abc", + SSHPublicKey="ssh-rsa AAAAB3", + SerialPort=0, + ) + + +def test_probe_sol_access_fails_on_client_error(monkeypatch: pytest.MonkeyPatch) -> None: + """Instance Connect API errors fail the SOL probe.""" + module = load_vm_script("component_key_access.py") + monkeypatch.setattr(module, "check_serial_access", lambda _ec2: {"enabled": True}) + eic = MagicMock() + eic.send_serial_console_ssh_public_key.side_effect = ClientError( + {"Error": {"Code": "InvalidInstanceID.NotFound", "Message": "missing"}}, + "SendSerialConsoleSSHPublicKey", + ) + + result = module._probe_sol_access(MagicMock(), eic, "i-missing", "ssh-rsa AAAAB3") + + assert result["passed"] is False + assert "missing" in result["error"] + + +def test_network_device_access_is_provider_hidden() -> None: + """AWS marks network-device SSH as provider-hidden rather than failing.""" + module = load_vm_script("component_key_access.py") + + result = module._probe_network_device_access() + + assert result["passed"] is True + assert result["provider_hidden"] is True + + +def test_my_isv_demo_component_key_access_emits_contract() -> None: + """Demo mode emits the AUTH03 JSON contract for make demo-test.""" + script = MY_ISV_VM_SCRIPTS / "component_key_access.py" + env = os.environ | {"ISVCTL_DEMO_MODE": "1"} + + completed = subprocess.run( + [ + sys.executable, + str(script), + "--instance-id", + "dummy-vm-0001", + "--key-file", + "/tmp/dummy-key.pem", + "--key-name", + "isv-test-gpu", + "--region", + "my-isv-region-1", + ], + check=False, + capture_output=True, + text=True, + env=env, + ) + + assert completed.returncode == 0, completed.stderr + payload: dict[str, Any] = json.loads(completed.stdout) + assert payload["success"] is True + assert payload["key_name"] == "isv-test-gpu" + assert payload["tests"]["sol_access"]["passed"] is True + assert payload["tests"]["network_device_access"]["passed"] is True diff --git a/isvtest/src/isvtest/validations/iam.py b/isvtest/src/isvtest/validations/iam.py index 82cb4a13b..637219d82 100644 --- a/isvtest/src/isvtest/validations/iam.py +++ b/isvtest/src/isvtest/validations/iam.py @@ -21,13 +21,43 @@ from typing import ClassVar -from isvtest.core.validation import BaseValidation +from isvtest.core.validation import BaseValidation, check_required_tests # ============================================================================= # Access Key Validations # ============================================================================= +class IamCredentialAccessCheck(BaseValidation): + """Validate created IAM credentials authenticate and reach authorized resources. + + Proves IAM03-01: a user created earlier can access the API (identity) and at + least one authorized resource (access). Scripts emit provider-neutral + ``tests.identity`` / ``tests.access`` subtests. + + Config: + step_output: The test_credentials step output to check + required_tests: Optional override of required subtest names + (default: identity, access) + + Step output: + tests.identity.passed: True when credentials authenticate to the API + tests.access.passed: True when at least one authorized resource is reachable + """ + + description: ClassVar[str] = "Check IAM credentials authenticate with authorized resource access" + + def run(self) -> None: + """Validate identity and authorized-resource access probes from step output.""" + required = self.config.get("required_tests", ["identity", "access"]) + if not check_required_tests(self, required, "IAM credential access tests failed"): + return + + step_output = self.config.get("step_output", {}) + account_id = step_output.get("account_id") or "unknown" + self.set_passed(f"IAM credentials authenticated with authorized resource access (account={account_id})") + + class AccessKeyCreatedCheck(BaseValidation): """Validate access key was created successfully. diff --git a/isvtest/src/isvtest/validations/instance.py b/isvtest/src/isvtest/validations/instance.py index bc7a1636d..c9f3631a6 100644 --- a/isvtest/src/isvtest/validations/instance.py +++ b/isvtest/src/isvtest/validations/instance.py @@ -20,7 +20,7 @@ from typing import ClassVar -from isvtest.core.validation import BaseValidation +from isvtest.core.validation import BaseValidation, check_required_tests from isvtest.validations.generic import check_operations_passed SERIAL_CONSOLE_RETENTION_DAYS_REQUIRED = 30 @@ -100,6 +100,47 @@ def run(self) -> None: self.set_passed(f"Instance {instance_id} launched with specified key '{requested_key_name}'") +class ComponentKeyAccessCheck(BaseValidation): + """Validate a specified key can access other components (SOL, network devices). + + Proves AUTH03-01 after AUTH02 launches an instance with a requested key. + Scripts emit provider-neutral ``tests.sol_access`` and + ``tests.network_device_access`` probes. Network-device access may be marked + ``provider_hidden`` when the platform does not expose tenant-visible device + SSH (for example AWS). + + Config: + step_output: The component_key_access step output to check + required_tests: Optional override of required subtest names + (default: sol_access, network_device_access) + + Step output: + key_name: Key used for component access (links to AUTH02) + tests.sol_access.passed: True when SOL/serial console accepts the key + tests.network_device_access.passed: True when network-device key access + succeeds, or is affirmatively provider-hidden + """ + + description: ClassVar[str] = "Check specified key accesses SOL and network devices" + + def run(self) -> None: + """Validate key-based SOL and network-device access probes from step output.""" + step_output = self.config.get("step_output", {}) + key_name = step_output.get("key_name") + if not key_name: + self.set_failed("No 'key_name' in step output") + return + + required = self.config.get("required_tests", ["sol_access", "network_device_access"]) + if not check_required_tests(self, required, "Component key access tests failed"): + return + + tests = step_output.get("tests", {}) + hidden = [name for name in required if tests[name].get("provider_hidden") is True] + hidden_note = f", provider_hidden={','.join(hidden)}" if hidden else "" + self.set_passed(f"Specified key '{key_name}' accessed required components{hidden_note}") + + class InstanceRebootCheck(BaseValidation): """Validate that an instance was rebooted successfully. diff --git a/isvtest/tests/test_iam.py b/isvtest/tests/test_iam.py index bacbe4734..62d2bce16 100644 --- a/isvtest/tests/test_iam.py +++ b/isvtest/tests/test_iam.py @@ -19,7 +19,7 @@ from typing import Any -from isvtest.validations.iam import ServiceAccountCredentialCheck +from isvtest.validations.iam import IamCredentialAccessCheck, ServiceAccountCredentialCheck def _sa_credential_output(**overrides: Any) -> dict[str, Any]: @@ -38,6 +38,69 @@ def _sa_credential_output(**overrides: Any) -> dict[str, Any]: return output +def _credential_access_output(**overrides: Any) -> dict[str, Any]: + """Return a valid IAM03 credential-access step output.""" + output: dict[str, Any] = { + "success": True, + "platform": "iam", + "account_id": "123456789012", + "tests": { + "identity": {"passed": True}, + "access": {"passed": True}, + }, + } + output.update(overrides) + return output + + +class TestIamCredentialAccessCheck: + """Tests for IamCredentialAccessCheck (IAM03-01).""" + + def test_passes_with_identity_and_access(self) -> None: + """Happy path: both identity and access probes pass.""" + result = IamCredentialAccessCheck(config={"step_output": _credential_access_output()}).execute() + + assert result["passed"] is True + assert "authenticated with authorized resource access" in result["output"] + assert "123456789012" in result["output"] + + def test_fails_when_identity_missing(self) -> None: + """Fails when the identity probe is absent.""" + out = _credential_access_output() + del out["tests"]["identity"] + + result = IamCredentialAccessCheck(config={"step_output": out}).execute() + + assert result["passed"] is False + assert "identity" in result["error"] + + def test_fails_when_access_fails(self) -> None: + """Fails when authorized-resource access does not pass.""" + result = IamCredentialAccessCheck( + config={ + "step_output": _credential_access_output( + tests={ + "identity": {"passed": True}, + "access": {"passed": False, "error": "permission denied"}, + } + ) + } + ).execute() + + assert result["passed"] is False + assert "access" in result["error"] + + def test_fails_when_tests_absent(self) -> None: + """Fails when the tests object is missing entirely.""" + out = _credential_access_output() + del out["tests"] + + result = IamCredentialAccessCheck(config={"step_output": out}).execute() + + assert result["passed"] is False + assert "tests" in result["error"] + + def test_sa_credential_check_passes_with_long_lived_key() -> None: """Passes for the AWS-style long-lived access-key source.""" result = ServiceAccountCredentialCheck(config={"step_output": _sa_credential_output()}).execute() diff --git a/isvtest/tests/test_instance.py b/isvtest/tests/test_instance.py index a1a6101d0..7797df47c 100644 --- a/isvtest/tests/test_instance.py +++ b/isvtest/tests/test_instance.py @@ -19,8 +19,11 @@ from typing import Any +import pytest + from isvtest.validations.instance import ( SERIAL_CONSOLE_RETENTION_DAYS_REQUIRED, + ComponentKeyAccessCheck, InstanceRebootCheck, InstanceSpecifiedKeyCheck, SerialConsoleRetentionCheck, @@ -203,6 +206,90 @@ def test_fails_when_actual_key_differs_from_requested_key(self) -> None: assert "expected key 'isv-test-key', got 'other-key'" in result["error"] +class TestComponentKeyAccessCheck: + """Tests for AUTH03-01 specified-key SOL / network-device access.""" + + def _output(self, **overrides: Any) -> dict[str, Any]: + """Build a minimal passing component_key_access step_output.""" + base: dict[str, Any] = { + "success": True, + "platform": "vm", + "instance_id": "i-abc123", + "key_name": "isv-test-key", + "tests": { + "sol_access": {"passed": True}, + "network_device_access": {"passed": True}, + }, + } + base.update(overrides) + return base + + def test_passes_with_sol_and_network_access(self) -> None: + """Happy path: both required component probes pass.""" + result = ComponentKeyAccessCheck(config={"step_output": self._output()}).execute() + + assert result["passed"] is True + assert "isv-test-key" in result["output"] + + def test_passes_with_provider_hidden_network_devices(self) -> None: + """Network-device access may be provider-hidden when not tenant-visible.""" + result = ComponentKeyAccessCheck( + config={ + "step_output": self._output( + tests={ + "sol_access": {"passed": True}, + "network_device_access": { + "passed": True, + "provider_hidden": True, + "message": "no tenant network-device SSH", + }, + } + ) + } + ).execute() + + assert result["passed"] is True + assert "provider_hidden=network_device_access" in result["output"] + + def test_fails_when_key_name_missing(self) -> None: + """The provider must report which key was used.""" + out = self._output() + del out["key_name"] + + result = ComponentKeyAccessCheck(config={"step_output": out}).execute() + + assert result["passed"] is False + assert "key_name" in result["error"] + + def test_fails_when_sol_access_fails(self) -> None: + """SOL access failure fails AUTH03.""" + result = ComponentKeyAccessCheck( + config={ + "step_output": self._output( + tests={ + "sol_access": {"passed": False, "error": "serial disabled"}, + "network_device_access": {"passed": True}, + } + ) + } + ).execute() + + assert result["passed"] is False + assert "sol_access" in result["error"] + + def test_skips_when_step_marked_skipped(self) -> None: + """Whole-step skip (e.g. serial console disabled) is a pytest skip.""" + with pytest.raises(pytest.skip.Exception, match="serial console"): + ComponentKeyAccessCheck( + config={ + "step_output": self._output( + skipped=True, + skip_reason="EC2 serial console access is disabled for this account or region", + ) + } + ).execute() + + class TestSerialConsoleRetentionCheck: """Tests for one-month serial console retention evidence validation."""