Created
October 19, 2022 08:42
-
-
Save ivyxjc/51a26521dd4abcda2be18f8e96ead20c to your computer and use it in GitHub Desktop.
Github Rest
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from base64 import b64encode | |
from nacl import encoding, public | |
import requests | |
env = "" | |
secret_map = {"AWS_ACCESS_KEY": "", "AWS_ACCESS_SECRET": ""} | |
token = "" | |
owner = "" | |
repos = ["",""] | |
headers = { | |
"Accept": "application/vnd.github+json", | |
"Authorization": f"Bearer {token}" | |
} | |
class NotFoundException(Exception): | |
pass | |
def get_repo_public_Key(repo_id, environment): | |
resp = requests.get(f"https://api.github.com/repositories/{repo_id}/environments/{environment}/secrets/public-key", | |
headers=headers) | |
if resp.status_code != 200: | |
raise Exception("fail to get repo public key") | |
map = resp.json() | |
return map['key_id'], map['key'] | |
def get_repo_id(owner, repo): | |
resp = requests.get(f"https://api.github.com/repos/{owner}/{repo}", headers=headers) | |
if resp.status_code != 200: | |
raise Exception("fail to get repo id") | |
map = resp.json() | |
return map['id'] | |
def encrypt(public_key: str, secret_value: str) -> str: | |
"""Encrypt a Unicode string using the public key.""" | |
public_key = public.PublicKey(public_key.encode("utf-8"), encoding.Base64Encoder()) | |
sealed_box = public.SealedBox(public_key) | |
encrypted = sealed_box.encrypt(secret_value.encode("utf-8")) | |
return b64encode(encrypted).decode("utf-8") | |
def check_repo_environment_secret_exists(repo_id, environment, secret_name): | |
resp = requests.get( | |
f"https://api.github.com/repositories/{repo_id}/environments/{environment}/secrets/{secret_name}", | |
headers=headers) | |
if resp.status_code != 200: | |
raise NotFoundException(f"Specific secret name does not exit,statu code: {resp.status_code}, body: {resp.text}") | |
def update_repo_environment_secret(repo_id, environment, secret_name, secret_value, key_id): | |
payload = {'encrypted_value': secret_value, 'key_id': key_id} | |
resp = requests.put( | |
f"https://api.github.com/repositories/{repo_id}/environments/{environment}/secrets/{secret_name}", | |
headers=headers, | |
json=payload) | |
if resp.status_code == 201: | |
print("[WARN] you create one secret") | |
if resp.status_code != 201 and resp.status_code != 204: | |
print(resp.text) | |
raise Exception(f"fail to update repo environment secret. status code: {resp.status_code}, resp: {resp.text}") | |
for r in repos: | |
print("start to handle repo: ", r) | |
repo_id = get_repo_id(owner, r) | |
key_id, key = get_repo_public_Key(repo_id, env) | |
for (secret_name, secret_value) in secret_map.items(): | |
try: | |
check_repo_environment_secret_exists(get_repo_id(owner, r), env, secret_name) | |
update_repo_environment_secret(repo_id, env, secret_name, | |
encrypt(key, secret_value), | |
key_id) | |
print(f"success update {secret_name}") | |
except NotFoundException as e: | |
print(f"secret name {secret_name} not found") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment