feat(notification): enable production publisher

Preserve production broker definitions and allow bounded SMTP delivery.

Ref: IT-1508
This commit is contained in:
2026-10-03 15:17:52 +03:00
parent a684caf7f7
commit 8d9ed96f2c
5 changed files with 425 additions and 11 deletions
@@ -6,8 +6,9 @@ Ingress, NodePort or unauthenticated administration. Application administration,
RabbitMQ management and the sandbox inbox are available through explicit local
`kubectl port-forward` sessions over SSH.
The initial channel is `Invemory.Staging.Sandbox`, template `TestEmail`, exact language
`en`. SMTP egress permits only Mailpit:1025. Mailpit has no relay configuration.
The initial sandbox channel is `Invemory.Staging.Sandbox`, template `TestEmail`, exact
language `en`. Its SMTP egress uses Mailpit:1025, which has no relay configuration.
The production add-on also permits the Worker to reach external SMTP on TCP 587.
Only its temporary inbox is ephemeral. Channels, templates, SMTP settings and usage
counters are persisted in PostgreSQL; durable broker queues, messages, users and
permissions are retained on the RabbitMQ volume. Never put real SMTP credentials
@@ -37,10 +38,59 @@ default-exchange publication. Resource ACLs constrain exchange/queue names, not
routing key of a publish operation: access to the default exchange permits routing
to other existing queues within the same vhost. This MVP shared-queue design therefore
does not provide security isolation between message ServiceNames or routing keys.
No production publisher or SMTP egress is configured.
The production publisher credential, result queue, network policies and SMTP egress
are provisioned separately below.
Additional environments need separately reviewed routing, users, permissions, exact
templates, provider settings and egress before they can join the shared service.
## Invemory production publisher credential
The additive `activate-production-publisher.py` provisions the dedicated
`invemory_production` user in `/notification`, durable result queue
`invemory.production.notification-results`, and the Notification Worker permissions
for that result queue. It preserves the staging user and queue. Run it on mini-pc-2
as `devops` from its reviewed copy; repeated runs verify the existing password and
ACLs. The generated password is retained only in
`/home/devops/shared-notification-private/invemory-production-rabbitmq-password.txt`
with mode `0600`. It is never printed by the provisioning script.
On mini-pc-2, the owner can display the password in their own terminal with:
```sh
sudo -u devops cat /home/devops/shared-notification-private/invemory-production-rabbitmq-password.txt
```
Use the `invemory_production` user, that password, broker
`rabbitmq.shared-notification.svc.cluster.local:5672`, vhost `/notification`, send
queue `notification.send-email`, result queue
`invemory.production.notification-results`, and the existing technical service name
`StoreMate-Prod` in the Invemory production notification settings. The visible sender
and `TestEmail` text still contain legacy branding; IT-1510 tracks that copy change.
### Live activation — 2026-10-03
`production-network.yaml` adds only production API/Outbox egress and RabbitMQ ingress
on TCP 5672, plus shared Worker egress to public SMTP on TCP 587 (private and cluster
address ranges excluded). The existing sandbox Mailpit allowance stays intact. The
file passed Kubernetes server dry-run and was applied on mini-pc-2. The API result
consumer and both Workers are running at one replica; RabbitMQ has one consumer on
`notification.send-email` and one on
`invemory.production.notification-results`. Two owned production test messages were
published, accepted by the Gmail SMTP channel, acknowledged as delivery successes in
Invemory Outbox, and confirmed in the recipient's Gmail conversation. Today's channel
usage increased by two. Both queues ended with zero ready and unacknowledged messages.
The activation helper also persists the production user, queue and expanded Worker ACL
in the `rabbit-definitions` Secret. The private password and matching RabbitMQ hash
are saved with mode `0600`; updated `notification.py` reproduces that Secret during
bootstrap without rotating passwords. The reviewed source is installed at
`/home/devops/notification-package/notification.py`, with its pre-activation copy
retained beside it. A broker restart should therefore reload production access;
verify the two consumers and an owned delivery afterward. The SMTP password remains
outside the repository. Apply `production-network.yaml` separately when reconciling
network policies. Do not use the initial sandbox-only render as a complete production
delivery configuration.
Web/API administration has no built-in authentication in this source revision. Its
Service stays ClusterIP and default-deny ingress; access through SSH/port-forward is
restricted to operators. The worker exposes no health HTTP endpoint. Pod rollout is
@@ -0,0 +1,224 @@
#!/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)
@@ -6,6 +6,7 @@ import base64
import contextlib
import datetime
import hashlib
import hmac
import json
import os
from pathlib import Path, PurePosixPath
@@ -20,6 +21,7 @@ MANAGED = "hrynco-shared-notification"
MIGRATIONS = ["20260501214629_InitialCreate", "20260502154249_PendingChanges"]
IMAGE = re.compile(r"registry\.hrynco\.com/[a-z0-9._/-]+@sha256:[a-f0-9]{64}$")
SHA = re.compile(r"[a-f0-9]{64}$")
PRODUCTION_RESULT_QUEUE = "invemory.production.notification-results"
# Marker must be provisioned only after findmnt proves the backing filesystem UUID.
# XFS/ext4 statfs plus exact marker prevents a missing-mount fallback onto another FS.
@@ -313,7 +315,7 @@ def network_policies(config):
return output
def rabbit_definitions(config, credentials):
def rabbit_definitions(config, credentials, production_password_hash=None):
def hashed(password, salt):
return base64.b64encode(salt + hashlib.sha256(salt + password.encode()).digest()).decode()
@@ -321,27 +323,58 @@ def rabbit_definitions(config, credentials):
for name, key, tag in (("notification_admin", "rabbitAdminPassword", "administrator"), ("notification_worker", "rabbitWorkerPassword", ""), ("invemory_staging", "rabbitPublisherPassword", "")):
salt = base64.b64decode(credentials[key + "Salt"])
users.append({"name": name, "password_hash": hashed(credentials[key], salt), "hashing_algorithm": "rabbit_password_hashing_sha256", "tags": tag})
if production_password_hash:
users.append({"name": "invemory_production", "password_hash": production_password_hash,
"hashing_algorithm": "rabbit_password_hashing_sha256", "tags": ""})
incoming = re.escape("notification.send-email")
reply = re.escape(config["replyQueue"])
production_reply = re.escape(PRODUCTION_RESULT_QUEUE)
worker_replies = reply + ("|" + production_reply if production_password_hash else "")
vhost = config["rabbitVirtualHost"]
return {"users": users, "vhosts": [{"name": vhost}], "permissions": [
permissions = [
{"user": "notification_admin", "vhost": vhost, "configure": ".*", "write": ".*", "read": ".*"},
{"user": "notification_worker", "vhost": vhost, "configure": "^(" + incoming + "|" + reply + ")$", "write": "^(amq\\.default|" + reply + ")$", "read": "^" + incoming + "$"},
{"user": "invemory_staging", "vhost": vhost, "configure": "^(" + incoming + "|" + reply + ")$", "write": "^(amq\\.default|" + incoming + ")$", "read": "^" + reply + "$"}],
"queues": [{"name": name, "vhost": vhost, "durable": True, "auto_delete": False, "arguments": {}} for name in ("notification.send-email", config["replyQueue"])], "exchanges": [], "bindings": []}
{"user": "notification_worker", "vhost": vhost, "configure": "^(" + incoming + "|" + worker_replies + ")$", "write": "^(amq\\.default|" + worker_replies + ")$", "read": "^" + incoming + "$"},
{"user": "invemory_staging", "vhost": vhost, "configure": "^(" + incoming + "|" + reply + ")$", "write": "^(amq\\.default|" + incoming + ")$", "read": "^" + reply + "$"}]
if production_password_hash:
permissions.append({"user": "invemory_production", "vhost": vhost,
"configure": "^(" + incoming + "|" + production_reply + ")$",
"write": "^(amq\\.default|" + incoming + ")$", "read": "^" + production_reply + "$"})
queues = ["notification.send-email", config["replyQueue"]]
if production_password_hash:
queues.append(PRODUCTION_RESULT_QUEUE)
return {"users": users, "vhosts": [{"name": vhost}], "permissions": permissions,
"queues": [{"name": name, "vhost": vhost, "durable": True, "auto_delete": False, "arguments": {}} for name in queues],
"exchanges": [], "bindings": []}
def secret_manifests(config, credentials, registry):
def secret_manifests(config, credentials, registry, production_password_hash=None):
def secret(name, data, kind="Opaque"):
return obj("Secret", name, type=kind, data={key: base64.b64encode(value.encode()).decode() for key, value in data.items()})
data = {key: credentials[key] for key in ("postgresPassword", "runtimePassword", "rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword", "seqApiKey")}
data["ownerConnectionString"] = "Host=postgres;Database=notification_service;Username=notification_owner;Password=" + credentials["postgresPassword"]
data["runtimeConnectionString"] = "Host=postgres;Database=notification_service;Username=notification_runtime;Password=" + credentials["runtimePassword"]
return [secret("notification-private", data), secret("rabbit-definitions", {"definitions.json": json.dumps(rabbit_definitions(config, credentials))}),
return [secret("notification-private", data), secret("rabbit-definitions", {"definitions.json": json.dumps(rabbit_definitions(config, credentials, production_password_hash))}),
secret("registry-pull", {".dockerconfigjson": json.dumps(registry)}, "kubernetes.io/dockerconfigjson")]
def preserved_production_password_hash(private_dir):
password_path = Path(private_dir) / "invemory-production-rabbitmq-password.txt"
hash_path = Path(private_dir) / "invemory-production-rabbitmq-password-hash.txt"
if password_path.exists() != hash_path.exists():
raise RuntimeError("production RabbitMQ password and boot hash must be preserved together")
if not password_path.exists():
return None
if any(path.stat().st_mode & 0o077 for path in (password_path, hash_path)):
raise RuntimeError("production RabbitMQ private file permissions are too broad")
password = password_path.read_text(encoding="utf-8").strip()
password_hash = hash_path.read_text(encoding="utf-8").strip()
decoded = base64.b64decode(password_hash, validate=True)
if len(decoded) != 36 or not hmac.compare_digest(decoded[4:], hashlib.sha256(decoded[:4] + password.encode()).digest()):
raise RuntimeError("production RabbitMQ boot hash does not match preserved password")
return password_hash
class Cluster:
def __init__(self, args):
self.command = [args.kubectl, "--kubeconfig=" + args.kubeconfig, "--request-timeout=20s"]
@@ -662,7 +695,7 @@ def main(argv=None):
if namespace and namespace["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED:
raise RuntimeError("refusing to adopt existing namespace")
cluster.apply([resources[0]])
values = secret_manifests(config, credentials, registry)
values = secret_manifests(config, credentials, registry, preserved_production_password_hash(args.private_dir))
for value in values:
current = cluster.get("secret", value["metadata"]["name"])
if current and current["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED:
@@ -0,0 +1,85 @@
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: production-to-rabbitmq
namespace: invemory-production
spec:
podSelector:
matchLabels:
app.kubernetes.io/instance: invemory-production
matchExpressions:
- key: app.kubernetes.io/component
operator: In
values: [api, worker]
policyTypes: [Egress]
egress:
- to:
- namespaceSelector:
matchLabels:
kubernetes.io/metadata.name: shared-notification
podSelector:
matchLabels:
app.kubernetes.io/component: rabbitmq
app.kubernetes.io/name: hrynco-shared-notification
app.kubernetes.io/managed-by: hrynco-shared-notification
ports:
- protocol: TCP
port: 5672
---
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: rabbitmq-from-invemory-production
namespace: shared-notification
spec:
podSelector:
matchLabels:
app.kubernetes.io/component: rabbitmq
app.kubernetes.io/name: hrynco-shared-notification
app.kubernetes.io/managed-by: hrynco-shared-notification
policyTypes: [Ingress]
ingress:
- from:
- namespaceSelector:
matchLabels:
kubernetes.io/metadata.name: invemory-production
podSelector:
matchLabels:
app.kubernetes.io/instance: invemory-production
invemory.com/environment: production
matchExpressions:
- key: app.kubernetes.io/component
operator: In
values: [api, worker]
ports:
- protocol: TCP
port: 5672
---
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: worker-to-external-smtp
namespace: shared-notification
spec:
podSelector:
matchLabels:
app.kubernetes.io/component: worker
app.kubernetes.io/name: hrynco-shared-notification
app.kubernetes.io/managed-by: hrynco-shared-notification
policyTypes: [Egress]
egress:
- to:
- ipBlock:
cidr: 0.0.0.0/0
except:
- 10.0.0.0/8
- 100.64.0.0/10
- 127.0.0.0/8
- 169.254.0.0/16
- 172.16.0.0/12
- 192.168.0.0/16
- 224.0.0.0/4
- 240.0.0.0/4
ports:
- protocol: TCP
port: 587
@@ -1,6 +1,7 @@
import argparse
import base64
import copy
import hashlib
import importlib.util
import json
from pathlib import Path
@@ -186,6 +187,27 @@ class ContractTests(unittest.TestCase):
self.assertIsNotNone(re.fullmatch(permissions["write"], "amq.default"))
self.assertIsNone(re.fullmatch(permissions["write"], "unreviewed.exchange"))
def test_production_boot_definitions_keep_result_publish_permission(self):
config, _ = fixture()
credentials = {}
for key in ("rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword"):
credentials[key] = "staging-test-value"
credentials[key + "Salt"] = base64.b64encode(b"salt").decode()
password = "production-test-value"
password_hash = base64.b64encode(b"prod" + hashlib.sha256(b"prod" + password.encode()).digest()).decode()
definitions = ns.rabbit_definitions(config, credentials, password_hash)
self.assertEqual([q["name"] for q in definitions["queues"]],
["notification.send-email", "invemory.staging.notification-results",
"invemory.production.notification-results"])
production = next(x for x in definitions["users"] if x["name"] == "invemory_production")
self.assertEqual(production["password_hash"], password_hash)
self.assertNotIn(password, json.dumps(definitions))
worker = next(x for x in definitions["permissions"] if x["user"] == "notification_worker")
self.assertIsNotNone(re.fullmatch(worker["configure"], "invemory.production.notification-results"))
self.assertIsNotNone(re.fullmatch(worker["configure"], "invemory.staging.notification-results"))
self.assertIsNotNone(re.fullmatch(worker["write"], "amq.default"))
self.assertEqual(worker["read"], r"^notification\.send\-email$")
def test_secrets_not_present_in_rendered_resources(self):
config, artifacts = fixture()
self.assertFalse(any(x["kind"] == "Secret" for x in ns.manifest(config, artifacts)))