From cf5d91b5ef29f6de67d78cf68a62c053907cbffa Mon Sep 17 00:00:00 2001 From: zortos293 <65777760+zortos293@users.noreply.github.com> Date: Thu, 29 Oct 2026 19:23:52 +0000 Subject: [PATCH 1/3] Complete browser-session launch and signaling recovery Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- app/src/WebSocketClient.cpp | 72 ++++- app/src/WebSocketClient.hpp | 8 +- app/src/gfn/cloud_session.cpp | 214 +++++-------- app/src/gfn/cloud_session_internal.hpp | 8 + app/src/gfn/cloud_session_protocol.cpp | 374 ++++++++++++++++------ app/src/gfn/cloud_session_state.cpp | 56 ++-- app/src/gfn_client.hpp | 6 + app/src/models.hpp | 11 +- app/src/signaling_client.cpp | 4 +- app/src/signaling_client.hpp | 5 +- app/src/ui_helpers.cpp | 10 +- app/src/webrtc/negotiation.cpp | 9 + app/src/webrtc/session.cpp | 58 +--- app/src/webrtc/signaling.cpp | 138 ++++++++ app/src/webrtc/startup_timeout_policy.hpp | 4 +- app/src/webrtc_session.hpp | 6 + scripts/test-host.sh | 5 + scripts/test-streaming-host.sh | 12 + tests/cloud_session_allocation_test.cpp | 43 +++ tests/cloud_session_recovery_test.cpp | 244 ++++++++++++++ tests/webrtc_signaling_test.cpp | 231 +++++++++++++ tests/websocket_client_test.cpp | 213 +++--------- tests/websocket_test_transport.hpp | 191 +++++++++++ 23 files changed, 1417 insertions(+), 505 deletions(-) create mode 100644 app/src/webrtc/signaling.cpp create mode 100644 tests/cloud_session_allocation_test.cpp create mode 100644 tests/cloud_session_recovery_test.cpp create mode 100644 tests/webrtc_signaling_test.cpp create mode 100644 tests/websocket_test_transport.hpp diff --git a/app/src/WebSocketClient.cpp b/app/src/WebSocketClient.cpp index ae3a5ae2..0ea68006 100644 --- a/app/src/WebSocketClient.cpp +++ b/app/src/WebSocketClient.cpp @@ -67,6 +67,14 @@ bool starts_with(const std::string& value, const char* prefix) return value.rfind(prefix, 0) == 0; } +bool transient_curl_error(CURLcode error) +{ + return error == CURLE_COULDNT_RESOLVE_HOST || error == CURLE_COULDNT_CONNECT || + error == CURLE_OPERATION_TIMEDOUT || error == CURLE_SEND_ERROR || + error == CURLE_RECV_ERROR || error == CURLE_GOT_NOTHING || + error == CURLE_SSL_CONNECT_ERROR; +} + } // namespace WebSocketClient::WebSocketClient(const std::string& url) : url_(url) {} @@ -76,11 +84,18 @@ WebSocketClient::~WebSocketClient() { close_transport(); } -bool WebSocketClient::connect() { +bool WebSocketClient::connect(Clock::time_point deadline) { disconnect(); close_transport(); last_error_.clear(); + failure_kind_ = FailureKind::Terminal; + deadline = std::min(deadline, Clock::now() + std::chrono::seconds(21)); + if (Clock::now() >= deadline) { + last_error_ = "Signaling connection deadline expired"; + failure_kind_ = FailureKind::Transient; + return false; + } curl_ = curl_easy_init(); if (!curl_) { last_error_ = "Could not initialize WebSocket transport"; @@ -126,11 +141,15 @@ bool WebSocketClient::connect() { close_transport(); return false; } - curl_easy_setopt(curl_, CURLOPT_CONNECTTIMEOUT, 15L); + const long remaining_ms = std::max(1L, static_cast( + std::chrono::duration_cast(deadline - Clock::now()).count())); + curl_easy_setopt(curl_, CURLOPT_CONNECTTIMEOUT_MS, std::min(15000L, remaining_ms)); + curl_easy_setopt(curl_, CURLOPT_TIMEOUT_MS, remaining_ms); curl_easy_setopt(curl_, CURLOPT_NOSIGNAL, 1L); CURLcode res = curl_easy_perform(curl_); if (res != CURLE_OK) { + failure_kind_ = transient_curl_error(res) ? FailureKind::Transient : FailureKind::Terminal; last_error_ = "Perform Failed: " + std::string(curl_easy_strerror(res)); brls::Logger::error("WebSocket connect failed: {}", curl_easy_strerror(res)); close_transport(); @@ -158,7 +177,7 @@ bool WebSocketClient::connect() { } handshake += "\r\n"; - if (!send_handshake(reinterpret_cast(handshake.data()), handshake.size())) { + if (!send_handshake(reinterpret_cast(handshake.data()), handshake.size(), deadline)) { close_transport(); return false; } @@ -168,9 +187,11 @@ bool WebSocketClient::connect() { size_t rcvd = 0; std::string response; - const auto receive_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + const auto receive_deadline = std::min(deadline, Clock::now() + std::chrono::seconds(5)); while (std::chrono::steady_clock::now() < receive_deadline) { res = curl_easy_recv(curl_, buf, sizeof(buf) - 1, &rcvd); + if (Clock::now() >= receive_deadline) + break; if (res == CURLE_OK && rcvd > 0) { response.append(buf, rcvd); if (response.size() > kMaximumHandshakeBytes) { @@ -187,22 +208,37 @@ bool WebSocketClient::connect() { rx_buffer_.insert(rx_buffer_.end(), response.begin() + body_start, response.end()); } connected_ = true; + failure_kind_ = FailureKind::None; brls::Logger::info("WebSocket connected!"); return true; } else { last_error_ = "Invalid WebSocket upgrade response"; + if ((response.starts_with("HTTP/1.1 ") || response.starts_with("HTTP/1.0 ")) && + response.size() >= 13 && (response[12] == ' ' || response[12] == '\r')) { + const std::string status = response.substr(9, 3); + if (status.find_first_not_of("0123456789") == std::string::npos && status != "101") { + last_error_ = "WebSocket upgrade rejected http=" + status; + if (status == "429" || status == "500" || status == "502" || + status == "503" || status == "504") + failure_kind_ = FailureKind::Transient; + } + } brls::Logger::error("Invalid WebSocket upgrade response"); break; } } else if (res == CURLE_AGAIN) { - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + std::this_thread::sleep_until(std::min(receive_deadline, Clock::now() + std::chrono::milliseconds(100))); } else { + failure_kind_ = FailureKind::Transient; last_error_ = "Recv error: " + std::string(curl_easy_strerror(res)); brls::Logger::error("WebSocket recv error: {}", curl_easy_strerror(res)); break; } } - if (last_error_.empty()) last_error_ = "Handshake timeout"; + if (last_error_.empty()) { + last_error_ = "Handshake timeout"; + failure_kind_ = FailureKind::Transient; + } close_transport(); return false; } @@ -258,33 +294,37 @@ bool WebSocketClient::drain_outgoing() { last_error_ = "Send stalled"; else last_error_ = "Send error: " + std::string(curl_easy_strerror(error)); + failure_kind_ = FailureKind::Transient; close_transport(); return false; } -bool WebSocketClient::send_handshake(const uint8_t* data, size_t length) { +bool WebSocketClient::send_handshake(const uint8_t* data, size_t length, Clock::time_point deadline) { if (!curl_) return false; size_t total_sent = 0; - const auto deadline = std::chrono::steady_clock::now() + opennow::websocket::WriteQueue::SendTimeout; + deadline = std::min(deadline, Clock::now() + opennow::websocket::WriteQueue::SendTimeout); while (total_sent < length) { if (std::chrono::steady_clock::now() >= deadline) { last_error_ = "Send timeout"; + failure_kind_ = FailureKind::Transient; return false; } size_t sent = 0; CURLcode res = curl_easy_send(curl_, data + total_sent, length - total_sent, &sent); if (res == CURLE_AGAIN) { - std::this_thread::sleep_for(std::chrono::milliseconds(10)); + std::this_thread::sleep_until(std::min(deadline, Clock::now() + std::chrono::milliseconds(10))); continue; } if (res != CURLE_OK) { last_error_ = "Send error: " + std::string(curl_easy_strerror(res)); + failure_kind_ = transient_curl_error(res) ? FailureKind::Transient : FailureKind::Terminal; connected_ = false; return false; } if (sent == 0) { last_error_ = "Send stalled"; + failure_kind_ = FailureKind::Transient; connected_ = false; return false; } @@ -301,6 +341,7 @@ bool WebSocketClient::send_frame(uint8_t opcode, const uint8_t* payload, size_t const size_t header_size = length < 126 ? 6 : (length <= 0xFFFF ? 8 : 14); if (length > kMaximumSignalingPayloadBytes || !tx_queue_.can_enqueue(length + header_size)) { last_error_ = "Outgoing WebSocket queue is full"; + failure_kind_ = FailureKind::Terminal; close_transport(); return false; } @@ -359,6 +400,7 @@ void WebSocketClient::poll() { (control && (!final || payload_len_7 > 125)) || (!control && ((opcode == 0) != (fragmented_opcode_ != 0)))) { last_error_ = "Invalid incoming WebSocket frame"; + failure_kind_ = FailureKind::Terminal; close_transport(); return; } @@ -379,6 +421,7 @@ void WebSocketClient::poll() { } if (big_len > static_cast(std::numeric_limits::max())) { last_error_ = "Incoming WebSocket frame is too large"; + failure_kind_ = FailureKind::Terminal; close_transport(); return; } @@ -389,6 +432,7 @@ void WebSocketClient::poll() { payload_len > std::numeric_limits::max() - header_size || (!control && payload_len > kMaximumSignalingPayloadBytes - fragmented_message_.size())) { last_error_ = "Incoming WebSocket frame is too large"; + failure_kind_ = FailureKind::Terminal; close_transport(); return; } @@ -411,11 +455,10 @@ void WebSocketClient::poll() { fragmented_opcode_ = opcode; fragmented_message_.append(payload.begin(), payload.end()); if (final) { - const bool text = fragmented_opcode_ == 1; std::string message = std::move(fragmented_message_); fragmented_message_.clear(); fragmented_opcode_ = 0; - if (text && on_message_) + if (on_message_) on_message_(message); } } else if (opcode == 0x08) { // Close frame @@ -425,6 +468,10 @@ void WebSocketClient::poll() { last_error_ = "Signaling closed"; if (close_code != 0) last_error_ += " code=" + std::to_string(close_code); + failure_kind_ = close_code == 0 || close_code == 1000 || close_code == 1001 || + close_code == 1006 || close_code == 1011 || close_code == 1012 || + close_code == 1013 || close_code == 1014 || close_code == 4001 + ? FailureKind::Transient : FailureKind::Terminal; if (!send_frame(0x88, payload.data(), payload.size())) return; connected_ = false; @@ -445,6 +492,7 @@ void WebSocketClient::poll() { if (frames_processed >= kMaximumFramesPerPoll) return; last_error_ = pending_receive_error_; + failure_kind_ = FailureKind::Transient; close_transport(); return; } @@ -460,6 +508,7 @@ void WebSocketClient::poll() { if (res == CURLE_OK && rcvd > 0) { if (rx_buffer_.size() > kMaximumSignalingPayloadBytes + 14 - rcvd) { last_error_ = "Incoming WebSocket buffer is too large"; + failure_kind_ = FailureKind::Terminal; close_transport(); return; } @@ -473,6 +522,7 @@ void WebSocketClient::poll() { } if (!pending_receive_error_.empty() && read_this_poll == 0) { last_error_ = pending_receive_error_; + failure_kind_ = FailureKind::Transient; close_transport(); } break; diff --git a/app/src/WebSocketClient.hpp b/app/src/WebSocketClient.hpp index c73e69d2..dcdf5e73 100644 --- a/app/src/WebSocketClient.hpp +++ b/app/src/WebSocketClient.hpp @@ -10,11 +10,13 @@ class WebSocketClient { public: using MessageCallback = std::function; + using Clock = std::chrono::steady_clock; + enum class FailureKind { None, Transient, Terminal }; WebSocketClient(const std::string& url); ~WebSocketClient(); - bool connect(); + bool connect(Clock::time_point deadline = Clock::time_point::max()); void disconnect(); void poll(); void send_message(const std::string& msg); @@ -23,6 +25,7 @@ class WebSocketClient { void set_custom_headers(const std::vector& headers) { custom_headers_ = headers; } bool is_connected() const { return connected_; } std::string get_last_error() const { return last_error_; } + FailureKind failure_kind() const { return failure_kind_; } private: std::string url_; @@ -32,6 +35,7 @@ class WebSocketClient { bool connected_ = false; bool closing_ = false; std::string last_error_; + FailureKind failure_kind_ = FailureKind::None; std::string pending_receive_error_; std::vector rx_buffer_; std::string fragmented_message_; @@ -40,6 +44,6 @@ class WebSocketClient { void close_transport(); bool drain_outgoing(); - bool send_handshake(const uint8_t* data, size_t length); + bool send_handshake(const uint8_t* data, size_t length, Clock::time_point deadline); bool send_frame(uint8_t opcode, const uint8_t* payload, size_t length); }; diff --git a/app/src/gfn/cloud_session.cpp b/app/src/gfn/cloud_session.cpp index 75091d93..d3dfb104 100644 --- a/app/src/gfn/cloud_session.cpp +++ b/app/src/gfn/cloud_session.cpp @@ -141,76 +141,16 @@ SessionInfo GfnClient::StartSession(AuthSession& session, const std::string& lau AppendSessionTraceLog("START response HTTP " + std::to_string(response.status_code)); AppendSessionTraceLog("START response body:\n" + JsonForTrace(response.body)); - JsonPtr root(nullptr, &json_decref); + SessionInfo info = ParseSessionResponse(response, "StartSession", {}, streaming_base_url); + RememberActiveCloudSession( + session, info.session_id, launch_app_id, info.streaming_base_url); + ad_session_key_.clear(); try { - root = LoadJson(response.body); - } catch (const std::exception& e) { - AppendSessionTraceLog(std::string("START JSON parse error: ") + e.what()); - } - - const bool app_patching = root && IsAppPatchingResponse(root.get()); - if (root) - { - json_t* req_status = json_object_get(root.get(), "requestStatus"); - if (req_status && GetInteger(req_status, "statusCode") != 1 && !app_patching) - { - const std::string error = BuildCloudMatchErrorMessage("StartSession", response.status_code, root.get()); - AppendSessionTraceLog("START api error: " + error); - throw std::runtime_error(error); - } - } - - if (response.status_code != 200 && response.status_code != 202 && !app_patching) - { - const std::string error = BuildCloudMatchErrorMessage("StartSession", response.status_code, root.get()); - AppendSessionTraceLog("START http error: " + error); - throw std::runtime_error(error); - } - - if (!root) { - AppendSessionTraceLog("START failed: empty or invalid JSON"); - throw std::runtime_error("StartSession failed: Empty or invalid JSON" + SessionTraceHint()); - } - - SessionInfo info; - info.app_patching = app_patching; - CollectIceServersFromObject(root.get(), info); - - json_t* sess = json_object_get(root.get(), "session"); - if (sess) - { - json_t* s_id = json_object_get(sess, "sessionId"); - if (s_id && json_is_string(s_id)) - info.session_id = json_string_value(s_id); - - json_t* status = json_object_get(sess, "status"); - if (status) - info.status = ParseSessionStatus(status); - - json_t* qp = json_object_get(sess, "queuePosition"); - if (qp && json_is_integer(qp)) - info.queue_position = json_integer_value(qp); - - ApplySeatSetupInfo(sess, info); - - ApplySessionNetworkInfo(sess, info); - } - - // Check root for queuePosition - json_t* root_qp = json_object_get(root.get(), "queuePosition"); - if (root_qp && json_is_integer(root_qp)) { - info.queue_position = json_integer_value(root_qp); + AnswerQueueAds(session, info, headers); + } catch (...) { + try { StopSession(session, info.session_id); } catch (...) {} + throw; } - - if (info.session_id.empty()) { - AppendSessionTraceLog("START failed: session ID missing after successful response"); - throw std::runtime_error("Failed to parse session ID from response" + SessionTraceHint()); - } - - if (app_patching) - AppendSessionTraceLog("START pending reason=app_patching; polling created session"); - RememberActiveCloudSession( - session, info.session_id, launch_app_id, streaming_base_url); AppendSessionTraceLog("START parsed: " + SessionInfoTraceSummary(info)); return info; } @@ -255,76 +195,77 @@ SessionInfo GfnClient::PollSession(AuthSession& session, const std::string& sess AppendSessionTraceLog("POLL response HTTP " + std::to_string(response.status_code)); AppendSessionTraceLog("POLL response body:\n" + JsonForTrace(response.body)); - JsonPtr root(nullptr, &json_decref); - try { - root = LoadJson(response.body); - } catch (const std::exception& e) { - AppendSessionTraceLog(std::string("POLL JSON parse error: ") + e.what()); - } - - const bool app_patching = root && IsAppPatchingResponse(root.get()); - if (root) - { - json_t* req_status = json_object_get(root.get(), "requestStatus"); - if (req_status && GetInteger(req_status, "statusCode") != 1 && !app_patching) - { - const std::string error = BuildCloudMatchErrorMessage("PollSession", response.status_code, root.get()); - AppendSessionTraceLog("POLL api error: " + error); - throw std::runtime_error(error); - } - } - - if (response.status_code != 200 && !app_patching) - { - const std::string error = BuildCloudMatchErrorMessage("PollSession", response.status_code, root.get()); - AppendSessionTraceLog("POLL http error: " + error); - throw std::runtime_error(error); - } + SessionInfo info = ParseSessionResponse(response, "PollSession", session_id, streaming_base_url); + UpdateActiveCloudSessionStreamingBaseUrl(session, session_id, info.streaming_base_url); + AnswerQueueAds(session, info, headers); + AppendSessionTraceLog("POLL parsed: " + SessionInfoTraceSummary(info)); + return info; +} - if (!root) { - AppendSessionTraceLog("POLL failed: empty or invalid JSON"); - throw std::runtime_error("PollSession failed: Empty or invalid JSON" + SessionTraceHint()); +void GfnClient::AnswerQueueAds(AuthSession& auth, SessionInfo& info, + std::vector headers) const +{ + if (info.status > 3 && info.status != 6) + return; + const std::string key = auth.user.user_id + ":" + info.session_id; + if (ad_session_key_ != key) { + ad_session_key_ = key; + ad_attempts_.clear(); + ad_pending_polls_ = 0; } - - // Keep diagnostics useful without persisting session tokens returned by CloudMatch. - const std::string redacted_session = JsonForTrace(response.body); - FILE* session_dump = fopen("sdmc:/switch/SwitchNOW/switchnow_session.json", "w"); - if (session_dump) { - fputs(redacted_session.c_str(), session_dump); - fclose(session_dump); + for (const std::string& id : info.ad_ids) { + if (ad_attempts_.size() >= 32 && !ad_attempts_.contains(id)) + throw std::runtime_error("NVIDIA returned too many queue ads."); + ad_attempts_.try_emplace(id, 0); } - - SessionInfo info; - info.session_id = session_id; - info.app_patching = app_patching; - CollectIceServersFromObject(root.get(), info); - - json_t* sess = json_object_get(root.get(), "session"); - if (sess) - { - json_t* status = json_object_get(sess, "status"); - if (status) - info.status = ParseSessionStatus(status); - - json_t* qp = json_object_get(sess, "queuePosition"); - if (qp && json_is_integer(qp)) - info.queue_position = json_integer_value(qp); - - ApplySeatSetupInfo(sess, info); - - ApplySessionNetworkInfo(sess, info); + bool pending = info.ads_pending && ad_attempts_.empty(); + for (auto& [id, attempts] : ad_attempts_) { + if (attempts < 0) + continue; + if (attempts >= 3) + throw std::runtime_error("Unable to report unsupported NVIDIA queue ad playback. Please try again."); + ++attempts; + HttpResponse response; + const std::string body = BuildQueueAdCancellationBody(id); + const std::string url = info.streaming_base_url + "v2/session/" + info.session_id; + try { + response = http_client_.Request("PUT", url, kCloudMatchUserAgent, headers, body); + } catch (const std::exception&) { + pending = true; + break; + } + if (response.status_code == 401) { + auth = ForceRefreshSavedSession(auth); + headers[0] = "Authorization: GFNJWT " + ResolveSessionJwt(auth); + try { + response = http_client_.Request("PUT", url, kCloudMatchUserAgent, headers, body); + } catch (const std::exception&) { + pending = true; + break; + } + } + if (CloudMatchResponseSucceeded(response)) { + attempts = -1; + AppendSessionTraceLog("QUEUEAD reported unsupported playback"); + continue; + } + if (response.status_code >= 400 && response.status_code < 500 && response.status_code != 429) + throw std::runtime_error("NVIDIA rejected the queue ad cancellation (HTTP " + + std::to_string(response.status_code) + ")."); + pending = true; + break; } - - // Sometimes queuePosition is at the root level! - json_t* root_qp = json_object_get(root.get(), "queuePosition"); - if (root_qp && json_is_integer(root_qp)) { - info.queue_position = json_integer_value(root_qp); + info.ads_pending = pending || std::any_of(ad_attempts_.begin(), ad_attempts_.end(), + [](const auto& entry) { return entry.second >= 0; }); + if (std::any_of(ad_attempts_.begin(), ad_attempts_.end(), + [](const auto& entry) { return entry.second >= 3; })) + throw std::runtime_error("Unable to report unsupported NVIDIA queue ad playback. Please try again."); + if (info.ads_pending) { + if (++ad_pending_polls_ >= 12) + throw std::runtime_error("NVIDIA queue ads did not become available for cancellation in time."); + } else { + ad_pending_polls_ = 0; } - - if (app_patching) - AppendSessionTraceLog("POLL pending reason=app_patching"); - AppendSessionTraceLog("POLL parsed: " + SessionInfoTraceSummary(info)); - return info; } void GfnClient::StopSession(AuthSession& session, const std::string& session_id) const @@ -368,9 +309,14 @@ void GfnClient::StopSession(AuthSession& session, const std::string& session_id) } AppendSessionTraceLog("STOP response HTTP " + std::to_string(response.status_code)); AppendSessionTraceLog("STOP response body:\n" + JsonForTrace(response.body)); - if ((response.status_code >= 200 && response.status_code < 300) || response.status_code == 404) + if (CloudMatchResponseSucceeded(response) || response.status_code == 404) { ForgetActiveCloudSession(session_id); + if (ad_session_key_ == session.user.user_id + ":" + session_id) { + ad_session_key_.clear(); + ad_attempts_.clear(); + ad_pending_polls_ = 0; + } return; } diff --git a/app/src/gfn/cloud_session_internal.hpp b/app/src/gfn/cloud_session_internal.hpp index d50387ce..38ea6499 100644 --- a/app/src/gfn/cloud_session_internal.hpp +++ b/app/src/gfn/cloud_session_internal.hpp @@ -17,6 +17,12 @@ inline constexpr const char* kCloudMatchUserAgent = void ApplySessionNetworkInfo(json_t* session, SessionInfo& info); void CollectIceServersFromObject(json_t* object, SessionInfo& info); +int WireInteger(json_t* value, int fallback = -1); +SessionInfo ParseSessionResponse(const HttpResponse& response, const std::string& stage, + const std::string& expected_id, const std::string& base_url); +std::string ResolveSessionControlBaseUrl(json_t* session, const std::string& fallback); +bool CloudMatchResponseSucceeded(const HttpResponse& response); +std::string BuildQueueAdCancellationBody(const std::string& id); void ResetSessionTraceLog(const std::string& reason); void AppendSessionTraceLog(const std::string& line); @@ -35,6 +41,8 @@ void RememberActiveCloudSession( const std::string& streaming_base_url); std::string LoadActiveCloudSessionStreamingBaseUrl( const AuthSession& auth, const std::string& session_id); +void UpdateActiveCloudSessionStreamingBaseUrl( + const AuthSession& auth, const std::string& session_id, const std::string& base_url); std::string ResolveConfiguredStreamingBaseUrl(const AuthSession& session); std::string LoadActiveCloudSessionId(const AuthSession& auth); void ForgetActiveCloudSession(const std::string& session_id); diff --git a/app/src/gfn/cloud_session_protocol.cpp b/app/src/gfn/cloud_session_protocol.cpp index b91bd967..97c6ad6c 100644 --- a/app/src/gfn/cloud_session_protocol.cpp +++ b/app/src/gfn/cloud_session_protocol.cpp @@ -2,10 +2,21 @@ #include "app_launch_mode_policy.hpp" #include +#include +#include +#include #include #include #include #include +#include + +#ifdef _WIN32 +#include +#include +#else +#include +#endif namespace opennow { @@ -21,106 +32,157 @@ bool HasPrefix(const std::string& value, const char* prefix) return value.rfind(prefix, 0) == 0; } -std::string ExtractHostFromUrlLike(const std::string& value) +bool ValidHost(const std::string& host) { - size_t start = value.find("://"); - if (start == std::string::npos) - return ""; - - start += 3; - const size_t path = value.find('/', start); - std::string host_port = value.substr(start, path == std::string::npos ? std::string::npos : path - start); - - if (!host_port.empty() && host_port.front() == '[') { - const size_t close = host_port.find(']'); - return close == std::string::npos ? "" : host_port.substr(1, close - 1); + if (host.empty() || host.size() > 253) + return false; + std::array ipv6 {}; + if (host.find(':') != std::string::npos) { + return inet_pton(AF_INET6, host.c_str(), ipv6.data()) == 1 && + std::any_of(ipv6.begin(), ipv6.end(), [](unsigned char byte) { return byte != 0; }); } - - const size_t colon = host_port.find(':'); - return colon == std::string::npos ? host_port : host_port.substr(0, colon); + in_addr ipv4 {}; + if (inet_pton(AF_INET, host.c_str(), &ipv4) == 1) + return ipv4.s_addr != 0; + if (std::all_of(host.begin(), host.end(), [](unsigned char ch) { return std::isdigit(ch) || ch == '.'; })) + return false; + size_t start = 0; + while (start < host.size()) { + const size_t end = host.find('.', start); + const std::string label = host.substr(start, end - start); + if (label.empty() || label.size() > 63 || label.front() == '-' || label.back() == '-' || + !std::all_of(label.begin(), label.end(), [](unsigned char ch) { + return ch < 128 && (std::isalnum(ch) || ch == '-'); + })) + return false; + if (end == std::string::npos) + return true; + start = end + 1; + } + return false; } -int ExtractPortFromUrlLike(const std::string& value) +struct Endpoint { - size_t start = value.find("://"); - if (start == std::string::npos) - return 0; + std::string host; + int port = 0; +}; - start += 3; - const size_t path = value.find('/', start); - const size_t host_end = path == std::string::npos ? value.size() : path; - - size_t colon = std::string::npos; - if (start < value.size() && value[start] == '[') { - const size_t close = value.find(']', start); - if (close != std::string::npos && close + 1 < host_end && value[close + 1] == ':') - colon = close + 1; +Endpoint ParseEndpoint(const std::string& value) +{ + const size_t scheme = value.find("://"); + if (scheme == std::string::npos || value.find_first_of("@#\\") != std::string::npos || + std::any_of(value.begin(), value.end(), [](unsigned char ch) { return ch <= 0x20 || ch >= 0x7f; })) + return {}; + const size_t start = scheme + 3; + const size_t end = value.find_first_of("/?", start); + const std::string authority = value.substr(start, end - start); + std::string host; + size_t port_start = std::string::npos; + if (!authority.empty() && authority.front() == '[') { + const size_t close = authority.find(']'); + if (close == std::string::npos) + return {}; + host = authority.substr(1, close - 1); + if (host.find(':') == std::string::npos) + return {}; + if (close + 1 < authority.size()) { + if (authority[close + 1] != ':') + return {}; + port_start = close + 2; + } } else { - colon = value.find(':', start); - if (colon >= host_end) - colon = std::string::npos; + const size_t colon = authority.find(':'); + host = authority.substr(0, colon); + if (colon != std::string::npos) + port_start = colon + 1; } + if (!ValidHost(host)) + return {}; + int port = 443; + if (port_start != std::string::npos) { + const char* first = authority.data() + port_start; + const char* last = authority.data() + authority.size(); + const auto parsed = std::from_chars(first, last, port); + if (parsed.ec != std::errc() || parsed.ptr != last || port <= 0 || port > 65535) + return {}; + } + return {host, port}; +} - if (colon == std::string::npos || colon + 1 >= host_end) - return 0; - - char* end = nullptr; - long port = std::strtol(value.c_str() + colon + 1, &end, 10); - if (port <= 0 || port > 65535) - return 0; +std::string ValidSignalingUrl(std::string value) +{ + if (!HasPrefix(value, "wss://") && !HasPrefix(value, "https://")) + return {}; + if (ParseEndpoint(value).host.empty()) + return {}; + if (HasPrefix(value, "https://")) + value.replace(0, 5, "wss"); + return value; +} - return static_cast(port); +bool IsNativeConnection(json_t* connection) +{ + const int protocol = WireInteger(json_object_get(connection, "appLevelProtocol")); + const std::string resource = GetString(connection, "resourcePath"); + return protocol == 1 || protocol == 6 || WireInteger(json_object_get(connection, "usage")) == 16 || + HasPrefix(resource, "rtsp://") || HasPrefix(resource, "rtsps://"); } -std::string GetConnectionIp(json_t* connection, const std::string& fallback_ip = "") +std::string GetConnectionIp(json_t* connection) { if (!connection || !json_is_object(connection)) - return fallback_ip; + return {}; json_t* ip = json_object_get(connection, "ip"); - if (json_is_string(ip)) + if (json_is_string(ip) && ValidHost(JsonString(ip))) return JsonString(ip); - if (json_is_array(ip) && json_array_size(ip) > 0) - return JsonString(json_array_get(ip, 0)); + if (json_is_array(ip)) { + size_t index; + json_t* entry; + json_array_foreach(ip, index, entry) { + if (ValidHost(JsonString(entry))) + return JsonString(entry); + } + } const std::string resource_path = GetString(connection, "resourcePath"); - const std::string host = ExtractHostFromUrlLike(resource_path); - return host.empty() ? fallback_ip : host; + const std::string host = ParseEndpoint(resource_path).host; + return host; } int GetConnectionPort(json_t* connection) { - const int direct_port = GetInteger(connection, "port"); - if (direct_port > 0) + const int direct_port = WireInteger(json_object_get(connection, "port")); + if (direct_port > 0 && direct_port <= 65535) return direct_port; - - return ExtractPortFromUrlLike(GetString(connection, "resourcePath")); + if (json_object_get(connection, "port")) + return 0; + const std::string resource = GetString(connection, "resourcePath"); + const size_t scheme = resource.find("://"); + if (scheme == std::string::npos) + return 0; + const std::string authority = resource.substr(scheme + 3, resource.find_first_of("/?", scheme + 3) - scheme - 3); + if (authority.find(':', authority.empty() || authority.front() != '[' ? 0 : authority.find(']')) == std::string::npos) + return 0; + return ParseEndpoint(resource).port; } -std::string BuildSignalingUrlFromConnection(json_t* connection, const std::string& fallback_ip) +std::string BuildSignalingUrlFromConnection(json_t* connection) { + if (IsNativeConnection(connection) || WireInteger(json_object_get(connection, "usage")) != 14) + return {}; const std::string resource_path = GetString(connection, "resourcePath"); - const std::string ip = GetConnectionIp(connection, fallback_ip); - int port = GetConnectionPort(connection); - if (port <= 0) - port = 443; - - if (HasPrefix(resource_path, "wss://")) - return resource_path; - - if (HasPrefix(resource_path, "https://")) - return "wss://" + resource_path.substr(strlen("https://")); - - if (HasPrefix(resource_path, "rtsps://") || HasPrefix(resource_path, "rtsp://")) { - const std::string host = ExtractHostFromUrlLike(resource_path); - return host.empty() ? "" : "wss://" + host + "/nvst/"; - } - - if (!resource_path.empty() && resource_path.front() == '/' && !ip.empty()) - return "wss://" + ip + ":" + std::to_string(port) + resource_path; - - return ip.empty() ? "" : "wss://" + ip + ":443/nvst/"; + if (resource_path.find("://") != std::string::npos) + return ValidSignalingUrl(resource_path); + if (!resource_path.empty() && resource_path.front() != '/') + return {}; + const std::string ip = GetConnectionIp(connection); + if (ip.empty()) + return {}; + const std::string host = ip.find(':') == std::string::npos ? ip : "[" + ip + "]"; + return ValidSignalingUrl("wss://" + host + ":443" + (resource_path.empty() ? "/nvst/" : resource_path)); } void AppendIceServer(SessionInfo& info, std::string url, std::string username, std::string credential) @@ -233,16 +295,7 @@ void ApplySessionNetworkInfo(json_t* sess, SessionInfo& info) if (!session_token.empty()) info.session_token = session_token; - const std::string server_ip = GetString(sess, "serverIp"); - if (!server_ip.empty()) - info.server_ip = server_ip; - - json_t* session_control = json_object_get(sess, "sessionControlInfo"); - CollectIceServersFromObject(session_control, info); - if (info.server_ip.empty() && session_control) - info.server_ip = GetString(session_control, "ip"); - - const std::string signaling_url = GetString(sess, "signalingUrl"); + const std::string signaling_url = ValidSignalingUrl(GetString(sess, "signalingUrl")); if (!signaling_url.empty()) info.signaling_url = signaling_url; @@ -258,13 +311,14 @@ void ApplySessionNetworkInfo(json_t* sess, SessionInfo& info) json_array_foreach(connection_info_array, index, conn_info) { CollectIceServersFromObject(conn_info, info); - if (GetInteger(conn_info, "usage") != priority) + if (WireInteger(json_object_get(conn_info, "usage")) != priority || IsNativeConnection(conn_info)) + continue; + if (priority == 14 && BuildSignalingUrlFromConnection(conn_info).empty()) continue; - const std::string fallback_ip = priority == 14 ? info.server_ip : ""; - const std::string ip = GetConnectionIp(conn_info, fallback_ip); + const std::string ip = GetConnectionIp(conn_info); const int port = GetConnectionPort(conn_info); - if (!ip.empty() && port > 0 && (priority != 14 || port > best_port)) { + if (!ip.empty() && port > 0 && (best_ip.empty() || (priority == 14 && port > best_port))) { best_ip = ip; best_port = port; } @@ -278,23 +332,146 @@ void ApplySessionNetworkInfo(json_t* sess, SessionInfo& info) } if (info.signaling_url.empty()) { - size_t index; - json_t* conn_info; - json_array_foreach(connection_info_array, index, conn_info) { - CollectIceServersFromObject(conn_info, info); - - if (GetInteger(conn_info, "usage") != 14) - continue; - - info.signaling_url = BuildSignalingUrlFromConnection(conn_info, info.server_ip); + for (bool explicit_url : {true, false}) { + size_t index; + json_t* conn_info; + json_array_foreach(connection_info_array, index, conn_info) { + CollectIceServersFromObject(conn_info, info); + + if (WireInteger(json_object_get(conn_info, "usage")) != 14) + continue; + const std::string resource = GetString(conn_info, "resourcePath"); + if ((HasPrefix(resource, "wss://") || HasPrefix(resource, "https://")) != explicit_url) + continue; + info.signaling_url = BuildSignalingUrlFromConnection(conn_info); + if (!info.signaling_url.empty()) + break; + } if (!info.signaling_url.empty()) break; } } } - if (info.signaling_url.empty() && session_control) - info.signaling_url = BuildSignalingUrlFromConnection(session_control, info.server_ip); + info.server_ip = ParseEndpoint(info.signaling_url).host; +} + +int WireInteger(json_t* value, int fallback) +{ + if (json_is_integer(value)) { + const json_int_t number = json_integer_value(value); + return number >= INT32_MIN && number <= INT32_MAX ? static_cast(number) : fallback; + } + if (!json_is_string(value)) + return fallback; + const std::string text = JsonString(value); + int number = 0; + const auto parsed = std::from_chars(text.data(), text.data() + text.size(), number); + return parsed.ec == std::errc() && parsed.ptr == text.data() + text.size() ? number : fallback; +} + +std::string ResolveSessionControlBaseUrl(json_t* session, const std::string& fallback) +{ + json_t* control = json_object_get(session, "sessionControlInfo"); + const std::string host = Lowercase(GetString(control, "ip")); + json_t* port = json_object_get(control, "port"); + if (!ValidHost(host) || (port && WireInteger(port) != 443)) + return fallback; + const Endpoint origin = ParseEndpoint(fallback); + if (host == Lowercase(origin.host)) + return "https://" + (host.find(':') == std::string::npos ? host : "[" + host + "]") + "/"; + const auto nvidia_host = [](const std::string& value) { + return value.ends_with(".cloudmatchbeta.nvidiagrid.net") || + value.ends_with(".cloudmatch.nvidiagrid.net") || + value.ends_with(".geforcenow.nvidiagrid.net"); + }; + if (nvidia_host(Lowercase(origin.host)) && nvidia_host(host) && HasPrefix(host, "np-")) + return "https://" + host + "/"; + return fallback; +} + +bool CloudMatchResponseSucceeded(const HttpResponse& response) +{ + if (response.status_code < 200 || response.status_code >= 300) + return false; + if (response.body.empty()) + return true; + try { + JsonPtr root = LoadJson(response.body); + return json_is_object(root.get()) && WireInteger(json_object_get( + json_object_get(root.get(), "requestStatus"), "statusCode"), -1) == 1; + } catch (...) { + return false; + } +} + +SessionInfo ParseSessionResponse(const HttpResponse& response, const std::string& stage, + const std::string& expected_id, const std::string& base_url) +{ + JsonPtr root(nullptr, &json_decref); + try { root = LoadJson(response.body); } catch (...) {} + const bool patching = IsAppPatchingResponse(root.get()); + json_t* request = json_object_get(root.get(), "requestStatus"); + if (!json_is_object(root.get()) || + (!patching && (response.status_code < 200 || response.status_code >= 300 || + WireInteger(json_object_get(request, "statusCode"), -1) != 1))) + throw std::runtime_error(BuildCloudMatchErrorMessage(stage, response.status_code, root.get())); + json_t* sess = json_object_get(root.get(), "session"); + SessionInfo info; + info.session_id = GetString(sess, "sessionId"); + if (info.session_id.empty() || info.session_id.find_first_of("/?#\\\r\n") != std::string::npos || + (!expected_id.empty() && expected_id != info.session_id)) + throw std::runtime_error(stage + " returned an invalid or different session ID."); + info.status = ParseSessionStatus(json_object_get(sess, "status")); + if (patching && !json_object_get(sess, "status")) + info.status = 1; + if (info.status < 0 || info.status > 8) + throw std::runtime_error(stage + " returned an invalid or missing session status."); + info.app_patching = patching; + info.queue_position = WireInteger(json_object_get(sess, "queuePosition"), 0); + ApplySeatSetupInfo(sess, info); + info.queue_position = WireInteger(json_object_get(root.get(), "queuePosition"), info.queue_position); + CollectIceServersFromObject(root.get(), info); + ApplySessionNetworkInfo(sess, info); + info.streaming_base_url = ResolveSessionControlBaseUrl(sess, base_url); + const auto required = [](json_t* value) { return json_is_true(value) || WireInteger(value, 0) == 1; }; + info.ads_pending = required(json_object_get(sess, "sessionAdsRequired")) || + required(json_object_get(sess, "isAdsRequired")) || + required(json_object_get(json_object_get(sess, "sessionProgress"), "isAdsRequired")); + json_t* ads = json_object_get(sess, "sessionAds"); + size_t index; + json_t* ad; + json_array_foreach(ads, index, ad) { + std::string id = GetString(ad, "adId"); + if (json_is_integer(json_object_get(ad, "adId"))) + id = std::to_string(json_integer_value(json_object_get(ad, "adId"))); + if (!id.empty() && id.size() <= 256 && + std::find(info.ad_ids.begin(), info.ad_ids.end(), id) == info.ad_ids.end()) { + if (info.ad_ids.size() >= 32) + throw std::runtime_error("NVIDIA returned too many queue ads."); + info.ad_ids.push_back(std::move(id)); + } + } + info.ads_pending = info.ads_pending || !info.ad_ids.empty(); + return info; +} + +std::string BuildQueueAdCancellationBody(const std::string& id) +{ + JsonPtr root(json_object(), &json_decref); + json_object_set_new(root.get(), "action", json_integer(6)); + json_t* updates = json_array(); + json_t* ad = json_object(); + json_object_set_new(ad, "adId", json_string(id.c_str())); + json_object_set_new(ad, "adAction", json_integer(5)); + json_object_set_new(ad, "clientTimestamp", json_integer(std::time(nullptr))); + json_object_set_new(ad, "cancelReason", json_string("FAILED_TO_REQUEST_ADS")); + json_object_set_new(ad, "watchedTimeInMs", json_integer(0)); + json_object_set_new(ad, "pausedTimeInMs", json_integer(0)); + json_object_set_new(ad, "adDurationInMs", json_integer(0)); + json_array_append_new(updates, ad); + json_object_set_new(root.get(), "adUpdates", updates); + return DumpJson(root.get()); } std::string BuildSessionBody( @@ -458,8 +635,9 @@ std::string CreateNetworkTestSession( int ParseSessionStatus(json_t* status) { - if (json_is_integer(status)) - return static_cast(json_integer_value(status)); + const int numeric = WireInteger(status, -1); + if (numeric >= 0) + return numeric; if (!json_is_string(status)) return -1; diff --git a/app/src/gfn/cloud_session_state.cpp b/app/src/gfn/cloud_session_state.cpp index f4e30b42..86663c21 100644 --- a/app/src/gfn/cloud_session_state.cpp +++ b/app/src/gfn/cloud_session_state.cpp @@ -139,7 +139,7 @@ std::string BuildCloudMatchErrorMessage(const std::string& stage, int http_statu const std::string details = CloudMatchErrorDetails(http_status, root); std::string message = stage + " failed: " + details; json_t* request_status = root ? json_object_get(root, "requestStatus") : nullptr; - if (request_status && GetInteger(request_status, "statusCode") == 81) { + if (request_status && WireInteger(json_object_get(request_status, "statusCode")) == 81) { message += "\nThis game is not available with the current GeForce NOW membership. " "If the game page says Premium, upgrade the membership to play it."; } @@ -149,7 +149,7 @@ std::string BuildCloudMatchErrorMessage(const std::string& stage, int http_statu bool IsAppPatchingResponse(json_t* root) { json_t* request_status = root ? json_object_get(root, "requestStatus") : nullptr; - if (!request_status || GetInteger(request_status, "statusCode") != 41) + if (!request_status || WireInteger(json_object_get(request_status, "statusCode")) != 41) return false; return GetString(request_status, "statusDescription").find("APP_PATCHING_STATUS") != std::string::npos; @@ -174,6 +174,19 @@ std::string GetActiveCloudSessionPath() return GetAppHome() + "/active_cloud_session.json"; } +JsonPtr LoadActiveCloudSession() +{ + for (const std::string& suffix : {std::string(), std::string(".bak")}) { + try { + JsonPtr root = LoadJson(ReadTextFile(GetActiveCloudSessionPath() + suffix)); + if (json_is_object(root.get()) && !GetString(root.get(), "session_id").empty() && + !GetString(root.get(), "user_id").empty()) + return root; + } catch (...) {} + } + return JsonPtr(nullptr, &json_decref); +} + } // namespace void RememberActiveCloudSession( @@ -206,13 +219,9 @@ std::string LoadActiveCloudSessionStreamingBaseUrl( const AuthSession& auth, const std::string& session_id) { - const std::string body = ReadTextFile(GetActiveCloudSessionPath()); - if (body.empty()) - return {}; - try { - JsonPtr root = LoadJson(body); + JsonPtr root = LoadActiveCloudSession(); if (GetString(root.get(), "user_id") != auth.user.user_id || GetString(root.get(), "session_id") != session_id) return {}; @@ -225,6 +234,19 @@ std::string LoadActiveCloudSessionStreamingBaseUrl( } } +void UpdateActiveCloudSessionStreamingBaseUrl( + const AuthSession& auth, const std::string& session_id, const std::string& base_url) +{ + JsonPtr root = LoadActiveCloudSession(); + if (GetString(root.get(), "user_id") != auth.user.user_id || + GetString(root.get(), "session_id") != session_id) + return; + if (GetString(root.get(), "streaming_base_url") == base_url) + return; + json_object_set_new(root.get(), "streaming_base_url", json_string(base_url.c_str())); + WriteJsonToFile(GetActiveCloudSessionPath(), root.get()); +} + std::string ResolveConfiguredStreamingBaseUrl(const AuthSession& session) { const StreamSettings settings = LoadStreamSettings(); @@ -237,13 +259,9 @@ std::string ResolveConfiguredStreamingBaseUrl(const AuthSession& session) std::string LoadActiveCloudSessionId(const AuthSession& auth) { - const std::string body = ReadTextFile(GetActiveCloudSessionPath()); - if (body.empty()) - return ""; - try { - JsonPtr root = LoadJson(body); + JsonPtr root = LoadActiveCloudSession(); if (GetString(root.get(), "user_id") != auth.user.user_id) return ""; return GetString(root.get(), "session_id"); @@ -256,17 +274,9 @@ std::string LoadActiveCloudSessionId(const AuthSession& auth) void ForgetActiveCloudSession(const std::string& session_id) { - const std::string body = ReadTextFile(GetActiveCloudSessionPath()); - if (!body.empty()) - { - try - { - JsonPtr root = LoadJson(body); - if (!session_id.empty() && GetString(root.get(), "session_id") != session_id) - return; - } - catch (...) {} - } + JsonPtr root = LoadActiveCloudSession(); + if (root && !session_id.empty() && GetString(root.get(), "session_id") != session_id) + return; std::remove(GetActiveCloudSessionPath().c_str()); std::remove((GetActiveCloudSessionPath() + ".bak").c_str()); diff --git a/app/src/gfn_client.hpp b/app/src/gfn_client.hpp index 82b84ec1..97f16e73 100644 --- a/app/src/gfn_client.hpp +++ b/app/src/gfn_client.hpp @@ -5,6 +5,7 @@ #include "stream_settings.hpp" #include +#include #include #include @@ -68,6 +69,11 @@ class GfnClient "NVIDIACEFClient/HEAD/debb5919f6 GFN-PC/2.0.80.173"; private: + void AnswerQueueAds(AuthSession& auth, SessionInfo& info, + std::vector headers) const; + mutable std::string ad_session_key_; + mutable std::map ad_attempts_; + mutable int ad_pending_polls_ = 0; HttpClient http_client_; std::string client_id_; }; diff --git a/app/src/models.hpp b/app/src/models.hpp index aa7b6fdc..58eee15c 100644 --- a/app/src/models.hpp +++ b/app/src/models.hpp @@ -152,7 +152,10 @@ struct SessionInfo int status = 0; int queue_position = 0; bool app_patching = false; - + std::string streaming_base_url; + std::vector ad_ids; + bool ads_pending = false; + // WebRTC / GFN Streaming details std::string session_token; std::string server_ip; @@ -160,6 +163,12 @@ struct SessionInfo std::string media_ip; int media_port = 0; std::vector ice_servers; + + bool ready_for_stream() const + { + return (status == 2 || status == 3) && !app_patching && + !ads_pending && !signaling_url.empty(); + } }; } // namespace opennow diff --git a/app/src/signaling_client.cpp b/app/src/signaling_client.cpp index 764f54f0..aa44288d 100644 --- a/app/src/signaling_client.cpp +++ b/app/src/signaling_client.cpp @@ -10,7 +10,7 @@ SignalingClient::~SignalingClient() { } } -bool SignalingClient::connect() { +bool SignalingClient::connect(WebSocketClient::Clock::time_point deadline) { if (!ws_) return false; ws_->set_on_message([this](const std::string& msg) { @@ -19,7 +19,7 @@ bool SignalingClient::connect() { } }); - if (ws_->connect()) { + if (ws_->connect(deadline)) { return true; } return false; diff --git a/app/src/signaling_client.hpp b/app/src/signaling_client.hpp index c03f97c1..20ad8171 100644 --- a/app/src/signaling_client.hpp +++ b/app/src/signaling_client.hpp @@ -9,7 +9,7 @@ class SignalingClient { SignalingClient(const std::string& url); ~SignalingClient(); - bool connect(); + bool connect(WebSocketClient::Clock::time_point deadline = WebSocketClient::Clock::time_point::max()); void send_message(const std::string& msg); void set_on_message(std::function cb) { on_message_ = cb; } void poll(); @@ -20,6 +20,9 @@ class SignalingClient { } std::string get_last_error() const { return ws_ ? ws_->get_last_error() : "No WS object"; } + WebSocketClient::FailureKind failure_kind() const { + return ws_ ? ws_->failure_kind() : WebSocketClient::FailureKind::Terminal; + } private: std::string url_; diff --git a/app/src/ui_helpers.cpp b/app/src/ui_helpers.cpp index 2cc45e5f..1ee6bc0c 100644 --- a/app/src/ui_helpers.cpp +++ b/app/src/ui_helpers.cpp @@ -310,17 +310,15 @@ void BeginLaunchSessionDialog(const GfnClient& client, const AuthSession& auth, return; } - int unknown_status_polls = 0; + int incomplete_ready_polls = 0; - while (info.status != 2 && info.status != 3) { + while (!info.ready_for_stream()) { + if ((info.status == 2 || info.status == 3) && ++incomplete_ready_polls >= 12) + throw std::runtime_error("NVIDIA did not return a usable streaming allocation in time."); if (info.status > 3 && info.status != 6) { throw std::runtime_error("Session ended or aborted by NVIDIA."); } - if (info.status < 0 && ++unknown_status_polls >= 6) { - throw std::runtime_error("NVIDIA returned an unknown session status for too long."); - } - std::string queue_title; std::string queue_detail; if (info.app_patching) diff --git a/app/src/webrtc/negotiation.cpp b/app/src/webrtc/negotiation.cpp index f64265ed..a7d91c09 100644 --- a/app/src/webrtc/negotiation.cpp +++ b/app/src/webrtc/negotiation.cpp @@ -662,6 +662,8 @@ extern "C" { } // namespace opennow::webrtc::internal void WebRtcSession::handle_signaling_message(const std::string& msg) { + if (signaling_server_rejected_) + return; signaling_rx_count_++; const std::string compact_message = opennow::webrtc::CompactSignalingMessage(msg); AppendStreamLog(compact_message); @@ -676,6 +678,13 @@ void WebRtcSession::handle_signaling_message(const std::string& msg) { json_t* root = json_loads(msg.c_str(), 0, &error); if (!root) return; + json_t* server_error = json_object_get(root, "error"); + if (server_error && !json_is_null(server_error)) { + signaling_server_rejected_ = true; + json_decref(root); + return; + } + json_t* peer_info = json_object_get(root, "peer_info"); if (peer_info && json_is_object(peer_info)) { const char* name = json_string_value(json_object_get(peer_info, "name")); diff --git a/app/src/webrtc/session.cpp b/app/src/webrtc/session.cpp index 9c51f48f..20168b8c 100644 --- a/app/src/webrtc/session.cpp +++ b/app/src/webrtc/session.cpp @@ -51,26 +51,6 @@ bool CreateNvdecCrashMarker() return marker.good(); } -std::string BuildSignInUrl(std::string base_url, const std::string& peer_name, const std::string& session_id) -{ - const size_t query_pos = base_url.find('?'); - if (query_pos != std::string::npos) - base_url.erase(query_pos); - - if (StartsWith(base_url, "http://")) - base_url.replace(0, 4, "ws"); - else if (StartsWith(base_url, "https://")) - base_url.replace(0, 5, "wss"); - - while (!base_url.empty() && base_url.back() == '/') - base_url.pop_back(); - - if (base_url.size() < 8 || base_url.substr(base_url.size() - 8) != "sign_in") - base_url += "/sign_in"; - - return base_url + "?peer_id=" + peer_name + "&version=2&peer_role=1&pairing_id=" + session_id; -} - std::string MakePeerName() { std::random_device rd; @@ -237,35 +217,9 @@ void WebRtcSession::start() { return; } - const std::string sign_in_url = BuildSignInUrl(signaling_url_, peer_name_, session_id_); - signaling_client_ = std::make_unique(sign_in_url); - signaling_client_->set_on_message([this](const std::string& msg) { - handle_signaling_message(msg); - }); - - std::vector headers; - headers.push_back("Origin: https://play.geforcenow.com"); - headers.push_back("User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/131.0.0.0 Safari/537.36"); - if (!session_id_.empty()) { - headers.push_back("Sec-WebSocket-Protocol: x-nv-sessionid." + session_id_); - } - signaling_client_->set_custom_headers(headers); - - if (!signaling_client_->connect()) { - current_state_ = "WebSocket Connect Failed: " + signaling_client_->get_last_error(); - peer_terminal_kind_.store( - static_cast(opennow::PeerTerminalKind::Failed), - std::memory_order_release); - peer_terminal_.store(true, std::memory_order_release); - AppendStreamLog("SESSION error websocket_connect_failed " + signaling_client_->get_last_error()); - signaling_ready_ = false; + connect_signaling(); + if (peer_terminal_.load(std::memory_order_acquire)) return; - } - signaling_ready_ = true; - current_state_ = "Signaling connected, waiting for offer"; - AppendStreamLog("SIGNAL connected"); - send_peer_info(); - send_heartbeat(); if (settings_.audio_enabled) audio_->start(); start_decoder_worker(); @@ -363,11 +317,9 @@ void WebRtcSession::poll() { std::lock_guard lock(peer_mutex_); - if (signaling_client_) { - signaling_client_->poll(); - } - - maybe_send_keepalive(); + poll_signaling(); + if (stop_requested_.load(std::memory_order_acquire)) + return; if (pc_ && signaling_ready_) { if (!remote_description_set_) { diff --git a/app/src/webrtc/signaling.cpp b/app/src/webrtc/signaling.cpp new file mode 100644 index 00000000..77226155 --- /dev/null +++ b/app/src/webrtc/signaling.cpp @@ -0,0 +1,138 @@ +#include "webrtc_session.hpp" +#include "internal.hpp" +#include "startup_timeout_policy.hpp" + +using namespace opennow::webrtc::internal; + +namespace +{ + +std::string EncodeQueryValue(const std::string& value) +{ + constexpr char hex[] = "0123456789ABCDEF"; + std::string encoded; + for (unsigned char byte : value) { + if ((byte >= 'a' && byte <= 'z') || (byte >= 'A' && byte <= 'Z') || + (byte >= '0' && byte <= '9') || byte == '-' || byte == '_' || byte == '.' || byte == '~') { + encoded += static_cast(byte); + } else { + encoded += '%'; + encoded += hex[byte >> 4]; + encoded += hex[byte & 15]; + } + } + return encoded; +} + +std::string BuildSignInUrl(std::string base_url, const std::string& peer_name, + const std::string& session_id, bool reconnect) +{ + const size_t query_pos = base_url.find_first_of("?#"); + if (query_pos != std::string::npos) + base_url.erase(query_pos); + + if (StartsWith(base_url, "http://")) + base_url.replace(0, 4, "ws"); + else if (StartsWith(base_url, "https://")) + base_url.replace(0, 5, "wss"); + + while (!base_url.empty() && base_url.back() == '/') + base_url.pop_back(); + + if (!base_url.ends_with("/sign_in")) + base_url += "/sign_in"; + + return base_url + "?peer_id=" + EncodeQueryValue(peer_name) + + (reconnect ? "&version=3.0" : "&version=2") + "&peer_role=1&pairing_id=" + + EncodeQueryValue(session_id) + (reconnect ? "&reconnect=1" : ""); +} + +} + +void WebRtcSession::connect_signaling() +{ + const auto deadline = session_started_at_ + opennow::webrtc::TransportStartupTimeout; + if (stop_requested_.load(std::memory_order_acquire) || offer_received_) + return; + if (std::chrono::steady_clock::now() >= deadline) { + handle_signaling_failure(WebSocketClient::FailureKind::Terminal); + return; + } + signaling_retry_due_ = {}; + signaling_server_rejected_ = false; + peer_id_ = 0; + remote_peer_id_ = 1; + ack_counter_ = 0; + const std::string sign_in_url = BuildSignInUrl( + signaling_url_, peer_name_, session_id_, signaling_retries_ > 0); + signaling_client_ = std::make_unique(sign_in_url); + signaling_client_->set_on_message([this](const std::string& msg) { + handle_signaling_message(msg); + }); + + std::vector headers; + headers.push_back("Origin: https://play.geforcenow.com"); + headers.push_back("User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/131.0.0.0 Safari/537.36"); + if (!session_id_.empty()) { + headers.push_back("Sec-WebSocket-Protocol: x-nv-sessionid." + session_id_); + } + signaling_client_->set_custom_headers(headers); + + if (!signaling_client_->connect(deadline)) { + handle_signaling_failure(signaling_client_->failure_kind()); + return; + } + signaling_ready_ = true; + current_state_ = "Signaling connected, waiting for offer"; + AppendStreamLog("SIGNAL connected"); + send_peer_info(); + send_heartbeat(); +} + +void WebRtcSession::poll_signaling() +{ + if (stop_requested_.load(std::memory_order_acquire)) + return; + if (signaling_retry_due_.time_since_epoch().count() != 0) { + if (std::chrono::steady_clock::now() >= signaling_retry_due_) + connect_signaling(); + return; + } + if (signaling_client_) { + signaling_client_->poll(); + if (signaling_server_rejected_ || !signaling_client_->is_connected()) { + handle_signaling_failure(signaling_server_rejected_ + ? WebSocketClient::FailureKind::Terminal : signaling_client_->failure_kind()); + return; + } + } + maybe_send_keepalive(); +} + +void WebRtcSession::handle_signaling_failure(WebSocketClient::FailureKind failure) +{ + const std::string detail = signaling_server_rejected_ ? "Signaling server rejected request" + : signaling_client_ ? signaling_client_->get_last_error() : "Signaling startup deadline expired"; + AppendStreamLog("SIGNAL failure " + detail); + signaling_client_.reset(); + signaling_ready_ = offer_received_; + if (offer_received_) + return; + + const auto now = std::chrono::steady_clock::now(); + const auto delay = std::chrono::seconds(1 << signaling_retries_); + if (failure == WebSocketClient::FailureKind::Transient && signaling_retries_ < 3 && + now + delay < session_started_at_ + opennow::webrtc::TransportStartupTimeout) { + signaling_retries_++; + signaling_retry_due_ = now + delay; + current_state_ = "Retrying signaling (" + std::to_string(signaling_retries_) + "/3)"; + AppendStreamLog("SIGNAL retry scheduled attempt=" + std::to_string(signaling_retries_)); + return; + } + + signaling_retry_due_ = {}; + current_state_ = detail; + peer_terminal_kind_.store(static_cast(opennow::PeerTerminalKind::Failed), std::memory_order_release); + peer_terminal_.store(true, std::memory_order_release); + request_stop(); +} diff --git a/app/src/webrtc/startup_timeout_policy.hpp b/app/src/webrtc/startup_timeout_policy.hpp index fa655bfc..1352849d 100644 --- a/app/src/webrtc/startup_timeout_policy.hpp +++ b/app/src/webrtc/startup_timeout_policy.hpp @@ -5,6 +5,8 @@ namespace opennow::webrtc { +inline constexpr auto TransportStartupTimeout = std::chrono::seconds(30); + enum class StartupTimeout { None, @@ -19,7 +21,7 @@ constexpr StartupTimeout DetectStartupTimeout( std::chrono::milliseconds peer_completed_elapsed) { if (!peer_completed) - return session_elapsed >= std::chrono::seconds(30) + return session_elapsed >= TransportStartupTimeout ? StartupTimeout::Transport : StartupTimeout::None; if (!frame_decoded && peer_completed_elapsed >= std::chrono::seconds(15)) return StartupTimeout::Video; diff --git a/app/src/webrtc_session.hpp b/app/src/webrtc_session.hpp index 078ecdcb..281b2d87 100644 --- a/app/src/webrtc_session.hpp +++ b/app/src/webrtc_session.hpp @@ -131,6 +131,9 @@ class WebRtcSession { int remote_peer_id_ = 1; int ack_counter_ = 0; bool signaling_ready_ = false; + bool signaling_server_rejected_ = false; + int signaling_retries_ = 0; + std::chrono::steady_clock::time_point signaling_retry_due_ {}; bool remote_description_set_ = false; bool offer_received_ = false; bool manual_candidate_added_ = false; @@ -261,6 +264,9 @@ class WebRtcSession { uint32_t last_rtp_access_units_dropped_ = 0; void handle_signaling_message(const std::string& msg); + void connect_signaling(); + void poll_signaling(); + void handle_signaling_failure(WebSocketClient::FailureKind failure); void setup_peer_connection(); void start_network_worker(); void network_loop(); diff --git a/scripts/test-host.sh b/scripts/test-host.sh index aa9ddb68..aa24efc4 100644 --- a/scripts/test-host.sh +++ b/scripts/test-host.sh @@ -49,6 +49,11 @@ for name in gfn_atomic_write gfn_device_identity gfn_trace_redaction; do done run_cpp catalog_response -ffunction-sections -fdata-sections -Wl,--gc-sections \ app/src/gfn/shared.cpp -ljansson +run_cpp cloud_session_allocation -ffunction-sections -fdata-sections -Wl,--gc-sections \ + app/src/gfn/cloud_session_protocol.cpp app/src/gfn/shared.cpp -ljansson +run_cpp cloud_session_recovery -ffunction-sections -fdata-sections -Wl,--gc-sections \ + app/src/gfn/cloud_session.cpp app/src/gfn/cloud_session_protocol.cpp \ + app/src/gfn/cloud_session_state.cpp app/src/gfn/shared.cpp -ljansson run_cpp cover_image_cache -Itests/cover_cache_stubs app/src/cover_image_cache.cpp run_cpp cached_image_admission -Itests/cover_cache_stubs app/src/cover_image_cache.cpp run_cpp http_client_error app/src/http_client.cpp -lcurl diff --git a/scripts/test-streaming-host.sh b/scripts/test-streaming-host.sh index e42bb06e..3bd6564f 100644 --- a/scripts/test-streaming-host.sh +++ b/scripts/test-streaming-host.sh @@ -56,6 +56,18 @@ run_cpp websocket_client -DFMT_HEADER_ONLY -isystem extern/ffmpeg/include \ -Iextern/libpeer/third_party/mbedtls/include "${sections[@]}" \ app/src/WebSocketClient.cpp app/src/signaling_client.cpp app/src/http_client.cpp "${crypto[@]}" run_cpp websocket_handshake -Iextern/libpeer/third_party/mbedtls/include -Wl,--gc-sections "${crypto[@]}" +run_cpp webrtc_signaling -DFMT_HEADER_ONLY -Iextern/libpeer/src \ + -Iextern/borealis/library/include/borealis/extern/nanovg \ + -isystem extern/ffmpeg/include -isystem extern/borealis/library/include \ + -isystem extern/borealis/library/lib/extern/fmt/include \ + -Iextern/libpeer/third_party/mbedtls/include "${sections[@]}" \ + app/src/webrtc/signaling.cpp app/src/webrtc/negotiation.cpp app/src/webrtc/input.cpp \ + app/src/webrtc/nvst_sdp.cpp app/src/WebSocketClient.cpp app/src/signaling_client.cpp \ + app/src/http_client.cpp "${crypto[@]}" -ljansson +for scenario in url rejected closed initial-transient terminal retry-limit deadline post-offer; do + "$out/webrtc_signaling" "$scenario" + printf 'PASS webrtc_signaling/%s\n' "$scenario" +done run_cpp signaling_diagnostics -ljansson run_cpp webrtc_input_handshake -Iextern/libpeer/src \ -Iextern/libpeer/third_party/mbedtls/include \ diff --git a/tests/cloud_session_allocation_test.cpp b/tests/cloud_session_allocation_test.cpp new file mode 100644 index 00000000..4763b6be --- /dev/null +++ b/tests/cloud_session_allocation_test.cpp @@ -0,0 +1,43 @@ +#include "gfn/cloud_session_internal.hpp" + +#include +#include +#include + +int main(int argc, char** argv) +{ + const int selected = argc > 1 ? std::atoi(argv[1]) : -1; + struct Case { const char* input; const char* signaling; const char* media; int port; }; + const Case cases[] = { + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":6,"resourcePath":"rtsps://classic.example:48322"},{"usage":14,"appLevelProtocol":5,"ip":"browser.example","port":443,"resourcePath":"/nvst/"}]})", "wss://browser.example:443/nvst/", "browser.example", 443}, + {R"({"connectionInfo":[{"usage":14,"ip":".cloudmatchbeta.nvidiagrid.net","port":443,"resourcePath":"/nvst/"},{"usage":14,"ip":"seat.example","port":443,"resourcePath":"/nvst/"}]})", "wss://seat.example:443/nvst/", "seat.example", 443}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":5,"ip":"seat.example","port":49003,"resourcePath":"/nvst/"}]})", "wss://seat.example:443/nvst/", "seat.example", 49003}, + {R"({"sessionControlInfo":{"ip":"np-ams-06.cloudmatchbeta.nvidiagrid.net","port":443}})", "", "", 0}, + {R"({"connectionInfo":[{"usage":14,"ip":"seat.example","port":443,"resourcePath":"/nvst/"},{"usage":14,"appLevelProtocol":6,"ip":"classic.example","port":48322,"resourcePath":"rtsps://classic.example:48322"}]})", "wss://seat.example:443/nvst/", "seat.example", 443}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":6,"resourcePath":"wss://classic.example/nvst/"},{"usage":14,"appLevelProtocol":5,"resourcePath":"https://browser.example:9443/custom/"},{"usage":17,"ip":"203.0.113.20","port":49005}]})", "wss://browser.example:9443/custom/", "203.0.113.20", 49005}, + {R"({"connectionInfo":[{"usage":14,"resourcePath":"wss://user:secret@bad.example/nvst/"},{"usage":14,"ip":"0.0.0.0","port":443},{"usage":14,"resourcePath":"wss://good.example:8443/nvst/"}]})", "wss://good.example:8443/nvst/", "good.example", 8443}, + {R"({"connectionInfo":[{"usage":14,"ip":"2001:db8::10","port":49003,"resourcePath":"/nvst/"}]})", "wss://[2001:db8::10]:443/nvst/", "2001:db8::10", 49003}, + {R"({"signalingUrl":"https://seat.example:9443/custom/","connectionInfo":[{"usage":2,"ip":"203.0.113.12","port":49002}]})", "wss://seat.example:9443/custom/", "203.0.113.12", 49002}, + {R"({"connectionInfo":[{"usage":16,"appLevelProtocol":6,"ip":"classic.example","port":48322}]})", "", "", 0}, + {R"({"connectionInfo":[{"usage":14,"resourcePath":"wss://[not-ip.example]/nvst/"},{"usage":14,"resourcePath":"wss://seat.example:65536/nvst/"},{"usage":14,"resourcePath":"wss://seat.example:443junk/nvst/"},{"usage":14,"resourcePath":"http://seat.example/nvst/"},{"usage":14,"resourcePath":"wss://good.example:9443/nvst/"}]})", "wss://good.example:9443/nvst/", "good.example", 9443}, + }; + for (int i = 0; i < static_cast(std::size(cases)); ++i) + { + if (selected >= 0 && selected != i) + continue; + auto root = opennow::gfn::detail::LoadJson(cases[i].input); + opennow::SessionInfo info; + opennow::gfn::cloud_session::ApplySessionNetworkInfo(root.get(), info); + assert(info.signaling_url == cases[i].signaling); + assert(info.media_ip == cases[i].media); + assert(info.media_port == cases[i].port); + } + if (selected < 0 || selected == 11) + { + for (const char* text : {"2", "\"2\"", "\"ready\""}) + { + auto root = opennow::gfn::detail::LoadJson(std::string("{\"status\":") + text + "}"); + assert(opennow::gfn::cloud_session::ParseSessionStatus(json_object_get(root.get(), "status")) == 2); + } + } +} diff --git a/tests/cloud_session_recovery_test.cpp b/tests/cloud_session_recovery_test.cpp new file mode 100644 index 00000000..846c7af8 --- /dev/null +++ b/tests/cloud_session_recovery_test.cpp @@ -0,0 +1,244 @@ +#include "gfn/cloud_session_internal.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace +{ +struct Reply +{ + std::string method; + std::string url; + opennow::HttpResponse response; + std::function inspect; +}; +std::deque replies; +opennow::StreamSettings settings; +const std::string initial = "https://np-ams-06.cloudmatchbeta.nvidiagrid.net/"; +const std::string assigned = "https://np-par-01.cloudmatchbeta.nvidiagrid.net/"; +const std::string queued = R"({"requestStatus":{"statusCode":1},"session":{"sessionId":"seat","status":1,"sessionControlInfo":{"ip":"np-par-01.cloudmatchbeta.nvidiagrid.net","port":443}}})"; +const std::string with_ads = R"({"requestStatus":{"statusCode":1},"session":{"sessionId":"seat","status":1,"sessionAdsRequired":1,"sessionAds":[{"adId":"ad-one"},{"adId":"ad-one"}],"sessionControlInfo":{"ip":"np-par-01.cloudmatchbeta.nvidiagrid.net","port":443}}})"; + +void expect(const std::string& method, const std::string& url, int code, const std::string& body, + std::function inspect = {}) +{ + replies.push_back({method, url, {code, body}, std::move(inspect)}); +} + +void expect_start(const std::string& body) +{ + expect("POST", initial + "v2/nettestsession", 200, + R"({"requestStatus":{"statusCode":1},"netTestSession":{"sessionId":"nettest"}})"); + expect("POST", initial + "v2/session?keyboardLayout=en-US_qwerty&languageCode=en_US", 200, body); +} + +void inspect_ad(const std::string& body) +{ + const auto root = opennow::gfn::detail::LoadJson(body); + assert(opennow::gfn::detail::GetInteger(root.get(), "action") == 6); + json_t* updates = json_object_get(root.get(), "adUpdates"); + assert(json_array_size(updates) == 1); + json_t* ad = json_array_get(updates, 0); + assert(opennow::gfn::detail::GetString(ad, "adId") == "ad-one"); + assert(opennow::gfn::detail::GetInteger(ad, "adAction") == 5); + assert(opennow::gfn::detail::GetString(ad, "cancelReason") == "FAILED_TO_REQUEST_ADS"); + for (const char* key : {"watchedTimeInMs", "pausedTimeInMs", "adDurationInMs"}) { + assert(json_is_integer(json_object_get(ad, key))); + assert(opennow::gfn::detail::GetInteger(ad, key) == 0); + } + assert(json_integer_value(json_object_get(ad, "clientTimestamp")) > 0); +} +} + +namespace opennow +{ +bool StreamDiagnosticsEnabled() { return false; } +StreamSettings LoadStreamSettings() { return settings; } +AuthSession GfnClient::EnsureFreshSavedSession(const AuthSession& auth) const { return auth; } +AuthSession GfnClient::ForceRefreshSavedSession(const AuthSession& auth) const { return auth; } +std::vector GfnClient::FetchStreamRegions(AuthSession&) const { return {}; } +std::vector GfnClient::MeasureStreamRegionLatencies(std::vector regions) const { return regions; } + +HttpResponse HttpClient::Request(const std::string& method, const std::string& url, + const std::string&, const std::vector&, const std::string& body, + const std::string&, HttpTransferControl) const +{ + assert(!replies.empty()); + auto reply = std::move(replies.front()); + replies.pop_front(); + assert(method == reply.method); + assert(url == reply.url); + if (reply.inspect) + reply.inspect(body); + return reply.response; +} +HttpResponse HttpClient::Get(const std::string& url, const std::string& ua, + const std::vector& headers, const std::string& proxy, HttpTransferControl control) const +{ + return Request("GET", url, ua, headers, {}, proxy, control); +} +HttpResponse HttpClient::Post(const std::string& url, const std::string& ua, + const std::vector& headers, const std::string& body, + const std::string& proxy, HttpTransferControl control) const +{ + return Request("POST", url, ua, headers, body, proxy, control); +} +} + +int main(int argc, char** argv) +{ + const int selected = argc > 1 ? std::atoi(argv[1]) : -1; + const auto cwd = std::filesystem::current_path(); + std::string pattern = (std::filesystem::temp_directory_path() / "opennow-recovery-XXXXXX").string(); + const char* directory = mkdtemp(pattern.data()); + assert(directory); + std::filesystem::current_path(directory); + std::filesystem::create_directories(opennow::gfn::detail::GetAppHome()); + settings.region = initial; + settings.game_language = "en_US"; + opennow::AuthSession auth; + auth.user.user_id = "synthetic-user"; + auth.tokens.id_token = "synthetic-token"; + auth.provider.streaming_service_url = initial; + opennow::GfnClient client; + if (selected < 0 || selected == 0) { + expect_start(queued); + client.StartSession(auth, "123", settings); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionStreamingBaseUrl(auth, "seat") == assigned); + expect("GET", assigned + "v2/session/seat", 200, queued); + client.PollSession(auth, "seat"); + expect("DELETE", assigned + "v2/session/seat", 204, ""); + client.StopSession(auth, "seat"); + } + if (selected < 0 || selected == 1) { + for (const char* body : {"{}", R"({"requestStatus":{"statusCode":1},"session":{"sessionId":"other","status":2}})", + R"({"requestStatus":{"statusCode":1},"session":{"sessionId":"seat"}})"}) { + expect("GET", initial + "v2/session/seat", 200, body); + bool threw = false; + try { client.PollSession(auth, "seat"); } catch (const std::exception&) { threw = true; } + assert(threw); + } + } + if (selected < 0 || selected == 2) { + opennow::gfn::cloud_session::RememberActiveCloudSession(auth, "seat", "123", initial); + expect("DELETE", initial + "v2/session/seat", 200, R"({"requestStatus":{"statusCode":4}})"); + bool threw = false; + try { client.StopSession(auth, "seat"); } catch (const std::exception&) { threw = true; } + assert(threw); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionId(auth) == "seat"); + opennow::gfn::cloud_session::ForgetActiveCloudSession("seat"); + } + if (selected < 0 || selected == 3) { + opennow::gfn::cloud_session::RememberActiveCloudSession(auth, "seat", "123", assigned); + opennow::gfn::cloud_session::RememberActiveCloudSession(auth, "seat", "123", assigned); + const std::string path = opennow::gfn::detail::GetAppHome() + "/active_cloud_session.json"; + std::filesystem::remove(path); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionId(auth) == "seat"); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionStreamingBaseUrl(auth, "seat") == assigned); + std::ofstream(path) << "{"; + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionId(auth) == "seat"); + opennow::gfn::cloud_session::ForgetActiveCloudSession("seat"); + } + if (selected < 0 || selected == 4) { + expect_start(with_ads); + expect("PUT", assigned + "v2/session/seat", 503, "{}", inspect_ad); + client.StartSession(auth, "123", settings); + expect("GET", assigned + "v2/session/seat", 200, queued); + expect("PUT", assigned + "v2/session/seat", 200, R"({"requestStatus":{"statusCode":1}})", inspect_ad); + client.PollSession(auth, "seat"); + expect("GET", assigned + "v2/session/seat", 200, with_ads); + client.PollSession(auth, "seat"); + assert(replies.empty()); + opennow::gfn::cloud_session::ForgetActiveCloudSession("seat"); + } + if (selected < 0 || selected == 5) { + for (const char* host : {"evil.example", "np-par-01.cloudmatchbeta.nvidiagrid.net.evil.example", + "np-par-01.cloudmatchbeta.nvidiagrid.net@evil.example", ".cloudmatchbeta.nvidiagrid.net", + "np..cloudmatchbeta.nvidiagrid.net", "http://np-par-01.cloudmatchbeta.nvidiagrid.net", + "np-par-01.cloudmatchbeta.nvidiagrid.net?x=1", "np-par-01.cloudmatchbeta.nvidiagrid.net#x"}) { + auto root = opennow::gfn::detail::LoadJson(std::string("{\"sessionControlInfo\":{\"ip\":\"") + host + "\",\"port\":443}}"); + assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(root.get(), initial) == initial); + } + for (const char* port : {"80", "8443", "0", "null", "\"bogus\""}) { + auto root = opennow::gfn::detail::LoadJson(std::string("{\"sessionControlInfo\":{\"ip\":\"np-par-01.cloudmatchbeta.nvidiagrid.net\",\"port\":") + port + "}}"); + assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(root.get(), initial) == initial); + } + auto root = opennow::gfn::detail::LoadJson(R"({"sessionControlInfo":{"ip":"seat.partner.example","port":443}})"); + assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(root.get(), "https://seat.partner.example/") == "https://seat.partner.example/"); + assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(root.get(), initial) == initial); + auto nvidia = opennow::gfn::detail::LoadJson(R"({"sessionControlInfo":{"ip":"NP-PAR-01.cloudmatchbeta.nvidiagrid.net","port":"443"}})"); + assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(nvidia.get(), initial) == assigned); + } + if (selected < 0 || selected == 6) { + const std::string ready = R"({"requestStatus":{"statusCode":"1"},"session":{"sessionId":"seat","status":"2","connectionInfo":[{"usage":"14","appLevelProtocol":"5","ip":"seat.example","port":"443","resourcePath":"/nvst/"}]}})"; + expect("GET", initial + "v2/session/seat", 200, ready); + auto info = client.PollSession(auth, "seat"); + assert(info.ready_for_stream()); + expect("GET", initial + "v2/session/seat", 200, R"({"requestStatus":{"statusCode":1},"session":{"sessionId":"seat","status":2}})"); + info = client.PollSession(auth, "seat"); + assert(!info.ready_for_stream()); + assert(info.signaling_url.empty()); + assert(info.media_ip.empty()); + assert(info.ice_servers.empty()); + expect("GET", initial + "v2/session/seat", 200, ready); + assert(client.PollSession(auth, "seat").ready_for_stream()); + } + if (selected < 0 || selected == 7) { + expect_start(with_ads); + expect("PUT", assigned + "v2/session/seat", 200, R"({"requestStatus":{"statusCode":4}})", inspect_ad); + auto info = client.StartSession(auth, "123", settings); + assert(info.ads_pending); + for (int attempt = 1; attempt < 3; ++attempt) { + expect("GET", assigned + "v2/session/seat", 200, queued); + expect("PUT", assigned + "v2/session/seat", 503, "{}", inspect_ad); + bool threw = false; + try { info = client.PollSession(auth, "seat"); } catch (const std::exception&) { threw = true; } + assert(threw == (attempt == 2)); + } + opennow::gfn::cloud_session::ForgetActiveCloudSession("seat"); + } + if (selected < 0 || selected == 8) { + for (const bool cleanup_succeeds : {false, true}) { + expect_start(with_ads); + expect("PUT", assigned + "v2/session/seat", 403, "{}", inspect_ad); + expect("DELETE", assigned + "v2/session/seat", cleanup_succeeds ? 204 : 503, ""); + bool threw = false; + try { client.StartSession(auth, "123", settings); } catch (const std::exception&) { threw = true; } + assert(threw); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionId(auth).empty() == cleanup_succeeds); + opennow::gfn::cloud_session::ForgetActiveCloudSession("seat"); + } + } + if (selected < 0 || selected == 9) { + opennow::gfn::cloud_session::RememberActiveCloudSession(auth, "seat", "123", initial); + expect("GET", initial + "v2/session/seat", 200, queued); + client.PollSession(auth, "seat"); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionStreamingBaseUrl(auth, "seat") == assigned); + opennow::AuthSession other = auth; + other.user.user_id = "different-user"; + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionId(other).empty()); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionStreamingBaseUrl(other, "seat").empty()); + expect("DELETE", assigned + "v2/session/seat", 200, R"({"requestStatus":{"statusCode":"1"}})"); + client.StopSession(auth, "seat"); + } + if (selected < 0 || selected == 10) { + const std::string required = R"({"requestStatus":{"statusCode":1},"session":{"sessionId":"seat","status":1,"sessionAdsRequired":true}})"; + expect_start(required); + assert(client.StartSession(auth, "123", settings).ads_pending); + for (int poll = 1; poll < 12; ++poll) { + expect("GET", initial + "v2/session/seat", 200, required); + bool threw = false; + try { client.PollSession(auth, "seat"); } catch (const std::exception&) { threw = true; } + assert(threw == (poll == 11)); + } + opennow::gfn::cloud_session::ForgetActiveCloudSession("seat"); + } + assert(replies.empty()); + std::filesystem::current_path(cwd); + std::filesystem::remove_all(directory); +} diff --git a/tests/webrtc_signaling_test.cpp b/tests/webrtc_signaling_test.cpp new file mode 100644 index 00000000..b2746472 --- /dev/null +++ b/tests/webrtc_signaling_test.cpp @@ -0,0 +1,231 @@ +#include "websocket_test_transport.hpp" + +#include +#include +#include +#include +#include +#include + +#define private public +#include "webrtc_session.hpp" +#undef private +#include "webrtc/internal.hpp" + +static std::vector stream_logs; +static int remote_descriptions = 0; +static int answers_created = 0; + +extern "C" { +int peer_init() { return 0; } +void peer_deinit() {} +void peer_connection_set_remote_description(PeerConnection*, const char*, SdpType) { ++remote_descriptions; } +const char* peer_connection_create_answer(PeerConnection*) { ++answers_created; return nullptr; } +int peer_connection_add_ice_candidate(PeerConnection*, char*) { return 0; } +PeerConnectionState peer_connection_get_state(PeerConnection*) { return PEER_CONNECTION_COMPLETED; } +const char* peer_connection_state_to_string(PeerConnectionState) { return "completed"; } +} + +namespace opennow { +bool StreamDiagnosticsEnabled() { return false; } +} + +namespace opennow::webrtc::internal { +bool StartsWith(const std::string& text, const char* prefix) { return text.rfind(prefix, 0) == 0; } +void AppendStreamLog(const std::string& line) { stream_logs.push_back(line); } +void AppendInputLog(const std::string&) {} +void AppendTraceLog(const std::string&) {} +void AppendSdpSummary(const std::string&, const std::string&) {} +} + +WebRtcSession::WebRtcSession(const std::string& url, const std::string&, const std::string& id, + const std::string&, int, const std::vector&, + const opennow::StreamSettings& settings) + : signaling_url_(url), session_id_(id), settings_(settings), peer_name_("peer-test") { + session_started_at_ = std::chrono::steady_clock::now(); +} +WebRtcSession::~WebRtcSession() = default; +void WebRtcSession::log_stream_summary(const char*) {} +void WebRtcSession::request_stop() { stop_requested_.store(true); } +FFmpegVideoDecoder::~FFmpegVideoDecoder() = default; +int FFmpegVideoDecoder::setup(int, int, int, int, void*, int) { return 0; } +void FFmpegVideoDecoder::cleanup() {} +int FFmpegVideoDecoder::submit_decode_unit(uint8_t*, int, int64_t) { return 0; } +int FFmpegVideoDecoder::capabilities() const { return 0; } +VideoDecodeStats* FFmpegVideoDecoder::video_decode_stats() { return nullptr; } +struct AudioPipeline::Impl {}; +AudioPipeline::~AudioPipeline() = default; + +static void deliver(WebRtcSession& session, const std::string& message, uint8_t opcode = 0x81) { + auto& fake = latest(); + assert(message.size() < 126); + fake.input = {opcode, static_cast(message.size())}; + fake.input.insert(fake.input.end(), message.begin(), message.end()); + session.poll_signaling(); + session.poll_signaling(); +} + +static void close_socket(WebRtcSession& session, uint16_t code) { + latest().input = {0x88, 2, static_cast(code >> 8), static_cast(code)}; + session.poll_signaling(); + session.poll_signaling(); +} + +static void retry_now(WebRtcSession& session) { + assert(session.signaling_retry_due_ > std::chrono::steady_clock::now()); + const auto dials = transports.size(); + session.poll_signaling(); + assert(transports.size() == dials); + session.signaling_retry_due_ = std::chrono::steady_clock::now(); + session.poll_signaling(); +} + +int main(int argc, char** argv) { + const std::string scenario = argc > 1 ? argv[1] : "url"; + if (scenario == "url") { + for (const auto* url : {"wss://rig.invalid/nvst/", "wss://rig.invalid/nvst/sign_in", + "https://rig.invalid/nvst/sign_in/?old=1#fragment"}) { + WebRtcSession session(url, "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + assert(session.signaling_ready_); + assert(latest().request.starts_with("GET /nvst/sign_in?peer_id=peer-test&version=2&peer_role=1&pairing_id=test-session HTTP/1.1\r\n")); + assert(latest().request.find("Sec-WebSocket-Protocol: x-nv-sessionid.test-session\r\n") != std::string::npos); + } + WebRtcSession session("wss://rig.invalid/nvst/sign_in#fragment", "", "s&?/#=+%", "", 0, {}, {}); + session.peer_name_ = "peer &?/#=+%\xc3\xa9"; + session.connect_signaling(); + assert(latest().request.starts_with("GET /nvst/sign_in?peer_id=peer%20%26%3F%2F%23%3D%2B%25%C3%A9&version=2&peer_role=1&pairing_id=s%26%3F%2F%23%3D%2B%25 HTTP/1.1\r\n")); + } else if (scenario == "rejected") { + for (const auto* message : {R"({"error":"peerRemoved"})", R"({"error":{"secret":"never-log-this"}})"}) { + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + deliver(session, message); + assert(session.is_terminal()); + assert(session.stop_requested_.load()); + assert(!session.signaling_ready_); + assert(session.signaling_retries_ == 0); + assert(session.current_state_ == "Signaling server rejected request"); + } + for (const auto& line : stream_logs) + assert(line.find("never-log-this") == std::string::npos); + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + deliver(session, R"({"error":null,"hb":1})"); + assert(!session.is_terminal() && session.heartbeat_rx_count_ == 1); + } else if (scenario == "closed") { + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + close_socket(session, 1013); + assert(!session.signaling_ready_); + assert(!session.is_terminal()); + assert(session.signaling_retries_ == 1); + assert(session.signaling_retry_due_ > std::chrono::steady_clock::now()); + const auto started = session.session_started_at_; + retry_now(session); + assert(session.session_started_at_ == started); + assert(session.signaling_ready_ && transports.size() == 2); + assert(latest().request.starts_with("GET /nvst/sign_in?peer_id=peer-test&version=3.0&peer_role=1&pairing_id=test-session&reconnect=1 HTTP/1.1\r\n")); + deliver(session, R"({"ackid":9,"peer_info":{"name":"peer-test","id":42}})"); + assert(session.peer_id_ == 42); + session.pc_ = reinterpret_cast(&session); + deliver(session, R"({"peer_msg":{"from":17,"msg":"{\"type\":\"offer\",\"sdp\":\"v=0\"}"}})", 0x82); + assert(session.offer_count_ == 1 && session.remote_peer_id_ == 17); + assert(remote_descriptions == 1 && answers_created == 1); + } else if (scenario == "initial-transient") { + for (bool http : {false, true}) { + if (http) next_upgrade = UpgradeResponse::Unavailable; + else next_perform_result = CURLE_COULDNT_CONNECT; + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.pc_ = reinterpret_cast(&session); + const auto started = session.session_started_at_; + session.connect_signaling(); + assert(!session.is_terminal() && !session.signaling_ready_); + assert(session.signaling_retries_ == 1); + retry_now(session); + assert(session.session_started_at_ == started); + assert(session.pc_ == reinterpret_cast(&session)); + assert(session.signaling_ready_); + assert(latest().request.find("version=3.0") != std::string::npos); + session.poll_signaling(); + const auto sent = frames(latest()); + assert(sent.size() == 2); + json_error_t error; + json_t* info = json_loads(sent.front().payload.c_str(), 0, &error); + assert(info); + json_t* peer = json_object_get(info, "peer_info"); + assert(std::string(json_string_value(json_object_get(peer, "name"))) == "peer-test"); + assert(json_integer_value(json_object_get(peer, "id")) == 0); + json_decref(info); + } + } else if (scenario == "terminal") { + for (int failure = 0; failure < 3; ++failure) { + if (failure == 0) next_upgrade = UpgradeResponse::Unauthorized; + else if (failure == 1) next_upgrade = UpgradeResponse::WrongAccept; + else next_perform_result = CURLE_PEER_FAILED_VERIFICATION; + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + assert(session.is_terminal() && !session.signaling_ready_); + assert(session.signaling_retries_ == 0); + assert(session.signaling_retry_due_.time_since_epoch().count() == 0); + const auto dials = transports.size(); + for (int i = 0; i < 20; ++i) session.poll_signaling(); + assert(transports.size() == dials); + } + } else if (scenario == "retry-limit") { + next_upgrade = UpgradeResponse::Unavailable; + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + for (int retry = 1; retry <= 3; ++retry) { + assert(session.signaling_retries_ == retry); + next_upgrade = UpgradeResponse::Unavailable; + retry_now(session); + } + assert(session.is_terminal()); + assert(transports.size() == 4); + } else if (scenario == "deadline") { + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + close_socket(session, 1013); + session.session_started_at_ = std::chrono::steady_clock::now() - std::chrono::seconds(28); + retry_now(session); + assert(session.signaling_ready_); + assert(latest().connect_timeout_ms > 0 && latest().connect_timeout_ms <= 2000); + assert(latest().timeout_ms > 0 && latest().timeout_ms <= 2000); + const auto dials = transports.size(); + close_socket(session, 1013); + assert(session.is_terminal()); + assert(transports.size() == dials); + + WebRtcSession expired("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + expired.connect_signaling(); + close_socket(expired, 1013); + const auto before = transports.size(); + expired.session_started_at_ = std::chrono::steady_clock::now() - std::chrono::seconds(31); + retry_now(expired); + assert(expired.is_terminal() && transports.size() == before); + } else if (scenario == "post-offer") { + for (int code : {4001, 1013, 1000, 1008, 0}) { + WebRtcSession session("wss://rig.invalid/nvst/", "", "test-session", "", 0, {}, {}); + session.connect_signaling(); + session.pc_ = reinterpret_cast(&session); + deliver(session, R"({"peer_msg":{"from":1,"msg":"{\"type\":\"offer\",\"sdp\":\"v=0\"}"}})"); + assert(session.offer_received_ && session.remote_description_set_); + session.peer_completed_seen_ = true; + session.frames_decoded_.store(1); + const auto dials = transports.size(); + const auto answers = answers_created; + if (code == 0) deliver(session, R"({"error":"peerRemoved"})"); + else close_socket(session, static_cast(code)); + const auto heartbeats = session.heartbeat_tx_count_; + for (int i = 0; i < 200; ++i) session.poll_signaling(); + assert(session.signaling_ready_ && !session.is_terminal() && !session.stop_requested_.load()); + assert(!session.signaling_client_); + assert(session.heartbeat_tx_count_ == heartbeats); + assert(session.signaling_retries_ == 0 && transports.size() == dials); + assert(answers_created == answers); + assert(session.pc_ == reinterpret_cast(&session)); + } + } else { + assert(false); + } +} diff --git a/tests/websocket_client_test.cpp b/tests/websocket_client_test.cpp index f91b5664..d58e211d 100644 --- a/tests/websocket_client_test.cpp +++ b/tests/websocket_client_test.cpp @@ -1,179 +1,46 @@ -#include "WebSocketClient.hpp" -#include "websocket_handshake.hpp" +#include "websocket_test_transport.hpp" -#include -#include -#include -#include -#include -#include - -namespace -{ - -enum class UpgradeResponse { Valid, Bare, WrongAccept, SlowDrip }; -UpgradeResponse next_upgrade = UpgradeResponse::Valid; - -struct FakeTransport +int main() { - struct Write + using Queue = opennow::websocket::WriteQueue; { - CURLcode result; - size_t bytes; - }; - - bool handshaking = true; - bool closed = false; - size_t calls = 0; - std::deque writes; - std::vector output; - std::vector input; - CURLcode receive_result = CURLE_AGAIN; - UpgradeResponse upgrade = std::exchange(next_upgrade, UpgradeResponse::Valid); - std::string request; - size_t upgrade_offset = 0; -}; - -std::vector> transports; - -FakeTransport& transport(CURL* curl) -{ - auto& result = *reinterpret_cast(curl); - assert(!result.closed); - return result; -} - -FakeTransport& latest() -{ - return *transports.back(); -} - -struct Frame -{ - uint8_t opcode; - std::string payload; -}; - -std::vector frames(const FakeTransport& fake) -{ - std::vector result; - size_t offset = 0; - while (offset < fake.output.size()) { - const auto& bytes = fake.output; - assert(bytes.size() - offset >= 6); - const uint8_t opcode = bytes[offset++]; - assert(opcode & 0x80); - assert(bytes[offset] & 0x80); - size_t length = bytes[offset++] & 0x7f; - if (length == 126 || length == 127) { - const size_t length_bytes = length == 126 ? 2 : 8; - assert(bytes.size() - offset >= length_bytes + 4); - length = 0; - for (size_t i = 0; i < length_bytes; ++i) - length = (length << 8) | bytes[offset++]; - } - const size_t mask = offset; - offset += 4; - assert(bytes.size() - offset >= length); - std::string payload; - for (size_t i = 0; i < length; ++i) - payload.push_back(bytes[offset++] ^ bytes[mask + i % 4]); - result.push_back({static_cast(opcode & 0x0f), std::move(payload)}); - } - return result; -} - -} - -extern "C" { - -CURL* curl_easy_init() -{ - transports.push_back(std::make_unique()); - return reinterpret_cast(transports.back().get()); -} - -void curl_easy_cleanup(CURL* curl) -{ - transport(curl).closed = true; -} - -CURLcode curl_easy_setopt(CURL*, CURLoption, ...) -{ - return CURLE_OK; -} - -CURLcode curl_easy_perform(CURL*) -{ - return CURLE_OK; -} - -const char* curl_easy_strerror(CURLcode) -{ - return "fake transport error"; -} - -CURLcode curl_easy_send(CURL* curl, const void* buffer, size_t length, size_t* sent) -{ - auto& fake = transport(curl); - if (fake.handshaking) { - fake.request.append(static_cast(buffer), length); - *sent = length; - return CURLE_OK; - } - ++fake.calls; - assert(length <= opennow::websocket::WriteQueue::MaximumWriteBytes); - FakeTransport::Write write{CURLE_OK, length}; - if (!fake.writes.empty()) { - write = fake.writes.front(); - fake.writes.pop_front(); + WebSocketClient client("ws://example.invalid"); + assert(client.connect()); + auto& fake = latest(); + std::vector messages; + client.set_on_message([&](const std::string& message) { messages.push_back(message); }); + fake.input = {0x82, 8, '{', '"', 'h', 'b', '"', ':', '1', '}'}; + client.poll(); + client.poll(); + assert(messages == std::vector{"{\"hb\":1}"}); + fake.input = {0x02, 3, '{', '"', 'h', 0x89, 1, 'p', + 0x00, 3, 'b', '"', ':', 0x80, 2, '1', '}'}; + client.poll(); + client.poll(); + assert(messages == std::vector({"{\"hb\":1}", "{\"hb\":1}"})); + assert(frames(fake).size() == 1 && frames(fake)[0].opcode == 0x0a); + fake.input = {0x02, 10, '{', '"', 't', 'e', 'x', 't', '"', ':', '"', 0xc3, + 0x80, 3, 0xa9, '"', '}'}; + client.poll(); + client.poll(); + assert(messages.back() == "{\"text\":\"\xc3\xa9\"}"); } - *sent = write.result == CURLE_OK ? std::min(length, write.bytes) : 0; - const auto* bytes = static_cast(buffer); - fake.output.insert(fake.output.end(), bytes, bytes + *sent); - return write.result; -} - -CURLcode curl_easy_recv(CURL* curl, void* buffer, size_t length, size_t* received) -{ - auto& fake = transport(curl); - if (fake.handshaking) { - const std::string key_header = "Sec-WebSocket-Key: "; - const size_t start = fake.request.find(key_header); - assert(start != std::string::npos); - const size_t value = start + key_header.size(); - const auto accept = opennow::websocket::AcceptForKey( - fake.request.substr(value, fake.request.find("\r\n", value) - value)); - assert(accept); - const std::string response = "HTTP/1.1 101 Switching Protocols\r\n" + - (fake.upgrade == UpgradeResponse::Bare ? std::string() : - "Upgrade: WebSocket\r\nConnection: keep-alive, Upgrade\r\nSec-WebSocket-Accept: " + - (fake.upgrade == UpgradeResponse::WrongAccept ? "synthetic-secret" : *accept) + "\r\n") + "\r\n"; - if (fake.upgrade == UpgradeResponse::SlowDrip && fake.upgrade_offset < 55) { - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - *received = 1; - } else { - *received = response.size() - fake.upgrade_offset; - } - assert(length >= *received); - std::memcpy(buffer, response.data() + fake.upgrade_offset, *received); - fake.upgrade_offset += *received; - fake.handshaking = fake.upgrade_offset != response.size(); - return CURLE_OK; + { + const auto dials = transports.size(); + WebSocketClient expired("ws://example.invalid"); + assert(!expired.connect(Queue::Clock::now())); + assert(expired.failure_kind() == WebSocketClient::FailureKind::Transient); + assert(transports.size() == dials); + next_upgrade = UpgradeResponse::SlowDrip; + WebSocketClient slow("ws://example.invalid"); + const auto started = Queue::Clock::now(); + assert(!slow.connect(started + std::chrono::milliseconds(50))); + assert(Queue::Clock::now() - started < std::chrono::seconds(1)); + assert(slow.failure_kind() == WebSocketClient::FailureKind::Transient); + assert(latest().closed); + assert(latest().connect_timeout_ms > 0 && latest().connect_timeout_ms <= 50); + assert(latest().timeout_ms > 0 && latest().timeout_ms <= 50); } - *received = std::min(length, fake.input.size()); - if (*received == 0) - return fake.receive_result; - std::memcpy(buffer, fake.input.data(), *received); - fake.input.erase(fake.input.begin(), fake.input.begin() + *received); - return CURLE_OK; -} - -} - -int main() -{ - using Queue = opennow::websocket::WriteQueue; for (const CURLcode error : {CURLE_OK, CURLE_RECV_ERROR}) { WebSocketClient client("ws://example.invalid"); assert(client.connect()); @@ -258,9 +125,9 @@ int main() fake.input.insert(fake.input.end(), {0x80, 0}); client.poll(); client.poll(); - assert(messages.empty()); + assert(messages == std::vector{"bi"}); client.poll(); - assert(messages == std::vector{std::string(20, 'x')}); + assert(messages == std::vector({"bi", std::string(20, 'x')})); } { diff --git a/tests/websocket_test_transport.hpp b/tests/websocket_test_transport.hpp new file mode 100644 index 00000000..961ae642 --- /dev/null +++ b/tests/websocket_test_transport.hpp @@ -0,0 +1,191 @@ +#pragma once + +#include "WebSocketClient.hpp" +#include "websocket_handshake.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace +{ + +enum class UpgradeResponse { Valid, Bare, WrongAccept, SlowDrip, Unavailable, Unauthorized }; +UpgradeResponse next_upgrade = UpgradeResponse::Valid; +CURLcode next_perform_result = CURLE_OK; + +struct FakeTransport +{ + struct Write + { + CURLcode result; + size_t bytes; + }; + + bool handshaking = true; + bool closed = false; + size_t calls = 0; + std::deque writes; + std::vector output; + std::vector input; + CURLcode receive_result = CURLE_AGAIN; + UpgradeResponse upgrade = std::exchange(next_upgrade, UpgradeResponse::Valid); + std::string request; + size_t upgrade_offset = 0; + CURLcode perform_result = std::exchange(next_perform_result, CURLE_OK); + long connect_timeout_ms = 0; + long timeout_ms = 0; +}; + +std::vector> transports; + +FakeTransport& transport(CURL* curl) +{ + auto& result = *reinterpret_cast(curl); + assert(!result.closed); + return result; +} + +FakeTransport& latest() +{ + return *transports.back(); +} + +struct Frame +{ + uint8_t opcode; + std::string payload; +}; + +inline std::vector frames(const FakeTransport& fake) +{ + std::vector result; + size_t offset = 0; + while (offset < fake.output.size()) { + const auto& bytes = fake.output; + assert(bytes.size() - offset >= 6); + const uint8_t opcode = bytes[offset++]; + assert(opcode & 0x80); + assert(bytes[offset] & 0x80); + size_t length = bytes[offset++] & 0x7f; + if (length == 126 || length == 127) { + const size_t length_bytes = length == 126 ? 2 : 8; + assert(bytes.size() - offset >= length_bytes + 4); + length = 0; + for (size_t i = 0; i < length_bytes; ++i) + length = (length << 8) | bytes[offset++]; + } + const size_t mask = offset; + offset += 4; + assert(bytes.size() - offset >= length); + std::string payload; + for (size_t i = 0; i < length; ++i) + payload.push_back(bytes[offset++] ^ bytes[mask + i % 4]); + result.push_back({static_cast(opcode & 0x0f), std::move(payload)}); + } + return result; +} + +} + +extern "C" { + +CURL* curl_easy_init() +{ + transports.push_back(std::make_unique()); + return reinterpret_cast(transports.back().get()); +} + +void curl_easy_cleanup(CURL* curl) +{ + transport(curl).closed = true; +} + +CURLcode curl_easy_setopt(CURL* curl, CURLoption option, ...) +{ + if (option == CURLOPT_CONNECTTIMEOUT_MS || option == CURLOPT_TIMEOUT_MS) { + va_list arguments; + va_start(arguments, option); + const long value = va_arg(arguments, long); + va_end(arguments); + if (option == CURLOPT_CONNECTTIMEOUT_MS) + transport(curl).connect_timeout_ms = value; + else + transport(curl).timeout_ms = value; + } + return CURLE_OK; +} + +CURLcode curl_easy_perform(CURL* curl) +{ + return transport(curl).perform_result; +} + +const char* curl_easy_strerror(CURLcode) +{ + return "fake transport error"; +} + +CURLcode curl_easy_send(CURL* curl, const void* buffer, size_t length, size_t* sent) +{ + auto& fake = transport(curl); + if (fake.handshaking) { + fake.request.append(static_cast(buffer), length); + *sent = length; + return CURLE_OK; + } + ++fake.calls; + assert(length <= opennow::websocket::WriteQueue::MaximumWriteBytes); + FakeTransport::Write write{CURLE_OK, length}; + if (!fake.writes.empty()) { + write = fake.writes.front(); + fake.writes.pop_front(); + } + *sent = write.result == CURLE_OK ? std::min(length, write.bytes) : 0; + const auto* bytes = static_cast(buffer); + fake.output.insert(fake.output.end(), bytes, bytes + *sent); + return write.result; +} + +CURLcode curl_easy_recv(CURL* curl, void* buffer, size_t length, size_t* received) +{ + auto& fake = transport(curl); + if (fake.handshaking) { + const std::string key_header = "Sec-WebSocket-Key: "; + const size_t start = fake.request.find(key_header); + assert(start != std::string::npos); + const size_t value = start + key_header.size(); + const auto accept = opennow::websocket::AcceptForKey( + fake.request.substr(value, fake.request.find("\r\n", value) - value)); + assert(accept); + const std::string status = fake.upgrade == UpgradeResponse::Unavailable ? "503 Unavailable" + : fake.upgrade == UpgradeResponse::Unauthorized ? "403 Forbidden" : "101 Switching Protocols"; + const std::string response = "HTTP/1.1 " + status + "\r\n" + + (fake.upgrade == UpgradeResponse::Bare ? std::string() : + "Upgrade: WebSocket\r\nConnection: keep-alive, Upgrade\r\nSec-WebSocket-Accept: " + + (fake.upgrade == UpgradeResponse::WrongAccept ? "synthetic-secret" : *accept) + "\r\n") + "\r\n"; + if (fake.upgrade == UpgradeResponse::SlowDrip && fake.upgrade_offset < 55) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + *received = 1; + } else { + *received = response.size() - fake.upgrade_offset; + } + assert(length >= *received); + std::memcpy(buffer, response.data() + fake.upgrade_offset, *received); + fake.upgrade_offset += *received; + fake.handshaking = fake.upgrade_offset != response.size(); + return CURLE_OK; + } + *received = std::min(length, fake.input.size()); + if (*received == 0) + return fake.receive_result; + std::memcpy(buffer, fake.input.data(), *received); + fake.input.erase(fake.input.begin(), fake.input.begin() + *received); + return CURLE_OK; +} + +} From 1d95835f4f01d9fb4d82f6077323fb830ff955c2 Mon Sep 17 00:00:00 2001 From: zortos293 <65777760+zortos293@users.noreply.github.com> Date: Thu, 29 Oct 2026 19:31:36 +0000 Subject: [PATCH 2/3] Preserve legacy signaling and provider endpoint ports Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- app/src/gfn/cloud_session_protocol.cpp | 55 ++++++++++++++++--------- tests/cloud_session_allocation_test.cpp | 11 ++++- tests/cloud_session_recovery_test.cpp | 1 + 3 files changed, 46 insertions(+), 21 deletions(-) diff --git a/app/src/gfn/cloud_session_protocol.cpp b/app/src/gfn/cloud_session_protocol.cpp index 97c6ad6c..86a76d73 100644 --- a/app/src/gfn/cloud_session_protocol.cpp +++ b/app/src/gfn/cloud_session_protocol.cpp @@ -169,20 +169,35 @@ int GetConnectionPort(json_t* connection) return ParseEndpoint(resource).port; } -std::string BuildSignalingUrlFromConnection(json_t* connection) +std::string BuildSignalingUrlFromConnection(json_t* connection, bool allow_legacy = false) { - if (IsNativeConnection(connection) || WireInteger(json_object_get(connection, "usage")) != 14) + const int usage = WireInteger(json_object_get(connection, "usage")); + const bool legacy = IsNativeConnection(connection); + if ((usage != 14 && usage != 16) || (legacy && !allow_legacy)) return {}; const std::string resource_path = GetString(connection, "resourcePath"); - if (resource_path.find("://") != std::string::npos) + if (HasPrefix(resource_path, "wss://") || HasPrefix(resource_path, "https://")) return ValidSignalingUrl(resource_path); - if (!resource_path.empty() && resource_path.front() != '/') + if (!resource_path.empty() && resource_path.front() != '/' && + !(legacy && (HasPrefix(resource_path, "rtsp://") || HasPrefix(resource_path, "rtsps://")))) return {}; - const std::string ip = GetConnectionIp(connection); + const bool legacy_resource = legacy && + (HasPrefix(resource_path, "rtsp://") || HasPrefix(resource_path, "rtsps://")); + const std::string ip = legacy_resource ? ParseEndpoint(resource_path).host : GetConnectionIp(connection); if (ip.empty()) return {}; const std::string host = ip.find(':') == std::string::npos ? ip : "[" + ip + "]"; - return ValidSignalingUrl("wss://" + host + ":443" + (resource_path.empty() ? "/nvst/" : resource_path)); + if (legacy) + return ValidSignalingUrl("wss://" + host + "/nvst/"); + int port = 443; + if (WireInteger(json_object_get(connection, "appLevelProtocol")) == 4 && + json_object_get(connection, "port")) { + port = GetConnectionPort(connection); + if (port <= 0) + return {}; + } + return ValidSignalingUrl("wss://" + host + ":" + std::to_string(port) + + (resource_path.empty() ? "/nvst/" : resource_path)); } void AppendIceServer(SessionInfo& info, std::string url, std::string username, std::string credential) @@ -332,18 +347,20 @@ void ApplySessionNetworkInfo(json_t* sess, SessionInfo& info) } if (info.signaling_url.empty()) { - for (bool explicit_url : {true, false}) { - size_t index; - json_t* conn_info; - json_array_foreach(connection_info_array, index, conn_info) { - CollectIceServersFromObject(conn_info, info); - - if (WireInteger(json_object_get(conn_info, "usage")) != 14) - continue; - const std::string resource = GetString(conn_info, "resourcePath"); - if ((HasPrefix(resource, "wss://") || HasPrefix(resource, "https://")) != explicit_url) - continue; - info.signaling_url = BuildSignalingUrlFromConnection(conn_info); + for (bool legacy : {false, true}) { + for (bool explicit_url : {true, false}) { + size_t index; + json_t* conn_info; + json_array_foreach(connection_info_array, index, conn_info) { + if (IsNativeConnection(conn_info) != legacy) + continue; + const std::string resource = GetString(conn_info, "resourcePath"); + if ((HasPrefix(resource, "wss://") || HasPrefix(resource, "https://")) != explicit_url) + continue; + info.signaling_url = BuildSignalingUrlFromConnection(conn_info, legacy); + if (!info.signaling_url.empty()) + break; + } if (!info.signaling_url.empty()) break; } @@ -379,7 +396,7 @@ std::string ResolveSessionControlBaseUrl(json_t* session, const std::string& fal return fallback; const Endpoint origin = ParseEndpoint(fallback); if (host == Lowercase(origin.host)) - return "https://" + (host.find(':') == std::string::npos ? host : "[" + host + "]") + "/"; + return fallback; const auto nvidia_host = [](const std::string& value) { return value.ends_with(".cloudmatchbeta.nvidiagrid.net") || value.ends_with(".cloudmatch.nvidiagrid.net") || diff --git a/tests/cloud_session_allocation_test.cpp b/tests/cloud_session_allocation_test.cpp index 4763b6be..0ff1b290 100644 --- a/tests/cloud_session_allocation_test.cpp +++ b/tests/cloud_session_allocation_test.cpp @@ -18,8 +18,15 @@ int main(int argc, char** argv) {R"({"connectionInfo":[{"usage":14,"resourcePath":"wss://user:secret@bad.example/nvst/"},{"usage":14,"ip":"0.0.0.0","port":443},{"usage":14,"resourcePath":"wss://good.example:8443/nvst/"}]})", "wss://good.example:8443/nvst/", "good.example", 8443}, {R"({"connectionInfo":[{"usage":14,"ip":"2001:db8::10","port":49003,"resourcePath":"/nvst/"}]})", "wss://[2001:db8::10]:443/nvst/", "2001:db8::10", 49003}, {R"({"signalingUrl":"https://seat.example:9443/custom/","connectionInfo":[{"usage":2,"ip":"203.0.113.12","port":49002}]})", "wss://seat.example:9443/custom/", "203.0.113.12", 49002}, - {R"({"connectionInfo":[{"usage":16,"appLevelProtocol":6,"ip":"classic.example","port":48322}]})", "", "", 0}, + {R"({"connectionInfo":[{"usage":16,"appLevelProtocol":6,"ip":"classic.example","port":48322}]})", "wss://classic.example/nvst/", "", 0}, {R"({"connectionInfo":[{"usage":14,"resourcePath":"wss://[not-ip.example]/nvst/"},{"usage":14,"resourcePath":"wss://seat.example:65536/nvst/"},{"usage":14,"resourcePath":"wss://seat.example:443junk/nvst/"},{"usage":14,"resourcePath":"http://seat.example/nvst/"},{"usage":14,"resourcePath":"wss://good.example:9443/nvst/"}]})", "wss://good.example:9443/nvst/", "good.example", 9443}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":6,"resourcePath":"rtsps://legacy.example:48322"}]})", "wss://legacy.example/nvst/", "", 0}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":1,"ip":"legacy.example","port":48010}]})", "wss://legacy.example/nvst/", "", 0}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":4,"ip":"webrtc.example","port":8443,"resourcePath":"/nvst/"}]})", "wss://webrtc.example:8443/nvst/", "webrtc.example", 8443}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":4,"ip":"2001:db8::1","port":8443}]})", "wss://[2001:db8::1]:8443/nvst/", "2001:db8::1", 8443}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":6,"resourcePath":"rtsps://legacy.example:322"},{"usage":14,"appLevelProtocol":6,"resourcePath":"wss://legacy-web.example:9443/custom/"}]})", "wss://legacy-web.example:9443/custom/", "", 0}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":6,"ip":"203.0.113.10","resourcePath":"rtsps://legacy.example:48322"}]})", "wss://legacy.example/nvst/", "", 0}, + {R"({"connectionInfo":[{"usage":14,"appLevelProtocol":6,"ip":"203.0.113.10","resourcePath":"rtsps://user:secret@bad.example:322"},{"usage":14,"appLevelProtocol":6,"resourcePath":"rtsps://legacy.example:48322"}]})", "wss://legacy.example/nvst/", "", 0}, }; for (int i = 0; i < static_cast(std::size(cases)); ++i) { @@ -32,7 +39,7 @@ int main(int argc, char** argv) assert(info.media_ip == cases[i].media); assert(info.media_port == cases[i].port); } - if (selected < 0 || selected == 11) + if (selected < 0 || selected == static_cast(std::size(cases))) { for (const char* text : {"2", "\"2\"", "\"ready\""}) { diff --git a/tests/cloud_session_recovery_test.cpp b/tests/cloud_session_recovery_test.cpp index 846c7af8..826f644a 100644 --- a/tests/cloud_session_recovery_test.cpp +++ b/tests/cloud_session_recovery_test.cpp @@ -170,6 +170,7 @@ int main(int argc, char** argv) } auto root = opennow::gfn::detail::LoadJson(R"({"sessionControlInfo":{"ip":"seat.partner.example","port":443}})"); assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(root.get(), "https://seat.partner.example/") == "https://seat.partner.example/"); + assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(root.get(), "https://seat.partner.example:8443/") == "https://seat.partner.example:8443/"); assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(root.get(), initial) == initial); auto nvidia = opennow::gfn::detail::LoadJson(R"({"sessionControlInfo":{"ip":"NP-PAR-01.cloudmatchbeta.nvidiagrid.net","port":"443"}})"); assert(opennow::gfn::cloud_session::ResolveSessionControlBaseUrl(nvidia.get(), initial) == assigned); From ac3f8dee655f9f40c16639d73c71c0f5bde472c3 Mon Sep 17 00:00:00 2001 From: zortos293 <65777760+zortos293@users.noreply.github.com> Date: Thu, 29 Oct 2026 19:34:54 +0000 Subject: [PATCH 3/3] Preserve sparse patching and empty stop acknowledgments Co-authored-by: capy-ai[bot] <230910855+capy-ai[bot]@users.noreply.github.com> --- app/src/gfn/cloud_session_protocol.cpp | 6 ++++-- tests/cloud_session_recovery_test.cpp | 13 +++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/app/src/gfn/cloud_session_protocol.cpp b/app/src/gfn/cloud_session_protocol.cpp index 86a76d73..b637a045 100644 --- a/app/src/gfn/cloud_session_protocol.cpp +++ b/app/src/gfn/cloud_session_protocol.cpp @@ -415,8 +415,8 @@ bool CloudMatchResponseSucceeded(const HttpResponse& response) return true; try { JsonPtr root = LoadJson(response.body); - return json_is_object(root.get()) && WireInteger(json_object_get( - json_object_get(root.get(), "requestStatus"), "statusCode"), -1) == 1; + return json_is_object(root.get()) && (json_object_size(root.get()) == 0 || + WireInteger(json_object_get(json_object_get(root.get(), "requestStatus"), "statusCode"), -1) == 1); } catch (...) { return false; } @@ -436,6 +436,8 @@ SessionInfo ParseSessionResponse(const HttpResponse& response, const std::string json_t* sess = json_object_get(root.get(), "session"); SessionInfo info; info.session_id = GetString(sess, "sessionId"); + if (patching && !json_object_get(sess, "sessionId")) + info.session_id = expected_id; if (info.session_id.empty() || info.session_id.find_first_of("/?#\\\r\n") != std::string::npos || (!expected_id.empty() && expected_id != info.session_id)) throw std::runtime_error(stage + " returned an invalid or different session ID."); diff --git a/tests/cloud_session_recovery_test.cpp b/tests/cloud_session_recovery_test.cpp index 826f644a..83441990 100644 --- a/tests/cloud_session_recovery_test.cpp +++ b/tests/cloud_session_recovery_test.cpp @@ -239,6 +239,19 @@ int main(int argc, char** argv) } opennow::gfn::cloud_session::ForgetActiveCloudSession("seat"); } + if (selected < 0 || selected == 11) { + expect("GET", initial + "v2/session/seat", 200, + R"({"requestStatus":{"statusCode":41,"statusDescription":"APP_PATCHING_STATUS"}})"); + const auto info = client.PollSession(auth, "seat"); + assert(info.session_id == "seat" && info.app_patching && info.status == 1); + assert(!info.ready_for_stream() && info.signaling_url.empty()); + } + if (selected < 0 || selected == 12) { + opennow::gfn::cloud_session::RememberActiveCloudSession(auth, "seat", "123", initial); + expect("DELETE", initial + "v2/session/seat", 200, "{}"); + client.StopSession(auth, "seat"); + assert(opennow::gfn::cloud_session::LoadActiveCloudSessionId(auth).empty()); + } assert(replies.empty()); std::filesystem::current_path(cwd); std::filesystem::remove_all(directory);