row editing

DBSyn/MySQLProtocol.swift

@@ -3,6 +3,7 @@ import CommonCrypto
 import CoreFoundation
 import CFNetwork
 import Security
+import Darwin
 
 // Native MySQL wire protocol client. No external mysql executable is used.
 // Initial scope: TCP, mysql_native_password and caching_sha2_password fast authentication.
@@ -198,9 +199,107 @@ private func mysqlRSAPublicKey(_ packet: [UInt8]) throws -> SecKey {
     return key
 }
 
+// SecureTransport operates on the same socket before and after MySQL SSLRequest.
+// CFStream SSL properties cannot reliably retrofit TLS onto already-open streams.
+private final class MySQLSocketTransport {
+    let fd: Int32
+    private var ssl: SSLContext?
+
+    init(host: String, port: Int) throws {
+        var hints = addrinfo()
+        hints.ai_family = AF_UNSPEC
+        hints.ai_socktype = Int32(SOCK_STREAM.rawValue)
+        hints.ai_protocol = Int32(IPPROTO_TCP)
+        var addresses: UnsafeMutablePointer<addrinfo>?
+        let result = getaddrinfo(host, String(port), &hints, &addresses)
+        guard result == 0 else { throw MySQLWireError(message: "Cannot resolve MySQL host: \(String(cString: gai_strerror(result)))") }
+        defer { freeaddrinfo(addresses) }
+        var candidate = addresses
+        var connected: Int32 = -1
+        while let item = candidate {
+            let info = item.pointee
+            let sock = Darwin.socket(info.ai_family, info.ai_socktype, info.ai_protocol)
+            if sock >= 0 {
+                if Darwin.connect(sock, info.ai_addr, info.ai_addrlen) == 0 {
+                    connected = sock
+                    break
+                }
+                _ = Darwin.close(sock)
+            }
+            candidate = info.ai_next
+        }
+        guard connected >= 0 else { throw MySQLWireError(message: "Cannot connect to MySQL at \(host):\(port): \(String(cString: strerror(errno)))") }
+        fd = connected
+    }
+
+    deinit {
+        if let ssl { _ = SSLClose(ssl) }
+        _ = Darwin.close(fd)
+    }
+
+    func read(_ buffer: UnsafeMutablePointer<UInt8>, count: Int) throws -> Int {
+        if let ssl {
+            var processed = 0
+            let status = SSLRead(ssl, buffer, count, &processed)
+            if status == errSSLSuccess || (status == errSSLWouldBlock && processed > 0) { return processed }
+            throw MySQLWireError(message: "MySQL TLS read failed (OSStatus \(status))")
+        }
+        let n = Darwin.recv(fd, buffer, count, 0)
+        guard n > 0 else { throw MySQLWireError(message: n == 0 ? "MySQL connection closed" : "MySQL read failed: \(String(cString: strerror(errno)))") }
+        return n
+    }
+
+    func write(_ buffer: UnsafePointer<UInt8>, count: Int) throws -> Int {
+        if let ssl {
+            var processed = 0
+            let status = SSLWrite(ssl, buffer, count, &processed)
+            if status == errSSLSuccess || (status == errSSLWouldBlock && processed > 0) { return processed }
+            throw MySQLWireError(message: "MySQL TLS write failed (OSStatus \(status))")
+        }
+        let n = Darwin.send(fd, buffer, count, 0)
+        guard n > 0 else { throw MySQLWireError(message: "MySQL write failed: \(String(cString: strerror(errno)))") }
+        return n
+    }
+
+    func startTLS(host: String, verify: Bool) throws {
+        guard let context = SSLCreateContext(nil, .clientSide, .streamType) else {
+            throw MySQLWireError(message: "Unable to create MySQL TLS context")
+        }
+        // SecureTransport callback connections refer to this retained transport.
+        let connection = Unmanaged.passUnretained(self).toOpaque()
+        var status = SSLSetIOFuncs(context, { connection, data, length in
+            guard let connection, let data, let length else { return errSecParam }
+            let transport = Unmanaged<MySQLSocketTransport>.fromOpaque(connection).takeUnretainedValue()
+            let requested = length.pointee
+            let n = Darwin.recv(transport.fd, data, requested, 0)
+            if n > 0 { length.pointee = n; return errSecSuccess }
+            length.pointee = 0
+            return n == 0 ? errSSLClosedGraceful : errSecIO
+        }, { connection, data, length in
+            guard let connection, let data, let length else { return errSecParam }
+            let transport = Unmanaged<MySQLSocketTransport>.fromOpaque(connection).takeUnretainedValue()
+            let n = Darwin.send(transport.fd, data, length.pointee, 0)
+            if n > 0 { length.pointee = n; return errSecSuccess }
+            length.pointee = 0
+            return errSecIO
+        })
+        guard status == errSecSuccess else { throw MySQLWireError(message: "TLS I/O setup failed (\(status))") }
+        status = SSLSetConnection(context, connection)
+        guard status == errSecSuccess else { throw MySQLWireError(message: "TLS socket setup failed (\(status))") }
+        status = host.withCString { SSLSetPeerDomainName(context, $0, host.utf8.count) }
+        guard status == errSecSuccess else { throw MySQLWireError(message: "TLS peer setup failed (\(status))") }
+        // REQUIRED encrypts but does not authenticate the server. Keep verification
+        // enabled for the verified mode; never silently fall back to REQUIRED.
+        status = SSLSetEnableCertVerify(context, verify)
+        guard status == errSecSuccess else { throw MySQLWireError(message: "TLS verification setup failed (\(status))") }
+        repeat { status = SSLHandshake(context) } while status == errSSLWouldBlock
+        guard status == errSecSuccess else { throw MySQLWireError(message: "MySQL TLS handshake failed (OSStatus \(status))") }
+        ssl = context
+    }
+}
+
 final class MySQLWireConnection {
-    private let input: InputStream
-    private let output: OutputStream
+    private let transport: MySQLSocketTransport
     private var sequence: UInt8 = 0
     private let maxPacket = 16_777_215
     private let serverHost: String
@@ -213,32 +312,16 @@ final class MySQLWireConnection {
         self.requireTLS = requireTLS
         self.verifyTLS = verifyTLS
         guard (1...65535).contains(port) else { throw MySQLWireError(message: "Invalid port") }
-        var read: Unmanaged<CFReadStream>?
-        var write: Unmanaged<CFWriteStream>?
-        CFStreamCreatePairWithSocketToHost(nil, host as CFString, UInt32(port), &read, &write)
-        guard let r = read?.takeRetainedValue(), let w = write?.takeRetainedValue() else {
-            throw MySQLWireError(message: "Cannot create connection")
-        }
-        input = unsafeBitCast(r, to: InputStream.self)
-        output = unsafeBitCast(w, to: OutputStream.self)
-        input.open(); output.open()
-        // A CFStream may still be .opening immediately after open(). The
-        // first blocking read below waits for the server greeting.
-        if input.streamStatus == .error || output.streamStatus == .error {
-            throw MySQLWireError(message: input.streamError?.localizedDescription ?? output.streamError?.localizedDescription ?? "Could not open MySQL connection")
-        }
+        transport = try MySQLSocketTransport(host: host, port: port)
     }
-    deinit { input.close(); output.close() }
 
     private func readExact(_ count: Int) throws -> [UInt8] {
         var buffer = [UInt8](repeating: 0, count: count)
         var done = 0
         while done < count {
-            let n = buffer.withUnsafeMutableBufferPointer { ptr -> Int in
-                guard let base = ptr.baseAddress else { return -1 }
-                return input.read(base.advanced(by: done), maxLength: count - done)
+            let n = try buffer.withUnsafeMutableBufferPointer { ptr -> Int in
+                try transport.read(ptr.baseAddress!.advanced(by: done), count: count - done)
             }
-            guard n > 0 else { throw MySQLWireError(message: input.streamError?.localizedDescription ?? "MySQL connection closed") }
             done += n
         }
         return buffer
@@ -247,9 +330,7 @@ final class MySQLWireConnection {
         let header = try readExact(4)
         let count = Int(header[0]) | Int(header[1]) << 8 | Int(header[2]) << 16
         sequence = header[3] &+ 1
-        guard count <= maxPacket else {
-            throw MySQLWireError(message: "MySQL packet exceeds protocol limit")
-        }
+        guard count <= maxPacket else { throw MySQLWireError(message: "MySQL packet exceeds protocol limit") }
         return try readExact(count)
     }
     private func send(_ payload: [UInt8]) throws {
@@ -259,30 +340,14 @@ final class MySQLWireConnection {
         sequence = sequence &+ 1
         var written = 0
         while written < packet.count {
-            let n = packet.withUnsafeBufferPointer { ptr -> Int in
-                guard let base = ptr.baseAddress else { return -1 }
-                return output.write(base.advanced(by: written), maxLength: packet.count - written)
+            let n = try packet.withUnsafeBufferPointer { ptr -> Int in
+                try transport.write(ptr.baseAddress!.advanced(by: written), count: packet.count - written)
             }
-            guard n > 0 else { throw MySQLWireError(message: output.streamError?.localizedDescription ?? "MySQL write failed") }
             written += n
         }
     }
-    // MySQL TLS is negotiated AFTER the server greeting: first send CLIENT_SSL,
-    // then upgrade the existing TCP streams, then send the authentication packet.
-    // Never send credentials if TLS negotiation or certificate verification fails.
     private func upgradeToTLS() throws {
-        let settings: [String: Any] = [
-            kCFStreamSSLLevel as String: kCFStreamSocketSecurityLevelNegotiatedSSL,
-            kCFStreamSSLValidatesCertificateChain as String: verifyTLS,
-            kCFStreamSSLPeerName as String: verifyTLS ? serverHost : kCFNull as Any
-        ]
-        let key = Stream.PropertyKey(rawValue: kCFStreamPropertySSLSettings as String)
-        guard input.setProperty(settings, forKey: key),
-              output.setProperty(settings, forKey: key) else {
-            throw MySQLWireError(message: "TLS stream upgrade unavailable; no credentials sent. Use SSH tunnelling for this server.")
-        }
-        // CFStream performs the TLS handshake before delivering encrypted data.
-        // Writing authentication triggers the handshake; any failure aborts it.
+        try transport.startTLS(host: serverHost, verify: verifyTLS)
         tlsActive = true
     }