Add authenticated canonical state reset

Author: chubes4

packages/runtime-cloudflare/src/request-routing.ts

@@ -4,6 +4,7 @@ export type WorkerRequestRoute =
   | { kind: "r2-state" }
   | { kind: "r2-mutate" }
   | { kind: "canonical-auth" }
+  | { kind: "operator-reset" }
   | { kind: "probe"; phase: string }
 
 export function routeWorkerRequest(request: Request): WorkerRequestRoute {
@@ -13,5 +14,6 @@ export function routeWorkerRequest(request: Request): WorkerRequestRoute {
   if (phase === "r2-state") return { kind: "r2-state" }
   if (phase === "r2-mutate") return { kind: "r2-mutate" }
   if (phase === "canonical-auth") return { kind: "canonical-auth" }
+  if (phase === "operator-reset") return { kind: "operator-reset" }
   return { kind: "probe", phase }
 }

packages/runtime-cloudflare/src/state-coordinator.ts

@@ -59,6 +59,7 @@ export class WordPressStateCoordinator implements DurableObject {
     if (action === "release") return Response.json(await this.release(body))
     if (action === "abort") return Response.json(await this.abort(body))
     if (action === "commit") return Response.json(await this.commit(body))
+    if (action === "reset") return Response.json(await this.reset())
     return new Response("Unknown coordinator action.", { status: 404 })
   }
 
@@ -120,6 +121,12 @@ export class WordPressStateCoordinator implements DurableObject {
     return lease
   }
 
+  private async reset(): Promise<{ reset: true }> {
+    await this.env.WORDPRESS_STATE_BUCKET.delete(POINTER_KEY)
+    await this.state.storage.delete(STORAGE_KEY)
+    return { reset: true }
+  }
+
   private async record(): Promise<CoordinatorRecord> {
     const stored = await this.state.storage.get<CoordinatorRecord>(STORAGE_KEY)
     if (stored?.initialized) return stored

packages/runtime-cloudflare/src/worker.ts

@@ -74,6 +74,7 @@ interface Env {
   WORDPRESS_STATE_BUCKET: R2Bucket
   WORDPRESS_ADMIN_PASSWORD?: string
   WORDPRESS_AUTH_SECRET?: string
+  WORDPRESS_OPERATOR_TOKEN?: string
 }
 
 export default {
@@ -82,6 +83,7 @@ export default {
     if (staticResponse) return staticResponse
     const route = routeWorkerRequest(request)
     const coordinator = env.WORDPRESS_STATE.getByName("default")
+    if (route.kind === "operator-reset") return resetCanonicalWordPress(request, env, coordinator)
     if (route.kind === "probe") {
       if (route.phase === "canonical-session") return canonicalSessionProbe(env.WORDPRESS_STATE_BUCKET, await coordinatorCall<{ pointer: MarkdownPointer | null }>(coordinator, request.url, "state"))
       return runBootProbe(route.phase, env.WORDPRESS_STATE_BUCKET)
@@ -135,6 +137,27 @@ interface Runtime {
 let cachedRuntime: { baseRevision: string; promise: Promise<Runtime> } | undefined
 const LEASE_ACQUISITION_TIMEOUT_MS = 100_000
 
+async function resetCanonicalWordPress(request: Request, env: Env, coordinator: DurableObjectStub): Promise<Response> {
+  if (request.method !== "POST") return new Response("Canonical reset requires POST.", { status: 405 })
+  const authorization = request.headers.get("authorization")
+  if (!env.WORDPRESS_OPERATOR_TOKEN || !authorization || !await secretsMatch(authorization, `Bearer ${env.WORDPRESS_OPERATOR_TOKEN}`)) {
+    return new Response("Canonical reset authorization failed.", { status: 401 })
+  }
+  const response = await coordinator.fetch(new Request(coordinatorUrl(request.url, "reset"), { method: "POST", headers: { "content-type": "application/json" }, body: "{}" }))
+  if (response.ok) await discardCachedRuntime()
+  return response
+}
+
+async function secretsMatch(left: string, right: string): Promise<boolean> {
+  const encoder = new TextEncoder()
+  const [leftHash, rightHash] = await Promise.all([crypto.subtle.digest("SHA-256", encoder.encode(left)), crypto.subtle.digest("SHA-256", encoder.encode(right))])
+  const leftBytes = new Uint8Array(leftHash)
+  const rightBytes = new Uint8Array(rightHash)
+  let difference = 0
+  for (let index = 0; index < leftBytes.length; index++) difference |= leftBytes[index] ^ rightBytes[index]
+  return difference === 0
+}
+
 async function runCoordinatedWordPressRequest(request: Request, env: Env, coordinator: DurableObjectStub, route: "wordpress" | "health" | "r2-mutate" | "canonical-auth"): Promise<Response> {
   if (route === "r2-mutate" && request.method !== "POST") return new Response("WordPress state mutation requires POST.", { status: 405 })
   let lease = await acquireLease(coordinator, request.url)

tests/cloudflare-runtime.test.ts

@@ -41,6 +41,7 @@ test("Cloudflare routing reserves phases while the phase-less route serves WordP
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=r2-state")), { kind: "r2-state" })
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=r2-mutate")), { kind: "r2-mutate" })
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=canonical-auth")), { kind: "canonical-auth" })
+  assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=operator-reset")), { kind: "operator-reset" })
   assert.deepEqual(routeWorkerRequest(new Request("https://worker.example/?phase=seeded-wordpress")), { kind: "probe", phase: "seeded-wordpress" })
 })
 
@@ -543,11 +544,13 @@ test("Cloudflare coordinator serializes leases, promotes with CAS, and recovers
     storage: {
       get: async <T>(key: string) => values.get(key) as T | undefined,
       put: async (key: string, value: unknown) => { values.set(key, value) },
+      delete: async (key: string) => { values.delete(key) },
     },
   }
   const bucket = {
     get: async (key: string) => objects.has(key) ? { json: async <T>() => JSON.parse(objects.get(key)!) as T } : null,
     put: async (key: string, value: string) => { objects.set(key, value) },
+    delete: async (key: string) => { objects.delete(key) },
   }
   const coordinator = new WordPressStateCoordinator(state as never, { WORDPRESS_STATE_BUCKET: bucket as never, COORDINATOR_LEASE_MS: 50 })
   const call = async (action: string, body: Record<string, unknown> = {}) => coordinator.fetch(new Request(`https://worker.example/?__wp_codebox_coordinator=${action}`, { method: action === "state" ? "GET" : "POST", headers: { "content-type": "application/json" }, body: action === "state" ? undefined : JSON.stringify(body) }))
@@ -566,6 +569,10 @@ test("Cloudflare coordinator serializes leases, promotes with CAS, and recovers
   const recovered = await (await call("begin")).json() as { token: string }
   assert.notEqual(recovered.token, stale.token)
   assert.equal((await call("release", { token: recovered.token })).status, 200)
+  assert.equal((await call("reset")).status, 200)
+  const resetState = await (await call("state")).json() as { pointer: null; version: number }
+  assert.equal(resetState.pointer, null)
+  assert.equal(resetState.version, 0)
 })
 
 test("serialized Cloudflare mutations use MDI flush paths and complete canonical state", async () => {