Metadata-Version: 2.4
Name: vs-security
Version: 0.1.13
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 |

---

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