"""Gmail account management endpoints.""" import asyncio import secrets from datetime import datetime, timezone from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_current_user, get_current_user_csrf, get_db, require_csrf from app.core.security import log_audit from app.models import GmailAccount, OAuthClientConfig, OAuthToken, User from app.services import gmail_service router = APIRouter(prefix="/gmail", tags=["gmail-accounts"]) class AccountUpdate(BaseModel): sending_method: str = "api" # api | smtp daily_limit: int = 200 campaign_limit: int = 200 class BulkSmtpAdd(BaseModel): """One line per account: 'email:app_password' or 'email,app_password' (also accepts 'email app_password' / 'email|app_password').""" lines: str = Field(..., max_length=200_000) daily_limit: int = Field(default=200, ge=1, le=2000) campaign_limit: int = Field(default=200, ge=1, le=2000) verify: bool = True # perform a real SMTP login test for each account before saving def _account_out(account: GmailAccount) -> dict: t = account.oauth_token return { "id": account.id, "email": account.email, "display_name": account.display_name, "status": account.status, "auth_provider": account.auth_provider, "sending_method": account.sending_method, "daily_limit": account.daily_limit, "campaign_limit": account.campaign_limit, "sends_today": account.sends_today, "last_success_at": account.last_success_at.isoformat() if account.last_success_at else None, "error_count": account.error_count, "last_error": (account.last_error or "")[:300], "cooldown_until": account.cooldown_until.isoformat() if account.cooldown_until else None, "has_refresh_token": bool(t and t.refresh_token_enc), "token_expires_at": t.expires_at.isoformat() if t and t.expires_at else None, "authorized": account.status == "authorized", } @router.get("/accounts") async def list_accounts(user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db)): rows = (await db.execute(select(GmailAccount).where( GmailAccount.user_id == user.id).order_by(GmailAccount.id))).scalars().all() return {"accounts": [_account_out(a) for a in rows], "count": len(rows)} @router.post("/accounts/connect") async def connect_start(user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): """Return the Google authorization URL for this user to open (real OAuth start). Uses the user's most recently imported client config when available.""" from sqlalchemy import select as _select cfg_row = (await db.execute( _select(OAuthClientConfig) .where(OAuthClientConfig.user_id == user.id, OAuthClientConfig.is_active == True) # noqa .order_by(OAuthClientConfig.id.desc()))).scalars().first() if cfg_row is not None: client = gmail_service.resolve_client_config(cfg_row) else: try: client = gmail_service.resolve_client_config() except gmail_service.GmailOauthError: raise HTTPException(400, "No Google OAuth client configuration imported yet. " "Use 'Import Gmail Configuration' first.") from app.api.routers.auth import google_start_impl return await google_start_impl(user, client) @router.post("/accounts/{account_id}/test") async def test_account(account_id: int, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): """Real connection test: verify OAuth creds and Gmail API profile.""" account = await db.get(GmailAccount, account_id) if not account or account.user_id != user.id: raise HTTPException(404, "Account not found") is_smtp_pw = account.auth_provider == "smtp_password" token = account.oauth_token if not is_smtp_pw and not token: return {"ok": False, "detail": "No OAuth token stored for this account"} from app.db.session import SyncSession with SyncSession() as s: acc_sync = s.get(GmailAccount, account.id) tok_sync = s.get(OAuthToken, token.id) if token else None try: client = gmail_service.build_gmail_client_from_account(acc_sync, tok_sync) result = client.verify_connection() except Exception as e: result = {"ok": False, "detail": str(e)[:400]} if result.get("ok"): account.status = "authorized" await db.commit() else: account.status = "expired" if "auth" in (result.get("detail") or "").lower() else account.status account.last_error = (result.get("detail") or "")[:500] await db.commit() await log_audit(db, user.id, "gmail.test", "gmail_account", account.id, {"ok": result.get("ok")}, None) return result @router.post("/accounts/{account_id}/reauthorize") async def reauthorize(account_id: int, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): account = await db.get(GmailAccount, account_id) if not account or account.user_id != user.id: raise HTTPException(404, "Account not found") try: client = gmail_service.resolve_client_config_for_account(account) except gmail_service.GmailOauthError as e: raise HTTPException(400, str(e)) state = gmail_service.generate_state() import json as _json import redis as _redis from app.core.config import settings try: r = _redis.from_url(settings.REDIS_URL, socket_timeout=2) r.setex(f"oauthstate:{state}", 600, _json.dumps({"user_id": user.id})) except Exception: raise HTTPException(503, "State store unavailable") url = gmail_service.build_authorization_url(client, state) return {"authorization_url": url} @router.post("/accounts/{account_id}/revoke") async def disconnect_account(account_id: int, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db), request: Request = None): """Disconnect: revoke token at Google and delete stored credentials.""" account = await db.get(GmailAccount, account_id) if not account or account.user_id != user.id: raise HTTPException(404, "Account not found") token = account.oauth_token revoked_at_google = False if token: from app.services.gmail_service import revoke_token revoked_at_google = revoke_token(token.refresh_token or token.access_token) account.status = "revoked" account.is_active = False account.revoked_by_user = True account.revoked_at = datetime.now(timezone.utc) if token: await db.delete(token) await db.commit() await log_audit(db, user.id, "gmail.disconnect", "gmail_account", account.id, {"email": account.email, "revoked_at_google": revoked_at_google}, request.client.host if request else None) return {"ok": True, "revoked_at_google": revoked_at_google} @router.delete("/accounts/{account_id}") async def delete_account(account_id: int, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db), request: Request = None): """DELETE semantics: revoke token at Google (best-effort) and REMOVE the account row (oauth_tokens cascade; campaign_recipients/send_* FKs are SET NULL).""" account = await db.get(GmailAccount, account_id) if not account or account.user_id != user.id: raise HTTPException(404, "Account not found") revoked_at_google = False token = account.oauth_token if token: revoked_at_google = gmail_service.revoke_token(token.refresh_token or token.access_token) email = account.email await db.delete(account) # token row cascades (delete-orphan + FK CASCADE) await db.commit() await log_audit(db, user.id, "gmail.delete", "gmail_account", account_id, {"email": email, "revoked_at_google": revoked_at_google}, request.client.host if request else None) return {"ok": True, "removed": True, "revoked_at_google": revoked_at_google} @router.post("/accounts/clear-all") async def clear_all_accounts(user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db), request: Request = None): """Clear All: revoke tokens at Google (best-effort) and remove all locally saved Gmail account rows. Audit trail preserved.""" rows = (await db.execute(select(GmailAccount).where( GmailAccount.user_id == user.id))).scalars().all() revoked = 0 for account in rows: token = account.oauth_token if token: if gmail_service.revoke_token(token.refresh_token or token.access_token): revoked += 1 await db.delete(account) # token cascades; related FKs are SET NULL await db.commit() await log_audit(db, user.id, "gmail.clear_all", "gmail_account", None, {"count": len(rows), "revoked_at_google": revoked}, request.client.host if request else None) return {"ok": True, "removed": len(rows), "revoked_at_google": revoked} @router.post("/accounts/import-config") async def import_client_config(file: UploadFile = File(...), user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db), request: Request = None): """Import & validate a Google OAuth client_secret JSON (never echoed back).""" from app.core.config import settings content = await file.read() if len(content) > 1024 * 512: raise HTTPException(413, "File too large (max 512 KB)") if not file.filename.lower().endswith(".json"): raise HTTPException(400, "Only .json configuration files are accepted") try: parsed = gmail_service.parse_client_json(content) except gmail_service.GmailOauthError as e: raise HTTPException(400, str(e)) row = OAuthClientConfig( name=parsed["project_id"] or file.filename[:100], user_id=user.id, client_id=parsed["client_id"], client_secret=parsed["client_secret"], # encrypted via property setter project_id=parsed["project_id"], redirect_uri=f"{settings.BASE_URL}/auth/google/callback", ) db.add(row) await db.commit() await log_audit(db, user.id, "oauth_config.imported", "oauth_client_config", row.id, {"project": parsed["project_id"]}, request.client.host if request else None) return {"ok": True, "id": row.id, "project_id": parsed["project_id"], "client_id_preview": parsed["client_id"][:12] + "…", "redirect_uri": row.redirect_uri} def _parse_bulk_smtp_lines(raw: str) -> list[tuple[str, str, str]]: """Parse free-form pasted lines into (raw_line, email, app_password) tuples. Accepts separators: ':' ',' '|' or whitespace. Blank lines / comments (#) skipped.""" import re as _re out = [] for line in raw.splitlines(): line = line.strip() if not line or line.startswith("#"): continue parts = _re.split(r"[:,|\s]+", line, maxsplit=1) if len(parts) != 2: out.append((line, "", "")) continue email, pwd = parts[0].strip(), parts[1].strip() out.append((line, email, pwd)) return out @router.post("/accounts/bulk-smtp") async def bulk_add_smtp_accounts(payload: BulkSmtpAdd, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db), request: Request = None): """Bulk-add Gmail accounts authenticated via plain SMTP app-passwords (no Google OAuth client / consent screen needed). Each line: 'email:app_password'. For every line we: 1. Parse & syntax-validate (16-letter app password). 2. Optionally perform a REAL SMTP login against smtp.gmail.com (STARTTLS + AUTH LOGIN) to prove the credential actually works before saving it - this is what prevents the "shows connected but nothing sends" problem. 3. Store the account with the password encrypted at rest (Fernet), never echoed back in any API response. Existing accounts (same email) are updated in place rather than duplicated. """ parsed = _parse_bulk_smtp_lines(payload.lines) if not parsed: raise HTTPException(400, "No lines found. Paste one 'email:app_password' per line.") if len(parsed) > 500: raise HTTPException(400, "Max 500 lines per bulk-add request") existing_rows = (await db.execute(select(GmailAccount).where( GmailAccount.user_id == user.id))).scalars().all() existing_by_email = {a.email.lower(): a for a in existing_rows} results = [] seen_in_batch = set() async def _verify(email: str, pw: str) -> dict: # smtplib is blocking; run in a thread so we don't stall the event loop # and so many accounts can be verified concurrently. return await asyncio.to_thread(gmail_service.verify_smtp_login, email, pw) verify_tasks = [] prepared = [] for raw_line, email, pwd in parsed: # Never echo the raw pasted line back (it contains the plaintext # password) - only the parsed email is safe to return. row = {"email": email, "ok": False, "detail": ""} if not email or not pwd: # Show only the first token (never the raw line - it may contain # a password fragment even on a parse failure). row["email"] = raw_line.split()[0][:120] if raw_line.split() else "" row["detail"] = "Could not parse line (expected 'email:app_password')" results.append(row) continue key = email.lower() if key in seen_in_batch: row["detail"] = "Duplicate line in this batch (skipped)" results.append(row) continue seen_in_batch.add(key) norm_pw = gmail_service.normalize_app_password(pwd) syntax_err = gmail_service.validate_smtp_credentials_syntax(email, norm_pw) if syntax_err: row["detail"] = syntax_err results.append(row) continue prepared.append((row, email, norm_pw)) if payload.verify: verify_tasks.append(_verify(email, norm_pw)) verify_results = await asyncio.gather(*verify_tasks) if verify_tasks else [] vi = 0 for row, email, norm_pw in prepared: if payload.verify: vres = verify_results[vi]; vi += 1 if not vres.get("ok"): row["detail"] = vres.get("detail", "SMTP login failed") results.append(row) continue key = email.lower() existing = existing_by_email.get(key) if existing: existing.smtp_password = norm_pw existing.auth_provider = "smtp_password" existing.sending_method = "smtp" existing.status = "authorized" existing.daily_limit = payload.daily_limit existing.campaign_limit = payload.campaign_limit existing.last_error = "" existing.is_active = True account = existing else: account = GmailAccount( user_id=user.id, email=email, display_name="", status="authorized", sending_method="smtp", auth_provider="smtp_password", daily_limit=payload.daily_limit, campaign_limit=payload.campaign_limit, ) account.smtp_password = norm_pw db.add(account) existing_by_email[key] = account row["ok"] = True row["detail"] = "Verified via real SMTP login" if payload.verify else "Added (not verified)" results.append(row) await db.commit() added_or_updated = sum(1 for r in results if r["ok"]) await log_audit(db, user.id, "gmail.bulk_smtp_add", "gmail_account", None, {"submitted": len(parsed), "ok": added_or_updated}, request.client.host if request else None) return {"ok": True, "total": len(parsed), "succeeded": added_or_updated, "failed": len(parsed) - added_or_updated, "results": results} @router.patch("/accounts/{account_id}") async def update_account(account_id: int, payload: AccountUpdate, user: User = Depends(get_current_user_csrf), db: AsyncSession = Depends(get_db)): account = await db.get(GmailAccount, account_id) if not account or account.user_id != user.id: raise HTTPException(404, "Account not found") if payload.sending_method in ("api", "smtp"): account.sending_method = payload.sending_method account.daily_limit = max(1, payload.daily_limit) account.campaign_limit = max(1, payload.campaign_limit) await db.commit() return _account_out(account)