Magic link auth via Gmail SMTP + password fallback

This commit is contained in:
2026-04-13 14:08:16 +00:00
parent 00e76e412d
commit f57dd9621f
2 changed files with 219 additions and 123 deletions
+88 -1
View File
@@ -3,12 +3,15 @@ from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, RedirectResponse
from sqlalchemy.orm import Session
from pydantic import BaseModel
from datetime import datetime, timezone
from datetime import datetime, timezone, timedelta
from typing import Optional
import hashlib
import json
import os
import secrets
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from database import get_db, init_db
from models import User, AgentCatalog, AgentInstance, Run, LLMProvider, Bridge
@@ -19,6 +22,15 @@ app = FastAPI(title="Agent Command Center", version="2026.04.12.01")
SESSION_SECRET = os.environ.get("SESSION_SECRET", secrets.token_hex(32))
_sessions: dict[str, dict] = {} # token -> {user_id, username, role}
_magic_links: dict[str, dict] = {} # token -> {user_id, email, expires}
# SMTP Config
SMTP_HOST = os.environ.get("SMTP_HOST", "smtp.gmail.com")
SMTP_PORT = int(os.environ.get("SMTP_PORT", "587"))
SMTP_USER = os.environ.get("SMTP_USER", "eric.jungbauer@gmail.com")
SMTP_PASS = os.environ.get("SMTP_PASS", "jozj oags ifqy auey")
SMTP_FROM = os.environ.get("SMTP_FROM", "eric.jungbauer@gmail.com")
APP_URL = os.environ.get("APP_URL", "https://agents.jfamily.io")
def hash_password(password: str) -> str:
@@ -64,6 +76,9 @@ class LoginRequest(BaseModel):
username: str # accepts username or email
password: str
class MagicLinkRequest(BaseModel):
email: str
class InstanceCreate(BaseModel):
catalog_id: str
name: Optional[str] = None
@@ -134,6 +149,78 @@ def logout(response: Response, session: Optional[str] = Cookie(None)):
return {"status": "ok"}
def send_magic_email(email: str, token: str):
"""Send a magic link email via SMTP."""
link = f"{APP_URL}/auth/verify?token={token}"
html = f"""
<div style="font-family:-apple-system,sans-serif;max-width:480px;margin:0 auto;padding:2rem">
<h2 style="color:#6c5ce7">Agent Command Center</h2>
<p>Click the button below to sign in:</p>
<a href="{link}" style="display:inline-block;background:#6c5ce7;color:#fff;padding:12px 32px;
border-radius:6px;text-decoration:none;font-weight:500;margin:1rem 0">Sign In</a>
<p style="color:#888;font-size:13px">This link expires in 15 minutes. If you didn't request this, ignore this email.</p>
<p style="color:#888;font-size:12px">{APP_URL}</p>
</div>
"""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Sign in to Agent Command Center"
msg["From"] = SMTP_FROM
msg["To"] = email
msg.attach(MIMEText(f"Sign in: {link}\n\nExpires in 15 minutes.", "plain"))
msg.attach(MIMEText(html, "html"))
with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
server.starttls()
server.login(SMTP_USER, SMTP_PASS)
server.send_message(msg)
@app.post("/api/auth/magic-link")
def request_magic_link(data: MagicLinkRequest, db: Session = Depends(get_db)):
"""Send a magic link to the user's email."""
user = db.query(User).filter(User.email == data.email).first()
if not user:
# Don't reveal whether email exists — return success either way
return {"status": "ok", "message": "If that email is registered, a sign-in link has been sent."}
# Generate token
token = secrets.token_urlsafe(32)
_magic_links[token] = {
"user_id": user.id,
"email": user.email,
"expires": datetime.now(timezone.utc) + timedelta(minutes=15),
}
# Send email
try:
send_magic_email(user.email, token)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to send email: {e}")
return {"status": "ok", "message": "If that email is registered, a sign-in link has been sent."}
@app.get("/auth/verify")
def verify_magic_link(token: str, response: Response, db: Session = Depends(get_db)):
"""Verify a magic link token and log the user in."""
link_data = _magic_links.pop(token, None)
if not link_data:
return RedirectResponse("/login?error=invalid", status_code=302)
if datetime.now(timezone.utc) > link_data["expires"]:
return RedirectResponse("/login?error=expired", status_code=302)
user = db.query(User).filter(User.id == link_data["user_id"]).first()
if not user:
return RedirectResponse("/login?error=invalid", status_code=302)
# Create session
session_token = create_session(user)
resp = RedirectResponse("/", status_code=302)
resp.set_cookie("session", session_token, httponly=True, samesite="lax", max_age=86400 * 7)
return resp
@app.get("/api/me")
def me(user: dict = Depends(require_auth), db: Session = Depends(get_db)):
u = db.query(User).filter(User.id == user["user_id"]).first()