feat(qconnect): add transactional LAN delegation

cx
This commit is contained in:
vicrodh committed 2026-08-30 15:33:23 -06:00
1 parent f30abd12ac
commit 74a41150e9
19 files changed
+4461 -297

No files matched your search

+12 -2
View File
@@ -4485,6 +4485,7 @@ dependencies = [
"tempfile",
"thiserror 2.0.20",
"tokio",
"zeroize",
]
[[package]]
@@ -4709,11 +4710,13 @@ name = "qconnect-app"
version = "2.0.2"
dependencies = [
"async-trait",
"futures-util",
"log",
"prost",
"qbz-models",
"qbz-player",
"qconnect-core",
"qconnect-lan",
"qconnect-protocol",
"qconnect-transport-ws",
"serde",
@@ -4736,17 +4739,22 @@ dependencies = [
name = "qconnect-lan"
version = "2.0.2"
dependencies = [
"if-addrs 0.15.0",
"base64 0.23.1",
"mdns-sd",
"qbz-credentials",
"qbz-qobuz",
"qconnect-core",
"qconnect-protocol",
"qconnect-transport-ws",
"reqwest",
"rustls",
"serde",
"serde_json",
"socket2",
"thiserror 2.0.20",
"tiny_http",
"tokio",
"url",
"uuid 1.26.0",
"zeroize",
]
@@ -4773,9 +4781,11 @@ dependencies = [
"prost",
"qconnect-protocol",
"serde",
"serde_json",
"thiserror 2.0.20",
"tokio",
"tokio-tungstenite",
"zeroize",
]
[[package]]
+1
View File
@@ -23,6 +23,7 @@ serde_json = { workspace = true }
# Crypto (for Qobuz auth)
md-5 = { workspace = true }
base64 = { workspace = true }
zeroize = "1"
# Time
chrono = { workspace = true }
+173 -15
View File
@@ -11,6 +11,7 @@ use super::auth::{
sign_request, sign_search, sign_session_start,
};
use super::bundle::{self, BundleTokens};
use super::delegated::{DelegatedApiConfigError, DelegatedAppCredentials};
use super::endpoints::{self, paths};
use super::error::{ApiError, Result};
use super::forbidden_breaker::ForbiddenBreaker;
@@ -18,6 +19,7 @@ use super::lyrics::{
merge_translation_into, QobuzLyricsContent, QobuzLyricsDocument, QobuzLyricsUrls,
};
use qbz_models::*;
use zeroize::Zeroizing;
const USER_AGENT: &str = "Mozilla/5.0 (X11; Linux x86_64; rv:120.0) Gecko/20100101 Firefox/120.0";
@@ -47,12 +49,45 @@ struct CmafSession {
expires_at: u64,
}
/// The app secret which has actually passed the API probe, together with the
/// app ID of the bundle generation it belongs to.
///
/// Keeping the provenance beside the secret prevents a background bundle
/// rotation from silently pairing a previously validated secret with a new
/// app ID. The secret is app-level material, never a user session token, and
/// is wiped when replaced or when the last client owner is dropped.
struct ValidatedBundleSecret {
app_id: String,
signing_secret: Zeroizing<String>,
}
impl ValidatedBundleSecret {
fn new(app_id: String, signing_secret: String) -> Self {
Self {
app_id,
signing_secret: Zeroizing::new(signing_secret),
}
}
fn is_current_for(&self, tokens: &BundleTokens) -> bool {
self.app_id == tokens.app_id
&& tokens
.secrets
.iter()
.any(|candidate| candidate == self.signing_secret.as_str())
}
fn copy_material(&self) -> (String, String) {
(self.app_id.clone(), self.signing_secret.to_string())
}
}
/// Qobuz API client
pub struct QobuzClient {
http: Client,
tokens: Arc<RwLock<Option<BundleTokens>>>,
session: Arc<RwLock<Option<UserSession>>>,
validated_secret: Arc<RwLock<Option<String>>>,
validated_secret: Arc<RwLock<Option<ValidatedBundleSecret>>>,
locale: Arc<RwLock<String>>,
cmaf_session: Arc<RwLock<Option<CmafSession>>>,
/// Where the regenerable bundle-token cache goes, when the embedder rather
@@ -169,6 +204,7 @@ impl QobuzClient {
let version = cached.bundle_version.clone();
log::info!("[Bundle] Using cached tokens (version {})", version);
*self.tokens.write().await = Some(cached.into());
*self.validated_secret.write().await = None;
// Cache reads are never gated, but the background refresh is a
// network request — gate it once before cloning the client into
@@ -178,6 +214,7 @@ impl QobuzClient {
Ok(client) => {
let client = client.clone();
let tokens_arc = Arc::clone(&self.tokens);
let validated_secret = Arc::clone(&self.validated_secret);
let cache_dir = self.bundle_cache_dir.clone();
tokio::spawn(async move {
if let Some(fresh) = bundle::refresh_bundle_if_changed_in(
@@ -188,6 +225,7 @@ impl QobuzClient {
.await
{
*tokens_arc.write().await = Some(fresh);
*validated_secret.write().await = None;
log::info!("[Bundle] Background refresh applied rotated tokens");
}
});
@@ -209,6 +247,7 @@ impl QobuzClient {
)
.await?;
*self.tokens.write().await = Some(tokens);
*self.validated_secret.write().await = None;
Ok(false)
}
@@ -259,29 +298,74 @@ impl QobuzClient {
Ok(&self.http)
}
/// Get validated secret (validates on first use)
pub(crate) async fn secret(&self) -> Result<String> {
// Check if we already have a validated secret
if let Some(secret) = self.validated_secret.read().await.clone() {
return Ok(secret);
}
/// Return app ID and signing secret from one validated bundle generation.
async fn validated_app_material(&self) -> Result<(String, String)> {
let (app_id, candidates) = {
let tokens = self.tokens.read().await;
let tokens = tokens.as_ref().ok_or_else(|| {
ApiError::BundleExtractionError("Client not initialized".to_string())
})?;
// Need to validate secrets
let tokens = self.tokens.read().await;
let tokens = tokens
.as_ref()
.ok_or_else(|| ApiError::BundleExtractionError("Client not initialized".to_string()))?;
if let Some(validated) = self.validated_secret.read().await.as_ref() {
if validated.is_current_for(tokens) {
return Ok(validated.copy_material());
}
}
for secret in &tokens.secrets {
(
tokens.app_id.clone(),
Zeroizing::new(tokens.secrets.clone()),
)
};
for secret in candidates.iter() {
if self.test_secret(secret).await? {
*self.validated_secret.write().await = Some(secret.clone());
return Ok(secret.clone());
// A background refresh is allowed while the network probe is
// running. Accept its result only if the exact app material is
// still present in the current bundle generation.
let tokens = self.tokens.read().await;
let current = tokens.as_ref().ok_or_else(|| {
ApiError::BundleExtractionError("Client not initialized".to_string())
})?;
if current.app_id != app_id
|| !current.secrets.iter().any(|candidate| candidate == secret)
{
return Err(ApiError::BundleExtractionError(
"Bundle credentials changed during validation".to_string(),
));
}
let validated = ValidatedBundleSecret::new(app_id.clone(), secret.clone());
let material = validated.copy_material();
*self.validated_secret.write().await = Some(validated);
return Ok(material);
}
}
Err(ApiError::InvalidAppSecret)
}
/// Get the validated signing secret (validates bundle candidates on first use).
pub(crate) async fn secret(&self) -> Result<String> {
self.validated_app_material()
.await
.map(|(_, signing_secret)| signing_secret)
}
/// Copy only validated app-level material into an isolated QConnect guest.
///
/// The returned value does not contain, share, or copy this client's
/// [`UserSession`], user auth token, OAuth private key, cookies, or HTTP
/// client. In particular, it uses the secret selected by the existing API
/// validation probe; it never guesses by taking the first bundle secret.
pub async fn delegated_app_credentials(&self) -> Result<DelegatedAppCredentials> {
let (app_id, signing_secret) = self.validated_app_material().await?;
DelegatedAppCredentials::new(app_id, signing_secret).map_err(|error| match error {
DelegatedApiConfigError::InvalidAppId => ApiError::InvalidAppId,
_ => ApiError::InvalidAppSecret,
})
}
/// Test if a secret is valid using a known track
async fn test_secret(&self, secret: &str) -> Result<bool> {
let test_track_id = 5966783u64; // Known test track
@@ -3198,6 +3282,80 @@ mod tests {
});
}
#[tokio::test]
async fn delegated_credentials_use_the_validated_secret_not_the_first_candidate() {
ensure_crypto_provider();
let client = QobuzClient::new().expect("client construction is local-only");
let app_id = "123456789";
let first_candidate = "first-unvalidated-secret";
let selected_secret = "second-validated-secret";
*client.tokens.write().await = Some(BundleTokens {
app_id: app_id.to_string(),
secrets: vec![first_candidate.to_string(), selected_secret.to_string()],
private_key: Some("owner-oauth-private-key".to_string()),
});
*client.validated_secret.write().await = Some(ValidatedBundleSecret::new(
app_id.to_string(),
selected_secret.to_string(),
));
*client.session.write().await = Some(UserSession {
user_auth_token: "owner-user-auth-token".to_string(),
user_id: 7,
email: "owner@example.test".to_string(),
display_name: "Owner".to_string(),
subscription_label: "Studio".to_string(),
subscription_valid_until: None,
country_code: Some("MX".to_string()),
language_code: Some("es".to_string()),
});
let credentials = client.delegated_app_credentials().await.unwrap();
let (delegated_app_id, delegated_secret) = credentials.values_for_test();
assert_eq!(delegated_app_id, app_id);
assert_eq!(delegated_secret, selected_secret);
assert_ne!(delegated_secret, first_candidate);
assert_ne!(delegated_secret, "owner-oauth-private-key");
assert_ne!(delegated_secret, "owner-user-auth-token");
}
#[test]
fn validated_secret_is_bound_to_its_bundle_generation() {
let validated =
ValidatedBundleSecret::new("123456789".to_string(), "selected-secret".to_string());
let matching = BundleTokens {
app_id: "123456789".to_string(),
secrets: vec!["other-secret".to_string(), "selected-secret".to_string()],
private_key: None,
};
let rotated_secret = BundleTokens {
app_id: "123456789".to_string(),
secrets: vec!["replacement-secret".to_string()],
private_key: None,
};
let rotated_app = BundleTokens {
app_id: "987654321".to_string(),
secrets: vec!["selected-secret".to_string()],
private_key: None,
};
assert!(validated.is_current_for(&matching));
assert!(!validated.is_current_for(&rotated_secret));
assert!(!validated.is_current_for(&rotated_app));
}
#[tokio::test]
async fn delegated_credentials_require_an_initialized_owner_client() {
ensure_crypto_provider();
let client = QobuzClient::new().expect("client construction is local-only");
assert!(matches!(
client.delegated_app_credentials().await,
Err(ApiError::BundleExtractionError(message))
if message == "Client not initialized"
));
}
/// With the offline gate closed, any public API method must fail fast
/// with the typed `ApiError::OfflineMode` — no network access, no
/// connect timeout. The gate is process-global and tests run in
+867
View File
@@ -0,0 +1,867 @@
//! Isolated Qobuz API access for a delegated QConnect guest session.
//!
//! This client owns its HTTP stack and credentials. It never reads or changes
//! [`crate::QobuzClient`] owner-session state, OAuth credentials, keyrings, or
//! global headers. The delegated JWT is sent only as `Authorization: Bearer`,
//! alongside `X-App-Id`; `X-User-Auth-Token` is deliberately unavailable.
//!
//! The surface is intentionally read-only and transient: access validation,
//! track metadata, and stream URL resolution. There is no persistence, token
//! refresh, library/account operation, or implicit owner fallback here.
use crate::auth::get_timestamp;
use crate::endpoints::paths;
use md5::{Digest, Md5};
use qbz_models::{Quality, StreamRestriction, StreamUrl, Track};
use reqwest::header::{HeaderMap, HeaderValue, AUTHORIZATION};
use reqwest::{Client, RequestBuilder, Response, StatusCode, Url};
use serde::de::DeserializeOwned;
use serde_json::Value;
use std::fmt;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use thiserror::Error;
use zeroize::Zeroizing;
const MAX_ENDPOINT_BYTES: usize = 2_048;
const MAX_JWT_BYTES: usize = 16 * 1_024;
const MAX_APP_ID_BYTES: usize = 256;
const MAX_SIGNING_SECRET_BYTES: usize = 16 * 1_024;
const MIN_INITIAL_TTL: Duration = Duration::from_secs(60);
const TRACKS_PER_BATCH: usize = 50;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
/// A configuration rejection which never embeds the rejected value.
#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
pub enum DelegatedApiConfigError {
#[error("delegated API endpoint is invalid")]
InvalidEndpoint,
#[error("delegated API endpoint exceeds the size limit")]
EndpointTooLong,
#[error("delegated API endpoint must use HTTPS on port 443")]
InsecureEndpoint,
#[error("delegated API endpoint host is not allowed")]
EndpointHostNotAllowed,
#[error("delegated API app ID is invalid")]
InvalidAppId,
#[error("delegated API signing secret is invalid")]
InvalidSigningSecret,
#[error("delegated API credential is invalid")]
InvalidJwt,
#[error("delegated API credential is expired")]
Expired,
#[error("delegated API credential has insufficient lifetime")]
InsufficientLifetime,
}
/// Runtime errors for delegated, read-only Qobuz API operations.
///
/// Network and response errors are intentionally sanitized: reqwest errors
/// may carry the delegated endpoint URL, which must not reach Debug/log output.
#[derive(Debug, Error, Clone, Copy, PartialEq, Eq)]
pub enum DelegatedApiError {
#[error(transparent)]
Configuration(#[from] DelegatedApiConfigError),
#[error("delegated API credential is expired")]
Expired,
#[error("offline mode is active - delegated Qobuz services are disabled")]
OfflineMode,
#[error("delegated API HTTP client could not be initialized")]
ClientInitialization,
#[error("delegated API network request failed")]
Network,
#[error("delegated API credential was rejected")]
Unauthorized,
#[error("delegated API access was forbidden")]
Forbidden,
#[error("delegated API request signature was rejected")]
InvalidSignature,
#[error("delegated API returned an invalid response")]
InvalidResponse,
#[error("delegated API request was rejected (HTTP {0})")]
RequestRejected(u16),
#[error("delegated API was rate limited")]
RateLimited,
#[error("delegated API server error (HTTP {0})")]
ServerError(u16),
#[error("track {0} is no longer available on Qobuz")]
TrackUnavailable(u64),
#[error("no valid quality is available for this track")]
NoQualityAvailable,
}
impl DelegatedApiError {
pub fn is_transient(&self) -> bool {
matches!(
self,
Self::Network | Self::RateLimited | Self::ServerError(_)
)
}
}
pub type DelegatedApiResult<T> = std::result::Result<T, DelegatedApiError>;
/// An HTTPS API base whose host has passed the caller's closed allowlist.
///
/// The allowlist belongs to the QConnect adapter because it is established by
/// official-client captures. Passing it explicitly prevents this lower-level
/// crate from silently broadening the accepted authority later.
pub struct DelegatedApiEndpoint {
base_url: Zeroizing<String>,
}
impl DelegatedApiEndpoint {
pub fn new(
endpoint: &str,
allowed_hosts: &[&str],
) -> std::result::Result<Self, DelegatedApiConfigError> {
if endpoint.len() > MAX_ENDPOINT_BYTES {
return Err(DelegatedApiConfigError::EndpointTooLong);
}
let mut base_url =
Url::parse(endpoint).map_err(|_| DelegatedApiConfigError::InvalidEndpoint)?;
if base_url.cannot_be_a_base()
|| !base_url.username().is_empty()
|| base_url.password().is_some()
|| base_url.query().is_some()
|| base_url.fragment().is_some()
{
return Err(DelegatedApiConfigError::InvalidEndpoint);
}
if base_url.scheme() != "https" || base_url.port_or_known_default() != Some(443) {
return Err(DelegatedApiConfigError::InsecureEndpoint);
}
let host = base_url
.host_str()
.ok_or(DelegatedApiConfigError::InvalidEndpoint)?;
if allowed_hosts.is_empty()
|| !allowed_hosts
.iter()
.any(|allowed| host.eq_ignore_ascii_case(allowed))
{
return Err(DelegatedApiConfigError::EndpointHostNotAllowed);
}
if !base_url.path().ends_with('/') {
let normalized_path = format!("{}/", base_url.path());
base_url.set_path(&normalized_path);
}
Ok(Self {
base_url: Zeroizing::new(base_url.to_string()),
})
}
fn join(&self, path: &str) -> DelegatedApiResult<Url> {
Url::parse(self.base_url.as_str())
.map_err(|_| {
DelegatedApiError::Configuration(DelegatedApiConfigError::InvalidEndpoint)
})?
.join(path.trim_start_matches('/'))
.map_err(|_| DelegatedApiError::Configuration(DelegatedApiConfigError::InvalidEndpoint))
}
}
impl fmt::Debug for DelegatedApiEndpoint {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("DelegatedApiEndpoint")
.field("base_url", &"<redacted>")
.finish()
}
}
/// App-level material copied into, and owned exclusively by, a guest client.
///
/// It intentionally does not implement `Clone`; the signing secret is wiped on
/// drop. Callers obtain this opaque value through
/// [`crate::QobuzClient::delegated_app_credentials`], which copies only the
/// app ID and the app secret selected by the owner's validation probe.
pub struct DelegatedAppCredentials {
app_id: String,
signing_secret: Zeroizing<String>,
}
impl DelegatedAppCredentials {
pub(crate) fn new(
app_id: String,
signing_secret: String,
) -> std::result::Result<Self, DelegatedApiConfigError> {
let signing_secret = Zeroizing::new(signing_secret);
if app_id.is_empty()
|| app_id.len() > MAX_APP_ID_BYTES
|| app_id.chars().any(char::is_whitespace)
|| HeaderValue::from_str(&app_id).is_err()
{
return Err(DelegatedApiConfigError::InvalidAppId);
}
if signing_secret.is_empty()
|| signing_secret.len() > MAX_SIGNING_SECRET_BYTES
|| signing_secret.chars().any(char::is_whitespace)
{
return Err(DelegatedApiConfigError::InvalidSigningSecret);
}
Ok(Self {
app_id,
signing_secret,
})
}
#[cfg(test)]
pub(crate) fn values_for_test(&self) -> (&str, &str) {
(&self.app_id, self.signing_secret.as_str())
}
}
impl fmt::Debug for DelegatedAppCredentials {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("DelegatedAppCredentials")
.field("app_id", &"<redacted>")
.field("signing_secret", &"<redacted>")
.finish()
}
}
/// Complete, in-memory configuration consumed when creating a guest client.
pub struct DelegatedApiConfig {
endpoint: DelegatedApiEndpoint,
expires_at: u64,
jwt_api: Zeroizing<String>,
app_credentials: DelegatedAppCredentials,
}
impl DelegatedApiConfig {
pub fn new(
endpoint: DelegatedApiEndpoint,
expires_at: u64,
jwt_api: String,
app_credentials: DelegatedAppCredentials,
) -> std::result::Result<Self, DelegatedApiConfigError> {
let jwt_api = Zeroizing::new(jwt_api);
let now = unix_timestamp();
if expires_at <= now {
return Err(DelegatedApiConfigError::Expired);
}
if expires_at.saturating_sub(now) < MIN_INITIAL_TTL.as_secs() {
return Err(DelegatedApiConfigError::InsufficientLifetime);
}
validate_jwt(&jwt_api)?;
Ok(Self {
endpoint,
expires_at,
jwt_api,
app_credentials,
})
}
}
impl fmt::Debug for DelegatedApiConfig {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("DelegatedApiConfig")
.field("endpoint", &self.endpoint)
.field("expires_at", &self.expires_at)
.field("jwt_api", &"<redacted>")
.field("app_credentials", &self.app_credentials)
.finish()
}
}
/// A read-only Qobuz client whose authority is one delegated `jwt_api`.
///
/// The type is not clonable. Callers that need shared ownership can place the
/// whole isolated context in an `Arc`; this never shares state with the owner
/// [`crate::QobuzClient`].
pub struct DelegatedQobuzClient {
http: Client,
endpoint: DelegatedApiEndpoint,
expires_at: u64,
jwt_api: Zeroizing<String>,
app_id: String,
signing_secret: Zeroizing<String>,
}
impl DelegatedQobuzClient {
pub fn new(config: DelegatedApiConfig) -> DelegatedApiResult<Self> {
let http = Client::builder()
.redirect(reqwest::redirect::Policy::none())
.connect_timeout(CONNECT_TIMEOUT)
.timeout(REQUEST_TIMEOUT)
.build()
.map_err(|_| DelegatedApiError::ClientInitialization)?;
Ok(Self {
http,
endpoint: config.endpoint,
expires_at: config.expires_at,
jwt_api: config.jwt_api,
app_id: config.app_credentials.app_id,
signing_secret: config.app_credentials.signing_secret,
})
}
pub fn expires_at(&self) -> u64 {
self.expires_at
}
/// Prove the delegated Bearer against the observed non-mutating endpoint.
pub async fn validate_access(&self) -> DelegatedApiResult<()> {
let url = self.endpoint.join(paths::USER_GET)?;
let response = self.execute(self.http.get(url)).await?;
ensure_success(response.status(), None, false)?;
let value: Value = read_json(response).await?;
let valid_user_shape = value.as_object().is_some_and(|object| {
object.contains_key("id")
|| object.contains_key("user")
|| object.contains_key("credential")
});
if !valid_user_shape {
return Err(DelegatedApiError::InvalidResponse);
}
Ok(())
}
/// Resolve read-only metadata for one track with delegated authority.
pub async fn get_track(&self, track_id: u64) -> DelegatedApiResult<Track> {
let response = self
.signed_get(
paths::TRACK_GET,
"trackget",
&[("track_id", track_id.to_string())],
)
.await?;
ensure_success(response.status(), Some(track_id), true)?;
read_json(response).await
}
/// Resolve read-only metadata in the API's 50-track request windows.
pub async fn get_tracks_batch(&self, track_ids: &[u64]) -> DelegatedApiResult<Vec<Track>> {
if track_ids.is_empty() {
return Ok(Vec::new());
}
let mut tracks = Vec::with_capacity(track_ids.len());
for chunk in track_ids.chunks(TRACKS_PER_BATCH) {
tracks.extend(self.get_tracks_batch_chunk(chunk).await?);
}
Ok(tracks)
}
/// Resolve one signed streaming URL using only the delegated Bearer.
pub async fn get_stream_url(
&self,
track_id: u64,
quality: Quality,
) -> DelegatedApiResult<StreamUrl> {
let timestamp = get_timestamp();
let track_id_param = track_id.to_string();
let format_id = quality.id().to_string();
let signing_params = [
("format_id", format_id.as_str()),
("intent", "stream"),
("track_id", track_id_param.as_str()),
];
let signature = Zeroizing::new(delegated_signature(
"trackgetFileUrl",
&signing_params,
timestamp,
self.signing_secret.as_str(),
));
let timestamp = timestamp.to_string();
let url = self.endpoint.join(paths::TRACK_GET_FILE_URL)?;
let response = self
.execute(self.http.get(url).query(&[
("track_id", track_id_param.as_str()),
("format_id", format_id.as_str()),
("intent", "stream"),
("request_ts", timestamp.as_str()),
("request_sig", signature.as_str()),
]))
.await?;
ensure_success(response.status(), Some(track_id), true)?;
let value: Value = read_json(response).await?;
let restrictions: Vec<StreamRestriction> = value
.get("restrictions")
.cloned()
.map(serde_json::from_value)
.transpose()
.map_err(|_| DelegatedApiError::InvalidResponse)?
.unwrap_or_default();
let stream_url = value
.get("url")
.and_then(Value::as_str)
.filter(|url| !url.is_empty())
.ok_or(DelegatedApiError::TrackUnavailable(track_id))?
.to_string();
Ok(StreamUrl {
url: stream_url,
format_id: value
.get("format_id")
.and_then(Value::as_u64)
.unwrap_or_default() as u32,
mime_type: value
.get("mime_type")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string(),
sampling_rate: value
.get("sampling_rate")
.and_then(Value::as_f64)
.unwrap_or_default(),
bit_depth: value
.get("bit_depth")
.and_then(Value::as_u64)
.map(|depth| depth as u32),
track_id,
restrictions,
})
}
/// Try the requested quality and lower tiers without ever switching to the
/// owner session or refreshing the delegated credential.
pub async fn get_stream_url_with_fallback(
&self,
track_id: u64,
preferred: Quality,
) -> DelegatedApiResult<StreamUrl> {
let qualities = Quality::fallback_order();
let start = qualities
.iter()
.position(|quality| *quality == preferred)
.unwrap_or_default();
let mut unavailable = false;
for quality in &qualities[start..] {
match self.get_stream_url(track_id, *quality).await {
Ok(stream) if !stream.has_restrictions() => return Ok(stream),
Ok(_) => continue,
Err(DelegatedApiError::TrackUnavailable(_)) => unavailable = true,
Err(error) => return Err(error),
}
}
if unavailable {
Err(DelegatedApiError::TrackUnavailable(track_id))
} else {
Err(DelegatedApiError::NoQualityAvailable)
}
}
async fn get_tracks_batch_chunk(&self, track_ids: &[u64]) -> DelegatedApiResult<Vec<Track>> {
let ids = track_ids
.iter()
.map(u64::to_string)
.collect::<Vec<_>>()
.join(",");
let timestamp = get_timestamp();
let signature = Zeroizing::new(delegated_signature(
"trackgetList",
&[("tracks_id", ids.as_str())],
timestamp,
self.signing_secret.as_str(),
));
let timestamp = timestamp.to_string();
let url = self.endpoint.join(paths::TRACK_GET_LIST)?;
let response = self
.execute(
self.http
.post(url)
.query(&[
("request_ts", timestamp.as_str()),
("request_sig", signature.as_str()),
])
.json(&serde_json::json!({ "tracks_id": track_ids })),
)
.await?;
ensure_success(response.status(), None, true)?;
let value: Value = read_json(response).await?;
let items = value
.get("tracks")
.and_then(|tracks| tracks.get("items"))
.cloned()
.ok_or(DelegatedApiError::InvalidResponse)?;
serde_json::from_value(items).map_err(|_| DelegatedApiError::InvalidResponse)
}
async fn signed_get(
&self,
path: &str,
method_name: &str,
params: &[(&str, String)],
) -> DelegatedApiResult<Response> {
let timestamp = get_timestamp();
let signing_params = params
.iter()
.map(|(key, value)| (*key, value.as_str()))
.collect::<Vec<_>>();
let signature = Zeroizing::new(delegated_signature(
method_name,
&signing_params,
timestamp,
self.signing_secret.as_str(),
));
let timestamp = timestamp.to_string();
let mut query = signing_params;
query.push(("request_ts", timestamp.as_str()));
query.push(("request_sig", signature.as_str()));
let url = self.endpoint.join(path)?;
self.execute(self.http.get(url).query(&query)).await
}
async fn execute(&self, request: RequestBuilder) -> DelegatedApiResult<Response> {
self.guard()?;
request
.headers(self.delegated_headers()?)
.send()
.await
.map_err(|_| DelegatedApiError::Network)
}
fn guard(&self) -> DelegatedApiResult<()> {
if crate::offline_gate::is_offline() {
return Err(DelegatedApiError::OfflineMode);
}
if self.expires_at <= unix_timestamp() {
return Err(DelegatedApiError::Expired);
}
Ok(())
}
fn delegated_headers(&self) -> DelegatedApiResult<HeaderMap> {
let mut headers = HeaderMap::new();
headers.insert(
"X-App-Id",
HeaderValue::from_str(&self.app_id).map_err(|_| {
DelegatedApiError::Configuration(DelegatedApiConfigError::InvalidAppId)
})?,
);
let mut authorization =
bearer_header(self.jwt_api.as_str()).map_err(DelegatedApiError::Configuration)?;
authorization.set_sensitive(true);
headers.insert(AUTHORIZATION, authorization);
Ok(headers)
}
}
impl fmt::Debug for DelegatedQobuzClient {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("DelegatedQobuzClient")
.field("endpoint", &"<redacted>")
.field("expires_at", &self.expires_at)
.field("jwt_api", &"<redacted>")
.field("app_id", &"<redacted>")
.field("signing_secret", &"<redacted>")
.finish()
}
}
fn validate_jwt(jwt: &str) -> std::result::Result<(), DelegatedApiConfigError> {
if jwt.is_empty()
|| jwt.len() > MAX_JWT_BYTES
|| !jwt.bytes().all(|byte| (0x21..=0x7e).contains(&byte))
{
return Err(DelegatedApiConfigError::InvalidJwt);
}
Ok(())
}
fn bearer_header(jwt: &str) -> std::result::Result<HeaderValue, DelegatedApiConfigError> {
let mut bearer = Zeroizing::new(String::with_capacity("Bearer ".len() + jwt.len()));
bearer.push_str("Bearer ");
bearer.push_str(jwt);
HeaderValue::from_str(bearer.as_str()).map_err(|_| DelegatedApiConfigError::InvalidJwt)
}
fn unix_timestamp() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
/// Sign without ever allocating a concatenated buffer containing the secret.
fn delegated_signature(
method: &str,
params: &[(&str, &str)],
timestamp: u64,
secret: &str,
) -> String {
let mut sorted = params.to_vec();
sorted.sort_by_key(|(key, _)| *key);
let mut hasher = Md5::new();
hasher.update(method.as_bytes());
for (key, value) in sorted {
hasher.update(key.as_bytes());
hasher.update(value.as_bytes());
}
hasher.update(timestamp.to_string().as_bytes());
hasher.update(secret.as_bytes());
format!("{:x}", hasher.finalize())
}
fn ensure_success(
status: StatusCode,
track_id: Option<u64>,
signed: bool,
) -> DelegatedApiResult<()> {
match status {
status if status.is_success() => Ok(()),
StatusCode::BAD_REQUEST if signed => Err(DelegatedApiError::InvalidSignature),
StatusCode::UNAUTHORIZED => Err(DelegatedApiError::Unauthorized),
StatusCode::FORBIDDEN => Err(DelegatedApiError::Forbidden),
StatusCode::NOT_FOUND => track_id.map_or_else(
|| Err(DelegatedApiError::RequestRejected(status.as_u16())),
|track_id| Err(DelegatedApiError::TrackUnavailable(track_id)),
),
StatusCode::TOO_MANY_REQUESTS => Err(DelegatedApiError::RateLimited),
status if status.is_server_error() => Err(DelegatedApiError::ServerError(status.as_u16())),
status => Err(DelegatedApiError::RequestRejected(status.as_u16())),
}
}
async fn read_json<T: DeserializeOwned>(response: Response) -> DelegatedApiResult<T> {
let body = response
.bytes()
.await
.map_err(|_| DelegatedApiError::Network)?;
serde_json::from_slice(&body).map_err(|_| DelegatedApiError::InvalidResponse)
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use std::sync::Once;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
const APP_ID: &str = "delegated-app-id";
const SIGNING_SECRET: &str = "delegated-signing-secret";
const JWT: &str = "delegated.jwt.value";
fn install_tls_provider() {
static INSTALL: Once = Once::new();
INSTALL.call_once(|| {
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
});
}
fn future_expiry() -> u64 {
unix_timestamp() + 600
}
fn test_client(base_url: String) -> DelegatedQobuzClient {
install_tls_provider();
let endpoint = DelegatedApiEndpoint {
base_url: Zeroizing::new(base_url),
};
let credentials =
DelegatedAppCredentials::new(APP_ID.to_string(), SIGNING_SECRET.to_string()).unwrap();
let config =
DelegatedApiConfig::new(endpoint, future_expiry(), JWT.to_string(), credentials)
.unwrap();
DelegatedQobuzClient::new(config).unwrap()
}
async fn mock_once(body: &'static str) -> (String, tokio::task::JoinHandle<String>) {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let task = tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.unwrap();
let mut request = Vec::new();
let mut chunk = [0_u8; 1024];
loop {
let read = socket.read(&mut chunk).await.unwrap();
if read == 0 {
break;
}
request.extend_from_slice(&chunk[..read]);
if request.windows(4).any(|window| window == b"\r\n\r\n") {
break;
}
}
let response = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body
);
socket.write_all(response.as_bytes()).await.unwrap();
String::from_utf8(request).unwrap()
});
(format!("http://{address}/api.json/0.2/"), task)
}
fn request_headers(request: &str) -> BTreeMap<String, String> {
request
.lines()
.skip(1)
.take_while(|line| !line.is_empty())
.filter_map(|line| line.split_once(':'))
.map(|(name, value)| (name.to_ascii_lowercase(), value.trim().to_string()))
.collect()
}
#[test]
fn endpoint_requires_https_and_a_caller_allowlist() {
let allowed = ["api.example.test"];
assert!(matches!(
DelegatedApiEndpoint::new("http://api.example.test/v1", &allowed),
Err(DelegatedApiConfigError::InsecureEndpoint)
));
assert!(matches!(
DelegatedApiEndpoint::new("https://other.example.test/v1", &allowed),
Err(DelegatedApiConfigError::EndpointHostNotAllowed)
));
assert!(matches!(
DelegatedApiEndpoint::new("https://user@api.example.test/v1", &allowed),
Err(DelegatedApiConfigError::InvalidEndpoint)
));
assert!(matches!(
DelegatedApiEndpoint::new("https://api.example.test/v1?token=nope", &allowed),
Err(DelegatedApiConfigError::InvalidEndpoint)
));
let endpoint = DelegatedApiEndpoint::new("https://api.example.test/v1", &allowed).unwrap();
assert_eq!(
endpoint.join("track/get").unwrap().as_str(),
"https://api.example.test/v1/track/get"
);
}
#[test]
fn config_rejects_expired_short_lived_and_invalid_credentials() {
let make_config = |expires_at, jwt: &str| {
let endpoint =
DelegatedApiEndpoint::new("https://api.example.test/v1", &["api.example.test"])
.unwrap();
let credentials =
DelegatedAppCredentials::new(APP_ID.to_string(), SIGNING_SECRET.to_string())
.unwrap();
DelegatedApiConfig::new(endpoint, expires_at, jwt.to_string(), credentials)
};
let now = unix_timestamp();
assert!(matches!(
make_config(now, JWT),
Err(DelegatedApiConfigError::Expired)
));
assert!(matches!(
make_config(now + 59, JWT),
Err(DelegatedApiConfigError::InsufficientLifetime)
));
assert!(matches!(
make_config(now + 600, "delegated jwt with spaces"),
Err(DelegatedApiConfigError::InvalidJwt)
));
}
#[test]
fn secrets_and_authorization_never_appear_in_debug() {
install_tls_provider();
let endpoint_text = "https://api.example.test/private/base";
let endpoint = DelegatedApiEndpoint::new(endpoint_text, &["api.example.test"]).unwrap();
let credentials =
DelegatedAppCredentials::new(APP_ID.to_string(), SIGNING_SECRET.to_string()).unwrap();
let config =
DelegatedApiConfig::new(endpoint, future_expiry(), JWT.to_string(), credentials)
.unwrap();
let config_debug = format!("{config:?}");
for forbidden in [
endpoint_text,
APP_ID,
SIGNING_SECRET,
JWT,
"Authorization",
"Bearer",
] {
assert!(!config_debug.contains(forbidden));
}
let client = DelegatedQobuzClient::new(config).unwrap();
let client_debug = format!("{client:?}");
for forbidden in [
endpoint_text,
APP_ID,
SIGNING_SECRET,
JWT,
"Authorization",
"Bearer",
] {
assert!(!client_debug.contains(forbidden));
}
let headers_debug = format!("{:?}", client.delegated_headers().unwrap());
assert!(headers_debug.contains("Sensitive"));
assert!(!headers_debug.contains(JWT));
assert!(!headers_debug.contains("Bearer"));
assert!(!headers_debug.contains("x-user-auth-token"));
}
#[tokio::test]
async fn access_preflight_uses_only_bearer_and_app_id() {
let _lock = crate::offline_gate::test_lock()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let _reset = crate::offline_gate::TestGateReset;
crate::offline_gate::set_offline(false);
let (base_url, server) = mock_once(r#"{"id": 7}"#).await;
let client = test_client(base_url);
client.validate_access().await.unwrap();
let request = server.await.unwrap();
let request_line = request.lines().next().unwrap();
let headers = request_headers(&request);
assert_eq!(request_line, "GET /api.json/0.2/user/get HTTP/1.1");
assert_eq!(headers.get("x-app-id").map(String::as_str), Some(APP_ID));
assert_eq!(
headers.get("authorization").map(String::as_str),
Some("Bearer delegated.jwt.value")
);
assert!(!headers.contains_key("x-user-auth-token"));
}
#[tokio::test]
async fn metadata_request_keeps_delegated_headers_and_is_signed() {
let _lock = crate::offline_gate::test_lock()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let _reset = crate::offline_gate::TestGateReset;
crate::offline_gate::set_offline(false);
let (base_url, server) = mock_once(r#"{"id": 42}"#).await;
let client = test_client(base_url);
let track = client.get_track(42).await.unwrap();
assert_eq!(track.id, 42);
let request = server.await.unwrap();
let request_line = request.lines().next().unwrap();
let headers = request_headers(&request);
assert_eq!(headers.get("x-app-id").map(String::as_str), Some(APP_ID));
assert_eq!(
headers.get("authorization").map(String::as_str),
Some("Bearer delegated.jwt.value")
);
assert!(!headers.contains_key("x-user-auth-token"));
let target = request_line.split_whitespace().nth(1).unwrap();
let url = Url::parse(&format!("http://localhost{target}")).unwrap();
let query = url.query_pairs().into_owned().collect::<BTreeMap<_, _>>();
assert_eq!(url.path(), "/api.json/0.2/track/get");
assert_eq!(query.get("track_id").map(String::as_str), Some("42"));
let timestamp = query.get("request_ts").unwrap().parse::<u64>().unwrap();
let expected =
crate::auth::sign_request("trackget", &[("track_id", "42")], timestamp, SIGNING_SECRET);
assert_eq!(query.get("request_sig"), Some(&expected));
}
#[test]
fn expired_credentials_fail_before_any_request() {
let _lock = crate::offline_gate::test_lock()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let _reset = crate::offline_gate::TestGateReset;
crate::offline_gate::set_offline(false);
let mut client = test_client("http://127.0.0.1:9/api.json/0.2/".to_string());
client.expires_at = 0;
assert_eq!(client.guard(), Err(DelegatedApiError::Expired));
}
}
+1
View File
@@ -5,6 +5,7 @@ pub const BASE_URL: &str = "https://www.qobuz.com/api.json/0.2";
/// Endpoint paths
pub mod paths {
// User
pub const USER_GET: &str = "/user/get";
pub const USER_LOGIN: &str = "/user/login";
// OAuth
+5
View File
@@ -34,6 +34,7 @@ pub mod auth;
pub mod bundle;
pub mod client;
pub mod cmaf;
pub mod delegated;
pub mod endpoints;
pub mod error;
pub mod forbidden_breaker;
@@ -56,6 +57,10 @@ pub use cmaf::{
setup_streaming as cmaf_setup_streaming, CmafProgressCallback, CmafProgressUpdate,
CmafRawBundle, CmafStreamingInfo, CMAF_PREFETCH_CONCURRENCY,
};
pub use delegated::{
DelegatedApiConfig, DelegatedApiConfigError, DelegatedApiEndpoint, DelegatedApiError,
DelegatedApiResult, DelegatedAppCredentials, DelegatedQobuzClient,
};
pub use error::{ApiError, Result};
pub use link_resolver::{resolve_link, LinkResolverError, ResolvedLink};
pub use lyrics::{
+2
View File
@@ -7,6 +7,7 @@ description = "UI-agnostic Qobuz Connect application adapter"
[dependencies]
qconnect-core = { path = "../qconnect-core" }
qconnect-lan = { path = "../qconnect-lan" }
qconnect-protocol = { path = "../qconnect-protocol" }
qconnect-transport-ws = { path = "../qconnect-transport-ws" }
qbz-models = { path = "../qbz-models" }
@@ -16,6 +17,7 @@ qbz-models = { path = "../qbz-models" }
# reached ONLY through the trait's start_track_stream and are never modified.
qbz-player = { path = "../qbz-player" }
async-trait = { workspace = true }
futures-util = { workspace = true }
tokio = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
+14 -27
View File
@@ -1730,22 +1730,6 @@ pub trait SessionLoopHost: Send + Sync {
async fn on_loop_error(&self, message: String);
}
/// Hex preview of up to `max_bytes` bytes for diagnostic logging. Mirrors the
/// Tauri adapter's `hex_preview` (relocated with the loop, slice 5).
fn hex_preview(data: &[u8], max_bytes: usize) -> String {
let take = data.len().min(max_bytes);
let hex: String = data[..take]
.iter()
.map(|b| format!("{b:02x}"))
.collect::<Vec<_>>()
.join("");
if data.len() > max_bytes {
format!("{hex}...({}B total)", data.len())
} else {
hex
}
}
impl<TTransport, TSink> QconnectApp<TTransport, TSink>
where
TTransport: WsTransport + 'static,
@@ -1866,9 +1850,10 @@ where
}
TransportEvent::InboundQueueServerEvent(evt) => {
log::debug!(
"[QConnect] <-- Inbound queue event: {} payload={}",
"[QConnect] <-- Inbound queue event: {} tracks={} autoplay_tracks={}",
evt.message_type(),
evt.payload
evt.payload.get("tracks").and_then(|value| value.as_array()).map_or(0, Vec::len),
evt.payload.get("autoplay_tracks").and_then(|value| value.as_array()).map_or(0, Vec::len),
);
self.sink
.on_event(QconnectAppEvent::Diagnostic {
@@ -1888,9 +1873,8 @@ where
}
TransportEvent::InboundRendererServerCommand(cmd) => {
log::debug!(
"[QConnect] <-- Inbound renderer command: {} payload={}",
cmd.message_type(),
cmd.payload
"[QConnect] <-- Inbound renderer command: {}",
cmd.message_type()
);
}
TransportEvent::InboundFrameDecoded {
@@ -1907,7 +1891,11 @@ where
cloud_message_type,
payload,
} => {
log::debug!("[QConnect/Transport] <-- Payload bytes: cloud_type={} len={} hex={}", cloud_message_type, payload.len(), hex_preview(payload, 64));
log::debug!(
"[QConnect/Transport] <-- Payload bytes: cloud_type={} len={}",
cloud_message_type,
payload.len()
);
}
TransportEvent::OutboundSent {
message_type,
@@ -2157,11 +2145,10 @@ mod tests {
}
fn test_config() -> WsTransportConfig {
WsTransportConfig {
endpoint_url: "wss://example.invalid/ws".to_string(),
subscribe_channels: vec![vec![1, 2, 3]],
..Default::default()
}
let mut config = WsTransportConfig::default();
config.endpoint_url = "wss://example.invalid/ws".to_string();
config.subscribe_channels = vec![vec![1, 2, 3]];
config
}
async fn build_connected_app() -> (
File diff suppressed because it is too large. Load diff
+6
View File
@@ -3,6 +3,7 @@
//! Application adapter that composes qconnect core + protocol + transport.
mod app;
mod delegation;
mod error;
mod events;
mod feature_flags;
@@ -20,6 +21,11 @@ pub use app::{
SessionStateTakeoverInput,
};
pub use startup::{compute_effective_startup, QconnectStartupMode};
pub use delegation::{
CommitRejected, CredentialOrigin, DelegationCancellation, DelegationCandidate,
DelegationCoordinator, DelegationCoordinatorConfig, DelegationCoordinatorError,
DelegationErrorCode, DelegationHost, DelegationPhase, DelegationSnapshot, RestoreReason,
};
pub use error::QconnectAppError;
pub use events::{NoOpEventSink, QconnectAppEvent, QconnectEventSink};
pub use session::{
+7 -2
View File
@@ -9,14 +9,19 @@ description = "Official Qobuz Connect LAN receiver wire and discovery"
serde = { workspace = true }
serde_json = { workspace = true }
thiserror = { workspace = true }
tiny_http = "0.12"
mdns-sd = "0.21"
if-addrs = "0.15"
socket2 = "0.6"
url = "2"
zeroize = "1"
[dev-dependencies]
base64 = { workspace = true }
qbz-credentials = { path = "../qbz-credentials" }
qbz-qobuz = { path = "../qbz-qobuz" }
qconnect-core = { path = "../qconnect-core" }
qconnect-protocol = { path = "../qconnect-protocol" }
qconnect-transport-ws = { path = "../qconnect-transport-ws" }
reqwest = { workspace = true }
rustls = { version = "0.23", features = ["aws_lc_rs"] }
tokio = { workspace = true }
uuid = { version = "1", features = ["v4"] }
@@ -0,0 +1,354 @@
//! Development-only official-client handoff probe.
//!
//! This binary advertises one ephemeral LAN renderer, accepts exactly the
//! official handoff shape, and validates the two delegated contexts without
//! printing token, endpoint, account or session values. It never starts the
//! player and never persists the handoff.
use std::collections::BTreeSet;
use std::error::Error;
use std::sync::mpsc;
use std::sync::Arc;
use std::time::Duration;
use base64::Engine;
use qbz_qobuz::{endpoints, QobuzClient};
use qconnect_core::QueueVersion;
use qconnect_lan::{
ConnectInfo, DeviceType, DisplayInfo, EndpointPolicy, HandoffCandidate, LanProjection,
LanService, LanServiceConfig, MaxAudioQuality,
};
use qconnect_protocol::{
build_qconnect_renderer_outbound_envelope, RendererBufferState, RendererCommandType,
RendererReport, RendererReportType,
};
use qconnect_transport_ws::{NativeWsTransport, TransportEvent, WsTransport, WsTransportConfig};
use reqwest::redirect::Policy;
use serde_json::{json, Value};
use uuid::Uuid;
use zeroize::Zeroize;
const WAIT_FOR_HANDOFF: Duration = Duration::from_secs(10 * 60);
const CLOUD_VALIDATION_TIMEOUT: Duration = Duration::from_secs(30);
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
let client = QobuzClient::new()?;
client.init().await?;
login_from_saved_oauth(&client).await?;
let app_id = client.app_id().await?;
let (qws_endpoint, _expires_at, mut owner_qws_jwt) =
client.create_qconnect_token().await?.into_parts();
owner_qws_jwt.zeroize();
let qws_endpoint = qws_endpoint.ok_or("owner QWS token omitted its endpoint")?;
let endpoint_policy =
EndpointPolicy::from_trusted_endpoints(endpoints::BASE_URL, &qws_endpoint)
.map_err(|_| "owner token supplied an invalid trusted endpoint")?;
let device_uuid = Uuid::new_v4().to_string();
let projection = LanProjection::new(
DisplayInfo {
friendly_name: "QBZ protocol probe".to_string(),
serial_number: device_uuid.clone(),
brand_display_name: "QBZ".to_string(),
model_display_name: "Development harness".to_string(),
max_audio_quality: MaxAudioQuality::UpToHires192,
device_type: DeviceType::Computer,
software_version: env!("CARGO_PKG_VERSION").to_string(),
},
ConnectInfo {
app_id: app_id.clone(),
current_session_id: None,
},
);
let (observations_tx, observations_rx) = mpsc::sync_channel(32);
let mut config = LanServiceConfig::new(projection, endpoint_policy, device_uuid.clone());
config.request_observer = Some(observations_tx);
let (mut lan, inbox) = LanService::start(config)?;
eprintln!("[probe] advertised 'QBZ protocol probe' for ten minutes");
eprintln!("[probe] select it from an official Qobuz client on this LAN");
let observation_task = tokio::task::spawn_blocking(move || {
while let Ok(observation) = observations_rx.recv() {
eprintln!("[probe] LAN {observation:?}");
}
});
let wait = tokio::task::spawn_blocking(move || inbox.take_timeout(WAIT_FOR_HANDOFF));
let candidate = tokio::select! {
result = wait => result.map_err(|_| "handoff waiter failed")?,
_ = tokio::signal::ctrl_c() => {
lan.shutdown();
return Ok(());
}
}
.ok_or("no official handoff arrived before the deadline")?;
lan.shutdown();
let _ = observation_task.await;
eprintln!("[probe] exact official handoff admitted; no secret was logged");
print_jwt_shape("jwt_api", candidate.api_token().jwt());
print_jwt_shape("jwt_qconnect", candidate.qconnect_token().jwt());
let api = probe_api_auth(&candidate, &app_id).await?;
eprintln!(
"[probe] delegated API user/get: baseline={}, x-user-auth-token={}, bearer={}",
api.baseline.label(),
api.x_user_auth_token.label(),
api.bearer.label()
);
validate_qws_session(&candidate, &device_uuid).await?;
eprintln!("[probe] delegated QWS accepted the controller-supplied session UUID");
Ok(())
}
async fn login_from_saved_oauth(client: &QobuzClient) -> Result<(), Box<dyn Error>> {
// Match the application's no-ambient-runtime credential derivation.
let (keyring, file) = std::thread::spawn(|| {
(
qbz_credentials::load_oauth_token().ok().flatten(),
qbz_credentials::load_oauth_token_from_file().ok().flatten(),
)
})
.join()
.map_err(|_| "credential loader panicked")?;
let mut candidates = Vec::new();
if let Some(token) = keyring {
candidates.push(token);
}
if let Some(token) = file {
if !candidates.iter().any(|existing| existing == &token) {
candidates.push(token);
}
}
for mut token in candidates {
let result = client.login_with_token(&token).await;
token.zeroize();
if result.is_ok() {
return Ok(());
}
}
Err("no saved OAuth token could initialize the probe".into())
}
fn print_jwt_shape(label: &str, jwt: &str) {
let segments = jwt.split('.').collect::<Vec<_>>();
if segments.len() != 3 {
eprintln!(
"[probe] {label}: opaque token with {} segment(s)",
segments.len()
);
return;
}
let header = json_object_keys(segments[0]);
let claims = json_object_keys(segments[1]);
eprintln!("[probe] {label}: JWT header_keys={header:?} claim_keys={claims:?}");
}
fn json_object_keys(segment: &str) -> BTreeSet<String> {
base64::engine::general_purpose::URL_SAFE_NO_PAD
.decode(segment)
.ok()
.and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
.and_then(|value| value.as_object().cloned())
.map(|object| object.into_iter().map(|(key, _)| key).collect())
.unwrap_or_default()
}
#[derive(Clone, Copy)]
struct ProbeResponse {
status: u16,
user_shape: bool,
}
impl ProbeResponse {
fn label(self) -> String {
format!(
"{}:{}",
self.status,
if self.user_shape { "user" } else { "other" }
)
}
}
struct ApiProbe {
baseline: ProbeResponse,
x_user_auth_token: ProbeResponse,
bearer: ProbeResponse,
}
async fn probe_api_auth(
candidate: &HandoffCandidate,
app_id: &str,
) -> Result<ApiProbe, Box<dyn Error>> {
let mut endpoint = url::Url::parse(candidate.api_token().endpoint())?;
if !endpoint.path().ends_with('/') {
endpoint.set_path(&format!("{}/", endpoint.path()));
}
let url = endpoint.join("user/get")?;
let http = reqwest::Client::builder()
.redirect(Policy::none())
.connect_timeout(Duration::from_secs(5))
.timeout(Duration::from_secs(10))
.build()?;
let baseline = inspect_api_response(
http.get(url.clone())
.header("X-App-Id", app_id)
.send()
.await?,
)
.await;
let x_user_auth_token = inspect_api_response(
http.get(url.clone())
.header("X-App-Id", app_id)
.header("X-User-Auth-Token", candidate.api_token().jwt())
.send()
.await?,
)
.await;
let bearer = inspect_api_response(
http.get(url)
.header("X-App-Id", app_id)
.bearer_auth(candidate.api_token().jwt())
.send()
.await?,
)
.await;
Ok(ApiProbe {
baseline,
x_user_auth_token,
bearer,
})
}
async fn inspect_api_response(response: reqwest::Response) -> ProbeResponse {
let status = response.status().as_u16();
let user_shape = response
.json::<Value>()
.await
.ok()
.and_then(|value| value.as_object().cloned())
.is_some_and(|object| {
object.contains_key("id")
|| object.contains_key("user")
|| object.contains_key("credential")
});
ProbeResponse { status, user_shape }
}
async fn validate_qws_session(
candidate: &HandoffCandidate,
device_uuid: &str,
) -> Result<(), Box<dyn Error>> {
let transport = Arc::new(NativeWsTransport::new());
let mut events = transport.subscribe();
let mut config = WsTransportConfig::default();
config.endpoint_url = candidate.qconnect_token().endpoint().to_string();
config.jwt_qws = Some(candidate.qconnect_token().jwt().to_string());
config.require_jwt = true;
config.subscribe_channels = vec![vec![0x01], vec![0x02], vec![0x03]];
config.reconnect_max_attempts = Some(1);
transport.connect(config).await?;
let validation = tokio::time::timeout(CLOUD_VALIDATION_TIMEOUT, async {
let mut joined = false;
loop {
match events.recv().await {
Ok(TransportEvent::Subscribed) if !joined => {
joined = true;
let report = RendererReport::new(
RendererReportType::RndrSrvrJoinSession,
Uuid::new_v4().to_string(),
QueueVersion::default(),
json!({
"session_uuid": candidate.session_id(),
"device_info": {
"device_uuid": device_uuid,
"friendly_name": "QBZ protocol probe",
"brand": "QBZ",
"model": "Development harness",
"device_type": 5,
"capabilities": {
"min_audio_quality": 1,
"max_audio_quality": 4,
"volume_remote_control": 2
},
"software_version": env!("CARGO_PKG_VERSION")
},
"is_active": candidate.become_active(),
"reason": 1,
"initial_state": {
"playing_state": 1,
"buffer_state": RendererBufferState::Ok.as_i32(),
"current_position": 0,
"duration": 0,
"queue_version": { "major": 0, "minor": 0 }
}
}),
);
let envelope = build_qconnect_renderer_outbound_envelope(report)?;
transport.send(envelope).await?;
}
Ok(TransportEvent::Connected) => eprintln!("[probe] QWS connected"),
Ok(TransportEvent::Authenticated) => eprintln!("[probe] QWS authenticated"),
Ok(TransportEvent::Subscribed) if joined => {
eprintln!("[probe] QWS subscribed again after delegated join")
}
Ok(TransportEvent::SessionEstablished) => {
eprintln!("[probe] QWS controller session established")
}
Ok(TransportEvent::CloudError { code, .. }) => {
return Err(format!("QWS rejected delegated join with code {code}").into());
}
Ok(TransportEvent::OutboundSent { message_type, .. }) => {
eprintln!("[probe] QWS outbound {message_type}");
}
Ok(TransportEvent::InboundFrameDecoded {
cloud_message_type,
payload_size,
}) => {
eprintln!(
"[probe] QWS inbound cloud_type={cloud_message_type} bytes={payload_size}"
);
}
Ok(TransportEvent::InboundQueueServerEvent(event)) => {
eprintln!("[probe] QWS inbound {}", event.message_type());
}
Ok(TransportEvent::InboundRendererServerCommand(command)) => {
eprintln!("[probe] QWS inbound {}", command.message_type());
if command.command_type == RendererCommandType::SrvrRndrSetActive
&& command.payload.get("active").and_then(Value::as_bool) == Some(true)
{
return Ok::<(), Box<dyn Error>>(());
}
}
Ok(TransportEvent::InboundReceived(_)) => {
eprintln!("[probe] QWS inbound JSON envelope");
}
Ok(TransportEvent::TransportError { stage, .. }) => {
eprintln!("[probe] QWS transport error at stage={stage}");
}
Ok(TransportEvent::ReconnectScheduled { attempt, .. }) => {
eprintln!("[probe] QWS reconnect scheduled attempt={attempt}");
}
Ok(TransportEvent::MaxReconnectAttemptsExceeded { attempts, .. }) => {
return Err(
format!("QWS validation exhausted after {attempts} attempt(s)").into(),
);
}
Ok(TransportEvent::Disconnected) => eprintln!("[probe] QWS disconnected"),
Ok(_) => {}
Err(_) => return Err("QWS validation event channel closed".into()),
}
}
})
.await;
let _ = transport.disconnect().await;
validation.map_err(|_| "QWS delegated session validation timed out")?
}
+4 -1
View File
@@ -16,5 +16,8 @@ pub use model::{
ConnectInfo, DeviceType, DisplayInfo, HandoffCandidate, LanJwtToken, MaxAudioQuality,
};
pub use projection::LanProjection;
pub use server::{LanError, LanService, LanServiceConfig, SERVICE_TYPE};
pub use server::{
LanError, LanHttpMethod, LanHttpRoute, LanRequestObservation, LanService, LanServiceConfig,
SERVICE_TYPE,
};
pub use validation::{EndpointPolicy, ValidationError, MAX_BODY_BYTES};
+83 -19
View File
@@ -1,11 +1,16 @@
use std::collections::HashMap;
use std::net::IpAddr;
use std::time::Duration;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::time::{Duration, Instant};
use mdns_sd::{ServiceDaemon, ServiceInfo};
use mdns_sd::{DaemonEvent, IfKind, ServiceDaemon, ServiceInfo, UnregisterStatus};
use crate::server::{LanError, SERVICE_TYPE};
const ANNOUNCE_TIMEOUT: Duration = Duration::from_secs(2);
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2);
const GOODBYE_RESEND_DELAY: Duration = Duration::from_millis(150);
const DAEMON_SHUTDOWN_RESERVE: Duration = Duration::from_millis(250);
pub(crate) struct MdnsRegistration {
daemon: ServiceDaemon,
fullname: String,
@@ -32,17 +37,60 @@ impl MdnsRegistration {
let info = service_info(device_uuid, port, sdk_version, &addresses)?;
let fullname = info.get_fullname().to_string();
let daemon = ServiceDaemon::new()?;
daemon.register(info)?;
let monitor = daemon.monitor()?;
if daemon.register(info).is_err() {
shutdown_daemon(&daemon, Some(&fullname), SHUTDOWN_TIMEOUT);
return Err(LanError::Mdns);
}
let deadline = Instant::now() + ANNOUNCE_TIMEOUT;
let announced = loop {
let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
break false;
};
match monitor.recv_timeout(remaining) {
Ok(DaemonEvent::Announce(announced, _))
if announced.eq_ignore_ascii_case(&fullname) =>
{
break true;
}
Ok(DaemonEvent::Error(_) | DaemonEvent::NameChange(_)) | Err(_) => break false,
Ok(_) => {}
}
};
if !announced {
shutdown_daemon(&daemon, Some(&fullname), SHUTDOWN_TIMEOUT);
return Err(LanError::Mdns);
}
Ok(Self { daemon, fullname })
}
pub(crate) fn shutdown(self) {
if let Ok(receiver) = self.daemon.unregister(&self.fullname) {
let _ = receiver.recv_timeout(Duration::from_secs(2));
shutdown_daemon(&self.daemon, Some(&self.fullname), SHUTDOWN_TIMEOUT);
}
}
fn shutdown_daemon(daemon: &ServiceDaemon, fullname: Option<&str>, timeout: Duration) {
let deadline = Instant::now() + timeout;
if let Some(fullname) = fullname {
if let Ok(receiver) = daemon.unregister(fullname) {
if let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
if matches!(receiver.recv_timeout(remaining), Ok(UnregisterStatus::OK)) {
// mdns-sd schedules a second goodbye 120 ms after the
// unregister acknowledgement. Keep the daemon alive long
// enough to send it, while reserving time for shutdown.
if let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
let goodbye_budget = remaining.saturating_sub(DAEMON_SHUTDOWN_RESERVE);
std::thread::sleep(GOODBYE_RESEND_DELAY.min(goodbye_budget));
}
}
}
}
if let Ok(receiver) = self.daemon.shutdown() {
let _ = receiver.recv_timeout(Duration::from_secs(2));
}
if let Ok(receiver) = daemon.shutdown() {
if let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
let _ = receiver.recv_timeout(remaining);
}
}
}
@@ -61,27 +109,36 @@ fn service_info(
("sdk_version".to_string(), sdk_version.to_string()),
("path".to_string(), String::new()),
]);
ServiceInfo::new(
let mut info = ServiceInfo::new(
SERVICE_TYPE,
&instance,
&hostname,
addresses,
port,
properties,
)
.map(ServiceInfo::enable_addr_auto)
)?;
info.set_interfaces(addresses.iter().copied().map(IfKind::Addr).collect());
Ok(info)
}
fn lan_ipv4_addresses() -> Result<Vec<IpAddr>, LanError> {
let mut addresses = if_addrs::get_if_addrs()
// A connected UDP socket performs route selection without sending data.
// Targeting the mDNS multicast group yields the IPv4 interface that can
// actually reach LAN controllers, excluding Docker/VM bridges and
// loopback records that make strict official clients choose a dead URL.
let socket = UdpSocket::bind(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)))
.map_err(|_| LanError::AddressDiscovery)?;
socket
.connect(SocketAddr::from((Ipv4Addr::new(224, 0, 0, 251), 5353)))
.map_err(|_| LanError::AddressDiscovery)?;
let address = socket
.local_addr()
.map_err(|_| LanError::AddressDiscovery)?
.into_iter()
.map(|interface| interface.ip())
.filter(|ip| ip.is_ipv4() && !ip.is_loopback() && !ip.is_unspecified())
.collect::<Vec<_>>();
addresses.sort_unstable();
addresses.dedup();
Ok(addresses)
.ip();
if !address.is_ipv4() || address.is_loopback() || address.is_unspecified() {
return Err(LanError::NoLanAddresses);
}
Ok(vec![address])
}
fn short_id(device_uuid: &str) -> String {
@@ -118,5 +175,12 @@ mod tests {
assert_eq!(info.get_property_val_str("device_uuid"), Some(uuid));
assert_eq!(info.get_property_val_str("sdk_version"), Some("0.9.5"));
assert_eq!(info.get_property_val_str("path"), Some(""));
assert!(!info.is_addr_auto());
assert_eq!(
info.get_addresses(),
&["192.0.2.10".parse::<IpAddr>().unwrap()]
.into_iter()
.collect()
);
}
}
+618 -155
View File
@@ -1,6 +1,6 @@
use std::collections::HashMap;
use std::io::Read;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener};
use std::io::{self, Read, Write};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::mpsc::{self, TrySendError};
use std::sync::{Arc, Mutex};
@@ -8,7 +8,7 @@ use std::thread::JoinHandle;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use socket2::{Domain, Protocol, Socket, Type};
use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
use zeroize::Zeroizing;
use crate::admission::{admission_channel, AdmissionInbox, AdmissionSender, SubmitError};
use crate::mdns::MdnsRegistration;
@@ -22,8 +22,36 @@ const POST_RATE_PER_MINUTE: f64 = 6.0;
const POST_BURST: f64 = 2.0;
const MAX_RATE_ORIGINS: usize = 256;
const HTTP_WORKERS: usize = 2;
const MAX_HEADER_BYTES: usize = 16 * 1024;
const ACCEPT_WAKE_TIMEOUT: Duration = Duration::from_millis(250);
const REJECTION_DRAIN_TIMEOUT: Duration = Duration::from_millis(25);
pub const DEFAULT_LOCAL_READ_TIMEOUT: Duration = Duration::from_secs(5);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LanHttpMethod {
Get,
Post,
Other,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LanHttpRoute {
GetDisplayInfo,
GetConnectInfo,
ConnectToQconnect,
Unknown,
}
/// Sanitized, opt-in request observation for protocol probes and diagnostics.
/// It deliberately excludes remote addresses, headers, URLs and bodies.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LanRequestObservation {
pub method: LanHttpMethod,
pub route: LanHttpRoute,
pub status_code: u16,
pub validation_error: Option<ValidationError>,
}
#[derive(Debug, thiserror::Error)]
pub enum LanError {
#[error("qconnect-lan-bind")]
@@ -51,15 +79,18 @@ pub struct LanServiceConfig {
pub endpoint_policy: EndpointPolicy,
pub device_uuid: String,
pub sdk_version: String,
/// Bounds stalled local request reads. The production default is the
/// contract's five-second local read/parse budget.
/// Bounds the complete local read and parse operation, including clients
/// that trickle bytes without ever going idle.
pub read_timeout: Duration,
/// Tests and environments without multicast can exercise the exact HTTP
/// wire while keeping discovery disabled. Production adapters leave true.
pub advertise_mdns: bool,
/// Optional explicit address set, primarily for deterministic integration
/// tests. Production uses interface discovery and mdns-sd auto updates.
/// tests. Production selects the interface routed to mDNS multicast.
pub advertised_addresses: Option<Vec<IpAddr>>,
/// Optional bounded diagnostics sink. Events are best-effort and dropped
/// when the receiver is busy or disconnected.
pub request_observer: Option<mpsc::SyncSender<LanRequestObservation>>,
}
impl LanServiceConfig {
@@ -77,40 +108,100 @@ impl LanServiceConfig {
read_timeout: DEFAULT_LOCAL_READ_TIMEOUT,
advertise_mdns: true,
advertised_addresses: None,
request_observer: None,
}
}
}
pub struct LanService {
port: u16,
server: Arc<Server>,
admission: AdmissionSender,
accepting: Arc<AtomicBool>,
accept_wake_addr: SocketAddr,
active_connections: Arc<ActiveConnections>,
accept_thread: Option<JoinHandle<()>>,
worker_threads: Vec<JoinHandle<()>>,
mdns: Option<MdnsRegistration>,
}
struct InFlightPermit {
count: Arc<AtomicUsize>,
}
impl InFlightPermit {
fn try_acquire(count: Arc<AtomicUsize>) -> Option<Self> {
count
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
(current < HTTP_WORKERS).then_some(current + 1)
})
.ok()
.map(|_| Self { count })
}
}
impl Drop for InFlightPermit {
fn drop(&mut self) {
self.count.fetch_sub(1, Ordering::AcqRel);
}
}
#[derive(Default)]
struct ActiveConnections {
next_id: AtomicUsize,
// Cloned handles exist only so shutdown can interrupt reads in workers.
// The worker-owned stream remains the sole reader and writer.
streams: Mutex<HashMap<usize, TcpStream>>,
}
impl ActiveConnections {
fn insert(&self, stream: TcpStream) -> usize {
let id = self.next_id.fetch_add(1, Ordering::Relaxed);
recover_lock(&self.streams).insert(id, stream);
id
}
fn remove(&self, id: usize) {
recover_lock(&self.streams).remove(&id);
}
fn shutdown_all(&self) {
for stream in recover_lock(&self.streams).values() {
let _ = stream.shutdown(Shutdown::Both);
}
}
}
struct ConnectionJob {
stream: TcpStream,
active_connections: Arc<ActiveConnections>,
active_id: usize,
_permit: InFlightPermit,
}
impl Drop for ConnectionJob {
fn drop(&mut self) {
self.active_connections.remove(self.active_id);
}
}
impl LanService {
/// Bind first, spawn HTTP workers, then publish DNS-SD. On any discovery
/// failure the bound listener is torn down before the error returns.
/// Bind first, spawn the fixed HTTP worker set, then publish DNS-SD. On
/// any discovery failure the listener is torn down before returning.
pub fn start(config: LanServiceConfig) -> Result<(Self, AdmissionInbox), LanError> {
validate_config(&config)?;
let server = Arc::new(bind_http(config.bind_addr, config.read_timeout)?);
let port = server
.server_addr()
.to_ip()
.map(|address| address.port())
.ok_or(LanError::Bind)?;
let listener = bind_http(config.bind_addr)?;
let listener_addr = listener.local_addr().map_err(|_| LanError::Bind)?;
let port = listener_addr.port();
let accept_wake_addr = accept_wake_addr(listener_addr);
let (admission, inbox) = admission_channel();
let accepting = Arc::new(AtomicBool::new(true));
let rate_limiter = Arc::new(Mutex::new(PostRateLimiter::default()));
// Count queued plus active work explicitly. This avoids a scheduling
// race where a worker exists but has not reached `recv` yet.
let (request_tx, request_rx) = mpsc::sync_channel::<Request>(HTTP_WORKERS);
let (request_tx, request_rx) = mpsc::sync_channel::<ConnectionJob>(HTTP_WORKERS);
let request_rx = Arc::new(Mutex::new(request_rx));
let in_flight = Arc::new(AtomicUsize::new(0));
let active_connections = Arc::new(ActiveConnections::default());
let mut worker_threads = Vec::with_capacity(HTTP_WORKERS);
for _ in 0..HTTP_WORKERS {
let worker_rx = Arc::clone(&request_rx);
@@ -119,83 +210,112 @@ impl LanService {
let worker_projection = config.projection.clone();
let worker_policy = config.endpoint_policy.clone();
let worker_limiter = Arc::clone(&rate_limiter);
let worker_in_flight = Arc::clone(&in_flight);
let worker_observer = config.request_observer.clone();
let read_timeout = config.read_timeout;
worker_threads.push(std::thread::spawn(move || loop {
let request = worker_rx
.lock()
.expect("LAN request queue lock poisoned")
.recv();
let request = match request {
Ok(request) => request,
let job = {
let receiver = recover_lock(&worker_rx);
receiver.recv()
};
let mut job = match job {
Ok(job) => job,
Err(_) => break,
};
if worker_accepting.load(Ordering::Acquire) {
handle_request(
request,
handle_connection(
&mut job.stream,
read_timeout,
&worker_projection,
&worker_policy,
&worker_admission,
&worker_limiter,
worker_observer.as_ref(),
);
} else {
let _ = request.respond(error_response(StatusCode(503), "unavailable"));
let _ = write_response(&mut job.stream, &error_response(503, "unavailable"));
}
worker_in_flight.fetch_sub(1, Ordering::AcqRel);
}));
}
let thread_server = Arc::clone(&server);
let thread_accepting = Arc::clone(&accepting);
let accept_in_flight = Arc::clone(&in_flight);
let accept_thread = std::thread::spawn(move || {
while thread_accepting.load(Ordering::Acquire) {
match thread_server.recv_timeout(Duration::from_millis(250)) {
Ok(Some(request)) => {
let admitted = accept_in_flight
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
(current < HTTP_WORKERS).then_some(current + 1)
})
.is_ok();
if !admitted {
let _ = request.respond(error_response(StatusCode(429), "busy"));
let accept_active_connections = Arc::clone(&active_connections);
let accept_admission = admission.clone();
let accept_observer = config.request_observer.clone();
// The listener deliberately has no SO_RCVTIMEO: that option also
// times out accept() on Linux. Shutdown wakes this blocking accept
// with one bounded local connection after clearing `accepting`.
let accept_thread = std::thread::spawn(move || loop {
match listener.accept() {
Ok((mut stream, _)) => {
if !thread_accepting.load(Ordering::Acquire) {
break;
}
let Some(permit) = InFlightPermit::try_acquire(Arc::clone(&accept_in_flight))
else {
observe_status(accept_observer.as_ref(), 429);
reject_connection(&mut stream, &error_response(429, "busy"));
continue;
};
let shutdown_stream = match stream.try_clone() {
Ok(stream) => stream,
Err(_) => {
drop(permit);
observe_status(accept_observer.as_ref(), 503);
reject_connection(&mut stream, &error_response(503, "unavailable"));
continue;
}
match request_tx.try_send(request) {
Ok(()) => {}
Err(TrySendError::Full(request)) => {
accept_in_flight.fetch_sub(1, Ordering::AcqRel);
let _ = request.respond(error_response(StatusCode(429), "busy"));
}
Err(TrySendError::Disconnected(request)) => {
accept_in_flight.fetch_sub(1, Ordering::AcqRel);
let _ =
request.respond(error_response(StatusCode(503), "unavailable"));
break;
}
};
let active_id = accept_active_connections.insert(shutdown_stream);
let job = ConnectionJob {
stream,
active_connections: Arc::clone(&accept_active_connections),
active_id,
_permit: permit,
};
match request_tx.try_send(job) {
Ok(()) => {}
Err(TrySendError::Full(mut job)) => {
observe_status(accept_observer.as_ref(), 429);
reject_connection(&mut job.stream, &error_response(429, "busy"));
}
Err(TrySendError::Disconnected(mut job)) => {
observe_status(accept_observer.as_ref(), 503);
reject_connection(&mut job.stream, &error_response(503, "unavailable"));
thread_accepting.store(false, Ordering::Release);
accept_admission.close();
accept_active_connections.shutdown_all();
break;
}
}
Ok(None) => {}
Err(_) if !thread_accepting.load(Ordering::Acquire) => break,
// Malformed or timed-out client connections are surfaced
// by tiny_http through this queue too; they must not kill
// discovery for every other controller.
Err(_) => {}
}
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
Err(_) if !thread_accepting.load(Ordering::Acquire) => break,
Err(_) => {
thread_accepting.store(false, Ordering::Release);
accept_admission.close();
accept_active_connections.shutdown_all();
break;
}
}
});
let mdns = if config.advertise_mdns {
let advertised_addresses = config.advertised_addresses.or_else(|| {
(!config.bind_addr.ip().is_unspecified()).then(|| vec![config.bind_addr.ip()])
});
match MdnsRegistration::register(
&config.device_uuid,
port,
&config.sdk_version,
config.advertised_addresses,
advertised_addresses,
) {
Ok(mdns) => Some(mdns),
Err(error) => {
admission.close();
accepting.store(false, Ordering::Release);
server.unblock();
active_connections.shutdown_all();
wake_acceptor(accept_wake_addr);
let _ = accept_thread.join();
for worker in worker_threads {
let _ = worker.join();
@@ -210,9 +330,10 @@ impl LanService {
Ok((
Self {
port,
server,
admission,
accepting,
accept_wake_addr,
active_connections,
accept_thread: Some(accept_thread),
worker_threads,
mdns,
@@ -227,12 +348,17 @@ impl LanService {
pub fn shutdown(&mut self) {
self.admission.close();
self.accepting.store(false, Ordering::Release);
// Stop advertising before making the HTTP endpoint unreachable. A
// strict official client may cache a discovered renderer for minutes,
// so publishing a dead port even briefly is worse than keeping the
// already-closed admission socket alive during the bounded goodbye.
if let Some(mdns) = self.mdns.take() {
mdns.shutdown();
}
self.accepting.store(false, Ordering::Release);
self.server.unblock();
self.active_connections.shutdown_all();
if let Some(thread) = self.accept_thread.take() {
wake_acceptor(self.accept_wake_addr);
let _ = thread.join();
}
for thread in self.worker_threads.drain(..) {
@@ -267,10 +393,28 @@ fn validate_config(config: &LanServiceConfig) -> Result<(), LanError> {
{
return Err(LanError::InvalidConfig);
}
if config.advertise_mdns {
let bind_ip = config.bind_addr.ip();
if bind_ip.is_loopback() {
return Err(LanError::InvalidConfig);
}
if let Some(addresses) = &config.advertised_addresses {
if addresses.is_empty()
|| addresses.iter().any(|address| {
!address.is_ipv4()
|| address.is_loopback()
|| address.is_unspecified()
|| (!bind_ip.is_unspecified() && *address != bind_ip)
})
{
return Err(LanError::InvalidConfig);
}
}
}
Ok(())
}
fn bind_http(bind_addr: SocketAddr, read_timeout: Duration) -> Result<Server, LanError> {
fn bind_http(bind_addr: SocketAddr) -> Result<TcpListener, LanError> {
let domain = if bind_addr.is_ipv4() {
Domain::IPV4
} else {
@@ -282,76 +426,289 @@ fn bind_http(bind_addr: SocketAddr, read_timeout: Duration) -> Result<Server, La
if bind_addr.is_ipv6() {
socket.set_only_v6(true).map_err(|_| LanError::Bind)?;
}
// Linux propagates SO_RCVTIMEO from a listening socket to accepted TCP
// sockets. That bounds stalled request reads inside tiny_http.
socket
.set_read_timeout(Some(read_timeout))
.map_err(|_| LanError::Bind)?;
socket.bind(&bind_addr.into()).map_err(|_| LanError::Bind)?;
socket.listen(128).map_err(|_| LanError::Bind)?;
let listener: TcpListener = socket.into();
Server::from_listener(listener, None).map_err(|_| LanError::Bind)
Ok(socket.into())
}
fn handle_request(
mut request: Request,
fn accept_wake_addr(listener_addr: SocketAddr) -> SocketAddr {
if !listener_addr.ip().is_unspecified() {
return listener_addr;
}
let loopback = if listener_addr.is_ipv4() {
IpAddr::V4(Ipv4Addr::LOCALHOST)
} else {
IpAddr::V6(Ipv6Addr::LOCALHOST)
};
SocketAddr::new(loopback, listener_addr.port())
}
fn wake_acceptor(address: SocketAddr) {
if let Ok(stream) = TcpStream::connect_timeout(&address, ACCEPT_WAKE_TIMEOUT) {
let _ = stream.shutdown(Shutdown::Both);
}
}
fn recover_lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
mutex
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReadRequestError {
Invalid,
TooLarge,
TimedOut,
ExpectationFailed,
}
struct ParsedRequest {
method: LanHttpMethod,
path: String,
content_type_json: bool,
body: Zeroizing<Vec<u8>>,
remote_ip: Option<IpAddr>,
}
fn handle_connection(
stream: &mut TcpStream,
read_timeout: Duration,
projection: &LanProjection,
policy: &EndpointPolicy,
admission: &AdmissionSender,
rate_limiter: &Arc<Mutex<PostRateLimiter>>,
observer: Option<&mpsc::SyncSender<LanRequestObservation>>,
) {
let method = request.method().clone();
let path = request.url().to_string();
let response = match path.as_str() {
"/get-display-info" if method == Method::Get => {
json_response(StatusCode(200), &projection.display_info())
let _ = stream.set_write_timeout(Some(read_timeout));
let request = match read_request(stream, read_timeout) {
Ok(request) => request,
Err(error) => {
let status = match error {
ReadRequestError::TooLarge => 413,
ReadRequestError::Invalid | ReadRequestError::TimedOut => 400,
ReadRequestError::ExpectationFailed => 417,
};
let code = match error {
ReadRequestError::TooLarge => "request_too_large",
ReadRequestError::Invalid => "invalid_request",
ReadRequestError::TimedOut => "request_timeout",
ReadRequestError::ExpectationFailed => "expectation_failed",
};
observe_status(observer, status);
let _ = write_response(stream, &error_response(status, code));
return;
}
"/get-connect-info" if method == Method::Get => {
json_response(StatusCode(200), &projection.connect_info())
}
"/connect-to-qconnect" if method == Method::Post => {
handle_connect(&mut request, policy, admission, rate_limiter)
}
"/get-display-info" | "/get-connect-info" | "/connect-to-qconnect" => {
error_response(StatusCode(405), "method_not_allowed")
}
_ => error_response(StatusCode(404), "not_found"),
};
let _ = request.respond(response);
let observed_route = route_for(&request.path);
let mut validation_error = None;
let response = handle_request(
&request,
projection,
policy,
admission,
rate_limiter,
&mut validation_error,
);
if let Some(observer) = observer {
let _ = observer.try_send(LanRequestObservation {
method: request.method,
route: observed_route,
status_code: response.status_code,
validation_error,
});
}
let _ = write_response(stream, &response);
}
fn handle_connect(
request: &mut Request,
fn observe_status(observer: Option<&mpsc::SyncSender<LanRequestObservation>>, status_code: u16) {
if let Some(observer) = observer {
let _ = observer.try_send(LanRequestObservation {
method: LanHttpMethod::Other,
route: LanHttpRoute::Unknown,
status_code,
validation_error: None,
});
}
}
fn read_request(
stream: &mut TcpStream,
timeout: Duration,
) -> Result<ParsedRequest, ReadRequestError> {
let deadline = Instant::now()
.checked_add(timeout)
.ok_or(ReadRequestError::Invalid)?;
let mut bytes = Zeroizing::new(Vec::with_capacity(4096));
let header_end = loop {
if let Some(end) = find_header_end(&bytes) {
if end > MAX_HEADER_BYTES {
return Err(ReadRequestError::TooLarge);
}
break end;
}
if bytes.len() >= MAX_HEADER_BYTES {
return Err(ReadRequestError::TooLarge);
}
read_more(stream, &mut bytes, deadline)?;
};
let head =
std::str::from_utf8(&bytes[..header_end - 4]).map_err(|_| ReadRequestError::Invalid)?;
let mut lines = head.split("\r\n");
let mut request_line = lines
.next()
.ok_or(ReadRequestError::Invalid)?
.split_ascii_whitespace();
let method = match request_line.next().ok_or(ReadRequestError::Invalid)? {
"GET" => LanHttpMethod::Get,
"POST" => LanHttpMethod::Post,
_ => LanHttpMethod::Other,
};
let path = request_line
.next()
.filter(|path| path.starts_with('/'))
.ok_or(ReadRequestError::Invalid)?
.to_string();
let version = request_line.next().ok_or(ReadRequestError::Invalid)?;
if request_line.next().is_some() || !matches!(version, "HTTP/1.0" | "HTTP/1.1") {
return Err(ReadRequestError::Invalid);
}
let mut content_length = None;
let mut content_type_json = false;
for line in lines {
let (name, value) = line.split_once(':').ok_or(ReadRequestError::Invalid)?;
let name = name.trim();
let value = value.trim();
if name.eq_ignore_ascii_case("content-length") {
if content_length.is_some() {
return Err(ReadRequestError::Invalid);
}
content_length = Some(
value
.parse::<usize>()
.map_err(|_| ReadRequestError::Invalid)?,
);
} else if name.eq_ignore_ascii_case("content-type") {
content_type_json = value
.split(';')
.next()
.is_some_and(|value| value.trim().eq_ignore_ascii_case("application/json"));
} else if name.eq_ignore_ascii_case("transfer-encoding") {
return Err(ReadRequestError::Invalid);
} else if name.eq_ignore_ascii_case("expect") {
return Err(ReadRequestError::ExpectationFailed);
}
}
let body_len = content_length.unwrap_or(0);
if body_len > MAX_BODY_BYTES {
return Err(ReadRequestError::TooLarge);
}
let total_len = header_end
.checked_add(body_len)
.ok_or(ReadRequestError::TooLarge)?;
while bytes.len() < total_len {
read_more(stream, &mut bytes, deadline)?;
}
if bytes.len() > total_len {
bytes[total_len..].fill(0);
}
bytes.truncate(total_len);
let body = Zeroizing::new(bytes[header_end..].to_vec());
Ok(ParsedRequest {
method,
path,
content_type_json,
body,
remote_ip: stream.peer_addr().ok().map(|address| address.ip()),
})
}
fn read_more(
stream: &mut TcpStream,
bytes: &mut Vec<u8>,
deadline: Instant,
) -> Result<(), ReadRequestError> {
let remaining = deadline
.checked_duration_since(Instant::now())
.filter(|remaining| !remaining.is_zero())
.ok_or(ReadRequestError::TimedOut)?;
stream
.set_read_timeout(Some(remaining))
.map_err(|_| ReadRequestError::Invalid)?;
let mut chunk = Zeroizing::new([0_u8; 4096]);
match stream.read(&mut chunk[..]) {
Ok(0) => Err(ReadRequestError::Invalid),
Ok(count) => {
bytes.extend_from_slice(&chunk[..count]);
Ok(())
}
Err(error)
if matches!(
error.kind(),
io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
) =>
{
Err(ReadRequestError::TimedOut)
}
Err(_) => Err(ReadRequestError::Invalid),
}
}
fn find_header_end(bytes: &[u8]) -> Option<usize> {
bytes
.windows(4)
.position(|window| window == b"\r\n\r\n")
.map(|position| position + 4)
}
fn route_for(path: &str) -> LanHttpRoute {
match path {
"/get-display-info" => LanHttpRoute::GetDisplayInfo,
"/get-connect-info" => LanHttpRoute::GetConnectInfo,
"/connect-to-qconnect" => LanHttpRoute::ConnectToQconnect,
_ => LanHttpRoute::Unknown,
}
}
fn handle_request(
request: &ParsedRequest,
projection: &LanProjection,
policy: &EndpointPolicy,
admission: &AdmissionSender,
rate_limiter: &Arc<Mutex<PostRateLimiter>>,
) -> Response<std::io::Cursor<Vec<u8>>> {
if !has_json_content_type(request) {
return error_response(StatusCode(415), "json_required");
}
if declared_body_too_large(request) {
return error_response(StatusCode(413), "body_too_large");
}
let remote_ip = request.remote_addr().map(SocketAddr::ip);
if !rate_limiter
.lock()
.expect("LAN rate-limit lock poisoned")
.allow(remote_ip, Instant::now())
{
return error_response(StatusCode(429), "rate_limited");
validation_error: &mut Option<ValidationError>,
) -> HttpResponse {
match (route_for(&request.path), request.method) {
(LanHttpRoute::GetDisplayInfo, LanHttpMethod::Get) => {
json_response(200, &projection.display_info())
}
(LanHttpRoute::GetConnectInfo, LanHttpMethod::Get) => {
json_response(200, &projection.connect_info())
}
(LanHttpRoute::ConnectToQconnect, LanHttpMethod::Post) => {
handle_connect(request, policy, admission, rate_limiter, validation_error)
}
(LanHttpRoute::Unknown, _) => error_response(404, "not_found"),
_ => error_response(405, "method_not_allowed"),
}
}
let mut body = Vec::new();
if request
.as_reader()
.take((MAX_BODY_BYTES + 1) as u64)
.read_to_end(&mut body)
.is_err()
{
return error_response(StatusCode(400), "body_read_failed");
fn handle_connect(
request: &ParsedRequest,
policy: &EndpointPolicy,
admission: &AdmissionSender,
rate_limiter: &Arc<Mutex<PostRateLimiter>>,
validation_error: &mut Option<ValidationError>,
) -> HttpResponse {
if !request.content_type_json {
return error_response(415, "json_required");
}
if body.len() > MAX_BODY_BYTES {
return error_response(StatusCode(413), "body_too_large");
if !recover_lock(rate_limiter).allow(request.remote_ip, Instant::now()) {
return error_response(429, "rate_limited");
}
let now = SystemTime::now()
@@ -359,60 +716,134 @@ fn handle_connect(
.unwrap_or_default()
.as_secs()
.min(i64::MAX as u64) as i64;
let candidate = match parse_and_validate(&body, policy, now) {
let candidate = match parse_and_validate(&request.body, policy, now) {
Ok(candidate) => candidate,
Err(ValidationError::BecomeActiveRequired) => {
return error_response(StatusCode(422), "active_required")
Err(error @ ValidationError::BecomeActiveRequired) => {
*validation_error = Some(error);
return error_response(422, "active_required");
}
Err(error) => {
*validation_error = Some(error);
return error_response(400, "invalid_handoff");
}
Err(_) => return error_response(StatusCode(400), "invalid_handoff"),
};
match admission.submit(candidate) {
Ok(()) => json_response(StatusCode(200), &serde_json::json!({})),
Err(SubmitError::Closed) => error_response(StatusCode(503), "unavailable"),
Ok(()) => json_response(200, &serde_json::json!({})),
Err(SubmitError::Closed) => error_response(503, "unavailable"),
}
}
fn has_json_content_type(request: &Request) -> bool {
request.headers().iter().any(|header| {
header.field.equiv("Content-Type")
&& header
.value
.as_str()
.split(';')
.next()
.is_some_and(|value| value.trim().eq_ignore_ascii_case("application/json"))
})
struct HttpResponse {
status_code: u16,
body: Vec<u8>,
}
fn declared_body_too_large(request: &Request) -> bool {
request.headers().iter().any(|header| {
header.field.equiv("Content-Length")
&& header
.value
.as_str()
.trim()
.parse::<usize>()
.is_ok_and(|length| length > MAX_BODY_BYTES)
})
fn json_response<T: serde::Serialize>(status_code: u16, value: &T) -> HttpResponse {
HttpResponse {
status_code,
body: serde_json::to_vec(value).unwrap_or_else(|_| b"{}".to_vec()),
}
}
fn json_response<T: serde::Serialize>(
status: StatusCode,
value: &T,
) -> Response<std::io::Cursor<Vec<u8>>> {
let body = serde_json::to_vec(value).unwrap_or_else(|_| b"{}".to_vec());
Response::from_data(body)
.with_status_code(status)
.with_header(json_content_type())
fn error_response(status_code: u16, code: &'static str) -> HttpResponse {
json_response(status_code, &serde_json::json!({ "error": code }))
}
fn error_response(status: StatusCode, code: &'static str) -> Response<std::io::Cursor<Vec<u8>>> {
json_response(status, &serde_json::json!({ "error": code }))
fn write_response(stream: &mut TcpStream, response: &HttpResponse) -> io::Result<()> {
let reason = match response.status_code {
200 => "OK",
400 => "Bad Request",
404 => "Not Found",
405 => "Method Not Allowed",
413 => "Payload Too Large",
415 => "Unsupported Media Type",
417 => "Expectation Failed",
422 => "Unprocessable Entity",
429 => "Too Many Requests",
503 => "Service Unavailable",
_ => "Error",
};
write!(
stream,
"HTTP/1.1 {} {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
response.status_code,
reason,
response.body.len()
)?;
stream.write_all(&response.body)?;
stream.flush()
}
fn json_content_type() -> Header {
Header::from_bytes("Content-Type", "application/json").expect("static JSON content-type header")
/// Send a bounded overload/shutdown response without immediately closing a
/// socket that still has unread request bytes. On Linux, closing with unread
/// bytes emits a reset that can hide an otherwise complete 429 from strict
/// clients. Read at most one declared request under an absolute 25 ms deadline,
/// then respond and close the write half. A trickling peer therefore cannot
/// retain the accept thread indefinitely, while a normal official POST gets a
/// complete HTTP response and clean EOF.
fn reject_connection(stream: &mut TcpStream, response: &HttpResponse) {
let Some(deadline) = Instant::now().checked_add(REJECTION_DRAIN_TIMEOUT) else {
return;
};
let drain_cap = MAX_HEADER_BYTES.saturating_add(MAX_BODY_BYTES);
let mut drained = Zeroizing::new(Vec::with_capacity(1024));
let mut scratch = Zeroizing::new([0_u8; 1024]);
while drained.len() < drain_cap {
let Some(remaining_time) = deadline
.checked_duration_since(Instant::now())
.filter(|remaining| !remaining.is_zero())
else {
break;
};
let _ = stream.set_read_timeout(Some(remaining_time));
let remaining_bytes = (drain_cap - drained.len()).min(scratch.len());
match stream.read(&mut scratch[..remaining_bytes]) {
Ok(0) => break,
Ok(count) => {
drained.extend_from_slice(&scratch[..count]);
if rejected_request_is_complete(&drained) {
break;
}
}
Err(error)
if matches!(
error.kind(),
io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
) =>
{
break;
}
Err(_) => break,
}
}
let _ = stream.set_write_timeout(Some(Duration::from_secs(1)));
let _ = write_response(stream, response);
let _ = stream.shutdown(Shutdown::Write);
}
fn rejected_request_is_complete(bytes: &[u8]) -> bool {
let Some(header_end) = find_header_end(bytes) else {
return false;
};
let Ok(head) = std::str::from_utf8(&bytes[..header_end.saturating_sub(4)]) else {
return true;
};
let mut content_length = 0_usize;
for line in head.split("\r\n").skip(1) {
let Some((name, value)) = line.split_once(':') else {
return true;
};
if name.trim().eq_ignore_ascii_case("content-length") {
let Ok(parsed) = value.trim().parse::<usize>() else {
return true;
};
content_length = parsed.min(MAX_BODY_BYTES);
break;
}
}
bytes.len() >= header_end.saturating_add(content_length)
}
#[derive(Debug, Clone, Copy)]
@@ -469,4 +900,36 @@ mod tests {
assert!(!limiter.allow(ip, now));
assert!(limiter.allow(ip, now + Duration::from_secs(10)));
}
#[test]
fn finds_only_complete_crlf_header_boundary() {
assert_eq!(find_header_end(b"GET / HTTP/1.1\r\n\r\nbody"), Some(18));
assert_eq!(find_header_end(b"GET / HTTP/1.1\n\n"), None);
}
#[test]
fn in_flight_permit_is_released_during_unwind() {
let count = Arc::new(AtomicUsize::new(0));
let unwind_count = Arc::clone(&count);
let result = std::panic::catch_unwind(move || {
let _permit = InFlightPermit::try_acquire(unwind_count).unwrap();
panic!("intentional permit-unwind regression");
});
assert!(result.is_err());
assert_eq!(count.load(Ordering::Acquire), 0);
}
#[test]
fn poisoned_server_lock_remains_recoverable() {
let value = Mutex::new(0_u8);
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _guard = value.lock().unwrap();
panic!("intentional lock-poison regression");
}));
assert!(result.is_err());
*recover_lock(&value) = 1;
assert_eq!(*recover_lock(&value), 1);
}
}
+312 -26
View File
@@ -1,6 +1,6 @@
use std::io::{Read, Write};
use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use qconnect_lan::{
ConnectInfo, DeviceType, DisplayInfo, EndpointPolicy, LanError, LanProjection, LanService,
@@ -81,20 +81,86 @@ fn test_config(bind_addr: SocketAddr) -> LanServiceConfig {
config
}
fn stalled_post(port: u16) -> TcpStream {
fn stalled_post(port: u16, content_length: usize) -> TcpStream {
let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, port)).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
write!(
stream,
"POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: 2048\r\nConnection: close\r\n\r\n{{"
"POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {content_length}\r\nConnection: close\r\n\r\n{{"
)
.unwrap();
stream.flush().unwrap();
stream
}
fn stalled_headers(port: u16) -> TcpStream {
let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, port)).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
stream
.write_all(b"POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Ty")
.unwrap();
stream.flush().unwrap();
stream
}
fn read_raw_response(stream: &mut TcpStream) -> String {
let mut response = String::new();
match stream.read_to_string(&mut response) {
Ok(_) => response,
Err(error) if error.kind() == std::io::ErrorKind::ConnectionReset => response,
Err(error) => panic!("failed to read raw HTTP response: {error}"),
}
}
fn read_raw_response_without_reset(stream: &mut TcpStream) -> String {
let mut response = String::new();
stream
.read_to_string(&mut response)
.expect("overload response must end with a clean EOF");
response
}
fn raw_exchange(port: u16, request: &[u8]) -> String {
let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, port)).unwrap();
stream
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
stream
.set_write_timeout(Some(Duration::from_secs(2)))
.unwrap();
stream.write_all(request).unwrap();
stream.flush().unwrap();
read_raw_response(&mut stream)
}
async fn wait_for_status(base: &str, expected: StatusCode, budget: Duration) {
let client = reqwest::Client::builder()
.timeout(Duration::from_secs(1))
.build()
.unwrap();
let deadline = Instant::now() + budget;
loop {
let status = client
.get(format!("{base}/get-display-info"))
.send()
.await
.unwrap()
.status();
if status == expected {
return;
}
assert!(
Instant::now() < deadline,
"expected HTTP {expected}, last response was {status}"
);
tokio::time::sleep(Duration::from_millis(10)).await;
}
}
fn handoff(session_id: &str, become_active: bool) -> serde_json::Value {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
@@ -285,36 +351,243 @@ async fn connect_projection_updates_without_restarting_listener() {
}
#[tokio::test]
async fn stalled_bodies_time_out_and_concurrency_stays_bounded() {
let (mut service, _, base, _) = start_service_with_timeout(Duration::from_millis(500));
let mut first = stalled_post(service.port());
let second = stalled_post(service.port());
let deadline = std::time::Instant::now() + Duration::from_millis(250);
loop {
let response = reqwest::get(format!("{base}/get-display-info"))
.await
.unwrap();
if response.status() == StatusCode::TOO_MANY_REQUESTS {
break;
}
assert_eq!(response.status(), StatusCode::OK);
assert!(
std::time::Instant::now() < deadline,
"workers never saturated"
);
tokio::time::sleep(Duration::from_millis(10)).await;
async fn listener_survives_idle_beyond_per_request_deadline() {
let (mut service, _, base, _) = start_service_with_timeout(Duration::from_millis(100));
tokio::time::sleep(Duration::from_millis(250)).await;
wait_for_status(&base, StatusCode::OK, Duration::from_secs(1)).await;
service.shutdown();
}
#[tokio::test]
async fn two_partial_bodies_are_bounded_and_capacity_recovers() {
let (mut service, _, base, _) = start_service_with_timeout(Duration::from_millis(750));
let mut first = stalled_post(service.port(), 512);
let mut second = stalled_post(service.port(), 512);
wait_for_status(
&base,
StatusCode::TOO_MANY_REQUESTS,
Duration::from_millis(300),
)
.await;
for stream in [&mut first, &mut second] {
let response = read_raw_response(stream);
assert!(response.starts_with("HTTP/1.1 400"), "{response}");
}
wait_for_status(&base, StatusCode::OK, Duration::from_secs(1)).await;
let mut response = String::new();
first.read_to_string(&mut response).unwrap();
assert!(response.starts_with("HTTP/1.1 400"), "{response}");
service.shutdown();
}
#[tokio::test]
async fn two_partial_headers_are_bounded_and_capacity_recovers() {
let (mut service, _, base, _) = start_service_with_timeout(Duration::from_millis(750));
let mut first = stalled_headers(service.port());
let mut second = stalled_headers(service.port());
wait_for_status(
&base,
StatusCode::TOO_MANY_REQUESTS,
Duration::from_millis(300),
)
.await;
for stream in [&mut first, &mut second] {
let response = read_raw_response(stream);
assert!(response.starts_with("HTTP/1.1 400"), "{response}");
}
wait_for_status(&base, StatusCode::OK, Duration::from_secs(1)).await;
service.shutdown();
}
#[test]
fn saturated_full_post_receives_complete_429_without_reset() {
let (mut service, _, _, _) = start_service_with_timeout(Duration::from_secs(2));
let first = stalled_post(service.port(), 512);
let second = stalled_post(service.port(), 512);
std::thread::sleep(Duration::from_millis(25));
let body = vec![b'x'; 32 * 1024];
let mut third = TcpStream::connect((Ipv4Addr::LOCALHOST, service.port())).unwrap();
third
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
write!(
third,
"POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
)
.unwrap();
third.write_all(&body).unwrap();
third.flush().unwrap();
let response = read_raw_response_without_reset(&mut third);
assert!(response.starts_with("HTTP/1.1 429"), "{response}");
let (head, response_body) = response.split_once("\r\n\r\n").unwrap();
let declared = head
.lines()
.find_map(|line| {
line.strip_prefix("Content-Length: ")
.and_then(|value| value.parse::<usize>().ok())
})
.unwrap();
assert_eq!(response_body.len(), declared);
assert_eq!(response_body, r#"{"error":"busy"}"#);
drop(first);
drop(second);
service.shutdown();
}
#[test]
fn mdns_failure_after_bind_releases_listener_and_never_advertises_loopback() {
fn rejected_header_trickle_cannot_delay_shutdown() {
let (mut service, _, _, _) = start_service_with_timeout(Duration::from_secs(2));
let first = stalled_post(service.port(), 512);
let second = stalled_post(service.port(), 512);
std::thread::sleep(Duration::from_millis(25));
let mut third = TcpStream::connect((Ipv4Addr::LOCALHOST, service.port())).unwrap();
third.write_all(b"P").unwrap();
third.flush().unwrap();
let drip = std::thread::spawn(move || {
for _ in 0..100 {
std::thread::sleep(Duration::from_millis(10));
if third.write_all(b"O").is_err() || third.flush().is_err() {
break;
}
}
});
std::thread::sleep(Duration::from_millis(75));
let started = Instant::now();
service.shutdown();
assert!(
started.elapsed() < Duration::from_millis(250),
"shutdown was retained by rejected trickle: {:?}",
started.elapsed()
);
drip.join().unwrap();
drop(first);
drop(second);
}
#[tokio::test]
async fn trickled_body_hits_absolute_deadline_and_capacity_recovers() {
let timeout = Duration::from_millis(500);
let (mut service, _, base, _) = start_service_with_timeout(timeout);
let mut writer = TcpStream::connect((Ipv4Addr::LOCALHOST, service.port())).unwrap();
writer
.set_write_timeout(Some(Duration::from_secs(1)))
.unwrap();
let started = Instant::now();
writer
.write_all(
b"POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: 64\r\nConnection: close\r\n\r\n{",
)
.unwrap();
writer.flush().unwrap();
let mut reader = writer.try_clone().unwrap();
reader
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let drip = std::thread::spawn(move || {
for _ in 0..4 {
std::thread::sleep(Duration::from_millis(80));
if writer.write_all(b"x").is_err() || writer.flush().is_err() {
break;
}
}
});
let response = read_raw_response(&mut reader);
let elapsed = started.elapsed();
drip.join().unwrap();
assert!(response.starts_with("HTTP/1.1 400"), "{response}");
assert!(
elapsed >= Duration::from_millis(300),
"elapsed: {elapsed:?}"
);
assert!(elapsed < Duration::from_millis(700), "elapsed: {elapsed:?}");
wait_for_status(&base, StatusCode::OK, Duration::from_secs(1)).await;
service.shutdown();
}
#[test]
fn oversized_declared_body_is_rejected_without_waiting_for_body() {
let (mut service, _, _, _) = start_service_with_timeout(Duration::from_secs(2));
let request = format!(
"POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
MAX_BODY_BYTES + 1
);
let started = Instant::now();
let response = raw_exchange(service.port(), request.as_bytes());
assert!(response.starts_with("HTTP/1.1 413"), "{response}");
assert!(started.elapsed() < Duration::from_secs(1));
service.shutdown();
}
#[test]
fn oversized_incomplete_header_is_rejected_at_the_header_cap() {
let (mut service, _, _, _) = start_service_with_timeout(Duration::from_secs(2));
let request = format!(
"GET /get-display-info HTTP/1.1\r\nHost: localhost\r\nX-Fill: {}",
"a".repeat(17 * 1024)
);
let response = raw_exchange(service.port(), request.as_bytes());
assert!(response.starts_with("HTTP/1.1 413"), "{response}");
service.shutdown();
}
#[test]
fn expect_continue_is_rejected_immediately_without_reading_the_body() {
let (mut service, _, _, _) = start_service_with_timeout(Duration::from_secs(2));
let request = b"POST /connect-to-qconnect HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: 1024\r\nExpect: 100-continue\r\nConnection: close\r\n\r\n";
let started = Instant::now();
let response = raw_exchange(service.port(), request);
assert!(response.starts_with("HTTP/1.1 417"), "{response}");
assert!(started.elapsed() < Duration::from_secs(1));
service.shutdown();
}
#[tokio::test]
async fn shutdown_interrupts_two_slowloris_connections_and_closes_the_port() {
let (mut service, inbox, base, _) = start_service_with_timeout(Duration::from_secs(5));
let port = service.port();
let first = stalled_headers(port);
let second = stalled_headers(port);
wait_for_status(
&base,
StatusCode::TOO_MANY_REQUESTS,
Duration::from_millis(500),
)
.await;
let started = Instant::now();
service.shutdown();
assert!(started.elapsed() < Duration::from_secs(1));
assert!(inbox.is_closed());
assert!(TcpStream::connect_timeout(
&SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
Duration::from_millis(100),
)
.is_err());
service.shutdown();
drop(first);
drop(second);
}
#[test]
fn invalid_mdns_address_is_rejected_before_bind() {
let reservation = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
let bind_addr = reservation.local_addr().unwrap();
drop(reservation);
@@ -326,8 +599,21 @@ fn mdns_failure_after_bind_releases_listener_and_never_advertises_loopback() {
Ok(_) => panic!("loopback mDNS address unexpectedly accepted"),
Err(error) => error,
};
assert!(matches!(error, LanError::NoLanAddresses));
assert!(matches!(error, LanError::InvalidConfig));
std::thread::sleep(Duration::from_millis(25));
assert!(TcpStream::connect_timeout(&bind_addr, Duration::from_millis(100)).is_err());
}
#[test]
fn explicit_mdns_address_must_match_specific_bind_address() {
let mut config = test_config(SocketAddr::from(([192, 0, 2, 10], 0)));
config.advertise_mdns = true;
config.advertised_addresses = Some(vec![IpAddr::V4(Ipv4Addr::new(192, 0, 2, 11))]);
let error = match LanService::start(config) {
Ok(_) => panic!("mismatched bind and advertised addresses unexpectedly accepted"),
Err(error) => error,
};
assert!(matches!(error, LanError::InvalidConfig));
}
+2
View File
@@ -10,8 +10,10 @@ qconnect-protocol = { path = "../qconnect-protocol" }
async-trait = { workspace = true }
tokio = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
thiserror = { workspace = true }
futures-util = { workspace = true }
tokio-tungstenite = { version = "0.30", features = ["rustls-tls-webpki-roots"] }
prost = "0.14"
log = "0.4"
zeroize = "1"
+83 -5
View File
@@ -1,4 +1,12 @@
#[derive(Debug, Clone)]
use std::fmt;
use zeroize::Zeroize;
/// WebSocket transport configuration.
///
/// The delegated endpoint can itself contain sensitive routing material, so
/// both it and `jwt_qws` are redacted from `Debug` and overwritten on drop.
#[derive(Clone)]
pub struct WsTransportConfig {
pub endpoint_url: String,
pub jwt_qws: Option<String>,
@@ -11,10 +19,9 @@ pub struct WsTransportConfig {
pub reconnect_backoff_max_ms: u64,
/// Maximum number of consecutive reconnect attempts before the transport
/// gives up and shuts down. The counter resets only when a session-level
/// join is confirmed (cloud emits MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE),
/// not when the WS / TCP connection succeeds — Qobuz cloud accepts the WS
/// connection before rejecting the session join, so a TCP-level reset
/// would mask the failure mode behind issue #358.
/// join is confirmed (`SRVR_CTRL_SESSION_STATE` for controllers or
/// targeted `SRVR_RNDR_SET_ACTIVE=true` for delegated renderers), not when
/// the WS / TCP connection succeeds.
///
/// `None` means unlimited (legacy behavior, retained for tests).
pub reconnect_max_attempts: Option<u32>,
@@ -31,6 +38,44 @@ pub struct WsTransportConfig {
pub qcloud_proto: u32,
}
impl WsTransportConfig {
/// Overwrite and remove delegated QWS routing material immediately.
///
/// This is also the single cleanup path used by `Drop`, which keeps the
/// behavior directly testable without inspecting freed memory.
fn clear_sensitive(&mut self) {
self.endpoint_url.zeroize();
if let Some(mut jwt_qws) = self.jwt_qws.take() {
jwt_qws.zeroize();
}
}
}
impl fmt::Debug for WsTransportConfig {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("WsTransportConfig")
.field("endpoint_url", &"[REDACTED]")
.field("jwt_qws", &self.jwt_qws.as_ref().map(|_| "[REDACTED]"))
.field("require_jwt", &self.require_jwt)
.field("reconnect_backoff_ms", &self.reconnect_backoff_ms)
.field("reconnect_backoff_max_ms", &self.reconnect_backoff_max_ms)
.field("reconnect_max_attempts", &self.reconnect_max_attempts)
.field("reconnect_idle_retry_ms", &self.reconnect_idle_retry_ms)
.field("connect_timeout_ms", &self.connect_timeout_ms)
.field("keepalive_interval_ms", &self.keepalive_interval_ms)
.field("auto_subscribe", &self.auto_subscribe)
.field("subscribe_channels", &self.subscribe_channels)
.field("qcloud_proto", &self.qcloud_proto)
.finish()
}
}
impl Drop for WsTransportConfig {
fn drop(&mut self) {
self.clear_sensitive();
}
}
impl Default for WsTransportConfig {
fn default() -> Self {
Self {
@@ -49,3 +94,36 @@ impl Default for WsTransportConfig {
}
}
}
#[cfg(test)]
mod tests {
use super::*;
const TEST_JWT: &str = "header.payload.secret-signature";
const TEST_ENDPOINT: &str = "wss://qws.example.test/ws?route=secret";
#[test]
fn debug_redacts_present_jwt() {
let mut config = WsTransportConfig::default();
config.endpoint_url = TEST_ENDPOINT.to_string();
config.jwt_qws = Some(TEST_JWT.to_string());
let debug = format!("{config:?}");
assert!(!debug.contains(TEST_JWT));
assert!(!debug.contains(TEST_ENDPOINT));
assert!(debug.contains("endpoint_url: \"[REDACTED]\""));
assert!(debug.contains("jwt_qws: Some(\"[REDACTED]\")"));
}
#[test]
fn cleanup_path_removes_jwt() {
let mut config = WsTransportConfig::default();
config.endpoint_url = TEST_ENDPOINT.to_string();
config.jwt_qws = Some(TEST_JWT.to_string());
config.clear_sensitive();
assert!(config.endpoint_url.is_empty());
assert!(config.jwt_qws.is_none());
}
}
+209 -45
View File
@@ -12,7 +12,7 @@ use prost::{
use qconnect_protocol::{
decode_inbound_json, decode_queue_server_events, decode_renderer_server_commands,
encode_outbound_payload_bytes, InboundEnvelope, OutboundEnvelope, QueueEventType,
QueueServerEvent, RendererServerCommand,
QueueServerEvent, RendererCommandType, RendererServerCommand,
};
use serde::{Deserialize, Serialize};
use tokio::{
@@ -21,8 +21,9 @@ use tokio::{
};
use tokio_tungstenite::{
connect_async_with_config,
tungstenite::{protocol::WebSocketConfig, Message as WsMessage},
tungstenite::{protocol::WebSocketConfig, Bytes, Message as WsMessage},
};
use zeroize::{Zeroize, Zeroizing};
use crate::{WsTransportConfig, WsTransportError};
@@ -38,11 +39,10 @@ pub enum TransportEvent {
Disconnected,
Authenticated,
Subscribed,
/// Emitted the first time we observe a `MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE`
/// frame on the active WS connection. This is the only proof the
/// session-level handshake actually went through (Qobuz cloud accepts the
/// WS upgrade well before the JOIN_SESSION negotiation, so `Connected` is
/// not a reliable signal — see issue #358).
/// Emitted on the first session-level acceptance proof for the active WS:
/// controller sessions receive `SRVR_CTRL_SESSION_STATE`; delegated
/// renderer handoffs receive targeted `SRVR_RNDR_SET_ACTIVE {active:true}`.
/// A WS upgrade alone is never sufficient proof.
SessionEstablished,
/// Emitted when the reconnect loop has exhausted
/// `reconnect_max_attempts` consecutive attempts without ever reaching
@@ -102,7 +102,6 @@ pub trait WsTransport: Send + Sync {
#[derive(Debug, Default)]
struct InMemoryState {
connected: bool,
last_config: Option<WsTransportConfig>,
sent_messages: Vec<OutboundEnvelope>,
}
@@ -152,9 +151,13 @@ impl WsTransport for InMemoryWsTransport {
}
state.connected = true;
state.last_config = Some(config);
drop(state);
// The in-memory transport never uses credentials. Dropping the owned
// config here invokes its zeroizing cleanup instead of retaining a
// needless copy for the lifetime of the transport.
drop(config);
self.events_tx
.send(TransportEvent::Connected)
.map_err(|_| WsTransportError::EventChannelClosed)?;
@@ -203,13 +206,26 @@ impl WsTransport for InMemoryWsTransport {
struct NativeState {
connected: bool,
running: bool,
last_config: Option<WsTransportConfig>,
}
struct NativeRuntime {
outbound_tx: mpsc::Sender<OutboundEnvelope>,
shutdown_tx: watch::Sender<bool>,
handle: JoinHandle<()>,
handle: Option<JoinHandle<()>>,
}
impl Drop for NativeRuntime {
fn drop(&mut self) {
let Some(handle) = self.handle.take() else {
return;
};
// A dropped JoinHandle normally detaches its task. Explicitly request
// shutdown and abort as a fallback so the task-owned config (and its
// delegated JWT) cannot remain retained after the transport is gone.
let _ = self.shutdown_tx.send(true);
handle.abort();
}
}
#[derive(Clone)]
@@ -251,7 +267,6 @@ impl WsTransport for NativeWsTransport {
{
let mut state = self.state.lock().await;
state.running = true;
state.last_config = Some(config.clone());
}
let (outbound_tx, outbound_rx) = mpsc::channel::<OutboundEnvelope>(512);
@@ -266,13 +281,13 @@ impl WsTransport for NativeWsTransport {
*runtime_guard = Some(NativeRuntime {
outbound_tx,
shutdown_tx,
handle,
handle: Some(handle),
});
Ok(())
}
async fn disconnect(&self) -> Result<(), WsTransportError> {
let runtime = self
let mut runtime = self
.runtime
.lock()
.await
@@ -283,6 +298,10 @@ impl WsTransport for NativeWsTransport {
runtime
.handle
.take()
.ok_or_else(|| {
WsTransportError::Internal("native transport task handle missing".to_string())
})?
.await
.map_err(|err| WsTransportError::Join(err.to_string()))?;
@@ -344,6 +363,11 @@ async fn run_native_transport_loop(
let (mut ws, _) = match connect_result {
Ok(Ok((ws, response))) => (ws, response),
Ok(Err(err)) => {
let reason = redact_sensitive(
format!("connect_error:{err}"),
config.jwt_qws.as_deref(),
Some(&config.endpoint_url),
);
match handle_reconnect_delay(
&events_tx,
&mut shutdown_rx,
@@ -351,7 +375,7 @@ async fn run_native_transport_loop(
&mut backoff,
max_backoff,
max_attempts,
format!("connect_error:{err}"),
reason,
)
.await
{
@@ -424,7 +448,11 @@ async fn run_native_transport_loop(
&events_tx,
TransportEvent::TransportError {
stage: "authenticate".to_string(),
message: err.to_string(),
message: redact_sensitive(
err.to_string(),
config.jwt_qws.as_deref(),
Some(&config.endpoint_url),
),
},
);
let _ = ws.close(None).await;
@@ -524,7 +552,11 @@ async fn run_native_transport_loop(
&events_tx,
TransportEvent::TransportError {
stage: "subscribe".to_string(),
message: err.to_string(),
message: redact_sensitive(
err.to_string(),
config.jwt_qws.as_deref(),
Some(&config.endpoint_url),
),
},
);
let _ = ws.close(None).await;
@@ -570,20 +602,15 @@ async fn run_native_transport_loop(
));
keepalive.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
// Per-connection latch. Only the first SESSION_STATE on this WS resets
// the reconnect counters — subsequent ones are no-ops, so we don't
// keep paying the lock cost.
// Per-connection latch. Only the first controller SESSION_STATE or
// delegated renderer SET_ACTIVE=true resets the reconnect counters.
let mut session_established = false;
// Per-connection keepalive half-open tracking (gap #6). The deadline is
// ~2.5× the ping interval: a healthy peer answers within one interval,
// so two unanswered pings plus 2.5× silence is a confident half-open
// signal without being trigger-happy.
let keepalive_deadline_ms = config
.keepalive_interval_ms
.max(1_000)
.saturating_mul(5)
/ 2;
let keepalive_deadline_ms = config.keepalive_interval_ms.max(1_000).saturating_mul(5) / 2;
let mut last_pong_at_ms = now_ms();
let mut outstanding_pings: u32 = 0;
@@ -640,9 +667,14 @@ async fn run_native_transport_loop(
incoming = ws.next() => {
match incoming {
Some(Ok(WsMessage::Binary(data))) => {
match handle_incoming_binary(&events_tx, &data) {
Ok(InboundFrameOutcome { session_state_seen }) => {
if session_state_seen && !session_established {
match handle_incoming_binary(
&events_tx,
&data,
config.jwt_qws.as_deref(),
Some(&config.endpoint_url),
) {
Ok(InboundFrameOutcome { session_established_seen }) => {
if session_established_seen && !session_established {
session_established = true;
reconnect_attempt = 0;
backoff = base_backoff;
@@ -654,7 +686,11 @@ async fn run_native_transport_loop(
&events_tx,
TransportEvent::TransportError {
stage: "decode_inbound_binary".to_string(),
message: err.to_string(),
message: redact_sensitive(
err.to_string(),
config.jwt_qws.as_deref(),
Some(&config.endpoint_url),
),
},
);
}
@@ -693,6 +729,11 @@ async fn run_native_transport_loop(
break;
}
let disconnect_reason = redact_sensitive(
disconnect_reason,
config.jwt_qws.as_deref(),
Some(&config.endpoint_url),
);
match handle_reconnect_delay(
&events_tx,
&mut shutdown_rx,
@@ -793,8 +834,8 @@ async fn handle_reconnect_delay(
/// and returns `true` so the caller can `continue` the loop.
///
/// NOTE: this resets the counters ONLY after Exhausted + a deliberate idle
/// wait — never on connect. The issue #358 latch (reset only on the first
/// SESSION_STATE) is therefore untouched.
/// wait — never on connect. The session-acceptance latch is therefore
/// untouched.
async fn idle_retry_after_exhausted(
shutdown_rx: &mut watch::Receiver<bool>,
reconnect_attempt: &mut u32,
@@ -818,18 +859,28 @@ async fn idle_retry_after_exhausted(
true
}
/// Side-channel result from `handle_incoming_binary`: we need to know whether
/// any decoded frame contained a `MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE` event,
/// because that is the signal the reconnect loop uses to reset its backoff
/// counters (see issue #358).
/// Side-channel result from `handle_incoming_binary`: the transport owns the
/// per-connection reconnect latch, so acceptance is recognized before decoded
/// events can be delayed in downstream broadcast consumers.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct InboundFrameOutcome {
session_state_seen: bool,
session_established_seen: bool,
}
fn is_session_established_command(command: &RendererServerCommand) -> bool {
command.command_type == RendererCommandType::SrvrRndrSetActive
&& command
.payload
.get("active")
.and_then(serde_json::Value::as_bool)
== Some(true)
}
fn handle_incoming_binary(
events_tx: &broadcast::Sender<TransportEvent>,
data: &[u8],
jwt_qws: Option<&str>,
sensitive_endpoint: Option<&str>,
) -> Result<InboundFrameOutcome, WsTransportError> {
let (cloud_message_type, payload) = decode_qcloud_frame(data)?;
let mut outcome = InboundFrameOutcome::default();
@@ -872,13 +923,14 @@ fn handle_incoming_binary(
log::debug!("[QConnect/Decode] Queue events decoded: {}", events.len());
for event in events {
if event.event_type == QueueEventType::SrvrCtrlSessionState {
outcome.session_state_seen = true;
outcome.session_established_seen = true;
}
emit(events_tx, TransportEvent::InboundQueueServerEvent(event));
}
}
Err(err) => {
log::warn!("[QConnect/Decode] Queue events decode error: {}", err);
let message = redact_sensitive(err.to_string(), jwt_qws, sensitive_endpoint);
log::warn!("[QConnect/Decode] Queue events decode error: {}", message);
}
}
@@ -891,6 +943,9 @@ fn handle_incoming_binary(
);
}
for command in commands {
if is_session_established_command(&command) {
outcome.session_established_seen = true;
}
emit(
events_tx,
TransportEvent::InboundRendererServerCommand(command),
@@ -898,7 +953,11 @@ fn handle_incoming_binary(
}
}
Err(err) => {
log::debug!("[QConnect/Decode] Renderer commands decode error: {}", err);
let message = redact_sensitive(err.to_string(), jwt_qws, sensitive_endpoint);
log::debug!(
"[QConnect/Decode] Renderer commands decode error: {}",
message
);
}
}
@@ -920,7 +979,11 @@ fn handle_incoming_binary(
Ok(error) => {
let msg_id = error.msg_id.unwrap_or(0);
let code = error.code.unwrap_or(0);
let descr = error.descr.unwrap_or_default();
let descr = redact_sensitive(
error.descr.unwrap_or_default(),
jwt_qws,
sensitive_endpoint,
);
log::warn!(
"[QConnect/Transport] Cloud error frame: msg_id={} code={} descr={:?}",
msg_id,
@@ -975,13 +1038,24 @@ async fn send_authenticate<S>(
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
let auth = Authenticate {
let mut auth = Authenticate {
msg_id: Some(next_msg_id(msg_id)),
msg_date: Some(now_ms()),
jwt: Some(jwt_qws.to_string()),
};
let frame = encode_qcloud_frame(MSG_TYPE_AUTHENTICATE, &auth.encode_to_vec());
ws.send(WsMessage::Binary(frame.into()))
// Encode directly into the qcloud frame so the protobuf payload is not a
// second temporary buffer containing the JWT. The unavoidable protobuf
// String copy is scrubbed before the first await point.
let frame_result = encode_qcloud_message(MSG_TYPE_AUTHENTICATE, &auth);
if let Some(jwt_qws) = auth.jwt.as_mut() {
jwt_qws.zeroize();
}
// Keep the encoded frame in an owner that scrubs its allocation when the
// last tungstenite `Bytes` reference is released.
let frame = Bytes::from_owner(Zeroizing::new(frame_result?));
ws.send(WsMessage::Binary(frame))
.await
.map_err(|err| WsTransportError::Protocol(format!("send authenticate: {err}")))
}
@@ -1043,6 +1117,42 @@ fn encode_qcloud_frame(msg_type: u8, payload: &[u8]) -> Vec<u8> {
frame
}
fn encode_qcloud_message<M: Message>(
msg_type: u8,
message: &M,
) -> Result<Vec<u8>, WsTransportError> {
let payload_len = message.encoded_len();
let mut frame = Vec::with_capacity(1 + 10 + payload_len);
frame.push(msg_type);
encode_varint(payload_len as u64, &mut frame);
message
.encode(&mut frame)
.map_err(|err| WsTransportError::Serialization(err.to_string()))?;
Ok(frame)
}
/// Remove delegated credentials and endpoint routing material from any
/// server-controlled diagnostic before it reaches logs or transport events.
/// Each replaced allocation is overwritten before the safe value continues.
fn redact_sensitive(
mut message: String,
jwt_qws: Option<&str>,
sensitive_endpoint: Option<&str>,
) -> String {
for sensitive in [jwt_qws, sensitive_endpoint]
.into_iter()
.flatten()
.filter(|value| !value.is_empty())
{
if message.contains(sensitive) {
let redacted = message.replace(sensitive, "[REDACTED]");
message.zeroize();
message = redacted;
}
}
message
}
fn decode_qcloud_frame(data: &[u8]) -> Result<(u8, &[u8]), WsTransportError> {
if data.is_empty() {
return Err(WsTransportError::Protocol("empty qcloud frame".to_string()));
@@ -1227,8 +1337,8 @@ mod tests {
.encode_to_vec();
let frame = encode_qcloud_frame(MSG_TYPE_ERROR, &error_payload);
let outcome = handle_incoming_binary(&events_tx, &frame).expect("decode frame");
assert!(!outcome.session_state_seen);
let outcome = handle_incoming_binary(&events_tx, &frame, None, None).expect("decode frame");
assert!(!outcome.session_established_seen);
// First event is always InboundFrameDecoded; skip it and assert the
// semantic event afterwards is CloudError, not the legacy
@@ -1250,6 +1360,60 @@ mod tests {
}
}
#[tokio::test]
async fn cloud_error_redacts_reflected_jwt() {
const JWT: &str = "header.payload.secret-signature";
let (events_tx, mut events_rx) = broadcast::channel::<TransportEvent>(8);
let error_payload = QwsErrorMessage {
msg_id: Some(7),
msg_date: None,
code: Some(403),
descr: Some(format!("rejected credential {JWT}")),
}
.encode_to_vec();
let frame = encode_qcloud_frame(MSG_TYPE_ERROR, &error_payload);
handle_incoming_binary(&events_tx, &frame, Some(JWT), None).expect("decode frame");
let _ = events_rx.recv().await.expect("frame event");
let event = events_rx.recv().await.expect("cloud error event");
let debug = format!("{event:?}");
assert!(!debug.contains(JWT));
match event {
TransportEvent::CloudError { descr, .. } => {
assert_eq!(descr, "rejected credential [REDACTED]");
}
other => panic!("expected CloudError, got {other:?}"),
}
}
#[test]
fn delegated_set_active_true_is_session_acceptance_proof() {
let active = RendererServerCommand {
command_type: RendererCommandType::SrvrRndrSetActive,
payload: serde_json::json!({ "active": true }),
};
let inactive = RendererServerCommand {
command_type: RendererCommandType::SrvrRndrSetActive,
payload: serde_json::json!({ "active": false }),
};
assert!(is_session_established_command(&active));
assert!(!is_session_established_command(&inactive));
}
#[test]
fn diagnostics_redact_both_delegated_secrets() {
const JWT: &str = "header.payload.secret-signature";
const ENDPOINT: &str = "wss://qws.example.test/ws?route=secret";
let message = format!("rejected {JWT} at {ENDPOINT}");
let redacted = redact_sensitive(message, Some(JWT), Some(ENDPOINT));
assert_eq!(redacted, "rejected [REDACTED] at [REDACTED]");
}
#[test]
fn qcloud_frame_rejects_truncated_payload() {
let payload = vec![0x0a, 0x0b, 0x0c];