feat(cloudkms): Add single tenant hsm scripts

kms/singletenanthsm/README.rst

@@ -0,0 +1,114 @@
+Google Cloud Key Management Service Python Samples
+===============================================================================
+
+.. image:: https://gstatic.com/cloudssh/images/open-btn.png
+   :target: https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/GoogleCloudPlatform/python-docs-samples&page=editor&open_in_editor=kms/singletenanthsm/README.rst
+
+
+This directory contains samples for Google Cloud Key Management Service. The `Cloud Key Management Service`_ allows you to create, import, and manage cryptographic keys and perform cryptographic operations in a single centralized cloud service.
+
+
+
+
+.. _Cloud Key Management Service: https://cloud.google.com/kms/docs/
+
+
+
+
+
+Setup
+-------------------------------------------------------------------------------
+
+
+Install Dependencies
+++++++++++++++++++++
+
+#. Clone python-kms and change directory to the sample directory you want to use.
+
+    .. code-block:: bash
+
+        $ git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
+
+#. Install `pip`_ and `virtualenv`_ if you do not already have them. You may want to refer to the `Python Development Environment Setup Guide`_ for Google Cloud Platform for instructions.
+
+   .. _Python Development Environment Setup Guide:
+       https://cloud.google.com/python/setup
+
+#. Create a virtualenv. Samples are compatible with Python 2.7 and 3.4+.
+
+    .. code-block:: bash
+
+        $ virtualenv env
+        $ source env/bin/activate
+
+#. Install the dependencies needed to run the samples.
+
+    .. code-block:: bash
+
+        $ pip install -r requirements.txt
+
+.. _pip: https://pip.pypa.io/
+.. _virtualenv: https://virtualenv.pypa.io/
+
+Samples
+-------------------------------------------------------------------------------
+
+Create a custom gcloud build to access the Single Tenant HSM service. 
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+
+
+This application creates a custom gcloud build to access the single tenant HSM service. The operation can be specified depending if the user
+wants to just generate rsa keys on all connected yubikeys(generate_rsa_keys), just generate the custom gcloud build to access the 
+single-tenant-hsm(build_custom_gcloud), or both generate keys and the custom gcloud build(generate_gcloud_and_keys). Yubikeys will need to be connected
+to run the `generate_rsa_keys` and `generate_gcloud_and_keys` operations.
+
+
+.. code-block:: bash
+
+    $ python3 setup.py
+
+    usage: setup.py [-h] [--operation] {build_custom_gcloud,generate_rsa_keys,generate_gcloud_and_keys}      
+
+    positional arguments:
+      operation  The type of setup operation you want to perform. This  includes build_custom_gcloud','generate_rsa_keys','generate_gcloud_and_keys'.
+
+    optional arguments:
+      -h, --help        show this help message and exit
+
+    # Below is an example of using the setup command to generate rsa private keys and the custom gcloud build:
+
+    $ python3 setup.py --operation=generate_gcloud_and_keys
+
+
+
+
+Approves a Single Tenant HSM Instance Proposal. 
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
+To run this sample:
+
+.. code-block:: bash
+
+    $ python3 approve_proposal.py
+
+    usage: approve_proposal.py [-h] [--proposal_resource PROPOSAL_RESOURCE]
+
+    This application fetches and approves the single tenant HSM instance proposal 
+    specified in the "proposal_resource" field.
+
+    For more information, visit https://cloud.google.com/kms/docs/attest-key.
+
+    positional arguments:
+        --proposal_resource PROPOSAL_RESOURCE
+                        The full name of the single tenant HSM instance proposal that needs to be approved.
+
+    optional arguments:
+        -h, --help            show this help message and exit
+
+    # Below is an example of using the approve script to fetch the challenges, sign the challenges, and send the signed challenges
+    # associated with the proposal 'my_proposal':
+
+    $ python3 approve_proposal.py --proposal_resource=projects/my-project/locations/us-east1/singleTenantHsmInstances/mysthi/proposals/my_proposal
+
+
+
+.. _Google Cloud SDK: https://cloud.google.com/sdk/

kms/singletenanthsm/approve_proposal.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python
+
+# Copyright 2025 Google LLC
+#
+# 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.
+
+import argparse
+import logging
+import json
+import os
+import sys
+from typing import List
+
+import gcloud_commands
+import ykman_utils
+
+
+def parse_challenges_into_files(sthi_output: str) -> List[bytes]:
+    """Parses the STHI output and writes the challenges and public keys to files.
+
+    Args:
+        sthi_output: The output of the STHI command.
+
+    Returns:
+        A list of the unsigned challenges.
+    """
+    logging.basicConfig(level=logging.INFO)
+    logger = logging.getLogger(__name__)
+
+    logger.info("Parsing challenges into files")
+    proposal_json = json.loads(sthi_output, strict=False)
+    challenges = proposal_json["quorumParameters"]["challenges"]
+
+    directory_path = "challenges"
+    if not os.path.exists(directory_path):
+        os.mkdir(directory_path)
+        logger.info(f"Directory '{directory_path}' created.")
+    else:
+        logger.info(f"Directory '{directory_path}' already exists.")
+
+    challenge_count = 0
+    unsigned_challenges = []
+    for challenge in challenges:
+        challenge_count += 1
+        try:
+            with open("challenges/challenge{0}.txt".format(challenge_count), "wb") as f:
+                binary_challenge = ykman_utils.urlsafe_base64_to_binary(challenge["challenge"])
+                f.write(binary_challenge)
+        except FileNotFoundError:
+            print(f"File not found: challenges/challenge{challenge_count}.txt")
+        except Exception as e:
+            print(f"An error occurred: {e}")
+        try:
+            with open("challenges/public_key{0}.pem".format(challenge_count), "w") as f:
+                f.write(challenge["publicKeyPem"].encode("utf-8").decode("unicode_escape"))
+        except FileNotFoundError:
+            print(f"File not found: challenges/public_key{challenge_count}.txt")
+        except Exception as e:
+            print(f"An error occurred: {e}")
+        unsigned_challenges.append(
+            ykman_utils.Challenge(binary_challenge, challenge["publicKeyPem"])
+        )
+
+    return unsigned_challenges
+
+
+def parse_args(args):
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--proposal_resource", type=str, required=True)
+    return parser.parse_args(args)
+
+
+def signed_challenges_to_files(
+    challenge_replies: list[ykman_utils.ChallengeReply],
+) -> None:
+    """Writes the signed challenges and public keys to files.
+
+    Args:
+        challenge_replies: A list of ChallengeReply objects.
+
+    Returns:
+        A list of tuples containing the signed challenge file path and the public
+        key file path.
+    """
+    signed_challenge_files = []
+    challenge_count = 0
+    logging.basicConfig(level=logging.INFO)
+    logger = logging.getLogger(__name__)
+
+    for challenge_reply in challenge_replies:
+        challenge_count += 1
+        directory_path = "signed_challenges"
+        if not os.path.exists(directory_path):
+            os.mkdir(directory_path)
+            logger.info(f"Directory '{directory_path}' created.")
+        else:
+            logger.info(f"Directory '{directory_path}' already exists.")
+        with open(
+            f"signed_challenges/public_key_{challenge_count}.pem", "w"
+        ) as public_key_file:
+
+            # Write public key to file
+            public_key_file.write(challenge_reply.public_key_pem)
+        with open(
+            f"signed_challenges/signed_challenge{challenge_count}.bin", "wb"
+        ) as binary_file:
+
+            # Write signed challenge to file
+            binary_file.write(challenge_reply.signed_challenge)
+            signed_challenge_files.append(
+                (
+                    f"signed_challenges/signed_challenge{challenge_count}.bin",
+                    f"signed_challenges/public_key_{challenge_count}.pem",
+                )
+            )
+    return signed_challenge_files
+
+
+def approve_proposal():
+    """Approves a proposal by fetching challenges, signing them, and sending them back to gcloud."""
+    parser = parse_args(sys.argv[1:])
+
+    # Fetch challenges
+    process = gcloud_commands.fetch_challenges(parser.proposal_resource)
+
+    # Parse challenges into files
+    unsigned_challenges = parse_challenges_into_files(process.stdout)
+
+    # Sign challenges
+    signed_challenges = ykman_utils.sign_challenges(unsigned_challenges)
+
+    # Parse signed challenges into files
+    signed_challenged_files = signed_challenges_to_files(signed_challenges)
+
+    # Return signed challenges to gcloud
+    gcloud_commands.send_signed_challenges(
+        signed_challenged_files, parser.proposal_resource
+    )
+
+
+if __name__ == "__main__":
+    approve_proposal()