Metadata-Version: 2.4
Name: vs-security
Version: 0.1.15
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.5
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 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

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

**Dependencies:**

| Package | Version | Purpose |
|---|---|---|
| `pydantic` | `>=2.0` | Schema validation |
| `pyjwt` | `>=2.8` | JWT encoding/decoding |
| `bcrypt` | `>=4.0` | Password hashing |
| `vs-common` | `>=0.1.5` | Config, logging, base exceptions |

---

## Shared request authentication

`VsRequestAuthenticator` is the transport-independent interface used by HTTP and MCP adapters:

```python
from vs_security import (
    VsRequestAuthenticator, VsAuthRequest, VsAuthContext,
    VsBearerTokenAuthenticator, VsJWTProvider,
    get_auth_context,
)

# Built-in adapter: extract Authorization: Bearer <token>, then validate the session.
authenticator = VsBearerTokenAuthenticator(VsJWTProvider(config))
# Pass authenticator=authenticator to VsAgentServer or VsFastMcpServer.
# For an existing session, use VsBearerTokenAuthenticator(session).
```

For stdio bearer credentials, pass `VsBearerTokenAuthenticator(session, stdio_token_env="VS_AGENT_TOKEN")`. The launcher supplies the token through that explicitly named environment variable. Each stdio request validates it through the same session; no token source is inferred. These credentials identify the launched process. Custom authenticators receive empty headers, `transport="stdio"`, and the MCP method name.

For custom authentication, subclass `VsRequestAuthenticator` and implement `async authenticate(self, request: VsAuthRequest) -> VsAuthContext`. Validate credentials before returning identity. Raise `VsAuthenticationError` for rejected credentials or `VsInsufficientRolesError` for insufficient permissions. Unexpected infrastructure and programming errors must propagate unchanged; do not convert them to credential rejection. The adapters log them and propagate server failures. Cancellation also propagates.

`VsAuthRequest` contains `headers`, `transport`, `path`, and `method`. Transport is validated at runtime and must be `http`, `mcp`, `stdio`, or `queue`; a typo such as `htpp` raises `ValueError`. Queue requests additionally expose `topic`, `message_id`, `reply_to`, and `correlation_id`; `get_header(name)` is case-insensitive. Headers are copied into an immutable mapping and omitted from its representation. The interface imports without FastAPI or MCP. Implementations must support concurrent requests across transport threads and event loops.

Return a context rather than modifying a `ContextVar`. HTTP/MCP adapters set the single shared context for the request and reset it in `finally`. Handlers and guards use `get_auth_context()`; old imports from `vs_security.guard.vs_security` and `vs_mcp_agent.auth.vs_mcp_auth` read the same context. Roles are supplied by your `VsAuthContext` subclass; the base context grants no roles.

---

## Legacy compatibility

The existing `VsBearerTokenFilter` delegates bearer parsing and session validation to `VsBearerTokenAuthenticator`. Its `enrich(context)` extension still runs after authentication, so existing applications can continue fetching fresh user details there.

`from vs_security import *` exports the framework-independent API without loading FastAPI. The deprecated `VsSecurity` dependency is intentionally excluded from star imports. Existing explicit imports (`from vs_security import VsSecurity`) still work when the FastAPI extra is installed.

---

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

| 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~~ **Deprecated** — use `VsSecurityFilter` + `get_auth_context()` via `vs-server` | — |

---

## Quick Start — Complete Example

### Step 1 — Define your auth context

`VsAuthContext` is a base class with a `metadata: Dict[str, Any]` bag for arbitrary extra data. Subclass it to add your own typed 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 — 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.

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

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

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

---

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

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

`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()`.

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

```python
context.metadata                      # Dict[str, Any] — arbitrary extra data 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

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.

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

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

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

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

Queue adapters authenticate the same way using message headers and routing metadata. Reply destinations require separate authorization through the agent's `queue_reply_guard`; invalid credentials never authorize a response destination.
