Protect a list endpoint

Protecting list endpoints is trickier than checking access to a single resource. Your endpoint might return 10,000 items, but the user can only see 50. You need to filter the list efficiently without checking every single item. This guide shows you the patterns you’ll see in production.

Prerequisites

This guide assumes you’ve already read Protect an endpoint and understand how to use Check() and CheckForUpdate() for individual permission checks.

The challenge with lists

When a user requests a list of resources, you need to answer: “Which of these can this user access?” You have two fundamental approaches:

1. Pre-filtering: Find out what the user can access, then query only those resources
2. Post-filtering: Fetch resources first, then check each one for access

Each approach has coarse-grained (workspace-level) and fine-grained (per-resource) variations. Pick the one that fits your data model and performance needs.

Pre-filtering (recommended)

Pre-filtering reduces database load by querying only what the user can access. It provides filter criteria to your database query before fetching results.

Coarse-grained: Workspace-level filtering

Uses StreamedListObjects to find all workspaces where the user has the required permission, then queries your database for resources in those workspaces.

How it works:

flowchart LR
    A[User requests list] --> B[StreamedListObjects:<br/>Get accessible workspaces]
    B --> C[Query DB:<br/>workspace_id IN ...]
    C --> D[Return filtered results]

    style A fill:#e1f5ff,stroke:#333,color:#000
    style B fill:#fff4e1,stroke:#333,color:#000
    style C fill:#fff4e1,stroke:#333,color:#000
    style D fill:#e8f5e9,stroke:#333,color:#000

When to use:

• Your resources are assigned to workspaces
• Workspace membership is the primary access control mechanism
• High-cardinality resources (thousands of items or more) — per-resource checks get expensive

Implementation:

Python

def get_accessible_workspaces(stub, user_id: str, permission: str):
    """Get all workspaces the user can access with the given permission.

    This uses the list_workspaces helper from kessel.rbac.v2 which
    handles StreamedListObjects pagination automatically.
    """
    subject = principal_subject(id=user_id, domain="redhat")

    workspace_ids = []
    for response in list_workspaces(stub, subject, permission):
        workspace_ids.append(response.object.resource_id)

    return workspace_ids

Go

// getAccessibleWorkspaces returns all workspaces the user can access with the given permission.
//
// This uses the ListWorkspaces helper from kessel.rbac.v2 which
// handles StreamedListObjects pagination automatically.
func getAccessibleWorkspaces(ctx context.Context, client v1beta2.KesselInventoryServiceClient, userID, permission string) ([]string, error) {
  subject := v2.PrincipalSubject(userID, "redhat")

  var workspaceIDs []string
  for response, err := range v2.ListWorkspaces(ctx, client, subject, permission, "") {
    if err != nil {
      return nil, fmt.Errorf("failed to list workspaces: %w", err)
    }
    workspaceIDs = append(workspaceIDs, response.Object.ResourceId)
  }

  return workspaceIDs, nil
}

TypeScript

/**
 * Get all workspaces the user can access with the given permission.
 *
 * This uses the listWorkspaces helper from kessel.rbac.v2 which
 * handles StreamedListObjects pagination automatically.
 */
async function getAccessibleWorkspaces(userId: string, permission: string): Promise<string[]> {
  const subject = principalSubject(userId, "redhat");

  const workspaceIds: string[] = [];
  for await (const response of listWorkspaces(client, subject, permission)) {
    if (response.object?.resourceId) {
      workspaceIds.push(response.object.resourceId);
    }
  }

  return workspaceIds;
}

Ruby

# Get all workspaces the user can access with the given permission.
#
# This uses the list_workspaces helper from Kessel::RBAC::V2 which
# handles StreamedListObjects pagination automatically.
def get_accessible_workspaces(client, user_id, permission)
  subject = principal_subject(user_id, "redhat")

  workspace_ids = []
  list_workspaces(client, subject, permission).each do |response|
    workspace_ids << response.object.resource_id
  end

  workspace_ids
end

Java

/**
 * Get all workspaces the user can access with the given permission.
 *
 * This uses the ListWorkspaces helper from kessel.rbac.v2 which
 * handles StreamedListObjects pagination automatically.
 */
public List<String> getAccessibleWorkspaces(String userId, String permission) {
    List<String> workspaceIds = new ArrayList<>();

    Iterable<StreamedListObjectsResponse> workspaces = ListWorkspaces.listWorkspaces(
        kesselClient,
        Utils.principalSubject(userId, "redhat"),
        permission
    );

    for (StreamedListObjectsResponse response : workspaces) {
        workspaceIds.add(response.getObject().getResourceId());
    }

    return workspaceIds;
}

Then use those workspace IDs to filter your database query:

Python

from sqlalchemy import select
from sqlalchemy.orm import Session

def list_integrations(session: Session, stub, user_id: str):
    """List integrations the user can access.

    Pre-filtering: Get allowed workspaces first, then query only those.
    """
    # Step 1: Get workspaces the user can access
    allowed_workspaces = get_accessible_workspaces(
        stub, user_id, "myservice_integration_view"
    )

    if not allowed_workspaces:
        return []

    # Step 2: Query database with workspace filter
    query = select(Integration).where(
        Integration.workspace_id.in_(allowed_workspaces)
    )

    result = session.execute(query)
    return result.scalars().all()

Go

// Integration represents an integration in your database
type Integration struct {
  ID          string
  Name        string
  WorkspaceID string
}

// Database is a placeholder for your database interface
type Database interface {
  Query(query string, args ...interface{}) ([]*Integration, error)
}

// listIntegrations returns integrations the user can access.
//
// Pre-filtering: Get allowed workspaces first, then query only those.
func listIntegrations(ctx context.Context, client v1beta2.KesselInventoryServiceClient, db Database, userID string) ([]*Integration, error) {
  // Step 1: Get workspaces the user can access
  allowedWorkspaces, err := getAccessibleWorkspaces(ctx, client, userID, "myservice_integration_view")
  if err != nil {
    return nil, err
  }

  if len(allowedWorkspaces) == 0 {
    return []*Integration{}, nil
  }

  // Step 2: Query database with workspace filter
  query := "SELECT id, name, workspace_id FROM integrations WHERE workspace_id = ANY($1)"
  integrations, err := db.Query(query, allowedWorkspaces)
  if err != nil {
    return nil, fmt.Errorf("failed to query integrations: %w", err)
  }

  return integrations, nil
}

TypeScript

/**
 * List integrations the user can access.
 *
 * Pre-filtering: Get allowed workspaces first, then query only those.
 */
async function listIntegrations(db: Database, userId: string): Promise<Integration[]> {
  // Step 1: Get workspaces the user can access
  const allowedWorkspaces = await getAccessibleWorkspaces(userId, "myservice_integration_view");

  if (allowedWorkspaces.length === 0) {
    return [];
  }

  // Step 2: Query database with workspace filter
  const integrations = await db.query(
    "SELECT * FROM integrations WHERE workspace_id = ANY($1)",
    [allowedWorkspaces]
  );

  return integrations;
}

// Database interface placeholder
interface Database {
  query(sql: string, params: any[]): Promise<Integration[]>;
}

interface Integration {
  id: string;
  name: string;
  workspaceId: string;
}

Ruby

# List integrations the user can access.
#
# Pre-filtering: Get allowed workspaces first, then query only those.
def list_integrations(db, client, user_id)
  # Step 1: Get workspaces the user can access
  allowed_workspaces = get_accessible_workspaces(client, user_id, "myservice_integration_view")

  return [] if allowed_workspaces.empty?

  # Step 2: Query database with workspace filter
  db.exec_params(
    "SELECT * FROM integrations WHERE workspace_id = ANY($1)",
    [allowed_workspaces]
  ).to_a
end

Java

/**
 * List integrations the user can access.
 *
 * Pre-filtering: Get allowed workspaces first, then query only those.
 */
public List<Integration> listIntegrations(Database db, String userId) {
    // Step 1: Get workspaces the user can access
    List<String> allowedWorkspaces = getAccessibleWorkspaces(userId, "myservice_integration_view");

    if (allowedWorkspaces.isEmpty()) {
        return List.of();
    }

    // Step 2: Query database with workspace filter
    String sql = "SELECT * FROM integrations WHERE workspace_id = ANY(?)";
    return db.query(sql, allowedWorkspaces);
}

// Database interface placeholder
interface Database {
    List<Integration> query(String sql, List<String> params);
}

static class Integration {
    String id;
    String name;
    String workspaceId;
}

What it costs:

• Initial cost: One Kessel API call to get workspace list
• Database cost: Single filtered query (efficient with proper indexes)
• Scales well: Performance depends on how many workspaces the user can access, not your total resource count

Tip

The workspace list result can be cached for the duration of the request. If you have multiple list endpoints in the same request flow, call StreamedListObjects once and reuse the workspace IDs.

Fine-grained: Resource-level filtering

Uses StreamedListObjects with your specific resource type to find exactly which resource IDs the user can access, then queries your database for only those resources.

When to use:

• Resources have permissions independent of workspace hierarchy
• Access is granted on individual resources, not workspace-level
• The accessible resource count stays under a few thousand

How it works:

1. Use StreamedListObjects with your resource type (not workspace) and permission
2. Get back a list of specific resource IDs the user can access
3. Query your database with WHERE id IN (...) for those exact IDs

Implementation:

Python

def get_accessible_resource_ids(stub, user_id: str, permission: str):
    """Get all resource IDs the user can access with the given permission.

    This uses StreamedListObjects with your specific resource type to find
    exactly which resource IDs the user can access.
    """
    subject = principal_subject(id=user_id, domain="redhat")
    resource_type = representation_type_pb2.RepresentationType(
        resource_type="integration",
        reporter_type="myservice"
    )

    request = streamed_list_objects_request_pb2.StreamedListObjectsRequest(
        object_type=resource_type,
        relation=permission,
        subject=subject
    )

    # Get accessible resource IDs
    resource_ids = []
    for response in stub.StreamedListObjects(request):
        resource_ids.append(response.object.resource_id)

    return resource_ids

Go

// getAccessibleResourceIDs gets all resource IDs the user can access with the given permission.
//
// This uses StreamedListObjects with your specific resource type to find
// exactly which resource IDs the user can access.
func getAccessibleResourceIDs(
  ctx context.Context,
  client v1beta2.KesselInventoryServiceClient,
  userID string,
  permission string,
) ([]string, error) {
    subject := v2.PrincipalSubject(userID, "redhat")
  reporterType := "myservice"
  resourceType := &v1beta2.RepresentationType{
    ResourceType: "integration",
    ReporterType: &reporterType,
  }

  request := &v1beta2.StreamedListObjectsRequest{
    ObjectType: resourceType,
    Relation:   permission,
    Subject:    subject,
  }

  stream, err := client.StreamedListObjects(ctx, request)
  if err != nil {
    return nil, fmt.Errorf("failed to start stream: %w", err)
  }

  // Get accessible resource IDs
  var resourceIDs []string
  for {
    response, err := stream.Recv()
    if err == io.EOF {
      break
    }
    if err != nil {
      return nil, fmt.Errorf("error receiving from stream: %w", err)
    }
    resourceIDs = append(resourceIDs, response.Object.ResourceId)
  }

  return resourceIDs, nil
}

TypeScript

/**
 * Get all resource IDs the user can access with the given permission.
 *
 * This uses StreamedListObjects with your specific resource type to find
 * exactly which resource IDs the user can access.
 */
async function getAccessibleResourceIds(userId: string, permission: string): Promise<string[]> {
  const subject = principalSubject(userId, "redhat");
  const resourceType: RepresentationType = {
    resourceType: "integration",
    reporterType: "myservice"
  };

  const request: StreamedListObjectsRequest = {
    objectType: resourceType,
    relation: permission,
    subject: subject
  };

  // Get accessible resource IDs
  const resourceIds: string[] = [];
  for await (const response of client.streamedListObjects(request)) {
    if (response.object?.resourceId) {
      resourceIds.push(response.object.resourceId);
    }
  }

  return resourceIds;
}

Ruby

# Get all resource IDs the user can access with the given permission.
#
# This uses StreamedListObjects with your specific resource type to find
# exactly which resource IDs the user can access.
def get_accessible_resource_ids(client, user_id, permission)
  subject = principal_subject(user_id, "redhat")
  resource_type = RepresentationType.new(
    resource_type: "integration",
    reporter_type: "myservice"
  )

  request = StreamedListObjectsRequest.new(
    object_type: resource_type,
    relation: permission,
    subject: subject
  )

  # Get accessible resource IDs
  resource_ids = []
  client.streamed_list_objects(request).each do |response|
    resource_ids << response.object.resource_id
  end

  resource_ids
end

Java

/**
 * Get all resource IDs the user can access with the given permission.
 *
 * This uses StreamedListObjects with your specific resource type to find
 * exactly which resource IDs the user can access.
 */
public List<String> getAccessibleResourceIDs(String userId, String permission) {
    SubjectReference subject = Utils.principalSubject(userId, "redhat");
    RepresentationType resourceType = RepresentationType.newBuilder()
        .setResourceType("integration")
        .setReporterType("myservice")
        .build();

    StreamedListObjectsRequest request = StreamedListObjectsRequest.newBuilder()
        .setObjectType(resourceType)
        .setRelation(permission)
        .setSubject(subject)
        .build();

    // Get accessible resource IDs
    List<String> resourceIDs = new ArrayList<>();
    Iterator<StreamedListObjectsResponse> responses = kesselClient.streamedListObjects(request);
    while (responses.hasNext()) {
        StreamedListObjectsResponse response = responses.next();
        resourceIDs.add(response.getObject().getResourceId());
    }

    return resourceIDs;
}

Then use those resource IDs to filter your database query:

Python

from sqlalchemy import select
from sqlalchemy.orm import Session

def list_integrations(session: Session, stub, user_id: str):
    """List integrations the user can access.

    Pre-filtering (fine-grained): Get allowed resource IDs first, then query only those.
    """
    # Step 1: Get resource IDs the user can access
    allowed_ids = get_accessible_resource_ids(stub, user_id, "view")

    if not allowed_ids:
        return []

    # Step 2: Query database with resource ID filter
    query = select(Integration).where(Integration.id.in_(allowed_ids))
    result = session.execute(query)
    return result.scalars().all()

Go

// Integration represents an integration in your database
type Integration struct {
  ID          string
  Name        string
  WorkspaceID string
}

// Database is a placeholder for your database interface
type Database interface {
  Query(query string, params ...interface{}) ([]*Integration, error)
}

// listIntegrations returns integrations the user can access.
//
// Pre-filtering (fine-grained): Get allowed resource IDs first, then query only those.
func listIntegrations(ctx context.Context, client v1beta2.KesselInventoryServiceClient, db Database, userID string) ([]*Integration, error) {
  // Step 1: Get resource IDs the user can access
  allowedIDs, err := getAccessibleResourceIDs(ctx, client, userID, "view")
  if err != nil {
    return nil, err
  }

  if len(allowedIDs) == 0 {
    return []*Integration{}, nil
  }

  // Step 2: Query database with resource ID filter
  query := "SELECT id, name, workspace_id FROM integrations WHERE id = ANY($1)"
  integrations, err := db.Query(query, allowedIDs)
  if err != nil {
    return nil, fmt.Errorf("failed to query integrations: %w", err)
  }

  return integrations, nil
}

TypeScript

/**
 * List integrations the user can access.
 *
 * Pre-filtering (fine-grained): Get allowed resource IDs first, then query only those.
 */
async function listIntegrations(db: Database, userId: string): Promise<Integration[]> {
  // Step 1: Get resource IDs the user can access
  const allowedIds = await getAccessibleResourceIds(userId, "view");

  if (allowedIds.length === 0) {
    return [];
  }

  // Step 2: Query database with resource ID filter
  const integrations = await db.query(
    "SELECT * FROM integrations WHERE id = ANY($1)",
    [allowedIds]
  );

  return integrations;
}

// Database interface placeholder
interface Database {
  query(sql: string, params: any[]): Promise<Integration[]>;
}

interface Integration {
  id: string;
  name: string;
  workspaceId: string;
}

Ruby

# List integrations the user can access.
#
# Pre-filtering (fine-grained): Get allowed resource IDs first, then query only those.
def list_integrations(db, client, user_id)
  # Step 1: Get resource IDs the user can access
  allowed_ids = get_accessible_resource_ids(client, user_id, "view")

  return [] if allowed_ids.empty?

  # Step 2: Query database with resource ID filter
  db.exec_params(
    "SELECT * FROM integrations WHERE id = ANY($1)",
    [allowed_ids]
  ).to_a
end

Java

/**
 * List integrations the user can access.
 *
 * Pre-filtering (fine-grained): Get allowed resource IDs first, then query only those.
 */
public List<Integration> listIntegrations(Database db, String userId) {
    // Step 1: Get resource IDs the user can access
    List<String> allowedIDs = getAccessibleResourceIDs(userId, "view");

    if (allowedIDs.isEmpty()) {
        return List.of();
    }

    // Step 2: Query database with resource ID filter
    String sql = "SELECT * FROM integrations WHERE id = ANY(?)";
    return db.query(sql, allowedIDs);
}

// Database interface placeholder
interface Database {
    List<Integration> query(String sql, List<String> params);
}

static class Integration {
    String id;
    String name;
    String workspaceId;
}

Speed tradeoffs:

• Kessel call: One API call to get the resource IDs
• Database: WHERE id IN (...) query — fast with an index
• Watch out: If the user can access most of your resources anyway, workspace-level filtering is simpler

Caution

Fine-grained pre-filtering works best when the accessible resource set is bounded. If a user has access to tens of thousands of specific resources, consider coarse-grained filtering instead.

Post-filtering

Post-filtering queries your database first, then checks permissions on the results before returning them to the client. Simpler to implement, but watch out — it gets inefficient with large result sets.

Coarse-grained: Workspace-level check

Query the database first, then check if the user has permission on the workspace(s) that each resource belongs to. Filter out resources where the user lacks workspace permission.

When to use:

• You get the data as-is with no way to filter upfront — Kafka consumers, batch processors, or fixed API responses
• Resources are assigned to workspaces (have a workspace_id field)
• The result set is small enough that checking a few workspace permissions is acceptable

How it works:

1. Query database and get all matching resources
2. Collect the unique workspace IDs from the resources
3. Use CheckBulk() to verify permission on those workspaces
4. Filter out resources where the user doesn’t have workspace permission

Implementation:

Python

def filter_by_workspace_permission(integrations, user_id: str, permission: str):
    """Filter integrations by checking workspace permissions.

    Post-filtering (coarse): Check workspace-level permissions for unique workspaces.
    """
    # Step 1: Get unique workspace IDs
    workspace_ids = list(set(i.workspace_id for i in integrations))

    # Step 2: Check workspace permissions with CheckBulk
    items = [
        check_bulk_request_pb2.CheckBulkRequestItem(
            object=workspace_resource(ws_id),
            relation=permission,
            subject=principal_subject(id=user_id, domain="redhat")
        )
        for ws_id in workspace_ids
    ]
    request = check_bulk_request_pb2.CheckBulkRequest(items=items)
    response = stub.CheckBulk(request)

    # Step 3: Build set of allowed workspaces
    allowed_workspaces = set()
    for index, pair in enumerate(response.pairs):
        if pair.item.allowed == allowed_pb2.ALLOWED_TRUE:
            allowed_workspaces.add(workspace_ids[index])

    # Step 4: Filter resources by allowed workspaces
    return [i for i in integrations if i.workspace_id in allowed_workspaces]

Go

// Integration represents an integration in your database
type Integration struct {
  ID          string
  Name        string
  WorkspaceID string
}

// filterByWorkspacePermission filters integrations by checking workspace permissions.
//
// Post-filtering (coarse): Check workspace-level permissions for unique workspaces.
func filterByWorkspacePermission(
  ctx context.Context,
  client v1beta2.KesselInventoryServiceClient,
  integrations []*Integration,
  userID string,
  permission string,
) ([]*Integration, error) {
  // Step 1: Get unique workspace IDs
  workspaceSet := make(map[string]bool)
  for _, integration := range integrations {
    workspaceSet[integration.WorkspaceID] = true
  }
  var workspaceIDs []string
  for wsID := range workspaceSet {
    workspaceIDs = append(workspaceIDs, wsID)
  }

  // Step 2: Check workspace permissions with CheckBulk
  var items []*v1beta2.CheckBulkRequestItem
  for _, wsID := range workspaceIDs {
    item := &v1beta2.CheckBulkRequestItem{
      Object:   v2.WorkspaceResource(wsID),
      Relation: permission,
      Subject:  v2.PrincipalSubject(userID, "redhat"),
    }
    items = append(items, item)
  }

  request := &v1beta2.CheckBulkRequest{Items: items}
  response, err := client.CheckBulk(ctx, request)
  if err != nil {
    return nil, fmt.Errorf("failed to check bulk: %w", err)
  }

  // Step 3: Build set of allowed workspaces
  allowedWorkspaces := make(map[string]bool)
  for index, pair := range response.Pairs {
    if item := pair.GetItem(); item != nil && item.Allowed == v1beta2.Allowed_ALLOWED_TRUE {
      allowedWorkspaces[workspaceIDs[index]] = true
    }
  }

  // Step 4: Filter resources by allowed workspaces
  var accessibleIntegrations []*Integration
  for _, integration := range integrations {
    if allowedWorkspaces[integration.WorkspaceID] {
      accessibleIntegrations = append(accessibleIntegrations, integration)
    }
  }

  return accessibleIntegrations, nil
}

TypeScript

/**
 * Filter integrations by checking workspace permissions.
 *
 * Post-filtering (coarse): Check workspace-level permissions for unique workspaces.
 */
async function filterByWorkspacePermission(
  integrations: Integration[],
  userId: string,
  permission: string
): Promise<Integration[]> {
  // Step 1: Get unique workspace IDs
  const workspaceIds = Array.from(new Set(integrations.map(i => i.workspaceId)));

  // Step 2: Check workspace permissions with CheckBulk
  const items: CheckBulkRequestItem[] = workspaceIds.map(wsId => ({
    object: workspaceResource(wsId),
    relation: permission,
    subject: principalSubject(userId, "redhat"),
  }));

  const request: CheckBulkRequest = { items };
  const response = await client.checkBulk(request);

  // Step 3: Build set of allowed workspaces
  const allowedWorkspaces = new Set<string>();
  response.pairs?.forEach((pair, index) => {
    if (pair.item?.allowed === Allowed.ALLOWED_TRUE) {
      allowedWorkspaces.add(workspaceIds[index]);
    }
  });

  // Step 4: Filter resources by allowed workspaces
  return integrations.filter(i => allowedWorkspaces.has(i.workspaceId));
}

Ruby

# Filter integrations by checking workspace permissions.
#
# Post-filtering (coarse): Check workspace-level permissions for unique workspaces.
def filter_by_workspace_permission(integrations, client, user_id, permission)
  # Step 1: Get unique workspace IDs
  workspace_ids = integrations.map { |i| i[:workspace_id] }.uniq

  # Step 2: Check workspace permissions with CheckBulk
  items = workspace_ids.map do |ws_id|
    CheckBulkRequestItem.new(
      object: workspace_resource(ws_id),
      relation: permission,
      subject: principal_subject(user_id, "redhat")
    )
  end

  request = CheckBulkRequest.new(items: items)
  response = client.check_bulk(request)

  # Step 3: Build set of allowed workspaces
  allowed_workspaces = Set.new
  response.pairs.each_with_index do |pair, index|
    if pair.item&.allowed == Allowed::ALLOWED_TRUE
      allowed_workspaces << workspace_ids[index]
    end
  end

  # Step 4: Filter resources by allowed workspaces
  integrations.select { |i| allowed_workspaces.include?(i[:workspace_id]) }
end

Java

/**
 * Filter integrations by checking workspace permissions.
 *
 * Post-filtering (coarse): Check workspace-level permissions for unique workspaces.
 */
public List<Integration> filterByWorkspacePermission(
        List<Integration> integrations,
        String userId,
        String permission) {

    // Step 1: Get unique workspace IDs
    Set<String> workspaceSet = new HashSet<>();
    for (Integration integration : integrations) {
        workspaceSet.add(integration.workspaceId);
    }
    List<String> workspaceIDs = new ArrayList<>(workspaceSet);

    // Step 2: Check workspace permissions with CheckBulk
    CheckBulkRequest.Builder requestBuilder = CheckBulkRequest.newBuilder();
    for (String wsId : workspaceIDs) {
        CheckBulkRequestItem item = CheckBulkRequestItem.newBuilder()
            .setObject(Utils.workspaceResource(wsId))
            .setRelation(permission)
            .setSubject(Utils.principalSubject(userId, "redhat"))
            .build();
        requestBuilder.addItems(item);
    }

    CheckBulkResponse response = kesselClient.checkBulk(requestBuilder.build());

    // Step 3: Build set of allowed workspaces
    Set<String> allowedWorkspaces = new HashSet<>();
    for (int index = 0; index < response.getPairsCount(); index++) {
        CheckBulkResponsePair pair = response.getPairs(index);
        if (pair.hasItem() && pair.getItem().getAllowed() == Allowed.ALLOWED_TRUE) {
            allowedWorkspaces.add(workspaceIDs.get(index));
        }
    }

    // Step 4: Filter resources by allowed workspaces
    List<Integration> accessibleIntegrations = new ArrayList<>();
    for (Integration integration : integrations) {
        if (allowedWorkspaces.contains(integration.workspaceId)) {
            accessibleIntegrations.add(integration);
        }
    }

    return accessibleIntegrations;
}

Full example:

Python

from sqlalchemy import select
from sqlalchemy.orm import Session

def list_integrations(session: Session, stub, user_id: str):
    """List integrations the user can access.

    Post-filtering (coarse): Query all integrations, then check workspace permissions.
    """
    # Step 1: Query database
    query = select(Integration).filter_by(status="active")
    result = session.execute(query)
    all_integrations = result.scalars().all()

    # Step 2: Filter by workspace permission
    accessible = filter_by_workspace_permission(
        all_integrations,
        user_id,
        "myservice_integration_view"
    )

    return accessible

Go

// Database is a placeholder for your database interface
type Database interface {
  Query(query string) ([]*Integration, error)
}

// listIntegrations returns integrations the user can access.
//
// Post-filtering (coarse): Query all integrations, then check workspace permissions.
func listIntegrations(ctx context.Context, client v1beta2.KesselInventoryServiceClient, db Database, userID string) ([]*Integration, error) {
  // Step 1: Query database
  allIntegrations, err := db.Query("SELECT * FROM integrations WHERE status = 'active'")
  if err != nil {
    return nil, fmt.Errorf("failed to query integrations: %w", err)
  }

  // Step 2: Filter by workspace permission
  accessible, err := filterByWorkspacePermission(
    ctx,
    client,
    allIntegrations,
    userID,
    "myservice_integration_view",
  )
  if err != nil {
    return nil, err
  }

  return accessible, nil
}

TypeScript

/**
 * List integrations the user can access.
 *
 * Post-filtering (coarse): Query all integrations, then check workspace permissions.
 */
async function listIntegrations(db: Database, userId: string): Promise<Integration[]> {
  // Step 1: Query database
  const allIntegrations = await db.query("SELECT * FROM integrations WHERE status = 'active'");

  // Step 2: Filter by workspace permission
  const accessible = await filterByWorkspacePermission(
    allIntegrations,
    userId,
    "myservice_integration_view"
  );

  return accessible;
}

// Database interface placeholder
interface Database {
  query(sql: string): Promise<Integration[]>;
}

interface Integration {
  id: string;
  name: string;
  workspaceId: string;
}

Ruby

# List integrations the user can access.
#
# Post-filtering (coarse): Query all integrations, then check workspace permissions.
def list_integrations(db, client, user_id)
  # Step 1: Query database
  all_integrations = db.exec("SELECT * FROM integrations WHERE status = 'active'").to_a

  # Step 2: Filter by workspace permission
  filter_by_workspace_permission(
    all_integrations,
    client,
    user_id,
    "myservice_integration_view"
  )
end

Java

/**
 * List integrations the user can access.
 *
 * Post-filtering (coarse): Query all integrations, then check workspace permissions.
 */
public List<Integration> listIntegrations(Database db, String userId) {
    // Step 1: Query database
    List<Integration> allIntegrations = db.query("SELECT * FROM integrations WHERE status = 'active'");

    // Step 2: Filter by workspace permission
    return filterByWorkspacePermission(
        allIntegrations,
        userId,
        "myservice_integration_view"
    );
}

// Database interface placeholder
interface Database {
    List<Integration> query(String sql);
}

static class Integration {
    String id;
    String name;
    String workspaceId;
}

The tradeoff:

• Database: Unfiltered query (might fetch resources user can’t access)
• Authorization: Check workspace permissions (fewer checks than per-resource)
• Network: One CheckBulk() call for all unique workspaces

Note

This is coarse-grained because you check workspace-level permissions. If the user typically has access to most workspaces in the result set, this is more efficient than pre-filtering.

Fine-grained: Batch check all results

Query your database first, then use CheckBulk to check permission on every returned resource. The middleware filters out any resources the user cannot access before returning to the client.

When to use:

• Small result sets (hundreds or low thousands)
• User likely has access to most results
• Resources have per-resource permissions (not just workspace-level)
• Pre-filtering isn’t possible due to permission model

Implementation:

Python

def filter_integrations_by_permission(integrations, user_id: str, permission: str):
    """Filter integrations using CheckBulk to batch permission checks.

    This is more efficient than calling Check() in a loop.
    """
    # Build bulk check request with one item per integration
    items = []
    for integration in integrations:
        item = check_bulk_request_pb2.CheckBulkRequestItem(
            object=workspace_resource(integration.workspace_id),
            relation=permission,
            subject=principal_subject(id=user_id, domain="redhat"),
        )
        items.append(item)

    request = check_bulk_request_pb2.CheckBulkRequest(items=items)

    # Make single API call to check all permissions
    response = stub.CheckBulk(request)

    # Filter integrations based on bulk check results
    accessible_integrations = []
    for index, pair in enumerate(response.pairs):
        if pair.item.allowed == allowed_pb2.ALLOWED_TRUE:
            accessible_integrations.append(integrations[index])

    return accessible_integrations

Go

// Integration represents an integration in your database
type Integration struct {
  ID          string
  Name        string
  WorkspaceID string
}

// filterIntegrationsByPermission filters integrations using CheckBulk to batch permission checks.
//
// This is more efficient than calling Check() in a loop.
func filterIntegrationsByPermission(
  ctx context.Context,
  client v1beta2.KesselInventoryServiceClient,
  integrations []*Integration,
  userID string,
  permission string,
) ([]*Integration, error) {
  // Build bulk check request with one item per integration
  var items []*v1beta2.CheckBulkRequestItem
  for _, integration := range integrations {
    item := &v1beta2.CheckBulkRequestItem{
      Object:   v2.WorkspaceResource(integration.WorkspaceID),
      Relation: permission,
      Subject:  v2.PrincipalSubject(userID, "redhat"),
    }
    items = append(items, item)
  }

  request := &v1beta2.CheckBulkRequest{Items: items}

  // Make single API call to check all permissions
  response, err := client.CheckBulk(ctx, request)
  if err != nil {
    return nil, fmt.Errorf("failed to check bulk: %w", err)
  }

  // Filter integrations based on bulk check results
  var accessibleIntegrations []*Integration
  for index, pair := range response.Pairs {
    if item := pair.GetItem(); item != nil && item.Allowed == v1beta2.Allowed_ALLOWED_TRUE {
      accessibleIntegrations = append(accessibleIntegrations, integrations[index])
    }
  }

  return accessibleIntegrations, nil
}

TypeScript

/**
 * Filter integrations using CheckBulk to batch permission checks.
 *
 * This is more efficient than calling Check() in a loop.
 */
async function filterIntegrationsByPermission(
  integrations: Integration[],
  userId: string,
  permission: string
): Promise<Integration[]> {
  // Build bulk check request with one item per integration
  const items: CheckBulkRequestItem[] = integrations.map((integration) => ({
    object: workspaceResource(integration.workspaceId),
    relation: permission,
    subject: principalSubject(userId, "redhat"),
  }));

  const request: CheckBulkRequest = { items };

  // Make single API call to check all permissions
  const response = await client.checkBulk(request);

  // Filter integrations based on bulk check results
  const accessibleIntegrations: Integration[] = [];
  response.pairs?.forEach((pair, index) => {
    if (pair.item?.allowed === Allowed.ALLOWED_TRUE) {
      accessibleIntegrations.push(integrations[index]);
    }
  });

  return accessibleIntegrations;
}

Ruby

# Filter integrations using CheckBulk to batch permission checks.
#
# This is more efficient than calling Check() in a loop.
def filter_integrations_by_permission(integrations, client, user_id, permission)
  # Build bulk check request with one item per integration
  items = integrations.map do |integration|
    CheckBulkRequestItem.new(
      object: workspace_resource(integration[:workspace_id]),
      relation: permission,
      subject: principal_subject(user_id, "redhat")
    )
  end

  request = CheckBulkRequest.new(items: items)

  # Make single API call to check all permissions
  response = client.check_bulk(request)

  # Filter integrations based on bulk check results
  accessible_integrations = []
  response.pairs.each_with_index do |pair, index|
    if pair.item&.allowed == Allowed::ALLOWED_TRUE
      accessible_integrations << integrations[index]
    end
  end

  accessible_integrations
end

Java

/**
 * Filter integrations using CheckBulk to batch permission checks.
 *
 * This is more efficient than calling Check() in a loop.
 */
public List<Integration> filterIntegrationsByPermission(
        List<Integration> integrations,
        String userId,
        String permission) {

    // Build bulk check request with one item per integration
    CheckBulkRequest.Builder requestBuilder = CheckBulkRequest.newBuilder();
    for (Integration integration : integrations) {
        CheckBulkRequestItem item = CheckBulkRequestItem.newBuilder()
            .setObject(Utils.workspaceResource(integration.workspaceId))
            .setRelation(permission)
            .setSubject(Utils.principalSubject(userId, "redhat"))
            .build();
        requestBuilder.addItems(item);
    }

    // Make single API call to check all permissions
    CheckBulkResponse response = kesselClient.checkBulk(requestBuilder.build());

    // Filter integrations based on bulk check results
    List<Integration> accessibleIntegrations = new ArrayList<>();
    for (int index = 0; index < response.getPairsCount(); index++) {
        CheckBulkResponsePair pair = response.getPairs(index);
        if (pair.hasItem() && pair.getItem().getAllowed() == Allowed.ALLOWED_TRUE) {
            accessibleIntegrations.add(integrations.get(index));
        }
    }

    return accessibleIntegrations;
}

Full example:

Python

from sqlalchemy import select
from sqlalchemy.orm import Session

def list_integrations(session: Session, stub, user_id: str):
    """List integrations the user can access.

    Post-filtering with CheckBulk: Query all integrations, then batch-check permissions.
    """
    # Step 1: Query all integrations from database
    query = select(Integration)
    result = session.execute(query)
    all_integrations = result.scalars().all()

    # Step 2: Use CheckBulk to filter by permission
    accessible = filter_integrations_by_permission(
        all_integrations,
        user_id,
        "myservice_integration_view"
    )

    return accessible

Go

// Database is a placeholder for your database interface
type Database interface {
  Query(query string) ([]*Integration, error)
}

// listIntegrations returns integrations the user can access.
//
// Post-filtering with CheckBulk: Query all integrations, then batch-check permissions.
func listIntegrations(ctx context.Context, client v1beta2.KesselInventoryServiceClient, db Database, userID string) ([]*Integration, error) {
  // Step 1: Query all integrations from database
  allIntegrations, err := db.Query("SELECT * FROM integrations")
  if err != nil {
    return nil, fmt.Errorf("failed to query integrations: %w", err)
  }

  // Step 2: Use CheckBulk to filter by permission
  accessible, err := filterIntegrationsByPermission(
    ctx,
    client,
    allIntegrations,
    userID,
    "myservice_integration_view",
  )
  if err != nil {
    return nil, err
  }

  return accessible, nil
}

TypeScript

/**
 * List integrations the user can access.
 *
 * Post-filtering with CheckBulk: Query all integrations, then batch-check permissions.
 */
async function listIntegrations(db: Database, userId: string): Promise<Integration[]> {
  // Step 1: Query all integrations from database
  const allIntegrations = await db.query("SELECT * FROM integrations");

  // Step 2: Use CheckBulk to filter by permission
  const accessible = await filterIntegrationsByPermission(
    allIntegrations,
    userId,
    "myservice_integration_view"
  );

  return accessible;
}

// Database interface placeholder
interface Database {
  query(sql: string): Promise<Integration[]>;
}

interface Integration {
  id: string;
  name: string;
  workspaceId: string;
}

Ruby

# List integrations the user can access.
#
# Post-filtering with CheckBulk: Query all integrations, then batch-check permissions.
def list_integrations(db, client, user_id)
  # Step 1: Query all integrations from database
  all_integrations = db.exec("SELECT * FROM integrations").to_a

  # Step 2: Use CheckBulk to filter by permission
  filter_integrations_by_permission(
    all_integrations,
    client,
    user_id,
    "myservice_integration_view"
  )
end

Java

/**
 * List integrations the user can access.
 *
 * Post-filtering with CheckBulk: Query all integrations, then batch-check permissions.
 */
public List<Integration> listIntegrations(Database db, String userId) {
    // Step 1: Query all integrations from database
    List<Integration> allIntegrations = db.query("SELECT * FROM integrations");

    // Step 2: Use CheckBulk to filter by permission
    return filterIntegrationsByPermission(
        allIntegrations,
        userId,
        "myservice_integration_view"
    );
}

// Database interface placeholder
interface Database {
    List<Integration> query(String sql);
}

static class Integration {
    String id;
    String name;
    String workspaceId;
}

The tradeoff:

• Database: You fetch everything, even stuff the user can’t see
• Authorization: One batched CheckBulk per 1,000 resources
• Network: One round-trip per batch
• When it makes sense: If the user can access most results anyway

Caution

Post-filtering with CheckBulk can waste database resources. If you query 10,000 resources but the user only has access to 100, you have over-fetched 9,900 rows. Use pre-filtering for high-cardinality resources instead.

Performance considerations

Approach	Database Load	Kessel API Calls	Best For
Pre-filter (coarse)	Low - workspace filter	1 StreamedListObjects for workspaces	High-cardinality resources, workspace-based access
Pre-filter (fine)	Low - resource ID filter	1 StreamedListObjects for resource IDs	Resource-level permissions, bounded accessible set
Post-filter (coarse)	Medium - varies by query	CheckBulk for unique workspaces	Data sources you can’t filter upfront
Post-filter (fine)	High - unfiltered query	1-N CheckBulk (1000 items/batch)	Small result sets, user likely accesses most

Key tradeoffs:

1. Database vs. Kessel load: Pre-filtering trades one API call for a faster database query
2. Cardinality: High-cardinality resources favor pre-filtering to avoid over-fetching
3. Access probability: If users usually access most of the results anyway, post-filtering can work
4. Pagination: Pre-filtering enables efficient pagination; post-filtering can skip results unpredictably

Tip

Default recommendation: Use pre-filtering with StreamedListObjects at the workspace level (coarse-grained). It scales well for high-cardinality resources.

Decision guide

flowchart TD
    START(["How should I protect<br/>this list endpoint?"])
    Q1{"Are resources<br/>assigned to workspaces?"}
    Q2{"Can you query by<br/>workspace ID efficiently?"}
    Q3{"Is the result set<br/>always small?<br/>(&lt; 100 items)"}
    Q4{"Do you need<br/>resource-level permissions<br/>(not workspace-level)?"}

    PRE_COARSE["✅ <strong>Pre-filter (coarse)</strong><br/>StreamedListObjects + workspace query<br/><em>Recommended for most cases</em>"]
    PRE_FINE["✅ <strong>Pre-filter (fine)</strong><br/>StreamedListObjects + resource ID query"]
    POST_COARSE["⚠️ <strong>Post-filter (coarse)</strong><br/>Query + workspace check per result<br/><em>Use only for small result sets</em>"]
    POST_FINE["⚠️ <strong>Post-filter (fine)</strong><br/>Query + resource check per result<br/><em>Last resort - performance cost</em>"]

    START --> Q1
    Q1 -- "Yes" --> Q2
    Q1 -- "No" --> Q4
    Q2 -- "Yes" --> PRE_COARSE
    Q2 -- "No" --> Q3
    Q3 -- "Yes" --> POST_COARSE
    Q3 -- "No" --> POST_FINE
    Q4 -- "Yes" --> PRE_FINE
    Q4 -- "No" --> POST_FINE

    style PRE_COARSE fill:#e8f5e9,stroke:#333,color:#000
    style PRE_FINE fill:#fff9e6,stroke:#333,color:#000
    style POST_COARSE fill:#fff4e1,stroke:#333,color:#000
    style POST_FINE fill:#ffe6e6,stroke:#333,color:#000

Real-world examples

Two Insights applications serve as reference implementations:

Digital Roadmap: Pre-filtering with StreamedListObjects

Digital Roadmap uses pre-filtering to show users only the roadmap items in workspaces they can access.

Pattern:

1. Call StreamedListObjects to get accessible workspaces
2. Query roadmap items with WHERE workspace_id IN (...)
3. Return filtered results

Key implementation: src/roadmap/common.py - uses StreamedListObjects to get accessible workspace IDs, then filters database queries with those IDs.

Why this works: Roadmap items are workspace-scoped, and users typically have access to fewer than 100 workspaces. Pre-filtering keeps the database query small and fast.

Config Manager: Default workspace pattern

Config Manager uses a simplified pattern for organization-level resources.

Pattern:

1. Look up the user’s default workspace (their organization’s root workspace)
2. Check permission on that single workspace
3. If allowed, return all resources for that organization

Key implementation: internal/http/middleware/authorization/kessel.go - the middleware checks the default workspace permission before allowing access to organization-wide configuration.

Why this works: Config Manager’s profiles are organization-wide settings, not workspace-scoped resources. One check tells you if the user can manage settings for their org — no need to check individual profiles.

Note

These are real production implementations. These patterns cover the most common scenarios for list endpoint authorization.

Next steps

Protect an endpoint Learn the basics of Check and CheckForUpdate for individual resource protection.

Relationships and permissions Understand how workspace hierarchy and permission inheritance work.

Consistency model Learn about consistency guarantees and when they matter for authorization.