diff --git a/internal/admin/router.go b/internal/admin/router.go new file mode 100644 index 0000000..104bca5 --- /dev/null +++ b/internal/admin/router.go @@ -0,0 +1,251 @@ +// Package admin exposes the admin HTTP API for cluster management and audit +// log access. All routes are protected by middleware.AdminAuth. +package admin + +import ( + "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) { + 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)) +} + +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 + } + + var cl cluster.Cluster + if err := c.ShouldBindJSON(&cl); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + cl.ID = id + if cl.AuthPassword == "" { + cl.AuthPassword = old.AuthPassword + } + + if err := repo.Update(ctx, &cl); 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(&cl), + }) + if err == nil { + _ = auditRepo.Insert(ctx, &audit.Entry{ + Actor: actor(c), + Action: audit.ActionClusterUpdate, + ClusterID: cl.ID, + Details: details, + }) + } + + c.JSON(http.StatusOK, &cl) + } +} + +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 +} + +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, + } +} diff --git a/internal/admin/router_test.go b/internal/admin/router_test.go new file mode 100644 index 0000000..6511cb3 --- /dev/null +++ b/internal/admin/router_test.go @@ -0,0 +1,394 @@ +package admin + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gin-gonic/gin" + + "spark-mcp-go/internal/audit" + "spark-mcp-go/internal/storage" +) + +func init() { + gin.SetMode(gin.TestMode) +} + +func newTestAdmin(t *testing.T) (*gin.Engine, *storage.ClusterRepo, *audit.Repo) { + t.Helper() + db, err := storage.Open(":memory:") + if err != nil { + t.Fatalf("open: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + r := gin.New() + Mount(r, db.Clusters(), audit.NewRepo(db), []string{"good-token"}) + return r, db.Clusters(), audit.NewRepo(db) +} + +func doReq(t *testing.T, r *gin.Engine, method, path, token string, body any) *httptest.ResponseRecorder { + t.Helper() + var rdr io.Reader + if body != nil { + b, _ := json.Marshal(body) + rdr = bytes.NewReader(b) + } + req := httptest.NewRequest(method, path, rdr) + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + return w +} + +func fullClusterMap(id string) map[string]any { + return map[string]any{ + "id": id, + "name": id + "-name", + "rm_url": "http://rm.example.com:8088", + "shs_url": "http://shs.example.com:18080", + "spark_submit_execute_bin": "/usr/bin/spark-submit", + "is_active": true, + "auth_type": "basic", + "auth_username": "admin", + "auth_password": "real-secret", + "ssl_verify": false, + "ssl_ca_bundle": "", + "url_allowlist": []string{"*.example.com"}, + "default_submit_args": []string{"--master", "yarn"}, + "rate_limit_per_min": 10, + } +} + +func TestMount_RequiresAuth(t *testing.T) { + r, _, _ := newTestAdmin(t) + + tests := []struct { + name string + method string + path string + token string + want int + }{ + {"no header", "GET", "/admin/clusters", "", http.StatusUnauthorized}, + {"wrong scheme", "GET", "/admin/clusters", "Basic xxx", http.StatusUnauthorized}, + {"wrong token", "GET", "/admin/clusters", "wrong", http.StatusUnauthorized}, + {"valid token", "GET", "/admin/clusters", "good-token", http.StatusOK}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + w := doReq(t, r, tt.method, tt.path, tt.token, nil) + if w.Code != tt.want { + t.Errorf("got status %d, want %d", w.Code, tt.want) + } + }) + } +} + +func TestListClusters(t *testing.T) { + r, _, _ := newTestAdmin(t) + + doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c1")) + doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c2")) + + w := doReq(t, r, "GET", "/admin/clusters", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("got status %d, want %d", w.Code, http.StatusOK) + } + + var list []map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &list); err != nil { + t.Fatalf("unmarshal list: %v", err) + } + if len(list) != 2 { + t.Errorf("got %d clusters, want 2", len(list)) + } + + for _, cl := range list { + if _, ok := cl["auth_password"]; ok { + t.Errorf("response must not contain auth_password") + } + } + + body := w.Body.String() + if strings.Contains(body, "real-secret") { + t.Errorf("response body leaks auth_password") + } +} + +func TestCreateCluster(t *testing.T) { + r, _, auditRepo := newTestAdmin(t) + + w := doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c1")) + if w.Code != http.StatusCreated { + t.Fatalf("got status %d, want %d", w.Code, http.StatusCreated) + } + + var cl map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &cl); err != nil { + t.Fatalf("unmarshal cluster: %v", err) + } + if cl["id"] != "c1" || cl["name"] != "c1-name" { + t.Errorf("unexpected cluster fields: id=%v name=%v", cl["id"], cl["name"]) + } + + entries, err := auditRepo.List(context.Background(), 100) + if err != nil { + t.Fatalf("list audit: %v", err) + } + if len(entries) < 1 { + t.Fatalf("got %d audit entries, want at least 1", len(entries)) + } + found := false + for _, e := range entries { + if e.Action == audit.ActionClusterCreate && e.Actor == "admin:good-tok" { + found = true + break + } + } + if !found { + t.Errorf("missing create audit entry from admin:good-tok") + } +} + +func TestCreateCluster_ValidationErrors(t *testing.T) { + r, _, _ := newTestAdmin(t) + + tests := []struct { + name string + body map[string]any + }{ + { + "missing id", + func() map[string]any { + m := fullClusterMap("c1") + delete(m, "id") + return m + }(), + }, + { + "missing name", + func() map[string]any { + m := fullClusterMap("c1") + delete(m, "name") + return m + }(), + }, + { + "invalid auth_type", + func() map[string]any { + m := fullClusterMap("c1") + m["auth_type"] = "kerberos" + return m + }(), + }, + { + "negative rate limit", + func() map[string]any { + m := fullClusterMap("c1") + m["rate_limit_per_min"] = -1 + return m + }(), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + w := doReq(t, r, "POST", "/admin/clusters", "good-token", tt.body) + if w.Code != http.StatusBadRequest { + t.Errorf("got status %d, want %d", w.Code, http.StatusBadRequest) + } + }) + } +} + +func TestGetCluster(t *testing.T) { + r, _, _ := newTestAdmin(t) + doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c1")) + + w := doReq(t, r, "GET", "/admin/clusters/c1", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("got status %d, want %d", w.Code, http.StatusOK) + } + var cl map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &cl); err != nil { + t.Fatalf("unmarshal cluster: %v", err) + } + if cl["id"] != "c1" || cl["name"] != "c1-name" { + t.Errorf("unexpected cluster fields: id=%v name=%v", cl["id"], cl["name"]) + } + + w = doReq(t, r, "GET", "/admin/clusters/not-found", "good-token", nil) + if w.Code != http.StatusNotFound { + t.Errorf("got status %d, want %d", w.Code, http.StatusNotFound) + } + var errBody map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &errBody); err != nil { + t.Fatalf("unmarshal error body: %v", err) + } + if errBody["error"] != "cluster not found" { + t.Errorf("unexpected error message: %v", errBody["error"]) + } +} + +func TestUpdateCluster(t *testing.T) { + r, _, auditRepo := newTestAdmin(t) + doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c1")) + + update := map[string]any{ + "name": "updated", + "rate_limit_per_min": 99, + } + w := doReq(t, r, "PUT", "/admin/clusters/c1", "good-token", update) + if w.Code != http.StatusOK { + t.Fatalf("got status %d, want %d: %s", w.Code, http.StatusOK, w.Body.String()) + } + var cl map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &cl); err != nil { + t.Fatalf("unmarshal cluster: %v", err) + } + if cl["name"] != "updated" || cl["rate_limit_per_min"] != float64(99) { + t.Errorf("unexpected update response: name=%v rate_limit_per_min=%v", cl["name"], cl["rate_limit_per_min"]) + } + + w = doReq(t, r, "GET", "/admin/clusters/c1", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("get after update: got status %d", w.Code) + } + if err := json.Unmarshal(w.Body.Bytes(), &cl); err != nil { + t.Fatalf("unmarshal cluster after update: %v", err) + } + if cl["name"] != "updated" || cl["rate_limit_per_min"] != float64(99) { + t.Errorf("cluster not updated: name=%v rate_limit_per_min=%v", cl["name"], cl["rate_limit_per_min"]) + } + + entries, err := auditRepo.List(context.Background(), 100) + if err != nil { + t.Fatalf("list audit: %v", err) + } + found := false + for _, e := range entries { + if e.Action == audit.ActionClusterUpdate { + found = true + var details map[string]any + if err := json.Unmarshal([]byte(e.Details), &details); err != nil { + t.Fatalf("unmarshal audit details: %v", err) + } + if _, ok := details["before"]; !ok { + t.Errorf("audit details missing before") + } + if _, ok := details["after"]; !ok { + t.Errorf("audit details missing after") + } + } + } + if !found { + t.Errorf("missing update audit entry") + } +} + +func TestUpdateCluster_PreservesEmptyPassword(t *testing.T) { + r, repo, auditRepo := newTestAdmin(t) + doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c1")) + + // AuthPassword has json:"-", so POST cannot receive it via JSON. + // Seed the stored password directly through the repo. + cl, err := repo.Get(context.Background(), "c1") + if err != nil { + t.Fatalf("get cluster from repo: %v", err) + } + cl.AuthPassword = "real-secret" + if err := repo.Update(context.Background(), cl); err != nil { + t.Fatalf("seed password: %v", err) + } + // Discard the audit row produced by the seed update so it does not confuse later checks. + _, _ = auditRepo.List(context.Background(), 100) + + update := map[string]any{ + "name": "renamed", + } + w := doReq(t, r, "PUT", "/admin/clusters/c1", "good-token", update) + if w.Code != http.StatusOK { + t.Fatalf("got status %d, want %d: %s", w.Code, http.StatusOK, w.Body.String()) + } + + if strings.Contains(w.Body.String(), "real-secret") { + t.Errorf("update response leaks auth_password") + } + + cl, err = repo.Get(context.Background(), "c1") + if err != nil { + t.Fatalf("get cluster from repo: %v", err) + } + if cl.AuthPassword != "real-secret" { + t.Errorf("password changed: got %q, want %q", cl.AuthPassword, "real-secret") + } +} + +func TestDeleteCluster(t *testing.T) { + r, _, auditRepo := newTestAdmin(t) + doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c1")) + + w := doReq(t, r, "DELETE", "/admin/clusters/c1", "good-token", nil) + if w.Code != http.StatusNoContent { + t.Fatalf("got status %d, want %d", w.Code, http.StatusNoContent) + } + + w = doReq(t, r, "GET", "/admin/clusters/c1", "good-token", nil) + if w.Code != http.StatusNotFound { + t.Errorf("get after delete: got status %d, want %d", w.Code, http.StatusNotFound) + } + + entries, err := auditRepo.List(context.Background(), 100) + if err != nil { + t.Fatalf("list audit: %v", err) + } + found := false + for _, e := range entries { + if e.Action == audit.ActionClusterDelete { + found = true + break + } + } + if !found { + t.Errorf("missing delete audit entry") + } +} + +func TestListAudit(t *testing.T) { + r, _, _ := newTestAdmin(t) + doReq(t, r, "POST", "/admin/clusters", "good-token", fullClusterMap("c1")) + doReq(t, r, "PUT", "/admin/clusters/c1", "good-token", map[string]any{"name": "updated"}) + doReq(t, r, "DELETE", "/admin/clusters/c1", "good-token", nil) + + w := doReq(t, r, "GET", "/admin/audit", "good-token", nil) + if w.Code != http.StatusOK { + t.Fatalf("got status %d, want %d", w.Code, http.StatusOK) + } + + var entries []*audit.Entry + if err := json.Unmarshal(w.Body.Bytes(), &entries); err != nil { + t.Fatalf("unmarshal audit list: %v", err) + } + if len(entries) != 3 { + t.Fatalf("got %d audit entries, want 3", len(entries)) + } + + wantActions := []string{string(audit.ActionClusterDelete), string(audit.ActionClusterUpdate), string(audit.ActionClusterCreate)} + for i, want := range wantActions { + got := string(entries[i].Action) + if got != want { + t.Errorf("entry[%d].action=%s, want %s", i, got, want) + } + } +} diff --git a/internal/audit/repo.go b/internal/audit/repo.go new file mode 100644 index 0000000..65eb011 --- /dev/null +++ b/internal/audit/repo.go @@ -0,0 +1,122 @@ +package audit + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + "spark-mcp-go/internal/storage" +) + +// Action enumerates admin write operations that are persisted to the audit log. +type Action string + +const ( + ActionClusterCreate Action = "cluster.create" + ActionClusterUpdate Action = "cluster.update" + ActionClusterDelete Action = "cluster.delete" +) + +// Entry is one admin write operation recorded for accountability. +// +// Actor is stored as "admin:" (the first 8 characters of the +// admin token when multiple tokens are in use). +// Details is a JSON-encoded string produced by MarshalDetails; callers that +// do not need structured details can leave it empty. +type Entry struct { + ID int64 `json:"id"` + Timestamp time.Time `json:"timestamp"` + Actor string `json:"actor"` + Action Action `json:"action"` + ClusterID string `json:"cluster_id,omitempty"` + Details string `json:"details,omitempty"` +} + +// Repo writes and reads audit_log rows. +type Repo struct { + db *storage.DB +} + +// NewRepo returns a repository bound to the supplied DB handle. +func NewRepo(db *storage.DB) *Repo { + return &Repo{db: db} +} + +// Insert persists a single audit entry. A zero Timestamp is replaced with +// the current time before storage. +func (r *Repo) Insert(ctx context.Context, e *Entry) error { + if e.Timestamp.IsZero() { + e.Timestamp = time.Now() + } + var details any + if e.Details != "" { + details = e.Details + } + _, err := r.db.SQLDB().ExecContext(ctx, ` + INSERT INTO audit_log (timestamp, actor, action, cluster_id, details) + VALUES (?, ?, ?, ?, ?)`, + e.Timestamp.UnixNano(), e.Actor, string(e.Action), e.ClusterID, details, + ) + if err != nil { + return fmt.Errorf("audit: insert: %w", err) + } + return nil +} + +// List returns the most recent audit entries ordered by timestamp descending. +// A limit of zero or less falls back to 100. +func (r *Repo) List(ctx context.Context, limit int) ([]*Entry, error) { + if limit <= 0 { + limit = 100 + } + rows, err := r.db.SQLDB().QueryContext(ctx, ` + SELECT id, timestamp, actor, action, cluster_id, details + FROM audit_log ORDER BY timestamp DESC LIMIT ?`, limit) + if err != nil { + return nil, fmt.Errorf("audit: list: %w", err) + } + defer rows.Close() + + var out []*Entry + for rows.Next() { + var e Entry + var ts int64 + var clusterID, details sql.NullString + var action string + if err := rows.Scan(&e.ID, &ts, &e.Actor, &action, &clusterID, &details); err != nil { + return nil, fmt.Errorf("audit: scan: %w", err) + } + e.Timestamp = time.Unix(0, ts) + e.Action = Action(action) + if clusterID.Valid { + e.ClusterID = clusterID.String + } + if details.Valid { + e.Details = details.String + } + out = append(out, &e) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("audit: rows: %w", err) + } + return out, nil +} + +// MarshalDetails encodes any value as a JSON string for Entry.Details. +func MarshalDetails(v any) (string, error) { + if v == nil { + return "", nil + } + b, err := json.Marshal(v) + if err != nil { + return "", err + } + return string(b), nil +} + +// ErrNotFound is unused in this package (single-entry lookups are not +// supported), but keeps the errors import available for future use. +var _ = errors.New diff --git a/internal/httpclient/auth.go b/internal/httpclient/auth.go new file mode 100644 index 0000000..b538a83 --- /dev/null +++ b/internal/httpclient/auth.go @@ -0,0 +1,53 @@ +package httpclient + +import ( + "errors" + "fmt" + "net/http" + "net/url" + + "spark-mcp-go/internal/cluster" +) + +// ApplyAuth rewrites req according to the cluster's configured AuthType. +// - none / empty: no-op +// - simple: appends ?user.name= (YARN SimpleAuth) +// - basic: attaches HTTP Basic credentials +// +// baseURL is kept for future use (e.g. reconstructing absolute URLs in Phase 5) +// and is intentionally unused in this version. +func ApplyAuth(req *http.Request, baseURL string, c *cluster.Cluster) error { + if baseURL == "" { + // baseURL is reserved for Phase 5 host rewriting. + } + + switch c.AuthType { + case cluster.AuthNone, "": + return nil + + case cluster.AuthSimple: + u, err := url.Parse(req.URL.String()) + if err != nil { + return fmt.Errorf("httpclient: simple auth parse url: %w", err) + } + q := u.Query() + user := c.AuthUsername + if user == "" { + user = "yarn" + } + q.Set("user.name", user) + u.RawQuery = q.Encode() + req.URL = u + return nil + + case cluster.AuthBasic: + if c.AuthUsername == "" || c.AuthPassword == "" { + return errors.New("httpclient: basic auth requires username and password") + } + req.SetBasicAuth(c.AuthUsername, c.AuthPassword) + return nil + + default: + return fmt.Errorf("httpclient: unknown auth type %q", c.AuthType) + } +} diff --git a/internal/httpclient/auth_test.go b/internal/httpclient/auth_test.go new file mode 100644 index 0000000..b36ca30 --- /dev/null +++ b/internal/httpclient/auth_test.go @@ -0,0 +1,99 @@ +package httpclient + +import ( + "encoding/base64" + "net/http" + "strings" + "testing" + + "spark-mcp-go/internal/cluster" +) + +func TestApplyAuth_None(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "http://example.com/foo", nil) + c := &cluster.Cluster{AuthType: cluster.AuthNone} + if err := ApplyAuth(req, "", c); err != nil { + t.Fatalf("apply auth: %v", err) + } + if req.Header.Get("Authorization") != "" { + t.Errorf("expected no Authorization header") + } + if req.URL.Query().Get("user.name") != "" { + t.Errorf("expected no user.name query parameter") + } +} + +func TestApplyAuth_Simple_DefaultUser(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "http://rm.example.com/ws/v1/cluster/apps", nil) + c := &cluster.Cluster{AuthType: cluster.AuthSimple} + if err := ApplyAuth(req, "", c); err != nil { + t.Fatalf("apply auth: %v", err) + } + if got := req.URL.Query().Get("user.name"); got != "yarn" { + t.Errorf("user.name=%q, want yarn", got) + } +} + +func TestApplyAuth_Simple_CustomUser(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "http://rm.example.com/ws/v1/cluster/apps", nil) + c := &cluster.Cluster{AuthType: cluster.AuthSimple, AuthUsername: "alice"} + if err := ApplyAuth(req, "", c); err != nil { + t.Fatalf("apply auth: %v", err) + } + if got := req.URL.Query().Get("user.name"); got != "alice" { + t.Errorf("user.name=%q, want alice", got) + } +} + +func TestApplyAuth_Simple_PreservesExistingQuery(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "http://rm.example.com/ws/v1/cluster/apps?foo=bar", nil) + c := &cluster.Cluster{AuthType: cluster.AuthSimple} + if err := ApplyAuth(req, "", c); err != nil { + t.Fatalf("apply auth: %v", err) + } + q := req.URL.Query() + if q.Get("foo") != "bar" { + t.Errorf("foo=%q, want bar", q.Get("foo")) + } + if q.Get("user.name") != "yarn" { + t.Errorf("user.name=%q, want yarn", q.Get("user.name")) + } + raw := req.URL.RawQuery + if !strings.Contains(raw, "foo=bar") || !strings.Contains(raw, "user.name=yarn") { + t.Errorf("raw query %q missing expected parameters", raw) + } +} + +func TestApplyAuth_Basic(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "http://rm.example.com/foo", nil) + c := &cluster.Cluster{AuthType: cluster.AuthBasic, AuthUsername: "u", AuthPassword: "p"} + if err := ApplyAuth(req, "", c); err != nil { + t.Fatalf("apply auth: %v", err) + } + want := "Basic " + base64.StdEncoding.EncodeToString([]byte("u:p")) + if got := req.Header.Get("Authorization"); got != want { + t.Errorf("Authorization=%q, want %q", got, want) + } +} + +func TestApplyAuth_Basic_MissingCreds(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "http://rm.example.com/foo", nil) + c := &cluster.Cluster{AuthType: cluster.AuthBasic, AuthUsername: "", AuthPassword: "p"} + if err := ApplyAuth(req, "", c); err == nil { + t.Error("expected error for missing username") + } + + req2, _ := http.NewRequest(http.MethodGet, "http://rm.example.com/foo", nil) + c2 := &cluster.Cluster{AuthType: cluster.AuthBasic, AuthUsername: "u", AuthPassword: ""} + if err := ApplyAuth(req2, "", c2); err == nil { + t.Error("expected error for missing password") + } +} + +func TestApplyAuth_UnknownAuthType(t *testing.T) { + req, _ := http.NewRequest(http.MethodGet, "http://rm.example.com/foo", nil) + c := &cluster.Cluster{AuthType: "kerberos"} + if err := ApplyAuth(req, "", c); err == nil { + t.Error("expected error for unknown auth type") + } +} diff --git a/internal/httpclient/client.go b/internal/httpclient/client.go new file mode 100644 index 0000000..8b3e344 --- /dev/null +++ b/internal/httpclient/client.go @@ -0,0 +1,60 @@ +package httpclient + +import ( + "context" + "io" + "net/http" + "time" +) + +// Config tunes the shared HTTP client. +type Config struct { + Timeout time.Duration // e.g. 30s + MaxResponseBytes int64 // e.g. 1 << 20 +} + +// Client is a thin, SSRF-aware wrapper around net/http.Client. +type Client struct { + cfg Config + hc *http.Client +} + +// New creates a Client with redirect following disabled. +// Redirect handling is intentionally left to DoWithRedirect. +func New(cfg Config) *Client { + return &Client{ + cfg: cfg, + hc: &http.Client{ + Timeout: cfg.Timeout, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + return http.ErrUseLastResponse + }, + }, + } +} + +// Do executes a single HTTP request. The response body is truncated to +// MaxResponseBytes. The request URL is validated by CheckURL before sending; +// hosts listed in allowedHosts bypass SSRF checks. +// +// TODO(prod): implement IP-pinned dialer to fully close TOCTOU between the +// SSRF check here and the actual TCP dial performed by net/http. +func (c *Client) Do(ctx context.Context, req *http.Request, allowedHosts ...string) (*http.Response, error) { + if err := CheckURL(req.URL.String(), allowedHosts...); err != nil { + return nil, err + } + + resp, err := c.hc.Do(req.WithContext(ctx)) + if err != nil { + return nil, err + } + + resp.Body = struct { + io.Reader + io.Closer + }{ + Reader: io.LimitReader(resp.Body, c.cfg.MaxResponseBytes), + Closer: resp.Body, + } + return resp, nil +} diff --git a/internal/httpclient/client_test.go b/internal/httpclient/client_test.go new file mode 100644 index 0000000..9e741ee --- /dev/null +++ b/internal/httpclient/client_test.go @@ -0,0 +1,56 @@ +package httpclient + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestClient_Timeout(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(2 * time.Second) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + c := New(Config{Timeout: 100 * time.Millisecond, MaxResponseBytes: 1024}) + req, err := http.NewRequest(http.MethodGet, srv.URL, nil) + if err != nil { + t.Fatalf("new request: %v", err) + } + _, err = c.Do(context.Background(), req) + if err == nil { + t.Fatal("expected timeout error, got nil") + } +} + +func TestClient_MaxResponseBytes(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte(strings.Repeat("a", 2048))) + })) + defer srv.Close() + + c := New(Config{Timeout: 5 * time.Second, MaxResponseBytes: 512}) + req, err := http.NewRequest(http.MethodGet, srv.URL, nil) + if err != nil { + t.Fatalf("new request: %v", err) + } + resp, err := c.Do(context.Background(), req, "127.0.0.1") + if err != nil { + t.Fatalf("do: %v", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if len(body) > 512 { + t.Fatalf("expected at most 512 bytes, got %d", len(body)) + } +} diff --git a/internal/httpclient/redirect.go b/internal/httpclient/redirect.go new file mode 100644 index 0000000..178c62e --- /dev/null +++ b/internal/httpclient/redirect.go @@ -0,0 +1,62 @@ +package httpclient + +import ( + "context" + "errors" + "fmt" + "net/http" +) + +const defaultMaxRedirects = 5 + +// DoWithRedirect 手动跟随 3xx,跨主机跳转保留 Authorization header。 +// maxRedirects == 0 禁止 redirect(等同 Do) +// maxRedirects < 0 用 defaultMaxRedirects +func (c *Client) DoWithRedirect(ctx context.Context, req *http.Request, maxRedirects int, allowedHosts ...string) (*http.Response, error) { + if maxRedirects == 0 { + return c.Do(ctx, req, allowedHosts...) + } + if maxRedirects < 0 { + maxRedirects = defaultMaxRedirects + } + + current := req + remaining := maxRedirects + for { + resp, err := c.Do(ctx, current, allowedHosts...) + if err != nil { + return nil, err + } + if resp.StatusCode < 300 || resp.StatusCode >= 400 { + return resp, nil + } + + // 3xx + if remaining == 0 { + _ = resp.Body.Close() + return nil, errors.New("httpclient: too many redirects") + } + remaining-- + + loc := resp.Header.Get("Location") + if loc == "" { + _ = resp.Body.Close() + return nil, fmt.Errorf("httpclient: %d %s without Location header", resp.StatusCode, resp.Status) + } + next, err := current.URL.Parse(loc) + if err != nil { + _ = resp.Body.Close() + return nil, fmt.Errorf("httpclient: parse redirect Location %q: %w", loc, err) + } + _ = resp.Body.Close() + + newReq, err := http.NewRequestWithContext(ctx, current.Method, next.String(), nil) + if err != nil { + return nil, fmt.Errorf("httpclient: build redirect request: %w", err) + } + // **保留** header (Authorization 等) + newReq.Header = current.Header.Clone() + newReq.Body = nil + current = newReq + } +} diff --git a/internal/httpclient/redirect_test.go b/internal/httpclient/redirect_test.go new file mode 100644 index 0000000..12c42ab --- /dev/null +++ b/internal/httpclient/redirect_test.go @@ -0,0 +1,142 @@ +package httpclient + +import ( + "context" + "encoding/base64" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +func TestDoWithRedirect_SameHost(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/start": + w.Header().Set("Location", "/final") + w.WriteHeader(http.StatusFound) + case "/final": + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) + default: + w.WriteHeader(http.StatusNotFound) + } + })) + defer srv.Close() + + c := New(Config{Timeout: 5 * time.Second, MaxResponseBytes: 1024}) + req, err := http.NewRequest(http.MethodGet, srv.URL+"/start", nil) + if err != nil { + t.Fatalf("new request: %v", err) + } + resp, err := c.DoWithRedirect(context.Background(), req, 5, "127.0.0.1") + if err != nil { + t.Fatalf("do with redirect: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status=%d, want 200", resp.StatusCode) + } + body, _ := io.ReadAll(resp.Body) + if string(body) != "ok" { + t.Fatalf("body=%q, want ok", string(body)) + } +} + +func TestDoWithRedirect_CrossHost_PreservesAuth(t *testing.T) { + srv2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte(r.Header.Get("Authorization"))) + })) + defer srv2.Close() + + srv1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/start" { + w.Header().Set("Location", srv2.URL+"/final") + w.WriteHeader(http.StatusTemporaryRedirect) + return + } + w.WriteHeader(http.StatusNotFound) + })) + defer srv1.Close() + + c := New(Config{Timeout: 5 * time.Second, MaxResponseBytes: 1024}) + req, err := http.NewRequest(http.MethodGet, srv1.URL+"/start", nil) + if err != nil { + t.Fatalf("new request: %v", err) + } + req.SetBasicAuth("u", "p") + + resp, err := c.DoWithRedirect(context.Background(), req, 5, "127.0.0.1") + if err != nil { + t.Fatalf("do with redirect: %v", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + want := "Basic " + base64.StdEncoding.EncodeToString([]byte("u:p")) + if string(body) != want { + t.Fatalf("body=%q, want %q", string(body), want) + } +} + +func TestDoWithRedirect_MaxLimit(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", "/loop") + w.WriteHeader(http.StatusFound) + })) + defer srv.Close() + + c := New(Config{Timeout: 5 * time.Second, MaxResponseBytes: 1024}) + req, _ := http.NewRequest(http.MethodGet, srv.URL+"/loop", nil) + _, err := c.DoWithRedirect(context.Background(), req, 2, "127.0.0.1") + if err == nil { + t.Fatal("expected too many redirects error") + } + if !strings.Contains(err.Error(), "too many redirects") { + t.Fatalf("expected too many redirects, got %v", err) + } +} + +func TestDoWithRedirect_NoLocation(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusFound) + })) + defer srv.Close() + + c := New(Config{Timeout: 5 * time.Second, MaxResponseBytes: 1024}) + req, _ := http.NewRequest(http.MethodGet, srv.URL, nil) + _, err := c.DoWithRedirect(context.Background(), req, 5, "127.0.0.1") + if err == nil { + t.Fatal("expected error for missing Location") + } + if !strings.Contains(err.Error(), "without Location header") { + t.Fatalf("expected missing Location error, got %v", err) + } +} + +func TestDoWithRedirect_ZeroDisallows(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Location", "/elsewhere") + w.WriteHeader(http.StatusFound) + })) + defer srv.Close() + + c := New(Config{Timeout: 5 * time.Second, MaxResponseBytes: 1024}) + req, _ := http.NewRequest(http.MethodGet, srv.URL, nil) + resp, err := c.DoWithRedirect(context.Background(), req, 0, "127.0.0.1") + if err != nil { + t.Fatalf("do: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusFound { + t.Fatalf("status=%d, want 302", resp.StatusCode) + } +} diff --git a/internal/httpclient/ssrf.go b/internal/httpclient/ssrf.go new file mode 100644 index 0000000..201cab3 --- /dev/null +++ b/internal/httpclient/ssrf.go @@ -0,0 +1,127 @@ +package httpclient + +import ( + "fmt" + "net" + "net/url" + "strings" +) + +// privateCIDRs lists IPv4 and IPv6 ranges that must not be reached by the +// outbound HTTP client (SSRF prevention). Includes RFC 1918, loopback, link- +// local, multicast, documentation, and IPv4-mapped IPv6 ranges. +var privateCIDRs = []string{ + "0.0.0.0/8", "10.0.0.0/8", "100.64.0.0/10", + "127.0.0.0/8", "169.254.0.0/16", + "172.16.0.0/12", "192.0.0.0/24", "192.168.0.0/16", + "198.18.0.0/15", "224.0.0.0/4", "240.0.0.0/4", + "::1/128", "fc00::/7", "fe80::/10", "::ffff:0:0/96", +} + +// privateNets holds the parsed CIDR blocks from privateCIDRs. +var privateNets []*net.IPNet + +func init() { + for _, cidr := range privateCIDRs { + _, ipNet, err := net.ParseCIDR(cidr) + if err != nil { + // net.ParseCIDR only fails on malformed input; the list is static. + panic(fmt.Sprintf("httpclient: unable to parse private CIDR %q: %v", cidr, err)) + } + privateNets = append(privateNets, ipNet) + } +} + +// CheckURL validates that rawURL is an http(s) URL and that its hostname does +// not resolve to a private/reserved IP address. Hosts matching any entry in +// allowedHosts bypass the IP check. +// +// allowedHosts entries support exact matches, suffix matches (e.g. +// "example.com" matches "rm.example.com"), and wildcard suffixes (e.g. +// "*.example.com"). +func CheckURL(rawURL string, allowedHosts ...string) error { + u, err := url.Parse(rawURL) + if err != nil { + return fmt.Errorf("httpclient: parse url: %w", err) + } + + if !strings.EqualFold(u.Scheme, "http") && !strings.EqualFold(u.Scheme, "https") { + return fmt.Errorf("httpclient: unsupported url scheme %q", u.Scheme) + } + + host := strings.ToLower(u.Hostname()) + if host == "" { + return fmt.Errorf("httpclient: url has no hostname") + } + + for _, allowed := range allowedHosts { + if matchAllowedHost(host, allowed) { + return nil + } + } + + ips, err := net.LookupIP(host) + if err != nil { + return fmt.Errorf("httpclient: lookup host %q: %w", host, err) + } + if len(ips) == 0 { + return fmt.Errorf("httpclient: host %q resolved to no IPs", host) + } + + for _, ip := range ips { + if isPrivateIP(ip) { + return fmt.Errorf("httpclient: host %q resolves to private IP %s", host, ip) + } + } + + return nil +} + +// isPrivateIP reports whether ip falls inside any of the reserved CIDR blocks. +func isPrivateIP(ip net.IP) bool { + if v4 := ip.To4(); v4 != nil { + for _, ipNet := range privateNets { + if isIPv4MappedNet(ipNet) { + continue + } + if ipNet.Contains(v4) { + return true + } + } + return false + } + for _, ipNet := range privateNets { + if ipNet.Contains(ip) { + return true + } + } + return false +} + +// isIPv4MappedNet reports whether n is an IPv4-mapped IPv6 CIDR such as +// ::ffff:0:0/96. These must not be used to classify plain IPv4 addresses. +func isIPv4MappedNet(n *net.IPNet) bool { + return len(n.IP) == net.IPv6len && n.IP.To4() != nil && n.IP[10] == 0xff && n.IP[11] == 0xff +} + +// matchAllowedHost reports whether host matches pattern. Patterns may be exact +// hostnames, domain suffixes ("example.com" matches "rm.example.com"), or +// wildcard suffixes ("*.example.com"). +func matchAllowedHost(host, pattern string) bool { + pattern = strings.ToLower(strings.TrimSpace(pattern)) + if pattern == "" { + return false + } + if pattern == host { + return true + } + + if strings.HasPrefix(pattern, "*.") { + pattern = pattern[2:] + } + if strings.HasPrefix(pattern, ".") { + pattern = pattern[1:] + } + + return host == pattern || strings.HasSuffix(host, "."+pattern) +} diff --git a/internal/httpclient/ssrf_test.go b/internal/httpclient/ssrf_test.go new file mode 100644 index 0000000..d67db1d --- /dev/null +++ b/internal/httpclient/ssrf_test.go @@ -0,0 +1,79 @@ +package httpclient + +import ( + "net" + "net/http" + "net/http/httptest" + "testing" +) + +func TestCheckURL_RejectsLoopback(t *testing.T) { + err := CheckURL("http://127.0.0.1:8080/path") + if err == nil { + t.Fatal("expected error for loopback URL") + } +} + +func TestCheckURL_RejectsLinkLocal(t *testing.T) { + err := CheckURL("http://169.254.169.254/latest/meta-data") + if err == nil { + t.Fatal("expected error for link-local URL") + } +} + +func TestCheckURL_AllowsAllowlisted(t *testing.T) { + err := CheckURL("http://127.0.0.1:8080", "127.0.0.1") + if err != nil { + t.Fatalf("expected allowlisted host to pass: %v", err) + } + if !matchAllowedHost("rm.example.com", "*.example.com") { + t.Fatal("expected *.example.com to match rm.example.com") + } +} + +func TestCheckURL_RejectsBadScheme(t *testing.T) { + cases := []string{"file:///etc/passwd", "gopher://x", "ftp://x"} + for _, raw := range cases { + err := CheckURL(raw) + if err == nil { + t.Errorf("expected error for %q", raw) + } + } +} + +func TestCheckURL_RejectsEmptyHost(t *testing.T) { + err := CheckURL("http:///path") + if err == nil { + t.Fatal("expected error for URL with empty host") + } +} + +func TestCheckURL_AcceptsPublic(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + err := CheckURL(srv.URL, "127.0.0.1") + if err != nil { + t.Fatalf("expected allowlisted server URL to pass: %v", err) + } +} + +func TestIsPrivateIP(t *testing.T) { + cases := []struct { + ip string + want bool + }{ + {"10.1.2.3", true}, + {"8.8.8.8", false}, + {"::1", true}, + {"2001:db8::1", false}, + } + for _, tc := range cases { + got := isPrivateIP(net.ParseIP(tc.ip)) + if got != tc.want { + t.Errorf("isPrivateIP(%q) = %v, want %v", tc.ip, got, tc.want) + } + } +} diff --git a/internal/middleware/auth.go b/internal/middleware/auth.go new file mode 100644 index 0000000..9b8d7ae --- /dev/null +++ b/internal/middleware/auth.go @@ -0,0 +1,66 @@ +package middleware + +import ( + "crypto/subtle" + "net/http" + "strings" + + "github.com/gin-gonic/gin" +) + +// AdminAuth validates Authorization: Bearer against a list of +// allowed admin tokens. On failure it aborts with 401 and a JSON error body. +// Successful requests have the matched token stored under "admin_token". +func AdminAuth(adminTokens []string) gin.HandlerFunc { + return func(c *gin.Context) { + token := bearerToken(c) + if token == "" { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing bearer token"}) + return + } + if !matchAny(token, adminTokens) { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid bearer token"}) + return + } + c.Set("admin_token", token) + c.Next() + } +} + +// AgentAuth validates a single Authorization: Bearer against the +// configured agent token. It is intended for the LLM-facing /mcp endpoint. +func AgentAuth(agentToken string) gin.HandlerFunc { + return func(c *gin.Context) { + token := bearerToken(c) + if token == "" || subtle.ConstantTimeCompare([]byte(token), []byte(agentToken)) != 1 { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"}) + return + } + c.Set("agent_token", token) + c.Next() + } +} + +func bearerToken(c *gin.Context) string { + h := c.GetHeader("Authorization") + if h == "" { + return "" + } + const prefix = "Bearer " + if !strings.HasPrefix(h, prefix) { + return "" + } + return strings.TrimSpace(h[len(prefix):]) +} + +// matchAny compares the provided token against every candidate using a +// constant-time comparison to reduce timing side-channels. +func matchAny(token string, candidates []string) bool { + t := []byte(token) + for _, c := range candidates { + if subtle.ConstantTimeCompare(t, []byte(c)) == 1 { + return true + } + } + return false +} diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index fdbdf9e..f0e94be 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -83,3 +83,6 @@ func (d *DB) Close() error { } return d.sqlDB.Close() } + +// SQLDB returns the underlying *sql.DB handle. +func (d *DB) SQLDB() *sql.DB { return d.sqlDB } diff --git a/main.go b/main.go index 6d44789..e8b21ba 100644 --- a/main.go +++ b/main.go @@ -1,7 +1,7 @@ // Command spark-mcp-go launches the Spark MCP Server. // -// Phase 0 scope: wire config + slog + Gin + /healthz. -// Full MCP transport, admin API, and Tool surface land in later phases. +// Phase 4 scope: wire config + slog + Gin + /healthz + /admin/* cluster CRUD +// with audit log. MCP transport and Tool surface land in later phases. package main import ( @@ -16,8 +16,11 @@ import ( "github.com/gin-gonic/gin" + "spark-mcp-go/internal/admin" + "spark-mcp-go/internal/audit" "spark-mcp-go/internal/config" "spark-mcp-go/internal/logging" + "spark-mcp-go/internal/storage" ) func main() { @@ -50,6 +53,13 @@ func run() error { return err } + db, err := storage.Open(cfg.SQLitePath) + if err != nil { + return err + } + defer db.Close() + logger.Info("storage.open", "path", cfg.SQLitePath) + gin.SetMode(gin.ReleaseMode) r := gin.New() r.Use(gin.Recovery()) @@ -58,6 +68,8 @@ func run() error { c.JSON(http.StatusOK, gin.H{"ok": true, "version": "0.0.0"}) }) + admin.Mount(r, db.Clusters(), audit.NewRepo(db), cfg.AdminTokens) + srv := &http.Server{ Addr: cfg.ListenAddr, Handler: r,