forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
6.5 KiB
1//! The advisory worker. When `[advisories] enabled = true`, it keeps an OSV dump per configured
2//! ecosystem in `<data_dir>/advisories/<ecosystem>/all.zip`. A dump older than `refresh_hours` is
3//! downloaded again (capped at [`MAX_DUMP_BYTES`], with timeouts, no redirects). A dump that is
4//! newer than the last import is imported, and then every repo is rescanned. It checks once an
5//! hour, and once at boot, so an existing dump is imported without a download.
6
7use std::collections::HashMap;
8use std::fs::{self, File};
9use std::io::{BufReader, Write};
10use std::path::{Path, PathBuf};
11use std::time::{Duration, SystemTime};
12
13use forge_core::advisories::{KNOWN_ECOSYSTEMS, read_dump};
14use forge_core::{Error, Result};
15
16use super::scan::scan_all;
17use crate::state::AppState;
18
19/// How often the worker checks the dumps.
20const CHECK_EVERY: Duration = Duration::from_secs(60 * 60);
21/// Largest dump accepted, in bytes.
22const MAX_DUMP_BYTES: u64 = 500 * 1024 * 1024;
23const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
24/// Time allowed for one whole download.
25const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(30 * 60);
26const USER_AGENT: &str = concat!("forge-advisories/", env!("CARGO_PKG_VERSION"));
27
28/// Starts the worker task, if advisories are enabled. It runs until the process exits.
29pub fn spawn(state: AppState) {
30 if !state.config.advisories.enabled {
31 return;
32 }
33 tokio::spawn(run(state));
34}
35
36/// The OSV dump URL for an ecosystem.
37pub fn dump_url(ecosystem: &str) -> String {
38 format!("https://osv-vulnerabilities.storage.googleapis.com/{ecosystem}/all.zip")
39}
40
41/// Where the dump of `ecosystem` is kept.
42pub fn dump_path(data_dir: &Path, ecosystem: &str) -> PathBuf {
43 data_dir.join("advisories").join(ecosystem).join("all.zip")
44}
45
46async fn run(state: AppState) {
47 let client = match reqwest::Client::builder()
48 .redirect(reqwest::redirect::Policy::none())
49 .connect_timeout(CONNECT_TIMEOUT)
50 .timeout(DOWNLOAD_TIMEOUT)
51 .user_agent(USER_AGENT)
52 .build()
53 {
54 Ok(client) => client,
55 Err(e) => {
56 tracing::error!(error = %e, "advisories: could not build the HTTP client");
57 return;
58 }
59 };
60 let mut imported: HashMap<String, SystemTime> = HashMap::new();
61 loop {
62 for ecosystem in &state.config.advisories.ecosystems {
63 if !KNOWN_ECOSYSTEMS.contains(&ecosystem.as_str()) {
64 tracing::warn!(ecosystem = %ecosystem, "advisories: unknown ecosystem, skipped");
65 continue;
66 }
67 if let Err(e) = refresh_one(&state, &client, ecosystem, &mut imported).await {
68 tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: refresh failed");
69 }
70 }
71 tokio::time::sleep(CHECK_EVERY).await;
72 }
73}
74
75/// Downloads `ecosystem`'s dump if it is stale, then imports it if it changed since the last import.
76async fn refresh_one(
77 state: &AppState,
78 client: &reqwest::Client,
79 ecosystem: &str,
80 imported: &mut HashMap<String, SystemTime>,
81) -> Result<()> {
82 let path = dump_path(&state.config.data_dir, ecosystem);
83 let max_age = Duration::from_secs(state.config.advisories.refresh_hours.max(1) * 3600);
84 if is_older_than(&path, max_age) {
85 download(client, ecosystem, &path).await?;
86 }
87 let Some(modified) = modified_time(&path) else {
88 return Ok(());
89 };
90 if imported.get(ecosystem) == Some(&modified) {
91 return Ok(());
92 }
93 import(state, ecosystem, &path).await?;
94 imported.insert(ecosystem.to_owned(), modified);
95 scan_all(state).await;
96 Ok(())
97}
98
99fn modified_time(path: &Path) -> Option<SystemTime> {
100 fs::metadata(path).and_then(|m| m.modified()).ok()
101}
102
103/// True when the file is missing or was written more than `max_age` ago.
104fn is_older_than(path: &Path, max_age: Duration) -> bool {
105 match modified_time(path) {
106 None => true,
107 Some(at) => SystemTime::now()
108 .duration_since(at)
109 .is_ok_and(|age| age > max_age),
110 }
111}
112
113/// Streams the dump to `<path>.part` and renames it into place once complete. A response that is
114/// too large, or that fails midway, leaves the previous dump in place.
115async fn download(client: &reqwest::Client, ecosystem: &str, path: &Path) -> Result<()> {
116 let url = dump_url(ecosystem);
117 let dir = path
118 .parent()
119 .ok_or_else(|| Error::Config("advisory dump path has no directory".into()))?;
120 fs::create_dir_all(dir)?;
121 let part = path.with_extension("zip.part");
122 let outcome = fetch_into(client, &url, &part).await;
123 match outcome {
124 Ok(bytes) => {
125 fs::rename(&part, path)?;
126 tracing::info!(ecosystem, bytes, "advisories: dump downloaded");
127 Ok(())
128 }
129 Err(e) => {
130 let _ = fs::remove_file(&part);
131 Err(e)
132 }
133 }
134}
135
136async fn fetch_into(client: &reqwest::Client, url: &str, part: &Path) -> Result<u64> {
137 let mut response = client
138 .get(url)
139 .send()
140 .await
141 .and_then(reqwest::Response::error_for_status)
142 .map_err(|e| Error::Invalid(format!("advisory download failed: {e}")))?;
143 if response
144 .content_length()
145 .is_some_and(|len| len > MAX_DUMP_BYTES)
146 {
147 return Err(Error::Invalid(
148 "advisory dump is larger than 500 MiB".into(),
149 ));
150 }
151 let mut file = File::create(part)?;
152 let mut total: u64 = 0;
153 while let Some(chunk) = response
154 .chunk()
155 .await
156 .map_err(|e| Error::Invalid(format!("advisory download interrupted: {e}")))?
157 {
158 total += chunk.len() as u64;
159 if total > MAX_DUMP_BYTES {
160 return Err(Error::Invalid(
161 "advisory dump is larger than 500 MiB".into(),
162 ));
163 }
164 file.write_all(&chunk)?;
165 }
166 file.sync_all()?;
167 Ok(total)
168}
169
170/// Parses the dump on a blocking thread and replaces the stored advisories of `ecosystem`.
171async fn import(state: &AppState, ecosystem: &str, path: &Path) -> Result<()> {
172 let owned_path = path.to_owned();
173 let eco = ecosystem.to_owned();
174 let (advisories, skipped) = tokio::task::spawn_blocking(move || -> Result<_> {
175 let file = File::open(&owned_path)?;
176 read_dump(BufReader::new(file), &eco)
177 })
178 .await
179 .map_err(|e| Error::Invalid(format!("advisory import task failed: {e}")))??;
180 state
181 .store
182 .replace_advisories(ecosystem, &advisories)
183 .await?;
184 tracing::info!(
185 ecosystem,
186 imported = advisories.len(),
187 skipped,
188 "advisories: dump imported"
189 );
190 Ok(())
191}