#!/usr/bin/env python3
# This bit has to stay at the very top of the script. It exists to ensure that
# running this script all by itself uses the python virtual environment with
# our dependencies installed. If will create that environment if it doesn't
# exist.
import os
import sys
import subprocess
args = [os.path.join("build", "bin", "python3")] + sys.argv
# Create the build root if it doesn't exist
if not os.path.exists("build"):
import venv
print("Build environment does not exist, creating...", file=sys.stderr)
venv.create("build", with_pip=True)
subprocess.run(["build/bin/pip", "install", "-U", "pip",
"pyhocon", "boto3", "PyYAML"])
print("Re-executing with builder python...", file=sys.stderr)
os.execv(args[0], args)
else:
# If the build root python is not running this script re-execute it with
# that python instead to ensure all of our dependencies exist.
if os.path.join(os.getcwd(), args[0]) != sys.executable:
print("Re-executing with builder python...", file=sys.stderr)
os.execv(args[0], args)
# Below here is the real script
import io
import os
import re
import sys
import glob
import json
import time
import shutil
import logging
import argparse
import textwrap
import subprocess
import urllib.error
from collections import defaultdict
from datetime import datetime, timedelta
from distutils.version import StrictVersion
from urllib.request import Request, urlopen
import yaml
import boto3
import pyhocon
class ColoredFormatter(logging.Formatter):
"""Log formatter that colors output based on level
"""
_colors = {
"red": "31",
"green": "32",
"yellow": "33",
"blue": "34",
"magenta": "35",
"cyan": "36",
"white": "37",
}
def _color_wrap(self, text, color, bold=False):
code = self._colors[color]
if bold:
code = "1;{}".format(code)
return "\033[{}m{}\033[0m".format(code, text)
def format(self, record):
msg = super().format(record)
# Levels: CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET
if record.levelno in {logging.ERROR, logging.CRITICAL}:
return self._color_wrap(msg, "red")
elif record.levelno == logging.WARNING:
return self._color_wrap(msg, "yellow")
else:
return self._color_wrap(msg, "green")
class IdentityBrokerClient:
"""Client for identity broker
Export IDENTITY_BROKER_ENDPOINT to override the default broker endpoint.
Export IDENTITY_BROKER_API_KEY to specify an API key for the broker.
See README_BROKER.md for more information and a spec.
"""
_DEFAULT_ENDPOINT = "https://aws-access.crute.us/api/account"
_DEFAULT_ACCOUNT = "alpine-amis-user"
def __init__(self, endpoint=None, key=None, account=None):
self.endpoint = endpoint or self._DEFAULT_ENDPOINT
self.account = account or self._DEFAULT_ACCOUNT
self.key = key
self._logger = logging.getLogger()
override_endpoint = os.environ.get("IDENTITY_BROKER_ENDPOINT")
if override_endpoint:
self.endpoint = override_endpoint
if not self.key:
self.key = os.environ.get("IDENTITY_BROKER_API_KEY")
if not self.key:
raise Exception("No identity broker key found")
def _get(self, path):
while True: # to handle rate limits
try:
res = urlopen(Request(path, headers={"X-API-Key": self.key}))
except urllib.error.HTTPError as ex:
if ex.headers.get("Location") == "/logout":
raise Exception("Identity broker token is expired")
if ex.status == 429:
self._logger.warning(
"Rate-limited by identity broker, sleeping 30 seconds")
time.sleep(30)
continue
raise ex
if res.status not in {200, 429}:
raise Exception(res.reason)
return json.load(res)
def get_credentials_url(self):
for account in self._get(self.endpoint):
if account["short_name"] == self.account:
return account["credentials_url"]
raise Exception("No account found")
def get_regions(self):
out = {}
for region in self._get(self.get_credentials_url()):
if region["enabled"]:
out[region["name"]] = region["credentials_url"]
return out
def get_credentials(self, region):
return self._get(self.get_regions()[region])
def _boto3_session_from_creds(self, creds, region):
return boto3.session.Session(
aws_access_key_id=creds["access_key"],
aws_secret_access_key=creds["secret_key"],
aws_session_token=creds["session_token"],
region_name=region)
def boto3_session_for_region(self, region):
return self._boto3_session_from_creds(
self.get_credentials(region), region)
def iter_regions(self):
for region, cred_url in self.get_regions().items():
yield self._boto3_session_from_creds(self._get(cred_url), region)
class ReleaseReadmeUpdater:
SECTION_TPL = textwrap.dedent("""
### Alpine Linux {release} ({date})
click to show/hide
{rows}