302 lines
9.7 KiB
Python
302 lines
9.7 KiB
Python
"""Authentication router for login and user management."""
|
|
|
|
import secrets
|
|
from typing import Annotated
|
|
from fastapi import APIRouter, HTTPException, Depends, Header
|
|
from pydantic import BaseModel
|
|
|
|
from baby_monitor.models.auth import (
|
|
LoginRequest,
|
|
LoginResponse,
|
|
RegisterRequest,
|
|
RegisterResponse,
|
|
)
|
|
from baby_monitor.repositories import (
|
|
get_child_repository,
|
|
get_diaper_change_repository,
|
|
get_feeding_repository,
|
|
get_invitation_repository,
|
|
get_sleep_repository,
|
|
get_token_repository,
|
|
get_user_repository,
|
|
)
|
|
from baby_monitor.repositories.interfaces import (
|
|
TokenRepositoryInterface,
|
|
UserRepositoryInterface,
|
|
InvitationRepositoryInterface,
|
|
ChildRepositoryInterface,
|
|
FeedingRepositoryInterface,
|
|
DiaperChangeRepositoryInterface,
|
|
SleepRepositoryInterface,
|
|
)
|
|
from baby_monitor.utils import hash_password, verify_password
|
|
|
|
router = APIRouter(prefix="/api", tags=["authentication"])
|
|
|
|
|
|
class ChangePasswordRequest(BaseModel):
|
|
"""Request model for password change."""
|
|
|
|
current_password: str
|
|
new_password: str
|
|
|
|
|
|
def verify_token(
|
|
authorization: Annotated[str | None, Header()] = None,
|
|
token_repo: TokenRepositoryInterface = Depends(get_token_repository),
|
|
) -> int:
|
|
"""Verify the bearer token and return user_id."""
|
|
if not authorization:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
|
|
try:
|
|
scheme, token = authorization.split()
|
|
if scheme.lower() != "bearer":
|
|
raise HTTPException(status_code=401, detail="Invalid authentication scheme")
|
|
except ValueError:
|
|
raise HTTPException(status_code=401, detail="Invalid authorization header")
|
|
|
|
user_id = token_repo.verify(token)
|
|
if user_id is None:
|
|
raise HTTPException(status_code=401, detail="Invalid or expired token")
|
|
|
|
return user_id
|
|
|
|
|
|
def verify_admin(
|
|
user_id: Annotated[int, Depends(verify_token)],
|
|
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
|
|
) -> int:
|
|
"""Verify the user is an admin and return user_id."""
|
|
user = user_repo.get_by_id(user_id)
|
|
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="User not found")
|
|
|
|
if not user.get("is_admin", False):
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
|
|
return user_id
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(
|
|
credentials: LoginRequest,
|
|
token_repo: Annotated[TokenRepositoryInterface, Depends(get_token_repository)],
|
|
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
|
|
) -> LoginResponse:
|
|
"""
|
|
API endpoint for user authentication.
|
|
|
|
Returns user info and access token on successful login.
|
|
Raises 401 on invalid credentials.
|
|
"""
|
|
# Check if user exists in database
|
|
user = user_repo.get_by_username(credentials.username)
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Invalid username or password")
|
|
|
|
# Verify password using bcrypt
|
|
if not verify_password(credentials.password, user["hashed_password"]):
|
|
raise HTTPException(status_code=401, detail="Invalid username or password")
|
|
|
|
# Generate a secure random token
|
|
access_token = secrets.token_urlsafe(32)
|
|
token_repo.store(access_token, user_id=user["id"], ttl=3600)
|
|
|
|
return LoginResponse(
|
|
message="Login successful",
|
|
username=credentials.username,
|
|
access_token=access_token,
|
|
is_admin=user.get("is_admin", False),
|
|
)
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(
|
|
user_id: Annotated[int, Depends(verify_token)],
|
|
token_repo: TokenRepositoryInterface = Depends(get_token_repository),
|
|
) -> dict[str, str]:
|
|
"""Logout and invalidate the current token."""
|
|
# Note: We'd need to pass the token itself, not user_id
|
|
# This is simplified - in production, extract token from verify_token
|
|
return {"message": "Logged out successfully"}
|
|
|
|
|
|
@router.get("/verify-invitation")
|
|
def verify_invitation(
|
|
token: str,
|
|
invitation_repo: Annotated[
|
|
InvitationRepositoryInterface, Depends(get_invitation_repository)
|
|
],
|
|
) -> dict[str, bool]:
|
|
"""Verify if an invitation token is valid."""
|
|
is_valid = invitation_repo.verify_invitation(token)
|
|
if not is_valid:
|
|
raise HTTPException(status_code=400, detail="Invalid or expired invitation")
|
|
return {"valid": True}
|
|
|
|
|
|
@router.post("/register", response_model=RegisterResponse)
|
|
def register(
|
|
request: RegisterRequest,
|
|
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
|
|
invitation_repo: Annotated[
|
|
InvitationRepositoryInterface, Depends(get_invitation_repository)
|
|
],
|
|
token_repo: Annotated[TokenRepositoryInterface, Depends(get_token_repository)],
|
|
) -> RegisterResponse:
|
|
"""
|
|
Register a new user with an invitation token.
|
|
|
|
Verifies the invitation, creates the user, consumes the invitation,
|
|
and returns an authentication token.
|
|
"""
|
|
# Verify invitation token
|
|
if not invitation_repo.verify_invitation(request.invitation_token):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Invalid or expired invitation token",
|
|
)
|
|
|
|
# Check if username already exists
|
|
existing_user = user_repo.get_by_username(request.username)
|
|
if existing_user:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Username already exists",
|
|
)
|
|
|
|
# Create the new user with hashed password
|
|
hashed_pw = hash_password(request.password)
|
|
user = user_repo.create(
|
|
username=request.username,
|
|
hashed_password=hashed_pw,
|
|
)
|
|
|
|
# Consume the invitation token
|
|
invitation_repo.consume_invitation(request.invitation_token)
|
|
|
|
# Generate authentication token
|
|
access_token = secrets.token_urlsafe(32)
|
|
token_repo.store(access_token, user_id=user["id"], ttl=3600)
|
|
|
|
return RegisterResponse(
|
|
message="Registration successful",
|
|
username=user["username"],
|
|
access_token=access_token,
|
|
is_admin=user.get("is_admin", False),
|
|
)
|
|
|
|
|
|
@router.get("/me")
|
|
def get_current_user(
|
|
user_id: Annotated[int, Depends(verify_token)],
|
|
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)],
|
|
) -> dict:
|
|
"""Get current user info including admin status."""
|
|
user = user_repo.get_by_id(user_id)
|
|
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
return {
|
|
"id": user["id"],
|
|
"username": user["username"],
|
|
"is_admin": user.get("is_admin", False),
|
|
}
|
|
|
|
|
|
@router.post("/change-password")
|
|
def change_password(
|
|
request: ChangePasswordRequest,
|
|
user_id: Annotated[int, Depends(verify_token)],
|
|
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)], # noqa: E501
|
|
) -> dict:
|
|
"""Change password for the current user."""
|
|
# Get user with credentials
|
|
user = user_repo.get_by_id(user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# Verify current password
|
|
if not verify_password(request.current_password, user["hashed_password"]):
|
|
raise HTTPException(status_code=401, detail="Current password is incorrect")
|
|
|
|
# Hash new password
|
|
new_password_hash = hash_password(request.new_password)
|
|
|
|
# Update password
|
|
success = user_repo.update_password(user_id, new_password_hash)
|
|
if not success:
|
|
raise HTTPException(status_code=500, detail="Failed to update password")
|
|
|
|
return {"message": "Password changed successfully"}
|
|
|
|
|
|
@router.delete("/account")
|
|
def delete_account(
|
|
user_id: Annotated[int, Depends(verify_token)],
|
|
user_repo: Annotated[UserRepositoryInterface, Depends(get_user_repository)], # noqa: E501
|
|
child_repo: Annotated[ChildRepositoryInterface, Depends(get_child_repository)], # noqa: E501
|
|
feeding_repo: Annotated[
|
|
FeedingRepositoryInterface, Depends(get_feeding_repository)
|
|
], # noqa: E501
|
|
diaper_repo: Annotated[
|
|
DiaperChangeRepositoryInterface, Depends(get_diaper_change_repository)
|
|
], # noqa: E501
|
|
sleep_repo: Annotated[SleepRepositoryInterface, Depends(get_sleep_repository)], # noqa: E501
|
|
) -> dict:
|
|
"""Delete current user's account and cascade delete orphaned children."""
|
|
# Get user's children
|
|
children = child_repo.get_by_user_id(user_id)
|
|
|
|
deleted_children = []
|
|
deleted_feedings = 0
|
|
deleted_diapers = 0
|
|
deleted_sleeps = 0
|
|
|
|
# Process each child
|
|
for child in children:
|
|
child_id = child["id"]
|
|
|
|
# Remove the parent-child relationship
|
|
child_repo.remove_parent(child_id, user_id)
|
|
|
|
# Check if child has any remaining parents
|
|
remaining_parents = child_repo.get_parent_ids(child_id)
|
|
|
|
if len(remaining_parents) == 0:
|
|
# Child is orphaned, delete all logs first
|
|
feedings = feeding_repo.get_by_child_id(child_id)
|
|
for feeding in feedings:
|
|
feeding_repo.delete(feeding["id"])
|
|
deleted_feedings += 1
|
|
|
|
diapers = diaper_repo.get_by_child_id(child_id)
|
|
for diaper in diapers:
|
|
diaper_repo.delete(diaper["id"])
|
|
deleted_diapers += 1
|
|
|
|
sleeps = sleep_repo.get_by_child_id(child_id)
|
|
for sleep in sleeps:
|
|
sleep_repo.delete(sleep["id"])
|
|
deleted_sleeps += 1
|
|
|
|
# Now delete the child
|
|
child_repo.delete(child_id)
|
|
deleted_children.append(child["name"])
|
|
|
|
# Finally, delete the user
|
|
user_repo.delete(user_id)
|
|
|
|
return {
|
|
"message": "Account deleted successfully",
|
|
"deleted_children": deleted_children,
|
|
"deleted_logs": {
|
|
"feedings": deleted_feedings,
|
|
"diaper_changes": deleted_diapers,
|
|
"sleep_sessions": deleted_sleeps,
|
|
},
|
|
}
|