Langchain Enterprise Rbac

LangChain Enterprise RBAC

Overview

Role-based access control for multi-tenant LangChain applications: permission models, model access gating, tenant-scoped vector stores, usage quotas, and FastAPI middleware integration.

Permission Model

// permissions.ts
enum Permission {
  CHAIN_INVOKE = "chain:invoke",
  CHAIN_STREAM = "chain:stream",
  MODEL_GPT4O = "model:gpt-4o",
  MODEL_GPT4O_MINI = "model:gpt-4o-mini",
  MODEL_CLAUDE = "model:claude-sonnet",
  TOOLS_EXECUTE = "tools:execute",
  ADMIN_CONFIG = "admin:config",
  ADMIN_USERS = "admin:users",
}

interface Role {
  name: string;
  permissions: Permission[];
}

const ROLES: Record<string, Role> = {
  viewer: {
    name: "viewer",
    permissions: [Permission.CHAIN_INVOKE, Permission.MODEL_GPT4O_MINI],
  },
  user: {
    name: "user",
    permissions: [
      Permission.CHAIN_INVOKE,
      Permission.CHAIN_STREAM,
      Permission.MODEL_GPT4O_MINI,
      Permission.TOOLS_EXECUTE,
    ],
  },
  power_user: {
    name: "power_user",
    permissions: [
      Permission.CHAIN_INVOKE,
      Permission.CHAIN_STREAM,
      Permission.MODEL_GPT4O_MINI,
      Permission.MODEL_GPT4O,
      Permission.MODEL_CLAUDE,
      Permission.TOOLS_EXECUTE,
    ],
  },
  admin: {
    name: "admin",
    permissions: Object.values(Permission),
  },
};

User and Tenant Models

interface Tenant {
  id: string;
  name: string;
  monthlyTokenLimit: number;
  tokensUsed: number;
  allowedModels: string[];
}

interface User {
  id: string;
  tenantId: string;
  role: string;
  email: string;
}

function hasPermission(user: User, permission: Permission): boolean {
  const role = ROLES[user.role];
  if (!role) return false;
  return role.permissions.includes(permission);
}

function canUseModel(user: User, tenant: Tenant, model: string): boolean {
  const modelPermission = `model:${model}` as Permission;
  return (
    hasPermission(user, modelPermission) &&
    tenant.allowedModels.includes(model)
  );
}

Middleware: Permission Enforcement

// Express middleware
import { Request, Response, NextFunction } from "express";

function requirePermission(permission: Permission) {
  return (req: Request, res: Response, next: NextFunction) => {
    const user = req.user as User;  // set by auth middleware
    if (!user) {
      return res.status(401).json({ error: "Authentication required" });
    }
    if (!hasPermission(user, permission)) {
      return res.status(403).json({
        error: `Missing permission: ${permission}`,
        role: user.role,
      });
    }
    next();
  };
}

// Usage
app.post("/api/chat",
  requirePermission(Permission.CHAIN_INVOKE),
  async (req, res) => {
    const result = await chain.invoke({ input: req.body.input });
    res.json({ result });
  }
);

app.post("/api/chat/stream",
  requirePermission(Permission.CHAIN_STREAM),
  async (req, res) => {
    // streaming endpoint...
  }
);

Model Access Control

import { ChatOpenAI } from "@langchain/openai";
import { ChatAnthropic } from "@langchain/anthropic";

class ModelGateway {
  private models: Record<string, any> = {};

  constructor() {
    this.models["gpt-4o-mini"] = new ChatOpenAI({ model: "gpt-4o-mini" });
    this.models["gpt-4o"] = new ChatOpenAI({ model: "gpt-4o" });
    this.models["claude-sonnet"] = new ChatAnthropic({ model: "claude-sonnet-4-20250514" });
  }

  getModel(modelName: string, user: User, tenant: Tenant) {
    if (!canUseModel(user, tenant, modelName)) {
      throw new Error(
        `User ${user.email} (role: ${user.role}) cannot access model: ${modelName}`
      );
    }

    if (tenant.tokensUsed >= tenant.monthlyTokenLimit) {
      throw new Error(
        `Tenant ${tenant.name} exceeded monthly token limit (${tenant.monthlyTokenLimit})`
      );
    }

    return this.models[modelName];
  }
}

Tenant-Scoped Vector Store

// Isolate each tenant's documents in the vector store
import { PineconeStore } from "@langchain/pinecone";

class TenantScopedRetriever {
  constructor(
    private vectorStore: PineconeStore,
    private tenantId: string,
  ) {}

  async retrieve(query: string, k = 4) {
    // Filter by tenant ID — prevents cross-tenant data leakage
    return this.vectorStore.similaritySearch(query, k, {
      tenantId: { $eq: this.tenantId },
    });
  }

  asRetriever(k = 4) {
    return this.vectorStore.asRetriever({
      k,
      filter: { tenantId: { $eq: this.tenantId } },
    });
  }
}

// When ingesting documents, always tag with tenant ID
async function ingestForTenant(tenantId: string, docs: any[]) {
  const tagged = docs.map((doc) => ({
    ...doc,
    metadata: { ...doc.metadata, tenantId },
  }));
  await vectorStore.addDocuments(tagged);
}

Usage Quota Tracking

class UsageQuotaManager {
  private usage = new Map<string, number>();

  async trackUsage(tenantId: string, tokens: number) {
    const current = this.usage.get(tenantId) ?? 0;
    this.usage.set(tenantId, current + tokens);
  }

  async checkQuota(tenant: Tenant): Promise<boolean> {
    const used = this.usage.get(tenant.id) ?? 0;
    return used < tenant.monthlyTokenLimit;
  }

  async getUsageReport(tenantId: string) {
    return {
      tenantId,
      tokensUsed: this.usage.get(tenantId) ?? 0,
      period: new Date().toISOString().slice(0, 7), // YYYY-MM
    };
  }
}

// Integrate with callback
class QuotaCallback extends BaseCallbackHandler {
  name = "QuotaCallback";

  constructor(
    private quotaManager: UsageQuotaManager,
    private tenantId: string,
  ) { super(); }

  async handleLLMEnd(output: any) {
    const tokens = output.llmOutput?.tokenUsage?.totalTokens ?? 0;
    await this.quotaManager.trackUsage(this.tenantId, tokens);
  }
}

Python FastAPI Equivalent

from fastapi import Depends, HTTPException

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_jwt(token)
    if not user:
        raise HTTPException(status_code=401)
    return user

def require_permission(permission: str):
    async def checker(user = Depends(get_current_user)):
        if permission not in ROLES[user.role]["permissions"]:
            raise HTTPException(status_code=403, detail=f"Missing: {permission}")
        return user
    return checker

@app.post("/api/chat")
async def chat(input: dict, user = Depends(require_permission("chain:invoke"))):
    return await chain.ainvoke({"input": input["text"]})

Error Handling

Resources

• RBAC Best Practices (Auth0)
• Multi-Tenant Architecture
• OAuth 2.0 Scopes

Next Steps

Use langchain-data-handling for tenant-scoped RAG pipelines.

Source: flight505/skill-forge — distributed by TomeVault.