diff --git a/docs/dot-permanent-subscriptions.md b/docs/dot-permanent-subscriptions.md new file mode 100644 index 0000000..5abc2ac --- /dev/null +++ b/docs/dot-permanent-subscriptions.md @@ -0,0 +1,15 @@ +# Permanent MCP Events subscriptions + +Following https://developers.openai.com/apps-sdk/build/mcp-events, an explicit JSON `ttlMs: null` grants a subscription without time expiration and returns `refreshBefore: null`. Omitting the field retains the one-hour default. Numeric durations retain the existing accepted range of 60 seconds through 24 hours; out-of-range values are rejected. A numeric refresh of a permanent subscription grants a finite lease. + +Encrypted event records and bridge subscription proofs carry an explicit `Permanent` flag with zero expiry. Zero expiry alone never grants permanence: existing invalidation and legacy records remain inactive. Public status reports an active permanent subscription with `renewal_state: current` and omitted `expires_at` and `refresh_before`. The HTTP subscribe result always includes `refreshBefore`, including JSON null. + +Permanence changes only time expiration. Delivery still requires enabled state and callback verification. Unsubscribe, revocation, disable/re-enable, authentication continuity changes and signing-secret replacement retain existing generation and trust-epoch invalidation rules. Event and bridge grants are intersected: a permanent event-only renewal cannot extend a finite bridge grant, and a finite event grant bounds a permanent bridge grant. + +A subscription does not grant answer/task capabilities by itself. Each mode still requires completion of its nonce proof. Once proven, continuous proof remains valid within the same binding, auth/pairing identity, consumer and trust epoch; permanent leases therefore need no manual ten-minute proof churn. Restart preserves this evidence. Invalidating continuity requires a fresh proof, even if a new permanent subscription uses the same deterministic ID. + +Callback challenge caching is separate from continuous mode proof. It remains bounded to five minutes and any finite source/target grant. Permanence does not relax HTTPS/SSRF checks, signed challenges, delivery verification, maxActiveSubs, delivery-record limits or encrypted-state size bounds. Existing finite records with null request parameters are not retrospectively upgraded on restart; a legacy bridge proof without a trust epoch remains within the existing epoch check. + +Magpie supports nullable permanent subscription dates at ad28f447 (PR #3), so its status card handles omitted `expires_at` and `refresh_before` values without zero dates. + +Persisted grants fail closed on load. A permanent event record requires zero expiry and retained `ttlMs: null`; finite event records require a non-zero expiry. A permanent bridge proof requires zero expiry and complete trust evidence. Legacy nullable records without the permanent flag remain finite. Invalid records make the encrypted state load as `corrupt` (events) or `bridge_state_corrupt` (bridge) instead of becoming active. diff --git a/internal/dotbridge/authorization.go b/internal/dotbridge/authorization.go index 0520c01..9caffc1 100644 --- a/internal/dotbridge/authorization.go +++ b/internal/dotbridge/authorization.go @@ -10,6 +10,7 @@ import ( ) type subscriptionProof struct { + Permanent bool BindingID string TrustEpoch string AuthHash string @@ -70,9 +71,7 @@ func (s *Service) activeSubscription(queue string) (subscriptionProof, bool) { for _, b := range bindings { if b.ID == sub.BindingID && b.Epoch == sub.TrustEpoch { found = true - if b.Expires.Before(sub.Expires) { - sub.Expires = b.Expires - } + sub = intersectLease(sub, b) } } if !found { @@ -80,9 +79,42 @@ func (s *Service) activeSubscription(queue string) (subscriptionProof, bool) { } } } - return sub, ok && time.Now().Before(sub.Expires) + return sub, ok && sub.leaseActive(time.Now()) } func consumerWorker(consumer, worker string) string { h := sha256.Sum256([]byte(consumer + "\x00" + worker)) return "consumer." + hex.EncodeToString(h[:]) } + +// A trusted record must carry complete binding evidence. Permanence is accepted +// only with zero expiry and the same continuity evidence as any trust epoch. +func completeSubscriptionProof(s subscriptionProof) bool { + if s.Permanent && (!s.Expires.IsZero() || s.TrustEpoch == "") { + return false + } + return s.TrustEpoch == "" || hexBindingID(s.BindingID) && s.AuthHash != "" && s.Consumer != "" +} + +// hexBindingID matches the subscription ids the event store derives, so a +// syntactically valid but non-hexadecimal id cannot stand in for one. +func hexBindingID(id string) bool { + if len(id) != 64 { + return false + } + _, err := hex.DecodeString(id) + return err == nil +} + +// A permanent grant still needs an explicit flag: zero expiry alone also means revoked. +func (s subscriptionProof) leaseActive(now time.Time) bool { + return s.Permanent || now.Before(s.Expires) +} + +// Both stores must authorize the lease; an event-only refresh cannot extend it. +func intersectLease(s subscriptionProof, b dotevents.TrustedBinding) subscriptionProof { + if !b.Permanent && (s.Permanent || b.Expires.Before(s.Expires)) { + s.Permanent = false + s.Expires = b.Expires + } + return s +} diff --git a/internal/dotbridge/permanent_load_test.go b/internal/dotbridge/permanent_load_test.go new file mode 100644 index 0000000..40cc905 --- /dev/null +++ b/internal/dotbridge/permanent_load_test.go @@ -0,0 +1,131 @@ +package dotbridge + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/dotcontrol/dotcontrol/internal/dotqueue" +) + +func rewriteBridgeSubscription(t *testing.T, s *Service, id string, change func(*subscriptionProof)) { + t.Helper() + data, err := securelyReadBridgeState(s.root) + if err != nil { + t.Fatal(err) + } + var st PersistedState + if err := json.Unmarshal(data, &st); err != nil { + t.Fatal(err) + } + sub := st.Subscriptions[id] + change(&sub) + st.Subscriptions[id] = sub + if err := securelyWriteBridgeState(s.root, mustMarshalJSON(st)); err != nil { + t.Fatal(err) + } +} + +func mustMarshalJSON(v any) []byte { + b, err := json.Marshal(v) + if err != nil { + panic(err) + } + return b +} + +func TestPermanentBridgeProofLoadFailsClosed(t *testing.T) { + cases := []struct { + name string + change func(*subscriptionProof) + }{ + {"finite expiry", func(s *subscriptionProof) { s.Expires = time.Now().Add(time.Hour) }}, + {"past expiry", func(s *subscriptionProof) { s.Expires = time.Now().Add(-time.Hour) }}, + {"empty trust epoch", func(s *subscriptionProof) { s.TrustEpoch = "" }}, + {"short binding", func(s *subscriptionProof) { s.BindingID = "short" }}, + {"non-hex binding", func(s *subscriptionProof) { s.BindingID = strings.Repeat("z", 64) }}, + {"empty auth hash", func(s *subscriptionProof) { s.AuthHash = "" }}, + {"empty consumer", func(s *subscriptionProof) { s.Consumer = "" }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s, _, id := openTestBridge(t) + p := testEventsParams(id) + p.TTL = json.RawMessage(`null`) + if _, err := s.EventsSubscribe(context.Background(), id, p); err != nil { + t.Fatal(err) + } + rewriteBridgeSubscription(t, s, id, func(sub *subscriptionProof) { + if !sub.Permanent { + t.Fatalf("expected permanent bridge proof: %#v", sub) + } + tc.change(sub) + }) + // Load the encrypted file through a fresh instance so no valid proof + // retained in the running service can mask the on-disk record. + + fresh := &Service{root: s.root, state: PersistedState{Version: APIVersion, Clients: map[string]ClientPolicy{}}} + err := fresh.loadState() + if err == nil { + t.Fatalf("inconsistent permanent bridge proof loaded: %#v", fresh.subscriptions[id]) + } + if err.Error() != "bridge_state_corrupt" { + t.Fatalf("load error = %v", err) + } + }) + } +} + +func TestLegacyBridgeProofWithinEpochStillLoads(t *testing.T) { + s, _, id := openTestBridge(t) + s.mu.Lock() + old := s.subscriptions[id] + old.TrustEpoch = "" + s.subscriptions[id] = old + err := s.saveState() + s.mu.Unlock() + if err != nil { + t.Fatal(err) + } + fresh := &Service{root: s.root, state: PersistedState{Version: APIVersion, Clients: map[string]ClientPolicy{}}} + if err := fresh.loadState(); err != nil { + t.Fatalf("legacy bridge proof rejected: %v", err) + } +} + +func TestPermanentBridgeProofLoadPreservesValidGrant(t *testing.T) { + s, _, id := openTestBridge(t) + p := testEventsParams(id) + p.TTL = json.RawMessage(`null`) + if _, err := s.EventsSubscribe(context.Background(), id, p); err != nil { + t.Fatal(err) + } + proveSustainable(t, s, id) + // Reproduce the persisted device proof from the pending probe. This isolates + // permanent bridge/event resume from the fixture's callback host rebinding. + s.mu.Lock() + policy := s.state.Clients[id] + proof := policy.proof(dotqueue.ModeAnswer) + proof.Verified, proof.Consumed = time.Now().Add(-time.Minute), true + policy.setProof(dotqueue.ModeAnswer, proof) + s.state.Clients[id] = policy + err := s.saveState() + s.mu.Unlock() + if err != nil { + t.Fatal(err) + } + restored, err := Open(context.Background(), Config{Root: s.root, Listen: s.listen, Vault: s.vault, MCPQueue: s.mcpQueue}) + if err != nil { + t.Fatal(err) + } + status, err := restored.StatusForClient(id) + if err != nil || !status.Ready || status.Subscription.State != "active" || status.Subscription.ExpiresAt != nil || status.Subscription.RefreshBefore != nil { + restored.Close() + t.Fatalf("valid permanent grant lost: %#v %v", status, err) + } + if err := restored.Close(); err != nil { + t.Fatal(err) + } +} diff --git a/internal/dotbridge/permanent_test.go b/internal/dotbridge/permanent_test.go new file mode 100644 index 0000000..354a74f --- /dev/null +++ b/internal/dotbridge/permanent_test.go @@ -0,0 +1,140 @@ +package dotbridge + +import ( + "context" + "encoding/json" + "github.com/dotcontrol/dotcontrol/internal/dotevents" + "strings" + "testing" + "time" +) + +func TestPermanentProofStatusAndRestart(t *testing.T) { + s, _, id := openTestBridge(t) + p := testEventsParams(id) + p.TTL = json.RawMessage(`null`) + if _, err := s.EventsSubscribe(context.Background(), id, p); err != nil { + t.Fatal(err) + } + status, _ := s.StatusForClient(id) + if status.Ready || status.Subscription.State != "active" || status.Subscription.ExpiresAt != nil || status.Subscription.RefreshBefore != nil || status.Subscription.RenewalState != "current" { + t.Fatalf("unproven permanent status: %#v", status) + } + proveSustainable(t, s, id) + s.mu.Lock() + policy := s.state.Clients[id] + policy.ProbeVerified = time.Now().Add(-24 * time.Hour) + s.state.Clients[id] = policy + err := s.saveState() + s.mu.Unlock() + if err != nil { + t.Fatal(err) + } + if err = s.Close(); err != nil { + t.Fatal(err) + } + reopened, err := Open(context.Background(), Config{Root: s.root, Listen: s.listen, Vault: s.vault, MCPQueue: s.mcpQueue}) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + status, err = reopened.StatusForClient(id) + if err != nil || !status.Ready || status.ExpiresAt != nil { + t.Fatalf("reopened permanent proof: %#v %v", status, err) + } + b, _ := json.Marshal(status) + if strings.Contains(string(b), "0001-01-01") { + t.Fatalf("zero date: %s", b) + } + if err = reopened.EventsUnsubscribe(context.Background(), id, p); err != nil { + t.Fatal(err) + } + status, _ = reopened.StatusForClient(id) + if status.Ready || status.Subscription.State == "active" { + t.Fatal("unsubscribe resurrected permanent lease") + } +} + +func TestPermanentInvalidationRequiresNewProof(t *testing.T) { + for _, action := range []string{"auth", "secret", "revoke", "disable", "unsubscribe"} { + t.Run(action, func(t *testing.T) { + s, _, id := openTestBridge(t) + p := testEventsParams(id) + p.TTL = json.RawMessage(`null`) + if _, err := s.EventsSubscribe(context.Background(), id, p); err != nil { + t.Fatal(err) + } + proveSustainable(t, s, id) + switch action { + case "auth": + s.mu.Lock() + policy := s.state.Clients[id] + policy.TokenHash = hashToken("new-auth") + s.state.Clients[id] = policy + s.mu.Unlock() + case "secret": + p.Delivery.Secret = "whsec_MTIzNDU2Nzg5MDEyMzQ1Njc4OTAxMjM0NTY3ODkwMTI=" + case "revoke": + if err := s.RevokeClient(id); err != nil { + t.Fatal(err) + } + case "disable": + if err := s.Disable(); err != nil { + t.Fatal(err) + } + if err := s.Enable(); err != nil { + t.Fatal(err) + } + case "unsubscribe": + if err := s.EventsUnsubscribe(context.Background(), id, p); err != nil { + t.Fatal(err) + } + } + if action != "revoke" { + if _, err := s.EventsSubscribe(context.Background(), id, p); err != nil { + t.Fatal(err) + } + } + status, _ := s.StatusForClient(id) + if status.Ready { + t.Fatal("invalidated proof revived") + } + }) + } +} + +func TestPermanentAndFiniteGrantsIntersect(t *testing.T) { + now := time.Now() + finite := now.Add(time.Hour) + cases := []struct { + name string + bridge subscriptionProof + event dotevents.TrustedBinding + permanent bool + }{ + {"both permanent", subscriptionProof{Permanent: true}, dotevents.TrustedBinding{Permanent: true}, true}, + {"bridge finite", subscriptionProof{Expires: finite}, dotevents.TrustedBinding{Permanent: true}, false}, + {"event finite", subscriptionProof{Permanent: true}, dotevents.TrustedBinding{Expires: finite}, false}, + {"revoked bridge", subscriptionProof{}, dotevents.TrustedBinding{Permanent: true}, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + grant := intersectLease(tc.bridge, tc.event) + if grant.Permanent != tc.permanent { + t.Fatal("permanence changed") + } + if tc.name == "revoked bridge" { + if grant.leaseActive(now) { + t.Fatal("revoked bridge revived") + } + return + } + if !grant.leaseActive(now) { + t.Fatal("active lease lost") + } + if !tc.permanent && (!grant.Expires.Equal(finite) || grant.leaseActive(finite)) { + t.Fatal("finite bound lost") + } + }) + } +} diff --git a/internal/dotbridge/probe.go b/internal/dotbridge/probe.go index ca7d5b6..9977c79 100644 --- a/internal/dotbridge/probe.go +++ b/internal/dotbridge/probe.go @@ -78,7 +78,7 @@ func (s *Service) StartProbeMode(clientID, mode string) (ProbeResult, error) { return ProbeResult{}, errors.New("bridge_paused") } current := s.subscriptions[policy.QueueID] - if !s.state.Enabled || !time.Now().Before(current.Expires) || current.Generation != subscription.Generation || !current.Expires.Equal(subscription.Expires) { + if !s.state.Enabled || !current.leaseActive(time.Now()) || current.Generation != subscription.Generation || (current.Permanent != subscription.Permanent || !current.Expires.Equal(subscription.Expires)) { return ProbeResult{}, errors.New("subscription_changed") } policy.setProof(mode, ProbeProof{TrustEpoch: subscription.TrustEpoch, JobID: job.ID, AnswerHash: hashToken(answer), Generation: subscription.Generation, SubscriptionExpires: subscription.Expires}) diff --git a/internal/dotbridge/service.go b/internal/dotbridge/service.go index 750c845..874f48d 100644 --- a/internal/dotbridge/service.go +++ b/internal/dotbridge/service.go @@ -256,24 +256,25 @@ func (s *Service) statusDTO(policy ClientPolicy) StatusDTO { current, exists := s.state.Clients[policy.ClientID] sub := s.subscriptions[current.QueueID] now := time.Now() - active := now.Before(sub.Expires) + active := sub.leaseActive(now) if sub.TrustEpoch != "" { active = false for _, b := range eventStatus.Bindings { - if b.ID == sub.BindingID && b.Epoch == sub.TrustEpoch && sub.AuthHash == current.TokenHash && sub.Consumer == s.mcpQueue && now.Before(b.Expires) { - if b.Expires.Before(sub.Expires) { - sub.Expires = b.Expires - } + if b.ID == sub.BindingID && b.Epoch == sub.TrustEpoch && sub.AuthHash == current.TokenHash && sub.Consumer == s.mcpQueue && (b.Permanent || now.Before(b.Expires)) { + sub = intersectLease(sub, b) sub.Refreshed = b.Refreshed // Both stores must authorize the effective lease. An event-only // renewal cannot extend the bridge's last committed grant. - active = now.Before(sub.Expires) + active = sub.leaseActive(now) } } } out := StatusDTO{Version: APIVersion, Enabled: s.state.Enabled, Paired: exists && !current.Revoked, Code: "probe_required", Subscription: SubscriptionDTO{State: "inactive"}} if active { - out.Subscription = SubscriptionDTO{State: "active", ExpiresAt: &sub.Expires, RefreshBefore: &sub.Expires, LastRefreshedAt: &sub.Refreshed, RenewalState: "current"} + out.Subscription = SubscriptionDTO{State: "active", LastRefreshedAt: &sub.Refreshed, RenewalState: "current"} + if !sub.Permanent { + out.Subscription.ExpiresAt, out.Subscription.RefreshBefore = &sub.Expires, &sub.Expires + } } else if sub.Generation == 0 && sub.Expires.IsZero() && e == nil && gateOK && exists && !current.Revoked && current.TokenHash == policy.TokenHash && s.capacityAllowedLocked() { // Old binaries did not persist the bridge subscription map. Report the // actual verified legacy lease without adopting it as authorization. @@ -326,10 +327,10 @@ func (s *Service) statusDTO(policy ClientPolicy) StatusDTO { if trusted { expiry = sub.Expires } - if sub.Expires.Before(expiry) { + if !sub.Permanent && (expiry.IsZero() || sub.Expires.Before(expiry)) { expiry = sub.Expires } - if out.ExpiresAt == nil || expiry.Before(*out.ExpiresAt) { + if !expiry.IsZero() && (out.ExpiresAt == nil || expiry.Before(*out.ExpiresAt)) { out.ExpiresAt = &expiry } verified := proof.Verified @@ -452,7 +453,7 @@ func (s *Service) EventsSubscribe(ctx context.Context, principal string, params bindings, readErr := reader.TrustedBindings(queue) matched := false for _, b := range bindings { - if b.ID == sub.ID && b.Epoch == sub.TrustEpoch && b.Expires.Equal(sub.ExpiresAt) { + if b.ID == sub.ID && b.Epoch == sub.TrustEpoch && b.Permanent == sub.Permanent && b.Expires.Equal(sub.ExpiresAt) { matched = true } } @@ -465,7 +466,7 @@ func (s *Service) EventsSubscribe(ctx context.Context, principal string, params if sub.TrustEpoch != "" && sub.TrustEpoch == old.TrustEpoch && old.BindingID == sub.ID && old.AuthHash == p.TokenHash && old.Consumer == s.mcpQueue { generation = old.Generation } - s.subscriptions[queue] = subscriptionProof{Expires: sub.ExpiresAt, Generation: generation, BindingID: sub.ID, TrustEpoch: sub.TrustEpoch, AuthHash: p.TokenHash, Consumer: s.mcpQueue, Refreshed: time.Now().UTC()} + s.subscriptions[queue] = subscriptionProof{Permanent: sub.Permanent, Expires: sub.ExpiresAt, Generation: generation, BindingID: sub.ID, TrustEpoch: sub.TrustEpoch, AuthHash: p.TokenHash, Consumer: s.mcpQueue, Refreshed: time.Now().UTC()} if generation != old.Generation { p.Capabilities = CapPolicy{} p.ProbeExpires = time.Time{} @@ -483,6 +484,7 @@ func (s *Service) EventsUnsubscribe(ctx context.Context, principal string, param s.mu.Lock() old := s.subscriptions[queue] old.Expires = time.Time{} + old.Permanent = false old.Generation++ s.subscriptions[queue] = old if saveErr := s.saveState(); e == nil { @@ -533,10 +535,7 @@ func (s *Service) loadState() error { if st.Subscriptions != nil { for queue, sub := range st.Subscriptions { policy, ok := st.Clients[queue] - if !ok || policy.QueueID != queue || sub.Generation == 0 { - return errors.New("bridge_state_corrupt") - } - if sub.TrustEpoch != "" && (len(sub.BindingID) != 64 || sub.AuthHash == "" || sub.Consumer == "") { + if !ok || policy.QueueID != queue || sub.Generation == 0 || !completeSubscriptionProof(sub) { return errors.New("bridge_state_corrupt") } } diff --git a/internal/dotbridge/worker.go b/internal/dotbridge/worker.go index 60a6033..85c66c6 100644 --- a/internal/dotbridge/worker.go +++ b/internal/dotbridge/worker.go @@ -140,7 +140,7 @@ func (s *Service) completeTask(jobID, token string, result dotqueue.Result) (dot s.mu.Lock() defer s.mu.Unlock() sub := s.subscriptions[b.QueueID] - active := time.Now().Before(sub.Expires) + active := sub.leaseActive(time.Now()) policy := s.state.Clients[b.QueueID] if policy.Revoked || (s.revoked != nil && s.revoked()) { return dotqueue.Job{}, errors.New("access_revoked") @@ -158,7 +158,7 @@ func (s *Service) completeTask(jobID, token string, result dotqueue.Result) (dot } proof.Consumed = true proof.SubscriptionExpires = sub.Expires - if sub.Expires.Before(proof.Expires) { + if !sub.Permanent && sub.Expires.Before(proof.Expires) { proof.Expires = sub.Expires } policy.setProof(job.Mode, proof) diff --git a/internal/dotevents/lifecycle_test.go b/internal/dotevents/lifecycle_test.go index ce30dee..e3f2d29 100644 --- a/internal/dotevents/lifecycle_test.go +++ b/internal/dotevents/lifecycle_test.go @@ -173,13 +173,13 @@ func TestUnsubscribeExpiryEncryptionAndSignatures(t *testing.T) { }) defer close() p := testParams("alice") - p.TTL = json.RawMessage(`null`) + p.TTL = nil sub, e := m.Subscribe(context.Background(), "alice", p) if e != nil { t.Fatal(e) } if time.Until(sub.ExpiresAt) > time.Hour { - t.Fatal("unbounded null ttl") + t.Fatal("unbounded default ttl") } if e = m.Deliver(context.Background(), "alice", "alice", Notice{EventID: "evt.signed", QueueID: "alice", JobID: "job.signed", Mode: "task", OccurredAt: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)}); e != nil { t.Fatal(e) diff --git a/internal/dotevents/manager.go b/internal/dotevents/manager.go index 2d32a01..1a87d57 100644 --- a/internal/dotevents/manager.go +++ b/internal/dotevents/manager.go @@ -75,10 +75,13 @@ func (m *Manager) subscribe(ctx context.Context, principal string, p Params, clo // Eligibility and the new grant start at the serialized mutation, not // request arrival: acquiring the store lock may outlive the old lease. now := clock().UTC() - rec.ExpiresAt = now.Add(time.Duration(ttl) * time.Millisecond) + rec.Permanent = ttl == 0 + if !rec.Permanent { + rec.ExpiresAt = now.Add(time.Duration(ttl) * time.Millisecond) + } rec.CreatedAt, rec.UpdatedAt = now, now for k, s := range st.Subscriptions { - if !s.ExpiresAt.After(now) { + if !s.leaseActive(now) { delete(st.Subscriptions, k) } } @@ -90,7 +93,7 @@ func (m *Manager) subscribe(ctx context.Context, principal string, p Params, clo // Cache is scoped to authenticated principal plus exact URL and bounded // by both verification lifetime and subscription expiry. for _, s := range st.Subscriptions { - if s.Principal == principal && s.Params.Delivery.URL == p.Delivery.URL && s.CallbackVerified && s.VerifiedUntil.After(now) && s.ExpiresAt.After(now) { + if s.Principal == principal && s.Params.Delivery.URL == p.Delivery.URL && s.CallbackVerified && s.VerifiedUntil.After(now) && s.leaseActive(now) { cached = true rec.CallbackVerified = true rec.VerifiedUntil = boundedVerificationDeadline(s.VerifiedUntil, s.ExpiresAt, rec.ExpiresAt, now) @@ -127,12 +130,12 @@ func (m *Manager) subscribe(ctx context.Context, principal string, p Params, clo err := m.store.withLock(false, true, func(st *persistenceState) error { now := clock().UTC() s := st.Subscriptions[id] - if !st.Enabled || s == nil || s.Generation != gen || !s.CallbackVerified || !s.ExpiresAt.After(now) { + if !st.Enabled || s == nil || s.Generation != gen || !s.CallbackVerified || !s.leaseActive(now) { return errf("stale") } // The cached path also reacquires the lock after persistence. A // delayed finalization cannot carry trust past the source lease. - if previous != nil && !previous.ExpiresAt.After(now) { + if previous != nil && !previous.leaseActive(now) { s.TrustEpoch, rec.TrustEpoch = gen, gen } return nil @@ -140,7 +143,7 @@ func (m *Manager) subscribe(ctx context.Context, principal string, p Params, clo if err != nil { return Subscription{}, err } - return Subscription{TrustEpoch: rec.TrustEpoch, ID: id, ExpiresAt: rec.ExpiresAt, CallbackVerified: true}, nil + return Subscription{Permanent: rec.Permanent, TrustEpoch: rec.TrustEpoch, ID: id, ExpiresAt: rec.ExpiresAt, CallbackVerified: true}, nil } if e = m.delivery.verify(ctx, rec); e != nil { // Restore a previously accepted refresh only if no revoke/refresh won @@ -162,7 +165,7 @@ func (m *Manager) subscribe(ctx context.Context, principal string, p Params, clo // original callback failure unless authorization was revoked. return nil } - if previous != nil && previous.CallbackVerified && previous.ExpiresAt.After(clock()) { + if previous != nil && previous.CallbackVerified && previous.leaseActive(clock()) { st.Subscriptions[id] = previous } return nil @@ -178,10 +181,10 @@ func (m *Manager) subscribe(ctx context.Context, principal string, p Params, clo } now := clock().UTC() s := st.Subscriptions[id] - if s == nil || s.Generation != gen || !s.ExpiresAt.After(now) { + if s == nil || s.Generation != gen || !s.leaseActive(now) { return errf("stale") } - if previous != nil && !previous.ExpiresAt.After(now) { + if previous != nil && !previous.leaseActive(now) { s.TrustEpoch = gen rec.TrustEpoch = gen } @@ -196,7 +199,7 @@ func (m *Manager) subscribe(ctx context.Context, principal string, p Params, clo if e != nil { return Subscription{}, e } - return Subscription{TrustEpoch: rec.TrustEpoch, ID: id, ExpiresAt: rec.ExpiresAt, CallbackVerified: true}, nil + return Subscription{Permanent: rec.Permanent, TrustEpoch: rec.TrustEpoch, ID: id, ExpiresAt: rec.ExpiresAt, CallbackVerified: true}, nil } // diagnosticReason accepts only stable categories, never arbitrary error text. @@ -229,7 +232,7 @@ func verifyStage(code string) string { // each reuse is capped by both subscription grants and the current cache TTL. func boundedVerificationDeadline(until, sourceExpiry, expiry, now time.Time) time.Time { for _, bound := range []time.Time{sourceExpiry, expiry, now.Add(verifiedCacheTTL)} { - if bound.Before(until) { + if !bound.IsZero() && bound.Before(until) { until = bound } } @@ -296,7 +299,7 @@ func (m *Manager) Deliver(ctx context.Context, principal, queueID string, n Noti return errf("disabled") } for id, s := range st.Subscriptions { - if s.Principal == principal && s.CallbackVerified && s.ExpiresAt.After(time.Now()) { + if s.Principal == principal && s.CallbackVerified && s.leaseActive(time.Now()) { ids = append(ids, id) } } @@ -330,7 +333,7 @@ func (m *Manager) deliverOne(ctx context.Context, principal, q, id string, n Not return errf("disabled") } s := st.Subscriptions[id] - if s == nil || !s.CallbackVerified || !s.ExpiresAt.After(time.Now()) { + if s == nil || !s.CallbackVerified || !s.leaseActive(time.Now()) { return errf("stale") } r := st.Deliveries[key] @@ -385,7 +388,7 @@ func (m *Manager) deliverOne(ctx context.Context, principal, q, id string, n Not return errf("disabled") } s := st.Subscriptions[id] - if s == nil || s.Generation != sub.Generation || !s.CallbackVerified || !s.ExpiresAt.After(time.Now()) { + if s == nil || s.Generation != sub.Generation || !s.CallbackVerified || !s.leaseActive(time.Now()) { return errf("stale") } r := st.Deliveries[key] @@ -424,13 +427,13 @@ func (m *Manager) Status(principal, q string) (Status, error) { var verifiedUpdated time.Time if st.Enabled { for _, s := range st.Subscriptions { - if s.Principal == principal && s.ExpiresAt.After(time.Now()) { + if s.Principal == principal && s.leaseActive(time.Now()) { out.Subscribed = true out.CallbackVerified = out.CallbackVerified || s.CallbackVerified if s.CallbackVerified { // Legacy verified leases remain real grants, but an empty // epoch supplies no evidence of durable mode trust. - out.Bindings = append(out.Bindings, TrustedBinding{ID: s.ID, Epoch: s.TrustEpoch, Expires: s.ExpiresAt, Refreshed: s.UpdatedAt}) + out.Bindings = append(out.Bindings, TrustedBinding{Permanent: s.Permanent, ID: s.ID, Epoch: s.TrustEpoch, Expires: s.ExpiresAt, Refreshed: s.UpdatedAt}) if s.UpdatedAt.After(verifiedUpdated) { verifiedUpdated = s.UpdatedAt } diff --git a/internal/dotevents/permanent_load_test.go b/internal/dotevents/permanent_load_test.go new file mode 100644 index 0000000..2d69dba --- /dev/null +++ b/internal/dotevents/permanent_load_test.go @@ -0,0 +1,109 @@ +package dotevents + +import ( + "context" + "encoding/json" + "net/http" + "testing" + "time" +) + +// rewritePersistedSubscription changes one decrypted record and seals it again, +// exercising the same encrypted load path as a restart. +func rewritePersistedSubscription(t *testing.T, m *Manager, id string, change func(*subscriptionRecord)) { + t.Helper() + if err := m.store.withLock(false, false, func(st *persistenceState) error { + rec := st.Subscriptions[id] + if rec == nil { + t.Fatalf("subscription %s missing", id) + } + cp := *rec + change(&cp) + st.Subscriptions[id] = &cp + return m.store.saveState(st) + }); err != nil { + t.Fatal(err) + } +} + +func reopenMustFail(t *testing.T, dir string) { + t.Helper() + reopened, err := Open(dir, testKey, nil) + if err == nil { + status, _ := reopened.Status("alice", "alice") + t.Fatalf("inconsistent permanent state loaded: active=%v", status.Subscribed) + } + if code := asEventsError(err).Code; code != "corrupt" && code != "unknown" { + t.Fatalf("load error code = %q, want corrupt/unknown: %v", code, err) + } +} + +func TestPermanentRecordLoadRejectsInconsistentPersistence(t *testing.T) { + finite := json.RawMessage(`3600000`) + cases := []struct { + name string + ttl json.RawMessage + change func(*subscriptionRecord) + }{ + {"permanent numeric ttl", finite, func(r *subscriptionRecord) { r.ExpiresAt = time.Time{} }}, + {"permanent omitted ttl", nil, func(r *subscriptionRecord) { r.ExpiresAt = time.Time{} }}, + {"permanent future expiry", nil, func(r *subscriptionRecord) { r.ExpiresAt = time.Now().Add(time.Hour) }}, + {"permanent past expiry", nil, func(r *subscriptionRecord) { r.ExpiresAt = time.Now().Add(-time.Hour) }}, + {"finite zero expiry", nil, func(r *subscriptionRecord) { r.Permanent = false; r.ExpiresAt = time.Time{} }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + m, dir, cleanup, _ := withFixtureServerURL(t, func(w http.ResponseWriter, r *http.Request) { echoVerification(w, r) }) + defer cleanup() + p := testParams("alice") + p.TTL = json.RawMessage(`null`) + sub, err := m.Subscribe(context.Background(), "alice", p) + if err != nil { + t.Fatal(err) + } + rewritePersistedSubscription(t, m, sub.ID, func(r *subscriptionRecord) { + r.Params.TTL = tc.ttl + tc.change(r) + }) + reopenMustFail(t, dir) + }) + } +} + +func TestPermanentRecordLoadPreservesGrantsAndLegacyFinite(t *testing.T) { + m, dir, cleanup, _ := withFixtureServerURL(t, func(w http.ResponseWriter, r *http.Request) { echoVerification(w, r) }) + defer cleanup() + p := testParams("alice") + p.TTL = json.RawMessage(`null`) + sub, err := m.Subscribe(context.Background(), "alice", p) + if err != nil { + t.Fatal(err) + } + if reopened, err := Open(dir, testKey, m.delivery); err != nil { + t.Fatal(err) + } else if status, _ := reopened.Status("alice", "alice"); !status.Subscribed || !status.Bindings[0].Permanent || !status.Bindings[0].Expires.IsZero() { + t.Fatalf("valid permanent grant lost: %#v", status) + } + // A nullable legacy record predating the permanent flag stays finite. + rewritePersistedSubscription(t, m, sub.ID, func(r *subscriptionRecord) { + r.Permanent = false + r.ExpiresAt = time.Now().Add(time.Hour) + }) + reopened, err := Open(dir, testKey, m.delivery) + if err != nil { + t.Fatal(err) + } + status, _ := reopened.Status("alice", "alice") + if !status.Subscribed || len(status.Bindings) != 1 || status.Bindings[0].Permanent || status.Bindings[0].Expires.IsZero() { + t.Fatalf("legacy finite lease changed: %#v", status) + } + rewritePersistedSubscription(t, reopened, sub.ID, func(r *subscriptionRecord) { r.ExpiresAt = time.Now().Add(-time.Second) }) + expired, err := Open(dir, testKey, nil) + if err != nil { + t.Fatal(err) + } + status, _ = expired.Status("alice", "alice") + if status.Subscribed || len(status.Bindings) != 0 { + t.Fatalf("expired legacy lease active: %#v", status) + } +} diff --git a/internal/dotevents/permanent_test.go b/internal/dotevents/permanent_test.go new file mode 100644 index 0000000..1a8bba5 --- /dev/null +++ b/internal/dotevents/permanent_test.go @@ -0,0 +1,106 @@ +package dotevents + +import ( + "context" + "encoding/json" + "net/http" + "testing" + "time" +) + +func TestPermanentLeaseLifecycle(t *testing.T) { + for _, stop := range []string{"unsubscribe", "disable"} { + t.Run(stop, func(t *testing.T) { + m, dir, cleanup, _ := withFixtureServerURL(t, func(w http.ResponseWriter, r *http.Request) { echoVerification(w, r) }) + defer cleanup() + p := testParams("alice") + p.TTL = json.RawMessage(`null`) + sub, err := m.subscribe(context.Background(), "alice", p, func() time.Time { return time.Now().Add(-365 * 24 * time.Hour) }) + if err != nil { + t.Fatal(err) + } + if !sub.ExpiresAt.IsZero() { + t.Fatalf("permanent grant has expiry %v", sub.ExpiresAt) + } + reopened, err := Open(dir, testKey, m.delivery) + if err != nil { + t.Fatal(err) + } + notice := Notice{OccurredAt: time.Now(), EventID: "permanent.event", QueueID: "alice", JobID: "job", Mode: "answer"} + if err = reopened.Deliver(context.Background(), "alice", "alice", notice); err != nil { + t.Fatal(err) + } + if stop == "unsubscribe" { + err = reopened.Unsubscribe(context.Background(), "alice", p) + } else { + err = reopened.SetEnabled(false) + if err == nil { + err = reopened.SetEnabled(true) + } + } + if err != nil { + t.Fatal(err) + } + notice.EventID = "stopped.event" + if err = reopened.Deliver(context.Background(), "alice", "alice", notice); err == nil { + t.Fatal("invalidated lease delivered") + } + }) + } +} + +func TestPermanentDowngradeAndFiniteTTL(t *testing.T) { + m, _, cleanup, _ := withFixtureServerURL(t, func(w http.ResponseWriter, r *http.Request) { echoVerification(w, r) }) + defer cleanup() + p := testParams("alice") + p.TTL = json.RawMessage(`null`) + if sub, err := m.Subscribe(context.Background(), "alice", p); err != nil || !sub.ExpiresAt.IsZero() { + t.Fatalf("permanent: %#v %v", sub, err) + } + for _, raw := range []string{"", "60000", "86400000"} { + p.TTL = json.RawMessage(raw) + start := time.Now() + sub, err := m.Subscribe(context.Background(), "alice", p) + ttl, _ := parseTTL(p.TTL) + if err != nil || sub.ExpiresAt.Before(start.Add(time.Duration(ttl)*time.Millisecond)) || sub.ExpiresAt.After(time.Now().Add(time.Duration(ttl)*time.Millisecond)) { + t.Fatalf("finite %q: %#v %v", raw, sub, err) + } + } + p.TTL = json.RawMessage(`86400001`) + if _, err := m.Subscribe(context.Background(), "alice", p); err == nil { + t.Fatal("numeric cap bypassed") + } +} + +func TestPermanentSubscriptionsCountTowardCapacity(t *testing.T) { + m, _, cleanup, _ := withFixtureServerURL(t, func(w http.ResponseWriter, r *http.Request) { echoVerification(w, r) }) + defer cleanup() + p := testParams("alice") + p.TTL = json.RawMessage(`null`) + if _, err := m.Subscribe(context.Background(), "alice", p); err != nil { + t.Fatal(err) + } + if err := m.store.withLock(false, true, func(st *persistenceState) error { + var source *subscriptionRecord + for _, s := range st.Subscriptions { + source = s + break + } + for i := 1; i < maxActiveSubs; i++ { + cp := *source + cp.Params.Delivery.URL += "&slot=" + time.Unix(int64(i), 0).Format("150405") + cp.ID = subscriptionID("alice", cp.Params) + st.Subscriptions[cp.ID] = &cp + } + if activeSubscriptions(st, time.Now().Add(365*24*time.Hour)) != maxActiveSubs { + t.Fatal("permanent leases not counted") + } + return nil + }); err != nil { + t.Fatal(err) + } + p.Delivery.URL += "&extra=1" + if _, err := m.Subscribe(context.Background(), "alice", p); err == nil || asEventsError(err).Code != "capacity" { + t.Fatalf("capacity: %v", err) + } +} diff --git a/internal/dotevents/state.go b/internal/dotevents/state.go index 3444f96..ca485e0 100644 --- a/internal/dotevents/state.go +++ b/internal/dotevents/state.go @@ -16,6 +16,7 @@ const ( ) type subscriptionRecord struct { + Permanent bool `json:"permanent,omitempty"` TrustEpoch string `json:"trust_epoch,omitempty"` VerifiedUntil time.Time `json:"verified_until"` Generation string `json:"generation"` @@ -99,7 +100,9 @@ func parsePersistenceState(data []byte) (*persistenceState, error) { } } for id, sub := range st.Subscriptions { - if sub != nil && sub.TrustEpoch != "" && validateID(sub.TrustEpoch) != nil { return nil, errf("corrupt") } + if sub != nil && sub.TrustEpoch != "" && validateID(sub.TrustEpoch) != nil { + return nil, errf("corrupt") + } if sub != nil && !sub.VerifiedUntil.IsZero() { // Validate the persisted shape, not elapsed wall time. Legacy // UpdatedAt predates lock/save/HTTP work, and wall clocks can step @@ -111,6 +114,14 @@ func parsePersistenceState(data []byte) (*persistenceState, error) { if sub == nil || sub.ID != id || sub.ID != subscriptionID(sub.Principal, sub.Params) || validateID(sub.Principal) != nil || validateParams(sub.Principal, sub.Params) != nil || validateSecret(sub.Secret) != nil || sub.Secret != sub.Params.Delivery.Secret || (sub.PreviousSecret != "" && validateSecret(sub.PreviousSecret) != nil) { return nil, errf("corrupt") } + // Permanence is accepted only when the retained request and expiry agree. + // A nullable legacy record without the flag remains a finite grant. + if sub.Permanent && (!sub.ExpiresAt.IsZero() || !explicitNullTTL(sub.Params.TTL)) { + return nil, errf("corrupt") + } + if !sub.Permanent && sub.ExpiresAt.IsZero() { + return nil, errf("corrupt") + } } for key, rec := range st.Deliveries { if rec == nil || deliveryKey(rec.SubscriptionID, rec.EventID) != key || validateID(rec.SubscriptionID) != nil || validateID(rec.EventID) != nil || rec.Attempts < 1 || rec.Attempts > 4 || len(rec.Body) > maxEventBodyBytes || !json.Valid(rec.Body) { @@ -171,10 +182,14 @@ func splitQueueKey(key string) [2]string { return parts } +func (s *subscriptionRecord) leaseActive(now time.Time) bool { + return s.Permanent || s.ExpiresAt.After(now) +} + func activeSubscriptions(st *persistenceState, now time.Time) int { n := 0 for _, sub := range st.Subscriptions { - if sub != nil && sub.ExpiresAt.After(now) { + if sub != nil && sub.leaseActive(now) { n++ } } diff --git a/internal/dotevents/types.go b/internal/dotevents/types.go index 92b5c7d..4416c3a 100644 --- a/internal/dotevents/types.go +++ b/internal/dotevents/types.go @@ -26,6 +26,7 @@ type Params struct { } type Subscription struct { + Permanent bool `json:"-"` TrustEpoch string `json:"-"` ID string `json:"id"` ExpiresAt time.Time `json:"expiresAt"` @@ -52,6 +53,7 @@ type Notice struct { // empty Epoch identifies a verified legacy lease, not durable mode trust. It // intentionally carries neither the callback URL nor any credential. type TrustedBinding struct { + Permanent bool ID, Epoch string Expires, Refreshed time.Time } diff --git a/internal/dotevents/validate.go b/internal/dotevents/validate.go index 6156148..48a62e5 100644 --- a/internal/dotevents/validate.go +++ b/internal/dotevents/validate.go @@ -93,10 +93,20 @@ func validateCursor(raw json.RawMessage) error { return nil } +// explicitNullTTL distinguishes a requested permanent lease from omission; +// parseTTL maps both to a duration, so persisted records need the raw form. +func explicitNullTTL(raw json.RawMessage) bool { + return len(raw) != 0 && bytes.Equal(bytes.TrimSpace(raw), []byte("null")) +} + func parseTTL(raw json.RawMessage) (int64, error) { - if len(raw) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) { + if len(raw) == 0 { return defaultTTL, nil } + // Zero is the internal duration sentinel for an explicitly permanent request. + if explicitNullTTL(raw) { + return 0, nil + } trimmed := bytes.TrimSpace(raw) if len(trimmed) == 0 || trimmed[0] < '0' || trimmed[0] > '9' { return 0, errf("unknown") diff --git a/internal/mcpserver/bridge.go b/internal/mcpserver/bridge.go index 4413546..d79f46e 100644 --- a/internal/mcpserver/bridge.go +++ b/internal/mcpserver/bridge.go @@ -86,10 +86,10 @@ type EventsSubscribeParams struct { type EventsSubscribeResult struct { mcp.ResultBase - ID string `json:"id"` - RefreshBefore time.Time `json:"refreshBefore"` - Cursor any `json:"cursor"` - Truncated bool `json:"truncated"` + ID string `json:"id"` + RefreshBefore *time.Time `json:"refreshBefore"` + Cursor any `json:"cursor"` + Truncated bool `json:"truncated"` } type EventsUnsubscribeParams struct { @@ -161,7 +161,11 @@ func NewWithBridge(backend service.Backend, bridge *dotbridge.Service) (*mcp.Ser if err != nil { return nil, eventsError(err) } - return &EventsSubscribeResult{ID: sub.ID, RefreshBefore: sub.ExpiresAt, Cursor: nil, Truncated: false}, nil + var refreshBefore *time.Time + if !sub.Permanent { + refreshBefore = &sub.ExpiresAt + } + return &EventsSubscribeResult{ID: sub.ID, RefreshBefore: refreshBefore, Cursor: nil, Truncated: false}, nil }); err != nil { return nil, err } diff --git a/internal/mcpserver/permanent_test.go b/internal/mcpserver/permanent_test.go new file mode 100644 index 0000000..2acf752 --- /dev/null +++ b/internal/mcpserver/permanent_test.go @@ -0,0 +1,55 @@ +package mcpserver + +import ( + "encoding/json" + "net/http/httptest" + "testing" +) + +func TestPermanentEventsSubscribeResponseOverHTTP(t *testing.T) { + s := openBridgeForMCPTest(t) + delivery, closeDelivery := NewTestEventsDelivery(t) + defer closeDelivery() + s.ReplaceEventsDeliveryForTest(delivery) + h, err := HTTPWithBridge(bridgeNoopBackend{}, s) + if err != nil { + t.Fatal(err) + } + server := httptest.NewServer(h) + defer server.Close() + env := eventsMCPRequest(t, server.URL, "events/subscribe", map[string]any{ + "_meta": map[string]any{ + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientCapabilities": map[string]any{}, + "io.modelcontextprotocol/clientInfo": map[string]any{"name": "fixture", "version": "1"}, + }, + "name": "dot.task.available", + "arguments": map[string]string{"queue_id": "client.a"}, + "delivery": map[string]string{ + "mode": "webhook", + "url": fixtureCallbackURL, + "secret": "whsec_MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=", + }, + "cursor": nil, "ttlMs": nil, + }) + var out map[string]json.RawMessage + if json.Unmarshal(env["result"], &out) != nil { + t.Fatalf("subscribe envelope: %s", env["result"]) + } + if out["id"] == nil || string(out["id"]) == `""` { + t.Fatalf("id missing: %s", env["result"]) + } + + if string(out["refreshBefore"]) != "null" { + t.Fatalf("refreshBefore missing: %s", env["result"]) + } + if string(out["cursor"]) != "null" || string(out["truncated"]) != "false" { + t.Fatalf("official cursor/truncated: %s", env["result"]) + } + if _, nested := out["subscription"]; nested { + t.Fatalf("nested subscription object: %s", env["result"]) + } + if _, leaked := out["secret"]; leaked { + t.Fatalf("secret leaked: %s", env["result"]) + } +}