diff --git a/pkg/connector/handle_message.go b/pkg/connector/handle_message.go index b645dfd..2c7fb29 100644 --- a/pkg/connector/handle_message.go +++ b/pkg/connector/handle_message.go @@ -35,6 +35,45 @@ func (lc *LineClient) newMessageHandler() *handlers.Handler { } } +func e2eeChunkLengths(chunks []string) []int { + lengths := make([]int, len(chunks)) + for i, chunk := range chunks { + lengths[i] = len(chunk) + } + return lengths +} + +func groupDecryptLogContext(evt *zerolog.Event, msg *line.Message, chatMID string, opType int) *zerolog.Event { + evt = evt. + Str("msg_id", msg.ID). + Str("chat_mid", chatMID). + Str("from", msg.From). + Str("to", msg.To). + Int("to_type", msg.ToType). + Int("op_type", opType). + Int("content_type", msg.ContentType). + Int("chunk_count", len(msg.Chunks)). + Ints("chunk_lengths", e2eeChunkLengths(msg.Chunks)) + + if version := msg.ContentMetadata["e2eeVersion"]; version != "" { + evt = evt.Str("e2ee_version", version) + } + if len(msg.Chunks) >= 5 { + if senderKeyID, err := e2ee.DecodeKeyID(msg.Chunks[len(msg.Chunks)-2]); err == nil { + evt = evt.Int("sender_key_id", senderKeyID) + } else { + evt = evt.Str("sender_key_id_error", err.Error()) + } + if groupKeyID, err := e2ee.DecodeKeyID(msg.Chunks[len(msg.Chunks)-1]); err == nil { + evt = evt.Int("group_key_id", groupKeyID) + } else { + evt = evt.Str("group_key_id_error", err.Error()) + } + } + + return evt +} + func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) { // Only process known content types; skip system messages (group created, member invited, etc.) if !isBridgeableContentType(msg) { @@ -53,7 +92,7 @@ func (lc *LineClient) queueIncomingMessage(msg *line.Message, opType int) { portalIDStr := portalMIDForMessage(msg, opType) portalKey := networkid.PortalKey{ID: makePortalID(portalIDStr), Receiver: lc.UserLogin.ID} - bodyText, unwrappedText := lc.decryptMessageBody(msg, portalIDStr) + bodyText, unwrappedText := lc.decryptMessageBody(msg, portalIDStr, opType) ts := lc.parseMessageTimestamp(msg) lc.UserLogin.Bridge.QueueRemoteEvent(lc.UserLogin, &simplevent.Message[line.Message]{ @@ -123,7 +162,7 @@ func (lc *LineClient) parseMessageTimestamp(msg *line.Message) time.Time { // decryptMessageBody runs E2EE decryption (when needed) for an inbound message // and returns the plaintext body plus the JSON-unwrapped text. Shared by the // live message path (queueIncomingMessage) and the backfill path (FetchMessages). -func (lc *LineClient) decryptMessageBody(msg *line.Message, portalIDStr string) (bodyText, unwrappedText string) { +func (lc *LineClient) decryptMessageBody(msg *line.Message, portalIDStr string, opType int) (bodyText, unwrappedText string) { // Handle Content bodyText = msg.Text if bodyText == "" && len(msg.Chunks) > 0 { @@ -144,7 +183,8 @@ func (lc *LineClient) decryptMessageBody(msg *line.Message, portalIDStr string) if len(msg.Chunks) >= 5 { if gkID, err := e2ee.DecodeKeyID(msg.Chunks[len(msg.Chunks)-1]); err == nil && gkID != 0 { if errFetch := lc.fetchAndUnwrapGroupKey(context.Background(), portalIDStr, gkID); errFetch != nil { - lc.UserLogin.Bridge.Log.Debug().Err(errFetch).Int("key_id", gkID).Str("chat_mid", portalIDStr).Msg("Prefetch group key before decrypt failed") + groupDecryptLogContext(lc.UserLogin.Bridge.Log.Debug().Err(errFetch), msg, portalIDStr, opType). + Msg("Prefetch group key before decrypt failed") } } } @@ -153,12 +193,17 @@ func (lc *LineClient) decryptMessageBody(msg *line.Message, portalIDStr string) if err == nil { bodyText = pt } else { - lc.UserLogin.Bridge.Log.Debug().Err(err).Int("key_id", keyID).Str("chat_mid", portalIDStr).Msg("DecryptGroupMessage failed, trying to fetch key") + groupDecryptLogContext(lc.UserLogin.Bridge.Log.Debug().Err(err), msg, portalIDStr, opType). + Msg("DecryptGroupMessage failed, trying to fetch key") if keyID != 0 { if errFetch := lc.fetchAndUnwrapGroupKey(context.Background(), portalIDStr, keyID); errFetch != nil { - lc.UserLogin.Bridge.Log.Warn().Err(errFetch).Int("key_id", keyID).Str("chat_mid", portalIDStr).Msg("Failed to fetch/unwrap group key") + groupDecryptLogContext(lc.UserLogin.Bridge.Log.Warn().Err(errFetch), msg, portalIDStr, opType). + Msg("Failed to fetch/unwrap group key") } else if ptRetry, _, errRetry := lc.E2EE.DecryptGroupMessage(msg, portalIDStr); errRetry == nil { bodyText = ptRetry + } else { + groupDecryptLogContext(lc.UserLogin.Bridge.Log.Warn().Err(errRetry), msg, portalIDStr, opType). + Msg("DecryptGroupMessage failed after group key refresh") } } } diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 3bd60d3..75081ac 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -178,7 +178,11 @@ func (lc *LineClient) FetchMessages(ctx context.Context, params bridgev2.FetchMe continue } - bodyText, unwrappedText := lc.decryptMessageBody(msg, chatMID) + opType := OpReceiveMessage + if msg.From == lc.Mid { + opType = OpSendMessage + } + bodyText, unwrappedText := lc.decryptMessageBody(msg, chatMID, int(opType)) converted, err := lc.convertLineMessage(ctx, params.Portal, intent, *msg, bodyText, unwrappedText) if err != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Str("msg_id", msg.ID).Str("chat_mid", chatMID).Msg("Failed to convert message for backfill") diff --git a/pkg/runner.go b/pkg/runner.go index ade9a05..77da7b0 100644 --- a/pkg/runner.go +++ b/pkg/runner.go @@ -138,6 +138,73 @@ func (r *Runner) getChannel(id int) (uint32, error) { return ptr, nil } +func ltsmPanicError(operation string, recovered any) error { + if err, ok := recovered.(error); ok { + return fmt.Errorf("%s panicked: %w", operation, err) + } + return fmt.Errorf("%s panicked: %v", operation, recovered) +} + +func (r *Runner) exportE2EEKeyPanicSafe(keyPtr uint32) (exported []byte, err error) { + defer func() { + if recovered := recover(); recovered != nil { + exported = nil + err = ltsmPanicError("ltsm E2EEKey.exportKey", recovered) + } + }() + return r.rt.E2EEKeyExportKey(keyPtr) +} + +func (r *Runner) unwrapGroupSharedKeyPanicSafe(chanPtr uint32, encKey []byte) (keyPtr uint32, err error) { + defer func() { + if recovered := recover(); recovered != nil { + keyPtr = 0 + err = ltsmPanicError("ltsm E2EEChannel.unwrapGroupSharedKey", recovered) + } + }() + return r.rt.E2EEChannelUnwrapGroupSharedKey(chanPtr, encKey) +} + +func (r *Runner) encryptV1PanicSafe(chanPtr uint32, plaintext []byte) (ciphertext []byte, err error) { + defer func() { + if recovered := recover(); recovered != nil { + ciphertext = nil + err = ltsmPanicError("ltsm E2EEChannel.encryptV1", recovered) + } + }() + return r.rt.E2EEChannelEncryptV1(chanPtr, plaintext) +} + +func (r *Runner) encryptV2PanicSafe(chanPtr uint32, to, from string, senderKeyID, receiverKeyID, contentType int, seq int64, plaintext []byte) (ciphertext []byte, err error) { + defer func() { + if recovered := recover(); recovered != nil { + ciphertext = nil + err = ltsmPanicError("ltsm E2EEChannel.encryptV2", recovered) + } + }() + return r.rt.E2EEChannelEncryptV2(chanPtr, to, from, senderKeyID, receiverKeyID, contentType, seq, plaintext) +} + +func (r *Runner) decryptV1PanicSafe(chanPtr uint32, ciphertext []byte) (plaintext []byte, err error) { + defer func() { + if recovered := recover(); recovered != nil { + plaintext = nil + err = ltsmPanicError("ltsm E2EEChannel.decryptV1", recovered) + } + }() + return r.rt.E2EEChannelDecryptV1(chanPtr, ciphertext) +} + +func (r *Runner) decryptV2PanicSafe(chanPtr uint32, to, from string, senderKeyID, receiverKeyID, contentType int, ciphertext []byte) (plaintext []byte, err error) { + defer func() { + if recovered := recover(); recovered != nil { + plaintext = nil + err = ltsmPanicError("ltsm E2EEChannel.decryptV2", recovered) + } + }() + return r.rt.E2EEChannelDecryptV2(chanPtr, to, from, senderKeyID, receiverKeyID, contentType, ciphertext) +} + func (r *Runner) GetSignature(reqPath, body, accessToken string) (string, error) { if reqPath == "" { reqPath = "/" @@ -305,7 +372,7 @@ func (r *Runner) LoginUnwrapKeyChain(serverPubB64, encryptedKeyChainB64 string) id := r.putKey(keyPtr) - exported, err := r.rt.E2EEKeyExportKey(keyPtr) + exported, err := r.exportE2EEKeyPanicSafe(keyPtr) if err != nil { return nil, fmt.Errorf("failed to export key %d: %w", i, err) } @@ -448,7 +515,7 @@ func (r *Runner) ChannelUnwrapGroupSharedKey(channelID int, encryptedSharedKeyB6 return 0, fmt.Errorf("invalid encrypted shared key: %w", err) } - keyPtr, err := r.rt.E2EEChannelUnwrapGroupSharedKey(chanPtr, encBytes) + keyPtr, err := r.unwrapGroupSharedKeyPanicSafe(chanPtr, encBytes) if err != nil { return 0, err } @@ -477,7 +544,7 @@ func (r *Runner) ChannelEncryptV1(channelID int, plaintext string) (string, erro return "", err } - ctBytes, err := r.rt.E2EEChannelEncryptV1(chanPtr, []byte(plaintext)) + ctBytes, err := r.encryptV1PanicSafe(chanPtr, []byte(plaintext)) if err != nil { return "", err } @@ -507,7 +574,7 @@ func (r *Runner) ChannelEncryptV2(channelID int, to, from string, senderKeyID, r return "", err } - ctBytes, err := r.rt.E2EEChannelEncryptV2(chanPtr, + ctBytes, err := r.encryptV2PanicSafe(chanPtr, to, from, senderKeyID, receiverKeyID, contentType, int64(seq), []byte(plaintext)) if err != nil { return "", err @@ -542,7 +609,7 @@ func (r *Runner) ChannelDecryptV1(channelID, senderKeyID, receiverKeyID int, cip return "", "", err } - ptBytes, err := r.rt.E2EEChannelDecryptV1(chanPtr, ctBytes) + ptBytes, err := r.decryptV1PanicSafe(chanPtr, ctBytes) if err != nil { return "", "", err } @@ -576,7 +643,7 @@ func (r *Runner) ChannelDecryptV2(channelID int, to, from string, senderKeyID, r return "", "", err } - ptBytes, err := r.rt.E2EEChannelDecryptV2(chanPtr, + ptBytes, err := r.decryptV2PanicSafe(chanPtr, to, from, senderKeyID, receiverKeyID, contentType, ctBytes) if err != nil { return "", "", err