diff --git a/modules/gitea/cmd/forge-account-provider/bus_test.go b/modules/gitea/cmd/forge-account-provider/bus_test.go new file mode 100644 index 00000000..b332cae7 --- /dev/null +++ b/modules/gitea/cmd/forge-account-provider/bus_test.go @@ -0,0 +1,335 @@ +package main + +// The provider whole, against a real bus (issue 955 F5): a JetStream server is linked into the test and started for +// it alone, and the provider is this test binary started again as the bundle, speaking MCP on its standard streams +// to a runtime played here, which does for the bundle's state what the tool runner does: each state verb and each +// watch on the bucket the controller makes from the manifest. The forge is a fake (an httptest server), and the +// forge's own sshd is a program named ssh-keyscan on the bundle's PATH. +// +// It starts a server and a child process, so it is left to the merge check: run the unit tests with -short. + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/provider" +) + +const bundleVar = "MESH_FORGE_ACCOUNT_TEST_BUNDLE" + +func TestMain(m *testing.M) { + if os.Getenv(bundleVar) == "1" { + main() + os.Exit(0) + } + os.Exit(m.Run()) +} + +type runtime struct { + t *testing.T + js jetstream.JetStream + kvs map[string]jetstream.KeyValue + in io.WriteCloser + mu sync.Mutex + seq int + ctx context.Context + // answers to what the runtime asked the bundle, by id. + wait map[string]chan json.RawMessage +} + +func startBus(t *testing.T) (*runtime, context.CancelFunc) { + t.Helper() + s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, + StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + s.Shutdown() + t.Fatal("a bus for this test did not come up within 30s") + } + t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() }) + nc, err := nats.Connect(s.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + rt := &runtime{t: t, js: js, kvs: map[string]jetstream.KeyValue{}, ctx: ctx, wait: map[string]chan json.RawMessage{}} + // The controller's job: a bucket per state the module declares (bound) or reads (the others). + for state, bucket := range map[string]string{"bound": "gitea_bound", "ssh-client.keys": "ssh-client_keys", + "mesh-controller.assignments": "mesh-controller_assignments"} { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1}) + if err != nil { + t.Fatal(err) + } + rt.kvs[state] = kv + } + return rt, cancel +} + +type rpc struct { + ID json.RawMessage `json:"id,omitempty"` + Method string `json:"method,omitempty"` + Params json.RawMessage `json:"params,omitempty"` + Result json.RawMessage `json:"result,omitempty"` +} + +func (r *runtime) say(m map[string]any) { + m["jsonrpc"] = "2.0" + raw, _ := json.Marshal(m) + r.mu.Lock() + defer r.mu.Unlock() + _, _ = r.in.Write(append(raw, '\n')) +} + +func (r *runtime) start(env []string) { + r.t.Helper() + cmd := exec.Command(os.Args[0], "-test.run=^$") + cmd.Env = append(os.Environ(), append(env, bundleVar+"=1")...) + in, err := cmd.StdinPipe() + if err != nil { + r.t.Fatal(err) + } + out, err := cmd.StdoutPipe() + if err != nil { + r.t.Fatal(err) + } + cmd.Stderr = testWriter{r.t} + if err := cmd.Start(); err != nil { + r.t.Fatal(err) + } + r.in = in + go func() { + sc := bufio.NewScanner(out) + sc.Buffer(make([]byte, 1<<20), 1<<24) + for sc.Scan() { + var m rpc + if json.Unmarshal(sc.Bytes(), &m) != nil { + continue + } + if m.Method == "" { + var id string + _ = json.Unmarshal(m.ID, &id) + r.mu.Lock() + ch := r.wait[id] + delete(r.wait, id) + r.mu.Unlock() + if ch != nil { + ch <- m.Result + } + continue + } + go r.answer(m) + } + }() + r.t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() }) +} + +type testWriter struct{ t *testing.T } + +func (w testWriter) Write(p []byte) (int, error) { w.t.Logf("bundle: %s", p); return len(p), nil } + +// push tells the bundle one change of a state it watches, and waits until it has taken it. +func (r *runtime) push(watch string, c map[string]any) { + r.mu.Lock() + r.seq++ + id := "runtime-" + strconv.Itoa(r.seq) + ch := make(chan json.RawMessage, 1) + r.wait[id] = ch + r.mu.Unlock() + r.say(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": c}}) + select { + case <-ch: + case <-time.After(20 * time.Second): + r.t.Errorf("the bundle did not take a change of %v in 20 s", c["key"]) + } +} + +func (r *runtime) answer(m rpc) { + var p struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + } + _ = json.Unmarshal(m.Params, &p) + result, err := func() (any, error) { + kv := r.kvs[p.State] + if kv == nil { + return nil, fmt.Errorf("the state %q is not one this bundle declares or reads", p.State) + } + switch m.Method { + case "mesh/state.get": + e, err := kv.Get(r.ctx, p.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return map[string]any{"key": p.Key, "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil + case "mesh/state.put": + rev, err := kv.Put(r.ctx, p.Key, p.Value) + return map[string]any{"revision": rev}, err + case "mesh/state.delete": + err := kv.Delete(r.ctx, p.Key) + return map[string]any{}, err + case "mesh/state.watch": + w, err := kv.WatchAll(r.ctx) + if err != nil { + return nil, err + } + id := strings.Trim(string(m.ID), `"`) + for { + var e jetstream.KeyValueEntry + select { + case e = <-w.Updates(): + case <-r.ctx.Done(): + return nil, r.ctx.Err() + } + if e == nil { // the current values are all handed + break + } + if e.Operation() == jetstream.KeyValuePut { + r.push(id, map[string]any{"state": p.State, "key": e.Key(), "op": "put", "revision": e.Revision(), + "current": true, "value": json.RawMessage(e.Value())}) + } + } + go func() { + for e := range w.Updates() { + change := map[string]any{"state": p.State, "key": e.Key(), "op": "put", "revision": e.Revision(), "current": false, + "value": json.RawMessage(e.Value())} + if e.Operation() != jetstream.KeyValuePut { + change["op"] = "delete" + delete(change, "value") + } + r.push(id, change) + } + }() + return map[string]any{}, nil + } + return nil, fmt.Errorf("the runtime offers no %s to this bundle", m.Method) + }() + if err != nil { + r.say(map[string]any{"id": m.ID, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + r.say(map[string]any{"id": m.ID, "result": result}) +} + +func (r *runtime) put(state, key string, v any) { + r.t.Helper() + raw, _ := json.Marshal(v) + if _, err := r.kvs[state].Put(r.ctx, key, raw); err != nil { + r.t.Fatal(err) + } +} + +func (r *runtime) boundEntry(key string) (provider.Entry, bool) { + e, err := r.kvs["bound"].Get(r.ctx, key) + if err != nil { + return provider.Entry{}, false + } + var out provider.Entry + _ = json.Unmarshal(e.Value(), &out) + return out, true +} + +func eventually(t *testing.T, what string, ok func() bool) { + t.Helper() + for deadline := time.Now().Add(40 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { + if ok() { + return + } + } + t.Fatalf("not within 40 s: %s", what) +} + +func TestTheProviderWholeAnswersAMachineThatReportedItsKey(t *testing.T) { + if testing.Short() { + t.Skip("starts a bus and a process: left to the merge check") + } + forge := newForge() + srv := httptest.NewServer(forge) + t.Cleanup(srv.Close) + dir := t.TempDir() + bin := filepath.Join(dir, "bin") + if err := os.Mkdir(bin, 0o755); err != nil { + t.Fatal(err) + } + files := map[string]string{ + "pw": "x\n", + "host": "novox.internal\n", + "origins": "https://git.novox.be\n", + "bin/ssh-keyscan": "#!/bin/sh\necho '[127.0.0.1]:222 ssh-ed25519 AAAAC3HOSTKEY'\n", + } + for name, c := range files { + mode := os.FileMode(0o600) + if strings.HasPrefix(name, "bin/") { + mode = 0o755 + } + if err := os.WriteFile(filepath.Join(dir, name), []byte(c), mode); err != nil { + t.Fatal(err) + } + } + rt, cancel := startBus(t) + t.Cleanup(cancel) + rt.put("mesh-controller.assignments", "g14.ssh-client", map[string]any{"node": "g14", "module": "ssh-client", + "provisioning": map[string]any{"provisions": []map[string]any{{"provision": "forge-account", "provider": "novox", + "provider-module": "gitea", "as": "mesh_g14_ssh_client"}}}}) + rt.put("ssh-client.keys", "g14.nox", map[string]string{"node": "g14", "account": "nox", "kind": "machine", + "public": pubA + " nox@g14 (mesh)", "fingerprint": "SHA256:x", "at": "2026-10-16T10:00:00Z"}) + rt.start([]string{"MESH_NODE=novox", "MESH_MODULE=gitea", "MESH_SERVED_MODULE=gitea", + "PATH=" + bin + ":" + os.Getenv("PATH"), + "MESH_GITEA_URL=" + srv.URL, "MESH_GITEA_ADMIN_USER=mesh-admin", + "MESH_GITEA_ADMIN_PASSWORD_FILE=" + filepath.Join(dir, "pw"), + "MESH_GITEA_ORIGINS_FILE=" + filepath.Join(dir, "origins"), + "MESH_FORGE_HOST_FILE=" + filepath.Join(dir, "host"), "MESH_GITEA_SSH_PORT=222"}) + + var e provider.Entry + eventually(t, "the bound entry of g14's ssh-client", func() bool { + var ok bool + e, ok = rt.boundEntry("g14.ssh-client.forge-account") + return ok + }) + if e.Values["user"] != "agent-g14" || e.Values["ssh_host"] != "novox.internal" || + e.Values["known_host"] != "[novox.internal]:222 ssh-ed25519 AAAAC3HOSTKEY" || + e.Values["ssh_base"] != "ssh://git@novox.internal:222/" || e.Values["https_base"] != "https://git.novox.be/" { + t.Fatalf("values = %v", e.Values) + } + if got := forge.titles("agent-g14"); len(got) != 1 || got[0] != "mesh:g14.nox" { + t.Fatalf("registered keys = %v", got) + } + + // A key that changes is answered without waiting for the next pass. + rt.put("ssh-client.keys", "g14.nox", map[string]string{"node": "g14", "account": "nox", "kind": "machine", + "public": pubB + " nox@g14 (mesh)", "fingerprint": "SHA256:y", "at": "2026-10-16T11:00:00Z"}) + eventually(t, "the new key registered and the old one gone", func() bool { + forge.mu.Lock() + defer forge.mu.Unlock() + k := forge.held["agent-g14"] + return len(k) == 1 && strings.HasPrefix(k[0]["key"].(string), pubB) + }) +} diff --git a/modules/gitea/go.mod b/modules/gitea/go.mod index 0fa6eee9..6ac66ac9 100644 --- a/modules/gitea/go.mod +++ b/modules/gitea/go.mod @@ -2,9 +2,21 @@ module gitea go 1.25.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 + github.com/nats-io/nats-server/v2 v2.11.17 + github.com/nats-io/nats.go v1.51.0 + golang.org/x/crypto v0.50.0 +) require ( - golang.org/x/crypto v0.50.0 + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect + github.com/klauspost/compress v1.18.5 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/sys v0.43.0 // indirect + golang.org/x/time v0.15.0 // indirect ) diff --git a/modules/gitea/go.sum b/modules/gitea/go.sum index 735d7c6c..83c75127 100644 --- a/modules/gitea/go.sum +++ b/modules/gitea/go.sum @@ -1,6 +1,27 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 h1:EK84so3Y8JdFT3BHrzj9cPQQrDUPa/d+IdOGnerlmEI= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= +github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= +github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= -golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 h1:EK84so3Y8JdFT3BHrzj9cPQQrDUPa/d+IdOGnerlmEI= golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= -golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=