langchain-enterprise-rbac

name	langchain-enterprise-rbac
description	Implement enterprise role-based access control for LangChain applications. Use when implementing user permissions, multi-tenant access, or enterprise security controls for LLM applications. Trigger with phrases like "langchain RBAC", "langchain permissions", "langchain access control", "langchain multi-tenant", "langchain enterprise auth".
allowed-tools	Read, Write, Edit
version	1.0.0
license	MIT
author	Jeremy Longshore <[e-mail]>

LangChain Enterprise RBAC

Overview

Implement role-based access control (RBAC) for LangChain applications with multi-tenant support and fine-grained permissions.

Prerequisites

• LangChain application with user authentication
• Identity provider (Auth0, Okta, Azure AD)
• Understanding of RBAC concepts

Instructions

Step 1: Define Permission Model

from enum import Enum
from typing import Set, Optional
from pydantic import BaseModel
from datetime import datetime

class Permission(str, Enum):
    # Chain permissions
    CHAIN_READ = "chain:read"
    CHAIN_EXECUTE = "chain:execute"
    CHAIN_CREATE = "chain:create"
    CHAIN_DELETE = "chain:delete"

    # Model permissions
    MODEL_GPT4 = "model:gpt-4"
    MODEL_GPT4_MINI = "model:gpt-4o-mini"
    MODEL_CLAUDE = "model:claude"

    # Feature permissions
    FEATURE_STREAMING = "feature:streaming"
    FEATURE_TOOLS = "feature:tools"
    FEATURE_RAG = "feature:rag"

    # Admin permissions
    ADMIN_USERS = "admin:users"
    ADMIN_BILLING = "admin:billing"
    ADMIN_AUDIT = "admin:audit"

class Role(BaseModel):
    name: str
    permissions: Set[Permission]
    description: str = ""

# Predefined roles
ROLES = {
    "viewer": Role(
        name="viewer",
        permissions={Permission.CHAIN_READ},
        description="Read-only access to chains"
    ),
    "user": Role(
        name="user",
        permissions={
            Permission.CHAIN_READ,
            Permission.CHAIN_EXECUTE,
            Permission.MODEL_GPT4_MINI,
        },
        description="Standard user with execution rights"
    ),
    "power_user": Role(
        name="power_user",
        permissions={
            Permission.CHAIN_READ,
            Permission.CHAIN_EXECUTE,
            Permission.CHAIN_CREATE,
            Permission.MODEL_GPT4_MINI,
            Permission.MODEL_GPT4,
            Permission.FEATURE_STREAMING,
            Permission.FEATURE_TOOLS,
        },
        description="Power user with advanced features"
    ),
    "admin": Role(
        name="admin",
        permissions=set(Permission),  # All permissions
        description="Full administrative access"
    ),
}

Step 2: User and Tenant Management

from typing import Dict, List
import uuid

class Tenant(BaseModel):
    id: str
    name: str
    allowed_models: List[str] = []
    monthly_token_limit: int = 1_000_000
    features: Set[str] = set()
    created_at: datetime = None

class User(BaseModel):
    id: str
    email: str
    tenant_id: str
    roles: List[str]
    created_at: datetime = None

    def get_permissions(self) -> Set[Permission]:
        """Get all permissions for user based on roles."""
        permissions = set()
        for role_name in self.roles:
            if role_name in ROLES:
                permissions.update(ROLES[role_name].permissions)
        return permissions

    def has_permission(self, permission: Permission) -> bool:
        """Check if user has specific permission."""
        return permission in .get_permissions()

 :
    

     ():
        .tenants: [, Tenant] = {}
        .users: [, User] = {}

     () -> Tenant:
        tenant = Tenant(
            =(uuid.uuid4()),
            name=name,
            created_at=datetime.now(),
            **kwargs
        )
        .tenants[tenant.] = tenant
         tenant

     () -> User:
         tenant_id   .tenants:
             ValueError()

        user = User(
            =(uuid.uuid4()),
            email=email,
            tenant_id=tenant_id,
            roles=roles  [],
            created_at=datetime.now()
        )
        .users[user.] = user
         user

     () -> [Tenant]:
        user = .users.get(user_id)
         user:
             .tenants.get(user.tenant_id)
         

Step 3: Permission Enforcement

from functools import wraps
from fastapi import HTTPException, Depends, Request
from typing import Callable

class PermissionChecker:
    """Check and enforce permissions."""

    def __init__(self, user_store: UserStore):
        self.user_store = user_store

    def require_permission(self, permission: Permission):
        """Decorator to require specific permission."""
        def decorator(func: Callable):
            @wraps(func)
            async def wrapper(request: Request, *args, **kwargs):
                user_id = request.state.user_id  # Set by auth middleware
                user = self.user_store.users.get(user_id)

                if not user:
                    raise HTTPException(status_code=401, detail="User not found")

                if not user.has_permission(permission):
                    raise HTTPException(
                        status_code=403,
                        detail=f"Permission denied: {permission.value}"
                    )

                return await func(request, *args, **kwargs)
             wrapper
         decorator

     ():
        
         ():

              ():
                user_id = request.state.user_id
                user = .user_store.users.get(user_id)

                  user:
                     HTTPException(status_code=)

                  (user.has_permission(p)  p  permissions):
                     HTTPException(status_code=)

                  func(request, *args, **kwargs)
             wrapper
         decorator


user_store = UserStore()
checker = PermissionChecker(user_store)



  ():
    
    

Step 4: Model Access Control

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

class ModelAccessController:
    """Control access to LLM models based on permissions."""

    MODEL_PERMISSIONS = {
        "gpt-4o": Permission.MODEL_GPT4,
        "gpt-4o-mini": Permission.MODEL_GPT4_MINI,
        "claude-3-5-sonnet-20241022": Permission.MODEL_CLAUDE,
    }

    def __init__(self, user_store: UserStore):
        self.user_store = user_store

    def get_allowed_models(self, user_id: str) -> List[str]:
        """Get list of models user can access."""
        user = self.user_store.users.get(user_id)
        if not user:
            return []

        permissions = user.get_permissions()
        tenant = self.user_store.get_user_tenant(user_id)

        allowed = []
        for model, permission in self.MODEL_PERMISSIONS.items():
            if permission in permissions:
                # Also check tenant restrictions
                if tenant and tenant.allowed_models:
                    if model in tenant.allowed_models:
                        allowed.append(model)
                else:
                    allowed.append(model)

        return allowed

    def create_llm():
        
        allowed = .get_allowed_models(user_id)

          allowed:
             PermissionError()

        
        model = model  allowed[]

         model   allowed:
             PermissionError()

         model.startswith():
             ChatOpenAI(model=model)
         model.startswith():
             ChatAnthropic(model=model)
        :
             ValueError()

Step 5: Tenant Isolation

from langchain_core.callbacks import BaseCallbackHandler
from contextvars import ContextVar

# Context variable for current tenant
current_tenant: ContextVar[str] = ContextVar("current_tenant")

class TenantIsolationMiddleware:
    """Middleware to enforce tenant isolation."""

    def __init__(self, user_store: UserStore):
        self.user_store = user_store

    async def __call__(self, request: Request, call_next):
        user_id = request.state.user_id
        user = self.user_store.users.get(user_id)

        if user:
            # Set tenant context
            token = current_tenant.set(user.tenant_id)
            try:
                response = await call_next(request)
            finally:
                current_tenant.reset(token)
            return response

        return await call_next(request)

class TenantAwareCallback(BaseCallbackHandler):
    """Tag all LLM calls with tenant ID."""

    def on_llm_start(self, serialized, prompts, **kwargs):
        tenant_id = current_tenant.get(None)
        if tenant_id:
            # Add tenant to metadata for billing/logging
            kwargs.setdefault("metadata", {})["tenant_id"] = tenant_id


 :
    

     ():
        .base_store = base_store

     ():
        tenant_id = current_tenant.get()
          tenant_id:
             ValueError()

        
        kwargs[] = kwargs.get(, {})
        kwargs[][] = tenant_id

         .base_store.similarity_search(query, **kwargs)

Step 6: Usage Quotas

from datetime import datetime, timedelta
from collections import defaultdict

class UsageQuotaManager:
    """Manage usage quotas per user and tenant."""

    def __init__(self, user_store: UserStore):
        self.user_store = user_store
        self.usage = defaultdict(lambda: {"tokens": 0, "requests": 0})
        self.reset_time = {}

    def check_quota(self, user_id: str, tokens: int = 0) -> bool:
        """Check if user has available quota."""
        user = self.user_store.users.get(user_id)
        tenant = self.user_store.get_user_tenant(user_id)

        if not tenant:
            return False

        # Reset monthly quota if needed
        self._maybe_reset(tenant.id)

        current = self.usage[tenant.id]["tokens"]
        return (current + tokens) <= tenant.monthly_token_limit

    def record_usage(self, user_id: str, tokens: int) -> None:
        """Record token usage."""
        tenant = .user_store.get_user_tenant(user_id)
         tenant:
            .usage[tenant.][] += tokens
            .usage[tenant.][] += 

     () -> :
        
        now = datetime.now()
        last_reset = .reset_time.get(tenant_id)

          last_reset  last_reset.month != now.month:
            .usage[tenant_id] = {: , : }
            .reset_time[tenant_id] = now

     () -> :
        
        tenant = .user_store.tenants.get(tenant_id)
        usage = .usage[tenant_id]

         {
            : tenant_id,
            : usage[],
            : tenant.monthly_token_limit  tenant  ,
            : usage[],
            : (usage[] / tenant.monthly_token_limit * )  tenant  
        }

Output

• Permission model with roles
• User and tenant management
• Model access control
• Tenant isolation
• Usage quotas

Resources

• RBAC Best Practices
• Multi-Tenant Architecture
• OAuth 2.0 Scopes

Next Steps

Use langchain-data-handling for data privacy controls.