Skip to content
Sign up

flagon-io/g1t

public
00

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
1,407 linesCodeBlameRaw
1//! The security service's tables in D1.
2
3use std::collections::HashMap;
4
5use g1t_contracts::security::{
6 AlertActivity, DismissReason, NewSecret, ScanState, SecretCounts, SecretFinding, SecretStatus, SecurityUpdate,
7 SeverityCounts, UpdateState, VersionUpdatesState, VulnStatus, Vulnerability,
8};
9use g1t_contracts::time::rfc3339;
10use g1t_contracts::new_id;
11use g1t_kit::now_ms;
12use serde::Deserialize;
13use worker::wasm_bindgen::JsValue;
14use worker::{D1Database, Result};
15
16pub fn optional(value: Option<&str>) -> JsValue {
17 value.map_or(JsValue::NULL, JsValue::from)
18}
19
20pub fn now() -> String {
21 rfc3339(now_ms())
22}
23
24/// A repository purged (`repo.purged`): every row kept for it, `?1` its
25/// id. Advisories are shared by every repository, so they stay.
26pub const PURGED: &[&str] = &[
27 "DELETE FROM secrets WHERE repo_id = ?1",
28 "DELETE FROM vulnerabilities WHERE repo_id = ?1",
29 "DELETE FROM upgrades WHERE repo_id = ?1",
30 "DELETE FROM updates WHERE repo_id = ?1",
31 "DELETE FROM alert_activity WHERE repo_id = ?1",
32 "DELETE FROM push_scans WHERE repo_id = ?1",
33 "DELETE FROM repos WHERE repo_id = ?1",
34];
35
36#[derive(Clone, Deserialize)]
37pub struct RepoRow {
38 pub repo_id: String,
39 pub namespace: String,
40 pub name: String,
41 pub upkeep: i64,
42 pub history: String,
43 pub history_cursor: Option<String>,
44 pub history_commits: i64,
45 pub history_finished_at: Option<String>,
46 pub deps_scanned_at: Option<String>,
47 pub deps_error: Option<String>,
48 pub lockfiles: String,
49 #[serde(default)]
50 pub version_updates: Option<String>,
51}
52
53impl RepoRow {
54 pub fn scan_state(&self) -> ScanState {
55 ScanState {
56 history: self.history.clone(),
57 commits_scanned: self.history_commits.max(0) as u32,
58 history_finished_at: self.history_finished_at.clone(),
59 dependencies_scanned_at: self.deps_scanned_at.clone(),
60 dependencies_error: self.deps_error.clone(),
61 lockfiles: serde_json::from_str(&self.lockfiles).unwrap_or_default(),
62 }
63 }
64
65 /// What `.g1t/dependencies.yml` said when it was last read.
66 pub fn version_updates(&self) -> VersionUpdatesState {
67 self.version_updates
68 .as_deref()
69 .and_then(|json| serde_json::from_str(json).ok())
70 .unwrap_or_default()
71 }
72}
73
74#[derive(Deserialize)]
75struct SecretRow {
76 id: String,
77 repo_id: String,
78 kind: String,
79 path: String,
80 line: i64,
81 commit_hash: String,
82 preview: String,
83 status: String,
84 source: String,
85 found_by: Option<String>,
86 found_at: String,
87 decided_by: Option<String>,
88 reason: Option<String>,
89 decided_at: Option<String>,
90 dismiss_reason: Option<String>,
91 test_value: Option<String>,
92 // The security suite's (migration 0004).
93 #[serde(default)]
94 validity: Option<String>,
95 #[serde(default)]
96 validity_checked_at: Option<String>,
97 #[serde(default)]
98 bypass_reason: Option<String>,
99 #[serde(default)]
100 bypass_comment: Option<String>,
101 #[serde(default)]
102 bypassed_by: Option<String>,
103 #[serde(default)]
104 bypassed_at: Option<String>,
105 #[serde(default)]
106 bypass_approved_by: Option<String>,
107 #[serde(default)]
108 pattern_id: Option<String>,
109 #[serde(default)]
110 pattern_name: Option<String>,
111 #[serde(default)]
112 location_count: Option<i64>,
113}
114
115/// Secrets with how many places each was found.
116/// Rewrites the work items kept by the old issue numbers of their
117/// repositories to their numbers in the workspace, keeping each item's id
118/// and workspace with it, `?1` a page of `work.migrated`'s items as JSON.
119/// Each new number is above every old one of its workspace, so a page heard
120/// twice changes nothing more; a row already on its new number from before
121/// the ids were kept (migration 0009) gets them.
122pub const WORK_MIGRATED: [&str; 6] = [
123 "UPDATE updates SET item = m.number, item_id = m.item_id, item_workspace = m.workspace FROM moved_items m
124 WHERE updates.repo_id = m.repo_id AND (updates.item = m.legacy OR (updates.item = m.number AND updates.item_id IS NULL))",
125 "UPDATE update_pulls SET item = m.number, item_id = m.item_id, item_workspace = m.workspace FROM moved_items m
126 WHERE update_pulls.repo_id = m.repo_id AND (update_pulls.item = m.legacy OR (update_pulls.item = m.number AND update_pulls.item_id IS NULL))",
127 "UPDATE upgrades SET item = m.number, item_id = m.item_id, item_workspace = m.workspace FROM moved_items m
128 WHERE upgrades.repo_id = m.repo_id AND (upgrades.item = m.legacy OR (upgrades.item = m.number AND upgrades.item_id IS NULL))",
129 "UPDATE code_alerts SET item = m.number, item_id = m.item_id, item_workspace = m.workspace FROM moved_items m
130 WHERE code_alerts.repo_id = m.repo_id AND (code_alerts.item = m.legacy OR (code_alerts.item = m.number AND code_alerts.item_id IS NULL))",
131 "UPDATE alert_fixes SET item = m.number, item_id = m.item_id, item_workspace = m.workspace FROM moved_items m
132 WHERE alert_fixes.repo_id = m.repo_id AND (alert_fixes.item = m.legacy OR (alert_fixes.item = m.number AND alert_fixes.item_id IS NULL))",
133 // Only an update that needed code changes named a work item; the
134 // other steps name pull requests, which kept their numbers.
135 "UPDATE alert_activity SET number = m.number FROM moved_items m
136 WHERE alert_activity.repo_id = m.repo_id AND alert_activity.number = m.legacy AND alert_activity.action = 'update_needs_code'",
137];
138
139/// `work.migrated`'s items (`?1`), as the rows [`WORK_MIGRATED`] reads.
140const MOVED_ITEMS: &str = "WITH moved_items AS (
141 SELECT json_extract(value, '$.repoId') AS repo_id, json_extract(value, '$.legacyNumber') AS legacy,
142 json_extract(value, '$.number') AS number, json_extract(value, '$.itemId') AS item_id,
143 json_extract(value, '$.workspace') AS workspace
144 FROM json_each(?1))";
145
146/// The work item a row names: its number, and, from `work.migrated`, its
147/// id and the workspace it is in.
148#[derive(Clone, Debug, PartialEq, Eq)]
149pub struct LinkedItem {
150 pub number: u32,
151 pub item_id: Option<String>,
152 pub workspace: Option<String>,
153}
154
155impl LinkedItem {
156 pub fn new(number: Option<i64>, item_id: Option<&str>, workspace: Option<&str>) -> Option<Self> {
157 number.map(|n| LinkedItem { number: n.max(0) as u32, item_id: item_id.map(str::to_owned), workspace: workspace.map(str::to_owned) })
158 }
159
160 /// The workspace to find it in: the one it was kept with, else the
161 /// repository's now (`repo_workspace`), for a row from before.
162 pub fn workspace_or<'a>(&'a self, repo_workspace: &'a str) -> &'a str {
163 self.workspace.as_deref().unwrap_or(repo_workspace)
164 }
165
166 /// Whether the item found by its number is this one: with its id kept,
167 /// only the item of that id is.
168 pub fn is(&self, found_id: &str) -> bool {
169 self.item_id.as_deref().is_none_or(|id| id == found_id)
170 }
171}
172
173/// One of [`WORK_MIGRATED`], ready to run.
174pub fn work_migrated_sql(statement: &str) -> String {
175 format!("{MOVED_ITEMS}
176{statement}")
177}
178
179const SECRET_SELECT: &str =
180 "SELECT s.*, (SELECT count(*) FROM secret_locations l WHERE l.secret_id = s.id) AS location_count FROM secrets s";
181
182impl From<SecretRow> for SecretFinding {
183 fn from(row: SecretRow) -> Self {
184 let status = SecretStatus::parse(&row.status).unwrap_or(SecretStatus::Open);
185 let bypass = match (row.bypass_reason.as_deref().and_then(g1t_contracts::security_suite::BypassReason::parse), row.bypassed_by, row.bypassed_at) {
186 (Some(reason), Some(by), Some(at)) => Some(g1t_contracts::security_suite::Bypass {
187 reason,
188 comment: row.bypass_comment,
189 by,
190 at,
191 approved_by: row.bypass_approved_by,
192 }),
193 _ => None,
194 };
195 SecretFinding {
196 label: crate::secret_alerts::label_of(&row.kind, row.pattern_name.as_deref()),
197 validity: row.validity,
198 validity_checked_at: row.validity_checked_at,
199 bypass,
200 pattern_id: row.pattern_id,
201 pattern_name: row.pattern_name,
202 locations: row.location_count.unwrap_or(0).max(0) as u32,
203 id: row.id,
204 repo_id: row.repo_id,
205 kind: row.kind,
206 path: row.path,
207 line: row.line.max(0) as u32,
208 commit: row.commit_hash,
209 preview: row.preview,
210 status,
211 source: row.source,
212 found_by: row.found_by,
213 found_at: row.found_at,
214 decided_by: row.decided_by,
215 reason: row.reason,
216 decided_at: row.decided_at,
217 dismissed_reason: row.dismiss_reason.as_deref().and_then(DismissReason::parse),
218 test_value: row.test_value,
219 state: status.state(),
220 }
221 }
222}
223
224#[derive(Deserialize)]
225pub struct VulnRow {
226 pub id: String,
227 pub repo_id: String,
228 pub ecosystem: String,
229 pub package: String,
230 pub version: String,
231 pub manifest: String,
232 pub osv_id: String,
233 pub advisory: String,
234 pub summary: String,
235 pub severity: String,
236 pub fixed_version: Option<String>,
237 pub status: String,
238 pub found_at: String,
239 pub fixed_at: Option<String>,
240 pub item: Option<i64>,
241 #[serde(default)]
242 pub dismiss_reason: Option<String>,
243 #[serde(default)]
244 pub dismiss_comment: Option<String>,
245 #[serde(default)]
246 pub dismissed_by: Option<String>,
247 #[serde(default)]
248 pub dismissed_at: Option<String>,
249}
250
251impl From<VulnRow> for Vulnerability {
252 fn from(row: VulnRow) -> Self {
253 let status = VulnStatus::parse(&row.status).unwrap_or(VulnStatus::Open);
254 let dismissed = status == VulnStatus::Dismissed;
255 Vulnerability {
256 id: row.id,
257 repo_id: row.repo_id,
258 ecosystem: row.ecosystem,
259 package: row.package,
260 version: row.version,
261 manifest: row.manifest,
262 advisory: row.advisory,
263 osv_id: row.osv_id,
264 summary: row.summary,
265 severity: row.severity,
266 fixed_version: row.fixed_version,
267 status,
268 item: row.item.map(|number| number as u32),
269 found_at: row.found_at,
270 fixed_at: row.fixed_at,
271 state: status.state(),
272 dismissed_by: row.dismissed_by.filter(|_| dismissed),
273 dismissed_reason: row.dismiss_reason.as_deref().and_then(DismissReason::parse).filter(|_| dismissed),
274 dismissed_comment: row.dismiss_comment.filter(|_| dismissed),
275 dismissed_at: row.dismissed_at.filter(|_| dismissed),
276 update: None,
277 }
278 }
279}
280
281/// A security update, as stored.
282#[derive(Clone, Deserialize)]
283pub struct UpdateRow {
284 pub repo_id: String,
285 pub ecosystem: String,
286 pub package: String,
287 pub target: String,
288 pub state: String,
289 pub branch: Option<String>,
290 pub pull: Option<i64>,
291 pub item: Option<i64>,
292 #[serde(default)]
293 pub item_id: Option<String>,
294 #[serde(default)]
295 pub item_workspace: Option<String>,
296 pub error: Option<String>,
297 pub updated_at: String,
298}
299
300impl UpdateRow {
301 pub fn linked_item(&self) -> Option<LinkedItem> {
302 LinkedItem::new(self.item, self.item_id.as_deref(), self.item_workspace.as_deref())
303 }
304
305 pub fn state(&self) -> UpdateState {
306 UpdateState::parse(&self.state).unwrap_or(UpdateState::Failed)
307 }
308
309 pub fn pull(&self) -> Option<u32> {
310 self.pull.map(|n| n.max(0) as u32)
311 }
312
313 pub fn item(&self) -> Option<u32> {
314 self.item.map(|n| n.max(0) as u32)
315 }
316
317 pub fn to_contract(&self) -> SecurityUpdate {
318 SecurityUpdate {
319 state: self.state(),
320 target: self.target.clone(),
321 branch: self.branch.clone(),
322 pull: self.pull(),
323 item: self.item(),
324 error: self.error.clone(),
325 updated_at: self.updated_at.clone(),
326 }
327 }
328}
329
330#[derive(Deserialize)]
331struct ActivityRow {
332 id: String,
333 alert_id: String,
334 action: String,
335 actor: Option<String>,
336 reason: Option<String>,
337 comment: Option<String>,
338 number: Option<i64>,
339 at: String,
340}
341
342impl From<ActivityRow> for AlertActivity {
343 fn from(row: ActivityRow) -> Self {
344 AlertActivity {
345 id: row.id,
346 alert_id: row.alert_id,
347 action: row.action,
348 actor: row.actor,
349 reason: row.reason.as_deref().and_then(DismissReason::parse),
350 comment: row.comment,
351 number: row.number.map(|n| n.max(0) as u32),
352 at: row.at,
353 }
354 }
355}
356
357/// A push too large to scan before it was stored, scanned after it landed.
358#[derive(Clone, Deserialize)]
359pub struct PushScanRow {
360 pub id: String,
361 pub repo_id: String,
362 pub git_ref: String,
363 pub head: String,
364 pub base: Option<String>,
365 pub cursor: Option<String>,
366 pub pusher: Option<String>,
367 pub pages: i64,
368}
369
370/// One action on an alert, to record.
371pub struct Activity<'a> {
372 pub alert_id: &'a str,
373 pub action: &'a str,
374 pub actor: Option<&'a str>,
375 pub reason: Option<DismissReason>,
376 pub comment: Option<&'a str>,
377 pub number: Option<u32>,
378}
379
380#[derive(Deserialize)]
381pub struct UpgradeRow {
382 pub item: i64,
383 #[serde(default)]
384 pub item_id: Option<String>,
385 #[serde(default)]
386 pub item_workspace: Option<String>,
387}
388
389impl UpgradeRow {
390 pub fn linked_item(&self) -> Option<LinkedItem> {
391 LinkedItem::new(Some(self.item), self.item_id.as_deref(), self.item_workspace.as_deref())
392 }
393}
394
395#[derive(Deserialize)]
396struct CountRow {
397 severity: String,
398 n: i64,
399}
400
401#[derive(Deserialize)]
402struct SecretCountRow {
403 status: String,
404 test: i64,
405 n: i64,
406}
407
408#[derive(Deserialize)]
409struct NumberRow {
410 n: i64,
411}
412
413#[derive(Deserialize)]
414struct FingerprintRow {
415 fingerprint: String,
416 id: String,
417 status: String,
418}
419
420#[derive(Deserialize)]
421struct BodyRow {
422 body: String,
423 fetched_at: String,
424}
425
426pub struct Store {
427 pub db: D1Database,
428}
429
430const VULN_COLUMNS: &str = "v.id, v.repo_id, v.ecosystem, v.package, v.version, v.manifest, v.osv_id, v.advisory,
431 v.summary, v.severity, v.fixed_version, v.status, v.found_at, v.fixed_at, v.dismiss_reason, v.dismiss_comment,
432 v.dismissed_by, v.dismissed_at, COALESCE(n.item, u.item) AS item";
433
434/// The work item an upgrade has: the security update's, when it went to
435/// g1t, or one opened before security updates.
436const VULN_JOINS: &str = "LEFT JOIN upgrades u ON u.repo_id = v.repo_id AND u.ecosystem = v.ecosystem AND u.package = v.package
437 LEFT JOIN updates n ON n.repo_id = v.repo_id AND n.ecosystem = v.ecosystem AND n.package = v.package";
438
439impl Store {
440 pub async fn repo(&self, repo_id: &str) -> Result<Option<RepoRow>> {
441 self.db
442 .prepare("SELECT * FROM repos WHERE repo_id = ?")
443 .bind(&[repo_id.into()])?
444 .first::<RepoRow>(None)
445 .await
446 }
447
448 /// Records a repository the first time it is seen, and keeps its
449 /// address current. Returns its row.
450 pub async fn register(&self, repo_id: &str, namespace: &str, name: &str) -> Result<RepoRow> {
451 self.db
452 .prepare(
453 "INSERT INTO repos (repo_id, namespace, name, created_at) VALUES (?1, ?2, ?3, ?4)
454 ON CONFLICT (repo_id) DO UPDATE SET namespace = ?2, name = ?3",
455 )
456 .bind(&[repo_id.into(), namespace.into(), name.into(), now().into()])?
457 .run()
458 .await?;
459 self.repo(repo_id)
460 .await?
461 .ok_or_else(|| worker::Error::RustError("the repository was not recorded".into()))
462 }
463
464 /// A repository's path changed: transferred or renamed.
465 pub async fn moved(&self, repo_id: &str, namespace: &str, name: &str) -> Result<()> {
466 self.db
467 .prepare("UPDATE repos SET namespace = ?, name = ? WHERE repo_id = ?")
468 .bind(&[namespace.into(), name.into(), repo_id.into()])?
469 .run()
470 .await?;
471 Ok(())
472 }
473
474 pub async fn rename_namespace(&self, stale: &[String], current: &str) -> Result<()> {
475 for slug in stale {
476 self.db
477 .prepare("UPDATE repos SET namespace = ? WHERE namespace = ?")
478 .bind(&[current.into(), slug.as_str().into()])?
479 .run()
480 .await?;
481 }
482 Ok(())
483 }
484
485 pub async fn in_namespace(&self, namespace: &str) -> Result<Vec<RepoRow>> {
486 self.db
487 .prepare("SELECT * FROM repos WHERE namespace = ? ORDER BY name")
488 .bind(&[namespace.into()])?
489 .all()
490 .await?
491 .results::<RepoRow>()
492 }
493
494 /// A repository purged: everything found in it goes. `?1` its id.
495 pub async fn purge(&self, repo_id: &str) -> Result<()> {
496 let mut batch = Vec::with_capacity(PURGED.len());
497 for sql in PURGED.iter().chain(crate::update_store::PURGED) {
498 batch.push(self.db.prepare(*sql).bind(&[repo_id.into()])?);
499 }
500 self.db.batch(batch).await?;
501 Ok(())
502 }
503
504 /// `work.migrated`: what was kept by old issue numbers names the work
505 /// items they became.
506 pub async fn work_migrated(&self, items: &[g1t_contracts::work::MigratedItem]) -> Result<()> {
507 if items.is_empty() {
508 return Ok(());
509 }
510 let page = serde_json::to_string(items)?;
511 let mut batch = Vec::with_capacity(WORK_MIGRATED.len());
512 for sql in WORK_MIGRATED {
513 batch.push(self.db.prepare(work_migrated_sql(sql)).bind(&[page.as_str().into()])?);
514 }
515 self.db.batch(batch).await?;
516 Ok(())
517 }
518
519 /// Records that a repository's daily dependency read was skipped, and
520 /// why, so the sweep moves on to others until the next day.
521 pub async fn skip_dependencies(&self, repo_id: &str, why: &str) -> Result<()> {
522 self.db
523 .prepare("UPDATE repos SET deps_scanned_at = ?, deps_error = ? WHERE repo_id = ?")
524 .bind(&[now().into(), why.into(), repo_id.into()])?
525 .run()
526 .await?;
527 Ok(())
528 }
529
530 /// Repositories whose history still has to be scanned, oldest first.
531 pub async fn unfinished_histories(&self, limit: u32) -> Result<Vec<RepoRow>> {
532 self.db
533 .prepare("SELECT * FROM repos WHERE history IN ('pending', 'running', 'stopped') ORDER BY created_at LIMIT ?")
534 .bind(&[limit.into()])?
535 .all()
536 .await?
537 .results::<RepoRow>()
538 }
539
540 /// Repositories whose dependencies were last read before `before`.
541 pub async fn stale_dependencies(&self, before: &str, limit: u32) -> Result<Vec<RepoRow>> {
542 self.db
543 .prepare(
544 "SELECT * FROM repos WHERE deps_scanned_at IS NULL OR deps_scanned_at < ?
545 ORDER BY deps_scanned_at LIMIT ?",
546 )
547 .bind(&[before.into(), limit.into()])?
548 .all()
549 .await?
550 .results::<RepoRow>()
551 }
552
553 pub async fn set_history(&self, repo_id: &str, state: &str, cursor: Option<&str>, commits: u32) -> Result<()> {
554 let finished = (state == "done").then(now);
555 self.db
556 .prepare(
557 "UPDATE repos SET history = ?, history_cursor = ?, history_commits = history_commits + ?,
558 history_finished_at = COALESCE(?, history_finished_at)
559 WHERE repo_id = ?",
560 )
561 .bind(&[state.into(), optional(cursor), commits.into(), optional(finished.as_deref()), repo_id.into()])?
562 .run()
563 .await?;
564 Ok(())
565 }
566
567 pub async fn restart_history(&self, repo_id: &str) -> Result<()> {
568 self.db
569 .prepare(
570 "UPDATE repos SET history = 'pending', history_cursor = NULL, history_commits = 0,
571 history_finished_at = NULL WHERE repo_id = ?",
572 )
573 .bind(&[repo_id.into()])?
574 .run()
575 .await?;
576 Ok(())
577 }
578
579 pub async fn set_dependencies_scanned(
580 &self,
581 repo_id: &str,
582 commit: Option<&str>,
583 lockfiles: &[String],
584 error: Option<&str>,
585 ) -> Result<()> {
586 self.db
587 .prepare("UPDATE repos SET deps_scanned_at = ?, deps_commit = ?, lockfiles = ?, deps_error = ? WHERE repo_id = ?")
588 .bind(&[now().into(), optional(commit), serde_json::to_string(lockfiles)?.into(), optional(error), repo_id.into()])?
589 .run()
590 .await?;
591 Ok(())
592 }
593
594 pub async fn set_version_updates(&self, repo_id: &str, state: &VersionUpdatesState) -> Result<()> {
595 self.db
596 .prepare("UPDATE repos SET version_updates = ? WHERE repo_id = ?")
597 .bind(&[serde_json::to_string(state)?.into(), repo_id.into()])?
598 .run()
599 .await?;
600 Ok(())
601 }
602
603 pub async fn set_upkeep(&self, repo_id: &str, enabled: bool, by: &str) -> Result<()> {
604 self.db
605 .prepare("UPDATE repos SET upkeep = ?, upkeep_by = ?, upkeep_at = ? WHERE repo_id = ?")
606 .bind(&[i32::from(enabled).into(), by.into(), now().into(), repo_id.into()])?
607 .run()
608 .await?;
609 Ok(())
610 }
611
612 // --- Secrets ------------------------------------------------------------
613
614 pub async fn secrets(&self, repo_id: &str) -> Result<Vec<SecretFinding>> {
615 let rows = self
616 .db
617 .prepare(format!(
618 "{SECRET_SELECT} WHERE s.repo_id = ?
619 ORDER BY CASE status WHEN 'open' THEN 0 WHEN 'blocked' THEN 1 WHEN 'allowed' THEN 2 ELSE 3 END,
620 test_value IS NOT NULL, found_at DESC LIMIT 500"
621 ))
622 .bind(&[repo_id.into()])?
623 .all()
624 .await?
625 .results::<SecretRow>()?;
626 Ok(rows.into_iter().map(SecretFinding::from).collect())
627 }
628
629 pub async fn secret(&self, repo_id: &str, id: &str) -> Result<Option<SecretFinding>> {
630 Ok(self
631 .db
632 .prepare(format!("{SECRET_SELECT} WHERE s.repo_id = ? AND s.id = ?"))
633 .bind(&[repo_id.into(), id.into()])?
634 .first::<SecretRow>(None)
635 .await?
636 .map(SecretFinding::from))
637 }
638
639 /// The findings already known for these fingerprints: fingerprint, id
640 /// and status.
641 pub async fn known(&self, repo_id: &str, fingerprints: &[String]) -> Result<Vec<(String, String, String)>> {
642 let mut known = Vec::new();
643 for chunk in fingerprints.chunks(90) {
644 let marks = vec!["?"; chunk.len()].join(", ");
645 let mut binds: Vec<JsValue> = vec![repo_id.into()];
646 binds.extend(chunk.iter().map(|fingerprint| JsValue::from(fingerprint.as_str())));
647 let rows = self
648 .db
649 .prepare(format!("SELECT fingerprint, id, status FROM secrets WHERE repo_id = ? AND fingerprint IN ({marks})"))
650 .bind(&binds)?
651 .all()
652 .await?
653 .results::<FingerprintRow>()?;
654 known.extend(rows.into_iter().map(|row| (row.fingerprint, row.id, row.status)));
655 }
656 Ok(known)
657 }
658
659 /// Records secrets not seen before in this repository, with `status`.
660 /// A secret already known keeps its record and its decision.
661 pub async fn add_secrets(
662 &self,
663 repo_id: &str,
664 secrets: &[NewSecret],
665 status: SecretStatus,
666 source: &str,
667 found_by: Option<&str>,
668 ) -> Result<()> {
669 if secrets.is_empty() {
670 return Ok(());
671 }
672 let now_ms = now_ms();
673 let found_at = rfc3339(now_ms);
674 let statements = secrets
675 .iter()
676 .map(|secret| {
677 self.db
678 .prepare(
679 "INSERT INTO secrets (id, repo_id, fingerprint, kind, path, line, commit_hash, preview,
680 status, source, found_by, found_at, test_value)
681 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
682 ON CONFLICT (repo_id, fingerprint) DO UPDATE SET test_value = excluded.test_value",
683 )
684 .bind(&[
685 new_id("sec", now_ms).into(),
686 repo_id.into(),
687 secret.fingerprint.as_str().into(),
688 secret.kind.as_str().into(),
689 secret.path.as_str().into(),
690 secret.line.into(),
691 secret.commit.as_str().into(),
692 secret.preview.as_str().into(),
693 status.as_str().into(),
694 source.into(),
695 optional(found_by),
696 found_at.as_str().into(),
697 optional(secret.test_value.as_deref()),
698 ])
699 })
700 .collect::<Result<Vec<_>>>()?;
701 self.db.batch(statements).await?;
702 Ok(())
703 }
704
705 /// A secret found in history that was only ever blocked has landed
706 /// after all (it was allowed, then pushed): it is open now unless
707 /// someone allowed it.
708 pub async fn landed(&self, repo_id: &str, fingerprints: &[String]) -> Result<()> {
709 for chunk in fingerprints.chunks(90) {
710 let marks = vec!["?"; chunk.len()].join(", ");
711 let mut binds: Vec<JsValue> = vec![repo_id.into()];
712 binds.extend(chunk.iter().map(|fingerprint| JsValue::from(fingerprint.as_str())));
713 self.db
714 .prepare(format!(
715 "UPDATE secrets SET status = 'open', source = 'history'
716 WHERE repo_id = ? AND status = 'blocked' AND fingerprint IN ({marks})"
717 ))
718 .bind(&binds)?
719 .run()
720 .await?;
721 }
722 Ok(())
723 }
724
725 /// Dismisses a secret: allowed, or resolved when it was revoked.
726 pub async fn dismiss_secret(&self, repo_id: &str, id: &str, reason: DismissReason, by: &str, comment: Option<&str>) -> Result<()> {
727 self.db
728 .prepare(
729 "UPDATE secrets SET status = ?, decided_by = ?, reason = ?, decided_at = ?, dismiss_reason = ?
730 WHERE repo_id = ? AND id = ?",
731 )
732 .bind(&[
733 reason.secret_status().as_str().into(),
734 by.into(),
735 optional(comment),
736 now().into(),
737 reason.as_str().into(),
738 repo_id.into(),
739 id.into(),
740 ])?
741 .run()
742 .await?;
743 Ok(())
744 }
745
746 /// Opens a secret again, as `status` (open, or blocked for one that
747 /// never landed).
748 pub async fn reopen_secret(&self, repo_id: &str, id: &str, status: SecretStatus) -> Result<()> {
749 self.db
750 .prepare(
751 "UPDATE secrets SET status = ?, decided_by = NULL, reason = NULL, decided_at = NULL, dismiss_reason = NULL
752 WHERE repo_id = ? AND id = ?",
753 )
754 .bind(&[status.as_str().into(), repo_id.into(), id.into()])?
755 .run()
756 .await?;
757 Ok(())
758 }
759
760 // --- Vulnerabilities ----------------------------------------------------
761
762 pub async fn vulnerabilities(&self, repo_id: &str) -> Result<Vec<Vulnerability>> {
763 let rows = self
764 .db
765 .prepare(format!(
766 "SELECT {VULN_COLUMNS} FROM vulnerabilities v
767 {VULN_JOINS}
768 WHERE v.repo_id = ?
769 ORDER BY CASE v.status WHEN 'open' THEN 0 WHEN 'dismissed' THEN 1 ELSE 2 END,
770 CASE v.severity WHEN 'critical' THEN 0 WHEN 'high' THEN 1 WHEN 'medium' THEN 2 WHEN 'low' THEN 3 ELSE 4 END,
771 v.package LIMIT 1000"
772 ))
773 .bind(&[repo_id.into()])?
774 .all()
775 .await?
776 .results::<VulnRow>()?;
777 let updates: HashMap<(String, String), UpdateRow> = self
778 .updates(repo_id)
779 .await?
780 .into_iter()
781 .map(|row| ((row.ecosystem.clone(), row.package.clone()), row))
782 .collect();
783 Ok(rows
784 .into_iter()
785 .map(|row| {
786 let update = updates.get(&(row.ecosystem.clone(), row.package.clone())).map(UpdateRow::to_contract);
787 Vulnerability { update, ..Vulnerability::from(row) }
788 })
789 .collect())
790 }
791
792 pub async fn vulnerability(&self, repo_id: &str, id: &str) -> Result<Option<Vulnerability>> {
793 let row = self
794 .db
795 .prepare(format!("SELECT {VULN_COLUMNS} FROM vulnerabilities v {VULN_JOINS} WHERE v.repo_id = ? AND v.id = ?"))
796 .bind(&[repo_id.into(), id.into()])?
797 .first::<VulnRow>(None)
798 .await?;
799 let Some(row) = row else { return Ok(None) };
800 let update = self.update(repo_id, &row.ecosystem, &row.package).await?.map(|row| row.to_contract());
801 Ok(Some(Vulnerability { update, ..Vulnerability::from(row) }))
802 }
803
804 pub async fn dismiss_vulnerability(
805 &self,
806 repo_id: &str,
807 id: &str,
808 reason: DismissReason,
809 by: &str,
810 comment: Option<&str>,
811 ) -> Result<()> {
812 self.db
813 .prepare(
814 "UPDATE vulnerabilities SET status = CASE status WHEN 'fixed' THEN 'fixed' ELSE 'dismissed' END,
815 dismiss_reason = ?, dismiss_comment = ?, dismissed_by = ?, dismissed_at = ?
816 WHERE repo_id = ? AND id = ?",
817 )
818 .bind(&[reason.as_str().into(), optional(comment), by.into(), now().into(), repo_id.into(), id.into()])?
819 .run()
820 .await?;
821 Ok(())
822 }
823
824 pub async fn reopen_vulnerability(&self, repo_id: &str, id: &str) -> Result<()> {
825 self.db
826 .prepare(
827 "UPDATE vulnerabilities SET status = CASE status WHEN 'dismissed' THEN 'open' ELSE status END,
828 dismiss_reason = NULL, dismiss_comment = NULL, dismissed_by = NULL, dismissed_at = NULL
829 WHERE repo_id = ? AND id = ?",
830 )
831 .bind(&[repo_id.into(), id.into()])?
832 .run()
833 .await?;
834 Ok(())
835 }
836
837 /// The ids of a package's open vulnerabilities.
838 pub async fn open_ids(&self, repo_id: &str, ecosystem: &str, package: &str) -> Result<Vec<String>> {
839 #[derive(Deserialize)]
840 struct IdRow {
841 id: String,
842 }
843 Ok(self
844 .db
845 .prepare("SELECT id FROM vulnerabilities WHERE repo_id = ? AND ecosystem = ? AND package = ? AND status = 'open'")
846 .bind(&[repo_id.into(), ecosystem.into(), package.into()])?
847 .all()
848 .await?
849 .results::<IdRow>()?
850 .into_iter()
851 .map(|row| row.id)
852 .collect())
853 }
854
855 pub async fn open_vulnerabilities(&self, repo_id: &str) -> Result<Vec<VulnRow>> {
856 self.db
857 .prepare(format!(
858 "SELECT {VULN_COLUMNS} FROM vulnerabilities v
859 {VULN_JOINS}
860 WHERE v.repo_id = ? AND v.status = 'open'"
861 ))
862 .bind(&[repo_id.into()])?
863 .all()
864 .await?
865 .results::<VulnRow>()
866 }
867
868 /// Every vulnerability alert of a repository: open, fixed and dismissed.
869 pub async fn all_vulnerabilities(&self, repo_id: &str) -> Result<Vec<VulnRow>> {
870 self.db
871 .prepare(format!("SELECT {VULN_COLUMNS} FROM vulnerabilities v {VULN_JOINS} WHERE v.repo_id = ?"))
872 .bind(&[repo_id.into()])?
873 .all()
874 .await?
875 .results::<VulnRow>()
876 }
877
878 /// Replaces what is known about a repository's dependencies with what a
879 /// scan found: new findings open, findings no longer true fixed, and
880 /// findings that came back open again, or dismissed again when someone
881 /// had dismissed them.
882 pub async fn replace_vulnerabilities(&self, repo_id: &str, found: &[Vulnerability]) -> Result<()> {
883 let now_ms = now_ms();
884 let now = rfc3339(now_ms);
885 let mut statements = vec![self
886 .db
887 .prepare(
888 "UPDATE vulnerabilities SET status = 'fixed', fixed_at = ? WHERE repo_id = ? AND status IN ('open', 'dismissed')",
889 )
890 .bind(&[now.as_str().into(), repo_id.into()])?];
891 for vuln in found {
892 statements.push(
893 self.db
894 .prepare(
895 "INSERT INTO vulnerabilities (id, repo_id, ecosystem, package, version, manifest, osv_id, advisory,
896 summary, severity, fixed_version, status, found_at)
897 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'open', ?)
898 ON CONFLICT (repo_id, ecosystem, package, version, manifest, osv_id) DO UPDATE SET
899 status = CASE WHEN vulnerabilities.dismiss_reason IS NULL THEN 'open' ELSE 'dismissed' END,
900 fixed_at = NULL, advisory = excluded.advisory, summary = excluded.summary,
901 severity = excluded.severity, fixed_version = excluded.fixed_version",
902 )
903 .bind(&[
904 new_id("vul", now_ms).into(),
905 repo_id.into(),
906 vuln.ecosystem.as_str().into(),
907 vuln.package.as_str().into(),
908 vuln.version.as_str().into(),
909 vuln.manifest.as_str().into(),
910 vuln.osv_id.as_str().into(),
911 vuln.advisory.as_str().into(),
912 vuln.summary.as_str().into(),
913 vuln.severity.as_str().into(),
914 optional(vuln.fixed_version.as_deref()),
915 now.as_str().into(),
916 ])?,
917 );
918 }
919 // D1 runs a batch as one transaction, so readers never see the
920 // moment between marking everything fixed and opening it again.
921 self.db.batch(statements).await?;
922 Ok(())
923 }
924
925 pub async fn counts(&self, repo_id: &str) -> Result<(SeverityCounts, u32, u32)> {
926 let rows = self
927 .db
928 .prepare("SELECT severity, count(*) AS n FROM vulnerabilities WHERE repo_id = ? AND status = 'open' GROUP BY severity")
929 .bind(&[repo_id.into()])?
930 .all()
931 .await?
932 .results::<CountRow>()?;
933 let mut counts = SeverityCounts::default();
934 let mut vulns = 0;
935 for row in rows {
936 let n = row.n.max(0) as u32;
937 vulns += n;
938 match row.severity.as_str() {
939 "critical" => counts.critical += n,
940 "high" => counts.high += n,
941 "medium" => counts.medium += n,
942 "low" => counts.low += n,
943 _ => counts.unknown += n,
944 }
945 }
946 let secrets = self.secret_counts(repo_id).await?;
947 // A secret in the history that looks real is the worst thing a
948 // repository can hold. One stopped at a push never landed, and a
949 // likely test value is no danger: neither is counted.
950 counts.critical += secrets.open;
951 Ok((counts, secrets.open + secrets.blocked, vulns))
952 }
953
954 pub async fn secret_counts(&self, repo_id: &str) -> Result<SecretCounts> {
955 let rows = self
956 .db
957 .prepare(
958 "SELECT status, (test_value IS NOT NULL) AS test, count(*) AS n FROM secrets WHERE repo_id = ?
959 GROUP BY status, test",
960 )
961 .bind(&[repo_id.into()])?
962 .all()
963 .await?
964 .results::<SecretCountRow>()?;
965 Ok(secret_counts(rows.iter().map(|row| (row.status.as_str(), row.test != 0, row.n.max(0) as u32))))
966 }
967
968 // --- Activity -----------------------------------------------------------
969
970 pub async fn record(&self, repo_id: &str, activity: &[Activity<'_>]) -> Result<()> {
971 if activity.is_empty() {
972 return Ok(());
973 }
974 let now_ms = now_ms();
975 let at = rfc3339(now_ms);
976 let statements = activity
977 .iter()
978 .map(|item| {
979 self.db
980 .prepare(
981 "INSERT INTO alert_activity (id, repo_id, alert_id, action, actor, reason, comment, number, at)
982 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
983 )
984 .bind(&[
985 new_id("act", now_ms).into(),
986 repo_id.into(),
987 item.alert_id.into(),
988 item.action.into(),
989 optional(item.actor),
990 optional(item.reason.map(DismissReason::as_str)),
991 optional(item.comment),
992 item.number.map_or(JsValue::NULL, JsValue::from),
993 at.as_str().into(),
994 ])
995 })
996 .collect::<Result<Vec<_>>>()?;
997 self.db.batch(statements).await?;
998 Ok(())
999 }
1000
1001 /// Each of `alert_ids`' latest action and comment, for the alerts that
1002 /// have any.
1003 pub async fn latest_activity(&self, repo_id: &str, alert_ids: &[String]) -> Result<HashMap<String, (String, Option<String>)>> {
1004 #[derive(Deserialize)]
1005 struct Latest {
1006 alert_id: String,
1007 action: String,
1008 comment: Option<String>,
1009 }
1010 if alert_ids.is_empty() {
1011 return Ok(HashMap::new());
1012 }
1013 let marks = vec!["?"; alert_ids.len()].join(", ");
1014 let mut binds: Vec<JsValue> = vec![repo_id.into()];
1015 binds.extend(alert_ids.iter().map(|id| JsValue::from(id.as_str())));
1016 let rows = self
1017 .db
1018 .prepare(format!(
1019 "SELECT alert_id, action, comment FROM alert_activity WHERE repo_id = ? AND alert_id IN ({marks})
1020 ORDER BY at DESC, id DESC"
1021 ))
1022 .bind(&binds)?
1023 .all()
1024 .await?
1025 .results::<Latest>()?;
1026 let mut latest = HashMap::new();
1027 for row in rows {
1028 latest.entry(row.alert_id).or_insert((row.action, row.comment));
1029 }
1030 Ok(latest)
1031 }
1032
1033 /// What happened to a repository's alerts, newest first.
1034 pub async fn activity(&self, repo_id: &str, limit: u32) -> Result<Vec<AlertActivity>> {
1035 Ok(self
1036 .db
1037 .prepare("SELECT * FROM alert_activity WHERE repo_id = ? ORDER BY at DESC, id DESC LIMIT ?")
1038 .bind(&[repo_id.into(), limit.into()])?
1039 .all()
1040 .await?
1041 .results::<ActivityRow>()?
1042 .into_iter()
1043 .map(AlertActivity::from)
1044 .collect())
1045 }
1046
1047 // --- Security updates ---------------------------------------------------
1048
1049 pub async fn updates(&self, repo_id: &str) -> Result<Vec<UpdateRow>> {
1050 self.db
1051 .prepare("SELECT * FROM updates WHERE repo_id = ?")
1052 .bind(&[repo_id.into()])?
1053 .all()
1054 .await?
1055 .results::<UpdateRow>()
1056 }
1057
1058 pub async fn update(&self, repo_id: &str, ecosystem: &str, package: &str) -> Result<Option<UpdateRow>> {
1059 self.db
1060 .prepare("SELECT * FROM updates WHERE repo_id = ? AND ecosystem = ? AND package = ?")
1061 .bind(&[repo_id.into(), ecosystem.into(), package.into()])?
1062 .first::<UpdateRow>(None)
1063 .await
1064 }
1065
1066 pub async fn update_by_branch(&self, repo_id: &str, branch: &str) -> Result<Option<UpdateRow>> {
1067 self.db
1068 .prepare("SELECT * FROM updates WHERE repo_id = ? AND branch = ?")
1069 .bind(&[repo_id.into(), branch.into()])?
1070 .first::<UpdateRow>(None)
1071 .await
1072 }
1073
1074 pub async fn update_by_pull(&self, repo_id: &str, pull: u32) -> Result<Option<UpdateRow>> {
1075 self.db
1076 .prepare("SELECT * FROM updates WHERE repo_id = ? AND pull = ?")
1077 .bind(&[repo_id.into(), pull.into()])?
1078 .first::<UpdateRow>(None)
1079 .await
1080 }
1081
1082 /// Updates asked of a sandbox before `before` that never pushed.
1083 /// Updates left in `needs_code`, which g1t opened a work item for before
1084 /// it stopped doing that.
1085 pub async fn needs_code_updates(&self, limit: u32) -> Result<Vec<UpdateRow>> {
1086 self.db
1087 .prepare("SELECT * FROM updates WHERE state = 'needs_code' ORDER BY updated_at LIMIT ?")
1088 .bind(&[limit.into()])?
1089 .all()
1090 .await?
1091 .results::<UpdateRow>()
1092 }
1093
1094 pub async fn stalled_updates(&self, before: &str, limit: u32) -> Result<Vec<UpdateRow>> {
1095 self.db
1096 .prepare("SELECT * FROM updates WHERE state = 'requested' AND updated_at < ? ORDER BY updated_at LIMIT ?")
1097 .bind(&[before.into(), limit.into()])?
1098 .all()
1099 .await?
1100 .results::<UpdateRow>()
1101 }
1102
1103 /// Asks for a security update: a sandbox is making the change on
1104 /// `branch`. An open pull request for an older version is kept until
1105 /// the new one opens, which supersedes it.
1106 pub async fn request_update(&self, repo_id: &str, ecosystem: &str, package: &str, target: &str, branch: &str) -> Result<()> {
1107 let now = now();
1108 self.db
1109 .prepare(
1110 "INSERT INTO updates (repo_id, ecosystem, package, target, state, branch, requested_at, updated_at)
1111 VALUES (?1, ?2, ?3, ?4, 'requested', ?5, ?6, ?6)
1112 ON CONFLICT (repo_id, ecosystem, package) DO UPDATE SET
1113 target = ?4, state = 'requested', branch = ?5, error = NULL, item = NULL,
1114 pull = CASE WHEN updates.state = 'open' THEN updates.pull ELSE NULL END,
1115 requested_at = ?6, updated_at = ?6",
1116 )
1117 .bind(&[repo_id.into(), ecosystem.into(), package.into(), target.into(), branch.into(), now.into()])?
1118 .run()
1119 .await?;
1120 Ok(())
1121 }
1122
1123 pub async fn set_update(
1124 &self,
1125 row: &UpdateRow,
1126 state: UpdateState,
1127 pull: Option<u32>,
1128 item: Option<u32>,
1129 error: Option<&str>,
1130 ) -> Result<()> {
1131 self.db
1132 .prepare(
1133 "UPDATE updates SET state = ?, pull = ?, item = ?, error = ?, updated_at = ?
1134 WHERE repo_id = ? AND ecosystem = ? AND package = ?",
1135 )
1136 .bind(&[
1137 state.as_str().into(),
1138 pull.map_or(JsValue::NULL, JsValue::from),
1139 item.map_or(JsValue::NULL, JsValue::from),
1140 optional(error),
1141 now().into(),
1142 row.repo_id.as_str().into(),
1143 row.ecosystem.as_str().into(),
1144 row.package.as_str().into(),
1145 ])?
1146 .run()
1147 .await?;
1148 Ok(())
1149 }
1150
1151 // --- Pushes scanned after they landed -----------------------------------
1152
1153 pub async fn add_push_scan(&self, repo_id: &str, git_ref: &str, head: &str, base: Option<&str>, pusher: Option<&str>) -> Result<String> {
1154 let id = new_id("psc", now_ms());
1155 self.db
1156 .prepare(
1157 "INSERT INTO push_scans (id, repo_id, git_ref, head, base, pusher, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
1158 )
1159 .bind(&[id.as_str().into(), repo_id.into(), git_ref.into(), head.into(), optional(base), optional(pusher), now().into()])?
1160 .run()
1161 .await?;
1162 Ok(id)
1163 }
1164
1165 pub async fn push_scan(&self, id: &str) -> Result<Option<PushScanRow>> {
1166 self.db
1167 .prepare("SELECT * FROM push_scans WHERE id = ?")
1168 .bind(&[id.into()])?
1169 .first::<PushScanRow>(None)
1170 .await
1171 }
1172
1173 pub async fn pending_push_scans(&self, limit: u32) -> Result<Vec<PushScanRow>> {
1174 self.db
1175 .prepare("SELECT * FROM push_scans WHERE state = 'pending' ORDER BY created_at LIMIT ?")
1176 .bind(&[limit.into()])?
1177 .all()
1178 .await?
1179 .results::<PushScanRow>()
1180 }
1181
1182 /// Where a push's scan stands after a page: its next page, or done.
1183 pub async fn advance_push_scan(&self, id: &str, next: Option<&str>, commits: u32, found: u32) -> Result<()> {
1184 let done = next.is_none();
1185 self.db
1186 .prepare(
1187 "UPDATE push_scans SET cursor = ?, state = ?, commits = commits + ?, pages = pages + 1, found = found + ?,
1188 finished_at = CASE WHEN ? THEN ? ELSE finished_at END
1189 WHERE id = ?",
1190 )
1191 .bind(&[
1192 optional(next),
1193 if done { "done" } else { "pending" }.into(),
1194 commits.into(),
1195 found.into(),
1196 done.into(),
1197 now().into(),
1198 id.into(),
1199 ])?
1200 .run()
1201 .await?;
1202 Ok(())
1203 }
1204
1205 // --- Upgrades -----------------------------------------------------------
1206
1207 /// The work item opened for a package before security updates, if any.
1208 pub async fn upgrade(&self, repo_id: &str, ecosystem: &str, package: &str) -> Result<Option<UpgradeRow>> {
1209 self.db
1210 .prepare("SELECT item, item_id, item_workspace FROM upgrades WHERE repo_id = ? AND ecosystem = ? AND package = ?")
1211 .bind(&[repo_id.into(), ecosystem.into(), package.into()])?
1212 .first::<UpgradeRow>(None)
1213 .await
1214 }
1215
1216 // --- Advisories and usage -----------------------------------------------
1217
1218 /// OSV's record of a vulnerability, if one was fetched after `after`.
1219 pub async fn advisory(&self, osv_id: &str, after: &str) -> Result<Option<serde_json::Value>> {
1220 let row = self
1221 .db
1222 .prepare("SELECT body, fetched_at FROM advisories WHERE osv_id = ?")
1223 .bind(&[osv_id.into()])?
1224 .first::<BodyRow>(None)
1225 .await?;
1226 Ok(row
1227 .filter(|row| row.fetched_at.as_str() >= after)
1228 .and_then(|row| serde_json::from_str(&row.body).ok()))
1229 }
1230
1231 pub async fn keep_advisory(&self, osv_id: &str, body: &serde_json::Value) -> Result<()> {
1232 self.db
1233 .prepare(
1234 "INSERT INTO advisories (osv_id, body, fetched_at) VALUES (?1, ?2, ?3)
1235 ON CONFLICT (osv_id) DO UPDATE SET body = ?2, fetched_at = ?3",
1236 )
1237 .bind(&[osv_id.into(), body.to_string().into(), now().into()])?
1238 .run()
1239 .await?;
1240 Ok(())
1241 }
1242
1243 /// Adds to a workspace's scanning this month; returns the month
1244 /// (`YYYY-MM`) with what it has cost so far, in millionths of a dollar,
1245 /// so the total is reported under the month it was counted in.
1246 pub async fn meter(&self, workspace: &str, reads: u32, commits: u32, osv_queries: u32, cost_micros: i64) -> Result<(String, i64)> {
1247 let month = now()[..7].to_owned();
1248 self.db
1249 .prepare(
1250 "INSERT INTO usage (workspace, month, reads, commits, osv_queries, cost_micros) VALUES (?1, ?2, ?3, ?4, ?5, ?6)
1251 ON CONFLICT (workspace, month) DO UPDATE SET
1252 reads = reads + ?3, commits = commits + ?4, osv_queries = osv_queries + ?5,
1253 cost_micros = cost_micros + ?6",
1254 )
1255 .bind(&[
1256 workspace.into(),
1257 month.as_str().into(),
1258 reads.into(),
1259 commits.into(),
1260 osv_queries.into(),
1261 JsValue::from(cost_micros as f64),
1262 ])?
1263 .run()
1264 .await?;
1265 let total = self
1266 .db
1267 .prepare("SELECT cost_micros AS n FROM usage WHERE workspace = ? AND month = ?")
1268 .bind(&[workspace.into(), month.as_str().into()])?
1269 .first::<NumberRow>(None)
1270 .await?
1271 .map_or(0, |row| row.n);
1272 Ok((month, total))
1273 }
1274}
1275
1276/// Secret alerts counted by where they stand, from (status, looks like a
1277/// test value, how many).
1278pub fn secret_counts<'a>(rows: impl Iterator<Item = (&'a str, bool, u32)>) -> SecretCounts {
1279 let mut counts = SecretCounts::default();
1280 for (status, test, n) in rows {
1281 match (SecretStatus::parse(status), test) {
1282 (Some(SecretStatus::Open | SecretStatus::Blocked), true) => counts.test_values += n,
1283 (Some(SecretStatus::Open), false) => counts.open += n,
1284 (Some(SecretStatus::Blocked), false) => counts.blocked += n,
1285 (Some(SecretStatus::Allowed), _) => counts.dismissed += n,
1286 (Some(SecretStatus::Resolved), _) => counts.fixed += n,
1287 (None, _) => {}
1288 }
1289 }
1290 counts
1291}
1292
1293#[cfg(test)]
1294mod tests {
1295 use super::*;
1296
1297 /// The migrations, in order, from `migrations/`.
1298 fn migrations() -> Vec<(String, String)> {
1299 let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
1300 let mut names: Vec<String> =
1301 std::fs::read_dir(&dir).unwrap().map(|entry| entry.unwrap().file_name().into_string().unwrap()).collect();
1302 names.sort();
1303 names.into_iter().map(|name| (name.clone(), std::fs::read_to_string(dir.join(&name)).unwrap())).collect()
1304 }
1305
1306 #[test]
1307 fn old_issue_numbers_become_their_work_items_numbers() {
1308 let db = rusqlite::Connection::open_in_memory().unwrap();
1309 let (before, after): (Vec<_>, Vec<_>) = migrations().into_iter().partition(|(name, _)| name.as_str() < "0007");
1310 for (name, sql) in &before {
1311 db.execute_batch(sql).unwrap_or_else(|error| panic!("{name}: {error}"));
1312 }
1313 db.execute_batch(
1314 "INSERT INTO updates (repo_id, ecosystem, package, target, state, pull, issue, requested_at, updated_at)
1315 VALUES ('rep_1', 'npm', 'a', '2', 'needs_code', NULL, 5, '', ''), ('rep_2', 'npm', 'a', '2', 'needs_code', NULL, 5, '', '');
1316 INSERT INTO upgrades (repo_id, ecosystem, package, number, target, opened_at) VALUES ('rep_1', 'npm', 'b', 3, '2', '');
1317 INSERT INTO alert_activity (id, repo_id, alert_id, action, number, at)
1318 VALUES ('a1', 'rep_1', 'vul_1', 'update_needs_code', 5, ''), ('a2', 'rep_1', 'vul_1', 'update_opened', 5, '');",
1319 )
1320 .unwrap();
1321 for (name, sql) in &after {
1322 db.execute_batch(sql).unwrap_or_else(|error| panic!("{name}: {error}"));
1323 }
1324 let page = serde_json::json!([
1325 { "repoId": "rep_1", "legacyNumber": 5, "workspace": "acme", "number": 41, "itemId": "iss_5" },
1326 { "repoId": "rep_1", "legacyNumber": 3, "workspace": "acme", "number": 40, "itemId": "iss_3" },
1327 ])
1328 .to_string();
1329 // Heard twice, as a queue may deliver it.
1330 for _ in 0..2 {
1331 for statement in WORK_MIGRATED {
1332 db.execute(&work_migrated_sql(statement), [&page]).unwrap();
1333 }
1334 }
1335 let item = |sql: &str| db.query_row(sql, [], |row| row.get::<_, Option<i64>>(0)).unwrap();
1336 assert_eq!(item("SELECT item FROM updates WHERE repo_id = 'rep_1'"), Some(41));
1337 // Another repository's number 5 was another issue.
1338 assert_eq!(item("SELECT item FROM updates WHERE repo_id = 'rep_2'"), Some(5));
1339 assert_eq!(item("SELECT item FROM upgrades WHERE repo_id = 'rep_1'"), Some(40));
1340 assert_eq!(item("SELECT number FROM alert_activity WHERE id = 'a1'"), Some(41));
1341 // A pull request kept its number.
1342 assert_eq!(item("SELECT number FROM alert_activity WHERE id = 'a2'"), Some(5));
1343 }
1344
1345 #[test]
1346 fn work_migrated_keeps_each_items_id_and_workspace() {
1347 let db = rusqlite::Connection::open_in_memory().unwrap();
1348 let (before, after): (Vec<_>, Vec<_>) = migrations().into_iter().partition(|(name, _)| name.as_str() < "0009");
1349 for (name, sql) in &before {
1350 db.execute_batch(sql).unwrap_or_else(|error| panic!("{name}: {error}"));
1351 }
1352 // rep_1's update is still on its old number; rep_2's upgrade was
1353 // rewritten by a page heard before ids were kept, and the catch-up
1354 // was done.
1355 db.execute_batch(
1356 "INSERT INTO updates (repo_id, ecosystem, package, target, state, pull, item, requested_at, updated_at)
1357 VALUES ('rep_1', 'npm', 'a', '2', 'needs_code', NULL, 5, '', '');
1358 INSERT INTO upgrades (repo_id, ecosystem, package, item, target, opened_at) VALUES ('rep_2', 'npm', 'b', 40, '2', '');
1359 UPDATE work_catch_up SET page = 3, done_at = 'x';",
1360 )
1361 .unwrap();
1362 for (name, sql) in &after {
1363 db.execute_batch(sql).unwrap_or_else(|error| panic!("{name}: {error}"));
1364 }
1365 // The catch-up reads the map again, for the ids of rows already moved.
1366 let done: Option<String> = db.query_row("SELECT done_at FROM work_catch_up", [], |row| row.get(0)).unwrap();
1367 assert_eq!(done, None);
1368 let page = serde_json::json!([
1369 { "repoId": "rep_1", "legacyNumber": 5, "workspace": "acme", "number": 41, "itemId": "iss_5" },
1370 { "repoId": "rep_2", "legacyNumber": 3, "workspace": "acme", "number": 40, "itemId": "iss_3" },
1371 ])
1372 .to_string();
1373 for _ in 0..2 {
1374 for statement in WORK_MIGRATED {
1375 db.execute(&work_migrated_sql(statement), [&page]).unwrap();
1376 }
1377 }
1378 let linked = |sql: &str| {
1379 db.query_row(sql, [], |row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option<String>>(1)?, row.get::<_, Option<String>>(2)?)))
1380 .unwrap()
1381 };
1382 assert_eq!(linked("SELECT item, item_id, item_workspace FROM updates"), (41, Some("iss_5".into()), Some("acme".into())));
1383 assert_eq!(linked("SELECT item, item_id, item_workspace FROM upgrades"), (40, Some("iss_3".into()), Some("acme".into())));
1384 }
1385
1386 #[test]
1387 fn a_linked_item_is_found_where_it_was_kept() {
1388 let kept = LinkedItem::new(Some(41), Some("iss_5"), Some("acme")).unwrap();
1389 // The repository moved to another workspace: its item did not.
1390 assert_eq!(kept.workspace_or("globex"), "acme");
1391 assert!(kept.is("iss_5"));
1392 // Another item that has the number now is not it.
1393 assert!(!kept.is("wki_9"));
1394 let before = LinkedItem::new(Some(41), None, None).unwrap();
1395 assert_eq!(before.workspace_or("globex"), "globex");
1396 assert!(before.is("wki_9"));
1397 }
1398
1399 #[test]
1400 fn only_real_secrets_in_history_are_open() {
1401 let counts = secret_counts(
1402 [("open", false, 2), ("open", true, 3), ("blocked", false, 1), ("blocked", true, 4), ("allowed", true, 5), ("resolved", false, 6)]
1403 .into_iter(),
1404 );
1405 assert_eq!((counts.open, counts.blocked, counts.test_values, counts.dismissed, counts.fixed), (2, 1, 7, 5, 6));
1406 }
1407}