Harden image admission, resource recovery and session renewal
Add exact-image compatibility policy, experimental admission, paginated build history and shared reauthentication checks. Preserve uncertain Docker creation records, bound frontend resource waits and renew runtime sessions proactively. Lazy-load and split Monaco, add regression and opt-in live Docker fault tests, and record full race and real-browser acceptance results.
This commit is contained in:
@@ -0,0 +1,407 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.misaka.ren/m1saka/dsh-web/internal/docker"
|
||||
"git.misaka.ren/m1saka/dsh-web/internal/store"
|
||||
)
|
||||
|
||||
const liveRecoveryLabel = "dsh-test.recovery-run"
|
||||
|
||||
type liveRecoveryResource struct{ kind, id string }
|
||||
type liveRecoveryFault struct {
|
||||
method, path string
|
||||
release chan struct{}
|
||||
done chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (f *liveRecoveryFault) unblock() { f.once.Do(func() { close(f.release) }) }
|
||||
|
||||
type liveRecoveryProxy struct {
|
||||
t *testing.T
|
||||
run string
|
||||
transport *http.Transport
|
||||
mu sync.Mutex
|
||||
handlers sync.WaitGroup
|
||||
resources []liveRecoveryResource
|
||||
fault *liveRecoveryFault
|
||||
}
|
||||
|
||||
func liveRequire(t *testing.T, err error) {
|
||||
t.Helper()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (p *liveRecoveryProxy) arm(method, path string, delayed bool) *liveRecoveryFault {
|
||||
f := &liveRecoveryFault{method: method, path: path, release: make(chan struct{}), done: make(chan struct{})}
|
||||
if !delayed {
|
||||
f.unblock()
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.fault = f
|
||||
p.mu.Unlock()
|
||||
p.t.Cleanup(f.unblock)
|
||||
return f
|
||||
}
|
||||
|
||||
func (p *liveRecoveryProxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
p.handlers.Add(1)
|
||||
defer p.handlers.Done()
|
||||
body, err := io.ReadAll(r.Body)
|
||||
if err != nil {
|
||||
p.t.Error(err)
|
||||
panic(http.ErrAbortHandler)
|
||||
}
|
||||
path := strings.TrimPrefix(r.URL.Path, "/v1.47")
|
||||
create := r.Method == "POST" && (path == "/containers/create" || path == "/networks/create")
|
||||
p.mu.Lock()
|
||||
allowed := r.Method == "GET"
|
||||
if create {
|
||||
var spec struct {
|
||||
Name string
|
||||
Labels map[string]string
|
||||
}
|
||||
if json.Unmarshal(body, &spec) == nil {
|
||||
name := spec.Name
|
||||
if path == "/containers/create" {
|
||||
name = r.URL.Query().Get("name")
|
||||
}
|
||||
allowed = spec.Labels[liveRecoveryLabel] == p.run && strings.Contains(name, p.run)
|
||||
}
|
||||
} else if !allowed {
|
||||
for _, resource := range p.resources {
|
||||
base := "/" + resource.kind + "/" + resource.id
|
||||
if path == base || strings.HasPrefix(path, base+"/") {
|
||||
allowed = true
|
||||
}
|
||||
}
|
||||
if strings.HasSuffix(path, "/connect") || strings.HasSuffix(path, "/disconnect") {
|
||||
var spec struct{ Container string }
|
||||
_ = json.Unmarshal(body, &spec)
|
||||
owned := false
|
||||
for _, resource := range p.resources {
|
||||
if resource.kind == "containers" && spec.Container == resource.id {
|
||||
owned = true
|
||||
}
|
||||
}
|
||||
allowed = allowed && owned
|
||||
}
|
||||
}
|
||||
var fault *liveRecoveryFault
|
||||
if p.fault != nil && p.fault.method == r.Method && p.fault.path == path {
|
||||
fault = p.fault
|
||||
p.fault = nil
|
||||
}
|
||||
p.mu.Unlock()
|
||||
if !allowed {
|
||||
p.t.Errorf("blocked out-of-scope request: %s %s", r.Method, path)
|
||||
panic(http.ErrAbortHandler)
|
||||
}
|
||||
if fault != nil {
|
||||
defer close(fault.done)
|
||||
select {
|
||||
case <-fault.release:
|
||||
default:
|
||||
conn, _, hijackErr := w.(http.Hijacker).Hijack()
|
||||
if hijackErr != nil {
|
||||
p.t.Error(hijackErr)
|
||||
return
|
||||
}
|
||||
_ = conn.Close()
|
||||
<-fault.release
|
||||
}
|
||||
}
|
||||
// A disconnected test client must not cancel the deliberately late daemon operation.
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
req := r.Clone(ctx)
|
||||
req.URL.Scheme, req.URL.Host, req.RequestURI = "http", "docker", ""
|
||||
req.Body = io.NopCloser(bytes.NewReader(body))
|
||||
if len(body) == 0 {
|
||||
req.Body = http.NoBody
|
||||
}
|
||||
resp, err := p.transport.RoundTrip(req)
|
||||
if err != nil {
|
||||
p.t.Error(err)
|
||||
panic(http.ErrAbortHandler)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
data, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
p.t.Error(err)
|
||||
panic(http.ErrAbortHandler)
|
||||
}
|
||||
if create && resp.StatusCode >= 200 && resp.StatusCode < 300 {
|
||||
var result struct{ ID string }
|
||||
if err := json.Unmarshal(data, &result); err != nil || result.ID == "" {
|
||||
p.t.Errorf("missing real create ID: %s", data)
|
||||
} else {
|
||||
kind := "containers"
|
||||
if path == "/networks/create" {
|
||||
kind = "networks"
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.resources = append(p.resources, liveRecoveryResource{kind, result.ID})
|
||||
p.mu.Unlock()
|
||||
p.t.Logf("created %s ID=%s run=%s", kind, result.ID, p.run)
|
||||
}
|
||||
}
|
||||
if fault != nil {
|
||||
p.t.Logf("dropped real response: %s %s status=%d", r.Method, path, resp.StatusCode)
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
p.t.Errorf("fault did not follow successful daemon operation: %s", data)
|
||||
}
|
||||
panic(http.ErrAbortHandler)
|
||||
}
|
||||
for key, values := range resp.Header {
|
||||
for _, value := range values {
|
||||
w.Header().Add(key, value)
|
||||
}
|
||||
}
|
||||
w.WriteHeader(resp.StatusCode)
|
||||
_, _ = w.Write(data)
|
||||
}
|
||||
|
||||
func newLiveRecovery(t *testing.T) (*API, *liveRecoveryProxy, string) {
|
||||
t.Helper()
|
||||
if os.Getenv("DSH_DOCKER_LIVE_TEST") != "1" {
|
||||
t.Skip("opt in with DSH_DOCKER_LIVE_TEST=1; uses /var/run/docker.sock and local node:24-alpine only")
|
||||
}
|
||||
t.Setenv("PANEL_CONTAINER_ID", "")
|
||||
var nonce [12]byte
|
||||
_, err := rand.Read(nonce[:])
|
||||
liveRequire(t, err)
|
||||
p := &liveRecoveryProxy{t: t, run: "live-" + hex.EncodeToString(nonce[:]), transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
return (&net.Dialer{}).DialContext(ctx, "unix", "/var/run/docker.sock")
|
||||
}}}
|
||||
real := docker.New("unix:///var/run/docker.sock")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
var image struct{ ID string }
|
||||
liveRequire(t, real.Do(ctx, "GET", "/images/node:24-alpine/json", nil, &image))
|
||||
s, err := store.Open(filepath.Join(t.TempDir(), "live.db"))
|
||||
liveRequire(t, err)
|
||||
t.Cleanup(s.Close)
|
||||
server := httptest.NewServer(p)
|
||||
t.Cleanup(func() {
|
||||
server.Close()
|
||||
p.handlers.Wait()
|
||||
p.transport.CloseIdleConnections()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
p.mu.Lock()
|
||||
resources := append([]liveRecoveryResource(nil), p.resources...)
|
||||
p.mu.Unlock()
|
||||
for _, kind := range []string{"containers", "networks"} {
|
||||
for _, resource := range resources {
|
||||
if resource.kind != kind {
|
||||
continue
|
||||
}
|
||||
path := "/" + kind + "/" + resource.id
|
||||
err := real.Do(ctx, "DELETE", path+"?force=1", nil, nil)
|
||||
if err != nil && !dockerNotFound(err) {
|
||||
t.Errorf("cleanup %s: %v", path, err)
|
||||
}
|
||||
if kind == "containers" {
|
||||
path += "/json"
|
||||
}
|
||||
if err := real.Do(ctx, "GET", path, nil, nil); !dockerNotFound(err) {
|
||||
t.Errorf("residual %s: %v", path, err)
|
||||
} else {
|
||||
t.Logf("cleanup confirmed absent %s %s", kind, resource.id)
|
||||
}
|
||||
}
|
||||
}
|
||||
filters, _ := json.Marshal(map[string][]string{"label": {liveRecoveryLabel + "=" + p.run}})
|
||||
for _, path := range []string{"/containers/json?all=1&", "/networks?"} {
|
||||
var residual []json.RawMessage
|
||||
if err := real.Do(ctx, "GET", path+"filters="+url.QueryEscape(string(filters)), nil, &residual); err != nil || len(residual) != 0 {
|
||||
t.Errorf("label residual check: %s count=%d err=%v", p.run, len(residual), err)
|
||||
} else {
|
||||
t.Logf("label residual check empty: %s %s", p.run, path)
|
||||
}
|
||||
}
|
||||
})
|
||||
return &API{s: s, root: t.TempDir(), docker: docker.New(server.URL)}, p, image.ID
|
||||
}
|
||||
|
||||
func liveContainerSpec(run, image string, labels map[string]string) map[string]any {
|
||||
labels[liveRecoveryLabel] = run
|
||||
return map[string]any{"Image": image, "Cmd": []string{"sleep", "600"}, "Labels": labels, "HostConfig": map[string]any{"NetworkMode": "none", "Memory": 64 << 20, "PidsLimit": 32}}
|
||||
}
|
||||
|
||||
func TestDockerLiveRecoveryNetwork(t *testing.T) {
|
||||
for _, panelMode := range []string{"unset", "unconnected", "connected-id", "connected-name"} {
|
||||
t.Run(panelMode, func(t *testing.T) {
|
||||
a, p, image := newLiveRecovery(t)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
|
||||
defer cancel()
|
||||
id := p.run
|
||||
liveRequire(t, a.setRuntimeCreatePending(id, "network_pending", true))
|
||||
p.arm("POST", "/networks/create", false)
|
||||
err := a.docker.Do(ctx, "POST", "/networks/create", map[string]any{"Name": runtimeNetwork(id), "Driver": "bridge", "Labels": map[string]string{"dsh-mgr.instance": id, liveRecoveryLabel: p.run}}, nil)
|
||||
if err == nil {
|
||||
t.Fatal("create response was not lost")
|
||||
}
|
||||
var network struct{ ID string }
|
||||
liveRequire(t, a.docker.Do(ctx, "GET", "/networks/"+runtimeNetwork(id), nil, &network))
|
||||
if panelMode != "unset" {
|
||||
name := p.run + "-panel"
|
||||
var panel struct{ ID string }
|
||||
spec := liveContainerSpec(p.run, image, map[string]string{})
|
||||
if strings.HasPrefix(panelMode, "connected") {
|
||||
spec["HostConfig"].(map[string]any)["NetworkMode"] = network.ID
|
||||
}
|
||||
liveRequire(t, a.docker.Do(ctx, "POST", "/containers/create?name="+name, spec, &panel))
|
||||
liveRequire(t, a.docker.Do(ctx, "POST", "/containers/"+panel.ID+"/start", nil, nil))
|
||||
t.Setenv("PANEL_CONTAINER_ID", panel.ID)
|
||||
if panelMode == "connected-name" {
|
||||
t.Setenv("PANEL_CONTAINER_ID", name)
|
||||
}
|
||||
var endpoints struct{ Containers map[string]json.RawMessage }
|
||||
liveRequire(t, a.docker.Do(ctx, "GET", "/networks/"+network.ID, nil, &endpoints))
|
||||
_, connected := endpoints.Containers[panel.ID]
|
||||
if connected != strings.HasPrefix(panelMode, "connected") {
|
||||
t.Fatalf("unexpected real panel connection: %v", connected)
|
||||
}
|
||||
}
|
||||
p.arm("DELETE", "/networks/"+network.ID, false)
|
||||
if err := a.removeRuntimeNetwork(ctx, id); err == nil {
|
||||
t.Fatal("delete response was not lost")
|
||||
}
|
||||
var recorded string
|
||||
liveRequire(t, a.s.DB.QueryRow("SELECT network_id FROM runtime_networks WHERE instance_id=?", id).Scan(&recorded))
|
||||
if recorded != network.ID {
|
||||
t.Fatalf("lost recorded identity: %s", recorded)
|
||||
}
|
||||
liveRequire(t, a.removeRuntimeNetwork(ctx, id))
|
||||
_, pending, err := a.pendingSince(id, "network_pending")
|
||||
liveRequire(t, err)
|
||||
if pending {
|
||||
t.Fatal("pending did not converge")
|
||||
}
|
||||
if err := a.docker.Do(ctx, "GET", "/networks/"+network.ID, nil, nil); !dockerNotFound(err) {
|
||||
t.Fatalf("network still exists: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDockerLiveRecoveryLateContainer(t *testing.T) {
|
||||
for _, build := range []bool{false, true} {
|
||||
name := "runtime"
|
||||
if build {
|
||||
name = "build-check"
|
||||
}
|
||||
t.Run(name, func(t *testing.T) {
|
||||
a, p, image := newLiveRecovery(t)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
|
||||
defer cancel()
|
||||
i := store.Instance{ID: p.run, UID: p.run, ImageID: image, Status: "error", Created: time.Now()}
|
||||
name := runtimeNetwork(i.ID)
|
||||
labels := map[string]string{"dsh-mgr.instance": i.ID, "dsh-mgr.owner": i.UID, "dsh-mgr.role": "dsh"}
|
||||
if build {
|
||||
name = buildCheckName(i.ID)
|
||||
labels = map[string]string{"dsh-mgr.build": i.ID, "dsh-mgr.role": "build-check"}
|
||||
liveRequire(t, a.ensureBuildChecks())
|
||||
_, err := a.s.DB.Exec("INSERT INTO image_build_checks(build_id,image_id,created_at) VALUES(?,?,?)", i.ID, image, time.Now().Add(-time.Hour).Format(time.RFC3339Nano))
|
||||
liveRequire(t, err)
|
||||
} else {
|
||||
liveRequire(t, a.s.AddInstance(i))
|
||||
liveRequire(t, a.setRuntimeCreatePending(i.ID, "container_pending", true))
|
||||
}
|
||||
fault := p.arm("POST", "/containers/create", true)
|
||||
if err := a.docker.Do(ctx, "POST", "/containers/create?name="+name, liveContainerSpec(p.run, image, labels), nil); err == nil {
|
||||
t.Fatal("expected disconnected create")
|
||||
}
|
||||
if _, err := a.docker.Inspect(ctx, name); !dockerNotFound(err) {
|
||||
t.Fatalf("expected real initial 404: %v", err)
|
||||
}
|
||||
for attempt := 0; attempt < 2; attempt++ {
|
||||
if build {
|
||||
if err := a.recoverBuildChecks(ctx); !errors.Is(err, errBuildCheckUncertain) {
|
||||
t.Fatalf("uncertainty lost: %v", err)
|
||||
}
|
||||
var count int
|
||||
liveRequire(t, a.s.DB.QueryRow("SELECT count(*) FROM image_build_checks WHERE build_id=?", i.ID).Scan(&count))
|
||||
if count != 1 {
|
||||
t.Fatal("build record lost")
|
||||
}
|
||||
} else {
|
||||
if err := a.recoverRuntimeContainer(ctx, &i); err == nil {
|
||||
t.Fatal("uncertainty lost")
|
||||
}
|
||||
_, pending, err := a.pendingSince(i.ID, "container_pending")
|
||||
liveRequire(t, err)
|
||||
if !pending {
|
||||
t.Fatal("pending lost")
|
||||
}
|
||||
}
|
||||
}
|
||||
fault.unblock()
|
||||
select {
|
||||
case <-fault.done:
|
||||
case <-ctx.Done():
|
||||
t.Fatal(ctx.Err())
|
||||
}
|
||||
container, err := a.docker.Inspect(ctx, name)
|
||||
liveRequire(t, err)
|
||||
p.arm("DELETE", "/containers/"+container.ID, false)
|
||||
if build {
|
||||
if err := a.recoverBuildChecks(ctx); err == nil {
|
||||
t.Fatal("delete response was not lost")
|
||||
}
|
||||
var recorded string
|
||||
liveRequire(t, a.s.DB.QueryRow("SELECT container_id FROM image_build_checks WHERE build_id=?", i.ID).Scan(&recorded))
|
||||
if recorded != container.ID {
|
||||
t.Fatal("build identity lost")
|
||||
}
|
||||
liveRequire(t, a.recoverBuildChecks(ctx))
|
||||
var count int
|
||||
liveRequire(t, a.s.DB.QueryRow("SELECT count(*) FROM image_build_checks").Scan(&count))
|
||||
if count != 0 {
|
||||
t.Fatal("build record retained")
|
||||
}
|
||||
} else {
|
||||
liveRequire(t, a.recoverRuntimeContainer(ctx, &i))
|
||||
_, pending, err := a.pendingSince(i.ID, "container_pending")
|
||||
liveRequire(t, err)
|
||||
if pending {
|
||||
t.Fatal("runtime pending did not converge")
|
||||
}
|
||||
stored, err := a.s.GetInstance(i.ID)
|
||||
liveRequire(t, err)
|
||||
if stored.ContainerID != container.ID {
|
||||
t.Fatal("runtime identity lost")
|
||||
}
|
||||
if err := a.removeRuntimeContainer(ctx, i.ContainerID); err == nil {
|
||||
t.Fatal("delete response was not lost")
|
||||
}
|
||||
liveRequire(t, a.removeRuntimeContainer(ctx, i.ContainerID))
|
||||
}
|
||||
if _, err := a.docker.Inspect(ctx, container.ID); !dockerNotFound(err) {
|
||||
t.Fatalf("late resource not reclaimed: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user