Class: AceMQ::AMQP::EncryptedCodec

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/crypto.rb

Overview

Encrypts whatever another codec produced, so the broker holds ciphertext.

keys = Keyring.of("orders-2026-09", Keys.generate)
codec = EncryptedCodec.wrapping(JSONCodec.new, keys)

mq = Connection.open(url, codec: codec)

It wraps a delegate rather than serialising anything itself, so choosing a format and choosing to encrypt stay independent: JSON in, AES-GCM out, and Avro just as well.

What is on the wire

0xAE  0x01  len  key identifier   12-byte nonce   ciphertext + 16-byte tag

Byte for byte what the Java, Go, .NET and Python libraries write, and what they read. The key identifier is in the message, in the clear. That is deliberate, and it is what makes rotation possible: a consumer reads which key a message needs rather than assuming the current one, so a new key can be introduced while messages written with the old one are still queued. Putting it in an AMQP header instead would have been tidier and would have lost it — headers are dropped by shovels, rewritten by federation, and absent from a message recovered out of a backup, and a ciphertext whose key nobody can name is gone.

The header is authenticated but not encrypted: GCM binds it as associated data, so an altered key identifier fails to decrypt rather than quietly decrypting as something else.

What this does not do

The broker can no longer read the message, and neither can the people who operate it. Decide what they do instead before turning this on: a dead-letter queue full of ciphertext is a queue nobody can triage, and the answer is usually a small internal tool holding the keyring rather than the management interface. EncryptedCodec.key_id_of tells an operator which key a message needs without holding any of them.

Encryption is not authorisation. Every service holding the keyring can read every message encrypted with those keys; the granularity is the key, so separate audiences mean separate keys. Nor does it authenticate the sender: anybody holding the key can write a message this codec will happily decrypt.

Nor does it hide the routing. Exchange, routing key, headers and message size stay in the clear, and for many systems the routing key is the sensitive part.

Constant Summary collapse

CONTENT_TYPE =

Deliberately not +...+json+, whatever the plaintext underneath is.

A json suffix is a promise that the bytes on the wire are JSON, and every JSON-aware consumer reads it that way. These bytes are ciphertext. Naming them json makes the JSON codec volunteer to decode them, which is how a message ends up failing in a parser rather than being refused by a codec that knows it cannot help.

"application/vnd.acemq.encrypted"
MAGIC =

Marks the framing as this codec's, so a message from elsewhere is refused rather than decrypted.

0xAE
VERSION =

Version 1. Present so a later framing can be told apart from this one by its first two bytes.

0x01
NONCE_BYTES =

Fixed at 12, which is the size GCM is defined for. Any other length is hashed into shape by the construction and loses the guarantee that two different nonces stay different.

12
TAG_BYTES =

A 128-bit authentication tag, as Java and Go both write.

16
PREFIX_BYTES =

The magic, the version and the length byte.

3

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(delegate, keyring) ⇒ EncryptedCodec

Returns a new instance of EncryptedCodec.



367
368
369
370
371
372
 
# File 'lib/acemq/amqp/crypto.rb', line 367

def initialize(delegate, keyring)
  @delegate = Codec.check!(delegate)
  @keyring = Keyring.check!(keyring)
  @delegate_takes_content_type = @delegate.method(:decode).arity != 1
  freeze
end

Instance Attribute Details

#delegate ⇒ Object (readonly)

The wrapped codec, whose output is what gets encrypted.



375
376
377
 
# File 'lib/acemq/amqp/crypto.rb', line 375

def delegate
  @delegate
end

Class Method Details

.header_for(key) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

The bytes GCM authenticates but does not encrypt.



464
465
466
467
 
# File 'lib/acemq/amqp/crypto.rb', line 464

def self.header_for(key)
  id = key.id_bytes
  [MAGIC, VERSION, id.bytesize].pack("C3") + id
end

.key_id_of(body) ⇒ String?

Reads which key a message needs, without needing the key.

For the operator looking at a dead-letter queue they can no longer read. The identifier is in the clear in front of the ciphertext, so this answers "which key does this need?" from the bytes alone — which is usually the question, because a queue full of undecryptable messages is normally a key that was retired too early rather than anything wrong with the messages.

Parameters:

  • body (String) —

    a message body

Returns:

  • (String, nil) —

    the key identifier, or nil when this was not written by this codec



450
451
452
453
454
455
456
457
458
459
 
# File 'lib/acemq/amqp/crypto.rb', line 450

def self.key_id_of(body)
  bytes = body.to_s.dup.force_encoding(Encoding::BINARY)
  return nil if bytes.bytesize < PREFIX_BYTES
  return nil unless bytes.getbyte(0) == MAGIC && bytes.getbyte(1) == VERSION

  length = bytes.getbyte(2)
  return nil if length.zero? || bytes.bytesize < PREFIX_BYTES + length

  bytes.byteslice(PREFIX_BYTES, length).force_encoding(Encoding::UTF_8)
end

.wrapping(delegate, keyring) ⇒ EncryptedCodec

Parameters:

  • delegate (#encode, #decode) —

    the codec whose output is encrypted

  • keyring (#current, #key_for) —

    the keys to write with and read with

Returns:



365
 
# File 'lib/acemq/amqp/crypto.rb', line 365

def self.wrapping(delegate, keyring) = new(delegate, keyring)

Instance Method Details

#can_decode?(content_type) ⇒ Boolean

Only its own. Volunteering for anything else means trying to decrypt plaintext and reporting the failure as a decode error, which sends whoever is debugging it in precisely the wrong direction.

Returns:

  • (Boolean) 


434
435
436
 
# File 'lib/acemq/amqp/crypto.rb', line 434

def can_decode?(content_type)
  content_type.to_s.downcase.start_with?(CONTENT_TYPE)
end

#content_type ⇒ Object 



377
 
# File 'lib/acemq/amqp/crypto.rb', line 377

def content_type = CONTENT_TYPE

#decode(body) ⇒ Object

Returns what the delegate made of the plaintext.

Parameters:

  • body (String) —

    the bytes as they arrived

Returns:

  • (Object) —

    what the delegate made of the plaintext

Raises:

  • (DecodeError) —

    when it was not written by this codec, names a key this keyring does not hold, or will not decrypt



411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
 
# File 'lib/acemq/amqp/crypto.rb', line 411

def decode(body)
  bytes = body.to_s.dup.force_encoding(Encoding::BINARY)
  key_id = self.class.key_id_of(bytes)
  if key_id.nil?
    raise DecodeError,
          "this message was not written by EncryptedCodec: it does not start with the " \
          "framing this codec writes. A consumer configured to decrypt has been " \
          "pointed at a queue carrying plaintext."
  end

  header_length = PREFIX_BYTES + key_id.bytesize
  if bytes.bytesize < header_length + NONCE_BYTES + TAG_BYTES
    raise DecodeError,
          "this message is too short to hold a nonce and an authentication tag, so it " \
          "was truncated somewhere between being written and being read."
  end

  decode_inner(unseal(bytes, header_length, @keyring.key_for(key_id)))
end

#encode(payload) ⇒ String

Returns the framed ciphertext, binary.

Parameters:

  • payload (Object) —

    anything the delegate will encode

Returns:

  • (String) —

    the framed ciphertext, binary

Raises:



382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
 
# File 'lib/acemq/amqp/crypto.rb', line 382

def encode(payload)
  key = @keyring.current
  header = self.class.header_for(key)
  # A fresh nonce per message, from OpenSSL's own source. Reusing one
  # under GCM does not weaken the encryption, it forfeits it: two
  # messages under the same key and nonce leak their difference outright,
  # so a counter is not an option however tempting it looks.
  nonce = OpenSSL::Random.random_bytes(NONCE_BYTES)
  plaintext = @delegate.encode(payload)

  cipher = OpenSSL::Cipher.new(key.cipher_name)
  cipher.encrypt
  cipher.key = key.secret
  cipher.iv = nonce
  cipher.auth_data = header
  sealed = cipher.update(plaintext.to_s) + cipher.final
  header + nonce + sealed + cipher.auth_tag(TAG_BYTES)
rescue OpenSSL::OpenSSLError => e
  # Without the payload. An exception that helpfully printed what could
  # not be encrypted would write the plaintext to the log, which is the
  # one place it was never supposed to reach.
  raise EncodeError,
        "could not encrypt a #{payload.class} with key #{key.id.inspect}: #{e.class}"
end

#inspect ⇒ Object 



470
 
# File 'lib/acemq/amqp/crypto.rb', line 470

def inspect = "#<AceMQ::AMQP::EncryptedCodec #{self}>"

#to_s ⇒ Object 



469
 
# File 'lib/acemq/amqp/crypto.rb', line 469

def to_s = "EncryptedCodec{#{@delegate}, key=#{@keyring.current.id}}"