vs-security

Authentication and authorization library for the Viveka Sutra platform. It provides a provider-based authentication model, generic session handling, JWT token management, and a framework-agnostic security filter — all wired together without boilerplate.

Installation

pip install vs-security

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? A VsSecurityFilter (in vs-server) runs as middleware on every request, validates credentials, and stores VsAuthContext in a ContextVar. Route guards then read that context to enforce roles — with no dependency on any HTTP framework.

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

┌──────────────────────────────────────────────────────────┐
│  REQUEST FLOW                                            │
│                                                          │
│  Incoming request                                        │
│         │                                                │
│  VsSecurityFilter.process()  (middleware, in vs-server)  │
│         │  calls do_filter() → validate → set context    │
│         ▼                                                │
│  VsAuthContext stored in ContextVar                      │
│         │                                                │
│  Route guard reads get_auth_context() → checks roles     │
└──────────────────────────────────────────────────────────┘

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 — Enable security on the server

In your app startup, call enable_security() on the server. The default uses Bearer token auth — pass a custom VsSecurityFilter subclass for any other mechanism.

from vs_server.filter.vs_security_filter import VsSecurityFilter
from vs_server.filter.vs_request import VsRequest

# custom filter — or skip this and use the default VsBearerTokenFilter
class MyFilter(VsSecurityFilter):
    async def do_filter(self, request: VsRequest) -> None:
        from vs_security.auth.vs_auth_manager import VsAuthManager
        from vs_security.error.vs_auth_error import VsAuthenticationError
        from vs_security.guard.vs_security import set_auth_context

        token = (request.get_header("Authorization") or "").removeprefix("Bearer ").strip()
        if not token:
            raise VsAuthenticationError("Missing token")
        context = await VsAuthManager.get_instance().session.validate(token)
        set_auth_context(context)

server = VsServerFactory.get("fastapi", config)
server.enable_security()                              # default: VsBearerTokenFilter
# server.enable_security(MyFilter(exclude_paths=["/health"]))  # custom filter
server.run()

Step 6 — Access context in handlers

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

@controller("/users")
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,
        }

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 the VsAuthContext — your VsTokenStore implementation knows how to extract the key from it.

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

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

Implementing VsTokenStore:

Your implementation receives the full VsAuthContext subclass and decides how to key the tokens — no coupling to a specific field type.

from vs_security.auth.vs_token_store import VsTokenStore
from vs_security.schema.vs_auth_context import VsAuthContext

class RedisTokenStore(VsTokenStore):

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

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

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

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

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

Security Filter (vs-server)

Authentication is handled by VsSecurityFilter in vs-server, not in vs-security. This keeps vs-security free of any HTTP framework dependency.

See the vs-server README for full documentation on VsSecurityFilter, VsRequest, VsBearerTokenFilter, and enable_security().

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}

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

VsSecurity — Deprecated

VsSecurity is kept for backwards compatibility but raises a DeprecationWarning on use. Migrate to VsSecurityFilter + get_auth_context().

# old — deprecated
from vs_security.guard.vs_security import VsSecurity
@controller("/users", guards=[VsSecurity.from_manager()])

# new — use enable_security() on the server at startup
server.enable_security()

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

All auth exceptions extend VsAuthenticationError, which extends VsBaseException from vs-common. VsAuthenticationError defines status_code as its own attribute (not inherited from VsBaseException, which has no HTTP concepts). When using vs-server, its global handler reads status_code via duck-typing and produces the correct HTTP response automatically — no try/catch needed in your controllers.

When using vs-server, all of these are handled globally — your login endpoint needs no try/catch:

@post("/auth/login", response_model=LoginResponse)
async def login(body: LoginRequest):
    session = await VsAuthManager.get_instance().authenticate(body.type, body)
    return LoginResponse(access_token=session.access_token, refresh_token=session.refresh_token)

If you are not using vs-server, catch them explicitly:

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