diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 00eeae11d..2227ff85a 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -31,9 +31,11 @@ jobs: - name: Test run: go test -v ./... + # Pinned: golang.org/x/tools v0.50.0 and later require Go 1.26, which + # breaks the Go 1.25 job. v0.49.0 is the newest release on Go 1.25. - name: Install goimports run: | - go install golang.org/x/tools/cmd/goimports@latest + go install golang.org/x/tools/cmd/goimports@v0.49.0 export PATH="$HOME/go/bin:$PATH" - name: Run pre-commit diff --git a/CHANGELOG.md b/CHANGELOG.md index 5a435af87..88a7245c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,18 @@ All notable HyperMeow changes are documented here. HyperMeow uses commit pseudo- ## [Unreleased] +### Added + +- Added connection lifecycle observers with the original `ConnectContext` for Connected, Disconnected, QR, LoggedOut and StreamReplaced. Existing generic event delivery remains unchanged; delayed lifecycle events retain their original context. + +- Added remote Signal keys: `NewRemoteSignalClient` runs a linked device whose Signal private keys, sessions and message plaintext live in an external store reached through `RemoteSignal`. The client keeps its socket, receipts, retry receipts, prekey uploads and QR pairing, hands each received payload to the store, receives only delivery metadata (`events.RemoteMessage`), and transmits stanzas the store built with `SendPrebuiltStanza`. The store also handles identity changes (`ForgetIdentity`) and phone-number-to-LID session moves (`MigratePNToLID`). Pair-code pairing, local `SendMessage` and `SendFBMessage` return `ErrRemoteSignalUnsupported` for such clients, and `PreRetryCallback` receives a nil message. + +### Changed + +- Synchronized native disconnect callback ownership during concurrent socket stop/close. + +- `RemoteSignal` implementations must provide `PreparePreKeyUpload` and `AcknowledgePreKeyUpload`. Remote prekey uploads recover the full durable pending batch across restarts and count changes, acknowledge exact ordered snapshots across ID wrap, and retry lost ACK responses without generating replacement keys. Fresh retry-receipt generation remains separate. See [remote Signal stores](README.md#remote-signal-stores). + ### Documentation - Added an evidence-backed comparison with upstream WhatsMeow. diff --git a/README.md b/README.md index 7b9e08e8d..893eb55e5 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,42 @@ HyperMeow retains upstream's core support for: New exported functionality is additive unless called out in the [changelog](CHANGELOG.md). +### Remote Signal stores + +`NewRemoteSignalClient` uses a customer-held Signal store through `RemoteSignal`. +The store supplies public prekeys and builds encrypted message stanzas; private +keys and message plaintext stay outside HyperMeow. Applications implementing the +interface must implement both durable upload operations: + +- `PreparePreKeyUpload(ctx, count)` returns a `RemotePreKeyUploadBatch` with an + opaque 32-character lowercase hexadecimal `BatchID` and 1–812 public `PreKeys`. + `count` selects a fresh batch size. An outstanding batch returns all surviving + keys in its original order, even when its size differs from `count`. Persist + that batch across service restarts; do not truncate it or generate a top-up. +- `AcknowledgePreKeyUpload(ctx, batchID, ids)` commits the complete upload after + the WhatsApp upload IQ succeeds. IDs can cross the 24-bit wrap boundary. The + acknowledgement covers every surviving key; consumed original members can + remain in the list and must not be recreated. Replaying the last exact ACK + after a lost response succeeds without changing a newer batch. + +The adapter captures independent public-key and ID snapshots. It checks the +literal last uploaded ID and acknowledges the entire captured list. A failed IQ +does not acknowledge the batch. After an attempted ACK fails, the adapter retains +it and resolves that exact ACK before preparing another batch. Unsupported remote +operations fail; there is no fresh-generation fallback. Upload serialization and +the adapter mutex keep upload state separate from retry-receipt generation. + +`GeneratePreKeys` and `MarkPreKeysUploaded` remain separate operations for fresh +retry-receipt keys. `GenOnePreKey` generates one fresh key and marks it published; +it never borrows from a prepared upload batch. Remote implementations must reject +legacy marking of prepared-batch members. Returned one-time prekeys must have +unique nonzero 24-bit IDs, a nonzero 32-byte public key, and no private key or +signed-prekey signature. + +This interface change requires matching remote-store implementations. It does +not establish that a service is deployed or a feature is available to customers. +Production consumers continue to pin reviewed `main` pseudo-versions. + ## Discord Join the [WhatsApp Web ecosystem Discord](https://whiskey.so/discord) and visit @@ -134,3 +170,22 @@ Polymorfa-added files: Copyright (c) 2026 Rajeh Taher, licensed under the [MIT License](LICENSE-MIT). See [LICENSING.md](LICENSING.md) for the file-level boundary. + +### Connection lifecycle observers + +`AddConnectionEventHandler(func(context.Context, any))` observes `Connected`, +`Disconnected`, `QR`, `LoggedOut` and `StreamReplaced`. The context is the original +connection's `ConnectContext` argument, captured before asynchronous delivery. +A delayed event can arrive after that context is cancelled. Compare it with the +connection authority your application owns before updating state; do not label an +old event using the newest connection. `RemoveConnectionEventHandler(id)` removes +a registration; an already captured callback can still arrive. Callbacks run +without the observer registry lock and can remove themselves. + +Generic `AddEventHandler` consumers still receive these events. Other event types +retain their existing API. Connection observers do not authorize a send, undo +bytes already sent or prove that a cancelled socket has finished closing. +Disconnect and wait for owned work to settle before accepting replacement +connection authority. `FrameSocket.SetDisconnectHandler` synchronizes callback +ownership with socket shutdown; assign the legacy field only before connecting. +These additions are unreleased until their reviewed main commit is adopted. diff --git a/client.go b/client.go index 53fcf872b..5aa4eb01a 100644 --- a/client.go +++ b/client.go @@ -163,9 +163,11 @@ type Client struct { // closed switch) lets a headless shadow Client dispatch injected nodes // synchronously and lets embedders that reach for it via reflection keep // working. - nodeHandlers map[string]nodeHandler - eventHandlers []wrappedEventHandler - eventHandlersLock sync.RWMutex + nodeHandlers map[string]nodeHandler + eventHandlers []wrappedEventHandler + eventHandlersLock sync.RWMutex + connectionEventHandlers []wrappedConnectionEventHandler + connectionEventHandlersLock sync.Mutex messageRetries map[string]int messageRetriesLock sync.Mutex @@ -208,6 +210,8 @@ type Client struct { GetMessageForRetry func(requester, to types.JID, id types.MessageID) *waE2E.Message // PreRetryCallback is called before a retry receipt is accepted. // If it returns false, the accepting will be cancelled and the retry receipt will be ignored. + // On a client from NewRemoteSignalClient, msg is always nil: the remote + // Signal store keeps sent messages and this client never has their content. PreRetryCallback func(receipt *events.Receipt, id types.MessageID, retryCount int, msg *waE2E.Message) bool // Should whatsmeow store recently sent messages in the database so that retry receipts can be accepted // even if the process is restarted? If false, only the in-memory cache and GetMessageForRetry will be used. @@ -236,6 +240,15 @@ type Client struct { // client (see [NewShadowClient]): it has no socket, Connect is guarded, // and outbound nodes plus Signal/keying oracle ops are delegated here. shadowRelay ShadowRelay + // remoteSignal holds this device's Signal keys when they live outside + // the client; see NewRemoteSignalClient. + remoteSignal RemoteSignal + // remoteMigratedPN records phone-number users whose Signal state the + // remote store has already moved to their LID. + remoteMigratedPN sync.Map + // testSendHook captures marshaled outbound nodes when there is no + // socket. Only tests set it. + testSendHook func(payload []byte) error // DisabledFeatures controls which built-in processing paths are // skipped. See [DisabledFeatures]. @@ -582,7 +595,7 @@ func (cli *Client) ConnectContext(ctx context.Context) error { err := cli.unlockedConnect(ctx) if isRetryableConnectError(err) && cli.InitialAutoReconnect && cli.EnableAutoReconnect { cli.Log.Errorf("Initial connection failed but reconnecting in background (%v)", err) - go cli.dispatchEvent(&events.Disconnected{}) + go cli.dispatchConnectionEvent(ctx, &events.Disconnected{}) go cli.autoReconnect(ctx) return nil } @@ -662,7 +675,7 @@ func (cli *Client) onDisconnect(ctx context.Context, ns *socket.NoiseSocket, rem cli.clearResponseWaiters(xmlStreamEndNode) if !cli.isExpectedDisconnect() && (cli.forceAutoReconnect.Swap(false) || remote) { cli.Log.Debugf("Emitting Disconnected event") - go cli.dispatchEvent(&events.Disconnected{}) + go cli.dispatchConnectionEvent(ctx, &events.Disconnected{}) go cli.autoReconnect(ctx) } else if remote { cli.Log.Debugf("OnDisconnect() called, but it was expected, so not emitting event") @@ -1064,6 +1077,13 @@ func (cli *Client) sendNodeAndGetData(ctx context.Context, node waBinary.Node) ( cli.sendLog.Debugf("%s", &node) return payload, cli.shadowRelay.SendNode(ctx, payload) } + if cli.testSendHook != nil { + payload, err := waBinary.Marshal(node) + if err != nil { + return nil, fmt.Errorf("failed to marshal node: %w", err) + } + return payload, cli.testSendHook(payload) + } return nil, ErrNotConnected } diff --git a/connection_lifecycle.go b/connection_lifecycle.go new file mode 100644 index 000000000..6c1c151df --- /dev/null +++ b/connection_lifecycle.go @@ -0,0 +1,73 @@ +// Copyright (c) 2026 Rajeh Taher +// Licensed under the MIT License. See LICENSE-MIT for details. + +package whatsmeow + +import ( + "context" + "sync/atomic" + + "github.com/polymorfa/hypermeow/types/events" +) + +// ConnectionEventHandler receives Connected, Disconnected, QR, LoggedOut and +// StreamReplaced with the original connection's ConnectContext. The context may +// already be cancelled when a delayed event arrives. Consumers must check their +// own current connection authority; an event does not establish ownership. +// Other events continue through the existing generic event API only. +type ConnectionEventHandler func(context.Context, any) + +type wrappedConnectionEventHandler struct { + fn ConnectionEventHandler + id uint32 +} + +// AddConnectionEventHandler adds a lifecycle observer without replacing generic +// event handlers. A handler may reenter this registry without holding its lock. +func (cli *Client) AddConnectionEventHandler(handler ConnectionEventHandler) uint32 { + id := atomic.AddUint32(&nextHandlerID, 1) + cli.connectionEventHandlersLock.Lock() + cli.connectionEventHandlers = append(cli.connectionEventHandlers, wrappedConnectionEventHandler{handler, id}) + cli.connectionEventHandlersLock.Unlock() + return id +} + +// RemoveConnectionEventHandler removes a lifecycle observer. A callback already +// captured by an in-flight dispatch can still arrive with its original context. +func (cli *Client) RemoveConnectionEventHandler(id uint32) bool { + cli.connectionEventHandlersLock.Lock() + defer cli.connectionEventHandlersLock.Unlock() + for i, handler := range cli.connectionEventHandlers { + if handler.id == id { + copy(cli.connectionEventHandlers[i:], cli.connectionEventHandlers[i+1:]) + cli.connectionEventHandlers[len(cli.connectionEventHandlers)-1] = wrappedConnectionEventHandler{} + cli.connectionEventHandlers = cli.connectionEventHandlers[:len(cli.connectionEventHandlers)-1] + return true + } + } + return false +} + +// ctx is supplied by the original connection producer before any asynchronous +// scheduling. Never replace it with client.BackgroundEventCtx or current state. +func (cli *Client) dispatchConnectionEvent(ctx context.Context, evt any) (handlerFailed bool) { + switch evt.(type) { + case *events.Connected, *events.Disconnected, *events.QR, *events.LoggedOut, *events.StreamReplaced: + cli.connectionEventHandlersLock.Lock() + handlers := append([]wrappedConnectionEventHandler(nil), cli.connectionEventHandlers...) + cli.connectionEventHandlersLock.Unlock() + for _, handler := range handlers { + invokeConnectionEventHandler(cli, handler.fn, ctx, evt) + } + } + return cli.dispatchEvent(evt) +} + +func invokeConnectionEventHandler(cli *Client, handler ConnectionEventHandler, ctx context.Context, evt any) { + defer func() { + if recover() != nil { + cli.Log.Errorf("Connection lifecycle event handler panicked") + } + }() + handler(ctx, evt) +} diff --git a/connection_lifecycle_test.go b/connection_lifecycle_test.go new file mode 100644 index 000000000..ceeff070e --- /dev/null +++ b/connection_lifecycle_test.go @@ -0,0 +1,74 @@ +// Copyright (c) 2026 Rajeh Taher +// Licensed under the MIT License. See LICENSE-MIT for details. + +package whatsmeow + +import ( + "context" + "sync/atomic" + "testing" + "time" + + waBinary "github.com/polymorfa/hypermeow/binary" + "github.com/polymorfa/hypermeow/types/events" +) + +func TestConnectionLifecycleRetainsOriginalContextAfterDelayedDispatch(t *testing.T) { + cli, _, _ := newTestRemote(t) + type sourceKey struct{} + original, cancel := context.WithCancel(context.WithValue(context.Background(), sourceKey{}, "old")) + defer cancel() + entered, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) + generic := atomic.Int32{} + cli.AddEventHandler(func(evt any) { + if _, ok := evt.(*events.StreamReplaced); ok { + generic.Add(1) + close(done) + } + }) + var got context.Context + id := cli.AddConnectionEventHandler(func(ctx context.Context, evt any) { + if _, ok := evt.(*events.StreamReplaced); !ok { + t.Error("wrong event") + } + got = ctx + close(entered) + <-release + }) + cli.handleStreamError(original, &waBinary.Node{Tag: "stream:error", Content: []waBinary.Node{{Tag: "conflict", Attrs: waBinary.Attrs{"type": "replaced"}}}}) + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("actual lifecycle producer did not dispatch") + } + cancel() + cli.BackgroundEventCtx = context.WithValue(context.Background(), sourceKey{}, "new") + close(release) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("generic callback lost") + } + if got != original || got.Value(sourceKey{}) != "old" || got.Err() != context.Canceled { + t.Fatal("late lifecycle event relabelled with current context") + } + if generic.Load() != 1 { + t.Fatal("generic consumers changed") + } + if !cli.RemoveConnectionEventHandler(id) || cli.RemoveConnectionEventHandler(id) { + t.Fatal("scoped callback removal failed") + } +} + +func TestConnectionCallbacksCanReenterRegistryAndGenericDeliveryRemains(t *testing.T) { + cli, _, _ := newTestRemote(t) + var scoped, generic int + var id uint32 + id = cli.AddConnectionEventHandler(func(context.Context, any) { scoped++; cli.RemoveConnectionEventHandler(id) }) + cli.AddEventHandler(func(any) { generic++ }) + cli.dispatchConnectionEvent(context.Background(), &events.Connected{}) + cli.dispatchConnectionEvent(context.Background(), &events.Connected{}) + if scoped != 1 || generic != 2 { + t.Fatal("scoped handler reentry or generic delivery failed") + } +} diff --git a/connection_native_retirement_test.go b/connection_native_retirement_test.go new file mode 100644 index 000000000..e39f2ff3c --- /dev/null +++ b/connection_native_retirement_test.go @@ -0,0 +1,250 @@ +// Copyright (c) 2026 Rajeh Taher +// Licensed under the MIT License. See LICENSE-MIT for details. + +package whatsmeow + +import ( + "context" + "encoding/binary" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/polymorfa/libsignal-protocol-go/ecc" + "google.golang.org/protobuf/proto" + + waBinary "github.com/polymorfa/hypermeow/binary" + "github.com/polymorfa/hypermeow/proto/waCert" + "github.com/polymorfa/hypermeow/proto/waWa6" + "github.com/polymorfa/hypermeow/socket" + "github.com/polymorfa/hypermeow/store" + "github.com/polymorfa/hypermeow/util/keys" + waLog "github.com/polymorfa/hypermeow/util/log" +) + +// nativeWriteGate uses a real loopback TCP connection. Once armed, it blocks +// the native writer until the underlying connection is physically closed. +// Cancellation of the per-request context alone cannot satisfy this barrier. +type nativeWriteGate struct { + net.Conn + armed atomic.Bool + entered, settled, closed chan struct{} + enterOnce, settleOnce, closeOnce sync.Once +} + +func (g *nativeWriteGate) Write(p []byte) (int, error) { + if g.armed.Load() { + g.enterOnce.Do(func() { close(g.entered) }) + <-g.closed + g.settleOnce.Do(func() { close(g.settled) }) + return 0, net.ErrClosed + } + return g.Conn.Write(p) +} +func (g *nativeWriteGate) Close() error { + g.closeOnce.Do(func() { close(g.closed) }) + return g.Conn.Close() +} + +func nativeCertificate(t *testing.T, root, intermediate, static *keys.KeyPair) []byte { + t.Helper() + detail := func(serial, issuer uint32, key []byte) []byte { + raw, err := proto.Marshal(&waCert.CertChain_NoiseCertificate_Details{Serial: proto.Uint32(serial), IssuerSerial: proto.Uint32(issuer), Key: key, NotBefore: proto.Uint64(uint64(time.Now().Add(-time.Hour).Unix())), NotAfter: proto.Uint64(uint64(time.Now().Add(time.Hour).Unix()))}) + if err != nil { + t.Fatal(err) + } + return raw + } + mid, leaf := detail(1, 0, intermediate.Pub[:]), detail(2, 1, static.Pub[:]) + sign := func(key *keys.KeyPair, raw []byte) []byte { + sig := ecc.CalculateSignature(ecc.NewDjbECPrivateKey(*key.Priv), raw) + return sig[:] + } + raw, err := proto.Marshal(&waCert.CertChain{Intermediate: &waCert.CertChain_NoiseCertificate{Details: mid, Signature: sign(root, mid)}, Leaf: &waCert.CertChain_NoiseCertificate{Details: leaf, Signature: sign(intermediate, leaf)}}) + if err != nil { + t.Fatal(err) + } + return raw +} + +// nativeRetirementClient completes real Noise XX using a synthetic local CA. +// The peer then intentionally never reads or responds to a close handshake. +func nativeRetirementClient(t *testing.T) (*Client, *nativeWriteGate, context.CancelFunc) { + t.Helper() + root, mid, static := keys.NewKeyPair(), keys.NewKeyPair(), keys.NewKeyPair() + cert := nativeCertificate(t, root, mid, static) + finish, serverStop := make(chan error, 1), make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ws, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true, CompressionMode: websocket.CompressionContextTakeover}) + if err != nil { + finish <- err + return + } + defer ws.CloseNow() + read := func(header bool) (*waWa6.HandshakeMessage, error) { + _, raw, err := ws.Read(r.Context()) + if err != nil { + return nil, err + } + if header { + if len(raw) < 4 { + return nil, io.ErrUnexpectedEOF + } + raw = raw[4:] + } + if len(raw) < 3 { + return nil, io.ErrUnexpectedEOF + } + n := int(raw[0])<<16 | int(raw[1])<<8 | int(raw[2]) + if n != len(raw)-3 { + return nil, errors.New("fixture frame length") + } + var msg waWa6.HandshakeMessage + err = proto.Unmarshal(raw[3:], &msg) + return &msg, err + } + hello, err := read(true) + if err != nil { + finish <- err + return + } + clientEphemeral := hello.GetClientHello().GetEphemeral() + if len(clientEphemeral) != 32 { + finish <- errors.New("fixture ephemeral") + return + } + ephem := keys.NewKeyPair() + nh := socket.NewNoiseHandshake() + nh.Start(socket.NoiseStartPattern, socket.WAConnHeader) + nh.Authenticate(clientEphemeral) + nh.Authenticate(ephem.Pub[:]) + if err = nh.MixSharedSecretIntoKey(*ephem.Priv, [32]byte(clientEphemeral)); err != nil { + finish <- err + return + } + encryptedStatic := nh.Encrypt(static.Pub[:]) + if err = nh.MixSharedSecretIntoKey(*static.Priv, [32]byte(clientEphemeral)); err != nil { + finish <- err + return + } + raw, _ := proto.Marshal(&waWa6.HandshakeMessage{ServerHello: &waWa6.HandshakeMessage_ServerHello{Ephemeral: ephem.Pub[:], Static: encryptedStatic, Payload: nh.Encrypt(cert)}}) + framed := make([]byte, len(raw)+3) + binary.BigEndian.PutUint16(framed[1:3], uint16(len(raw))) + copy(framed[3:], raw) + if err = ws.Write(r.Context(), websocket.MessageBinary, framed); err != nil { + finish <- err + return + } + final, err := read(false) + if err != nil { + finish <- err + return + } + clientStatic, err := nh.Decrypt(final.GetClientFinish().GetStatic()) + if err != nil || len(clientStatic) != 32 { + finish <- errors.New("fixture client static") + return + } + if err = nh.MixSharedSecretIntoKey(*ephem.Priv, [32]byte(clientStatic)); err != nil { + finish <- err + return + } + if _, err = nh.Decrypt(final.GetClientFinish().GetPayload()); err != nil { + finish <- err + return + } + finish <- nil + <-serverStop + })) + t.Cleanup(func() { close(serverStop); server.Close() }) + var gate *nativeWriteGate + transport := &http.Transport{DialContext: func(ctx context.Context, network, address string) (net.Conn, error) { + c, err := (&net.Dialer{}).DialContext(ctx, network, address) + if err != nil { + return nil, err + } + gate = &nativeWriteGate{Conn: c, entered: make(chan struct{}), settled: make(chan struct{}), closed: make(chan struct{})} + return gate, nil + }} + client := NewRemoteSignalClient(&store.Device{NoiseKey: keys.NewKeyPair()}, testRemoteIdentity, &fakeRemoteSignal{}, waLog.Noop) + client.EnableAutoReconnect = false + client.SocketConfig = &SocketConfig{URL: "ws" + strings.TrimPrefix(server.URL, "http"), NoiseCertificateAuthority: root.Pub} + client.SetPreLoginHTTPClient(&http.Client{Transport: transport}) + client.SetWebsocketHTTPClient(&http.Client{Transport: transport}) + lifetime, cancel := context.WithCancel(context.Background()) + t.Cleanup(func() { cancel(); client.Disconnect(); transport.CloseIdleConnections() }) + if err := client.ConnectContext(lifetime); err != nil { + t.Fatal(err) + } + select { + case err := <-finish: + if err != nil { + t.Fatal(err) + } + case <-time.After(3 * time.Second): + t.Fatal("native handshake did not complete") + } + return client, gate, cancel +} + +func TestPinnedNativeSocketCancellationRequiresPhysicalRetirement(t *testing.T) { + client, gate, cancelLifetime := nativeRetirementClient(t) + raw, err := waBinary.Marshal(waBinary.Node{Tag: "message", Attrs: waBinary.Attrs{"id": "synthetic-retirement", "to": "1@s.whatsapp.net"}, Content: []waBinary.Node{{Tag: "enc", Attrs: waBinary.Attrs{"type": "msg", "v": "2"}, Content: []byte{1, 2, 3}}}}) + if err != nil { + t.Fatal(err) + } + gate.armed.Store(true) + req, cancelRequest := context.WithCancel(context.Background()) + returned := make(chan error, 1) + go func() { _, err := client.SendPrebuiltStanza(req, raw, "", 30*time.Second); returned <- err }() + select { + case <-gate.entered: + case <-time.After(3 * time.Second): + t.Fatal("native socket write not entered") + } + cancelRequest() + select { + case err := <-returned: + if err == nil { + t.Fatal("cancelled request succeeded") + } + case <-time.After(3 * time.Second): + t.Fatal("native request ignored cancellation") + } + select { + case <-gate.closed: + t.Fatal("per-request cancellation unexpectedly closed physical socket") + default: + } + select { + case <-gate.settled: + t.Fatal("detached write falsely settled") + default: + } + cancelLifetime() + stopped := make(chan struct{}) + go func() { client.Disconnect(); close(stopped) }() + select { + case <-stopped: + case <-time.After(3 * time.Second): + t.Fatal("native Disconnect did not settle after lease cancellation") + } + select { + case <-gate.closed: + default: + t.Fatal("Disconnect returned with old TCP socket open") + } + select { + case <-gate.settled: + case <-time.After(3 * time.Second): + t.Fatal("detached writer did not settle") + } +} diff --git a/connectionevents.go b/connectionevents.go index 3e52853d3..b701a950f 100644 --- a/connectionevents.go +++ b/connectionevents.go @@ -40,7 +40,7 @@ func (cli *Client) handleStreamError(ctx context.Context, node *waBinary.Node) { case code == "401" && conflictType == "device_removed": cli.expectDisconnect() cli.Log.Infof("Got device removed stream error, sending LoggedOut event and deleting session") - go cli.dispatchEvent(&events.LoggedOut{OnConnect: false, Reason: events.ConnectFailureLoggedOut}) + go cli.dispatchConnectionEvent(ctx, &events.LoggedOut{OnConnect: false, Reason: events.ConnectFailureLoggedOut}) err := cli.Store.Delete(ctx) if err != nil { cli.Log.Warnf("Failed to delete store after device_removed error: %v", err) @@ -48,7 +48,7 @@ func (cli *Client) handleStreamError(ctx context.Context, node *waBinary.Node) { case conflictType == "replaced": cli.expectDisconnect() cli.Log.Infof("Got replaced stream error, sending StreamReplaced event") - go cli.dispatchEvent(&events.StreamReplaced{}) + go cli.dispatchConnectionEvent(ctx, &events.StreamReplaced{}) case code == "503": // This seems to happen when the server wants to restart or something. // The disconnection will be emitted as an events.Disconnected and then the auto-reconnect will do its thing. @@ -125,7 +125,7 @@ func (cli *Client) handleConnectFailure(ctx context.Context, node *waBinary.Node } if reason.IsLoggedOut() { cli.Log.Infof("Got %s connect failure, sending LoggedOut event and deleting session", reason) - go cli.dispatchEvent(&events.LoggedOut{OnConnect: true, Reason: reason}) + go cli.dispatchConnectionEvent(ctx, &events.LoggedOut{OnConnect: true, Reason: reason}) err := cli.Store.Delete(ctx) if err != nil { cli.Log.Warnf("Failed to delete store after %d failure: %v", int(reason), err) @@ -201,7 +201,7 @@ func (cli *Client) handleConnectSuccess(ctx context.Context, node *waBinary.Node if err != nil { cli.Log.Warnf("Failed to send post-connect passive IQ: %v", err) } - cli.dispatchEvent(&events.Connected{}) + cli.dispatchConnectionEvent(ctx, &events.Connected{}) cli.closeSocketWaitChan() }() } diff --git a/message.go b/message.go index 17f8511ca..05f7de092 100644 --- a/message.go +++ b/message.go @@ -331,6 +331,10 @@ func (cli *Client) handlePlaintextMessage(ctx context.Context, info *types.Messa } func (cli *Client) migrateSessionStore(ctx context.Context, pn, lid types.JID) { + if cli.isRemoteSignal() { + cli.migrateRemoteSessions(ctx, pn, lid) + return + } err := cli.Store.Sessions.MigratePNToLID(ctx, pn, lid) if err != nil { cli.Log.Errorf("Failed to migrate signal store from %s to %s: %v", pn, lid, err) @@ -381,6 +385,10 @@ func (cli *Client) decryptMessages(ctx context.Context, info *types.MessageInfo, info.SenderAlt = lid } } + if cli.isRemoteSignal() { + cli.decryptMessagesRemote(ctx, info, node, senderEncryptionJID) + return + } var recognizedStanza, protobufFailed bool for _, child := range children { if child.Tag != "enc" { diff --git a/notification.go b/notification.go index 7bfe51954..69f3c9ebc 100644 --- a/notification.go +++ b/notification.go @@ -64,6 +64,10 @@ func (cli *Client) handleEncryptNotification(ctx context.Context, node *waBinary } func (cli *Client) deleteIdentityChangeState(ctx context.Context, from types.JID) { + if cli.isRemoteSignal() { + cli.forgetRemoteIdentity(ctx, from) + return + } users := identityChangeSignalUsers(from) if cli.Store.LIDs != nil { var alternate types.JID diff --git a/pair-code.go b/pair-code.go index b7f2c0db8..ca5867790 100644 --- a/pair-code.go +++ b/pair-code.go @@ -90,6 +90,11 @@ func generateCompanionEphemeralKey() (ephemeralKeyPair *keys.KeyPair, ephemeralK // // See https://faq.whatsapp.com/1324084875126592 for more info func (cli *Client) PairPhone(ctx context.Context, phone string, showPushNotification bool, clientType PairClientType, clientDisplayName string) (string, error) { + if cli.isRemoteSignal() { + // Code pairing runs X25519 with the identity private key; the remote + // store does not expose that step yet. + return "", fmt.Errorf("pair-code pairing: %w", ErrRemoteSignalUnsupported) + } if cli == nil { return "", ErrClientIsNil } @@ -150,6 +155,9 @@ func (cli *Client) tryHandleCodePairNotification(ctx context.Context, parentNode } func (cli *Client) handleCodePairNotification(ctx context.Context, parentNode *waBinary.Node) error { + if cli.isRemoteSignal() { + return fmt.Errorf("pair-code pairing: %w", ErrRemoteSignalUnsupported) + } node, ok := parentNode.GetOptionalChildByTag("link_code_companion_reg") if !ok { return &ElementMissingError{ diff --git a/pair.go b/pair.go index 0968ea599..47ff02e7b 100644 --- a/pair.go +++ b/pair.go @@ -77,7 +77,7 @@ func (cli *Client) handlePairDevice(ctx context.Context, node *waBinary.Node) { evt.Codes = append(evt.Codes, cli.makeQRData(content, cli.getQRClientType())) } - cli.dispatchEvent(evt) + cli.dispatchConnectionEvent(ctx, evt) } func (cli *Client) getQRClientType() PairClientType { @@ -189,7 +189,16 @@ func (cli *Client) handlePair(ctx context.Context, deviceIdentityBytes []byte, r return ErrPairInvalidDeviceSignature } - deviceIdentity.DeviceSignature = generateDeviceSignature(&deviceIdentity, cli.Store.IdentityKey)[:] + if cli.isRemoteSignal() { + signature, err := cli.remoteDeviceSignature(ctx, deviceIdentityBytes, &deviceIdentity) + if err != nil { + cli.sendPairError(ctx, reqID, 500, "internal-error") + return err + } + deviceIdentity.DeviceSignature = signature + } else { + deviceIdentity.DeviceSignature = generateDeviceSignature(&deviceIdentity, cli.Store.IdentityKey)[:] + } if cli.PrePairCallback != nil && !cli.PrePairCallback(jid, platform, businessName) { cli.sendPairError(ctx, reqID, 500, "internal-error") @@ -219,6 +228,13 @@ func (cli *Client) handlePair(ctx context.Context, deviceIdentityBytes []byte, r return &PairDatabaseError{"failed to save device store", err} } cli.StoreLIDPNMapping(ctx, lid, jid) + if cli.isRemoteSignal() { + if err = cli.remoteSignal.SetOwnJIDs(ctx, jid, lid); err != nil { + _ = cli.Store.Delete(ctx) + cli.sendPairError(ctx, reqID, 500, "internal-error") + return fmt.Errorf("failed to give the remote Signal store its own JIDs: %w", err) + } + } err = cli.Store.Identities.PutIdentity(ctx, mainDeviceLID.SignalAddress().String(), mainDeviceIdentity) if err != nil { _ = cli.Store.Delete(ctx) diff --git a/remote_prekey_upload_test.go b/remote_prekey_upload_test.go new file mode 100644 index 000000000..cf3436b25 --- /dev/null +++ b/remote_prekey_upload_test.go @@ -0,0 +1,390 @@ +// Copyright (c) 2026 Rajeh Taher +// Licensed under the MIT License. See LICENSE-MIT for details. + +package whatsmeow + +import ( + "context" + "encoding/binary" + "errors" + "fmt" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + waBinary "github.com/polymorfa/hypermeow/binary" + "github.com/polymorfa/hypermeow/store" + "github.com/polymorfa/hypermeow/util/keys" +) + +func preKeyIDs(batch []*keys.PreKey) []uint32 { + ids := make([]uint32, len(batch)) + for i, key := range batch { + ids[i] = key.KeyID + } + return ids +} + +func publicTestPreKey(id uint32) *keys.PreKey { + pub := [32]byte{1} + return &keys.PreKey{KeyPair: keys.KeyPair{Pub: &pub}, KeyID: id} +} + +func TestRemotePreKeyUploadWrapAndConsumedMembers(t *testing.T) { + for _, consumeBeforePrepare := range []bool{false, true} { + t.Run(fmt.Sprint(consumeBeforePrepare), func(t *testing.T) { + remote := &fakeRemoteSignal{batch: &RemotePreKeyUploadBatch{ + BatchID: strings.Repeat("a", 32), + PreKeys: []*keys.PreKey{publicTestPreKey(0xFFFFFE), publicTestPreKey(0xFFFFFF), publicTestPreKey(2)}, + }} + pk := &remotePreKeyStore{remote: remote} + if consumeBeforePrepare { + remote.batch.PreKeys = remote.batch.PreKeys[2:] + } + batch, err := pk.GetOrGenPreKeys(context.Background(), 50) + if err != nil { + t.Fatal(err) + } + want := []uint32{0xFFFFFE, 0xFFFFFF, 2} + if consumeBeforePrepare { + want = []uint32{2} + } else { + // These original members are consumed after preparation, but ACK + // must retain the complete immutable upload snapshot. + remote.batch.PreKeys = remote.batch.PreKeys[2:] + } + if !equalIDs(preKeyIDs(batch), want) || remote.nextID != 0 { + t.Fatalf("keys=%v allocated=%d", preKeyIDs(batch), remote.nextID) + } + if err := pk.MarkPreKeysAsUploaded(context.Background(), 0xFFFFFF); err == nil || len(remote.acks) != 0 { + t.Fatal("nonlast numeric maximum acknowledged a wrapped batch") + } + if err := pk.MarkPreKeysAsUploaded(context.Background(), 2); err != nil { + t.Fatal(err) + } + if len(remote.acks) != 1 || !equalIDs(remote.acks[0].ids, want) || remote.batch != nil || len(remote.marked) != 0 { + t.Fatalf("acks=%v legacy marks=%v", remote.acks, remote.marked) + } + if err := pk.MarkPreKeysAsUploaded(context.Background(), 2); err == nil || len(remote.acks) != 1 { + t.Fatal("completed snapshot remained usable") + } + }) + } +} + +type boundaryPreKeyRemote struct { + *fakeRemoteSignal + prepared *RemotePreKeyUploadBatch + prepareErr error + fresh []*keys.PreKey + ack func(string, []uint32) error +} + +func (r *boundaryPreKeyRemote) PreparePreKeyUpload(context.Context, uint32) (*RemotePreKeyUploadBatch, error) { + return r.prepared, r.prepareErr +} + +func (r *boundaryPreKeyRemote) GeneratePreKeys(context.Context, uint32) ([]*keys.PreKey, error) { + return r.fresh, nil +} + +func (r *boundaryPreKeyRemote) AcknowledgePreKeyUpload(_ context.Context, token string, ids []uint32) error { + return r.ack(token, ids) +} + +func TestRemotePreKeyUploadDefensiveSnapshotsAndACKRetry(t *testing.T) { + input := []*keys.PreKey{publicTestPreKey(10), publicTestPreKey(11)} + var calls []testPreKeyACK + remote := &boundaryPreKeyRemote{fakeRemoteSignal: &fakeRemoteSignal{}, prepared: &RemotePreKeyUploadBatch{BatchID: strings.Repeat("b", 32), PreKeys: input}} + remote.ack = func(token string, ids []uint32) error { + calls = append(calls, testPreKeyACK{token, append([]uint32(nil), ids...)}) + ids[0] = 999 // A callback must not corrupt the next retry's authority. + if len(calls) == 1 { + return errors.New("synthetic response loss") + } + return nil + } + pk := &remotePreKeyStore{remote: remote} + batch, err := pk.GetOrGenPreKeys(context.Background(), 1) + if err != nil { + t.Fatal(err) + } + input[0].Pub[0] = 42 + input[0].KeyID = 30 + if batch[0].Pub[0] != 1 || batch[0].KeyID != 10 { + t.Fatal("remote response pointers escaped without a copy") + } + batch[0].KeyID, batch[1].KeyID = 100, 101 + batch[0].Pub[0] = 99 + if input[0].Pub[0] != 42 { + t.Fatal("caller mutated remote public key storage") + } + if err := pk.MarkPreKeysAsUploaded(context.Background(), 101); err == nil || len(calls) != 0 { + t.Fatal("caller-mutated last ID became ACK authority") + } + if err := pk.MarkPreKeysAsUploaded(context.Background(), 11); err == nil { + t.Fatal("synthetic first ACK failure was hidden") + } + if err := pk.MarkPreKeysAsUploaded(context.Background(), 11); err != nil { + t.Fatal(err) + } + for _, call := range calls { + if call.batchID != strings.Repeat("b", 32) || !equalIDs(call.ids, []uint32{10, 11}) { + t.Fatalf("ACK authority changed: %v", calls) + } + } +} + +func TestRemotePreKeyUploadRejectsMalformedResponsesWithoutFreshFallback(t *testing.T) { + valid := func() *RemotePreKeyUploadBatch { + return &RemotePreKeyUploadBatch{BatchID: strings.Repeat("a", 32), PreKeys: []*keys.PreKey{publicTestPreKey(1)}} + } + cases := map[string]func() *RemotePreKeyUploadBatch{ + "nil": func() *RemotePreKeyUploadBatch { return nil }, + "empty token": func() *RemotePreKeyUploadBatch { b := valid(); b.BatchID = ""; return b }, + "uppercase token": func() *RemotePreKeyUploadBatch { b := valid(); b.BatchID = strings.Repeat("A", 32); return b }, + "foreign token": func() *RemotePreKeyUploadBatch { b := valid(); b.BatchID = strings.Repeat("sensitive", 4); return b }, + "empty": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys = nil; return b }, + "too many": func() *RemotePreKeyUploadBatch { + b := valid() + for i := uint32(2); i <= 813; i++ { + b.PreKeys = append(b.PreKeys, publicTestPreKey(i)) + } + return b + }, + "nil key": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys[0] = nil; return b }, + "missing public": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys[0].Pub = nil; return b }, + "zero public": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys[0].Pub = new([32]byte); return b }, + "private key": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys[0].Priv = new([32]byte); return b }, + "signed key": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys[0].Signature = new([64]byte); return b }, + "zero ID": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys[0].KeyID = 0; return b }, + "wide ID": func() *RemotePreKeyUploadBatch { b := valid(); b.PreKeys[0].KeyID = 0x1000000; return b }, + "duplicate": func() *RemotePreKeyUploadBatch { + b := valid() + b.PreKeys = append(b.PreKeys, publicTestPreKey(1)) + return b + }, + } + for name, makeBatch := range cases { + t.Run(name, func(t *testing.T) { + fake := &fakeRemoteSignal{} + remote := &boundaryPreKeyRemote{fakeRemoteSignal: fake, prepared: makeBatch(), ack: func(string, []uint32) error { t.Fatal("invalid batch acknowledged"); return nil }} + pk := &remotePreKeyStore{remote: remote} + if _, err := pk.GetOrGenPreKeys(context.Background(), 50); err == nil || strings.Contains(err.Error(), "sensitive") { + t.Fatalf("invalid batch error=%v", err) + } + if err := pk.MarkPreKeysAsUploaded(context.Background(), 1); err == nil { + t.Fatal("invalid response became usable") + } + if len(fake.marked) != 0 || fake.nextID != 0 { + t.Fatal("invalid prepare fell back to fresh generation/marking") + } + }) + } + for _, count := range []uint32{0, 813, ^uint32(0)} { + fake := &fakeRemoteSignal{} + if _, err := (&remotePreKeyStore{remote: fake}).GetOrGenPreKeys(context.Background(), count); err == nil || len(fake.prepares) != 0 { + t.Fatalf("invalid count %d reached remote", count) + } + } + remoteErr := errors.New("unsupported prepare operation") + remote := &boundaryPreKeyRemote{fakeRemoteSignal: &fakeRemoteSignal{}, prepareErr: remoteErr} + if _, err := (&remotePreKeyStore{remote: remote}).GetOrGenPreKeys(context.Background(), 50); !errors.Is(err, remoteErr) { + t.Fatal("unsupported prepare did not fail closed") + } +} + +func TestRemoteFreshRetryPreKeyRejectsMalformedMaterial(t *testing.T) { + for _, generated := range [][]*keys.PreKey{nil, {nil}, {publicTestPreKey(0)}, {publicTestPreKey(1), publicTestPreKey(2)}} { + fake := &fakeRemoteSignal{} + remote := &boundaryPreKeyRemote{fakeRemoteSignal: fake, fresh: generated} + if _, err := (&remotePreKeyStore{remote: remote}).GenOnePreKey(context.Background()); err == nil || len(fake.marked) != 0 { + t.Fatal("malformed fresh key was returned or marked") + } + } +} + +func TestRemoteFreshRetryCannotBorrowFromPreparedUpload(t *testing.T) { + fake := &fakeRemoteSignal{} + remote := &boundaryPreKeyRemote{fakeRemoteSignal: fake, + prepared: &RemotePreKeyUploadBatch{BatchID: strings.Repeat("a", 32), PreKeys: []*keys.PreKey{publicTestPreKey(1)}}, + fresh: []*keys.PreKey{publicTestPreKey(1)}, + } + pk := &remotePreKeyStore{remote: remote} + if _, err := pk.GetOrGenPreKeys(context.Background(), 50); err != nil { + t.Fatal(err) + } + if _, err := pk.GenOnePreKey(context.Background()); err == nil || len(fake.marked) != 0 { + t.Fatal("pending upload key was accepted as a fresh retry key") + } +} + +func installUploadIQ(t *testing.T, cli *Client, remote *fakeRemoteSignal, failure string, want int) { + t.Helper() + cli.testSendHook = func(raw []byte) error { + unpacked, err := waBinary.Unpack(raw) + if err != nil { + return err + } + node, err := waBinary.Unmarshal(unpacked) + if err != nil { + return err + } + if node.Tag != "iq" || node.AttrGetter().OptionalString("xmlns") != "encrypt" || node.AttrGetter().OptionalString("type") != "set" { + return errors.New("unexpected upload IQ") + } + listNode := node.GetChildByTag("list") + list := listNode.GetChildren() + if len(list) != want { + t.Errorf("IQ keys=%d, want%d", len(list), want) + } + remote.mu.Lock() + remote.operations = append(remote.operations, "iq") + for i, item := range list { + id, ok := item.GetChildByTag("id").Content.([]byte) + if !ok || len(id) != 3 || i >= len(remote.batch.PreKeys) || binary.BigEndian.Uint32(append([]byte{0}, id...)) != remote.batch.PreKeys[i].KeyID { + t.Error("IQ did not contain the complete prepared ordered public batch") + } + } + if len(remote.acks) != 0 { + t.Error("batch was acknowledged before successful IQ") + } + remote.mu.Unlock() + if failure == "transport" { + return errors.New("synthetic IQ transport failure") + } + response := &waBinary.Node{Tag: "iq", Attrs: waBinary.Attrs{"id": node.AttrGetter().OptionalString("id"), "type": "result"}} + if failure == "server" { + response.Attrs["type"] = "error" + response.Content = []waBinary.Node{{Tag: "error", Attrs: waBinary.Attrs{"code": "500", "text": "synthetic"}}} + } + if !cli.receiveResponse(context.Background(), response) { + t.Error("upload IQ response had no registered waiter") + } + return nil + } +} + +func TestRemoteActualUploadIQFailureNeverAcknowledgesAndReopenRecovers812(t *testing.T) { + cli, remote, _ := newTestRemote(t) + installUploadIQ(t, cli, remote, "transport", 812) + cli.uploadPreKeys(context.Background(), true) + if remote.batch == nil || len(remote.batch.PreKeys) != 812 || len(remote.acks) != 0 || len(remote.marked) != 0 { + t.Fatal("failed IQ lost or acknowledged prepared keys") + } + // A different client/store instance simulates reopening against the same + // durable remote state. A normal50 request still uploads the full812. + second := NewRemoteSignalClient(&store.Device{}, testRemoteIdentity, remote, nil) + installUploadIQ(t, second, remote, "", 812) + second.uploadPreKeys(context.Background(), false) + if len(remote.prepares) != 2 || remote.prepares[0] != 812 || remote.prepares[1] != 50 || len(remote.acks) != 1 || len(remote.acks[0].ids) != 812 || remote.nextID != 812 || len(remote.marked) != 0 { + t.Fatalf("prepares=%v ACKs=%d allocated=%d", remote.prepares, len(remote.acks), remote.nextID) + } +} + +func TestRemoteActualUploadIQServerErrorDoesNotAcknowledgeOnPrepare(t *testing.T) { + cli, remote, _ := newTestRemote(t) + installUploadIQ(t, cli, remote, "server", 50) + cli.uploadPreKeys(context.Background(), false) + first := remote.batch.BatchID + batch, err := cli.Store.PreKeys.GetOrGenPreKeys(context.Background(), 812) + if err != nil || len(batch) != 50 || remote.batch.BatchID != first || len(remote.acks) != 0 || len(remote.marked) != 0 || remote.nextID != 50 { + t.Fatalf("failed IQ prepare len=%d err=%v ACKs=%v", len(batch), err, remote.acks) + } +} + +func TestRemoteLostACKIsReplayedBeforePreparingAnotherBatch(t *testing.T) { + cli, remote, _ := newTestRemote(t) + remote.ackLost = true + installUploadIQ(t, cli, remote, "", 50) + cli.uploadPreKeys(context.Background(), false) + if remote.batch != nil || len(remote.acks) != 1 || !cli.lastPreKeyUpload.IsZero() { + t.Fatal("lost ACK fixture did not apply the first batch before response loss") + } + pk := cli.Store.PreKeys.(*remotePreKeyStore) + if _, err := pk.GetOrGenPreKeys(context.Background(), 50); err != nil { + t.Fatal(err) + } + if len(remote.acks) != 2 || remote.acks[0].batchID != remote.acks[1].batchID || !equalIDs(remote.acks[0].ids, remote.acks[1].ids) || remote.batch == nil || remote.batch.BatchID == remote.acks[0].batchID || remote.nextID != 100 { + t.Fatalf("lost ACK was not resolved before new prepare: acks=%v", remote.acks) + } + if strings.Join(remote.operations, ",") != "prepare,iq,ack,ack,prepare" { + t.Fatalf("lost ACK replay ordering=%v", remote.operations) + } + // A persistent ACK failure must not permit even a prepare allocation. + batch, err := pk.GetOrGenPreKeys(context.Background(), 50) + if err != nil { + t.Fatal(err) + } + remote.ackErr = errors.New("synthetic unavailable ACK") + if err = pk.MarkPreKeysAsUploaded(context.Background(), batch[len(batch)-1].KeyID); err == nil { + t.Fatal("ACK failure hidden") + } + prepares := len(remote.prepares) + if _, err = pk.GetOrGenPreKeys(context.Background(), 50); err == nil || len(remote.prepares) != prepares || remote.nextID != 100 { + t.Fatal("prepare replaced an unresolved attempted ACK") + } +} + +type serializedPreKeyRemote struct { + *fakeRemoteSignal + active atomic.Int32 + max atomic.Int32 +} + +func (r *serializedPreKeyRemote) enter() func() { + n := r.active.Add(1) + for previous := r.max.Load(); n > previous; previous = r.max.Load() { + if r.max.CompareAndSwap(previous, n) { + break + } + } + time.Sleep(time.Millisecond) + return func() { r.active.Add(-1) } +} + +func (r *serializedPreKeyRemote) PreparePreKeyUpload(ctx context.Context, count uint32) (*RemotePreKeyUploadBatch, error) { + defer r.enter()() + return r.fakeRemoteSignal.PreparePreKeyUpload(ctx, count) +} +func (r *serializedPreKeyRemote) AcknowledgePreKeyUpload(ctx context.Context, token string, ids []uint32) error { + defer r.enter()() + return r.fakeRemoteSignal.AcknowledgePreKeyUpload(ctx, token, ids) +} +func (r *serializedPreKeyRemote) GeneratePreKeys(ctx context.Context, count uint32) ([]*keys.PreKey, error) { + defer r.enter()() + return r.fakeRemoteSignal.GeneratePreKeys(ctx, count) +} +func (r *serializedPreKeyRemote) MarkPreKeysUploaded(ctx context.Context, ids []uint32) error { + defer r.enter()() + return r.fakeRemoteSignal.MarkPreKeysUploaded(ctx, ids) +} + +func TestRemotePreKeyUploadSerializesPrepareMarkAndFreshGeneration(t *testing.T) { + remote := &serializedPreKeyRemote{fakeRemoteSignal: &fakeRemoteSignal{}} + pk := &remotePreKeyStore{remote: remote} + var wg sync.WaitGroup + for i := 0; i < 12; i++ { + wg.Go(func() { + for j := 0; j < 4; j++ { + batch, err := pk.GetOrGenPreKeys(context.Background(), 3) + if err != nil { + t.Error(err) + return + } + // Other complete callers may have retired this snapshot, so a + // literal-last mismatch is expected and must perform zero ACK. + _ = pk.MarkPreKeysAsUploaded(context.Background(), batch[len(batch)-1].KeyID) + if _, err := pk.GenOnePreKey(context.Background()); err != nil { + t.Error(err) + } + } + }) + } + wg.Wait() + if remote.max.Load() != 1 || len(remote.marked) != 48 || len(remote.acks) == 0 { + t.Fatalf("parallel calls=%d fresh=%d ACK=%d", remote.max.Load(), len(remote.marked), len(remote.acks)) + } +} diff --git a/remote_signal.go b/remote_signal.go new file mode 100644 index 000000000..26c2b8041 --- /dev/null +++ b/remote_signal.go @@ -0,0 +1,672 @@ +// Copyright (c) 2026 Rajeh Taher +// Licensed under the MIT License. See LICENSE-MIT for details. + +package whatsmeow + +import ( + "bytes" + "context" + "errors" + "fmt" + "sync" + "time" + + "google.golang.org/protobuf/proto" + + "github.com/polymorfa/libsignal-protocol-go/ecc" + "github.com/polymorfa/libsignal-protocol-go/keys/prekey" + + waBinary "github.com/polymorfa/hypermeow/binary" + "github.com/polymorfa/hypermeow/proto/waAdv" + "github.com/polymorfa/hypermeow/proto/waE2E" + "github.com/polymorfa/hypermeow/store" + "github.com/polymorfa/hypermeow/types" + "github.com/polymorfa/hypermeow/types/events" + "github.com/polymorfa/hypermeow/util/keys" + waLog "github.com/polymorfa/hypermeow/util/log" +) + +// RemoteSignal holds a linked device's Signal private keys, sessions and +// message plaintext outside this client. +// +// A client created with [NewRemoteSignalClient] keeps its socket, receipts, +// retries, prekey uploads and pairing flow, but never decrypts or encrypts a +// message itself and never sees plaintext. Every operation that needs a +// private key goes through this interface. Outgoing messages arrive as +// finished stanzas built by the remote store; see [Client.SendPrebuiltStanza]. +type RemoteSignal interface { + // Receive hands one `` payload to the remote store, which decrypts, + // processes and delivers it to the application. Only delivery metadata + // comes back. A decryption failure is reported as *RemoteDecryptError. + Receive(ctx context.Context, req RemoteReceive) (*RemoteReceiveResult, error) + // GeneratePreKeys creates one-time prekeys. Only public halves and IDs + // are returned; private halves stay remote. + GeneratePreKeys(ctx context.Context, count uint32) ([]*keys.PreKey, error) + // PreparePreKeyUpload durably prepares a public upload batch. count is + // only the allocation size when no live pending batch exists: an existing + // batch is returned in full, in its original order, without topping up. + PreparePreKeyUpload(ctx context.Context, count uint32) (*RemotePreKeyUploadBatch, error) + // AcknowledgePreKeyUpload atomically acknowledges all surviving keys in + // the prepared batch after the upload IQ succeeds. Exact ACKs are replayable + // after a lost response; consumed original members may remain in ids. + AcknowledgePreKeyUpload(ctx context.Context, batchID string, ids []uint32) error + // MarkPreKeysUploaded marks fresh retry-receipt keys only. Prepared upload + // batches require AcknowledgePreKeyUpload and must never use this operation. + MarkPreKeysUploaded(ctx context.Context, ids []uint32) error + // UploadedPreKeyCount reports how many uploaded prekeys remain unused. + UploadedPreKeyCount(ctx context.Context) (int, error) + // SignPairing verifies the primary's pair-success container and returns + // the encoded ADVSignedDeviceIdentity with this device's signature. + SignPairing(ctx context.Context, deviceIdentityContainer []byte) ([]byte, error) + // SetOwnJIDs tells the remote store this device's JIDs after pairing. + SetOwnJIDs(ctx context.Context, pn, lid types.JID) error + // Resend answers a retry receipt with a re-encrypted stanza for the + // requester, built from the remote store's own record of the message. + Resend(ctx context.Context, req RemoteResend) ([]byte, error) + // ForgetIdentity drops the stored identity and sessions for every device + // of user after the server reports an identity change. + ForgetIdentity(ctx context.Context, user types.JID) error + // MigratePNToLID moves the sessions and identities of a phone-number + // user's devices, including hosted devices, to the user's LID. Calling it + // again after a successful migration is a no-op. + MigratePNToLID(ctx context.Context, pn, lid types.JID) error +} + +// RemotePreKeyUploadBatch contains only public one-time prekeys and an opaque +// durable batch token. Private halves and pending records stay in the remote +// store. A batch contains 1..812 unique IDs in 1..0xFFFFFF; BatchID is a +// 32-character lowercase hexadecimal token, never an ordering key. +type RemotePreKeyUploadBatch struct { + BatchID string + PreKeys []*keys.PreKey +} + +// RemoteIdentity is the public material a remote-keyed device registers with. +type RemoteIdentity struct { + RegistrationID uint32 + IdentityKey [32]byte + SignedPreKeyID uint32 + SignedPreKey [32]byte + SignedPreKeySignature [64]byte + // AdvSecret is the pairing HMAC key shared with the primary through the + // QR code. It authenticates pairing; it decrypts nothing. + AdvSecret [32]byte +} + +// RemoteReceive is one `` payload of a received message. +type RemoteReceive struct { + // Timestamp is the provider message time; a missing value remains zero. + Timestamp time.Time + // SenderAlt is the other provider address when Sender is normalized for + // Signal encryption (for example, PN when Sender is a LID). + SenderAlt types.JID + Chat types.JID + Sender types.JID + MessageID types.MessageID + Kind string // "pkmsg", "msg" or "skmsg" + Ciphertext []byte + PaddingVersion int + IsFromMe bool +} + +// RemoteReceiveResult is what the client learns about a received payload. +type RemoteReceiveResult struct { + // AlreadyDelivered means the application has had this payload before; + // the client only needs to acknowledge it again. + AlreadyDelivered bool + // Redelivered means the payload was served from the remote buffer. + Redelivered bool + // SenderKeyRejected means a sender-key distribution in the message could + // not be stored. + SenderKeyRejected bool + // AppStateKeyShare is set only for shares from this account's own + // devices; it is the one piece of content the client receives. + AppStateKeyShare *waE2E.AppStateSyncKeyShare +} + +// RemoteDecryptFailure classifies why the remote store could not decrypt. +type RemoteDecryptFailure int + +const ( + // RemoteDecryptOther is any failure that should trigger a retry receipt. + RemoteDecryptOther RemoteDecryptFailure = iota + // RemoteDecryptOldCounter is a message older than the session can decrypt. + RemoteDecryptOldCounter + // RemoteDecryptNoSenderKey is a group message with no sender key. + RemoteDecryptNoSenderKey + // RemoteDecryptUndecodable decrypted but did not decode as a message. + RemoteDecryptUndecodable +) + +// RemoteDecryptError is returned by [RemoteSignal.Receive] for payloads that +// could not be decrypted. +type RemoteDecryptError struct { + Failure RemoteDecryptFailure + Err error +} + +func (e *RemoteDecryptError) Error() string { + return fmt.Sprintf("remote decrypt failed (%d): %v", e.Failure, e.Err) +} + +func (e *RemoteDecryptError) Unwrap() error { return e.Err } + +// RemoteResend asks the remote store to re-encrypt a sent message. +type RemoteResend struct { + Chat types.JID + MessageID types.MessageID + Requester types.JID + EncryptionJID types.JID + // Recipient is the original chat user when the requester is one of this + // account's own devices on a direct message. + Recipient types.JID + IsGroup bool + // IsBroadcast marks a broadcast-list retry; group addressing is unset. + IsBroadcast bool + AddressingMode types.AddressingMode + RetryCount int + // Bundle carries keys from the retry receipt, used to start a fresh + // session. Nil when the receipt had none. + Bundle *prekey.Bundle +} + +// ErrRemoteSignalUnsupported is returned for flows a remote-keyed client +// cannot run yet. +var ErrRemoteSignalUnsupported = errors.New("operation is not supported with remote Signal keys") + +// ErrRemoteSignalRequired is returned for operations that only a remote-keyed +// client performs. +var ErrRemoteSignalRequired = errors.New("operation requires a client with remote Signal keys") + +// NewRemoteSignalClient constructs a client whose Signal private keys live in +// remote. The device store holds only public material: private key fields +// are zero and the device is marked [store.Device.RemoteKeys], so a local +// Signal path reached by mistake fails loudly instead of using zero keys. +// Persisting the device stores those zero fields; the caller sets the same +// identity again when it loads a remote-keyed device. +func NewRemoteSignalClient(deviceStore *store.Device, identity RemoteIdentity, remote RemoteSignal, log waLog.Logger) *Client { + if remote == nil || isTypedNil(remote) { + panic("whatsmeow: NewRemoteSignalClient requires a non-nil RemoteSignal") + } + if deviceStore == nil { + panic("whatsmeow: NewRemoteSignalClient requires a non-nil device store") + } + applyRemoteIdentity(deviceStore, identity, remote) + cli := NewClient(deviceStore, log) + cli.remoteSignal = remote + return cli +} + +func applyRemoteIdentity(device *store.Device, identity RemoteIdentity, remote RemoteSignal) { + identityPub, signedPreKeyPub, signature := identity.IdentityKey, identity.SignedPreKey, identity.SignedPreKeySignature + device.RegistrationID = identity.RegistrationID + device.IdentityKey = &keys.KeyPair{Pub: &identityPub, Priv: new([32]byte)} + device.SignedPreKey = &keys.PreKey{ + KeyPair: keys.KeyPair{Pub: &signedPreKeyPub, Priv: new([32]byte)}, + KeyID: identity.SignedPreKeyID, + Signature: &signature, + } + device.AdvSecretKey = append([]byte(nil), identity.AdvSecret[:]...) + device.PreKeys = &remotePreKeyStore{remote: remote} + device.RemoteKeys = true +} + +// isRemoteSignal reports whether this client's Signal keys are remote. +func (cli *Client) isRemoteSignal() bool { + return cli != nil && cli.remoteSignal != nil +} + +// remotePreKeyStore serves prekey uploads and retry receipts from the remote +// store. Only public halves pass through it. +type remotePreKeyStore struct { + remote RemoteSignal + + mu sync.Mutex + // The ACK authority is captured before returning mutable public key objects + // to callers. It never relies on their pointers or numeric ID ordering. + batchID string + ids []uint32 + // awaitingACK means the IQ succeeded and an ACK was attempted. Resolve that + // exact receipt before preparing another batch if its response was lost. + awaitingACK bool +} + +var _ store.PreKeyStore = (*remotePreKeyStore)(nil) + +// GetOrGenPreKeys returns the complete durable pending batch. A process restart +// or a changed requested count cannot truncate it or allocate a fresh top-up. +func (s *remotePreKeyStore) GetOrGenPreKeys(ctx context.Context, count uint32) ([]*keys.PreKey, error) { + if count == 0 || count > 812 { + return nil, errors.New("remote prekey upload count must be between 1 and 812") + } + s.mu.Lock() + defer s.mu.Unlock() + if s.awaitingACK { + if err := s.acknowledgeLocked(ctx); err != nil { + return nil, err + } + } + batch, err := s.remote.PreparePreKeyUpload(ctx, count) + if err != nil { + return nil, err + } + if batch == nil || !validRemotePreKeyBatchID(batch.BatchID) || len(batch.PreKeys) == 0 || len(batch.PreKeys) > 812 { + return nil, errors.New("remote store returned an invalid prekey upload batch") + } + public, ids, err := copyRemotePreKeys(batch.PreKeys) + if err != nil { + return nil, err + } + s.batchID, s.ids = batch.BatchID, ids + return public, nil +} + +func validRemotePreKeyBatchID(id string) bool { + if len(id) != 32 { + return false + } + for _, c := range []byte(id) { + if !(c >= '0' && c <= '9' || c >= 'a' && c <= 'f') { + return false + } + } + return true +} + +// copyRemotePreKeys validates the public-only boundary and returns independent +// public key objects plus an ID snapshot that neither caller nor remote can edit. +func copyRemotePreKeys(input []*keys.PreKey) ([]*keys.PreKey, []uint32, error) { + public := make([]*keys.PreKey, 0, len(input)) + ids := make([]uint32, 0, len(input)) + seen := make(map[uint32]bool, len(input)) + for _, key := range input { + if key == nil || key.Pub == nil || *key.Pub == [32]byte{} || key.Priv != nil || key.Signature != nil || key.KeyID == 0 || key.KeyID > 0xFFFFFF || seen[key.KeyID] { + return nil, nil, errors.New("remote store returned an invalid public prekey") + } + pub := *key.Pub + public = append(public, &keys.PreKey{KeyPair: keys.KeyPair{Pub: &pub}, KeyID: key.KeyID}) + ids = append(ids, key.KeyID) + seen[key.KeyID] = true + } + return public, ids, nil +} + +// GenOnePreKey creates a key for a retry receipt. The receipt publishes it, +// so it is marked uploaded at once, as the SQL store does. +func (s *remotePreKeyStore) GenOnePreKey(ctx context.Context) (*keys.PreKey, error) { + s.mu.Lock() + defer s.mu.Unlock() + generated, err := s.remote.GeneratePreKeys(ctx, 1) + if err != nil { + return nil, err + } + if len(generated) != 1 { + return nil, fmt.Errorf("remote store returned %d prekeys, want 1", len(generated)) + } + public, ids, err := copyRemotePreKeys(generated) + if err != nil { + return nil, err + } + for _, pendingID := range s.ids { + if ids[0] == pendingID { + return nil, errors.New("remote store returned a prepared upload key for a fresh retry") + } + } + if err = s.remote.MarkPreKeysUploaded(ctx, ids); err != nil { + return nil, fmt.Errorf("failed to mark retry prekey as published: %w", err) + } + return public[0], nil +} + +func (s *remotePreKeyStore) GetPreKey(ctx context.Context, id uint32) (*keys.PreKey, error) { + return nil, fmt.Errorf("GetPreKey(%d): %w", id, ErrRemoteSignalUnsupported) +} + +func (s *remotePreKeyStore) RemovePreKey(ctx context.Context, id uint32) error { + return fmt.Errorf("RemovePreKey(%d): %w", id, ErrRemoteSignalUnsupported) +} + +// MarkPreKeysAsUploaded requires the literal last returned ID and acknowledges +// every ID captured for that upload, including IDs across 24-bit wrap. +func (s *remotePreKeyStore) MarkPreKeysAsUploaded(ctx context.Context, upToID uint32) error { + s.mu.Lock() + defer s.mu.Unlock() + if len(s.ids) == 0 || upToID != s.ids[len(s.ids)-1] { + return errors.New("prekey upload acknowledgement does not match the prepared batch") + } + s.awaitingACK = true + return s.acknowledgeLocked(ctx) +} + +func (s *remotePreKeyStore) acknowledgeLocked(ctx context.Context) error { + // Implementations may retain or mutate arguments. Preserve our retry authority. + if err := s.remote.AcknowledgePreKeyUpload(ctx, s.batchID, append([]uint32(nil), s.ids...)); err != nil { + return err + } + s.batchID, s.ids, s.awaitingACK = "", nil, false + return nil +} + +func (s *remotePreKeyStore) UploadedPreKeyCount(ctx context.Context) (int, error) { + return s.remote.UploadedPreKeyCount(ctx) +} + +// decryptMessagesRemote is decryptMessages for a remote-keyed client: each +// `` payload goes to the remote store, and only delivery metadata comes +// back. Receipts, retry receipts and acks follow the local path. +func (cli *Client) decryptMessagesRemote(ctx context.Context, info *types.MessageInfo, node *waBinary.Node, senderEncryptionJID types.JID) { + var recognizedStanza, undecodable, containsDirectMsg bool + for _, child := range node.GetChildren() { + if child.Tag != "enc" { + continue + } + recognizedStanza = true + ag := child.AttrGetter() + encType, ok := ag.GetString("type", false) + if !ok { + continue + } + switch { + case encType == "pkmsg" || encType == "msg": + containsDirectMsg = true + case encType == "skmsg" && info.IsGroup: + case encType == "msmsg": + // Bot message secrets are kept by the remote store, which this + // path does not reach yet. + cli.backgroundIfAsyncAck(func() { + cli.sendAck(ctx, node, NackMissingMessageSecret) + }) + return + default: + cli.Log.Warnf("Unhandled encrypted message (type %s) from %s", encType, info.SourceString()) + continue + } + ciphertext, ok := child.Content.([]byte) + if !ok { + cli.Log.Warnf("Encrypted payload from %s is not a byte slice", info.SourceString()) + continue + } + senderAlt := info.SenderAlt + // The encryption address can be the alternate LID of a PN-addressed + // stanza. Keep the PN as its counterpart instead of forwarding the + // same LID as both sender fields. + if senderEncryptionJID == info.SenderAlt && senderEncryptionJID != info.Sender { + senderAlt = info.Sender + } + result, err := cli.remoteSignal.Receive(ctx, RemoteReceive{ + Timestamp: info.Timestamp, + SenderAlt: senderAlt, + Chat: info.Chat, + Sender: senderEncryptionJID, + MessageID: info.ID, + Kind: encType, + Ciphertext: ciphertext, + PaddingVersion: ag.OptionalInt("v"), + IsFromMe: info.IsFromMe, + }) + var decryptErr *RemoteDecryptError + isDecryptErr := errors.As(err, &decryptErr) + switch { + case err == nil: + case isDecryptErr && decryptErr.Failure == RemoteDecryptOldCounter: + cli.Log.Warnf("Ignoring message %s from %s: %v", info.ID, info.SourceString(), err) + continue + case isDecryptErr && decryptErr.Failure == RemoteDecryptUndecodable: + cli.Log.Warnf("Remote store could not decode message %s from %s: %v", info.ID, info.SourceString(), err) + undecodable = true + continue + case !isDecryptErr: + // The remote store may never have seen the payload (outage, + // timeout). Neither ack nor send a retry receipt: the server + // redelivers the stanza, and acknowledging it could lose the + // message for good. + cli.Log.Errorf("Remote Signal store unavailable for message %s from %s: %v", info.ID, info.SourceString(), err) + return + default: + cli.Log.Warnf("Error decrypting message %s from %s remotely: %v", info.ID, info.SourceString(), err) + if ctx.Err() != nil || errors.Is(err, context.Canceled) { + return + } + isUnavailable := encType == "skmsg" && !containsDirectMsg && + isDecryptErr && decryptErr.Failure == RemoteDecryptNoSenderKey + if cli.SynchronousAck { + cli.sendRetryReceipt(ctx, node, info, isUnavailable) + cli.sendAck(ctx, node, 0) + } else { + go cli.sendRetryReceipt(context.WithoutCancel(ctx), node, info, isUnavailable) + go cli.sendAck(ctx, node, 0) + } + cli.dispatchEvent(&events.UndecryptableMessage{ + Info: *info, + IsUnavailable: isUnavailable, + DecryptFailMode: events.DecryptFailMode(ag.OptionalString("decrypt-fail")), + }) + return + } + undecodable = false + cli.cancelDelayedRequestFromPhone(info.ID) + if result.AppStateKeyShare != nil { + cli.handleAppStateSyncKeyShare(ctx, result.AppStateKeyShare) + } + if !result.AlreadyDelivered { + cli.dispatchEvent(&events.RemoteMessage{ + Info: *info, + Redelivered: result.Redelivered, + SenderKeyRejected: result.SenderKeyRejected, + }) + } + } + cli.backgroundIfAsyncAck(func() { + if !recognizedStanza { + cli.sendAck(ctx, node, NackUnrecognizedStanza) + } else if undecodable { + cli.sendAck(ctx, node, NackInvalidProtobuf) + } else { + cli.sendMessageReceipt(ctx, info, node) + } + }) +} + +// remoteDeviceSignature asks the remote store to sign the pairing identity. +// The store verifies the container itself and signs only the ADV message it +// builds; the client still checks that the result covers the details it +// verified locally. +func (cli *Client) remoteDeviceSignature(ctx context.Context, container []byte, verified *waAdv.ADVSignedDeviceIdentity) ([]byte, error) { + signed, err := cli.remoteSignal.SignPairing(ctx, container) + if err != nil { + return nil, fmt.Errorf("remote Signal store refused pairing: %w", err) + } + var identity waAdv.ADVSignedDeviceIdentity + if err = proto.Unmarshal(signed, &identity); err != nil { + return nil, &PairProtoError{"failed to parse remotely signed device identity", err} + } + if !bytes.Equal(identity.GetDetails(), verified.GetDetails()) { + return nil, fmt.Errorf("remote Signal store signed different device identity details") + } + if len(identity.GetDeviceSignature()) != 64 { + return nil, fmt.Errorf("remote Signal store returned a %d-byte device signature", len(identity.GetDeviceSignature())) + } + // Check it is a valid signature by the advertised identity key over the + // exact message generateDeviceSignature signs. + message := concatBytes(AdvDeviceSignaturePrefix, verified.Details, cli.Store.IdentityKey.Pub[:], verified.AccountSignatureKey) + signature := *(*[64]byte)(identity.GetDeviceSignature()) + if !ecc.VerifySignature(ecc.NewDjbECPublicKey(*cli.Store.IdentityKey.Pub), message, signature) { + return nil, fmt.Errorf("remote Signal store returned a device signature that does not verify") + } + return identity.GetDeviceSignature(), nil +} + +// handleRetryReceiptRemote answers a retry receipt with a stanza the remote +// store re-encrypts from its own record of the message. +func (cli *Client) handleRetryReceiptRemote(ctx context.Context, receipt *events.Receipt, node *waBinary.Node, messageID types.MessageID, retryCount int) error { + if !cli.acceptIncomingRetry(receipt.Sender, messageID) { + return nil + } + // The remote store holds the message, so the callback gets no content. + if cli.PreRetryCallback != nil && !cli.PreRetryCallback(receipt, messageID, retryCount, nil) { + cli.Log.Debugf("Cancelled retry receipt in PreRetryCallback") + return nil + } + req := RemoteResend{ + Chat: receipt.Chat, + MessageID: messageID, + Requester: receipt.Sender, + EncryptionJID: receipt.Sender, + RetryCount: retryCount, + } + if _, hasKeys := node.GetOptionalChildByTag("keys"); hasKeys { + bundle, err := nodeToPreKeyBundle(uint32(receipt.Sender.Device), *node) + if err != nil { + return fmt.Errorf("failed to read prekey bundle in retry receipt: %w", err) + } + req.Bundle = bundle + } + if receipt.Sender.Server == types.DefaultUserServer { + if lid, err := cli.Store.LIDs.GetLIDForPN(ctx, receipt.Sender); err != nil { + cli.Log.Warnf("Failed to get LID for %s: %v", receipt.Sender, err) + } else if !lid.IsEmpty() { + req.EncryptionJID = lid + } + } + switch { + case receipt.Chat.Server == types.BroadcastServer: + req.IsBroadcast = true + case receipt.IsGroup: + req.IsGroup = true + if group, err := cli.getCachedGroupData(ctx, receipt.Chat); err == nil && group != nil { + req.AddressingMode = group.AddressingMode + } + case receipt.IsFromMe: + req.Recipient = receipt.Chat + } + stanza, err := cli.remoteSignal.Resend(ctx, req) + if err != nil { + return fmt.Errorf("remote Signal store could not resend %s: %w", messageID, err) + } + return cli.sendRemoteStanza(ctx, stanza) +} + +func decodeRemoteStanza(stanza []byte) (*waBinary.Node, error) { + unpacked, err := waBinary.Unpack(stanza) + if err != nil { + return nil, fmt.Errorf("failed to unpack remote stanza: %w", err) + } + node, err := waBinary.Unmarshal(unpacked) + if err != nil { + return nil, fmt.Errorf("failed to decode remote stanza: %w", err) + } + if node.Tag != "message" { + return nil, fmt.Errorf("remote stanza is <%s>, want ", node.Tag) + } + return node, nil +} + +func (cli *Client) sendRemoteStanza(ctx context.Context, stanza []byte) error { + node, err := decodeRemoteStanza(stanza) + if err != nil { + return err + } + return cli.sendNode(ctx, *node) +} + +// SendPrebuiltStanza transmits a `` stanza the remote Signal store +// built and waits for the server's ack. phash is the participant hash the +// store computed, compared with the ack as SendMessage does. A zero timeout +// waits until ctx ends. +func (cli *Client) SendPrebuiltStanza(ctx context.Context, stanza []byte, phash string, timeout time.Duration) (SendResponse, error) { + var resp SendResponse + if !cli.isRemoteSignal() { + return resp, fmt.Errorf("SendPrebuiltStanza: %w", ErrRemoteSignalRequired) + } + node, err := decodeRemoteStanza(stanza) + if err != nil { + return resp, err + } + id, _ := node.Attrs["id"].(string) + if id == "" { + return resp, fmt.Errorf("remote stanza has no id") + } + resp.ID = types.MessageID(id) + respChan := cli.waitResponse(id) + data, err := cli.sendNodeAndGetData(ctx, *node) + if err != nil { + cli.cancelResponse(id, respChan) + return resp, err + } + var timeoutChan <-chan time.Time + if timeout > 0 { + timeoutChan = time.After(timeout) + } + var respNode *waBinary.Node + select { + case respNode = <-respChan: + case <-timeoutChan: + cli.cancelResponse(id, respChan) + return resp, ErrMessageTimedOut + case <-ctx.Done(): + cli.cancelResponse(id, respChan) + return resp, ctx.Err() + } + if isDisconnectNode(respNode) { + if respNode, err = cli.retryFrame(ctx, "prebuilt message send", id, data, respNode, 0); err != nil { + return resp, err + } + } + ag := respNode.AttrGetter() + resp.ServerID = types.MessageServerID(ag.OptionalInt("server_id")) + resp.Timestamp = ag.UnixTime("t") + if errorCode := ag.Int("error"); errorCode != 0 { + return resp, fmt.Errorf("%w %d", ErrServerReturnedError, errorCode) + } + setParticipantHashMismatch(&resp, phash, ag.OptionalString("phash")) + return resp, nil +} + +// forgetRemoteIdentity asks the remote store to drop identity and session +// state for from and its PN/LID alternate, mirroring deleteIdentityChangeState. +// The store clears a phone-number or LID user's hosted devices with it, so a +// hosted JID is sent as its base user. +func (cli *Client) forgetRemoteIdentity(ctx context.Context, from types.JID) { + base := from.ToNonAD() + switch base.Server { + case types.HostedServer: + base.Server = types.DefaultUserServer + case types.HostedLIDServer: + base.Server = types.HiddenUserServer + } + users := []types.JID{base} + if cli.Store.LIDs != nil { + var alternate types.JID + var err error + switch from.Server { + case types.DefaultUserServer, types.HostedServer: + alternate, err = cli.Store.LIDs.GetLIDForPN(ctx, types.NewJID(from.User, types.DefaultUserServer)) + case types.HiddenUserServer, types.HostedLIDServer: + alternate, err = cli.Store.LIDs.GetPNForLID(ctx, types.NewJID(from.User, types.HiddenUserServer)) + } + if err != nil { + cli.Log.Warnf("Failed to resolve alternate JID for identity change from %s: %v", from, err) + } else if !alternate.IsEmpty() { + users = append(users, alternate.ToNonAD()) + } + } + for _, user := range users { + if err := cli.remoteSignal.ForgetIdentity(ctx, user); err != nil { + cli.Log.Warnf("Remote Signal store failed to forget identity of %s: %v", user, err) + } + } +} + +// migrateRemoteSessions asks the remote store to move pn's Signal state to +// lid, once per user per client, as SQLStore.MigratePNToLID does locally. +func (cli *Client) migrateRemoteSessions(ctx context.Context, pn, lid types.JID) { + pn, lid = pn.ToNonAD(), lid.ToNonAD() + if _, done := cli.remoteMigratedPN.Load(pn.User); done { + return + } + if err := cli.remoteSignal.MigratePNToLID(ctx, pn, lid); err != nil { + cli.Log.Errorf("Remote Signal store failed to migrate %s to %s: %v", pn, lid, err) + return + } + cli.remoteMigratedPN.Store(pn.User, struct{}{}) +} diff --git a/remote_signal_test.go b/remote_signal_test.go new file mode 100644 index 000000000..1a325b852 --- /dev/null +++ b/remote_signal_test.go @@ -0,0 +1,665 @@ +// Copyright (c) 2026 Rajeh Taher +// Licensed under the MIT License. See LICENSE-MIT for details. + +package whatsmeow + +import ( + "context" + "errors" + "fmt" + "slices" + "strings" + "sync" + "testing" + "time" + + "google.golang.org/protobuf/proto" + + waBinary "github.com/polymorfa/hypermeow/binary" + "github.com/polymorfa/hypermeow/proto/waAdv" + "github.com/polymorfa/hypermeow/proto/waE2E" + "github.com/polymorfa/hypermeow/store" + "github.com/polymorfa/hypermeow/types" + "github.com/polymorfa/hypermeow/types/events" + "github.com/polymorfa/hypermeow/util/keys" +) + +type fakeRemoteSignal struct { + mu sync.Mutex + + receives []RemoteReceive + receiveErr error + result RemoteReceiveResult + nextID uint32 + marked [][]uint32 + signed []byte + ownJIDs []types.JID + resends []RemoteResend + resendWith []byte + forgotten []types.JID + migrated [][2]types.JID + migrateErr error + batchSeq uint32 + batch *RemotePreKeyUploadBatch + prepares []uint32 + acks []testPreKeyACK + ackErr error + ackLost bool + lastACK testPreKeyACK + operations []string +} + +type testPreKeyACK struct { + batchID string + ids []uint32 +} + +func (f *fakeRemoteSignal) PreparePreKeyUpload(ctx context.Context, count uint32) (*RemotePreKeyUploadBatch, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.prepares = append(f.prepares, count) + f.operations = append(f.operations, "prepare") + if f.batch == nil { + f.batchSeq++ + f.batch = &RemotePreKeyUploadBatch{BatchID: fmt.Sprintf("%032x", f.batchSeq), PreKeys: f.generateLocked(count)} + } + return f.batch, nil +} + +func (f *fakeRemoteSignal) AcknowledgePreKeyUpload(ctx context.Context, batchID string, ids []uint32) error { + f.mu.Lock() + defer f.mu.Unlock() + ack := testPreKeyACK{batchID, append([]uint32(nil), ids...)} + f.acks = append(f.acks, ack) + f.operations = append(f.operations, "ack") + if f.ackErr != nil { + return f.ackErr + } + if batchID == f.lastACK.batchID && equalIDs(ids, f.lastACK.ids) { + return nil + } + if f.batch == nil || f.batch.BatchID != batchID { + return errors.New("unexpected batch ACK") + } + for _, key := range f.batch.PreKeys { + if !slices.Contains(ids, key.KeyID) { + return errors.New("batch ACK omitted a surviving key") + } + } + f.lastACK, f.batch = ack, nil + if f.ackLost { + f.ackLost = false + return errors.New("synthetic lost ACK response") + } + return nil +} + +func (f *fakeRemoteSignal) MigratePNToLID(ctx context.Context, pn, lid types.JID) error { + f.mu.Lock() + defer f.mu.Unlock() + f.migrated = append(f.migrated, [2]types.JID{pn, lid}) + return f.migrateErr +} + +func (f *fakeRemoteSignal) Receive(ctx context.Context, req RemoteReceive) (*RemoteReceiveResult, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.receives = append(f.receives, req) + if f.receiveErr != nil { + return nil, f.receiveErr + } + result := f.result + return &result, nil +} + +func (f *fakeRemoteSignal) GeneratePreKeys(ctx context.Context, count uint32) ([]*keys.PreKey, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.generateLocked(count), nil +} + +func (f *fakeRemoteSignal) generateLocked(count uint32) []*keys.PreKey { + out := make([]*keys.PreKey, 0, count) + for i := uint32(0); i < count; i++ { + f.nextID++ + pub := [32]byte{byte(f.nextID%255 + 1)} + out = append(out, &keys.PreKey{KeyPair: keys.KeyPair{Pub: &pub}, KeyID: f.nextID}) + } + return out +} + +func (f *fakeRemoteSignal) MarkPreKeysUploaded(ctx context.Context, ids []uint32) error { + f.mu.Lock() + defer f.mu.Unlock() + f.marked = append(f.marked, append([]uint32(nil), ids...)) + return nil +} + +func (f *fakeRemoteSignal) UploadedPreKeyCount(ctx context.Context) (int, error) { return 7, nil } + +func (f *fakeRemoteSignal) SignPairing(ctx context.Context, container []byte) ([]byte, error) { + return f.signed, nil +} + +func (f *fakeRemoteSignal) SetOwnJIDs(ctx context.Context, pn, lid types.JID) error { + f.mu.Lock() + defer f.mu.Unlock() + f.ownJIDs = append(f.ownJIDs, pn, lid) + return nil +} + +func (f *fakeRemoteSignal) ForgetIdentity(ctx context.Context, user types.JID) error { + f.mu.Lock() + defer f.mu.Unlock() + f.forgotten = append(f.forgotten, user) + return nil +} + +func (f *fakeRemoteSignal) Resend(ctx context.Context, req RemoteResend) ([]byte, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.resends = append(f.resends, req) + return f.resendWith, nil +} + +var _ RemoteSignal = (*fakeRemoteSignal)(nil) + +type sentNodes struct { + mu sync.Mutex + nodes []*waBinary.Node +} + +func (s *sentNodes) hook(payload []byte) error { + unpacked, err := waBinary.Unpack(payload) + if err != nil { + return err + } + node, err := waBinary.Unmarshal(unpacked) + if err != nil { + return err + } + s.mu.Lock() + s.nodes = append(s.nodes, node) + s.mu.Unlock() + return nil +} + +func (s *sentNodes) tags() []string { + s.mu.Lock() + defer s.mu.Unlock() + var out []string + for _, n := range s.nodes { + out = append(out, n.Tag+":"+n.AttrGetter().OptionalString("type")) + } + return out +} + +var testRemoteIdentity = RemoteIdentity{ + RegistrationID: 4242, + IdentityKey: [32]byte{1}, + SignedPreKeyID: 3, + SignedPreKey: [32]byte{2}, + SignedPreKeySignature: [64]byte{3}, + AdvSecret: [32]byte{4}, +} + +func newTestRemote(t *testing.T) (*Client, *fakeRemoteSignal, *sentNodes) { + t.Helper() + remote := &fakeRemoteSignal{} + cli := NewRemoteSignalClient(&store.Device{}, testRemoteIdentity, remote, nil) + sent := &sentNodes{} + cli.testSendHook = sent.hook + cli.SynchronousAck = true + return cli, remote, sent +} + +func TestRemoteSignalClientHoldsOnlyPublicKeys(t *testing.T) { + cli, _, _ := newTestRemote(t) + dev := cli.Store + if !dev.RemoteKeys { + t.Fatal("device is not marked remote-keyed") + } + if *dev.IdentityKey.Pub != testRemoteIdentity.IdentityKey || *dev.IdentityKey.Priv != [32]byte{} { + t.Fatal("identity key should be public-only") + } + if *dev.SignedPreKey.Priv != [32]byte{} || *dev.SignedPreKey.Signature != testRemoteIdentity.SignedPreKeySignature { + t.Fatal("signed prekey should carry the remote signature and no private key") + } + if dev.RegistrationID != 4242 || string(dev.AdvSecretKey) != string(testRemoteIdentity.AdvSecret[:]) { + t.Fatal("registration ID or ADV secret not applied") + } + defer func() { + if r := recover(); r == nil || !strings.Contains(r.(string), "remote keys") { + t.Fatalf("local identity read on a remote-keyed device did not panic: %v", r) + } + }() + dev.GetIdentityKeyPair() +} + +func TestNewRemoteSignalClientRejectsNilRemote(t *testing.T) { + defer func() { + if recover() == nil { + t.Fatal("expected a panic for a nil RemoteSignal") + } + }() + NewRemoteSignalClient(&store.Device{}, testRemoteIdentity, nil, nil) +} + +func encMessage(encType string) (*types.MessageInfo, *waBinary.Node) { + sender := types.NewJID("123", types.HiddenUserServer) + info := &types.MessageInfo{ + MessageSource: types.MessageSource{Chat: sender, Sender: sender}, + ID: "MSG1", + } + node := &waBinary.Node{ + Tag: "message", + Attrs: waBinary.Attrs{"id": "MSG1", "from": sender, "t": "1700000000"}, + Content: []waBinary.Node{{ + Tag: "enc", + Attrs: waBinary.Attrs{"type": encType, "v": "2"}, + Content: []byte("ciphertext"), + }}, + } + return info, node +} + +func TestRemoteReceiveDeliversMetadataAndAcknowledges(t *testing.T) { + cli, remote, sent := newTestRemote(t) + remote.result = RemoteReceiveResult{Redelivered: true} + var got []*events.RemoteMessage + cli.AddEventHandler(func(evt any) { + if e, ok := evt.(*events.RemoteMessage); ok { + got = append(got, e) + } + }) + info, node := encMessage("pkmsg") + cli.decryptMessages(context.Background(), info, node) + + if len(remote.receives) != 1 { + t.Fatalf("Receive calls = %d, want 1", len(remote.receives)) + } + req := remote.receives[0] + if req.Kind != "pkmsg" || string(req.Ciphertext) != "ciphertext" || req.PaddingVersion != 2 || req.Sender != info.Sender { + t.Fatalf("unexpected receive request: %+v", req) + } + if len(got) != 1 || !got[0].Redelivered || got[0].Info.ID != "MSG1" { + t.Fatalf("RemoteMessage events = %+v", got) + } + if tags := sent.tags(); len(tags) != 1 || !strings.HasPrefix(tags[0], "receipt:") { + t.Fatalf("sent %v, want one delivery receipt", tags) + } +} + +func TestRemoteReceivePreservesProviderTimeAndAlternateSender(t *testing.T) { + pn := types.NewJID("15551234567", types.DefaultUserServer) + lid := types.NewJID("123456789", types.HiddenUserServer) + stamp := time.Unix(1720000000, 123456789).UTC() + for _, tc := range []struct { + name string + sender, alternate types.JID + wantSender, wantAlt types.JID + timestamp time.Time + }{ + {"lid addressed", lid, pn, lid, pn, stamp}, + {"pn normalized to lid", pn, lid, lid, pn, stamp}, + {"absent metadata", lid, types.JID{}, lid, types.JID{}, time.Time{}}, + } { + t.Run(tc.name, func(t *testing.T) { + cli, remote, sent := newTestRemote(t) + info, node := encMessage("pkmsg") + info.Sender, info.SenderAlt = tc.sender, tc.alternate + info.Chat = tc.sender.ToNonAD() + info.Timestamp = tc.timestamp + cli.decryptMessages(context.Background(), info, node) + if len(remote.receives) != 1 { + t.Fatalf("Receive calls = %d, want 1", len(remote.receives)) + } + req := remote.receives[0] + if !req.Timestamp.Equal(tc.timestamp) || req.Sender != tc.wantSender || req.SenderAlt != tc.wantAlt { + t.Fatalf("receive metadata = (%v, %s, %s), want (%v, %s, %s)", req.Timestamp, req.Sender, req.SenderAlt, tc.timestamp, tc.wantSender, tc.wantAlt) + } + if req.Kind != "pkmsg" || req.MessageID != info.ID || req.Chat != info.Chat || string(req.Ciphertext) != "ciphertext" { + t.Fatal("adding receive metadata changed the encrypted payload or routing") + } + if tags := sent.tags(); len(tags) != 1 || !strings.HasPrefix(tags[0], "receipt:") { + t.Fatalf("sent %v, want the unchanged delivery receipt", tags) + } + }) + } +} + +func TestRemoteAlreadyDeliveredOnlyAcknowledges(t *testing.T) { + cli, remote, sent := newTestRemote(t) + remote.result = RemoteReceiveResult{AlreadyDelivered: true} + events_ := 0 + cli.AddEventHandler(func(evt any) { + if _, ok := evt.(*events.RemoteMessage); ok { + events_++ + } + }) + info, node := encMessage("msg") + cli.decryptMessages(context.Background(), info, node) + if events_ != 0 { + t.Fatal("an already-delivered payload must not be announced again") + } + if tags := sent.tags(); len(tags) != 1 || !strings.HasPrefix(tags[0], "receipt:") { + t.Fatalf("sent %v, want the receipt again", tags) + } +} + +func TestRemoteDecryptFailureSendsRetryReceipt(t *testing.T) { + cli, remote, sent := newTestRemote(t) + remote.receiveErr = &RemoteDecryptError{Failure: RemoteDecryptOther, Err: errors.New("bad mac")} + var undecryptable int + cli.AddEventHandler(func(evt any) { + if _, ok := evt.(*events.UndecryptableMessage); ok { + undecryptable++ + } + }) + info, node := encMessage("msg") + cli.decryptMessages(context.Background(), info, node) + tags := sent.tags() + if len(tags) < 1 || tags[0] != "receipt:retry" { + t.Fatalf("sent %v, want a retry receipt first", tags) + } + if undecryptable != 1 { + t.Fatalf("UndecryptableMessage events = %d, want 1", undecryptable) + } +} + +func TestRemoteUndecodableMessageIsNacked(t *testing.T) { + cli, remote, sent := newTestRemote(t) + remote.receiveErr = &RemoteDecryptError{Failure: RemoteDecryptUndecodable, Err: errors.New("bad proto")} + info, node := encMessage("msg") + cli.decryptMessages(context.Background(), info, node) + tags := sent.tags() + if len(tags) != 1 || !strings.HasPrefix(tags[0], "ack:") { + t.Fatalf("sent %v, want a nack", tags) + } + if code := sent.nodes[0].AttrGetter().OptionalInt("error"); code != NackInvalidProtobuf { + t.Fatalf("nack code = %d, want %d", code, NackInvalidProtobuf) + } +} + +func TestRemotePreKeyStoreDoesNotTruncateAnOutstandingInitialUpload(t *testing.T) { + _, remote, _ := newTestRemote(t) + pk := &remotePreKeyStore{remote: remote} + first, err := pk.GetOrGenPreKeys(context.Background(), 812) + if err != nil || len(first) != 812 { + t.Fatalf("initial batch len=%d err=%v", len(first), err) + } + recovered, err := pk.GetOrGenPreKeys(context.Background(), 50) + if err != nil || len(recovered) != 812 { + t.Fatalf("outstanding batch len=%d err=%v, want all812", len(recovered), err) + } +} + +func TestRemotePreKeyStoreReusesFullBatchAndKeepsFreshRetryKeysSeparate(t *testing.T) { + _, remote, _ := newTestRemote(t) + pk := &remotePreKeyStore{remote: remote} + ctx := context.Background() + ids := func(batch []*keys.PreKey) []uint32 { + var out []uint32 + for _, k := range batch { + out = append(out, k.KeyID) + } + return out + } + first, _ := pk.GetOrGenPreKeys(ctx, 3) // 1..3; the upload then fails, nothing is marked + again, _ := pk.GetOrGenPreKeys(ctx, 3) + if !equalIDs(ids(again), ids(first)) || remote.nextID != 3 { + t.Fatalf("retry after a failed upload got %v (generated up to %d), want the same keys", ids(again), remote.nextID) + } + full, _ := pk.GetOrGenPreKeys(ctx, 2) // requested count cannot truncate pending three + if !equalIDs(ids(full), []uint32{1, 2, 3}) { + t.Fatalf("pending batch was truncated: %v", ids(full)) + } + if err := pk.MarkPreKeysAsUploaded(ctx, full[len(full)-1].KeyID); err != nil { + t.Fatal(err) + } + if len(remote.acks) != 1 || !equalIDs(remote.acks[0].ids, []uint32{1, 2, 3}) || len(remote.marked) != 0 { + t.Fatalf("acks=%v fresh marks=%v, want complete batch ACK only", remote.acks, remote.marked) + } + next, _ := pk.GetOrGenPreKeys(ctx, 2) + if !equalIDs(ids(next), []uint32{4, 5}) { + t.Fatalf("next batch %v, want [4 5]", ids(next)) + } + retry, err := pk.GenOnePreKey(ctx) + if err != nil { + t.Fatal(err) + } + last := remote.marked[len(remote.marked)-1] + if !equalIDs(last, []uint32{retry.KeyID}) { + t.Fatalf("retry-receipt prekey %d was not marked published (last mark %v)", retry.KeyID, last) + } + if n, _ := pk.UploadedPreKeyCount(ctx); n != 7 { + t.Fatalf("UploadedPreKeyCount = %d", n) + } +} + +func equalIDs(a, b []uint32) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +func TestRemoteDeviceSignatureMustVerify(t *testing.T) { + identity := keys.NewKeyPair() + remoteIdentity := testRemoteIdentity + remoteIdentity.IdentityKey = *identity.Pub + remote := &fakeRemoteSignal{} + cli := NewRemoteSignalClient(&store.Device{}, remoteIdentity, remote, nil) + account := keys.NewKeyPair() + verified := &waAdv.ADVSignedDeviceIdentity{Details: []byte("details"), AccountSignatureKey: account.Pub[:]} + signedBy := func(details []byte, signature []byte) []byte { + b, _ := proto.Marshal(&waAdv.ADVSignedDeviceIdentity{Details: details, DeviceSignature: signature}) + return b + } + valid := generateDeviceSignature(verified, identity)[:] + + remote.signed = signedBy([]byte("other details"), valid) + if _, err := cli.remoteDeviceSignature(context.Background(), nil, verified); err == nil { + t.Fatal("a signature over different details was accepted") + } + remote.signed = signedBy([]byte("details"), make([]byte, 64)) + if _, err := cli.remoteDeviceSignature(context.Background(), nil, verified); err == nil { + t.Fatal("a 64-byte signature that does not verify was accepted") + } + other := generateDeviceSignature(verified, keys.NewKeyPair())[:] + remote.signed = signedBy([]byte("details"), other) + if _, err := cli.remoteDeviceSignature(context.Background(), nil, verified); err == nil { + t.Fatal("a signature by a different identity was accepted") + } + remote.signed = signedBy([]byte("details"), valid) + if sig, err := cli.remoteDeviceSignature(context.Background(), nil, verified); err != nil || len(sig) != 64 { + t.Fatalf("valid signature rejected: %v", err) + } +} + +func marshalStanza(t *testing.T, node waBinary.Node) []byte { + t.Helper() + b, err := waBinary.Marshal(node) + if err != nil { + t.Fatal(err) + } + return b +} + +func TestSendPrebuiltStanzaWaitsForTheAck(t *testing.T) { + cli, _, sent := newTestRemote(t) + stanza := marshalStanza(t, waBinary.Node{ + Tag: "message", + Attrs: waBinary.Attrs{"id": "OUT1", "to": types.NewJID("123", types.HiddenUserServer), "type": "text"}, + }) + done := make(chan struct{}) + var resp SendResponse + var err error + go func() { + resp, err = cli.SendPrebuiltStanza(context.Background(), stanza, "", 5*time.Second) + close(done) + }() + deadline := time.Now().Add(5 * time.Second) + for len(sent.tags()) == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if err := cli.InjectNode(context.Background(), &waBinary.Node{ + Tag: "ack", + Attrs: waBinary.Attrs{"id": "OUT1", "class": "message", "t": "1700000001"}, + }); err != nil { + t.Fatal(err) + } + <-done + if err != nil || resp.ID != "OUT1" || resp.Timestamp.Unix() != 1700000001 { + t.Fatalf("resp=%+v err=%v", resp, err) + } + if tags := sent.tags(); len(tags) != 1 || tags[0] != "message:text" { + t.Fatalf("sent %v", tags) + } +} + +func TestSendPrebuiltStanzaRejectsNonMessages(t *testing.T) { + cli, _, _ := newTestRemote(t) + stanza := marshalStanza(t, waBinary.Node{Tag: "iq", Attrs: waBinary.Attrs{"id": "X"}}) + if _, err := cli.SendPrebuiltStanza(context.Background(), stanza, "", time.Second); err == nil { + t.Fatal("a non-message stanza was sent") + } +} + +func TestRemoteRetryReceiptAsksTheStoreToResend(t *testing.T) { + cli, remote, sent := newTestRemote(t) + requester := types.JID{User: "123", Server: types.HiddenUserServer, Device: 2} + remote.resendWith = marshalStanza(t, waBinary.Node{ + Tag: "message", + Attrs: waBinary.Attrs{"id": "MSG9", "to": requester, "type": "text"}, + }) + receipt := &events.Receipt{ + MessageSource: types.MessageSource{Chat: requester.ToNonAD(), Sender: requester}, + } + node := &waBinary.Node{ + Tag: "receipt", + Attrs: waBinary.Attrs{"id": "MSG9", "from": requester, "type": "retry"}, + Content: []waBinary.Node{{ + Tag: "retry", + Attrs: waBinary.Attrs{"id": "MSG9", "t": "1700000000", "count": "1"}, + }}, + } + if err := cli.handleRetryReceipt(context.Background(), receipt, node); err != nil { + t.Fatal(err) + } + if len(remote.resends) != 1 || remote.resends[0].MessageID != "MSG9" || remote.resends[0].RetryCount != 1 { + t.Fatalf("resends = %+v", remote.resends) + } + if tags := sent.tags(); len(tags) != 1 || tags[0] != "message:text" { + t.Fatalf("sent %v", tags) + } +} + +func TestRemoteClientRefusesLocalSends(t *testing.T) { + cli, _, _ := newTestRemote(t) + _, err := cli.SendMessage(context.Background(), types.NewJID("123", types.DefaultUserServer), nil) + if !errors.Is(err, ErrRemoteSignalUnsupported) { + t.Fatalf("SendMessage err = %v", err) + } + if _, err := cli.PairPhone(context.Background(), "15550001111", false, PairClientChrome, "Chrome (Linux)"); !errors.Is(err, ErrRemoteSignalUnsupported) { + t.Fatalf("PairPhone err = %v", err) + } +} + +func TestRemoteStoreOutageNeitherAcksNorRetries(t *testing.T) { + cli, remote, sent := newTestRemote(t) + remote.receiveErr = errors.New("connection refused") + announced := 0 + cli.AddEventHandler(func(evt any) { + switch evt.(type) { + case *events.RemoteMessage, *events.UndecryptableMessage: + announced++ + } + }) + info, node := encMessage("msg") + cli.decryptMessages(context.Background(), info, node) + if tags := sent.tags(); len(tags) != 0 { + t.Fatalf("sent %v during a store outage; the server must redeliver", tags) + } + if announced != 0 { + t.Fatal("no event should be emitted for a payload the store never processed") + } +} + +func TestRemoteRetryHonoursPreRetryCallback(t *testing.T) { + cli, remote, sent := newTestRemote(t) + var gotMessage bool + cli.PreRetryCallback = func(_ *events.Receipt, id types.MessageID, _ int, msg *waE2E.Message) bool { + gotMessage = msg != nil + return false + } + requester := types.JID{User: "123", Server: types.HiddenUserServer, Device: 2} + receipt := &events.Receipt{MessageSource: types.MessageSource{Chat: requester.ToNonAD(), Sender: requester}} + node := &waBinary.Node{ + Tag: "receipt", + Attrs: waBinary.Attrs{"id": "MSG8", "from": requester, "type": "retry"}, + Content: []waBinary.Node{{ + Tag: "retry", + Attrs: waBinary.Attrs{"id": "MSG8", "t": "1700000000", "count": "1"}, + }}, + } + if err := cli.handleRetryReceipt(context.Background(), receipt, node); err != nil { + t.Fatal(err) + } + if len(remote.resends) != 0 || len(sent.tags()) != 0 || gotMessage { + t.Fatalf("rejected retry still resent (resends=%d, sent=%v, message=%v)", len(remote.resends), sent.tags(), gotMessage) + } +} + +func TestRemoteIdentityChangeReachesTheStore(t *testing.T) { + for _, tc := range []struct{ from, want types.JID }{ + {types.JID{User: "123", Server: types.HiddenUserServer, Device: 3}, types.NewJID("123", types.HiddenUserServer)}, + {types.JID{User: "123", Server: types.HostedLIDServer, Device: 99}, types.NewJID("123", types.HiddenUserServer)}, + {types.JID{User: "5511", Server: types.HostedServer, Device: 99}, types.NewJID("5511", types.DefaultUserServer)}, + } { + cli, remote, _ := newTestRemote(t) + cli.deleteIdentityChangeState(context.Background(), tc.from) + if len(remote.forgotten) != 1 || remote.forgotten[0] != tc.want { + t.Fatalf("from %s: forgotten = %v, want [%s]", tc.from, remote.forgotten, tc.want) + } + } +} + +func TestRemoteClientRefusesLocalFBSends(t *testing.T) { + cli, _, _ := newTestRemote(t) + _, err := cli.SendFBMessage(context.Background(), types.NewJID("123", types.DefaultUserServer), nil, nil) + if !errors.Is(err, ErrRemoteSignalUnsupported) { + t.Fatalf("SendFBMessage err = %v", err) + } +} + +func TestRemotePNSenderMigratesSessionsInTheRemoteStore(t *testing.T) { + cli, remote, _ := newTestRemote(t) + pn := types.JID{User: "5511", Server: types.DefaultUserServer, Device: 3} + lid := types.JID{User: "777", Server: types.HiddenUserServer, Device: 3} + receive := func() { + info, node := encMessage("msg") + info.Chat, info.Sender, info.SenderAlt = pn.ToNonAD(), pn, lid + node.Attrs["from"] = pn + cli.decryptMessages(context.Background(), info, node) + } + + remote.migrateErr = errors.New("store offline") + receive() + remote.migrateErr = nil + receive() + receive() + + want := [2]types.JID{pn.ToNonAD(), lid.ToNonAD()} + if len(remote.migrated) != 2 || remote.migrated[0] != want || remote.migrated[1] != want { + t.Fatalf("migrations = %v, want a failed then one successful %v and no more", remote.migrated, want) + } + if len(remote.receives) != 3 || remote.receives[2].Sender != lid { + t.Fatalf("receives should decrypt as the LID %s: %+v", lid, remote.receives) + } +} diff --git a/retry.go b/retry.go index 343578bf0..71cb761fe 100644 --- a/retry.go +++ b/retry.go @@ -228,6 +228,30 @@ func (cli *Client) tryHandleRetryReceipt(ctx context.Context, receipt *events.Re } // handleRetryReceipt handles an incoming retry receipt for an outgoing message. +// acceptIncomingRetry counts a retry request from sender for messageID and +// reports whether it is still within the limits. +func (cli *Client) acceptIncomingRetry(sender types.JID, messageID types.MessageID) bool { + retryKey := incomingRetryKey{sender, messageID} + cli.incomingRetryRequestCounterLock.Lock() + internalCounter, accepted := incrementBoundedCounter( + ensureMap(&cli.incomingRetryRequestCounter), + retryKey, + maxIncomingRetryEntries, + &cli.incomingRetryRequestCounterReset, + time.Now(), + ) + cli.incomingRetryRequestCounterLock.Unlock() + if !accepted { + cli.Log.Warnf("Dropping retry request from %s for %s: retry counter capacity reached", messageID, sender) + return false + } + if internalCounter >= 10 { + cli.Log.Warnf("Dropping retry request from %s for %s: internal retry counter is %d", messageID, sender, internalCounter) + return false + } + return true +} + func (cli *Client) handleRetryReceipt(ctx context.Context, receipt *events.Receipt, node *waBinary.Node) error { retryChild, ok := node.GetOptionalChildByTag("retry") if !ok { @@ -240,6 +264,11 @@ func (cli *Client) handleRetryReceipt(ctx context.Context, receipt *events.Recei if !ag.OK() { return ag.Error() } + if cli.isRemoteSignal() { + // The remote store keeps the sent message; this client never had + // its plaintext. + return cli.handleRetryReceiptRemote(ctx, receipt, node, messageID, retryCount) + } msg, err := cli.getMessageForRetry(ctx, receipt, messageID) if err != nil { return err @@ -257,22 +286,7 @@ func (cli *Client) handleRetryReceipt(ctx context.Context, receipt *events.Recei } } - retryKey := incomingRetryKey{receipt.Sender, messageID} - cli.incomingRetryRequestCounterLock.Lock() - internalCounter, accepted := incrementBoundedCounter( - ensureMap(&cli.incomingRetryRequestCounter), - retryKey, - maxIncomingRetryEntries, - &cli.incomingRetryRequestCounterReset, - time.Now(), - ) - cli.incomingRetryRequestCounterLock.Unlock() - if !accepted { - cli.Log.Warnf("Dropping retry request from %s for %s: retry counter capacity reached", messageID, receipt.Sender) - return nil - } - if internalCounter >= 10 { - cli.Log.Warnf("Dropping retry request from %s for %s: internal retry counter is %d", messageID, receipt.Sender, internalCounter) + if !cli.acceptIncomingRetry(receipt.Sender, messageID) { return nil } diff --git a/send.go b/send.go index 3af53db66..be04556ea 100644 --- a/send.go +++ b/send.go @@ -195,6 +195,12 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E err = ErrClientIsNil return } + if cli.isRemoteSignal() { + // The remote Signal store builds and encrypts outgoing stanzas; this + // client only transmits them. + err = fmt.Errorf("SendMessage: %w; use SendPrebuiltStanza", ErrRemoteSignalUnsupported) + return + } var req SendRequestExtra if len(extra) > 1 { err = errors.New("only one extra parameter may be provided to SendMessage") diff --git a/sendfb.go b/sendfb.go index b26bb1b24..0eb95530b 100644 --- a/sendfb.go +++ b/sendfb.go @@ -52,6 +52,10 @@ func (cli *Client) SendFBMessage( err = ErrClientIsNil return } + if cli.isRemoteSignal() { + err = fmt.Errorf("SendFBMessage: %w; use SendPrebuiltStanza", ErrRemoteSignalUnsupported) + return + } var req SendRequestExtra if len(extra) > 1 { err = errors.New("only one extra parameter may be provided to SendMessage") diff --git a/socket/framesocket.go b/socket/framesocket.go index bcf107b2f..1c4ba0671 100644 --- a/socket/framesocket.go +++ b/socket/framesocket.go @@ -31,7 +31,9 @@ type FrameSocket struct { HTTPHeaders http.Header HTTPClient *http.Client - Frames chan []byte + Frames chan []byte + // OnDisconnect may be assigned before Connect. During socket lifetime, use + // SetDisconnectHandler to synchronize with concurrent Close/Stop. OnDisconnect func(ctx context.Context, remote bool) Header []byte @@ -64,6 +66,13 @@ func (fs *FrameSocket) IsConnected() bool { return fs.conn.Load() != nil } +// SetDisconnectHandler updates callback ownership under the same lock as Close. +func (fs *FrameSocket) SetDisconnectHandler(handler func(context.Context, bool)) { + fs.lock.Lock() + fs.OnDisconnect = handler + fs.lock.Unlock() +} + func (fs *FrameSocket) Close(code websocket.StatusCode) { fs.lock.Lock() defer fs.lock.Unlock() diff --git a/socket/noisesocket.go b/socket/noisesocket.go index 8175868ce..3cf60251e 100644 --- a/socket/noisesocket.go +++ b/socket/noisesocket.go @@ -45,9 +45,9 @@ func newNoiseSocket( onFrame: frameHandler, stopConsumer: make(chan struct{}), } - fs.OnDisconnect = func(ctx context.Context, remote bool) { + fs.SetDisconnectHandler(func(ctx context.Context, remote bool) { disconnectHandler(ctx, ns, remote) - } + }) go ns.consumeFrames(ctx, fs.Frames) return ns, nil } @@ -80,7 +80,7 @@ func (ns *NoiseSocket) Stop(disconnect, allowOnDisconnect bool) { if ns.destroyed.CompareAndSwap(false, true) { close(ns.stopConsumer) if !allowOnDisconnect { - ns.fs.OnDisconnect = nil + ns.fs.SetDisconnectHandler(nil) } if disconnect { ns.fs.Close(websocket.StatusNormalClosure) diff --git a/store/signal.go b/store/signal.go index 6cd816b43..728289bcf 100644 --- a/store/signal.go +++ b/store/signal.go @@ -26,7 +26,14 @@ var ErrUnsupportedSignalStoreOperation = errors.New("signal store operation is n var _ store.SignalProtocol = (*Device)(nil) +// errRemoteKeys explains a local Signal operation reached on a device whose +// private keys are remote. Reaching it is a programming error. +const errRemoteKeys = "whatsmeow: local Signal operation on a device with remote keys" + func (device *Device) GetIdentityKeyPair() *identity.KeyPair { + if device.RemoteKeys { + panic(errRemoteKeys + ": identity key") + } return identity.NewKeyPair( identity.NewKey(ecc.NewDjbECPublicKey(*device.IdentityKey.Pub)), ecc.NewDjbECPrivateKey(*device.IdentityKey.Priv), @@ -170,6 +177,9 @@ func (device *Device) LoadSignedPreKey(ctx context.Context, signedPreKeyID uint3 return nil, nil } if signedPreKeyID == device.SignedPreKey.KeyID { + if device.RemoteKeys { + panic(errRemoteKeys + ": signed prekey") + } return record.NewSignedPreKey(signedPreKeyID, 0, ecc.NewECKeyPair( ecc.NewDjbECPublicKey(*device.SignedPreKey.Pub), ecc.NewDjbECPrivateKey(*device.SignedPreKey.Priv), diff --git a/store/store.go b/store/store.go index 1cefa7c93..1038861b1 100644 --- a/store/store.go +++ b/store/store.go @@ -265,8 +265,12 @@ type Device struct { FacebookUUID uuid.UUID - Initialized bool - Deleted bool + Initialized bool + Deleted bool + // RemoteKeys marks a device whose Signal private keys live outside this + // process. Its private key fields are zero and must never be used; the + // local Signal store panics instead of computing with them. + RemoteKeys bool Identities IdentityStore Sessions SessionStore PreKeys PreKeyStore diff --git a/types/events/events.go b/types/events/events.go index ab2b1ece6..91d79de02 100644 --- a/types/events/events.go +++ b/types/events/events.go @@ -288,6 +288,19 @@ const ( UnavailableTypeViewOnce UnavailableType = "view_once" ) +// RemoteMessage is emitted by a client whose Signal keys are remote (see +// whatsmeow.NewRemoteSignalClient) when a message payload was handed to the +// remote store. The remote store delivers the content to the application; this +// event carries only delivery metadata. +type RemoteMessage struct { + Info types.MessageInfo + // Redelivered means the remote store served the payload from its buffer. + Redelivered bool + // SenderKeyRejected means a sender-key distribution in the message could + // not be stored. + SenderKeyRejected bool +} + // UndecryptableMessage is emitted when receiving a new message that failed to decrypt. // // The library will automatically ask the sender to retry. If the sender resends the message,