package main

import (

"encoding/json"

"fmt"

"sync"

"sync/atomic"

"testing"

"time"

"github.com/daptin/daptin/server/resource"

"github.com/imroc/req"

"golang.org/x/net/websocket"

)

const wsBaseAddress = "http://localhost:6337"

const wsURL = "ws://localhost:6337/live"

var wsServerOnce sync.Once

func ensureServer() {

wsServerOnce.Do(func() {

createServer()

})

}

// wsPayload matches WebSocketPayload on the server side.

type wsPayload struct {

Id string `json:"id,omitempty"`

Method string `json:"method"`

Payload map[string]interface{} `json:"attributes"`

}

var wsReqCounter atomic.Int64

func nextReqId() string {

return fmt.Sprintf("req-%d", wsReqCounter.Add(1))

}

// dialWS opens an authenticated websocket connection with retry for transient failures.

// It consumes the initial session-open message before returning.

func dialWS(t testing.TB, token string) *websocket.Conn {

t.Helper()

for attempt := 0; attempt < 5; attempt++ {

config, err := websocket.NewConfig(wsURL, wsBaseAddress)

if err != nil {

t.Fatalf("websocket config: %v", err)

}

config.Header.Set("Authorization", "Bearer "+token)

config.Header.Set("Cookie", "token="+token)

ws, err := websocket.DialConfig(config)

if err != nil {

// per-IP limiter may reject during high churn — retry after backoff

time.Sleep(time.Duration(50*(attempt+1)) * time.Millisecond)

continue

}

// consume the session-open message

ws.SetReadDeadline(time.Now().Add(5 * time.Second))

var sessionMsg resource.WsOutMessage

if err := websocket.JSON.Receive(ws, &sessionMsg); err != nil {

t.Fatalf("failed to read session-open: %v", err)

}

if sessionMsg.Type != "session" || sessionMsg.Status != "open" {

t.Fatalf("expected session-open, got type=%q status=%q", sessionMsg.Type, sessionMsg.Status)

}

return ws

}

t.Fatalf("websocket dial: failed after 5 attempts")

return nil

}

// dialWSRaw opens an authenticated websocket without consuming session-open.

func dialWSRaw(t testing.TB, token string) *websocket.Conn {

t.Helper()

for attempt := 0; attempt < 5; attempt++ {

config, err := websocket.NewConfig(wsURL, wsBaseAddress)

if err != nil {

t.Fatalf("websocket config: %v", err)

}

config.Header.Set("Authorization", "Bearer "+token)

config.Header.Set("Cookie", "token="+token)

ws, err := websocket.DialConfig(config)

if err != nil {

time.Sleep(time.Duration(50*(attempt+1)) * time.Millisecond)

continue

}

return ws

}

t.Fatalf("websocket dial: failed after 5 attempts")

return nil

}

// sendJSON sends a JSON payload over the websocket.

func sendJSON(t testing.TB, ws *websocket.Conn, v interface{}) {

t.Helper()

if err := websocket.JSON.Send(ws, v); err != nil {

t.Fatalf("websocket send: %v", err)

}

}

// recvJSON reads a JSON message with a timeout.

func recvJSON(t testing.TB, ws *websocket.Conn, timeout time.Duration) resource.WsOutMessage {

t.Helper()

ws.SetReadDeadline(time.Now().Add(timeout))

var msg resource.WsOutMessage

if err := websocket.JSON.Receive(ws, &msg); err != nil {

t.Fatalf("websocket recv: %v", err)

}

return msg

}

// tryRecvJSON reads a JSON message with a timeout, returns ok=false on timeout.

func tryRecvJSON(ws *websocket.Conn, timeout time.Duration) (resource.WsOutMessage, bool) {

ws.SetReadDeadline(time.Now().Add(timeout))

var msg resource.WsOutMessage

if err := websocket.JSON.Receive(ws, &msg); err != nil {

return msg, false

}

return msg, true

}

var wsTokenOnce sync.Once

var wsToken string

// signUpAndGetToken returns a JWT token, creating a user if needed.

// Works both standalone (guest signup open) and after TestServerApis (guest signup locked).

func signUpAndGetToken(t testing.TB) string {

if t != nil {

t.Helper()

}

wsTokenOnce.Do(func() {

client := req.New()

client.SetTimeout(30 * time.Second)

// Try signing in as test@gmail.com (created by TestServerApis)

resp, err := client.Post(wsBaseAddress+"/action/user_account/signin", req.BodyJSON(map[string]interface{}{

"attributes": map[string]interface{}{

"email": "test@gmail.com",

"password": "tester123",

},

}))

if err == nil {

if tok := extractToken(resp); tok != "" {

wsToken = tok

return

}

}

// TestServerApis hasn't run — sign up as guest

client.Post(wsBaseAddress+"/action/user_account/signup", req.BodyJSON(map[string]interface{}{

"attributes": map[string]interface{}{

"email": "test@gmail.com",

"name": "test",

"password": "tester123",

"passwordConfirm": "tester123",

},

}))

resp, err = client.Post(wsBaseAddress+"/action/user_account/signin", req.BodyJSON(map[string]interface{}{

"attributes": map[string]interface{}{

"email": "test@gmail.com",

"password": "tester123",

},

}))

if err != nil {

panic(fmt.Sprintf("signin failed: %v", err))

}

wsToken = extractToken(resp)

if wsToken == "" {

panic("no token after signup+signin")

}

})

return wsToken

}

func extractToken(resp *req.Resp) string {

var signInResp []interface{}

resp.ToJSON(&signInResp)

for _, item := range signInResp {

attrs, ok := item.(map[string]interface{})

if ok && attrs["ResponseType"] == "client.store.set" {

return attrs["Attributes"].(map[string]interface{})["value"].(string)

}

}

return ""

}

// ===== E2E TESTS =====

func TestWebSocketSessionOpen(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWSRaw(t, token)

defer ws.Close()

// first message should be session-open

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "session" {

t.Errorf("expected type=session, got %q", msg.Type)

}

if msg.Status != "open" {

t.Errorf("expected status=open, got %q", msg.Status)

}

var data map[string]interface{}

json.Unmarshal(msg.Data, &data)

if data["user"] == nil {

t.Errorf("session-open missing user field")

}

if data["sessionId"] == nil {

t.Errorf("session-open missing sessionId field")

}

t.Logf("session-open: user=%v sessionId=%v", data["user"], data["sessionId"])

}

func TestWebSocketConnect(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

// subscribe to a known system topic to confirm bidirectional communication

id := nextReqId()

sendJSON(t, ws, wsPayload{

Id: id,

Method: "subscribe",

Payload: map[string]interface{}{"topicName": "user_account"},

})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" {

t.Errorf("expected type=response, got %q", msg.Type)

}

if msg.Method != "subscribe" {

t.Errorf("expected method=subscribe, got %q", msg.Method)

}

if msg.Id != id {

t.Errorf("expected id=%q, got %q", id, msg.Id)

}

}

func TestWebSocketSubscribeAck(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

topicName := "user_account"

t.Logf("subscribing to topic: %s", topicName)

id := nextReqId()

sendJSON(t, ws, wsPayload{

Id: id,

Method: "subscribe",

Payload: map[string]interface{}{"topicName": topicName},

})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || !*msg.Ok {

t.Errorf("expected successful response, got type=%q ok=%v", msg.Type, msg.Ok)

}

if msg.Method != "subscribe" {

t.Errorf("expected method=subscribe, got %q", msg.Method)

}

if msg.Id != id {

t.Errorf("expected id=%q, got %q", id, msg.Id)

}

var data map[string]interface{}

json.Unmarshal(msg.Data, &data)

if data["topic"] != topicName {

t.Errorf("expected topic=%q in data, got %v", topicName, data["topic"])

}

}

func TestWebSocketSubscribeNonexistentTopicError(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

id := nextReqId()

sendJSON(t, ws, wsPayload{

Id: id,

Method: "subscribe",

Payload: map[string]interface{}{"topicName": "nonexistent_topic_xyz_12345"},

})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" {

t.Errorf("expected type=response, got %q", msg.Type)

}

if msg.Ok == nil || *msg.Ok {

t.Errorf("expected ok=false, got %v", msg.Ok)

}

if msg.Method != "subscribe" {

t.Errorf("expected method=subscribe, got %q", msg.Method)

}

}

func TestWebSocketUnsubscribeAck(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

topicName := "user_account"

// subscribe first

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

recvJSON(t, ws, 5*time.Second) // subscribe ack

// unsubscribe

id := nextReqId()

sendJSON(t, ws, wsPayload{Id: id, Method: "unsubscribe", Payload: map[string]interface{}{"topicName": topicName}})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || !*msg.Ok {

t.Errorf("expected successful unsubscribe response, got type=%q ok=%v", msg.Type, msg.Ok)

}

if msg.Method != "unsubscribe" {

t.Errorf("expected method=unsubscribe, got %q", msg.Method)

}

var data map[string]interface{}

json.Unmarshal(msg.Data, &data)

if data["topic"] != topicName {

t.Errorf("expected topic=%q, got %v", topicName, data["topic"])

}

}

func TestWebSocketNewMessageErrors(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

// missing topicName

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "new-message", Payload: map[string]interface{}{}})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || *msg.Ok {

t.Errorf("expected error response for missing topicName, got type=%q ok=%v", msg.Type, msg.Ok)

}

// nonexistent topic

sendJSON(t, ws, wsPayload{

Id: nextReqId(),

Method: "new-message",

Payload: map[string]interface{}{

"topicName": "does_not_exist_xyz",

"message": map[string]interface{}{"text": "hello"},

},

})

msg = recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || *msg.Ok {

t.Errorf("expected error response for nonexistent topic, got type=%q ok=%v", msg.Type, msg.Ok)

}

}

func TestWebSocketCreateDestroyTopic(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

topicName := fmt.Sprintf("test-topic-%d", time.Now().UnixNano())

// create topic

id := nextReqId()

sendJSON(t, ws, wsPayload{

Id: id,

Method: "create-topicName",

Payload: map[string]interface{}{"name": topicName},

})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || !*msg.Ok {

t.Errorf("expected successful create response, got type=%q ok=%v error=%q", msg.Type, msg.Ok, msg.Error)

}

// destroy it

sendJSON(t, ws, wsPayload{

Id: nextReqId(),

Method: "destroy-topicName",

Payload: map[string]interface{}{"name": topicName},

})

msg = recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || !*msg.Ok {

t.Errorf("expected successful destroy response, got type=%q ok=%v error=%q", msg.Type, msg.Ok, msg.Error)

}

}

func TestWebSocketDestroySystemTopicError(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

// "world" is always a system topic

sendJSON(t, ws, wsPayload{

Id: nextReqId(),

Method: "destroy-topicName",

Payload: map[string]interface{}{"name": "world"},

})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || *msg.Ok {

t.Errorf("expected error for system topic delete, got type=%q ok=%v", msg.Type, msg.Ok)

}

}

func TestWebSocketCreateDuplicateTopicError(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

topicName := fmt.Sprintf("dup-topic-%d", time.Now().UnixNano())

// create first

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

recvJSON(t, ws, 5*time.Second) // success response

// create duplicate

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || *msg.Ok {

t.Errorf("expected error for duplicate topic, got type=%q ok=%v", msg.Type, msg.Ok)

}

}

func TestWebSocketPingPong(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

sendJSON(t, ws, wsPayload{Method: "ping", Payload: map[string]interface{}{}})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "pong" {

t.Errorf("expected type=pong, got %q", msg.Type)

}

}

func TestWebSocketPubSubRoundTrip(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

topicName := fmt.Sprintf("pubsub-rt-%d", time.Now().UnixNano())

// publisher connection

wsPub := dialWS(t, token)

defer wsPub.Close()

// subscriber connection

wsSub := dialWS(t, token)

defer wsSub.Close()

// create topic on publisher

sendJSON(t, wsPub, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

recvJSON(t, wsPub, 5*time.Second) // create response

time.Sleep(300 * time.Millisecond)

// subscribe on subscriber

sendJSON(t, wsSub, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

subAck := recvJSON(t, wsSub, 5*time.Second)

if subAck.Type != "response" || subAck.Ok == nil || !*subAck.Ok {

t.Fatalf("expected successful subscribe, got type=%q ok=%v", subAck.Type, subAck.Ok)

}

// publish a message

sendJSON(t, wsPub, wsPayload{

Id: nextReqId(),

Method: "new-message",

Payload: map[string]interface{}{

"topicName": topicName,

"message": map[string]interface{}{"text": "hello world"},

},

})

// subscriber should receive an event

msg := recvJSON(t, wsSub, 5*time.Second)

if msg.Type != "event" {

t.Errorf("expected type=event, got %q", msg.Type)

}

if msg.Event != "new-message" {

t.Errorf("expected event=new-message, got %q", msg.Event)

}

if msg.Topic != topicName {

t.Errorf("expected topic=%q, got %q", topicName, msg.Topic)

}

var payload map[string]interface{}

json.Unmarshal(msg.Data, &payload)

if payload["text"] != "hello world" {

t.Errorf("expected text='hello world', got %v", payload["text"])

}

}

func TestWebSocketNoSuchMethod(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

id := nextReqId()

sendJSON(t, ws, wsPayload{Id: id, Method: "nonexistent-method", Payload: map[string]interface{}{}})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" || msg.Ok == nil || *msg.Ok {

t.Errorf("expected error for unknown method, got type=%q ok=%v", msg.Type, msg.Ok)

}

if msg.Error != "no such method" {

t.Errorf("expected error='no such method', got %q", msg.Error)

}

if msg.Method != "nonexistent-method" {

t.Errorf("expected method echoed back, got %q", msg.Method)

}

}

func TestWebSocketRequestCorrelation(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

// send with id — response should echo id

id := nextReqId()

sendJSON(t, ws, wsPayload{Id: id, Method: "subscribe", Payload: map[string]interface{}{"topicName": "user_account"}})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Id != id {

t.Errorf("expected id=%q echoed, got %q", id, msg.Id)

}

}

func TestWebSocketNoIdOmitted(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

// send without id — response should have empty id (omitted)

sendJSON(t, ws, wsPayload{Method: "subscribe", Payload: map[string]interface{}{"topicName": "user_account"}})

msg := recvJSON(t, ws, 5*time.Second)

if msg.Id != "" {

t.Errorf("expected empty id when not sent, got %q", msg.Id)

}

}

// ===== AUTHORIZATION TESTS =====

// expectResponse reads the next message and checks it's a response with the expected ok value.

func expectResponse(t testing.TB, ws *websocket.Conn, method string, wantOk bool) resource.WsOutMessage {

t.Helper()

msg := recvJSON(t, ws, 5*time.Second)

if msg.Type != "response" {

t.Fatalf("expected type=response, got %q (method=%q)", msg.Type, msg.Method)

}

if msg.Method != method {

t.Fatalf("expected method=%q, got %q", method, msg.Method)

}

if msg.Ok == nil {

t.Fatalf("response has nil Ok for method=%q", method)

}

if *msg.Ok != wantOk {

t.Fatalf("expected ok=%v for method=%q, got ok=%v error=%q", wantOk, method, *msg.Ok, msg.Error)

}

return msg

}

// TestAuthzUserTopicDefaultPermOwnerOnly verifies that a newly created user topic

// has owner-only permissions (UserCRUD|UserExecute). A second user should be denied

// subscribe, publish, destroy, and get-topic-permission.

func TestAuthzUserTopicDefaultPermOwnerOnly(t *testing.T) {

ensureServer()

// user1 (owner) creates a topic

token1 := signUpAndGetToken(t)

ws1 := dialWS(t, token1)

defer ws1.Close()

topicName := fmt.Sprintf("authz-default-%d", time.Now().UnixNano())

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws1, "create-topicName", true)

// user2 (non-owner) tries to interact

token2, err := signUpUser("authz-other@test.com", "tester123")

if err != nil {

t.Fatalf("signup user2: %v", err)

}

ws2 := dialWS(t, token2)

defer ws2.Close()

// subscribe should fail (CanRead denied — no GuestRead bit)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", false)

// new-message should fail (CanExecute denied — no GuestExecute bit)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "new-message", Payload: map[string]interface{}{

"topicName": topicName,

"message": map[string]interface{}{"text": "hello"},

}})

expectResponse(t, ws2, "new-message", false)

// destroy should fail (CanDelete denied)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "destroy-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws2, "destroy-topicName", false)

// get-topic-permission should fail (CanPeek denied — no GuestPeek bit)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "get-topic-permission", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "get-topic-permission", false)

}

// TestAuthzOwnerCanAccessOwnTopic verifies the owner can subscribe, publish,

// get/set permissions, and destroy their own topic.

func TestAuthzOwnerCanAccessOwnTopic(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

topicName := fmt.Sprintf("authz-owner-%d", time.Now().UnixNano())

// create

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws, "create-topicName", true)

// subscribe (CanRead — UserRead set)

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws, "subscribe", true)

// publish (CanExecute — UserExecute set)

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "new-message", Payload: map[string]interface{}{

"topicName": topicName,

"message": map[string]interface{}{"text": "owner msg"},

}})

// owner gets the event back on their subscription — drain it

for {

msg, ok := tryRecvJSON(ws, 2*time.Second)

if !ok {

break

}

if msg.Type == "event" {

continue

}

break

}

// get-topic-permission (CanPeek — UserPeek set)

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "get-topic-permission", Payload: map[string]interface{}{"topicName": topicName}})

resp := expectResponse(t, ws, "get-topic-permission", true)

var permData map[string]interface{}

json.Unmarshal(resp.Data, &permData)

if permData["type"] != "user" {

t.Errorf("expected type=user, got %v", permData["type"])

}

// set-topic-permission (owner check passes)

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(2097151), // ALLOW_ALL

}})

expectResponse(t, ws, "set-topic-permission", true)

// unsubscribe first before destroy

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "unsubscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws, "unsubscribe", true)

// destroy (CanDelete — UserDelete set)

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "destroy-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws, "destroy-topicName", true)

}

// TestAuthzSetPermissionNonOwnerDenied verifies that a non-owner cannot

// set-topic-permission even if they have other access.

func TestAuthzSetPermissionNonOwnerDenied(t *testing.T) {

ensureServer()

token1 := signUpAndGetToken(t)

ws1 := dialWS(t, token1)

defer ws1.Close()

topicName := fmt.Sprintf("authz-setperm-%d", time.Now().UnixNano())

// owner creates topic with ALLOW_ALL so user2 can do most things

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws1, "create-topicName", true)

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(2097151),

}})

expectResponse(t, ws1, "set-topic-permission", true)

// user2 tries set-topic-permission — should fail (owner-only)

token2, err := signUpUser("authz-setperm@test.com", "tester123")

if err != nil {

t.Fatalf("signup user2: %v", err)

}

ws2 := dialWS(t, token2)

defer ws2.Close()

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(0),

}})

resp := expectResponse(t, ws2, "set-topic-permission", false)

if resp.Error != "only owner or admin can change permissions" {

t.Errorf("expected owner-only error, got %q", resp.Error)

}

}

// TestAuthzGrantGuestReadAllowsNonOwnerSubscribe verifies that when the owner

// adds GuestRead to the topic permission, a non-owner can subscribe.

func TestAuthzGrantGuestReadAllowsNonOwnerSubscribe(t *testing.T) {

ensureServer()

token1 := signUpAndGetToken(t)

ws1 := dialWS(t, token1)

defer ws1.Close()

topicName := fmt.Sprintf("authz-gread-%d", time.Now().UnixNano())

// create with default (owner-only)

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws1, "create-topicName", true)

token2, err := signUpUser("authz-gread@test.com", "tester123")

if err != nil {

t.Fatalf("signup user2: %v", err)

}

ws2 := dialWS(t, token2)

defer ws2.Close()

// user2 cannot subscribe yet (no GuestRead)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", false)

// owner grants GuestRead|GuestPeek + keeps UserCRUD|UserExecute

// GuestPeek=1, GuestRead=2, UserCRUD=16256, UserExecute=8192

newPerm := float64(1 | 2 | 16256 | 8192)

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": newPerm,

}})

expectResponse(t, ws1, "set-topic-permission", true)

// user2 can now subscribe

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", true)

// but user2 still cannot publish (no GuestExecute=64)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "new-message", Payload: map[string]interface{}{

"topicName": topicName,

"message": map[string]interface{}{"text": "blocked"},

}})

expectResponse(t, ws2, "new-message", false)

}

// TestAuthzGrantGuestExecuteAllowsNonOwnerPublish verifies that when the owner

// adds GuestExecute to the topic permission, a non-owner can publish.

func TestAuthzGrantGuestExecuteAllowsNonOwnerPublish(t *testing.T) {

ensureServer()

token1 := signUpAndGetToken(t)

ws1 := dialWS(t, token1)

defer ws1.Close()

topicName := fmt.Sprintf("authz-gexec-%d", time.Now().UnixNano())

// create with GuestExecute + owner perms

// GuestExecute=32 (1<<5), UserCRUD|UserExecute=16256

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws1, "create-topicName", true)

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(32 | 16256),

}})

expectResponse(t, ws1, "set-topic-permission", true)

token2, err := signUpUser("authz-gexec@test.com", "tester123")

if err != nil {

t.Fatalf("signup user2: %v", err)

}

ws2 := dialWS(t, token2)

defer ws2.Close()

// user2 can publish (GuestExecute granted)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "new-message", Payload: map[string]interface{}{

"topicName": topicName,

"message": map[string]interface{}{"text": "allowed"},

}})

// new-message doesn't send a response on success for user topics, so we just check no error

// Actually it does - let me check by reading the handler flow

// The handler sends events, not responses for new-message on user topics

// Let's just verify no error response arrives

msg, ok := tryRecvJSON(ws2, 2*time.Second)

if ok && msg.Type == "response" && msg.Ok != nil && !*msg.Ok {

t.Errorf("publish should have been allowed but got error: %q", msg.Error)

}

// user2 cannot subscribe (no GuestRead=2)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", false)

}

// TestAuthzDestroyTopicNonOwnerDenied verifies a non-owner cannot destroy

// a topic even with ALLOW_ALL minus GuestDelete.

func TestAuthzDestroyTopicNonOwnerDenied(t *testing.T) {

ensureServer()

token1 := signUpAndGetToken(t)

ws1 := dialWS(t, token1)

defer ws1.Close()

topicName := fmt.Sprintf("authz-destroy-%d", time.Now().UnixNano())

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws1, "create-topicName", true)

// Grant everything except GuestDelete (bit 16)

// ALLOW_ALL=2097151, GuestDelete=16 → 2097151 & ^16 = 2097135

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(2097151 &^ 16),

}})

expectResponse(t, ws1, "set-topic-permission", true)

token2, err := signUpUser("authz-destroy@test.com", "tester123")

if err != nil {

t.Fatalf("signup user2: %v", err)

}

ws2 := dialWS(t, token2)

defer ws2.Close()

// user2 can subscribe (GuestRead is set)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", true)

// user2 cannot destroy (GuestDelete removed)

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "destroy-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws2, "destroy-topicName", false)

}

// TestAuthzGetTopicPermissionRequiresPeek verifies that get-topic-permission

// requires CanPeek, which for non-owners means GuestPeek must be set.

func TestAuthzGetTopicPermissionRequiresPeek(t *testing.T) {

ensureServer()

token1 := signUpAndGetToken(t)

ws1 := dialWS(t, token1)

defer ws1.Close()

topicName := fmt.Sprintf("authz-getperm-%d", time.Now().UnixNano())

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws1, "create-topicName", true)

token2, err := signUpUser("authz-getperm@test.com", "tester123")

if err != nil {

t.Fatalf("signup user2: %v", err)

}

ws2 := dialWS(t, token2)

defer ws2.Close()

// default perms: no GuestPeek → denied

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "get-topic-permission", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "get-topic-permission", false)

// owner adds GuestPeek (bit 1) to perm

// UserCRUD=16256, UserExecute=8192, GuestPeek=1

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(1 | 16256 | 8192),

}})

expectResponse(t, ws1, "set-topic-permission", true)

// now user2 can get-topic-permission

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "get-topic-permission", Payload: map[string]interface{}{"topicName": topicName}})

resp := expectResponse(t, ws2, "get-topic-permission", true)

var permData map[string]interface{}

json.Unmarshal(resp.Data, &permData)

if permData["type"] != "user" {

t.Errorf("expected type=user, got %v", permData["type"])

}

if permData["owner"] == nil {

t.Errorf("expected owner field in response")

}

}

// TestAuthzSystemTopicCannotCreateOrDestroy verifies system topics cannot be

// created or destroyed via the WebSocket protocol.

func TestAuthzSystemTopicCannotCreateOrDestroy(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

// create-topicName with system name should fail

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": "world"}})

resp := expectResponse(t, ws, "create-topicName", false)

if resp.Error != "cannot create topic with reserved name" {

t.Errorf("expected reserved name error, got %q", resp.Error)

}

// destroy-topicName with system name should fail

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "destroy-topicName", Payload: map[string]interface{}{"name": "user_account"}})

resp = expectResponse(t, ws, "destroy-topicName", false)

if resp.Error != "cannot delete system topic" {

t.Errorf("expected system topic error, got %q", resp.Error)

}

}

// TestAuthzSystemTopicSetPermissionDenied verifies that set-topic-permission

// cannot be used on system topics.

func TestAuthzSystemTopicSetPermissionDenied(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)

ws := dialWS(t, token)

defer ws.Close()

sendJSON(t, ws, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": "world",

"permission": float64(2097151),

}})

resp := expectResponse(t, ws, "set-topic-permission", false)

if resp.Error != "cannot modify system topic permissions" {

t.Errorf("expected system topic error, got %q", resp.Error)

}

}

// TestAuthzPermissionChangeAffectsAccess verifies that when the owner changes

// permissions, subsequent requests from non-owners reflect the new permissions.

func TestAuthzPermissionChangeAffectsAccess(t *testing.T) {

ensureServer()

token1 := signUpAndGetToken(t)

ws1 := dialWS(t, token1)

defer ws1.Close()

topicName := fmt.Sprintf("authz-change-%d", time.Now().UnixNano())

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})

expectResponse(t, ws1, "create-topicName", true)

token2, err := signUpUser("authz-change@test.com", "tester123")

if err != nil {

t.Fatalf("signup user2: %v", err)

}

ws2 := dialWS(t, token2)

defer ws2.Close()

// Step 1: default perms — user2 denied subscribe

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", false)

// Step 2: owner grants GuestRead+GuestPeek

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(1 | 2 | 16256 | 8192), // GuestPeek|GuestRead|UserCRUD|UserExecute

}})

expectResponse(t, ws1, "set-topic-permission", true)

// Step 3: user2 can now subscribe

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", true)

// Step 4: owner revokes GuestRead (back to owner-only)

sendJSON(t, ws1, wsPayload{Id: nextReqId(), Method: "set-topic-permission", Payload: map[string]interface{}{

"topicName": topicName,

"permission": float64(16256 | 8192), // UserCRUD|UserExecute only

}})

expectResponse(t, ws1, "set-topic-permission", true)

// Step 5: user2 unsubscribe (if subscribed) and try again — denied

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "unsubscribe", Payload: map[string]interface{}{"topicName": topicName}})

recvJSON(t, ws2, 5*time.Second) // unsubscribe response

sendJSON(t, ws2, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})

expectResponse(t, ws2, "subscribe", false)

}

// TestAuthzNonexistentTopicPermissionOperations verifies that get/set permission

// on a nonexistent topic returns appropriate errors.

func TestAuthzNonexistentTopicPermissionOperations(t *testing.T) {

ensureServer()

token := signUpAndGetToken(t)