# Auth Session Pattern (Refresh Token Rotation)

## Overview

A Clean Architecture pattern for mobile-safe authentication sessions with refresh token rotation, SHA-256 hashed storage, and backward-compatible wiring.

## Data Model

### `user_sessions` Table

| Column | Type | Notes |
|--------|------|-------|
| `id` | int (PK, autoincrement) | |
| `user_id` | int (FK → users.id, CASCADE) | Indexed |
| `refresh_token_hash` | str(255) | SHA-256 of refresh token |
| `device_name` | str(255) | Nullable, for session display |
| `user_agent` | str(512) | Nullable |
| `ip_address` | str(45) | Nullable (supports IPv6) |
| `expires_at` | datetime(tz) | From config `REFRESH_TOKEN_EXPIRE_DAYS` |
| `revoked_at` | datetime(tz) | Nullable — set on rotation/logout |
| `created_at` | datetime(tz) | |
| `updated_at` | datetime(tz) | |

## Layer Implementation

### Domain (`domain/entities.py`)

```python
@dataclass(slots=True)
class SessionEntity:
    id: int
    user_id: int
    refresh_token_hash: str
    device_name: str | None
    user_agent: str | None
    ip_address: str | None
    expires_at: datetime
    revoked_at: datetime | None
    created_at: datetime
    updated_at: datetime
```

### Application Port (`application/interfaces.py`)

```python
class SessionRepository(Protocol):
    async def create(self, *, user_id, refresh_token_hash, device_name,
                     user_agent, ip_address, expires_at) -> SessionEntity: ...
    async def get_by_refresh_hash(self, token_hash: str) -> SessionEntity | None: ...
    async def get_valid_session(self, user_id: int, session_id: int) -> SessionEntity | None: ...
    async def revoke(self, session_id: int) -> None: ...
    async def revoke_all_for_user(self, user_id: int, except_session_id: int | None = None) -> None: ...
    async def get_user_sessions(self, user_id: int) -> list[SessionEntity]: ...
```

### SQLAlchemy Model (`infrastructure/models.py`)

```python
class UserSessionModel(Base):
    __tablename__ = "user_sessions"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
    refresh_token_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    device_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
    user_agent: Mapped[str | None] = mapped_column(String(512), nullable=True)
    ip_address: Mapped[str | None] = mapped_column(String(45), nullable=True)
    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

    user: Mapped["UserModel"] = relationship("UserModel", back_populates="sessions")
```

Add a relationship on `UserModel`:

```python
sessions: Mapped[list["UserSessionModel"]] = relationship(
    "UserSessionModel", back_populates="user", cascade="all, delete-orphan"
)
```

### Repository (`infrastructure/repositories.py`)

Key implementation details:

- **`get_by_refresh_hash`**: Select by `refresh_token_hash` column
- **`get_valid_session`**: AND query: `id == x AND user_id == y AND revoked_at IS NULL AND expires_at > now`
- **`revoke`**: Set `revoked_at = datetime.now(UTC)` on the model
- **`revoke_all_for_user`**: Bulk `update()` with `.where(user_id == x, revoked_at.is_(None))`, optionally except one session
- **`get_user_sessions`**: Select with AND `revoked_at IS NULL` and `expires_at > now`, ordered by `created_at DESC`

### Config (`core/config.py`)

```python
REFRESH_TOKEN_EXPIRE_DAYS: int = 30
```

### Security (`core/security.py`)

```python
def create_refresh_token(subject: str | Any, expires_delta: timedelta | None = None) -> str:
    """JWT with typ='refresh', longer expiry than access token."""
    to_encode = {
        "sub": str(subject),
        "typ": "refresh",
        "iss": settings.JWT_ISSUER,
        "aud": settings.JWT_AUDIENCE,
        "iat": now,
        "nbf": now,
        "exp": now + (expires_delta or timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)),
        "jti": str(uuid4()),
    }
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

def decode_refresh_token(token: str) -> dict[str, Any]:
    """Decode and verify typ='refresh'. Rejects access tokens used as refresh."""
    payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"],
                         issuer=settings.JWT_ISSUER, audience=settings.JWT_AUDIENCE,
                         options={"require": ["sub", "exp", "iat", "nbf", "iss", "aud", "typ", "jti"]})
    if payload.get("typ") != "refresh":
        raise InvalidTokenError("Invalid token type")
    return payload
```

### Token Hashing (`application/use_cases.py`)

```python
import hashlib

def _hash_token(token: str) -> str:
    """SHA-256 hash. Never store raw refresh tokens in the DB."""
    return hashlib.sha256(token.encode()).hexdigest()
```

## Use Case Flow

### Login (with session)

```python
async def login(self, data: UserLoginEntity) -> dict:
    user = await self.user_repo.get_by_email(data.email)
    # ... validate password, check is_active ...

    access_token = self.token_service.create_access_token(subject=str(user.id))

    result = {"access_token": access_token, "token_type": "bearer", "user": {...}}

    if self.session_repo is not None:     # ← Optional for backward compat
        refresh_token = create_refresh_token(subject=str(user.id))
        expires_at = datetime.now(UTC) + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
        await self.session_repo.create(
            user_id=user.id,
            refresh_token_hash=_hash_token(refresh_token),
            device_name=None, user_agent=None, ip_address=None,
            expires_at=expires_at,
        )
        result["refresh_token"] = refresh_token

    return result
```

### Refresh (rotation)

```python
async def refresh_token(self, refresh_token: str) -> dict:
    payload = decode_refresh_token(refresh_token)       # Verify JWT
    user_id = int(payload["sub"])
    token_hash = _hash_token(refresh_token)
    session = await self.session_repo.get_by_refresh_hash(token_hash)

    # Guard: session exists, not revoked, user active
    if session is None: raise ValueError("Session not found")
    if session.revoked_at is not None: raise ValueError("Session revoked")
    user = await self.user_repo.get_by_id(user_id)
    if user is None or not user.is_active: raise ValueError("Account deactivated")

    await self.session_repo.revoke(session.id)           # Rotate: revoke old

    # Issue new tokens + create new session
    new_access = create_access_token(subject=str(user_id))
    new_refresh = create_refresh_token(subject=str(user_id))
    await self.session_repo.create(
        user_id=user_id,
        refresh_token_hash=_hash_token(new_refresh),
        device_name=session.device_name,   # Preserve device metadata
        user_agent=session.user_agent,
        ip_address=session.ip_address,
        expires_at=...,                     # New 30-day expiry
    )
    return {"access_token": new_access, "refresh_token": new_refresh}
```

### Logout / Session Management

```python
async def logout(self, user_id: int) -> None:
    """Revoke ALL sessions for the user."""
    await self.session_repo.revoke_all_for_user(user_id)

async def revoke_session(self, user_id: int, session_id: int) -> None:
    """Revoke one specific session (ownership check required)."""
    session = await self.session_repo.get_valid_session(user_id, session_id)
    if session is None:
        raise ValueError("Session not found")
    await self.session_repo.revoke(session_id)
```

## Backward Compatibility

The session repo is **optional** — `session_repo: SessionRepository | None = None` in the use case constructor. Without it:
- Login returns `access_token` only (no `refresh_token`)
- `refresh_token()` raises `ValueError("Session management not configured")`
- `logout()` is a no-op
- `get_sessions()` returns `[]`

This means existing code that constructs `UserUseCases(repo, hasher, token)` continues working without changes.

## Endpoints

| Method | Path | Auth | Response |
|--------|------|------|----------|
| POST | `/auth/refresh` | None (body: `refresh_token`) | 200: `{access_token, refresh_token}` |
| POST | `/auth/logout` | Bearer access | 204 No Content |
| GET | `/auth/sessions` | Bearer access | 200: `[{id, device_name, created_at, expires_at}]` |
| DELETE | `/auth/sessions/{id}` | Bearer access | 204 No Content |

**Login response delta:** Login now returns `refresh_token` in addition to `access_token` and `user`. The response model is `LoginResponse` (extends old `TokenResponse`). Old clients that ignore the new field continue working.

## Pitfalls

### Pitfall: Stale JWT after user deactivation
The refresh token can still be valid (JWT expiry) even after the user is deactivated. Always check `user.is_active` during refresh, not just JWT validity.

### Pitfall: Session repo as None triggers confusing errors
If `session_repo` is None and code calls `refresh_token()`, the error is `"Session management not configured"`. Document this in the API error responses.

### Pitfall: `get_valid_session` must check user_id
Always scope session lookups to the requesting user. `get_valid_session(user_id, session_id)` prevents session-ID enumeration across users.

### Pitfall: Rate-limiting refresh
Refresh endpoints should be rate-limited differently than login — a compromised refresh token could be brute-forced if not protected.