diff --git a/internal/dwn/feed_grants_test.go b/internal/dwn/feed_grants_test.go new file mode 100644 index 0000000..a0932c4 --- /dev/null +++ b/internal/dwn/feed_grants_test.go @@ -0,0 +1,207 @@ +package dwn + +import ( + "errors" + "testing" + "time" +) + +const ( + feedGrantor = "did:dht:owner" + feedGrantee = "did:jwk:delegate" + meshProto = "https://example.com/protocols/mesh" + auditProto = "https://example.com/protocols/audit" +) + +var feedNow = time.Date(2026, 7, 26, 12, 0, 0, 0, time.UTC) + +func feedGrant(id, protocol, protocolPath, contextID string) *PermissionGrant { + return &PermissionGrant{ + ID: id, + Grantor: feedGrantor, + Grantee: feedGrantee, + DateGranted: "2026-07-01T00:00:00.000000Z", + DateExpires: "2026-08-01T00:00:00.000000Z", + Scope: PermissionScope{ + Interface: DwnScopeInterfaceMessages, + Method: DwnScopeMethodRead, + Protocol: protocol, + ProtocolPath: protocolPath, + ContextID: contextID, + }, + } +} + +func grantIDsOf(grants []*PermissionGrant) []string { + ids := make([]string, len(grants)) + for i, g := range grants { + ids[i] = g.ID + } + return ids +} + +func TestSelectFeedGrantsFullScope(t *testing.T) { + unscoped := feedGrant("grant-unscoped", "", "", "") + protocolScoped := feedGrant("grant-mesh", meshProto, "", "") + + selected, err := SelectFeedGrants( + []*PermissionGrant{protocolScoped, unscoped}, + feedGrantor, feedGrantee, feedNow, FullSyncScope(), + ) + if err != nil { + t.Fatalf("SelectFeedGrants: %v", err) + } + // Full scope invokes only unscoped grants; protocol-scoped ones cover a + // strict subset and do not participate. + if got := grantIDsOf(selected); len(got) != 1 || got[0] != "grant-unscoped" { + t.Errorf("selected = %v, want [grant-unscoped]", got) + } + + _, err = SelectFeedGrants( + []*PermissionGrant{protocolScoped}, + feedGrantor, feedGrantee, feedNow, FullSyncScope(), + ) + var coverage *FeedGrantCoverageError + if !errors.As(err, &coverage) || coverage.Protocol != "" { + t.Fatalf("error = %v, want full-scope FeedGrantCoverageError", err) + } +} + +func TestSelectFeedGrantsProtocolSet(t *testing.T) { + scope, err := ProtocolSetSyncScope([]string{meshProto, auditProto}) + if err != nil { + t.Fatalf("ProtocolSetSyncScope: %v", err) + } + + t.Run("protocol roots cover the set and all participants are invoked", func(t *testing.T) { + mesh := feedGrant("b-mesh", meshProto, "", "") + audit := feedGrant("a-audit", auditProto, "", "") + unrelated := feedGrant("c-other", "https://example.com/protocols/unrelated", "", "") + + selected, err := SelectFeedGrants( + []*PermissionGrant{mesh, audit, unrelated}, + feedGrantor, feedGrantee, feedNow, scope, + ) + if err != nil { + t.Fatalf("SelectFeedGrants: %v", err) + } + got := grantIDsOf(selected) + if len(got) != 2 || got[0] != "a-audit" || got[1] != "b-mesh" { + t.Errorf("selected = %v, want [a-audit b-mesh] sorted by id", got) + } + }) + + t.Run("unscoped grant covers every protocol and participates", func(t *testing.T) { + unscoped := feedGrant("only-unscoped", "", "", "") + selected, err := SelectFeedGrants( + []*PermissionGrant{unscoped}, + feedGrantor, feedGrantee, feedNow, scope, + ) + if err != nil { + t.Fatalf("SelectFeedGrants: %v", err) + } + if got := grantIDsOf(selected); len(got) != 1 || got[0] != "only-unscoped" { + t.Errorf("selected = %v, want [only-unscoped]", got) + } + }) + + t.Run("missing coverage names the uncovered protocol", func(t *testing.T) { + mesh := feedGrant("b-mesh", meshProto, "", "") + _, err := SelectFeedGrants( + []*PermissionGrant{mesh}, + feedGrantor, feedGrantee, feedNow, scope, + ) + var coverage *FeedGrantCoverageError + if !errors.As(err, &coverage) || coverage.Protocol != auditProto { + t.Fatalf("error = %v, want FeedGrantCoverageError{Protocol: %q}", err, auditProto) + } + }) + + t.Run("subtree-scoped grants never authorize feed scopes", func(t *testing.T) { + pathScoped := feedGrant("path-scoped", meshProto, "network/node", "") + contextScoped := feedGrant("ctx-scoped", auditProto, "", "ctx-root") + _, err := SelectFeedGrants( + []*PermissionGrant{pathScoped, contextScoped}, + feedGrantor, feedGrantee, feedNow, scope, + ) + var coverage *FeedGrantCoverageError + if !errors.As(err, &coverage) { + t.Fatalf("error = %v, want FeedGrantCoverageError", err) + } + }) + + t.Run("selected IDs use UTF-16 code-unit order", func(t *testing.T) { + // U+1F600 encodes as a surrogate pair and sorts BEFORE U+FFFD in + // UTF-16 order, unlike UTF-8 byte order. + astral := feedGrant("\U0001F600-grant", meshProto, "", "") + replacement := feedGrant("�-grant", auditProto, "", "") + selected, err := SelectFeedGrants( + []*PermissionGrant{replacement, astral}, + feedGrantor, feedGrantee, feedNow, scope, + ) + if err != nil { + t.Fatalf("SelectFeedGrants: %v", err) + } + got := grantIDsOf(selected) + if len(got) != 2 || got[0] != "\U0001F600-grant" || got[1] != "�-grant" { + t.Errorf("selected = %q, want astral grant first (UTF-16 order)", got) + } + }) +} + +func TestSelectFeedGrantsActiveWindow(t *testing.T) { + scope, err := ProtocolSetSyncScope([]string{meshProto}) + if err != nil { + t.Fatalf("ProtocolSetSyncScope: %v", err) + } + + tests := []struct { + name string + mutate func(*PermissionGrant) + }{ + {"expired (dateExpires == now excluded)", func(g *PermissionGrant) { + g.DateExpires = Timestamp(feedNow) + }}, + {"not yet active", func(g *PermissionGrant) { + g.DateGranted = "2026-07-27T00:00:00.000000Z" + }}, + {"wrong grantor", func(g *PermissionGrant) { g.Grantor = "did:dht:someone-else" }}, + {"wrong grantee", func(g *PermissionGrant) { g.Grantee = "did:jwk:someone-else" }}, + {"records scope excluded", func(g *PermissionGrant) { g.Scope.Interface = DwnScopeInterfaceRecords }}, + {"messages query method excluded", func(g *PermissionGrant) { g.Scope.Method = DwnScopeMethodQuery }}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + grant := feedGrant("g", meshProto, "", "") + tc.mutate(grant) + _, err := SelectFeedGrants([]*PermissionGrant{grant}, feedGrantor, feedGrantee, feedNow, scope) + var coverage *FeedGrantCoverageError + if !errors.As(err, &coverage) { + t.Fatalf("error = %v, want FeedGrantCoverageError (grant must be inactive)", err) + } + }) + } + + t.Run("dateGranted == now is active", func(t *testing.T) { + grant := feedGrant("g", meshProto, "", "") + grant.DateGranted = Timestamp(feedNow) + selected, err := SelectFeedGrants([]*PermissionGrant{grant}, feedGrantor, feedGrantee, feedNow, scope) + if err != nil || len(selected) != 1 { + t.Fatalf("selected, err = %v, %v; want the boundary grant active", grantIDsOf(selected), err) + } + }) +} + +func TestFeedGrantIDs(t *testing.T) { + grants := []*PermissionGrant{ + feedGrant("b", meshProto, "", ""), + feedGrant("a", meshProto, "", ""), + } + ids := FeedGrantIDs(grants) + if len(ids) != 2 || ids[0] != "a" || ids[1] != "b" { + t.Errorf("FeedGrantIDs = %v, want [a b]", ids) + } + if FeedGrantIDs(nil) != nil { + t.Error("FeedGrantIDs(nil) must be nil") + } +} diff --git a/internal/dwn/link_identity.go b/internal/dwn/link_identity.go new file mode 100644 index 0000000..edcd6d8 --- /dev/null +++ b/internal/dwn/link_identity.go @@ -0,0 +1,237 @@ +package dwn + +import ( + "bytes" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "errors" + "fmt" +) + +// Replication-link identity: canonical sync scopes, projection IDs, +// authorization epochs, and link keys. These port the enbox agent's +// types/sync.ts and sync-link-key.ts hash algorithms: base64url (no padding) +// SHA-256 over canonical JSON — JSON.stringify with keys in ALPHABETICAL +// order and no extra whitespace. Key order is part of the hash input, so the +// canonical structs below declare fields in alphabetical JSON-key order. + +// Version strings of the link-identity hash algorithms. +const ( + // SyncProjectionRootVersion identifies the projection-ID algorithm used + // by durable message-feed full/protocol sync scopes. + SyncProjectionRootVersion = "replication-log-feed-v1" + + // SyncAuthorizationEpochVersion identifies the authorization-epoch + // algorithm used by delegated Messages.Read sync links. + SyncAuthorizationEpochVersion = "messages-read-grants-v1" +) + +// LinkKeySeparator joins the components of compound replication link +// identifiers. +const LinkKeySeparator = "^" + +// SyncScope kinds. +const ( + SyncScopeKindFull = "full" + SyncScopeKindProtocolSet = "protocolSet" +) + +// ErrSyncScopeEmptyProtocols is returned when a protocol-set scope has no +// protocol URIs. +var ErrSyncScopeEmptyProtocols = errors.New("protocol-set scope requires at least one protocol URI") + +// SyncScope describes the primary CID set a replication link syncs: the full +// tenant feed or a set of protocol roots. Narrow protocolPath/contextId sync +// is not representable; a delegate must have grant coverage for every full +// protocol in the protocol set. +type SyncScope struct { + Kind string + Protocols []string +} + +// FullSyncScope returns the full-tenant scope. +func FullSyncScope() SyncScope { + return SyncScope{Kind: SyncScopeKindFull} +} + +// ProtocolSetSyncScope returns a protocol-set scope with the protocol list +// normalized to canonical scope-union order. +func ProtocolSetSyncScope(protocols []string) (SyncScope, error) { + normalized, err := NormalizeSyncProtocols(protocols) + if err != nil { + return SyncScope{}, err + } + return SyncScope{Kind: SyncScopeKindProtocolSet, Protocols: normalized}, nil +} + +// NormalizeSyncProtocols normalizes a protocol list into canonical +// scope-union order: deduplicated and sorted by UTF-16 code-unit order +// (JavaScript string order). An empty result is an error. +func NormalizeSyncProtocols(protocols []string) ([]string, error) { + seen := make(map[string]struct{}, len(protocols)) + unique := make([]string, 0, len(protocols)) + for _, p := range protocols { + if _, ok := seen[p]; ok { + continue + } + seen[p] = struct{}{} + unique = append(unique, p) + } + if len(unique) == 0 { + return nil, ErrSyncScopeEmptyProtocols + } + sortStringsUTF16(unique) + return unique, nil +} + +// canonicalize validates the scope and returns it in canonical form. +func (s SyncScope) canonicalize() (canonicalSyncScope, error) { + switch s.Kind { + case SyncScopeKindFull: + return canonicalSyncScope{Kind: SyncScopeKindFull}, nil + case SyncScopeKindProtocolSet: + normalized, err := NormalizeSyncProtocols(s.Protocols) + if err != nil { + return canonicalSyncScope{}, err + } + return canonicalSyncScope{Kind: SyncScopeKindProtocolSet, Protocols: normalized}, nil + default: + return canonicalSyncScope{}, fmt.Errorf("unknown sync scope kind %q", s.Kind) + } +} + +// Canonical JSON shapes. Field declaration order IS the serialized key order +// and must stay alphabetical — reordering silently changes every persisted +// projection ID and authorization epoch. +type canonicalSyncScope struct { + Kind string `json:"kind"` + Protocols []string `json:"protocols,omitempty"` +} + +type canonicalProjectionIDInput struct { + Scope canonicalSyncScope `json:"scope"` + TenantDID string `json:"tenantDid"` + Version string `json:"version"` +} + +type canonicalEpochGrant struct { + DateExpires string `json:"dateExpires"` + DateGranted string `json:"dateGranted,omitempty"` + ID string `json:"id"` +} + +type canonicalOwnerEpochInput struct { + Kind string `json:"kind"` + Version string `json:"version"` +} + +type canonicalDelegateEpochInput struct { + DelegateDID string `json:"delegateDid"` + Grants []canonicalEpochGrant `json:"grants"` + Kind string `json:"kind"` + Version string `json:"version"` +} + +// hashCanonicalObject computes base64url-no-padding(SHA-256(canonical JSON)). +// JSON.stringify does not HTML-escape, so the encoder must not either. +func hashCanonicalObject(value any) (string, error) { + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + enc.SetEscapeHTML(false) + if err := enc.Encode(value); err != nil { + return "", fmt.Errorf("encoding canonical object: %w", err) + } + canonical := bytes.TrimSuffix(buf.Bytes(), []byte("\n")) + sum := sha256.Sum256(canonical) + return base64.RawURLEncoding.EncodeToString(sum[:]), nil +} + +// ComputeProjectionID computes the deterministic projection ID of a sync +// scope for a tenant. It intentionally excludes endpoint, grant IDs, and +// authorization epoch. +func ComputeProjectionID(tenantDID string, scope SyncScope) (string, error) { + canonicalScope, err := scope.canonicalize() + if err != nil { + return "", err + } + return hashCanonicalObject(canonicalProjectionIDInput{ + Scope: canonicalScope, + TenantDID: tenantDID, + Version: SyncProjectionRootVersion, + }) +} + +// SyncAuthorizationGrant is the grant metadata that participates in +// delegated authorization-epoch hashing. DateGranted may be empty, in which +// case it is omitted from the canonical form (the SDK's undefined). +type SyncAuthorizationGrant struct { + ID string + DateGranted string + DateExpires string +} + +// SyncAuthorizationGrantsFromPermissionGrants maps parsed permission grants +// to authorization-epoch inputs. +func SyncAuthorizationGrantsFromPermissionGrants(grants []*PermissionGrant) []SyncAuthorizationGrant { + out := make([]SyncAuthorizationGrant, 0, len(grants)) + for _, grant := range grants { + out = append(out, SyncAuthorizationGrant{ + ID: grant.ID, + DateGranted: grant.DateGranted, + DateExpires: grant.DateExpires, + }) + } + return out +} + +// ComputeOwnerAuthorizationEpoch computes the stable authorization epoch of +// an owner link (owner links do not invoke grants). +func ComputeOwnerAuthorizationEpoch() string { + epoch, err := hashCanonicalObject(canonicalOwnerEpochInput{ + Kind: "owner", + Version: SyncAuthorizationEpochVersion, + }) + if err != nil { + // Static input; marshaling cannot fail. + panic(err) + } + return epoch +} + +// ComputeDelegateAuthorizationEpoch computes the authorization epoch of a +// delegate link from the delegate DID plus the active grant set. A changed +// grant set creates a new link key even when the projection ID is unchanged. +func ComputeDelegateAuthorizationEpoch(delegateDID string, grants []SyncAuthorizationGrant) (string, error) { + if len(grants) == 0 { + return "", errors.New("delegate authorization requires at least one grant") + } + sorted := make([]SyncAuthorizationGrant, len(grants)) + copy(sorted, grants) + for i := 1; i < len(sorted); i++ { + for j := i; j > 0 && compareStringsUTF16(sorted[j].ID, sorted[j-1].ID) < 0; j-- { + sorted[j], sorted[j-1] = sorted[j-1], sorted[j] + } + } + canonicalGrants := make([]canonicalEpochGrant, len(sorted)) + for i, grant := range sorted { + canonicalGrants[i] = canonicalEpochGrant{ + DateExpires: grant.DateExpires, + DateGranted: grant.DateGranted, + ID: grant.ID, + } + } + return hashCanonicalObject(canonicalDelegateEpochInput{ + DelegateDID: delegateDID, + Grants: canonicalGrants, + Kind: "delegate", + Version: SyncAuthorizationEpochVersion, + }) +} + +// BuildLinkKey builds the runtime identifier of a replication link: +// (tenantDid, remoteEndpoint, projectionId, authorizationEpoch) joined with +// "^". +func BuildLinkKey(tenantDID, remoteEndpoint, projectionID, authorizationEpoch string) string { + return tenantDID + LinkKeySeparator + remoteEndpoint + LinkKeySeparator + projectionID + LinkKeySeparator + authorizationEpoch +} diff --git a/internal/dwn/link_identity_test.go b/internal/dwn/link_identity_test.go new file mode 100644 index 0000000..5a28662 --- /dev/null +++ b/internal/dwn/link_identity_test.go @@ -0,0 +1,176 @@ +package dwn + +import ( + "encoding/json" + "errors" + "os" + "reflect" + "testing" +) + +// Fixtures generated by executing the enbox agent implementation +// (packages/agent/src/types/sync.ts + sync-link-key.ts); see the provenance +// note in messages_test.go. +type linkIdentityFixtureFile struct { + ProjectionIDs []struct { + Name string `json:"name"` + TenantDID string `json:"tenantDid"` + Scope struct { + Kind string `json:"kind"` + Protocols []string `json:"protocols"` + } `json:"scope"` + ProjectionID string `json:"projectionId"` + } `json:"projectionIds"` + AuthorizationEpochs []struct { + Name string `json:"name"` + Input struct { + Kind string `json:"kind"` + DelegateDID string `json:"delegateDid"` + Grants []struct { + ID string `json:"id"` + DateExpires string `json:"dateExpires"` + DateGranted string `json:"dateGranted"` + } `json:"grants"` + } `json:"input"` + Epoch string `json:"epoch"` + } `json:"authorizationEpochs"` + LinkKeys []struct { + Name string `json:"name"` + TenantDID string `json:"tenantDid"` + RemoteEndpoint string `json:"remoteEndpoint"` + ProjectionID string `json:"projectionId"` + AuthorizationEpoch string `json:"authorizationEpoch"` + LinkKey string `json:"linkKey"` + } `json:"linkKeys"` + NormalizeSyncProtocols []struct { + Name string `json:"name"` + Input []string `json:"input"` + Normalized []string `json:"normalized"` + } `json:"normalizeSyncProtocols"` +} + +func loadLinkIdentityFixtures(t *testing.T) *linkIdentityFixtureFile { + t.Helper() + data, err := os.ReadFile("testdata/link_identity_fixtures.json") + if err != nil { + t.Fatalf("reading fixtures: %v", err) + } + var file linkIdentityFixtureFile + if err := json.Unmarshal(data, &file); err != nil { + t.Fatalf("parsing fixtures: %v", err) + } + return &file +} + +func TestComputeProjectionIDMatchesTypeScriptFixtures(t *testing.T) { + file := loadLinkIdentityFixtures(t) + if len(file.ProjectionIDs) == 0 { + t.Fatal("no projection ID vectors") + } + for _, vector := range file.ProjectionIDs { + t.Run(vector.Name, func(t *testing.T) { + scope := SyncScope{Kind: vector.Scope.Kind, Protocols: vector.Scope.Protocols} + got, err := ComputeProjectionID(vector.TenantDID, scope) + if err != nil { + t.Fatalf("ComputeProjectionID: %v", err) + } + if got != vector.ProjectionID { + t.Errorf("projection ID = %s, want %s", got, vector.ProjectionID) + } + }) + } +} + +func TestComputeAuthorizationEpochMatchesTypeScriptFixtures(t *testing.T) { + file := loadLinkIdentityFixtures(t) + if len(file.AuthorizationEpochs) == 0 { + t.Fatal("no authorization epoch vectors") + } + for _, vector := range file.AuthorizationEpochs { + t.Run(vector.Name, func(t *testing.T) { + var ( + got string + err error + ) + switch vector.Input.Kind { + case "owner": + got = ComputeOwnerAuthorizationEpoch() + case "delegate": + grants := make([]SyncAuthorizationGrant, len(vector.Input.Grants)) + for i, grant := range vector.Input.Grants { + grants[i] = SyncAuthorizationGrant{ + ID: grant.ID, + DateGranted: grant.DateGranted, + DateExpires: grant.DateExpires, + } + } + got, err = ComputeDelegateAuthorizationEpoch(vector.Input.DelegateDID, grants) + if err != nil { + t.Fatalf("ComputeDelegateAuthorizationEpoch: %v", err) + } + default: + t.Fatalf("unknown epoch kind %q", vector.Input.Kind) + } + if got != vector.Epoch { + t.Errorf("epoch = %s, want %s", got, vector.Epoch) + } + }) + } +} + +func TestBuildLinkKeyMatchesTypeScriptFixtures(t *testing.T) { + file := loadLinkIdentityFixtures(t) + for _, vector := range file.LinkKeys { + got := BuildLinkKey(vector.TenantDID, vector.RemoteEndpoint, vector.ProjectionID, vector.AuthorizationEpoch) + if got != vector.LinkKey { + t.Errorf("%s: link key = %s, want %s", vector.Name, got, vector.LinkKey) + } + } +} + +func TestNormalizeSyncProtocolsMatchesTypeScriptFixtures(t *testing.T) { + file := loadLinkIdentityFixtures(t) + for _, vector := range file.NormalizeSyncProtocols { + got, err := NormalizeSyncProtocols(vector.Input) + if err != nil { + t.Fatalf("%s: NormalizeSyncProtocols: %v", vector.Name, err) + } + if !reflect.DeepEqual(got, vector.Normalized) { + t.Errorf("%s: normalized = %q, want %q", vector.Name, got, vector.Normalized) + } + } +} + +func TestSyncScopeValidation(t *testing.T) { + if _, err := NormalizeSyncProtocols(nil); !errors.Is(err, ErrSyncScopeEmptyProtocols) { + t.Errorf("empty protocols error = %v, want ErrSyncScopeEmptyProtocols", err) + } + if _, err := ProtocolSetSyncScope(nil); !errors.Is(err, ErrSyncScopeEmptyProtocols) { + t.Errorf("empty scope error = %v, want ErrSyncScopeEmptyProtocols", err) + } + if _, err := ComputeProjectionID("did:dht:x", SyncScope{Kind: "bogus"}); err == nil { + t.Error("unknown scope kind must be rejected") + } + if _, err := ComputeDelegateAuthorizationEpoch("did:jwk:x", nil); err == nil { + t.Error("delegate epoch without grants must be rejected") + } + + // The projection ID must not depend on input protocol order/duplicates. + a, err := ComputeProjectionID("did:dht:x", SyncScope{Kind: SyncScopeKindProtocolSet, Protocols: []string{"b", "a", "b"}}) + if err != nil { + t.Fatalf("ComputeProjectionID: %v", err) + } + b, err := ComputeProjectionID("did:dht:x", SyncScope{Kind: SyncScopeKindProtocolSet, Protocols: []string{"a", "b"}}) + if err != nil { + t.Fatalf("ComputeProjectionID: %v", err) + } + if a != b { + t.Error("projection ID must be order/duplicate independent") + } + + if got := SyncAuthorizationGrantsFromPermissionGrants([]*PermissionGrant{ + {ID: "g1", DateGranted: "2026-01-01T00:00:00.000000Z", DateExpires: "2026-02-01T00:00:00.000000Z"}, + }); len(got) != 1 || got[0].ID != "g1" || got[0].DateExpires != "2026-02-01T00:00:00.000000Z" { + t.Errorf("SyncAuthorizationGrantsFromPermissionGrants = %+v", got) + } +} diff --git a/internal/dwn/messages.go b/internal/dwn/messages.go new file mode 100644 index 0000000..f81b82c --- /dev/null +++ b/internal/dwn/messages.go @@ -0,0 +1,589 @@ +package dwn + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "unicode/utf16" + + gocid "github.com/ipfs/go-cid" +) + +// Messages interface support: MessagesQuery, MessagesRead, MessagesSubscribe. +// +// These are the replication-log feed read surfaces of the enbox DWN. Unlike +// Records/Protocols operations, Messages operations invoke permission grants +// as a PLURAL, canonical `permissionGrantIds` list mirrored in the descriptor +// AND the signature payload, and never use protocol roles or delegated grants. + +// Sentinel errors mirroring the dwn-sdk-js permission grant invocation rules. +var ( + // ErrPermissionGrantInvocationAmbiguous mirrors + // MessagePermissionGrantCreateInvocationAmbiguous: a message may invoke + // the singular permissionGrantId or the plural permissionGrantIds, never + // both. + ErrPermissionGrantInvocationAmbiguous = errors.New("permission grant invocation must use either permissionGrantId or permissionGrantIds, not both") + + // ErrPermissionGrantIDsEmpty mirrors MessagePermissionGrantIdsEmpty: an + // explicitly supplied permissionGrantIds list must contain at least one + // grant ID. + ErrPermissionGrantIDsEmpty = errors.New("permissionGrantIds must contain at least one grant ID") + + // ErrPermissionGrantIDsNotCanonical mirrors + // MessagePermissionGrantIdsNotCanonical: a received permissionGrantIds + // list must already be sorted and deduplicated. + ErrPermissionGrantIDsNotCanonical = errors.New("permissionGrantIds must be sorted and deduplicated") + + // ErrMessagesEntryNotFound is returned by MessagesRead when the target + // DWN does not have a message with the requested CID. + ErrMessagesEntryNotFound = errors.New("messages entry not found") +) + +// +// --- Canonical permissionGrantIds --- +// + +// compareStringsUTF16 compares two strings by UTF-16 code units, matching +// JavaScript's `<`/`>` string comparison (used by the SDK's +// lexicographicalCompare). For ASCII this equals bytewise order, but astral +// (surrogate-pair) characters sort BEFORE U+E000..U+FFFF, unlike UTF-8 byte +// order — the golden fixtures pin this down. +func compareStringsUTF16(a, b string) int { + ua := utf16.Encode([]rune(a)) + ub := utf16.Encode([]rune(b)) + for i := 0; i < len(ua) && i < len(ub); i++ { + if ua[i] != ub[i] { + if ua[i] < ub[i] { + return -1 + } + return 1 + } + } + switch { + case len(ua) < len(ub): + return -1 + case len(ua) > len(ub): + return 1 + default: + return 0 + } +} + +// sortStringsUTF16 sorts in place by UTF-16 code-unit order. +func sortStringsUTF16(values []string) { + // Insertion sort keeps this dependency-free; grant ID lists are small. + for i := 1; i < len(values); i++ { + for j := i; j > 0 && compareStringsUTF16(values[j], values[j-1]) < 0; j-- { + values[j], values[j-1] = values[j-1], values[j] + } + } +} + +// CanonicalPermissionGrantIDs normalizes a permission grant ID list to the +// canonical wire form: deduplicated and sorted by UTF-16 code-unit order +// (JavaScript string order). A nil list means "no grants invoked" and returns +// nil; an explicitly supplied empty list is an error, mirroring the SDK. +func CanonicalPermissionGrantIDs(ids []string) ([]string, error) { + if ids == nil { + return nil, nil + } + if len(ids) == 0 { + return nil, ErrPermissionGrantIDsEmpty + } + seen := make(map[string]struct{}, len(ids)) + canonical := make([]string, 0, len(ids)) + for _, id := range ids { + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + canonical = append(canonical, id) + } + sortStringsUTF16(canonical) + return canonical, nil +} + +// ValidateCanonicalPermissionGrantIDs verifies a received permissionGrantIds +// list is already in canonical form (sorted, deduplicated, non-empty), +// mirroring the SDK's parse-time validation. +func ValidateCanonicalPermissionGrantIDs(ids []string) error { + if ids == nil { + return nil + } + canonical, err := CanonicalPermissionGrantIDs(ids) + if err != nil { + return err + } + if len(canonical) != len(ids) { + return ErrPermissionGrantIDsNotCanonical + } + for i := range ids { + if ids[i] != canonical[i] { + return ErrPermissionGrantIDsNotCanonical + } + } + return nil +} + +// NormalizePermissionGrantInvocation applies the SDK's grant invocation +// rules: singular and plural forms are mutually exclusive, and a supplied +// plural list is canonicalized. It returns the normalized plural list (nil +// when the invocation is singular or absent). +func NormalizePermissionGrantInvocation(permissionGrantID string, permissionGrantIDs []string) ([]string, error) { + if permissionGrantID != "" && permissionGrantIDs != nil { + return nil, ErrPermissionGrantInvocationAmbiguous + } + if permissionGrantID != "" { + return nil, nil + } + return CanonicalPermissionGrantIDs(permissionGrantIDs) +} + +// +// --- MessagesFilter --- +// + +// MessagesTimestampRange is the messageTimestamp range criterion of a +// MessagesFilter. At least one bound must be set. +type MessagesTimestampRange struct { + From string `json:"from,omitempty"` + To string `json:"to,omitempty"` +} + +// MessagesFilter selects messages from the replication-log feed. +// +// The wire schema is additionalProperties:false with minProperties:1 — only +// these fields may appear and at least one must be set. ProtocolPathPrefix +// and ContextIDPrefix are segment-aware prefixes: they match a candidate that +// equals the prefix or starts with the prefix followed by "/". +type MessagesFilter struct { + Interface string `json:"interface,omitempty"` + Method string `json:"method,omitempty"` + Protocol string `json:"protocol,omitempty"` + ProtocolPathPrefix string `json:"protocolPathPrefix,omitempty"` + ContextIDPrefix string `json:"contextIdPrefix,omitempty"` + MessageTimestamp *MessagesTimestampRange `json:"messageTimestamp,omitempty"` +} + +// messagesFilterToMap converts a filter to its wire map, omitting zero-value +// fields. Empty filters are rejected (schema minProperties: 1). +func messagesFilterToMap(f MessagesFilter) (map[string]any, error) { + m := make(map[string]any) + if f.Interface != "" { + m["interface"] = f.Interface + } + if f.Method != "" { + m["method"] = f.Method + } + if f.Protocol != "" { + m["protocol"] = f.Protocol + } + if f.ProtocolPathPrefix != "" { + m["protocolPathPrefix"] = f.ProtocolPathPrefix + } + if f.ContextIDPrefix != "" { + m["contextIdPrefix"] = f.ContextIDPrefix + } + if f.MessageTimestamp != nil { + r := make(map[string]any) + if f.MessageTimestamp.From != "" { + r["from"] = f.MessageTimestamp.From + } + if f.MessageTimestamp.To != "" { + r["to"] = f.MessageTimestamp.To + } + if len(r) == 0 { + return nil, fmt.Errorf("messages filter messageTimestamp range requires from or to") + } + m["messageTimestamp"] = r + } + if len(m) == 0 { + return nil, fmt.Errorf("messages filter requires at least one field") + } + return m, nil +} + +// messagesFiltersToWire converts a filter list to the descriptor's `filters` +// value. The field is ALWAYS present on the wire, as an empty array when no +// filters are given. +func messagesFiltersToWire(filters []MessagesFilter) ([]any, error) { + wire := make([]any, 0, len(filters)) + for i, f := range filters { + m, err := messagesFilterToMap(f) + if err != nil { + return nil, fmt.Errorf("filter %d: %w", i, err) + } + wire = append(wire, m) + } + return wire, nil +} + +// progressTokenToMap converts a ProgressToken to its wire map for embedding +// inside a signed descriptor. +func progressTokenToMap(t *ProgressToken) map[string]any { + m := map[string]any{ + "streamId": t.StreamID, + "epoch": t.Epoch, + "position": t.Position, + } + if t.MessageCID != "" { + m["messageCid"] = t.MessageCID + } + return m +} + +// +// --- Messages message builders --- +// + +// messagesSignaturePayload is the JWS payload for Messages interface +// operations: {descriptorCid, permissionGrantIds}. The SAME canonical grant +// ID list appears in the descriptor and in this payload. +type messagesSignaturePayload struct { + DescriptorCID string `json:"descriptorCid"` + PermissionGrantIDs []string `json:"permissionGrantIds,omitempty"` +} + +// signMessagesMessage adds the canonical grant IDs to the descriptor, signs +// the Messages-specific payload, and assembles the message. Messages +// operations never use protocol roles or authorDelegatedGrant. +func signMessagesMessage(s *Signer, desc map[string]any, canonicalGrantIDs []string) (*Message, error) { + if len(canonicalGrantIDs) > 0 { + desc["permissionGrantIds"] = canonicalGrantIDs + } + + descriptorCID, err := ComputeCID(desc) + if err != nil { + return nil, fmt.Errorf("computing descriptor CID: %w", err) + } + + sigPayload := messagesSignaturePayload{ + DescriptorCID: descriptorCID, + PermissionGrantIDs: canonicalGrantIDs, + } + + jws, err := SignJWS(sigPayload, s.KeyID(), s.PrivateKey) + if err != nil { + return nil, fmt.Errorf("signing: %w", err) + } + + return &Message{ + Descriptor: desc, + Authorization: &Authorization{Signature: jws}, + }, nil +} + +// MessagesQueryOptions configures a MessagesQuery message. +type MessagesQueryOptions struct { + // Filters selects the feed scope; an empty list queries the whole feed. + // The filters field is always present on the wire. + Filters []MessagesFilter + + // PermissionGrantIDs invokes a grant set. The list is canonicalized + // (deduplicated, sorted) before signing; nil means owner-authored. + PermissionGrantIDs []string + + // Cursor resumes the feed after this progress token. + Cursor *ProgressToken + + // Limit bounds the page size when > 0. + Limit int + + // CidsOnly requests entries without the full message (seq/messageCid + // metadata only) — used for fingerprint convergence probes. + CidsOnly bool + + // MessageTimestamp overrides the descriptor timestamp. Leave empty for + // the current time; used for deterministic fixtures. + MessageTimestamp string +} + +// BuildMessagesQuery constructs and signs a MessagesQuery message. +func BuildMessagesQuery(s *Signer, opts MessagesQueryOptions) (*Message, error) { + canonicalIDs, err := CanonicalPermissionGrantIDs(opts.PermissionGrantIDs) + if err != nil { + return nil, err + } + filters, err := messagesFiltersToWire(opts.Filters) + if err != nil { + return nil, err + } + + timestamp := opts.MessageTimestamp + if timestamp == "" { + timestamp = Now() + } + + desc := map[string]any{ + "interface": "Messages", + "method": "Query", + "messageTimestamp": timestamp, + "filters": filters, + } + if opts.Cursor != nil { + if !opts.Cursor.valid() { + return nil, fmt.Errorf("invalid progress token cursor") + } + desc["cursor"] = progressTokenToMap(opts.Cursor) + } + if opts.Limit > 0 { + desc["limit"] = opts.Limit + } + if opts.CidsOnly { + desc["cidsOnly"] = true + } + + return signMessagesMessage(s, desc, canonicalIDs) +} + +// MessagesReadOptions configures a MessagesRead message. +type MessagesReadOptions struct { + // MessageCID is the CID of the message to read. + MessageCID string + + // PermissionGrantIDs invokes a grant set; nil means owner-authored. + PermissionGrantIDs []string + + // MessageTimestamp overrides the descriptor timestamp (see + // MessagesQueryOptions.MessageTimestamp). + MessageTimestamp string +} + +// BuildMessagesRead constructs and signs a MessagesRead message. +func BuildMessagesRead(s *Signer, opts MessagesReadOptions) (*Message, error) { + if _, err := gocid.Decode(opts.MessageCID); err != nil { + return nil, fmt.Errorf("%q is not a valid CID: %w", opts.MessageCID, err) + } + canonicalIDs, err := CanonicalPermissionGrantIDs(opts.PermissionGrantIDs) + if err != nil { + return nil, err + } + + timestamp := opts.MessageTimestamp + if timestamp == "" { + timestamp = Now() + } + + desc := map[string]any{ + "interface": "Messages", + "method": "Read", + "messageTimestamp": timestamp, + "messageCid": opts.MessageCID, + } + + return signMessagesMessage(s, desc, canonicalIDs) +} + +// MessagesSubscribeOptions configures a MessagesSubscribe message. +type MessagesSubscribeOptions struct { + // Filters selects the feed scope; an empty list subscribes to the whole + // feed. The filters field is always present on the wire. + Filters []MessagesFilter + + // PermissionGrantIDs invokes a grant set; nil means owner-authored. + PermissionGrantIDs []string + + // Cursor resumes from this progress token: stored events after it are + // replayed, then an EOSE marker, then live events. + Cursor *ProgressToken + + // MessageTimestamp overrides the descriptor timestamp (see + // MessagesQueryOptions.MessageTimestamp). + MessageTimestamp string +} + +// BuildMessagesSubscribe constructs and signs a MessagesSubscribe message. +func BuildMessagesSubscribe(s *Signer, opts MessagesSubscribeOptions) (*Message, error) { + canonicalIDs, err := CanonicalPermissionGrantIDs(opts.PermissionGrantIDs) + if err != nil { + return nil, err + } + filters, err := messagesFiltersToWire(opts.Filters) + if err != nil { + return nil, err + } + + timestamp := opts.MessageTimestamp + if timestamp == "" { + timestamp = Now() + } + + desc := map[string]any{ + "interface": "Messages", + "method": "Subscribe", + "messageTimestamp": timestamp, + "filters": filters, + } + if opts.Cursor != nil { + if !opts.Cursor.valid() { + return nil, fmt.Errorf("invalid progress token cursor") + } + desc["cursor"] = progressTokenToMap(opts.Cursor) + } + + return signMessagesMessage(s, desc, canonicalIDs) +} + +// +// --- Client methods --- +// + +// ProgressGapError reports a 410 Gone reply: the supplied progress token can +// no longer be resumed against the remote replication log. +type ProgressGapError struct { + Info *ProgressGapInfo +} + +func (e *ProgressGapError) Error() string { + if e == nil || e.Info == nil { + return "progress gap" + } + return fmt.Sprintf("progress gap: %s", e.Info.Reason) +} + +// parseProgressGapReply extracts the structured gap metadata from a 410 reply. +func parseProgressGapReply(reply *DwnReply) *ProgressGapError { + gap := &ProgressGapInfo{} + if len(reply.Error) > 0 { + _ = json.Unmarshal(reply.Error, gap) + } + return &ProgressGapError{Info: gap} +} + +// MessagesQueryEntry is one entry of a MessagesQuery reply. Message is empty +// when the query was cidsOnly. +type MessagesQueryEntry struct { + Seq string `json:"seq"` + MessageCID string `json:"messageCid"` + IsLatestBaseState bool `json:"isLatestBaseState"` + Protocol string `json:"protocol,omitempty"` + Message json.RawMessage `json:"message,omitempty"` + EncodedData string `json:"encodedData,omitempty"` +} + +// MessagesQueryResult is the parsed reply of a MessagesQuery. +type MessagesQueryResult struct { + Entries []MessagesQueryEntry + + // Cursor is the high-water progress token for the next page. + Cursor *ProgressToken + + // Drained reports whether the scan reached the log head. + Drained bool + + // Fingerprint is the server's set fingerprint over the queried scope + // (64 lowercase hex chars). Present only for the empty filter set or a + // pure protocol-only filter set over non-core protocols. + Fingerprint string +} + +// MessagesQuery pages the target tenant's replication-log feed. +// +// A 410 reply returns a *ProgressGapError carrying the structured gap +// metadata so callers can reset their checkpoint and re-baseline. +func (c *Client) MessagesQuery(ctx context.Context, target string, opts MessagesQueryOptions) (*MessagesQueryResult, error) { + msg, err := BuildMessagesQuery(c.signer, opts) + if err != nil { + return nil, fmt.Errorf("building MessagesQuery: %w", err) + } + + result, err := c.transport.Send(ctx, target, msg, nil) + if err != nil { + return nil, err + } + reply := result.Reply + if reply == nil { + return nil, fmt.Errorf("messages query: missing reply") + } + switch reply.Status.Code { + case http.StatusOK: + case http.StatusGone: + return nil, parseProgressGapReply(reply) + case http.StatusTooManyRequests: + return nil, queryRateLimitError(reply.Status) + default: + return nil, fmt.Errorf("messages query failed: %d %s", reply.Status.Code, reply.Status.Detail) + } + + queryResult := &MessagesQueryResult{ + Drained: reply.Drained, + Fingerprint: reply.Fingerprint, + } + if len(reply.Entries) > 0 { + if err := json.Unmarshal(reply.Entries, &queryResult.Entries); err != nil { + return nil, fmt.Errorf("unmarshaling messages query entries: %w", err) + } + } + if len(reply.Cursor) > 0 { + cursor := &ProgressToken{} + if err := json.Unmarshal(reply.Cursor, cursor); err != nil { + return nil, fmt.Errorf("unmarshaling messages query cursor: %w", err) + } + queryResult.Cursor = cursor + } + return queryResult, nil +} + +// MessagesReadResult is the parsed reply of a MessagesRead. +type MessagesReadResult struct { + // MessageCID is the CID of the returned message. + MessageCID string + + // Message is the stored message, verbatim. + Message json.RawMessage + + // Data is the record data when the server returned any (RecordsWrite + // messages with data). It rides the dwn-response header/body transport + // split exactly like RecordsRead. + Data []byte +} + +// MessagesRead fetches a single message (and its data, if any) by CID from +// the target tenant's DWN. A 404 reply returns an error wrapping +// ErrMessagesEntryNotFound. +func (c *Client) MessagesRead(ctx context.Context, target string, messageCid string, permissionGrantIDs []string) (*MessagesReadResult, error) { + msg, err := BuildMessagesRead(c.signer, MessagesReadOptions{ + MessageCID: messageCid, + PermissionGrantIDs: permissionGrantIDs, + }) + if err != nil { + return nil, fmt.Errorf("building MessagesRead: %w", err) + } + + result, err := c.transport.Send(ctx, target, msg, nil) + if err != nil { + return nil, err + } + reply := result.Reply + if reply == nil { + return nil, fmt.Errorf("messages read: missing reply") + } + switch reply.Status.Code { + case http.StatusOK: + case http.StatusNotFound: + return nil, fmt.Errorf("messages read %s: %w", messageCid, ErrMessagesEntryNotFound) + case http.StatusTooManyRequests: + return nil, queryRateLimitError(reply.Status) + default: + return nil, fmt.Errorf("messages read failed: %d %s", reply.Status.Code, reply.Status.Detail) + } + + var entry struct { + MessageCID string `json:"messageCid"` + Message json.RawMessage `json:"message"` + } + if len(reply.Entry) == 0 { + return nil, fmt.Errorf("messages read: missing entry") + } + if err := json.Unmarshal(reply.Entry, &entry); err != nil { + return nil, fmt.Errorf("unmarshaling messages read entry: %w", err) + } + + return &MessagesReadResult{ + MessageCID: entry.MessageCID, + Message: entry.Message, + Data: result.Data, + }, nil +} diff --git a/internal/dwn/messages_subscribe_test.go b/internal/dwn/messages_subscribe_test.go new file mode 100644 index 0000000..244cdfe --- /dev/null +++ b/internal/dwn/messages_subscribe_test.go @@ -0,0 +1,372 @@ +package dwn + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "reflect" + "testing" + "time" + + "github.com/coder/websocket" +) + +// writeMessagesSubscribeReply confirms a subscription and includes the +// MessagesSubscribe reply feed snapshot (head + fingerprint). +func writeMessagesSubscribeReply(ctx context.Context, conn *websocket.Conn, requestID, sdkSubscriptionID string, head *ProgressToken, fingerprint string) error { + response := JsonRpcResponse{ + JSONRPC: "2.0", + ID: requestID, + Result: &JsonRpcResult{ + Reply: &DwnReply{ + Status: Status{Code: http.StatusOK, Detail: "OK"}, + Head: head, + Fingerprint: fingerprint, + Subscription: &SubscriptionConfirm{ID: sdkSubscriptionID}, + }, + }, + } + return writeWSJSON(ctx, conn, response) +} + +// verifyMessagesSubscribeRequest checks the signed MessagesSubscribe wire +// shape: descriptor fields plus the canonical permissionGrantIds mirrored in +// descriptor AND signature payload. +func verifyMessagesSubscribeRequest(request *JsonRpcRequest, wantFilters []map[string]any, wantGrantIDs []string, wantCursor *ProgressToken) error { + desc := request.Params.Message.Descriptor + if desc["interface"] != "Messages" || desc["method"] != "Subscribe" { + return fmt.Errorf("descriptor interface/method = %v/%v, want Messages/Subscribe", desc["interface"], desc["method"]) + } + + filtersJSON, err := json.Marshal(desc["filters"]) + if err != nil { + return fmt.Errorf("marshal filters: %w", err) + } + var gotFilters []map[string]any + if err := json.Unmarshal(filtersJSON, &gotFilters); err != nil { + return fmt.Errorf("parse filters: %w", err) + } + if gotFilters == nil { + return fmt.Errorf("filters must always be present, got null") + } + if !reflect.DeepEqual(gotFilters, wantFilters) { + return fmt.Errorf("filters = %v, want %v", gotFilters, wantFilters) + } + + var gotDescriptorIDs []string + if raw, ok := desc["permissionGrantIds"]; ok { + idsJSON, err := json.Marshal(raw) + if err != nil { + return fmt.Errorf("marshal permissionGrantIds: %w", err) + } + if err := json.Unmarshal(idsJSON, &gotDescriptorIDs); err != nil { + return fmt.Errorf("parse permissionGrantIds: %w", err) + } + } + if !reflect.DeepEqual(gotDescriptorIDs, wantGrantIDs) { + return fmt.Errorf("descriptor permissionGrantIds = %v, want %v", gotDescriptorIDs, wantGrantIDs) + } + + payloadBytes, err := base64.RawURLEncoding.DecodeString(request.Params.Message.Authorization.Signature.Payload) + if err != nil { + return fmt.Errorf("decode signature payload: %w", err) + } + var payload struct { + DescriptorCid string `json:"descriptorCid"` + PermissionGrantIDs []string `json:"permissionGrantIds"` + } + if err := json.Unmarshal(payloadBytes, &payload); err != nil { + return fmt.Errorf("parse signature payload: %w", err) + } + if !reflect.DeepEqual(payload.PermissionGrantIDs, wantGrantIDs) { + return fmt.Errorf("payload permissionGrantIds = %v, want %v", payload.PermissionGrantIDs, wantGrantIDs) + } + if payload.DescriptorCid == "" { + return fmt.Errorf("payload missing descriptorCid") + } + + if wantCursor == nil { + if _, ok := desc["cursor"]; ok { + return fmt.Errorf("fresh subscription unexpectedly carried cursor") + } + return nil + } + return expectDescriptorCursor(request, *wantCursor) +} + +func TestMessagesSubscribeHandshakeAndFeedSnapshot(t *testing.T) { + serverErrors := make(chan error, 8) + serverDone := make(chan struct{}) + + resumeCursor := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "10", MessageCID: "cid-10"} + head := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "12"} + wantFilters := []map[string]any{ + {"protocol": "https://example.com/protocols/mesh"}, + {"interface": "Records", "method": "Write", "protocolPathPrefix": "network/node"}, + } + wantGrantIDs := []string{"grant-a", "grant-b"} + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(serverDone) + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- fmt.Errorf("accept: %w", err) + return + } + defer conn.CloseNow() + + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + if err := verifyMessagesSubscribeRequest(request, wantFilters, wantGrantIDs, &resumeCursor); err != nil { + serverErrors <- err + return + } + if err := writeMessagesSubscribeReply(ctx, conn, request.ID, "messages-subscribe-cid", &head, "ab12"); err != nil { + serverErrors <- err + return + } + + // One replayed event, then EOSE. + eventToken := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "11", MessageCID: "cid-11"} + event := &SubscriptionMessage{ + Type: SubscriptionEventType, + Cursor: &eventToken, + Seq: "11", + MessageCID: "cid-11", + Protocol: "https://example.com/protocols/mesh", + EncodedData: "ZGF0YQ", + Event: &RecordEvent{ + Message: json.RawMessage(`{"descriptor":{"interface":"Records","method":"Write"}}`), + }, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, event); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, eventToken); err != nil { + serverErrors <- err + return + } + eoseToken := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "12"} + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, &SubscriptionMessage{Type: SubscriptionEOSEType, Cursor: &eoseToken}); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, eoseToken); err != nil { + serverErrors <- err + return + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + messages := make(chan SubscriptionMessage, 4) + lifecycle := make(chan SubscriptionLifecycleEvent, 4) + manager := NewSubscriptionManager(server.URL, slog.Default()) + sub, err := manager.SubscribeMessagesWithLifecycle( + ctx, + "did:dht:tenant", + newTestSigner(t), + MessagesSubscribeOptions{ + Filters: []MessagesFilter{ + {Protocol: "https://example.com/protocols/mesh"}, + {Interface: "Records", Method: "Write", ProtocolPathPrefix: "network/node"}, + }, + // Unsorted with duplicates: the wire message must carry the + // canonical list. + PermissionGrantIDs: []string{"grant-b", "grant-a", "grant-b"}, + Cursor: &resumeCursor, + }, + func(message *SubscriptionMessage) error { + copyMessage := *message + copyMessage.Cursor = message.Cursor.clone() + messages <- copyMessage + return nil + }, + func(event SubscriptionLifecycleEvent) { + lifecycle <- event + }, + ) + if err != nil { + t.Fatalf("SubscribeMessagesWithLifecycle: %v", err) + } + defer sub.Close() + + var established SubscriptionLifecycleEvent + select { + case established = <-lifecycle: + case <-ctx.Done(): + t.Fatalf("waiting for established: %v", ctx.Err()) + } + if established.Kind != SubscriptionLifecycleEstablished { + t.Fatalf("first lifecycle = %+v, want established", established) + } + if established.NeedsFullRefresh { + t.Error("cursor-seeded establishment must not require full refresh") + } + if established.Head == nil || *established.Head != head { + t.Errorf("established head = %+v, want %+v", established.Head, head) + } + if established.Fingerprint != "ab12" { + t.Errorf("established fingerprint = %q, want ab12", established.Fingerprint) + } + + gotMessages := make([]SubscriptionMessage, 0, 2) + for len(gotMessages) < 2 { + select { + case message := <-messages: + gotMessages = append(gotMessages, message) + case <-ctx.Done(): + t.Fatalf("waiting for messages: %v; got %d", ctx.Err(), len(gotMessages)) + } + } + if gotMessages[0].Type != SubscriptionEventType || gotMessages[0].Seq != "11" || + gotMessages[0].MessageCID != "cid-11" || gotMessages[0].EncodedData != "ZGF0YQ" { + t.Errorf("first message = %+v, want replay event with feed metadata", gotMessages[0]) + } + if gotMessages[1].Type != SubscriptionEOSEType { + t.Errorf("second message = %+v, want eose", gotMessages[1]) + } + + select { + case <-serverDone: + case <-ctx.Done(): + t.Fatalf("waiting for server: %v", ctx.Err()) + } + drainServerErrors(t, serverErrors) + + if cursor := sub.Cursor(); cursor == nil || cursor.Position != "12" { + t.Errorf("Cursor() = %+v, want EOSE cursor", cursor) + } +} + +func TestMessagesSubscribeTerminalDeliveryErrorCodes(t *testing.T) { + tests := []struct { + code string + wantRevoked bool + wantTerminal bool + }{ + {code: MessagesSubscribeDeliveryAuthorizationFailedCode, wantRevoked: true, wantTerminal: true}, + {code: MessagesSubscribeDeliveryFailedCode, wantRevoked: true, wantTerminal: true}, + {code: "SomeOtherTerminalCode", wantRevoked: false, wantTerminal: true}, + } + + for _, tc := range tests { + t.Run(tc.code, func(t *testing.T) { + serverErrors := make(chan error, 4) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + serverErrors <- fmt.Errorf("accept: %w", err) + return + } + defer conn.CloseNow() + + ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) + defer cancel() + + request, err := readWSRequest(ctx, conn) + if err != nil { + serverErrors <- err + return + } + if err := writeMessagesSubscribeReply(ctx, conn, request.ID, "sub-cid", nil, ""); err != nil { + serverErrors <- err + return + } + token := ProgressToken{StreamID: "stream", Epoch: "epoch", Position: "1"} + terminal := &SubscriptionMessage{ + Type: SubscriptionErrorType, + Cursor: &token, + Error: &SubscriptionError{Code: tc.code, Detail: "delivery stopped"}, + } + if err := writeSubscriptionMessage(ctx, conn, request.Subscription.ID, terminal); err != nil { + serverErrors <- err + return + } + if err := expectAck(ctx, conn, request.Subscription.ID, token); err != nil { + serverErrors <- err + } + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + lifecycle := make(chan SubscriptionLifecycleEvent, 4) + manager := NewSubscriptionManager(server.URL, slog.Default()) + sub, err := manager.SubscribeMessagesWithLifecycle( + ctx, + "did:dht:tenant", + newTestSigner(t), + MessagesSubscribeOptions{}, + func(*SubscriptionMessage) error { return nil }, + func(event SubscriptionLifecycleEvent) { lifecycle <- event }, + ) + if err != nil { + t.Fatalf("SubscribeMessagesWithLifecycle: %v", err) + } + defer sub.Close() + + var terminalEvent SubscriptionLifecycleEvent + terminalSeen := false + for !terminalSeen { + select { + case event := <-lifecycle: + if event.Kind == SubscriptionLifecycleTerminal { + terminalEvent = event + terminalSeen = true + } + case <-ctx.Done(): + t.Fatalf("waiting for terminal lifecycle: %v", ctx.Err()) + } + } + if !tc.wantTerminal { + t.Fatal("test table only models terminal cases") + } + if got := errors.Is(terminalEvent.Err, ErrSubscriptionAuthorizationRevoked); got != tc.wantRevoked { + t.Errorf("errors.Is(err, ErrSubscriptionAuthorizationRevoked) = %v, want %v (err = %v)", + got, tc.wantRevoked, terminalEvent.Err) + } + drainServerErrors(t, serverErrors) + }) + } +} + +func TestMessagesSubscribeValidatesOptions(t *testing.T) { + manager := NewSubscriptionManager("http://127.0.0.1:0", slog.Default()) + ctx := context.Background() + signer := newTestSigner(t) + handler := func(*SubscriptionMessage) error { return nil } + + if _, err := manager.SubscribeMessages(ctx, "did:dht:tenant", signer, MessagesSubscribeOptions{ + PermissionGrantIDs: []string{}, + }, handler); !errors.Is(err, ErrPermissionGrantIDsEmpty) { + t.Errorf("empty grant list error = %v, want ErrPermissionGrantIDsEmpty", err) + } + + if _, err := manager.SubscribeMessages(ctx, "did:dht:tenant", signer, MessagesSubscribeOptions{ + Filters: []MessagesFilter{{}}, + }, handler); err == nil { + t.Error("empty filter must be rejected at subscribe time") + } + + if _, err := manager.SubscribeMessages(ctx, "did:dht:tenant", signer, MessagesSubscribeOptions{ + Cursor: &ProgressToken{StreamID: "s"}, + }, handler); err == nil { + t.Error("invalid cursor must be rejected at subscribe time") + } +} diff --git a/internal/dwn/messages_test.go b/internal/dwn/messages_test.go new file mode 100644 index 0000000..9c4cedf --- /dev/null +++ b/internal/dwn/messages_test.go @@ -0,0 +1,559 @@ +package dwn + +import ( + "context" + "crypto/ed25519" + "encoding/base64" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "os" + "reflect" + "testing" +) + +// +// --- Golden-fixture parity with the TypeScript implementation --- +// +// Fixtures under testdata/ were generated by executing the ACTUAL enbox +// TypeScript implementation (dwn-sdk-js + agent sources at +// ~/src/enboxorg/enbox) with a fixed Ed25519 key and fixed timestamps: +// +// bun scratchpad/fixtures/generate.ts internal/dwn/testdata +// +// (script kept in the session scratchpad; the generatedBy field in each +// fixture file records the provenance). +// +// The parity assertions cover: byte-identical descriptors (deep JSON +// equality AND equal descriptor CIDs), byte-identical signature payloads, +// and equal CIDs computed over the full TS-signed message. Full-JWS byte +// equality is NOT asserted: dwn-sdk-js serializes the protected header as +// {"kid","alg"} while meshd emits {"alg","kid"} — both are valid JWS and the +// server never compares header bytes, but the signatures differ. +// + +type messagesWireFixtureFile struct { + GeneratedBy string `json:"generatedBy"` + DID string `json:"did"` + Ed25519SeedBase64URL string `json:"ed25519SeedBase64Url"` + Fixtures []messagesWireFixture `json:"fixtures"` +} + +type messagesWireFixture struct { + Name string `json:"name"` + Inputs json.RawMessage `json:"inputs"` + Message json.RawMessage `json:"message"` + Descriptor json.RawMessage `json:"descriptor"` + SignaturePayloadBase64URL string `json:"signaturePayloadBase64Url"` + DescriptorCid string `json:"descriptorCid"` + MessageCid string `json:"messageCid"` +} + +type messagesWireFixtureInputs struct { + MessageTimestamp string `json:"messageTimestamp"` + Filters []MessagesFilter `json:"filters"` + PermissionGrantIDs []string `json:"permissionGrantIds"` + Cursor *ProgressToken `json:"cursor"` + Limit int `json:"limit"` + CidsOnly bool `json:"cidsOnly"` + MessageCid string `json:"messageCid"` +} + +func loadMessagesWireFixtures(t *testing.T) (*Signer, []messagesWireFixture) { + t.Helper() + data, err := os.ReadFile("testdata/messages_wire_fixtures.json") + if err != nil { + t.Fatalf("reading fixtures: %v", err) + } + var file messagesWireFixtureFile + if err := json.Unmarshal(data, &file); err != nil { + t.Fatalf("parsing fixtures: %v", err) + } + seed, err := base64.RawURLEncoding.DecodeString(file.Ed25519SeedBase64URL) + if err != nil { + t.Fatalf("decoding fixture seed: %v", err) + } + signer := &Signer{ + DID: file.DID, + PrivateKey: ed25519.NewKeyFromSeed(seed), + } + return signer, file.Fixtures +} + +func TestMessagesBuildersMatchTypeScriptFixtures(t *testing.T) { + signer, fixtures := loadMessagesWireFixtures(t) + if len(fixtures) == 0 { + t.Fatal("no fixtures loaded") + } + + for _, fixture := range fixtures { + t.Run(fixture.Name, func(t *testing.T) { + var inputs messagesWireFixtureInputs + if err := json.Unmarshal(fixture.Inputs, &inputs); err != nil { + t.Fatalf("parsing inputs: %v", err) + } + + var ( + msg *Message + err error + ) + var fixtureMethod struct { + Descriptor struct { + Method string `json:"method"` + } `json:"descriptor"` + } + if err := json.Unmarshal(fixture.Message, &fixtureMethod); err != nil { + t.Fatalf("parsing fixture message method: %v", err) + } + switch fixtureMethod.Descriptor.Method { + case "Query": + msg, err = BuildMessagesQuery(signer, MessagesQueryOptions{ + Filters: inputs.Filters, + PermissionGrantIDs: inputs.PermissionGrantIDs, + Cursor: inputs.Cursor, + Limit: inputs.Limit, + CidsOnly: inputs.CidsOnly, + MessageTimestamp: inputs.MessageTimestamp, + }) + case "Read": + msg, err = BuildMessagesRead(signer, MessagesReadOptions{ + MessageCID: inputs.MessageCid, + PermissionGrantIDs: inputs.PermissionGrantIDs, + MessageTimestamp: inputs.MessageTimestamp, + }) + case "Subscribe": + msg, err = BuildMessagesSubscribe(signer, MessagesSubscribeOptions{ + Filters: inputs.Filters, + PermissionGrantIDs: inputs.PermissionGrantIDs, + Cursor: inputs.Cursor, + MessageTimestamp: inputs.MessageTimestamp, + }) + default: + t.Fatalf("unknown fixture method %q", fixtureMethod.Descriptor.Method) + } + if err != nil { + t.Fatalf("building message: %v", err) + } + + // 1. Descriptors must be identical (deep JSON equality). + var wantDescriptor any + if err := json.Unmarshal(fixture.Descriptor, &wantDescriptor); err != nil { + t.Fatalf("parsing fixture descriptor: %v", err) + } + gotDescriptorJSON, err := json.Marshal(msg.Descriptor) + if err != nil { + t.Fatalf("marshaling built descriptor: %v", err) + } + var gotDescriptor any + if err := json.Unmarshal(gotDescriptorJSON, &gotDescriptor); err != nil { + t.Fatalf("round-tripping built descriptor: %v", err) + } + if !reflect.DeepEqual(gotDescriptor, wantDescriptor) { + t.Errorf("descriptor mismatch:\n got: %s\nwant: %s", gotDescriptorJSON, fixture.Descriptor) + } + + // 2. Signature payloads must be byte-identical. + if got := msg.Authorization.Signature.Payload; got != fixture.SignaturePayloadBase64URL { + gotBytes, _ := base64.RawURLEncoding.DecodeString(got) + wantBytes, _ := base64.RawURLEncoding.DecodeString(fixture.SignaturePayloadBase64URL) + t.Errorf("signature payload mismatch:\n got: %s\nwant: %s", gotBytes, wantBytes) + } + + // 3. The Go descriptor CID must match the TS descriptor CID. + gotDescriptorCid, err := ComputeCID(msg.Descriptor) + if err != nil { + t.Fatalf("computing descriptor CID: %v", err) + } + if gotDescriptorCid != fixture.DescriptorCid { + t.Errorf("descriptor CID = %s, want %s", gotDescriptorCid, fixture.DescriptorCid) + } + + // 4. Go's CID discipline must reproduce the TS message CID over + // the TS-signed message (dag-cbor, CIDv1, sha2-256). + var fixtureMessage map[string]any + if err := json.Unmarshal(fixture.Message, &fixtureMessage); err != nil { + t.Fatalf("parsing fixture message: %v", err) + } + gotMessageCid, err := ComputeCID(fixtureMessage) + if err != nil { + t.Fatalf("computing message CID: %v", err) + } + if gotMessageCid != fixture.MessageCid { + t.Errorf("message CID over TS message = %s, want %s", gotMessageCid, fixture.MessageCid) + } + }) + } +} + +type grantIDsFixtureFile struct { + Vectors []struct { + Name string `json:"name"` + Input struct { + PermissionGrantID string `json:"permissionGrantId"` + PermissionGrantIDs []string `json:"permissionGrantIds"` + } `json:"input"` + Canonical []string `json:"canonical"` + ErrorCode string `json:"errorCode"` + } `json:"vectors"` +} + +func TestPermissionGrantIDCanonicalizationMatchesTypeScriptFixtures(t *testing.T) { + data, err := os.ReadFile("testdata/grant_ids_fixtures.json") + if err != nil { + t.Fatalf("reading fixtures: %v", err) + } + var file grantIDsFixtureFile + if err := json.Unmarshal(data, &file); err != nil { + t.Fatalf("parsing fixtures: %v", err) + } + if len(file.Vectors) == 0 { + t.Fatal("no vectors loaded") + } + + wantErrs := map[string]error{ + "MessagePermissionGrantIdsEmpty": ErrPermissionGrantIDsEmpty, + "MessagePermissionGrantCreateInvocationAmbiguous": ErrPermissionGrantInvocationAmbiguous, + } + + for _, vector := range file.Vectors { + t.Run(vector.Name, func(t *testing.T) { + canonical, err := NormalizePermissionGrantInvocation(vector.Input.PermissionGrantID, vector.Input.PermissionGrantIDs) + if vector.ErrorCode != "" { + wantErr, ok := wantErrs[vector.ErrorCode] + if !ok { + t.Fatalf("fixture uses unmapped error code %q", vector.ErrorCode) + } + if !errors.Is(err, wantErr) { + t.Fatalf("error = %v, want %v", err, wantErr) + } + return + } + if err != nil { + t.Fatalf("NormalizePermissionGrantInvocation: %v", err) + } + if !reflect.DeepEqual(canonical, vector.Canonical) { + t.Errorf("canonical = %q, want %q", canonical, vector.Canonical) + } + }) + } +} + +func TestValidateCanonicalPermissionGrantIDs(t *testing.T) { + if err := ValidateCanonicalPermissionGrantIDs(nil); err != nil { + t.Errorf("nil list: %v", err) + } + if err := ValidateCanonicalPermissionGrantIDs([]string{"a", "b"}); err != nil { + t.Errorf("canonical list: %v", err) + } + if err := ValidateCanonicalPermissionGrantIDs([]string{"b", "a"}); !errors.Is(err, ErrPermissionGrantIDsNotCanonical) { + t.Errorf("unsorted list error = %v, want ErrPermissionGrantIDsNotCanonical", err) + } + if err := ValidateCanonicalPermissionGrantIDs([]string{"a", "a"}); !errors.Is(err, ErrPermissionGrantIDsNotCanonical) { + t.Errorf("duplicated list error = %v, want ErrPermissionGrantIDsNotCanonical", err) + } + if err := ValidateCanonicalPermissionGrantIDs([]string{}); !errors.Is(err, ErrPermissionGrantIDsEmpty) { + t.Errorf("empty list error = %v, want ErrPermissionGrantIDsEmpty", err) + } +} + +// +// --- Builder unit tests --- +// + +func TestBuildMessagesQueryDescriptorShape(t *testing.T) { + signer := newTestSigner(t) + + t.Run("filters always present", func(t *testing.T) { + msg, err := BuildMessagesQuery(signer, MessagesQueryOptions{}) + if err != nil { + t.Fatalf("BuildMessagesQuery: %v", err) + } + filters, ok := msg.Descriptor["filters"].([]any) + if !ok || len(filters) != 0 { + t.Fatalf("filters = %#v, want empty array", msg.Descriptor["filters"]) + } + wire, err := json.Marshal(msg.Descriptor) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var onWire map[string]json.RawMessage + if err := json.Unmarshal(wire, &onWire); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if string(onWire["filters"]) != "[]" { + t.Errorf("wire filters = %s, want []", onWire["filters"]) + } + for _, absent := range []string{"limit", "cidsOnly", "cursor", "permissionGrantIds"} { + if _, ok := onWire[absent]; ok { + t.Errorf("zero-value field %q must be omitted", absent) + } + } + }) + + t.Run("grant IDs canonicalized in descriptor and payload", func(t *testing.T) { + msg, err := BuildMessagesQuery(signer, MessagesQueryOptions{ + PermissionGrantIDs: []string{"b", "a", "b"}, + }) + if err != nil { + t.Fatalf("BuildMessagesQuery: %v", err) + } + want := []string{"a", "b"} + got, ok := msg.Descriptor["permissionGrantIds"].([]string) + if !ok || !reflect.DeepEqual(got, want) { + t.Errorf("descriptor permissionGrantIds = %#v, want %v", msg.Descriptor["permissionGrantIds"], want) + } + payloadBytes, err := base64.RawURLEncoding.DecodeString(msg.Authorization.Signature.Payload) + if err != nil { + t.Fatalf("decoding payload: %v", err) + } + var payload struct { + DescriptorCid string `json:"descriptorCid"` + PermissionGrantIDs []string `json:"permissionGrantIds"` + } + if err := json.Unmarshal(payloadBytes, &payload); err != nil { + t.Fatalf("parsing payload: %v", err) + } + if !reflect.DeepEqual(payload.PermissionGrantIDs, want) { + t.Errorf("payload permissionGrantIds = %v, want %v", payload.PermissionGrantIDs, want) + } + wantCid, err := ComputeCID(msg.Descriptor) + if err != nil { + t.Fatalf("ComputeCID: %v", err) + } + if payload.DescriptorCid != wantCid { + t.Errorf("payload descriptorCid = %s, want %s", payload.DescriptorCid, wantCid) + } + }) + + t.Run("explicit empty grant list rejected", func(t *testing.T) { + _, err := BuildMessagesQuery(signer, MessagesQueryOptions{PermissionGrantIDs: []string{}}) + if !errors.Is(err, ErrPermissionGrantIDsEmpty) { + t.Errorf("error = %v, want ErrPermissionGrantIDsEmpty", err) + } + }) + + t.Run("empty filter rejected", func(t *testing.T) { + if _, err := BuildMessagesQuery(signer, MessagesQueryOptions{Filters: []MessagesFilter{{}}}); err == nil { + t.Error("empty filter must be rejected (schema minProperties: 1)") + } + if _, err := BuildMessagesQuery(signer, MessagesQueryOptions{ + Filters: []MessagesFilter{{MessageTimestamp: &MessagesTimestampRange{}}}, + }); err == nil { + t.Error("empty messageTimestamp range must be rejected") + } + }) + + t.Run("invalid cursor rejected", func(t *testing.T) { + if _, err := BuildMessagesQuery(signer, MessagesQueryOptions{Cursor: &ProgressToken{StreamID: "s"}}); err == nil { + t.Error("invalid cursor must be rejected") + } + }) +} + +func TestBuildMessagesReadValidatesCID(t *testing.T) { + signer := newTestSigner(t) + if _, err := BuildMessagesRead(signer, MessagesReadOptions{MessageCID: "not-a-cid"}); err == nil { + t.Error("invalid CID must be rejected") + } +} + +// +// --- Client method tests --- +// + +// messagesTestServer returns a Client pointed at an httptest server that +// responds to each request with the given responder. +func messagesTestServer(t *testing.T, responder func(w http.ResponseWriter, req *JsonRpcRequest)) *Client { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var rpcReq JsonRpcRequest + if err := json.Unmarshal([]byte(r.Header.Get("dwn-request")), &rpcReq); err != nil { + t.Errorf("parsing dwn-request header: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + responder(w, &rpcReq) + })) + t.Cleanup(server.Close) + return NewClient(server.URL, newTestSigner(t)) +} + +func writeJSONRPCReply(w http.ResponseWriter, id string, reply any) { + response := map[string]any{ + "jsonrpc": "2.0", + "id": id, + "result": map[string]any{"reply": reply}, + } + _ = json.NewEncoder(w).Encode(response) +} + +func TestClientMessagesQueryParsesReply(t *testing.T) { + client := messagesTestServer(t, func(w http.ResponseWriter, req *JsonRpcRequest) { + desc := req.Params.Message.Descriptor + if desc["interface"] != "Messages" || desc["method"] != "Query" { + t.Errorf("descriptor = %v, want Messages/Query", desc) + } + writeJSONRPCReply(w, req.ID, map[string]any{ + "status": map[string]any{"code": 200, "detail": "OK"}, + "entries": []map[string]any{ + { + "seq": "7", + "messageCid": "bafycid1", + "isLatestBaseState": true, + "protocol": "https://example.com/protocols/mesh", + "message": map[string]any{"descriptor": map[string]any{"interface": "Records"}}, + "encodedData": "aGVsbG8", + }, + { + // cidsOnly-style entry: message omitted. + "seq": "8", + "messageCid": "bafycid2", + "isLatestBaseState": false, + }, + }, + "cursor": map[string]any{"streamId": "s1", "epoch": "e1", "position": "8", "messageCid": "bafycid2"}, + "drained": true, + "fingerprint": "00ff", + }) + }) + + result, err := client.MessagesQuery(context.Background(), "did:dht:tenant", MessagesQueryOptions{ + Filters: []MessagesFilter{{Protocol: "https://example.com/protocols/mesh"}}, + }) + if err != nil { + t.Fatalf("MessagesQuery: %v", err) + } + if len(result.Entries) != 2 { + t.Fatalf("entries = %d, want 2", len(result.Entries)) + } + first := result.Entries[0] + if first.Seq != "7" || first.MessageCID != "bafycid1" || !first.IsLatestBaseState || + first.Protocol != "https://example.com/protocols/mesh" || first.EncodedData != "aGVsbG8" || len(first.Message) == 0 { + t.Errorf("first entry = %+v", first) + } + if second := result.Entries[1]; second.Message != nil || second.IsLatestBaseState { + t.Errorf("second entry = %+v, want cids-only shape", second) + } + if result.Cursor == nil || result.Cursor.Position != "8" || result.Cursor.MessageCID != "bafycid2" { + t.Errorf("cursor = %+v", result.Cursor) + } + if !result.Drained || result.Fingerprint != "00ff" { + t.Errorf("drained/fingerprint = %v/%q", result.Drained, result.Fingerprint) + } +} + +func TestClientMessagesQueryProgressGap(t *testing.T) { + client := messagesTestServer(t, func(w http.ResponseWriter, req *JsonRpcRequest) { + writeJSONRPCReply(w, req.ID, map[string]any{ + "status": map[string]any{"code": 410, "detail": "Progress token gap"}, + "error": map[string]any{ + "code": "ProgressGap", + "requested": map[string]any{"streamId": "s1", "epoch": "old", "position": "5"}, + "oldestAvailable": map[string]any{"streamId": "s1", "epoch": "new", "position": "0"}, + "latestAvailable": map[string]any{"streamId": "s1", "epoch": "new", "position": "9"}, + "reason": "epoch_mismatch", + }, + }) + }) + + _, err := client.MessagesQuery(context.Background(), "did:dht:tenant", MessagesQueryOptions{ + Cursor: &ProgressToken{StreamID: "s1", Epoch: "old", Position: "5"}, + }) + var gap *ProgressGapError + if !errors.As(err, &gap) { + t.Fatalf("error = %v, want *ProgressGapError", err) + } + if gap.Info == nil || gap.Info.Reason != "epoch_mismatch" || + gap.Info.Requested.Epoch != "old" || gap.Info.LatestAvailable.Position != "9" { + t.Errorf("gap info = %+v", gap.Info) + } +} + +func TestClientMessagesQueryRateLimited(t *testing.T) { + client := messagesTestServer(t, func(w http.ResponseWriter, req *JsonRpcRequest) { + writeJSONRPCReply(w, req.ID, map[string]any{ + "status": map[string]any{"code": 429, "detail": "throttled, retry after 2s"}, + }) + }) + + _, err := client.MessagesQuery(context.Background(), "did:dht:tenant", MessagesQueryOptions{}) + if !errors.Is(err, ErrRateLimited) { + t.Fatalf("error = %v, want ErrRateLimited", err) + } +} + +func TestClientMessagesReadDataTransportSplit(t *testing.T) { + // Compute a valid CID for the request. + targetCid, err := ComputeCID(map[string]any{"hello": "world"}) + if err != nil { + t.Fatalf("ComputeCID: %v", err) + } + recordData := []byte("record data bytes") + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var rpcReq JsonRpcRequest + if err := json.Unmarshal([]byte(r.Header.Get("dwn-request")), &rpcReq); err != nil { + t.Errorf("parsing dwn-request header: %v", err) + return + } + desc := rpcReq.Params.Message.Descriptor + if desc["interface"] != "Messages" || desc["method"] != "Read" || desc["messageCid"] != targetCid { + t.Errorf("descriptor = %v", desc) + } + if grantIDs, ok := desc["permissionGrantIds"].([]any); !ok || len(grantIDs) != 1 || grantIDs[0] != "grant-1" { + t.Errorf("permissionGrantIds = %v", desc["permissionGrantIds"]) + } + + // Data replies: JSON-RPC envelope in the dwn-response header, raw + // data as the body — same split as RecordsRead. + envelope, _ := json.Marshal(map[string]any{ + "jsonrpc": "2.0", + "id": rpcReq.ID, + "result": map[string]any{ + "reply": map[string]any{ + "status": map[string]any{"code": 200, "detail": "OK"}, + "entry": map[string]any{ + "messageCid": targetCid, + "message": map[string]any{"descriptor": map[string]any{"interface": "Records", "method": "Write"}}, + }, + }, + }, + }) + w.Header().Set("dwn-response", string(envelope)) + _, _ = w.Write(recordData) + })) + defer server.Close() + + client := NewClient(server.URL, newTestSigner(t)) + result, err := client.MessagesRead(context.Background(), "did:dht:tenant", targetCid, []string{"grant-1"}) + if err != nil { + t.Fatalf("MessagesRead: %v", err) + } + if result.MessageCID != targetCid { + t.Errorf("messageCid = %s, want %s", result.MessageCID, targetCid) + } + if len(result.Message) == 0 { + t.Error("message missing") + } + if string(result.Data) != string(recordData) { + t.Errorf("data = %q, want %q", result.Data, recordData) + } +} + +func TestClientMessagesReadNotFound(t *testing.T) { + targetCid, err := ComputeCID(map[string]any{"missing": true}) + if err != nil { + t.Fatalf("ComputeCID: %v", err) + } + client := messagesTestServer(t, func(w http.ResponseWriter, req *JsonRpcRequest) { + writeJSONRPCReply(w, req.ID, map[string]any{ + "status": map[string]any{"code": 404, "detail": "Not Found"}, + }) + }) + + _, err = client.MessagesRead(context.Background(), "did:dht:tenant", targetCid, nil) + if !errors.Is(err, ErrMessagesEntryNotFound) { + t.Fatalf("error = %v, want ErrMessagesEntryNotFound", err) + } +} diff --git a/internal/dwn/permission_grants.go b/internal/dwn/permission_grants.go index 6904bfc..2b9cea9 100644 --- a/internal/dwn/permission_grants.go +++ b/internal/dwn/permission_grants.go @@ -11,6 +11,7 @@ import ( const ( DwnScopeInterfaceRecords = "Records" DwnScopeInterfaceProtocols = "Protocols" + DwnScopeInterfaceMessages = "Messages" DwnScopeMethodRead = "Read" DwnScopeMethodWrite = "Write" @@ -257,3 +258,157 @@ func signerDIDFromJWS(jws *GeneralJWS) (string, error) { } return header.KID, nil } + +// +// --- Feed-grant selection (Messages replication-log scopes) --- +// + +// FeedGrantCoverageError reports that no active grant covers part of a +// requested feed scope. Protocol is empty when the full-tenant scope lacked +// an unscoped Messages/Read grant, otherwise it names the uncovered +// protocol. +type FeedGrantCoverageError struct { + Protocol string +} + +func (e *FeedGrantCoverageError) Error() string { + if e.Protocol == "" { + return "no active unscoped Messages/Read permission grant covers the full feed scope" + } + return fmt.Sprintf("no active protocol-root Messages/Read permission grant covers protocol %s", e.Protocol) +} + +// SelectFeedGrants returns the permission grants a delegate must invoke for +// a Messages feed operation over the given scope, porting the enbox agent's +// resolveMessagesScopes rules: +// +// - Only active Messages/Read grants participate: grantor and grantee +// match, and dateGranted <= now < dateExpires (compared as DWN timestamp +// strings, like the agent). +// - Full scope requires at least one UNSCOPED grant; the invoked set is +// every active unscoped grant. +// - Protocol-set scope requires EVERY protocol to be covered by a +// protocol-root grant (scope.protocol == p with no protocolPath or +// contextId) or an unscoped grant; the invoked set is every active grant +// participating in that root set. +// - Exact protocolPath/contextId grants never authorize feed scopes; they +// cover a strict subset of the messages in those scopes. +// +// The returned grants are sorted by ID (UTF-16 code-unit order). Missing +// coverage returns a *FeedGrantCoverageError naming the uncovered protocol. +// Revoked grants must be excluded by the caller before selection. +func SelectFeedGrants(grants []*PermissionGrant, grantor, grantee string, now time.Time, scope SyncScope) ([]*PermissionGrant, error) { + if now.IsZero() { + now = time.Now() + } + nowTimestamp := Timestamp(now) + + active := make([]*PermissionGrant, 0, len(grants)) + for _, grant := range grants { + if isActiveMessagesFeedGrant(grant, grantor, grantee, nowTimestamp) { + active = append(active, grant) + } + } + + var selected []*PermissionGrant + switch scope.Kind { + case SyncScopeKindFull: + for _, grant := range active { + if grantScopeIsUnscoped(grant.Scope) { + selected = append(selected, grant) + } + } + if len(selected) == 0 { + return nil, &FeedGrantCoverageError{} + } + case SyncScopeKindProtocolSet: + protocols, err := NormalizeSyncProtocols(scope.Protocols) + if err != nil { + return nil, err + } + for _, protocol := range protocols { + if !anyGrantMatchesProtocolRoot(active, protocol) { + return nil, &FeedGrantCoverageError{Protocol: protocol} + } + } + for _, grant := range active { + if grantParticipatesInProtocolSet(grant, protocols) { + selected = append(selected, grant) + } + } + default: + return nil, fmt.Errorf("unknown sync scope kind %q", scope.Kind) + } + + for i := 1; i < len(selected); i++ { + for j := i; j > 0 && compareStringsUTF16(selected[j].ID, selected[j-1].ID) < 0; j-- { + selected[j], selected[j-1] = selected[j-1], selected[j] + } + } + return selected, nil +} + +// FeedGrantIDs extracts the canonical invoked ID list from a feed-grant +// selection, ready for permissionGrantIds invocation. +func FeedGrantIDs(grants []*PermissionGrant) []string { + if len(grants) == 0 { + return nil + } + ids := make([]string, len(grants)) + for i, grant := range grants { + ids[i] = grant.ID + } + sortStringsUTF16(ids) + return ids +} + +// isActiveMessagesFeedGrant mirrors the agent's isActiveMessagesGrant: +// grantor/grantee match, active window dateGranted <= now < dateExpires by +// timestamp-string comparison, and a Messages/Read scope. +func isActiveMessagesFeedGrant(grant *PermissionGrant, grantor, grantee, now string) bool { + if grant == nil { + return false + } + if grant.Grantor != grantor || grant.Grantee != grantee { + return false + } + if grant.DateGranted > now || grant.DateExpires <= now { + return false + } + return grant.Scope.Interface == DwnScopeInterfaceMessages && grant.Scope.Method == DwnScopeMethodRead +} + +// grantScopeIsUnscoped reports whether the scope covers the full tenant feed. +func grantScopeIsUnscoped(scope PermissionScope) bool { + return scope.Protocol == "" && scope.ProtocolPath == "" && scope.ContextID == "" +} + +// grantMatchesProtocolRoot reports whether a grant authorizes the root feed +// of a protocol: unscoped, or scoped to exactly that protocol with no +// protocolPath or contextId narrowing. +func grantMatchesProtocolRoot(grant *PermissionGrant, protocol string) bool { + if grantScopeIsUnscoped(grant.Scope) { + return true + } + return grant.Scope.Protocol == protocol && + grant.Scope.ProtocolPath == "" && + grant.Scope.ContextID == "" +} + +func anyGrantMatchesProtocolRoot(grants []*PermissionGrant, protocol string) bool { + for _, grant := range grants { + if grantMatchesProtocolRoot(grant, protocol) { + return true + } + } + return false +} + +func grantParticipatesInProtocolSet(grant *PermissionGrant, protocols []string) bool { + for _, protocol := range protocols { + if grantMatchesProtocolRoot(grant, protocol) { + return true + } + } + return false +} diff --git a/internal/dwn/permission_scope.go b/internal/dwn/permission_scope.go new file mode 100644 index 0000000..e8d9846 --- /dev/null +++ b/internal/dwn/permission_scope.go @@ -0,0 +1,55 @@ +package dwn + +import "strings" + +// ProtocolScope is the minimal protocol-bound scope shape shared by +// permission grant authorization. Permission grants and candidate +// records/filters are projected down to these fields before matching so +// grant selection and server enforcement use the same rules. +// +// An empty string means "unset" (the SDK's undefined). +type ProtocolScope struct { + Protocol string + ProtocolPath string + ContextID string +} + +// MatchesPermissionScope determines whether a candidate target is within a +// grant scope, porting the SDK's PermissionScopeMatcher.matches exactly. +// +// Matching fails closed for invalid combinations: protocolPath and contextId +// are mutually exclusive, and either field requires protocol. protocol +// matches by exact equality. protocolPath and contextId scopes match +// subtrees: the target value must equal the scoped value or begin with the +// scoped value followed by "/". +func MatchesPermissionScope(scope, target ProtocolScope) bool { + if scope.ProtocolPath != "" && scope.ContextID != "" { + return false + } + + if (scope.ProtocolPath != "" || scope.ContextID != "") && scope.Protocol == "" { + return false + } + + if scope.Protocol != "" && scope.Protocol != target.Protocol { + return false + } + + if scope.ProtocolPath != "" { + return MatchesSubtree(scope.ProtocolPath, target.ProtocolPath) + } + + if scope.ContextID != "" { + return MatchesSubtree(scope.ContextID, target.ContextID) + } + + return true +} + +// MatchesSubtree reports whether candidate is the selected hierarchical path +// or one of its proper "/"-delimited descendants, porting the SDK's +// FilterUtility.matchesSubtree. An unset (empty) candidate never matches. +func MatchesSubtree(subtree, candidate string) bool { + return candidate != "" && + (candidate == subtree || strings.HasPrefix(candidate, subtree+"/")) +} diff --git a/internal/dwn/permission_scope_test.go b/internal/dwn/permission_scope_test.go new file mode 100644 index 0000000..e850deb --- /dev/null +++ b/internal/dwn/permission_scope_test.go @@ -0,0 +1,162 @@ +package dwn + +import "testing" + +// Table tests mirroring dwn-sdk-js PermissionScopeMatcher.matches and +// FilterUtility.matchesSubtree. +func TestMatchesPermissionScope(t *testing.T) { + const ( + mesh = "https://example.com/protocols/mesh" + other = "https://example.com/protocols/other" + ) + + tests := []struct { + name string + scope ProtocolScope + target ProtocolScope + want bool + }{ + { + name: "unscoped matches anything", + scope: ProtocolScope{}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "a/b", ContextID: "ctx/1"}, + want: true, + }, + { + name: "unscoped matches empty target", + scope: ProtocolScope{}, + target: ProtocolScope{}, + want: true, + }, + { + name: "protocol exact equality match", + scope: ProtocolScope{Protocol: mesh}, + target: ProtocolScope{Protocol: mesh}, + want: true, + }, + { + name: "protocol mismatch", + scope: ProtocolScope{Protocol: mesh}, + target: ProtocolScope{Protocol: other}, + want: false, + }, + { + name: "protocol scope vs unset target protocol", + scope: ProtocolScope{Protocol: mesh}, + target: ProtocolScope{}, + want: false, + }, + { + name: "protocol scope ignores target path and context", + scope: ProtocolScope{Protocol: mesh}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "a/b", ContextID: "ctx"}, + want: true, + }, + { + name: "protocolPath and contextId together fail closed", + scope: ProtocolScope{Protocol: mesh, ProtocolPath: "a", ContextID: "ctx"}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "a", ContextID: "ctx"}, + want: false, + }, + { + name: "protocolPath without protocol fails closed", + scope: ProtocolScope{ProtocolPath: "a"}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "a"}, + want: false, + }, + { + name: "contextId without protocol fails closed", + scope: ProtocolScope{ContextID: "ctx"}, + target: ProtocolScope{Protocol: mesh, ContextID: "ctx"}, + want: false, + }, + { + name: "protocolPath exact match", + scope: ProtocolScope{Protocol: mesh, ProtocolPath: "network/node"}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "network/node"}, + want: true, + }, + { + name: "protocolPath subtree descendant", + scope: ProtocolScope{Protocol: mesh, ProtocolPath: "network/node"}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "network/node/endpoint"}, + want: true, + }, + { + name: "protocolPath boundary-aware: sibling prefix does not match", + scope: ProtocolScope{Protocol: mesh, ProtocolPath: "network/node"}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "network/nodeInfo"}, + want: false, + }, + { + name: "protocolPath parent not covered", + scope: ProtocolScope{Protocol: mesh, ProtocolPath: "network/node"}, + target: ProtocolScope{Protocol: mesh, ProtocolPath: "network"}, + want: false, + }, + { + name: "protocolPath scope vs unset target path", + scope: ProtocolScope{Protocol: mesh, ProtocolPath: "network/node"}, + target: ProtocolScope{Protocol: mesh}, + want: false, + }, + { + name: "contextId exact match", + scope: ProtocolScope{Protocol: mesh, ContextID: "root/child"}, + target: ProtocolScope{Protocol: mesh, ContextID: "root/child"}, + want: true, + }, + { + name: "contextId subtree descendant", + scope: ProtocolScope{Protocol: mesh, ContextID: "root"}, + target: ProtocolScope{Protocol: mesh, ContextID: "root/child/grandchild"}, + want: true, + }, + { + name: "contextId boundary-aware: sibling prefix does not match", + scope: ProtocolScope{Protocol: mesh, ContextID: "root"}, + target: ProtocolScope{Protocol: mesh, ContextID: "rooted"}, + want: false, + }, + { + name: "contextId scope vs unset target context", + scope: ProtocolScope{Protocol: mesh, ContextID: "root"}, + target: ProtocolScope{Protocol: mesh}, + want: false, + }, + { + name: "contextId scope with wrong protocol", + scope: ProtocolScope{Protocol: mesh, ContextID: "root"}, + target: ProtocolScope{Protocol: other, ContextID: "root"}, + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := MatchesPermissionScope(tc.scope, tc.target); got != tc.want { + t.Errorf("MatchesPermissionScope(%+v, %+v) = %v, want %v", tc.scope, tc.target, got, tc.want) + } + }) + } +} + +func TestMatchesSubtree(t *testing.T) { + tests := []struct { + subtree string + candidate string + want bool + }{ + {"a/b", "a/b", true}, + {"a/b", "a/b/c", true}, + {"a/b", "a/bc", false}, + {"a/b", "a", false}, + {"a/b", "", false}, + {"a", "a/b/c", true}, + } + for _, tc := range tests { + if got := MatchesSubtree(tc.subtree, tc.candidate); got != tc.want { + t.Errorf("MatchesSubtree(%q, %q) = %v, want %v", tc.subtree, tc.candidate, got, tc.want) + } + } +} diff --git a/internal/dwn/subscribe.go b/internal/dwn/subscribe.go index 3bc714d..212556a 100644 --- a/internal/dwn/subscribe.go +++ b/internal/dwn/subscribe.go @@ -113,6 +113,23 @@ type SubscriptionError struct { Detail string `json:"detail"` } +// Wire error codes of terminal Messages subscription delivery failures. The +// server emits these in-band (type:"error") and then closes the subscription: +// AuthorizationFailed when delivery-time grant revalidation fails (grant +// revoked or expired mid-subscription), DeliveryFailed for any other +// unrecoverable delivery error. +const ( + MessagesSubscribeDeliveryAuthorizationFailedCode = "MessagesSubscribeDeliveryAuthorizationFailed" + MessagesSubscribeDeliveryFailedCode = "MessagesSubscribeDeliveryFailed" +) + +// ErrSubscriptionAuthorizationRevoked marks a terminal subscription error +// caused by the server permanently stopping delivery — grant revocation, +// grant expiry, or an unrecoverable delivery failure. The terminal lifecycle +// error wraps this sentinel so callers can distinguish "re-resolve the grant +// set" from other terminal conditions via errors.Is. +var ErrSubscriptionAuthorizationRevoked = errors.New("subscription delivery authorization revoked") + // ProgressGapInfo describes why a stored cursor cannot be resumed. type ProgressGapInfo struct { Code string `json:"code,omitempty"` @@ -139,6 +156,16 @@ type SubscriptionLifecycleEvent struct { RetryIn time.Duration Gap *ProgressGapInfo Err error + + // Head is the replication-log high-water token from a MessagesSubscribe + // reply, set on established events when the server provided a feed + // snapshot. NOT a delivery cursor: adopt as a checkpoint only after + // Fingerprint verifies against the local feed. + Head *ProgressToken + + // Fingerprint is the feed set fingerprint from a MessagesSubscribe + // reply, set on established events alongside Head. + Fingerprint string } // SubscriptionLifecycleHandler receives serialized connection lifecycle events. @@ -187,6 +214,7 @@ type Subscription struct { filter RecordsFilter signer *Signer auth MessageAuth + messages *messagesSubscribeConfig // nil for Records subscriptions handler EventHandler lifecycleHandler SubscriptionLifecycleHandler cursor *ProgressToken // last known cursor for reconnection @@ -297,18 +325,6 @@ func (m *SubscriptionManager) SubscribeWithAuthAndLifecycle( handler EventHandler, lifecycleHandler SubscriptionLifecycleHandler, ) (*Subscription, error) { - if ctx == nil { - return nil, fmt.Errorf("subscription context is required") - } - if target == "" { - return nil, fmt.Errorf("subscription target is required") - } - if signer == nil { - return nil, fmt.Errorf("subscription signer is required") - } - if handler == nil { - return nil, fmt.Errorf("subscription event handler is required") - } if err := auth.validate(); err != nil { return nil, fmt.Errorf("subscription auth: %w", err) } @@ -316,24 +332,109 @@ func (m *SubscriptionManager) SubscribeWithAuthAndLifecycle( filter = cloneSubscriptionFilter(filter) auth.DelegatedGrant = append(json.RawMessage(nil), auth.DelegatedGrant...) - subCtx, cancel := context.WithCancel(ctx) - subscriptionID := uuid.New().String() - - sub := &Subscription{ - SubscriptionID: subscriptionID, + return m.start(ctx, &Subscription{ target: target, filter: filter, signer: signer, auth: auth, handler: handler, lifecycleHandler: lifecycleHandler, - logger: m.logger.With( - slog.String("subscription_id", subscriptionID), - slog.String("target", target), - ), - cancel: cancel, - done: make(chan struct{}), + }) +} + +// messagesSubscribeConfig holds the immutable wire parameters of a Messages +// feed subscription; each (re)connect signs a fresh MessagesSubscribe from +// them plus the current resume cursor. +type messagesSubscribeConfig struct { + filters []MessagesFilter + permissionGrantIDs []string // canonical +} + +// SubscribeMessages starts a Messages-interface feed subscription against the +// target tenant's replication log. See SubscribeMessagesWithLifecycle. +func (m *SubscriptionManager) SubscribeMessages( + ctx context.Context, + target string, + signer *Signer, + opts MessagesSubscribeOptions, + handler EventHandler, +) (*Subscription, error) { + return m.SubscribeMessagesWithLifecycle(ctx, target, signer, opts, handler, nil) +} + +// SubscribeMessagesWithLifecycle starts a Messages-interface feed +// subscription and reports transport lifecycle state. +// +// The signed message is a MessagesSubscribe descriptor (filters + +// permissionGrantIds + cursor). opts.Cursor seeds the resume position; on +// reconnect the last delivered cursor is used, exactly like Records +// subscriptions. Established lifecycle events surface the subscribe reply's +// head token and fingerprint when the server provides a feed snapshot. +func (m *SubscriptionManager) SubscribeMessagesWithLifecycle( + ctx context.Context, + target string, + signer *Signer, + opts MessagesSubscribeOptions, + handler EventHandler, + lifecycleHandler SubscriptionLifecycleHandler, +) (*Subscription, error) { + // Canonicalize and validate once so a bad grant list or filter fails + // fast instead of on every reconnect. + canonicalIDs, err := CanonicalPermissionGrantIDs(opts.PermissionGrantIDs) + if err != nil { + return nil, fmt.Errorf("subscription permission grant IDs: %w", err) + } + filters := cloneMessagesFilters(opts.Filters) + if _, err := messagesFiltersToWire(filters); err != nil { + return nil, fmt.Errorf("subscription filters: %w", err) } + var initialCursor *ProgressToken + if opts.Cursor != nil { + if !opts.Cursor.valid() { + return nil, fmt.Errorf("subscription cursor is not a valid progress token") + } + initialCursor = opts.Cursor.clone() + } + + return m.start(ctx, &Subscription{ + target: target, + signer: signer, + messages: &messagesSubscribeConfig{ + filters: filters, + permissionGrantIDs: canonicalIDs, + }, + handler: handler, + lifecycleHandler: lifecycleHandler, + cursor: initialCursor, + }) +} + +// start validates and registers a subscription with the shared +// reconnect/ack/close machinery, then launches its run loop. +func (m *SubscriptionManager) start(ctx context.Context, sub *Subscription) (*Subscription, error) { + if ctx == nil { + return nil, fmt.Errorf("subscription context is required") + } + if sub.target == "" { + return nil, fmt.Errorf("subscription target is required") + } + if sub.signer == nil { + return nil, fmt.Errorf("subscription signer is required") + } + if sub.handler == nil { + return nil, fmt.Errorf("subscription event handler is required") + } + + subCtx, cancel := context.WithCancel(ctx) + subscriptionID := uuid.New().String() + + sub.SubscriptionID = subscriptionID + sub.logger = m.logger.With( + slog.String("subscription_id", subscriptionID), + slog.String("target", sub.target), + ) + sub.cancel = cancel + sub.done = make(chan struct{}) m.mu.Lock() if m.closed { @@ -359,6 +460,23 @@ func (m *SubscriptionManager) SubscribeWithAuthAndLifecycle( return sub, nil } +// cloneMessagesFilters deep-copies a Messages filter list so later caller +// mutations cannot change what reconnects sign. +func cloneMessagesFilters(filters []MessagesFilter) []MessagesFilter { + if filters == nil { + return nil + } + clone := make([]MessagesFilter, len(filters)) + for i, f := range filters { + if f.MessageTimestamp != nil { + r := *f.MessageTimestamp + f.MessageTimestamp = &r + } + clone[i] = f + } + return clone +} + // Close stops a subscription and sends rpc.subscribe.close to the server. func (s *Subscription) Close() { // Callbacks run on the subscription goroutine to preserve event, cursor, @@ -451,12 +569,12 @@ func (s *Subscription) run(ctx context.Context, endpoint string) { return } - var gapErr *progressGapError + var gapErr *ProgressGapError if errors.As(err, &gapErr) { s.clearCursor() s.emitLifecycle(SubscriptionLifecycleEvent{ Kind: SubscriptionLifecycleProgressGap, - Gap: gapErr.info, + Gap: gapErr.Info, Err: err, }) backoff = time.Second @@ -522,7 +640,7 @@ func (s *Subscription) connect(ctx context.Context, endpoint string) (bool, erro func (s *Subscription) sendSubscribeRequest(ctx context.Context, conn *websocket.Conn) (string, *ProgressToken, error) { cursor := s.Cursor() - msg, err := buildSubscribeMessage(s.signer, s.filter, cursor, s.auth) + msg, err := s.buildSubscribeWireMessage(cursor) if err != nil { return "", nil, fmt.Errorf("building subscribe message: %w", err) } @@ -540,18 +658,6 @@ func (s *Subscription) sendSubscribeRequest(ctx context.Context, conn *websocket return rpcReq.ID, cursor, nil } -// progressGapError marks a cursor that production can no longer resume. -type progressGapError struct { - info *ProgressGapInfo -} - -func (e *progressGapError) Error() string { - if e == nil || e.info == nil { - return "subscription progress gap" - } - return fmt.Sprintf("subscription progress gap: %s", e.info.Reason) -} - // readSubscribeResponse demultiplexes replay events and the request reply. // Production may deliver subscription-ID replay frames before the request-ID // confirmation because it installs and drains the listener synchronously. @@ -589,11 +695,7 @@ func (s *Subscription) readSubscribeResponse(ctx context.Context, conn *websocke reply := rpcResp.Result.Reply if reply.Status.Code == http.StatusGone { - gap := &ProgressGapInfo{} - if len(reply.Error) > 0 { - _ = json.Unmarshal(reply.Error, gap) - } - return &progressGapError{info: gap} + return parseProgressGapReply(reply) } if reply.Status.Code != http.StatusOK { failure := fmt.Errorf("subscription failed: %d %s", reply.Status.Code, reply.Status.Detail) @@ -609,6 +711,8 @@ func (s *Subscription) readSubscribeResponse(ctx context.Context, conn *websocke s.emitLifecycle(SubscriptionLifecycleEvent{ Kind: SubscriptionLifecycleEstablished, NeedsFullRefresh: resumeCursor == nil, + Head: reply.Head.clone(), + Fingerprint: reply.Fingerprint, }) s.logger.InfoContext(ctx, "subscription established") return nil @@ -692,7 +796,15 @@ func (s *Subscription) handleSubscriptionFrame(ctx context.Context, conn *websoc } var terminalErr error if messageType == SubscriptionErrorType { - terminalErr = fmt.Errorf("terminal subscription error: %s: %s", message.Error.Code, message.Error.Detail) + switch message.Error.Code { + case MessagesSubscribeDeliveryAuthorizationFailedCode, MessagesSubscribeDeliveryFailedCode: + // Grant revocation/expiry (or unrecoverable delivery failure): + // wrap the sentinel so callers can react specifically. + terminalErr = fmt.Errorf("terminal subscription error: %s: %s: %w", + message.Error.Code, message.Error.Detail, ErrSubscriptionAuthorizationRevoked) + default: + terminalErr = fmt.Errorf("terminal subscription error: %s: %s", message.Error.Code, message.Error.Detail) + } } if s.handler != nil { @@ -781,7 +893,7 @@ func subscriptionCursorGap(current, candidate *ProgressToken, reason string) err if candidate != nil { info.LatestAvailable = *candidate.clone() } - return &progressGapError{info: info} + return &ProgressGapError{Info: info} } // sendAck sends an rpc.ack message for flow control. @@ -815,6 +927,20 @@ func (s *Subscription) closeSubscription(conn *websocket.Conn) { _ = conn.Write(ctx, websocket.MessageText, data) } +// buildSubscribeWireMessage signs the subscribe message for the current +// connection attempt: a MessagesSubscribe descriptor for Messages feed +// subscriptions, otherwise a RecordsSubscribe descriptor. +func (s *Subscription) buildSubscribeWireMessage(cursor *ProgressToken) (*Message, error) { + if s.messages != nil { + return BuildMessagesSubscribe(s.signer, MessagesSubscribeOptions{ + Filters: s.messages.filters, + PermissionGrantIDs: s.messages.permissionGrantIDs, + Cursor: cursor, + }) + } + return buildSubscribeMessage(s.signer, s.filter, cursor, s.auth) +} + // buildSubscribeMessage creates a RecordsSubscribe DWN message. // // Like the other Records builders it supports protocol-role, plain-grant diff --git a/internal/dwn/subscribe_hardening_test.go b/internal/dwn/subscribe_hardening_test.go index 54664c4..79919df 100644 --- a/internal/dwn/subscribe_hardening_test.go +++ b/internal/dwn/subscribe_hardening_test.go @@ -190,15 +190,15 @@ func TestSubscriptionCursorProgressionAllowsOnlyExactEOSE(t *testing.T) { } err := sub.validateCursorProgression(SubscriptionEventType, current.clone()) - var gap *progressGapError - if !errors.As(err, &gap) || gap.info == nil || gap.info.Reason != "token_too_old" { + var gap *ProgressGapError + if !errors.As(err, &gap) || gap.Info == nil || gap.Info.Reason != "token_too_old" { t.Fatalf("exact event token error = %v, want token_too_old gap", err) } epochMismatch := current epochMismatch.Epoch = "other-epoch" err = sub.validateCursorProgression(SubscriptionEOSEType, &epochMismatch) - if !errors.As(err, &gap) || gap.info == nil || gap.info.Reason != "epoch_mismatch" { + if !errors.As(err, &gap) || gap.Info == nil || gap.Info.Reason != "epoch_mismatch" { t.Fatalf("cross-epoch EOSE error = %v, want epoch_mismatch gap", err) } } diff --git a/internal/dwn/testdata/grant_ids_fixtures.json b/internal/dwn/testdata/grant_ids_fixtures.json new file mode 100644 index 0000000..05f40b7 --- /dev/null +++ b/internal/dwn/testdata/grant_ids_fixtures.json @@ -0,0 +1,99 @@ +{ + "generatedBy": "bun scratchpad/fixtures/generate.ts (enbox@main, dwn-sdk-js src executed directly); fixed Ed25519 seed 0x01..0x20", + "vectors": [ + { + "name": "sorted-dedupe-ascii", + "input": { + "permissionGrantIds": [ + "b", + "a", + "b", + "c", + "a" + ] + }, + "canonical": [ + "a", + "b", + "c" + ] + }, + { + "name": "single", + "input": { + "permissionGrantIds": [ + "only" + ] + }, + "canonical": [ + "only" + ] + }, + { + "name": "unicode-bmp", + "input": { + "permissionGrantIds": [ + "é", + "z", + "a", + "É" + ] + }, + "canonical": [ + "a", + "z", + "É", + "é" + ] + }, + { + "name": "unicode-astral-vs-bmp", + "input": { + "permissionGrantIds": [ + "�", + "😀", + "A" + ] + }, + "canonical": [ + "A", + "😀", + "�" + ] + }, + { + "name": "unicode-mixed", + "input": { + "permissionGrantIds": [ + "😀-grant", + "filigature", + "ascii", + "ff" + ] + }, + "canonical": [ + "ascii", + "😀-grant", + "ff", + "filigature" + ] + }, + { + "name": "empty-array-error", + "input": { + "permissionGrantIds": [] + }, + "errorCode": "MessagePermissionGrantIdsEmpty" + }, + { + "name": "ambiguous-error", + "input": { + "permissionGrantId": "x", + "permissionGrantIds": [ + "y" + ] + }, + "errorCode": "MessagePermissionGrantCreateInvocationAmbiguous" + } + ] +} diff --git a/internal/dwn/testdata/link_identity_fixtures.json b/internal/dwn/testdata/link_identity_fixtures.json new file mode 100644 index 0000000..65b7a55 --- /dev/null +++ b/internal/dwn/testdata/link_identity_fixtures.json @@ -0,0 +1,131 @@ +{ + "generatedBy": "bun scratchpad/fixtures/generate.ts (enbox@main, dwn-sdk-js src executed directly); fixed Ed25519 seed 0x01..0x20", + "projectionIds": [ + { + "name": "full-scope", + "tenantDid": "did:jwk:eyJjcnYiOiJFZDI1NTE5Iiwia3R5IjoiT0tQIiwieCI6ImViVldMb19tVlBsQWVMRVM2S21McDVBZmhUcm1sYjdYNE9PUkM2MEVsbVEifQ", + "scope": { + "kind": "full" + }, + "projectionId": "5Tk4-lTjMLHdR7yZo4JvkWc92s5GgLvoNjkpHq3mRg4" + }, + { + "name": "protocol-set-single", + "tenantDid": "did:jwk:eyJjcnYiOiJFZDI1NTE5Iiwia3R5IjoiT0tQIiwieCI6ImViVldMb19tVlBsQWVMRVM2S21McDVBZmhUcm1sYjdYNE9PUkM2MEVsbVEifQ", + "scope": { + "kind": "protocolSet", + "protocols": [ + "https://example.com/protocols/mesh" + ] + }, + "projectionId": "EJ9blfEE7H6KMZjjZ0z-reF73-TFWfJ92sXZyKzmhYA" + }, + { + "name": "protocol-set-unsorted-dupes", + "tenantDid": "did:dht:tenant123", + "scope": { + "kind": "protocolSet", + "protocols": [ + "https://example.com/z", + "https://example.com/a", + "https://example.com/z" + ] + }, + "projectionId": "4wjeR3RyuLut_sm0fANvL4QV0yF6OlDdNr7ym7NAHrQ" + }, + { + "name": "protocol-set-unicode-and-html-chars", + "tenantDid": "did:dht:tenant123", + "scope": { + "kind": "protocolSet", + "protocols": [ + "https://example.com/prötocols/😀", + "https://example.com/a?x=1&y=<2>", + "https://example.com/�" + ] + }, + "projectionId": "_KbAXt4vJYa0H17nJu2Do5uDIhsxJGbjDYaH21I0SRA" + } + ], + "authorizationEpochs": [ + { + "name": "owner", + "input": { + "kind": "owner" + }, + "epoch": "DSJcSmkcjSfZDpCUWogVinCvpvCCRNRFre6GL0PG2iQ" + }, + { + "name": "delegate-two-grants-unsorted", + "input": { + "kind": "delegate", + "delegateDid": "did:jwk:eyJjcnYiOiJFZDI1NTE5Iiwia3R5IjoiT0tQIiwieCI6ImViVldMb19tVlBsQWVMRVM2S21McDVBZmhUcm1sYjdYNE9PUkM2MEVsbVEifQ", + "grants": [ + { + "id": "bafyreigrantbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "dateExpires": "2026-08-01T00:00:00.000000Z", + "dateGranted": "2026-07-01T00:00:00.000000Z" + }, + { + "id": "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "dateExpires": "2026-09-01T00:00:00.000000Z", + "dateGranted": "2026-07-02T10:20:30.400500Z" + } + ] + }, + "epoch": "BzCEaopdOQJwS8WXIQ6TqKw_xZOl4monwJ3dIiTsBGE" + }, + { + "name": "delegate-unicode-optional-dategranted", + "input": { + "kind": "delegate", + "delegateDid": "did:dht:delegate<&>😀", + "grants": [ + { + "id": "z-grant-😀", + "dateExpires": "2027-01-01T00:00:00.000000Z" + }, + { + "id": "a-grant-�", + "dateExpires": "2026-12-31T23:59:59.999999Z", + "dateGranted": "2026-06-01T00:00:00.000000Z" + }, + { + "id": "😀-astral-first", + "dateExpires": "2026-10-01T00:00:00.000000Z", + "dateGranted": "2026-05-01T00:00:00.000000Z" + } + ] + }, + "epoch": "TMYbD9KAcUn0lbqC0-bFbUUmyLfenTVIRulE0_-Q9pA" + } + ], + "linkKeys": [ + { + "name": "basic", + "tenantDid": "did:dht:tenant123", + "remoteEndpoint": "https://dwn.example.com", + "projectionId": "proj-id", + "authorizationEpoch": "epoch-id", + "linkKey": "did:dht:tenant123^https://dwn.example.com^proj-id^epoch-id" + } + ], + "normalizeSyncProtocols": [ + { + "name": "unsorted-dupes-unicode", + "input": [ + "https://z.example", + "https://a.example", + "https://z.example", + "https://example.com/😀", + "https://example.com/�" + ], + "normalized": [ + "https://a.example", + "https://example.com/😀", + "https://example.com/�", + "https://z.example" + ] + } + ] +} diff --git a/internal/dwn/testdata/messages_wire_fixtures.json b/internal/dwn/testdata/messages_wire_fixtures.json new file mode 100644 index 0000000..ea60dd6 --- /dev/null +++ b/internal/dwn/testdata/messages_wire_fixtures.json @@ -0,0 +1,380 @@ +{ + "generatedBy": "bun scratchpad/fixtures/generate.ts (enbox@main, dwn-sdk-js src executed directly); fixed Ed25519 seed 0x01..0x20", + "did": "did:jwk:eyJjcnYiOiJFZDI1NTE5Iiwia3R5IjoiT0tQIiwieCI6ImViVldMb19tVlBsQWVMRVM2S21McDVBZmhUcm1sYjdYNE9PUkM2MEVsbVEifQ", + "keyId": "did:jwk:eyJjcnYiOiJFZDI1NTE5Iiwia3R5IjoiT0tQIiwieCI6ImViVldMb19tVlBsQWVMRVM2S21McDVBZmhUcm1sYjdYNE9PUkM2MEVsbVEifQ#0", + "ed25519SeedBase64Url": "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA", + "fixtures": [ + { + "name": "query-grants-cursor-limit", + "inputs": { + "messageTimestamp": "2026-07-26T12:00:00.123456Z", + "filters": [ + { + "protocol": "https://example.com/protocols/mesh" + } + ], + "permissionGrantIds": [ + "bafyreigrantbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "bafyreigrantbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ], + "cursor": { + "streamId": "a1b2c3d4e5f60718", + "epoch": "018f2c6e-2222-7333-8444-555566667777", + "position": "42", + "messageCid": "bafyreib3n2ku2tzt2ie7yhqhotozcxfjnaubqhgifzfdkbmtjhaqmyzvce" + }, + "limit": 25 + }, + "message": { + "descriptor": { + "interface": "Messages", + "method": "Query", + "filters": [ + { + "protocol": "https://example.com/protocols/mesh" + } + ], + "messageTimestamp": "2026-07-26T12:00:00.123456Z", + "cursor": { + "streamId": "a1b2c3d4e5f60718", + "epoch": "018f2c6e-2222-7333-8444-555566667777", + "position": "42", + "messageCid": "bafyreib3n2ku2tzt2ie7yhqhotozcxfjnaubqhgifzfdkbmtjhaqmyzvce" + }, + "limit": 25, + "permissionGrantIds": [ + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "bafyreigrantbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ] + }, + "authorization": { + "signature": { + "payload": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWR1aXJraTVyYW03YTdiaGF4aDR4ZGk1ZTZ2bW1oNTVhc2tlMjducTRnNGRwNGdlcnd3M3UiLCJwZXJtaXNzaW9uR3JhbnRJZHMiOlsiYmFmeXJlaWdyYW50YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhIiwiYmFmeXJlaWdyYW50YmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiIl19", + "signatures": [ + { + "protected": "eyJraWQiOiJkaWQ6andrOmV5SmpjbllpT2lKRlpESTFOVEU1SWl3aWEzUjVJam9pVDB0UUlpd2llQ0k2SW1WaVZsZE1iMTl0VmxCc1FXVk1SVk0yUzIxTWNEVkJabWhVY20xc1lqZFlORTlQVWtNMk1FVnNiVkVpZlEjMCIsImFsZyI6IkVkRFNBIn0", + "signature": "Avpe4jKLsErzKyvNsVgmQ1MKMteBYuayWbLD3AkaQl_ERFpyQf0CQZCtzmF7iiJrVlyFgXVCJpm0uxb5_R-JDw" + } + ] + } + } + }, + "descriptor": { + "interface": "Messages", + "method": "Query", + "filters": [ + { + "protocol": "https://example.com/protocols/mesh" + } + ], + "messageTimestamp": "2026-07-26T12:00:00.123456Z", + "cursor": { + "streamId": "a1b2c3d4e5f60718", + "epoch": "018f2c6e-2222-7333-8444-555566667777", + "position": "42", + "messageCid": "bafyreib3n2ku2tzt2ie7yhqhotozcxfjnaubqhgifzfdkbmtjhaqmyzvce" + }, + "limit": 25, + "permissionGrantIds": [ + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "bafyreigrantbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ] + }, + "signaturePayloadBase64Url": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWR1aXJraTVyYW03YTdiaGF4aDR4ZGk1ZTZ2bW1oNTVhc2tlMjducTRnNGRwNGdlcnd3M3UiLCJwZXJtaXNzaW9uR3JhbnRJZHMiOlsiYmFmeXJlaWdyYW50YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhIiwiYmFmeXJlaWdyYW50YmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiYmJiIl19", + "descriptorCid": "bafyreiduirki5ram7a7bhaxh4xdi5e6vmmh55aske27nq4g4dp4gerww3u", + "messageCid": "bafyreibrdlobrviygpvvbrzckotlloppxwxyblt4fkkthrpvdwryluew5q" + }, + { + "name": "query-cids-only-owner", + "inputs": { + "messageTimestamp": "2026-07-26T12:00:01.000001Z", + "cidsOnly": true, + "limit": 1 + }, + "message": { + "descriptor": { + "interface": "Messages", + "method": "Query", + "filters": [], + "messageTimestamp": "2026-07-26T12:00:01.000001Z", + "limit": 1, + "cidsOnly": true + }, + "authorization": { + "signature": { + "payload": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWdvYXVoMzN2Z2dmeG9tNzNsenV0aHZ0ZWFhNXJzbm91YnpxenZkZTVlaTdhNnVrN3lyN2kifQ", + "signatures": [ + { + "protected": "eyJraWQiOiJkaWQ6andrOmV5SmpjbllpT2lKRlpESTFOVEU1SWl3aWEzUjVJam9pVDB0UUlpd2llQ0k2SW1WaVZsZE1iMTl0VmxCc1FXVk1SVk0yUzIxTWNEVkJabWhVY20xc1lqZFlORTlQVWtNMk1FVnNiVkVpZlEjMCIsImFsZyI6IkVkRFNBIn0", + "signature": "cQqXDPnyJxI8mPHte_RnJRpPaNlza5T3YGklkPXMDh1HEEBrF23VhewF0JN08FFs4TX8EiWl6e5DMKdHoIc8DA" + } + ] + } + } + }, + "descriptor": { + "interface": "Messages", + "method": "Query", + "filters": [], + "messageTimestamp": "2026-07-26T12:00:01.000001Z", + "limit": 1, + "cidsOnly": true + }, + "signaturePayloadBase64Url": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWdvYXVoMzN2Z2dmeG9tNzNsenV0aHZ0ZWFhNXJzbm91YnpxenZkZTVlaTdhNnVrN3lyN2kifQ", + "descriptorCid": "bafyreigoauh33vggfxom73lzuthvteaa5rsnoubzqzvde5ei7a6uk7yr7i", + "messageCid": "bafyreibur3g6ezyef2kibcxpta4nxah3cg3h64ca4jtrkddeaghh5qumnu" + }, + { + "name": "query-filter-shapes-unicode", + "inputs": { + "messageTimestamp": "2026-07-26T12:00:02.999999Z", + "filters": [ + { + "interface": "Records", + "method": "Write", + "protocol": "https://example.com/protocols/mesh", + "protocolPathPrefix": "network/node" + }, + { + "contextIdPrefix": "ctx-root/child", + "messageTimestamp": { + "from": "2026-01-01T00:00:00.000000Z", + "to": "2026-07-26T12:00:00.123456Z" + } + }, + { + "protocol": "https://example.com/prötocols/😀" + } + ] + }, + "message": { + "descriptor": { + "interface": "Messages", + "method": "Query", + "filters": [ + { + "interface": "Records", + "method": "Write", + "protocol": "https://example.com/protocols/mesh", + "protocolPathPrefix": "network/node" + }, + { + "contextIdPrefix": "ctx-root/child", + "messageTimestamp": { + "from": "2026-01-01T00:00:00.000000Z", + "to": "2026-07-26T12:00:00.123456Z" + } + }, + { + "protocol": "https://example.com/prötocols/😀" + } + ], + "messageTimestamp": "2026-07-26T12:00:02.999999Z" + }, + "authorization": { + "signature": { + "payload": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWN3cTV6cmxkdmN0eWozcG5nbXJ2dTJwYTdlZnE3amtkNnB1dmd3amRibW80aHhmcXN2NTQifQ", + "signatures": [ + { + "protected": "eyJraWQiOiJkaWQ6andrOmV5SmpjbllpT2lKRlpESTFOVEU1SWl3aWEzUjVJam9pVDB0UUlpd2llQ0k2SW1WaVZsZE1iMTl0VmxCc1FXVk1SVk0yUzIxTWNEVkJabWhVY20xc1lqZFlORTlQVWtNMk1FVnNiVkVpZlEjMCIsImFsZyI6IkVkRFNBIn0", + "signature": "V8hb5mAFI6wwvjJpgt_tcrhsHuYJVy9YX71p31TqzeixZ4af4JDjxv4OmZJM0BWjnH9xQ_uwKvPIVFVrWyymBw" + } + ] + } + } + }, + "descriptor": { + "interface": "Messages", + "method": "Query", + "filters": [ + { + "interface": "Records", + "method": "Write", + "protocol": "https://example.com/protocols/mesh", + "protocolPathPrefix": "network/node" + }, + { + "contextIdPrefix": "ctx-root/child", + "messageTimestamp": { + "from": "2026-01-01T00:00:00.000000Z", + "to": "2026-07-26T12:00:00.123456Z" + } + }, + { + "protocol": "https://example.com/prötocols/😀" + } + ], + "messageTimestamp": "2026-07-26T12:00:02.999999Z" + }, + "signaturePayloadBase64Url": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWN3cTV6cmxkdmN0eWozcG5nbXJ2dTJwYTdlZnE3amtkNnB1dmd3amRibW80aHhmcXN2NTQifQ", + "descriptorCid": "bafyreicwq5zrldvctyj3pngmrvu2pa7efq7jkd6puvgwjdbmo4hxfqsv54", + "messageCid": "bafyreidncf3a32mxw6js4su2v6mws3m5tqdgpmkuflpohag77nxzgvzdwq" + }, + { + "name": "read-single-grant", + "inputs": { + "messageTimestamp": "2026-07-26T12:00:00.123456Z", + "messageCid": "bafyreidykglsfhoixmivffc5uwhcgshx4j465xwqntbmu43nb2dzqwfvae", + "permissionGrantIds": [ + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + ] + }, + "message": { + "descriptor": { + "interface": "Messages", + "method": "Read", + "messageCid": "bafyreidykglsfhoixmivffc5uwhcgshx4j465xwqntbmu43nb2dzqwfvae", + "messageTimestamp": "2026-07-26T12:00:00.123456Z", + "permissionGrantIds": [ + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + ] + }, + "authorization": { + "signature": { + "payload": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWRkZmt5ajVxYnhndG9jdHZta28yeHRrZjRqNTZ1bDZ0aDZ0bWw1c3lpaHNld3I0bGVqYWkiLCJwZXJtaXNzaW9uR3JhbnRJZHMiOlsiYmFmeXJlaWdyYW50YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhIl19", + "signatures": [ + { + "protected": "eyJraWQiOiJkaWQ6andrOmV5SmpjbllpT2lKRlpESTFOVEU1SWl3aWEzUjVJam9pVDB0UUlpd2llQ0k2SW1WaVZsZE1iMTl0VmxCc1FXVk1SVk0yUzIxTWNEVkJabWhVY20xc1lqZFlORTlQVWtNMk1FVnNiVkVpZlEjMCIsImFsZyI6IkVkRFNBIn0", + "signature": "LIxyTmhtnFNF17M-TxoJkc4rce9InuEjB1aH-TnoyrDww3UR6MQBitd6Sk20OhX-Cew6dd1mH8NQU5oBAoJSBQ" + } + ] + } + } + }, + "descriptor": { + "interface": "Messages", + "method": "Read", + "messageCid": "bafyreidykglsfhoixmivffc5uwhcgshx4j465xwqntbmu43nb2dzqwfvae", + "messageTimestamp": "2026-07-26T12:00:00.123456Z", + "permissionGrantIds": [ + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + ] + }, + "signaturePayloadBase64Url": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWRkZmt5ajVxYnhndG9jdHZta28yeHRrZjRqNTZ1bDZ0aDZ0bWw1c3lpaHNld3I0bGVqYWkiLCJwZXJtaXNzaW9uR3JhbnRJZHMiOlsiYmFmeXJlaWdyYW50YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhIl19", + "descriptorCid": "bafyreiddfkyj5qbxgtoctvmko2xtkf4j56ul6th6tml5syihsewr4lejai", + "messageCid": "bafyreifxixp6i3cwqp3axwxtaui7ko3to2vsshymhlez5wvh4jjr37ojbm" + }, + { + "name": "subscribe-grants-cursor", + "inputs": { + "messageTimestamp": "2026-07-26T12:00:01.000001Z", + "filters": [ + { + "protocol": "https://example.com/protocols/mesh" + }, + { + "protocol": "https://example.com/protocols/audit" + } + ], + "permissionGrantIds": [ + "bafyreigrantccccccccccccccccccccccccccccccccccccccccccccccc4", + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + ], + "cursor": { + "streamId": "a1b2c3d4e5f60718", + "epoch": "018f2c6e-2222-7333-8444-555566667777", + "position": "42", + "messageCid": "bafyreib3n2ku2tzt2ie7yhqhotozcxfjnaubqhgifzfdkbmtjhaqmyzvce" + } + }, + "message": { + "descriptor": { + "interface": "Messages", + "method": "Subscribe", + "filters": [ + { + "protocol": "https://example.com/protocols/mesh" + }, + { + "protocol": "https://example.com/protocols/audit" + } + ], + "messageTimestamp": "2026-07-26T12:00:01.000001Z", + "cursor": { + "streamId": "a1b2c3d4e5f60718", + "epoch": "018f2c6e-2222-7333-8444-555566667777", + "position": "42", + "messageCid": "bafyreib3n2ku2tzt2ie7yhqhotozcxfjnaubqhgifzfdkbmtjhaqmyzvce" + }, + "permissionGrantIds": [ + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "bafyreigrantccccccccccccccccccccccccccccccccccccccccccccccc4" + ] + }, + "authorization": { + "signature": { + "payload": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWF4dWtpNGFtNHBhM2ozaGtmaG1qbGdhZmlkYnVhNHM2Z2NwMmxmZTQ0NDVtdHJxNmhpb2kiLCJwZXJtaXNzaW9uR3JhbnRJZHMiOlsiYmFmeXJlaWdyYW50YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhIiwiYmFmeXJlaWdyYW50Y2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2M0Il19", + "signatures": [ + { + "protected": "eyJraWQiOiJkaWQ6andrOmV5SmpjbllpT2lKRlpESTFOVEU1SWl3aWEzUjVJam9pVDB0UUlpd2llQ0k2SW1WaVZsZE1iMTl0VmxCc1FXVk1SVk0yUzIxTWNEVkJabWhVY20xc1lqZFlORTlQVWtNMk1FVnNiVkVpZlEjMCIsImFsZyI6IkVkRFNBIn0", + "signature": "2M-5H3UyxnmB01wptRvzT_XH5tbcZJ-lIELv3g3ecJ4T41lwKSGiykBFVW_4ppMgjShd5X88BJP_qAwbePxSDg" + } + ] + } + } + }, + "descriptor": { + "interface": "Messages", + "method": "Subscribe", + "filters": [ + { + "protocol": "https://example.com/protocols/mesh" + }, + { + "protocol": "https://example.com/protocols/audit" + } + ], + "messageTimestamp": "2026-07-26T12:00:01.000001Z", + "cursor": { + "streamId": "a1b2c3d4e5f60718", + "epoch": "018f2c6e-2222-7333-8444-555566667777", + "position": "42", + "messageCid": "bafyreib3n2ku2tzt2ie7yhqhotozcxfjnaubqhgifzfdkbmtjhaqmyzvce" + }, + "permissionGrantIds": [ + "bafyreigrantaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "bafyreigrantccccccccccccccccccccccccccccccccccccccccccccccc4" + ] + }, + "signaturePayloadBase64Url": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWF4dWtpNGFtNHBhM2ozaGtmaG1qbGdhZmlkYnVhNHM2Z2NwMmxmZTQ0NDVtdHJxNmhpb2kiLCJwZXJtaXNzaW9uR3JhbnRJZHMiOlsiYmFmeXJlaWdyYW50YWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhYWFhIiwiYmFmeXJlaWdyYW50Y2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2M0Il19", + "descriptorCid": "bafyreiaxuki4am4pa3j3hkfhmjlgafidbua4s6gcp2lfe4445mtrq6hioi", + "messageCid": "bafyreibhgynupgzqsms2hgqrhivytzshlfjljfxmgikt7beptzomzgcggu" + }, + { + "name": "subscribe-empty-filters-owner", + "inputs": { + "messageTimestamp": "2026-07-26T12:00:02.999999Z", + "filters": [] + }, + "message": { + "descriptor": { + "interface": "Messages", + "method": "Subscribe", + "filters": [], + "messageTimestamp": "2026-07-26T12:00:02.999999Z" + }, + "authorization": { + "signature": { + "payload": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWIycmtjZmtqZGtpdTdkcTZ6NnJxZHR1d2p1NzQzbXF3a3hkNGJvbnFhdDRjNGRhbmFud3UifQ", + "signatures": [ + { + "protected": "eyJraWQiOiJkaWQ6andrOmV5SmpjbllpT2lKRlpESTFOVEU1SWl3aWEzUjVJam9pVDB0UUlpd2llQ0k2SW1WaVZsZE1iMTl0VmxCc1FXVk1SVk0yUzIxTWNEVkJabWhVY20xc1lqZFlORTlQVWtNMk1FVnNiVkVpZlEjMCIsImFsZyI6IkVkRFNBIn0", + "signature": "VXN-yVWeZgL9k-ciyQ2v5F-t9HTf4Vta1oFbOxLkpSRPlHLUQoHQbnh9aEGnbDtaFNE4R1Yn1t8WW_cbgZ2nBQ" + } + ] + } + } + }, + "descriptor": { + "interface": "Messages", + "method": "Subscribe", + "filters": [], + "messageTimestamp": "2026-07-26T12:00:02.999999Z" + }, + "signaturePayloadBase64Url": "eyJkZXNjcmlwdG9yQ2lkIjoiYmFmeXJlaWIycmtjZmtqZGtpdTdkcTZ6NnJxZHR1d2p1NzQzbXF3a3hkNGJvbnFhdDRjNGRhbmFud3UifQ", + "descriptorCid": "bafyreib2rkcfkjdkiu7dq6z6rqdtuwju743mqwkxd4bonqat4c4dananwu", + "messageCid": "bafyreiboffei44bdiqhmjowjow4pydmrkgehvcgq77bgkbbcycvhuvj5qm" + } + ] +} diff --git a/internal/dwn/transport.go b/internal/dwn/transport.go index 366e8d8..46c7f9d 100644 --- a/internal/dwn/transport.go +++ b/internal/dwn/transport.go @@ -104,6 +104,19 @@ type DwnReply struct { Record json.RawMessage `json:"record,omitempty"` Error json.RawMessage `json:"error,omitempty"` + // Drained reports whether a MessagesQuery scan reached the log head. + Drained bool `json:"drained,omitempty"` + + // Fingerprint is the feed set fingerprint on MessagesQuery and + // MessagesSubscribe replies, when the filters map onto fingerprint + // domains. + Fingerprint string `json:"fingerprint,omitempty"` + + // Head is the replication-log high-water token on MessagesSubscribe + // replies. NOT a delivery cursor: adopt as a checkpoint only after the + // fingerprint verifies against the local feed. + Head *ProgressToken `json:"head,omitempty"` + // Subscription is populated in the initial subscribe response. Subscription *SubscriptionConfirm `json:"subscription,omitempty"` }