Payload encryption

Chosen fields of a job are sealed (AES-256-GCM) before the job is stored, and opened again only for the worker that runs it. They are encrypted at rest in Redis, in backups, in dead letters and in the console. This is Core.

• SDKs: list the fields in the job definition (encrypted_fields, encryptedFields, EncryptedFields); the client sends them as encrypt_fields. The SDK gateway seals them with JOOBQUI_PAYLOAD_KEYS before it writes the job and opens them when it hands the task out. The handler sees plain values; Redis never holds them. The fields travel in plain text from your app to the gateway, so reach the gateway over HTTPS.
• Crystal: declare the field with encrypted_property and an explicit job_name:; the process seals it itself with the same keys.

Crystal

struct SealedJob
  include JoobQ::Job

  @queue = SEALED
  @retries = 3
  @max_retries = 3
  @expires = 1.day.from_now.to_unix_ms

  property probe : String = ""
  property enqueued_before_ms : Int64 = 0_i64
  encrypted_property card_number : String = "", job_name: "Conf::SealedJob"

  def initialize
  end

  def initialize(@probe : String, @card_number : String, @enqueued_before_ms : Int64 = 0_i64)
  end

  def perform
    Conf.report("sealed", probe, {"card_sha256" => Conf.any(Digest::SHA256.hexdigest(card_number)), "enqueued_before_ms" => Conf.any(enqueued_before_ms)})
  end
end

Python

@job("conf.sealed", queue=QUEUE, encrypted_fields=["card_number"])
def sealed_job(payload, ctx):
    report(ctx, "sealed", payload, card_sha256=hashlib.sha256(payload["card_number"].encode()).hexdigest(),
           enqueued_before_ms=payload.get("enqueued_before_ms"))

TypeScript

const sealed = defineJob<App>({
  name: 'conf.sealed', queue: QUEUE, encryptedFields: ['card_number'],
  perform: async (p, ctx) => report(ctx, 'sealed', p, {
    card_sha256: createHash('sha256').update(p.card_number).digest('hex'), enqueued_before_ms: p.enqueued_before_ms,
  }),
})

Ruby

class SealedJob
  include JoobQ::Job
  job_name "conf.sealed"
  queue QUEUE
  encrypted_fields "card_number"
  def self.perform(p, ctx)
    report(ctx, "sealed", p, card_sha256: Digest::SHA256.hexdigest(p["card_number"]), enqueued_before_ms: p["enqueued_before_ms"])
  end
end

Go

sealed := joobq.DefineJob[*App]("conf.sealed", joobq.JobOptions{Queue: queue, EncryptedFields: []string{"card_number"}},
	func(p map[string]any, ec *EC) (any, error) {
		sum := sha256.Sum256([]byte(p["card_number"].(string)))
		return report(ec, "sealed", p, map[string]any{"card_sha256": hex.EncodeToString(sum[:]), "enqueued_before_ms": p["enqueued_before_ms"]})
	})

Rust

let sealed = Arc::new(JobDef::new("conf.sealed", queue.clone(), |p: Value, ctx: &Ctx| {
    let sum = Sha256::digest(p["card_number"].as_str().unwrap_or("").as_bytes());
    report(ctx, "sealed", &p, json!({"card_sha256": sum.iter().map(|b| format!("{:02x}", b)).collect::<String>(), "enqueued_before_ms": p["enqueued_before_ms"]}))
}).encrypted_fields(vec!["card_number".into()]));

Java

JobDefinition<App> sealed = JobDefinition.<App>of("conf.sealed", QUEUE, 3, (p, ctx) -> {
    byte[] sum = MessageDigest.getInstance("SHA-256").digest(((String) m(p).get("card_number")).getBytes(StandardCharsets.UTF_8));
    StringBuilder hex = new StringBuilder();
    for (byte b : sum) hex.append(String.format("%02x", b));
    return report(ctx, "sealed", p, map("card_sha256", hex.toString(), "enqueued_before_ms", m(p).get("enqueued_before_ms")));
}).encryptedFields("card_number");

PHP

final class SealedJob extends ConfJob
{
    public static array $encryptedFields = ['card_number'];
    public static function jobName(): string { return 'conf.sealed'; }
    public static function perform(array $p, ExecutionContext $ctx): mixed
    {
        return report($ctx, 'sealed', $p, ['card_sha256' => hash('sha256', $p['card_number']), 'enqueued_before_ms' => $p['enqueued_before_ms'] ?? null]);
    }
}

.NET

var sealedJob = JobDefinition<App>.Define("conf.sealed", Queue, 3, (p, ctx) =>
    Report(ctx, "sealed", p, new()
    {
        ["card_sha256"] = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(p!["card_number"]!.GetValue<string>()))).ToLowerInvariant(),
        ["enqueued_before_ms"] = p?["enqueued_before_ms"] == null ? null : Num(p["enqueued_before_ms"]),
    }),
    encryptedFields: new[] { "card_number" });

Tested against a real server in every language shown.

Keys

JOOBQUI_PAYLOAD_KEYS is a comma-separated list of id:key entries; a key is 64 hex digits or the base64 of 32 bytes (id:$(openssl rand -base64 32)). The first entry seals, every entry opens. Set the same value on every process that seals or opens such a job: the server (for SDK jobs) and every Crystal process.

• Without a key, the gateway refuses a job that asks for encrypt_fields (422: encrypt_fields requested, but JOOBQUI_PAYLOAD_KEYS is not configured).
• A Crystal worker whose queues include an encrypted job class refuses to start without a key, and any process refuses a malformed value.
• JOOBQUI_PAYLOAD_KEY is the earlier name, read when JOOBQUI_PAYLOAD_KEYS is unset.

Rotate by adding the new key first (it then seals), and drop the old one only when no job sealed with it can still exist (queued, delayed, retrying or dead). Renaming a Crystal job class or field makes its existing values unreadable, which is why job_name: is explicit.

Plain-text jobs of an encrypted class are dead-lettered by default, because anyone who can write to Redis could otherwise run an unsealed job. JOOBQUI_ALLOW_PLAINTEXT_PAYLOADS=true lets a worker run them, only while draining jobs written before encryption.

What it is not

It is not encryption of jobs in general: the other fields, the job's name and its queue are plain, the value is plain while the job runs and in whatever your code logs, and the length of a sealed value is not hidden.