# Phase 07.3: Security — ES256 Algorithm Upgrade - Pattern Map

**Mapped:** 2026-06-05
**Files analyzed:** 7 new/modified files
**Analogs found:** 7 / 7

---

## File Classification

| New/Modified File | Role | Data Flow | Closest Analog | Match Quality |
|-------------------|------|-----------|----------------|---------------|
| `backend/services/auth.py` | service | request-response | self (4 signing sites + TTL param) | exact — modify in-place |
| `backend/config.py` | config | — | self (add 3 fields) | exact — modify in-place |
| `backend/main.py` | config/startup | event-driven | `backend/services/ai_config.py` (`seed_system_settings_from_env`) | exact — same startup-hook + upsert pattern |
| `backend/api/auth.py` | controller | request-response | self (`LoginRequest` + `_set_refresh_cookie`) | exact — modify in-place |
| `frontend/src/views/auth/LoginView.vue` | component | request-response | self (step-based login form) | exact — modify in-place |
| `frontend/src/stores/auth.js` | store | request-response | self (`login()` action) | exact — modify in-place |
| `backend/tests/test_auth_es256.py` | test | — | `backend/tests/test_task2_auth_service.py` + `backend/tests/test_auth_api.py` | role-match — same unit + integration pattern |

---

## Pattern Assignments

### `backend/services/auth.py` — ES256 signing sites + TTL param

**Analog:** self — 4 sites to change in-place

**Current imports block** (lines 18–38):
```python
from __future__ import annotations

import hashlib
import hmac
import logging
import re
import secrets
import uuid
from datetime import datetime, timezone, timedelta
from typing import Optional

import httpx
import jwt
import pyotp
from pwdlib import PasswordHash
from pwdlib.hashers.argon2 import Argon2Hasher
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

from config import settings
from db.models import BackupCode, Quota, RefreshToken, User
```
Add `import base64` to this block (no new packages).

**Site 1 — `create_access_token`** (lines 86–99, current HS256):
```python
def create_access_token(user_id: str, role: str) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": str(user_id),
        "role": role,
        "typ": "access",
        "iat": now,
        "exp": now + timedelta(minutes=settings.access_token_expire_minutes),
    }
    return jwt.encode(payload, settings.secret_key, algorithm="HS256")
```
Change last line to:
```python
    private_pem = base64.b64decode(settings.jwt_private_key).decode()
    return jwt.encode(payload, private_pem, algorithm="ES256")
```

**Site 2 — `decode_access_token`** (lines 102–117, current HS256):
```python
def decode_access_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
    except jwt.ExpiredSignatureError as exc:
        raise ValueError("Token has expired") from exc
    except jwt.PyJWTError as exc:
        raise ValueError(f"Invalid token: {exc}") from exc

    if payload.get("typ") != "access":
        raise ValueError("Token type mismatch: expected 'access'")
    return payload
```
Change the `jwt.decode` call to:
```python
        public_pem = base64.b64decode(settings.jwt_public_key).decode()
        payload = jwt.decode(token, public_pem, algorithms=["ES256"])
```

**Site 3 — `create_password_reset_token`** (lines 120–132):
```python
    return jwt.encode(payload, settings.secret_key, algorithm="HS256")
```
Change to:
```python
    private_pem = base64.b64decode(settings.jwt_private_key).decode()
    return jwt.encode(payload, private_pem, algorithm="ES256")
```

**Site 4 — `decode_password_reset_token`** (lines 135–149):
```python
        payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
```
Change to:
```python
        public_pem = base64.b64decode(settings.jwt_public_key).decode()
        payload = jwt.decode(token, public_pem, algorithms=["ES256"])
```

**`create_refresh_token` TTL param** (lines 154–172 — current signature):
```python
async def create_refresh_token(session: AsyncSession, user_id: uuid.UUID) -> str:
    raw = secrets.token_urlsafe(32)
    token_hash = hashlib.sha256(raw.encode()).hexdigest()
    now = datetime.now(timezone.utc)
    row = RefreshToken(
        id=uuid.uuid4(),
        user_id=user_id,
        token_hash=token_hash,
        expires_at=now + timedelta(days=settings.refresh_token_expire_days),
        revoked=False,
    )
    session.add(row)
    await session.commit()
    return raw
```
Add `remember_me: bool = False` param; change TTL line to:
```python
    ttl = (
        timedelta(days=settings.refresh_token_expire_days)
        if remember_me
        else timedelta(hours=settings.refresh_token_expire_hours)
    )
    ...
    expires_at=now + ttl,
```

---

### `backend/config.py` — add 3 new settings

**Analog:** self — add after `refresh_token_expire_days` at line 35

**Current JWT block** (lines 33–35):
```python
    # Auth / JWT (Phase 2)
    access_token_expire_minutes: int = 15
    refresh_token_expire_days: int = 30
```
Extend to:
```python
    # Auth / JWT (Phase 2)
    access_token_expire_minutes: int = 15
    refresh_token_expire_days: int = 30
    refresh_token_expire_hours: int = 16          # Phase 7.3: default short session

    # JWT key pair (Phase 7.3 — ES256; both required in production)
    # Values are base64-encoded PEM strings (single-line, shell-safe).
    jwt_private_key: str = ""   # JWT_PRIVATE_KEY env var — required
    jwt_public_key: str = ""    # JWT_PUBLIC_KEY  env var — required
```
No methods on Settings. Callers decode inline with `base64.b64decode(settings.jwt_private_key).decode()` — consistent with how other base64 secrets are handled in this codebase (e.g., HKDF in `ai_config.py` decodes `cloud_creds_key` inline).

---

### `backend/main.py` — lifespan ES256 rotation check

**Primary analog:** `backend/services/ai_config.py:seed_system_settings_from_env` (lines 201–244) — exact pattern for reading and writing a `SystemSettings` key-value row at startup.

**Secondary analog:** existing lifespan try/except block (lines 172–180) — exact wrapping pattern for skipping when the table doesn't exist yet.

**Existing startup try/except block to copy** (lines 172–180):
```python
    try:
        async with AsyncSessionLocal() as session:
            await seed_system_settings_from_env(session)
            await session.commit()
    except Exception as _seed_exc:
        import logging as _logging
        _logging.getLogger(__name__).warning(
            "AI provider seed skipped (table may not exist yet): %s", _seed_exc
        )
```

**Pattern to follow for the new rotation block** — slot in after the AI seed block (after line 180, before `yield`):
```python
    # ES256 startup rotation (Phase 7.3 — D-04):
    # If jwt_algorithm in system_settings differs from "ES256" (or is absent),
    # bulk-revoke all refresh tokens and record the new algorithm.
    # Wrapped in try/except so a missing table before migrations doesn't crash.
    try:
        async with AsyncSessionLocal() as session:
            await _rotate_tokens_on_algorithm_change(session)
    except Exception as _es256_exc:
        import logging as _logging
        _logging.getLogger(__name__).warning(
            "ES256 rotation check skipped (table may not exist yet): %s", _es256_exc
        )
```

**`_rotate_tokens_on_algorithm_change` helper** — define as a module-level async function above `lifespan`, using the `seed_system_settings_from_env` upsert pattern as the direct template:

```python
# seed_system_settings_from_env upsert pattern (ai_config.py lines 219–244):
stmt = select(SystemSettings).where(SystemSettings.provider_id == provider_id)
result = await session.execute(stmt)
existing = result.scalar_one_or_none()
if existing is not None:
    return
row = SystemSettings(
    provider_id=provider_id,
    model_name=model_name,
    context_chars=context_chars,
    is_active=True,
    api_key_enc=None,
    base_url=None,
)
session.add(row)
```

For the rotation helper, use `provider_id="jwt_algorithm"`, `model_name="ES256"`, `context_chars=0`, `is_active=False` (D-03 anti-pattern: never `is_active=True` for metadata rows). Add `text()` bulk UPDATE before the upsert when the algorithm differs:

```python
from sqlalchemy import text, select
from db.models import SystemSettings, RefreshToken

async def _rotate_tokens_on_algorithm_change(session) -> None:
    """Idempotent ES256 migration. Bulk-revokes tokens if jwt_algorithm changed."""
    result = await session.execute(
        select(SystemSettings).where(SystemSettings.provider_id == "jwt_algorithm")
    )
    row = result.scalar_one_or_none()
    if row is not None and row.model_name == "ES256":
        return  # Already migrated
    # Bulk-revoke all active refresh tokens (single SQL statement, no Python iteration)
    await session.execute(
        text("UPDATE refresh_tokens SET revoked = true WHERE revoked = false")
    )
    if row is None:
        import uuid as _uuid
        session.add(SystemSettings(
            id=_uuid.uuid4(),
            provider_id="jwt_algorithm",
            model_name="ES256",
            context_chars=0,
            is_active=False,
            api_key_enc=None,
            base_url=None,
        ))
    else:
        row.model_name = "ES256"
    await session.commit()
```

**Required import addition at top of `main.py`** (alongside existing `from sqlalchemy import text`):
```python
from sqlalchemy import text, select   # text already imported; add select
```

---

### `backend/api/auth.py` — LoginRequest + `_set_refresh_cookie`

**Analog:** self — modify `LoginRequest` at lines 56–60 and `_set_refresh_cookie` at lines 70–80; thread `remember_me` through the login handler at line 278–279.

**Current `LoginRequest`** (lines 56–60):
```python
class LoginRequest(BaseModel):
    email: EmailStr
    password: str
    totp_code: Optional[str] = None
    backup_code: Optional[str] = None
```
Add one field:
```python
    remember_me: bool = False
```

**Current `_set_refresh_cookie`** (lines 70–80):
```python
def _set_refresh_cookie(response: Response, raw_token: str) -> None:
    """Set the httpOnly Secure SameSite=Strict refresh cookie (CLAUDE.md constraint)."""
    response.set_cookie(
        key="refresh_token",
        value=raw_token,
        httponly=True,
        secure=True,
        samesite="strict",
        path="/api/auth/refresh",
        max_age=settings.refresh_token_expire_days * 86400,
    )
```
Add `remember_me: bool = False` param; change `max_age` to:
```python
        max_age=(
            settings.refresh_token_expire_days * 86400
            if remember_me
            else settings.refresh_token_expire_hours * 3600
        ),
```

**Login handler token-issue block** (lines 277–279):
```python
    access_token = auth_service.create_access_token(str(user.id), user.role)
    raw_refresh = await auth_service.create_refresh_token(session, user.id)
    _set_refresh_cookie(response, raw_refresh)
```
Pass `remember_me` through:
```python
    access_token = auth_service.create_access_token(str(user.id), user.role)
    raw_refresh = await auth_service.create_refresh_token(session, user.id, remember_me=body.remember_me)
    _set_refresh_cookie(response, raw_refresh, remember_me=body.remember_me)
```

---

### `frontend/src/views/auth/LoginView.vue` — "Stay signed in" checkbox

**Analog:** self — the password step form (lines 1–59 of template); the `submitPassword` / `submitTotp` / `submitBackupCode` functions (lines 224–265).

**Current reactive state** (lines 188–193):
```javascript
const email = ref('')
const password = ref('')
const totpInput = ref('')
const backupCodeInput = ref('')
const loading = ref(false)
const error = ref(null)
```
Add one ref:
```javascript
const rememberMe = ref(false)
```

**Checkbox markup** — insert inside the password step form (`<form @submit.prevent="submitPassword">`), between the password `<div>` block and the error `<div>`. Copy existing input style (Tailwind classes) from the form:
```html
<div class="flex items-center gap-2">
  <input
    v-model="rememberMe"
    id="remember-me"
    type="checkbox"
    class="rounded border-gray-300 text-indigo-600 focus:ring-indigo-500"
  />
  <label for="remember-me" class="text-sm text-gray-600">
    Stay signed in for 30 days
  </label>
</div>
```

**`submitPassword` update** (lines 224–235):
```javascript
async function submitPassword() {
  loading.value = true
  error.value = null
  try {
    const result = await authStore.login(email.value, password.value)
    await handleLoginResult(result)
  } catch (e) {
    error.value = e.message
  } finally {
    loading.value = false
  }
}
```
Change the `authStore.login` call to pass `rememberMe`:
```javascript
    const result = await authStore.login(email.value, password.value, {
      rememberMe: rememberMe.value,
    })
```

**`submitTotp` and `submitBackupCode`** — these already pass `options` objects (lines 241–264). Add `rememberMe: rememberMe.value` to each call's options object, alongside the existing `totpCode` / `backupCode` keys. The checkbox state persists across steps since it is a single `ref` scoped to the component.

---

### `frontend/src/stores/auth.js` — `login()` action

**Analog:** self — `login()` at lines 60–88.

**Current `login()` action** (lines 60–88):
```javascript
async function login(email, password, options = {}) {
  loading.value = true
  error.value = null
  try {
    const data = await api.login({
      email,
      password,
      totp_code: options.totpCode ?? null,
      backup_code: options.backupCode ?? null,
    })
    // ... response handling unchanged
  }
}
```
Add `remember_me` to the API call body:
```javascript
      remember_me: options.rememberMe ?? false,
```
Insert after `backup_code`. The rest of `login()` is unchanged.

---

### `backend/tests/test_auth_es256.py` — NEW test file

**Analog:** `backend/tests/test_task2_auth_service.py` (unit pattern) + `backend/tests/test_auth_api.py` (integration + FakeRedis pattern)

**File header pattern** (from `test_task2_auth_service.py` lines 1–7):
```python
"""
TDD tests for Phase 7.3: ES256 algorithm upgrade, startup token rotation,
and remember_me session TTL.
"""
import pytest
import pytest_asyncio
```

**Unit test structure** (from `test_task2_auth_service.py` lines 28–57 — no async, no DB):
```python
def test_access_token_uses_es256():
    from services.auth import create_access_token
    t = create_access_token("test-uid", "user")
    # Header is base64url({"alg":"ES256",...}) — check alg claim
    import base64, json
    header = json.loads(base64.urlsafe_b64decode(t.split(".")[0] + "=="))
    assert header["alg"] == "ES256"
```

**xfail stub pattern** — for tests needing DB fixtures not yet wired, use `pytest.mark.xfail(strict=True, reason="...")`. Follow the existing xfail convention in the test suite (used in Phase 7.2 test stubs). Stubs that need `db_session` use the `async def` + `pytest_asyncio.fixture` pattern from `conftest.py`:
```python
@pytest.mark.xfail(strict=True, reason="ES256-04: not yet implemented")
@pytest.mark.asyncio
async def test_startup_rotation_revokes_tokens(db_session):
    pytest.xfail("not yet implemented")
```

**FakeRedis pattern** — for integration tests hitting the login endpoint, import and apply the `FakeRedis` from `test_auth_api.py`:
```python
from test_auth_api import FakeRedis, _register, _login
```
Or copy the minimal `FakeRedis` class and `_login` helper into the new file — the `test_auth_api.py` pattern is lines 47–97 of that file.

**Key env var fixture** — because ES256 needs real key material in `settings`, add a module-level fixture that patches settings before the module runs:
```python
import base64
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization

@pytest.fixture(autouse=True)
def es256_keys(monkeypatch):
    """Patch settings with a freshly generated P-256 key pair for each test."""
    k = ec.generate_private_key(ec.SECP256R1())
    priv = base64.b64encode(
        k.private_bytes(serialization.Encoding.PEM,
                        serialization.PrivateFormat.PKCS8,
                        serialization.NoEncryption())
    ).decode()
    pub = base64.b64encode(
        k.public_key().public_bytes(serialization.Encoding.PEM,
                                    serialization.PublicFormat.SubjectPublicKeyInfo)
    ).decode()
    monkeypatch.setattr("config.settings.jwt_private_key", priv)
    monkeypatch.setattr("config.settings.jwt_public_key", pub)
```

**`conftest.py` `auth_user` fixture** (lines 207–245) — the integration tests that assert on refresh-token TTL need a `db_session` + `async_client`. Reuse these fixtures from `conftest.py` directly (they are autoimported by pytest from the package-level conftest).

---

## Shared Patterns

### Base64-decoded secret inline at call site
**Source:** `backend/services/ai_config.py` lines 63–71 (HKDF derive pattern)
**Apply to:** All 4 JWT signing/decoding sites in `services/auth.py`
```python
# Pattern: decode inline, never cache as module-level global
private_pem = base64.b64decode(settings.jwt_private_key).decode()
public_pem  = base64.b64decode(settings.jwt_public_key).decode()
```

### Startup idempotency check (SystemSettings read + upsert)
**Source:** `backend/services/ai_config.py:seed_system_settings_from_env` lines 219–244
**Apply to:** `_rotate_tokens_on_algorithm_change` in `main.py`
```python
stmt = select(SystemSettings).where(SystemSettings.provider_id == "<key>")
result = await session.execute(stmt)
existing = result.scalar_one_or_none()
if existing is not None:
    return  # idempotent — skip
# ... insert or update
```

### Startup try/except wrapping for missing table
**Source:** `backend/main.py` lines 172–180
**Apply to:** ES256 rotation block in `main.py`
```python
try:
    async with AsyncSessionLocal() as session:
        await <startup_fn>(session)
except Exception as _exc:
    import logging as _logging
    _logging.getLogger(__name__).warning(
        "<task> skipped (table may not exist yet): %s", _exc
    )
```

### httpOnly cookie set pattern
**Source:** `backend/api/auth.py:_set_refresh_cookie` lines 70–80
**Apply to:** modified `_set_refresh_cookie` signature (add `remember_me` param)
```python
response.set_cookie(
    key="refresh_token",
    value=raw_token,
    httponly=True,
    secure=True,
    samesite="strict",
    path="/api/auth/refresh",
    max_age=<computed>,
)
```

### Vue 3 Options API checkbox (script setup)
**Source:** `frontend/src/views/auth/LoginView.vue` lines 177–193 (existing ref pattern)
**Apply to:** new `rememberMe` ref + checkbox in `LoginView.vue`
```javascript
// Existing pattern: all form state is a ref()
const rememberMe = ref(false)

// Existing options object pattern (submitTotp, line 241):
const result = await authStore.login(email.value, password.value, {
  totpCode: totpInput.value,
})
// Extend with:
  rememberMe: rememberMe.value,
```

### Bulk SQL update (no Python-layer iteration)
**Source:** `backend/services/auth.py:revoke_all_refresh_tokens` provides context; the **preferred pattern for bulk updates** in this codebase is a single `session.execute(text(...))` as documented in CONTEXT.md §Specific Ideas
**Apply to:** `_rotate_tokens_on_algorithm_change` bulk revoke
```python
await session.execute(
    text("UPDATE refresh_tokens SET revoked = true WHERE revoked = false")
)
```

---

## No Analog Found

All files have close analogs. No gaps.

---

## Metadata

**Analog search scope:** `backend/services/`, `backend/api/`, `backend/config.py`, `backend/main.py`, `frontend/src/stores/`, `frontend/src/views/auth/`, `backend/tests/`
**Files read:** 12 source files + 2 planning documents
**Pattern extraction date:** 2026-06-05