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.

```python
# 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:**
```json
{
  "sub": "3f8a1c2d-...(user UUID)",
  "role": "FIELD_AGENT",
  "legacy_id": "e1",
  "type": "access",
  "iat": 1740400000,
  "exp": 1740400900
}
```

**Refresh token:**
```json
{
  "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

```python
# 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`

```python
# 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`.

```python
# 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.

```python
# 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:**
```python
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

```python
# 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:**
```python
# 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):**
```python
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:**
```python
# 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):**
```python
class RefreshResponse(BaseModel):
    message: str = "Token refreshed"
    # New access_token set as cookie; not in body for security
```

**Router Implementation:**
```python
@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):**
```python
class LogoutResponse(BaseModel):
    message: str = "Logged out successfully"
```

**Router Implementation:**
```python
@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:**
```python
@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):

```python
# 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.

```python
# 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.*