Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/services/security/src/security_updates.rs

462 lines22,076 bytesCodeBlame
1//! 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}