82 lines
3.4 KiB
Python
Executable File
82 lines
3.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""One-time administrator bootstrap for dedicated RouterOS API users and Secret."""
|
|
|
|
import argparse
|
|
import configparser
|
|
import os
|
|
import secrets
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
USERNAME = "wifi-vlan-reconciler"
|
|
GROUP = "wifi-vlan-reconciler"
|
|
SECRET = "routeros-wifi-vlan-credentials"
|
|
|
|
|
|
def run(*args):
|
|
return subprocess.run(args, check=True, capture_output=True, text=True)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("values", type=Path, help="homelab values.yaml with controller and caps")
|
|
parser.add_argument("state", type=Path, help="private temporary credential file; reuse on retry")
|
|
args = parser.parse_args()
|
|
values = yaml.safe_load(args.values.read_text())
|
|
hosts = [values["controller"], *values["caps"]]
|
|
if len(hosts) != len(set(hosts)):
|
|
raise SystemExit("duplicate device hostname")
|
|
if run("kubectl", "config", "current-context").stdout.strip() != "kubernetes-admin@kubernetes":
|
|
raise SystemExit("unexpected kubectl context")
|
|
|
|
config = configparser.ConfigParser(interpolation=None)
|
|
if args.state.exists():
|
|
config.read(args.state)
|
|
if set(config.sections()) != {"default", *hosts}:
|
|
raise SystemExit("saved credential file does not match the device list")
|
|
else:
|
|
config["default"] = {"port": "8728", "use_ssl": "false",
|
|
"plaintext_login": "true", "ssl_certificate_verify": "false"}
|
|
for host in hosts:
|
|
config[host] = {"hostname": host, "username": USERNAME,
|
|
"password": secrets.token_hex(24)}
|
|
fd = os.open(args.state, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "w") as handle:
|
|
config.write(handle)
|
|
|
|
failures = []
|
|
for host in hosts:
|
|
password = config[host]["password"]
|
|
command = (
|
|
f':if ([:len [/user/group/find where name="{GROUP}"]] = 0) '
|
|
f'do={{/user/group/add name={GROUP} policy=api,read,write}}; '
|
|
f':if ([:len [/user/find where name="{USERNAME}"]] = 0) '
|
|
f'do={{/user/add name={USERNAME} group={GROUP} password="{password}"}} '
|
|
f'else={{/user/set [find where name="{USERNAME}"] group={GROUP} password="{password}"}}; '
|
|
f':put [/user/get [find where name="{USERNAME}"] group]'
|
|
)
|
|
try:
|
|
result = run("ssh", "-n", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
|
|
f"admin@{host}", command)
|
|
if result.stdout.replace("\r", "").strip() != GROUP:
|
|
raise RuntimeError("group verification failed")
|
|
print(f"{host}: dedicated API account ready")
|
|
except Exception:
|
|
print(f"{host}: account provisioning failed", file=sys.stderr)
|
|
failures.append(host)
|
|
manifest = run("kubectl", "-n", "default", "create", "secret", "generic", SECRET,
|
|
f"--from-file=mktxp.conf={args.state}", "--dry-run=client", "-o", "yaml")
|
|
subprocess.run(["kubectl", "-n", "default", "apply", "-f", "-"], input=manifest.stdout,
|
|
text=True, check=True, capture_output=True)
|
|
print(f"default/{SECRET}: created from the dedicated per-device credentials")
|
|
if failures:
|
|
raise SystemExit("Provisioning incomplete; rerun with the same private state file when devices return")
|
|
args.state.unlink()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|