Menu
Public repository / a7e96191cab7cc0297487b9e56f4e19a4c057957
AkurAI-Mail / scripts/akurai-mail-server
Latest change aa0081daf17648c0a6e369ad847bdc30c0c7d321 - feat(status): realtime process list, animated charts, auto-polling 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
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", "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 webmail_state(args):
data = load_webmail()
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"], reverse=True)
print(json.dumps({
"ok": True,
"mailbox": address,
"folder": folder,
"folders": folder_counts(address),
"messages": rows[: int(args.limit)],
"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 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("dns"); s.set_defaults(fn=dns)
s = sub.add_parser("anti-spam-config"); s.add_argument("payload", nargs="?", default="{}"); s.set_defaults(fn=anti_spam_config)
s = sub.add_parser("anti-spam-scan"); s.add_argument("email", nargs="?"); s.set_defaults(fn=anti_spam_scan)
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-state"); s.add_argument("email"); s.add_argument("--folder", default="Inbox"); s.add_argument("--query", default=""); s.add_argument("--limit", default="100"); s.set_defaults(fn=webmail_state)
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)
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()