diff --git a/migrations/068_team_enrollment.sql b/migrations/068_team_enrollment.sql new file mode 100644 index 0000000..1a1de30 --- /dev/null +++ b/migrations/068_team_enrollment.sql @@ -0,0 +1,53 @@ +ALTER TABLE users ADD COLUMN IF NOT EXISTS email_proven_at TIMESTAMPTZ NULL; + +CREATE OR REPLACE FUNCTION users_email_change_clears_proof() RETURNS trigger AS $$ +BEGIN + NEW.email_proven_at := NULL; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS users_email_change_clears_proof ON users; +CREATE TRIGGER users_email_change_clears_proof + BEFORE UPDATE OF email ON users + FOR EACH ROW WHEN (OLD.email IS DISTINCT FROM NEW.email) + EXECUTE FUNCTION users_email_change_clears_proof(); + +CREATE TABLE IF NOT EXISTS team_enrollment_tokens ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + team_id UUID NOT NULL REFERENCES teams(id) ON DELETE CASCADE, + secret_hash BYTEA NOT NULL UNIQUE, + role TEXT NOT NULL DEFAULT 'member', + allowed_domains TEXT[] NOT NULL DEFAULT '{}', + require_approval BOOLEAN NOT NULL, + expires_at TIMESTAMPTZ NULL, + created_by UUID NULL REFERENCES users(id) ON DELETE SET NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + revoked_at TIMESTAMPTZ NULL, + last_used_at TIMESTAMPTZ NULL, + CONSTRAINT team_enrollment_tokens_guard CHECK (cardinality(allowed_domains) > 0 OR require_approval) +); + +CREATE UNIQUE INDEX IF NOT EXISTS team_enrollment_tokens_one_active + ON team_enrollment_tokens (team_id) WHERE revoked_at IS NULL; + +CREATE TABLE IF NOT EXISTS team_join_requests ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + team_id UUID NOT NULL REFERENCES teams(id) ON DELETE CASCADE, + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + token_id UUID NULL REFERENCES team_enrollment_tokens(id) ON DELETE SET NULL, + status TEXT NOT NULL CHECK (status IN ('pending', 'approved', 'denied')), + reason TEXT NULL CHECK (reason IN ('approval', 'email_unverified', 'seat_limit')), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + decided_by UUID NULL REFERENCES users(id) ON DELETE SET NULL, + decided_at TIMESTAMPTZ NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS team_join_requests_one_pending + ON team_join_requests (team_id, user_id) WHERE status = 'pending'; + +CREATE INDEX IF NOT EXISTS team_join_requests_pending_user + ON team_join_requests (user_id) WHERE status = 'pending'; + +CREATE INDEX IF NOT EXISTS team_join_requests_token + ON team_join_requests (token_id); diff --git a/src/db.rs b/src/db.rs index 7b4ed73..956d836 100644 --- a/src/db.rs +++ b/src/db.rs @@ -163,4 +163,17 @@ mod tests { lenient.run(&mut *tx).await.unwrap(); tx.rollback().await.unwrap(); } + + #[tokio::test] + async fn every_foreign_key_to_users_has_a_delete_rule() { + let pool = test_pool_or_skip!(); + let without_rule: Vec = sqlx::query_scalar( + "SELECT conrelid::regclass::text || '.' || conname FROM pg_constraint \ + WHERE contype = 'f' AND confrelid = 'users'::regclass AND confdeltype = 'a' ORDER BY 1", + ) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(without_rule, Vec::::new()); + } } diff --git a/src/enrollment.rs b/src/enrollment.rs new file mode 100644 index 0000000..8d98c3a --- /dev/null +++ b/src/enrollment.rs @@ -0,0 +1,502 @@ +use axum::http::StatusCode; +use base64::Engine; +use rand::RngCore; +use serde::ser::{Serialize, SerializeStruct, Serializer}; +use serde_json::json; +use sqlx::{PgConnection, PgPool}; +use tracing::{error, warn}; +use uuid::Uuid; + +use crate::permissions::PERM_INVITE_MEMBERS; +use crate::routes::audit::write_audit_event; +use crate::routes::invitations::{admit_member, notify_member_added}; +use crate::routes::team_enrollment::{EnrollError, TokenRow}; +use crate::routes::team_grants::ensure_public_key; +use crate::routes::teams::ensure_seat_available; +use crate::sync_notifier::{notify_team_permission_holders, SyncNotifier}; +use crate::team_authority::lock_billing_owner; +use crate::team_plan::team_locked; + +const DEFAULT_ROLE: &str = "member"; + +pub(crate) fn role_or_default(role: Option<&str>) -> &str { + role.unwrap_or(DEFAULT_ROLE) +} + +pub fn mint_secret() -> String { + let mut bytes = [0u8; 32]; + rand::thread_rng().fill_bytes(&mut bytes); + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PendingReason { + Approval, + EmailUnverified, + SeatLimit, +} + +impl PendingReason { + pub fn as_str(self) -> &'static str { + match self { + PendingReason::Approval => "approval", + PendingReason::EmailUnverified => "email_unverified", + PendingReason::SeatLimit => "seat_limit", + } + } + + fn from_db(value: &str) -> Option { + [PendingReason::Approval, PendingReason::EmailUnverified, PendingReason::SeatLimit] + .into_iter() + .find(|reason| reason.as_str() == value) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Admission { + Admit, + Pending(PendingReason), + DomainMismatch, +} + +#[derive(Debug, Clone)] +pub struct Guard { + pub allowed_domains: Vec, + pub require_approval: bool, +} + +pub fn domain_allowed(guard: &Guard, email: &str) -> bool { + if guard.allowed_domains.is_empty() { + return true; + } + email + .rsplit_once('@') + .map(|(_, d)| d.to_lowercase()) + .is_some_and(|d| guard.allowed_domains.contains(&d)) +} + +pub fn admission(guard: &Guard, email: &str, email_proven: bool, seat_available: bool) -> Admission { + if !domain_allowed(guard, email) { + return Admission::DomainMismatch; + } + if guard.require_approval { + let proof_missing = !guard.allowed_domains.is_empty() && !email_proven; + return Admission::Pending(if proof_missing { PendingReason::EmailUnverified } else { PendingReason::Approval }); + } + if !email_proven { + return Admission::Pending(PendingReason::EmailUnverified); + } + if !seat_available { + return Admission::Pending(PendingReason::SeatLimit); + } + Admission::Admit +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RedeemOutcome { + Member, + Admitted, + Pending(PendingReason), +} + +impl Serialize for RedeemOutcome { + fn serialize(&self, serializer: S) -> Result { + let (status, reason) = match self { + RedeemOutcome::Member => ("member", None), + RedeemOutcome::Admitted => ("admitted", None), + RedeemOutcome::Pending(reason) => ("pending", Some(reason.as_str())), + }; + let mut out = serializer.serialize_struct("RedeemOutcome", 1 + usize::from(reason.is_some()))?; + out.serialize_field("status", status)?; + if let Some(reason) = reason { + out.serialize_field("reason", reason)?; + } + out.end() + } +} + +#[derive(sqlx::FromRow)] +struct PendingRequest { + user_id: Uuid, + token_id: Option, +} + +#[derive(sqlx::FromRow)] +struct TokenState { + role: String, + allowed_domains: Vec, + require_approval: bool, + created_by: Option, + live: bool, +} + +pub(crate) async fn audit_request(pool: &PgPool, team_id: Uuid, actor: Uuid, action: &str, request_id: Uuid, user_id: Uuid) { + write_audit_event( + pool.clone(), + team_id, + actor, + action, + Some("join_request"), + Some(request_id.to_string()), + None, + Some(json!({ "user_id": user_id })), + ) + .await; +} + +pub(crate) async fn notify_join_requests_changed(pool: &PgPool, notifier: &SyncNotifier, team_id: Uuid) { + notify_team_permission_holders(pool, notifier, team_id, PERM_INVITE_MEMBERS, format!("join_requests:{team_id}")).await; +} + +pub(crate) async fn decide_pending( + conn: &mut PgConnection, + request_id: Uuid, + team_id: Uuid, + status: &str, + decided_by: Option, +) -> Result, sqlx::Error> { + sqlx::query_scalar( + "UPDATE team_join_requests SET status = $3, reason = NULL, decided_by = $4, decided_at = now() \ + WHERE id = $1 AND team_id = $2 AND status = 'pending' RETURNING user_id", + ) + .bind(request_id) + .bind(team_id) + .bind(status) + .bind(decided_by) + .fetch_optional(conn) + .await +} + +async fn is_member(conn: &mut PgConnection, team_id: Uuid, user_id: Uuid) -> Result { + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM team_members WHERE team_id = $1 AND user_id = $2)") + .bind(team_id) + .bind(user_id) + .fetch_one(conn) + .await +} + +async fn user_identity(conn: &mut PgConnection, user_id: Uuid) -> Result<(String, bool), sqlx::Error> { + sqlx::query_as("SELECT email, email_proven_at IS NOT NULL FROM users WHERE id = $1") + .bind(user_id) + .fetch_one(conn) + .await +} + +async fn seat_available(pool: &PgPool, owner_id: Uuid, user_id: Uuid) -> Result { + match ensure_seat_available(pool, owner_id, Some(user_id)).await { + Ok(()) => Ok(true), + Err(StatusCode::PAYMENT_REQUIRED) => Ok(false), + Err(status) => Err(status.into()), + } +} + +fn proof_available(guard: &Guard, proven: bool) -> bool { + proven || (guard.require_approval && !crate::email::delivery_configured()) +} + +pub(crate) async fn redeem_for( + pool: &PgPool, + notifier: &SyncNotifier, + token: &TokenRow, + user_id: Uuid, + public_key: Option<&str>, +) -> Result { + let mut conn = pool.acquire().await?; + if is_member(&mut conn, token.team_id, user_id).await? { + return Ok(RedeemOutcome::Member); + } + let (email, proven) = user_identity(&mut conn, user_id).await?; + let guard = token.guard(); + let first = admission(&guard, &email, proof_available(&guard, proven), true); + let reason = match first { + Admission::DomainMismatch => return Err(EnrollError::DomainMismatch), + Admission::Pending(reason) => reason, + Admission::Admit => PendingReason::Approval, + }; + ensure_public_key(&mut conn, user_id, public_key).await?; + let (request_id, inserted, prior) = sqlx::query_as::<_, (Uuid, bool, Option)>( + "WITH prior AS (SELECT reason FROM team_join_requests WHERE team_id = $1 AND user_id = $2 AND status = 'pending') \ + INSERT INTO team_join_requests (team_id, user_id, token_id, status, reason) VALUES ($1, $2, $3, 'pending', $4) \ + ON CONFLICT (team_id, user_id) WHERE status = 'pending' \ + DO UPDATE SET reason = EXCLUDED.reason, token_id = EXCLUDED.token_id \ + RETURNING id, (xmax = 0), (SELECT reason FROM prior)", + ) + .bind(token.team_id) + .bind(user_id) + .bind(token.id) + .bind(reason.as_str()) + .fetch_one(&mut *conn) + .await?; + sqlx::query("UPDATE team_enrollment_tokens SET last_used_at = now() WHERE id = $1") + .bind(token.id) + .execute(&mut *conn) + .await?; + drop(conn); + + let outcome = match first { + Admission::Pending(reason) => RedeemOutcome::Pending(reason), + _ => match try_admit(pool, notifier, request_id, None).await { + Err(EnrollError::NotFound) => settled_state(pool, token.team_id, user_id).await?, + other => other?, + }, + }; + if let RedeemOutcome::Pending(reason) = outcome { + if inserted { + audit_request(pool, token.team_id, user_id, "join_request.created", request_id, user_id).await; + } + if prior.as_deref() != Some(reason.as_str()) { + notify_join_requests_changed(pool, notifier, token.team_id).await; + } + } + Ok(outcome) +} + +async fn settled_state(pool: &PgPool, team_id: Uuid, user_id: Uuid) -> Result { + let mut conn = pool.acquire().await?; + if is_member(&mut conn, team_id, user_id).await? { + return Ok(RedeemOutcome::Member); + } + let reason = sqlx::query_scalar::<_, Option>( + "SELECT reason FROM team_join_requests WHERE team_id = $1 AND user_id = $2 AND status = 'pending'", + ) + .bind(team_id) + .bind(user_id) + .fetch_optional(&mut *conn) + .await?; + reason + .flatten() + .as_deref() + .and_then(PendingReason::from_db) + .map(RedeemOutcome::Pending) + .ok_or(EnrollError::NotFound) +} + +pub(crate) async fn try_admit( + pool: &PgPool, + notifier: &SyncNotifier, + request_id: Uuid, + approved_by: Option, +) -> Result { + let mut tx = pool.begin().await?; + let team_id = sqlx::query_scalar::<_, Uuid>("SELECT team_id FROM team_join_requests WHERE id = $1 AND status = 'pending'") + .bind(request_id) + .fetch_optional(&mut *tx) + .await? + .ok_or(EnrollError::NotFound)?; + let owner_id = lock_billing_owner(&mut tx, team_id).await?; + let PendingRequest { user_id, token_id } = sqlx::query_as::<_, PendingRequest>( + "SELECT user_id, token_id FROM team_join_requests WHERE id = $1 AND team_id = $2 AND status = 'pending' FOR UPDATE", + ) + .bind(request_id) + .bind(team_id) + .fetch_optional(&mut *tx) + .await? + .ok_or(EnrollError::NotFound)?; + if team_locked(pool, team_id).await? { + return Err(EnrollError::Unavailable); + } + + if is_member(&mut tx, team_id, user_id).await? { + decide_pending(&mut tx, request_id, team_id, "approved", approved_by).await?; + tx.commit().await?; + notify_join_requests_changed(pool, notifier, team_id).await; + return Ok(RedeemOutcome::Member); + } + + let token = sqlx::query_as::<_, TokenState>( + "SELECT role, allowed_domains, require_approval, created_by, \ + (revoked_at IS NULL AND COALESCE(expires_at > now(), true)) AS live \ + FROM team_enrollment_tokens WHERE id = $1", + ) + .bind(token_id) + .fetch_optional(&mut *tx) + .await?; + let (email, proven) = user_identity(&mut tx, user_id).await?; + let (guard, proven) = match (&token, approved_by) { + (token, Some(_)) => { + let allowed_domains = token.as_ref().map(|t| t.allowed_domains.clone()).unwrap_or_default(); + let needs_proof = !allowed_domains.is_empty() && crate::email::delivery_configured(); + (Guard { allowed_domains, require_approval: false }, proven || !needs_proof) + } + (Some(token), None) if token.live => { + let guard = Guard { allowed_domains: token.allowed_domains.clone(), require_approval: token.require_approval }; + let proven = proof_available(&guard, proven); + (guard, proven) + } + (_, None) => (Guard { allowed_domains: Vec::new(), require_approval: true }, proven), + }; + + match admission(&guard, &email, proven, seat_available(pool, owner_id, user_id).await?) { + Admission::DomainMismatch => Err(EnrollError::DomainMismatch), + Admission::Pending(reason) => { + let changed = sqlx::query("UPDATE team_join_requests SET reason = $2 WHERE id = $1 AND reason IS DISTINCT FROM $2") + .bind(request_id) + .bind(reason.as_str()) + .execute(&mut *tx) + .await? + .rows_affected() + > 0; + tx.commit().await?; + if changed { + notify_join_requests_changed(pool, notifier, team_id).await; + } + if approved_by.is_some() && reason == PendingReason::SeatLimit { + return Err(EnrollError::NoSeat); + } + Ok(RedeemOutcome::Pending(reason)) + } + Admission::Admit => { + let role = role_or_default(token.as_ref().map(|t| t.role.as_str())); + admit_member(&mut tx, team_id, user_id, token.as_ref().and_then(|t| t.created_by), role, &[], None).await?; + decide_pending(&mut tx, request_id, team_id, "approved", approved_by).await?; + tx.commit().await?; + + if let Some(admin) = approved_by { + audit_request(pool, team_id, admin, "join_request.approved", request_id, user_id).await; + } + write_audit_event( + pool.clone(), + team_id, + approved_by.unwrap_or(user_id), + "member.admitted", + Some("user"), + Some(user_id.to_string()), + None, + Some(json!({ "via": "enrollment" })), + ) + .await; + notify_member_added(pool, notifier, team_id, user_id).await; + notify_join_requests_changed(pool, notifier, team_id).await; + Ok(RedeemOutcome::Admitted) + } + } +} + +pub(crate) async fn admit_verified_requests(pool: &PgPool, notifier: &SyncNotifier, user_id: Uuid) { + let requests = sqlx::query_scalar::<_, Uuid>( + "SELECT id FROM team_join_requests WHERE user_id = $1 AND status = 'pending' AND reason = 'email_unverified'", + ) + .bind(user_id) + .fetch_all(pool) + .await + .unwrap_or_else(|e| { + error!(error = %e, user_id = %user_id, "Failed to list join requests awaiting verification"); + Vec::new() + }); + for request_id in requests { + match try_admit(pool, notifier, request_id, None).await { + Ok(_) => {} + Err(err @ (EnrollError::DomainMismatch | EnrollError::NotFound | EnrollError::Unavailable)) => { + warn!(?err, request_id = %request_id, "Join request not admitted after email verification"); + } + Err(err) => error!(?err, request_id = %request_id, "Failed to admit a join request after email verification"), + } + } +} + +fn is_hostname(domain: &str) -> bool { + domain.len() <= 253 + && domain.split('.').count() >= 2 + && domain.split('.').all(|label| { + (1..=63).contains(&label.len()) + && !label.starts_with('-') + && !label.ends_with('-') + && label.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') + }) +} + +pub fn normalize_domains(input: &[String]) -> Result, ()> { + let mut out: Vec = Vec::new(); + for raw in input { + let domain = raw.trim().to_lowercase(); + let domain = domain.strip_prefix('@').unwrap_or(&domain).to_string(); + if !is_hostname(&domain) { + return Err(()); + } + if !out.contains(&domain) { + out.push(domain); + } + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn guard(domains: &[&str], approval: bool) -> Guard { + Guard { allowed_domains: domains.iter().map(|d| d.to_string()).collect(), require_approval: approval } + } + + #[test] + fn domain_guard_admits_only_exact_proven_matches_with_a_seat() { + let g = guard(&["acme.com"], false); + assert_eq!(admission(&g, "a@acme.com", true, true), Admission::Admit); + assert_eq!(admission(&g, "A@ACME.COM", true, true), Admission::Admit); + assert_eq!(admission(&g, "a@acme.com", false, true), Admission::Pending(PendingReason::EmailUnverified)); + assert_eq!(admission(&g, "a@acme.com", true, false), Admission::Pending(PendingReason::SeatLimit)); + assert_eq!(admission(&g, "a@gmail.com", true, true), Admission::DomainMismatch); + assert_eq!(admission(&g, "a@sub.acme.com", true, true), Admission::DomainMismatch); + } + + #[test] + fn the_domain_is_the_part_after_the_last_at() { + let g = guard(&["acme.com"], false); + assert_eq!(admission(&g, "noat", true, true), Admission::DomainMismatch); + assert_eq!(admission(&g, "a@b@acme.com", true, true), Admission::Admit); + } + + #[test] + fn approval_only_guard_makes_everyone_pending() { + let g = guard(&[], true); + assert_eq!(admission(&g, "a@gmail.com", true, true), Admission::Pending(PendingReason::Approval)); + } + + #[test] + fn domain_mismatch_wins_over_approval() { + let g = guard(&["acme.com"], true); + assert_eq!(admission(&g, "a@gmail.com", true, true), Admission::DomainMismatch); + assert_eq!(admission(&g, "a@acme.com", true, true), Admission::Pending(PendingReason::Approval)); + assert_eq!(admission(&g, "a@acme.com", false, true), Admission::Pending(PendingReason::EmailUnverified)); + } + + #[test] + fn pending_reasons_match_the_database_values() { + for (reason, value) in [ + (PendingReason::Approval, "approval"), + (PendingReason::EmailUnverified, "email_unverified"), + (PendingReason::SeatLimit, "seat_limit"), + ] { + assert_eq!(reason.as_str(), value); + assert_eq!(PendingReason::from_db(value), Some(reason)); + } + assert_eq!(PendingReason::from_db("other"), None); + } + + #[test] + fn normalize_domains_cleans_and_dedupes() { + let input = vec![" @Acme.COM ".to_string(), "acme.com".to_string()]; + assert_eq!(normalize_domains(&input), Ok(vec!["acme.com".to_string()])); + } + + #[test] + fn normalize_domains_rejects_non_hostnames() { + for bad in ["acme", "a b.com", "", "-a.com", "a-.com", "a..com", "a.com/x"] { + assert!(normalize_domains(&[bad.to_string()]).is_err(), "{bad}"); + } + assert!(normalize_domains(&[format!("{}.com", "a".repeat(64))]).is_err()); + } + + #[test] + fn normalize_domains_accepts_punycode_and_rejects_a_trailing_dot() { + assert_eq!(normalize_domains(&["xn--bcher-kva.example".to_string()]), Ok(vec!["xn--bcher-kva.example".to_string()])); + assert!(normalize_domains(&["acme.com.".to_string()]).is_err()); + } + + #[test] + fn mint_secret_is_43_url_safe_chars() { + let s = mint_secret(); + assert_eq!(s.len(), 43); + assert_ne!(s, mint_secret()); + } +} diff --git a/src/features.rs b/src/features.rs index 1d7b51d..492d0ff 100644 --- a/src/features.rs +++ b/src/features.rs @@ -4,8 +4,8 @@ use axum::{ extract::Request, http::StatusCode, middleware::Next, - response::{IntoResponse, Response}, - Extension, Json, + response::Response, + Extension, }; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -24,6 +24,13 @@ impl Features { } } + pub(crate) fn registration_open(&self) -> Result<(), crate::routes::ApiError> { + if self.registration { + return Ok(()); + } + Err(crate::routes::ApiError::coded(StatusCode::FORBIDDEN, "REGISTRATION_DISABLED")) + } + #[cfg(test)] pub fn open() -> Self { Self { registration: true, team_invites: true, handles_from_email: false } @@ -49,15 +56,7 @@ async fn deny_unless(enabled: bool, code: &'static str, req: Request, next: Next if enabled { return next.run(req).await; } - (StatusCode::FORBIDDEN, Json(serde_json::json!({ "error": code }))).into_response() -} - -pub async fn require_registration( - Extension(features): Extension, - req: Request, - next: Next, -) -> Response { - deny_unless(features.registration, "REGISTRATION_DISABLED", req, next).await + crate::routes::coded_error(StatusCode::FORBIDDEN, code) } pub async fn require_team_invites( @@ -86,9 +85,16 @@ mod tests { StatusCode::OK } + async fn register_handler( + Extension(features): Extension, + ) -> Result { + features.registration_open()?; + Ok(StatusCode::OK) + } + async fn call(features: Features) -> (StatusCode, StatusCode) { let app = Router::new() - .route("/register", post(ok_handler).layer(from_fn(require_registration))) + .route("/register", post(register_handler)) .route("/invite", post(ok_handler).layer(from_fn(require_team_invites))) .layer(Extension(features)); let mut statuses = Vec::new(); @@ -141,8 +147,7 @@ mod tests { #[tokio::test] async fn refusal_names_the_disabled_feature() { let app = Router::new() - .route("/register", post(ok_handler)) - .layer(from_fn(require_registration)) + .route("/register", post(register_handler)) .layer(Extension(Features { registration: false, ..Features::open() })); let resp = app .oneshot(Request::post("/register").body(Body::empty()).unwrap()) diff --git a/src/main.rs b/src/main.rs index 76243e9..08c8ec0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,7 @@ mod auth; mod db; mod email; +mod enrollment; mod entitlement; mod features; mod handles; @@ -35,8 +36,8 @@ use axum::{ }; use dashmap::{DashMap, DashSet}; use rate_limit::{ - GrantMintRateLimiter, GrantRedeemRateLimiter, InviteRateLimiter, KnockRateLimiter, RateLimiter, - RedeemRateLimiter, RegisterRateLimiter, SearchRateLimiter, SessionCodeRateLimiter, + EnrollmentRedeemRateLimiter, EnrollmentRegisterLimiter, EnrollmentResolveLimiter, GrantMintRateLimiter, GrantRedeemRateLimiter, InviteRateLimiter, KnockRateLimiter, RateLimiter, + RedeemRateLimiter, RegisterRateLimiter, ResendVerificationRateLimiter, SearchRateLimiter, SessionCodeRateLimiter, SyncRateLimiter, WaitlistRateLimiter, }; use routes::audit::AuditClientRateLimiter; @@ -58,6 +59,42 @@ use tower_http::trace::{DefaultMakeSpan, DefaultOnFailure, DefaultOnResponse, Tr use tracing::Level; use tracing_subscriber::EnvFilter; +fn signup_routes( + notifier: sync_notifier::SyncNotifier, + auth_limiter: RateLimiter, + register_limiter: RegisterRateLimiter, + resolve_limiter: EnrollmentResolveLimiter, + token_limiter: EnrollmentRegisterLimiter, +) -> Router { + Router::new() + .route("/v1/auth/register", post(routes::auth::register)) + .route( + "/v1/enrollment/resolve", + post(routes::team_enrollment::resolve).layer(middleware::from_fn(features::require_team_invites)), + ) + .layer(Extension(notifier)) + .layer(Extension(auth_limiter)) + .layer(Extension(register_limiter)) + .layer(Extension(resolve_limiter)) + .layer(Extension(token_limiter)) + .layer(middleware::from_fn(rate_limit::attach_client_ip)) +} + +fn enrollment_routes() -> Router { + Router::new() + .route( + "/v1/teams/:team_id/enrollment-token", + post(routes::team_enrollment::create_token) + .get(routes::team_enrollment::get_token) + .delete(routes::team_enrollment::revoke_token), + ) + .route("/v1/enrollment/redeem", post(routes::team_enrollment::redeem)) + .route("/v1/teams/:team_id/join-requests", get(routes::team_enrollment::list_requests)) + .route("/v1/teams/:team_id/join-requests/:request_id/approve", post(routes::team_enrollment::approve)) + .route("/v1/teams/:team_id/join-requests/:request_id/deny", post(routes::team_enrollment::deny)) + .route_layer(middleware::from_fn(features::require_team_invites)) +} + // Merged routers keep the last one's fallback, layers included; without this an // unknown path would run whatever middleware the last merged router carries. async fn not_found() -> axum::http::StatusCode { @@ -234,6 +271,10 @@ async fn main() { register_per_day, Duration::from_secs(86400), )); + let enrollment_resolve_limiter = + EnrollmentResolveLimiter(RateLimiter::::new(60, Duration::from_secs(60))); + let enrollment_register_limiter = + EnrollmentRegisterLimiter(RateLimiter::::new(500, Duration::from_secs(86400))); let invite_limiter = InviteRateLimiter(RateLimiter::::new( invite_per_hour, Duration::from_secs(3600), @@ -260,6 +301,10 @@ async fn main() { GrantMintRateLimiter(RateLimiter::::new(30, Duration::from_secs(3600))); let grant_redeem_limiter = GrantRedeemRateLimiter(RateLimiter::::new(20, Duration::from_secs(3600))); + let enrollment_redeem_limiter = + EnrollmentRedeemRateLimiter(RateLimiter::::new(20, Duration::from_secs(3600))); + let resend_verification_limiter = + ResendVerificationRateLimiter(RateLimiter::::new(5, Duration::from_secs(3600))); // Lemon Squeezy live metrics cache (background refresh every 5 min). let ls_cache = lemonsqueezy::LsCache::default(); @@ -290,15 +335,13 @@ async fn main() { "Configured two-factor authentication" ); - // Register — stricter limit: 5/day per IP on top of the general auth 10/min - let register_route = Router::new() - .route("/v1/auth/register", post(routes::auth::register)) - .layer(Extension(notifier.clone())) - .layer(middleware::from_fn(rate_limit::register_rate_limit)) - .layer(Extension(register_limiter)) - .layer(middleware::from_fn(rate_limit::auth_rate_limit)) - .layer(Extension(auth_limiter.clone())) - .layer(middleware::from_fn(features::require_registration)); + let register_route = signup_routes( + notifier.clone(), + auth_limiter.clone(), + register_limiter, + enrollment_resolve_limiter, + enrollment_register_limiter, + ); // Public auth + public invitation lookup — rate limited at 10/min per IP let public = Router::new() @@ -312,7 +355,8 @@ async fn main() { get(routes::invitations::get_invitation), ) .layer(middleware::from_fn(rate_limit::auth_rate_limit)) - .layer(Extension(auth_limiter)); + .layer(Extension(auth_limiter)) + .layer(Extension(notifier.clone())); // A revoked session retries refresh every few seconds, and a passkey ceremony runs in the // browser beside the sign-in it completes; neither may spend the sign-in budget. @@ -453,6 +497,7 @@ async fn main() { ) .route("/v1/teams/:team_id", delete(routes::teams::delete_team).patch(routes::teams::rename_team)) .route("/v1/teams/:team_id/billing-owner", put(routes::team_ownership::transfer_billing_owner)) + .merge(enrollment_routes()) .route( "/v1/teams/:team_id/lock-policy", put(routes::team_lock_policy::set_lock_policy).delete(routes::team_lock_policy::clear_lock_policy), @@ -742,6 +787,8 @@ async fn main() { .layer(Extension(redeem_limiter)) .layer(Extension(grant_mint_limiter)) .layer(Extension(grant_redeem_limiter)) + .layer(Extension(enrollment_redeem_limiter)) + .layer(Extension(resend_verification_limiter)) .layer(middleware::from_fn(auth::auth_middleware)) .layer(Extension(notifier.clone())) .layer(Extension(terminal_manager.clone())) diff --git a/src/object_authz.rs b/src/object_authz.rs index 436cd11..8e2849f 100644 --- a/src/object_authz.rs +++ b/src/object_authz.rs @@ -64,6 +64,12 @@ pub struct MemberRow { pub role_ids: Vec, } +impl MemberContext { + pub fn effective(&self) -> i64 { + crate::permissions::with_dependencies(self.base) + } +} + impl MemberRow { pub fn context(&self, locked: bool) -> MemberContext { let base = if locked { diff --git a/src/permissions.rs b/src/permissions.rs index 424b1c2..585acd5 100644 --- a/src/permissions.rs +++ b/src/permissions.rs @@ -91,7 +91,7 @@ pub async fn effective_permissions( Ok(crate::object_authz::member_contexts(pool, team_id, Some(user_id)) .await? .pop() - .map_or(0, |m| with_dependencies(m.base))) + .map_or(0, |m| m.effective())) } /// Returns true if any of (team_id, user_id)'s roles grant `permission`. diff --git a/src/rate_limit.rs b/src/rate_limit.rs index d172bb7..59bd1c5 100644 --- a/src/rate_limit.rs +++ b/src/rate_limit.rs @@ -5,7 +5,7 @@ use std::net::IpAddr; use std::sync::{Arc, OnceLock}; use std::time::{Duration, Instant}; use tokio::sync::Mutex; -use tracing::warn; +use tracing::{warn, Instrument}; use uuid::Uuid; /// Per-key sliding window rate limiter. Key is typically IpAddr or Uuid. @@ -216,29 +216,39 @@ pub struct GrantMintRateLimiter(pub RateLimiter); #[derive(Clone)] pub struct GrantRedeemRateLimiter(pub RateLimiter); -/// Register endpoint: N registrations/day per IP. -pub async fn register_rate_limit( - axum::Extension(RegisterRateLimiter(limiter)): axum::Extension, - req: Request, - next: Next, -) -> Result { +/// Enrolment-token redemptions per user, so a leaked token cannot be probed through one account. +#[derive(Clone)] +pub struct EnrollmentRedeemRateLimiter(pub RateLimiter); + +/// Handlers that charge IP limiters themselves read the address from here. +#[derive(Clone, Copy)] +pub struct ClientIp(pub IpAddr); + +pub async fn attach_client_ip(mut req: Request, next: Next) -> Response { let ip = extract_ip(&req); - if !limiter.check(ip).await { - warn!(%ip, "Register rate limit exceeded"); - return Err(StatusCode::TOO_MANY_REQUESTS); - } - Ok(next.run(req).await) + req.extensions_mut().insert(ClientIp(ip)); + next.run(req).await } -/// Checks a per-user limiter, warning and returning 429 on exhaustion. The -/// one place this shape is expressed; middlewares and handlers alike call it. -pub async fn check_user_budget( - limiter: &RateLimiter, - user: Uuid, +/// Public token lookups (resolve, and the token branch of register), per IP. +#[derive(Clone)] +pub struct EnrollmentResolveLimiter(pub RateLimiter); + +/// Sign-ups through one enrolment token, so a fleet rollout behind one NAT is not capped per IP. +#[derive(Clone)] +pub struct EnrollmentRegisterLimiter(pub RateLimiter); + +#[derive(Clone)] +pub struct ResendVerificationRateLimiter(pub RateLimiter); + +pub async fn check_budget( + limiter: &RateLimiter, + key: K, label: &str, ) -> Result<(), StatusCode> { - if !limiter.check(user).await { - warn!(user_id = %user, limiter = label, "Rate limit exceeded"); + let shown = format!("{key}"); + if !limiter.check(key).await { + warn!(key = %shown, limiter = label, "Rate limit exceeded"); return Err(StatusCode::TOO_MANY_REQUESTS); } Ok(()) @@ -253,7 +263,7 @@ async fn user_keyed_limit( req: Request, next: Next, ) -> Result { - check_user_budget(limiter, user, label).await?; + check_budget(limiter, user, label).await?; Ok(next.run(req).await) } @@ -283,13 +293,8 @@ pub async fn auth_rate_limit( req: Request, next: Next, ) -> Result { - let ip = extract_ip(&req); - let method = req.method().clone(); - let path = req.uri().path().to_owned(); - if !limiter.check(ip).await { - warn!(%ip, method = %method, path = %path, "Auth rate limit exceeded"); - return Err(StatusCode::TOO_MANY_REQUESTS); - } + let span = tracing::warn_span!("auth_request", method = %req.method(), path = %req.uri().path()); + check_budget(&limiter, extract_ip(&req), "auth").instrument(span).await?; Ok(next.run(req).await) } @@ -299,11 +304,7 @@ pub async fn waitlist_rate_limit( req: Request, next: Next, ) -> Result { - let ip = extract_ip(&req); - if !limiter.check(ip).await { - warn!(%ip, "Waitlist rate limit exceeded"); - return Err(StatusCode::TOO_MANY_REQUESTS); - } + check_budget(&limiter, extract_ip(&req), "waitlist").await?; Ok(next.run(req).await) } diff --git a/src/routes/auth.rs b/src/routes/auth.rs index ba01dd4..750ae58 100644 --- a/src/routes/auth.rs +++ b/src/routes/auth.rs @@ -1,5 +1,5 @@ use axum::{ - extract::State, + extract::{rejection::JsonRejection, State}, http::StatusCode, response::{IntoResponse, Response}, Json, @@ -18,6 +18,13 @@ use crate::auth::{ password::{hash_auth_key, verify_auth_key}, AuthUser, }; +use crate::enrollment::{admit_verified_requests, domain_allowed, redeem_for, RedeemOutcome}; +use crate::rate_limit::{ + check_budget, ClientIp, EnrollmentRegisterLimiter, EnrollmentResolveLimiter, RateLimiter, RegisterRateLimiter, + ResendVerificationRateLimiter, +}; +use crate::routes::team_enrollment::load_live_token; +use crate::routes::ApiError; use crate::email::{send_password_changed_email, send_verification_email}; use crate::routes::email_undeliverable_response; use crate::routes::invitations::{accept_pending_invitation, PendingInvitation, PENDING_INVITATION_COLUMNS}; @@ -255,6 +262,22 @@ pub struct RegisterRequest { pub wrapped_user_secrets: Option, #[serde(default)] pub machine_fingerprint: Option, + #[serde(default)] + pub enrollment: Option, +} + +#[derive(Deserialize)] +pub struct EnrollmentCredential { + pub id: Uuid, + pub secret: String, +} + +#[derive(Serialize)] +pub struct RegisterResponse { + #[serde(flatten)] + session: AuthResponse, + #[serde(skip_serializing_if = "Option::is_none")] + enrollment: Option, } #[derive(Serialize)] @@ -321,13 +344,78 @@ async fn insert_user_with_handle_fallback( } } +async fn charge_ip_gates( + auth: &RateLimiter, + register: &RateLimiter, + ip: std::net::IpAddr, +) -> Result<(), StatusCode> { + check_budget(auth, ip, "auth").await?; + check_budget(register, ip, "register").await +} + +async fn charge_untokened_gates( + features: &crate::features::Features, + auth: &RateLimiter, + register: &RateLimiter, + ip: std::net::IpAddr, +) -> Result<(), ApiError> { + features.registration_open()?; + Ok(charge_ip_gates(auth, register, ip).await?) +} + +#[allow(clippy::too_many_arguments)] pub async fn register( State(pool): State, axum::Extension(features): axum::Extension, axum::Extension(notifier): axum::Extension, - Json(body): Json, -) -> Result<(StatusCode, Json), StatusCode> { - register_with(pool, ¬ifier, features, crate::email::delivery_configured(), body).await + axum::Extension(ClientIp(ip)): axum::Extension, + axum::Extension(auth_limiter): axum::Extension, + axum::Extension(RegisterRateLimiter(register_limiter)): axum::Extension, + axum::Extension(EnrollmentResolveLimiter(resolve_limiter)): axum::Extension, + axum::Extension(EnrollmentRegisterLimiter(token_limiter)): axum::Extension, + body: Result, JsonRejection>, +) -> Result<(StatusCode, Json), ApiError> { + let Json(body) = match body { + Ok(body) => body, + Err(rejection) => { + charge_untokened_gates(&features, &auth_limiter, ®ister_limiter, ip).await?; + return Err(rejection.status().into()); + } + }; + let credential = body.enrollment.as_ref().filter(|_| features.team_invites); + let token = match credential { + Some(credential) => match load_live_token(&pool, credential.id, &credential.secret).await { + Ok(token) => { + check_budget(&token_limiter, token.id, "enrollment_register").await?; + Some(token) + } + Err(err) => { + check_budget(&resolve_limiter, ip, "enrollment_resolve").await?; + charge_ip_gates(&auth_limiter, ®ister_limiter, ip).await?; + return Err(err.into()); + } + }, + None => { + charge_untokened_gates(&features, &auth_limiter, ®ister_limiter, ip).await?; + None + } + }; + if let Some(token) = &token { + if !domain_allowed(&token.guard(), &crate::email::normalize(&body.email)) { + return Err(ApiError::coded(StatusCode::FORBIDDEN, "ENROLLMENT_DOMAIN_MISMATCH")); + } + } + let public_key = body.public_key.clone(); + let (status, Json(session)) = + register_with(pool.clone(), ¬ifier, features, crate::email::delivery_configured(), body).await?; + let enrollment = match &token { + Some(token) => redeem_for(&pool, ¬ifier, token, session.user_id, public_key.as_deref()) + .await + .map_err(|err| warn!(?err, user_id = %session.user_id, "Enrolment redeem after registration failed")) + .ok(), + None => None, + }; + Ok((status, Json(RegisterResponse { session, enrollment }))) } async fn register_with( @@ -687,6 +775,7 @@ async fn derive_handle_on_verify(tx: &mut sqlx::PgConnection, user_id: Uuid, ema pub async fn verify_email( State(pool): State, axum::Extension(features): axum::Extension, + axum::Extension(notifier): axum::Extension, Json(body): Json, ) -> Result, StatusCode> { let db_err = |what: &'static str| { @@ -704,7 +793,8 @@ pub async fn verify_email( RETURNING user_id ) UPDATE users - SET email_verified = TRUE, email_verified_at = COALESCE(email_verified_at, now()) + SET email_verified = TRUE, email_verified_at = COALESCE(email_verified_at, now()), + email_proven_at = COALESCE(email_proven_at, now()) FROM consumed WHERE users.id = consumed.user_id RETURNING users.email, users.id, users.handle, users.handle_is_custom", @@ -720,6 +810,7 @@ pub async fn verify_email( } tx.commit().await.map_err(db_err("Failed to commit email verification"))?; info!(email = %email, "User email verified"); + admit_verified_requests(&pool, ¬ifier, user_id).await; return Ok(Json(VerifyEmailResponse { email, user_id })); } drop(tx); @@ -772,14 +863,15 @@ async fn issue_verification_token(conn: &mut sqlx::PgConnection, user_id: Uuid) pub async fn resend_verification_email( State(pool): State, axum::Extension(auth): axum::Extension, + axum::Extension(ResendVerificationRateLimiter(limiter)): axum::Extension, ) -> Result { let mut tx = pool.begin().await.map_err(|e| { error!(error = %e, "Failed to begin verification resend transaction"); StatusCode::INTERNAL_SERVER_ERROR })?; - let (email, verified, undeliverable) = sqlx::query_as::<_, (String, bool, bool)>( - "SELECT email, email_verified, email_undeliverable_at IS NOT NULL FROM users WHERE id = $1 FOR UPDATE", + let (email, proven, undeliverable) = sqlx::query_as::<_, (String, bool, bool)>( + "SELECT email, email_proven_at IS NOT NULL, email_undeliverable_at IS NOT NULL FROM users WHERE id = $1 FOR UPDATE", ) .bind(auth.0) .fetch_one(&mut *tx) @@ -789,12 +881,13 @@ pub async fn resend_verification_email( StatusCode::INTERNAL_SERVER_ERROR })?; - if verified { + if proven { return Ok(StatusCode::OK.into_response()); } if undeliverable { return Ok(email_undeliverable_response()); } + check_budget(&limiter, auth.0, "resend_verification").await?; let token = issue_verification_token(&mut tx, auth.0).await?; @@ -1052,7 +1145,8 @@ mod handler_tests { use crate::auth::jwt::validate_token; use crate::test_pool_or_skip; use crate::test_support::{ - add_member, seed_team, seed_user, seed_user_with_credentials, set_user_tier, set_user_trial, + add_member, seed_team, seed_user, seed_user_with_credentials, set_user_tier, set_user_trial, BillingMode, + DeliveryGuard, }; use axum::extract::State; use axum::Extension; @@ -1071,6 +1165,15 @@ mod handler_tests { } } + async fn register( + State(pool): State, + Extension(features): Extension, + Extension(notifier): Extension, + Json(body): Json, + ) -> Result<(StatusCode, Json), StatusCode> { + register_with(pool, ¬ifier, features, crate::email::delivery_configured(), body).await + } + fn register_req(email: &str, account_id: Uuid, fp: Option<&str>) -> RegisterRequest { RegisterRequest { email: email.to_string(), @@ -1079,6 +1182,7 @@ mod handler_tests { public_key: None, wrapped_user_secrets: None, machine_fingerprint: fp.map(str::to_string), + enrollment: None, } } @@ -1919,7 +2023,7 @@ mod handler_tests { } async fn verify(pool: &PgPool, switch: bool, token: String) { - let _ = verify_email(State(pool.clone()), features(switch), Json(VerifyEmailRequest { token })).await.unwrap(); + let _ = verify_email(State(pool.clone()), features(switch), Extension(SyncNotifier::new()), Json(VerifyEmailRequest { token })).await.unwrap(); } #[tokio::test] @@ -1958,6 +2062,150 @@ mod handler_tests { assert_eq!(handle_row(&pool, off_id).await, (generated, false)); } + async fn proven_at(pool: &PgPool, id: Uuid) -> Option> { + sqlx::query_scalar("SELECT email_proven_at FROM users WHERE id = $1") + .bind(id) + .fetch_one(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn verify_email_sets_email_proven_at() { + let _env = EnvGuard::new(true); + let pool = test_pool_or_skip!(); + let (id, token) = registered_unverified(&pool, false, &format!("{}@corp.test", unique_handle("pv"))).await; + assert!(proven_at(&pool, id).await.is_none()); + verify(&pool, false, token).await; + assert!(proven_at(&pool, id).await.is_some()); + } + + #[tokio::test] + async fn register_auto_verify_leaves_email_proven_at_null() { + let _env = EnvGuard::new(true); + let pool = test_pool_or_skip!(); + let email = format!("{}@corp.test", unique_handle("av")); + let (_, Json(resp)) = register_with(pool.clone(), &SyncNotifier::new(), features(false).0, false, register_req(&email, Uuid::new_v4(), None)) + .await + .unwrap(); + let verified: bool = sqlx::query_scalar("SELECT email_verified FROM users WHERE id = $1") + .bind(resp.user_id) + .fetch_one(&pool) + .await + .unwrap(); + assert!(verified); + assert!(proven_at(&pool, resp.user_id).await.is_none()); + } + + async fn verification_token_count(pool: &PgPool, id: Uuid) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM email_verification_tokens WHERE user_id = $1") + .bind(id) + .fetch_one(pool) + .await + .unwrap() + } + + async fn resend_with(pool: &PgPool, id: Uuid, budget: &ResendVerificationRateLimiter) -> StatusCode { + match resend_verification_email(State(pool.clone()), Extension(AuthUser(id)), Extension(budget.clone())).await { + Ok(response) => response.status(), + Err(status) => status, + } + } + + async fn resend(pool: &PgPool, id: Uuid) -> StatusCode { + resend_with(pool, id, &crate::test_support::default_resend_limiter()).await + } + + #[tokio::test] + async fn resend_is_capped_at_five_an_hour_per_user() { + let _env = DeliveryGuard::off(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let budget = crate::test_support::default_resend_limiter(); + let (id, other) = (seed_user(&pool).await, seed_user(&pool).await); + for _ in 0..5 { + assert_eq!(resend_with(&pool, id, &budget).await, StatusCode::OK); + } + assert_eq!(resend_with(&pool, id, &budget).await, StatusCode::TOO_MANY_REQUESTS); + assert_eq!(resend_with(&pool, other, &budget).await, StatusCode::OK); + } + + #[tokio::test] + async fn resend_sends_when_verified_but_not_proven() { + let _env = DeliveryGuard::off(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let id = crate::test_support::seed_user(&pool).await; + sqlx::query("UPDATE users SET email_verified = TRUE, email_verified_at = now() WHERE id = $1") + .bind(id) + .execute(&pool) + .await + .unwrap(); + assert_eq!(verification_token_count(&pool, id).await, 0); + assert_eq!(resend(&pool, id).await, StatusCode::OK); + assert_eq!(verification_token_count(&pool, id).await, 1); + } + + #[tokio::test] + async fn changing_email_clears_the_proven_address_and_resend_issues_a_link() { + let _env = DeliveryGuard::off(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let account_id = Uuid::new_v4(); + let id = crate::test_support::seed_user_with_credentials(&pool, account_id, "auth-key-secret").await; + sqlx::query("UPDATE users SET email_verified = TRUE, email_proven_at = now() WHERE id = $1") + .bind(id) + .execute(&pool) + .await + .unwrap(); + let status = update_email( + State(pool.clone()), + Extension(AuthUser(id)), + Json(UpdateEmailRequest { + new_email: format!("{}@moved.test", unique_handle("nm")), + auth_key: "auth-key-secret".to_string(), + }), + ) + .await + .unwrap(); + assert_eq!(status, StatusCode::NO_CONTENT); + assert!(proven_at(&pool, id).await.is_none()); + let after_change = verification_token_count(&pool, id).await; + assert_eq!(resend(&pool, id).await, StatusCode::OK); + assert_eq!(verification_token_count(&pool, id).await, after_change + 1); + } + + #[tokio::test] + async fn any_email_change_clears_the_proof() { + let pool = test_pool_or_skip!(); + let id = crate::test_support::seed_user(&pool).await; + sqlx::query("UPDATE users SET email_proven_at = now() WHERE id = $1").bind(id).execute(&pool).await.unwrap(); + sqlx::query("UPDATE users SET email = email, handle_is_custom = handle_is_custom WHERE id = $1") + .bind(id) + .execute(&pool) + .await + .unwrap(); + assert!(proven_at(&pool, id).await.is_some()); + + sqlx::query("UPDATE users SET email = $1 WHERE id = $2") + .bind(format!("{}@moved.test", unique_handle("tr"))) + .bind(id) + .execute(&pool) + .await + .unwrap(); + assert!(proven_at(&pool, id).await.is_none()); + } + + #[tokio::test] + async fn resend_is_a_no_op_once_proven() { + let pool = test_pool_or_skip!(); + let id = crate::test_support::seed_user(&pool).await; + sqlx::query("UPDATE users SET email_verified = TRUE, email_proven_at = now() WHERE id = $1") + .bind(id) + .execute(&pool) + .await + .unwrap(); + assert_eq!(resend(&pool, id).await, StatusCode::OK); + assert_eq!(verification_token_count(&pool, id).await, 0); + } + async fn stored_wrapped(pool: &PgPool, id: Uuid) -> Option { sqlx::query_scalar("SELECT wrapped_user_secrets FROM users WHERE id = $1") .bind(id) diff --git a/src/routes/invitations.rs b/src/routes/invitations.rs index 1df06f8..a55c5b9 100644 --- a/src/routes/invitations.rs +++ b/src/routes/invitations.rs @@ -196,6 +196,10 @@ pub(crate) async fn announce_member_joined( Some(metadata), ) .await; + notify_member_added(pool, notifier, team_id, user_id).await; +} + +pub(crate) async fn notify_member_added(pool: &PgPool, notifier: &SyncNotifier, team_id: Uuid, user_id: Uuid) { notifier.notify_membership_changed(user_id, team_id, true); notify_team_members_changed(pool, notifier, team_id).await; } diff --git a/src/routes/meta.rs b/src/routes/meta.rs index f1c033c..969375b 100644 --- a/src/routes/meta.rs +++ b/src/routes/meta.rs @@ -27,6 +27,8 @@ pub struct MetaResponse { pub wrapped_secrets_cas: bool, /// Only Owners confer the Owner role, and `PUT /v1/teams/:id/billing-owner` exists. pub multi_owner: bool, + /// The server can send email; without it registration auto-verifies and enrolment tokens must require approval. + pub email_delivery: bool, } pub async fn get_meta( @@ -34,8 +36,12 @@ pub async fn get_meta( Extension(cipher): Extension, Extension(passkeys): Extension, ) -> Json { + Json(meta_response(&features, &cipher, &passkeys, crate::email::delivery_configured())) +} + +fn meta_response(features: &Features, cipher: &SeedCipher, passkeys: &Passkeys, email_delivery: bool) -> MetaResponse { let self_hosted = self_host::is_self_hosted(); - Json(MetaResponse { + MetaResponse { self_hosted, billing_enabled: !self_hosted, registration_enabled: features.registration, @@ -47,7 +53,8 @@ pub async fn get_meta( passkeys: passkeys.available(), wrapped_secrets_cas: true, multi_owner: true, - }) + email_delivery, + } } #[cfg(test)] @@ -99,6 +106,14 @@ mod tests { assert_eq!(serde_json::to_value(&meta).unwrap()["wrapped_secrets_cas"], serde_json::json!(true)); } + #[test] + fn meta_reports_email_delivery() { + for configured in [true, false] { + let meta = meta_response(&Features::open(), &SeedCipher::for_tests(), &Passkeys::for_tests(), configured); + assert_eq!(serde_json::to_value(&meta).unwrap()["email_delivery"], serde_json::json!(configured)); + } + } + #[tokio::test] async fn meta_advertises_multi_owner() { let Json(meta) = get_meta(Extension(Features::open()), Extension(SeedCipher::for_tests()), Extension(Passkeys::for_tests())).await; diff --git a/src/routes/mod.rs b/src/routes/mod.rs index cb31d2b..b5ed7c8 100644 --- a/src/routes/mod.rs +++ b/src/routes/mod.rs @@ -14,6 +14,7 @@ pub mod presence; pub mod resend_webhook; pub mod session_codes; pub mod sync; +pub mod team_enrollment; pub mod team_grants; pub mod team_lock_policy; pub mod team_policy; @@ -51,3 +52,27 @@ pub(crate) fn email_undeliverable_response() -> Response { pub(crate) fn coded_error(status: StatusCode, code: &'static str) -> Response { (status, Json(serde_json::json!({ "error": code }))).into_response() } + +#[derive(Debug)] +pub(crate) struct ApiError(pub StatusCode, pub Option<&'static str>); + +impl ApiError { + pub(crate) fn coded(status: StatusCode, code: &'static str) -> Self { + ApiError(status, Some(code)) + } +} + +impl From for ApiError { + fn from(status: StatusCode) -> Self { + ApiError(status, None) + } +} + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + match self.1 { + Some(code) => coded_error(self.0, code), + None => self.0.into_response(), + } + } +} diff --git a/src/routes/resend_webhook.rs b/src/routes/resend_webhook.rs index c85b75b..82003af 100644 --- a/src/routes/resend_webhook.rs +++ b/src/routes/resend_webhook.rs @@ -253,6 +253,7 @@ mod tests { let resend = crate::routes::auth::resend_verification_email( State(pool.clone()), axum::Extension(crate::auth::AuthUser(unverified)), + axum::Extension(crate::test_support::default_resend_limiter()), ) .await .unwrap(); diff --git a/src/routes/session_codes.rs b/src/routes/session_codes.rs index 4179b93..3697fee 100644 --- a/src/routes/session_codes.rs +++ b/src/routes/session_codes.rs @@ -8,7 +8,7 @@ use tracing::warn; use uuid::Uuid; use crate::auth::AuthUser; -use crate::rate_limit::{check_user_budget, RedeemRateLimiter, SessionCodeRateLimiter}; +use crate::rate_limit::{check_budget, RedeemRateLimiter, SessionCodeRateLimiter}; use crate::session_grants; #[derive(Debug, Serialize)] @@ -34,7 +34,7 @@ pub async fn create_code( Extension(SessionCodeRateLimiter(limiter)): Extension, Path(session_id): Path, ) -> Result<(StatusCode, Json), StatusCode> { - check_user_budget(&limiter, auth.0, "session_code_mint").await?; + check_budget(&limiter, auth.0, "session_code_mint").await?; crate::routes::terminal::require_active_session_host(&pool, session_id, auth.0).await?; // Only invite_link sessions admit through the short-code/grant path @@ -68,7 +68,7 @@ pub async fn redeem_code( Extension(RedeemRateLimiter(limiter)): Extension, Json(body): Json, ) -> Result, StatusCode> { - check_user_budget(&limiter, auth.0, "session_code_redeem").await?; + check_budget(&limiter, auth.0, "session_code_redeem").await?; // Unknown, malformed, expired and revoked all answer 404: no response // distinguishes a real code from a wrong one. diff --git a/src/routes/team_enrollment.rs b/src/routes/team_enrollment.rs new file mode 100644 index 0000000..5044a80 --- /dev/null +++ b/src/routes/team_enrollment.rs @@ -0,0 +1,1794 @@ +use axum::extract::{Path, Query, State}; +use axum::http::StatusCode; +use axum::{Extension, Json}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use sqlx::{PgConnection, PgPool}; +use tracing::{error, info}; +use uuid::Uuid; + +use crate::auth::AuthUser; +use crate::enrollment::{ + audit_request, decide_pending, mint_secret, normalize_domains, notify_join_requests_changed, redeem_for, try_admit, + Guard, RedeemOutcome, role_or_default, +}; +use crate::rate_limit::{check_budget, ClientIp, EnrollmentRedeemRateLimiter, EnrollmentResolveLimiter, GrantMintRateLimiter}; +use crate::routes::audit::write_audit_event; +use crate::routes::ApiError; +use crate::routes::team_grants::{require_grant_admin, require_grantable_role, validate_public_key}; +use crate::session_grants::hash_secret; +use crate::sync_notifier::SyncNotifier; +use crate::team_authority::{lock_billing_owner, require_may_confer_builtin}; +use crate::team_plan::{require_granular, team_locked}; + +fn internal(what: &'static str) -> impl FnOnce(sqlx::Error) -> ApiError { + move |e| { + error!(error = %e, "{what}"); + StatusCode::INTERNAL_SERVER_ERROR.into() + } +} + +async fn revoke_live(conn: &mut PgConnection, team_id: Uuid) -> Result, ApiError> { + sqlx::query_scalar::<_, Uuid>( + "UPDATE team_enrollment_tokens SET revoked_at = now() WHERE team_id = $1 AND revoked_at IS NULL RETURNING id", + ) + .bind(team_id) + .fetch_optional(conn) + .await + .map_err(internal("Failed to revoke enrollment token")) +} + +async fn audit(pool: &PgPool, team_id: Uuid, actor: Uuid, action: &str, token_id: Uuid, metadata: Option) { + write_audit_event( + pool.clone(), + team_id, + actor, + action, + Some("enrollment_token"), + Some(token_id.to_string()), + None, + metadata, + ) + .await; +} + +#[derive(Deserialize)] +pub struct CreateTokenRequest { + pub role: Option, + pub allowed_domains: Vec, + pub require_approval: bool, + pub expires_at: Option>, +} + +#[derive(Serialize)] +pub struct CreateTokenResponse { + pub id: Uuid, + pub secret: String, + pub role: String, + pub allowed_domains: Vec, + pub require_approval: bool, + pub expires_at: Option>, + pub created_at: DateTime, +} + +impl std::fmt::Debug for CreateTokenResponse { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CreateTokenResponse") + .field("id", &self.id) + .field("secret", &"") + .field("role", &self.role) + .field("allowed_domains", &self.allowed_domains) + .field("require_approval", &self.require_approval) + .field("expires_at", &self.expires_at) + .field("created_at", &self.created_at) + .finish() + } +} + +const MAX_DOMAINS: usize = 50; + +fn validated_domains(body: &CreateTokenRequest) -> Result, ApiError> { + if body.allowed_domains.len() > MAX_DOMAINS { + return Err(StatusCode::BAD_REQUEST.into()); + } + let domains = normalize_domains(&body.allowed_domains).map_err(|()| ApiError::from(StatusCode::BAD_REQUEST))?; + if domains.is_empty() && !body.require_approval { + return Err(ApiError::coded(StatusCode::BAD_REQUEST, "ENROLLMENT_NEEDS_GUARD")); + } + if !body.require_approval && !crate::email::delivery_configured() { + return Err(ApiError::coded(StatusCode::BAD_REQUEST, "ENROLLMENT_NEEDS_APPROVAL")); + } + if body.expires_at.is_some_and(|at| at <= Utc::now()) { + return Err(StatusCode::BAD_REQUEST.into()); + } + Ok(domains) +} + +pub async fn create_token( + State(pool): State, + Extension(auth): Extension, + Extension(GrantMintRateLimiter(limiter)): Extension, + Path(team_id): Path, + Json(body): Json, +) -> Result<(StatusCode, Json), ApiError> { + check_budget(&limiter, auth.0, "enrollment_token_mint").await?; + require_grant_admin(&pool, team_id, auth.0).await?; + let role = role_or_default(body.role.as_deref()).to_string(); + require_grantable_role(&pool, team_id, auth.0, &role).await?; + let domains = validated_domains(&body)?; + require_granular(&pool, team_id, false).await?; + + let secret = mint_secret(); + let mut tx = pool.begin().await.map_err(internal("Failed to begin enrollment token transaction"))?; + lock_billing_owner(&mut tx, team_id).await?; + let replaced = revoke_live(&mut tx, team_id).await?; + let (id, created_at) = sqlx::query_as::<_, (Uuid, DateTime)>( + "INSERT INTO team_enrollment_tokens \ + (team_id, secret_hash, role, allowed_domains, require_approval, expires_at, created_by) \ + VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id, created_at", + ) + .bind(team_id) + .bind(hash_secret(&secret)) + .bind(&role) + .bind(&domains) + .bind(body.require_approval) + .bind(body.expires_at) + .bind(auth.0) + .fetch_one(&mut *tx) + .await + .map_err(internal("Failed to insert enrollment token"))?; + tx.commit().await.map_err(internal("Failed to commit enrollment token"))?; + + info!(team_id = %team_id, token_id = %id, rotated = replaced.is_some(), "Enrollment token created"); + let action = if replaced.is_some() { "enrollment_token.rotated" } else { "enrollment_token.created" }; + let metadata = json!({ + "role": role, + "allowed_domains": domains, + "require_approval": body.require_approval, + "expires_at": body.expires_at, + }); + audit(&pool, team_id, auth.0, action, id, Some(metadata)).await; + + Ok(( + StatusCode::CREATED, + Json(CreateTokenResponse { + id, + secret, + role, + allowed_domains: domains, + require_approval: body.require_approval, + expires_at: body.expires_at, + created_at, + }), + )) +} + +#[derive(Debug, Serialize, sqlx::FromRow)] +pub struct TokenView { + pub id: Uuid, + pub role: String, + pub allowed_domains: Vec, + pub require_approval: bool, + pub expires_at: Option>, + pub created_at: DateTime, + pub created_by_name: Option, + pub last_used_at: Option>, + pub admitted_count: i64, + pub pending_count: i64, +} + +#[derive(Debug, Serialize)] +pub struct GetTokenResponse { + pub token: Option, +} + +pub async fn get_token( + State(pool): State, + Extension(auth): Extension, + Path(team_id): Path, +) -> Result, ApiError> { + require_grant_admin(&pool, team_id, auth.0).await?; + let token = sqlx::query_as::<_, TokenView>( + "SELECT t.id, t.role, t.allowed_domains, t.require_approval, t.expires_at, t.created_at, \ + u.handle AS created_by_name, t.last_used_at, \ + (SELECT count(*) FROM team_join_requests r WHERE r.token_id = t.id AND r.status = 'approved') AS admitted_count, \ + (SELECT count(*) FROM team_join_requests r WHERE r.token_id = t.id AND r.status = 'pending') AS pending_count \ + FROM team_enrollment_tokens t LEFT JOIN users u ON u.id = t.created_by \ + WHERE t.team_id = $1 AND t.revoked_at IS NULL", + ) + .bind(team_id) + .fetch_optional(&pool) + .await + .map_err(internal("Failed to read enrollment token"))?; + Ok(Json(GetTokenResponse { token })) +} + +pub async fn revoke_token( + State(pool): State, + Extension(auth): Extension, + Path(team_id): Path, +) -> Result { + require_grant_admin(&pool, team_id, auth.0).await?; + let mut conn = pool.acquire().await.map_err(internal("Failed to acquire connection"))?; + let revoked = revoke_live(&mut conn, team_id).await?; + if let Some(id) = revoked { + info!(team_id = %team_id, token_id = %id, "Enrollment token revoked"); + audit(&pool, team_id, auth.0, "enrollment_token.revoked", id, None).await; + } + Ok(StatusCode::NO_CONTENT) +} + +#[derive(Debug, sqlx::FromRow)] +pub(crate) struct TokenRow { + pub id: Uuid, + pub team_id: Uuid, + pub team_name: String, + pub allowed_domains: Vec, + pub require_approval: bool, +} + +impl TokenRow { + pub(crate) fn guard(&self) -> Guard { + Guard { allowed_domains: self.allowed_domains.clone(), require_approval: self.require_approval } + } +} + +#[derive(Debug)] +pub(crate) enum EnrollError { + NotFound, + Gone, + Unavailable, + DomainMismatch, + NoSeat, + Status(StatusCode), +} + +impl From for EnrollError { + fn from(e: sqlx::Error) -> Self { + error!(error = %e, "Enrollment query failed"); + EnrollError::Status(StatusCode::INTERNAL_SERVER_ERROR) + } +} + +impl From for EnrollError { + fn from(status: StatusCode) -> Self { + EnrollError::Status(status) + } +} + +impl From for ApiError { + fn from(err: EnrollError) -> Self { + match err { + EnrollError::NotFound => StatusCode::NOT_FOUND.into(), + EnrollError::Gone => StatusCode::GONE.into(), + EnrollError::Unavailable => ApiError::coded(StatusCode::PAYMENT_REQUIRED, "ENROLLMENT_UNAVAILABLE"), + EnrollError::DomainMismatch => ApiError::coded(StatusCode::FORBIDDEN, "ENROLLMENT_DOMAIN_MISMATCH"), + EnrollError::NoSeat => StatusCode::PAYMENT_REQUIRED.into(), + EnrollError::Status(status) => status.into(), + } + } +} + +#[derive(sqlx::FromRow)] +struct LiveTokenLookup { + #[sqlx(flatten)] + row: TokenRow, + revoked: bool, + expired: bool, +} + +pub(crate) async fn load_live_token(pool: &PgPool, id: Uuid, secret: &str) -> Result { + let found = sqlx::query_as::<_, LiveTokenLookup>( + "SELECT t.id, t.team_id, tm.name AS team_name, t.allowed_domains, t.require_approval, \ + t.revoked_at IS NOT NULL AS revoked, COALESCE(t.expires_at <= now(), false) AS expired \ + FROM team_enrollment_tokens t JOIN teams tm ON tm.id = t.team_id \ + WHERE t.id = $1 AND t.secret_hash = $2", + ) + .bind(id) + .bind(hash_secret(secret)) + .fetch_optional(pool) + .await? + .ok_or(EnrollError::NotFound)?; + if found.revoked || found.expired { + return Err(EnrollError::Gone); + } + if team_locked(pool, found.row.team_id).await? { + return Err(EnrollError::Unavailable); + } + Ok(found.row) +} + +#[derive(Deserialize)] +pub struct ResolveRequest { + pub id: Uuid, + pub secret: String, +} + +pub async fn resolve( + State(pool): State, + Extension(ClientIp(ip)): Extension, + Extension(EnrollmentResolveLimiter(limiter)): Extension, + Json(body): Json, +) -> Result, ApiError> { + check_budget(&limiter, ip, "enrollment_resolve").await?; + let token = load_live_token(&pool, body.id, &body.secret).await?; + Ok(Json(json!({ + "team_name": token.team_name, + "allowed_domains": token.allowed_domains, + "require_approval": token.require_approval, + "registration_open": true, + }))) +} + +#[derive(Deserialize)] +pub struct RedeemRequest { + pub id: Uuid, + pub secret: String, + pub public_key: Option, +} + +pub async fn redeem( + State(pool): State, + Extension(auth): Extension, + Extension(notifier): Extension, + Extension(EnrollmentRedeemRateLimiter(limiter)): Extension, + Json(body): Json, +) -> Result, ApiError> { + check_budget(&limiter, auth.0, "enrollment_redeem").await?; + validate_public_key(body.public_key.as_deref())?; + let token = load_live_token(&pool, body.id, &body.secret).await?; + Ok(Json(redeem_for(&pool, ¬ifier, &token, auth.0, body.public_key.as_deref()).await?)) +} + +#[derive(Deserialize)] +pub struct ListRequestsQuery { + pub status: Option, +} + +#[derive(Debug, Serialize, sqlx::FromRow)] +pub struct JoinRequestView { + pub id: Uuid, + pub user_id: Uuid, + pub email: String, + pub display_name: String, + pub reason: Option, + pub email_proven: bool, + pub created_at: DateTime, +} + +pub async fn list_requests( + State(pool): State, + Extension(auth): Extension, + Path(team_id): Path, + Query(query): Query, +) -> Result>, ApiError> { + require_grant_admin(&pool, team_id, auth.0).await?; + let status = query.status.unwrap_or_else(|| "pending".to_string()); + if !["pending", "approved", "denied"].contains(&status.as_str()) { + return Err(StatusCode::BAD_REQUEST.into()); + } + let rows = sqlx::query_as::<_, JoinRequestView>( + "SELECT r.id, r.user_id, u.email, u.handle AS display_name, r.reason, u.email_proven_at IS NOT NULL AS email_proven, r.created_at \ + FROM team_join_requests r JOIN users u ON u.id = r.user_id \ + WHERE r.team_id = $1 AND r.status = $2 ORDER BY r.created_at, r.id", + ) + .bind(team_id) + .bind(status) + .fetch_all(&pool) + .await + .map_err(internal("Failed to list join requests"))?; + Ok(Json(rows)) +} + +pub async fn approve( + State(pool): State, + Extension(auth): Extension, + Extension(notifier): Extension, + Path((team_id, request_id)): Path<(Uuid, Uuid)>, +) -> Result, ApiError> { + require_grant_admin(&pool, team_id, auth.0).await?; + let role = sqlx::query_scalar::<_, Option>( + "SELECT t.role FROM team_join_requests r \ + LEFT JOIN team_enrollment_tokens t ON t.id = r.token_id \ + WHERE r.id = $1 AND r.team_id = $2 AND r.status = 'pending'", + ) + .bind(request_id) + .bind(team_id) + .fetch_optional(&pool) + .await + .map_err(internal("Failed to read join request"))? + .ok_or(StatusCode::NOT_FOUND)?; + require_may_confer_builtin(&pool, team_id, auth.0, role_or_default(role.as_deref())).await?; + Ok(Json(try_admit(&pool, ¬ifier, request_id, Some(auth.0)).await?)) +} + +pub async fn deny( + State(pool): State, + Extension(auth): Extension, + Extension(notifier): Extension, + Path((team_id, request_id)): Path<(Uuid, Uuid)>, +) -> Result { + require_grant_admin(&pool, team_id, auth.0).await?; + let mut conn = pool.acquire().await.map_err(internal("Failed to acquire connection"))?; + let user_id = decide_pending(&mut conn, request_id, team_id, "denied", Some(auth.0)) + .await + .map_err(internal("Failed to deny join request"))? + .ok_or(StatusCode::NOT_FOUND)?; + drop(conn); + audit_request(&pool, team_id, auth.0, "join_request.denied", request_id, user_id).await; + notify_join_requests_changed(&pool, ¬ifier, team_id).await; + Ok(StatusCode::NO_CONTENT) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::features::Features; + use crate::permissions::{PERM_CONNECT, PERM_INVITE_MEMBERS}; + use crate::enrollment::{try_admit, PendingReason}; + use crate::rate_limit::{EnrollmentRegisterLimiter, RateLimiter, RegisterRateLimiter}; + use crate::sync_notifier::SyncEvent; + use crate::test_pool_or_skip; + use crate::test_support::{ + add_member, grant_builtin_role, member_with_role, seed_team_with_roles, seed_user, set_user_tier, + team_audit_count, told, BillingMode, DeliveryGuard, + }; + use axum::body::Body; + use axum::http::Request; + use std::net::SocketAddr; + use std::time::Duration; + use tower::ServiceExt; + + fn budget() -> GrantMintRateLimiter { + GrantMintRateLimiter(RateLimiter::new(100, Duration::from_secs(3600))) + } + + async fn team_with_manager(pool: &PgPool) -> (Uuid, Uuid, Uuid) { + let owner = seed_user(pool).await; + let team = seed_team_with_roles(pool, owner).await; + let manager = seed_user(pool).await; + add_member(pool, team, manager).await; + grant_builtin_role(pool, team, manager, "manager").await; + (team, owner, manager) + } + + fn body(role: &str, domains: &[&str], approval: bool) -> CreateTokenRequest { + CreateTokenRequest { + role: Some(role.to_string()), + allowed_domains: domains.iter().map(|d| d.to_string()).collect(), + require_approval: approval, + expires_at: None, + } + } + + async fn create( + pool: &PgPool, + team: Uuid, + actor: Uuid, + req: CreateTokenRequest, + ) -> Result<(StatusCode, Json), ApiError> { + create_token( + State(pool.clone()), + Extension(AuthUser(actor)), + Extension(budget()), + Path(team), + Json(req), + ) + .await + } + + async fn created(pool: &PgPool, team: Uuid, actor: Uuid, req: CreateTokenRequest) -> CreateTokenResponse { + let (status, Json(token)) = create(pool, team, actor, req).await.expect("create token"); + assert_eq!(status, StatusCode::CREATED); + token + } + + async fn fetch(pool: &PgPool, team: Uuid, actor: Uuid) -> serde_json::Value { + let Json(value) = get_token(State(pool.clone()), Extension(AuthUser(actor)), Path(team)) + .await + .expect("get token"); + serde_json::to_value(value).unwrap() + } + + async fn revoke(pool: &PgPool, team: Uuid, actor: Uuid) -> Result { + revoke_token(State(pool.clone()), Extension(AuthUser(actor)), Path(team)).await + } + + fn error_code(err: ApiError) -> (StatusCode, Option<&'static str>) { + (err.0, err.1) + } + + async fn business_team(pool: &PgPool) -> (Uuid, Uuid, Uuid) { + let (team, owner, manager) = team_with_manager(pool).await; + set_user_tier(pool, owner, "business").await; + (team, owner, manager) + } + + #[tokio::test] + async fn create_returns_secret_once_and_get_never_does() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + + let token = created(&pool, team, manager, body("member", &["Example.com"], true)).await; + assert_eq!(token.secret.len(), 43); + assert_eq!(token.allowed_domains, vec!["example.com".to_string()]); + assert!(!format!("{token:?}").contains(&token.secret)); + + let stored: Vec = sqlx::query_scalar("SELECT secret_hash FROM team_enrollment_tokens WHERE id = $1") + .bind(token.id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(stored, crate::session_grants::hash_secret(&token.secret)); + + let view = fetch(&pool, team, manager).await; + let live = &view["token"]; + assert_eq!(live["id"], token.id.to_string()); + assert_eq!(live["role"], "member"); + assert_eq!(live["admitted_count"], 0); + assert_eq!(live["pending_count"], 0); + assert!(live["created_by_name"].is_string()); + assert!(!view.to_string().contains(&token.secret)); + assert!(live.get("secret").is_none()); + } + + #[tokio::test] + async fn get_counts_admitted_and_pending_for_the_live_token() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + assert!(fetch(&pool, team, manager).await["token"].is_null()); + + let token = created(&pool, team, manager, body("member", &[], true)).await; + for status in ["approved", "approved", "pending"] { + let user = seed_user(&pool).await; + sqlx::query("INSERT INTO team_join_requests (team_id, user_id, token_id, status) VALUES ($1, $2, $3, $4)") + .bind(team) + .bind(user) + .bind(token.id) + .bind(status) + .execute(&pool) + .await + .unwrap(); + } + let view = fetch(&pool, team, manager).await; + assert_eq!(view["token"]["admitted_count"], 2); + assert_eq!(view["token"]["pending_count"], 1); + } + + #[tokio::test] + async fn create_without_guard_is_400_needs_guard() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let err = create(&pool, team, manager, body("member", &[], false)).await.unwrap_err(); + assert_eq!(error_code(err), (StatusCode::BAD_REQUEST, Some("ENROLLMENT_NEEDS_GUARD"))); + } + + #[tokio::test] + async fn create_without_approval_on_server_without_email_is_400_needs_approval() { + let _env = DeliveryGuard::off(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + + let without = create(&pool, team, manager, body("member", &["example.com"], false)).await; + let with = create(&pool, team, manager, body("member", &["example.com"], true)).await; + + assert_eq!( + error_code(without.unwrap_err()), + (StatusCode::BAD_REQUEST, Some("ENROLLMENT_NEEDS_APPROVAL")) + ); + assert!(with.is_ok()); + } + + #[tokio::test] + async fn domains_without_approval_are_allowed_when_email_is_configured() { + let _env = DeliveryGuard::on(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + assert!(create(&pool, team, manager, body("member", &["example.com"], false)).await.is_ok()); + } + + #[tokio::test] + async fn create_without_role_defaults_to_member() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let token = created(&pool, team, manager, CreateTokenRequest { role: None, ..body("member", &[], true) }).await; + assert_eq!(token.role, "member"); + assert_eq!(fetch(&pool, team, manager).await["token"]["role"], "member"); + } + + #[tokio::test] + async fn more_than_fifty_domains_is_400() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let domains: Vec = (0..51).map(|i| format!("d{i}.example")).collect(); + let domains: Vec<&str> = domains.iter().map(String::as_str).collect(); + assert_eq!(create(&pool, team, manager, body("member", &domains, true)).await.unwrap_err().0, StatusCode::BAD_REQUEST); + assert!(create(&pool, team, manager, body("member", &domains[..50], true)).await.is_ok()); + } + + #[tokio::test] + async fn invalid_domains_role_and_expiry_are_400() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + + let bad_domain = create(&pool, team, manager, body("member", &["a@b.com"], true)).await.unwrap_err(); + assert_eq!(bad_domain.0, StatusCode::BAD_REQUEST); + + let mut past = body("member", &[], true); + past.expires_at = Some(Utc::now() - chrono::Duration::minutes(1)); + assert_eq!(create(&pool, team, manager, past).await.unwrap_err().0, StatusCode::BAD_REQUEST); + + let mut future = body("member", &[], true); + future.expires_at = Some(Utc::now() + chrono::Duration::days(1)); + assert!(create(&pool, team, manager, future).await.is_ok()); + } + + #[tokio::test] + async fn create_owner_role_is_400() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, owner, _manager) = team_with_manager(&pool).await; + for role in ["owner", "no-such-role"] { + let err = create(&pool, team, owner, body(role, &[], true)).await.unwrap_err(); + assert_eq!(err.0, StatusCode::BAD_REQUEST); + } + } + + #[tokio::test] + async fn create_role_above_actor_is_403() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, _manager) = team_with_manager(&pool).await; + let inviter = crate::test_support::inviter_with(&pool, team, PERM_INVITE_MEMBERS).await; + let err = create(&pool, team, inviter, body("manager", &[], true)).await.unwrap_err(); + assert_eq!(err.0, StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn member_without_invite_permission_is_403() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let plain = member_with_role(&pool, team, PERM_CONNECT).await; + created(&pool, team, manager, body("member", &[], true)).await; + + for actor in [plain, seed_user(&pool).await] { + assert_eq!(create(&pool, team, actor, body("member", &[], true)).await.unwrap_err().0, StatusCode::FORBIDDEN); + assert_eq!( + get_token(State(pool.clone()), Extension(AuthUser(actor)), Path(team)).await.unwrap_err().0, + StatusCode::FORBIDDEN + ); + assert_eq!(revoke(&pool, team, actor).await.unwrap_err().0, StatusCode::FORBIDDEN); + } + } + + #[tokio::test] + async fn rotate_revokes_previous_and_old_secret_hash_is_dead() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + + let first = created(&pool, team, manager, body("member", &[], true)).await; + let second = created(&pool, team, manager, body("editor", &["example.com"], true)).await; + assert_ne!(first.id, second.id); + assert_ne!(first.secret, second.secret); + + let live: Vec = sqlx::query_scalar( + "SELECT id FROM team_enrollment_tokens WHERE team_id = $1 AND revoked_at IS NULL", + ) + .bind(team) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(live, vec![second.id]); + + let old_live: Option = sqlx::query_scalar( + "SELECT id FROM team_enrollment_tokens WHERE secret_hash = $1 AND revoked_at IS NULL", + ) + .bind(crate::session_grants::hash_secret(&first.secret)) + .fetch_optional(&pool) + .await + .unwrap(); + assert_eq!(old_live, None); + assert_eq!(fetch(&pool, team, manager).await["token"]["role"], "editor"); + } + + #[tokio::test] + async fn below_business_create_402_revoke_204() { + let _billing = BillingMode::hosted(); + let pool = test_pool_or_skip!(); + let (team, owner, manager) = business_team(&pool).await; + let token = created(&pool, team, manager, body("member", &[], true)).await; + + set_user_tier(&pool, owner, "free").await; + let err = create(&pool, team, manager, body("member", &[], true)).await.unwrap_err(); + assert_eq!(err.0, StatusCode::PAYMENT_REQUIRED); + assert_eq!(fetch(&pool, team, manager).await["token"]["id"], token.id.to_string()); + + assert_eq!(revoke(&pool, team, manager).await.unwrap(), StatusCode::NO_CONTENT); + assert!(fetch(&pool, team, manager).await["token"].is_null()); + assert_eq!(revoke(&pool, team, manager).await.unwrap(), StatusCode::NO_CONTENT); + } + + #[tokio::test] + async fn team_invites_disabled_blocks_all_routes() { + let app = crate::enrollment_routes() + .layer(Extension(Features { team_invites: false, ..Features::open() })) + .with_state(crate::test_support::dead_pool().await); + let (team, request) = (Uuid::new_v4(), Uuid::new_v4()); + let token = format!("/v1/teams/{team}/enrollment-token"); + let requests = format!("/v1/teams/{team}/join-requests"); + + for (method, uri) in [ + ("POST", token.clone()), + ("GET", token.clone()), + ("DELETE", token), + ("POST", "/v1/enrollment/redeem".to_string()), + ("GET", requests.clone()), + ("POST", format!("{requests}/{request}/approve")), + ("POST", format!("{requests}/{request}/deny")), + ] { + let resp = app + .clone() + .oneshot(Request::builder().method(method).uri(&uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::FORBIDDEN, "{method} {uri}"); + } + } + + #[tokio::test] + async fn audit_created_rotated_revoked() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + + created(&pool, team, manager, body("member", &[], true)).await; + assert_eq!(team_audit_count(&pool, team, "enrollment_token.created").await, 1); + assert_eq!(team_audit_count(&pool, team, "enrollment_token.rotated").await, 0); + + created(&pool, team, manager, body("member", &["example.com"], true)).await; + assert_eq!(team_audit_count(&pool, team, "enrollment_token.created").await, 1); + assert_eq!(team_audit_count(&pool, team, "enrollment_token.rotated").await, 1); + + revoke(&pool, team, manager).await.unwrap(); + revoke(&pool, team, manager).await.unwrap(); + assert_eq!(team_audit_count(&pool, team, "enrollment_token.revoked").await, 1); + + let metadata: serde_json::Value = sqlx::query_scalar( + "SELECT metadata FROM audit_logs WHERE team_id = $1 AND action = 'enrollment_token.rotated'", + ) + .bind(team) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(metadata["role"], "member"); + assert_eq!(metadata["allowed_domains"], serde_json::json!(["example.com"])); + assert_eq!(metadata["require_approval"], true); + } + + #[tokio::test] + async fn two_simultaneous_creates_leave_one_live_token() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let (team, _owner, manager) = team_with_manager(&pool).await; + + let (a, b) = tokio::join!( + create(&pool, team, manager, body("member", &[], true)), + create(&pool, team, manager, body("editor", &[], true)), + ); + assert!(a.is_ok() && b.is_ok()); + + let live: i64 = sqlx::query_scalar( + "SELECT count(*) FROM team_enrollment_tokens WHERE team_id = $1 AND revoked_at IS NULL", + ) + .bind(team) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(live, 1); + } + + struct SignupApp { + app: axum::Router, + pool: PgPool, + } + + fn signup_app(pool: &PgPool, features: Features, register_per_day: usize) -> SignupApp { + signup_app_with_token_budget(pool, features, register_per_day, 500) + } + + fn signup_app_with_token_budget(pool: &PgPool, features: Features, register_per_day: usize, token_per_day: usize) -> SignupApp { + std::env::set_var("JWT_SECRET", "ci-test-secret"); + let auth_limiter = RateLimiter::new(10, Duration::from_secs(60)); + let app = crate::signup_routes( + crate::sync_notifier::SyncNotifier::new(), + auth_limiter.clone(), + RegisterRateLimiter(RateLimiter::new(register_per_day, Duration::from_secs(86400))), + EnrollmentResolveLimiter(RateLimiter::new(60, Duration::from_secs(60))), + EnrollmentRegisterLimiter(RateLimiter::new(token_per_day, Duration::from_secs(86400))), + ) + .merge( + axum::Router::new() + .route("/v1/auth/login", axum::routing::post(|| async { StatusCode::OK })) + .layer(axum::middleware::from_fn(crate::rate_limit::auth_rate_limit)) + .layer(Extension(auth_limiter)), + ) + .layer(Extension(features)) + .with_state(pool.clone()); + SignupApp { app, pool: pool.clone() } + } + + impl SignupApp { + async fn post(&self, uri: &str, ip: &str, body: serde_json::Value) -> (StatusCode, serde_json::Value) { + let mut req = Request::builder() + .method("POST") + .uri(uri) + .header("content-type", "application/json") + .body(Body::from(body.to_string())) + .unwrap(); + let addr = SocketAddr::new(ip.parse().unwrap(), 4000); + req.extensions_mut().insert(axum::extract::ConnectInfo(addr)); + let resp = self.app.clone().oneshot(req).await.unwrap(); + let status = resp.status(); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX).await.unwrap(); + (status, serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null)) + } + + async fn resolve(&self, ip: &str, id: Uuid, secret: &str) -> (StatusCode, serde_json::Value) { + self.post("/v1/enrollment/resolve", ip, json!({ "id": id, "secret": secret })).await + } + + async fn register(&self, ip: &str, email: &str, enrollment: Option<(Uuid, &str)>) -> (StatusCode, serde_json::Value) { + let mut body = json!({ "email": email, "account_id": Uuid::new_v4(), "auth_key": "auth-key-secret", "public_key": "pk" }); + if let Some((id, secret)) = enrollment { + body["enrollment"] = json!({ "id": id, "secret": secret }); + } + self.post("/v1/auth/register", ip, body).await + } + + async fn register_raw(&self, ip: &str, body: &str) -> StatusCode { + let mut req = Request::builder() + .method("POST") + .uri("/v1/auth/register") + .header("content-type", "application/json") + .body(Body::from(body.to_owned())) + .unwrap(); + req.extensions_mut().insert(axum::extract::ConnectInfo(SocketAddr::new(ip.parse().unwrap(), 4000))); + self.app.clone().oneshot(req).await.unwrap().status() + } + + async fn user_exists(&self, email: &str) -> bool { + sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM users WHERE email = $1)") + .bind(email) + .fetch_one(&self.pool) + .await + .unwrap() + } + } + + fn fresh_email(domain: &str) -> String { + format!("u{}@{domain}", Uuid::new_v4().simple()) + } + + async fn acme_token(pool: &PgPool) -> CreateTokenResponse { + let (team, _owner, manager) = team_with_manager(pool).await; + created(pool, team, manager, body("member", &["acme.com"], true)).await + } + + #[tokio::test] + async fn resolve_returns_team_name_and_guards() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + let (team, _owner, manager) = team_with_manager(&pool).await; + let token = created(&pool, team, manager, body("member", &["acme.com"], true)).await; + let name: String = sqlx::query_scalar("SELECT name FROM teams WHERE id = $1").bind(team).fetch_one(&pool).await.unwrap(); + + let (status, value) = app.resolve("10.0.0.1", token.id, &token.secret).await; + assert_eq!(status, StatusCode::OK); + assert_eq!( + value, + json!({ + "team_name": name, + "allowed_domains": ["acme.com"], + "require_approval": true, + "registration_open": true, + }) + ); + } + + #[tokio::test] + async fn resolve_unknown_404_revoked_410_expired_410_lapsed_402() { + let _billing = BillingMode::hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + let (team, owner, manager) = business_team(&pool).await; + let token = created(&pool, team, manager, body("member", &[], true)).await; + + assert_eq!(app.resolve("10.0.0.2", Uuid::new_v4(), &token.secret).await.0, StatusCode::NOT_FOUND); + assert_eq!(app.resolve("10.0.0.2", token.id, "wrong-secret").await.0, StatusCode::NOT_FOUND); + assert_eq!(app.resolve("10.0.0.2", token.id, &token.secret).await.0, StatusCode::OK); + + sqlx::query("UPDATE team_enrollment_tokens SET expires_at = now() - interval '1 minute' WHERE id = $1") + .bind(token.id) + .execute(&pool) + .await + .unwrap(); + assert_eq!(app.resolve("10.0.0.2", token.id, &token.secret).await.0, StatusCode::GONE); + + let fresh = created(&pool, team, manager, body("member", &[], true)).await; + set_user_tier(&pool, owner, "free").await; + let (status, value) = app.resolve("10.0.0.2", fresh.id, &fresh.secret).await; + assert_eq!((status, value), (StatusCode::PAYMENT_REQUIRED, json!({ "error": "ENROLLMENT_UNAVAILABLE" }))); + + set_user_tier(&pool, owner, "business").await; + revoke(&pool, team, manager).await.unwrap(); + assert_eq!(app.resolve("10.0.0.2", fresh.id, &fresh.secret).await.0, StatusCode::GONE); + } + + #[tokio::test] + async fn resolve_has_own_ip_bucket() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + for _ in 0..60 { + assert_eq!(app.resolve("10.0.0.3", Uuid::new_v4(), "x").await.0, StatusCode::NOT_FOUND); + } + assert_eq!(app.resolve("10.0.0.3", Uuid::new_v4(), "x").await.0, StatusCode::TOO_MANY_REQUESTS); + assert_eq!(app.post("/v1/auth/login", "10.0.0.3", json!({})).await.0, StatusCode::OK); + assert_eq!(app.resolve("10.0.0.4", Uuid::new_v4(), "x").await.0, StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn register_with_token_works_when_registration_disabled() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features { registration: false, ..Features::open() }, 5); + let token = acme_token(&pool).await; + let email = fresh_email("acme.com"); + let (status, _) = app.register("10.0.1.1", &email, Some((token.id, &token.secret))).await; + assert_eq!(status, StatusCode::CREATED); + assert!(app.user_exists(&email).await); + } + + #[tokio::test] + async fn register_without_token_still_blocked_when_disabled() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features { registration: false, ..Features::open() }, 5); + let email = fresh_email("acme.com"); + let (status, value) = app.register("10.0.1.2", &email, None).await; + assert_eq!((status, value), (StatusCode::FORBIDDEN, json!({ "error": "REGISTRATION_DISABLED" }))); + assert!(!app.user_exists(&email).await); + } + + #[tokio::test] + async fn register_with_token_skips_ip_daily_limit() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + let token = acme_token(&pool).await; + for _ in 0..6 { + let (status, _) = app.register("10.0.1.3", &fresh_email("acme.com"), Some((token.id, &token.secret))).await; + assert_eq!(status, StatusCode::CREATED); + } + assert_eq!(app.register("10.0.1.3", &fresh_email("acme.com"), None).await.0, StatusCode::CREATED); + } + + #[tokio::test] + async fn register_with_token_skips_shared_auth_bucket() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + let token = acme_token(&pool).await; + for _ in 0..11 { + let (status, _) = app.register("10.0.1.4", &fresh_email("acme.com"), Some((token.id, &token.secret))).await; + assert_eq!(status, StatusCode::CREATED); + } + assert_eq!(app.post("/v1/auth/login", "10.0.1.4", json!({})).await.0, StatusCode::OK); + } + + #[tokio::test] + async fn register_domain_mismatch_creates_no_user() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + let token = acme_token(&pool).await; + let email = fresh_email("gmail.com"); + let (status, value) = app.register("10.0.1.5", &email, Some((token.id, &token.secret))).await; + assert_eq!((status, value), (StatusCode::FORBIDDEN, json!({ "error": "ENROLLMENT_DOMAIN_MISMATCH" }))); + assert!(!app.user_exists(&email).await); + } + + #[tokio::test] + async fn register_with_bad_token_counts_against_ip_limit() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + for _ in 0..5 { + let (status, _) = app.register("10.0.1.6", &fresh_email("acme.com"), Some((Uuid::new_v4(), "nope"))).await; + assert_eq!(status, StatusCode::NOT_FOUND); + } + assert_eq!(app.register("10.0.1.6", &fresh_email("acme.com"), None).await.0, StatusCode::TOO_MANY_REQUESTS); + } + + #[tokio::test] + async fn register_ignores_the_token_when_team_invites_are_off() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features { team_invites: false, registration: false, ..Features::open() }, 5); + let token = acme_token(&pool).await; + let (status, value) = app.register("10.0.1.7", &fresh_email("acme.com"), Some((token.id, &token.secret))).await; + assert_eq!((status, value), (StatusCode::FORBIDDEN, json!({ "error": "REGISTRATION_DISABLED" }))); + } + + #[tokio::test] + async fn malformed_body_hits_the_registration_gate_first() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features { registration: false, ..Features::open() }, 5); + assert_eq!(app.register_raw("10.0.1.8", "{not json").await, StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn malformed_body_is_charged_to_the_ip_buckets() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 2); + for _ in 0..2 { + assert!(app.register_raw("10.0.1.9", "{not json").await.is_client_error()); + } + assert_eq!(app.register_raw("10.0.1.9", "{not json").await, StatusCode::TOO_MANY_REQUESTS); + } + + #[tokio::test] + async fn fifty_machines_resolve_and_register_from_one_ip() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + let token = acme_token(&pool).await; + for machine in 0..50 { + assert_eq!(app.resolve("10.0.2.3", token.id, &token.secret).await.0, StatusCode::OK, "resolve {machine}"); + let (status, _) = app.register("10.0.2.3", &fresh_email("acme.com"), Some((token.id, &token.secret))).await; + assert_eq!(status, StatusCode::CREATED, "register {machine}"); + } + } + + #[tokio::test] + async fn token_budget_is_keyed_on_the_token() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app_with_token_budget(&pool, Features::open(), 5, 2); + let a = acme_token(&pool).await; + let b = acme_token(&pool).await; + for _ in 0..2 { + assert_eq!(app.register("10.0.2.1", &fresh_email("acme.com"), Some((a.id, &a.secret))).await.0, StatusCode::CREATED); + } + let blocked = fresh_email("acme.com"); + assert_eq!(app.register("10.0.2.1", &blocked, Some((a.id, &a.secret))).await.0, StatusCode::TOO_MANY_REQUESTS); + assert!(!app.user_exists(&blocked).await); + assert_eq!(app.register("10.0.2.1", &fresh_email("acme.com"), Some((b.id, &b.secret))).await.0, StatusCode::CREATED); + } + + #[tokio::test] + async fn dead_token_on_a_closed_server_creates_no_user_and_is_charged_to_the_ip() { + let _billing = BillingMode::hosted(); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features { registration: false, ..Features::open() }, 2); + let (team, owner, manager) = business_team(&pool).await; + let revoked = created(&pool, team, manager, body("member", &[], true)).await; + revoke(&pool, team, manager).await.unwrap(); + let lapsed = created(&pool, team, manager, body("member", &[], true)).await; + set_user_tier(&pool, owner, "free").await; + + let gone = fresh_email("acme.com"); + let (status, _) = app.register("10.0.2.2", &gone, Some((revoked.id, &revoked.secret))).await; + assert_eq!(status, StatusCode::GONE); + let unavailable = fresh_email("acme.com"); + let (status, value) = app.register("10.0.2.2", &unavailable, Some((lapsed.id, &lapsed.secret))).await; + assert_eq!((status, value), (StatusCode::PAYMENT_REQUIRED, json!({ "error": "ENROLLMENT_UNAVAILABLE" }))); + assert!(!app.user_exists(&gone).await && !app.user_exists(&unavailable).await); + let (status, _) = app.register("10.0.2.2", &fresh_email("acme.com"), Some((revoked.id, &revoked.secret))).await; + assert_eq!(status, StatusCode::TOO_MANY_REQUESTS); + } + + fn redeem_budget() -> EnrollmentRedeemRateLimiter { + EnrollmentRedeemRateLimiter(RateLimiter::new(100, Duration::from_secs(3600))) + } + + async fn insert_token(pool: &PgPool, team: Uuid, creator: Uuid, role: &str, domains: &[&str], approval: bool) -> (Uuid, String) { + let secret = mint_secret(); + let id: Uuid = sqlx::query_scalar( + "INSERT INTO team_enrollment_tokens (team_id, secret_hash, role, allowed_domains, require_approval, created_by) \ + VALUES ($1, $2, $3, $4, $5, $6) RETURNING id", + ) + .bind(team) + .bind(hash_secret(&secret)) + .bind(role) + .bind(domains.iter().map(|d| d.to_string()).collect::>()) + .bind(approval) + .bind(creator) + .fetch_one(pool) + .await + .unwrap(); + (id, secret) + } + + async fn user_at(pool: &PgPool, email: &str, proven: bool) -> Uuid { + let user = seed_user(pool).await; + sqlx::query("UPDATE users SET email = $1 WHERE id = $2").bind(email).bind(user).execute(pool).await.unwrap(); + sqlx::query("UPDATE users SET email_proven_at = CASE WHEN $1 THEN now() END WHERE id = $2") + .bind(proven) + .bind(user) + .execute(pool) + .await + .unwrap(); + user + } + + async fn redeem_as(pool: &PgPool, notifier: &SyncNotifier, user: Uuid, id: Uuid, secret: &str) -> Result { + redeem_with_key(pool, notifier, user, id, secret, None).await + } + + async fn redeem_with_key( + pool: &PgPool, + notifier: &SyncNotifier, + user: Uuid, + id: Uuid, + secret: &str, + public_key: Option<&str>, + ) -> Result { + let Json(outcome) = redeem( + State(pool.clone()), + Extension(AuthUser(user)), + Extension(notifier.clone()), + Extension(redeem_budget()), + Json(RedeemRequest { id, secret: secret.to_string(), public_key: public_key.map(str::to_string) }), + ) + .await?; + Ok(serde_json::to_value(outcome).unwrap()) + } + + async fn approve_as(pool: &PgPool, notifier: &SyncNotifier, actor: Uuid, team: Uuid, request: Uuid) -> Result { + let Json(outcome) = + approve(State(pool.clone()), Extension(AuthUser(actor)), Extension(notifier.clone()), Path((team, request))).await?; + Ok(serde_json::to_value(outcome).unwrap()) + } + + async fn deny_as(pool: &PgPool, notifier: &SyncNotifier, actor: Uuid, team: Uuid, request: Uuid) -> Result { + deny(State(pool.clone()), Extension(AuthUser(actor)), Extension(notifier.clone()), Path((team, request))).await + } + + async fn pending_request(pool: &PgPool, team: Uuid, user: Uuid) -> (Uuid, Option) { + sqlx::query_as("SELECT id, reason FROM team_join_requests WHERE team_id = $1 AND user_id = $2 AND status = 'pending'") + .bind(team) + .bind(user) + .fetch_one(pool) + .await + .unwrap() + } + + async fn assert_pending_reasons_set(pool: &PgPool, team: Uuid) { + let unset: i64 = sqlx::query_scalar("SELECT count(*) FROM team_join_requests WHERE team_id = $1 AND status = 'pending' AND reason IS NULL") + .bind(team) + .fetch_one(pool) + .await + .unwrap(); + assert_eq!(unset, 0); + } + + async fn request_count(pool: &PgPool, team: Uuid, user: Uuid) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM team_join_requests WHERE team_id = $1 AND user_id = $2") + .bind(team) + .bind(user) + .fetch_one(pool) + .await + .unwrap() + } + + async fn membership_count(pool: &PgPool, team: Uuid, user: Uuid) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM team_members WHERE team_id = $1 AND user_id = $2") + .bind(team) + .bind(user) + .fetch_one(pool) + .await + .unwrap() + } + + async fn builtin_roles_of(pool: &PgPool, team: Uuid, user: Uuid) -> Vec { + sqlx::query_scalar( + "SELECT tr.name FROM team_member_roles mr JOIN team_roles tr ON tr.id = mr.role_id WHERE mr.team_id = $1 AND mr.user_id = $2", + ) + .bind(team) + .bind(user) + .fetch_all(pool) + .await + .unwrap() + } + + #[tokio::test] + async fn redeem_domain_only_proven_email_admits_and_uses_a_seat() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + let used_before = crate::routes::teams::owner_seats_used(&pool, owner).await.unwrap(); + + let outcome = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + + assert_eq!(outcome, json!({ "status": "admitted" })); + assert_eq!(membership_count(&pool, team, user).await, 1); + assert_eq!(builtin_roles_of(&pool, team, user).await, vec!["member".to_string()]); + assert_eq!(crate::routes::teams::owner_seats_used(&pool, owner).await.unwrap(), used_before + 1); + let invited_by: Option = sqlx::query_scalar("SELECT invited_by FROM team_members WHERE team_id = $1 AND user_id = $2") + .bind(team) + .bind(user) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(invited_by, Some(manager)); + let (status, reason): (String, Option) = + sqlx::query_as("SELECT status, reason FROM team_join_requests WHERE team_id = $1 AND user_id = $2") + .bind(team) + .bind(user) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!((status.as_str(), reason), ("approved", None)); + assert_eq!(team_audit_count(&pool, team, "join_request.created").await, 0); + assert_eq!(team_audit_count(&pool, team, "member.admitted").await, 1); + let via: serde_json::Value = + sqlx::query_scalar("SELECT metadata FROM audit_logs WHERE team_id = $1 AND action = 'member.admitted'") + .bind(team) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(via, json!({ "via": "enrollment" })); + let last_used: Option> = sqlx::query_scalar("SELECT last_used_at FROM team_enrollment_tokens WHERE id = $1") + .bind(id) + .fetch_one(&pool) + .await + .unwrap(); + assert!(last_used.is_some()); + } + + #[tokio::test] + async fn redeem_unproven_email_is_pending_then_verify_email_admits() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), false).await; + + let outcome = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + assert_eq!(outcome, json!({ "status": "pending", "reason": "email_unverified" })); + assert_eq!(membership_count(&pool, team, user).await, 0); + + let token: String = sqlx::query_scalar("INSERT INTO email_verification_tokens (user_id) VALUES ($1) RETURNING token") + .bind(user) + .fetch_one(&pool) + .await + .unwrap(); + let _ = crate::routes::auth::verify_email( + State(pool.clone()), + Extension(Features::open()), + Extension(notifier.clone()), + Json(crate::routes::auth::VerifyEmailRequest { token }), + ) + .await + .unwrap(); + + assert_eq!(membership_count(&pool, team, user).await, 1); + let status: String = sqlx::query_scalar("SELECT status FROM team_join_requests WHERE team_id = $1 AND user_id = $2") + .bind(team) + .bind(user) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(status, "approved"); + } + + #[tokio::test] + async fn redeem_with_approval_is_pending_until_approve() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("anywhere.org"), true).await; + + let outcome = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + assert_eq!(outcome, json!({ "status": "pending", "reason": "approval" })); + assert_eq!(membership_count(&pool, team, user).await, 0); + + let Json(listed) = list_requests( + State(pool.clone()), + Extension(AuthUser(manager)), + Path(team), + Query(ListRequestsQuery { status: Some("pending".to_string()) }), + ) + .await + .unwrap(); + let listed = serde_json::to_value(listed).unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + assert_eq!(listed.as_array().unwrap().len(), 1); + assert_eq!(listed[0]["id"], request.to_string()); + assert_eq!(listed[0]["user_id"], user.to_string()); + assert_eq!(listed[0]["reason"], "approval"); + assert_eq!(listed[0]["email_proven"], true); + assert!(listed[0]["email"].is_string() && listed[0]["display_name"].is_string() && listed[0]["created_at"].is_string()); + + let (other_team, _, other_manager) = team_with_manager(&pool).await; + assert_eq!(approve_as(&pool, ¬ifier, other_manager, other_team, request).await.unwrap_err().0, StatusCode::NOT_FOUND); + assert_eq!(deny_as(&pool, ¬ifier, other_manager, other_team, request).await.unwrap_err().0, StatusCode::NOT_FOUND); + + assert_eq!(approve_as(&pool, ¬ifier, manager, team, request).await.unwrap(), json!({ "status": "admitted" })); + assert_eq!(membership_count(&pool, team, user).await, 1); + assert_eq!(team_audit_count(&pool, team, "join_request.approved").await, 1); + assert_eq!(approve_as(&pool, ¬ifier, manager, team, request).await.unwrap_err().0, StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn approve_without_seat_is_402_and_stays_pending_seat_limit() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, owner, manager) = team_with_manager(&pool).await; + crate::test_support::set_user_seats(&pool, owner, 1).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + + let err = approve_as(&pool, ¬ifier, manager, team, request).await.unwrap_err(); + assert_eq!(error_code(err), (StatusCode::PAYMENT_REQUIRED, None)); + assert_eq!(pending_request(&pool, team, user).await, (request, Some("seat_limit".to_string()))); + assert_eq!(membership_count(&pool, team, user).await, 0); + + crate::test_support::set_user_seats(&pool, owner, 5).await; + assert_eq!(approve_as(&pool, ¬ifier, manager, team, request).await.unwrap(), json!({ "status": "admitted" })); + } + + #[tokio::test] + async fn approve_that_parks_on_the_seat_limit_notifies_holders() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, owner, manager) = team_with_manager(&pool).await; + crate::test_support::set_user_seats(&pool, owner, 1).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + let mut rx = notifier.subscribe(); + + assert_eq!(approve_as(&pool, ¬ifier, manager, team, request).await.unwrap_err().0, StatusCode::PAYMENT_REQUIRED); + + assert!(told(&mut rx, &format!("join_requests:{team}")).contains(&manager)); + } + + #[tokio::test] + async fn re_redeem_that_changes_the_reason_notifies_holders() { + let _env = DeliveryGuard::on(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), false).await; + let first = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + assert_eq!(first, json!({ "status": "pending", "reason": "email_unverified" })); + sqlx::query("UPDATE users SET email_proven_at = now() WHERE id = $1").bind(user).execute(&pool).await.unwrap(); + let mut rx = notifier.subscribe(); + + let second = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + + assert_eq!(second, json!({ "status": "pending", "reason": "approval" })); + assert!(told(&mut rx, &format!("join_requests:{team}")).contains(&manager)); + assert_eq!(request_count(&pool, team, user).await, 1); + assert_eq!(team_audit_count(&pool, team, "join_request.created").await, 1); + } + + #[tokio::test] + async fn one_user_redeeming_twice_at_once_never_sees_not_found() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + + let mut gate = pool.begin().await.unwrap(); + sqlx::query("SELECT 1 FROM teams WHERE id = $1 FOR NO KEY UPDATE").bind(team).fetch_one(&mut *gate).await.unwrap(); + let racers = [(); 2].map(|()| { + let (pool, notifier, secret) = (pool.clone(), notifier.clone(), secret.clone()); + tokio::spawn(async move { redeem_as(&pool, ¬ifier, user, id, &secret).await.map_err(error_code) }) + }); + tokio::time::sleep(Duration::from_millis(500)).await; + gate.commit().await.unwrap(); + let mut outcomes = Vec::new(); + for racer in racers { + outcomes.push(racer.await.unwrap().unwrap()["status"].clone()); + } + outcomes.sort_by_key(|status| status.to_string()); + + assert_eq!(outcomes, vec![json!("admitted"), json!("member")]); + assert_eq!(membership_count(&pool, team, user).await, 1); + } + + #[tokio::test] + async fn deleting_the_token_creator_keeps_the_token_admitting() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + + sqlx::query("DELETE FROM users WHERE id = $1").bind(manager).execute(&pool).await.unwrap(); + + let creator: Option = sqlx::query_scalar("SELECT created_by FROM team_enrollment_tokens WHERE id = $1") + .bind(id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(creator, None); + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + assert_eq!(redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(), json!({ "status": "admitted" })); + assert_eq!(membership_count(&pool, team, user).await, 1); + } + + #[tokio::test] + async fn redeem_without_seat_reports_pending_seat_limit() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, owner, manager) = team_with_manager(&pool).await; + crate::test_support::set_user_seats(&pool, owner, 1).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + + let outcome = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + assert_eq!(outcome, json!({ "status": "pending", "reason": "seat_limit" })); + assert_eq!(pending_request(&pool, team, user).await.1, Some("seat_limit".to_string())); + } + + #[tokio::test] + async fn redeem_twice_keeps_one_pending_row() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + + for _ in 0..2 { + let outcome = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + assert_eq!(outcome, json!({ "status": "pending", "reason": "approval" })); + } + assert_eq!(request_count(&pool, team, user).await, 1); + assert_eq!(team_audit_count(&pool, team, "join_request.created").await, 1); + assert_pending_reasons_set(&pool, team).await; + } + + #[tokio::test] + async fn existing_member_gets_member() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], true).await; + let user = user_at(&pool, &fresh_email("gmail.com"), true).await; + add_member(&pool, team, user).await; + + assert_eq!(redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(), json!({ "status": "member" })); + assert_eq!(request_count(&pool, team, user).await, 0); + } + + #[tokio::test] + async fn domain_mismatch_creates_no_request() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], true).await; + let user = user_at(&pool, &fresh_email("gmail.com"), true).await; + + let err = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap_err(); + assert_eq!(error_code(err), (StatusCode::FORBIDDEN, Some("ENROLLMENT_DOMAIN_MISMATCH"))); + assert_eq!(request_count(&pool, team, user).await, 0); + } + + #[tokio::test] + async fn deny_then_redeem_again_creates_a_new_pending_request() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (first, _) = pending_request(&pool, team, user).await; + + assert_eq!(deny_as(&pool, ¬ifier, manager, team, first).await.unwrap(), StatusCode::NO_CONTENT); + assert_eq!(team_audit_count(&pool, team, "join_request.denied").await, 1); + assert_eq!(deny_as(&pool, ¬ifier, manager, team, first).await.unwrap_err().0, StatusCode::NOT_FOUND); + + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (second, _) = pending_request(&pool, team, user).await; + assert_ne!(first, second); + assert_eq!(request_count(&pool, team, user).await, 2); + assert_eq!(membership_count(&pool, team, user).await, 0); + } + + #[tokio::test] + async fn lapsed_team_redeem_402_approve_402_deny_204() { + let _billing = BillingMode::hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, owner, manager) = business_team(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let waiting = user_at(&pool, &fresh_email("acme.com"), true).await; + redeem_as(&pool, ¬ifier, waiting, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, waiting).await; + + set_user_tier(&pool, owner, "free").await; + let newcomer = user_at(&pool, &fresh_email("acme.com"), true).await; + let unavailable = (StatusCode::PAYMENT_REQUIRED, Some("ENROLLMENT_UNAVAILABLE")); + assert_eq!(error_code(redeem_as(&pool, ¬ifier, newcomer, id, &secret).await.unwrap_err()), unavailable); + assert_eq!(error_code(approve_as(&pool, ¬ifier, manager, team, request).await.unwrap_err()), unavailable); + assert_eq!(request_count(&pool, team, newcomer).await, 0); + assert_pending_reasons_set(&pool, team).await; + assert_eq!(deny_as(&pool, ¬ifier, manager, team, request).await.unwrap(), StatusCode::NO_CONTENT); + } + + #[tokio::test] + async fn rotated_token_requests_still_decidable() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let old = created(&pool, team, manager, body("editor", &[], true)).await; + let a = user_at(&pool, &fresh_email("acme.com"), true).await; + let b = user_at(&pool, &fresh_email("acme.com"), true).await; + let c = user_at(&pool, &fresh_email("acme.com"), true).await; + for user in [a, b, c] { + redeem_as(&pool, ¬ifier, user, old.id, &old.secret).await.unwrap(); + } + + created(&pool, team, manager, body("member", &[], true)).await; + assert_eq!( + error_code(redeem_as(&pool, ¬ifier, seed_user(&pool).await, old.id, &old.secret).await.unwrap_err()), + (StatusCode::GONE, None) + ); + let (request_a, _) = pending_request(&pool, team, a).await; + assert_eq!(approve_as(&pool, ¬ifier, manager, team, request_a).await.unwrap(), json!({ "status": "admitted" })); + assert_eq!(builtin_roles_of(&pool, team, a).await, vec!["editor".to_string()]); + + let (request_b, _) = pending_request(&pool, team, b).await; + assert_eq!(deny_as(&pool, ¬ifier, manager, team, request_b).await.unwrap(), StatusCode::NO_CONTENT); + + sqlx::query("DELETE FROM team_enrollment_tokens WHERE id = $1").bind(old.id).execute(&pool).await.unwrap(); + let (request_c, _) = pending_request(&pool, team, c).await; + assert_eq!(approve_as(&pool, ¬ifier, manager, team, request_c).await.unwrap(), json!({ "status": "admitted" })); + assert_eq!(builtin_roles_of(&pool, team, c).await, vec!["member".to_string()]); + } + + #[tokio::test] + async fn join_requests_event_reaches_only_invite_permission_holders() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let plain = member_with_role(&pool, team, PERM_CONNECT).await; + let inviter = member_with_role(&pool, team, PERM_INVITE_MEMBERS).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + let mut rx = notifier.subscribe(); + + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + + let told = told(&mut rx, &format!("join_requests:{team}")); + assert!(told.contains(&manager) && told.contains(&inviter)); + assert!(!told.contains(&plain) && !told.contains(&user)); + } + + #[tokio::test] + async fn admission_sends_membership_changed() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + let mut rx = notifier.subscribe(); + + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + + let mut added = false; + let mut roster_told = std::collections::HashSet::new(); + while let Ok(event) = rx.try_recv() { + match event { + SyncEvent::MembershipChanged { user_id, team_id, added: true } if user_id == user && team_id == team => added = true, + SyncEvent::BlobPushed { user_id, device_id } if device_id == format!("team_members:{team}") => { + roster_told.insert(user_id); + } + _ => {} + } + } + assert!(added); + assert!(roster_told.contains(&manager) && roster_told.contains(&user)); + } + + #[tokio::test] + async fn approve_and_verify_racing_admit_once() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), false).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + sqlx::query("UPDATE users SET email_proven_at = now() WHERE id = $1").bind(user).execute(&pool).await.unwrap(); + + let mut gate = pool.begin().await.unwrap(); + sqlx::query("SELECT 1 FROM team_join_requests WHERE id = $1 FOR UPDATE").bind(request).fetch_one(&mut *gate).await.unwrap(); + let racers = [Some(manager), None].map(|by| { + let (pool, notifier) = (pool.clone(), notifier.clone()); + tokio::spawn(async move { try_admit(&pool, ¬ifier, request, by).await }) + }); + tokio::time::sleep(Duration::from_millis(300)).await; + gate.commit().await.unwrap(); + let [by_admin, by_verify] = racers; + let (by_admin, by_verify) = (by_admin.await.unwrap(), by_verify.await.unwrap()); + + assert_eq!([by_admin.is_ok(), by_verify.is_ok()].iter().filter(|ok| **ok).count(), 1); + assert_eq!(membership_count(&pool, team, user).await, 1); + let approved: i64 = sqlx::query_scalar("SELECT count(*) FROM team_join_requests WHERE id = $1 AND status = 'approved'") + .bind(request) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(approved, 1); + assert_eq!(team_audit_count(&pool, team, "member.admitted").await, 1); + } + + #[tokio::test] + async fn a_dead_token_never_auto_admits_a_verified_request() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), false).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + sqlx::query("UPDATE team_enrollment_tokens SET revoked_at = now() WHERE id = $1").bind(id).execute(&pool).await.unwrap(); + sqlx::query("UPDATE users SET email_proven_at = now() WHERE id = $1").bind(user).execute(&pool).await.unwrap(); + + let outcome = try_admit(&pool, ¬ifier, request, None).await.unwrap(); + assert_eq!(outcome, RedeemOutcome::Pending(PendingReason::Approval)); + assert_eq!(membership_count(&pool, team, user).await, 0); + } + + #[tokio::test] + async fn register_with_token_redeems_in_the_same_call() { + let _env = DeliveryGuard::off(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let app = signup_app(&pool, Features::open(), 5); + let (team, _owner, manager) = team_with_manager(&pool).await; + let token = created(&pool, team, manager, body("member", &["acme.com"], true)).await; + let email = fresh_email("acme.com"); + + let (status, value) = app.register("10.0.3.1", &email, Some((token.id, &token.secret))).await; + + assert_eq!(status, StatusCode::CREATED); + assert_eq!(value["enrollment"], json!({ "status": "pending", "reason": "approval" })); + assert!(value["jwt_token"].is_string()); + let user: Uuid = sqlx::query_scalar("SELECT id FROM users WHERE email = $1").bind(&email).fetch_one(&pool).await.unwrap(); + assert_eq!(pending_request(&pool, team, user).await.1, Some("approval".to_string())); + + let (_, plain) = app.register("10.0.3.1", &fresh_email("acme.com"), None).await; + assert!(plain.get("enrollment").is_none()); + } + + #[tokio::test] + async fn redeem_without_public_key_is_400_and_creates_no_request() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + sqlx::query("UPDATE users SET public_key = NULL WHERE id = $1").bind(user).execute(&pool).await.unwrap(); + + let err = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap_err(); + assert_eq!(error_code(err), (StatusCode::BAD_REQUEST, None)); + assert_eq!(request_count(&pool, team, user).await, 0); + + let outcome = redeem_with_key(&pool, ¬ifier, user, id, &secret, Some("fresh-key")).await.unwrap(); + assert_eq!(outcome, json!({ "status": "pending", "reason": "approval" })); + } + + #[tokio::test] + async fn both_guard_token_with_unproven_email_waits_for_verification_not_approval() { + let _env = DeliveryGuard::on(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), false).await; + + let redeemed = redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + let approved = approve_as(&pool, ¬ifier, manager, team, request).await.unwrap(); + + let pending = json!({ "status": "pending", "reason": "email_unverified" }); + assert_eq!((redeemed, approved), (pending.clone(), pending)); + assert_eq!(membership_count(&pool, team, user).await, 0); + assert_eq!(pending_request(&pool, team, user).await.1, Some("email_unverified".to_string())); + } + + #[tokio::test] + async fn domain_only_unproven_approve_stays_pending() { + let _env = DeliveryGuard::on(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &["acme.com"], false).await; + let user = user_at(&pool, &fresh_email("acme.com"), false).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + + let approved = approve_as(&pool, ¬ifier, manager, team, request).await.unwrap(); + + assert_eq!(approved, json!({ "status": "pending", "reason": "email_unverified" })); + assert_eq!(membership_count(&pool, team, user).await, 0); + } + + #[tokio::test] + async fn approval_only_token_with_unproven_email_is_admitted_on_approve() { + let _env = DeliveryGuard::on(BillingMode::self_hosted()); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), false).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, reason) = pending_request(&pool, team, user).await; + assert_eq!(reason, Some("approval".to_string())); + + let approved = approve_as(&pool, ¬ifier, manager, team, request).await.unwrap(); + + assert_eq!(approved, json!({ "status": "admitted" })); + assert_eq!(membership_count(&pool, team, user).await, 1); + } + + #[tokio::test] + async fn already_member_decision_notifies_join_request_holders() { + let _billing = BillingMode::self_hosted(); + let pool = test_pool_or_skip!(); + let notifier = SyncNotifier::new(); + let (team, _owner, manager) = team_with_manager(&pool).await; + let (id, secret) = insert_token(&pool, team, manager, "member", &[], true).await; + let user = user_at(&pool, &fresh_email("acme.com"), true).await; + redeem_as(&pool, ¬ifier, user, id, &secret).await.unwrap(); + let (request, _) = pending_request(&pool, team, user).await; + add_member(&pool, team, user).await; + let mut rx = notifier.subscribe(); + + assert_eq!(try_admit(&pool, ¬ifier, request, Some(manager)).await.unwrap(), RedeemOutcome::Member); + + assert!(told(&mut rx, &format!("join_requests:{team}")).contains(&manager)); + } +} diff --git a/src/routes/team_grants.rs b/src/routes/team_grants.rs index c21e0da..9d5f441 100644 --- a/src/routes/team_grants.rs +++ b/src/routes/team_grants.rs @@ -21,8 +21,9 @@ use tracing::{error, info, warn}; use uuid::Uuid; use crate::auth::AuthUser; +use crate::enrollment::role_or_default; use crate::permissions::{require_all_team_permissions, PERM_INVITE_MEMBERS}; -use crate::rate_limit::{check_user_budget, GrantMintRateLimiter, GrantRedeemRateLimiter}; +use crate::rate_limit::{check_budget, GrantMintRateLimiter, GrantRedeemRateLimiter}; use crate::routes::audit::write_audit_event; use crate::routes::invitations::{admit_member, announce_member_joined}; use crate::routes::teams::{ensure_seat_available, notify_team_members_changed, team_owner}; @@ -36,10 +37,75 @@ const MAX_PUBLIC_KEY_LEN: usize = 256; /// The create/list/revoke gate, in one place: the same permission bit that /// guards `POST /v1/teams/:team_id/invite`. Preview and redeem are deliberately /// not gated on it — they are authenticated as whoever holds the link. -async fn require_grant_admin(pool: &PgPool, team_id: Uuid, user: Uuid) -> Result<(), StatusCode> { +pub(crate) async fn require_grant_admin(pool: &PgPool, team_id: Uuid, user: Uuid) -> Result<(), StatusCode> { require_all_team_permissions(pool, team_id, user, &[PERM_INVITE_MEMBERS]).await } +/// Link-borne roles are the built-in ones below owner, and the actor may not confer more than they hold. +pub(crate) async fn require_grantable_role( + pool: &PgPool, + team_id: Uuid, + actor: Uuid, + role: &str, +) -> Result<(), StatusCode> { + if !grants::is_grantable_role(role) { + warn!(team_id = %team_id, user_id = %actor, role = %role, "Rejected non-grantable role"); + return Err(StatusCode::BAD_REQUEST); + } + crate::team_authority::require_may_confer_builtin(pool, team_id, actor, role).await +} + +pub(crate) fn validate_public_key(key: Option<&str>) -> Result<(), StatusCode> { + match key { + Some(key) if key.trim().is_empty() || key.len() > MAX_PUBLIC_KEY_LEN => Err(StatusCode::BAD_REQUEST), + _ => Ok(()), + } +} + +/// Overwriting a published key would orphan every vault key already wrapped to it; rotation is PUT /v1/auth/public-key. +async fn store_public_key_if_empty( + conn: &mut sqlx::PgConnection, + user_id: Uuid, + key: Option<&str>, +) -> Result<(), StatusCode> { + let Some(key) = key else { return Ok(()) }; + sqlx::query( + "UPDATE users SET public_key = $1, updated_at = now() \ + WHERE id = $2 AND (public_key IS NULL OR public_key = '')", + ) + .bind(key) + .bind(user_id) + .execute(conn) + .await + .map_err(|e| { + error!(error = %e, "Failed to record redeemer public key"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(()) +} + +/// Without a key on the roster row no key-holder can wrap for the member, so the join is refused. +pub(crate) async fn ensure_public_key( + conn: &mut sqlx::PgConnection, + user_id: Uuid, + key: Option<&str>, +) -> Result<(), StatusCode> { + store_public_key_if_empty(&mut *conn, user_id, key).await?; + let has_key = sqlx::query_scalar::<_, bool>("SELECT public_key IS NOT NULL AND public_key <> '' FROM users WHERE id = $1") + .bind(user_id) + .fetch_one(conn) + .await + .map_err(|e| { + error!(error = %e, "Failed to verify redeemer public key"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + if !has_key { + warn!(user_id = %user_id, "Join without a public key"); + return Err(StatusCode::BAD_REQUEST); + } + Ok(()) +} + fn reject(rejection: GrantRejection) -> StatusCode { match rejection { // Wrong id and wrong secret answer alike: nothing distinguishes a real @@ -98,15 +164,11 @@ pub async fn create_grant( Path(team_id): Path, Json(body): Json, ) -> Result<(StatusCode, Json), StatusCode> { - check_user_budget(&limiter, auth.0, "team_grant_mint").await?; + check_budget(&limiter, auth.0, "team_grant_mint").await?; require_grant_admin(&pool, team_id, auth.0).await?; - let role = body.role.as_deref().unwrap_or("member").to_string(); - if !grants::is_grantable_role(&role) { - warn!(team_id = %team_id, user_id = %auth.0, role = %role, "Rejected non-grantable role"); - return Err(StatusCode::BAD_REQUEST); - } - crate::team_authority::require_may_confer_builtin(&pool, team_id, auth.0, &role).await?; + let role = role_or_default(body.role.as_deref()).to_string(); + require_grantable_role(&pool, team_id, auth.0, &role).await?; let max_uses = grants::clamp_max_uses(body.max_uses); let ttl = grants::clamp_ttl(body.expires_in_secs); @@ -219,7 +281,7 @@ pub async fn preview_grant( Path(grant_id): Path, Json(body): Json, ) -> Result, StatusCode> { - check_user_budget(&limiter, auth.0, "team_grant_preview").await?; + check_budget(&limiter, auth.0, "team_grant_preview").await?; let preview = grants::preview(&pool, grant_id, &body.secret) .await @@ -258,13 +320,9 @@ pub async fn redeem_grant( Path(grant_id): Path, Json(body): Json, ) -> Result, StatusCode> { - check_user_budget(&limiter, auth.0, "team_grant_redeem").await?; + check_budget(&limiter, auth.0, "team_grant_redeem").await?; - if let Some(key) = body.public_key.as_deref() { - if key.trim().is_empty() || key.len() > MAX_PUBLIC_KEY_LEN { - return Err(StatusCode::BAD_REQUEST); - } - } + validate_public_key(body.public_key.as_deref())?; let mut tx = pool.begin().await.map_err(|e| { error!(error = %e, "Failed to begin transaction for grant redemption"); @@ -311,39 +369,7 @@ pub async fn redeem_grant( // to whoever minted the link. admit_member(&mut tx, locked.team_id, auth.0, Some(locked.created_by), &locked.role, &[], None).await?; - // Fill a missing key only. Overwriting a key a user already published - // would orphan every vault key already wrapped to it, in this team and - // every other; rotation belongs to PUT /v1/auth/public-key. - if let Some(key) = body.public_key.as_deref() { - sqlx::query( - "UPDATE users SET public_key = $1, updated_at = now() \ - WHERE id = $2 AND (public_key IS NULL OR public_key = '')", - ) - .bind(key) - .bind(auth.0) - .execute(&mut *tx) - .await - .map_err(|e| { - error!(error = %e, "Failed to record redeemer public key"); - StatusCode::INTERNAL_SERVER_ERROR - })?; - } - - // Without a key on the roster row, no key-holder can wrap for this member - // and the joiner lands in a vault that can never fill. Refuse the join - // rather than create that state. - let has_key = sqlx::query_scalar::<_, bool>( - "SELECT public_key IS NOT NULL AND public_key <> '' FROM users WHERE id = $1", - ) - .bind(auth.0) - .fetch_one(&mut *tx) - .await - .map_err(|e| { error!(error = %e, "Failed to verify redeemer public key"); StatusCode::INTERNAL_SERVER_ERROR })?; - - if !has_key { - warn!(user_id = %auth.0, "Join grant redemption without a public key"); - return Err(StatusCode::BAD_REQUEST); - } + ensure_public_key(&mut tx, auth.0, body.public_key.as_deref()).await?; tx.commit().await.map_err(|e| { error!(error = %e, "Failed to commit grant redemption"); @@ -1110,9 +1136,9 @@ mod tests { #[test] fn generated_secrets_are_url_safe_and_high_entropy() { - let secret = grants::generate_secret(); + let secret = crate::enrollment::mint_secret(); assert_eq!(secret.len(), 43, "43 base64url characters is 256 bits"); assert!(secret.chars().all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')); - assert_ne!(secret, grants::generate_secret()); + assert_ne!(secret, crate::enrollment::mint_secret()); } } diff --git a/src/routes/teams.rs b/src/routes/teams.rs index 5f85535..9914e7e 100644 --- a/src/routes/teams.rs +++ b/src/routes/teams.rs @@ -3216,10 +3216,8 @@ mod authz_tests { #[tokio::test] async fn email_invite_reports_when_the_email_was_not_sent() { + let _env = crate::test_support::DeliveryGuard::off(BillingMode::self_hosted()); let pool = test_pool_or_skip!(); - if crate::email::delivery_configured() { - return; - } let owner = seed_user(&pool).await; let team = seed_team(&pool, owner).await; let inviter = inviter_with(&pool, team, PERM_INVITE_MEMBERS).await; diff --git a/src/sync_notifier.rs b/src/sync_notifier.rs index 754d8d6..b961224 100644 --- a/src/sync_notifier.rs +++ b/src/sync_notifier.rs @@ -161,6 +161,19 @@ pub async fn notify_team(pool: &PgPool, notifier: &SyncNotifier, team_id: Uuid, .await; } +pub async fn notify_team_permission_holders( + pool: &PgPool, + notifier: &SyncNotifier, + team_id: Uuid, + perm: i64, + payload: String, +) { + let members = crate::object_authz::member_contexts(pool, team_id, None).await.unwrap_or_default(); + for member in members.iter().filter(|m| m.effective() & perm != 0) { + notifier.notify(member.user_id, payload.clone()); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/team_join_grants.rs b/src/team_join_grants.rs index 345ca48..62667a4 100644 --- a/src/team_join_grants.rs +++ b/src/team_join_grants.rs @@ -14,9 +14,7 @@ //! [`crate::session_grants`] — a code minted for one purpose being redeemable //! on another path is exactly the bug that produced that rule. -use base64::Engine; use chrono::{DateTime, Duration, Utc}; -use rand::RngCore; use sqlx::PgPool; use uuid::Uuid; @@ -40,14 +38,6 @@ pub fn is_grantable_role(role: &str) -> bool { GRANTABLE_ROLES.contains(&role) } -/// 256 bits of URL-safe randomness. Longer than the session-grant token -/// because this one travels in a shareable link with a multi-day life. -pub fn generate_secret() -> String { - let mut bytes = [0u8; 32]; - rand::thread_rng().fill_bytes(&mut bytes); - base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes) -} - /// Clamp a requested TTL into `[60, MAX_TTL_SECS]`, defaulting when absent. pub fn clamp_ttl(requested: Option) -> Duration { let secs = requested @@ -79,7 +69,7 @@ pub async fn create( ttl: Duration, created_by: Uuid, ) -> Result<(GrantRow, String), sqlx::Error> { - let secret = generate_secret(); + let secret = crate::enrollment::mint_secret(); let expires_at = Utc::now() + ttl; let (id, created_at_role): (Uuid, String) = sqlx::query_as( diff --git a/src/test_support.rs b/src/test_support.rs index 038b91d..fe15f78 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -41,11 +41,7 @@ impl BillingMode { fn set(key: Option<&str>) -> Self { let lock = env_lock(); - let prev = std::env::var("LEMONSQUEEZY_API_KEY").ok(); - match key { - Some(k) => std::env::set_var("LEMONSQUEEZY_API_KEY", k), - None => std::env::remove_var("LEMONSQUEEZY_API_KEY"), - } + let prev = swap_env("LEMONSQUEEZY_API_KEY", key); let prev_mode = crate::self_host::set_test_billing_mode(Some(key.is_none())); BillingMode(lock, prev, prev_mode) } @@ -53,14 +49,45 @@ impl BillingMode { impl Drop for BillingMode { fn drop(&mut self) { - match &self.1 { - Some(v) => std::env::set_var("LEMONSQUEEZY_API_KEY", v), - None => std::env::remove_var("LEMONSQUEEZY_API_KEY"), - } + swap_env("LEMONSQUEEZY_API_KEY", self.1.as_deref()); crate::self_host::set_test_billing_mode(self.2); } } +fn swap_env(name: &str, value: Option<&str>) -> Option { + let prev = std::env::var(name).ok(); + match value { + Some(v) => std::env::set_var(name, v), + None => std::env::remove_var(name), + } + prev +} + +/// Email delivery on or off for the test, under the env lock the billing mode holds. +#[allow(dead_code)] +pub struct DeliveryGuard(BillingMode, Option); + +impl DeliveryGuard { + pub fn on(billing: BillingMode) -> Self { + Self::set(billing, Some("test-key")) + } + + pub fn off(billing: BillingMode) -> Self { + Self::set(billing, None) + } + + fn set(billing: BillingMode, key: Option<&str>) -> Self { + let prev = swap_env("RESEND_API_KEY", key); + DeliveryGuard(billing, prev) + } +} + +impl Drop for DeliveryGuard { + fn drop(&mut self) { + swap_env("RESEND_API_KEY", self.1.as_deref()); + } +} + /// Serializes tests that read or write `users.last_seen_on`. Activity counts are /// whole-table aggregates, so a concurrent test stamping a user would shift the /// totals mid-assertion. Any test touching that column must hold this lock. @@ -124,6 +151,13 @@ pub fn default_knock_limiter() -> crate::rate_limit::KnockRateLimiter { )) } +pub fn default_resend_limiter() -> crate::rate_limit::ResendVerificationRateLimiter { + crate::rate_limit::ResendVerificationRateLimiter(crate::rate_limit::RateLimiter::new( + 5, + std::time::Duration::from_secs(3600), + )) +} + /// Insert a minimal valid user and return its id. Each call uses fresh UUIDs so /// tests never collide on the unique `email`/`account_id` columns. pub async fn seed_user(pool: &PgPool) -> Uuid { @@ -318,11 +352,17 @@ pub fn team_members_told( rx: &mut tokio::sync::broadcast::Receiver, team: Uuid, ) -> std::collections::HashSet { - let want = format!("team_members:{team}"); + told(rx, &format!("team_members:{team}")) +} + +pub fn told( + rx: &mut tokio::sync::broadcast::Receiver, + payload: &str, +) -> std::collections::HashSet { let mut told = std::collections::HashSet::new(); while let Ok(ev) = rx.try_recv() { if let crate::sync_notifier::SyncEvent::BlobPushed { user_id, device_id } = ev { - if device_id == want { + if device_id == payload { told.insert(user_id); } }