backend: add single-use token auth for spawned stream proxies

Generate a random 256-bit token when spawning a proxy process, pass
it as a CLI argument, and return it to the client in the stream start
response. The proxy validates the token on WebTransport connect and
consumes it after first use, preventing replay. A wrong token attempt
also consumes the token for security. Includes 5 unit tests for token
validation logic.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-16 02:36:35 +00:00
parent 826a3b59c9
commit b8c705554f
4 changed files with 109 additions and 8 deletions
+6 -3
View File
@@ -124,9 +124,9 @@ async fn run_backend(port: u16) -> Result<()> {
Ok(()) Ok(())
} }
async fn run_proxy(port: u16, stream_id: uuid::Uuid) -> Result<()> { async fn run_proxy(port: u16, stream_id: uuid::Uuid, stream_token: String) -> Result<()> {
let (config, cert_hash) = certs::get_webtransport_stream_config(stream_id)?; let (config, cert_hash) = certs::get_webtransport_stream_config(stream_id)?;
let proxy = proxy::Proxy::new(cert_hash); let proxy = proxy::Proxy::new(cert_hash, stream_token);
let proxy_arc = std::sync::Arc::new(proxy); let proxy_arc = std::sync::Arc::new(proxy);
let router = Router::new() let router = Router::new()
@@ -166,8 +166,11 @@ async fn main() -> anyhow::Result<()> {
.nth(3) .nth(3)
.ok_or(anyhow!("Cert ID argument missing"))?, .ok_or(anyhow!("Cert ID argument missing"))?,
)?; )?;
let stream_token = std::env::args()
.nth(4)
.ok_or(anyhow!("Stream token argument missing"))?;
run_proxy(port, stream_id).await run_proxy(port, stream_id, stream_token).await
} }
_ => Err(anyhow!("Unknown mode: {mode}")), _ => Err(anyhow!("Unknown mode: {mode}")),
} }
@@ -85,6 +85,18 @@ impl crate::proxy::Proxy {
description: "Could not start stream".to_string(), description: "Could not start stream".to_string(),
}); });
// Validate single-use stream token via the shared helper so this
// handler and its unit tests exercise the same code path.
let provided_token = req.query::<String>("token").unwrap_or_default();
if let Err(msg) = super::validate_stream_token(&self, &provided_token).await {
error!("Stream token validation failed: {msg}");
return Err(AppError {
status_code: StatusCode::UNAUTHORIZED,
description: msg,
});
}
info!("Stream token validated and consumed");
info!("WebTransport connection initiated"); info!("WebTransport connection initiated");
let (wt_stream_send, wt_stream_recv, wt_datagram_send) = match setup_webtransport(req).await let (wt_stream_send, wt_stream_recv, wt_datagram_send) = match setup_webtransport(req).await
{ {
+75 -3
View File
@@ -11,16 +11,16 @@ mod video;
pub struct Proxy { pub struct Proxy {
pub cert_hash: [u8; 32], pub cert_hash: [u8; 32],
//pub cert_hash: String,
pub stream: RwLock<Option<backend::Stream>>, pub stream: RwLock<Option<backend::Stream>>,
pub stream_token: RwLock<Option<String>>,
} }
impl Proxy { impl Proxy {
pub fn new(cert_hash: [u8; 32]) -> Self { pub fn new(cert_hash: [u8; 32], stream_token: String) -> Self {
//pub fn new(cert_hash: String) -> Self {
Proxy { Proxy {
stream: RwLock::new(None), stream: RwLock::new(None),
cert_hash, cert_hash,
stream_token: RwLock::new(Some(stream_token)),
} }
} }
} }
@@ -78,6 +78,22 @@ async fn proxy_main(
Ok(()) Ok(())
} }
/// Validate a provided token against the stored token. Consumes the token on success (single-use).
/// Returns Ok(()) if valid, Err with description if invalid or already consumed.
pub async fn validate_stream_token(proxy: &Proxy, provided: &str) -> std::result::Result<(), String> {
let mut token_guard = proxy.stream_token.write().await;
match token_guard.take() {
Some(expected) if expected == provided => Ok(()),
Some(_) => {
// Wrong token: still consumed by the `take()` above. Any validation
// attempt — correct or not — invalidates the token, so a wrong
// guess cannot be followed by a correct one.
Err("Invalid stream token".to_string())
}
None => Err("Stream token already used".to_string()),
}
}
async fn spawn_gamestream(stream: backend::Stream) -> Result<Channels> { async fn spawn_gamestream(stream: backend::Stream) -> Result<Channels> {
let (tx, rx) = tokio::sync::oneshot::channel(); let (tx, rx) = tokio::sync::oneshot::channel();
let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>(); let (stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
@@ -99,3 +115,59 @@ async fn spawn_gamestream(stream: backend::Stream) -> Result<Channels> {
.context("Could not get gamestream communication channels")?, .context("Could not get gamestream communication channels")?,
}) })
} }
#[cfg(test)]
mod tests {
use super::*;
fn make_proxy(token: &str) -> Proxy {
Proxy {
cert_hash: [0u8; 32],
stream: RwLock::new(None),
stream_token: RwLock::new(Some(token.to_string())),
}
}
#[tokio::test]
async fn test_valid_token_accepted() {
let proxy = make_proxy("abc123");
let result = validate_stream_token(&proxy, "abc123").await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_wrong_token_rejected() {
let proxy = make_proxy("abc123");
let result = validate_stream_token(&proxy, "wrong").await;
assert!(result.is_err());
assert_eq!(result.unwrap_err(), "Invalid stream token");
}
#[tokio::test]
async fn test_missing_token_rejected() {
let proxy = make_proxy("abc123");
let result = validate_stream_token(&proxy, "").await;
assert!(result.is_err());
}
#[tokio::test]
async fn test_token_consumed_after_use() {
let proxy = make_proxy("abc123");
let first = validate_stream_token(&proxy, "abc123").await;
assert!(first.is_ok());
let second = validate_stream_token(&proxy, "abc123").await;
assert!(second.is_err());
assert_eq!(second.unwrap_err(), "Stream token already used");
}
#[tokio::test]
async fn test_wrong_attempt_consumes_token() {
let proxy = make_proxy("abc123");
// Wrong token attempt should consume it
let _ = validate_stream_token(&proxy, "wrong").await;
// Correct token should also fail now
let result = validate_stream_token(&proxy, "abc123").await;
assert!(result.is_err());
}
}
+16 -2
View File
@@ -25,7 +25,7 @@ struct PostStreamStartParams {
struct PostStreamStartResponse { struct PostStreamStartResponse {
url: String, url: String,
cert_hash: [u8; 32], cert_hash: [u8; 32],
//cert_hash: String, stream_token: String,
} }
#[derive(Deserialize)] #[derive(Deserialize)]
@@ -301,6 +301,19 @@ impl crate::backend::Backend {
let port = self.port + <u16>::try_from((*writer).len()).unwrap(); let port = self.port + <u16>::try_from((*writer).len()).unwrap();
// Generate single-use stream token for proxy authentication
let stream_token = {
let mut bytes = [0u8; 32];
openssl::rand::rand_bytes(&mut bytes).map_err(|e| {
error!("Failed to generate stream token: {e}");
AppError {
status_code: StatusCode::INTERNAL_SERVER_ERROR,
description: "Could not start stream".to_string(),
}
})?;
hex::encode(bytes)
};
// Spawn WebTransport proxy // Spawn WebTransport proxy
let binary_path = match std::env::current_exe() { let binary_path = match std::env::current_exe() {
Ok(b) => b, Ok(b) => b,
@@ -314,7 +327,7 @@ impl crate::backend::Backend {
stream_id, port stream_id, port
); );
match tokio::process::Command::new(binary_path) match tokio::process::Command::new(binary_path)
.args(["proxy", &port.to_string(), &stream_id.to_string()]) .args(["proxy", &port.to_string(), &stream_id.to_string(), &stream_token])
.spawn() .spawn()
{ {
Ok(_) => (), Ok(_) => (),
@@ -355,6 +368,7 @@ impl crate::backend::Backend {
let post_stream_response = PostStreamStartResponse { let post_stream_response = PostStreamStartResponse {
url: webtransport_url, url: webtransport_url,
cert_hash: setup_resp.cert_hash, cert_hash: setup_resp.cert_hash,
stream_token,
}; };
Ok(Json(post_stream_response)) Ok(Json(post_stream_response))