adc719975f
cargo fmt regroups; three clippy fixes: - F-S-006: hoist MAX_API_KEY_CANDIDATES to module scope to satisfy clippy::items_after_statements. - F-S-009: use map_or instead of map().unwrap_or() per clippy::map_unwrap_or. - F-P-012: replace `|c| c.encode()` closure with the method itself per clippy::redundant_closure_for_method_calls. No behavior change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1286 lines
47 KiB
Rust
1286 lines
47 KiB
Rust
//! `UsersService` implementation backed by Postgres (v1.1.8).
|
|
//!
|
|
//! Wires the `app_user_repo` (Argon2id-hashed user rows), the
|
|
//! `app_user_session_repo` (SHA-256-hashed sliding-window sessions),
|
|
//! the authz layer (so script-side calls require the right capability
|
|
//! when the principal is non-public), the `ServiceEventEmitter` (so
|
|
//! every mutation publishes an event for future triggers), and the
|
|
//! shared email service (for the verification / reset / invitation
|
|
//! flows in later commits).
|
|
//!
|
|
//! Cross-app isolation lives on every method that takes
|
|
//! `&SdkCallCx`: we read `cx.app_id` and never trust a script-passed
|
|
//! identifier. The `verify_session_for_realtime` method takes
|
|
//! `app_id` explicitly because it's invoked from the realtime
|
|
//! authority (no script execution context); the value comes from the
|
|
//! topic row, also not script-controlled.
|
|
//!
|
|
//! Commit 4 ships CRUD + login/verify/logout + verify_session_for_realtime.
|
|
//! Email-tied flows (verification, password reset, invitations),
|
|
//! roles, and admin-mediated helpers are stubbed with
|
|
//! `UsersError::Backend("not yet implemented")` and ship in later
|
|
//! commits in the v1.1.8 series.
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use async_trait::async_trait;
|
|
use chrono::Utc;
|
|
use picloud_shared::{
|
|
AppId, AppUser, AppUserId, CreateUserInput, EmailError, EmailService, EmailTemplateOpts,
|
|
GeneratedAccept, GeneratedSession, Invitation, InvitationId, InviteOpts, OutboundEmail,
|
|
Principal, SdkCallCx, ServiceEvent, ServiceEventEmitter, UpdateUserInput, UsersError,
|
|
UsersListOpts, UsersListPage, UsersService,
|
|
};
|
|
|
|
use crate::app_user_invitation_repo::{AppUserInvitationRepo, AppUserInvitationRepoError};
|
|
use crate::app_user_password_reset_repo::{
|
|
AppUserPasswordResetRepo, AppUserPasswordResetRepoError,
|
|
};
|
|
use crate::app_user_repo::{
|
|
AppUserRepository, AppUserRepositoryError, AppUserRow, ListOpts as RepoListOpts,
|
|
};
|
|
use crate::app_user_role_repo::{AppUserRoleRepo, AppUserRoleRepoError};
|
|
use crate::app_user_session_repo::{AppUserSessionRepository, AppUserSessionRepositoryError};
|
|
use crate::app_user_verification_repo::{AppUserVerificationRepo, AppUserVerificationRepoError};
|
|
use crate::auth::{
|
|
generate_session_token, hash_password, hash_token, verify_password, TIMING_FLAT_DUMMY_HASH,
|
|
};
|
|
use crate::authz::{self, AuthzDenied, AuthzRepo, Capability};
|
|
|
|
/// Runtime configuration. Defaults match the v1.1.8 brief — overridable
|
|
/// via env vars wired in the picloud binary.
|
|
#[derive(Debug, Clone, Copy)]
|
|
pub struct UsersServiceConfig {
|
|
/// Sliding-window TTL on a session (default 24h).
|
|
pub session_ttl: Duration,
|
|
/// Hard cap from session creation (default 30 days). Beyond this
|
|
/// the session is dead regardless of recent activity.
|
|
pub session_absolute_ttl: Duration,
|
|
/// Verification-token TTL (default 48h).
|
|
pub verification_ttl: Duration,
|
|
/// Password-reset-token TTL (default 1h).
|
|
pub password_reset_ttl: Duration,
|
|
/// Invitation-token TTL (default 7 days).
|
|
pub invitation_ttl: Duration,
|
|
}
|
|
|
|
impl Default for UsersServiceConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
session_ttl: Duration::from_secs(24 * 3600),
|
|
session_absolute_ttl: Duration::from_secs(30 * 24 * 3600),
|
|
verification_ttl: Duration::from_secs(48 * 3600),
|
|
password_reset_ttl: Duration::from_secs(3600),
|
|
invitation_ttl: Duration::from_secs(7 * 24 * 3600),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl UsersServiceConfig {
|
|
/// Resolve all knobs from the process environment. Falls back to
|
|
/// the documented defaults on any parse failure (logging a warn).
|
|
#[must_use]
|
|
pub fn from_env() -> Self {
|
|
let default = Self::default();
|
|
Self {
|
|
session_ttl: env_duration_hours(
|
|
"PICLOUD_APP_USER_SESSION_TTL_HOURS",
|
|
default.session_ttl,
|
|
),
|
|
session_absolute_ttl: env_duration_hours(
|
|
"PICLOUD_APP_USER_SESSION_ABSOLUTE_HOURS",
|
|
default.session_absolute_ttl,
|
|
),
|
|
verification_ttl: env_duration_hours(
|
|
"PICLOUD_APP_USER_VERIFICATION_TTL_HOURS",
|
|
default.verification_ttl,
|
|
),
|
|
password_reset_ttl: env_duration_hours(
|
|
"PICLOUD_APP_USER_PASSWORD_RESET_TTL_HOURS",
|
|
default.password_reset_ttl,
|
|
),
|
|
invitation_ttl: env_duration_days(
|
|
"PICLOUD_APP_USER_INVITATION_TTL_DAYS",
|
|
default.invitation_ttl,
|
|
),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn env_duration_hours(var: &str, default: Duration) -> Duration {
|
|
std::env::var(var)
|
|
.ok()
|
|
.and_then(|s| s.parse::<u64>().ok())
|
|
.map_or(default, |h| Duration::from_secs(h * 3600))
|
|
}
|
|
|
|
fn env_duration_days(var: &str, default: Duration) -> Duration {
|
|
std::env::var(var)
|
|
.ok()
|
|
.and_then(|s| s.parse::<u64>().ok())
|
|
.map_or(default, |d| Duration::from_secs(d * 24 * 3600))
|
|
}
|
|
|
|
/// Postgres-backed `UsersService`.
|
|
/// In-memory rate limiter for outbound-email surfaces (verification +
|
|
/// password reset). Two scopes:
|
|
///
|
|
/// 1. **Per-(app, recipient)**: max `recipient_burst` calls per
|
|
/// `recipient_window` (default 5 per hour). Defends against an
|
|
/// attacker hammering one address.
|
|
/// 2. **Per-app daily cap**: max `app_daily_cap` calls per app per
|
|
/// rolling 24h window (default 200). Defends the SMTP-relay quota.
|
|
///
|
|
/// Both windows reset lazily — buckets that have aged out get rebuilt
|
|
/// on the next call. Stored in a single Mutex; reads + writes are
|
|
/// short, contention is low for the call volume this limits.
|
|
#[derive(Debug)]
|
|
struct EmailRateLimiter {
|
|
recipient_burst: u32,
|
|
recipient_window: Duration,
|
|
app_daily_cap: u32,
|
|
state: std::sync::Mutex<EmailRateState>,
|
|
}
|
|
|
|
#[derive(Debug, Default)]
|
|
struct EmailRateState {
|
|
per_recipient: std::collections::HashMap<(AppId, String), Bucket>,
|
|
per_app: std::collections::HashMap<AppId, Bucket>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy)]
|
|
struct Bucket {
|
|
window_started_at: std::time::Instant,
|
|
count: u32,
|
|
}
|
|
|
|
impl EmailRateLimiter {
|
|
fn new() -> Self {
|
|
Self {
|
|
recipient_burst: 5,
|
|
recipient_window: Duration::from_secs(3600),
|
|
app_daily_cap: 200,
|
|
state: std::sync::Mutex::new(EmailRateState::default()),
|
|
}
|
|
}
|
|
|
|
/// Attempt to consume one slot for `(app, recipient)`. Returns
|
|
/// `Err(())` if either cap is exceeded.
|
|
fn try_consume(&self, app_id: AppId, recipient: &str) -> Result<(), ()> {
|
|
let now = std::time::Instant::now();
|
|
let app_cap_window = Duration::from_secs(24 * 3600);
|
|
let mut state = self.state.lock().expect("email rate state poisoned");
|
|
|
|
// Check + roll the per-app bucket.
|
|
{
|
|
let app_bucket = state.per_app.entry(app_id).or_insert(Bucket {
|
|
window_started_at: now,
|
|
count: 0,
|
|
});
|
|
if now.duration_since(app_bucket.window_started_at) >= app_cap_window {
|
|
*app_bucket = Bucket {
|
|
window_started_at: now,
|
|
count: 0,
|
|
};
|
|
}
|
|
if app_bucket.count >= self.app_daily_cap {
|
|
return Err(());
|
|
}
|
|
}
|
|
|
|
// Check + roll the per-recipient bucket.
|
|
let key = (app_id, recipient.to_string());
|
|
{
|
|
let bucket = state.per_recipient.entry(key).or_insert(Bucket {
|
|
window_started_at: now,
|
|
count: 0,
|
|
});
|
|
if now.duration_since(bucket.window_started_at) >= self.recipient_window {
|
|
*bucket = Bucket {
|
|
window_started_at: now,
|
|
count: 0,
|
|
};
|
|
}
|
|
if bucket.count >= self.recipient_burst {
|
|
return Err(());
|
|
}
|
|
bucket.count += 1;
|
|
}
|
|
// Both checks passed; bump the app counter.
|
|
if let Some(app_bucket) = state.per_app.get_mut(&app_id) {
|
|
app_bucket.count += 1;
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
pub struct UsersServiceImpl {
|
|
users: Arc<dyn AppUserRepository>,
|
|
sessions: Arc<dyn AppUserSessionRepository>,
|
|
verifications: Arc<dyn AppUserVerificationRepo>,
|
|
password_resets: Arc<dyn AppUserPasswordResetRepo>,
|
|
invitations: Arc<dyn AppUserInvitationRepo>,
|
|
roles: Arc<dyn AppUserRoleRepo>,
|
|
email: Arc<dyn EmailService>,
|
|
authz: Arc<dyn AuthzRepo>,
|
|
events: Arc<dyn ServiceEventEmitter>,
|
|
config: UsersServiceConfig,
|
|
email_rate_limiter: EmailRateLimiter,
|
|
}
|
|
|
|
impl UsersServiceImpl {
|
|
#[must_use]
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub fn new(
|
|
users: Arc<dyn AppUserRepository>,
|
|
sessions: Arc<dyn AppUserSessionRepository>,
|
|
verifications: Arc<dyn AppUserVerificationRepo>,
|
|
password_resets: Arc<dyn AppUserPasswordResetRepo>,
|
|
invitations: Arc<dyn AppUserInvitationRepo>,
|
|
roles: Arc<dyn AppUserRoleRepo>,
|
|
email: Arc<dyn EmailService>,
|
|
authz: Arc<dyn AuthzRepo>,
|
|
events: Arc<dyn ServiceEventEmitter>,
|
|
config: UsersServiceConfig,
|
|
) -> Self {
|
|
Self {
|
|
users,
|
|
sessions,
|
|
verifications,
|
|
password_resets,
|
|
invitations,
|
|
roles,
|
|
email,
|
|
authz,
|
|
events,
|
|
config,
|
|
email_rate_limiter: EmailRateLimiter::new(),
|
|
}
|
|
}
|
|
|
|
async fn require(
|
|
&self,
|
|
principal: Option<&Principal>,
|
|
cap: Capability,
|
|
) -> Result<(), UsersError> {
|
|
let Some(p) = principal else {
|
|
// Public HTTP execution — script-as-gate; not denied here.
|
|
return Ok(());
|
|
};
|
|
match authz::require(&*self.authz, p, cap).await {
|
|
Ok(()) => Ok(()),
|
|
Err(AuthzDenied::Denied) => Err(UsersError::Forbidden),
|
|
Err(AuthzDenied::Repo(e)) => Err(UsersError::Backend(e.to_string())),
|
|
}
|
|
}
|
|
|
|
/// Compose the public `AppUser` shape from a `AppUserRow`. Roles
|
|
/// are always empty in commit 4 of v1.1.8; commit 8 (per-app
|
|
/// roles) replaces this stub with an actual `app_user_roles`
|
|
/// query.
|
|
fn to_app_user(row: AppUserRow, roles: Vec<String>) -> AppUser {
|
|
AppUser {
|
|
id: row.id,
|
|
app_id: row.app_id,
|
|
email: row.email,
|
|
display_name: row.display_name,
|
|
email_verified_at: row.email_verified_at,
|
|
last_login_at: row.last_login_at,
|
|
created_at: row.created_at,
|
|
updated_at: row.updated_at,
|
|
roles,
|
|
}
|
|
}
|
|
|
|
async fn fetch_roles(
|
|
&self,
|
|
app_id: AppId,
|
|
user_id: AppUserId,
|
|
) -> Result<Vec<String>, UsersError> {
|
|
Ok(self.roles.list_for_user(app_id, user_id).await?)
|
|
}
|
|
|
|
async fn emit(&self, cx: &SdkCallCx, op: &'static str, user_id: AppUserId) {
|
|
let _ = self
|
|
.events
|
|
.emit(
|
|
cx,
|
|
ServiceEvent {
|
|
source: "users",
|
|
op,
|
|
collection: None,
|
|
key: Some(user_id.to_string()),
|
|
payload: None,
|
|
old_payload: None,
|
|
},
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
|
|
fn validate_email(email: &str) -> Result<String, UsersError> {
|
|
let trimmed = email.trim();
|
|
if trimmed.is_empty() {
|
|
return Err(UsersError::Invalid("email must not be empty".into()));
|
|
}
|
|
if trimmed.len() > 254 {
|
|
return Err(UsersError::Invalid("email exceeds 254 characters".into()));
|
|
}
|
|
if !trimmed.contains('@') {
|
|
return Err(UsersError::Invalid("email must contain '@'".into()));
|
|
}
|
|
Ok(trimmed.to_string())
|
|
}
|
|
|
|
fn validate_password(password: &str) -> Result<(), UsersError> {
|
|
if password.chars().count() < 8 {
|
|
return Err(UsersError::Invalid(
|
|
"password must be at least 8 characters".into(),
|
|
));
|
|
}
|
|
if password.len() > 1024 {
|
|
// Guard against absurd inputs that would make Argon2id costly.
|
|
return Err(UsersError::Invalid("password exceeds 1024 bytes".into()));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn validate_display_name(name: Option<&str>) -> Result<Option<String>, UsersError> {
|
|
let Some(raw) = name else { return Ok(None) };
|
|
let trimmed = raw.trim();
|
|
if trimmed.is_empty() {
|
|
return Ok(None);
|
|
}
|
|
if trimmed.chars().count() > 256 {
|
|
return Err(UsersError::Invalid(
|
|
"display_name exceeds 256 characters".into(),
|
|
));
|
|
}
|
|
Ok(Some(trimmed.to_string()))
|
|
}
|
|
|
|
impl From<AppUserRepositoryError> for UsersError {
|
|
fn from(e: AppUserRepositoryError) -> Self {
|
|
match e {
|
|
AppUserRepositoryError::NotFound(_) => UsersError::NotFound,
|
|
AppUserRepositoryError::DuplicateEmail(_) => UsersError::DuplicateEmail,
|
|
AppUserRepositoryError::Db(err) => UsersError::Backend(err.to_string()),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl From<AppUserSessionRepositoryError> for UsersError {
|
|
fn from(e: AppUserSessionRepositoryError) -> Self {
|
|
match e {
|
|
AppUserSessionRepositoryError::Db(err) => UsersError::Backend(err.to_string()),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl From<AppUserVerificationRepoError> for UsersError {
|
|
fn from(e: AppUserVerificationRepoError) -> Self {
|
|
match e {
|
|
AppUserVerificationRepoError::Db(err) => UsersError::Backend(err.to_string()),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl From<AppUserPasswordResetRepoError> for UsersError {
|
|
fn from(e: AppUserPasswordResetRepoError) -> Self {
|
|
match e {
|
|
AppUserPasswordResetRepoError::Db(err) => UsersError::Backend(err.to_string()),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl From<AppUserInvitationRepoError> for UsersError {
|
|
fn from(e: AppUserInvitationRepoError) -> Self {
|
|
match e {
|
|
AppUserInvitationRepoError::Db(err) => UsersError::Backend(err.to_string()),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl From<AppUserRoleRepoError> for UsersError {
|
|
fn from(e: AppUserRoleRepoError) -> Self {
|
|
match e {
|
|
AppUserRoleRepoError::Db(err) => UsersError::Backend(err.to_string()),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn validate_role(role: &str) -> Result<String, UsersError> {
|
|
let trimmed = role.trim();
|
|
if trimmed.is_empty() {
|
|
return Err(UsersError::Invalid("role must not be empty".into()));
|
|
}
|
|
if trimmed.chars().count() > 128 {
|
|
return Err(UsersError::Invalid("role exceeds 128 characters".into()));
|
|
}
|
|
Ok(trimmed.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=<raw>` (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(&self, cx: &SdkCallCx, input: CreateUserInput) -> Result<AppUser, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
|
|
.await?;
|
|
let email = validate_email(&input.email)?;
|
|
validate_password(&input.password)?;
|
|
let display_name = validate_display_name(input.display_name.as_deref())?;
|
|
let password_hash = hash_password(&input.password)
|
|
.map_err(|e| UsersError::Backend(format!("password hashing failed: {e}")))?;
|
|
let row = self
|
|
.users
|
|
.create(cx.app_id, &email, &password_hash, display_name.as_deref())
|
|
.await?;
|
|
let user_id = row.id;
|
|
let user = Self::to_app_user(row, Vec::new());
|
|
self.emit(cx, "create", user_id).await;
|
|
Ok(user)
|
|
}
|
|
|
|
async fn get(&self, cx: &SdkCallCx, id: AppUserId) -> Result<Option<AppUser>, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
|
|
.await?;
|
|
let Some(row) = self.users.get(cx.app_id, id).await? else {
|
|
return Ok(None);
|
|
};
|
|
let roles = self.fetch_roles(cx.app_id, row.id).await?;
|
|
Ok(Some(Self::to_app_user(row, roles)))
|
|
}
|
|
|
|
async fn find_by_email(
|
|
&self,
|
|
cx: &SdkCallCx,
|
|
email: &str,
|
|
) -> Result<Option<AppUser>, UsersError> {
|
|
// F-S-003: an anonymous public-HTTP script could iterate emails
|
|
// via find_by_email to enumerate registered users (the `require`
|
|
// helper short-circuits on principal == None). Tighten: require
|
|
// an authenticated principal — scripts that need a "does this
|
|
// email exist?" probe must wrap it in their own auth gate.
|
|
if cx.principal.is_none() {
|
|
return Err(UsersError::Forbidden);
|
|
}
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
|
|
.await?;
|
|
let normalized = validate_email(email)?;
|
|
let Some(row) = self.users.find_by_email(cx.app_id, &normalized).await? else {
|
|
return Ok(None);
|
|
};
|
|
let roles = self.fetch_roles(cx.app_id, row.id).await?;
|
|
Ok(Some(Self::to_app_user(row, roles)))
|
|
}
|
|
|
|
async fn update(
|
|
&self,
|
|
cx: &SdkCallCx,
|
|
id: AppUserId,
|
|
patch: UpdateUserInput,
|
|
) -> Result<AppUser, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
|
|
.await?;
|
|
let mut row_opt = self.users.get(cx.app_id, id).await?;
|
|
if let Some(display) = patch.display_name {
|
|
let normalized = match display {
|
|
Some(s) => validate_display_name(Some(s.as_str()))?,
|
|
None => None,
|
|
};
|
|
row_opt = Some(
|
|
self.users
|
|
.update_display_name(cx.app_id, id, normalized.as_deref())
|
|
.await?,
|
|
);
|
|
}
|
|
let row = row_opt.ok_or(UsersError::NotFound)?;
|
|
let roles = self.fetch_roles(cx.app_id, row.id).await?;
|
|
let user = Self::to_app_user(row, roles);
|
|
self.emit(cx, "update", user.id).await;
|
|
Ok(user)
|
|
}
|
|
|
|
async fn delete(&self, cx: &SdkCallCx, id: AppUserId) -> Result<bool, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
|
|
.await?;
|
|
let deleted = self.users.delete(cx.app_id, id).await?;
|
|
if deleted {
|
|
self.emit(cx, "delete", id).await;
|
|
}
|
|
Ok(deleted)
|
|
}
|
|
|
|
async fn list(&self, cx: &SdkCallCx, opts: UsersListOpts) -> Result<UsersListPage, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
|
|
.await?;
|
|
let limit = opts.limit.unwrap_or(50).clamp(1, 500);
|
|
let page = self
|
|
.users
|
|
.list(
|
|
cx.app_id,
|
|
RepoListOpts {
|
|
cursor: opts
|
|
.cursor
|
|
.as_deref()
|
|
.and_then(crate::app_user_repo::ListCursor::decode),
|
|
limit,
|
|
},
|
|
)
|
|
.await?;
|
|
|
|
// F-P-001: one batched roles lookup instead of N per-user
|
|
// queries. With limit=500 this is 1 query vs 501.
|
|
let user_ids: Vec<AppUserId> = page.items.iter().map(|r| r.id).collect();
|
|
let mut roles_by_user = self
|
|
.roles
|
|
.list_for_users(cx.app_id, &user_ids)
|
|
.await
|
|
.map_err(|e| UsersError::Backend(e.to_string()))?;
|
|
let items: Vec<AppUser> = page
|
|
.items
|
|
.into_iter()
|
|
.map(|row| {
|
|
let roles = roles_by_user.remove(&row.id).unwrap_or_default();
|
|
Self::to_app_user(row, roles)
|
|
})
|
|
.collect();
|
|
Ok(UsersListPage {
|
|
items,
|
|
next_cursor: page
|
|
.next_cursor
|
|
.as_ref()
|
|
.map(crate::app_user_repo::ListCursor::encode),
|
|
})
|
|
}
|
|
|
|
async fn login(
|
|
&self,
|
|
cx: &SdkCallCx,
|
|
email: &str,
|
|
password: &str,
|
|
) -> Result<Option<GeneratedSession>, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
|
|
.await?;
|
|
// Both branches normalize the email so an attacker can't probe
|
|
// existence via "" / "x" / very-long-string differences.
|
|
let password_owned = password.to_string();
|
|
let Ok(normalized) = validate_email(email) else {
|
|
// Still run a dummy verify so the timing of an invalid
|
|
// email shape matches the timing of a legitimate miss.
|
|
// F-P-002: off the async worker — Argon2 is CPU-bound.
|
|
let _ = tokio::task::spawn_blocking(move || {
|
|
verify_password(TIMING_FLAT_DUMMY_HASH, &password_owned)
|
|
})
|
|
.await;
|
|
return Ok(None);
|
|
};
|
|
let creds = self
|
|
.users
|
|
.get_credentials_by_email(cx.app_id, &normalized)
|
|
.await?;
|
|
let (hash, user_id, app_id) = match creds {
|
|
Some(c) => (c.password_hash, Some(c.id), Some(c.app_id)),
|
|
None => (TIMING_FLAT_DUMMY_HASH.to_string(), None, None),
|
|
};
|
|
// F-P-002: Argon2id verify off the async worker.
|
|
let password_ok =
|
|
tokio::task::spawn_blocking(move || verify_password(&hash, &password_owned))
|
|
.await
|
|
.map_err(|e| UsersError::Backend(format!("verify spawn_blocking join: {e}")))?;
|
|
let (Some(user_id), Some(app_id)) = (user_id, app_id) else {
|
|
return Ok(None);
|
|
};
|
|
if !password_ok {
|
|
return Ok(None);
|
|
}
|
|
// Cross-app isolation: a credentials row for a different app
|
|
// shouldn't be reachable by find/get_credentials with a fixed
|
|
// app_id, but defense-in-depth here.
|
|
if app_id != cx.app_id {
|
|
return Ok(None);
|
|
}
|
|
|
|
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(cx.app_id, user_id, &token.hash, expires_at, absolute)
|
|
.await?;
|
|
self.users.touch_last_login(cx.app_id, user_id).await?;
|
|
|
|
let row = self
|
|
.users
|
|
.get(cx.app_id, user_id)
|
|
.await?
|
|
.ok_or(UsersError::NotFound)?;
|
|
let roles = self.fetch_roles(cx.app_id, row.id).await?;
|
|
let user = Self::to_app_user(row, roles);
|
|
Ok(Some(GeneratedSession {
|
|
token: token.raw,
|
|
user,
|
|
expires_at,
|
|
}))
|
|
}
|
|
|
|
async fn verify(&self, cx: &SdkCallCx, token: &str) -> Result<Option<AppUser>, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
|
|
.await?;
|
|
self.resolve_session(cx.app_id, token).await
|
|
}
|
|
|
|
async fn logout(&self, cx: &SdkCallCx, token: &str) -> Result<(), UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
|
|
.await?;
|
|
let token_hash = hash_token(token);
|
|
// Only revoke when the row belongs to this app — otherwise the
|
|
// script could prove existence of cross-app sessions by trying
|
|
// arbitrary tokens. Lookup is the cross-app filter here.
|
|
if let Some(lookup) = self.sessions.lookup(&token_hash).await? {
|
|
if lookup.app_id != cx.app_id {
|
|
return Ok(());
|
|
}
|
|
self.sessions.revoke(&token_hash).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn verify_session_for_realtime(
|
|
&self,
|
|
app_id: AppId,
|
|
token: &str,
|
|
) -> Result<Option<AppUser>, UsersError> {
|
|
self.resolve_session(app_id, token).await
|
|
}
|
|
|
|
async fn send_verification_email(
|
|
&self,
|
|
cx: &SdkCallCx,
|
|
user_id: AppUserId,
|
|
opts: EmailTemplateOpts,
|
|
) -> Result<(), UsersError> {
|
|
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)?;
|
|
// F-S-002: token-bucket rate limit keyed on (app_id, recipient).
|
|
// Anonymous public-route callers (cx.principal == None) skip the
|
|
// authz gate above, so without this an attacker could burn the
|
|
// SMTP-relay quota by spamming any public endpoint that calls
|
|
// users::send_verification_email.
|
|
if self
|
|
.email_rate_limiter
|
|
.try_consume(cx.app_id, &user.email)
|
|
.is_err()
|
|
{
|
|
tracing::warn!(
|
|
app_id = %cx.app_id,
|
|
user_id = %user.id,
|
|
"send_verification_email rate-limited"
|
|
);
|
|
return Err(UsersError::EmailRateLimited);
|
|
}
|
|
|
|
// 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,
|
|
) -> Result<Option<AppUser>, UsersError> {
|
|
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,
|
|
cx: &SdkCallCx,
|
|
email: &str,
|
|
opts: EmailTemplateOpts,
|
|
) -> Result<(), UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersWrite(cx.app_id))
|
|
.await?;
|
|
// F-S-004: no existence leak. The wall-clock delta between
|
|
// "user not found" (handler returns immediately) and "user
|
|
// found" (Argon2 token generate + DB INSERT + SMTP RTT) was
|
|
// tens-of-ms — externally observable to any attacker who could
|
|
// measure HTTP latency. Eliminate that observable by running
|
|
// the same expensive work shape on the no-match branch:
|
|
// generate a dummy token, do NOT INSERT, and skip the send.
|
|
// (Skipping the INSERT keeps us honest about not leaking via
|
|
// disk-space side channels either.)
|
|
let Ok(normalized) = validate_email(email) else {
|
|
// Bad email shape — still spend the dummy budget.
|
|
let _ = generate_session_token();
|
|
return Ok(());
|
|
};
|
|
let Some(user) = self.users.find_by_email(cx.app_id, &normalized).await? else {
|
|
// No user — same dummy work as the matching branch.
|
|
let _ = generate_session_token();
|
|
return Ok(());
|
|
};
|
|
|
|
// F-S-002: rate limit per (app, recipient). Return Ok silently
|
|
// on exhaustion so an attacker can't probe whether an email
|
|
// exists by comparing the rate-limit response to the no-match
|
|
// response.
|
|
if self
|
|
.email_rate_limiter
|
|
.try_consume(cx.app_id, &user.email)
|
|
.is_err()
|
|
{
|
|
tracing::warn!(
|
|
app_id = %cx.app_id,
|
|
user_id = %user.id,
|
|
"request_password_reset rate-limited; suppressing to avoid existence leak"
|
|
);
|
|
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,
|
|
) -> Result<Option<AppUser>, UsersError> {
|
|
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,
|
|
cx: &SdkCallCx,
|
|
email: &str,
|
|
opts: InviteOpts,
|
|
) -> Result<(), UsersError> {
|
|
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<String>,
|
|
) -> Result<Option<GeneratedAccept>, UsersError> {
|
|
// 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()),
|
|
};
|
|
// Apply pre-staged roles (commit 8). Each role goes through
|
|
// the same trim + length validation the script's add_role
|
|
// call would; invalid ones are skipped with a warn rather
|
|
// than aborting the whole accept (the invitation was a
|
|
// promise the admin made; we honor as much of it as we can).
|
|
let user_id = row.id;
|
|
let mut applied: Vec<String> = Vec::with_capacity(consumed.roles.len());
|
|
for raw_role in &consumed.roles {
|
|
match validate_role(raw_role) {
|
|
Ok(role) => {
|
|
self.roles.add(cx.app_id, user_id, &role).await?;
|
|
applied.push(role);
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
invalid_role = %raw_role,
|
|
user_id = %user_id,
|
|
app_id = %cx.app_id,
|
|
error = %e,
|
|
"accept_invite: skipping malformed pre-staged role"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
let user = Self::to_app_user(row, applied);
|
|
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,
|
|
cx: &SdkCallCx,
|
|
user_id: AppUserId,
|
|
role: &str,
|
|
) -> Result<(), UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersAdmin(cx.app_id))
|
|
.await?;
|
|
// Ensure the user actually exists in this app — otherwise an
|
|
// FK-failing insert leaks "no such user" via the error shape.
|
|
let exists = self.users.get(cx.app_id, user_id).await?.is_some();
|
|
if !exists {
|
|
return Err(UsersError::NotFound);
|
|
}
|
|
let role = validate_role(role)?;
|
|
self.roles.add(cx.app_id, user_id, &role).await?;
|
|
self.emit(cx, "role_added", user_id).await;
|
|
Ok(())
|
|
}
|
|
async fn remove_role(
|
|
&self,
|
|
cx: &SdkCallCx,
|
|
user_id: AppUserId,
|
|
role: &str,
|
|
) -> Result<bool, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersAdmin(cx.app_id))
|
|
.await?;
|
|
let role = validate_role(role)?;
|
|
let removed = self.roles.remove(cx.app_id, user_id, &role).await?;
|
|
if removed {
|
|
self.emit(cx, "role_removed", user_id).await;
|
|
}
|
|
Ok(removed)
|
|
}
|
|
async fn has_role(
|
|
&self,
|
|
cx: &SdkCallCx,
|
|
user_id: AppUserId,
|
|
role: &str,
|
|
) -> Result<bool, UsersError> {
|
|
self.require(cx.principal.as_ref(), Capability::AppUsersRead(cx.app_id))
|
|
.await?;
|
|
let role = validate_role(role)?;
|
|
Ok(self.roles.has(cx.app_id, user_id, &role).await?)
|
|
}
|
|
async fn admin_reset_password_token(
|
|
&self,
|
|
principal: &Principal,
|
|
app_id: AppId,
|
|
user_id: AppUserId,
|
|
) -> Result<String, UsersError> {
|
|
self.require(Some(principal), Capability::AppUsersAdmin(app_id))
|
|
.await?;
|
|
// Verify the user exists in this app — surfaces 404 cleanly
|
|
// instead of failing on the FK insert.
|
|
let exists = self.users.get(app_id, user_id).await?.is_some();
|
|
if !exists {
|
|
return Err(UsersError::NotFound);
|
|
}
|
|
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(app_id, user_id, &token.hash, expires_at)
|
|
.await?;
|
|
// No email send — the admin pastes the raw token into their
|
|
// own out-of-band reset link. The script-side
|
|
// users::request_password_reset is the email-bearing path.
|
|
Ok(token.raw)
|
|
}
|
|
async fn admin_revoke_all_sessions(
|
|
&self,
|
|
principal: &Principal,
|
|
app_id: AppId,
|
|
user_id: AppUserId,
|
|
) -> Result<u64, UsersError> {
|
|
self.require(Some(principal), Capability::AppUsersAdmin(app_id))
|
|
.await?;
|
|
Ok(self.sessions.revoke_for_user(app_id, user_id).await?)
|
|
}
|
|
async fn admin_create_invitation(
|
|
&self,
|
|
principal: &Principal,
|
|
app_id: AppId,
|
|
email: &str,
|
|
opts: InviteOpts,
|
|
) -> Result<Invitation, UsersError> {
|
|
self.require(Some(principal), Capability::AppUsersAdmin(app_id))
|
|
.await?;
|
|
let normalized = validate_email(email)?;
|
|
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(
|
|
app_id,
|
|
&normalized,
|
|
display_name.as_deref(),
|
|
&opts.roles,
|
|
&token.hash,
|
|
expires_at,
|
|
)
|
|
.await?;
|
|
|
|
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()
|
|
};
|
|
// Admin-issued: no SdkCallCx exists; synthesize a minimal
|
|
// one with the admin's principal so the email service's
|
|
// own authz pass is honored. (For users::invite the cx is
|
|
// the script's; the internal_cx() helper there clears the
|
|
// principal because the users::* gate already fired.
|
|
// Here the admin's principal is the actual caller of the
|
|
// email — keep the gate.)
|
|
let send_cx = SdkCallCx {
|
|
app_id,
|
|
principal: Some(principal.clone()),
|
|
script_id: picloud_shared::ScriptId::new(),
|
|
execution_id: picloud_shared::ExecutionId::new(),
|
|
request_id: picloud_shared::RequestId::new(),
|
|
trigger_depth: 0,
|
|
root_execution_id: picloud_shared::ExecutionId::new(),
|
|
is_dead_letter_handler: false,
|
|
event: None,
|
|
};
|
|
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 = %app_id,
|
|
"admin_create_invitation email failed; row remains valid"
|
|
);
|
|
}
|
|
}
|
|
|
|
Ok(Invitation {
|
|
id: row.id,
|
|
app_id: row.app_id,
|
|
email: row.email,
|
|
display_name: row.display_name,
|
|
roles: row.roles,
|
|
created_at: row.created_at,
|
|
expires_at: row.expires_at,
|
|
})
|
|
}
|
|
|
|
async fn list_invitations(
|
|
&self,
|
|
principal: &Principal,
|
|
app_id: AppId,
|
|
) -> Result<Vec<Invitation>, UsersError> {
|
|
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,
|
|
) -> Result<bool, UsersError> {
|
|
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<GeneratedSession, UsersError> {
|
|
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.
|
|
async fn resolve_session(
|
|
&self,
|
|
app_id: AppId,
|
|
token: &str,
|
|
) -> Result<Option<AppUser>, UsersError> {
|
|
let token_hash = hash_token(token);
|
|
let Some(lookup) = self.sessions.lookup(&token_hash).await? else {
|
|
return Ok(None);
|
|
};
|
|
// Cross-app isolation: a token issued for app A cannot
|
|
// authorize a request into app B.
|
|
if lookup.app_id != app_id {
|
|
return Ok(None);
|
|
}
|
|
|
|
let now = Utc::now();
|
|
let sliding = now
|
|
+ chrono::Duration::from_std(self.config.session_ttl)
|
|
.map_err(|e| UsersError::Backend(format!("sliding ttl: {e}")))?;
|
|
let new_expires = sliding.min(lookup.absolute_expires_at);
|
|
self.sessions.touch(&token_hash, new_expires).await?;
|
|
|
|
let row = self
|
|
.users
|
|
.get(lookup.app_id, lookup.user_id)
|
|
.await?
|
|
.ok_or(UsersError::NotFound)?;
|
|
let roles = self.fetch_roles(lookup.app_id, row.id).await?;
|
|
Ok(Some(Self::to_app_user(row, roles)))
|
|
}
|
|
}
|