Daemon: authenticate socket peers

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<Int32>.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

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<timeval>.size))
         setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, socklen_t(MemoryLayout<timeval>.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<Int32>.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
 

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<PeerCredentials, PeerCredentialError>
+
+    init(ownerUID: uid_t,
+         credentials: @escaping (Int32) -> Result<PeerCredentials, PeerCredentialError> = 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<PeerCredentials, PeerCredentialError> {
+        // 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"
+    }
+}