747 lines
33 KiB
Swift
747 lines
33 KiB
Swift
import Foundation
|
|
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.
|
|
struct MySQLWireError: LocalizedError {
|
|
let message: String
|
|
var errorDescription: String? { message }
|
|
}
|
|
|
|
// SSH transport uses the system OpenSSH client and key-based authentication.
|
|
// No remote MySQL password is passed to ssh. SSH host-key verification remains enabled.
|
|
private final class DBSynSSHTunnels {
|
|
static let shared = DBSynSSHTunnels()
|
|
private let lock = NSLock()
|
|
private var running: [UUID: (Process, ConnectionProfile, Int)] = [:]
|
|
|
|
func endpoint(for profile: ConnectionProfile) throws -> (String, Int) {
|
|
guard profile.useSSHTunnel else { return (profile.host, profile.port) }
|
|
lock.lock(); defer { lock.unlock() }
|
|
if let current = running[profile.id] {
|
|
if current.0.isRunning && current.1 == profile { return ("127.0.0.1", current.2) }
|
|
if current.0.isRunning { current.0.terminate() }
|
|
running.removeValue(forKey: profile.id)
|
|
}
|
|
guard !profile.sshHost.isEmpty, !profile.sshUser.isEmpty else {
|
|
throw MySQLWireError(message: "Configure SSH hostname and SSH username in Connection Settings")
|
|
}
|
|
// A stable local port per connection; ExitOnForwardFailure rejects conflicts.
|
|
let bytes = Array(profile.id.uuidString.utf8)
|
|
let checksum = bytes.reduce(0) { ($0 &* 31 &+ Int($1)) % 10000 }
|
|
let localPort = 45000 + checksum
|
|
let process = Process()
|
|
process.executableURL = URL(fileURLWithPath: "/usr/bin/ssh")
|
|
var args = ["-N", "-T", "-o", "BatchMode=yes", "-o", "ExitOnForwardFailure=yes", "-o", "ConnectTimeout=8", "-o", "StrictHostKeyChecking=yes", "-p", String(profile.sshPort)]
|
|
if !profile.sshIdentityFile.isEmpty { args += ["-i", (profile.sshIdentityFile as NSString).expandingTildeInPath] }
|
|
args += ["-L", "127.0.0.1:\(localPort):\(profile.host):\(profile.port)", "\(profile.sshUser)@\(profile.sshHost)"]
|
|
process.arguments = args
|
|
process.standardInput = FileHandle.nullDevice
|
|
process.standardOutput = FileHandle.nullDevice
|
|
process.standardError = FileHandle.nullDevice
|
|
do { try process.run() } catch { throw MySQLWireError(message: "Cannot launch SSH: \(error.localizedDescription)") }
|
|
// SSH may take several seconds to authenticate. A running process does
|
|
// not mean its local listener is ready. Probe the forwarded endpoint
|
|
// before handing it to the MySQL handshake.
|
|
let deadline = Date().addingTimeInterval(10)
|
|
var ready = false
|
|
while Date() < deadline && process.isRunning {
|
|
let probe = Process()
|
|
probe.executableURL = URL(fileURLWithPath: "/usr/bin/nc")
|
|
probe.arguments = ["-z", "-G", "1", "127.0.0.1", String(localPort)]
|
|
probe.standardInput = FileHandle.nullDevice
|
|
probe.standardOutput = FileHandle.nullDevice
|
|
probe.standardError = FileHandle.nullDevice
|
|
if (try? probe.run()) != nil {
|
|
probe.waitUntilExit()
|
|
if probe.terminationStatus == 0 { ready = true; break }
|
|
}
|
|
Thread.sleep(forTimeInterval: 0.15)
|
|
}
|
|
guard ready else {
|
|
if process.isRunning { process.terminate() }
|
|
throw MySQLWireError(message: "SSH forwarding did not become ready within 10 seconds (local port \(localPort)). Check SSH credentials, host key and target address.")
|
|
}
|
|
running[profile.id] = (process, profile, localPort)
|
|
return ("127.0.0.1", localPort)
|
|
}
|
|
func stop(_ id: UUID) {
|
|
lock.lock(); defer { lock.unlock() }
|
|
if let existing = running.removeValue(forKey: id), existing.0.isRunning { existing.0.terminate() }
|
|
}
|
|
}
|
|
|
|
private struct ByteCursor {
|
|
let bytes: [UInt8]
|
|
var offset = 0
|
|
mutating func take(_ n: Int) throws -> [UInt8] {
|
|
guard n >= 0, offset + n <= bytes.count else { throw MySQLWireError(message: "Malformed MySQL packet") }
|
|
defer { offset += n }
|
|
return Array(bytes[offset..<(offset+n)])
|
|
}
|
|
mutating func u8() throws -> UInt8 { try take(1)[0] }
|
|
mutating func u16() throws -> Int { let b = try take(2); return Int(b[0]) | (Int(b[1]) << 8) }
|
|
mutating func u32() throws -> Int { let b = try take(4); return Int(b[0]) | (Int(b[1]) << 8) | (Int(b[2]) << 16) | (Int(b[3]) << 24) }
|
|
mutating func nulString() throws -> String {
|
|
let start = offset
|
|
while offset < bytes.count && bytes[offset] != 0 { offset += 1 }
|
|
guard offset < bytes.count else { throw MySQLWireError(message: "Unterminated server string") }
|
|
let s = String(decoding: bytes[start..<offset], as: UTF8.self)
|
|
offset += 1
|
|
return s
|
|
}
|
|
mutating func lenInt() throws -> Int? {
|
|
let first = try u8()
|
|
switch first {
|
|
case 0xfb: return nil
|
|
case 0xfc: return try u16()
|
|
case 0xfd:
|
|
let b = try take(3); return Int(b[0]) | Int(b[1]) << 8 | Int(b[2]) << 16
|
|
case 0xfe:
|
|
let b = try take(8)
|
|
var v: UInt64 = 0
|
|
for (i, byte) in b.enumerated() { v |= UInt64(byte) << (8 * i) }
|
|
guard v <= UInt64(Int.max) else { throw MySQLWireError(message: "Field too large") }
|
|
return Int(v)
|
|
default: return Int(first)
|
|
}
|
|
}
|
|
mutating func lenString() throws -> String? {
|
|
guard let n = try lenInt() else { return nil }
|
|
return String(decoding: try take(n), as: UTF8.self)
|
|
}
|
|
}
|
|
|
|
private func le32(_ v: Int) -> [UInt8] { (0..<4).map { UInt8(truncatingIfNeeded: v >> ($0 * 8)) } }
|
|
private func sha1(_ data: [UInt8]) -> [UInt8] {
|
|
var result = [UInt8](repeating: 0, count: Int(CC_SHA1_DIGEST_LENGTH))
|
|
data.withUnsafeBytes { ptr in _ = CC_SHA1(ptr.baseAddress, CC_LONG(data.count), &result) }
|
|
return result
|
|
}
|
|
private func sha256(_ data: [UInt8]) -> [UInt8] {
|
|
var result = [UInt8](repeating: 0, count: Int(CC_SHA256_DIGEST_LENGTH))
|
|
data.withUnsafeBytes { ptr in _ = CC_SHA256(ptr.baseAddress, CC_LONG(data.count), &result) }
|
|
return result
|
|
}
|
|
private func authResponse(password: String, salt: [UInt8], plugin: String) throws -> [UInt8] {
|
|
if password.isEmpty { return [] }
|
|
let pwd = Array(password.utf8)
|
|
if plugin == "mysql_native_password" {
|
|
let first = sha1(pwd), second = sha1(sha1(pwd))
|
|
let last = sha1(salt + second)
|
|
return zip(first, last).map { $0 ^ $1 }
|
|
}
|
|
if plugin == "caching_sha2_password" {
|
|
// MySQL caching_sha2_password authentication:
|
|
// XOR(SHA256(password),
|
|
// SHA256(SHA256(SHA256(password)) + challenge))
|
|
let first = sha256(pwd)
|
|
let second = sha256(first)
|
|
let third = sha256(second)
|
|
let last = sha256(third + salt)
|
|
|
|
return zip(first, last).map { $0 ^ $1 }
|
|
}
|
|
throw MySQLWireError(message: "Authentication plugin \(plugin) is not yet supported")
|
|
}
|
|
|
|
// Parse an RSA SubjectPublicKeyInfo PEM supplied by MySQL. Security.framework
|
|
// expects the inner PKCS#1 RSAPublicKey DER for SecKeyCreateWithData.
|
|
private func mysqlRSAPublicKey(_ packet: [UInt8]) throws -> SecKey {
|
|
guard packet.first == 0x01 else {
|
|
throw MySQLWireError(message: "Expected caching_sha2_password public-key response")
|
|
}
|
|
let pem = String(decoding: packet.dropFirst(), as: UTF8.self)
|
|
let base64 = pem.components(separatedBy: .newlines)
|
|
.filter { !$0.hasPrefix("-----") }.joined()
|
|
.trimmingCharacters(in: .whitespacesAndNewlines)
|
|
guard let spki = Data(base64Encoded: base64) else {
|
|
throw MySQLWireError(message: "Invalid MySQL RSA public key")
|
|
}
|
|
let der = [UInt8](spki)
|
|
func length(_ index: inout Int) throws -> Int {
|
|
guard index < der.count else { throw MySQLWireError(message: "Invalid RSA DER length") }
|
|
let first = Int(der[index]); index += 1
|
|
if first < 128 { return first }
|
|
let bytes = first & 0x7f
|
|
guard bytes > 0, bytes <= 4, index + bytes <= der.count else {
|
|
throw MySQLWireError(message: "Invalid RSA DER length")
|
|
}
|
|
var value = 0
|
|
for _ in 0..<bytes { value = (value << 8) | Int(der[index]); index += 1 }
|
|
return value
|
|
}
|
|
func element(_ index: inout Int, tag: UInt8) throws -> Range<Int> {
|
|
guard index < der.count, der[index] == tag else {
|
|
throw MySQLWireError(message: "Invalid RSA DER element")
|
|
}
|
|
index += 1
|
|
let count = try length(&index)
|
|
guard count <= der.count - index else { throw MySQLWireError(message: "Truncated RSA DER") }
|
|
let range = index..<(index + count)
|
|
index += count
|
|
return range
|
|
}
|
|
var i = 0
|
|
let outer = try element(&i, tag: 0x30)
|
|
i = outer.lowerBound
|
|
_ = try element(&i, tag: 0x30) // AlgorithmIdentifier
|
|
let bits = try element(&i, tag: 0x03)
|
|
guard bits.count > 1, der[bits.lowerBound] == 0 else {
|
|
throw MySQLWireError(message: "Invalid RSA public key bit string")
|
|
}
|
|
let pkcs1 = Data(der[(bits.lowerBound + 1)..<bits.upperBound])
|
|
var error: Unmanaged<CFError>?
|
|
guard let key = SecKeyCreateWithData(pkcs1 as CFData, [
|
|
kSecAttrKeyType as String: kSecAttrKeyTypeRSA,
|
|
kSecAttrKeyClass as String: kSecAttrKeyClassPublic
|
|
] as CFDictionary, &error) else {
|
|
throw MySQLWireError(message: "Cannot import MySQL RSA public key: \(error?.takeRetainedValue().localizedDescription ?? "unknown error")")
|
|
}
|
|
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)
|
|
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 == errSecSuccess || (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 == errSecSuccess || (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 {
|
|
NSLog("DBSyn TLS: host=%@ verify=%@", host, verify ? "ON" : "OFF")
|
|
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
|
|
|
|
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
|
|
|
|
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.
|
|
// Certificate verification is handled after the TLS handshake.
|
|
status = SSLSetSessionOption(context, .breakOnServerAuth, true)
|
|
guard status == errSecSuccess else {
|
|
throw MySQLWireError(message: "TLS authentication setup failed (\(status))")
|
|
}
|
|
|
|
repeat {
|
|
status = SSLHandshake(context)
|
|
} while status == errSSLWouldBlock
|
|
|
|
if status == OSStatus(-9841) {
|
|
if verify {
|
|
var trust: SecTrust?
|
|
let trustStatus = SSLCopyPeerTrust(context, &trust)
|
|
|
|
guard trustStatus == errSecSuccess, let trust else {
|
|
throw MySQLWireError(message: "Unable to obtain MySQL server trust")
|
|
}
|
|
|
|
var trustError: CFError?
|
|
|
|
guard SecTrustEvaluateWithError(trust, &trustError) else {
|
|
throw MySQLWireError(
|
|
message: "MySQL server verification failed (verifyTLS=\(verify)): \(trustError?.localizedDescription ?? "untrusted server")"
|
|
)
|
|
}
|
|
}
|
|
|
|
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 transport: MySQLSocketTransport
|
|
private var sequence: UInt8 = 0
|
|
private let maxPacket = 16_777_215
|
|
private let serverHost: String
|
|
private let requireTLS: Bool
|
|
private let verifyTLS: Bool
|
|
private var tlsActive = false
|
|
|
|
init(host: String, port: Int, requireTLS: Bool = false, verifyTLS: Bool = true) throws {
|
|
self.serverHost = host
|
|
self.requireTLS = requireTLS
|
|
self.verifyTLS = verifyTLS
|
|
guard (1...65535).contains(port) else { throw MySQLWireError(message: "Invalid port") }
|
|
transport = try MySQLSocketTransport(host: host, port: port)
|
|
}
|
|
|
|
private func readExact(_ count: Int) throws -> [UInt8] {
|
|
var buffer = [UInt8](repeating: 0, count: count)
|
|
var done = 0
|
|
while done < count {
|
|
let n = try buffer.withUnsafeMutableBufferPointer { ptr -> Int in
|
|
try transport.read(ptr.baseAddress!.advanced(by: done), count: count - done)
|
|
}
|
|
done += n
|
|
}
|
|
return buffer
|
|
}
|
|
private func readPacket() throws -> [UInt8] {
|
|
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") }
|
|
return try readExact(count)
|
|
}
|
|
private func send(_ payload: [UInt8]) throws {
|
|
guard payload.count < maxPacket else { throw MySQLWireError(message: "Query exceeds initial client packet limit") }
|
|
let count = payload.count
|
|
let packet = [UInt8(count & 255), UInt8((count >> 8) & 255), UInt8((count >> 16) & 255), sequence] + payload
|
|
sequence = sequence &+ 1
|
|
var written = 0
|
|
while written < packet.count {
|
|
let n = try packet.withUnsafeBufferPointer { ptr -> Int in
|
|
try transport.write(ptr.baseAddress!.advanced(by: written), count: packet.count - written)
|
|
}
|
|
written += n
|
|
}
|
|
}
|
|
private func upgradeToTLS() throws {
|
|
try transport.startTLS(host: serverHost, verify: verifyTLS)
|
|
tlsActive = true
|
|
}
|
|
|
|
private func checkError(_ p: [UInt8]) throws {
|
|
guard p.first == 0xff else { return }
|
|
let details = String(decoding: p.dropFirst(3), as: UTF8.self)
|
|
throw MySQLWireError(message: "MySQL error: \(details)")
|
|
}
|
|
func authenticate(user: String, password: String, database: String) throws {
|
|
let hello = try readPacket()
|
|
var c = ByteCursor(bytes: hello)
|
|
let version = try c.u8()
|
|
guard version == 10 else { throw MySQLWireError(message: "Unsupported MySQL handshake") }
|
|
_ = try c.nulString()
|
|
_ = try c.u32()
|
|
let salt1 = try c.take(8)
|
|
_ = try c.u8()
|
|
let lower = try c.u16()
|
|
_ = try c.u8() // character set
|
|
_ = try c.u16() // status
|
|
let upper = try c.u16()
|
|
let capabilities = lower | (upper << 16)
|
|
let authLen = Int(try c.u8())
|
|
_ = try c.take(10)
|
|
let salt2 = try c.take(max(13, authLen - 8))
|
|
let salt = salt1 + salt2.prefix(max(0, authLen - 9))
|
|
|
|
// MySQL authentication uses the 20-byte server challenge.
|
|
guard salt.count == 20 else {
|
|
throw MySQLWireError(
|
|
message: "Invalid MySQL authentication challenge length: \(salt.count)"
|
|
)
|
|
}
|
|
let plugin = (try? c.nulString()) ?? "mysql_native_password"
|
|
|
|
NSLog(
|
|
"DBSyn MySQL authentication: plugin=%@ saltLength=%d tlsRequired=%@",
|
|
plugin,
|
|
salt.count,
|
|
requireTLS ? "YES" : "NO"
|
|
)
|
|
if requireTLS {
|
|
let clientSSL = 0x00000800
|
|
guard capabilities & clientSSL != 0 else {
|
|
throw MySQLWireError(message: "MySQL server does not advertise TLS; refusing to send credentials")
|
|
}
|
|
let sslFlags = 0x00000001 | 0x00000004 | 0x00000200 | 0x00000800 | 0x00002000 | 0x00008000 | 0x00080000
|
|
let request = le32(sslFlags & capabilities) + le32(16_777_216) + [UInt8(45)] + [UInt8](repeating: 0, count: 23)
|
|
try send(request)
|
|
try upgradeToTLS()
|
|
} else if !hostIsLoopback {
|
|
throw MySQLWireError(message: "Remote connections require verified TLS or an SSH tunnel")
|
|
}
|
|
// CLIENT_LONG_PASSWORD, LONG_FLAG, PROTOCOL_41, TRANSACTIONS,
|
|
// SECURE_CONNECTION and PLUGIN_AUTH. Do not set DEPRECATE_EOF.
|
|
let clientFlags = 0x00000001 | 0x00000004 | 0x00000200 | (requireTLS ? 0x00000800 : 0) | 0x00002000 | 0x00008000 | 0x00080000
|
|
let response = try authResponse(password: password, salt: Array(salt), plugin: plugin)
|
|
var body = le32(clientFlags & capabilities) + le32(16_777_216) + [UInt8(45)] + [UInt8](repeating: 0, count: 23)
|
|
body += Array(user.utf8) + [0, UInt8(response.count)] + response
|
|
body += Array(plugin.utf8) + [0]
|
|
try send(body)
|
|
var reply = try readPacket()
|
|
|
|
NSLog(
|
|
"DBSyn MySQL initial authentication response: type=0x%02X length=%d",
|
|
Int(reply.first ?? 0),
|
|
reply.count
|
|
)
|
|
|
|
try checkError(reply)
|
|
var currentPlugin = plugin
|
|
var currentSalt = Array(salt)
|
|
if reply.first == 0xfe {
|
|
var switchCursor = ByteCursor(bytes: Array(reply.dropFirst()))
|
|
let newPlugin = try switchCursor.nulString()
|
|
let rawSalt = Array(reply.dropFirst(1 + newPlugin.utf8.count + 1))
|
|
let newSalt = rawSalt.last == 0 ? Array(rawSalt.dropLast()) : rawSalt
|
|
currentPlugin = newPlugin
|
|
currentSalt = newSalt
|
|
try send(authResponse(password: password, salt: newSalt, plugin: newPlugin))
|
|
reply = try readPacket()
|
|
try checkError(reply)
|
|
}
|
|
if reply.first == 0x01 {
|
|
guard currentPlugin == "caching_sha2_password", reply.count >= 2 else {
|
|
throw MySQLWireError(message: "Unsupported authentication continuation")
|
|
}
|
|
NSLog(
|
|
"DBSyn MySQL authentication continuation: code=0x%02X TLS=%@",
|
|
Int(reply[1]),
|
|
tlsActive ? "YES" : "NO"
|
|
)
|
|
|
|
switch reply[1] {
|
|
case 0x03: // Fast authentication accepted; wait for OK.
|
|
reply = try readPacket()
|
|
case 0x04: // Full authentication required.
|
|
guard hostIsLoopback || tlsActive else {
|
|
throw MySQLWireError(message: "Full authentication requires encrypted transport")
|
|
}
|
|
if tlsActive {
|
|
try send(Array(password.utf8) + [0])
|
|
} else if password.isEmpty {
|
|
try send([0])
|
|
} else {
|
|
// RSA fallback is restricted to trusted loopback transport.
|
|
try send([0x02])
|
|
let keyPacket = try readPacket()
|
|
try checkError(keyPacket)
|
|
let key = try mysqlRSAPublicKey(keyPacket)
|
|
guard !currentSalt.isEmpty else {
|
|
throw MySQLWireError(message: "Missing MySQL authentication nonce")
|
|
}
|
|
var scrambled = Array(password.utf8) + [0]
|
|
for i in scrambled.indices { scrambled[i] ^= currentSalt[i % currentSalt.count] }
|
|
var rsaError: Unmanaged<CFError>?
|
|
guard let encrypted = SecKeyCreateEncryptedData(
|
|
key, .rsaEncryptionOAEPSHA1, Data(scrambled) as CFData, &rsaError
|
|
) as Data? else {
|
|
throw MySQLWireError(message: "MySQL RSA encryption failed: \(rsaError?.takeRetainedValue().localizedDescription ?? "unknown error")")
|
|
}
|
|
try send(Array(encrypted))
|
|
}
|
|
reply = try readPacket()
|
|
default:
|
|
throw MySQLWireError(message: "Unsupported caching_sha2_password response: \(reply[1])")
|
|
}
|
|
}
|
|
NSLog(
|
|
"DBSyn MySQL final authentication response: type=0x%02X length=%d",
|
|
Int(reply.first ?? 0),
|
|
reply.count
|
|
)
|
|
|
|
try checkError(reply)
|
|
guard reply.first == 0x00 else { throw MySQLWireError(message: "Authentication did not complete") }
|
|
if !database.isEmpty { _ = try query("USE `\(database.replacingOccurrences(of: "`", with: "``"))`") }
|
|
}
|
|
private var hostIsLoopback = false
|
|
func markHost(_ host: String) { hostIsLoopback = ["localhost", "127.0.0.1", "::1"].contains(host.lowercased()) }
|
|
|
|
// Streams result rows without the query editor's 100,000-row limit.
|
|
// The callback is invoked on the session's serial background queue.
|
|
func streamFirstColumn(_ sql: String, onRow: (String) throws -> Void) throws -> Int64 {
|
|
sequence = 0
|
|
try send([0x03] + Array(sql.utf8))
|
|
let first = try readPacket()
|
|
try checkError(first)
|
|
guard first.first != 0x00 else { return 0 }
|
|
var cursor = ByteCursor(bytes: first)
|
|
guard let columnCount = try cursor.lenInt(), columnCount == 1 else {
|
|
throw MySQLWireError(message: "Export query must return exactly one column")
|
|
}
|
|
_ = try readPacket() // Column metadata
|
|
_ = try readPacket() // Column EOF
|
|
var count: Int64 = 0
|
|
while true {
|
|
let packet = try readPacket()
|
|
try checkError(packet)
|
|
if packet.first == 0xfe && packet.count < 9 { break }
|
|
var row = ByteCursor(bytes: packet)
|
|
guard let value = try row.lenString() else {
|
|
throw MySQLWireError(message: "Unexpected NULL export row")
|
|
}
|
|
try onRow(value)
|
|
count += 1
|
|
}
|
|
return count
|
|
}
|
|
|
|
func query(_ sql: String) throws -> QueryResult {
|
|
sequence = 0
|
|
try send([0x03] + Array(sql.utf8))
|
|
let first = try readPacket()
|
|
try checkError(first)
|
|
if first.first == 0x00 { return QueryResult(message: "Statement executed successfully") }
|
|
var c = ByteCursor(bytes: first)
|
|
guard let columnCount = try c.lenInt(), columnCount <= 4096 else { throw MySQLWireError(message: "Invalid column count") }
|
|
var names: [String] = []
|
|
for _ in 0..<columnCount {
|
|
var field = ByteCursor(bytes: try readPacket())
|
|
for _ in 0..<4 { _ = try field.lenString() }
|
|
names.append(try field.lenString() ?? "")
|
|
}
|
|
_ = try readPacket() // column EOF (initial protocol requires EOF)
|
|
var rows: [[String]] = []
|
|
while true {
|
|
let packet = try readPacket()
|
|
try checkError(packet)
|
|
if packet.first == 0xfe && packet.count < 9 { break }
|
|
var row = ByteCursor(bytes: packet)
|
|
var values: [String] = []
|
|
for _ in 0..<columnCount { values.append(try row.lenString() ?? "NULL") }
|
|
rows.append(values)
|
|
if rows.count > 100_000 { throw MySQLWireError(message: "Result exceeds 100,000 row safety limit") }
|
|
}
|
|
return QueryResult(columns: names, rows: rows, message: "\(rows.count) row(s)")
|
|
}
|
|
}
|
|
|
|
|
|
// A session owns one authenticated TCP connection. Access is serialized by the
|
|
// queue below because MySQLWireConnection is not safe for concurrent queries.
|
|
private final class PersistentMySQLSessions {
|
|
static let shared = PersistentMySQLSessions()
|
|
|
|
private struct Entry {
|
|
let profile: ConnectionProfile
|
|
let password: String
|
|
let connection: MySQLWireConnection
|
|
}
|
|
|
|
// Each profile has its own serial executor. An unreachable server cannot
|
|
// block requests to any other saved connection.
|
|
private final class SessionSlot {
|
|
let queue: DispatchQueue
|
|
var entry: Entry?
|
|
// Only accessed on queue except generation, which is lock protected.
|
|
private let generationLock = NSLock()
|
|
private var currentGeneration: UInt64 = 0
|
|
|
|
init(id: UUID) {
|
|
queue = DispatchQueue(label: "DBSyn.MySQLSession.\(id.uuidString)")
|
|
}
|
|
func generation() -> UInt64 {
|
|
generationLock.lock(); defer { generationLock.unlock() }
|
|
return currentGeneration
|
|
}
|
|
func invalidate() {
|
|
generationLock.lock(); currentGeneration &+= 1; generationLock.unlock()
|
|
}
|
|
}
|
|
|
|
private let slotsLock = NSLock()
|
|
private var slots: [UUID: SessionSlot] = [:]
|
|
|
|
private func slot(for id: UUID) -> SessionSlot {
|
|
slotsLock.lock(); defer { slotsLock.unlock() }
|
|
if let existing = slots[id] { return existing }
|
|
let created = SessionSlot(id: id)
|
|
slots[id] = created
|
|
return created
|
|
}
|
|
|
|
func execute(profile: ConnectionProfile, database: String,
|
|
password: String, sql: String) throws -> QueryResult {
|
|
let session = slot(for: profile.id)
|
|
let expectedGeneration = session.generation()
|
|
return try session.queue.sync {
|
|
guard session.generation() == expectedGeneration else {
|
|
throw MySQLWireError(message: "Connection request superseded")
|
|
}
|
|
if let previous = session.entry,
|
|
(previous.profile != profile || previous.password != password) {
|
|
session.entry = nil
|
|
DBSynSSHTunnels.shared.stop(profile.id)
|
|
}
|
|
if session.entry == nil {
|
|
let endpoint = try DBSynSSHTunnels.shared.endpoint(for: profile)
|
|
let connection = try MySQLWireConnection(
|
|
host: endpoint.0, port: endpoint.1,
|
|
requireTLS: !profile.useSSHTunnel &&
|
|
!["127.0.0.1", "localhost", "::1"].contains(endpoint.0.lowercased()),
|
|
verifyTLS: profile.verifyTLS
|
|
)
|
|
connection.markHost(endpoint.0)
|
|
try connection.authenticate(user: profile.username,
|
|
password: password, database: database)
|
|
guard session.generation() == expectedGeneration else {
|
|
throw MySQLWireError(message: "Connection request superseded")
|
|
}
|
|
session.entry = Entry(profile: profile, password: password,
|
|
connection: connection)
|
|
}
|
|
guard let entry = session.entry else {
|
|
throw MySQLWireError(message: "MySQL session was not created")
|
|
}
|
|
do {
|
|
let result = try entry.connection.query(sql)
|
|
guard session.generation() == expectedGeneration else {
|
|
throw MySQLWireError(message: "Connection request superseded")
|
|
}
|
|
return result
|
|
} catch {
|
|
session.entry = nil
|
|
throw error
|
|
}
|
|
}
|
|
}
|
|
|
|
func stream(profile: ConnectionProfile, database: String,
|
|
password: String, sql: String,
|
|
onRow: (String) throws -> Void) throws -> Int64 {
|
|
let session = slot(for: profile.id)
|
|
let expectedGeneration = session.generation()
|
|
return try session.queue.sync {
|
|
guard session.generation() == expectedGeneration,
|
|
let entry = session.entry,
|
|
entry.profile == profile, entry.password == password else {
|
|
throw MySQLWireError(message: "Export session is no longer connected")
|
|
}
|
|
do {
|
|
return try entry.connection.streamFirstColumn(sql, onRow: { row in
|
|
guard session.generation() == expectedGeneration else {
|
|
throw MySQLWireError(message: "Export cancelled: connection switched")
|
|
}
|
|
try onRow(row)
|
|
})
|
|
} catch {
|
|
session.entry = nil
|
|
throw error
|
|
}
|
|
}
|
|
}
|
|
|
|
func disconnect(profileID: UUID) {
|
|
let session = slot(for: profileID)
|
|
// Invalidate immediately without waiting for a blocked network read.
|
|
session.invalidate()
|
|
session.queue.async {
|
|
session.entry = nil
|
|
DBSynSSHTunnels.shared.stop(profileID)
|
|
}
|
|
}
|
|
}
|
|
|
|
enum NativeMySQLRunner {
|
|
static func run(profile: ConnectionProfile, database: String,
|
|
password: String, sql: String) async throws -> QueryResult {
|
|
try await Task.detached(priority: .userInitiated) {
|
|
try PersistentMySQLSessions.shared.execute(
|
|
profile: profile, database: database, password: password, sql: sql
|
|
)
|
|
}.value
|
|
}
|
|
|
|
static func stream(profile: ConnectionProfile, database: String,
|
|
password: String, sql: String,
|
|
onRow: @escaping (String) throws -> Void) async throws -> Int64 {
|
|
try await Task.detached(priority: .userInitiated) {
|
|
try PersistentMySQLSessions.shared.stream(
|
|
profile: profile, database: database, password: password,
|
|
sql: sql, onRow: { row in
|
|
try Task<Never, Never>.checkCancellation()
|
|
try onRow(row)
|
|
}
|
|
)
|
|
}.value
|
|
}
|
|
|
|
// Connection shutdown may wait behind a slow network operation on the
|
|
// session's serial queue. Never make the SwiftUI main thread wait for it.
|
|
static func disconnect(profileID: UUID) {
|
|
PersistentMySQLSessions.shared.disconnect(profileID: profileID)
|
|
}
|
|
}
|