From 8759a6e4e1b9ea48865d2ce12df9243c308355a7 Mon Sep 17 00:00:00 2001 From: Ngo Quoc Dat Date: Sun, 20 Sep 2026 12:58:18 +0700 Subject: [PATCH] fix(plugin-etcd): classify etcd faults by their gRPC code, not the HTTP status --- CHANGELOG.md | 8 + .../EtcdDriverPlugin/EtcdCommandParser.swift | 12 +- .../EtcdDriverPlugin/EtcdGatewayRoute.swift | 34 + Plugins/EtcdDriverPlugin/EtcdHttpClient.swift | 597 +++++++++++------- .../EtcdDriverPlugin/EtcdPluginDriver.swift | 20 +- .../EtcdRequestRecovery.swift | 35 + .../EtcdDriverPlugin/EtcdServerFault.swift | 98 +++ .../Plugins/EtcdCommandParserTests.swift | 48 ++ .../Plugins/EtcdGatewayRouteTests.swift | 54 ++ .../Plugins/EtcdRequestRecoveryTests.swift | 104 +++ .../Plugins/EtcdServerFaultTests.swift | 152 +++++ docs/databases/etcd.mdx | 30 +- project.yml | 3 + scripts/check-etcd-auth-faults.sh | 138 ++++ 14 files changed, 1075 insertions(+), 258 deletions(-) create mode 100644 Plugins/EtcdDriverPlugin/EtcdGatewayRoute.swift create mode 100644 Plugins/EtcdDriverPlugin/EtcdRequestRecovery.swift create mode 100644 Plugins/EtcdDriverPlugin/EtcdServerFault.swift create mode 100644 TableProTests/Plugins/EtcdGatewayRouteTests.swift create mode 100644 TableProTests/Plugins/EtcdRequestRecoveryTests.swift create mode 100644 TableProTests/Plugins/EtcdServerFaultTests.swift create mode 100755 scripts/check-etcd-auth-faults.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index 55eb638659..6ebe82a0a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- etcd connections failing with `Unexpected HTTP 400 from v3/maintenance/status` once authentication is enabled. (#2994) +- etcd connections carrying a username refusing a server that has authentication disabled. +- Raw JSON shown instead of etcd's own message when a username or password is wrong. +- etcd connections reported as unreachable every 30 seconds for a user without the root role. +- An etcd `watch` returning no events instead of an error when the session is not authenticated. +- `The request timed out` from an etcd `watch --timeout` above 60 seconds. +- Stop in an etcd tab cancelling an unrelated request and leaving the running one alone. +- Crash from an etcd `watch --timeout` with a negative or out-of-range value. - Oracle PL/SQL blocks split at their inner semicolons and sent as fragments, failing with PLS-00103. (#2984) - Oracle procedures, packages and triggers created from the editor stored INVALID while the run reported success. - SQL*Plus `/` lines, `q'[…]'` literals and backslashes in strings misread in Oracle scripts. diff --git a/Plugins/EtcdDriverPlugin/EtcdCommandParser.swift b/Plugins/EtcdDriverPlugin/EtcdCommandParser.swift index b73c6d5bab..78e114e0cc 100644 --- a/Plugins/EtcdDriverPlugin/EtcdCommandParser.swift +++ b/Plugins/EtcdDriverPlugin/EtcdCommandParser.swift @@ -82,6 +82,9 @@ extension EtcdParseError: PluginDriverError { struct EtcdCommandParser { private static let logger = Logger(subsystem: "com.TablePro", category: "EtcdCommandParser") + static let defaultWatchTimeout: TimeInterval = 30 + static let maximumWatchTimeout: TimeInterval = 3_000 + // MARK: - Public API static func parse(_ input: String) throws -> EtcdOperation { @@ -198,11 +201,16 @@ struct EtcdCommandParser { let prefix = flags.has("prefix") - var timeout: TimeInterval = 30 + var timeout = Self.defaultWatchTimeout if let timeoutStr = flags.value(for: "timeout") { - guard let parsed = TimeInterval(timeoutStr) else { + guard let parsed = TimeInterval(timeoutStr), parsed.isFinite else { throw EtcdParseError.invalidArgument("--timeout must be a number") } + guard parsed >= 0, parsed <= Self.maximumWatchTimeout else { + throw EtcdParseError.invalidArgument( + "--timeout must be between 0 and \(Int(Self.maximumWatchTimeout)) seconds" + ) + } timeout = parsed } diff --git a/Plugins/EtcdDriverPlugin/EtcdGatewayRoute.swift b/Plugins/EtcdDriverPlugin/EtcdGatewayRoute.swift new file mode 100644 index 0000000000..cef1900794 --- /dev/null +++ b/Plugins/EtcdDriverPlugin/EtcdGatewayRoute.swift @@ -0,0 +1,34 @@ +// +// EtcdGatewayRoute.swift +// EtcdDriverPlugin +// +// Decides whether an etcd v3 gateway prefix is routed, from the answer alone. +// + +import Foundation + +internal enum EtcdGatewayRoute: Equatable, Sendable { + case routed + case notRouted + case notEtcd +} + +internal extension EtcdGatewayRoute { + static let candidatePrefixes = ["v3", "v3beta", "v3alpha"] + + static func classify(httpStatus: Int, body: Data) -> EtcdGatewayRoute { + guard httpStatus != notFoundStatus else { return .notRouted } + guard isJsonObject(body) else { return .notEtcd } + return .routed + } +} + +private extension EtcdGatewayRoute { + static let notFoundStatus = 404 + + static func isJsonObject(_ body: Data) -> Bool { + guard !body.isEmpty else { return false } + guard let object = try? JSONSerialization.jsonObject(with: body) else { return false } + return object is [String: Any] + } +} diff --git a/Plugins/EtcdDriverPlugin/EtcdHttpClient.swift b/Plugins/EtcdDriverPlugin/EtcdHttpClient.swift index df0df9853d..0883fd67b8 100644 --- a/Plugins/EtcdDriverPlugin/EtcdHttpClient.swift +++ b/Plugins/EtcdDriverPlugin/EtcdHttpClient.swift @@ -15,6 +15,7 @@ internal enum EtcdError: Error, LocalizedError { case connectionFailed(String) case serverError(String) case authFailed(String) + case fault(EtcdServerFault) case requestCancelled var errorDescription: String? { @@ -27,10 +28,17 @@ internal enum EtcdError: Error, LocalizedError { return String(format: String(localized: "Server error: %@"), detail) case .authFailed(let detail): return String(format: String(localized: "Authentication failed: %@"), detail) + case .fault(let fault): + return fault.localizedDescription case .requestCancelled: return String(localized: "Request was cancelled") } } + + var serverFault: EtcdServerFault? { + guard case .fault(let fault) = self else { return nil } + return fault + } } // MARK: - Codable Types @@ -298,10 +306,8 @@ internal struct EtcdCompactionRequest: Encodable { // MARK: - Generic Error Response -private struct EtcdErrorResponse: Decodable { - let error: String? - let message: String? - let code: Int? +internal struct EtcdVersionResponse: Decodable { + let etcdserver: String? } // MARK: - HTTP Client @@ -311,9 +317,9 @@ internal final class EtcdHttpClient: @unchecked Sendable { private let lock = NSLock() private var session: URLSession? private var sessionGeneration: UInt64 = 0 - private var currentTask: URLSessionDataTask? + private var activeQueryTasks: [ObjectIdentifier: URLSessionDataTask] = [:] private var authToken: String? - private var _isAuthenticating = false + private var authTask: Task? private var apiPrefix = "v3" private let queryTimeout = HttpQueryTimeoutBox() @@ -381,114 +387,126 @@ internal final class EtcdHttpClient: @unchecked Sendable { do { try await detectApiPrefix() - } catch let etcdError as EtcdError { - lock.withLock { - session?.invalidateAndCancel() - session = nil + if hasCredentials { + try await refreshToken(replacing: nil) } - Self.logger.error("Connection test failed: \(etcdError.localizedDescription)") + try await healthCheck() + } catch let etcdError as EtcdError { + invalidateSession() + Self.logger.error("Connection failed: \(etcdError.localizedDescription)") throw etcdError } catch { - lock.withLock { - session?.invalidateAndCancel() - session = nil - } - Self.logger.error("Connection test failed: \(error.localizedDescription)") + invalidateSession() + Self.logger.error("Connection failed: \(error.localizedDescription)") throw EtcdError.connectionFailed(error.localizedDescription) } - if !config.username.isEmpty { - do { - try await authenticate() - } catch { - lock.withLock { - session?.invalidateAndCancel() - session = nil - } - throw error - } - } - Self.logger.debug("Connected to etcd at \(self.config.host):\(self.config.port)") } + private var hasCredentials: Bool { + !config.username.isEmpty + } + + private func invalidateSession() { + lock.withLock { + authTask?.cancel() + authTask = nil + authToken = nil + session?.invalidateAndCancel() + session = nil + } + } + func disconnect() { + let pending = takeActiveQueryTasks() lock.lock() sessionGeneration &+= 1 - currentTask?.cancel() - currentTask = nil + authTask?.cancel() + authTask = nil session?.invalidateAndCancel() session = nil authToken = nil - _isAuthenticating = false apiPrefix = "v3" lock.unlock() + for task in pending { + task.cancel() + } } func ping() async throws { - let _: EtcdStatusResponse = try await post(path: apiPath("maintenance/status"), body: EmptyBody()) + try await healthCheck() } - /// Probes etcd gateway prefixes in order and selects the first that responds - /// with a non-404 status. Covers all etcd versions: - /// 3.5+ → /v3/ only - /// 3.4 → /v3/ + /v3beta/ - /// 3.3 → /v3beta/ + /v3alpha/ - /// 3.2- → /v3alpha/ only - private func detectApiPrefix() async throws { - let candidates = ["v3", "v3beta", "v3alpha"] - - let session = try lock.withLock { () -> URLSession in - guard let currentSession = self.session else { throw EtcdError.notConnected } - return currentSession + func healthCheck() async throws { + do { + let request = EtcdRangeRequest( + key: Self.base64Encode(Self.healthProbeKey), + limit: 1, + keysOnly: true + ) + _ = try await send( + path: apiPath("kv/range"), + body: request, + cancellable: false + ) + } catch let EtcdError.fault(fault) where fault.provesLiveSession { + return } + } - for candidate in candidates { - guard let url = URL(string: "\(baseUrl)/\(candidate)/maintenance/status") else { - continue - } - - var request = URLRequest(url: url) - request.httpMethod = "POST" - request.setValue("application/json", forHTTPHeaderField: "Content-Type") - request.httpBody = try JSONEncoder().encode(EmptyBody()) - - let response: URLResponse - do { - (_, response) = try await session.data(for: request) - } catch { - // Network-level failure — server is unreachable regardless of prefix - throw error - } - - guard let httpResponse = response as? HTTPURLResponse else { - throw EtcdError.serverError("Invalid response type") - } + private func detectApiPrefix() async throws { + var sawForeignResponse = false - switch httpResponse.statusCode { - case 404: - continue - case 200: + for candidate in EtcdGatewayRoute.candidatePrefixes { + switch try await probeGatewayRoute(prefix: candidate) { + case .routed: lock.withLock { apiPrefix = candidate } Self.logger.debug("Detected etcd API prefix: \(candidate)") return - case 401 where !config.username.isEmpty: - // Auth required but credentials are configured — prefix is valid, - // authenticate() will run after detection - lock.withLock { apiPrefix = candidate } - Self.logger.debug("Detected etcd API prefix: \(candidate) (auth required)") - return - case 401: - throw EtcdError.authFailed("Authentication required") - default: - Self.logger.warning("Prefix probe \(candidate) returned HTTP \(httpResponse.statusCode)") - throw EtcdError.serverError("Unexpected HTTP \(httpResponse.statusCode) from \(candidate)/maintenance/status") + case .notRouted: + continue + case .notEtcd: + sawForeignResponse = true } } - throw EtcdError.serverError( - "No supported etcd API found (tried: \(candidates.joined(separator: ", ")))" + guard !sawForeignResponse else { + throw EtcdError.connectionFailed(String(localized: """ + The server answered but is not an etcd v3 JSON gateway. + """)) + } + throw EtcdError.connectionFailed(String(format: String(localized: """ + No etcd v3 API found at %@. Point the connection at the client port, 2379 by default, \ + and not the peer port on 2380. + """), "\(config.host):\(config.port)")) + } + + private func probeGatewayRoute(prefix: String) async throws -> EtcdGatewayRoute { + let session = try lock.withLock { () -> URLSession in + guard let currentSession = self.session else { throw EtcdError.notConnected } + return currentSession + } + + let probe = EtcdRangeRequest( + key: Self.base64Encode(Self.healthProbeKey), + limit: 1, + keysOnly: true ) + guard let url = URL(string: "\(baseUrl)/\(prefix)/kv/range") else { + return .notRouted + } + + var request = URLRequest(url: url) + request.httpMethod = "POST" + request.setValue("application/json", forHTTPHeaderField: "Content-Type") + request.httpBody = try JSONEncoder().encode(probe) + + let (data, response) = try await session.data(for: request) + guard let httpResponse = response as? HTTPURLResponse else { + throw EtcdError.serverError(String(localized: "Invalid response type")) + } + return EtcdGatewayRoute.classify(httpStatus: httpResponse.statusCode, body: data) } // MARK: - KV Operations @@ -536,81 +554,126 @@ internal final class EtcdHttpClient: @unchecked Sendable { try await post(path: apiPath("maintenance/status"), body: EmptyBody()) } + func serverVersion() async -> String? { + if let status = try? await endpointStatus(), let version = status.version { + return version + } + return await gatewayVersion() + } + + private func gatewayVersion() async -> String? { + guard let session = lock.withLock({ self.session }) else { return nil } + guard let url = URL(string: "\(baseUrl)/version") else { return nil } + + var request = URLRequest(url: url) + request.timeoutInterval = HttpQueryTimeout.sessionBootstrapRequestTimeout + guard let (data, response) = try? await session.data(for: request), + let httpResponse = response as? HTTPURLResponse, + httpResponse.statusCode == 200, + let payload = try? JSONDecoder().decode(EtcdVersionResponse.self, from: data) else { + return nil + } + return payload.etcdserver + } + // MARK: - Watch func watch(key: String, prefix: Bool, timeout: TimeInterval) async throws -> [EtcdWatchEvent] { - let (token, generation) = try lock.withLock { () -> (String?, UInt64) in + try await watch(key: key, prefix: prefix, timeout: timeout, isRetry: false) + } + + private func watch( + key: String, + prefix: Bool, + timeout: TimeInterval, + isRetry: Bool + ) async throws -> [EtcdWatchEvent] { + let token = try lock.withLock { () -> String? in guard session != nil else { throw EtcdError.notConnected } - return (authToken, sessionGeneration) + return authToken } - let b64Key = Self.base64Encode(key) - var createReq = EtcdWatchCreateRequest(key: b64Key) + var createRequest = EtcdWatchCreateRequest(key: Self.base64Encode(key)) if prefix { - createReq.rangeEnd = Self.base64Encode(Self.prefixRangeEnd(for: key)) + createRequest.rangeEnd = Self.base64Encode(Self.prefixRangeEnd(for: key)) } - let watchReq = EtcdWatchRequest(createRequest: createReq) + let window = Self.watchWindow(timeout) + let watchRequest = try buildRequest( + path: apiPath("watch"), + body: EtcdWatchRequest(createRequest: createRequest), + token: token, + timeout: window + Self.watchTransportGrace + ) - let watchPath = apiPath("watch") - guard let url = URL(string: "\(baseUrl)/\(watchPath)") else { - throw EtcdError.serverError("Invalid URL: \(baseUrl)/\(watchPath)") - } + let outcome = try await streamWatch(request: watchRequest, timeout: window) - var request = URLRequest(url: url) - request.httpMethod = "POST" - request.setValue("application/json", forHTTPHeaderField: "Content-Type") - if let token { - request.setValue(token, forHTTPHeaderField: "Authorization") + if let status = outcome.httpStatus, status >= 400 { + let fault = EtcdServerFault.decode(httpStatus: status, body: outcome.data) + let recovery = EtcdRequestRecovery.action( + for: fault, + hasCredentials: hasCredentials, + isRetry: isRetry + ) + guard recovery == .reauthenticateAndRetry else { throw EtcdError.fault(fault) } + try await refreshToken(replacing: token) + return try await watch(key: key, prefix: prefix, timeout: timeout, isRetry: true) } - request.httpBody = try JSONEncoder().encode(watchReq) - let watchRequest = request - return try await withThrowingTaskGroup(of: [EtcdWatchEvent].self) { group in - let collectedData = DataCollector() + return Self.parseWatchEvents(from: outcome.data) + } + + private func streamWatch( + request: URLRequest, + timeout: TimeInterval + ) async throws -> (data: Data, httpStatus: Int?) { + try await withThrowingTaskGroup(of: (data: Data, httpStatus: Int?)?.self) { group in + let handle = TaskHandle() + let generation = try lock.withLock { () -> UInt64 in + guard session != nil else { throw EtcdError.notConnected } + return sessionGeneration + } group.addTask { - let data: Data = try await withCheckedThrowingContinuation { continuation in - let result: (session: URLSession, task: URLSessionDataTask)? = self.lock.withLock { + try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<(data: Data, httpStatus: Int?)?, Error>) in + let task: URLSessionDataTask? = self.lock.withLock { guard self.sessionGeneration == generation, let currentSession = self.session else { return nil } - let dataTask = currentSession.dataTask(with: watchRequest) { data, _, error in + return currentSession.dataTask(with: request) { data, response, error in + self.releaseActiveQueryTask(handle.task) + let status = (response as? HTTPURLResponse)?.statusCode if let error { - // URLError.cancelled is expected when we cancel after timeout if (error as? URLError)?.code == .cancelled { - continuation.resume(returning: data ?? Data()) + continuation.resume(returning: (data ?? Data(), status)) } else { continuation.resume(throwing: error) } return } - continuation.resume(returning: data ?? Data()) + continuation.resume(returning: (data ?? Data(), status)) } - self.currentTask = dataTask - return (currentSession, dataTask) } - guard let result else { + guard let task else { continuation.resume(throwing: EtcdError.notConnected) return } - collectedData.setTask(result.task) - result.task.resume() + self.registerActiveQueryTask(task) + handle.adopt(task) + task.resume() } - return Self.parseWatchEvents(from: data) } group.addTask { - try await Task.sleep(nanoseconds: UInt64(timeout * 1_000_000_000)) - collectedData.cancelTask() - return [] + try await Task.sleep(nanoseconds: Self.nanoseconds(from: timeout)) + handle.cancel() + return nil } - var allEvents: [EtcdWatchEvent] = [] - for try await events in group { - allEvents.append(contentsOf: events) + defer { group.cancelAll() } + for try await outcome in group where outcome != nil { + return outcome ?? (Data(), nil) } - group.cancelAll() - return allEvents + return (Data(), nil) } } @@ -674,50 +737,110 @@ internal final class EtcdHttpClient: @unchecked Sendable { // MARK: - Cancellation func cancelCurrentRequest() { - lock.lock() - currentTask?.cancel() - currentTask = nil - lock.unlock() + for task in takeActiveQueryTasks() { + task.cancel() + } } // MARK: - Internal Transport private func post(path: String, body: Req) async throws -> Res { - let data = try await performRequest(path: path, body: body) - do { - let decoder = JSONDecoder() - return try decoder.decode(Res.self, from: data) - } catch { - let bodyStr = String(data: data, encoding: .utf8) ?? "" - Self.logger.error("Failed to decode response for \(path): \(bodyStr)") - throw EtcdError.serverError("Failed to decode response: \(error.localizedDescription)") - } + let data = try await send(path: path, body: body) + return try decode(data, from: path) } private func postVoid(path: String, body: Req) async throws { - _ = try await performRequest(path: path, body: body) + _ = try await send(path: path, body: body) + } + + private func decode(_ data: Data, from path: String) throws -> Res { + do { + return try JSONDecoder().decode(Res.self, from: data) + } catch { + let bodyText = String(data: data, encoding: .utf8) ?? "" + Self.logger.error("Failed to decode response for \(path): \(bodyText)") + throw EtcdError.serverError( + String(format: String(localized: "Failed to decode response: %@"), error.localizedDescription) + ) + } } - private func performRequest(path: String, body: Req, allowReauth: Bool = true) async throws -> Data { - let (token, generation) = try lock.withLock { () -> (String?, UInt64) in + private func send( + path: String, + body: Req, + authorized: Bool = true, + cancellable: Bool = true, + isRetry: Bool = false + ) async throws -> Data { + let token = try lock.withLock { () -> String? in guard session != nil else { throw EtcdError.notConnected } - return (authToken, sessionGeneration) + return authToken } - guard let url = URL(string: "\(baseUrl)/\(path)") else { - throw EtcdError.serverError("Invalid URL: \(baseUrl)/\(path)") + let request = try buildRequest( + path: path, + body: body, + token: authorized ? token : nil, + timeout: queryTimeout.requestTimeoutInterval + ) + let (data, response) = try await perform(request: request, cancellable: cancellable) + + guard let httpResponse = response as? HTTPURLResponse else { + throw EtcdError.serverError(String(localized: "Invalid response type")) } + guard httpResponse.statusCode >= 400 else { return data } + + let fault = EtcdServerFault.decode(httpStatus: httpResponse.statusCode, body: data) + let recovery = EtcdRequestRecovery.action( + for: fault, + hasCredentials: authorized && hasCredentials, + isRetry: isRetry + ) + guard recovery == .reauthenticateAndRetry else { throw EtcdError.fault(fault) } + + try await refreshToken(replacing: token) + return try await send( + path: path, + body: body, + authorized: authorized, + cancellable: cancellable, + isRetry: true + ) + } + private func buildRequest( + path: String, + body: Req, + token: String?, + timeout: TimeInterval + ) throws -> URLRequest { + guard let url = URL(string: "\(baseUrl)/\(path)") else { + throw EtcdError.serverError( + String(format: String(localized: "Invalid URL: %@"), "\(baseUrl)/\(path)") + ) + } var request = URLRequest(url: url) request.httpMethod = "POST" - request.timeoutInterval = queryTimeout.requestTimeoutInterval + request.timeoutInterval = timeout request.setValue("application/json", forHTTPHeaderField: "Content-Type") if let token { request.setValue(token, forHTTPHeaderField: "Authorization") } request.httpBody = try JSONEncoder().encode(body) + return request + } + + private func perform( + request: URLRequest, + cancellable: Bool + ) async throws -> (Data, URLResponse) { + let generation = try lock.withLock { () -> UInt64 in + guard session != nil else { throw EtcdError.notConnected } + return sessionGeneration + } + let handle = TaskHandle() - let (data, response) = try await withTaskCancellationHandler { + return try await withTaskCancellationHandler { try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<(Data, URLResponse), Error>) in self.lock.lock() guard self.sessionGeneration == generation, let currentSession = self.session else { @@ -726,128 +849,97 @@ internal final class EtcdHttpClient: @unchecked Sendable { return } let task = currentSession.dataTask(with: request) { data, response, error in + if cancellable { + self.releaseActiveQueryTask(handle.task) + } if let error { continuation.resume(throwing: error) return } guard let data, let response else { - continuation.resume(throwing: EtcdError.serverError("Empty response from server")) + continuation.resume( + throwing: EtcdError.serverError(String(localized: "Empty response from server")) + ) return } continuation.resume(returning: (data, response)) } - self.currentTask = task + if cancellable { + self.activeQueryTasks[ObjectIdentifier(task)] = task + } self.lock.unlock() + handle.adopt(task) task.resume() } } onCancel: { - self.lock.lock() - self.currentTask?.cancel() - self.currentTask = nil - self.lock.unlock() - } - - lock.withLock { currentTask = nil } - - guard let httpResponse = response as? HTTPURLResponse else { - throw EtcdError.serverError("Invalid response type") + handle.cancel() } + } - if httpResponse.statusCode == 401 { - // Attempt token refresh if not already authenticating and credentials are available - let alreadyAuthenticating = lock.withLock { _isAuthenticating } + private func registerActiveQueryTask(_ task: URLSessionDataTask) { + lock.withLock { activeQueryTasks[ObjectIdentifier(task)] = task } + } - if allowReauth, !alreadyAuthenticating, !config.username.isEmpty { - try await authenticate() - return try await performRequest(path: path, body: body, allowReauth: false) - } - let errorBody = String(data: data, encoding: .utf8) ?? "Unauthorized" - throw EtcdError.authFailed(errorBody) - } + private func releaseActiveQueryTask(_ task: URLSessionDataTask?) { + guard let task else { return } + lock.withLock { activeQueryTasks.removeValue(forKey: ObjectIdentifier(task)) } + } - if httpResponse.statusCode >= 400 { - let errorBody = String(data: data, encoding: .utf8) ?? "Unknown error" - if let errorResp = try? JSONDecoder().decode(EtcdErrorResponse.self, from: data), - let message = errorResp.error ?? errorResp.message { - throw EtcdError.serverError(message) - } - throw EtcdError.serverError(errorBody.trimmingCharacters(in: .whitespacesAndNewlines)) + private func takeActiveQueryTasks() -> [URLSessionDataTask] { + lock.withLock { () -> [URLSessionDataTask] in + let pending = Array(activeQueryTasks.values) + activeQueryTasks.removeAll() + return pending } - - return data } // MARK: - Authentication - private func authenticate() async throws { - let startedAuthenticating = try lock.withLock { () -> Bool in - guard session != nil else { throw EtcdError.notConnected } - guard !_isAuthenticating else { return false } - _isAuthenticating = true - return true - } - guard startedAuthenticating else { return } - - defer { - lock.withLock { _isAuthenticating = false } + private func refreshToken(replacing staleToken: String?) async throws { + enum Pending { + case alreadyRefreshed + case task(Task) } - let authReq = EtcdAuthRequest(name: config.username, password: config.password) - let authPath = apiPath("auth/authenticate") - guard let url = URL(string: "\(baseUrl)/\(authPath)") else { - throw EtcdError.serverError("Invalid auth URL") - } - - var request = URLRequest(url: url) - request.httpMethod = "POST" - request.setValue("application/json", forHTTPHeaderField: "Content-Type") - request.httpBody = try JSONEncoder().encode(authReq) - - let generation = try lock.withLock { () -> UInt64 in + let pending: Pending = try lock.withLock { () -> Pending in guard session != nil else { throw EtcdError.notConnected } - return sessionGeneration - } - - let (data, response) = try await withCheckedThrowingContinuation { - (continuation: CheckedContinuation<(Data, URLResponse), Error>) in - self.lock.lock() - guard self.sessionGeneration == generation, let currentSession = self.session else { - self.lock.unlock() - continuation.resume(throwing: EtcdError.notConnected) - return - } - let task = currentSession.dataTask(with: request) { data, response, error in - if let error { - continuation.resume(throwing: error) - return - } - guard let data, let response else { - continuation.resume(throwing: EtcdError.authFailed("Empty response")) - return - } - continuation.resume(returning: (data, response)) + if let authToken, authToken != staleToken { return .alreadyRefreshed } + if let authTask { return .task(authTask) } + let task = Task { + defer { self.lock.withLock { self.authTask = nil } } + try await self.authenticate() } - self.lock.unlock() - task.resume() + authTask = task + return .task(task) } - guard let httpResponse = response as? HTTPURLResponse else { - throw EtcdError.authFailed("Invalid response type") - } + guard case .task(let task) = pending else { return } + try await task.value + } - if httpResponse.statusCode >= 400 { - let errorBody = String(data: data, encoding: .utf8) ?? "Authentication failed" - throw EtcdError.authFailed(errorBody) + private func authenticate() async throws { + let credentials = EtcdAuthRequest(name: config.username, password: config.password) + let data: Data + do { + data = try await send( + path: apiPath("auth/authenticate"), + body: credentials, + authorized: false, + cancellable: false + ) + } catch let EtcdError.fault(fault) where fault.kind == .authNotEnabled { + lock.withLock { authToken = nil } + Self.logger.info("etcd reports authentication is not enabled; continuing without a token") + return } - let authResp = try JSONDecoder().decode(EtcdAuthResponse.self, from: data) - guard let token = authResp.token, !token.isEmpty else { - throw EtcdError.authFailed("No token in response") + let response: EtcdAuthResponse = try decode(data, from: "auth/authenticate") + guard let token = response.token, !token.isEmpty else { + throw EtcdError.authFailed(String(localized: "No token in response")) } lock.withLock { authToken = token } - Self.logger.debug("Authenticated with etcd successfully") } @@ -908,23 +1000,46 @@ internal final class EtcdHttpClient: @unchecked Sendable { private struct EmptyBody: Encodable {} - // MARK: - Data Collector for Watch + // MARK: - Request Handle - private final class DataCollector: @unchecked Sendable { + private static let healthProbeKey = "health" + private static let watchTransportGrace = TimeInterval(HttpQueryTimeout.defaultGraceSeconds) + private static let maximumWatchWindow = + TimeInterval(HttpQueryTimeout.resourceCeilingSeconds) - watchTransportGrace + + private static func watchWindow(_ seconds: TimeInterval) -> TimeInterval { + guard seconds.isFinite else { return maximumWatchWindow } + return min(max(seconds, 0), maximumWatchWindow) + } + + private static func nanoseconds(from seconds: TimeInterval) -> UInt64 { + UInt64(watchWindow(seconds) * 1_000_000_000) + } + + private final class TaskHandle: @unchecked Sendable { private let lock = NSLock() - private var _task: URLSessionDataTask? + private var storedTask: URLSessionDataTask? + private var cancelRequested = false - func setTask(_ task: URLSessionDataTask) { - lock.lock() - _task = task - lock.unlock() + var task: URLSessionDataTask? { + lock.withLock { storedTask } } - func cancelTask() { - lock.lock() - let task = _task - lock.unlock() - task?.cancel() + func adopt(_ task: URLSessionDataTask) { + let alreadyCancelled = lock.withLock { () -> Bool in + storedTask = task + return cancelRequested + } + guard alreadyCancelled else { return } + task.cancel() + } + + func cancel() { + let pending = lock.withLock { () -> URLSessionDataTask? in + cancelRequested = true + return storedTask + } + pending?.cancel() } } diff --git a/Plugins/EtcdDriverPlugin/EtcdPluginDriver.swift b/Plugins/EtcdDriverPlugin/EtcdPluginDriver.swift index b4d77bc37d..c7293d3f90 100644 --- a/Plugins/EtcdDriverPlugin/EtcdPluginDriver.swift +++ b/Plugins/EtcdDriverPlugin/EtcdPluginDriver.swift @@ -88,9 +88,9 @@ final class EtcdPluginDriver: PluginDatabaseDriver, @unchecked Sendable { let client = EtcdHttpClient(config: config) try await client.connect() - let status = try? await client.endpointStatus() + let version = await client.serverVersion() lock.withLock { - _serverVersion = status?.version + _serverVersion = version _httpClient = client } } @@ -404,12 +404,12 @@ final class EtcdPluginDriver: PluginDatabaseDriver, @unchecked Sendable { throw EtcdError.notConnected } - let status = try await client.endpointStatus() - let dbSizeBytes = Int64(status.dbSize ?? "0") - return PluginDatabaseMetadata( - name: database, - sizeBytes: dbSizeBytes - ) + do { + let status = try await client.endpointStatus() + return PluginDatabaseMetadata(name: database, sizeBytes: Int64(status.dbSize ?? "0")) + } catch let EtcdError.fault(fault) where fault.provesLiveSession { + return PluginDatabaseMetadata(name: database) + } } // MARK: - NoSQL Query Building Hooks @@ -860,8 +860,8 @@ final class EtcdPluginDriver: PluginDatabaseDriver, @unchecked Sendable { private func dispatchEndpointHealth( client: EtcdHttpClient, startTime: Date ) async throws -> PluginQueryResult { - try await client.ping() - return singleMessageResult("endpoint is healthy", startTime: startTime) + try await client.healthCheck() + return singleMessageResult(String(localized: "endpoint is healthy"), startTime: startTime) } // MARK: - Tagged Query Execution diff --git a/Plugins/EtcdDriverPlugin/EtcdRequestRecovery.swift b/Plugins/EtcdDriverPlugin/EtcdRequestRecovery.swift new file mode 100644 index 0000000000..4dba58d249 --- /dev/null +++ b/Plugins/EtcdDriverPlugin/EtcdRequestRecovery.swift @@ -0,0 +1,35 @@ +// +// EtcdRequestRecovery.swift +// EtcdDriverPlugin +// +// Decides what a failed etcd request should do next, the way clientv3 does. +// + +import Foundation + +internal enum EtcdRequestRecovery: Equatable, Sendable { + case reauthenticateAndRetry + case surface +} + +internal extension EtcdRequestRecovery { + static func action( + for fault: EtcdServerFault, + hasCredentials: Bool, + isRetry: Bool + ) -> EtcdRequestRecovery { + guard !isRetry, hasCredentials else { return .surface } + switch fault.kind { + case .credentialsRequired, .tokenRejected, .authRevisionStale: + return .reauthenticateAndRetry + case .credentialsRejected, .authNotEnabled, .permissionDenied, .unclassified: + return .surface + } + } +} + +internal extension EtcdServerFault { + var provesLiveSession: Bool { + kind == .permissionDenied + } +} diff --git a/Plugins/EtcdDriverPlugin/EtcdServerFault.swift b/Plugins/EtcdDriverPlugin/EtcdServerFault.swift new file mode 100644 index 0000000000..d2a57ec3cb --- /dev/null +++ b/Plugins/EtcdDriverPlugin/EtcdServerFault.swift @@ -0,0 +1,98 @@ +// +// EtcdServerFault.swift +// EtcdDriverPlugin +// +// Classifies an etcd v3 gateway failure by the gRPC status code in its body. +// + +import Foundation + +internal struct EtcdServerFault: Equatable, Sendable { + internal let grpcCode: Int? + internal let message: String +} + +internal extension EtcdServerFault { + enum Kind: Equatable, Sendable { + case credentialsRequired + case credentialsRejected + case tokenRejected + case authRevisionStale + case authNotEnabled + case permissionDenied + case unclassified + } + + static func decode(httpStatus: Int, body: Data) -> EtcdServerFault { + if let envelope = try? JSONDecoder().decode(Envelope.self, from: body), + let message = envelope.message ?? envelope.error, + !message.isEmpty { + return EtcdServerFault(grpcCode: envelope.code, message: message) + } + let text = String(data: body, encoding: .utf8)? + .trimmingCharacters(in: .whitespacesAndNewlines) ?? "" + guard !text.isEmpty else { + return EtcdServerFault( + grpcCode: nil, + message: String(format: String(localized: "etcd returned HTTP %d"), httpStatus) + ) + } + return EtcdServerFault(grpcCode: nil, message: String(text.prefix(maximumMessageLength))) + } + + var kind: Kind { + switch grpcCode { + case Self.unauthenticatedCode: + return .tokenRejected + case Self.permissionDeniedCode: + return .permissionDenied + case Self.failedPreconditionCode: + return message.contains(Self.authNotEnabledMarker) ? .authNotEnabled : .unclassified + case Self.invalidArgumentCode: + return invalidArgumentKind + default: + return .unclassified + } + } + + var localizedDescription: String { + switch kind { + case .credentialsRequired: + return String(localized: """ + This etcd server has authentication enabled. Add a username and password to the \ + connection, then connect again. + """) + case .credentialsRejected, .tokenRejected: + return String(format: String(localized: "Authentication failed: %@"), message) + default: + return message + } + } +} + +private extension EtcdServerFault { + struct Envelope: Decodable { + let error: String? + let message: String? + let code: Int? + } + + static let maximumMessageLength = 512 + + static let invalidArgumentCode = 3 + static let permissionDeniedCode = 7 + static let failedPreconditionCode = 9 + static let unauthenticatedCode = 16 + + static let userEmptyMarker = "user name is empty" + static let authFailedMarker = "authentication failed" + static let authRevisionMarker = "revision of auth store is old" + static let authNotEnabledMarker = "authentication is not enabled" + + var invalidArgumentKind: Kind { + if message.contains(Self.userEmptyMarker) { return .credentialsRequired } + if message.contains(Self.authFailedMarker) { return .credentialsRejected } + if message.contains(Self.authRevisionMarker) { return .authRevisionStale } + return .unclassified + } +} diff --git a/TableProTests/Plugins/EtcdCommandParserTests.swift b/TableProTests/Plugins/EtcdCommandParserTests.swift index 70fdb67bab..dbf93bad2a 100644 --- a/TableProTests/Plugins/EtcdCommandParserTests.swift +++ b/TableProTests/Plugins/EtcdCommandParserTests.swift @@ -307,6 +307,54 @@ struct EtcdCommandParserWatchTests { try EtcdCommandParser.parse("watch") } } + + @Test("A negative timeout is rejected instead of reaching the transport") + func watchNegativeTimeout() { + #expect(throws: EtcdParseError.self) { + try EtcdCommandParser.parse("watch key --timeout -1") + } + } + + @Test("A timeout past the transport ceiling is rejected") + func watchTimeoutTooLarge() { + #expect(throws: EtcdParseError.self) { + try EtcdCommandParser.parse("watch key --timeout 1e30") + } + #expect(throws: EtcdParseError.self) { + try EtcdCommandParser.parse( + "watch key --timeout \(Int(EtcdCommandParser.maximumWatchTimeout) + 1)" + ) + } + } + + @Test("A non-finite timeout is rejected") + func watchTimeoutNotFinite() { + #expect(throws: EtcdParseError.self) { + try EtcdCommandParser.parse("watch key --timeout inf") + } + #expect(throws: EtcdParseError.self) { + try EtcdCommandParser.parse("watch key --timeout nan") + } + } + + @Test("Zero and the ceiling are accepted") + func watchTimeoutBounds() throws { + let zero = try EtcdCommandParser.parse("watch key --timeout 0") + guard case .watch(_, _, let zeroTimeout) = zero else { + Issue.record("Expected .watch") + return + } + #expect(zeroTimeout == 0) + + let ceiling = try EtcdCommandParser.parse( + "watch key --timeout \(Int(EtcdCommandParser.maximumWatchTimeout))" + ) + guard case .watch(_, _, let ceilingTimeout) = ceiling else { + Issue.record("Expected .watch") + return + } + #expect(ceilingTimeout == EtcdCommandParser.maximumWatchTimeout) + } } // MARK: - Lease Commands diff --git a/TableProTests/Plugins/EtcdGatewayRouteTests.swift b/TableProTests/Plugins/EtcdGatewayRouteTests.swift new file mode 100644 index 0000000000..b2dee632c1 --- /dev/null +++ b/TableProTests/Plugins/EtcdGatewayRouteTests.swift @@ -0,0 +1,54 @@ +// +// EtcdGatewayRouteTests.swift +// TableProTests +// +// Which gateway prefix an etcd server routes is a 404-or-not question. Measured: etcd 3.2 +// serves only /v3alpha, 3.3 only /v3beta, and 3.4 onward /v3. +// + +import Foundation +import Testing + +private func gatewayBody(_ text: String) -> Data { + Data(text.utf8) +} + +@Suite("EtcdGatewayRoute") +struct EtcdGatewayRouteTests { + @Test("v3 is tried before the legacy prefixes") + func prefixOrder() { + #expect(EtcdGatewayRoute.candidatePrefixes == ["v3", "v3beta", "v3alpha"]) + } + + @Test("Only 404 means the prefix is not routed") + func notFoundIsTheOnlyRejection() { + #expect( + EtcdGatewayRoute.classify(httpStatus: 404, body: gatewayBody("404 page not found")) == .notRouted + ) + #expect( + EtcdGatewayRoute.classify(httpStatus: 404, body: gatewayBody(#"{"code":5}"#)) == .notRouted + ) + } + + @Test("An auth fault proves the prefix is routed") + func authFaultIsRouted() { + let fault = gatewayBody(#"{"code":3, "message":"etcdserver: user name is empty"}"#) + #expect(EtcdGatewayRoute.classify(httpStatus: 400, body: fault) == .routed) + #expect(EtcdGatewayRoute.classify(httpStatus: 401, body: fault) == .routed) + #expect(EtcdGatewayRoute.classify(httpStatus: 403, body: fault) == .routed) + } + + @Test("A successful answer is routed") + func successIsRouted() { + #expect( + EtcdGatewayRoute.classify(httpStatus: 200, body: gatewayBody(#"{"header":{}}"#)) == .routed + ) + } + + @Test("A non-etcd answer is not accepted as a gateway") + func nonEtcdAnswers() { + #expect(EtcdGatewayRoute.classify(httpStatus: 200, body: gatewayBody("")) == .notEtcd) + #expect(EtcdGatewayRoute.classify(httpStatus: 200, body: Data()) == .notEtcd) + #expect(EtcdGatewayRoute.classify(httpStatus: 200, body: gatewayBody("[1,2,3]")) == .notEtcd) + } +} diff --git a/TableProTests/Plugins/EtcdRequestRecoveryTests.swift b/TableProTests/Plugins/EtcdRequestRecoveryTests.swift new file mode 100644 index 0000000000..269a24e9a8 --- /dev/null +++ b/TableProTests/Plugins/EtcdRequestRecoveryTests.swift @@ -0,0 +1,104 @@ +// +// EtcdRequestRecoveryTests.swift +// TableProTests +// +// Mirrors clientv3's shouldRefreshToken: refresh on an invalid token, on a stale auth store +// revision, and on a missing token when credentials exist. Never on permission denied. +// + +import Foundation +import Testing + +private func recoveryFault(code: Int, message: String) -> EtcdServerFault { + EtcdServerFault(grpcCode: code, message: message) +} + +private let missingToken = recoveryFault(code: 3, message: "etcdserver: user name is empty") +private let wrongPassword = recoveryFault( + code: 3, + message: "etcdserver: authentication failed, invalid user ID or password" +) +private let staleRevision = recoveryFault(code: 3, message: "etcdserver: revision of auth store is old") +private let rejectedToken = recoveryFault(code: 16, message: "etcdserver: invalid auth token") +private let deniedPermission = recoveryFault(code: 7, message: "etcdserver: permission denied") +private let authOff = recoveryFault(code: 9, message: "etcdserver: authentication is not enabled") + +@Suite("EtcdRequestRecovery") +struct EtcdRequestRecoveryTests { + @Test("A rejected token is refreshed once") + func rejectedTokenRefreshes() { + #expect( + EtcdRequestRecovery.action(for: rejectedToken, hasCredentials: true, isRetry: false) + == .reauthenticateAndRetry + ) + } + + @Test("A missing token is refreshed when credentials exist") + func missingTokenRefreshes() { + #expect( + EtcdRequestRecovery.action(for: missingToken, hasCredentials: true, isRetry: false) + == .reauthenticateAndRetry + ) + } + + @Test("A stale auth store revision is refreshed") + func staleRevisionRefreshes() { + #expect( + EtcdRequestRecovery.action(for: staleRevision, hasCredentials: true, isRetry: false) + == .reauthenticateAndRetry + ) + } + + @Test("Permission denied is never retried") + func permissionDeniedSurfaces() { + #expect( + EtcdRequestRecovery.action(for: deniedPermission, hasCredentials: true, isRetry: false) + == .surface + ) + } + + @Test("A wrong password is never retried") + func wrongPasswordSurfaces() { + #expect( + EtcdRequestRecovery.action(for: wrongPassword, hasCredentials: true, isRetry: false) + == .surface + ) + } + + @Test("Authentication being off is never retried") + func authOffSurfaces() { + #expect( + EtcdRequestRecovery.action(for: authOff, hasCredentials: true, isRetry: false) == .surface + ) + } + + @Test("A connection with no credentials cannot recover") + func noCredentialsSurfaces() { + #expect( + EtcdRequestRecovery.action(for: missingToken, hasCredentials: false, isRetry: false) + == .surface + ) + #expect( + EtcdRequestRecovery.action(for: rejectedToken, hasCredentials: false, isRetry: false) + == .surface + ) + } + + @Test("A retry never refreshes again") + func retryDoesNotLoop() { + #expect( + EtcdRequestRecovery.action(for: rejectedToken, hasCredentials: true, isRetry: true) == .surface + ) + #expect( + EtcdRequestRecovery.action(for: missingToken, hasCredentials: true, isRetry: true) == .surface + ) + } + + @Test("An unrecognised fault is surfaced") + func unclassifiedSurfaces() { + let fault = recoveryFault(code: 5, message: "etcdserver: key is not provided") + #expect( + EtcdRequestRecovery.action(for: fault, hasCredentials: true, isRetry: false) == .surface + ) + } +} diff --git a/TableProTests/Plugins/EtcdServerFaultTests.swift b/TableProTests/Plugins/EtcdServerFaultTests.swift new file mode 100644 index 0000000000..383ffcfa50 --- /dev/null +++ b/TableProTests/Plugins/EtcdServerFaultTests.swift @@ -0,0 +1,152 @@ +// +// EtcdServerFaultTests.swift +// TableProTests +// +// etcd's gateway reports every failure as a gRPC code plus a message, and grpc-gateway maps +// two distinct codes onto HTTP 400. These pin the classification to the code, measured against +// etcd 3.5.17 and 3.6.1. +// + +import Foundation +import Testing + +private func etcdBody(_ json: String) -> Data { + Data(json.utf8) +} + +@Suite("EtcdServerFault - classification") +struct EtcdServerFaultClassificationTests { + @Test("etcd 3.6 reports a missing token as InvalidArgument, not Unauthorized") + func missingTokenOnEtcd36() { + let fault = EtcdServerFault.decode( + httpStatus: 400, + body: etcdBody(#"{"code":3, "message":"etcdserver: user name is empty"}"#) + ) + #expect(fault.grpcCode == 3) + #expect(fault.kind == .credentialsRequired) + } + + @Test("etcd 3.5 carries the same fault with an extra error key") + func missingTokenOnEtcd35() { + let fault = EtcdServerFault.decode( + httpStatus: 400, + body: etcdBody( + #"{"error":"etcdserver: user name is empty","code":3,"message":"etcdserver: user name is empty"}"# + ) + ) + #expect(fault.kind == .credentialsRequired) + #expect(fault.message == "etcdserver: user name is empty") + } + + @Test("A wrong password shares the gRPC code of a missing token") + func wrongPassword() { + let fault = EtcdServerFault.decode( + httpStatus: 400, + body: etcdBody( + #"{"code":3, "message":"etcdserver: authentication failed, invalid user ID or password"}"# + ) + ) + #expect(fault.kind == .credentialsRejected) + } + + @Test("A stale auth store revision is its own kind") + func staleAuthRevision() { + let fault = EtcdServerFault.decode( + httpStatus: 400, + body: etcdBody(#"{"code":3, "message":"etcdserver: revision of auth store is old"}"#) + ) + #expect(fault.kind == .authRevisionStale) + } + + @Test("Authentication disabled on the server is recognised") + func authNotEnabled() { + let fault = EtcdServerFault.decode( + httpStatus: 400, + body: etcdBody(#"{"code":9, "message":"etcdserver: authentication is not enabled"}"#) + ) + #expect(fault.kind == .authNotEnabled) + } + + @Test("Another FailedPrecondition stays unclassified") + func roleNotFound() { + let fault = EtcdServerFault.decode( + httpStatus: 400, + body: etcdBody(#"{"code":9, "message":"etcdserver: role name not found"}"#) + ) + #expect(fault.kind == .unclassified) + } + + @Test("Permission denied proves the session is live") + func permissionDenied() { + let fault = EtcdServerFault.decode( + httpStatus: 403, + body: etcdBody(#"{"code":7, "message":"etcdserver: permission denied"}"#) + ) + #expect(fault.kind == .permissionDenied) + #expect(fault.provesLiveSession) + } + + @Test("A rejected token is its own kind") + func invalidToken() { + let fault = EtcdServerFault.decode( + httpStatus: 401, + body: etcdBody(#"{"code":16, "message":"etcdserver: invalid auth token"}"#) + ) + #expect(fault.kind == .tokenRejected) + #expect(!fault.provesLiveSession) + } + + @Test("The HTTP status never decides the classification") + func statusIsNotConsulted() { + let payload = etcdBody(#"{"code":16, "message":"etcdserver: invalid auth token"}"#) + #expect(EtcdServerFault.decode(httpStatus: 401, body: payload).kind == .tokenRejected) + #expect(EtcdServerFault.decode(httpStatus: 500, body: payload).kind == .tokenRejected) + #expect(EtcdServerFault.decode(httpStatus: 200, body: payload).kind == .tokenRejected) + } +} + +@Suite("EtcdServerFault - decoding") +struct EtcdServerFaultDecodingTests { + @Test("A plain text body survives as the message") + func plainTextBody() { + let fault = EtcdServerFault.decode(httpStatus: 404, body: etcdBody("404 page not found\n")) + #expect(fault.grpcCode == nil) + #expect(fault.message == "404 page not found") + #expect(fault.kind == .unclassified) + } + + @Test("An empty body falls back to the HTTP status") + func emptyBody() { + let fault = EtcdServerFault.decode(httpStatus: 502, body: Data()) + #expect(fault.grpcCode == nil) + #expect(fault.message.contains("502")) + } + + @Test("A long plain text body is capped") + func longBody() { + let fault = EtcdServerFault.decode( + httpStatus: 500, + body: etcdBody(String(repeating: "x", count: 4_000)) + ) + #expect(fault.message.count == 512) + } + + @Test("A missing token asks for credentials instead of quoting etcd") + func credentialsRequiredMessage() { + let fault = EtcdServerFault.decode( + httpStatus: 400, + body: etcdBody(#"{"code":3, "message":"etcdserver: user name is empty"}"#) + ) + #expect(fault.localizedDescription.contains("username")) + #expect(!fault.localizedDescription.contains("user name is empty")) + } + + @Test("Permission denied reads as etcd wrote it") + func permissionDeniedMessage() { + let fault = EtcdServerFault.decode( + httpStatus: 403, + body: etcdBody(#"{"code":7, "message":"etcdserver: permission denied"}"#) + ) + #expect(fault.localizedDescription == "etcdserver: permission denied") + } +} diff --git a/docs/databases/etcd.mdx b/docs/databases/etcd.mdx index 91c2ed82f8..b0bd16f732 100644 --- a/docs/databases/etcd.mdx +++ b/docs/databases/etcd.mdx @@ -19,7 +19,7 @@ Click **New Connection…**, select **etcd**, enter host and port, then click ** |-------|-------------| | **Host** | etcd server hostname or IP. `localhost` by default | | **Port** | Client port, `2379` by default | -| **Username / Password** | Only when etcd authentication is enabled | +| **Username / Password** | The etcd user to authenticate as. Leave both empty on a server that has no authentication | ### Advanced fields @@ -30,6 +30,14 @@ Click **New Connection…**, select **etcd**, enter host and port, then click ** | **CA Certificate** | Path to `ca.pem`, for Verify CA and Verify Identity | | **Client Certificate** / **Client Key** | Paths to the client certificate and private key, for mutual TLS | +## Authentication + +Fill in **Username** and **Password** once the server has run `auth enable`. Credentials against a server that has authentication off still connect, unauthenticated, the way `etcdctl --user` does. + +The user does not need the `root` role. A user whose roles cover part of the keyspace connects and browses what those roles allow. `endpoint status` and the `user` and `role` commands do need `root`, and answer `etcdserver: permission denied` without it. On etcd 3.6.0 through 3.6.11 that also leaves the database size blank. + +Running `auth enable` from the command editor turns authentication on for the whole cluster. A connection that already holds a username and password keeps working; one that holds neither cannot, and has to be reopened with credentials. + ## Connection URL ```text @@ -88,13 +96,25 @@ TLS here is set by **TLS Mode** in the Advanced fields, not by the connection's ## Troubleshooting -### No supported etcd API found +### No etcd v3 API found + +The gateway answered 404 on `/v3`, `/v3beta` and `/v3alpha`. Point the connection at the client port, 2379 by default, and not the peer port on 2380. + +### The server answered but is not an etcd v3 JSON gateway + +Something is listening on that port and it is not etcd. A reverse proxy that rewrites 404 into a branded page does this too. + +### This etcd server has authentication enabled + +The server wants a token and the connection carries no credentials. Add a **Username** and **Password** and connect again. + +### etcdserver: authentication failed, invalid user ID or password -The HTTP/JSON gateway answered on none of the v3 paths. Point the connection at the client port, 2379 by default, and not the peer port on 2380. +etcd rejected the credentials. Check them with `etcdctl --user=name:password endpoint health`. -### Authentication failed +### etcdserver: permission denied -The username and password were rejected, or the user holds no role covering the key. Check both, and that `auth enable` has been run. +The credentials were accepted and the user's roles do not cover that key or command. `endpoint status`, `user` and `role` need the `root` role. ## Related diff --git a/project.yml b/project.yml index 3d748151bf..d7cdd27ae6 100644 --- a/project.yml +++ b/project.yml @@ -445,7 +445,10 @@ targets: - Plugins/ElasticsearchDriverPlugin/ElasticsearchQueryBuilder.swift - Plugins/ElasticsearchDriverPlugin/ElasticsearchStatementGenerator.swift - Plugins/EtcdDriverPlugin/EtcdCommandParser.swift + - Plugins/EtcdDriverPlugin/EtcdGatewayRoute.swift - Plugins/EtcdDriverPlugin/EtcdQueryBuilder.swift + - Plugins/EtcdDriverPlugin/EtcdRequestRecovery.swift + - Plugins/EtcdDriverPlugin/EtcdServerFault.swift - Plugins/EtcdDriverPlugin/EtcdStatementGenerator.swift - Plugins/TypesenseDriverPlugin/TypesenseApiKeys.swift - Plugins/TypesenseDriverPlugin/TypesenseConsoleParser.swift diff --git a/scripts/check-etcd-auth-faults.sh b/scripts/check-etcd-auth-faults.sh new file mode 100755 index 0000000000..e1dd47807d --- /dev/null +++ b/scripts/check-etcd-auth-faults.sh @@ -0,0 +1,138 @@ +#!/usr/bin/env bash +# +# Compare EtcdServerFault's classification against a real etcd server. +# +# etcd's HTTP/JSON gateway reports every failure as a gRPC status code plus an English message, +# and grpc-gateway's HTTPStatusFromCode maps InvalidArgument(3) and FailedPrecondition(9) onto +# the same HTTP 400. The code alone therefore cannot separate "you sent no token" from "your +# password is wrong" from "auth is not enabled", so EtcdServerFault matches a substring of the +# message. Those substrings are a hand transcription of etcd's own error strings that nothing at +# runtime checks: if a release reworded one, TablePro would stop re-authenticating and would show +# the wrong message, silently. This drives a live etcd through each fault and diffs what it says +# against the markers in the Swift source. +# +# Usage: +# scripts/check-etcd-auth-faults.sh [host] [port] +# +# Needs curl, python3, and an etcd whose authentication is ENABLED with a known user. Set +# ETCD_USER and ETCD_PASSWORD (default root/root). This is a manual check, not a CI gate. +# +# Not covered: the auth-store-old-revision marker. Reproducing it needs the server's auth +# revision to move under a token that is still otherwise valid, which grant, revoke and password +# changes do not produce (they answer permission-denied or invalid-token instead). Re-read that +# one marker by hand against api/v3rpc/rpctypes/error.go when etcd is upgraded. + +set -uo pipefail + +HOST="${1:-127.0.0.1}" +PORT="${2:-2379}" +USER_NAME="${ETCD_USER:-root}" +PASSWORD="${ETCD_PASSWORD:-root}" +BASE="http://$HOST:$PORT/v3" +SOURCE="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)/Plugins/EtcdDriverPlugin/EtcdServerFault.swift" +HEALTH_KEY="aGVhbHRo" + +for tool in curl python3; do + command -v "$tool" > /dev/null || { + echo "$tool not found" >&2 + exit 3 + } +done +[ -f "$SOURCE" ] || { + echo "not found: $SOURCE" >&2 + exit 3 +} +if ! curl -s --max-time 5 -XPOST "$BASE/kv/range" -d "{\"key\":\"$HEALTH_KEY\"}" > /dev/null; then + echo "no etcd v3 gateway at $HOST:$PORT" >&2 + exit 3 +fi + +marker() { + python3 - "$SOURCE" "$1" <<'PY' +import re +import sys + +source, name = sys.argv[1], sys.argv[2] +text = open(source, encoding="utf-8").read() +match = re.search(r'static let %s = "([^"]+)"' % re.escape(name), text) +print(match.group(1) if match else "") +PY +} + +# Print "\t" for one request. +observe() { + local path="$1" body="$2" header="${3:-}" + local response + if [ -n "$header" ]; then + response="$(curl -s --max-time 10 -XPOST -H "Authorization: $header" "$BASE/$path" -d "$body")" + else + response="$(curl -s --max-time 10 -XPOST "$BASE/$path" -d "$body")" + fi + printf '%s' "$response" | python3 -c ' +import json +import sys + +try: + payload = json.load(sys.stdin) +except Exception: + print("\t") + sys.exit() +print("%s\t%s" % (payload.get("code", ""), payload.get("message") or payload.get("error") or "")) +' +} + +TOKEN="$(curl -s --max-time 10 -XPOST "$BASE/auth/authenticate" \ + -d "{\"name\":\"$USER_NAME\",\"password\":\"$PASSWORD\"}" | + python3 -c 'import json,sys; print(json.load(sys.stdin).get("token",""))')" +if [ -z "$TOKEN" ]; then + echo "could not authenticate as $USER_NAME; is authentication enabled on this server?" >&2 + exit 3 +fi + +STATUS=0 + +expect() { + local label="$1" want_code="$2" marker_name="$3" observed="$4" + local code="${observed%%$'\t'*}" message="${observed#*$'\t'}" + + if [ "$code" != "$want_code" ]; then + echo "FAIL $label: etcd answered gRPC code '$code', EtcdServerFault classifies $want_code" + STATUS=1 + return + fi + if [ -n "$marker_name" ]; then + local want_marker + want_marker="$(marker "$marker_name")" + if [ -z "$want_marker" ]; then + echo "FAIL $label: EtcdServerFault has no marker named $marker_name" + STATUS=1 + return + fi + case "$message" in + *"$want_marker"*) ;; + *) + echo "FAIL $label: etcd says \"$message\", EtcdServerFault looks for \"$want_marker\"" + STATUS=1 + return + ;; + esac + fi + echo "ok $label: code $code, \"$message\"" +} + +expect "missing token" 3 userEmptyMarker \ + "$(observe kv/range "{\"key\":\"$HEALTH_KEY\"}")" +expect "wrong password" 3 authFailedMarker \ + "$(observe auth/authenticate "{\"name\":\"$USER_NAME\",\"password\":\"$PASSWORD.wrong\"}")" +expect "rejected token" 16 "" \ + "$(observe kv/range "{\"key\":\"$HEALTH_KEY\"}" 'not-a-real-token.1')" +expect "auth already enabled" 9 "" \ + "$(observe auth/enable '{}' "$TOKEN")" + +echo +if [ "$STATUS" -eq 0 ]; then + echo "EtcdServerFault agrees with $HOST:$PORT" +else + echo "EtcdServerFault disagrees with $HOST:$PORT; update the markers or the classification" +fi +exit "$STATUS"