Events API uses consistent error codes

http/events.go

@@ -50,7 +50,10 @@ func handleEventsSubscribeWebsocket(args eventSubscribeArgs) (websocket.StatusCo
 			logger.Debug("Sending message to websocket", "message", message)
 			var messageBytes []byte
 			if args.json {
-				messageBytes, err = protojson.Marshal(message)
+				opts := protojson.MarshalOptions{
+					UseProtoNames: true,
+				}
+				messageBytes, err = opts.Marshal(message)
 			} else {
 				messageBytes, err = proto.Marshal(message)
 			}
@@ -78,7 +81,7 @@ func handleEventsSubscribe(core *vault.Core, req *logical.Request) http.Handler
 		_, _, err := core.CheckToken(ctx, req, false)
 		if err != nil {
 			if errors.Is(err, logical.ErrPermissionDenied) {
-				respondError(w, http.StatusUnauthorized, logical.ErrPermissionDenied)
+				respondError(w, http.StatusForbidden, logical.ErrPermissionDenied)
 				return
 			}
 			logger.Debug("Error validating token", "error", err)

http/events_test.go

@@ -2,13 +2,15 @@ package http
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/vault/api"
 	"github.com/hashicorp/vault/helper/namespace"
 	"github.com/hashicorp/vault/sdk/logical"
 	"github.com/hashicorp/vault/vault"
@@ -33,7 +35,7 @@ func TestEventsSubscribe(t *testing.T) {
 
 	stop := atomic.Bool{}
 
-	eventType := "abc"
+	const eventType = "abc"
 
 	// send some events
 	go func() {
@@ -42,7 +44,10 @@ func TestEventsSubscribe(t *testing.T) {
 			if err != nil {
 				core.Logger().Info("Error generating UUID, exiting sender", "error", err)
 			}
-			err = core.Events().SendInternal(namespace.RootContext(context.Background()), namespace.RootNamespace, nil, logical.EventType(eventType), &logical.EventData{
+			pluginInfo := &logical.EventPluginInfo{
+				MountPath: "secret",
+			}
+			err = core.Events().SendInternal(namespace.RootContext(context.Background()), namespace.RootNamespace, pluginInfo, logical.EventType(eventType), &logical.EventData{
 				Id:        id,
 				Metadata:  nil,
 				EntityIds: nil,
@@ -59,19 +64,9 @@ func TestEventsSubscribe(t *testing.T) {
 		stop.Store(true)
 	})
 
-	ctx, cancelFunc := context.WithTimeout(context.Background(), 5*time.Second)
-	t.Cleanup(cancelFunc)
-
+	ctx := context.Background()
 	wsAddr := strings.Replace(addr, "http", "ws", 1)
 
-	// check that the connection fails if we don't have a token
-	_, _, err := websocket.Dial(ctx, wsAddr+"/v1/sys/events/subscribe/"+eventType+"?json=true", nil)
-	if err == nil {
-		t.Error("Expected websocket error but got none")
-	} else if !strings.HasSuffix(err.Error(), "401") {
-		t.Errorf("Expected 401 websocket but got %v", err)
-	}
-
 	conn, _, err := websocket.Dial(ctx, wsAddr+"/v1/sys/events/subscribe/"+eventType+"?json=true", &websocket.DialOptions{
 		HTTPHeader: http.Header{"x-vault-token": []string{token}},
 	})
@@ -83,8 +78,80 @@ func TestEventsSubscribe(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	msgJson := strings.TrimSpace(string(msg))
-	if !strings.HasPrefix(msgJson, "{") || !strings.HasSuffix(msgJson, "}") {
-		t.Errorf("Expected to get JSON event but got: %v", msgJson)
+	data := map[string]interface{}{}
+	err = json.Unmarshal(msg, &data)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if actualType := data["event_type"].(string); actualType != eventType {
+		t.Fatalf("Expeced event type %s, got %s", eventType, actualType)
+	}
+	pluginInfo, ok := data["plugin_info"].(map[string]interface{})
+	if !ok || pluginInfo == nil {
+		t.Fatalf("No plugin_info object: %v", data)
+	}
+	mountPath, ok := pluginInfo["mount_path"].(string)
+	if !ok || mountPath != "secret" {
+		t.Fatalf("Wrong mount_path: %v", data)
+	}
+}
+
+func TestEventsSubscribeAuth(t *testing.T) {
+	core := vault.TestCore(t)
+	ln, addr := TestServer(t, core)
+	defer ln.Close()
+
+	// unseal the core
+	keys, root := vault.TestCoreInit(t, core)
+	for _, key := range keys {
+		_, err := core.Unseal(key)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	var nonPrivilegedToken string
+	// Fetch a valid non privileged token.
+	{
+		config := api.DefaultConfig()
+		config.Address = addr
+
+		client, err := api.NewClient(config)
+		if err != nil {
+			t.Fatal(err)
+		}
+		client.SetToken(root)
+
+		secret, err := client.Auth().Token().Create(&api.TokenCreateRequest{Policies: []string{"default"}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if secret.Auth.ClientToken == "" {
+			t.Fatal("Failed to fetch a non privileged token")
+		}
+		nonPrivilegedToken = secret.Auth.ClientToken
+	}
+
+	ctx := context.Background()
+	wsAddr := strings.Replace(addr, "http", "ws", 1)
+
+	// Get a 403 with no token.
+	_, resp, err := websocket.Dial(ctx, wsAddr+"/v1/sys/events/subscribe/abc", nil)
+	if err == nil {
+		t.Error("Expected websocket error but got none")
+	}
+	if resp == nil || resp.StatusCode != http.StatusForbidden {
+		t.Errorf("Expected 403 but got %+v", resp)
+	}
+
+	// Get a 403 with a non privileged token.
+	_, resp, err = websocket.Dial(ctx, wsAddr+"/v1/sys/events/subscribe/abc", &websocket.DialOptions{
+		HTTPHeader: http.Header{"x-vault-token": []string{nonPrivilegedToken}},
+	})
+	if err == nil {
+		t.Error("Expected websocket error but got none")
+	}
+	if resp == nil || resp.StatusCode != http.StatusForbidden {
+		t.Errorf("Expected 403 but got %+v", resp)
 	}
 }