| """ |
| Authentication Service |
| Handles JWT token generation, validation, and password hashing. |
| """ |
|
|
| import os |
| from datetime import datetime, timedelta, timezone |
| from jose import JWTError, jwt |
| import bcrypt |
| from dotenv import load_dotenv |
|
|
| load_dotenv() |
|
|
| SECRET_KEY = os.environ.get("JWT_SECRET_KEY") |
| if not SECRET_KEY: |
| raise ValueError("JWT_SECRET_KEY is not set in environment") |
|
|
| ALGORITHM = os.environ.get("JWT_ALGORITHM") |
| ACCESS_TOKEN_EXPIRE_MINUTES = int(os.environ.get("JWT_ACCESS_TOKEN_EXPIRE_MINUTES")) |
|
|
| def verify_password(plain_password: str, hashed_password: str) -> bool: |
| return bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8')) |
|
|
| def get_password_hash(password: str) -> str: |
| salt = bcrypt.gensalt() |
| return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8') |
|
|
| def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str: |
| to_encode = data.copy() |
| if expires_delta: |
| expire = datetime.now(timezone.utc) + expires_delta |
| else: |
| 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 |
|
|
| def decode_access_token(token: str) -> dict | None: |
| try: |
| payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) |
| return payload |
| except JWTError: |
| return None |
|
|