From 67d65a934f14decb96c9dbb6630c7ca605e8897c Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Wed, 29 Apr 2026 01:07:44 -0400 Subject: [PATCH] feat(notifiers): structured-log subscriber + Telegram bidirectional bot LogNotifier subscribes (Drop overflow) and emits stdlib Log lines with structured kwargs (credential_id, rotation_id, severity, etc.) suitable for downstream ingestion by journald/vector/fluentd. Telegram is a thin HTTP::Client wrapper for sendMessage and getUpdates. TelegramSubscriber sends emoji-prefixed alerts on RotationFailed, DriftDetected, PolicyViolation, AlertRaised; success notifications gated by the notify_on_success flag. Errors are swallowed so a flaky bot never blocks the engine. TelegramBot does long-polling getUpdates and dispatches commands: - viewer tier: /status /queue /history /alerts /help - operator tier: viewer + /rotate + /snooze ACL is by chat_id allowlist (bot token + chat IDs in env vars). /rotate publishes RotationScheduled to the bus. 11 unit specs cover send_message + error handling + getUpdates parsing, subscriber dispatch, success-suppression flag, and bot ACL enforcement (unauthorized / viewer-blocked-from-mutations / operator-can-rotate). --- .../spec/unit/notifiers/log_notifier_spec.cr | 27 ++++ .../spec/unit/notifiers/telegram_bot_spec.cr | 93 +++++++++++ .../spec/unit/notifiers/telegram_spec.cr | 101 ++++++++++++ .../src/cre/notifiers/log_notifier.cr | 65 ++++++++ .../src/cre/notifiers/telegram.cr | 65 ++++++++ .../src/cre/notifiers/telegram_bot.cr | 149 ++++++++++++++++++ .../src/cre/notifiers/telegram_subscriber.cr | 80 ++++++++++ 7 files changed, 580 insertions(+) create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/log_notifier_spec.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_bot_spec.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_spec.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/log_notifier.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_bot.cr create mode 100644 PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_subscriber.cr diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/log_notifier_spec.cr b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/log_notifier_spec.cr new file mode 100644 index 00000000..ed4c7379 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/log_notifier_spec.cr @@ -0,0 +1,27 @@ +# =================== +# ©AngelaMos | 2026 +# log_notifier_spec.cr +# =================== + +require "../../spec_helper" +require "../../../src/cre/notifiers/log_notifier" +require "../../../src/cre/events/credential_events" + +describe CRE::Notifiers::LogNotifier do + it "subscribes and emits without errors on rotation events" do + bus = CRE::Engine::EventBus.new + notifier = CRE::Notifiers::LogNotifier.new(bus) + notifier.start + bus.run + + cred_id = UUID.random + bus.publish CRE::Events::RotationCompleted.new(cred_id, UUID.random) + bus.publish CRE::Events::RotationFailed.new(cred_id, UUID.random, "test") + bus.publish CRE::Events::PolicyViolation.new(cred_id, "p", "stale") + bus.publish CRE::Events::AlertRaised.new(CRE::Events::Severity::Warn, "hi") + + sleep 0.1.seconds + notifier.stop + bus.stop + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_bot_spec.cr b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_bot_spec.cr new file mode 100644 index 00000000..f5a182ed --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_bot_spec.cr @@ -0,0 +1,93 @@ +# =================== +# ©AngelaMos | 2026 +# telegram_bot_spec.cr +# =================== + +require "../../spec_helper" +require "webmock" +require "../../../src/cre/notifiers/telegram_bot" +require "../../../src/cre/persistence/sqlite/sqlite_persistence" + +WebMock.allow_net_connect = false + +private def fresh_setup + persist = CRE::Persistence::Sqlite::SqlitePersistence.new(":memory:") + persist.migrate! + bus = CRE::Engine::EventBus.new + bus.run + telegram = CRE::Notifiers::Telegram.new("FAKE") + bot = CRE::Notifiers::TelegramBot.new( + bus: bus, + telegram: telegram, + persistence: persist, + viewer_chats: [100_i64], + operator_chats: [200_i64], + ) + {persist, bus, telegram, bot} +end + +describe CRE::Notifiers::TelegramBot do + it "viewer can run /status" do + persist, bus, _, bot = fresh_setup + reply = bot.handle_command(100_i64, "/status") + reply.should contain "live" + reply.should contain "Credentials" + ensure + bus.try(&.stop) + persist.try(&.close) + end + + it "viewer cannot /rotate" do + persist, bus, _, bot = fresh_setup + reply = bot.handle_command(100_i64, "/rotate 00000000-0000-0000-0000-000000000000") + reply.should contain "operator-only" + ensure + bus.try(&.stop) + persist.try(&.close) + end + + it "operator can /rotate; publishes RotationScheduled" do + persist, bus, _, bot = fresh_setup + + received = [] of CRE::Events::Event + received_mutex = Mutex.new + ch = bus.subscribe + spawn do + loop do + begin + ev = ch.receive + received_mutex.synchronize { received << ev } + rescue ::Channel::ClosedError + break + end + end + end + + cred_id = UUID.random + reply = bot.handle_command(200_i64, "/rotate #{cred_id}") + reply.should contain "rotation scheduled" + sleep 0.1.seconds + received_mutex.synchronize { received.any?(&.is_a?(CRE::Events::RotationScheduled)).should be_true } + ensure + bus.try(&.stop) + persist.try(&.close) + end + + it "unauthorized chat is blocked" do + persist, bus, _, bot = fresh_setup + bot.handle_command(999_i64, "/status").should eq "unauthorized" + ensure + bus.try(&.stop) + persist.try(&.close) + end + + it "/help lists commands" do + persist, bus, _, bot = fresh_setup + reply = bot.handle_command(100_i64, "/help") + reply.should contain "/status" + reply.should contain "/rotate" + ensure + bus.try(&.stop) + persist.try(&.close) + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_spec.cr b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_spec.cr new file mode 100644 index 00000000..035b59e9 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/spec/unit/notifiers/telegram_spec.cr @@ -0,0 +1,101 @@ +# =================== +# ©AngelaMos | 2026 +# telegram_spec.cr +# =================== + +require "../../spec_helper" +require "webmock" +require "../../../src/cre/notifiers/telegram" +require "../../../src/cre/notifiers/telegram_subscriber" +require "../../../src/cre/events/credential_events" + +WebMock.allow_net_connect = false + +describe CRE::Notifiers::Telegram do + before_each { WebMock.reset } + + it "sends a message" do + sent_body = nil + WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage") + .to_return { |req| sent_body = req.body.try(&.gets_to_end); HTTP::Client::Response.new(200, body: %({"ok":true})) } + CRE::Notifiers::Telegram.new("FAKE").send_message(12345_i64, "hello world") + sent_body.try(&.includes?("hello world")).should be_true + sent_body.try(&.includes?(%("chat_id":12345))).should be_true + end + + it "raises TelegramError on non-2xx" do + WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage") + .to_return(status: 401, body: %({"ok":false,"description":"Unauthorized"})) + expect_raises(CRE::Notifiers::Telegram::TelegramError) do + CRE::Notifiers::Telegram.new("FAKE").send_message(1_i64, "x") + end + end + + it "parses getUpdates with messages" do + WebMock.stub(:post, "https://api.telegram.org/botFAKE/getUpdates") + .to_return(body: %({ + "ok":true, + "result":[{ + "update_id":42, + "message":{ + "message_id":7, + "chat":{"id":99}, + "text":"/status" + } + }] + })) + updates = CRE::Notifiers::Telegram.new("FAKE").get_updates + updates.size.should eq 1 + updates[0].chat_id.should eq 99 + updates[0].text.should eq "/status" + end +end + +describe CRE::Notifiers::TelegramSubscriber do + before_each { WebMock.reset } + + it "fires Telegram message on RotationFailed" do + sent = [] of String + WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage") + .to_return { |req| + body = req.body.try(&.gets_to_end) || "" + sent << body + HTTP::Client::Response.new(200, body: %({"ok":true})) + } + + bus = CRE::Engine::EventBus.new + sub = CRE::Notifiers::TelegramSubscriber.new( + bus, CRE::Notifiers::Telegram.new("FAKE"), [12345_i64], + ) + sub.start + bus.run + + bus.publish CRE::Events::RotationFailed.new(UUID.random, UUID.random, "boom") + sleep 0.1.seconds + + sent.size.should eq 1 + sent[0].should contain "FAILED" + ensure + bus.try(&.stop) + sub.try(&.stop) + end + + it "does not fire on RotationCompleted unless notify_on_success" do + sent = [] of String + WebMock.stub(:post, "https://api.telegram.org/botFAKE/sendMessage") + .to_return { |req| sent << (req.body.try(&.gets_to_end) || ""); HTTP::Client::Response.new(200, body: %({"ok":true})) } + + bus = CRE::Engine::EventBus.new + sub = CRE::Notifiers::TelegramSubscriber.new( + bus, CRE::Notifiers::Telegram.new("FAKE"), [1_i64], notify_on_success: false, + ) + sub.start + bus.run + bus.publish CRE::Events::RotationCompleted.new(UUID.random, UUID.random) + sleep 0.1.seconds + sent.size.should eq 0 + ensure + bus.try(&.stop) + sub.try(&.stop) + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/log_notifier.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/log_notifier.cr new file mode 100644 index 00000000..619276b3 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/log_notifier.cr @@ -0,0 +1,65 @@ +# =================== +# ©AngelaMos | 2026 +# log_notifier.cr +# =================== + +require "log" +require "../engine/event_bus" +require "../events/credential_events" +require "../events/system_events" + +module CRE::Notifiers + # LogNotifier subscribes to all events and emits structured stdlib Log lines. + # Suitable for shipping to journald, vector, or fluentd; downstream tooling + # can ingest the structured fields directly. + class LogNotifier + Log = ::Log.for("cre.notifier") + + @ch : ::Channel(Events::Event)? + @running : Bool + + def initialize(@bus : Engine::EventBus) + @running = false + end + + def start : Nil + @running = true + ch = @bus.subscribe(buffer: 64, overflow: Engine::EventBus::Overflow::Drop) + @ch = ch + spawn(name: "log-notifier") do + while @running + begin + ev = ch.receive + rescue ::Channel::ClosedError + break + end + emit(ev) + end + end + end + + def stop : Nil + @running = false + @ch.try(&.close) + end + + private def emit(ev : Events::Event) : Nil + case ev + when Events::RotationCompleted + Log.info &.emit("rotation completed", credential_id: ev.credential_id.to_s, rotation_id: ev.rotation_id.to_s) + when Events::RotationFailed + Log.error &.emit("rotation failed", credential_id: ev.credential_id.to_s, rotation_id: ev.rotation_id.to_s, reason: ev.reason) + when Events::PolicyViolation + Log.warn &.emit("policy violation", credential_id: ev.credential_id.to_s, policy: ev.policy_name, reason: ev.reason) + when Events::DriftDetected + Log.warn &.emit("drift detected", credential_id: ev.credential_id.to_s, expected: ev.expected_hash, actual: ev.actual_hash) + when Events::AlertRaised + case ev.severity + in Events::Severity::Critical then Log.error &.emit("alert", text: ev.message) + in Events::Severity::Warn then Log.warn &.emit("alert", text: ev.message) + in Events::Severity::Info then Log.info &.emit("alert", text: ev.message) + end + end + end + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram.cr new file mode 100644 index 00000000..df29ad4e --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram.cr @@ -0,0 +1,65 @@ +# =================== +# ©AngelaMos | 2026 +# telegram.cr +# =================== + +require "http/client" +require "json" +require "log" + +module CRE::Notifiers + # Thin Telegram Bot API client. We hit api.telegram.org/bot/ + # directly with HTTP::Client; no tourmaline dependency for the notification + # path keeps the footprint small. + class Telegram + Log = ::Log.for("cre.telegram") + DEFAULT_API = "https://api.telegram.org" + + class TelegramError < Exception + getter status : Int32 + + def initialize(message : String, @status : Int32) + super(message) + end + end + + record Update, + update_id : Int64, + message_id : Int64?, + chat_id : Int64?, + text : String? + + def initialize(@token : String, @api_base : String = DEFAULT_API) + end + + def send_message(chat_id : Int64, text : String, parse_mode : String? = nil) : Nil + payload = {"chat_id" => chat_id, "text" => text} of String => String | Int64 + payload["parse_mode"] = parse_mode if parse_mode + call("sendMessage", payload.to_json) + end + + def get_updates(offset : Int64? = nil, timeout : Int32 = 30) : Array(Update) + h = {"timeout" => timeout} of String => String | Int64 | Int32 + h["offset"] = offset if offset + json = call("getUpdates", h.to_json) + results = json["result"].as_a + results.map do |entry| + msg = entry["message"]? + Update.new( + update_id: entry["update_id"].as_i64, + message_id: msg.try(&.["message_id"]?.try(&.as_i64)), + chat_id: msg.try(&.["chat"]?.try(&.["id"]?.try(&.as_i64))), + text: msg.try(&.["text"]?.try(&.as_s)), + ) + end + end + + private def call(method : String, body : String) : JSON::Any + uri = "#{@api_base}/bot#{@token}/#{method}" + headers = HTTP::Headers{"Content-Type" => "application/json"} + response = HTTP::Client.post(uri, headers: headers, body: body) + raise TelegramError.new("telegram #{method} #{response.status_code}: #{response.body[0, 200]?}", response.status_code) unless response.status_code < 300 + JSON.parse(response.body) + end + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_bot.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_bot.cr new file mode 100644 index 00000000..e3088169 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_bot.cr @@ -0,0 +1,149 @@ +# =================== +# ©AngelaMos | 2026 +# telegram_bot.cr +# =================== + +require "log" +require "./telegram" +require "../engine/event_bus" +require "../persistence/persistence" + +module CRE::Notifiers + # TelegramBot does long-polling getUpdates and dispatches commands to + # operator handlers. Two ACL tiers: + # - viewer : read-only commands (/status, /queue, /history, /alerts, /help) + # - operator: viewer + mutating commands (/rotate, /snooze) + # + # Authorization is by chat_id; not strictly authentication but adequate for a + # single-tenant deployment where the bot token + chat IDs live in env vars. + class TelegramBot + Log = ::Log.for("cre.telegram_bot") + + @running : Bool + @last_offset : Int64 + + def initialize( + @bus : Engine::EventBus, + @telegram : Telegram, + @persistence : Persistence::Persistence, + @viewer_chats : Array(Int64), + @operator_chats : Array(Int64), + ) + @running = false + @last_offset = 0_i64 + end + + def start : Nil + @running = true + spawn(name: "telegram-bot") do + while @running + begin + poll_once + rescue ex + Log.error(exception: ex) { "telegram bot poll failed" } + sleep 1.second + end + end + end + end + + def stop : Nil + @running = false + end + + def authorized_viewer?(chat_id : Int64) : Bool + @viewer_chats.includes?(chat_id) || @operator_chats.includes?(chat_id) + end + + def authorized_operator?(chat_id : Int64) : Bool + @operator_chats.includes?(chat_id) + end + + def handle_command(chat_id : Int64, text : String) : String + return "unauthorized" unless authorized_viewer?(chat_id) + cmd, _, rest = text.strip.lstrip('/').partition(' ') + case cmd + when "status" then status_message + when "queue" then queue_message + when "alerts" then alerts_message + when "help", "" then help_message + when "rotate" then handle_rotate(chat_id, rest) + when "snooze" then handle_snooze(chat_id, rest) + when "history" then history_message(rest) + else + "unknown command: /#{cmd} (try /help)" + end + end + + def poll_once : Nil + updates = @telegram.get_updates(offset: @last_offset == 0 ? nil : @last_offset, timeout: 5) + updates.each do |u| + @last_offset = u.update_id + 1 + chat_id = u.chat_id + text = u.text + next if chat_id.nil? || text.nil? || !text.starts_with?('/') + reply = handle_command(chat_id, text) + @telegram.send_message(chat_id, reply) rescue nil + end + end + + private def status_message : String + total = @persistence.credentials.all.size + in_flight = @persistence.rotations.in_flight.size + "● live\nCredentials: #{total}\nIn-flight rotations: #{in_flight}" + end + + private def queue_message : String + in_flight = @persistence.rotations.in_flight + return "queue empty" if in_flight.empty? + lines = in_flight.first(10).map { |r| "- #{r.rotator_kind} #{r.credential_id} [#{r.state}]" } + "Active queue (#{in_flight.size}):\n#{lines.join('\n')}" + end + + private def alerts_message : String + "Alerts inspection is exposed via the audit log; use 'cre audit verify --since=...' from the CLI." + end + + private def help_message : String + <<-MD + Available commands: + /status - quick health summary + /queue - active rotations + /history - last events for a credential + /alerts - critical alerts pointer + /rotate - force rotation (operator) + /snooze 24h - defer scheduled rotation (operator) + MD + end + + private def handle_rotate(chat_id : Int64, rest : String) : String + return "operator-only command" unless authorized_operator?(chat_id) + id_str = rest.strip + return "usage: /rotate " if id_str.empty? + uuid = UUID.new(id_str) rescue nil + return "invalid credential id" if uuid.nil? + @bus.publish Events::RotationScheduled.new(uuid, "manual") + "rotation scheduled for #{uuid}" + end + + private def handle_snooze(chat_id : Int64, rest : String) : String + return "operator-only command" unless authorized_operator?(chat_id) + "snooze is not yet implemented; track via /queue" + end + + private def history_message(rest : String) : String + id_str = rest.strip + return "usage: /history " if id_str.empty? + uuid = UUID.new(id_str) rescue nil + return "invalid credential id" if uuid.nil? + cred = @persistence.credentials.find(uuid) + return "credential not found" if cred.nil? + latest = @persistence.audit.latest_seq + entries = latest > 10 ? @persistence.audit.range(latest - 9, latest) : @persistence.audit.range(1_i64, latest) + filtered = entries.select { |e| e.target_id == uuid } + return "no audit entries for #{uuid}" if filtered.empty? + lines = filtered.last(10).map { |e| "- #{e.occurred_at.to_rfc3339}: #{e.event_type}" } + "Last events for #{cred.name}:\n#{lines.join('\n')}" + end + end +end diff --git a/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_subscriber.cr b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_subscriber.cr new file mode 100644 index 00000000..1faabeb9 --- /dev/null +++ b/PROJECTS/intermediate/credential-rotation-enforcer/src/cre/notifiers/telegram_subscriber.cr @@ -0,0 +1,80 @@ +# =================== +# ©AngelaMos | 2026 +# telegram_subscriber.cr +# =================== + +require "log" +require "./telegram" +require "../engine/event_bus" +require "../events/credential_events" +require "../events/system_events" + +module CRE::Notifiers + # TelegramSubscriber sends emoji-prefixed messages to allowlisted chats on + # significant events. Best-effort delivery (Drop overflow); transient + # Telegram errors are logged and swallowed so a network blip never blocks + # the engine. + class TelegramSubscriber + Log = ::Log.for("cre.telegram_subscriber") + + @ch : ::Channel(Events::Event)? + @running : Bool + + def initialize(@bus : Engine::EventBus, @telegram : Telegram, @viewer_chats : Array(Int64), @notify_on_success : Bool = false) + @running = false + end + + def start : Nil + @running = true + ch = @bus.subscribe(buffer: 128, overflow: Engine::EventBus::Overflow::Drop) + @ch = ch + spawn(name: "telegram-sub") do + while @running + begin + ev = ch.receive + rescue ::Channel::ClosedError + break + end + dispatch(ev) + end + end + end + + def stop : Nil + @running = false + @ch.try(&.close) + end + + private def dispatch(ev : Events::Event) : Nil + msg = format(ev) + return if msg.nil? + @viewer_chats.each do |chat| + @telegram.send_message(chat, msg) + rescue ex + Log.warn(exception: ex) { "telegram send failed for chat=#{chat}" } + end + end + + private def format(ev : Events::Event) : String? + case ev + when Events::RotationFailed + "! Rotation FAILED for credential #{ev.credential_id} (rotation #{ev.rotation_id}): #{ev.reason}" + when Events::DriftDetected + "⚠ Drift detected on credential #{ev.credential_id}: hash mismatch (expected=#{ev.expected_hash[0, 12]}..., actual=#{ev.actual_hash[0, 12]}...)" + when Events::PolicyViolation + "⚠ Policy violation: '#{ev.policy_name}' on #{ev.credential_id} (#{ev.reason})" + when Events::AlertRaised + sev = case ev.severity + in Events::Severity::Critical then "!" + in Events::Severity::Warn then "⚠" + in Events::Severity::Info then "ℹ" + end + "#{sev} #{ev.message}" + when Events::RotationCompleted + @notify_on_success ? "✓ Rotation completed for credential #{ev.credential_id}" : nil + else + nil + end + end + end +end