ClickHouse/tests/ci/runner_token_rotation_lambda/app.py

93 lines
2.7 KiB
Python
Raw Normal View History

2021-09-28 11:52:51 +00:00
#!/usr/bin/env python3
import argparse
import sys
2022-11-28 16:31:22 +00:00
import boto3 # type: ignore
import requests # type: ignore
from lambda_shared.token import get_cached_access_token, get_access_token_by_key_app
2021-09-28 11:52:51 +00:00
def get_runner_registration_token(access_token):
headers = {
"Authorization": f"token {access_token}",
"Accept": "application/vnd.github.v3+json",
}
response = requests.post(
"https://api.github.com/orgs/ClickHouse/actions/runners/registration-token",
headers=headers,
)
2021-09-28 11:52:51 +00:00
response.raise_for_status()
data = response.json()
return data["token"]
2021-09-28 11:52:51 +00:00
def main(access_token, push_to_ssm, ssm_parameter_name):
2021-09-28 11:52:51 +00:00
runner_registration_token = get_runner_registration_token(access_token)
if push_to_ssm:
print("Trying to put params into ssm manager")
client = boto3.client("ssm")
2021-09-28 11:52:51 +00:00
client.put_parameter(
Name=ssm_parameter_name,
Value=runner_registration_token,
Type="SecureString",
Overwrite=True,
)
2021-09-28 11:52:51 +00:00
else:
print(
"Not push token to AWS Parameter Store, just print:",
runner_registration_token,
)
2021-09-28 11:52:51 +00:00
def handler(event, context):
main(get_cached_access_token(), True, "github_runner_registration_token")
2021-09-28 11:52:51 +00:00
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Get new token from github to add runners"
)
parser.add_argument(
"-p", "--private-key-path", help="Path to file with private key"
)
parser.add_argument("-k", "--private-key", help="Private key")
parser.add_argument(
"-a", "--app-id", type=int, help="GitHub application ID", required=True
)
parser.add_argument(
"--push-to-ssm",
action="store_true",
help="Store received token in parameter store",
)
parser.add_argument(
"--ssm-parameter-name",
default="github_runner_registration_token",
help="AWS paramater store parameter name",
)
2021-09-28 11:52:51 +00:00
args = parser.parse_args()
if not args.private_key_path and not args.private_key:
print(
"Either --private-key-path or --private-key must be specified",
file=sys.stderr,
)
2021-09-28 11:52:51 +00:00
if args.private_key_path and args.private_key:
print(
"Either --private-key-path or --private-key must be specified",
file=sys.stderr,
)
2021-09-28 11:52:51 +00:00
if args.private_key:
private_key = args.private_key
else:
with open(args.private_key_path, "r") as key_file:
2021-09-28 11:52:51 +00:00
private_key = key_file.read()
token = get_access_token_by_key_app(private_key, args.app_id)
main(token, args.push_to_ssm, args.ssm_parameter_name)