forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
8.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`. Each dump is downloaded again
3//! `refresh_hours` after its last refresh (the file's modification time), sleeping until then
4//! (capped at [`MAX_DUMP_BYTES`], with timeouts, no redirects). A failed download is retried with
5//! a backoff that doubles from [`RETRY_FIRST`] up to [`RETRY_MAX`]. A dump that is newer than the
6//! last import is imported, and then every repo is rescanned. At boot, a fresh dump is imported
7//! without a download.
8
9use std::collections::HashMap;
10use std::fs::{self, File};
11use std::io::{BufReader, Write};
12use std::path::{Path, PathBuf};
13use std::time::{Duration, SystemTime};
14
15use forge_core::advisories::{KNOWN_ECOSYSTEMS, read_dump};
16use forge_core::{Error, Result};
17
18use super::scan::scan_all;
19use crate::state::AppState;
20
21/// First wait after a failed download. It doubles with each further failure.
22const RETRY_FIRST: Duration = Duration::from_secs(5 * 60);
23/// Longest wait after a failed download.
24const RETRY_MAX: Duration = Duration::from_secs(60 * 60);
25/// Largest dump accepted, in bytes.
26const MAX_DUMP_BYTES: u64 = 500 * 1024 * 1024;
27const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
28/// Time allowed for one whole download.
29const DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(30 * 60);
30const USER_AGENT: &str = concat!("forge-advisories/", env!("CARGO_PKG_VERSION"));
31
32/// Starts the worker task, if advisories are enabled. It runs until the process exits.
33pub fn spawn(state: AppState) {
34 if !state.config.advisories.enabled {
35 return;
36 }
37 tokio::spawn(run(state));
38}
39
40/// The OSV dump URL for an ecosystem.
41pub fn dump_url(ecosystem: &str) -> String {
42 format!("https://osv-vulnerabilities.storage.googleapis.com/{ecosystem}/all.zip")
43}
44
45/// Where the dump of `ecosystem` is kept.
46pub fn dump_path(data_dir: &Path, ecosystem: &str) -> PathBuf {
47 data_dir.join("advisories").join(ecosystem).join("all.zip")
48}
49
50async fn run(state: AppState) {
51 let client = match reqwest::Client::builder()
52 .redirect(reqwest::redirect::Policy::none())
53 .connect_timeout(CONNECT_TIMEOUT)
54 .timeout(DOWNLOAD_TIMEOUT)
55 .user_agent(USER_AGENT)
56 .build()
57 {
58 Ok(client) => client,
59 Err(e) => {
60 tracing::error!(error = %e, "advisories: could not build the HTTP client");
61 return;
62 }
63 };
64 let refresh_after = Duration::from_secs(state.config.advisories.refresh_hours.max(1) * 3600);
65 let mut imported: HashMap<String, SystemTime> = HashMap::new();
66 // When each ecosystem's dump is next downloaded, and how many downloads in a row failed.
67 let mut due: HashMap<String, SystemTime> = HashMap::new();
68 let mut failures: HashMap<String, u32> = HashMap::new();
69 loop {
70 let mut wake_at: Option<SystemTime> = None;
71 for ecosystem in &state.config.advisories.ecosystems {
72 if !KNOWN_ECOSYSTEMS.contains(&ecosystem.as_str()) {
73 tracing::warn!(ecosystem = %ecosystem, "advisories: unknown ecosystem, skipped");
74 continue;
75 }
76 let path = dump_path(&state.config.data_dir, ecosystem);
77 let mut refresh_at = *due
78 .entry(ecosystem.clone())
79 .or_insert_with(|| next_refresh_at(&path, refresh_after));
80 let now = SystemTime::now();
81 if refresh_at <= now {
82 refresh_at = match download(&client, ecosystem, &path).await {
83 Ok(()) => {
84 failures.remove(ecosystem);
85 next_refresh_at(&path, refresh_after)
86 }
87 Err(e) => {
88 tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: refresh failed");
89 let count = failures.entry(ecosystem.clone()).or_insert(0);
90 *count += 1;
91 now + retry_delay(*count)
92 }
93 };
94 due.insert(ecosystem.clone(), refresh_at);
95 }
96 if let Err(e) = import_if_changed(&state, ecosystem, &path, &mut imported).await {
97 tracing::error!(error = %e, ecosystem = %ecosystem, "advisories: import failed");
98 }
99 wake_at = Some(wake_at.map_or(refresh_at, |at| at.min(refresh_at)));
100 }
101 let Some(wake_at) = wake_at else {
102 tracing::info!("advisories: no known ecosystems configured, worker stopped");
103 return;
104 };
105 let wait = wake_at
106 .duration_since(SystemTime::now())
107 .unwrap_or(Duration::ZERO);
108 tokio::time::sleep(wait).await;
109 }
110}
111
112/// When the dump at `path` is due for its next download: `refresh_after` past its last refresh.
113/// A missing dump is due now.
114fn next_refresh_at(path: &Path, refresh_after: Duration) -> SystemTime {
115 match modified_time(path) {
116 None => SystemTime::now(),
117 Some(last) => last + refresh_after,
118 }
119}
120
121/// The wait after the `failures`-th failed download in a row: 5 minutes, doubling, capped at an hour.
122fn retry_delay(failures: u32) -> Duration {
123 let shift = failures.saturating_sub(1).min(16);
124 (RETRY_FIRST * 2u32.pow(shift)).min(RETRY_MAX)
125}
126
127/// Imports the dump of `ecosystem` if it changed since the last import, then rescans every repo.
128async fn import_if_changed(
129 state: &AppState,
130 ecosystem: &str,
131 path: &Path,
132 imported: &mut HashMap<String, SystemTime>,
133) -> Result<()> {
134 let Some(modified) = modified_time(path) else {
135 return Ok(());
136 };
137 if imported.get(ecosystem) == Some(&modified) {
138 return Ok(());
139 }
140 import(state, ecosystem, path).await?;
141 imported.insert(ecosystem.to_owned(), modified);
142 scan_all(state).await;
143 Ok(())
144}
145
146fn modified_time(path: &Path) -> Option<SystemTime> {
147 fs::metadata(path).and_then(|m| m.modified()).ok()
148}
149
150/// Streams the dump to `<path>.part` and renames it into place once complete. A response that is
151/// too large, or that fails midway, leaves the previous dump in place.
152async fn download(client: &reqwest::Client, ecosystem: &str, path: &Path) -> Result<()> {
153 let url = dump_url(ecosystem);
154 let dir = path
155 .parent()
156 .ok_or_else(|| Error::Config("advisory dump path has no directory".into()))?;
157 fs::create_dir_all(dir)?;
158 let part = path.with_extension("zip.part");
159 let outcome = fetch_into(client, &url, &part).await;
160 match outcome {
161 Ok(bytes) => {
162 fs::rename(&part, path)?;
163 tracing::info!(ecosystem, bytes, "advisories: dump downloaded");
164 Ok(())
165 }
166 Err(e) => {
167 let _ = fs::remove_file(&part);
168 Err(e)
169 }
170 }
171}
172
173async fn fetch_into(client: &reqwest::Client, url: &str, part: &Path) -> Result<u64> {
174 let mut response = client
175 .get(url)
176 .send()
177 .await
178 .and_then(reqwest::Response::error_for_status)
179 .map_err(|e| Error::Invalid(format!("advisory download failed: {e}")))?;
180 if response
181 .content_length()
182 .is_some_and(|len| len > MAX_DUMP_BYTES)
183 {
184 return Err(Error::Invalid(
185 "advisory dump is larger than 500 MiB".into(),
186 ));
187 }
188 let mut file = File::create(part)?;
189 let mut total: u64 = 0;
190 while let Some(chunk) = response
191 .chunk()
192 .await
193 .map_err(|e| Error::Invalid(format!("advisory download interrupted: {e}")))?
194 {
195 total += chunk.len() as u64;
196 if total > MAX_DUMP_BYTES {
197 return Err(Error::Invalid(
198 "advisory dump is larger than 500 MiB".into(),
199 ));
200 }
201 file.write_all(&chunk)?;
202 }
203 file.sync_all()?;
204 Ok(total)
205}
206
207/// Parses the dump on a blocking thread and replaces the stored advisories of `ecosystem`.
208async fn import(state: &AppState, ecosystem: &str, path: &Path) -> Result<()> {
209 let owned_path = path.to_owned();
210 let eco = ecosystem.to_owned();
211 let (advisories, skipped) = tokio::task::spawn_blocking(move || -> Result<_> {
212 let file = File::open(&owned_path)?;
213 read_dump(BufReader::new(file), &eco)
214 })
215 .await
216 .map_err(|e| Error::Invalid(format!("advisory import task failed: {e}")))??;
217 state
218 .store
219 .replace_advisories(ecosystem, &advisories)
220 .await?;
221 tracing::info!(
222 ecosystem,
223 imported = advisories.len(),
224 skipped,
225 "advisories: dump imported"
226 );
227 Ok(())
228}