diff --git a/docs/test-plan.adoc b/docs/test-plan.adoc index ef82d64e..bee365fe 100644 --- a/docs/test-plan.adoc +++ b/docs/test-plan.adoc @@ -190,7 +190,7 @@ a| https://github.com/NVIDIA/ai-cloud-validation/issues/247[#247] | [[AUTH03-01]]AUTH03-01 | AUTH03 | https://github.com/NVIDIA/ai-cloud-validation/issues/248[#248] -| +| bare_metal, iam, security, vm | | P1 a| https://github.com/NVIDIA/ai-cloud-validation/issues/248[#248] diff --git a/docs/test-plan.yaml b/docs/test-plan.yaml index e1635216..e6587078 100644 --- a/docs/test-plan.yaml +++ b/docs/test-plan.yaml @@ -180,6 +180,9 @@ domains: github_issues: - "#248" labels: + - bare_metal + - iam + - security - vm - name: Identity and Asset Mgmt (IAM) capabilities: diff --git a/isvctl/configs/providers/nico/config/bare_metal.yaml b/isvctl/configs/providers/nico/config/bare_metal.yaml index 37696bed..02c962f5 100644 --- a/isvctl/configs/providers/nico/config/bare_metal.yaml +++ b/isvctl/configs/providers/nico/config/bare_metal.yaml @@ -57,6 +57,11 @@ # nico-admin-cli into the provider-neutral attestation contract (SEC22-01 / # CNP09-02). It requires operator/admin CLI credentials in addition to the # tenant REST credentials used for machine enumeration. +# - query_key_access.py reads the site's SSH Key Groups and serial-console +# configuration and reports whether a tenant-specified key can access +# out-of-band components; SpecifiedKeyAccessCheck validates it (AUTH-XX-03). +# This is a read-only step (like the other checks here); the auto-provisioning +# variant that mints a throwaway key lives in the dedicated key_access.yaml. # - list_instances.py / describe_instance.py query existing instances; # InstanceListCheck and InstanceStateCheck validate the JSON contract. # @@ -312,6 +317,26 @@ commands: - "{{nico_api_base}}" timeout: 120 + # ─── Specified-Key Access (AUTH-XX-03) ────────────────────────── + # Reads the site's SSH Key Groups and serial-console configuration from + # the NICo API (/carbide/ segment, like the other NICo steps) and reports + # whether a tenant-specified key can access out-of-band components. The serial + # console (SOL) is verifiable from the tenant API (provider-enabled + + # SSH-key auth + key synced); network-device access is provider-managed + # and reported as unverified. SpecifiedKeyAccessCheck validates it. + - name: query_key_access + phase: test + continue_on_failure: true + command: "python ../scripts/auth/query_key_access.py" + args: + - "--org" + - "{{org}}" + - "--site-id" + - "{{site_id}}" + - "--api-base" + - "{{nico_api_base}}" + timeout: 120 + # ─── List Existing Instances ─────────────────────────────────── # Lists instances visible to NICo at the configured site. - name: list_instances diff --git a/isvctl/configs/providers/nico/config/key_access.yaml b/isvctl/configs/providers/nico/config/key_access.yaml new file mode 100644 index 00000000..efbcb34e --- /dev/null +++ b/isvctl/configs/providers/nico/config/key_access.yaml @@ -0,0 +1,125 @@ +# 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. + +# NICo Specified-Key Access Validation Configuration (AUTH-XX-03) +# +# Self-contained, auto-provisioning variant of the AUTH-XX-03 check. Unlike the +# read-only query_key_access step in bare_metal.yaml (which skips when no SSH +# key group is synced to the site), this config provisions a throwaway specified +# key, runs the check, then tears the key down -- so it can go green in CI +# without any manual key setup on the cluster. +# +# Architecture (imports the bare_metal suite for the SpecifiedKeyAccessCheck): +# setup setup_key_access.py - create an ephemeral SSH key + SSH Key +# Group synced to the site (private key is +# generated and discarded immediately). +# test query_key_access.py - report whether the specified key can +# access the serial console (SOL); +# SpecifiedKeyAccessCheck validates it. +# teardown teardown_key_access.py - delete the key group + key created in +# setup (IDs come from the setup output). +# +# Only the specified_key_access validation has its step wired here; the rest of +# the imported bare_metal suite skips (steps not configured), exactly like the +# other single-purpose NICo runs. +# +# Prerequisites: +# - NICO_API_BASE set to the NICo API base URL +# - NICO_BEARER_TOKEN set, or OIDC client credentials configured via +# NICO_SSA_ISSUER, NICO_CLIENT_ID, and NICO_CLIENT_SECRET +# - NICO_ORGANIZATION and NICO_SITE_ID environment variables set +# - ssh-keygen available on the host (used to mint the throwaway key) +# - The site's serial console must be provider-enabled for a full pass; +# otherwise the check reports the serial console as unverified. +# +# Usage: +# NICO_BEARER_TOKEN= NICO_ORGANIZATION= NICO_SITE_ID= \ +# ISVTEST_INCLUDE_UNRELEASED=1 \ +# uv run isvctl test run -f isvctl/configs/providers/nico/config/key_access.yaml + +import: + - ../../../suites/bare_metal.yaml + +version: "1.0" + +commands: + bare_metal: + phases: ["setup", "test", "teardown"] + steps: + # ─── Provision a Throwaway Specified Key (setup) ──────────────── + # Creates an ephemeral SSH key and a key group that syncs it to the + # site. Emits sshkey_id / sshkeygroup_id / restore_ssh_keys_enabled for + # teardown to consume. + - name: setup_key_access + phase: setup + command: "python ../scripts/auth/setup_key_access.py" + args: + - "--org" + - "{{org}}" + - "--site-id" + - "{{site_id}}" + - "--api-base" + - "{{nico_api_base}}" + timeout: 360 + + # ─── Specified-Key Access (AUTH-XX-03, test) ──────────────────── + # Same step the bare_metal suite validates with SpecifiedKeyAccessCheck. + - name: query_key_access + phase: test + command: "python ../scripts/auth/query_key_access.py" + args: + - "--org" + - "{{org}}" + - "--site-id" + - "{{site_id}}" + - "--api-base" + - "{{nico_api_base}}" + timeout: 120 + + # ─── Remove the Throwaway Specified Key (teardown) ────────────── + # Best-effort cleanup of the key group + key created in setup, and + # restores the site's SSH-key SOL flag when setup flipped it. + - name: teardown_key_access + phase: teardown + command: "python ../scripts/auth/teardown_key_access.py" + args: + - "--org" + - "{{org}}" + - "--site-id" + - "{{site_id}}" + - "--api-base" + - "{{nico_api_base}}" + - "--sshkeygroup-id" + - "{{steps.setup_key_access.sshkeygroup_id}}" + - "--sshkey-id" + - "{{steps.setup_key_access.sshkey_id}}" + - "--restore-ssh-keys-enabled" + - "{{steps.setup_key_access.restore_ssh_keys_enabled}}" + timeout: 120 + +tests: + cluster_name: "nico-key-access-validation" + description: "Verify a specified key can access out-of-band components (AUTH-XX-03)" + + settings: + org: "{{env.NICO_ORGANIZATION}}" + site_id: "{{env.NICO_SITE_ID}}" + nico_api_base: "{{env.NICO_API_BASE}}" + # The imported bare_metal suite defines these for the full instance + # lifecycle, which this config does not run. Blank them so the suite's + # {{instance_type}} self-reference does not emit missing-variable warnings. + region: "" + instance_type: "" + teardown_flag: "" diff --git a/isvctl/configs/providers/nico/scripts/auth/query_key_access.py b/isvctl/configs/providers/nico/scripts/auth/query_key_access.py new file mode 100644 index 00000000..069dbb22 --- /dev/null +++ b/isvctl/configs/providers/nico/scripts/auth/query_key_access.py @@ -0,0 +1,258 @@ +#!/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. + +"""Report specified-key access to out-of-band components for a NICo site (AUTH-XX-03). + +AUTH-XX-03 verifies that a tenant-specified key (an SSH key) can be used to +access other components "as possible", with the serial console (SOL) and +network devices called out as examples. + +NICo models a tenant-specified key as an SSH Key, grouped into an SSH Key Group +that is synced down to one or more Sites. A Site exposes a serial console (SOL) +whose access the provider enables (``isSerialConsoleEnabled``) and whose +SSH-key authentication the tenant enables (``isSerialConsoleSSHKeysEnabled``). +When a key group with at least one key is synced to a site whose serial console +is enabled with SSH-key auth, that key grants SOL access to the site's machines. + +This script gathers the provider-neutral evidence for that access path: + +- **specified keys** -- the SSH keys in the key groups synced to the site. Only + the count and fingerprint-derived posture are surfaced; key material is never + emitted. +- **serial console (SOL)** -- per-site serial-console configuration and whether + the synced key can reach it. +- **network devices** -- reported as ``key_access_enabled: null`` (unverified): + tenant key access to switches is provider-managed and not exposed by the NICo + tenant REST API, so it is neither asserted nor falsely passed. + +NICo API endpoints used (the ``/carbide/`` segment is the current deployed name +for what newer docs call ``/nico/``; the other NICo scripts use it too): + GET /{org}/carbide/site/{site_id} + GET /{org}/carbide/sshkeygroup?siteId={site_id} + +Auth: + - NICO_BEARER_TOKEN, or OIDC client_credentials + (NICO_SSA_ISSUER / NICO_CLIENT_ID / NICO_CLIENT_SECRET). + +When no SSH key group with a key is synced to the site there is nothing to +evidence access with, so the script emits a structured skip (``skipped: true`` ++ ``skip_reason``) carrying an ``org_key_groups`` count, distinguishing "no key +groups exist at all" from "key groups exist but none are synced to this site". + +Required JSON output fields: + { + "success": true, + "platform": "nico", + "site_id": "...", + "specified_keys": 1, + "access_targets": [ + { + "type": "serial_console", + "name": " serial console (SOL)", + "key_access_enabled": true, // tri-state: true | false | null (unverified) + "reachable": true, // endpoint present AND key synced to site + "detail": "..." + }, + { + "type": "network_device", + "name": "...", + "key_access_enabled": null, + "reachable": false, + "detail": "..." + } + ] + } + +Usage: + NICO_BEARER_TOKEN= \ + python query_key_access.py --org --site-id --api-base + + Wired via the bare_metal suite: + uv run isvctl test run -f isvctl/configs/providers/nico/config/bare_metal.yaml + +Reference: + infra-controller rest-api/openapi/spec.yaml (Site, SshKeyGroup schemas) +""" + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +# Allow importing from sibling common/ directory +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from common.nico_client import ( + NicoAuthError, + forge_get, + forge_get_all, + resolve_auth, + sshkeygroup_synced_to_site, +) + + +def _count_specified_keys(groups: list[dict[str, Any]]) -> int: + """Count distinct SSH keys across the key groups synced to the site.""" + key_ids: set[str] = set() + keys_without_id = 0 + for group in groups: + for key in group.get("sshKeys") or []: + key_id = key.get("id") if isinstance(key, dict) else None + if isinstance(key_id, str) and key_id: + key_ids.add(key_id) + else: + keys_without_id += 1 + return len(key_ids) + keys_without_id + + +def _keys_synced_to_site(groups: list[dict[str, Any]], site_id: str) -> bool: + """Return whether a key group with at least one key is synced to the site.""" + return any((group.get("sshKeys") or []) and sshkeygroup_synced_to_site(group, site_id) for group in groups) + + +def _serial_console_target(site: dict[str, Any]) -> dict[str, Any]: + """Build the serial-console (SOL) access target from the site config. + + Only called once a specified key is synced to the site, so reachability + reduces to the console endpoint being present. + """ + name = f"{site.get('name') or site.get('id') or 'site'} serial console (SOL)" + + if not site.get("isSerialConsoleEnabled"): + return { + "type": "serial_console", + "name": name, + "key_access_enabled": None, + "reachable": False, + "detail": "Provider has not enabled the serial console (SOL) for this site", + } + + key_access_enabled = bool(site.get("isSerialConsoleSSHKeysEnabled")) + hostname = (site.get("serialConsoleHostname") or "").strip() + + if not key_access_enabled: + detail = "Serial console is enabled but SSH-key access is disabled (isSerialConsoleSSHKeysEnabled is false)" + elif not hostname: + detail = "Serial console SSH-key access is enabled but no serial console hostname is configured" + else: + detail = "Specified key can access the serial console (SOL): provider-enabled, SSH-key auth on, key synced" + + return { + "type": "serial_console", + "name": name, + "key_access_enabled": key_access_enabled, + "reachable": bool(hostname), + "detail": detail, + } + + +def _network_device_target() -> dict[str, Any]: + """Build the (unverified) network-device access target. + + Tenant key access to network switches is provider-managed and is not + exposed by the NICo tenant REST API, so it is reported as unverified rather + than asserted or falsely passed. + """ + return { + "type": "network_device", + "name": "Network devices", + "key_access_enabled": None, + "reachable": False, + "detail": ( + "Tenant key access to network devices is provider-managed and not exposed by the " + "NICo tenant REST API; not verifiable from the tenant API" + ), + } + + +def main() -> int: + """Gather specified-key access evidence and print the JSON contract to stdout.""" + parser = argparse.ArgumentParser( + description="Report specified-key (SSH) access to out-of-band components on a NICo site" + ) + parser.add_argument("--org", required=True, help="NGC org name") + parser.add_argument("--site-id", required=True, help="NICo site UUID") + parser.add_argument("--api-base", required=True, help="NICo API base URL") + args = parser.parse_args() + + result: dict[str, Any] = { + "success": False, + "platform": "nico", + "site_id": args.site_id, + "specified_keys": 0, + "access_targets": [], + } + + try: + auth = resolve_auth() + + groups = forge_get_all( + args.org, + "sshkeygroup", + auth.token, + base_url=args.api_base, + params={"siteId": args.site_id}, + result_key="sshKeyGroups", + ) + + result["specified_keys"] = _count_specified_keys(groups) + + if not _keys_synced_to_site(groups, args.site_id): + # Nothing to evidence access with on this site. Query the org-wide + # key groups (no siteId filter) so the skip reason can distinguish + # "no key groups exist at all" from "key groups exist but none are + # synced to this site" -- the operator needs to know which. + org_groups = forge_get_all( + args.org, + "sshkeygroup", + auth.token, + base_url=args.api_base, + result_key="sshKeyGroups", + ) + result["org_key_groups"] = len(org_groups) + result["skipped"] = True + if org_groups: + result["skip_reason"] = ( + f"{len(org_groups)} SSH key group(s) exist for the org but none are synced to this site; " + "sync a key group containing an SSH key to the site to enable key-based SOL access" + ) + else: + result["skip_reason"] = ( + "No SSH key groups exist for the org; create one with an SSH key and sync it to the site " + "to enable key-based SOL access" + ) + else: + site = forge_get(args.org, f"site/{args.site_id}", auth.token, base_url=args.api_base) + result["access_targets"] = [ + _serial_console_target(site), + _network_device_target(), + ] + + result["success"] = True + + except NicoAuthError as e: + result["error_type"] = "auth" + result["error"] = str(e) + except Exception as e: + result["error"] = f"{type(e).__name__}: {e}" + + 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/nico/scripts/auth/setup_key_access.py b/isvctl/configs/providers/nico/scripts/auth/setup_key_access.py new file mode 100644 index 00000000..b585486f --- /dev/null +++ b/isvctl/configs/providers/nico/scripts/auth/setup_key_access.py @@ -0,0 +1,193 @@ +#!/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. + +"""Provision a throwaway specified key synced to a NICo site (AUTH-XX-03 setup). + +Creates an ephemeral SSH key and an SSH Key Group that syncs it to the target +site, so query_key_access.py can evidence key-based serial-console (SOL) access +end-to-end without manual setup. The matching private key is generated in a temp +dir and immediately discarded -- only the public key is registered, so the +credential is unusable by anyone and safe to leave until teardown removes it. + +teardown_key_access.py deletes everything this script creates (it reads the +created IDs from this step's output), so the IDs are always emitted -- even on a +mid-provision failure -- to keep cleanup reliable. + +NICo API endpoints used (``/carbide/`` segment, like the other NICo scripts): + POST /{org}/carbide/sshkey + POST /{org}/carbide/sshkeygroup + GET /{org}/carbide/sshkeygroup/{id} (poll for sync) + PATCH /{org}/carbide/site/{site_id} (best-effort; older API only) + +Auth: + - NICO_BEARER_TOKEN, or OIDC client_credentials + (NICO_SSA_ISSUER / NICO_CLIENT_ID / NICO_CLIENT_SECRET). + +Output fields (consumed by teardown_key_access.py via Jinja step references): + { + "success": true, + "platform": "nico", + "site_id": "...", + "sshkey_id": "...", # "" when not created + "sshkeygroup_id": "...", # "" when not created + "synced": true, + "restore_ssh_keys_enabled": false | null # prior site flag to restore, or null + } + +Usage: + NICO_BEARER_TOKEN= \ + python setup_key_access.py --org --site-id --api-base +""" + +import argparse +import json +import os +import subprocess +import sys +import tempfile +import time +import uuid +from pathlib import Path +from typing import Any + +# Allow importing from sibling common/ directory +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from common.nico_client import ( + NicoAuthError, + forge_get, + forge_patch, + forge_post, + resolve_auth, + sshkeygroup_synced_to_site, +) + +SYNC_POLL_TIMEOUT_SECONDS = 180 +SYNC_POLL_INTERVAL_SECONDS = 5 + + +def _generate_public_key(comment: str) -> str: + """Generate an ed25519 keypair and return only the public key. + + The private key lives in a temp dir that is removed on return, so the + registered public key has no usable counterpart. + """ + with tempfile.TemporaryDirectory() as tmp: + key_path = os.path.join(tmp, "id_ed25519") + subprocess.run( + ["ssh-keygen", "-t", "ed25519", "-N", "", "-C", comment, "-f", key_path, "-q"], + check=True, + capture_output=True, + ) + return Path(f"{key_path}.pub").read_text().strip() + + +def _wait_for_sync(org: str, group_id: str, site_id: str, token: str, *, base_url: str) -> bool: + """Poll the key group until it is synced to the site or the timeout elapses.""" + if not group_id: + # No group id means nothing to poll (the create response had no id). + return False + deadline = time.monotonic() + SYNC_POLL_TIMEOUT_SECONDS + while True: + group = forge_get(org, f"sshkeygroup/{group_id}", token, base_url=base_url) + if sshkeygroup_synced_to_site(group, site_id): + return True + if time.monotonic() >= deadline: + return False + time.sleep(SYNC_POLL_INTERVAL_SECONDS) + + +def main() -> int: + """Provision the throwaway specified key and print the JSON contract to stdout.""" + parser = argparse.ArgumentParser(description="Provision a throwaway SSH key synced to a NICo site") + parser.add_argument("--org", required=True, help="NGC org name") + parser.add_argument("--site-id", required=True, help="NICo site UUID") + parser.add_argument("--api-base", required=True, help="NICo API base URL") + args = parser.parse_args() + + suffix = uuid.uuid4().hex[:8] + name = f"isvtest-auth-xx-03-{suffix}" + + result: dict[str, Any] = { + "success": False, + "platform": "nico", + "site_id": args.site_id, + "sshkey_id": "", + "sshkeygroup_id": "", + "synced": False, + "restore_ssh_keys_enabled": None, + } + + try: + auth = resolve_auth() + + public_key = _generate_public_key(comment=name) + key = forge_post( + args.org, "sshkey", auth.token, base_url=args.api_base, body={"name": name, "publicKey": public_key} + ) + result["sshkey_id"] = key.get("id") or "" + + group = forge_post( + args.org, + "sshkeygroup", + auth.token, + base_url=args.api_base, + body={"name": name, "sshKeyIds": [result["sshkey_id"]], "siteIds": [args.site_id]}, + ) + result["sshkeygroup_id"] = group.get("id") or "" + + result["synced"] = _wait_for_sync( + args.org, result["sshkeygroup_id"], args.site_id, auth.token, base_url=args.api_base + ) + + if not result["synced"]: + result["error"] = "SSH key group did not sync to the site before timeout" + else: + # Older clusters gate SSH-key SOL access on a tenant-settable site flag; + # newer ones derive it from key-group sync (the flag is deprecated and + # the PATCH may be rejected). Enabling it is therefore best-effort: only + # record a restore value when we actually flipped it off->on. + site = forge_get(args.org, f"site/{args.site_id}", auth.token, base_url=args.api_base) + if not site.get("isSerialConsoleSSHKeysEnabled"): + try: + forge_patch( + args.org, + f"site/{args.site_id}", + auth.token, + base_url=args.api_base, + body={"isSerialConsoleSSHKeysEnabled": True}, + ) + result["restore_ssh_keys_enabled"] = False + except Exception: + # Deprecated/derived on this API version; nothing to restore. + pass + + result["success"] = True + + except NicoAuthError as e: + result["error_type"] = "auth" + result["error"] = str(e) + except FileNotFoundError: + result["error"] = "ssh-keygen not found; it is required to generate the throwaway specified key" + except Exception as e: + result["error"] = f"{type(e).__name__}: {e}" + + 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/nico/scripts/auth/teardown_key_access.py b/isvctl/configs/providers/nico/scripts/auth/teardown_key_access.py new file mode 100644 index 00000000..18f6a7e8 --- /dev/null +++ b/isvctl/configs/providers/nico/scripts/auth/teardown_key_access.py @@ -0,0 +1,138 @@ +#!/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. + +"""Remove the throwaway specified key provisioned for AUTH-XX-03 (teardown). + +Deletes the SSH Key Group and SSH Key created by setup_key_access.py and, when +setup flipped it, restores the site's serial-console SSH-key flag. The IDs are +passed in from the setup step output via Jinja; empty IDs mean setup created +nothing, so this is a no-op. DELETE 404 is treated as already removed +(idempotent teardown). Other failures on one resource do not stop the others; +errors are reported in ``cleanup_errors`` and fail the step. + +NICo API endpoints used (``/carbide/`` segment, like the other NICo scripts): + DELETE /{org}/carbide/sshkeygroup/{id} + DELETE /{org}/carbide/sshkey/{id} + PATCH /{org}/carbide/site/{site_id} (only when restoring the flag) + +Usage (wired via the key_access config; IDs come from the setup step): + python teardown_key_access.py --org --site-id --api-base \ + --sshkeygroup-id --sshkey-id --restore-ssh-keys-enabled +""" + +import argparse +import json +import sys +from pathlib import Path +from typing import Any +from urllib.error import HTTPError + +# Allow importing from sibling common/ directory +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from common.nico_client import NicoAuthError, forge_delete, forge_patch, resolve_auth + + +def _delete_if_present(org: str, path: str, token: str, *, base_url: str) -> None: + """DELETE a NICo resource; treat 404 as already removed.""" + try: + forge_delete(org, path, token, base_url=base_url) + except HTTPError as e: + if e.code == 404: + return + raise + + +def _as_bool(value: str) -> bool | None: + """Parse a Jinja-rendered flag into True/False, or None when unset.""" + normalized = (value or "").strip().lower() + if normalized in ("", "none", "null"): + return None + return {"true": True, "false": False}.get(normalized) + + +def main() -> int: + """Tear down the throwaway specified key and print the JSON contract to stdout.""" + parser = argparse.ArgumentParser(description="Remove the throwaway SSH key/group provisioned for AUTH-XX-03") + parser.add_argument("--org", required=True, help="NGC org name") + parser.add_argument("--site-id", required=True, help="NICo site UUID") + parser.add_argument("--api-base", required=True, help="NICo API base URL") + parser.add_argument("--sshkeygroup-id", default="", help="SSH Key Group ID created by setup") + parser.add_argument("--sshkey-id", default="", help="SSH Key ID created by setup") + parser.add_argument( + "--restore-ssh-keys-enabled", + default="", + help="Prior site isSerialConsoleSSHKeysEnabled value to restore (blank = leave as-is)", + ) + args = parser.parse_args() + + result: dict[str, Any] = { + "success": False, + "platform": "nico", + "site_id": args.site_id, + "cleanup_errors": [], + } + + group_id = (args.sshkeygroup_id or "").strip() + key_id = (args.sshkey_id or "").strip() + restore = _as_bool(args.restore_ssh_keys_enabled) + + try: + auth = resolve_auth() + + # Delete the group before the key: deleting the group removes the key's + # group membership, so the key can then be deleted on its own. + if group_id: + try: + _delete_if_present(args.org, f"sshkeygroup/{group_id}", auth.token, base_url=args.api_base) + except Exception as e: + result["cleanup_errors"].append(f"sshkeygroup {group_id}: {type(e).__name__}: {e}") + + if key_id: + try: + _delete_if_present(args.org, f"sshkey/{key_id}", auth.token, base_url=args.api_base) + except Exception as e: + result["cleanup_errors"].append(f"sshkey {key_id}: {type(e).__name__}: {e}") + + if restore is not None: + try: + forge_patch( + args.org, + f"site/{args.site_id}", + auth.token, + base_url=args.api_base, + body={"isSerialConsoleSSHKeysEnabled": restore}, + ) + except Exception as e: + result["cleanup_errors"].append(f"restore site flag: {type(e).__name__}: {e}") + + # A failure on one resource does not stop the others (the framework + # already runs teardown steps best-effort), but any failed deletion + # fails the step so a leaked key/group is visible. + result["success"] = not result["cleanup_errors"] + if result["cleanup_errors"]: + result["error"] = "; ".join(result["cleanup_errors"]) + + except NicoAuthError as e: + result["error_type"] = "auth" + result["error"] = str(e) + + 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/nico/scripts/common/nico_client.py b/isvctl/configs/providers/nico/scripts/common/nico_client.py index 6081fbd3..b54685bd 100644 --- a/isvctl/configs/providers/nico/scripts/common/nico_client.py +++ b/isvctl/configs/providers/nico/scripts/common/nico_client.py @@ -181,18 +181,38 @@ def forge_get( params: dict[str, str] | None = None, timeout: int = 30, ) -> dict[str, Any]: - """Make an authenticated GET request to a single NICo API page. + """Make an authenticated GET request to a single NICo API page.""" + return forge_request(org, path, token, base_url=base_url, params=params, timeout=timeout) + + +def forge_request( + org: str, + path: str, + token: str, + *, + base_url: str, + method: str = "GET", + params: dict[str, str] | None = None, + body: dict[str, Any] | None = None, + timeout: int = 30, +) -> dict[str, Any]: + """Make an authenticated request to the NICo API. Args: org: NGC org name. - path: API path relative to the API name segment (e.g., "machine", "expected-machine"). + path: API path relative to the API name segment (e.g., "sshkey", "sshkeygroup/{id}"). token: Bearer token. base_url: NICo API base URL. + method: HTTP method ("GET", "POST", "PATCH", "DELETE"). params: Query parameters (will be URL-encoded). + body: Optional JSON request body. timeout: Request timeout in seconds. Returns: - Parsed JSON response as a dict. + Parsed JSON response, or ``{}`` when the response body is empty. DELETE + also returns ``{}`` for a non-JSON body (legacy carbide answers with a + bare ``OK`` string); other methods raise ``JSONDecodeError`` on a + non-JSON body. Raises: HTTPError: On non-2xx response. @@ -200,16 +220,48 @@ def forge_get( url = f"{base_url}/{org}/{nico_api_name()}/{path}" if params: url = f"{url}?{urlencode(params)}" + data = json.dumps(body).encode() if body is not None else None + headers = {"Authorization": f"Bearer {token}"} + if data is not None: + headers["Content-Type"] = "application/json" - req = Request(url, headers={"Authorization": f"Bearer {token}"}) + req = Request(url, data=data, headers=headers, method=method) try: with urlopen(req, timeout=timeout) as resp: - return json.loads(resp.read().decode()) + raw = resp.read().decode() + if not raw.strip(): + return {} + try: + return json.loads(raw) + except json.JSONDecodeError: + if method.upper() == "DELETE": + # Legacy carbide DELETE may return a bare "OK" string. + return {} + raise except HTTPError as e: - body = "" + err_body = "" if e.fp: - body = e.fp.read().decode(errors="replace")[:500] - raise type(e)(e.url, e.code, f"{e.reason}: {body}", e.headers, None) from e + err_body = e.fp.read().decode(errors="replace")[:500] + raise type(e)(e.url, e.code, f"{e.reason}: {err_body}", e.headers, None) from e + + +def forge_post( + org: str, path: str, token: str, *, base_url: str, body: dict[str, Any], timeout: int = 30 +) -> dict[str, Any]: + """POST a JSON body to a NICo API path.""" + return forge_request(org, path, token, base_url=base_url, method="POST", body=body, timeout=timeout) + + +def forge_patch( + org: str, path: str, token: str, *, base_url: str, body: dict[str, Any], timeout: int = 30 +) -> dict[str, Any]: + """PATCH a JSON body to a NICo API path.""" + return forge_request(org, path, token, base_url=base_url, method="PATCH", body=body, timeout=timeout) + + +def forge_delete(org: str, path: str, token: str, *, base_url: str, timeout: int = 30) -> dict[str, Any]: + """DELETE a NICo API resource.""" + return forge_request(org, path, token, base_url=base_url, method="DELETE", timeout=timeout) def forge_get_all( @@ -279,6 +331,27 @@ def forge_get_all( return all_items +# SSH Key Group / site-association status that means the group's keys have +# fully propagated to the site and are therefore usable for access. +SYNCED_STATUS = "Synced" + + +def sshkeygroup_synced_to_site(group: dict[str, Any], site_id: str) -> bool: + """Return whether an SSH key group has reached ``Synced`` for the site. + + The group is accepted when either its own status is ``Synced`` or the + association for this specific site is ``Synced``. + """ + if group.get("status") == SYNCED_STATUS: + return True + for assoc in group.get("siteAssociations") or []: + if not isinstance(assoc, dict): + continue + if (assoc.get("site") or {}).get("id") == site_id and assoc.get("status") == SYNCED_STATUS: + return True + return False + + def classify_health(health: dict[str, Any]) -> str: """Classify machine health as 'healthy' or 'unhealthy'.""" alerts = health.get("alerts", []) diff --git a/isvctl/configs/suites/bare_metal.yaml b/isvctl/configs/suites/bare_metal.yaml index a1c030da..35276c43 100644 --- a/isvctl/configs/suites/bare_metal.yaml +++ b/isvctl/configs/suites/bare_metal.yaml @@ -653,5 +653,18 @@ tests: test_id: "CNP09-02" labels: ["attestation", "bare_metal", "firmware", "min_req", "security"] + # Specified-key access to out-of-band components (AUTH-XX-03). Verifies a + # tenant-specified key (e.g. an SSH key) can be used to access other + # components "as possible" -- the serial console (SOL) being the verifiable + # example, with network-device access reported as unverified (provider- + # managed). Only runs for providers that implement the query_key_access + # step. + specified_key_access: + step: query_key_access + checks: + SpecifiedKeyAccessCheck: + test_id: "AUTH03-01" + labels: ["bare_metal", "iam", "security", "vm"] + exclude: labels: [] diff --git a/isvctl/configs/suites/vm.yaml b/isvctl/configs/suites/vm.yaml index 92f11e6a..b7ee93dc 100644 --- a/isvctl/configs/suites/vm.yaml +++ b/isvctl/configs/suites/vm.yaml @@ -75,7 +75,7 @@ tests: checks: ComponentKeyAccessCheck: test_id: "AUTH03-01" - labels: ["vm"] + labels: ["bare_metal", "iam", "security", "vm"] instance_created: step: launch_instance diff --git a/isvctl/tests/providers/nico/test_nico_provider.py b/isvctl/tests/providers/nico/test_nico_provider.py index 69b22610..06d7aa21 100644 --- a/isvctl/tests/providers/nico/test_nico_provider.py +++ b/isvctl/tests/providers/nico/test_nico_provider.py @@ -27,6 +27,7 @@ from pathlib import Path from types import ModuleType, SimpleNamespace from typing import Any +from urllib.error import HTTPError from urllib.parse import parse_qs import pytest @@ -230,6 +231,21 @@ def _load_oob_health_script() -> ModuleType: return _load_nico_script("health/query_oob_health.py", "test_query_oob_health") +def _load_query_key_access_script() -> ModuleType: + """Load the query_key_access script as a module for direct unit testing.""" + return _load_nico_script("auth/query_key_access.py", "test_query_key_access") + + +def _load_setup_key_access_script() -> ModuleType: + """Load the setup_key_access script as a module for direct unit testing.""" + return _load_nico_script("auth/setup_key_access.py", "test_setup_key_access") + + +def _load_teardown_key_access_script() -> ModuleType: + """Load the teardown_key_access script as a module for direct unit testing.""" + return _load_nico_script("auth/teardown_key_access.py", "test_teardown_key_access") + + def test_nico_auth_prefers_explicit_bearer_token(monkeypatch: pytest.MonkeyPatch) -> None: """A locally supplied NICo bearer token should be the simplest auth path.""" module = _load_nico_client() @@ -354,6 +370,52 @@ def fake_urlopen(request: Any, timeout: int = 30) -> _Response: assert seen["url"] == f"http://127.0.0.1:8080/v2/org/ncx/{segment}/site/site-1" +@pytest.mark.parametrize("body", ["OK", "", " "]) +def test_forge_delete_tolerates_non_json_success_body(monkeypatch: pytest.MonkeyPatch, body: str) -> None: + """NICo DELETE answers with a bare ``OK`` (not JSON); a 2xx body must not raise.""" + module = _load_nico_client() + + class _RawResponse: + """Context-managed byte response for mocked HTTP calls.""" + + def __enter__(self) -> _RawResponse: + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self) -> bytes: + return body.encode() + + monkeypatch.setattr(module, "urlopen", lambda request, timeout=30: _RawResponse()) + + result = module.forge_delete("ncx", "sshkey/key-1", "tok", base_url="http://x") + + assert result == {} + + +def test_forge_post_rejects_non_json_success_body(monkeypatch: pytest.MonkeyPatch) -> None: + """POST responses must be JSON so resource IDs are not silently dropped.""" + + class _RawResponse: + """Context-managed byte response for mocked HTTP calls.""" + + def __enter__(self) -> _RawResponse: + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self) -> bytes: + return b"OK" + + module = _load_nico_client() + monkeypatch.setattr(module, "urlopen", lambda request, timeout=30: _RawResponse()) + + with pytest.raises(json.JSONDecodeError): + module.forge_post("ncx", "sshkey", "tok", base_url="http://x", body={"name": "k"}) + + @pytest.mark.parametrize( "step_name", [ @@ -368,6 +430,7 @@ def fake_urlopen(request: Any, timeout: int = 30) -> _Response: "query_sanitization", "query_stable_ips", "query_oob_health", + "query_key_access", ], ) def test_nico_bare_metal_config_exposes_api_base_setting(step_name: str) -> None: @@ -1019,6 +1082,7 @@ def test_nico_network_inventory_scripts_skip_when_site_has_no_network_inventory( ("query_sanitization.py", _load_sanitization_script), ("query_stable_ips.py", _load_stable_ips_script), ("query_oob_health.py", _load_oob_health_script), + ("query_key_access.py", _load_query_key_access_script), ], ) def test_nico_scripts_require_api_base( @@ -3047,3 +3111,369 @@ def test_topology_script_output_satisfies_check( bad.run() assert bad._passed is False assert "m-1" in bad._error + + +# =========================================================================== +# Specified-key access scripts (AUTH-XX-03): query / setup / teardown +# =========================================================================== + + +def _run_script_main( + module: ModuleType, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + argv: list[str], +) -> tuple[int, dict[str, Any]]: + """Run a NICo script's main() with argv and return (exit_code, parsed JSON).""" + monkeypatch.setattr(sys, "argv", argv) + code = module.main() + return code, json.loads(capsys.readouterr().out) + + +def test_query_key_access_reports_serial_console_accessible( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A key synced to a SOL-enabled, SSH-key-auth site yields a reachable target.""" + module = _load_query_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr( + module, + "forge_get", + lambda org, path, token, **kw: { + "name": "sjc-1", + "isSerialConsoleEnabled": True, + "isSerialConsoleSSHKeysEnabled": True, + "serialConsoleHostname": "sol.example.com", + }, + ) + monkeypatch.setattr( + module, + "forge_get_all", + lambda org, path, token, **kw: [ + { + "status": "Synced", + "sshKeys": [{"id": "k1"}], + "siteAssociations": [{"site": {"id": "site-1"}, "status": "Synced"}], + } + ], + ) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + ["query_key_access.py", "--org", "o", "--site-id", "site-1", "--api-base", "http://x"], + ) + + assert code == 0 + assert out["success"] is True + assert out["specified_keys"] == 1 + sol = next(t for t in out["access_targets"] if t["type"] == "serial_console") + assert sol["key_access_enabled"] is True + assert sol["reachable"] is True + + +def test_query_key_access_skips_when_no_key_groups( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """No key groups anywhere yields a structured skip with org_key_groups == 0.""" + module = _load_query_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "forge_get", lambda *a, **k: {"isSerialConsoleEnabled": True}) + monkeypatch.setattr(module, "forge_get_all", lambda *a, **k: []) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + ["query_key_access.py", "--org", "o", "--site-id", "site-1", "--api-base", "http://x"], + ) + + assert code == 0 + assert out["success"] is True + assert out["skipped"] is True + assert out["org_key_groups"] == 0 + assert "No SSH key groups exist" in out["skip_reason"] + + +def test_query_key_access_skip_distinguishes_unsynced_groups( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Groups that exist but are not synced to the site yield a distinct skip reason.""" + module = _load_query_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "forge_get", lambda *a, **k: {"isSerialConsoleEnabled": True}) + + def fake_all(org: str, path: str, token: str, *, base_url: str, params: dict | None = None, **kw: Any) -> list: + # Site-filtered query finds nothing; the org-wide query finds one group. + if params and "siteId" in params: + return [] + return [{"status": "Syncing", "sshKeys": [{"id": "k"}], "siteAssociations": []}] + + monkeypatch.setattr(module, "forge_get_all", fake_all) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + ["query_key_access.py", "--org", "o", "--site-id", "site-1", "--api-base", "http://x"], + ) + + assert code == 0 + assert out["skipped"] is True + assert out["org_key_groups"] == 1 + assert "none are synced to this site" in out["skip_reason"] + + +def test_setup_key_access_provisions_and_records_restore( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Setup creates the key + synced group and records the site flag to restore.""" + module = _load_setup_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "_generate_public_key", lambda comment: "ssh-ed25519 AAAAtest") + + def fake_post(org: str, path: str, token: str, *, base_url: str, body: dict, **kw: Any) -> dict: + return {"id": "key-1"} if path == "sshkey" else {"id": "kg-1"} + + def fake_get(org: str, path: str, token: str, **kw: Any) -> dict: + if path.startswith("sshkeygroup/"): + return {"status": "Synced", "siteAssociations": [{"site": {"id": "site-1"}, "status": "Synced"}]} + return {"isSerialConsoleSSHKeysEnabled": False} + + patched: dict[str, Any] = {} + + def fake_patch(org: str, path: str, token: str, *, base_url: str, body: dict, **kw: Any) -> dict: + patched.update(body) + return {} + + monkeypatch.setattr(module, "forge_post", fake_post) + monkeypatch.setattr(module, "forge_get", fake_get) + monkeypatch.setattr(module, "forge_patch", fake_patch) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + ["setup_key_access.py", "--org", "o", "--site-id", "site-1", "--api-base", "http://x"], + ) + + assert code == 0 + assert out["success"] is True + assert out["sshkey_id"] == "key-1" + assert out["sshkeygroup_id"] == "kg-1" + assert out["synced"] is True + assert out["restore_ssh_keys_enabled"] is False + assert patched == {"isSerialConsoleSSHKeysEnabled": True} + + +def test_setup_key_access_fails_when_group_does_not_sync( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Setup fails when the key group never syncs, but still emits IDs for teardown.""" + module = _load_setup_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "_generate_public_key", lambda comment: "ssh-ed25519 AAAAtest") + monkeypatch.setattr(module, "_wait_for_sync", lambda *a, **k: False) + monkeypatch.setattr( + module, + "forge_post", + lambda org, path, token, *, base_url, body, **kw: {"id": "key-1"} if path == "sshkey" else {"id": "kg-1"}, + ) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + ["setup_key_access.py", "--org", "o", "--site-id", "site-1", "--api-base", "http://x"], + ) + + assert code == 1 + assert out["success"] is False + assert out["synced"] is False + assert out["sshkey_id"] == "key-1" + assert out["sshkeygroup_id"] == "kg-1" + assert "did not sync" in out["error"] + + +def test_setup_key_access_emits_key_id_on_group_failure( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A created key id is emitted even when the group create fails, so teardown can clean up.""" + module = _load_setup_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr(module, "_generate_public_key", lambda comment: "ssh-ed25519 AAAAtest") + + def fake_post(org: str, path: str, token: str, *, base_url: str, body: dict, **kw: Any) -> dict: + if path == "sshkey": + return {"id": "key-1"} + raise RuntimeError("group create failed") + + monkeypatch.setattr(module, "forge_post", fake_post) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + ["setup_key_access.py", "--org", "o", "--site-id", "site-1", "--api-base", "http://x"], + ) + + assert code == 1 + assert out["success"] is False + assert out["sshkey_id"] == "key-1" + assert out["sshkeygroup_id"] == "" + + +def test_teardown_key_access_deletes_resources_and_restores_flag( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Teardown deletes the group then the key and restores the site flag.""" + module = _load_teardown_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + deletes: list[str] = [] + patched: dict[str, Any] = {} + monkeypatch.setattr(module, "forge_delete", lambda org, path, token, **kw: deletes.append(path) or {}) + monkeypatch.setattr( + module, "forge_patch", lambda org, path, token, *, base_url, body, **kw: patched.update(body) or {} + ) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + [ + "teardown_key_access.py", + "--org", + "o", + "--site-id", + "site-1", + "--api-base", + "http://x", + "--sshkeygroup-id", + "kg-1", + "--sshkey-id", + "key-1", + "--restore-ssh-keys-enabled", + "false", + ], + ) + + assert code == 0 + assert out["success"] is True + assert deletes == ["sshkeygroup/kg-1", "sshkey/key-1"] + assert patched == {"isSerialConsoleSSHKeysEnabled": False} + assert out["cleanup_errors"] == [] + + +def test_teardown_key_access_treats_404_delete_as_success( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """DELETE 404 means the resource is already gone; teardown should still succeed.""" + module = _load_teardown_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + monkeypatch.setattr( + module, + "forge_delete", + lambda *a, **k: (_ for _ in ()).throw(HTTPError("http://x", 404, "Not Found", None, None)), + ) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + [ + "teardown_key_access.py", + "--org", + "o", + "--site-id", + "site-1", + "--api-base", + "http://x", + "--sshkeygroup-id", + "kg-1", + "--sshkey-id", + "key-1", + "--restore-ssh-keys-enabled", + "", + ], + ) + + assert code == 0 + assert out["success"] is True + assert out["cleanup_errors"] == [] + + +def test_teardown_key_access_treats_none_restore_flag_as_unset( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Jinja renders Python None as the literal string 'None'; do not PATCH the site.""" + module = _load_teardown_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + patches: list[dict[str, Any]] = [] + monkeypatch.setattr(module, "forge_delete", lambda *a, **k: {}) + monkeypatch.setattr( + module, + "forge_patch", + lambda org, path, token, *, base_url, body, **kw: patches.append(body) or {}, + ) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + [ + "teardown_key_access.py", + "--org", + "o", + "--site-id", + "site-1", + "--api-base", + "http://x", + "--sshkeygroup-id", + "kg-1", + "--sshkey-id", + "key-1", + "--restore-ssh-keys-enabled", + "None", + ], + ) + + assert code == 0 + assert out["success"] is True + assert patches == [] + + +def test_teardown_key_access_is_noop_without_ids( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Empty IDs (setup created nothing) make teardown a no-op.""" + module = _load_teardown_key_access_script() + monkeypatch.setattr(module, "resolve_auth", lambda: SimpleNamespace(token="t")) + calls: list[str] = [] + monkeypatch.setattr(module, "forge_delete", lambda *a, **k: calls.append("delete") or {}) + monkeypatch.setattr(module, "forge_patch", lambda *a, **k: calls.append("patch") or {}) + + code, out = _run_script_main( + module, + monkeypatch, + capsys, + [ + "teardown_key_access.py", + "--org", + "o", + "--site-id", + "site-1", + "--api-base", + "http://x", + "--sshkeygroup-id", + "", + "--sshkey-id", + "", + "--restore-ssh-keys-enabled", + "", + ], + ) + + assert code == 0 + assert out["success"] is True + assert calls == [] diff --git a/isvtest/src/isvtest/validations/key_access.py b/isvtest/src/isvtest/validations/key_access.py new file mode 100644 index 00000000..507cc238 --- /dev/null +++ b/isvtest/src/isvtest/validations/key_access.py @@ -0,0 +1,155 @@ +# 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. + +"""Key-secret-management validations (requirement AUTH-XX). + +``SpecifiedKeyAccessCheck`` (AUTH-XX-03): verify that a tenant-specified key +can be used to access other components "as possible" -- the serial console +(SOL) and network devices being the called-out examples. + +The validation is provider-neutral: a step script reports the number of +specified keys available plus a list of access targets, each carrying a +tri-state ``key_access_enabled`` flag (``true`` = key access is enabled, +``false`` = explicitly disabled, ``null`` = could not be verified) and a +``reachable`` flag (the key has actually propagated and the component endpoint +is present). The check passes when at least one target is reachable with the +key, fails when a target's key-access path is explicitly disabled or broken, +and skips when access can only be left unverified. +""" + +from __future__ import annotations + +from typing import ClassVar + +import pytest + +from isvtest.core.validation import BaseValidation + + +class SpecifiedKeyAccessCheck(BaseValidation): + """Validate specified-key access to out-of-band components (AUTH-XX-03). + + AUTH-XX-03 requires that a tenant-supplied key (e.g. an SSH key) can be used + to reach other components, with the serial console (SOL) and network devices + given as examples. Proving this reduces to checking the end-to-end access + path for each reported target: + + * a specified key exists (``specified_keys >= 1``); without one there is + nothing to evidence access with, so the check skips; + * for each access target, ``key_access_enabled`` records whether key-based + access to that component is enabled, and ``reachable`` records whether the + key has actually propagated and the component endpoint is present. + + A target is "accessible" when key access is enabled *and* it is reachable. + The check passes when at least ``min_accessible_targets`` targets are + accessible. A target whose key access is explicitly disabled, or enabled but + not reachable, is a concrete failure. A target the script could only leave + unverified (``key_access_enabled`` is ``null``, e.g. provider-managed + network-device access) neither passes nor fails -- if no verifiable target + is accessible, the check skips rather than fabricating a pass. + + Config: + step_output: Step output containing the specified-key access evidence. + min_accessible_targets: Minimum number of targets that must be accessible + via the specified key for the check to pass (default: 1). + + Step output (from query_key_access.py): + success: bool + platform: str + site_id: str + specified_keys: int -- distinct tenant-specified keys synced to the site + access_targets: list[dict]: + type: str -- e.g. "serial_console", "network_device" + name: str + key_access_enabled: bool | None -- true / false / null (unverified) + reachable: bool + detail: str + """ + + description: ClassVar[str] = "Check a tenant-specified key can access out-of-band components (SOL, network devices)" + timeout: ClassVar[int] = 120 + + def run(self) -> None: + """Validate that a specified key can access at least one reported component.""" + step_output = self.config.get("step_output", {}) + + if step_output.get("skipped") is True: + pytest.skip(step_output.get("skip_reason") or "Specified-key access validation skipped") + + if not step_output.get("success"): + self.set_failed(f"Specified-key access step failed: {step_output.get('error', 'Unknown error')}") + return + + targets = step_output.get("access_targets") + if not isinstance(targets, list): + self.set_failed("Specified-key access step output is missing the 'access_targets' list") + return + + specified_keys = step_output.get("specified_keys") + if not isinstance(specified_keys, int) or isinstance(specified_keys, bool): + self.set_failed("Specified-key access step output is missing integer 'specified_keys'") + return + + min_accessible = self._parse_positive_int("min_accessible_targets", default=1) + if min_accessible is None: + return + + if specified_keys < 1: + pytest.skip("No tenant-specified key is registered/synced to the site; cannot evidence key-based access") + + accessible = 0 + concrete_failures: list[str] = [] + unverified: list[str] = [] + + for idx, raw_target in enumerate(targets): + target = raw_target if isinstance(raw_target, dict) else {} + label = target.get("name") or target.get("type") or f"target_{idx}" + subtest_name = f"target_{target.get('type') or idx}" + enabled = target.get("key_access_enabled") + reachable = target.get("reachable") is True + detail = target.get("detail") or "" + + if enabled is True and reachable: + accessible += 1 + self.report_subtest( + subtest_name, passed=True, message=f"{label}: accessible via specified key ({detail})" + ) + elif enabled is True: + concrete_failures.append(f"{label} (enabled but not reachable: {detail})") + self.report_subtest( + subtest_name, passed=False, message=f"{label}: key access enabled but not reachable ({detail})" + ) + elif enabled is False: + concrete_failures.append(f"{label} (key access disabled: {detail})") + self.report_subtest(subtest_name, passed=False, message=f"{label}: key access disabled ({detail})") + else: + unverified.append(f"{label} ({detail})" if detail else label) + self.report_subtest(subtest_name, passed=False, skipped=True, message=f"{label}: unverified ({detail})") + + if concrete_failures: + self.set_failed(f"Specified-key access not established: {'; '.join(concrete_failures)}") + return + + if accessible >= min_accessible: + self.set_passed( + f"Specified key ({specified_keys} key(s)) can access " + f"{accessible} of {len(targets)} reported component(s)" + ) + return + + if unverified: + pytest.skip(f"Specified-key access incomplete; could not verify target(s): {', '.join(unverified)}") + + self.set_failed(f"No component is accessible via the specified key (need {min_accessible}, found {accessible})") diff --git a/isvtest/tests/test_key_access.py b/isvtest/tests/test_key_access.py new file mode 100644 index 00000000..d0910879 --- /dev/null +++ b/isvtest/tests/test_key_access.py @@ -0,0 +1,178 @@ +# 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 the specified-key access validation (AUTH-XX-03).""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from isvtest.validations.key_access import SpecifiedKeyAccessCheck + + +def _serial_target( + *, + key_access_enabled: bool | None = True, + reachable: bool = True, + name: str = "sjc-1 serial console (SOL)", + detail: str = "ok", +) -> dict[str, Any]: + """Build a serial-console (SOL) access target.""" + return { + "type": "serial_console", + "name": name, + "key_access_enabled": key_access_enabled, + "reachable": reachable, + "detail": detail, + } + + +def _network_target() -> dict[str, Any]: + """Build the (unverified) network-device access target.""" + return { + "type": "network_device", + "name": "Network devices", + "key_access_enabled": None, + "reachable": False, + "detail": "provider-managed; not verifiable from tenant API", + } + + +def _output( + *, + success: bool = True, + specified_keys: int = 1, + targets: list[dict[str, Any]] | None = None, + error: str = "", +) -> dict[str, Any]: + """Build a specified-key access step output.""" + if targets is None: + targets = [_serial_target(), _network_target()] + return { + "success": success, + "platform": "nico", + "site_id": "test-site-001", + "specified_keys": specified_keys, + "access_targets": targets, + "error": error, + } + + +class TestSpecifiedKeyAccessCheck: + """Tests for SpecifiedKeyAccessCheck (AUTH-XX-03).""" + + def test_serial_console_accessible_passes(self) -> None: + """A key synced to a SOL-enabled, SSH-key-auth site passes.""" + check = SpecifiedKeyAccessCheck(config={"step_output": _output()}) + check.run() + assert check._passed is True, check._error + assert "can access" in check._output + # The network-device target is reported as a skipped subtest, not a failure. + net = next(r for r in check._subtest_results if r["name"] == "target_network_device") + assert net["skipped"] is True + sol = next(r for r in check._subtest_results if r["name"] == "target_serial_console") + assert sol["passed"] is True + + def test_step_failure(self) -> None: + """A failed step is reported with its error detail.""" + check = SpecifiedKeyAccessCheck(config={"step_output": _output(success=False, error="API timeout")}) + check.run() + assert check._passed is False + assert "API timeout" in check._error + + def test_structured_skip(self) -> None: + """A structured skip skips the validation.""" + check = SpecifiedKeyAccessCheck( + config={"step_output": {"success": True, "skipped": True, "skip_reason": "SOL not configured"}} + ) + with pytest.raises(pytest.skip.Exception, match="SOL not configured"): + check.run() + + def test_no_specified_keys_skips(self) -> None: + """Without any specified key there is nothing to evidence access with.""" + check = SpecifiedKeyAccessCheck(config={"step_output": _output(specified_keys=0)}) + with pytest.raises(pytest.skip.Exception, match="No tenant-specified key"): + check.run() + + def test_key_access_disabled_fails(self) -> None: + """A component whose key access is explicitly disabled fails.""" + targets = [_serial_target(key_access_enabled=False, reachable=False, detail="SSH-key access disabled")] + check = SpecifiedKeyAccessCheck(config={"step_output": _output(targets=targets)}) + check.run() + assert check._passed is False + assert "key access disabled" in check._error + + def test_enabled_but_unreachable_fails(self) -> None: + """Key access enabled but the key has not propagated is a broken path.""" + targets = [_serial_target(key_access_enabled=True, reachable=False, detail="key not synced")] + check = SpecifiedKeyAccessCheck(config={"step_output": _output(targets=targets)}) + check.run() + assert check._passed is False + assert "not reachable" in check._error + + def test_string_reachable_false_does_not_count_as_accessible(self) -> None: + """Malformed provider output must not coerce the string 'false' to reachable.""" + targets = [ + { + "type": "serial_console", + "name": "sjc-1 serial console (SOL)", + "key_access_enabled": True, + "reachable": "false", + "detail": "bad type", + } + ] + check = SpecifiedKeyAccessCheck(config={"step_output": _output(targets=targets)}) + check.run() + assert check._passed is False + assert "not reachable" in check._error + + def test_only_unverified_targets_skips(self) -> None: + """When the only targets are unverifiable, the check skips.""" + check = SpecifiedKeyAccessCheck(config={"step_output": _output(targets=[_network_target()])}) + with pytest.raises(pytest.skip.Exception, match="could not verify"): + check.run() + + def test_missing_targets_list_fails(self) -> None: + """A non-list access_targets field fails.""" + output = _output() + output["access_targets"] = None + check = SpecifiedKeyAccessCheck(config={"step_output": output}) + check.run() + assert check._passed is False + assert "access_targets" in check._error + + def test_missing_specified_keys_fails(self) -> None: + """A missing/non-int specified_keys field fails.""" + output = _output() + output["specified_keys"] = "1" + check = SpecifiedKeyAccessCheck(config={"step_output": output}) + check.run() + assert check._passed is False + assert "specified_keys" in check._error + + def test_min_accessible_targets_enforced(self) -> None: + """Requiring more accessible targets than available skips (network unverified).""" + check = SpecifiedKeyAccessCheck(config={"step_output": _output(), "min_accessible_targets": 2}) + with pytest.raises(pytest.skip.Exception, match="could not verify"): + check.run() + + def test_invalid_min_accessible_targets_fails(self) -> None: + """A non-integer min_accessible_targets is rejected.""" + check = SpecifiedKeyAccessCheck(config={"step_output": _output(), "min_accessible_targets": "two"}) + check.run() + assert check._passed is False + assert "min_accessible_targets" in check._error