From e55720c263ace1acedcfbd256bc0cc9421bfdcdb Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 15:28:56 -0400 Subject: [PATCH 1/9] fix: authenticate inbound LINE media before decryption Adapt https://github.com/beeper/line/pull/196 (a6d06b0cde098719ac78ed9dd673bfa7244cea17) to the shared media decryptor, covering both uploads and direct media. Preserve upstream fail-closed key fallback and route video to chunk-hash HMAC verification. Add valid, tampered and wrong-mode coverage for all media kinds. Co-authored-by: Adrien Vermersch --- pkg/connector/handle_message.go | 11 +++--- pkg/connector/handlers/handler.go | 10 ++++-- pkg/connector/handlers/handler_test.go | 2 +- pkg/connector/media.go | 25 +++++++++++--- pkg/connector/media_hmac_test.go | 48 ++++++++++++++++++++++++++ 5 files changed, 84 insertions(+), 12 deletions(-) create mode 100644 pkg/connector/media_hmac_test.go diff --git a/pkg/connector/handle_message.go b/pkg/connector/handle_message.go index 57640eaa..d2859c64 100644 --- a/pkg/connector/handle_message.go +++ b/pkg/connector/handle_message.go @@ -29,11 +29,12 @@ const ( func (lc *LineClient) newMessageHandler() *handlers.Handler { handler := &handlers.Handler{ - Log: lc.UserLogin.Bridge.Log, - HTTPClient: lc.HTTPClient, - RecoverClient: lc.recoverClientAfterAuthError, - NewClient: func() *line.Client { return lc.newClient() }, - DecryptMedia: lc.decryptImageData, + Log: lc.UserLogin.Bridge.Log, + HTTPClient: lc.HTTPClient, + RecoverClient: lc.recoverClientAfterAuthError, + NewClient: func() *line.Client { return lc.newClient() }, + DecryptMedia: lc.decryptImageData, + DecryptVideoMedia: lc.decryptVideoData, } connector, ok := lc.UserLogin.Bridge.Network.(*LineConnector) if ok && connector.directMedia.Load() { diff --git a/pkg/connector/handlers/handler.go b/pkg/connector/handlers/handler.go index 768604e0..bc421d0c 100644 --- a/pkg/connector/handlers/handler.go +++ b/pkg/connector/handlers/handler.go @@ -40,6 +40,8 @@ type Handler struct { // DecryptMedia decrypts E2EE encrypted media data using the given key material. DecryptMedia func(data []byte, keyMaterial string) ([]byte, error) + // DecryptVideoMedia verifies the video-specific chunk-hash HMAC before decryption. + DecryptVideoMedia func(data []byte, keyMaterial string) ([]byte, error) } func (h *Handler) decryptDownloadedMedia(data []byte, decryptedBody string, metadata map[string]string, kind string) ([]byte, error) { @@ -90,13 +92,17 @@ func (h *Handler) decryptMediaWithKeys(data []byte, keys []string, encrypted boo if len(keys) == 0 { return nil, fmt.Errorf("%w: encrypted %s has no usable media key", bridgev2.ErrIgnoringRemoteEvent, strings.ToLower(kind)) } - if h.DecryptMedia == nil { + decrypt := h.DecryptMedia + if kind == "video" { + decrypt = h.DecryptVideoMedia + } + if decrypt == nil { return nil, fmt.Errorf("%w: encrypted %s has no media decryptor", bridgev2.ErrIgnoringRemoteEvent, strings.ToLower(kind)) } decryptErrors := make([]error, 0, len(keys)) for _, key := range keys { - decrypted, err := h.DecryptMedia(data, key) + decrypted, err := decrypt(data, key) if err == nil { return decrypted, nil } diff --git a/pkg/connector/handlers/handler_test.go b/pkg/connector/handlers/handler_test.go index 19057e03..c825b11d 100644 --- a/pkg/connector/handlers/handler_test.go +++ b/pkg/connector/handlers/handler_test.go @@ -88,7 +88,7 @@ func TestDecryptDownloadedMediaFailsClosed(t *testing.T) { ciphertext := []byte("ciphertext") decryptErr := errors.New("invalid media key") var calls int - h := &Handler{DecryptMedia: func([]byte, string) ([]byte, error) { + h := &Handler{DecryptVideoMedia: func([]byte, string) ([]byte, error) { calls++ return nil, decryptErr }} diff --git a/pkg/connector/media.go b/pkg/connector/media.go index a65c3add..36b77d1f 100644 --- a/pkg/connector/media.go +++ b/pkg/connector/media.go @@ -92,6 +92,16 @@ func convertHEICToJPEG(ctx context.Context, imageData []byte) ([]byte, error) { // LINE's E2EE file format: [encrypted_data][32-byte HMAC] // The keyMaterial is derived using HKDF to get encKey (32), macKey (32), and nonce (12 bytes) func (lc *LineClient) decryptImageData(encryptedData []byte, keyMaterialB64 string) ([]byte, error) { + return lc.decryptMediaData(encryptedData, keyMaterialB64, func(ciphertext []byte) []byte { + return ciphertext + }) +} + +func (lc *LineClient) decryptVideoData(encryptedData []byte, keyMaterialB64 string) ([]byte, error) { + return lc.decryptMediaData(encryptedData, keyMaterialB64, generateChunkHashes) +} + +func (lc *LineClient) decryptMediaData(encryptedData []byte, keyMaterialB64 string, macInput func([]byte) []byte) ([]byte, error) { keyMaterial, err := base64.StdEncoding.DecodeString(keyMaterialB64) if err != nil { return nil, fmt.Errorf("failed to decode key material: %w", err) @@ -106,7 +116,7 @@ func (lc *LineClient) decryptImageData(encryptedData []byte, keyMaterialB64 stri } encKey := derived[0:32] - // macKey := derived[32:64] // for HMAC verification + macKey := derived[32:64] nonce := derived[64:76] // Create 16-byte counter: nonce(12 bytes) + zero counter(4 bytes) @@ -117,7 +127,14 @@ func (lc *LineClient) decryptImageData(encryptedData []byte, keyMaterialB64 stri if len(encryptedData) < 32 { return nil, fmt.Errorf("encrypted data too short (< 32 bytes for HMAC)") } - encryptedData = encryptedData[:len(encryptedData)-32] + ciphertext := encryptedData[:len(encryptedData)-32] + receivedMAC := encryptedData[len(encryptedData)-32:] + + h := hmac.New(sha256.New, macKey) + h.Write(macInput(ciphertext)) + if !hmac.Equal(receivedMAC, h.Sum(nil)) { + return nil, fmt.Errorf("media HMAC verification failed") + } block, err := aes.NewCipher(encKey) if err != nil { @@ -126,8 +143,8 @@ func (lc *LineClient) decryptImageData(encryptedData []byte, keyMaterialB64 stri stream := cipher.NewCTR(block, counter) - decrypted := make([]byte, len(encryptedData)) - stream.XORKeyStream(decrypted, encryptedData) + decrypted := make([]byte, len(ciphertext)) + stream.XORKeyStream(decrypted, ciphertext) return decrypted, nil } diff --git a/pkg/connector/media_hmac_test.go b/pkg/connector/media_hmac_test.go new file mode 100644 index 00000000..57fa7425 --- /dev/null +++ b/pkg/connector/media_hmac_test.go @@ -0,0 +1,48 @@ +package connector + +import ( + "bytes" + "testing" +) + +// Adapted from beeper/line#196. +func TestMediaHMAC(t *testing.T) { + lc := &LineClient{} + for _, kind := range []string{"image", "file", "audio", "video", "thumbnail"} { + t.Run(kind, func(t *testing.T) { + plaintext := bytes.Repeat([]byte("authenticated media"), 40000) + encrypt := lc.encryptFileData + if kind == "video" { + encrypt = lc.encryptVideoData + } + ciphertext, key, err := encrypt(plaintext) + if err != nil { + t.Fatal(err) + } + if kind == "thumbnail" { + ciphertext, err = encryptThumbnail(plaintext, key) + if err != nil { + t.Fatal(err) + } + } + decrypt, wrongMode := lc.decryptImageData, lc.decryptVideoData + if kind == "video" { + decrypt, wrongMode = wrongMode, decrypt + } + decrypted, err := decrypt(ciphertext, key) + if err != nil || !bytes.Equal(decrypted, plaintext) { + t.Fatalf("valid media: err=%v, equal=%v", err, bytes.Equal(decrypted, plaintext)) + } + for _, offset := range []int{0, len(ciphertext) - 1} { + tampered := bytes.Clone(ciphertext) + tampered[offset] ^= 1 + if got, err := decrypt(tampered, key); err == nil || got != nil { + t.Fatalf("tampering at %d returned plaintext or no error", offset) + } + } + if got, err := wrongMode(ciphertext, key); err == nil || got != nil { + t.Fatal("accepted incorrect HMAC mode") + } + }) + } +} From ca81ee360659ba02a16168d27b56f36334dcc772 Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 15:30:16 -0400 Subject: [PATCH 2/9] fix: attach detected metadata to inbound LINE images Select the image metadata changes from https://github.com/beeper/line/pull/189 at a4d8c11884a88527e5dbb50c80f7c1dd52b8ad64, retaining current download, size-limit and authentication handling. Keep the separate 60-retry OBS policy out of this scoped fix. Includes bounded ISO-BMFF brand detection and regression tests. Co-authored-by: Adrien Vermersch --- pkg/connector/handlers/image.go | 12 ++- pkg/connector/handlers/image_media.go | 98 ++++++++++++++++++++++ pkg/connector/handlers/image_media_test.go | 70 ++++++++++++++++ 3 files changed, 178 insertions(+), 2 deletions(-) create mode 100644 pkg/connector/handlers/image_media.go create mode 100644 pkg/connector/handlers/image_media_test.go diff --git a/pkg/connector/handlers/image.go b/pkg/connector/handlers/image.go index 7c3ccdbd..ac3c41f7 100644 --- a/pkg/connector/handlers/image.go +++ b/pkg/connector/handlers/image.go @@ -112,8 +112,15 @@ func (h *Handler) ConvertImage(ctx context.Context, portal *bridgev2.Portal, int } // Upload to Matrix + imageMedia := lineImageMediaInfo(imgData) + if imageMedia.usedMimeFallback { + h.Log.Debug().Int("size_bytes", len(imgData)).Msg("Falling back to JPEG MIME type for LINE image") + } + if imageMedia.decodeErr != nil { + h.Log.Debug().Err(imageMedia.decodeErr).Str("mime_type", imageMedia.mimeType).Msg("Could not decode LINE image dimensions") + } uploadStart := time.Now() - mxc, file, err := intent.UploadMedia(ctx, portal.MXID, imgData, "image.jpg", "image/jpeg") + mxc, file, err := intent.UploadMedia(ctx, portal.MXID, imgData, imageMedia.fileName, imageMedia.mimeType) uploadDuration := time.Since(uploadStart) if err != nil { h.Log.Error(). @@ -145,7 +152,8 @@ func (h *Handler) ConvertImage(ctx context.Context, portal *bridgev2.Portal, int Type: event.EventMessage, Content: &event.MessageEventContent{ MsgType: event.MsgImage, - Body: "image.jpg", + Body: imageMedia.fileName, + Info: imageMedia.info, URL: mxc, File: file, RelatesTo: relatesTo, diff --git a/pkg/connector/handlers/image_media.go b/pkg/connector/handlers/image_media.go new file mode 100644 index 00000000..74bd4975 --- /dev/null +++ b/pkg/connector/handlers/image_media.go @@ -0,0 +1,98 @@ +package handlers + +import ( + "bytes" + "encoding/binary" + "image" + _ "image/gif" + _ "image/jpeg" + _ "image/png" + "net/http" + "strings" + + _ "golang.org/x/image/webp" + "maunium.net/go/mautrix/event" +) + +// Adapted from beeper/line#189; OBS retry policy remains independent. +type lineImageMedia struct { + fileName string + mimeType string + info *event.FileInfo + usedMimeFallback bool + decodeErr error +} + +func lineImageMediaInfo(data []byte) lineImageMedia { + mimeType, usedFallback := detectLineImageMimeType(data) + info := &event.FileInfo{MimeType: mimeType, Size: len(data)} + var decodeErr error + if config, _, err := image.DecodeConfig(bytes.NewReader(data)); err == nil { + info.Width = config.Width + info.Height = config.Height + } else if !usedFallback { + switch mimeType { + case "image/jpeg", "image/png", "image/gif", "image/webp": + decodeErr = err + } + } + return lineImageMedia{ + fileName: "image." + imageExtensionForMIME(mimeType), mimeType: mimeType, + info: info, usedMimeFallback: usedFallback, decodeErr: decodeErr, + } +} + +func detectLineImageMimeType(data []byte) (string, bool) { + mimeType := normalizedImageMIMEType(http.DetectContentType(data)) + if strings.HasPrefix(mimeType, "image/") { + return mimeType, false + } + if len(data) >= 12 && lineImageBrandAt(data, 4) == [4]byte{'f', 't', 'y', 'p'} { + boxSize := binary.BigEndian.Uint32(data[:4]) + if boxSize >= 16 && uint64(boxSize) <= uint64(len(data)) { + detected := "" + for offset, end := 8, int(boxSize); offset+4 <= end; { + switch lineImageBrandAt(data, offset) { + case [4]byte{'a', 'v', 'i', 'f'}, [4]byte{'a', 'v', 'i', 's'}: + return "image/avif", false + case [4]byte{'h', 'e', 'i', 'c'}, [4]byte{'h', 'e', 'i', 'x'}, [4]byte{'h', 'e', 'v', 'c'}, [4]byte{'h', 'e', 'v', 'x'}: + return "image/heic", false + case [4]byte{'h', 'e', 'i', 'f'}, [4]byte{'h', 'e', 'i', 'm'}, [4]byte{'h', 'e', 'i', 's'}, [4]byte{'m', 'i', 'f', '1'}, [4]byte{'m', 's', 'f', '1'}: + detected = "image/heif" + } + if offset == 8 { + offset = 16 + } else { + offset += 4 + } + } + if detected != "" { + return detected, false + } + } + } + return "image/jpeg", true +} + +func lineImageBrandAt(data []byte, offset int) [4]byte { + return [4]byte{data[offset], data[offset+1], data[offset+2], data[offset+3]} +} + +func imageExtensionForMIME(mimeType string) string { + switch normalizedMIMEType := normalizedImageMIMEType(mimeType); normalizedMIMEType { + case "image/jpeg": + return "jpg" + case "image/svg+xml": + return "svg" + case "image/x-icon", "image/vnd.microsoft.icon": + return "ico" + case "image/png", "image/gif", "image/webp", "image/heic", "image/heif", "image/avif", "image/bmp", "image/tiff": + return strings.TrimPrefix(normalizedMIMEType, "image/") + } + return "jpg" +} + +func normalizedImageMIMEType(mimeType string) string { + normalizedMIMEType, _, _ := strings.Cut(mimeType, ";") + return strings.ToLower(strings.TrimSpace(normalizedMIMEType)) +} diff --git a/pkg/connector/handlers/image_media_test.go b/pkg/connector/handlers/image_media_test.go new file mode 100644 index 00000000..cc831ead --- /dev/null +++ b/pkg/connector/handlers/image_media_test.go @@ -0,0 +1,70 @@ +package handlers + +import ( + "bytes" + "encoding/binary" + "image" + "image/jpeg" + "image/png" + "testing" +) + +func TestLineImageMediaInfo(t *testing.T) { + for _, format := range []string{"jpeg", "png"} { + t.Run(format, func(t *testing.T) { + var buf bytes.Buffer + img := image.NewRGBA(image.Rect(0, 0, 23, 17)) + var err error + if format == "jpeg" { + err = jpeg.Encode(&buf, img, nil) + } else { + err = png.Encode(&buf, img) + } + if err != nil { + t.Fatal(err) + } + got := lineImageMediaInfo(buf.Bytes()) + if got.mimeType != "image/"+format || got.info.Width != 23 || got.info.Height != 17 || got.info.Size != buf.Len() || got.info.MimeType != got.mimeType || got.usedMimeFallback || got.decodeErr != nil { + t.Fatalf("incorrect image metadata: %+v, info=%+v", got, got.info) + } + }) + } +} + +func TestLineImageBrands(t *testing.T) { + for _, tt := range []struct{ major, compatible, want string }{ + {"heic", "", "image/heic"}, {"heif", "", "image/heif"}, + {"mif1", "heic", "image/heic"}, {"mif1", "avif", "image/avif"}, + {"avif", "", "image/avif"}, + } { + t.Run(tt.major+tt.compatible, func(t *testing.T) { + data := make([]byte, 20) + binary.BigEndian.PutUint32(data, 20) + copy(data[4:], "ftyp") + copy(data[8:], tt.major) + copy(data[16:], tt.compatible) + got := lineImageMediaInfo(data) + if got.mimeType != tt.want || got.usedMimeFallback || got.decodeErr != nil { + t.Fatalf("unexpected metadata: %+v", got) + } + for _, size := range []uint32{0, 12, 1024, 0xffffffff} { + binary.BigEndian.PutUint32(data, size) + if got := lineImageMediaInfo(data); !got.usedMimeFallback { + t.Fatalf("accepted invalid ftyp size %d", size) + } + } + }) + } +} + +func TestLineImageExtensions(t *testing.T) { + for mimeType, want := range map[string]string{ + "image/jpeg": "jpg", "image/png; charset=binary": "png", "image/gif": "gif", + "image/webp": "webp", "image/heic": "heic", "image/heif": "heif", "image/avif": "avif", + "image/bmp": "bmp", "image/tiff": "tiff", "image/svg+xml": "svg", "image/x-icon": "ico", "unknown": "jpg", + } { + if got := imageExtensionForMIME(mimeType); got != want { + t.Errorf("%s: got %s, want %s", mimeType, got, want) + } + } +} From 3aa69d8cbbec47dcb71b44d217ee509162f42c04 Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 16:00:36 -0400 Subject: [PATCH 3/9] test: make signing and cancellation checks race-stable Replace 6000 expensive eviction cache misses with three complete cache rotations, per-call victim and LRU assertions, and explicit refresh-boundary coverage. Preserve the separate 6000-call reuse and real heap-exhaustion tests. Give cold signing a separate setup budget while retaining the one-second cancellation bound; cancel and join the request worker on every exit path. --- pkg/line/methods_context_test.go | 17 +++++++++++++- pkg/runner_secret_test.go | 40 +++++++++++++++++++++++++++++++- 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/pkg/line/methods_context_test.go b/pkg/line/methods_context_test.go index b6b0bcf1..ff591417 100644 --- a/pkg/line/methods_context_test.go +++ b/pkg/line/methods_context_test.go @@ -21,14 +21,29 @@ func testRPCContextCancellation(t *testing.T, method string, call func(*Client, } ctx, cancel := context.WithCancel(context.Background()) + defer cancel() result := make(chan error, 1) + done := make(chan struct{}) go func() { + defer close(done) result <- call(client, ctx) }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(30 * time.Second): + t.Errorf("%s worker did not exit after cleanup cancellation", method) + } + }) select { case <-requestStarted: - case <-time.After(time.Second): + case err := <-result: + t.Fatalf("%s failed before reaching transport: %v", method, err) + // Cold white-box signing happens before RoundTrip and is substantially + // slower under -race. This is a setup bound, not the cancellation bound. + case <-time.After(30 * time.Second): t.Fatalf("%s request did not start", method) } cancel() diff --git a/pkg/runner_secret_test.go b/pkg/runner_secret_test.go index 3ce7a194..45127d0f 100644 --- a/pkg/runner_secret_test.go +++ b/pkg/runner_secret_test.go @@ -2,6 +2,7 @@ package line import ( "bytes" + "crypto/sha256" "encoding/base64" "errors" "fmt" @@ -185,8 +186,16 @@ func TestGetSignatureEvictsSigningContexts(t *testing.T) { } r := newSigningTestRunner(t) wantByToken := make(map[string]string) - for i := 0; i < 6000; i++ { + t.Cleanup(r.Close) + // Three complete rotations exercise initial fill, eviction and recreation + // of every token. Thousands of white-box key derivations test throughput, + // not additional cache states, and exceed the race suite's time budget. + for i := 0; i < 3*(signingCacheCapacity+2); i++ { token := fmt.Sprintf("token-%d", i%(signingCacheCapacity+2)) + var victim [sha256.Size]byte + if len(r.signingLRU) == signingCacheCapacity { + victim = r.signingLRU[0] + } signature, err := r.GetSignature("/api/test", "{}", token) if err != nil { t.Fatalf("signature %d failed: %v", i+1, err) @@ -196,10 +205,39 @@ func TestGetSignatureEvictsSigningContexts(t *testing.T) { } else if signature != want { t.Fatalf("signature for %q changed: got %q want %q", token, signature, want) } + if got := len(r.signingContexts); got != min(i+1, signingCacheCapacity) { + t.Fatalf("signature %d: cache size %d", i+1, got) + } + if len(r.signingLRU) != len(r.signingContexts) { + t.Fatal("LRU and context map sizes differ") + } + if _, retained := r.signingContexts[victim]; retained { + t.Fatal("least recently used signing context was not evicted") + } + key := sha256.Sum256([]byte(token)) + if r.signingLRU[len(r.signingLRU)-1] != key { + t.Fatal("new signing context is not most recently used") + } } if got := len(r.signingContexts); got != signingCacheCapacity { t.Fatalf("unexpected signing cache size: got %d want %d", got, signingCacheCapacity) } + // Exercise the refresh boundary with a populated, churned cache without + // deriving thousands of keys just to advance the successful-call counter. + oldRuntime := r.signingRT + r.signingCalls = signingRefreshCalls - 1 + for i := 0; i < 2; i++ { + got, err := r.GetSignature("/api/test", "{}", "token-5") + if err != nil || got != wantByToken["token-5"] { + t.Fatalf("signature across refresh = %q, %v", got, err) + } + if (r.signingRT != oldRuntime) != (i == 1) { + t.Fatal("signing runtime refreshed at the wrong call boundary") + } + } + if len(r.signingContexts) != 1 || len(r.signingLRU) != 1 { + t.Fatal("refresh retained stale signing contexts") + } } func TestGetSignatureRecoversFromExhaustedRuntime(t *testing.T) { From 15ff9e49a4855f91476eb194e4481c88a487ebfc Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 17:03:48 -0400 Subject: [PATCH 4/9] fix: resolve public LINE IDs before creating DM portals Share exact identifier lookup with search, validate canonical user MIDs and reuse returned contact data. Delay portal persistence until remote resolution succeeds, honor lookup-only requests, and pass complete DM information to bridgev2. Add cancellable lookup RPCs and isolated database regressions for identity consistency, errors, cancellation, concurrent portal reuse and Matrix creation retries. --- pkg/connector/resolve_identifier.go | 92 ++++++ pkg/connector/resolve_identifier_test.go | 346 +++++++++++++++++++++++ pkg/connector/userinfo.go | 66 +++-- pkg/line/methods.go | 24 +- pkg/line/methods_context_test.go | 20 ++ 5 files changed, 519 insertions(+), 29 deletions(-) create mode 100644 pkg/connector/resolve_identifier.go create mode 100644 pkg/connector/resolve_identifier_test.go diff --git a/pkg/connector/resolve_identifier.go b/pkg/connector/resolve_identifier.go new file mode 100644 index 00000000..76772dab --- /dev/null +++ b/pkg/connector/resolve_identifier.go @@ -0,0 +1,92 @@ +package connector + +import ( + "context" + "fmt" + "strings" + "time" + "unicode" + + "github.com/beeper/line/pkg/line" +) + +func isCanonicalUserMID(identifier string) bool { + if len(identifier) != 33 || identifier[0] != 'u' { + return false + } + for _, ch := range identifier[1:] { + if !(ch >= '0' && ch <= '9' || ch >= 'a' && ch <= 'f') { + return false + } + } + return true +} + +// resolveContactIdentifier never matches display names or creates portals. +// Public IDs are resolved afresh: caching an ID->MID mapping could route a +// reassigned public ID to the wrong recipient. Contact data is cached by MID. +func (lc *LineClient) resolveContactIdentifier(ctx context.Context, identifier string) (*line.Contact, error) { + if err := lc.checkLineCall(ctx); err != nil { + return nil, err + } + identifier = strings.ToLower(strings.TrimSpace(identifier)) + if identifier == "" || strings.ContainsFunc(identifier, unicode.IsSpace) || strings.ContainsFunc(identifier, unicode.IsControl) { + return nil, fmt.Errorf("provide a LINE public ID or user MID, not a display name") + } + var contact *line.Contact + var err error + if isCanonicalUserMID(identifier) { + lc.cacheMu.Lock() + cached, ok := lc.contactCache[identifier] + lc.cacheMu.Unlock() + if ok && cached.Contact.Mid == identifier && time.Since(cached.cachedAt) < contactCacheTTL { + return &cached.Contact, nil + } + if lc.isOwnMID(identifier) { + _, profile, lookupErr := callLineResult(lc, ctx, func(client *line.Client) (*line.Profile, error) { + return client.GetProfileContext(ctx) + }) + err = lookupErr + if profile != nil { + contact = &line.Contact{Mid: profile.Mid, DisplayName: profile.DisplayName, PicturePath: profile.PicturePath} + } + } else { + _, contacts, lookupErr := callLineResult(lc, ctx, func(client *line.Client) (*line.ContactsResponse, error) { + return client.GetContactsV2Context(ctx, []string{identifier}) + }) + err = lookupErr + if contacts != nil { + if wrapper, found := contacts.Contacts[identifier]; found { + contact = &wrapper.Contact + } + } + if err == nil && contact == nil { + _, buddy, buddyErr := callLineResult(lc, ctx, func(client *line.Client) (*line.BuddyProfile, error) { + return client.GetBuddyProfileContext(ctx, identifier) + }) + err = buddyErr + if buddy != nil && buddy.Mid != "" { + contact = &line.Contact{Mid: buddy.Mid, DisplayName: buddy.DisplayName, PicturePath: buddy.PicturePath} + } + } + } + } else { + _, contact, err = callLineResult(lc, ctx, func(client *line.Client) (*line.Contact, error) { + return client.FindContactByUseridContext(ctx, identifier) + }) + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + if err != nil { + return nil, fmt.Errorf("LINE identifier lookup failed (the account may restrict ID lookup): %w", err) + } + if contact == nil || contact.Mid == "" { + return nil, nil + } + if !isCanonicalUserMID(contact.Mid) || isCanonicalUserMID(identifier) && contact.Mid != identifier { + return nil, fmt.Errorf("LINE lookup returned an invalid or mismatched user MID") + } + lc.setCachedContact(contact.Mid, *contact) + return contact, nil +} diff --git a/pkg/connector/resolve_identifier_test.go b/pkg/connector/resolve_identifier_test.go new file mode 100644 index 00000000..b653cf05 --- /dev/null +++ b/pkg/connector/resolve_identifier_test.go @@ -0,0 +1,346 @@ +package connector + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/rs/zerolog" + "go.mau.fi/util/dbutil" + "maunium.net/go/mautrix" + "maunium.net/go/mautrix/bridgev2" + "maunium.net/go/mautrix/bridgev2/bridgeconfig" + "maunium.net/go/mautrix/bridgev2/database" + "maunium.net/go/mautrix/bridgev2/networkid" + "maunium.net/go/mautrix/bridgev2/provisionutil" + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" + + "github.com/beeper/line/pkg/line" +) + +const resolutionMID = "u0123456789abcdef0123456789abcdef" +const resolutionOwnMID = "ufedcba9876543210fedcba9876543210" + +type resolutionTransport func(*http.Request) (*http.Response, error) + +func (f resolutionTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +type resolutionMatrix struct { + bridgev2.MatrixConnector + mu sync.Mutex + created int + failCreate bool +} + +func (*resolutionMatrix) Init(*bridgev2.Bridge) {} +func (m *resolutionMatrix) BotIntent() bridgev2.MatrixAPI { + return &resolutionIntent{mxid: "@bot:example.com", matrix: m} +} +func (m *resolutionMatrix) GhostIntent(userID networkid.UserID) bridgev2.MatrixAPI { + return &resolutionIntent{mxid: id.UserID("@line_" + userID + ":example.com"), matrix: m} +} +func (*resolutionMatrix) GetCapabilities() *bridgev2.MatrixCapabilities { + return &bridgev2.MatrixCapabilities{AutoJoinInvites: true} +} +func (*resolutionMatrix) ParseGhostMXID(id.UserID) (networkid.UserID, bool) { return "", false } +func (*resolutionMatrix) GenerateDeterministicRoomID(networkid.PortalKey) id.RoomID { return "" } +func (*resolutionMatrix) GetMembers(context.Context, id.RoomID) (map[id.UserID]*event.MemberEventContent, error) { + return map[id.UserID]*event.MemberEventContent{"@owner:example.com": {Membership: event.MembershipJoin}}, nil +} + +type resolutionIntent struct { + bridgev2.MatrixAPI + mxid id.UserID + matrix *resolutionMatrix +} + +func (i *resolutionIntent) GetMXID() id.UserID { return i.mxid } +func (*resolutionIntent) IsDoublePuppet() bool { return false } +func (*resolutionIntent) SetDisplayName(context.Context, string) error { return nil } +func (*resolutionIntent) SetAvatarURL(context.Context, id.ContentURIString) error { return nil } +func (*resolutionIntent) SetExtraProfileMeta(context.Context, any) error { return nil } +func (*resolutionIntent) EnsureInvited(context.Context, id.RoomID, id.UserID) error { return nil } +func (*resolutionIntent) EnsureJoined(context.Context, id.RoomID, ...bridgev2.EnsureJoinedParams) error { + return nil +} +func (*resolutionIntent) SendState(context.Context, id.RoomID, event.Type, string, *event.Content, time.Time) (*mautrix.RespSendEvent, error) { + return &mautrix.RespSendEvent{EventID: "$state"}, nil +} +func (i *resolutionIntent) CreateRoom(ctx context.Context, req *mautrix.ReqCreateRoom) (id.RoomID, error) { + i.matrix.mu.Lock() + defer i.matrix.mu.Unlock() + if i.matrix.failCreate { + return "", errors.New("injected Matrix create failure") + } + i.matrix.created++ + return id.RoomID(fmt.Sprintf("!created%d:example.com", i.matrix.created)), nil +} + +func newResolutionClient(t *testing.T, transport resolutionTransport) *LineClient { + t.Helper() + db, err := dbutil.NewWithDialect("file:"+filepath.ToSlash(filepath.Join(t.TempDir(), "bridge.db")), "sqlite3") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + connector := &LineConnector{} + br := bridgev2.NewBridge("test", db, zerolog.New(io.Discard), &bridgeconfig.BridgeConfig{SplitPortals: true}, &resolutionMatrix{}, connector, func(*bridgev2.Bridge) bridgev2.CommandProcessor { return nil }) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + br.BackgroundCtx = ctx + if err = br.DB.Upgrade(ctx); err != nil { + t.Fatal(err) + } + user := &database.User{MXID: "@owner:example.com"} + if err = br.DB.User.Insert(ctx, user); err != nil { + t.Fatal(err) + } + stored := &database.UserLogin{ID: resolutionOwnMID, UserMXID: user.MXID, Metadata: &UserLoginMetadata{AccessToken: "test-token", Mid: resolutionOwnMID}} + if err = br.DB.UserLogin.Insert(ctx, stored); err != nil { + t.Fatal(err) + } + login, err := br.GetExistingUserLoginByID(ctx, stored.ID) + if err != nil { + t.Fatal(err) + } + t.Cleanup(login.BridgeState.Destroy) + lc := login.Client.(*LineClient) + lc.setCachedContact(resolutionOwnMID, line.Contact{Mid: resolutionOwnMID, DisplayName: "Owner"}) + original := newLineAPIClient + newLineAPIClient = func(token string) *line.Client { + client := line.NewClient(token) + client.HTTPClient.Transport = transport + return client + } + t.Cleanup(func() { newLineAPIClient = original }) + return lc +} + +func resolutionResponse(body string) *http.Response { + return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(body))} +} + +func assertPortalCount(t *testing.T, lc *LineClient, want int) { + t.Helper() + portals, err := lc.UserLogin.Bridge.DB.Portal.GetAll(context.Background()) + if err != nil || len(portals) != want { + t.Fatalf("portal count = %d, err=%v, want %d", len(portals), err, want) + } +} + +func TestResolveIdentifierPublicMIDAndSearch(t *testing.T) { + var calls []string + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + calls = append(calls, req.URL.Path) + if !strings.HasSuffix(req.URL.Path, "/findContactByUserid") { + t.Fatalf("unexpected RPC (possible bulk sync or duplicate lookup): %s", req.URL.Path) + } + var args []string + if err := json.NewDecoder(req.Body).Decode(&args); err != nil || len(args) != 1 || args[0] != "carol.id" { + t.Fatalf("public ID arguments = %v, %v", args, err) + } + return resolutionResponse(`{"code":0,"data":{"mid":"` + resolutionMID + `","displayName":"Carol"}}`), nil + }) + ctx := context.Background() + searched, err := lc.SearchUsers(ctx, " CAROL.ID ") + if err != nil || len(searched) != 1 || searched[0].UserID != resolutionMID { + t.Fatalf("search = %+v, %v", searched, err) + } + assertPortalCount(t, lc, 0) + resolved, err := lc.ResolveIdentifier(ctx, "carol.id", true) + if err != nil { + t.Fatal(err) + } + if resolved.UserID != searched[0].UserID || resolved.Chat.PortalKey.ID != resolutionMID || resolved.Chat.PortalKey.Receiver != lc.UserLogin.ID { + t.Fatalf("incorrect canonical portal: %+v", resolved) + } + if resolved.Chat.Portal != nil || resolved.Chat.PortalInfo.Members == nil || *resolved.Chat.PortalInfo.Type != database.RoomTypeDM || *resolved.Chat.PortalInfo.Name != "Carol" { + t.Fatalf("missing prevalidated DM info: %+v", resolved.Chat) + } + assertPortalCount(t, lc, 0) + midResult, err := lc.ResolveIdentifier(ctx, resolutionMID, true) + if err != nil || midResult.Chat.PortalKey != resolved.Chat.PortalKey { + t.Fatalf("MID disagrees with public ID: %+v, %v", midResult, err) + } + lookupOnly, err := lc.ResolveIdentifier(ctx, resolutionMID, false) + if err != nil || lookupOnly.Chat != nil { + t.Fatalf("lookup-only created chat: %+v, %v", lookupOnly, err) + } + if len(calls) != 2 { + t.Fatalf("got %d RPCs, want one per public lookup and no redundant contact fetch", len(calls)) + } +} + +func TestResolveIdentifierMIDLookup(t *testing.T) { + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + if !strings.HasSuffix(req.URL.Path, "/getContactsV2") { + t.Fatalf("MID sent to wrong endpoint: %s", req.URL.Path) + } + return resolutionResponse(`{"code":0,"data":{"contacts":{"` + resolutionMID + `":{"contact":{"mid":"` + resolutionMID + `","displayName":"MID contact"}}}}}`), nil + }) + result, err := lc.ResolveIdentifier(context.Background(), strings.ToUpper(resolutionMID), true) + if err != nil || result == nil || result.Chat.PortalKey.ID != resolutionMID { + t.Fatalf("MID lookup = %+v, %v", result, err) + } +} + +func TestResolveIdentifierFailuresDoNotPersistPortals(t *testing.T) { + for _, tc := range []struct { + name, query, response string + wantErr bool + }{ + {"unknown", "unknown.id", `{"code":0,"data":{}}`, false}, + {"restricted", "restricted.id", `{"code":1,"message":"ID search restricted"}`, true}, + {"invalid response", "carol.id", `{"code":0,"data":{"mid":"carol.id"}}`, true}, + {"group response", "carol.id", `{"code":0,"data":{"mid":"c0123456789abcdef0123456789abcdef"}}`, true}, + {"display name", "Carol Smith", "", true}, + {"empty", " ", "", true}, + } { + t.Run(tc.name, func(t *testing.T) { + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + if tc.response == "" { + t.Fatal("invalid identifier made a network request") + } + return resolutionResponse(tc.response), nil + }) + result, err := lc.ResolveIdentifier(context.Background(), tc.query, true) + if result != nil || (err != nil) != tc.wantErr { + t.Fatalf("result=%+v, err=%v", result, err) + } + assertPortalCount(t, lc, 0) + }) + } +} + +func TestResolveIdentifierCancellation(t *testing.T) { + started := make(chan struct{}) + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + close(started) + <-req.Context().Done() + return nil, req.Context().Err() + }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { _, err := lc.ResolveIdentifier(ctx, "carol.id", true); done <- err }() + select { + case <-started: + case <-time.After(30 * time.Second): + t.Fatal("lookup did not start") + } + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v", err) + } + case <-time.After(time.Second): + t.Fatal("lookup ignored cancellation") + } + assertPortalCount(t, lc, 0) +} + +func TestResolveIdentifierConcurrentCanonicalPortalReuse(t *testing.T) { + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + if !strings.HasSuffix(req.URL.Path, "/findContactByUserid") { + return nil, fmt.Errorf("unexpected RPC: %s", req.URL.Path) + } + return resolutionResponse(`{"code":0,"data":{"mid":"` + resolutionMID + `","displayName":"Carol"}}`), nil + }) + ctx := context.Background() + const count = 12 + portals := make(chan *bridgev2.Portal, count) + errors := make(chan error, count) + var wg sync.WaitGroup + for range count { + wg.Go(func() { + resolved, err := lc.ResolveIdentifier(ctx, "carol.id", true) + if err != nil { + errors <- err + return + } + portal, err := lc.UserLogin.Bridge.GetPortalByKey(ctx, resolved.Chat.PortalKey) + if err != nil { + errors <- err + return + } + portals <- portal + }) + } + wg.Wait() + close(errors) + for err := range errors { + t.Error(err) + } + close(portals) + var first *bridgev2.Portal + for portal := range portals { + if first == nil { + first = portal + } else if first != portal { + t.Fatal("duplicate cached portal") + } + } + assertPortalCount(t, lc, 1) + if first == nil || first.ID != resolutionMID { + t.Fatal("missing canonical portal") + } + // An existing Matrix mapping must survive subsequent resolution. + first.MXID = "!existing:example.com" + if err := first.Save(ctx); err != nil { + t.Fatal(err) + } + resolved, err := lc.ResolveIdentifier(ctx, resolutionMID, true) + if err != nil { + t.Fatal(err) + } + existing, err := lc.UserLogin.Bridge.GetPortalByKey(ctx, resolved.Chat.PortalKey) + if err != nil || existing.MXID != first.MXID { + t.Fatalf("existing portal not reused: %v", err) + } +} + +func TestStartChatCreatesReusesAndRetriesRoom(t *testing.T) { + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + if !strings.HasSuffix(req.URL.Path, "/findContactByUserid") { + return nil, fmt.Errorf("unexpected RPC: %s", req.URL.Path) + } + return resolutionResponse(`{"code":0,"data":{"mid":"` + resolutionMID + `","displayName":"Carol"}}`), nil + }) + matrix := lc.UserLogin.Bridge.Matrix.(*resolutionMatrix) + ctx := context.Background() + matrix.failCreate = true + if _, err := provisionutil.ResolveIdentifier(ctx, lc.UserLogin, "carol.id", true); err == nil { + t.Fatal("injected room creation failure was ignored") + } + // bridgev2 intentionally retains a canonical, retryable portal row after + // Matrix creation fails; no public-ID alias or wrong-recipient row exists. + assertPortalCount(t, lc, 1) + key := networkid.PortalKey{ID: resolutionMID, Receiver: lc.UserLogin.ID} + row, err := lc.UserLogin.Bridge.DB.Portal.GetByKey(ctx, key) + if err != nil || row == nil || row.MXID != "" { + t.Fatalf("failed room state: %+v %v", row, err) + } + matrix.failCreate = false + created, err := provisionutil.ResolveIdentifier(ctx, lc.UserLogin, "carol.id", true) + if err != nil || created == nil || created.DMRoomID == "" { + t.Fatalf("create: %+v %v", created, err) + } + existing, err := provisionutil.ResolveIdentifier(ctx, lc.UserLogin, resolutionMID, true) + if err != nil || existing.DMRoomID != created.DMRoomID || existing.JustCreated { + t.Fatalf("reuse: %+v %v", existing, err) + } + if matrix.created != 1 { + t.Fatalf("created %d rooms", matrix.created) + } + assertPortalCount(t, lc, 1) +} diff --git a/pkg/connector/userinfo.go b/pkg/connector/userinfo.go index ecda636e..869a24f3 100644 --- a/pkg/connector/userinfo.go +++ b/pkg/connector/userinfo.go @@ -193,6 +193,10 @@ func (lc *LineClient) GetChatInfo(ctx context.Context, portal *bridgev2.Portal) } contact := lc.getContact(ctx, string(portal.ID)) + return lc.directChatInfo(mid, contact), nil +} + +func (lc *LineClient) directChatInfo(mid string, contact line.Contact) *bridgev2.ChatInfo { dmType := database.RoomTypeDM chatName := contact.EffectiveDisplayName() return &bridgev2.ChatInfo{ @@ -213,14 +217,14 @@ func (lc *LineClient) GetChatInfo(ctx context.Context, portal *bridgev2.Portal) }, { EventSender: bridgev2.EventSender{ - Sender: networkid.UserID(portal.ID), + Sender: networkid.UserID(mid), }, Membership: event.MembershipJoin, PowerLevel: ptr.Ptr(0), }, }), }, - }, nil + } } func (lc *LineClient) GetUserInfo(ctx context.Context, ghost *bridgev2.Ghost) (*bridgev2.UserInfo, error) { @@ -294,24 +298,30 @@ func (lc *LineClient) setCachedContact(mid string, contact line.Contact) { } func (lc *LineClient) ResolveIdentifier(ctx context.Context, identifier string, createChat bool) (*bridgev2.ResolveIdentifierResponse, error) { - userID := makeUserID(strings.TrimSpace(identifier)) - portalID := networkid.PortalKey{ID: makePortalID(string(userID)), Receiver: lc.UserLogin.ID} + contact, err := lc.resolveContactIdentifier(ctx, identifier) + if err != nil || contact == nil { + return nil, err + } + userID := makeUserID(contact.Mid) ghost, err := lc.UserLogin.Bridge.GetGhostByID(ctx, userID) if err != nil { return nil, fmt.Errorf("failed to get ghost: %w", err) } - portal, err := lc.UserLogin.Bridge.GetPortalByKey(ctx, portalID) - if err != nil { - return nil, fmt.Errorf("failed to get portal: %w", err) - } - ghostInfo, _ := lc.GetUserInfo(ctx, ghost) - portalInfo, _ := lc.GetChatInfo(ctx, portal) - return &bridgev2.ResolveIdentifierResponse{ + name := contact.EffectiveDisplayName() + resp := &bridgev2.ResolveIdentifierResponse{ Ghost: ghost, UserID: userID, - UserInfo: ghostInfo, - Chat: &bridgev2.CreateChatResponse{Portal: portal, PortalKey: portalID, PortalInfo: portalInfo}, - }, nil + UserInfo: &bridgev2.UserInfo{Identifiers: []string{contact.Mid}, Name: &name, Avatar: lc.avatarFromPicturePath(contact.PicturePath)}, + } + if createChat { + // Resolve and validate before bridgev2 persists anything. Its canonical + // portal cache and room-creation event loop handle reuse and concurrency. + resp.Chat = &bridgev2.CreateChatResponse{ + PortalKey: networkid.PortalKey{ID: makePortalID(contact.Mid), Receiver: lc.UserLogin.ID}, + PortalInfo: lc.directChatInfo(contact.Mid, *contact), + } + } + return resp, nil } func (lc *LineClient) midToResolveIdentifier(ctx context.Context, mid string) *bridgev2.ResolveIdentifierResponse { @@ -334,27 +344,29 @@ func (lc *LineClient) midToResolveIdentifier(ctx context.Context, mid string) *b func (lc *LineClient) SearchUsers(ctx context.Context, query string) ([]*bridgev2.ResolveIdentifierResponse, error) { var results []*bridgev2.ResolveIdentifierResponse - // Try by LINE user ID first + // Exact public ID and MID lookup is shared with start-chat. Only search + // falls back to non-unique display-name matches. lowerQuery := strings.ToLower(strings.TrimSpace(query)) if lowerQuery != "" { - _, contact, err := callLineResult(lc, ctx, func(client *line.Client) (*line.Contact, error) { - return client.FindContactByUserid(lowerQuery) - }) - if err == nil && contact != nil && contact.Mid != "" { + contact, err := lc.resolveContactIdentifier(ctx, query) + if ctx.Err() != nil { + return nil, ctx.Err() + } + if err == nil && contact != nil { if r := lc.midToResolveIdentifier(ctx, contact.Mid); r != nil { - results = append(results, r) - if len(results) >= 20 { - return results, nil - } + return []*bridgev2.ResolveIdentifierResponse{r}, nil } } } // Search contacts by display name client, allMids, err := callLineResult(lc, ctx, func(client *line.Client) ([]string, error) { - return client.GetAllContactIds() + return client.GetAllContactIdsContext(ctx) }) if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to get all contact IDs for search") return results, nil } @@ -366,8 +378,11 @@ func (lc *LineClient) SearchUsers(ctx context.Context, query string) ([]*bridgev end = len(allMids) } batch := allMids[i:end] - contactsResp, err := client.GetContactsV2(batch) + contactsResp, err := client.GetContactsV2Context(ctx, batch) if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } continue } for _, wrapper := range contactsResp.Contacts { @@ -377,6 +392,7 @@ func (lc *LineClient) SearchUsers(ctx context.Context, query string) ([]*bridgev } name := strings.ToLower(c.EffectiveDisplayName()) if lowerQuery == "" || strings.Contains(name, lowerQuery) { + lc.setCachedContact(c.Mid, c) if r := lc.midToResolveIdentifier(ctx, c.Mid); r != nil { results = append(results, r) } diff --git a/pkg/line/methods.go b/pkg/line/methods.go index 08a0ca04..ac67286f 100644 --- a/pkg/line/methods.go +++ b/pkg/line/methods.go @@ -487,8 +487,12 @@ func (c *Client) SendChatChecked(chatMid, messageID string) error { // GetContactsV2 fetches contact details for a list of MIDs. func (c *Client) GetContactsV2(mids []string) (*ContactsResponse, error) { + return c.GetContactsV2Context(c.requestContext(), mids) +} + +func (c *Client) GetContactsV2Context(ctx context.Context, mids []string) (*ContactsResponse, error) { req := GetContactsV2Request{TargetUserMids: mids} - resp, err := c.callRPC("TalkService", "getContactsV2", req, 2) + resp, err := c.callRPCContext(ctx, "TalkService", "getContactsV2", req, 2) if err != nil { return nil, err } @@ -508,7 +512,11 @@ func (c *Client) GetContactsV2(mids []string) (*ContactsResponse, error) { // GetBuddyProfile fetches the profile of a LINE official/business account (buddy). func (c *Client) GetBuddyProfile(mid string) (*BuddyProfile, error) { - resp, err := c.callRPC("BuddyService", "getBuddyProfile", mid) + return c.GetBuddyProfileContext(c.requestContext(), mid) +} + +func (c *Client) GetBuddyProfileContext(ctx context.Context, mid string) (*BuddyProfile, error) { + resp, err := c.callRPCContext(ctx, "BuddyService", "getBuddyProfile", mid) if err != nil { return nil, err } @@ -1039,7 +1047,11 @@ func (c *Client) RejectChatInvitationContext(ctx context.Context, reqSeq int64, // FindContactByUserid looks up a LINE user by their user ID (not MID). func (c *Client) FindContactByUserid(userid string) (*Contact, error) { - resp, err := c.callRPC("TalkService", "findContactByUserid", userid) + return c.FindContactByUseridContext(c.requestContext(), userid) +} + +func (c *Client) FindContactByUseridContext(ctx context.Context, userid string) (*Contact, error) { + resp, err := c.callRPCContext(ctx, "TalkService", "findContactByUserid", userid) if err != nil { return nil, err } @@ -1059,7 +1071,11 @@ func (c *Client) FindContactByUserid(userid string) (*Contact, error) { // GetAllContactIds returns the MIDs of all contacts. func (c *Client) GetAllContactIds() ([]string, error) { - resp, err := c.callRPC("TalkService", "getAllContactIds") + return c.GetAllContactIdsContext(c.requestContext()) +} + +func (c *Client) GetAllContactIdsContext(ctx context.Context) ([]string, error) { + resp, err := c.callRPCContext(ctx, "TalkService", "getAllContactIds") if err != nil { return nil, err } diff --git a/pkg/line/methods_context_test.go b/pkg/line/methods_context_test.go index ff591417..65f10dff 100644 --- a/pkg/line/methods_context_test.go +++ b/pkg/line/methods_context_test.go @@ -71,3 +71,23 @@ func TestGetProfileContextCancellation(t *testing.T) { return err }) } + +func TestIdentifierLookupContextCancellation(t *testing.T) { + for name, call := range map[string]func(*Client, context.Context) error{ + "findContactByUserid": func(c *Client, ctx context.Context) error { + _, err := c.FindContactByUseridContext(ctx, "carol") + return err + }, + "getContactsV2": func(c *Client, ctx context.Context) error { + _, err := c.GetContactsV2Context(ctx, []string{"u0123456789abcdef0123456789abcdef"}) + return err + }, + "getBuddyProfile": func(c *Client, ctx context.Context) error { + _, err := c.GetBuddyProfileContext(ctx, "u0123456789abcdef0123456789abcdef") + return err + }, + "getAllContactIds": func(c *Client, ctx context.Context) error { _, err := c.GetAllContactIdsContext(ctx); return err }, + } { + t.Run(name, func(t *testing.T) { testRPCContextCancellation(t, name, call) }) + } +} From 5802c3fa806d9453f10c2eb99fc425ecfd0f2b96 Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 17:08:01 -0400 Subject: [PATCH 5/9] fix: serialize QR login operations and preserve deadline errors Reject overlapping Start/Wait calls before they consume polling stages or mutate authentication state. Preserve caller and attempt expiry instead of reporting every canceled start as context.Canceled. Reproduce both defects and cover mocked QR presentation, scan/PIN approval, token verification, missing-keychain persistence rejection, expiry and cancellation. --- pkg/connector/login_qr.go | 48 ++++++-- pkg/connector/login_qr_test.go | 193 +++++++++++++++++++++++++++++++++ 2 files changed, 231 insertions(+), 10 deletions(-) create mode 100644 pkg/connector/login_qr_test.go diff --git a/pkg/connector/login_qr.go b/pkg/connector/login_qr.go index ac998ed9..a608947d 100644 --- a/pkg/connector/login_qr.go +++ b/pkg/connector/login_qr.go @@ -3,6 +3,7 @@ package connector import ( "context" "fmt" + "sync" "time" "maunium.net/go/mautrix/bridgev2" @@ -11,18 +12,23 @@ import ( ) type LineQRLogin struct { - login *LineEmailLogin - client *line.Client - session string - qr *line.QRCodeResponse - poll chan error - pin bool + operationMu sync.Mutex + login *LineEmailLogin + client *line.Client + session string + qr *line.QRCodeResponse + poll chan error + pin bool } var _ bridgev2.LoginProcessDisplayAndWait = (*LineQRLogin)(nil) var _ bridgev2.LoginProcessWithOverride = (*LineQRLogin)(nil) func (lq *LineQRLogin) StartWithOverride(ctx context.Context, override *bridgev2.UserLogin) (*bridgev2.LoginStep, error) { + if !lq.operationMu.TryLock() { + return nil, fmt.Errorf("QR login operation already in progress") + } + defer lq.operationMu.Unlock() meta, ok := override.Metadata.(*UserLoginMetadata) if !ok { return nil, fmt.Errorf("existing LINE login metadata has unexpected type %T", override.Metadata) @@ -32,10 +38,21 @@ func (lq *LineQRLogin) StartWithOverride(ctx context.Context, override *bridgev2 if meta.ForceFullE2EELogin || len(meta.ExportedKeyMap) == 0 { lq.login.Certificate = "" } - return lq.Start(ctx) + return lq.start(ctx) } func (lq *LineQRLogin) Start(ctx context.Context) (*bridgev2.LoginStep, error) { + if !lq.operationMu.TryLock() { + return nil, fmt.Errorf("QR login operation already in progress") + } + defer lq.operationMu.Unlock() + return lq.start(ctx) +} + +func (lq *LineQRLogin) start(ctx context.Context) (*bridgev2.LoginStep, error) { + if err := ctx.Err(); err != nil { + return nil, err + } ll := lq.login ll.mu.Lock() if ll.canceled || ll.attemptCtx != nil { @@ -54,10 +71,15 @@ func (lq *LineQRLogin) Start(ctx context.Context) (*bridgev2.LoginStep, error) { canceled := ll.canceled || ctx.Err() != nil || processCtx.Err() != nil ll.mu.Unlock() if err != nil || canceled { - lq.Cancel() - if canceled { - return nil, context.Canceled + // Preserve expiry before Cancel changes the attempt's context error. + if ctx.Err() != nil { + err = ctx.Err() + } else if processCtx.Err() != nil { + err = processCtx.Err() + } else if canceled { + err = context.Canceled } + lq.Cancel() return nil, err } lq.client, lq.session, lq.qr = client, session, qr @@ -100,6 +122,12 @@ func (lq *LineQRLogin) startPoll(check func(context.Context, string) error, retr } func (lq *LineQRLogin) Wait(ctx context.Context) (*bridgev2.LoginStep, error) { + // Overlapping provisioning calls must not consume each other's polling + // results or advance the PIN/token exchange twice. Cancel doesn't take this lock. + if !lq.operationMu.TryLock() { + return nil, fmt.Errorf("QR login operation already in progress") + } + defer lq.operationMu.Unlock() if lq.poll == nil { return nil, fmt.Errorf("QR login has not started") } diff --git a/pkg/connector/login_qr_test.go b/pkg/connector/login_qr_test.go new file mode 100644 index 00000000..f35be5b5 --- /dev/null +++ b/pkg/connector/login_qr_test.go @@ -0,0 +1,193 @@ +package connector + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "net/http" + "net/url" + "strings" + "sync" + "testing" + "time" + + "maunium.net/go/mautrix/bridgev2" +) + +func TestQRConcurrentWaitIsRejected(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + lq := &LineQRLogin{login: &LineEmailLogin{attemptCtx: ctx, attemptCancel: cancel}, poll: make(chan error, 1)} + results := make(chan error, 2) + for range 2 { + go func() { _, err := lq.Wait(context.Background()); results <- err }() + } + select { + case err := <-results: + if err == nil || !strings.Contains(err.Error(), "already in progress") { + t.Errorf("unexpected concurrent Wait result: %v", err) + } + case <-time.After(time.Second): + t.Error("concurrent Wait calls both blocked instead of rejecting the duplicate") + cancel() + <-results + } + cancel() + select { + case err := <-results: + if !errors.Is(err, context.Canceled) { + t.Errorf("active Wait ignored cancellation: %v", err) + } + case <-time.After(time.Second): + t.Fatal("active Wait did not stop") + } +} + +func TestQRApprovalPINAndTokenValidation(t *testing.T) { + approved, pinApproved := make(chan struct{}), make(chan struct{}) + var mu sync.Mutex + var calls []string + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + method := req.URL.Path[strings.LastIndex(req.URL.Path, "/")+1:] + mu.Lock() + calls = append(calls, method) + mu.Unlock() + switch method { + case "createSession": + return resolutionResponse(`{"code":0,"data":{"authSessionId":"qr-session"}}`), nil + case "createQrCode": + return resolutionResponse(`{"code":0,"data":{"callbackUrl":"line://au/q/test","longPollingMaxCount":1}}`), nil + case "checkQrCodeVerified", "checkPinCodeVerified": + if req.Header.Get("X-Line-Session-ID") != "qr-session" || req.Header.Get("x-hmac") == "" { + return nil, errors.New("missing QR authentication headers") + } + gate := approved + if method == "checkPinCodeVerified" { + gate = pinApproved + } + select { + case <-gate: + case <-req.Context().Done(): + return nil, req.Context().Err() + } + return resolutionResponse(`{"code":0,"data":{}}`), nil + case "verifyCertificate": + resp := resolutionResponse(`{"code":1,"data":{}}`) + resp.StatusCode = http.StatusBadRequest + return resp, nil + case "createPinCode": + return resolutionResponse(`{"code":0,"data":{"pinCode":"123456"}}`), nil + case "qrCodeLoginV2": + return resolutionResponse(`{"code":0,"data":{"mid":"` + resolutionOwnMID + `","tokenV3IssueResult":{"accessToken":"qr-token","refreshToken":"qr-refresh"},"metaData":{}}}`), nil + case "getProfile": + if req.Header.Get("x-line-access") != "qr-token" { + return nil, errors.New("QR token was not used for verification") + } + return resolutionResponse(`{"code":0,"data":{"mid":"` + resolutionOwnMID + `","displayName":"Owner"}}`), nil + default: + return nil, fmt.Errorf("unexpected QR RPC: %s", method) + } + }) + lq := &LineQRLogin{login: &LineEmailLogin{User: lc.UserLogin.User}} + defer lq.Cancel() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + step, err := lq.Start(ctx) + if err != nil || step.DisplayAndWaitParams.Type != bridgev2.LoginDisplayTypeQR { + t.Fatalf("QR start: %+v %v", step, err) + } + callback, err := url.Parse(step.DisplayAndWaitParams.Data) + if err != nil { + t.Fatal(err) + } + key, err := base64.StdEncoding.DecodeString(callback.Query().Get("secret")) + if err != nil || len(key) != 32 || callback.Query().Get("e2eeVersion") != "1" { + t.Fatal("QR callback missing login key material") + } + close(approved) + step, err = lq.Wait(ctx) + if err != nil || step.DisplayAndWaitParams.Type != bridgev2.LoginDisplayTypeCode || step.DisplayAndWaitParams.Data != "123456" { + t.Fatalf("PIN step: %+v %v", step, err) + } + close(pinApproved) + step, err = lq.Wait(ctx) + if step != nil || !errors.Is(err, ErrLoginNoKeychain) { + t.Fatalf("missing E2EE keychain was accepted: %+v %v", step, err) + } + stored, err := lc.UserLogin.Bridge.DB.UserLogin.GetByID(ctx, lc.UserLogin.ID) + if err != nil || stored.Metadata.(*UserLoginMetadata).AccessToken != "test-token" { + t.Fatalf("incomplete QR login changed stored session: %v", err) + } + mu.Lock() + defer mu.Unlock() + want := "[createSession createQrCode checkQrCodeVerified verifyCertificate createPinCode checkPinCodeVerified qrCodeLoginV2 getProfile]" + if fmt.Sprint(calls) != want { + t.Fatalf("unexpected QR state transitions: %v", calls) + } +} + +func TestQRExpiryAndCancellation(t *testing.T) { + for _, expire := range []bool{false, true} { + t.Run(fmt.Sprint(expire), func(t *testing.T) { + pollStarted, pollStopped := make(chan struct{}), make(chan struct{}) + newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + switch { + case strings.HasSuffix(req.URL.Path, "/createSession"): + return resolutionResponse(`{"code":0,"data":{"authSessionId":"session"}}`), nil + case strings.HasSuffix(req.URL.Path, "/createQrCode"): + return resolutionResponse(`{"code":0,"data":{"callbackUrl":"line://au/q/test","longPollingMaxCount":1}}`), nil + case strings.HasSuffix(req.URL.Path, "/checkQrCodeVerified"): + close(pollStarted) + defer close(pollStopped) + if expire { + return resolutionResponse(`{"code":10052,"data":{"statusCode":410}}`), nil + } + <-req.Context().Done() + return nil, req.Context().Err() + default: + return nil, fmt.Errorf("approval was bypassed: %s", req.URL.Path) + } + }) + lq := &LineQRLogin{login: &LineEmailLogin{}} + defer lq.Cancel() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if _, err := lq.Start(ctx); err != nil { + t.Fatal(err) + } + select { + case <-pollStarted: + case <-ctx.Done(): + t.Fatal("poll did not start") + } + if !expire { + lq.Cancel() + } + step, err := lq.Wait(ctx) + want := error(context.Canceled) + if expire { + want = ErrLoginVerificationFailed + } + if step != nil || !errors.Is(err, want) { + t.Fatalf("unexpected terminal state: %+v %v", step, err) + } + select { + case <-pollStopped: + case <-ctx.Done(): + t.Fatal("poll worker leaked") + } + }) + } +} + +func TestQRStartPreservesDeadline(t *testing.T) { + ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer cancel() + lq := &LineQRLogin{login: &LineEmailLogin{}} + newResolutionClient(t, func(*http.Request) (*http.Response, error) { return nil, ctx.Err() }) + step, err := lq.Start(ctx) + if step != nil || !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("start after deadline = %+v, %v; want DeadlineExceeded", step, err) + } +} From 0597689427d74415fccdd17e36044555736a8032 Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 17:11:02 -0400 Subject: [PATCH 6/9] fix: preserve outbound cancellation and LINE send errors Pass the Matrix request context to sendMessage, preserve structured TalkException details for existing recovery logic, and reject missing acknowledgements instead of returning success or dereferencing nil. Add correlated send-stage logs without message bodies and regressions for ownership, canonical recipients, API failures and cancellation. These defects are reproduced locally; the reported live Element delivery failure still needs event-level evidence. --- pkg/connector/outbound_pipeline_test.go | 119 ++++++++++++++++++++++++ pkg/connector/send_message.go | 15 ++- pkg/line/methods.go | 13 ++- pkg/line/send_message_response_test.go | 34 +++++++ 4 files changed, 178 insertions(+), 3 deletions(-) create mode 100644 pkg/connector/outbound_pipeline_test.go create mode 100644 pkg/line/send_message_response_test.go diff --git a/pkg/connector/outbound_pipeline_test.go b/pkg/connector/outbound_pipeline_test.go new file mode 100644 index 00000000..2970f5fc --- /dev/null +++ b/pkg/connector/outbound_pipeline_test.go @@ -0,0 +1,119 @@ +package connector + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "testing" + "time" + + "maunium.net/go/mautrix/bridgev2" + "maunium.net/go/mautrix/bridgev2/database" + "maunium.net/go/mautrix/bridgev2/networkid" + "maunium.net/go/mautrix/event" + + "github.com/beeper/line/pkg/line" +) + +func outboundTestMessage(t *testing.T, lc *LineClient) *bridgev2.MatrixMessage { + t.Helper() + portal, err := lc.UserLogin.Bridge.GetPortalByKey(context.Background(), networkid.PortalKey{ID: resolutionMID, Receiver: lc.UserLogin.ID}) + if err != nil { + t.Fatal(err) + } + return &bridgev2.MatrixMessage{MatrixEventBase: bridgev2.MatrixEventBase[*event.MessageEventContent]{ + Portal: portal, + Event: &event.Event{ID: "$test-outbound", Sender: lc.UserLogin.UserMXID, RoomID: "!test:example.com"}, + Content: &event.MessageEventContent{MsgType: event.MsgText, Body: "test message"}, + }} +} + +func TestOutboundOwnerAndCanonicalRecipient(t *testing.T) { + for _, fail := range []bool{false, true} { + t.Run(fmt.Sprint(fail), func(t *testing.T) { + requests := 0 + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + requests++ + if !strings.HasSuffix(req.URL.Path, "/sendMessage") { + return nil, fmt.Errorf("unexpected endpoint: %s", req.URL.Path) + } + var args []json.RawMessage + if err := json.NewDecoder(req.Body).Decode(&args); err != nil || len(args) != 2 { + t.Fatalf("invalid send args: %v %v", args, err) + } + var sent line.Message + if err := json.Unmarshal(args[1], &sent); err != nil { + t.Fatal(err) + } + if sent.To != resolutionMID || sent.From != resolutionOwnMID || sent.Text != "test message" { + t.Fatalf("misrouted/incorrect message: %+v", sent) + } + if fail { + return resolutionResponse(`{"code":42,"message":"send rejected"}`), nil + } + return resolutionResponse(`{"code":0,"data":{"id":"line-message-id"}}`), nil + }) + msg := outboundTestMessage(t, lc) + login, _, err := msg.Portal.FindPreferredLogin(context.Background(), lc.UserLogin.User, true) + if err != nil || login != lc.UserLogin { + t.Fatalf("owner login association: %v", err) + } + other := &bridgev2.User{User: &database.User{MXID: "@other:example.com"}, Bridge: lc.UserLogin.Bridge} + if _, _, err = msg.Portal.FindPreferredLogin(context.Background(), other, true); !errors.Is(err, bridgev2.ErrNotLoggedIn) { + t.Fatalf("unrelated sender was allowed to use owner's login: %v", err) + } + resp, err := lc.HandleMatrixMessage(context.Background(), msg) + if fail { + if err == nil || resp != nil { + t.Fatalf("rejected send reported success: %+v %v", resp, err) + } + } else if err != nil || resp == nil || resp.DB.ID != "line-message-id" { + t.Fatalf("successful send lost LINE acknowledgment: %+v %v", resp, err) + } + if requests != 1 { + t.Fatalf("send RPCs=%d", requests) + } + }) + } +} + +func TestOutboundRequestCancellation(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + close(started) + select { + case <-req.Context().Done(): + return nil, req.Context().Err() + case <-release: + return nil, context.Canceled + } + }) + msg := outboundTestMessage(t, lc) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { _, err := lc.HandleMatrixMessage(ctx, msg); done <- err }() + select { + case <-started: + case <-time.After(30 * time.Second): + cancel() + close(release) + <-done + t.Fatal("send did not start") + } + cancel() + select { + case err := <-done: + close(release) + if !errors.Is(err, context.Canceled) { + t.Fatalf("send returned %v", err) + } + case <-time.After(time.Second): + close(release) + <-done + t.Fatal("LINE send did not inherit Matrix request cancellation") + } +} diff --git a/pkg/connector/send_message.go b/pkg/connector/send_message.go index 9632ad22..f29912f4 100644 --- a/pkg/connector/send_message.go +++ b/pkg/connector/send_message.go @@ -63,6 +63,14 @@ func lineGroupE2EEFetchFailureError(err error) error { } func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.MatrixMessage) (*bridgev2.MatrixMessageResponse, error) { + logContext := lc.UserLogin.Bridge.Log.With(). + Str("line_login_id", string(lc.UserLogin.ID)). + Str("line_recipient", string(msg.Portal.ID)) + if msg.Event != nil { + logContext = logContext.Stringer("matrix_event_id", msg.Event.ID).Stringer("matrix_sender", msg.Event.Sender) + } + log := logContext.Logger() + log.Debug().Msg("Handling outbound Matrix message for LINE") if err := lc.checkLineCall(ctx); err != nil { return nil, err } @@ -82,8 +90,13 @@ func (lc *LineClient) HandleMatrixMessage(ctx context.Context, msg *bridgev2.Mat var err error var sentMsg *line.Message client, sentMsg, err = callLineResultUsing(lc, ctx, client, func(client *line.Client) (*line.Message, error) { - return client.SendMessage(int64(reqSeq), lineMsg) + return client.SendMessageContext(ctx, int64(reqSeq), lineMsg) }) + resultLog := log.Debug().Bool("success", err == nil) + if sentMsg != nil { + resultLog = resultLog.Str("line_message_id", sentMsg.ID) + } + resultLog.Msg("Finished LINE API send attempt") return sentMsg, err } portalMid := string(msg.Portal.ID) diff --git a/pkg/line/methods.go b/pkg/line/methods.go index ac67286f..39686a73 100644 --- a/pkg/line/methods.go +++ b/pkg/line/methods.go @@ -391,7 +391,11 @@ func (c *Client) GetE2EEPublicKey(mid string, keyVersion, keyID int) (*E2EEPubli } func (c *Client) SendMessage(reqSeq int64, msg *Message) (*Message, error) { - resp, err := c.callRPC("TalkService", "sendMessage", reqSeq, msg) + return c.SendMessageContext(c.requestContext(), reqSeq, msg) +} + +func (c *Client) SendMessageContext(ctx context.Context, reqSeq int64, msg *Message) (*Message, error) { + resp, err := c.callRPCContext(ctx, "TalkService", "sendMessage", reqSeq, msg) if err != nil { return nil, err } @@ -404,7 +408,12 @@ func (c *Client) SendMessage(reqSeq int64, msg *Message) (*Message, error) { return nil, err } if wrapper.Code != 0 { - return nil, fmt.Errorf("sendMessage failed: %s", wrapper.Message) + // Keep TalkException codes available to auth recovery and send retries, + // including when LINE wraps an error in an HTTP 200 response. + return nil, fmt.Errorf("sendMessage failed: %s", resp) + } + if wrapper.Data == nil || wrapper.Data.ID == "" { + return nil, fmt.Errorf("sendMessage returned no message acknowledgement") } return wrapper.Data, nil } diff --git a/pkg/line/send_message_response_test.go b/pkg/line/send_message_response_test.go new file mode 100644 index 00000000..fff1be9b --- /dev/null +++ b/pkg/line/send_message_response_test.go @@ -0,0 +1,34 @@ +package line + +import ( + "io" + "net/http" + "strings" + "testing" +) + +func TestSendMessageRejectsMissingAcknowledgement(t *testing.T) { + for _, body := range []string{`{"code":0}`, `{"code":0,"data":null}`, `{"code":0,"data":{}}`} { + t.Run(body, func(t *testing.T) { + client := NewClient("test-token") + client.HTTPClient.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(body))}, nil + }) + msg, err := client.SendMessage(1, &Message{To: "u0123456789abcdef0123456789abcdef", Text: "test"}) + if err == nil || msg != nil { + t.Fatalf("missing acknowledgement accepted: %+v, %v", msg, err) + } + }) + } +} + +func TestSendMessagePreservesAuthenticationError(t *testing.T) { + client := NewClient("test-token") + client.HTTPClient.Transport = roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: 200, Body: io.NopCloser(strings.NewReader(`{"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":119,"reason":"Access token refresh required"}}`))}, nil + }) + _, err := client.SendMessage(1, &Message{To: "u0123456789abcdef0123456789abcdef", Text: "test"}) + if !IsAuthError(err) { + t.Fatalf("send error lost auth classification: %v", err) + } +} From b3fb1acde9a5436f1a42d7a867f747d3640f5c2a Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 17:13:49 -0400 Subject: [PATCH 7/9] test: cover DM reopening and passwordless QR reconnects Verify public-ID reopening after leaving a Matrix room, rejection of mismatched MIDs and expired lookup requests, and mandatory fresh approval for passwordless QR sessions after token recovery is unavailable. --- pkg/connector/login_qr_test.go | 10 +++++ pkg/connector/resolve_identifier_test.go | 47 +++++++++++++++++++++++- 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/pkg/connector/login_qr_test.go b/pkg/connector/login_qr_test.go index f35be5b5..ec146999 100644 --- a/pkg/connector/login_qr_test.go +++ b/pkg/connector/login_qr_test.go @@ -191,3 +191,13 @@ func TestQRStartPreservesDeadline(t *testing.T) { t.Fatalf("start after deadline = %+v, %v; want DeadlineExceeded", step, err) } } + +func TestQRPasswordlessReconnectNeedsApproval(t *testing.T) { + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + return nil, fmt.Errorf("unexpected passwordless authentication request: %s", req.URL.Path) + }) + lc.UserLogin.Bridge.Network.(*LineConnector).Config.QRLogin = true + if err := lc.tryLogin(context.Background()); !errors.Is(err, errLineQRLoginRequired) { + t.Fatalf("passwordless QR session must request fresh approval: %v", err) + } +} diff --git a/pkg/connector/resolve_identifier_test.go b/pkg/connector/resolve_identifier_test.go index b653cf05..1d2c5c1b 100644 --- a/pkg/connector/resolve_identifier_test.go +++ b/pkg/connector/resolve_identifier_test.go @@ -39,6 +39,8 @@ type resolutionMatrix struct { mu sync.Mutex created int failCreate bool + ownerLeft bool + deleted int } func (*resolutionMatrix) Init(*bridgev2.Bridge) {} @@ -53,8 +55,14 @@ func (*resolutionMatrix) GetCapabilities() *bridgev2.MatrixCapabilities { } func (*resolutionMatrix) ParseGhostMXID(id.UserID) (networkid.UserID, bool) { return "", false } func (*resolutionMatrix) GenerateDeterministicRoomID(networkid.PortalKey) id.RoomID { return "" } -func (*resolutionMatrix) GetMembers(context.Context, id.RoomID) (map[id.UserID]*event.MemberEventContent, error) { - return map[id.UserID]*event.MemberEventContent{"@owner:example.com": {Membership: event.MembershipJoin}}, nil +func (m *resolutionMatrix) GetMembers(context.Context, id.RoomID) (map[id.UserID]*event.MemberEventContent, error) { + m.mu.Lock() + defer m.mu.Unlock() + membership := event.MembershipJoin + if m.ownerLeft { + membership = event.MembershipLeave + } + return map[id.UserID]*event.MemberEventContent{"@owner:example.com": {Membership: membership}}, nil } type resolutionIntent struct { @@ -85,6 +93,13 @@ func (i *resolutionIntent) CreateRoom(ctx context.Context, req *mautrix.ReqCreat return id.RoomID(fmt.Sprintf("!created%d:example.com", i.matrix.created)), nil } +func (i *resolutionIntent) DeleteRoom(context.Context, id.RoomID, bool) error { + i.matrix.mu.Lock() + defer i.matrix.mu.Unlock() + i.matrix.deleted++ + return nil +} + func newResolutionClient(t *testing.T, transport resolutionTransport) *LineClient { t.Helper() db, err := dbutil.NewWithDialect("file:"+filepath.ToSlash(filepath.Join(t.TempDir(), "bridge.db")), "sqlite3") @@ -249,6 +264,24 @@ func TestResolveIdentifierCancellation(t *testing.T) { assertPortalCount(t, lc, 0) } +func TestResolveIdentifierDeadlineAndMismatchedMID(t *testing.T) { + lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { + if !strings.HasSuffix(req.URL.Path, "/getContactsV2") { + return nil, fmt.Errorf("unexpected endpoint: %s", req.URL.Path) + } + return resolutionResponse(`{"code":0,"data":{"contacts":{"` + resolutionMID + `":{"contact":{"mid":"u22222222222222222222222222222222","displayName":"Wrong recipient"}}}}}`), nil + }) + ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer cancel() + if resp, err := lc.ResolveIdentifier(ctx, resolutionMID, true); resp != nil || !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expired lookup: %+v %v", resp, err) + } + if resp, err := lc.ResolveIdentifier(context.Background(), resolutionMID, true); resp != nil || err == nil { + t.Fatalf("mismatched MID accepted: %+v %v", resp, err) + } + assertPortalCount(t, lc, 0) +} + func TestResolveIdentifierConcurrentCanonicalPortalReuse(t *testing.T) { lc := newResolutionClient(t, func(req *http.Request) (*http.Response, error) { if !strings.HasSuffix(req.URL.Path, "/findContactByUserid") { @@ -343,4 +376,14 @@ func TestStartChatCreatesReusesAndRetriesRoom(t *testing.T) { t.Fatalf("created %d rooms", matrix.created) } assertPortalCount(t, lc, 1) + oldRoomID := created.DMRoomID + matrix.ownerLeft = true + reopened, err := provisionutil.ResolveIdentifier(ctx, lc.UserLogin, "carol.id", true) + if err != nil || reopened.DMRoomID == oldRoomID || reopened.DMRoomID == "" { + t.Fatalf("reopen after leaving: %+v %v", reopened, err) + } + if matrix.created != 2 || matrix.deleted != 1 { + t.Fatalf("create/delete counts = %d/%d", matrix.created, matrix.deleted) + } + assertPortalCount(t, lc, 1) } From 6deb64eb0c543d8e485aa54ac08331960aa9ff8a Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 17:15:20 -0400 Subject: [PATCH 8/9] docs: trace chat sync and outbound authentication diagnostics Label existing connect, LINE fullSync and post-backfill synchronization triggers without changing policy. Document canonical DM resolution, retained retryable portal rows, outbound stage/account diagnostics, independent QR lifecycle findings and live-verification boundaries. --- pkg/connector/client.go | 1 + pkg/connector/sync.go | 3 +- readme/DEBUGGING.md | 152 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 155 insertions(+), 1 deletion(-) create mode 100644 readme/DEBUGGING.md diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 97113f3d..7a768f06 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -690,6 +690,7 @@ func (lc *LineClient) Connect(ctx context.Context) { // deliver messages. Otherwise an existing LINE group whose Matrix room // doesn't exist yet may be created by the first message, which makes the // sender's existing membership look like a fresh join. + lc.UserLogin.Bridge.Log.Info().Str("sync_trigger", "connect").Msg("Starting LINE conversation synchronization") lc.syncChatsNow(ctx) if ctx.Err() != nil { return diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index ad50eca7..04805b62 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -768,6 +768,7 @@ func (lc *LineClient) prefetchMessages(ctx context.Context) { // a hidden resync so a truncated recent-message window cannot leave a room // at an old membership or name. if backfilledSystemEvents.Load() && ctx.Err() == nil { + lc.UserLogin.Bridge.Log.Debug().Str("sync_trigger", "backfill_state_restore").Msg("Starting LINE conversation synchronization") lc.syncChatsNow(ctx) } } @@ -1646,7 +1647,7 @@ func (lc *LineClient) pollLoop(ctx context.Context) { // handle fullsync requests if eventType == "fullSync" { - lc.UserLogin.Bridge.Log.Info().Msg("Received fullSync request") + lc.UserLogin.Bridge.Log.Info().Str("sync_trigger", "line_full_sync").Msg("Received fullSync request") var fsPayload struct { NextRevision string `json:"nextRevision"` diff --git a/readme/DEBUGGING.md b/readme/DEBUGGING.md new file mode 100644 index 00000000..12089944 --- /dev/null +++ b/readme/DEBUGGING.md @@ -0,0 +1,152 @@ +# Direct chats, synchronization, outbound messages and QR login + +## Starting a direct chat + +In the bot management room: + +```text +search carol.id +start-chat carol.id +start-chat u0123456789abcdef0123456789abcdef +``` + +These are illustrative identifiers, not live test accounts. In another Matrix +room, use the configured command prefix (default `!line`), for example +`!line start-chat carol.id`. With several LINE logins, the framework also accepts +`start-chat `. + +Public LINE IDs use the existing `TalkService.findContactByUserid` RPC. The +bridge validates the returned user MID before creating a ghost or portal. User +MIDs are `u` followed by 32 hexadecimal digits; surrounding whitespace and input +case are normalized. A direct MID uses cached contact data or `getContactsV2`, +with the existing BuddyService fallback for official accounts. The bridge never +chooses a recipient by display name. `search` can still return display-name +matches when there is no exact ID match, but `start-chat` cannot use that fallback. + +Both commands share exact identifier resolution. A public-ID result already +contains the contact profile: it is reused for ghost/DM information rather than +re-fetching the same contact. A public ID is resolved afresh on each request so a +reassigned public ID cannot use a stale ID-to-MID alias cache. + +The former implementation treated every argument as a MID, created a portal +record immediately, and ignored the error returned by `GetChatInfo`. An argument +beginning with `c` was then routed to `getChats`; an empty result produced +`chat not found`. Framework room creation retried this missing chat-info lookup +and reported `Failed to create portal room`. + +The corrected response contains the canonical MID, login-scoped portal key and +complete DM information. bridgev2 retains responsibility for its locked portal +cache, database identity, room-creation event loop, membership checks and reuse. +Lookup-only requests do not create portal records. Failed remote lookups create +neither invalid portal records nor public-ID aliases. If Matrix room creation +fails later, bridgev2 retains one valid canonical row with no room ID for retry; +deleting it blindly would race another request. Leaving an existing room follows +bridgev2's existing orphaned-DM cleanup/recreation behavior. + +Historical invalid public-ID portal records are not migrated or deleted +automatically. Review such records separately before any production cleanup. + +## Why many conversations may appear + +Neither exact ID resolution nor its error path calls conversation synchronization. +The search-only display-name fallback enumerates contacts, not conversations, and +does not create portal rooms. Authentication recovery on these calls refreshes or +re-authenticates the session without invoking `Connect` or a broad chat sync. + +Existing legitimate broad synchronization sources are: + +* `LineClient.Connect`: group synchronization, DM synchronization and recent-message + prefetch on connection/startup. +* LINE SSE `fullSync`: refreshes block state, group/DM state and recent messages; + this can occur when server/client revision state requires reconciliation. +* Recent-message backfill containing system events: re-fetches current group state + after replay so historical membership/name events cannot leave stale state. + +New `sync_trigger` log fields identify `connect`, `line_full_sync`, and +`backfill_state_restore`. Existing portal mappings, message deduplication rows, +LINE revision state and persisted chat-deletion cutoffs affect what those paths +create or skip. No legitimate sync path has been disabled. The reported temporal +association with `start-chat` is not sufficient to establish a causal connection. +The earlier #196/#189 integrations changed media processing, not these sync paths. + +## Diagnosing an outbound message + +Trace one Matrix event ID rather than retrying across unrelated rooms/accounts: + +1. Matrix connector filtering and decryption. Relevant messages include + `Dropping ... event from user with no permission to send events`, + `Dropping unencrypted event as encryption is configured to be required`, + `Decrypting received event`, and `Failed to decrypt event`. +2. Portal lookup and `FindPreferredLogin`. A login-scoped portal uses its receiver + login. A different Matrix sender cannot use that login unless an explicitly + configured relay permits it. `Failed to get user login to handle Matrix event` + distinguishes this stage. Working bot commands do not establish that every + portal event has the same sender/login association or encryption session. +3. `Handling outbound Matrix message for LINE`: the connector has been reached. + Correlation fields include `matrix_event_id`, `matrix_sender`, `line_login_id` + and `line_recipient`. The recipient is the portal's canonical remote ID. +4. Message conversion and Letter Sealing selection/encryption, then the signed + `TalkService.sendMessage` RPC. `Finished LINE API send attempt` records + `success` and, on acknowledgement, `line_message_id`. Message bodies, access + tokens and QR approval data are not included in these new diagnostics. +5. The returned LINE message ID becomes the bridge message record. bridgev2 sends + the corresponding success/error status. An API acknowledgement is not proof of + recipient-side delivery/read status. + +Three defects were reproduced with mock responses and fixed: the send RPC did not +inherit the Matrix request context; HTTP-200 error wrappers lost TalkException +codes required by existing recovery logic; and a missing acknowledgement could +produce success with an empty ID or a nil dereference. Authentication/permission +rules, supported encryption paths and inbound conversion are unchanged. + +The actual Element delivery incident still needs a failing event's correlated +stage/error/status to identify its cause. Useful evidence is the event ID, sender, +portal receiver/login ownership, and the messages above, with tokens, message +bodies and approval codes removed. Do not infer a Synapse connectivity failure +from this report; command delivery and MID-based portal creation already work. + +## QR login: separate state machine + +Enable `network.qr_login: true` in the runtime configuration, then use: + +```text +login dev.highest.matrix.line.qr_login +relogin dev.highest.matrix.line.qr_login +cancel +``` + +Outside the management room, prefix these with `!line` (or the configured prefix). +The bridge requests a session and QR callback, adds the login public key, and +returns a display-and-wait step. bridgev2 renders the QR image for Matrix. After +LINE scan approval, a reusable certificate may satisfy verification; otherwise +the user must enter/approve a PIN on the primary device. Only then does token +exchange occur. The bridge verifies the token's profile/MID and required E2EE +keys before persisting a login, and checks cancellation/account identity during +finalization. + +Overlapping Start/Wait operations are now rejected instead of consuming another +waiter's polling result or blocking indefinitely. Startup deadline errors retain +`context.DeadlineExceeded`, rather than always becoming `context.Canceled`. +Cancel still interrupts the underlying attempt without waiting on the operation +guard. Expired polling is retried only within the existing server count/attempt +deadline; exhausted verification requires a fresh login attempt and QR scan. +There is no new speculative refresh endpoint or device-approval bypass. + +V3 refresh tokens are persisted by the existing finalizer. If refresh fails for a +passwordless QR account, reconnect must request a new QR approval rather than +inventing email/password credentials. Mock tests cover this and ensure that a +token response without a usable required keychain does not replace the stored +session. These lifecycle fixes do not establish the cause of an unspecified live +QR scanning/token-exchange failure. Successful live QR login and reconnection +have not been validated in this workspace. + +## Test boundaries + +New tests use mock LINE HTTP responses and isolated SQLite databases. They cover +shared resolution, invalid identities, lookup failures, cancellation/deadlines, +canonical portal-cache concurrency, existing-room reuse, creation failure/retry, +reopening after leaving, outbound ownership/routing/error propagation and QR state +transitions. The production PostgreSQL database and live Matrix/LINE accounts are +not used. Full live command concurrency, recipient delivery, and successful live +QR/session restoration require operator validation after deployment; mocked tests +are not a claim that those account-specific workflows have been verified. From 587164f7ced99bc5765ac7df4eafc11314e6d602 Mon Sep 17 00:00:00 2001 From: Michael Ramsey Date: Tue, 20 Oct 2026 17:15:28 -0400 Subject: [PATCH 9/9] diagnostics: identify synchronization triggers and document chat flows Label connection, LINE fullSync and historical-state reconciliation without changing sync behavior. Document canonical identity resolution, portal retry/reopening, outbound correlation and ownership checks, QR lifecycle fixes, mocked coverage and remaining live-verification requirements. --- README.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/README.md b/README.md index 732232fe..3c0f14f5 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,9 @@ Based on the [mautrix-twilio](https://github.com/mautrix/twilio) bridge > need to log out of the bridge first as the bridge will try to automatically > log back in and invalidate the LINE Chrome session again. +For direct-chat identifiers, synchronization triggers, outbound-message diagnostics, +and QR login troubleshooting, see [the debugging guide](readme/DEBUGGING.md). + ## Features - [x] Messages