350 lines
15 KiB
Go
350 lines
15 KiB
Go
package controlplane
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestReactFrontendIsEmbedded(t *testing.T) {
|
|
server, err := NewServer(ServerConfig{DataFile: filepath.Join(t.TempDir(), "state.json")})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
httpServer := httptest.NewServer(server.Handler())
|
|
defer httpServer.Close()
|
|
defer server.Close()
|
|
response, err := http.Get(httpServer.URL + "/")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer response.Body.Close()
|
|
body, err := io.ReadAll(response.Body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if response.StatusCode != http.StatusOK || !strings.Contains(string(body), "<div id=\"root\"></div>") || !strings.Contains(string(body), "/assets/") {
|
|
t.Fatalf("React index was not served: status=%d body=%s", response.StatusCode, body)
|
|
}
|
|
if strings.Contains(response.Header.Get("Content-Security-Policy"), "unsafe-inline") {
|
|
t.Fatal("React frontend still permits inline scripts")
|
|
}
|
|
marker := `src="/assets/`
|
|
start := strings.Index(string(body), marker)
|
|
if start < 0 {
|
|
t.Fatal("React asset reference is missing")
|
|
}
|
|
start += len(marker)
|
|
end := strings.Index(string(body)[start:], "\"")
|
|
if end < 0 {
|
|
t.Fatal("React asset reference is malformed")
|
|
}
|
|
assetPath := string(body)[start : start+end]
|
|
asset, err := http.Get(httpServer.URL + "/assets/" + assetPath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer asset.Body.Close()
|
|
if asset.StatusCode != http.StatusOK {
|
|
t.Fatalf("React asset status = %d", asset.StatusCode)
|
|
}
|
|
}
|
|
|
|
func TestNodeWebTaskAndEventFlow(t *testing.T) {
|
|
t.Parallel()
|
|
dataFile := filepath.Join(t.TempDir(), "control-plane.json")
|
|
server, err := NewServer(ServerConfig{
|
|
DataFile: dataFile,
|
|
NodeTokens: map[string]string{"node-1": "node-secret", "node-2": "other-secret"},
|
|
WebUsers: map[string]string{"admin": "web-secret"},
|
|
LeaseTTL: time.Minute,
|
|
HeartbeatTimeout: time.Minute,
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
httpServer := httptest.NewServer(server.Handler())
|
|
defer httpServer.Close()
|
|
defer server.Close()
|
|
client := httpServer.Client()
|
|
|
|
if response := doJSON(t, client, http.MethodGet, httpServer.URL+"/v1/nodes", "", nil); response.Code != http.StatusUnauthorized {
|
|
t.Fatalf("unauthenticated node list status = %d", response.Code)
|
|
}
|
|
login := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/auth/login", "", map[string]string{"username": "admin", "password": "web-secret"})
|
|
if login.Code != http.StatusOK {
|
|
t.Fatalf("login status = %d: %s", login.Code, login.Body.String())
|
|
}
|
|
var loginBody struct {
|
|
AccessToken string `json:"access_token"`
|
|
}
|
|
decodeBody(t, login, &loginBody)
|
|
if loginBody.AccessToken == "" {
|
|
t.Fatal("login returned no session token")
|
|
}
|
|
webAuth := "Bearer " + loginBody.AccessToken
|
|
|
|
registration := NodeRegistration{NodeID: "node-1", AgentVersion: "test", ProtocolVersion: ProtocolVersion,
|
|
Capabilities: []string{"heartbeat", "poll-tasks", "send-text"}, ReportingConfigVersion: 1,
|
|
Accounts: []AccountSummary{{AccountID: "account-a", Active: true, Verified: true}}}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/register", "Bearer node-secret", registration); response.Code != http.StatusOK {
|
|
t.Fatalf("registration status = %d: %s", response.Code, response.Body.String())
|
|
}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/heartbeat", "Bearer node-secret", Heartbeat{
|
|
NodeID: "node-1", AgentVersion: "test", ProtocolVersion: ProtocolVersion, NodeStatus: NodeOnline,
|
|
WechatRunning: true, WechatLoggedIn: true, ActiveAccountID: "account-a", ReportingConfigVersion: 1, CorrelationID: "heartbeat-1",
|
|
}); response.Code != http.StatusOK {
|
|
t.Fatalf("heartbeat status = %d", response.Code)
|
|
}
|
|
|
|
payload := json.RawMessage(`{"target_id":"chat-a","text":"hello","confirmed":true}`)
|
|
created := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/tasks", webAuth, TaskSubmission{
|
|
NodeID: "node-1", AccountID: "account-a", Kind: "send-text", IdempotencyKey: "same-key", Payload: payload,
|
|
})
|
|
if created.Code != http.StatusAccepted {
|
|
t.Fatalf("task create status = %d: %s", created.Code, created.Body.String())
|
|
}
|
|
var createdBody TaskSubmissionResponse
|
|
decodeBody(t, created, &createdBody)
|
|
if createdBody.TaskID == "" || createdBody.Status != TaskPending {
|
|
t.Fatalf("unexpected task create response: %+v", createdBody)
|
|
}
|
|
|
|
poll := doJSON(t, client, http.MethodGet, httpServer.URL+"/v1/nodes/node-1/tasks?account_id=account-a", "Bearer node-secret", nil)
|
|
if poll.Code != http.StatusOK {
|
|
t.Fatalf("poll status = %d", poll.Code)
|
|
}
|
|
var batch TaskBatch
|
|
decodeBody(t, poll, &batch)
|
|
if len(batch.Tasks) != 1 || batch.Tasks[0].LeaseGeneration != 1 {
|
|
t.Fatalf("unexpected task batch: %+v", batch)
|
|
}
|
|
task := batch.Tasks[0]
|
|
ack := TaskAck{TaskID: task.TaskID, AccountID: task.AccountID, LeaseGeneration: task.LeaseGeneration}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/tasks/"+task.TaskID+"/ack", "Bearer node-secret", ack); response.Code != http.StatusOK {
|
|
t.Fatalf("ack status = %d: %s", response.Code, response.Body.String())
|
|
}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/tasks/"+task.TaskID+"/start", "Bearer node-secret", ack); response.Code != http.StatusOK {
|
|
t.Fatalf("start status = %d: %s", response.Code, response.Body.String())
|
|
}
|
|
result := TaskResult{TaskID: task.TaskID, AccountID: task.AccountID, LeaseGeneration: task.LeaseGeneration, Status: TaskSucceeded, HasSideEffect: true, CorrelationID: "result-1"}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/tasks/"+task.TaskID+"/result", "Bearer node-secret", result); response.Code != http.StatusOK {
|
|
t.Fatalf("result status = %d: %s", response.Code, response.Body.String())
|
|
}
|
|
|
|
duplicate := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/tasks", webAuth, TaskSubmission{
|
|
NodeID: "node-1", AccountID: "account-a", Kind: "send-text", IdempotencyKey: "same-key", Payload: payload,
|
|
})
|
|
var duplicateBody TaskSubmissionResponse
|
|
decodeBody(t, duplicate, &duplicateBody)
|
|
if duplicate.Code != http.StatusAccepted || !duplicateBody.Duplicate || duplicateBody.TaskID != task.TaskID {
|
|
t.Fatalf("unexpected duplicate response: %+v", duplicateBody)
|
|
}
|
|
conflict := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/tasks", webAuth, TaskSubmission{
|
|
NodeID: "node-1", AccountID: "account-a", Kind: "send-text", IdempotencyKey: "same-key",
|
|
Payload: json.RawMessage(`{"target_id":"chat-b","text":"different","confirmed":true}`),
|
|
})
|
|
if conflict.Code != http.StatusConflict {
|
|
t.Fatalf("idempotency conflict status = %d", conflict.Code)
|
|
}
|
|
|
|
unauthorizedEvent := MessageEvent{NodeID: "node-1", AccountID: "account-a", ChatID: "chat-not-allowed", ChatType: ChatPrivate,
|
|
EventSeq: 1, EventType: "message", OccurredAt: time.Now().UTC(), Content: "must-not-store", ConfigVersion: 1, AuthorizationVersion: 1, Authorized: false}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/events", "Bearer node-secret", unauthorizedEvent); response.Code != http.StatusForbidden {
|
|
t.Fatalf("unauthorized event status = %d", response.Code)
|
|
}
|
|
authorizedEvent := unauthorizedEvent
|
|
authorizedEvent.ChatID = "chat-a"
|
|
authorizedEvent.Authorized = true
|
|
accepted := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/events", "Bearer node-secret", authorizedEvent)
|
|
if accepted.Code != http.StatusAccepted {
|
|
t.Fatalf("authorized event status = %d: %s", accepted.Code, accepted.Body.String())
|
|
}
|
|
duplicateEvent := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/events", "Bearer node-secret", authorizedEvent)
|
|
var duplicateEventBody EventReceipt
|
|
decodeBody(t, duplicateEvent, &duplicateEventBody)
|
|
if duplicateEvent.Code != http.StatusAccepted || !duplicateEventBody.Duplicate {
|
|
t.Fatalf("unexpected duplicate event: %+v", duplicateEventBody)
|
|
}
|
|
crossAccount := authorizedEvent
|
|
crossAccount.AccountID = "account-b"
|
|
crossAccount.ChatID = "chat-a"
|
|
crossAccount.EventSeq = 1
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/events", "Bearer node-secret", crossAccount); response.Code != http.StatusAccepted {
|
|
t.Fatalf("cross-account event status = %d", response.Code)
|
|
}
|
|
changed := authorizedEvent
|
|
changed.Content = "different"
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/events", "Bearer node-secret", changed); response.Code != http.StatusConflict {
|
|
t.Fatalf("event idempotency conflict status = %d", response.Code)
|
|
}
|
|
|
|
events := doJSON(t, client, http.MethodGet, httpServer.URL+"/v1/events?limit=20", webAuth, nil)
|
|
if events.Code != http.StatusOK || bytes.Contains(events.Body.Bytes(), []byte("chat-not-allowed")) {
|
|
t.Fatalf("event listing leaked rejected event: status=%d body=%s", events.Code, events.Body.String())
|
|
}
|
|
var eventList struct {
|
|
Events []StoredEvent `json:"events"`
|
|
}
|
|
decodeBody(t, events, &eventList)
|
|
if len(eventList.Events) != 2 {
|
|
t.Fatalf("expected two account-scoped events, got %d", len(eventList.Events))
|
|
}
|
|
|
|
if err := server.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
restarted, err := NewServer(ServerConfig{DataFile: dataFile, NodeTokens: map[string]string{"node-1": "node-secret"}, WebUsers: map[string]string{"admin": "web-secret"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
restartedHTTP := httptest.NewServer(restarted.Handler())
|
|
defer restartedHTTP.Close()
|
|
loginAgain := doJSON(t, restartedHTTP.Client(), http.MethodPost, restartedHTTP.URL+"/v1/auth/login", "", map[string]string{"username": "admin", "password": "web-secret"})
|
|
var loginAgainBody struct {
|
|
AccessToken string `json:"access_token"`
|
|
}
|
|
decodeBody(t, loginAgain, &loginAgainBody)
|
|
stored := doJSON(t, restartedHTTP.Client(), http.MethodGet, restartedHTTP.URL+"/v1/tasks/"+task.TaskID, "Bearer "+loginAgainBody.AccessToken, nil)
|
|
var storedTask Task
|
|
decodeBody(t, stored, &storedTask)
|
|
if stored.Code != http.StatusOK || storedTask.Status != TaskSucceeded {
|
|
t.Fatalf("task was not durable across restart: status=%d task=%+v", stored.Code, storedTask)
|
|
}
|
|
}
|
|
|
|
func TestRemoteReadRoutesUseTheTaskQueueAndRetainReadContent(t *testing.T) {
|
|
server, err := NewServer(ServerConfig{
|
|
DataFile: filepath.Join(t.TempDir(), "control-plane.json"),
|
|
NodeTokens: map[string]string{"node-1": "node-secret"},
|
|
WebUsers: map[string]string{"admin": "web-secret"},
|
|
LeaseTTL: time.Minute,
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
httpServer := httptest.NewServer(server.Handler())
|
|
defer httpServer.Close()
|
|
defer server.Close()
|
|
client := httpServer.Client()
|
|
|
|
login := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/auth/login", "", map[string]string{"username": "admin", "password": "web-secret"})
|
|
var loginBody struct {
|
|
AccessToken string `json:"access_token"`
|
|
}
|
|
decodeBody(t, login, &loginBody)
|
|
webAuth := "Bearer " + loginBody.AccessToken
|
|
registration := NodeRegistration{NodeID: "node-1", AgentVersion: "test", ProtocolVersion: ProtocolVersion,
|
|
Capabilities: []string{"heartbeat", "poll-tasks", "read-sessions", "read-contacts", "read-messages"},
|
|
Accounts: []AccountSummary{{AccountID: "account-a", Active: true, Verified: true}}}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/register", "Bearer node-secret", registration); response.Code != http.StatusOK {
|
|
t.Fatalf("registration status = %d: %s", response.Code, response.Body.String())
|
|
}
|
|
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/reads/sessions", "", ReadTaskSubmission{
|
|
NodeID: "node-1", AccountID: "account-a", IdempotencyKey: "read-unauthorized",
|
|
}); response.Code != http.StatusUnauthorized {
|
|
t.Fatalf("unauthorized read status = %d", response.Code)
|
|
}
|
|
created := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/reads/sessions", webAuth, ReadTaskSubmission{
|
|
NodeID: "node-1", AccountID: "account-a", IdempotencyKey: "read-sessions-1", Limit: 20,
|
|
})
|
|
if created.Code != http.StatusAccepted {
|
|
t.Fatalf("read create status = %d: %s", created.Code, created.Body.String())
|
|
}
|
|
var createdBody TaskSubmissionResponse
|
|
decodeBody(t, created, &createdBody)
|
|
stored := doJSON(t, client, http.MethodGet, httpServer.URL+"/v1/tasks/"+createdBody.TaskID, webAuth, nil)
|
|
var createdTask Task
|
|
decodeBody(t, stored, &createdTask)
|
|
if createdTask.Kind != "read-sessions" || createdTask.Payload == nil {
|
|
t.Fatalf("unexpected read task: %+v", createdTask)
|
|
}
|
|
|
|
poll := doJSON(t, client, http.MethodGet, httpServer.URL+"/v1/nodes/node-1/tasks?account_id=account-a", "Bearer node-secret", nil)
|
|
var batch TaskBatch
|
|
decodeBody(t, poll, &batch)
|
|
if len(batch.Tasks) != 1 {
|
|
t.Fatalf("expected one read task, got %d", len(batch.Tasks))
|
|
}
|
|
task := batch.Tasks[0]
|
|
ack := TaskAck{TaskID: task.TaskID, AccountID: task.AccountID, LeaseGeneration: task.LeaseGeneration}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/tasks/"+task.TaskID+"/ack", "Bearer node-secret", ack); response.Code != http.StatusOK {
|
|
t.Fatalf("ack status = %d: %s", response.Code, response.Body.String())
|
|
}
|
|
if response := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/tasks/"+task.TaskID+"/start", "Bearer node-secret", ack); response.Code != http.StatusOK {
|
|
t.Fatalf("start status = %d: %s", response.Code, response.Body.String())
|
|
}
|
|
content := json.RawMessage(`{"items":[],"limit":20,"offset":0,"hasMore":false}`)
|
|
result := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/nodes/node-1/tasks/"+task.TaskID+"/result", "Bearer node-secret", TaskResult{
|
|
TaskID: task.TaskID, AccountID: task.AccountID, LeaseGeneration: task.LeaseGeneration,
|
|
Status: TaskSucceeded, Content: content, CorrelationID: "read-result-1",
|
|
})
|
|
if result.Code != http.StatusOK {
|
|
t.Fatalf("read result status = %d: %s", result.Code, result.Body.String())
|
|
}
|
|
var resultTask Task
|
|
decodeBody(t, result, &resultTask)
|
|
if resultTask.Result == nil || string(resultTask.Result.Content) != string(content) {
|
|
t.Fatalf("read content was not retained: %+v", resultTask.Result)
|
|
}
|
|
|
|
invalid := doJSON(t, client, http.MethodPost, httpServer.URL+"/v1/reads/messages", webAuth, ReadTaskSubmission{
|
|
NodeID: "node-1", AccountID: "account-a", IdempotencyKey: "read-messages-invalid",
|
|
})
|
|
if invalid.Code != http.StatusBadRequest {
|
|
t.Fatalf("invalid message read status = %d: %s", invalid.Code, invalid.Body.String())
|
|
}
|
|
}
|
|
|
|
type responseBody struct {
|
|
*http.Response
|
|
Body *bytes.Buffer
|
|
Code int
|
|
}
|
|
|
|
func doJSON(t *testing.T, client *http.Client, method, endpoint, authorization string, value any) responseBody {
|
|
t.Helper()
|
|
var body *bytes.Reader
|
|
if value == nil {
|
|
body = bytes.NewReader(nil)
|
|
} else {
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
body = bytes.NewReader(data)
|
|
}
|
|
request, err := http.NewRequest(method, endpoint, body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
request.Header.Set("Content-Type", "application/json")
|
|
if authorization != "" {
|
|
request.Header.Set("Authorization", authorization)
|
|
}
|
|
response, err := client.Do(request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer response.Body.Close()
|
|
data, err := io.ReadAll(response.Body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return responseBody{Response: response, Body: bytes.NewBuffer(data), Code: response.StatusCode}
|
|
}
|
|
|
|
func decodeBody(t *testing.T, response responseBody, target any) {
|
|
t.Helper()
|
|
if err := json.Unmarshal(response.Body.Bytes(), target); err != nil {
|
|
t.Fatalf("decode response %d: %v; body=%s", response.Code, err, response.Body.String())
|
|
}
|
|
}
|