pagesift_sink/audit/
mod.rs

1//! 审计日志：每次抓取写一行 JSON（只有元数据），结果可疑时另存当时的 HTML，供事后复现。
2//!
3//! 单进程长期运行的服务（`serve`、`mcp http`）用它当真实流量的测试场：抓了什么、走了哪个引擎、
4//! 结果怎样、哪里不对，都能从日志里按站点和调用方汇总，可疑页面能用快照在本地反复重放。
5//!
6//! - 日志按大小轮转：超过上限就把当前文件改名为 `<path>.1`（覆盖旧的）再新开，磁盘占用不超过两份上限。
7//! - URL 默认去掉用户名密码，查询参数里像凭证的值（token、key、signature……）换成 `REDACTED`。
8//! - 快照只在结果可疑时写，每天有条数上限，单个文件有大小上限。
9
10use std::io;
11use std::path::{Path, PathBuf};
12use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
13use std::time::{Duration, SystemTime, UNIX_EPOCH};
14
15use pagesift_api::types::{Document, ScrapeOptions};
16use pagesift_api::{ApiError, ErrorCode};
17use serde_json::{Value, json};
18use url::Url;
19
20mod writer;
21
22tokio::task_local! {
23    /// 发起这次抓取的客户端（MCP 的 clientInfo 名称等），只用于审计日志。
24    pub static CLIENT: String;
25    /// 多服务器工人：这个任务写出的审计记录另收集一份，随结果交给网关统一追加（见 `pagesift` 的 `cluster` 模块）。
26    pub static COLLECT: std::cell::RefCell<Vec<Value>>;
27}
28
29/// 审计日志配置。
30#[derive(Debug, Clone)]
31pub struct AuditConfig {
32    /// JSON Lines 日志文件。
33    pub path: PathBuf,
34    /// 当前日志文件超过这个大小就轮转。
35    pub max_bytes: u64,
36    /// 结果可疑时保存 HTML 快照的目录；`None` 表示不存快照。
37    pub snapshot_dir: Option<PathBuf>,
38    /// 单个快照最多保存多少字节，超出部分截掉。
39    pub snapshot_max_bytes: usize,
40    /// 每天最多保存多少个快照。
41    pub snapshots_per_day: u32,
42    /// URL 里的凭证是否脱敏。
43    pub redact: bool,
44    /// 运行方式（`serve`、`mcp-http`、`scrape`……），写进每条记录。
45    pub source: String,
46}
47
48impl AuditConfig {
49    pub fn new(path: impl Into<PathBuf>) -> Self {
50        let path = path.into();
51        let snapshot_dir = path.parent().map(|p| p.join("snapshots"));
52        Self {
53            path,
54            // [策略] 运维策略，都可在 `[audit]` 里改。默认值的依据：一条记录约 0.4 KiB（本机服务日志 64 条实测均值 411 B），64 MiB 约 16 万条，
55            // 够一台服务一天的量，轮转后最多占两份；单份快照 2 MiB 覆盖 99.8% 的 HTML 页（本仓库 AEB、WCXB、
56            // WebMainBench 共 1237 页实测：中位数 86 KiB、p99 745 KiB、最大 2.8 MiB），每天 200 份把快照
57            // 目录限制在每天 400 MiB 内。
58            max_bytes: 64 * 1024 * 1024,
59            snapshot_dir,
60            snapshot_max_bytes: 2 * 1024 * 1024,
61            snapshots_per_day: 200,
62            redact: true,
63            source: "pagesift".into(),
64        }
65    }
66}
67
68pub struct Audit {
69    cfg: AuditConfig,
70    /// 只收集、不落本地文件时（没配置审计日志的工人）为 `None`。
71    log: Option<writer::LogWriter>,
72    /// 快照计数所在的那一天（Unix 天数）和当天已存的个数。
73    snap_day: AtomicU64,
74    snap_count: AtomicU32,
75}
76
77/// 这一类问题值得留快照：页面抓到了，但正文可能丢了。
78pub const SUSPICIOUS_WARNINGS: &[&str] = &[
79    "empty_content",
80    "thin_content",
81    "partial_content",
82    "replies_omitted",
83    "pages_missing",
84    "content_missing",
85    "render_unavailable",
86    "render_skipped_",
87];
88/// 抓到 200 的 HTML 页面却少于这么多字，也算可疑。
89const SHORT_TEXT_CHARS: usize = 200;
90
91impl Audit {
92    pub fn open(cfg: AuditConfig) -> io::Result<Self> {
93        let log = Some(writer::LogWriter::open(&cfg.path, cfg.max_bytes)?);
94        Ok(Self {
95            cfg,
96            log,
97            snap_day: AtomicU64::new(0),
98            snap_count: AtomicU32::new(0),
99        })
100    }
101
102    /// 快照要用渲染或抓取得到的原始 HTML：开着快照时内部多要一份 rawHtml。
103    pub fn wants_raw_html(&self) -> bool {
104        self.cfg.snapshot_dir.is_some()
105    }
106
107    /// 记一次抓取。`raw_html` 是这次得到的 HTML（缓存命中、出错时没有）。
108    pub fn record_scrape(
109        &self,
110        kind: &str,
111        url: &str,
112        opts: &ScrapeOptions,
113        result: &Result<Document, ApiError>,
114        raw_html: Option<&str>,
115        elapsed: Duration,
116    ) {
117        let rec = self.scrape_record(kind, url, opts, result, raw_html, elapsed);
118        self.write(&rec);
119    }
120
121    /// 同 [`Audit::record_scrape`]，这次抓取实际忽略了 robots.txt：记录里另带 `robotsIgnored`（原因）。
122    #[allow(clippy::too_many_arguments)]
123    pub fn record_scrape_ignoring_robots(
124        &self,
125        kind: &str,
126        url: &str,
127        opts: &ScrapeOptions,
128        result: &Result<Document, ApiError>,
129        raw_html: Option<&str>,
130        elapsed: Duration,
131        reason: &str,
132    ) {
133        let mut rec = self.scrape_record(kind, url, opts, result, raw_html, elapsed);
134        rec["robotsIgnored"] = json!(reason);
135        self.write(&rec);
136    }
137
138    /// 一次整站爬取或 map 实际忽略了 robots.txt：起始 URL、请求类型、原因（调用方见 `client`）。
139    /// 爬取里每一页的抓取记录另带 `robotsIgnored`。
140    pub fn record_robots_ignored(&self, request: &str, url: &str, reason: &str) {
141        let mut rec = self.base("robots_ignored");
142        rec["url"] = json!(self.clean_url(url));
143        rec["request"] = json!(request);
144        rec["reason"] = json!(reason);
145        self.write(&rec);
146    }
147
148    /// 记一次上传文件解析（`/v2/parse`）：`name` 是 `upload:<文件名>`，另记文件大小；不存快照。
149    pub fn record_parse(
150        &self,
151        name: &str,
152        file_bytes: usize,
153        opts: &ScrapeOptions,
154        result: &Result<Document, ApiError>,
155        elapsed: Duration,
156    ) {
157        let mut rec = self.scrape_record("parse", name, opts, result, None, elapsed);
158        rec["fileBytes"] = json!(file_bytes);
159        self.write(&rec);
160    }
161
162    fn scrape_record(
163        &self,
164        kind: &str,
165        url: &str,
166        opts: &ScrapeOptions,
167        result: &Result<Document, ApiError>,
168        raw_html: Option<&str>,
169        elapsed: Duration,
170    ) -> Value {
171        let mut rec = self.base(kind);
172        rec["url"] = json!(self.clean_url(url));
173        rec["engineRequested"] = json!(opts.engine.as_deref().unwrap_or("auto"));
174        rec["ms"] = json!((elapsed.as_secs_f64() * 1000.0 * 10.0).round() / 10.0);
175        let mut suspicious: Option<String> = None;
176        match result {
177            Ok(doc) => {
178                let m = &doc.metadata;
179                let p = m.pagesift.as_ref();
180                if let Some(u) = m.url.as_deref().filter(|u| *u != url) {
181                    rec["finalUrl"] = json!(self.clean_url(u));
182                }
183                rec["status"] = json!(m.status_code);
184                rec["contentType"] = json!(m.content_type);
185                rec["engine"] = json!(p.and_then(|p| p.engine.as_deref()));
186                rec["browserEngine"] = json!(p.and_then(|p| p.browser_engine.as_deref()));
187                rec["cache"] = json!(m.cache_state);
188                let chars = doc
189                    .markdown
190                    .as_deref()
191                    .map(|md| md.chars().filter(|c| !c.is_whitespace()).count());
192                rec["textChars"] = json!(chars);
193                rec["links"] = json!(doc.links.as_ref().map(Vec::len));
194                if let Some(b) = raw_html.map(str::len) {
195                    rec["htmlBytes"] = json!(b);
196                }
197                let warnings: &[String] = p.map(|p| p.warnings.as_slice()).unwrap_or_default();
198                if !warnings.is_empty() {
199                    rec["warnings"] = json!(warnings);
200                }
201                if let Some(w) = warnings
202                    .iter()
203                    .find(|w| SUSPICIOUS_WARNINGS.iter().any(|s| w.starts_with(s)))
204                {
205                    suspicious = Some(w.split(':').next().unwrap_or(w).to_string());
206                } else if m.status_code == 200
207                    && m.cache_state.as_deref() != Some("hit")
208                    && m.content_type
209                        .as_deref()
210                        .is_some_and(|t| t.contains("html"))
211                    && chars.is_some_and(|c| c < SHORT_TEXT_CHARS)
212                {
213                    suspicious = Some("short_text".into());
214                }
215            }
216            Err(e) => {
217                rec["errorCode"] = json!(e.code.as_str());
218                rec["error"] = json!(truncate(&e.message, 300));
219                if let Some(rule) = e
220                    .details
221                    .as_ref()
222                    .and_then(|d| d.get("ruleId"))
223                    .and_then(Value::as_str)
224                {
225                    rec["barrierRule"] = json!(rule);
226                }
227                // robots 禁止、URL 不合法、内网地址、不支持的参数是调用方的事，不算 pagesift 的可疑结果
228                if !matches!(
229                    e.code,
230                    ErrorCode::RobotsDisallowed
231                        | ErrorCode::InvalidRequest
232                        | ErrorCode::SsrfBlocked
233                        | ErrorCode::UnsupportedOption
234                ) {
235                    suspicious = Some("error".into());
236                }
237            }
238        }
239        if let Some(reason) = &suspicious {
240            rec["suspicious"] = json!(reason);
241            if let Some(html) = raw_html
242                && let Some(path) = self.snapshot(url, html)
243            {
244                rec["snapshot"] = json!(path.display().to_string());
245            }
246        }
247        rec
248    }
249
250    /// 调用方（MCP 的 `report_issue`）报告的问题，和抓取记录写进同一个日志。
251    pub fn record_report(&self, report: Value) {
252        let mut rec = self.base("report");
253        if let Some(u) = report.get("url").and_then(Value::as_str) {
254            rec["url"] = json!(self.clean_url(u));
255        }
256        for key in ["params", "expected", "actual", "note"] {
257            if let Some(v) = report.get(key) {
258                rec[key] = match v {
259                    Value::String(s) => json!(truncate(s, 2000)),
260                    other => other.clone(),
261                };
262            }
263        }
264        self.write(&rec);
265    }
266
267    fn base(&self, kind: &str) -> Value {
268        let mut rec = json!({
269            "ts": pagesift_api::rfc3339(SystemTime::now()),
270            "kind": kind,
271            "version": env!("CARGO_PKG_VERSION"),
272            "source": self.cfg.source,
273        });
274        if let Ok(client) = CLIENT.try_with(Clone::clone) {
275            rec["client"] = json!(client);
276        }
277        rec
278    }
279
280    /// 只收集记录、不写本地文件（没配置审计日志的工人：记录随结果交给网关）。不存快照。
281    pub fn collect_only(source: &str) -> Self {
282        let mut cfg = AuditConfig::new("");
283        cfg.snapshot_dir = None;
284        cfg.source = source.into();
285        Self {
286            cfg,
287            log: None,
288            snap_day: AtomicU64::new(0),
289            snap_count: AtomicU32::new(0),
290        }
291    }
292
293    /// 网关：追加一条工人交来的记录，带上工人 id。
294    pub fn append_from(&self, mut rec: Value, worker: &str) {
295        rec["worker"] = json!(worker);
296        self.write(&rec);
297    }
298
299    fn write(&self, rec: &Value) {
300        let _ = COLLECT.try_with(|c| c.borrow_mut().push(rec.clone()));
301        if let Some(log) = &self.log {
302            let mut line = rec.to_string();
303            line.push('\n');
304            log.write(line);
305        }
306    }
307
308    /// 等已记录的行都写到文件里（测试和需要立刻读日志的调用方用）。
309    pub fn flush(&self) {
310        if let Some(log) = &self.log {
311            log.flush();
312        }
313    }
314
315    fn snapshot(&self, url: &str, html: &str) -> Option<PathBuf> {
316        let dir = self.cfg.snapshot_dir.as_ref()?;
317        let now = SystemTime::now().duration_since(UNIX_EPOCH).ok()?;
318        let day = now.as_secs() / 86_400;
319        if self.snap_day.swap(day, Ordering::Relaxed) != day {
320            self.snap_count.store(0, Ordering::Relaxed);
321        }
322        if self.snap_count.fetch_add(1, Ordering::Relaxed) >= self.cfg.snapshots_per_day {
323            return None;
324        }
325        let date = pagesift_api::rfc3339(SystemTime::now());
326        let dir = dir.join(date.get(..10).unwrap_or("unknown"));
327        let hash = pagesift_content::hash_hex(url.as_bytes());
328        let path = dir.join(format!(
329            "{}-{}.html",
330            now.as_millis(),
331            hash.get(..12).unwrap_or(&hash)
332        ));
333        let mut end = html.len().min(self.cfg.snapshot_max_bytes);
334        while !html.is_char_boundary(end) {
335            end -= 1;
336        }
337        // 快照可能有几 MB，不在异步线程上写
338        let body = format!("<!-- pagesift snapshot of {url} -->\n{}", &html[..end]);
339        let target = path.clone();
340        let write = move || -> io::Result<()> {
341            std::fs::create_dir_all(target.parent().unwrap_or(Path::new(".")))?;
342            std::fs::write(&target, body)
343        };
344        match tokio::runtime::Handle::try_current() {
345            Ok(h) => {
346                h.spawn_blocking(move || {
347                    if let Err(e) = write() {
348                        tracing::warn!("audit snapshot write failed: {e}");
349                    }
350                });
351            }
352            Err(_) => write().ok()?,
353        }
354        Some(path)
355    }
356
357    fn clean_url(&self, raw: &str) -> String {
358        if !self.cfg.redact {
359            return raw.to_string();
360        }
361        redact_url(raw)
362    }
363}
364
365/// 去掉 URL 里的用户名密码，查询参数里像凭证的值换成 `REDACTED`。
366pub fn redact_url(raw: &str) -> String {
367    let Ok(mut u) = Url::parse(raw) else {
368        return raw.to_string();
369    };
370    let _ = u.set_username("");
371    let _ = u.set_password(None);
372    if u.query().is_some() {
373        let pairs: Vec<(String, String)> = u
374            .query_pairs()
375            .map(|(k, v)| {
376                let v = if secret_param(&k) {
377                    "REDACTED".to_string()
378                } else {
379                    v.into_owned()
380                };
381                (k.into_owned(), v)
382            })
383            .collect();
384        u.query_pairs_mut().clear().extend_pairs(pairs);
385    }
386    u.to_string()
387}
388
389fn secret_param(name: &str) -> bool {
390    const HINTS: &[&str] = &[
391        "token",
392        "key",
393        "secret",
394        "password",
395        "passwd",
396        "pwd",
397        "auth",
398        "session",
399        "sig",
400        "jwt",
401        "credential",
402        "code",
403        "ticket",
404    ];
405    let n = name.to_ascii_lowercase();
406    HINTS.iter().any(|h| n.contains(h))
407}
408
409fn truncate(s: &str, max: usize) -> &str {
410    let mut end = s.len().min(max);
411    while !s.is_char_boundary(end) {
412        end -= 1;
413    }
414    &s[..end]
415}
416
417#[cfg(test)]
418mod tests {
419    use super::*;
420
421    #[test]
422    fn urls_lose_credentials_but_keep_ordinary_parameters() {
423        assert_eq!(
424            redact_url("https://u:p@example.com/a?q=rust&api_key=abc&X-Amz-Signature=zz&page=2"),
425            "https://example.com/a?q=rust&api_key=REDACTED&X-Amz-Signature=REDACTED&page=2"
426        );
427        assert_eq!(redact_url("not a url"), "not a url");
428    }
429
430    #[test]
431    fn records_rotate_and_flag_suspicious_results() {
432        let dir = tempfile::tempdir().unwrap();
433        let mut cfg = AuditConfig::new(dir.path().join("audit.jsonl"));
434        cfg.max_bytes = 600;
435        cfg.source = "test".into();
436        let audit = Audit::open(cfg).unwrap();
437        let mut doc = Document {
438            markdown: Some("tiny".into()),
439            ..Document::default()
440        };
441        doc.metadata.status_code = 200;
442        doc.metadata.content_type = Some("text/html".into());
443        let opts = ScrapeOptions::default();
444        for _ in 0..4 {
445            audit.record_scrape(
446                "scrape",
447                "https://example.com/p?token=t",
448                &opts,
449                &Ok(doc.clone()),
450                Some("<html>x</html>"),
451                Duration::from_millis(5),
452            );
453        }
454        audit.flush();
455        let log = std::fs::read_to_string(dir.path().join("audit.jsonl")).unwrap();
456        let old = std::fs::read_to_string(dir.path().join("audit.jsonl.1")).unwrap();
457        assert!(!log.is_empty() && !old.is_empty(), "rotated");
458        let rec: Value = serde_json::from_str(log.lines().last().unwrap()).unwrap();
459        assert_eq!(rec["suspicious"], "short_text");
460        assert_eq!(rec["url"], "https://example.com/p?token=REDACTED");
461        assert_eq!(rec["textChars"], 4);
462        let snap = rec["snapshot"].as_str().unwrap();
463        assert!(std::path::Path::new(snap).exists());
464
465        let err = ApiError::new(ErrorCode::RobotsDisallowed, "disallowed");
466        audit.record_scrape(
467            "scrape",
468            "https://example.com/x",
469            &opts,
470            &Err(err),
471            None,
472            Duration::ZERO,
473        );
474        audit.flush();
475        let log = std::fs::read_to_string(dir.path().join("audit.jsonl")).unwrap();
476        let rec: Value = serde_json::from_str(log.lines().last().unwrap()).unwrap();
477        assert_eq!(rec["errorCode"], "ROBOTS_DISALLOWED");
478        assert!(rec.get("suspicious").is_none());
479    }
480
481    #[test]
482    fn ignored_robots_are_recorded_with_url_client_and_reason() {
483        let dir = tempfile::tempdir().unwrap();
484        let audit = Audit::open(AuditConfig::new(dir.path().join("audit.jsonl"))).unwrap();
485        let mut doc = Document::default();
486        doc.metadata.status_code = 200;
487        let opts = ScrapeOptions::default();
488        CLIENT.sync_scope("agent-x".to_string(), || {
489            audit.record_scrape_ignoring_robots(
490                "scrape",
491                "https://example.com/private?token=t",
492                &opts,
493                &Ok(doc),
494                None,
495                Duration::ZERO,
496                "ignoreRobotsTxt",
497            );
498            audit.record_robots_ignored(
499                "crawl",
500                "https://example.com/",
501                "ignoreRobotsTxt+allow_crawl_ignore_robots",
502            );
503        });
504        audit.flush();
505        let log = std::fs::read_to_string(dir.path().join("audit.jsonl")).unwrap();
506        let recs: Vec<Value> = log
507            .lines()
508            .map(|l| serde_json::from_str(l).unwrap())
509            .collect();
510        assert_eq!(recs[0]["robotsIgnored"], "ignoreRobotsTxt");
511        assert_eq!(recs[0]["url"], "https://example.com/private?token=REDACTED");
512        assert_eq!(recs[0]["client"], "agent-x");
513        assert_eq!(recs[1]["kind"], "robots_ignored");
514        assert_eq!(recs[1]["request"], "crawl");
515        assert_eq!(
516            recs[1]["reason"],
517            "ignoreRobotsTxt+allow_crawl_ignore_robots"
518        );
519        assert_eq!(recs[1]["client"], "agent-x");
520    }
521}
pagesift_sink/audit/mod.rs

pagesift_sink/audit/
mod.rs
