security.py•2.01 kB
from passlib.context import CryptContext
from jose import jwt
from datetime import datetime, timedelta
import uuid
from typing import Optional, Dict, Any
import os
from app.core.config import settings
# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify a password against a hash."""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""Generate a password hash."""
return pwd_context.hash(password)
# JWT token functions
def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
"""Create a new JWT access token."""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({
"exp": expire,
"iat": datetime.utcnow(),
"iss": settings.MCP_ISSUER_URL,
"jti": str(uuid.uuid4())
})
return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
def create_refresh_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
"""Create a new JWT refresh token."""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
to_encode.update({
"exp": expire,
"iat": datetime.utcnow(),
"iss": settings.MCP_ISSUER_URL,
"jti": str(uuid.uuid4()),
"token_type": "refresh"
})
return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
def decode_token(token: str) -> Dict[str, Any]:
"""Decode a JWT token."""
return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])