Menu
Public repository / 1c2217cda20adcbcf6a4d0b6e8aa22d39f062555
AkurAI-Mail / scripts/akurai-mail-server
Latest change aa05f301181c865428b6faf9c750f0631324c73a - feat: compute reply-needed, follow-up and priority signals for conversations (AKURAI-MAIL-7) by Ólafur Búi Ólafsson
#!/usr/bin/env python3
import argparse, base64, crypt, email, hashlib, html, json, mailbox, mimetypes, os, re, secrets, shutil, socket, string, struct, subprocess, sys, time
from email.message import EmailMessage
from email.policy import default
from email.utils import formatdate, make_msgid, parseaddr
from pathlib import Path
import ctypes.util, select, urllib.error, urllib.request
STATE = Path("/var/lib/akurai-mail/state.json")
WEBMAIL_STATE = Path("/var/lib/akurai-mail/webmail.json")
DOVECOT_USERS = Path("/etc/dovecot/users")
POSTFIX_VMAILBOX = Path("/etc/postfix/vmailbox")
POSTFIX_VIRTUAL = Path("/etc/postfix/virtual")
DOMAIN = "olibuijr.com"
HOSTNAME = "mail.olibuijr.com"
PUBLIC_IP = "3.94.46.219"
DOMAINS_STATE = Path("/var/lib/akurai-mail/domains.json")
OPENDKIM_KEYTABLE = Path("/etc/opendkim/KeyTable")
OPENDKIM_SIGNINGTABLE = Path("/etc/opendkim/SigningTable")
OPENDKIM_TRUSTEDHOSTS = Path("/etc/opendkim/TrustedHosts")
OPENDKIM_KEYS_DIR = Path("/etc/opendkim/keys")
SPAM_POLICY_DEFAULT = {
"enabled": True,
"autoMove": True,
"threshold": 5,
"blocklist": [],
"allowlist": [],
"domainBlocklist": [],
"subjectKeywords": ["crypto", "lottery", "winner", "urgent payment", "gift card", "wire transfer"],
"attachmentExtensions": ["exe", "scr", "bat", "cmd", "js", "vbs", "ps1", "jar"],
"maxLinks": 10,
"stats": {"scanned": 0, "moved": 0, "lastScan": 0},
}
DEFAULT = {"domain": DOMAIN, "hostname": HOSTNAME, "publicIp": PUBLIC_IP, "users": [], "aliases": [], "events": [], "metricsHistory": [], "spamPolicy": SPAM_POLICY_DEFAULT}
WEBMAIL_DEFAULT = {"contacts": [], "signatures": {}, "vacation": {}, "vacationSent": {}, "rules": [], "preferences": {"density": "comfortable", "theme": "light"}, "audit": []}
FOLDERS = {
"Inbox": "",
"Sent": ".Sent",
"Drafts": ".Drafts",
"Archive": ".Archive",
"Spam": ".Junk",
"Trash": ".Trash",
}
MAX_ATTACHMENT_BYTES = 8 * 1024 * 1024
def load():
if not STATE.exists():
save(DEFAULT.copy())
with STATE.open() as f:
data = json.load(f)
for key, value in DEFAULT.items():
data.setdefault(key, value.copy() if isinstance(value, (list, dict)) else value)
return data
def save(data):
STATE.parent.mkdir(parents=True, exist_ok=True)
tmp = STATE.with_suffix(".tmp")
tmp.write_text(json.dumps(data, indent=2) + "\n")
tmp.replace(STATE)
def spam_policy(data):
policy = data.setdefault("spamPolicy", {})
for key, value in SPAM_POLICY_DEFAULT.items():
if isinstance(value, dict):
policy.setdefault(key, value.copy())
elif isinstance(value, list):
policy.setdefault(key, value.copy())
else:
policy.setdefault(key, value)
return policy
def load_webmail():
if not WEBMAIL_STATE.exists():
save_webmail(WEBMAIL_DEFAULT.copy())
with WEBMAIL_STATE.open() as f:
data = json.load(f)
for key, value in WEBMAIL_DEFAULT.items():
data.setdefault(key, value.copy() if isinstance(value, (list, dict)) else value)
return data
def save_webmail(data):
WEBMAIL_STATE.parent.mkdir(parents=True, exist_ok=True)
tmp = WEBMAIL_STATE.with_suffix(".tmp")
tmp.write_text(json.dumps(data, indent=2) + "\n")
tmp.replace(WEBMAIL_STATE)
def audit(data, text):
data.setdefault("audit", []).insert(0, {"at": int(time.time()), "text": text})
data["audit"] = data["audit"][:100]
def event(data, text):
data.setdefault("events", []).insert(0, {"at": int(time.time()), "text": text})
data["events"] = data["events"][:40]
def pw(n=24):
alphabet = string.ascii_letters + string.digits + "-_!@#%+"
return "".join(secrets.choice(alphabet) for _ in range(n))
def run(args):
return subprocess.run(args, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
def services():
names = ["postfix", "dovecot", "opendkim", "nginx", "akurai-mail-ui", "akurai-mail-hook", "fail2ban"]
out = {}
for name in names:
r = run(["systemctl", "is-active", name])
out[name] = r.stdout.strip() or "unknown"
return out
def mem_info():
values = {}
for line in Path("/proc/meminfo").read_text().splitlines():
key, raw = line.split(":", 1)
values[key] = int(raw.strip().split()[0])
total = values.get("MemTotal", 1)
available = values.get("MemAvailable", 0)
used = max(total - available, 0)
return {"totalMb": round(total / 1024), "usedMb": round(used / 1024), "percent": round((used / total) * 100, 1)}
def cpu_times():
parts = Path("/proc/stat").read_text().splitlines()[0].split()[1:]
nums = [int(x) for x in parts]
idle = nums[3] + (nums[4] if len(nums) > 4 else 0)
return sum(nums), idle
def cpu_percent(data):
total, idle = cpu_times()
prev = data.get("lastCpu")
data["lastCpu"] = {"total": total, "idle": idle}
if not prev:
return 0.0
diff_total = max(total - int(prev.get("total", total)), 1)
diff_idle = max(idle - int(prev.get("idle", idle)), 0)
return round(max(0, min(100, (1 - diff_idle / diff_total) * 100)), 1)
def disk_info(path="/"):
usage = shutil.disk_usage(path)
used = usage.total - usage.free
return {
"totalGb": round(usage.total / (1024 ** 3), 1),
"usedGb": round(used / (1024 ** 3), 1),
"percent": round((used / usage.total) * 100, 1),
}
def vm_metrics(data):
metrics = {"at": int(time.time()), "cpu": cpu_percent(data), "memory": mem_info(), "disk": disk_info("/")}
history = data.setdefault("metricsHistory", [])
history.append({"at": metrics["at"], "cpu": metrics["cpu"], "memory": metrics["memory"]["percent"], "disk": metrics["disk"]["percent"]})
data["metricsHistory"] = history[-48:]
return metrics
def dovecot_users():
users = []
if not DOVECOT_USERS.exists():
return users
for line in DOVECOT_USERS.read_text().splitlines():
if not line.strip() or line.startswith("#") or ":" not in line:
continue
email = line.split(":", 1)[0].strip()
if "@" not in email:
continue
home = ""
parts = line.split(":")
if len(parts) > 5:
home = parts[5]
size = mailbox_size(home) if home else 0
users.append({"email": email, "createdAt": 0, "status": "active", "home": home, "hasPassword": True, "quotaUsed": size, "quotaUsedMb": round(size / (1024 * 1024), 2), "lastDelivery": mailbox_last_delivery(home) if home else 0})
return sorted(users, key=lambda u: u["email"])
def user_home(address):
for user in dovecot_users():
if user["email"] == address:
return Path(user["home"])
raise ValueError("mailbox not found")
def mailbox_size(path):
total = 0
root = Path(path)
if not root.exists():
return 0
for item in root.rglob("*"):
if item.is_file():
total += item.stat().st_size
return total
def mailbox_last_delivery(path):
root = Path(path) / "Maildir" / "new"
if not root.exists():
return 0
mtimes = [item.stat().st_mtime for item in root.glob("*") if item.is_file()]
return int(max(mtimes)) if mtimes else 0
def mail_queue():
result = run(["mailq"])
text = result.stdout.strip()
if not text or "Mail queue is empty" in text:
return {"count": 0, "summary": "Mail queue is empty"}
count = len([line for line in text.splitlines() if re.match(r"^[A-F0-9]", line)])
return {"count": count, "summary": text.splitlines()[-1] if text.splitlines() else "Queue has messages"}
def delivery_errors():
result = run(["journalctl", "-u", "postfix", "--since", "24 hours ago", "--no-pager", "-n", "80"])
rows = []
for line in result.stdout.splitlines():
if re.search(r"\b(reject|warning|error|fatal|bounced)\b", line, re.I):
rows.append(line[-240:])
return rows[-12:]
def dns_query(name, qtype, server="1.1.1.1", timeout=3):
qmap = {"A": 1, "MX": 15, "TXT": 16}
typ = qmap[qtype]
tid = int(time.time() * 1000) & 0xFFFF
packet = struct.pack("!HHHHHH", tid, 0x0100, 1, 0, 0, 0)
for part in name.rstrip(".").split("."):
packet += bytes([len(part)]) + part.encode()
packet += b"\0" + struct.pack("!HH", typ, 1)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(timeout)
try:
sock.sendto(packet, (server, 53))
data, _ = sock.recvfrom(4096)
finally:
sock.close()
qd, an = struct.unpack("!HH", data[4:8])
pos = 12
def read_name(p):
labels = []
jumped = False
end = p
while True:
length = data[p]
if length & 0xC0 == 0xC0:
ptr = ((length & 0x3F) << 8) | data[p + 1]
if not jumped:
end = p + 2
p = ptr
jumped = True
continue
if length == 0:
if not jumped:
end = p + 1
break
p += 1
labels.append(data[p:p + length].decode(errors="replace"))
p += length
return ".".join(labels), end
for _ in range(qd):
_, pos = read_name(pos)
pos += 4
vals = []
for _ in range(an):
_, pos = read_name(pos)
atype, _, _, rdlen = struct.unpack("!HHIH", data[pos:pos + 10])
pos += 10
start = pos
rdata = data[pos:pos + rdlen]
pos += rdlen
if atype != typ:
continue
if atype == 1:
vals.append(socket.inet_ntoa(rdata))
elif atype == 15:
pref = struct.unpack("!H", rdata[:2])[0]
exch, _ = read_name(start + 2)
vals.append(f"{pref} {exch}")
elif atype == 16:
txt = []
p = 0
while p < len(rdata):
ln = rdata[p]
p += 1
txt.append(rdata[p:p + ln].decode(errors="replace"))
p += ln
vals.append("".join(txt))
return vals
def dns_health():
checks = [
{"name": HOSTNAME, "type": "A", "expect": PUBLIC_IP},
{"name": DOMAIN, "type": "MX", "expect": "10 mail.olibuijr.com"},
{"name": DOMAIN, "type": "TXT", "expect": "v=spf1 ip4:3.94.46.219 mx -all"},
{"name": "_dmarc." + DOMAIN, "type": "TXT", "expect": "v=DMARC1; p=quarantine; rua=mailto:postmaster@olibuijr.com"},
]
for check in checks:
try:
values = dns_query(check["name"], check["type"])
check["values"] = values
check["ok"] = check["expect"] in values
except Exception as e:
check["values"] = []
check["ok"] = False
check["error"] = str(e)
return checks
def folder_root(address, folder):
if folder not in FOLDERS:
raise ValueError("unknown folder")
root = user_home(address) / "Maildir"
suffix = FOLDERS[folder]
path = root if suffix == "" else root / suffix
for name in ("cur", "new", "tmp"):
(path / name).mkdir(parents=True, exist_ok=True)
return path
def msg_id(folder, subdir, name):
raw = json.dumps([folder, subdir, name], separators=(",", ":")).encode()
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def decode_msg_id(value):
padded = value + "=" * (-len(value) % 4)
return json.loads(base64.urlsafe_b64decode(padded.encode()).decode())
def message_path(address, value):
folder, subdir, name = decode_msg_id(value)
root = folder_root(address, folder)
path = root / subdir / name
if not path.exists() or subdir not in ("cur", "new"):
raise ValueError("message not found")
return folder, subdir, path
def clean_addr(value):
return str(value or "").replace("\n", " ").replace("\r", " ").strip()
def normalized_email(value):
return parseaddr(str(value or ""))[1].lower().strip()
def sender_domain(value):
address = normalized_email(value)
return address.rsplit("@", 1)[1] if "@" in address else ""
def message_links(text):
return len(re.findall(r"https?://|www\.", text, flags=re.I))
def spam_score(msg, policy):
sender = normalized_email(msg.get("from"))
domain = sender_domain(msg.get("from"))
allowlist = {str(x).lower() for x in policy.get("allowlist", [])}
if sender in allowlist or domain in allowlist:
return 0, ["allowlisted"]
score = 0
reasons = []
if sender and sender in {str(x).lower() for x in policy.get("blocklist", [])}:
score += 10
reasons.append("blocked sender")
if domain and domain in {str(x).lower().lstrip("@") for x in policy.get("domainBlocklist", [])}:
score += 8
reasons.append("blocked domain")
subject = str(msg.get("subject") or "")
body = plain_from_message(msg)
search = f"{subject}\n{body}".lower()
for word in [str(x).lower().strip() for x in policy.get("subjectKeywords", []) if str(x).strip()]:
if word in search:
score += 2
reasons.append(f"keyword: {word}")
auth = " ".join(msg.get_all("authentication-results", [])).lower()
for signal in ("spf=fail", "dkim=fail", "dmarc=fail", "spf=softfail"):
if signal in auth:
score += 3
reasons.append(signal)
ext_block = {str(x).lower().lstrip(".") for x in policy.get("attachmentExtensions", [])}
for part in msg.walk():
filename = part.get_filename()
if filename and "." in filename:
ext = filename.rsplit(".", 1)[1].lower()
if ext in ext_block:
score += 6
reasons.append(f"attachment: .{ext}")
links = message_links(search)
if links > int(policy.get("maxLinks", 10)):
score += 2
reasons.append(f"link count: {links}")
return score, reasons[:8]
def plain_from_message(msg):
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain" and not part.get_filename():
try:
return part.get_content()
except Exception:
return ""
for part in msg.walk():
if part.get_content_type() == "text/html" and not part.get_filename():
try:
return re.sub(r"<[^>]+>", " ", part.get_content())
except Exception:
return ""
return ""
try:
return msg.get_content()
except Exception:
return ""
def html_from_message(msg):
for part in msg.walk() if msg.is_multipart() else [msg]:
if part.get_content_type() == "text/html" and not part.get_filename():
try:
return sanitize_html(part.get_content())
except Exception:
return ""
text = html.escape(plain_from_message(msg))
return "<br>".join(text.splitlines())
def sanitize_html(value):
value = re.sub(r"(?is)<(script|style|iframe|object|embed|link|meta|base|form)[^>]*>.*?</\1>", "", value)
value = re.sub(r"(?is)<(script|style|iframe|object|embed|link|meta|base|form)[^>]*/?>", "", value)
value = re.sub(r"\s+on[a-zA-Z]+\s*=\s*(['\"]).*?\1", "", value)
value = re.sub(r"\s+(href|src)\s*=\s*(['\"])\s*javascript:.*?\2", r' \1="#"', value, flags=re.I)
return value
def message_summary(address, folder, subdir, path):
with path.open("rb") as f:
msg = email.message_from_binary_file(f, policy=default)
name = path.name
flags = name.split(":2,", 1)[1] if ":2," in name else ""
text = " ".join(plain_from_message(msg).split())
attachments = [p.get_filename() for p in msg.walk() if p.get_filename()]
policy = spam_policy(load())
score, reasons = spam_score(msg, policy)
return {
"id": msg_id(folder, subdir, name),
"folder": folder,
"from": clean_addr(msg.get("from")),
"to": clean_addr(msg.get("to")),
"subject": str(msg.get("subject") or "(no subject)"),
"date": str(msg.get("date") or ""),
"timestamp": path.stat().st_mtime,
"preview": text[:180],
"unread": subdir == "new" or "S" not in flags,
"flagged": "F" in flags,
"attachments": [a for a in attachments if a],
"spamScore": score,
"spamReasons": reasons,
}
def set_seen(path):
if path.parent.name != "new":
return path
target = path.parent.parent / "cur" / (path.name + ":2,S" if ":2," not in path.name else path.name)
target.parent.mkdir(parents=True, exist_ok=True)
path.rename(target)
return target
def update_flag(path, flag, enabled):
name = path.name
base, flags = (name.split(":2,", 1) + [""])[:2] if ":2," in name else (name, "")
flags_set = set(flags)
if enabled:
flags_set.add(flag)
else:
flags_set.discard(flag)
new_name = f"{base}:2,{''.join(sorted(flags_set))}"
target = path.with_name(new_name)
path.rename(target)
return target
def move_message_path(address, message_id, target_folder):
_, _, path = message_path(address, message_id)
target_root = folder_root(address, target_folder)
target = target_root / "cur" / path.name
suffix = 0
while target.exists():
suffix += 1
target = target_root / "cur" / f"{path.name}.{suffix}"
path.rename(target)
return target
def move_path_to_folder(address, path, target_folder):
target_root = folder_root(address, target_folder)
target = target_root / "cur" / path.name
suffix = 0
while target.exists():
suffix += 1
target = target_root / "cur" / f"{path.name}.{suffix}"
path.rename(target)
return target
def folder_counts(address):
counts = []
for folder in FOLDERS:
root = folder_root(address, folder)
total = len(list((root / "cur").glob("*"))) + len(list((root / "new").glob("*")))
unread = len(list((root / "new").glob("*")))
counts.append({"name": folder, "total": total, "unread": unread})
return counts
def decode_page_cursor(value):
if not value:
return None
try:
raw = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)).decode()
timestamp, message_id = raw.split("|", 1)
return float(timestamp), message_id
except (ValueError, UnicodeDecodeError, TypeError):
raise ValueError("invalid page cursor")
def encode_page_cursor(row):
raw = f"{row['timestamp']}|{row['id']}".encode()
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
def page_rows(rows, cursor, limit):
marker = decode_page_cursor(cursor)
if marker:
rows = [
row for row in rows
if (row["timestamp"], row["id"]) < marker
]
limit = max(1, min(int(limit), 100))
page = rows[:limit]
next_cursor = encode_page_cursor(page[-1]) if len(rows) > len(page) else None
return page, next_cursor
def normalize_subject(subject):
"""Strip Re:/Fwd: prefixes for conversation grouping."""
s = str(subject or "").strip()
while True:
m = re_module.match(r"(?i)^\s*(re|fwd?|aw|sv|vs|wg)\s*[\:\-\]]\s*", s)
if not m:
break
s = s[m.end():]
return s.strip() or s
def is_autoreply(msg):
"""Detect auto-replies and no-reply senders."""
from_addr = normalized_email(msg.get("from", ""))
if not from_addr:
return True
for pattern in ("no-?reply@", "noreply@", "mailer-daemon@", "mailer@",
"postmaster@", "bounce", "auto-reply", "autoreply"):
if re_module.search(pattern, from_addr):
return True
subject = str(msg.get("subject", "")).lower()
for phrase in ("out of office", "automatic reply", "auto reply",
"vacation", "away from", "undeliverable", "delivery status",
"out of the office", "on leave"):
if phrase in subject:
return True
return False
def is_self_mail(msg, user_address):
"""Check if the message was sent by the user themselves."""
from_addr = normalized_email(msg.get("from", ""))
return from_addr == normalized_email(user_address)
def extract_deadline(text):
"""Extract explicit deadline/dates from text. Returns (date_str, confidence)."""
if not text:
return None
# Look for patterns like "by Friday", "due 2024-01-15", "deadline: tomorrow"
patterns = [
(r"(?i)(?:deadline|due|by|before|needs? (?:to be done )?by)\s*[: ]\s*(\d{4}-\d{2}-\d{2})", 0.9),
(r"(?i)(?:deadline|due|by|before)\s*[: ]\s*(tomorrow)", 0.7),
(r"(?i)(?:deadline|due|by|before)\s*[: ]\s*(today)", 0.7),
(r"(?i)urgent.*?(?:reply|respond|action)", 0.5),
(r"(?i)(?:need|needs|needed)\s+(?:this|by)\s+(monday|tuesday|wednesday|thursday|friday|saturday|sunday)", 0.6),
]
for pat, conf in patterns:
m = re_module.search(pat, text)
if m:
return (m.group(1) if m.lastindex else m.group(0), conf)
return None
def priority_keywords(text):
"""Score priority from body/subject text."""
if not text:
return 0, []
text_lower = text.lower()
high_terms = ["urgent", "asap", "critical", "emergency", "immediate action",
"time sensitive", "deadline", "overdue", "escalation"]
low_terms = ["newsletter", "unsubscribe", "promotion", "sale", "offer",
"weekly digest", "monthly update", "notification", "receipt",
"order confirmation", "shipping", "tracking", "verification code"]
high_hits = [t for t in high_terms if t in text_lower]
low_hits = [t for t in low_terms if t in text_lower]
if high_hits:
return 3, high_hits[:3]
if low_hits:
return -1, low_hits[:3]
return 0, []
def compute_signals(address, folder, messages_by_conversation):
"""Compute per-conversation signals deterministically.
Returns a dict keyed by conversation_id with signal objects.
Each signal has: value, provenance, confidence, evidence, messageIds.
"""
signals = {}
now = time.time()
for conv_id, msgs in messages_by_conversation.items():
if not msgs:
continue
# Sort by timestamp ascending
sorted_msgs = sorted(msgs, key=lambda m: m.get("timestamp", 0))
msg_ids = [m["id"] for m in sorted_msgs]
# Separate inbound (from others) and outbound (from user)
inbound = [m for m in sorted_msgs if not is_self_mail(m, address)]
outbound = [m for m in sorted_msgs if is_self_mail(m, address)]
# Last activity
last_msg = sorted_msgs[-1]
last_inbound = inbound[-1] if inbound else None
last_outbound = outbound[-1] if outbound else None
last_inbound_at = last_inbound.get("timestamp", 0) if last_inbound else 0
last_outbound_at = last_outbound.get("timestamp", 0) if last_outbound else 0
# Participants
participants = set()
for m in sorted_msgs:
addr = normalized_email(m.get("from", ""))
if addr:
participants.add(addr)
for to_field in [m.get("to", "")]:
for a in re_module.split(r"[,;]", str(to_field)):
na = normalized_email(a)
if na:
participants.add(na)
# Reply-needed calculation
reply_needed = None
if last_inbound and not is_autoreply({"from": last_inbound.get("from", ""),
"subject": last_inbound.get("subject", "")}):
# Last message is from someone else (not user)
if not is_self_mail(last_msg, address):
# User hasn't replied after the last inbound
if last_outbound_at < last_inbound_at:
age_seconds = now - last_inbound_at
age_days = max(0, int(age_seconds / 86400))
age_hours = int((age_seconds % 86400) / 3600)
# Build evidence
evidence = [f"Last message from {normalized_email(last_msg.get('from', ''))} on {last_msg.get('date', 'unknown')}"]
if age_days > 0:
evidence.append(f"Unanswered for {age_days}d {age_hours}h")
elif age_hours > 0:
evidence.append(f"Unanswered for {age_hours}h")
confidence = 0.95
if age_days < 1:
confidence = 0.6 # Recently received, might still be reading
elif age_days > 7:
confidence = 0.85 # Might have been handled elsewhere
reply_needed = {
"value": True,
"provenance": "deterministic",
"confidence": confidence,
"evidence": evidence,
"messageIds": msg_ids,
"unansweredAge": {"days": age_days, "hours": age_hours},
}
else:
# User has already replied
reply_needed = {
"value": False,
"provenance": "deterministic",
"confidence": 0.95,
"evidence": ["Already answered"],
"messageIds": msg_ids,
}
else:
# Last message is from user — no reply needed
reply_needed = {
"value": False,
"provenance": "deterministic",
"confidence": 1.0,
"evidence": ["Last message is your own — sent mail or draft"],
"messageIds": msg_ids,
}
elif last_inbound and is_autoreply({"from": last_inbound.get("from", ""),
"subject": last_inbound.get("subject", "")}):
reply_needed = {
"value": False,
"provenance": "deterministic",
"confidence": 0.95,
"evidence": ["Last message is an auto-reply or no-reply sender"],
"messageIds": msg_ids,
}
elif not last_inbound:
# Only outbound (sent mail, drafts)
reply_needed = {
"value": False,
"provenance": "deterministic",
"confidence": 1.0,
"evidence": ["Conversation has only your messages — sent or draft"],
"messageIds": msg_ids,
}
# Follow-up-due calculation
follow_up_due = None
if last_outbound and last_outbound_at > last_inbound_at:
# User sent last message, no reply received
age_seconds = now - last_outbound_at
age_days = max(0, int(age_seconds / 86400))
if age_days >= 2:
follow_up_due = {
"value": True,
"provenance": "deterministic",
"confidence": 0.85,
"evidence": [f"Sent message {age_days}d ago with no reply"],
"messageIds": msg_ids,
}
elif age_days >= 1:
follow_up_due = {
"value": True,
"provenance": "deterministic",
"confidence": 0.5,
"evidence": [f"Sent message {age_days}d ago with no reply yet"],
"messageIds": msg_ids,
}
# Check for explicit deadlines
deadline_detected = None
all_text = " ".join([
str(m.get("subject", "")) + " " + str(m.get("preview", ""))
for m in sorted_msgs
])
deadline = extract_deadline(all_text)
if deadline:
deadline_detected = {
"value": deadline[0],
"provenance": "deterministic",
"confidence": deadline[1],
"evidence": [f"Deadline mention found: \"{deadline[0]}\""],
"messageIds": msg_ids,
}
# Upgrade follow-up if deadline is close
if follow_up_due is None:
follow_up_due = {
"value": True,
"provenance": "deterministic",
"confidence": deadline[1] * 0.8,
"evidence": [f"Deadline detected: {deadline[0]}"],
"messageIds": msg_ids,
}
# Priority calculation
prio_score, prio_evidence = priority_keywords(all_text)
priority = {
"value": "normal",
"provenance": "deterministic",
"confidence": 0.7,
"evidence": [],
"messageIds": msg_ids,
}
if prio_score >= 3:
priority["value"] = "high"
priority["confidence"] = 0.85
priority["evidence"] = [f"High-priority keywords: {', '.join(prio_evidence)}"]
elif prio_score < 0:
priority["value"] = "low"
priority["confidence"] = 0.8
priority["evidence"] = [f"Low-priority indicators: {', '.join(prio_evidence)}"]
# Summary
participant_list = sorted([p for p in participants if p != normalized_email(address)])[:5]
summary_text = f"{len(sorted_msgs)} message{'s' if len(sorted_msgs) != 1 else ''}"
if participant_list:
summary_text += f" with {', '.join(participant_list[:3])}"
subj = normalize_subject(sorted_msgs[0].get("subject", ""))
if subj:
summary_text += f" about \"{subj[:80]}\""
conv_signals = {
"conversationId": conv_id,
"participants": list(participants),
"messageCount": len(sorted_msgs),
"lastInboundAt": last_inbound.get("date", "") if last_inbound else None,
"lastInboundFrom": normalized_email(last_inbound.get("from", "")) if last_inbound else None,
"lastOutboundAt": last_outbound.get("date", "") if last_outbound else None,
"replyNeeded": reply_needed,
"followUpDue": follow_up_due,
"priority": priority,
"deadlineDetected": deadline_detected,
"summary": {
"value": summary_text,
"provenance": "deterministic",
"confidence": 1.0,
"evidence": ["Computed from thread metadata"],
"messageIds": msg_ids,
},
}
signals[conv_id] = conv_signals
return signals
def webmail_signals(args):
"""Compute and return conversation signals for a mailbox folder."""
data = load_webmail()
address = args.email
folder = args.folder
root = folder_root(address, folder)
# Read all messages
all_messages = {}
for subdir in ("new", "cur"):
for path in (root / subdir).glob("*"):
if path.name.startswith("."):
continue
try:
item = message_summary(address, folder, subdir, path)
all_messages[item["id"]] = item
except Exception:
continue
if not all_messages:
return print(json.dumps({"ok": True, "signals": {}, "messageSignals": {}}))
# Group by conversation (normalized subject)
conversations = {}
for mid, msg in all_messages.items():
conv_key = normalize_subject(msg.get("subject", ""))
if not conv_key:
conv_key = f"__no_subject_{mid}"
conversations.setdefault(conv_key, []).append(msg)
# Compute signals
corrections = data.get("signalCorrections", {})
signals = compute_signals(address, folder, conversations)
# Apply user corrections on top
for conv_id, corr in corrections.items():
if conv_id in signals:
for signal_type, correction in corr.items():
if signal_type in signals[conv_id]:
sig = signals[conv_id][signal_type]
if sig is not None and isinstance(sig, dict):
if correction.get("dismissed"):
sig["dismissed"] = correction["dismissed"]
if correction.get("correctedValue") is not None:
sig["correctedValue"] = correction["correctedValue"]
sig["corrected"] = True
sig["correctedAt"] = correction.get("at", 0)
sig["provenance"] = "user-corrected"
# Create per-message mapping
message_signals = {}
for conv_id, conv_sigs in signals.items():
for mid in conv_sigs.get("summary", {}).get("messageIds", []):
message_signals[mid] = conv_id
print(json.dumps({
"ok": True,
"signals": signals,
"messageSignals": message_signals,
}))
def webmail_signal_correct(args):
"""Persist a signal correction or dismissal."""
data = load_webmail()
address = args.email
payload = parse_json_arg(args.payload)
conversation_id = payload.get("conversationId")
signal_type = payload.get("signalType")
action = payload.get("action") # "dismiss" or "correct"
value = payload.get("value") # only for correct
if not conversation_id or not signal_type or action not in ("dismiss", "correct"):
raise ValueError("conversationId, signalType, and action (dismiss/correct) required")
corrections = data.setdefault("signalCorrections", {})
conv_corr = corrections.setdefault(conversation_id, {})
signal_corr = conv_corr.setdefault(signal_type, {})
if action == "dismiss":
signal_corr["dismissed"] = True
signal_corr["at"] = int(time.time())
elif action == "correct":
signal_corr["correctedValue"] = value
signal_corr["dismissed"] = False
signal_corr["at"] = int(time.time())
save_webmail(data)
print(json.dumps({"ok": True, "action": action}))
def webmail_state(args):
data = load_webmail()
action = getattr(args, "action", "")
if action == "signals":
return webmail_signals(args)
if action == "signal-correct":
return webmail_signal_correct(args)
address = args.email
folder = args.folder
process_incoming(address, folder)
root = folder_root(address, folder)
rows = []
query = (args.query or "").lower()
for subdir in ("new", "cur"):
for path in (root / subdir).glob("*"):
if path.name.startswith("."):
continue
try:
item = message_summary(address, folder, subdir, path)
except Exception:
continue
haystack = " ".join([item["from"], item["to"], item["subject"], item["preview"]]).lower()
if query and query not in haystack:
continue
rows.append(item)
rows.sort(key=lambda row: (row["timestamp"], row["id"]), reverse=True)
page, next_cursor = page_rows(rows, args.cursor, args.limit)
print(json.dumps({
"ok": True,
"mailbox": address,
"folder": folder,
"folders": folder_counts(address),
"messages": page,
"total": len(rows),
"nextCursor": next_cursor,
"contacts": data.get("contacts", []),
"signatures": data.get("signatures", {}),
"vacation": data.get("vacation", {}),
"rules": data.get("rules", []),
"preferences": data.get("preferences", {}),
"audit": data.get("audit", [])[:20],
}))
def read_message(args):
folder, subdir, path = message_path(args.email, args.id)
if args.mark_read:
path = set_seen(path)
subdir = "cur"
with path.open("rb") as f:
msg = email.message_from_binary_file(f, policy=default)
summary = message_summary(args.email, folder, subdir, path)
attachments = []
for index, part in enumerate(msg.walk()):
filename = part.get_filename()
if filename:
payload = part.get_payload(decode=True) or b""
attachments.append({"index": index, "filename": filename, "contentType": part.get_content_type(), "size": len(payload)})
print(json.dumps({"ok": True, "message": {**summary, "bodyText": plain_from_message(msg), "bodyHtml": html_from_message(msg), "attachments": attachments}}))
def save_to_folder(address, folder, msg):
root = folder_root(address, folder)
box = mailbox.Maildir(str(root), create=True)
item = mailbox.MaildirMessage(msg)
if folder in ("Sent", "Drafts"):
item.set_subdir("cur")
item.set_flags("S")
key = box.add(item)
box.flush()
return key
def parse_json_arg(value):
try:
return json.loads(value or "{}")
except json.JSONDecodeError:
raise ValueError("invalid JSON payload")
def send_message(args):
payload = parse_json_arg(args.payload)
sender = clean_addr(payload.get("from") or args.email)
if sender not in {u["email"] for u in dovecot_users()}:
raise ValueError("sender mailbox not found")
recipients = []
for key in ("to", "cc", "bcc"):
raw = payload.get(key) or ""
recipients.extend([clean_addr(x) for x in re.split(r"[,;]", raw) if clean_addr(x)])
if not recipients:
raise ValueError("at least one recipient is required")
body = str(payload.get("body") or "")
signature = str(payload.get("signature") or "")
if signature:
body = body.rstrip() + "\n\n" + signature.strip()
msg = EmailMessage()
msg["From"] = sender
msg["To"] = clean_addr(payload.get("to"))
if payload.get("cc"):
msg["Cc"] = clean_addr(payload.get("cc"))
msg["Subject"] = clean_addr(payload.get("subject") or "(no subject)")
msg["Date"] = formatdate(localtime=True)
msg["Message-ID"] = make_msgid(domain=DOMAIN)
msg.set_content(body)
for item in payload.get("attachments", [])[:5]:
name = clean_addr(item.get("name") or "attachment")
encoded = item.get("data") or ""
content = base64.b64decode(encoded.split(",", 1)[-1], validate=False)
if len(content) > MAX_ATTACHMENT_BYTES:
raise ValueError(f"attachment too large: {name}")
ctype = item.get("type") or mimetypes.guess_type(name)[0] or "application/octet-stream"
maintype, subtype = ctype.split("/", 1) if "/" in ctype else ("application", "octet-stream")
msg.add_attachment(content, maintype=maintype, subtype=subtype, filename=name)
proc = subprocess.run(["/usr/sbin/sendmail", "-t", "-oi"], input=msg.as_bytes(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if proc.returncode != 0:
raise ValueError(proc.stderr.decode(errors="replace")[:400] or "sendmail failed")
save_to_folder(sender, "Sent", msg)
data = load_webmail()
event(load(), f"Message sent from {sender}")
print(json.dumps({"ok": True, "sent": True, "recipients": recipients}))
def send_auto_reply(sender, recipient, original):
data = load_webmail()
vacation = data.get("vacation", {})
if not vacation.get("enabled"):
return False
source = clean_addr(original.get("from"))
if not source or DOMAIN in source:
return False
key = hashlib.sha1((source + str(original.get("message-id", ""))).encode()).hexdigest()
sent = data.setdefault("vacationSent", {})
if sent.get(key):
return False
msg = EmailMessage()
msg["From"] = recipient
msg["To"] = source
msg["Subject"] = clean_addr(vacation.get("subject") or "Out of office")
msg["Date"] = formatdate(localtime=True)
msg["Message-ID"] = make_msgid(domain=DOMAIN)
msg.set_content(str(vacation.get("body") or "I am currently away."))
proc = subprocess.run(["/usr/sbin/sendmail", "-t", "-oi"], input=msg.as_bytes(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if proc.returncode == 0:
sent[key] = int(time.time())
audit(data, f"Vacation reply sent to {source}")
save_webmail(data)
return True
return False
def process_incoming(address, folder):
if folder != "Inbox":
return
state = load()
policy = spam_policy(state)
data = load_webmail()
rules = [r for r in data.get("rules", []) if r.get("enabled")]
root = folder_root(address, "Inbox")
moved = 0
spam_moved = 0
scanned = 0
for subdir in ("new", "cur"):
for path in list((root / subdir).glob("*")):
try:
with path.open("rb") as f:
msg = email.message_from_binary_file(f, policy=default)
item = message_summary(address, "Inbox", subdir, path)
except Exception:
continue
scanned += 1
score, reasons = spam_score(msg, policy)
if policy.get("enabled", True) and score >= int(policy.get("threshold", 5)):
if policy.get("autoMove", True):
move_path_to_folder(address, path, "Spam")
spam_moved += 1
continue
send_auto_reply(address, address, msg)
for rule in rules:
value = str(item.get(rule.get("field", "from"), "")).lower()
if rule.get("contains", "").lower() and rule.get("contains", "").lower() in value:
target = rule.get("action", "Archive")
if target in FOLDERS and target != "Inbox":
move_message_path(address, item["id"], target)
moved += 1
break
if scanned:
stats = policy.setdefault("stats", {})
stats["scanned"] = int(stats.get("scanned", 0)) + scanned
stats["moved"] = int(stats.get("moved", 0)) + spam_moved
stats["lastScan"] = int(time.time())
save(state)
if spam_moved:
audit(data, f"Anti-spam quarantined {spam_moved} message(s)")
if moved:
audit(data, f"Automatic rules moved {moved} message(s)")
if spam_moved or moved:
save_webmail(data)
def save_draft(args):
payload = parse_json_arg(args.payload)
msg = EmailMessage()
msg["From"] = clean_addr(payload.get("from") or args.email)
msg["To"] = clean_addr(payload.get("to") or "")
msg["Subject"] = clean_addr(payload.get("subject") or "(draft)")
msg.set_content(str(payload.get("body") or ""))
key = save_to_folder(args.email, "Drafts", msg)
print(json.dumps({"ok": True, "draft": key}))
def message_action(args):
if args.action == "delete":
move_message_path(args.email, args.id, "Trash")
elif args.action == "archive":
move_message_path(args.email, args.id, "Archive")
elif args.action == "spam":
_, _, path = message_path(args.email, args.id)
with path.open("rb") as f:
msg = email.message_from_binary_file(f, policy=default)
sender = normalized_email(msg.get("from"))
state = load()
policy = spam_policy(state)
if sender and sender not in policy.setdefault("blocklist", []):
policy["blocklist"].append(sender)
event(state, f"Spam sender blocked: {sender}")
save(state)
move_message_path(args.email, args.id, "Spam")
elif args.action == "move":
move_message_path(args.email, args.id, args.target)
elif args.action == "read":
_, _, path = message_path(args.email, args.id)
set_seen(path)
elif args.action == "unread":
_, _, path = message_path(args.email, args.id)
path = update_flag(path, "S", False)
target = path.parent.parent / "new" / path.name
path.rename(target)
elif args.action == "star":
_, _, path = message_path(args.email, args.id)
update_flag(path, "F", True)
elif args.action == "unstar":
_, _, path = message_path(args.email, args.id)
update_flag(path, "F", False)
else:
raise ValueError("unknown message action")
print(json.dumps({"ok": True, "action": args.action}))
def webmail_config(args):
data = load_webmail()
payload = parse_json_arg(args.payload)
if args.kind == "contact":
contacts = [c for c in data.get("contacts", []) if c.get("email") != payload.get("email")]
contacts.append({"name": clean_addr(payload.get("name")), "email": clean_addr(payload.get("email"))})
data["contacts"] = sorted([c for c in contacts if c["email"]], key=lambda c: c["email"])
elif args.kind == "signature":
email_addr = clean_addr(payload.get("email") or args.email)
data.setdefault("signatures", {})[email_addr] = str(payload.get("signature") or "")
elif args.kind == "vacation":
data["vacation"] = {"enabled": bool(payload.get("enabled")), "subject": clean_addr(payload.get("subject")), "body": str(payload.get("body") or "")}
elif args.kind == "rule":
rules = data.setdefault("rules", [])
rule = {"id": hashlib.sha1(json.dumps(payload, sort_keys=True).encode()).hexdigest()[:12], "field": payload.get("field", "from"), "contains": str(payload.get("contains", "")), "action": payload.get("action", "Archive"), "enabled": bool(payload.get("enabled", True))}
rules.append(rule)
data["rules"] = rules[-50:]
elif args.kind == "preferences":
data["preferences"] = {**data.get("preferences", {}), **payload}
else:
raise ValueError("unknown config kind")
save_webmail(data)
print(json.dumps({"ok": True, "webmail": data}))
def webmail_export(args):
print(json.dumps({"ok": True, "webmail": load_webmail()}))
def webmail_import(args):
payload = parse_json_arg(args.payload)
data = load_webmail()
for key in ("contacts", "signatures", "vacation", "rules", "preferences"):
if key in payload:
data[key] = payload[key]
audit(data, "Settings imported")
save_webmail(data)
print(json.dumps({"ok": True, "webmail": data}))
def attachment_download(args):
_, _, path = message_path(args.email, args.id)
with path.open("rb") as f:
msg = email.message_from_binary_file(f, policy=default)
index = int(args.index)
for current, part in enumerate(msg.walk()):
if current == index and part.get_filename():
payload = part.get_payload(decode=True) or b""
print(json.dumps({"ok": True, "filename": part.get_filename(), "contentType": part.get_content_type(), "data": base64.b64encode(payload).decode()}))
return
raise ValueError("attachment not found")
def apply_rules(args):
data = load_webmail()
moved = 0
rules = [r for r in data.get("rules", []) if r.get("enabled")]
root = folder_root(args.email, args.folder)
for subdir in ("new", "cur"):
for path in list((root / subdir).glob("*")):
try:
item = message_summary(args.email, args.folder, subdir, path)
except Exception:
continue
for rule in rules:
value = str(item.get(rule.get("field", "from"), "")).lower()
if rule.get("contains", "").lower() in value:
target = rule.get("action", "Archive")
if target in FOLDERS and target != args.folder:
move_message_path(args.email, item["id"], target)
moved += 1
break
print(json.dumps({"ok": True, "moved": moved}))
def aliases():
out = []
if POSTFIX_VIRTUAL.exists():
for line in POSTFIX_VIRTUAL.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#") or " " not in line:
continue
alias, target = line.split(None, 1)
out.append({"alias": alias, "target": target, "createdAt": 0, "status": "active"})
return sorted(out, key=lambda a: a["alias"])
def spam_folder_count(address):
try:
root = folder_root(address, "Spam")
return len(list((root / "cur").glob("*"))) + len(list((root / "new").glob("*")))
except Exception:
return 0
def spam_summary(data):
policy = spam_policy(data)
counts = [{"email": user["email"], "spam": spam_folder_count(user["email"])} for user in dovecot_users()]
return {"policy": policy, "mailboxes": counts, "totalSpam": sum(row["spam"] for row in counts)}
def clean_list(values, domain=False):
rows = []
for value in values:
for item in re.split(r"[\n,;]", str(value or "")):
item = item.strip().lower()
if not item:
continue
rows.append(item.lstrip("@") if domain else item)
return sorted(set(rows))
def update_policy_list(policy, field, mode, values):
current = clean_list(policy.get(field, []), field == "domainBlocklist")
incoming = clean_list(values, field == "domainBlocklist")
if mode == "set":
current = incoming
elif mode == "remove":
current = [item for item in current if item not in incoming]
else:
current = sorted(set(current + incoming))
policy[field] = current[:200]
def anti_spam_config(args):
data = load()
policy = spam_policy(data)
payload = parse_json_arg(args.payload)
if "enabled" in payload:
policy["enabled"] = bool(payload.get("enabled"))
if "autoMove" in payload:
policy["autoMove"] = bool(payload.get("autoMove"))
if "threshold" in payload:
policy["threshold"] = max(1, min(20, int(payload.get("threshold") or policy.get("threshold", 5))))
if "maxLinks" in payload:
policy["maxLinks"] = max(0, min(100, int(payload.get("maxLinks") or policy.get("maxLinks", 10))))
field = payload.get("field")
if field in {"blocklist", "allowlist", "domainBlocklist", "subjectKeywords", "attachmentExtensions"}:
raw_values = payload.get("values", payload.get("value", ""))
values = raw_values if isinstance(raw_values, list) else [raw_values]
update_policy_list(policy, field, payload.get("mode", "add"), values)
event(data, "Anti-spam policy updated")
save(data)
print(json.dumps({"ok": True, "antiSpam": spam_summary(data)}))
def scan_mailbox_for_spam(address, policy):
root = folder_root(address, "Inbox")
scanned = 0
moved = 0
for subdir in ("new", "cur"):
for path in list((root / subdir).glob("*")):
try:
with path.open("rb") as f:
msg = email.message_from_binary_file(f, policy=default)
except Exception:
continue
scanned += 1
score, _ = spam_score(msg, policy)
if policy.get("enabled", True) and score >= int(policy.get("threshold", 5)) and policy.get("autoMove", True):
move_path_to_folder(address, path, "Spam")
moved += 1
return {"email": address, "scanned": scanned, "moved": moved}
def anti_spam_scan(args):
data = load()
policy = spam_policy(data)
users = [args.email] if args.email else [user["email"] for user in dovecot_users()]
results = [scan_mailbox_for_spam(address, policy) for address in users]
scanned = sum(row["scanned"] for row in results)
moved = sum(row["moved"] for row in results)
stats = policy.setdefault("stats", {})
stats["scanned"] = int(stats.get("scanned", 0)) + scanned
stats["moved"] = int(stats.get("moved", 0)) + moved
stats["lastScan"] = int(time.time())
event(data, f"Anti-spam scan moved {moved} of {scanned} message(s)")
save(data)
webmail = load_webmail()
if moved:
audit(webmail, f"Anti-spam scan quarantined {moved} message(s)")
save_webmail(webmail)
print(json.dumps({"ok": True, "scanned": scanned, "moved": moved, "results": results, "antiSpam": spam_summary(data)}))
def require_root():
if os.geteuid() == 0:
return
os.execvp("sudo", ["sudo", "-n", sys.argv[0], *sys.argv[1:]])
def ensure_mailbox(email, password):
local, _, domain = email.partition("@")
known_domains = {DOMAIN}
if DOMAINS_STATE.exists():
dd = load_domains()
known_domains.update(d["domain"] for d in dd.get("domains", []))
if domain not in known_domains or not local:
raise ValueError(f"mailbox must be in a configured domain (have: {', '.join(sorted(known_domains))})")
mail_root = Path("/var/vmail") / DOMAIN / local
mail_root.mkdir(parents=True, exist_ok=True)
subprocess.run(["chown", "-R", "vmail:mail", str(mail_root)], check=False)
subprocess.run(["chmod", "0770", str(mail_root)], check=False)
hash_value = crypt.crypt(password, crypt.mksalt(crypt.METHOD_BLOWFISH))
existing = []
if DOVECOT_USERS.exists():
existing = [line for line in DOVECOT_USERS.read_text().splitlines() if line and not line.startswith(email + ":")]
existing.append(f"{email}:{{BLF-CRYPT}}{hash_value}:5000:8::{mail_root}::userdb_mail=maildir:{mail_root}/Maildir")
DOVECOT_USERS.write_text("\n".join(existing) + "\n")
subprocess.run(["chown", "root:dovecot", str(DOVECOT_USERS)], check=False)
subprocess.run(["chmod", "0640", str(DOVECOT_USERS)], check=False)
vmailbox = []
if POSTFIX_VMAILBOX.exists():
vmailbox = [line for line in POSTFIX_VMAILBOX.read_text().splitlines() if line and not line.startswith(email + " ")]
vmailbox.append(f"{email} {domain}/{local}/")
POSTFIX_VMAILBOX.write_text("\n".join(vmailbox) + "\n")
subprocess.run(["postmap", str(POSTFIX_VMAILBOX)], check=False)
subprocess.run(["systemctl", "reload", "dovecot"], check=False)
subprocess.run(["systemctl", "reload", "postfix"], check=False)
def top_processes(n=15):
"""Return top N processes by CPU, including PID, user, cpu%, mem%, command."""
r = run(["ps", "aux", "--sort=-pcpu"])
lines = r.stdout.strip().splitlines()[1:n+1]
out = []
for line in lines:
parts = line.split(None, 10)
if len(parts) < 11:
continue
out.append({
"pid": int(parts[1]),
"user": parts[0],
"cpu": float(parts[2]),
"mem": float(parts[3]),
"vsz": int(parts[4]),
"rss": int(parts[5]),
"command": parts[10][:120],
})
return out
def metrics_quick(args):
"""Lightweight metrics endpoint for realtime polling."""
data = load()
m = vm_metrics(data)
save(data)
print(json.dumps({
"metrics": m,
"services": services(),
"processes": top_processes(),
"at": int(time.time()),
}))
def status(args):
data = load()
data["services"] = services()
data["metrics"] = vm_metrics(data)
data["users"] = dovecot_users()
data["aliases"] = aliases()
data["queue"] = mail_queue()
data["deliveryErrors"] = delivery_errors()
data["dnsHealth"] = dns_health()
data["antiSpam"] = spam_summary(data)
data["generatedAt"] = int(time.time())
save(data)
print(json.dumps(data))
def add_user(args):
require_root()
data = load()
password = args.password or pw()
try:
ensure_mailbox(args.email, password)
except ValueError as e:
print(json.dumps({"ok": False, "error": str(e)}))
return 0
event(data, f"Mailbox active: {args.email}")
save(data)
print(json.dumps({"ok": True, "email": args.email, "password": password}))
def set_password(args):
require_root()
data = load()
existing = {u["email"] for u in dovecot_users()}
if args.email not in existing:
print(json.dumps({"ok": False, "error": "user not found"}))
return 0
password = args.password or pw()
ensure_mailbox(args.email, password)
event(data, f"Password rotated: {args.email}")
save(data)
print(json.dumps({"ok": True, "email": args.email, "password": password}))
def add_alias(args):
require_root()
data = load()
lines = []
if POSTFIX_VIRTUAL.exists():
lines = [line for line in POSTFIX_VIRTUAL.read_text().splitlines() if line and not line.startswith(args.alias + " ")]
lines.append(f"{args.alias} {args.target}")
POSTFIX_VIRTUAL.write_text("\n".join(lines) + "\n")
subprocess.run(["postmap", str(POSTFIX_VIRTUAL)], check=False)
subprocess.run(["systemctl", "reload", "postfix"], check=False)
event(data, f"Alias active: {args.alias} -> {args.target}")
save(data)
print(json.dumps({"ok": True, "alias": args.alias, "target": args.target}))
def dns(args):
print(json.dumps({"records": [
{"type": "A", "name": HOSTNAME, "value": PUBLIC_IP},
{"type": "A", "name": "smtp." + DOMAIN, "value": PUBLIC_IP},
{"type": "A", "name": "imap." + DOMAIN, "value": PUBLIC_IP},
{"type": "A", "name": "webmail." + DOMAIN, "value": PUBLIC_IP},
{"type": "MX", "name": DOMAIN, "value": "10 " + HOSTNAME},
{"type": "TXT", "name": DOMAIN, "value": "v=spf1 ip4:" + PUBLIC_IP + " mx -all"},
{"type": "TXT", "name": "_dmarc." + DOMAIN, "value": "v=DMARC1; p=quarantine; rua=mailto:postmaster@" + DOMAIN},
{"type": "TXT", "name": "mail._domainkey." + DOMAIN, "value": "See /etc/opendkim/keys/olibuijr.com/mail.txt"},
]}))
def load_domains():
if not DOMAINS_STATE.exists():
default_domains = {"domains": [{"domain": DOMAIN, "addedAt": 0, "status": "active"}]}
save_domains(default_domains)
with DOMAINS_STATE.open() as f:
data = json.load(f)
return data
def save_domains(data):
DOMAINS_STATE.parent.mkdir(parents=True, exist_ok=True)
tmp = DOMAINS_STATE.with_suffix(".tmp")
tmp.write_text(json.dumps(data, indent=2) + "\n")
tmp.replace(DOMAINS_STATE)
def domain_dns_records(domain):
records = [
{"type": "MX", "name": "@", "value": f"10 {HOSTNAME}."},
{"type": "TXT", "name": "@", "value": f"v=spf1 mx a:{HOSTNAME} -all"},
{"type": "TXT", "name": "_dmarc", "value": f"v=DMARC1; p=quarantine; rua=mailto:postmaster@{domain}"},
]
dkim_key = read_dkim_public_key(domain)
if dkim_key:
records.append({"type": "TXT", "name": "mail._domainkey", "value": dkim_key})
else:
records.append({"type": "TXT", "name": "mail._domainkey", "value": "(DKIM key not yet generated)"})
return records
def read_dkim_public_key(domain):
key_file = OPENDKIM_KEYS_DIR / domain / "mail.txt"
if not key_file.exists():
return ""
text = key_file.read_text()
# Parse the bind-format TXT record into a single value
# e.g. mail._domainkey IN TXT ( "v=DKIM1; ..." "p=..." ) ; ...
parts = re.findall(r'"([^"]*)"', text)
return "".join(parts)
def dns_check_domain(domain):
checks = [
{"name": domain, "type": "MX", "expect": f"10 {HOSTNAME.rstrip('.')}"},
]
# Check SPF
checks.append({"name": domain, "type": "TXT", "expect_contains": "v=spf1"})
# Check DMARC
checks.append({"name": f"_dmarc.{domain}", "type": "TXT", "expect_contains": "v=DMARC1"})
# Check DKIM
dkim_key = read_dkim_public_key(domain)
if dkim_key:
checks.append({"name": f"mail._domainkey.{domain}", "type": "TXT", "expect_contains": "v=DKIM1"})
for check in checks:
try:
values = dns_query(check["name"], check["type"])
check["values"] = values
if "expect" in check:
check["ok"] = check["expect"] in values
elif "expect_contains" in check:
check["ok"] = any(check["expect_contains"] in v for v in values)
else:
check["ok"] = len(values) > 0
except Exception as e:
check["values"] = []
check["ok"] = False
check["error"] = str(e)
# Clean internal fields
check.pop("expect", None)
check.pop("expect_contains", None)
return checks
def configure_domain_mail(domain):
"""Configure Postfix virtual_mailbox_domains and OpenDKIM for a new domain."""
# Add to Postfix virtual_mailbox_domains if not already there
vmd_file = Path("/etc/postfix/vmailbox_domains")
existing = []
if vmd_file.exists():
existing = [line.strip() for line in vmd_file.read_text().splitlines() if line.strip()]
if domain not in existing:
existing.append(domain)
vmd_file.write_text("\n".join(existing) + "\n")
# Generate DKIM keys if they don't exist
key_dir = OPENDKIM_KEYS_DIR / domain
if not (key_dir / "mail.private").exists():
key_dir.mkdir(parents=True, exist_ok=True)
subprocess.run(
["opendkim-genkey", "-s", "mail", "-d", domain, "-D", str(key_dir)],
check=True,
)
subprocess.run(["chown", "-R", "opendkim:opendkim", str(key_dir)], check=False)
subprocess.run(["chmod", "0600", str(key_dir / "mail.private")], check=False)
# Update OpenDKIM KeyTable
keytable_entry = f"mail._domainkey.{domain} {domain}:mail:{key_dir}/mail.private"
if OPENDKIM_KEYTABLE.exists():
lines = [line for line in OPENDKIM_KEYTABLE.read_text().splitlines() if domain not in line]
else:
lines = []
lines.append(keytable_entry)
OPENDKIM_KEYTABLE.write_text("\n".join(lines) + "\n")
# Update OpenDKIM SigningTable
signing_entry = f"*@{domain} mail._domainkey.{domain}"
if OPENDKIM_SIGNINGTABLE.exists():
lines = [line for line in OPENDKIM_SIGNINGTABLE.read_text().splitlines() if domain not in line]
else:
lines = []
lines.append(signing_entry)
OPENDKIM_SIGNINGTABLE.write_text("\n".join(lines) + "\n")
# Update TrustedHosts
if OPENDKIM_TRUSTEDHOSTS.exists():
lines = OPENDKIM_TRUSTEDHOSTS.read_text().splitlines()
else:
lines = ["127.0.0.1", "::1", "localhost"]
if domain not in lines:
lines.append(domain)
OPENDKIM_TRUSTEDHOSTS.write_text("\n".join(lines) + "\n")
# Create vmail directory for the domain
vmail_dir = Path("/var/vmail") / domain
vmail_dir.mkdir(parents=True, exist_ok=True)
subprocess.run(["chown", "-R", "vmail:mail", str(vmail_dir)], check=False)
# Reload services
subprocess.run(["systemctl", "reload", "opendkim"], check=False)
subprocess.run(["systemctl", "reload", "postfix"], check=False)
def domain_list(args):
data = load_domains()
domains = data.get("domains", [])
result = []
for d in domains:
domain = d["domain"]
dns_records = domain_dns_records(domain)
dns_checks = dns_check_domain(domain)
all_ok = all(c.get("ok", False) for c in dns_checks)
result.append({
"domain": domain,
"addedAt": d.get("addedAt", 0),
"status": "verified" if all_ok else "pending",
"dnsRecords": dns_records,
"dnsChecks": dns_checks,
})
print(json.dumps({"ok": True, "domains": result}))
def domain_add(args):
require_root()
domain = args.domain.strip().lower()
if not re.match(r"^[a-z0-9]([a-z0-9\-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9\-]*[a-z0-9])?)+$", domain):
print(json.dumps({"ok": False, "error": "invalid domain name"}))
return 0
data = load_domains()
existing = [d["domain"] for d in data.get("domains", [])]
if domain in existing:
print(json.dumps({"ok": False, "error": "domain already exists"}))
return 0
configure_domain_mail(domain)
data.setdefault("domains", []).append({"domain": domain, "addedAt": int(time.time()), "status": "active"})
save_domains(data)
state = load()
event(state, f"Domain added: {domain}")
save(state)
dns_records = domain_dns_records(domain)
print(json.dumps({"ok": True, "domain": domain, "dnsRecords": dns_records}))
def domain_check(args):
domain = args.domain.strip().lower()
dns_records = domain_dns_records(domain)
dns_checks = dns_check_domain(domain)
all_ok = all(c.get("ok", False) for c in dns_checks)
print(json.dumps({"ok": True, "domain": domain, "status": "verified" if all_ok else "pending", "dnsRecords": dns_records, "dnsChecks": dns_checks}))
def domain_remove(args):
require_root()
domain = args.domain.strip().lower()
if domain == DOMAIN:
print(json.dumps({"ok": False, "error": "cannot remove the primary domain"}))
return 0
data = load_domains()
data["domains"] = [d for d in data.get("domains", []) if d["domain"] != domain]
save_domains(data)
# Remove from OpenDKIM tables (leave keys on disk for safety)
for table_file in [OPENDKIM_KEYTABLE, OPENDKIM_SIGNINGTABLE]:
if table_file.exists():
lines = [line for line in table_file.read_text().splitlines() if domain not in line]
table_file.write_text("\n".join(lines) + "\n")
if OPENDKIM_TRUSTEDHOSTS.exists():
lines = [line for line in OPENDKIM_TRUSTEDHOSTS.read_text().splitlines() if line.strip() != domain]
OPENDKIM_TRUSTEDHOSTS.write_text("\n".join(lines) + "\n")
# Remove from vmailbox_domains
vmd_file = Path("/etc/postfix/vmailbox_domains")
if vmd_file.exists():
lines = [line for line in vmd_file.read_text().splitlines() if line.strip() != domain]
vmd_file.write_text("\n".join(lines) + "\n")
subprocess.run(["systemctl", "reload", "opendkim"], check=False)
subprocess.run(["systemctl", "reload", "postfix"], check=False)
state = load()
event(state, f"Domain removed: {domain}")
save(state)
print(json.dumps({"ok": True, "domain": domain, "removed": True}))
def validate_login(args):
if not DOVECOT_USERS.exists():
print(json.dumps({"ok": False, "error": "mailbox auth unavailable"}))
return 0
for line in DOVECOT_USERS.read_text().splitlines():
if not line.startswith(args.email + ":"):
continue
stored = line.split(":", 2)[1]
if stored.startswith("{BLF-CRYPT}"):
stored = stored.removeprefix("{BLF-CRYPT}")
ok = bool(stored and crypt.crypt(args.password, stored) == stored)
print(json.dumps({"ok": True, "email": args.email} if ok else {"ok": False, "error": "invalid mailbox credentials"}))
return 0
print(json.dumps({"ok": False, "error": "mailbox not found"}))
return 0
def post_pibun(account):
"""POST a new-mail signal to pi-bun's /api/ingest-mail over the AkurAI-VPN mesh."""
url = os.environ.get("AKURAI_MAIL_PIBUN_URL", "http://100.88.0.9:4173/api/ingest-mail")
token = os.environ.get("AKURAI_MAIL_PIBUN_TOKEN", "").strip()
if not token:
print("[hook] AKURAI_MAIL_PIBUN_TOKEN is not set; skipping pi-bun notify", file=sys.stderr)
return False
body = json.dumps({"account": account}).encode()
req = urllib.request.Request(url, data=body, method="POST", headers={"content-type": "application/json", "authorization": "Bearer " + token})
try:
with urllib.request.urlopen(req, timeout=10) as resp:
resp.read()
print("[hook] notified pi-bun (%s)" % (account or "mailbox"), file=sys.stderr)
return True
except urllib.error.HTTPError as e:
print("[hook] pi-bun notify returned HTTP %s" % e.code, file=sys.stderr)
return False
except Exception as e:
print("[hook] pi-bun notify failed: %s" % e, file=sys.stderr)
return False
def hook_notify(args):
"""One-shot notify (manual/testing): POST once to pi-bun for an optional account."""
ok = post_pibun(getattr(args, "account", "") or "")
print(json.dumps({"ok": ok}))
return 0 if ok else 1
def hook_watch(args):
"""Long-running inotify watcher: on Inbox delivery, notify pi-bun (push, not poll)."""
libc = ctypes.CDLL(ctypes.util.find_library("c") or "libc.so.6", use_errno=True)
fd = libc.inotify_init()
if fd < 0:
raise OSError(ctypes.get_errno(), "inotify_init failed")
IN_CREATE, IN_MOVED_TO, IN_CLOSE_WRITE = 0x100, 0x80, 0x8
event_mask = IN_CREATE | IN_MOVED_TO | IN_CLOSE_WRITE
root = os.environ.get("AKURAI_MAIL_WATCH_ROOT", "/var/vmail")
watched = {} # wd -> (Path new_dir, str account_email)
def refresh():
known = {str(p) for p, _ in watched.values()}
for new_dir in sorted(Path(root).glob("*/*/Maildir/new")):
if str(new_dir) in known or not new_dir.is_dir():
continue
parts = new_dir.parts
email = "%s@%s" % (parts[-3], parts[-4])
wd = libc.inotify_add_watch(fd, str(new_dir).encode(), event_mask)
if wd >= 0:
watched[wd] = (new_dir, email)
print("[hook] watching %s (%s)" % (new_dir, email), file=sys.stderr)
refresh()
if not watched:
print("[hook] no Maildir/new directories under %s yet" % root, file=sys.stderr)
last_scan = time.time()
while True:
ready, _, _ = select.select([fd], [], [], 60)
if time.time() - last_scan >= 60:
refresh()
last_scan = time.time()
if not ready:
continue
raw = os.read(fd, 65536)
touched = set()
pos = 0
while pos + 16 <= len(raw):
wd, _mask, _cookie, namelen = struct.unpack_from("iIII", raw, pos)
pos += 16
name = raw[pos:pos + namelen].split(b"\0", 1)[0]
pos += namelen
entry = watched.get(wd)
if entry is None or name.startswith(b"."):
continue
touched.add(entry[1])
if not touched:
continue
time.sleep(0.3) # coalesce a delivery burst; pi-bun also de-dupes overlapping syncs
for email in touched:
post_pibun(email)
def main():
p = argparse.ArgumentParser()
sub = p.add_subparsers(required=True)
s = sub.add_parser("status"); s.set_defaults(fn=status)
s = sub.add_parser("metrics"); s.set_defaults(fn=metrics_quick)
s = sub.add_parser("add-user"); s.add_argument("email"); s.add_argument("--password"); s.set_defaults(fn=add_user)
s = sub.add_parser("set-password"); s.add_argument("email"); s.add_argument("--password"); s.set_defaults(fn=set_password)
s = sub.add_parser("add-alias"); s.add_argument("alias"); s.add_argument("target"); s.set_defaults(fn=add_alias)
s = sub.add_parser("webmail-state"); s.add_argument("email"); s.add_argument("--folder", default="Inbox"); s.add_argument("--query", default=""); s.add_argument("--limit", default="40"); s.add_argument("--cursor", default=""); s.add_argument("--action", default=""); s.add_argument("--payload", default="{}"); s.set_defaults(fn=webmail_state)
s = sub.add_parser("anti-spam-config"); s.add_argument("payload", nargs="?", default="{}"); s.set_defaults(fn=anti_spam_config)
s = sub.add_parser("domain-list"); s.set_defaults(fn=domain_list)
s = sub.add_parser("domain-add"); s.add_argument("domain"); s.set_defaults(fn=domain_add)
s = sub.add_parser("domain-check"); s.add_argument("domain"); s.set_defaults(fn=domain_check)
s = sub.add_parser("domain-remove"); s.add_argument("domain"); s.set_defaults(fn=domain_remove)
s = sub.add_parser("validate-login"); s.add_argument("email"); s.add_argument("password"); s.set_defaults(fn=validate_login)
s = sub.add_parser("webmail-read"); s.add_argument("email"); s.add_argument("id"); s.add_argument("--mark-read", action="store_true"); s.set_defaults(fn=read_message)
s = sub.add_parser("webmail-send"); s.add_argument("email"); s.add_argument("payload"); s.set_defaults(fn=send_message)
s = sub.add_parser("webmail-draft"); s.add_argument("email"); s.add_argument("payload"); s.set_defaults(fn=save_draft)
s = sub.add_parser("webmail-action"); s.add_argument("email"); s.add_argument("id"); s.add_argument("action"); s.add_argument("--target", default="Archive"); s.set_defaults(fn=message_action)
s = sub.add_parser("webmail-config"); s.add_argument("email"); s.add_argument("kind"); s.add_argument("payload"); s.set_defaults(fn=webmail_config)
s = sub.add_parser("webmail-apply-rules"); s.add_argument("email"); s.add_argument("--folder", default="Inbox"); s.set_defaults(fn=apply_rules)
s = sub.add_parser("webmail-export"); s.add_argument("email"); s.set_defaults(fn=webmail_export)
s = sub.add_parser("webmail-import"); s.add_argument("email"); s.add_argument("payload"); s.set_defaults(fn=webmail_import)
s = sub.add_parser("webmail-attachment"); s.add_argument("email"); s.add_argument("id"); s.add_argument("index"); s.set_defaults(fn=attachment_download)
s = sub.add_parser("webmail-signals"); s.add_argument("email"); s.add_argument("--folder", default="Inbox"); s.set_defaults(fn=webmail_signals)
s = sub.add_parser("webmail-signal-correct"); s.add_argument("email"); s.add_argument("payload"); s.set_defaults(fn=webmail_signal_correct)
s = sub.add_parser("hook-watch"); s.set_defaults(fn=hook_watch)
s = sub.add_parser("hook-notify"); s.add_argument("account", nargs="?", default=""); s.set_defaults(fn=hook_notify)
a = p.parse_args()
try:
r = a.fn(a)
except Exception as e:
print(json.dumps({"ok": False, "error": str(e)}))
r = 0
sys.exit(r or 0)
if __name__ == "__main__":
main()