Something went wrong. Try again.
Cloudflare Tunnel client with NetBSD support
Something went wrong. Try again.
cloudflared github_message.py
5.3 kB · 171 lines
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172#!/usr/bin/python3"""Create Github Releases Notes with binary checksums from Workers KV"""
import argparseimport loggingimport osimport requests
from github import Github, UnknownObjectException
FORMAT = "%(levelname)s - %(asctime)s: %(message)s"logging.basicConfig(format=FORMAT, level=logging.INFO)
CLOUDFLARED_REPO = os.environ.get("GITHUB_REPO", "cloudflare/cloudflared")GITHUB_CONFLICT_CODE = "already_exists"BASE_KV_URL = 'https://api.cloudflare.com/client/v4/accounts/'
def kv_get_keys(prefix, account, namespace, api_token): """ get the KV keys for a given prefix """ response = requests.get( BASE_KV_URL + account + "/storage/kv/namespaces/" + namespace + "/keys" + "?prefix=" + prefix, headers={ "Content-Type": "application/json", "Authorization": "Bearer " + api_token, }, ) if response.status_code != 200: jsonResponse = response.json() errors = jsonResponse["errors"] if len(errors) > 0: raise Exception("failed to get checksums: {0}", errors[0]) return response.json()["result"]
def kv_get_value(key, account, namespace, api_token): """ get the KV value for a provided key """ response = requests.get( BASE_KV_URL + account + "/storage/kv/namespaces/" + namespace + "/values/" + key, headers={ "Content-Type": "application/json", "Authorization": "Bearer " + api_token, }, ) if response.status_code != 200: jsonResponse = response.json() errors = jsonResponse["errors"] if len(errors) > 0: raise Exception("failed to get checksums: {0}", errors[0]) return response.text
def update_or_add_message(msg, name, sha): """ updates or builds the github version message for each new asset's sha256. Searches the existing message string to update or create. """ new_text = '{0}: {1}\n'.format(name, sha) start = msg.find(name) if (start != -1): end = msg.find("\n", start) if (end != -1): return msg.replace(msg[start:end+1], new_text) back = msg.rfind("```") if (back != -1): return '{0}{1}```'.format(msg[:back], new_text) return '{0} \n### SHA256 Checksums:\n```\n{1}```'.format(msg, new_text)
def get_release(repo, version): """ Get a Github Release matching the version tag. """ try: release = repo.get_release(version) logging.info("Release %s found", version) return release except UnknownObjectException: logging.info("Release %s not found", version)
def parse_args(): """ Parse and validate args """ parser = argparse.ArgumentParser( description="Updates a Github Release with checksums from KV" ) parser.add_argument( "--api-key", default=os.environ.get("API_KEY"), help="Github API key" ) parser.add_argument( "--kv-namespace-id", default=os.environ.get("KV_NAMESPACE"), help="workers KV namespace id" ) parser.add_argument( "--kv-account-id", default=os.environ.get("KV_ACCOUNT"), help="workers KV account id" ) parser.add_argument( "--kv-api-token", default=os.environ.get("KV_API_TOKEN"), help="workers KV API Token" ) parser.add_argument( "--release-version", metavar="version", default=os.environ.get("VERSION"), help="Release version", ) parser.add_argument( "--dry-run", action="store_true", help="Do not modify the release message" )
args = parser.parse_args() is_valid = True if not args.release_version: logging.error("Missing release version") is_valid = False
if not args.api_key: logging.error("Missing API key") is_valid = False
if not args.kv_namespace_id: logging.error("Missing KV namespace id") is_valid = False
if not args.kv_account_id: logging.error("Missing KV account id") is_valid = False
if not args.kv_api_token: logging.error("Missing KV API token") is_valid = False
if is_valid: return args
parser.print_usage() exit(1)
def main(): """ Attempts to update the Github Release message with the github asset's checksums """ try: args = parse_args() client = Github(args.api_key) repo = client.get_repo(CLOUDFLARED_REPO) release = get_release(repo, args.release_version)
msg = ""
prefix = f"update_{args.release_version}_" keys = kv_get_keys(prefix, args.kv_account_id, args.kv_namespace_id, args.kv_api_token) for key in [k["name"] for k in keys]: checksum = kv_get_value( key, args.kv_account_id, args.kv_namespace_id, args.kv_api_token) binary_name = key[len(prefix):] msg = update_or_add_message(msg, binary_name, checksum)
if args.dry_run: logging.info("Skipping release message update because of dry-run") logging.info(f"Github message:\n{msg}") return
# update the release body text release.update_release(args.release_version, msg)
except Exception as e: logging.exception(e) exit(1)
main()