You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by la...@apache.org on 2026/07/23 19:18:36 UTC
(airavata-custos) 04/35: Add access request endpoints and approval orchestration
This is an automated email from the ASF dual-hosted git repository.
lahirujayathilake pushed a commit to branch pearc26
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit 9c1df2e1d5eec2989faec4ad3f01f204fb4405df
Author: lahiruj <la...@gmail.com>
AuthorDate: Thu Jul 16 01:18:25 2026 -0400
Add access request endpoints and approval orchestration
---
api/core.openapi.yaml | 230 +++++++++++++++
docs/privileges.md | 10 +-
internal/server/access_request.go | 199 +++++++++++++
internal/server/server.go | 8 +
pkg/models/access_request.go | 3 +
pkg/models/privilege.go | 5 +
pkg/service/access_request.go | 329 +++++++++++++++++++++
pkg/service/access_request_integration_test.go | 387 +++++++++++++++++++++++++
8 files changed, 1169 insertions(+), 2 deletions(-)
diff --git a/api/core.openapi.yaml b/api/core.openapi.yaml
index 6d5e96fe7..57ea4b058 100644
--- a/api/core.openapi.yaml
+++ b/api/core.openapi.yaml
@@ -1,5 +1,44 @@
basePath: /
definitions:
+ AccessRequest:
+ properties:
+ approver_id:
+ type: string
+ created_user_id:
+ type: string
+ deny_reason:
+ type: string
+ email:
+ type: string
+ event_code:
+ type: string
+ expires_at:
+ type: string
+ id:
+ type: string
+ institution:
+ type: string
+ name:
+ type: string
+ oidc_sub:
+ type: string
+ reason:
+ type: string
+ status:
+ $ref: '#/definitions/AccessRequestStatus'
+ timestamp:
+ type: string
+ type: object
+ AccessRequestStatus:
+ enum:
+ - PENDING
+ - APPROVED
+ - DENIED
+ type: string
+ x-enum-varnames:
+ - AccessRequestPending
+ - AccessRequestApproved
+ - AccessRequestDenied
AllocationMembershipResponse:
properties:
compute_allocation_id:
@@ -369,6 +408,8 @@ definitions:
- core:traces:read
- core:privileges:grant
- core:roles:manage
+ - core:access-requests:read
+ - core:access-requests:write
type: string
x-enum-varnames:
- ClustersRead
@@ -384,6 +425,8 @@ definitions:
- TracesRead
- PrivilegesGrant
- RolesManage
+ - AccessRequestsRead
+ - AccessRequestsWrite
Project:
properties:
created_time:
@@ -752,6 +795,187 @@ info:
title: Apache Custos Core API
version: 0.2.0
paths:
+ /access-requests:
+ get:
+ parameters:
+ - description: Filter by status
+ in: query
+ name: status
+ type: string
+ - description: Filter by event code
+ in: query
+ name: event
+ type: string
+ - description: Maximum rows (default 50)
+ in: query
+ name: limit
+ type: integer
+ produces:
+ - application/json
+ responses:
+ "200":
+ description: OK
+ schema:
+ items:
+ $ref: '#/definitions/AccessRequest'
+ type: array
+ security:
+ - BearerAuth: []
+ summary: List access requests
+ tags:
+ - Access Requests
+ post:
+ consumes:
+ - application/json
+ parameters:
+ - description: Access request payload
+ in: body
+ name: request
+ required: true
+ schema:
+ properties:
+ event_code:
+ type: string
+ institution:
+ type: string
+ reason:
+ type: string
+ type: object
+ produces:
+ - application/json
+ responses:
+ "201":
+ description: Created
+ schema:
+ $ref: '#/definitions/AccessRequest'
+ "400":
+ description: Bad Request
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ "404":
+ description: Not Found
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ "409":
+ description: Conflict
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ security:
+ - BearerAuth: []
+ summary: Submit an access request
+ tags:
+ - Access Requests
+ /access-requests/{id}:
+ put:
+ consumes:
+ - application/json
+ parameters:
+ - description: Access request ID
+ in: path
+ name: id
+ required: true
+ type: string
+ - description: Decision payload
+ in: body
+ name: request
+ required: true
+ schema:
+ properties:
+ deny_reason:
+ type: string
+ expires_at:
+ type: string
+ status:
+ type: string
+ type: object
+ produces:
+ - application/json
+ responses:
+ "200":
+ description: OK
+ schema:
+ $ref: '#/definitions/AccessRequest'
+ "400":
+ description: Bad Request
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ "404":
+ description: Not Found
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ security:
+ - BearerAuth: []
+ summary: Approve or deny an access request
+ tags:
+ - Access Requests
+ /access-requests/events/{code}:
+ get:
+ parameters:
+ - description: Event code
+ in: path
+ name: code
+ required: true
+ type: string
+ produces:
+ - application/json
+ responses:
+ "200":
+ description: OK
+ schema:
+ properties:
+ code:
+ type: string
+ name:
+ type: string
+ type: object
+ "404":
+ description: Not Found
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ security:
+ - BearerAuth: []
+ summary: Resolve an access event code
+ tags:
+ - Access Requests
+ /access-requests/me:
+ get:
+ produces:
+ - application/json
+ responses:
+ "200":
+ description: OK
+ schema:
+ $ref: '#/definitions/AccessRequest'
+ "404":
+ description: Not Found
+ schema:
+ properties:
+ error:
+ type: string
+ type: object
+ security:
+ - BearerAuth: []
+ summary: Get the caller's latest access request
+ tags:
+ - Access Requests
/audit/events:
get:
parameters:
@@ -2864,6 +3088,8 @@ paths:
- core:traces:read
- core:privileges:grant
- core:roles:manage
+ - core:access-requests:read
+ - core:access-requests:write
in: path
name: key
required: true
@@ -3389,6 +3615,8 @@ paths:
- core:traces:read
- core:privileges:grant
- core:roles:manage
+ - core:access-requests:read
+ - core:access-requests:write
in: path
name: key
required: true
@@ -3985,6 +4213,8 @@ paths:
- core:traces:read
- core:privileges:grant
- core:roles:manage
+ - core:access-requests:read
+ - core:access-requests:write
in: path
name: key
required: true
diff --git a/docs/privileges.md b/docs/privileges.md
index afedecd13..5ce531187 100644
--- a/docs/privileges.md
+++ b/docs/privileges.md
@@ -34,12 +34,18 @@ should be migrated.
| `core:organizations:read` | `GET /organizations`, `GET /organizations/{id}` |
| `core:organizations:write` | `POST /organizations` |
| `core:traces:read` | `GET /audit/traces`, `GET /audit/traces/{trace_id}`, `GET /audit/events`, `GET /audit/sources` |
+| `core:access-requests:read` | `GET /access-requests` |
+| `core:access-requests:write` | `PUT /access-requests/{id}` |
| `core:privileges:grant` | `GET /privileges/catalog`, `GET /users/{id}/privileges`, `GET /privileges/{key}/holders`, `POST /users/{id}/privileges`, `DELETE /users/{id}/privileges/{key}` |
| `core:roles:manage` | `GET /roles`, `POST /roles`, `PUT /roles/{id}`, `DELETE /roles/{id}`, and the role-assignment endpoints under `/users/{id}/roles` |
-Only three routes skip the privilege check by design: `GET /healthz` is
+A few routes skip the privilege check by design: `GET /healthz` is
public, and `GET /me` / `GET /user/privileges` accept any verified caller
-so users can read their own profile and effective privilege set.
+so users can read their own profile and effective privilege set. The
+access-request self-service routes (`GET /access-requests/events/{code}`,
+`POST /access-requests`, `GET /access-requests/me`) accept any verified
+token, even one not yet linked to a user, so a new person can request
+access.
The bootstrap super_admin role carries every privilege registered at
startup, core and connector alike. Individual grants beyond that are made
diff --git a/internal/server/access_request.go b/internal/server/access_request.go
new file mode 100644
index 000000000..3f23f7f5f
--- /dev/null
+++ b/internal/server/access_request.go
@@ -0,0 +1,199 @@
+// 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 (
+ "errors"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/apache/airavata-custos/internal/store"
+ "github.com/apache/airavata-custos/pkg/common"
+ "github.com/apache/airavata-custos/pkg/identity"
+ "github.com/apache/airavata-custos/pkg/models"
+)
+
+type createAccessRequestBody struct {
+ Institution string `json:"institution"`
+ EventCode string `json:"event_code"`
+ Reason string `json:"reason"`
+}
+
+type decideAccessRequestBody struct {
+ Status string `json:"status"`
+ ExpiresAt *time.Time `json:"expires_at"`
+ DenyReason string `json:"deny_reason"`
+}
+
+// @Summary Resolve an access event code
+// @Tags Access Requests
+// @Security BearerAuth
+// @Produce json
+// @Param code path string true "Event code"
+// @Success 200 {object} object{code=string,name=string}
+// @Failure 404 {object} object{error=string}
+// @Router /access-requests/events/{code} [get]
+func (s *Server) getAccessEventByCode(w http.ResponseWriter, r *http.Request) {
+ ev, err := s.svc.GetAccessEvent(r.Context(), r.PathValue("code"))
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ alloc, err := s.svc.GetComputeAllocation(r.Context(), ev.ComputeAllocationID)
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ // Only the code and allocation name go out; the caller is not a user yet.
+ common.WriteJSON(w, http.StatusOK, map[string]string{"code": ev.Code, "name": alloc.Name})
+}
+
+// @Summary Submit an access request
+// @Tags Access Requests
+// @Security BearerAuth
+// @Accept json
+// @Produce json
+// @Param request body object{institution=string,event_code=string,reason=string} true "Access request payload"
+// @Success 201 {object} models.AccessRequest
+// @Failure 400 {object} object{error=string}
+// @Failure 404 {object} object{error=string}
+// @Failure 409 {object} object{error=string}
+// @Router /access-requests [post]
+func (s *Server) createAccessRequest(w http.ResponseWriter, r *http.Request) {
+ claims := identity.ClaimsFromContext(r.Context())
+ if claims == nil {
+ common.WriteError(w, http.StatusUnauthorized, errors.New("verified token required"))
+ return
+ }
+ var body createAccessRequestBody
+ if err := common.DecodeJSON(r, &body); err != nil {
+ common.WriteError(w, http.StatusBadRequest, err)
+ return
+ }
+ if body.Institution == "" || body.EventCode == "" {
+ common.WriteError(w, http.StatusBadRequest, errors.New("institution and event_code are required"))
+ return
+ }
+ name := claims.Name
+ if name == "" {
+ name = strings.TrimSpace(claims.GivenName + " " + claims.FamilyName)
+ }
+ created, err := s.svc.CreateAccessRequest(r.Context(), &models.AccessRequest{
+ OIDCSub: claims.Sub,
+ Email: claims.Email,
+ Name: name,
+ Institution: body.Institution,
+ EventCode: body.EventCode,
+ Reason: body.Reason,
+ })
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ common.WriteJSON(w, http.StatusCreated, created)
+}
+
+// @Summary Get the caller's latest access request
+// @Tags Access Requests
+// @Security BearerAuth
+// @Produce json
+// @Success 200 {object} models.AccessRequest
+// @Failure 404 {object} object{error=string}
+// @Router /access-requests/me [get]
+func (s *Server) getOwnAccessRequest(w http.ResponseWriter, r *http.Request) {
+ claims := identity.ClaimsFromContext(r.Context())
+ if claims == nil {
+ common.WriteError(w, http.StatusUnauthorized, errors.New("verified token required"))
+ return
+ }
+ req, err := s.svc.GetLatestAccessRequestBySub(r.Context(), claims.Sub)
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ common.WriteJSON(w, http.StatusOK, req)
+}
+
+// @Summary List access requests
+// @Tags Access Requests
+// @Security BearerAuth
+// @Produce json
+// @Param status query string false "Filter by status"
+// @Param event query string false "Filter by event code"
+// @Param limit query integer false "Maximum rows (default 50)"
+// @Success 200 {array} models.AccessRequest
+// @Router /access-requests [get]
+func (s *Server) listAccessRequests(w http.ResponseWriter, r *http.Request) {
+ q := r.URL.Query()
+ rows, err := s.svc.ListAccessRequests(r.Context(), store.AccessRequestListFilter{
+ Status: q.Get("status"),
+ EventCode: q.Get("event"),
+ Limit: atoiOr(q.Get("limit"), 50),
+ })
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ if rows == nil {
+ rows = []models.AccessRequest{}
+ }
+ common.WriteJSON(w, http.StatusOK, rows)
+}
+
+// @Summary Approve or deny an access request
+// @Tags Access Requests
+// @Security BearerAuth
+// @Accept json
+// @Produce json
+// @Param id path string true "Access request ID"
+// @Param request body object{status=string,expires_at=string,deny_reason=string} true "Decision payload"
+// @Success 200 {object} models.AccessRequest
+// @Failure 400 {object} object{error=string}
+// @Failure 404 {object} object{error=string}
+// @Router /access-requests/{id} [put]
+func (s *Server) decideAccessRequest(w http.ResponseWriter, r *http.Request) {
+ caller := identity.CallerFromContext(r.Context())
+ if caller == nil {
+ common.WriteError(w, http.StatusUnauthorized, errors.New("linked caller required"))
+ return
+ }
+ var body decideAccessRequestBody
+ if err := common.DecodeJSON(r, &body); err != nil {
+ common.WriteError(w, http.StatusBadRequest, err)
+ return
+ }
+ var (
+ updated *models.AccessRequest
+ err error
+ )
+ switch models.AccessRequestStatus(body.Status) {
+ case models.AccessRequestApproved:
+ updated, err = s.svc.ApproveAccessRequest(r.Context(), r.PathValue("id"), caller.UserID, body.ExpiresAt)
+ case models.AccessRequestDenied:
+ updated, err = s.svc.DenyAccessRequest(r.Context(), r.PathValue("id"), caller.UserID, body.DenyReason)
+ default:
+ common.WriteError(w, http.StatusBadRequest, errors.New("status must be APPROVED or DENIED"))
+ return
+ }
+ if err != nil {
+ common.WriteServiceError(w, err)
+ return
+ }
+ common.WriteJSON(w, http.StatusOK, updated)
+}
diff --git a/internal/server/server.go b/internal/server/server.go
index 18e0af764..e4ef6ef29 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -164,6 +164,14 @@ func (s *Server) routes() {
s.router.RequirePrivilege("GET /user-identities/oidc-subjects/{oidcSub}", models.UsersRead, s.getUserIdentityByOIDCSub)
s.router.RequirePrivilege("GET /users/{id}/user-identities", models.UsersRead, s.listUserIdentitiesForUser)
+ // Self-service routes reachable with a verified token whose identity is
+ // not linked to a user yet.
+ s.router.RequireToken("GET /access-requests/events/{code}", s.getAccessEventByCode)
+ s.router.RequireToken("POST /access-requests", s.createAccessRequest)
+ s.router.RequireToken("GET /access-requests/me", s.getOwnAccessRequest)
+ s.router.RequirePrivilege("GET /access-requests", models.AccessRequestsRead, s.listAccessRequests)
+ s.router.RequirePrivilege("PUT /access-requests/{id}", models.AccessRequestsWrite, s.decideAccessRequest)
+
// A caller reads their own profile; no privilege needed.
s.router.RequireAuth("GET /user/privileges", s.getCallerPrivileges)
s.router.RequireAuth("GET /me", s.getCallerProfile)
diff --git a/pkg/models/access_request.go b/pkg/models/access_request.go
index 925260ec2..591f6d50a 100644
--- a/pkg/models/access_request.go
+++ b/pkg/models/access_request.go
@@ -32,6 +32,9 @@ const (
AccessRequestEventCreated = "CREATED"
AccessRequestEventApproved = "APPROVED"
AccessRequestEventDenied = "DENIED"
+ // AccessRequestEventFailed records an approval attempt whose provisioning
+ // failed; the request itself stays PENDING so it can be re-approved.
+ AccessRequestEventFailed = "FAILED"
)
// AccessEvent maps an event code to the allocation and organization an
diff --git a/pkg/models/privilege.go b/pkg/models/privilege.go
index 1ef25f988..58f8ddd8a 100644
--- a/pkg/models/privilege.go
+++ b/pkg/models/privilege.go
@@ -40,6 +40,9 @@ const (
TracesRead PrivilegeKey = "core:traces:read"
PrivilegesGrant PrivilegeKey = "core:privileges:grant"
RolesManage PrivilegeKey = "core:roles:manage"
+
+ AccessRequestsRead PrivilegeKey = "core:access-requests:read"
+ AccessRequestsWrite PrivilegeKey = "core:access-requests:write"
)
var (
@@ -62,6 +65,8 @@ func init() {
TracesRead,
PrivilegesGrant,
RolesManage,
+ AccessRequestsRead,
+ AccessRequestsWrite,
)
}
diff --git a/pkg/service/access_request.go b/pkg/service/access_request.go
new file mode 100644
index 000000000..2ee21bbe6
--- /dev/null
+++ b/pkg/service/access_request.go
@@ -0,0 +1,329 @@
+// 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 service
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/apache/airavata-custos/internal/store"
+ "github.com/apache/airavata-custos/pkg/models"
+)
+
+// defaultAccessRequestLifetime is how long an approved trial account lasts
+// when the approver does not pick an explicit expiry.
+const defaultAccessRequestLifetime = 30 * 24 * time.Hour
+
+// GetAccessEvent retrieves an access event by its code. Returns ErrNotFound
+// when no event matches.
+func (s *Service) GetAccessEvent(ctx context.Context, code string) (*models.AccessEvent, error) {
+ if code == "" {
+ return nil, fmt.Errorf("%w: event code is required", ErrInvalidInput)
+ }
+ ev, err := s.accessEvents.FindByCode(ctx, code)
+ if err != nil {
+ return nil, fmt.Errorf("get access event: %w", err)
+ }
+ if ev == nil {
+ return nil, fmt.Errorf("%w: access event %q", ErrNotFound, code)
+ }
+ return ev, nil
+}
+
+// CreateAccessRequest records a self-service access request identified by a
+// verified token subject. The event code must resolve and only one PENDING
+// request per subject is allowed. A "CREATED" event is appended to the
+// request's audit log in the same transaction.
+func (s *Service) CreateAccessRequest(ctx context.Context, req *models.AccessRequest) (*models.AccessRequest, error) {
+ if req == nil {
+ return nil, fmt.Errorf("%w: access request is nil", ErrInvalidInput)
+ }
+ if req.OIDCSub == "" {
+ return nil, fmt.Errorf("%w: oidc_sub is required", ErrInvalidInput)
+ }
+ if req.Email == "" {
+ return nil, fmt.Errorf("%w: email is required", ErrInvalidInput)
+ }
+ if req.Institution == "" {
+ return nil, fmt.Errorf("%w: institution is required", ErrInvalidInput)
+ }
+ if req.EventCode == "" {
+ return nil, fmt.Errorf("%w: event_code is required", ErrInvalidInput)
+ }
+
+ if ev, err := s.accessEvents.FindByCode(ctx, req.EventCode); err != nil {
+ return nil, fmt.Errorf("lookup access event: %w", err)
+ } else if ev == nil {
+ return nil, fmt.Errorf("%w: access event %q", ErrNotFound, req.EventCode)
+ }
+
+ if pending, err := s.accessRequests.HasPendingBySub(ctx, req.OIDCSub); err != nil {
+ return nil, fmt.Errorf("check pending access request: %w", err)
+ } else if pending {
+ return nil, fmt.Errorf("%w: a pending access request already exists for this subject", ErrAlreadyExists)
+ }
+
+ if req.ID == "" {
+ req.ID = newID()
+ }
+ if req.Status == "" {
+ req.Status = models.AccessRequestPending
+ }
+ if req.Timestamp.IsZero() {
+ req.Timestamp = nowUTC()
+ }
+
+ if err := s.inTx(ctx, func(tx *sql.Tx) error {
+ if err := s.accessRequests.Create(ctx, tx, req); err != nil {
+ return err
+ }
+ return s.accessRequestEvents.Create(ctx, tx, &models.AccessRequestEvent{
+ ID: newID(),
+ AccessRequestID: req.ID,
+ EventType: models.AccessRequestEventCreated,
+ Description: "Access request created",
+ Timestamp: req.Timestamp,
+ })
+ }); err != nil {
+ return nil, fmt.Errorf("create access request: %w", err)
+ }
+ return req, nil
+}
+
+// GetLatestAccessRequestBySub returns the newest access request submitted by
+// the given token subject. Returns ErrNotFound when the subject has none.
+func (s *Service) GetLatestAccessRequestBySub(ctx context.Context, sub string) (*models.AccessRequest, error) {
+ if sub == "" {
+ return nil, fmt.Errorf("%w: oidc_sub is required", ErrInvalidInput)
+ }
+ req, err := s.accessRequests.FindLatestBySub(ctx, sub)
+ if err != nil {
+ return nil, fmt.Errorf("get latest access request: %w", err)
+ }
+ if req == nil {
+ return nil, ErrNotFound
+ }
+ return req, nil
+}
+
+// ListAccessRequests returns access requests filtered by the supplied criteria.
+func (s *Service) ListAccessRequests(ctx context.Context, f store.AccessRequestListFilter) ([]models.AccessRequest, error) {
+ rows, err := s.accessRequests.List(ctx, f)
+ if err != nil {
+ return nil, fmt.Errorf("list access requests: %w", err)
+ }
+ return rows, nil
+}
+
+// ApproveAccessRequest transitions a PENDING request to APPROVED. Account
+// provisioning runs first; if any step fails the request stays PENDING with
+// a FAILED event row and the error is returned, so a later re-approve can
+// pick up where the partial run stopped. On success the status transition
+// and an APPROVED event row commit in one transaction.
+func (s *Service) ApproveAccessRequest(ctx context.Context, id, approverID string, expiresAt *time.Time) (*models.AccessRequest, error) {
+ if id == "" {
+ return nil, fmt.Errorf("%w: access request id is required", ErrInvalidInput)
+ }
+ if approverID == "" {
+ return nil, fmt.Errorf("%w: approver id is required", ErrInvalidInput)
+ }
+ req, err := s.accessRequests.FindByID(ctx, id)
+ if err != nil {
+ return nil, fmt.Errorf("lookup access request: %w", err)
+ }
+ if req == nil {
+ return nil, ErrNotFound
+ }
+ if req.Status != models.AccessRequestPending {
+ return nil, fmt.Errorf("%w: access request is %s, only PENDING can be approved", ErrInvalidInput, req.Status)
+ }
+
+ exp := nowUTC().Add(defaultAccessRequestLifetime)
+ if expiresAt != nil {
+ exp = expiresAt.UTC()
+ }
+
+ user, err := s.provisionAccessRequest(ctx, req, exp)
+ if err != nil {
+ s.appendAccessRequestEvent(ctx, req.ID, models.AccessRequestEventFailed, fmt.Sprintf("Approval failed: %v", err))
+ return nil, err
+ }
+
+ req.Status = models.AccessRequestApproved
+ req.ApproverID = approverID
+ req.ExpiresAt = &exp
+ req.CreatedUserID = user.ID
+ if err := s.inTx(ctx, func(tx *sql.Tx) error {
+ if err := s.accessRequests.Update(ctx, tx, req); err != nil {
+ return err
+ }
+ return s.accessRequestEvents.Create(ctx, tx, &models.AccessRequestEvent{
+ ID: newID(),
+ AccessRequestID: req.ID,
+ EventType: models.AccessRequestEventApproved,
+ Description: "Access request approved",
+ Timestamp: nowUTC(),
+ })
+ }); err != nil {
+ return nil, fmt.Errorf("approve access request: %w", err)
+ }
+ return req, nil
+}
+
+// provisionAccessRequest creates the user, identity binding, cluster user,
+// and allocation membership for an approval. Every step tolerates leftovers
+// from an earlier partial run so a failed approval can be retried.
+func (s *Service) provisionAccessRequest(ctx context.Context, req *models.AccessRequest, expiresAt time.Time) (*models.User, error) {
+ ev, err := s.accessEvents.FindByCode(ctx, req.EventCode)
+ if err != nil {
+ return nil, fmt.Errorf("lookup access event: %w", err)
+ }
+ if ev == nil {
+ return nil, fmt.Errorf("%w: access event %q", ErrNotFound, req.EventCode)
+ }
+ alloc, err := s.allocs.FindByID(ctx, ev.ComputeAllocationID)
+ if err != nil {
+ return nil, fmt.Errorf("lookup compute allocation: %w", err)
+ }
+ if alloc == nil {
+ return nil, fmt.Errorf("%w: compute allocation %q not found", ErrInvalidInput, ev.ComputeAllocationID)
+ }
+ if alloc.Status != models.ACTIVE {
+ return nil, fmt.Errorf("%w: compute allocation %q is %s, must be ACTIVE", ErrInvalidInput, alloc.ID, alloc.Status)
+ }
+
+ first, last := splitName(req.Name)
+ user, err := s.CreateUser(ctx, &models.User{
+ OrganizationID: ev.OrganizationID,
+ FirstName: first,
+ LastName: last,
+ Email: req.Email,
+ Status: models.UserActive,
+ })
+ if errors.Is(err, ErrAlreadyExists) {
+ user, err = s.users.FindByEmail(ctx, req.Email)
+ if err == nil && user == nil {
+ err = fmt.Errorf("user with email %q vanished during approval", req.Email)
+ }
+ }
+ if err != nil {
+ return nil, fmt.Errorf("create user: %w", err)
+ }
+
+ if _, err := s.CreateUserIdentity(ctx, &models.UserIdentity{
+ UserID: user.ID,
+ Source: identitySourceOIDC,
+ ExternalID: req.OIDCSub,
+ OIDCSub: req.OIDCSub,
+ Email: req.Email,
+ }); err != nil && !errors.Is(err, ErrAlreadyExists) {
+ return nil, fmt.Errorf("create user identity: %w", err)
+ }
+
+ // Check the pair first: the allocator treats a pair duplicate as a
+ // username collision and would burn its retry budget on the same pair.
+ if existing, err := s.clusterUsers.FindByPair(ctx, alloc.ComputeClusterID, user.ID); err != nil {
+ return nil, fmt.Errorf("lookup compute cluster user: %w", err)
+ } else if existing == nil {
+ if _, err := s.AllocateComputeClusterUser(ctx, user, alloc.ComputeClusterID); err != nil {
+ return nil, fmt.Errorf("allocate compute cluster user: %w", err)
+ }
+ }
+
+ if _, err := s.CreateComputeAllocationMembership(ctx, &models.ComputeAllocationMembership{
+ ComputeAllocationID: alloc.ID,
+ UserID: user.ID,
+ StartTime: nowUTC(),
+ EndTime: expiresAt,
+ MembershipStatus: models.ACTIVE,
+ }); err != nil && !errors.Is(err, ErrAlreadyExists) {
+ return nil, fmt.Errorf("create compute allocation membership: %w", err)
+ }
+ return user, nil
+}
+
+// DenyAccessRequest transitions a PENDING request to DENIED, recording the
+// reason and the approver. A DENIED event row is appended in-tx.
+func (s *Service) DenyAccessRequest(ctx context.Context, id, approverID, denyReason string) (*models.AccessRequest, error) {
+ if id == "" {
+ return nil, fmt.Errorf("%w: access request id is required", ErrInvalidInput)
+ }
+ if approverID == "" {
+ return nil, fmt.Errorf("%w: approver id is required", ErrInvalidInput)
+ }
+ req, err := s.accessRequests.FindByID(ctx, id)
+ if err != nil {
+ return nil, fmt.Errorf("lookup access request: %w", err)
+ }
+ if req == nil {
+ return nil, ErrNotFound
+ }
+ if req.Status != models.AccessRequestPending {
+ return nil, fmt.Errorf("%w: access request is %s, only PENDING can be denied", ErrInvalidInput, req.Status)
+ }
+
+ req.Status = models.AccessRequestDenied
+ req.ApproverID = approverID
+ req.DenyReason = denyReason
+ if err := s.inTx(ctx, func(tx *sql.Tx) error {
+ if err := s.accessRequests.Update(ctx, tx, req); err != nil {
+ return err
+ }
+ return s.accessRequestEvents.Create(ctx, tx, &models.AccessRequestEvent{
+ ID: newID(),
+ AccessRequestID: req.ID,
+ EventType: models.AccessRequestEventDenied,
+ Description: "Access request denied",
+ Timestamp: nowUTC(),
+ })
+ }); err != nil {
+ return nil, fmt.Errorf("deny access request: %w", err)
+ }
+ return req, nil
+}
+
+// appendAccessRequestEvent writes a best-effort audit row outside the main
+// flow; a failure here must not mask the error being reported.
+func (s *Service) appendAccessRequestEvent(ctx context.Context, requestID, eventType, description string) {
+ _ = s.inTx(ctx, func(tx *sql.Tx) error {
+ return s.accessRequestEvents.Create(ctx, tx, &models.AccessRequestEvent{
+ ID: newID(),
+ AccessRequestID: requestID,
+ EventType: eventType,
+ Description: description,
+ Timestamp: nowUTC(),
+ })
+ })
+}
+
+// splitName maps a display name onto first/last: first token, then the rest.
+func splitName(full string) (first, last string) {
+ parts := strings.Fields(full)
+ switch len(parts) {
+ case 0:
+ return "", ""
+ case 1:
+ return parts[0], ""
+ default:
+ return parts[0], strings.Join(parts[1:], " ")
+ }
+}
diff --git a/pkg/service/access_request_integration_test.go b/pkg/service/access_request_integration_test.go
new file mode 100644
index 000000000..34ebc629c
--- /dev/null
+++ b/pkg/service/access_request_integration_test.go
@@ -0,0 +1,387 @@
+//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 service
+
+import (
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+ "github.com/jmoiron/sqlx"
+
+ "github.com/apache/airavata-custos/internal/store"
+ "github.com/apache/airavata-custos/pkg/models"
+ "github.com/apache/airavata-custos/pkg/posix"
+)
+
+type accessTestEnv struct {
+ db *sqlx.DB
+ svc *Service
+ orgID string
+ adminID string
+ clusterID string
+ allocID string
+ code string
+}
+
+// setupAccessEnv builds the org/admin/cluster/project/allocation chain an
+// access event depends on and inserts the access_events row.
+func setupAccessEnv(t *testing.T) *accessTestEnv {
+ t.Helper()
+ database := setupTestDB(t)
+ // setupTestDB truncates only the core auth tables; clear the access and
+ // allocation chains this suite touches.
+ if _, err := database.Exec("SET FOREIGN_KEY_CHECKS = 0"); err != nil {
+ t.Fatalf("disable FK: %v", err)
+ }
+ for _, tbl := range []string{
+ "access_request_events", "access_requests", "access_events",
+ "compute_allocation_memberships", "compute_cluster_users",
+ "compute_allocations", "projects", "compute_clusters",
+ } {
+ if _, err := database.Exec("TRUNCATE TABLE " + tbl); err != nil {
+ t.Fatalf("truncate %s: %v", tbl, err)
+ }
+ }
+ if _, err := database.Exec("SET FOREIGN_KEY_CHECKS = 1"); err != nil {
+ t.Fatalf("re-enable FK: %v", err)
+ }
+
+ svc := newTestService(database)
+ org, err := svc.CreateOrganization(ctx(), &models.Organization{
+ OriginatedID: uuid.NewString(),
+ Name: "access-org-" + uuid.NewString()[:8],
+ })
+ if err != nil {
+ t.Fatalf("create org: %v", err)
+ }
+ admin, err := svc.CreateUser(ctx(), &models.User{
+ OrganizationID: org.ID,
+ FirstName: "Admin",
+ LastName: "Approver",
+ Email: "admin-" + uuid.NewString()[:8] + "@example.invalid",
+ Status: models.UserActive,
+ })
+ if err != nil {
+ t.Fatalf("create admin: %v", err)
+ }
+ cluster, err := svc.CreateComputeCluster(ctx(), &models.ComputeCluster{
+ Name: "access-cluster-" + uuid.NewString()[:8],
+ })
+ if err != nil {
+ t.Fatalf("create cluster: %v", err)
+ }
+ proj, err := svc.CreateProject(ctx(), &models.Project{
+ OriginatedID: uuid.NewString(),
+ Title: "Access Test Project",
+ Origination: "TEST",
+ ProjectPIID: admin.ID,
+ })
+ if err != nil {
+ t.Fatalf("create project: %v", err)
+ }
+ alloc, err := svc.CreateComputeAllocation(ctx(), &models.ComputeAllocation{
+ ProjectID: proj.ID,
+ Name: "access-alloc-" + uuid.NewString()[:8],
+ ComputeClusterID: cluster.ID,
+ InitialSUAmount: 1000,
+ StartTime: nowUTC(),
+ EndTime: nowUTC().Add(365 * 24 * time.Hour),
+ })
+ if err != nil {
+ t.Fatalf("create allocation: %v", err)
+ }
+ code := "EVT-" + uuid.NewString()[:8]
+ if _, err := database.Exec(
+ "INSERT INTO access_events (code, compute_allocation_id, organization_id) VALUES (?, ?, ?)",
+ code, alloc.ID, org.ID,
+ ); err != nil {
+ t.Fatalf("seed access event: %v", err)
+ }
+ return &accessTestEnv{
+ db: database, svc: svc,
+ orgID: org.ID, adminID: admin.ID,
+ clusterID: cluster.ID, allocID: alloc.ID, code: code,
+ }
+}
+
+func newAccessRequestFor(env *accessTestEnv, sub string) *models.AccessRequest {
+ return &models.AccessRequest{
+ OIDCSub: sub,
+ Email: sub + "@example.invalid",
+ Name: "Trial Person",
+ Institution: "Test University",
+ EventCode: env.code,
+ Reason: "tutorial",
+ }
+}
+
+func requestEventTypes(t *testing.T, env *accessTestEnv, requestID string) []string {
+ t.Helper()
+ var types []string
+ if err := env.db.Select(&types,
+ "SELECT event_type FROM access_request_events WHERE access_request_id = ? ORDER BY timestamp, event_type",
+ requestID,
+ ); err != nil {
+ t.Fatalf("list request events: %v", err)
+ }
+ return types
+}
+
+func TestCreateAccessRequest_Guards(t *testing.T) {
+ env := setupAccessEnv(t)
+ sub := "sub-" + uuid.NewString()[:8]
+
+ bad := newAccessRequestFor(env, sub)
+ bad.EventCode = "NO-SUCH-EVENT"
+ if _, err := env.svc.CreateAccessRequest(ctx(), bad); !errors.Is(err, ErrNotFound) {
+ t.Fatalf("unknown event: got %v, want ErrNotFound", err)
+ }
+
+ created, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, sub))
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+ if created.Status != models.AccessRequestPending {
+ t.Errorf("status = %s, want PENDING", created.Status)
+ }
+ if got := requestEventTypes(t, env, created.ID); len(got) != 1 || got[0] != models.AccessRequestEventCreated {
+ t.Errorf("event rows = %v, want [CREATED]", got)
+ }
+
+ if _, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, sub)); !errors.Is(err, ErrAlreadyExists) {
+ t.Fatalf("duplicate pending: got %v, want ErrAlreadyExists", err)
+ }
+}
+
+func TestApproveAccessRequest_HappyPath(t *testing.T) {
+ env := setupAccessEnv(t)
+ sub := "sub-" + uuid.NewString()[:8]
+ req, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, sub))
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+
+ exp := nowUTC().Add(48 * time.Hour).Truncate(time.Second)
+ approved, err := env.svc.ApproveAccessRequest(ctx(), req.ID, env.adminID, &exp)
+ if err != nil {
+ t.Fatalf("approve: %v", err)
+ }
+ if approved.Status != models.AccessRequestApproved {
+ t.Errorf("status = %s, want APPROVED", approved.Status)
+ }
+ if approved.ApproverID != env.adminID {
+ t.Errorf("approver = %q, want %q", approved.ApproverID, env.adminID)
+ }
+ if approved.CreatedUserID == "" {
+ t.Fatal("created_user_id is empty")
+ }
+
+ user, err := env.svc.GetUser(ctx(), approved.CreatedUserID)
+ if err != nil {
+ t.Fatalf("get created user: %v", err)
+ }
+ if user.Status != models.UserActive {
+ t.Errorf("user status = %s, want ACTIVE", user.Status)
+ }
+ if user.OrganizationID != env.orgID {
+ t.Errorf("user org = %q, want %q", user.OrganizationID, env.orgID)
+ }
+
+ ident, err := env.svc.GetUserIdentityBySourceAndExternalID(ctx(), identitySourceOIDC, sub)
+ if err != nil {
+ t.Fatalf("get identity: %v", err)
+ }
+ if ident.UserID != user.ID {
+ t.Errorf("identity user = %q, want %q", ident.UserID, user.ID)
+ }
+
+ ccu, err := env.svc.GetComputeClusterUserByPair(ctx(), env.clusterID, user.ID)
+ if err != nil {
+ t.Fatalf("get cluster user: %v", err)
+ }
+ if !strings.HasPrefix(ccu.LocalUsername, posix.Prefix()+"-") {
+ t.Errorf("local username %q lacks prefix %q", ccu.LocalUsername, posix.Prefix()+"-")
+ }
+
+ var m struct {
+ Status string `db:"membership_status"`
+ EndTime time.Time `db:"end_time"`
+ }
+ if err := env.db.Get(&m,
+ "SELECT membership_status, end_time FROM compute_allocation_memberships WHERE compute_allocation_id = ? AND user_id = ?",
+ env.allocID, user.ID,
+ ); err != nil {
+ t.Fatalf("get membership: %v", err)
+ }
+ if m.Status != string(models.ACTIVE) {
+ t.Errorf("membership status = %s, want ACTIVE", m.Status)
+ }
+ if d := m.EndTime.UTC().Sub(exp); d > 2*time.Second || d < -2*time.Second {
+ t.Errorf("membership end_time = %v, want %v", m.EndTime.UTC(), exp)
+ }
+
+ if got := requestEventTypes(t, env, req.ID); len(got) != 2 || got[0] != models.AccessRequestEventCreated || got[1] != models.AccessRequestEventApproved {
+ t.Errorf("event rows = %v, want [CREATED APPROVED]", got)
+ }
+
+ latest, err := env.svc.GetLatestAccessRequestBySub(ctx(), sub)
+ if err != nil {
+ t.Fatalf("latest by sub: %v", err)
+ }
+ if latest.Status != models.AccessRequestApproved {
+ t.Errorf("latest status = %s, want APPROVED", latest.Status)
+ }
+}
+
+func TestApproveAccessRequest_ReusesExistingUser(t *testing.T) {
+ env := setupAccessEnv(t)
+ sub := "sub-" + uuid.NewString()[:8]
+ req, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, sub))
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+ pre, err := env.svc.CreateUser(ctx(), &models.User{
+ OrganizationID: env.orgID,
+ FirstName: "Trial",
+ LastName: "Person",
+ Email: req.Email,
+ Status: models.UserActive,
+ })
+ if err != nil {
+ t.Fatalf("pre-create user: %v", err)
+ }
+
+ approved, err := env.svc.ApproveAccessRequest(ctx(), req.ID, env.adminID, nil)
+ if err != nil {
+ t.Fatalf("approve with existing user: %v", err)
+ }
+ if approved.CreatedUserID != pre.ID {
+ t.Errorf("created_user_id = %q, want reused %q", approved.CreatedUserID, pre.ID)
+ }
+ if approved.ExpiresAt == nil {
+ t.Fatal("expires_at not defaulted")
+ }
+ want := nowUTC().Add(defaultAccessRequestLifetime)
+ if d := approved.ExpiresAt.Sub(want); d > time.Hour || d < -time.Hour {
+ t.Errorf("default expires_at = %v, want about %v", approved.ExpiresAt, want)
+ }
+ if _, err := env.svc.GetComputeClusterUserByPair(ctx(), env.clusterID, pre.ID); err != nil {
+ t.Errorf("cluster user for reused user: %v", err)
+ }
+}
+
+func TestApproveAccessRequest_FailureLeavesPending(t *testing.T) {
+ env := setupAccessEnv(t)
+ sub := "sub-" + uuid.NewString()[:8]
+ req, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, sub))
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+
+ if _, err := env.db.Exec("UPDATE compute_allocations SET status = 'INACTIVE' WHERE id = ?", env.allocID); err != nil {
+ t.Fatalf("deactivate allocation: %v", err)
+ }
+ if _, err := env.svc.ApproveAccessRequest(ctx(), req.ID, env.adminID, nil); !errors.Is(err, ErrInvalidInput) {
+ t.Fatalf("approve inactive allocation: got %v, want ErrInvalidInput", err)
+ }
+
+ latest, err := env.svc.GetLatestAccessRequestBySub(ctx(), sub)
+ if err != nil {
+ t.Fatalf("latest by sub: %v", err)
+ }
+ if latest.Status != models.AccessRequestPending {
+ t.Errorf("status after failed approve = %s, want PENDING", latest.Status)
+ }
+ got := requestEventTypes(t, env, req.ID)
+ if len(got) != 2 || got[1] != models.AccessRequestEventFailed {
+ t.Errorf("event rows = %v, want [CREATED FAILED]", got)
+ }
+
+ if _, err := env.db.Exec("UPDATE compute_allocations SET status = 'ACTIVE' WHERE id = ?", env.allocID); err != nil {
+ t.Fatalf("reactivate allocation: %v", err)
+ }
+ if _, err := env.svc.ApproveAccessRequest(ctx(), req.ID, env.adminID, nil); err != nil {
+ t.Fatalf("re-approve after failure: %v", err)
+ }
+}
+
+func TestDenyAccessRequest(t *testing.T) {
+ env := setupAccessEnv(t)
+ sub := "sub-" + uuid.NewString()[:8]
+ req, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, sub))
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+
+ denied, err := env.svc.DenyAccessRequest(ctx(), req.ID, env.adminID, "not eligible")
+ if err != nil {
+ t.Fatalf("deny: %v", err)
+ }
+ if denied.Status != models.AccessRequestDenied {
+ t.Errorf("status = %s, want DENIED", denied.Status)
+ }
+ if denied.DenyReason != "not eligible" {
+ t.Errorf("deny_reason = %q", denied.DenyReason)
+ }
+ if got := requestEventTypes(t, env, req.ID); len(got) != 2 || got[1] != models.AccessRequestEventDenied {
+ t.Errorf("event rows = %v, want [CREATED DENIED]", got)
+ }
+
+ if _, err := env.svc.ApproveAccessRequest(ctx(), req.ID, env.adminID, nil); !errors.Is(err, ErrInvalidInput) {
+ t.Errorf("approve denied request: got %v, want ErrInvalidInput", err)
+ }
+ if _, err := env.svc.DenyAccessRequest(ctx(), req.ID, env.adminID, "again"); !errors.Is(err, ErrInvalidInput) {
+ t.Errorf("deny denied request: got %v, want ErrInvalidInput", err)
+ }
+}
+
+func TestListAccessRequests_Filter(t *testing.T) {
+ env := setupAccessEnv(t)
+ subA := "sub-" + uuid.NewString()[:8]
+ subB := "sub-" + uuid.NewString()[:8]
+ reqA, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, subA))
+ if err != nil {
+ t.Fatalf("create A: %v", err)
+ }
+ if _, err := env.svc.CreateAccessRequest(ctx(), newAccessRequestFor(env, subB)); err != nil {
+ t.Fatalf("create B: %v", err)
+ }
+ if _, err := env.svc.DenyAccessRequest(ctx(), reqA.ID, env.adminID, "no"); err != nil {
+ t.Fatalf("deny A: %v", err)
+ }
+
+ pending, err := env.svc.ListAccessRequests(ctx(), store.AccessRequestListFilter{Status: string(models.AccessRequestPending), EventCode: env.code})
+ if err != nil {
+ t.Fatalf("list pending: %v", err)
+ }
+ if len(pending) != 1 || pending[0].OIDCSub != subB {
+ t.Errorf("pending list = %+v, want only sub %s", pending, subB)
+ }
+ all, err := env.svc.ListAccessRequests(ctx(), store.AccessRequestListFilter{EventCode: env.code})
+ if err != nil {
+ t.Fatalf("list all: %v", err)
+ }
+ if len(all) != 2 {
+ t.Errorf("all list len = %d, want 2", len(all))
+ }
+}