232 lines
7.9 KiB
Python
Executable file
232 lines
7.9 KiB
Python
Executable file
#!/usr/bin/env python3
|
||
"""Send a tagged probe mail via sink SMTP, then IMAP-search INBOX vs Junk/spam.
|
||
|
||
SUT remains Taler mail domains elsewhere; this only exercises mon* sinks
|
||
(and optional g0v) for inbox-vs-spam visibility.
|
||
|
||
Env:
|
||
MAIL_DELIV_IMAP_PASS / MAIL_DELIV_ALERT_PASS required (shared default)
|
||
MAIL_DELIV_PASS_mon001 / _mon002 / _mon003 optional per-mailbox override
|
||
MAIL_DELIV_PROBE_FROM default mon001@hacktivism.ch
|
||
MAIL_DELIV_PROBE_TO default mon001 mon002 mon003 (keep all — odd-looking addrs)
|
||
MAIL_DELIV_PROBE_SMTP default mail.cyon.ch:587
|
||
MAIL_DELIV_IMAP_HOST default mail.cyon.ch
|
||
MAIL_DELIV_PROBE_WAIT_S default 25
|
||
MAIL_DELIV_PROBE_ALERT_TO optional extra alert recipients on spam/miss
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import imaplib
|
||
import os
|
||
import smtplib
|
||
import ssl
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from email.message import EmailMessage
|
||
from pathlib import Path
|
||
|
||
|
||
def _env(name: str, default: str = "") -> str:
|
||
return (os.environ.get(name) or default).strip()
|
||
|
||
|
||
def _pass_for(addr: str) -> str:
|
||
"""Shared pass, or MAIL_DELIV_PASS_<localpart> (e.g. MAIL_DELIV_PASS_mon002)."""
|
||
local = addr.split("@", 1)[0].replace(".", "_").replace("+", "_")
|
||
specific = _env(f"MAIL_DELIV_PASS_{local}") or _env(f"MAIL_DELIV_IMAP_PASS_{local}")
|
||
if specific:
|
||
return specific
|
||
return _env("MAIL_DELIV_IMAP_PASS") or _env("MAIL_DELIV_ALERT_PASS")
|
||
|
||
|
||
def smtp_send(host: str, port: int, user: str, pw: str, to: str, subj: str, body: str, msgid: str) -> None:
|
||
msg = EmailMessage()
|
||
msg["From"] = user
|
||
msg["To"] = to
|
||
msg["Subject"] = subj
|
||
msg["Message-ID"] = msgid
|
||
msg["X-Taler-Mon-Probe"] = msgid.strip("<>")
|
||
msg.set_content(body)
|
||
ctx = ssl.create_default_context()
|
||
with smtplib.SMTP(host, port, timeout=45) as s:
|
||
s.ehlo()
|
||
s.starttls(context=ctx)
|
||
s.ehlo()
|
||
s.login(user, pw)
|
||
s.send_message(msg)
|
||
|
||
|
||
def imap_find(host: str, user: str, pw: str, token: str) -> dict:
|
||
"""Return {folder: count} for folders where token appears in Message-ID/Subject/body."""
|
||
ctx = ssl.create_default_context()
|
||
M = imaplib.IMAP4_SSL(host, 993, ssl_context=ctx)
|
||
M.login(user, pw)
|
||
typ, bl = M.list()
|
||
folders = []
|
||
for raw in bl or []:
|
||
line = raw.decode(errors="replace")
|
||
# LIST (\HasNoChildren) "." INBOX.Junk
|
||
if '"/"' in line or ' "." ' in line or ' "."' in line:
|
||
name = line.split(' "." ')[-1].strip().strip('"')
|
||
elif ' "/" ' in line:
|
||
name = line.split(' "/" ')[-1].strip().strip('"')
|
||
else:
|
||
parts = line.rsplit(" ", 1)
|
||
name = parts[-1].strip().strip('"') if parts else ""
|
||
if name:
|
||
folders.append(name)
|
||
# ensure common names
|
||
for n in ("INBOX", "INBOX.Junk", "INBOX.spam", "Junk", "Spam"):
|
||
if n not in folders:
|
||
folders.append(n)
|
||
|
||
hits: dict[str, int] = {}
|
||
needle = token.encode()
|
||
for folder in folders:
|
||
try:
|
||
typ, _ = M.select(folder, readonly=True)
|
||
if typ != "OK":
|
||
continue
|
||
except Exception:
|
||
continue
|
||
# SEARCH TEXT is broad; prefer HEADER Message-ID
|
||
found = 0
|
||
for crit in (f'(HEADER Message-ID "{token}")', f'(HEADER X-Taler-Mon-Probe "{token}")', f'(SUBJECT "{token}")'):
|
||
try:
|
||
typ, data = M.search(None, crit)
|
||
if typ == "OK" and data and data[0]:
|
||
ids = data[0].split()
|
||
found = max(found, len(ids))
|
||
except Exception:
|
||
continue
|
||
if found:
|
||
hits[folder] = found
|
||
M.logout()
|
||
return hits
|
||
|
||
|
||
def classify(hits: dict[str, int]) -> str:
|
||
if not hits:
|
||
return "missing"
|
||
# prefer INBOX without junk
|
||
junkish = []
|
||
inboxish = []
|
||
for f, n in hits.items():
|
||
fl = f.lower()
|
||
if "junk" in fl or "spam" in fl:
|
||
junkish.append(f)
|
||
elif fl == "inbox" or fl.endswith(".inbox") or fl == "inbox.inbox":
|
||
inboxish.append(f)
|
||
elif "inbox" in fl and "junk" not in fl and "spam" not in fl and "trash" not in fl:
|
||
inboxish.append(f)
|
||
if inboxish and not junkish:
|
||
return "inbox"
|
||
if junkish and not inboxish:
|
||
return "spam"
|
||
if inboxish and junkish:
|
||
return "both"
|
||
return "other:" + ",".join(hits.keys())
|
||
|
||
|
||
def main() -> int:
|
||
frm = _env("MAIL_DELIV_PROBE_FROM", "mon001@hacktivism.ch")
|
||
# Keep mon001–003 even if one address looks odd / one pass differs.
|
||
tos = _env(
|
||
"MAIL_DELIV_PROBE_TO",
|
||
"mon001@hacktivism.ch mon002@hacktivism.ch mon003@hacktivism.ch",
|
||
).split()
|
||
pw_from = _pass_for(frm)
|
||
if not pw_from:
|
||
print("FAIL missing MAIL_DELIV_IMAP_PASS (or per-mailbox MAIL_DELIV_PASS_*)", file=sys.stderr)
|
||
return 2
|
||
smtp = _env("MAIL_DELIV_PROBE_SMTP", "mail.cyon.ch:587")
|
||
imap_host = _env("MAIL_DELIV_IMAP_HOST", "mail.cyon.ch")
|
||
wait_s = float(_env("MAIL_DELIV_PROBE_WAIT_S", "25") or "25")
|
||
host, _, port_s = smtp.partition(":")
|
||
port = int(port_s or "587")
|
||
|
||
token = uuid.uuid4().hex
|
||
msgid = f"<{token}@taler-monitoring.probe>"
|
||
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
||
subj = f"[taler-mon probe] {token[:12]}"
|
||
body = (
|
||
f"taler-monitoring regular mon* probe\n"
|
||
f"token={token}\n"
|
||
f"when={now}\n"
|
||
f"Hallo Hernâni, hallo Florian — Inbox-vs-Spam check.\n"
|
||
)
|
||
|
||
results = []
|
||
sent_ok = []
|
||
for to in tos:
|
||
try:
|
||
smtp_send(host, port, frm, pw_from, to, subj, body, msgid)
|
||
print(f"SENT to={to} token={token[:12]}")
|
||
sent_ok.append(to)
|
||
except Exception as e:
|
||
print(f"SEND_FAIL to={to} {type(e).__name__}: {e}")
|
||
results.append((to, "send_fail", {}))
|
||
|
||
time.sleep(max(5.0, wait_s))
|
||
|
||
rc = 0
|
||
for to in tos:
|
||
pw = _pass_for(to)
|
||
if not pw:
|
||
print(f"IMAP_SKIP to={to} no password (set MAIL_DELIV_PASS_{to.split('@')[0]})")
|
||
results.append((to, "no_pass", {}))
|
||
rc = 1
|
||
continue
|
||
try:
|
||
hits = imap_find(imap_host, to, pw, token)
|
||
kind = classify(hits)
|
||
print(f"IMAP to={to} result={kind} hits={hits}")
|
||
results.append((to, kind, hits))
|
||
if kind != "inbox":
|
||
rc = 1
|
||
except Exception as e:
|
||
# Keep name in rotation; soft note — odd addr / wrong pass still listed
|
||
print(f"IMAP_FAIL to={to} {type(e).__name__}: {e}")
|
||
results.append((to, "imap_fail", {}))
|
||
rc = 1
|
||
|
||
# optional notify alert recipients if any non-inbox
|
||
alert_to = _env("MAIL_DELIV_PROBE_ALERT_TO") or _env("MAIL_DELIV_ALERT_TO")
|
||
bad = [r for r in results if r[1] not in ("inbox",)]
|
||
if alert_to and bad:
|
||
summary = "\n".join(f"- {t}: {k} {h}" for t, k, h in results)
|
||
for ato in alert_to.split():
|
||
try:
|
||
smtp_send(
|
||
host,
|
||
port,
|
||
frm,
|
||
pw_from,
|
||
ato,
|
||
f"[taler-mon probe] spam/miss {token[:12]}",
|
||
f"Probe issues:\n{summary}\n",
|
||
f"<{uuid.uuid4().hex}@taler-monitoring.alert>",
|
||
)
|
||
print(f"ALERT → {ato}")
|
||
except Exception as e:
|
||
print(f"ALERT_FAIL → {ato} {e}")
|
||
|
||
# stamp state
|
||
state = Path(
|
||
_env(
|
||
"TALER_MONITORD_STATE_DIR",
|
||
str(Path.home() / ".local/state/taler-monitord"),
|
||
)
|
||
)
|
||
state.mkdir(parents=True, exist_ok=True)
|
||
(state / "mail-deliv-probe.last").write_text(
|
||
f"ts={now}\ntoken={token}\nrc={rc}\n" + "\n".join(f"{t}\t{k}\t{h}" for t, k, h in results) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
return rc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|