diff --git a/docs/native-send-admission.md b/docs/native-send-admission.md new file mode 100644 index 000000000..29174cdd3 --- /dev/null +++ b/docs/native-send-admission.md @@ -0,0 +1,28 @@ +# Optional native send admission (dark integration seam) + +`SendRequestExtra.NativeSend` is process-local and nil by default. The existing +send path is unchanged when nil. Initial support is one ordinary text DM with +an explicitly preallocated message ID; group, peer, bot, media and extra-node +paths are refused before preparation. This is not a renderer or a grant. + +`Admit(ctx, id, write)` runs after native preparation and after the Noise socket +write lock. It must synchronously invoke `write` at most once inside the final +local/authority fence. Returning without invocation refuses the send. A retained +callback cannot invoke after Admit returns. A refusal consumes no Noise nonce +or first-frame header. `Entered()` is a non-blocking notification immediately +before the actual WebSocket write call; it is evidence of entry, not provider +acceptance. It cannot prove whether bytes reached a provider. + +The hook returns before the provider ACK wait. Callers must not hold Campaign +locks through that wait. Persist an invocation CAS marker and preallocated ID +before writing; an existing marker with no durable outcome is unknown and must +never be reinvoked. No goroutine-start or AMQP-publish callback counts as entry. +A frame-write failure or post-entry authority loss is not proof of non-delivery. + +Guarded sends do not populate HyperMeow's retry content cache/store and do not +use the disconnect frame resend path. Integrators must also keep their +`GetMessageForRetry` fallback from returning guarded Campaign content and must +not implement an automatic retry around this API. Durable retry/receipt policy +and the regional invocation/result outbox belong to the owning application. +No Campaign consumer, positive authority adapter, database schema or runtime +mount is supplied here. This seam alone does not establish native acceptance. diff --git a/native_send.go b/native_send.go new file mode 100644 index 000000000..fb96f53ae --- /dev/null +++ b/native_send.go @@ -0,0 +1,63 @@ +package whatsmeow + +import ( + "context" + "errors" + + "google.golang.org/protobuf/reflect/protoreflect" + + waBinary "github.com/polymorfa/hypermeow/binary" + "github.com/polymorfa/hypermeow/proto/waE2E" + "github.com/polymorfa/hypermeow/types" +) + +var ErrNativeSendUnsupported = errors.New("native admission requires a preallocated ordinary text DM") + +// NativeSendHooks are process-local. Admit must synchronously invoke write at +// most once under the caller's final local/authority fence. Entered is a short +// notification immediately before the actual socket write; it must not block. +// The hooks finish before SendMessage waits for the provider ACK. +type NativeSendHooks struct { + Admit func(context.Context, types.MessageID, func() error) error + Entered func() +} + +func (cli *Client) sendNativeMessageNode(ctx context.Context, node waBinary.Node, id types.MessageID, hooks *NativeSendHooks) ([]byte, error) { + if hooks == nil { + return cli.sendNodeAndGetData(ctx, node) + } + if hooks.Admit == nil || hooks.Entered == nil || id == "" { + return nil, ErrNativeSendUnsupported + } + cli.socketLock.RLock() + sock := cli.socket + cli.socketLock.RUnlock() + // A shadow relay is a separate transport, not proof of native socket entry. + if sock == nil { + return nil, ErrNotConnected + } + payload, err := waBinary.Marshal(node) + if err != nil { + return nil, err + } + err = sock.SendFrameAdmitted(ctx, payload, func(ctx context.Context, write func() error) error { + return hooks.Admit(ctx, id, write) + }, hooks.Entered) + return payload, err +} + +// Restrict the initial hook to one ordinary text payload. Unknown fields and +// sibling message variants must not smuggle additional native effects. +func validNativeText(message *waE2E.Message) bool { + if message == nil || message.Conversation == nil || len(message.ProtoReflect().GetUnknown()) != 0 { + return false + } + valid := true + message.ProtoReflect().Range(func(field protoreflect.FieldDescriptor, _ protoreflect.Value) bool { + if field.Name() != "conversation" { + valid = false + } + return valid + }) + return valid +} diff --git a/native_send_test.go b/native_send_test.go new file mode 100644 index 000000000..1e18b41e2 --- /dev/null +++ b/native_send_test.go @@ -0,0 +1,39 @@ +package whatsmeow + +import ( + "context" + "errors" + "testing" + + "google.golang.org/protobuf/proto" + + "github.com/polymorfa/hypermeow/proto/waE2E" + "github.com/polymorfa/hypermeow/types" +) + +func TestNativeSendRejectsUnsupportedBeforePreparation(t *testing.T) { + cli := &Client{} + to := types.NewJID("15550000000", types.DefaultUserServer) + hooks := &NativeSendHooks{Admit: func(context.Context, types.MessageID, func() error) error { + t.Fatal("unexpected admission") + return nil + }, Entered: func() { t.Fatal("unexpected entry") }} + for _, tc := range []struct { + name string + message *waE2E.Message + extra SendRequestExtra + }{ + {"missing ID", &waE2E.Message{Conversation: proto.String("text")}, SendRequestExtra{NativeSend: hooks}}, + {"media", &waE2E.Message{ImageMessage: &waE2E.ImageMessage{}}, SendRequestExtra{ID: "original", NativeSend: hooks}}, + {"sibling", &waE2E.Message{Conversation: proto.String("text"), ImageMessage: &waE2E.ImageMessage{}}, SendRequestExtra{ID: "original", NativeSend: hooks}}, + {"peer", &waE2E.Message{Conversation: proto.String("text")}, SendRequestExtra{ID: "original", NativeSend: hooks, Peer: true}}, + {"missing callback", &waE2E.Message{Conversation: proto.String("text")}, SendRequestExtra{ID: "original", NativeSend: &NativeSendHooks{}}}, + } { + t.Run(tc.name, func(t *testing.T) { + _, err := cli.SendMessage(t.Context(), to, tc.message, tc.extra) + if !errors.Is(err, ErrNativeSendUnsupported) { + t.Fatalf("wrong refusal: %v", err) + } + }) + } +} diff --git a/send.go b/send.go index 48eafd8f9..192277e39 100644 --- a/send.go +++ b/send.go @@ -148,6 +148,10 @@ func setParticipantHashMismatch(resp *SendResponse, sent, acknowledged string) b // // Trying to add multiple extra parameters will return an error. type SendRequestExtra struct { + // NativeSend is an optional process-local final-frame admission. Supported + // only for a preallocated-ID ordinary one-to-one text send. Nil preserves + // the existing send path. It never crosses a serialized request. + NativeSend *NativeSendHooks // The message ID to use when sending. If this is not provided, a random message ID will be generated ID types.MessageID // JID of the bot to be invoked (optional) @@ -202,6 +206,15 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E } else if len(extra) == 1 { req = extra[0] } + if req.NativeSend != nil { + if req.ID == "" || req.NativeSend.Admit == nil || req.NativeSend.Entered == nil || req.Peer || !req.InlineBotJID.IsEmpty() || + (to.Server != types.DefaultUserServer && to.Server != types.HiddenUserServer) || !validNativeText(message) || req.Meta != nil || req.AdditionalNodes != nil || req.MediaHandle != "" || to.IsBot() { + err = ErrNativeSendUnsupported + return + } + hooks := *req.NativeSend + req.NativeSend = &hooks + } if to.Device > 0 && !req.Peer { err = ErrRecipientADJID return @@ -300,6 +313,8 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E } } + extraParams.nativeSend = req.NativeSend + var groupParticipants []types.JID if to.Server == types.GroupServer || to.Server == types.BroadcastServer { start := time.Now() @@ -377,7 +392,7 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E defer cli.messageSendLock.Unlock() // Peer message retries aren't implemented yet - if !req.Peer { + if !req.Peer && req.NativeSend == nil { err = cli.addRecentMessage(ctx, to, req.ID, message, nil) if err != nil { return @@ -434,6 +449,10 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E return } resp.DebugTimings.Resp = time.Since(start) + if isDisconnectNode(respNode) && req.NativeSend != nil { + err = &DisconnectedError{Action: "guarded message send"} + return + } if isDisconnectNode(respNode) { start = time.Now() respNode, err = cli.retryFrame(ctx, "message send", req.ID, data, respNode, 0) @@ -741,6 +760,7 @@ func (cli *Client) sendNewsletter( } type nodeExtraParams struct { + nativeSend *NativeSendHooks botNode *waBinary.Node metaNode *waBinary.Node additionalNodes *[]waBinary.Node @@ -892,7 +912,7 @@ func (cli *Client) sendDM( } start = time.Now() - data, err := cli.sendNodeAndGetData(ctx, *node) + data, err := cli.sendNativeMessageNode(ctx, *node, id, extraParams.nativeSend) timings.Send = time.Since(start) if err != nil { return "", nil, fmt.Errorf("failed to send message node: %w", err) diff --git a/socket/framesocket.go b/socket/framesocket.go index bcf107b2f..decb9684d 100644 --- a/socket/framesocket.go +++ b/socket/framesocket.go @@ -123,6 +123,10 @@ func (fs *FrameSocket) Context() context.Context { } func (fs *FrameSocket) SendFrame(data []byte) error { + return fs.sendFrame(data, nil, nil) +} + +func (fs *FrameSocket) sendFrame(data []byte, entered func(), writeCtx context.Context) error { conn := fs.conn.Load() if conn == nil { return ErrSocketClosed @@ -151,7 +155,13 @@ func (fs *FrameSocket) SendFrame(data []byte) error { // Copy actual frame data copy(wholeFrame[headerLength+FrameLengthSize:], data) - return conn.Write(fs.cancelCtx, websocket.MessageBinary, wholeFrame) + if entered != nil { + entered() + } + if writeCtx == nil { + writeCtx = fs.cancelCtx + } + return conn.Write(writeCtx, websocket.MessageBinary, wholeFrame) } func (fs *FrameSocket) frameComplete() { diff --git a/socket/native_admission.go b/socket/native_admission.go new file mode 100644 index 000000000..5256a806d --- /dev/null +++ b/socket/native_admission.go @@ -0,0 +1,50 @@ +package socket + +import ( + "context" + "errors" + "sync" +) + +var ErrNativeAdmissionClosed = errors.New("native send admission closed") + +// SendFrameAdmitted joins a process-local admission to an actual frame write. +// It runs after the Noise write lock, never around goroutine scheduling or ACK +// waiting. A refusal consumes neither the Noise nonce nor the first-frame header. +func (ns *NoiseSocket) SendFrameAdmitted(ctx context.Context, plaintext []byte, admit func(context.Context, func() error) error, entered func()) error { + if admit == nil || entered == nil { + return ErrNativeAdmissionClosed + } + ns.writeLock.Lock() + defer ns.writeLock.Unlock() + if err := ctx.Err(); err != nil { + return err + } + ciphertext := ns.writeKey.Seal(nil, generateIV(ns.writeCounter), plaintext, nil) + var mu sync.Mutex + active, invoked := true, false + var writeErr error + err := admit(ctx, func() error { + mu.Lock() + defer mu.Unlock() + if !active || invoked { + return ErrNativeAdmissionClosed + } + if err := ctx.Err(); err != nil { + return err + } + invoked = true + ns.writeCounter++ + writeErr = ns.fs.sendFrame(ciphertext, entered, ctx) + return writeErr + }) + mu.Lock() + active = false + didInvoke := invoked + result := writeErr + mu.Unlock() + if !didInvoke && err == nil { + return ErrNativeAdmissionClosed + } + return errors.Join(err, result) +} diff --git a/socket/native_admission_test.go b/socket/native_admission_test.go new file mode 100644 index 000000000..4db114492 --- /dev/null +++ b/socket/native_admission_test.go @@ -0,0 +1,140 @@ +package socket + +import ( + "bytes" + "context" + "crypto/aes" + "crypto/cipher" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "github.com/coder/websocket" + + waLog "github.com/polymorfa/hypermeow/util/log" +) + +func testAdmittedSocket(t *testing.T) *NoiseSocket { + t.Helper() + block, err := aes.NewCipher(make([]byte, 32)) + if err != nil { + t.Fatal(err) + } + key, err := cipher.NewGCM(block) + if err != nil { + t.Fatal(err) + } + return &NoiseSocket{fs: newNativeTestFrameSocket(t), writeKey: key} +} + +func TestNativeAdmissionDenialPreservesNonceAndHeader(t *testing.T) { + ns := testAdmittedSocket(t) + header := bytes.Clone(ns.fs.Header) + denied := errors.New("revoked") + entered := false + err := ns.SendFrameAdmitted(t.Context(), []byte("message"), func(context.Context, func() error) error { return denied }, func() { entered = true }) + if !errors.Is(err, denied) || entered || ns.writeCounter != 0 || !bytes.Equal(ns.fs.Header, header) { + t.Fatalf("denial changed write state: %v, %v, %d", err, entered, ns.writeCounter) + } +} + +func TestNativeAdmissionRejectsRetainedAndRepeatedCallback(t *testing.T) { + ns := testAdmittedSocket(t) + var retained func() error + err := ns.SendFrameAdmitted(t.Context(), nil, func(_ context.Context, write func() error) error { retained = write; return nil }, func() { t.Error("entered after refusal") }) + if !errors.Is(err, ErrNativeAdmissionClosed) || !errors.Is(retained(), ErrNativeAdmissionClosed) || ns.writeCounter != 0 { + t.Fatal("retained callback remained usable") + } + count := 0 + err = ns.SendFrameAdmitted(t.Context(), []byte("message"), func(_ context.Context, write func() error) error { + if err := write(); err != nil { + return err + } + if !errors.Is(write(), ErrNativeAdmissionClosed) { + t.Error("second invocation accepted") + } + return nil + }, func() { count++ }) + if err != nil || count != 1 || ns.writeCounter != 1 { + t.Fatalf("single write failed: %v, count %d", err, count) + } +} + +func TestNativeAdmissionRechecksAfterWriteLock(t *testing.T) { + ns := testAdmittedSocket(t) + var revoked atomic.Bool + started := make(chan struct{}) + done := make(chan error, 1) + ns.writeLock.Lock() + go func() { + close(started) + done <- ns.SendFrameAdmitted(t.Context(), nil, func(_ context.Context, write func() error) error { + if revoked.Load() { + return ErrNativeAdmissionClosed + } + return write() + }, func() { t.Error("revoked send entered") }) + }() + <-started + revoked.Store(true) + ns.writeLock.Unlock() + if err := <-done; !errors.Is(err, ErrNativeAdmissionClosed) || ns.writeCounter != 0 { + t.Fatalf("lock wait bypassed authority: %v", err) + } +} + +func TestNativeAdmissionCancellationAndClosedSocketNeverEnter(t *testing.T) { + for _, closed := range []bool{false, true} { + t.Run(map[bool]string{false: "cancelled", true: "closed"}[closed], func(t *testing.T) { + ns := testAdmittedSocket(t) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + if closed { + ns.fs.Close(0) + } else { + cancel() + } + err := ns.SendFrameAdmitted(ctx, nil, func(_ context.Context, write func() error) error { return write() }, func() { t.Error("non-writable socket entered") }) + if err == nil { + t.Fatal("missing refusal") + } + }) + } +} + +func TestNativeAdmissionEntryIsInsideWriteAndBeforeReturn(t *testing.T) { + ns := testAdmittedSocket(t) + order := []string{} + err := ns.SendFrameAdmitted(t.Context(), []byte("message"), func(_ context.Context, write func() error) error { + order = append(order, "admit") + err := write() + order = append(order, "written") + return err + }, func() { order = append(order, "entered") }) + if err != nil || len(order) != 3 || order[0] != "admit" || order[1] != "entered" || order[2] != "written" { + t.Fatalf("wrong entry order: %v, %v", order, err) + } +} + +func newNativeTestFrameSocket(t *testing.T) *FrameSocket { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true}) + if err != nil { + t.Error(err) + return + } + defer conn.CloseNow() + <-conn.CloseRead(r.Context()).Done() + })) + t.Cleanup(server.Close) + fs := NewFrameSocket(waLog.Noop, server.Client()) + fs.URL = server.URL + if err := fs.Connect(t.Context()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { fs.Close(0) }) + return fs +}