Added Support of API Key Authentication

backend/app/api/deps.py

@@ -1,10 +1,10 @@
 from collections.abc import Generator
-from typing import Annotated
+from typing import Annotated, Optional
 
 import jwt
-from fastapi import Depends, HTTPException, status, Request
+from fastapi import Depends, HTTPException, status, Request, Header, Security
 from fastapi.responses import JSONResponse
-from fastapi.security import OAuth2PasswordBearer
+from fastapi.security import OAuth2PasswordBearer, APIKeyHeader
 from jwt.exceptions import InvalidTokenError
 from pydantic import ValidationError
 from sqlmodel import Session, select
@@ -13,42 +13,67 @@
 from app.core.config import settings
 from app.core.db import engine
 from app.utils import APIResponse
-from app.models import TokenPayload, User, UserProjectOrg, ProjectUser, Project, Organization
+from app.crud.organization import validate_organization
+from app.crud.api_key import get_api_key_by_value
+from app.models import TokenPayload, User, UserProjectOrg, UserOrganization, ProjectUser, Project, Organization
 
 reusable_oauth2 = OAuth2PasswordBearer(
-    tokenUrl=f"{settings.API_V1_STR}/login/access-token"
+    tokenUrl=f"{settings.API_V1_STR}/login/access-token",
+    auto_error= False
 )
 
 
 def get_db() -> Generator[Session, None, None]:
     with Session(engine) as session:
         yield session
 
-
+api_key_header = APIKeyHeader(name="X-API-KEY", auto_error=False)
 SessionDep = Annotated[Session, Depends(get_db)]
 TokenDep = Annotated[str, Depends(reusable_oauth2)]
 
+def get_current_user(
+    session: SessionDep,
+    token: TokenDep,
+    api_key: Annotated[str, Depends(api_key_header)],
+) -> UserOrganization:
+    """Authenticate user via API Key first, fallback to JWT token."""
 
-def get_current_user(session: SessionDep, token: TokenDep) -> User:
-    try:
-        payload = jwt.decode(
-            token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
-        )
-        token_data = TokenPayload(**payload)
-    except (InvalidTokenError, ValidationError):
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Could not validate credentials",
-        )
-    user = session.get(User, token_data.sub)
-    if not user:
-        raise HTTPException(status_code=404, detail="User not found")
-    if not user.is_active:
-        raise HTTPException(status_code=400, detail="Inactive user")
-    return user
+    if api_key:
+        api_key_record = get_api_key_by_value(session, api_key)
+        if not api_key_record:
+            raise HTTPException(status_code=401, detail="Invalid API Key")
+
+        user = session.get(User, api_key_record.user_id)
+        if not user:
+            raise HTTPException(status_code=404, detail="User linked to API Key not found")
 
+        validate_organization(session, api_key_record.organization_id)
 
-CurrentUser = Annotated[User, Depends(get_current_user)]
+        # Return UserOrganization model with organization ID
+        return UserOrganization(**user.model_dump(), organization_id=api_key_record.organization_id)
+
+    if token:
+        try:
+            payload = jwt.decode(
+                token, settings.SECRET_KEY, algorithms=[security.ALGORITHM]
+            )
+            token_data = TokenPayload(**payload)
+        except (InvalidTokenError, ValidationError):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Could not validate credentials",
+            )
+        user = session.get(User, token_data.sub)
+        if not user:
+            raise HTTPException(status_code=404, detail="User not found")
+        if not user.is_active:
+            raise HTTPException(status_code=400, detail="Inactive user")
+
+        return UserOrganization(**user.model_dump(), organization_id=None)
+
+    raise HTTPException(status_code=401, detail="Invalid Authorization format")
+
+CurrentUser = Annotated[UserOrganization, Depends(get_current_user)]
 
 
 def get_current_active_superuser(current_user: CurrentUser) -> User:
@@ -78,6 +103,8 @@ def verify_user_project_organization(
     Verify that the authenticated user is part of the project
     and that the project belongs to the organization.
     """
+    if current_user.organization_id and current_user.organization_id != organization_id:
+        raise HTTPException(status_code=403, detail="User does not belong to the specified organization")
 
     project_organization = db.exec(
         select(Project, Organization)
@@ -105,9 +132,11 @@ def verify_user_project_organization(
         raise HTTPException(status_code=403, detail="Project does not belong to the organization")
 
 
+    current_user.organization_id = organization_id
+
     # Superuser bypasses all checks
     if current_user.is_superuser:
-        return UserProjectOrg(**current_user.model_dump(), project_id=project_id, organization_id=organization_id)
+        return UserProjectOrg(**current_user.model_dump(), project_id=project_id)
 
     # Check if the user is part of the project
     user_in_project = db.exec(
@@ -121,4 +150,4 @@ def verify_user_project_organization(
     if not user_in_project:
         raise HTTPException(status_code=403, detail="User is not part of the project")
 
-    return UserProjectOrg(**current_user.model_dump(), project_id=project_id, organization_id=organization_id)
+    return UserProjectOrg(**current_user.model_dump(), project_id=project_id)

backend/app/api/main.py

@@ -1,5 +1,5 @@
 from fastapi import APIRouter
-from app.api.routes import items, login, private, users, utils,project,organization, project_user
+from app.api.routes import items, login, private, users, utils,project,organization, project_user, api_keys
 from app.core.config import settings
 
 api_router = APIRouter()
@@ -10,6 +10,7 @@
 api_router.include_router(organization.router)
 api_router.include_router(project.router)
 api_router.include_router(project_user.router)
+api_router.include_router(api_keys.router)
 
 
 if settings.ENVIRONMENT == "local":

backend/app/api/routes/api_keys.py

@@ -0,0 +1,98 @@
+import uuid
+from fastapi import APIRouter, Depends, HTTPException
+from sqlmodel import Session
+from app.api.deps import get_db, get_current_active_superuser
+from app.crud.api_key import create_api_key, get_api_key, get_api_keys_by_organization, delete_api_key, get_api_key_by_user_org
+from app.crud.organization import get_organization_by_id, validate_organization
+from app.crud.project_user import is_user_part_of_organization
+from app.models import APIKeyPublic, User
+from app.utils import APIResponse
+
+router = APIRouter(prefix="/apikeys", tags=["API Keys"])
+
+
+# Create API Key
+@router.post("/", response_model=APIResponse[APIKeyPublic])
+def create_key(
+    organization_id: int,
+    user_id: uuid.UUID,
+    session: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_superuser)
+):
+    """
+    Generate a new API key for the user's organization.
+    """
+    try:
+        # Validate organization
+        validate_organization(session, organization_id)
+
+        # Check if user belongs to organization
+        if not is_user_part_of_organization(session, user_id, organization_id):
+            raise HTTPException(status_code=403, detail="User is not part of any project in the organization")
+
+        existing_api_key = get_api_key_by_user_org(session, organization_id, user_id)
+        if existing_api_key:
+            raise HTTPException(status_code=400, detail="API Key already exists for this user and organization")
+
+        # Create and return API key
+        api_key = create_api_key(session, organization_id=organization_id, user_id=user_id)
+        return APIResponse.success_response(api_key)
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+# List API Keys
+@router.get("/", response_model=APIResponse[list[APIKeyPublic]])
+def list_keys(
+    organization_id: int,
+    session: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_superuser),
+):
+    """
+    Retrieve all API keys for the user's organization.
+    """
+    try:
+        # Validate organization
+        validate_organization(session, organization_id)
+
+        # Retrieve API keys
+        api_keys = get_api_keys_by_organization(session, organization_id)
+        return APIResponse.success_response(api_keys)
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+# Get API Key by ID
+@router.get("/{api_key_id}", response_model=APIResponse[APIKeyPublic])
+def get_key(
+    api_key_id: int,
+    session: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_superuser)
+):
+    """
+    Retrieve an API key by ID.
+    """
+    api_key = get_api_key(session, api_key_id)
+    if not api_key:
+        raise HTTPException(status_code=404, detail="API Key does not exist")
+
+    return APIResponse.success_response(api_key)
+
+
+# Revoke API Key (Soft Delete)
+@router.delete("/{api_key_id}", response_model=APIResponse[dict])
+def revoke_key(
+    api_key_id: int,
+    session: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_superuser)
+):
+    """
+    Soft delete an API key (revoke access).
+    """
+    try:
+        delete_api_key(session, api_key_id)
+        return APIResponse.success_response({"message": "API key revoked successfully"})
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))

backend/app/crud/api_key.py

@@ -0,0 +1,81 @@
+import uuid
+import secrets
+from datetime import datetime
+from sqlmodel import Session, select
+
+from app.models import APIKey, APIKeyPublic
+
+
+# Create API Key
+def create_api_key(session: Session, organization_id: uuid.UUID, user_id: uuid.UUID) -> APIKeyPublic:
+    """
+    Generates a new API key for an organization and associates it with a user.
+    """
+    api_key = APIKey(
+        key='ApiKey '+secrets.token_urlsafe(32),
+        organization_id=organization_id,
+        user_id=user_id
+    )
+
+    session.add(api_key)
+    session.commit()
+    session.refresh(api_key)
+
+    return APIKeyPublic.model_validate(api_key)
+
+
+# Get API Key by ID
+def get_api_key(session: Session, api_key_id: int) -> APIKeyPublic | None:
+    """
+    Retrieves an API key by its ID if it exists and is not deleted.
+    """
+    api_key = session.exec(
+        select(APIKey).where(APIKey.id == api_key_id, APIKey.is_deleted == False)
+    ).first()
+
+    return APIKeyPublic.model_validate(api_key) if api_key else None
+
+
+# Get API Keys for an Organization
+def get_api_keys_by_organization(session: Session, organization_id: uuid.UUID) -> list[APIKeyPublic]:
+    """
+    Retrieves all active API keys associated with an organization.
+    """
+    api_keys = session.exec(
+        select(APIKey).where(APIKey.organization_id == organization_id, APIKey.is_deleted == False)
+    ).all()
+
+    return [APIKeyPublic.model_validate(api_key) for api_key in api_keys]
+
+
+# Soft Delete (Revoke) API Key
+def delete_api_key(session: Session, api_key_id: int) -> None:
+    """
+    Soft deletes (revokes) an API key by marking it as deleted.
+    """
+    api_key = session.get(APIKey, api_key_id)
+
+    if not api_key or api_key.is_deleted:
+        raise ValueError("API key not found or already deleted.")
+
+    api_key.is_deleted = True
+    api_key.deleted_at = datetime.utcnow()
+
+    session.add(api_key)
+    session.commit()
+
+def get_api_key_by_value(session: Session, api_key_value: str) -> APIKey | None:
+    """
+    Retrieve an API Key record by its value.
+    """
+    return session.exec(select(APIKey).where(APIKey.key == api_key_value, APIKey.is_deleted == False)).first()
+
+def get_api_key_by_user_org(session: Session, organization_id: int, user_id: str) -> APIKey | None:
+    """
+    Retrieve an API key for a specific user and organization.
+    """
+    statement = select(APIKey).where(
+        APIKey.organization_id == organization_id,
+        APIKey.user_id == user_id
+    )
+    return session.exec(statement).first()

backend/app/crud/organization.py

@@ -12,10 +12,25 @@ def create_organization(*, session: Session, org_create: OrganizationCreate) ->
     return db_org
 
 
-def get_organization_by_id(*, session: Session, org_id: int) -> Optional[Organization]:
+# Get organization by ID
+def get_organization_by_id(session: Session, org_id: int) -> Optional[Organization]:
     statement = select(Organization).where(Organization.id == org_id)
     return session.exec(statement).first()
 
 def get_organization_by_name(*, session: Session, name: str) -> Optional[Organization]:
     statement = select(Organization).where(Organization.name == name)
     return session.exec(statement).first()
+
+# Validate if organization exists and is active
+def validate_organization(session: Session, org_id: int) -> Organization:
+    """
+    Ensures that an organization exists and is active.
+    """
+    organization = get_organization_by_id(session, org_id)
+    if not organization:
+        raise ValueError("Organization not found")
+
+    if not organization.is_active:
+        raise ValueError("Organization is not active")
+
+    return organization

backend/app/crud/project_user.py

@@ -1,6 +1,6 @@
 import uuid
 from sqlmodel import Session, select, delete, func
-from app.models import ProjectUser, ProjectUserPublic, User
+from app.models import ProjectUser, ProjectUserPublic, User, Project
 from datetime import datetime
 
 
@@ -80,3 +80,21 @@ def get_users_by_project(
     users = session.exec(statement).all()
 
     return [ProjectUserPublic.model_validate(user) for user in users], total_count
+
+
+# Check if a user belongs to an at least one project in organization
+def is_user_part_of_organization(session: Session, user_id: uuid.UUID, org_id: int) -> bool:
+    """
+    Checks if a user is part of at least one project within the organization.
+    """
+    user_in_org = session.exec(
+        select(ProjectUser)
+        .join(Project, ProjectUser.project_id == Project.id)
+        .where(
+            Project.organization_id == org_id,
+            ProjectUser.user_id == user_id,
+            ProjectUser.is_deleted == False
+        )
+    ).first()
+
+    return bool(user_in_org)