//! `RealtimeAuthorityImpl` — the DB-backed SSE subscribe gate (v1.1.6). //! //! Backs the [`picloud_shared::RealtimeAuthority`] trait the SSE handler //! in orchestrator-core calls. All `topics`-table reads and signing-key //! material stay inside this impl so the data-plane crate never touches //! the key. //! //! Verdict mapping (see [`SubscribeDenied`]): //! * topic missing OR not externally subscribable → `NotFound` (404). //! Both collapse to 404 so the endpoint can't probe internal topics. //! * `auth_mode = 'public'` → allow. //! * `auth_mode = 'token'` → verify the HMAC token (present, signed by //! this app's key, unexpired, scoped to this topic) → allow, else //! `Unauthorized` (401, generic — never says which check failed). //! //! Signing keys never change in v1.1.6 (no rotation API), so a small //! in-memory cache avoids a per-subscribe DB read once an app's key has //! been seen. The cache is purely an optimization — a cold miss reads //! the row. use std::collections::HashMap; use std::sync::{Arc, Mutex}; use async_trait::async_trait; use picloud_shared::{subscriber_token, AppId, RealtimeAuthority, SubscribeDenied, UsersService}; use crate::app_secrets_repo::AppSecretsRepo; use crate::topic_repo::{TopicAuthMode, TopicRepo}; /// F-S-008: TTL on cached signing-key entries. Once key rotation /// lands, every running process keeps accepting tokens signed by the /// old key until restart — bounded eviction is the simplest defence. /// Also bounds memory growth on a many-app install. const KEY_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(300); pub struct RealtimeAuthorityImpl { topics: Arc, secrets: Arc, users: Arc, key_cache: Mutex)>>, } impl RealtimeAuthorityImpl { #[must_use] pub fn new( topics: Arc, secrets: Arc, users: Arc, ) -> Self { Self { topics, secrets, users, key_cache: Mutex::new(HashMap::new()), } } /// Fetch the app's signing key, consulting the cache first. Returns /// `None` when the app has no key (no token ever minted) — which the /// caller maps to `Unauthorized`. async fn signing_key(&self, app_id: AppId) -> Result>, SubscribeDenied> { if let Ok(cache) = self.key_cache.lock() { if let Some((ts, k)) = cache.get(&app_id) { if ts.elapsed() <= KEY_CACHE_TTL { return Ok(Some(k.clone())); } } } let key = self .secrets .signing_key(app_id) .await .map_err(|e| SubscribeDenied::Backend(e.to_string()))?; if let Some(ref k) = key { if let Ok(mut cache) = self.key_cache.lock() { cache.insert(app_id, (std::time::Instant::now(), k.clone())); } } Ok(key) } /// F-S-008 + future key-rotation hook: drop the cached signing key /// for an app. Wired into the app-secrets rotation path so a fresh /// key takes effect on the next request instead of waiting for the /// TTL. pub fn invalidate_signing_key(&self, app_id: AppId) { if let Ok(mut cache) = self.key_cache.lock() { cache.remove(&app_id); } } } #[async_trait] impl RealtimeAuthority for RealtimeAuthorityImpl { async fn authorize_subscribe( &self, app_id: AppId, topic: &str, token: Option<&str>, ) -> Result<(), SubscribeDenied> { let registered = self .topics .get(app_id, topic) .await .map_err(|e| SubscribeDenied::Backend(e.to_string()))?; // Missing topic AND internal-only topic both 404 — don't leak // which internal topics exist. let Some(t) = registered.filter(|t| t.external_subscribable) else { return Err(SubscribeDenied::NotFound); }; match t.auth_mode { TopicAuthMode::Public => Ok(()), TopicAuthMode::Token => { let token = token.ok_or(SubscribeDenied::Unauthorized)?; let key = self .signing_key(app_id) .await? .ok_or(SubscribeDenied::Unauthorized)?; let now = chrono::Utc::now().timestamp(); let claims = subscriber_token::verify(&key, token, now) .map_err(|_| SubscribeDenied::Unauthorized)?; // Per-app key already makes a cross-app token fail the // signature check; this is belt-and-suspenders. if claims.app_id != app_id || !claims.allows_topic(topic) { return Err(SubscribeDenied::Unauthorized); } Ok(()) } TopicAuthMode::Session => { // v1.1.8 F3: per-app user session authorizes the // subscribe. The token comes off the // `Authorization: Bearer ...` header or `?token=...` // query (extraction is on the SSE handler side, // unchanged from the Token branch). UsersService // bumps the sliding TTL on success and rejects // cross-app tokens internally. let token = token.ok_or(SubscribeDenied::Unauthorized)?; let user = self .users .verify_session_for_realtime(app_id, token) .await .map_err(|e| SubscribeDenied::Backend(e.to_string()))? .ok_or(SubscribeDenied::Unauthorized)?; // Defense-in-depth: the service already cross-app // checks, but if the impl drifted we still reject. if user.app_id != app_id { return Err(SubscribeDenied::Unauthorized); } Ok(()) } } } } #[cfg(test)] mod tests { use super::*; use crate::app_secrets_repo::AppSecretsRepoError; use crate::topic_repo::{Topic, TopicRepoError}; use chrono::Utc; use picloud_shared::subscriber_token::{sign, TokenClaims}; struct FakeTopics(Vec<(AppId, Topic)>); #[async_trait] impl TopicRepo for FakeTopics { async fn create( &self, _: AppId, _: &str, _: bool, _: TopicAuthMode, ) -> Result { unimplemented!() } async fn list(&self, _: AppId) -> Result, TopicRepoError> { unimplemented!() } async fn get(&self, app_id: AppId, name: &str) -> Result, TopicRepoError> { Ok(self .0 .iter() .find(|(a, t)| *a == app_id && t.name == name) .map(|(_, t)| t.clone())) } async fn update( &self, _: AppId, _: &str, _: Option, _: Option, ) -> Result, TopicRepoError> { unimplemented!() } async fn delete(&self, _: AppId, _: &str) -> Result { unimplemented!() } } struct FakeSecrets(AppId, Vec); #[async_trait] impl AppSecretsRepo for FakeSecrets { async fn get_or_create_signing_key( &self, _: AppId, ) -> Result, AppSecretsRepoError> { Ok(self.1.clone()) } async fn signing_key(&self, app_id: AppId) -> Result>, AppSecretsRepoError> { Ok((app_id == self.0).then(|| self.1.clone())) } } fn topic(name: &str, external: bool, mode: TopicAuthMode) -> Topic { Topic { name: name.to_string(), external_subscribable: external, auth_mode: mode, created_at: Utc::now(), updated_at: Utc::now(), } } fn authority( topics: Vec<(AppId, Topic)>, key_app: AppId, key: Vec, ) -> RealtimeAuthorityImpl { RealtimeAuthorityImpl::new( Arc::new(FakeTopics(topics)), Arc::new(FakeSecrets(key_app, key)), Arc::new(picloud_shared::NoopUsersService), ) } #[tokio::test] async fn missing_topic_is_not_found() { let app = AppId::new(); let a = authority(vec![], app, vec![0u8; 32]); assert_eq!( a.authorize_subscribe(app, "ghost", None).await, Err(SubscribeDenied::NotFound) ); } #[tokio::test] async fn internal_only_topic_is_not_found() { let app = AppId::new(); let a = authority( vec![(app, topic("internal", false, TopicAuthMode::Public))], app, vec![0u8; 32], ); assert_eq!( a.authorize_subscribe(app, "internal", None).await, Err(SubscribeDenied::NotFound) ); } #[tokio::test] async fn public_topic_allows_without_token() { let app = AppId::new(); let a = authority( vec![(app, topic("news", true, TopicAuthMode::Public))], app, vec![0u8; 32], ); assert!(a.authorize_subscribe(app, "news", None).await.is_ok()); } #[tokio::test] async fn token_topic_without_token_is_unauthorized() { let app = AppId::new(); let a = authority( vec![(app, topic("chat", true, TopicAuthMode::Token))], app, vec![7u8; 32], ); assert_eq!( a.authorize_subscribe(app, "chat", None).await, Err(SubscribeDenied::Unauthorized) ); } #[tokio::test] async fn token_topic_with_valid_token_allows() { let app = AppId::new(); let key = vec![9u8; 32]; let a = authority( vec![(app, topic("chat", true, TopicAuthMode::Token))], app, key.clone(), ); let token = sign( &key, &TokenClaims { app_id: app, topics: vec!["chat".into()], iat: Utc::now().timestamp(), exp: Utc::now().timestamp() + 60, }, ); assert!(a .authorize_subscribe(app, "chat", Some(&token)) .await .is_ok()); } #[tokio::test] async fn token_for_other_topic_is_unauthorized() { let app = AppId::new(); let key = vec![9u8; 32]; let a = authority( vec![(app, topic("chat", true, TopicAuthMode::Token))], app, key.clone(), ); let token = sign( &key, &TokenClaims { app_id: app, topics: vec!["other".into()], iat: Utc::now().timestamp(), exp: Utc::now().timestamp() + 60, }, ); assert_eq!( a.authorize_subscribe(app, "chat", Some(&token)).await, Err(SubscribeDenied::Unauthorized) ); } #[tokio::test] async fn expired_token_is_unauthorized() { let app = AppId::new(); let key = vec![9u8; 32]; let a = authority( vec![(app, topic("chat", true, TopicAuthMode::Token))], app, key.clone(), ); let token = sign( &key, &TokenClaims { app_id: app, topics: vec!["chat".into()], iat: Utc::now().timestamp() - 120, exp: Utc::now().timestamp() - 60, }, ); assert_eq!( a.authorize_subscribe(app, "chat", Some(&token)).await, Err(SubscribeDenied::Unauthorized) ); } /// Fake UsersService that resolves a specific (token, app_id) pair /// into a stubbed AppUser. All other methods return Backend /// errors via NoopUsersService — only verify_session_for_realtime /// is exercised by these tests. struct FakeUsersForRealtime { token: String, app_id: AppId, user: picloud_shared::AppUser, } #[async_trait] impl UsersService for FakeUsersForRealtime { async fn verify_session_for_realtime( &self, app_id: AppId, token: &str, ) -> Result, picloud_shared::UsersError> { if app_id == self.app_id && token == self.token { Ok(Some(self.user.clone())) } else { Ok(None) } } async fn create( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::CreateUserInput, ) -> Result { unimplemented!() } async fn get( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::AppUserId, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn find_by_email( &self, _: &picloud_shared::SdkCallCx, _: &str, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn update( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::AppUserId, _: picloud_shared::UpdateUserInput, ) -> Result { unimplemented!() } async fn delete( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::AppUserId, ) -> Result { unimplemented!() } async fn list( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::UsersListOpts, ) -> Result { unimplemented!() } async fn login( &self, _: &picloud_shared::SdkCallCx, _: &str, _: &str, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn verify( &self, _: &picloud_shared::SdkCallCx, _: &str, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn logout( &self, _: &picloud_shared::SdkCallCx, _: &str, ) -> Result<(), picloud_shared::UsersError> { unimplemented!() } async fn send_verification_email( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::AppUserId, _: picloud_shared::EmailTemplateOpts, ) -> Result<(), picloud_shared::UsersError> { unimplemented!() } async fn verify_email( &self, _: &picloud_shared::SdkCallCx, _: &str, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn request_password_reset( &self, _: &picloud_shared::SdkCallCx, _: &str, _: picloud_shared::EmailTemplateOpts, ) -> Result<(), picloud_shared::UsersError> { unimplemented!() } async fn complete_password_reset( &self, _: &picloud_shared::SdkCallCx, _: &str, _: &str, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn invite( &self, _: &picloud_shared::SdkCallCx, _: &str, _: picloud_shared::InviteOpts, ) -> Result<(), picloud_shared::UsersError> { unimplemented!() } async fn accept_invite( &self, _: &picloud_shared::SdkCallCx, _: &str, _: &str, _: Option, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn add_role( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::AppUserId, _: &str, ) -> Result<(), picloud_shared::UsersError> { unimplemented!() } async fn remove_role( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::AppUserId, _: &str, ) -> Result { unimplemented!() } async fn has_role( &self, _: &picloud_shared::SdkCallCx, _: picloud_shared::AppUserId, _: &str, ) -> Result { unimplemented!() } async fn admin_reset_password_token( &self, _: &picloud_shared::Principal, _: AppId, _: picloud_shared::AppUserId, ) -> Result { unimplemented!() } async fn admin_revoke_all_sessions( &self, _: &picloud_shared::Principal, _: AppId, _: picloud_shared::AppUserId, ) -> Result { unimplemented!() } async fn admin_create_invitation( &self, _: &picloud_shared::Principal, _: AppId, _: &str, _: picloud_shared::InviteOpts, ) -> Result { unimplemented!() } async fn list_invitations( &self, _: &picloud_shared::Principal, _: AppId, ) -> Result, picloud_shared::UsersError> { unimplemented!() } async fn revoke_invitation( &self, _: &picloud_shared::Principal, _: AppId, _: picloud_shared::InvitationId, ) -> Result { unimplemented!() } } fn stub_user(app_id: AppId) -> picloud_shared::AppUser { picloud_shared::AppUser { id: picloud_shared::AppUserId::new(), app_id, email: "alice@example.com".into(), display_name: Some("Alice".into()), email_verified_at: None, last_login_at: None, created_at: Utc::now(), updated_at: Utc::now(), roles: vec![], } } fn session_authority( topics: Vec<(AppId, Topic)>, app_id: AppId, token: &str, ) -> RealtimeAuthorityImpl { RealtimeAuthorityImpl::new( Arc::new(FakeTopics(topics)), Arc::new(FakeSecrets(app_id, vec![0u8; 32])), Arc::new(FakeUsersForRealtime { token: token.to_string(), app_id, user: stub_user(app_id), }), ) } #[tokio::test] async fn session_topic_with_valid_session_token_allows() { let app = AppId::new(); let a = session_authority( vec![(app, topic("chat", true, TopicAuthMode::Session))], app, "valid-session-token", ); assert!(a .authorize_subscribe(app, "chat", Some("valid-session-token")) .await .is_ok()); } #[tokio::test] async fn session_topic_without_token_is_unauthorized() { let app = AppId::new(); let a = session_authority( vec![(app, topic("chat", true, TopicAuthMode::Session))], app, "valid-session-token", ); assert_eq!( a.authorize_subscribe(app, "chat", None).await, Err(SubscribeDenied::Unauthorized) ); } #[tokio::test] async fn session_topic_with_wrong_token_is_unauthorized() { let app = AppId::new(); let a = session_authority( vec![(app, topic("chat", true, TopicAuthMode::Session))], app, "valid-session-token", ); assert_eq!( a.authorize_subscribe(app, "chat", Some("wrong-token")) .await, Err(SubscribeDenied::Unauthorized) ); } #[tokio::test] async fn session_token_for_other_app_is_unauthorized() { let app_a = AppId::new(); let app_b = AppId::new(); // Authority for app B; users service knows app A's token. let a = RealtimeAuthorityImpl::new( Arc::new(FakeTopics(vec![( app_b, topic("chat", true, TopicAuthMode::Session), )])), Arc::new(FakeSecrets(app_b, vec![0u8; 32])), Arc::new(FakeUsersForRealtime { token: "app-a-token".into(), app_id: app_a, user: stub_user(app_a), }), ); assert_eq!( a.authorize_subscribe(app_b, "chat", Some("app-a-token")) .await, Err(SubscribeDenied::Unauthorized) ); } #[tokio::test] async fn token_signed_by_other_app_key_is_unauthorized() { let app_a = AppId::new(); let app_b = AppId::new(); let key_a = vec![1u8; 32]; let key_b = vec![2u8; 32]; // Authority for app B; its key is key_b. let a = authority( vec![(app_b, topic("chat", true, TopicAuthMode::Token))], app_b, key_b, ); // Token signed by app A's key, claiming app A. let token = sign( &key_a, &TokenClaims { app_id: app_a, topics: vec!["chat".into()], iat: Utc::now().timestamp(), exp: Utc::now().timestamp() + 60, }, ); assert_eq!( a.authorize_subscribe(app_b, "chat", Some(&token)).await, Err(SubscribeDenied::Unauthorized) ); } }