feat(node)!: [high] support JWT auth token revocation

Makefile

@@ -156,7 +156,8 @@ lint: lint-imports
 ## test-unit: Run unit tests.
 test-unit:
 	@echo "--> Running unit tests"
-	@go test $(VERBOSE) -tags=fibre -covermode=atomic -coverprofile=coverage.txt `go list ./... | grep -v nodebuilder/tests` $(LOG_AND_FILTER)
+# Test-node port allocation is process-local, so coverage packages must run serially.
+	@go test $(VERBOSE) -p 1 -tags=fibre -covermode=atomic -coverprofile=coverage.txt `go list ./... | grep -v nodebuilder/tests` $(LOG_AND_FILTER)
 .PHONY: test-unit
 
 ## test-unit-fast: Run unit tests without coverage instrumentation.

README.md

@@ -68,6 +68,14 @@ For more information on setting up a node and the hardware requirements needed,
 
 The celestia-node public API is documented [here](https://node-rpc-docs.celestia.org/).
 
+### Revoke an RPC token
+
+Use an admin token to revoke a previously issued token on a running node:
+
+```sh
+celestia node revoke-token "$TOKEN_TO_REVOKE" --url http://localhost:26658 --token "$ADMIN_TOKEN"
+```
+
 ## Node types
 
 - **Bridge** nodes - relay blocks from the celestia consensus network to the celestia data availability (DA) network

api/rpc/server.go

@@ -75,7 +75,7 @@ type Server struct {
 	tlsKeyPath  string
 
 	signer   jwt.Signer
-	verifier jwt.Verifier
+	verifier *authtoken.Verifier
 
 	metrics *rpcMetrics
 }
@@ -93,7 +93,7 @@ func NewServer(
 	tlsConfig TLSConfig,
 	rateLimitCfg RateLimitConfig,
 	signer jwt.Signer,
-	verifier jwt.Verifier,
+	verifier *authtoken.Verifier,
 ) *Server {
 	srv := &Server{
 		signer:       signer,
@@ -181,22 +181,28 @@ func (s *Server) WithMetrics() error {
 	return nil
 }
 
-// verifyAuth is the RPC server's auth middleware. This middleware is only
-// reached if a token is provided in the header of the request, otherwise only
-// methods with `read` permissions are accessible.
-func (s *Server) verifyAuth(_ context.Context, token string) ([]auth.Permission, error) {
-	if s.authDisabled {
-		return perms.AllPerms, nil
-	}
-	return authtoken.ExtractSignedPermissions(s.verifier, token)
-}
-
 // authHandler wraps the handler with authentication.
 func (s *Server) authHandler(next http.Handler) http.Handler {
-	return &auth.Handler{
-		Verify: s.verifyAuth,
-		Next:   next.ServeHTTP,
-	}
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx, cancel := context.WithCancel(r.Context())
+		defer cancel()
+		var release func()
+		defer func() {
+			if release != nil {
+				release()
+			}
+		}()
+		handler := auth.Handler{
+			Verify: func(ctx context.Context, token string) ([]auth.Permission, error) {
+				permissions, unregister, err := s.verifier.VerifyAndRegister(ctx, token, cancel)
+				release = unregister
+				return permissions, err
+			},
+			Next: next.ServeHTTP,
+		}
+		// go-jsonrpc closes WebSockets and subscriptions when this context is canceled.
+		handler.ServeHTTP(w, r.WithContext(ctx))
+	})
 }
 
 // corsAny applies permissive CORS (allows all origins, methods, headers)

api/rpc/server_test.go

@@ -15,16 +15,20 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"slices"
 	"testing"
 	"time"
 
 	"github.com/cristalhq/jwt/v5"
 	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-jsonrpc/auth"
+	"github.com/ipfs/go-datastore"
+	ds_sync "github.com/ipfs/go-datastore/sync"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/celestiaorg/celestia-node/api/rpc/perms"
+	"github.com/celestiaorg/celestia-node/libs/authtoken"
 )
 
 // TestServer_HandlerStackSelection tests that the correct middleware stack is selected
@@ -265,8 +269,8 @@ func TestServer_AuthMiddleware(t *testing.T) {
 	}
 }
 
-// TestServer_VerifyAuth tests the auth verification logic
-func TestServer_VerifyAuth(t *testing.T) {
+// TestServer_AuthPermissions tests permissions passed through the auth middleware.
+func TestServer_AuthPermissions(t *testing.T) {
 	signer, verifier := createTestJWT(t)
 
 	tests := []struct {
@@ -311,13 +315,27 @@ func TestServer_VerifyAuth(t *testing.T) {
 				verifier,
 			)
 
-			permissions, err := server.verifyAuth(context.Background(), tt.token)
-
+			called := false
+			handler := server.newHandlerStack(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				called = true
+				for _, permission := range perms.AllPerms {
+					assert.Equal(t, slices.Contains(tt.expectedPerms, permission),
+						auth.HasPerm(r.Context(), perms.AllPerms, permission))
+				}
+				w.WriteHeader(http.StatusOK)
+			}))
+			req := httptest.NewRequest(http.MethodGet, "/", nil)
+			if tt.token != "" {
+				req.Header.Set("Authorization", "Bearer "+tt.token)
+			}
+			w := httptest.NewRecorder()
+			handler.ServeHTTP(w, req)
 			if tt.expectError {
-				assert.Error(t, err)
+				assert.Equal(t, http.StatusUnauthorized, w.Code)
+				assert.False(t, called)
 			} else {
-				assert.NoError(t, err)
-				assert.Equal(t, tt.expectedPerms, permissions)
+				assert.Equal(t, http.StatusOK, w.Code)
+				assert.True(t, called)
 			}
 		})
 	}
@@ -507,15 +525,15 @@ func generateSelfSignedCert(t *testing.T) (certPath, keyPath string) {
 	return certPath, keyPath
 }
 
-func createTestJWT(t *testing.T) (jwt.Signer, jwt.Verifier) {
+func createTestJWT(t *testing.T) (jwt.Signer, *authtoken.Verifier) {
 	key := []byte("test-secret-key-for-jwt-signing-32")
 	signer, err := jwt.NewSignerHS(jwt.HS256, key)
 	require.NoError(t, err)
 
 	verifier, err := jwt.NewVerifierHS(jwt.HS256, key)
 	require.NoError(t, err)
 
-	return signer, verifier
+	return signer, authtoken.NewVerifier(verifier, ds_sync.MutexWrap(datastore.NewMapDatastore()))
 }
 
 func createValidToken(t *testing.T, signer jwt.Signer, permissions []auth.Permission) string {

libs/authtoken/authtoken.go

@@ -13,12 +13,16 @@ import (
 )
 
 // ExtractSignedPermissions returns the permissions granted to the token by the passed signer.
-// If the token isn't signed by the signer, it will not pass verification.
+// It checks signature and expiry only; use Verifier.Verify to also check revocation.
 func ExtractSignedPermissions(verifier jwt.Verifier, token string) ([]auth.Permission, error) {
 	tk, err := jwt.Parse([]byte(token), verifier)
 	if err != nil {
 		return nil, err
 	}
+	return extractPermissions(tk)
+}
+
+func extractPermissions(tk *jwt.Token) ([]auth.Permission, error) {
 	p := new(perms.JWTPayload)
 
 	if err := json.Unmarshal(tk.Claims(), p); err != nil {

libs/authtoken/verifier.go

@@ -0,0 +1,121 @@
+package authtoken
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/cristalhq/jwt/v5"
+	"github.com/filecoin-project/go-jsonrpc/auth"
+	"github.com/ipfs/go-datastore"
+)
+
+// ErrRevoked indicates that a token has been revoked.
+var ErrRevoked = errors.New("token revoked")
+
+// Verifier checks JWT signatures, expiry and persistent revocations.
+// Active requests are canceled when their token is revoked.
+type Verifier struct {
+	verifier jwt.Verifier
+	store    datastore.Datastore
+	mu       sync.Mutex
+	sessions map[datastore.Key]map[context.Context]context.CancelFunc
+}
+
+// NewVerifier uses the node's datastore to persist revocations across restarts.
+func NewVerifier(verifier jwt.Verifier, store datastore.Datastore) *Verifier {
+	return &Verifier{
+		verifier: verifier,
+		store:    store,
+		sessions: make(map[datastore.Key]map[context.Context]context.CancelFunc),
+	}
+}
+
+// Verify returns the permissions of a valid, unrevoked token.
+func (v *Verifier) Verify(ctx context.Context, token string) ([]auth.Permission, error) {
+	tk, err := jwt.Parse([]byte(token), v.verifier)
+	if err != nil {
+		return nil, err
+	}
+	return v.verify(ctx, tk)
+}
+
+// VerifyAndRegister verifies a token and cancels the request when it is revoked.
+// Call the returned function when the request finishes. The context must be unique per request.
+func (v *Verifier) VerifyAndRegister(
+	ctx context.Context, token string, cancel context.CancelFunc,
+) ([]auth.Permission, func(), error) {
+	tk, err := jwt.Parse([]byte(token), v.verifier)
+	if err != nil {
+		return nil, nil, err
+	}
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	// Checking under the same lock as Revoke closes the authentication/registration race.
+	permissions, err := v.verify(ctx, tk)
+	if err != nil {
+		return nil, nil, err
+	}
+	key := tokenKey(tk)
+	if v.sessions[key] == nil {
+		v.sessions[key] = make(map[context.Context]context.CancelFunc)
+	}
+	v.sessions[key][ctx] = cancel
+	return permissions, func() {
+		v.mu.Lock()
+		defer v.mu.Unlock()
+		delete(v.sessions[key], ctx)
+		if len(v.sessions[key]) == 0 {
+			delete(v.sessions, key)
+		}
+	}, nil
+}
+
+// Revoke permanently invalidates a token signed by this node, including tokens
+// issued before nonces were introduced. Repeated revocations are safe.
+func (v *Verifier) Revoke(ctx context.Context, token string) error {
+	tk, err := jwt.Parse([]byte(token), v.verifier)
+	if err != nil {
+		return err
+	}
+	key := tokenKey(tk)
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	if err := v.store.Put(ctx, key, nil); err != nil {
+		return fmt.Errorf("persist token revocation: %w", err)
+	}
+	err = v.store.Sync(ctx, key)
+	// Even if syncing fails, reject the token and terminate its active sessions.
+	for _, cancel := range v.sessions[key] {
+		cancel()
+	}
+	delete(v.sessions, key)
+	if err != nil {
+		return fmt.Errorf("sync token revocation: %w", err)
+	}
+	return nil
+}
+
+func (v *Verifier) verify(ctx context.Context, tk *jwt.Token) ([]auth.Permission, error) {
+	permissions, err := extractPermissions(tk)
+	if err != nil {
+		return nil, err
+	}
+	revoked, err := v.store.Has(ctx, tokenKey(tk))
+	if err != nil {
+		return nil, fmt.Errorf("check token revocation: %w", err)
+	}
+	if revoked {
+		return nil, ErrRevoked
+	}
+	return permissions, nil
+}
+
+func tokenKey(tk *jwt.Token) datastore.Key {
+	// Hash decoded signatures so alternate base64 encodings cannot bypass revocation.
+	digest := sha256.Sum256(tk.Signature())
+	return datastore.NewKey("/auth/revoked/" + hex.EncodeToString(digest[:]))
+}