#!/usr/bin/env python3 """Add the Invemory production RabbitMQ publisher without exposing its password.""" import base64 import hashlib import importlib.util import json import os import re import secrets import select import subprocess import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path KUBECTL = "/snap/microk8s/current/kubectl" KUBECONFIG = "/home/devops/invemory-production/kubeconfig" ADMIN_CREDENTIALS = Path("/home/devops/shared-notification-private/credentials.json") PASSWORD_FILE = Path("/home/devops/shared-notification-private/invemory-production-rabbitmq-password.txt") PASSWORD_HASH_FILE = Path("/home/devops/shared-notification-private/invemory-production-rabbitmq-password-hash.txt") CONFIG_FILE = Path("/home/devops/notification-package/config.local.json") USER = "invemory_production" VHOST = "/notification" SEND_QUEUE = "notification.send-email" STAGING_RESULT = "invemory.staging.notification-results" PRODUCTION_RESULT = "invemory.production.notification-results" MODULE_PATH = Path(__file__).with_name("notification.py") SPEC = importlib.util.spec_from_file_location("shared_notification_boot", MODULE_PATH) notification = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(notification) def read_private(path): stat = path.stat() if stat.st_mode & 0o077: raise RuntimeError(f"private file permissions are too broad: {path}") return path.read_text(encoding="utf-8").strip() def new_password(): if PASSWORD_FILE.exists(): return read_private(PASSWORD_FILE), False password = secrets.token_urlsafe(36) fd = os.open(PASSWORD_FILE, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as output: output.write(password + "\n") return password, True def password_matches(user, password): if user.get("hashing_algorithm") != "rabbit_password_hashing_sha256": return False hashed = base64.b64decode(user["password_hash"]) return len(hashed) == 36 and hashed[4:] == hashlib.sha256(hashed[:4] + password.encode()).digest() def kubectl(*arguments, data=None): result = subprocess.run([KUBECTL, "--kubeconfig=" + KUBECONFIG, "-n", "shared-notification", *arguments], input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=30) if result.returncode: raise RuntimeError("RabbitMQ boot definitions Kubernetes operation failed") return result.stdout def preserve_boot_definitions(credentials, user, password): if not password_matches(user, password): raise RuntimeError("production broker password hash does not match private password") password_hash = user["password_hash"] if PASSWORD_HASH_FILE.exists(): if read_private(PASSWORD_HASH_FILE) != password_hash: raise RuntimeError("preserved production broker boot hash differs") else: fd = os.open(PASSWORD_HASH_FILE, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as output: output.write(password_hash + "\n") if notification.preserved_production_password_hash(PASSWORD_FILE.parent) != password_hash: raise RuntimeError("bootstrap cannot reproduce production broker password hash") config = json.loads(CONFIG_FILE.read_text(encoding="utf-8")) before = notification.rabbit_definitions(config, credentials) after = notification.rabbit_definitions(config, credentials, password_hash) secret = json.loads(kubectl("get", "secret", "rabbit-definitions", "-o", "json")) current = json.loads(base64.b64decode(secret["data"]["definitions.json"])) if current not in (before, after): raise RuntimeError("RabbitMQ boot definitions differ from the reviewed base or production extension") if current == before: metadata = {key: secret["metadata"][key] for key in ("name", "namespace", "resourceVersion", "labels")} if "annotations" in secret["metadata"]: metadata["annotations"] = secret["metadata"]["annotations"] replacement = {"apiVersion": "v1", "kind": "Secret", "metadata": metadata, "type": "Opaque", "data": {"definitions.json": base64.b64encode(json.dumps(after).encode()).decode()}} kubectl("replace", "-f", "-", data=json.dumps(replacement)) actual = json.loads(kubectl("get", "secret", "rabbit-definitions", "-o", "json")) if json.loads(base64.b64decode(actual["data"]["definitions.json"])) != after: raise RuntimeError("production RabbitMQ boot definitions readback failed") def port_forward(): process = subprocess.Popen( [KUBECTL, "--kubeconfig=" + KUBECONFIG, "-n", "shared-notification", "port-forward", "--address", "127.0.0.1", "service/rabbitmq", ":15672"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, ) deadline = time.monotonic() + 20 while time.monotonic() < deadline: if process.poll() is not None: raise RuntimeError("RabbitMQ port-forward exited") ready, _, _ = select.select([process.stdout], [], [], 0.5) if ready: line = process.stdout.readline() match = re.search(r"Forwarding from 127\.0\.0\.1:(\d+) -> 15672", line) if match: return process, "http://127.0.0.1:" + match.group(1) process.terminate() raise RuntimeError("RabbitMQ port-forward timed out") def main(): credentials = json.loads(read_private(ADMIN_CREDENTIALS)) admin_password = credentials["rabbitAdminPassword"] password, created = new_password() process, base = port_forward() def request(method, path, body=None): authorization = "notification_admin:" + admin_password headers = {"Authorization": "Basic " + base64.b64encode(authorization.encode()).decode()} if body is not None: headers["Content-Type"] = "application/json" data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(base + path, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=10) as response: return response.status, json.loads(response.read() or b"{}") except urllib.error.HTTPError as error: if error.code == 404: return 404, None raise RuntimeError(f"RabbitMQ API returned HTTP {error.code} for {method} {path}") from None try: status, vhost = request("GET", "/api/vhosts/" + urllib.parse.quote(VHOST, safe="")) if status != 200 or vhost["name"] != VHOST: raise RuntimeError("expected RabbitMQ vhost is absent") for queue in (SEND_QUEUE, STAGING_RESULT): status, value = request("GET", "/api/queues/%2Fnotification/" + queue) if status != 200 or not value["durable"]: raise RuntimeError("expected shared RabbitMQ queue is absent") status, existing = request("GET", "/api/users/" + USER) if status == 404: request("PUT", "/api/users/" + USER, {"password": password, "tags": ""}) elif status == 200: if not password_matches(existing, password): raise RuntimeError("existing production user has a different password") else: raise RuntimeError("cannot inspect production RabbitMQ user") status, permissions = request("GET", "/api/permissions/%2Fnotification/" + USER) incoming = re.escape(SEND_QUEUE) staging_reply = re.escape(STAGING_RESULT) production_reply = re.escape(PRODUCTION_RESULT) expected = { "configure": "^(" + incoming + "|" + production_reply + ")$", "write": "^(amq\\.default|" + incoming + ")$", "read": "^" + production_reply + "$", } if status == 404: request("PUT", "/api/permissions/%2Fnotification/" + USER, expected) elif status != 200 or any(permissions[key] != value for key, value in expected.items()): raise RuntimeError("production user permissions differ from the reviewed contract") status, queue = request("GET", "/api/queues/%2Fnotification/" + PRODUCTION_RESULT) if status == 404: request("PUT", "/api/queues/%2Fnotification/" + PRODUCTION_RESULT, {"durable": True, "auto_delete": False, "arguments": {}}) elif status != 200 or not queue["durable"] or queue["auto_delete"]: raise RuntimeError("production result queue differs from the reviewed contract") # Worker publishes result messages to ReplyTo through the default exchange. status, worker = request("GET", "/api/permissions/%2Fnotification/notification_worker") old = { "configure": "^(" + incoming + "|" + staging_reply + ")$", "write": "^(amq\\.default|" + staging_reply + ")$", "read": "^" + incoming + "$", } updated = { "configure": "^(" + incoming + "|" + staging_reply + "|" + production_reply + ")$", "write": "^(amq\\.default|" + staging_reply + "|" + production_reply + ")$", "read": old["read"], } if status != 200: raise RuntimeError("expected Notification Worker permissions are absent") if all(worker[key] == value for key, value in old.items()): request("PUT", "/api/permissions/%2Fnotification/notification_worker", updated) elif any(worker[key] != value for key, value in updated.items()): raise RuntimeError("Notification Worker permissions differ from the reviewed contract") status, identity = request("GET", "/api/users/" + USER) if status != 200 or not password_matches(identity, password): raise RuntimeError("production RabbitMQ credential verification failed") preserve_boot_definitions(credentials, identity, password) print(json.dumps({"result": "PASS", "user": USER, "vhost": VHOST, "resultQueue": PRODUCTION_RESULT, "passwordFile": str(PASSWORD_FILE), "passwordCreated": created, "bootDefinitionsSaved": True, "passwordPrinted": False})) finally: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) if __name__ == "__main__": try: main() except Exception as error: print(json.dumps({"result": "FAIL", "error": str(error), "passwordPrinted": False}), file=sys.stderr) sys.exit(1)