You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by ya...@apache.org on 2026/09/30 22:16:20 UTC
(airavata-custos) 01/02: Require admin approval before a cluster account is created (#578)
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit d4879e40df8dee417b3d6689d1d3ff0f0de8f4c2
Author: Lahiru Jayathilake <la...@gmail.com>
AuthorDate: Wed Sep 30 18:16:11 2026 -0400
Require admin approval before a cluster account is created (#578)
---
api/core.openapi.yaml | 181 +++++++++++++++++++++
.../internal/subscribers/cluster_user.go | 8 +-
docs/contributing/writing-a-connector.md | 4 +-
.../db/migrations/000002_compute_clusters.up.sql | 11 +-
.../server/cluster_account_integration_test.go | 116 +++++++++++++
internal/server/compute_cluster.go | 98 +++++++++++
internal/server/responses.go | 14 ++
internal/server/server.go | 3 +
internal/store/compute_cluster_user_store.go | 52 +++++-
internal/store/store.go | 4 +
pkg/events/compute_cluster_user_subscribe.go | 7 +
pkg/events/types.go | 7 +-
pkg/models/allocation.go | 26 ++-
pkg/service/compute_cluster_user.go | 141 +++++++++++++++-
.../compute_cluster_user_integration_test.go | 176 ++++++++++++++++++++
pkg/service/user_onboard.go | 11 +-
16 files changed, 837 insertions(+), 22 deletions(-)
diff --git a/api/core.openapi.yaml b/api/core.openapi.yaml
index 8f0e49ba0..e9393567b 100644
--- a/api/core.openapi.yaml
+++ b/api/core.openapi.yaml
@@ -72,6 +72,59 @@ definitions:
x-enum-varnames:
- ClusterAccessUser
- ClusterAccessAdmin
+ ClusterAccountApproval:
+ enum:
+ - PENDING
+ - APPROVED
+ - DENIED
+ type: string
+ x-enum-varnames:
+ - ClusterAccountPending
+ - ClusterAccountApproved
+ - ClusterAccountDenied
+ ClusterAccountListResponse:
+ properties:
+ items:
+ items:
+ $ref: '#/definitions/ClusterAccountResponse'
+ type: array
+ total:
+ type: integer
+ type: object
+ ClusterAccountResponse:
+ properties:
+ access_level:
+ $ref: '#/definitions/ClusterAccessLevel'
+ approval_status:
+ $ref: '#/definitions/ClusterAccountApproval'
+ cluster_name:
+ type: string
+ compute_cluster_id:
+ type: string
+ display_name:
+ type: string
+ email:
+ type: string
+ id:
+ type: string
+ local_username:
+ description: The username of the user on the compute cluster, which may be
+ different from their Airavata Custos username.
+ type: string
+ provisioned_at:
+ description: When the account was provisioned into the registry; nil until
+ then.
+ type: string
+ review_note:
+ type: string
+ reviewed_at:
+ type: string
+ reviewed_by:
+ description: The admin who approved or denied the account.
+ type: string
+ user_id:
+ type: string
+ type: object
ComputeAllocation:
properties:
compute_cluster_id:
@@ -333,6 +386,8 @@ definitions:
properties:
access_level:
$ref: '#/definitions/ClusterAccessLevel'
+ approval_status:
+ $ref: '#/definitions/ClusterAccountApproval'
compute_cluster_id:
type: string
id:
@@ -345,6 +400,13 @@ definitions:
description: When the account was provisioned into the registry; nil until
then.
type: string
+ review_note:
+ type: string
+ reviewed_at:
+ type: string
+ reviewed_by:
+ description: The admin who approved or denied the account.
+ type: string
user_id:
type: string
type: object
@@ -792,6 +854,11 @@ definitions:
username:
type: string
type: object
+ denyClusterAccountRequest:
+ properties:
+ note:
+ type: string
+ type: object
grantPrivilegeRequest:
properties:
privilege:
@@ -2592,6 +2659,41 @@ paths:
tags:
- Compute Allocation Usages
/compute-cluster-users:
+ get:
+ description: Cluster users with their user and cluster, newest first. Every
+ status unless one is given.
+ parameters:
+ - description: PENDING | APPROVED | DENIED
+ in: query
+ name: approval_status
+ type: string
+ - description: Page size
+ in: query
+ name: limit
+ type: integer
+ - description: Page offset
+ in: query
+ name: offset
+ type: integer
+ produces:
+ - application/json
+ responses:
+ "200":
+ description: OK
+ schema:
+ $ref: '#/definitions/ClusterAccountListResponse'
+ "400":
+ description: Bad Request
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ security:
+ - BearerAuth: []
+ summary: List cluster accounts by approval status
+ tags:
+ - Compute Cluster Users
post:
consumes:
- application/json
@@ -2711,6 +2813,85 @@ paths:
summary: Update a compute cluster user
tags:
- Compute Cluster Users
+ /compute-cluster-users/{id}/approve:
+ post:
+ description: Records the caller as the approver and starts creating the account
+ on the cluster.
+ parameters:
+ - description: Compute cluster user ID
+ in: path
+ name: id
+ required: true
+ type: string
+ produces:
+ - application/json
+ responses:
+ "200":
+ description: OK
+ schema:
+ $ref: '#/definitions/ComputeClusterUser'
+ "404":
+ description: Not Found
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ "409":
+ description: Already approved
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ security:
+ - BearerAuth: []
+ summary: Approve a cluster account
+ tags:
+ - Compute Cluster Users
+ /compute-cluster-users/{id}/deny:
+ post:
+ consumes:
+ - application/json
+ description: Records the caller's decision and the note. No account is created
+ on the cluster. The user keeps portal access.
+ parameters:
+ - description: Compute cluster user ID
+ in: path
+ name: id
+ required: true
+ type: string
+ - description: Optional note
+ in: body
+ name: request
+ schema:
+ $ref: '#/definitions/denyClusterAccountRequest'
+ produces:
+ - application/json
+ responses:
+ "200":
+ description: OK
+ schema:
+ $ref: '#/definitions/ComputeClusterUser'
+ "404":
+ description: Not Found
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ "409":
+ description: Already approved or denied
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ security:
+ - BearerAuth: []
+ summary: Deny a cluster account
+ tags:
+ - Compute Cluster Users
/compute-clusters:
get:
produces:
diff --git a/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go b/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go
index 71a554192..7e86b1cd9 100644
--- a/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go
+++ b/connectors/COmanage/Identity-Provisioner/internal/subscribers/cluster_user.go
@@ -55,7 +55,9 @@ func NewClusterUserSubscriber(c *client.Client, bus *events.Bus, core *service.S
// RegisterSubscribers subscribes the handlers under the given subscriber name.
func (s *ClusterUserSubscriber) RegisterSubscribers(subscriber string) {
- s.bus.SubscribeComputeClusterUserCreated(subscriber, s.handleClusterUserCreate)
+ // Accounts are created on the cluster only after an admin approves them,
+ // so the `approve event` is what starts provisioning, not the `create event`.
+ s.bus.SubscribeComputeClusterUserApproved(subscriber, s.handleClusterUserApproved)
s.bus.SubscribeUserIdentityCreated(subscriber, s.handleUserIdentityCreate)
}
@@ -94,9 +96,9 @@ func (s *ClusterUserSubscriber) handleUserIdentityCreate(ctx context.Context, id
return nil
}
-func (s *ClusterUserSubscriber) handleClusterUserCreate(ctx context.Context, cu models.ComputeClusterUser) error {
+func (s *ClusterUserSubscriber) handleClusterUserApproved(ctx context.Context, cu models.ComputeClusterUser) error {
ctx = audit.WithSource(ctx, "comanage")
- ctx, span := tracing.Start(ctx, "comanage.cluster_user_create")
+ ctx, span := tracing.Start(ctx, "comanage.cluster_user_approved")
defer span.End()
if cu.ComputeClusterID != s.custosClusterID {
diff --git a/docs/contributing/writing-a-connector.md b/docs/contributing/writing-a-connector.md
index c2c3caa2a..f9ff790a3 100644
--- a/docs/contributing/writing-a-connector.md
+++ b/docs/contributing/writing-a-connector.md
@@ -373,10 +373,10 @@ core *service.Service
}
func (s *ClusterUserSubscriber) RegisterSubscribers() {
-s.bus.SubscribeComputeClusterUserCreated(comanage.Type, s.handleCreate)
+s.bus.SubscribeComputeClusterUserApproved(comanage.Type, s.handleApproved)
}
-func (s *ClusterUserSubscriber) handleCreate(ctx context.Context, cu models.ComputeClusterUser) error {
+func (s *ClusterUserSubscriber) handleApproved(ctx context.Context, cu models.ComputeClusterUser) error {
// ...react...
return nil
}
diff --git a/internal/db/migrations/000002_compute_clusters.up.sql b/internal/db/migrations/000002_compute_clusters.up.sql
index 8537a26e8..c303d035a 100644
--- a/internal/db/migrations/000002_compute_clusters.up.sql
+++ b/internal/db/migrations/000002_compute_clusters.up.sql
@@ -33,16 +33,25 @@ CREATE TABLE IF NOT EXISTS compute_cluster_users
local_username VARCHAR(32) NOT NULL,
access_level VARCHAR(16) NOT NULL DEFAULT 'USER',
provisioned_at TIMESTAMPTZ(6) NULL,
+ -- An admin with the right privilege has to approve the account before it is created on the cluster.
+ -- A user an admin adds is approved at once, a user from an allocation source waits for review.
+ approval_status VARCHAR(16) NOT NULL DEFAULT 'PENDING',
+ reviewed_at TIMESTAMPTZ(6) NULL,
+ reviewed_by VARCHAR(255) NULL,
+ review_note TEXT NULL,
created_at TIMESTAMPTZ(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
updated_at TIMESTAMPTZ(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
PRIMARY KEY (id),
CONSTRAINT chk_compute_cluster_users_access_level CHECK (access_level IN ('USER', 'ADMIN')),
+ CONSTRAINT chk_compute_cluster_users_approval_status CHECK (approval_status IN ('PENDING', 'APPROVED', 'DENIED')),
CONSTRAINT uq_compute_cluster_users_pair UNIQUE (compute_cluster_id, user_id),
CONSTRAINT uq_compute_cluster_users_local_username UNIQUE (compute_cluster_id, local_username),
CONSTRAINT fk_compute_cluster_users_cluster FOREIGN KEY (compute_cluster_id) REFERENCES compute_clusters (id) ON DELETE CASCADE,
- CONSTRAINT fk_compute_cluster_users_user FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE CASCADE
+ CONSTRAINT fk_compute_cluster_users_user FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE CASCADE,
+ CONSTRAINT fk_compute_cluster_users_reviewed_by FOREIGN KEY (reviewed_by) REFERENCES users (id) ON DELETE SET NULL
);
CREATE INDEX IF NOT EXISTS idx_compute_cluster_users_user ON compute_cluster_users (user_id);
+CREATE INDEX IF NOT EXISTS idx_compute_cluster_users_approval_status ON compute_cluster_users (approval_status);
CREATE OR REPLACE TRIGGER trg_compute_clusters_updated_at
BEFORE UPDATE ON compute_clusters
diff --git a/internal/server/cluster_account_integration_test.go b/internal/server/cluster_account_integration_test.go
new file mode 100644
index 000000000..6237a4c57
--- /dev/null
+++ b/internal/server/cluster_account_integration_test.go
@@ -0,0 +1,116 @@
+//go:build integration
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package server
+
+import (
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ "github.com/google/uuid"
+ "github.com/jmoiron/sqlx"
+
+ "github.com/apache/airavata-custos/pkg/models"
+ "github.com/apache/airavata-custos/pkg/service"
+)
+
+// seedClusterUser creates a cluster and a user, then a cluster user with no approval given, and returns it.
+func seedClusterUser(t *testing.T, database *sqlx.DB, svc *service.Service) *models.ComputeClusterUser {
+ t.Helper()
+ cluster, err := svc.CreateComputeCluster(t.Context(), &models.ComputeCluster{Name: "appr-" + uuid.NewString()[:8]})
+ if err != nil {
+ t.Fatalf("create cluster: %v", err)
+ }
+ userID := seedUser(t, database, fmt.Sprintf("member-%s@example.edu", uuid.NewString()))
+ cu, err := svc.CreateComputeClusterUser(t.Context(), &models.ComputeClusterUser{
+ ComputeClusterID: cluster.ID,
+ UserID: userID,
+ LocalUsername: "appr-" + uuid.NewString()[:8],
+ })
+ if err != nil {
+ t.Fatalf("create cluster user: %v", err)
+ }
+ return cu
+}
+
+// Make sure the list can be narrowed to the accounts waiting for approval and
+// carries the user's name, email and the cluster name the admin needs.
+func TestListClusterAccounts_FiltersByApprovalStatus(t *testing.T) {
+ database, svc, srv := setupTestStack(t)
+ admin := seedUser(t, database, fmt.Sprintf("admin-%s@example.edu", uuid.NewString()))
+ pending := seedClusterUser(t, database, svc)
+ approved := seedClusterUser(t, database, svc)
+ if _, err := svc.ApproveComputeClusterUser(t.Context(), approved.ID, admin); err != nil {
+ t.Fatalf("approve: %v", err)
+ }
+
+ rr := httptest.NewRecorder()
+ req := withTestCaller(httptest.NewRequest(http.MethodGet, "/compute-cluster-users?approval_status=PENDING", nil), admin, models.ClustersRead)
+ srv.ServeHTTP(rr, req)
+ if rr.Code != http.StatusOK {
+ t.Fatalf("status: got %d, want 200: %s", rr.Code, rr.Body.String())
+ }
+ var body ClusterAccountListResponse
+ if err := json.NewDecoder(rr.Body).Decode(&body); err != nil {
+ t.Fatalf("decode: %v", err)
+ }
+ if body.Total != 1 || len(body.Items) != 1 || body.Items[0].ID != pending.ID {
+ t.Fatalf("expected only the pending account, got %+v", body)
+ }
+ if body.Items[0].DisplayName != "Test User" || body.Items[0].Email == "" || body.Items[0].ClusterName == "" {
+ t.Fatalf("expected the user and cluster fields on the row, got %+v", body.Items[0])
+ }
+}
+
+// Make sure approving through the API records the caller as the reviewer.
+func TestApproveClusterAccount_RecordsCaller(t *testing.T) {
+ database, svc, srv := setupTestStack(t)
+ admin := seedUser(t, database, fmt.Sprintf("admin-%s@example.edu", uuid.NewString()))
+ cu := seedClusterUser(t, database, svc)
+
+ rr := httptest.NewRecorder()
+ req := withTestCaller(httptest.NewRequest(http.MethodPost, "/compute-cluster-users/"+cu.ID+"/approve", nil), admin, models.ClustersWrite)
+ srv.ServeHTTP(rr, req)
+ if rr.Code != http.StatusOK {
+ t.Fatalf("status: got %d, want 200: %s", rr.Code, rr.Body.String())
+ }
+ var got models.ComputeClusterUser
+ if err := json.NewDecoder(rr.Body).Decode(&got); err != nil {
+ t.Fatalf("decode: %v", err)
+ }
+ if got.ApprovalStatus != models.ClusterAccountApproved || got.ReviewedBy == nil || *got.ReviewedBy != admin {
+ t.Fatalf("expected approval by the caller, got %+v", got)
+ }
+}
+
+// Make sure the read privilege is not enough to approve or deny an account.
+func TestReviewClusterAccount_RequiresClustersWrite(t *testing.T) {
+ _, _, srv := setupTestStack(t)
+ for _, action := range []string{"approve", "deny"} {
+ rr := httptest.NewRecorder()
+ req := withTestCaller(httptest.NewRequest(http.MethodPost, "/compute-cluster-users/any/"+action, nil), "u-1", models.ClustersRead)
+ srv.ServeHTTP(rr, req)
+ if rr.Code != http.StatusForbidden {
+ t.Fatalf("%s status: got %d, want 403", action, rr.Code)
+ }
+ }
+}
diff --git a/internal/server/compute_cluster.go b/internal/server/compute_cluster.go
index f0b6f8f1d..05a8f81ca 100644
--- a/internal/server/compute_cluster.go
+++ b/internal/server/compute_cluster.go
@@ -19,6 +19,7 @@ package server
import (
"net/http"
+ "strings"
"github.com/apache/airavata-custos/pkg/common"
"github.com/apache/airavata-custos/pkg/models"
@@ -101,13 +102,20 @@ func (s *Server) createComputeClusterUser(w http.ResponseWriter, r *http.Request
common.WriteError(w, http.StatusBadRequest, err)
return
}
+ caller := requireCaller(w, r)
+ if caller == nil {
+ return
+ }
// Access level is left out on purpose. Granting cluster admin goes through
// user onboarding, which checks roles:manage first.
// TODO - accept an access level here once the caller can be checked for ADMIN on the same cluster
+ // An admin creating the account by hand is the approval.
created, err := s.svc.CreateComputeClusterUser(r.Context(), &models.ComputeClusterUser{
ComputeClusterID: req.ComputeClusterID,
UserID: req.UserID,
LocalUsername: req.LocalUsername,
+ ApprovalStatus: models.ClusterAccountApproved,
+ ReviewedBy: &caller.UserID,
})
if err != nil {
common.WriteServiceError(w, err)
@@ -180,6 +188,96 @@ func (s *Server) deleteComputeClusterUser(w http.ResponseWriter, r *http.Request
w.WriteHeader(http.StatusNoContent)
}
+// @Summary List cluster accounts by approval status
+// @Description Cluster users with their user and cluster, newest first. Every status unless one is given.
+// @Tags Compute Cluster Users
+// @Security BearerAuth
+// @Produce json
+// @Param approval_status query string false "PENDING | APPROVED | DENIED"
+// @Param limit query integer false "Page size"
+// @Param offset query integer false "Page offset"
+// @Success 200 {object} ClusterAccountListResponse
+// @Failure 400 {object} object{error=string}
+// @Router /compute-cluster-users [get]
+func (s *Server) listClusterAccounts(w http.ResponseWriter, r *http.Request) {
+ q := r.URL.Query()
+ status := models.ClusterAccountApproval(q.Get("approval_status"))
+ rows, total, err := s.svc.ListComputeClusterUsersByApproval(r.Context(), status, atoiOr(q.Get("limit"), 50), atoiOr(q.Get("offset"), 0))
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ items := make([]ClusterAccountResponse, 0, len(rows))
+ for _, row := range rows {
+ items = append(items, ClusterAccountResponse{
+ ComputeClusterUser: row.ComputeClusterUser,
+ DisplayName: strings.TrimSpace(row.FirstName + " " + row.LastName),
+ Email: row.Email,
+ ClusterName: row.ClusterName,
+ })
+ }
+ common.WriteJSON(w, http.StatusOK, ClusterAccountListResponse{Items: items, Total: total})
+}
+
+// @Summary Approve a cluster account
+// @Description Records the caller as the approver and starts creating the account on the cluster.
+// @Tags Compute Cluster Users
+// @Security BearerAuth
+// @Produce json
+// @Param id path string true "Compute cluster user ID"
+// @Success 200 {object} models.ComputeClusterUser
+// @Failure 404 {object} object{error=string}
+// @Failure 409 {object} object{error=string} "Already approved"
+// @Router /compute-cluster-users/{id}/approve [post]
+func (s *Server) approveClusterAccount(w http.ResponseWriter, r *http.Request) {
+ caller := requireCaller(w, r)
+ if caller == nil {
+ return
+ }
+ cu, err := s.svc.ApproveComputeClusterUser(r.Context(), r.PathValue("id"), caller.UserID)
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ common.WriteJSON(w, http.StatusOK, cu)
+}
+
+type denyClusterAccountRequest struct {
+ Note string `json:"note"`
+}
+
+// @Summary Deny a cluster account
+// @Description Records the caller's decision and the note. No account is created on the cluster. The user keeps portal access.
+// @Tags Compute Cluster Users
+// @Security BearerAuth
+// @Accept json
+// @Produce json
+// @Param id path string true "Compute cluster user ID"
+// @Param request body denyClusterAccountRequest false "Optional note"
+// @Success 200 {object} models.ComputeClusterUser
+// @Failure 404 {object} object{error=string}
+// @Failure 409 {object} object{error=string} "Already approved or denied"
+// @Router /compute-cluster-users/{id}/deny [post]
+func (s *Server) denyClusterAccount(w http.ResponseWriter, r *http.Request) {
+ caller := requireCaller(w, r)
+ if caller == nil {
+ return
+ }
+ var req denyClusterAccountRequest
+ if r.ContentLength != 0 {
+ if err := common.DecodeJSON(r, &req); err != nil {
+ common.WriteError(w, http.StatusBadRequest, err)
+ return
+ }
+ }
+ cu, err := s.svc.DenyComputeClusterUser(r.Context(), r.PathValue("id"), caller.UserID, strings.TrimSpace(req.Note))
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ common.WriteJSON(w, http.StatusOK, cu)
+}
+
// @Summary List users on a compute cluster
// @Tags Compute Cluster Users
// @Security BearerAuth
diff --git a/internal/server/responses.go b/internal/server/responses.go
index 7072d01e8..3e9518eca 100644
--- a/internal/server/responses.go
+++ b/internal/server/responses.go
@@ -83,6 +83,20 @@ type AllocationMembershipResponse struct {
Email string `json:"email,omitempty"`
}
+// ClusterAccountResponse is a cluster user with the user's name and email and the cluster name, for the review list.
+type ClusterAccountResponse struct {
+ models.ComputeClusterUser
+ DisplayName string `json:"display_name"`
+ Email string `json:"email"`
+ ClusterName string `json:"cluster_name"`
+}
+
+// ClusterAccountListResponse is the paginated list envelope for cluster accounts.
+type ClusterAccountListResponse struct {
+ Items []ClusterAccountResponse `json:"items"`
+ Total int `json:"total"`
+}
+
// ComputeAllocationListResponse is the paginated list envelope for compute
// allocations.
type ComputeAllocationListResponse struct {
diff --git a/internal/server/server.go b/internal/server/server.go
index 733acd16d..6c646c6c7 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -86,6 +86,9 @@ func (s *Server) routes() {
s.router.RequirePrivilege("GET /compute-clusters/{id}", models.ClustersRead, s.getComputeCluster)
s.router.RequirePrivilege("POST /compute-cluster-users", models.ClustersWrite, s.createComputeClusterUser)
+ s.router.RequirePrivilege("GET /compute-cluster-users", models.ClustersRead, s.listClusterAccounts)
+ s.router.RequirePrivilege("POST /compute-cluster-users/{id}/approve", models.ClustersWrite, s.approveClusterAccount)
+ s.router.RequirePrivilege("POST /compute-cluster-users/{id}/deny", models.ClustersWrite, s.denyClusterAccount)
s.router.RequirePrivilege("GET /compute-cluster-users/{id}", models.ClustersRead, s.getComputeClusterUser)
s.router.RequirePrivilege("PUT /compute-cluster-users/{id}", models.ClustersWrite, s.updateComputeClusterUser)
s.router.RequirePrivilege("DELETE /compute-cluster-users/{id}", models.ClustersWrite, s.deleteComputeClusterUser)
diff --git a/internal/store/compute_cluster_user_store.go b/internal/store/compute_cluster_user_store.go
index cffaae98d..5769a81c8 100644
--- a/internal/store/compute_cluster_user_store.go
+++ b/internal/store/compute_cluster_user_store.go
@@ -21,12 +21,23 @@ import (
"context"
"database/sql"
"errors"
+ "time"
"github.com/jmoiron/sqlx"
"github.com/apache/airavata-custos/pkg/models"
)
+// ComputeClusterUserWithUser is a cluster user joined with the fields the
+// review list shows.
+type ComputeClusterUserWithUser struct {
+ models.ComputeClusterUser
+ FirstName string `db:"first_name"`
+ LastName string `db:"last_name"`
+ Email string `db:"user_email"`
+ ClusterName string `db:"cluster_name"`
+}
+
type pgComputeClusterUserStore struct {
db *sqlx.DB
}
@@ -36,7 +47,7 @@ func NewComputeClusterUserStore(db *sqlx.DB) ComputeClusterUserStore {
return &pgComputeClusterUserStore{db: db}
}
-const computeClusterUserColumns = `id, compute_cluster_id, user_id, local_username, access_level, provisioned_at`
+const computeClusterUserColumns = `id, compute_cluster_id, user_id, local_username, access_level, provisioned_at, approval_status, reviewed_at, reviewed_by, review_note`
func (s *pgComputeClusterUserStore) FindByID(ctx context.Context, id string) (*models.ComputeClusterUser, error) {
var c models.ComputeClusterUser
@@ -110,9 +121,42 @@ func (s *pgComputeClusterUserStore) FindByUser(ctx context.Context, userID strin
func (s *pgComputeClusterUserStore) Create(ctx context.Context, tx *sql.Tx, c *models.ComputeClusterUser) error {
_, err := tx.ExecContext(ctx,
- `INSERT INTO compute_cluster_users (id, compute_cluster_id, user_id, local_username, access_level)
- VALUES ($1, $2, $3, $4, $5)`,
- c.ID, c.ComputeClusterID, c.UserID, c.LocalUsername, c.AccessLevel)
+ `INSERT INTO compute_cluster_users (id, compute_cluster_id, user_id, local_username, access_level, approval_status, reviewed_at, reviewed_by)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
+ c.ID, c.ComputeClusterID, c.UserID, c.LocalUsername, c.AccessLevel, c.ApprovalStatus, c.ReviewedAt, c.ReviewedBy)
+ return err
+}
+
+func (s *pgComputeClusterUserStore) ListByApprovalStatus(ctx context.Context, status models.ClusterAccountApproval, limit, offset int) ([]ComputeClusterUserWithUser, int, error) {
+ var rows []ComputeClusterUserWithUser
+ err := s.db.SelectContext(ctx, &rows,
+ `SELECT cu.id, cu.compute_cluster_id, cu.user_id, cu.local_username, cu.access_level, cu.provisioned_at,
+ cu.approval_status, cu.reviewed_at, cu.reviewed_by, cu.review_note,
+ u.first_name, u.last_name, u.email AS user_email, c.name AS cluster_name
+ FROM compute_cluster_users cu
+ JOIN users u ON u.id = cu.user_id
+ JOIN compute_clusters c ON c.id = cu.compute_cluster_id
+ WHERE $1 = '' OR cu.approval_status = $1
+ ORDER BY cu.created_at DESC
+ LIMIT $2 OFFSET $3`,
+ status, limit, offset)
+ if err != nil {
+ return nil, 0, err
+ }
+ var total int
+ if err := s.db.GetContext(ctx, &total,
+ `SELECT COUNT(*) FROM compute_cluster_users WHERE $1 = '' OR approval_status = $1`, status); err != nil {
+ return nil, 0, err
+ }
+ return rows, total, nil
+}
+
+func (s *pgComputeClusterUserStore) Review(ctx context.Context, tx *sql.Tx, id string, status models.ClusterAccountApproval, reviewedBy, note string, at time.Time) error {
+ _, err := tx.ExecContext(ctx,
+ `UPDATE compute_cluster_users
+ SET approval_status = $1, reviewed_by = $2, reviewed_at = $3, review_note = NULLIF($4, '')
+ WHERE id = $5`,
+ status, reviewedBy, at, note, id)
return err
}
diff --git a/internal/store/store.go b/internal/store/store.go
index 2eb7df980..433834cba 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -93,8 +93,12 @@ type ComputeClusterUserStore interface {
FindByCluster(ctx context.Context, clusterID string) ([]models.ComputeClusterUser, error)
// FindByUser returns every cluster mapping held by the given Custos user.
FindByUser(ctx context.Context, userID string) ([]models.ComputeClusterUser, error)
+ // ListByApprovalStatus returns mappings joined with the user and cluster, newest first, plus the total. An empty status means any status.
+ ListByApprovalStatus(ctx context.Context, status models.ClusterAccountApproval, limit, offset int) ([]ComputeClusterUserWithUser, int, error)
// Create inserts a new mapping within the provided transaction.
Create(ctx context.Context, tx *sql.Tx, c *models.ComputeClusterUser) error
+ // Review records the admin's decision on the mapping within the provided transaction.
+ Review(ctx context.Context, tx *sql.Tx, id string, status models.ClusterAccountApproval, reviewedBy, note string, at time.Time) error
// Update replaces mutable fields of an existing mapping within the provided transaction.
Update(ctx context.Context, tx *sql.Tx, c *models.ComputeClusterUser) error
// MarkProvisioned stamps provisioned_at on the mapping within the provided transaction.
diff --git a/pkg/events/compute_cluster_user_subscribe.go b/pkg/events/compute_cluster_user_subscribe.go
index 6a4749d35..6a0650029 100644
--- a/pkg/events/compute_cluster_user_subscribe.go
+++ b/pkg/events/compute_cluster_user_subscribe.go
@@ -33,6 +33,13 @@ func (b *Bus) SubscribeComputeClusterUserCreated(subscriber string, handler Comp
subscribeTyped(b, subscriber, ComputeClusterUserCreateEvent, handler)
}
+// SubscribeComputeClusterUserApproved registers a typed handler invoked
+// whenever a compute_cluster_user::approve event is published, which is when
+// an admin approves the account, and it can be created on the cluster.
+func (b *Bus) SubscribeComputeClusterUserApproved(subscriber string, handler ComputeClusterUserHandler) {
+ subscribeTyped(b, subscriber, ComputeClusterUserApproveEvent, handler)
+}
+
// SubscribeComputeClusterUserUpdated registers a typed handler invoked
// whenever a compute_cluster_user::update event is published.
func (b *Bus) SubscribeComputeClusterUserUpdated(subscriber string, handler ComputeClusterUserHandler) {
diff --git a/pkg/events/types.go b/pkg/events/types.go
index b86254b0b..a5bc7e5c9 100644
--- a/pkg/events/types.go
+++ b/pkg/events/types.go
@@ -61,9 +61,10 @@ const (
// ComputeClusterUser lifecycle message types.
const (
- ComputeClusterUserCreateEvent EventType = "compute_cluster_user::create"
- ComputeClusterUserUpdateEvent EventType = "compute_cluster_user::update"
- ComputeClusterUserDeleteEvent EventType = "compute_cluster_user::delete"
+ ComputeClusterUserCreateEvent EventType = "compute_cluster_user::create"
+ ComputeClusterUserUpdateEvent EventType = "compute_cluster_user::update"
+ ComputeClusterUserDeleteEvent EventType = "compute_cluster_user::delete"
+ ComputeClusterUserApproveEvent EventType = "compute_cluster_user::approve"
)
// ClusterAccount lifecycle message types.
diff --git a/pkg/models/allocation.go b/pkg/models/allocation.go
index e6adefc20..675815cfc 100644
--- a/pkg/models/allocation.go
+++ b/pkg/models/allocation.go
@@ -41,13 +41,27 @@ const (
ClusterAccessAdmin ClusterAccessLevel = "ADMIN"
)
+// ClusterAccountApproval is the admin's decision on a cluster account.
+// The account is created on the cluster only after it is approved.
+type ClusterAccountApproval string
+
+const (
+ ClusterAccountPending ClusterAccountApproval = "PENDING"
+ ClusterAccountApproved ClusterAccountApproval = "APPROVED"
+ ClusterAccountDenied ClusterAccountApproval = "DENIED"
+)
+
type ComputeClusterUser struct {
- ID string `json:"id" db:"id"`
- ComputeClusterID string `json:"compute_cluster_id" db:"compute_cluster_id"`
- UserID string `json:"user_id" db:"user_id"`
- LocalUsername string `json:"local_username" db:"local_username"` // The username of the user on the compute cluster, which may be different from their Airavata Custos username.
- AccessLevel ClusterAccessLevel `json:"access_level" db:"access_level"`
- ProvisionedAt *time.Time `json:"provisioned_at" db:"provisioned_at"` // When the account was provisioned into the registry; nil until then.
+ ID string `json:"id" db:"id"`
+ ComputeClusterID string `json:"compute_cluster_id" db:"compute_cluster_id"`
+ UserID string `json:"user_id" db:"user_id"`
+ LocalUsername string `json:"local_username" db:"local_username"` // The username of the user on the compute cluster, which may be different from their Airavata Custos username.
+ AccessLevel ClusterAccessLevel `json:"access_level" db:"access_level"`
+ ProvisionedAt *time.Time `json:"provisioned_at" db:"provisioned_at"` // When the account was provisioned into the registry; nil until then.
+ ApprovalStatus ClusterAccountApproval `json:"approval_status" db:"approval_status"`
+ ReviewedAt *time.Time `json:"reviewed_at,omitempty" db:"reviewed_at"`
+ ReviewedBy *string `json:"reviewed_by,omitempty" db:"reviewed_by"` // The admin who approved or denied the cluster account.
+ ReviewNote *string `json:"review_note,omitempty" db:"review_note"`
}
type ComputeAllocation struct {
diff --git a/pkg/service/compute_cluster_user.go b/pkg/service/compute_cluster_user.go
index 950a79552..0e59f08ee 100644
--- a/pkg/service/compute_cluster_user.go
+++ b/pkg/service/compute_cluster_user.go
@@ -24,11 +24,16 @@ import (
"fmt"
"strings"
+ "github.com/apache/airavata-custos/internal/store"
"github.com/apache/airavata-custos/pkg/events"
"github.com/apache/airavata-custos/pkg/models"
)
-const clusterUserAuditAdminGranted = "CLUSTER_ADMIN_GRANTED"
+const (
+ clusterUserAuditAdminGranted = "CLUSTER_ADMIN_GRANTED"
+ clusterUserAuditApproved = "CLUSTER_ACCOUNT_APPROVED"
+ clusterUserAuditDenied = "CLUSTER_ACCOUNT_DENIED"
+)
func (s *Service) writeClusterUserAuditTx(ctx context.Context, tx *sql.Tx, eventType, entityID string, details map[string]any) error {
payload, err := json.Marshal(details)
@@ -71,6 +76,24 @@ func (s *Service) CreateComputeClusterUser(ctx context.Context, cu *models.Compu
return nil, fmt.Errorf("%w: unknown access level %q for user %s on cluster %s, must be %s or %s",
ErrInvalidInput, cu.AccessLevel, cu.UserID, cu.ComputeClusterID, models.ClusterAccessUser, models.ClusterAccessAdmin)
}
+ // A new row waits for review. The admin routes pass it as approved, with the admin who added it as the reviewer.
+ if cu.ApprovalStatus == "" {
+ cu.ApprovalStatus = models.ClusterAccountPending
+ }
+ switch cu.ApprovalStatus {
+ case models.ClusterAccountPending:
+ case models.ClusterAccountApproved:
+ if cu.ReviewedBy == nil || *cu.ReviewedBy == "" {
+ return nil, fmt.Errorf("%w: an approved cluster user needs a reviewer", ErrInvalidInput)
+ }
+ if cu.ReviewedAt == nil {
+ now := nowUTC()
+ cu.ReviewedAt = &now
+ }
+ default:
+ return nil, fmt.Errorf("%w: a new cluster user can be %s or %s, not %s",
+ ErrInvalidInput, models.ClusterAccountPending, models.ClusterAccountApproved, cu.ApprovalStatus)
+ }
if cluster, err := s.clusters.FindByID(ctx, cu.ComputeClusterID); err != nil {
return nil, fmt.Errorf("lookup compute cluster: %w", err)
@@ -94,7 +117,14 @@ func (s *Service) CreateComputeClusterUser(ctx context.Context, cu *models.Compu
if err := s.clusterUsers.Create(ctx, tx, cu); err != nil {
return err
}
- return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserCreateEvent, cu)
+ if err := s.eventBus.Publish(ctx, tx, events.ComputeClusterUserCreateEvent, cu); err != nil {
+ return err
+ }
+ // Approved at creation, so provisioning starts now.
+ if cu.ApprovalStatus == models.ClusterAccountApproved {
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserApproveEvent, cu)
+ }
+ return nil
}); err != nil {
switch {
case isLocalUsernameDuplicate(err):
@@ -109,6 +139,113 @@ func (s *Service) CreateComputeClusterUser(ctx context.Context, cu *models.Compu
return cu, nil
}
+// ListComputeClusterUsersByApproval returns cluster users in the given
+// approval status with their user and cluster, newest first. An empty status means any status.
+func (s *Service) ListComputeClusterUsersByApproval(ctx context.Context, status models.ClusterAccountApproval, limit, offset int) ([]store.ComputeClusterUserWithUser, int, error) {
+ switch status {
+ case "", models.ClusterAccountPending, models.ClusterAccountApproved, models.ClusterAccountDenied:
+ default:
+ return nil, 0, fmt.Errorf("%w: unknown approval status %q", ErrInvalidInput, status)
+ }
+ rows, total, err := s.clusterUsers.ListByApprovalStatus(ctx, status, limit, offset)
+ if err != nil {
+ return nil, 0, fmt.Errorf("list compute cluster users by approval: %w", err)
+ }
+ return rows, total, nil
+}
+
+// ApproveComputeClusterUser records the admin's approval and publishes the
+// `approve event`, which is what creates the account on the cluster. A denied account can be approved later.
+func (s *Service) ApproveComputeClusterUser(ctx context.Context, id, reviewerID string) (*models.ComputeClusterUser, error) {
+ cu, err := s.clusterUserForReview(ctx, id, reviewerID)
+ if err != nil {
+ return nil, err
+ }
+
+ // Only a denied or pending cluster account can be approved
+ if cu.ApprovalStatus == models.ClusterAccountApproved {
+ return nil, fmt.Errorf("%w: cluster user %q is already approved", ErrAlreadyExists, id)
+ }
+ now := nowUTC()
+ cu.ApprovalStatus = models.ClusterAccountApproved
+ cu.ReviewedAt = &now
+ cu.ReviewedBy = &reviewerID
+ cu.ReviewNote = nil
+ if err := s.inTx(ctx, func(tx *sql.Tx) error {
+ if err := s.clusterUsers.Review(ctx, tx, id, models.ClusterAccountApproved, reviewerID, "", now); err != nil {
+ return err
+ }
+ if err := s.writeClusterUserAuditTx(ctx, tx, clusterUserAuditApproved, id, map[string]any{
+ "actor_id": reviewerID,
+ "user_id": cu.UserID,
+ "cluster_id": cu.ComputeClusterID,
+ "local_username": cu.LocalUsername,
+ }); err != nil {
+ return err
+ }
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserApproveEvent, cu)
+ }); err != nil {
+ return nil, fmt.Errorf("approve compute cluster user: %w", err)
+ }
+ return cu, nil
+}
+
+// DenyComputeClusterUser records the admin's denial. No event is published,
+// so the account is never created on the cluster. An approved account cannot
+// be denied, since it may already exist on the cluster.
+func (s *Service) DenyComputeClusterUser(ctx context.Context, id, reviewerID, note string) (*models.ComputeClusterUser, error) {
+ cu, err := s.clusterUserForReview(ctx, id, reviewerID)
+ if err != nil {
+ return nil, err
+ }
+ switch cu.ApprovalStatus {
+ case models.ClusterAccountApproved:
+ return nil, fmt.Errorf("%w: cluster user %q is already approved, remove it instead", ErrAlreadyExists, id)
+ case models.ClusterAccountDenied:
+ return nil, fmt.Errorf("%w: cluster user %q is already denied", ErrAlreadyExists, id)
+ }
+ now := nowUTC()
+ cu.ApprovalStatus = models.ClusterAccountDenied
+ cu.ReviewedAt = &now
+ cu.ReviewedBy = &reviewerID
+ cu.ReviewNote = nil
+ if note != "" {
+ cu.ReviewNote = ¬e
+ }
+ if err := s.inTx(ctx, func(tx *sql.Tx) error {
+ if err := s.clusterUsers.Review(ctx, tx, id, models.ClusterAccountDenied, reviewerID, note, now); err != nil {
+ return err
+ }
+ return s.writeClusterUserAuditTx(ctx, tx, clusterUserAuditDenied, id, map[string]any{
+ "actor_id": reviewerID,
+ "user_id": cu.UserID,
+ "cluster_id": cu.ComputeClusterID,
+ "local_username": cu.LocalUsername,
+ "note": note,
+ })
+ }); err != nil {
+ return nil, fmt.Errorf("deny compute cluster user: %w", err)
+ }
+ return cu, nil
+}
+
+func (s *Service) clusterUserForReview(ctx context.Context, id, reviewerID string) (*models.ComputeClusterUser, error) {
+ if id == "" {
+ return nil, fmt.Errorf("%w: compute cluster user id is required", ErrInvalidInput)
+ }
+ if reviewerID == "" {
+ return nil, fmt.Errorf("%w: reviewer is required", ErrInvalidInput)
+ }
+ cu, err := s.clusterUsers.FindByID(ctx, id)
+ if err != nil {
+ return nil, fmt.Errorf("lookup compute cluster user: %w", err)
+ }
+ if cu == nil {
+ return nil, ErrNotFound
+ }
+ return cu, nil
+}
+
func isLocalUsernameDuplicate(err error) bool {
return err != nil && strings.Contains(err.Error(), "uq_compute_cluster_users_local_username")
}
diff --git a/pkg/service/compute_cluster_user_integration_test.go b/pkg/service/compute_cluster_user_integration_test.go
index 1818d21b5..fe631f32a 100644
--- a/pkg/service/compute_cluster_user_integration_test.go
+++ b/pkg/service/compute_cluster_user_integration_test.go
@@ -20,11 +20,13 @@
package service
import (
+ "context"
"errors"
"fmt"
"testing"
"github.com/google/uuid"
+ "github.com/jmoiron/sqlx"
"github.com/apache/airavata-custos/pkg/models"
)
@@ -84,3 +86,177 @@ func TestMarkComputeClusterUserProvisioned_RoundTrip(t *testing.T) {
t.Fatalf("expected ErrNotFound for unknown id, got %v", err)
}
}
+
+// newClusterUser creates a cluster, a user, and a cluster user with no approval
+// given, the way an allocation source does, and returns the cluster user.
+func newClusterUser(t *testing.T, svc *Service, database *sqlx.DB) *models.ComputeClusterUser {
+ t.Helper()
+ cluster, err := svc.CreateComputeCluster(ctx(), &models.ComputeCluster{Name: "appr-" + uuid.NewString()[:8]})
+ if err != nil {
+ t.Fatalf("create cluster: %v", err)
+ }
+ userID := seedUser(t, database, fmt.Sprintf("member-%s@example.edu", uuid.NewString()))
+ cu, err := svc.CreateComputeClusterUser(ctx(), &models.ComputeClusterUser{
+ ComputeClusterID: cluster.ID,
+ UserID: userID,
+ LocalUsername: "appr-" + uuid.NewString()[:8],
+ })
+ if err != nil {
+ t.Fatalf("create compute cluster user: %v", err)
+ }
+ return cu
+}
+
+func countDeliveries(t *testing.T, svc *Service) int {
+ t.Helper()
+ rows, err := svc.EventBus().ListDeliveries(ctx(), "", 10)
+ if err != nil {
+ t.Fatalf("list deliveries: %v", err)
+ }
+ return len(rows)
+}
+
+// Make sure a cluster user created without an approval waits for an admin
+// instead of being created on the cluster.
+func TestCreateComputeClusterUser_WaitsForApproval(t *testing.T) {
+ database := setupTestDB(t)
+ svc := newTestService(database)
+
+ cu := newClusterUser(t, svc, database)
+ if cu.ApprovalStatus != models.ClusterAccountPending || cu.ReviewedBy != nil {
+ t.Fatalf("expected a pending row with no reviewer, got %+v", cu)
+ }
+}
+
+// Make sure a cluster user an admin adds directly is approved and provisioning
+// starts at once, without a separate approval step.
+func TestCreateComputeClusterUser_ApprovedByAdminStartsProvisioning(t *testing.T) {
+ database := setupTestDB(t)
+ svc := newTestService(database)
+ svc.EventBus().SubscribeComputeClusterUserApproved("test-subscriber", func(context.Context, models.ComputeClusterUser) error { return nil })
+ cluster, err := svc.CreateComputeCluster(ctx(), &models.ComputeCluster{Name: "appr-" + uuid.NewString()[:8]})
+ if err != nil {
+ t.Fatalf("create cluster: %v", err)
+ }
+ admin := seedUser(t, database, fmt.Sprintf("admin-%s@example.edu", uuid.NewString()))
+ member := seedUser(t, database, fmt.Sprintf("member-%s@example.edu", uuid.NewString()))
+
+ cu, err := svc.CreateComputeClusterUser(ctx(), &models.ComputeClusterUser{
+ ComputeClusterID: cluster.ID,
+ UserID: member,
+ LocalUsername: "appr-" + uuid.NewString()[:8],
+ ApprovalStatus: models.ClusterAccountApproved,
+ ReviewedBy: &admin,
+ })
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+ if cu.ApprovalStatus != models.ClusterAccountApproved || cu.ReviewedAt == nil {
+ t.Fatalf("expected an approved row with a review time, got %+v", cu)
+ }
+ if n := countDeliveries(t, svc); n != 1 {
+ t.Fatalf("expected one delivery of the approve event, got %d", n)
+ }
+}
+
+// Make sure an approval stores who approved, leaves an audit row, and publishes
+// the approve event so the account gets created on the cluster.
+func TestApproveComputeClusterUser_PublishesApproveEvent(t *testing.T) {
+ database := setupTestDB(t)
+ svc := newTestService(database)
+ svc.EventBus().SubscribeComputeClusterUserApproved("test-subscriber", func(context.Context, models.ComputeClusterUser) error { return nil })
+ cu := newClusterUser(t, svc, database)
+ admin := seedUser(t, database, fmt.Sprintf("admin-%s@example.edu", uuid.NewString()))
+
+ got, err := svc.ApproveComputeClusterUser(ctx(), cu.ID, admin)
+ if err != nil {
+ t.Fatalf("approve: %v", err)
+ }
+ if got.ApprovalStatus != models.ClusterAccountApproved || got.ReviewedBy == nil || *got.ReviewedBy != admin || got.ReviewedAt == nil {
+ t.Fatalf("expected an approved row reviewed by %s, got %+v", admin, got)
+ }
+ if n := countDeliveries(t, svc); n != 1 {
+ t.Fatalf("expected one delivery of the approve event, got %d", n)
+ }
+ if n := countAuditEventsOfType(t, database, clusterUserAuditApproved, cu.ID); n != 1 {
+ t.Fatalf("expected one approval audit row, got %d", n)
+ }
+}
+
+// Make sure a denial stores who denied and the note, leaves an audit row, and
+// publishes nothing, so the account is never created on the cluster.
+func TestDenyComputeClusterUser_PublishesNothing(t *testing.T) {
+ database := setupTestDB(t)
+ svc := newTestService(database)
+ svc.EventBus().SubscribeComputeClusterUserApproved("test-subscriber", func(context.Context, models.ComputeClusterUser) error { return nil })
+ cu := newClusterUser(t, svc, database)
+ admin := seedUser(t, database, fmt.Sprintf("admin-%s@example.edu", uuid.NewString()))
+
+ got, err := svc.DenyComputeClusterUser(ctx(), cu.ID, admin, "not on the collaborator list")
+ if err != nil {
+ t.Fatalf("deny: %v", err)
+ }
+ if got.ApprovalStatus != models.ClusterAccountDenied || got.ReviewNote == nil || *got.ReviewNote != "not on the collaborator list" {
+ t.Fatalf("expected a denied row with the note, got %+v", got)
+ }
+ if n := countDeliveries(t, svc); n != 0 {
+ t.Fatalf("expected no delivery for a denied account, got %d", n)
+ }
+ if n := countAuditEventsOfType(t, database, clusterUserAuditDenied, cu.ID); n != 1 {
+ t.Fatalf("expected one denial audit row, got %d", n)
+ }
+}
+
+// Make sure an approved account cannot be approved again or denied, since it
+// may already exist on the cluster, while a denied one can still be approved.
+func TestReviewComputeClusterUser_ApprovalIsFinal(t *testing.T) {
+ database := setupTestDB(t)
+ svc := newTestService(database)
+ cu := newClusterUser(t, svc, database)
+ admin := seedUser(t, database, fmt.Sprintf("admin-%s@example.edu", uuid.NewString()))
+
+ if _, err := svc.DenyComputeClusterUser(ctx(), cu.ID, admin, ""); err != nil {
+ t.Fatalf("deny: %v", err)
+ }
+ if _, err := svc.ApproveComputeClusterUser(ctx(), cu.ID, admin); err != nil {
+ t.Fatalf("approve after deny: %v", err)
+ }
+ if _, err := svc.ApproveComputeClusterUser(ctx(), cu.ID, admin); !errors.Is(err, ErrAlreadyExists) {
+ t.Fatalf("expected ErrAlreadyExists on a second approval, got %v", err)
+ }
+ if _, err := svc.DenyComputeClusterUser(ctx(), cu.ID, admin, ""); !errors.Is(err, ErrAlreadyExists) {
+ t.Fatalf("expected ErrAlreadyExists denying an approved account, got %v", err)
+ }
+}
+
+// Make sure a cluster user created through onboarding is approved by the
+// onboarding admin and provisioning starts at once, since adding the user by hand is the approval.
+func TestOnboardUser_ClusterAccountIsApproved(t *testing.T) {
+ database := setupTestDB(t)
+ svc := newTestService(database)
+ svc.EventBus().SubscribeComputeClusterUserApproved("test-subscriber", func(context.Context, models.ComputeClusterUser) error { return nil })
+ onboarder := seedUser(t, database, fmt.Sprintf("onboarder-%s@example.edu", uuid.NewString()))
+ cluster, err := svc.CreateComputeCluster(ctx(), &models.ComputeCluster{Name: "onb-" + uuid.NewString()[:8]})
+ if err != nil {
+ t.Fatalf("create cluster: %v", err)
+ }
+ in := onboardInput(fmt.Sprintf("researcher-%s@example.edu", uuid.NewString()))
+ in.Username = "onb-" + uuid.NewString()[:8]
+ in.ComputeClusterID = cluster.ID
+ in.OnboardedBy = onboarder
+
+ created, err := svc.OnboardUser(ctx(), in)
+ if err != nil {
+ t.Fatalf("onboard: %v", err)
+ }
+ cu, err := svc.GetComputeClusterUserByPair(ctx(), cluster.ID, created.ID)
+ if err != nil {
+ t.Fatalf("cluster user: %v", err)
+ }
+ if cu.ApprovalStatus != models.ClusterAccountApproved || cu.ReviewedBy == nil || *cu.ReviewedBy != onboarder {
+ t.Fatalf("expected approval by the onboarding admin, got %+v", cu)
+ }
+ if n := countDeliveries(t, svc); n != 1 {
+ t.Fatalf("expected one delivery of the approve event, got %d", n)
+ }
+}
diff --git a/pkg/service/user_onboard.go b/pkg/service/user_onboard.go
index 516342482..4733e773a 100644
--- a/pkg/service/user_onboard.go
+++ b/pkg/service/user_onboard.go
@@ -182,12 +182,17 @@ func (s *Service) OnboardUser(ctx context.Context, in OnboardUserInput) (*models
}
}
+ // Onboarding is an admin creating the user account on purpose, so it counts as the approval.
+ reviewedAt := nowUTC()
clusterUser := &models.ComputeClusterUser{
ID: newID(),
ComputeClusterID: in.ComputeClusterID,
UserID: created.ID,
LocalUsername: in.Username,
AccessLevel: models.ClusterAccessUser,
+ ApprovalStatus: models.ClusterAccountApproved,
+ ReviewedAt: &reviewedAt,
+ ReviewedBy: &in.OnboardedBy,
}
if in.ClusterAdmin {
clusterUser.AccessLevel = models.ClusterAccessAdmin
@@ -257,7 +262,11 @@ func (s *Service) OnboardUser(ctx context.Context, in OnboardUserInput) (*models
return err
}
if needsCluster {
- return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserCreateEvent, clusterUser)
+ if err := s.eventBus.Publish(ctx, tx, events.ComputeClusterUserCreateEvent, clusterUser); err != nil {
+ return err
+ }
+ // Approved at creation, so provisioning starts now.
+ return s.eventBus.Publish(ctx, tx, events.ComputeClusterUserApproveEvent, clusterUser)
}
return nil
}); err != nil {