collective
Browse / github.com/storytold/artcraft-services / patch
Verify FAL webhook signatures
Verify FAL webhook signatures `/v1/webhooks/fal` accepts any request today (`TODO(bt, 2025-06-03): Handle webhook crypto authentication`). FAL signs every webhook with ED25519 and publishes the public keys as a JWKS, so this checks those signatures as [FAL's webhook docs](https://fal.ai/docs/model-apis/model-endpoints/webhooks) describe. The check runs before the body is parsed or any job is looked up. ## How it works **`fal_client::webhook_signature`** (new): - Reads `X-Fal-Webhook-Request-Id`, `-User-Id`, `-Timestamp` and `-Signature`. - The signed message is `request_id \n user_id \n timestamp \n sha256_hex(raw body)`
- Status
- open
- Score
- 0.00
- Adopters
- 0 other operators
- Author
- guillehr2 (imported from pull request #150, open)
- Base
24cb90e16b5794797c6228040495d087c724dfdc- Size
- +1548 −2 in 24 files
- Review
- pending
- Shared
- 2026-10-28
- Id
87022ee97838791672d80704
How it went
No other operator has reported using it yet.
Diff
raw · Cargo.lock Cargo.toml build/selfhost/.env.example build/selfhost/README.md crates/api_clients/fal_client/Cargo.toml crates/api_clients/fal_client/src/lib.rs crates/api_clients/fal_client/src/webhook_signature/fal_jwks_cache.rs crates/api_clients/fal_client/src/webhook_signature/fal_webhook_public_key.rs crates/api_clients/fal_client/src/webhook_signature/fal_webhook_signature_error.rs crates/api_clients/fal_client/src/webhook_signature/fal_webhook_signature_headers.rs crates/api_clients/fal_client/src/webhook_signature/mod.rs crates/api_clients/fal_client/src/webhook_signature/parse_fal_jwks.rs crates/api_clients/fal_client/src/webhook_signature/verify_fal_webhook_request.rs crates/api_clients/fal_client/src/webhook_signature/verify_fal_webhook_signature.rs crates/service/web/storyteller_web/Cargo.toml crates/service/web/storyteller_web/src/http_server/endpoints/omni_gen/generate/video/tests/support.rs crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal/fal_webhook_handler.rs crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal/fal_webhook_handler_tests.rs crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal/mod.rs crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal/signature/check_fal_webhook_signature.rs crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal/signature/fal_webhook_signature_mode.rs crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal/signature/fal_webhook_signature_settings.rs crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal/signature/mod.rs crates/service/web/storyteller_web/src/startup/setup_inference_providers.rs
@@ -3189,13 +3189,17 @@ name = "fal_client" version = "0.0.1" dependencies = [ "anyhow",+ "base64 0.22.1", "errors", "file_to_base64_url", "futures 0.3.32",+ "hex", "log", "reqwest 0.12.23",+ "ring", "serde", "serde_json",+ "sha2 0.11.0-rc.5", "strum", "test_data", "test_utils",@@ -8741,6 +8745,7 @@ dependencies = [ "google_sign_in", "grok_api_client", "hashing",+ "hex", "hostname 0.4.0", "http_server_common", "images",@@ -8776,6 +8781,7 @@ dependencies = [ "reqwest 0.12.23", "resend-rs", "reusable_types",+ "ring", "rootly_client", "rootly_config", "rust-s3",@@ -307,6 +307,7 @@ elasticsearch = { version = "8.5.0-alpha.1", default-features = false, features env = "1.0.1" env_logger = "0.11.9" futures = { version = "0.3.32" }+hex = "0.4.3" hex-literal = "1.1.0" hostname = "0.4.0" log = "0.4.29"@@ -319,6 +320,7 @@ rand = "0.9.1" redis = { version = "1.0.2", features = ["r2d2", "tokio-rustls-comp", "tokio-comp", "cluster-async", "aio"] } regex = "1.12.3" reqwest = { version = "0.12.23", default-features = false, features = [ "charset", "http2", "system-proxy", "json", "cookies", "rustls-tls", "gzip" ] }+ring = "0.17.8" reqwest-websocket = { version = "0.5.0" } rust-s3 = { version = "0.36.0-beta.2", default-features = false, features = ["fail-on-err", "tags", "tokio-rustls-tls"] } erased-serde = "0.4"@@ -28,6 +28,9 @@ NUM_WORKERS=4 FAL_API_KEY=unset FAL_WEBHOOK_URL=http://localhost:12345/v1/webhooks/fal+# What the FAL webhook endpoint does with a request that isn't signed by FAL:+# enforce (reject it), report (log a warning and accept it; the default) or off.+FAL_WEBHOOK_SIGNATURE_MODE=report GMICLOUD_API_KEY=unset GROK_API_KEY=unset OPENAI_API_KEY=unset@@ -50,6 +50,12 @@ OpenAI, Stripe, Resend and others) to be present at startup. `.env.example` sets them to the placeholder `unset`, which lets the server start. Features that call a provider only work after you add a real key for it. +`FAL_WEBHOOK_SIGNATURE_MODE` controls the check that FAL webhooks are signed+by FAL (with the public keys FAL publishes at+`https://rest.fal.ai/.well-known/jwks.json`): `enforce` rejects requests that+don't verify, `report` (the default) logs them and accepts them, and `off`+skips the check.+ Provider webhooks, such as `FAL_WEBHOOK_URL`, must be reachable from the internet for asynchronous generation results to arrive. @@ -22,11 +22,15 @@ file_to_base64_url.workspace = true # External anyhow.workspace = true+base64.workspace = true futures.workspace = true+hex.workspace = true log.workspace = true reqwest.workspace = true+ring.workspace = true serde.workspace = true serde_json.workspace = true+sha2.workspace = true strum.workspace = true thiserror.workspace = true tokio.workspace = true@@ -38,4 +42,4 @@ test_utils.workspace = true test_data.workspace = true # External Dev-tokio = { workspace = true, features = ["rt", "macros", "net", "io-util"] }+tokio = { workspace = true, features = ["rt", "macros", "net", "io-util", "test-util"] }@@ -7,3 +7,4 @@ pub mod requests; pub mod requests_old; pub mod utils; pub mod webhook_payload;+pub mod webhook_signature;new file mode 100644@@ -0,0 +1,359 @@+use std::future::Future;+use std::pin::Pin;+use std::sync::Arc;+use std::time::Duration;++use log::{info, warn};+use tokio::sync::Mutex;+use tokio::time::Instant;++use crate::webhook_signature::fal_webhook_public_key::FalWebhookPublicKey;+use crate::webhook_signature::fal_webhook_signature_error::FalWebhookSignatureError;+use crate::webhook_signature::parse_fal_jwks::parse_fal_jwks;++/// Where FAL publishes the public keys its webhooks are signed with.+pub const FAL_JWKS_URL: &str = "https://rest.fal.ai/.well-known/jwks.json";++/// Fetched keys are used for this long before they are fetched again.+pub const REFRESH_AFTER: Duration = Duration::from_secs(12 * 60 * 60);++/// Keys are never used once they are this old, even when a refresh fails: FAL's docs say the+/// JWKS may be cached for up to 24 hours, because the keys can change.+pub const MAX_KEY_AGE: Duration = Duration::from_secs(24 * 60 * 60);++/// A signature that no cached key verifies triggers an early refresh (FAL may have rotated its+/// keys), but at most once per this interval, so bad requests can't make us hammer FAL.+pub const MIN_UNKNOWN_KEY_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60);++/// After a failed fetch, no new fetch is attempted for this long. The cache's lock is held+/// while fetching, so without this a JWKS outage would make every webhook wait on a timeout.+pub const RETRY_AFTER_FAILED_FETCH: Duration = Duration::from_secs(60);++const FETCH_TIMEOUT: Duration = Duration::from_secs(10);++/// Returns the JWKS document as text.+pub type JwksFetchFuture = Pin<Box<dyn Future<Output = Result<String, FalWebhookSignatureError>> + Send>>;+pub type JwksFetcher = Arc<dyn Fn() -> JwksFetchFuture + Send + Sync>;++/// Caches FAL's webhook public keys, fetching them when they are missing or stale.+///+/// Concurrent callers share one fetch: the lock is held while fetching.+pub struct FalJwksCache {+ fetcher: JwksFetcher,+ state: Mutex<CacheState>,+}++struct CacheState {+ keys: Vec<FalWebhookPublicKey>,+ fetched_at: Option<Instant>,+ last_unknown_key_refresh: Option<Instant>,+ /// No fetch is attempted before this (set after a failed fetch).+ maybe_retry_at: Option<Instant>,+}++impl FalJwksCache {+ /// A cache that fetches FAL's JWKS from [`FAL_JWKS_URL`].+ pub fn for_fal() -> Self {+ Self::with_fetcher(Arc::new(|| Box::pin(fetch_fal_jwks())))+ }++ /// A cache that gets the JWKS document from `fetcher` (used by tests and alternative sources).+ pub fn with_fetcher(fetcher: JwksFetcher) -> Self {+ Self {+ fetcher,+ state: Mutex::new(CacheState {+ keys: Vec::new(),+ fetched_at: None,+ last_unknown_key_refresh: None,+ maybe_retry_at: None,+ }),+ }+ }++ /// The keys to verify with: the cached ones while they are fresh, otherwise newly fetched.+ ///+ /// If a refresh fails, keys younger than [`MAX_KEY_AGE`] are still used (and the failure is+ /// logged), so a brief JWKS outage doesn't reject genuine webhooks. After a failure, fetching+ /// waits [`RETRY_AFTER_FAILED_FETCH`] before it is tried again.+ pub async fn keys(&self) -> Result<Vec<FalWebhookPublicKey>, FalWebhookSignatureError> {+ let mut state = self.state.lock().await;+ let now = Instant::now();++ let age = state.fetched_at.map(|fetched_at| now.saturating_duration_since(fetched_at));+ let usable = age.is_some_and(|age| age < MAX_KEY_AGE);++ if age.is_some_and(|age| age < REFRESH_AFTER) {+ return Ok(state.keys.clone());+ }++ if state.maybe_retry_at.is_some_and(|retry_at| now < retry_at) {+ return if usable {+ Ok(state.keys.clone())+ } else {+ Err(FalWebhookSignatureError::JwksFetchFailed("waiting to retry after a failed fetch".to_string()))+ };+ }++ match self.fetch().await {+ Ok(keys) => {+ state.keys = keys;+ state.fetched_at = Some(now);+ state.maybe_retry_at = None;+ Ok(state.keys.clone())+ }+ Err(err) => {+ state.maybe_retry_at = Some(now + RETRY_AFTER_FAILED_FETCH);+ if usable {+ warn!("Could not refresh the FAL JWKS, using keys fetched {:?} ago: {}", age, err);+ Ok(state.keys.clone())+ } else {+ Err(err)+ }+ }+ }+ }++ /// Fetch the keys again now, because a signature matched none of the cached ones and FAL may+ /// have rotated its keys. Rate-limited to once per [`MIN_UNKNOWN_KEY_REFRESH_INTERVAL`].+ ///+ /// Returns whether new keys were fetched (so a retry can make a difference).+ pub async fn refresh_after_unknown_key(&self) -> Result<bool, FalWebhookSignatureError> {+ let mut state = self.state.lock().await;+ let now = Instant::now();++ let recently = state.last_unknown_key_refresh+ .is_some_and(|last| now.saturating_duration_since(last) < MIN_UNKNOWN_KEY_REFRESH_INTERVAL);++ if recently {+ return Ok(false);+ }++ state.last_unknown_key_refresh = Some(now);++ let keys = self.fetch().await?;+ let changed = keys != state.keys;++ state.keys = keys;+ state.fetched_at = Some(now);++ if changed {+ info!("The FAL JWKS changed; using the new keys");+ }++ Ok(changed)+ }++ async fn fetch(&self) -> Result<Vec<FalWebhookPublicKey>, FalWebhookSignatureError> {+ let body = (self.fetcher)().await?;+ parse_fal_jwks(&body)+ }+}++async fn fetch_fal_jwks() -> Result<String, FalWebhookSignatureError> {+ let failed = |err: reqwest::Error| FalWebhookSignatureError::JwksFetchFailed(err.to_string());++ let response = reqwest::Client::builder()+ .timeout(FETCH_TIMEOUT)+ .build()+ .map_err(failed)?+ .get(FAL_JWKS_URL)+ .send()+ .await+ .map_err(failed)?;++ let status = response.status();++ if !status.is_success() {+ return Err(FalWebhookSignatureError::JwksFetchFailed(format!("HTTP {}", status)));+ }++ response.text().await.map_err(failed)+}++#[cfg(test)]+mod tests {+ use std::sync::atomic::{AtomicUsize, Ordering};+ use std::sync::Mutex as StdMutex;++ use super::*;++ const KEY_A: &str = r#"{"keys":[{"kty":"OKP","crv":"Ed25519","x":"Cghs6Ge6h5B3_vMIurJPYmN23Yk-VcTk-juiIEQ6zuY="}]}"#;+ const KEY_B: &str = r#"{"keys":[{"kty":"OKP","crv":"Ed25519","x":"TZE-E0LB1iT29Ai1LbAh9bam1YQJkK7fEGWAmmkRm_I="}]}"#;++ mod caching {+ use super::*;++ #[tokio::test(start_paused = true)]+ async fn fetches_once_while_the_keys_are_fresh() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ cache.keys().await.unwrap();+ tokio::time::advance(REFRESH_AFTER - Duration::from_secs(1)).await;+ cache.keys().await.unwrap();++ assert_eq!(source.fetches(), 1);+ }++ #[tokio::test(start_paused = true)]+ async fn refetches_once_the_keys_are_stale() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ let before = cache.keys().await.unwrap();+ source.serve(KEY_B);+ tokio::time::advance(REFRESH_AFTER).await;+ let after = cache.keys().await.unwrap();++ assert_eq!(source.fetches(), 2);+ assert_ne!(before, after);+ }++ #[tokio::test(start_paused = true)]+ async fn keeps_using_stale_keys_through_a_failed_refresh() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ let fetched = cache.keys().await.unwrap();+ source.fail();+ tokio::time::advance(REFRESH_AFTER).await;++ assert_eq!(cache.keys().await.unwrap(), fetched);+ }++ #[tokio::test(start_paused = true)]+ async fn never_uses_keys_older_than_the_max_age() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ cache.keys().await.unwrap();+ source.fail();+ tokio::time::advance(MAX_KEY_AGE).await;++ assert!(matches!(cache.keys().await, Err(FalWebhookSignatureError::JwksFetchFailed(_))));+ }++ #[tokio::test(start_paused = true)]+ async fn waits_before_retrying_a_failed_fetch() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ cache.keys().await.unwrap();+ source.fail();+ tokio::time::advance(REFRESH_AFTER).await;+ cache.keys().await.unwrap();+ cache.keys().await.unwrap();++ assert_eq!(source.fetches(), 2);++ source.serve(KEY_B);+ tokio::time::advance(RETRY_AFTER_FAILED_FETCH).await;++ assert_eq!(cache.keys().await.unwrap(), parse_fal_jwks(KEY_B).unwrap());+ assert_eq!(source.fetches(), 3);+ }++ #[tokio::test(start_paused = true)]+ async fn reports_a_failed_first_fetch() {+ let source = FakeSource::new(KEY_A);+ source.fail();++ assert!(matches!(source.cache().keys().await, Err(FalWebhookSignatureError::JwksFetchFailed(_))));+ }+ }++ mod unknown_key_refresh {+ use super::*;++ #[tokio::test(start_paused = true)]+ async fn picks_up_rotated_keys() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ cache.keys().await.unwrap();+ source.serve(KEY_B);++ assert!(cache.refresh_after_unknown_key().await.unwrap());+ assert_eq!(cache.keys().await.unwrap(), parse_fal_jwks(KEY_B).unwrap());+ }++ #[tokio::test(start_paused = true)]+ async fn reports_no_change_when_the_keys_are_the_same() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ cache.keys().await.unwrap();++ assert!(!cache.refresh_after_unknown_key().await.unwrap());+ }++ #[tokio::test(start_paused = true)]+ async fn is_rate_limited() {+ let source = FakeSource::new(KEY_A);+ let cache = source.cache();++ cache.keys().await.unwrap();+ cache.refresh_after_unknown_key().await.unwrap();+ source.serve(KEY_B);+ tokio::time::advance(MIN_UNKNOWN_KEY_REFRESH_INTERVAL - Duration::from_secs(1)).await;++ assert!(!cache.refresh_after_unknown_key().await.unwrap());+ assert_eq!(source.fetches(), 2);++ tokio::time::advance(Duration::from_secs(1)).await;++ assert!(cache.refresh_after_unknown_key().await.unwrap());+ assert_eq!(source.fetches(), 3);+ }+ }++ /// Fetches FAL's real JWKS. Needs network access, so it only runs on request:+ /// `cargo test -p fal_client -- --ignored fetches_fals_live_jwks`+ #[tokio::test]+ #[ignore]+ async fn fetches_fals_live_jwks() {+ let keys = FalJwksCache::for_fal().keys().await.unwrap();++ assert!(!keys.is_empty());+ }++ /// A JWKS source the test controls: what it serves, whether it fails, and how often it was asked.+ struct FakeSource {+ body: Arc<StdMutex<Option<String>>>,+ fetches: Arc<AtomicUsize>,+ }++ impl FakeSource {+ fn new(body: &str) -> Self {+ Self {+ body: Arc::new(StdMutex::new(Some(body.to_string()))),+ fetches: Arc::new(AtomicUsize::new(0)),+ }+ }++ fn cache(&self) -> FalJwksCache {+ let body = self.body.clone();+ let fetches = self.fetches.clone();++ FalJwksCache::with_fetcher(Arc::new(move || {+ fetches.fetch_add(1, Ordering::SeqCst);+ let served = body.lock().unwrap().clone();+ Box::pin(async move {+ served.ok_or_else(|| FalWebhookSignatureError::JwksFetchFailed("test source is down".to_string()))+ })+ }))+ }++ fn serve(&self, body: &str) {+ *self.body.lock().unwrap() = Some(body.to_string());+ }++ fn fail(&self) {+ *self.body.lock().unwrap() = None;+ }++ fn fetches(&self) -> usize {+ self.fetches.load(Ordering::SeqCst)+ }+ }+}new file mode 100644@@ -0,0 +1,24 @@+use std::fmt::{Debug, Formatter};++/// An ED25519 public key that FAL signs its webhooks with.+///+/// FAL publishes these in its JWKS (see `parse_fal_jwks`); each key's `x` member is the+/// 32-byte public key.+#[derive(Clone, PartialEq, Eq)]+pub struct FalWebhookPublicKey([u8; 32]);++impl FalWebhookPublicKey {+ pub fn from_bytes(bytes: [u8; 32]) -> Self {+ Self(bytes)+ }++ pub fn as_bytes(&self) -> &[u8; 32] {+ &self.0+ }+}++impl Debug for FalWebhookPublicKey {+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {+ write!(f, "FalWebhookPublicKey({})", hex::encode(self.0))+ }+}new file mode 100644@@ -0,0 +1,49 @@+use std::error::Error;+use std::fmt::{Display, Formatter};++/// Why a FAL webhook request could not be verified as coming from FAL.+#[derive(Debug, Clone, PartialEq, Eq)]+pub enum FalWebhookSignatureError {+ /// A required `X-Fal-Webhook-*` header is absent (or not valid UTF-8).+ MissingHeader(&'static str),++ /// `X-Fal-Webhook-Timestamp` is not a whole number of seconds.+ InvalidTimestamp(String),++ /// The timestamp is further than the tolerance from the current time (a stale or replayed request).+ TimestampOutOfTolerance { timestamp: i64, now: i64 },++ /// `X-Fal-Webhook-Signature` is not 64 bytes of hex.+ InvalidSignatureEncoding,++ /// There are no public keys to verify against.+ NoPublicKeys,++ /// The signature is well formed but no known public key verifies it.+ SignatureMismatch,++ /// The JWKS could not be fetched.+ JwksFetchFailed(String),++ /// The JWKS could not be parsed, or holds no usable ED25519 key.+ JwksParseFailed(String),+}++impl Error for FalWebhookSignatureError {}++impl Display for FalWebhookSignatureError {+ fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {+ match self {+ Self::MissingHeader(name) => write!(f, "missing webhook header {}", name),+ Self::InvalidTimestamp(value) => write!(f, "webhook timestamp is not an integer: {:?}", value),+ Self::TimestampOutOfTolerance { timestamp, now } => {+ write!(f, "webhook timestamp {} is too far from the current time {}", timestamp, now)+ }+ Self::InvalidSignatureEncoding => write!(f, "webhook signature is not 64 bytes of hex"),+ Self::NoPublicKeys => write!(f, "no FAL webhook public keys to verify against"),+ Self::SignatureMismatch => write!(f, "webhook signature does not match any FAL public key"),+ Self::JwksFetchFailed(reason) => write!(f, "could not fetch the FAL JWKS: {}", reason),+ Self::JwksParseFailed(reason) => write!(f, "could not parse the FAL JWKS: {}", reason),+ }+ }+}new file mode 100644@@ -0,0 +1,84 @@+use crate::webhook_signature::fal_webhook_signature_error::FalWebhookSignatureError;++pub const REQUEST_ID_HEADER: &str = "X-Fal-Webhook-Request-Id";+pub const USER_ID_HEADER: &str = "X-Fal-Webhook-User-Id";+pub const TIMESTAMP_HEADER: &str = "X-Fal-Webhook-Timestamp";+pub const SIGNATURE_HEADER: &str = "X-Fal-Webhook-Signature";++/// The four headers FAL sends with every webhook so the receiver can verify it.+///+/// Values are kept exactly as received: the timestamp in particular is signed as the+/// header's text, not as a re-formatted number.+#[derive(Debug, Clone, PartialEq, Eq)]+pub struct FalWebhookSignatureHeaders {+ pub request_id: String,+ pub user_id: String,+ pub timestamp: String,+ /// Hex-encoded ED25519 signature.+ pub signature: String,+}++impl FalWebhookSignatureHeaders {+ /// Read the headers through `get_header`, which looks one up by name.+ ///+ /// HTTP header names are case-insensitive; `get_header` is expected to honour that (as the+ /// header maps of actix-web, reqwest and `http` do).+ pub fn from_lookup<'a, F>(get_header: F) -> Result<Self, FalWebhookSignatureError>+ where+ F: Fn(&str) -> Option<&'a str>,+ {+ let required = |name: &'static str| {+ get_header(name)+ .map(str::to_string)+ .ok_or(FalWebhookSignatureError::MissingHeader(name))+ };++ Ok(Self {+ request_id: required(REQUEST_ID_HEADER)?,+ user_id: required(USER_ID_HEADER)?,+ timestamp: required(TIMESTAMP_HEADER)?,+ signature: required(SIGNATURE_HEADER)?,+ })+ }+}++#[cfg(test)]+mod tests {+ use std::collections::HashMap;++ use super::*;++ #[test]+ fn reads_all_four_headers() {+ let headers = header_map(&[+ (REQUEST_ID_HEADER, "req-1"),+ (USER_ID_HEADER, "user-1"),+ (TIMESTAMP_HEADER, "1700000000"),+ (SIGNATURE_HEADER, "abcd"),+ ]);++ let parsed = FalWebhookSignatureHeaders::from_lookup(|name| headers.get(name).map(String::as_str)).unwrap();++ assert_eq!(parsed.request_id, "req-1");+ assert_eq!(parsed.user_id, "user-1");+ assert_eq!(parsed.timestamp, "1700000000");+ assert_eq!(parsed.signature, "abcd");+ }++ #[test]+ fn names_the_first_missing_header() {+ let headers = header_map(&[+ (REQUEST_ID_HEADER, "req-1"),+ (USER_ID_HEADER, "user-1"),+ (SIGNATURE_HEADER, "abcd"),+ ]);++ let err = FalWebhookSignatureHeaders::from_lookup(|name| headers.get(name).map(String::as_str)).unwrap_err();++ assert_eq!(err, FalWebhookSignatureError::MissingHeader(TIMESTAMP_HEADER));+ }++ fn header_map(pairs: &[(&str, &str)]) -> HashMap<String, String> {+ pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()+ }+}new file mode 100644@@ -0,0 +1,7 @@+pub mod fal_jwks_cache;+pub mod fal_webhook_public_key;+pub mod fal_webhook_signature_error;+pub mod fal_webhook_signature_headers;+pub mod parse_fal_jwks;+pub mod verify_fal_webhook_request;+pub mod verify_fal_webhook_signature;new file mode 100644@@ -0,0 +1,102 @@+use base64::engine::general_purpose::URL_SAFE_NO_PAD;+use base64::Engine;+use serde::Deserialize;++use crate::webhook_signature::fal_webhook_public_key::FalWebhookPublicKey;+use crate::webhook_signature::fal_webhook_signature_error::FalWebhookSignatureError;++#[derive(Deserialize)]+struct Jwks {+ keys: Vec<Jwk>,+}++#[derive(Deserialize)]+struct Jwk {+ kty: Option<String>,+ crv: Option<String>,+ x: Option<String>,+}++/// Parse FAL's JWKS (`https://rest.fal.ai/.well-known/jwks.json`) into its ED25519 public keys.+///+/// Keys of any other type are skipped. It is an error if no ED25519 key is left, because then+/// no webhook could ever verify.+///+/// FAL encodes `x` as base64url *with* `=` padding, although RFC 7515 calls for none, so both+/// forms are accepted.+pub fn parse_fal_jwks(json: &str) -> Result<Vec<FalWebhookPublicKey>, FalWebhookSignatureError> {+ let jwks: Jwks = serde_json::from_str(json)+ .map_err(|err| FalWebhookSignatureError::JwksParseFailed(err.to_string()))?;++ let keys: Vec<FalWebhookPublicKey> = jwks.keys.iter().filter_map(ed25519_key).collect();++ if keys.is_empty() {+ return Err(FalWebhookSignatureError::JwksParseFailed("the JWKS has no ED25519 keys".to_string()));+ }++ Ok(keys)+}++fn ed25519_key(jwk: &Jwk) -> Option<FalWebhookPublicKey> {+ let is_okp = jwk.kty.as_deref().map_or(true, |kty| kty == "OKP");+ if !is_okp || jwk.crv.as_deref() != Some("Ed25519") {+ return None;+ }++ let x = jwk.x.as_deref()?.trim_end_matches('=');+ let bytes: [u8; 32] = URL_SAFE_NO_PAD.decode(x).ok()?.try_into().ok()?;++ Some(FalWebhookPublicKey::from_bytes(bytes))+}++#[cfg(test)]+mod tests {+ use super::*;++ /// FAL's JWKS as served on 2026-10-15: two keys (so keys can rotate), `x` padded.+ const FAL_JWKS_2026_10_15: &str = r#"{"keys":[{"kty":"OKP","crv":"Ed25519","kid":"Eyv1ENn6NyZnEorJHsWnEDHRQ45-hnGqP0Lb-yX7Itw","x":"Cghs6Ge6h5B3_vMIurJPYmN23Yk-VcTk-juiIEQ6zuY=","use":"sig"},{"kty":"OKP","crv":"Ed25519","kid":"q_RdVYvPKoXPwp5YOM4bhSvtWynZeoAWGOyxQiXgsmQ","x":"TZE-E0LB1iT29Ai1LbAh9bam1YQJkK7fEGWAmmkRm_I=","use":"sig"}]}"#;++ #[test]+ fn parses_fals_published_jwks() {+ let keys = parse_fal_jwks(FAL_JWKS_2026_10_15).unwrap();++ assert_eq!(keys.len(), 2);+ assert_eq!(+ hex::encode(keys[0].as_bytes()),+ "0a086ce867ba879077fef308bab24f626376dd893e55c4e4fa3ba220443acee6",+ );+ }++ #[test]+ fn accepts_x_without_padding() {+ let keys = parse_fal_jwks(r#"{"keys":[{"kty":"OKP","crv":"Ed25519","x":"Cghs6Ge6h5B3_vMIurJPYmN23Yk-VcTk-juiIEQ6zuY"}]}"#).unwrap();++ assert_eq!(keys.len(), 1);+ }++ #[test]+ fn skips_keys_that_are_not_ed25519() {+ let json = r#"{"keys":[+ {"kty":"RSA","n":"abc","e":"AQAB"},+ {"kty":"OKP","crv":"X25519","x":"Cghs6Ge6h5B3_vMIurJPYmN23Yk-VcTk-juiIEQ6zuY"},+ {"kty":"OKP","crv":"Ed25519","x":"TZE-E0LB1iT29Ai1LbAh9bam1YQJkK7fEGWAmmkRm_I="}+ ]}"#;++ let keys = parse_fal_jwks(json).unwrap();++ assert_eq!(keys.len(), 1);+ }++ #[test]+ fn skips_keys_of_the_wrong_length() {+ let err = parse_fal_jwks(r#"{"keys":[{"kty":"OKP","crv":"Ed25519","x":"AAAA"}]}"#).unwrap_err();++ assert!(matches!(err, FalWebhookSignatureError::JwksParseFailed(_)));+ }++ #[test]+ fn rejects_a_jwks_without_usable_keys() {+ assert!(matches!(parse_fal_jwks(r#"{"keys":[]}"#), Err(FalWebhookSignatureError::JwksParseFailed(_))));+ assert!(matches!(parse_fal_jwks("not json"), Err(FalWebhookSignatureError::JwksParseFailed(_))));+ }+}new file mode 100644@@ -0,0 +1,160 @@+use crate::webhook_signature::fal_jwks_cache::FalJwksCache;+use crate::webhook_signature::fal_webhook_signature_error::FalWebhookSignatureError;+use crate::webhook_signature::fal_webhook_signature_headers::FalWebhookSignatureHeaders;+use crate::webhook_signature::verify_fal_webhook_signature::verify_fal_webhook_signature;++/// Verify a webhook request against FAL's published keys, fetching them through `jwks_cache`.+///+/// When no cached key matches, the keys are fetched again once (FAL may have rotated them) and+/// the signature is checked against the new ones. Other failures (stale timestamp, malformed+/// headers) never trigger a fetch.+pub async fn verify_fal_webhook_request(+ jwks_cache: &FalJwksCache,+ headers: &FalWebhookSignatureHeaders,+ raw_body: &[u8],+ now_unix_seconds: i64,+) -> Result<(), FalWebhookSignatureError> {+ let keys = jwks_cache.keys().await?;++ match verify_fal_webhook_signature(headers, raw_body, &keys, now_unix_seconds) {+ Err(FalWebhookSignatureError::SignatureMismatch) => {+ if jwks_cache.refresh_after_unknown_key().await? {+ let keys = jwks_cache.keys().await?;+ verify_fal_webhook_signature(headers, raw_body, &keys, now_unix_seconds)+ } else {+ Err(FalWebhookSignatureError::SignatureMismatch)+ }+ }+ result => result,+ }+}++#[cfg(test)]+mod tests {+ use std::sync::atomic::{AtomicUsize, Ordering};+ use std::sync::{Arc, Mutex};++ use base64::engine::general_purpose::URL_SAFE;+ use base64::Engine;+ use ring::rand::SystemRandom;+ use ring::signature::{Ed25519KeyPair, KeyPair};+ use sha2::{Digest, Sha256};++ use super::*;++ const NOW: i64 = 1_760_500_000;+ const BODY: &[u8] = br#"{"request_id":"req-1","status":"OK","payload":{}}"#;++ #[tokio::test(start_paused = true)]+ async fn verifies_with_the_cached_keys() {+ let signer = Signer::new();+ let source = Source::serving(&[&signer]);++ let result = verify_fal_webhook_request(&source.cache(), &signer.sign(BODY), BODY, NOW).await;++ assert_eq!(result, Ok(()));+ assert_eq!(source.fetches(), 1);+ }++ #[tokio::test(start_paused = true)]+ async fn picks_up_a_rotated_key_with_one_extra_fetch() {+ let old = Signer::new();+ let new = Signer::new();+ let source = Source::serving(&[&old]);+ let cache = source.cache();+ cache.keys().await.unwrap();++ source.serve(&[&new]);+ let result = verify_fal_webhook_request(&cache, &new.sign(BODY), BODY, NOW).await;++ assert_eq!(result, Ok(()));+ assert_eq!(source.fetches(), 2);+ }++ #[tokio::test(start_paused = true)]+ async fn a_forged_signature_fails_after_at_most_one_extra_fetch() {+ let genuine = Signer::new();+ let forger = Signer::new();+ let source = Source::serving(&[&genuine]);+ let cache = source.cache();++ for _ in 0..3 {+ let result = verify_fal_webhook_request(&cache, &forger.sign(BODY), BODY, NOW).await;+ assert_eq!(result, Err(FalWebhookSignatureError::SignatureMismatch));+ }++ assert_eq!(source.fetches(), 2);+ }++ #[tokio::test(start_paused = true)]+ async fn a_stale_timestamp_never_triggers_a_fetch() {+ let signer = Signer::new();+ let source = Source::serving(&[&signer]);+ let cache = source.cache();++ let result = verify_fal_webhook_request(&cache, &signer.sign(BODY), BODY, NOW + 3600).await;++ assert!(matches!(result, Err(FalWebhookSignatureError::TimestampOutOfTolerance { .. })));+ assert_eq!(source.fetches(), 1);+ }++ struct Signer {+ key_pair: Ed25519KeyPair,+ }++ impl Signer {+ fn new() -> Self {+ let pkcs8 = Ed25519KeyPair::generate_pkcs8(&SystemRandom::new()).unwrap();+ Self { key_pair: Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap() }+ }++ /// The JWK FAL would publish for this key (`x` padded, as FAL serves it).+ fn jwk(&self) -> String {+ let x = URL_SAFE.encode(self.key_pair.public_key().as_ref());+ format!(r#"{{"kty":"OKP","crv":"Ed25519","x":"{x}","use":"sig"}}"#)+ }++ fn sign(&self, body: &[u8]) -> FalWebhookSignatureHeaders {+ let (request_id, user_id, timestamp) = ("req-1", "user-1", NOW.to_string());+ let message = format!("{request_id}\n{user_id}\n{timestamp}\n{}", hex::encode(Sha256::digest(body)));+ FalWebhookSignatureHeaders {+ request_id: request_id.to_string(),+ user_id: user_id.to_string(),+ timestamp,+ signature: hex::encode(self.key_pair.sign(message.as_bytes()).as_ref()),+ }+ }+ }++ struct Source {+ body: Arc<Mutex<String>>,+ fetches: Arc<AtomicUsize>,+ }++ impl Source {+ fn serving(signers: &[&Signer]) -> Self {+ let source = Self { body: Arc::new(Mutex::new(String::new())), fetches: Arc::new(AtomicUsize::new(0)) };+ source.serve(signers);+ source+ }++ fn serve(&self, signers: &[&Signer]) {+ let keys: Vec<String> = signers.iter().map(|signer| signer.jwk()).collect();+ *self.body.lock().unwrap() = format!(r#"{{"keys":[{}]}}"#, keys.join(","));+ }++ fn cache(&self) -> FalJwksCache {+ let body = self.body.clone();+ let fetches = self.fetches.clone();+ FalJwksCache::with_fetcher(Arc::new(move || {+ fetches.fetch_add(1, Ordering::SeqCst);+ let served = body.lock().unwrap().clone();+ Box::pin(async move { Ok(served) })+ }))+ }++ fn fetches(&self) -> usize {+ self.fetches.load(Ordering::SeqCst)+ }+ }+}new file mode 100644@@ -0,0 +1,245 @@+use ring::signature::{UnparsedPublicKey, ED25519};+use sha2::{Digest, Sha256};++use crate::webhook_signature::fal_webhook_public_key::FalWebhookPublicKey;+use crate::webhook_signature::fal_webhook_signature_error::FalWebhookSignatureError;+use crate::webhook_signature::fal_webhook_signature_headers::FalWebhookSignatureHeaders;++/// FAL webhooks older or newer than this (in seconds) are rejected, as FAL's docs require.+pub const TIMESTAMP_TOLERANCE_SECONDS: i64 = 300;++const ED25519_SIGNATURE_LENGTH: usize = 64;++/// Verify that a webhook was signed by FAL, following FAL's published scheme:+///+/// 1. The timestamp must be within [`TIMESTAMP_TOLERANCE_SECONDS`] of `now_unix_seconds`.+/// 2. The signed message is the request ID, user ID, timestamp and the hex SHA-256 of the+/// *raw* request body, joined by `\n`.+/// 3. The hex signature must verify as ED25519 under at least one of `public_keys`.+///+/// This checks the signature only. It does not compare the header's request ID with the one in+/// the body; callers that act on the body's request ID should do that as well.+pub fn verify_fal_webhook_signature(+ headers: &FalWebhookSignatureHeaders,+ raw_body: &[u8],+ public_keys: &[FalWebhookPublicKey],+ now_unix_seconds: i64,+) -> Result<(), FalWebhookSignatureError> {+ let timestamp: i64 = headers.timestamp.trim().parse()+ .map_err(|_| FalWebhookSignatureError::InvalidTimestamp(headers.timestamp.clone()))?;++ // i128 so that extreme (hostile) timestamps can't overflow the subtraction.+ if (now_unix_seconds as i128 - timestamp as i128).abs() > TIMESTAMP_TOLERANCE_SECONDS as i128 {+ return Err(FalWebhookSignatureError::TimestampOutOfTolerance { timestamp, now: now_unix_seconds });+ }++ let signature = hex::decode(headers.signature.trim())+ .map_err(|_| FalWebhookSignatureError::InvalidSignatureEncoding)?;++ if signature.len() != ED25519_SIGNATURE_LENGTH {+ return Err(FalWebhookSignatureError::InvalidSignatureEncoding);+ }++ if public_keys.is_empty() {+ return Err(FalWebhookSignatureError::NoPublicKeys);+ }++ let message = signed_message(headers, raw_body);++ let verified = public_keys.iter().any(|key| {+ UnparsedPublicKey::new(&ED25519, key.as_bytes())+ .verify(message.as_bytes(), &signature)+ .is_ok()+ });++ if verified {+ Ok(())+ } else {+ Err(FalWebhookSignatureError::SignatureMismatch)+ }+}++/// The exact bytes FAL signs. The timestamp is the header's own text, not a re-formatted number.+fn signed_message(headers: &FalWebhookSignatureHeaders, raw_body: &[u8]) -> String {+ let body_hash = hex::encode(Sha256::digest(raw_body));+ format!("{}\n{}\n{}\n{}", headers.request_id, headers.user_id, headers.timestamp, body_hash)+}++#[cfg(test)]+mod tests {+ use ring::rand::SystemRandom;+ use ring::signature::{Ed25519KeyPair, KeyPair};++ use super::*;++ // An independent test vector, made with PyNaCl (libsodium) by following FAL's documented+ // algorithm, so the message format isn't only checked against this file's own code:+ //+ // sk = nacl.signing.SigningKey(bytes(range(32)))+ // message = "\n".join([request_id, user_id, timestamp, sha256(body).hexdigest()])+ // sk.sign(message.encode()).signature.hex()+ const VECTOR_PUBLIC_KEY: &str = "03a107bff3ce10be1d70dd18e74bc09967e4d6309ba50d5f1ddc8664125531b8";+ const VECTOR_REQUEST_ID: &str = "019e18d8-8c36-7bc1-aa77-2bc2f70268c6";+ const VECTOR_USER_ID: &str = "test-user-1";+ const VECTOR_TIMESTAMP: &str = "1760500000";+ const VECTOR_BODY: &[u8] = br#"{"request_id":"019e18d8-8c36-7bc1-aa77-2bc2f70268c6","status":"OK","payload":{"images":[]}}"#;+ const VECTOR_SIGNATURE: &str = "d925f70845ad2e5a410c3b671a443a27cf538bf1f20b9918b6ce583f0f3640c6f9b2a407a442fa3d770e914ea27f4d4bfc07fd64f3496672e0da6af9a930040e";+ const VECTOR_NOW: i64 = 1760500000;++ mod independent_vector {+ use super::*;++ #[test]+ fn verifies() {+ assert_eq!(verify(&vector_headers(), VECTOR_BODY, &[vector_key()], VECTOR_NOW), Ok(()));+ }++ #[test]+ fn verifies_with_the_right_key_among_others() {+ let keys = [random_key_pair().1, vector_key(), random_key_pair().1];++ assert_eq!(verify(&vector_headers(), VECTOR_BODY, &keys, VECTOR_NOW), Ok(()));+ }++ #[test]+ fn rejects_a_changed_body() {+ let body = br#"{"request_id":"019e18d8-8c36-7bc1-aa77-2bc2f70268c6","status":"OK","payload":{"images":[1]}}"#;++ assert_eq!(verify(&vector_headers(), body, &[vector_key()], VECTOR_NOW), Err(FalWebhookSignatureError::SignatureMismatch));+ }++ #[test]+ fn rejects_a_changed_request_id_user_id_or_timestamp() {+ let changed = [+ FalWebhookSignatureHeaders { request_id: "019e18d8-8c36-7bc1-aa77-2bc2f70268c7".to_string(), ..vector_headers() },+ FalWebhookSignatureHeaders { user_id: "test-user-2".to_string(), ..vector_headers() },+ FalWebhookSignatureHeaders { timestamp: "1760500001".to_string(), ..vector_headers() },+ ];++ for headers in changed {+ assert_eq!(verify(&headers, VECTOR_BODY, &[vector_key()], VECTOR_NOW), Err(FalWebhookSignatureError::SignatureMismatch), "{headers:?}");+ }+ }++ #[test]+ fn rejects_a_signature_from_another_key() {+ assert_eq!(+ verify(&vector_headers(), VECTOR_BODY, &[random_key_pair().1], VECTOR_NOW),+ Err(FalWebhookSignatureError::SignatureMismatch),+ );+ }++ #[test]+ fn accepts_uppercase_hex() {+ let headers = FalWebhookSignatureHeaders { signature: VECTOR_SIGNATURE.to_uppercase(), ..vector_headers() };++ assert_eq!(verify(&headers, VECTOR_BODY, &[vector_key()], VECTOR_NOW), Ok(()));+ }+ }++ mod timestamps {+ use super::*;++ #[test]+ fn accepts_the_edges_of_the_tolerance() {+ for now in [VECTOR_NOW - TIMESTAMP_TOLERANCE_SECONDS, VECTOR_NOW + TIMESTAMP_TOLERANCE_SECONDS] {+ assert_eq!(verify(&vector_headers(), VECTOR_BODY, &[vector_key()], now), Ok(()), "now = {now}");+ }+ }++ #[test]+ fn rejects_stale_and_future_requests() {+ for now in [VECTOR_NOW - TIMESTAMP_TOLERANCE_SECONDS - 1, VECTOR_NOW + TIMESTAMP_TOLERANCE_SECONDS + 1] {+ assert_eq!(+ verify(&vector_headers(), VECTOR_BODY, &[vector_key()], now),+ Err(FalWebhookSignatureError::TimestampOutOfTolerance { timestamp: 1760500000, now }),+ );+ }+ }++ #[test]+ fn rejects_a_timestamp_that_is_not_a_number() {+ for timestamp in ["", "soon", "1760500000.5", "99999999999999999999999"] {+ let headers = FalWebhookSignatureHeaders { timestamp: timestamp.to_string(), ..vector_headers() };++ assert!(matches!(verify(&headers, VECTOR_BODY, &[vector_key()], VECTOR_NOW), Err(FalWebhookSignatureError::InvalidTimestamp(_))), "{timestamp:?}");+ }+ }++ #[test]+ fn extreme_timestamps_do_not_overflow() {+ let headers = FalWebhookSignatureHeaders { timestamp: i64::MIN.to_string(), ..vector_headers() };++ assert!(matches!(+ verify(&headers, VECTOR_BODY, &[vector_key()], i64::MAX),+ Err(FalWebhookSignatureError::TimestampOutOfTolerance { .. }),+ ));+ }+ }++ mod malformed_input {+ use super::*;++ #[test]+ fn rejects_signatures_that_are_not_64_bytes_of_hex() {+ for signature in ["", "zz", "abcd", &VECTOR_SIGNATURE[..126], &format!("{VECTOR_SIGNATURE}00")] {+ let headers = FalWebhookSignatureHeaders { signature: signature.to_string(), ..vector_headers() };++ assert_eq!(verify(&headers, VECTOR_BODY, &[vector_key()], VECTOR_NOW), Err(FalWebhookSignatureError::InvalidSignatureEncoding), "{signature:?}");+ }+ }++ #[test]+ fn rejects_when_there_are_no_keys() {+ assert_eq!(verify(&vector_headers(), VECTOR_BODY, &[], VECTOR_NOW), Err(FalWebhookSignatureError::NoPublicKeys));+ }+ }++ mod round_trip {+ use super::*;++ #[test]+ fn a_freshly_signed_webhook_verifies() {+ let (key_pair, public_key) = random_key_pair();+ let body = br#"{"status":"ERROR","request_id":"abc","error":"boom"}"#;+ let mut headers = FalWebhookSignatureHeaders {+ request_id: "abc".to_string(),+ user_id: "user".to_string(),+ timestamp: VECTOR_NOW.to_string(),+ signature: String::new(),+ };+ headers.signature = hex::encode(key_pair.sign(signed_message(&headers, body).as_bytes()).as_ref());++ assert_eq!(verify(&headers, body, &[public_key], VECTOR_NOW), Ok(()));+ }+ }++ fn verify(+ headers: &FalWebhookSignatureHeaders,+ body: &[u8],+ keys: &[FalWebhookPublicKey],+ now: i64,+ ) -> Result<(), FalWebhookSignatureError> {+ verify_fal_webhook_signature(headers, body, keys, now)+ }++ fn vector_headers() -> FalWebhookSignatureHeaders {+ FalWebhookSignatureHeaders {+ request_id: VECTOR_REQUEST_ID.to_string(),+ user_id: VECTOR_USER_ID.to_string(),+ timestamp: VECTOR_TIMESTAMP.to_string(),+ signature: VECTOR_SIGNATURE.to_string(),+ }+ }++ fn vector_key() -> FalWebhookPublicKey {+ FalWebhookPublicKey::from_bytes(hex::decode(VECTOR_PUBLIC_KEY).unwrap().try_into().unwrap())+ }++ fn random_key_pair() -> (Ed25519KeyPair, FalWebhookPublicKey) {+ let pkcs8 = Ed25519KeyPair::generate_pkcs8(&SystemRandom::new()).unwrap();+ let key_pair = Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap();+ let public_key = FalWebhookPublicKey::from_bytes(key_pair.public_key().as_ref().try_into().unwrap());+ (key_pair, public_key)+ }+}@@ -218,7 +218,9 @@ rustls = { version = "0.23", features = ["aws-lc-rs"] } test_utils = { path = "../../../testing/test_utils" } # Testing/Dev Deps: External+hex.workspace = true reqwest.workspace = true+ring.workspace = true speculoos.workspace = true strum.workspace = true tempfile.workspace = true@@ -56,6 +56,9 @@ use crate::http_server::user_lookup::user_session::session_utils::session_checke use crate::http_server::web_utils::scoped_temp_dir_creator::ScopedTempDirCreator; use crate::http_server::web_utils::web_opaque_cursor_encoder_v2::WebOpaqueCursorEncoderV2; use crate::http_server::web_utils::web_sort_key_crypto::WebSortKeyCrypto;+use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_mode::FalWebhookSignatureMode;+use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_settings::FalWebhookSignatureSettings;+use fal_client::webhook_signature::fal_jwks_cache::FalJwksCache; use crate::startup::setup_inference_providers::{ BeebleData, FalData, GmiCloudData, GrokApiData, InferenceProviders, KinoviWebData, OpenAiData, WorldLabsData,@@ -678,6 +681,10 @@ pub(crate) fn build_test_server_state(pool: MySqlPool, media_cdn_override_url: S fal: FalData { api_key: fal_client::creds::fal_api_key::FalApiKey::new("test".to_string()), webhook_url: "http://localhost/webhook".to_string(),+ webhook_signature: FalWebhookSignatureSettings {+ mode: FalWebhookSignatureMode::Off,+ jwks_cache: Arc::new(FalJwksCache::for_fal()),+ }, }, gmicloud: GmiCloudData { api_key: gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey::new("test".to_string()),@@ -3,6 +3,7 @@ use std::sync::Arc; use crate::http_server::common_responses::common_web_error::CommonWebError; use crate::http_server::endpoints::webhooks::fal::process_failure::handle_failed_fal_webhook::handle_failed_fal_webhook; use crate::http_server::endpoints::webhooks::fal::process_success::handle_successful_fal_webhook::handle_successful_fal_webhook;+use crate::http_server::endpoints::webhooks::fal::signature::check_fal_webhook_signature::check_fal_webhook_signature; use crate::state::server_state::ServerState; use actix_web::web::Bytes; use actix_web::web::Json;@@ -17,13 +18,17 @@ use pager::notification::notification_urgency::NotificationUrgency; // =============== Handler =============== -// TODO(bt, 2025-06-03): Handle webhook crypto authentication pub async fn fal_webhook_handler( http_request: HttpRequest, request_body_bytes: Bytes, server_state: web::Data<Arc<ServerState>>, ) -> Result<Json<SimpleGenericJsonSuccess>, CommonWebError> { + // Step 0: Check that FAL sent it (signed with one of FAL's published keys). What happens to+ // a request that doesn't verify depends on FAL_WEBHOOK_SIGNATURE_MODE; see+ // `check_fal_webhook_signature`.+ check_fal_webhook_signature(&http_request, &request_body_bytes, &server_state.inference_providers.fal.webhook_signature).await?;+ // Step 1: Parse bytes into a UTF-8 string and log it. let raw_body = String::from_utf8(request_body_bytes.to_vec()) .map_err(|err| {new file mode 100644@@ -0,0 +1,156 @@+//! End-to-end tests of the FAL webhook signature check: real HTTP requests through the real+//! route and handler, with a test MySQL database behind them.+//!+//! These are DATABASE TESTS (see `mysql_testing`). Skip them on machines without MySQL with+//! `--features skip_database_tests`.++use std::sync::Arc;+use std::time::{SystemTime, UNIX_EPOCH};++use actix_web::http::StatusCode;+use actix_web::{test, web, App};+use base64::engine::general_purpose::URL_SAFE;+use base64::Engine;+use fal_client::webhook_signature::fal_jwks_cache::FalJwksCache;+use fal_client::webhook_signature::fal_webhook_signature_headers::{REQUEST_ID_HEADER, SIGNATURE_HEADER, TIMESTAMP_HEADER, USER_ID_HEADER};+use ring::rand::SystemRandom;+use ring::signature::{Ed25519KeyPair, KeyPair};+use sha2::{Digest, Sha256};++use crate::http_server::endpoints::omni_gen::generate::video::tests::support::build_test_server_state;+use crate::http_server::endpoints::webhooks::fal::fal_webhook_handler::fal_webhook_handler;+use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_mode::FalWebhookSignatureMode;+use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_settings::FalWebhookSignatureSettings;++const WEBHOOK_PATH: &str = "/v1/webhooks/fal";++/// No job has this FAL request ID, so a webhook that gets past the signature check is answered+/// 404 by the job lookup. That shows the request reached the database, without needing a job.+const UNKNOWN_REQUEST_ID: &str = "00000000-0000-0000-0000-signature-test";++mod enforce {+ use super::*;++ #[actix_web::test]+ #[cfg_attr(feature = "skip_database_tests", ignore)]+ async fn an_unsigned_webhook_is_rejected_before_the_job_lookup() {+ let fal = FakeFal::new();++ let status = post(&fal, FalWebhookSignatureMode::Enforce, unsigned_webhook()).await;++ assert_eq!(status, StatusCode::UNAUTHORIZED);+ }++ #[actix_web::test]+ #[cfg_attr(feature = "skip_database_tests", ignore)]+ async fn a_webhook_signed_by_another_key_is_rejected() {+ let fal = FakeFal::new();+ let forger = FakeFal::new();++ let status = post(&fal, FalWebhookSignatureMode::Enforce, forger.signed_webhook(&webhook_body())).await;++ assert_eq!(status, StatusCode::UNAUTHORIZED);+ }++ #[actix_web::test]+ #[cfg_attr(feature = "skip_database_tests", ignore)]+ async fn a_genuine_webhook_goes_on_to_the_job_lookup() {+ let fal = FakeFal::new();++ let status = post(&fal, FalWebhookSignatureMode::Enforce, fal.signed_webhook(&webhook_body())).await;++ assert_eq!(status, StatusCode::NOT_FOUND);+ }+}++mod report {+ use super::*;++ #[actix_web::test]+ #[cfg_attr(feature = "skip_database_tests", ignore)]+ async fn an_unsigned_webhook_is_still_processed() {+ let fal = FakeFal::new();++ let status = post(&fal, FalWebhookSignatureMode::Report, unsigned_webhook()).await;++ assert_eq!(status, StatusCode::NOT_FOUND);+ }+}++/// A webhook request: its headers and raw body.+struct Webhook {+ headers: Vec<(&'static str, String)>,+ body: Vec<u8>,+}++/// Plays FAL: an ED25519 key pair, published as a JWKS, that signs webhooks the way FAL does.+struct FakeFal {+ key_pair: Ed25519KeyPair,+}++impl FakeFal {+ fn new() -> Self {+ let pkcs8 = Ed25519KeyPair::generate_pkcs8(&SystemRandom::new()).expect("generate key");+ Self { key_pair: Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).expect("parse key") }+ }++ fn jwks(&self) -> String {+ let x = URL_SAFE.encode(self.key_pair.public_key().as_ref());+ format!(r#"{{"keys":[{{"kty":"OKP","crv":"Ed25519","x":"{x}","use":"sig"}}]}}"#)+ }++ fn signed_webhook(&self, body: &[u8]) -> Webhook {+ let user_id = "fal-test-user";+ let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs().to_string();+ let message = format!("{UNKNOWN_REQUEST_ID}\n{user_id}\n{timestamp}\n{}", hex::encode(Sha256::digest(body)));+ let signature = hex::encode(self.key_pair.sign(message.as_bytes()).as_ref());++ Webhook {+ headers: vec![+ (REQUEST_ID_HEADER, UNKNOWN_REQUEST_ID.to_string()),+ (USER_ID_HEADER, user_id.to_string()),+ (TIMESTAMP_HEADER, timestamp),+ (SIGNATURE_HEADER, signature),+ ],+ body: body.to_vec(),+ }+ }+}++fn webhook_body() -> Vec<u8> {+ format!(r#"{{"request_id":"{UNKNOWN_REQUEST_ID}","gateway_request_id":"{UNKNOWN_REQUEST_ID}","status":"OK","payload":{{"images":[]}}}}"#).into_bytes()+}++fn unsigned_webhook() -> Webhook {+ Webhook { headers: Vec::new(), body: webhook_body() }+}++/// POST `webhook` to the FAL webhook handler of a server that trusts `fal`'s key, in `mode`.+async fn post(fal: &FakeFal, mode: FalWebhookSignatureMode, webhook: Webhook) -> StatusCode {+ let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();++ let pool = mysql_testing::pool::create_test_pool().await;+ let mut server_state = build_test_server_state(pool, "http://localhost".to_string());++ let jwks = fal.jwks();+ server_state.inference_providers.fal.webhook_signature = FalWebhookSignatureSettings {+ mode,+ jwks_cache: Arc::new(FalJwksCache::with_fetcher(Arc::new(move || {+ let jwks = jwks.clone();+ Box::pin(async move { Ok(jwks) })+ }))),+ };++ let app = test::init_service(+ App::new()+ .app_data(web::Data::new(Arc::new(server_state)))+ .route(WEBHOOK_PATH, web::post().to(fal_webhook_handler))+ ).await;++ let mut request = test::TestRequest::post().uri(WEBHOOK_PATH);+ for (name, value) in webhook.headers {+ request = request.insert_header((name, value));+ }++ test::call_service(&app, request.set_payload(webhook.body).to_request()).await.status()+}@@ -1,3 +1,6 @@ pub mod fal_webhook_handler;+#[cfg(test)]+mod fal_webhook_handler_tests; pub(super) mod process_failure; pub(super) mod process_success;+pub mod signature;new file mode 100644@@ -0,0 +1,220 @@+use std::time::{SystemTime, UNIX_EPOCH};++use actix_web::HttpRequest;+use errors::anyhow;+use fal_client::webhook_signature::fal_webhook_signature_error::FalWebhookSignatureError;+use fal_client::webhook_signature::fal_webhook_signature_headers::FalWebhookSignatureHeaders;+use fal_client::webhook_signature::verify_fal_webhook_request::verify_fal_webhook_request;+use log::{error, warn};++use crate::http_server::common_responses::common_web_error::CommonWebError;+use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_mode::{FalWebhookSignatureMode, FAL_WEBHOOK_SIGNATURE_MODE_ENV};+use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_settings::FalWebhookSignatureSettings;++/// Check that a webhook request was sent by FAL, as the configured mode asks.+///+/// - `Enforce`: a request that doesn't verify is rejected here, before its body is parsed or+/// any job is looked up. A bad or missing signature is a 401; FAL's keys being unavailable is+/// a 500, so that FAL retries the delivery rather than having it refused.+/// - `Report`: a failure is logged and the request goes on to be processed.+/// - `Off`: nothing is checked.+pub async fn check_fal_webhook_signature(+ http_request: &HttpRequest,+ raw_body: &[u8],+ settings: &FalWebhookSignatureSettings,+) -> Result<(), CommonWebError> {+ check_fal_webhook_signature_at(http_request, raw_body, settings, unix_now()).await+}++async fn check_fal_webhook_signature_at(+ http_request: &HttpRequest,+ raw_body: &[u8],+ settings: &FalWebhookSignatureSettings,+ now_unix_seconds: i64,+) -> Result<(), CommonWebError> {+ if settings.mode == FalWebhookSignatureMode::Off {+ return Ok(());+ }++ let verification = match read_headers(http_request) {+ Ok(headers) => verify_fal_webhook_request(&settings.jwks_cache, &headers, raw_body, now_unix_seconds).await,+ Err(err) => Err(err),+ };++ let Err(err) = verification else {+ return Ok(());+ };++ match settings.mode {+ FalWebhookSignatureMode::Enforce if is_our_problem(&err) => {+ error!("Could not verify a FAL webhook because FAL's keys are unavailable: {}", err);+ Err(CommonWebError::from_anyhow_error(anyhow!("FAL webhook keys unavailable: {}", err)))+ }+ FalWebhookSignatureMode::Enforce => {+ warn!("Rejected a FAL webhook that failed its signature check: {}", err);+ Err(CommonWebError::NotAuthorized)+ }+ FalWebhookSignatureMode::Report => {+ warn!(+ "A FAL webhook failed its signature check ({}); processing it anyway because {}=report",+ err, FAL_WEBHOOK_SIGNATURE_MODE_ENV,+ );+ Ok(())+ }+ FalWebhookSignatureMode::Off => Ok(()),+ }+}++fn read_headers(http_request: &HttpRequest) -> Result<FalWebhookSignatureHeaders, FalWebhookSignatureError> {+ FalWebhookSignatureHeaders::from_lookup(|name| {+ http_request.headers().get(name).and_then(|value| value.to_str().ok())+ })+}++/// Failures that are about our copy of FAL's keys, not about the request.+fn is_our_problem(err: &FalWebhookSignatureError) -> bool {+ matches!(+ err,+ FalWebhookSignatureError::JwksFetchFailed(_)+ | FalWebhookSignatureError::JwksParseFailed(_)+ | FalWebhookSignatureError::NoPublicKeys+ )+}++fn unix_now() -> i64 {+ SystemTime::now()+ .duration_since(UNIX_EPOCH)+ .map(|elapsed| elapsed.as_secs() as i64)+ .unwrap_or(0)+}++#[cfg(test)]+mod tests {+ use std::sync::Arc;++ use actix_web::http::StatusCode;+ use actix_web::test::TestRequest;+ use actix_web::ResponseError;+ use fal_client::webhook_signature::fal_jwks_cache::FalJwksCache;+ use fal_client::webhook_signature::fal_webhook_signature_error::FalWebhookSignatureError;+ use fal_client::webhook_signature::fal_webhook_signature_headers::{REQUEST_ID_HEADER, SIGNATURE_HEADER, TIMESTAMP_HEADER, USER_ID_HEADER};++ use super::*;++ // A webhook signed with PyNaCl by following FAL's documented algorithm (the same vector as+ // `fal_client::webhook_signature::verify_fal_webhook_signature`'s tests).+ const PUBLIC_KEY_JWKS: &str = r#"{"keys":[{"kty":"OKP","crv":"Ed25519","x":"A6EHv_POEL4dcN0Y50vAmWfk1jCbpQ1fHdyGZBJVMbg=","use":"sig"}]}"#;+ const REQUEST_ID: &str = "019e18d8-8c36-7bc1-aa77-2bc2f70268c6";+ const USER_ID: &str = "test-user-1";+ const TIMESTAMP: &str = "1760500000";+ const BODY: &[u8] = br#"{"request_id":"019e18d8-8c36-7bc1-aa77-2bc2f70268c6","status":"OK","payload":{"images":[]}}"#;+ const SIGNATURE: &str = "d925f70845ad2e5a410c3b671a443a27cf538bf1f20b9918b6ce583f0f3640c6f9b2a407a442fa3d770e914ea27f4d4bfc07fd64f3496672e0da6af9a930040e";+ const NOW: i64 = 1760500000;++ mod enforce {+ use super::*;++ #[tokio::test]+ async fn accepts_a_genuine_webhook() {+ let result = check(FalWebhookSignatureMode::Enforce, &signed_request(), BODY).await;++ assert!(result.is_ok());+ }++ #[tokio::test]+ async fn rejects_an_unsigned_webhook_with_401() {+ let result = check(FalWebhookSignatureMode::Enforce, &TestRequest::post().to_http_request(), BODY).await;++ assert_eq!(status_of(result), StatusCode::UNAUTHORIZED);+ }++ #[tokio::test]+ async fn rejects_a_tampered_body_with_401() {+ let tampered = br#"{"request_id":"019e18d8-8c36-7bc1-aa77-2bc2f70268c6","status":"OK","payload":{"images":["https://example.com/x.png"]}}"#;++ let result = check(FalWebhookSignatureMode::Enforce, &signed_request(), tampered).await;++ assert_eq!(status_of(result), StatusCode::UNAUTHORIZED);+ }++ #[tokio::test]+ async fn rejects_a_replayed_webhook_with_401() {+ let settings = settings(FalWebhookSignatureMode::Enforce, Some(PUBLIC_KEY_JWKS));++ let result = check_fal_webhook_signature_at(&signed_request(), BODY, &settings, NOW + 3600).await;++ assert_eq!(status_of(result), StatusCode::UNAUTHORIZED);+ }++ #[tokio::test]+ async fn answers_500_when_fals_keys_are_unavailable() {+ let settings = settings(FalWebhookSignatureMode::Enforce, None);++ let result = check_fal_webhook_signature_at(&signed_request(), BODY, &settings, NOW).await;++ assert_eq!(status_of(result), StatusCode::INTERNAL_SERVER_ERROR);+ }+ }++ mod report_and_off {+ use super::*;++ #[tokio::test]+ async fn report_lets_a_failing_webhook_through() {+ let result = check(FalWebhookSignatureMode::Report, &TestRequest::post().to_http_request(), BODY).await;++ assert!(result.is_ok());+ }++ #[tokio::test]+ async fn report_lets_a_webhook_through_when_the_keys_are_unavailable() {+ let settings = settings(FalWebhookSignatureMode::Report, None);++ let result = check_fal_webhook_signature_at(&signed_request(), BODY, &settings, NOW).await;++ assert!(result.is_ok());+ }++ #[tokio::test]+ async fn off_checks_nothing_and_fetches_no_keys() {+ // The key source fails, so any attempt to verify would surface as an error in Enforce+ // mode; Off must not even try.+ let settings = settings(FalWebhookSignatureMode::Off, None);++ let result = check_fal_webhook_signature_at(&TestRequest::post().to_http_request(), BODY, &settings, NOW).await;++ assert!(result.is_ok());+ }+ }++ async fn check(mode: FalWebhookSignatureMode, request: &HttpRequest, body: &[u8]) -> Result<(), CommonWebError> {+ check_fal_webhook_signature_at(request, body, &settings(mode, Some(PUBLIC_KEY_JWKS)), NOW).await+ }++ /// Settings whose key source serves `maybe_jwks`, or fails when it is `None`.+ fn settings(mode: FalWebhookSignatureMode, maybe_jwks: Option<&'static str>) -> FalWebhookSignatureSettings {+ FalWebhookSignatureSettings {+ mode,+ jwks_cache: Arc::new(FalJwksCache::with_fetcher(Arc::new(move || {+ Box::pin(async move {+ maybe_jwks+ .map(str::to_string)+ .ok_or_else(|| FalWebhookSignatureError::JwksFetchFailed("test source is down".to_string()))+ })+ }))),+ }+ }++ fn signed_request() -> HttpRequest {+ TestRequest::post()+ .insert_header((REQUEST_ID_HEADER, REQUEST_ID))+ .insert_header((USER_ID_HEADER, USER_ID))+ .insert_header((TIMESTAMP_HEADER, TIMESTAMP))+ .insert_header((SIGNATURE_HEADER, SIGNATURE))+ .to_http_request()+ }++ fn status_of(result: Result<(), CommonWebError>) -> StatusCode {+ result.expect_err("expected the webhook to be rejected").status_code()+ }+}new file mode 100644@@ -0,0 +1,65 @@+use errors::{anyhow, AnyhowResult};++/// Environment variable that sets [`FalWebhookSignatureMode`]: `enforce`, `report` or `off`.+pub const FAL_WEBHOOK_SIGNATURE_MODE_ENV: &str = "FAL_WEBHOOK_SIGNATURE_MODE";++/// What the FAL webhook endpoint does with a request whose signature doesn't verify.+#[derive(Clone, Copy, Debug, PartialEq, Eq)]+pub enum FalWebhookSignatureMode {+ /// Reject it before reading the body: 401, or 500 if FAL's keys can't be fetched (so FAL+ /// retries the delivery).+ Enforce,++ /// Log a warning and process it anyway. This is the default, so that failures can be seen+ /// in the logs before switching to `Enforce`.+ Report,++ /// Don't check signatures.+ Off,+}++impl FalWebhookSignatureMode {+ /// Read [`FAL_WEBHOOK_SIGNATURE_MODE_ENV`], defaulting to `Report`. An unrecognised value is an+ /// error, so a typo can't silently turn checking off.+ pub fn from_env() -> AnyhowResult<Self> {+ match easyenv::get_env_string_optional(FAL_WEBHOOK_SIGNATURE_MODE_ENV) {+ None => Ok(Self::Report),+ Some(value) => Self::parse(&value).ok_or_else(|| {+ anyhow!("{} must be one of enforce, report or off (got {:?})", FAL_WEBHOOK_SIGNATURE_MODE_ENV, value)+ }),+ }+ }++ pub fn parse(value: &str) -> Option<Self> {+ match value.trim().to_ascii_lowercase().as_str() {+ "enforce" => Some(Self::Enforce),+ "report" => Some(Self::Report),+ "off" => Some(Self::Off),+ _ => None,+ }+ }+}++#[cfg(test)]+mod tests {+ use super::*;++ #[test]+ fn parses_the_three_modes() {+ assert_eq!(FalWebhookSignatureMode::parse("enforce"), Some(FalWebhookSignatureMode::Enforce));+ assert_eq!(FalWebhookSignatureMode::parse("report"), Some(FalWebhookSignatureMode::Report));+ assert_eq!(FalWebhookSignatureMode::parse("off"), Some(FalWebhookSignatureMode::Off));+ }++ #[test]+ fn ignores_case_and_surrounding_whitespace() {+ assert_eq!(FalWebhookSignatureMode::parse(" Enforce \n"), Some(FalWebhookSignatureMode::Enforce));+ }++ #[test]+ fn rejects_anything_else() {+ for value in ["", "true", "on", "enforced", "1"] {+ assert_eq!(FalWebhookSignatureMode::parse(value), None, "{value:?}");+ }+ }+}new file mode 100644@@ -0,0 +1,24 @@+use std::sync::Arc;++use fal_client::webhook_signature::fal_jwks_cache::FalJwksCache;++use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_mode::FalWebhookSignatureMode;++/// How the FAL webhook endpoint verifies that requests come from FAL.+#[derive(Clone)]+pub struct FalWebhookSignatureSettings {+ pub mode: FalWebhookSignatureMode,++ /// FAL's public keys, shared by all requests.+ pub jwks_cache: Arc<FalJwksCache>,+}++impl FalWebhookSignatureSettings {+ /// Settings for production: the mode from the environment, keys fetched from FAL.+ pub fn from_env() -> errors::AnyhowResult<Self> {+ Ok(Self {+ mode: FalWebhookSignatureMode::from_env()?,+ jwks_cache: Arc::new(FalJwksCache::for_fal()),+ })+ }+}new file mode 100644@@ -0,0 +1,3 @@+pub mod check_fal_webhook_signature;+pub mod fal_webhook_signature_mode;+pub mod fal_webhook_signature_settings;@@ -1,6 +1,9 @@ use beeble_client::creds::beeble_api_key::BeebleApiKey; use errors::AnyhowResult; use fal_client::creds::fal_api_key::FalApiKey;+use log::info;++use crate::http_server::endpoints::webhooks::fal::signature::fal_webhook_signature_settings::FalWebhookSignatureSettings; /// All third-party inference provider credentials and configuration grouped together. pub struct InferenceProviders {@@ -18,6 +21,8 @@ pub struct InferenceProviders { pub struct FalData { pub api_key: FalApiKey, pub webhook_url: String,+ /// How webhooks at `webhook_url` are checked for FAL's signature.+ pub webhook_signature: FalWebhookSignatureSettings, } /// GmiCloud integration@@ -62,10 +67,14 @@ pub struct WorldLabsData { } pub fn setup_inference_providers() -> AnyhowResult<InferenceProviders> {+ let fal_webhook_signature = FalWebhookSignatureSettings::from_env()?;+ info!("FAL webhook signature mode: {:?}", fal_webhook_signature.mode);+ Ok(InferenceProviders { fal: FalData { api_key: FalApiKey::new(easyenv::get_env_string_required("FAL_API_KEY")?), webhook_url: easyenv::get_env_string_required("FAL_WEBHOOK_URL")?,+ webhook_signature: fal_webhook_signature, }, gmicloud: GmiCloudData { api_key: gmicloud_client::creds::gmicloud_api_key::GmiCloudApiKey::new(