Security & Authorization

ContextBrain enforces security at multiple layers. No data operation can bypass tenant isolation.

Two-Layer Defense

Layer 1: gRPC Permission Interceptor

BrainPermissionInterceptor maps every RPC method to a required permission before the handler executes:

RPC_PERMISSION_MAP = {
    # Knowledge
    "Search": Permissions.BRAIN_READ,
    "Upsert": Permissions.BRAIN_WRITE,
    "GraphSearch": Permissions.BRAIN_READ,
    "CreateKGRelation": Permissions.BRAIN_WRITE,
    # Memory
    "AddEpisode": Permissions.MEMORY_WRITE,
    "GetRecentEpisodes": Permissions.MEMORY_READ,
    "UpsertFact": Permissions.MEMORY_WRITE,
    "GetUserFacts": Permissions.MEMORY_READ,
    # Traces
    "LogTrace": Permissions.TRACE_WRITE,
    "GetTraces": Permissions.TRACE_READ,
}

Unmapped RPCs log a warning and are denied by default.

Layer 2: Handler Authorization

Inside each handler, validate_token_for_read() / validate_token_for_write() enforce tenant binding:

from contextunity.brain.service.helpers import validate_token_for_read

async def Search(self, request, context):
    token = validate_token_for_read(context, params.tenant_id)
    # Token validated: possesses 'brain:read' AND can access this tenant

Domain Permissions

Domain	Read	Write
Knowledge	brain:read	brain:write
Memory	memory:read	memory:write
Traces	trace:read	trace:write
Commerce	brain:read	brain:write
News	brain:read	brain:write

Row-Level Security (PostgreSQL)

Even if application-level checks are circumvented, PostgreSQL RLS prevents cross-tenant data leakage. Every database operation sets session scope:

SET LOCAL app.current_tenant = '{tenant_id}';
SET LOCAL app.current_user = '{user_id}';

Database Roles

Role	RLS	Purpose
brain_app	Enforced	Used by ContextBrain service
brain_admin	Bypassed	Used by ContextView dashboard

The brain_app role has Row-Level Security enforced. Only brain_admin can bypass it (wildcard '*' tenant for dashboard access).

Error Handling

Brain uses @grpc_error_handler to automatically map exceptions to gRPC status codes:

Exception	gRPC Status	Description
StorageError	UNAVAILABLE	Database query or connection failure
SecurityError	PERMISSION_DENIED	Token validation failure
ValidationError	INVALID_ARGUMENT	Invalid request parameters
ContextbrainError	INTERNAL	Unclassified service error

from contextunity.core.exceptions import grpc_error_handler

@grpc_error_handler
async def Search(self, request, context):
    # Exceptions auto-mapped to gRPC status codes
    ...