RBAC and Doc updates

api/main.py

@@ -2,26 +2,40 @@
 Mailing List Management API
 FastAPI-based REST API for managing mailing lists and members
 """
-from fastapi import FastAPI, HTTPException, Depends, Header
+from fastapi import FastAPI, HTTPException, Depends, Header, status, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from pydantic import BaseModel, EmailStr
-from typing import List, Optional
+from typing import List, Optional, Annotated
 import mysql.connector
 from mysql.connector import Error
 import os
 import csv
 import io
 from contextlib import contextmanager
+from datetime import datetime, timedelta
+import secrets
+import bcrypt
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from enum import Enum
 
 # Configuration
-API_TOKEN = os.getenv('API_TOKEN', 'change-this-token')
+API_TOKEN = os.getenv('API_TOKEN', 'change-this-token')  # Keep for backward compatibility during transition
+JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'your-secret-key-change-this-in-production')
+JWT_ALGORITHM = "HS256"
+JWT_ACCESS_TOKEN_EXPIRE_MINUTES = 30
+SESSION_EXPIRE_HOURS = 24
+
 MYSQL_HOST = os.getenv('MYSQL_HOST', 'mysql')
 MYSQL_PORT = int(os.getenv('MYSQL_PORT', 3306))
 MYSQL_DATABASE = os.getenv('MYSQL_DATABASE', 'maillist')
 MYSQL_USER = os.getenv('MYSQL_USER', 'maillist')
 MYSQL_PASSWORD = os.getenv('MYSQL_PASSWORD', '')
 
+# Password hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
 # FastAPI app
 app = FastAPI(
     title="Mailing List Manager API",
@@ -60,14 +74,144 @@ def get_db():
         if connection and connection.is_connected():
             connection.close()
 
-# Authentication
+# Role-based access control
+class UserRole(str, Enum):
+    ADMINISTRATOR = "administrator"
+    OPERATOR = "operator"
+    READ_ONLY = "read-only"
+
+class CurrentUser(BaseModel):
+    user_id: int
+    username: str
+    role: UserRole
+    active: bool
+
+# Authentication and authorization
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verify a password against its hash"""
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password: str) -> str:
+    """Hash a password"""
+    return pwd_context.hash(password)
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    """Create a JWT access token"""
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
+    return encoded_jwt
+
+def create_session_id() -> str:
+    """Create a secure session ID"""
+    return secrets.token_urlsafe(32)
+
+async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> CurrentUser:
+    """Get current authenticated user from JWT token"""
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    
+    try:
+        # Check if it's the old API token (for backward compatibility during transition)
+        if credentials.credentials == API_TOKEN:
+            # Return a fake admin user for legacy API token
+            return CurrentUser(
+                user_id=0,
+                username="legacy_admin",
+                role=UserRole.ADMINISTRATOR,
+                active=True
+            )
+        
+        # Try to decode JWT token
+        payload = jwt.decode(credentials.credentials, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+    except JWTError:
+        raise credentials_exception
+    
+    # Get user from database
+    with get_db() as conn:
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute("SELECT * FROM users WHERE username = %s AND active = 1", (username,))
+        user = cursor.fetchone()
+        cursor.close()
+        
+        if user is None:
+            raise credentials_exception
+        
+        return CurrentUser(
+            user_id=user["user_id"],
+            username=user["username"],
+            role=UserRole(user["role"]),
+            active=user["active"]
+        )
+
+def require_role(required_roles: List[UserRole]):
+    """Decorator factory for role-based access control"""
+    def role_checker(current_user: CurrentUser = Depends(get_current_user)) -> CurrentUser:
+        if current_user.role not in required_roles:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"Insufficient permissions. Required roles: {[role.value for role in required_roles]}"
+            )
+        return current_user
+    return role_checker
+
+# Convenience functions for common role requirements
+def require_admin() -> CurrentUser:
+    return Depends(require_role([UserRole.ADMINISTRATOR]))
+
+def require_write_access() -> CurrentUser:
+    return Depends(require_role([UserRole.ADMINISTRATOR, UserRole.OPERATOR]))
+
+def require_read_access() -> CurrentUser:
+    return Depends(require_role([UserRole.ADMINISTRATOR, UserRole.OPERATOR, UserRole.READ_ONLY]))
+
+# Legacy authentication (for backward compatibility)
 def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
-    """Verify API token"""
+    """Legacy API token verification - deprecated, use get_current_user instead"""
     if credentials.credentials != API_TOKEN:
         raise HTTPException(status_code=401, detail="Invalid authentication token")
     return credentials.credentials
 
-# Pydantic models
+# Pydantic models for authentication
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    expires_in: int
+    user: dict
+
+class CreateUserRequest(BaseModel):
+    username: str
+    password: str
+    role: UserRole
+
+class UpdateUserRequest(BaseModel):
+    password: Optional[str] = None
+    role: Optional[UserRole] = None
+    active: Optional[bool] = None
+
+class UserResponse(BaseModel):
+    user_id: int
+    username: str
+    role: UserRole
+    created_at: datetime
+    last_login: Optional[datetime] = None
+    active: bool
+
+# Pydantic models for mailing list functionality
 class MailingList(BaseModel):
     list_id: Optional[int] = None
     list_name: str
@@ -107,6 +251,179 @@ class BulkImportResult(BaseModel):
     subscriptions_added: int
     errors: List[str]
 
+# Authentication routes
+@app.post("/auth/login", response_model=TokenResponse)
+async def login(login_request: LoginRequest, request: Request):
+    """Authenticate user and return JWT token"""
+    with get_db() as conn:
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute("SELECT * FROM users WHERE username = %s AND active = 1", (login_request.username,))
+        user = cursor.fetchone()
+        
+        if not user or not verify_password(login_request.password, user["password_hash"]):
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Incorrect username or password"
+            )
+        
+        # Update last login
+        cursor.execute("UPDATE users SET last_login = %s WHERE user_id = %s", 
+                      (datetime.utcnow(), user["user_id"]))
+        conn.commit()
+        
+        # Create session record
+        session_id = create_session_id()
+        expires_at = datetime.utcnow() + timedelta(hours=SESSION_EXPIRE_HOURS)
+        
+        client_ip = request.client.host if request.client else None
+        user_agent = request.headers.get("user-agent", "")
+        
+        cursor.execute("""
+            INSERT INTO user_sessions (session_id, user_id, expires_at, ip_address, user_agent)
+            VALUES (%s, %s, %s, %s, %s)
+        """, (session_id, user["user_id"], expires_at, client_ip, user_agent))
+        conn.commit()
+        cursor.close()
+        
+        # Create JWT token
+        access_token_expires = timedelta(minutes=JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
+        access_token = create_access_token(
+            data={"sub": user["username"]}, expires_delta=access_token_expires
+        )
+        
+        return TokenResponse(
+            access_token=access_token,
+            expires_in=JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60,
+            user={
+                "user_id": user["user_id"],
+                "username": user["username"],
+                "role": user["role"],
+                "active": user["active"]
+            }
+        )
+
+@app.post("/auth/logout")
+async def logout(current_user: CurrentUser = Depends(get_current_user)):
+    """Logout current user (invalidate sessions)"""
+    if current_user.user_id == 0:  # Legacy admin user
+        return {"message": "Logged out successfully"}
+    
+    with get_db() as conn:
+        cursor = conn.cursor()
+        cursor.execute("UPDATE user_sessions SET active = 0 WHERE user_id = %s", (current_user.user_id,))
+        conn.commit()
+        cursor.close()
+    
+    return {"message": "Logged out successfully"}
+
+@app.get("/users", response_model=List[UserResponse])
+async def get_users(current_user: CurrentUser = require_admin()):
+    """Get all users (admin only)"""
+    with get_db() as conn:
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute("SELECT user_id, username, role, created_at, last_login, active FROM users ORDER BY username")
+        users = cursor.fetchall()
+        cursor.close()
+        return users
+
+@app.post("/users", response_model=UserResponse, status_code=201)
+async def create_user(user_request: CreateUserRequest, current_user: CurrentUser = require_admin()):
+    """Create a new user (admin only)"""
+    # Check if username already exists
+    with get_db() as conn:
+        cursor = conn.cursor(dictionary=True)
+        cursor.execute("SELECT user_id FROM users WHERE username = %s", (user_request.username,))
+        existing_user = cursor.fetchone()
+        
+        if existing_user:
+            raise HTTPException(status_code=400, detail="Username already exists")
+        
+        # Hash password and create user
+        password_hash = get_password_hash(user_request.password)
+        cursor.execute("""
+            INSERT INTO users (username, password_hash, role)
+            VALUES (%s, %s, %s)
+        """, (user_request.username, password_hash, user_request.role.value))
+        
+        conn.commit()
+        user_id = cursor.lastrowid
+        
+        # Return created user
+        cursor.execute("""
+            SELECT user_id, username, role, created_at, last_login, active 
+            FROM users WHERE user_id = %s
+        """, (user_id,))
+        new_user = cursor.fetchone()
+        cursor.close()
+        
+        return new_user
+
+@app.patch("/users/{user_id}", response_model=UserResponse)
+async def update_user(user_id: int, updates: UpdateUserRequest, current_user: CurrentUser = require_admin()):
+    """Update a user (admin only)"""
+    with get_db() as conn:
+        cursor = conn.cursor(dictionary=True)
+        
+        # Build update query dynamically
+        update_fields = []
+        values = []
+        
+        if updates.password is not None:
+            update_fields.append("password_hash = %s")
+            values.append(get_password_hash(updates.password))
+        if updates.role is not None:
+            update_fields.append("role = %s")
+            values.append(updates.role.value)
+        if updates.active is not None:
+            update_fields.append("active = %s")
+            values.append(updates.active)
+        
+        if not update_fields:
+            raise HTTPException(status_code=400, detail="No fields to update")
+        
+        values.append(user_id)
+        query = f"UPDATE users SET {', '.join(update_fields)} WHERE user_id = %s"
+        
+        cursor.execute(query, values)
+        conn.commit()
+        
+        # Return updated user
+        cursor.execute("""
+            SELECT user_id, username, role, created_at, last_login, active 
+            FROM users WHERE user_id = %s
+        """, (user_id,))
+        updated_user = cursor.fetchone()
+        cursor.close()
+        
+        if not updated_user:
+            raise HTTPException(status_code=404, detail="User not found")
+        return updated_user
+
+@app.delete("/users/{user_id}", status_code=204)
+async def delete_user(user_id: int, current_user: CurrentUser = require_admin()):
+    """Delete a user (admin only)"""
+    if user_id == current_user.user_id:
+        raise HTTPException(status_code=400, detail="Cannot delete your own account")
+    
+    with get_db() as conn:
+        cursor = conn.cursor()
+        cursor.execute("DELETE FROM users WHERE user_id = %s", (user_id,))
+        conn.commit()
+        
+        if cursor.rowcount == 0:
+            raise HTTPException(status_code=404, detail="User not found")
+        cursor.close()
+
+@app.get("/auth/me", response_model=dict)
+async def get_current_user_info(current_user: CurrentUser = Depends(get_current_user)):
+    """Get current user information"""
+    return {
+        "user_id": current_user.user_id,
+        "username": current_user.username,
+        "role": current_user.role.value,
+        "active": current_user.active
+    }
+
 # Routes
 @app.get("/")
 async def root():
@@ -132,7 +449,7 @@ async def health():
 
 # Mailing Lists endpoints
 @app.get("/lists", response_model=List[MailingList])
-async def get_lists(token: str = Depends(verify_token)):
+async def get_lists(current_user: CurrentUser = require_read_access()):
     """Get all mailing lists"""
     with get_db() as conn:
         cursor = conn.cursor(dictionary=True)
@@ -142,7 +459,7 @@ async def get_lists(token: str = Depends(verify_token)):
         return lists
 
 @app.get("/lists/{list_id}", response_model=MailingList)
-async def get_list(list_id: int, token: str = Depends(verify_token)):
+async def get_list(list_id: int, current_user: CurrentUser = require_read_access()):
     """Get a specific mailing list"""
     with get_db() as conn:
         cursor = conn.cursor(dictionary=True)
@@ -155,7 +472,7 @@ async def get_list(list_id: int, token: str = Depends(verify_token)):
         return mailing_list
 
 @app.post("/lists", response_model=MailingList, status_code=201)
-async def create_list(mailing_list: MailingList, token: str = Depends(verify_token)):
+async def create_list(mailing_list: MailingList, current_user: CurrentUser = require_write_access()):
     """Create a new mailing list"""
     with get_db() as conn:
         cursor = conn.cursor()
@@ -172,7 +489,7 @@ async def create_list(mailing_list: MailingList, token: str = Depends(verify_tok
             raise HTTPException(status_code=400, detail=f"Failed to create list: {str(e)}")
 
 @app.patch("/lists/{list_id}", response_model=MailingList)
-async def update_list(list_id: int, updates: MailingListUpdate, token: str = Depends(verify_token)):
+async def update_list(list_id: int, updates: MailingListUpdate, current_user: CurrentUser = require_write_access()):
     """Update a mailing list"""
     with get_db() as conn:
         cursor = conn.cursor(dictionary=True)
@@ -210,7 +527,7 @@ async def update_list(list_id: int, updates: MailingListUpdate, token: str = Dep
         return updated_list
 
 @app.delete("/lists/{list_id}", status_code=204)
-async def delete_list(list_id: int, token: str = Depends(verify_token)):
+async def delete_list(list_id: int, current_user: CurrentUser = require_write_access()):
     """Delete a mailing list"""
     with get_db() as conn:
         cursor = conn.cursor()
@@ -223,7 +540,7 @@ async def delete_list(list_id: int, token: str = Depends(verify_token)):
 
 # Members endpoints
 @app.get("/members", response_model=List[Member])
-async def get_members(token: str = Depends(verify_token)):
+async def get_members(current_user: CurrentUser = require_read_access()):
     """Get all members"""
     with get_db() as conn:
         cursor = conn.cursor(dictionary=True)
@@ -233,7 +550,7 @@ async def get_members(token: str = Depends(verify_token)):
         return members
 
 @app.get("/members/{member_id}", response_model=Member)
-async def get_member(member_id: int, token: str = Depends(verify_token)):
+async def get_member(member_id: int, current_user: CurrentUser = require_read_access()):
     """Get a specific member"""
     with get_db() as conn:
         cursor = conn.cursor(dictionary=True)
@@ -246,7 +563,7 @@ async def get_member(member_id: int, token: str = Depends(verify_token)):
         return member
 
 @app.post("/members", response_model=Member, status_code=201)
-async def create_member(member: Member, token: str = Depends(verify_token)):
+async def create_member(member: Member, current_user: CurrentUser = require_write_access()):
     """Create a new member"""
     with get_db() as conn:
         cursor = conn.cursor()
@@ -263,7 +580,7 @@ async def create_member(member: Member, token: str = Depends(verify_token)):
             raise HTTPException(status_code=400, detail=f"Failed to create member: {str(e)}")
 
 @app.patch("/members/{member_id}", response_model=Member)
-async def update_member(member_id: int, updates: MemberUpdate, token: str = Depends(verify_token)):
+async def update_member(member_id: int, updates: MemberUpdate, current_user: CurrentUser = require_write_access()):
     """Update a member"""
     with get_db() as conn:
         cursor = conn.cursor(dictionary=True)
@@ -296,7 +613,7 @@ async def update_member(member_id: int, updates: MemberUpdate, token: str = Depe
         return updated_member
 
 @app.delete("/members/{member_id}", status_code=204)
-async def delete_member(member_id: int, token: str = Depends(verify_token)):
+async def delete_member(member_id: int, current_user: CurrentUser = require_write_access()):
     """Delete a member"""
     with get_db() as conn:
         cursor = conn.cursor()
@@ -309,7 +626,7 @@ async def delete_member(member_id: int, token: str = Depends(verify_token)):
 
 # Subscription endpoints
 @app.get("/lists/{list_id}/members", response_model=List[Member])
-async def get_list_members(list_id: int, token: str = Depends(verify_token)):
+async def get_list_members(list_id: int, current_user: CurrentUser = require_read_access()):
     """Get all members of a specific list"""
     with get_db() as conn:
         cursor = conn.cursor(dictionary=True)
@@ -325,7 +642,7 @@ async def get_list_members(list_id: int, token: str = Depends(verify_token)):
         return members
 
 @app.post("/subscriptions", status_code=201)
-async def subscribe_member(subscription: Subscription, token: str = Depends(verify_token)):
+async def subscribe_member(subscription: Subscription, current_user: CurrentUser = require_write_access()):
     """Subscribe a member to a list"""
     with get_db() as conn:
         cursor = conn.cursor()
@@ -359,7 +676,7 @@ async def subscribe_member(subscription: Subscription, token: str = Depends(veri
             raise HTTPException(status_code=400, detail=f"Failed to create subscription: {str(e)}")
 
 @app.delete("/subscriptions")
-async def unsubscribe_member(list_email: EmailStr, member_email: EmailStr, token: str = Depends(verify_token)):
+async def unsubscribe_member(list_email: EmailStr, member_email: EmailStr, current_user: CurrentUser = require_write_access()):
     """Unsubscribe a member from a list"""
     with get_db() as conn:
         cursor = conn.cursor()
@@ -380,7 +697,7 @@ async def unsubscribe_member(list_email: EmailStr, member_email: EmailStr, token
         return {"message": "Unsubscribed successfully"}
 
 @app.post("/bulk-import", response_model=BulkImportResult)
-async def bulk_import_members(bulk_request: BulkImportRequest, token: str = Depends(verify_token)):
+async def bulk_import_members(bulk_request: BulkImportRequest, current_user: CurrentUser = require_write_access()):
     """Bulk import members from CSV data and subscribe them to specified lists"""
     
     result = BulkImportResult(