diff --git a/crates/manager-core/migrations/0030_app_user_invitations.sql b/crates/manager-core/migrations/0030_app_user_invitations.sql new file mode 100644 index 0000000..a639a30 --- /dev/null +++ b/crates/manager-core/migrations/0030_app_user_invitations.sql @@ -0,0 +1,29 @@ +-- v1.1.8 User Management — invitations. +-- +-- Unlike verification + reset tokens, invitations don't carry a +-- user_id — the user doesn't exist yet. Instead they pre-stage the +-- email, an optional display name, and a roles array applied on +-- accept (once the per-app role table exists in migration 0031). +-- +-- token_hash UNIQUE is the lookup key; the surrogate `id` UUID is +-- what the admin invitations UI references (rotation-safe; an +-- admin can list pending invites by id without leaking tokens). +-- +-- accepted_at gates one-shot semantics: the consume path is an +-- atomic UPDATE WHERE accepted_at IS NULL. Stale accept attempts get +-- nothing, so a leaked / cached token can't be replayed. + +CREATE TABLE app_user_invitations ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + token_hash TEXT NOT NULL UNIQUE, + app_id UUID NOT NULL REFERENCES apps(id) ON DELETE CASCADE, + email TEXT NOT NULL, + display_name TEXT, + roles TEXT[] NOT NULL DEFAULT '{}', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + expires_at TIMESTAMPTZ NOT NULL, + accepted_at TIMESTAMPTZ +); + +CREATE INDEX idx_app_user_invitations_app_pending + ON app_user_invitations (app_id) WHERE accepted_at IS NULL; diff --git a/crates/manager-core/src/app_user_invitation_repo.rs b/crates/manager-core/src/app_user_invitation_repo.rs new file mode 100644 index 0000000..985a11d --- /dev/null +++ b/crates/manager-core/src/app_user_invitation_repo.rs @@ -0,0 +1,219 @@ +//! CRUD over `app_user_invitations` (v1.1.8 commit 7). +//! +//! Distinct shape from the verification / reset repos: pre-stages +//! email + display_name + roles for a user that doesn't exist yet. +//! The admin UI references rows by surrogate `id`; the consumer +//! references by `token_hash`. Both views are scoped by `app_id`. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use picloud_shared::{AppId, InvitationId}; +use sqlx::PgPool; + +#[derive(Debug, thiserror::Error)] +pub enum AppUserInvitationRepoError { + #[error("database error: {0}")] + Db(#[from] sqlx::Error), +} + +#[derive(Debug, Clone)] +pub struct InvitationRow { + pub id: InvitationId, + pub app_id: AppId, + pub email: String, + pub display_name: Option, + pub roles: Vec, + pub created_at: DateTime, + pub expires_at: DateTime, + pub accepted_at: Option>, +} + +/// Snapshot returned by `consume` — just the fields needed to create +/// the new user. Excludes timestamps and ids since the caller already +/// has the token hash and the consume happened atomically. +#[derive(Debug, Clone)] +pub struct ConsumedInvitation { + pub email: String, + pub display_name: Option, + pub roles: Vec, +} + +#[async_trait] +pub trait AppUserInvitationRepo: Send + Sync { + async fn create( + &self, + app_id: AppId, + email: &str, + display_name: Option<&str>, + roles: &[String], + token_hash: &str, + expires_at: DateTime, + ) -> Result; + + /// Pending invitations (accepted_at IS NULL) for the admin UI. + async fn list_pending( + &self, + app_id: AppId, + ) -> Result, AppUserInvitationRepoError>; + + /// Admin revoke (deletes the pending row). Returns whether a row + /// was actually removed. + async fn revoke( + &self, + app_id: AppId, + invite_id: InvitationId, + ) -> Result; + + /// Atomic one-shot consume by token. Returns the staged fields + /// for the caller to create the user from; None on + /// missing / already-accepted / expired / wrong app. + async fn consume_by_token( + &self, + app_id: AppId, + token_hash: &str, + ) -> Result, AppUserInvitationRepoError>; + + /// GC: drop accepted or expired rows. + async fn gc(&self, batch_size: i64) -> Result; +} + +pub struct PostgresAppUserInvitationRepo { + pool: PgPool, +} + +impl PostgresAppUserInvitationRepo { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[async_trait] +impl AppUserInvitationRepo for PostgresAppUserInvitationRepo { + async fn create( + &self, + app_id: AppId, + email: &str, + display_name: Option<&str>, + roles: &[String], + token_hash: &str, + expires_at: DateTime, + ) -> Result { + let row: InvitationRecord = sqlx::query_as( + "INSERT INTO app_user_invitations \ + (token_hash, app_id, email, display_name, roles, expires_at) \ + VALUES ($1, $2, $3, $4, $5, $6) \ + RETURNING id, app_id, email, display_name, roles, \ + created_at, expires_at, accepted_at", + ) + .bind(token_hash) + .bind(app_id.into_inner()) + .bind(email) + .bind(display_name) + .bind(roles) + .bind(expires_at) + .fetch_one(&self.pool) + .await?; + Ok(row.into()) + } + + async fn list_pending( + &self, + app_id: AppId, + ) -> Result, AppUserInvitationRepoError> { + let rows: Vec = sqlx::query_as( + "SELECT id, app_id, email, display_name, roles, \ + created_at, expires_at, accepted_at \ + FROM app_user_invitations \ + WHERE app_id = $1 AND accepted_at IS NULL \ + ORDER BY created_at DESC", + ) + .bind(app_id.into_inner()) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(Into::into).collect()) + } + + async fn revoke( + &self, + app_id: AppId, + invite_id: InvitationId, + ) -> Result { + let res = sqlx::query( + "DELETE FROM app_user_invitations \ + WHERE app_id = $1 AND id = $2 AND accepted_at IS NULL", + ) + .bind(app_id.into_inner()) + .bind(invite_id.into_inner()) + .execute(&self.pool) + .await?; + Ok(res.rows_affected() > 0) + } + + async fn consume_by_token( + &self, + app_id: AppId, + token_hash: &str, + ) -> Result, AppUserInvitationRepoError> { + let row: Option<(String, Option, Vec)> = sqlx::query_as( + "UPDATE app_user_invitations \ + SET accepted_at = NOW() \ + WHERE token_hash = $1 \ + AND app_id = $2 \ + AND accepted_at IS NULL \ + AND expires_at > NOW() \ + RETURNING email, display_name, roles", + ) + .bind(token_hash) + .bind(app_id.into_inner()) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|(email, display_name, roles)| ConsumedInvitation { + email, + display_name, + roles, + })) + } + + async fn gc(&self, batch_size: i64) -> Result { + let res = sqlx::query( + "DELETE FROM app_user_invitations WHERE id IN ( \ + SELECT id FROM app_user_invitations \ + WHERE expires_at <= NOW() OR accepted_at IS NOT NULL \ + LIMIT $1 \ + FOR UPDATE SKIP LOCKED \ + )", + ) + .bind(batch_size) + .execute(&self.pool) + .await?; + Ok(res.rows_affected()) + } +} + +#[derive(sqlx::FromRow)] +struct InvitationRecord { + id: uuid::Uuid, + app_id: uuid::Uuid, + email: String, + display_name: Option, + roles: Vec, + created_at: DateTime, + expires_at: DateTime, + accepted_at: Option>, +} + +impl From for InvitationRow { + fn from(r: InvitationRecord) -> Self { + Self { + id: r.id.into(), + app_id: r.app_id.into(), + email: r.email, + display_name: r.display_name, + roles: r.roles, + created_at: r.created_at, + expires_at: r.expires_at, + accepted_at: r.accepted_at, + } + } +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index c979aac..6eb97c0 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -17,6 +17,7 @@ pub mod app_members_api; pub mod app_members_repo; pub mod app_repo; pub mod app_secrets_repo; +pub mod app_user_invitation_repo; pub mod app_user_password_reset_repo; pub mod app_user_repo; pub mod app_user_session_repo; @@ -89,6 +90,10 @@ pub use app_user_session_repo::{ AppUserSessionLookup, AppUserSessionRepository, AppUserSessionRepositoryError, PostgresAppUserSessionRepository, }; +pub use app_user_invitation_repo::{ + AppUserInvitationRepo, AppUserInvitationRepoError, ConsumedInvitation, InvitationRow, + PostgresAppUserInvitationRepo, +}; pub use app_user_password_reset_repo::{ AppUserPasswordResetRepo, AppUserPasswordResetRepoError, PostgresAppUserPasswordResetRepo, }; diff --git a/crates/manager-core/src/users_service.rs b/crates/manager-core/src/users_service.rs index c4bf0ba..fa1de04 100644 --- a/crates/manager-core/src/users_service.rs +++ b/crates/manager-core/src/users_service.rs @@ -33,6 +33,7 @@ use picloud_shared::{ UsersListOpts, UsersListPage, UsersService, }; +use crate::app_user_invitation_repo::{AppUserInvitationRepo, AppUserInvitationRepoError}; use crate::app_user_password_reset_repo::{ AppUserPasswordResetRepo, AppUserPasswordResetRepoError, }; @@ -127,6 +128,7 @@ pub struct UsersServiceImpl { sessions: Arc, verifications: Arc, password_resets: Arc, + invitations: Arc, email: Arc, authz: Arc, events: Arc, @@ -141,6 +143,7 @@ impl UsersServiceImpl { sessions: Arc, verifications: Arc, password_resets: Arc, + invitations: Arc, email: Arc, authz: Arc, events: Arc, @@ -151,6 +154,7 @@ impl UsersServiceImpl { sessions, verifications, password_resets, + invitations, email, authz, events, @@ -287,6 +291,14 @@ impl From for UsersError { } } +impl From for UsersError { + fn from(e: AppUserInvitationRepoError) -> Self { + match e { + AppUserInvitationRepoError::Db(err) => UsersError::Backend(err.to_string()), + } + } +} + fn map_email_error(e: EmailError) -> UsersError { match e { EmailError::NotConfigured => UsersError::EmailNotConfigured, @@ -698,20 +710,125 @@ impl UsersService for UsersServiceImpl { } async fn invite( &self, - _cx: &SdkCallCx, - _email: &str, - _opts: InviteOpts, + cx: &SdkCallCx, + email: &str, + opts: InviteOpts, ) -> Result<(), UsersError> { - Err(UsersError::Backend(NOT_YET_IMPL.into())) + self.require(cx.principal.as_ref(), Capability::AppUsersAdmin(cx.app_id)) + .await?; + let normalized = validate_email(email)?; + // Roles are stored verbatim; commit 8 (per-app roles) is what + // actually applies them on accept. v1.1.8 ships with empty + // role lists or whatever the script supplies — the storage + // round-trip is identical either way. + let display_name = validate_display_name(opts.display_name.as_deref())?; + let token = generate_session_token(); + let expires_at = Utc::now() + + chrono::Duration::from_std(self.config.invitation_ttl) + .map_err(|e| UsersError::Backend(format!("invitation ttl: {e}")))?; + let row = self + .invitations + .create( + cx.app_id, + &normalized, + display_name.as_deref(), + &opts.roles, + &token.hash, + expires_at, + ) + .await?; + + // Send the invitation email if a template was supplied. + // Inviters can skip the email by omitting the template — useful + // when the invitation flow is driven by an external delivery + // (e.g., printed onboarding letter). + if let Some(template) = opts.template { + let link = build_link(&template.link_base, &token.raw); + let body = template.body_template.replace("{link}", &link); + let outbound = OutboundEmail { + to: vec![normalized.clone()], + from: template.from, + subject: template.subject, + text: Some(body), + ..Default::default() + }; + let send_cx = internal_cx(cx); + if let Err(e) = self.email.send(&send_cx, outbound).await { + if matches!(e, EmailError::NotConfigured) { + return Err(UsersError::EmailNotConfigured); + } + tracing::warn!( + error = %e, + invite_id = %row.id, + app_id = %cx.app_id, + "invite email send failed; invitation row remains valid" + ); + } + } + Ok(()) } + async fn accept_invite( &self, - _cx: &SdkCallCx, - _token: &str, - _password: &str, - _display_name: Option, + cx: &SdkCallCx, + token: &str, + password: &str, + display_name: Option, ) -> Result, UsersError> { - Err(UsersError::Backend(NOT_YET_IMPL.into())) + // No require: the invitation token IS the authorization. + validate_password(password)?; + let token_hash = hash_token(token); + let Some(consumed) = self.invitations.consume_by_token(cx.app_id, &token_hash).await? else { + return Ok(None); + }; + + // Display name on the new user: caller override beats the + // invitation's pre-staged value beats None. + let final_display = display_name + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string) + .or(consumed.display_name); + + let password_hash = hash_password(password) + .map_err(|e| UsersError::Backend(format!("password hashing failed: {e}")))?; + let row = match self + .users + .create( + cx.app_id, + &consumed.email, + &password_hash, + final_display.as_deref(), + ) + .await + { + Ok(row) => row, + Err(AppUserRepositoryError::DuplicateEmail(_)) => { + // User already exists (sign-up beat acceptance). The + // invitation is consumed — they'll need to log in + // normally with their existing password. + return Ok(None); + } + Err(e) => return Err(e.into()), + }; + // Commit 8 wires the role-application path; in commit 7 the + // staged roles are stored on the invitation row but not + // applied (the role table doesn't exist yet). Logged so an + // operator can audit the gap. + if !consumed.roles.is_empty() { + tracing::info!( + user_id = %row.id, + app_id = %cx.app_id, + roles = ?consumed.roles, + "accept_invite: pre-staged roles will be applied once migration 0031 lands" + ); + } + let user_id = row.id; + let user = Self::to_app_user(row, Vec::new()); + let session = self.mint_session(cx.app_id, user_id, &user).await?; + self.emit(cx, "create", user_id).await; + Ok(Some(GeneratedAccept { user, session })) } async fn add_role( &self, @@ -755,22 +872,69 @@ impl UsersService for UsersServiceImpl { } async fn list_invitations( &self, - _principal: &Principal, - _app_id: AppId, + principal: &Principal, + app_id: AppId, ) -> Result, UsersError> { - Err(UsersError::Backend(NOT_YET_IMPL.into())) + self.require(Some(principal), Capability::AppUsersAdmin(app_id)) + .await?; + let rows = self.invitations.list_pending(app_id).await?; + Ok(rows + .into_iter() + .map(|r| Invitation { + id: r.id, + app_id: r.app_id, + email: r.email, + display_name: r.display_name, + roles: r.roles, + created_at: r.created_at, + expires_at: r.expires_at, + }) + .collect()) } async fn revoke_invitation( &self, - _principal: &Principal, - _app_id: AppId, - _invite_id: InvitationId, + principal: &Principal, + app_id: AppId, + invite_id: InvitationId, ) -> Result { - Err(UsersError::Backend(NOT_YET_IMPL.into())) + self.require(Some(principal), Capability::AppUsersAdmin(app_id)) + .await?; + Ok(self.invitations.revoke(app_id, invite_id).await?) } } impl UsersServiceImpl { + /// Mint a session for an already-authenticated user. Used by + /// `login` (after password verify) and `accept_invite` (after + /// invitation consume). Returns the raw token to embed in the + /// response; only the SHA-256 hash hits storage. + async fn mint_session( + &self, + app_id: AppId, + user_id: AppUserId, + user: &AppUser, + ) -> Result { + let now = Utc::now(); + let absolute = now + + chrono::Duration::from_std(self.config.session_absolute_ttl) + .map_err(|e| UsersError::Backend(format!("absolute ttl: {e}")))?; + let sliding = now + + chrono::Duration::from_std(self.config.session_ttl) + .map_err(|e| UsersError::Backend(format!("sliding ttl: {e}")))?; + let expires_at = sliding.min(absolute); + + let token = generate_session_token(); + self.sessions + .create(app_id, user_id, &token.hash, expires_at, absolute) + .await?; + self.users.touch_last_login(app_id, user_id).await?; + Ok(GeneratedSession { + token: token.raw, + user: user.clone(), + expires_at, + }) + } + /// Shared lookup logic for `verify` and `verify_session_for_realtime`. /// Honors the absolute cap and cross-app isolation, then bumps the /// sliding window (capped at the absolute cap) on success. diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 4371d2e..7324380 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -21,9 +21,10 @@ use picloud_manager_core::{ FilesServiceImpl, FsFilesRepo, HttpConfig, HttpServiceImpl, KvServiceImpl, OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository, - PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserPasswordResetRepo, - PostgresAppUserRepository, PostgresAppUserSessionRepository, - PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, PostgresDeadLetterService, + PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserInvitationRepo, + PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, + PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, + PostgresDeadLetterService, PostgresDocsRepo, PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo, PrincipalResolver, @@ -255,11 +256,14 @@ pub async fn build_app( Arc::new(PostgresAppUserVerificationRepo::new(pool.clone())); let app_user_password_resets_repo = Arc::new(PostgresAppUserPasswordResetRepo::new(pool.clone())); + let app_user_invitations_repo = + Arc::new(PostgresAppUserInvitationRepo::new(pool.clone())); let users: Arc = Arc::new(UsersServiceImpl::new( app_users_repo.clone(), app_user_sessions_repo.clone(), app_user_verifications_repo.clone(), app_user_password_resets_repo.clone(), + app_user_invitations_repo.clone(), email.clone(), authz.clone(), events.clone(),