diff --git a/crates/executor-core/src/sdk/users.rs b/crates/executor-core/src/sdk/users.rs index a5314a1..c35d8c3 100644 --- a/crates/executor-core/src/sdk/users.rs +++ b/crates/executor-core/src/sdk/users.rs @@ -530,6 +530,7 @@ fn parse_user_id(s: &str, ctx: &str) -> Result> { fn parse_email_template(opts: &Map, ctx: &str) -> Result> { Ok(EmailTemplateOpts { link_base: required_string(opts, "link_base", ctx)?, + from: required_string(opts, "from", ctx)?, subject: required_string(opts, "subject", ctx)?, body_template: required_string(opts, "body_template", ctx)?, }) diff --git a/crates/manager-core/migrations/0028_app_user_email_verifications.sql b/crates/manager-core/migrations/0028_app_user_email_verifications.sql new file mode 100644 index 0000000..a607b5c --- /dev/null +++ b/crates/manager-core/migrations/0028_app_user_email_verifications.sql @@ -0,0 +1,19 @@ +-- v1.1.8 User Management — email verification tokens. +-- +-- Created by `users::send_verification_email`; consumed by +-- `users::verify_email`. Same SHA-256 token-hash shape as +-- `app_user_sessions`. Single-use: the consume path is an atomic +-- UPDATE WHERE consumed_at IS NULL so a replayed token returns +-- "missing" the second time around without spurious side effects. + +CREATE TABLE app_user_email_verifications ( + token_hash TEXT PRIMARY KEY, + app_id UUID NOT NULL REFERENCES apps(id) ON DELETE CASCADE, + user_id UUID NOT NULL REFERENCES app_users(id) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + expires_at TIMESTAMPTZ NOT NULL, + consumed_at TIMESTAMPTZ +); + +CREATE INDEX idx_app_user_email_verifications_user + ON app_user_email_verifications (app_id, user_id); diff --git a/crates/manager-core/src/app_user_verification_repo.rs b/crates/manager-core/src/app_user_verification_repo.rs new file mode 100644 index 0000000..d766660 --- /dev/null +++ b/crates/manager-core/src/app_user_verification_repo.rs @@ -0,0 +1,114 @@ +//! CRUD over `app_user_email_verifications` (v1.1.8 commit 5). +//! +//! One-shot tokens. The consume path is an atomic UPDATE WHERE +//! consumed_at IS NULL: rowcount = 1 means the token was valid and is +//! now used; rowcount = 0 means missing / already-consumed / wrong +//! app. The repo never returns the raw token — only the user id the +//! consumed token belonged to. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use picloud_shared::{AppId, AppUserId}; +use sqlx::PgPool; + +#[derive(Debug, thiserror::Error)] +pub enum AppUserVerificationRepoError { + #[error("database error: {0}")] + Db(#[from] sqlx::Error), +} + +#[async_trait] +pub trait AppUserVerificationRepo: Send + Sync { + async fn create( + &self, + app_id: AppId, + user_id: AppUserId, + token_hash: &str, + expires_at: DateTime, + ) -> Result<(), AppUserVerificationRepoError>; + + /// Atomic consume. Returns `Some(user_id)` on success (rowcount = 1), + /// `None` on missing / already-consumed / expired / wrong app. + /// Scoped to `app_id` so a token issued for app A cannot be + /// presented to app B. + async fn consume( + &self, + app_id: AppId, + token_hash: &str, + ) -> Result, AppUserVerificationRepoError>; + + /// GC: drop consumed or expired rows, matching the dead-letter / + /// session sweep pattern. + async fn gc(&self, batch_size: i64) -> Result; +} + +pub struct PostgresAppUserVerificationRepo { + pool: PgPool, +} + +impl PostgresAppUserVerificationRepo { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[async_trait] +impl AppUserVerificationRepo for PostgresAppUserVerificationRepo { + async fn create( + &self, + app_id: AppId, + user_id: AppUserId, + token_hash: &str, + expires_at: DateTime, + ) -> Result<(), AppUserVerificationRepoError> { + sqlx::query( + "INSERT INTO app_user_email_verifications \ + (token_hash, app_id, user_id, expires_at) \ + VALUES ($1, $2, $3, $4)", + ) + .bind(token_hash) + .bind(app_id.into_inner()) + .bind(user_id.into_inner()) + .bind(expires_at) + .execute(&self.pool) + .await?; + Ok(()) + } + + async fn consume( + &self, + app_id: AppId, + token_hash: &str, + ) -> Result, AppUserVerificationRepoError> { + let row: Option<(uuid::Uuid,)> = sqlx::query_as( + "UPDATE app_user_email_verifications \ + SET consumed_at = NOW() \ + WHERE token_hash = $1 \ + AND app_id = $2 \ + AND consumed_at IS NULL \ + AND expires_at > NOW() \ + RETURNING user_id", + ) + .bind(token_hash) + .bind(app_id.into_inner()) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(|(uid,)| uid.into())) + } + + async fn gc(&self, batch_size: i64) -> Result { + let res = sqlx::query( + "DELETE FROM app_user_email_verifications WHERE token_hash IN ( \ + SELECT token_hash FROM app_user_email_verifications \ + WHERE expires_at <= NOW() OR consumed_at IS NOT NULL \ + LIMIT $1 \ + FOR UPDATE SKIP LOCKED \ + )", + ) + .bind(batch_size) + .execute(&self.pool) + .await?; + Ok(res.rows_affected()) + } +} diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index d1fe7b5..a375b6e 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -19,6 +19,7 @@ pub mod app_repo; pub mod app_secrets_repo; pub mod app_user_repo; pub mod app_user_session_repo; +pub mod app_user_verification_repo; pub mod users_service; pub mod apps_api; pub mod auth; @@ -87,6 +88,9 @@ pub use app_user_session_repo::{ AppUserSessionLookup, AppUserSessionRepository, AppUserSessionRepositoryError, PostgresAppUserSessionRepository, }; +pub use app_user_verification_repo::{ + AppUserVerificationRepo, AppUserVerificationRepoError, PostgresAppUserVerificationRepo, +}; pub use users_service::{UsersServiceConfig, UsersServiceImpl}; pub use admin_users_api::{admins_router, AdminsState}; pub use api::{admin_router, AdminState}; diff --git a/crates/manager-core/src/users_service.rs b/crates/manager-core/src/users_service.rs index eada01f..ece5d70 100644 --- a/crates/manager-core/src/users_service.rs +++ b/crates/manager-core/src/users_service.rs @@ -27,15 +27,17 @@ use std::time::Duration; use async_trait::async_trait; use chrono::Utc; use picloud_shared::{ - AppId, AppUser, AppUserId, CreateUserInput, EmailTemplateOpts, GeneratedAccept, - GeneratedSession, InviteOpts, Invitation, InvitationId, Principal, SdkCallCx, ServiceEvent, - ServiceEventEmitter, UpdateUserInput, UsersError, UsersListOpts, UsersListPage, UsersService, + AppId, AppUser, AppUserId, CreateUserInput, EmailError, EmailService, EmailTemplateOpts, + GeneratedAccept, GeneratedSession, InviteOpts, Invitation, InvitationId, OutboundEmail, + Principal, SdkCallCx, ServiceEvent, ServiceEventEmitter, UpdateUserInput, UsersError, + UsersListOpts, UsersListPage, UsersService, }; use crate::app_user_repo::{ AppUserRepository, AppUserRepositoryError, AppUserRow, ListOpts as RepoListOpts, }; use crate::app_user_session_repo::{AppUserSessionRepository, AppUserSessionRepositoryError}; +use crate::app_user_verification_repo::{AppUserVerificationRepo, AppUserVerificationRepoError}; use crate::auth::{ self, generate_session_token, hash_password, hash_token, verify_password, TIMING_FLAT_DUMMY_HASH, }; @@ -120,6 +122,8 @@ fn env_duration_days(var: &str, default: Duration) -> Duration { pub struct UsersServiceImpl { users: Arc, sessions: Arc, + verifications: Arc, + email: Arc, authz: Arc, events: Arc, config: UsersServiceConfig, @@ -127,9 +131,12 @@ pub struct UsersServiceImpl { impl UsersServiceImpl { #[must_use] + #[allow(clippy::too_many_arguments)] pub fn new( users: Arc, sessions: Arc, + verifications: Arc, + email: Arc, authz: Arc, events: Arc, config: UsersServiceConfig, @@ -137,6 +144,8 @@ impl UsersServiceImpl { Self { users, sessions, + verifications, + email, authz, events, config, @@ -256,6 +265,54 @@ impl From for UsersError { } } +impl From for UsersError { + fn from(e: AppUserVerificationRepoError) -> Self { + match e { + AppUserVerificationRepoError::Db(err) => UsersError::Backend(err.to_string()), + } + } +} + +fn map_email_error(e: EmailError) -> UsersError { + match e { + EmailError::NotConfigured => UsersError::EmailNotConfigured, + EmailError::Forbidden => UsersError::Forbidden, + EmailError::MissingField(_) + | EmailError::InvalidAddress(_) + | EmailError::TooLarge { .. } => UsersError::EmailTransport(e.to_string()), + EmailError::Transport(s) => UsersError::EmailTransport(s), + } +} + +/// Build the token-bearing URL: `link_base?token=` (or `&token=...` +/// if the base already has a query string). The raw token never lives +/// in storage; it appears only here, in the email body, and in the +/// user's inbox. +fn build_link(link_base: &str, raw_token: &str) -> String { + let sep = if link_base.contains('?') { '&' } else { '?' }; + format!("{link_base}{sep}token={raw_token}") +} + +/// Synthesize an `SdkCallCx` with `principal: None` for the internal +/// email send. The users::* surface has already done its own authz +/// check (AppUsersWrite / AppUsersAdmin); the underlying email send is +/// a system-level action, not the user's direct invocation. Mirrors +/// how downstream services would internally hop without re-running +/// the caller's authz. +fn internal_cx(cx: &SdkCallCx) -> SdkCallCx { + SdkCallCx { + app_id: cx.app_id, + principal: None, + script_id: cx.script_id, + execution_id: cx.execution_id, + request_id: cx.request_id, + trigger_depth: cx.trigger_depth, + root_execution_id: cx.root_execution_id, + is_dead_letter_handler: cx.is_dead_letter_handler, + event: cx.event.clone(), + } +} + #[async_trait] impl UsersService for UsersServiceImpl { async fn create( @@ -479,18 +536,64 @@ impl UsersService for UsersServiceImpl { async fn send_verification_email( &self, - _cx: &SdkCallCx, - _user_id: AppUserId, - _opts: EmailTemplateOpts, + cx: &SdkCallCx, + user_id: AppUserId, + opts: EmailTemplateOpts, ) -> Result<(), UsersError> { - Err(UsersError::Backend(NOT_YET_IMPL.into())) + self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id)) + .await?; + let user = self + .users + .get(cx.app_id, user_id) + .await? + .ok_or(UsersError::NotFound)?; + + // Mint a single-use token. The hash hits the verifications + // table; the raw token only ever appears in the email body. + let token = generate_session_token(); + let expires_at = Utc::now() + + chrono::Duration::from_std(self.config.verification_ttl) + .map_err(|e| UsersError::Backend(format!("verification ttl: {e}")))?; + self.verifications + .create(cx.app_id, user.id, &token.hash, expires_at) + .await?; + + let link = build_link(&opts.link_base, &token.raw); + let body = opts.body_template.replace("{link}", &link); + let outbound = OutboundEmail { + to: vec![user.email.clone()], + from: opts.from, + subject: opts.subject, + text: Some(body), + ..Default::default() + }; + let send_cx = internal_cx(cx); + self.email + .send(&send_cx, outbound) + .await + .map_err(map_email_error)?; + Ok(()) } + async fn verify_email( &self, - _cx: &SdkCallCx, - _token: &str, + cx: &SdkCallCx, + token: &str, ) -> Result, UsersError> { - Err(UsersError::Backend(NOT_YET_IMPL.into())) + self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id)) + .await?; + let token_hash = hash_token(token); + let Some(user_id) = self.verifications.consume(cx.app_id, &token_hash).await? else { + return Ok(None); + }; + let row = self + .users + .mark_email_verified(cx.app_id, user_id) + .await?; + let roles = self.fetch_roles(cx.app_id, row.id).await?; + let user = Self::to_app_user(row, roles); + self.emit(cx, "email_verified", user.id).await; + Ok(Some(user)) } async fn request_password_reset( &self, diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index cd05e74..40e2463 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -22,7 +22,8 @@ use picloud_manager_core::{ OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository, PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserRepository, - PostgresAppUserSessionRepository, PostgresDeadLetterRepo, PostgresDeadLetterService, + PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, + PostgresDeadLetterService, PostgresDocsRepo, PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo, PrincipalResolver, @@ -250,9 +251,13 @@ pub async fn build_app( // `users::*` namespace and the admin /apps/{id}/users HTTP surface. let app_users_repo = Arc::new(PostgresAppUserRepository::new(pool.clone())); let app_user_sessions_repo = Arc::new(PostgresAppUserSessionRepository::new(pool.clone())); + let app_user_verifications_repo = + Arc::new(PostgresAppUserVerificationRepo::new(pool.clone())); let users: Arc = Arc::new(UsersServiceImpl::new( app_users_repo.clone(), app_user_sessions_repo.clone(), + app_user_verifications_repo.clone(), + email.clone(), authz.clone(), events.clone(), UsersServiceConfig::from_env(), diff --git a/crates/shared/src/users.rs b/crates/shared/src/users.rs index adab41d..37a2902 100644 --- a/crates/shared/src/users.rs +++ b/crates/shared/src/users.rs @@ -104,9 +104,15 @@ pub struct GeneratedAccept { /// flows (`send_verification_email`, `request_password_reset`, `invite`). /// The script owns the body; PiCloud only injects `?token=` into /// the `link_base`. +/// +/// `from` is required because the underlying `EmailService::send` needs +/// an envelope sender — the brief example omitted it for brevity, but +/// the service can't synthesize a sensible default. Documented as a +/// shape deviation in HANDBACK §7. #[derive(Debug, Clone)] pub struct EmailTemplateOpts { pub link_base: String, + pub from: String, pub subject: String, /// Body template with `{link}` substituted for the /// `link_base?token=` URL. Plain text body — the v1.1.7