Changed module name from my_auth to myauth
Changed encryption algorithm to argon2 Added unit tests
This commit is contained in:
6
src/myauth/__init__.py
Normal file
6
src/myauth/__init__.py
Normal file
@@ -0,0 +1,6 @@
|
||||
from .factory import create_sqlite_auth_service, create_app_router_for_sqlite
|
||||
|
||||
__all__ = [
|
||||
'create_sqlite_auth_service',
|
||||
'create_app_router_for_sqlite',
|
||||
]
|
||||
3
src/myauth/api/__init__.py
Normal file
3
src/myauth/api/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
from .routes import create_auth_router
|
||||
|
||||
__all__ = ["create_auth_router"]
|
||||
379
src/myauth/api/routes.py
Normal file
379
src/myauth/api/routes.py
Normal file
@@ -0,0 +1,379 @@
|
||||
"""
|
||||
FastAPI routes for authentication module.
|
||||
|
||||
This module provides ready-to-use FastAPI routes for all authentication
|
||||
operations. Routes are organized in an APIRouter with /auth prefix.
|
||||
"""
|
||||
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
|
||||
from ..core.auth import AuthService
|
||||
from ..exceptions import AuthError
|
||||
from ..models.email_verification import (
|
||||
EmailVerificationRequest,
|
||||
PasswordResetRequest,
|
||||
PasswordResetConfirm
|
||||
)
|
||||
from ..models.token import AccessTokenResponse, RefreshTokenRequest
|
||||
from ..models.user import UserCreate, UserResponse
|
||||
|
||||
# OAuth2 scheme for token authentication
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
|
||||
|
||||
|
||||
def create_auth_router(auth_service: AuthService) -> APIRouter:
|
||||
"""
|
||||
Create and configure the authentication router.
|
||||
|
||||
This factory function creates an APIRouter with all authentication
|
||||
endpoints configured. The router includes automatic exception handling
|
||||
for authentication errors.
|
||||
|
||||
Args:
|
||||
auth_service: Configured authentication service instance.
|
||||
|
||||
Returns:
|
||||
Configured APIRouter ready to be included in a FastAPI app.
|
||||
|
||||
Example:
|
||||
>>> from fastapi import FastAPI
|
||||
>>> from auth_module.api.routes import create_auth_router
|
||||
>>>
|
||||
>>> app = FastAPI()
|
||||
>>> auth_router = create_auth_router(auth_service)
|
||||
>>> app.include_router(auth_router)
|
||||
"""
|
||||
router = APIRouter(prefix="/auth", tags=["authentication"])
|
||||
|
||||
def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> UserResponse:
|
||||
"""
|
||||
Dependency to extract and validate the current user from access token.
|
||||
|
||||
This dependency can be used in any route that requires authentication.
|
||||
It extracts the Bearer token from the Authorization header and validates it.
|
||||
|
||||
Args:
|
||||
token: JWT access token from Authorization header.
|
||||
|
||||
Returns:
|
||||
The current authenticated user (as UserResponse).
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if token is invalid or expired.
|
||||
|
||||
Example:
|
||||
>>> @app.get("/protected")
|
||||
>>> def protected_route(user: UserResponse = Depends(get_current_user)):
|
||||
>>> return {"user_id": user.id}
|
||||
"""
|
||||
try:
|
||||
user = auth_service.get_current_user(token)
|
||||
return UserResponse(
|
||||
id=user.id,
|
||||
email=user.email,
|
||||
username=user.username,
|
||||
roles=user.roles,
|
||||
user_settings=user.user_settings,
|
||||
created_at=user.created_at,
|
||||
updated_at=user.updated_at
|
||||
)
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/register",
|
||||
response_model=UserResponse,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
summary="Register a new user",
|
||||
description="Create a new user account with email, username, and password."
|
||||
)
|
||||
def register(user_data: UserCreate) -> UserResponse:
|
||||
"""
|
||||
Register a new user.
|
||||
|
||||
Creates a new user account with the provided credentials. The password
|
||||
is automatically hashed before storage. Email verification is optional
|
||||
and the account is created with is_verified=False.
|
||||
|
||||
Args:
|
||||
user_data: User registration data including email, username, and password.
|
||||
|
||||
Returns:
|
||||
The created user information (without password).
|
||||
|
||||
Raises:
|
||||
HTTPException: 409 if email already exists.
|
||||
HTTPException: 422 if validation fails (password strength, etc.).
|
||||
"""
|
||||
try:
|
||||
user = auth_service.register(user_data)
|
||||
return UserResponse(
|
||||
id=user.id,
|
||||
email=user.email,
|
||||
username=user.username,
|
||||
roles=user.roles,
|
||||
user_settings=user.user_settings,
|
||||
created_at=user.created_at,
|
||||
updated_at=user.updated_at
|
||||
)
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/login",
|
||||
response_model=AccessTokenResponse,
|
||||
summary="Login with email and password",
|
||||
description="Authenticate a user and receive access and refresh tokens."
|
||||
)
|
||||
def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]) -> AccessTokenResponse:
|
||||
"""
|
||||
Authenticate a user and generate tokens.
|
||||
|
||||
This endpoint accepts form data with username (which should contain the email)
|
||||
and password. It returns both access and refresh tokens upon successful authentication.
|
||||
|
||||
Note: OAuth2PasswordRequestForm uses 'username' field, but we treat it as email.
|
||||
|
||||
Args:
|
||||
form_data: OAuth2 form with username (email) and password fields.
|
||||
|
||||
Returns:
|
||||
Access token (JWT) and refresh token with token_type="bearer".
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if credentials are invalid.
|
||||
HTTPException: 403 if account is disabled.
|
||||
"""
|
||||
try:
|
||||
# OAuth2PasswordRequestForm uses 'username' but we treat it as email
|
||||
user, tokens = auth_service.login(form_data.username, form_data.password)
|
||||
return tokens
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/logout",
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
summary="Logout user",
|
||||
description="Revoke the refresh token to logout the user."
|
||||
)
|
||||
def logout(request: RefreshTokenRequest) -> None:
|
||||
"""
|
||||
Logout a user by revoking their refresh token.
|
||||
|
||||
This prevents the refresh token from being used to obtain new access tokens.
|
||||
The current access token remains valid until it expires naturally (30 minutes).
|
||||
|
||||
Args:
|
||||
request: Request body containing the refresh token to revoke.
|
||||
|
||||
Returns:
|
||||
204 No Content on success.
|
||||
"""
|
||||
auth_service.logout(request.refresh_token)
|
||||
return None
|
||||
|
||||
@router.post(
|
||||
"/refresh",
|
||||
response_model=AccessTokenResponse,
|
||||
summary="Refresh access token",
|
||||
description="Exchange a refresh token for new access and refresh tokens."
|
||||
)
|
||||
def refresh_token(request: RefreshTokenRequest) -> AccessTokenResponse:
|
||||
"""
|
||||
Obtain a new access token using a refresh token.
|
||||
|
||||
This endpoint allows clients to obtain a new access token without
|
||||
requiring the user to re-enter their password. The old refresh token
|
||||
is revoked and a new one is issued.
|
||||
|
||||
Args:
|
||||
request: Request body containing the refresh token.
|
||||
|
||||
Returns:
|
||||
New access token and refresh token.
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if refresh token is invalid, expired, or revoked.
|
||||
"""
|
||||
try:
|
||||
tokens = auth_service.refresh_access_token(request.refresh_token)
|
||||
return tokens
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/password-reset-request",
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Request password reset",
|
||||
description="Generate a password reset token and return it (to be sent via email)."
|
||||
)
|
||||
def request_password_reset(request: PasswordResetRequest) -> dict:
|
||||
"""
|
||||
Request a password reset token.
|
||||
|
||||
This endpoint generates a secure token that can be used to reset the password.
|
||||
In production, this token should be sent via email. For this module, the token
|
||||
is returned in the response so the consuming application can handle email delivery.
|
||||
|
||||
Args:
|
||||
request: Request body containing the email address.
|
||||
|
||||
Returns:
|
||||
Dictionary with the reset token and a message.
|
||||
|
||||
Raises:
|
||||
HTTPException: 404 if email is not registered.
|
||||
"""
|
||||
try:
|
||||
token = auth_service.request_password_reset(request.email)
|
||||
return {
|
||||
"message": "Password reset token generated",
|
||||
"token": token
|
||||
}
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/password-reset",
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Reset password with token",
|
||||
description="Reset user password using a valid reset token."
|
||||
)
|
||||
def reset_password(request: PasswordResetConfirm) -> dict:
|
||||
"""
|
||||
Reset a user's password using a reset token.
|
||||
|
||||
This endpoint validates the reset token and updates the user's password.
|
||||
All existing refresh tokens are revoked for security.
|
||||
|
||||
Args:
|
||||
request: Request body containing the reset token and new password.
|
||||
|
||||
Returns:
|
||||
Success message.
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if token is invalid, expired, or already used.
|
||||
HTTPException: 422 if new password doesn't meet requirements.
|
||||
"""
|
||||
try:
|
||||
auth_service.reset_password(request.token, request.new_password)
|
||||
return {"message": "Password reset successfully"}
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/verify-email-request",
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Request email verification",
|
||||
description="Generate an email verification token and return it (to be sent via email)."
|
||||
)
|
||||
def request_email_verification(request: EmailVerificationRequest) -> dict:
|
||||
"""
|
||||
Request an email verification token.
|
||||
|
||||
This endpoint generates a JWT token for email verification. In production,
|
||||
this token should be sent via email with a verification link. For this module,
|
||||
the token is returned in the response so the consuming application can handle
|
||||
email delivery.
|
||||
|
||||
Args:
|
||||
request: Request body containing the email address.
|
||||
|
||||
Returns:
|
||||
Dictionary with the verification token and a message.
|
||||
|
||||
Raises:
|
||||
HTTPException: 404 if email is not registered.
|
||||
"""
|
||||
try:
|
||||
token = auth_service.request_email_verification(request.email)
|
||||
return {
|
||||
"message": "Email verification token generated",
|
||||
"token": token
|
||||
}
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.get(
|
||||
"/verify-email",
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Verify email with token",
|
||||
description="Verify user email using a verification token from query parameter."
|
||||
)
|
||||
def verify_email(token: str) -> dict:
|
||||
"""
|
||||
Verify a user's email address.
|
||||
|
||||
This endpoint validates the verification token and marks the user's
|
||||
email as verified. It uses a query parameter so it can be easily
|
||||
accessed via a link in an email.
|
||||
|
||||
Args:
|
||||
token: Email verification token (JWT) from query parameter.
|
||||
|
||||
Returns:
|
||||
Success message.
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if token is invalid or expired.
|
||||
"""
|
||||
try:
|
||||
auth_service.verify_email(token)
|
||||
return {"message": "Email verified successfully"}
|
||||
except AuthError as e:
|
||||
raise HTTPException(
|
||||
status_code=e.status_code,
|
||||
detail=e.message
|
||||
)
|
||||
|
||||
@router.get(
|
||||
"/me",
|
||||
response_model=UserResponse,
|
||||
summary="Get current user",
|
||||
description="Get information about the currently authenticated user."
|
||||
)
|
||||
def get_me(current_user: Annotated[UserResponse, Depends(get_current_user)]) -> UserResponse:
|
||||
"""
|
||||
Get current authenticated user information.
|
||||
|
||||
This is a protected route that requires a valid access token in the
|
||||
Authorization header (Bearer token).
|
||||
|
||||
Args:
|
||||
current_user: The authenticated user (injected by dependency).
|
||||
|
||||
Returns:
|
||||
Current user information.
|
||||
|
||||
Raises:
|
||||
HTTPException: 401 if token is invalid or expired.
|
||||
"""
|
||||
return current_user
|
||||
|
||||
return router
|
||||
5
src/myauth/core/__init__.py
Normal file
5
src/myauth/core/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
from .auth import AuthService
|
||||
from .password import PasswordManager
|
||||
from .token import TokenManager
|
||||
|
||||
__all__ = ["AuthService", "PasswordManager", "TokenManager"]
|
||||
446
src/myauth/core/auth.py
Normal file
446
src/myauth/core/auth.py
Normal file
@@ -0,0 +1,446 @@
|
||||
"""
|
||||
Authentication service for the authentication module.
|
||||
|
||||
This module provides the main authentication service that orchestrates
|
||||
all authentication operations including registration, login, token management,
|
||||
password reset, and email verification.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from .password import PasswordManager
|
||||
from .token import TokenManager
|
||||
from ..persistence.base import UserRepository, TokenRepository
|
||||
from ..models.user import UserCreate, UserInDB, UserUpdate
|
||||
from ..models.token import AccessTokenResponse, TokenData
|
||||
from ..emailing.base import EmailService
|
||||
from ..exceptions import (
|
||||
InvalidCredentialsError,
|
||||
UserNotFoundError,
|
||||
AccountDisabledError,
|
||||
ExpiredTokenError,
|
||||
InvalidTokenError,
|
||||
RevokedTokenError
|
||||
)
|
||||
|
||||
|
||||
class AuthService:
|
||||
"""
|
||||
Main authentication service.
|
||||
|
||||
This service orchestrates all authentication-related operations by
|
||||
coordinating between password management, token management, and
|
||||
persistence layers.
|
||||
|
||||
Attributes:
|
||||
user_repository: Repository for user persistence operations.
|
||||
token_repository: Repository for token persistence operations.
|
||||
password_manager: Manager for password hashing and verification.
|
||||
token_manager: Manager for token creation and validation.
|
||||
email_service: Optional service for sending emails.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
user_repository: UserRepository,
|
||||
token_repository: TokenRepository,
|
||||
password_manager: PasswordManager,
|
||||
token_manager: TokenManager,
|
||||
email_service: Optional[EmailService] = None
|
||||
):
|
||||
"""
|
||||
Initialize the authentication service.
|
||||
|
||||
Args:
|
||||
user_repository: Repository for user persistence.
|
||||
token_repository: Repository for token persistence.
|
||||
password_manager: Manager for password hashing and verification.
|
||||
token_manager: Manager for token creation and validation.
|
||||
email_service: Optional service for sending emails (password reset, verification).
|
||||
"""
|
||||
self.user_repository = user_repository
|
||||
self.token_repository = token_repository
|
||||
self.password_manager = password_manager
|
||||
self.token_manager = token_manager
|
||||
self.email_service = email_service
|
||||
|
||||
def register(self, user_data: UserCreate) -> UserInDB:
|
||||
"""
|
||||
Register a new user.
|
||||
|
||||
This method creates a new user account with hashed password.
|
||||
The user's email is initially unverified.
|
||||
|
||||
Args:
|
||||
user_data: User registration data including password.
|
||||
|
||||
Returns:
|
||||
The created user (without password).
|
||||
|
||||
Raises:
|
||||
UserAlreadyExistsError: If email is already registered.
|
||||
|
||||
Example:
|
||||
>>> user_data = UserCreate(
|
||||
... email="user@example.com",
|
||||
... username="john_doe",
|
||||
... password="SecurePass123!"
|
||||
... )
|
||||
>>> user = auth_service.register(user_data)
|
||||
"""
|
||||
hashed_password = self.password_manager.hash_password(user_data.password)
|
||||
user = self.user_repository.create_user(user_data, hashed_password)
|
||||
return user
|
||||
|
||||
def login(self, email: str, password: str) -> tuple[UserInDB, AccessTokenResponse]:
|
||||
"""
|
||||
Authenticate a user and create tokens.
|
||||
|
||||
This method verifies credentials, checks account status, and generates
|
||||
both access and refresh tokens.
|
||||
|
||||
Args:
|
||||
email: User's email address.
|
||||
password: User's plain text password.
|
||||
|
||||
Returns:
|
||||
Tuple of (user, tokens) where tokens contains access_token and refresh_token.
|
||||
|
||||
Raises:
|
||||
InvalidCredentialsError: If email or password is incorrect.
|
||||
AccountDisabledError: If user account is disabled.
|
||||
|
||||
Example:
|
||||
>>> user, tokens = auth_service.login("user@example.com", "password")
|
||||
>>> print(tokens.access_token)
|
||||
"""
|
||||
# Get user by email
|
||||
user = self.user_repository.get_user_by_email(email)
|
||||
if not user:
|
||||
raise InvalidCredentialsError()
|
||||
|
||||
# Verify password
|
||||
if not self.password_manager.verify_password(password, user.hashed_password):
|
||||
raise InvalidCredentialsError()
|
||||
|
||||
# Check if account is active
|
||||
if not user.is_active:
|
||||
raise AccountDisabledError()
|
||||
|
||||
# Create tokens
|
||||
access_token = self.token_manager.create_access_token(user)
|
||||
refresh_token = self.token_manager.create_refresh_token()
|
||||
|
||||
# Store refresh token in database
|
||||
token_data = TokenData(
|
||||
token=refresh_token,
|
||||
token_type="refresh",
|
||||
user_id=user.id,
|
||||
expires_at=self.token_manager.get_refresh_token_expiration(),
|
||||
created_at=datetime.utcnow(),
|
||||
is_revoked=False
|
||||
)
|
||||
self.token_repository.save_token(token_data)
|
||||
|
||||
tokens = AccessTokenResponse(
|
||||
access_token=access_token,
|
||||
refresh_token=refresh_token,
|
||||
token_type="bearer"
|
||||
)
|
||||
|
||||
return user, tokens
|
||||
|
||||
def refresh_access_token(self, refresh_token: str) -> AccessTokenResponse:
|
||||
"""
|
||||
Create a new access token using a refresh token.
|
||||
|
||||
This method validates the refresh token and generates a new access token
|
||||
without requiring the user to re-enter their password.
|
||||
|
||||
Args:
|
||||
refresh_token: The refresh token to exchange.
|
||||
|
||||
Returns:
|
||||
New access and refresh tokens.
|
||||
|
||||
Raises:
|
||||
InvalidTokenError: If refresh token is invalid.
|
||||
ExpiredTokenError: If refresh token has expired.
|
||||
RevokedTokenError: If refresh token has been revoked.
|
||||
UserNotFoundError: If user no longer exists.
|
||||
AccountDisabledError: If user account is disabled.
|
||||
|
||||
Example:
|
||||
>>> tokens = auth_service.refresh_access_token(old_refresh_token)
|
||||
>>> print(tokens.access_token)
|
||||
"""
|
||||
# Validate refresh token
|
||||
token_data = self.token_repository.get_token(refresh_token, "refresh")
|
||||
if not token_data:
|
||||
raise InvalidTokenError("Invalid refresh token")
|
||||
|
||||
if token_data.is_revoked:
|
||||
raise RevokedTokenError()
|
||||
|
||||
if token_data.expires_at < datetime.utcnow():
|
||||
raise ExpiredTokenError()
|
||||
|
||||
# Get user
|
||||
user = self.user_repository.get_user_by_id(token_data.user_id)
|
||||
if not user:
|
||||
raise UserNotFoundError()
|
||||
|
||||
if not user.is_active:
|
||||
raise AccountDisabledError()
|
||||
|
||||
# Create new tokens
|
||||
access_token = self.token_manager.create_access_token(user)
|
||||
new_refresh_token = self.token_manager.create_refresh_token()
|
||||
|
||||
# Revoke old refresh token
|
||||
self.token_repository.revoke_token(refresh_token)
|
||||
|
||||
# Store new refresh token
|
||||
new_token_data = TokenData(
|
||||
token=new_refresh_token,
|
||||
token_type="refresh",
|
||||
user_id=user.id,
|
||||
expires_at=self.token_manager.get_refresh_token_expiration(),
|
||||
created_at=datetime.utcnow(),
|
||||
is_revoked=False
|
||||
)
|
||||
self.token_repository.save_token(new_token_data)
|
||||
|
||||
return AccessTokenResponse(
|
||||
access_token=access_token,
|
||||
refresh_token=new_refresh_token,
|
||||
token_type="bearer"
|
||||
)
|
||||
|
||||
def logout(self, refresh_token: str) -> bool:
|
||||
"""
|
||||
Logout a user by revoking their refresh token.
|
||||
|
||||
This prevents the refresh token from being used to obtain new
|
||||
access tokens. The current access token will remain valid until
|
||||
it expires naturally.
|
||||
|
||||
Args:
|
||||
refresh_token: The refresh token to revoke.
|
||||
|
||||
Returns:
|
||||
True if logout was successful, False if token not found.
|
||||
|
||||
Example:
|
||||
>>> auth_service.logout(refresh_token)
|
||||
True
|
||||
"""
|
||||
return self.token_repository.revoke_token(refresh_token)
|
||||
|
||||
def request_password_reset(self, email: str) -> str:
|
||||
"""
|
||||
Generate a password reset token for a user.
|
||||
|
||||
This method creates a secure token that can be sent to the user
|
||||
via email to reset their password. If an email service is configured,
|
||||
the email is sent automatically. Otherwise, the token is returned
|
||||
for manual handling.
|
||||
|
||||
Args:
|
||||
email: User's email address.
|
||||
|
||||
Returns:
|
||||
The password reset token to be sent via email.
|
||||
|
||||
Raises:
|
||||
UserNotFoundError: If email is not registered.
|
||||
|
||||
Example:
|
||||
>>> token = auth_service.request_password_reset("user@example.com")
|
||||
>>> # Token is automatically sent via email if service is configured
|
||||
"""
|
||||
user = self.user_repository.get_user_by_email(email)
|
||||
if not user:
|
||||
raise UserNotFoundError(f"No user found with email {email}")
|
||||
|
||||
# Create reset token
|
||||
reset_token = self.token_manager.create_password_reset_token()
|
||||
|
||||
# Store token in database
|
||||
token_data = TokenData(
|
||||
token=reset_token,
|
||||
token_type="password_reset",
|
||||
user_id=user.id,
|
||||
expires_at=self.token_manager.get_password_reset_token_expiration(),
|
||||
created_at=datetime.utcnow(),
|
||||
is_revoked=False
|
||||
)
|
||||
self.token_repository.save_token(token_data)
|
||||
|
||||
# Send email if service is configured
|
||||
if self.email_service:
|
||||
self.email_service.send_password_reset_email(email, reset_token)
|
||||
|
||||
return reset_token
|
||||
|
||||
def reset_password(self, token: str, new_password: str) -> bool:
|
||||
"""
|
||||
Reset a user's password using a reset token.
|
||||
|
||||
This method validates the reset token and updates the user's password.
|
||||
All existing refresh tokens for the user are revoked for security.
|
||||
|
||||
Args:
|
||||
token: Password reset token.
|
||||
new_password: New plain text password (will be hashed).
|
||||
|
||||
Returns:
|
||||
True if password was reset successfully.
|
||||
|
||||
Raises:
|
||||
InvalidTokenError: If reset token is invalid.
|
||||
ExpiredTokenError: If reset token has expired.
|
||||
RevokedTokenError: If reset token has been used.
|
||||
UserNotFoundError: If user no longer exists.
|
||||
|
||||
Example:
|
||||
>>> auth_service.reset_password(token, "NewSecurePass123!")
|
||||
True
|
||||
"""
|
||||
# Validate reset token
|
||||
token_data = self.token_repository.get_token(token, "password_reset")
|
||||
if not token_data:
|
||||
raise InvalidTokenError("Invalid password reset token")
|
||||
|
||||
if token_data.is_revoked:
|
||||
raise RevokedTokenError("Password reset token has already been used")
|
||||
|
||||
if token_data.expires_at < datetime.utcnow():
|
||||
raise ExpiredTokenError("Password reset token has expired")
|
||||
|
||||
# Get user
|
||||
user = self.user_repository.get_user_by_id(token_data.user_id)
|
||||
if not user:
|
||||
raise UserNotFoundError()
|
||||
|
||||
# Hash new password
|
||||
hashed_password = self.password_manager.hash_password(new_password)
|
||||
|
||||
# Update user password
|
||||
updates = UserUpdate(password=hashed_password)
|
||||
self.user_repository.update_user(user.id, updates)
|
||||
|
||||
# Revoke the reset token
|
||||
self.token_repository.revoke_token(token)
|
||||
|
||||
# Revoke all user's refresh tokens for security
|
||||
self.token_repository.revoke_all_user_tokens(user.id, "refresh")
|
||||
|
||||
return True
|
||||
|
||||
def request_email_verification(self, email: str) -> str:
|
||||
"""
|
||||
Generate an email verification token for a user.
|
||||
|
||||
This method creates a JWT token that can be sent to the user
|
||||
to verify their email address. If an email service is configured,
|
||||
the email is sent automatically. Otherwise, the token is returned
|
||||
for manual handling.
|
||||
|
||||
Args:
|
||||
email: User's email address.
|
||||
|
||||
Returns:
|
||||
The email verification token (JWT) to be sent via email.
|
||||
|
||||
Raises:
|
||||
UserNotFoundError: If email is not registered.
|
||||
|
||||
Example:
|
||||
>>> token = auth_service.request_email_verification("user@example.com")
|
||||
>>> # Token is automatically sent via email if service is configured
|
||||
"""
|
||||
user = self.user_repository.get_user_by_email(email)
|
||||
if not user:
|
||||
raise UserNotFoundError(f"No user found with email {email}")
|
||||
|
||||
verification_token = self.token_manager.create_email_verification_token(email)
|
||||
|
||||
# Send email if service is configured
|
||||
if self.email_service:
|
||||
self.email_service.send_verification_email(email, verification_token)
|
||||
|
||||
return verification_token
|
||||
|
||||
def verify_email(self, token: str) -> bool:
|
||||
"""
|
||||
Verify a user's email address using a verification token.
|
||||
|
||||
This method decodes the JWT token, extracts the email, and marks
|
||||
the user's email as verified.
|
||||
|
||||
Args:
|
||||
token: Email verification token (JWT).
|
||||
|
||||
Returns:
|
||||
True if email was verified successfully.
|
||||
|
||||
Raises:
|
||||
InvalidTokenError: If verification token is invalid.
|
||||
ExpiredTokenError: If verification token has expired.
|
||||
UserNotFoundError: If user no longer exists.
|
||||
|
||||
Example:
|
||||
>>> auth_service.verify_email(token)
|
||||
True
|
||||
"""
|
||||
# Decode and validate token
|
||||
email = self.token_manager.decode_email_verification_token(token)
|
||||
|
||||
# Get user
|
||||
user = self.user_repository.get_user_by_email(email)
|
||||
if not user:
|
||||
raise UserNotFoundError(f"No user found with email {email}")
|
||||
|
||||
# Update user's verified status
|
||||
updates = UserUpdate(is_verified=True)
|
||||
self.user_repository.update_user(user.id, updates)
|
||||
|
||||
return True
|
||||
|
||||
def get_current_user(self, access_token: str) -> UserInDB:
|
||||
"""
|
||||
Retrieve the current user from an access token.
|
||||
|
||||
This method decodes and validates the JWT access token and
|
||||
retrieves the corresponding user from the database.
|
||||
|
||||
Args:
|
||||
access_token: JWT access token.
|
||||
|
||||
Returns:
|
||||
The user associated with the token.
|
||||
|
||||
Raises:
|
||||
InvalidTokenError: If access token is invalid.
|
||||
ExpiredTokenError: If access token has expired.
|
||||
UserNotFoundError: If user no longer exists.
|
||||
AccountDisabledError: If user account is disabled.
|
||||
|
||||
Example:
|
||||
>>> user = auth_service.get_current_user(access_token)
|
||||
>>> print(user.email)
|
||||
"""
|
||||
# Decode and validate token
|
||||
token_payload = self.token_manager.decode_access_token(access_token)
|
||||
|
||||
# Get user
|
||||
user = self.user_repository.get_user_by_id(token_payload.sub)
|
||||
if not user:
|
||||
raise UserNotFoundError()
|
||||
|
||||
if not user.is_active:
|
||||
raise AccountDisabledError()
|
||||
|
||||
return user
|
||||
77
src/myauth/core/password.py
Normal file
77
src/myauth/core/password.py
Normal file
@@ -0,0 +1,77 @@
|
||||
"""
|
||||
Password management for authentication module.
|
||||
|
||||
This module provides password hashing and verification functionality using
|
||||
bcrypt. It handles secure password storage and validation.
|
||||
"""
|
||||
|
||||
from passlib.context import CryptContext
|
||||
|
||||
|
||||
class PasswordManager:
|
||||
"""
|
||||
Manager for password hashing and verification operations.
|
||||
|
||||
This class uses argon2 for secure password hashing
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""
|
||||
Initialize the password manager.
|
||||
|
||||
"""
|
||||
self._context = CryptContext(
|
||||
schemes=["argon2"],
|
||||
deprecated="auto",
|
||||
# argon2__time_cost=3, # number of iterations (increases CPU time)
|
||||
# argon2__memory_cost=65536, # memory usage in KiB (64 MiB)
|
||||
# argon2__parallelism=2, # number of parallel threads
|
||||
# argon2__salt_len=16 # length of the random salt in bytes
|
||||
)
|
||||
|
||||
def hash_password(self, password: str) -> str:
|
||||
"""
|
||||
Hash a plain text password using bcrypt.
|
||||
|
||||
This method creates a secure hash of the password that can be
|
||||
safely stored in the database. Each hash includes a salt, making
|
||||
rainbow table attacks ineffective.
|
||||
|
||||
Args:
|
||||
password: Plain text password to hash.
|
||||
|
||||
Returns:
|
||||
Bcrypt hashed password string.
|
||||
|
||||
Example:
|
||||
>>> pm = PasswordManager()
|
||||
>>> hashed = pm.hash_password("SecurePassword123!")
|
||||
>>> print(hashed)
|
||||
$2b$12$abcdefghijklmnopqrstuvwxyz...
|
||||
"""
|
||||
return self._context.hash(password)
|
||||
|
||||
def verify_password(self, plain_password: str, hashed_password: str) -> bool:
|
||||
"""
|
||||
Verify a plain text password against a hashed password.
|
||||
|
||||
This method performs constant-time comparison to prevent timing
|
||||
attacks. It returns True only if the plain password matches the
|
||||
hash.
|
||||
|
||||
Args:
|
||||
plain_password: Plain text password to verify.
|
||||
hashed_password: Bcrypt hashed password from database.
|
||||
|
||||
Returns:
|
||||
True if password matches, False otherwise.
|
||||
|
||||
Example:
|
||||
>>> pm = PasswordManager()
|
||||
>>> hashed = pm.hash_password("SecurePassword123!")
|
||||
>>> pm.verify_password("SecurePassword123!", hashed)
|
||||
True
|
||||
>>> pm.verify_password("WrongPassword", hashed)
|
||||
False
|
||||
"""
|
||||
return self._context.verify(plain_password, hashed_password)
|
||||
265
src/myauth/core/token.py
Normal file
265
src/myauth/core/token.py
Normal file
@@ -0,0 +1,265 @@
|
||||
"""
|
||||
Token management for authentication module.
|
||||
|
||||
This module provides functionality for creating and validating different types
|
||||
of tokens: JWT access tokens, opaque refresh tokens, password reset tokens,
|
||||
and email verification tokens.
|
||||
"""
|
||||
|
||||
import secrets
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from jose import JWTError, jwt
|
||||
|
||||
from ..exceptions import InvalidTokenError, ExpiredTokenError
|
||||
from ..models.token import TokenPayload
|
||||
from ..models.user import UserInDB
|
||||
|
||||
|
||||
class TokenManager:
|
||||
"""
|
||||
Manager for token creation and validation operations.
|
||||
|
||||
This class handles multiple token types:
|
||||
- JWT access tokens (stateless, short-lived)
|
||||
- Opaque refresh tokens (stored in DB, long-lived)
|
||||
- Opaque password reset tokens (stored in DB, short-lived)
|
||||
- JWT email verification tokens (stateless)
|
||||
|
||||
Attributes:
|
||||
jwt_secret: Secret key for signing JWT tokens.
|
||||
jwt_algorithm: Algorithm for JWT encoding (default: HS256).
|
||||
access_token_expire_minutes: Validity duration for access tokens.
|
||||
refresh_token_expire_days: Validity duration for refresh tokens.
|
||||
password_reset_token_expire_minutes: Validity duration for reset tokens.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
jwt_secret: str,
|
||||
jwt_algorithm: str = "HS256",
|
||||
access_token_expire_minutes: int = 30,
|
||||
refresh_token_expire_days: int = 7,
|
||||
password_reset_token_expire_minutes: int = 15
|
||||
):
|
||||
"""
|
||||
Initialize the token manager.
|
||||
|
||||
Args:
|
||||
jwt_secret: Secret key for signing JWT tokens.
|
||||
jwt_algorithm: Algorithm for JWT encoding (default: HS256).
|
||||
access_token_expire_minutes: Validity duration for access tokens.
|
||||
refresh_token_expire_days: Validity duration for refresh tokens.
|
||||
password_reset_token_expire_minutes: Validity duration for reset tokens.
|
||||
"""
|
||||
if not jwt_secret:
|
||||
raise ValueError("JWT secret cannot be empty")
|
||||
|
||||
self.jwt_secret = jwt_secret
|
||||
self.jwt_algorithm = jwt_algorithm
|
||||
self.access_token_expire_minutes = access_token_expire_minutes
|
||||
self.refresh_token_expire_days = refresh_token_expire_days
|
||||
self.password_reset_token_expire_minutes = password_reset_token_expire_minutes
|
||||
|
||||
def create_access_token(self, user: UserInDB) -> str:
|
||||
"""
|
||||
Create a JWT access token for a user.
|
||||
|
||||
The token contains the user's ID and email, and is signed with the
|
||||
configured secret. It expires after the configured duration.
|
||||
|
||||
Args:
|
||||
user: The user for whom to create the token.
|
||||
|
||||
Returns:
|
||||
Encoded JWT access token string.
|
||||
|
||||
Example:
|
||||
>>> tm = TokenManager(jwt_secret="secret")
|
||||
>>> token = tm.create_access_token(user)
|
||||
>>> print(token)
|
||||
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...
|
||||
"""
|
||||
expires_at = datetime.now() + timedelta(minutes=self.access_token_expire_minutes)
|
||||
|
||||
payload = {
|
||||
"sub": user.id,
|
||||
"email": user.email,
|
||||
"exp": int(expires_at.timestamp()),
|
||||
"type": "access"
|
||||
}
|
||||
|
||||
return jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm)
|
||||
|
||||
def create_refresh_token(self) -> str:
|
||||
"""
|
||||
Create an opaque refresh token.
|
||||
|
||||
This generates a cryptographically secure random token that will be
|
||||
stored in the database along with its expiration time and user association.
|
||||
|
||||
Returns:
|
||||
Secure random token string (64 characters hex).
|
||||
|
||||
Example:
|
||||
>>> tm = TokenManager(jwt_secret="secret")
|
||||
>>> token = tm.create_refresh_token()
|
||||
>>> len(token)
|
||||
64
|
||||
"""
|
||||
return secrets.token_hex(32)
|
||||
|
||||
def create_password_reset_token(self) -> str:
|
||||
"""
|
||||
Create an opaque password reset token.
|
||||
|
||||
This generates a cryptographically secure random token that will be
|
||||
stored in the database with a short expiration time (15 minutes by default).
|
||||
|
||||
Returns:
|
||||
Secure random token string (64 characters hex).
|
||||
|
||||
Example:
|
||||
>>> tm = TokenManager(jwt_secret="secret")
|
||||
>>> token = tm.create_password_reset_token()
|
||||
>>> len(token)
|
||||
64
|
||||
"""
|
||||
return secrets.token_hex(32)
|
||||
|
||||
def create_email_verification_token(self, email: str) -> str:
|
||||
"""
|
||||
Create a JWT email verification token.
|
||||
|
||||
This token is stateless (not stored in DB) and contains the email
|
||||
to be verified. It has no explicit expiration in the payload but
|
||||
should be validated for reasonable freshness.
|
||||
|
||||
Args:
|
||||
email: The email address to encode in the token.
|
||||
|
||||
Returns:
|
||||
Encoded JWT verification token string.
|
||||
|
||||
Example:
|
||||
>>> tm = TokenManager(jwt_secret="secret")
|
||||
>>> token = tm.create_email_verification_token("user@example.com")
|
||||
"""
|
||||
# Set expiration to 7 days for email verification
|
||||
expires_at = datetime.now() + timedelta(days=7)
|
||||
|
||||
payload = {
|
||||
"email": email,
|
||||
"exp": int(expires_at.timestamp()),
|
||||
"type": "email_verification"
|
||||
}
|
||||
|
||||
return jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm)
|
||||
|
||||
def decode_access_token(self, token: str) -> TokenPayload:
|
||||
"""
|
||||
Decode and validate a JWT access token.
|
||||
|
||||
This method verifies the token signature, checks expiration,
|
||||
and validates the token type.
|
||||
|
||||
Args:
|
||||
token: The JWT access token to decode.
|
||||
|
||||
Returns:
|
||||
TokenPayload containing the decoded claims.
|
||||
|
||||
Raises:
|
||||
InvalidTokenError: If token is malformed or has invalid signature.
|
||||
ExpiredTokenError: If token has expired.
|
||||
|
||||
Example:
|
||||
>>> tm = TokenManager(jwt_secret="secret")
|
||||
>>> payload = tm.decode_access_token(token)
|
||||
>>> print(payload.sub) # user_id
|
||||
"""
|
||||
try:
|
||||
payload = jwt.decode(
|
||||
token,
|
||||
self.jwt_secret,
|
||||
algorithms=[self.jwt_algorithm]
|
||||
)
|
||||
|
||||
# Validate token type
|
||||
if payload.get("type") != "access":
|
||||
raise InvalidTokenError("Invalid token type")
|
||||
|
||||
return TokenPayload(
|
||||
sub=payload.get("sub"),
|
||||
email=payload.get("email"),
|
||||
exp=payload.get("exp"),
|
||||
type=payload.get("type")
|
||||
)
|
||||
|
||||
except JWTError as e:
|
||||
if "expired" in str(e).lower():
|
||||
raise ExpiredTokenError("Access token has expired")
|
||||
raise InvalidTokenError(f"Invalid access token: {str(e)}")
|
||||
|
||||
def decode_email_verification_token(self, token: str) -> str:
|
||||
"""
|
||||
Decode and validate an email verification JWT token.
|
||||
|
||||
This method verifies the token signature, checks expiration,
|
||||
and extracts the email address.
|
||||
|
||||
Args:
|
||||
token: The JWT email verification token to decode.
|
||||
|
||||
Returns:
|
||||
The email address contained in the token.
|
||||
|
||||
Raises:
|
||||
InvalidTokenError: If token is malformed or has invalid signature.
|
||||
ExpiredTokenError: If token has expired.
|
||||
|
||||
Example:
|
||||
>>> tm = TokenManager(jwt_secret="secret")
|
||||
>>> email = tm.decode_email_verification_token(token)
|
||||
>>> print(email)
|
||||
user@example.com
|
||||
"""
|
||||
try:
|
||||
payload = jwt.decode(
|
||||
token,
|
||||
self.jwt_secret,
|
||||
algorithms=[self.jwt_algorithm]
|
||||
)
|
||||
|
||||
# Validate token type
|
||||
if payload.get("type") != "email_verification":
|
||||
raise InvalidTokenError("Invalid token type")
|
||||
|
||||
email = payload.get("email")
|
||||
if not email:
|
||||
raise InvalidTokenError("Email not found in token")
|
||||
|
||||
return email
|
||||
|
||||
except JWTError as e:
|
||||
if "expired" in str(e).lower():
|
||||
raise ExpiredTokenError("Email verification token has expired")
|
||||
raise InvalidTokenError(f"Invalid email verification token: {str(e)}")
|
||||
|
||||
def get_refresh_token_expiration(self) -> datetime:
|
||||
"""
|
||||
Calculate expiration datetime for a refresh token.
|
||||
|
||||
Returns:
|
||||
Datetime when a newly created refresh token should expire.
|
||||
"""
|
||||
return datetime.now() + timedelta(days=self.refresh_token_expire_days)
|
||||
|
||||
def get_password_reset_token_expiration(self) -> datetime:
|
||||
"""
|
||||
Calculate expiration datetime for a password reset token.
|
||||
|
||||
Returns:
|
||||
Datetime when a newly created reset token should expire.
|
||||
"""
|
||||
return datetime.now() + timedelta(minutes=self.password_reset_token_expire_minutes)
|
||||
4
src/myauth/emailing/__init__.py
Normal file
4
src/myauth/emailing/__init__.py
Normal file
@@ -0,0 +1,4 @@
|
||||
from .base import EmailService
|
||||
from .smtp import SMTPEmailService
|
||||
|
||||
__all__ = ["EmailService", "SMTPEmailService"]
|
||||
64
src/myauth/emailing/base.py
Normal file
64
src/myauth/emailing/base.py
Normal file
@@ -0,0 +1,64 @@
|
||||
"""
|
||||
Abstract base class for email service.
|
||||
|
||||
This module defines the interface that all email service implementations
|
||||
must follow. It provides methods for sending authentication-related emails.
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class EmailService(ABC):
|
||||
"""
|
||||
Abstract base class for email service operations.
|
||||
|
||||
This interface defines all methods required for sending authentication-related
|
||||
emails. Concrete implementations can use different email providers (SMTP,
|
||||
SendGrid, AWS SES, etc.).
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def send_verification_email(self, email: str, token: str) -> None:
|
||||
"""
|
||||
Send an email verification link to the user.
|
||||
|
||||
This method should send an email containing a verification link or token
|
||||
that the user can use to verify their email address.
|
||||
|
||||
Args:
|
||||
email: The recipient's email address.
|
||||
token: The verification token (JWT) to include in the email.
|
||||
|
||||
Raises:
|
||||
Exception: If email sending fails (implementation-specific).
|
||||
|
||||
Example:
|
||||
>>> email_service.send_verification_email(
|
||||
... "user@example.com",
|
||||
... "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
|
||||
... )
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def send_password_reset_email(self, email: str, token: str) -> None:
|
||||
"""
|
||||
Send a password reset link to the user.
|
||||
|
||||
This method should send an email containing a password reset link or token
|
||||
that the user can use to reset their password.
|
||||
|
||||
Args:
|
||||
email: The recipient's email address.
|
||||
token: The password reset token to include in the email.
|
||||
|
||||
Raises:
|
||||
Exception: If email sending fails (implementation-specific).
|
||||
|
||||
Example:
|
||||
>>> email_service.send_password_reset_email(
|
||||
... "user@example.com",
|
||||
... "a1b2c3d4e5f6..."
|
||||
... )
|
||||
"""
|
||||
pass
|
||||
211
src/myauth/emailing/smtp.py
Normal file
211
src/myauth/emailing/smtp.py
Normal file
@@ -0,0 +1,211 @@
|
||||
"""
|
||||
SMTP email service implementation.
|
||||
|
||||
This module provides an SMTP-based implementation of the email service
|
||||
for sending authentication-related emails.
|
||||
"""
|
||||
|
||||
import smtplib
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
from typing import Optional
|
||||
|
||||
from .base import EmailService
|
||||
|
||||
# Default email templates
|
||||
DEFAULT_VERIFICATION_TEMPLATE = """
|
||||
<html>
|
||||
<body>
|
||||
<h2>Email Verification</h2>
|
||||
<p>Please click the link below to verify your email address:</p>
|
||||
<p><a href="{verification_link}">Verify Email</a></p>
|
||||
<p>Or copy and paste this link into your browser:</p>
|
||||
<p>{verification_link}</p>
|
||||
<p>This link will expire in 7 days.</p>
|
||||
<p>If you did not request this verification, please ignore this email.</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
DEFAULT_PASSWORD_RESET_TEMPLATE = """
|
||||
<html>
|
||||
<body>
|
||||
<h2>Password Reset</h2>
|
||||
<p>You have requested to reset your password. Please click the link below:</p>
|
||||
<p><a href="{reset_link}">Reset Password</a></p>
|
||||
<p>Or copy and paste this link into your browser:</p>
|
||||
<p>{reset_link}</p>
|
||||
<p>This link will expire in 15 minutes.</p>
|
||||
<p>If you did not request a password reset, please ignore this email.</p>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
|
||||
class SMTPEmailService(EmailService):
|
||||
"""
|
||||
SMTP implementation of EmailService.
|
||||
|
||||
This implementation uses standard SMTP protocol to send emails.
|
||||
It supports both TLS and SSL connections and allows custom HTML
|
||||
templates for email content.
|
||||
|
||||
Attributes:
|
||||
host: SMTP server hostname.
|
||||
port: SMTP server port (587 for TLS, 465 for SSL).
|
||||
username: SMTP authentication username.
|
||||
password: SMTP authentication password.
|
||||
use_tls: Whether to use TLS (default: True).
|
||||
from_email: Email address to use as sender.
|
||||
from_name: Display name for sender (optional).
|
||||
base_url: Base URL for constructing verification/reset links.
|
||||
verification_template: HTML template for verification emails.
|
||||
password_reset_template: HTML template for password reset emails.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
host: str,
|
||||
port: int,
|
||||
username: str,
|
||||
password: str,
|
||||
from_email: str,
|
||||
base_url: str,
|
||||
use_tls: bool = True,
|
||||
from_name: Optional[str] = None,
|
||||
verification_template: Optional[str] = None,
|
||||
password_reset_template: Optional[str] = None
|
||||
):
|
||||
"""
|
||||
Initialize SMTP email service.
|
||||
|
||||
Args:
|
||||
host: SMTP server hostname (e.g., "smtp.gmail.com").
|
||||
port: SMTP server port (587 for TLS, 465 for SSL).
|
||||
username: SMTP authentication username.
|
||||
password: SMTP authentication password.
|
||||
from_email: Email address to use as sender.
|
||||
base_url: Base URL for your application (e.g., "https://myapp.com").
|
||||
use_tls: Whether to use TLS encryption (default: True).
|
||||
from_name: Display name for sender (default: None).
|
||||
verification_template: Custom HTML template for verification emails
|
||||
(must include {verification_link} placeholder).
|
||||
password_reset_template: Custom HTML template for reset emails
|
||||
(must include {reset_link} placeholder).
|
||||
|
||||
Example:
|
||||
>>> email_service = SMTPEmailService(
|
||||
... host="smtp.gmail.com",
|
||||
... port=587,
|
||||
... username="noreply@myapp.com",
|
||||
... password="app_password",
|
||||
... from_email="noreply@myapp.com",
|
||||
... base_url="https://myapp.com",
|
||||
... from_name="My Application"
|
||||
... )
|
||||
"""
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.username = username
|
||||
self.password = password
|
||||
self.use_tls = use_tls
|
||||
self.from_email = from_email
|
||||
self.from_name = from_name
|
||||
self.base_url = base_url.rstrip('/')
|
||||
|
||||
# Use custom templates or defaults
|
||||
self.verification_template = verification_template or DEFAULT_VERIFICATION_TEMPLATE
|
||||
self.password_reset_template = password_reset_template or DEFAULT_PASSWORD_RESET_TEMPLATE
|
||||
|
||||
def _send_email(self, to_email: str, subject: str, html_content: str) -> None:
|
||||
"""
|
||||
Send an email via SMTP.
|
||||
|
||||
Internal method that handles the actual SMTP connection and sending.
|
||||
|
||||
Args:
|
||||
to_email: Recipient email address.
|
||||
subject: Email subject line.
|
||||
html_content: HTML content of the email.
|
||||
|
||||
Raises:
|
||||
smtplib.SMTPException: If email sending fails.
|
||||
"""
|
||||
# Create message
|
||||
message = MIMEMultipart("alternative")
|
||||
message["Subject"] = subject
|
||||
message["From"] = f"{self.from_name} <{self.from_email}>" if self.from_name else self.from_email
|
||||
message["To"] = to_email
|
||||
|
||||
# Attach HTML content
|
||||
html_part = MIMEText(html_content, "html")
|
||||
message.attach(html_part)
|
||||
|
||||
# Send email
|
||||
if self.use_tls:
|
||||
with smtplib.SMTP(self.host, self.port) as server:
|
||||
server.starttls()
|
||||
server.login(self.username, self.password)
|
||||
server.send_message(message)
|
||||
else:
|
||||
with smtplib.SMTP_SSL(self.host, self.port) as server:
|
||||
server.login(self.username, self.password)
|
||||
server.send_message(message)
|
||||
|
||||
def send_verification_email(self, email: str, token: str) -> None:
|
||||
"""
|
||||
Send an email verification link to the user.
|
||||
|
||||
Constructs a verification link using the base_url and token,
|
||||
then sends an email using the configured verification template.
|
||||
|
||||
Args:
|
||||
email: The recipient's email address.
|
||||
token: The verification token (JWT).
|
||||
|
||||
Raises:
|
||||
smtplib.SMTPException: If email sending fails.
|
||||
|
||||
Example:
|
||||
>>> email_service.send_verification_email(
|
||||
... "user@example.com",
|
||||
... "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
|
||||
... )
|
||||
"""
|
||||
verification_link = f"{self.base_url}/auth/verify-email?token={token}"
|
||||
html_content = self.verification_template.format(verification_link=verification_link)
|
||||
|
||||
self._send_email(
|
||||
to_email=email,
|
||||
subject="Verify Your Email Address",
|
||||
html_content=html_content
|
||||
)
|
||||
|
||||
def send_password_reset_email(self, email: str, token: str) -> None:
|
||||
"""
|
||||
Send a password reset link to the user.
|
||||
|
||||
Constructs a password reset link using the base_url and token,
|
||||
then sends an email using the configured password reset template.
|
||||
|
||||
Args:
|
||||
email: The recipient's email address.
|
||||
token: The password reset token.
|
||||
|
||||
Raises:
|
||||
smtplib.SMTPException: If email sending fails.
|
||||
|
||||
Example:
|
||||
>>> email_service.send_password_reset_email(
|
||||
... "user@example.com",
|
||||
... "a1b2c3d4e5f6..."
|
||||
... )
|
||||
"""
|
||||
reset_link = f"{self.base_url}/reset-password?token={token}"
|
||||
html_content = self.password_reset_template.format(reset_link=reset_link)
|
||||
|
||||
self._send_email(
|
||||
to_email=email,
|
||||
subject="Reset Your Password",
|
||||
html_content=html_content
|
||||
)
|
||||
198
src/myauth/exceptions.py
Normal file
198
src/myauth/exceptions.py
Normal file
@@ -0,0 +1,198 @@
|
||||
"""
|
||||
Custom exceptions for authentication module.
|
||||
|
||||
This module defines all custom exceptions used throughout the authentication
|
||||
system. Each exception is designed to be caught and converted to appropriate
|
||||
HTTP responses by FastAPI exception handlers.
|
||||
"""
|
||||
|
||||
|
||||
class AuthError(Exception):
|
||||
"""
|
||||
Base exception for all authentication-related errors.
|
||||
|
||||
This base class provides common attributes for all auth exceptions
|
||||
and facilitates centralized error handling.
|
||||
|
||||
Attributes:
|
||||
message: Human-readable error message.
|
||||
status_code: HTTP status code associated with this error.
|
||||
error_code: Unique error code for client-side handling (optional).
|
||||
"""
|
||||
|
||||
def __init__(self, message: str, status_code: int = 500, error_code: str = None):
|
||||
"""
|
||||
Initialize the authentication error.
|
||||
|
||||
Args:
|
||||
message: Human-readable error message.
|
||||
status_code: HTTP status code (default: 500).
|
||||
error_code: Unique error code for client identification.
|
||||
"""
|
||||
self.message = message
|
||||
self.status_code = status_code
|
||||
self.error_code = error_code or self.__class__.__name__
|
||||
super().__init__(self.message)
|
||||
|
||||
|
||||
class InvalidCredentialsError(AuthError):
|
||||
"""
|
||||
Exception raised when login credentials are invalid.
|
||||
|
||||
This exception is raised during authentication when the provided
|
||||
email or password does not match any user in the database.
|
||||
|
||||
HTTP Status: 401 Unauthorized
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "Invalid email or password"):
|
||||
"""
|
||||
Initialize invalid credentials error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "Invalid email or password").
|
||||
"""
|
||||
super().__init__(message=message, status_code=401)
|
||||
|
||||
|
||||
class UserAlreadyExistsError(AuthError):
|
||||
"""
|
||||
Exception raised when attempting to create a user with an existing email.
|
||||
|
||||
This exception is raised during registration when the provided email
|
||||
address is already associated with an existing user account.
|
||||
|
||||
HTTP Status: 409 Conflict
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "User with this email already exists"):
|
||||
"""
|
||||
Initialize user already exists error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "User with this email already exists").
|
||||
"""
|
||||
super().__init__(message=message, status_code=409)
|
||||
|
||||
|
||||
class UserNotFoundError(AuthError):
|
||||
"""
|
||||
Exception raised when a requested user cannot be found.
|
||||
|
||||
This exception is raised when attempting to retrieve, update, or delete
|
||||
a user that does not exist in the database.
|
||||
|
||||
HTTP Status: 404 Not Found
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "User not found"):
|
||||
"""
|
||||
Initialize user not found error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "User not found").
|
||||
"""
|
||||
super().__init__(message=message, status_code=404)
|
||||
|
||||
|
||||
class InvalidTokenError(AuthError):
|
||||
"""
|
||||
Exception raised when a token is invalid or malformed.
|
||||
|
||||
This exception is raised when a token cannot be decoded, has an invalid
|
||||
signature, or contains invalid data. This applies to JWT tokens, refresh
|
||||
tokens, and password reset tokens.
|
||||
|
||||
HTTP Status: 401 Unauthorized
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "Invalid token"):
|
||||
"""
|
||||
Initialize invalid token error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "Invalid token").
|
||||
"""
|
||||
super().__init__(message=message, status_code=401)
|
||||
|
||||
|
||||
class ExpiredTokenError(AuthError):
|
||||
"""
|
||||
Exception raised when a token has expired.
|
||||
|
||||
This exception is raised when attempting to use a token that has
|
||||
passed its expiration time. This applies to all token types.
|
||||
|
||||
HTTP Status: 401 Unauthorized
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "Token has expired"):
|
||||
"""
|
||||
Initialize expired token error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "Token has expired").
|
||||
"""
|
||||
super().__init__(message=message, status_code=401)
|
||||
|
||||
|
||||
class RevokedTokenError(AuthError):
|
||||
"""
|
||||
Exception raised when attempting to use a revoked token.
|
||||
|
||||
This exception is raised when a token has been explicitly revoked,
|
||||
typically after logout or when a user's tokens are invalidated for
|
||||
security reasons.
|
||||
|
||||
HTTP Status: 401 Unauthorized
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "Token has been revoked"):
|
||||
"""
|
||||
Initialize revoked token error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "Token has been revoked").
|
||||
"""
|
||||
super().__init__(message=message, status_code=401)
|
||||
|
||||
|
||||
class EmailNotVerifiedError(AuthError):
|
||||
"""
|
||||
Exception raised when an action requires email verification.
|
||||
|
||||
This exception is raised when a user attempts to perform an action
|
||||
that requires their email to be verified, but their account email
|
||||
has not been verified yet.
|
||||
|
||||
HTTP Status: 403 Forbidden
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "Email address is not verified"):
|
||||
"""
|
||||
Initialize email not verified error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "Email address is not verified").
|
||||
"""
|
||||
super().__init__(message=message, status_code=403)
|
||||
|
||||
|
||||
class AccountDisabledError(AuthError):
|
||||
"""
|
||||
Exception raised when attempting to access a disabled account.
|
||||
|
||||
This exception is raised when a user attempts to login or perform
|
||||
actions with an account that has been disabled by an administrator.
|
||||
|
||||
HTTP Status: 403 Forbidden
|
||||
"""
|
||||
|
||||
def __init__(self, message: str = "Account has been disabled"):
|
||||
"""
|
||||
Initialize account disabled error.
|
||||
|
||||
Args:
|
||||
message: Custom error message (default: "Account has been disabled").
|
||||
"""
|
||||
super().__init__(message=message, status_code=403)
|
||||
78
src/myauth/factory.py
Normal file
78
src/myauth/factory.py
Normal file
@@ -0,0 +1,78 @@
|
||||
from typing import Optional
|
||||
|
||||
from .core import AuthService
|
||||
from .core.password import PasswordManager
|
||||
from .core.token import TokenManager
|
||||
from .emailing.base import EmailService
|
||||
|
||||
|
||||
def create_sqlite_auth_service(db_path: str,
|
||||
jwt_secret: str,
|
||||
email_service: Optional[EmailService],
|
||||
password_manager: PasswordManager = None,
|
||||
token_manager: TokenManager = None,
|
||||
):
|
||||
"""
|
||||
Creates and configures an authentication service using SQLite as the underlying
|
||||
data store.
|
||||
|
||||
This function is responsible for setting up the necessary repositories
|
||||
and managers required for the AuthService. It uses SQLite repositories
|
||||
for user and token data and initializes the required managers for password
|
||||
and token processing, as well as an optional email service.
|
||||
|
||||
:param db_path: Path to the SQLite database file used to persist authentication
|
||||
data.
|
||||
:type db_path: str
|
||||
:param jwt_secret: Secret key used to encode and decode JSON Web Tokens (JWTs).
|
||||
:type jwt_secret: str
|
||||
:param email_service: Optional email service instance for managing email-based
|
||||
communication.
|
||||
:type email_service: Optional[EmailService]
|
||||
:param password_manager: Optional custom PasswordManager instance responsible for
|
||||
password-related operations. Defaults to a new instance.
|
||||
:type password_manager: PasswordManager
|
||||
:param token_manager: Optional custom TokenManager instance responsible for token
|
||||
generation and verification. Defaults to a new instance
|
||||
configured with `jwt_secret`.
|
||||
:type token_manager: TokenManager
|
||||
:return: A fully configured instance of AuthService with the specified
|
||||
repositories and managers.
|
||||
:rtype: AuthService
|
||||
"""
|
||||
from .persistence.sqlite import SQLiteUserRepository
|
||||
from .persistence.sqlite import SQLiteTokenRepository
|
||||
user_repository = SQLiteUserRepository(db_path)
|
||||
token_repository = SQLiteTokenRepository(db_path)
|
||||
password_manager = password_manager or PasswordManager()
|
||||
token_manager = token_manager or TokenManager(jwt_secret)
|
||||
|
||||
auth_service = AuthService(
|
||||
user_repository=user_repository,
|
||||
token_repository=token_repository,
|
||||
password_manager=password_manager,
|
||||
token_manager=token_manager,
|
||||
email_service=email_service)
|
||||
|
||||
return auth_service
|
||||
|
||||
|
||||
def create_app_router_for_sqlite(db_path: str,
|
||||
jwt_secret: str,
|
||||
email_service: Optional[EmailService] = None):
|
||||
"""
|
||||
Creates an application router designed to use with an SQLite database backend.
|
||||
This function initializes an authentication service using the provided database
|
||||
path and JWT secret, optionally integrating an email service for email-based
|
||||
authentication functionalities. The authentication service is then used to
|
||||
create an application router for handling authentication-related API routes.
|
||||
|
||||
:param db_path: Path to the SQLite database file.
|
||||
:param jwt_secret: A secret string used for signing and verifying JWT tokens.
|
||||
:param email_service: An optional email service instance for managing email-related
|
||||
communication and functionalities during authentication.
|
||||
:return: An application router configured for handling authentication API routes.
|
||||
"""
|
||||
auth_service = create_sqlite_auth_service(db_path, jwt_secret, email_service)
|
||||
from .api.routes import create_auth_router
|
||||
return create_auth_router(auth_service)
|
||||
0
src/myauth/models/__init__.py
Normal file
0
src/myauth/models/__init__.py
Normal file
91
src/myauth/models/email_verification.py
Normal file
91
src/myauth/models/email_verification.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""
|
||||
Email verification and password reset models for authentication module.
|
||||
|
||||
This module defines Pydantic models for email verification and password
|
||||
reset operations including request and confirmation models.
|
||||
"""
|
||||
|
||||
from pydantic import BaseModel, EmailStr, field_validator
|
||||
|
||||
from ..models.validators import validate_password_strength
|
||||
|
||||
|
||||
class EmailVerificationRequest(BaseModel):
|
||||
"""
|
||||
Request model for email verification.
|
||||
|
||||
This model is used when a user requests an email verification link
|
||||
to be sent to their email address.
|
||||
|
||||
Attributes:
|
||||
email: The email address to send the verification link to.
|
||||
"""
|
||||
|
||||
email: EmailStr
|
||||
|
||||
|
||||
class EmailVerificationConfirm(BaseModel):
|
||||
"""
|
||||
Confirmation model for email verification.
|
||||
|
||||
This model is used when a user clicks the verification link and
|
||||
submits the token to confirm their email address.
|
||||
|
||||
Attributes:
|
||||
token: JWT token received in the verification email.
|
||||
"""
|
||||
|
||||
token: str
|
||||
|
||||
|
||||
class PasswordResetRequest(BaseModel):
|
||||
"""
|
||||
Request model for password reset.
|
||||
|
||||
This model is used when a user requests a password reset link
|
||||
to be sent to their email address.
|
||||
|
||||
Attributes:
|
||||
email: The email address to send the password reset link to.
|
||||
"""
|
||||
|
||||
email: EmailStr
|
||||
|
||||
|
||||
class PasswordResetConfirm(BaseModel):
|
||||
"""
|
||||
Confirmation model for password reset.
|
||||
|
||||
This model is used when a user submits a new password along with
|
||||
their reset token. The new password is validated with the same
|
||||
strict rules as user registration.
|
||||
|
||||
Attributes:
|
||||
token: Random secure token received in the password reset email.
|
||||
new_password: The new password that must meet security requirements:
|
||||
- Minimum 8 characters
|
||||
- At least 1 uppercase letter
|
||||
- At least 1 lowercase letter
|
||||
- At least 1 digit
|
||||
- At least 1 special character
|
||||
"""
|
||||
|
||||
token: str
|
||||
new_password: str
|
||||
|
||||
@field_validator('new_password')
|
||||
@classmethod
|
||||
def validate_password(cls, value: str) -> str:
|
||||
"""
|
||||
Validate new password meets security requirements.
|
||||
|
||||
Args:
|
||||
value: The new password to validate.
|
||||
|
||||
Returns:
|
||||
The validated password.
|
||||
|
||||
Raises:
|
||||
ValueError: If password does not meet security requirements.
|
||||
"""
|
||||
return validate_password_strength(value)
|
||||
88
src/myauth/models/token.py
Normal file
88
src/myauth/models/token.py
Normal file
@@ -0,0 +1,88 @@
|
||||
"""
|
||||
Token models for authentication module.
|
||||
|
||||
This module defines Pydantic models for token-related operations including
|
||||
JWT payloads, API responses, requests, and database storage.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class TokenPayload(BaseModel):
|
||||
"""
|
||||
JWT token payload structure.
|
||||
|
||||
This model represents the data contained within a JWT access token.
|
||||
It follows the standard JWT claims format.
|
||||
|
||||
Attributes:
|
||||
sub: Subject (user ID).
|
||||
email: User's email address.
|
||||
exp: Expiration timestamp (Unix timestamp).
|
||||
type: Token type identifier (always "access" for access tokens).
|
||||
"""
|
||||
|
||||
sub: str
|
||||
email: str
|
||||
exp: int
|
||||
type: Literal["access"] = "access"
|
||||
|
||||
|
||||
class AccessTokenResponse(BaseModel):
|
||||
"""
|
||||
Response model after successful authentication.
|
||||
|
||||
This model is returned after login or token refresh operations.
|
||||
It contains both access and refresh tokens.
|
||||
|
||||
Attributes:
|
||||
access_token: JWT access token for API authentication.
|
||||
refresh_token: Opaque refresh token for obtaining new access tokens.
|
||||
token_type: OAuth2 token type (always "bearer").
|
||||
"""
|
||||
|
||||
access_token: str
|
||||
refresh_token: str
|
||||
token_type: str = "bearer"
|
||||
|
||||
|
||||
class RefreshTokenRequest(BaseModel):
|
||||
"""
|
||||
Request model for token refresh operation.
|
||||
|
||||
This model is used when a client wants to obtain a new access token
|
||||
using their refresh token.
|
||||
|
||||
Attributes:
|
||||
refresh_token: The refresh token to exchange for a new access token.
|
||||
"""
|
||||
|
||||
refresh_token: str
|
||||
|
||||
|
||||
class TokenData(BaseModel):
|
||||
"""
|
||||
Token storage model for database.
|
||||
|
||||
This model represents tokens that need to be persisted in the database.
|
||||
It uses a discriminator field to distinguish between different token types
|
||||
(refresh tokens and password reset tokens) in a single collection/table.
|
||||
|
||||
Attributes:
|
||||
token: The token string (random secure string for refresh and reset).
|
||||
token_type: Discriminator field ("refresh" or "password_reset").
|
||||
user_id: ID of the user this token belongs to.
|
||||
expires_at: When the token expires.
|
||||
created_at: When the token was created.
|
||||
is_revoked: Whether the token has been revoked (for logout/security).
|
||||
"""
|
||||
|
||||
token: str
|
||||
token_type: Literal["refresh", "password_reset"]
|
||||
user_id: str
|
||||
expires_at: datetime
|
||||
created_at: datetime
|
||||
is_revoked: bool = False
|
||||
189
src/myauth/models/user.py
Normal file
189
src/myauth/models/user.py
Normal file
@@ -0,0 +1,189 @@
|
||||
"""
|
||||
User models for authentication module.
|
||||
|
||||
This module defines Pydantic models for user-related operations including
|
||||
creation, updates, database representation, and API responses.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel, EmailStr, Field, field_validator
|
||||
|
||||
from ..models.validators import validate_password_strength, validate_username_not_empty
|
||||
|
||||
|
||||
class UserBase(BaseModel):
|
||||
"""
|
||||
Base user model with common fields.
|
||||
|
||||
This model contains the shared fields used across different user representations.
|
||||
It serves as the foundation for other user models.
|
||||
|
||||
Attributes:
|
||||
email: User's email address (unique in database).
|
||||
username: User's display name (required, non-unique).
|
||||
roles: List of user roles (free-form strings, no defaults).
|
||||
user_settings: Dictionary for storing custom user settings.
|
||||
"""
|
||||
|
||||
email: EmailStr
|
||||
username: str
|
||||
roles: list[str] = Field(default_factory=list)
|
||||
user_settings: dict = Field(default_factory=dict)
|
||||
|
||||
|
||||
class UserCreate(UserBase):
|
||||
"""
|
||||
Model for user creation (registration).
|
||||
|
||||
This model extends UserBase with a password field and enforces strict
|
||||
password validation rules.
|
||||
|
||||
Attributes:
|
||||
password: Plain text password that must meet security requirements:
|
||||
- Minimum 8 characters
|
||||
- At least 1 uppercase letter
|
||||
- At least 1 lowercase letter
|
||||
- At least 1 digit
|
||||
- At least 1 special character
|
||||
"""
|
||||
|
||||
password: str
|
||||
|
||||
@field_validator('password')
|
||||
@classmethod
|
||||
def validate_password(cls, value: str) -> Optional[str]:
|
||||
"""
|
||||
Validate password meets security requirements.
|
||||
|
||||
Args:
|
||||
value: The password to validate.
|
||||
|
||||
Returns:
|
||||
The validated password.
|
||||
|
||||
Raises:
|
||||
ValueError: If password does not meet security requirements.
|
||||
"""
|
||||
return validate_password_strength(value)
|
||||
|
||||
@field_validator('username')
|
||||
@classmethod
|
||||
def validate_username(cls, value: str) -> str:
|
||||
"""
|
||||
Validate username is not empty and has reasonable length.
|
||||
|
||||
Args:
|
||||
value: The username to validate.
|
||||
|
||||
Returns:
|
||||
The validated username.
|
||||
|
||||
Raises:
|
||||
ValueError: If username is empty or too long.
|
||||
"""
|
||||
return validate_username_not_empty(value)
|
||||
|
||||
|
||||
class UserUpdate(BaseModel):
|
||||
"""
|
||||
Model for user updates.
|
||||
|
||||
All fields are optional to allow partial updates. Password updates
|
||||
are validated with the same strict rules as UserCreate.
|
||||
|
||||
Attributes:
|
||||
email: Optional new email address.
|
||||
username: Optional new username.
|
||||
password: Optional new password (will be hashed).
|
||||
roles: Optional new roles list.
|
||||
user_settings: Optional new settings dict.
|
||||
is_verified: Optional email verification status.
|
||||
is_active: Optional account active status.
|
||||
"""
|
||||
|
||||
email: Optional[EmailStr] = None
|
||||
username: Optional[str] = None
|
||||
password: Optional[str] = None
|
||||
roles: Optional[list[str]] = None
|
||||
user_settings: Optional[dict] = None
|
||||
is_verified: Optional[bool] = None
|
||||
is_active: Optional[bool] = None
|
||||
|
||||
@field_validator('password')
|
||||
@classmethod
|
||||
def validate_password_strength(cls, value: Optional[str]) -> Optional[str]:
|
||||
"""
|
||||
Validate password meets security requirements if provided.
|
||||
|
||||
Args:
|
||||
value: The password to validate (can be None).
|
||||
|
||||
Returns:
|
||||
The validated password or None.
|
||||
|
||||
Raises:
|
||||
ValueError: If password is provided but does not meet security requirements.
|
||||
"""
|
||||
return validate_password_strength(value)
|
||||
|
||||
@field_validator('username')
|
||||
@classmethod
|
||||
def validate_username(cls, value: Optional[str]) -> Optional[str]:
|
||||
"""
|
||||
Validate username if provided.
|
||||
|
||||
Args:
|
||||
value: The username to validate (can be None).
|
||||
|
||||
Returns:
|
||||
The validated username or None.
|
||||
|
||||
Raises:
|
||||
ValueError: If username is provided but empty or too long.
|
||||
"""
|
||||
return validate_username_not_empty(value)
|
||||
|
||||
|
||||
class UserInDB(UserBase):
|
||||
"""
|
||||
Complete user model as stored in database.
|
||||
|
||||
This model represents the full user entity including security-sensitive
|
||||
fields like hashed_password. It should not be directly exposed via API.
|
||||
|
||||
Attributes:
|
||||
id: Unique user identifier (string for database compatibility).
|
||||
hashed_password: Bcrypt hashed password.
|
||||
is_verified: Whether the user's email has been verified.
|
||||
is_active: Whether the user account is active.
|
||||
created_at: Timestamp of account creation.
|
||||
updated_at: Timestamp of last update.
|
||||
"""
|
||||
|
||||
id: str
|
||||
hashed_password: str
|
||||
is_verified: bool = False
|
||||
is_active: bool = True
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
|
||||
class UserResponse(UserBase):
|
||||
"""
|
||||
User model for API responses.
|
||||
|
||||
This is a safe representation of the user without sensitive fields
|
||||
like hashed_password. Used for API responses where user information
|
||||
needs to be exposed.
|
||||
|
||||
Attributes:
|
||||
id: Unique user identifier.
|
||||
created_at: Timestamp of account creation.
|
||||
updated_at: Timestamp of last update.
|
||||
"""
|
||||
|
||||
id: str
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
57
src/myauth/models/validators.py
Normal file
57
src/myauth/models/validators.py
Normal file
@@ -0,0 +1,57 @@
|
||||
from typing import Optional
|
||||
|
||||
def validate_password_strength(password: str) -> Optional[str]:
|
||||
"""
|
||||
Validate password meets security requirements.
|
||||
|
||||
Args:
|
||||
password: The password to validate.
|
||||
|
||||
Returns:
|
||||
The validated password.
|
||||
|
||||
Raises:
|
||||
ValueError: If password does not meet security requirements.
|
||||
"""
|
||||
if password is None:
|
||||
return password
|
||||
|
||||
if len(password) < 8:
|
||||
raise ValueError('Password must be at least 8 characters long')
|
||||
|
||||
if not any(char.isupper() for char in password):
|
||||
raise ValueError('Password must contain at least one uppercase letter')
|
||||
|
||||
if not any(char.islower() for char in password):
|
||||
raise ValueError('Password must contain at least one lowercase letter')
|
||||
|
||||
if not any(char.isdigit() for char in password):
|
||||
raise ValueError('Password must contain at least one digit')
|
||||
|
||||
special_characters = "!@#$%^&*()_+-=[]{}|;:,.<>?/~`"
|
||||
if not any(char in special_characters for char in password):
|
||||
raise ValueError('Password must contain at least one special character')
|
||||
|
||||
return password
|
||||
|
||||
|
||||
def validate_username_not_empty(user_name: str) -> str:
|
||||
"""
|
||||
Validate username is not empty and has reasonable length.
|
||||
|
||||
Args:
|
||||
user_name: The username to validate.
|
||||
|
||||
Returns:
|
||||
The validated username.
|
||||
|
||||
Raises:
|
||||
ValueError: If username is empty or too long.
|
||||
"""
|
||||
if not user_name or not user_name.strip():
|
||||
raise ValueError('Username cannot be empty')
|
||||
|
||||
if len(user_name) > 100:
|
||||
raise ValueError('Username cannot exceed 100 characters')
|
||||
|
||||
return user_name.strip()
|
||||
6
src/myauth/persistence/__init__.py
Normal file
6
src/myauth/persistence/__init__.py
Normal file
@@ -0,0 +1,6 @@
|
||||
from .sqlite import UserRepository, TokenRepository
|
||||
|
||||
__all__ = [
|
||||
"UserRepository",
|
||||
"TokenRepository",
|
||||
]
|
||||
226
src/myauth/persistence/base.py
Normal file
226
src/myauth/persistence/base.py
Normal file
@@ -0,0 +1,226 @@
|
||||
"""
|
||||
Abstract base classes for persistence layer.
|
||||
|
||||
This module defines the interfaces that all database implementations must
|
||||
follow. It provides abstract base classes for user and token repositories,
|
||||
ensuring consistency across different database backends (MongoDB, SQLite,
|
||||
PostgreSQL, custom engines, etc.).
|
||||
"""
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from ..models.token import TokenData
|
||||
from ..models.user import UserCreate, UserInDB, UserUpdate
|
||||
|
||||
|
||||
class UserRepository(ABC):
|
||||
"""
|
||||
Abstract base class for user persistence operations.
|
||||
|
||||
This interface defines all methods required for user management in the
|
||||
database. Concrete implementations must provide all these methods for
|
||||
their specific database backend.
|
||||
|
||||
All methods should raise appropriate exceptions (UserNotFoundError,
|
||||
UserAlreadyExistsError, etc.) when operations fail.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def create_user(self, user_data: UserCreate, hashed_password: str) -> UserInDB:
|
||||
"""
|
||||
Create a new user in the database.
|
||||
|
||||
Args:
|
||||
user_data: User information from registration.
|
||||
hashed_password: Pre-hashed password (hashing is done by the service layer).
|
||||
|
||||
Returns:
|
||||
The created user with database-assigned ID and timestamps.
|
||||
|
||||
Raises:
|
||||
UserAlreadyExistsError: If email already exists in database.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_user_by_email(self, email: str) -> UserInDB | None:
|
||||
"""
|
||||
Retrieve a user by their email address.
|
||||
|
||||
Args:
|
||||
email: The email address to search for.
|
||||
|
||||
Returns:
|
||||
The user if found, None otherwise.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_user_by_id(self, user_id: str) -> UserInDB | None:
|
||||
"""
|
||||
Retrieve a user by their unique identifier.
|
||||
|
||||
Args:
|
||||
user_id: The unique user identifier.
|
||||
|
||||
Returns:
|
||||
The user if found, None otherwise.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def update_user(self, user_id: str, updates: UserUpdate) -> UserInDB:
|
||||
"""
|
||||
Update an existing user's information.
|
||||
|
||||
Only fields present in the updates object (non-None) should be updated.
|
||||
The updated_at timestamp should be automatically set to the current time.
|
||||
|
||||
Args:
|
||||
user_id: The unique user identifier.
|
||||
updates: Pydantic model containing fields to update.
|
||||
|
||||
Returns:
|
||||
The updated user.
|
||||
|
||||
Raises:
|
||||
UserNotFoundError: If user does not exist.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete_user(self, user_id: str) -> bool:
|
||||
"""
|
||||
Delete a user from the database.
|
||||
|
||||
Implementation can choose between soft delete (setting is_active=False)
|
||||
or hard delete (removing from database). The choice depends on the
|
||||
application's requirements for data retention.
|
||||
|
||||
Args:
|
||||
user_id: The unique user identifier.
|
||||
|
||||
Returns:
|
||||
True if user was deleted, False if user was not found.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def email_exists(self, email: str) -> bool:
|
||||
"""
|
||||
Check if an email address is already registered.
|
||||
|
||||
This is useful for validating registration requests without
|
||||
retrieving the full user object.
|
||||
|
||||
Args:
|
||||
email: The email address to check.
|
||||
|
||||
Returns:
|
||||
True if email exists, False otherwise.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class TokenRepository(ABC):
|
||||
"""
|
||||
Abstract base class for token persistence operations.
|
||||
|
||||
This interface defines all methods required for token management in the
|
||||
database. It handles both refresh tokens and password reset tokens using
|
||||
a discriminator field (token_type).
|
||||
|
||||
All methods should raise appropriate exceptions (InvalidTokenError,
|
||||
ExpiredTokenError, etc.) when operations fail.
|
||||
"""
|
||||
|
||||
@abstractmethod
|
||||
def save_token(self, token_data: TokenData) -> None:
|
||||
"""
|
||||
Save a token to the database.
|
||||
|
||||
This is used for both refresh tokens and password reset tokens.
|
||||
The token_type field in TokenData distinguishes between them.
|
||||
|
||||
Args:
|
||||
token_data: Complete token information including type, user_id, expiration.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_token(self, token: str, token_type: str) -> TokenData | None:
|
||||
"""
|
||||
Retrieve a token from the database.
|
||||
|
||||
Args:
|
||||
token: The token string to search for.
|
||||
token_type: Type of token ("refresh" or "password_reset").
|
||||
|
||||
Returns:
|
||||
The token data if found, None otherwise.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def revoke_token(self, token: str) -> bool:
|
||||
"""
|
||||
Revoke a specific token.
|
||||
|
||||
This sets the is_revoked flag to True, preventing the token from
|
||||
being used again without physically deleting it (useful for audit trails).
|
||||
|
||||
Args:
|
||||
token: The token string to revoke.
|
||||
|
||||
Returns:
|
||||
True if token was revoked, False if token was not found.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def revoke_all_user_tokens(self, user_id: str, token_type: str) -> int:
|
||||
"""
|
||||
Revoke all tokens of a specific type for a user.
|
||||
|
||||
This is useful for logout-all-devices functionality or when a user's
|
||||
password is changed (invalidating all refresh tokens).
|
||||
|
||||
Args:
|
||||
user_id: The user whose tokens should be revoked.
|
||||
token_type: Type of tokens to revoke ("refresh" or "password_reset").
|
||||
|
||||
Returns:
|
||||
Number of tokens revoked.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete_expired_tokens(self) -> int:
|
||||
"""
|
||||
Delete all expired tokens from the database.
|
||||
|
||||
This maintenance operation should be called periodically to keep
|
||||
the tokens collection clean. It permanently removes tokens that
|
||||
have passed their expiration time.
|
||||
|
||||
Returns:
|
||||
Number of tokens deleted.
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def is_token_valid(self, token: str, token_type: str) -> bool:
|
||||
"""
|
||||
Check if a token is valid (exists, not revoked, not expired).
|
||||
|
||||
This is a convenience method that combines multiple checks into
|
||||
a single boolean result.
|
||||
|
||||
Args:
|
||||
token: The token string to validate.
|
||||
token_type: Type of token ("refresh" or "password_reset").
|
||||
|
||||
Returns:
|
||||
True if token is valid and can be used, False otherwise.
|
||||
"""
|
||||
pass
|
||||
611
src/myauth/persistence/sqlite.py
Normal file
611
src/myauth/persistence/sqlite.py
Normal file
@@ -0,0 +1,611 @@
|
||||
"""
|
||||
SQLite implementation of persistence layer.
|
||||
|
||||
This module provides SQLite database implementations for user and token
|
||||
repositories. It uses the standard library sqlite3 module and handles
|
||||
JSON serialization for complex fields.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
from uuid import uuid4
|
||||
|
||||
from .base import UserRepository, TokenRepository
|
||||
from ..models.user import UserCreate, UserInDB, UserUpdate
|
||||
from ..models.token import TokenData
|
||||
from ..exceptions import UserAlreadyExistsError, UserNotFoundError
|
||||
|
||||
|
||||
class SQLiteUserRepository(UserRepository):
|
||||
"""
|
||||
SQLite implementation of UserRepository.
|
||||
|
||||
This implementation uses sqlite3 to manage user data. JSON fields
|
||||
(roles, user_settings) are serialized/deserialized automatically.
|
||||
The database schema is created automatically on initialization.
|
||||
|
||||
Attributes:
|
||||
db_path: Path to the SQLite database file.
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: str):
|
||||
"""
|
||||
Initialize SQLite user repository.
|
||||
|
||||
Creates the users table and indexes if they don't exist.
|
||||
|
||||
Args:
|
||||
db_path: Path to the SQLite database file.
|
||||
"""
|
||||
self.db_path = db_path
|
||||
self._create_tables()
|
||||
|
||||
def _create_tables(self) -> None:
|
||||
"""
|
||||
Create users table and indexes if they don't exist.
|
||||
|
||||
The table schema includes:
|
||||
- id: Primary key (UUID string)
|
||||
- email: Unique, indexed for fast lookups
|
||||
- username: Non-unique display name
|
||||
- hashed_password: Bcrypt hash
|
||||
- roles: JSON array of role strings
|
||||
- user_settings: JSON object for custom settings
|
||||
- is_verified: Boolean (stored as INTEGER)
|
||||
- is_active: Boolean (stored as INTEGER)
|
||||
- created_at: ISO format timestamp
|
||||
- updated_at: ISO format timestamp
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Create users table
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS users
|
||||
(
|
||||
id
|
||||
TEXT
|
||||
PRIMARY
|
||||
KEY,
|
||||
email
|
||||
TEXT
|
||||
UNIQUE
|
||||
NOT
|
||||
NULL,
|
||||
username
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
hashed_password
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
roles
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
user_settings
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
is_verified
|
||||
INTEGER
|
||||
NOT
|
||||
NULL
|
||||
DEFAULT
|
||||
0,
|
||||
is_active
|
||||
INTEGER
|
||||
NOT
|
||||
NULL
|
||||
DEFAULT
|
||||
1,
|
||||
created_at
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
updated_at
|
||||
TEXT
|
||||
NOT
|
||||
NULL
|
||||
)
|
||||
""")
|
||||
|
||||
# Create index on email for fast lookups
|
||||
cursor.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_users_email
|
||||
ON users(email)
|
||||
""")
|
||||
|
||||
conn.commit()
|
||||
|
||||
def _row_to_user(self, row: tuple) -> UserInDB:
|
||||
"""
|
||||
Convert a database row to a UserInDB model.
|
||||
|
||||
Args:
|
||||
row: Database row tuple.
|
||||
|
||||
Returns:
|
||||
UserInDB model instance.
|
||||
"""
|
||||
return UserInDB(
|
||||
id=row[0],
|
||||
email=row[1],
|
||||
username=row[2],
|
||||
hashed_password=row[3],
|
||||
roles=json.loads(row[4]),
|
||||
user_settings=json.loads(row[5]),
|
||||
is_verified=bool(row[6]),
|
||||
is_active=bool(row[7]),
|
||||
created_at=datetime.fromisoformat(row[8]),
|
||||
updated_at=datetime.fromisoformat(row[9])
|
||||
)
|
||||
|
||||
def create_user(self, user_data: UserCreate, hashed_password: str) -> UserInDB:
|
||||
"""
|
||||
Create a new user in the database.
|
||||
|
||||
Args:
|
||||
user_data: User information from registration.
|
||||
hashed_password: Pre-hashed password.
|
||||
|
||||
Returns:
|
||||
The created user with generated ID and timestamps.
|
||||
|
||||
Raises:
|
||||
UserAlreadyExistsError: If email already exists.
|
||||
"""
|
||||
if self.email_exists(user_data.email):
|
||||
raise UserAlreadyExistsError(f"User with email {user_data.email} already exists")
|
||||
|
||||
user_id = str(uuid4())
|
||||
now = datetime.now()
|
||||
|
||||
user = UserInDB(
|
||||
id=user_id,
|
||||
email=user_data.email,
|
||||
username=user_data.username,
|
||||
hashed_password=hashed_password,
|
||||
roles=user_data.roles,
|
||||
user_settings=user_data.user_settings,
|
||||
is_verified=False,
|
||||
is_active=True,
|
||||
created_at=now,
|
||||
updated_at=now
|
||||
)
|
||||
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO users (id, email, username, hashed_password, roles,
|
||||
user_settings, is_verified, is_active, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""", (
|
||||
user.id,
|
||||
user.email,
|
||||
user.username,
|
||||
user.hashed_password,
|
||||
json.dumps(user.roles),
|
||||
json.dumps(user.user_settings),
|
||||
int(user.is_verified),
|
||||
int(user.is_active),
|
||||
user.created_at.isoformat(),
|
||||
user.updated_at.isoformat()
|
||||
))
|
||||
conn.commit()
|
||||
|
||||
return user
|
||||
|
||||
def get_user_by_email(self, email: str) -> Optional[UserInDB]:
|
||||
"""
|
||||
Retrieve a user by their email address.
|
||||
|
||||
Args:
|
||||
email: The email address to search for.
|
||||
|
||||
Returns:
|
||||
The user if found, None otherwise.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT id,
|
||||
email,
|
||||
username,
|
||||
hashed_password,
|
||||
roles,
|
||||
user_settings,
|
||||
is_verified,
|
||||
is_active,
|
||||
created_at,
|
||||
updated_at
|
||||
FROM users
|
||||
WHERE email = ?
|
||||
""", (email,))
|
||||
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return self._row_to_user(row)
|
||||
return None
|
||||
|
||||
def get_user_by_id(self, user_id: str) -> Optional[UserInDB]:
|
||||
"""
|
||||
Retrieve a user by their unique identifier.
|
||||
|
||||
Args:
|
||||
user_id: The unique user identifier.
|
||||
|
||||
Returns:
|
||||
The user if found, None otherwise.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT id,
|
||||
email,
|
||||
username,
|
||||
hashed_password,
|
||||
roles,
|
||||
user_settings,
|
||||
is_verified,
|
||||
is_active,
|
||||
created_at,
|
||||
updated_at
|
||||
FROM users
|
||||
WHERE id = ?
|
||||
""", (user_id,))
|
||||
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return self._row_to_user(row)
|
||||
return None
|
||||
|
||||
def update_user(self, user_id: str, updates: UserUpdate) -> UserInDB:
|
||||
"""
|
||||
Update an existing user's information.
|
||||
|
||||
Only non-None fields in the updates object are applied.
|
||||
The updated_at timestamp is automatically set to current time.
|
||||
|
||||
Args:
|
||||
user_id: The unique user identifier.
|
||||
updates: Pydantic model containing fields to update.
|
||||
|
||||
Returns:
|
||||
The updated user.
|
||||
|
||||
Raises:
|
||||
UserNotFoundError: If user does not exist.
|
||||
"""
|
||||
user = self.get_user_by_id(user_id)
|
||||
if not user:
|
||||
raise UserNotFoundError(f"User with id {user_id} not found")
|
||||
|
||||
# Build update query dynamically based on non-None fields
|
||||
update_fields = []
|
||||
update_values = []
|
||||
|
||||
update_data = updates.model_dump(exclude_unset=True)
|
||||
|
||||
for field, value in update_data.items():
|
||||
if value is not None:
|
||||
if field == "roles":
|
||||
update_fields.append("roles = ?")
|
||||
update_values.append(json.dumps(value))
|
||||
elif field == "user_settings":
|
||||
update_fields.append("user_settings = ?")
|
||||
update_values.append(json.dumps(value))
|
||||
elif field == "password":
|
||||
# Password should be hashed before calling this method
|
||||
# This field should contain hashed_password value
|
||||
update_fields.append("hashed_password = ?")
|
||||
update_values.append(value)
|
||||
elif field in ["is_verified", "is_active"]:
|
||||
update_fields.append(f"{field} = ?")
|
||||
update_values.append(int(value))
|
||||
else:
|
||||
update_fields.append(f"{field} = ?")
|
||||
update_values.append(value)
|
||||
|
||||
if not update_fields:
|
||||
return user
|
||||
|
||||
# Always update the updated_at timestamp
|
||||
update_fields.append("updated_at = ?")
|
||||
update_values.append(datetime.now().isoformat())
|
||||
|
||||
# Add user_id for WHERE clause
|
||||
update_values.append(user_id)
|
||||
|
||||
query = f"UPDATE users SET {', '.join(update_fields)} WHERE id = ?"
|
||||
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(query, update_values)
|
||||
conn.commit()
|
||||
|
||||
# Return updated user
|
||||
updated_user = self.get_user_by_id(user_id)
|
||||
if not updated_user:
|
||||
raise UserNotFoundError(f"User with id {user_id} not found after update")
|
||||
|
||||
return updated_user
|
||||
|
||||
def delete_user(self, user_id: str) -> bool:
|
||||
"""
|
||||
Delete a user from the database (hard delete).
|
||||
|
||||
Args:
|
||||
user_id: The unique user identifier.
|
||||
|
||||
Returns:
|
||||
True if user was deleted, False if user was not found.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
||||
conn.commit()
|
||||
return cursor.rowcount > 0
|
||||
|
||||
def email_exists(self, email: str) -> bool:
|
||||
"""
|
||||
Check if an email address is already registered.
|
||||
|
||||
Args:
|
||||
email: The email address to check.
|
||||
|
||||
Returns:
|
||||
True if email exists, False otherwise.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT 1 FROM users WHERE email = ? LIMIT 1", (email,))
|
||||
return cursor.fetchone() is not None
|
||||
|
||||
|
||||
class SQLiteTokenRepository(TokenRepository):
|
||||
"""
|
||||
SQLite implementation of TokenRepository.
|
||||
|
||||
This implementation manages both refresh tokens and password reset tokens
|
||||
in a single table with a discriminator field (token_type).
|
||||
|
||||
Attributes:
|
||||
db_path: Path to the SQLite database file.
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: str):
|
||||
"""
|
||||
Initialize SQLite token repository.
|
||||
|
||||
Creates the tokens table and indexes if they don't exist.
|
||||
|
||||
Args:
|
||||
db_path: Path to the SQLite database file.
|
||||
"""
|
||||
self.db_path = db_path
|
||||
self._create_tables()
|
||||
|
||||
def _create_tables(self) -> None:
|
||||
"""
|
||||
Create tokens table and indexes if they don't exist.
|
||||
|
||||
The table schema includes:
|
||||
- token: Primary key (random secure string)
|
||||
- token_type: Discriminator ("refresh" or "password_reset")
|
||||
- user_id: Foreign key to users
|
||||
- expires_at: ISO format timestamp
|
||||
- created_at: ISO format timestamp
|
||||
- is_revoked: Boolean (stored as INTEGER)
|
||||
|
||||
Indexes are created for efficient queries.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Create tokens table
|
||||
cursor.execute("""
|
||||
CREATE TABLE IF NOT EXISTS tokens
|
||||
(
|
||||
token
|
||||
TEXT
|
||||
PRIMARY
|
||||
KEY,
|
||||
token_type
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
user_id
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
expires_at
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
created_at
|
||||
TEXT
|
||||
NOT
|
||||
NULL,
|
||||
is_revoked
|
||||
INTEGER
|
||||
NOT
|
||||
NULL
|
||||
DEFAULT
|
||||
0
|
||||
)
|
||||
""")
|
||||
|
||||
# Create index on user_id for revoking all user tokens
|
||||
cursor.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_tokens_user_id
|
||||
ON tokens(user_id)
|
||||
""")
|
||||
|
||||
# Create composite index on token_type and user_id
|
||||
cursor.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_tokens_type_user
|
||||
ON tokens(token_type, user_id)
|
||||
""")
|
||||
|
||||
# Create index on expires_at for cleanup operations
|
||||
cursor.execute("""
|
||||
CREATE INDEX IF NOT EXISTS idx_tokens_expires_at
|
||||
ON tokens(expires_at)
|
||||
""")
|
||||
|
||||
conn.commit()
|
||||
|
||||
def _row_to_token(self, row: tuple) -> TokenData:
|
||||
"""
|
||||
Convert a database row to a TokenData model.
|
||||
|
||||
Args:
|
||||
row: Database row tuple.
|
||||
|
||||
Returns:
|
||||
TokenData model instance.
|
||||
"""
|
||||
return TokenData(
|
||||
token=row[0],
|
||||
token_type=row[1],
|
||||
user_id=row[2],
|
||||
expires_at=datetime.fromisoformat(row[3]),
|
||||
created_at=datetime.fromisoformat(row[4]),
|
||||
is_revoked=bool(row[5])
|
||||
)
|
||||
|
||||
def save_token(self, token_data: TokenData) -> None:
|
||||
"""
|
||||
Save a token to the database.
|
||||
|
||||
Args:
|
||||
token_data: Complete token information.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
INSERT INTO tokens (token, token_type, user_id, expires_at, created_at, is_revoked)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""", (
|
||||
token_data.token,
|
||||
token_data.token_type,
|
||||
token_data.user_id,
|
||||
token_data.expires_at.isoformat(),
|
||||
token_data.created_at.isoformat(),
|
||||
int(token_data.is_revoked)
|
||||
))
|
||||
conn.commit()
|
||||
|
||||
def get_token(self, token: str, token_type: str) -> Optional[TokenData]:
|
||||
"""
|
||||
Retrieve a token from the database.
|
||||
|
||||
Args:
|
||||
token: The token string to search for.
|
||||
token_type: Type of token ("refresh" or "password_reset").
|
||||
|
||||
Returns:
|
||||
The token data if found, None otherwise.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT token, token_type, user_id, expires_at, created_at, is_revoked
|
||||
FROM tokens
|
||||
WHERE token = ?
|
||||
AND token_type = ?
|
||||
""", (token, token_type))
|
||||
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return self._row_to_token(row)
|
||||
return None
|
||||
|
||||
def revoke_token(self, token: str) -> bool:
|
||||
"""
|
||||
Revoke a specific token.
|
||||
|
||||
Args:
|
||||
token: The token string to revoke.
|
||||
|
||||
Returns:
|
||||
True if token was revoked, False if token was not found.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
UPDATE tokens
|
||||
SET is_revoked = 1
|
||||
WHERE token = ?
|
||||
""", (token,))
|
||||
conn.commit()
|
||||
return cursor.rowcount > 0
|
||||
|
||||
def revoke_all_user_tokens(self, user_id: str, token_type: str) -> int:
|
||||
"""
|
||||
Revoke all tokens of a specific type for a user.
|
||||
|
||||
Args:
|
||||
user_id: The user whose tokens should be revoked.
|
||||
token_type: Type of tokens to revoke.
|
||||
|
||||
Returns:
|
||||
Number of tokens revoked.
|
||||
"""
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
UPDATE tokens
|
||||
SET is_revoked = 1
|
||||
WHERE user_id = ?
|
||||
AND token_type = ?
|
||||
""", (user_id, token_type))
|
||||
conn.commit()
|
||||
return cursor.rowcount
|
||||
|
||||
def delete_expired_tokens(self) -> int:
|
||||
"""
|
||||
Delete all expired tokens from the database.
|
||||
|
||||
Returns:
|
||||
Number of tokens deleted.
|
||||
"""
|
||||
now = datetime.now().isoformat()
|
||||
|
||||
with sqlite3.connect(self.db_path) as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
DELETE
|
||||
FROM tokens
|
||||
WHERE expires_at < ?
|
||||
""", (now,))
|
||||
conn.commit()
|
||||
return cursor.rowcount
|
||||
|
||||
def is_token_valid(self, token: str, token_type: str) -> bool:
|
||||
"""
|
||||
Check if a token is valid (exists, not revoked, not expired).
|
||||
|
||||
Args:
|
||||
token: The token string to validate.
|
||||
token_type: Type of token.
|
||||
|
||||
Returns:
|
||||
True if token is valid, False otherwise.
|
||||
"""
|
||||
token_data = self.get_token(token, token_type)
|
||||
|
||||
if not token_data:
|
||||
return False
|
||||
|
||||
if token_data.is_revoked:
|
||||
return False
|
||||
|
||||
if token_data.expires_at < datetime.now():
|
||||
return False
|
||||
|
||||
return True
|
||||
Reference in New Issue
Block a user