Enable TLS with Kessel SDKs (Client-side)

The Kessel SDKs support TLS for secure communication with Kessel services. To use OAuth 2.0 authentication, you must also configure TLS transport credentials.

Prerequisites

Before configuring TLS, your process will need to have access to the CA certificate(s) used to sign the Kessel service certificates.

Kubernetes/OpenShift Deployment using OpenShift Service CA

When deploying in Kubernetes or OpenShift, you can mount CA certificates using ConfigMaps:

volumeMounts:
  - name: ca-certs
    mountPath: "/ca-certs"
volumes:
  - name: ca-certs
    configMap:
      name: openshift-service-ca.crt

This mounts the CA certificate at /ca-certs/service-ca.crt, which you can reference in your application code.

Setup

Configure Transport Credentials. Load your CA certificate to verify the server.

Python
Go

def configure_tls(ca_cert_path: str) -> grpc.ChannelCredentials:
    """Load CA certificate and create TLS credentials."""
    with open(ca_cert_path, 'rb') as f:
        ca_cert = f.read()
    return grpc.ssl_channel_credentials(root_certificates=ca_cert)

func configureTLS(caPath string) (credentials.TransportCredentials, error) {
  caCert, err := os.ReadFile(caPath)
  if err != nil {
    return nil, fmt.Errorf("failed to read the ca cert file: %w", err)
  }

  certPool := x509.NewCertPool()
  if !certPool.AppendCertsFromPEM(caCert) {
    return nil, fmt.Errorf("failed to add server CA certificate")
  }

  config := &tls.Config{
    RootCAs: certPool,
  }
  return credentials.NewTLS(config), nil
}

Configure Call Credentials. Set up OAuth 2.0 authentication for your client.
- Python
- Go
# Configure OAuth 2.0 credentials discovery = fetch_oidc_discovery(os.getenv("ISSUER_URL")) token_endpoint = discovery.token_endpoint # Create OAuth2 credentials with the discovered token endpoint auth_credentials = OAuth2ClientCredentials( client_id=os.getenv("CLIENT_ID"), client_secret=os.getenv("CLIENT_SECRET"), token_endpoint=token_endpoint, )
// Configure OAuth 2.0 credentials oauthCredentials := auth.NewOAuth2ClientCredentials( os.Getenv("CLIENT_ID"), os.Getenv("CLIENT_SECRET"), os.Getenv("TOKEN_ENDPOINT"), )

Combine Both. Use your SDK’s builder to apply both credential types and create the client.

Python
Go

# Configure TLS transport credentials
ssl_credentials = configure_tls("/ca-certs/service-ca.crt")

# Build authenticated client with TLS
stub, channel = ClientBuilder(os.getenv("KESSEL_ENDPOINT")).oauth2_client_authenticated(auth_credentials, ssl_credentials).build()

// Configure TLS transport credentials
channelCreds, err := configureTLS("/ca-certs/service-ca.crt")
if err != nil {
  log.Fatalf("Failed to setup TLS: %v", err)
}

// Build authenticated client with TLS
client, conn, err := v1beta2.NewClientBuilder(os.Getenv("KESSEL_ENDPOINT")).
  Authenticated(kesselgrpc.OAuth2CallCredentials(&oauthCredentials), channelCreds).
  Build()
if err != nil {
  log.Fatalf("Failed to create client: %v", err)
}
defer conn.Close()

import os
import grpc
from google.protobuf import struct_pb2
from kessel.auth import OAuth2ClientCredentials, fetch_oidc_discovery
from kessel.inventory.v1beta2 import (
    ClientBuilder,
    report_resource_request_pb2,
    resource_representations_pb2,
    representation_metadata_pb2,
)


def configure_tls(ca_cert_path: str) -> grpc.ChannelCredentials:
    """Load CA certificate and create TLS credentials."""
    with open(ca_cert_path, 'rb') as f:
        ca_cert = f.read()
    return grpc.ssl_channel_credentials(root_certificates=ca_cert)


def main():
    # Configure OAuth 2.0 credentials
    discovery = fetch_oidc_discovery(os.getenv("ISSUER_URL"))
    token_endpoint = discovery.token_endpoint

    # Create OAuth2 credentials with the discovered token endpoint
    auth_credentials = OAuth2ClientCredentials(
        client_id=os.getenv("CLIENT_ID"),
        client_secret=os.getenv("CLIENT_SECRET"),
        token_endpoint=token_endpoint,
    )

    # Configure TLS transport credentials
    ssl_credentials = configure_tls("/ca-certs/service-ca.crt")

    # Build authenticated client with TLS
    stub, channel = ClientBuilder(os.getenv("KESSEL_ENDPOINT")).oauth2_client_authenticated(auth_credentials, ssl_credentials).build()

    with channel:
        # Example: Report a resource
        common_struct = struct_pb2.Struct()
        common_struct.update({"workspace_id": "workspace-1"})

        metadata = representation_metadata_pb2.RepresentationMetadata(
            local_resource_id="doc-123",
            api_href="https://drive.example.com/document/123",
        )

        representations = resource_representations_pb2.ResourceRepresentations(
            metadata=metadata,
            common=common_struct,
        )

        report_req = report_resource_request_pb2.ReportResourceRequest(
            type="document",
            reporter_type="drive",
            reporter_instance_id="drive-1",
            representations=representations,
        )

        try:
            response = stub.ReportResource(report_req)
            print("Successfully reported resource")
            print(response)
        except grpc.RpcError as e:
            print(f"gRPC error: {e.code()} - {e.details()}")


if __name__ == "__main__":
    main()

package main

import (
  "context"
  "crypto/tls"
  "crypto/x509"
  "fmt"
  "log"
  "os"

  "google.golang.org/grpc/credentials"
  "google.golang.org/protobuf/types/known/structpb"

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

func configureTLS(caPath string) (credentials.TransportCredentials, error) {
  caCert, err := os.ReadFile(caPath)
  if err != nil {
    return nil, fmt.Errorf("failed to read the ca cert file: %w", err)
  }

  certPool := x509.NewCertPool()
  if !certPool.AppendCertsFromPEM(caCert) {
    return nil, fmt.Errorf("failed to add server CA certificate")
  }

  config := &tls.Config{
    RootCAs: certPool,
  }
  return credentials.NewTLS(config), nil
}


func main() {
  // Configure OAuth 2.0 credentials
  oauthCredentials := auth.NewOAuth2ClientCredentials(
    os.Getenv("CLIENT_ID"),
    os.Getenv("CLIENT_SECRET"),
    os.Getenv("TOKEN_ENDPOINT"),
  )

  // Configure TLS transport credentials
  channelCreds, err := configureTLS("/ca-certs/service-ca.crt")
  if err != nil {
    log.Fatalf("Failed to setup TLS: %v", err)
  }

  // Build authenticated client with TLS
  client, conn, err := v1beta2.NewClientBuilder(os.Getenv("KESSEL_ENDPOINT")).
    Authenticated(kesselgrpc.OAuth2CallCredentials(&oauthCredentials), channelCreds).
    Build()
  if err != nil {
    log.Fatalf("Failed to create client: %v", err)
  }
  defer conn.Close()

  ctx := context.Background()

  // Example: Report a resource
  reportRequest := &v1beta2.ReportResourceRequest{
    Type:               "document",
    ReporterType:       "drive",
    ReporterInstanceId: "drive-1",
    Representations: &v1beta2.ResourceRepresentations{
      Metadata: &v1beta2.RepresentationMetadata{
        LocalResourceId: "doc-123",
        ApiHref:         "https://drive.example.com/document/123",
      },
      Common: &structpb.Struct{
        Fields: map[string]*structpb.Value{
          "workspace_id": structpb.NewStringValue("workspace-1"),
        },
      },
    },
  }

  response, err := client.ReportResource(ctx, reportRequest)
  if err != nil {
    log.Fatalf("Failed to report resource: %v", err)
  }

  fmt.Printf("Successfully reported resource: %+v\n", response)
}