diff --git a/isvctl/configs/providers/nico/config/bare_metal.yaml b/isvctl/configs/providers/nico/config/bare_metal.yaml index 182fa6a9..6bb0ceb9 100644 --- a/isvctl/configs/providers/nico/config/bare_metal.yaml +++ b/isvctl/configs/providers/nico/config/bare_metal.yaml @@ -52,6 +52,14 @@ # 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 from the NICo API and reports whether a +# tenant-specified key can access out-of-band components (serial console / +# SOL is verifiable; network devices are provider-managed). The +# setup_key_access.py and teardown_key_access.py steps provision a +# throwaway key group for the run, then remove it, so +# SpecifiedKeyAccessCheck (AUTH-XX-03) can validate the path without +# persistent manual key setup. # # Prerequisites: # - NICO_API_BASE set to the NICo API base URL @@ -59,6 +67,7 @@ # NICO_SSA_ISSUER, NICO_CLIENT_ID, and NICO_CLIENT_SECRET # - Expected machines pre-registered; machines ingested and DPUs initialized # - NICO_ORGANIZATION and NICO_SITE_ID environment variables set +# - ssh-keygen available on the host (used to mint the throwaway key) # # Usage: # NICO_BEARER_TOKEN= NICO_ORGANIZATION= NICO_SITE_ID= \ @@ -71,8 +80,26 @@ version: "1.0" commands: bare_metal: - phases: ["test"] + phases: ["setup", "test", "teardown"] steps: + # ─── Provision a Throwaway Specified Key (AUTH-XX-03 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" + requires_available_validations: + - SpecifiedKeyAccessCheck + args: + - "--org" + - "{{org}}" + - "--site-id" + - "{{site_id}}" + - "--api-base" + - "{{nico_api_base}}" + timeout: 240 + # ─── Verify Hardware Ingestion ───────────────────────────────── # Compares expected-machine manifest against discovered machines. # Output includes per-machine status, health, GPU/DPU counts. @@ -223,6 +250,50 @@ 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 + command: "python ../scripts/auth/query_key_access.py" + requires_available_validations: + - SpecifiedKeyAccessCheck + args: + - "--org" + - "{{org}}" + - "--site-id" + - "{{site_id}}" + - "--api-base" + - "{{nico_api_base}}" + timeout: 120 + + # ─── Remove the Throwaway Specified Key (AUTH-XX-03 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" + requires_available_validations: + - SpecifiedKeyAccessCheck + 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-bare-metal-validation" description: "Verify NICo bare-metal hardware ingestion and DPU health" 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..98c8117f --- /dev/null +++ b/isvctl/configs/providers/nico/scripts/auth/query_key_access.py @@ -0,0 +1,282 @@ +#!/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, + "key_groups": 1, + "keys_synced_to_site": true, + "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 + +# SSH Key Group / site-association status that means the key has fully +# propagated to the site and is therefore usable for access. +SYNCED_STATUS = "Synced" + + +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. + + A key is usable on the site once its group's association reaches ``Synced``. + The group is accepted when either its own status is ``Synced`` or the + association for this specific site is ``Synced``. + """ + for group in groups: + if not (group.get("sshKeys") or []): + continue + if group.get("status") == SYNCED_STATUS: + return True + for assoc in group.get("siteAssociations") or []: + if not isinstance(assoc, dict): + continue + assoc_site = (assoc.get("site") or {}).get("id") + if assoc_site == site_id and assoc.get("status") == SYNCED_STATUS: + return True + return False + + +def _serial_console_target(site: dict[str, Any], keys_synced: bool) -> dict[str, Any]: + """Build the serial-console (SOL) access target from the site config.""" + 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() + reachable = bool(hostname) and keys_synced + + 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" + elif not keys_synced: + detail = "Serial console SSH-key access is enabled but no specified key is synced to this site" + 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": reachable, + "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, + "key_groups": 0, + "keys_synced_to_site": False, + "access_targets": [], + } + + try: + auth = resolve_auth() + + site = forge_get(args.org, f"site/{args.site_id}", auth.token, base_url=args.api_base) + + groups = forge_get_all( + args.org, + "sshkeygroup", + auth.token, + base_url=args.api_base, + params={"siteId": args.site_id}, + result_key="sshKeyGroups", + ) + + keys_synced = _keys_synced_to_site(groups, args.site_id) + + result["specified_keys"] = _count_specified_keys(groups) + result["key_groups"] = len(groups) + result["keys_synced_to_site"] = keys_synced + + if not keys_synced: + # 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" + ) + result["success"] = True + print(json.dumps(result, indent=2)) + return 0 + + result["access_targets"] = [ + _serial_console_target(site, keys_synced), + _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..0cd57746 --- /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 + +SYNCED_STATUS = "Synced" +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 _group_synced_to_site(group: dict[str, Any], site_id: str) -> bool: + """Return whether the key group has reached Synced for the site.""" + 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 _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.""" + deadline = time.monotonic() + SYNC_POLL_TIMEOUT_SECONDS + while True: + group = forge_get(org, f"sshkeygroup/{group_id}", token, base_url=base_url) + if _group_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 + ) + + # 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. + result["restore_ssh_keys_enabled"] = None + + 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..26276edb --- /dev/null +++ b/isvctl/configs/providers/nico/scripts/auth/teardown_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. + +"""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. Cleanup is best-effort: a failure on one resource +does not stop the others, and errors are reported in ``cleanup_errors``. + +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 + +# 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 _as_bool(value: str) -> bool | None: + """Parse a Jinja-rendered flag into True/False, or None when unset.""" + text = (value or "").strip().lower() + if text in ("true", "1", "yes", "on"): + return True + if text in ("false", "0", "no", "off"): + return False + return None + + +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: a key cannot be removed while it is + # still attached to a group. + if group_id: + try: + forge_delete(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: + forge_delete(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}") + + # Best-effort: teardown succeeds as long as it ran; individual failures + # are surfaced in cleanup_errors without blocking the others. + result["success"] = True + + 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 b78017b6..423cd120 100644 --- a/isvctl/configs/providers/nico/scripts/common/nico_client.py +++ b/isvctl/configs/providers/nico/scripts/common/nico_client.py @@ -206,6 +206,71 @@ def forge_get( raise type(e)(e.url, e.code, f"{e.reason}: {body}", e.headers, None) from e +def forge_request( + org: str, + path: str, + token: str, + *, + base_url: str, + method: str, + body: dict[str, Any] | None = None, + timeout: int = 30, +) -> dict[str, Any]: + """Make an authenticated write request (POST/PATCH/DELETE) to the NICo API. + + Args: + org: NGC org name. + path: API path relative to /carbide/ (e.g., "sshkey", "sshkeygroup/{id}"). + token: Bearer token. + base_url: NICo API base URL. + method: HTTP method ("POST", "PATCH", "DELETE"). + body: Optional JSON request body. + timeout: Request timeout in seconds. + + Returns: + Parsed JSON response, or ``{}`` when the response body is empty (e.g. a + 202/204 from DELETE). + + Raises: + HTTPError: On non-2xx response. + """ + url = f"{base_url}/{org}/carbide/{path}" + 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, data=data, headers=headers, method=method) + try: + with urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode() + return json.loads(raw) if raw.strip() else {} + except HTTPError as e: + err_body = "" + if e.fp: + 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 (``/carbide/`` segment).""" + 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 (``/carbide/`` segment).""" + 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 (``/carbide/`` segment).""" + return forge_request(org, path, token, base_url=base_url, method="DELETE", timeout=timeout) + + def forge_get_all( org: str, path: str, diff --git a/isvctl/configs/suites/bare_metal.yaml b/isvctl/configs/suites/bare_metal.yaml index e08ff942..510f9a15 100644 --- a/isvctl/configs/suites/bare_metal.yaml +++ b/isvctl/configs/suites/bare_metal.yaml @@ -442,5 +442,17 @@ tests: checks: FirmwareAttestationCheck: {} + # 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: + min_accessible_targets: 1 + exclude: labels: [] diff --git a/isvctl/tests/providers/nico/test_nico_provider.py b/isvctl/tests/providers/nico/test_nico_provider.py index 084b3b66..c6e1da99 100644 --- a/isvctl/tests/providers/nico/test_nico_provider.py +++ b/isvctl/tests/providers/nico/test_nico_provider.py @@ -202,6 +202,39 @@ def _load_sanitization_script() -> ModuleType: return module +def _load_query_key_access_script() -> ModuleType: + """Load the query_key_access script as a module for direct unit testing.""" + script_path = NICO_SCRIPTS / "auth" / "query_key_access.py" + spec = importlib.util.spec_from_file_location("test_query_key_access", script_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + with _isolated_common_imports(): + spec.loader.exec_module(module) + return module + + +def _load_setup_key_access_script() -> ModuleType: + """Load the setup_key_access script as a module for direct unit testing.""" + script_path = NICO_SCRIPTS / "auth" / "setup_key_access.py" + spec = importlib.util.spec_from_file_location("test_setup_key_access", script_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + with _isolated_common_imports(): + spec.loader.exec_module(module) + return module + + +def _load_teardown_key_access_script() -> ModuleType: + """Load the teardown_key_access script as a module for direct unit testing.""" + script_path = NICO_SCRIPTS / "auth" / "teardown_key_access.py" + spec = importlib.util.spec_from_file_location("test_teardown_key_access", script_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + with _isolated_common_imports(): + spec.loader.exec_module(module) + return module + + 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() @@ -306,6 +339,7 @@ def fake_forge_get(org, path, token, *, base_url, params=None, timeout=30): @pytest.mark.parametrize( "step_name", [ + "setup_key_access", "verify_ingestion", "check_dpu_health", "query_governance_metrics", @@ -315,6 +349,8 @@ def fake_forge_get(org, path, token, *, base_url, params=None, timeout=30): "query_ib_tenant_isolation", "query_ib_keys", "query_sanitization", + "query_key_access", + "teardown_key_access", ], ) def test_nico_bare_metal_config_exposes_api_base_setting(step_name: str) -> None: @@ -329,6 +365,15 @@ def test_nico_bare_metal_config_exposes_api_base_setting(step_name: str) -> None assert "{{nico_api_base}}" in step["args"] +def test_nico_bare_metal_key_access_steps_are_gated() -> None: + """AUTH-XX-03's key-mutating steps should only run when its check is available.""" + merged = merge_yaml_files([NICO_CONFIG / "bare_metal.yaml"]) + steps = {s["name"]: s for s in merged["commands"]["bare_metal"]["steps"]} + + for step_name in ("setup_key_access", "query_key_access", "teardown_key_access"): + assert steps[step_name]["requires_available_validations"] == ["SpecifiedKeyAccessCheck"] + + @pytest.mark.parametrize( ("script_name", "load_script"), [ @@ -341,6 +386,7 @@ def test_nico_bare_metal_config_exposes_api_base_setting(step_name: str) -> None ("query_ib_tenant_isolation.py", _load_ib_tenant_isolation_script), ("query_ib_keys.py", _load_ib_keys_script), ("query_sanitization.py", _load_sanitization_script), + ("query_key_access.py", _load_query_key_access_script), ], ) def test_nico_scripts_require_api_base( @@ -1800,3 +1846,262 @@ def test_sanitization_script_output_satisfies_disk_check( assert "1/1 machine(s)" in bad._error sub = next(r for r in bad._subtest_results if r["name"] == "disk_m-1") assert "without sanitization" in sub["message"] + + +# =========================================================================== +# 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["keys_synced_to_site"] is True + 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_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_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..740cbd21 --- /dev/null +++ b/isvtest/src/isvtest/validations/key_access.py @@ -0,0 +1,158 @@ +# 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 + key_groups: int + keys_synced_to_site: bool + 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 + labels: ClassVar[tuple[str, ...]] = ("bare_metal", "security", "iam") + + 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 = bool(target.get("reachable")) + 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..df1d28ef --- /dev/null +++ b/isvtest/tests/test_key_access.py @@ -0,0 +1,164 @@ +# 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, + "key_groups": 1, + "keys_synced_to_site": True, + "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_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