"""SMTP manager listing for the desktop's SMTPs screen. Shows `smtp_manager` rows for a given (type, mail_type) -- Preprints SMTPs are type 3, mail_type 1 -- with the publisher name resolved from `users` via the integer `publisher_id`. """ import asyncio import smtplib import ssl from datetime import datetime, timedelta, timezone from email.message import EmailMessage from typing import Annotated from bson import ObjectId from bson.errors import InvalidId from fastapi import APIRouter, HTTPException, Query, status from ..db import get_db from ..models import SmtpOut, SmtpTestRequest, SmtpTestResult router = APIRouter(prefix="/smtp", tags=["smtp"]) # Publisher logos live here (relative to the web root); the file name is the # publisher's users.image, e.g. writable/uploads/publisher_logo/HSPI.png. LOGO_DIR = "writable/uploads/publisher_logo" # Daily used-count (DUL) resets at IST midnight. IST = timezone(timedelta(hours=5, minutes=30)) async def _publishers() -> dict: """publisher_id (int) -> {name, image} from the publisher's users row.""" out: dict = {} async for u in get_db()["users"].find( {"publisher_id": {"$exists": True}}, {"publisher_id": 1, "fname": 1, "image": 1}): pid = u.get("publisher_id") if pid is not None: out[pid] = {"name": u.get("fname"), "image": u.get("image")} return out @router.get("", response_model=list[SmtpOut], summary="SMTPs for a type / mail_type") async def list_smtp( type: Annotated[int, Query(description="smtp_manager.type (Preprints = 3)")] = 3, mail_type: Annotated[int, Query(description="smtp_manager.mail_type (SMTP = 1)")] = 1, ): pubs = await _publishers() # Match int or string, in case the field is stored either way. query = {"type": {"$in": [type, str(type)]}, "mail_type": {"$in": [mail_type, str(mail_type)]}} coll = get_db()["smtp_manager"] rows = await coll.find(query).sort("smtp_id", 1).to_list(length=None) today_ist = datetime.now(IST).strftime("%Y-%m-%d") out = [] for r in rows: # Daily DUL reset: zero used_count once per IST day so the view starts at 0. if str(r.get("used_reset_on") or "") != today_ist: await coll.update_one({"_id": r["_id"]}, {"$set": {"used_count": 0, "used_reset_on": today_ist}}) r["used_count"] = 0 r["used_reset_on"] = today_ist pub = pubs.get(r.get("publisher_id")) or {} image = pub.get("image") logo = f"{LOGO_DIR}/{image}" if image else None out.append(SmtpOut(publisher=pub.get("name"), publisher_logo=logo, **r)) return out async def _smtp_creds(doc: dict) -> tuple[str, str, str]: """Resolve (from_addr, login_user, login_password) for a test send. Both kinds pick a random subuser of this SMTP for the From address: * mail_type 1 (SMTP): log in with the smtp_manager user / password. * mail_type 2 (Mail): log in with the SUBUSER's full address as the username -- @ (e.g. editor.review@scholarsubmit.org) -- and its `password` from smtp_subusers; host/port still come from smtp_manager. """ domain = (doc.get("domain") or doc.get("host") or "").strip() subuser_doc = None if str(doc.get("type")) == "3" and str(doc.get("mail_type")) in ("1", "2"): pipeline = [ {"$match": {"smtp_id": {"$in": [doc["_id"], str(doc["_id"])]}}}, {"$sample": {"size": 1}}, ] rows = await (await get_db()["smtp_subusers"].aggregate(pipeline)).to_list(length=1) subuser_doc = rows[0] if rows else None sub = ((subuser_doc or {}).get("subuser") or "").strip() if sub: from_addr = sub if "@" in sub else f"{sub}@{domain}" else: user = (doc.get("user") or "").strip() from_addr = user if "@" in user else f"noreply@{domain}" if str(doc.get("mail_type")) == "2" and subuser_doc: login_user = from_addr # subuser@domain, e.g. editor.review@scholarsubmit.org login_password = subuser_doc.get("password") or "" else: login_user = (doc.get("user") or "").strip() login_password = doc.get("password") or "" return from_addr, login_user, login_password def _smtp_port(doc: dict) -> int: """The port to connect on: smtp_manager.port when set, else 587.""" raw = doc.get("port") return int(raw) if str(raw).strip().isdigit() else 587 def _send_test(doc: dict, to_email: str, from_addr: str, login_user: str, login_password: str) -> None: """Send one test message through the SMTP described by `doc`. Blocking -- call it via a worker thread. Raises on any failure.""" host = (doc.get("host") or "").strip() port = _smtp_port(doc) user = login_user password = login_password # EHLO/HELO name -> the sender domain, so the internal machine FQDN # (e.g. Linux-Server-01.peertechz.local) never lands in the Received header. helo = from_addr.split("@")[-1].strip() or (doc.get("domain") or "").strip() or None msg = EmailMessage() msg["Subject"] = "PHIDESK SMTP test" msg["From"] = from_addr msg["To"] = to_email msg.set_content("This is a test email from PHIDESK to verify this SMTP account is working.") context = ssl.create_default_context() # mail_type 2 servers are self-hosted (Postal) and present a certificate for # a different hostname than the branded SMTP host, so hostname verification # would always fail. Skip cert verification for those only. if str(doc.get("mail_type")) == "2": context.check_hostname = False context.verify_mode = ssl.CERT_NONE if port == 465: with smtplib.SMTP_SSL(host, port, timeout=30, context=context, local_hostname=helo) as server: if user and password: server.login(user, password) server.send_message(msg) else: with smtplib.SMTP(host, port, timeout=30, local_hostname=helo) as server: server.ehlo() if server.has_extn("starttls"): server.starttls(context=context) server.ehlo() if user and password: server.login(user, password) server.send_message(msg) @router.get("/{oid}/send-info", summary="Resolved SMTP send details (no send)") async def smtp_send_info(oid: str): """What the next test/send would use: host, port, From, login username and where the credentials come from. The subuser shown is a sample (a random one is picked per send).""" try: object_id = ObjectId(oid) except (InvalidId, TypeError): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid smtp id") doc = await get_db()["smtp_manager"].find_one({"_id": object_id}) if not doc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="SMTP not found") from_addr, login_user, _pw = await _smtp_creds(doc) mail_type = str(doc.get("mail_type")) return { "smtp_id": doc.get("smtp_id"), "host": doc.get("host"), "port": _smtp_port(doc), "domain": doc.get("domain"), "mail_type": doc.get("mail_type"), "from_addr": from_addr, "login_user": login_user, "auth_source": ("smtp_manager user / password" if mail_type == "1" else "smtp_subusers: subuser@domain / password"), } @router.post("/{oid}/test", response_model=SmtpTestResult, summary="Send a test email through one SMTP (by its _id) and record the result") async def test_smtp(oid: str, payload: SmtpTestRequest): coll = get_db()["smtp_manager"] try: object_id = ObjectId(oid) except (InvalidId, TypeError): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="invalid smtp id") # smtp_id (the integer) is NOT unique -- always resolve by the ObjectId _id. doc = await coll.find_one({"_id": object_id}) if not doc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="SMTP not found") email = payload.email.strip() from_addr, login_user, login_password = await _smtp_creds(doc) try: await asyncio.to_thread(_send_test, doc, email, from_addr, login_user, login_password) ok, detail = True, f"Test email sent to {email} (from {from_addr})." except Exception as exc: # surface the reason to the desktop ok, detail = False, f"{type(exc).__name__}: {exc}" # Persist the outcome so the SMTPs list shows it on the next load. await coll.update_one( {"_id": doc["_id"]}, {"$set": { "last_test_ok": ok, "last_test_status": "sent" if ok else "failed", "last_tested_on": datetime.now(timezone.utc), "last_test_detail": detail, "last_test_email": email, }}, ) return SmtpTestResult(ok=ok, detail=detail)