LynkedUpPro_CRM/docs/backend/01_authentication_module.md

B1 — Authentication Module

Module: B1 Depends on: B0 (Project Overview), B2 (Database Schema — users table) Read before: B3, B4, B5, B6, B7, B8 (every router imports require_role from here)

---

1. Overview

This module defines the complete authentication and authorization system for LynkedUpPro. It covers:

• Password hashing (bcrypt via passlib)
• JWT access + refresh token lifecycle
• httpOnly cookie storage strategy
• All 7 login flows across 7 roles
• The get_current_user() and require_role() FastAPI dependencies used by every other router
• The permission system derived from src/utils/permissions.js
• All auth endpoints (/auth/login, /auth/refresh, /auth/logout, /auth/me)

No other module should re-implement any of this. All routers import and use the dependencies defined here.

---

2. Files to Create

app/
├── routers/auth.py          ← All /api/v1/auth/* endpoints
├── services/auth_service.py ← Business logic: verify password, create tokens, rotate refresh
├── utils/security.py        ← JWT encode/decode, bcrypt helpers (no business logic)
└── dependencies.py          ← get_db(), get_current_user(), require_role() — used everywhere

---

3. Password Hashing

Use passlib with bcrypt scheme. Never store or compare plain text passwords.

# app/utils/security.py
from passlib.context import CryptContext

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

def hash_password(plain: str) -> str:
    """Hash a plain-text password. Use during user creation and seed."""
    return pwd_context.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    """Verify plain text against stored bcrypt hash. Use during login."""
    return pwd_context.verify(plain, hashed)

Seed note: All 17 mock users have password: "password". The seed script must bcrypt-hash this before inserting. The plain text "password" must never appear in the database.

---

4. JWT Tokens

Token Types

Type	Lifetime	Stored In	Purpose
Access token	15 minutes	httpOnly cookie (access_token)	Authorizes API requests
Refresh token	7 days	httpOnly cookie (refresh_token)	Issues new access tokens silently

JWT Payload (Claims)

Access token:

{
  "sub": "3f8a1c2d-...(user UUID)",
  "role": "FIELD_AGENT",
  "legacy_id": "e1",
  "type": "access",
  "iat": 1740400000,
  "exp": 1740400900
}

Refresh token:

{
  "sub": "3f8a1c2d-...(user UUID)",
  "type": "refresh",
  "iat": 1740400000,
  "exp": 1741004800
}

Why legacy_id in the access token: During the transition period, some frontend components still look up users by their mock ID (e1, own_001). Including it in the token avoids an extra DB lookup per request.

Token Creation

# app/utils/security.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from app.config import settings

def create_access_token(user_id: str, role: str, legacy_id: str | None = None) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
    payload = {
        "sub": str(user_id),
        "role": role,
        "type": "access",
        "exp": expire,
    }
    if legacy_id:
        payload["legacy_id"] = legacy_id
    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)

def create_refresh_token(user_id: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(days=settings.refresh_token_expire_days)
    payload = {
        "sub": str(user_id),
        "type": "refresh",
        "exp": expire,
    }
    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)

def decode_token(token: str) -> dict:
    """Raises JWTError if invalid or expired."""
    return jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])

Refresh Token Rotation Security

Refresh tokens are rotated on every use. The hashed value of the current refresh token is stored in users.refresh_token_hash. On refresh:

1. Decode the incoming refresh token (verify not expired)
2. Load user from DB, get refresh_token_hash
3. verify_password(incoming_token, stored_hash) — if mismatch, revoke all tokens (possible reuse attack)
4. If match: issue new access + refresh token pair, update refresh_token_hash

# app/utils/security.py
def hash_token(token: str) -> str:
    """Hash a refresh token for safe storage. Uses same bcrypt context."""
    return pwd_context.hash(token)

def verify_token_hash(token: str, hashed: str) -> bool:
    return pwd_context.verify(token, hashed)

---

5. Cookie Configuration

All tokens are delivered as httpOnly cookies. This prevents XSS attacks from stealing tokens via document.cookie or localStorage.

# Shared cookie settings used in auth router
ACCESS_COOKIE_PARAMS = {
    "key": "access_token",
    "httponly": True,
    "secure": True,        # HTTPS only in production; set False in dev
    "samesite": "strict",  # Prevents CSRF
    "max_age": 60 * 15,    # 15 minutes in seconds
    "path": "/",
}

REFRESH_COOKIE_PARAMS = {
    "key": "refresh_token",
    "httponly": True,
    "secure": True,
    "samesite": "strict",
    "max_age": 60 * 60 * 24 * 7,  # 7 days in seconds
    "path": "/api/v1/auth",        # Scoped — only sent to auth endpoints
}

Why path="/api/v1/auth" for the refresh token: Limits the refresh token cookie to auth endpoints only. It won't be sent with every API request, reducing exposure.

---

6. FastAPI Dependencies

These are the three shared dependencies imported by every other router.

# app/dependencies.py
from fastapi import Cookie, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from jose import JWTError

from app.database import get_db
from app.models.user import User
from app.utils.security import decode_token

async def get_current_user(
    access_token: str | None = Cookie(default=None),
    db: AsyncSession = Depends(get_db),
) -> User:
    """
    Extracts and validates the JWT from the httpOnly cookie.
    Raises 401 if missing, expired, or invalid.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Not authenticated",
        headers={"WWW-Authenticate": "Bearer"},
    )

    if not access_token:
        raise credentials_exception

    try:
        payload = decode_token(access_token)
        if payload.get("type") != "access":
            raise credentials_exception
        user_id: str = payload.get("sub")
        if not user_id:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await db.get(User, user_id)
    if user is None or user.deleted_at is not None or not user.is_active:
        raise credentials_exception

    return user


def require_role(*roles: str):
    """
    Returns a dependency that enforces role-based access.
    Usage:  current_user: User = Depends(require_role("ADMIN", "OWNER"))
    """
    async def checker(current_user: User = Depends(get_current_user)) -> User:
        if current_user.role not in roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Access denied. Required roles: {list(roles)}",
            )
        return current_user
    return checker

Usage example in any router:

from app.dependencies import require_role

@router.get("/admin/stats")
async def get_admin_stats(
    current_user: User = Depends(require_role("ADMIN", "OWNER")),
    db: AsyncSession = Depends(get_db),
):
    ...

---

7. Login Flows by Role

The frontend sends different type values from Login.jsx. This table maps each UI login type to the DB lookup strategy.

Frontend loginType	Backend strategy	DB column matched	Roles possible
customer	username + password	users.username	CUSTOMER
employee	emp_id + password	users.emp_id	FIELD_AGENT, ADMIN
owner	username or email + password	users.username or users.email	OWNER
contractor	username or email + password	users.username or users.email	CONTRACTOR
subcontractor	username or email + password	users.username or users.email	SUBCONTRACTOR
vendor	username or email + password	users.username or users.email	VENDOR

Note on employee type: The login form sends type: "employee" for both FIELD_AGENT and ADMIN (they use emp_id like FA001, ADM01). All other non-customer roles send their own type string and match on username.

Auth Service Implementation

# app/services/auth_service.py
from sqlalchemy import select, or_
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User
from app.utils.security import verify_password, create_access_token, create_refresh_token, hash_token

async def authenticate_user(
    db: AsyncSession,
    identifier: str,
    password: str,
    login_type: str,
) -> User | None:
    """
    Finds and verifies a user. Returns User on success, None on failure.
    Never raises — caller decides how to handle None.
    """
    stmt = None

    if login_type == "customer":
        stmt = select(User).where(
            User.username == identifier,
            User.role == "CUSTOMER",
            User.deleted_at.is_(None),
            User.is_active.is_(True),
        )
    elif login_type == "employee":
        stmt = select(User).where(
            User.emp_id == identifier,
            User.role.in_(["FIELD_AGENT", "ADMIN"]),
            User.deleted_at.is_(None),
            User.is_active.is_(True),
        )
    elif login_type in ("owner", "contractor", "subcontractor", "vendor"):
        role_map = {
            "owner": "OWNER",
            "contractor": "CONTRACTOR",
            "subcontractor": "SUBCONTRACTOR",
            "vendor": "VENDOR",
        }
        stmt = select(User).where(
            or_(User.username == identifier, User.email == identifier),
            User.role == role_map[login_type],
            User.deleted_at.is_(None),
            User.is_active.is_(True),
        )
    else:
        return None

    result = await db.execute(stmt)
    user = result.scalar_one_or_none()

    if user is None:
        return None
    if not verify_password(password, user.password_hash):
        return None

    return user


async def create_token_pair(db: AsyncSession, user: User) -> tuple[str, str]:
    """
    Creates access + refresh tokens and persists the refresh token hash.
    Returns (access_token, refresh_token).
    """
    access_token = create_access_token(
        user_id=str(user.id),
        role=user.role,
        legacy_id=user.legacy_id,
    )
    refresh_token = create_refresh_token(user_id=str(user.id))

    # Store hashed refresh token for rotation validation
    user.refresh_token_hash = hash_token(refresh_token)
    user.last_login_at = datetime.now(timezone.utc)
    await db.commit()

    return access_token, refresh_token

---

8. API Endpoints

8.1 POST /api/v1/auth/login

Purpose: Authenticate user, issue tokens as httpOnly cookies.

Request Body:

# app/schemas/auth.py
class LoginRequest(BaseModel):
    identifier: str    # username, emp_id, or email depending on type
    password: str
    type: Literal[
        "customer", "employee", "owner",
        "contractor", "subcontractor", "vendor"
    ]

Response Body (200):

class UserPublic(BaseModel):
    id: str            # UUID as string
    legacy_id: str | None
    full_name: str
    email: str
    role: str          # 'OWNER', 'ADMIN', 'FIELD_AGENT', etc.
    company_name: str | None
    xp: int | None
    streak_days: int | None
    achievements: list[str]

class LoginResponse(BaseModel):
    user: UserPublic
    message: str = "Login successful"

Error Responses:

• 401 — Invalid credentials (always generic — never reveal which field was wrong)
• 403 — Account deactivated (is_active = false)
• 422 — Missing or invalid request body fields

Router Implementation:

# app/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, Response, status
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/login", response_model=LoginResponse)
async def login(
    request: LoginRequest,
    response: Response,
    db: AsyncSession = Depends(get_db),
):
    user = await authenticate_user(db, request.identifier, request.password, request.type)

    if user is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials",
        )

    access_token, refresh_token = await create_token_pair(db, user)

    # Set httpOnly cookies
    response.set_cookie(value=access_token, **ACCESS_COOKIE_PARAMS)
    response.set_cookie(value=refresh_token, **REFRESH_COOKIE_PARAMS)

    # Write audit log
    await write_audit_log(db, actor_id=user.id, action="auth.login", resource_type="user", resource_id=str(user.id))

    return LoginResponse(user=UserPublic.model_validate(user))

---

8.2 POST /api/v1/auth/refresh

Purpose: Issue a new access token using the refresh token cookie.

Request: No body. Reads refresh_token from httpOnly cookie automatically.

Response (200):

class RefreshResponse(BaseModel):
    message: str = "Token refreshed"
    # New access_token set as cookie; not in body for security

Router Implementation:

@router.post("/refresh", response_model=RefreshResponse)
async def refresh_token(
    response: Response,
    refresh_token: str | None = Cookie(default=None),
    db: AsyncSession = Depends(get_db),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid or expired refresh token",
    )

    if not refresh_token:
        raise credentials_exception

    try:
        payload = decode_token(refresh_token)
        if payload.get("type") != "refresh":
            raise credentials_exception
        user_id = payload.get("sub")
    except JWTError:
        raise credentials_exception

    user = await db.get(User, user_id)
    if user is None or not user.is_active or user.deleted_at is not None:
        raise credentials_exception

    # Validate token hash (detect reuse attacks)
    if not user.refresh_token_hash or not verify_token_hash(refresh_token, user.refresh_token_hash):
        # Possible token reuse — revoke all sessions
        user.refresh_token_hash = None
        await db.commit()
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh token reuse detected. Please log in again.",
        )

    # Issue new token pair
    new_access, new_refresh = await create_token_pair(db, user)
    response.set_cookie(value=new_access, **ACCESS_COOKIE_PARAMS)
    response.set_cookie(value=new_refresh, **REFRESH_COOKIE_PARAMS)

    return RefreshResponse()

Error Responses:

• 401 — No cookie, expired token, tampered token, or reuse detected

---

8.3 POST /api/v1/auth/logout

Purpose: Clear auth cookies and invalidate refresh token in DB.

Request: No body. User must be authenticated.

Response (200):

class LogoutResponse(BaseModel):
    message: str = "Logged out successfully"

Router Implementation:

@router.post("/logout", response_model=LogoutResponse)
async def logout(
    response: Response,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    # Invalidate refresh token in DB
    current_user.refresh_token_hash = None
    await db.commit()

    # Clear both cookies
    response.delete_cookie("access_token", path="/")
    response.delete_cookie("refresh_token", path="/api/v1/auth")

    await write_audit_log(db, actor_id=current_user.id, action="auth.logout", resource_type="user", resource_id=str(current_user.id))

    return LogoutResponse()

---

8.4 GET /api/v1/auth/me

Purpose: Return the current user's profile. Called by the frontend on app load to restore session after page refresh.

Request: No body. Reads access_token cookie.

Response (200): UserPublic schema (same as login response)

Router Implementation:

@router.get("/me", response_model=UserPublic)
async def get_me(current_user: User = Depends(get_current_user)):
    return UserPublic.model_validate(current_user)

Why this matters for the frontend: Currently, the frontend has no session persistence — user state is lost on page refresh. Once GET /auth/me is implemented, AuthContext can call it on mount and restore the user from the cookie automatically.

---

9. Role → Post-Login Redirect Map

The frontend (Login.jsx) redirects to a role-specific page after successful login. The backend LoginResponse includes role so the frontend can make this decision. This table is the authoritative mapping:

Role	Post-login Redirect
CUSTOMER	/portal/profile
OWNER	/owner/snapshot
ADMIN	/emp/fa/dashboard
FIELD_AGENT	/emp/fa/dashboard
CONTRACTOR	/contractor/dashboard
VENDOR	/vendor/dashboard
SUBCONTRACTOR	/subcontractor/dashboard

---

10. Permission System

The frontend's src/utils/permissions.js defines a PERMISSIONS object and a ROLE_PERMISSIONS map. The backend must enforce the same rules at the API layer. The canonical permission matrix:

Permission	OWNER	ADMIN	CONTRACTOR	SUBCONTRACTOR	FIELD_AGENT	VENDOR	CUSTOMER
VIEW_ALL_PROJECTS	✅	✅	❌	❌	❌	❌	❌
VIEW_ALL_PERSONNEL	✅	✅	❌	❌	❌	❌	❌
VIEW_FINANCIALS	✅	✅	❌	❌	❌	❌	❌
VIEW_SENSITIVE_DATA	✅	❌	❌	❌	❌	❌	❌
MANAGE_USERS	✅	✅	❌	❌	❌	❌	❌
APPROVE_DOCUMENTS	✅	✅	❌	❌	❌	❌	❌
VIEW_ASSIGNED_PROJECTS	✅	✅	✅	✅	❌	❌	❌
UPLOAD_DOCUMENTS	✅	✅	✅	❌	❌	✅	❌
SUBMIT_INVOICES	✅	✅	✅	❌	❌	✅	❌
MANAGE_OWN_CREW	❌	❌	✅	❌	❌	❌	❌
VIEW_ASSIGNED_TASKS	✅	✅	✅	✅	❌	❌	❌
UPDATE_TASK_STATUS	✅	✅	✅	✅	❌	❌	❌

Implement as a utility used inside service functions (not in routers — routers only check role, services check fine-grained permissions):

# app/utils/permissions.py
ROLE_PERMISSIONS: dict[str, list[str]] = {
    "OWNER": [
        "VIEW_ALL_PROJECTS", "VIEW_ALL_PERSONNEL", "VIEW_FINANCIALS",
        "VIEW_SENSITIVE_DATA", "MANAGE_USERS", "APPROVE_DOCUMENTS",
        "VIEW_ASSIGNED_PROJECTS", "UPLOAD_DOCUMENTS", "SUBMIT_INVOICES",
        "VIEW_ASSIGNED_TASKS", "UPDATE_TASK_STATUS",
    ],
    "ADMIN": [
        "VIEW_ALL_PROJECTS", "VIEW_ALL_PERSONNEL", "VIEW_FINANCIALS",
        "MANAGE_USERS", "APPROVE_DOCUMENTS", "VIEW_ASSIGNED_PROJECTS",
        "UPLOAD_DOCUMENTS", "SUBMIT_INVOICES", "VIEW_ASSIGNED_TASKS", "UPDATE_TASK_STATUS",
    ],
    "CONTRACTOR": [
        "VIEW_ASSIGNED_PROJECTS", "UPLOAD_DOCUMENTS",
        "SUBMIT_INVOICES", "MANAGE_OWN_CREW", "VIEW_ASSIGNED_TASKS", "UPDATE_TASK_STATUS",
    ],
    "SUBCONTRACTOR": ["VIEW_ASSIGNED_TASKS", "UPDATE_TASK_STATUS"],
    "FIELD_AGENT": [],
    "VENDOR": ["UPLOAD_DOCUMENTS", "SUBMIT_INVOICES"],
    "CUSTOMER": [],
}

def has_permission(role: str, permission: str) -> bool:
    return permission in ROLE_PERMISSIONS.get(role, [])

---

11. Sensitive Data Masking

VIEW_SENSITIVE_DATA is OWNER-only. Any endpoint that returns sensitive fields (SSN, EIN, bank account numbers) must mask them for all other roles.

# app/utils/mask.py

def mask_ssn(ssn: str | None) -> str:
    """'123-45-6789' → '***-**-6789'"""
    if not ssn:
        return ""
    return f"***-**-{ssn[-4:]}"

def mask_bank_account(account: str | None) -> str:
    """'9876543210' → '******3210'"""
    if not account:
        return ""
    return f"{'*' * (len(account) - 4)}{account[-4:]}"

def mask_ein(ein: str | None) -> str:
    """'12-3456789' → '**-***6789'"""
    if not ein:
        return ""
    return f"**-***{ein[-4:]}"

---

12. Seed Credentials

All 17 mock users are seeded with password = "password" (bcrypt-hashed). Use these during development and integration testing:

Role	Login Type (UI)	Identifier	Portal
CUSTOMER	customer	alice	/portal/profile
FIELD_AGENT	employee	FA001	/emp/fa/dashboard
FIELD_AGENT	employee	FA002	/emp/fa/dashboard
FIELD_AGENT	employee	FA003	/emp/fa/dashboard
FIELD_AGENT	employee	FA004	/emp/fa/dashboard
FIELD_AGENT	employee	FA005	/emp/fa/dashboard
ADMIN	employee	ADM01	/emp/fa/dashboard
ADMIN	employee	ADM02	/emp/fa/dashboard
ADMIN	employee	ADM03	/emp/fa/dashboard
OWNER	owner	justin	/owner/snapshot
OWNER	owner	diana	/owner/snapshot
CONTRACTOR	contractor	mike	/contractor/dashboard
SUBCONTRACTOR	subcontractor	carlos	/subcontractor/dashboard
VENDOR	vendor	abc_supply	/vendor/dashboard

---

13. Security Checklist

Before shipping auth to production:

• SECRET_KEY is generated with openssl rand -hex 32 — not a guessable string
• secure=True on cookies (HTTPS only)
• samesite="strict" on cookies
• Refresh token path scoped to /api/v1/auth
• Login endpoint has rate limiting (e.g. 10 attempts / minute per IP)
• Failed login attempts do NOT reveal which field was wrong
• Refresh token rotation implemented (reuse = revoke all)
• Logout clears both cookies AND invalidates DB token hash
• GET /auth/me is the only session-restore mechanism (no localStorage token storage)
• Audit log entry written for every login and logout

---

Next: Read 03_properties_module.md or 04_users_people_module.md — both depend on require_role() defined here.