diff --git a/CMakeLists.txt b/CMakeLists.txt index 68016730aa..682c0836de 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -447,17 +447,31 @@ if(DESKTOP) set(CURL_DISABLE_INSTALL ON) if(WIN32) - # Use Windows native Schannel/SSPI for TLS instead of OpenSSL/BoringSSL. + # Use Windows native Schannel/SSPI (Crypt32) for TLS instead of + # OpenSSL/BoringSSL. Schannel verifies server certificates against the + # Windows Certificate Store and enforces native OCSP/CRL revocation + # checking (SCH_CRED_REVOCATION_CHECK_CHAIN_EXCLUDE_ROOT) as well as + # Microsoft Certificate Trust List (CTL) / disallowed-CA updates. set(CURL_USE_SCHANNEL ON) elseif(APPLE) # Enable Apple Security.framework (SecTrust) native OS certificate # verification alongside OpenSSL/BoringSSL, so macOS builds verify peers - # against the system Keychain rather than requiring a CA bundle file path. + # against the macOS System Keychain (SecTrustEvaluateWithError) and enforce + # Apple's native OCSP/CRL policies and Valid/trustd revocation updates + # rather than relying on a static CA bundle file path. set(USE_APPLE_SECTRUST ON) # When curl detects static BoringSSL with a Clang toolchain, it appends # ${_curl_stdlib_cxx} to its link interface. Override curl's default of # "stdc++" (removed from modern Xcode SDKs) to Apple's "c++" (libc++). set(_curl_stdlib_cxx "c++") + else() + # On Linux, libcurl uses OpenSSL/BoringSSL and verifies server certificates + # against the OS trust store managed by the distribution's ca-certificates + # package (which removes or distrusts compromised root and intermediate CAs). + # Note: TransportCurl and WebSocketClientImpl resolve CURLOPT_CAINFO and + # CURLOPT_CAPATH at runtime across standard Linux distribution paths so + # prebuilt binaries do not depend solely on the build machine's detected + # CURL_CA_BUNDLE path. endif() # Disable optional external libraries that curl 8.23.0 enables by default: @@ -481,7 +495,8 @@ if(DESKTOP) set(CURL_ZSTD OFF CACHE STRING "Build curl with zstd support" FORCE) # curl 8.23.0 removed the legacy umbrella HTTP_ONLY option. Explicitly disable - # every non-HTTP(S) protocol so only HTTP and HTTPS are compiled into libcurl. + # every unused protocol so only HTTP, HTTPS, WS, and WSS (used by the REST + # layer and Realtime Database WebSockets) are compiled into libcurl. set(CURL_DISABLE_DICT ON) set(CURL_DISABLE_FILE ON) set(CURL_DISABLE_FTP ON) @@ -496,7 +511,7 @@ if(DESKTOP) set(CURL_DISABLE_SMTP ON) set(CURL_DISABLE_TELNET ON) set(CURL_DISABLE_TFTP ON) - set(CURL_DISABLE_WEBSOCKETS ON) + set(CURL_DISABLE_WEBSOCKETS OFF) # Skip building curl's test suite and example binaries. set(BUILD_TESTING OFF) @@ -505,77 +520,11 @@ if(DESKTOP) add_external_library(curl) # Clean up generic normal variables set specifically for curl so they do not - # leak into subsequent add_subdirectory() calls (libuv, zlib, firestore, etc.). + # leak into subsequent add_subdirectory() calls (zlib, firestore, etc.). unset(BUILD_STATIC_LIBS) unset(BUILD_EXAMPLES) - add_external_library(libuv) - add_external_library(zlib) - - add_external_library(uWebSockets) - - # Binutils on Mac doesn't support thread-local storage (required by - # websockets), but because we only use websockets via the scheduler, we don't - # need it. Deactivate this by blanking out the __thread keyword. - set(websockets_additional_defines "-D__thread=") - - # uWebSockets does not come with a CMakeLists file, so define the target. - # Note that since it depends on OpenSSL, only do so if that was found. - if (OPENSSL_FOUND) - if (MSVC OR APPLE) - set(uWebSockets_extra_src) - else() - # Epoll is only used on Linux, otherwise LibUV is used. - set(uWebSockets_extra_src - ${UWEBSOCKETS_SOURCE_DIR}/src/Epoll.cpp) - endif() - add_library(libuWS STATIC - ${uWebSockets_extra_src} - ${UWEBSOCKETS_SOURCE_DIR}/src/Extensions.cpp - ${UWEBSOCKETS_SOURCE_DIR}/src/Group.cpp - ${UWEBSOCKETS_SOURCE_DIR}/src/HTTPSocket.cpp - ${UWEBSOCKETS_SOURCE_DIR}/src/Hub.cpp - ${UWEBSOCKETS_SOURCE_DIR}/src/Networking.cpp - ${UWEBSOCKETS_SOURCE_DIR}/src/Node.cpp - ${UWEBSOCKETS_SOURCE_DIR}/src/Socket.cpp - ${UWEBSOCKETS_SOURCE_DIR}/src/WebSocket.cpp) - if(MSVC) - set(websockets_additional_defines - ${websockets_additional_defines} - -DWIN32_LEAN_AND_MEAN # Ensure that windows doesn't include winsock.h by - # default, as it can cause issues when libraries try - # to include winsock2.h later on in the process. - -DUSE_LIBUV=1 - ) - elseif(APPLE) - set(websockets_additional_defines - ${websockets_additional_defines} - -DUSE_LIBUV=1 - ) - endif() - - target_compile_definitions(libuWS - PRIVATE - ${websockets_additional_defines} - ) - target_include_directories(libuWS - PUBLIC - ${LIBUV_SOURCE_DIR}/include - ${UWEBSOCKETS_SOURCE_DIR} - ${ZLIB_SOURCE_DIR} - ${ZLIB_BINARY_DIR} - PRIVATE - ${OPENSSL_INCLUDE_DIR} - ) - target_link_libraries(libuWS - PRIVATE - OpenSSL::SSL - OpenSSL::Crypto - uv_a - zlibstatic - ) - endif() endif() if(DESKTOP AND NOT CMAKE_CROSSCOMPILING) diff --git a/README.md b/README.md index d8651d95ba..7ed28b2711 100644 --- a/README.md +++ b/README.md @@ -157,11 +157,9 @@ Currently, the third party libraries that can be provided this way are: | Library | | ------- | -| CURL | +| CURL (8.14.0 or newer, built with WebSocket support) | | FLATBUFFERS | -| LIBUV | | NANOPB | -| UWEBSOCKETS | | ZLIB | #### Building with CMake for iOS diff --git a/app/rest/CMakeLists.txt b/app/rest/CMakeLists.txt index cc22dd0171..f051d1f9fd 100644 --- a/app/rest/CMakeLists.txt +++ b/app/rest/CMakeLists.txt @@ -26,6 +26,8 @@ enable_language(CXX) set(rest_SRCS controller_curl.cc controller_interface.cc + curl/curl_handles.cc + curl/tls_policy.cc gzipheader.cc request.cc request_binary_gzip.cc @@ -75,8 +77,8 @@ if (MSVC) # libcurl on Windows (Schannel/SSPI) depends on these Windows SDK libraries: # crypt32 (certificate store), bcrypt (BCryptGenRandom), iphlpapi # (if_nametoindex) and secur32 (InitSecurityInterface). These are also - # pulled in via #pragma comment(lib) in transport_curl.cc for consumers of - # the merged firebase_app.lib. + # pulled in via #pragma comment(lib) in curl_handles.cc and transport_curl.cc + # for consumers of the merged firebase_app.lib. set(FIREBASE_CURL_DEPENDENCY ${FIREBASE_CURL_DEPENDENCY} crypt32 bcrypt iphlpapi secur32) endif() diff --git a/app/rest/curl/curl_handles.cc b/app/rest/curl/curl_handles.cc new file mode 100644 index 0000000000..788d6059b9 --- /dev/null +++ b/app/rest/curl/curl_handles.cc @@ -0,0 +1,66 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "app/rest/curl/curl_handles.h" + +#include +#include + +#include "app/src/assert.h" + +#if defined(_MSC_VER) +#pragma comment(lib, "advapi32.lib") +#pragma comment(lib, "bcrypt.lib") +#pragma comment(lib, "crypt32.lib") +#pragma comment(lib, "iphlpapi.lib") +#pragma comment(lib, "secur32.lib") +#pragma comment(lib, "ws2_32.lib") +#endif // defined(_MSC_VER) + +namespace firebase { +namespace rest { +namespace curl { + +namespace { + +std::mutex g_curl_global_mutex; +int64_t g_curl_global_refcount = 0; + +} // namespace + +void AcquireCurlGlobal() { + std::lock_guard lock(g_curl_global_mutex); + if (g_curl_global_refcount++ == 0) { + CURLcode code = curl_global_init(CURL_GLOBAL_ALL); + FIREBASE_ASSERT_MESSAGE(code == CURLE_OK, + "curl_global_init(CURL_GLOBAL_ALL) failed: %s", + curl_easy_strerror(code)); + } +} + +void ReleaseCurlGlobal() { + std::lock_guard lock(g_curl_global_mutex); + FIREBASE_ASSERT_MESSAGE( + g_curl_global_refcount > 0, + "ReleaseCurlGlobal called more times than AcquireCurlGlobal"); + if (--g_curl_global_refcount == 0) { + curl_global_cleanup(); + } +} + +} // namespace curl +} // namespace rest +} // namespace firebase diff --git a/app/rest/curl/curl_handles.h b/app/rest/curl/curl_handles.h new file mode 100644 index 0000000000..3759b49ab5 --- /dev/null +++ b/app/rest/curl/curl_handles.h @@ -0,0 +1,74 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef FIREBASE_APP_REST_CURL_CURL_HANDLES_H_ +#define FIREBASE_APP_REST_CURL_CURL_HANDLES_H_ + +#include + +#include "curl/curl.h" + +// The desktop SDK drives libcurl through APIs that only exist in recent +// releases: curl_multi_poll()/curl_multi_wakeup() (7.68), CURLOPT_PROTOCOLS_STR +// (7.85), CURLOPT_CONNECT_ONLY=2 WebSocket handshakes (7.86) and +// CURLWS_NOAUTOPONG (8.14). Fail at compile time with a clear message rather +// than with scattered "undeclared identifier" errors when a user points +// CURL_SOURCE_DIR at an older tree. +static_assert(LIBCURL_VERSION_NUM >= 0x080e00, + "Firebase C++ SDK requires libcurl >= 8.14.0 (WebSockets with " + "CURLWS_NOAUTOPONG)"); + +namespace firebase { +namespace rest { +namespace curl { + +// RAII ownership for libcurl easy, multi, and slist handles. All deleters are +// null-safe. These are plain std::unique_ptr aliases so ownership transfer, +// release() and reset() behave exactly as for any other unique_ptr. +struct EasyDeleter { + void operator()(CURL* easy) const { + if (easy != nullptr) curl_easy_cleanup(easy); + } +}; + +struct MultiDeleter { + void operator()(CURLM* multi) const { + if (multi != nullptr) curl_multi_cleanup(multi); + } +}; + +struct SListDeleter { + void operator()(curl_slist* list) const { + if (list != nullptr) curl_slist_free_all(list); + } +}; + +using UniqueEasy = std::unique_ptr; +using UniqueMulti = std::unique_ptr; +using UniqueSList = std::unique_ptr; + +// Reference-counted curl_global_init() / curl_global_cleanup(). Thread-safe. +// Every libcurl user in the SDK holds a reference while it owns any CURL +// handle, so libcurl's process-wide state is initialized exactly once and torn +// down only after the last user is gone. +void AcquireCurlGlobal(); +void ReleaseCurlGlobal(); + +} // namespace curl +} // namespace rest +} // namespace firebase + +#endif // FIREBASE_APP_REST_CURL_CURL_HANDLES_H_ diff --git a/app/rest/curl/tls_policy.cc b/app/rest/curl/tls_policy.cc new file mode 100644 index 0000000000..c7abb0ba75 --- /dev/null +++ b/app/rest/curl/tls_policy.cc @@ -0,0 +1,165 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "app/rest/curl/tls_policy.h" + +#ifdef FIREBASE_SSL_CAPATH +#include +#include + +#include +#endif // FIREBASE_SSL_CAPATH + +namespace firebase { +namespace rest { +namespace curl { + +namespace { + +// Applies a single option and folds the result into *first_error, keeping only +// the first failure so the caller sees the root cause. +template +void ApplyOpt(CURL* easy, CURLoption option, T value, CURLcode* first_error) { + CURLcode code = curl_easy_setopt(easy, option, value); + if (code != CURLE_OK && *first_error == CURLE_OK) *first_error = code; +} + +#ifdef FIREBASE_SSL_CAPATH +bool IsReadableRegularFile(const char* path) { + if (path == nullptr || path[0] == '\0') return false; + struct stat st; + return stat(path, &st) == 0 && S_ISREG(st.st_mode) && access(path, R_OK) == 0; +} + +bool IsReadableDirectory(const char* path) { + if (path == nullptr || path[0] == '\0') return false; + struct stat st; + return stat(path, &st) == 0 && S_ISDIR(st.st_mode) && + access(path, R_OK | X_OK) == 0; +} +#endif // FIREBASE_SSL_CAPATH + +} // namespace + +CURLcode ApplyTlsPolicy(CURL* easy) { + CURLcode first_error = CURLE_OK; + // Verify the authenticity of the server's TLS certificate chain against the + // trusted CA store (1L = enabled). + ApplyOpt(easy, CURLOPT_SSL_VERIFYPEER, 1L, &first_error); + // Verify that the server certificate's Subject Alternative Name (SAN) or + // Common Name (CN) matches the target hostname (2L = strict hostname match; + // 1L is deprecated in libcurl). + ApplyOpt(easy, CURLOPT_SSL_VERIFYHOST, 2L, &first_error); + // Require TLS 1.2 or newer (CURL_SSLVERSION_TLSv1_2 sets the minimum allowed + // TLS protocol version in libcurl >= 7.54.0, permitting TLS 1.2 and TLS 1.3 + // while rejecting TLS 1.0 and 1.1). + ApplyOpt(easy, CURLOPT_SSLVERSION, + static_cast(CURL_SSLVERSION_TLSv1_2), // NOLINT + &first_error); +#ifdef FIREBASE_SSL_CAPATH + // Explicitly setting CURLOPT_CAINFO (even to nullptr) marks + // ssl.custom_cafile = TRUE inside libcurl, which stops it from substituting + // the build machine's compile-time CURL_CA_BUNDLE path. See the comment on + // LinuxCaBundlePath() for why that matters. + ApplyOpt(easy, CURLOPT_CAINFO, LinuxCaBundlePath(), &first_error); + CURLcode capath_code = + curl_easy_setopt(easy, CURLOPT_CAPATH, LinuxCaDirectoryPath()); + if (capath_code != CURLE_OK && capath_code != CURLE_NOT_BUILT_IN && + first_error == CURLE_OK) { + first_error = capath_code; + } +#endif // FIREBASE_SSL_CAPATH + return first_error; +} + +// Standard system CA bundle locations across major Linux distributions +// (matching Go's crypto/x509/root_linux.go and libcurl's CMakeLists.txt / +// acinclude.m4 CURL_CHECK_CA_BUNDLE detection): +// - /etc/ssl/certs/ca-certificates.crt: Debian, Ubuntu, Gentoo, Arch, Alpine +// - /etc/pki/tls/certs/ca-bundle.crt: Fedora, RHEL, CentOS, Rocky, Alma, +// Amazon Linux +// - /etc/ssl/ca-bundle.pem: openSUSE, SLES +// - /etc/pki/tls/cacert.pem: OpenELEC +// - /etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem: CentOS/RHEL/Fedora +// p11-kit trust store +// - /etc/ssl/cert.pem: Alpine, Arch, Fedora, FreeBSD/OpenBSD symlink +// +// Why we probe at runtime instead of relying on libcurl's compile-time +// CURL_CA_BUNDLE: +// 1. At CMake configure time, libcurl detects the CA bundle path on the +// *build* host (typically Ubuntu's /etc/ssl/certs/ca-certificates.crt) and +// compiles that single path into CURL_CA_BUNDLE. +// 2. In libcurl's OpenSSL backend (ossl_populate_x509_store in +// lib/vtls/openssl.c), ssl_cafile is loaded before ssl_capath (via +// X509_STORE_load_file on OpenSSL 3.0+, or X509_STORE_load_locations on +// OpenSSL < 3.0 and BoringSSL). If ssl_cafile points to a build-host +// CURL_CA_BUNDLE file that does not exist on the runtime machine, libcurl +// immediately aborts with CURLE_SSL_CACERT_BADFILE before ever loading +// ssl_capath. +// 3. Conversely, some Linux distributions and container images only ship a +// combined CA bundle file without populating OpenSSL c_rehash (.0) +// symlinks in /etc/ssl/certs, so setting only CURLOPT_CAPATH is +// insufficient. +// Explicitly setting CURLOPT_CAINFO to LinuxCaBundlePath() (even when it +// returns nullptr) marks ssl.custom_cafile = TRUE in libcurl +// (lib/vtls/vtls_config.c), preventing a missing build-host CURL_CA_BUNDLE +// path from breaking verification while loading both the runtime +// distribution's CA bundle and hashed CA directory when available. +const char* LinuxCaBundlePath() { +#ifdef FIREBASE_SSL_CAPATH + static const char* const kCaBundleEnvVars[] = { + "CURL_CA_BUNDLE", + "SSL_CERT_FILE", + }; + for (const char* env_var : kCaBundleEnvVars) { + const char* env_path = std::getenv(env_var); + if (IsReadableRegularFile(env_path)) { + return env_path; + } + } + static const char* const kCaBundlePaths[] = { + "/etc/ssl/certs/ca-certificates.crt", + "/etc/pki/tls/certs/ca-bundle.crt", + "/etc/ssl/ca-bundle.pem", + "/etc/pki/tls/cacert.pem", + "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem", + "/etc/ssl/cert.pem", + }; + for (const char* path : kCaBundlePaths) { + if (IsReadableRegularFile(path)) { + return path; + } + } +#endif // FIREBASE_SSL_CAPATH + return nullptr; +} + +const char* LinuxCaDirectoryPath() { +#ifdef FIREBASE_SSL_CAPATH + const char* env_dir = std::getenv("SSL_CERT_DIR"); + if (IsReadableDirectory(env_dir)) { + return env_dir; + } + if (IsReadableDirectory(FIREBASE_SSL_CAPATH)) { + return FIREBASE_SSL_CAPATH; + } +#endif // FIREBASE_SSL_CAPATH + return nullptr; +} + +} // namespace curl +} // namespace rest +} // namespace firebase diff --git a/app/rest/curl/tls_policy.h b/app/rest/curl/tls_policy.h new file mode 100644 index 0000000000..ac4f55b6bf --- /dev/null +++ b/app/rest/curl/tls_policy.h @@ -0,0 +1,64 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef FIREBASE_APP_REST_CURL_TLS_POLICY_H_ +#define FIREBASE_APP_REST_CURL_TLS_POLICY_H_ + +#include "curl/curl.h" + +namespace firebase { +namespace rest { +namespace curl { + +// Applies the SDK-wide TLS policy to a libcurl easy handle. This is the single +// place where transport security is configured for every HTTPS and WSS +// connection the SDK opens (REST transport and Realtime Database WebSockets). +// +// The policy is: +// * Always verify the peer certificate chain (CURLOPT_SSL_VERIFYPEER=1) and +// that the certificate matches the requested hostname/SANs +// (CURLOPT_SSL_VERIFYHOST=2). +// * Require TLS 1.2 or newer (CURLOPT_SSLVERSION=TLSv1_2). TLS 1.3 is still +// negotiated when available; TLS 1.0/1.1 downgrades are refused. +// * Trust the operating system's certificate store so that root and +// intermediate CAs revoked or distrusted by the OS vendor are rejected: +// - Windows (CURL_USE_SCHANNEL): Windows Certificate Store via Crypt32, +// with native OCSP/CRL and Microsoft CTL revocation checks. +// - macOS (USE_APPLE_SECTRUST): System Keychain via Security.framework +// (SecTrustEvaluateWithError), with Apple OCSP/CRL and Valid/trustd +// revocation checks. +// - Linux (OpenSSL/BoringSSL): the distribution's ca-certificates trust +// store, resolved at *runtime* by LinuxCaBundlePath() and +// LinuxCaDirectoryPath() rather than relying on the path libcurl +// detected on the build machine. +// +// Returns the first non-CURLE_OK result from curl_easy_setopt(), or CURLE_OK. +CURLcode ApplyTlsPolicy(CURL* easy); + +// Returns the first readable system CA bundle file on the current Linux host +// (probing standard Debian/Ubuntu, RHEL/Fedora, openSUSE, and Alpine paths), +// or nullptr if no bundle file is found or on non-Linux platforms. +const char* LinuxCaBundlePath(); + +// Returns FIREBASE_SSL_CAPATH ("/etc/ssl/certs") if it exists as a readable +// directory on the current Linux host, or nullptr otherwise. +const char* LinuxCaDirectoryPath(); + +} // namespace curl +} // namespace rest +} // namespace firebase + +#endif // FIREBASE_APP_REST_CURL_TLS_POLICY_H_ diff --git a/app/rest/transport_curl.cc b/app/rest/transport_curl.cc index 613bbaee28..0aaac9eb08 100644 --- a/app/rest/transport_curl.cc +++ b/app/rest/transport_curl.cc @@ -22,6 +22,8 @@ #include #include "app/rest/controller_curl.h" +#include "app/rest/curl/curl_handles.h" +#include "app/rest/curl/tls_policy.h" #include "app/rest/util.h" #include "app/src/assert.h" #include "app/src/include/firebase/internal/mutex.h" @@ -329,10 +331,7 @@ void InitTransportCurl() { MutexLock lock(*g_initialize_mutex); if (g_initialize_count == 0) { // Initialize curl. - CURLcode global_init_code = curl_global_init(CURL_GLOBAL_ALL); - FIREBASE_ASSERT_MESSAGE(global_init_code == CURLE_OK, - "curl global init failed with code %d", - global_init_code); + curl::AcquireCurlGlobal(); // Kick off background thread. assert(!g_curl_thread); @@ -351,7 +350,7 @@ void CleanupTransportCurl() { g_curl_thread = nullptr; // Clean up curl. - curl_global_cleanup(); + curl::ReleaseCurlGlobal(); } } @@ -457,14 +456,9 @@ bool BackgroundTransportCurl::PerformBackground(Request* request) { CheckOk(curl_easy_setopt(curl_, CURLOPT_PROTOCOLS_STR, "http,https"), "set valid protocols"); - // Verify SSL. - CheckOk(curl_easy_setopt(curl_, CURLOPT_SSL_VERIFYPEER, 1L), "verify peer"); - CheckOk(curl_easy_setopt(curl_, CURLOPT_SSL_VERIFYHOST, 2L), "verify host"); - -#ifdef FIREBASE_SSL_CAPATH - CheckOk(curl_easy_setopt(curl_, CURLOPT_CAPATH, FIREBASE_SSL_CAPATH), - "CA Path"); -#endif + // Peer/hostname verification, TLS 1.2 minimum, and OS trust store selection + // are centralized in app/rest/curl/tls_policy.h. + CheckOk(curl::ApplyTlsPolicy(curl_), "apply TLS policy"); // Set callback functions. CheckOk(curl_easy_setopt(curl_, CURLOPT_HEADERFUNCTION, CurlHeaderCallback), diff --git a/app/rest/util.cc b/app/rest/util.cc index 10bb146b57..1ee52e8ee2 100644 --- a/app/rest/util.cc +++ b/app/rest/util.cc @@ -21,6 +21,7 @@ #include #include +#include "app/rest/curl/curl_handles.h" #include "app/src/variant_util.h" #include "curl/curl.h" #include "flatbuffers/flatbuffers.h" @@ -50,14 +51,25 @@ static ::firebase::Mutex* g_util_curl_mutex = new Mutex(); CURL* g_curl_instance = nullptr; int g_init_ref_count = 0; +// Returns an opaque void* (matching TransportCurl::curl_ in transport_curl.h) +// so public/internal headers that include util.h do not need in +// their include path. void* CreateCurlPtr() { // NOLINT MutexLock lock(*g_util_curl_mutex); - return curl_easy_init(); + curl::AcquireCurlGlobal(); + CURL* curl = curl_easy_init(); + if (curl == nullptr) { + curl::ReleaseCurlGlobal(); + } + return curl; } void DestroyCurlPtr(void* curl_ptr) { MutexLock lock(*g_util_curl_mutex); - curl_easy_cleanup(static_cast(curl_ptr)); + if (curl_ptr != nullptr) { + curl_easy_cleanup(static_cast(curl_ptr)); + curl::ReleaseCurlGlobal(); + } } void Initialize() { diff --git a/app/rest/util.h b/app/rest/util.h index 32c543aa59..4e26b15f1d 100644 --- a/app/rest/util.h +++ b/app/rest/util.h @@ -64,10 +64,13 @@ void Initialize(); // exiting. Reference counted for convenience. void Terminate(); -// Returns a new CURL pointer. Mutex-locked, for thread safety. +// Returns a new CURL* easy handle as an opaque void* (so headers like util.h +// and transport_curl.h do not need to include ). Mutex-locked and +// paired with AcquireCurlGlobal() for thread safety. void* CreateCurlPtr(); -// Disposes of a CURL pointer. Mutex-locked, for thread safety. +// Disposes of a CURL* easy handle created by CreateCurlPtr() and releases its +// global libcurl reference. Mutex-locked, for thread safety. void DestroyCurlPtr(void* curl_ptr); // Trim the string's leading and trailing white space. diff --git a/app/src/logger.cc b/app/src/logger.cc index 7bbfb5791e..c7f6716529 100644 --- a/app/src/logger.cc +++ b/app/src/logger.cc @@ -95,7 +95,9 @@ LogLevel Logger::GetLogLevel() const { return log_level_; } void Logger::LogMessageImplV(LogLevel log_level, const char* format, va_list args) const { - parent_logger_->LogMessageV(log_level, format, args); + if (parent_logger_ != nullptr) { + parent_logger_->LogMessageV(log_level, format, args); + } } } // namespace firebase diff --git a/app/tests/logger_test.cc b/app/tests/logger_test.cc index 9b6e6855f4..926fc6aad6 100644 --- a/app/tests/logger_test.cc +++ b/app/tests/logger_test.cc @@ -63,6 +63,10 @@ TEST(LoggerTest, GetSetLogLevel) { EXPECT_EQ(logger2.GetLogLevel(), kLogLevelDebug); logger2.SetLogLevel(kLogLevelInfo); EXPECT_EQ(logger2.GetLogLevel(), kLogLevelInfo); + + // Logging at any level with a null parent logger must be a safe no-op. + logger.LogWarning("ignored warning"); + logger.LogError("ignored error"); } TEST(LoggerTest, LogWithEachFunction) { diff --git a/build_scripts/desktop/package.sh b/build_scripts/desktop/package.sh index a741931e6e..d58de21eb9 100755 --- a/build_scripts/desktop/package.sh +++ b/build_scripts/desktop/package.sh @@ -220,8 +220,6 @@ readonly deps_hidden_firebase_app=" */firestore-build/*/nanopb-build*/${prefix}*.${ext} " readonly deps_hidden_firebase_database=" -*/${subdir}${prefix}uv_a.${ext} -*/${subdir}${prefix}libuWS.${ext} " readonly deps_hidden_firebase_firestore=" */firestore-build/Firestore/*/${prefix}*.${ext} @@ -375,17 +373,33 @@ for product in ${product_list[*]}; do echo "rm -f ${deps[*]}" >> "${merge_libraries_tmp}/merge_${product}.sh" fi chmod u+x "${merge_libraries_tmp}/merge_${product}.sh" - if [[ ${run_in_parallel} -eq 0 ]]; then - # Run immediately if not set to run in parallel. - "${merge_libraries_tmp}/merge_${product}.sh" - else + if [[ ${run_in_parallel} -ne 0 ]]; then echo "echo \"${libfile_out}\" DONE" >> "${merge_libraries_tmp}/merge_${product}.sh" fi done -if [[ ${run_in_parallel} -ne 0 ]]; then - # Analytics is the smallest SDK, so it should be the shortest job. +if [[ ${run_in_parallel} -eq 0 ]]; then + # Run sequentially only after all merge_${product}.sh scripts are generated so + # `find . -path ${dep}` above resolves shared hidden deps (e.g. libcurl.a in + # deps_hidden_firebase_app) before merge_app.sh deletes them under -D. The + # first script to run populates ${cache_file} from --scan_libs="${allfiles}". + for product in ${product_list[*]}; do + if [[ -x "${merge_libraries_tmp}/merge_${product}.sh" ]]; then + "${merge_libraries_tmp}/merge_${product}.sh" + fi + done +else + # Analytics is the smallest SDK, so it should be the shortest job; fall back + # to the first built product if analytics was excluded from the CMake build. shortest=analytics + if [[ ! -x "${merge_libraries_tmp}/merge_${shortest}.sh" ]]; then + for product in ${product_list[*]}; do + if [[ -x "${merge_libraries_tmp}/merge_${product}.sh" ]]; then + shortest="${product}" + break + fi + done + fi echo "There are ${#product_list[@]} jobs to run." echo "Running shortest job to populate cache, then remaining jobs in parallel..." "${merge_libraries_tmp}/merge_${shortest}.sh" diff --git a/cmake/CTestCustom.cmake b/cmake/CTestCustom.cmake index c233161fa6..5946957f71 100644 --- a/cmake/CTestCustom.cmake +++ b/cmake/CTestCustom.cmake @@ -15,9 +15,6 @@ # Define the tests that need to be ignored. set(CTEST_CUSTOM_TESTS_IGNORE ${CTEST_CUSTOM_TESTS_IGNORE} - # Tests from libuv, that can't be disabled normally. - uv_test - uv_test_a # Tests from zlib, that can't be disabled normally. example example64 diff --git a/cmake/external/CMakeLists.txt b/cmake/external/CMakeLists.txt index 2ca2da355c..3cfdd8e830 100644 --- a/cmake/external/CMakeLists.txt +++ b/cmake/external/CMakeLists.txt @@ -30,9 +30,7 @@ include(firestore) if (${FIREBASE_EXTERNAL_PLATFORM} STREQUAL "DESKTOP") include(boringssl) include(curl) - include(libuv) include(leveldb) - include(uWebSockets) include(zlib) endif() diff --git a/cmake/external/libuv.cmake b/cmake/external/libuv.cmake deleted file mode 100644 index 162358ad1e..0000000000 --- a/cmake/external/libuv.cmake +++ /dev/null @@ -1,37 +0,0 @@ -# Copyright 2019 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -include(ExternalProject) - -if(TARGET libuv OR NOT DOWNLOAD_LIBUV) - return() -endif() - -set(version v1.33.1) - -ExternalProject_Add( - libuv - - DOWNLOAD_DIR ${FIREBASE_DOWNLOAD_DIR} - DOWNLOAD_NAME libuv-${version}.tar.gz - URL https://github.com/libuv/libuv/archive/${version}.tar.gz - - PREFIX ${PROJECT_BINARY_DIR} - - CONFIGURE_COMMAND "" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" - HTTP_HEADER "${EXTERNAL_PROJECT_HTTP_HEADER}" -) diff --git a/cmake/external/uWebSockets.cmake b/cmake/external/uWebSockets.cmake deleted file mode 100644 index 0efd51d1eb..0000000000 --- a/cmake/external/uWebSockets.cmake +++ /dev/null @@ -1,39 +0,0 @@ -# Copyright 2019 Google LLC -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -include(ExternalProject) - -if(TARGET uWebSockets OR NOT DOWNLOAD_UWEBSOCKETS) - return() -endif() - -set(uwebsockets_commit_tag 4d94401b9c98346f9afd838556fdc7dce30561eb) -set(patch_file - ${CMAKE_CURRENT_LIST_DIR}/../../scripts/git/patches/uWebSockets/0001-fix-want-write-and-sprintf-deprecation.patch) - -ExternalProject_Add( - uWebSockets - - DOWNLOAD_COMMAND - COMMAND git init uWebSockets - COMMAND cd uWebSockets && git fetch --depth=1 https://github.com/uNetworking/uWebSockets.git ${uwebsockets_commit_tag} && git reset --hard FETCH_HEAD - - PATCH_COMMAND git apply ${patch_file} && git gc --aggressive - PREFIX ${PROJECT_BINARY_DIR} - CONFIGURE_COMMAND "" - BUILD_COMMAND "" - INSTALL_COMMAND "" - TEST_COMMAND "" - HTTP_HEADER "${EXTERNAL_PROJECT_HTTP_HEADER}" -) diff --git a/cmake/external_rules.cmake b/cmake/external_rules.cmake index 4a3d846883..813f9a779c 100644 --- a/cmake/external_rules.cmake +++ b/cmake/external_rules.cmake @@ -49,8 +49,6 @@ function(download_external_sources) check_use_local_directory(BORINGSSL) check_use_local_directory(CURL) check_use_local_directory(FLATBUFFERS) - check_use_local_directory(LIBUV) - check_use_local_directory(UWEBSOCKETS) check_use_local_directory(ZLIB) check_use_local_directory(FIREBASE_IOS_SDK) @@ -83,8 +81,6 @@ function(download_external_sources) -DDOWNLOAD_CURL=${DOWNLOAD_CURL} -DDOWNLOAD_FLATBUFFERS=${DOWNLOAD_FLATBUFFERS} -DDOWNLOAD_GOOGLETEST=${FIREBASE_DOWNLOAD_GTEST} - -DDOWNLOAD_LIBUV=${DOWNLOAD_LIBUV} - -DDOWNLOAD_UWEBSOCKETS=${DOWNLOAD_UWEBSOCKETS} -DDOWNLOAD_ZLIB=${DOWNLOAD_ZLIB} -DDOWNLOAD_FIREBASE_IOS_SDK=${DOWNLOAD_FIREBASE_IOS_SDK} -DEXTERNAL_PROJECT_HTTP_HEADER=${EXTERNAL_PROJECT_HTTP_HEADER} diff --git a/database/CMakeLists.txt b/database/CMakeLists.txt index 1df98b1f3e..a6b07b3ddf 100644 --- a/database/CMakeLists.txt +++ b/database/CMakeLists.txt @@ -165,15 +165,14 @@ if(ANDROID OR IOS) set(additional_DEFINES) else() set(additional_include_DIR + ${CURL_SOURCE_DIR}/include ${FLATBUFFERS_SOURCE_DIR}/include - ${OPENSSL_INCLUDE_DIR} - ${UWEBSOCKETS_SOURCE_DIR}/..) + ${OPENSSL_INCLUDE_DIR}) set(additional_link_LIB firebase_rest_lib leveldb - libuWS - OpenSSL::SSL + libcurl OpenSSL::Crypto) set(additional_DEFINES @@ -186,6 +185,7 @@ if(MSVC) -DWIN32_LEAN_AND_MEAN # Ensure that windows doesn't include winsock.h by # default, as it can cause issues when libraries try # to include winsock2.h later on in the process. + -DCURL_STATICLIB=1 ) endif() @@ -246,7 +246,3 @@ endif() cpp_pack_library(firebase_database "") cpp_pack_public_headers() -if (NOT ANDROID AND NOT IOS) - cpp_pack_library(uv_a "deps/database/external") - cpp_pack_library(libuWS "deps/database/external") -endif() diff --git a/database/integration_test/CMakeLists.txt b/database/integration_test/CMakeLists.txt index 09fc404bec..a21079377d 100644 --- a/database/integration_test/CMakeLists.txt +++ b/database/integration_test/CMakeLists.txt @@ -211,7 +211,7 @@ else() "-framework SystemConfiguration" ) elseif(MSVC) - set(ADDITIONAL_LIBS advapi32 ws2_32 crypt32 bcrypt iphlpapi secur32 psapi userenv shell32) + set(ADDITIONAL_LIBS advapi32 ws2_32 crypt32 bcrypt iphlpapi secur32 shell32) else() set(ADDITIONAL_LIBS pthread) endif() diff --git a/database/src/desktop/connection/connection.cc b/database/src/desktop/connection/connection.cc index 21e90f680c..6608f20df8 100644 --- a/database/src/desktop/connection/connection.cc +++ b/database/src/desktop/connection/connection.cc @@ -229,20 +229,14 @@ void Connection::OnError(const WebSocketClientErrorData& error_data) { logger_->LogDebug("%s websocket error occurred. Uri: %s", log_id_.c_str(), error_data.GetUri().c_str()); - scheduler_->Schedule(new callback::CallbackValue1( - safe_this_, [](ConnectionRef conn_ref) { - ConnectionRefLock lock(&conn_ref); - auto connection = lock.GetReference(); - if (connection != nullptr) { - // If error occurs before the connection is opened, it is due to - // connection failed (ex. incorrect url). Otherwise, it can be - // any lower-level error during connection. - DisconnectReason reason = connection->ws_connected_ - ? kDisconnectReasonWebsocketError - : kDisconnectReasonConnectionFailed; - connection->Close(reason); - } - })); + if (state_ != kStateDisconnected) { + // If error occurs before the connection is opened, it is due to + // connection failed (ex. incorrect url). Otherwise, it can be + // any lower-level error during connection. + DisconnectReason reason = ws_connected_ ? kDisconnectReasonWebsocketError + : kDisconnectReasonConnectionFailed; + Close(reason); + } } void Connection::HandleIncomingFrame(const char* msg) { diff --git a/database/src/desktop/connection/util_connection.cc b/database/src/desktop/connection/util_connection.cc index dbac9b53dd..f8a39b5bb4 100644 --- a/database/src/desktop/connection/util_connection.cc +++ b/database/src/desktop/connection/util_connection.cc @@ -27,7 +27,6 @@ std::unique_ptr CreateWebSocketClient( const HostInfo& info, WebSocketClientEventHandler* delegate, const char* opt_last_session_id, Logger* logger, scheduler::Scheduler* scheduler, const std::string& app_check_token) { - // Currently we use uWebSockets implementation. std::string uri = info.GetConnectionUrl(opt_last_session_id); return std::make_unique( uri, info.user_agent(), logger, scheduler, app_check_token, delegate); diff --git a/database/src/desktop/connection/web_socket_client_impl.cc b/database/src/desktop/connection/web_socket_client_impl.cc index 2195c0c03a..46ad72f02c 100644 --- a/database/src/desktop/connection/web_socket_client_impl.cc +++ b/database/src/desktop/connection/web_socket_client_impl.cc @@ -14,288 +14,634 @@ #include "database/src/desktop/connection/web_socket_client_impl.h" +#include #include -#include - +#include +#include +#include +#include +#include + +#include "app/rest/curl/curl_handles.h" +#include "app/rest/curl/tls_policy.h" #include "app/src/app_common.h" #include "app/src/assert.h" #include "app/src/include/firebase/app.h" #include "app/src/include/firebase/internal/mutex.h" -#include "app/src/log.h" -#include "app/src/thread.h" +#include "curl/curl.h" namespace firebase { namespace database { namespace internal { namespace connection { +using ::firebase::rest::curl::AcquireCurlGlobal; +using ::firebase::rest::curl::ReleaseCurlGlobal; +using ::firebase::rest::curl::UniqueEasy; + +namespace { +// Size of the stack buffer used for each curl_ws_recv() read pass. +constexpr size_t kRecvBufferSize = 16384; +// RFC 6455 section 5.5 caps all WebSocket control frame payloads at 125 bytes +// and forbids fragmenting control frames. +constexpr size_t kMaxPingPayloadBytes = 125; +// Upper bound on a single reassembled inbound WebSocket message (256 MiB, +// matching Realtime Database's single-read payload limit) so a misbehaving peer +// streaming CURLWS_CONT fragments cannot grow memory without bound. +constexpr size_t kMaxMessagePayloadBytes = 256 * 1024 * 1024; +// WebSocket close status code 1000 (Normal Closure, RFC 6455 section 7.4.1) +// encoded in network byte order (big-endian) for CURLWS_CLOSE frames. +// libcurl's defines frame flag bits (CURLWS_CLOSE) but +// does not define RFC 6455 status code constants. +constexpr uint16_t kCloseStatusNormal = 1000; +const unsigned char kCloseNormalPayload[2] = { + static_cast((kCloseStatusNormal >> 8) & 0xFF), + static_cast(kCloseStatusNormal & 0xFF), +}; +// Upper bound on how long the I/O thread sleeps in curl_multi_poll() when +// nothing happens. Post() wakes it immediately via curl_multi_wakeup(). +constexpr int kPollTimeoutMs = 1000; +} // namespace + WebSocketClientImpl::WebSocketClientImpl( const std::string& uri, const std::string& user_agent, Logger* logger, scheduler::Scheduler* scheduler, const std::string& app_check_token, WebSocketClientEventHandler* handler /*=nullptr*/) : uri_(uri), handler_(handler), - thread_(nullptr), - hub_(), - keep_loop_alive_(nullptr), - process_queue_async_(nullptr), - callback_queue_(), - callback_queue_mutex_(Mutex::kModeNonRecursive), - is_destructing_(0), - websocket_(nullptr), + multi_(nullptr), + post_mutex_(Mutex::kModeNonRecursive), + posted_(), + quit_(false), + easy_(nullptr), + request_headers_(nullptr), + socket_(CURL_SOCKET_BAD), + state_(kStateIdle), + handshake_deadline_(std::chrono::steady_clock::time_point::max()), + incoming_message_buffer_(), + incoming_ping_buffer_(), + send_queue_(), + frame_in_progress_(false), + current_send_offset_(0), user_agent_(user_agent), + app_check_token_mutex_(Mutex::kModeNonRecursive), + app_check_token_(app_check_token), logger_(logger), scheduler_(scheduler), safe_this_(this), - app_check_token_(app_check_token) { - // Bind callback function - hub_.onError(WebSocketClientImpl::OnError); - hub_.onConnection(WebSocketClientImpl::OnConnection); - hub_.onMessage(WebSocketClientImpl::OnMessage); - hub_.onDisconnection(WebSocketClientImpl::OnDisconnection); - - // Create a async object to keep the loop alive and close all async handler - // during destruction. - keep_loop_alive_ = new uS::Async(hub_.getLoop()); - keep_loop_alive_->setData(this); - keep_loop_alive_->start([](uS::Async* async) { - assert(async); - assert(async->getData()); - - WebSocketClientImpl* client = - static_cast(async->getData()); - - // Close all async process - if (client) { - client->CloseSync(); - - client->keep_loop_alive_->close(); - client->keep_loop_alive_ = nullptr; - client->process_queue_async_->close(); - client->process_queue_async_ = nullptr; - } - }); - - // Initiate async handler to process callback queue. The callback will only - // be triggered after process_queue_async_->send(). - process_queue_async_ = new uS::Async(hub_.getLoop()); - process_queue_async_->setData(this); - process_queue_async_->start(ProcessCallbackQueue); - - // Start the event loop - thread_ = std::make_unique(EventLoopRoutine, this); -} - -void WebSocketClientImpl::EventLoopRoutine(void* data) { - assert(data != nullptr); - WebSocketClientImpl* client = static_cast(data); - Logger* logger = client->logger_; - - logger->LogDebug("=== uWebSockets Event Loop Start ==="); - client->hub_.run(); - logger->LogDebug("=== uWebSockets Event Loop End ==="); + thread_() { + AcquireCurlGlobal(); + multi_.reset(curl_multi_init()); + FIREBASE_ASSERT_MESSAGE(multi_ != nullptr, "curl_multi_init() failed"); + // Started last: everything RunEventLoop() touches is initialized above. + thread_ = Thread(EventLoopRoutine, this); } WebSocketClientImpl::~WebSocketClientImpl() { - // Clear safe reference immediately so that scheduled callback can skip - // executing code which requires reference to this. + // Clear safe reference immediately so that scheduled callbacks can skip + // executing code which requires a reference to this. safe_this_.ClearReference(); - is_destructing_.store(1); + // Release the socket and the easy handle on the I/O thread, then stop it. + // Nothing touches this object from the I/O thread once Join() returns. + Post([this]() { + TeardownOnLoop(); + quit_ = true; + }); + thread_.Join(); - // Remove the handler to keep event loop alive - if (keep_loop_alive_ != nullptr) { - keep_loop_alive_->send(); - } + handler_ = nullptr; + assert(easy_ == nullptr); + assert(socket_ == CURL_SOCKET_BAD); - // Wait for the thread to end. - if (thread_) { - thread_->Join(); - thread_.reset(nullptr); - } + multi_.reset(); + ReleaseCurlGlobal(); +} - handler_ = nullptr; +// --------------------------------------------------------------------------- +// Cross-thread entry points. +// --------------------------------------------------------------------------- - // websocket_ should be cleared now or OnDisconnection() probably is not - // called properly. - assert(websocket_ == nullptr); +void WebSocketClientImpl::Post(std::function fn) { + MutexLock lock(post_mutex_); + posted_.push_back(std::move(fn)); + // curl_multi_wakeup() is thread-safe; multi_ outlives the I/O thread. + curl_multi_wakeup(multi_.get()); } void WebSocketClientImpl::Connect(int timeout_ms) { - ScheduleOnce( - [](WebSocketClientImpl* client, int timeout_ms, const std::string&) { - Logger* logger = client->logger_; - if (client->websocket_ == nullptr) { - std::map headers; - headers["User-Agent"] = client->user_agent_; - headers[app_common::kApiClientHeader] = App::GetUserAgent(); - if (!client->app_check_token_.empty()) { - headers["X-Firebase-AppCheck"] = client->app_check_token_; - } - client->hub_.connect(client->uri_, client, headers, timeout_ms); - } else { - logger->LogWarning("websocket has already been connected to %s", - client->uri_.c_str()); - } - }, - timeout_ms, ""); + Post([this, timeout_ms]() { ConnectOnLoop(timeout_ms); }); } void WebSocketClientImpl::Close() { - ScheduleOnce([](WebSocketClientImpl* client, int, - const std::string&) { client->CloseSync(); }, - 0, ""); -} - -void WebSocketClientImpl::CloseSync() { - if (IsWebSocketAvailable()) { - websocket_->close(1000); - } + Post([this]() { CloseOnLoop(); }); } void WebSocketClientImpl::Send(const char* msg) { assert(msg != nullptr); - - ScheduleOnce( - [](WebSocketClientImpl* client, int, const std::string& msg) { - Logger* logger = client->logger_; - if (client->IsWebSocketAvailable()) { - client->websocket_->send(msg.c_str()); - } else { - logger->LogWarning( - "Cannot send message. websocket is not available"); - } - }, - 0, msg); + std::string message(msg); + Post([this, message = std::move(message)]() { SendOnLoop(message); }); } void WebSocketClientImpl::RefreshAppCheckToken(const std::string& token) { + MutexLock lock(app_check_token_mutex_); app_check_token_ = token; } -void WebSocketClientImpl::OnError(void* data) { - assert(data != nullptr); - WebSocketClientImpl* client = static_cast(data); - Logger* logger = client->logger_; +// --------------------------------------------------------------------------- +// The I/O thread. +// --------------------------------------------------------------------------- + +void WebSocketClientImpl::EventLoopRoutine(void* client) { + static_cast(client)->RunEventLoop(); +} - if (client->handler_) { - // TODO(b/71873743): Modify uWebSockets to provide more context, ex. reasons - WebSocketClientErrorData error(client->uri_.c_str()); - client->handler_->OnError(error); +void WebSocketClientImpl::RunPosted() { + std::deque> batch; + { + MutexLock lock(post_mutex_); + batch.swap(posted_); } + for (std::function& fn : batch) fn(); +} + +void WebSocketClientImpl::RunEventLoop() { + while (!quit_) { + RunPosted(); + if (quit_) break; + + // Drive the handshake (a no-op once the connect-only handle is done). + int running_handles = 0; + curl_multi_perform(multi_.get(), &running_handles); + int messages_left = 0; + CURLMsg* message = nullptr; + while ((message = curl_multi_info_read(multi_.get(), &messages_left)) != + nullptr) { + if (message->msg == CURLMSG_DONE && message->easy_handle == easy_.get()) { + OnHandshakeDone(message->data.result); + } + } - logger->LogDebug("Error occurred while establishing connection to %s", - client->uri_.c_str()); + // libcurl bounds only DNS/TCP/TLS for a connect-only handle, not the HTTP + // Upgrade exchange, so enforce Connect()'s timeout here. + int timeout_ms = kPollTimeoutMs; + if (state_ == kStateConnecting && + handshake_deadline_ != std::chrono::steady_clock::time_point::max()) { + const auto remaining_ms = + std::chrono::duration_cast( + handshake_deadline_ - std::chrono::steady_clock::now()) + .count(); + if (remaining_ms <= 0) { + logger_->LogDebug("WebSocket handshake to %s timed out", uri_.c_str()); + CleanupEasyHandle(); + state_ = kStateClosed; + OnError(); + continue; + } + timeout_ms = static_cast( + std::min(timeout_ms, static_cast(remaining_ms))); + } + + // Sleep until the handshake makes progress, the open socket is ready, + // Post() wakes us, or the timeout elapses. + curl_waitfd wait_fd; + wait_fd.fd = socket_; + wait_fd.events = CURL_WAIT_POLLIN; + if (!send_queue_.empty()) wait_fd.events |= CURL_WAIT_POLLOUT; + wait_fd.revents = 0; + const bool watch_socket = + state_ == kStateOpen && socket_ != CURL_SOCKET_BAD; + int num_fds = 0; + curl_multi_poll(multi_.get(), watch_socket ? &wait_fd : nullptr, + watch_socket ? 1 : 0, timeout_ms, &num_fds); + if (watch_socket && wait_fd.revents != 0) Pump(wait_fd.revents); + } } -void WebSocketClientImpl::OnConnection(ClientWebSocket* ws, - uWS::HttpRequest req) { - assert(ws != nullptr); - assert(ws->getUserData() != nullptr); - WebSocketClientImpl* client = - static_cast(ws->getUserData()); - - // There should be only one connection per client. However, hub_ can have - // multiple connnections at a time, and current implementation does not - // prevent Connect() from being called when another connection is - // establishing. Use assert for now to prevent this from happening. - assert(client->websocket_ == nullptr); - client->websocket_ = ws; - - if (client->handler_) { - client->scheduler_->Schedule(new callback::CallbackValue1( - client->safe_this_, [](ClientRef client_ref) { - ClientRefLock lock(&client_ref); - auto client = lock.GetReference(); - if (client != nullptr && client->handler_ != nullptr) { - client->handler_->OnOpen(); - } - })); +// --------------------------------------------------------------------------- +// Handshake. +// --------------------------------------------------------------------------- + +void WebSocketClientImpl::ConnectOnLoop(int timeout_ms) { + if (state_ == kStateConnecting || state_ == kStateOpen) { + logger_->LogWarning("websocket has already been connected to %s", + uri_.c_str()); + return; + } + + CleanupEasyHandle(); + + UniqueEasy easy(curl_easy_init()); + if (easy == nullptr) { + logger_->LogError("Failed to initialize curl easy handle for %s", + uri_.c_str()); + OnError(); + return; + } + + curl_easy_setopt(easy.get(), CURLOPT_URL, uri_.c_str()); + // Restrict schemes to ws/wss so system-libcurl builds that compile in extra + // protocols (file://, http://, ftp://, ...) cannot open non-WebSocket URIs. + CURLcode proto_result = + curl_easy_setopt(easy.get(), CURLOPT_PROTOCOLS_STR, "ws,wss"); + if (proto_result != CURLE_OK) { + logger_->LogError("Failed to restrict WebSocket protocols for %s: (%d) %s", + uri_.c_str(), static_cast(proto_result), + curl_easy_strerror(proto_result)); + OnError(); + return; + } + // CURLOPT_CONNECT_ONLY = 2L instructs libcurl to perform the HTTP/1.1 + // WebSocket Upgrade handshake and stop after receiving 101 Switching + // Protocols so framing can be driven via curl_ws_recv() and curl_ws_send(). + curl_easy_setopt(easy.get(), CURLOPT_CONNECT_ONLY, 2L); + // Never let libcurl touch process-wide signal handlers from the I/O thread. + curl_easy_setopt(easy.get(), CURLOPT_NOSIGNAL, 1L); + // Handle PING frames explicitly in ReceiveFrames() so standalone PINGs flush + // their PONG response immediately without waiting for another frame. + curl_easy_setopt(easy.get(), CURLOPT_WS_OPTIONS, + static_cast(CURLWS_NOAUTOPONG)); // NOLINT + // Peer/hostname verification, TLS 1.2 minimum, and OS trust store selection + // are centralized in app/rest/curl/tls_policy.h so REST and WebSocket + // connections can never drift apart. + CURLcode tls_result = firebase::rest::curl::ApplyTlsPolicy(easy.get()); + if (tls_result != CURLE_OK) { + logger_->LogError("Failed to apply TLS policy for %s: (%d) %s", + uri_.c_str(), static_cast(tls_result), + curl_easy_strerror(tls_result)); + OnError(); + return; + } + if (timeout_ms > 0) { + // Bounds DNS + TCP + TLS; the HTTP Upgrade exchange is bounded by + // handshake_deadline_ in RunEventLoop(). + curl_easy_setopt(easy.get(), CURLOPT_CONNECTTIMEOUT_MS, + static_cast(timeout_ms)); // NOLINT + } + + std::string app_check_token; + { + MutexLock lock(app_check_token_mutex_); + app_check_token = app_check_token_; + } + + auto append_header = [this](const char* header) { + curl_slist* next = curl_slist_append(request_headers_.get(), header); + if (next == nullptr) return false; + request_headers_.release(); + request_headers_.reset(next); + return true; + }; + std::string user_agent_header = "User-Agent: " + user_agent_; + if (!append_header(user_agent_header.c_str())) { + OnError(); + return; + } + + const char* api_client = App::GetUserAgent(); + if (api_client != nullptr && api_client[0] != '\0') { + std::string api_client_header = + std::string(app_common::kApiClientHeader) + ": " + api_client; + if (!append_header(api_client_header.c_str())) { + OnError(); + return; + } + } + + if (!app_check_token.empty()) { + std::string app_check_header = "X-Firebase-AppCheck: " + app_check_token; + if (!append_header(app_check_header.c_str())) { + OnError(); + return; + } } - // Just in case of the client is delete right after connection request is sent - // This may have racing condition with the destructor. But since the - // destructor schedule another task, which runs on the same thread to this - // call, in order to close the websocket, it should not be a big issue. - if (client->is_destructing_.load() > 0) { - client->CloseSync(); + curl_easy_setopt(easy.get(), CURLOPT_HTTPHEADER, request_headers_.get()); + + easy_ = std::move(easy); + incoming_message_buffer_.clear(); + incoming_ping_buffer_.clear(); + send_queue_.clear(); + frame_in_progress_ = false; + current_send_offset_ = 0; + + CURLMcode mc = curl_multi_add_handle(multi_.get(), easy_.get()); + if (mc != CURLM_OK) { + logger_->LogError("curl_multi_add_handle failed: %s", + curl_multi_strerror(mc)); + easy_.reset(); + request_headers_.reset(); + OnError(); + return; } + state_ = kStateConnecting; + handshake_deadline_ = timeout_ms > 0 + ? std::chrono::steady_clock::now() + + std::chrono::milliseconds(timeout_ms) + : std::chrono::steady_clock::time_point::max(); } -void WebSocketClientImpl::OnMessage(ClientWebSocket* ws, char* message, - size_t length, uWS::OpCode opCode) { - assert(ws != nullptr); - assert(ws->getUserData() != nullptr); - WebSocketClientImpl* client = - static_cast(ws->getUserData()); - - if (client->handler_) { - std::string message_string(message, length); - client->scheduler_->Schedule( - new callback::CallbackValue2( - client->safe_this_, message_string, - [](ClientRef client_ref, std::string msg) { - ClientRefLock lock(&client_ref); - auto client = lock.GetReference(); - if (client != nullptr && client->handler_ != nullptr) { - client->handler_->OnMessage(msg.c_str()); - } - })); +void WebSocketClientImpl::OnHandshakeDone(CURLcode result) { + // The handle stays attached to multi_: libcurl closes a connect-only + // connection the moment its handle is removed. + if (result != CURLE_OK) { + logger_->LogDebug( + "Error occurred while establishing connection to %s: (%d) %s", + uri_.c_str(), static_cast(result), curl_easy_strerror(result)); + CleanupEasyHandle(); + state_ = kStateClosed; + OnError(); + return; } + + curl_socket_t sockfd = CURL_SOCKET_BAD; + curl_easy_getinfo(easy_.get(), CURLINFO_ACTIVESOCKET, &sockfd); + if (sockfd == CURL_SOCKET_BAD) { + logger_->LogError("No active socket after WebSocket upgrade to %s", + uri_.c_str()); + CleanupEasyHandle(); + state_ = kStateClosed; + OnError(); + return; + } + + socket_ = sockfd; + state_ = kStateOpen; + OnConnection(); + // Drain any WebSocket frames that arrived in the same packet or TLS record + // as the HTTP 101 Switching Protocols response, and flush anything queued. + Pump(CURL_WAIT_POLLIN | CURL_WAIT_POLLOUT); } -void WebSocketClientImpl::OnDisconnection(ClientWebSocket* ws, int code, - char* message, size_t length) { - assert(ws != nullptr); - assert(ws->getUserData() != nullptr); - WebSocketClientImpl* client = - static_cast(ws->getUserData()); - - assert(client->websocket_ != nullptr); - client->websocket_ = nullptr; - - if (client->handler_) { - client->scheduler_->Schedule(new callback::CallbackValue1( - client->safe_this_, [](ClientRef client_ref) { - ClientRefLock lock(&client_ref); - auto client = lock.GetReference(); - if (client != nullptr && client->handler_ != nullptr) { - client->handler_->OnClose(); - } - })); +// --------------------------------------------------------------------------- +// I/O-thread operations. +// --------------------------------------------------------------------------- + +void WebSocketClientImpl::CloseOnLoop() { + if (state_ == kStateOpen && easy_ != nullptr) { + if (!frame_in_progress_) { + size_t sent = 0; + (void)curl_ws_send(easy_.get(), kCloseNormalPayload, + sizeof(kCloseNormalPayload), &sent, 0, CURLWS_CLOSE); + } + CleanupEasyHandle(); + state_ = kStateClosed; + OnDisconnection(); + } else if (state_ == kStateConnecting && easy_ != nullptr) { + CleanupEasyHandle(); + state_ = kStateClosed; } } -void WebSocketClientImpl::ScheduleOnce(Callback cb, int int_value, - const std::string& string_value) { - assert(cb != nullptr); +void WebSocketClientImpl::SendOnLoop(const std::string& msg) { + if (IsWebSocketAvailable()) { + send_queue_.push_back(OutboundFrame{msg, CURLWS_TEXT}); + // If a frame is already partially written the socket is backpressured; + // RunEventLoop() polls for POLLOUT while send_queue_ is non-empty. + if (!frame_in_progress_) FlushSendQueue(); + } else { + logger_->LogWarning("Cannot send message. websocket is not available"); + } +} - MutexLock lock(callback_queue_mutex_); - callback_queue_.push(CallbackData(cb, this, int_value, string_value)); +void WebSocketClientImpl::TeardownOnLoop() { + if (state_ == kStateOpen && easy_ != nullptr && !frame_in_progress_) { + size_t sent = 0; + (void)curl_ws_send(easy_.get(), kCloseNormalPayload, + sizeof(kCloseNormalPayload), &sent, 0, CURLWS_CLOSE); + } + CleanupEasyHandle(); + state_ = kStateClosed; +} - // Signal the event loop to trigger the async callback. - process_queue_async_->send(); +void WebSocketClientImpl::Pump(int revents) { + if (state_ != kStateOpen) return; + // Only (re)enter curl_ws_send() when the socket signalled writability or + // nothing is mid-frame: a POLLIN wakeup (server PING, data) while a frame is + // backpressured has nothing to flush yet. + if (!frame_in_progress_ || (revents & CURL_WAIT_POLLOUT) != 0) { + FlushSendQueue(); + } + // Receiving is always attempted: curl_ws_recv() is non-blocking and TLS may + // hold already-decrypted frames that the socket no longer reports as + // readable. + if (state_ == kStateOpen) ReceiveFrames(); } -void WebSocketClientImpl::ProcessCallbackQueue(uS::Async* async) { - assert(async); - assert(async->getData()); +void WebSocketClientImpl::FlushSendQueue() { + if (!IsWebSocketAvailable()) { + return; + } - WebSocketClientImpl* client = - static_cast(async->getData()); + CURL* easy = easy_.get(); + while (!send_queue_.empty() && state_ == kStateOpen) { + const OutboundFrame& frame = send_queue_.front(); + const char* data = frame.payload.data() + current_send_offset_; + size_t remaining = frame.payload.size() - current_send_offset_; + size_t sent = 0; + + frame_in_progress_ = true; + CURLcode res = curl_ws_send(easy, data, remaining, &sent, 0, frame.flags); + if (res == CURLE_OK || res == CURLE_AGAIN) { + current_send_offset_ += sent; + if (current_send_offset_ >= frame.payload.size() && res == CURLE_OK) { + send_queue_.pop_front(); + current_send_offset_ = 0; + frame_in_progress_ = false; + } else { + // Socket write buffer is full; wait for CURL_WAIT_POLLOUT. + break; + } + } else { + logger_->LogError("curl_ws_send failed for %s: (%d) %s", uri_.c_str(), + static_cast(res), curl_easy_strerror(res)); + CleanupEasyHandle(); + state_ = kStateClosed; + OnError(); + OnDisconnection(); + break; + } + } +} - MutexLock lock(client->callback_queue_mutex_); - while (!client->callback_queue_.empty()) { - auto& callback_data = client->callback_queue_.front(); - callback_data.callback(callback_data.client, callback_data.int_value, - callback_data.string_value); - client->callback_queue_.pop(); +void WebSocketClientImpl::ReceiveFrames() { + if (!IsWebSocketAvailable()) { + return; } + + CURL* easy = easy_.get(); + char buffer[kRecvBufferSize]; + + while (state_ == kStateOpen) { + size_t rlen = 0; + const struct curl_ws_frame* meta = nullptr; + CURLcode res = curl_ws_recv(easy, buffer, sizeof(buffer), &rlen, &meta); + + if (res == CURLE_AGAIN) { + break; + } + if (res == CURLE_GOT_NOTHING) { + CleanupEasyHandle(); + state_ = kStateClosed; + OnDisconnection(); + break; + } + if (res != CURLE_OK) { + logger_->LogDebug("curl_ws_recv error on %s: (%d) %s", uri_.c_str(), + static_cast(res), curl_easy_strerror(res)); + CleanupEasyHandle(); + state_ = kStateClosed; + OnError(); + OnDisconnection(); + break; + } + + if (meta == nullptr) { + break; + } + + if (meta->flags & CURLWS_CLOSE) { + if (!frame_in_progress_) { + size_t sent = 0; + (void)curl_ws_send(easy, "", 0, &sent, 0, CURLWS_CLOSE); + } + CleanupEasyHandle(); + state_ = kStateClosed; + OnDisconnection(); + break; + } + + const uint64_t frame_remaining = + meta->bytesleft > 0 ? static_cast(meta->bytesleft) : 0; + + if (meta->flags & CURLWS_PING) { + if ((meta->flags & CURLWS_CONT) != 0 || + rlen > kMaxPingPayloadBytes - incoming_ping_buffer_.size() || + frame_remaining > + kMaxPingPayloadBytes - (incoming_ping_buffer_.size() + rlen)) { + logger_->LogError( + "Invalid or oversized WebSocket PING frame from %s; closing", + uri_.c_str()); + CleanupEasyHandle(); + state_ = kStateClosed; + OnError(); + OnDisconnection(); + break; + } + if (rlen > 0) { + incoming_ping_buffer_.append(buffer, rlen); + } + if (meta->bytesleft == 0) { + OutboundFrame pong{std::move(incoming_ping_buffer_), CURLWS_PONG}; + incoming_ping_buffer_.clear(); + if (frame_in_progress_ && !send_queue_.empty()) { + // A frame is already partially buffered/sent in libcurl's sendbuf; + // queue the PONG immediately behind it (coalescing if a PONG is + // already waiting, per RFC 6455 section 5.5.3) so we never splice a + // control frame mid-frame. + if (send_queue_.size() > 1 && send_queue_[1].flags == CURLWS_PONG) { + send_queue_[1].payload = std::move(pong.payload); + } else { + send_queue_.insert(send_queue_.begin() + 1, std::move(pong)); + } + } else { + send_queue_.push_front(std::move(pong)); + FlushSendQueue(); + } + } + continue; + } + + if (meta->flags & CURLWS_PONG) { + continue; + } + + if (meta->flags & (CURLWS_TEXT | CURLWS_BINARY | CURLWS_CONT)) { + if (rlen > kMaxMessagePayloadBytes - incoming_message_buffer_.size() || + frame_remaining > kMaxMessagePayloadBytes - + (incoming_message_buffer_.size() + rlen)) { + logger_->LogError( + "Inbound WebSocket message from %s exceeds %zu-byte limit; closing", + uri_.c_str(), kMaxMessagePayloadBytes); + CleanupEasyHandle(); + state_ = kStateClosed; + OnError(); + OnDisconnection(); + break; + } + if (rlen > 0) { + incoming_message_buffer_.append(buffer, rlen); + } + if (meta->bytesleft == 0 && (meta->flags & CURLWS_CONT) == 0) { + std::string completed_message; + completed_message.swap(incoming_message_buffer_); + OnMessage(completed_message.data(), completed_message.size()); + } + } + } +} + +void WebSocketClientImpl::CleanupEasyHandle() { + socket_ = CURL_SOCKET_BAD; + if (easy_ != nullptr) { + // Detaching aborts an in-flight handshake or closes the open connection. + curl_multi_remove_handle(multi_.get(), easy_.get()); + easy_.reset(); + } + request_headers_.reset(); + incoming_message_buffer_.clear(); + incoming_ping_buffer_.clear(); + send_queue_.clear(); + frame_in_progress_ = false; + current_send_offset_ = 0; } bool WebSocketClientImpl::IsWebSocketAvailable() const { - return websocket_ != nullptr && !websocket_->isClosed() && - !websocket_->isShuttingDown(); + return state_ == kStateOpen && easy_ != nullptr; +} + +// --------------------------------------------------------------------------- +// Handler notifications, marshalled to the Database scheduler. +// --------------------------------------------------------------------------- + +void WebSocketClientImpl::ScheduleHandlerEvent( + std::function fn) { + if (handler_ == nullptr) return; + ClientRef client_ref = safe_this_; + scheduler_->Schedule([client_ref, fn = std::move(fn)]() mutable { + ClientRefLock lock(&client_ref); + WebSocketClientImpl* client = lock.GetReference(); + if (client != nullptr && client->handler_ != nullptr) { + fn(client->handler_); + } + }); +} + +void WebSocketClientImpl::OnError() { + std::string uri = uri_; + ScheduleHandlerEvent( + [uri = std::move(uri)](WebSocketClientEventHandler* handler) { + WebSocketClientErrorData error(uri.c_str()); + handler->OnError(error); + }); +} + +void WebSocketClientImpl::OnConnection() { + ScheduleHandlerEvent( + [](WebSocketClientEventHandler* handler) { handler->OnOpen(); }); +} + +void WebSocketClientImpl::OnMessage(const char* message, size_t length) { + std::string payload(message, length); + ScheduleHandlerEvent( + [payload = std::move(payload)](WebSocketClientEventHandler* handler) { + handler->OnMessage(payload.c_str()); + }); +} + +void WebSocketClientImpl::OnDisconnection() { + ScheduleHandlerEvent( + [](WebSocketClientEventHandler* handler) { handler->OnClose(); }); } } // namespace connection diff --git a/database/src/desktop/connection/web_socket_client_impl.h b/database/src/desktop/connection/web_socket_client_impl.h index fecd1863c8..6843c5bb95 100644 --- a/database/src/desktop/connection/web_socket_client_impl.h +++ b/database/src/desktop/connection/web_socket_client_impl.h @@ -15,24 +15,47 @@ #ifndef FIREBASE_DATABASE_SRC_DESKTOP_CONNECTION_WEB_SOCKET_CLIENT_IMPL_H_ #define FIREBASE_DATABASE_SRC_DESKTOP_CONNECTION_WEB_SOCKET_CLIENT_IMPL_H_ -#include +#include +#include +#include +#include +#include #include -#include #include +#include "app/rest/curl/curl_handles.h" #include "app/src/include/firebase/internal/mutex.h" #include "app/src/logger.h" #include "app/src/safe_reference.h" #include "app/src/scheduler.h" #include "app/src/thread.h" +#include "curl/curl.h" #include "database/src/desktop/connection/web_socket_client_interface.h" -#include "uWebSockets/src/uWS.h" namespace firebase { namespace database { namespace internal { namespace connection { +// libcurl-based WebSocket client. +// +// Threading model +// --------------- +// * Each client owns one I/O thread for the lifetime of the object. That +// thread drives a private CURLM multi handle through the HTTP Upgrade +// handshake (CURLOPT_CONNECT_ONLY = 2) and then services the connected +// socket with curl_ws_recv()/curl_ws_send(), sleeping in curl_multi_poll() +// between events. Nothing is shared with the REST transport except +// libcurl's global state (AcquireCurlGlobal()) and the TLS policy. +// * Connect(), Close(), Send() and the destructor may be called from any +// thread (in practice the Database scheduler thread). They post work to the +// I/O thread and return immediately; the destructor posts a teardown and +// joins the thread, so no I/O-thread work touches this object after it +// returns. +// * WebSocketClientEventHandler callbacks (OnOpen/OnMessage/OnClose/OnError) +// are never invoked on the I/O thread: they are marshalled to the Database +// `scheduler_` through `safe_this_`, which also makes them no-ops after the +// client is destroyed. class WebSocketClientImpl : public WebSocketClientInterface { public: WebSocketClientImpl(const std::string& uri, const std::string& user_agent, @@ -58,107 +81,125 @@ class WebSocketClientImpl : public WebSocketClientInterface { void RefreshAppCheckToken(const std::string& token); private: - typedef uWS::WebSocket ClientWebSocket; - typedef void (*Callback)(WebSocketClientImpl* client, int int_value, - const std::string& string_value); - - // Callback for hub_ when connection error occurs - static void OnError(void* data); - - // Callback for hub_ when connection is established - static void OnConnection(ClientWebSocket* ws, uWS::HttpRequest req); - - // Callback for hub_ when a message is received from the server - static void OnMessage(ClientWebSocket* ws, char* message, size_t length, - uWS::OpCode opCode); - - // Callback for hub_ when the connection is closed - static void OnDisconnection(ClientWebSocket* ws, int code, char* message, - size_t length); - - // The thread routine to host the event loop of hub_ - static void EventLoopRoutine(void* data); - - // Synchronously request to close the websocket. Should only be called in - // evnet loop thread. - void CloseSync(); - - // Schedule an async callback to be trigger in the next iteration of the event - // loop. This call is thread-safe and is to prevent multiple threads fighting - // for the same resource, such as websocket_ - void ScheduleOnce(Callback cb, int int_value, - const std::string& string_value); - - // Process callback queue in event loop thread - static void ProcessCallbackQueue(uS::Async* async); + enum State { + kStateIdle = 0, + kStateConnecting, + kStateOpen, + kStateClosed, + }; - // Check if the websocket is available and not closed. - // Only call this in event loop. + // Queues fn to run on the I/O thread and wakes it. Thread-safe. + void Post(std::function fn); + + // Entry point and event loop for the background WebSocket I/O thread. + static void EventLoopRoutine(void* client); + void RunEventLoop(); + // Runs every queued function in `posted_`. I/O thread only. + void RunPosted(); + + // The following run on the I/O thread only. + + // Start the asynchronous WebSocket handshake. + void ConnectOnLoop(int timeout_ms); + // Handshake finished (CURLMSG_DONE): either fail or switch to kStateOpen. + void OnHandshakeDone(CURLcode result); + // Gracefully close the WebSocket connection and notify the handler. + void CloseOnLoop(); + // Queue and flush an outbound text message. + void SendOnLoop(const std::string& msg); + // Destructor path: release every I/O-thread resource without callbacks. + void TeardownOnLoop(); + + // Socket readiness while open (`revents` is a CURL_WAIT_* mask): flush and + // receive. + void Pump(int revents); + // Attempt to flush any queued outbound messages via curl_ws_send(). + void FlushSendQueue(); + // Drain all available incoming WebSocket frames via curl_ws_recv(). + void ReceiveFrames(); + // Detach and destroy the easy handle and reset all per-connection buffers. + void CleanupEasyHandle(); + // Check if the WebSocket connection is currently established and open. bool IsWebSocketAvailable() const; - // The server uri to connect to. Should start with "ws://" for non-SSL - // connection or "wss://" for SSL connection. + // Event notifications matching the WebSocketClientEventHandler contract; + // each one marshals to scheduler_ via ScheduleHandlerEvent(). + void ScheduleHandlerEvent( + std::function fn); + void OnError(); + void OnConnection(); + void OnMessage(const char* message, size_t length); + void OnDisconnection(); + + // The server URI to connect to ("ws://..." or "wss://..."). const std::string uri_; // The event handler for connection events. WebSocketClientEventHandler* handler_; - // The thread to host the event loop of hub_ - std::unique_ptr thread_; + // Private multi handle: drives the handshake and provides + // curl_multi_poll()/curl_multi_wakeup() for the connected socket. + firebase::rest::curl::UniqueMulti multi_; - // The access point for uWebSockets which contains event loops and - // different sockets. - uWS::Hub hub_; + // Mutex protecting `posted_`, the queue of closures posted via Post() from + // other threads (Connect(), Send(), Close(), and ~WebSocketClientImpl()) to + // be drained on the background I/O thread. + Mutex post_mutex_; + std::deque> posted_; - // The handler to keep the event loop of hub_ alive even there is no - // connection at all. Otherwise the loop would stop when there is nothing to - // handle anymore. Also used to asynchronously close all async handles and - // WebSocket connection in destructor. - uS::Async* keep_loop_alive_; + // Set by the destructor's posted teardown to exit RunEventLoop(). + bool quit_; - // Async handler to process callback queue in event look thread. - uS::Async* process_queue_async_; + // The active libcurl easy handle for the WebSocket connection. Attached to + // multi_ from ConnectOnLoop() until CleanupEasyHandle(): libcurl closes a + // connect-only connection as soon as its handle is removed. + firebase::rest::curl::UniqueEasy easy_; - // A blob of callback data which is added to callback queue and triggered in - // the event loop thread. The callback member is called with a reference to - // the client, int_value and string_value stored in this data structure. - struct CallbackData { - explicit CallbackData(Callback c, WebSocketClientImpl* ws_client, int i, - const std::string& str) - : callback(c), client(ws_client), int_value(i), string_value(str) {} + // HTTP Upgrade request headers associated with easy_. + firebase::rest::curl::UniqueSList request_headers_; - Callback callback; + // The connected socket while open, otherwise CURL_SOCKET_BAD. + curl_socket_t socket_; - WebSocketClientImpl* client; + // Current connection state. + State state_; - // Generic purposed int value. Used as timeout in this case. - int int_value; + // Deadline for the HTTP Upgrade exchange. libcurl bounds only DNS/TCP/TLS + // for a CURLOPT_CONNECT_ONLY handle, so RunEventLoop() enforces this itself. + std::chrono::steady_clock::time_point handshake_deadline_; - // Generic purposed string value. Used as message to send in this case. - std::string string_value; + // Buffer for reassembling multi-chunk or fragmented (CURLWS_CONT) incoming + // WebSocket data frames. + std::string incoming_message_buffer_; + + struct OutboundFrame { + std::string payload; + unsigned int flags; }; - // A queue of callback to be triggered in event loop thread. - std::queue callback_queue_; + // Buffer for reassembling incoming PING payloads if split across chunks. + std::string incoming_ping_buffer_; - // Mutex to guard callback_queue_ - Mutex callback_queue_mutex_; + // Queue of outbound WebSocket frames (CURLWS_TEXT data frames and + // CURLWS_PONG control replies) waiting to be sent over the socket. + std::deque send_queue_; - // Flagged when this object starts to be destructed. This helps the other - // thread to handle situation accordingly, ex. if the connection is - // established after this object starts to be deleted. Note that this flag is - // is not guarded by any synchronization primitives and is changed outside of - // the event loop. If referenced during the event loop, make sure the racing - // condition will not cause the problem. - std::atomic is_destructing_; + // True once curl_ws_send() has been invoked for send_queue_.front() and + // libcurl may hold encoded header/payload bytes in its internal sendbuf, + // even when curl_ws_send() returned CURLE_AGAIN with sent == 0. + bool frame_in_progress_; - // The websocket handler to send message, to receive message and to close the - // connection. Should only be used in the event loop. Not thread safe - ClientWebSocket* websocket_; + // Byte offset into send_queue_.front().payload for partial non-blocking + // writes. + size_t current_send_offset_; // User agent used when opening the connection. std::string user_agent_; + // Mutex guarding app_check_token_ (written from the scheduler thread, read + // on the I/O thread when connecting). + Mutex app_check_token_mutex_; + // App Check token used when opening the connection. std::string app_check_token_; @@ -166,14 +207,16 @@ class WebSocketClientImpl : public WebSocketClientInterface { scheduler::Scheduler* scheduler_; - // Safe reference to this. Set in constructor and cleared in destructor - // Should be safe to be copied in any thread because the std::shared_ptr never - // changes, until safe_this_ is completely destroyed. + // Safe reference to this. Set in constructor and cleared in destructor. typedef firebase::internal::SafeReference ClientRef; typedef firebase::internal::SafeReferenceLock ClientRefLock; ClientRef safe_this_; + // Declared last so it is the first member destroyed; it is joined in the + // destructor body before that anyway. + Thread thread_; + friend class PersistentConnectionTest; }; diff --git a/database/src/desktop/database_desktop.cc b/database/src/desktop/database_desktop.cc index a7e0a5e544..459886e1a8 100644 --- a/database/src/desktop/database_desktop.cc +++ b/database/src/desktop/database_desktop.cc @@ -74,6 +74,7 @@ DatabaseInternal::DatabaseInternal(App* app, const char* url) cleanup_(), database_url_(url), constructor_url_(url), + persistence_enabled_(false), logger_(app_common::FindAppLoggerByName(app->name())), repo_(nullptr) { assert(app); diff --git a/database/src/desktop/database_desktop.h b/database/src/desktop/database_desktop.h index 8cd3d24388..6dc991111d 100644 --- a/database/src/desktop/database_desktop.h +++ b/database/src/desktop/database_desktop.h @@ -177,7 +177,7 @@ class DatabaseInternal { // We keep it so that we can find the database in our cache. std::string constructor_url_; - bool persistence_enabled_; + bool persistence_enabled_ = false; // The logger for this instance of the database. Logger logger_; diff --git a/database/tests/CMakeLists.txt b/database/tests/CMakeLists.txt index 4db43fd9a2..d99db4e30d 100644 --- a/database/tests/CMakeLists.txt +++ b/database/tests/CMakeLists.txt @@ -342,19 +342,30 @@ firebase_cpp_cc_test( SOURCES desktop/connection/web_socket_client_impl_test.cc INCLUDES + ${CURL_SOURCE_DIR}/include ${OPENSSL_INCLUDE_DIR} - ${UWEBSOCKETS_SOURCE_DIR}/.. DEPENDS firebase_database firebase_testing + libcurl ${OPENSSL_CRYPTO_LIBRARY} - libuWS ) if(MSVC) + # - WIN32_LEAN_AND_MEAN prevents from pulling in legacy + # , which conflicts with . + # - CURL_STATICLIB=1 tells (included via + # web_socket_client_impl.h) that libcurl is linked statically rather than + # imported from a DLL (__declspec(dllimport)). target_compile_definitions(firebase_rtdb_desktop_connection_web_socket_client_impl_test PRIVATE -DWIN32_LEAN_AND_MEAN + -DCURL_STATICLIB=1 + ) + # TestWebSocketEchoServer calls Winsock2 socket APIs directly. + target_link_libraries(firebase_rtdb_desktop_connection_web_socket_client_impl_test + PRIVATE + ws2_32 ) endif() @@ -364,10 +375,8 @@ firebase_cpp_cc_test( desktop/connection/connection_test.cc INCLUDES ${OPENSSL_INCLUDE_DIR} - ${UWEBSOCKETS_SOURCE_DIR}/.. DEPENDS ${OPENSSL_CRYPTO_DIR} - libuWS firebase_app_for_testing firebase_database firebase_testing diff --git a/database/tests/desktop/connection/web_socket_client_impl_test.cc b/database/tests/desktop/connection/web_socket_client_impl_test.cc index c0f219b410..74ee91c589 100644 --- a/database/tests/desktop/connection/web_socket_client_impl_test.cc +++ b/database/tests/desktop/connection/web_socket_client_impl_test.cc @@ -14,133 +14,492 @@ #include "database/src/desktop/connection/web_socket_client_impl.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "app/src/include/firebase/internal/platform.h" #include "app/src/semaphore.h" +#include "app/src/time.h" +#include "database/src/desktop/util_desktop.h" #include "gmock/gmock.h" #include "gtest/gtest.h" +#if FIREBASE_PLATFORM_WINDOWS +#include +#include +#else +#include +#include +#include +#include +#include +#include +#include +#endif + namespace firebase { namespace database { namespace internal { namespace connection { -// Simple WebSocket based Echo Server using third_party/uWebSockets -// It has some quirk. Ex. hub_ needs a handler (async_) to wake the loop before -// closing it or the event loop will never stop. +namespace { + +// Platform socket compatibility helpers: +// - Windows (Winsock2) represents sockets as unsigned `SOCKET` handles, uses +// `INVALID_SOCKET` as the error sentinel, closes sockets via `closesocket()`, +// and reports socket errors via `WSAGetLastError()`. +// - POSIX (Linux/macOS) represents sockets as signed `int` file descriptors, +// uses `-1` as the error sentinel, closes sockets via `close()`, and reports +// socket errors via `errno`. +#if FIREBASE_PLATFORM_WINDOWS +typedef SOCKET SocketHandle; +constexpr SocketHandle kInvalidSocket = INVALID_SOCKET; +void CloseSocket(SocketHandle s) { closesocket(s); } +bool LastSocketErrorWasTimeout() { + int err = WSAGetLastError(); + return err == WSAETIMEDOUT || err == WSAEWOULDBLOCK; +} +#else +typedef int SocketHandle; +constexpr SocketHandle kInvalidSocket = -1; +void CloseSocket(SocketHandle s) { close(s); } +bool LastSocketErrorWasTimeout() { + return errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR; +} +#endif + +// RFC 6455 section 4.2.2 magic GUID concatenated with the client's +// `Sec-WebSocket-Key` header before SHA-1 hashing and Base64 encoding to form +// `Sec-WebSocket-Accept`. +constexpr char kWebSocketGuid[] = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; + +int64_t ElapsedMs(std::chrono::steady_clock::time_point start) { + return std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); +} + +// Self-contained RFC 6455 localhost WebSocket test server. class TestWebSocketEchoServer { public: - explicit TestWebSocketEchoServer(int port) - : port_(port), run_(false), thread_(nullptr), keep_alive_(nullptr) { - hub_.onMessage([](uWS::WebSocket* ws, char* message, - size_t length, uWS::OpCode opCode) { - // Echo back immediately - ws->send(message, length, opCode); - }); - hub_.onConnection( - [](uWS::WebSocket* ws, uWS::HttpRequest request) { - LogDebug("[Server] Received connection from (%s) %s port: %d", - ws->getAddress().family, ws->getAddress().address, - ws->getAddress().port); - }); - hub_.onDisconnection([](uWS::WebSocket* ws, int code, - char* message, size_t length) { - LogDebug("[Server] Disconnected from (%s) %s port: %d", - ws->getAddress().family, ws->getAddress().address, - ws->getAddress().port); - }); + enum ServerBehavior { + // Echoes client data frames back, replies to PING with PONG, and echoes + // CLOSE before disconnecting. + kBehaviorEcho = 0, + // Immediately after the 101 handshake, sends a 3-fragment text message + // with a PING control frame interleaved between fragments 1 and 2, then + // enters the normal echo loop. + kBehaviorSendFragmentedWithInterleavedPing, + // Immediately after the 101 handshake, sends a normal CLOSE frame (1000) + // and closes the connection. + kBehaviorServerInitiatedClose, + // Accepts the TCP connection and reads the HTTP Upgrade request but never + // sends the 101 response, exercising the client's handshake deadline + // watchdog. + kBehaviorNeverAnswerHandshake, + }; + + explicit TestWebSocketEchoServer(int port = 0, + ServerBehavior behavior = kBehaviorEcho) + : port_(port), + behavior_(behavior), + run_(false), + listen_sock_(kInvalidSocket), + thread_(nullptr), + pong_received_(false) { +#if FIREBASE_PLATFORM_WINDOWS + // Winsock2 requires explicit per-process initialization (reference-counted + // by the OS) before any socket APIs (`socket`, `bind`, `listen`, etc.) can + // be called. + WSADATA wsa_data; + WSAStartup(MAKEWORD(2, 2), &wsa_data); +#else + // On POSIX, calling `send()` on a socket after the peer has closed its end + // raises `SIGPIPE`, which terminates the test process by default. Ignore + // `SIGPIPE` so `send()` instead fails cleanly with `EPIPE`. + signal(SIGPIPE, SIG_IGN); +#endif } - ~TestWebSocketEchoServer() { Stop(); } + ~TestWebSocketEchoServer() { + Stop(); +#if FIREBASE_PLATFORM_WINDOWS + WSACleanup(); +#endif + } void Start() { - keep_alive_ = new uS::Async(hub_.getLoop()); - keep_alive_->setData(this); - keep_alive_->start([](uS::Async* async) { - TestWebSocketEchoServer* server = - static_cast(async->getData()); - assert(server != nullptr); - // close ths group in event loop thread - server->hub_.getDefaultGroup().close(); - async->close(); - }); - - run_ = true; - thread_ = new std::thread([this]() { - auto listen = [&](int port) { - if (hub_.listen(port)) { - LogDebug("[Server] Starts to listen to port %d", port); - return true; - } else { - LogDebug("[Server] Cannot listen to port %d", port); - return false; - } - }; - - if (port_ == 0) { - int attempts = 1000; - int port = 0; - bool res = false; - - do { - --attempts; - port = 10000 + (rand() % 55000); // NOLINT - res = listen(port); - } while (run_ == true && res == false && attempts != 0); - - if (res) { - port_ = port; - hub_.run(); // Blocks until done - } else if (attempts == 0) { - LogError("Failed to find free port after 1000 attempts"); - } - } else { - if (listen(port_) == true) { - hub_.run(); // Blocks until done - } else { - LogWarning("[Server] Cannot listen to port %d", port_.load()); - } - } - - run_ = false; - }); + listen_sock_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + ASSERT_NE(listen_sock_, kInvalidSocket); + + int opt = 1; + setsockopt(listen_sock_, SOL_SOCKET, SO_REUSEADDR, + reinterpret_cast(&opt), sizeof(opt)); + + // Bind to 127.0.0.1 (`INADDR_LOOPBACK`). When `port_` is 0, the OS kernel + // assigns an available ephemeral port, which we query below via + // `getsockname()`. + sockaddr_in addr; + std::memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = htons(static_cast(port_.load())); + + ASSERT_EQ( + bind(listen_sock_, reinterpret_cast(&addr), sizeof(addr)), + 0); + ASSERT_EQ(listen(listen_sock_, 128), 0); + + sockaddr_in bound_addr; + socklen_t addr_len = sizeof(bound_addr); + ASSERT_EQ(getsockname(listen_sock_, + reinterpret_cast(&bound_addr), &addr_len), + 0); + port_.store(ntohs(bound_addr.sin_port)); + + run_.store(true); + thread_ = new std::thread([this]() { RunAcceptLoop(); }); } void Stop() { - run_ = false; - - if (keep_alive_) { - keep_alive_->send(); - keep_alive_ = nullptr; - } - + run_.store(false); if (thread_ != nullptr) { thread_->join(); delete thread_; thread_ = nullptr; } + if (listen_sock_ != kInvalidSocket) { + CloseSocket(listen_sock_); + listen_sock_ = kInvalidSocket; + } } int GetPort(bool waitForPort = false) const { - while (waitForPort == true && run_ == true && port_ == 0) { + while (waitForPort && run_.load() && port_.load() == 0) { firebase::internal::Sleep(10); } + return port_.load(); + } - return port_; + std::string GetLastHandshakeRequest() const { + std::lock_guard lock(mutex_); + return last_handshake_request_; + } + + bool pong_received() const { return pong_received_.load(); } + + std::string last_pong_payload() const { + std::lock_guard lock(mutex_); + return last_pong_payload_; } private: + static bool SendAll(SocketHandle sock, const void* data, size_t len) { + const char* ptr = static_cast(data); + int flags = 0; +#ifdef MSG_NOSIGNAL + flags |= MSG_NOSIGNAL; +#endif + while (len > 0) { + auto sent = send(sock, ptr, static_cast(len), flags); + if (sent <= 0) { + return false; + } + ptr += sent; + len -= static_cast(sent); + } + return true; + } + + bool RecvExact(SocketHandle sock, void* buf, size_t len) { + char* ptr = static_cast(buf); + while (len > 0) { + auto r = recv(sock, ptr, static_cast(len), 0); + if (r > 0) { + ptr += r; + len -= static_cast(r); + continue; + } + if (r == 0) { + return false; + } + if (LastSocketErrorWasTimeout() && run_.load()) { + continue; + } + return false; + } + return true; + } + + static bool SendFrame(SocketHandle sock, uint8_t first_byte, + const void* payload, size_t len) { + uint8_t header[10]; + size_t header_len = 0; + header[0] = first_byte; + if (len < 126) { + header[1] = static_cast(len); + header_len = 2; + } else if (len <= 0xFFFF) { + header[1] = 126; + header[2] = static_cast((len >> 8) & 0xFF); + header[3] = static_cast(len & 0xFF); + header_len = 4; + } else { + header[1] = 127; + uint64_t l64 = static_cast(len); + for (int i = 0; i < 8; ++i) { + header[2 + i] = static_cast((l64 >> ((7 - i) * 8)) & 0xFF); + } + header_len = 10; + } + if (!SendAll(sock, header, header_len)) { + return false; + } + if (len > 0 && !SendAll(sock, payload, len)) { + return false; + } + return true; + } + + // Sleeps in 10ms slices so Stop() wakes the server thread promptly. + void SleepWhileRunning(int ms) { + for (int waited = 0; waited < ms && run_.load(); waited += 10) { + firebase::internal::Sleep(10); + } + } + + void RunAcceptLoop() { + while (run_.load()) { + SocketHandle lsock = listen_sock_; + if (lsock == kInvalidSocket) { + break; + } +#if !FIREBASE_PLATFORM_WINDOWS + if (lsock >= FD_SETSIZE) { + break; + } +#endif + fd_set read_fds; + FD_ZERO(&read_fds); + FD_SET(lsock, &read_fds); + timeval tv; + tv.tv_sec = 0; + tv.tv_usec = 50 * 1000; // 50ms + int sel = + select(static_cast(lsock + 1), &read_fds, nullptr, nullptr, &tv); + if (sel <= 0 || !run_.load()) { + continue; + } + + sockaddr_in client_addr; + socklen_t client_len = sizeof(client_addr); + SocketHandle client_sock = + accept(lsock, reinterpret_cast(&client_addr), &client_len); + if (client_sock == kInvalidSocket) { + continue; + } + HandleClient(client_sock); + CloseSocket(client_sock); + } + } + + void HandleClient(SocketHandle sock) { + // Set a 200ms receive timeout (`SO_RCVTIMEO`) so blocking `recv()` calls + // wake periodically to check `run_` when `Stop()` is called, and so + // aborted client connections in `TestEdgeCase2` that close before sending + // a complete HTTP Upgrade request do not block the server thread. + // Windows `setsockopt(SO_RCVTIMEO)` takes a `DWORD` millisecond count, + // whereas POSIX takes a `struct timeval`. +#if FIREBASE_PLATFORM_WINDOWS + DWORD timeout_ms = 200; + setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, + reinterpret_cast(&timeout_ms), sizeof(timeout_ms)); +#else + timeval tv; + tv.tv_sec = 0; + tv.tv_usec = 200 * 1000; + setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); +#endif + + // Read HTTP Upgrade request headers until "\r\n\r\n". + std::string request; + char ch = 0; + while (request.find("\r\n\r\n") == std::string::npos) { + auto r = recv(sock, &ch, 1, 0); + if (r <= 0) { + return; + } + request.push_back(ch); + if (request.size() > 8192) { + return; + } + } + + { + std::lock_guard lock(mutex_); + last_handshake_request_ = request; + } + + // Extract Sec-WebSocket-Key header (case-insensitive). + std::string ws_key; + std::istringstream stream(request); + std::string line; + while (std::getline(stream, line)) { + if (!line.empty() && line.back() == '\r') { + line.pop_back(); + } + const std::string kPrefix = "Sec-WebSocket-Key:"; + if (line.size() >= kPrefix.size()) { + bool match = true; + for (size_t i = 0; i < kPrefix.size(); ++i) { + char a = line[i]; + char b = kPrefix[i]; + if (a >= 'A' && a <= 'Z') a = static_cast(a - 'A' + 'a'); + if (b >= 'A' && b <= 'Z') b = static_cast(b - 'A' + 'a'); + if (a != b) { + match = false; + break; + } + } + if (match) { + size_t pos = kPrefix.size(); + while (pos < line.size() && (line[pos] == ' ' || line[pos] == '\t')) { + ++pos; + } + ws_key = line.substr(pos); + break; + } + } + } + if (ws_key.empty()) { + return; + } + + if (behavior_ == kBehaviorNeverAnswerHandshake) { + // Hold the connection open without replying until the server stops. + SleepWhileRunning(30000); + return; + } + + std::string accept_key; + GetBase64SHA1(ws_key + kWebSocketGuid, &accept_key); + + std::string response = + "HTTP/1.1 101 Switching Protocols\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + "Sec-WebSocket-Accept: " + + accept_key + "\r\n\r\n"; + if (!SendAll(sock, response.data(), response.size())) { + return; + } + + if (behavior_ == kBehaviorServerInitiatedClose) { + const uint8_t close_code[2] = {0x03, 0xe8}; // 1000 Normal Closure + SendFrame(sock, 0x88, close_code, sizeof(close_code)); + return; + } + + if (behavior_ == kBehaviorSendFragmentedWithInterleavedPing) { + // Send a 3-fragment text message with a PING control frame interleaved + // between fragment 1 and fragment 2: + // Fragment 1: FIN=0, opcode=0x1 (TEXT), payload="Part1-" + SendFrame(sock, 0x01, "Part1-", 6); + // Interleaved PING: FIN=1, opcode=0x9 (PING), payload="ping-data" + SendFrame(sock, 0x89, "ping-data", 9); + // Fragment 2: FIN=0, opcode=0x0 (CONT), payload="Part2-" + SendFrame(sock, 0x00, "Part2-", 6); + // Fragment 3: FIN=1, opcode=0x0 (CONT), payload="Part3" + SendFrame(sock, 0x80, "Part3", 5); + } + + // Read client frames and echo data frames back. + while (run_.load()) { + uint8_t hdr[2]; + if (!RecvExact(sock, hdr, 2)) { + break; + } + uint8_t opcode = hdr[0] & 0x0F; + bool masked = (hdr[1] & 0x80) != 0; + uint64_t payload_len = hdr[1] & 0x7F; + if (payload_len == 126) { + uint8_t ext[2]; + if (!RecvExact(sock, ext, 2)) break; + payload_len = (static_cast(ext[0]) << 8) | ext[1]; + } else if (payload_len == 127) { + uint8_t ext[8]; + if (!RecvExact(sock, ext, 8)) break; + payload_len = 0; + for (int i = 0; i < 8; ++i) { + payload_len = (payload_len << 8) | ext[i]; + } + } + + uint8_t mask_key[4] = {0, 0, 0, 0}; + if (masked && !RecvExact(sock, mask_key, 4)) { + break; + } + + std::vector payload(static_cast(payload_len)); + if (payload_len > 0 && !RecvExact(sock, payload.data(), payload.size())) { + break; + } + if (masked) { + for (size_t i = 0; i < payload.size(); ++i) { + payload[i] ^= mask_key[i & 3]; + } + } + + if (opcode == 0x8) { + // CLOSE frame: echo CLOSE back and exit. + SendFrame(sock, 0x88, payload.data(), payload.size()); + break; + } else if (opcode == 0x9) { + // PING frame: reply with PONG. + SendFrame(sock, 0x8A, payload.data(), payload.size()); + } else if (opcode == 0xA) { + // PONG frame: record payload. + { + std::lock_guard lock(mutex_); + last_pong_payload_.assign( + reinterpret_cast(payload.data()), payload.size()); + } + pong_received_.store(true); + } else { + // Echo the frame back to the client. + if (!SendFrame(sock, hdr[0], payload.data(), payload.size())) { + break; + } + } + } + } + std::atomic port_; - std::atomic run_; // Is the listen thread started and running - uWS::Hub hub_; + ServerBehavior behavior_; + std::atomic run_; + SocketHandle listen_sock_; std::thread* thread_; - uS::Async* keep_alive_; + mutable std::mutex mutex_; + std::string last_handshake_request_; + std::atomic pong_received_; + std::string last_pong_payload_; }; std::string GetLocalHostUri(int port) { std::stringstream ss; - ss << "ws://localhost:" << port; + ss << "ws://127.0.0.1:" << port; return ss.str(); } +} // namespace + class TestClientEventHandler : public WebSocketClientEventHandler { public: explicit TestClientEventHandler(Semaphore* s) @@ -150,7 +509,7 @@ class TestClientEventHandler : public WebSocketClientEventHandler { is_closed_(false), is_error_(false), semaphore_(s) {} - ~TestClientEventHandler() override{}; + ~TestClientEventHandler() override {} void OnOpen() override { is_connected_ = true; @@ -170,6 +529,7 @@ class TestClientEventHandler : public WebSocketClientEventHandler { void OnError(const WebSocketClientErrorData& error_data) override { is_error_ = true; + error_uri_ = error_data.GetUri(); semaphore_->Post(); } @@ -178,68 +538,58 @@ class TestClientEventHandler : public WebSocketClientEventHandler { std::string msg_received_; bool is_closed_ = false; bool is_error_ = false; + std::string error_uri_; private: Semaphore* semaphore_; }; // Test if the client can connect to a local echo server, send a message, -// receive message and close the connection properly. +// receive a message and close the connection properly. TEST(WebSocketClientImpl, Test1) { - // Launch a local echo server TestWebSocketEchoServer server(0); server.Start(); auto uri = GetLocalHostUri(server.GetPort(true)); - Semaphore semaphore(1); + Semaphore semaphore(0); TestClientEventHandler handler(&semaphore); Logger logger(nullptr); scheduler::Scheduler scheduler; - WebSocketClientImpl ws_client(uri.c_str(), "", &logger, &scheduler, "", - &handler); + WebSocketClientImpl ws_client(uri.c_str(), "test-agent/1.0", &logger, + &scheduler, "", &handler); - // Connect to local server - LogDebug("[Client] Connecting to %s", uri.c_str()); - EXPECT_TRUE(semaphore.TryWait()); + // Connect to local server. ws_client.Connect(5000); - semaphore.Wait(); - semaphore.Post(); + EXPECT_TRUE(semaphore.TimedWait(5000)); EXPECT_TRUE(handler.is_connected_ && !handler.is_error_); - // Send a message and wait for the response - EXPECT_TRUE(semaphore.TryWait()); + // Send a message and wait for the echoed response. ws_client.Send("Hello World"); - semaphore.Wait(); - semaphore.Post(); + EXPECT_TRUE(semaphore.TimedWait(5000)); EXPECT_TRUE(handler.is_msg_received_ && !handler.is_error_); EXPECT_STREQ("Hello World", handler.msg_received_.c_str()); - // Close the connection - EXPECT_TRUE(semaphore.TryWait()); + // Close the connection. ws_client.Close(); - semaphore.Wait(); - semaphore.Post(); + EXPECT_TRUE(semaphore.TimedWait(5000)); EXPECT_TRUE(handler.is_closed_ && !handler.is_error_); - // Stop the server server.Stop(); } // Test if it is safe to create the client and destroy it immediately. -// This is to test if the destructor can properly end the event loop. -// Otherwise, it would block forever and timeout +// This verifies that the destructor cleanly joins the I/O thread even when +// Connect() was never called. TEST(WebSocketClientImpl, TestEdgeCase1) { Logger logger(nullptr); scheduler::Scheduler scheduler; - WebSocketClientImpl ws_client("ws://localhost", "", &logger, &scheduler, ""); + WebSocketClientImpl ws_client("ws://127.0.0.1", "", &logger, &scheduler, ""); } -// Test if it is safe to connect to a server and destroy the client immediately. -// This is to test if the destructor can properly end the event loop -// Otherwise, it would block forever and timeout +// Test if it is safe to connect to a server and destroy the client while the +// handshake is in flight. TEST(WebSocketClientImpl, TestEdgeCase2) { - // Launch a local echo server TestWebSocketEchoServer server(0); server.Start(); Logger logger(nullptr); @@ -247,23 +597,194 @@ TEST(WebSocketClientImpl, TestEdgeCase2) { auto uri = GetLocalHostUri(server.GetPort(true)); - int count = 0; - while ((count++) < 10000) { + for (int count = 0; count < 50; ++count) { WebSocketClientImpl* ws_client = new WebSocketClientImpl(uri.c_str(), "", &logger, &scheduler, ""); - - // Connect to local server - LogDebug("[Client][%d] Connecting to %s", count, uri.c_str()); ws_client->Connect(5000); - - // Immediately destroy the client right after connect request delete ws_client; } - // Stop the server server.Stop(); } +// Test large messages (>16KB kRecvBufferSize, requiring multiple curl_ws_recv +// chunks), multi-frame fragmented messages (CURLWS_CONT) with an interleaved +// PING, and HTTP Upgrade headers (User-Agent and X-Firebase-AppCheck). +TEST(WebSocketClientImpl, LargeAndFragmentedMessagesWithInterleavedPing) { + TestWebSocketEchoServer server( + 0, TestWebSocketEchoServer::kBehaviorSendFragmentedWithInterleavedPing); + server.Start(); + + auto uri = GetLocalHostUri(server.GetPort(true)); + + Semaphore semaphore(0); + TestClientEventHandler handler(&semaphore); + Logger logger(nullptr); + scheduler::Scheduler scheduler; + WebSocketClientImpl ws_client(uri.c_str(), "custom-agent/2.0", &logger, + &scheduler, "initial-token", &handler); + ws_client.RefreshAppCheckToken("refreshed-app-check-token"); + + ws_client.Connect(5000); + // Wait for OnOpen() AND the initial fragmented message sent by the server. + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_connected_); + EXPECT_FALSE(handler.is_error_); + EXPECT_TRUE(handler.is_msg_received_); + EXPECT_EQ(handler.msg_received_, "Part1-Part2-Part3"); + + // Send a 100KB payload to exercise multi-chunk curl_ws_recv reassembly and + // flush the client's PONG back to the server. + std::string large_payload(100 * 1024, 'A'); + for (size_t i = 0; i < large_payload.size(); ++i) { + large_payload[i] = static_cast('A' + (i % 26)); + } + handler.is_msg_received_ = false; + handler.msg_received_.clear(); + ws_client.Send(large_payload.c_str()); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_msg_received_); + EXPECT_EQ(handler.msg_received_, large_payload); + + // Verify the server received the PONG matching its interleaved PING. + EXPECT_TRUE(server.pong_received()); + EXPECT_EQ(server.last_pong_payload(), "ping-data"); + + // Verify HTTP Upgrade request headers included User-Agent and the refreshed + // X-Firebase-AppCheck token. + std::string handshake = server.GetLastHandshakeRequest(); + EXPECT_NE(handshake.find("User-Agent: custom-agent/2.0\r\n"), + std::string::npos); + EXPECT_NE( + handshake.find("X-Firebase-AppCheck: refreshed-app-check-token\r\n"), + std::string::npos); + + ws_client.Close(); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_closed_); + + server.Stop(); +} + +// Test server-initiated CLOSE frame triggers OnClose() cleanly. +TEST(WebSocketClientImpl, ServerInitiatedClose) { + TestWebSocketEchoServer server( + 0, TestWebSocketEchoServer::kBehaviorServerInitiatedClose); + server.Start(); + + auto uri = GetLocalHostUri(server.GetPort(true)); + + Semaphore semaphore(0); + TestClientEventHandler handler(&semaphore); + Logger logger(nullptr); + scheduler::Scheduler scheduler; + WebSocketClientImpl ws_client(uri.c_str(), "", &logger, &scheduler, "", + &handler); + + ws_client.Connect(5000); + // Expect OnOpen() followed by OnClose(). + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_connected_); + EXPECT_TRUE(handler.is_closed_); + EXPECT_FALSE(handler.is_error_); + + server.Stop(); +} + +// Test connection failure triggers OnError(). +TEST(WebSocketClientImpl, ConnectionFailureTriggersOnError) { + // Bind and immediately stop a server to obtain an unused port. + int closed_port = 0; + { + TestWebSocketEchoServer server(0); + server.Start(); + closed_port = server.GetPort(true); + server.Stop(); + } + + auto uri = GetLocalHostUri(closed_port); + Semaphore semaphore(0); + TestClientEventHandler handler(&semaphore); + Logger logger(nullptr); + scheduler::Scheduler scheduler; + WebSocketClientImpl ws_client(uri.c_str(), "", &logger, &scheduler, "", + &handler); + + ws_client.Connect(2000); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_error_); + EXPECT_FALSE(handler.is_connected_); + EXPECT_EQ(handler.error_uri_, uri); +} + +// libcurl bounds only DNS/TCP/TLS for a CURLOPT_CONNECT_ONLY handle; the HTTP +// Upgrade exchange itself has no libcurl-side timeout. The client's own +// watchdog must turn a server that accepts TCP but never answers the Upgrade +// into OnError() within Connect()'s timeout. +TEST(WebSocketClientImpl, HandshakeStallTimesOutViaWatchdog) { + TestWebSocketEchoServer server( + 0, TestWebSocketEchoServer::kBehaviorNeverAnswerHandshake); + server.Start(); + + auto uri = GetLocalHostUri(server.GetPort(true)); + Semaphore semaphore(0); + TestClientEventHandler handler(&semaphore); + Logger logger(nullptr); + scheduler::Scheduler scheduler; + WebSocketClientImpl ws_client(uri.c_str(), "", &logger, &scheduler, "", + &handler); + + auto start = std::chrono::steady_clock::now(); + ws_client.Connect(500); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_error_); + EXPECT_FALSE(handler.is_connected_); + EXPECT_EQ(handler.error_uri_, uri); + // Fired by the watchdog (~500 ms), not by a TCP-level failure. + EXPECT_GE(ElapsedMs(start), 400); + EXPECT_LT(ElapsedMs(start), 4000); + // The server did receive the Upgrade request; it just never answered. + EXPECT_FALSE(server.GetLastHandshakeRequest().empty()); + + server.Stop(); +} + +// Non-ws/wss URI schemes (e.g. http://, file://) must be rejected before +// opening a connection even when linked against a multi-protocol system +// libcurl. +TEST(WebSocketClientImpl, RejectsNonWebSocketUriSchemes) { + Logger logger(nullptr); + scheduler::Scheduler scheduler; + + const std::string http_uri = "http://127.0.0.1:1/"; + { + Semaphore semaphore(0); + TestClientEventHandler handler(&semaphore); + WebSocketClientImpl ws_client(http_uri, "", &logger, &scheduler, "", + &handler); + ws_client.Connect(5000); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_error_); + EXPECT_FALSE(handler.is_connected_); + EXPECT_EQ(http_uri, handler.error_uri_); + } + + const std::string file_uri = "file:///etc/hosts"; + { + Semaphore semaphore(0); + TestClientEventHandler handler(&semaphore); + WebSocketClientImpl ws_client(file_uri, "", &logger, &scheduler, "", + &handler); + ws_client.Connect(5000); + EXPECT_TRUE(semaphore.TimedWait(5000)); + EXPECT_TRUE(handler.is_error_); + EXPECT_FALSE(handler.is_connected_); + EXPECT_EQ(file_uri, handler.error_uri_); + } +} + } // namespace connection } // namespace internal } // namespace database diff --git a/release_build_files/NOTICES b/release_build_files/NOTICES index 6b3c23ba45..e15144b92f 100644 --- a/release_build_files/NOTICES +++ b/release_build_files/NOTICES @@ -1354,81 +1354,6 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -===== -libuv -===== -libuv is licensed for use as follows: - -==== -Copyright (c) 2015-present libuv project contributors. - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to -deal in the Software without restriction, including without limitation the -rights to use, copy, modify, merge, publish, distribute, sublicense, and/or -sell copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in -all copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS -IN THE SOFTWARE. -==== - -This license applies to parts of libuv originating from the -https://github.com/joyent/libuv repository: - -==== - -Copyright Joyent, Inc. and other Node contributors. All rights reserved. -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to -deal in the Software without restriction, including without limitation the -rights to use, copy, modify, merge, publish, distribute, sublicense, and/or -sell copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in -all copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING -FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS -IN THE SOFTWARE. - -==== - -This license applies to all parts of libuv that are not externally -maintained libraries. - -The externally maintained libraries used by libuv are: - - - tree.h (from FreeBSD), copyright Niels Provos. Two clause BSD license. - - - inet_pton and inet_ntop implementations, contained in src/inet.c, are - copyright the Internet Systems Consortium, Inc., and licensed under the ISC - license. - - - stdint-msvc2008.h (from msinttypes), copyright Alexander Chemeris. Three - clause BSD license. - - - pthread-fixes.c, copyright Google Inc. and Sony Mobile Communications AB. - Three clause BSD license. - - - android-ifaddrs.h, android-ifaddrs.c, copyright Berkeley Software Design - Inc, Kenneth MacKay and Emergya (Cloud4all, FP7/2007-2013, grant agreement - n° 289016). Three clause BSD license. - - ====== Nanopb ====== @@ -1452,29 +1377,6 @@ redistribute it freely, subject to the following restrictions: 3. This notice may not be removed or altered from any source distribution.a - - - -=========== -uWebSockets -=========== -Copyright (c) 2016 Alex Hultman and contributors - -This software is provided 'as-is', without any express or implied -warranty. In no event will the authors be held liable for any damages -arising from the use of this software. - -Permission is granted to anyone to use this software for any purpose, -including commercial applications, and to alter it and redistribute it -freely, subject to the following restrictions: - -1. The origin of this software must not be misrepresented; you must not - claim that you wrote the original software. If you use this software - in a product, an acknowledgement in the product documentation would be - appreciated but is not required. -2. Altered source versions must be plainly marked as such, and must not be - misrepresented as being the original software. -3. This notice may not be removed or altered from any source distribution. ============================= diff --git a/release_build_files/readme.md b/release_build_files/readme.md index 0bd7234921..4c52efec08 100644 --- a/release_build_files/readme.md +++ b/release_build_files/readme.md @@ -461,7 +461,7 @@ Authentication | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32` App Check | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32` Firestore | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32, rpcrt4, ole32, shell32, dbghelp` Functions | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32, rpcrt4, ole32` -Realtime Database | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32, psapi, userenv, shell32` +Realtime Database | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32, shell32` Remote Config | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32, rpcrt4, ole32, icu` Storage | `advapi32, ws2_32, crypt32, bcrypt, iphlpapi, secur32` @@ -618,7 +618,18 @@ code. - General (Desktop): Updated the bundled libcurl from 7.73.0 to 8.23.0. Windows apps linking the desktop libraries must now also link the `bcrypt`, `iphlpapi`, and `secur32` Windows SDK libraries, and macOS apps - must now also link the `SystemConfiguration` framework. + must now also link the `SystemConfiguration` framework. Builds that + supply their own libcurl via `CURL_SOURCE_DIR` require libcurl 8.14.0 or + newer built with WebSocket support. + - Database (Desktop): Replaced the uWebSockets/libuv WebSocket transport + with libcurl's WebSocket implementation, driven by one lightweight + event-driven thread per connection. Windows apps no longer need to link + `psapi` or `userenv` for the Realtime Database. + - General (Desktop): REST and WebSocket connections now share one TLS + policy: peer and hostname verification are always on, TLS 1.2 is the + minimum version, and certificates are validated against the OS trust + store (Schannel on Windows, the keychain on macOS) or the distribution's + CA bundle on Linux. - General (Linux Desktop): Fixed a potential 1-byte buffer underflow and file descriptor leak when reading `/etc/timezone` in `GetTimezone()`. - Database, Functions, Storage (Android): Fixed a crash in `GetInstance()` diff --git a/scripts/gha/print_matrix_configuration.py b/scripts/gha/print_matrix_configuration.py index 6384cf3751..c918d481ea 100644 --- a/scripts/gha/print_matrix_configuration.py +++ b/scripts/gha/print_matrix_configuration.py @@ -326,8 +326,6 @@ def filter_values_on_diff(parm_key, value, auto_diff): # ignored completely (set to None), or scan the changes more in-depth (set # to scan_changes function) "cmake/external/firestore.cmake": "firestore", - "cmake/external/libuv.cmake": "database", - "cmake/external/uWebSockets.cmake": "database", "ios_spm/Package.swift":scan_changes_in_file, "Android/firebase_dependencies.gradle":scan_changes_in_file, "release_build_files/Android/firebase_dependencies.gradle":scan_changes_in_file, diff --git a/scripts/git/patches/uWebSockets/0001-fix-want-write-and-sprintf-deprecation.patch b/scripts/git/patches/uWebSockets/0001-fix-want-write-and-sprintf-deprecation.patch deleted file mode 100644 index 793240d324..0000000000 --- a/scripts/git/patches/uWebSockets/0001-fix-want-write-and-sprintf-deprecation.patch +++ /dev/null @@ -1,73 +0,0 @@ -From 300161c1e2e1414fdd8df7ae6f036a6213c3672f Mon Sep 17 00:00:00 2001 -From: "drsanta@google.com" -Date: Thu, 8 Dec 2022 18:52:28 -0500 -Subject: [PATCH] fix want write and sprintf deprecation - ---- - src/HTTPSocket.h | 2 +- - src/Socket.h | 34 ++++++++++++++++++---------------- - 2 files changed, 19 insertions(+), 17 deletions(-) - -diff --git a/src/HTTPSocket.h b/src/HTTPSocket.h -index 5cc7a7f..8b1ab20 100644 ---- a/src/HTTPSocket.h -+++ b/src/HTTPSocket.h -@@ -204,7 +204,7 @@ struct HttpResponse { - - static size_t transform(const char *src, char *dst, size_t length, TransformData transformData) { - // todo: sprintf is extremely slow -- int offset = transformData.hasHead ? 0 : std::sprintf(dst, "HTTP/1.1 200 OK\r\nContent-Length: %u\r\n\r\n", (unsigned int) length); -+ int offset = transformData.hasHead ? 0 : std::snprintf(dst, length, "HTTP/1.1 200 OK\r\nContent-Length: %u\r\n\r\n", (unsigned int) length); - memcpy(dst + offset, src, length); - return length + offset; - } -diff --git a/src/Socket.h b/src/Socket.h -index 2179ff8..521e798 100644 ---- a/src/Socket.h -+++ b/src/Socket.h -@@ -306,24 +306,26 @@ protected: - if (messageQueue.empty()) { - - if (ssl) { -- sent = SSL_write(ssl, message->data, message->length); -- if (sent == (ssize_t) message->length) { -- wasTransferred = false; -- return true; -- } else if (sent < 0) { -- switch (SSL_get_error(ssl, sent)) { -- case SSL_ERROR_WANT_READ: -- break; -- case SSL_ERROR_WANT_WRITE: -- if ((getPoll() & UV_WRITABLE) == 0) { -- setPoll(getPoll() | UV_WRITABLE); -- changePoll(this); -+ /* BEG Patched by firebase-cpp-sdk 00001-fix-want-write-crash.patch */ -+ bool continue_loop = true; -+ do { -+ sent = SSL_write(ssl, message->data, message->length); -+ if (sent == (ssize_t) message->length) { -+ wasTransferred = false; -+ return true; -+ } else if (sent < 0) { -+ switch (SSL_get_error(ssl, sent)) { -+ case SSL_ERROR_WANT_READ: -+ continue_loop = false; -+ break; -+ case SSL_ERROR_WANT_WRITE: -+ break; -+ default: -+ return false; - } -- break; -- default: -- return false; - } -- } -+ } while (continue_loop); -+ /* END Patched by firebase-cpp-sdk 00001-fix-want-write-crash.patch */ - } else { - sent = ::send(getFd(), message->data, message->length, MSG_NOSIGNAL); - if (sent == (ssize_t) message->length) { --- -2.39.0.rc1.256.g54fd8350bd-goog -