Skip to content
Kessel

Protect an endpoint

Protect your API endpoints by checking user permissions before granting access. You’ll define permissions in KSL, initialize the SDK, and add permission checks to your middleware.

Prerequisites

Section titled “Prerequisites”

Define your service permissions

Section titled “Define your service permissions”

The base RBAC schema (rbac.ksl) is already deployed to Kessel, so you only need to register your service’s permissions.

Register workspace-level permissions

Section titled “Register workspace-level permissions”

Create a schema file for your service. For example, myservice.ksl:

version 0.2
namespace myservice


import rbac


// View permission for integrations - allows reading integration configurations
@rbac.add_unified_permission(app:'myservice', resource:'integration', verb:'view');


// Edit permission for integrations - allows creating/updating integration configurations
@rbac.add_unified_permission(app:'myservice', resource:'integration', verb:'edit');

In this schema:

  • import rbac brings in the existing RBAC workspace hierarchy
  • app is your service name (myservice)
  • resource is the resource type within your service (integration)
  • verb is the action—typically view or edit
  • This creates permission names like myservice_integration_view and myservice_integration_edit for Check API calls

These permissions can be checked at the workspace level and are automatically inherited through the workspace hierarchy.

Initialize the SDK client

Section titled “Initialize the SDK client”

The SDK handles authentication and TLS for you.

Install the SDK:

Terminal window
go get github.com/project-kessel/kessel-sdk-go

Create the client:

package main


import (
  "context"
  "log"
  "os"


  "github.com/project-kessel/kessel-sdk-go/kessel/inventory/v1beta2"
)


func main() {
  ctx := context.Background()


  // Create client using environment configuration
  inventoryClient, conn, err := v1beta2.NewClientBuilder(os.Getenv("KESSEL_ENDPOINT")).
    Insecure(). // For local dev - use authenticated connection in production
    Build()
  if err != nil {
    log.Fatal("Failed to create client:", err)
  }
  defer conn.Close()


  // Use inventoryClient for permission checks
}

Implement permission checks

Section titled “Implement permission checks”

Use Check() for reads and CheckForUpdate() for writes. The difference is consistency: reads can tolerate slight staleness, writes need the latest state.

When to use Check vs CheckForUpdate

Section titled “When to use Check vs CheckForUpdate”
MethodConsistencyUse ForExample Operations
Check()Eventually consistent (configurable)Read operations, UI visibility decisions, standard authorizationGET requests, list filtering, showing/hiding UI elements
CheckForUpdate()Strongly consistent (always)Write operations, sensitive operationsPUT/POST/DELETE requests, resource modification, credential access

Check: Read operations

Section titled “Check: Read operations”

Use Check() for reads. Eventual consistency is fine here—users won’t notice a few hundred milliseconds of staleness.

import (
  "context"
  "log"


  "github.com/project-kessel/kessel-sdk-go/kessel/inventory/v1beta2"
  v2 "github.com/project-kessel/kessel-sdk-go/kessel/rbac/v2"
)


func checkReadPermission(ctx context.Context, client v1beta2.KesselInventoryServiceClient, userID, workspaceID string) (bool, error) {
  checkRequest := &v1beta2.CheckRequest{
    Object:   v2.WorkspaceResource(workspaceID),
    Relation: "myservice_integration_view",
    Subject:  v2.PrincipalSubject(userID, "redhat"),
  }


  response, err := client.Check(ctx, checkRequest)
  if err != nil {
    return false, err
  }


  return response.Allowed == v1beta2.Allowed_ALLOWED_TRUE, nil
}

CheckForUpdate: Write operations

Section titled “CheckForUpdate: Write operations”

Use CheckForUpdate() for writes. This ensures you’re checking against the latest permission state.

func checkWritePermission(ctx context.Context, client v1beta2.KesselInventoryServiceClient, userID, workspaceID string) (bool, error) {
  checkRequest := &v1beta2.CheckForUpdateRequest{
    Object:   v2.WorkspaceResource(workspaceID),
    Relation: "myservice_integration_edit",
    Subject:  v2.PrincipalSubject(userID, "redhat"),
  }


  response, err := client.CheckForUpdate(ctx, checkRequest)
  if err != nil {
    return false, err
  }


  return response.Allowed == v1beta2.Allowed_ALLOWED_TRUE, nil
}

Integrate with middleware

Section titled “Integrate with middleware”

Add permission checks to your framework’s middleware:

Client initialization:

# Initialize Kessel client
stub, channel = ClientBuilder("localhost:9000").insecure().build()

Middleware implementation:

def require_permission(relation: str):
    """Decorator to require permission check before executing endpoint."""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # Extract user from request headers
            user_id = request.headers.get("X-User-ID")
            user_domain = request.headers.get("X-User-Domain", "redhat")


            # Extract workspace from header
            workspace_id = request.headers.get("X-Workspace-ID")


            if not user_id or not workspace_id:
                abort(400, "Missing required headers")


            try:
                # Build check request
                check_req = check_request_pb2.CheckRequest(
                    subject=principal_subject(user_id, user_domain),
                    relation=relation,
                    object=workspace_resource(workspace_id),
                )


                # Perform permission check
                response = stub.Check(check_req)


                if response.allowed != allowed_pb2.ALLOWED_TRUE:
                    abort(403, f"Permission denied for relation '{relation}'")


                # Permission granted, proceed
                return f(*args, **kwargs)


            except ConnectError as e:
                abort(500, f"Permission check failed: {e.message}")


        return decorated_function
    return decorator

Using the middleware:

@app.route("/integrations/<integration_id>", methods=["GET"])
@require_permission(relation="myservice_integration_view")
def get_integration(integration_id):
    return {"integration_id": integration_id, "name": "Example Integration"}


@app.route("/integrations/<integration_id>", methods=["PUT"])
@require_permission(relation="myservice_integration_edit")
def update_integration(integration_id):
    return {"status": "updated"}


with channel:
    app.run(debug=True)

Handle errors

Section titled “Handle errors”

If Kessel is unavailable, decide whether to fail closed (deny access) or fail open (allow access). Fail closed is safer for most cases.

func checkPermissionSafe(ctx context.Context, client v1beta2.KesselInventoryServiceClient, req *v1beta2.CheckRequest) bool {
  response, err := client.Check(ctx, req)
  if err != nil {
    // Log the error for monitoring
    log.Printf("Permission check failed: %v", err)


    // Fail closed: deny access when check fails
    return false
  }


  return response.Allowed == v1beta2.Allowed_ALLOWED_TRUE
}

Next steps

Section titled “Next steps”
Kessel Schema Language Design complex permission models with relationships and custom resource types.
Report resources Create the resource instances you're protecting using the inventory API.
Relationships and permissions Understand how permissions are evaluated through the relationship graph.