feat(activitypub): double-knock delivery signatures and per-host circuit breaker

.changeset/activitypub-delivery-circuit-breaker.md

@@ -0,0 +1,5 @@
+---
+"@dwk/activitypub": minor
+---
+
+Add a per-host delivery circuit breaker. After 5 consecutive retryable failures to a host, its deliveries are deferred, without spending retry attempts, for a cooldown that starts at 10 minutes and doubles on each failed probe up to 24 hours. A success or a signed inbound activity from the host closes the circuit, and a host failing for 7 days has its deliveries dropped while the circuit is open. New log/metric events: `activitypub.delivery.deferred`, `activitypub.delivery.circuit_opened`, `activitypub.delivery.circuit_closed`.

.changeset/activitypub-double-knock.md

@@ -0,0 +1,5 @@
+---
+"@dwk/activitypub": minor
+---
+
+Sign outbound deliveries with RFC 9421 HTTP Message Signatures first, falling back to draft-cavage when a peer rejects them ("double-knock"). Peers that only accept draft-cavage are remembered per actor for 7 days. Adds `signRequestRfc9421`; `deliverActivity`'s trailing `now`/`timeoutMs` arguments move into a `DeliveryOptions` object (with the new `preference`), and `DeliveryResult` now reports the accepted `profile` on success.

packages/activitypub/README.md

@@ -94,19 +94,35 @@ No KV is used for any authoritative state, per
 [`spec/non-functional-requirements.md`](../../spec/non-functional-requirements.md):
 follower lists, dedup, and the delivery queue all live in the DO's SQLite.
 
+## Unreachable peers
+
+Each actor's Durable Object keeps a per-host delivery circuit breaker. After 5
+consecutive retryable failures to a host, its queued deliveries are deferred
+(not attempted, and not charged a retry) for a cooldown that starts at 10
+minutes and doubles on each failed probe, up to 24 hours. A successful
+delivery, or any signed activity received from that host, closes the circuit.
+A host that has been failing for 7 days has its deliveries dropped while the
+circuit is open, the same window Mastodon uses before it marks a domain
+unavailable. This keeps one dead peer from spending every alarm wake on
+10-second timeouts.
+
 ## HTTP signatures
 
-Inbound `POST /inbox` deliveries are authenticated, and outbound deliveries are
-signed, with the de-facto fediverse `draft-cavage-http-signatures` profile
-(RSA-SHA256 over a covered header set, body integrity via `Digest`). The
+Inbound `POST /inbox` deliveries are verified against either the de-facto
+fediverse `draft-cavage-http-signatures` profile (RSA-SHA256 over a covered
+header set, body integrity via `Digest`) or
+[RFC 9421](https://www.rfc-editor.org/rfc/rfc9421) HTTP Message Signatures
+(via [`@dwk/http-signatures`](../http-signatures/README.md)). The
 implementation is RSA-only with an explicit algorithm allow-list (no `none`, no
 symmetric algorithms), mirroring the [`@dwk/dpop`](../dpop/README.md) hardening
 posture.
 
-Signing/verification sit behind the `verifyInboxSignature` config seam, so the
-forthcoming cross-standard `@dwk/http-signatures` package (RFC 9421 +
-draft-cavage; [#59](https://github.com/davidwkeith/workers/issues/59)) can be
-swapped in unchanged once it lands.
+Outbound deliveries **double-knock**: each is signed with RFC 9421 first, and
+if the peer answers `400`/`401`/`403` the same body is re-signed with
+draft-cavage and sent once more. A host that only accepts draft-cavage is
+remembered in the actor's Durable Object for 7 days, so later deliveries skip
+the first knock (the same strategy [Fedify](https://fedify.dev/manual/send#double-knocking-http-signatures)
+uses). Outbound `GET`s (authorized-fetch key resolution) stay draft-cavage.
 
 Every outbound request (key resolution, delivery, discovery) sends a
 descriptive `User-Agent` (`userAgent` in `ActivityPubConfig`, defaulting to

packages/activitypub/src/delivery.test.ts

@@ -99,13 +99,20 @@ describe("deliverActivity", () => {
       TEST_UA,
     );
 
-    expect(result).toEqual<DeliveryResult>({ ok: true, status: 202 });
+    expect(result).toEqual<DeliveryResult>({
+      ok: true,
+      status: 202,
+      profile: "rfc9421",
+    });
     expect(fetchImpl).toHaveBeenCalledOnce();
     const init = (fetchImpl as unknown as ReturnType<typeof vi.fn>).mock
       .calls[0]![1] as RequestInit;
     const headers = init.headers as Record<string, string>;
-    expect(headers.Signature).toContain('keyId="');
-    expect(headers.Digest).toMatch(/^SHA-256=/);
+    expect(headers["Signature-Input"]).toContain(
+      'keyid="https://example.com/users/bob#main-key"',
+    );
+    expect(headers.Signature).toMatch(/^sig1=:/);
+    expect(headers["Content-Digest"]).toMatch(/^sha-256=:/);
     expect(headers.Date).toBeTruthy();
   });
 
@@ -131,7 +138,7 @@ describe("deliverActivity", () => {
       .calls[0]![1] as RequestInit;
     const headers = init.headers as Record<string, string>;
     expect(headers["User-Agent"]).toBe(TEST_UA);
-    expect(headers.Signature).toContain('keyId="');
+    expect(headers["Signature-Input"]).toContain('keyid="');
   });
 
   it("releases its timeout after a successful delivery", async () => {
@@ -150,8 +157,7 @@ describe("deliverActivity", () => {
         { keyId: "https://example.com/users/bob#main-key", privateKeyPem: pem },
         fetchImpl,
         TEST_UA,
-        undefined,
-        10,
+        { timeoutMs: 10 },
       );
       await vi.advanceTimersByTimeAsync(10);
 
@@ -161,8 +167,84 @@ describe("deliverActivity", () => {
     }
   });
 
+  it("double-knocks with draft-cavage when RFC 9421 is rejected", async () => {
+    const pem = await privateKeyPem();
+    const fetchImpl = vi.fn(async (_url: string, init?: RequestInit) => {
+      const headers = init?.headers as Record<string, string>;
+      return new Response(null, {
+        status: headers["Signature-Input"] ? 401 : 202,
+      });
+    }) as unknown as typeof fetch;
+
+    const result = await deliverActivity(
+      "https://remote.example/inbox",
+      JSON.stringify({ type: "Accept" }),
+      { keyId: "https://example.com/users/bob#main-key", privateKeyPem: pem },
+      fetchImpl,
+      TEST_UA,
+    );
+
+    expect(result).toEqual<DeliveryResult>({
+      ok: true,
+      status: 202,
+      profile: "cavage",
+    });
+    const calls = (fetchImpl as unknown as ReturnType<typeof vi.fn>).mock
+      .calls as [string, RequestInit][];
+    expect(calls).toHaveLength(2);
+    const second = calls[1]![1].headers as Record<string, string>;
+    expect(second["Signature-Input"]).toBeUndefined();
+    expect(second.Signature).toContain(
+      'keyId="https://example.com/users/bob#main-key"',
+    );
+    expect(second.Digest).toMatch(/^SHA-256=/);
+  });
+
+  it("skips the RFC 9421 knock for a known draft-cavage peer", async () => {
+    const pem = await privateKeyPem();
+    const fetchImpl = vi.fn(
+      async () => new Response(null, { status: 202 }),
+    ) as unknown as typeof fetch;
+
+    const result = await deliverActivity(
+      "https://remote.example/inbox",
+      "{}",
+      { keyId: "k", privateKeyPem: pem },
+      fetchImpl,
+      TEST_UA,
+      { preference: "cavage" },
+    );
+
+    expect(result).toEqual({ ok: true, status: 202, profile: "cavage" });
+    expect(fetchImpl).toHaveBeenCalledOnce();
+    const init = (fetchImpl as unknown as ReturnType<typeof vi.fn>).mock
+      .calls[0]![1] as RequestInit;
+    const headers = init.headers as Record<string, string>;
+    expect(headers["Signature-Input"]).toBeUndefined();
+    expect(headers.Signature).toContain('keyId="k"');
+  });
+
+  it.each([404, 410, 422, 500])(
+    "does not knock again on a non-signature %i",
+    async (status) => {
+      const pem = await privateKeyPem();
+      const fetchImpl = vi.fn(
+        async () => new Response(null, { status }),
+      ) as unknown as typeof fetch;
+      await deliverActivity(
+        "https://remote.example/inbox",
+        "{}",
+        { keyId: "k", privateKeyPem: pem },
+        fetchImpl,
+        TEST_UA,
+      );
+      expect(fetchImpl).toHaveBeenCalledOnce();
+    },
+  );
+
   it("classifies a 4xx as a permanent (non-retryable) failure", async () => {
     const pem = await privateKeyPem();
+    // Both knocks rejected: the draft-cavage answer is the one reported.
     const fetchImpl = (async () =>
       new Response("bad", { status: 400 })) as unknown as typeof fetch;
     const result = await deliverActivity(

packages/activitypub/src/delivery.ts

@@ -8,13 +8,40 @@
  * malicious actor cannot point a delivery at the Worker's own network. (DNS
  * rebinding is out of scope: the Workers runtime does not expose name
  * resolution to user code; see `@dwk/webmention`'s safe-fetch for the same
- * limitation.) The forthcoming `@dwk/http-signatures` package owns the signing
- * primitive; this module wires it to the network.
+ * limitation.) `signature.ts` owns the signing primitives; this module wires
+ * them to the network.
+ *
+ * Signing "double-knocks": a peer with no known preference is sent an RFC 9421
+ * signature first, and if it rejects that with a signature-shaped status
+ * (`400`/`401`/`403`) the same body is re-signed with draft-cavage and sent
+ * once more. The result reports which profile the peer accepted so the caller
+ * can remember draft-cavage peers and skip the first knock next time.
  */
 
 import { assertPublicUrl, safeFetch, SsrfError } from "@dwk/safe-fetch";
 
-import { signRequest, type SignerKey } from "./signature.js";
+import type { SignatureProfile } from "@dwk/http-signatures";
+
+import {
+  signRequest,
+  signRequestRfc9421,
+  type SignedRequest,
+  type SignerKey,
+} from "./signature.js";
+
+/**
+ * Statuses that may mean "I could not verify that signature", and so warrant a
+ * second knock with draft-cavage. Mastodon answers `401`; others use `400` or
+ * `403`. A `404`/`410`/`422`/`5xx` is not about the signature, so it never
+ * triggers the fallback.
+ *
+ * `403` is also what a peer that blocks our domain or actor answers, so each
+ * delivery to such a peer is sent twice (both knocks fail, and the row is
+ * dropped as permanent). That doubled traffic is accepted — the same trade-off
+ * Fedify makes — rather than dropping `403`, which would leave peers that use
+ * it for signature failures unreachable.
+ */
+const KNOCK_AGAIN_STATUSES: ReadonlySet<number> = new Set([400, 401, 403]);
 
 /** Machine-readable cause of a blocked delivery target. */
 export type BlockedReason =
@@ -62,34 +89,99 @@ function reasonFor(error: SsrfError): BlockedReason {
 
 /** The outcome of a single delivery attempt. */
 export type DeliveryResult =
-  | { readonly ok: true; readonly status: number }
+  | {
+      readonly ok: true;
+      readonly status: number;
+      /** The signature profile the peer accepted. */
+      readonly profile: SignatureProfile;
+    }
   | {
       readonly ok: false;
       readonly status: number;
       readonly retryable: boolean;
     };
 
+/** Optional knobs for {@link deliverActivity}. */
+export interface DeliveryOptions {
+  /** Clock, epoch milliseconds (deterministic tests). Defaults to `Date.now`. */
+  readonly now?: () => number;
+  /** Timeout for each `POST`, in milliseconds. Defaults to 10 s. */
+  readonly timeoutMs?: number;
+  /**
+   * The signature profile the peer is known to want: `"cavage"` skips the
+   * RFC 9421 knock; anything else (including unknown) double-knocks — RFC 9421
+   * first, then draft-cavage if the first answer is in
+   * {@link KNOCK_AGAIN_STATUSES}.
+   */
+  readonly preference?: SignatureProfile;
+}
+
 /**
  * Sign and `POST` one activity to a remote inbox. A `2xx` is success; `4xx`
  * (except `408`/`429`) is a permanent failure (drop) since the peer rejected the
  * request; `5xx`, `408`, `429`, and network errors are retryable.
+ *
+ * {@link DeliveryOptions.preference} selects the signing strategy — see there.
  */
 export async function deliverActivity(
   inboxUrl: string,
   activityJson: string,
   signer: SignerKey,
   fetchImpl: typeof fetch,
   userAgent: string,
-  now: () => number = () => Date.now(),
-  timeoutMs = 10_000,
+  options: DeliveryOptions = {},
 ): Promise<DeliveryResult> {
+  const now = options.now ?? (() => Date.now());
+  const timeoutMs = options.timeoutMs ?? 10_000;
+  const { preference } = options;
   // Throws DeliveryBlockedError for an unsafe target — the caller drops the row.
   assertPublicHttpsTarget(inboxUrl);
 
   const body = new TextEncoder().encode(activityJson);
-  const signed = await signRequest(inboxUrl, body, signer, { now });
+  const post = (signed: SignedRequest): Promise<Response | null> =>
+    postSigned(fetchImpl, inboxUrl, signed, userAgent, timeoutMs);
+
+  let profile: SignatureProfile =
+    preference === "cavage" ? "cavage" : "rfc9421";
+  let response =
+    profile === "cavage"
+      ? await post(await signRequest(inboxUrl, body, signer, { now }))
+      : await post(await signRequestRfc9421(inboxUrl, body, signer, { now }));
+  if (
+    profile === "rfc9421" &&
+    response !== null &&
+    KNOCK_AGAIN_STATUSES.has(response.status)
+  ) {
+    // The first knock's body is never read; release it before knocking again.
+    await response.body?.cancel();
+    profile = "cavage";
+    response = await post(await signRequest(inboxUrl, body, signer, { now }));
+  }
+  // Network error or timeout: worth retrying.
+  if (response === null) return { ok: false, status: 0, retryable: true };
+
+  if (response.status >= 200 && response.status < 300) {
+    return { ok: true, status: response.status, profile };
+  }
+  const retryable =
+    response.status >= 500 ||
+    response.status === 408 ||
+    response.status === 429;
+  return { ok: false, status: response.status, retryable };
+}
 
-  let response: Response;
+/**
+ * `POST` one signed delivery. Returns the response, or `null` on a network
+ * error or timeout; throws {@link DeliveryBlockedError} if a redirect hop lands
+ * on a blocked target.
+ */
+async function postSigned(
+  fetchImpl: typeof fetch,
+  inboxUrl: string,
+  signed: SignedRequest,
+  userAgent: string,
+  timeoutMs: number,
+): Promise<Response | null> {
   try {
     // Route the POST itself through the SSRF-safe wrapper too, not just the
     // pre-flight check on `inboxUrl`: a hostile inbox can 3xx-redirect a
@@ -102,30 +194,20 @@ export async function deliverActivity(
       {
         method: "POST",
         // `User-Agent` is layered on top of (not part of) the signed header
-        // set `signRequest` returns — it isn't covered by the signature, so
-        // adding it here doesn't touch signing at all.
+        // set — it isn't covered by the signature, so adding it here doesn't
+        // touch signing at all.
         headers: { ...signed.headers, "User-Agent": userAgent },
         body: signed.body as BufferSource,
       },
       { allowedSchemes: ["https:"], timeoutMs },
     );
-    response = result.response;
+    return result.response;
   } catch (error) {
     if (error instanceof SsrfError) {
       // A redirect hop landed on a blocked target — this inbox can never be
       // delivered to safely, same as failing the pre-flight check.
       throw new DeliveryBlockedError(reasonFor(error));
     }
-    // Network error or timeout: worth retrying.
-    return { ok: false, status: 0, retryable: true };
+    return null;
   }
-
-  if (response.status >= 200 && response.status < 300) {
-    return { ok: true, status: response.status };
-  }
-  const retryable =
-    response.status >= 500 ||
-    response.status === 408 ||
-    response.status === 429;
-  return { ok: false, status: response.status, retryable };
 }