// Copyright © 2022 Rangeproof Pty Ltd. All rights reserved. import Foundation import Combine import Sodium import GRDB import SessionUtilitiesKit public final class SnodeAPI { internal static let sodium: Atomic = Atomic(Sodium()) private static var hasLoadedSnodePool: Atomic = Atomic(false) private static var loadedSwarms: Atomic> = Atomic([]) private static var getSnodePoolPublisher: Atomic, Error>?> = Atomic(nil) /// - Note: Should only be accessed from `Threading.workQueue` to avoid race conditions. internal static var snodeFailureCount: Atomic<[Snode: UInt]> = Atomic([:]) /// - Note: Should only be accessed from `Threading.workQueue` to avoid race conditions. internal static var snodePool: Atomic> = Atomic([]) /// The offset between the user's clock and the Service Node's clock. Used in cases where the /// user's clock is incorrect. /// /// - Note: Should only be accessed from `Threading.workQueue` to avoid race conditions. public static var clockOffsetMs: Atomic = Atomic(0) /// - Note: Should only be accessed from `Threading.workQueue` to avoid race conditions. public static var swarmCache: Atomic<[String: Set]> = Atomic([:]) // MARK: - Hardfork version public static var hardfork = UserDefaults.standard[.hardfork] public static var softfork = UserDefaults.standard[.softfork] // MARK: - Settings internal static let maxRetryCount: Int = 8 private static let minSwarmSnodeCount: Int = 3 private static let seedNodePool: Set = { guard !Features.useTestnet else { return [ "http://public.loki.foundation:38157" ] } return [ "https://seed1.getsession.org:4432", "https://seed2.getsession.org:4432", "https://seed3.getsession.org:4432" ] }() private static let snodeFailureThreshold: Int = 3 private static let minSnodePoolCount: Int = 12 public static func currentOffsetTimestampMs(using dependencies: Dependencies = Dependencies()) -> Int64 { let clockOffsetMs: Int64 = SnodeAPI.clockOffsetMs.wrappedValue return (Int64(floor(dependencies.dateNow.timeIntervalSince1970 * 1000)) + clockOffsetMs) } // MARK: - Snode Pool Interaction private static var hasInsufficientSnodes: Bool { snodePool.wrappedValue.count < minSnodePoolCount } private static func loadSnodePoolIfNeeded( using dependencies: Dependencies = Dependencies() ) { guard !hasLoadedSnodePool.wrappedValue else { return } let fetchedSnodePool: Set = dependencies[singleton: .storage] .read { db in try Snode.fetchSet(db) } .defaulting(to: []) snodePool.mutate { $0 = fetchedSnodePool } hasLoadedSnodePool.mutate { $0 = true } } private static func setSnodePool( _ db: Database? = nil, to newValue: Set, using dependencies: Dependencies = Dependencies() ) { guard let db: Database = db else { dependencies[singleton: .storage].write { db in setSnodePool(db, to: newValue, using: dependencies) } return } snodePool.mutate { $0 = newValue } _ = try? Snode.deleteAll(db) newValue.forEach { try? $0.save(db) } } private static func dropSnodeFromSnodePool(_ snode: Snode) { var snodePool = SnodeAPI.snodePool.wrappedValue snodePool.remove(snode) setSnodePool(to: snodePool) } @objc public static func clearSnodePool() { snodePool.mutate { $0.removeAll() } Threading.workQueue.async { setSnodePool(to: []) } } // MARK: - Swarm Interaction private static func loadSwarmIfNeeded( for publicKey: String, using dependencies: Dependencies = Dependencies() ) { guard !loadedSwarms.wrappedValue.contains(publicKey) else { return } let updatedCacheForKey: Set = dependencies[singleton: .storage] .read { db in try Snode.fetchSet(db, publicKey: publicKey) } .defaulting(to: []) swarmCache.mutate { $0[publicKey] = updatedCacheForKey } loadedSwarms.mutate { $0.insert(publicKey) } } private static func setSwarm( to newValue: Set, for publicKey: String, persist: Bool = true, using dependencies: Dependencies ) { swarmCache.mutate { $0[publicKey] = newValue } guard persist else { return } dependencies[singleton: .storage].write { db in try? newValue.save(db, key: publicKey) } } public static func dropSnodeFromSwarmIfNeeded( _ snode: Snode, publicKey: String, using dependencies: Dependencies = Dependencies() ) { let swarmOrNil = swarmCache.wrappedValue[publicKey] guard var swarm = swarmOrNil, let index = swarm.firstIndex(of: snode) else { return } swarm.remove(at: index) setSwarm(to: swarm, for: publicKey, using: dependencies) } // MARK: - Snode API public static func hasCachedSnodesIncludingExpired( using dependencies: Dependencies = Dependencies() ) -> Bool { loadSnodePoolIfNeeded(using: dependencies) return !hasInsufficientSnodes } public static func getSnodePool( using dependencies: Dependencies = Dependencies() ) -> AnyPublisher, Error> { loadSnodePoolIfNeeded(using: dependencies) let now: Date = Date() let hasSnodePoolExpired: Bool = dependencies[singleton: .storage][.lastSnodePoolRefreshDate] .map { now.timeIntervalSince($0) > 2 * 60 * 60 } .defaulting(to: true) let snodePool: Set = SnodeAPI.snodePool.wrappedValue guard hasInsufficientSnodes || hasSnodePoolExpired else { return Just(snodePool) .setFailureType(to: Error.self) .eraseToAnyPublisher() } if let getSnodePoolPublisher: AnyPublisher, Error> = getSnodePoolPublisher.wrappedValue { return getSnodePoolPublisher } return getSnodePoolPublisher.mutate { result in /// It was possible for multiple threads to call this at the same time resulting in duplicate promises getting created, while /// this should no longer be possible (as the `wrappedValue` should now properly be blocked) this is a sanity check /// to make sure we don't create an additional promise when one already exists if let previouslyBlockedPublisher: AnyPublisher, Error> = result { return previouslyBlockedPublisher } let targetPublisher: AnyPublisher, Error> = { guard snodePool.count >= minSnodePoolCount else { return getSnodePoolFromSeedNode(using: dependencies) } return getSnodePoolFromSnode(using: dependencies) .catch { _ in getSnodePoolFromSeedNode(using: dependencies) } .eraseToAnyPublisher() }() /// Need to include the post-request code and a `shareReplay` within the publisher otherwise it can still be executed /// multiple times as a result of multiple subscribers let publisher: AnyPublisher, Error> = targetPublisher .tryFlatMap { snodePool -> AnyPublisher, Error> in guard !snodePool.isEmpty else { throw SnodeAPIError.snodePoolUpdatingFailed } return dependencies[singleton: .storage] .writePublisher { db in db[.lastSnodePoolRefreshDate] = now setSnodePool(db, to: snodePool, using: dependencies) return snodePool } .eraseToAnyPublisher() } .handleEvents( receiveCompletion: { _ in getSnodePoolPublisher.mutate { $0 = nil } } ) .shareReplay(1) .eraseToAnyPublisher() /// Actually assign the atomic value result = publisher return publisher } } public static func getSwarm( for publicKey: String, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher, Error> { loadSwarmIfNeeded(for: publicKey, using: dependencies) if let cachedSwarm = swarmCache.wrappedValue[publicKey], cachedSwarm.count >= minSwarmSnodeCount { return Just(cachedSwarm) .setFailureType(to: Error.self) .eraseToAnyPublisher() } let currentUserPublicKey: String = getUserHexEncodedPublicKey(using: dependencies) SNLog("Getting swarm for: \((publicKey == currentUserPublicKey) ? "self" : publicKey).") return getRandomSnode(using: dependencies) .flatMap { snode in SnodeAPI.send( request: SnodeRequest( endpoint: .getSwarm, body: GetSwarmRequest(pubkey: publicKey) ), to: snode, associatedWith: publicKey, using: dependencies ) .retry(4) .eraseToAnyPublisher() } .map { _, responseData in parseSnodes(from: responseData, using: dependencies) } .handleEvents( receiveOutput: { swarm in setSwarm(to: swarm, for: publicKey, using: dependencies) } ) .eraseToAnyPublisher() } // MARK: - Batching & Polling public typealias PollResponse = [SnodeAPI.Namespace: (info: ResponseInfoType, data: PreparedGetMessagesResponse?)] public static func preparedPoll( _ db: Database, namespaces: [SnodeAPI.Namespace], refreshingConfigHashes: [String] = [], from snode: Snode, authInfo: SnodeAPI.AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) throws -> HTTP.PreparedRequest { // Determine the maxSize each namespace in the request should take up var requests: [any ErasedPreparedRequest] = [] let namespaceMaxSizeMap: [SnodeAPI.Namespace: Int64] = SnodeAPI.Namespace.maxSizeMap(for: namespaces) let fallbackSize: Int64 = (namespaceMaxSizeMap.values.min() ?? 1) // If we have any config hashes to refresh TTLs then add those requests first if !refreshingConfigHashes.isEmpty { requests.append( try SnodeAPI.prepareRequest( request: Request( endpoint: .expire, publicKey: authInfo.publicKey, body: UpdateExpiryRequest( messageHashes: refreshingConfigHashes, expiryMs: UInt64( SnodeAPI.currentOffsetTimestampMs() + (30 * 24 * 60 * 60 * 1000) // 30 days ), extend: true, authInfo: authInfo ) ), responseType: UpdateExpiryResponse.self ) ) } // Add the various 'getMessages' requests requests.append( contentsOf: try namespaces.map { namespace -> any ErasedPreparedRequest in try SnodeAPI.preparedGetMessages( db, in: namespace, snode: snode, maxSize: namespaceMaxSizeMap[namespace] .defaulting(to: fallbackSize), authInfo: authInfo, using: dependencies ) } ) return try preparedBatch( db, requests: requests, requireAllBatchResponses: true, associatedWith: authInfo.publicKey, using: dependencies ) .map { (_: ResponseInfoType, batchResponse: HTTP.BatchResponse) -> [SnodeAPI.Namespace: (info: ResponseInfoType, data: PreparedGetMessagesResponse?)] in let messageResponses: [HTTP.BatchSubResponse] = batchResponse .compactMap { $0 as? HTTP.BatchSubResponse } /// Since we have extended the TTL for a number of messages we need to make sure we update the local /// `SnodeReceivedMessageInfo.expirationDateMs` values so we don't end up deleting them /// incorrectly before they actually expire on the swarm if !refreshingConfigHashes.isEmpty, let refreshTTLSubReponse: HTTP.BatchSubResponse = batchResponse .first(where: { $0 is HTTP.BatchSubResponse }) .asType(HTTP.BatchSubResponse.self), let refreshTTLResponse: UpdateExpiryResponse = refreshTTLSubReponse.body, // TODO: Need to fix this for the group config polling let validResults: [String: UpdateExpiryResponseResult] = try? refreshTTLResponse.validResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: getUserHexEncodedPublicKey(using: dependencies), validationData: refreshingConfigHashes ), let targetResult: UpdateExpiryResponseResult = validResults[snode.ed25519PublicKey], let groupedExpiryResult: [UInt64: [String]] = targetResult.changed .updated(with: targetResult.unchanged) .groupedByValue() .nullIfEmpty() { dependencies[singleton: .storage].writeAsync { db in try groupedExpiryResult.forEach { updatedExpiry, hashes in try SnodeReceivedMessageInfo .filter(hashes.contains(SnodeReceivedMessageInfo.Columns.hash)) .updateAll( db, SnodeReceivedMessageInfo.Columns.expirationDateMs .set(to: updatedExpiry) ) } } } return zip(namespaces, messageResponses) .reduce(into: [:]) { result, next in guard let messageResponse: PreparedGetMessagesResponse = next.1.body else { return } result[next.0] = (next.1, messageResponse) } } } private static func preparedBatch( _ db: Database, requests: [any ErasedPreparedRequest], requireAllBatchResponses: Bool, associatedWith publicKey: String, using dependencies: Dependencies ) throws -> HTTP.PreparedRequest { return try SnodeAPI .prepareRequest( request: Request( endpoint: .batch, publicKey: publicKey, body: HTTP.BatchRequest(requestsKey: .requests, requests: requests) ), responseType: HTTP.BatchResponse.self, requireAllBatchResponses: requireAllBatchResponses, using: dependencies ) } public static func preparedSequence( _ db: Database, requests: [any ErasedPreparedRequest], requireAllBatchResponses: Bool, associatedWith publicKey: String, using dependencies: Dependencies ) throws -> HTTP.PreparedRequest { return try SnodeAPI .prepareRequest( request: Request( endpoint: .sequence, publicKey: publicKey, body: HTTP.BatchRequest(requestsKey: .requests, requests: requests) ), responseType: HTTP.BatchResponse.self, requireAllBatchResponses: requireAllBatchResponses, using: dependencies ) } // MARK: - Retrieve public typealias PreparedGetMessagesResponse = (messages: [SnodeReceivedMessage], lastHash: String?) public static func preparedGetMessages( _ db: Database, in namespace: SnodeAPI.Namespace, snode: Snode, maxSize: Int64? = nil, authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) throws -> HTTP.PreparedRequest { // Prune expired message hashes for this namespace on this service node try SnodeReceivedMessageInfo.pruneExpiredMessageHashInfo( db, for: snode, namespace: namespace, associatedWith: authInfo.publicKey, using: dependencies ) let maybeLastHash: String? = try SnodeReceivedMessageInfo .fetchLastNotExpired( db, for: snode, namespace: namespace, associatedWith: authInfo.publicKey, using: dependencies )? .hash let preparedRequest: HTTP.PreparedRequest = try { // Check if this namespace requires authentication guard namespace.requiresReadAuthentication else { return try SnodeAPI.prepareRequest( request: Request( endpoint: .getMessages, publicKey: authInfo.publicKey, body: LegacyGetMessagesRequest( pubkey: authInfo.publicKey, lastHash: (maybeLastHash ?? ""), namespace: namespace, maxCount: nil, maxSize: maxSize ) ), responseType: GetMessagesResponse.self ) } return try SnodeAPI.prepareRequest( request: Request( endpoint: .getMessages, publicKey: authInfo.publicKey, body: GetMessagesRequest( lastHash: (maybeLastHash ?? ""), namespace: namespace, authInfo: authInfo, timestampMs: UInt64(SnodeAPI.currentOffsetTimestampMs(using: dependencies)), maxSize: maxSize ) ), responseType: GetMessagesResponse.self ) }() return preparedRequest .map { _, response -> (messages: [SnodeReceivedMessage], lastHash: String?) in return ( response.messages.compactMap { rawMessage -> SnodeReceivedMessage? in SnodeReceivedMessage( snode: snode, publicKey: authInfo.publicKey, namespace: namespace, rawMessage: rawMessage ) }, maybeLastHash ) } } public static func getSessionID( for onsName: String, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher { let validationCount = 3 // The name must be lowercased let onsName = onsName.lowercased() // Hash the ONS name using BLAKE2b let nameAsData = [UInt8](onsName.data(using: String.Encoding.utf8)!) guard let nameHash = sodium.wrappedValue.genericHash.hash(message: nameAsData) else { return Fail(error: SnodeAPIError.hashingFailed) .eraseToAnyPublisher() } // Ask 3 different snodes for the Session ID associated with the given name hash let base64EncodedNameHash = nameHash.toBase64() return Publishers .MergeMany( (0.. AnyPublisher in SnodeAPI .send( request: SnodeRequest( endpoint: .oxenDaemonRPCCall, body: OxenDaemonRPCRequest( endpoint: .daemonOnsResolve, body: ONSResolveRequest( type: 0, // type 0 means Session base64EncodedNameHash: base64EncodedNameHash ) ) ), to: snode, associatedWith: nil, using: dependencies ) .decoded(as: ONSResolveResponse.self) .tryMap { _, response -> String in try response.sessionId( sodium: sodium.wrappedValue, nameBytes: nameAsData, nameHashBytes: nameHash ) } .retry(4) .eraseToAnyPublisher() } } ) .collect() .tryMap { results -> String in guard results.count == validationCount, Set(results).count == 1 else { throw SnodeAPIError.validationFailed } return results[0] } .eraseToAnyPublisher() } public static func getExpiries( from snode: Snode, of serverHashes: [String], authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher<(ResponseInfoType, GetExpiriesResponse), Error> { let sendTimestamp: UInt64 = UInt64(SnodeAPI.currentOffsetTimestampMs(using: dependencies)) // FIXME: There is a bug on SS now that a single-hash lookup is not working. Remove it when the bug is fixed let serverHashes: [String] = serverHashes.appending("fakehash") return SnodeAPI .send( request: SnodeRequest( endpoint: .getExpiries, body: GetExpiriesRequest( messageHashes: serverHashes, authInfo: authInfo, timestampMs: sendTimestamp ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: GetExpiriesResponse.self, using: dependencies) .eraseToAnyPublisher() } // MARK: - Store public static func sendMessage( _ message: SnodeMessage, in namespace: Namespace, authInfo: AuthenticationInfo, using dependencies: Dependencies ) -> AnyPublisher<(ResponseInfoType, SendMessagesResponse), Error> { let userX25519PublicKey: String = getUserHexEncodedPublicKey(using: dependencies) let sendTimestamp: UInt64 = UInt64(SnodeAPI.currentOffsetTimestampMs(using: dependencies)) // Create a convenience method to send a message to an individual Snode func sendMessage(to snode: Snode) throws -> AnyPublisher<(any ResponseInfoType, SendMessagesResponse), Error> { guard namespace.requiresWriteAuthentication else { return SnodeAPI .send( request: SnodeRequest( endpoint: .sendMessage, body: LegacySendMessagesRequest( message: message, namespace: namespace ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: SendMessagesResponse.self, using: dependencies) .eraseToAnyPublisher() } return SnodeAPI .send( request: SnodeRequest( endpoint: .sendMessage, body: SendMessageRequest( message: message, namespace: namespace, authInfo: authInfo, timestampMs: sendTimestamp ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: SendMessagesResponse.self, using: dependencies) .eraseToAnyPublisher() } return getSwarm(for: authInfo.publicKey) .tryFlatMapWithRandomSnode(retry: maxRetryCount) { snode -> AnyPublisher<(ResponseInfoType, SendMessagesResponse), Error> in try sendMessage(to: snode) .tryMap { info, response -> (ResponseInfoType, SendMessagesResponse) in try response.validateResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: userX25519PublicKey ) return (info, response) } .eraseToAnyPublisher() } } public static func preparedSendMessage( _ db: Database, message: SnodeMessage, in namespace: Namespace, authInfo: AuthenticationInfo, using dependencies: Dependencies ) throws -> HTTP.PreparedRequest { let userX25519PublicKey: String = getUserHexEncodedPublicKey(db, using: dependencies) let request: HTTP.PreparedRequest = try { // Check if this namespace requires authentication guard namespace.requiresWriteAuthentication else { return try SnodeAPI.prepareRequest( request: Request( endpoint: .sendMessage, publicKey: authInfo.publicKey, body: LegacySendMessagesRequest( message: message, namespace: namespace ) ), responseType: SendMessagesResponse.self ) } return try SnodeAPI.prepareRequest( request: Request( endpoint: .sendMessage, publicKey: authInfo.publicKey, body: SendMessageRequest( message: message, namespace: namespace, authInfo: authInfo, timestampMs: UInt64(SnodeAPI.currentOffsetTimestampMs(using: dependencies)) ) ), responseType: SendMessagesResponse.self ) }() return request .tryMap { _, response -> SendMessagesResponse in try response.validateResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: userX25519PublicKey ) return response } } // MARK: - Edit public static func updateExpiry( serverHashes: [String], updatedExpiryMs: Int64, shortenOnly: Bool? = nil, extendOnly: Bool? = nil, authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher<[String: UpdateExpiryResponseResult], Error> { // ShortenOnly and extendOnly cannot be true at the same time guard shortenOnly == nil || extendOnly == nil else { return Fail(error: SnodeAPIError.generic) .eraseToAnyPublisher() } return getSwarm(for: authInfo.publicKey) .tryFlatMapWithRandomSnode(retry: maxRetryCount) { snode -> AnyPublisher<[String: UpdateExpiryResponseResult], Error> in SnodeAPI .send( request: SnodeRequest( endpoint: .expire, body: UpdateExpiryRequest( messageHashes: serverHashes, expiryMs: UInt64(updatedExpiryMs), shorten: shortenOnly, extend: extendOnly, authInfo: authInfo ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: UpdateExpiryResponse.self, using: dependencies) .tryMap { _, response -> [String: UpdateExpiryResponseResult] in try response.validResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: getUserHexEncodedPublicKey(using: dependencies), validationData: serverHashes ) } .eraseToAnyPublisher() } } public static func revokeSubkey( subkeyToRevoke: String, authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher { return getSwarm(for: authInfo.publicKey) .tryFlatMapWithRandomSnode(retry: maxRetryCount) { snode -> AnyPublisher in SnodeAPI .send( request: SnodeRequest( endpoint: .revokeSubkey, body: RevokeSubkeyRequest( subkeyToRevoke: subkeyToRevoke, authInfo: authInfo ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: RevokeSubkeyResponse.self, using: dependencies) .tryMap { _, response -> Void in try response.validateResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: getUserHexEncodedPublicKey(using: dependencies), validationData: subkeyToRevoke ) return () } .eraseToAnyPublisher() } } // MARK: Delete public static func deleteMessages( serverHashes: [String], authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher<[String: Bool], Error> { let userX25519PublicKey: String = getUserHexEncodedPublicKey(using: dependencies) return getSwarm(for: authInfo.publicKey, using: dependencies) .tryFlatMapWithRandomSnode(retry: maxRetryCount) { snode -> AnyPublisher<[String: Bool], Error> in SnodeAPI .send( request: SnodeRequest( endpoint: .deleteMessages, body: DeleteMessagesRequest( messageHashes: serverHashes, requireSuccessfulDeletion: false, authInfo: authInfo ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: DeleteMessagesResponse.self, using: dependencies) .tryMap { _, response -> [String: Bool] in let validResultMap: [String: Bool] = try response.validResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: userX25519PublicKey, validationData: serverHashes ) // If `validResultMap` didn't throw then at least one service node // deleted successfully so we should mark the hash as invalid so we // don't try to fetch updates using that hash going forward (if we // do we would end up re-fetching all old messages) dependencies[singleton: .storage].writeAsync { db in try? SnodeReceivedMessageInfo.handlePotentialDeletedOrInvalidHash( db, potentiallyInvalidHashes: serverHashes ) } return validResultMap } .eraseToAnyPublisher() } } public static func preparedDeleteMessages( serverHashes: [String], requireSuccessfulDeletion: Bool, authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) throws -> HTTP.PreparedRequest<[String: Bool]> { let userX25519PublicKey: String = getUserHexEncodedPublicKey(using: dependencies) return try SnodeAPI .prepareRequest( request: Request( endpoint: .deleteMessages, publicKey: authInfo.publicKey, body: DeleteMessagesRequest( messageHashes: serverHashes, requireSuccessfulDeletion: requireSuccessfulDeletion, authInfo: authInfo ) ), responseType: DeleteMessagesResponse.self ) .tryMap { _, response -> [String: Bool] in let validResultMap: [String: Bool] = try response.validResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: userX25519PublicKey, validationData: serverHashes ) // If `validResultMap` didn't throw then at least one service node // deleted successfully so we should mark the hash as invalid so we // don't try to fetch updates using that hash going forward (if we // do we would end up re-fetching all old messages) dependencies[singleton: .storage].writeAsync { db in try? SnodeReceivedMessageInfo.handlePotentialDeletedOrInvalidHash( db, potentiallyInvalidHashes: serverHashes ) } return validResultMap } } /// Clears all the user's data from their swarm. Returns a dictionary of snode public key to deletion confirmation. public static func deleteAllMessages( namespace: SnodeAPI.Namespace, authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher<[String: Bool], Error> { let userX25519PublicKey: String = getUserHexEncodedPublicKey(using: dependencies) return getSwarm(for: authInfo.publicKey, using: dependencies) .tryFlatMapWithRandomSnode(retry: maxRetryCount) { snode -> AnyPublisher<[String: Bool], Error> in getNetworkTime(from: snode) .flatMap { timestampMs -> AnyPublisher<[String: Bool], Error> in SnodeAPI .send( request: SnodeRequest( endpoint: .deleteAll, body: DeleteAllMessagesRequest( namespace: namespace, authInfo: authInfo, timestampMs: timestampMs ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: DeleteAllMessagesResponse.self, using: dependencies) .tryMap { _, response -> [String: Bool] in try response.validResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: userX25519PublicKey, validationData: timestampMs ) } .eraseToAnyPublisher() } .eraseToAnyPublisher() } } /// Clears all the user's data from their swarm. Returns a dictionary of snode public key to deletion confirmation. public static func deleteAllMessages( beforeMs: UInt64, namespace: SnodeAPI.Namespace, authInfo: AuthenticationInfo, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher<[String: Bool], Error> { let userX25519PublicKey: String = getUserHexEncodedPublicKey(using: dependencies) return getSwarm(for: authInfo.publicKey, using: dependencies) .tryFlatMapWithRandomSnode(retry: maxRetryCount) { snode -> AnyPublisher<[String: Bool], Error> in getNetworkTime(from: snode) .flatMap { timestampMs -> AnyPublisher<[String: Bool], Error> in SnodeAPI .send( request: SnodeRequest( endpoint: .deleteAllBefore, body: DeleteAllBeforeRequest( beforeMs: beforeMs, namespace: namespace, authInfo: authInfo, timestampMs: timestampMs ) ), to: snode, associatedWith: authInfo.publicKey, using: dependencies ) .decoded(as: DeleteAllBeforeResponse.self, using: dependencies) .tryMap { _, response -> [String: Bool] in try response.validResultMap( sodium: sodium.wrappedValue, userX25519PublicKey: userX25519PublicKey, validationData: beforeMs ) } .eraseToAnyPublisher() } .eraseToAnyPublisher() } } // MARK: - Internal API public static func getNetworkTime( from snode: Snode, using dependencies: Dependencies = Dependencies() ) -> AnyPublisher { return SnodeAPI .send( request: SnodeRequest<[String: String]>( endpoint: .getInfo, body: [:] ), to: snode, associatedWith: nil, using: dependencies ) .decoded(as: GetNetworkTimestampResponse.self, using: dependencies) .map { _, response in // Assume we've fetched the networkTime in order to send a message to the specified snode, in // which case we want to update the 'clockOffsetMs' value for subsequent requests let offset = (Int64(response.timestamp) - Int64(floor(dependencies.dateNow.timeIntervalSince1970 * 1000))) SnodeAPI.clockOffsetMs.mutate { $0 = offset } return response.timestamp } .eraseToAnyPublisher() } public static func preparedGetNetworkTime( from snode: Snode, using dependencies: Dependencies = Dependencies() ) throws -> HTTP.PreparedRequest { return try SnodeAPI .prepareRequest( request: Request, Endpoint>( endpoint: .getInfo, publicKey: snode.x25519PublicKey, body: [:] ), responseType: GetNetworkTimestampResponse.self ) .map { _, response in // Assume we've fetched the networkTime in order to send a message to the specified snode, in // which case we want to update the 'clockOffsetMs' value for subsequent requests let offset = (Int64(response.timestamp) - Int64(floor(dependencies.dateNow.timeIntervalSince1970 * 1000))) SnodeAPI.clockOffsetMs.mutate { $0 = offset } return response.timestamp } } internal static func getRandomSnode( using dependencies: Dependencies ) -> AnyPublisher { // randomElement() uses the system's default random generator, which is cryptographically secure return getSnodePool(using: dependencies) .map { $0.randomElement()! } .eraseToAnyPublisher() } private static func getSnodePoolFromSeedNode( using dependencies: Dependencies ) -> AnyPublisher, Error> { let request: SnodeRequest = SnodeRequest( endpoint: .jsonGetNServiceNodes, body: GetServiceNodesRequest( activeOnly: true, limit: 256, fields: GetServiceNodesRequest.Fields( publicIp: true, storagePort: true, pubkeyEd25519: true, pubkeyX25519: true ) ) ) guard let target: String = seedNodePool.randomElement() else { return Fail(error: SnodeAPIError.snodePoolUpdatingFailed) .eraseToAnyPublisher() } guard let payload: Data = try? JSONEncoder(using: dependencies).encode(request) else { return Fail(error: HTTPError.invalidJSON) .eraseToAnyPublisher() } SNLog("Populating snode pool using seed node: \(target).") return HTTP .execute( .post, "\(target)/json_rpc", body: payload, useSeedNodeURLSession: true ) .decoded(as: SnodePoolResponse.self, using: dependencies) .mapError { error in switch error { case HTTPError.parsingFailed: return SnodeAPIError.snodePoolUpdatingFailed default: return error } } .map { snodePool -> Set in snodePool.result .serviceNodeStates .compactMap { $0.value } .asSet() } .retry(2) .handleEvents( receiveCompletion: { result in switch result { case .finished: SNLog("Got snode pool from seed node: \(target).") case .failure: SNLog("Failed to contact seed node at: \(target).") } } ) .eraseToAnyPublisher() } private static func getSnodePoolFromSnode( using dependencies: Dependencies ) -> AnyPublisher, Error> { var snodePool = SnodeAPI.snodePool.wrappedValue var snodes: Set = [] (0..<3).forEach { _ in guard let snode = snodePool.randomElement() else { return } snodePool.remove(snode) snodes.insert(snode) } return Publishers .MergeMany( snodes .map { snode -> AnyPublisher, Error> in // Don't specify a limit in the request. Service nodes return a shuffled // list of nodes so if we specify a limit the 3 responses we get might have // very little overlap. SnodeAPI .send( request: SnodeRequest( endpoint: .oxenDaemonRPCCall, body: OxenDaemonRPCRequest( endpoint: .daemonGetServiceNodes, body: GetServiceNodesRequest( activeOnly: true, limit: nil, fields: GetServiceNodesRequest.Fields( publicIp: true, storagePort: true, pubkeyEd25519: true, pubkeyX25519: true ) ) ) ), to: snode, associatedWith: nil, using: dependencies ) .decoded(as: SnodePoolResponse.self, using: dependencies) .mapError { error -> Error in switch error { case HTTPError.parsingFailed: return SnodeAPIError.snodePoolUpdatingFailed default: return error } } .map { _, snodePool -> Set in snodePool.result .serviceNodeStates .compactMap { $0.value } .asSet() } .retry(4) .eraseToAnyPublisher() } ) .collect() .tryMap { results -> Set in let result: Set = results.reduce(Set()) { prev, next in prev.intersection(next) } // We want the snodes to agree on at least this many snodes guard result.count > 24 else { throw SnodeAPIError.inconsistentSnodePools } // Limit the snode pool size to 256 so that we don't go too long without // refreshing it return Set(result.prefix(256)) } .eraseToAnyPublisher() } private static func send( request: SnodeRequest, to snode: Snode, associatedWith publicKey: String?, using dependencies: Dependencies ) -> AnyPublisher<(ResponseInfoType, Data?), Error> { guard let payload: Data = try? JSONEncoder(using: dependencies).encode(request) else { return Fail(error: HTTPError.invalidJSON) .eraseToAnyPublisher() } guard Features.useOnionRequests else { return HTTP .execute( .post, "\(snode.address):\(snode.port)/storage_rpc/v1", body: payload ) .map { response in (HTTP.ResponseInfo(code: -1, headers: [:]), response) } .mapError { error in switch error { case HTTPError.httpRequestFailed(let statusCode, let data): return SnodeAPI .handleError( withStatusCode: statusCode, data: data, forSnode: snode, associatedWith: publicKey, using: dependencies ) .defaulting(to: error) default: return error } } .eraseToAnyPublisher() } return dependencies[singleton: .network] .send(.onionRequest(payload, to: snode)) .mapError { error in switch error { case HTTPError.httpRequestFailed(let statusCode, let data): return SnodeAPI .handleError( withStatusCode: statusCode, data: data, forSnode: snode, associatedWith: publicKey, using: dependencies ) .defaulting(to: error) default: return error } } .handleEvents( receiveOutput: { _, maybeData in // Extract and store hard fork information if returned guard let data: Data = maybeData, let snodeResponse: SnodeResponse = try? JSONDecoder(using: dependencies) .decode(SnodeResponse.self, from: data) else { return } if snodeResponse.hardFork[1] > softfork { softfork = snodeResponse.hardFork[1] UserDefaults.standard[.softfork] = softfork } if snodeResponse.hardFork[0] > hardfork { hardfork = snodeResponse.hardFork[0] UserDefaults.standard[.hardfork] = hardfork softfork = snodeResponse.hardFork[1] UserDefaults.standard[.softfork] = softfork } } ) .eraseToAnyPublisher() } // MARK: - Parsing // The parsing utilities below use a best attempt approach to parsing; they warn for parsing // failures but don't throw exceptions. private static func parseSnodes( from responseData: Data?, using dependencies: Dependencies ) -> Set { guard let responseData: Data = responseData, let responseJson: JSON = try? JSONSerialization.jsonObject( with: responseData, options: [ .fragmentsAllowed ] ) as? JSON else { SNLog("Failed to parse snodes from response data.") return [] } guard let rawSnodes = responseJson["snodes"] as? [JSON] else { SNLog("Failed to parse snodes from: \(responseJson).") return [] } guard let snodeData: Data = try? JSONSerialization.data(withJSONObject: rawSnodes, options: []) else { return [] } // FIXME: Hopefully at some point this different Snode structure will be deprecated and can be removed if let swarmSnodes: [SwarmSnode] = try? JSONDecoder(using: dependencies) .decode([Failable].self, from: snodeData) .compactMap({ $0.value }), !swarmSnodes.isEmpty { return swarmSnodes.map { $0.toSnode() }.asSet() } return ((try? JSONDecoder(using: dependencies).decode([Failable].self, from: snodeData)) ?? []) .compactMap { $0.value } .asSet() } // MARK: - Error Handling @discardableResult internal static func handleError( withStatusCode statusCode: UInt, data: Data?, forSnode snode: Snode, associatedWith publicKey: String? = nil, using dependencies: Dependencies ) -> Error? { func handleBadSnode(using dependencies: Dependencies) { let oldFailureCount = (SnodeAPI.snodeFailureCount.wrappedValue[snode] ?? 0) let newFailureCount = oldFailureCount + 1 SnodeAPI.snodeFailureCount.mutate { $0[snode] = newFailureCount } SNLog("Couldn't reach snode at: \(snode); setting failure count to \(newFailureCount).") if newFailureCount >= SnodeAPI.snodeFailureThreshold { SNLog("Failure threshold reached for: \(snode); dropping it.") if let publicKey = publicKey { SnodeAPI.dropSnodeFromSwarmIfNeeded(snode, publicKey: publicKey, using: dependencies) } SnodeAPI.dropSnodeFromSnodePool(snode) SNLog("Snode pool count: \(snodePool.wrappedValue.count).") SnodeAPI.snodeFailureCount.mutate { $0[snode] = 0 } } } switch statusCode { case 500, 502, 503: // The snode is unreachable handleBadSnode(using: dependencies) case 404: // May caused by invalid open groups SNLog("Can't reach the server.") case 406: SNLog("The user's clock is out of sync with the service node network.") return SnodeAPIError.clockOutOfSync case 421: // The snode isn't associated with the given public key anymore if let publicKey = publicKey { func invalidateSwarm() { SNLog("Invalidating swarm for: \(publicKey).") SnodeAPI.dropSnodeFromSwarmIfNeeded(snode, publicKey: publicKey, using: dependencies) } if let data: Data = data { let snodes = parseSnodes(from: data, using: dependencies) if !snodes.isEmpty { setSwarm(to: snodes, for: publicKey, using: dependencies) } else { invalidateSwarm() } } else { invalidateSwarm() } } else { SNLog("Got a 421 without an associated public key.") } default: handleBadSnode(using: dependencies) let message: String = { if let data: Data = data, let stringFromData = String(data: data, encoding: .utf8) { return stringFromData } return "Empty data." }() SNLog("Unhandled response code: \(statusCode), messasge: \(message)") } return nil } // MARK: - Convenience private static func prepareRequest( request: Request, responseType: R.Type, requireAllBatchResponses: Bool = true, retryCount: Int = 0, timeout: TimeInterval = HTTP.defaultTimeout, using dependencies: Dependencies = Dependencies() ) throws -> HTTP.PreparedRequest { return HTTP.PreparedRequest( request: request, urlRequest: try request.generateUrlRequest(using: dependencies), responseType: responseType, requireAllBatchResponses: requireAllBatchResponses, retryCount: retryCount, timeout: timeout ) } } @objc(SNSnodeAPI) public final class SNSnodeAPI: NSObject { @objc(currentOffsetTimestampMs) public static func currentOffsetTimestampMs() -> UInt64 { return UInt64(SnodeAPI.currentOffsetTimestampMs()) } } // MARK: - Publisher Convenience public extension Publisher where Output == Set { func tryFlatMapWithRandomSnode( maxPublishers: Subscribers.Demand = .unlimited, retry retries: Int = 0, drainBehaviour: Atomic = .alwaysRandom, _ transform: @escaping (Snode) throws -> P ) -> AnyPublisher where T == P.Output, P: Publisher, P.Failure == Error { return self .mapError { $0 } .flatMap(maxPublishers: maxPublishers) { swarm -> AnyPublisher in // If we don't want to reuse a specific snode multiple times then just grab a // random one from the swarm every time var remainingSnodes: Set = { switch drainBehaviour.wrappedValue { case .alwaysRandom: return swarm case .limitedReuse(_, let targetSnode, _, let usedSnodes): // If we've used all of the snodes then reset the used list guard targetSnode != nil || usedSnodes != swarm else { drainBehaviour.mutate { $0 = $0.reset() } return swarm } return swarm.subtracting(usedSnodes) } }() return Just(()) .setFailureType(to: Error.self) .tryFlatMap(maxPublishers: maxPublishers) { _ -> AnyPublisher in let snode: Snode = try { switch drainBehaviour.wrappedValue { case .limitedReuse(_, .some(let targetSnode), _, _): return targetSnode default: break } // Select the next snode return try remainingSnodes.popRandomElement() ?? { throw SnodeAPIError.generic }() }() drainBehaviour.mutate { $0 = $0.use(snode: snode) } return try transform(snode) .eraseToAnyPublisher() } .retry(retries) .eraseToAnyPublisher() } .eraseToAnyPublisher() } } // MARK: - Snode Convenience private extension Snode { var server: String { "\(address):\(port)" } var urlString: String { "\(address):\(port)/storage_rpc/v1" } } // MARK: - Request Convenience private extension Request { init( endpoint: SnodeAPI.Endpoint, publicKey: String, body: B ) where T == SnodeRequest, Endpoint == SnodeAPI.Endpoint { self = Request( method: .post, endpoint: endpoint, publicKey: publicKey, body: SnodeRequest( endpoint: endpoint, body: body ) ) } }