native macOS codings agent orchestrator prowl.onev.cat
Something went wrong. Try again.
37 kB · 918 lines
Swift
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919import Foundationimport Networkimport Observationimport Security
@MainActor@Observablefinal class MirrorHost { private(set) var isRunning = false private(set) var isStarting = false private(set) var error: String? private(set) var pairingKey = "" private(set) var subscriberCount = 0 { // Subscription membership changes publish count, including takeovers with an unchanged count. didSet { updateDeviceActivity() } } private var devicePaneIDs: [UUID: Set<UUID>] = [:] var address: String var port: String @ObservationIgnored var commandService: MirrorCommandService? @ObservationIgnored private var commandPeers: [UUID: MirrorCommandRequest] = [:] @ObservationIgnored private var commandDevices: [UUID: UUID] = [:] var onStarted: (() -> Void)? var onStopped: (() -> Void)? @ObservationIgnored private let enabled: Bool @ObservationIgnored private let source: any MirrorPaneSource @ObservationIgnored private let defaults: UserDefaults @ObservationIgnored private var listener: NWListener? @ObservationIgnored private var retiringListener: NWListener? @ObservationIgnored private var peers: [UUID: MirrorConnection] = [:] @ObservationIgnored private var pendingPeers: [UUID: MirrorConnection] = [:] @ObservationIgnored private var pendingOrder: [UUID] = [] private(set) var pendingHandshakeCount = 0 static let maximumPendingHandshakes = 8 @ObservationIgnored private var subscriptions: [UUID: Subscription] = [:] @ObservationIgnored private var pollTask: Task<Void, Never>? private(set) var hostRunID: UUID? @ObservationIgnored private var connectionAttempts = MirrorConnectionAttempts()
private(set) var onlineDeviceIDs: Set<UUID> = [] private(set) var devices: [MirrorPairedDevice] = [] /// The device that completed the current pairing window; cleared when a new code is issued. private(set) var lastPairedDevice: MirrorPairedDevice? private(set) var pairingExpiresAt: Date? @ObservationIgnored private var identity: MirrorHostIdentity? @ObservationIgnored private var pairingTask: Task<Void, Never>? @ObservationIgnored private var challenges: [UUID: Data] = [:] @ObservationIgnored private var pendingPairings: [UUID: MirrorDeviceCredential] = [:] @ObservationIgnored private var devicePeers: [UUID: UUID] = [:] @ObservationIgnored private var authenticationDeadlines: [UUID: Task<Void, Never>] = [:] @ObservationIgnored private let clock: any Clock<Duration> @ObservationIgnored private let loadIdentity: () throws -> MirrorHostIdentity? @ObservationIgnored private let saveIdentity: (MirrorHostIdentity) throws -> Void
func isOnline(_ device: UUID) -> Bool { onlineDeviceIDs.contains(device) }
func mirroredPanes(for deviceID: UUID) -> [MirrorPaneDescriptor] { guard let ids = devicePaneIDs[deviceID], !ids.isEmpty else { return [] } return source.panes().filter { ids.contains($0.id) } }
func subscriptionID(for paneID: UUID, deviceID: UUID) -> UUID? { subscriptions.first { peerID, subscription in devicePeers[peerID] == deviceID && subscription.paneID == paneID }?.value.id }
func disconnect(subscriptionID: UUID) { // Bind confirmation to the original lease, not a replacement on the same pane. guard let (peerID, _) = subscriptions.first(where: { $0.value.id == subscriptionID }), let peer = peers[peerID] else { return } subscriptions.removeValue(forKey: peerID) subscriberCount = subscriptions.count if subscriptions.isEmpty { pollTask?.cancel() pollTask = nil } cancelCommand(peerID) peer.onMessage = nil peer.send( .failure(.init(error: "This mirror was disconnected by Host.", subscriptionID: subscriptionID)), closeAfterSending: true) }
private func updateDeviceActivity() { var next: [UUID: Set<UUID>] = [:] for (peerID, subscription) in subscriptions { guard let deviceID = devicePeers[peerID] else { continue } next[deviceID, default: []].insert(subscription.paneID) } if next != devicePaneIDs { devicePaneIDs = next } }
func cancelPairing() { lastPairedDevice = nil guard pairingExpiresAt != nil else { return } expirePairing() }
func addDevice() { guard isRunning, !isStarting else { return } do { pairingKey = try MirrorPairingCode.generate() lastPairedDevice = nil pairingExpiresAt = Date().addingTimeInterval(60) pairingTask?.cancel() let clock = clock pairingTask = Task { [weak self] in do { try await clock.sleep(for: .seconds(60)) } catch { return } self?.expirePairing() } try rebuildListener() } catch { ProwlLogger("RemoteMirror").warning("Host device operation failed: \(error)") self.error = error.localizedDescription } }
private func expirePairing() { pairingKey = "" pairingExpiresAt = nil pairingTask?.cancel() pairingTask = nil do { if isRunning { try rebuildListener() } } catch { ProwlLogger("RemoteMirror").warning("Host device operation failed: \(error)") self.error = error.localizedDescription } }
func revoke(_ deviceID: UUID) { guard var next = identity else { return } next.devices.removeAll { $0.id == deviceID } do { try saveIdentity(next) identity = next devices = next.devices for (peerID, owner) in commandDevices where owner == deviceID { cancelCommand(peerID, includingCreate: true) } for (peerID, owner) in devicePeers where owner == deviceID { peers[peerID]?.close("Device access was revoked.") } try rebuildListener() } catch { ProwlLogger("RemoteMirror").warning("Host device operation failed: \(error)") self.error = error.localizedDescription } }
func revokeAllDevices() { guard var next = identity else { return } next.devices.removeAll() do { // Persist first: a failed vault write must not pretend access was revoked. try saveIdentity(next) identity = next devices = [] error = nil pairingTask?.cancel() pairingTask = nil pairingKey = "" pairingExpiresAt = nil lastPairedDevice = nil pendingPairings.removeAll() for peerID in Array(commandPeers.keys) { cancelCommand(peerID, includingCreate: true) } // Include unauthenticated peers so an in-flight pairing cannot survive the reset. for peer in Array(peers.values) + Array(pendingPeers.values) { peer.close("All device access was revoked. Pair again to connect.") } if isRunning || isStarting { try rebuildListener() } } catch { ProwlLogger("RemoteMirror").warning("Host device reset failed: \(error)") self.error = error.localizedDescription } }
private struct Subscription { let paneID: UUID let id = UUID() var representation: MirrorMessage.Representation = .terminal var includeViewportText = false var includeStyledScrollback = false var includeScrollState = false var unstableCaptures = 0 var gate = MirrorFrameGate() var textGate = MirrorTextFrameGate() var history: MirrorHistory? var scroll: ScrollRequest? }
private struct ScrollRequest { let payload: MirrorMessage.ScrollPayload var isReady = false var sequence: UInt64? var failure: String?
var isPending: Bool { sequence == nil && failure == nil } }
init( source: any MirrorPaneSource, defaults: UserDefaults = .standard, enabled: Bool, clock: any Clock<Duration> = ContinuousClock(), loadIdentity: @escaping () throws -> MirrorHostIdentity? = { try MirrorCredentialVault.load(MirrorHostIdentity.self, account: "host") }, saveIdentity: @escaping (MirrorHostIdentity) throws -> Void = { try MirrorCredentialVault.save($0, account: "host") } ) { self.clock = clock self.loadIdentity = loadIdentity self.saveIdentity = saveIdentity self.enabled = enabled self.source = source self.defaults = defaults address = defaults.string(forKey: "remoteMirrorHostAddress") ?? "0.0.0.0" port = defaults.string(forKey: "remoteMirrorHostPort") ?? "7880" }
func start() { // Code security: hidden experimental UI must not leave a reachable listener. guard enabled, !isStarting, !isRunning, listener == nil else { return } error = nil guard let portNumber = UInt16(port), portNumber > 0, IPv4Address(address) != nil || IPv6Address(address) != nil else { self.error = String(localized: "Enter a listen address of this Mac and a port between 1 and 65535.") return } do { let saved = try loadIdentity() ?? MirrorHostIdentity(id: UUID(), devices: []) try saveIdentity(saved) identity = saved devices = saved.devices hostRunID = UUID() isStarting = true try rebuildListener() } catch { ProwlLogger("RemoteMirror").warning("Host start failed: \(error)") self.error = String(localized: "Cannot start Host: \(error.localizedDescription)") isStarting = false } }
private func rebuildListener() throws { isStarting = true if retiringListener != nil { return } if listener != nil { retireListener() return } try installListener() }
private func retireListener() { guard let previous = listener else { return } listener = nil retiringListener = previous previous.stateUpdateHandler = { [weak self, weak previous] state in guard case .cancelled = state else { return } Task { @MainActor in guard let self, let previous, self.retiringListener === previous else { return } self.retiringListener = nil guard self.hostRunID != nil else { return } do { try self.installListener() } catch { self.stop() self.error = error.localizedDescription } } } previous.cancel() }
private func installListener() throws { guard let identity, let portNumber = UInt16(port), portNumber > 0 else { throw MirrorProtocolError.invalidMessage } var keys = identity.devices.map { ($0.id.uuidString, $0.key) } if !pairingKey.isEmpty { keys.append(("pair", Data(try MirrorPairingCode.normalized(pairingKey).utf8))) } // An unpairable listener still requires an unpredictable key, never certificate fallback. if keys.isEmpty { keys.append(("closed", try MirrorAuthentication.randomKey())) } let parameters = try MirrorConnection.parameters(keys: keys) let bindHost: NWEndpoint.Host if let ipv4 = IPv4Address(address) { bindHost = .ipv4(ipv4) } else if let ipv6 = IPv6Address(address) { bindHost = .ipv6(ipv6) } else { throw MirrorProtocolError.invalidMessage } parameters.requiredLocalEndpoint = .hostPort( host: bindHost, port: .init(rawValue: portNumber)!) let listener = try NWListener(using: parameters) self.listener = listener isStarting = true listener.stateUpdateHandler = { [weak self, weak listener] state in Task { @MainActor in guard let self, let listener, self.listener === listener else { return } switch state { case .ready: self.isRunning = true self.isStarting = false self.defaults.set(self.address, forKey: "remoteMirrorHostAddress") self.defaults.set(self.port, forKey: "remoteMirrorHostPort") self.completePairings() self.onStarted?() case .failed(let error): ProwlLogger("RemoteMirror").warning("Host listener failed: \(error)") self.stop() self.error = MirrorConnectionFailure(error).listenerMessage(address: self.address, port: self.port) default: break } } } listener.newConnectionHandler = { [weak self, weak listener] connection in Task { @MainActor in guard let self, let listener, self.listener === listener else { connection.cancel() return } self.accept(connection) } } listener.start(queue: .main) }
func stop() { onStopped?() pairingTask?.cancel() pairingTask = nil pairingExpiresAt = nil lastPairedDevice = nil for task in authenticationDeadlines.values { task.cancel() } authenticationDeadlines.removeAll() challenges.removeAll() pendingPairings.removeAll() devicePeers.removeAll() onlineDeviceIDs.removeAll() hostRunID = nil retireListener() pollTask?.cancel() pollTask = nil let connections = Array(peers.values) let pending = Array(pendingPeers.values) pendingPeers.removeAll() pendingOrder.removeAll() pendingHandshakeCount = 0 for peer in pending { peer.close() } for peerID in commandPeers.keys { cancelCommand(peerID, includingCreate: true) } for peer in connections { end(peer, reason: .hostStopped) } peers.removeAll() commandPeers.removeAll() commandDevices.removeAll() subscriptions.removeAll() hostRunID = nil subscriberCount = 0 isRunning = false isStarting = false pairingKey = "" connectionAttempts = MirrorConnectionAttempts() }
private func accept(_ connection: NWConnection) { guard listener != nil, case .hostPort(let address, _) = connection.endpoint, connectionAttempts.allows( source: String(describing: address), now: ProcessInfo.processInfo.systemUptime) else { connection.cancel() return } let sourceAddress = String(describing: address) // Code security: incomplete handshakes never consume authenticated capacity. // Evict the oldest pending connection so a silent full pool can still admit a valid client. if pendingOrder.count >= Self.maximumPendingHandshakes, let oldest = pendingOrder.first { pendingPeers[oldest]?.close() } let peer = MirrorConnection(connection, handshakeTimeout: .seconds(5)) pendingPeers[peer.id] = peer pendingOrder.append(peer.id) pendingHandshakeCount = pendingPeers.count peer.onReady = { [weak self, weak peer] in guard let self, let peer, self.pendingPeers[peer.id] === peer else { return } self.pendingPeers.removeValue(forKey: peer.id) self.pendingOrder.removeAll { $0 == peer.id } self.pendingHandshakeCount = self.pendingPeers.count guard self.listener != nil, self.peers.count < 16 else { peer.close() return } do { guard let identity = self.identity else { throw MirrorProtocolError.invalidMessage } let nonce = try MirrorAuthentication.randomKey() self.peers[peer.id] = peer self.challenges[peer.id] = nonce peer.send(.challenge(.init(hostID: identity.id, nonce: nonce))) let clock = self.clock self.authenticationDeadlines[peer.id] = Task { [weak peer] in do { try await clock.sleep(for: .seconds(5)) } catch { return } peer?.close("Device authentication timed out.") } } catch { peer.close(error.localizedDescription) } } peer.onHandshakeFailure = { [weak self, weak peer] in guard let self, let peer, self.pendingPeers[peer.id] === peer else { return } self.connectionAttempts.recordFailure( source: sourceAddress, now: ProcessInfo.processInfo.systemUptime) } peer.onMessage = { [weak self, weak peer] message in guard let self, let peer, self.peers[peer.id] === peer else { return } if self.devicePeers[peer.id] == nil { self.authenticate(message, peer: peer) } else { self.handle(message, from: peer) } } peer.onClose = { [weak self, weak peer] _ in guard let self, let peer else { return } self.pendingPeers.removeValue(forKey: peer.id) self.pendingOrder.removeAll { $0 == peer.id } self.pendingHandshakeCount = self.pendingPeers.count self.authenticationDeadlines.removeValue(forKey: peer.id)?.cancel() self.challenges.removeValue(forKey: peer.id) self.pendingPairings.removeValue(forKey: peer.id) self.devicePeers.removeValue(forKey: peer.id) self.onlineDeviceIDs = Set(self.devicePeers.values) self.cancelCommand(peer.id) self.peers.removeValue(forKey: peer.id) // Accepted create requests can outlive a lost connection. Keep their device // ownership until completion so explicit revoke or stop can still cancel them. self.subscriptions.removeValue(forKey: peer.id) self.subscriberCount = self.subscriptions.count if self.subscriptions.isEmpty { self.pollTask?.cancel() self.pollTask = nil } } peer.start() }
private func authenticate(_ message: MirrorMessage, peer: MirrorConnection) { do { guard var identity, let nonce = challenges[peer.id] else { throw MirrorProtocolError.invalidMessage } switch message { case .authenticate(let request): // Code security: TLS membership alone is insufficient; prove this specific device's secret. guard let device = identity.devices.first(where: { $0.id == request.deviceID }), MirrorAuthentication.verify( request.proof, key: device.key, challenge: .init(hostID: identity.id, nonce: nonce), purpose: "device", identity: device.id.uuidString) else { throw MirrorProtocolError.invalidMessage } if let index = identity.devices.firstIndex(where: { $0.id == device.id }) { identity.devices[index].lastSeen = Date() } try saveIdentity(identity) self.identity = identity devices = identity.devices devicePeers[peer.id] = device.id onlineDeviceIDs = Set(devicePeers.values) challenges.removeValue(forKey: peer.id) authenticationDeadlines.removeValue(forKey: peer.id)?.cancel() peer.send(.authenticated(identity.id)) case .pair(let request): guard let expires = pairingExpiresAt, expires > Date(), !pairingKey.isEmpty, !request.name.isEmpty, request.name.utf8.count <= 240, !request.name.unicodeScalars.contains(where: { CharacterSet.controlCharacters.contains($0) }), identity.devices.count < 64, MirrorAuthentication.verify( request.proof, key: Data(try MirrorPairingCode.normalized(pairingKey).utf8), challenge: .init(hostID: identity.id, nonce: nonce), purpose: "pair", identity: request.name) else { throw MirrorProtocolError.invalidMessage } let device = MirrorPairedDevice( id: UUID(), name: request.name, key: try MirrorAuthentication.randomKey(), pairedAt: Date()) identity.devices.append(device) try saveIdentity(identity) self.identity = identity devices = identity.devices // Consume before any other peer can pair; a lost response requires a new pairing window. pendingPairings[peer.id] = .init(hostID: identity.id, deviceID: device.id, key: device.key) expirePairing() default: throw MirrorProtocolError.invalidMessage } } catch { if case .hostPort(let address, _) = peer.connection.endpoint { connectionAttempts.recordFailure( source: String(describing: address), now: ProcessInfo.processInfo.systemUptime) } peer.close("Device authentication failed or pairing expired.") } }
private func completePairings() { let pending = pendingPairings pendingPairings.removeAll() for (peerID, credential) in pending { guard devices.contains(where: { $0.id == credential.deviceID }) else { peers[peerID]?.close("Device access was revoked.") continue } // The client reconnects immediately with this key, so the replacement listener must be ready. peers[peerID]?.send(.paired(credential), closeAfterSending: true) lastPairedDevice = devices.first { $0.id == credential.deviceID } } }
private func handle(_ message: MirrorMessage, from peer: MirrorConnection) { do { guard let hostRunID else { throw MirrorProtocolError.invalidMessage } switch message.kind { case .list: let busy = Set(subscriptions.values.map(\.paneID)) let panes = source.panes().map { MirrorPaneDescriptor( id: $0.id, title: $0.title, directory: $0.directory, busy: busy.contains($0.id), projectName: $0.projectName, subtitle: $0.subtitle, role: $0.role) } peer.send( .panes( .init( panes: panes, capabilities: ["vt-v1", "text-v1", "takeover", "refresh"] + (commandService != nil ? ["launch-profile", "launch-shell", "agents-dispatch", "agent-input"] : []) + (source.supportsBoundedHistory ? ["history"] : []) + (source.supportsRemoteScroll ? ["remote-scroll"] : []) + (source.supportsViewportText ? ["viewport-text-v1"] : []) + (source.supportsStyledScrollback ? ["styled-scrollback-v1"] : []) + (source.supportsScrollState ? ["scroll-state-v1"] : []), hostRunID: hostRunID))) case .command: try handleCommand(message, peer: peer) case .commandReceipt: try handleReceipt(message, peer: peer) case .subscribe: try subscribe(message, peer: peer) case .acknowledge: try acknowledge(message, peer: peer) case .refresh: guard var subscription = subscription(for: message, peer: peer) else { throw MirrorProtocolError.invalidMessage } subscription.gate.requestRefresh() subscription.textGate.requestRefresh() subscriptions[peer.id] = subscription case .input: guard let subscription = subscription(for: message, peer: peer), subscription.representation == .terminal, let bytes = message.bytes, !bytes.isEmpty, bytes.count <= MirrorWire.maximumInput else { throw MirrorProtocolError.invalidMessage } try source.write(bytes, to: subscription.paneID) case .scroll: try scroll(message, peer: peer) case .history: try sendHistory(message, to: peer) default: throw MirrorProtocolError.invalidMessage } } catch { peer.close(error.localizedDescription) } }
private func scroll(_ message: MirrorMessage, peer: MirrorConnection) throws { guard var subscription = subscription(for: message, peer: peer), case .scroll(let request) = message else { throw MirrorProtocolError.invalidMessage } if let previous = subscription.scroll, previous.payload.requestID == request.requestID { guard previous.payload.direction == request.direction else { throw MirrorProtocolError.invalidMessage } if let sequence = previous.sequence { peer.send( .scrollResult(.init(requestID: request.requestID, sequence: sequence, subscriptionID: subscription.id))) } else if let failure = previous.failure { peer.send(.failure(.init(error: failure, subscriptionID: subscription.id, requestID: request.requestID))) } return } guard subscription.scroll?.isPending != true else { peer.send( .failure( .init( error: "SCROLL_BUSY: Wait for the current scroll to finish.", subscriptionID: subscription.id, requestID: request.requestID))) return } guard source.supportsRemoteScroll else { peer.send( .failure( .init( error: "SCROLL_UNAVAILABLE: Host cannot scroll this pane.", subscriptionID: subscription.id, requestID: request.requestID))) return } subscription.scroll = ScrollRequest(payload: request) do { try source.scroll(request.direction, to: subscription.paneID) } catch { let failure = "SCROLL_UNAVAILABLE: Host could not scroll this pane." subscription.scroll?.failure = failure subscriptions[peer.id] = subscription peer.send(.failure(.init(error: failure, subscriptionID: subscription.id, requestID: request.requestID))) return } subscriptions[peer.id] = subscription let clock = clock Task { @MainActor [weak self, weak peer] in // TUI mouse input is processed asynchronously. Return a later capture, not // the screen that was present synchronously when the wheel event was sent. do { try await clock.sleep(for: .milliseconds(200)) } catch { return } guard let self, let peer, var current = self.subscriptions[peer.id], current.id == request.subscriptionID, current.scroll?.payload.requestID == request.requestID, current.scroll?.isPending == true else { return } current.scroll?.isReady = true self.subscriptions[peer.id] = current self.poll() // Prefer a changed screen, but a boundary/no-op must also finish. This // bounded refresh is not a receipt from the application processing input. do { try await clock.sleep(for: .milliseconds(600)) } catch { return } guard var latest = self.subscriptions[peer.id], latest.id == request.subscriptionID, latest.scroll?.payload.requestID == request.requestID, latest.scroll?.isPending == true else { return } latest.gate.requestRefresh() latest.textGate.requestRefresh() self.subscriptions[peer.id] = latest self.poll() } }
private func handleCommand(_ message: MirrorMessage, peer: MirrorConnection) throws { guard let service = commandService, let request = message.commandRequest else { throw MirrorProtocolError.invalidMessage } let lease: UUID? // Code security: commands cannot escape the authenticated connection’s current pane lease. if let paneID = request.request.command.targetPaneID { guard let active = subscription(for: message, peer: peer), paneID == active.paneID else { throw MirrorProtocolError.invalidMessage } lease = active.id } else { if subscriptions[peer.id] != nil { guard case .list = request.request.command, let active = subscription(for: message, peer: peer) else { throw MirrorProtocolError.invalidMessage } lease = active.id } else { lease = nil } } if let pending = commandPeers[peer.id] { guard pending == request else { throw MirrorProtocolError.invalidMessage } return } commandPeers[peer.id] = request commandDevices[peer.id] = devicePeers[peer.id] let peerID = peer.id Task { @MainActor [weak self, weak peer] in let response = await service.execute(request) { [weak self, weak peer] in guard let self, let peer, self.peers[peer.id] === peer else { return false } return lease == nil || self.subscriptions[peer.id]?.id == lease } guard let self else { return } self.commandPeers.removeValue(forKey: peerID) self.commandDevices.removeValue(forKey: peerID) guard let peer, self.peers[peerID] === peer else { return } peer.send(.commandResult(.init(commandResponse: response))) } }
private func cancelCommand(_ peerID: UUID, includingCreate: Bool = false) { guard let request = commandPeers[peerID], includingCreate || request.request.command.targetPaneID != nil else { return } commandService?.cancel(request.requestID) }
private func handleReceipt(_ message: MirrorMessage, peer: MirrorConnection) throws { guard let active = subscription(for: message, peer: peer), let requestID = message.commandReceiptID, let service = commandService else { throw MirrorProtocolError.invalidMessage } Task { @MainActor [weak self, weak peer] in let response = await service.receipt(requestID, paneID: active.paneID) guard let self, let peer, self.subscriptions[peer.id]?.id == active.id else { return } peer.send(.commandResult(.init(commandResponse: response))) } }
private func subscription(for message: MirrorMessage, peer: MirrorConnection) -> Subscription? { guard let subscription = subscriptions[peer.id] else { return nil } guard message.subscriptionID == subscription.id else { return nil } return subscription }
private func acknowledge(_ message: MirrorMessage, peer: MirrorConnection) throws { guard var subscription = subscription(for: message, peer: peer), let sequence = message.sequence else { throw MirrorProtocolError.invalidMessage } if subscription.representation == .text { try subscription.textGate.acknowledge(sequence) } else { try subscription.gate.acknowledge(sequence) } subscriptions[peer.id] = subscription }
private func subscribe(_ message: MirrorMessage, peer: MirrorConnection) throws { guard let hostRunID, subscriptions[peer.id] == nil, let paneID = message.paneID, let representation = message.representation, source.panes().contains(where: { $0.id == paneID }) else { throw MirrorProtocolError.invalidMessage } let previous = subscriptions.first { $0.value.paneID == paneID } guard previous == nil || message.intent == .takeover else { peer.send( .failure( .init(error: "PANE_BUSY: This pane already has a remote mirror.", subscriptionID: nil))) return } var next = Subscription( paneID: paneID, representation: representation, includeViewportText: representation == .terminal && message.includeViewportText == true && source.supportsViewportText, includeStyledScrollback: representation == .terminal && message.includeViewportText == true && message.includeStyledScrollback == true && source.supportsStyledScrollback, includeScrollState: message.includeScrollState == true && source.supportsScrollState) // Prepare and encode before revoking the old lease. Capture failure leaves it intact. let first = try capture(&next) if let first { _ = try MirrorWire.encode(first) } if let previous { subscriptions.removeValue(forKey: previous.key) if let oldPeer = peers[previous.key] { end(oldPeer, reason: .takenOver) } } subscriptions[peer.id] = next subscriberCount = subscriptions.count peer.send(.subscribed(.init(paneID: paneID, subscriptionID: next.id, hostRunID: hostRunID))) if let first { send(first, subscription: next, to: peer) }
if pollTask == nil { let clock = clock pollTask = Task { [weak self] in while !Task.isCancelled { do { try await clock.sleep(for: .milliseconds(200)) } catch { return } self?.poll() } } } }
private func end(_ peer: MirrorConnection, reason: MirrorMessage.EndReason) { cancelCommand(peer.id) peer.send(.ended(.init(reason: reason)), closeAfterSending: true) }
private func capture(_ subscription: inout Subscription) throws -> MirrorMessage? { if let scroll = subscription.scroll, scroll.isPending, !scroll.isReady { return nil } do { let frame = try captureAvailable(&subscription) subscription.unstableCaptures = 0 return frame } catch MirrorPaneSourceError.captureChanged { subscription.unstableCaptures += 1 guard subscription.unstableCaptures < 10 else { throw MirrorProtocolError.invalidMessage } return nil } }
private func captureAvailable(_ subscription: inout Subscription) throws -> MirrorMessage? { if subscription.representation == .text { guard subscription.textGate.outstanding == nil else { return nil } let captured = try source.textSnapshot(subscription.paneID) let text = captured.text let bounds = subscription.includeScrollState ? captured.scrollBounds : nil guard let sequence = subscription.textGate.offer( text, columns: captured.columns, rows: captured.rows, truncated: captured.truncated, scrollBounds: bounds) else { return nil } return .textFrame( .init( columns: captured.columns, rows: captured.rows, truncated: captured.truncated, sequence: sequence, text: text, subscriptionID: subscription.id, scrollBounds: bounds)) } guard subscription.gate.outstanding == nil else { return nil } var frame = try source.snapshot( subscription.paneID, styledScrollback: subscription.includeStyledScrollback) if !subscription.includeViewportText { frame.viewportText = nil } if !subscription.includeScrollState { frame.scrollBounds = nil } guard let sequence = subscription.gate.offer(frame) else { return nil } return .frame(.init(frame: frame, sequence: sequence, subscriptionID: subscription.id)) }
private func send(_ message: MirrorMessage, subscription: Subscription, to peer: MirrorConnection) { if subscription.includeViewportText, case .frame(let payload) = message { peer.send( .viewport( .init( styledScrollback: payload.frame.styledScrollback, text: payload.frame.viewportText, sequence: payload.sequence, subscriptionID: subscription.id))) } if subscription.includeScrollState, let sequence = message.sequence { let bounds: MirrorScrollBounds? switch message { case .frame(let payload): bounds = payload.frame.scrollBounds case .textFrame(let payload): bounds = payload.scrollBounds default: bounds = nil } peer.send( .scrollState( .init( atTop: bounds?.atTop, atBottom: bounds?.atBottom, sequence: sequence, subscriptionID: subscription.id))) } peer.send(message) }
private func sendHistory(_ message: MirrorMessage, to peer: MirrorConnection) throws { guard var subscription = subscription(for: message, peer: peer) else { throw MirrorProtocolError.invalidMessage } if message.historyID == nil { do { do { guard source.supportsBoundedHistory else { throw MirrorProtocolError.invalidMessage } let captured = try source.boundedRetainedText(subscription.paneID) subscription.history = MirrorHistory(text: captured.text, truncated: captured.truncated) } catch { peer.send( .failure( .init( error: "HISTORY_UNAVAILABLE: Cannot capture history within the supported limits.", subscriptionID: subscription.id))) return } } } guard let history = subscription.history, message.historyID == nil || message.historyID == history.id else { throw MirrorProtocolError.invalidMessage } let page = try history.page(before: message.offset ?? history.lines.count) subscriptions[peer.id] = subscription peer.send( .historyPage( .init( historyID: history.id, offset: page.start, lines: page.lines, total: history.lines.count, subscriptionID: subscription.id, capturedAt: history.capturedAt, truncated: history.truncated))) }
private func poll() { for (id, var subscription) in subscriptions { guard let peer = peers[id] else { continue } guard source.panes().contains(where: { $0.id == subscription.paneID }) else { subscriptions.removeValue(forKey: id) subscriberCount = subscriptions.count end(peer, reason: .paneClosed) continue } do { if let frame = try capture(&subscription) { var result: MirrorMessage? if let scroll = subscription.scroll, scroll.isPending, scroll.isReady, let sequence = frame.sequence { subscription.scroll?.sequence = sequence result = .scrollResult( .init(requestID: scroll.payload.requestID, sequence: sequence, subscriptionID: subscription.id)) } subscriptions[id] = subscription send(frame, subscription: subscription, to: peer) if let result { peer.send(result) } } else { subscriptions[id] = subscription } } catch { peer.close("Host pane is unavailable: \(error.localizedDescription)") } } }}
nonisolated struct MirrorConnectionAttempts { private var failures: [String: [TimeInterval]] = [:]
mutating func allows(source: String, now: TimeInterval) -> Bool { guard now.isFinite else { return false } prune(now: now) return (failures[source]?.count ?? 0) < 12 }
mutating func recordFailure(source: String, now: TimeInterval) { guard now.isFinite else { return } prune(now: now) // Bound bookkeeping even when many source addresses fail authentication. if failures[source] == nil, failures.count >= 256, let oldest = failures.min(by: { ($0.value.last ?? 0) < ($1.value.last ?? 0) })?.key { failures.removeValue(forKey: oldest) } if (failures[source]?.count ?? 0) < 12 { failures[source, default: []].append(now) } }
private mutating func prune(now: TimeInterval) { failures = failures.compactMapValues { times in let recent = times.filter { now < $0 + 60 } return recent.isEmpty ? nil : recent } }}