pagesift_api/
webhook.rs

1//! Firecrawl v2 的任务 webhook：`started`、`page`、`completed`、`failed` 四种事件。
2//!
3//! - 每个任务一个投递任务，事件按产生顺序逐个 POST，失败按指数退避重试，次数有上限；
4//! - 投递不阻塞爬取：页面事件进有界队列，队列满了就丢弃并计数；开始和结束事件一定排上队；
5//! - 配置了密钥时带 `X-Firecrawl-Signature: sha256=<HMAC-SHA256(密钥, 请求体)>`，和 Firecrawl 一样；
6//! - 目标地址过 SSRF 校验：请求时校验 URL，投递时的 DNS 解析结果逐个校验，连接的正是校验过的地址，
7//!   不跟随重定向，不走环境变量里的代理。
8
9use std::collections::BTreeMap;
10use std::net::SocketAddr;
11use std::sync::Arc;
12use std::time::Duration;
13
14use pagesift_net::{DnsLookup, SsrfPolicy, SystemLookup};
15use serde::{Deserialize, Serialize};
16use serde_json::{Map, Value, json};
17use tokio::sync::mpsc;
18
19use crate::error::{ApiError, ErrorCode};
20use crate::metrics::Metrics;
21use crate::types::Document;
22
23/// 请求里的 `webhook`：Firecrawl 接受一个 URL 字符串，或者带 headers、metadata、events 的对象。
24#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
25pub struct WebhookConfig {
26    pub url: String,
27    #[serde(default)]
28    pub headers: BTreeMap<String, String>,
29    #[serde(default)]
30    pub metadata: Map<String, Value>,
31    #[serde(default = "all_events")]
32    pub events: Vec<String>,
33}
34
35const EVENTS: [&str; 4] = ["started", "page", "completed", "failed"];
36
37fn all_events() -> Vec<String> {
38    EVENTS.iter().map(|e| e.to_string()).collect()
39}
40
41impl WebhookConfig {
42    /// 解析并校验请求里的 `webhook` 字段（含 SSRF 校验）。未知字段忽略。
43    pub(crate) fn parse(v: &Value, policy: &SsrfPolicy) -> Result<Self, ApiError> {
44        let cfg: Self = match v {
45            Value::String(u) => Self {
46                url: u.clone(),
47                headers: BTreeMap::new(),
48                metadata: Map::new(),
49                events: all_events(),
50            },
51            Value::Object(_) => serde_json::from_value(v.clone())
52                .map_err(|e| ApiError::invalid(format!("invalid webhook: {e}")))?,
53            _ => {
54                return Err(ApiError::invalid(
55                    "webhook must be a URL or an object with a url",
56                ));
57            }
58        };
59        if let Some(e) = cfg.events.iter().find(|e| !EVENTS.contains(&e.as_str())) {
60            return Err(ApiError::invalid(format!(
61                "unknown webhook event {e:?}: use started, page, completed or failed"
62            )));
63        }
64        for (k, v) in &cfg.headers {
65            if reqwest::header::HeaderName::from_bytes(k.as_bytes()).is_err()
66                || reqwest::header::HeaderValue::from_str(v).is_err()
67            {
68                return Err(ApiError::invalid(format!("invalid webhook header {k:?}")));
69            }
70        }
71        policy.check_url(&cfg.url).map_err(|e| {
72            ApiError::new(
73                ErrorCode::SsrfBlocked,
74                format!("webhook URL is not allowed: {e}"),
75            )
76        })?;
77        Ok(cfg)
78    }
79
80    fn wants(&self, event: Event) -> bool {
81        self.events.iter().any(|e| e == event.name())
82    }
83}
84
85#[derive(Debug, Clone, Copy, PartialEq, Eq)]
86pub(crate) enum Event {
87    Started,
88    Page,
89    Completed,
90    Failed,
91}
92
93impl Event {
94    fn name(self) -> &'static str {
95        match self {
96            Event::Started => "started",
97            Event::Page => "page",
98            Event::Completed => "completed",
99            Event::Failed => "failed",
100        }
101    }
102}
103
104/// webhook 投递的设置（服务端配置，不来自请求）。
105#[derive(Clone)]
106pub struct WebhookSettings {
107    /// 目标地址的 SSRF 策略，默认拒绝一切非公网地址。
108    pub ssrf: SsrfPolicy,
109    pub dns: Arc<dyn DnsLookup>,
110    /// 每个事件最多投递几次（含第一次）。
111    pub max_attempts: u32,
112    /// 第一次重试前等多久，之后每次翻倍，不超过 `max_backoff`。
113    pub initial_backoff: Duration,
114    pub max_backoff: Duration,
115    /// 单次投递的超时。
116    pub timeout: Duration,
117    /// 每个任务排队等待投递的事件数上限，满了丢弃页面事件。
118    pub queue: usize,
119    /// 签名密钥：设了就给每个请求加 `X-Firecrawl-Signature`，接收方用它验证来源。
120    pub secret: Option<String>,
121}
122
123impl Default for WebhookSettings {
124    fn default() -> Self {
125        Self {
126            ssrf: SsrfPolicy::default(),
127            dns: Arc::new(SystemLookup),
128            max_attempts: 5,
129            initial_backoff: Duration::from_secs(1),
130            max_backoff: Duration::from_secs(60),
131            timeout: Duration::from_secs(10),
132            queue: 1024,
133            secret: None,
134        }
135    }
136}
137
138impl std::fmt::Debug for WebhookSettings {
139    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
140        f.debug_struct("WebhookSettings")
141            .field("ssrf", &self.ssrf)
142            .field("max_attempts", &self.max_attempts)
143            .field("initial_backoff", &self.initial_backoff)
144            .field("max_backoff", &self.max_backoff)
145            .field("timeout", &self.timeout)
146            .field("queue", &self.queue)
147            .field("secret", &self.secret.is_some())
148            .finish_non_exhaustive()
149    }
150}
151
152/// 所有任务共用的投递器（HTTP 客户端和设置）。
153#[derive(Clone)]
154pub(crate) struct Webhooks {
155    client: reqwest::Client,
156    settings: Arc<WebhookSettings>,
157    metrics: Arc<Metrics>,
158}
159
160/// 解析失败或地址被 SSRF 策略拒绝：投递时从错误链里认出来，不再重试。
161#[derive(Debug)]
162struct Blocked(String);
163
164impl std::fmt::Display for Blocked {
165    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
166        write!(f, "webhook address blocked: {}", self.0)
167    }
168}
169
170impl std::error::Error for Blocked {}
171
172/// 按 SSRF 策略校验 DNS 解析结果的解析器。
173struct GuardedDns {
174    dns: Arc<dyn DnsLookup>,
175    policy: SsrfPolicy,
176}
177
178impl reqwest::dns::Resolve for GuardedDns {
179    fn resolve(&self, name: reqwest::dns::Name) -> reqwest::dns::Resolving {
180        let (dns, policy) = (self.dns.clone(), self.policy.clone());
181        Box::pin(async move {
182            let host = name.as_str().to_string();
183            let ips = dns.lookup(&host).await?;
184            let blocked = |why: String| -> Box<dyn std::error::Error + Send + Sync> {
185                Box::new(Blocked(why))
186            };
187            if ips.is_empty() {
188                return Err(blocked(format!("no address for {host}")));
189            }
190            for ip in &ips {
191                policy.check_ip(*ip).map_err(|e| blocked(e.to_string()))?;
192            }
193            let addrs: reqwest::dns::Addrs =
194                Box::new(ips.into_iter().map(|ip| SocketAddr::new(ip, 0)));
195            Ok(addrs)
196        })
197    }
198}
199
200fn is_blocked(e: &reqwest::Error) -> bool {
201    let mut src: Option<&(dyn std::error::Error + 'static)> = Some(e);
202    while let Some(s) = src {
203        if s.is::<Blocked>() {
204            return true;
205        }
206        src = s.source();
207    }
208    false
209}
210
211impl Webhooks {
212    pub(crate) fn new(settings: WebhookSettings, metrics: Arc<Metrics>) -> Self {
213        let client = reqwest::Client::builder()
214            .no_proxy()
215            .redirect(reqwest::redirect::Policy::none())
216            .dns_resolver(Arc::new(GuardedDns {
217                dns: settings.dns.clone(),
218                policy: settings.ssrf.clone(),
219            }))
220            .user_agent(pagesift_net::DEFAULT_USER_AGENT)
221            .timeout(settings.timeout)
222            .build()
223            .expect("webhook client builds with static settings");
224        Self {
225            client,
226            settings: Arc::new(settings),
227            metrics,
228        }
229    }
230
231    pub(crate) fn ssrf(&self) -> &SsrfPolicy {
232        &self.settings.ssrf
233    }
234
235    /// 为一个任务启动投递任务。`kind` 是事件类型前缀：`crawl` 或 `batch_scrape`。
236    pub(crate) fn start(&self, job_id: &str, kind: &'static str, cfg: WebhookConfig) -> Hook {
237        let (tx, mut rx) = mpsc::channel::<Value>(self.settings.queue.max(1));
238        let this = self.clone();
239        let cfg = Arc::new(cfg);
240        let worker_cfg = cfg.clone();
241        tokio::spawn(async move {
242            while let Some(body) = rx.recv().await {
243                this.deliver(&worker_cfg, &body).await;
244            }
245        });
246        Hook {
247            tx,
248            cfg,
249            kind,
250            job_id: job_id.into(),
251            metrics: self.metrics.clone(),
252        }
253    }
254
255    /// 投递一个事件，失败时按指数退避重试。
256    async fn deliver(&self, cfg: &WebhookConfig, body: &Value) {
257        let s = &self.settings;
258        // IP 字面量不经过解析器，这里再按当前策略校验一次（任务可能是重启前按旧配置收下的）
259        if let Err(e) = s.ssrf.check_url(&cfg.url) {
260            self.metrics.webhook("blocked");
261            tracing::warn!(url = %cfg.url, error = %e, "webhook target blocked by the SSRF policy");
262            return;
263        }
264        let payload = body.to_string();
265        let signature = s.secret.as_deref().map(|k| sign(k, payload.as_bytes()));
266        let mut backoff = s.initial_backoff;
267        for attempt in 1..=s.max_attempts.max(1) {
268            let mut req = self
269                .client
270                .post(&cfg.url)
271                .header(reqwest::header::CONTENT_TYPE, "application/json")
272                .body(payload.clone());
273            if let Some(sig) = &signature {
274                req = req.header("x-firecrawl-signature", sig);
275            }
276            for (k, v) in &cfg.headers {
277                req = req.header(k, v);
278            }
279            let (retry, why) = match req.send().await {
280                Ok(r) if r.status().is_success() => {
281                    self.metrics.webhook("delivered");
282                    return;
283                }
284                Ok(r) => {
285                    let st = r.status();
286                    // 超时、限流和服务端错误可以重试；其他 4xx 和重定向重试也没用
287                    let retry = st.is_server_error()
288                        || st == reqwest::StatusCode::TOO_MANY_REQUESTS
289                        || st == reqwest::StatusCode::REQUEST_TIMEOUT;
290                    (retry, format!("status {st}"))
291                }
292                Err(e) if is_blocked(&e) => {
293                    self.metrics.webhook("blocked");
294                    tracing::warn!(url = %cfg.url, error = %e, "webhook target blocked by the SSRF policy");
295                    return;
296                }
297                Err(e) => (true, e.to_string()),
298            };
299            if !retry || attempt == s.max_attempts {
300                self.metrics.webhook("failed");
301                tracing::warn!(url = %cfg.url, attempt, reason = %why, "webhook delivery failed");
302                return;
303            }
304            self.metrics.webhook("retried");
305            tracing::debug!(url = %cfg.url, attempt, reason = %why, ?backoff, "retrying webhook");
306            tokio::time::sleep(backoff).await;
307            backoff = (backoff * 2).min(s.max_backoff);
308        }
309    }
310}
311
312/// `sha256=` 加上请求体的 HMAC-SHA256（小写十六进制）。
313fn sign(secret: &str, payload: &[u8]) -> String {
314    let key = ring::hmac::Key::new(ring::hmac::HMAC_SHA256, secret.as_bytes());
315    let tag = ring::hmac::sign(&key, payload);
316    let mut out = String::from("sha256=");
317    for b in tag.as_ref() {
318        out.push_str(&format!("{b:02x}"));
319    }
320    out
321}
322
323/// 一个任务的 webhook 句柄。全部句柄释放后，投递任务发完排队的事件就退出。
324#[derive(Clone)]
325pub(crate) struct Hook {
326    tx: mpsc::Sender<Value>,
327    cfg: Arc<WebhookConfig>,
328    kind: &'static str,
329    job_id: Arc<str>,
330    metrics: Arc<Metrics>,
331}
332
333impl Hook {
334    pub(crate) fn send(&self, event: Event, data: Vec<Document>, error: Option<String>) {
335        if !self.cfg.wants(event) {
336            return;
337        }
338        let mut body = json!({
339            "success": error.is_none(),
340            "type": format!("{}.{}", self.kind, event.name()),
341            "id": &*self.job_id,
342            "data": data,
343            "metadata": self.cfg.metadata,
344        });
345        if let Some(e) = error {
346            body["error"] = e.into();
347        }
348        match self.tx.try_send(body) {
349            Ok(()) => {}
350            Err(mpsc::error::TrySendError::Full(body)) if event != Event::Page => {
351                // 开始和结束事件不能丢：另起一个任务等队列空出来，爬取照样不等
352                let tx = self.tx.clone();
353                tokio::spawn(async move {
354                    let _ = tx.send(body).await;
355                });
356            }
357            Err(_) => {
358                self.metrics.webhook("dropped");
359            }
360        }
361    }
362}
363
364#[cfg(test)]
365mod tests {
366    use super::*;
367
368    #[test]
369    fn signatures_match_rfc_4231() {
370        // RFC 4231 测试用例 2
371        assert_eq!(
372            sign("Jefe", b"what do ya want for nothing?"),
373            "sha256=5bdcc146bf60754e6a042426089575c75a003f089d2739839dec58b964ec3843"
374        );
375    }
376
377    #[test]
378    fn parses_strings_objects_and_rejects_private_targets() {
379        let policy = SsrfPolicy::default();
380        let c = WebhookConfig::parse(&json!("https://hooks.example.com/x"), &policy).unwrap();
381        assert_eq!(c.events, all_events());
382        let c = WebhookConfig::parse(
383            &json!({"url": "https://hooks.example.com/x", "headers": {"X-Token": "t"},
384                    "metadata": {"run": "1"}, "events": ["completed"], "future": true}),
385            &policy,
386        )
387        .unwrap();
388        assert!(c.wants(Event::Completed) && !c.wants(Event::Page));
389        for (v, code) in [
390            (json!("http://127.0.0.1:9000/hook"), ErrorCode::SsrfBlocked),
391            (
392                json!("http://169.254.169.254/latest"),
393                ErrorCode::SsrfBlocked,
394            ),
395            (json!("http://metadata.internal/"), ErrorCode::SsrfBlocked),
396            (json!("file:///etc/passwd"), ErrorCode::SsrfBlocked),
397            (
398                json!({"url": "https://a.example", "events": ["done"]}),
399                ErrorCode::InvalidRequest,
400            ),
401            (json!(42), ErrorCode::InvalidRequest),
402        ] {
403            assert_eq!(
404                WebhookConfig::parse(&v, &policy).unwrap_err().code,
405                code,
406                "{v}"
407            );
408        }
409    }
410}
pagesift_api/webhook.rs

pagesift_api/
webhook.rs
