feat: add control-plane identity and management console

control-plane/server.go

@@ -86,9 +86,9 @@ type Server struct {
 	config          ServerConfig
 	store           *Store
 	accountStores   *AccountStoreManager
+	userStore       *UserStore
 	tlsConfig       *tls.Config
 	nodeTokenHashes map[string][32]byte
-	userPasswords   map[string][32]byte
 	sessionMu       sync.Mutex
 	sessions        map[string]session
 	aiProvider      AIProvider
@@ -101,8 +101,8 @@ type Server struct {
 }
 
 type session struct {
-	Username string
-	Expires  time.Time
+	UserID  string
+	Expires time.Time
 }
 
 type requestError struct {
@@ -204,14 +204,26 @@ func NewServer(config ServerConfig) (*Server, error) {
 		_ = store.Close()
 		return nil, err
 	}
+	userStore, err := NewUserStore(accountStores.catalog)
+	if err != nil {
+		_ = accountStores.Close()
+		_ = store.Close()
+		return nil, err
+	}
+	if err := userStore.BootstrapAdmins(context.Background(), config.WebUsers); err != nil {
+		_ = accountStores.Close()
+		_ = store.Close()
+		return nil, err
+	}
+	config.WebUsers = nil
 	aiCtx, aiCancel := context.WithCancel(context.Background())
 	s := &Server{
 		config:          config,
 		store:           store,
 		accountStores:   accountStores,
+		userStore:       userStore,
 		tlsConfig:       tlsConfig,
 		nodeTokenHashes: map[string][32]byte{},
-		userPasswords:   map[string][32]byte{},
 		sessions:        map[string]session{},
 		aiProvider:      config.AIProvider,
 		aiCtx:           aiCtx,
@@ -223,11 +235,6 @@ func NewServer(config ServerConfig) (*Server, error) {
 			s.nodeTokenHashes[nodeID] = sha256.Sum256([]byte(token))
 		}
 	}
-	for username, password := range config.WebUsers {
-		if username != "" && password != "" {
-			s.userPasswords[username] = sha256.Sum256([]byte(password))
-		}
-	}
 	s.aiWG.Add(3)
 	go s.aiWorker()
 	go s.aiScheduler()
@@ -310,6 +317,9 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	case r.URL.Path == "/" && r.Method == http.MethodGet:
 		s.serveFrontend(w, "dist/index.html")
 		return
+	case r.Method == http.MethodGet && isFrontendRoute(r.URL.Path):
+		s.serveFrontend(w, "dist/index.html")
+		return
 	case strings.HasPrefix(r.URL.Path, "/assets/") && r.Method == http.MethodGet:
 		s.serveFrontend(w, path.Join("dist", strings.TrimPrefix(r.URL.Path, "/")))
 		return
@@ -320,6 +330,10 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		err = s.ready(w, correlationID)
 	case r.URL.Path == "/v1/auth/login" && r.Method == http.MethodPost:
 		err = s.login(w, r, correlationID)
+	case r.URL.Path == "/v1/auth/me" || r.URL.Path == "/v1/auth/logout":
+		err = s.webAuthRoute(w, r, correlationID)
+	case r.URL.Path == "/v1/users" || strings.HasPrefix(r.URL.Path, "/v1/users/"):
+		err = s.usersRoute(w, r, correlationID)
 	case r.URL.Path == "/v1/nodes" && r.Method == http.MethodGet:
 		err = s.listNodes(w, r)
 	case r.URL.Path == "/v1/nodes/register" && r.Method == http.MethodPost:
@@ -348,6 +362,15 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func isFrontendRoute(route string) bool {
+	switch route {
+	case "/overview", "/messages", "/contacts", "/tasks", "/agents", "/users", "/broadcast", "/diagnostics":
+		return true
+	default:
+		return false
+	}
+}
+
 func (s *Server) ready(w http.ResponseWriter, correlationID string) error {
 	if err := s.store.Read(func(PersistedState) error { return nil }); err != nil {
 		return requestError{status: http.StatusServiceUnavailable, code: "NotReady", message: "The control plane store is not ready."}
@@ -364,17 +387,19 @@ func (s *Server) login(w http.ResponseWriter, r *http.Request, correlationID str
 	if err := decodeJSON(r, &request, 8*1024); err != nil {
 		return err
 	}
-	password, exists := s.userPasswords[request.Username]
-	provided := sha256.Sum256([]byte(request.Password))
-	if !exists || subtle.ConstantTimeCompare(password[:], provided[:]) != 1 {
+	user, ok, err := s.userStore.Authenticate(r.Context(), request.Username, request.Password)
+	if err != nil {
+		return requestError{status: http.StatusInternalServerError, code: "AuthenticationUnavailable", message: "Authentication is temporarily unavailable."}
+	}
+	if !ok {
 		return requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Authentication failed."}
 	}
 	token := randomID()
 	s.sessionMu.Lock()
-	s.sessions[token] = session{Username: request.Username, Expires: time.Now().UTC().Add(s.config.SessionTTL)}
+	s.sessions[webSessionKey(token)] = session{UserID: user.ID, Expires: time.Now().UTC().Add(s.config.SessionTTL)}
 	s.sessionMu.Unlock()
-	_ = s.appendAudit(request.Username, "login", "session", correlationID, "success")
-	writeJSON(w, http.StatusOK, map[string]any{"access_token": token, "token_type": "Bearer", "expires_in": int(s.config.SessionTTL.Seconds()), "correlation_id": correlationID})
+	_ = s.appendAudit(user.Username, "login", "session", correlationID, "success")
+	writeJSON(w, http.StatusOK, map[string]any{"access_token": token, "token_type": "Bearer", "expires_in": int(s.config.SessionTTL.Seconds()), "user": user, "correlation_id": correlationID})
 	return nil
 }
 
@@ -464,13 +489,17 @@ func (s *Server) nodeRoute(w http.ResponseWriter, r *http.Request, correlationID
 }
 
 func (s *Server) listNodes(w http.ResponseWriter, r *http.Request) error {
-	if _, err := s.authenticateWeb(r); err != nil {
+	principal, err := s.authenticateWebPrincipal(r)
+	if err != nil {
 		return err
 	}
 	var nodes []Node
 	if err := s.store.Mutate(func(state *PersistedState) error {
 		now := time.Now().UTC()
 		for nodeID, value := range state.Nodes {
+			if !principal.CanAccessNode(nodeID) {
+				continue
+			}
 			node := value
 			nodeChanged := false
 			if node.LastHeartbeatAt == nil || now.Sub(*node.LastHeartbeatAt) > s.config.HeartbeatTimeout {
@@ -886,7 +915,7 @@ func (s *Server) recordTaskResult(w http.ResponseWriter, r *http.Request, nodeID
 }
 
 func (s *Server) readRoute(w http.ResponseWriter, r *http.Request, correlationID string) error {
-	username, err := s.authenticateWeb(r)
+	principal, err := s.authenticateWebPrincipal(r)
 	if err != nil {
 		return err
 	}
@@ -906,10 +935,13 @@ func (s *Server) readRoute(w http.ResponseWriter, r *http.Request, correlationID
 	if err != nil {
 		return err
 	}
+	if !principal.CanAccessNode(request.NodeID) {
+		return inaccessibleResource()
+	}
 	return s.createTaskSubmission(w, TaskSubmission{
 		NodeID: request.NodeID, AccountID: request.AccountID, Kind: kind,
 		IdempotencyKey: request.IdempotencyKey, Payload: payload, NotAfter: request.NotAfter,
-	}, username, correlationID)
+	}, principal.Username, correlationID)
 }
 
 type readSessionsPayload struct {
@@ -1016,7 +1048,7 @@ func validPagination(limit, offset int) bool { return limit >= 1 && limit <= 200
 func hasTaskContent(content jsonRaw) bool { return len(content) != 0 && string(content) != "null" }
 
 func (s *Server) taskRoute(w http.ResponseWriter, r *http.Request, correlationID string) error {
-	username, err := s.authenticateWeb(r)
+	principal, err := s.authenticateWebPrincipal(r)
 	if err != nil {
 		return err
 	}
@@ -1024,31 +1056,34 @@ func (s *Server) taskRoute(w http.ResponseWriter, r *http.Request, correlationID
 	if len(parts) == 2 && parts[0] == "v1" && parts[1] == "tasks" {
 		switch r.Method {
 		case http.MethodGet:
-			return s.listTasks(w, r)
+			return s.listTasks(w, r, principal)
 		case http.MethodPost:
-			return s.createTask(w, r, username, correlationID)
+			return s.createTask(w, r, principal, correlationID)
 		default:
 			return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
 		}
 	}
 	if len(parts) == 3 && r.Method == http.MethodGet {
-		return s.getTask(w, parts[2])
+		return s.getTask(w, parts[2], principal)
 	}
 	if len(parts) == 4 && parts[3] == "cancel" && r.Method == http.MethodPost {
-		return s.cancelTask(w, parts[2], username, correlationID)
+		return s.cancelTask(w, parts[2], principal, correlationID)
 	}
 	if len(parts) == 4 && parts[3] == "resume" && r.Method == http.MethodPost {
-		return s.resumeTask(w, parts[2], username, correlationID)
+		return s.resumeTask(w, parts[2], principal, correlationID)
 	}
 	return requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}
 }
 
-func (s *Server) createTask(w http.ResponseWriter, r *http.Request, username, correlationID string) error {
+func (s *Server) createTask(w http.ResponseWriter, r *http.Request, principal WebPrincipal, correlationID string) error {
 	var request TaskSubmission
 	if err := decodeJSON(r, &request, 128*1024); err != nil {
 		return err
 	}
-	return s.createTaskSubmission(w, request, username, correlationID)
+	if !principal.CanAccessNode(request.NodeID) {
+		return inaccessibleResource()
+	}
+	return s.createTaskSubmission(w, request, principal.Username, correlationID)
 }
 
 func (s *Server) createTaskSubmission(w http.ResponseWriter, request TaskSubmission, username, correlationID string) error {
@@ -1103,14 +1138,14 @@ func (s *Server) createTaskSubmission(w http.ResponseWriter, request TaskSubmiss
 	return nil
 }
 
-func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) error {
+func (s *Server) listTasks(w http.ResponseWriter, r *http.Request, principal WebPrincipal) error {
 	nodeID := r.URL.Query().Get("node_id")
 	accountID := r.URL.Query().Get("account_id")
 	limit := queryLimit(r.URL.Query().Get("limit"))
 	var tasks []Task
 	if err := s.store.Read(func(state PersistedState) error {
 		for _, task := range state.Tasks {
-			if nodeID != "" && task.NodeID != nodeID || accountID != "" && task.AccountID != accountID {
+			if !principal.CanAccessNode(task.NodeID) || nodeID != "" && task.NodeID != nodeID || accountID != "" && task.AccountID != accountID {
 				continue
 			}
 			tasks = append(tasks, task)
@@ -1127,14 +1162,14 @@ func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) error {
 	return nil
 }
 
-func (s *Server) getTask(w http.ResponseWriter, taskID string) error {
+func (s *Server) getTask(w http.ResponseWriter, taskID string, principal WebPrincipal) error {
 	if !validIdentifier(taskID, 200) {
 		return requestError{status: http.StatusBadRequest, code: "InvalidTask", message: "Task ID is invalid."}
 	}
 	var task Task
 	if err := s.store.Read(func(state PersistedState) error {
 		value, ok := state.Tasks[taskID]
-		if !ok {
+		if !ok || !principal.CanAccessNode(value.NodeID) {
 			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
 		}
 		task = value
@@ -1146,13 +1181,16 @@ func (s *Server) getTask(w http.ResponseWriter, taskID string) error {
 	return nil
 }
 
-func (s *Server) cancelTask(w http.ResponseWriter, taskID, username, correlationID string) error {
+func (s *Server) cancelTask(w http.ResponseWriter, taskID string, principal WebPrincipal, correlationID string) error {
 	var task Task
 	err := s.store.Mutate(func(state *PersistedState) error {
 		value, ok := state.Tasks[taskID]
 		if !ok {
 			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
 		}
+		if !principal.CanAccessNode(value.NodeID) {
+			return inaccessibleResource()
+		}
 		task = value
 		if terminal(task.Status) {
 			return nil
@@ -1167,7 +1205,7 @@ func (s *Server) cancelTask(w http.ResponseWriter, taskID, username, correlation
 				task.LeaseExpiresAt = nil
 			}
 			task.UpdatedAt = now
-			state.Audit = appendAudit(state.Audit, "user:"+username, "task.cancel-request", taskID, correlationID, "success", now)
+			state.Audit = appendAudit(state.Audit, "user:"+principal.Username, "task.cancel-request", taskID, correlationID, "success", now)
 			state.Tasks[taskID] = task
 		}
 		return nil
@@ -1179,13 +1217,16 @@ func (s *Server) cancelTask(w http.ResponseWriter, taskID, username, correlation
 	return nil
 }
 
-func (s *Server) resumeTask(w http.ResponseWriter, taskID, username, correlationID string) error {
+func (s *Server) resumeTask(w http.ResponseWriter, taskID string, principal WebPrincipal, correlationID string) error {
 	var task Task
 	err := s.store.Mutate(func(state *PersistedState) error {
 		value, ok := state.Tasks[taskID]
 		if !ok {
 			return requestError{status: http.StatusNotFound, code: "TaskNotFound", message: "Task was not found."}
 		}
+		if !principal.CanAccessNode(value.NodeID) {
+			return inaccessibleResource()
+		}
 		task = value
 		if terminal(task.Status) {
 			return requestError{status: http.StatusConflict, code: "TerminalState", message: "The task cannot be resumed after it reached a terminal state."}
@@ -1212,7 +1253,7 @@ func (s *Server) resumeTask(w http.ResponseWriter, taskID, username, correlation
 			task.LastCorrelationID = correlationID
 		}
 		state.Tasks[taskID] = task
-		state.Audit = appendAudit(state.Audit, "user:"+username, "task.resume", taskID, correlationID, "success", now)
+		state.Audit = appendAudit(state.Audit, "user:"+principal.Username, "task.resume", taskID, correlationID, "success", now)
 		return nil
 	})
 	if err != nil {
@@ -1226,10 +1267,11 @@ func (s *Server) eventRoute(w http.ResponseWriter, r *http.Request, _ string) er
 	if r.Method != http.MethodGet {
 		return requestError{status: http.StatusMethodNotAllowed, code: "MethodNotAllowed", message: "Method is not allowed."}
 	}
-	if _, err := s.authenticateWeb(r); err != nil {
+	principal, err := s.authenticateWebPrincipal(r)
+	if err != nil {
 		return err
 	}
-	return s.listEvents(w, r)
+	return s.listEvents(w, r, principal)
 }
 
 func (s *Server) ingestEvent(w http.ResponseWriter, r *http.Request, nodeID, correlationID string) error {
@@ -1286,7 +1328,7 @@ func (s *Server) ingestEvent(w http.ResponseWriter, r *http.Request, nodeID, cor
 	return nil
 }
 
-func (s *Server) listEvents(w http.ResponseWriter, r *http.Request) error {
+func (s *Server) listEvents(w http.ResponseWriter, r *http.Request, principal WebPrincipal) error {
 	query := r.URL.Query()
 	nodeID, accountID, chatID := query.Get("node_id"), query.Get("account_id"), query.Get("chat_id")
 	limit := queryLimit(query.Get("limit"))
@@ -1294,7 +1336,7 @@ func (s *Server) listEvents(w http.ResponseWriter, r *http.Request) error {
 	if err := s.store.Read(func(state PersistedState) error {
 		for index := len(state.Events) - 1; index >= 0 && len(events) < limit; index-- {
 			event := state.Events[index]
-			if nodeID != "" && event.NodeID != nodeID || accountID != "" && event.AccountID != accountID || chatID != "" && event.ChatID != chatID {
+			if !principal.CanAccessNode(event.NodeID) || nodeID != "" && event.NodeID != nodeID || accountID != "" && event.AccountID != accountID || chatID != "" && event.ChatID != chatID {
 				continue
 			}
 			events = append(events, event)
@@ -1308,9 +1350,13 @@ func (s *Server) listEvents(w http.ResponseWriter, r *http.Request) error {
 }
 
 func (s *Server) auditRoute(w http.ResponseWriter, r *http.Request, _ string) error {
-	if _, err := s.authenticateWeb(r); err != nil {
+	principal, err := s.authenticateWebPrincipal(r)
+	if err != nil {
 		return err
 	}
+	if principal.Role != RoleAdmin {
+		return requestError{status: http.StatusForbidden, code: "AdminRequired", message: "Administrator access is required."}
+	}
 	limit := queryLimit(r.URL.Query().Get("limit"))
 	var audit []AuditEntry
 	if err := s.store.Read(func(state PersistedState) error {
@@ -1343,22 +1389,40 @@ func (s *Server) authenticateNode(r *http.Request) (string, error) {
 }
 
 func (s *Server) authenticateWeb(r *http.Request) (string, error) {
+	principal, err := s.authenticateWebPrincipal(r)
+	return principal.Username, err
+}
+
+func (s *Server) authenticateWebPrincipal(r *http.Request) (WebPrincipal, error) {
 	token := bearerToken(r)
 	if token == "" {
-		return "", requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Web authentication is required."}
+		return WebPrincipal{}, requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Web authentication is required."}
 	}
+	key := webSessionKey(token)
 	now := time.Now().UTC()
 	s.sessionMu.Lock()
-	value, ok := s.sessions[token]
+	value, ok := s.sessions[key]
 	if ok && !now.Before(value.Expires) {
-		delete(s.sessions, token)
+		delete(s.sessions, key)
 		ok = false
 	}
 	s.sessionMu.Unlock()
 	if !ok {
-		return "", requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Web session is missing or expired."}
+		return WebPrincipal{}, requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Web session is missing or expired."}
 	}
-	return value.Username, nil
+	user, err := s.userStore.Get(r.Context(), value.UserID)
+	if errors.Is(err, ErrUserNotFound) || err == nil && !user.Active {
+		return WebPrincipal{}, requestError{status: http.StatusUnauthorized, code: "Unauthorized", message: "Web session is missing or expired."}
+	}
+	if err != nil {
+		return WebPrincipal{}, requestError{status: http.StatusInternalServerError, code: "AuthenticationUnavailable", message: "Authentication is temporarily unavailable."}
+	}
+	return WebPrincipal{WebUser: user}, nil
+}
+
+func webSessionKey(token string) string {
+	digest := sha256.Sum256([]byte(token))
+	return string(digest[:])
 }
 
 func (s *Server) appendAudit(principal, action, resource, correlationID, outcome string) error {
@@ -1378,12 +1442,32 @@ func (s *Server) serveFrontend(w http.ResponseWriter, name string) {
 		writeError(w, requestError{status: http.StatusNotFound, code: "NotFound", message: "Resource was not found."}, randomID())
 		return
 	}
+	nonce := ""
+	if path.Base(name) == "index.html" {
+		random := make([]byte, 16)
+		if _, err := rand.Read(random); err != nil {
+			writeError(w, requestError{status: http.StatusInternalServerError, code: "InternalError", message: "The web application could not be initialized."}, randomID())
+			return
+		}
+		nonce = base64.RawStdEncoding.EncodeToString(random)
+		html := string(content)
+		if !strings.Contains(html, "__WX_CSP_NONCE__") {
+			writeError(w, requestError{status: http.StatusInternalServerError, code: "InternalError", message: "The web application could not be initialized."}, randomID())
+			return
+		}
+		content = []byte(strings.Replace(html, "__WX_CSP_NONCE__", nonce, 1))
+		w.Header().Set("Cache-Control", "no-store")
+	}
 	contentType := mime.TypeByExtension(path.Ext(name))
 	if contentType == "" {
 		contentType = "application/octet-stream"
 	}
 	w.Header().Set("Content-Type", contentType+"; charset=utf-8")
-	w.Header().Set("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src 'self'; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
+	styleSource := "'self'"
+	if nonce != "" {
+		styleSource += " 'nonce-" + nonce + "'"
+	}
+	w.Header().Set("Content-Security-Policy", "default-src 'self'; script-src 'self'; style-src "+styleSource+"; style-src-attr 'unsafe-inline'; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
 	w.WriteHeader(http.StatusOK)
 	_, _ = w.Write(content)
 }