Files
DoGaMa-serv/internal/agent/agent_server.go
T
codex b575b5260d
CI / validate (pull_request) Failing after 8m3s
fix(deploy): start current Palworld template
2026-08-15 21:43:48 +02:00

513 lines
20 KiB
Go

// Package agent implements the deny-by-default privileged Docker boundary.
package agent
import (
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"regexp"
"strings"
"syscall"
"time"
"git.zaynet.fr/DoGaMa/DoGaMa-serv/internal/agentwire"
)
const maxDiskPaths = 16
var diagnosticSecretPattern = regexp.MustCompile(`(?i)(password|token|secret|api[_-]?key)\s*[:=]\s*[^\s,;]+`)
type service struct {
paths *PathPolicy
registry *Registry
plans *PlanPolicy
docker DockerRuntime
logger *slog.Logger
disk DiskChecker
}
type DiskChecker interface {
AvailableBytes(string) (uint64, uint64, error)
}
type statfsDiskChecker struct{}
func (statfsDiskChecker) AvailableBytes(path string) (uint64, uint64, error) {
var filesystem syscall.Statfs_t
if err := syscall.Statfs(path, &filesystem); err != nil || filesystem.Bsize <= 0 {
return 0, 0, errors.New("filesystem unavailable")
}
blockSize := uint64(filesystem.Bsize)
return blockSize * filesystem.Bavail, blockSize * filesystem.Blocks, nil
}
// NewHandler constructs the complete authenticated private agent API.
func NewHandler(authenticator *Authenticator, paths *PathPolicy, plans *PlanPolicy, registry *Registry, docker DockerRuntime, logger *slog.Logger) http.Handler {
return NewHandlerWithDiskChecker(authenticator, paths, plans, registry, docker, statfsDiskChecker{}, logger)
}
func NewHandlerWithDiskChecker(authenticator *Authenticator, paths *PathPolicy, plans *PlanPolicy, registry *Registry, docker DockerRuntime, disk DiskChecker, logger *slog.Logger) http.Handler {
server := &service{paths: paths, plans: plans, registry: registry, docker: docker, disk: disk, logger: logger}
mux := http.NewServeMux()
mux.HandleFunc("GET /v1/health", server.health)
mux.HandleFunc("POST /v1/check-disk", server.checkDisk)
mux.HandleFunc("GET /v1/instances", server.listInstances)
mux.HandleFunc("POST /v1/check-ports", server.checkPorts)
mux.HandleFunc("POST /v1/instances", server.createInstance)
mux.HandleFunc("PUT /v1/instances/{id}", server.replaceInstance)
mux.HandleFunc("GET /v1/instances/{id}", server.inspectInstance)
mux.HandleFunc("GET /v1/instances/{id}/stats", server.instanceStats)
mux.HandleFunc("POST /v1/instances/{id}/start", server.startInstance)
mux.HandleFunc("POST /v1/instances/{id}/stop", server.stopInstance)
mux.HandleFunc("POST /v1/instances/{id}/restart", server.restartInstance)
mux.HandleFunc("DELETE /v1/instances/{id}", server.deleteInstance)
return server.headers(authenticator.Middleware(mux))
}
func (s *service) health(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
defer cancel()
if err := s.docker.Ping(ctx); err != nil {
s.logger.Warn("Docker daemon unavailable", "event", "agent.docker.unavailable")
writeProblem(w, http.StatusServiceUnavailable, "docker_unavailable", "The Docker daemon is unavailable.")
return
}
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}
func (s *service) checkDisk(w http.ResponseWriter, r *http.Request) {
var request struct {
Paths []string `json:"paths"`
}
if err := decodeJSON(r.Body, &request); err != nil || len(request.Paths) == 0 || len(request.Paths) > maxDiskPaths {
writeProblem(w, http.StatusBadRequest, "invalid_request", "The request is invalid.")
return
}
type diskInfo struct {
Path string `json:"path"`
BytesAvailable uint64 `json:"bytes_available"`
BytesTotal uint64 `json:"bytes_total"`
}
response := struct {
Paths []diskInfo `json:"paths"`
}{Paths: make([]diskInfo, 0, len(request.Paths))}
seen := make(map[string]struct{}, len(request.Paths))
for _, requested := range request.Paths {
canonical, err := s.paths.Resolve(requested)
if err != nil {
writeProblem(w, http.StatusUnprocessableEntity, "path_not_allowed", "A requested path is not allowed.")
return
}
if _, exists := seen[canonical]; exists {
continue
}
seen[canonical] = struct{}{}
available, total, err := s.disk.AvailableBytes(canonical)
if err != nil {
writeProblem(w, http.StatusUnprocessableEntity, "path_unavailable", "A requested path is unavailable.")
return
}
response.Paths = append(response.Paths, diskInfo{
Path: requested,
BytesAvailable: available,
BytesTotal: total,
})
}
writeJSON(w, http.StatusOK, response)
}
func (s *service) listInstances(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, struct {
Instances []RegisteredInstance `json:"instances"`
}{Instances: s.registry.List()})
}
func (s *service) checkPorts(w http.ResponseWriter, r *http.Request) {
var request struct {
Ports []agentwire.PlanPort `json:"ports"`
}
if decodeJSON(r.Body, &request) != nil || len(request.Ports) == 0 || len(request.Ports) > 32 {
writeProblem(w, http.StatusBadRequest, "invalid_request", "The request is invalid.")
return
}
if err := s.docker.CheckPorts(r.Context(), request.Ports); err != nil {
writeProblem(w, http.StatusConflict, "port_unavailable", "A requested host port is unavailable.")
return
}
writeJSON(w, http.StatusOK, map[string]bool{"available": true})
}
func (s *service) createInstance(w http.ResponseWriter, r *http.Request) {
var plan agentwire.DeploymentPlan
if err := decodeJSON(r.Body, &plan); err != nil {
writeDiagnosticProblem(w, http.StatusUnprocessableEntity, "invalid_plan", "The deployment plan is invalid.", map[string]any{"phase": "preparation", "operation_error": redactDiagnostic(err.Error())})
return
}
if err := s.plans.Validate(plan); err != nil {
s.logger.Error("instance plan rejected", "event", "instance.create.plan_rejected", "instance_id", plan.InstanceID, "template_id", plan.TemplateID, "template_version", plan.TemplateVersion, "template_digest", plan.TemplateDigest, "error", err)
writeDiagnosticProblem(w, http.StatusUnprocessableEntity, "invalid_plan", "The deployment plan is invalid.", map[string]any{"phase": "preparation", "template_id": plan.TemplateID, "template_version": plan.TemplateVersion, "template_digest": plan.TemplateDigest, "operation_error": redactDiagnostic(err.Error())})
return
}
for index := range plan.Mounts {
mount := &plan.Mounts[index]
canonical, err := s.paths.Prepare(mount.HostPath)
if err != nil {
writeProblem(w, http.StatusUnprocessableEntity, "path_not_allowed", "A deployment path is not allowed.")
return
}
mount.HostPath = canonical
available, _, diskErr := s.disk.AvailableBytes(canonical)
if diskErr != nil || available < uint64(plan.Resources.StorageGB)*1024*1024*1024 {
writeProblem(w, http.StatusUnprocessableEntity, "insufficient_disk", "A deployment path has insufficient disk space.")
return
}
}
if existing, ok := s.registry.Get(plan.InstanceID); ok {
if existing.PlanDigest != plan.PlanDigest {
writeProblem(w, http.StatusConflict, "registration_conflict", "The instance registration conflicts with the deployment plan.")
return
}
state, err := s.boundState(r.Context(), existing)
if err != nil {
writeProblem(w, http.StatusConflict, "registration_mismatch", "The registered container binding is invalid.")
return
}
writeJSON(w, http.StatusOK, state)
return
}
if err := s.docker.CheckPorts(r.Context(), plan.Ports); err != nil {
writeProblem(w, http.StatusConflict, "port_unavailable", "A requested host port is unavailable.")
return
}
assets, err := s.prepareAssets(plan)
if err != nil {
writeProblem(w, http.StatusUnprocessableEntity, "asset_prepare_failed", "Approved template assets could not be prepared.")
return
}
containerID, err := s.docker.Create(r.Context(), plan, assets)
if err != nil {
s.logger.Error("instance create failed", "event", "instance.create.failed", "instance_id", plan.InstanceID, "error", err)
writeDiagnosticProblem(w, http.StatusBadGateway, "container_create_failed", "The container could not be created.", map[string]any{"operation_error": redactDiagnostic(err.Error())})
return
}
entry := RegisteredInstance{InstanceID: plan.InstanceID, ContainerID: containerID, PlanDigest: plan.PlanDigest}
if err := s.registry.Register(entry); err != nil {
_ = s.docker.Delete(r.Context(), containerID)
writeProblem(w, http.StatusInternalServerError, "registration_failed", "The container registration could not be persisted.")
return
}
writeJSON(w, http.StatusCreated, agentwire.InstanceState{InstanceID: plan.InstanceID, ContainerID: containerID, PlanDigest: plan.PlanDigest, Health: "stopped"})
}
func (s *service) replaceInstance(w http.ResponseWriter, r *http.Request) {
var plan agentwire.DeploymentPlan
id := r.PathValue("id")
if decodeJSON(r.Body, &plan) != nil || plan.InstanceID != id || s.plans.Validate(plan) != nil {
writeProblem(w, http.StatusUnprocessableEntity, "invalid_plan", "The replacement plan is invalid.")
return
}
entry, ok := s.registry.Get(id)
if !ok {
writeProblem(w, http.StatusNotFound, "registration_not_found", "The instance is not registered.")
return
}
oldState, err := s.boundState(r.Context(), entry)
if err != nil {
writeProblem(w, http.StatusConflict, "registration_mismatch", "The registered container binding is invalid.")
return
}
for index := range plan.Mounts {
canonical, pathErr := s.paths.Resolve(plan.Mounts[index].HostPath)
if pathErr != nil {
writeProblem(w, http.StatusUnprocessableEntity, "path_not_allowed", "A deployment path is not allowed.")
return
}
plan.Mounts[index].HostPath = canonical
}
assets, err := s.prepareAssets(plan)
if err != nil {
writeProblem(w, http.StatusUnprocessableEntity, "asset_prepare_failed", "Approved template assets could not be prepared.")
return
}
if oldState.Running {
if err := s.docker.Stop(r.Context(), entry.ContainerID, plan.StopTimeoutSeconds); err != nil {
writeProblem(w, http.StatusBadGateway, "container_stop_failed", "The previous container could not be stopped.")
return
}
}
if err := s.docker.Delete(r.Context(), entry.ContainerID); err != nil {
writeProblem(w, http.StatusBadGateway, "container_delete_failed", "The previous container could not be removed.")
return
}
containerID, err := s.docker.Create(r.Context(), plan, assets)
if err != nil {
writeProblem(w, http.StatusBadGateway, "container_replace_failed", "The replacement container could not be created; persistent data was preserved.")
return
}
replacement := RegisteredInstance{InstanceID: id, ContainerID: containerID, PlanDigest: plan.PlanDigest}
if err := s.registry.Replace(replacement); err != nil {
_ = s.docker.Delete(r.Context(), containerID)
writeProblem(w, http.StatusInternalServerError, "registration_failed", "The replacement registration could not be persisted.")
return
}
writeJSON(w, http.StatusOK, agentwire.InstanceState{InstanceID: id, ContainerID: containerID, PlanDigest: plan.PlanDigest, Health: "stopped"})
}
func (s *service) inspectInstance(w http.ResponseWriter, r *http.Request) {
entry, ok := s.registration(w, r.PathValue("id"))
if !ok {
return
}
state, err := s.boundState(r.Context(), entry)
if err != nil {
writeProblem(w, http.StatusConflict, "registration_mismatch", "The registered container binding is invalid.")
return
}
writeJSON(w, http.StatusOK, state)
}
func (s *service) instanceStats(w http.ResponseWriter, r *http.Request) {
entry, ok := s.registration(w, r.PathValue("id"))
if !ok {
return
}
if _, err := s.boundState(r.Context(), entry); err != nil {
writeProblem(w, http.StatusConflict, "registration_mismatch", "The registered container binding is invalid.")
return
}
stats, err := s.docker.Stats(r.Context(), entry.ContainerID)
if err != nil {
writeProblem(w, http.StatusBadGateway, "stats_unavailable", "Container statistics are unavailable.")
return
}
stats.InstanceID = entry.InstanceID
writeJSON(w, http.StatusOK, stats)
}
func (s *service) startInstance(w http.ResponseWriter, r *http.Request) {
entry, ok := s.registration(w, r.PathValue("id"))
if !ok {
return
}
state, err := s.boundState(r.Context(), entry)
if err != nil {
s.bindingProblem(w)
return
}
if !state.Running {
if err := s.docker.Start(r.Context(), entry.ContainerID); err != nil {
s.logger.Error("instance start failed", "event", "instance.start.failed", "instance_id", entry.InstanceID, "container_id", entry.ContainerID, "error", err)
writeDiagnosticProblem(w, http.StatusBadGateway, "start_failed", "The registered container could not be started.", s.containerDiagnostic(r.Context(), entry.ContainerID, err))
return
}
inspection, inspectErr := s.docker.Inspect(r.Context(), entry.ContainerID)
if inspectErr != nil || (!inspection.Running && (inspection.Status != "" || inspection.FinishedAt != "")) {
s.logger.Error("instance exited after start", "event", "container.exited", "instance_id", entry.InstanceID, "container_id", entry.ContainerID, "error", err)
writeDiagnosticProblem(w, http.StatusBadGateway, "start_exited", "The registered container exited immediately after start.", s.containerDiagnostic(r.Context(), entry.ContainerID, inspectErr))
return
}
if inspection.Running {
state, err = s.boundState(r.Context(), entry)
if err != nil {
writeDiagnosticProblem(w, http.StatusBadGateway, "start_inspect_failed", "The started container could not be inspected.", s.containerDiagnostic(r.Context(), entry.ContainerID, err))
return
}
} else {
state.Running, state.Health = true, "starting"
}
}
writeJSON(w, http.StatusOK, state)
}
func (s *service) stopInstance(w http.ResponseWriter, r *http.Request) {
entry, ok := s.registration(w, r.PathValue("id"))
if !ok {
return
}
state, err := s.boundState(r.Context(), entry)
if err != nil {
s.bindingProblem(w)
return
}
var request struct {
TimeoutSeconds int `json:"timeout_seconds"`
}
if decodeJSON(r.Body, &request) != nil || request.TimeoutSeconds < 5 || request.TimeoutSeconds > 900 {
writeProblem(w, http.StatusBadRequest, "invalid_request", "The stop timeout is invalid.")
return
}
if state.Running {
if err := s.docker.Stop(r.Context(), entry.ContainerID, request.TimeoutSeconds); err != nil {
writeProblem(w, http.StatusBadGateway, "stop_failed", "The registered container could not be stopped.")
return
}
state.Running, state.Ready, state.Health = false, false, "stopped"
}
writeJSON(w, http.StatusOK, state)
}
func (s *service) restartInstance(w http.ResponseWriter, r *http.Request) {
entry, ok := s.registration(w, r.PathValue("id"))
if !ok {
return
}
state, err := s.boundState(r.Context(), entry)
if err != nil {
s.bindingProblem(w)
return
}
var request struct {
TimeoutSeconds int `json:"timeout_seconds"`
}
if decodeJSON(r.Body, &request) != nil || request.TimeoutSeconds < 5 || request.TimeoutSeconds > 900 {
writeProblem(w, http.StatusBadRequest, "invalid_request", "The restart timeout is invalid.")
return
}
if state.Running {
err = s.docker.Restart(r.Context(), entry.ContainerID, request.TimeoutSeconds)
} else {
err = s.docker.Start(r.Context(), entry.ContainerID)
}
if err != nil {
writeProblem(w, http.StatusBadGateway, "restart_failed", "The registered container could not be restarted.")
return
}
state.Running, state.Ready, state.Health = true, false, "starting"
writeJSON(w, http.StatusOK, state)
}
func (s *service) deleteInstance(w http.ResponseWriter, r *http.Request) {
entry, ok := s.registry.Get(r.PathValue("id"))
if !ok {
w.WriteHeader(http.StatusNoContent)
return
}
state, err := s.boundState(r.Context(), entry)
if err != nil {
s.bindingProblem(w)
return
}
if state.Running {
writeProblem(w, http.StatusConflict, "instance_running", "The instance must be stopped before container deletion.")
return
}
if err := s.docker.Delete(r.Context(), entry.ContainerID); err != nil {
writeProblem(w, http.StatusBadGateway, "delete_failed", "The registered container could not be deleted.")
return
}
if err := s.registry.Remove(entry.InstanceID); err != nil {
writeProblem(w, http.StatusInternalServerError, "registry_update_failed", "The agent registry could not be updated.")
return
}
w.WriteHeader(http.StatusNoContent)
}
func (s *service) registration(w http.ResponseWriter, id string) (RegisteredInstance, bool) {
entry, ok := s.registry.Get(id)
if !ok {
writeProblem(w, http.StatusNotFound, "instance_not_registered", "The instance is not registered.")
}
return entry, ok
}
func (s *service) boundState(ctx context.Context, entry RegisteredInstance) (agentwire.InstanceState, error) {
inspection, err := s.docker.Inspect(ctx, entry.ContainerID)
if err != nil || inspection.ContainerID != entry.ContainerID || inspection.Labels["io.dogama.managed"] != "true" || inspection.Labels["io.dogama.instance-id"] != entry.InstanceID || inspection.Labels["io.dogama.plan-digest"] != entry.PlanDigest {
return agentwire.InstanceState{}, errors.New("registration binding mismatch")
}
health := inspection.Health
if !inspection.Running {
health = "stopped"
}
return agentwire.InstanceState{InstanceID: entry.InstanceID, ContainerID: entry.ContainerID, PlanDigest: entry.PlanDigest, Running: inspection.Running, Ready: inspection.Running && (inspection.Health == "healthy" || inspection.Health == "none"), Health: health, ExitCode: inspection.ExitCode}, nil
}
func (s *service) bindingProblem(w http.ResponseWriter) {
writeProblem(w, http.StatusConflict, "registration_mismatch", "The registered container binding is invalid.")
}
// containerDiagnostic deliberately contains Docker state rather than request
// payloads: it is safe to return over the authenticated private agent link and
// remains useful when stdout/stderr is empty.
func (s *service) containerDiagnostic(ctx context.Context, containerID string, cause error) map[string]any {
diagnostic := map[string]any{"container_id": containerID}
if cause != nil {
diagnostic["operation_error"] = redactDiagnostic(cause.Error())
}
inspection, err := s.docker.Inspect(ctx, containerID)
if err != nil {
diagnostic["inspect_error"] = redactDiagnostic(err.Error())
return diagnostic
}
diagnostic["state"] = inspection.Status
diagnostic["running"] = inspection.Running
diagnostic["restarting"] = inspection.Restarting
diagnostic["paused"] = inspection.Paused
diagnostic["oom_killed"] = inspection.OOMKilled
diagnostic["dead"] = inspection.Dead
diagnostic["exit_code"] = inspection.ExitCode
diagnostic["error"] = redactDiagnostic(inspection.Error)
diagnostic["started_at"] = inspection.StartedAt
diagnostic["finished_at"] = inspection.FinishedAt
diagnostic["health"] = inspection.Health
diagnostic["restart_count"] = inspection.RestartCount
if logs, logsErr := s.docker.Logs(ctx, containerID, 100); logsErr == nil {
diagnostic["logs_tail"] = redactDiagnostic(logs)
} else {
diagnostic["logs_error"] = redactDiagnostic(logsErr.Error())
}
return diagnostic
}
func redactDiagnostic(value string) string {
return diagnosticSecretPattern.ReplaceAllStringFunc(value, func(match string) string {
separator := "="
if strings.Contains(match, ":") {
separator = ":"
}
parts := strings.SplitN(match, separator, 2)
return parts[0] + separator + "[REDACTED]"
})
}
func (s *service) headers(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Cache-Control", "no-store")
w.Header().Set("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'")
w.Header().Set("X-Content-Type-Options", "nosniff")
next.ServeHTTP(w, r)
})
}
func decodeJSON(body io.Reader, destination any) error {
decoder := json.NewDecoder(body)
decoder.DisallowUnknownFields()
if err := decoder.Decode(destination); err != nil {
return err
}
if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
return errors.New("multiple JSON values")
}
return nil
}
func writeJSON(w http.ResponseWriter, status int, value any) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(status)
_ = json.NewEncoder(w).Encode(value)
}
func writeProblem(w http.ResponseWriter, status int, code, message string) {
writeDiagnosticProblem(w, status, code, message, nil)
}
func writeDiagnosticProblem(w http.ResponseWriter, status int, code, message string, details map[string]any) {
writeJSON(w, status, struct {
Code string `json:"code"`
Message string `json:"message"`
Details map[string]any `json:"details,omitempty"`
}{Code: code, Message: message, Details: details})
}