vs-security

JWT-based authentication and authorization library for FastAPI services in the Viveka Sutra platform. Drop it in, configure a secret key, and your endpoints are protected in minutes.

Installation

pip install vs-security

With FastAPI guard support (recommended):

pip install "vs-security[fastapi]"

Dependencies: pyjwt, bcrypt, pydantic, vs-common

How It Works

Login Request
     │
     ▼
VsAuthManager.authenticate()
     │
     ▼
VsAuthProvider (e.g. VsUsernamePasswordAuthProvider)
     │  verifies credentials, returns VsAuthContext
     ▼
VsJWTProvider.generate_token()
     │  mints access + refresh JWT pair
     ▼
VsTokenPair  ──────────────────────────────────┐
                                               │
Protected Route Request (Bearer token)         │
     │                                         │
     ▼                                         │
VsSecurity (FastAPI dependency)                │
     │  verifies token, checks roles           │
     ▼                                         │
VsAuthContext available via get_auth_context() ◄┘

Quick Start

1. Initialize at startup

from vs_security.auth.vs_jwt_provider import VsJWTProvider
from vs_security.auth.vs_auth_manager import VsAuthManager
from vs_security.auth.vs_username_password_provider import VsUsernamePasswordAuthProvider
from vs_security.guard.vs_security_factory import VsSecurityFactory

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

auth_manager = VsAuthManager(jwt_provider=jwt_provider)
auth_manager.register("username_password", VsUsernamePasswordAuthProvider(
    jwt_provider=jwt_provider,
    user_loader=my_user_loader,   # see below
))

VsSecurityFactory.init(
    secret_key=config.get("auth.secret_key"),
    algorithm="HS256",
)

2. Implement a user loader

async def my_user_loader(username: str):
    identity = await identity_repo.find_by_username(username)
    if not identity:
        return None
    context = VsAuthContext(
        user_id=identity.user_id,
        username=identity.username,
        roles=identity.roles,
        provider="email",
    )
    return context, identity.hashed_password  # hashed_password is bcrypt hash

3. Add a login endpoint

from vs_security.schema.vs_credentials import VsUsernamePasswordCredentials
from vs_security.error.vs_auth_error import VsAuthenticationError

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

4. Protect routes

from vs_security.guard.vs_security_factory import VsSecurityFactory
from vs_security.guard.vs_security import get_auth_context

# Protect all routes in a controller
@controller("/llm", guards=[VsSecurityFactory.get()])
class LlmController:

    @get("/data")
    async def get_data(self):
        context = get_auth_context()   # available anywhere after guard runs
        return {"user_id": str(context.user_id)}

5. Role-based access

# Only users with "admin" role can access
@controller("/admin", guards=[VsSecurityFactory.with_roles(["admin"])])
class AdminController:
    ...

# Or check roles manually inside a handler
context = get_auth_context()
if not context.has_role("admin"):
    raise HTTPException(status_code=403)

Configuration

Set these in your config.ini (read by VsBaseConfig):

Token Lifecycle

# Refresh an access token
new_pair = await jwt_provider.refresh_token(body.refresh_token)

# Revoke a single refresh token (logout)
await jwt_provider.revoke_token(user_id, body.refresh_token)

# Revoke all tokens (logout everywhere)
await jwt_provider.revoke_all_tokens(user_id)

Token revocation requires a VsTokenStore implementation. Implement the interface and pass it to VsJWTProvider:

class VsTokenStore(ABC):
    async def save(self, user_id: UUID, refresh_token: str) -> None: ...
    async def get_all(self, user_id: UUID) -> List[str]: ...
    async def delete(self, user_id: UUID, refresh_token: str) -> None: ...
    async def delete_all(self, user_id: UUID) -> None: ...

Custom Auth Providers

Extend VsAuthProvider to support OAuth, API keys, or any other credential type:

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

class MyApiKeyCredentials(VsCredentials):
    api_key: str

class ApiKeyAuthProvider(VsAuthProvider):
    async def authenticate(self, credentials: VsCredentials) -> VsAuthContext:
        # validate api_key, return VsAuthContext
        ...

auth_manager.register("api_key", ApiKeyAuthProvider())

Error Reference

VsSecurity maps these automatically to HTTP responses — you only need to catch them in your login/refresh endpoints.

VsAuthContext Reference

Available inside any protected route via get_auth_context():

context.user_id          # UUID
context.username         # str
context.roles            # List[str]
context.provider         # str  (e.g. "email")

context.has_role("admin")              # bool
context.has_any_role("admin", "mod")   # bool
context.has_all_roles("admin", "mod")  # bool