vs-security

Authentication and authorization library for FastAPI services in the Viveka Sutra platform. It provides a provider-based authentication model, flexible session generation, 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 and roles.

Session generation — what do we give back? A VsSessionGenerator receives the authenticated VsAuthContext and produces a VsSession. By default this mints a JWT pair. You can swap in any other mechanism — Redis sessions, opaque tokens, signed cookies — without changing anything else.

Authorization — are you allowed? VsSecurity is a FastAPI dependency that verifies the incoming token on every protected request and makes VsAuthContext available inside the handler.

┌──────────────────────────────────────────────────────────┐
│  LOGIN FLOW                                              │
│                                                          │
│  credentials ──► VsAuthProvider ──► VsAuthContext        │
│                                          │               │
│                                   VsSessionGenerator     │
│                                          │               │
│                                       VsSession          │
│                                  (access_token etc.)     │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│  REQUEST FLOW                                            │
│                                                          │
│  Authorization: Bearer <token>                           │
│         │                                                │
│      VsSecurity (FastAPI dependency/guard)               │
│         │  verifies token, checks roles                  │
│         ▼                                                │
│  VsAuthContext ◄── get_auth_context()                    │
│  available anywhere inside the handler                   │
└──────────────────────────────────────────────────────────┘

The key components and their relationships:

Quick Start — Complete Example

This shows a fully wired JWT-based auth setup from startup to protected route.

Step 1 — Implement your user loader

VsUserLoader is the bridge between your database and the auth system. It is the only class you must implement for username/password auth.

# app/auth/my_user_loader.py
from vs_security.auth.vs_user_loader import VsUserLoader
from vs_security.schema.vs_auth_context import VsAuthContext
from app.repo.user_identity_repo import UserIdentityRepo

class MyUserLoader(VsUserLoader):

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

    async def load(self, username: str):
        """
        Look up a user by username.
        Return (VsAuthContext, stored_credential) if found, None if not.
        stored_credential is the hashed password string from your database.
        """
        identity = await self._repo.find_by_email(username)
        if not identity:
            return None

        context = VsAuthContext(
            user_id=identity.user_id,
            username=identity.email,
            roles=identity.roles,   # e.g. ["user", "admin"]
            provider="email",
        )
        return context, identity.hashed_password  # hashed_password is a bcrypt hash

    # Optional — override only if you use a hashing scheme other than bcrypt
    def verify_password(self, plain: str, stored: str) -> bool:
        import bcrypt
        return bcrypt.checkpw(plain.encode(), stored.encode())

Step 2 — Wire everything at startup

Providers are registered on the VsAuthManager class (static registry) before any instance is created. The instance only needs the session generator.

# app/startup.py
from vs_common.config.vs_base_config import VsBaseConfig
from vs_security.auth.vs_auth_manager import VsAuthManager
from vs_security.auth.vs_jwt_provider import VsJWTProvider
from vs_security.auth.vs_jwt_session_generator import VsJWTSessionGenerator
from vs_security.auth.vs_username_password_provider import VsUsernamePasswordAuthProvider
from vs_security.guard.vs_security_factory import VsSecurityFactory
from app.auth.my_user_loader import MyUserLoader

config = VsBaseConfig("config.ini")

# 1. Register auth providers on the class — this happens once at module load
VsAuthManager.register(
    "username_password",
    VsUsernamePasswordAuthProvider(user_loader=MyUserLoader())
)

# 2. Create the JWT provider — reads secret key and expiry from config
jwt_provider = VsJWTProvider(config=config)

# 3. Create the auth manager with JWT session generation
auth_manager = VsAuthManager(
    session_generator=VsJWTSessionGenerator(jwt_provider=jwt_provider)
)

# 4. Initialize the guard factory so protected routes can verify tokens
VsSecurityFactory.init(
    secret_key=config.get("auth.secret_key"),
    algorithm="HS256",
)

Step 3 — Add a login endpoint

authenticate() returns a VsSession. For VsJWTSessionGenerator, it contains access_token and refresh_token.

# app/controller/auth_controller.py
from pydantic import BaseModel
from fastapi import HTTPException
from vs_server.decorator.vs_controller_decorator import controller, post
from vs_security.auth.vs_auth_manager import VsAuthManager
from vs_security.schema.vs_credentials import VsUsernamePasswordCredentials
from vs_security.error.vs_auth_error import VsAuthenticationError

class LoginRequest(BaseModel):
    username: str
    password: str

class LoginResponse(BaseModel):
    access_token: str
    refresh_token: str

@controller("/auth")
class AuthController:

    def __init__(self, auth_manager: VsAuthManager):
        self._auth = auth_manager

    @post("/login", response_model=LoginResponse, status_code=200)
    async def login(self, body: LoginRequest):
        try:
            session = await self._auth.authenticate(
                "username_password",
                VsUsernamePasswordCredentials(
                    username=body.username,
                    password=body.password,
                ),
            )
            return LoginResponse(
                access_token=session.access_token,
                refresh_token=session.refresh_token,
            )
        except VsAuthenticationError as e:
            raise HTTPException(status_code=401, detail=str(e))

Step 4 — Protect routes

Apply VsSecurityFactory.get() as a guard to any controller or individual route. The guard automatically extracts the Authorization: Bearer <token> header, verifies it, and makes the user context available.

# app/controller/user_controller.py
from vs_server.decorator.vs_controller_decorator import controller, get
from vs_security.guard.vs_security_factory import VsSecurityFactory
from vs_security.guard.vs_security import get_auth_context

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

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

    @get("/admin-only", guards=[VsSecurityFactory.with_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 the central coordinator for the login flow. It does two things:

1. Routes incoming credentials to the correct VsAuthProvider based on the provider name string.
2. Calls VsSessionGenerator.generate() on the resulting VsAuthContext to produce a VsSession.

The provider registry is a class-level dict shared across all instances. This means you register providers once at startup (before creating any instance), and every instance sees them.

# Registration — happens once at startup, on the class
VsAuthManager.register("username_password", VsUsernamePasswordAuthProvider(...))
VsAuthManager.register("google", GoogleAuthProvider(...))
VsAuthManager.register("api_key", ApiKeyAuthProvider(...))

# Instantiation — happens once, takes the session strategy
auth_manager = VsAuthManager(session_generator=VsJWTSessionGenerator(jwt_provider))

# Usage — at login time
session = await auth_manager.authenticate(
    "username_password",                         # which provider to use
    VsUsernamePasswordCredentials(...),          # credentials for that provider
)

The provider name is arbitrary — it just has to match between register() and authenticate(). Using a string (not a type) means a single manager can support multiple providers simultaneously. A mobile app might call authenticate("google", ...) while a web form calls authenticate("username_password", ...) — same manager, different providers.

VsAuthProvider — In Depth

VsAuthProvider is the place where credential verification happens. Each provider handles one credential type. It receives raw credentials and returns a VsAuthContext or raises a VsAuthenticationError.

from abc import ABC, abstractmethod
from vs_security.schema.vs_auth_context import VsAuthContext
from vs_security.schema.vs_credentials import VsCredentials

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.
• The provider does not know or care about sessions or tokens. Its only job is to say "yes, this is user X with roles Y".

Built-in provider: VsUsernamePasswordAuthProvider handles email/password login via VsUserLoader.

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
from pydantic import BaseModel

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 VsAuthContext(
            user_id=record.owner_id,
            username=record.owner_email,
            roles=record.roles,
            provider="api_key",
        )

# Register at startup
VsAuthManager.register("api_key", ApiKeyAuthProvider(key_repo=ApiKeyRepo()))

# Login
session = await auth_manager.authenticate(
    "api_key",
    ApiKeyCredentials(api_key=request.headers["X-Api-Key"]),
)

VsUserLoader — In Depth

VsUserLoader is used exclusively by VsUsernamePasswordAuthProvider. It separates the database concern (finding the user) from the authentication concern (verifying the password).

class VsUserLoader(ABC):

    @abstractmethod
    async def load(self, username: str) -> Optional[Tuple[VsAuthContext, str]]:
        """Returns (VsAuthContext, stored_credential) or None if user not found."""
        ...

    def verify_password(self, plain: str, stored: str) -> bool:
        """Defaults to bcrypt. Override for argon2, scrypt, or any other scheme."""
        import bcrypt
        return bcrypt.checkpw(plain.encode(), stored.encode())

load() — receives the username (email, phone, or whatever your system uses) and returns a tuple of (VsAuthContext, stored_credential). The stored_credential is the hashed password string exactly as stored in your database. Return None if the user is not found — the provider will raise VsInvalidCredentialsError automatically.

verify_password() — receives the raw plaintext password the user submitted and the stored credential from load(). The default implementation uses bcrypt. Override this if you hash passwords with a different algorithm.

# argon2 example
def verify_password(self, plain: str, stored: str) -> bool:
    from argon2 import PasswordHasher
    ph = PasswordHasher()
    try:
        return ph.verify(stored, plain)
    except Exception:
        return False

VsAuthContext can be a subclass. If your system has extra fields (tenant ID, subscription plan, etc.) you want available inside protected routes, extend VsAuthContext:

from vs_security.schema.vs_auth_context import VsAuthContext

class AppAuthContext(VsAuthContext):
    tenant_id: str
    plan: str

# In your loader's load() method:
return AppAuthContext(
    user_id=user.id,
    username=user.email,
    roles=user.roles,
    provider="email",
    tenant_id=user.tenant_id,
    plan=user.subscription_plan,
), user.hashed_password

# Inside a protected handler:
context = get_auth_context()          # returns the AppAuthContext you set
print(context.tenant_id)              # works — it's the same object

VsSessionGenerator — In Depth

VsSessionGenerator defines what the login response looks like. After VsAuthProvider confirms identity, VsAuthManager hands the VsAuthContext to the session generator to produce a VsSession.

class VsSessionGenerator(ABC):

    @abstractmethod
    async def generate(self, context: VsAuthContext) -> VsSession: ...

VsSession holds the result:

class VsSession(BaseModel):
    access_token: Optional[str] = None    # JWT access token, opaque token, etc.
    refresh_token: Optional[str] = None   # JWT refresh token, session renewal token, etc.
    data: Dict[str, Any] = {}             # any extra fields your implementation needs

Built-in — VsJWTSessionGenerator:

Wraps VsJWTProvider to mint a JWT access/refresh pair. This is the default for most services.

jwt_provider = VsJWTProvider(config=config, token_store=token_store)
auth_manager = VsAuthManager(session_generator=VsJWTSessionGenerator(jwt_provider=jwt_provider))

# After authenticate():
# session.access_token  → signed JWT, e.g. "eyJhbGciOi..."
# session.refresh_token → signed JWT for token renewal
# session.data          → {} (empty for JWT)

Custom — Redis session store:

If you want server-side sessions instead of stateless JWTs, implement VsSessionGenerator:

from vs_security.auth.vs_session_generator import VsSessionGenerator
from vs_security.schema.vs_auth_context import VsAuthContext
from vs_security.schema.vs_session import VsSession
import uuid

class RedisSessionGenerator(VsSessionGenerator):

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

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

# Wire it in
auth_manager = VsAuthManager(session_generator=RedisSessionGenerator(redis_client=redis))

# After authenticate():
# session.access_token  → None
# session.refresh_token → None
# session.data          → {"session_id": "abc-123", "expires_in": 3600}

Custom — opaque token:

class OpaqueTokenGenerator(VsSessionGenerator):

    def __init__(self, token_store):
        self._store = token_store

    async def generate(self, context: VsAuthContext) -> VsSession:
        import secrets
        token = secrets.token_urlsafe(32)
        await self._store.save(token, context)
        return VsSession(access_token=token)

VsJWTProvider — Token Lifecycle

VsJWTProvider handles everything JWT-related: minting tokens, verifying them, refreshing them, and revoking them. It reads its configuration from config.ini.

jwt_provider = VsJWTProvider(config=config, token_store=token_store)

token_store is optional. Without it, refresh tokens are stateless (cannot be revoked). With it, refresh tokens are tracked and can be revoked individually or all at once.

Refresh a token:

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

Revoke tokens (logout):

@post("/auth/logout")
async def logout(body: LogoutRequest):
    context = get_auth_context()
    await jwt_provider.revoke_token(context.user_id, body.refresh_token)
    return {"message": "logged out"}

@post("/auth/logout-all")
async def logout_all():
    context = get_auth_context()
    await jwt_provider.revoke_all_tokens(context.user_id)
    return {"message": "logged out from all devices"}

Implementing VsTokenStore:

from vs_security.auth.vs_token_store import VsTokenStore
from uuid import UUID

class RedisTokenStore(VsTokenStore):

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

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

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

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

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

VsSecurity — Route Guards

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

1. Reads the Authorization: Bearer <token> header.
2. Verifies the token signature and expiry via VsJWTProvider.
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.

Guards are configured via VsSecurityFactory:

# Protect all routes in a controller — any authenticated user
@controller("/users", guards=[VsSecurityFactory.get()])
class UserController: ...

# Protect specific routes — only users with "admin" role
@controller("/admin")
class AdminController:

    @get("/report", guards=[VsSecurityFactory.with_roles(["admin"])])
    async def get_report(self): ...

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

VsSecurityFactory.init() must be called once at startup before any guards are used:

VsSecurityFactory.init(
    secret_key=config.get("auth.secret_key"),
    algorithm="HS256",   # must match what VsJWTProvider uses
)

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 = get_auth_context()
    return {
        "user_id": str(context.user_id),
        "username": context.username,
        "roles": context.roles,
        "provider": context.provider,
    }

get_auth_context() uses Python's contextvars — it is safe in async code and always returns the context for the current request, even under concurrent load.

VsAuthContext Reference

VsAuthContext is the canonical representation of an authenticated user. It is available inside any guarded handler via get_auth_context().

context.user_id          # UUID   — unique user identifier
context.username         # str    — display name or email
context.roles            # List[str] — e.g. ["user", "admin"]
context.provider         # str    — which provider authenticated the user ("email", "google", etc.)

context.has_role("admin")                    # True if "admin" is in roles
context.has_any_role("admin", "moderator")   # True if any of these roles match
context.has_all_roles("admin", "moderator")  # True if all roles match

To attach custom fields, extend VsAuthContext and return the subclass from your VsUserLoader.load():

class AppAuthContext(VsAuthContext):
    tenant_id: str

# In load():
return AppAuthContext(..., tenant_id=user.tenant_id), user.hashed_password

# In a handler:
context: AppAuthContext = get_auth_context()
print(context.tenant_id)

Error Reference

VsSecurity catches VsAuthenticationError automatically and converts it to an HTTP 401 response. You only need to catch these exceptions in your login and refresh endpoints.

@post("/auth/login")
async def login(body: LoginRequest):
    try:
        session = await auth_manager.authenticate(...)
        return {"access_token": session.access_token}
    except VsInvalidCredentialsError:
        raise HTTPException(status_code=401, detail="Invalid username or password")
    except VsAuthenticationError as e:
        raise HTTPException(status_code=401, detail=str(e))