Sign in

flagon-io/g1t

public

g1t/services/security/src/security_updates.rs

462 lines22,076 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Git storage hardened, pages in tens of milliseconds, honest security alerts, and costs reconciled daily1//! Security updates: for each vulnerable package with a fix, g1t itself
2//! opens a pull request that raises its version.
3//!
4//! 1. A dependency scan asks the runner for a `bump` (most severe first):
5//! a sandbox raises the package in each lockfile with the ecosystem's
6//! own tool and pushes that to `g1t/security/<package>-<version>`.
7//! 2. That push (`git.push`) opens the pull request, authored by g1t
8//! (`User::system`). It lands through the branch's required checks like
9//! any other. An older one for the same package is closed as superseded.
10//! 3. When its checks fail because code has to change, or the sandbox
11//! never pushed, the pull request is closed and an issue is opened for
12//! g1t-agent, started by g1t. That is the only time an agent is used.
13//! 4. A package no longer vulnerable closes its update as superseded.
14//!
15//! Each step is written to the alerts' activity log.
16
17use std::collections::BTreeMap;
18
19use g1t_contracts::repos::RepoPath;
20use g1t_contracts::security::{BumpArgs, UpdateState, update_branch};
21use g1t_contracts::time::rfc3339;
22use g1t_contracts::work::{OpenIssueArgs, OpenPullArgs, Pull, PullActionArgs, PullDetail, PullStatus, Runtime, ViewArgs};
23use g1t_contracts::{Outcome, User};
24use g1t_kit::now_ms;
25use g1t_scan::osv::{self, Severity};
26use g1t_scan::version;
27use serde_json::{Value, json};
28use worker::Result;
29
30use crate::Security;
31use crate::deps::{advisory_table, issue_text};
32use crate::store::{Activity, RepoRow, UpdateRow, VulnRow};
33
34/// Security updates asked for per scan, most severe first.
35const MAX_NEW_UPDATES: usize = 8;
36/// A sandbox that has not pushed its branch after this long is taken to
37/// have failed.
38const STALLED_MS: u64 = 45 * 60 * 1000;
39/// A failed update is tried again after this long.
40const RETRY_FAILED_MS: u64 = 24 * 60 * 60 * 1000;
41
42/// What to do about a package's existing update, given the version it
43/// should now reach.
44#[derive(Debug, PartialEq, Eq)]
45pub enum Next {
46 /// Leave it: in flight, done, or someone closed it.
47 Wait,
48 /// Ask for a new one.
49 Request,
50}
51
52/// Whether a package with an update in `state` to `current`, last changed
53/// at `updated_at`, needs a new one to reach `target`. A higher target
54/// always does; the same one only when the last was superseded (it is
55/// vulnerable again) or failed long enough ago.
56pub fn next_step(state: UpdateState, current: &str, target: &str, updated_at: &str, retry_after: &str) -> Next {
57 if version::compare(target, current) == std::cmp::Ordering::Greater {
58 return Next::Request;
59 }
60 match state {
61 UpdateState::Superseded => Next::Request,
62 UpdateState::Failed if updated_at < retry_after => Next::Request,
63 _ => Next::Wait,
64 }
65}
66
67/// The pull request's title.
68pub fn pull_title(package: &str, target: &str) -> String {
69 format!("Upgrade {package} to {target}").chars().take(200).collect()
70}
71
72/// The pull request's body: what it fixes, where, and what happens if
73/// raising the version is not enough.
74pub fn pull_text(ecosystem: &str, package: &str, target: &str, vulns: &[&VulnRow]) -> String {
75 let mut from: Vec<&str> = vulns.iter().map(|vuln| vuln.version.as_str()).collect();
76 from.sort();
77 from.dedup();
78 let mut lockfiles: Vec<&str> = vulns.iter().map(|vuln| vuln.manifest.as_str()).collect();
79 lockfiles.sort();
80 lockfiles.dedup();
81 let mut body = format!(
82 "Upgrades `{package}` ({ecosystem}) from {} to **{target}**, which fixes these known vulnerabilities:\n\n",
83 from.join(", ")
84 );
85 body.push_str(&advisory_table(vulns));
86 body.push_str(&format!(
87 "\nLockfiles changed: {}.\n\n\
88 Only the version changes. This pull request lands through this branch's required checks like any other. \
89 If they fail because code has to change, g1t closes it and puts g1t-agent on an issue to make the change.\n\n\
90 ---\n_Opened by g1t's security updates. Turn them off for this project on its Security page._",
91 lockfiles.iter().map(|path| format!("`{path}`")).collect::<Vec<_>>().join(", ")
92 ));
93 body
94}
95
96impl Security {
97 fn path_of(repo: &RepoRow) -> RepoPath {
98 RepoPath { namespace: repo.namespace.clone(), name: repo.name.clone() }
99 }
100
101 async fn get_pull(&self, repo: &RepoRow, number: u32) -> Result<Option<Pull>> {
102 let found: Outcome<PullDetail> = g1t_kit::call(
103 &self.work,
104 "get_pull",
105 &ViewArgs { repo: Self::path_of(repo), number, viewer: Some(User::system(&repo.namespace)), after_seq: 0 },
106 )
107 .await?;
108 Ok(found.into_result().ok().map(|detail| detail.pull))
109 }
110
111 /// Closes one of g1t's pull requests, saying why first.
112 async fn close_with(&self, repo: &RepoRow, number: u32, why: String) -> Result<()> {
113 let system = User::system(&repo.namespace);
114 self.comment(&system, &Self::path_of(repo), number, why).await?;
115 let _: Outcome<Value> = g1t_kit::call(
116 &self.work,
117 "close_pull",
118 &PullActionArgs {
119 actor: system,
120 repo: Self::path_of(repo),
121 number,
122 summary: String::new(),
123 keep_issue_open: false,
124 ignore_checks: false,
125 },
126 )
127 .await?;
128 Ok(())
129 }
130
131 /// Records `action` on every open alert of an update's package.
132 async fn note(&self, row: &UpdateRow, action: &str, actor: Option<&str>, number: Option<u32>, comment: Option<&str>) -> Result<()> {
133 let ids = self.store.open_ids(&row.repo_id, &row.ecosystem, &row.package).await?;
134 let activity: Vec<Activity> = ids
135 .iter()
136 .map(|id| Activity { alert_id: id, action, actor, reason: None, comment, number })
137 .collect();
138 self.store.record(&row.repo_id, &activity).await
139 }
140
141 /// After a dependency scan: asks for an update for each vulnerable
142 /// package with a fix (when `enabled`), and supersedes those whose
143 /// package is no longer vulnerable.
144 pub async fn security_updates(&self, repo: &RepoRow, enabled: bool) -> Result<()> {
145 let open = self.store.open_vulnerabilities(&repo.repo_id).await?;
146 let mut by_package: BTreeMap<(String, String), Vec<&VulnRow>> = BTreeMap::new();
147 for vuln in &open {
148 by_package.entry((vuln.ecosystem.clone(), vuln.package.clone())).or_default().push(vuln);
149 }
150 // In flight for a package that is no longer vulnerable: no longer needed.
151 for row in self.store.updates(&repo.repo_id).await? {
152 if !row.state().in_progress() || by_package.contains_key(&(row.ecosystem.clone(), row.package.clone())) {
153 continue;
154 }
155 self.supersede(repo, &row, format!("`{}` is no longer vulnerable here, so this is no longer needed.", row.package))
156 .await?;
157 }
158 if !enabled {
159 return Ok(());
160 }
161 let mut groups: Vec<((String, String), Vec<&VulnRow>)> = by_package
162 .into_iter()
163 .filter(|(_, vulns)| vulns.iter().any(|vuln| vuln.fixed_version.is_some()))
164 .collect();
165 groups.sort_by_key(|(_, vulns)| std::cmp::Reverse(vulns.iter().map(|v| Severity::parse(&v.severity)).max()));
166 let retry_after = rfc3339(now_ms().saturating_sub(RETRY_FAILED_MS));
167 let mut asked = 0;
168 for ((ecosystem, package), vulns) in groups {
169 if asked >= MAX_NEW_UPDATES {
170 break;
171 }
172 let Some(target) = osv::upgrade_target(vulns.iter().filter_map(|v| v.fixed_version.as_deref())) else {
173 continue;
174 };
175 match self.store.update(&repo.repo_id, &ecosystem, &package).await? {
176 Some(row) => {
177 if next_step(row.state(), &row.target, &target, &row.updated_at, &retry_after) == Next::Wait {
178 continue;
179 }
180 }
181 None => {
182 // An upgrade issue from before security updates, still
183 // open, is left to the agent on it.
184 if let Some(existing) = self.store.upgrade(&repo.repo_id, &ecosystem, &package).await?
185 && self
186 .issue(&User::system(&repo.namespace), &Self::path_of(repo), existing.number as u32)
187 .await?
188 .is_some_and(|issue| issue.state == g1t_contracts::work::State::Open)
189 {
190 continue;
191 }
192 }
193 }
194 asked += 1;
195 self.request(repo, &ecosystem, &package, &target, &vulns).await?;
196 }
197 Ok(())
198 }
199
200 /// Asks the runner to make the change on its branch.
201 async fn request(&self, repo: &RepoRow, ecosystem: &str, package: &str, target: &str, vulns: &[&VulnRow]) -> Result<()> {
202 let branch = update_branch(package, target);
203 let mut lockfiles: Vec<String> = vulns.iter().map(|vuln| vuln.manifest.clone()).collect();
204 lockfiles.sort();
205 lockfiles.dedup();
206 let args = BumpArgs {
207 repo: Self::path_of(repo),
208 ecosystem: ecosystem.to_owned(),
209 package: package.to_owned(),
210 version: target.to_owned(),
211 lockfiles,
212 branch: branch.clone(),
213 message: format!("Upgrade {package} to {target}"),
214 };
215 let started: std::result::Result<(), String> = match g1t_kit::call::<_, Outcome<bool>>(&self.runner, "bump", &args).await {
216 Ok(Outcome::Ok(_)) => Ok(()),
217 Ok(Outcome::Fail(refused)) => Err(refused.message),
218 Err(error) => Err(format!("the runner could not be reached: {error}")),
219 };
220 self.store.request_update(&repo.repo_id, ecosystem, package, target, &branch).await?;
221 let Some(row) = self.store.update(&repo.repo_id, ecosystem, package).await? else {
222 return Ok(());
223 };
224 match started {
225 Ok(()) => self.note(&row, "update_requested", Some(g1t_contracts::system::USERNAME), None, None).await,
226 Err(reason) => {
227 let error = format!("g1t could not start the security update: {reason}");
228 self.store.set_update(&row, UpdateState::Failed, row.pull(), None, Some(&error)).await?;
229 self.note(&row, "update_failed", Some(g1t_contracts::system::USERNAME), None, Some(&error)).await
230 }
231 }
232 }
233
234 /// A security update's branch was pushed: its pull request opens, and
235 /// an older one for the same package is closed as superseded.
236 pub async fn update_pushed(&self, repo_id: &str, branch: &str) -> Result<()> {
237 let Some(row) = self.store.update_by_branch(repo_id, branch).await? else {
238 return Ok(());
239 };
240 if row.state() != UpdateState::Requested {
241 return Ok(());
242 }
243 let Some(repo) = self.store.repo(repo_id).await? else {
244 return Ok(());
245 };
246 let open = self.store.open_vulnerabilities(repo_id).await?;
247 let vulns: Vec<&VulnRow> = open
248 .iter()
249 .filter(|vuln| vuln.ecosystem == row.ecosystem && vuln.package == row.package)
250 .collect();
251 let system = User::system(&repo.namespace);
252 let opened: Outcome<Pull> = g1t_kit::call(
253 &self.work,
254 "open_pull",
255 &OpenPullArgs {
256 actor: system,
257 repo: Self::path_of(&repo),
258 issue: None,
259 title: pull_title(&row.package, &row.target),
260 body: pull_text(&row.ecosystem, &row.package, &row.target, &vulns),
261 branch: Some(branch.to_owned()),
262 agent: String::new(),
263 runtime: Runtime::External,
264 },
265 )
266 .await?;
267 let pull = match opened {
268 Outcome::Ok(pull) => pull,
269 Outcome::Fail(refused) => {
270 let error = format!("The pull request could not be opened: {}", refused.message);
271 self.store.set_update(&row, UpdateState::Failed, row.pull(), None, Some(&error)).await?;
272 return self.note(&row, "update_failed", Some(g1t_contracts::system::USERNAME), None, Some(&error)).await;
273 }
274 };
275 if let Some(older) = row.pull().filter(|older| *older != pull.number) {
276 self.close_with(&repo, older, format!("Superseded by #{}, which upgrades `{}` to {}.", pull.number, row.package, row.target))
277 .await?;
278 }
279 self.store.set_update(&row, UpdateState::Open, Some(pull.number), None, None).await?;
280 self.note(&row, "update_opened", Some(g1t_contracts::system::USERNAME), Some(pull.number), None).await
281 }
282
283 /// A pull request merged, closed or checked: if it is a security
284 /// update's, where the update stands now.
285 pub async fn update_pull_event(&self, kind: &str, repo_id: &str, number: u32, status: Option<&str>, actor: Option<&str>) -> Result<()> {
286 let Some(row) = self.store.update_by_pull(repo_id, number).await? else {
287 return Ok(());
288 };
289 if row.state() != UpdateState::Open {
290 return Ok(());
291 }
292 match kind {
293 "pull.merged" => {
294 self.store.set_update(&row, UpdateState::Merged, Some(number), None, None).await?;
295 self.note(&row, "update_merged", actor, Some(number), None).await
296 }
297 "pull.closed" => {
298 self.store.set_update(&row, UpdateState::Closed, Some(number), None, None).await?;
299 self.note(&row, "update_closed", actor, Some(number), None).await
300 }
301 "checks.completed" if status == Some("failed") => {
302 let Some(repo) = self.store.repo(repo_id).await? else {
303 return Ok(());
304 };
305 self.needs_code(&repo, &row, "Raising the version alone fails this branch's required checks").await
306 }
307 _ => Ok(()),
308 }
309 }
310
311 /// Closes an update that is no longer needed: its pull request, if it
312 /// has one still open (one that merged meanwhile is recorded merged).
313 async fn supersede(&self, repo: &RepoRow, row: &UpdateRow, why: String) -> Result<()> {
314 if let Some(number) = row.pull() {
315 match self.get_pull(repo, number).await? {
316 Some(pull) if pull.status == PullStatus::Merged => {
317 return self.store.set_update(row, UpdateState::Merged, Some(number), row.issue(), None).await;
318 }
319 Some(pull) if matches!(pull.status, PullStatus::Open | PullStatus::Draft) => {
320 self.store.set_update(row, UpdateState::Superseded, Some(number), row.issue(), None).await?;
321 self.close_with(repo, number, why).await?;
322 return self.note(row, "update_superseded", Some(g1t_contracts::system::USERNAME), Some(number), None).await;
323 }
324 _ => {}
325 }
326 }
327 self.store.set_update(row, UpdateState::Superseded, row.pull(), row.issue(), None).await
328 }
329
330 /// Updates whose sandbox never pushed: raising the version did not
331 /// work, so g1t-agent gets an issue for it.
332 pub async fn stalled_updates(&self) -> Result<()> {
333 let before = rfc3339(now_ms().saturating_sub(STALLED_MS));
334 for row in self.store.stalled_updates(&before, 10).await? {
335 let Some(repo) = self.store.repo(&row.repo_id).await? else { continue };
336 if repo.upkeep == 0 || !self.active(&repo.repo_id).await? {
337 self.store
338 .set_update(&row, UpdateState::Failed, row.pull(), None, Some("The version could not be raised in a sandbox."))
339 .await?;
340 continue;
341 }
342 self.needs_code(&repo, &row, "The version could not be raised in a sandbox on its own").await?;
343 }
344 Ok(())
345 }
346
347 /// Raising the version is not enough: closes g1t's pull request, opens
348 /// an issue for the change, and puts g1t-agent on it, started by g1t.
349 async fn needs_code(&self, repo: &RepoRow, row: &UpdateRow, why: &str) -> Result<()> {
350 let path = Self::path_of(repo);
351 let system = User::system(&repo.namespace);
352 let open = self.store.open_vulnerabilities(&repo.repo_id).await?;
353 let vulns: Vec<&VulnRow> = open
354 .iter()
355 .filter(|vuln| vuln.ecosystem == row.ecosystem && vuln.package == row.package)
356 .collect();
357 if vulns.is_empty() {
358 return self.supersede(repo, row, format!("`{}` is no longer vulnerable here.", row.package)).await;
359 }
360 let issue: Outcome<g1t_contracts::work::Issue> = g1t_kit::call(
361 &self.work,
362 "open_issue",
363 &OpenIssueArgs {
364 actor: system.clone(),
365 repo: path.clone(),
366 title: format!("Upgrade {} to {}: needs code changes", row.package, row.target).chars().take(200).collect(),
367 body: issue_text(&row.ecosystem, &row.package, &row.target, &vulns, &[]),
368 labels: vec!["dependencies".to_owned(), "security".to_owned()],
369 checks: Vec::new(),
370 },
371 )
372 .await?;
373 let issue = match issue {
374 Outcome::Ok(issue) => issue,
375 Outcome::Fail(refused) => {
376 let error = format!("{why}, and the issue for it could not be opened: {}", refused.message);
377 self.store.set_update(row, UpdateState::Failed, row.pull(), None, Some(&error)).await?;
378 return self.note(row, "update_failed", Some(g1t_contracts::system::USERNAME), None, Some(&error)).await;
379 }
380 };
381 self.store
382 .set_update(row, UpdateState::NeedsCode, row.pull(), Some(issue.number), Some(why))
383 .await?;
384 if let Some(number) = row.pull() {
385 self.close_with(repo, number, format!("{why}, so code has to change too. g1t-agent is making the change in #{}.", issue.number))
386 .await?;
387 }
388 let started: Outcome<Value> =
389 g1t_kit::call(&self.runner, "run", &json!({ "actor": system, "repo": path, "issue": issue.number })).await?;
390 if let Outcome::Fail(refused) = started {
391 self.comment(&system, &path, issue.number, format!(
392 "g1t could not put an agent on this upgrade: {}\n\nAssign it to g1t-agent once agents can run here, or upgrade it by hand.",
393 refused.message
394 ))
395 .await?;
396 }
397 self.note(row, "update_needs_code", Some(g1t_contracts::system::USERNAME), Some(issue.number), Some(why)).await
398 }
399}
400
401#[cfg(test)]
402mod tests {
403 use super::*;
404
405 fn row(version: &str, fixed: &str) -> VulnRow {
406 VulnRow {
407 id: "vul_1".into(),
408 repo_id: "rep_1".into(),
409 ecosystem: "npm".into(),
410 package: "lodash".into(),
411 version: version.into(),
412 manifest: "web/package-lock.json".into(),
413 osv_id: "GHSA-35jh-r3h4-6jhm".into(),
414 advisory: "GHSA-35jh-r3h4-6jhm".into(),
415 summary: "Command Injection in lodash".into(),
416 severity: "high".into(),
417 fixed_version: Some(fixed.into()),
418 status: "open".into(),
419 found_at: "2026-10-04T00:00:00Z".into(),
420 fixed_at: None,
421 number: None,
422 dismiss_reason: None,
423 dismiss_comment: None,
424 dismissed_by: None,
425 dismissed_at: None,
426 }
427 }
428
429 #[test]
430 fn a_newer_fix_asks_again_and_the_same_one_waits() {
431 let retry = "2026-10-05T00:00:00Z";
432 let at = "2026-10-06T00:00:00Z";
433 assert_eq!(next_step(UpdateState::Open, "4.17.20", "4.17.21", at, retry), Next::Request);
434 assert_eq!(next_step(UpdateState::Open, "4.17.21", "4.17.21", at, retry), Next::Wait);
435 assert_eq!(next_step(UpdateState::Requested, "4.17.21", "4.17.21", at, retry), Next::Wait);
436 assert_eq!(next_step(UpdateState::Merged, "4.17.21", "4.17.21", at, retry), Next::Wait);
437 // A person closed it: left alone until a newer fix.
438 assert_eq!(next_step(UpdateState::Closed, "4.17.21", "4.17.21", at, retry), Next::Wait);
439 assert_eq!(next_step(UpdateState::Closed, "4.17.21", "4.17.22", at, retry), Next::Request);
440 // Vulnerable again after it was no longer needed.
441 assert_eq!(next_step(UpdateState::Superseded, "4.17.21", "4.17.21", at, retry), Next::Request);
442 // Failed: tried again a day later.
443 assert_eq!(next_step(UpdateState::Failed, "4.17.21", "4.17.21", at, retry), Next::Wait);
444 assert_eq!(next_step(UpdateState::Failed, "4.17.21", "4.17.21", "2026-10-04T00:00:00Z", retry), Next::Request);
445 assert_eq!(next_step(UpdateState::NeedsCode, "4.17.21", "4.17.21", at, retry), Next::Wait);
446 }
447
448 #[test]
449 fn the_pull_request_says_what_it_fixes_and_where() {
450 let a = row("4.17.20", "4.17.21");
451 let mut b = row("4.17.19", "4.17.21");
452 b.manifest = "package-lock.json".into();
453 let body = pull_text("npm", "lodash", "4.17.21", &[&a, &b]);
454 assert!(body.starts_with("Upgrades `lodash` (npm) from 4.17.19, 4.17.20 to **4.17.21**"));
455 assert!(body.contains("[GHSA-35jh-r3h4-6jhm](https://osv.dev/vulnerability/GHSA-35jh-r3h4-6jhm)"));
456 assert!(body.contains("Lockfiles changed: `package-lock.json`, `web/package-lock.json`."));
457 assert!(body.contains("required checks"));
458 assert_eq!(pull_title("lodash", "4.17.21"), "Upgrade lodash to 4.17.21");
459 assert_eq!(update_branch("@babel/core", "7.24.1"), "g1t/security/babel-core-7.24.1");
460 assert_eq!(update_branch("golang.org/x/net", "v0.23.0"), "g1t/security/golang.org-x-net-v0.23.0");
461 }
462}