Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/encp-resume-hook-seal.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@workflow/core': minor
---

`resumeHook()` now seals its payload to the target run's published public key when one is present, removing the cross-deployment `run-key` API round trip from hook resumption. Readers resolve the full key capability so they can open sealed payloads.
4 changes: 2 additions & 2 deletions packages/core/src/private.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

import { withResolvers } from '@workflow/utils';
import type { WorldCapabilities } from '@workflow/world';
import type { CryptoKey } from './encryption.js';
import type { PayloadKey } from './serialization/encryption.js';
import type { EventsConsumer } from './events-consumer.js';
import type { QueueItem } from './global.js';
import type { ReplayPayloadCache } from './replay-payload-cache.js';
Expand Down Expand Up @@ -132,7 +132,7 @@ export function getStepFunction(stepId: string): StepFunction | undefined {

export interface WorkflowOrchestratorContext {
runId: string;
encryptionKey: CryptoKey | undefined;
encryptionKey: PayloadKey | undefined;
worldCapabilities?: WorldCapabilities;
globalThis: typeof globalThis;
eventsConsumer: EventsConsumer;
Expand Down
4 changes: 2 additions & 2 deletions packages/core/src/replay-payload-cache.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import type { Event, WorkflowRun } from '@workflow/world';
import type { CryptoKey } from './encryption.js';
import type { PayloadKey } from './serialization/encryption.js';
import {
type PreparedReplayPayload,
prepareReplayPayload,
Expand Down Expand Up @@ -43,7 +43,7 @@ export class ReplayPayloadCache {
private nextUnscannedEventIndex = 0;

constructor(
private readonly encryptionKey: CryptoKey | undefined,
private readonly encryptionKey: PayloadKey | undefined,
private readonly preparer: ReplayPayloadPreparer = prepareReplayPayload
) {}

Expand Down
72 changes: 72 additions & 0 deletions packages/core/src/runtime/helpers.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,20 @@ import { PreconditionFailedError, WorkflowWorldError } from '@workflow/errors';
import type { Event, World } from '@workflow/world';
import { ulid } from 'ulid';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { deriveRunKeyPair, seal } from '../sealed-box.js';
import {
decrypt,
encodeWithFormatPrefix,
encrypt,
peekFormatPrefix,
SerializationFormat,
} from '../serialization.js';
import {
getWorkflowQueueName,
healthCheck,
latestEventStateUpdatedAt,
loadWorkflowRunEvents,
memoizeEncryptionKey,
type MutableEventLog,
withPreconditionRetry,
} from './helpers.js';
Expand Down Expand Up @@ -638,3 +647,66 @@ describe('withPreconditionRetry', () => {
expect(eventsListMock).not.toHaveBeenCalled();
});
});

describe('memoizeEncryptionKey', () => {
const MATERIAL = new Uint8Array(32).fill(0x6b);

function worldWithKey(getEncryptionKeyForRun: unknown): World {
return { getEncryptionKeyForRun } as unknown as World;
}

it('resolves a key that can open payloads sealed to the run', async () => {
// A run reading its own event log may find sealed ('encp') payloads that
// another run wrote to it — a cross-deployment hook resumption, say. If
// this resolved only the symmetric key, those payloads would fail to open
// and wedge the run, so the sealed capability must be part of what every
// reader gets by default.
const getKey = memoizeEncryptionKey(
worldWithKey(vi.fn().mockResolvedValue(MATERIAL)),
'wrun_1'
);
const resolved = await getKey();
expect(resolved).toBeDefined();

const { publicKey } = await deriveRunKeyPair(MATERIAL);
const sealed = await seal(publicKey, new TextEncoder().encode('"hi"'));
const prefixed = encodeWithFormatPrefix(
SerializationFormat.SEALED,
sealed
) as Uint8Array;

await expect(decrypt(prefixed, resolved)).resolves.toEqual(
new TextEncoder().encode('"hi"')
);
});

it("resolves a key that still opens the run's own symmetric payloads", async () => {
const getKey = memoizeEncryptionKey(
worldWithKey(vi.fn().mockResolvedValue(MATERIAL)),
'wrun_1'
);
const resolved = await getKey();

const encrypted = await encrypt(new TextEncoder().encode('"hi"'), resolved);
expect(peekFormatPrefix(encrypted)).toBe(SerializationFormat.ENCRYPTED);
await expect(decrypt(encrypted, resolved)).resolves.toEqual(
new TextEncoder().encode('"hi"')
);
});

it('memoizes so the key is derived once per run', async () => {
// Derivation now costs several Web Crypto round trips (HKDF + a PKCS#8
// import + a JWK export), so re-deriving per payload would be wasteful.
const spy = vi.fn().mockResolvedValue(MATERIAL);
const getKey = memoizeEncryptionKey(worldWithKey(spy), 'wrun_1');

const [a, b] = await Promise.all([getKey(), getKey()]);
expect(a).toBe(b);
expect(spy).toHaveBeenCalledTimes(1);
});

it('resolves undefined when encryption is not configured', async () => {
const getKey = memoizeEncryptionKey(worldWithKey(undefined), 'wrun_1');
await expect(getKey()).resolves.toBeUndefined();
});
});
36 changes: 25 additions & 11 deletions packages/core/src/runtime/helpers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,11 @@ import {
ulidToDate,
} from '@workflow/world';
import { monotonicFactory } from 'ulid';

import { type CryptoKey, importKey } from '../encryption.js';
import { runtimeLogger } from '../logger.js';
import {
deriveRunPayloadKeys,
type PayloadKey,
} from '../serialization/encryption.js';
import * as Attribute from '../telemetry/semantic-conventions.js';
import { getSpanKind, trace } from '../telemetry.js';
import { version as workflowCoreVersion } from '../version.js';
Expand Down Expand Up @@ -792,13 +794,21 @@ export function getQueueOverhead(message: { requestedAt?: Date }) {
}

/**
* Returns a memoized accessor for the per-run AES-256 encryption key.
* Returns a memoized accessor for a run's full encryption capability.
*
* The first call resolves the run's key material via
* `world.getEncryptionKeyForRun` (which may do HKDF derivation locally on
* Vercel, or a network fetch from external contexts) and derives a
* {@link PayloadKey} from it; subsequent calls await the same cached promise.
* If the world doesn't support encryption or the run has no key configured,
* the cached value is `undefined`.
*
* The first call resolves the key via `world.getEncryptionKeyForRun` (which
* may do HKDF derivation locally on Vercel, or a network fetch from
* external contexts) and imports it as a `CryptoKey`; subsequent calls
* await the same cached promise. If the world doesn't support encryption
* or the run has no key configured, the cached value is `undefined`.
* The resolved value is deliberately the *full* capability — the symmetric AES
* key plus the run's X25519 keypair — not just a `CryptoKey`. A run reading
* its own event log can encounter sealed (`encp`) payloads that another run
* wrote to it (a cross-deployment hook resumption, say), and opening those
* needs the keypair. Resolving only the symmetric key would leave those
* payloads unopenable and wedge the run.
*
* Used by step / workflow handlers to defer the (potentially expensive)
* key fetch until the first code path that actually needs it — typically
Expand All @@ -816,8 +826,8 @@ export function getQueueOverhead(message: { requestedAt?: Date }) {
export function memoizeEncryptionKey(
world: World,
runOrId: WorkflowRun | string
): () => Promise<CryptoKey | undefined> {
let cached: Promise<CryptoKey | undefined> | undefined;
): () => Promise<PayloadKey | undefined> {
let cached: Promise<PayloadKey | undefined> | undefined;
return () => {
if (!cached) {
cached = (async () => {
Expand All @@ -828,7 +838,11 @@ export function memoizeEncryptionKey(
typeof runOrId === 'string'
? await world.getEncryptionKeyForRun?.(runOrId)
: await world.getEncryptionKeyForRun?.(runOrId);
return rawKey ? await importKey(rawKey) : undefined;
// Resolve the *full* capability, not just the symmetric key: a run
// reading its own event log may encounter sealed (`encp`) payloads
// that another run wrote to it, and opening those needs the run's
// X25519 scalar as well.
return rawKey ? await deriveRunPayloadKeys(rawKey) : undefined;
})();
}
return cached;
Expand Down
Loading
Loading