kaarmic-auth-python 1.0.0

Official Python implementation of Kaarmic.Auth authentication library.

Kaarmic.Auth Python (kaarmic-auth)

Official Python implementation of the Kaarmic.Auth authentication and session management library for Flask applications.

---

Installation

pip install kaarmic-auth

---

Critical Notes (Read Before Integrating)

These are the things that won't cause an import error but will silently break your app.

1. You must serve a login page at login_url

KaarmicAuth(app, ..., login_url="/login") only tells the library where to redirect unauthenticated users. You must create that route yourself to render your login HTML. Without it, unauthenticated page visits loop forever with a 302.

@app.route("/login")
def login_page():
    return render_template("login.html")   # or return your login HTML

Also add "/login" to the public list in your enforce_auth hook (the skeleton below does this). If you change login_url to something else (e.g. "/auth/login"), update both places.

---

2. current_user() returns a string user ID, not a user object

uid = current_user()   # e.g. "64a1b2c3d4e5f6a7b8c9d0e1"  ← just a string

# To get user details, query your own DB:
user = db.users.find_one({"_id": ObjectId(uid)})

---

3. Token is read from cookie OR Authorization: Bearer <token> header

The library checks both automatically. Cookie-based (browser sessions) and header-based (API clients / mobile) both work out of the box. You do not need to manually extract or pass the token anywhere — current_context() / current_user() will be populated on every authenticated request regardless of how the token was sent.

---

4. AuthenticationContext.session_id is auto-generated

You do not need to provide session_id when creating an AuthenticationContext for login. The library generates a UUID internally.

# Correct — session_id is optional, library fills it in
ctx = AuthenticationContext(user_id=str(user["_id"]), metadata={"username": username})

# The returned token_pair will contain the generated session_id
token_pair = login_service.issue_tokens(ctx)

---

5. metadata dict is passthrough — retrieve it from current_context()

5. AuthenticationContext — declare your user data here, once, at login

AuthenticationContext is the dict you fill in at login time. Everything in it gets stored inside the token and is available on every subsequent request via current_context() — no DB query needed.

Where to declare it: inside your api_login route, after the password check.

from kaarmic_auth import AuthenticationContext

# Inside api_login(), after verifying the user:
ctx = AuthenticationContext(
    user_id  = str(user["_id"]),       # REQUIRED — string ID from your DB

    metadata = {                        # OPTIONAL — any key/value pairs you want in the token
        "username": user["username"],
        "role":     user.get("role", "user"),
        "email":    user.get("email"),
        # ... anything else you want available on every request without a DB call
    },

    scopes   = ["read", "write"],      # OPTIONAL — permission strings for your own authz logic

    tenant_id = "tenant-abc",          # OPTIONAL — for multi-tenant apps
    client_id = "web-client",          # OPTIONAL — for OAuth-style apps
)

token_pair = login_service.issue_tokens(ctx)

Where to read it back: on any authenticated request, call current_context():

@app.route("/api/profile")
def profile():
    ctx = current_context()

    ctx.user_id            # "64a1b2..."  — the primary key from your DB
    ctx.metadata["username"]   # "krish"
    ctx.metadata["role"]       # "admin"
    ctx.scopes             # ["read", "write"]
    ctx.tenant_id          # "tenant-abc"
    ctx.session_id         # auto-generated UUID — unique per login session

Rule of thumb: put data you need on every request (username, role, permissions) in metadata. For heavy or sensitive data (full profile, address), just store user_id and query your DB when needed.

---

Where does the code go?

Everything goes into your main app.py (or whichever file you start Flask from), in this exact order:

app.py
│
├── 1. Pick storage backend    ← copy ONE of Option A / B / C / D below
├── 2. Wire up services        ← copy Step 2 block (same for all options)
├── 3. Auth enforcement hooks  ← copy Step 3 block (same for all options)
├── 4. Login + logout routes   ← copy Step 4 block (same for all options)
└── 5. Your own routes         ← add @login_required where needed

Concrete example — if you choose Option B (in-memory + MongoDB), your app.py will look like:

from flask import Flask, request, redirect, jsonify, g
from pymongo import MongoClient
from kaarmic_auth import (
    TokenCryptoService, TokenValidationService, LoginService,
    AuthenticationContext, RevocationReason,
)
from kaarmic_auth.stores import (
    MemoryAccessTokenStore, MongoRefreshTokenStore, MongoUserSessionIndexStore,
)
from kaarmic_auth.flask_ext import KaarmicAuth, login_required, current_user, current_context

app = Flask(__name__)
db  = MongoClient("mongodb://localhost:27017/")["your_db_name"]

# ── [Step 1] Storage ──────────────────────────────────────────────────────────
access_store        = MemoryAccessTokenStore()
refresh_store       = MongoRefreshTokenStore(db.auth_refresh_tokens)
session_index_store = MongoUserSessionIndexStore(db.auth_session_index)

# ── [Step 2] Services ─────────────────────────────────────────────────────────
crypto_service     = TokenCryptoService()
validation_service = TokenValidationService(access_store, crypto_service)
login_service      = LoginService(
    access_token_store  = access_store,
    refresh_token_store = refresh_store,
    session_index_store = session_index_store,
    crypto_service      = crypto_service,
    default_access_minutes = 30.0,
    default_refresh_days   = 30.0,
)
kaarmic_auth = KaarmicAuth(app, validation_service, login_service, login_url="/login")

# ── [Step 3] Auth enforcement ─────────────────────────────────────────────────
@app.before_request
def enforce_auth():
    public = ["/login", "/api/auth/login", "/api/auth/logout"]
    if request.path in public or request.path.startswith("/static"):
        return
    if not current_context() or not current_context().is_valid:
        refresh_token = request.cookies.get("refresh_token")
        if refresh_token:
            try:
                new_pair = login_service.refresh_token(refresh_token)
                res = validation_service.validate_access_token(new_pair.access_token)
                if res.is_valid:
                    g.auth_context = res.context
                    g.rotated_tokens = new_pair
                    return
            except Exception:
                pass
        if request.path.startswith("/api/") or request.is_json:
            return jsonify({"error": "Unauthorized"}), 401
        return redirect("/login")

@app.after_request
def attach_rotated_cookies(response):
    rotated = getattr(g, "rotated_tokens", None)
    if rotated:
        response.set_cookie("access_token",  rotated.access_token,  httponly=True, samesite="Lax")
        response.set_cookie("refresh_token", rotated.refresh_token, httponly=True, samesite="Lax")
        response.set_cookie("user_id",       rotated.user_id,                      samesite="Lax")
    return response

# ── [Step 4] Login / logout routes ────────────────────────────────────────────
import hashlib

@app.route("/api/auth/login", methods=["POST"])
def api_login():
    data     = request.json or {}
    username = data.get("username", "").strip()
    password = data.get("password", "")
    user = db.users.find_one({"username": username})   # your own user lookup
    if not user or user["password_hash"] != hashlib.sha256(password.encode()).hexdigest():
        return jsonify({"error": "Invalid credentials"}), 401
    ctx = AuthenticationContext(user_id=str(user["_id"]), metadata={"username": username})
    token_pair = login_service.issue_tokens(ctx)
    resp = jsonify({"status": "success"})
    resp.set_cookie("access_token",  token_pair.access_token,  httponly=True, samesite="Lax")
    resp.set_cookie("refresh_token", token_pair.refresh_token, httponly=True, samesite="Lax")
    resp.set_cookie("user_id",       token_pair.user_id,                      samesite="Lax")
    return resp

@app.route("/api/auth/logout", methods=["POST"])
def api_logout():
    access_token  = request.cookies.get("access_token")
    refresh_token = request.cookies.get("refresh_token")
    if access_token:
        access_store.remove(crypto_service.compute_hash(access_token))
    if refresh_token:
        login_service.revoke_refresh_token(refresh_token, RevocationReason.LOGOUT)
    ctx = current_context()
    if ctx:
        login_service.revoke_session(ctx.user_id, ctx.session_id)
    resp = jsonify({"status": "logged out"})
    resp.delete_cookie("access_token",  path="/", samesite="Lax")
    resp.delete_cookie("refresh_token", path="/", samesite="Lax")
    resp.delete_cookie("user_id",       path="/", samesite="Lax")
    return resp

# ── [Step 5] Your own routes ──────────────────────────────────────────────────
@app.route("/dashboard")
@login_required
def dashboard():
    return f"Hello {current_user()}"

if __name__ == "__main__":
    app.run(debug=True)

Only Steps 1 and 4 need customisation:

• Step 1 — swap in your chosen storage backend (Options A/B/C/D below)
• Step 4 login — replace db.users.find_one(...) with your own user lookup and password check

Steps 2, 3, and the logout route are identical boilerplate regardless of your stack.

---

How it works

Kaarmic.Auth uses three storage tiers. Each is a separate interface you can implement with any backend:

Tier	What it stores	Lifetime	You must provide
Access Token Store	Short-lived session tokens for per-request validation	Minutes (e.g. 30 min)	In-memory (built-in) or Redis (custom)
Refresh Token Store	Long-lived tokens for silent session renewal	Days (e.g. 30 days)	MongoDB (built-in) or any DB (custom)
Session Index Store	Map of user_id → active sessions for O(1) revocation	Same as refresh	MongoDB (built-in), memory (built-in), or any DB (custom)

You choose a backend for each tier independently.

---

Step 1 — Pick your storage configuration

Option A: Everything in-memory (development / testing only)

No external dependencies. All data is lost on server restart.

import dataclasses
from kaarmic_auth.stores import MemoryAccessTokenStore, MemoryUserSessionIndexStore
from kaarmic_auth import IRefreshTokenStore
from kaarmic_auth.models import RefreshTokenEntry, RevocationReason

# A minimal in-memory refresh store for dev/test
class MemoryRefreshTokenStore(IRefreshTokenStore):
    def __init__(self):
        self._store = {}

    def store(self, entry: RefreshTokenEntry) -> None:
        self._store[entry.token_hash] = entry

    def get_by_hash(self, token_hash: str):
        return self._store.get(token_hash)

    def try_rotate(self, old_entry: RefreshTokenEntry, new_entry: RefreshTokenEntry) -> bool:
        if old_entry.token_hash not in self._store:
            return False
        self._store[old_entry.token_hash] = old_entry  # already marked is_revoked=True by caller
        self._store[new_entry.token_hash] = new_entry
        return True

    def revoke(self, token_hash: str, reason: RevocationReason) -> None:
        entry = self._store.get(token_hash)
        if entry:
            # dataclasses.replace() is the correct way to copy-and-update a dataclass
            self._store[token_hash] = dataclasses.replace(
                entry, is_revoked=True, revoked_reason=reason
            )

    def cleanup_expired(self, cutoff_utc) -> int:
        return 0

access_store        = MemoryAccessTokenStore()
refresh_store       = MemoryRefreshTokenStore()
session_index_store = MemoryUserSessionIndexStore()

---

Option B: In-memory access tokens + MongoDB refresh/session (recommended default)

Access tokens stay fast in RAM. Refresh tokens and sessions survive server restarts.

from pymongo import MongoClient
from kaarmic_auth.stores import (
    MemoryAccessTokenStore,
    MongoRefreshTokenStore,
    MongoUserSessionIndexStore,
)

db = MongoClient("mongodb://localhost:27017/")["your_db_name"]

access_store        = MemoryAccessTokenStore()
refresh_store       = MongoRefreshTokenStore(db.auth_refresh_tokens)
session_index_store = MongoUserSessionIndexStore(db.auth_session_index)

---

Option C: Redis access tokens + MongoDB refresh/session (production scale)

Redis handles access token TTL expiry natively. No manual cleanup needed.

import redis
import json
from datetime import datetime, timezone
from pymongo import MongoClient
from kaarmic_auth import IAccessTokenStore
from kaarmic_auth.models import AccessTokenEntry, AuthenticationContext
from kaarmic_auth.stores import MongoRefreshTokenStore, MongoUserSessionIndexStore

class RedisAccessTokenStore(IAccessTokenStore):
    def __init__(self, redis_client):
        self._r = redis_client

    def store(self, entry: AccessTokenEntry) -> None:
        expires_utc = entry.expires_utc
        if expires_utc.tzinfo is None:
            expires_utc = expires_utc.replace(tzinfo=timezone.utc)
        ttl = max(1, int((expires_utc - datetime.now(timezone.utc)).total_seconds()))
        payload = json.dumps({
            "user_id":    entry.context.user_id,
            "session_id": entry.context.session_id,
            "issued":     entry.issued_utc.isoformat(),
            "expires":    expires_utc.isoformat(),
        })
        self._r.setex(entry.token_hash, ttl, payload)

    def get_by_hash(self, token_hash: str):
        data = self._r.get(token_hash)
        if not data:
            return None
        d = json.loads(data)
        ctx     = AuthenticationContext(user_id=d["user_id"], session_id=d["session_id"])
        issued  = datetime.fromisoformat(d["issued"])
        expires = datetime.fromisoformat(d["expires"])
        return AccessTokenEntry(
            token_hash=token_hash,
            context=ctx,
            issued_utc=issued,
            expires_utc=expires,
            parameters={},
        )

    def remove(self, token_hash: str) -> None:
        self._r.delete(token_hash)   # O(1), instant

    def cleanup_expired(self, cutoff_utc) -> int:
        return 0   # Redis TTL handles this natively — no-op

db = MongoClient("mongodb://localhost:27017/")["your_db_name"]

access_store        = RedisAccessTokenStore(redis.Redis(host="localhost", port=6379))
refresh_store       = MongoRefreshTokenStore(db.auth_refresh_tokens)
session_index_store = MongoUserSessionIndexStore(db.auth_session_index)

---

Option D: Custom refresh token store (SQL, PostgreSQL, DynamoDB, etc.)

Implement IRefreshTokenStore with any backend:

from kaarmic_auth import IRefreshTokenStore
from kaarmic_auth.models import RefreshTokenEntry, RevocationReason, AuthenticationContext
from datetime import datetime, timezone

class PostgresRefreshTokenStore(IRefreshTokenStore):
    def __init__(self, db_connection):
        self._db = db_connection
        # Run once during DB setup:
        # CREATE TABLE refresh_tokens (
        #   token_hash TEXT PRIMARY KEY,
        #   user_id TEXT, session_id TEXT,
        #   expires_at TIMESTAMPTZ, absolute_expires_at TIMESTAMPTZ,
        #   is_revoked BOOLEAN DEFAULT false,
        #   revoked_reason INT DEFAULT 0,
        #   replaced_by TEXT,
        #   version INT DEFAULT 1
        # );

    def store(self, entry: RefreshTokenEntry) -> None:
        self._db.execute(
            """
            INSERT INTO refresh_tokens
                (token_hash, user_id, session_id, expires_at, absolute_expires_at,
                 is_revoked, revoked_reason, replaced_by, version)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (token_hash) DO UPDATE SET
                is_revoked=EXCLUDED.is_revoked,
                revoked_reason=EXCLUDED.revoked_reason,
                replaced_by=EXCLUDED.replaced_by,
                version=EXCLUDED.version
            """,
            (entry.token_hash, entry.context.user_id, entry.context.session_id,
             entry.refresh_expiry_utc, entry.absolute_expiry_utc,
             entry.is_revoked, int(entry.revoked_reason),
             entry.replaced_by_token_hash, entry.version)
        )

    def get_by_hash(self, token_hash: str):
        row = self._db.execute(
            "SELECT * FROM refresh_tokens WHERE token_hash = %s", (token_hash,)
        ).fetchone()
        if not row:
            return None
        ctx = AuthenticationContext(user_id=row["user_id"], session_id=row["session_id"])
        return RefreshTokenEntry(
            token_hash=row["token_hash"],
            context=ctx,
            created_utc=datetime.now(timezone.utc),
            refresh_expiry_utc=row["expires_at"],
            absolute_expiry_utc=row["absolute_expires_at"],
            version=row["version"],
            is_revoked=row["is_revoked"],
            revoked_reason=RevocationReason(row["revoked_reason"]),
            replaced_by_token_hash=row["replaced_by"],
        )

    def try_rotate(self, old_entry: RefreshTokenEntry, new_entry: RefreshTokenEntry) -> bool:
        # Optimistic concurrency: only mark rotated if version matches exactly
        updated = self._db.execute(
            """
            UPDATE refresh_tokens
            SET is_revoked=true, revoked_reason=%s, replaced_by=%s, version=%s
            WHERE token_hash=%s AND version=%s AND is_revoked=false
            """,
            (int(old_entry.revoked_reason), old_entry.replaced_by_token_hash,
             old_entry.version, old_entry.token_hash, old_entry.version - 1)
        ).rowcount
        if not updated:
            return False
        self.store(new_entry)
        return True

    def revoke(self, token_hash: str, reason: RevocationReason) -> None:
        self._db.execute(
            "UPDATE refresh_tokens SET is_revoked=true, revoked_reason=%s WHERE token_hash=%s",
            (int(reason), token_hash)
        )

    def cleanup_expired(self, cutoff_utc: datetime) -> int:
        return self._db.execute(
            "DELETE FROM refresh_tokens WHERE expires_at <= %s", (cutoff_utc,)
        ).rowcount

---

Step 2 — Wire up services (same for all options above)

from flask import Flask
from kaarmic_auth import TokenCryptoService, TokenValidationService, LoginService
from kaarmic_auth.flask_ext import KaarmicAuth, login_required, current_user, current_context

app = Flask(__name__)

crypto_service     = TokenCryptoService()
validation_service = TokenValidationService(access_store, crypto_service)
login_service      = LoginService(
    access_token_store  = access_store,
    refresh_token_store = refresh_store,
    session_index_store = session_index_store,
    crypto_service      = crypto_service,
    default_access_minutes = 30.0,    # access token lifetime
    default_refresh_days   = 30.0,    # refresh token lifetime
)

kaarmic_auth = KaarmicAuth(app, validation_service, login_service, login_url="/login")

---

Step 3 — Add the auth enforcement hook

from flask import request, redirect, jsonify, g

@app.before_request
def enforce_auth():
    public = ["/login", "/api/auth/login", "/api/auth/logout"]
    if request.path in public or request.path.startswith("/static"):
        return

    if not current_context() or not current_context().is_valid:
        refresh_token = request.cookies.get("refresh_token")
        if refresh_token:
            try:
                new_pair = login_service.refresh_token(refresh_token)
                res = validation_service.validate_access_token(new_pair.access_token)
                if res.is_valid:
                    g.auth_context = res.context
                    g.rotated_tokens = new_pair
                    return
            except Exception:
                pass
        if request.path.startswith("/api/") or request.is_json:
            return jsonify({"error": "Unauthorized"}), 401
        return redirect("/login")

@app.after_request
def attach_rotated_cookies(response):
    rotated = getattr(g, "rotated_tokens", None)
    if rotated:
        response.set_cookie("access_token",  rotated.access_token,  httponly=True, samesite="Lax")
        response.set_cookie("refresh_token", rotated.refresh_token, httponly=True, samesite="Lax")
        response.set_cookie("user_id",       rotated.user_id,                      samesite="Lax")
    return response

---

Step 4 — Add login and logout routes

import hashlib
from kaarmic_auth import AuthenticationContext, RevocationReason

@app.route("/api/auth/login", methods=["POST"])
def api_login():
    data     = request.json or {}
    username = data.get("username", "").strip()
    password = data.get("password", "")

    # --- your own user lookup here ---
    user = your_db.find_user(username=username)
    if not user or user.password_hash != hashlib.sha256(password.encode()).hexdigest():
        return jsonify({"error": "Invalid credentials"}), 401
    # ----------------------------------

    ctx = AuthenticationContext(user_id=str(user.id), metadata={"username": username})
    token_pair = login_service.issue_tokens(ctx)

    resp = jsonify({"status": "success"})
    resp.set_cookie("access_token",  token_pair.access_token,  httponly=True, samesite="Lax")
    resp.set_cookie("refresh_token", token_pair.refresh_token, httponly=True, samesite="Lax")
    resp.set_cookie("user_id",       token_pair.user_id,                      samesite="Lax")
    return resp


@app.route("/api/auth/logout", methods=["POST"])
def api_logout():
    access_token  = request.cookies.get("access_token")
    refresh_token = request.cookies.get("refresh_token")

    if access_token:
        access_store.remove(crypto_service.compute_hash(access_token))
    if refresh_token:
        login_service.revoke_refresh_token(refresh_token, RevocationReason.LOGOUT)
    ctx = current_context()
    if ctx:
        login_service.revoke_session(ctx.user_id, ctx.session_id)

    resp = jsonify({"status": "logged out"})
    resp.delete_cookie("access_token",  path="/", samesite="Lax")
    resp.delete_cookie("refresh_token", path="/", samesite="Lax")
    resp.delete_cookie("user_id",       path="/", samesite="Lax")
    return resp

---

Step 5 — Protect your routes

@app.route("/dashboard")
@login_required
def dashboard():
    uid = current_user()         # returns the authenticated user_id string
    ctx = current_context()      # returns the full AuthenticationContext
    return f"Hello {uid}"

@app.route("/api/data")
def api_data():
    ctx = current_context()
    if not ctx:
        return jsonify({"error": "Unauthorized"}), 401
    # filter data by ctx.user_id

---

Django Integration

Kaarmic.Auth includes a native Django adapter (django_ext) that integrates via standard Django middleware and decorators.

1. Register Services in settings.py

Instantiate your storage and services in your Django settings or an initializer file, and register them as settings attributes so the middleware can discover them:

# settings.py
MIDDLEWARE = [
    # ... standard django middleware ...
    "kaarmic_auth.django_ext.KaarmicAuthMiddleware",
]

# Configure your login redirect page
KAARMIC_AUTH_LOGIN_URL = "/login"

# Setup your preferred storage backends (e.g. Memory + MongoDB)
from pymongo import MongoClient
from kaarmic_auth import TokenCryptoService, TokenValidationService, LoginService
from kaarmic_auth.stores import MemoryAccessTokenStore, MongoRefreshTokenStore, MongoUserSessionIndexStore

db = MongoClient("mongodb://localhost:27017/")["your_db_name"]

access_store        = MemoryAccessTokenStore()
refresh_store       = MongoRefreshTokenStore(db.auth_refresh_tokens)
session_index_store = MongoUserSessionIndexStore(db.auth_session_index)

crypto_service     = TokenCryptoService()

# Export these services for the middleware
KAARMIC_AUTH_VALIDATION_SERVICE = TokenValidationService(access_store, crypto_service)
KAARMIC_AUTH_LOGIN_SERVICE      = LoginService(
    access_token_store  = access_store,
    refresh_token_store = refresh_store,
    session_index_store = session_index_store,
    crypto_service      = crypto_service,
)

---

2. Protect Views and Read Context

Import the Django-specific components (prefixed with django_ to avoid conflicts if co-existing):

# views.py
from django.http import JsonResponse, HttpResponse
from kaarmic_auth import (
    django_login_required,
    django_current_user,
    django_current_context,
)

@django_login_required
def dashboard_view(request):
    # Retrieve user_id or full context by passing the request object explicitly
    user_id = django_current_user(request)
    ctx     = django_current_context(request)
    
    return HttpResponse(f"Hello Django User {user_id} (session: {ctx.session_id})")

---

3. Login and Logout views in Django

# views.py
import hashlib
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.conf import settings
from kaarmic_auth import AuthenticationContext, RevocationReason
from kaarmic_auth.django_ext import current_context

@csrf_exempt
def api_login(request):
    import json
    data     = json.loads(request.body) if request.body else {}
    username = data.get("username", "").strip()
    password = data.get("password", "")

    # Perform your own user authentication
    # user = User.objects.get(username=username) ...
    
    ctx = AuthenticationContext(user_id="user_id_123", metadata={"username": username})
    
    # Retrieve the configured LoginService
    login_service = settings.KAARMIC_AUTH_LOGIN_SERVICE
    token_pair = login_service.issue_tokens(ctx)

    resp = JsonResponse({"status": "success"})
    resp.set_cookie("access_token",  token_pair.access_token,  httponly=True, samesite="Lax")
    resp.set_cookie("refresh_token", token_pair.refresh_token, httponly=True, samesite="Lax")
    resp.set_cookie("user_id",       token_pair.user_id,                      samesite="Lax")
    return resp

@csrf_exempt
def api_logout(request):
    access_token  = request.COOKIES.get("access_token")
    refresh_token = request.COOKIES.get("refresh_token")
    
    # Retrieve configured services
    validation_service = settings.KAARMIC_AUTH_VALIDATION_SERVICE
    login_service      = settings.KAARMIC_AUTH_LOGIN_SERVICE

    if access_token:
        # access_store can be accessed via validation_service._access_token_store
        validation_service._access_token_store.remove(
            validation_service._crypto_service.compute_hash(access_token)
        )
    if refresh_token:
        login_service.revoke_refresh_token(refresh_token, RevocationReason.LOGOUT)
        
    ctx = current_context(request)
    if ctx:
        login_service.revoke_session(ctx.user_id, ctx.session_id)

    resp = JsonResponse({"status": "logged out"})
    resp.delete_cookie("access_token",  path="/")
    resp.delete_cookie("refresh_token", path="/")
    resp.delete_cookie("user_id",       path="/")
    return resp

---

Security Invariants (automatic — no config needed)

• Zero-Plaintext Storage — only SHA256(token) is ever stored; plaintext never persists.
• Lazy Eviction — expired access token hashes are removed from the store at the moment of detection, with no background threads.
• Reuse Attack Defense — presenting a previously rotated refresh token immediately revokes all active sessions for that user.
• Optimistic Concurrency — concurrent refresh rotations are rejected via version field checking.
• Atomic Rollback — if any store throws during issue_tokens, already-written tiers are compensated automatically.

---

Maintainers & Authors

• Krishna (@bitlakrishnasai) - Founder & Core Architect
• Supported and sponsored by Kaarmic Labs (kaarmiclabs.com)