Metadata-Version: 2.4
Name: vs-security
Version: 0.1.4
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, flexible session generation, 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 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:

| Class | Role | Extend? |
|---|---|---|
| `VsAuthManager` | Coordinator — routes credentials to the right provider and calls the session generator | No |
| `VsAuthProvider` | Verifies credentials, returns `VsAuthContext` | Yes — one per auth mechanism |
| `VsSessionGenerator` | Converts `VsAuthContext` into a `VsSession` | Yes — to change what is returned after login |
| `VsUserLoader` | Loads a user from your database for username/password auth | Yes — required for `VsUsernamePasswordAuthProvider` |
| `VsJWTProvider` | Mints, verifies, refreshes, and revokes JWT tokens. Also implements `VsSessionGenerator` — pass it directly as the session generator | No — use as-is |
| `VsSecurity` | FastAPI dependency that guards routes | No — configure via `VsSecurityFactory` |

---

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

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

```python
# 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_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
#    VsJWTProvider implements VsSessionGenerator, so pass it directly
jwt_provider = VsJWTProvider(config=config)

# 3. Initialize the singleton — call once at startup
auth_manager = VsAuthManager.init(session_generator=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`. When using `VsJWTProvider` as the session generator, it contains `access_token` and `refresh_token`.

```python
# 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):
        self._auth = VsAuthManager.get_instance()

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

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

```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 — keep this out of source control |
| `auth.algorithm` | `HS256` | JWT signing algorithm |
| `auth.access_expiry_minutes` | `15` | How long an access token is valid |
| `auth.refresh_expiry_days` | `7` | How long a refresh token is valid |

---

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

```python
# 1. Register providers — on the class, before init
VsAuthManager.register("username_password", VsUsernamePasswordAuthProvider(...))
VsAuthManager.register("google", GoogleAuthProvider(...))
VsAuthManager.register("api_key", ApiKeyAuthProvider(...))

# 2. Initialize the singleton once at startup
VsAuthManager.init(session_generator=jwt_provider)

# 3. Retrieve the instance anywhere in the app
auth_manager = VsAuthManager.get_instance()

# 4. Authenticate 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`.

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

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

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

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

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

```python
class VsSessionGenerator(ABC):

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

`VsSession` holds the result:

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

`VsJWTProvider` implements `VsSessionGenerator` directly — its `generate()` method mints a JWT access/refresh pair. Pass it as the session generator without any wrapper.

```python
jwt_provider = VsJWTProvider(config=config, token_store=token_store)
auth_manager = VsAuthManager(session_generator=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`:

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

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

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

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

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

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

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

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

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

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

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

| Exception | HTTP Status | Raised By | When |
|-----------|-------------|-----------|------|
| `VsInvalidCredentialsError` | 401 | `VsUsernamePasswordAuthProvider` | Wrong username or password |
| `VsAuthenticationError` | 401 | Any provider | Generic auth failure — base class for all auth errors |
| `VsTokenExpiredError` | 401 | `VsJWTProvider` | JWT access or refresh token 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 an HTTP 401 response. You only need to catch these exceptions in your login and refresh endpoints.

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