kite/.planning/phases/07.3-security-es256-algorithm-upgrade-inserted/07.3-PATTERNS.md

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):

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):

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:

    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):

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:

        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):

    return jwt.encode(payload, settings.secret_key, algorithm="HS256")

Change to:

    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):

        payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])

Change to:

        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):

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:

    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):

    # Auth / JWT (Phase 2)
    access_token_expire_minutes: int = 15
    refresh_token_expire_days: int = 30

Extend to:

    # 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):

    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):

    # 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:

# 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:

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):

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):

class LoginRequest(BaseModel):
    email: EmailStr
    password: str
    totp_code: Optional[str] = None
    backup_code: Optional[str] = None

Add one field:

    remember_me: bool = False

Current _set_refresh_cookie (lines 70–80):

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:

        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):

    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:

    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):

const email = ref('')
const password = ref('')
const totpInput = ref('')
const backupCodeInput = ref('')
const loading = ref(false)
const error = ref(null)

Add one ref:

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:

<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):

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:

    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):

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:

      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):

"""
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):

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:

@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:

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:

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

# 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

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

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)

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

// 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

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