Add encrypted pathway payload envelopes

src/pathways/builder.ts

@@ -86,6 +86,16 @@ import {
 import { FileEventSchema, FileInputSchema } from "./types.ts"
 import type { Buffer } from "node:buffer"
 import process from "node:process"
+import {
+  createPathwayEncryptionProvider,
+  decryptPayloadEnvelope,
+  encryptPayloadEnvelope,
+  PATHWAY_ENCRYPTED_METADATA_KEY,
+  PATHWAY_ENCRYPTION_SCHEME,
+  PATHWAY_ENCRYPTION_SCHEME_METADATA_KEY,
+  type PathwayEncryptionConfig,
+  type PathwayEncryptionProvider,
+} from "./encryption.ts"
 
 /**
  * Default timeout for pathway processing in milliseconds (10 seconds)
@@ -247,6 +257,15 @@ export interface PathwaysBuilderConfig {
    */
   provisionFailure?: ProvisionFailureMode | ProvisionFailureConfig
   managedConfig?: ManagedPathwayConfig
+  /**
+   * Optional symmetric encryption for pathways marked with `encrypted`.
+   *
+   * When omitted, `mode: "none"`, or `mode: "symmetric"` without a key, encrypted
+   * pathway payloads are left unchanged. When a symmetric key is supplied, the
+   * full payload is AES-256-GCM encrypted on write and decrypted before
+   * process-time schema validation and handlers.
+   */
+  encryption?: PathwayEncryptionConfig
 }
 
 /**
@@ -386,6 +405,7 @@ export class PathwaysBuilder<
   private readonly writable: Record<keyof TPathway, boolean> = {} as Record<keyof TPathway, boolean>
   private readonly subscribed: Record<keyof TPathway, boolean> = {} as Record<keyof TPathway, boolean>
   private readonly pumpGroups: Record<keyof TPathway, string> = {} as Record<keyof TPathway, string>
+  private readonly encryptedPathways: Record<keyof TPathway, boolean> = {} as Record<keyof TPathway, boolean>
   private readonly timeouts: Record<keyof TPathway, number> = {} as Record<keyof TPathway, number>
   private readonly maxRetries: Record<keyof TPathway, number> = {} as Record<keyof TPathway, number>
   private readonly retryDelays: Record<keyof TPathway, number> = {} as Record<keyof TPathway, number>
@@ -403,6 +423,7 @@ export class PathwaysBuilder<
 
   // Logger instance (but not using it yet due to TypeScript errors)
   private readonly logger: Logger
+  private readonly encryptionProvider: PathwayEncryptionProvider | null
 
   // Configuration values needed for cloning
   private readonly baseUrl: string
@@ -481,6 +502,7 @@ export class PathwaysBuilder<
     defaultAutoProvision,
     provisionFailure,
     managedConfig,
+    encryption,
   }: PathwaysBuilderConfig) {
     // Initialize logger (use NoopLogger if none provided)
     this.logger = logger ?? new NoopLogger()
@@ -505,6 +527,7 @@ export class PathwaysBuilder<
     this.dataCoreAccessControl = dataCoreAccessControl ?? "private"
     this.dataCoreDeleteProtection = dataCoreDeleteProtection ?? false
     this.provisionFailure = provisionFailure
+    this.encryptionProvider = createPathwayEncryptionProvider(encryption)
 
     // Store virtual pathway auto-provisioning config
     this.pathwayName = pathwayName
@@ -662,6 +685,8 @@ export class PathwaysBuilder<
       throw new Error(error)
     }
 
+    data.payload = this.decryptPathwayPayload(pathway, data.payload, data.metadata)
+
     // Validate event payload against schema if available
     if (this.schemas[pathway]) {
       const parsedPayload = this.schemas[pathway].safeParse(data.payload)
@@ -845,6 +870,7 @@ export class PathwaysBuilder<
       maxRetries?: number
       retryDelayMs?: number
       isFilePathway?: FP
+      encrypted?: boolean
     },
   ): PathwaysBuilder<
     & TPathway
@@ -857,6 +883,10 @@ export class PathwaysBuilder<
     const path = `${contract.flowType}/${contract.eventType}` as PathwayKey<F, E>
     const writable = contract.writable ?? true
     const subscribe = contract.subscribe ?? true
+    const encrypted = contract.encrypted === true
+    if (contract.isFilePathway && encrypted) {
+      throw new Error(`Pathway ${path} is a file pathway and cannot be encrypted`)
+    }
     if (contract.pumpGroup !== undefined && contract.pumpGroup.trim() === "") {
       throw new Error(
         `Pathway ${path} has an empty pumpGroup — pumpGroup must be a non-empty string when set`,
@@ -871,6 +901,7 @@ export class PathwaysBuilder<
       writable,
       subscribe,
       pumpGroup,
+      encrypted,
       isFilePathway: contract.isFilePathway,
       timeoutMs: contract.timeoutMs,
       maxRetries: contract.maxRetries,
@@ -921,6 +952,7 @@ export class PathwaysBuilder<
     this.writable[path] = writable
     this.subscribed[path] = subscribe
     this.pumpGroups[path] = pumpGroup
+    this.encryptedPathways[path] = encrypted
 
     // Store provisioning descriptions
     if (contract.description !== undefined) {
@@ -937,6 +969,7 @@ export class PathwaysBuilder<
       writable,
       subscribe,
       pumpGroup,
+      encrypted,
       isFilePathway: contract.isFilePathway,
     })
 
@@ -1208,11 +1241,17 @@ export class PathwaysBuilder<
       finalMetadata[AUDIT_SESSION_ID] = options.sessionId
     }
 
+    const { data: eventData, encrypted } = this.encryptPathwayPayload(path, data, Boolean(batch))
+    if (encrypted) {
+      finalMetadata[PATHWAY_ENCRYPTED_METADATA_KEY] = "true"
+      finalMetadata[PATHWAY_ENCRYPTION_SCHEME_METADATA_KEY] = PATHWAY_ENCRYPTION_SCHEME
+    }
+
     let eventIds: string | string[] = []
     this.logger.debug(`Writing webhook data to pathway`, { pathway: pathStr, batch })
     if (batch) {
       eventIds = await (this.batchWriters[path] as SendWebhookBatch<TPathway[TPath]["output"]>)(
-        data as unknown as TPathway[TPath]["output"][],
+        eventData as unknown as TPathway[TPath]["output"][],
         finalMetadata,
         options,
       ).catch((error) => {
@@ -1223,7 +1262,7 @@ export class PathwaysBuilder<
         throw error
       })
     } else if (this.filePathways.has(path)) {
-      const { fileId, fileName, fileContent, ...additionalProperties } = data as z.infer<typeof FileInputSchema>
+      const { fileId, fileName, fileContent, ...additionalProperties } = eventData as z.infer<typeof FileInputSchema>
       const fileType = await fileTypeFromBuffer(fileContent as Buffer)
       process.env.DEBUG?.includes("pathways") && console.log("additionalProperties", additionalProperties)
       eventIds = await (this.fileWriters[path] as SendFilehook)(
@@ -1246,7 +1285,7 @@ export class PathwaysBuilder<
         throw error
       })
     } else {
-      eventIds = await (this.writers[path] as SendWebhook<TPathway[TPath]["output"]>)(data, finalMetadata, options)
+      eventIds = await (this.writers[path] as SendWebhook<TPathway[TPath]["output"]>)(eventData, finalMetadata, options)
         .catch((error) => {
           this.logger.error(`Error writing webhook to pathway`, {
             pathway: pathStr,
@@ -1278,6 +1317,58 @@ export class PathwaysBuilder<
     return eventIds
   }
 
+  private encryptPathwayPayload<TPath extends keyof TPathway>(
+    path: TPath,
+    data: unknown,
+    batch: boolean,
+  ): { data: unknown; encrypted: boolean } {
+    if (!this.encryptedPathways[path] || !this.encryptionProvider) {
+      return { data, encrypted: false }
+    }
+
+    if (batch) {
+      if (!Array.isArray(data)) {
+        return { data, encrypted: false }
+      }
+      return {
+        data: data.map((item) => encryptPayloadEnvelope(item, this.encryptionProvider!)),
+        encrypted: true,
+      }
+    }
+
+    return {
+      data: encryptPayloadEnvelope(data, this.encryptionProvider),
+      encrypted: true,
+    }
+  }
+
+  private decryptPathwayPayload<TPath extends keyof TPathway>(
+    path: TPath,
+    payload: unknown,
+    metadata: unknown,
+  ): unknown {
+    if (!this.encryptedPathways[path] || !this.hasEncryptedPayloadMetadata(metadata)) {
+      return payload
+    }
+
+    if (!this.encryptionProvider) {
+      throw new Error(
+        `Pathway ${String(path)} received encrypted payload but no symmetric encryption key is configured`,
+      )
+    }
+
+    return decryptPayloadEnvelope(payload, this.encryptionProvider)
+  }
+
+  private hasEncryptedPayloadMetadata(metadata: unknown): boolean {
+    if (!metadata || typeof metadata !== "object") {
+      return false
+    }
+
+    const value = (metadata as Record<string, unknown>)[PATHWAY_ENCRYPTED_METADATA_KEY]
+    return value === true || value === "true"
+  }
+
   /**
    * Waits for a specific event to be processed
    *

src/pathways/encryption.ts

@@ -0,0 +1,114 @@
+import { createCipheriv, createDecipheriv, createHash, randomBytes } from "node:crypto"
+import { Buffer } from "node:buffer"
+
+const ALGORITHM = "aes-256-gcm"
+const IV_LENGTH = 12
+const TAG_LENGTH = 16
+const MIN_KEY_LENGTH = 32
+
+export const ENCRYPTED_PAYLOAD_FIELD = "encryptedPayload"
+export const PATHWAY_ENCRYPTED_METADATA_KEY = "pathways/encrypted"
+export const PATHWAY_ENCRYPTION_SCHEME_METADATA_KEY = "pathways/encryption-scheme"
+export const PATHWAY_ENCRYPTION_SCHEME = "aes-256-gcm-sha256-v1"
+
+export type PathwayEncryptionMode = "none" | "symmetric"
+
+export interface PathwayEncryptionConfig {
+  mode?: PathwayEncryptionMode
+  key?: string
+}
+
+export interface PathwayEncryptionProvider {
+  encrypt(plaintext: string): string
+  decrypt(payload: string): string
+}
+
+export function deriveEncryptionKey(secret: string): Buffer {
+  return createHash("sha256").update(secret).digest()
+}
+
+export function aesGcmEncrypt(plaintext: string, key: Buffer): string {
+  const iv = randomBytes(IV_LENGTH)
+  const cipher = createCipheriv(ALGORITHM, key, iv, { authTagLength: TAG_LENGTH })
+  const ciphertext = Buffer.concat([cipher.update(plaintext, "utf8"), cipher.final()])
+  const authTag = cipher.getAuthTag()
+  return `${iv.toString("base64")}.${ciphertext.toString("base64")}.${authTag.toString("base64")}`
+}
+
+export function aesGcmDecrypt(payload: string, key: Buffer): string {
+  const parts = payload.split(".")
+  if (parts.length !== 3) {
+    throw new Error("Invalid AES-256-GCM payload format")
+  }
+
+  const [ivB64, ciphertextB64, authTagB64] = parts
+  try {
+    const decipher = createDecipheriv(ALGORITHM, key, Buffer.from(ivB64, "base64"), {
+      authTagLength: TAG_LENGTH,
+    })
+    decipher.setAuthTag(Buffer.from(authTagB64, "base64"))
+    return Buffer.concat([
+      decipher.update(Buffer.from(ciphertextB64, "base64")),
+      decipher.final(),
+    ]).toString("utf8")
+  } catch {
+    throw new Error("AES-256-GCM decryption failed: auth tag mismatch (wrong key or tampered payload)")
+  }
+}
+
+export function createPathwayEncryptionProvider(
+  config?: PathwayEncryptionConfig,
+): PathwayEncryptionProvider | null {
+  const mode = config?.mode ?? (config?.key ? "symmetric" : "none")
+  if (mode === "none") {
+    return null
+  }
+
+  if (mode !== "symmetric") {
+    throw new Error(`Unknown encryption mode: ${String(mode)}`)
+  }
+
+  const secret = config?.key
+  if (!secret) {
+    return null
+  }
+
+  if (secret.length < MIN_KEY_LENGTH) {
+    throw new Error(
+      "Pathways symmetric encryption key must be at least 32 characters (generate with: openssl rand -hex 32)",
+    )
+  }
+
+  const key = deriveEncryptionKey(secret)
+  return {
+    encrypt: (plaintext: string) => aesGcmEncrypt(plaintext, key),
+    decrypt: (payload: string) => aesGcmDecrypt(payload, key),
+  }
+}
+
+export function encryptPayloadEnvelope(payload: unknown, provider: PathwayEncryptionProvider): Record<string, string> {
+  return {
+    [ENCRYPTED_PAYLOAD_FIELD]: provider.encrypt(JSON.stringify(payload)),
+  }
+}
+
+export function decryptPayloadEnvelope(payload: unknown, provider: PathwayEncryptionProvider): unknown {
+  const encryptedPayload = typeof payload === "string"
+    ? payload
+    : payload && typeof payload === "object" && !Array.isArray(payload)
+    ? (payload as Record<string, unknown>)[ENCRYPTED_PAYLOAD_FIELD]
+    : undefined
+
+  if (typeof encryptedPayload !== "string") {
+    throw new Error(`Encrypted pathway payload must be a string or an object with ${ENCRYPTED_PAYLOAD_FIELD}`)
+  }
+
+  try {
+    return JSON.parse(provider.decrypt(encryptedPayload))
+  } catch (error) {
+    if (error instanceof SyntaxError) {
+      throw new Error("Encrypted pathway payload decrypted to invalid JSON")
+    }
+    throw error
+  }
+}

src/pathways/index.ts

@@ -21,6 +21,7 @@ export type {
   ProvisionFailureMode,
 } from "./provisioner.ts"
 export * from "./constants.ts"
+export * from "./encryption.ts"
 export * from "./internal-pathway.state.ts"
 export * from "./kv/kv-adapter.ts"
 export * from "./logger.ts"

src/pathways/types.ts

@@ -85,6 +85,14 @@ export interface PathwayContract<F extends string, E extends string, T extends A
    */
   isFilePathway?: boolean
 
+  /**
+   * Enables whole-payload encryption for this pathway.
+   *
+   * When a symmetric key is configured on the builder, writes send a Flowcore-compatible
+   * encrypted envelope and processing decrypts marked events before validation.
+   */
+  encrypted?: boolean
+
   /**
    * Description for the event type. When provided, the provisioner will create/update
    * this event type on the platform. When undefined, the event type must pre-exist.