native macOS codings agent orchestrator prowl.onev.cat
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248import Foundationimport Networkimport Security
@MainActorfinal class MirrorConnection { let id = UUID() let connection: NWConnection var onMessage: ((MirrorMessage) -> Void)? var onReady: (() -> Void)? var onHandshakeFailure: (() -> Void)? var onClose: ((String?) -> Void)? /// Set when the connection ends before it became ready. private(set) var failure: MirrorConnectionFailure? private var closed = false private var finishing = false private var queuedBytes = 0 private var heartbeat: Task<Void, Never>? private var deadline: Task<Void, Never>? private let clock: any Clock<Duration> private var becameReady = false private let handshakeTimeout: Duration
init( _ connection: NWConnection, clock: any Clock<Duration> = ContinuousClock(), handshakeTimeout: Duration = .seconds(30) ) { self.connection = connection self.clock = clock self.handshakeTimeout = handshakeTimeout }
static func parameters(pairingKey: String) throws -> NWParameters { let code = try MirrorPairingCode.normalized(pairingKey) return try parameters(keys: [("pair", Data(code.utf8))]) }
static func parameters(keys: [(String, Data)]) throws -> NWParameters { let tls = NWProtocolTLS.Options() for (name, secret) in keys { let identity = Data(name.utf8) secret.withUnsafeBytes { keyBytes in identity.withUnsafeBytes { identityBytes in sec_protocol_options_add_pre_shared_key( tls.securityProtocolOptions, DispatchData(bytes: keyBytes) as __DispatchData, DispatchData(bytes: identityBytes) as __DispatchData) } } } sec_protocol_options_set_min_tls_protocol_version(tls.securityProtocolOptions, .TLSv12) sec_protocol_options_set_max_tls_protocol_version(tls.securityProtocolOptions, .TLSv12) // Security's Swift enum omits the PSK suites supported by Network.framework. guard let suite = tls_ciphersuite_t( rawValue: UInt16(TLS_ECDHE_PSK_WITH_CHACHA20_POLY1305_SHA256) ) else { throw MirrorProtocolError.invalidMessage } sec_protocol_options_append_tls_ciphersuite(tls.securityProtocolOptions, suite) let tcp = NWProtocolTCP.Options() tcp.enableKeepalive = true tcp.keepaliveIdle = 15 let parameters = NWParameters(tls: tls, tcp: tcp) parameters.allowLocalEndpointReuse = true return parameters }
func start() { connection.stateUpdateHandler = { [weak self] state in Task { @MainActor in guard let self, !self.closed else { return } switch state { case .ready: self.becameReady = true self.resetDeadline() self.onReady?() guard !self.closed else { return } self.readHeader() let clock = self.clock self.heartbeat = Task { [weak self] in while !Task.isCancelled { do { try await clock.sleep(for: .seconds(2)) } catch { return } self?.send(.ping) } } case .failed(let error): if !self.becameReady { self.onHandshakeFailure?() self.failure = MirrorConnectionFailure(error) } self.close(error.localizedDescription) case .cancelled: self.close(nil) case .waiting(let error): if self.becameReady { self.close(String(localized: "Connection lost: \(error.localizedDescription)")) } else { // A refused port or rejected TLS key reports as waiting; there is nothing to wait for. self.onHandshakeFailure?() self.failure = MirrorConnectionFailure(error) self.close(error.localizedDescription) } default: break } } } resetDeadline() connection.start(queue: .main) }
func send(_ message: MirrorMessage, closeAfterSending: Bool = false) { guard !closed, !finishing else { return } do { let bytes = try MirrorWire.encode(message) guard queuedBytes + bytes.count <= 2 * MirrorWire.maximumPayload else { close(String(localized: "Remote receiver is too slow.")) return } queuedBytes += bytes.count if closeAfterSending { finishing = true heartbeat?.cancel() } connection.send( content: bytes, completion: .contentProcessed { [self] error in Task { @MainActor in guard !self.closed else { return } self.queuedBytes -= bytes.count if let error { self.close(error.localizedDescription) } else if closeAfterSending { self.close() } } }) } catch { close(error.localizedDescription) } }
func close(_ reason: String? = nil) { guard !closed else { return } #if DEBUG ProwlLogger("MirrorPairing").notice("Connection \(id) closed: \(reason ?? "no reason")") #endif closed = true deadline?.cancel() heartbeat?.cancel() connection.stateUpdateHandler = nil connection.cancel() let callback = onClose onMessage = nil onReady = nil onHandshakeFailure = nil onClose = nil callback?(reason) }
private func resetDeadline() { deadline?.cancel() let clock = clock let timeout: Duration = becameReady ? .seconds(8) : handshakeTimeout deadline = Task { [weak self] in do { try await clock.sleep(for: timeout) } catch { return } guard let self else { return } if !self.becameReady { self.failure = .timedOut } self.close( String(localized: "Remote connection timed out. The other side is no longer responding.") ) } }
private func readHeader() { read(count: 4) { [weak self] header in guard let self else { return } do { let length = try MirrorWire.length(header) self.read(count: length) { [weak self] payload in guard let self else { return } do { let message = try MirrorWire.decode(payload) self.receive(message) if !self.closed { self.readHeader() } } catch { self.close(error.localizedDescription) } } } catch { self.close(error.localizedDescription) } } }
func receive(_ message: MirrorMessage) { // A revoked lease can still have input in flight. Do not let its rejection // cancel the final status message while that message is draining. guard !closed, !finishing else { return } resetDeadline() if message.kind == .ping { send(.pong) } else if message.kind != .pong { onMessage?(message) } }
private func read(count: Int, completion: @escaping @MainActor (Data) -> Void) { connection.receive(minimumIncompleteLength: count, maximumLength: count) { [weak self] data, _, done, error in Task { @MainActor in guard let self, !self.closed else { return } guard let data, data.count == count, error == nil else { let fallback = done ? String(localized: "Host disconnected.") : String(localized: "Incomplete remote message.") self.close(error?.localizedDescription ?? fallback) return } completion(data) } } }}
nonisolated enum MirrorPairingCode { static let alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
static func normalized(_ input: String) throws -> String { let trimmed = input.trimmingCharacters(in: .whitespacesAndNewlines) let code = trimmed.uppercased().filter { $0 != "-" && !$0.isWhitespace } guard code.count == 8, code.allSatisfy({ alphabet.contains($0) }) else { throw MirrorProtocolError.invalidPairingKey } return code }
/// Live formatting for a code field: uppercase, at most eight symbols, a hyphen after four. static func formatted(_ input: String) -> String { let symbols = input.uppercased().filter { ($0.isLetter || $0.isNumber) && $0.isASCII }.prefix(8) guard symbols.count > 4 else { return String(symbols) } return String(symbols.prefix(4)) + "-" + String(symbols.dropFirst(4)) }
static func generate() throws -> String { var bytes = [UInt8](repeating: 0, count: 8) guard SecRandomCopyBytes(kSecRandomDefault, bytes.count, &bytes) == errSecSuccess else { throw MirrorProtocolError.invalidPairingKey } let symbols = Array(alphabet) let code = String(bytes.map { symbols[Int($0 & 31)] }) return String(code.prefix(4)) + "-" + code.suffix(4) }}