BerriAI/litellm · error · ValueError

sqs_app_encryption_key_b64 is required when encryption is en

Error message

sqs_app_encryption_key_b64 is required when encryption is enabled.

What it means

SQSLogger.__init__ enables application-level encryption (AES-GCM via AppCrypto) when sqs_aws_use_application_level_encryption is true, and requires a base64-encoded key from litellm.aws_sqs_callback_params['sqs_app_encryption_key_b64'] or the constructor arg. Without the key there is nothing to encrypt with, so construction aborts with ValueError.

Source

Thrown at litellm/integrations/sqs.py:197

            litellm.aws_sqs_callback_params.get("sqs_strip_base64_files", False) or sqs_strip_base64_files
        )

        self.sqs_aws_use_application_level_encryption = (
            litellm.aws_sqs_callback_params.get("sqs_aws_use_application_level_encryption", False)
            or sqs_aws_use_application_level_encryption
        )
        self.sqs_app_encryption_key_b64 = (
            litellm.aws_sqs_callback_params.get("sqs_app_encryption_key_b64") or sqs_app_encryption_key_b64
        )
        self.sqs_app_encryption_aad = (
            litellm.aws_sqs_callback_params.get("sqs_app_encryption_aad") or sqs_app_encryption_aad
        )
        self.app_crypto: AppCrypto | None = None
        if self.sqs_aws_use_application_level_encryption:
            from litellm.litellm_core_utils.app_crypto import AppCrypto

            if not self.sqs_app_encryption_key_b64:
                raise ValueError("sqs_app_encryption_key_b64 is required when encryption is enabled.")
            key = base64.b64decode(self.sqs_app_encryption_key_b64)
            self.app_crypto = AppCrypto(key)
            verbose_logger.debug("SQSLogger: Application-level encryption enabled.")
        self.sqs_config = litellm.aws_sqs_callback_params.get("sqs_config") or sqs_config

    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:
        try:
            verbose_logger.debug("SQS Logging - Enters logging function for model %s", kwargs)
            standard_logging_payload = kwargs.get("standard_logging_object")
            if self.sqs_strip_base64_files:
                standard_logging_payload = await self._strip_base64_from_messages(standard_logging_payload)
            if standard_logging_payload is None:
                raise ValueError("standard_logging_payload is None")

            self.log_queue.append(standard_logging_payload)
            verbose_logger.debug(
                "sqs logging: queue length %s, batch size %s",
                len(self.log_queue),

View on GitHub (pinned to 6c2dcb801b)

Solutions

  1. Set litellm.aws_sqs_callback_params = {'sqs_app_encryption_key_b64': '<base64 key>', ...} (or pass sqs_app_encryption_key_b64= to the constructor) alongside enabling encryption
  2. Generate a valid key: python -c "import os,base64; print(base64.b64encode(os.urandom(32)).decode())" and distribute it to every producer/consumer that must decrypt
  3. If encryption is not actually required, leave sqs_aws_use_application_level_encryption unset/false

Example fix

# before
litellm.aws_sqs_callback_params = {
    "sqs_aws_use_application_level_encryption": True,
    # key missing -> ValueError
}

# after
import base64, os
litellm.aws_sqs_callback_params = {
    "sqs_aws_use_application_level_encryption": True,
    "sqs_app_encryption_key_b64": base64.b64encode(os.urandom(32)).decode(),
    "sqs_app_encryption_aad": "litellm-logs",
}
Defensive patterns

Strategy: validation

Validate before calling

import base64, binascii

def valid_b64_key(v: str | None) -> bool:
    if not v:
        return False
    try:
        return len(base64.b64decode(v, validate=True)) in (16, 24, 32)
    except (binascii.Error, ValueError):
        return False

if litellm.aws_sqs_callback_params.get("sqs_aws_use_application_level_encryption"):
    assert valid_b64_key(
        litellm.aws_sqs_callback_params.get("sqs_app_encryption_key_b64")
    ), "sqs_app_encryption_key_b64 must be a base64 AES key"

Type guard

from typing import Any, TypeGuard

def is_b64_aes_key(v: Any) -> TypeGuard[str]:
    import base64, binascii
    if not isinstance(v, str) or not v:
        return False
    try:
        return len(base64.b64decode(v, validate=True)) in (16, 24, 32)
    except (binascii.Error, ValueError):
        return False

Try / catch

try:
    sqs_logger = SQSLogger()
except ValueError as e:
    if "sqs_app_encryption_key_b64 is required" in str(e):
        raise RuntimeError("SQS encryption enabled but key missing — refusing to start") from e
    raise

Prevention

When it happens

Trigger: Turning on sqs_aws_use_application_level_encryption (env var LITELLM_SQS_USE_APPLICATION_LEVEL_ENCRYPTION or callback config) without supplying the key; passing the key under a different name or only via constructor while reading config from litellm.aws_sqs_callback_params; base64 key set but empty string (falsy) after .get().

Common situations: Enabling encryption in staging from a runbook that omits the key step; secrets mounted as empty files (key resolves to ''); key stored in the proxy YAML under the wrong field name.

Related errors


AI-assisted analysis of BerriAI/litellm@6c2dcb801b (2026-08-15). Data as JSON: /api/errors/3cfbb52e96d4686f. Report an issue: GitHub.