株式会社カサレアル BSブログ

import argparse
import subprocess
import time
from typing import List, Optional, Tuple
import uuid
import requests
import os


SSH_KEY_PATH = '/tmp/dev-user'



def execute(
    cmd: List[str],
    cwd: Optional[str] = None,
    capture_output: Optional[bool] = False,
    env: Optional[dict] = None,
    raise_errors: Optional[bool] = True
) -> Tuple[int, str]:
    print(f'Executing 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 run_ssh(cmd: str, private_key_file: str, username: str, hostname: str) -> str:
    ssh_command = [
        'ssh',
        '-i', private_key_file,
        '-o', 'StrictHostKeyChecking=no',
        '-o', 'UserKnownHostsFile=/dev/null',
        f'{username}@{hostname}',
        cmd,
    ]
    print(f"Executing 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': ''},
            )
        except subprocess.CalledProcessError as e:
            print('ERROR_output:', e.output)
            print('ERROR_stderr:', e.stderr)
            print('ERROR_stdout:', e.stdout)
            time.sleep(5)
            tries += 1
        else:
            print(ssh.stdout)
            return ssh.stdout



def main(
    request,
) -> str:
    cmd = 'uname -a'
    project = 'xxx-yyyy' #プロジェクト名
    instance = 'instance-centos' # インスタンス名
    zone = 'asia-northeast1-b' # zone名
    username = 'dev-user' # ssh接続ユーザー名
    hostname = f'{instance}.{zone}.c.{project}.internal'
    private_key_file = SSH_KEY_PATH


    # プライベートキーを指定パスに配置
    with open(private_key_file, 'w') as f:
        f.write(os.environ.get('SSH_PRIVATE_KEY', 'Specified environment variable is not set.'))


    # 権限を変更
    os.chmod(private_key_file, 0o400)
    
    # ssh実行
    result = run_ssh(cmd, private_key_file, username, hostname)
    print('result:',result)


    # プライベートキーを削除
    execute(['rm', private_key_file])
    return 'OK'