initial codes

2025-10-28 05:31:25 +02:00
parent 96aea02b0c
commit 34d32ec5cb
7 changed files with 472 additions and 0 deletions
--- a/files/entrypoint.sh
+++ 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}" ]_]()]()
--- a/files/healthcheck.sh
+++ 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
--- a/files/nats-registration-listener.sh
+++ 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[@]}"
--- a/files/nats_registration_listener.py
+++ 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