Spaces:
Runtime error
Runtime error
File size: 4,449 Bytes
00a4363 070c931 00a4363 070c931 00a4363 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 | """Account resolution. One provider per account for life: Google accounts never get a password and
email accounts never get linked to Google."""
from dataclasses import dataclass
from datetime import UTC, datetime
import bcrypt
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import AuthProvider, Subscription, User, Workspace, WorkspaceMember
from app.services.plans import PLANS
class IdentityError(Exception):
def __init__(self, code: str, message: str, status: int = 400):
super().__init__(message)
self.code = code
self.message = message
self.status = status
@dataclass
class Resolved:
user: User
created: bool
def normalize_email(email: str) -> str:
return email.strip().lower()
def hash_password(password: str) -> str:
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
def verify_password(password: str, password_hash: str | None) -> bool:
if not password_hash:
return False
return bcrypt.checkpw(password.encode(), password_hash.encode())
def validate_password(password: str) -> None:
if len(password) < 8:
raise IdentityError("weak_password", "Use at least 8 characters.")
def wrong_provider_error(user: User) -> IdentityError:
if user.auth_provider == AuthProvider.GOOGLE:
return IdentityError("wrong_provider", "This email uses Google sign in. Continue with Google.", 409)
return IdentityError("wrong_provider", "This email uses a password. Sign in with email instead.", 409)
def find_user(db: Session, email: str) -> User | None:
return db.scalar(select(User).where(User.email == normalize_email(email)))
def _ensure_active(user: User) -> None:
if user.deleted_at or not user.is_active:
raise IdentityError("account_deleted", "This account was deleted. Contact support to restore it.", 403)
def _bootstrap_workspace(db: Session, user: User) -> None:
first = (user.name or user.email.split("@")[0]).split(" ")[0]
workspace = Workspace(name=f"{first}'s workspace", owner_id=user.id)
db.add(workspace)
db.flush()
db.add(WorkspaceMember(workspace_id=workspace.id, user_id=user.id, role="owner"))
free = PLANS["free"]
db.add(Subscription(workspace_id=workspace.id, plan="free", video_plan=free.id, video_limit=free.videos,
videos_period_start=datetime.now(UTC)))
def resolve_or_create_google_user(
db: Session, *, email: str, google_id: str, name: str | None, avatar_url: str | None
) -> Resolved:
email = normalize_email(email)
user = db.scalar(select(User).where(User.google_id == google_id)) or find_user(db, email)
if user:
_ensure_active(user)
if user.auth_provider != AuthProvider.GOOGLE:
raise wrong_provider_error(user)
user.name = user.name or name
user.avatar_url = avatar_url or user.avatar_url
return Resolved(user, False)
user = User(
email=email, name=name, avatar_url=avatar_url, auth_provider=AuthProvider.GOOGLE, google_id=google_id
)
db.add(user)
db.flush()
_bootstrap_workspace(db, user)
return Resolved(user, True)
def create_password_user(db: Session, *, email: str, name: str | None, password_hash: str) -> User:
email = normalize_email(email)
existing = find_user(db, email)
if existing:
_ensure_active(existing)
if existing.auth_provider != AuthProvider.EMAIL:
raise wrong_provider_error(existing)
raise IdentityError("exists", "An account with this email already exists. Sign in instead.", 409)
user = User(email=email, name=name, auth_provider=AuthProvider.EMAIL, password_hash=password_hash)
db.add(user)
db.flush()
_bootstrap_workspace(db, user)
return user
def resolve_password_user(db: Session, *, email: str, password: str) -> User:
user = find_user(db, email)
if not user:
raise IdentityError("invalid_credentials", "Email or password is incorrect.", 401)
_ensure_active(user)
if user.auth_provider != AuthProvider.EMAIL:
raise wrong_provider_error(user)
if not verify_password(password, user.password_hash):
raise IdentityError("invalid_credentials", "Email or password is incorrect.", 401)
return user
def soft_delete(db: Session, user: User) -> None:
user.deleted_at = datetime.now(UTC)
user.is_active = False
user.token_version += 1
|