feat!: dict-primitive messaging payloads and unified Azure credential…

CLAUDE.md

@@ -44,9 +44,21 @@ The document category deliberately has **no `base.py` ABC and no backend wrapper
 
   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)
+### Azure credentials — one chain, defined once
 
-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.
+Every Azure `from_managed_identity` builds its credential through **`cloudrift/core/azure_credentials.py`** (`build_async_credential` / `build_credential`), never by instantiating an `azure.identity` class directly. The chain is `DefaultAzureCredential` with the developer-machine sources excluded, yielding **workload identity → managed identity → az CLI**. Excluding `environment` is the point, not an accident: ambient `AZURE_CLIENT_ID`/`AZURE_CLIENT_SECRET` would otherwise shadow the workload's real identity — the same reasoning as SQS's `exclude_env_credentials`. When adding an Azure backend, call the helper; **do not** reintroduce `ManagedIdentityCredential`.
+
+Each constructor takes `credential_options: dict | None = None`, forwarded verbatim to `DefaultAzureCredential` (overrides win over the defaults). It is deliberately an explicit dict rather than `**kwargs`: with `**kwargs` a misspelled *backend* option (e.g. `session_enable` for `session_enabled`) would be silently swallowed as a credential option instead of raising `TypeError`. Two backends pass it positionally-adjacent to their own `**kwargs` (`crypto`, `sql/mssql`), which is the same reason. `azure-identity` is pinned `>=1.15.0`, which predates `exclude_broker_credential` — so that one is not set by default; callers on newer releases can pass it through `credential_options`.
+
+Credentials are **not** shared between backends: each owns the one it built and closes it in `close()`, so a module-level singleton would let one backend's shutdown break another's.
+
+### Messaging payload contract (dict-primitive)
+
+The messaging primitive is a **dict + a flat `attributes` map** (string → string). Unlike every other category, `send`/`send_batch` are **concrete on the ABC**, not abstract: `MessagingBackend.send(payload: dict, attributes=None, delay=0, *, group_id, dedup_id)` serializes through the module-level `to_json(payload, default=...)` — the single serialization point — and calls the backend's `_send_json(body: str, ...)`. `send_batch(messages: list[OutgoingMessage], ...)` where `OutgoingMessage` is `{body: dict, attributes: dict[str,str] | None}` likewise calls `_send_json_batch(items: list[tuple[str, dict|None]], ...)`. **When adding a backend, implement `_send_json`/`_send_json_batch` and never override `send`/`send_batch`** — that is what keeps serialization uniform. `to_json` rejects non-dicts with `TypeError`, so a backend can never see an unserialized payload and a caller can never send raw bytes. The class attribute `json_default` (default `str`) is the `json.dumps` fallback for `datetime`/`Decimal`/`UUID`; override it per backend if fidelity matters.
+
+The hook passes a **`str`, not `bytes`**, because neither provider wants bytes: SQS's `MessageBody` is typed `string` (so bytes would force an `encode`/`decode` round-trip), and `ServiceBusMessage(str)` produces a byte-identical AMQP `DATA` body to `ServiceBusMessage(bytes)`.
+
+Attributes map to SQS `MessageAttributes` (String type) and Service Bus `application_properties`. On the **receive** side `Message.body` is still `bytes` — deliberately asymmetric, so a malformed or non-UTF-8 payload from a foreign producer stays inspectable for DLQ triage — with `Message.data` as the `json.loads`'d dict. `Message.attributes` is the stringified provider attributes. Receipt-handle / ack / FIFO semantics are unchanged.
 
 ### Redis cache specifics
 

README.md

@@ -86,6 +86,43 @@ storage = get_storage(
 )
 ```
 
+### Azure authentication
+
+Every Azure backend resolves its identity through the same chain, so one code
+path works in all three environments a service runs in:
+
+```
+Workload Identity  →  Managed Identity  →  Azure CLI
+   (AKS)              (App Service /        (local dev,
+                       Container Apps /      after `az login`)
+                       VM)
+```
+
+That means `get_storage("azure_blob", account_url=...)` with no credentials works
+unchanged on AKS, on App Service, and on your laptop. Pass `client_id=` to select a
+*user-assigned* managed identity; omit it for the system-assigned one.
+
+Developer-machine credential sources are deliberately excluded: ambient
+`AZURE_CLIENT_ID` / `AZURE_CLIENT_SECRET` env vars (which would silently shadow the
+workload's real identity — the Azure counterpart of SQS's `exclude_env_credentials`),
+plus the shared token cache, VS Code, PowerShell, and `azd`.
+
+To override — for example to lock a production service down to managed identity only:
+
+```python
+storage = get_storage("azure_blob", account_url="...", container="c",
+                      credential_options={"exclude_cli_credential": True})
+```
+
+`credential_options` is accepted by every Azure `from_managed_identity` constructor and
+is forwarded verbatim to `DefaultAzureCredential`. The chain itself is defined once in
+`cloudrift/core/azure_credentials.py`.
+
+> On a developer machine this means code intended to exercise managed identity will
+> instead succeed using your `az login` identity rather than failing. Pass
+> `credential_options={"exclude_cli_credential": True}` when you need to test the
+> production path.
+
 ---
 
 ## Storage
@@ -146,26 +183,22 @@ bus = get_queue("azure_bus", fully_qualified_namespace="ns.servicebus.windows.ne
                 queue_name="my-queue")  # managed identity
 ```
 
-**Operations** — the payload primitive is **raw bytes** plus an optional flat
-`attributes` map (string → string). Attributes map to SQS `MessageAttributes`
+**Operations** — pass a **dict**; cloudrift serializes it to JSON. Optionally add a
+flat `attributes` map (string → string), which maps to SQS `MessageAttributes`
 (String type) / Service Bus `application_properties`:
 
 ```python
-from cloudrift.messaging import OutgoingMessage, send_json
+from cloudrift.messaging import OutgoingMessage
 
-# Raw bytes + attributes
-msg_id = await queue.send(b"raw payload", attributes={"content_type": "text/plain"})
+msg_id = await queue.send({"action": "process", "id": 42}, attributes={"v": "1"})
 ids = await queue.send_batch([
-    OutgoingMessage(body=b"a", attributes={"k": "1"}),
-    OutgoingMessage(body=b"b"),
+    OutgoingMessage(body={"n": 1}, attributes={"k": "1"}),
+    OutgoingMessage(body={"n": 2}),
 ])
 
-# 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.json())         # or m.body for the raw bytes
+    handle_job(m.data)           # the payload dict; m.body is 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
@@ -174,6 +207,14 @@ await queue.purge()
 await queue.close()
 ```
 
+`send()` takes a `dict` and nothing else — passing bytes, a string, or a list raises
+`TypeError`. Values `json.dumps` can't encode natively (`datetime`, `Decimal`, `UUID`)
+are stringified via the backend's `json_default`, which defaults to `str`.
+
+`Message.body` stays **raw bytes** on the receive path so a malformed or non-UTF-8
+payload from a foreign producer is still inspectable for dead-letter triage;
+`Message.data` is the decoded dict.
+
 > **Azure Service Bus note:** receipt handles are lock tokens — they are only
 > valid on the same backend instance that received the message, and only within
 > the message lock duration. SQS receipt handles, by contrast, are plain strings
@@ -188,13 +229,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 send_json(fifo, {"task": "extract"}, group_id="owner-123", dedup_id="evt-abc")
+await fifo.send({"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 send_json(bus, {"task": "extract"}, group_id="owner-123", dedup_id="evt-abc")
+await bus.send({"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

@@ -8,7 +8,7 @@
 from cloudrift.pubsub import get_pubsub
 from cloudrift.email import get_email
 
-__version__ = "0.2.6"
+__version__ = "0.2.7"
 __all__ = [
     "get_storage",
     "get_queue",

cloudrift/cache/redis_azure.py

@@ -64,24 +64,29 @@ def from_managed_identity(
         ssl: bool = True,
         client_id: str | None = None,
         decode_responses: bool = False,
+        credential_options: dict | None = None,
     ) -> "AzureRedisCacheBackend":
-        """Authenticate via Azure Managed Identity (Entra ID token auth).
+        """Authenticate via Azure AD (Entra ID token auth).
 
+        Resolves an identity through workload identity → managed identity → az CLI.
         Requires the cache to have *Microsoft Entra Authentication* enabled and the
-        managed identity to have a Redis data-access role assigned.
+        identity to have a Redis data-access role assigned.
 
         Args:
             host: e.g. ``<name>.redis.cache.windows.net``
-            username: The object ID (or configured Redis username) of the managed identity.
+            username: The object ID (or configured Redis username) of the identity.
             port: Redis SSL port (default 6380).
             db: Database index (default 0).
             ssl: Enable TLS (default ``True``).
             client_id: Optional client ID for a user-assigned managed identity.
                        Omit to use the system-assigned identity.
+            credential_options: Forwarded to ``DefaultAzureCredential`` — see
+                       :mod:`cloudrift.core.azure_credentials`.
         """
         try:
-            from azure.identity import ManagedIdentityCredential
-            credential = ManagedIdentityCredential(client_id=client_id) if client_id else ManagedIdentityCredential()
+            from cloudrift.core.azure_credentials import build_credential
+
+            credential = build_credential(client_id, **(credential_options or {}))
             provider = _AzureEntraCredentialProvider(credential, username)
             client = aioredis.Redis(
                 host=host,

cloudrift/core/azure_credentials.py

@@ -0,0 +1,80 @@
+"""Shared Azure AD credential construction.
+
+Every Azure backend authenticates the same way, so the chain is defined once
+here instead of being copy-pasted into each provider module.
+
+The chain that results from the defaults below is::
+
+    Workload Identity  ->  Managed Identity  ->  Azure CLI
+
+which covers the three environments a Lyzr service actually runs in — AKS with
+workload identity, App Service / Container Apps / VM with a managed identity,
+and a developer machine with ``az login`` — without any per-environment code.
+
+Everything excluded below is a developer-machine credential source that is
+either ambiguous or actively harmful in a service:
+
+- ``environment`` — ambient ``AZURE_CLIENT_ID`` / ``AZURE_CLIENT_SECRET`` env
+  vars would silently shadow the workload's real identity. This is the Azure
+  counterpart of the SQS ``exclude_env_credentials`` option on
+  ``AWSSQSBackend.from_iam_role``.
+- ``shared_token_cache``, ``visual_studio_code``, ``powershell``,
+  ``developer_cli`` — stale or user-scoped caches that must never authenticate
+  a production workload.
+
+Azure SDK imports are deliberately lazy so a service installing only
+``cloudrift[aws]`` never imports ``azure.identity``.
+
+Note: ``azure-identity`` is pinned ``>=1.15.0``, which predates
+``exclude_broker_credential``. It is therefore not set by default; pass it
+through ``**overrides`` if you are on a newer release and want it.
+"""
+
+_EXCLUDED_BY_DEFAULT = {
+    "exclude_environment_credential": True,
+    "exclude_shared_token_cache_credential": True,
+    "exclude_visual_studio_code_credential": True,
+    "exclude_powershell_credential": True,
+    "exclude_developer_cli_credential": True,
+}
+
+
+def _credential_options(client_id: str | None, overrides: dict) -> dict:
+    """Merge the house defaults, the managed-identity client ID, and caller overrides.
+
+    Overrides are applied last so a caller can always re-enable a source (or
+    exclude one that is on by default).
+    """
+    options = dict(_EXCLUDED_BY_DEFAULT)
+    if client_id:
+        options["managed_identity_client_id"] = client_id
+    options.update(overrides)
+    return options
+
+
+def build_async_credential(client_id: str | None = None, **overrides):
+    """Return an async ``DefaultAzureCredential`` for the standard chain.
+
+    Args:
+        client_id: Client ID of a *user-assigned* managed identity. Omit to use
+            the system-assigned identity.
+        **overrides: Passed straight to ``DefaultAzureCredential``, applied after
+            the defaults — e.g. ``exclude_cli_credential=True`` to lock a
+            production service down to managed identity only.
+
+    The caller owns the returned credential and must ``await credential.close()``.
+    """
+    from azure.identity.aio import DefaultAzureCredential
+
+    return DefaultAzureCredential(**_credential_options(client_id, overrides))
+
+
+def build_credential(client_id: str | None = None, **overrides):
+    """Return a synchronous ``DefaultAzureCredential`` for the standard chain.
+
+    Sync twin of :func:`build_async_credential`, for the backends whose SDK has
+    no async client (Redis/Entra, ACS email, MS SQL token provider).
+    """
+    from azure.identity import DefaultAzureCredential
+
+    return DefaultAzureCredential(**_credential_options(client_id, overrides))

cloudrift/crypto/azure_keyvault_keys.py

@@ -4,11 +4,7 @@
     ClientAuthenticationError,
     ResourceNotFoundError,
 )
-from azure.identity.aio import (
-    ClientSecretCredential,
-    DefaultAzureCredential,
-    ManagedIdentityCredential,
-)
+from azure.identity.aio import ClientSecretCredential
 from azure.keyvault.keys.crypto import EncryptionAlgorithm
 from azure.keyvault.keys.crypto.aio import CryptographyClient
 
@@ -34,8 +30,7 @@ class AzureKeyVaultKeysBackend(CryptoBackend):
 
     Construct via:
     - ``from_service_principal`` — tenant_id / client_id / client_secret
-    - ``from_managed_identity``  — managed identity (optionally a client_id),
-      else DefaultAzureCredential
+    - ``from_managed_identity``  — workload identity → managed identity → az CLI
     """
 
     def __init__(
@@ -77,14 +72,20 @@ def from_managed_identity(
         cls,
         key_id: str,
         client_id: str | None = None,
+        credential_options: dict | None = None,
         **kwargs,
     ) -> "AzureKeyVaultKeysBackend":
-        """Authenticate via managed identity (or DefaultAzureCredential)."""
-        credential = (
-            ManagedIdentityCredential(client_id=client_id)
-            if client_id
-            else DefaultAzureCredential()
-        )
+        """Authenticate via Azure AD: workload identity → managed identity → az CLI.
+
+        ``client_id`` selects a user-assigned managed identity; omit it for the
+        system-assigned one. ``credential_options`` is forwarded to
+        ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`.
+        (A dict rather than ``**kwargs`` here because ``**kwargs`` already
+        carries backend options such as ``algorithm``.)
+        """
+        from cloudrift.core.azure_credentials import build_async_credential
+
+        credential = build_async_credential(client_id, **(credential_options or {}))
         return cls(key_id, credential, **kwargs)
 
     # ------------------------------------------------------------------

cloudrift/email/azure_acs.py

@@ -65,15 +65,17 @@ def from_managed_identity(
         endpoint: str,
         default_from: str | None = None,
         client_id: str | None = None,
+        credential_options: dict | None = None,
     ) -> "AzureACSEmailBackend":
-        """Authenticate via Azure Managed Identity."""
-        from azure.identity import ManagedIdentityCredential
+        """Authenticate via Azure AD: workload identity → managed identity → az CLI.
 
-        credential = (
-            ManagedIdentityCredential(client_id=client_id)
-            if client_id
-            else ManagedIdentityCredential()
-        )
+        ``client_id`` selects a user-assigned managed identity; omit it for the
+        system-assigned one. ``credential_options`` is forwarded to
+        ``DefaultAzureCredential`` — see :mod:`cloudrift.core.azure_credentials`.
+        """
+        from cloudrift.core.azure_credentials import build_credential
+
+        credential = build_credential(client_id, **(credential_options or {}))
         return cls(endpoint=endpoint, default_from=default_from, credential=credential)
 
     @classmethod

cloudrift/messaging/__init__.py

@@ -2,7 +2,7 @@
     Message,
     MessagingBackend,
     OutgoingMessage,
-    send_json,
+    to_json,
 )
 
 
@@ -54,5 +54,5 @@ def get_queue(provider: str, **kwargs) -> MessagingBackend:
     "MessagingBackend",
     "OutgoingMessage",
     "get_queue",
-    "send_json",
+    "to_json",
 ]