diff --git a/docs/api-overview.md b/docs/api-overview.md index 895ee2ba9..726fe9834 100644 --- a/docs/api-overview.md +++ b/docs/api-overview.md @@ -316,6 +316,94 @@ counts as not matched (`false`), and an `evaluator` expression that cannot be evaluated fails the flow with `Failed to evaluate expression: ...`, including the `ExpressionError` detail. +## Triggers + +Trigger adapters (`@loushy/build-ai-agent/triggers`) wake an agent up from an +inbound webhook, a schedule or a Slack message. Wire any of them with +`listen(agent, onEvent)`, where `onEvent` runs the agent. + +### Webhook authentication + +`WebhookTriggerAdapter` starts an HTTP server. **Always set `auth` for a +webhook that is reachable from outside your machine**: without it, anyone who +can reach the port can run your agent (and spend your tokens). If you listen +on a non-loopback host with no `auth`, the adapter logs a one-time warning +through `options.logger`. + +```ts +import { createAgent, createMockProvider } from '@loushy/build-ai-agent'; +import { WebhookTriggerAdapter } from '@loushy/build-ai-agent/triggers'; + +const agent = createAgent({ prompt: 'You are helpful.', provider: createMockProvider() }); + +// HMAC of the RAW request body (GitHub / Shopify style): header `x-signature-256: sha256=`. +new WebhookTriggerAdapter({ + port: 8787, + auth: { type: 'hmac', secret: process.env.WEBHOOK_SECRET ?? '' }, +}).listen(agent, (input) => agent.send(input)); + +// Replay protection: the signed payload becomes `${timestamp}.${body}` and +// requests more than `toleranceSeconds` (default 300) old are rejected. +new WebhookTriggerAdapter({ + auth: { + type: 'hmac', + secret: process.env.WEBHOOK_SECRET ?? '', + header: 'x-signature', + timestampHeader: 'x-timestamp', + toleranceSeconds: 120, + }, +}); + +// A shared bearer token (`Authorization: Bearer `). +new WebhookTriggerAdapter({ auth: { type: 'bearer', token: process.env.WEBHOOK_TOKEN ?? '' } }); + +// Anything else: return true to accept. `rawBody` is a Buffer of the exact bytes received. +new WebhookTriggerAdapter({ + auth: { type: 'custom', verify: (req) => req.headers['x-api-key'] === process.env.API_KEY }, +}); +``` + +HMAC options: `header` (default `x-signature-256`), `algorithm` (`sha256` or +`sha1`, default `sha256`), `prefix` (default `sha256=`; `''` for a bare +digest), `timestampHeader` and `toleranceSeconds`. Signatures and bearer +tokens are compared in constant time. A request that fails authentication gets +a generic `401 {"error":"Unauthorized"}` - the response never says which check +failed - and the reason (never a secret or signature) is logged at `warn` +level. Serve webhooks over HTTPS (terminate TLS in front of the adapter) so +tokens and payloads are not sent in clear text. + +### Cron schedules + +`CronTriggerAdapter` takes either a fixed `intervalMs` or a real cron +expression: + +```ts +import { createAgent, createMockProvider } from '@loushy/build-ai-agent'; +import { CronTriggerAdapter } from '@loushy/build-ai-agent/triggers'; + +const agent = createAgent({ prompt: 'You are helpful.', provider: createMockProvider() }); + +new CronTriggerAdapter({ + cron: '*/15 9-17 * * MON-FRI', // minute hour day-of-month month day-of-week + timezone: 'Europe/Paris', // IANA name; defaults to the machine's local zone + input: 'Check the support queue', + onResult: (result, error) => console.log(error ?? result?.text), +}).listen(agent, (input) => agent.send(input)); +``` + +Supported syntax: `*`, lists (`1,15`), ranges (`1-5`), steps (`*/15`, +`10-40/10`), month names (`JAN`) and weekday names (`MON`), with `0` and `7` +both meaning Sunday, plus `@hourly`, `@daily`, `@weekly` and `@monthly`. As in +classic cron, when both day-of-month and day-of-week are restricted a day +matches if either does. An invalid expression throws a `CronExpressionError` +naming the field and showing a valid example. `parseCronExpression(expr, +timezone).nextRun(after)` is exported if you need the next fire time. + +Around daylight-saving changes, a time that does not exist (spring forward) is +skipped for that day, and a time that happens twice (fall back) fires once; +an every-hour schedule keeps firing hourly. The timer is re-armed after each +run from the scheduled time (no drift, no double fire), and `stop()` clears it. + ## Deployment - `DeploymentAdapter`, `registerAdapter()`, `getAdapter()`, `listAdapters()` - diff --git a/src/triggers/adapters/CronTriggerAdapter.test.ts b/src/triggers/adapters/CronTriggerAdapter.test.ts index 19e19776c..94e3772bc 100644 --- a/src/triggers/adapters/CronTriggerAdapter.test.ts +++ b/src/triggers/adapters/CronTriggerAdapter.test.ts @@ -96,3 +96,112 @@ describe('CronTriggerAdapter', () => { expect(adapter.reply).toBeUndefined(); }); }); + +describe('CronTriggerAdapter with a cron expression', () => { + const start = (options: { cron: string; timezone?: string; fireImmediately?: boolean }) => { + const onEvent = vi.fn().mockResolvedValue(fakeResult('tick')); + const onResult = vi.fn(); + const adapter = new CronTriggerAdapter({ input: 'scheduled', onResult, ...options }); + return { onEvent, onResult, handle: adapter.listen(noopAgent, onEvent) }; + }; + + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-01-01T00:00:30Z')); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('fires at each matching minute, without drift or double-firing, with a single timer', async () => { + const { onEvent, onResult, handle } = start({ cron: '* * * * *' }); + expect(vi.getTimerCount()).toBe(1); + + await vi.advanceTimersByTimeAsync(29_999); + expect(onEvent).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(onEvent).toHaveBeenCalledTimes(1); + expect(onEvent).toHaveBeenCalledWith( + 'scheduled', + expect.objectContaining({ firedAt: '2026-01-01T00:01:00.000Z', tick: 0 }) + ); + expect(onResult).toHaveBeenCalledWith(fakeResult('tick'), undefined, expect.anything()); + expect(vi.getTimerCount()).toBe(1); // re-armed + + await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(onEvent).toHaveBeenCalledTimes(11); + expect(new Date().toISOString()).toBe('2026-01-01T00:11:00.000Z'); + handle.stop(); + }); + + it('honours the time zone', async () => { + const { onEvent, handle } = start({ cron: '0 9 * * *', timezone: 'Asia/Kolkata' }); + await vi.advanceTimersByTimeAsync(3 * 3600_000 + 29 * 60_000); // 03:29:30Z + expect(onEvent).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(60_000); + expect(onEvent).toHaveBeenCalledTimes(1); + handle.stop(); + }); + + it("chains timers for delays beyond setTimeout's maximum (~24.8 days) and fires once", async () => { + const { onEvent, handle } = start({ cron: '0 0 1 1 *', timezone: 'UTC' }); // next: 2027-01-01, ~365 days away + expect(vi.getTimerCount()).toBe(1); + await vi.advanceTimersByTimeAsync(364 * 86_400_000); + expect(onEvent).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(1); + await vi.advanceTimersByTimeAsync(2 * 86_400_000); + expect(onEvent).toHaveBeenCalledTimes(1); + expect(onEvent).toHaveBeenCalledWith( + 'scheduled', + expect.objectContaining({ firedAt: '2027-01-01T00:00:00.000Z' }) + ); + handle.stop(); + }); + + it('fireImmediately fires once right away in addition to the schedule', async () => { + const { onEvent, handle } = start({ cron: '* * * * *', fireImmediately: true }); + expect(onEvent).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(30_000); + expect(onEvent).toHaveBeenCalledTimes(2); + handle.stop(); + }); + + it('keeps the schedule alive when a run fails', async () => { + const onEvent = vi.fn().mockRejectedValue(new Error('boom')); + const onResult = vi.fn(); + const handle = new CronTriggerAdapter({ cron: '* * * * *', input: 'x', onResult }).listen(noopAgent, onEvent); + await vi.advanceTimersByTimeAsync(90_000); + expect(onEvent).toHaveBeenCalledTimes(2); + expect(onResult).toHaveBeenCalledWith(undefined, expect.any(Error), expect.anything()); + handle.stop(); + }); + + it('stop() clears the timer and nothing fires afterwards', async () => { + const { onEvent, handle } = start({ cron: '* * * * *' }); + handle.stop(); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(10 * 60_000); + expect(onEvent).not.toHaveBeenCalled(); + }); + + it('skips missed runs after a long suspension instead of firing a burst', async () => { + const { onEvent, handle } = start({ cron: '* * * * *' }); + vi.setSystemTime(new Date('2026-01-01T05:00:10Z')); // clock jumped (e.g. laptop sleep) + await vi.advanceTimersByTimeAsync(30_000); + expect(onEvent).toHaveBeenCalledTimes(1); + handle.stop(); + }); + + it('validates its options', () => { + const base = { input: 'x', onResult: vi.fn() }; + expect(() => new CronTriggerAdapter({ ...base, cron: 'nope' })).toThrow(/expected 5 space-separated fields/); + expect(() => new CronTriggerAdapter({ ...base, cron: '* * * * *', timezone: 'Nowhere/Land' })).toThrow( + /timezone/ + ); + expect(() => new CronTriggerAdapter({ ...base } as never)).toThrow(/exactly one of/); + expect(() => new CronTriggerAdapter({ ...base, intervalMs: 1000, cron: '* * * * *' } as never)).toThrow( + /exactly one of/ + ); + }); +}); diff --git a/src/triggers/adapters/CronTriggerAdapter.ts b/src/triggers/adapters/CronTriggerAdapter.ts index 95cdd397c..86dbfc008 100644 --- a/src/triggers/adapters/CronTriggerAdapter.ts +++ b/src/triggers/adapters/CronTriggerAdapter.ts @@ -1,15 +1,10 @@ /** - * Cron trigger adapter (LOU-T5). + * Cron trigger adapter (LOU-T5, LOU-D13). * - * Fires an agent run on a fixed interval rather than in response to an - * inbound request. Deliberately scheduled by a plain millisecond interval - * (`intervalMs`) rather than parsing real cron syntax ('* * * * *') - this - * SDK has no existing cron-expression dependency, and adding one (plus a - * scheduler loop, timezone handling, etc.) is out of scope for landing the - * TriggerAdapter interface itself. `intervalMs` covers the same "run - * periodically" need; a `CronExpressionTriggerAdapter` that layers real - * cron syntax on top of this same interface is a natural, separately - * scoped follow-up. + * Fires an agent run on a schedule rather than in response to an inbound + * request. The schedule is either a fixed interval (`intervalMs`) or a + * standard 5-field cron expression (`cron`, with an optional IANA + * `timezone`), parsed by the dependency-free parser in ../cronExpression.ts. * * Reply semantics: there is no caller waiting on a response the way an * HTTP request or a Slack message is - so results are handed to the @@ -20,24 +15,85 @@ import type { ExecutionResult } from '../../execution/AgentExecutor'; import { RunnableAgent, TriggerAdapter, TriggerContext, TriggerHandle } from '../types'; -export interface CronTriggerAdapterOptions { - /** How often to fire, in milliseconds. Must be a positive integer. */ - intervalMs: number; +import { CronSchedule, parseCronExpression } from '../cronExpression'; + +interface CronTriggerAdapterBaseOptions { /** The input to run the agent with on each tick. */ input: string; /** Called with the outcome of each scheduled run (success or failure). */ onResult: (result: ExecutionResult | undefined, error: unknown, context: TriggerContext) => void; - /** If true, fires once immediately in addition to the interval. Defaults to false. */ + /** If true, fires once immediately in addition to the schedule. Defaults to false. */ fireImmediately?: boolean; } +/** Fire every `intervalMs` milliseconds. */ +export interface CronIntervalOptions extends CronTriggerAdapterBaseOptions { + /** How often to fire, in milliseconds. Must be a positive number. */ + intervalMs: number; + cron?: never; + timezone?: never; +} + +/** Fire on a cron expression. */ +export interface CronExpressionOptions extends CronTriggerAdapterBaseOptions { + /** + * Standard 5-field cron expression (`minute hour day-of-month month day-of-week`) + * or `@hourly` / `@daily` / `@weekly` / `@monthly`. Supports `*`, lists, ranges, + * steps and month/weekday names; 0 and 7 are both Sunday. + */ + cron: string; + /** IANA time zone the expression is evaluated in (e.g. `'America/New_York'`). Defaults to the machine's local zone. */ + timezone?: string; + intervalMs?: never; +} + +/** + * Options for {@link CronTriggerAdapter}: pass either `intervalMs` or `cron`. + * + * @example + * ```ts + * new CronTriggerAdapter({ + * cron: '0 9 * * MON-FRI', + * timezone: 'Europe/Paris', + * input: 'Summarise yesterday', + * onResult: (result) => console.log(result?.text), + * }); + * ``` + */ +export type CronTriggerAdapterOptions = CronIntervalOptions | CronExpressionOptions; + +/** setTimeout stores its delay in a signed 32-bit int; anything larger fires immediately. */ +const MAX_TIMEOUT_MS = 2 ** 31 - 1; + +type Schedule = { kind: 'interval'; intervalMs: number } | { kind: 'cron'; schedule: CronSchedule }; + +function resolveSchedule(options: CronTriggerAdapterOptions): Schedule { + const hasInterval = options.intervalMs !== undefined; + if (hasInterval === (options.cron !== undefined)) { + throw new Error("CronTriggerAdapter: pass exactly one of options.intervalMs or options.cron (e.g. { cron: '*/5 * * * *' })."); + } + if (options.cron !== undefined) { + return { kind: 'cron', schedule: parseCronExpression(options.cron, options.timezone) }; + } + const intervalMs = options.intervalMs as number; + if (!Number.isFinite(intervalMs) || intervalMs <= 0) { + throw new Error('CronTriggerAdapter: options.intervalMs must be a positive number'); + } + return { kind: 'interval', intervalMs }; +} + +/** Don't keep the process alive solely because a cron trigger is listening; callers that want that hold their own handle. */ +function unref(timer: NodeJS.Timeout): void { + if (typeof timer.unref === 'function') timer.unref(); +} + export class CronTriggerAdapter implements TriggerAdapter { public readonly type = 'cron'; + private readonly schedule: Schedule; + constructor(private readonly options: CronTriggerAdapterOptions) { - if (!Number.isFinite(options.intervalMs) || options.intervalMs <= 0) { - throw new Error('CronTriggerAdapter: options.intervalMs must be a positive number'); - } + this.schedule = resolveSchedule(options); } public listen( @@ -57,19 +113,53 @@ export class CronTriggerAdapter implements TriggerAdapter { if (this.options.fireImmediately) { fire(); } - const timer = setInterval(fire, this.options.intervalMs); - // Don't keep the process alive solely because a cron trigger is - // listening (mirrors how e.g. Node's own http/net timers behave when - // `.unref()`-ed); a caller that wants to keep the process alive should - // hold its own reference/handle open. - if (typeof timer.unref === 'function') { - timer.unref(); - } - - return { - stop: () => { - clearInterval(timer); - }, - }; + return this.schedule.kind === 'interval' + ? startInterval(this.schedule.intervalMs, fire) + : startCron(this.schedule.schedule, fire); } } + +function startInterval(intervalMs: number, fire: () => void): TriggerHandle { + const timer = setInterval(fire, intervalMs); + unref(timer); + return { stop: () => clearInterval(timer) }; +} + +/** + * One timer, always set to the next run. Each run is computed from the + * previous scheduled time (cron times are absolute, so nothing drifts) and + * strictly after it (so nothing fires twice). Delays beyond setTimeout's + * ~24.8 day limit are chained. + */ +function startCron(schedule: CronSchedule, fire: () => void): TriggerHandle { + let timer: NodeJS.Timeout | undefined; + let stopped = false; + + const waitUntil = (target: number) => { + const delay = Math.max(0, target - Date.now()); + timer = setTimeout(() => { + if (stopped) return; + if (Date.now() < target) { + waitUntil(target); + return; + } + // Re-arm first so a throwing handler can never end the schedule. If the + // process was suspended past later runs, skip them rather than firing a burst. + arm(new Date(Math.max(target, Date.now()))); + fire(); + }, Math.min(delay, MAX_TIMEOUT_MS)); + unref(timer); + }; + + const arm = (after: Date) => { + if (!stopped) waitUntil(schedule.nextRun(after).getTime()); + }; + + arm(new Date()); + return { + stop: () => { + stopped = true; + if (timer) clearTimeout(timer); + }, + }; +} diff --git a/src/triggers/adapters/WebhookTriggerAdapter.auth.test.ts b/src/triggers/adapters/WebhookTriggerAdapter.auth.test.ts new file mode 100644 index 000000000..d58052c8e --- /dev/null +++ b/src/triggers/adapters/WebhookTriggerAdapter.auth.test.ts @@ -0,0 +1,234 @@ +import { describe, it, expect, vi, afterEach } from 'vitest'; +import http from 'node:http'; +import { createHmac } from 'node:crypto'; +import { WebhookTriggerAdapter, WebhookTriggerHandle, WebhookTriggerAdapterOptions } from './WebhookTriggerAdapter'; +import { ExecutionResult } from '../../execution/AgentExecutor'; +import { Logger } from '../../execution/logger'; +import { RunnableAgent } from '../types'; + +const SECRET = 'whsec_test_secret'; +const result: ExecutionResult = { + text: 'ok', + messages: [], + toolCalls: [], + usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 }, + finishReason: 'stop', + steps: 1, +}; +const noopAgent: RunnableAgent = { send: vi.fn() }; + +function makeLogger(): Logger & { warn: ReturnType } { + return { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }; +} + +function post(port: number, body: string, headers: Record): Promise<{ status: number; text: string }> { + return new Promise((resolve, reject) => { + const req = http.request({ host: '127.0.0.1', port, path: '/', method: 'POST', headers }, (res) => { + let data = ''; + res.on('data', (chunk) => (data += chunk)); + res.on('end', () => resolve({ status: res.statusCode ?? 0, text: data })); + }); + req.on('error', reject); + req.end(body); + }); +} + +function sign(body: string, secret = SECRET, algorithm: 'sha256' | 'sha1' = 'sha256'): string { + return `${algorithm}=${createHmac(algorithm, secret).update(body).digest('hex')}`; +} + +async function waitForPort(handle: WebhookTriggerHandle): Promise { + for (let i = 0; i < 100 && handle.port === 0; i++) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + return handle.port; +} + +describe('WebhookTriggerAdapter auth', () => { + let handle: WebhookTriggerHandle | undefined; + const onEvent = vi.fn(); + + afterEach(async () => { + onEvent.mockReset(); + await handle?.stop(); + handle = undefined; + }); + + async function start(options: WebhookTriggerAdapterOptions): Promise { + onEvent.mockResolvedValue(result); + handle = new WebhookTriggerAdapter({ host: '127.0.0.1', ...options }).listen(noopAgent, onEvent); + return waitForPort(handle); + } + + describe('hmac', () => { + const body = JSON.stringify({ input: 'hello' }); + + it('accepts a valid signature', async () => { + const port = await start({ auth: { type: 'hmac', secret: SECRET } }); + const res = await post(port, body, { 'x-signature-256': sign(body) }); + expect(res.status).toBe(200); + expect(onEvent).toHaveBeenCalledWith('hello', expect.anything()); + }); + + it('verifies the raw bytes, not re-serialized JSON', async () => { + const raw = '{ "input": "spaced" }'; + const port = await start({ auth: { type: 'hmac', secret: SECRET } }); + const res = await post(port, raw, { 'x-signature-256': sign(raw) }); + expect(res.status).toBe(200); + expect(onEvent).toHaveBeenCalledWith('spaced', expect.anything()); + }); + + it('rejects an invalid signature, a missing header and a wrong-length signature with 401', async () => { + const port = await start({ auth: { type: 'hmac', secret: SECRET } }); + for (const headers of [ + { 'x-signature-256': sign(body, 'wrong-secret') }, + {}, + { 'x-signature-256': 'sha256=abc' }, + { 'x-signature-256': 'sha256=' }, + { 'x-signature-256': sign(body).slice(7) }, + ] as Record[]) { + expect((await post(port, body, headers)).status).toBe(401); + } + expect(onEvent).not.toHaveBeenCalled(); + }); + + it('rejects a tampered body', async () => { + const port = await start({ auth: { type: 'hmac', secret: SECRET } }); + const res = await post(port, JSON.stringify({ input: 'evil' }), { 'x-signature-256': sign(body) }); + expect(res.status).toBe(401); + expect(onEvent).not.toHaveBeenCalled(); + }); + + it('supports a custom header, sha1 and an empty prefix', async () => { + const port = await start({ + auth: { type: 'hmac', secret: SECRET, header: 'X-Hub-Signature', algorithm: 'sha1' }, + }); + expect((await post(port, body, { 'x-hub-signature': sign(body, SECRET, 'sha1') })).status).toBe(200); + await handle?.stop(); + const bare = await start({ auth: { type: 'hmac', secret: SECRET, prefix: '' } }); + const digest = createHmac('sha256', SECRET).update(body).digest('hex'); + expect((await post(bare, body, { 'x-signature-256': digest })).status).toBe(200); + }); + + describe('with a timestamp header', () => { + const auth = { type: 'hmac', secret: SECRET, timestampHeader: 'x-timestamp', toleranceSeconds: 60 } as const; + const signed = (ts: string) => + `sha256=${createHmac('sha256', SECRET).update(`${ts}.${body}`).digest('hex')}`; + const now = () => String(Math.floor(Date.now() / 1000)); + + it('accepts a fresh timestamp signed together with the body', async () => { + const port = await start({ auth }); + const ts = now(); + const res = await post(port, body, { 'x-signature-256': signed(ts), 'x-timestamp': ts }); + expect(res.status).toBe(200); + }); + + it('rejects a stale (replayed) request even though the signature is valid', async () => { + const port = await start({ auth }); + const ts = String(Math.floor(Date.now() / 1000) - 3600); + const res = await post(port, body, { 'x-signature-256': signed(ts), 'x-timestamp': ts }); + expect(res.status).toBe(401); + }); + + it('rejects a missing or non-numeric timestamp, and a timestamp not covered by the signature', async () => { + const port = await start({ auth }); + const ts = now(); + expect((await post(port, body, { 'x-signature-256': signed(ts) })).status).toBe(401); + expect((await post(port, body, { 'x-signature-256': signed('abc'), 'x-timestamp': 'abc' })).status).toBe(401); + expect((await post(port, body, { 'x-signature-256': sign(body), 'x-timestamp': ts })).status).toBe(401); + }); + }); + }); + + describe('bearer', () => { + it('accepts the right token and rejects wrong, missing and malformed ones', async () => { + const port = await start({ auth: { type: 'bearer', token: 's3cret-token' } }); + expect((await post(port, '{}', { authorization: 'Bearer s3cret-token' })).status).toBe(200); + expect((await post(port, '{}', { authorization: 'Bearer nope' })).status).toBe(401); + expect((await post(port, '{}', { authorization: 'Bearer s3cret-tokeX' })).status).toBe(401); + expect((await post(port, '{}', { authorization: 's3cret-token' })).status).toBe(401); + expect((await post(port, '{}', {})).status).toBe(401); + expect(onEvent).toHaveBeenCalledTimes(1); + }); + }); + + describe('custom', () => { + it('uses the verifier (sync or async) with the request and raw body', async () => { + const verify = vi.fn(async (req: http.IncomingMessage, raw: Buffer) => req.headers['x-key'] === 'k' && raw.toString() === 'raw!'); + const port = await start({ auth: { type: 'custom', verify } }); + expect((await post(port, 'raw!', { 'x-key': 'k' })).status).toBe(200); + expect((await post(port, 'raw!', { 'x-key': 'bad' })).status).toBe(401); + }); + + it('treats a throwing verifier as a rejection', async () => { + const port = await start({ + auth: { + type: 'custom', + verify: () => { + throw new Error('db down with secret=hunter2'); + }, + }, + }); + const res = await post(port, '{}', {}); + expect(res.status).toBe(401); + expect(res.text).not.toContain('hunter2'); + }); + }); + + describe('401 response and logging', () => { + it('is generic: no reason, no expected signature', async () => { + const logger = makeLogger(); + const port = await start({ auth: { type: 'hmac', secret: SECRET }, logger }); + const body = '{}'; + const bad = sign(body, 'other'); + const res = await post(port, body, { 'x-signature-256': bad }); + expect(res.status).toBe(401); + expect(JSON.parse(res.text)).toEqual({ error: 'Unauthorized' }); + expect(res.text).not.toContain(sign(body)); + expect(res.text).not.toContain(bad); + }); + + it('logs failures through the logger without secrets or signatures', async () => { + const logger = makeLogger(); + const port = await start({ auth: { type: 'hmac', secret: SECRET }, logger }); + const bad = sign('{}', 'other'); + await post(port, '{}', { 'x-signature-256': bad }); + expect(logger.warn).toHaveBeenCalledTimes(1); + const logged = JSON.stringify(logger.warn.mock.calls); + expect(logged).not.toContain(SECRET); + expect(logged).not.toContain(bad); + expect(logged).not.toContain(sign('{}')); + }); + }); + + describe('configuration', () => { + it('rejects empty secrets/tokens and toleranceSeconds without timestampHeader', () => { + expect(() => new WebhookTriggerAdapter({ auth: { type: 'hmac', secret: '' } })).toThrow(/secret/); + expect(() => new WebhookTriggerAdapter({ auth: { type: 'bearer', token: '' } })).toThrow(/token/); + expect(() => new WebhookTriggerAdapter({ auth: { type: 'hmac', secret: 's', toleranceSeconds: 5 } })).toThrow( + /timestampHeader/ + ); + expect(() => new WebhookTriggerAdapter({ auth: { type: 'nope' } as never })).toThrow(/auth\.type/); + }); + + it('warns once when listening on a non-loopback host without auth', async () => { + const logger = makeLogger(); + const adapter = new WebhookTriggerAdapter({ host: '0.0.0.0', logger }); + const first = adapter.listen(noopAgent, onEvent); + const second = adapter.listen(noopAgent, onEvent); + expect(logger.warn).toHaveBeenCalledTimes(1); + expect(String(logger.warn.mock.calls[0][0])).toContain('auth'); + await Promise.all([waitForPort(first), waitForPort(second)]); + await Promise.all([first.stop(), second.stop()]); + }); + + it('does not warn on loopback hosts or when auth is configured', async () => { + const logger = makeLogger(); + const a = new WebhookTriggerAdapter({ host: '127.0.0.1', logger }).listen(noopAgent, onEvent); + const b = new WebhookTriggerAdapter({ host: '0.0.0.0', logger, auth: { type: 'bearer', token: 't' } }).listen(noopAgent, onEvent); + expect(logger.warn).not.toHaveBeenCalled(); + await Promise.all([waitForPort(a), waitForPort(b)]); + await Promise.all([a.stop(), b.stop()]); + }); + }); +}); diff --git a/src/triggers/adapters/WebhookTriggerAdapter.ts b/src/triggers/adapters/WebhookTriggerAdapter.ts index 642c65000..432ef3ba9 100644 --- a/src/triggers/adapters/WebhookTriggerAdapter.ts +++ b/src/triggers/adapters/WebhookTriggerAdapter.ts @@ -11,10 +11,15 @@ * Reply semantics: the reply target IS the still-open HTTP response - no * separate `reply()` method is implemented (see types.ts's doc comment for * why this differs from Slack/cron). + * + * Authentication (LOU-D13): `options.auth` verifies every request (HMAC over + * the raw body, bearer token, or a custom verifier) before the agent runs. */ import * as http from 'node:http'; import type { ExecutionResult } from '../../execution/AgentExecutor'; +import { Logger, noopLogger } from '../../execution/logger'; import { RunnableAgent, TriggerAdapter, TriggerContext, TriggerHandle } from '../types'; +import { WebhookAuth, assertValidWebhookAuth, checkWebhookAuth } from '../webhookAuth'; export interface WebhookTriggerAdapterOptions { /** Port to listen on. Defaults to 0 (OS-assigned ephemeral port - inspect `handle.port` after `listen()`). */ @@ -23,6 +28,21 @@ export interface WebhookTriggerAdapterOptions { host?: string; /** Only requests to this path are handled; everything else gets a 404. Defaults to '/'. */ path?: string; + /** + * Authenticate every request (HMAC signature over the raw body, bearer + * token, or a custom verifier). Failures get a generic `401`. Strongly + * recommended for any webhook reachable from outside your machine. + * + * @example + * ```ts + * new WebhookTriggerAdapter({ + * auth: { type: 'hmac', secret: process.env.WEBHOOK_SECRET ?? '' }, + * }); + * ``` + */ + auth?: WebhookAuth; + /** Receives auth failures and the "no auth configured" warning. Defaults to a no-op logger. */ + logger?: Logger; } export interface WebhookTriggerHandle extends TriggerHandle { @@ -30,13 +50,13 @@ export interface WebhookTriggerHandle extends TriggerHandle { readonly port: number; } -function readBody(req: http.IncomingMessage): Promise { +function readBody(req: http.IncomingMessage): Promise { return new Promise((resolve, reject) => { - let data = ''; - req.on('data', (chunk) => { - data += chunk; + const chunks: Buffer[] = []; + req.on('data', (chunk: Buffer | string) => { + chunks.push(typeof chunk === 'string' ? Buffer.from(chunk) : chunk); }); - req.on('end', () => resolve(data)); + req.on('end', () => resolve(Buffer.concat(chunks))); req.on('error', reject); }); } @@ -58,14 +78,41 @@ function parseWebhookInput(raw: string): string { type WebhookOnEvent = (input: string, context: TriggerContext) => Promise; -async function respondWithAgentResult( +interface WebhookRuntime { + path: string; + onEvent: WebhookOnEvent; + auth?: WebhookAuth; + logger: Logger; +} + +/** Authenticates the request (when configured); on failure writes a generic 401 and returns false. */ +async function authorize( + runtime: WebhookRuntime, req: http.IncomingMessage, res: http.ServerResponse, - onEvent: WebhookOnEvent + rawBody: Buffer +): Promise { + if (!runtime.auth) return true; + const failure = await checkWebhookAuth(runtime.auth, req, rawBody); + if (failure === undefined) return true; + runtime.logger.warn('webhook request rejected: authentication failed', { + reason: failure, + remoteAddress: req.socket.remoteAddress, + }); + writeJson(res, 401, { error: 'Unauthorized' }); + return false; +} + +async function respondWithAgentResult( + runtime: WebhookRuntime, + req: http.IncomingMessage, + res: http.ServerResponse ): Promise { try { - const input = parseWebhookInput(await readBody(req)); - const result = await onEvent(input, { channel: res, request: req }); + const rawBody = await readBody(req); + if (!(await authorize(runtime, req, res, rawBody))) return; + const input = parseWebhookInput(rawBody.toString('utf8')); + const result = await runtime.onEvent(input, { channel: res, request: req }); writeJson(res, 200, result); } catch (error) { writeJson(res, 500, { error: (error as Error).message }); @@ -73,34 +120,53 @@ async function respondWithAgentResult( } async function handleWebhookRequest( + runtime: WebhookRuntime, req: http.IncomingMessage, - res: http.ServerResponse, - path: string, - onEvent: WebhookOnEvent + res: http.ServerResponse ): Promise { - if (req.method !== 'POST' || req.url !== path) { + if (req.method !== 'POST' || req.url !== runtime.path) { writeJson(res, 404, { error: 'Not found' }); return; } - await respondWithAgentResult(req, res, onEvent); + await respondWithAgentResult(runtime, req, res); +} + +function isLoopbackHost(host: string): boolean { + return host === 'localhost' || host === '::1' || host === '[::1]' || /^127\./.test(host); } export class WebhookTriggerAdapter implements TriggerAdapter { public readonly type = 'webhook'; - constructor(private readonly options: WebhookTriggerAdapterOptions = {}) {} + private warnedUnauthenticated = false; + + constructor(private readonly options: WebhookTriggerAdapterOptions = {}) { + if (options.auth) assertValidWebhookAuth(options.auth); + } + + private warnIfUnauthenticated(host: string, logger: Logger): void { + if (this.options.auth || this.warnedUnauthenticated || isLoopbackHost(host)) return; + this.warnedUnauthenticated = true; + logger.warn( + `WebhookTriggerAdapter is listening on ${host} without \`auth\`: anyone who can reach it can run your agent. ` + + 'Configure options.auth (hmac, bearer or custom); see docs/api-overview.md#triggers.' + ); + } public listen( agent: RunnableAgent, onEvent: WebhookOnEvent ): WebhookTriggerHandle { - const path = this.options.path ?? '/'; + const host = this.options.host ?? '0.0.0.0'; + const logger = this.options.logger ?? noopLogger; + const runtime: WebhookRuntime = { path: this.options.path ?? '/', onEvent, auth: this.options.auth, logger }; + this.warnIfUnauthenticated(host, logger); const server = http.createServer((req, res) => { - void handleWebhookRequest(req, res, path, onEvent); + void handleWebhookRequest(runtime, req, res); }); - server.listen(this.options.port ?? 0, this.options.host ?? '0.0.0.0'); + server.listen(this.options.port ?? 0, host); void agent; // available for adapters/callers that want a default onEvent; unused by this adapter's own logic. diff --git a/src/triggers/cronExpression.test.ts b/src/triggers/cronExpression.test.ts new file mode 100644 index 000000000..5022b4f2e --- /dev/null +++ b/src/triggers/cronExpression.test.ts @@ -0,0 +1,118 @@ +import { describe, it, expect } from 'vitest'; +import { CronExpressionError, parseCronExpression } from './cronExpression'; + +const next = (expr: string, after: string, tz = 'UTC') => + parseCronExpression(expr, tz).nextRun(new Date(after)).toISOString(); + +describe('parseCronExpression().nextRun()', () => { + const table: Array<[string, string, string, string?]> = [ + // expression, after, expected, timezone + ['* * * * *', '2026-01-01T00:00:30Z', '2026-01-01T00:01:00.000Z'], + ['* * * * *', '2026-01-01T00:00:00Z', '2026-01-01T00:01:00.000Z'], + ['*/15 * * * *', '2026-01-01T10:07:00Z', '2026-01-01T10:15:00.000Z'], + ['*/15 * * * *', '2026-01-01T10:45:00Z', '2026-01-01T11:00:00.000Z'], + ['10-40/10 * * * *', '2026-01-01T12:40:00Z', '2026-01-01T13:10:00.000Z'], + ['10-40/10 * * * *', '2026-01-01T12:05:00Z', '2026-01-01T12:10:00.000Z'], + ['5/20 * * * *', '2026-01-01T12:06:00Z', '2026-01-01T12:25:00.000Z'], + ['0 9 * * MON-FRI', '2026-01-02T09:00:00Z', '2026-01-05T09:00:00.000Z'], + ['0 9 * * mon-fri', '2026-01-02T08:59:00Z', '2026-01-02T09:00:00.000Z'], + ['30 4 1,15 * *', '2026-01-15T04:30:00Z', '2026-02-01T04:30:00.000Z'], + ['0 0 1 JAN *', '2026-01-01T00:00:00Z', '2027-01-01T00:00:00.000Z'], + ['0 0 1 MAR-MAY *', '2026-05-01T00:00:00Z', '2027-03-01T00:00:00.000Z'], + ['0 0 31 * *', '2026-01-31T00:00:00Z', '2026-03-31T00:00:00.000Z'], + ['59 23 31 12 *', '2026-12-31T23:59:00Z', '2027-12-31T23:59:00.000Z'], + // leap day + ['0 0 29 2 *', '2026-01-01T00:00:00Z', '2028-02-29T00:00:00.000Z'], + ['0 0 29 2 *', '2028-02-29T00:00:00Z', '2032-02-29T00:00:00.000Z'], + // Sunday as 0, 7 and SUN, and inside a range + ['0 12 * * 0', '2026-01-01T00:00:00Z', '2026-01-04T12:00:00.000Z'], + ['0 12 * * 7', '2026-01-01T00:00:00Z', '2026-01-04T12:00:00.000Z'], + ['0 12 * * SUN', '2026-01-01T00:00:00Z', '2026-01-04T12:00:00.000Z'], + ['0 12 * * 5-7', '2026-01-03T12:00:00Z', '2026-01-04T12:00:00.000Z'], + // day-of-month OR day-of-week when both are restricted + ['0 0 13 * FRI', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00.000Z'], + ['0 0 13 * FRI', '2026-01-10T00:00:00Z', '2026-01-13T00:00:00.000Z'], + ['0 0 13 * FRI', '2026-01-13T00:00:00Z', '2026-01-16T00:00:00.000Z'], + // ...but AND when one of them is a wildcard + ['0 0 13 * *', '2026-01-01T00:00:00Z', '2026-01-13T00:00:00.000Z'], + ['0 0 * * MON', '2026-01-01T00:00:00Z', '2026-01-05T00:00:00.000Z'], + // shortcuts + ['@hourly', '2026-01-01T10:30:00Z', '2026-01-01T11:00:00.000Z'], + ['@daily', '2026-01-01T10:30:00Z', '2026-01-02T00:00:00.000Z'], + ['@weekly', '2026-01-01T10:30:00Z', '2026-01-04T00:00:00.000Z'], + ['@monthly', '2026-01-01T10:30:00Z', '2026-02-01T00:00:00.000Z'], + [' 0 0 * * * ', '2026-01-01T10:30:00Z', '2026-01-02T00:00:00.000Z'], + // fixed-offset time zone + ['0 9 * * *', '2026-01-01T00:00:00Z', '2026-01-01T03:30:00.000Z', 'Asia/Kolkata'], + // New York spring forward (2026-03-08, 02:00 EST -> 03:00 EDT) + ['30 1 * * *', '2026-03-07T06:30:00Z', '2026-03-08T06:30:00.000Z', 'America/New_York'], + ['30 3 * * *', '2026-03-08T06:30:00Z', '2026-03-08T07:30:00.000Z', 'America/New_York'], + ['0 * * * *', '2026-03-08T06:30:00Z', '2026-03-08T07:00:00.000Z', 'America/New_York'], + // a time inside the skipped hour does not exist: skipped that day + ['30 2 * * *', '2026-03-07T07:30:00Z', '2026-03-09T06:30:00.000Z', 'America/New_York'], + // New York fall back (2026-11-01, 02:00 EDT -> 01:00 EST): fixed times fire once... + ['30 1 * * *', '2026-11-01T04:00:00Z', '2026-11-01T05:30:00.000Z', 'America/New_York'], + ['30 1 * * *', '2026-11-01T05:30:00Z', '2026-11-02T06:30:00.000Z', 'America/New_York'], + // ...while an every-hour schedule keeps its real-time cadence + ['0 * * * *', '2026-11-01T05:30:00Z', '2026-11-01T06:00:00.000Z', 'America/New_York'], + ['0 * * * *', '2026-11-01T06:00:00Z', '2026-11-01T07:00:00.000Z', 'America/New_York'], + // the zone decides the weekday/day, not UTC + ['0 0 * * MON', '2026-01-01T00:00:00Z', '2026-01-05T05:00:00.000Z', 'America/New_York'], + ]; + + it.each(table)('%s after %s -> %s (%s)', (expr, after, expected, tz) => { + expect(next(expr, after, tz)).toBe(expected); + }); + + it('is strictly after `after` and successive calls walk forward without repeats', () => { + const schedule = parseCronExpression('*/20 * * * *', 'UTC'); + let t = new Date('2026-06-01T00:00:00Z'); + const seen: string[] = []; + for (let i = 0; i < 4; i++) { + t = schedule.nextRun(t); + seen.push(t.toISOString()); + } + expect(seen).toEqual([ + '2026-06-01T00:20:00.000Z', + '2026-06-01T00:40:00.000Z', + '2026-06-01T01:00:00.000Z', + '2026-06-01T01:20:00.000Z', + ]); + }); + + it('works in the machine local zone when no timezone is given', () => { + const run = parseCronExpression('* * * * *').nextRun(new Date('2026-01-01T00:00:10Z')); + expect(run.toISOString()).toBe('2026-01-01T00:01:00.000Z'); + }); +}); + +describe('parseCronExpression() errors', () => { + const invalid: Array<[string, RegExp]> = [ + ['', /expected 5 space-separated fields/], + ['* * * *', /expected 5 space-separated fields/], + ['* * * * * *', /expected 5 space-separated fields/], + ['@fortnightly', /expected 5 space-separated fields/], + ['60 * * * *', /minute field "60".*out of range.*0-59.*\*\/15/], + ['* 24 * * *', /hour field "24".*out of range.*0-23/], + ['* * 0 * *', /day-of-month field "0".*out of range.*1-31/], + ['* * * 13 *', /month field "13".*out of range.*1-12/], + ['* * * * 8', /day-of-week field "8".*out of range.*0-7.*MON-FRI/], + ['*/0 * * * *', /minute field "\*\/0".*step "0"/], + ['*/x * * * *', /step "x"/], + ['5-1 * * * *', /minute field "5-1".*ends before it starts/], + ['a * * * *', /minute field "a".*not a number/], + ['* * * FOO *', /month field "FOO".*not a number or name/], + ['1,,2 * * * *', /minute field.*empty or malformed list item/], + ['1-2-3 * * * *', /minute field.*malformed range/], + ['0 0 31 2 *', /never matches/], + ]; + + it.each(invalid)('rejects %j', (expr, message) => { + expect(() => parseCronExpression(expr)).toThrow(CronExpressionError); + expect(() => parseCronExpression(expr)).toThrow(message); + }); + + it('rejects an unknown time zone with an example', () => { + expect(() => parseCronExpression('* * * * *', 'Mars/Olympus')).toThrow(/Invalid timezone "Mars\/Olympus".*America\/New_York/); + }); +}); diff --git a/src/triggers/cronExpression.ts b/src/triggers/cronExpression.ts new file mode 100644 index 000000000..c4583574b --- /dev/null +++ b/src/triggers/cronExpression.ts @@ -0,0 +1,285 @@ +/** + * Dependency-free cron expression parser and scheduler math (LOU-D13). + * + * Supports the standard 5-field format `minute hour day-of-month month + * day-of-week` with `*`, lists (`1,15`), ranges (`1-5`), steps (`*\/15`, + * `10-40/10`), month names (`JAN`) and weekday names (`MON`), with both `0` + * and `7` meaning Sunday. When both day-of-month and day-of-week are + * restricted a day matches if EITHER does (classic Vixie cron semantics). + * The shortcuts `@hourly`, `@daily`, `@weekly`, `@monthly` (and `@yearly`) + * are accepted too. + * + * Time zones use `Intl.DateTimeFormat` (any IANA name). Around DST changes: + * a wall-clock time that does not exist (spring forward) is skipped that + * day; a wall-clock time that happens twice (fall back) fires once if the + * hour field is restricted, while an every-hour schedule keeps firing + * hourly in real time. + */ + +/** Thrown for an invalid cron expression or time zone; the message names the field and shows a valid example. */ +export class CronExpressionError extends Error { + constructor(message: string) { + super(message); + this.name = 'CronExpressionError'; + } +} + +/** A parsed cron expression. */ +export interface CronSchedule { + /** The first fire time strictly after `after`. */ + nextRun(after: Date): Date; +} + +interface FieldSpec { + name: string; + min: number; + max: number; + example: string; + names?: readonly string[]; + /** Added to a name's index to get its numeric value. */ + nameOffset?: number; +} + +const MONTHS = ['JAN', 'FEB', 'MAR', 'APR', 'MAY', 'JUN', 'JUL', 'AUG', 'SEP', 'OCT', 'NOV', 'DEC']; +const WEEKDAYS = ['SUN', 'MON', 'TUE', 'WED', 'THU', 'FRI', 'SAT']; + +const FIELDS: readonly FieldSpec[] = [ + { name: 'minute', min: 0, max: 59, example: '*/15' }, + { name: 'hour', min: 0, max: 23, example: '9-17' }, + { name: 'day-of-month', min: 1, max: 31, example: '1,15' }, + { name: 'month', min: 1, max: 12, example: 'JAN-JUN', names: MONTHS, nameOffset: 1 }, + { name: 'day-of-week', min: 0, max: 7, example: 'MON-FRI', names: WEEKDAYS, nameOffset: 0 }, +]; + +const SHORTCUTS: Record = { + '@hourly': '0 * * * *', + '@daily': '0 0 * * *', + '@midnight': '0 0 * * *', + '@weekly': '0 0 * * 0', + '@monthly': '0 0 1 * *', + '@yearly': '0 0 1 1 *', + '@annually': '0 0 1 1 *', +}; + +const EXAMPLE_EXPRESSION = '"*/15 9-17 * * MON-FRI" or "@daily"'; + +function fieldError(spec: FieldSpec, text: string, problem: string): CronExpressionError { + return new CronExpressionError( + `Invalid cron expression: ${spec.name} field "${text}" ${problem}. ` + + `Allowed values are ${spec.min}-${spec.max}${spec.names ? ` or names like ${spec.names[0]}` : ''}; e.g. "${spec.example}".` + ); +} + +function parseValue(spec: FieldSpec, token: string, whole: string): number { + const nameIndex = spec.names?.indexOf(token.toUpperCase()) ?? -1; + const value = nameIndex >= 0 ? nameIndex + (spec.nameOffset ?? 0) : /^\d+$/.test(token) ? Number(token) : NaN; + if (Number.isNaN(value)) throw fieldError(spec, whole, `has "${token}", which is not a number${spec.names ? ' or name' : ''}`); + if (value < spec.min || value > spec.max) throw fieldError(spec, whole, `has ${value}, which is out of range`); + return value; +} + +function parseStep(spec: FieldSpec, stepText: string | undefined, whole: string): number { + if (stepText === undefined) return 1; + if (!/^\d+$/.test(stepText) || Number(stepText) < 1) { + throw fieldError(spec, whole, `has step "${stepText}", which must be a positive integer`); + } + return Number(stepText); +} + +function parseRange(spec: FieldSpec, rangeText: string, hasStep: boolean, whole: string): [number, number] { + if (rangeText === '*') return [spec.min, spec.max]; + const [startText, endText, ...extra] = rangeText.split('-'); + if (extra.length > 0 || startText === '') throw fieldError(spec, whole, `has a malformed range "${rangeText}"`); + const start = parseValue(spec, startText, whole); + if (endText !== undefined) { + const end = parseValue(spec, endText, whole); + if (end < start) throw fieldError(spec, whole, `has range "${rangeText}" that ends before it starts`); + return [start, end]; + } + return [start, hasStep ? spec.max : start]; +} + +function parseField(spec: FieldSpec, text: string): Set { + const values = new Set(); + for (const part of text.split(',')) { + const [rangeText, stepText, ...extra] = part.split('/'); + if (part === '' || extra.length > 0) throw fieldError(spec, text, 'has an empty or malformed list item'); + const step = parseStep(spec, stepText, text); + const [start, end] = parseRange(spec, rangeText, stepText !== undefined, text); + for (let v = start; v <= end; v += step) values.add(spec.name === 'day-of-week' && v === 7 ? 0 : v); + } + return values; +} + +interface ZonedParts { + year: number; + month: number; + day: number; + hour: number; + minute: number; + weekday: number; +} + +const WEEKDAY_INDEX: Record = { Sun: 0, Mon: 1, Tue: 2, Wed: 3, Thu: 4, Fri: 5, Sat: 6 }; + +function makeFormatter(timeZone: string | undefined): Intl.DateTimeFormat { + try { + return new Intl.DateTimeFormat('en-US', { + timeZone, + hourCycle: 'h23', + year: 'numeric', + month: 'numeric', + day: 'numeric', + hour: 'numeric', + minute: 'numeric', + weekday: 'short', + }); + } catch { + throw new CronExpressionError( + `Invalid timezone "${timeZone}". Use an IANA name such as "America/New_York" or "UTC".` + ); + } +} + +function zonedParts(formatter: Intl.DateTimeFormat, date: Date): ZonedParts { + const p: Record = {}; + for (const part of formatter.formatToParts(date)) p[part.type] = part.value; + return { + year: Number(p.year), + month: Number(p.month), + day: Number(p.day), + hour: Number(p.hour) % 24, + minute: Number(p.minute), + weekday: WEEKDAY_INDEX[p.weekday], + }; +} + +/** The wall-clock fields read as if they were UTC, for offset arithmetic. */ +function asUtcMs(p: ZonedParts): number { + return Date.UTC(p.year, p.month - 1, p.day, p.hour, p.minute); +} + +const MINUTE_MS = 60_000; +const HOUR_MS = 3_600_000; +const MAX_SEARCH_YEARS = 10; + +class CronExpressionSchedule implements CronSchedule { + private readonly formatter: Intl.DateTimeFormat; + + constructor( + private readonly minutes: Set, + private readonly hours: Set, + private readonly daysOfMonth: Set, + private readonly months: Set, + private readonly daysOfWeek: Set, + private readonly flags: { domRestricted: boolean; dowRestricted: boolean; hourRestricted: boolean }, + timeZone: string | undefined + ) { + this.formatter = makeFormatter(timeZone); + } + + private dayMatches(p: ZonedParts): boolean { + const domMatch = this.daysOfMonth.has(p.day); + const dowMatch = this.daysOfWeek.has(p.weekday); + if (this.flags.domRestricted && this.flags.dowRestricted) return domMatch || dowMatch; + return domMatch && dowMatch; + } + + /** The earliest instant whose wall clock reads the given local midnight (or just after it if midnight does not exist). */ + private startOfDay(rawYear: number, rawMonth: number, rawDay: number): number { + const guess = Date.UTC(rawYear, rawMonth - 1, rawDay); + const target = new Date(guess); + const [year, month, day] = [target.getUTCFullYear(), target.getUTCMonth() + 1, target.getUTCDate()]; + for (const offset of [-HOUR_MS * 14, HOUR_MS * 14]) { + const probe = new Date(guess + offset); + const candidate = guess - (asUtcMs(zonedParts(this.formatter, probe)) - probe.getTime()); + const p = zonedParts(this.formatter, new Date(candidate)); + if (p.year === year && p.month === month && p.day === day && p.hour === 0 && p.minute === 0) { + return candidate; + } + } + // Midnight is skipped by a DST change: the first instant of that local day. + let t = guess - HOUR_MS * 14; + while (zonedParts(this.formatter, new Date(t)).day !== day) t += 15 * MINUTE_MS; + return t; + } + + /** True for the second occurrence of a wall-clock time that happens twice (DST fall back). */ + private isRepeatedHour(t: number, p: ZonedParts): boolean { + const earlier = zonedParts(this.formatter, new Date(t - HOUR_MS)); + return earlier.hour === p.hour && earlier.day === p.day && earlier.minute === p.minute; + } + + private nextInSet(set: Set, from: number): number | undefined { + let best: number | undefined; + for (const v of set) if (v >= from && (best === undefined || v < best)) best = v; + return best; + } + + /** Advance past a non-matching time unit; returns the next instant worth examining. */ + private step(t: number, p: ZonedParts): number { + if (!this.months.has(p.month)) return this.startOfDay(p.year, p.month + 1, 1); + if (!this.dayMatches(p)) return this.startOfDay(p.year, p.month, p.day + 1); + if (!this.hours.has(p.hour)) return t + (60 - p.minute) * MINUTE_MS; + const next = this.nextInSet(this.minutes, p.minute + 1); + return t + ((next ?? 60) - p.minute) * MINUTE_MS; + } + + private matches(t: number, p: ZonedParts): boolean { + if (!this.months.has(p.month) || !this.dayMatches(p) || !this.hours.has(p.hour) || !this.minutes.has(p.minute)) { + return false; + } + return !(this.flags.hourRestricted && this.isRepeatedHour(t, p)); + } + + public nextRun(after: Date): Date { + const limit = after.getTime() + MAX_SEARCH_YEARS * 366 * 24 * HOUR_MS; + let t = Math.floor(after.getTime() / MINUTE_MS) * MINUTE_MS + MINUTE_MS; + while (t <= limit) { + const p = zonedParts(this.formatter, new Date(t)); + if (this.matches(t, p)) return new Date(t); + t = this.step(t, p); + } + throw new CronExpressionError('Cron expression never matches a real date (for example "0 0 31 2 *"). Check the day-of-month and month fields.'); + } +} + +/** + * Parses a cron expression into a schedule. + * + * @param expression A 5-field cron expression or one of `@hourly`, `@daily`, `@weekly`, `@monthly`, `@yearly`. + * @param timeZone IANA time zone name (e.g. `'America/New_York'`). Defaults to the machine's local zone. + * @throws {CronExpressionError} With the offending field and an example of a valid value. + * + * @example + * ```ts + * const schedule = parseCronExpression('*\/15 9-17 * * MON-FRI', 'Europe/Paris'); + * const next = schedule.nextRun(new Date()); + * ``` + */ +export function parseCronExpression(expression: string, timeZone?: string): CronSchedule { + const trimmed = expression.trim(); + const fields = (SHORTCUTS[trimmed.toLowerCase()] ?? trimmed).split(/\s+/); + if (trimmed === '' || fields.length !== FIELDS.length) { + throw new CronExpressionError( + `Invalid cron expression "${expression}": expected 5 space-separated fields (minute hour day-of-month month day-of-week) ` + + `or a shortcut (@hourly, @daily, @weekly, @monthly), e.g. ${EXAMPLE_EXPRESSION}.` + ); + } + const [minutes, hours, daysOfMonth, months, daysOfWeek] = FIELDS.map((spec, i) => parseField(spec, fields[i])); + const schedule = new CronExpressionSchedule( + minutes, + hours, + daysOfMonth, + months, + daysOfWeek, + { + domRestricted: !fields[2].startsWith('*'), + dowRestricted: !fields[4].startsWith('*'), + hourRestricted: !fields[1].startsWith('*'), + }, + timeZone + ); + schedule.nextRun(new Date(0)); // fail fast on an impossible schedule such as "0 0 31 2 *" + return schedule; +} diff --git a/src/triggers/index.ts b/src/triggers/index.ts index 3310f9826..d43a9faeb 100644 --- a/src/triggers/index.ts +++ b/src/triggers/index.ts @@ -9,5 +9,7 @@ export * from './types'; export * from './TriggerRegistry'; export * from './adapters/WebhookTriggerAdapter'; +export * from './webhookAuth'; +export * from './cronExpression'; export * from './adapters/CronTriggerAdapter'; export * from './adapters/SlackTriggerAdapter'; diff --git a/src/triggers/webhookAuth.ts b/src/triggers/webhookAuth.ts new file mode 100644 index 000000000..78dabdac5 --- /dev/null +++ b/src/triggers/webhookAuth.ts @@ -0,0 +1,166 @@ +/** + * Webhook request authentication (LOU-D13). + * + * Pure verification helpers used by `WebhookTriggerAdapter`. They never + * throw for a bad request - they return the (internal) reason a request was + * rejected so the adapter can log it, while the HTTP response stays generic. + */ +import { createHash, createHmac, timingSafeEqual } from 'node:crypto'; +import type { IncomingMessage } from 'node:http'; + +/** + * Verify a signature computed with a shared secret over the RAW request + * body bytes (the format GitHub, Shopify and many others use). + * + * @example + * ```ts + * new WebhookTriggerAdapter({ auth: { type: 'hmac', secret: process.env.WEBHOOK_SECRET! } }); + * ``` + */ +export interface HmacWebhookAuth { + type: 'hmac'; + /** Shared secret used to compute the HMAC. */ + secret: string; + /** Request header carrying the signature. Defaults to `'x-signature-256'`. */ + header?: string; + /** HMAC hash algorithm. Defaults to `'sha256'`. */ + algorithm?: 'sha256' | 'sha1'; + /** Text the header value starts with before the hex digest. Defaults to `` `${algorithm}=` `` (`'sha256='`). Use `''` for a bare digest. */ + prefix?: string; + /** + * Header carrying a Unix timestamp (seconds, or milliseconds). When set, + * the signed payload becomes `${timestamp}.${rawBody}` and requests whose + * timestamp is further than `toleranceSeconds` from now are rejected, + * which prevents replaying a captured request. + */ + timestampHeader?: string; + /** Maximum accepted clock difference in seconds. Defaults to 300. Requires `timestampHeader`. */ + toleranceSeconds?: number; +} + +/** + * Require an `Authorization: Bearer ` header. + * + * @example + * ```ts + * new WebhookTriggerAdapter({ auth: { type: 'bearer', token: process.env.WEBHOOK_TOKEN! } }); + * ``` + */ +export interface BearerWebhookAuth { + type: 'bearer'; + /** The expected token. */ + token: string; +} + +/** + * Bring your own check. `rawBody` is the exact bytes received (decode with + * `rawBody.toString('utf8')`). Return `false` (or throw) to reject with 401. + * + * @example + * ```ts + * new WebhookTriggerAdapter({ + * auth: { type: 'custom', verify: (req) => req.headers['x-api-key'] === process.env.API_KEY }, + * }); + * ``` + */ +export interface CustomWebhookAuth { + type: 'custom'; + verify(req: IncomingMessage, rawBody: Buffer): boolean | Promise; +} + +/** How `WebhookTriggerAdapter` authenticates inbound requests. */ +export type WebhookAuth = HmacWebhookAuth | BearerWebhookAuth | CustomWebhookAuth; + +const DEFAULT_TOLERANCE_SECONDS = 300; + +/** Fails fast, at construction time, on a configuration that could never authenticate (or never reject) anything. */ +export function assertValidWebhookAuth(auth: WebhookAuth): void { + if (auth.type === 'hmac') { + if (!auth.secret) { + throw new Error("WebhookTriggerAdapter: auth.secret must be a non-empty string (e.g. { type: 'hmac', secret: process.env.WEBHOOK_SECRET })."); + } + if (auth.toleranceSeconds !== undefined && !auth.timestampHeader) { + throw new Error('WebhookTriggerAdapter: auth.toleranceSeconds has no effect without auth.timestampHeader. Set timestampHeader (e.g. "x-timestamp") or remove toleranceSeconds.'); + } + } else if (auth.type === 'bearer') { + if (!auth.token) { + throw new Error("WebhookTriggerAdapter: auth.token must be a non-empty string (e.g. { type: 'bearer', token: process.env.WEBHOOK_TOKEN })."); + } + } else if (auth.type !== 'custom' || typeof auth.verify !== 'function') { + throw new Error("WebhookTriggerAdapter: auth.type must be 'hmac', 'bearer' or 'custom' (custom needs a verify(req, rawBody) function)."); + } +} + +/** + * Constant-time string comparison. Both sides are hashed first, so the + * buffers always have equal length (`timingSafeEqual` throws otherwise) and + * the length of the expected value is not leaked either. + */ +function safeEqual(a: string, b: string): boolean { + const digest = (value: string) => createHash('sha256').update(value).digest(); + return timingSafeEqual(digest(a), digest(b)); +} + +function headerValue(req: IncomingMessage, name: string): string | undefined { + const value = req.headers[name.toLowerCase()]; + return typeof value === 'string' ? value : undefined; +} + +function isFresh(timestamp: string, toleranceSeconds: number, nowMs: number): boolean { + const parsed = Number(timestamp); + if (!Number.isFinite(parsed)) return false; + const seconds = parsed > 1e12 ? parsed / 1000 : parsed; + return Math.abs(nowMs / 1000 - seconds) <= toleranceSeconds; +} + +function checkTimestamp(auth: HmacWebhookAuth, req: IncomingMessage, nowMs: number): string | undefined { + if (!auth.timestampHeader) return undefined; + const timestamp = headerValue(req, auth.timestampHeader); + if (timestamp === undefined) return 'missing timestamp'; + return isFresh(timestamp, auth.toleranceSeconds ?? DEFAULT_TOLERANCE_SECONDS, nowMs) ? undefined : 'stale timestamp'; +} + +function checkHmac(auth: HmacWebhookAuth, req: IncomingMessage, rawBody: Buffer, nowMs: number): string | undefined { + const algorithm = auth.algorithm ?? 'sha256'; + const prefix = auth.prefix ?? `${algorithm}=`; + const header = headerValue(req, auth.header ?? 'x-signature-256'); + if (header === undefined || !header.startsWith(prefix)) return 'missing or malformed signature'; + + const timestampFailure = checkTimestamp(auth, req, nowMs); + if (timestampFailure) return timestampFailure; + + const hmac = createHmac(algorithm, auth.secret); + if (auth.timestampHeader) hmac.update(`${headerValue(req, auth.timestampHeader)}.`); + const expected = hmac.update(rawBody).digest('hex'); + return safeEqual(header.slice(prefix.length).toLowerCase(), expected) ? undefined : 'signature mismatch'; +} + +function checkBearer(auth: BearerWebhookAuth, req: IncomingMessage): string | undefined { + const match = /^Bearer (.+)$/i.exec(headerValue(req, 'authorization') ?? ''); + if (!match) return 'missing bearer token'; + return safeEqual(match[1], auth.token) ? undefined : 'bearer token mismatch'; +} + +async function checkCustom(auth: CustomWebhookAuth, req: IncomingMessage, rawBody: Buffer): Promise { + try { + return (await auth.verify(req, rawBody)) === true ? undefined : 'custom verifier rejected the request'; + } catch { + return 'custom verifier threw'; + } +} + +/** + * Checks a request against `auth`. Resolves to `undefined` when the request + * is authentic, or to a short internal reason (safe to log - it contains no + * secret, signature or token) when it is not. + */ +export async function checkWebhookAuth( + auth: WebhookAuth, + req: IncomingMessage, + rawBody: Buffer, + nowMs: number = Date.now() +): Promise { + if (auth.type === 'hmac') return checkHmac(auth, req, rawBody, nowMs); + if (auth.type === 'bearer') return checkBearer(auth, req); + return checkCustom(auth, req, rawBody); +}