Enforce fine-grained CloudEvent grants

MERGER_REVIEW.md

@@ -80,3 +80,12 @@ Not functional, but the rename effort stopped at the public surface:
 ### 7. Empty `Cleanup()`
 
 `internal/app/app.go:106` — `cleanup: func() {}`. Both source repos also left this effectively empty, so it's not a merger regression, just a pre-existing TODO carried over. ClickHouse / S3 connections won't be closed on shutdown.
+
+## Fine-grained CloudEvent grants — RESTORED
+
+Telemetry-api could honor a narrow `cloud_events` JWT claim (authorize specific event types / sources / ids rather than "read all"); fetch-api never had this, and the merged service inherited fetch-api's coarse-only check. The capability is now restored at the GraphQL/MCP layer in `internal/graph/auth_helpers.go`:
+
+- `requireSubjectOptsByDID` now authorizes a CloudEvent request via **either** the existing full-access permissions (`GetRawData`, or `GetLocationHistory` + `GetNonLocationHistory`) **or** a covering `cloud_events` grant, evaluated independently of the permissions enum. Subject-DID scoping still applies in both cases.
+- Enforcement is up-front and fail-closed (`cloudEventRequestAllowed` / `grantCovers`, ported from telemetry-api's `validCloudEventRequest`): a request is allowed only if its filter falls entirely within a single grant; unset filter dimensions default to the `*` wildcard. We validate rather than post-filter results, so a request that can't be fully served is rejected instead of partially serviced.
+- The single chokepoint covers all three resolvers (`cloudEvents`, `latestCloudEvent`, `availableCloudEventTypes`) and the MCP surface (same resolvers via `@mcpTool`). Tests in `internal/graph/cloud_events_auth_test.go`.
+- **Out of scope:** the internal gRPC `FetchService` remains an unauthenticated, network-isolated, trusted surface by design — grants are not enforced there. Grant `tags` are not checked (no `tags` field on `CloudEventFilter`; telemetry-api ignored them too).

README.md

@@ -13,6 +13,35 @@ This service accepts subject and permission-scoped tokens. These are generally s
 }
 ```
 
+### CloudEvent access
+
+There are two independent ways a token can authorize the CloudEvent queries (`cloudEvents`,
+`latestCloudEvent`, `availableCloudEventTypes`). A request is permitted if **either** applies:
+
+1. **Full access via `permissions`.** A token holding `privilege:GetRawData`, or the combination of
+   `privilege:GetLocationHistory` and `privilege:GetNonLocationHistory`, may read every CloudEvent for
+   the subject.
+2. **A scoped `cloud_events` grant.** Independently of the `permissions` enum, a token may carry a
+   `cloud_events` claim that authorizes only specific event types, from specific sources, optionally
+   pinned to specific IDs. `"*"` is the wildcard ("any").
+
+```json
+{
+  "asset": "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:42",
+  "cloud_events": {
+    "events": [
+      { "event_type": "dimo.attestation", "source": "*", "ids": ["*"] }
+    ]
+  }
+}
+```
+
+A grant-scoped request is authorized only when its filter falls entirely within a single grant. Any
+dimension the filter leaves unset defaults to the wildcard `"*"`, so a narrow grant requires the caller
+to scope their query to match it. For example, the grant above permits
+`cloudEvents(subject: ..., filter: {type: "dimo.attestation"})` but rejects an unfiltered
+`cloudEvents(subject: ...)` or a request for a different `type`.
+
 ## Migrating
 
 ### From Telemetry

internal/graph/auth_helpers.go

@@ -18,31 +18,90 @@ const (
 )
 
 func (r *queryResolver) requireSubjectOptsByDID(ctx context.Context, requestedDID string, filter *model.CloudEventFilter) (*grpc.AdvancedSearchOptions, error) {
-	token, err := requireRawDataToken(ctx)
-	if err != nil {
-		return nil, err
+	tok, _ := ctx.Value(ClaimsContextKey{}).(*tokenclaims.Token)
+	if tok == nil {
+		return nil, fmt.Errorf("%s", errNoTokenClaims)
 	}
-	tokenSubjectDID := token.Asset
-	searchSubject, err := r.ensureRequestedDIDLinkedToPermissionedSubject(ctx, requestedDID, tokenSubjectDID)
+	// Two independent authorization paths, OR'd together:
+	//   1. The permissions enum grants full access to ALL of the subject's events.
+	//   2. The cloud_events grant, evaluated independently of the enum, authorizes exactly the
+	//      types/sources/ids its rules describe.
+	if !hasFullCloudEventAccess(tok) && !cloudEventRequestAllowed(tok.CloudEvents, filter) {
+		return nil, fmt.Errorf("%s", errNoPermission)
+	}
+	// Subject scoping applies regardless of which path authorized: a grant narrows WHICH
+	// types/sources/ids within the subject, it is not a license to read a different subject.
+	searchSubject, err := r.ensureRequestedDIDLinkedToPermissionedSubject(ctx, requestedDID, tok.Asset)
 	if err != nil {
 		return nil, err
 	}
 	return filterToAdvancedSearchOptions(filter, searchSubject), nil
 }
 
-func requireRawDataToken(ctx context.Context) (*tokenclaims.Token, error) {
-	tok, _ := ctx.Value(ClaimsContextKey{}).(*tokenclaims.Token)
-	if tok == nil {
-		return nil, fmt.Errorf("%s", errNoTokenClaims)
-	}
+// hasFullCloudEventAccess reports whether the permissions enum grants access to ALL of the
+// subject's events: either raw-data, or the (location + non-location) combination.
+func hasFullCloudEventAccess(tok *tokenclaims.Token) bool {
 	hasGetRawData := slices.Contains(tok.Permissions, tokenclaims.PermissionGetRawData)
 	hasLocationHistory := slices.Contains(tok.Permissions, tokenclaims.PermissionGetLocationHistory)
 	hasNonLocationHistory := slices.Contains(tok.Permissions, tokenclaims.PermissionGetNonLocationHistory)
-	hasAllTimeData := hasLocationHistory && hasNonLocationHistory
-	if !hasGetRawData && !hasAllTimeData {
-		return nil, fmt.Errorf("%s", errNoPermission)
+	return hasGetRawData || (hasLocationHistory && hasNonLocationHistory)
+}
+
+// cloudEventRequestAllowed reports whether every event the given filter could match falls within
+// at least one of the bearer's cloud_events grants. It is evaluated independently of the
+// permissions enum and is fail-closed: an absent grant, or a filter broader than any grant,
+// returns false.
+func cloudEventRequestAllowed(grants *tokenclaims.CloudEvents, filter *model.CloudEventFilter) bool {
+	if grants == nil || len(grants.Events) == 0 {
+		return false
+	}
+	// Requested types = filter.Type + filter.Types. An empty set means "any type", which only a
+	// wildcard grant can cover.
+	var types []string
+	if filter != nil {
+		if filter.Type != nil {
+			types = append(types, *filter.Type)
+		}
+		types = append(types, filter.Types...)
+	}
+	if len(types) == 0 {
+		types = []string{tokenclaims.GlobalIdentifier}
+	}
+	source := tokenclaims.GlobalIdentifier
+	id := tokenclaims.GlobalIdentifier
+	if filter != nil {
+		if filter.Source != nil {
+			source = *filter.Source
+		}
+		if filter.ID != nil {
+			id = *filter.ID
+		}
+	}
+	// The query returns the union over requested types, so EACH requested type must be covered by
+	// some grant.
+	for _, t := range types {
+		if !grantCovers(grants.Events, t, source, id) {
+			return false
+		}
+	}
+	return true
+}
+
+// grantCovers reports whether any single grant authorizes the (type, source, id) tuple. A grant
+// matches a dimension when it equals the requested value or carries the GlobalIdentifier wildcard.
+func grantCovers(events []tokenclaims.Event, evtType, source, id string) bool {
+	for _, ce := range events {
+		if ce.EventType != evtType && ce.EventType != tokenclaims.GlobalIdentifier {
+			continue
+		}
+		if ce.Source != source && ce.Source != tokenclaims.GlobalIdentifier {
+			continue
+		}
+		if slices.Contains(ce.IDs, id) || slices.Contains(ce.IDs, tokenclaims.GlobalIdentifier) {
+			return true
+		}
 	}
-	return tok, nil
+	return false
 }
 
 func (r *queryResolver) ensureRequestedDIDLinkedToPermissionedSubject(ctx context.Context, requestedDID string, tokenSubjectDID string) (string, error) {

internal/graph/cloud_events_auth_test.go

@@ -0,0 +1,238 @@
+package graph
+
+import (
+	"context"
+	"testing"
+
+	"github.com/DIMO-Network/dq/internal/graph/model"
+	"github.com/DIMO-Network/token-exchange-api/pkg/tokenclaims"
+)
+
+const (
+	testSource = "0xAAaaAAaaAAaaAAaaAAaaAAaaAAaaAAaaAAaaAAaa"
+	otherSrc   = "0xBBbbBBbbBBbbBBbbBBbbBBbbBBbbBBbbBBbbBBbb"
+	idOne      = "id-1"
+	idTwo      = "id-2"
+	idThree    = "id-3"
+)
+
+func strPtr(s string) *string { return &s }
+
+func TestCloudEventRequestAllowed(t *testing.T) {
+	wildcard := &tokenclaims.CloudEvents{Events: []tokenclaims.Event{{
+		EventType: tokenclaims.GlobalIdentifier,
+		Source:    tokenclaims.GlobalIdentifier,
+		IDs:       []string{tokenclaims.GlobalIdentifier},
+	}}}
+	attFromSource := &tokenclaims.CloudEvents{Events: []tokenclaims.Event{{
+		EventType: "dimo.attestation",
+		Source:    testSource,
+		IDs:       []string{tokenclaims.GlobalIdentifier},
+	}}}
+	specificIDs := &tokenclaims.CloudEvents{Events: []tokenclaims.Event{{
+		EventType: "dimo.attestation",
+		Source:    testSource,
+		IDs:       []string{idOne, idTwo},
+	}}}
+
+	tests := []struct {
+		name   string
+		grants *tokenclaims.CloudEvents
+		filter *model.CloudEventFilter
+		want   bool
+	}{
+		{
+			name:   "nil grants denied",
+			grants: nil,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation")},
+			want:   false,
+		},
+		{
+			name:   "empty events denied",
+			grants: &tokenclaims.CloudEvents{},
+			filter: nil,
+			want:   false,
+		},
+		{
+			name:   "wildcard grant, no filter, allowed",
+			grants: wildcard,
+			filter: nil,
+			want:   true,
+		},
+		{
+			name:   "wildcard grant, specific filter, allowed",
+			grants: wildcard,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.status"), Source: strPtr(otherSrc), ID: strPtr(idThree)},
+			want:   true,
+		},
+		{
+			name:   "narrow grant, matching filter, allowed",
+			grants: attFromSource,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation"), Source: strPtr(testSource)},
+			want:   true,
+		},
+		{
+			name:   "narrow grant, no filter, denied (defaults to wildcards)",
+			grants: attFromSource,
+			filter: nil,
+			want:   false,
+		},
+		{
+			name:   "narrow grant, missing source filter, denied",
+			grants: attFromSource,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation")},
+			want:   false,
+		},
+		{
+			name:   "narrow grant, wrong source, denied",
+			grants: attFromSource,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation"), Source: strPtr(otherSrc)},
+			want:   false,
+		},
+		{
+			name:   "narrow grant, wrong type, denied",
+			grants: attFromSource,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.status"), Source: strPtr(testSource)},
+			want:   false,
+		},
+		{
+			name:   "specific ids, granted id, allowed",
+			grants: specificIDs,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation"), Source: strPtr(testSource), ID: strPtr(idOne)},
+			want:   true,
+		},
+		{
+			name:   "specific ids, ungranted id, denied",
+			grants: specificIDs,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation"), Source: strPtr(testSource), ID: strPtr(idThree)},
+			want:   false,
+		},
+		{
+			name:   "specific ids, no id filter, denied (defaults to wildcard id)",
+			grants: specificIDs,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation"), Source: strPtr(testSource)},
+			want:   false,
+		},
+		{
+			name:   "plural types, every type covered, allowed",
+			grants: wildcard,
+			filter: &model.CloudEventFilter{Types: []string{"dimo.attestation", "dimo.status"}},
+			want:   true,
+		},
+		{
+			name:   "plural types, one type uncovered, denied",
+			grants: attFromSource,
+			filter: &model.CloudEventFilter{Types: []string{"dimo.attestation", "dimo.status"}, Source: strPtr(testSource)},
+			want:   false,
+		},
+		{
+			name:   "type and types merged, all covered, allowed",
+			grants: wildcard,
+			filter: &model.CloudEventFilter{Type: strPtr("dimo.attestation"), Types: []string{"dimo.status"}},
+			want:   true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := cloudEventRequestAllowed(tt.grants, tt.filter); got != tt.want {
+				t.Errorf("cloudEventRequestAllowed() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestRequireSubjectOptsByDID(t *testing.T) {
+	const subject = "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:42"
+
+	rawDataToken := &tokenclaims.Token{CustomClaims: tokenclaims.CustomClaims{
+		Asset:       subject,
+		Permissions: []string{tokenclaims.PermissionGetRawData},
+	}}
+	locComboToken := &tokenclaims.Token{CustomClaims: tokenclaims.CustomClaims{
+		Asset:       subject,
+		Permissions: []string{tokenclaims.PermissionGetLocationHistory, tokenclaims.PermissionGetNonLocationHistory},
+	}}
+	grantOnlyToken := &tokenclaims.Token{CustomClaims: tokenclaims.CustomClaims{
+		Asset: subject,
+		CloudEvents: &tokenclaims.CloudEvents{Events: []tokenclaims.Event{{
+			EventType: "dimo.attestation",
+			Source:    tokenclaims.GlobalIdentifier,
+			IDs:       []string{tokenclaims.GlobalIdentifier},
+		}}},
+	}}
+	noAccessToken := &tokenclaims.Token{CustomClaims: tokenclaims.CustomClaims{
+		Asset:       subject,
+		Permissions: []string{tokenclaims.PermissionExecuteCommands},
+	}}
+
+	tests := []struct {
+		name    string
+		token   *tokenclaims.Token
+		reqDID  string
+		filter  *model.CloudEventFilter
+		wantErr bool
+	}{
+		{
+			name:    "no token claims denied",
+			token:   nil,
+			reqDID:  subject,
+			wantErr: true,
+		},
+		{
+			name:    "raw-data token, any filter, allowed",
+			token:   rawDataToken,
+			reqDID:  subject,
+			filter:  &model.CloudEventFilter{Type: strPtr("dimo.status")},
+			wantErr: false,
+		},
+		{
+			name:    "location combo token, no filter, allowed",
+			token:   locComboToken,
+			reqDID:  subject,
+			wantErr: false,
+		},
+		{
+			name:    "grant-only token, covered filter, allowed",
+			token:   grantOnlyToken,
+			reqDID:  subject,
+			filter:  &model.CloudEventFilter{Type: strPtr("dimo.attestation")},
+			wantErr: false,
+		},
+		{
+			name:    "grant-only token, uncovered filter, denied",
+			token:   grantOnlyToken,
+			reqDID:  subject,
+			filter:  &model.CloudEventFilter{Type: strPtr("dimo.status")},
+			wantErr: true,
+		},
+		{
+			name:    "no access token denied",
+			token:   noAccessToken,
+			reqDID:  subject,
+			filter:  &model.CloudEventFilter{Type: strPtr("dimo.attestation")},
+			wantErr: true,
+		},
+		{
+			name:    "raw-data token, mismatched subject, denied",
+			token:   rawDataToken,
+			reqDID:  "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:99",
+			filter:  &model.CloudEventFilter{Type: strPtr("dimo.status")},
+			wantErr: true,
+		},
+	}
+
+	r := &queryResolver{&Resolver{}}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			if tt.token != nil {
+				ctx = context.WithValue(ctx, ClaimsContextKey{}, tt.token)
+			}
+			_, err := r.requireSubjectOptsByDID(ctx, tt.reqDID, tt.filter)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("requireSubjectOptsByDID() err = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}