flowCreate.solutions

Authentication & Authorization

This document details the standard authentication and authorization implementation patterns, including JWT tokens, role-based access control (RBAC), and dependency injection.

Authentication Methods

1. JWT Tokens (Primary)

OAuth2 with JWT (JSON Web Tokens) for API authentication.

Token Generation Pattern:

from jose import jwt
from datetime import datetime, timedelta
import os

SECRET_KEY = os.getenv("SECRET_KEY")

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(hours=24)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")
    return encoded_jwt

Token Payload Example:

{
  "user_id": "user-abc-123",
  "email": "[email protected]",
  "user_type": "user",
  "active_tenant_id": "tenant-xyz-789",
  "exp": 1704412800
}

Usage in Requests:

curl -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIs..." \
  https://api.example.com/endpoint

2. API Keys (Integrations)

For third-party integrations.

Storage: Hashed with bcrypt in the configuration database.

Authentication:

curl -H "X-CLIENT-ID: client_123" \
     -H "X-API-KEY: fdc0b295d9e4fe54e6df..." \
  https://api.example.com/v1/endpoint

3. HTTP Basic Auth (Documentation)

For /docs and /redoc endpoints only.

Configuration:

DOCS_USERNAME=admin
DOCS_PASSWORD=secure-password

Browser prompts for credentials when accessing documentation.

Authorization

Role-Based Access Control (RBAC)

User Types:

  • super_admin - System administrators (cross-tenant access, system operations)
  • user - Regular users (tenant-level access based on role)

Tenant Roles (for regular users):

  • admin - Administrators (full tenant access)
  • owner - Owners (full tenant access)
  • member - Members (limited access)

Implementation Patterns

Basic Authentication Dependency:

from database.dependencies import get_current_user

@router.get("/protected-endpoint")
async def protected_endpoint(
    token: dict = Depends(get_current_user)
):
    user_id = token["user_id"]
    # Endpoint implementation

Admin-Only Endpoints:

from database.dependencies import get_current_user, require_admin_role

@router.post("/admin-endpoint")
async def admin_endpoint(
    token: dict = Depends(get_current_user),
    _: None = Depends(require_admin_role)
):
    # Only admins and owners can access
    pass

Super Admin Endpoints:

from utils.auth import get_current_super_admin

@router.post("/system/admin-endpoint")
async def system_admin_endpoint(
    token: dict = Depends(get_current_super_admin)
):
    # Only super_admin user type can access
    pass

Resource Isolation

Every request validates access to the requested resource (e.g., tenant/company isolation):

async def has_resource_access(
    db: AsyncSession,
    user_id: str,
    resource_id: str
) -> bool:
    """Verify user has access to the specific resource/tenant."""
    user = await get_user_by_id(db, user_id)
    
    # Super admins have access to everything
    if user.user_type == "super_admin":
        return True
    
    # Regular users must belong to the tenant owning the resource
    return user.tenant_id == resource_id

Authentication Flow

Login Flow

1. User submits credentials → POST /api/v1/auth/login
2. Server validates credentials (bcrypt compare)
3. Server generates JWT token with user data
4. Token returned to client
5. Client stores token (localStorage/sessionStorage)
6. Client includes token in Authorization header for subsequent requests
7. Server validates token on each request

Token Validation Dependency

from jose import jwt, JWTError
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        # Decode and validate token
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        
        # Extract user data
        user_id = payload.get("user_id")
        if user_id is None:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        return payload
        
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Password Security

Password Hashing

Uses bcrypt for secure password hashing:

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# Hash password on registration
hashed_password = pwd_context.hash(plain_password)

# Verify password on login
is_valid = pwd_context.verify(plain_password, hashed_password)

Password Requirements

  • Minimum length: 8 characters
  • Maximum length: 128 characters
  • No sanitization/mutation: Passwords must never be sanitized or transformed (credentials are validated, not rewritten).
  • Complexity: No specific requirements (encourages passphrases)

Password Validation Schema

class UserCreate(BaseModel):
    email: str = Field(..., max_length=255)
    password: str = Field(..., min_length=8, max_length=128)
    full_name: str = Field(..., max_length=100)
    
    # Do not sanitize/mutate passwords. Prevent XSS by:
    # - never reflecting passwords in responses/errors
    # - never logging passwords
    # - hashing at rest (bcrypt/passlib)
    
    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        from email_validator import validate_email, EmailNotValidError
        try:
            validate_email(v)
            return v.lower()
        except EmailNotValidError:
            raise ValueError("Invalid email address")

Security Best Practices

Token Management

DO:

  • ✅ Use HTTPS only in production
  • ✅ Set reasonable expiration (24 hours)
  • ✅ Store tokens securely (httpOnly cookies or secure storage)
  • ✅ Invalidate tokens on logout
  • ✅ Rotate SECRET_KEY periodically (every 90 days)
  • ✅ Include user context in token payload

DON'T:

  • ❌ Store tokens in URLs
  • ❌ Send tokens in GET parameters
  • ❌ Share tokens between users
  • ❌ Use expired tokens
  • ❌ Log full tokens (log last 4 chars only)

Secret Key Management

Generate Strong Key:

python -c "import secrets; print(secrets.token_urlsafe(32))"

Store Securely:

  • Environment variable (NEVER in code)
  • Secret management service
  • Encrypted configuration file

Rotate Regularly:

  • Every 90 days minimum
  • Immediately if compromised
  • When team members leave

API Key Management

For integration API keys:

  • Store hashed (bcrypt)
  • Show plain text only once (during generation)
  • Allow regeneration to invalidate old keys
  • Scope per tenant/resource