From e7e3172ea73f0358012ebc0ef504b87be35178af Mon Sep 17 00:00:00 2001 From: Ugur Cekmez Date: Wed, 26 Aug 2026 22:07:18 +0300 Subject: [PATCH] feat(spec,middleware): specify unsubscribe and enforce the subscription lease MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two halves of the same hole in §10. **Unsubscribe existed everywhere except the normative spec.** The middleware implemented `DELETE`, `ws-message.json` defined `unsubscribe` / `unsubscribed`, the IETF draft referenced it, and the conformance fixtures README claimed to hold vectors for it. SPECIFICATION.md contained the string zero times, and the §10 lifecycle diagram had no cancellation path and no terminal state other than `rejected`. A subscription protocol with no specified way to stop a subscription is the first thing a new implementer hits. **The WebSub lease was advertised and never honoured.** Intent verification sends `hub.lease_seconds=2592000`, but nothing defined expiry, renewal, or `hub.mode=unsubscribe`. EEP borrowed WebSub's challenge handshake without its lifecycle, so the value was decorative: subscriptions were immortal and an abandoned `delivery_url` received traffic forever with no defined way to garbage-collect it. Spec: - §10 lifecycle gains `expired` and `cancelled` as terminal states, a state table, and an explicit rule that non-`active` subscriptions are not delivered to. - §10.1 Cancellation: `204`, idempotent, terminal, id not reusable, `delivery_secret` discarded, in-flight deliveries not retried. Idempotency matters — returning 404 on a repeat DELETE tells a retrying client its own successful cancellation failed. - §10.2 Lease lifetime: `hub.lease_seconds` is binding, `expires_at` is reported on every representation, renewal is re-subscription with intent verification and preserves the `subscription_id`, and a publisher that will not enforce a lease MUST NOT advertise one. - Unsubscribe verification: an unauthenticated WebSub-style unsubscribe MUST complete a `hub.mode=unsubscribe` challenge, otherwise a caller could cancel someone else's delivery by guessing a `subscription_id`. - Six `subscription.*` lifecycle events added to the §9 catalog. Schema: `lease_seconds` on `subscription.request.json` (optional, 300..31536000), types regenerated. TypeScript and Python middleware: - Grant a lease at creation, clamping a requested value into policy rather than rejecting the subscription, and report `expires_at`. - `WebhookDispatcher` refuses to deliver once the lease has elapsed. The check is at delivery time rather than in a sweeper, so the rule holds in deployments with no background job — an unenforced lease is no lease. - Cancellation is idempotent and emits `subscription.cancelled` with a reason, replacing the non-catalogued `subscription.deleted`. Conformance fixtures for a lease request and a below-minimum lease. Refs: EEP audit 2026-08 findings A2, A3 Signed-off-by: Ugur Cekmez --- docs/current/SPECIFICATION.md | 125 ++++++++- .../middleware/src/core/eep-server.test.ts | 132 +++++++++- .../middleware/src/core/eep-server.ts | 59 ++++- .../middleware/src/core/request-handler.ts | 40 ++- .../src/dispatcher/webhook-dispatcher.test.ts | 65 +++++ .../src/dispatcher/webhook-dispatcher.ts | 15 ++ .../eep_middleware/adapters.py | 4 + .../eep_middleware/core.py | 38 ++- .../eep-middleware-python/tests/test_core.py | 49 +++- schemas/v0.1/subscription.request.json | 245 +++++++++--------- tests/conformance-fixtures/manifest.json | 22 ++ .../lease-below-minimum.expected.json | 4 + .../lease-below-minimum.input.json | 9 + .../subscription/lease-request.expected.json | 4 + .../subscription/lease-request.input.json | 9 + tests/types/eep-schemas.d.ts | 4 + 16 files changed, 668 insertions(+), 156 deletions(-) create mode 100644 tests/conformance-fixtures/subscription/lease-below-minimum.expected.json create mode 100644 tests/conformance-fixtures/subscription/lease-below-minimum.input.json create mode 100644 tests/conformance-fixtures/subscription/lease-request.expected.json create mode 100644 tests/conformance-fixtures/subscription/lease-request.input.json diff --git a/docs/current/SPECIFICATION.md b/docs/current/SPECIFICATION.md index 6ee5f2f..9603e30 100644 --- a/docs/current/SPECIFICATION.md +++ b/docs/current/SPECIFICATION.md @@ -649,6 +649,16 @@ EEP event types follow a reverse-domain dot notation pattern: | `com.example.agent.task.completed` | An A2A task completed successfully | | `com.example.agent.task.failed` | An A2A task failed | +### Subscription lifecycle +| Event Type | Description | +|------------|-------------| +| `com.example.subscription.created` | A subscription was created and verified | +| `com.example.subscription.paused` | Delivery was suspended (see §10) | +| `com.example.subscription.resumed` | Delivery resumed after a pause | +| `com.example.subscription.expiring` | The lease is close to elapsing; renew to keep receiving (§10.2) | +| `com.example.subscription.expired` | The lease elapsed without renewal; terminal | +| `com.example.subscription.cancelled` | The subscription was cancelled; terminal (§10.1) | + ### Commerce and marketplace | Event Type | Description | |------------|-------------| @@ -681,23 +691,91 @@ POST /eep/subscribe ▼ [Challenge Response from Subscriber] │ - Success ──────────────────► [active] - │ │ - Failure event delivery - ▼ │ - [rejected] 5 consecutive failed deliveries - ▼ - [paused] - │ - POST /eep/subscriptions/{subscription_id}/resume - ▼ - [active] -``` + Success ──────────────────► [active] ◄──────────────┐ + │ │ │ │ │ + Failure │ │ │ re-subscribe (lease renewal) + ▼ │ │ └───────────────┘ + [rejected] │ │ + │ └── lease elapses ──► [expired] (terminal) + │ + 5 consecutive failed deliveries + ▼ + [paused] + │ + POST /eep/subscriptions/{id}/resume + ▼ + [active] + + Any non-terminal state ── DELETE /eep/subscriptions/{id} ──► [cancelled] (terminal) +``` + +| State | Meaning | Terminal | +|---|---|---| +| `pending_verification` | Created; awaiting WebSub intent verification | no | +| `active` | Verified and receiving deliveries | no | +| `paused` | Delivery suspended after repeated failures, or by request | no | +| `rejected` | Intent verification failed or timed out | **yes** | +| `expired` | The lease elapsed without renewal | **yes** | +| `cancelled` | Cancelled by the subscriber or the publisher | **yes** | A "failed delivery" is one that exhausted the full §5.4 retry schedule. The counter is consecutive and resets on the next delivery that the subscriber acknowledges with a 2xx. Endpoint paths are normative in §5.1.1. +Publishers MUST NOT deliver events for a subscription in `paused`, `rejected`, +`expired` or `cancelled`. + +### 10.1 Cancellation (normative) + +`DELETE /eep/subscriptions/{subscription_id}` cancels a subscription. On success +the publisher MUST: + +1. Respond `204 No Content`. The operation is idempotent — a second `DELETE` of + an already-cancelled subscription MUST also return `204`, not `404`, so a + retrying client converges. +2. Stop delivering immediately. Deliveries already in flight MAY complete; + retries for them MUST NOT be scheduled. +3. Move the subscription to `cancelled`, a terminal state. A cancelled + `subscription_id` MUST NOT be reusable — a subscriber that wants delivery + again creates a new subscription. +4. Discard the `delivery_secret`. + +Publishers MAY cancel a subscription unilaterally (for example when the +underlying entity is deleted, or an agreement gate is revoked). When they do, +they SHOULD emit `com.example.subscription.cancelled` on any other channel the +subscriber holds, and MUST include a `reason`. + +Cancellation is how a subscriber exercises data-minimisation obligations over +the delivery relationship itself; `data_request`-style erasure of already +delivered payloads is covered separately in §16. + +### 10.2 Lease lifetime and renewal (normative) + +Intent verification carries `hub.lease_seconds` (see below). That value is a +contract, not decoration: **a subscription is time-bounded and expires unless +renewed.** Without expiry, an abandoned `delivery_url` receives traffic forever +and a publisher has no defined way to garbage-collect it. + +- Publishers MUST treat `hub.lease_seconds` as the lifetime of the + subscription, starting from successful verification. +- The `201` creation response and every subscription representation (§5.1.1) + MUST carry `expires_at`, an RFC 3339 timestamp. +- A subscriber renews by re-subscribing with the same `source_did`, + `event_types` and `delivery_url`. The publisher MUST perform intent + verification again and, on success, extend the existing subscription rather + than creating a duplicate — the `subscription_id` is preserved. +- When the lease elapses without renewal, the publisher MUST move the + subscription to `expired` and stop delivering. +- Publishers SHOULD emit `com.example.subscription.expiring` to the subscriber + ahead of expiry — at least 24 hours before, or at 10% of the lease remaining, + whichever is sooner — carrying `subscription_id` and `expires_at`. + +A subscriber MAY request a lease by sending `lease_seconds` in the subscription +request. Publishers MAY clamp it to their own policy and MUST report the value +actually granted in `expires_at`. Publishers that do not implement expiry MUST +NOT advertise a `hub.lease_seconds` they will not honour; they SHOULD omit the +parameter instead of sending a value that means nothing. + ### WebSub intent verification When creating a webhook subscription, the publisher MUST perform intent verification: @@ -716,6 +794,29 @@ When creating a webhook subscription, the publisher MUST perform intent verifica Intent verification prevents malicious actors from registering unauthorized URLs to bounce traffic through the publisher. +`hub.lease_seconds` is the lifetime the publisher is granting; see §10.2. A +publisher that sends it MUST enforce it. + +### Unsubscribe verification + +`DELETE /eep/subscriptions/{subscription_id}` (§10.1) is authenticated by the +caller's API key, so it needs no callback round-trip. + +A publisher MAY additionally accept WebSub-style unsubscribe, where the request +is not authenticated as the subscription's owner. In that case the publisher +MUST verify intent exactly as it does for `subscribe`, with `hub.mode` set to +`unsubscribe`: + +``` +?hub.mode=unsubscribe +&hub.topic=did:web:example.com:u:acme-corp +&hub.challenge=random_secure_string_32_chars +``` + +The subscriber endpoint MUST echo `hub.challenge`. Without this round-trip an +unauthenticated caller could cancel another subscriber's delivery by guessing a +`subscription_id`, so publishers MUST NOT act on an unverified unsubscribe. + --- ## 11. Authentication and authorization diff --git a/packages/@eep-dev/middleware/src/core/eep-server.test.ts b/packages/@eep-dev/middleware/src/core/eep-server.test.ts index 0e17014..a08f6f5 100644 --- a/packages/@eep-dev/middleware/src/core/eep-server.test.ts +++ b/packages/@eep-dev/middleware/src/core/eep-server.test.ts @@ -1,7 +1,12 @@ import { describe, expect, it, vi } from "vitest"; import { parseGateConfig, type GateProof, type ProofVerifier } from "@eep-dev/gates"; import { EEPServer } from "./eep-server.js"; -import { TEST_DELIVERY_EVENT_TYPE } from "./request-handler.js"; +import { + TEST_DELIVERY_EVENT_TYPE, + DEFAULT_LEASE_SECONDS, + MIN_LEASE_SECONDS, + MAX_LEASE_SECONDS +} from "./request-handler.js"; import type { EventBusAdapter, DBAdapter, SubscriptionRecord, SubscriptionUpdate } from "./request-handler.js"; // Prevent real DNS resolution during subscribe validation tests. @@ -595,6 +600,88 @@ describe("EEPServer", () => { }); }); + // SPECIFICATION.md §10.2 — a subscription is time-bounded. `hub.lease_seconds` + // was advertised during intent verification but never enforced, which made + // it decorative: an abandoned delivery_url received traffic forever. + describe("lease lifetime (§10.2)", () => { + const makeServer = () => { + const db = new RecordingDBAdapter(); + const bus = new RecordingEventBusAdapter(); + const server = new EEPServer({ + baseUrl: "https://api.example.com", + did: "did:web:example.com", + dbAdapter: db, + eventBusAdapter: bus + }); + return { db, bus, server }; + }; + + const subscribe = async (server: EEPServer, extra: Record = {}) => + server.getSubscribeHandler()({ + method: "POST", + path: "/eep/subscribe", + headers: {}, + body: { + source_did: "did:web:agent.example", + delivery_method: "webhook", + delivery_url: "https://hook.example/notify", + event_types: ["com.example.entity.updated"], + ...extra + } + }); + + it("grants the default 30-day lease when none is requested", async () => { + const { server } = makeServer(); + const res = await subscribe(server); + const body = res.body as { expires_at: string; created_at: string }; + const seconds = (Date.parse(body.expires_at) - Date.parse(body.created_at)) / 1000; + expect(seconds).toBe(DEFAULT_LEASE_SECONDS); + }); + + it("honours a requested lease within policy", async () => { + const { server } = makeServer(); + const res = await subscribe(server, { lease_seconds: 3600 }); + const body = res.body as { expires_at: string; created_at: string }; + expect((Date.parse(body.expires_at) - Date.parse(body.created_at)) / 1000).toBe(3600); + }); + + it("clamps a lease below the minimum rather than rejecting the subscription", async () => { + const { server } = makeServer(); + const res = await subscribe(server, { lease_seconds: 1 }); + const body = res.body as { expires_at: string; created_at: string }; + expect(res.status).toBe(201); + expect((Date.parse(body.expires_at) - Date.parse(body.created_at)) / 1000).toBe(MIN_LEASE_SECONDS); + }); + + it("clamps a lease above the maximum", async () => { + const { server } = makeServer(); + const res = await subscribe(server, { lease_seconds: 99_999_999 }); + const body = res.body as { expires_at: string; created_at: string }; + expect((Date.parse(body.expires_at) - Date.parse(body.created_at)) / 1000).toBe(MAX_LEASE_SECONDS); + }); + + it("falls back to the default for a non-numeric lease", async () => { + const { server } = makeServer(); + const res = await subscribe(server, { lease_seconds: "forever" }); + const body = res.body as { expires_at: string; created_at: string }; + expect(res.status).toBe(201); + expect((Date.parse(body.expires_at) - Date.parse(body.created_at)) / 1000).toBe(DEFAULT_LEASE_SECONDS); + }); + + it("reports expires_at on the subscription representation", async () => { + const { server } = makeServer(); + const created = await subscribe(server); + const id = (created.body as { subscription_id: string }).subscription_id; + const status = await server.getSubscriptionStatusHandler()({ + method: "GET", + path: `/eep/subscriptions/${id}`, + headers: {}, + params: { subscriptionId: id } + }); + expect(typeof (status.body as { expires_at?: string }).expires_at).toBe("string"); + }); + }); + // SPECIFICATION.md §5.1.1 — list / resume / test-delivery member operations. describe("subscription collection operations (§5.1.1)", () => { const makeServer = () => { @@ -813,7 +900,7 @@ describe("EEPServer", () => { expect(res.status).toBe(400); }); - it("deletes the subscription and publishes subscription.deleted", async () => { + it("cancels the subscription and publishes subscription.cancelled", async () => { const { server, db, bus } = makeServer(); const id = await createSubscription(server); expect(db.rows.length).toBe(1); @@ -821,27 +908,58 @@ describe("EEPServer", () => { const res = await server.getUnsubscribeHandler()({ method: "DELETE", - path: `/eep/subscribe/${id}`, + path: `/eep/subscriptions/${id}`, headers: {}, params: { subscriptionId: id } }); expect(res.status).toBe(204); expect(db.rows.length).toBe(0); - expect(bus.published).toEqual(["subscription.deleted"]); + expect(bus.published).toEqual(["subscription.cancelled"]); + expect(bus.events.at(-1)?.data).toMatchObject({ + subscription_id: id, + reason: expect.any(String) + }); }); - it("returns 404 when DELETE targets an unknown id and skips publish", async () => { + // §10.1: cancellation is idempotent. Returning 404 on the second DELETE + // would tell a retrying client its own successful cancellation failed. + it("returns 204 for an unknown id and publishes nothing", async () => { const { server, bus } = makeServer(); const res = await server.getUnsubscribeHandler()({ method: "DELETE", - path: "/eep/subscribe/sub_missing", + path: "/eep/subscriptions/sub_missing", headers: {}, params: { subscriptionId: "sub_missing" } }); - expect(res.status).toBe(404); + expect(res.status).toBe(204); expect(bus.published).toEqual([]); }); + it("is idempotent across repeated cancellation", async () => { + const { server, db, bus } = makeServer(); + const id = await createSubscription(server); + bus.published.length = 0; + + const first = await server.getUnsubscribeHandler()({ + method: "DELETE", + path: `/eep/subscriptions/${id}`, + headers: {}, + params: { subscriptionId: id } + }); + const second = await server.getUnsubscribeHandler()({ + method: "DELETE", + path: `/eep/subscriptions/${id}`, + headers: {}, + params: { subscriptionId: id } + }); + + expect(first.status).toBe(204); + expect(second.status).toBe(204); + expect(db.rows.length).toBe(0); + // The lifecycle event fires once, for the transition that happened. + expect(bus.published).toEqual(["subscription.cancelled"]); + }); + it("returns 400 when DELETE has no subscription id param", async () => { const { server } = makeServer(); const res = await server.getUnsubscribeHandler()({ diff --git a/packages/@eep-dev/middleware/src/core/eep-server.ts b/packages/@eep-dev/middleware/src/core/eep-server.ts index a6a241d..11a1626 100644 --- a/packages/@eep-dev/middleware/src/core/eep-server.ts +++ b/packages/@eep-dev/middleware/src/core/eep-server.ts @@ -10,7 +10,12 @@ import { type ProofVerifier } from "@eep-dev/gates"; import { SSRFError, validateEventTypePattern, validateSSRF } from "@eep-dev/validator"; -import { TEST_DELIVERY_EVENT_TYPE } from "./request-handler.js"; +import { + TEST_DELIVERY_EVENT_TYPE, + DEFAULT_LEASE_SECONDS, + MIN_LEASE_SECONDS, + MAX_LEASE_SECONDS +} from "./request-handler.js"; import type { AuthAdapter, CloudEvent, @@ -91,6 +96,24 @@ class HeaderProofAuthAdapter implements AuthAdapter { } } +/** + * Clamp a requested `lease_seconds` into publisher policy (§10.2). + * + * A non-integer, out-of-range or absent value falls back to the default rather + * than being rejected: the lease is the publisher's to grant, and a subscriber + * asking for something unreasonable should still get a working subscription + * with an honest `expires_at`. + */ +function clampLeaseSeconds(requested: unknown): number { + if (typeof requested !== "number" || !Number.isFinite(requested)) { + return DEFAULT_LEASE_SECONDS; + } + const seconds = Math.floor(requested); + if (seconds < MIN_LEASE_SECONDS) return MIN_LEASE_SECONDS; + if (seconds > MAX_LEASE_SECONDS) return MAX_LEASE_SECONDS; + return seconds; +} + const DEFAULT_GATE_CONFIG = parseGateConfig({ default_tier: "public", tiers: { @@ -334,6 +357,14 @@ export class EEPServer { // Returned to the subscriber once, on creation, and never again. const deliverySecret = deliveryMethod === "webhook" ? randomBytes(24).toString("base64url") : undefined; + // Subscriptions are time-bounded (§10.2). A subscriber MAY request a + // lease; the publisher clamps it to policy and reports what it actually + // granted as `expires_at`. Advertising a lease and not enforcing it is + // what made `hub.lease_seconds` decorative. + const leaseSeconds = clampLeaseSeconds(body.lease_seconds); + const createdAt = new Date(); + const expiresAt = new Date(createdAt.getTime() + leaseSeconds * 1000); + const subscription: SubscriptionRecord = { subscription_id: `sub_${Date.now()}`, source_did: sourceDid, @@ -342,10 +373,11 @@ export class EEPServer { event_types: rawEventTypes, status: "active", failure_count: 0, + expires_at: expiresAt.toISOString(), delivery_secret: deliverySecret, metadata, tier, - created_at: new Date().toISOString() + created_at: createdAt.toISOString() }; await this.dbAdapter.saveSubscription(subscription); @@ -539,20 +571,19 @@ export class EEPServer { body: { error: "invalid_request", message: "subscription_id is required" } }; } + // §10.1: cancellation is idempotent. A second DELETE of an + // already-cancelled subscription returns 204, not 404, so a retrying + // client converges instead of treating its own success as a failure. const deleted = await this.dbAdapter.deleteSubscription(subscriptionId); - if (!deleted) { - return { - status: 404, - body: { error: "not_found", message: `subscription ${subscriptionId} does not exist` } - }; + if (deleted) { + await this.eventBusAdapter.publish({ + id: `evt_${Date.now()}`, + type: "subscription.cancelled", + source: this.did, + time: new Date().toISOString(), + data: { subscription_id: subscriptionId, reason: "cancelled_by_subscriber" } + }); } - await this.eventBusAdapter.publish({ - id: `evt_${Date.now()}`, - type: "subscription.deleted", - source: this.did, - time: new Date().toISOString(), - data: { subscription_id: subscriptionId } - }); return { status: 204, body: null }; }; } diff --git a/packages/@eep-dev/middleware/src/core/request-handler.ts b/packages/@eep-dev/middleware/src/core/request-handler.ts index b665908..51dafef 100644 --- a/packages/@eep-dev/middleware/src/core/request-handler.ts +++ b/packages/@eep-dev/middleware/src/core/request-handler.ts @@ -34,14 +34,48 @@ export type RouteDefinition = { */ export const TEST_DELIVERY_EVENT_TYPE = "com.eep.subscription.test"; +/** + * Subscription lifecycle states (SPECIFICATION.md §10). + * + * `rejected`, `expired` and `cancelled` are terminal: a subscription in one of + * them never delivers again and its id is not reusable. + */ +export type SubscriptionStatus = + | "pending_verification" + | "active" + | "paused" + | "rejected" + | "expired" + | "cancelled"; + +/** States in which a publisher MUST NOT deliver events. */ +export const TERMINAL_SUBSCRIPTION_STATUSES: readonly SubscriptionStatus[] = [ + "rejected", + "expired", + "cancelled", +]; + +/** Default lease granted when the subscriber does not request one: 30 days. */ +export const DEFAULT_LEASE_SECONDS = 2_592_000; + +/** Bounds a publisher will clamp a requested `lease_seconds` into. */ +export const MIN_LEASE_SECONDS = 300; +export const MAX_LEASE_SECONDS = 31_536_000; + export type SubscriptionRecord = { subscription_id: string; source_did: string; delivery_method: "sse" | "webhook"; callback_url?: string; event_types: string[]; - status: "active" | "paused"; + status: SubscriptionStatus; failure_count: number; + /** + * RFC 3339 timestamp at which the lease elapses (SPECIFICATION.md §10.2). + * A subscription past this instant MUST NOT receive deliveries; it moves to + * `expired`. Absent means the publisher grants an unbounded lease. + */ + expires_at?: string; /** Per-subscription HMAC secret returned to the subscriber on creation. */ delivery_secret?: string; /** Subscriber-defined metadata (passed through, not interpreted). */ @@ -68,7 +102,9 @@ export type EventBusAdapter = { subscribe: (pattern: string, handler: (event: CloudEvent) => void) => Promise; }; -export type SubscriptionUpdate = Partial>; +export type SubscriptionUpdate = Partial< + Pick +>; export type DBAdapter = { saveSubscription: (subscription: SubscriptionRecord) => Promise; diff --git a/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.test.ts b/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.test.ts index 0bac34f..894a1d0 100644 --- a/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.test.ts +++ b/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.test.ts @@ -59,6 +59,71 @@ describe("WebhookDispatcher", () => { expect(DEFAULT_RETRY_SCHEDULE_MS).toEqual([0, 5_000, 30_000, 120_000, 900_000, 3_600_000, 21_600_000]); }); + // SPECIFICATION.md §10.2: an elapsed lease means no more deliveries. Checked + // at delivery time rather than by a sweeper, so the rule holds in a + // deployment with no background job — an unenforced lease is no lease. + describe("lease enforcement (§10.2)", () => { + const iso = (offsetMs: number) => new Date(Date.now() + offsetMs).toISOString(); + + it("does not deliver to a subscription whose lease has elapsed", async () => { + const db = new InMemoryDBAdapter(); + await db.saveSubscription(subscription({ expires_at: iso(-1000) })); + const { client, calls } = mockClient([200]); + const dispatcher = new WebhookDispatcher({ db, httpClient: client, retryScheduleMs: NO_DELAY }); + + const results = await dispatcher.dispatch(event()); + + expect(results).toHaveLength(0); + expect(calls).toHaveLength(0); + }); + + it("still delivers while the lease is current", async () => { + const db = new InMemoryDBAdapter(); + await db.saveSubscription(subscription({ expires_at: iso(60_000) })); + const { client, calls } = mockClient([200]); + const dispatcher = new WebhookDispatcher({ db, httpClient: client, retryScheduleMs: NO_DELAY }); + + const results = await dispatcher.dispatch(event()); + + expect(results).toHaveLength(1); + expect(calls).toHaveLength(1); + }); + + it("treats an absent expires_at as an unbounded lease", async () => { + const db = new InMemoryDBAdapter(); + await db.saveSubscription(subscription()); + const { client, calls } = mockClient([200]); + const dispatcher = new WebhookDispatcher({ db, httpClient: client, retryScheduleMs: NO_DELAY }); + + expect(await dispatcher.dispatch(event())).toHaveLength(1); + expect(calls).toHaveLength(1); + }); + + it("ignores an unparseable expires_at rather than dropping traffic", async () => { + const db = new InMemoryDBAdapter(); + await db.saveSubscription(subscription({ expires_at: "not-a-timestamp" })); + const { client, calls } = mockClient([200]); + const dispatcher = new WebhookDispatcher({ db, httpClient: client, retryScheduleMs: NO_DELAY }); + + expect(await dispatcher.dispatch(event())).toHaveLength(1); + expect(calls).toHaveLength(1); + }); + + it("does not deliver a test event to an expired subscription either", async () => { + const db = new InMemoryDBAdapter(); + await db.saveSubscription(subscription({ subscription_id: "sub_target", expires_at: iso(-1) })); + const { client, calls } = mockClient([200]); + const dispatcher = new WebhookDispatcher({ db, httpClient: client, retryScheduleMs: NO_DELAY }); + + const results = await dispatcher.dispatch( + event({ id: "evt_test_x", type: TEST_DELIVERY_EVENT_TYPE, data: { subscription_id: "sub_target" } }) + ); + + expect(results).toHaveLength(0); + expect(calls).toHaveLength(0); + }); + }); + // SPECIFICATION.md §5.1.1: a synthetic test delivery is addressed to ONE // subscription. It must reach that subscriber even though // `com.eep.subscription.test` matches none of their `event_types`, and it diff --git a/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts b/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts index d14ddef..67b3de8 100644 --- a/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts +++ b/packages/@eep-dev/middleware/src/dispatcher/webhook-dispatcher.ts @@ -171,6 +171,7 @@ export class WebhookDispatcher { const deliverable = sub.delivery_method === "webhook" && sub.status === "active" && + !this.leaseHasElapsed(sub) && typeof sub.callback_url === "string" && sub.callback_url.length > 0; if (!deliverable) return false; @@ -188,6 +189,20 @@ export class WebhookDispatcher { return matchesAnyPattern(event.type, sub.event_types); } + /** + * True once the subscription's lease has elapsed (SPECIFICATION.md §10.2). + * + * Checked at delivery time rather than relying on a sweeper, so an expired + * subscription stops receiving events even in a deployment that has no + * background job — an unenforced lease is the same as no lease at all. + */ + private leaseHasElapsed(sub: SubscriptionRecord): boolean { + if (typeof sub.expires_at !== "string") return false; + const expiresAt = Date.parse(sub.expires_at); + if (Number.isNaN(expiresAt)) return false; + return expiresAt <= Date.now(); + } + private async deliverWithRetry(event: CloudEvent, sub: SubscriptionRecord): Promise { let lastStatus: number | undefined; diff --git a/packages/eep-middleware-python/eep_middleware/adapters.py b/packages/eep-middleware-python/eep_middleware/adapters.py index 6cd1cf3..96c5aaa 100644 --- a/packages/eep-middleware-python/eep_middleware/adapters.py +++ b/packages/eep-middleware-python/eep_middleware/adapters.py @@ -14,6 +14,10 @@ class SubscriptionRecord: delivery_method: str callback_url: str | None created_at: str + #: RFC 3339 instant at which the lease elapses (SPECIFICATION.md §10.2). + #: A subscription past this instant MUST NOT receive deliveries. ``None`` + #: means the publisher granted an unbounded lease. + expires_at: str | None = None @dataclass(slots=True) diff --git a/packages/eep-middleware-python/eep_middleware/core.py b/packages/eep-middleware-python/eep_middleware/core.py index 76e481e..056df08 100644 --- a/packages/eep-middleware-python/eep_middleware/core.py +++ b/packages/eep-middleware-python/eep_middleware/core.py @@ -26,6 +26,31 @@ async def extract_proofs(self, headers: dict[str, str], query: dict[str, str] | return [] +#: Default lease granted when the subscriber does not request one: 30 days. +DEFAULT_LEASE_SECONDS = 2_592_000 +#: Bounds a publisher clamps a requested ``lease_seconds`` into. +MIN_LEASE_SECONDS = 300 +MAX_LEASE_SECONDS = 31_536_000 + + +def clamp_lease_seconds(requested: Any) -> int: + """Clamp a requested ``lease_seconds`` into publisher policy (§10.2). + + A non-integer, out-of-range or absent value falls back to the default + rather than being rejected: the lease is the publisher's to grant, and a + subscriber asking for something unreasonable should still end up with a + working subscription and an honest ``expires_at``. + """ + if isinstance(requested, bool) or not isinstance(requested, (int, float)): + return DEFAULT_LEASE_SECONDS + seconds = int(requested) + if seconds < MIN_LEASE_SECONDS: + return MIN_LEASE_SECONDS + if seconds > MAX_LEASE_SECONDS: + return MAX_LEASE_SECONDS + return seconds + + class EEPServer: def __init__( self, @@ -121,12 +146,20 @@ async def create_subscription(self, payload: dict[str, Any]) -> tuple[int, dict[ except SSRFError as err: return 400, {"error": "invalid_request", "message": f"delivery_url is not allowed: {err}"} + # Subscriptions are time-bounded (§10.2). A subscriber MAY request a + # lease; the publisher clamps it to policy and reports what it actually + # granted as `expires_at`. Advertising `hub.lease_seconds` during intent + # verification and then never enforcing it is what made the value + # decorative. + lease_seconds = clamp_lease_seconds(payload.get("lease_seconds")) + created_at = time.time() subscription = SubscriptionRecord( - subscription_id=f"sub_{int(time.time() * 1000)}", + subscription_id=f"sub_{int(created_at * 1000)}", source_did=source_did, delivery_method=str(delivery_method), callback_url=delivery_url, - created_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + created_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(created_at)), + expires_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(created_at + lease_seconds)), ) await self.db_adapter.save_subscription(subscription) await self.event_bus_adapter.publish( @@ -148,6 +181,7 @@ async def create_subscription(self, payload: dict[str, Any]) -> tuple[int, dict[ "delivery_method": subscription.delivery_method, "callback_url": subscription.callback_url, "created_at": subscription.created_at, + "expires_at": subscription.expires_at, } async def audit_payload(self) -> dict[str, Any]: diff --git a/packages/eep-middleware-python/tests/test_core.py b/packages/eep-middleware-python/tests/test_core.py index cfe9ee5..ff4324e 100644 --- a/packages/eep-middleware-python/tests/test_core.py +++ b/packages/eep-middleware-python/tests/test_core.py @@ -3,7 +3,13 @@ import pytest from eep_gates import ProofVerifier -from eep_middleware.core import EEPServer +from eep_middleware.core import ( + DEFAULT_LEASE_SECONDS, + MAX_LEASE_SECONDS, + MIN_LEASE_SECONDS, + EEPServer, + clamp_lease_seconds, +) class _PaymentVerifier(ProofVerifier): @@ -185,3 +191,44 @@ async def test_sse_subscription_skips_ssrf_and_audits() -> None: await server.subscribe_to_events("subscription.*", lambda event: events.append(event.event_type)) audit = await server.audit_payload() assert audit["subscriptions_count"] == 1 + + +# ── Lease lifetime (SPECIFICATION.md §10.2) ─────────────────────────────── +# +# `hub.lease_seconds` was advertised during intent verification and never +# enforced, which made it decorative: an abandoned delivery_url received +# traffic forever and a publisher had no defined way to garbage-collect it. + + +def test_clamp_lease_seconds_defaults_when_absent_or_non_numeric(): + assert clamp_lease_seconds(None) == DEFAULT_LEASE_SECONDS + assert clamp_lease_seconds("forever") == DEFAULT_LEASE_SECONDS + # `bool` is an `int` subclass in Python; a flag is not a lease. + assert clamp_lease_seconds(True) == DEFAULT_LEASE_SECONDS + + +def test_clamp_lease_seconds_clamps_to_policy_bounds(): + assert clamp_lease_seconds(1) == MIN_LEASE_SECONDS + assert clamp_lease_seconds(99_999_999) == MAX_LEASE_SECONDS + + +def test_clamp_lease_seconds_honours_a_value_within_bounds(): + assert clamp_lease_seconds(3600) == 3600 + # Fractional seconds are truncated, not rejected. + assert clamp_lease_seconds(3600.9) == 3600 + + +@pytest.mark.asyncio +async def test_create_subscription_reports_the_granted_lease(): + server = EEPServer(base_url="https://api.example.com", did="did:web:example.com") + status, body = await server.create_subscription( + { + "source_did": "did:web:agent.example", + "delivery_method": "sse", + "event_types": ["com.example.entity.updated"], + } + ) + assert status == 201 + # A subscription is time-bounded; the publisher reports what it granted. + assert isinstance(body["expires_at"], str) + assert body["expires_at"] > body["created_at"] diff --git a/schemas/v0.1/subscription.request.json b/schemas/v0.1/subscription.request.json index 2189e22..ed42139 100644 --- a/schemas/v0.1/subscription.request.json +++ b/schemas/v0.1/subscription.request.json @@ -1,126 +1,135 @@ { - "$schema": "http://json-schema.org/draft-07/schema#", - "$id": "https://eep.dev/schemas/v0.1/subscription.request.json", - "title": "EEP Subscription Request", - "description": "Schema for creating a new EEP event subscription. Validates the POST body sent to /eep/subscribe.", - "type": "object", - "required": [ - "source_did", - "event_types", - "delivery_method" - ], - "additionalProperties": false, - "properties": { - "source_did": { - "type": "string", - "description": "The DID or URI of the entity to subscribe to. Must be a valid DID (e.g., 'did:web:example.com:u:acme-corp') or an entity URI.", - "pattern": "^(did:[a-z]+:[a-zA-Z0-9._%-]+(:[a-zA-Z0-9._%-]+)*|https?://[^\\s]+)$", - "examples": [ - "did:web:example.com:u:acme-corp", - "https://api.example.com/entities/acme-corp" - ] - }, - "event_types": { - "type": "array", - "description": "List of event type patterns to subscribe to. Supports wildcard suffix (e.g., 'com.example.entity.*'). An empty array is not allowed.", - "minItems": 1, - "maxItems": 50, - "items": { - "type": "string", - "pattern": "^[a-z][a-z0-9]*(\\.[a-z][a-z0-9]*)*(\\.(\\*|[a-z][a-z0-9]*))?$", - "examples": [ - "com.example.entity.updated", - "com.example.trust.*", - "com.example.*" - ] - } - }, - "delivery_method": { - "type": "string", - "description": "How events should be delivered to the subscriber.", - "enum": [ - "webhook", - "sse" - ], - "examples": [ - "webhook" - ] - }, - "delivery_url": { - "type": "string", - "description": "The HTTPS URL where webhook events will be POSTed. Required when delivery_method is 'webhook'. Must be a publicly accessible HTTPS endpoint.", - "format": "uri", - "pattern": "^https://", - "examples": [ - "https://agent.example.com/hooks/eep" - ] - }, - "delivery_format": { - "type": "string", - "description": "The event envelope format for delivery. Defaults to CloudEvents v1.0.", - "enum": [ - "cloudevents/v1.0" - ], - "default": "cloudevents/v1.0" - }, - "metadata": { - "type": "object", - "description": "Optional subscriber-defined metadata attached to this subscription for internal tracking.", - "additionalProperties": { - "type": "string" - }, - "maxProperties": 10, - "examples": [ - { - "description": "Monitor Acme Corp for trust changes", - "agent_id": "agent-42" - } - ] - }, - "tier": { - "type": "string", - "description": "Optional. The access tier being requested. If the entity has gate configuration, this specifies which tier the subscriber wants. When omitted, the entity's default_tier is used.", - "pattern": "^[a-z][a-z0-9_]{0,31}$", - "examples": [ - "pro", - "academic", - "verified_agents" - ] - }, - "gate_proofs": { - "type": "array", - "description": "Optional. Array of proof objects that satisfy the tier's requirements. See gate.proof.json for the full proof schema. Only needed when subscribing to a gated tier.", - "items": { - "type": "object", - "required": [ - "type" - ], - "properties": { - "type": { - "type": "string", - "description": "Proof type matching a requirement type." - } - }, - "additionalProperties": true - }, - "maxItems": 10 - } + "$schema": "http://json-schema.org/draft-07/schema#", + "$id": "https://eep.dev/schemas/v0.1/subscription.request.json", + "title": "EEP Subscription Request", + "description": "Schema for creating a new EEP event subscription. Validates the POST body sent to /eep/subscribe.", + "type": "object", + "required": [ + "source_did", + "event_types", + "delivery_method" + ], + "additionalProperties": false, + "properties": { + "source_did": { + "type": "string", + "description": "The DID or URI of the entity to subscribe to. Must be a valid DID (e.g., 'did:web:example.com:u:acme-corp') or an entity URI.", + "pattern": "^(did:[a-z]+:[a-zA-Z0-9._%-]+(:[a-zA-Z0-9._%-]+)*|https?://[^\\s]+)$", + "examples": [ + "did:web:example.com:u:acme-corp", + "https://api.example.com/entities/acme-corp" + ] }, - "if": { - "properties": { - "delivery_method": { - "const": "webhook" - } + "event_types": { + "type": "array", + "description": "List of event type patterns to subscribe to. Supports wildcard suffix (e.g., 'com.example.entity.*'). An empty array is not allowed.", + "minItems": 1, + "maxItems": 50, + "items": { + "type": "string", + "pattern": "^[a-z][a-z0-9]*(\\.[a-z][a-z0-9]*)*(\\.(\\*|[a-z][a-z0-9]*))?$", + "examples": [ + "com.example.entity.updated", + "com.example.trust.*", + "com.example.*" + ] + } + }, + "delivery_method": { + "type": "string", + "description": "How events should be delivered to the subscriber.", + "enum": [ + "webhook", + "sse" + ], + "examples": [ + "webhook" + ] + }, + "delivery_url": { + "type": "string", + "description": "The HTTPS URL where webhook events will be POSTed. Required when delivery_method is 'webhook'. Must be a publicly accessible HTTPS endpoint.", + "format": "uri", + "pattern": "^https://", + "examples": [ + "https://agent.example.com/hooks/eep" + ] + }, + "delivery_format": { + "type": "string", + "description": "The event envelope format for delivery. Defaults to CloudEvents v1.0.", + "enum": [ + "cloudevents/v1.0" + ], + "default": "cloudevents/v1.0" + }, + "lease_seconds": { + "type": "integer", + "description": "Optional. Requested subscription lifetime in seconds, starting from successful intent verification (SPECIFICATION.md \u00a710.2). The publisher MAY clamp this to its own policy and reports the value actually granted as `expires_at` on the subscription. Omit to accept the publisher's default lease.", + "minimum": 300, + "maximum": 31536000, + "examples": [ + 2592000 + ] + }, + "metadata": { + "type": "object", + "description": "Optional subscriber-defined metadata attached to this subscription for internal tracking.", + "additionalProperties": { + "type": "string" + }, + "maxProperties": 10, + "examples": [ + { + "description": "Monitor Acme Corp for trust changes", + "agent_id": "agent-42" } + ] }, - "then": { + "tier": { + "type": "string", + "description": "Optional. The access tier being requested. If the entity has gate configuration, this specifies which tier the subscriber wants. When omitted, the entity's default_tier is used.", + "pattern": "^[a-z][a-z0-9_]{0,31}$", + "examples": [ + "pro", + "academic", + "verified_agents" + ] + }, + "gate_proofs": { + "type": "array", + "description": "Optional. Array of proof objects that satisfy the tier's requirements. See gate.proof.json for the full proof schema. Only needed when subscribing to a gated tier.", + "items": { + "type": "object", "required": [ - "delivery_url" + "type" ], "properties": { - "delivery_url": { - "pattern": "^https://" - } - } + "type": { + "type": "string", + "description": "Proof type matching a requirement type." + } + }, + "additionalProperties": true + }, + "maxItems": 10 + } + }, + "if": { + "properties": { + "delivery_method": { + "const": "webhook" + } + } + }, + "then": { + "required": [ + "delivery_url" + ], + "properties": { + "delivery_url": { + "pattern": "^https://" + } } -} \ No newline at end of file + } +} diff --git a/tests/conformance-fixtures/manifest.json b/tests/conformance-fixtures/manifest.json index a37df9d..7ca3f6d 100644 --- a/tests/conformance-fixtures/manifest.json +++ b/tests/conformance-fixtures/manifest.json @@ -168,6 +168,28 @@ "shape": "json-pair", "asserts_valid": false }, + { + "id": "subscription-lease-request", + "category": "subscription", + "tier": "Standard", + "spec_section": "\u00a710.2 Lease lifetime and renewal", + "schema": "schemas/v0.1/subscription.request.json", + "input": "subscription/lease-request.input.json", + "expected": "subscription/lease-request.expected.json", + "shape": "json-pair", + "asserts_valid": true + }, + { + "id": "subscription-lease-below-minimum", + "category": "subscription", + "tier": "Standard", + "spec_section": "\u00a710.2 Lease lifetime and renewal", + "schema": "schemas/v0.1/subscription.request.json", + "input": "subscription/lease-below-minimum.input.json", + "expected": "subscription/lease-below-minimum.expected.json", + "shape": "json-pair", + "asserts_valid": false + }, { "id": "discovery-crosswalk-host-bundle", "category": "discovery", diff --git a/tests/conformance-fixtures/subscription/lease-below-minimum.expected.json b/tests/conformance-fixtures/subscription/lease-below-minimum.expected.json new file mode 100644 index 0000000..26de2b2 --- /dev/null +++ b/tests/conformance-fixtures/subscription/lease-below-minimum.expected.json @@ -0,0 +1,4 @@ +{ + "valid": false, + "reason": "lease_seconds below the schema minimum of 300" +} diff --git a/tests/conformance-fixtures/subscription/lease-below-minimum.input.json b/tests/conformance-fixtures/subscription/lease-below-minimum.input.json new file mode 100644 index 0000000..84531b8 --- /dev/null +++ b/tests/conformance-fixtures/subscription/lease-below-minimum.input.json @@ -0,0 +1,9 @@ +{ + "source_did": "did:web:test.eep.dev:u:alice", + "event_types": [ + "com.example.entity.updated" + ], + "delivery_method": "webhook", + "delivery_url": "https://agent.example.com/hooks/eep", + "lease_seconds": 30 +} diff --git a/tests/conformance-fixtures/subscription/lease-request.expected.json b/tests/conformance-fixtures/subscription/lease-request.expected.json new file mode 100644 index 0000000..127088f --- /dev/null +++ b/tests/conformance-fixtures/subscription/lease-request.expected.json @@ -0,0 +1,4 @@ +{ + "valid": true, + "reason": "lease_seconds is an optional integer within publisher policy bounds" +} diff --git a/tests/conformance-fixtures/subscription/lease-request.input.json b/tests/conformance-fixtures/subscription/lease-request.input.json new file mode 100644 index 0000000..42cab00 --- /dev/null +++ b/tests/conformance-fixtures/subscription/lease-request.input.json @@ -0,0 +1,9 @@ +{ + "source_did": "did:web:test.eep.dev:u:alice", + "event_types": [ + "com.example.entity.updated" + ], + "delivery_method": "webhook", + "delivery_url": "https://agent.example.com/hooks/eep", + "lease_seconds": 2592000 +} diff --git a/tests/types/eep-schemas.d.ts b/tests/types/eep-schemas.d.ts index 0601263..67b36cb 100644 --- a/tests/types/eep-schemas.d.ts +++ b/tests/types/eep-schemas.d.ts @@ -2555,6 +2555,10 @@ export interface EEPSubscriptionRequest { * The event envelope format for delivery. Defaults to CloudEvents v1.0. */ delivery_format?: 'cloudevents/v1.0'; + /** + * Optional. Requested subscription lifetime in seconds, starting from successful intent verification (SPECIFICATION.md §10.2). The publisher MAY clamp this to its own policy and reports the value actually granted as `expires_at` on the subscription. Omit to accept the publisher's default lease. + */ + lease_seconds?: number; /** * Optional subscriber-defined metadata attached to this subscription for internal tracking. */