Context

When setting up S3 blob storage for self-hosted LangSmith, you may want to enable KMS encryption to ensure your data is encrypted at rest. This requires specific configuration parameters and proper IAM permissions to work correctly.

Answer

To enable KMS encryption for S3 storage in LangSmith, you need to configure the following environment variables and ensure proper permissions are in place.

Required Configuration

Set these environment variables in your platform-backend service:

S3_KMS_ENCRYPTION_ENABLED: "true"
S3_KMS_KEY_ARN: "arn:aws:kms:us-east-1:<account-id>:key/<key-id>"

Additionally, ensure you have the standard S3 configuration:

FF_S3_STORAGE_ENABLED: "true"
FF_BLOB_STORAGE_ENABLED: "true"
BLOB_STORAGE_ENGINE: "S3"
S3_BUCKET_NAME: "your-bucket-name"
S3_RUN_MANIFEST_BUCKET_NAME: "your-bucket-name"

Version Requirements

KMS encryption is supported in:

• Helm version: 0.12.12 or later

• Docker image version: 0.12.42 or later

IAM Configuration

Configure your backend service account with the appropriate IAM role:

backend:
  serviceAccount:
    annotations:
      eks.amazonaws.com/role-arn: "arn:aws:iam::<account-id>:role/langsmith-backend-role"

Ensure your IAM role has the necessary KMS permissions as outlined in the blob storage documentation.

Verification

To verify that KMS encryption is working:

1. Check if objects in your bucket are encrypted using the AWS CLI:

   aws s3api head-object \
     --bucket your-bucket-name \
     --key <any-object-in-your-bucket>

   Look for ServerSideEncryption: "aws:kms" and SSEKMSKeyId in the response.

2. Check backend logs for the "Configuring s3 connection" message at INFO level, or use the troubleshooting script to collect pod logs.

3. Use a Python script to verify encryption across all objects:

   import boto3
   
   def check_bucket_encryption(bucket_name):
       s3 = boto3.client("s3")
       
       paginator = s3.get_paginator("list_objects_v2")
       page_iterator = paginator.paginate(Bucket=bucket_name)
       
       for page in page_iterator:
           for obj in page.get("Contents", []):
               key = obj["Key"]
               head = s3.head_object(Bucket=bucket_name, Key=key)
               
               enc = head.get("ServerSideEncryption")
               kms_key = head.get("SSEKMSKeyId")
               
               if enc:
                   print(f"{key}: ENCRYPTED ({enc})", end="")
                   if kms_key:
                       print(f" using KMS key {kms_key}")
                   else:
                       print()
               else:
                   print(f"{key}: NOT encrypted")
   
   check_bucket_encryption("your-bucket-name")

Sources: LangSmith Self-Host Blob Storage Documentation