SSH を使用するようにアプリを構成する

---

このドキュメントでは、SSH と OS Login を使用して 2 つの仮想マシン（VM）インスタンスをプログラムで接続するアプリを構成する方法について説明します。アプリが SSH を使用できるようになると、システム管理プロセスの自動化に役立ちます。

このガイドで使用するすべてのコードサンプルは、GitHub の GoogleCloudPlatform/python-docs-samples ページに置かれています。

準備

• サービス アカウント用に SSH を設定します。
• プロジェクトか、サービス アカウントとして実行する VM で OS Login を設定します。
• まだ設定していない場合は、認証を設定します。認証とは、Google Cloud サービスと API にアクセスするために ID を確認するプロセスです。ローカル開発環境からコードまたはサンプルを実行するには、次のように Compute Engine に対する認証を行います。
  

  Select the tab for how you plan to use the samples on this page:

  Console

  When you use the Google Cloud console to access Google Cloud services and APIs, you don't need to set up authentication.

  gcloud

  1. Install the Google Cloud CLI, then initialize it by running the following command:

     gcloud init

     Note: If you installed the gcloud CLI previously, make sure you have the latest version by running gcloud components update.
  2. Set a default region and zone.

SSH アプリを設定する

SSH 認証鍵を管理し、Compute Engine VM への SSH 接続を開始するようにアプリを設定します。大まかには、アプリで次のことを行う必要があります。

1. Google OS Login ライブラリをインポートしてクライアント ライブラリをビルドし、OS Login API で認証できるようにします。
2. アプリで OS Login を使用できるようにするには、OS Login クライアント オブジェクトを初期化します。
3. VM のサービス アカウントの SSH 認証鍵を生成し、サービス アカウントに公開鍵を追加する create_ssh_key() メソッドを実装します。
4. サービス アカウントが使用する POSIX ユーザー名を取得するために、OS Login ライブラリから get_login_profile() メソッドを呼び出します。
5. リモート SSH コマンドを実行する run_ssh() メソッドを実装します。
6. 一時 SSH 認証鍵ファイルを削除します。

SSH アプリのサンプル

oslogin_service_account_ssh.py サンプルアプリでは、SSH アプリの実装の一例を示します。この例では、アプリが run_ssh() メソッドを使用してリモート インスタンス上でコマンドを実行し、コマンド出力を返します。

"""
Example of using the OS Login API to apply public SSH keys for a service
account, and use that service account to run commands on a remote
instance over SSH. This example uses zonal DNS names to address instances
on the same internal VPC network.
"""
from __future__ import annotations

import argparse
import subprocess
import time
from typing import Optional
import uuid

from google.cloud import oslogin_v1
import requests

SERVICE_ACCOUNT_METADATA_URL = (
    "http://metadata.google.internal/computeMetadata/v1/instance/"
    "service-accounts/default/email"
)
HEADERS = {"Metadata-Flavor": "Google"}


def execute(
    cmd: list[str],
    cwd: Optional[str] = None,
    capture_output: bool = False,
    env: Optional[dict] = None,
    raise_errors: bool = True,
) -> tuple[int, str]:
    """
    Run an external command (wrapper for Python subprocess).

    Args:
        cmd: The command to be run.
        cwd: Directory in which to run the command.
        capture_output: Should the command output be captured and returned or just ignored.
        env: Environmental variables passed to the child process.
        raise_errors: Should errors in run command raise exceptions.

    Returns:
        Return code and captured output.
    """
    print(f"Running command: {cmd}")
    process = subprocess.run(
        cmd,
        cwd=cwd,
        stdout=subprocess.PIPE if capture_output else subprocess.DEVNULL,
        stderr=subprocess.STDOUT,
        text=True,
        env=env,
        check=raise_errors,
    )
    output = process.stdout
    returncode = process.returncode

    if returncode:
        print(f"Command returned error status {returncode}")
        if capture_output:
            print(f"With output: {output}")

    return returncode, output


def create_ssh_key(
    oslogin_client: oslogin_v1.OsLoginServiceClient,
    account: str,
    expire_time: int = 300,
) -> str:
    """
    Generates a temporary SSH key pair and apply it to the specified account.

    Args:
        oslogin_client: OS Login client object.
        account: Name of the service account this key will be assigned to.
            This should be in form of `user/<service_account_username>`.
        expire_time: How many seconds from now should this key be valid.

    Returns:
        The path to private SSH key. Public key can be found by appending `.pub`
        to the file name.

    """
    private_key_file = f"/tmp/key-{uuid.uuid4()}"
    execute(["ssh-keygen", "-t", "rsa", "-N", "", "-f", private_key_file])

    with open(f"{private_key_file}.pub") as original:
        public_key = original.read().strip()

    # Expiration time is in microseconds.
    expiration = int((time.time() + expire_time) * 1000000)

    request = oslogin_v1.ImportSshPublicKeyRequest()
    request.parent = account
    request.ssh_public_key.key = public_key
    request.ssh_public_key.expiration_time_usec = expiration

    print(f"Setting key for {account}...")
    oslogin_client.import_ssh_public_key(request)

    # Let the key properly propagate
    time.sleep(5)

    return private_key_file


def run_ssh(cmd: str, private_key_file: str, username: str, hostname: str) -> str:
    """
    Runs a command on a remote system.

    Args:
        cmd: command to be run.
        private_key_file: private SSH key to be used for authentication.
        username: username to be used for authentication.
        hostname: hostname of the machine you want to run the command on.

    Returns:
        Output of the executed command.
    """
    ssh_command = [
        "ssh",
        "-i",
        private_key_file,
        "-o",
        "StrictHostKeyChecking=no",
        "-o",
        "UserKnownHostsFile=/dev/null",
        f"{username}@{hostname}",
        cmd,
    ]
    print(f"Running ssh command: {' '.join(ssh_command)}")
    tries = 0
    while tries < 3:
        try:
            ssh = subprocess.run(
                ssh_command,
                shell=False,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                check=True,
                env={"SSH_AUTH_SOCK": ""},
                timeout=10,
            )
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as err:
            time.sleep(10)
            tries += 1
            if tries == 3:
                if isinstance(err, subprocess.CalledProcessError):
                    print(
                        f"Failed to run SSH command (return code: {err.returncode}. Output received: {err.output}"
                    )
                else:
                    print("Failed to run SSH - timed out.")
                raise err
        else:
            return ssh.stdout


def main(
    cmd: str,
    project: str,
    instance: Optional[str] = None,
    zone: Optional[str] = None,
    account: Optional[str] = None,
    hostname: Optional[str] = None,
    oslogin: Optional[oslogin_v1.OsLoginServiceClient] = None,
) -> str:
    """
    Runs a command on a remote system.

    Args:
        cmd: command to be executed on the remote host.
        project: name of the project in which te remote instance is hosted.
        instance: name of the remote system instance.
        zone: zone in which the remote system resides. I.e. us-west3-a
        account: account to be used for authentication.
        hostname: hostname of the remote system.
        oslogin: OSLogin service client object. If not provided, a new client will be created.

    Returns:
        The commands output.
    """
    # Create the OS Login API object.
    if oslogin is None:
        oslogin = oslogin_v1.OsLoginServiceClient()

    # Identify the service account ID if it is not already provided.
    account = (
        account or requests.get(SERVICE_ACCOUNT_METADATA_URL, headers=HEADERS).text
    )

    if not account.startswith("users/"):
        account = f"users/{account}"

    # Create a new SSH key pair and associate it with the service account.
    private_key_file = create_ssh_key(oslogin, account)
    try:
        # Using the OS Login API, get the POSIX username from the login profile
        # for the service account.
        profile = oslogin.get_login_profile(name=account)
        username = profile.posix_accounts[0].username

        # Create the hostname of the target instance using the instance name,
        # the zone where the instance is located, and the project that owns the
        # instance.
        hostname = hostname or f"{instance}.{zone}.c.{project}.internal"

        # Run a command on the remote instance over SSH.
        result = run_ssh(cmd, private_key_file, username, hostname)

        # Print the command line output from the remote instance.
        print(result)
        return result
    finally:
        # Shred the private key and delete the pair.
        execute(["shred", private_key_file])
        execute(["rm", private_key_file])
        execute(["rm", f"{private_key_file}.pub"])


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--cmd", default="uname -a", help="The command to run on the remote instance."
    )
    parser.add_argument("--project", help="Your Google Cloud project ID.")
    parser.add_argument("--zone", help="The zone where the target instance is located.")
    parser.add_argument("--instance", help="The target instance for the ssh command.")
    parser.add_argument("--account", help="The service account email.")
    parser.add_argument(
        "--hostname",
        help="The external IP address or hostname for the target instance.",
    )
    args = parser.parse_args()

    main(
        args.cmd,
        args.project,
        instance=args.instance,
        zone=args.zone,
        account=args.account,
        hostname=args.hostname,
    )

SSH アプリを実行する

SSH を使用するアプリの作成後は、次の例のようなプロセスに沿ってアプリを実行できます。このプロセスでは、oslogin_service_account_ssh.py サンプルアプリをインストールして実行します。インストールするライブラリは、アプリで使用しているプログラミング言語によって異なる場合があります。

別の方法として、oslogin_service_account_ssh.py をインポートして直接実行するアプリを作成することもできます。

1. SSH アプリをホストする VM に接続します。

2. VM で、pip と Python 3 クライアント ライブラリをインストールします。

   sudo apt update && sudo apt install python3-pip -y && pip install --upgrade google-cloud-os-login requests
   

3. 省略可: oslogin_service_account_ssh.py サンプルアプリを使用する場合は、GoogleCloudPlatform/python-docs-samples からダウンロードします。

   curl -O https://raw.githubusercontent.com/GoogleCloudPlatform/python-docs-samples/master/compute/oslogin/oslogin_service_account_ssh.py
   

4. SSH アプリを実行します。サンプルアプリは、argparse を使用してコマンドラインから変数を受け入れます。この例では、プロジェクト内の別の VM に cowsay をインストールして実行するようアプリに指示します。

   python3 service_account_ssh.py \
      --cmd 'sudo apt install cowsay -y && cowsay "It works!"' \
      --project=PROJECT_ID --instance=VM_NAME --zone=ZONE
   

   次のように置き換えます。

   • PROJECT_ID: アプリが接続する VM のプロジェクト ID。
   • VM_NAME: アプリが接続する VM の名前。
   • ZONE: アプリが接続する VM のゾーン。

   出力は次のようになります。

   ⋮
   ___________
    It works!
   -----------
         \   ^__^
          \  (oo)\_______
             (__)\       )\/\
                 ||----w |
                 ||     ||
   

次のステップ

• 完全なコードサンプルをダウンロードして確認する。完全なサンプルには、これらすべてのメソッドを同時に使用する一例が含まれています。ご自由にダウンロードして、ニーズに合わせて変更し、実行してください。
• SSH 認証鍵の構成やストレージなど、Compute Engine での SSH 接続の動作の詳細を確認する。