diff --git a/crates/manager-core/migrations/0033_topics_auth_mode_session.sql b/crates/manager-core/migrations/0033_topics_auth_mode_session.sql new file mode 100644 index 0000000..fb505fa --- /dev/null +++ b/crates/manager-core/migrations/0033_topics_auth_mode_session.sql @@ -0,0 +1,14 @@ +-- v1.1.8 F3 — extend topics.auth_mode CHECK to allow 'session'. +-- +-- v1.1.6 shipped 'public' + 'token'. v1.1.8 adds 'session' so a +-- topic can authorize SSE subscribers against a per-app user session +-- minted by users::login. The realtime authority delegates verify to +-- UsersService::verify_session_for_realtime, which returns the user +-- on success; the subscription proceeds with that principal. +-- +-- 'script' (v1.2 script-mediated subscribe auth) extends the +-- constraint a third time later. + +ALTER TABLE topics DROP CONSTRAINT IF EXISTS topics_auth_mode_check; +ALTER TABLE topics ADD CONSTRAINT topics_auth_mode_check + CHECK (auth_mode IN ('public', 'token', 'session')); diff --git a/crates/manager-core/src/realtime_authority.rs b/crates/manager-core/src/realtime_authority.rs index 848e34e..8cdf6fc 100644 --- a/crates/manager-core/src/realtime_authority.rs +++ b/crates/manager-core/src/realtime_authority.rs @@ -22,7 +22,7 @@ use std::collections::HashMap; use std::sync::{Arc, Mutex}; use async_trait::async_trait; -use picloud_shared::{subscriber_token, AppId, RealtimeAuthority, SubscribeDenied}; +use picloud_shared::{subscriber_token, AppId, RealtimeAuthority, SubscribeDenied, UsersService}; use crate::app_secrets_repo::AppSecretsRepo; use crate::topic_repo::{TopicAuthMode, TopicRepo}; @@ -30,15 +30,21 @@ use crate::topic_repo::{TopicAuthMode, TopicRepo}; pub struct RealtimeAuthorityImpl { topics: Arc, secrets: Arc, + users: Arc, key_cache: Mutex>>, } impl RealtimeAuthorityImpl { #[must_use] - pub fn new(topics: Arc, secrets: Arc) -> Self { + pub fn new( + topics: Arc, + secrets: Arc, + users: Arc, + ) -> Self { Self { topics, secrets, + users, key_cache: Mutex::new(HashMap::new()), } } @@ -104,6 +110,28 @@ impl RealtimeAuthority for RealtimeAuthorityImpl { } 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(()) + } } } } @@ -184,6 +212,7 @@ mod tests { RealtimeAuthorityImpl::new( Arc::new(FakeTopics(topics)), Arc::new(FakeSecrets(key_app, key)), + Arc::new(picloud_shared::NoopUsersService), ) } @@ -308,6 +337,304 @@ mod tests { ); } + /// 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(); diff --git a/crates/manager-core/src/topic_repo.rs b/crates/manager-core/src/topic_repo.rs index f425ac9..f256f03 100644 --- a/crates/manager-core/src/topic_repo.rs +++ b/crates/manager-core/src/topic_repo.rs @@ -13,14 +13,16 @@ use picloud_shared::AppId; use serde::{Deserialize, Serialize}; use sqlx::PgPool; -/// External-subscriber auth gate for a topic. `'public'` + `'token'` in -/// v1.1.6; `'session'` (v1.1.8) and `'script'` (v1.2) extend the DB -/// CHECK constraint and this enum later. +/// External-subscriber auth gate for a topic. v1.1.6 shipped +/// `'public'` + `'token'`; v1.1.8 F3 adds `'session'` (per-app user +/// sessions authorize subscription); `'script'` (v1.2) extends the +/// DB CHECK constraint and this enum later. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] pub enum TopicAuthMode { Public, Token, + Session, } impl TopicAuthMode { @@ -29,6 +31,7 @@ impl TopicAuthMode { match self { Self::Public => "public", Self::Token => "token", + Self::Session => "session", } } @@ -36,6 +39,7 @@ impl TopicAuthMode { match s { "public" => Ok(Self::Public), "token" => Ok(Self::Token), + "session" => Ok(Self::Session), other => Err(TopicRepoError::Backend(format!( "unknown auth_mode in DB: {other}" ))), diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 74b2cbe..30f35fa 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -201,11 +201,6 @@ pub async fn build_app( // column. Operators upgrading from v1.1.6 or earlier MUST apply // v1.1.7 first; the 0032 migration's guard refuses to run // otherwise. - let realtime_authority: Arc = Arc::new(RealtimeAuthorityImpl::new( - topic_repo.clone(), - app_secrets_repo.clone(), - )); - // v1.1.5 durable pub/sub, extended in v1.1.6 with the realtime // broadcast + subscriber-token mint. Publishes fan out to matching // pubsub triggers at publish time (one outbox row each, delivered by @@ -216,7 +211,7 @@ pub async fn build_app( PubsubServiceImpl::new(pubsub_repo, authz.clone()).with_realtime( broadcaster.clone(), topic_repo.clone(), - app_secrets_repo, + app_secrets_repo.clone(), SubscriberTokenConfig::from_env(), ), ); @@ -261,6 +256,14 @@ pub async fn build_app( events.clone(), UsersServiceConfig::from_env(), )); + // v1.1.8 F3: build the realtime authority now that UsersService + // is constructed; the Session arm of authorize_subscribe calls + // users.verify_session_for_realtime. + let realtime_authority: Arc = Arc::new(RealtimeAuthorityImpl::new( + topic_repo.clone(), + app_secrets_repo.clone(), + users.clone(), + )); let services = Services::new( kv, docs, diff --git a/dashboard/src/lib/api.ts b/dashboard/src/lib/api.ts index 8e5ce36..f5377f4 100644 --- a/dashboard/src/lib/api.ts +++ b/dashboard/src/lib/api.ts @@ -348,7 +348,7 @@ export interface CreatePubsubTriggerInput { } // v1.1.6 — externally-subscribable realtime topics. -export type TopicAuthMode = 'public' | 'token'; +export type TopicAuthMode = 'public' | 'token' | 'session'; export interface Topic { name: string; diff --git a/dashboard/src/routes/apps/[slug]/+page.svelte b/dashboard/src/routes/apps/[slug]/+page.svelte index 1b5ab75..9423cc4 100644 --- a/dashboard/src/routes/apps/[slug]/+page.svelte +++ b/dashboard/src/routes/apps/[slug]/+page.svelte @@ -1265,6 +1265,13 @@ token — requires a valid subscriber token + {/if} {#if createTopicError} @@ -1583,6 +1590,13 @@ token — requires a valid subscriber token + {/if} {#if editFlipToExternal}