// Package admin exposes the admin HTTP API for cluster management and audit // log access. All routes are protected by middleware.AdminAuth. package admin import ( "encoding/json" "errors" "net/http" "strconv" "github.com/gin-gonic/gin" "spark-mcp-go/internal/audit" "spark-mcp-go/internal/cluster" "spark-mcp-go/internal/middleware" "spark-mcp-go/internal/storage" ) // Mount attaches the /admin sub-router to r, protecting every route with // bearer-token admin authentication. func Mount(r *gin.Engine, repo *storage.ClusterRepo, auditRepo *audit.Repo, adminTokens []string) { // HTML page is public — its own modal prompts for the token. // All other /admin/* endpoints (API + OpenAPI docs) still require auth. gPublic := r.Group("/admin") gPublic.GET("", webHandler) gPublic.GET("/", webHandler) g := r.Group("/admin", middleware.AdminAuth(adminTokens)) g.GET("/clusters", listClusters(repo)) g.POST("/clusters", createCluster(repo, auditRepo)) g.GET("/clusters/:id", getCluster(repo)) g.PUT("/clusters/:id", updateCluster(repo, auditRepo)) g.DELETE("/clusters/:id", deleteCluster(repo, auditRepo)) g.GET("/audit", listAudit(auditRepo)) g.GET("/docs", DocsHandler) g.GET("/docs/spec", OpenAPISpecHandler) } func listClusters(repo *storage.ClusterRepo) gin.HandlerFunc { return func(c *gin.Context) { clusters, err := repo.List(c.Request.Context()) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, clusters) } } func createCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.HandlerFunc { return func(c *gin.Context) { var cl cluster.Cluster if err := c.ShouldBindJSON(&cl); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } if err := validateClusterCreate(&cl); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } ctx := c.Request.Context() if err := repo.Create(ctx, &cl); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } details, err := audit.MarshalDetails(clusterFields(&cl)) if err == nil { _ = auditRepo.Insert(ctx, &audit.Entry{ Actor: actor(c), Action: audit.ActionClusterCreate, ClusterID: cl.ID, Details: details, }) } c.JSON(http.StatusCreated, &cl) } } func getCluster(repo *storage.ClusterRepo) gin.HandlerFunc { return func(c *gin.Context) { cl, err := repo.Get(c.Request.Context(), c.Param("id")) if err != nil { if errors.Is(err, storage.ErrNotFound) { c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"}) return } c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, cl) } } func updateCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.HandlerFunc { return func(c *gin.Context) { id := c.Param("id") ctx := c.Request.Context() old, err := repo.Get(ctx, id) if err != nil { if errors.Is(err, storage.ErrNotFound) { c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"}) return } c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } // Parse the body as a generic map first so we can tell which fields // the client actually sent. A struct + ShouldBindJSON can't tell // "absent" from "zero value" for bools, ints, and empty strings — // which silently flips true→false and 1→0 on PATCH-style updates. rawBody := make(map[string]json.RawMessage) if err := c.ShouldBindJSON(&rawBody); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } merged := *old merged.ID = id // Strings: absent = keep old; empty string = keep old (no way to // distinguish "set to empty" from "not set" in JSON; treat them // the same — clients should send the new value or omit). if v, ok := rawBody["name"]; ok { merged.Name = decodeString(v) } if v, ok := rawBody["rm_url"]; ok { merged.RMURL = decodeString(v) } if v, ok := rawBody["shs_url"]; ok { merged.SHSURL = decodeString(v) } if v, ok := rawBody["spark_submit_execute_bin"]; ok { merged.SparkSubmitExecuteBin = decodeString(v) } if v, ok := rawBody["auth_type"]; ok { merged.AuthType = cluster.AuthType(decodeString(v)) } if v, ok := rawBody["auth_username"]; ok { merged.AuthUsername = decodeString(v) } if v, ok := rawBody["auth_password"]; ok { // Field is json:"-" so it never round-trips through admin JSON, // but a future dedicated password endpoint could set it here. merged.AuthPassword = decodeString(v) } if v, ok := rawBody["ssl_ca_bundle"]; ok { merged.SSLCABundle = decodeString(v) } // Ints. if v, ok := rawBody["rate_limit_per_min"]; ok { merged.RateLimitPerMin = decodeInt(v) } // Bools. if v, ok := rawBody["is_active"]; ok { merged.IsActive = decodeBool(v) } if v, ok := rawBody["ssl_verify"]; ok { merged.SSLVerify = decodeBool(v) } // Slices: null = keep old; [] = replace with empty. if v, ok := rawBody["url_allowlist"]; ok { merged.URLAllowlist = decodeStringSlice(v) } if v, ok := rawBody["default_submit_args"]; ok { merged.DefaultSubmitArgs = decodeStringSlice(v) } // Re-validate after merging so a missing auth_type on a fresh // cluster (defaulted by DB to "none") doesn't get clobbered. if err := validateClusterUpdate(&merged); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) return } if err := repo.Update(ctx, &merged); err != nil { if errors.Is(err, storage.ErrNotFound) { c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"}) return } c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } details, err := audit.MarshalDetails(map[string]any{ "before": clusterFields(old), "after": clusterFields(&merged), }) if err == nil { _ = auditRepo.Insert(ctx, &audit.Entry{ Actor: actor(c), Action: audit.ActionClusterUpdate, ClusterID: merged.ID, Details: details, }) } c.JSON(http.StatusOK, &merged) } } func decodeString(raw json.RawMessage) string { var s string _ = json.Unmarshal(raw, &s) return s } func decodeInt(raw json.RawMessage) int { var n int _ = json.Unmarshal(raw, &n) return n } func decodeBool(raw json.RawMessage) bool { var b bool _ = json.Unmarshal(raw, &b) return b } func decodeStringSlice(raw json.RawMessage) []string { // Handle both "key": null (clear to nil — caller treats as keep-old // since they would have skipped this field) and "key": [...] var s []string _ = json.Unmarshal(raw, &s) return s } func deleteCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.HandlerFunc { return func(c *gin.Context) { id := c.Param("id") ctx := c.Request.Context() old, err := repo.Get(ctx, id) if err != nil { if errors.Is(err, storage.ErrNotFound) { c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"}) return } c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } if err := repo.Delete(ctx, id); err != nil { if errors.Is(err, storage.ErrNotFound) { c.JSON(http.StatusNotFound, gin.H{"error": "cluster not found"}) return } c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } details, err := audit.MarshalDetails(map[string]any{ "id": id, "name": old.Name, }) if err == nil { _ = auditRepo.Insert(ctx, &audit.Entry{ Actor: actor(c), Action: audit.ActionClusterDelete, ClusterID: id, Details: details, }) } c.Status(http.StatusNoContent) } } func listAudit(auditRepo *audit.Repo) gin.HandlerFunc { return func(c *gin.Context) { limitStr := c.DefaultQuery("limit", "100") limit, err := strconv.Atoi(limitStr) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid limit"}) return } entries, err := auditRepo.List(c.Request.Context(), limit) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } c.JSON(http.StatusOK, entries) } } func actor(c *gin.Context) string { tok, _ := c.Get("admin_token") s, _ := tok.(string) if len(s) > 8 { s = s[:8] } return "admin:" + s } func validateClusterCreate(c *cluster.Cluster) error { if c.ID == "" { return errors.New("id is required") } if c.Name == "" { return errors.New("name is required") } if c.RMURL == "" { return errors.New("rm_url is required") } if c.SHSURL == "" { return errors.New("shs_url is required") } if c.SparkSubmitExecuteBin == "" { return errors.New("spark_submit_execute_bin is required") } switch c.AuthType { case cluster.AuthNone, cluster.AuthSimple, cluster.AuthBasic: default: return errors.New("auth_type must be one of none, simple, basic") } if c.RateLimitPerMin < 0 { return errors.New("rate_limit_per_min must be >= 0") } return nil } // validateClusterUpdate is the lighter validation for PATCH semantics: only // fields that constrain cluster behavior are checked, since the existing row // already supplies the rest. func validateClusterUpdate(c *cluster.Cluster) error { switch c.AuthType { case cluster.AuthNone, cluster.AuthSimple, cluster.AuthBasic: // ok case "": // Should not happen — defaults to "none" in DB. default: return errors.New("auth_type must be one of none, simple, basic") } if c.RateLimitPerMin < 0 { return errors.New("rate_limit_per_min must be >= 0") } return nil } func clusterFields(c *cluster.Cluster) map[string]any { return map[string]any{ "id": c.ID, "name": c.Name, "rm_url": c.RMURL, "shs_url": c.SHSURL, "spark_submit_execute_bin": c.SparkSubmitExecuteBin, "is_active": c.IsActive, "auth_type": c.AuthType, "auth_username": c.AuthUsername, "ssl_verify": c.SSLVerify, "ssl_ca_bundle": c.SSLCABundle, "url_allowlist": c.URLAllowlist, "default_submit_args": c.DefaultSubmitArgs, "rate_limit_per_min": c.RateLimitPerMin, } }