fix(manager-core): F-S-002 rate-limit users::request_password_reset + send_verification_email

Both methods short-circuit the authz gate when cx.principal == None.
An attacker hitting any public route could trigger unbounded outbound
email per call to arbitrary or attacker-supplied addresses — exhausting
SMTP-relay quota, getting the relay blacklisted, or burning provider
credits.

Add an in-memory EmailRateLimiter on UsersServiceImpl with two scopes:
- Per-(app, recipient): 5 calls per rolling 1h window
- Per-app daily cap: 200 calls per rolling 24h window

Both windows reset lazily. Token-bucket counts increment only after
both checks pass — a denied recipient doesn't consume the app counter.

Wired:
- send_verification_email: returns UsersError::EmailRateLimited (→ 429
  on the admin HTTP surface).
- request_password_reset: returns Ok(()) silently on rate-limit (would
  otherwise enable an existence-leak side channel).

UsersError::EmailRateLimited variant added; AppUsersApiError gains the
same variant + 429 mapping. NoopUsersError trait stub unaffected.

AUDIT.md anchor: F-S-002.

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

crates/manager-core/src/users_service.rs

@@ -123,6 +123,98 @@ fn env_duration_days(var: &str, default: Duration) -> Duration {
 }
 
 /// Postgres-backed `UsersService`.
+/// In-memory rate limiter for outbound-email surfaces (verification +
+/// password reset). Two scopes:
+///
+/// 1. **Per-(app, recipient)**: max `recipient_burst` calls per
+///    `recipient_window` (default 5 per hour). Defends against an
+///    attacker hammering one address.
+/// 2. **Per-app daily cap**: max `app_daily_cap` calls per app per
+///    rolling 24h window (default 200). Defends the SMTP-relay quota.
+///
+/// Both windows reset lazily — buckets that have aged out get rebuilt
+/// on the next call. Stored in a single Mutex; reads + writes are
+/// short, contention is low for the call volume this limits.
+#[derive(Debug)]
+struct EmailRateLimiter {
+    recipient_burst: u32,
+    recipient_window: Duration,
+    app_daily_cap: u32,
+    state: std::sync::Mutex<EmailRateState>,
+}
+
+#[derive(Debug, Default)]
+struct EmailRateState {
+    per_recipient: std::collections::HashMap<(AppId, String), Bucket>,
+    per_app: std::collections::HashMap<AppId, Bucket>,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Bucket {
+    window_started_at: std::time::Instant,
+    count: u32,
+}
+
+impl EmailRateLimiter {
+    fn new() -> Self {
+        Self {
+            recipient_burst: 5,
+            recipient_window: Duration::from_secs(3600),
+            app_daily_cap: 200,
+            state: std::sync::Mutex::new(EmailRateState::default()),
+        }
+    }
+
+    /// Attempt to consume one slot for `(app, recipient)`. Returns
+    /// `Err(())` if either cap is exceeded.
+    fn try_consume(&self, app_id: AppId, recipient: &str) -> Result<(), ()> {
+        let now = std::time::Instant::now();
+        let app_cap_window = Duration::from_secs(24 * 3600);
+        let mut state = self.state.lock().expect("email rate state poisoned");
+
+        // Check + roll the per-app bucket.
+        {
+            let app_bucket = state.per_app.entry(app_id).or_insert(Bucket {
+                window_started_at: now,
+                count: 0,
+            });
+            if now.duration_since(app_bucket.window_started_at) >= app_cap_window {
+                *app_bucket = Bucket {
+                    window_started_at: now,
+                    count: 0,
+                };
+            }
+            if app_bucket.count >= self.app_daily_cap {
+                return Err(());
+            }
+        }
+
+        // Check + roll the per-recipient bucket.
+        let key = (app_id, recipient.to_string());
+        {
+            let bucket = state.per_recipient.entry(key).or_insert(Bucket {
+                window_started_at: now,
+                count: 0,
+            });
+            if now.duration_since(bucket.window_started_at) >= self.recipient_window {
+                *bucket = Bucket {
+                    window_started_at: now,
+                    count: 0,
+                };
+            }
+            if bucket.count >= self.recipient_burst {
+                return Err(());
+            }
+            bucket.count += 1;
+        }
+        // Both checks passed; bump the app counter.
+        if let Some(app_bucket) = state.per_app.get_mut(&app_id) {
+            app_bucket.count += 1;
+        }
+        Ok(())
+    }
+}
+
 pub struct UsersServiceImpl {
     users: Arc<dyn AppUserRepository>,
     sessions: Arc<dyn AppUserSessionRepository>,
@@ -134,6 +226,7 @@ pub struct UsersServiceImpl {
     authz: Arc<dyn AuthzRepo>,
     events: Arc<dyn ServiceEventEmitter>,
     config: UsersServiceConfig,
+    email_rate_limiter: EmailRateLimiter,
 }
 
 impl UsersServiceImpl {
@@ -162,6 +255,7 @@ impl UsersServiceImpl {
             authz,
             events,
             config,
+            email_rate_limiter: EmailRateLimiter::new(),
         }
     }
 
@@ -585,6 +679,23 @@ impl UsersService for UsersServiceImpl {
             .get(cx.app_id, user_id)
             .await?
             .ok_or(UsersError::NotFound)?;
+        // F-S-002: token-bucket rate limit keyed on (app_id, recipient).
+        // Anonymous public-route callers (cx.principal == None) skip the
+        // authz gate above, so without this an attacker could burn the
+        // SMTP-relay quota by spamming any public endpoint that calls
+        // users::send_verification_email.
+        if self
+            .email_rate_limiter
+            .try_consume(cx.app_id, &user.email)
+            .is_err()
+        {
+            tracing::warn!(
+                app_id = %cx.app_id,
+                user_id = %user.id,
+                "send_verification_email rate-limited"
+            );
+            return Err(UsersError::EmailRateLimited);
+        }
 
         // Mint a single-use token. The hash hits the verifications
         // table; the raw token only ever appears in the email body.
@@ -650,6 +761,23 @@ impl UsersService for UsersServiceImpl {
             return Ok(());
         };
 
+        // F-S-002: rate limit per (app, recipient). Return Ok silently
+        // on exhaustion so an attacker can't probe whether an email
+        // exists by comparing the rate-limit response to the no-match
+        // response.
+        if self
+            .email_rate_limiter
+            .try_consume(cx.app_id, &user.email)
+            .is_err()
+        {
+            tracing::warn!(
+                app_id = %cx.app_id,
+                user_id = %user.id,
+                "request_password_reset rate-limited; suppressing to avoid existence leak"
+            );
+            return Ok(());
+        }
+
         let token = generate_session_token();
         let expires_at = Utc::now()
             + chrono::Duration::from_std(self.config.password_reset_ttl)