egutierrez
2f5b372a80
A secured bus freezes per-subject permissions at connect time, so a peer that creates or joins a room after connecting cannot pub/sub on it until it reconnects (RefreshSession). No client called it, so under enforce+ACL the demos failed closed — pushing the operator to disable the ACL (a security regression at the operator's discretion). Wire the membership-change contract into every client: - cmd/worker: RefreshSession after CreateRoom, before publishing. - cmd/chat (simple): RefreshSession after CreateRoom+Join, before Subscribe. - cmd/chat (encrypted demo): A refreshes after CreateRoom; B refreshes after the invite+join, both before pub/sub. - local_files/bridge (gateway): RefreshSession after CreateRoom+Join, before Subscribe. - mobile: new Session.RefreshSession wrapper + the contract documented for callers. Contract (documented on the wrappers): after ANY membership change, call RefreshSession BEFORE pub/sub on the new room (it drops active subs, so it must precede Subscribe). On an unsecured/dev bus it is a harmless reconnect. Test: - TestClientCreateRoomRefreshPublishFlow: end-to-end under enforce+ACL, a peer creates a room, refreshes, invites a second peer who joins+refreshes+subscribes, and the publish is received — no manual intervention, the ACL stays on. CGO_ENABLED=0 go build/vet/test green; govulncheck 0 reachable. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
80 lines
2.4 KiB
Go
80 lines
2.4 KiB
Go
// Command worker is a demo peer: it creates (or joins) a cleartext room and
|
|
// publishes an incrementing counter once per second, to both stdout and the
|
|
// bus. It demonstrates that a process is a first-class bus peer, uniform with
|
|
// the human chat client.
|
|
package main
|
|
|
|
import (
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/enmanuel/unibus/pkg/client"
|
|
"github.com/enmanuel/unibus/pkg/room"
|
|
)
|
|
|
|
func main() {
|
|
var (
|
|
natsURL = flag.String("nats-url", "nats://127.0.0.1:4250", "NATS url")
|
|
ctrlURL = flag.String("ctrl-url", "http://127.0.0.1:8470", "membershipd control-plane url")
|
|
roomSub = flag.String("room", "proc.test.ticks", "room subject to publish to")
|
|
idFile = flag.String("id-file", "./local_files/worker.id", "identity file path")
|
|
caFile = flag.String("ca", "", "path to the bus CA cert (ca.crt); set to connect with TLS + nkey to a secured bus")
|
|
)
|
|
flag.Parse()
|
|
|
|
log.SetFlags(log.LstdFlags | log.Lmsgprefix)
|
|
log.SetPrefix("[worker] ")
|
|
|
|
id, err := client.LoadOrCreateIdentity(*idFile)
|
|
if err != nil {
|
|
log.Fatalf("identity: %v", err)
|
|
}
|
|
c, err := client.Connect(*natsURL, *ctrlURL, id, *caFile)
|
|
if err != nil {
|
|
log.Fatalf("connect: %v", err)
|
|
}
|
|
defer c.Close()
|
|
log.Printf("endpoint: %s", c.Endpoint().ID)
|
|
|
|
// Create the room; if it already exists we cannot recreate it under a known
|
|
// id (rooms get fresh ULIDs), so for the demo each worker run owns its room.
|
|
roomID, err := c.CreateRoom(*roomSub, room.ModeNATS)
|
|
if err != nil {
|
|
log.Fatalf("create room: %v", err)
|
|
}
|
|
// Membership-change contract (issue 0006e): the bus freezes per-subject
|
|
// permissions at connect time, and this room did not exist then. Refresh the
|
|
// session so the new room's subject becomes publishable under enforce+ACL. On
|
|
// an unsecured/dev bus this is a harmless reconnect.
|
|
if err := c.RefreshSession(); err != nil {
|
|
log.Fatalf("refresh session after create room: %v", err)
|
|
}
|
|
log.Printf("room %q -> %s (subject %s, cleartext)", *roomSub, roomID, *roomSub)
|
|
|
|
stop := make(chan os.Signal, 1)
|
|
signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
|
|
|
|
ticker := time.NewTicker(1 * time.Second)
|
|
defer ticker.Stop()
|
|
n := 0
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
n++
|
|
payload := fmt.Sprintf("tick %d @ %s", n, time.Now().UTC().Format(time.RFC3339))
|
|
fmt.Println(payload)
|
|
if err := c.Publish(roomID, []byte(payload)); err != nil {
|
|
log.Printf("publish: %v", err)
|
|
}
|
|
case <-stop:
|
|
log.Printf("stopping after %d ticks", n)
|
|
return
|
|
}
|
|
}
|
|
}
|