Fix/gateway stream keepalive

.github/workflows/dev-containers.yml

@@ -20,6 +20,7 @@ on:
         options:
           - all
           - server
+          - agent
           - web
           - workspace
           - workspace-lite
@@ -91,6 +92,14 @@ jobs:
             workspace_tags: "false"
             build_args: "ZAKURA_AGENT_VERSION=${{ github.sha }}"
             target: ""
+          - name: agent
+            image: zakura-agent-dev
+            dockerfile: docker/agent/Dockerfile
+            context: .
+            timeout: 60
+            workspace_tags: "false"
+            build_args: "ZAKURA_AGENT_VERSION=${{ github.sha }}"
+            target: ""
           - name: web
             image: zakura-web-dev
             dockerfile: docker/web/Dockerfile

.github/workflows/release-containers.yml

@@ -19,6 +19,7 @@ on:
           - all
           - app
           - server
+          - agent
           - web
           - workspace
 
@@ -100,6 +101,13 @@ jobs:
             timeout: 60
             workspace_tags: "false"
             build_args: "ZAKURA_AGENT_VERSION=${{ github.sha }}"
+          - name: agent
+            image: zakura-agent
+            dockerfile: docker/agent/Dockerfile
+            context: .
+            timeout: 60
+            workspace_tags: "false"
+            build_args: "ZAKURA_AGENT_VERSION=${{ github.sha }}"
           - name: web
             image: zakura-web
             dockerfile: docker/web/Dockerfile
@@ -246,7 +254,7 @@ jobs:
             echo "### Pull (Docker Hub)"
             echo
             echo '```bash'
-            for img in zakura zakura-server zakura-web zakura-workspace; do
+            for img in zakura zakura-server zakura-agent zakura-web zakura-workspace; do
               if [[ -f "digests/${img}.txt" ]]; then
                 echo "docker pull ${DH_USER}/${img}:${VERSION}"
               fi
@@ -256,7 +264,7 @@ jobs:
             echo "### Pull (GHCR)"
             echo
             echo '```bash'
-            for img in zakura zakura-server zakura-web zakura-workspace; do
+            for img in zakura zakura-server zakura-agent zakura-web zakura-workspace; do
               if [[ -f "digests/${img}.txt" ]]; then
                 echo "docker pull ${REGISTRY}/${OWNER}/${img}:${VERSION}"
               fi
@@ -265,7 +273,7 @@ jobs:
             echo
             echo "### Digests"
             echo
-            for img in zakura zakura-server zakura-web zakura-workspace; do
+            for img in zakura zakura-server zakura-agent zakura-web zakura-workspace; do
               if [[ -f "digests/${img}.txt" ]]; then
                 digest="$(tr -d '[:space:]' < "digests/${img}.txt")"
                 echo "- \`${DH_USER}/${img}@${digest}\`"
@@ -283,6 +291,9 @@ jobs:
             if [[ -f digests/zakura-server.txt ]]; then
               echo "| \`${DH_USER}/zakura-server\` / \`${REGISTRY}/${OWNER}/zakura-server\` | \`${VERSION}\`, minor/major, \`latest\` |"
             fi
+            if [[ -f digests/zakura-agent.txt ]]; then
+              echo "| \`${DH_USER}/zakura-agent\` / \`${REGISTRY}/${OWNER}/zakura-agent\` | \`${VERSION}\`, minor/major, \`latest\` |"
+            fi
             if [[ -f digests/zakura-web.txt ]]; then
               echo "| \`${DH_USER}/zakura-web\` / \`${REGISTRY}/${OWNER}/zakura-web\` | \`${VERSION}\`, minor/major, \`latest\` |"
             fi

apps/agent/bootstrap/bootstrap.go

@@ -0,0 +1,79 @@
+// Package bootstrap handles a local installation's private node enrollment
+// material. Neither bootstrap secrets nor node tokens belong in logs or UI.
+package bootstrap
+
+import (
+	"context"
+	"crypto/rand"
+	"errors"
+	"io/fs"
+
+	"github.com/Moonrend/Zakura/apps/agent/internal/statefs"
+	"github.com/Moonrend/Zakura/apps/agent/workspace"
+)
+
+type Installation struct {
+	Version int    `json:"version"`
+	ID      string `json:"id"`
+	Secret  string `json:"secret"`
+}
+
+func (i Installation) Valid() bool {
+	return i.Version == 1 && workspace.ValidIdentity(i.ID) && len(i.Secret) >= 40 && len(i.Secret) <= 256
+}
+
+// Initialize is called only by the control plane with a dedicated writable
+// bootstrap volume. The node receives this volume read-only.
+func Initialize(ctx context.Context, directory string) (Installation, error) {
+	root, err := statefs.OpenRoot(directory)
+	if err != nil {
+		return Installation{}, err
+	}
+	defer root.Close()
+	unlock, err := statefs.Lock(ctx, root)
+	if err != nil {
+		return Installation{}, err
+	}
+	defer unlock()
+	var installation Installation
+	err = statefs.Read(root, "installation.json", &installation)
+	if errors.Is(err, fs.ErrNotExist) {
+		installation = Installation{Version: 1, ID: rand.Text(), Secret: rand.Text() + rand.Text()}
+		err = statefs.Write(root, "installation.json", installation, false)
+	}
+	if err != nil {
+		return Installation{}, err
+	}
+	if !installation.Valid() {
+		return Installation{}, errors.New("invalid node bootstrap installation")
+	}
+	return installation, nil
+}
+
+func Read(directory string) (Installation, error) {
+	root, err := statefs.OpenRoot(directory)
+	if err != nil {
+		return Installation{}, err
+	}
+	defer root.Close()
+	var installation Installation
+	if err := statefs.Read(root, "installation.json", &installation); err != nil {
+		return Installation{}, err
+	}
+	if !installation.Valid() {
+		return Installation{}, errors.New("invalid node bootstrap installation")
+	}
+	return installation, nil
+}
+
+type Request struct {
+	InstallationID  string `json:"installationId"`
+	NodeIncarnation string `json:"nodeIncarnation"`
+}
+
+type Credential struct {
+	InstallationID  string `json:"installationId"`
+	NodeID          string `json:"nodeId"`
+	NodeIncarnation string `json:"nodeIncarnation"`
+	Token           string `json:"token"`
+}

apps/agent/bootstrap/bootstrap_test.go

@@ -0,0 +1,136 @@
+package bootstrap
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestInstallationConcurrentInitialization(t *testing.T) {
+	directory := t.TempDir()
+	var wg sync.WaitGroup
+	installations := make([]Installation, 8)
+	errorsSeen := make([]error, 8)
+	for i := range installations {
+		wg.Go(func() { installations[i], errorsSeen[i] = Initialize(context.Background(), directory) })
+	}
+	wg.Wait()
+	for i, err := range errorsSeen {
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !installations[i].Valid() || installations[i] != installations[0] {
+			t.Fatal("initialization changed installation identity")
+		}
+	}
+	i, err := Read(directory)
+	if err != nil || i != installations[0] {
+		t.Fatalf("read after restart: %v", err)
+	}
+	if runtime.GOOS != "windows" {
+		info, err := os.Stat(filepath.Join(directory, "installation.json"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if info.Mode().Perm() != 0600 {
+			t.Fatalf("secret permissions: %v", info.Mode())
+		}
+	}
+}
+
+func TestEnrollmentPersistsWithoutContactingServerAgain(t *testing.T) {
+	bootstrapDirectory, state := t.TempDir(), t.TempDir()
+	i, err := Initialize(context.Background(), bootstrapDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := Credential{i.ID, "node", "incarnation", "rnr_" + strings.Repeat("a", 43)}
+	var requests atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		if r.Method != http.MethodPost || r.URL.Path != "/api/runtime-nodes/installation/enroll" || r.Header.Get("Authorization") != "Bearer "+i.Secret {
+			t.Error("wrong enrollment request")
+			w.WriteHeader(401)
+			return
+		}
+		var request Request
+		if json.NewDecoder(r.Body).Decode(&request) != nil || request.InstallationID != i.ID || request.NodeIncarnation != "incarnation" {
+			t.Error("wrong identity")
+			w.WriteHeader(400)
+			return
+		}
+		_ = json.NewEncoder(w).Encode(want)
+	}))
+	defer server.Close()
+	credential, err := Wait(context.Background(), server.URL, bootstrapDirectory, state, "incarnation")
+	if err != nil || credential != want {
+		t.Fatalf("enroll: %v", err)
+	}
+	server.Close()
+	again, err := Wait(context.Background(), server.URL, bootstrapDirectory, state, "incarnation")
+	if err != nil || again != want || requests.Load() != 1 {
+		t.Fatalf("restart: %v requests=%d", err, requests.Load())
+	}
+	if _, err := Wait(context.Background(), server.URL, bootstrapDirectory, state, "replacement"); err == nil {
+		t.Fatal("changed identity reused credential")
+	}
+	if _, err := Wait(context.Background(), server.URL+"/other", bootstrapDirectory, state, "incarnation"); err == nil {
+		t.Fatal("changed control plane reused credential")
+	}
+}
+
+func TestEnrollmentResponseHandling(t *testing.T) {
+	i, err := Initialize(context.Background(), t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, test := range []struct {
+		name   string
+		status int
+		retry  bool
+		body   string
+	}{
+		{"setup", 409, true, ""}, {"startup", 503, true, ""}, {"revoked", 403, false, i.Secret},
+		{"redirect", 302, false, ""}, {"malformed", 200, false, "{}"},
+		{"trailing", 200, false, `{"installationId":"` + i.ID + `","nodeId":"node","nodeIncarnation":"incarnation","token":"rnr_` + strings.Repeat("a", 43) + `"} {}`},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(test.status)
+				_, _ = w.Write([]byte(test.body))
+			}))
+			defer server.Close()
+			_, retry, err := enroll(context.Background(), server.Client(), server.URL, t.TempDir(), "incarnation", i)
+			if retry != test.retry || !retry && err == nil {
+				t.Fatalf("retry=%v err=%v", retry, err)
+			}
+			if err != nil && strings.Contains(err.Error(), i.Secret) {
+				t.Fatal("response body leaked")
+			}
+		})
+	}
+}
+
+func TestEnrollmentWaitCancelsBeforeSetup(t *testing.T) {
+	directory := t.TempDir()
+	if _, err := Initialize(context.Background(), directory); err != nil {
+		t.Fatal(err)
+	}
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusConflict) }))
+	defer server.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	if _, err := Wait(ctx, server.URL, directory, t.TempDir(), "incarnation"); !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("cancel: %v", err)
+	}
+}