Skip to content
Sign up

flagon-io/g1t

public
0
320 linesCodeBlameRaw
1//! The security digest: security alerts that are not urgent (medium and
2//! low) wait here for each person's next digest, once a day or once a week
3//! as they chose (`InboxSettings::security_digest`), and go out as one
4//! inbox item and one email per workspace: "Security digest for acme: 4
5//! new medium vulnerabilities in 2 repositories, 1 new low secret in 1
6//! repository". Critical and high alerts, refused pushes and vulnerability
7//! reports never wait here (inbox.rs decides, `security_delivery`).
8//!
9//! The cron (wrangler.jsonc, `DIGEST_CRON`) runs once a day in the morning,
10//! UTC; daily digests go out on every run, weekly ones on Monday's. Rows
11//! are deleted as their digest goes out, so a retried run sends nothing
12//! twice; a person whose cadence became "at once" or "off" after rows were
13//! queued gets what waited once, in a digest, rather than losing it.
14
15use std::collections::BTreeMap;
16
17use g1t_contracts::inbox::{DigestCadence, Reason, Severity};
18use g1t_contracts::new_id;
19use g1t_contracts::repos::RepoPath;
20use g1t_contracts::security::SeverityCounts;
21use g1t_contracts::security_suite::{AlertType, SecuritySummary};
22use g1t_contracts::time::rfc3339;
23use serde::Deserialize;
24use worker::wasm_bindgen::JsValue;
25use worker::{D1Database, D1PreparedStatement, Fetcher, Result};
26
27use crate::inbox::{self, Notice};
28use crate::subscriptions;
29
30/// The cron that sends digests: 08:00 UTC daily.
31pub const DIGEST_CRON: &str = "0 8 * * *";
32/// Rows read per run, at most.
33const BATCH: u32 = 5000;
34/// What the digest's inbox item records as its event type.
35const EVENT: &str = "security.digest";
36
37/// One scan summary waiting for a person's digest.
38#[derive(Clone, Debug, Deserialize, PartialEq)]
39pub struct Queued {
40 pub id: String,
41 pub username: String,
42 pub workspace: String,
43 pub repo_id: String,
44 pub repo: String,
45 pub alert_type: String,
46 pub opened: f64,
47 /// JSON `SeverityCounts`.
48 pub counts: String,
49 pub worst: Option<String>,
50 pub title: String,
51 pub link: String,
52}
53
54/// The statement that queues a summary for one person's digest.
55pub fn queue(db: &D1Database, username: &str, path: &RepoPath, summary: &SecuritySummary, event_id: &str, now_ms: u64) -> Result<D1PreparedStatement> {
56 db.prepare(
57 "INSERT OR IGNORE INTO inbox_security_digest
58 (id, username, workspace, repo_id, repo, alert_type, event_id, opened, counts, worst, title, link, created_at)
59 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
60 )
61 .bind(&[
62 new_id("dig", now_ms).into(),
63 username.to_lowercase().into(),
64 path.namespace.to_lowercase().into(),
65 summary.repo_id.as_str().into(),
66 format!("{}/{}", path.namespace, path.name).to_lowercase().into(),
67 summary.alert_type.as_str().into(),
68 event_id.into(),
69 summary.opened_count.into(),
70 serde_json::to_string(&summary.new_by_severity)?.into(),
71 summary.worst.as_deref().map_or(JsValue::NULL, JsValue::from),
72 summary.latest.as_deref().unwrap_or_default().into(),
73 summary.link.as_str().into(),
74 rfc3339(now_ms).into(),
75 ])
76}
77
78/// The day of the week at `now_ms`: 0 is Sunday. 1970-01-01 was a Thursday.
79pub fn weekday(now_ms: u64) -> u8 {
80 ((now_ms / 86_400_000 + 4) % 7) as u8
81}
82
83/// Whether a person's digest goes out on this run: daily ones every day,
84/// weekly ones on Monday; what waited for someone who since chose "at
85/// once" or "off" goes out too, once, rather than being lost.
86pub fn due(cadence: DigestCadence, weekday: u8) -> bool {
87 match cadence {
88 DigestCadence::Daily | DigestCadence::AtOnce | DigestCadence::Off => true,
89 DigestCadence::Weekly => weekday == 1,
90 }
91}
92
93/// "vulnerability" or "vulnerabilities".
94fn what(alert_type: &str, n: u32) -> &'static str {
95 match (AlertType::parse(alert_type), n) {
96 (Some(AlertType::SecretScanning), 1) => "secret",
97 (Some(AlertType::SecretScanning), _) => "secrets",
98 (Some(AlertType::CodeScanning), 1) => "code scanning alert",
99 (Some(AlertType::CodeScanning), _) => "code scanning alerts",
100 (_, 1) => "vulnerability",
101 _ => "vulnerabilities",
102 }
103}
104
105/// "in 2 repositories".
106fn in_repos(n: usize) -> String {
107 if n == 1 { "in 1 repository".to_owned() } else { format!("in {n} repositories") }
108}
109
110/// The digest's title and body for one workspace's rows: the title counts
111/// what is new per severity and alert type, worst first; the body names up
112/// to five of the alerts, newest first, each with its repository.
113pub fn digest_text(workspace: &str, rows: &[Queued]) -> (String, String) {
114 // (severity rank, alert type) -> (count, repositories)
115 let mut groups: BTreeMap<(u8, &str), (u32, std::collections::BTreeSet<&str>)> = BTreeMap::new();
116 for row in rows {
117 let counts: SeverityCounts = serde_json::from_str(&row.counts).unwrap_or_default();
118 for (rank, severity, n) in [(0u8, "critical", counts.critical), (1, "high", counts.high), (2, "medium", counts.medium), (3, "low", counts.low), (4, "", counts.unknown)] {
119 if n == 0 {
120 continue;
121 }
122 let _ = severity;
123 let group = groups.entry((rank, row.alert_type.as_str())).or_default();
124 group.0 += n;
125 group.1.insert(row.repo.as_str());
126 }
127 }
128 let parts: Vec<String> = groups
129 .iter()
130 .map(|((rank, alert_type), (n, repos))| {
131 let severity = match rank {
132 0 => "critical ",
133 1 => "high ",
134 2 => "medium ",
135 3 => "low ",
136 _ => "",
137 };
138 format!("{n} new {severity}{} {}", what(alert_type, *n), in_repos(repos.len()))
139 })
140 .collect();
141 let total: u32 = rows.iter().map(|row| row.opened as u32).sum();
142 let title = if parts.is_empty() {
143 format!("Security digest for {workspace}: {total} new security alerts")
144 } else {
145 format!("Security digest for {workspace}: {}", parts.join(", "))
146 };
147 let mut named: Vec<String> = rows.iter().rev().filter(|row| !row.title.is_empty()).take(5).map(|row| format!("{}: {}", row.repo, row.title)).collect();
148 if rows.len() > named.len() && !named.is_empty() {
149 named.push(format!("and {} more", rows.iter().filter(|row| !row.title.is_empty()).count().saturating_sub(named.len())));
150 }
151 (title, named.join(" · "))
152}
153
154/// Sends every digest that is due, and returns how many went out.
155pub async fn send_due(db: &D1Database, identity: &Fetcher, notify: Option<&Fetcher>, now_ms: u64) -> Result<u32> {
156 let rows = db
157 .prepare(
158 "SELECT id, username, workspace, repo_id, repo, alert_type, opened, counts, worst, title, link
159 FROM inbox_security_digest ORDER BY username, workspace, created_at LIMIT ?",
160 )
161 .bind(&[BATCH.into()])?
162 .all()
163 .await?
164 .results::<Queued>()?;
165 if rows.is_empty() {
166 return Ok(0);
167 }
168 let mut by_person: BTreeMap<String, Vec<Queued>> = BTreeMap::new();
169 for row in rows {
170 by_person.entry(row.username.clone()).or_default().push(row);
171 }
172 let people: Vec<String> = by_person.keys().cloned().collect();
173 let settings = subscriptions::settings_of(db, &people).await?;
174 let today = weekday(now_ms);
175 let now = rfc3339(now_ms);
176 let mut sent = 0;
177 for (username, rows) in by_person {
178 let setting = settings.get(&username).cloned().unwrap_or_default();
179 if !due(setting.security_digest, today) {
180 continue;
181 }
182 let mut by_workspace: BTreeMap<String, Vec<Queued>> = BTreeMap::new();
183 for row in rows {
184 by_workspace.entry(row.workspace.clone()).or_default().push(row);
185 }
186 for (workspace, rows) in by_workspace {
187 let (title, body) = digest_text(&workspace, &rows);
188 let notice = Notice { username: username.clone(), reason: Reason::SecurityAlert, severity: Severity::Info, title, body };
189 let thread = format!("workspace:{workspace}/security/digest");
190 let link = format!("/{workspace}/-/security");
191 let event_id = new_id("dig", now_ms);
192 let mut statements = vec![
193 db.prepare(inbox::BUMP).bind(&[
194 new_id("ntf", now_ms).into(),
195 username.as_str().into(),
196 thread.as_str().into(),
197 event_id.as_str().into(),
198 EVENT.into(),
199 notice.reason.as_str().into(),
200 notice.severity.as_str().into(),
201 notice.title.as_str().into(),
202 notice.body.as_str().into(),
203 workspace.as_str().into(),
204 JsValue::NULL,
205 JsValue::NULL,
206 JsValue::NULL,
207 JsValue::NULL,
208 JsValue::NULL,
209 link.as_str().into(),
210 JsValue::NULL,
211 now.as_str().into(),
212 new_id("ntf", now_ms).into(),
213 ])?,
214 db.prepare(
215 "INSERT OR IGNORE INTO inbox_activity (id, item_id, username, event_id, event, reason, severity, title, body, actor, created_at)
216 SELECT ?1, id, ?2, ?4, ?5, ?6, ?7, ?8, ?9, NULL, ?10 FROM inbox_items WHERE username = ?2 AND thread = ?3",
217 )
218 .bind(&[
219 new_id("ntf", now_ms).into(),
220 username.as_str().into(),
221 thread.as_str().into(),
222 event_id.as_str().into(),
223 EVENT.into(),
224 notice.reason.as_str().into(),
225 notice.severity.as_str().into(),
226 notice.title.as_str().into(),
227 notice.body.as_str().into(),
228 now.as_str().into(),
229 ])?,
230 ];
231 let ids: Vec<&str> = rows.iter().map(|row| row.id.as_str()).collect();
232 statements.push(
233 db.prepare("DELETE FROM inbox_security_digest WHERE id IN (SELECT value FROM json_each(?))")
234 .bind(&[serde_json::to_string(&ids)?.into()])?,
235 );
236 db.batch(statements).await?;
237 sent += 1;
238 if let Some(notify) = notify {
239 inbox::tell_live(notify, vec![(username.clone(), inbox::live_notification(&event_id, &notice, &link, &now))]).await;
240 inbox::tell_inbox_count(db, notify, &username).await;
241 }
242 // Emailed as a security alert, to someone who can still read the
243 // first repository named: identity checks that, as for any item.
244 if setting.email.contains(&Reason::SecurityAlert)
245 && let Some(first) = rows.first()
246 {
247 let sent_mail: Result<bool> = g1t_kit::call(
248 identity,
249 "notify_by_email",
250 &g1t_contracts::inbox::NotifyByEmailArgs {
251 username: username.clone(),
252 repo_id: first.repo_id.clone(),
253 subject: notice.title.clone(),
254 intro: notice.body.clone(),
255 quote: None,
256 path: link.clone(),
257 reason: Reason::SecurityAlert,
258 },
259 )
260 .await;
261 if let Err(error) = sent_mail {
262 worker::console_error!("security digest: email to {username} not sent: {error}");
263 }
264 }
265 }
266 }
267 Ok(sent)
268}
269
270#[cfg(test)]
271mod tests {
272 use super::*;
273
274 fn row(repo: &str, alert_type: &str, counts: &str, title: &str) -> Queued {
275 Queued {
276 id: "dig_1".into(),
277 username: "ana".into(),
278 workspace: "acme".into(),
279 repo_id: "rep_1".into(),
280 repo: repo.into(),
281 alert_type: alert_type.into(),
282 opened: 1.0,
283 counts: counts.into(),
284 worst: None,
285 title: title.into(),
286 link: "/acme/rocket/security/vulnerabilities".into(),
287 }
288 }
289
290 #[test]
291 fn the_digest_counts_what_is_new_worst_first_across_repositories() {
292 let rows = vec![
293 row("acme/rocket", "vulnerability", r#"{"critical":0,"high":0,"medium":3,"low":0,"unknown":0}"#, "lodash 4.17.20 in package-lock.json: GHSA-1"),
294 row("acme/probe", "vulnerability", r#"{"critical":0,"high":0,"medium":1,"low":0,"unknown":0}"#, "ms 2.0.0 in package-lock.json: GHSA-2"),
295 row("acme/rocket", "secret_scanning", r#"{"critical":0,"high":0,"medium":0,"low":1,"unknown":0}"#, "An example key in README.md"),
296 ];
297 let (title, body) = digest_text("acme", &rows);
298 assert_eq!(title, "Security digest for acme: 4 new medium vulnerabilities in 2 repositories, 1 new low secret in 1 repository");
299 assert!(body.starts_with("acme/rocket: An example key in README.md · acme/probe: ms 2.0.0"));
300 assert!(!body.contains("and "));
301 }
302
303 #[test]
304 fn weekly_digests_go_out_on_mondays_and_nothing_queued_is_lost() {
305 // 2026-10-12 is a Monday; 2026-10-11 a Sunday.
306 let monday = 1_791_763_200_000;
307 assert_eq!(weekday(monday), 1);
308 assert_eq!(weekday(monday - 86_400_000), 0);
309 assert!(due(DigestCadence::Weekly, 1) && !due(DigestCadence::Weekly, 0));
310 assert!(due(DigestCadence::Daily, 0) && due(DigestCadence::Off, 3) && due(DigestCadence::AtOnce, 5));
311 }
312
313 #[test]
314 fn a_digest_of_many_names_five_and_counts_the_rest() {
315 let rows: Vec<Queued> = (0..8).map(|n| row(&format!("acme/r{n}"), "code_scanning", r#"{"critical":0,"high":0,"medium":1,"low":0,"unknown":0}"#, &format!("Rule {n}"))).collect();
316 let (title, body) = digest_text("acme", &rows);
317 assert_eq!(title, "Security digest for acme: 8 new medium code scanning alerts in 8 repositories");
318 assert!(body.ends_with("and 3 more"));
319 }
320}