package main import ( "encoding/hex" "fmt" "strings" "sync" cs "fn-registry/functions/cybersecurity" "github.com/enmanuel/unibus/pkg/busauth" "github.com/enmanuel/unibus/pkg/client" "github.com/enmanuel/unibus/pkg/frame" "github.com/enmanuel/unibus/pkg/room" ) // gateway is the live web gateway: it owns the operator's identity and a single // connected unibus client, and turns the bus's crypto-bearing API into the plain // REST/SSE surface the browser consumes. The browser never signs, never speaks // NATS, and never sees a private key — the gateway is the legitimate room member // that seals/opens payloads on the browser's behalf. // // TRUST MODEL: content stays end-to-end encrypted on the wire. The gateway can // read plaintext because it acts AS the operator's client — a real member of // each room, holding the room key K like any peer. It is the same trust a native // desktop client has. In the wallet phase (per-browser WebCrypto identity) the // decryption can move into the browser; today, for the single-operator MVP, the // gateway decrypts server-side and pushes cleartext over a loopback/authenticated // SSE channel. type gateway struct { id cs.Identity endpoint string cli *client.Client refreshACL bool // call RefreshSession after a membership change (needed under a per-subject ACL bus) mu sync.Mutex hubs map[string]*roomHub // roomID -> live fan-out of decrypted frames to SSE clients } // gatewayConfig wires a live gateway. type gatewayConfig struct { Identity cs.Identity NatsURL string CtrlURL string CtrlURLs []string NatsURLs []string CAPath string // bus CA; empty => plaintext dev connection (matches a loopback membershipd) } // newGateway connects the unibus client with the operator identity following the // same posture seam every peer uses: a non-empty CA path means TLS + nkey, empty // means plaintext dev. When a CA is configured the bus is assumed to enforce a // per-subject ACL, so membership changes trigger a session refresh. func newGateway(cfg gatewayConfig) (*gateway, error) { opts := client.Options{ CtrlURLs: cfg.CtrlURLs, NatsServers: cfg.NatsURLs, } if cfg.CAPath != "" { tlsCfg, err := busauth.LoadCATLSConfig(cfg.CAPath) if err != nil { return nil, fmt.Errorf("webgw: load bus CA %q: %w", cfg.CAPath, err) } opts.UseNkey = true opts.TLS = tlsCfg opts.CtrlTLS = tlsCfg } cli, err := client.NewWithOptions(cfg.NatsURL, cfg.CtrlURL, cfg.Identity, opts) if err != nil { return nil, fmt.Errorf("webgw: connect bus client: %w", err) } return &gateway{ id: cfg.Identity, endpoint: frame.EndpointID(cfg.Identity.SignPub), cli: cli, refreshACL: cfg.CAPath != "", hubs: map[string]*roomHub{}, }, nil } // Close stops every hub and releases the bus client connection. func (g *gateway) Close() error { g.mu.Lock() for _, h := range g.hubs { h.stop() } g.hubs = map[string]*roomHub{} g.mu.Unlock() if g.cli != nil { return g.cli.Close() } return nil } // ---- wire types (browser-facing JSON) ------------------------------------ // meInfo is what GET /api/me returns: the operator identity the gateway acts as. type meInfo struct { Endpoint string `json:"endpoint"` SignPub string `json:"sign_pub"` } // roomWire is the browser view of a room. It deliberately omits messages: those // stream over SSE (GET /api/rooms/{id}/stream), not in the room list. type roomWire struct { ID string `json:"id"` Subject string `json:"subject"` Name string `json:"name"` Epoch int `json:"epoch"` Encrypt bool `json:"encrypt"` Persist bool `json:"persist"` SignMsgs bool `json:"sign_msgs"` Role string `json:"role"` } // createRoomReq is the POST /api/rooms body. Encrypt/Persist/SignMsgs are // pointers so an omitted field falls back to the chat default rather than to the // Go zero value (false). The common case — the browser sending only {subject, // encrypted} — maps encrypted onto all three (the Matrix-like chat policy). type createRoomReq struct { Subject string `json:"subject"` Encrypted *bool `json:"encrypted,omitempty"` Encrypt *bool `json:"encrypt,omitempty"` Persist *bool `json:"persist,omitempty"` SignMsgs *bool `json:"sign_msgs,omitempty"` } // policy resolves the requested policy. A bare {subject} defaults to the // Matrix-like chat room (encrypted + persisted + signed) so a created room keeps // durable, end-to-end-encrypted, authored history. Callers can override any leg. func (r createRoomReq) policy() room.Policy { enc, per, sig := true, true, true if r.Encrypted != nil { enc, per, sig = *r.Encrypted, *r.Encrypted, *r.Encrypted } if r.Encrypt != nil { enc = *r.Encrypt } if r.Persist != nil { per = *r.Persist } if r.SignMsgs != nil { sig = *r.SignMsgs } return room.Policy{Encrypt: enc, Persist: per, SignMsgs: sig} } // sendReq is the POST /api/rooms/{id}/send body. type sendReq struct { Body string `json:"body"` } // msgWire is one decrypted message pushed over SSE. type msgWire struct { ID string `json:"id"` Sender string `json:"sender"` Body string `json:"body"` TS int64 `json:"ts"` // epoch ms (decoded from the frame's ULID id) Mine bool `json:"mine"` } // ---- operations ----------------------------------------------------------- func (g *gateway) me() meInfo { return meInfo{Endpoint: g.endpoint, SignPub: hex.EncodeToString(g.id.SignPub)} } // subjectName derives a short, human-friendly room name from its bus subject by // dropping the leading namespace segment (room., test., proc., agent.). It is a // display nicety only; the canonical identity stays the subject/room id. func subjectName(subject string) string { for _, p := range []string{"room.", "test.", "proc.", "agent.", "rpc."} { if strings.HasPrefix(subject, p) { return strings.TrimPrefix(subject, p) } } return subject } func (g *gateway) listRooms() ([]roomWire, error) { rooms, err := g.cli.ListMyRooms() if err != nil { return nil, err } out := make([]roomWire, 0, len(rooms)) for _, rm := range rooms { out = append(out, roomWire{ ID: rm.RoomID, Subject: rm.Subject, Name: subjectName(rm.Subject), Epoch: rm.Epoch, Encrypt: rm.Policy.Encrypt, Persist: rm.Policy.Persist, SignMsgs: rm.Policy.SignMsgs, Role: rm.Role, }) } return out, nil } func (g *gateway) createRoom(req createRoomReq) (roomWire, error) { subject := strings.TrimSpace(req.Subject) if subject == "" { return roomWire{}, fmt.Errorf("webgw: subject required") } p := req.policy() roomID, err := g.cli.CreateRoom(subject, p) if err != nil { return roomWire{}, err } // Under a per-subject ACL the operator's frozen NATS permissions do not yet // cover the new room's subject; refresh so subsequent data-plane use works. On // a plaintext/non-ACL dev bus this is unnecessary and would needlessly drop any // live SSE subscriptions, so it is gated on the secured posture. if g.refreshACL { _ = g.cli.RefreshSession() } return roomWire{ ID: roomID, Subject: subject, Name: subjectName(subject), Epoch: 1, Encrypt: p.Encrypt, Persist: p.Persist, SignMsgs: p.SignMsgs, Role: "owner", }, nil } // join resolves room metadata and (for encrypted rooms) fetches the room key so // the gateway can later open payloads. Idempotent. func (g *gateway) join(roomID string) error { if err := g.cli.Join(roomID); err != nil { return err } if g.refreshACL { _ = g.cli.RefreshSession() } return nil } // send publishes plaintext to a room. The unibus client seals it with the room // key (encrypted rooms) and signs it (signed rooms) before it leaves the process. func (g *gateway) send(roomID, body string) error { return g.cli.Publish(roomID, []byte(body)) }