acemq_amqp.codecs.encrypted

  1# Copyright 2026 AceMQ.
  2#
  3# Licensed under the Apache License, Version 2.0 (the "License");
  4# you may not use this file except in compliance with the License.
  5# You may obtain a copy of the License at
  6#
  7#     https://www.apache.org/licenses/LICENSE-2.0
  8#
  9# Unless required by applicable law or agreed to in writing, software
 10# distributed under the License is distributed on an "AS IS" BASIS,
 11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12# See the License for the specific language governing permissions and
 13# limitations under the License.
 14
 15"""Encrypts whatever another codec produced, so the broker holds ciphertext::
 16
 17    pip install "acemq-amqp[crypto]"
 18
 19    from acemq_amqp.codecs.encrypted import EncryptedCodec, Keyring, generate_key
 20
 21    keyring = Keyring.of("2026-01", generate_key())
 22    mq = await connect(url, codec=EncryptedCodec(JsonCodec(), keyring))
 23
 24It wraps a delegate rather than serialising anything itself, so choosing a
 25format and choosing to encrypt stay independent: JSON in, AES-GCM out, and Avro
 26just as well.
 27
 28What is on the wire
 29-------------------
 30
 31::
 32
 33    0xAE  0x01  len  key identifier   12-byte nonce   ciphertext + 16-byte tag
 34
 35**The key identifier is in the message, in the clear.** That is deliberate, and
 36it is what makes rotation possible: a consumer reads which key a message needs
 37rather than assuming the current one, so a new key can be introduced while
 38messages written with the old one are still queued. Putting it in an AMQP header
 39instead would have been tidier and would have lost it — headers are dropped by
 40shovels, rewritten by federation, and absent from a message recovered out of a
 41backup, and a ciphertext whose key nobody can name is gone.
 42
 43The header is authenticated but not encrypted: GCM binds all of it — magic,
 44version, length and identifier — as associated data, so an altered key
 45identifier makes the message fail to open rather than quietly opening as
 46something else. :func:`key_id_of` reads it back without needing any key, which
 47is what an operator staring at an unreadable dead-letter queue actually wants.
 48
 49This is the family framing, byte for byte, and **all four of the other AceMQ
 50libraries write and read it too**:
 51
 52===========  ======  =======  ==========  ==========  ================  ==========
 53library      magic   version  id length   iv          cipher            tag
 54===========  ======  =======  ==========  ==========  ================  ==========
 55Java         0xAE    0x01     1 byte      12-byte     AES-GCM           16 bytes
 56**Python**   0xAE    0x01     1 byte      12-byte     AES-GCM           16 bytes
 57Ruby         0xAE    0x01     1 byte      12-byte     AES-GCM           16 bytes
 58Go           0xAE    0x01     1 byte      12-byte     AES-GCM           16 bytes
 59.NET         0xAE    0x01     1 byte      12-byte     AES-GCM           16 bytes
 60===========  ======  =======  ==========  ==========  ================  ==========
 61
 62A body written by any of the five opens in any of the other four, given the key.
 63.NET used to be the exception and no longer is: it wrote AES-256-CBC with a
 64separate HMAC-SHA-256 up to its own 0.3.0 and moved here in the same round Go
 65dropped the magic-less variant of its own. .NET still reads the bodies it wrote
 66before that change, so a queue filled before it can be drained; nothing writes
 67that framing now, and nothing else has ever read it.
 68
 69The first byte is what makes the refusal honest. A body that does not begin
 70``0xAE`` is not this framing, and it is refused as that — visibly, naming the
 71framing — rather than being decrypted into something else or reported as a key
 72problem.
 73
 74``tests/test_encrypted.py`` holds a complete test vector for a known key, nonce
 75and plaintext. All five test suites pin the same one.
 76
 77What this does not do
 78---------------------
 79
 80The broker can no longer read the message, and neither can the people who
 81operate it. **Decide what they do instead before turning this on**: a
 82dead-letter queue full of ciphertext is a queue nobody can triage, and the
 83answer is usually a small internal tool holding the keyring rather than the
 84management UI.
 85
 86Encryption is not authorisation. Every service holding the keyring can read
 87every message encrypted with those keys; the granularity is the key, so separate
 88audiences mean separate keys. Anybody holding a key can also *write* a message
 89this codec will decrypt without complaint, so a key is a shared secret and not a
 90signature.
 91
 92Nor does it hide the routing. Exchange, routing key, headers and message size
 93stay in the clear, and for many systems the routing key is the sensitive part.
 94
 95Nothing here ever puts a plaintext, a key, or any part of either into an
 96exception, a log line or a :func:`repr`. A failure says which key identifier the
 97message named and stops there, and the tests pin it.
 98"""
 99
100from __future__ import annotations
101
102import base64
103import os
104from typing import Any
105
106from ..ack import FatalError
107from ..codec import Codec
108from ..errors import AceMQError
109
110#: What this codec writes, and what Java writes.
111#:
112#: Deliberately not ``...+json``, whatever the plaintext underneath is. A
113#: ``+json`` suffix is a promise that the bytes on the wire are JSON, and every
114#: JSON-aware consumer reads it that way. These bytes are ciphertext. Naming
115#: them ``+json`` makes the JSON codec volunteer to decode them, which is how a
116#: message ends up failing in a parser rather than being refused by a codec that
117#: knows it cannot help.
118ENCRYPTED_CONTENT_TYPE = "application/vnd.acemq.encrypted"
119
120#: Marks the framing as this codec's, so a message from elsewhere is refused
121#: rather than decrypted.
122MAGIC = 0xAE
123
124#: Version 1. Present so a later framing can be told apart from this one by its
125#: first two bytes.
126VERSION = 0x01
127
128#: GCM's nonce, in bytes. Twelve is the size the construction is defined for;
129#: any other length sends the nonce through an extra hash and buys nothing.
130NONCE_BYTES = 12
131
132#: The authentication tag, in bytes. 128 bits, which is what Java and Go use and
133#: the only length worth using: a truncated tag is a cheaper forgery.
134TAG_BYTES = 16
135
136#: The longest a key identifier may be, in UTF-8 bytes. It is one byte of
137#: framing in front of every message, so it is a name rather than a description.
138MAX_KEY_ID_BYTES = 255
139
140#: The key lengths AES takes. 32 bytes — AES-256 — is what :func:`generate_key`
141#: produces; 16 and 24 are accepted because Java accepts them and a key that
142#: already exists should not need re-issuing to be readable here.
143KEY_SIZES = (16, 24, 32)
144
145
146def generate_key() -> bytes:
147    """Draws a fresh 32-byte key from the operating system's random source.
148
149    :returns: a new AES-256 key
150    """
151    return os.urandom(32)
152
153
154def key_from_base64(encoded: str) -> bytes:
155    """Reads a key out of Base64, which is how one travels in configuration.
156
157    :param encoded: the Base64 text, with surrounding whitespace ignored
158    :returns: the key bytes
159    :raises AceMQError: when the text is not Base64, or does not decode to a
160        length AES takes. The message never quotes the input, because the input
161        is a key
162    """
163    try:
164        key = base64.b64decode(encoded.strip(), validate=True)
165    except (ValueError, TypeError) as failure:
166        # Deliberately not echoing ``encoded``: it is the key, and an exception
167        # is the one place it must never be written down.
168        raise AceMQError(
169            "acemq: this is not Base64, so no key could be read from it: "
170            f"{type(failure).__name__}"
171        ) from None
172    _check_key_size(key)
173    return key
174
175
176def key_to_base64(key: bytes) -> str:
177    """Renders a key as Base64.
178
179    The inverse of :func:`key_from_base64`, for writing a generated key into a
180    secret store. Whatever this returns is the key itself, so it belongs in the
181    same place the database password does and nowhere else.
182
183    :param key: the key bytes
184    :returns: the Base64 text
185    """
186    return base64.b64encode(key).decode("ascii")
187
188
189def _check_key_size(key: bytes) -> None:
190    if len(key) not in KEY_SIZES:
191        raise AceMQError(
192            f"acemq: an AES key is 16, 24 or 32 bytes and this one is {len(key)}. If it came "
193            "from a passphrase it needs a key derivation function such as PBKDF2, scrypt or "
194            "Argon2 rather than being used as it stands"
195        )
196
197
198class EncryptionKey:
199    """One key, and the name a message calls it by.
200
201    :param id: what the key is called on the wire. Something datelike —
202        ``2026-01`` — reads well when somebody is working out what to rotate.
203        At most 255 UTF-8 bytes, because it travels in front of every message
204    :param key: 16, 24 or 32 bytes of key material
205    :raises AceMQError: when the identifier is empty or too long, or the key is
206        not a length AES takes
207    """
208
209    __slots__ = ("_id", "_key")
210
211    def __init__(self, id: str, key: bytes) -> None:
212        if not id:
213            raise AceMQError(
214                "acemq: a key identifier cannot be empty: it is what a reader looks "
215                "the key up by"
216            )
217        encoded = id.encode("utf-8")
218        if len(encoded) > MAX_KEY_ID_BYTES:
219            raise AceMQError(
220                f"acemq: a key identifier is at most {MAX_KEY_ID_BYTES} bytes and {id!r} is "
221                f"{len(encoded)}. It travels in front of every message, so it is a name rather "
222                "than a description"
223            )
224        if "\x00" in id:
225            raise AceMQError(f"acemq: the key identifier {id!r} contains a null byte")
226        _check_key_size(key)
227        self._id = id
228        self._key = bytes(key)
229
230    @property
231    def id(self) -> str:
232        """What a message calls this key."""
233        return self._id
234
235    @property
236    def key(self) -> bytes:
237        """The key material."""
238        return self._key
239
240    def __eq__(self, other: object) -> bool:
241        if not isinstance(other, EncryptionKey):
242            return NotImplemented
243        return self._id == other._id and self._key == other._key
244
245    def __hash__(self) -> int:
246        return hash((self._id, self._key))
247
248    def __repr__(self) -> str:
249        """Names the key and never shows it.
250
251        A key logged by accident — in a dataclass dump, in an exception, in a
252        debugger's variables pane — prints its identifier and nothing else. That
253        is the whole reason this is a type rather than a tuple, and a test pins
254        it.
255        """
256        return f"EncryptionKey(id={self._id!r})"
257
258    __str__ = __repr__
259
260
261class Keyring:
262    """The keys a process can use, and which one it writes with.
263
264    More than one, because rotation needs an overlap: new messages are written
265    with the newest key while messages written with the old one are still being
266    read. A keyring with one key cannot rotate without an outage, and the order
267    to do it in is *add the new key everywhere first*, so every consumer can read
268    it, and only then :meth:`use` it somewhere.
269
270    :param keys: the keys to hold. The first is the one written with, unless
271        :meth:`use` says otherwise
272    :raises AceMQError: when given no keys at all
273    """
274
275    __slots__ = ("_current", "_keys")
276
277    def __init__(self, *keys: EncryptionKey) -> None:
278        if not keys:
279            raise AceMQError(
280                "acemq: a keyring needs at least one key; there is nothing to write with"
281            )
282        self._keys: dict[str, EncryptionKey] = {}
283        for key in keys:
284            self._keys[key.id] = key
285        self._current = keys[0].id
286
287    @classmethod
288    def of(cls, id: str, key: bytes) -> Keyring:
289        """A keyring holding one key, which is where every rotation starts.
290
291        :param id: what the key is called on the wire
292        :param key: the key material
293        :returns: the keyring
294        """
295        return cls(EncryptionKey(id, key))
296
297    def add(self, key: EncryptionKey) -> None:
298        """Puts a key on the ring without making it the one written with.
299
300        :param key: the key to hold
301        """
302        self._keys[key.id] = key
303
304    def use(self, id: str) -> None:
305        """Makes a key the one new messages are written with.
306
307        :param id: which key
308        :raises AceMQError: when that key is not on the ring
309        """
310        if id not in self._keys:
311            raise AceMQError(f"acemq: there is no key {id!r} on this keyring")
312        self._current = id
313
314    @property
315    def current(self) -> EncryptionKey:
316        """The key new messages are written with."""
317        return self._keys[self._current]
318
319    @property
320    def ids(self) -> tuple[str, ...]:
321        """The keys on the ring, in the order they were added. Never the keys
322        themselves — this is for a health endpoint or a line at start-up."""
323        return tuple(self._keys)
324
325    def key_for(self, id: str) -> EncryptionKey:
326        """Looks a key up by the name a message called it.
327
328        :param id: the identifier read off the message
329        :returns: the key
330        :raises FatalError: when the ring does not hold it. Fatal rather than
331            retryable, because the same bytes will name the same missing key on
332            every attempt and retrying only holds the queue open until the
333            message ages out
334        """
335        key = self._keys.get(id)
336        if key is None:
337            raise FatalError(
338                f"acemq: this message was encrypted with key {id!r}, which is not on this "
339                f"keyring (it holds {', '.join(self._keys) or 'nothing'}). Either the key was "
340                "retired while messages written with it were still queued, or the message came "
341                "from a service using a different keyring"
342            )
343        return key
344
345    def __repr__(self) -> str:
346        return f"Keyring(current={self._current!r}, holds={list(self._keys)!r})"
347
348
349def frame(key_id: str) -> bytes:
350    """The header a message written with ``key_id`` carries.
351
352    Magic, version, one length byte and the identifier in UTF-8 — the bytes
353    ahead of the nonce, and the bytes GCM authenticates. Public because a test
354    that wants to prove the framing has to be able to build it, and because a
355    tool reading a queue may want to recognise it.
356
357    :param key_id: the identifier to frame
358    :returns: the header bytes
359    """
360    encoded = key_id.encode("utf-8")
361    return bytes((MAGIC, VERSION, len(encoded))) + encoded
362
363
364def key_id_of(body: bytes) -> str | None:
365    """Reads which key a message needs, without needing the key.
366
367    For the operator looking at a dead-letter queue they can no longer read. The
368    identifier is in the clear in front of the ciphertext, so this answers "which
369    key does this need?" from the bytes alone — which is usually the question,
370    because a queue full of undecryptable messages is normally a key that was
371    retired too early rather than anything wrong with the messages.
372
373    :param body: a message body
374    :returns: the key identifier, or ``None`` when this was not written by this
375        codec
376    """
377    if len(body) < 3 or body[0] != MAGIC or body[1] != VERSION:
378        return None
379    length = body[2]
380    if length == 0 or len(body) < 3 + length:
381        return None
382    try:
383        return body[3 : 3 + length].decode("utf-8")
384    except UnicodeDecodeError:
385        return None
386
387
388class EncryptedCodec:
389    """Encrypts what another codec produced, and decrypts before handing it back.
390
391    :param delegate: the codec that turns objects into bytes; those bytes are
392        what gets encrypted
393    :param keyring: the keys to write with and read with
394    :raises AceMQError: when ``cryptography`` is not installed
395    """
396
397    def __init__(self, delegate: Codec, keyring: Keyring) -> None:
398        try:
399            from cryptography.hazmat.primitives.ciphers.aead import AESGCM
400        except ImportError as missing:  # pragma: no cover - depends on the install
401            raise AceMQError(
402                "acemq: EncryptedCodec needs cryptography, which is an optional extra. "
403                'Install it with pip install "acemq-amqp[crypto]"'
404            ) from missing
405
406        # AES and GCM come from cryptography, which wraps OpenSSL. Writing
407        # either by hand is how a side channel gets into a message path, and
408        # there is no version of that decision that ends well.
409        self._aesgcm: Any = AESGCM
410        self._delegate = delegate
411        self._keyring = keyring
412
413    @property
414    def content_type(self) -> str:
415        return ENCRYPTED_CONTENT_TYPE
416
417    @property
418    def delegate(self) -> Codec:
419        """The wrapped codec, whose output is what gets encrypted.
420
421        Its content type is not visible on the wire, because saying "this is
422        encrypted JSON" tells an observer more than they need. A consumer knows
423        what to expect from its own configuration.
424        """
425        return self._delegate
426
427    def encode(self, payload: Any) -> bytes:
428        key = self._keyring.current
429        header = frame(key.id)
430
431        # A fresh nonce per message, from the operating system. Reusing one
432        # under GCM does not weaken the encryption, it forfeits it: two messages
433        # under the same key and nonce leak their difference outright, and a
434        # counter that restarts with the process — which every counter in a
435        # container eventually does — reuses them by design.
436        nonce = os.urandom(NONCE_BYTES)
437
438        plaintext = self._delegate.encode(payload)
439        try:
440            sealed: bytes = self._aesgcm(key.key).encrypt(nonce, plaintext, header)
441        except Exception as failure:
442            # Without the payload, and without the key. An exception that
443            # helpfully printed what could not be encrypted would write the
444            # plaintext to the log, which is the one place it was never supposed
445            # to reach. The type of the failure is all that comes out.
446            raise AceMQError(
447                f"acemq: could not encrypt a {type(payload).__name__} with key "
448                f"{key.id!r}: {type(failure).__name__}"
449            ) from None
450        return header + nonce + sealed
451
452    def decode(self, body: bytes, content_type: str | None = None) -> Any:
453        key_id = key_id_of(body)
454        if key_id is None:
455            raise FatalError(
456                "acemq: this message was not written by EncryptedCodec — it does not "
457                "start with the framing this codec writes. A consumer configured to "
458                "decrypt has been pointed at a queue carrying something else"
459            )
460
461        header_length = 3 + len(key_id.encode("utf-8"))
462        if len(body) < header_length + NONCE_BYTES + TAG_BYTES:
463            raise FatalError(
464                "acemq: this message is too short to hold a nonce and an authentication "
465                "tag, so it was truncated somewhere between being written and being read"
466            )
467
468        key = self._keyring.key_for(key_id)
469        header = body[:header_length]
470        nonce = body[header_length : header_length + NONCE_BYTES]
471        sealed = body[header_length + NONCE_BYTES :]
472
473        try:
474            plaintext: bytes = self._aesgcm(key.key).decrypt(nonce, sealed, header)
475        except Exception:
476            # One answer for every way this can fail. GCM authenticates as well
477            # as encrypts, so a wrong key and a tampered ciphertext are the same
478            # event here, and they are told apart by nothing — no separate
479            # message, no separate type, no earlier check that would have
480            # distinguished them. An error that said which one it was would be a
481            # padding oracle with better manners.
482            raise FatalError(
483                f"acemq: this message did not decrypt with key {key_id!r}. Either that is "
484                "not the key it was written with, or it was altered after it was written"
485            ) from None
486
487        return self._delegate.decode(plaintext, self._delegate.content_type)
488
489    def can_decode(self, content_type: str | None) -> bool:
490        """Accepts only its own content type.
491
492        Volunteering for anything else means trying to decrypt plaintext and
493        reporting the failure as a decode error, which sends whoever is
494        debugging it in precisely the wrong direction. Never a message with no
495        content type either, for the same reason.
496        """
497        if not content_type:
498            return False
499        return content_type.lower().startswith(ENCRYPTED_CONTENT_TYPE)
500
501    def __repr__(self) -> str:
502        """Names the codec and the current key, and shows neither key nor body."""
503        return f"EncryptedCodec({self._delegate!r}, key={self._keyring.current.id!r})"