Metadata-Version: 2.4
Name: vs-security
Version: 0.1.10
Summary: Security library for Viveka Sutra — authentication, JWT, and authorization
Project-URL: Homepage, https://vivekasutra.com/
Project-URL: Source, https://github.com/vivekasutra/viveka-mula
Keywords: security,auth,jwt,authentication,viveka,vs
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: Other/Proprietary License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Framework :: AsyncIO
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: pydantic>=2.0
Requires-Dist: pyjwt>=2.8
Requires-Dist: bcrypt>=4.0
Requires-Dist: vs-common>=0.1.3
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110; extra == "fastapi"
Provides-Extra: dev
Requires-Dist: build; extra == "dev"
Requires-Dist: twine; extra == "dev"
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"

# 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

```bash
pip install vs-security
```

With FastAPI guard support (recommended):

```bash
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                   │
└──────────────────────────────────────────────────────────┘
```

| Class | Role | Extend? |
|---|---|---|
| `VsAuthManager` | Coordinator — routes credentials to the right provider and calls the session | No |
| `VsAuthProvider` | Verifies credentials, returns `VsAuthContext` | Yes — one per auth mechanism |
| `VsSession[T]` | Abstract base — implement `generate()` and `validate()` to define what a session looks like and how it is verified | Yes — to change the session format |
| `VsUserLoader` | Loads a user from your database for username/password auth | Yes — required for `VsUsernamePasswordAuthProvider` |
| `VsJWTProvider` | Built-in `VsSession[VsTokenPair]` — mints, verifies, refreshes, and revokes JWTs. Accepts `context_class` to control which subclass is reconstructed on decode | No — use as-is |
| `VsSecurity` | FastAPI dependency that guards routes | No — use `VsSecurity.from_manager()` |

---

## 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.

```python
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

```python
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.

```python
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`.

```python
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

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

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

| Key | Default | Description |
|-----|---------|-------------|
| `auth.secret_key` | required | JWT signing secret |
| `auth.algorithm` | `HS256` | JWT signing algorithm |
| `auth.access_expiry_minutes` | `15` | Access token lifetime |
| `auth.refresh_expiry_days` | `7` | Refresh token lifetime |

---

## VsAuthManager — In Depth

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

```python
# 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.

```python
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:**

```python
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.

```python
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:**

```python
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.

```python
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:**

```python
@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.

```python
@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.

```python
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}")
```

---

## 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:

```python
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:**

```python
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:

```python
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:**

```python
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

| Exception | HTTP Status | Raised By | When |
|-----------|-------------|-----------|------|
| `VsInvalidCredentialsError` | 401 | `VsUsernamePasswordAuthProvider` | Wrong username or password |
| `VsAuthenticationError` | 401 | Any provider or session | Generic auth failure — base class for all auth errors |
| `VsTokenExpiredError` | 401 | `VsJWTProvider` | JWT has expired |
| `VsTokenRevokedError` | 401 | `VsJWTProvider` | Refresh token was explicitly revoked |
| `VsInsufficientRolesError` | 403 | `VsSecurity` | User authenticated but lacks required roles |

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

```python
@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))
```
