8f5e26d2ee
Add email+password auth alongside Google: argon2id hashing; users gains password_hash/email_verified/is_active and google_sub becomes nullable (migration 0022); a single-use, hashed auth_tokens table for email verification + password reset. Registration creates a pending (inactive, unverified) account + an access request; sign-in needs verified email AND admin approval. Anti-enumeration: uniform register/login/reset responses (a correct-password owner still gets a specific pending reason). allow_registration flag (DB-tunable). Admin users-list + role endpoint (guards: not self, not demo, not the last admin); approving access (or a manual whitelist add) now activates the matching pending account. current_user rejects deactivated accounts. Verified end-to-end via curl + DB.
123 lines
4.4 KiB
Python
123 lines
4.4 KiB
Python
"""Outbound email for onboarding (access requests + approval notices).
|
|
|
|
Gmail SMTP + App Password by default. Deliberately fail-soft: if SMTP isn't configured
|
|
or a send errors, we log and return False so the caller can fall back to in-app
|
|
notifications — email is never load-bearing for the app's core flow.
|
|
"""
|
|
import logging
|
|
import smtplib
|
|
import ssl
|
|
from email.message import EmailMessage
|
|
from email.utils import formatdate
|
|
|
|
from app import sysconfig
|
|
from app.config import settings
|
|
from app.db import SessionLocal
|
|
|
|
log = logging.getLogger("subfeed.email")
|
|
|
|
|
|
def _smtp() -> dict:
|
|
"""Effective SMTP config (admin DB override over env defaults). Opens its own short
|
|
session because email is often sent from a BackgroundTask, outside a request's db."""
|
|
with SessionLocal() as db:
|
|
return {
|
|
"host": sysconfig.get_str(db, "smtp_host"),
|
|
"port": sysconfig.get_int(db, "smtp_port"),
|
|
"user": sysconfig.get_str(db, "smtp_user"),
|
|
"password": sysconfig.get_str(db, "smtp_password"),
|
|
"from": sysconfig.get_str(db, "smtp_from"),
|
|
}
|
|
|
|
|
|
def email_enabled() -> bool:
|
|
c = _smtp()
|
|
return bool(c["host"] and c["user"] and c["password"])
|
|
|
|
|
|
def _admin_contact() -> str | None:
|
|
return next(iter(sorted(settings.admin_email_set)), None)
|
|
|
|
|
|
def _send(to: list[str], subject: str, body: str, reply_to: str | None = None) -> bool:
|
|
recipients = [e for e in to if e]
|
|
if not recipients:
|
|
return False
|
|
c = _smtp()
|
|
if not (c["host"] and c["user"] and c["password"]):
|
|
log.info("SMTP not configured; skipping email %r to %s", subject, recipients)
|
|
return False
|
|
msg = EmailMessage()
|
|
msg["Subject"] = subject
|
|
msg["From"] = c["from"] or c["user"]
|
|
msg["To"] = ", ".join(recipients)
|
|
# A real Date and a Reply-To (so it's a conversation, not a no-reply blast) both
|
|
# nudge spam filters the right way; reputation still does most of the work.
|
|
msg["Date"] = formatdate(localtime=True)
|
|
if reply_to:
|
|
msg["Reply-To"] = reply_to
|
|
msg.set_content(body)
|
|
try:
|
|
with smtplib.SMTP(c["host"], c["port"], timeout=20) as s:
|
|
s.starttls(context=ssl.create_default_context())
|
|
s.login(c["user"], c["password"])
|
|
s.send_message(msg)
|
|
log.info("Sent email %r to %s", subject, recipients)
|
|
return True
|
|
except Exception:
|
|
log.exception("SMTP send failed (%r to %s)", subject, recipients)
|
|
return False
|
|
|
|
|
|
def send_access_approved(email: str) -> bool:
|
|
admin = _admin_contact()
|
|
body = (
|
|
"Hi,\n\n"
|
|
"Your request to use Siftlode has been approved — welcome aboard.\n\n"
|
|
"Just open Siftlode and sign in with this Google account, and your YouTube\n"
|
|
"subscriptions feed will be ready.\n\n"
|
|
"If you weren't expecting this, you can ignore the message.\n\n"
|
|
"— Siftlode"
|
|
+ (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "")
|
|
)
|
|
return _send([email], "You're in — your Siftlode access is approved", body, reply_to=admin)
|
|
|
|
|
|
def send_admin_new_request(admins: list[str], requester: str) -> bool:
|
|
body = (
|
|
f"{requester} just requested access to Siftlode.\n\n"
|
|
"Open Siftlode → Settings → Account → Access requests to approve or deny it.\n\n"
|
|
"(Reply to this email to reach the requester directly.)\n"
|
|
)
|
|
return _send(admins, f"Siftlode access request from {requester}", body, reply_to=requester)
|
|
|
|
|
|
def send_verify_email(to: str, link: str) -> bool:
|
|
body = (
|
|
"Hi,\n\n"
|
|
"Confirm your email to finish creating your Siftlode account:\n\n"
|
|
f"{link}\n\n"
|
|
"If you didn't sign up, you can ignore this message.\n\n"
|
|
"— Siftlode"
|
|
)
|
|
return _send([to], "Confirm your Siftlode email", body)
|
|
|
|
|
|
def send_password_reset(to: str, link: str) -> bool:
|
|
body = (
|
|
"Hi,\n\n"
|
|
"Use this link to set a new Siftlode password (it expires in an hour):\n\n"
|
|
f"{link}\n\n"
|
|
"If you didn't request this, you can ignore this message — your password is unchanged.\n\n"
|
|
"— Siftlode"
|
|
)
|
|
return _send([to], "Reset your Siftlode password", body)
|
|
|
|
|
|
def send_test(to: str) -> bool:
|
|
body = (
|
|
"This is a test email from Siftlode.\n\n"
|
|
"If you're reading this, your SMTP settings work.\n\n"
|
|
"— Siftlode"
|
|
)
|
|
return _send([to], "Siftlode SMTP test", body)
|