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`

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<synctv_proto::client::CreateRoomPublishKeyResponse, ApiError> {
-        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(""),

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(),

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<Option<PublishKeyOptions>, 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<PublishKeyOptions>,
 ) -> Result<CreateRoomPublishKeyResponse, ApiError> {
-    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<CreateRoomPublishKeyResponse, ApiError> {
         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(

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<CreateRoomPublishKeyRequest>,
         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<AppState>,
     Path(path): Path<RoomStreamPath>,
+    req: Option<Json<CreateRoomPublishKeyRequest>>,
 ) -> AppResult<Json<CreateRoomPublishKeyResponse>> {
     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,

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()?;

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<T>(&self, token: &str) -> Result<T>
+    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)]

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::<NonExpiringClaims>(&token).is_err());
+    assert!(jwt
+        .verify_custom_with_optional_exp::<NonExpiringClaims>(&token)
+        .is_ok());
+}
+
 #[test]
 fn test_custom_token_wrong_secret_rejected() {
     #[derive(Debug, serde::Serialize, serde::Deserialize)]

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;

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<i64>,
+    /// 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<i64>,
+}
+
+impl PublishKeyOptions {
+    fn validate(self, now: i64) -> Result<Self> {
+        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<i64>,
     /// 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<u64> {
-    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<dyn Clock>,
+    token_ttl_hours: i64,
+    jti_store: Arc<dyn JtiStore>,
 }
 
 fn token_lifetime_secs(token_ttl_hours: i64) -> Result<i64> {
@@ -82,17 +127,19 @@ fn token_lifetime_secs(token_ttl_hours: i64) -> Result<i64> {
         .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<dyn Clock>,
+fn default_publish_key_options(
+    clock: &dyn Clock,
     token_ttl_hours: i64,
-    jti_store: Arc<dyn JtiStore>,
+) -> Result<PublishKeyOptions> {
+    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<PublishKey>;
 
+    fn generate_publish_key_with_options(
+        &self,
+        room_id: &RoomId,
+        media_id: &MediaId,
+        user_id: &UserId,
+        options: PublishKeyOptions,
+    ) -> Result<PublishKey>;
+
     async fn validate_publish_key(&self, token: &str) -> Result<PublishClaims>;
 
     async fn validate_publish_key_for_stream_claims(
@@ -138,13 +193,23 @@ impl PublishKeyService {
     fn decode_publish_claims(&self, token: &str) -> Result<PublishClaims> {
         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<dyn Clock>,
         token_ttl_hours: i64,
     ) -> Result<Self> {
-        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<dyn Clock>) -> Result<Self> {
         Self::new(jwt_service, clock, 24)
     }
@@ -233,12 +302,7 @@ impl PublishKeyService {
         redis_runtime: Arc<dyn RedisConnectionRuntime>,
         key_prefix: String,
     ) -> Result<Self> {
-        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<dyn RedisConnectionRuntime>,
         key_prefix: String,
     ) -> Result<Self> {
-        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<PublishKey> {
+        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<PublishKey> {
         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<PublishKey> {
+        PublishKeyService::generate_publish_key_with_options(
+            self, room_id, media_id, user_id, options,
+        )
+    }
+
     async fn validate_publish_key(&self, token: &str) -> Result<PublishClaims> {
         PublishKeyService::validate_publish_key(self, token).await
     }

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<dyn RedisConnectionRuntime>,
     key_builder: KeyBuilder,
-    local_cache: moka::future::Cache<String, ()>,
+    local_cache: moka::future::Cache<String, u64>,
     fail_closed: bool,
 }
 
+struct JtiExpiry;
+
+impl moka::Expiry<String, u64> for JtiExpiry {
+    fn expire_after_create(
+        &self,
+        _key: &String,
+        ttl_secs: &u64,
+        _created_at: std::time::Instant,
+    ) -> Option<Duration> {
+        Some(Duration::from_secs(*ttl_secs))
+    }
+}
+
 impl RedisJtiStore {
     #[must_use]
     pub fn from_runtime(
         redis_runtime: Arc<dyn RedisConnectionRuntime>,
         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<dyn RedisConnectionRuntime>,
         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<String, ()>,
+    cache: moka::future::Cache<String, u64>,
 }
 
 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<bool> {
+    async fn try_claim(&self, jti: &str, ttl_secs: u64) -> Result<bool> {
         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;

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),

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 {

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,

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,

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 {

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<i64>,
+}
+
+#[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)]

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)

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<String>,
+    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(),
         }
     }
 }

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:?}"),
     }

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),

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",
     )