From 4a5f7630ad25ce61dcd91ead9994d7e32ffe51e0 Mon Sep 17 00:00:00 2001 From: zijiren <84728412+zijiren233@users.noreply.github.com> Date: Tue, 18 Aug 2026 00:16:00 +0800 Subject: [PATCH] feat: add publish key lifecycles (#418) ## Summary - add single-use, reusable expiring, and permanent RTMP publish keys - validate lifecycle and expiration combinations in the core service - expose lifecycle fields through HTTP, management gRPC, and CLI - keep JTI consumption limited to single-use keys ## Validation - `cargo fmt --all -- --check` - `make clippy-check` - `make check-all-targets` - `cargo test -p synctv-core service::publish_key::tests` --- .../src/impls/admin/livestream.rs | 10 +- synctv-api-common/src/impls/admin/tests.rs | 6 +- synctv-api-common/src/impls/client/stream.rs | 79 +++++++- synctv-api-http/src/http/room/streams.rs | 7 +- synctv-api-http/src/openapi.rs | 19 ++ synctv-core/src/service/auth/jwt.rs | 27 +++ synctv-core/src/service/auth/jwt/tests.rs | 21 ++ synctv-core/src/service/mod.rs | 4 +- synctv-core/src/service/publish_key.rs | 184 +++++++++++++----- synctv-core/src/service/publish_key/jti.rs | 39 ++-- synctv-core/src/service/publish_key/tests.rs | 155 ++++++++++++++- synctv-management/proto/management.proto | 2 + synctv-management/src/admin_runtime.rs | 2 +- synctv-management/src/service.rs | 7 +- synctv-proto/proto/client.proto | 14 +- synctv/src/cli/commands/room/stream.rs | 26 ++- synctv/src/cli/execute/room.rs | 2 + synctv/src/cli/human_output.rs | 12 +- synctv/src/cli/tests.rs | 3 + synctv/src/management_runtime/admin.rs | 4 +- synctv/tests/full_stack_e2e_tests.rs | 8 + 21 files changed, 541 insertions(+), 90 deletions(-) diff --git a/synctv-api-common/src/impls/admin/livestream.rs b/synctv-api-common/src/impls/admin/livestream.rs index ea5f2100..83597815 100644 --- a/synctv-api-common/src/impls/admin/livestream.rs +++ b/synctv-api-common/src/impls/admin/livestream.rs @@ -209,15 +209,14 @@ impl AdminApiImpl { pub async fn create_publish_key_for_actor( &self, room_id: &str, - media_id: &str, + request: synctv_proto::client::CreateRoomPublishKeyRequest, actor_user_id: &UserId, admin_user_id: &UserId, ctx: &RequestContext, ) -> Result { - let request = synctv_proto::client::CreateRoomPublishKeyRequest { - media_id: media_id.to_string(), - }; crate::impls::validate_proto_request(&request)?; + let options = crate::impls::client::stream::publish_key_options(&request)?; + let public_media_id = request.media_id.clone(); let room_id_value = crate::impls::parse_room_id_param(room_id, "room_id", &self.public_id_codec)?; let media_id_value = @@ -247,11 +246,12 @@ impl AdminApiImpl { room_id_value, media_id_value, actor_user_id, + options, )?; tracing::info!( room_id, - media_id, + media_id = public_media_id, actor_user_id = %actor_user_id, admin_user_id = %admin_user_id, ip_address = ctx.ip_address.as_deref().unwrap_or(""), diff --git a/synctv-api-common/src/impls/admin/tests.rs b/synctv-api-common/src/impls/admin/tests.rs index e7c015d4..351fa7da 100644 --- a/synctv-api-common/src/impls/admin/tests.rs +++ b/synctv-api-common/src/impls/admin/tests.rs @@ -4791,7 +4791,11 @@ async fn test_create_publish_key_bypasses_room_membership_requirement_for_global admin_api .create_publish_key_for_actor( &public_room_id, - &public_media_id, + synctv_proto::client::CreateRoomPublishKeyRequest { + media_id: public_media_id.clone(), + r#type: synctv_proto::client::PublishKeyType::SingleUse as i32, + expires_at: Some(chrono::Utc::now().timestamp() + 3600), + }, &owner.id, &global_admin.id, &RequestContext::default(), diff --git a/synctv-api-common/src/impls/client/stream.rs b/synctv-api-common/src/impls/client/stream.rs index 037687d7..8b06ba31 100644 --- a/synctv-api-common/src/impls/client/stream.rs +++ b/synctv-api-common/src/impls/client/stream.rs @@ -1,10 +1,13 @@ -use synctv_core::models::{MediaId, Room, RoomId, UserId}; +use synctv_core::{ + models::{MediaId, Room, RoomId, UserId}, + service::{PublishKeyOptions, PublishKeyType as CorePublishKeyType}, +}; use crate::impls::{ApiError, ClientApiImpl}; use synctv_proto::client::{ CreateRoomPublishKeyRequest, CreateRoomPublishKeyResponse, GetRoomStreamInfoRequest, GetRoomStreamInfoResponse, KickRoomStreamRequest, ListRoomStreamsRequest, - ListRoomStreamsResponse, RoomStreamPublisherInfo, SortDirection, StreamEntry, + ListRoomStreamsResponse, PublishKeyType, RoomStreamPublisherInfo, SortDirection, StreamEntry, }; const LIVESTREAM_UNAVAILABLE_MESSAGE: &str = "Live streaming is not available on this server."; @@ -111,6 +114,29 @@ fn build_publish_rtmp_url(runtime_settings: &crate::ApiRuntimeSettings, room_id: format!("rtmp://{rtmp_host}:{rtmp_port}/{room_id}") } +pub(crate) fn publish_key_options( + req: &CreateRoomPublishKeyRequest, +) -> Result, ApiError> { + let key_type = match PublishKeyType::try_from(req.r#type) + .map_err(|_| ApiError::InvalidInput("publish key type is invalid".to_string()))? + { + PublishKeyType::SingleUse => CorePublishKeyType::SingleUse, + PublishKeyType::Expiring => CorePublishKeyType::Expiring, + PublishKeyType::Permanent => CorePublishKeyType::Permanent, + PublishKeyType::Unspecified if req.expires_at.is_none() => return Ok(None), + PublishKeyType::Unspecified => { + return Err(ApiError::InvalidInput( + "publish key type is required when expiration is provided".to_string(), + )); + } + }; + + Ok(Some(PublishKeyOptions { + key_type, + expires_at: req.expires_at, + })) +} + pub(crate) fn issue_room_publish_key( publish_key_service: &dyn synctv_core::service::StreamingPublishKeyService, runtime_settings: &crate::ApiRuntimeSettings, @@ -118,10 +144,18 @@ pub(crate) fn issue_room_publish_key( room_id: RoomId, media_id: MediaId, actor_user_id: &UserId, + options: Option, ) -> Result { - let publish_key = publish_key_service - .generate_publish_key(&room_id, &media_id, actor_user_id) - .map_err(|error| ApiError::Internal(format!("Failed to generate publish key: {error}")))?; + let publish_key = match options { + Some(options) => publish_key_service.generate_publish_key_with_options( + &room_id, + &media_id, + actor_user_id, + options, + ), + None => publish_key_service.generate_publish_key(&room_id, &media_id, actor_user_id), + } + .map_err(|error| ApiError::InvalidInput(error.to_string()))?; let room_id = public_id_codec .encode_room_id(room_id) .map_err(|error| ApiError::Internal(format!("Failed to encode room id: {error}")))?; @@ -135,6 +169,11 @@ pub(crate) fn issue_room_publish_key( rtmp_url: build_publish_rtmp_url(runtime_settings, &room_id), stream_key, expires_at: publish_key.expires_at, + r#type: match publish_key.key_type { + CorePublishKeyType::SingleUse => PublishKeyType::SingleUse as i32, + CorePublishKeyType::Expiring => PublishKeyType::Expiring as i32, + CorePublishKeyType::Permanent => PublishKeyType::Permanent as i32, + }, }) } @@ -177,6 +216,7 @@ impl ClientApiImpl { req: CreateRoomPublishKeyRequest, ) -> Result { crate::impls::validate_proto_request(&req)?; + let options = publish_key_options(&req)?; let uid = *user_id; let rid = self.parse_room_id(room_id)?; let media_id = self @@ -222,6 +262,7 @@ impl ClientApiImpl { rid, media_id, &uid, + options, ) } @@ -337,6 +378,7 @@ impl ClientApiImpl { mod tests { use super::{ build_room_streams_request, build_room_streams_response, ensure_room_accepts_live_publish, + publish_key_options, }; use crate::impls::ApiError; @@ -381,6 +423,33 @@ mod tests { Ok(()) } + #[test] + fn publish_key_options_preserves_legacy_default() -> TestResult { + let options = api_ok(publish_key_options( + &synctv_proto::client::CreateRoomPublishKeyRequest { + media_id: "med_AbC123".to_string(), + ..Default::default() + }, + ))?; + + assert!(options.is_none()); + Ok(()) + } + + #[test] + fn publish_key_options_rejects_ambiguous_legacy_expiration() -> TestResult { + let error = api_err(publish_key_options( + &synctv_proto::client::CreateRoomPublishKeyRequest { + media_id: "med_AbC123".to_string(), + expires_at: Some(1_800_000_000), + ..Default::default() + }, + ))?; + + assert!(error.is_invalid_argument(), "{error:?}"); + Ok(()) + } + #[test] fn build_room_streams_request_normalizes_defaults() -> TestResult { let req = api_ok(build_room_streams_request( diff --git a/synctv-api-http/src/http/room/streams.rs b/synctv-api-http/src/http/room/streams.rs index 3df06385..541bf61c 100644 --- a/synctv-api-http/src/http/room/streams.rs +++ b/synctv-api-http/src/http/room/streams.rs @@ -64,6 +64,7 @@ pub async fn list_room_streams( post, path = "/api/playback-providers/{roomId}/rtmp/{mediaId}/publish-key", tag = "RTMP Playback Provider", + request_body = Option, params( ("roomId" = String, Path, description = "Room ID"), ("mediaId" = String, Path, description = "Media ID") @@ -84,11 +85,11 @@ pub async fn create_room_publish_key( request_meta: RequestMetadata, State(state): State, Path(path): Path, + req: Option>, ) -> AppResult> { let room_id = path.room_id; - let req = CreateRoomPublishKeyRequest { - media_id: path.media_id, - }; + let mut req = req.map(|Json(req)| req).unwrap_or_default(); + req.media_id = path.media_id; let response = execute_user_endpoint( &state, request_meta, diff --git a/synctv-api-http/src/openapi.rs b/synctv-api-http/src/openapi.rs index 0a60f55b..9321c67d 100644 --- a/synctv-api-http/src/openapi.rs +++ b/synctv-api-http/src/openapi.rs @@ -496,6 +496,7 @@ pub struct GoogleRpcStatusSchema { client::RefreshTokenRequest, client::RefreshTokenResponse, client::LogoutResponse, + client::CreateRoomPublishKeyRequest, client::CreateRoomPublishKeyResponse, client::RequestPasswordResetRequest, client::RequestPasswordResetResponse, @@ -1978,6 +1979,24 @@ mod tests { Ok(()) } + #[test] + fn openapi_marks_publish_key_body_optional() -> TestResult { + let doc = openapi_json()?; + + let request_body = &doc["paths"] + ["/api/playback-providers/{roomId}/rtmp/{mediaId}/publish-key"]["post"]["requestBody"]; + assert!( + request_body.is_object(), + "publish-key creation should document its request body schema" + ); + assert_ne!( + request_body["required"].as_bool(), + Some(true), + "legacy publish-key creation may omit the request body" + ); + Ok(()) + } + #[test] fn openapi_documents_admin_user_preferences_routes() -> TestResult { let doc = openapi_json()?; diff --git a/synctv-core/src/service/auth/jwt.rs b/synctv-core/src/service/auth/jwt.rs index c2bd8242..a3175574 100644 --- a/synctv-core/src/service/auth/jwt.rs +++ b/synctv-core/src/service/auth/jwt.rs @@ -767,6 +767,33 @@ impl JwtService { Ok(token_data.claims) } + + /// Verify a custom token whose claims may intentionally omit expiration. + /// + /// This is reserved for credential types with their own lifecycle checks, + /// such as permanent publish keys. General custom tokens must use + /// [`Self::verify_custom`] so an expiration claim remains required. + pub fn verify_custom_with_optional_exp(&self, token: &str) -> Result + where + T: DeserializeOwned + Serialize, + { + let mut validation = Validation::new(self.algorithm); + validation.validate_exp = false; + validation.validate_nbf = false; + validation.required_spec_claims.remove("exp"); + + let token_data = decode(token, &self.decoding_key, &validation) + .map_err(|e| map_jwt_error(&e, "Token"))?; + + validate_serialized_claims_expiration( + self.clock.as_ref(), + &token_data.claims, + self.clock_skew_leeway_secs, + "Token", + )?; + + Ok(token_data.claims) + } } #[cfg(test)] diff --git a/synctv-core/src/service/auth/jwt/tests.rs b/synctv-core/src/service/auth/jwt/tests.rs index 691e3c3b..0c64a6c0 100644 --- a/synctv-core/src/service/auth/jwt/tests.rs +++ b/synctv-core/src/service/auth/jwt/tests.rs @@ -673,6 +673,27 @@ fn test_sign_and_verify_custom_token() { assert_eq!(verified.exp, claims.exp); } +#[test] +fn test_custom_token_requires_expiration_by_default() { + #[derive(Debug, serde::Serialize, serde::Deserialize)] + struct NonExpiringClaims { + sub: String, + } + + let jwt = create_jwt_service(); + let token = ok( + jwt.sign_custom(&NonExpiringClaims { + sub: "custom_subject".to_string(), + }), + "non-expiring custom token should sign", + ); + + assert!(jwt.verify_custom::(&token).is_err()); + assert!(jwt + .verify_custom_with_optional_exp::(&token) + .is_ok()); +} + #[test] fn test_custom_token_wrong_secret_rejected() { #[derive(Debug, serde::Serialize, serde::Deserialize)] diff --git a/synctv-core/src/service/mod.rs b/synctv-core/src/service/mod.rs index 15370213..9ed3aec6 100644 --- a/synctv-core/src/service/mod.rs +++ b/synctv-core/src/service/mod.rs @@ -181,8 +181,8 @@ pub use presence::{ }; pub use providers_manager::{LocalProviderHttpOptions, MediaProvidersOptions, ProvidersManager}; pub use publish_key::{ - InMemoryJtiStore, JtiStore, PublishClaims, PublishKey, PublishKeyService, RedisJtiStore, - StreamingPublishKeyService, + InMemoryJtiStore, JtiStore, PublishClaims, PublishKey, PublishKeyOptions, PublishKeyService, + PublishKeyType, RedisJtiStore, StreamingPublishKeyService, }; pub use rate_limit::{RateLimitConfig, RateLimitError, RateLimiter, RequestRateLimiterService}; pub use realtime_outbox::RealtimeOutboxService; diff --git a/synctv-core/src/service/publish_key.rs b/synctv-core/src/service/publish_key.rs index 6935f0bd..47ffe264 100644 --- a/synctv-core/src/service/publish_key.rs +++ b/synctv-core/src/service/publish_key.rs @@ -1,7 +1,7 @@ //! Publish key generation for RTMP live streaming //! //! Generates JWT tokens for RTMP push authentication. -//! Includes single-use enforcement to prevent TOCTOU races. +//! Supports single-use, reusable expiring, and permanent publish keys. //! //! ## JTI Deduplication Backends //! @@ -32,8 +32,54 @@ pub struct PublishKey { pub media_id: String, /// User ID who requested the key pub user_id: String, - /// Expiration timestamp - pub expires_at: i64, + /// Expiration timestamp, absent for permanent keys + pub expires_at: Option, + /// Key lifecycle type + pub key_type: PublishKeyType, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PublishKeyType { + #[default] + SingleUse, + Expiring, + Permanent, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PublishKeyOptions { + pub key_type: PublishKeyType, + pub expires_at: Option, +} + +impl PublishKeyOptions { + fn validate(self, now: i64) -> Result { + match (self.key_type, self.expires_at) { + (PublishKeyType::SingleUse | PublishKeyType::Expiring, Some(expires_at)) + if expires_at > now => + { + Ok(self) + } + (PublishKeyType::SingleUse | PublishKeyType::Expiring, Some(_)) => Err( + Error::InvalidInput("publish key expiration must be in the future".to_string()), + ), + (PublishKeyType::SingleUse | PublishKeyType::Expiring, None) => { + Err(Error::InvalidInput(format!( + "{} publish keys require an expiration time", + match self.key_type { + PublishKeyType::SingleUse => "single-use", + PublishKeyType::Expiring => "expiring", + PublishKeyType::Permanent => unreachable!(), + } + ))) + } + (PublishKeyType::Permanent, None) => Ok(self), + (PublishKeyType::Permanent, Some(_)) => Err(Error::InvalidInput( + "permanent publish keys must not have an expiration time".to_string(), + )), + } + } } /// Claims for RTMP publish token @@ -49,26 +95,25 @@ pub struct PublishClaims { pub perm_manage_live_streams: bool, /// Issued at timestamp pub iat: i64, - /// Expiration timestamp - pub exp: i64, + /// Expiration timestamp, absent for permanent keys + #[serde(skip_serializing_if = "Option::is_none")] + pub exp: Option, /// JWT ID (unique token identifier) pub jti: String, + /// Key lifecycle type + #[serde(default)] + pub key_type: PublishKeyType, } -fn cache_ttl_secs(token_ttl_hours: i64) -> Result { - if token_ttl_hours <= 0 { - tracing::warn!( - token_ttl_hours, - "Publish key token TTL must be positive; using deduplication grace window only" - ); - return Ok(300); - } - let hours = u64::try_from(token_ttl_hours) - .map_err(|_| Error::InvalidInput("publish key token TTL is invalid".to_string()))?; - hours - .checked_mul(3600) - .and_then(|seconds| seconds.checked_add(300)) - .ok_or_else(|| Error::InvalidInput("publish key cache TTL is too large".to_string())) +/// Publish key service for generating RTMP streaming tokens. +/// +/// Single-use keys use a pluggable `JtiStore` backend for deduplication. +#[derive(Clone)] +pub struct PublishKeyService { + jwt_service: JwtService, + clock: Arc, + token_ttl_hours: i64, + jti_store: Arc, } fn token_lifetime_secs(token_ttl_hours: i64) -> Result { @@ -82,17 +127,19 @@ fn token_lifetime_secs(token_ttl_hours: i64) -> Result { .ok_or_else(|| Error::InvalidInput("publish key token TTL is too large".to_string())) } -/// Publish key service for generating RTMP streaming tokens. -/// -/// Includes single-use enforcement: each publish key `jti` can only be -/// consumed once by `validate_publish_key`. Uses a pluggable `JtiStore` -/// backend for deduplication. -#[derive(Clone)] -pub struct PublishKeyService { - jwt_service: JwtService, - clock: Arc, +fn default_publish_key_options( + clock: &dyn Clock, token_ttl_hours: i64, - jti_store: Arc, +) -> Result { + let expires_at = clock + .now() + .timestamp() + .checked_add(token_lifetime_secs(token_ttl_hours)?) + .ok_or_else(|| Error::InvalidInput("publish key expiration overflow".to_string()))?; + Ok(PublishKeyOptions { + key_type: PublishKeyType::SingleUse, + expires_at: Some(expires_at), + }) } #[async_trait] @@ -104,6 +151,14 @@ pub trait StreamingPublishKeyService: Send + Sync { user_id: &UserId, ) -> Result; + fn generate_publish_key_with_options( + &self, + room_id: &RoomId, + media_id: &MediaId, + user_id: &UserId, + options: PublishKeyOptions, + ) -> Result; + async fn validate_publish_key(&self, token: &str) -> Result; async fn validate_publish_key_for_stream_claims( @@ -138,13 +193,23 @@ impl PublishKeyService { fn decode_publish_claims(&self, token: &str) -> Result { let claims: PublishClaims = self .jwt_service - .verify_custom(token) + .verify_custom_with_optional_exp(token) .map_err(|e| Error::Authentication(format!("Invalid token format: {e}")))?; let now = self.clock.now().timestamp(); - if now > claims.exp { - return Err(Error::Authentication("Token has expired".to_string())); + match (claims.key_type, claims.exp) { + (PublishKeyType::SingleUse | PublishKeyType::Expiring, Some(expires_at)) => { + if now > expires_at { + return Err(Error::Authentication("Token has expired".to_string())); + } + } + (PublishKeyType::Permanent, None) => {} + _ => { + return Err(Error::Authentication( + "Publish key lifecycle claims are invalid".to_string(), + )); + } } if !claims.perm_manage_live_streams { @@ -157,7 +222,13 @@ impl PublishKeyService { } async fn claim_publish_key(&self, claims: &PublishClaims) -> Result<()> { - let ttl_secs = (claims.exp - claims.iat) + if claims.key_type != PublishKeyType::SingleUse { + return Ok(()); + } + let expires_at = claims.exp.ok_or_else(|| { + Error::Authentication("Single-use publish key has no expiration".to_string()) + })?; + let ttl_secs = (expires_at - self.clock.now().timestamp()) .max(0) .cast_unsigned() .saturating_add(300); @@ -216,12 +287,10 @@ impl PublishKeyService { clock: Arc, token_ttl_hours: i64, ) -> Result { - let cache_ttl_secs = cache_ttl_secs(token_ttl_hours)?; - let store = Arc::new(InMemoryJtiStore::new(cache_ttl_secs)); + let store = Arc::new(InMemoryJtiStore::new(0)); Ok(Self::from_store(jwt_service, clock, token_ttl_hours, store)) } - /// Create a new publish key service with default TTL (24 hours) pub fn with_default_ttl(jwt_service: JwtService, clock: Arc) -> Result { Self::new(jwt_service, clock, 24) } @@ -233,12 +302,7 @@ impl PublishKeyService { redis_runtime: Arc, key_prefix: String, ) -> Result { - let cache_ttl_secs = cache_ttl_secs(token_ttl_hours)?; - let store = Arc::new(RedisJtiStore::from_runtime( - redis_runtime, - key_prefix, - cache_ttl_secs, - )); + let store = Arc::new(RedisJtiStore::from_runtime(redis_runtime, key_prefix, 0)); Ok(Self::from_store(jwt_service, clock, token_ttl_hours, store)) } @@ -249,11 +313,10 @@ impl PublishKeyService { redis_runtime: Arc, key_prefix: String, ) -> Result { - let cache_ttl_secs = cache_ttl_secs(token_ttl_hours)?; let store = Arc::new(RedisJtiStore::from_runtime_fail_closed( redis_runtime, key_prefix, - cache_ttl_secs, + 0, )); Ok(Self::from_store(jwt_service, clock, token_ttl_hours, store)) } @@ -289,11 +352,20 @@ impl PublishKeyService { room_id: &RoomId, media_id: &MediaId, user_id: &UserId, + ) -> Result { + let options = default_publish_key_options(self.clock.as_ref(), self.token_ttl_hours)?; + self.generate_publish_key_with_options(room_id, media_id, user_id, options) + } + + pub fn generate_publish_key_with_options( + &self, + room_id: &RoomId, + media_id: &MediaId, + user_id: &UserId, + options: PublishKeyOptions, ) -> Result { let now = self.clock.now().timestamp(); - let exp = now - .checked_add(token_lifetime_secs(self.token_ttl_hours)?) - .ok_or_else(|| Error::InvalidInput("publish key expiration overflow".to_string()))?; + let options = options.validate(now)?; let claims = PublishClaims { room_id: room_id.to_string(), @@ -301,8 +373,9 @@ impl PublishKeyService { user_id: user_id.to_string(), perm_manage_live_streams: true, iat: now, - exp, + exp: options.expires_at, jti: synctv_common::snanoid!(32), + key_type: options.key_type, }; let token = self.jwt_service.sign_custom(&claims)?; @@ -312,7 +385,8 @@ impl PublishKeyService { room_id: room_id.to_string(), media_id: media_id.to_string(), user_id: user_id.to_string(), - expires_at: exp, + expires_at: options.expires_at, + key_type: options.key_type, }) } @@ -400,6 +474,18 @@ impl StreamingPublishKeyService for PublishKeyService { PublishKeyService::generate_publish_key(self, room_id, media_id, user_id) } + fn generate_publish_key_with_options( + &self, + room_id: &RoomId, + media_id: &MediaId, + user_id: &UserId, + options: PublishKeyOptions, + ) -> Result { + PublishKeyService::generate_publish_key_with_options( + self, room_id, media_id, user_id, options, + ) + } + async fn validate_publish_key(&self, token: &str) -> Result { PublishKeyService::validate_publish_key(self, token).await } diff --git a/synctv-core/src/service/publish_key/jti.rs b/synctv-core/src/service/publish_key/jti.rs index 2b77b461..60bd7978 100644 --- a/synctv-core/src/service/publish_key/jti.rs +++ b/synctv-core/src/service/publish_key/jti.rs @@ -36,23 +36,36 @@ pub trait JtiStore: Send + Sync { pub struct RedisJtiStore { pub(super) redis_runtime: Arc, key_builder: KeyBuilder, - local_cache: moka::future::Cache, + local_cache: moka::future::Cache, fail_closed: bool, } +struct JtiExpiry; + +impl moka::Expiry for JtiExpiry { + fn expire_after_create( + &self, + _key: &String, + ttl_secs: &u64, + _created_at: std::time::Instant, + ) -> Option { + Some(Duration::from_secs(*ttl_secs)) + } +} + impl RedisJtiStore { #[must_use] pub fn from_runtime( redis_runtime: Arc, key_prefix: String, - cache_ttl_secs: u64, + _cache_ttl_secs: u64, ) -> Self { Self { redis_runtime, key_builder: KeyBuilder::new(key_prefix), local_cache: moka::future::Cache::builder() .max_capacity(100_000) - .time_to_live(Duration::from_secs(cache_ttl_secs)) + .expire_after(JtiExpiry) .build(), fail_closed: false, } @@ -75,14 +88,14 @@ impl RedisJtiStore { pub fn from_runtime_fail_closed( redis_runtime: Arc, key_prefix: String, - cache_ttl_secs: u64, + _cache_ttl_secs: u64, ) -> Self { Self { redis_runtime, key_builder: KeyBuilder::new(key_prefix), local_cache: moka::future::Cache::builder() .max_capacity(100_000) - .time_to_live(Duration::from_secs(cache_ttl_secs)) + .expire_after(JtiExpiry) .build(), fail_closed: true, } @@ -128,11 +141,11 @@ impl JtiStore for RedisJtiStore { match set_result { Ok(Some(_)) => { - self.local_cache.insert(jti.to_string(), ()).await; + self.local_cache.insert(jti.to_string(), ttl_secs).await; Ok(true) } Ok(None) => { - self.local_cache.insert(jti.to_string(), ()).await; + self.local_cache.insert(jti.to_string(), ttl_secs).await; Ok(false) } Err(error) => { @@ -153,7 +166,7 @@ impl JtiStore for RedisJtiStore { if self.local_cache.contains_key(jti) { Ok(false) } else { - self.local_cache.insert(jti.to_string(), ()).await; + self.local_cache.insert(jti.to_string(), ttl_secs).await; Ok(true) } } @@ -174,16 +187,16 @@ impl JtiStore for RedisJtiStore { } pub struct InMemoryJtiStore { - cache: moka::future::Cache, + cache: moka::future::Cache, } impl InMemoryJtiStore { #[must_use] - pub fn new(cache_ttl_secs: u64) -> Self { + pub fn new(_cache_ttl_secs: u64) -> Self { Self { cache: moka::future::Cache::builder() .max_capacity(100_000) - .time_to_live(Duration::from_secs(cache_ttl_secs)) + .expire_after(JtiExpiry) .build(), } } @@ -191,7 +204,7 @@ impl InMemoryJtiStore { #[async_trait] impl JtiStore for InMemoryJtiStore { - async fn try_claim(&self, jti: &str, _ttl_secs: u64) -> Result { + async fn try_claim(&self, jti: &str, ttl_secs: u64) -> Result { use moka::ops::compute::Op; let entry = self .cache @@ -200,7 +213,7 @@ impl JtiStore for InMemoryJtiStore { if maybe_entry.is_some() { Op::Nop } else { - Op::Put(()) + Op::Put(ttl_secs) } }) .await; diff --git a/synctv-core/src/service/publish_key/tests.rs b/synctv-core/src/service/publish_key/tests.rs index 9556bc75..6ec9de7b 100644 --- a/synctv-core/src/service/publish_key/tests.rs +++ b/synctv-core/src/service/publish_key/tests.rs @@ -214,7 +214,7 @@ async fn test_generate_publish_key_returns_valid_token() { assert_eq!(key.room_id, room_id.to_string()); assert_eq!(key.media_id, media_id.to_string()); assert_eq!(key.user_id, user_id.to_string()); - assert!(key.expires_at > 0); + assert!(key.expires_at.is_some_and(|expires_at| expires_at > 0)); } #[tokio::test] @@ -232,13 +232,159 @@ async fn test_generate_publish_key_expiration_matches_ttl() { let now = test_clock().now().timestamp(); let expected_exp = now + (2 * 3600); - let diff = (key.expires_at - expected_exp).abs(); + let diff = (key.expires_at.expect("default key must expire") - expected_exp).abs(); assert!( diff < 5, "Expiration time is off by more than 5 seconds: diff={diff}" ); } +#[tokio::test] +async fn explicit_publish_key_types_enforce_their_lifecycle() { + let now = 1_700_000_000i64; + let clock = fixed_clock(now * 1000); + let service = ok( + PublishKeyService::new(create_jwt_service_with_clock(clock.clone()), clock, 24), + "publish key service should build", + ); + let room_id = RoomId::new(); + let media_id = MediaId::new(); + let user_id = UserId::new(); + + let single_use = ok( + service.generate_publish_key_with_options( + &room_id, + &media_id, + &user_id, + PublishKeyOptions { + key_type: PublishKeyType::SingleUse, + expires_at: Some(now + 3600), + }, + ), + "single-use key should generate", + ); + assert!(service + .validate_publish_key(&single_use.token) + .await + .is_ok()); + assert!(service + .validate_publish_key(&single_use.token) + .await + .is_err()); + + let expiring = ok( + service.generate_publish_key_with_options( + &room_id, + &media_id, + &user_id, + PublishKeyOptions { + key_type: PublishKeyType::Expiring, + expires_at: Some(now + 3600), + }, + ), + "expiring key should generate", + ); + assert!(service.validate_publish_key(&expiring.token).await.is_ok()); + assert!(service.validate_publish_key(&expiring.token).await.is_ok()); + + let permanent = ok( + service.generate_publish_key_with_options( + &room_id, + &media_id, + &user_id, + PublishKeyOptions { + key_type: PublishKeyType::Permanent, + expires_at: None, + }, + ), + "permanent key should generate", + ); + assert_eq!(permanent.expires_at, None); + assert!(service.validate_publish_key(&permanent.token).await.is_ok()); + assert!(service.validate_publish_key(&permanent.token).await.is_ok()); +} + +#[tokio::test] +async fn legacy_publish_keys_without_key_type_default_to_single_use() { + #[derive(Serialize)] + struct LegacyPublishClaims { + room_id: String, + media_id: String, + user_id: String, + perm_manage_live_streams: bool, + iat: i64, + exp: i64, + jti: String, + } + + let now = 1_700_000_000i64; + let clock = fixed_clock(now * 1000); + let jwt = create_jwt_service_with_clock(clock.clone()); + let service = ok( + PublishKeyService::new(jwt.clone(), clock, 24), + "publish key service should build", + ); + let room_id = RoomId::new(); + let media_id = MediaId::new(); + let user_id = UserId::new(); + let token = ok( + jwt.sign_custom(&LegacyPublishClaims { + room_id: room_id.to_string(), + media_id: media_id.to_string(), + user_id: user_id.to_string(), + perm_manage_live_streams: true, + iat: now, + exp: now + 3600, + jti: "legacy-publish-key".to_string(), + }), + "legacy publish key should sign", + ); + + let claims = ok( + service + .validate_publish_key_for_stream_claims(&token, &room_id, &media_id) + .await, + "legacy publish key should remain valid", + ); + assert_eq!(claims.key_type, PublishKeyType::SingleUse); + assert!(service + .validate_publish_key_for_stream_claims(&token, &room_id, &media_id) + .await + .is_err()); +} + +#[test] +fn publish_key_options_reject_inconsistent_expiration() { + let now = 1_700_000_000i64; + let clock = fixed_clock(now * 1000); + let service = ok( + PublishKeyService::new(create_jwt_service_with_clock(clock.clone()), clock, 24), + "publish key service should build", + ); + let room_id = RoomId::new(); + let media_id = MediaId::new(); + let user_id = UserId::new(); + + for options in [ + PublishKeyOptions { + key_type: PublishKeyType::SingleUse, + expires_at: None, + }, + PublishKeyOptions { + key_type: PublishKeyType::Expiring, + expires_at: Some(now), + }, + PublishKeyOptions { + key_type: PublishKeyType::Permanent, + expires_at: Some(now + 3600), + }, + ] { + assert!(service + .generate_publish_key_with_options(&room_id, &media_id, &user_id, options) + .is_err()); + } +} + #[tokio::test] async fn test_publish_key_uses_configured_clock_for_issue_and_validation() { let now_secs = 2_000_000_000; @@ -262,7 +408,7 @@ async fn test_publish_key_uses_configured_clock_for_issue_and_validation() { ); assert_eq!(claims.iat, now_secs); - assert_eq!(claims.exp, now_secs + 7_200); + assert_eq!(claims.exp, Some(now_secs + 7_200)); assert_eq!(key.expires_at, claims.exp); } @@ -311,8 +457,9 @@ async fn test_validate_publish_key_rejects_expired_token() { user_id: UserId::new().to_string(), perm_manage_live_streams: true, iat: now - 7200, - exp: now - 3600, + exp: Some(now - 3600), jti: "expired_publish_key_test".to_string(), + key_type: PublishKeyType::SingleUse, }; let token = ok( jwt_service.sign_custom(&expired_claims), diff --git a/synctv-management/proto/management.proto b/synctv-management/proto/management.proto index 58304dae..60093154 100644 --- a/synctv-management/proto/management.proto +++ b/synctv-management/proto/management.proto @@ -1052,6 +1052,8 @@ message CreatePublishKeyRequest { UserRef actor = 1; string room_id = 2; string media_id = 3; + synctv.client.PublishKeyType type = 4; + optional int64 expires_at = 5; } message GetStreamInfoRequest { diff --git a/synctv-management/src/admin_runtime.rs b/synctv-management/src/admin_runtime.rs index 075a949a..e6b9593c 100644 --- a/synctv-management/src/admin_runtime.rs +++ b/synctv-management/src/admin_runtime.rs @@ -897,7 +897,7 @@ pub trait AdminRuntime: Send + Sync { async fn create_publish_key_for_actor( &self, room_id: &str, - media_id: &str, + request: client_proto::CreateRoomPublishKeyRequest, actor_user_id: &UserId, admin_user_id: &UserId, ctx: &RequestContext, diff --git a/synctv-management/src/service.rs b/synctv-management/src/service.rs index d412d4bc..989319a9 100644 --- a/synctv-management/src/service.rs +++ b/synctv-management/src/service.rs @@ -3147,11 +3147,16 @@ impl ManagementService for ManagementServiceImpl { let ctx = self.grpc_request_context(&request); let req = request.into_inner(); let actor_user_id = self.resolve_client_actor_user_id(req.actor).await?; + let publish_key_request = client_proto::CreateRoomPublishKeyRequest { + media_id: req.media_id, + r#type: req.r#type, + expires_at: req.expires_at, + }; let response = self .admin_api .create_publish_key_for_actor( &req.room_id, - &req.media_id, + publish_key_request, &actor_user_id, &validated.user_id, &ctx, diff --git a/synctv-proto/proto/client.proto b/synctv-proto/proto/client.proto index f3051128..2daafcd8 100644 --- a/synctv-proto/proto/client.proto +++ b/synctv-proto/proto/client.proto @@ -1752,19 +1752,31 @@ message ListRoomStreamsResponse { int32 total = 2; } +enum PublishKeyType { + PUBLISH_KEY_TYPE_UNSPECIFIED = 0; + PUBLISH_KEY_TYPE_SINGLE_USE = 1; + PUBLISH_KEY_TYPE_EXPIRING = 2; + PUBLISH_KEY_TYPE_PERMANENT = 3; +} + message CreateRoomPublishKeyRequest { string media_id = 1 [(buf.validate.field).string = { min_len: 1 max_len: 64 pattern: "^med_[A-Za-z0-9]+$" }]; + PublishKeyType type = 2 [(buf.validate.field).enum = { + defined_only: true + }]; + optional int64 expires_at = 3; } message CreateRoomPublishKeyResponse { string publish_key = 1; string rtmp_url = 2; string stream_key = 3; - int64 expires_at = 4; + optional int64 expires_at = 4; + PublishKeyType type = 5; } message GetRoomStreamInfoRequest { diff --git a/synctv/src/cli/commands/room/stream.rs b/synctv/src/cli/commands/room/stream.rs index 7faa7b32..b032419f 100644 --- a/synctv/src/cli/commands/room/stream.rs +++ b/synctv/src/cli/commands/room/stream.rs @@ -10,7 +10,7 @@ pub struct RoomStreamCommand { pub enum RoomStreamSubcommand { /// List active RTMP publish sessions in a room List(RoomStreamListArgs), - /// Create a single-use RTMP publish key for a room media item + /// Create an RTMP publish key for a room media item PublishKey(RoomStreamPublishKeyArgs), /// Get the active RTMP stream state for one room media item Get(RoomStreamGetArgs), @@ -61,6 +61,30 @@ pub struct RoomStreamPublishKeyArgs { #[arg(long, allow_hyphen_values = true)] pub media_id: String, + + #[arg(long, value_enum)] + pub key_type: CliPublishKeyType, + + /// Unix timestamp. Required for single-use and expiring keys. + #[arg(long)] + pub expires_at: Option, +} + +#[derive(Debug, Clone, Copy, ValueEnum)] +pub enum CliPublishKeyType { + SingleUse, + Expiring, + Permanent, +} + +impl CliPublishKeyType { + pub const fn to_proto(self) -> i32 { + match self { + Self::SingleUse => synctv_proto::client::PublishKeyType::SingleUse as i32, + Self::Expiring => synctv_proto::client::PublishKeyType::Expiring as i32, + Self::Permanent => synctv_proto::client::PublishKeyType::Permanent as i32, + } + } } #[derive(Debug, Args)] diff --git a/synctv/src/cli/execute/room.rs b/synctv/src/cli/execute/room.rs index 599b2fa5..2de1b8d7 100644 --- a/synctv/src/cli/execute/room.rs +++ b/synctv/src/cli/execute/room.rs @@ -546,6 +546,8 @@ pub(super) async fn execute_room(room_command: RoomCommand) -> Result<()> { actor: Some(args.actor.to_management_proto()?), room_id: args.room.room_id, media_id: args.media_id, + r#type: args.key_type.to_proto(), + expires_at: args.expires_at, } )?; args.room.remote.print_output(&response) diff --git a/synctv/src/cli/human_output.rs b/synctv/src/cli/human_output.rs index fc6ea628..2de4aa22 100644 --- a/synctv/src/cli/human_output.rs +++ b/synctv/src/cli/human_output.rs @@ -397,7 +397,8 @@ pub(in crate::cli) struct HumanCreatePublishKeyResponse { publish_key: String, rtmp_url: String, stream_key: String, - expires_at: String, + expires_at: Option, + key_type: String, } #[derive(Debug, Clone, Serialize)] @@ -1987,7 +1988,14 @@ impl ToHuman for synctv_proto::client::CreateRoomPublishKeyResponse { publish_key: self.publish_key.clone(), rtmp_url: self.rtmp_url.clone(), stream_key: self.stream_key.clone(), - expires_at: humanize_timestamp(self.expires_at), + expires_at: self.expires_at.map(humanize_timestamp), + key_type: match self.r#type() { + synctv_proto::client::PublishKeyType::SingleUse => "single_use", + synctv_proto::client::PublishKeyType::Expiring => "expiring", + synctv_proto::client::PublishKeyType::Permanent => "permanent", + synctv_proto::client::PublishKeyType::Unspecified => "unspecified", + } + .to_string(), } } } diff --git a/synctv/src/cli/tests.rs b/synctv/src/cli/tests.rs index 9c069a59..6b2a09e0 100644 --- a/synctv/src/cli/tests.rs +++ b/synctv/src/cli/tests.rs @@ -3961,6 +3961,8 @@ fn cli_parses_room_stream_publish_key_and_get() { "alice", "--media-id", "media-1", + "--key-type", + "single-use", ]); match cli_publish_key.command { Commands::Room(RoomCommand { @@ -3972,6 +3974,7 @@ fn cli_parses_room_stream_publish_key_and_get() { assert_eq!(args.room.room_id, "room-1"); assert_eq!(args.actor.username.as_deref(), Some("alice")); assert_eq!(args.media_id, "media-1"); + assert!(matches!(args.key_type, CliPublishKeyType::SingleUse)); } other => panic!("unexpected command parsed: {other:?}"), } diff --git a/synctv/src/management_runtime/admin.rs b/synctv/src/management_runtime/admin.rs index b04566c2..dccce4af 100644 --- a/synctv/src/management_runtime/admin.rs +++ b/synctv/src/management_runtime/admin.rs @@ -1141,7 +1141,7 @@ impl AdminRuntime for ManagementAdminRuntime { async fn create_publish_key_for_actor( &self, room_id: &str, - media_id: &str, + request: client_proto::CreateRoomPublishKeyRequest, actor_user_id: &UserId, admin_user_id: &UserId, ctx: &RequestContext, @@ -1149,7 +1149,7 @@ impl AdminRuntime for ManagementAdminRuntime { self.inner .create_publish_key_for_actor( room_id, - media_id, + request, actor_user_id, admin_user_id, &api_request_context(ctx), diff --git a/synctv/tests/full_stack_e2e_tests.rs b/synctv/tests/full_stack_e2e_tests.rs index d5ec23ab..f0a66293 100644 --- a/synctv/tests/full_stack_e2e_tests.rs +++ b/synctv/tests/full_stack_e2e_tests.rs @@ -4668,6 +4668,8 @@ async fn full_stack_cli_stream_commands_cover_publish_list_get_and_kick_with_rea &media_id, "--username", &owner_username, + "--key-type", + "permanent", ], "create rtmp publish key", ) @@ -5054,6 +5056,8 @@ async fn full_stack_public_rtmp_playback_serves_signed_flv_and_hls_over_http() { &media_id, "--username", &owner_username, + "--key-type", + "permanent", ], "create HTTP playback RTMP publish key", ) @@ -5507,6 +5511,8 @@ async fn full_stack_cli_management_actor_state_constraints_reject_invalid_room_o &media_id, "--username", &banned_username, + "--key-type", + "permanent", ], "banned actor publish key", ) @@ -5541,6 +5547,8 @@ async fn full_stack_cli_management_actor_state_constraints_reject_invalid_room_o &media_id, "--username", &owner_username, + "--key-type", + "permanent", ], "media creator publish key in banned room", )