fix(audit-2026-06-11/H-B2): inbound_secret mandatory + bad-signature lockout on email triggers

Two gaps:

1. /api/v1/admin/apps/{id}/triggers/email accepted `inbound_secret: null`
   and treated it as "this trigger is unsigned". The runtime then
   skipped HMAC verification entirely. URL discovery (logs, ops
   dashboards, bruteforce on the 122-bit TriggerId space) then fan-out-
   amplified into the outbox for free — anonymous DoS via the
   dispatcher.

2. Even signed triggers had no per-trigger rate limit on signature-
   verify failures, so an attacker who knew the URL could pump unsigned
   POSTs to force Argon2-equivalent work (HMAC verify + nonce-dedup +
   stored-secret decrypt) at line rate.

Fix:

* triggers_api::create_email_trigger now requires a non-empty
  inbound_secret. 400 on missing / empty / whitespace-only.
* email_inbound_api::receive_inbound_email returns 401 immediately when
  the resolved target has no inbound_secret_encrypted column (pre-fix
  rows). The previous `if let Some(ct), Some(nonce)` branch is gone.
* New `BadSignatureLimiter`: per-`(app_id, trigger_id)` sliding-window
  bucket (10 fails / 60 s). On lockout the receiver returns 429 with
  Retry-After instead of 401. record_failure runs on both the
  "no-secret" 401 path and any verify_signature failure.

Test fallout:
* email_inbound integration tests that relied on None secret: removed
  the now-impossible `unsigned_trigger_accepts_without_signature` test,
  replaced with `create_without_inbound_secret_is_rejected` covering
  null / "" / "   "; updated `malformed_body_is_422` and
  `cross_app_path_is_404` to pass + sign with a secret.
* Two new unit tests pin the limiter behavior (burst lockout + per-
  trigger independence).

Audit refs: security_audit/08_dos_resource.md#h-3,
security_audit/09_external_integrations.md#f-ext-m-001.

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

crates/manager-core/src/email_inbound_api.rs

@@ -21,7 +21,7 @@
 
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use axum::body::Bytes;
 use axum::extract::{Path, State};
@@ -88,6 +88,73 @@ impl InboundNonceDedup {
     }
 }
 
+/// Audit 2026-06-11 H-B2 — per-`(app_id, trigger_id)` bad-signature
+/// limiter. After [`BAD_SIG_BURST`] failed verifies within
+/// [`BAD_SIG_WINDOW`], subsequent attempts return 429 with a retry-
+/// after hint rather than 401, so the dispatcher's Argon2-equivalent
+/// (HMAC verify + JSON parse + outbox insert) can't be hammered.
+const BAD_SIG_BURST: u32 = 10;
+const BAD_SIG_WINDOW: Duration = Duration::from_secs(60);
+
+#[derive(Debug, Clone, Copy)]
+struct BadSigBucket {
+    window_started_at: Instant,
+    count: u32,
+}
+
+#[derive(Default)]
+pub struct BadSignatureLimiter {
+    inner: Mutex<HashMap<(AppId, TriggerId), BadSigBucket>>,
+}
+
+impl BadSignatureLimiter {
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Check whether this trigger is currently locked out. Does NOT
+    /// increment — that happens in [`Self::record_failure`].
+    fn check(&self, app_id: AppId, trigger_id: TriggerId) -> Result<(), Duration> {
+        let now = Instant::now();
+        let mut map = self.inner.lock().expect("bad-sig limiter poisoned");
+        // Lazy sweep when crossing soft cap.
+        if map.len() > 4096 {
+            map.retain(|_, b| now.duration_since(b.window_started_at) < BAD_SIG_WINDOW);
+        }
+        let Some(b) = map.get_mut(&(app_id, trigger_id)) else {
+            return Ok(());
+        };
+        if now.duration_since(b.window_started_at) >= BAD_SIG_WINDOW {
+            *b = BadSigBucket {
+                window_started_at: now,
+                count: 0,
+            };
+        }
+        if b.count >= BAD_SIG_BURST {
+            return Err(BAD_SIG_WINDOW.saturating_sub(now.duration_since(b.window_started_at)));
+        }
+        Ok(())
+    }
+
+    /// Record a failed signature verify.
+    fn record_failure(&self, app_id: AppId, trigger_id: TriggerId) {
+        let now = Instant::now();
+        let mut map = self.inner.lock().expect("bad-sig limiter poisoned");
+        let b = map.entry((app_id, trigger_id)).or_insert(BadSigBucket {
+            window_started_at: now,
+            count: 0,
+        });
+        if now.duration_since(b.window_started_at) >= BAD_SIG_WINDOW {
+            *b = BadSigBucket {
+                window_started_at: now,
+                count: 0,
+            };
+        }
+        b.count = b.count.saturating_add(1);
+    }
+}
+
 #[derive(Clone)]
 pub struct EmailInboundState {
     pub triggers: Arc<dyn TriggerRepo>,
@@ -97,6 +164,10 @@ pub struct EmailInboundState {
     /// — cluster mode (v1.3+) will need a shared store. Per-process is
     /// fine for single-node MVP and dev.
     pub nonce_dedup: Arc<InboundNonceDedup>,
+    /// Audit 2026-06-11 H-B2 — bad-signature lockout per
+    /// `(app_id, trigger_id)`. Process-local; cluster needs a shared
+    /// store but v1.1.x is single-node.
+    pub bad_sig_limiter: Arc<BadSignatureLimiter>,
 }
 
 pub fn email_inbound_router(state: EmailInboundState) -> Router {
@@ -146,13 +217,28 @@ async fn receive_inbound_email(
         return Err(EmailInboundError::NotFound);
     }
 
-    // HMAC verification (only when the trigger has a secret configured).
-    if let (Some(ct), Some(nonce)) = (
+    // Audit 2026-06-11 H-B2 — bad-signature lockout. After BAD_SIG_BURST
+    // failures within the window, refuse to even attempt verification
+    // so HMAC + nonce-dedup + outbox-insert can't be hammered.
+    if let Err(retry_after) = s.bad_sig_limiter.check(app_id, trigger_id) {
+        return Err(EmailInboundError::TooManyBadSignatures { retry_after });
+    }
+
+    // Audit 2026-06-11 H-B2 — fail closed when no inbound_secret is
+    // present. Creation now requires a secret (triggers_api), so this
+    // path is unreachable for new triggers; remains as defense-in-depth
+    // for any pre-audit row that still has a null secret column.
+    let (Some(ct), Some(nonce)) = (
         target.inbound_secret_encrypted.as_ref(),
         target.inbound_secret_nonce.as_ref(),
-    ) {
-        let secret = decrypt_secret(&s.master_key, ct, nonce)?;
-        verify_signature(&headers, &body, secret.as_bytes(), &s.nonce_dedup)?;
+    ) else {
+        s.bad_sig_limiter.record_failure(app_id, trigger_id);
+        return Err(EmailInboundError::Unauthorized);
+    };
+    let secret = decrypt_secret(&s.master_key, ct, nonce)?;
+    if let Err(err) = verify_signature(&headers, &body, secret.as_bytes(), &s.nonce_dedup) {
+        s.bad_sig_limiter.record_failure(app_id, trigger_id);
+        return Err(err);
     }
 
     // Parse the generic JSON shape. Malformed → 422.
@@ -266,6 +352,10 @@ pub enum EmailInboundError {
     NotFound,
     #[error("invalid signature")]
     Unauthorized,
+    /// Audit 2026-06-11 H-B2 — too many bad signatures for this trigger;
+    /// locked out until the window rolls over.
+    #[error("too many bad signatures; retry after {retry_after:?}")]
+    TooManyBadSignatures { retry_after: Duration },
     #[error("malformed body: {0}")]
     Malformed(String),
     #[error("backend: {0}")]
@@ -274,28 +364,46 @@ pub enum EmailInboundError {
 
 impl IntoResponse for EmailInboundError {
     fn into_response(self) -> Response {
-        let (status, body) = match &self {
+        let (status, body, retry_after) = match &self {
             Self::NotFound => (
                 StatusCode::NOT_FOUND,
                 json!({ "error": "trigger not found" }),
+                None,
             ),
             Self::Unauthorized => (
                 StatusCode::UNAUTHORIZED,
                 json!({ "error": "invalid or missing signature" }),
+                None,
             ),
+            Self::TooManyBadSignatures { retry_after } => {
+                let secs = retry_after.as_secs().max(1);
+                (
+                    StatusCode::TOO_MANY_REQUESTS,
+                    json!({ "error": "too many bad signatures; trigger locked" }),
+                    Some(secs),
+                )
+            }
             Self::Malformed(m) => (
                 StatusCode::UNPROCESSABLE_ENTITY,
                 json!({ "error": format!("malformed inbound email body: {m}") }),
+                None,
             ),
             Self::Backend(e) => {
                 tracing::error!(error = %e, "inbound email receiver backend error");
                 (
                     StatusCode::INTERNAL_SERVER_ERROR,
                     json!({ "error": "internal error" }),
+                    None,
                 )
             }
         };
-        (status, Json(body)).into_response()
+        let mut resp = (status, Json(body)).into_response();
+        if let Some(secs) = retry_after {
+            if let Ok(v) = axum::http::HeaderValue::from_str(&secs.to_string()) {
+                resp.headers_mut().insert("retry-after", v);
+            }
+        }
+        resp
     }
 }
 
@@ -438,4 +546,34 @@ mod tests {
         let bad: Result<InboundPayload, _> = serde_json::from_slice(br#"{"subject":"hi"}"#);
         assert!(bad.is_err());
     }
+
+    #[test]
+    fn bad_sig_limiter_locks_out_after_burst() {
+        // Audit 2026-06-11 H-B2 closure.
+        let limiter = BadSignatureLimiter::new();
+        let app = AppId::new();
+        let trig = TriggerId::new();
+        // BAD_SIG_BURST consecutive failures still allow check() (the
+        // BURST'th failure is the one that bumps the bucket to full);
+        // the BURST+1th check should be Err.
+        for _ in 0..BAD_SIG_BURST {
+            assert!(limiter.check(app, trig).is_ok());
+            limiter.record_failure(app, trig);
+        }
+        let retry = limiter.check(app, trig).expect_err("should be locked out");
+        assert!(retry <= BAD_SIG_WINDOW);
+    }
+
+    #[test]
+    fn bad_sig_limiter_keys_independently_per_trigger() {
+        let limiter = BadSignatureLimiter::new();
+        let app = AppId::new();
+        let a = TriggerId::new();
+        let b = TriggerId::new();
+        for _ in 0..BAD_SIG_BURST {
+            limiter.record_failure(app, a);
+        }
+        assert!(limiter.check(app, a).is_err(), "a locked");
+        assert!(limiter.check(app, b).is_ok(), "b independent");
+    }
 }

crates/manager-core/src/triggers_api.rs

@@ -583,25 +583,33 @@ async fn create_email_trigger(
     .await?;
     validate_trigger_target(&*s.scripts, app_id, input.script_id).await?;
 
-    // Encrypt the inbound HMAC secret at rest (user-approved deviation
-    // from the brief's plaintext column). An empty/whitespace secret is
-    // treated as "no secret" (unsigned trigger).
-    let (inbound_secret_encrypted, inbound_secret_nonce) = match input.inbound_secret {
-        Some(secret) if !secret.trim().is_empty() => {
-            // 64 KB cap is irrelevant for a signing secret, but `seal`
-            // takes one; reuse the secrets default.
-            let (ct, nonce) = seal(
-                &s.master_key,
-                &serde_json::Value::String(secret),
-                crate::secrets_service::DEFAULT_SECRET_MAX_VALUE_BYTES,
-            )
-            .map_err(|e| {
-                TriggersApiError::Invalid(format!("could not seal inbound_secret: {e}"))
-            })?;
-            (Some(ct), Some(nonce.to_vec()))
+    // Audit 2026-06-11 H-B2 — inbound_secret is mandatory on create.
+    // The previous behavior accepted None / empty and skipped HMAC
+    // verification at runtime, which left the public webhook URL as the
+    // only secret; URL discovery (logs, ops dashboards, brute force on
+    // 122-bit TriggerId space) then fan-out-amplified into the outbox
+    // for free.
+    let secret = match input.inbound_secret {
+        Some(s) if !s.trim().is_empty() => s,
+        _ => {
+            return Err(TriggersApiError::Invalid(
+                "inbound_secret is required (audit 2026-06-11 H-B2): \
+                 an HMAC-signed webhook URL is the only credential separating \
+                 a real provider from an attacker"
+                    .into(),
+            ))
         }
-        _ => (None, None),
     };
+    // 64 KB cap is irrelevant for a signing secret, but `seal` takes one;
+    // reuse the secrets default.
+    let (ct, nonce) = seal(
+        &s.master_key,
+        &serde_json::Value::String(secret),
+        crate::secrets_service::DEFAULT_SECRET_MAX_VALUE_BYTES,
+    )
+    .map_err(|e| TriggersApiError::Invalid(format!("could not seal inbound_secret: {e}")))?;
+    let inbound_secret_encrypted = Some(ct);
+    let inbound_secret_nonce = Some(nonce.to_vec());
 
     let req = CreateEmailTrigger {
         script_id: input.script_id,