feat(v1.1.8): password reset flow (migration 0029 + revokes sessions)

migration 0029: app_user_password_resets table — same shape as
verification (token_hash PK, app_id + user_id FKs, expires_at,
consumed_at). One-shot via atomic UPDATE WHERE consumed_at IS NULL.
Default TTL 1h (shorter than verification's 48h — reset tokens are
higher-risk).

UsersServiceImpl gains password_resets: Arc<dyn AppUserPasswordResetRepo>.

users::request_password_reset(email, opts):
  * Returns Ok(()) regardless of whether the email matched — no
    existence-leak signal in script-land (per brief).
  * Email-not-configured surfaces as NotConfigured so scripts can
    fall back to a synchronous reset path. Other email errors are
    silently swallowed and logged server-side; surfacing them would
    leak which addresses produced a "real send attempted" signal vs
    a no-op.

users::complete_password_reset(token, new_password):
  * Atomically consumes the token, updates the Argon2id hash, and
    revokes EVERY active session for that user (anyone with a stale
    token shouldn't be able to ride out the reset). Emits
    users::password_changed.
  * Returns the user on success, () on bad/expired/already-used.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/manager-core/migrations/0029_app_user_password_resets.sql

@@ -0,0 +1,23 @@
+-- v1.1.8 User Management — password reset tokens.
+--
+-- Identical shape to `app_user_email_verifications`. Same one-shot
+-- semantics via atomic UPDATE WHERE consumed_at IS NULL. Default TTL
+-- is shorter (1h vs 48h) — reset tokens are higher-risk than email
+-- verification (whoever holds them can change the password).
+--
+-- `users::request_password_reset` deliberately returns no signal to
+-- script-land about whether the email matched, so probing this table
+-- via mass-replay isn't a clean enumeration vector. The application
+-- enforces "no existence leak"; this migration is just storage.
+
+CREATE TABLE app_user_password_resets (
+    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_password_resets_user
+    ON app_user_password_resets (app_id, user_id);

crates/manager-core/src/app_user_password_reset_repo.rs

@@ -0,0 +1,111 @@
+//! CRUD over `app_user_password_resets` (v1.1.8 commit 6).
+//!
+//! Same shape and one-shot semantics as
+//! `app_user_verification_repo`. Kept as a separate repo so the
+//! distinct lifecycles (consume + revoke-sessions for reset; consume
+//! + mark-email-verified for verification) stay obvious at the call
+//! site. A future v1.2 cleanup may merge them behind a `kind`
+//! discriminator if the duplication becomes load-bearing.
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use picloud_shared::{AppId, AppUserId};
+use sqlx::PgPool;
+
+#[derive(Debug, thiserror::Error)]
+pub enum AppUserPasswordResetRepoError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+#[async_trait]
+pub trait AppUserPasswordResetRepo: Send + Sync {
+    async fn create(
+        &self,
+        app_id: AppId,
+        user_id: AppUserId,
+        token_hash: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), AppUserPasswordResetRepoError>;
+
+    /// Atomic consume. Returns `Some(user_id)` on success, `None` on
+    /// missing / already-consumed / expired / wrong app.
+    async fn consume(
+        &self,
+        app_id: AppId,
+        token_hash: &str,
+    ) -> Result<Option<AppUserId>, AppUserPasswordResetRepoError>;
+
+    async fn gc(&self, batch_size: i64) -> Result<u64, AppUserPasswordResetRepoError>;
+}
+
+pub struct PostgresAppUserPasswordResetRepo {
+    pool: PgPool,
+}
+
+impl PostgresAppUserPasswordResetRepo {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+#[async_trait]
+impl AppUserPasswordResetRepo for PostgresAppUserPasswordResetRepo {
+    async fn create(
+        &self,
+        app_id: AppId,
+        user_id: AppUserId,
+        token_hash: &str,
+        expires_at: DateTime<Utc>,
+    ) -> Result<(), AppUserPasswordResetRepoError> {
+        sqlx::query(
+            "INSERT INTO app_user_password_resets \
+                 (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<Option<AppUserId>, AppUserPasswordResetRepoError> {
+        let row: Option<(uuid::Uuid,)> = sqlx::query_as(
+            "UPDATE app_user_password_resets \
+             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<u64, AppUserPasswordResetRepoError> {
+        let res = sqlx::query(
+            "DELETE FROM app_user_password_resets WHERE token_hash IN ( \
+                 SELECT token_hash FROM app_user_password_resets \
+                 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())
+    }
+}

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_password_reset_repo;
 pub mod app_user_repo;
 pub mod app_user_session_repo;
 pub mod app_user_verification_repo;
@@ -88,6 +89,9 @@ pub use app_user_session_repo::{
     AppUserSessionLookup, AppUserSessionRepository, AppUserSessionRepositoryError,
     PostgresAppUserSessionRepository,
 };
+pub use app_user_password_reset_repo::{
+    AppUserPasswordResetRepo, AppUserPasswordResetRepoError, PostgresAppUserPasswordResetRepo,
+};
 pub use app_user_verification_repo::{
     AppUserVerificationRepo, AppUserVerificationRepoError, PostgresAppUserVerificationRepo,
 };

crates/manager-core/src/users_service.rs

@@ -33,6 +33,9 @@ use picloud_shared::{
     UsersListOpts, UsersListPage, UsersService,
 };
 
+use crate::app_user_password_reset_repo::{
+    AppUserPasswordResetRepo, AppUserPasswordResetRepoError,
+};
 use crate::app_user_repo::{
     AppUserRepository, AppUserRepositoryError, AppUserRow, ListOpts as RepoListOpts,
 };
@@ -123,6 +126,7 @@ pub struct UsersServiceImpl {
     users: Arc<dyn AppUserRepository>,
     sessions: Arc<dyn AppUserSessionRepository>,
     verifications: Arc<dyn AppUserVerificationRepo>,
+    password_resets: Arc<dyn AppUserPasswordResetRepo>,
     email: Arc<dyn EmailService>,
     authz: Arc<dyn AuthzRepo>,
     events: Arc<dyn ServiceEventEmitter>,
@@ -136,6 +140,7 @@ impl UsersServiceImpl {
         users: Arc<dyn AppUserRepository>,
         sessions: Arc<dyn AppUserSessionRepository>,
         verifications: Arc<dyn AppUserVerificationRepo>,
+        password_resets: Arc<dyn AppUserPasswordResetRepo>,
         email: Arc<dyn EmailService>,
         authz: Arc<dyn AuthzRepo>,
         events: Arc<dyn ServiceEventEmitter>,
@@ -145,6 +150,7 @@ impl UsersServiceImpl {
             users,
             sessions,
             verifications,
+            password_resets,
             email,
             authz,
             events,
@@ -273,6 +279,14 @@ impl From<AppUserVerificationRepoError> for UsersError {
     }
 }
 
+impl From<AppUserPasswordResetRepoError> for UsersError {
+    fn from(e: AppUserPasswordResetRepoError) -> Self {
+        match e {
+            AppUserPasswordResetRepoError::Db(err) => UsersError::Backend(err.to_string()),
+        }
+    }
+}
+
 fn map_email_error(e: EmailError) -> UsersError {
     match e {
         EmailError::NotConfigured => UsersError::EmailNotConfigured,
@@ -597,19 +611,90 @@ impl UsersService for UsersServiceImpl {
     }
     async fn request_password_reset(
         &self,
-        _cx: &SdkCallCx,
-        _email: &str,
-        _opts: EmailTemplateOpts,
+        cx: &SdkCallCx,
+        email: &str,
+        opts: EmailTemplateOpts,
     ) -> Result<(), UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
+            .await?;
+        // No existence leak: an invalid email shape or a missing user
+        // both return Ok(()) silently. The script-side script cannot
+        // tell the difference between "user exists" and "user
+        // doesn't exist" — the only observable is whether an email
+        // actually arrives, which the script can't see.
+        let Ok(normalized) = validate_email(email) else {
+            return Ok(());
+        };
+        let Some(user) = self.users.find_by_email(cx.app_id, &normalized).await? else {
+            return Ok(());
+        };
+
+        let token = generate_session_token();
+        let expires_at = Utc::now()
+            + chrono::Duration::from_std(self.config.password_reset_ttl)
+                .map_err(|e| UsersError::Backend(format!("reset ttl: {e}")))?;
+        self.password_resets
+            .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);
+        // Email-not-configured isn't a security signal — it's the
+        // operator's known state. Surface it to script-land so the
+        // script can fall back to a sync reset flow. But: never
+        // surface email send failures (TooLarge/InvalidAddress/
+        // Transport) as anything other than a generic "email failed"
+        // — those still could be used to probe whether the user
+        // existed; clamp the visible variants.
+        match self.email.send(&send_cx, outbound).await {
+            Ok(()) => Ok(()),
+            Err(EmailError::NotConfigured) => Err(UsersError::EmailNotConfigured),
+            Err(e) => {
+                tracing::warn!(
+                    error = %e,
+                    user_id = %user.id,
+                    app_id = %cx.app_id,
+                    "request_password_reset: email send failed; suppressing to avoid existence leak"
+                );
+                Ok(())
+            }
+        }
     }
+
     async fn complete_password_reset(
         &self,
-        _cx: &SdkCallCx,
-        _token: &str,
-        _new_password: &str,
+        cx: &SdkCallCx,
+        token: &str,
+        new_password: &str,
     ) -> Result<Option<AppUser>, UsersError> {
-        Err(UsersError::Backend(NOT_YET_IMPL.into()))
+        self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
+            .await?;
+        validate_password(new_password)?;
+        let token_hash = hash_token(token);
+        let Some(user_id) = self.password_resets.consume(cx.app_id, &token_hash).await? else {
+            return Ok(None);
+        };
+        let password_hash = hash_password(new_password)
+            .map_err(|e| UsersError::Backend(format!("password hashing failed: {e}")))?;
+        let row = self
+            .users
+            .update_password_hash(cx.app_id, user_id, &password_hash)
+            .await?;
+        // Revoke every active session for the user — anyone with a
+        // stale token shouldn't be able to ride out the reset.
+        let _ = self.sessions.revoke_for_user(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, "password_changed", user.id).await;
+        Ok(Some(user))
     }
     async fn invite(
         &self,

crates/picloud/src/lib.rs

@@ -21,9 +21,9 @@ use picloud_manager_core::{
     FilesServiceImpl, FsFilesRepo, HttpConfig, HttpServiceImpl, KvServiceImpl, OutboxEventEmitter,
     OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository,
     PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository,
-    PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserRepository,
-    PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo,
-    PostgresDeadLetterService,
+    PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserPasswordResetRepo,
+    PostgresAppUserRepository, PostgresAppUserSessionRepository,
+    PostgresAppUserVerificationRepo, PostgresDeadLetterRepo, PostgresDeadLetterService,
     PostgresDocsRepo, PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresKvRepo,
     PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository,
     PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo, PrincipalResolver,
@@ -253,10 +253,13 @@ pub async fn build_app(
     let app_user_sessions_repo = Arc::new(PostgresAppUserSessionRepository::new(pool.clone()));
     let app_user_verifications_repo =
         Arc::new(PostgresAppUserVerificationRepo::new(pool.clone()));
+    let app_user_password_resets_repo =
+        Arc::new(PostgresAppUserPasswordResetRepo::new(pool.clone()));
     let users: Arc<dyn UsersService> = Arc::new(UsersServiceImpl::new(
         app_users_repo.clone(),
         app_user_sessions_repo.clone(),
         app_user_verifications_repo.clone(),
+        app_user_password_resets_repo.clone(),
         email.clone(),
         authz.clone(),
         events.clone(),