mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-06 05:49:12 +02:00
e288eff2c5
When bootstrap is enabled and no PAT is available from the request header or Authorization bearer token, the server now generates a new PAT automatically and forwards it to the cloud service. This removes the need for the frontend to pass X-User-PAT — the server handles it entirely.
285 lines
8.3 KiB
Go
285 lines
8.3 KiB
Go
package handler
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
|
|
chimw "github.com/go-chi/chi/v5/middleware"
|
|
"github.com/multica-ai/multica/server/internal/auth"
|
|
"github.com/multica-ai/multica/server/internal/cloudruntime"
|
|
"github.com/multica-ai/multica/server/internal/logger"
|
|
db "github.com/multica-ai/multica/server/pkg/db/generated"
|
|
)
|
|
|
|
const maxCloudRuntimeRequestBodySize = 1 << 20
|
|
|
|
type cloudRuntimeProxyOptions struct {
|
|
withUserID bool
|
|
withQuery bool
|
|
withBody bool
|
|
withUserPAT bool
|
|
}
|
|
|
|
func (h *Handler) GetCloudRuntimeService(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) GetCloudRuntimeHealth(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodGet, "/healthz", cloudRuntimeProxyOptions{})
|
|
}
|
|
|
|
func (h *Handler) GetCloudRuntimeReady(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodGet, "/readyz", cloudRuntimeProxyOptions{})
|
|
}
|
|
|
|
func (h *Handler) ListCloudRuntimeNodes(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodGet, "/api/v1/nodes", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withQuery: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) CreateCloudRuntimeNode(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/nodes", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withBody: true,
|
|
withUserPAT: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) DeleteCloudRuntimeNode(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodDelete, "/api/v1/nodes", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withBody: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) StartCloudRuntimeNode(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/nodes/start", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withBody: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) StopCloudRuntimeNode(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/nodes/stop", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withBody: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) RebootCloudRuntimeNode(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/nodes/reboot", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withBody: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) GetCloudRuntimeNodeStatus(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/nodes/status", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withBody: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) ExecCloudRuntimeNode(w http.ResponseWriter, r *http.Request) {
|
|
h.proxyCloudRuntime(w, r, http.MethodPost, "/api/v1/nodes/exec", cloudRuntimeProxyOptions{
|
|
withUserID: true,
|
|
withBody: true,
|
|
})
|
|
}
|
|
|
|
func (h *Handler) proxyCloudRuntime(w http.ResponseWriter, r *http.Request, method, path string, opts cloudRuntimeProxyOptions) {
|
|
if h.CloudRuntime == nil || !h.CloudRuntime.Enabled() {
|
|
writeError(w, http.StatusServiceUnavailable, "cloud runtime is not configured")
|
|
return
|
|
}
|
|
|
|
var userID string
|
|
if opts.withUserID {
|
|
var ok bool
|
|
userID, ok = requireUserID(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
}
|
|
|
|
var body []byte
|
|
if opts.withBody {
|
|
var ok bool
|
|
body, ok = readCloudRuntimeJSONBody(w, r)
|
|
if !ok {
|
|
return
|
|
}
|
|
}
|
|
|
|
userPAT, patGenerated, ok := h.cloudRuntimeUserPAT(w, r, userID, opts.withUserPAT)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
var query url.Values
|
|
if opts.withQuery {
|
|
query = r.URL.Query()
|
|
}
|
|
|
|
resp, err := h.CloudRuntime.Do(r.Context(), cloudruntime.Request{
|
|
Method: method,
|
|
Path: path,
|
|
Query: query,
|
|
Body: body,
|
|
UserID: userID,
|
|
UserPAT: userPAT,
|
|
RequestID: cloudRuntimeRequestID(r),
|
|
})
|
|
if err != nil {
|
|
if patGenerated {
|
|
h.revokeGeneratedPAT(r.Context(), userPAT)
|
|
}
|
|
writeCloudRuntimeError(w, r, err)
|
|
return
|
|
}
|
|
if patGenerated && resp.StatusCode >= 300 {
|
|
h.revokeGeneratedPAT(r.Context(), userPAT)
|
|
}
|
|
writeCloudRuntimeResponse(w, resp)
|
|
}
|
|
|
|
func readCloudRuntimeJSONBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) {
|
|
r.Body = http.MaxBytesReader(w, r.Body, maxCloudRuntimeRequestBodySize)
|
|
data, err := io.ReadAll(r.Body)
|
|
if err != nil {
|
|
var maxErr *http.MaxBytesError
|
|
if errors.As(err, &maxErr) {
|
|
writeError(w, http.StatusRequestEntityTooLarge, "request body is too large")
|
|
return nil, false
|
|
}
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return nil, false
|
|
}
|
|
if len(bytes.TrimSpace(data)) == 0 {
|
|
writeError(w, http.StatusBadRequest, "request body is required")
|
|
return nil, false
|
|
}
|
|
var raw json.RawMessage
|
|
if err := json.Unmarshal(data, &raw); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid request body")
|
|
return nil, false
|
|
}
|
|
return data, true
|
|
}
|
|
|
|
func cloudRuntimeRequestID(r *http.Request) string {
|
|
if id := r.Header.Get("X-Request-ID"); id != "" {
|
|
return id
|
|
}
|
|
return chimw.GetReqID(r.Context())
|
|
}
|
|
|
|
func (h *Handler) cloudRuntimeUserPAT(w http.ResponseWriter, r *http.Request, userID string, enabled bool) (pat string, generated bool, ok bool) {
|
|
if !enabled {
|
|
return "", false, true
|
|
}
|
|
if p := strings.TrimSpace(r.Header.Get("X-User-PAT")); p != "" {
|
|
if !strings.HasPrefix(p, "mul_") {
|
|
writeError(w, http.StatusBadRequest, "invalid X-User-PAT")
|
|
return "", false, false
|
|
}
|
|
if h.Queries == nil {
|
|
writeError(w, http.StatusInternalServerError, "failed to validate X-User-PAT")
|
|
return "", false, false
|
|
}
|
|
token, err := h.Queries.GetPersonalAccessTokenByHash(r.Context(), auth.HashToken(p))
|
|
if err != nil || uuidToString(token.UserID) != userID {
|
|
writeError(w, http.StatusForbidden, "invalid X-User-PAT")
|
|
return "", false, false
|
|
}
|
|
return p, false, true
|
|
}
|
|
|
|
authHeader := strings.TrimSpace(r.Header.Get("Authorization"))
|
|
const prefix = "Bearer "
|
|
if strings.HasPrefix(authHeader, prefix+"mul_") {
|
|
return strings.TrimPrefix(authHeader, prefix), false, true
|
|
}
|
|
|
|
// Auto-generate a PAT for cloud runtime bootstrap.
|
|
p, err := h.generateCloudRuntimePAT(r.Context(), userID)
|
|
if err != nil {
|
|
slog.Error("failed to generate cloud runtime PAT", append(logger.RequestAttrs(r), "error", err)...)
|
|
writeError(w, http.StatusInternalServerError, "failed to generate cloud runtime PAT")
|
|
return "", false, false
|
|
}
|
|
return p, true, true
|
|
}
|
|
|
|
func (h *Handler) generateCloudRuntimePAT(ctx context.Context, userID string) (string, error) {
|
|
rawToken, err := auth.GeneratePATToken()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
prefix := rawToken[:12]
|
|
_, err = h.Queries.CreatePersonalAccessToken(ctx, db.CreatePersonalAccessTokenParams{
|
|
UserID: parseUUID(userID),
|
|
Name: "Cloud Runtime (auto)",
|
|
TokenHash: auth.HashToken(rawToken),
|
|
TokenPrefix: prefix,
|
|
})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return rawToken, nil
|
|
}
|
|
|
|
func (h *Handler) revokeGeneratedPAT(ctx context.Context, rawToken string) {
|
|
ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
|
defer cancel()
|
|
hash := auth.HashToken(rawToken)
|
|
if _, err := h.DB.Exec(ctx, `UPDATE personal_access_token SET revoked = TRUE WHERE token_hash = $1`, hash); err != nil {
|
|
slog.Warn("failed to revoke auto-generated cloud runtime PAT", "error", err)
|
|
}
|
|
h.PATCache.Invalidate(ctx, hash)
|
|
}
|
|
|
|
func writeCloudRuntimeResponse(w http.ResponseWriter, resp *cloudruntime.Response) {
|
|
if requestID := resp.Header.Get("X-Request-ID"); requestID != "" {
|
|
w.Header().Set("X-Request-ID", requestID)
|
|
}
|
|
body := bytes.TrimSpace(resp.Body)
|
|
if len(body) == 0 {
|
|
w.WriteHeader(resp.StatusCode)
|
|
return
|
|
}
|
|
if json.Valid(body) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(resp.StatusCode)
|
|
w.Write(body)
|
|
return
|
|
}
|
|
writeJSON(w, resp.StatusCode, map[string]string{"error": string(body)})
|
|
}
|
|
|
|
func writeCloudRuntimeError(w http.ResponseWriter, r *http.Request, err error) {
|
|
switch {
|
|
case errors.Is(err, cloudruntime.ErrDisabled):
|
|
writeError(w, http.StatusServiceUnavailable, "cloud runtime is not configured")
|
|
case errors.Is(err, cloudruntime.ErrInvalidBaseURL):
|
|
writeError(w, http.StatusServiceUnavailable, "cloud runtime is misconfigured")
|
|
case errors.Is(err, context.DeadlineExceeded):
|
|
writeError(w, http.StatusGatewayTimeout, "cloud runtime request timed out")
|
|
default:
|
|
slog.Warn("cloud runtime request failed", append(logger.RequestAttrs(r), "error", err)...)
|
|
writeError(w, http.StatusBadGateway, "cloud runtime request failed")
|
|
}
|
|
}
|