diff --git a/.env b/.env new file mode 100644 index 0000000..cb074a0 --- /dev/null +++ b/.env @@ -0,0 +1,57 @@ +# --- Required --- +# Space-separated list of NATS URLs. Use tls:// if your server requires TLS. +# Examples: +# NATS_SERVERS="tls://nats01.example:4222" +# NATS_SERVERS="nats://127.0.0.1:4222" +NATS_SERVERS="tls://cloud.ikeja.co.za:4222" + +# --- Optional --- +# Queue group (leave empty to receive every registration message) +# NATS_QUEUE="registration-listeners" +NATS_NAME="registration-listener" +# PRINT_RAW=1 + +# --- Auth (pick one; comment the rest) --- +# NATS_TOKEN="REDACTED_TOKEN" +# NATS_USER="username" +# NATS_PASS="password" + +# If your NATS uses username/password: +NATS_USER="agent" +NATS_PASS="eiTh2chi-sahp3iJ9-PeeR4AiD-ahXa8EeG-eeLeew4i-fu4Aejei" + +# If you have a .creds file mounted by the volume, reference it directly: +# NATS_CREDS="/opt/containers/nats-registration-listener/data/user.creds" + +# Or inject it dynamically as base64 (entrypoint will write to DATA_DIR): +# NATS_CREDS_B64="BASE64_OF_USER.CREDS" + +# --- TLS (server-auth only) --- +# Option A: pre-mounted CA path (host ./data -> container /opt/.../data) +# TLS_CA="/opt/containers/nats-registration-listener/data/8devices-root-ca.pem" + +# Option B: inline PEM (entrypoint writes to DATA_DIR/tls-ca.pem) +# TLS_CA_PEM="-----BEGIN CERTIFICATE----- +# MIICbDCCAdUCFCFi1ReBpGi0EQ+iZylOYXXF15elMA0GCSqGSIb3DQEBCwUAMHQx +# CzAJBgNVBAYTAkxUMQowCAYDVQQIDAEtMRAwDgYDVQQHDAdWaWxuaXVzMRUwEwYD +# VQQKDAw4REVWSUNFUyBVQUIxDDAKBgNVBAsMA1ImRDEiMCAGA1UEAwwZOERFVklD +# RVMgUHJvZHVjdHMgUm9vdCBDQTAgFw0yMDA0MTYxMzIzMjBaGA8yMTIwMDMyMzEz +# MjMyMFowdDELMAkGA1UEBhMCTFQxCjAIBgNVBAgMAS0xEDAOBgNVBAcMB1ZpbG5p +# dXMxFTATBgNVBAoMDDhERVZJQ0VTIFVBQjEMMAoGA1UECwwDUiZEMSIwIAYDVQQD +# DBk4REVWSUNFUyBQcm9kdWN0cyBSb290IENBMIGfMA0GCSqGSIb3DQEBAQUAA4GN +# ADCBiQKBgQC0e3L19mgARZgb7yc331S03L85F/q0MHhXGeNA8XJCnzfSIny0EzgM +# nrWKVvPSUdmb6lF/M7SCnbw3Suf1UD8AQYgjDNNh513zK5MhnZCwNfnXxv6upx78 +# PAMzLgKmEixErl8syRB1pfkz1Fc6GbOtdtAvzbFBD8FsHuNBYJxgqQIDAQABMA0G +# CSqGSIb3DQEBCwUAA4GBADq6R9k9Ob6aqHda9ZZ+0BbPtQzC38QNJ8QCh4AFSIsK +# RrJEcyns1Qnh7/5+ZOY/UhU4h++uBuUVacH4DmOJgbxae/OXNAMcpbvYM2SZ6KfR +# 5eSeEAJ+t49b2SaanYhZFusrn1fi8JhJQ0wi5LYfmLz/Kf4zGGZj+IgR9QmwIcLf +# -----END CERTIFICATE-----" + +# Option C: base64 of the PEM (safer for .env with multi-line) +# TLS_CA_PEM_B64="BASE64_OF_PEM" + +# Option D: fetch from URL at startup +# TLS_CA_URL="https://example.internal/8devices-root-ca.pem" + +# Only disable hostname verification if absolutely necessary: +# TLS_INSECURE=0 diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..8b62237 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,35 @@ +FROM ubuntu:24.04 + +ENV DEBIAN_FRONTEND=noninteractive +ENV APP_DIR=/opt/containers/nats-registration-listener +ENV DATA_DIR=${APP_DIR}/data + +# Minimal deps + Python + dos2unix for CRLF→LF +RUN apt-get update && apt-get install -y --no-install-recommends \ + bash curl jq ca-certificates tini python3 python3-pip dos2unix \ + && rm -rf /var/lib/apt/lists/* + +# Python deps +RUN pip3 install --no-cache-dir --upgrade nats-py + +# App payload (immutable inside image) +RUN mkdir -p ${APP_DIR}/app/bin +COPY files/nats-registration-listener.sh ${APP_DIR}/app/bin/nats-registration-listener.sh +COPY files/entrypoint.sh /usr/local/bin/nats-registration-entrypoint +COPY files/healthcheck.sh /usr/local/bin/nats-registration-healthcheck +COPY files/nats_registration_listener.py /usr/local/bin/nats_registration_listener.py + +# Normalize line endings (in case files came from Windows) and set exec bits +RUN dos2unix /usr/local/bin/nats-registration-entrypoint \ + /usr/local/bin/nats-registration-healthcheck \ + ${APP_DIR}/app/bin/nats-registration-listener.sh \ + /usr/local/bin/nats_registration_listener.py \ + && chmod +x /usr/local/bin/nats-registration-entrypoint \ + /usr/local/bin/nats-registration-healthcheck \ + ${APP_DIR}/app/bin/nats-registration-listener.sh + +WORKDIR ${APP_DIR}/app + +# Use tini as PID1; entrypoint validates env and execs the main script +ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/nats-registration-entrypoint"] +CMD ["/opt/containers/nats-registration-listener/app/bin/nats-registration-listener.sh"] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..4f0b4a5 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,30 @@ +services: + nats-registration-listener: + build: + context: . + dockerfile: Dockerfile + image: nats-registration-listener:latest + + # Share host network so 'localhost:4222' talks to NATS on the host (if that's your setup) + network_mode: host + + working_dir: /opt/containers/nats-registration-listener/app + entrypoint: ["/usr/bin/tini", "--", "/usr/local/bin/nats-registration-entrypoint"] + command: ["/opt/containers/nats-registration-listener/app/bin/nats-registration-listener.sh"] + + restart: unless-stopped + tty: true + stdin_open: true + + # One .env to rule them all + env_file: + - .env + + volumes: + - ./data:/opt/containers/nats-registration-listener/data:rw + + healthcheck: + test: ["CMD-SHELL", "/usr/local/bin/nats-registration-healthcheck"] + interval: 30s + timeout: 10s + retries: 5 diff --git a/files/entrypoint.sh b/files/entrypoint.sh new file mode 100644 index 0000000..2390914 --- /dev/null +++ b/files/entrypoint.sh @@ -0,0 +1,75 @@ +#!/usr/bin/env bash +set -euo pipefail + +log(){ echo "[nats-registration][entrypoint] $*"; } +err(){ echo "[nats-registration][entrypoint][ERROR] $*" >&2; } + +# Sanity: tools +for bin in curl jq python3; do + command -v "$bin" >/dev/null 2>&1 || { err "Missing $bin"; exit 2; } +done + +# Paths (match Dockerfile ENV) +: "${APP_DIR:=/opt/containers/nats-registration-listener}" +: "${DATA_DIR:=${APP_DIR}/data}" +mkdir -p "${DATA_DIR}" + +# Required (fed from .env via compose) +: "${NATS_SERVERS:?NATS_SERVERS required (space-separated NATS URLs, e.g. 'tls://nats01:4222')}" +: "${NATS_SUBJECT:?NATS_SUBJECT required (e.g. 'v1.registration')}" + +# Optional with defaults +: "${NATS_NAME:=registration-listener}" +: "${PRINT_RAW:=}" +: "${NATS_QUEUE:=}" + +# TLS input options (dynamic) +# Priority: +# 1) TLS_CA -> use as a path (pre-mounted file) +# 2) TLS_CA_PEM -> write PEM content to ${DATA_DIR}/tls-ca.pem +# 3) TLS_CA_PEM_B64 -> base64 decode to ${DATA_DIR}/tls-ca.pem +# 4) TLS_CA_URL -> curl to ${DATA_DIR}/tls-ca.pem +: "${TLS_CA:=}" +: "${TLS_CA_PEM:=}" +: "${TLS_CA_PEM_B64:=}" +: "${TLS_CA_URL:=}" +: "${TLS_INSECURE:=}" + +# Auth (optional) +: "${NATS_USER:=}" +: "${NATS_PASS:=}" +: "${NATS_TOKEN:=}" +: "${NATS_CREDS:=}" # path to a .creds file if you have one +: "${NATS_CREDS_B64:=}" # optional: base64-encoded creds; will be written to DATA_DIR + +# ---- Dynamic materialization of CA (and optional .creds) ---- +CA_TARGET="${DATA_DIR}/tls-ca.pem" +CREDS_TARGET="${DATA_DIR}/user.creds" + +materialized_ca="" + +if [[ -z "${TLS_CA}" ]]; then + if [[ -n "${TLS_CA_PEM}" ]]; then + log "Materializing TLS CA from TLS_CA_PEM -> ${CA_TARGET}" + printf '%s\n' "${TLS_CA_PEM}" > "${CA_TARGET}" + chmod 0644 "${CA_TARGET}" + TLS_CA="${CA_TARGET}" + materialized_ca="1" + elif [[ -n "${TLS_CA_PEM_B64}" ]]; then + log "Materializing TLS CA from TLS_CA_PEM_B64 -> ${CA_TARGET}" + echo "${TLS_CA_PEM_B64}" | base64 -d > "${CA_TARGET}" + chmod 0644 "${CA_TARGET}" + TLS_CA="${CA_TARGET}" + materialized_ca="1" + elif [[ -n "${TLS_CA_URL}" ]]; then + log "Fetching TLS CA from TLS_CA_URL -> ${CA_TARGET}" + curl -fsSL "${TLS_CA_URL}" -o "${CA_TARGET}" + chmod 0644 "${CA_TARGET}" + TLS_CA="${CA_TARGET}" + materialized_ca="1" + fi +fi + +# Validate CA if present +if [[ -n "${TLS_CA}" ]]; then + if [[ ! -s "${TLS_CA}" ]_]()]() diff --git a/files/healthcheck.sh b/files/healthcheck.sh new file mode 100644 index 0000000..020124f --- /dev/null +++ b/files/healthcheck.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Simple TCP healthcheck against the first server in NATS_SERVERS +# Accepts schemes nats:// or tls:// +first_server=$(awk '{print $1}' <<< "${NATS_SERVERS:-}") +if [[ -z "${first_server}" ]]; then + echo "NATS_SERVERS not set" >&2 + exit 1 +fi + +# Extract host and port +scheme="${first_server%%://*}" +rest="${first_server#*://}" + +host="${rest%:*}" +port="${rest##*:}" + +# If rest still contains '/', strip path +host="${host%%/*}" +port="${port%%/*}" + +: "${port:=4222}" + +# Try TCP connect (bash /dev/tcp) +timeout 3 bash -c "cat < /dev/null > /dev/tcp/${host}/${port}" 2>/dev/null || { + echo "Cannot connect to ${host}:${port}" >&2 + exit 1 +} + +# Optionally add more checks later (e.g., NATS /varz on 8222) +exit 0 diff --git a/files/nats-registration-listener.sh b/files/nats-registration-listener.sh new file mode 100644 index 0000000..f5d6546 --- /dev/null +++ b/files/nats-registration-listener.sh @@ -0,0 +1,60 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Build args for the Python listener based on env +args=() + +# Servers: split on spaces +read -r -a servers <<< "${NATS_SERVERS}" +for s in "${servers[@]}"; do + args+=(--servers "$s") +done + +# Subject (required) +args+=(--subject "${NATS_SUBJECT}") + +# Optional queue group +if [[ -n "${NATS_QUEUE:-}" ]]; then + args+=(--queue "${NATS_QUEUE}") +fi + +# Client name +if [[ -n "${NATS_NAME:-}" ]]; then + args+=(--name "${NATS_NAME}") +fi + +# Auth +if [[ -n "${NATS_TOKEN:-}" ]]; then + args+=(--token "${NATS_TOKEN}") +fi +if [[ -n "${NATS_USER:-}" ]]; then + args+=(--user "${NATS_USER}") +fi +if [[ -n "${NATS_PASS:-}" ]]; then + args+=(--password "${NATS_PASS}") +fi +if [[ -n "${NATS_CREDS:-}" ]]; then + args+=(--creds "${NATS_CREDS}") +fi + +# TLS +if [[ -n "${TLS_CA:-}" ]]; then + args+=(--tls-ca "${TLS_CA}") +fi +if [[ -n "${TLS_CERT:-}" && -n "${TLS_KEY:-}" ]]; then + args+=(--tls-cert "${TLS_CERT}" --tls-key "${TLS_KEY}") +fi +if [[ -n "${TLS_INSECURE:-}" ]]; then + case "${TLS_INSECURE}" in + 1|true|TRUE|yes|YES) args+=(--insecure) ;; + esac +fi + +# Output mode +if [[ -n "${PRINT_RAW:-}" ]]; then + case "${PRINT_RAW}" in + 1|true|TRUE|yes|YES) args+=(--print-raw) ;; + esac +fi + +exec python3 /usr/local/bin/nats_registration_listener.py "${args[@]}" diff --git a/files/nats_registration_listener.py b/files/nats_registration_listener.py new file mode 100644 index 0000000..8c4eba9 --- /dev/null +++ b/files/nats_registration_listener.py @@ -0,0 +1,183 @@ +#!/usr/bin/env python3 +""" +NATS Registration Listener +-------------------------- +Listens on a NATS subject (default: 'registration') and prints messages to STDOUT. + +Install deps: + pip install --upgrade nats-py + +Examples: + python nats_registration_listener.py --servers nats://127.0.0.1:4222 --subject registration + python nats_registration_listener.py --servers nats://n1:4222 nats://n2:4222 --token $NATS_TOKEN + python nats_registration_listener.py --servers nats://host:4222 --creds /path/to/user.creds + python nats_registration_listener.py --servers tls://host:4443 --tls-ca ca.pem --tls-cert client.crt --tls-key client.key + +Notes: +- If messages are JSON, they will be pretty-printed; otherwise raw text (UTF-8) or bytes length is shown. +- Reconnects automatically with exponential backoff, prints connection status to STDERR. +""" + +import argparse +import asyncio +import json +import os +import signal +import ssl +import sys +from datetime import datetime, timezone +from typing import Optional + +import nats + + +def parse_args(): + p = argparse.ArgumentParser(description="Listen to a NATS subject and print messages.") + p.add_argument( + "--servers", + nargs="+", + help="One or more NATS server URLs (e.g., nats://127.0.0.1:4222).", + default=[os.environ.get("NATS_URL", "nats://127.0.0.1:4222")], + ) + p.add_argument( + "--subject", + help="NATS subject to subscribe to.", + default=os.environ.get("NATS_SUBJECT", "registration"), + ) + p.add_argument("--queue", help="Optional queue group name.", default=os.environ.get("NATS_QUEUE")) + p.add_argument("--name", help="Client connection name.", default="registration-listener") + + # Auth options + p.add_argument("--creds", help="Path to .creds file (JWT + NKey).") + p.add_argument("--user", help="Username.") + p.add_argument("--password", help="Password.") + p.add_argument("--token", help="Auth token.") + + # TLS options + p.add_argument("--tls-ca", help="Path to CA certificate for TLS.") + p.add_argument("--tls-cert", help="Path to client certificate for TLS.") + p.add_argument("--tls-key", help="Path to client key for TLS.") + p.add_argument("--insecure", action="store_true", help="Disable TLS hostname verification.") + + p.add_argument("--print-raw", action="store_true", help="Always print raw message bytes (no decoding / pretty JSON).") + + return p.parse_args() + + +def make_ssl_context(args) -> Optional[ssl.SSLContext]: + if not any([args.tls_ca, args.tls_cert, args.tls_key]) and not any(url.startswith("tls://") for url in args.servers): + return None + + ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH) + if args.tls_ca: + ctx.load_verify_locations(args.tls_ca) + + if args.tls_cert and args.tls_key: + ctx.load_cert_chain(certfile=args.tls_cert, keyfile=args.tls_key) + + if args.insecure: + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + + return ctx + + +def ts() -> str: + return datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S%z") + + +async def main(): + args = parse_args() + ssl_ctx = make_ssl_context(args) + + reconnect_time_wait = 2 # seconds base + max_reconnect_attempts = -1 # infinite + + async def disconnected_cb(): + print(f"[{ts()}] Disconnected from NATS.", file=sys.stderr, flush=True) + + async def reconnected_cb(): + print(f"[{ts()}] Reconnected to NATS.", file=sys.stderr, flush=True) + + async def error_cb(e): + print(f"[{ts()}] Error: {e!r}", file=sys.stderr, flush=True) + + async def closed_cb(): + print(f"[{ts()}] Connection closed.", file=sys.stderr, flush=True) + + nc = await nats.connect( + servers=args.servers, + name=args.name, + allow_reconnect=True, + reconnect_time_wait=reconnect_time_wait, + max_reconnect_attempts=max_reconnect_attempts, + disconnected_cb=disconnected_cb, + reconnected_cb=reconnected_cb, + error_cb=error_cb, + closed_cb=closed_cb, + user_credentials=args.creds if args.creds else None, + user=args.user, + password=args.password, + token=args.token, + tls=ssl_ctx, + ) + + async def message_handler(msg: nats.aio.msg.Msg): + payload = msg.data + now = ts() + meta = f'subject="{msg.subject}"' + if msg.reply: + meta += f' reply="{msg.reply}"' + meta += f" size={len(payload)}B" + + if getattr(args, "print_raw", False): + sys.stdout.write(f"[{now}] {meta} bytes={payload!r}\n") + sys.stdout.flush() + return + + # Try to decode as UTF-8 and pretty print JSON if applicable + try: + text = payload.decode("utf-8", errors="replace") + try: + obj = json.loads(text) + pretty = json.dumps(obj, indent=2, ensure_ascii=False, sort_keys=True) + print(f"[{now}] {meta}\n{pretty}\n", flush=True) + except json.JSONDecodeError: + print(f"[{now}] {meta}\n{text}\n", flush=True) + except Exception: + print(f"[{now}] {meta} (non-UTF8) bytes={payload!r}\n", flush=True) + + if args.queue: + await nc.subscribe(args.subject, queue=args.queue, cb=message_handler) + else: + await nc.subscribe(args.subject, cb=message_handler) + + print(f"[{ts()}] Listening on subject '{args.subject}' (queue={getattr(args, 'queue', None) or '-'}) via {args.servers}", file=sys.stderr, flush=True) + + # Graceful shutdown + stop_event = asyncio.Event() + + def handle_signal(*_): + print(f"[{ts()}] Received stop signal, draining...", file=sys.stderr, flush=True) + stop_event.set() + + loop = asyncio.get_running_loop() + for s in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(s, handle_signal) + except NotImplementedError: + # Windows + signal.signal(s, lambda *_: handle_signal()) + + await stop_event.wait() + try: + await nc.drain() + finally: + await nc.close() + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + pass