Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions docs/dot-permanent-subscriptions.md
Original file line number Diff line number Diff line change
@@ -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.
40 changes: 36 additions & 4 deletions internal/dotbridge/authorization.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
)

type subscriptionProof struct {
Permanent bool
BindingID string
TrustEpoch string
AuthHash string
Expand Down Expand Up @@ -70,19 +71,50 @@ 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 {
return sub, false
}
}
}
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
}
131 changes: 131 additions & 0 deletions internal/dotbridge/permanent_load_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
140 changes: 140 additions & 0 deletions internal/dotbridge/permanent_test.go
Original file line number Diff line number Diff line change
@@ -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")
}
})
}
}
2 changes: 1 addition & 1 deletion internal/dotbridge/probe.go
Original file line number Diff line number Diff line change
Expand Up @@ -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})
Expand Down
Loading
Loading