136 lines
4.5 KiB
Python
Executable File
136 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Build a secret-bearing ECS user-data file without printing its contents."""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from .render_asterisk import render
|
|
except ImportError: # Direct execution: python3 deploy/build_asterisk_userdata.py
|
|
from render_asterisk import render
|
|
|
|
CONFIG_FILES = ("http.conf", "ari.conf", "pjsip.conf", "rtp.conf", "extensions.conf")
|
|
IMAGE_RE = re.compile(r"^[A-Za-z0-9._/-]+(?::[A-Za-z0-9_.-]+)?@sha256:[0-9a-f]{64}$")
|
|
|
|
|
|
def read_secret(path):
|
|
if path is None:
|
|
value = os.environ.get("ARI_PASSWORD", "")
|
|
else:
|
|
path = Path(path)
|
|
if path.is_symlink() or not path.is_file():
|
|
raise ValueError("ARI password file must be an existing non-symlink file")
|
|
value = path.read_text().rstrip("\n")
|
|
if not value:
|
|
raise ValueError("set ARI_PASSWORD or --ari-password-file")
|
|
return value
|
|
|
|
|
|
def immutable_image(value):
|
|
if not isinstance(value, str) or not IMAGE_RE.fullmatch(value):
|
|
raise ValueError("--image must be a registry image pinned by @sha256:<64 hex>")
|
|
return value
|
|
|
|
|
|
def build(cfg, image, password, config_gid=1000, template=None, environment=None):
|
|
if (
|
|
isinstance(config_gid, bool)
|
|
or not isinstance(config_gid, int)
|
|
or not 1 <= config_gid <= 65535
|
|
):
|
|
raise ValueError("config GID must be between 1 and 65535")
|
|
env = dict(os.environ if environment is None else environment)
|
|
env["ARI_PASSWORD"] = password
|
|
rendered = render(cfg, env)
|
|
payloads = {
|
|
name.upper().replace(".", "_") + "_B64": base64.b64encode(
|
|
rendered[name].encode()
|
|
).decode()
|
|
for name in CONFIG_FILES
|
|
}
|
|
variables = [
|
|
"#!/usr/bin/env bash",
|
|
"# Generated by deploy/build_asterisk_userdata.py; do not commit this file.",
|
|
f"ASTERISK_IMAGE={shlex.quote(image)}",
|
|
f"ASTERISK_CONFIG_GID={config_gid}",
|
|
]
|
|
variables.extend(f"{key}={shlex.quote(value)}" for key, value in payloads.items())
|
|
variables.append("")
|
|
if template is None:
|
|
template = Path(__file__).with_name("asterisk_bootstrap.sh")
|
|
template_text = Path(template).read_text()
|
|
if template_text.startswith("#!"):
|
|
template_text = template_text.split("\n", 1)[1]
|
|
return "\n".join(variables) + template_text
|
|
|
|
|
|
def write_output(content, path):
|
|
path = Path(path)
|
|
if path.exists() or path.is_symlink():
|
|
raise ValueError(
|
|
"output already exists; choose a new local secret-bearing path"
|
|
)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary_path = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
dir=path.parent,
|
|
prefix=".userdata-",
|
|
delete=False,
|
|
) as temporary:
|
|
temporary.write(content)
|
|
temporary.flush()
|
|
os.fsync(temporary.fileno())
|
|
temporary_path = Path(temporary.name)
|
|
temporary_path.chmod(0o600)
|
|
temporary_path.replace(path)
|
|
path.chmod(0o600)
|
|
except Exception:
|
|
if temporary_path is not None:
|
|
temporary_path.unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--config", required=True, help="render_asterisk JSON config")
|
|
parser.add_argument("--image", help="immutable registry image@sha256:digest")
|
|
parser.add_argument("--ari-password-file")
|
|
parser.add_argument(
|
|
"--output", required=True, help="local secret-bearing user-data path"
|
|
)
|
|
parser.add_argument("--config-gid", type=int, default=1000)
|
|
args = parser.parse_args()
|
|
try:
|
|
cfg = json.loads(Path(args.config).read_text())
|
|
image = immutable_image(
|
|
args.image or cfg.get("asterisk_image") or os.environ.get("ASTERISK_IMAGE")
|
|
)
|
|
content = build(
|
|
cfg, image, read_secret(args.ari_password_file), args.config_gid
|
|
)
|
|
if len(content.encode()) > 16 * 1024:
|
|
raise ValueError("generated user-data exceeds the ECS 16 KiB limit")
|
|
write_output(content, args.output)
|
|
print(
|
|
f"wrote {args.output} ({len(content.encode())} bytes); "
|
|
"secret payload not printed"
|
|
)
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|