From eead80c0c8cab04809feafc51e639296b1dbd50c Mon Sep 17 00:00:00 2001 From: Producer Guy <254332139+ProducerGuy@users.noreply.github.com> Date: Sat, 3 Oct 2026 15:33:07 -0500 Subject: [PATCH] daemon: authenticate socket peers (root or owner only) Each accepted control-socket connection is checked with getpeereid(): only uid 0 and the daemon's owner uid are served, as a second layer behind the socket's 0600 permissions. Rejected connections are closed before any byte is read, never hold a slot, and are logged at a bounded rate. Accepts are capped at 64 per event. Daemon and client sockets set SO_NOSIGPIPE so a peer that hangs up costs an EPIPE, not the process. Socket tests run serialized with outcome-based waits. --- .../ThermalForgeCore/ConnectionServer.swift | 66 ++- Sources/ThermalForgeCore/Daemon.swift | 22 +- .../ThermalForgeCore/DaemonInvariants.swift | 100 +++++ .../ConnectionServerTests.swift | 117 ++--- Tests/ThermalForgeTests/PeerAuthTests.swift | 424 ++++++++++++++++++ Tests/ThermalForgeTests/SIGPIPETests.swift | 145 ++++++ Tests/ThermalForgeTests/SocketSuites.swift | 19 + 7 files changed, 831 insertions(+), 62 deletions(-) create mode 100644 Tests/ThermalForgeTests/PeerAuthTests.swift create mode 100644 Tests/ThermalForgeTests/SIGPIPETests.swift create mode 100644 Tests/ThermalForgeTests/SocketSuites.swift diff --git a/Sources/ThermalForgeCore/ConnectionServer.swift b/Sources/ThermalForgeCore/ConnectionServer.swift index 4f34c88e..f93b0b7d 100644 --- a/Sources/ThermalForgeCore/ConnectionServer.swift +++ b/Sources/ThermalForgeCore/ConnectionServer.swift @@ -7,7 +7,8 @@ // deadline (replacing Phase 0's SO_RCVTIMEO). Decoupled from DaemonServer so it can be // tested against a plain bound AF_UNIX socket — the daemon's request processing is // injected as `handle`. Framing-level replies (legacy peer, oversized) live here; the -// verb dispatch does not. +// verb dispatch does not. Every accepted fd passes the injected peer check before +// anything else happens to it. // import Darwin @@ -16,6 +17,11 @@ import Foundation final class ConnectionServer: @unchecked Sendable { private let listenFD: Int32 private let maxConnections: Int + /// Accepts per accept event before returning, so one burst of connects can't hold + /// the accept queue in a single handler call. The read source is level-triggered: + /// anything still pending fires it again. + private let maxAcceptsPerEvent: Int + private let authorizer: any PeerAuthorizing private let headerDeadline: TimeInterval private let requestDeadline: TimeInterval /// Processes a decoded request body → response. DaemonServer wraps its call in @@ -26,14 +32,29 @@ final class ConnectionServer: @unchecked Sendable { private var acceptSource: DispatchSourceRead? private var activeConnections = 0 // acceptQueue-confined private var accepting = true // acceptQueue-confined + private var rejectionLog: RejectionLogLimiter // acceptQueue-confined + private var summaryScheduled = false // acceptQueue-confined + private let summaryDelay: TimeInterval + private let log: (String) -> Void + /// `authorizer` is required, with no allow-all default, so no caller can build an + /// unauthenticated server by omission. init(listenFD: Int32, + authorizer: any PeerAuthorizing, maxConnections: Int = 8, + maxAcceptsPerEvent: Int = 64, headerDeadline: TimeInterval = 1.0, requestDeadline: TimeInterval = 5.0, + summaryDelay: TimeInterval = 60, + log: @escaping (String) -> Void = { NSLog("%@", $0) }, handle: @escaping (Data) -> DaemonResponse) { self.listenFD = listenFD + self.authorizer = authorizer self.maxConnections = maxConnections + self.maxAcceptsPerEvent = maxAcceptsPerEvent + self.rejectionLog = RejectionLogLimiter(now: Date()) + self.summaryDelay = summaryDelay + self.log = log self.headerDeadline = headerDeadline self.requestDeadline = requestDeadline self.handle = handle @@ -42,13 +63,26 @@ final class ConnectionServer: @unchecked Sendable { /// Accept via a DispatchSource on the (non-blocking) listen fd, bounded to /// maxConnections concurrent handlers. At capacity we suspend accepting; a finishing /// connection resumes it — pending connects wait in the listen backlog (queueing). + /// + /// The peer check runs right after accept, before the fd is counted, made + /// non-blocking, or wrapped in DispatchIO. A rejected fd is closed inline: no byte + /// is read from it or written to it, it never holds a slot, and connectionFinished + /// never runs for it. func start() { _ = fcntl(listenFD, F_SETFL, fcntl(listenFD, F_GETFL, 0) | O_NONBLOCK) let source = DispatchSource.makeReadSource(fileDescriptor: listenFD, queue: acceptQueue) source.setEventHandler { [self] in - while activeConnections < maxConnections { + var accepted = 0 + while activeConnections < maxConnections, accepted < maxAcceptsPerEvent { let clientFD = accept(listenFD, nil, nil) if clientFD < 0 { break } // EAGAIN (no more pending) or error + accepted += 1 + let decision = authorizer.decide(fd: clientFD) + guard case .allow = decision else { + close(clientFD) + logRejection(decision) + continue + } activeConnections += 1 handleConnection(clientFD) } @@ -66,6 +100,29 @@ final class ConnectionServer: @unchecked Sendable { source.resume() } + /// Rate-limited (see RejectionLogLimiter). The first suppressed rejection arms a + /// one-shot summary so a flood that stops still gets its count logged. + private func logRejection(_ decision: PeerDecision) { + let message: String + switch decision { + case .reject(let peer): + message = "ThermalForge daemon: rejected peer euid \(peer.uid) egid \(peer.gid) (allowed: \(authorizer.allowedDescription))" + case .unavailable(let err): + message = "ThermalForge daemon: rejected peer, credentials unavailable: errno \(err) (allowed: \(authorizer.allowedDescription))" + case .allow: + return + } + if let line = rejectionLog.record(message, now: Date()) { + log(line) + } else if !summaryScheduled { + summaryScheduled = true + acceptQueue.asyncAfter(deadline: .now() + summaryDelay) { [self] in + summaryScheduled = false + if let summary = rejectionLog.flush() { log(summary) } + } + } + } + private func connectionFinished() { activeConnections -= 1 if !accepting, activeConnections < maxConnections { @@ -79,6 +136,11 @@ final class ConnectionServer: @unchecked Sendable { /// concurrently. The header deadline closes a connect-and-hang fast (freeing its slot /// so a queued request isn't starved); the full deadline bounds a slow/partial body. private func handleConnection(_ fd: Int32) { + // A client that hangs up before its reply is written must cost us an EPIPE, + // not a SIGPIPE: the default action would terminate the daemon. Any new socket + // the daemon writes to must set SO_NOSIGPIPE too. + var on: Int32 = 1 + _ = setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &on, socklen_t(MemoryLayout.size)) _ = fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK) let q = DispatchQueue(label: "com.thermalforge.conn") let io = DispatchIO(type: .stream, fileDescriptor: fd, queue: q) { [self] _ in diff --git a/Sources/ThermalForgeCore/Daemon.swift b/Sources/ThermalForgeCore/Daemon.swift index 27982986..35de4129 100644 --- a/Sources/ThermalForgeCore/Daemon.swift +++ b/Sources/ThermalForgeCore/Daemon.swift @@ -142,7 +142,13 @@ public struct FanApplyResult: Equatable { } public final class DaemonClient { - public init() {} + private let socketPath: String + + public init() { self.socketPath = ThermalForgeDaemon.socketPath } + + /// Test seam: point the client at a plain bound AF_UNIX socket instead of the + /// daemon's. Production callers use `init()`. + init(socketPath: String) { self.socketPath = socketPath } /// Read the daemon's current hold (what's set and who owns it) so the menu /// bar app can reflect a CLI hold instead of fighting or wiping it. @@ -188,10 +194,15 @@ public final class DaemonClient { var tv = timeval(tv_sec: whole, tv_usec: Int32((timeout - Double(whole)) * 1_000_000)) setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, socklen_t(MemoryLayout.size)) setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, socklen_t(MemoryLayout.size)) + // A daemon that closes before replying (restart, or a rejected peer) must make + // the write fail with EPIPE, not raise SIGPIPE and kill the app or CLI. Any new + // client socket must set SO_NOSIGPIPE too. + var noSigPipe: Int32 = 1 + setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &noSigPipe, socklen_t(MemoryLayout.size)) var addr = sockaddr_un() addr.sun_family = sa_family_t(AF_UNIX) - setPath(&addr, ThermalForgeDaemon.socketPath) + setPath(&addr, socketPath) // connect() must ALSO be bounded, not just read/write. A wedged daemon // (accept loop stalled in a slow handleClient, listen backlog full) makes a @@ -380,7 +391,12 @@ public final class DaemonServer { // Accept connections concurrently (bounded) so one hung connection can't stall // others, and — the security fix — a slow-reading client can no longer hold // smcLock during the response write (processFrame takes it only around process()). - let server = ConnectionServer(listenFD: socketFD) { [self] body in processFrame(body) } + // Every accepted connection is checked against the peer's kernel credentials + // (root or ownerUID only), a second layer behind the socket's 0600 permissions. + let server = ConnectionServer(listenFD: socketFD, + authorizer: PeerAuthorizer(ownerUID: ownerUID)) { [self] body in + processFrame(body) + } server.start() connectionServer = server diff --git a/Sources/ThermalForgeCore/DaemonInvariants.swift b/Sources/ThermalForgeCore/DaemonInvariants.swift index 764351b7..87220a2e 100644 --- a/Sources/ThermalForgeCore/DaemonInvariants.swift +++ b/Sources/ThermalForgeCore/DaemonInvariants.swift @@ -71,3 +71,103 @@ public struct ThermalFloor { return .engage } } + +// MARK: - Peer authentication + +/// A connected peer's effective uid/gid, as the kernel recorded them at connect(). +struct PeerCredentials: Equatable { + let uid: uid_t + let gid: gid_t +} + +/// Why a peer's credentials couldn't be read. Any such failure is a rejection. +struct PeerCredentialError: Error, Equatable { + let errno: Int32 +} + +/// The per-connection peer check's verdict. Only `.allow` gets a handler. +enum PeerDecision: Equatable { + case allow(PeerCredentials) + case reject(PeerCredentials) + case unavailable(errno: Int32) +} + +/// What ConnectionServer needs from a peer check: a verdict per accepted fd, and the +/// allowed set for the rejection log. Injected so tests can supply a fake. +protocol PeerAuthorizing { + func decide(fd: Int32) -> PeerDecision + var allowedDescription: String { get } +} + +/// Second layer behind the socket's 0600 permissions: only root and the daemon's +/// owner may talk to it. Credentials come from getpeereid(), which reads the +/// LOCAL_PEERCRED copy the kernel took at connect() — still valid after the peer +/// exits, so the check can't race the peer going away. +struct PeerAuthorizer: PeerAuthorizing { + let ownerUID: uid_t + private let credentials: (Int32) -> Result + + init(ownerUID: uid_t, + credentials: @escaping (Int32) -> Result = PeerAuthorizer.kernelCredentials) { + self.ownerUID = ownerUID + self.credentials = credentials + } + + /// True only for root and the owner. + static func allows(uid: uid_t, ownerUID: uid_t) -> Bool { + uid == 0 || uid == ownerUID + } + + func decide(fd: Int32) -> PeerDecision { + switch credentials(fd) { + case .success(let peer): + return Self.allows(uid: peer.uid, ownerUID: ownerUID) ? .allow(peer) : .reject(peer) + case .failure(let error): + return .unavailable(errno: error.errno) + } + } + + var allowedDescription: String { "uid 0, owner uid \(ownerUID)" } + + static func kernelCredentials(_ fd: Int32) -> Result { + // Start from values that can never pass the policy, so no path defaults to root. + var uid = uid_t.max + var gid = gid_t.max + guard getpeereid(fd, &uid, &gid) == 0 else { + return .failure(PeerCredentialError(errno: errno)) + } + return .success(PeerCredentials(uid: uid, gid: gid)) + } +} + +/// Bounds the rejection log so a hostile local process can't flood the system log: +/// a burst of 5 lines, then one per minute. Suppressed rejections are counted; the +/// count rides on the next line that gets through, or on the summary the caller +/// emits from its one-shot timer (`flush`). The clock is injected for tests. +struct RejectionLogLimiter { + private var limiter: RateLimiter + private(set) var suppressed = 0 + + init(now: Date) { + limiter = RateLimiter(capacity: 5, refillPerSecond: 1.0 / 60, now: now) + } + + /// The line to log for this rejection, or nil when it's suppressed. + mutating func record(_ message: String, now: Date) -> String? { + guard limiter.allow(now: now) else { + suppressed += 1 + return nil + } + defer { suppressed = 0 } + return suppressed > 0 + ? "\(message) (\(suppressed) earlier rejections suppressed)" + : message + } + + /// Summary of rejections suppressed since the last line, or nil if none. + mutating func flush() -> String? { + guard suppressed > 0 else { return nil } + defer { suppressed = 0 } + return "ThermalForge daemon: \(suppressed) further peer rejections suppressed" + } +} diff --git a/Tests/ThermalForgeTests/ConnectionServerTests.swift b/Tests/ThermalForgeTests/ConnectionServerTests.swift index 591808a1..659545e1 100644 --- a/Tests/ThermalForgeTests/ConnectionServerTests.swift +++ b/Tests/ThermalForgeTests/ConnectionServerTests.swift @@ -13,74 +13,77 @@ import Testing @testable import ThermalForgeCore -@Suite("Connection server (Phase 4)") -struct ConnectionServerTests { +extension SocketSuites { + @Suite("Connection server (Phase 4)") + struct ConnectionServerTests { - private func setPath(_ addr: inout sockaddr_un, _ path: String) { - withUnsafeMutablePointer(to: &addr.sun_path) { ptr in - ptr.withMemoryRebound(to: CChar.self, capacity: 104) { _ = strlcpy($0, path, 104) } + private func setPath(_ addr: inout sockaddr_un, _ path: String) { + withUnsafeMutablePointer(to: &addr.sun_path) { ptr in + ptr.withMemoryRebound(to: CChar.self, capacity: 104) { _ = strlcpy($0, path, 104) } + } } - } - private func bindListener(_ path: String) -> Int32 { - unlink(path) - let fd = socket(AF_UNIX, SOCK_STREAM, 0) - var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) - let len = socklen_t(MemoryLayout.size) - let r = withUnsafePointer(to: &addr) { - $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, len) } + private func bindListener(_ path: String) -> Int32 { + unlink(path) + let fd = socket(AF_UNIX, SOCK_STREAM, 0) + var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) + let len = socklen_t(MemoryLayout.size) + let r = withUnsafePointer(to: &addr) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, len) } + } + #expect(r == 0) + #expect(listen(fd, 16) == 0) + return fd } - #expect(r == 0) - #expect(listen(fd, 16) == 0) - return fd - } - private func connectClient(_ path: String) -> Int32 { - let fd = socket(AF_UNIX, SOCK_STREAM, 0) - var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) - let len = socklen_t(MemoryLayout.size) - let r = withUnsafePointer(to: &addr) { - $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { connect(fd, $0, len) } + private func connectClient(_ path: String) -> Int32 { + let fd = socket(AF_UNIX, SOCK_STREAM, 0) + var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) + let len = socklen_t(MemoryLayout.size) + let r = withUnsafePointer(to: &addr) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { connect(fd, $0, len) } + } + #expect(r == 0) + return fd } - #expect(r == 0) - return fd - } - @Test("8 connect-and-hang sockets don't starve a real request — served within ~1s") - func floodThenRealRequest() throws { - let path = "/tmp/tf-conn-test.sock" - let listenFD = bindListener(path) - defer { close(listenFD); unlink(path) } + @Test("8 connect-and-hang sockets don't starve a real request — served within ~1s") + func floodThenRealRequest() throws { + let path = "/tmp/tf-conn-test.sock" + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } - let server = ConnectionServer(listenFD: listenFD, maxConnections: 8, - headerDeadline: 1.0, requestDeadline: 5.0) { _ in - .versionResponse("served") // prove the request was actually processed - } - server.start() + let server = ConnectionServer(listenFD: listenFD, authorizer: FakeAuthorizer.allowAll(), + maxConnections: 8, + headerDeadline: 1.0, requestDeadline: 5.0) { _ in + .versionResponse("served") // prove the request was actually processed + } + server.start() - // Fill the pool: 8 sockets that connect and send NOTHING. - var hangs: [Int32] = [] - for _ in 0..<8 { hangs.append(connectClient(path)) } - defer { hangs.forEach { close($0) } } - Thread.sleep(forTimeInterval: 0.2) // let the acceptor take all 8 + // Fill the pool: 8 sockets that connect and send NOTHING. + var hangs: [Int32] = [] + for _ in 0..<8 { hangs.append(connectClient(path)) } + defer { hangs.forEach { close($0) } } + Thread.sleep(forTimeInterval: 0.2) // let the acceptor take all 8 - // A real request behind the flood. - let client = connectClient(path) - defer { close(client) } - let frame = try DaemonProtocol.encodeFrame(DaemonRequest(verb: .version), - max: DaemonProtocol.maxRequestBytes) - _ = frame.withUnsafeBytes { write(client, $0.baseAddress, frame.count) } - var tv = timeval(tv_sec: 3, tv_usec: 0) - setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, &tv, socklen_t(MemoryLayout.size)) + // A real request behind the flood. + let client = connectClient(path) + defer { close(client) } + let frame = try DaemonProtocol.encodeFrame(DaemonRequest(verb: .version), + max: DaemonProtocol.maxRequestBytes) + _ = frame.withUnsafeBytes { write(client, $0.baseAddress, frame.count) } + var tv = timeval(tv_sec: 3, tv_usec: 0) + setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, &tv, socklen_t(MemoryLayout.size)) - let t0 = Date() - let body = try DaemonProtocol.readFrame(client, max: DaemonProtocol.maxResponseBytes) - let elapsed = Date().timeIntervalSince(t0) - let resp = try DaemonProtocol.decode(DaemonResponse.self, from: body) + let t0 = Date() + let body = try DaemonProtocol.readFrame(client, max: DaemonProtocol.maxResponseBytes) + let elapsed = Date().timeIntervalSince(t0) + let resp = try DaemonProtocol.decode(DaemonResponse.self, from: body) - print(String(format: "FLOOD: real request served in %.3fs behind 8 connect-and-hang sockets", elapsed)) - #expect(resp.ok) - #expect(resp.version == "served") - #expect(elapsed < 1.3) // freed by the ~1s header deadline, not the 5s full deadline + print(String(format: "FLOOD: real request served in %.3fs behind 8 connect-and-hang sockets", elapsed)) + #expect(resp.ok) + #expect(resp.version == "served") + #expect(elapsed < 1.3) // freed by the ~1s header deadline, not the 5s full deadline + } } } diff --git a/Tests/ThermalForgeTests/PeerAuthTests.swift b/Tests/ThermalForgeTests/PeerAuthTests.swift new file mode 100644 index 00000000..3b909810 --- /dev/null +++ b/Tests/ThermalForgeTests/PeerAuthTests.swift @@ -0,0 +1,424 @@ +// +// PeerAuthTests.swift +// ThermalForge +// +// The per-connection peer check: the pure root-or-owner policy, rejection through +// ConnectionServer (closed before any byte is read, never holding a slot), the real +// getpeereid() path, and the bounded rejection log. Single-user safe: a real-kernel +// rejection comes from setting ownerUID to a uid that isn't ours. +// +// No assertion here depends on machine speed. Servers get deadlines far beyond any +// wait, so a deadline can never stand in for the behavior under test; waits are for +// outcomes, with a generous cap that only bounds a genuine failure. +// + +import Darwin +import Foundation +import Testing + +@testable import ThermalForgeCore + +/// A peer check whose verdict the test decides. `allowAll` stands in where a test is +/// about something other than authentication. +struct FakeAuthorizer: PeerAuthorizing { + let verdict: (Int32) -> PeerDecision + func decide(fd: Int32) -> PeerDecision { verdict(fd) } + var allowedDescription: String { "test" } + + static func allowAll() -> FakeAuthorizer { + FakeAuthorizer { _ in .allow(PeerCredentials(uid: getuid(), gid: getgid())) } + } + static func rejectAll() -> FakeAuthorizer { + FakeAuthorizer { _ in .reject(PeerCredentials(uid: 4294967294, gid: 4294967294)) } + } +} + +/// Upper bounds that only cap a genuine failure, never a pass. +enum SocketTestLimits { + /// Longest any test waits for an outcome. + static let wait: TimeInterval = 30 + /// Server header/request deadlines: longer than any wait, so they never fire first. + static let serverDeadline: TimeInterval = 120 +} + +/// Thread-safe recorder for values produced on the server's queues, with a wait for an +/// outcome instead of a fixed sleep. +final class Box: @unchecked Sendable { + private let cond = NSCondition() + private var items: [T] = [] + func append(_ item: T) { cond.lock(); items.append(item); cond.broadcast(); cond.unlock() } + var all: [T] { cond.lock(); defer { cond.unlock() }; return items } + + /// Block until `done(items)` holds; false only if `cap` passes first. + func wait(cap: TimeInterval = SocketTestLimits.wait, until done: ([T]) -> Bool) -> Bool { + let deadline = Date().addingTimeInterval(cap) + cond.lock(); defer { cond.unlock() } + while !done(items) { + if !cond.wait(until: deadline) { return done(items) } + } + return true + } +} + +extension SocketSuites { + @Suite("Peer authentication") + struct PeerAuthTests { + + // MARK: - Socket helpers + + private func setPath(_ addr: inout sockaddr_un, _ path: String) { + withUnsafeMutablePointer(to: &addr.sun_path) { ptr in + ptr.withMemoryRebound(to: CChar.self, capacity: 104) { _ = strlcpy($0, path, 104) } + } + } + + private func bindListener(_ path: String) -> Int32 { + unlink(path) + let fd = socket(AF_UNIX, SOCK_STREAM, 0) + var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) + let len = socklen_t(MemoryLayout.size) + let r = withUnsafePointer(to: &addr) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, len) } + } + #expect(r == 0) + #expect(listen(fd, 16) == 0) + return fd + } + + /// Connected client with SO_NOSIGPIPE, so writing to a rejected (closed) socket + /// fails with EPIPE instead of killing the test runner. + private func connectClient(_ path: String) -> Int32 { + let fd = socket(AF_UNIX, SOCK_STREAM, 0) + var on: Int32 = 1 + setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &on, socklen_t(MemoryLayout.size)) + var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) + let len = socklen_t(MemoryLayout.size) + let r = withUnsafePointer(to: &addr) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { connect(fd, $0, len) } + } + #expect(r == 0) + return fd + } + + /// Start a server whose deadlines never fire within a test. The dispatch source's + /// handler retains the server, so it lives for the rest of the process. + private func startServer(_ listenFD: Int32, _ authorizer: any PeerAuthorizing, + maxConnections: Int = 8, maxAcceptsPerEvent: Int = 64, + summaryDelay: TimeInterval = 60, + log: @escaping (String) -> Void = { _ in }, + handle: @escaping (Data) -> DaemonResponse = { _ in .versionResponse("served") }) { + ConnectionServer(listenFD: listenFD, authorizer: authorizer, + maxConnections: maxConnections, + maxAcceptsPerEvent: maxAcceptsPerEvent, + headerDeadline: SocketTestLimits.serverDeadline, + requestDeadline: SocketTestLimits.serverDeadline, + summaryDelay: summaryDelay, log: log, handle: handle).start() + } + + private func request(_ path: String) throws -> DaemonResponse { + try DaemonClient(socketPath: path).request(DaemonRequest(verb: .version), + timeout: SocketTestLimits.wait) + } + + /// Send a valid frame, then read. A rejected connection yields EOF (0) or + /// ECONNRESET. The server's deadlines outlast the read timeout, so neither can come + /// from a deadline, only from the rejection close; a timeout would read -1/EAGAIN. + private func sendAndRead(_ fd: Int32) throws -> (n: Int, err: Int32) { + var tv = timeval(tv_sec: Int(SocketTestLimits.wait), tv_usec: 0) + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, socklen_t(MemoryLayout.size)) + let frame = try DaemonProtocol.encodeFrame(DaemonRequest(verb: .version), + max: DaemonProtocol.maxRequestBytes) + _ = frame.withUnsafeBytes { write(fd, $0.baseAddress, frame.count) } + var buf = [UInt8](repeating: 0, count: 64) + let n = read(fd, &buf, buf.count) + return (n, n < 0 ? errno : 0) + } + + private func uniquePath(_ tag: String) -> String { + "/tmp/tf-peer-\(tag)-\(getpid())-\(UInt32.random(in: 0...UInt32.max)).sock" + } + + /// The integer between `prefix` and `suffix` in `line`, if both are present. + private static func count(in line: String, after prefix: String, before suffix: String) -> Int? { + guard let end = line.range(of: suffix, options: .backwards), + let start = line.range(of: prefix, options: .backwards, + range: line.startIndex..) -> PeerAuthorizer { + PeerAuthorizer(ownerUID: 501, credentials: { _ in r }) + } + let owner = PeerCredentials(uid: 501, gid: 20) + let root = PeerCredentials(uid: 0, gid: 0) + let other = PeerCredentials(uid: 502, gid: 20) + #expect(authorizer(.success(owner)).decide(fd: -1) == .allow(owner)) + #expect(authorizer(.success(root)).decide(fd: -1) == .allow(root)) + #expect(authorizer(.success(other)).decide(fd: -1) == .reject(other)) + #expect(authorizer(.failure(PeerCredentialError(errno: ENOTCONN))).decide(fd: -1) + == .unavailable(errno: ENOTCONN)) + } + + // MARK: 2. Injected authorizer always rejects + + @Test("Rejected peer is closed unread: EOF, handler never runs, one log line") + func rejectedPeerIsClosedUnread() throws { + let path = uniquePath("reject") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + let handled = Box() + let logged = Box() + startServer(listenFD, FakeAuthorizer.rejectAll(), log: { logged.append($0) }) { _ in + handled.append(1) + return .versionResponse("served") + } + + let client = connectClient(path) + defer { close(client) } + let r = try sendAndRead(client) + + #expect(r.n == 0 || (r.n < 0 && r.err == ECONNRESET)) + #expect(logged.wait { !$0.isEmpty }) + #expect(handled.all.isEmpty) + #expect(logged.all.count == 1) + #expect(logged.all.first?.contains("euid 4294967294 egid 4294967294 (allowed: test)") == true) + } + + // MARK: 3. Rejections don't use slots + + @Test("50 rejections hold no slot: the next request is still served", + arguments: [8, 1]) + func rejectionsHoldNoSlots(maxConnections: Int) throws { + let path = uniquePath("slots\(maxConnections)") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + // A rejection that held a slot would never give it back (rejected fds never + // reach connectionFinished), and no deadline fires within the wait, so the + // request below could only fail. + let decisions = Box() + let fake = FakeAuthorizer { _ in + decisions.append(1) + return decisions.all.count <= 50 + ? .reject(PeerCredentials(uid: 4294967294, gid: 4294967294)) + : .allow(PeerCredentials(uid: getuid(), gid: getgid())) + } + startServer(listenFD, fake, maxConnections: maxConnections) + + // Paced on the server's decisions so the test's 16-deep backlog never overflows. + var rejected: [Int32] = [] + defer { rejected.forEach { close($0) } } + for i in 1...50 { + rejected.append(connectClient(path)) + #expect(decisions.wait { $0.count >= i }) + } + + let resp = try request(path) + #expect(resp.version == "served") + #expect(decisions.all.count == 51) + } + + @Test("Accept cap of 1 per event still drains a queued burst: the source fires again") + func acceptCapRefires() throws { + let path = uniquePath("cap") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + // 15 connects queued in the backlog BEFORE the server starts: one readable event + // with 15 pending, which a 1-per-event cap can only drain by firing again. + let queued = 15 + var rejected: [Int32] = [] + defer { rejected.forEach { close($0) } } + for _ in 0..() + let fake = FakeAuthorizer { _ in + decisions.append(1) + return decisions.all.count <= queued + ? .reject(PeerCredentials(uid: 4294967294, gid: 4294967294)) + : .allow(PeerCredentials(uid: getuid(), gid: getgid())) + } + startServer(listenFD, fake, maxAcceptsPerEvent: 1) + + let resp = try request(path) + #expect(resp.version == "served") + #expect(decisions.all.count == queued + 1) + } + + // MARK: 4. Credential source fails + + @Test("Unreadable credentials are a rejection: closed, handler never runs, errno logged") + func credentialErrorRejects() throws { + let path = uniquePath("errno") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + let handled = Box() + let logged = Box() + let authorizer = PeerAuthorizer(ownerUID: getuid(), + credentials: { _ in .failure(PeerCredentialError(errno: EINVAL)) }) + startServer(listenFD, authorizer, log: { logged.append($0) }) { _ in + handled.append(1) + return .versionResponse("served") + } + + let client = connectClient(path) + defer { close(client) } + let r = try sendAndRead(client) + + #expect(r.n == 0 || (r.n < 0 && r.err == ECONNRESET)) + #expect(logged.wait { !$0.isEmpty }) + #expect(handled.all.isEmpty) + #expect(logged.all.first?.contains("credentials unavailable: errno \(EINVAL)") == true) + } + + // MARK: 5–6. Real kernel credentials + + @Test("Real getpeereid: owner = our uid is served") + func realKernelAllowsOwner() throws { + let path = uniquePath("real-allow") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + startServer(listenFD, PeerAuthorizer(ownerUID: getuid())) + + let resp = try request(path) + #expect(resp.version == "served") + } + + @Test("Real getpeereid: owner = another uid rejects us", .enabled(if: getuid() != 0)) + func realKernelRejectsNonOwner() throws { + let path = uniquePath("real-reject") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + let handled = Box() + let logged = Box() + startServer(listenFD, PeerAuthorizer(ownerUID: getuid() + 1), + log: { logged.append($0) }) { _ in + handled.append(1) + return .versionResponse("served") + } + + // Not served, AND the server says why: a rejection line naming our uid. A + // request that merely timed out would leave no such line. + #expect(throws: (any Error).self) { try request(path) } + #expect(logged.wait { !$0.isEmpty }) + #expect(handled.all.isEmpty) + #expect(logged.all.first?.contains("euid \(getuid()) egid \(getegid())") == true) + #expect(logged.all.first?.contains("owner uid \(getuid() + 1)") == true) + } + + // MARK: 7. Peer gone before the check + + @Test("Peer closed before the check: real credentials still read as our uid") + func credentialsSurvivePeerClose() throws { + let path = uniquePath("gone") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + // Connect and hang up while the connection still sits in the backlog, then start + // the server: the check runs against a peer that no longer exists. + let client = connectClient(path) + close(client) + + let decisions = Box() + let real = PeerAuthorizer(ownerUID: getuid()) + let recording = FakeAuthorizer { fd in + let d = real.decide(fd: fd) + decisions.append(d) + return d + } + startServer(listenFD, recording) + + #expect(decisions.wait { !$0.isEmpty }) + #expect(decisions.all.first == .allow(PeerCredentials(uid: getuid(), gid: getegid()))) + } + + // MARK: 8. Bounded rejection log + + @Test("1000 rejections in one window: 5 lines, then one summary with the right count") + func logBoundSummary() { + let t0 = Date(timeIntervalSince1970: 1_000_000) + var limiter = RejectionLogLimiter(now: t0) + var lines = 0 + for _ in 0..<1000 { + if limiter.record("rejected", now: t0) != nil { lines += 1 } + } + #expect(lines == 5) + #expect(limiter.suppressed == 995) + #expect(limiter.flush() == "ThermalForge daemon: 995 further peer rejections suppressed") + #expect(limiter.flush() == nil) + } + + @Test("The suppressed count rides on the next line that gets through") + func logBoundCarryForward() { + let t0 = Date(timeIntervalSince1970: 1_000_000) + var limiter = RejectionLogLimiter(now: t0) + for _ in 0..<1000 { _ = limiter.record("rejected", now: t0) } + #expect(limiter.record("rejected", now: t0.addingTimeInterval(30)) == nil) // no token yet + #expect(limiter.record("rejected", now: t0.addingTimeInterval(61)) + == "rejected (996 earlier rejections suppressed)") + #expect(limiter.suppressed == 0) + } + + @Test("Through the server: a 100-rejection flood is logged in bounded lines that account for every rejection") + func logBoundThroughServer() throws { + let path = uniquePath("flood") + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + let logged = Box() + let decisions = Box() + let rejectAll = FakeAuthorizer.rejectAll() + let counting = FakeAuthorizer { fd in decisions.append(1); return rejectAll.decide(fd: fd) } + let start = Date() + startServer(listenFD, counting, summaryDelay: 0.5, log: { logged.append($0) }) + + var fds: [Int32] = [] + defer { fds.forEach { close($0) } } + for i in 1...100 { // paced so the test's 16-deep backlog never overflows + fds.append(connectClient(path)) + #expect(decisions.wait { $0.count >= i }) + } + + // Every rejection is accounted for exactly once: its own line, a count carried + // on a later line, or a timed summary. Wait for that outcome, not for a time. + func accounted(_ lines: [String]) -> Int { + lines.reduce(0) { total, line in + if let n = Self.count(in: line, after: "ThermalForge daemon: ", + before: " further peer rejections suppressed") { + return total + n + } + return total + 1 + (Self.count(in: line, after: "(", before: " earlier rejections suppressed)") ?? 0) + } + } + #expect(logged.wait { accounted($0) == 100 }) + + // The 5-line burst, plus at most one refill per full minute the test ran (the + // limiter reads real time here; its exact boundaries are unit-tested above). + let rejectionLines = logged.all.filter { $0.contains("rejected peer") }.count + let refills = Int(Date().timeIntervalSince(start) / 60) + #expect(rejectionLines >= 5) + #expect(rejectionLines <= 5 + refills) + #expect(logged.all.contains { $0.hasSuffix(" further peer rejections suppressed") }) + } + } +} diff --git a/Tests/ThermalForgeTests/SIGPIPETests.swift b/Tests/ThermalForgeTests/SIGPIPETests.swift new file mode 100644 index 00000000..cc182d27 --- /dev/null +++ b/Tests/ThermalForgeTests/SIGPIPETests.swift @@ -0,0 +1,145 @@ +// +// SIGPIPETests.swift +// ThermalForge +// +// A write to a socket whose peer has gone away must fail with EPIPE, never raise +// SIGPIPE — whose default action terminates the process. Daemon side: a client sends +// a valid frame and closes before the reply is written. Client side: the daemon closes +// before replying. A counting handler stands in for the default action so a raised +// SIGPIPE is observed instead of killing the test runner. +// + +import Darwin +import Foundation +import Testing + +@testable import ThermalForgeCore + +nonisolated(unsafe) private var sigpipeCount: sig_atomic_t = 0 +private func countSIGPIPE(_: Int32) { sigpipeCount += 1 } + +/// Run `body` with SIGPIPE counted rather than fatal; returns how many were raised. +/// Restores the previous disposition afterwards. +private func countingSIGPIPE(_ body: () throws -> Void) rethrows -> Int { + sigpipeCount = 0 + let previous = signal(SIGPIPE, countSIGPIPE) + defer { signal(SIGPIPE, previous) } + // 0 = SIG_DFL (terminate), 1 = SIG_IGN. Shows what the raised signal would have done. + print("SIGPIPE disposition before test: \(unsafeBitCast(previous, to: Int.self))") + try body() + return Int(sigpipeCount) +} + +extension SocketSuites { + @Suite("SIGPIPE on a vanished peer") + struct SIGPIPETests { + + private func setPath(_ addr: inout sockaddr_un, _ path: String) { + withUnsafeMutablePointer(to: &addr.sun_path) { ptr in + ptr.withMemoryRebound(to: CChar.self, capacity: 104) { _ = strlcpy($0, path, 104) } + } + } + + private func bindListener(_ path: String) -> Int32 { + unlink(path) + let fd = socket(AF_UNIX, SOCK_STREAM, 0) + var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) + let len = socklen_t(MemoryLayout.size) + let r = withUnsafePointer(to: &addr) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { bind(fd, $0, len) } + } + #expect(r == 0) + #expect(listen(fd, 16) == 0) + return fd + } + + private func connectClient(_ path: String) -> Int32 { + let fd = socket(AF_UNIX, SOCK_STREAM, 0) + var addr = sockaddr_un(); addr.sun_family = sa_family_t(AF_UNIX); setPath(&addr, path) + let len = socklen_t(MemoryLayout.size) + let r = withUnsafePointer(to: &addr) { + $0.withMemoryRebound(to: sockaddr.self, capacity: 1) { connect(fd, $0, len) } + } + #expect(r == 0) + return fd + } + + @Test("Daemon: client closes before the reply is written — no SIGPIPE, daemon keeps serving") + func daemonSurvivesVanishedClient() throws { + let path = "/tmp/tf-sigpipe-daemon-\(getpid()).sock" + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + // The first request parks in `handle` until the client has closed, so the + // reply is written to a dead peer. Later requests are answered normally. + let requestArrived = DispatchSemaphore(value: 0) + let clientClosed = DispatchSemaphore(value: 0) + let first = NSLock(); var isFirst = true + // maxConnections 1: the follow-up request below can only be accepted once this + // connection has finished, i.e. after its reply write to the dead peer completed. + // So a served follow-up proves the write happened while SIGPIPE was being counted. + let server = ConnectionServer(listenFD: listenFD, authorizer: FakeAuthorizer.allowAll(), + maxConnections: 1, + headerDeadline: SocketTestLimits.serverDeadline, + requestDeadline: SocketTestLimits.serverDeadline) { _ in + first.lock(); let park = isFirst; isFirst = false; first.unlock() + if park { + requestArrived.signal() + clientClosed.wait() + } + return .versionResponse("served") + } + server.start() + + let frame = try DaemonProtocol.encodeFrame(DaemonRequest(verb: .version), + max: DaemonProtocol.maxRequestBytes) + let raised = try countingSIGPIPE { + let client = connectClient(path) + _ = frame.withUnsafeBytes { write(client, $0.baseAddress, frame.count) } + #expect(requestArrived.wait(timeout: .now() + SocketTestLimits.wait) == .success) + close(client) + clientClosed.signal() + + // The daemon must still be up and serving. + let resp = try DaemonClient(socketPath: path).request(DaemonRequest(verb: .version), + timeout: SocketTestLimits.wait) + #expect(resp.version == "served") + } + print("SIGPIPE daemon-side: \(raised) raised") + #expect(raised == 0) + } + + @Test("Client: daemon closes before replying — request throws, no SIGPIPE") + func clientSurvivesVanishedDaemon() throws { + let path = "/tmp/tf-sigpipe-client-\(getpid()).sock" + let listenFD = bindListener(path) + defer { close(listenFD); unlink(path) } + + // A peer that accepts and immediately closes without reading. Whether a given + // client write lands before or after that close is a race, so run many rounds: + // any round where the close wins makes an unprotected write raise SIGPIPE. + let rounds = 300 + let acceptor = Thread { + for _ in 0..= 0 { close(fd) } + } + } + acceptor.start() + + var threw = 0 + let raised = countingSIGPIPE { + let client = DaemonClient(socketPath: path) + for _ in 0..