vs-security

Authentication and authorization library for FastAPI services in the Viveka Sutra platform. It provides a provider-based authentication model, generic session handling, JWT token management, and route-level guards — all wired together without boilerplate.

Installation

pip install vs-security

With FastAPI guard support (recommended):

pip install "vs-security[fastapi]"

Dependencies: pyjwt, bcrypt, pydantic, vs-common

Architecture Overview

The library is built around three independent concerns that compose together:

Authentication — who are you? A VsAuthProvider receives credentials (username/password, OAuth token, API key — anything) and returns a VsAuthContext confirming the user's identity.

Session — what do we give back, and how do we verify it? A VsSession[T] has two responsibilities: generate() produces a session of type T after login, and validate() verifies an incoming token and returns the VsAuthContext. Both live in one class so the generate and validate logic stays together. VsJWTProvider is the built-in implementation — it returns VsTokenPair.

Authorization — are you allowed? VsSecurity is a FastAPI dependency that calls validate() on every protected request and makes VsAuthContext available inside the handler.

┌──────────────────────────────────────────────────────────┐
│  LOGIN FLOW                                              │
│                                                          │
│  credentials ──► VsAuthProvider ──► VsAuthContext        │
│                                          │               │
│                                   VsSession.generate()   │
│                                          │               │
│                                          T               │
│                                  (VsTokenPair, etc.)     │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│  REQUEST FLOW                                            │
│                                                          │
│  Authorization: Bearer <token>                           │
│         │                                                │
│      VsSecurity (FastAPI dependency/guard)               │
│         │  calls VsSession.validate(token)               │
│         ▼                                                │
│  VsAuthContext ◄── get_auth_context()                    │
│  available anywhere inside the handler                   │
└──────────────────────────────────────────────────────────┘

Quick Start — Complete Example

Step 1 — Define your auth context

VsAuthContext is a base class with only metadata. Subclass it to add your own fields and control how they serialize into the JWT.

from uuid import UUID
from typing import List
from vs_security.schema.vs_auth_context import VsAuthContext

class AppAuthContext(VsAuthContext):
    user_id: UUID
    username: str
    roles: List[str] = []
    provider: str = ""

    def get_roles(self) -> List[str]:
        return self.roles

    def to_claims(self) -> dict:
        return {
            "sub": str(self.user_id),
            "username": self.username,
            "roles": self.roles,
            "provider": self.provider,
        }

    @classmethod
    def from_claims(cls, payload: dict) -> "AppAuthContext":
        return cls(
            user_id=UUID(payload["sub"]),
            username=payload.get("username", ""),
            roles=payload.get("roles", []),
            provider=payload.get("provider", ""),
        )

Step 2 — Implement your user loader

from vs_security.auth.vs_user_loader import VsUserLoader

class MyUserLoader(VsUserLoader):

    def __init__(self):
        self._repo = UserIdentityRepo()

    async def load(self, username: str):
        identity = await self._repo.find_by_email(username)
        if not identity:
            return None
        context = AppAuthContext(
            user_id=identity.user_id,
            username=identity.email,
            roles=identity.roles,
            provider="email",
        )
        return context, identity.hashed_password

    def verify_password(self, plain: str, stored: str) -> bool:
        import bcrypt
        return bcrypt.checkpw(plain.encode(), stored.encode())

Step 3 — Wire everything at startup

Pass context_class to VsJWTProvider so it knows which subclass to reconstruct on token decode.

from vs_security.auth.vs_auth_manager import VsAuthManager
from vs_security.auth.vs_jwt_provider import VsJWTProvider
from vs_security.auth.vs_username_password_provider import VsUsernamePasswordAuthProvider

jwt_provider = VsJWTProvider(config=config, context_class=AppAuthContext)

VsAuthManager.init(session=jwt_provider)
VsAuthManager.register("username_password", VsUsernamePasswordAuthProvider(
    user_loader=MyUserLoader()
))

# Controllers are imported after this point so VsSecurity.from_manager() resolves correctly
from app.controller.auth_controller import AuthController
from app.controller.user_controller import UserController

Step 4 — Add a login endpoint

authenticate() returns whatever T your VsSession produces. With VsJWTProvider, that is VsTokenPair.

from vs_security.auth.vs_auth_manager import VsAuthManager
from vs_security.schema.vs_credentials import VsUsernamePasswordCredentials
from vs_security.schema.vs_token_pair import VsTokenPair
from vs_security.error.vs_auth_error import VsAuthenticationError
from vs_server.decorator.vs_controller_decorator import controller, post

@controller("/auth")
class AuthController:

    @post("/login", response_model=VsTokenPair)
    async def login(self, body: LoginRequest):
        try:
            return await VsAuthManager.get_instance().authenticate(
                "username_password",
                VsUsernamePasswordCredentials(
                    username=body.username,
                    password=body.password,
                ),
            )
        except VsAuthenticationError as e:
            raise HTTPException(status_code=401, detail=str(e))

Step 5 — Protect routes

from vs_security.guard.vs_security import VsSecurity, get_auth_context
from vs_server.decorator.vs_controller_decorator import controller, get

@controller("/users", guards=[VsSecurity.from_manager()])
class UserController:

    @get("/me")
    async def get_me(self):
        context: AppAuthContext = get_auth_context()
        return {
            "user_id": str(context.user_id),
            "username": context.username,
            "roles": context.roles,
        }

    @get("/admin-only", guards=[VsSecurity.from_manager(roles=["admin"])])
    async def admin_only(self):
        return {"message": "you have admin access"}

Configuration (config.ini)

[auth]
secret_key              = your-very-secret-key-here
algorithm               = HS256
access_expiry_minutes   = 15
refresh_expiry_days     = 7

VsAuthManager — In Depth

VsAuthManager is a singleton coordinator. It routes credentials to the correct provider and hands the resulting VsAuthContext to VsSession.generate().

# Initialize once at startup — pass a VsSession[T] implementation
VsAuthManager.init(session=VsJWTProvider(config=config))

# Register providers — call before importing controllers
VsAuthManager.register("username_password", VsUsernamePasswordAuthProvider(...))
VsAuthManager.register("google", GoogleAuthProvider(...))
VsAuthManager.register("api_key", ApiKeyAuthProvider(...))

# Retrieve anywhere in the app
auth_manager = VsAuthManager.get_instance()

# At login time
token_pair = await auth_manager.authenticate(
    "username_password",
    VsUsernamePasswordCredentials(username=..., password=...),
)

VsAuthProvider — In Depth

VsAuthProvider verifies credentials and returns who the user is. It knows nothing about sessions or tokens.

class VsAuthProvider(ABC):

    @abstractmethod
    async def authenticate(self, credentials: VsCredentials) -> VsAuthContext: ...

Rules:

• Return VsAuthContext on success.
• Raise VsAuthenticationError (or a subclass) on failure — never return None.

Custom provider example — API key:

from vs_security.auth.vs_auth_provider import VsAuthProvider
from vs_security.schema.vs_auth_context import VsAuthContext
from vs_security.schema.vs_credentials import VsCredentials
from vs_security.error.vs_auth_error import VsInvalidCredentialsError

class ApiKeyCredentials(VsCredentials):
    api_key: str

class ApiKeyAuthProvider(VsAuthProvider):

    def __init__(self, key_repo):
        self._repo = key_repo

    async def authenticate(self, credentials: VsCredentials) -> VsAuthContext:
        if not isinstance(credentials, ApiKeyCredentials):
            raise TypeError("Expected ApiKeyCredentials")

        record = await self._repo.find_by_key(credentials.api_key)
        if not record or not record.is_active:
            raise VsInvalidCredentialsError()

        return AppAuthContext(
            user_id=record.owner_id,
            roles=record.roles,
            provider="api_key",
        )

VsAuthManager.register("api_key", ApiKeyAuthProvider(key_repo=ApiKeyRepo()))

VsSession — In Depth

VsSession[T] is the abstract base for both session generation (login) and token validation (requests). Implement it when you want a session format other than JWT.

class VsSession(ABC, Generic[T]):

    @abstractmethod
    async def generate(self, context: VsAuthContext) -> T:
        """Called after login — produce a session from the authenticated context."""
        ...

    @abstractmethod
    async def validate(self, token: str) -> VsAuthContext:
        """Called on each request — verify the token and return the auth context."""
        ...

Custom example — Redis session store:

from vs_security.schema.vs_session import VsSession
from vs_security.schema.vs_auth_context import VsAuthContext
from pydantic import BaseModel
import uuid

class RedisSessionData(BaseModel):
    session_id: str
    expires_in: int

class RedisSession(VsSession[RedisSessionData]):

    def __init__(self, redis_client, ttl: int = 3600):
        self._redis = redis_client
        self._ttl = ttl

    async def generate(self, context: VsAuthContext) -> RedisSessionData:
        session_id = str(uuid.uuid4())
        await self._redis.setex(f"session:{session_id}", self._ttl, context.model_dump_json())
        return RedisSessionData(session_id=session_id, expires_in=self._ttl)

    async def validate(self, token: str) -> VsAuthContext:
        data = await self._redis.get(f"session:{token}")
        if not data:
            raise VsAuthenticationError("Session not found or expired")
        return AppAuthContext.model_validate_json(data)

# Wire it in
VsAuthManager.init(session=RedisSession(redis_client=redis))

VsJWTProvider — Token Lifecycle

VsJWTProvider is the built-in VsSession[VsTokenPair]. It handles minting, verifying, refreshing, and revoking JWTs.

jwt_provider = VsJWTProvider(config=config, context_class=AppAuthContext, token_store=token_store)
VsAuthManager.init(session=jwt_provider)

context_class tells the provider which subclass to reconstruct when decoding a token — defaults to VsAuthContext. token_store is optional; without it, refresh tokens are stateless (cannot be revoked).

Refresh a token:

@post("/auth/refresh")
async def refresh(body: RefreshRequest):
    try:
        pair = await jwt_provider.refresh_token(body.refresh_token)
        return pair
    except VsAuthenticationError as e:
        raise HTTPException(status_code=401, detail=str(e))

Revoke tokens (logout):

revoke_token and revoke_all_tokens take a subject: str — the same value you put in "sub" inside to_claims().

@post("/auth/logout")
async def logout(body: LogoutRequest):
    context: AppAuthContext = get_auth_context()
    await jwt_provider.revoke_token(str(context.user_id), body.refresh_token)

@post("/auth/logout-all")
async def logout_all():
    context: AppAuthContext = get_auth_context()
    await jwt_provider.revoke_all_tokens(str(context.user_id))

Implementing VsTokenStore:

from vs_security.auth.vs_token_store import VsTokenStore

class RedisTokenStore(VsTokenStore):

    def __init__(self, redis_client):
        self._redis = redis_client

    async def save(self, subject: str, refresh_token: str) -> None:
        await self._redis.sadd(f"tokens:{subject}", refresh_token)

    async def get_all(self, subject: str) -> list[str]:
        return list(await self._redis.smembers(f"tokens:{subject}"))

    async def delete(self, subject: str, refresh_token: str) -> None:
        await self._redis.srem(f"tokens:{subject}", refresh_token)

    async def delete_all(self, subject: str) -> None:
        await self._redis.delete(f"tokens:{subject}")

VsSecurity — Route Guards

VsSecurity is a FastAPI callable dependency. On every request it:

1. Reads the Authorization: Bearer <token> header.
2. Calls VsSession.validate(token) to verify and decode it.
3. Checks that the user has the required roles (if any).
4. Stores VsAuthContext in a context variable so get_auth_context() works anywhere in the call stack.

Use VsSecurity.from_manager() to create a guard backed by the initialized VsAuthManager session:

from vs_security.guard.vs_security import VsSecurity

# Any authenticated user
@controller("/users", guards=[VsSecurity.from_manager()])
class UserController: ...

# Only users with "admin" role
@get("/report", guards=[VsSecurity.from_manager(roles=["admin"])])
async def get_report(self): ...

# Multiple roles — user must have at least one
@get("/dashboard", guards=[VsSecurity.from_manager(roles=["admin", "manager"])])
async def dashboard(): ...

Important: Controllers must be imported after VsAuthManager.init() is called, since VsSecurity.from_manager() resolves the session at decoration time.

Accessing the auth context inside a handler:

from vs_security.guard.vs_security import get_auth_context

@get("/me")
async def get_me(self):
    context: AppAuthContext = get_auth_context()
    return {
        "user_id": str(context.user_id),
        "username": context.username,
        "roles": context.roles,
    }

get_auth_context() uses Python's contextvars — safe in async code and always returns the context for the current request.

VsAuthContext Reference

VsAuthContext is a minimal base class. It has no user_id, username, or roles fields — those belong in your subclass so you control the types. The base provides:

context.metadata                      # VsAuthMetadata — extensible metadata bag

context.get_roles() -> List[str]      # override to expose roles for VsSecurity guard
context.has_role("admin")
context.has_any_role("admin", "moderator")
context.has_all_roles("admin", "moderator")

context.to_claims() -> dict           # override to control JWT payload
VsAuthContext.from_claims(payload)    # override to reconstruct from JWT payload

Subclass example:

from uuid import UUID
from typing import List
from vs_security.schema.vs_auth_context import VsAuthContext

class AppAuthContext(VsAuthContext):
    user_id: UUID
    username: str
    roles: List[str] = []

    def get_roles(self) -> List[str]:
        return self.roles

    def to_claims(self) -> dict:
        return {
            "sub": str(self.user_id),
            "username": self.username,
            "roles": self.roles,
        }

    @classmethod
    def from_claims(cls, payload: dict) -> "AppAuthContext":
        return cls(
            user_id=UUID(payload["sub"]),
            username=payload.get("username", ""),
            roles=payload.get("roles", []),
        )

The JWT provider uses to_claims() to build the token payload and from_claims() to reconstruct the context on decode. If you use VsTokenStore, to_claims() must include a "sub" key — that string is used as the token store key.

Error Reference

VsSecurity catches VsAuthenticationError automatically and converts it to HTTP 401. Only catch these in your login and refresh endpoints:

@post("/auth/login")
async def login(body: LoginRequest):
    try:
        return await VsAuthManager.get_instance().authenticate(...)
    except VsInvalidCredentialsError:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    except VsAuthenticationError as e:
        raise HTTPException(status_code=401, detail=str(e))