package sourceagent import ( "bytes" "context" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" "net" "net/http" "net/url" "os" "os/exec" "path/filepath" "strconv" "strings" "time" ) type DockerController struct { socket string composeBinary string http *http.Client apiPrefix string engineVersion string apiVersion string } func NewDockerController(ctx context.Context, socket, composeBinary string) (*DockerController, error) { socket = strings.TrimSpace(socket) if socket == "" { socket = "/var/run/docker.sock" } if composeBinary == "" { composeBinary = "docker" } transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { return (&net.Dialer{Timeout: 5 * time.Second}).DialContext(ctx, "unix", socket) }} d := &DockerController{socket: socket, composeBinary: composeBinary, http: &http.Client{Transport: transport, Timeout: 2 * time.Minute}} if err := d.negotiate(ctx); err != nil { return nil, err } return d, nil } func (d *DockerController) negotiate(ctx context.Context) error { var version struct { Version string `json:"Version"` APIVersion string `json:"ApiVersion"` } if err := d.json(ctx, http.MethodGet, "/version", nil, &version); err != nil { return fmt.Errorf("docker socket unavailable: %w", err) } if strings.TrimSpace(version.APIVersion) == "" { return errors.New("docker engine did not report ApiVersion") } d.engineVersion, d.apiVersion = version.Version, version.APIVersion d.apiPrefix = "/v" + version.APIVersion return nil } func (d *DockerController) request(ctx context.Context, method, path string, body any) (*http.Response, error) { var r io.Reader if body != nil { data, err := json.Marshal(body) if err != nil { return nil, err } r = bytes.NewReader(data) } req, err := http.NewRequestWithContext(ctx, method, "http://docker"+path, r) if err != nil { return nil, err } if body != nil { req.Header.Set("Content-Type", "application/json") } return d.http.Do(req) } func (d *DockerController) json(ctx context.Context, method, path string, body, out any) error { resp, err := d.request(ctx, method, path, body) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode/100 != 2 { data, _ := io.ReadAll(io.LimitReader(resp.Body, 8192)) return fmt.Errorf("docker API %s %s: HTTP %d: %s", method, path, resp.StatusCode, strings.TrimSpace(string(data))) } if out == nil { _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<20)) return nil } return json.NewDecoder(io.LimitReader(resp.Body, 16<<20)).Decode(out) } func (d *DockerController) ComposeAvailable(ctx context.Context) bool { path, err := exec.LookPath(d.composeBinary) if err != nil { return false } cctx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() cmd := exec.CommandContext(cctx, path, "compose", "version", "--short") cmd.Env = append(os.Environ(), "DOCKER_HOST=unix://"+d.socket) return cmd.Run() == nil } func (d *DockerController) Status(ctx context.Context) DockerControllerStatus { st := DockerControllerStatus{Enabled: true, Socket: d.socket, Reachable: true, EngineVersion: d.engineVersion, APIVersion: d.apiVersion, ComposeAvailable: d.ComposeAvailable(ctx), LastRefresh: time.Now().UTC()} inv, err := d.Inventory(ctx) if err != nil { st.Reachable = false st.LastError = err.Error() return st } // Heartbeats are intentionally bounded. The dashboard needs a useful // inventory sample, not every Docker metadata field from a very large host. if len(inv.Containers) > 100 { inv.Containers = inv.Containers[:100] } if len(inv.Networks) > 100 { inv.Networks = inv.Networks[:100] } if len(inv.Volumes) > 100 { inv.Volumes = inv.Volumes[:100] } st.Inventory = inv st.Containers = len(inv.Containers) st.Networks = len(inv.Networks) st.Volumes = len(inv.Volumes) for _, c := range inv.Containers { if c.State == "running" { st.Running++ } if c.Health == "unhealthy" { st.Unhealthy++ } } return st } func (d *DockerController) Inventory(ctx context.Context) (DockerInventory, error) { var rawContainers []struct { ID string `json:"Id"` Names []string `json:"Names"` Image string `json:"Image"` State string `json:"State"` Status string `json:"Status"` Labels map[string]string `json:"Labels"` } if err := d.json(ctx, http.MethodGet, d.apiPrefix+"/containers/json?all=1", nil, &rawContainers); err != nil { return DockerInventory{}, err } out := DockerInventory{Containers: make([]DockerContainerSummary, 0, len(rawContainers))} for _, c := range rawContainers { health := "" lower := strings.ToLower(c.Status) if strings.Contains(lower, "(healthy)") { health = "healthy" } else if strings.Contains(lower, "(unhealthy)") { health = "unhealthy" } else if strings.Contains(lower, "health: starting") { health = "starting" } // Labels may contain large Compose/Kubernetes metadata and are not needed // by the controller dashboard. Omitting them keeps the signed heartbeat // inventory comfortably inside its 64 KiB persistence budget. out.Containers = append(out.Containers, DockerContainerSummary{ID: c.ID, Names: c.Names, Image: c.Image, State: c.State, Status: c.Status, Health: health}) } var rawNetworks []struct { ID string `json:"Id"` Name, Driver, Scope string } if err := d.json(ctx, http.MethodGet, d.apiPrefix+"/networks", nil, &rawNetworks); err == nil { for _, n := range rawNetworks { out.Networks = append(out.Networks, DockerNetworkSummary{ID: n.ID, Name: n.Name, Driver: n.Driver, Scope: n.Scope}) } } var rawVolumes struct { Volumes []struct{ Name, Driver, Scope string } } if err := d.json(ctx, http.MethodGet, d.apiPrefix+"/volumes", nil, &rawVolumes); err == nil { for _, v := range rawVolumes.Volumes { out.Volumes = append(out.Volumes, DockerVolumeSummary{Name: v.Name, Driver: v.Driver, Scope: v.Scope}) } } return out, nil } func (d *DockerController) Execute(ctx context.Context, claim ControllerClaim) ControllerJobResult { started := time.Now().UTC() result := ControllerJobResult{JobID: claim.Job.ID, Status: ControllerJobSucceeded, Result: map[string]any{}} if err := validateControllerJob(claim.Job, claim.Policy); err != nil { result.Status = ControllerJobFailed result.Error = err.Error() return result } if claim.Job.DryRun || claim.Policy.DryRun { result.Result = map[string]any{"dry_run": true, "validated": true, "kind": claim.Job.Kind, "parameters": claim.Job.Parameters} result.DurationMS = time.Since(started).Milliseconds() return result } var err error switch claim.Job.Kind { case "inventory_refresh": var inv DockerInventory inv, err = d.Inventory(ctx) if err == nil { result.Result = map[string]any{"inventory": inv} } case "container_start", "container_stop", "container_restart", "container_remove": err = d.containerLifecycle(ctx, claim.Job, claim.Policy, result.Result) case "container_create": err = d.containerCreate(ctx, claim.Job, claim.Policy, result.Result) case "network_create", "network_remove": err = d.networkAction(ctx, claim.Job, claim.Policy, result.Result) case "volume_create", "volume_remove": err = d.volumeAction(ctx, claim.Job, claim.Policy, result.Result) case "compose_up", "compose_down", "compose_restart", "compose_pull", "compose_ps", "compute_capacity_compose", "compose_smoke_test": err = d.composeAction(ctx, claim.Job, claim.Policy, result.Result) case "evidence_http_probe": err = d.evidenceProbe(ctx, claim.Job, claim.Policy, result.Result) case "health_recovery": job := claim.Job job.Kind = "container_restart" err = d.containerLifecycle(ctx, job, claim.Policy, result.Result) default: err = errors.New("unsupported controller action") } if err != nil { result.Status = ControllerJobFailed result.Error = err.Error() } result.DurationMS = time.Since(started).Milliseconds() return result } func validateControllerJob(job ControllerJob, policy ControllerPolicy) error { if !policy.Enabled { return errors.New("docker controller master switch disabled") } if job.Autonomous && !policy.AutonomousEnabled { return errors.New("autonomous docker controller actions disabled") } if job.Autonomous { switch job.Kind { case "evidence_http_probe", "health_recovery", "compute_capacity_compose", "compose_smoke_test": default: return errors.New("job kind is not permitted for autonomous controller execution") } } if isDestructiveControllerKind(job.Kind) && !policy.AllowDestructive { return errors.New("destructive docker controller actions are disabled") } return nil } func isDestructiveControllerKind(kind string) bool { switch kind { case "container_remove", "network_remove", "volume_remove", "compose_down": return true } return false } func (d *DockerController) containerLifecycle(ctx context.Context, job ControllerJob, policy ControllerPolicy, out map[string]any) error { name := paramString(job.Parameters, "container") if name == "" { return errors.New("container is required") } if protectedName(name, policy.ProtectedContainers) { return errors.New("container is protected by controller policy") } id := url.PathEscape(name) var method, path string = http.MethodPost, d.apiPrefix + "/containers/" + id + "/" + strings.TrimPrefix(job.Kind, "container_") if job.Kind == "container_stop" || job.Kind == "container_restart" { path += "?t=15" } if job.Kind == "container_remove" { method = http.MethodDelete path = d.apiPrefix + "/containers/" + id + "?force=false&v=false" } if err := d.json(ctx, method, path, nil, nil); err != nil { return err } out["container"] = name out["action"] = job.Kind return nil } func (d *DockerController) containerCreate(ctx context.Context, job ControllerJob, policy ControllerPolicy, out map[string]any) error { image := paramString(job.Parameters, "image") name := paramString(job.Parameters, "name") if image == "" || name == "" { return errors.New("image and name are required") } if !imageAllowed(image, policy.AllowedImages) { return errors.New("image is not allowed by controller policy") } if protectedName(name, policy.ProtectedContainers) { return errors.New("container name is protected") } network := paramString(job.Parameters, "network") if network == "" { network = "bridge" } if network == "host" || network == "container" || strings.HasPrefix(network, "container:") { return errors.New("host/container network modes are forbidden") } cmd := paramStrings(job.Parameters, "command", 32, 2048) env := paramStrings(job.Parameters, "env", 64, 4096) mounts, err := namedVolumeMounts(job.Parameters["volumes"], policy) if err != nil { return err } memory := paramInt64(job.Parameters, "memory_bytes", 512<<20) if memory < 32<<20 { memory = 32 << 20 } if memory > 8<<30 { memory = 8 << 30 } nano := paramInt64(job.Parameters, "nano_cpus", 1_000_000_000) if nano < 100_000_000 { nano = 100_000_000 } if nano > 4_000_000_000 { nano = 4_000_000_000 } body := map[string]any{"Image": image, "Cmd": cmd, "Env": env, "Tty": false, "HostConfig": map[string]any{"NetworkMode": network, "Privileged": false, "ReadonlyRootfs": true, "CapDrop": []string{"ALL"}, "SecurityOpt": []string{"no-new-privileges:true"}, "Memory": memory, "NanoCpus": nano, "Mounts": mounts}} var created struct { ID string `json:"Id"` } path := d.apiPrefix + "/containers/create?name=" + url.QueryEscape(name) if err := d.json(ctx, http.MethodPost, path, body, &created); err != nil { return err } out["container_id"] = created.ID out["name"] = name if paramBool(job.Parameters, "start", true) { if err := d.json(ctx, http.MethodPost, d.apiPrefix+"/containers/"+url.PathEscape(created.ID)+"/start", nil, nil); err != nil { return err } out["started"] = true } return nil } func (d *DockerController) networkAction(ctx context.Context, job ControllerJob, policy ControllerPolicy, out map[string]any) error { name := paramString(job.Parameters, "network") if name == "" { return errors.New("network is required") } if protectedName(name, policy.ProtectedNetworks) { return errors.New("network is protected") } if job.Kind == "network_create" { driver := paramString(job.Parameters, "driver") if driver == "" { driver = "bridge" } if driver != "bridge" { return errors.New("only bridge networks may be created by the controller") } var r struct{ ID string } if err := d.json(ctx, http.MethodPost, d.apiPrefix+"/networks/create", map[string]any{"Name": name, "Driver": driver, "CheckDuplicate": true, "Internal": paramBool(job.Parameters, "internal", false)}, &r); err != nil { return err } out["network_id"] = r.ID } else { if err := d.json(ctx, http.MethodDelete, d.apiPrefix+"/networks/"+url.PathEscape(name), nil, nil); err != nil { return err } } out["network"] = name return nil } func (d *DockerController) volumeAction(ctx context.Context, job ControllerJob, policy ControllerPolicy, out map[string]any) error { name := paramString(job.Parameters, "volume") if name == "" { return errors.New("volume is required") } if protectedName(name, policy.ProtectedVolumes) { return errors.New("volume is protected") } if job.Kind == "volume_create" { var r map[string]any if err := d.json(ctx, http.MethodPost, d.apiPrefix+"/volumes/create", map[string]any{"Name": name, "Driver": "local"}, &r); err != nil { return err } out["volume"] = r } else { if err := d.json(ctx, http.MethodDelete, d.apiPrefix+"/volumes/"+url.PathEscape(name)+"?force=false", nil, nil); err != nil { return err } out["volume"] = name } return nil } func (d *DockerController) composeAction(ctx context.Context, job ControllerJob, policy ControllerPolicy, out map[string]any) error { if !d.ComposeAvailable(ctx) { return errors.New("docker compose CLI is not available in this controller agent") } file := paramString(job.Parameters, "compose_file") safe, err := allowedComposeFile(file, policy.AllowedComposeRoots) if err != nil { return err } project := paramString(job.Parameters, "project") if project == "" { project = "brain-controller" } service := paramString(job.Parameters, "service") run := func(args ...string) (string, error) { base := []string{"compose", "-f", safe, "-p", project} base = append(base, args...) cmd := exec.CommandContext(ctx, d.composeBinary, base...) cmd.Env = append(os.Environ(), "DOCKER_HOST=unix://"+d.socket) data, err := cmd.CombinedOutput() if len(data) > 256<<10 { data = data[:256<<10] } return strings.TrimSpace(string(data)), err } out["compose_file"] = safe out["project"] = project if job.Kind == "compose_smoke_test" { args := []string{"up", "-d", "--wait", "--wait-timeout", "120"} if service != "" { args = append(args, service) } up, err := run(args...) out["up_output"] = up if err != nil { return fmt.Errorf("compose smoke-test up failed: %w: %s", err, up) } ps, psErr := run("ps", "--format", "json") out["ps_output"] = ps out["smoke_test_passed"] = psErr == nil if paramBool(job.Parameters, "cleanup", false) { if !policy.AllowDestructive { return errors.New("smoke-test cleanup requires destructive actions to be enabled") } down, downErr := run("down", "--remove-orphans") out["cleanup_output"] = down if downErr != nil && psErr == nil { psErr = downErr } } if psErr != nil { return fmt.Errorf("compose smoke-test verification failed: %w", psErr) } return nil } action := strings.TrimPrefix(job.Kind, "compose_") if job.Kind == "compute_capacity_compose" { action = "up" } var args []string switch action { case "up": args = []string{"up", "-d"} case "down": args = []string{"down", "--remove-orphans"} case "restart": args = []string{"restart"} case "pull": args = []string{"pull"} case "ps": args = []string{"ps", "--format", "json"} default: return errors.New("unsupported compose action") } if service != "" && action != "down" { args = append(args, service) } data, err := run(args...) out["output"] = data out["action"] = action if err != nil { return fmt.Errorf("docker compose %s failed: %w: %s", action, err, data) } return nil } func (d *DockerController) evidenceProbe(ctx context.Context, job ControllerJob, policy ControllerPolicy, out map[string]any) error { target := paramString(job.Parameters, "url") u, err := url.Parse(target) if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") { return errors.New("evidence probe requires absolute http(s) url") } if err := validateEvidenceProbeTarget(ctx, u); err != nil { return err } image := paramString(job.Parameters, "image") if image == "" { image = "curlimages/curl:8.11.1" } if !imageAllowed(image, policy.AllowedImages) { return errors.New("evidence probe image is not allowed") } _ = d.pullImage(ctx, image) name := "brain-evidence-" + strings.TrimPrefix(randomID("p"), "p-") marker := "__BRAIN_PROBE_META__" cmd := []string{"--silent", "--show-error", "--location", "--max-time", "25", "--connect-timeout", "10", "--max-filesize", "524288", "--output", "-", "--write-out", "\n" + marker + "%{http_code}|%{url_effective}|%{content_type}|%{size_download}", target} body := map[string]any{"Image": image, "Cmd": cmd, "Tty": true, "AttachStdout": true, "AttachStderr": true, "HostConfig": map[string]any{"NetworkMode": "bridge", "AutoRemove": false, "Privileged": false, "ReadonlyRootfs": true, "CapDrop": []string{"ALL"}, "SecurityOpt": []string{"no-new-privileges:true"}, "Memory": int64(128 << 20), "NanoCpus": int64(500_000_000)}} var created struct { ID string `json:"Id"` } if err := d.json(ctx, http.MethodPost, d.apiPrefix+"/containers/create?name="+url.QueryEscape(name), body, &created); err != nil { return err } defer d.json(context.Background(), http.MethodDelete, d.apiPrefix+"/containers/"+url.PathEscape(created.ID)+"?force=true&v=false", nil, nil) if err := d.json(ctx, http.MethodPost, d.apiPrefix+"/containers/"+url.PathEscape(created.ID)+"/start", nil, nil); err != nil { return err } var wait struct{ StatusCode int } if err := d.json(ctx, http.MethodPost, d.apiPrefix+"/containers/"+url.PathEscape(created.ID)+"/wait?condition=not-running", nil, &wait); err != nil { return err } resp, err := d.request(ctx, http.MethodGet, d.apiPrefix+"/containers/"+url.PathEscape(created.ID)+"/logs?stdout=1&stderr=1&tail=all", nil) if err != nil { return err } defer resp.Body.Close() raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) text := string(raw) idx := strings.LastIndex(text, "\n"+marker) bodyBytes := raw meta := "" if idx >= 0 { bodyBytes = raw[:idx] meta = strings.TrimSpace(text[idx+1+len(marker):]) } sum := sha256.Sum256(bodyBytes) out["url"] = target out["body_sha256"] = hex.EncodeToString(sum[:]) out["captured_bytes"] = len(bodyBytes) out["exit_code"] = wait.StatusCode if meta != "" { parts := strings.Split(meta, "|") if len(parts) >= 4 { out["http_status"] = parts[0] out["effective_url"] = parts[1] out["content_type"] = parts[2] out["download_bytes"] = parts[3] } } if wait.StatusCode != 0 { return fmt.Errorf("evidence probe exited with code %d", wait.StatusCode) } return nil } func validateEvidenceProbeTarget(ctx context.Context, u *url.URL) error { if u == nil || u.User != nil { return errors.New("evidence probe target must not contain credentials") } host := strings.TrimSpace(strings.ToLower(u.Hostname())) if host == "" || host == "localhost" || strings.HasSuffix(host, ".localhost") || strings.HasSuffix(host, ".local") || strings.HasSuffix(host, ".internal") { return errors.New("evidence probe target must be a public host") } unsafeIP := func(ip net.IP) bool { return ip == nil || ip.IsLoopback() || ip.IsPrivate() || ip.IsUnspecified() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() } if literal := net.ParseIP(host); literal != nil { if unsafeIP(literal) { return errors.New("evidence probe target must not use a private or local IP") } return nil } lookupCtx, cancel := context.WithTimeout(ctx, 4*time.Second) defer cancel() addresses, err := net.DefaultResolver.LookupIPAddr(lookupCtx, host) if err != nil || len(addresses) == 0 { return errors.New("evidence probe target could not be resolved safely") } for _, address := range addresses { if unsafeIP(address.IP) { return errors.New("evidence probe target resolves to a private or local address") } } return nil } func (d *DockerController) pullImage(ctx context.Context, image string) error { repository, tag := splitDockerImageReference(image) path := d.apiPrefix + "/images/create?fromImage=" + url.QueryEscape(repository) if tag != "" { path += "&tag=" + url.QueryEscape(tag) } resp, err := d.request(ctx, http.MethodPost, path, nil) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode/100 != 2 { return fmt.Errorf("docker image pull HTTP %d", resp.StatusCode) } _, err = io.Copy(io.Discard, io.LimitReader(resp.Body, 16<<20)) return err } func splitDockerImageReference(image string) (repository, tag string) { image = strings.TrimSpace(image) if image == "" { return "", "" } // Digests are complete references and must not be split into tag syntax. if strings.Contains(image, "@") { return image, "" } lastSlash := strings.LastIndex(image, "/") lastColon := strings.LastIndex(image, ":") // A colon before the final slash belongs to a registry port, not to a tag. if lastColon > lastSlash { return image[:lastColon], image[lastColon+1:] } return image, "" } func imageAllowed(image string, allowed []string) bool { for _, a := range allowed { a = strings.TrimSpace(a) if a == "*" || image == a { return true } // Prefix allow rules must be explicit. A trailing ':', '/', or '@' // expresses "all tags/children/digests"; plain names stay exact so an // allowlist entry like "vendor/tool" cannot also permit // "vendor/tool-malicious". if (strings.HasSuffix(a, ":") || strings.HasSuffix(a, "/") || strings.HasSuffix(a, "@")) && strings.HasPrefix(image, a) { return true } } return false } func protectedName(name string, patterns []string) bool { name = strings.TrimPrefix(strings.TrimSpace(name), "/") for _, p := range patterns { p = strings.TrimSpace(p) if p == "" { continue } if ok, _ := filepath.Match(p, name); ok || strings.EqualFold(p, name) { return true } } return false } func allowedComposeFile(file string, roots []string) (string, error) { if strings.TrimSpace(file) == "" { return "", errors.New("compose_file is required") } abs, err := filepath.Abs(file) if err != nil { return "", err } resolved, err := filepath.EvalSymlinks(abs) if err != nil { return "", err } for _, root := range roots { rr, err := filepath.EvalSymlinks(root) if err != nil { rr = filepath.Clean(root) } rel, err := filepath.Rel(rr, resolved) if err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) { return resolved, nil } } return "", errors.New("compose file is outside allowed compose roots") } func namedVolumeMounts(raw any, policy ControllerPolicy) ([]map[string]any, error) { arr, ok := raw.([]any) if !ok && raw != nil { return nil, errors.New("volumes must be an array") } var out []map[string]any for _, item := range arr { m, ok := item.(map[string]any) if !ok { return nil, errors.New("invalid volume mount") } source := strings.TrimSpace(fmt.Sprint(m["source"])) target := strings.TrimSpace(fmt.Sprint(m["target"])) if source == "" || target == "" || !strings.HasPrefix(target, "/") { return nil, errors.New("named volume source and absolute target are required") } if protectedName(source, policy.ProtectedVolumes) { return nil, errors.New("volume is protected") } out = append(out, map[string]any{"Type": "volume", "Source": source, "Target": target, "ReadOnly": toBool(m["read_only"], false)}) } return out, nil } func paramString(m map[string]any, key string) string { if m == nil { return "" } return strings.TrimSpace(fmt.Sprint(m[key])) } func paramBool(m map[string]any, key string, def bool) bool { if m == nil { return def } v, ok := m[key] if !ok { return def } return toBool(v, def) } func toBool(v any, def bool) bool { switch x := v.(type) { case bool: return x case string: b, e := strconv.ParseBool(x) if e == nil { return b } case float64: return x != 0 } return def } func paramInt64(m map[string]any, key string, def int64) int64 { if m == nil { return def } switch v := m[key].(type) { case float64: return int64(v) case int: return int64(v) case int64: return v case string: n, e := strconv.ParseInt(v, 10, 64) if e == nil { return n } } return def } func paramStrings(m map[string]any, key string, max, each int) []string { if m == nil { return nil } raw, ok := m[key].([]any) if !ok { if ss, ok := m[key].([]string); ok { return ss } return nil } if len(raw) > max { raw = raw[:max] } out := make([]string, 0, len(raw)) for _, v := range raw { s := fmt.Sprint(v) if len(s) > each { s = s[:each] } out = append(out, s) } return out }