Spaces:
Sleeping
Sleeping
Debug - auth
Browse files- app/auth.py +48 -0
app/auth.py
CHANGED
|
@@ -0,0 +1,48 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
import os
|
| 2 |
+
from datetime import datetime, timedelta, timezone
|
| 3 |
+
from passlib.context import CryptContext
|
| 4 |
+
from itsdangerous import URLSafeTimedSerializer, SignatureExpired, BadSignature
|
| 5 |
+
from dotenv import load_dotenv
|
| 6 |
+
from typing import Optional
|
| 7 |
+
from . import crud, models
|
| 8 |
+
|
| 9 |
+
load_dotenv()
|
| 10 |
+
|
| 11 |
+
SECRET_KEY = os.getenv("SECRET_KEY", "super-secret") # Fallback, but .env should be used
|
| 12 |
+
# Use URLSafeTimedSerializer for session tokens that expire
|
| 13 |
+
serializer = URLSafeTimedSerializer(SECRET_KEY)
|
| 14 |
+
|
| 15 |
+
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
| 16 |
+
|
| 17 |
+
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
| 18 |
+
return pwd_context.verify(plain_password, hashed_password)
|
| 19 |
+
|
| 20 |
+
def get_password_hash(password: str) -> str:
|
| 21 |
+
return pwd_context.hash(password)
|
| 22 |
+
|
| 23 |
+
# Session Token generation (using itsdangerous for simplicity)
|
| 24 |
+
# Stores user_id securely signed with a timestamp
|
| 25 |
+
def create_session_token(user_id: int) -> str:
|
| 26 |
+
return serializer.dumps(user_id)
|
| 27 |
+
|
| 28 |
+
# Session Token verification
|
| 29 |
+
async def get_user_id_from_token(token: str) -> Optional[int]:
|
| 30 |
+
if not token:
|
| 31 |
+
return None
|
| 32 |
+
try:
|
| 33 |
+
# Set max_age to something reasonable, e.g., 1 day
|
| 34 |
+
user_id = serializer.loads(token, max_age=86400) # 24 hours * 60 min * 60 sec
|
| 35 |
+
return int(user_id)
|
| 36 |
+
except (SignatureExpired, BadSignature, ValueError):
|
| 37 |
+
return None
|
| 38 |
+
|
| 39 |
+
# Function to get current user from token
|
| 40 |
+
async def get_current_user_from_token(token: str) -> Optional[models.User]:
|
| 41 |
+
user_id = await get_user_id_from_token(token)
|
| 42 |
+
if user_id is None:
|
| 43 |
+
return None
|
| 44 |
+
user = await crud.get_user_by_id(user_id)
|
| 45 |
+
if user:
|
| 46 |
+
# Return the public User model, not UserInDB
|
| 47 |
+
return models.User(id=user.id, email=user.email)
|
| 48 |
+
return None
|