Added and fixed Create auction API endpoint. Added middleware, session-based authentication and logout endpoint

This commit is contained in:
2024-10-17 05:24:56 +00:00
parent fbbdad51c2
commit dca14861de
7 changed files with 358 additions and 42 deletions

View File

@ -1,13 +1,19 @@
import bcrypt
from jose import jwt, JWTError
from datetime import datetime, timedelta
from fastapi import HTTPException, status
from datetime import datetime, timedelta, timezone
from fastapi import HTTPException, status, Request
import os
from pydantic import BaseModel
from typing import Optional
# Secret and algorithm for JWT
SECRET_KEY = os.getenv('SECRET_KEY', 'your_jwt_secret_key') # Ensure this is set in your environment
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 14
class TokenData(BaseModel):
user_id: Optional[int] = None
# Hash password using bcrypt directly
def get_password_hash(password: str) -> str:
@ -26,29 +32,37 @@ def create_access_token(data: dict, expires_delta: timedelta = None):
"""Creates a JWT token with expiration time."""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# Verify JWT token
def verify_access_token(token: str):
"""Verifies the JWT token and returns the user_id if valid."""
def verify_access_token(request: Request) -> int:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
# First, check Authorization header (for cases where the JWT is passed in headers)
auth_header = request.headers.get("Authorization")
token = None
if auth_header and auth_header.startswith("Bearer "):
token = auth_header.split(" ")[1]
# If no Authorization header, fallback to cookies
if not token:
token = request.cookies.get("access_token")
if not token:
raise credentials_exception
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("user_id")
user_id: int = payload.get("user_id")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
headers={"WWW-Authenticate": "Bearer"},
)
raise credentials_exception
return user_id
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
headers={"WWW-Authenticate": "Bearer"},
)
raise credentials_exception