diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index cc69cca..393f5dd 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -8,13 +8,11 @@ package main import ( "context" - "errors" "fmt" "io" "log" "os" "os/signal" - "strings" "sync" "syscall" "time" @@ -31,7 +29,7 @@ import ( ) func main() { - if os.Getenv("MESH_BROKER_FILE") == "" { + if os.Getenv(process.BrokerFileEnv) == "" { // Started by the tool runner, with no credential of its own: the put's tools (ADR 0546), served through the runtime, // since a process never serves (ADR 0436). The process the manifest runs names its credential. if err := serveTools(); err != nil { @@ -118,33 +116,15 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-assigner: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to read the bus with") - } - account, err := process.ReadAccount(path) + account, err := process.ReadBrokerAccount(getenv) if err != nil { return err } - opts := []nats.Option{ - nats.Name(account.Node + "." + account.Module), - nats.UserInfo(account.User, account.Password), - nats.CustomInboxPrefix(account.InboxPrefix()), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), - } - if strings.TrimSpace(account.Fingerprint) != "" { - opts = append(opts, nats.Secure(process.PinnedToFingerprint(account.Fingerprint))) - } - conn, err := nats.Connect(account.URL, opts...) + conn, js, err := account.Dial() if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) + return err } defer conn.Close() - js, err := jetstream.New(conn) - if err != nil { - return err - } kvs, err := open(ctx, js, logger.Printf) if err != nil { return err diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 73e5df7..d2b7ff8 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index 0082dd1..fa4f833 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -1,17 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index ee96827..b944b16 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-builder/internal/build" ) @@ -345,7 +346,7 @@ func (b *bed) archive(t *testing.T, name, path, at string) string { func (b *bed) start(t *testing.T, node string, env map[string]string) { t.Helper() broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: b.srvURL, Node: node, Module: "mesh-delivery-builder", User: node + ".mesh-delivery-builder", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: b.srvURL, Node: node, Module: "mesh-delivery-builder", User: node + ".mesh-delivery-builder", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 15f126c..d602f8e 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -6,12 +6,6 @@ package main import ( "context" - "crypto/sha256" - "crypto/tls" - "crypto/x509" - "encoding/hex" - "encoding/json" - "errors" "fmt" "io" "log" @@ -51,80 +45,18 @@ func main() { } } -// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate -// fingerprint to pin, the principal and its password (as the indexer reads it; issue 1242 moves this into the SDK). -type credential struct { - URL string `json:"url"` - Fingerprint string `json:"fingerprint"` - Node string `json:"node"` - Module string `json:"module"` - User string `json:"user"` - Password string `json:"password"` -} - -func readCredential(path string) (credential, error) { - raw, err := os.ReadFile(path) - if err != nil { - return credential{}, err - } - var c credential - if err := json.Unmarshal(raw, &c); err != nil { - return credential{}, fmt.Errorf("the broker credential is not one: %w", err) - } - if c.URL == "" || c.User == "" { - return credential{}, errors.New("the broker credential names no bus address or user") - } - return c, nil -} - -// pinned accepts exactly the certificate with this SHA-256 and no other. -func pinned(want string) *tls.Config { - return &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter - MinVersion: tls.VersionTLS12, - VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - if len(rawCerts) == 0 { - return errors.New("the bus presented no certificate") - } - sum := sha256.Sum256(rawCerts[0]) - if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { - return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) - } - return nil - }, - } -} - // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-builder: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with") - } - cred, err := readCredential(path) + cred, err := process.ReadBrokerAccount(getenv) if err != nil { return err } - opts := []nats.Option{ - nats.Name(cred.Node + "." + cred.Module), - nats.UserInfo(cred.User, cred.Password), - nats.CustomInboxPrefix("_INBOX." + cred.User), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), - } - if strings.TrimSpace(cred.Fingerprint) != "" { - opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) - } - conn, err := nats.Connect(cred.URL, opts...) + conn, js, err := cred.Dial() if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) + return err } defer conn.Close() - js, err := jetstream.New(conn) - if err != nil { - return err - } builds, err := process.EssentialBucket(ctx, js, "builder", delivery.BuilderBuildsBucket, "claim or write a build key") if err != nil { return err @@ -163,11 +95,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again() }) // The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs // (the seat verb held, novox/hq issue 1937) before it leaves the key done. - account, err := process.ReadAccount(path) - if err != nil { - return err - } - calls, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) if err != nil { return err } @@ -232,13 +160,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // module. for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} { stream := toolchainStreams[module] - cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module) + cons, err := cred.EventConsumer(ctx, js, stream) if err != nil { logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the builder starts again: %v", cred.Node, cred.Module, stream, module, err) continue } - cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) }, - jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + cc, err := process.HearEvents(cons, func(m jetstream.Msg) { watch.Reported(m) }, logger.Printf) if err != nil { return err } @@ -247,13 +174,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The controller says artifacts-gone when its take-in found a key's blob gone (issue 1937): the builder's durable // consumer on EVENTS, made by the controller from the manifest's consumes. Where it is not there yet the builder says // so; it still asks held at its start and at every module value. - switch cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module); { + switch cons, err := cred.EventConsumer(ctx, js, eventsStream); { case err != nil: logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v", cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err) default: - cc, err := cons.Consume(func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) }, - jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + cc, err := process.HearEvents(cons, func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) }, logger.Printf) if err != nil { return err } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go index 48bcd29..de72b5c 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-sdk/go/process" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -53,12 +54,12 @@ func reAsked(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e jetstre // emitAskedAgain says the builder's event build.asked-again (issue 1945): mesh.mod..event.build.asked-again, // with the SDK's envelope headers and the new ask's id as the bus's dedup id. -func emitAskedAgain(ctx context.Context, js jetstream.JetStream, cred credential, e build.AskedAgain) error { +func emitAskedAgain(ctx context.Context, js jetstream.JetStream, cred process.Account, e build.AskedAgain) error { data, err := json.Marshal(e) if err != nil { return err } - msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + build.AskedAgainEvent) + msg := nats.NewMsg(cred.EventSubject(build.AskedAgainEvent)) id := "asked-again-" + e.New msg.Header.Set("x-event-id", id) msg.Header.Set("x-source", cred.Module) diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 3c79bbe..e5c4d4e 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-builder go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index e793069..fa4f833 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index 3e01724..6b39ca3 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -25,6 +25,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" + "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-checker/internal/check" ) @@ -423,7 +424,7 @@ func pull(repository string, number int, head string) delivery.CandidatePull { func (b *bed) start(t *testing.T, env map[string]string) { t.Helper() broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: b.srvURL, Node: "n1", Module: "mesh-delivery-checker", User: "n1.mesh-delivery-checker", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: b.srvURL, Node: "n1", Module: "mesh-delivery-checker", User: "n1.mesh-delivery-checker", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index c4d3902..5dc2441 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -7,12 +7,7 @@ package main import ( "context" - "crypto/sha256" - "crypto/tls" - "crypto/x509" - "encoding/hex" "encoding/json" - "errors" "fmt" "io" "log" @@ -36,7 +31,7 @@ import ( // eventSubject is the subject of an event of the checker's stream checks (the bus names the stream CHECKS): the indexer's // watch of the previews bucket and the forge bridge's watch of the owed bucket are woken by them. func eventSubject(event string) string { - return "mesh.stream." + delivery.CheckerModule + ".event." + event + return process.ModuleStreamSubject(delivery.CheckerModule, event) } const ( @@ -53,50 +48,6 @@ func main() { } } -// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate -// fingerprint to pin, the principal and its password (as the indexer reads it; issue 1242 moves this into the SDK). -type credential struct { - URL string `json:"url"` - Fingerprint string `json:"fingerprint"` - Node string `json:"node"` - Module string `json:"module"` - User string `json:"user"` - Password string `json:"password"` -} - -func readCredential(path string) (credential, error) { - raw, err := os.ReadFile(path) - if err != nil { - return credential{}, err - } - var c credential - if err := json.Unmarshal(raw, &c); err != nil { - return credential{}, fmt.Errorf("the broker credential is not one: %w", err) - } - if c.URL == "" || c.User == "" { - return credential{}, errors.New("the broker credential names no bus address or user") - } - return c, nil -} - -// pinned accepts exactly the certificate with this SHA-256 and no other. -func pinned(want string) *tls.Config { - return &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter - MinVersion: tls.VersionTLS12, - VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - if len(rawCerts) == 0 { - return errors.New("the bus presented no certificate") - } - sum := sha256.Sum256(rawCerts[0]) - if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { - return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) - } - return nil - }, - } -} - // durationEnv reads a duration setting from the environment, def when it is not set. func durationEnv(getenv func(string) string, name string, def time.Duration) (time.Duration, error) { v := strings.TrimSpace(getenv(name)) @@ -128,33 +79,15 @@ var keptStates = []announce.State{ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-checker: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with") - } - cred, err := readCredential(path) + cred, err := process.ReadBrokerAccount(getenv) if err != nil { return err } - opts := []nats.Option{ - nats.Name(cred.Node + "." + cred.Module), - nats.UserInfo(cred.User, cred.Password), - nats.CustomInboxPrefix("_INBOX." + cred.User), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), - } - if strings.TrimSpace(cred.Fingerprint) != "" { - opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) - } - conn, err := nats.Connect(cred.URL, opts...) + conn, js, err := cred.Dial() if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) + return err } defer conn.Close() - js, err := jetstream.New(conn) - if err != nil { - return err - } // Its own states, the candidates it checks, and the modules and manifests that say which build modules judge a repository: // a check without one of them would be wrong or unwritten, so each stops the start, naming its grant (issue 1945). buckets := map[string]jetstream.KeyValue{} @@ -166,11 +99,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // Calls to the build modules go out as this module's own account, naming the machine (mesh-sdk process.CallOn); the // buckets are read and written on the connection above. - account, err := process.ReadAccount(path) - if err != nil { - return err - } - caller, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + caller, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) if err != nil { return err } @@ -287,13 +216,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // module. for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} { stream := toolchainStreams[module] - cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module) + cons, err := cred.EventConsumer(ctx, js, stream) if err != nil { logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the checker starts again: %v", cred.Node, cred.Module, stream, module, err) continue } - cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) }, - jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + cc, err := process.HearEvents(cons, func(m jetstream.Msg) { watch.Reported(m) }, logger.Printf) if err != nil { return err } diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 08c70f0..c34164e 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 6cf00d5..fa4f833 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,11 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026233910-a2759ec92ec9 h1:MS7i4NVNQHMAMqRRFMdzbto31BHZY+BuEZY1ql3p4Rs= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026233910-a2759ec92ec9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 05497d4..59ed05e 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -21,6 +21,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" ) // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, @@ -271,7 +272,7 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, } broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-forge-bridge", User: "n1.mesh-delivery-forge-bridge", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-forge-bridge", User: "n1.mesh-delivery-forge-bridge", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index dbdf411..9bfb202 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -6,18 +6,12 @@ package main import ( "context" - "crypto/sha256" - "crypto/tls" - "crypto/x509" - "encoding/hex" - "encoding/json" "errors" "fmt" "io" "log" "os" "os/signal" - "strings" "sync" "syscall" "time" @@ -39,7 +33,7 @@ const eventsStream = "EVENTS" // The bridge's own stream FORGE takes `mesh.stream..event.` (mesh-control internal/broker // modstreams.go ModuleStreamSubject). func forgeSubject(event string) string { - return "mesh.stream." + delivery.ForgeBridgeModule + ".event." + event + return process.ModuleStreamSubject(delivery.ForgeBridgeModule, event) } func main() { @@ -51,50 +45,6 @@ func main() { } } -// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate -// fingerprint to pin, the principal and its password (as the watchman reads it; issue 1242 moves this into the SDK). -type credential struct { - URL string `json:"url"` - Fingerprint string `json:"fingerprint"` - Node string `json:"node"` - Module string `json:"module"` - User string `json:"user"` - Password string `json:"password"` -} - -func readCredential(path string) (credential, error) { - raw, err := os.ReadFile(path) - if err != nil { - return credential{}, err - } - var c credential - if err := json.Unmarshal(raw, &c); err != nil { - return credential{}, fmt.Errorf("the broker credential is not one: %w", err) - } - if c.URL == "" || c.User == "" { - return credential{}, errors.New("the broker credential names no bus address or user") - } - return c, nil -} - -// pinned accepts exactly the certificate with this SHA-256 and no other. -func pinned(want string) *tls.Config { - return &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter - MinVersion: tls.VersionTLS12, - VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - if len(rawCerts) == 0 { - return errors.New("the bus presented no certificate") - } - sum := sha256.Sum256(rawCerts[0]) - if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { - return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) - } - return nil - }, - } -} - // delivered is the part of a JetStream message the bridge uses. type delivered interface { Subject() string @@ -152,40 +102,23 @@ func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-forge-bridge: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with") - } - cred, err := readCredential(path) + cred, err := process.ReadBrokerAccount(getenv) if err != nil { return err } again := make(chan struct{}, 1) - opts := []nats.Option{ - nats.Name(cred.Node + "." + cred.Module), - nats.UserInfo(cred.User, cred.Password), - nats.CustomInboxPrefix("_INBOX." + cred.User), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), + conn, js, err := cred.Dial( nats.ReconnectHandler(func(*nats.Conn) { select { case again <- struct{}{}: default: } }), - } - if strings.TrimSpace(cred.Fingerprint) != "" { - opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) - } - conn, err := nats.Connect(cred.URL, opts...) - if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) - } - defer conn.Close() - js, err := jetstream.New(conn) + ) if err != nil { return err } + defer conn.Close() // What the forge holds open, said before any event is heard, and again at each reconnect (ADR 0429 §13). call := callerOn(conn, cred.User) // No forge event is heard while the open pull requests are read and said (ADR 0443 §5): events and syncs take turns. @@ -227,20 +160,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } go onReconnect(ctx, again, said, logger) - cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module) + cons, err := cred.EventConsumer(ctx, js, eventsStream) if err != nil { // Essential: the forge's events are what the bridge carries. return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, eventsStream, err) } - cc, err := cons.Consume(func(m jetstream.Msg) { + cc, err := process.HearEvents(cons, func(m jetstream.Msg) { hear(&turn, func() { // Bounded: the turn is held, and no sync or other event goes on without it. hctx, cancel := context.WithTimeout(ctx, syncDeadline) defer cancel() handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger) }) - }, - jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + }, logger.Printf) if err != nil { return err } diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index 75272b1..b1bc526 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-forge-bridge/go.sum b/modules/mesh-delivery-forge-bridge/go.sum index 0082dd1..fa4f833 100644 --- a/modules/mesh-delivery-forge-bridge/go.sum +++ b/modules/mesh-delivery-forge-bridge/go.sum @@ -1,17 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 815e319..edecd2f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -20,6 +20,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" ) // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the @@ -171,7 +172,7 @@ func startIndexer(t *testing.T, put func(context.Context, jetstream.KeyValue), b } broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-indexer", User: "n1.mesh-delivery-indexer", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 3d1d1b1..2049c7f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -10,10 +10,6 @@ package main import ( "context" - "crypto/sha256" - "crypto/tls" - "crypto/x509" - "encoding/hex" "encoding/json" "errors" "fmt" @@ -21,7 +17,6 @@ import ( "log" "os" "os/signal" - "strings" "syscall" "github.com/nats-io/nats.go" @@ -47,50 +42,6 @@ func main() { } } -// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate -// fingerprint to pin, the principal and its password (as the watchman reads it; issue 1242 moves this into the SDK). -type credential struct { - URL string `json:"url"` - Fingerprint string `json:"fingerprint"` - Node string `json:"node"` - Module string `json:"module"` - User string `json:"user"` - Password string `json:"password"` -} - -func readCredential(path string) (credential, error) { - raw, err := os.ReadFile(path) - if err != nil { - return credential{}, err - } - var c credential - if err := json.Unmarshal(raw, &c); err != nil { - return credential{}, fmt.Errorf("the broker credential is not one: %w", err) - } - if c.URL == "" || c.User == "" { - return credential{}, errors.New("the broker credential names no bus address or user") - } - return c, nil -} - -// pinned accepts exactly the certificate with this SHA-256 and no other. -func pinned(want string) *tls.Config { - return &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter - MinVersion: tls.VersionTLS12, - VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - if len(rawCerts) == 0 { - return errors.New("the bus presented no certificate") - } - sum := sha256.Sum256(rawCerts[0]) - if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { - return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) - } - return nil - }, - } -} - // awaitedStore is a state bound when it appears (process.AwaitBucket) as the indexer's store: until then every call is // process.ErrNotYet. type awaitedStore struct{ b *process.AwaitedBucket } @@ -272,33 +223,15 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-indexer: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with") - } - cred, err := readCredential(path) + cred, err := process.ReadBrokerAccount(getenv) if err != nil { return err } - opts := []nats.Option{ - nats.Name(cred.Node + "." + cred.Module), - nats.UserInfo(cred.User, cred.Password), - nats.CustomInboxPrefix("_INBOX." + cred.User), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), - } - if strings.TrimSpace(cred.Fingerprint) != "" { - opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) - } - conn, err := nats.Connect(cred.URL, opts...) + conn, js, err := cred.Dial() if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) + return err } defer conn.Close() - js, err := jetstream.New(conn) - if err != nil { - return err - } candidates, err := process.EssentialBucket(ctx, js, "indexer", delivery.CandidatesBucket, "write a candidate's fingerprints") if err != nil { return err @@ -344,11 +277,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // The indexer asks the forge for the objects of a manifest's named paths as its own account (ADR 0499 clause 3). - account, err := process.ReadAccount(path) - if err != nil { - return err - } - calls, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) if err != nil { return err } @@ -449,30 +378,28 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // with the build module, as with the gate's source. for _, module := range sortedModules() { stream := index.ToolchainStreams[module] - tcons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module) + tcons, err := cred.EventConsumer(ctx, js, stream) if err != nil { logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the indexer starts again: %v", cred.Node, cred.Module, stream, module, err) continue } - tc, err := tcons.Consume(func(m jetstream.Msg) { tools.Reported(m) }, - jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + tc, err := process.HearEvents(tcons, func(m jetstream.Msg) { tools.Reported(m) }, logger.Printf) if err != nil { return err } defer tc.Stop() } - switch pcons, err := js.Consumer(ctx, PreviewsStream, cred.Node+"_"+cred.Module); { + switch pcons, err := cred.EventConsumer(ctx, js, PreviewsStream); { case err != nil: logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's previews are not heard until the indexer starts again: %v", cred.Node, cred.Module, PreviewsStream, err) default: - pc, err := pcons.Consume(func(m jetstream.Msg) { previews.Written(m) }, - jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + pc, err := process.HearEvents(pcons, func(m jetstream.Msg) { previews.Written(m) }, logger.Printf) if err != nil { return err } defer pc.Stop() } - cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module) + cons, err := cred.EventConsumer(ctx, js, forgeStream) if err != nil { // Essential: the forge's merges are what the indexer indexes. return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the indexer hears no merge without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, forgeStream, err) @@ -487,8 +414,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return info.NumPending == 0 && info.NumAckPending == 0, nil } taken.Ready(func(jetstream.KeyValue) { ix.KickFill(ctx) }) - cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) }, - jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, logger); ix.KickFill(ctx) }, logger.Printf) if err != nil { return err } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go index 3b95a25..21f0b64 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go @@ -3,11 +3,12 @@ package main import ( "context" "errors" - "git.novox.be/novox/mesh-sdk/go/process" "log" "strings" "sync" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index b35bfdf..1c8c4e9 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index 9259a48..fa4f833 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,17 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index bb24aa4..e32088b 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -19,6 +19,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" ) // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the @@ -90,7 +91,7 @@ func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2 defer cancel() broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -321,7 +322,7 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2 b := startFull(t) tools := b.as(t, "n1.node-tools") broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -411,7 +412,7 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1 b := startFull(t) tools := b.as(t, "n1.node-tools") broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -483,7 +484,7 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC b := startFull(t) assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view") broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -556,7 +557,7 @@ func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) { b := startFull(t) assigner := b.as(t, "n1.mesh-delivery-assigner") broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -650,7 +651,7 @@ func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) { b := startFull(t) assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view") broker := filepath.Join(t.TempDir(), "broker") - cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index d73893d..4e2c9c8 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -8,23 +8,16 @@ package main import ( "context" - "crypto/sha256" - "crypto/tls" - "crypto/x509" - "encoding/hex" - "encoding/json" "errors" "fmt" "io" "log" "os" "os/signal" - "strings" "syscall" "git.novox.be/novox/mesh-sdk/go/process" - "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/asker" @@ -40,80 +33,18 @@ func main() { } } -// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate -// fingerprint to pin, the principal and its password (as the watchman reads it; issue 1242 moves this into the SDK). -type credential struct { - URL string `json:"url"` - Fingerprint string `json:"fingerprint"` - Node string `json:"node"` - Module string `json:"module"` - User string `json:"user"` - Password string `json:"password"` -} - -func readCredential(path string) (credential, error) { - raw, err := os.ReadFile(path) - if err != nil { - return credential{}, err - } - var c credential - if err := json.Unmarshal(raw, &c); err != nil { - return credential{}, fmt.Errorf("the broker credential is not one: %w", err) - } - if c.URL == "" || c.User == "" { - return credential{}, errors.New("the broker credential names no bus address or user") - } - return c, nil -} - -// pinned accepts exactly the certificate with this SHA-256 and no other. -func pinned(want string) *tls.Config { - return &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter - MinVersion: tls.VersionTLS12, - VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - if len(rawCerts) == 0 { - return errors.New("the bus presented no certificate") - } - sum := sha256.Sum256(rawCerts[0]) - if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { - return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) - } - return nil - }, - } -} - // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-teller: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to be on the bus with") - } - cred, err := readCredential(path) + cred, err := process.ReadBrokerAccount(getenv) if err != nil { return err } - opts := []nats.Option{ - nats.Name(cred.Node + "." + cred.Module), - nats.UserInfo(cred.User, cred.Password), - nats.CustomInboxPrefix("_INBOX." + cred.User), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), - } - if strings.TrimSpace(cred.Fingerprint) != "" { - opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) - } - conn, err := nats.Connect(cred.URL, opts...) + conn, js, err := cred.Dial() if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) + return err } defer conn.Close() - js, err := jetstream.New(conn) - if err != nil { - return err - } gate, err := js.KeyValue(ctx, delivery.AsksBucket) if errors.Is(err, jetstream.ErrBucketNotFound) { logger.Print(waiting) diff --git a/modules/mesh-delivery-teller/go.mod b/modules/mesh-delivery-teller/go.mod index c8156b4..d89c1a0 100644 --- a/modules/mesh-delivery-teller/go.mod +++ b/modules/mesh-delivery-teller/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-teller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-teller/go.sum b/modules/mesh-delivery-teller/go.sum index b3dab4f..fa4f833 100644 --- a/modules/mesh-delivery-teller/go.sum +++ b/modules/mesh-delivery-teller/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index c13bd86..fde1855 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -177,3 +177,28 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR t.Fatalf("read %d source files, want the program's", files) } } + +// Issue 1242: no delivery part hand-writes its bus connection. A part's own credential, its pinned dial, the subjects of its +// streams and its durable consumer come from mesh-sdk's process package (ReadBrokerAccount, Account.Dial, Account.EventConsumer, +// ModuleStreamSubject); a part whose main.go dials with nats.Connect, pins with its own certificate check or binds the +// consumer name itself is refused here. mesh-grants is outside it until its own seats change (issue 1945) lands. +func TestNoPartHandWritesItsBusConnection(t *testing.T) { + mains, err := filepath.Glob("../mesh-delivery-*/cmd/*/*.go") + if err != nil || len(mains) == 0 { + t.Fatalf("no delivery part found: %v", err) + } + for _, f := range mains { + if strings.HasSuffix(f, "_test.go") { + continue + } + raw, err := os.ReadFile(f) + if err != nil { + t.Fatal(err) + } + for _, banned := range []string{"nats.Connect(", "VerifyPeerCertificate", `cred.Node+"_"+cred.Module`, `"MESH_BROKER_FILE"`} { + if bytes.Contains(raw, []byte(banned)) { + t.Errorf("%s hand-writes the bus connection (%s): use mesh-sdk's process.ReadBrokerAccount, Account.Dial, Account.EventConsumer or ModuleStreamSubject", f, banned) + } + } + } +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index 69d4247..55eedf8 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -18,6 +18,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-view/internal/view" ) @@ -81,7 +82,7 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8 } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -180,7 +181,7 @@ func TestProcessWholeShowsWhatTheWatchmanSays(t *testing.T) { // 1168/AC5 } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -246,7 +247,7 @@ func TestProcessWholeShowsTheIndexersBucketsOnceTheyExist(t *testing.T) { // 124 } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -332,7 +333,7 @@ func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { // } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -422,7 +423,7 @@ func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -519,7 +520,7 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129 } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -670,7 +671,7 @@ func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -738,7 +739,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -834,7 +835,7 @@ func TestProcessWholeSaysModulesInMotion(t *testing.T) { } t.Cleanup(func() { os.RemoveAll(dir) }) broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index aa41f4f..0fcca90 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -7,8 +7,6 @@ package main import ( "context" "crypto/sha256" - "crypto/tls" - "crypto/x509" "encoding/hex" "encoding/json" "errors" @@ -26,6 +24,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-view/internal/view" ) @@ -44,58 +43,10 @@ func main() { } } -// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate -// fingerprint to pin, the principal and its password. -type credential struct { - URL string `json:"url"` - Fingerprint string `json:"fingerprint"` - Node string `json:"node"` - Module string `json:"module"` - User string `json:"user"` - Password string `json:"password"` -} - -func readCredential(path string) (credential, error) { - raw, err := os.ReadFile(path) - if err != nil { - return credential{}, err - } - var c credential - if err := json.Unmarshal(raw, &c); err != nil { - return credential{}, fmt.Errorf("the broker credential is not one: %w", err) - } - if c.URL == "" || c.User == "" { - return credential{}, errors.New("the broker credential names no bus address or user") - } - return c, nil -} - -// pinned accepts exactly the certificate with this SHA-256 and no other. -func pinned(want string) *tls.Config { - return &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter - MinVersion: tls.VersionTLS12, - VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - if len(rawCerts) == 0 { - return errors.New("the bus presented no certificate") - } - sum := sha256.Sum256(rawCerts[0]) - if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { - return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) - } - return nil - }, - } -} - // run is the program: it returns when ctx ends or a listener cannot be made. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-view: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to watch the buckets with") - } - cred, err := readCredential(path) + cred, err := process.ReadBrokerAccount(getenv) if err != nil { return err } @@ -118,25 +69,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { webUp <- srv.ListenAndServe() }() go view.ServeSocket(ctx, l, model) - opts := []nats.Option{ - nats.Name(cred.Node + "." + cred.Module), - nats.UserInfo(cred.User, cred.Password), - nats.CustomInboxPrefix("_INBOX." + cred.User), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), - } - if strings.TrimSpace(cred.Fingerprint) != "" { - opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) - } - conn, err := nats.Connect(cred.URL, opts...) - if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) - } - defer conn.Close() - js, err := jetstream.New(conn) + conn, js, err := cred.Dial() if err != nil { return err } + defer conn.Close() model.SetWriter(kvWriter{ctx: ctx, js: js}) model.SetMotionEvent(func(e view.MotionEvent) { if err := emitEvent(ctx, js, cred, e.Type, e.Body, e.ID); err != nil { @@ -263,12 +200,12 @@ const DeliveredEventType = "delivered" // emitEvent publishes an event of the view's own namespace, mesh.mod..event., with the SDK's envelope headers // (as mesh-sdk's process.Conn.Emit does) and the id as the bus's own dedup id, so the same revision said twice in the stream's // duplicate window is kept once. -func emitEvent(ctx context.Context, js jetstream.JetStream, cred credential, eventType string, body any, id string) error { +func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account, eventType string, body any, id string) error { data, err := json.Marshal(body) if err != nil { return err } - msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + eventType) + msg := nats.NewMsg(cred.EventSubject(eventType)) msg.Header.Set("x-event-id", id) msg.Header.Set("x-source", cred.Module) msg.Header.Set("x-node", cred.Node) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go index f536aab..e60edff 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go @@ -3,29 +3,10 @@ package main import ( "context" "io" - "os" - "path/filepath" "strings" "testing" ) -func TestCredentialIsTheBrokerFile(t *testing.T) { - path := filepath.Join(t.TempDir(), "broker") - os.WriteFile(path, []byte(`{"url":"nats://127.0.0.1:4222","fingerprint":"sha256:ab","node":"n","module":"mesh-delivery-view","user":"u","password":"p"}`), 0o600) - c, err := readCredential(path) - if err != nil || c.URL != "nats://127.0.0.1:4222" || c.User != "u" || c.Password != "p" || c.Fingerprint != "sha256:ab" { - t.Fatalf("credential = %+v, %v", c, err) - } - os.WriteFile(path, []byte(`{"url":"x"}`), 0o600) - if _, err := readCredential(path); err == nil { - t.Fatal("a credential naming no user must be refused") - } - os.WriteFile(path, []byte(`not json`), 0o600) - if _, err := readCredential(path); err == nil { - t.Fatal("a file that is not a credential must be refused") - } -} - func TestRunNeedsItsCredential(t *testing.T) { err := run(context.Background(), func(string) string { return "" }, io.Discard) if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 8609300..a3bb0c8 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-view/go.sum b/modules/mesh-delivery-view/go.sum index 9259a48..fa4f833 100644 --- a/modules/mesh-delivery-view/go.sum +++ b/modules/mesh-delivery-view/go.sum @@ -1,17 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index 6e99302..27cf64c 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -18,6 +18,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" ) // busConf is the composed accounts file's shape (mesh-controller internal/broker): the account MESH imports the system @@ -114,7 +115,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8 connect("node.n0") broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -286,7 +287,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909 t.Fatal(err) } broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } @@ -388,7 +389,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) { putHave("n2", "web", fpOld) putWant("n2", "web", fpNew) // stood before the start broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) + cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"}) if err := os.WriteFile(broker, cred, 0o600); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 57994b4..c349da7 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -7,11 +7,6 @@ package main import ( "context" - "crypto/sha256" - "crypto/tls" - "crypto/x509" - "encoding/hex" - "encoding/json" "errors" "fmt" "io" @@ -49,50 +44,6 @@ func main() { } } -// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate -// fingerprint to pin, the principal and its password. -type credential struct { - URL string `json:"url"` - Fingerprint string `json:"fingerprint"` - Node string `json:"node"` - Module string `json:"module"` - User string `json:"user"` - Password string `json:"password"` -} - -func readCredential(path string) (credential, error) { - raw, err := os.ReadFile(path) - if err != nil { - return credential{}, err - } - var c credential - if err := json.Unmarshal(raw, &c); err != nil { - return credential{}, fmt.Errorf("the broker credential is not one: %w", err) - } - if c.URL == "" || c.User == "" { - return credential{}, errors.New("the broker credential names no bus address or user") - } - return c, nil -} - -// pinned accepts exactly the certificate with this SHA-256 and no other. -func pinned(want string) *tls.Config { - return &tls.Config{ - InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter - MinVersion: tls.VersionTLS12, - VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error { - if len(rawCerts) == 0 { - return errors.New("the bus presented no certificate") - } - sum := sha256.Sum256(rawCerts[0]) - if got := "sha256:" + hex.EncodeToString(sum[:]); got != want { - return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got) - } - return nil - }, - } -} - // kvOut writes the watchman's three buckets. type kvOut struct { ctx context.Context @@ -190,40 +141,23 @@ func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) { // run is the program: it returns when ctx ends or the bus cannot be used. func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { logger := log.New(logOut, "mesh-delivery-watchman: ", log.LstdFlags) - path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) - if path == "" { - return errors.New("MESH_BROKER_FILE is not set, so there is no credential to watch the bus with") - } - cred, err := readCredential(path) + cred, err := process.ReadBrokerAccount(getenv) if err != nil { return err } again := make(chan struct{}, 1) - opts := []nats.Option{ - nats.Name(cred.Node + "." + cred.Module), - nats.UserInfo(cred.User, cred.Password), - nats.CustomInboxPrefix("_INBOX." + cred.User), - nats.MaxReconnects(-1), - nats.RetryOnFailedConnect(true), + conn, js, err := cred.Dial( nats.ReconnectHandler(func(*nats.Conn) { select { case again <- struct{}{}: default: } }), - } - if strings.TrimSpace(cred.Fingerprint) != "" { - opts = append(opts, nats.Secure(pinned(cred.Fingerprint))) - } - conn, err := nats.Connect(cred.URL, opts...) - if err != nil { - return fmt.Errorf("connecting to the bus: %w", err) - } - defer conn.Close() - js, err := jetstream.New(conn) + ) if err != nil { return err } + defer conn.Close() // Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants // and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build // seats' steps, and says that moves are not watched (issue 1945). diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index 58fe0a7..66ce007 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -5,8 +5,6 @@ import ( "errors" "fmt" "io" - "os" - "path/filepath" "strings" "testing" @@ -15,23 +13,6 @@ import ( "github.com/nats-io/nats.go/jetstream" ) -func TestCredentialIsTheBrokerFile(t *testing.T) { // 1135/AC6 - path := filepath.Join(t.TempDir(), "broker") - os.WriteFile(path, []byte(`{"url":"nats://127.0.0.1:4222","fingerprint":"sha256:ab","node":"n","module":"mesh-delivery-watchman","user":"u","password":"p"}`), 0o600) - c, err := readCredential(path) - if err != nil || c.URL != "nats://127.0.0.1:4222" || c.User != "u" || c.Password != "p" || c.Fingerprint != "sha256:ab" { - t.Fatalf("credential = %+v, %v", c, err) - } - os.WriteFile(path, []byte(`{"url":"x"}`), 0o600) - if _, err := readCredential(path); err == nil { - t.Fatal("a credential naming no user must be refused") - } - os.WriteFile(path, []byte(`not json`), 0o600) - if _, err := readCredential(path); err == nil { - t.Fatal("a file that is not a credential must be refused") - } -} - func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) { refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x") diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index b154f77..d42c521 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-watchman go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-watchman/go.sum b/modules/mesh-delivery-watchman/go.sum index 3c78612..084deeb 100644 --- a/modules/mesh-delivery-watchman/go.sum +++ b/modules/mesh-delivery-watchman/go.sum @@ -1,7 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c h1:pY2YpHSSsfDPBrOEiWTNSIPn0zdnAayRmtmmyQ3hCHY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952 h1:TmveT3Bija643RGWkilvkzP/h8UWlVvHFZZtQfJK5oU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027193820-8ad9372b1952/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=