Spaces:
Runtime error
Runtime error
File size: 6,519 Bytes
f3997d4 6fc5402 f3997d4 4f772f1 f3997d4 4f772f1 f3997d4 4f772f1 f3997d4 6fc5402 f3997d4 6fc5402 f3997d4 4f772f1 f3997d4 4f772f1 f3997d4 4f772f1 f3997d4 | 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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | from sqlalchemy.orm import Session
from typing import Optional
import httpx
from app.database.models import User
from app.utils.password import hash_password, verify_password
from app.utils.jwt import create_access_token, create_refresh_token
from app.utils.helpers import generate_id
try:
from google.oauth2 import id_token
from google.auth.transport import requests as google_requests
GOOGLE_AUTH_AVAILABLE = True
except ImportError:
GOOGLE_AUTH_AVAILABLE = False
print("⚠️ WARNING: google-auth libraries not found. Google Login will fail.")
class AuthService:
"""Service for authentication operations."""
@staticmethod
def create_user(db: Session, email: str, password: str, name: Optional[str] = None) -> User:
"""
Create a new user with email and password.
Args:
db: Database session
email: User email
password: Plain text password
name: Optional user name
Returns:
Created User object
Raises:
ValueError: If email already exists
"""
# Check if user exists
existing_user = db.query(User).filter(User.email == email).first()
if existing_user:
raise ValueError("Email already registered")
# Hash password
password_hash = hash_password(password)
# Create user
import os
admin_email = os.environ.get('ADMIN_EMAIL')
is_admin_user = 1 if admin_email and email == admin_email else 0
user = User(
id=generate_id(),
email=email,
password_hash=password_hash,
name=name,
is_admin=is_admin_user
)
db.add(user)
db.commit()
db.refresh(user)
if is_admin_user:
print(f"✅ User {email} registered and granted ADMIN privileges!")
return user
@staticmethod
def authenticate_user(db: Session, email: str, password: str) -> Optional[User]:
"""
Authenticate a user with email and password.
Args:
db: Database session
email: User email
password: Plain text password
Returns:
User object if authentication successful, None otherwise
"""
user = db.query(User).filter(User.email == email).first()
if not user or not user.password_hash:
return None
if not verify_password(password, user.password_hash):
return None
return user
@staticmethod
async def verify_google_token(credential: str, client_id: str) -> dict:
"""
Verify Google OAuth token and extract user info.
Args:
credential: Google JWT token from frontend
client_id: Google OAuth client ID from settings
Returns:
Dictionary with user info (email, name, google_id, picture)
Raises:
ValueError: If token is invalid or verification fails
"""
if not GOOGLE_AUTH_AVAILABLE:
raise ValueError("Google Auth libraries are not installed on this server.")
try:
# Verify the token with Google
idinfo = id_token.verify_oauth2_token(
credential,
google_requests.Request(),
client_id
)
# Extract user information from the token
return {
"email": idinfo.get("email"),
"name": idinfo.get("name"),
"google_id": idinfo.get("sub"), # 'sub' is the Google user ID
"picture": idinfo.get("picture")
}
except Exception as e:
print(f"❌ Error verifying Google token: {e}")
raise ValueError(f"Invalid Google token: {str(e)}")
@staticmethod
def create_user_from_google(
db: Session,
google_id: str,
email: str,
name: Optional[str] = None
) -> User:
"""
Create or get user from Google OAuth.
Args:
db: Database session
google_id: Google user ID
email: User email
name: Optional user name
Returns:
User object
"""
# Check if user exists with this Google ID
user = db.query(User).filter(User.google_id == google_id).first()
if user:
return user
# Check if user exists with this email
user = db.query(User).filter(User.email == email).first()
if user:
# Link Google account
user.google_id = google_id
if name and not user.name:
user.name = name
db.commit()
db.refresh(user)
return user
# Create new user
import os
admin_email = os.environ.get('ADMIN_EMAIL')
is_admin_user = 1 if admin_email and email == admin_email else 0
user = User(
id=generate_id(),
email=email,
google_id=google_id,
name=name,
is_admin=is_admin_user
)
db.add(user)
db.commit()
db.refresh(user)
if is_admin_user:
print(f"✅ User {email} registered via Google and granted ADMIN privileges!")
return user
@staticmethod
def get_user_by_id(db: Session, user_id: str) -> Optional[User]:
"""
Get user by ID.
Args:
db: Database session
user_id: User ID
Returns:
User object if found, None otherwise
"""
return db.query(User).filter(User.id == user_id).first()
@staticmethod
def generate_tokens(user: User) -> dict:
"""
Generate access and refresh tokens for a user.
Args:
user: User object
Returns:
Dictionary with access_token and refresh_token
"""
access_token = create_access_token(user.id)
refresh_token = create_refresh_token(user.id)
return {
"access_token": access_token,
"refresh_token": refresh_token
}
|