feat: add crypto (KMS / Key Vault) backend and SQS exclude_env_creden…

CLAUDE.md

@@ -42,6 +42,12 @@ The document category deliberately has **no `base.py` ABC and no backend wrapper
 - **Cache** uses an explicit auth-method argument: `get_cache(provider, auth_method, **kwargs)` where `auth_method` is the literal `from_*` method name (e.g. `get_cache("redis", "from_url", url=...)`).
 - **All five other categories** infer the constructor from **which credential keys are present** in `**kwargs`: `get_storage(provider, **kwargs)` calls `from_access_key` if `aws_access_key_id` is present, `from_connection_string` if `connection_string` is present, and falls through to the managed-identity/IAM-role default. When adding an auth method here, add both the constructor (a `from_*` classmethod, or a `connect_*` function for document) and a routing branch in the factory — for document, in **both** `get_mongodb` and `get_mongodb_sync`.
 
+  AWS `from_assume_role` (STS AssumeRole, cross-account) is wired for **SQS** (`get_queue`) and **S3** (`get_storage` + `get_storage_client`): `role_arn` present → assume-role path, checked **before** `aws_access_key_id`. It does a synchronous `boto3` `sts:AssumeRole` (optional `external_id` → `ExternalId`) and threads the temporary creds into the `aioboto3.Session`. Temp creds are not auto-refreshed — construct a fresh backend when the session expires.
+
+### Messaging payload contract (byte-primitive)
+
+The messaging primitive is **raw bytes + a flat `attributes` map** (string → string), not JSON. `MessagingBackend.send(body: bytes, attributes=None, delay=0, *, group_id, dedup_id)` and `send_batch(messages: list[OutgoingMessage], ...)` where `OutgoingMessage` (in `base.py`) is `{body: bytes, attributes: dict[str,str] | None}`. Attributes map to SQS `MessageAttributes` (String type) and Service Bus `application_properties`. Received `Message.body` is `bytes`; `Message.attributes` is the stringified provider attributes. JSON convenience: the module-level `send_json(backend, message: dict, attributes=None, delay=0, ...)` helper (works for any backend — `json.dumps(...).encode()` then `send`) plus `Message.json()` (decodes the body with `json.loads`). Receipt-handle / ack / FIFO semantics are unchanged.
+
 ### Redis cache specifics
 
 The three Redis backends (`redis`, `elasticache`, `azure_redis`) share **all** their operation logic through `_RedisMixin` in `cache/base.py` — the per-provider modules contain *only* the `from_*` constructors that build the `aioredis.Redis` client with the right auth (URL, IAM SigV4 auto-refresh, access key, managed identity). A new Redis command is implemented **once** in `_RedisMixin` and added as an `@abstractmethod` on `CacheBackend`; never reimplement it per provider. `CacheBackend.pipeline()` ships a no-atomicity `_SequentialPipeline` fallback that the Redis mixin overrides with a real server-side transactional pipeline.

README.md

@@ -98,6 +98,9 @@ s3 = get_storage("s3", bucket="b", region="us-east-1")                       # I
 s3 = get_storage("s3", bucket="b", aws_access_key_id="...",                  # static keys
                  aws_secret_access_key="...", region="us-east-1")
 s3 = get_storage("s3", bucket="b", profile_name="dev")                       # ~/.aws/credentials
+s3 = get_storage("s3", bucket="b",                                           # STS AssumeRole
+                 role_arn="arn:aws:iam::123456789012:role/cross",
+                 external_id="my-external-id", region="us-east-1")
 
 # Azure Blob
 blob = get_storage("azure_blob", connection_string="...", container="c")
@@ -132,21 +135,38 @@ from cloudrift.messaging import get_queue
 sqs = get_queue("sqs", queue_url="https://sqs.us-east-1.amazonaws.com/.../q",
                 region="us-east-1")
 
+# AWS SQS via STS AssumeRole (cross-account)
+sqs = get_queue("sqs", queue_url="https://sqs.../q",
+                role_arn="arn:aws:iam::123456789012:role/cross-account",
+                external_id="my-external-id", region="us-east-1")
+
 # Azure Service Bus
 bus = get_queue("azure_bus", connection_string="...", queue_name="my-queue")
 bus = get_queue("azure_bus", fully_qualified_namespace="ns.servicebus.windows.net",
                 queue_name="my-queue")  # managed identity
 ```
 
-**Operations**:
+**Operations** — the payload primitive is **raw bytes** plus an optional flat
+`attributes` map (string → string). Attributes map to SQS `MessageAttributes`
+(String type) / Service Bus `application_properties`:
 
 ```python
-msg_id = await queue.send({"action": "process", "id": 42}, delay=0)
-ids = await queue.send_batch([{"n": 1}, {"n": 2}])
+from cloudrift.messaging import OutgoingMessage, send_json
+
+# Raw bytes + attributes
+msg_id = await queue.send(b"raw payload", attributes={"content_type": "text/plain"})
+ids = await queue.send_batch([
+    OutgoingMessage(body=b"a", attributes={"k": "1"}),
+    OutgoingMessage(body=b"b"),
+])
+
+# JSON convenience: send_json() dumps+encodes, m.json() loads the bytes body
+await send_json(queue, {"action": "process", "id": 42}, attributes={"v": "1"})
 
 messages = await queue.receive(max_messages=10, wait_time=20)   # long-poll
 for m in messages:
-    handle_job(m.body)
+    handle_job(m.json())         # or m.body for the raw bytes
+    print(m.attributes)          # str -> str map
     await queue.delete(m.receipt_handle)   # ack
     # or: await queue.nack(m.receipt_handle)  # return for immediate redelivery
 
@@ -168,13 +188,13 @@ queues, pass `group_id` (ordering key) and `dedup_id` (deduplication key):
 # SQS FIFO — group_id is required, dedup_id optional if the queue has
 # content-based deduplication enabled
 fifo = get_queue("sqs", queue_url="https://sqs.../jobs.fifo", region="us-east-1")
-await fifo.send({"task": "extract"}, group_id="owner-123", dedup_id="evt-abc")
+await send_json(fifo, {"task": "extract"}, group_id="owner-123", dedup_id="evt-abc")
 
 # Azure Service Bus — queue must be created with sessions enabled;
 # pass session_enabled=True so the backend uses session receivers
 bus = get_queue("azure_bus", connection_string="...", queue_name="jobs",
                 session_enabled=True)
-await bus.send({"task": "extract"}, group_id="owner-123", dedup_id="evt-abc")
+await send_json(bus, {"task": "extract"}, group_id="owner-123", dedup_id="evt-abc")
 
 messages = await fifo.receive(max_messages=10, wait_time=20, visibility_timeout=300)
 for m in messages:

cloudrift/__init__.py

@@ -4,10 +4,11 @@
 from cloudrift.cache import get_cache, cache_broker_url
 from cloudrift.secrets import get_secrets
 from cloudrift.sql import get_sql
+from cloudrift.crypto import get_crypto
 from cloudrift.pubsub import get_pubsub
 from cloudrift.email import get_email
 
-__version__ = "0.2.5"
+__version__ = "0.2.6"
 __all__ = [
     "get_storage",
     "get_queue",
@@ -17,6 +18,7 @@
     "cache_broker_url",
     "get_secrets",
     "get_sql",
+    "get_crypto",
     "get_pubsub",
     "get_email",
 ]

cloudrift/core/exceptions.py

@@ -78,6 +78,19 @@ class SQLAuthError(SQLError):
     Azure AD access token or RDS IAM token could not be obtained)."""
 
 
+# Crypto (KMS / Key Vault keys) exceptions
+class CryptoError(CloudRiftError):
+    """Base exception for cryptographic (KMS / Key Vault) operations."""
+
+
+class CryptoKeyNotFoundError(CryptoError):
+    """Raised when the referenced key does not exist or is unavailable."""
+
+
+class CryptoPermissionError(CryptoError):
+    """Raised on key access / crypto-operation permission failures."""
+
+
 # Pub/Sub exceptions
 class PubSubError(CloudRiftError):
     """Base exception for pub/sub operations."""

cloudrift/crypto/__init__.py

@@ -0,0 +1,50 @@
+from cloudrift.crypto.base import CryptoBackend
+
+
+def get_crypto(provider: str, **kwargs) -> CryptoBackend:
+    """Factory to instantiate a key-management crypto backend.
+
+    The cross-cloud analog of AWS KMS: encrypt/decrypt small payloads against a
+    managed key. AWS KMS ↔ Azure Key Vault keys.
+
+    Args:
+        provider: ``"aws_kms"`` or ``"azure_keyvault"``.
+        **kwargs: Provider-specific config. The AWS factory routes to the
+            appropriate ``from_*`` classmethod based on which credential keys are
+            present; the Azure factory routes by whether ``client_secret`` is set.
+
+    Returns:
+        A :class:`CryptoBackend` instance.
+
+    Examples:
+        get_crypto("aws_kms", key_id="arn:aws:kms:...:key/abc", region="us-east-1")  # IAM/env
+        get_crypto("aws_kms", key_id="alias/my-key", aws_access_key_id="AKIA...",
+                   aws_secret_access_key="...", region="us-east-1")
+        get_crypto("azure_keyvault",
+                   key_id="https://myvault.vault.azure.net/keys/mykey",
+                   tenant_id="...", client_id="...", client_secret="...")
+        get_crypto("azure_keyvault",
+                   key_id="https://myvault.vault.azure.net/keys/mykey")  # managed identity
+    """
+    if provider in ("aws_kms", "kms"):
+        from cloudrift.crypto.aws_kms import AWSKMSBackend
+
+        if "aws_access_key_id" in kwargs:
+            return AWSKMSBackend.from_access_key(**kwargs)
+        if "profile_name" in kwargs:
+            return AWSKMSBackend.from_profile(**kwargs)
+        return AWSKMSBackend.from_iam_role(**kwargs)
+
+    if provider in ("azure_keyvault", "azure_keyvault_keys"):
+        from cloudrift.crypto.azure_keyvault_keys import AzureKeyVaultKeysBackend
+
+        if "client_secret" in kwargs:
+            return AzureKeyVaultKeysBackend.from_service_principal(**kwargs)
+        return AzureKeyVaultKeysBackend.from_managed_identity(**kwargs)
+
+    raise ValueError(
+        f"Unknown crypto provider: {provider!r}. Choose 'aws_kms' or 'azure_keyvault'."
+    )
+
+
+__all__ = ["CryptoBackend", "get_crypto"]

cloudrift/crypto/aws_kms.py

@@ -0,0 +1,166 @@
+import asyncio
+
+import aioboto3
+from botocore.config import Config
+from botocore.exceptions import ClientError
+
+from cloudrift.core.exceptions import (
+    CryptoError,
+    CryptoKeyNotFoundError,
+    CryptoPermissionError,
+)
+from cloudrift.crypto.base import CryptoBackend
+
+
+class AWSKMSBackend(CryptoBackend):
+    """AWS KMS crypto backend (native async via ``aioboto3``).
+
+    Encrypts/decrypts directly against a symmetric KMS key — the same
+    ``Encrypt`` / ``Decrypt`` calls the AWS SDK makes, so ciphertext is
+    interchangeable with anything encrypted by raw boto3 against the same key.
+    A single async client is created lazily and reused.
+
+    Construct via:
+    - ``from_access_key`` — static credentials (+ optional session token)
+    - ``from_iam_role``   — instance profile / environment / ECS task role
+    - ``from_profile``    — named profile from ``~/.aws/credentials``
+
+    ``key_id`` (key id, ARN, or alias) is required to encrypt. It is optional for
+    decrypt-only backends — KMS resolves the key from the ciphertext.
+    """
+
+    def __init__(
+        self,
+        session: aioboto3.Session,
+        key_id: str | None = None,
+        *,
+        encryption_context: dict | None = None,
+        endpoint_url: str | None = None,
+        max_pool_connections: int = 25,
+        connect_timeout: float = 10.0,
+        read_timeout: float = 30.0,
+        client_kwargs: dict | None = None,
+    ) -> None:
+        self._session = session
+        self._key_id = key_id
+        self._encryption_context = encryption_context or {}
+        self._endpoint_url = endpoint_url
+        self._config = Config(
+            max_pool_connections=max_pool_connections,
+            connect_timeout=connect_timeout,
+            read_timeout=read_timeout,
+        )
+        self._client_kwargs = client_kwargs or {}
+        self._client_cm = None
+        self._client = None
+        self._lock = asyncio.Lock()
+
+    # ------------------------------------------------------------------
+    # Factory constructors
+    # ------------------------------------------------------------------
+
+    @classmethod
+    def from_access_key(
+        cls,
+        aws_access_key_id: str,
+        aws_secret_access_key: str,
+        key_id: str | None = None,
+        region: str = "us-east-1",
+        aws_session_token: str | None = None,
+        **kwargs,
+    ) -> "AWSKMSBackend":
+        """Authenticate with explicit access key / secret."""
+        session = aioboto3.Session(
+            aws_access_key_id=aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key,
+            aws_session_token=aws_session_token,
+            region_name=region,
+        )
+        return cls(session, key_id, **kwargs)
+
+    @classmethod
+    def from_iam_role(
+        cls,
+        key_id: str | None = None,
+        region: str = "us-east-1",
+        **kwargs,
+    ) -> "AWSKMSBackend":
+        """Authenticate via IAM role / instance profile / environment variables."""
+        session = aioboto3.Session(region_name=region)
+        return cls(session, key_id, **kwargs)
+
+    @classmethod
+    def from_profile(
+        cls,
+        profile_name: str,
+        key_id: str | None = None,
+        region: str = "us-east-1",
+        **kwargs,
+    ) -> "AWSKMSBackend":
+        """Authenticate using a named profile from ``~/.aws/credentials``."""
+        session = aioboto3.Session(profile_name=profile_name, region_name=region)
+        return cls(session, key_id, **kwargs)
+
+    # ------------------------------------------------------------------
+    # Internal lifecycle
+    # ------------------------------------------------------------------
+
+    async def _ensure(self):
+        if self._client is not None:
+            return self._client
+        async with self._lock:
+            if self._client is None:
+                self._client_cm = self._session.client(
+                    "kms",
+                    endpoint_url=self._endpoint_url,
+                    config=self._config,
+                    **self._client_kwargs,
+                )
+                try:
+                    self._client = await self._client_cm.__aenter__()
+                except Exception:
+                    self._client_cm = None
+                    raise
+        return self._client
+
+    async def close(self) -> None:
+        client_cm, self._client_cm = self._client_cm, None
+        self._client = None
+        if client_cm is not None:
+            await client_cm.__aexit__(None, None, None)
+
+    # ------------------------------------------------------------------
+    # CryptoBackend implementation
+    # ------------------------------------------------------------------
+
+    async def encrypt(self, plaintext: bytes) -> bytes:
+        if not self._key_id:
+            raise CryptoError("AWS KMS key_id is required to encrypt")
+        client = await self._ensure()
+        try:
+            kwargs: dict = {"KeyId": self._key_id, "Plaintext": plaintext}
+            if self._encryption_context:
+                kwargs["EncryptionContext"] = self._encryption_context
+            response = await client.encrypt(**kwargs)
+            return response["CiphertextBlob"]
+        except ClientError as e:
+            self._raise(e)
+
+    async def decrypt(self, ciphertext: bytes) -> bytes:
+        client = await self._ensure()
+        try:
+            kwargs: dict = {"CiphertextBlob": ciphertext}
+            if self._encryption_context:
+                kwargs["EncryptionContext"] = self._encryption_context
+            response = await client.decrypt(**kwargs)
+            return response["Plaintext"]
+        except ClientError as e:
+            self._raise(e)
+
+    def _raise(self, exc: ClientError):
+        code = exc.response["Error"]["Code"]
+        if code in ("NotFoundException", "KeyUnavailableException"):
+            raise CryptoKeyNotFoundError(str(exc)) from exc
+        if code in ("AccessDeniedException", "UnauthorizedAccess"):
+            raise CryptoPermissionError(str(exc)) from exc
+        raise CryptoError(str(exc)) from exc