added burger menu with admin page and user invitation link generation

2025-11-07 22:33:57 +01:00
parent b8e04fe816
commit d31ebc8c22
17 changed files with 1326 additions and 25 deletions
@@ -0,0 +1,66 @@
+"""Admin router for administrative functions."""
+
+import secrets
+from datetime import datetime, timedelta, UTC
+from typing import Annotated
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+
+from baby_monitor.routers.auth import verify_admin
+from baby_monitor.repositories.dependencies.get_invitation_repository import (
+    get_invitation_repository,
+)
+from baby_monitor.repositories.interfaces import (
+    InvitationRepositoryInterface,
+)
+
+router = APIRouter(prefix="/api/admin", tags=["admin"])
+
+
+class InvitationResponse(BaseModel):
+    """Response model for invitation link generation."""
+
+    token: str
+    expires_at: str
+    message: str
+
+
+@router.post("/generate-invitation", response_model=InvitationResponse)
+async def generate_invitation_link(
+    user_id: Annotated[int, Depends(verify_admin)],
+    invitation_repository: Annotated[
+        InvitationRepositoryInterface,
+        Depends(get_invitation_repository),
+    ],
+) -> InvitationResponse:
+    """Generate a new invitation link for user registration.
+    
+    The invitation token is valid for 24 hours and can be used once.
+    Requires admin role.
+    
+    Args:
+        user_id: Admin user ID (from verify_admin)
+        invitation_repository: Invitation storage backend
+    
+    Returns:
+        InvitationResponse with token and expiration details
+    """
+    
+    # Generate secure random token
+    invitation_token = secrets.token_urlsafe(32)
+    
+    # Calculate expiration (24 hours from now)
+    expires_at = datetime.now(UTC) + timedelta(hours=24)
+    
+    # Store invitation token in database
+    invitation_repository.create_invitation(
+        token=invitation_token,
+        created_by_user_id=user_id,
+        expires_at=expires_at,
+    )
+    
+    return InvitationResponse(
+        token=invitation_token,
+        expires_at=expires_at.isoformat(),
+        message="Invitation link generated successfully. Valid for 24 hours.",
+    )
@@ -4,14 +4,25 @@ import secrets
 from fastapi import APIRouter, HTTPException, Depends, Header
 from typing import Annotated

-from baby_monitor.models.auth import LoginRequest, LoginResponse
+from baby_monitor.models.auth import (
+    LoginRequest,
+    LoginResponse,
+    RegisterRequest,
+    RegisterResponse,
+)
 from baby_monitor.repositories import (
    get_token_repository,
    get_credentials_repository,
+    get_user_repository,
+)
+from baby_monitor.repositories.dependencies.get_invitation_repository import (
+    get_invitation_repository,
 )
 from baby_monitor.repositories.interfaces import (
    TokenRepositoryInterface,
    CredentialsRepositoryInterface,
+    UserRepositoryInterface,
+    InvitationRepositoryInterface,
 )

 router = APIRouter(prefix="/api", tags=["authentication"])
@@ -39,11 +50,36 @@ def verify_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: TokenRepositoryInterface = Depends(get_token_repository),
-    creds_repo: CredentialsRepositoryInterface = Depends(get_credentials_repository),
+    token_repo: Annotated[
+        TokenRepositoryInterface, Depends(get_token_repository)
+    ],
+    user_repo: Annotated[
+        UserRepositoryInterface, Depends(get_user_repository)
+    ],
+    creds_repo: Annotated[
+        CredentialsRepositoryInterface, Depends(get_credentials_repository)
+    ],
 ) -> LoginResponse:
    """
    API endpoint for user authentication.
@@ -51,11 +87,29 @@ def login(
    Returns user info and access token on successful login.
    Raises 401 on invalid credentials.
    """
-    # Verify credentials using repository
-    if creds_repo.verify_admin_credentials(credentials.username, credentials.password):
+    # First check if it's a database user
+    user = user_repo.get_by_username(credentials.username)
+    if user:
+        # TODO: Use proper password hashing (bcrypt/argon2)
+        # For now, compare plain text (matches registration)
+        if user["hashed_password"] == credentials.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,
+            )
+    
+    # Fallback to admin credentials from environment
+    if creds_repo.verify_admin_credentials(
+        credentials.username, credentials.password
+    ):
        # Generate a secure random token
        access_token = secrets.token_urlsafe(32)
-        # Store token with user_id (hardcoded 1 for now)
+        # Store token with user_id (hardcoded 1 for admin)
        token_repo.store(access_token, user_id=1, ttl=3600)

        return LoginResponse(
@@ -63,8 +117,10 @@ def login(
            username=credentials.username,
            access_token=access_token,
        )
-    else:
-        raise HTTPException(status_code=401, detail="Invalid username or password")
+    
+    raise HTTPException(
+        status_code=401, detail="Invalid username or password"
+    )


@router.post("/logout")
@@ -76,3 +132,92 @@ def logout(
    # 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
+    # TODO: Hash password before storing (currently plain text)
+    user = user_repo.create(
+        username=request.username,
+        hashed_password=request.password,
+    )
+    
+    # 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,
+    )
+
+
+@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),
+    }