X-Git-Url: https://git.nmode.ca/signal-cli/blobdiff_plain/c7a7d00da56f8f90fe8d6a11042fa2cd7cb3b985..9439de1c153be43f695bd30170fedf05536674da:/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java diff --git a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java index 1af99ff1..0e100d72 100644 --- a/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java +++ b/lib/src/main/java/org/asamk/signal/manager/ManagerImpl.java @@ -1,5 +1,5 @@ /* - Copyright (C) 2015-2021 AsamK and contributors + Copyright (C) 2015-2022 AsamK and contributors This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by @@ -16,19 +16,22 @@ */ package org.asamk.signal.manager; -import org.asamk.signal.manager.actions.HandleAction; import org.asamk.signal.manager.api.Configuration; import org.asamk.signal.manager.api.Device; import org.asamk.signal.manager.api.Group; import org.asamk.signal.manager.api.Identity; import org.asamk.signal.manager.api.InactiveGroupLinkException; import org.asamk.signal.manager.api.InvalidDeviceLinkException; +import org.asamk.signal.manager.api.InvalidStickerException; import org.asamk.signal.manager.api.Message; import org.asamk.signal.manager.api.Pair; import org.asamk.signal.manager.api.RecipientIdentifier; import org.asamk.signal.manager.api.SendGroupMessageResults; import org.asamk.signal.manager.api.SendMessageResult; import org.asamk.signal.manager.api.SendMessageResults; +import org.asamk.signal.manager.api.StickerPack; +import org.asamk.signal.manager.api.StickerPackId; +import org.asamk.signal.manager.api.StickerPackUrl; import org.asamk.signal.manager.api.TypingAction; import org.asamk.signal.manager.api.UnregisteredRecipientException; import org.asamk.signal.manager.api.UpdateGroup; @@ -43,32 +46,24 @@ import org.asamk.signal.manager.helper.Context; import org.asamk.signal.manager.storage.SignalAccount; import org.asamk.signal.manager.storage.groups.GroupInfo; import org.asamk.signal.manager.storage.identities.IdentityInfo; -import org.asamk.signal.manager.storage.messageCache.CachedMessage; import org.asamk.signal.manager.storage.recipients.Contact; import org.asamk.signal.manager.storage.recipients.Profile; import org.asamk.signal.manager.storage.recipients.RecipientAddress; import org.asamk.signal.manager.storage.recipients.RecipientId; import org.asamk.signal.manager.storage.stickers.Sticker; -import org.asamk.signal.manager.storage.stickers.StickerPackId; +import org.asamk.signal.manager.util.AttachmentUtils; import org.asamk.signal.manager.util.KeyUtils; import org.asamk.signal.manager.util.StickerUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.whispersystems.libsignal.InvalidKeyException; -import org.whispersystems.libsignal.ecc.ECPublicKey; import org.whispersystems.libsignal.util.guava.Optional; import org.whispersystems.signalservice.api.SignalSessionLock; import org.whispersystems.signalservice.api.messages.SignalServiceDataMessage; -import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope; import org.whispersystems.signalservice.api.messages.SignalServiceReceiptMessage; import org.whispersystems.signalservice.api.messages.SignalServiceTypingMessage; -import org.whispersystems.signalservice.api.push.ACI; -import org.whispersystems.signalservice.api.push.exceptions.AuthorizationFailedException; import org.whispersystems.signalservice.api.util.DeviceNameUtil; import org.whispersystems.signalservice.api.util.InvalidNumberException; import org.whispersystems.signalservice.api.util.PhoneNumberFormatter; -import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState; -import org.whispersystems.signalservice.api.websocket.WebSocketUnavailableException; import org.whispersystems.signalservice.internal.util.DynamicCredentialsProvider; import org.whispersystems.signalservice.internal.util.Hex; import org.whispersystems.signalservice.internal.util.Util; @@ -76,12 +71,8 @@ import org.whispersystems.signalservice.internal.util.Util; import java.io.File; import java.io.IOException; import java.net.URI; -import java.net.URISyntaxException; -import java.net.URLEncoder; -import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; -import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -90,38 +81,29 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.Stream; -import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.schedulers.Schedulers; - -import static org.asamk.signal.manager.config.ServiceConfig.capabilities; +import io.reactivex.rxjava3.disposables.CompositeDisposable; public class ManagerImpl implements Manager { private final static Logger logger = LoggerFactory.getLogger(ManagerImpl.class); - private final SignalDependencies dependencies; - private SignalAccount account; - - private final ExecutorService executor = Executors.newCachedThreadPool(); - + private final SignalDependencies dependencies; private final Context context; - private boolean hasCaughtUpWithOldMessages = false; - private boolean ignoreAttachments = false; + private final ExecutorService executor = Executors.newCachedThreadPool(); private Thread receiveThread; + private boolean isReceivingSynchronous; private final Set weakHandlers = new HashSet<>(); private final Set messageHandlers = new HashSet<>(); private final List closedListeners = new ArrayList<>(); - private boolean isReceivingSynchronous; - private boolean needsToRetryFailedMessages = false; + private final CompositeDisposable disposable = new CompositeDisposable(); ManagerImpl( SignalAccount account, @@ -155,6 +137,27 @@ public class ManagerImpl implements Manager { final var stickerPackStore = new StickerPackStore(pathConfig.stickerPacksPath()); this.context = new Context(account, dependencies, avatarStore, attachmentStore, stickerPackStore); + this.context.getAccountHelper().setUnregisteredListener(this::close); + this.context.getReceiveHelper().setAuthenticationFailureListener(this::close); + this.context.getReceiveHelper().setCaughtUpWithOldMessagesListener(() -> { + synchronized (this) { + this.notifyAll(); + } + }); + disposable.add(account.getIdentityKeyStore().getIdentityChanges().subscribe(recipientId -> { + logger.trace("Archiving old sessions for {}", recipientId); + account.getSessionStore().archiveSessions(recipientId); + account.getSenderKeyStore().deleteSharedWith(recipientId); + final var profile = account.getRecipientStore().getProfile(recipientId); + if (profile != null) { + account.getRecipientStore() + .storeProfile(recipientId, + Profile.newBuilder(profile) + .withUnidentifiedAccessMode(Profile.UnidentifiedAccessMode.UNKNOWN) + .withLastUpdateTimestamp(0) + .build()); + } + })); } @Override @@ -164,36 +167,9 @@ public class ManagerImpl implements Manager { @Override public void checkAccountState() throws IOException { - if (account.getLastReceiveTimestamp() == 0) { - logger.info("The Signal protocol expects that incoming messages are regularly received."); - } else { - var diffInMilliseconds = System.currentTimeMillis() - account.getLastReceiveTimestamp(); - long days = TimeUnit.DAYS.convert(diffInMilliseconds, TimeUnit.MILLISECONDS); - if (days > 7) { - logger.warn( - "Messages have been last received {} days ago. The Signal protocol expects that incoming messages are regularly received.", - days); - } - } - try { - context.getPreKeyHelper().refreshPreKeysIfNecessary(); - if (account.getAci() == null) { - account.setAci(ACI.parseOrNull(dependencies.getAccountManager().getWhoAmI().getAci())); - } - updateAccountAttributes(null); - } catch (AuthorizationFailedException e) { - account.setRegistered(false); - throw e; - } + context.getAccountHelper().checkAccountState(); } - /** - * This is used for checking a set of phone numbers for registration on Signal - * - * @param numbers The set of phone number in question - * @return A map of numbers to canonicalized number and uuid. If a number is not registered the uuid is null. - * @throws IOException if it's unable to get the contacts to check if they're registered - */ @Override public Map> areUsersRegistered(Set numbers) throws IOException { final var canonicalizedNumbers = numbers.stream().collect(Collectors.toMap(n -> n, n -> { @@ -224,40 +200,22 @@ public class ManagerImpl implements Manager { @Override public void updateAccountAttributes(String deviceName) throws IOException { - final String encryptedDeviceName; - if (deviceName == null) { - encryptedDeviceName = account.getEncryptedDeviceName(); - } else { - final var privateKey = account.getIdentityKeyPair().getPrivateKey(); - encryptedDeviceName = DeviceNameUtil.encryptDeviceName(deviceName, privateKey); - account.setEncryptedDeviceName(encryptedDeviceName); + if (deviceName != null) { + context.getAccountHelper().setDeviceName(deviceName); } - dependencies.getAccountManager() - .setAccountAttributes(encryptedDeviceName, - null, - account.getLocalRegistrationId(), - true, - null, - account.getPinMasterKey() == null ? null : account.getPinMasterKey().deriveRegistrationLock(), - account.getSelfUnidentifiedAccessKey(), - account.isUnrestrictedUnidentifiedAccess(), - capabilities, - account.isDiscoverableByPhoneNumber()); + context.getAccountHelper().updateAccountAttributes(); } @Override public Configuration getConfiguration() { final var configurationStore = account.getConfigurationStore(); - return new Configuration(java.util.Optional.ofNullable(configurationStore.getReadReceipts()), - java.util.Optional.ofNullable(configurationStore.getUnidentifiedDeliveryIndicators()), - java.util.Optional.ofNullable(configurationStore.getTypingIndicators()), - java.util.Optional.ofNullable(configurationStore.getLinkPreviews())); + return Configuration.from(configurationStore); } @Override public void updateConfiguration( Configuration configuration - ) throws IOException, NotMasterDeviceException { + ) throws NotMasterDeviceException { if (!account.isMasterDevice()) { throw new NotMasterDeviceException(); } @@ -278,13 +236,6 @@ public class ManagerImpl implements Manager { context.getSyncHelper().sendConfigurationMessage(); } - /** - * @param givenName if null, the previous givenName will be kept - * @param familyName if null, the previous familyName will be kept - * @param about if null, the previous about text will be kept - * @param aboutEmoji if null, the previous about emoji will be kept - * @param avatar if avatar is null the image from the local avatar store is used (if present), - */ @Override public void setProfile( String givenName, final String familyName, String about, String aboutEmoji, java.util.Optional avatar @@ -300,28 +251,12 @@ public class ManagerImpl implements Manager { @Override public void unregister() throws IOException { - // When setting an empty GCM id, the Signal-Server also sets the fetchesMessages property to false. - // If this is the master device, other users can't send messages to this number anymore. - // If this is a linked device, other users can still send messages, but this device doesn't receive them anymore. - dependencies.getAccountManager().setGcmId(Optional.absent()); - - account.setRegistered(false); - close(); + context.getAccountHelper().unregister(); } @Override public void deleteAccount() throws IOException { - try { - context.getPinHelper().removeRegistrationLockPin(); - } catch (IOException e) { - logger.warn("Failed to remove registration lock pin"); - } - account.setRegistrationLockPin(null, null); - - dependencies.getAccountManager().deleteAccount(); - - account.setRegistered(false); - close(); + context.getAccountHelper().deleteAccount(); } @Override @@ -354,56 +289,25 @@ public class ManagerImpl implements Manager { } @Override - public void removeLinkedDevices(long deviceId) throws IOException { - dependencies.getAccountManager().removeDevice(deviceId); - var devices = dependencies.getAccountManager().getDevices(); - account.setMultiDevice(devices.size() > 1); + public void removeLinkedDevices(int deviceId) throws IOException { + context.getAccountHelper().removeLinkedDevices(deviceId); } @Override public void addDeviceLink(URI linkUri) throws IOException, InvalidDeviceLinkException { - var info = DeviceLinkInfo.parseDeviceLinkUri(linkUri); - - addDevice(info.deviceIdentifier(), info.deviceKey()); - } - - private void addDevice( - String deviceIdentifier, ECPublicKey deviceKey - ) throws IOException, InvalidDeviceLinkException { - var identityKeyPair = account.getIdentityKeyPair(); - var verificationCode = dependencies.getAccountManager().getNewDeviceVerificationCode(); - - try { - dependencies.getAccountManager() - .addDevice(deviceIdentifier, - deviceKey, - identityKeyPair, - Optional.of(account.getProfileKey().serialize()), - verificationCode); - } catch (InvalidKeyException e) { - throw new InvalidDeviceLinkException("Invalid device link", e); - } - account.setMultiDevice(true); + var deviceLinkInfo = DeviceLinkInfo.parseDeviceLinkUri(linkUri); + context.getAccountHelper().addDevice(deviceLinkInfo); } @Override - public void setRegistrationLockPin(java.util.Optional pin) throws IOException { + public void setRegistrationLockPin(java.util.Optional pin) throws IOException, NotMasterDeviceException { if (!account.isMasterDevice()) { - throw new RuntimeException("Only master device can set a PIN"); + throw new NotMasterDeviceException(); } if (pin.isPresent()) { - final var masterKey = account.getPinMasterKey() != null - ? account.getPinMasterKey() - : KeyUtils.createMasterKey(); - - context.getPinHelper().setRegistrationLockPin(pin.get(), masterKey); - - account.setRegistrationLockPin(pin.get(), masterKey); + context.getAccountHelper().setRegistrationPin(pin.get()); } else { - // Remove KBS Pin - context.getPinHelper().removeRegistrationLockPin(); - - account.setRegistrationLockPin(null, null); + context.getAccountHelper().removeRegistrationPin(); } } @@ -426,33 +330,9 @@ public class ManagerImpl implements Manager { return null; } - return new Group(groupInfo.getGroupId(), - groupInfo.getTitle(), - groupInfo.getDescription(), - groupInfo.getGroupInviteLink(), - groupInfo.getMembers() - .stream() - .map(account.getRecipientStore()::resolveRecipientAddress) - .collect(Collectors.toSet()), - groupInfo.getPendingMembers() - .stream() - .map(account.getRecipientStore()::resolveRecipientAddress) - .collect(Collectors.toSet()), - groupInfo.getRequestingMembers() - .stream() - .map(account.getRecipientStore()::resolveRecipientAddress) - .collect(Collectors.toSet()), - groupInfo.getAdminMembers() - .stream() - .map(account.getRecipientStore()::resolveRecipientAddress) - .collect(Collectors.toSet()), - groupInfo.isBlocked(), - groupInfo.getMessageExpirationTimer(), - groupInfo.getPermissionAddMember(), - groupInfo.getPermissionEditDetails(), - groupInfo.getPermissionSendMessage(), - groupInfo.isMember(account.getSelfRecipientId()), - groupInfo.isAdmin(account.getSelfRecipientId())); + return Group.from(groupInfo, + account.getRecipientStore()::resolveRecipientAddress, + account.getSelfRecipientId()); } @Override @@ -525,33 +405,28 @@ public class ManagerImpl implements Manager { try { final var recipientId = context.getRecipientHelper().resolveRecipient(single); final var result = context.getSendHelper().sendMessage(messageBuilder, recipientId); - results.put(recipient, - List.of(SendMessageResult.from(result, - account.getRecipientStore(), - account.getRecipientStore()::resolveRecipientAddress))); + results.put(recipient, List.of(toSendMessageResult(result))); } catch (UnregisteredRecipientException e) { results.put(recipient, List.of(SendMessageResult.unregisteredFailure(single.toPartialRecipientAddress()))); } } else if (recipient instanceof RecipientIdentifier.NoteToSelf) { final var result = context.getSendHelper().sendSelfMessage(messageBuilder); - results.put(recipient, - List.of(SendMessageResult.from(result, - account.getRecipientStore(), - account.getRecipientStore()::resolveRecipientAddress))); + results.put(recipient, List.of(toSendMessageResult(result))); } else if (recipient instanceof RecipientIdentifier.Group group) { final var result = context.getSendHelper().sendAsGroupMessage(messageBuilder, group.groupId()); - results.put(recipient, - result.stream() - .map(sendMessageResult -> SendMessageResult.from(sendMessageResult, - account.getRecipientStore(), - account.getRecipientStore()::resolveRecipientAddress)) - .toList()); + results.put(recipient, result.stream().map(this::toSendMessageResult).toList()); } } return new SendMessageResults(timestamp, results); } + private SendMessageResult toSendMessageResult(final org.whispersystems.signalservice.api.messages.SendMessageResult result) { + return SendMessageResult.from(result, + account.getRecipientStore(), + account.getRecipientStore()::resolveRecipientAddress); + } + private SendMessageResults sendTypingMessage( SignalServiceTypingMessage.Action action, Set recipients ) throws IOException, NotAGroupMemberException, GroupNotFoundException, GroupSendingNotAllowedException { @@ -563,10 +438,7 @@ public class ManagerImpl implements Manager { try { final var recipientId = context.getRecipientHelper().resolveRecipient(single); final var result = context.getSendHelper().sendTypingMessage(message, recipientId); - results.put(recipient, - List.of(SendMessageResult.from(result, - account.getRecipientStore(), - account.getRecipientStore()::resolveRecipientAddress))); + results.put(recipient, List.of(toSendMessageResult(result))); } catch (UnregisteredRecipientException e) { results.put(recipient, List.of(SendMessageResult.unregisteredFailure(single.toPartialRecipientAddress()))); @@ -575,12 +447,7 @@ public class ManagerImpl implements Manager { final var groupId = ((RecipientIdentifier.Group) recipient).groupId(); final var message = new SignalServiceTypingMessage(action, timestamp, Optional.of(groupId.serialize())); final var result = context.getSendHelper().sendGroupTypingMessage(message, groupId); - results.put(recipient, - result.stream() - .map(r -> SendMessageResult.from(r, - account.getRecipientStore(), - account.getRecipientStore()::resolveRecipientAddress)) - .toList()); + results.put(recipient, result.stream().map(this::toSendMessageResult).toList()); } } return new SendMessageResults(timestamp, results); @@ -625,11 +492,7 @@ public class ManagerImpl implements Manager { try { final var result = context.getSendHelper() .sendReceiptMessage(receiptMessage, context.getRecipientHelper().resolveRecipient(sender)); - return new SendMessageResults(timestamp, - Map.of(sender, - List.of(SendMessageResult.from(result, - account.getRecipientStore(), - account.getRecipientStore()::resolveRecipientAddress)))); + return new SendMessageResults(timestamp, Map.of(sender, List.of(toSendMessageResult(result)))); } catch (UnregisteredRecipientException e) { return new SendMessageResults(timestamp, Map.of(sender, List.of(SendMessageResult.unregisteredFailure(sender.toPartialRecipientAddress())))); @@ -639,7 +502,7 @@ public class ManagerImpl implements Manager { @Override public SendMessageResults sendMessage( Message message, Set recipients - ) throws IOException, AttachmentInvalidException, NotAGroupMemberException, GroupNotFoundException, GroupSendingNotAllowedException, UnregisteredRecipientException { + ) throws IOException, AttachmentInvalidException, NotAGroupMemberException, GroupNotFoundException, GroupSendingNotAllowedException, UnregisteredRecipientException, InvalidStickerException { final var messageBuilder = SignalServiceDataMessage.newBuilder(); applyMessage(messageBuilder, message); return sendMessage(messageBuilder, recipients); @@ -647,7 +510,7 @@ public class ManagerImpl implements Manager { private void applyMessage( final SignalServiceDataMessage.Builder messageBuilder, final Message message - ) throws AttachmentInvalidException, IOException, UnregisteredRecipientException { + ) throws AttachmentInvalidException, IOException, UnregisteredRecipientException, InvalidStickerException { messageBuilder.withBody(message.messageText()); final var attachments = message.attachments(); if (attachments != null) { @@ -665,6 +528,30 @@ public class ManagerImpl implements Manager { List.of(), resolveMentions(quote.mentions()))); } + if (message.sticker().isPresent()) { + final var sticker = message.sticker().get(); + final var packId = StickerPackId.deserialize(sticker.packId()); + final var stickerId = sticker.stickerId(); + + final var stickerPack = context.getAccount().getStickerStore().getStickerPack(packId); + if (stickerPack == null) { + throw new InvalidStickerException("Sticker pack not found"); + } + final var manifest = context.getStickerHelper().getOrRetrieveStickerPack(packId, stickerPack.getPackKey()); + if (manifest.stickers().size() <= stickerId) { + throw new InvalidStickerException("Sticker id not part of this pack"); + } + final var manifestSticker = manifest.stickers().get(stickerId); + final var streamDetails = context.getStickerPackStore().retrieveSticker(packId, stickerId); + if (streamDetails == null) { + throw new InvalidStickerException("Missing local sticker file"); + } + messageBuilder.withSticker(new SignalServiceDataMessage.Sticker(packId.serialize(), + stickerPack.getPackKey(), + stickerId, + manifestSticker.emoji(), + AttachmentUtils.createAttachment(streamDetails, Optional.absent()))); + } } private ArrayList resolveMentions(final List mentionList) throws IOException, UnregisteredRecipientException { @@ -762,7 +649,7 @@ public class ManagerImpl implements Manager { @Override public void setGroupBlocked( final GroupId groupId, final boolean blocked - ) throws GroupNotFoundException, IOException, NotMasterDeviceException { + ) throws GroupNotFoundException, NotMasterDeviceException { if (!account.isMasterDevice()) { throw new NotMasterDeviceException(); } @@ -771,9 +658,6 @@ public class ManagerImpl implements Manager { context.getSyncHelper().sendBlockedList(); } - /** - * Change the expiration timer for a contact - */ @Override public void setExpirationTimer( RecipientIdentifier.Single recipient, int messageExpirationTimer @@ -788,14 +672,8 @@ public class ManagerImpl implements Manager { } } - /** - * Upload the sticker pack from path. - * - * @param path Path can be a path to a manifest.json file or to a zip file that contains a manifest.json file - * @return if successful, returns the URL to install the sticker pack in the signal app - */ @Override - public URI uploadStickerPack(File path) throws IOException, StickerPackInvalidException { + public StickerPackUrl uploadStickerPack(File path) throws IOException, StickerPackInvalidException { var manifest = StickerUtils.getSignalServiceStickerManifestUpload(path); var messageSender = dependencies.getMessageSender(); @@ -807,17 +685,30 @@ public class ManagerImpl implements Manager { var sticker = new Sticker(packId, packKey); account.getStickerStore().updateSticker(sticker); - try { - return new URI("https", - "signal.art", - "/addstickers/", - "pack_id=" - + URLEncoder.encode(Hex.toStringCondensed(packId.serialize()), StandardCharsets.UTF_8) - + "&pack_key=" - + URLEncoder.encode(Hex.toStringCondensed(packKey), StandardCharsets.UTF_8)); - } catch (URISyntaxException e) { - throw new AssertionError(e); - } + return new StickerPackUrl(packId, packKey); + } + + @Override + public List getStickerPacks() { + final var stickerPackStore = context.getStickerPackStore(); + return account.getStickerStore().getStickerPacks().stream().map(pack -> { + if (stickerPackStore.existsStickerPack(pack.getPackId())) { + try { + final var manifest = stickerPackStore.retrieveManifest(pack.getPackId()); + return new StickerPack(pack.getPackId(), + new StickerPackUrl(pack.getPackId(), pack.getPackKey()), + pack.isInstalled(), + manifest.title(), + manifest.author(), + java.util.Optional.ofNullable(manifest.cover() == null ? null : manifest.cover().toApi()), + manifest.stickers().stream().map(JsonStickerPack.JsonSticker::toApi).toList()); + } catch (Exception e) { + logger.warn("Failed to read local sticker pack manifest: {}", e.getMessage(), e); + } + } + + return new StickerPack(pack.getPackId(), pack.getPackKey(), pack.isInstalled()); + }).toList(); } @Override @@ -832,54 +723,6 @@ public class ManagerImpl implements Manager { } } - private void retryFailedReceivedMessages(ReceiveMessageHandler handler) { - Set queuedActions = new HashSet<>(); - for (var cachedMessage : account.getMessageCache().getCachedMessages()) { - var actions = retryFailedReceivedMessage(handler, cachedMessage); - if (actions != null) { - queuedActions.addAll(actions); - } - } - handleQueuedActions(queuedActions); - } - - private List retryFailedReceivedMessage( - final ReceiveMessageHandler handler, final CachedMessage cachedMessage - ) { - var envelope = cachedMessage.loadEnvelope(); - if (envelope == null) { - cachedMessage.delete(); - return null; - } - - final var result = context.getIncomingMessageHandler() - .handleRetryEnvelope(envelope, ignoreAttachments, handler); - final var actions = result.first(); - final var exception = result.second(); - - if (exception instanceof UntrustedIdentityException) { - if (System.currentTimeMillis() - envelope.getServerDeliveredTimestamp() > 1000L * 60 * 60 * 24 * 30) { - // Envelope is more than a month old, cleaning up. - cachedMessage.delete(); - return null; - } - if (!envelope.hasSourceUuid()) { - final var identifier = ((UntrustedIdentityException) exception).getSender(); - final var recipientId = account.getRecipientStore().resolveRecipient(identifier); - try { - account.getMessageCache().replaceSender(cachedMessage, recipientId); - } catch (IOException ioException) { - logger.warn("Failed to move cached message to recipient folder: {}", ioException.getMessage()); - } - } - return null; - } - - // If successful and for all other errors that are not recoverable, delete the cached message - cachedMessage.delete(); - return actions; - } - @Override public void addReceiveHandler(final ReceiveMessageHandler handler, final boolean isWeakListener) { if (isReceivingSynchronous) { @@ -903,7 +746,7 @@ public class ManagerImpl implements Manager { logger.debug("Starting receiving messages"); while (!Thread.interrupted()) { try { - receiveMessagesInternal(Duration.ofMinutes(1), false, (envelope, e) -> { + context.getReceiveHelper().receiveMessages(Duration.ofMinutes(1), false, (envelope, e) -> { synchronized (messageHandlers) { Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> { try { @@ -920,7 +763,6 @@ public class ManagerImpl implements Manager { } } logger.debug("Finished receiving messages"); - hasCaughtUpWithOldMessages = false; synchronized (messageHandlers) { receiveThread = null; @@ -988,180 +830,21 @@ public class ManagerImpl implements Manager { isReceivingSynchronous = true; receiveThread = Thread.currentThread(); try { - receiveMessagesInternal(timeout, returnOnTimeout, handler); + context.getReceiveHelper().receiveMessages(timeout, returnOnTimeout, handler); } finally { receiveThread = null; - hasCaughtUpWithOldMessages = false; isReceivingSynchronous = false; } } - private void receiveMessagesInternal( - Duration timeout, boolean returnOnTimeout, ReceiveMessageHandler handler - ) throws IOException { - needsToRetryFailedMessages = true; - - // Use a Map here because java Set doesn't have a get method ... - Map queuedActions = new HashMap<>(); - - final var signalWebSocket = dependencies.getSignalWebSocket(); - final var webSocketStateDisposable = Observable.merge(signalWebSocket.getUnidentifiedWebSocketState(), - signalWebSocket.getWebSocketState()) - .subscribeOn(Schedulers.computation()) - .observeOn(Schedulers.computation()) - .distinctUntilChanged() - .subscribe(this::onWebSocketStateChange); - signalWebSocket.connect(); - - hasCaughtUpWithOldMessages = false; - var backOffCounter = 0; - final var MAX_BACKOFF_COUNTER = 9; - - while (!Thread.interrupted()) { - if (needsToRetryFailedMessages) { - retryFailedReceivedMessages(handler); - needsToRetryFailedMessages = false; - } - SignalServiceEnvelope envelope; - final CachedMessage[] cachedMessage = {null}; - final var nowMillis = System.currentTimeMillis(); - if (nowMillis - account.getLastReceiveTimestamp() > 60000) { - account.setLastReceiveTimestamp(nowMillis); - } - logger.debug("Checking for new message from server"); - try { - var result = signalWebSocket.readOrEmpty(timeout.toMillis(), envelope1 -> { - final var recipientId = envelope1.hasSourceUuid() ? account.getRecipientStore() - .resolveRecipient(envelope1.getSourceAddress()) : null; - // store message on disk, before acknowledging receipt to the server - cachedMessage[0] = account.getMessageCache().cacheMessage(envelope1, recipientId); - }); - backOffCounter = 0; - - if (result.isPresent()) { - envelope = result.get(); - logger.debug("New message received from server"); - } else { - logger.debug("Received indicator that server queue is empty"); - handleQueuedActions(queuedActions.keySet()); - queuedActions.clear(); - - hasCaughtUpWithOldMessages = true; - synchronized (this) { - this.notifyAll(); - } - - // Continue to wait another timeout for new messages - continue; - } - } catch (AssertionError e) { - if (e.getCause() instanceof InterruptedException) { - Thread.currentThread().interrupt(); - break; - } else { - throw e; - } - } catch (IOException e) { - logger.debug("Pipe unexpectedly unavailable: {}", e.getMessage()); - if (e instanceof WebSocketUnavailableException || "Connection closed!".equals(e.getMessage())) { - final var sleepMilliseconds = 100 * (long) Math.pow(2, backOffCounter); - backOffCounter = Math.min(backOffCounter + 1, MAX_BACKOFF_COUNTER); - logger.warn("Connection closed unexpectedly, reconnecting in {} ms", sleepMilliseconds); - try { - Thread.sleep(sleepMilliseconds); - } catch (InterruptedException interruptedException) { - return; - } - hasCaughtUpWithOldMessages = false; - signalWebSocket.connect(); - continue; - } - throw e; - } catch (TimeoutException e) { - backOffCounter = 0; - if (returnOnTimeout) return; - continue; - } - - final var result = context.getIncomingMessageHandler().handleEnvelope(envelope, ignoreAttachments, handler); - for (final var h : result.first()) { - final var existingAction = queuedActions.get(h); - if (existingAction == null) { - queuedActions.put(h, h); - } else { - existingAction.mergeOther(h); - } - } - final var exception = result.second(); - - if (hasCaughtUpWithOldMessages) { - handleQueuedActions(queuedActions.keySet()); - queuedActions.clear(); - } - if (cachedMessage[0] != null) { - if (exception instanceof UntrustedIdentityException) { - logger.debug("Keeping message with untrusted identity in message cache"); - final var address = ((UntrustedIdentityException) exception).getSender(); - final var recipientId = account.getRecipientStore().resolveRecipient(address); - if (!envelope.hasSourceUuid()) { - try { - cachedMessage[0] = account.getMessageCache().replaceSender(cachedMessage[0], recipientId); - } catch (IOException ioException) { - logger.warn("Failed to move cached message to recipient folder: {}", - ioException.getMessage()); - } - } - } else { - cachedMessage[0].delete(); - } - } - } - handleQueuedActions(queuedActions.keySet()); - queuedActions.clear(); - dependencies.getSignalWebSocket().disconnect(); - webSocketStateDisposable.dispose(); - } - - private void onWebSocketStateChange(final WebSocketConnectionState s) { - if (s.equals(WebSocketConnectionState.AUTHENTICATION_FAILED)) { - account.setRegistered(false); - try { - close(); - } catch (IOException e) { - e.printStackTrace(); - } - } - } - @Override public void setIgnoreAttachments(final boolean ignoreAttachments) { - this.ignoreAttachments = ignoreAttachments; + context.getReceiveHelper().setIgnoreAttachments(ignoreAttachments); } @Override public boolean hasCaughtUpWithOldMessages() { - return hasCaughtUpWithOldMessages; - } - - private void handleQueuedActions(final Collection queuedActions) { - logger.debug("Handling message actions"); - var interrupted = false; - for (var action : queuedActions) { - logger.debug("Executing action {}", action.getClass().getSimpleName()); - try { - action.execute(context); - } catch (Throwable e) { - if ((e instanceof AssertionError || e instanceof RuntimeException) - && e.getCause() instanceof InterruptedException) { - interrupted = true; - continue; - } - logger.warn("Message action failed.", e); - } - } - if (interrupted) { - Thread.currentThread().interrupt(); - } + return context.getReceiveHelper().hasCaughtUpWithOldMessages(); } @Override @@ -1250,91 +933,46 @@ public class ManagerImpl implements Manager { return identity == null ? List.of() : List.of(toIdentity(identity)); } - /** - * Trust this the identity with this fingerprint - * - * @param recipient account of the identity - * @param fingerprint Fingerprint - */ @Override public boolean trustIdentityVerified( RecipientIdentifier.Single recipient, byte[] fingerprint ) throws UnregisteredRecipientException { - RecipientId recipientId; - try { - recipientId = context.getRecipientHelper().resolveRecipient(recipient); - } catch (IOException e) { - return false; - } - final var updated = context.getIdentityHelper().trustIdentityVerified(recipientId, fingerprint); - if (updated && this.isReceiving()) { - needsToRetryFailedMessages = true; - } - return updated; + return trustIdentity(recipient, r -> context.getIdentityHelper().trustIdentityVerified(r, fingerprint)); } - /** - * Trust this the identity with this safety number - * - * @param recipient account of the identity - * @param safetyNumber Safety number - */ @Override public boolean trustIdentityVerifiedSafetyNumber( RecipientIdentifier.Single recipient, String safetyNumber ) throws UnregisteredRecipientException { - RecipientId recipientId; - try { - recipientId = context.getRecipientHelper().resolveRecipient(recipient); - } catch (IOException e) { - return false; - } - final var updated = context.getIdentityHelper().trustIdentityVerifiedSafetyNumber(recipientId, safetyNumber); - if (updated && this.isReceiving()) { - needsToRetryFailedMessages = true; - } - return updated; + return trustIdentity(recipient, + r -> context.getIdentityHelper().trustIdentityVerifiedSafetyNumber(r, safetyNumber)); } - /** - * Trust this the identity with this scannable safety number - * - * @param recipient account of the identity - * @param safetyNumber Scannable safety number - */ @Override public boolean trustIdentityVerifiedSafetyNumber( RecipientIdentifier.Single recipient, byte[] safetyNumber ) throws UnregisteredRecipientException { - RecipientId recipientId; - try { - recipientId = context.getRecipientHelper().resolveRecipient(recipient); - } catch (IOException e) { - return false; - } - final var updated = context.getIdentityHelper().trustIdentityVerifiedSafetyNumber(recipientId, safetyNumber); - if (updated && this.isReceiving()) { - needsToRetryFailedMessages = true; - } - return updated; + return trustIdentity(recipient, + r -> context.getIdentityHelper().trustIdentityVerifiedSafetyNumber(r, safetyNumber)); } - /** - * Trust all keys of this identity without verification - * - * @param recipient account of the identity - */ @Override public boolean trustIdentityAllKeys(RecipientIdentifier.Single recipient) throws UnregisteredRecipientException { + return trustIdentity(recipient, r -> context.getIdentityHelper().trustIdentityAllKeys(r)); + } + + private boolean trustIdentity( + RecipientIdentifier.Single recipient, Function trustMethod + ) throws UnregisteredRecipientException { RecipientId recipientId; try { recipientId = context.getRecipientHelper().resolveRecipient(recipient); } catch (IOException e) { return false; } - final var updated = context.getIdentityHelper().trustIdentityAllKeys(recipientId); + final var updated = trustMethod.apply(recipientId); if (updated && this.isReceiving()) { - needsToRetryFailedMessages = true; + context.getReceiveHelper().setNeedsToRetryFailedMessages(true); } return updated; } @@ -1347,7 +985,7 @@ public class ManagerImpl implements Manager { } @Override - public void close() throws IOException { + public void close() { Thread thread; synchronized (messageHandlers) { weakHandlers.clear(); @@ -1361,6 +999,7 @@ public class ManagerImpl implements Manager { executor.shutdown(); dependencies.getSignalWebSocket().disconnect(); + disposable.dispose(); synchronized (closedListeners) { closedListeners.forEach(Runnable::run);