mirror of
git://f0xx.org/ac/ac-deploy
synced 2026-07-29 05:59:05 +03:00
372 lines
15 KiB
Python
372 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Alertmanager webhook receiver — alert-notifier.py
|
||
Runs on cast04:9099. Receives JSON from Alertmanager, then for each batch:
|
||
1. Shortens the alert URL via the project URL shortener (s.f0xx.org)
|
||
2. Fetches QR PNG from the shortener's built-in QR endpoint
|
||
3. Sends HTML e-mail with the QR embedded as inline image
|
||
4. Sends Telegram message with the QR image attached
|
||
|
||
Config is read from alert-notifier.env in the same directory as this script.
|
||
"""
|
||
|
||
import http.server
|
||
import json
|
||
import smtplib
|
||
import ssl
|
||
import subprocess
|
||
import urllib.request
|
||
import urllib.parse
|
||
import email.mime.multipart
|
||
import email.mime.text
|
||
import email.mime.image
|
||
import email.utils
|
||
import logging
|
||
import os
|
||
import sys
|
||
import pathlib
|
||
|
||
# ── Config defaults (overridden by .env file) ───────────────────────────────
|
||
SHORTENER_INTERNAL = os.environ.get('SHORTENER_INTERNAL', 'http://10.7.16.128')
|
||
SHORTENER_HOST_HDR = os.environ.get('SHORTENER_HOST_HDR', 's.f0xx.org')
|
||
SHORTENER_BEARER = os.environ.get('SHORTENER_BEARER', '')
|
||
SHORTENER_TTL = int(os.environ.get('SHORTENER_TTL', '86400')) # 1 day default
|
||
|
||
SMTP_HOST = os.environ.get('SMTP_HOST', 'smtp.gmail.com')
|
||
SMTP_PORT = int(os.environ.get('SMTP_PORT', '587'))
|
||
SMTP_USER = os.environ.get('SMTP_USER', 'bestcastr@gmail.com')
|
||
SMTP_PASS = os.environ.get('SMTP_PASS', '')
|
||
EMAIL_FROM = os.environ.get('EMAIL_FROM', 'Android Cast Alerts <bestcastr@gmail.com>')
|
||
EMAIL_TO = os.environ.get('EMAIL_TO', 'a.afanasieff@gmail.com')
|
||
|
||
TELEGRAM_BOT = os.environ.get('TELEGRAM_BOT', '')
|
||
TELEGRAM_CHAT = int(os.environ.get('TELEGRAM_CHAT', '0'))
|
||
|
||
LISTEN_PORT = int(os.environ.get('LISTEN_PORT', '9099'))
|
||
|
||
# Default URL to shorten when no alert-specific URL is available
|
||
FALLBACK_URL = os.environ.get(
|
||
'FALLBACK_URL',
|
||
'https://apps.f0xx.org/app/androidcast_project/alertmanager/'
|
||
)
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s %(levelname)s %(message)s',
|
||
stream=sys.stdout,
|
||
)
|
||
log = logging.getLogger('alert-notifier')
|
||
|
||
|
||
# ── URL shortener ────────────────────────────────────────────────────────────
|
||
|
||
def shorten(url: str) -> dict:
|
||
"""
|
||
POST the destination URL (no ?src=qr) to the project URL shortener.
|
||
Returns dict with keys:
|
||
short_url – plain https://s.f0xx.org/SLUG (used as click-link in email/TG)
|
||
qr_url – https://s.f0xx.org/SLUG?src=qr (encoded into the QR image)
|
||
slug – bare slug
|
||
Raises on failure.
|
||
"""
|
||
payload = json.dumps({
|
||
'url': url,
|
||
'bearer': SHORTENER_BEARER,
|
||
'ttl': SHORTENER_TTL,
|
||
}).encode()
|
||
req = urllib.request.Request(
|
||
f'{SHORTENER_INTERNAL}/api/v1/shorten',
|
||
data=payload,
|
||
headers={
|
||
'Content-Type': 'application/json',
|
||
'Host': SHORTENER_HOST_HDR,
|
||
},
|
||
method='POST',
|
||
)
|
||
with urllib.request.urlopen(req, timeout=8) as resp:
|
||
data = json.loads(resp.read())
|
||
if data.get('code', '') != '0':
|
||
raise RuntimeError(f"shortener error: {data}")
|
||
short_url = data['u']
|
||
slug = short_url.rsplit('/', 1)[-1]
|
||
# The QR image encodes the short URL WITH ?src=qr so every mobile scan
|
||
# is counted separately from normal link clicks on the backend.
|
||
qr_url = short_url + ('&src=qr' if '?' in short_url else '?src=qr')
|
||
return {
|
||
'short_url': short_url,
|
||
'qr_url': qr_url,
|
||
'slug': slug,
|
||
}
|
||
|
||
|
||
def make_qr_png(url: str) -> bytes:
|
||
"""
|
||
Generate a QR code PNG for *url* using the system qrencode binary.
|
||
Returns raw PNG bytes. Raises on failure.
|
||
"""
|
||
result = subprocess.run(
|
||
['qrencode', '-t', 'PNG', '-o', '-', '-s', '6', '--', url],
|
||
capture_output=True,
|
||
check=True,
|
||
)
|
||
return result.stdout
|
||
|
||
|
||
# ── E-mail ───────────────────────────────────────────────────────────────────
|
||
|
||
SEVERITY_EMOJI = {
|
||
'critical': '🔴',
|
||
'warning': '🟡',
|
||
'info': '🔵',
|
||
}
|
||
|
||
STATUS_EMOJI = {
|
||
'firing': '🔥',
|
||
'resolved': '✅',
|
||
}
|
||
|
||
|
||
def _alert_html_row(alert: dict) -> str:
|
||
status = alert.get('status', 'firing')
|
||
labels = alert.get('labels', {})
|
||
anns = alert.get('annotations', {})
|
||
name = labels.get('alertname', '(unknown)')
|
||
instance = labels.get('instance', '')
|
||
severity = labels.get('severity', '')
|
||
summary = anns.get('summary', name)
|
||
desc = anns.get('description', '')
|
||
starts = alert.get('startsAt', '')[:19].replace('T', ' ')
|
||
s_emoji = STATUS_EMOJI.get(status, '')
|
||
sv_emoji = SEVERITY_EMOJI.get(severity, '')
|
||
color = '#c0392b' if severity == 'critical' else '#e67e22' if severity == 'warning' else '#2980b9'
|
||
return f"""
|
||
<tr>
|
||
<td style="padding:8px;border-bottom:1px solid #eee;">
|
||
<b style="color:{color};">{s_emoji} {sv_emoji} {status.upper()}</b>
|
||
— <b>{name}</b>
|
||
{f'<span style="color:#666;font-size:0.9em"> @ {instance}</span>' if instance else ''}
|
||
</td>
|
||
</tr>
|
||
<tr>
|
||
<td style="padding:4px 8px 12px 8px;border-bottom:1px solid #eee;color:#333;">
|
||
{f'<b>Summary:</b> {summary}<br>' if summary else ''}
|
||
{f'<b>Detail:</b> {desc}<br>' if desc else ''}
|
||
{f'<b>Fired:</b> {starts} UTC' if starts else ''}
|
||
</td>
|
||
</tr>"""
|
||
|
||
|
||
def send_email(subject: str, alerts: list, short_url: str, qr_png: bytes) -> None:
|
||
msg = email.mime.multipart.MIMEMultipart('related')
|
||
msg['From'] = EMAIL_FROM
|
||
msg['To'] = EMAIL_TO
|
||
msg['Subject'] = subject
|
||
msg['Date'] = email.utils.formatdate(localtime=False)
|
||
msg['Message-ID'] = email.utils.make_msgid(domain='cast04.intra.raptor.org')
|
||
|
||
rows = ''.join(_alert_html_row(a) for a in alerts)
|
||
|
||
html = f"""<!DOCTYPE html>
|
||
<html><body style="font-family:Arial,sans-serif;max-width:640px;margin:auto;">
|
||
<div style="background:#1a1a2e;padding:16px;border-radius:6px 6px 0 0;">
|
||
<h2 style="color:#fff;margin:0;">Android Cast — Monitoring Alert</h2>
|
||
</div>
|
||
<div style="border:1px solid #ddd;border-top:none;padding:16px;border-radius:0 0 6px 6px;">
|
||
<table width="100%" cellpadding="0" cellspacing="0">{rows}</table>
|
||
<hr style="margin:16px 0;">
|
||
<p style="margin:0;">
|
||
<b>View in Alertmanager:</b>
|
||
<a href="{short_url}">{short_url}</a>
|
||
</p>
|
||
<p style="margin:8px 0 4px 0;color:#555;font-size:0.85em;">
|
||
Scan QR to open on mobile (tracked via url-shortener):
|
||
</p>
|
||
<img src="cid:alert_qr" alt="QR code" width="200" height="200"
|
||
style="display:block;border:1px solid #ccc;padding:4px;border-radius:4px;">
|
||
</div>
|
||
<p style="color:#999;font-size:0.75em;text-align:center;margin-top:8px;">
|
||
cast04 monitoring · <a href="https://apps.f0xx.org/app/androidcast_project/monitor/">Grafana</a>
|
||
</p>
|
||
</body></html>"""
|
||
|
||
alternative = email.mime.multipart.MIMEMultipart('alternative')
|
||
alternative.attach(email.mime.text.MIMEText(html, 'html', 'utf-8'))
|
||
msg.attach(alternative)
|
||
|
||
qr_img = email.mime.image.MIMEImage(qr_png, 'png')
|
||
qr_img.add_header('Content-ID', '<alert_qr>')
|
||
qr_img.add_header('Content-Disposition', 'inline', filename='alert-qr.png')
|
||
msg.attach(qr_img)
|
||
|
||
ctx = ssl.create_default_context()
|
||
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as s:
|
||
s.ehlo()
|
||
s.starttls(context=ctx)
|
||
s.ehlo()
|
||
s.login(SMTP_USER, SMTP_PASS)
|
||
s.sendmail(SMTP_USER, EMAIL_TO, msg.as_bytes())
|
||
log.info('email sent to %s subject=%r', EMAIL_TO, subject)
|
||
|
||
|
||
# ── Telegram ─────────────────────────────────────────────────────────────────
|
||
|
||
def send_telegram(text: str, qr_png: bytes) -> None:
|
||
if not TELEGRAM_BOT or not TELEGRAM_CHAT:
|
||
return
|
||
|
||
import io, email.generator
|
||
|
||
boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW'
|
||
|
||
# Build multipart/form-data body manually
|
||
body = (
|
||
f'--{boundary}\r\n'
|
||
f'Content-Disposition: form-data; name="chat_id"\r\n\r\n'
|
||
f'{TELEGRAM_CHAT}\r\n'
|
||
f'--{boundary}\r\n'
|
||
f'Content-Disposition: form-data; name="caption"\r\n\r\n'
|
||
f'{text}\r\n'
|
||
f'--{boundary}\r\n'
|
||
f'Content-Disposition: form-data; name="parse_mode"\r\n\r\n'
|
||
f'HTML\r\n'
|
||
f'--{boundary}\r\n'
|
||
f'Content-Disposition: form-data; name="photo"; filename="alert-qr.png"\r\n'
|
||
f'Content-Type: image/png\r\n\r\n'
|
||
).encode() + qr_png + f'\r\n--{boundary}--\r\n'.encode()
|
||
|
||
req = urllib.request.Request(
|
||
f'https://api.telegram.org/bot{TELEGRAM_BOT}/sendPhoto',
|
||
data=body,
|
||
headers={'Content-Type': f'multipart/form-data; boundary={boundary}'},
|
||
)
|
||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||
result = json.loads(resp.read())
|
||
if not result.get('ok'):
|
||
raise RuntimeError(f"Telegram error: {result}")
|
||
log.info('telegram photo sent to chat_id=%s', TELEGRAM_CHAT)
|
||
|
||
|
||
# ── HTTP handler ─────────────────────────────────────────────────────────────
|
||
|
||
class WebhookHandler(http.server.BaseHTTPRequestHandler):
|
||
def log_message(self, fmt, *args): # suppress default access log
|
||
log.debug(fmt, *args)
|
||
|
||
def do_POST(self):
|
||
if self.path != '/webhook':
|
||
self.send_response(404)
|
||
self.end_headers()
|
||
return
|
||
|
||
length = int(self.headers.get('Content-Length', 0))
|
||
body = self.rfile.read(length)
|
||
try:
|
||
payload = json.loads(body)
|
||
except json.JSONDecodeError:
|
||
self.send_response(400)
|
||
self.end_headers()
|
||
return
|
||
|
||
self.send_response(200)
|
||
self.end_headers()
|
||
|
||
# Process asynchronously (we already returned 200)
|
||
try:
|
||
self._handle(payload)
|
||
except Exception as exc:
|
||
log.error('webhook handler error: %s', exc, exc_info=True)
|
||
|
||
def _handle(self, payload: dict) -> None:
|
||
alerts = payload.get('alerts', [])
|
||
status = payload.get('status', 'firing')
|
||
ext_url = payload.get('externalURL', FALLBACK_URL)
|
||
|
||
if not alerts:
|
||
return
|
||
|
||
# ── shorten & QR ────────────────────────────────────────────────────
|
||
try:
|
||
short = shorten(ext_url)
|
||
# QR image encodes short_url?src=qr — generated locally via qrencode
|
||
qr_png = make_qr_png(short['qr_url'])
|
||
log.info('shortened %s → %s qr_url=%s (%d B)',
|
||
ext_url, short['short_url'], short['qr_url'], len(qr_png))
|
||
except Exception as exc:
|
||
log.error('shorten/qr failed: %s', exc)
|
||
return
|
||
|
||
# ── subject / Telegram text ──────────────────────────────────────────
|
||
firing_names = [a['labels'].get('alertname', '?') for a in alerts if a.get('status') == 'firing']
|
||
resolved_names = [a['labels'].get('alertname', '?') for a in alerts if a.get('status') == 'resolved']
|
||
|
||
if status == 'resolved':
|
||
subject = f'[RESOLVED] {", ".join(resolved_names or firing_names)}'
|
||
else:
|
||
subject = f'[ALERT] {", ".join(firing_names or [a["labels"].get("alertname","?") for a in alerts])}'
|
||
|
||
tg_lines = []
|
||
for a in alerts:
|
||
s = a.get('status', 'firing')
|
||
lbl = a.get('labels', {})
|
||
ann = a.get('annotations', {})
|
||
emoji = STATUS_EMOJI.get(s, '') + ' ' + SEVERITY_EMOJI.get(lbl.get('severity',''), '')
|
||
tg_lines.append(
|
||
f'{emoji} <b>{s.upper()}</b> {lbl.get("alertname","?")} @ {lbl.get("instance","")}\n'
|
||
f'{ann.get("summary","")}\n'
|
||
f'{ann.get("description","")}'
|
||
)
|
||
|
||
tg_text = '\n\n'.join(tg_lines) + f'\n\n🔗 {short["short_url"]}'
|
||
|
||
# ── send ─────────────────────────────────────────────────────────────
|
||
try:
|
||
send_email(subject, alerts, short['short_url'], qr_png)
|
||
except Exception as exc:
|
||
log.error('email failed: %s', exc)
|
||
|
||
try:
|
||
send_telegram(tg_text, qr_png)
|
||
except Exception as exc:
|
||
log.error('telegram failed: %s', exc)
|
||
|
||
|
||
# ── Entry point ───────────────────────────────────────────────────────────────
|
||
|
||
def load_env(path: str) -> None:
|
||
"""Load KEY=VALUE pairs from an .env file into os.environ."""
|
||
try:
|
||
with open(path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith('#') or '=' not in line:
|
||
continue
|
||
k, _, v = line.partition('=')
|
||
os.environ.setdefault(k.strip(), v.strip())
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
|
||
if __name__ == '__main__':
|
||
env_file = pathlib.Path(__file__).with_suffix('.env')
|
||
load_env(str(env_file))
|
||
|
||
# Re-read globals that may have been overridden by .env
|
||
SHORTENER_INTERNAL = os.environ.get('SHORTENER_INTERNAL', SHORTENER_INTERNAL)
|
||
SHORTENER_HOST_HDR = os.environ.get('SHORTENER_HOST_HDR', SHORTENER_HOST_HDR)
|
||
SHORTENER_BEARER = os.environ.get('SHORTENER_BEARER', SHORTENER_BEARER)
|
||
SHORTENER_TTL = int(os.environ.get('SHORTENER_TTL', str(SHORTENER_TTL)))
|
||
SMTP_HOST = os.environ.get('SMTP_HOST', SMTP_HOST)
|
||
SMTP_PORT = int(os.environ.get('SMTP_PORT', str(SMTP_PORT)))
|
||
SMTP_USER = os.environ.get('SMTP_USER', SMTP_USER)
|
||
SMTP_PASS = os.environ.get('SMTP_PASS', SMTP_PASS)
|
||
EMAIL_FROM = os.environ.get('EMAIL_FROM', EMAIL_FROM)
|
||
EMAIL_TO = os.environ.get('EMAIL_TO', EMAIL_TO)
|
||
TELEGRAM_BOT = os.environ.get('TELEGRAM_BOT', TELEGRAM_BOT)
|
||
TELEGRAM_CHAT = int(os.environ.get('TELEGRAM_CHAT', str(TELEGRAM_CHAT)))
|
||
LISTEN_PORT = int(os.environ.get('LISTEN_PORT', str(LISTEN_PORT)))
|
||
FALLBACK_URL = os.environ.get('FALLBACK_URL', FALLBACK_URL)
|
||
|
||
server = http.server.HTTPServer(('127.0.0.1', LISTEN_PORT), WebhookHandler)
|
||
log.info('alert-notifier listening on 127.0.0.1:%d', LISTEN_PORT)
|
||
server.serve_forever()
|