Secret scanning: scan existing history, remove an allowed fingerprint cb7b3cd45dc0

-- Secret scanning, part 2 (oxygenhub#277): findings from the background

-- scan of each repository's history beside those of refused pushes, with

-- the change and bookmark they were found in and a maintainer's mark that

-- the secret was revoked; and each repository's scan progress, so a pass

-- resumes where the last one stopped and a push is scanned incrementally.

ALTER TABLE secret_findings ADD COLUMN source TEXT NOT NULL DEFAULT 'push';

ALTER TABLE secret_findings ADD COLUMN change_hash TEXT NOT NULL DEFAULT '';

ALTER TABLE secret_findings ADD COLUMN bookmark TEXT NOT NULL DEFAULT '';

ALTER TABLE secret_findings ADD COLUMN resolved_ms BIGINT;

ALTER TABLE secret_findings ADD COLUMN resolved_by TEXT NOT NULL DEFAULT '';

ALTER TABLE secret_findings ADD COLUMN resolved_reason TEXT NOT NULL DEFAULT '';

-- One history finding per value and file, however many changes carry it.

CREATE UNIQUE INDEX secret_findings_history ON secret_findings(repo_id, fingerprint, path) WHERE source = 'history';

CREATE TABLE secret_scans (

    repo_id       BIGINT PRIMARY KEY REFERENCES repositories(id) ON DELETE CASCADE,

    rules_version BIGINT NOT NULL DEFAULT 0,

    done_heads    TEXT NOT NULL DEFAULT '',

    target_heads  TEXT NOT NULL DEFAULT '',

    position      BIGINT NOT NULL DEFAULT 0,

    pending       BIGINT NOT NULL DEFAULT 0,

    changes       BIGINT NOT NULL DEFAULT 0,

    bytes         BIGINT NOT NULL DEFAULT 0,

    requested_ms  BIGINT,

    pass_ms       BIGINT,

    finished_ms   BIGINT,

    error         TEXT NOT NULL DEFAULT ''

);

-- Secret scanning, part 2 (oxygenhub#277): findings from the background

-- scan of each repository's history beside those of refused pushes, with

-- the change and bookmark they were found in and a maintainer's mark that

-- the secret was revoked; and each repository's scan progress, so a pass

-- resumes where the last one stopped and a push is scanned incrementally.

ALTER TABLE secret_findings ADD COLUMN source TEXT NOT NULL DEFAULT 'push';

ALTER TABLE secret_findings ADD COLUMN change_hash TEXT NOT NULL DEFAULT '';

ALTER TABLE secret_findings ADD COLUMN bookmark TEXT NOT NULL DEFAULT '';

ALTER TABLE secret_findings ADD COLUMN resolved_ms INTEGER;

ALTER TABLE secret_findings ADD COLUMN resolved_by TEXT NOT NULL DEFAULT '';

ALTER TABLE secret_findings ADD COLUMN resolved_reason TEXT NOT NULL DEFAULT '';

-- One history finding per value and file, however many changes carry it.

CREATE UNIQUE INDEX secret_findings_history ON secret_findings(repo_id, fingerprint, path) WHERE source = 'history';

CREATE TABLE secret_scans (

    repo_id       INTEGER PRIMARY KEY REFERENCES repositories(id) ON DELETE CASCADE,

    rules_version INTEGER NOT NULL DEFAULT 0,

    done_heads    TEXT NOT NULL DEFAULT '',

    target_heads  TEXT NOT NULL DEFAULT '',

    position      INTEGER NOT NULL DEFAULT 0,

    pending       INTEGER NOT NULL DEFAULT 0,

    changes       INTEGER NOT NULL DEFAULT 0,

    bytes         INTEGER NOT NULL DEFAULT 0,

    requested_ms  INTEGER,

    pass_ms       INTEGER,

    finished_ms   INTEGER,

    error         TEXT NOT NULL DEFAULT ''

);

//! The history scan (hub spec §15.4, oxygenhub#277): push protection's

//! rules run over what a repository already holds, so a secret committed

//! before push protection existed, or while it was off, is found too.

//!

//! What is scanned is every change reachable from the repository's

//! bookmarks: the lines it inserts and the text blobs it puts in place of

//! a file's content, exactly what push protection reads in a push. A

//! repository's progress is a row in `secret_scans`: the heads whose

//! history is scanned (`done_heads`), and while a scan is under way the

//! heads it works toward (`target_heads`) and how far it got (`position`

//! in its list of changes: those reachable from the target and not from

//! the done heads, in clock order then by hash, so the same list every

//! pass). A pass scans within a [`Budget`] and saves where it stopped; the

//! next one resumes there. Once the list is done the target becomes the

//! done heads, so after a push only the new changes are scanned. A change

//! of rules (`version` in `push_protection.toml`) or a maintainer's "Scan

//! history now" scans the whole history again; findings are kept one per

//! value and file, so a rescan adds nothing already known and keeps what

//! was resolved.

//!

//! The scan is the worker role's `secrets.history_scan` task (docs/roles.md),

//! one repository at a time, one pass each in turn, on one blocking thread

//! that rests as long as it works: at most half a core, so it never starves

//! requests. It reads repositories with the shared lock only, as pages do.

//! Repositories with push protection off are not scanned.

use crate::bus::{Event, Topic};

use crate::config::Config;

use crate::db::{self, push_protection as pdb, repos::RepoRow};

use crate::push_protection::{self as pp, Finding};

use crate::repos::{RepoError, RepoName};

use crate::Shared;

use oxy_core::hash::{ChangeHash, ObjectType};

use oxy_core::objects::{Blob, Edit};

use oxy_core::protocol::Bundle;

use oxy_core::repo::Repo;

use std::collections::{BTreeSet, HashMap, HashSet};

use std::time::{Duration, Instant};

/// How much one pass over one repository may scan: `bytes` of text (a

/// change counts at least one, however little it adds) and `time`

/// wall-clock (resting included), each file or inserted text at most

/// `file` bytes. At least one change is scanned per pass, so a pass always

/// gets somewhere.

#[derive(Clone, Copy, Debug)]

pub struct Budget {

    pub bytes: u64,

    pub time: Duration,

    pub file: u64,

}

impl Default for Budget {

    fn default() -> Self {

        Budget { bytes: 64 << 20, time: Duration::from_secs(5), file: pp::Budget::default().file }

    }

}

impl Budget {

    /// The hub's budget from `[limits]`; `None` when the scan is off.

    pub fn of(config: &Config) -> Option<Budget> {

        let l = &config.limits;

        (l.history_scan_limit.0 > 0).then(|| Budget {

            bytes: l.history_scan_limit.0,

            time: Duration::from_millis(l.history_scan_time_ms),

            file: l.scan_file_limit.0,

        })

    }

}

/// What one pass did.

#[derive(Clone, Debug, Default, PartialEq, Eq)]

pub struct Pass {

    /// Changes and bytes scanned in this pass.

    pub changes: usize,

    pub bytes: u64,

    /// Findings new to this repository.

    pub found: usize,

    /// Changes still to scan toward the target.

    pub left: usize,

    /// Nothing to scan: the history up to the bookmarks is scanned.

    pub idle: bool,

}

/// How long the worker waits after starting before its first sweep, so a

/// starting hub serves first.

const START_DELAY: Duration = Duration::from_secs(10);

/// How long it rests between repositories.

const PAUSE: Duration = Duration::from_millis(200);

/// Every repository is looked at again this often, for the pushes the bus

/// did not tell of (it delivers at most once) and bookmarks moved without

/// a push (landings).

const SWEEP_EVERY: Duration = Duration::from_secs(15 * 60);

/// The bus topic a "Scan history now" is announced on, its kind the

/// repository's id.

pub const REQUEST_TOPIC: &str = "secrets:scan";

/// The worker role's task: every repository first, then those a push, a

/// request or a sweep says have something new, each a pass at a time.

pub async fn worker(app: Shared) {

    let Some(budget) = Budget::of(&app.config) else {

        // Off in `[limits]`: the task holds its lock and does nothing.

        return std::future::pending().await;

    };

    let mut events = app.bus.subscribe(Topic::Live);

    tokio::time::sleep(START_DELAY).await;

    let mut sweep = true;

    let mut dirty: BTreeSet<i64> = BTreeSet::new();

    loop {

        let mut order = pdb::requested(&app.db).await.unwrap_or_default();

        if sweep {

            order.extend(db::repos::list(&app.db).await.unwrap_or_default().into_iter().map(|r| r.id));

            sweep = false;

        }

        order.extend(std::mem::take(&mut dirty));

        let mut seen = HashSet::new();

        order.retain(|id| seen.insert(*id));

        for id in order {

            sweep |= hear(&mut events, &mut dirty);

            let Ok(Some(row)) = db::repos::get_by_id(&app.db, id).await else { continue };

            if !row.push_protection {

                continue;

            }

            match pass(&app, &row, budget).await {

                Ok(p) => {

                    if p.left > 0 {

                        dirty.insert(id);

                    }

                    if p.found > 0 {

                        tracing::warn!(

                            "history scan: {}/{}: {} possible secret(s) found in history; see Settings › Push protection",

                            row.owner,

                            row.name,

                            p.found

                        );

                    }

                }

                Err(e) => tracing::warn!("history scan: {}/{}: {e:#}", row.owner, row.name),

            }

            tokio::time::sleep(PAUSE).await;

        }

        sweep |= hear(&mut events, &mut dirty);

        if sweep || !dirty.is_empty() {

            continue;

        }

        // Nothing left: wait for a push, a request or the next sweep.

        let wake = tokio::time::sleep(SWEEP_EVERY);

        tokio::pin!(wake);

        loop {

            tokio::select! {

                ev = events.recv() => match ev {

                    Ok(ev) => if let Some(id) = wakes(&ev) { dirty.insert(id); break; },

                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { sweep = true; break; }

                    Err(tokio::sync::broadcast::error::RecvError::Closed) => { (&mut wake).await; sweep = true; break; }

                },

                _ = &mut wake => { sweep = true; break; }

            }

        }

        if !sweep {

            // Let a burst of pushes settle into one pass each.

            tokio::time::sleep(Duration::from_secs(1)).await;

            hear(&mut events, &mut dirty);

        }

    }

}

/// Take the events waiting: the repositories they name go in `dirty`.

/// Whether events were lost, so everything should be looked at.

fn hear(events: &mut tokio::sync::broadcast::Receiver<Event>, dirty: &mut BTreeSet<i64>) -> bool {

    use tokio::sync::broadcast::error::TryRecvError;

    loop {

        match events.try_recv() {

            Ok(ev) => {

                if let Some(id) = wakes(&ev) {

                    dirty.insert(id);

                }

            }

            Err(TryRecvError::Lagged(_)) => return true,

            Err(_) => return false,

        }

    }

}

/// The repository an event says has something to scan: a push to it

/// (`repo:N`, `push`), or a maintainer's request.

fn wakes(ev: &Event) -> Option<i64> {

    let Event::Live { topic, kind } = ev else { return None };

    if topic == REQUEST_TOPIC {

        return kind.parse().ok();

    }

    if kind != "push" {

        return None;

    }

    topic.strip_prefix("repo:")?.parse().ok()

}

/// Ask for repository `repo_id`'s whole history to be scanned again now

/// ("Scan history now", maintain role).

pub async fn request(app: &Shared, repo_id: i64) -> sqlx::Result<()> {

    pdb::request_scan(&app.db, repo_id).await?;

    wake(app, repo_id);

    Ok(())

}

/// Tell the worker that repository `repo_id` may have history to scan (its

/// push protection was turned back on).

pub fn wake(app: &Shared, repo_id: i64) {

    app.bus.live(REQUEST_TOPIC, &repo_id.to_string());

}

fn heads_text(heads: &[ChangeHash]) -> String {

    heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>().join(",")

}

fn heads_of(text: &str) -> Vec<ChangeHash> {

    text.split(',').filter_map(ChangeHash::from_hex).collect()

}

/// One pass over `row`'s history within `budget`: resume the scan under

/// way, or start one toward the bookmarks if they moved past what was

/// scanned; record what it finds and where it stopped.

pub async fn pass(app: &Shared, row: &RepoRow, budget: Budget) -> anyhow::Result<Pass> {

    let name = RepoName::new(&row.owner, &row.name).ok_or_else(|| anyhow::anyhow!("not a repository name"))?;

    let loaded = pdb::scan_of(&app.db, row.id).await?.unwrap_or(pdb::ScanRow { repo_id: row.id, ..Default::default() });

    let mut s = loaded.clone();

    let version = pp::rules().version as i64;

    if s.rules_version != version {

        // New rules: the whole history again.

        s.rules_version = version;

        s.done_heads.clear();

        s.target_heads.clear();

        s.position = 0;

    }

    let from = Progress { done: heads_of(&s.done_heads), target: heads_of(&s.target_heads), position: s.position.max(0) as usize };

    let default = row.default_bookmark.clone();

    let started = db::now_ms();

    let work = app.repos.read(&name, move |repo, _| scan_in(repo, from, &default, budget)).await;

    let work = match work {

        Ok(w) => w,

        Err(e) => {

            s.error = e.to_string();

            s.pass_ms = Some(started);

            save(app, &loaded, &s).await?;

            return Err(e.into());

        }

    };

    if work.idle && s == loaded {

        // Up to date, as the row already says: nothing to write.

        return Ok(Pass { idle: true, ..Default::default() });

    }

    let records: Vec<pdb::NewHistoryFinding> = work

        .findings

        .iter()

        .map(|(f, bookmark)| pdb::NewHistoryFinding {

            finding: pdb::NewFinding {

                rule: &f.rule,

                description: &f.description,

                path: &f.path,

                line: f.line as i64,

                fingerprint: &f.fingerprint,

                excerpt: &f.excerpt,

            },

            change_hash: f.change_hash.map(|h| h.to_hex()).unwrap_or_default(),

            bookmark: bookmark.clone(),

        })

        .collect();

    let found = pdb::record_history(&app.db, row.id, &records).await?;

    let left = work.pending - work.position;

    if work.idle || left == 0 {

        s.done_heads = heads_text(&work.target);

        s.target_heads.clear();

        s.position = 0;

        s.pending = 0;

        s.requested_ms = None;

        if !work.idle {

            s.finished_ms = Some(db::now_ms());

        }

    } else {

        s.target_heads = heads_text(&work.target);

        s.position = work.position as i64;

        s.pending = work.pending as i64;

    }

    s.changes += work.changes as i64;

    s.bytes += work.bytes as i64;

    s.pass_ms = Some(started);

    s.error.clear();

    save(app, &loaded, &s).await?;

    Ok(Pass { changes: work.changes, bytes: work.bytes, found, left, idle: work.idle })

}

/// Save `s` unless a maintainer asked for a new scan while the pass ran:

/// that request's fresh start wins.

async fn save(app: &Shared, loaded: &pdb::ScanRow, s: &pdb::ScanRow) -> sqlx::Result<()> {

    let now = pdb::scan_of(&app.db, s.repo_id).await?;

    if now.as_ref().and_then(|n| n.requested_ms) != loaded.requested_ms {

        return Ok(());

    }

    pdb::save_scan(&app.db, s).await

}

/// Where a scan stands, as read from its row.

struct Progress {

    done: Vec<ChangeHash>,

    target: Vec<ChangeHash>,

    position: usize,

}

/// What [`scan_in`] did.

struct Work {

    /// The heads it scanned toward, and how far it got of how many.

    target: Vec<ChangeHash>,

    position: usize,

    pending: usize,

    idle: bool,

    changes: usize,

    bytes: u64,

    /// Each finding, placed, with the bookmark it was found on.

    findings: Vec<(Finding, String)>,

}

/// The pass itself, on a blocking thread with the repository open.

fn scan_in(repo: &Repo, from: Progress, default: &str, budget: Budget) -> Result<Work, RepoError> {

    let started = Instant::now();

    let view = repo.view()?;

    let mut bookmarks: Vec<(String, Vec<ChangeHash>)> = Vec::new();

    for name in view.bookmarks.keys() {

        bookmarks.push((name.clone(), repo.heads_of(&view, name)?));

    }

    // The default bookmark first: a finding is named by the first bookmark

    // that reaches it.

    bookmarks.sort_by_key(|(n, _)| (n != default, n.clone()));

    let mut current: Vec<ChangeHash> = bookmarks.iter().flat_map(|(_, h)| h.iter().copied()).collect();

    current.sort();

    current.dedup();

    let mut done = from.done;

    done.sort();

    let (mut target, mut position) = (from.target, from.position);

    if target.is_empty() {

        if current == done {

            return Ok(Work { target: done, position: 0, pending: 0, idle: true, changes: 0, bytes: 0, findings: vec![] });

        }

        target = current.clone();

        position = 0;

    }

    let state = match repo.load_state_history(&target) {

        Ok(s) => s,

        // What the scan worked toward is gone (a bookmark rewritten, then

        // collected): start again toward the bookmarks as they are.

        Err(_) if target != current => {

            target = current.clone();

            position = 0;

            repo.load_state_history(&target)?

        }

        Err(e) => return Err(e.into()),

    };

    // What is scanned already: the done heads' history.

    let scanned: HashSet<ChangeHash> = if done.iter().all(|h| state.contains(h)) {

        closure(&done, |h| state.get(h).map(|m| m.deps.clone()))

    } else {

        match repo.load_state_history(&done) {

            Ok(old) => old.changes().map(|(h, _)| *h).collect(),

            Err(_) => HashSet::new(),

        }

    };

    let mut list: Vec<(oxy_core::objects::Hlc, ChangeHash)> =

        state.changes().filter(|(h, _)| !scanned.contains(*h)).map(|(h, m)| (m.hlc, *h)).collect();

    list.sort();

    let pending = list.len();

    if position > pending {

        position = 0;

    }

    let file = pp::Budget { bytes: u64::MAX, file: budget.file, time: Duration::MAX };

    let mut work = Work { target, position, pending, idle: false, changes: 0, bytes: 0, findings: vec![] };

    let mut reach: HashMap<String, HashSet<ChangeHash>> = HashMap::new();

    let mut busy = Duration::ZERO;

    let mut spent: u64 = 0;

    while work.position < pending {

        if work.changes > 0 && (spent >= budget.bytes || started.elapsed() >= budget.time) {

            break;

        }

        let t = Instant::now();

        let h = list[work.position].1;

        let bundle = bundle_of(repo, &h, budget.file)?;

        let scan = pp::scan(&bundle, file);

        work.bytes += scan.scanned;

        spent += scan.scanned.max(1);

        if !scan.findings.is_empty() {

            let mut findings = scan.findings;

            pp::locate(repo, &bundle, &mut findings);

            let on = bookmarks

                .iter()

                .find(|(name, heads)| {

                    reach.entry(name.clone()).or_insert_with(|| closure(heads, |c| state.get(c).map(|m| m.deps.clone()))).contains(&h)

                })

                .map(|(name, _)| name.clone())

                .unwrap_or_default();

            work.findings.extend(findings.into_iter().map(|f| (f, on.clone())));

        }

        work.position += 1;

        work.changes += 1;

        // Rest as long as it worked, so the scan takes at most half a core.

        busy += t.elapsed();

        if busy >= Duration::from_millis(20) {

            std::thread::sleep(busy);

            busy = Duration::ZERO;

        }

    }

    Ok(work)

}

/// `heads` and every change they depend on, by `deps`.

fn closure(heads: &[ChangeHash], deps: impl Fn(&ChangeHash) -> Option<Vec<ChangeHash>>) -> HashSet<ChangeHash> {

    let mut seen: HashSet<ChangeHash> = HashSet::new();

    let mut stack: Vec<ChangeHash> = heads.to_vec();

    while let Some(h) = stack.pop() {

        if !seen.insert(h) {

            continue;

        }

        if let Some(d) = deps(&h) {

            stack.extend(d.into_iter().filter(|x| !seen.contains(x)));

        }

    }

    seen

}

/// One stored change as push protection reads a push: the change, and the

/// blobs (with their chunks, unless over the file budget) its edits put in

/// place of a file's content. A blob the store does not hold (lazy) is

/// left out, so it is not scanned.

fn bundle_of(repo: &Repo, h: &ChangeHash, file: u64) -> Result<Bundle, RepoError> {

    let store = repo.store();

    let change = store.get_change(h)?;

    let mut objects = Vec::new();

    for e in &change.edits {

        let Edit::Replace { new: Some(r), .. } = e else { continue };

        let Ok(body) = store.get_typed(r, ObjectType::Blob) else { continue };

        let Ok(blob) = Blob::decode(&body) else { continue };

        if blob.length <= file {

            for c in &blob.chunks {

                if let Ok(chunk) = store.get_typed(c, ObjectType::Chunk) {

                    objects.push((ObjectType::Chunk, chunk));

                }

            }

        }

        objects.push((ObjectType::Blob, body));

    }

    Ok(Bundle { changes: vec![(change, vec![])], hashes: vec![*h], objects, ..Default::default() })

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn pushes_and_requests_wake_the_scan() {

        let live = |topic: &str, kind: &str| Event::Live { topic: topic.into(), kind: kind.into() };

        assert_eq!(wakes(&live("repo:7", "push")), Some(7));

        assert_eq!(wakes(&live("repo:7", "issue")), None);

        assert_eq!(wakes(&live("repo:7/proposal:2", "push")), None);

        assert_eq!(wakes(&live(REQUEST_TOPIC, "9")), Some(9));

        assert_eq!(wakes(&Event::DeliveryWaiting), None);

    }

    #[test]

    fn heads_round_trip() {

        let a = ChangeHash::from_hex(&"ab".repeat(32)).unwrap();

        let b = ChangeHash::from_hex(&"cd".repeat(32)).unwrap();

        assert_eq!(heads_of(&heads_text(&[a, b])), [a, b]);

        assert!(heads_of("").is_empty());

    }

}

//! The history scan (oxygenhub#277, hub spec §15.4): a secret that reached

//! a repository before push protection existed, or while it was off, is

//! found in its history with the change, file and line; a maintainer

//! allows it or marks it resolved; an allowed fingerprint can be removed;

//! after a push only the new changes are scanned; a pass stops at its

//! budget and the next resumes there; the worker scans every repository

//! when it starts and each after a push.

//!

//! Every credential here is made up and put together at run time, so this

//! file is not itself a finding.

use axum::body::Body;

use axum::http::{header, Request, StatusCode};

use http_body_util::BodyExt;

use oxy_core::hash::ChangeHash;

use oxy_core::repo::Repo;

use oxygenhub::auth::random_hex;

use oxygenhub::config::Config;

use oxygenhub::db;

use oxygenhub::history_scan::{pass, Budget};

use oxygenhub::push_protection::fingerprint;

use oxygenhub::repos::RepoName;

use oxygenhub::Shared;

use std::path::Path;

use std::time::Duration;

use tower::ServiceExt;

fn hermetic() {

    static ONCE: std::sync::Once = std::sync::Once::new();

    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();

    ONCE.call_once(|| {

        let d = tempfile::tempdir().unwrap();

        std::env::set_var("OXY_HOME", d.path());

        DIR.set(d).ok();

    });

}

fn write(root: &Path, rel: &str, content: &[u8]) {

    let p = root.join(rel);

    std::fs::create_dir_all(p.parent().unwrap()).unwrap();

    std::fs::write(p, content).unwrap();

}

struct Hub {

    app: Shared,

    base: String,

    _dir: tempfile::TempDir,

}

/// A hub; with `worker` false the background scan is off and the tests

/// run its passes themselves.

async fn hub(worker: bool) -> Hub {

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let mut config = Config::load(dir.path()).unwrap();

    config.registration.setup_code = "test-setup".into();

    config.registration.mode = "open".into();

    if !worker {

        config.limits.history_scan_limit = oxygenhub::config::ByteSize(0);

    }

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let port = listener.local_addr().unwrap().port();

    config.server.base_url = format!("http://127.0.0.1:{port}");

    let app = oxygenhub::build(config).await.unwrap();

    let router = oxygenhub::router(app.clone());

    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });

    Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir }

}

struct Res {

    status: StatusCode,

    body: String,

    cookie: Option<String>,

}

async fn send(app: &Shared, method: &str, path: &str, cookie: Option<&str>, form: Option<&str>) -> Res {

    let mut req = Request::builder().method(method).uri(path);

    if let Some(c) = cookie {

        req = req.header(header::COOKIE, format!("oxyhub={c}"));

    }

    let body = match form {

        Some(f) => {

            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");

            Body::from(f.to_string())

        }

        None => Body::empty(),

    };

    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();

    let status = res.status();

    let cookie = res

        .headers()

        .get(header::SET_COOKIE)

        .and_then(|v| v.to_str().ok())

        .and_then(|v| v.split(';').next())

        .and_then(|kv| kv.split_once('='))

        .map(|(_, v)| v.to_string());

    let bytes = res.into_body().collect().await.unwrap().to_bytes();

    Res { status, body: String::from_utf8_lossy(&bytes).into_owned(), cookie }

}

async fn get(app: &Shared, path: &str, cookie: Option<&str>) -> Res {

    send(app, "GET", path, cookie, None).await

}

async fn post(app: &Shared, path: &str, cookie: Option<&str>, form: &str) -> Res {

    send(app, "POST", path, cookie, Some(form)).await

}

async fn register(app: &Shared, user: &str) -> String {

    let r = post(

        app,

        "/register",

        None,

        &format!("username={user}&email={user}%40example.com&password=correct-horse-battery&setup=test-setup"),

    )

    .await;

    assert_eq!(r.status, StatusCode::SEE_OTHER, "{}", r.body);

    r.cookie.unwrap()

}

fn csrf_of(page: &str) -> String {

    let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();

    page[i..].split('"').next().unwrap().to_string()

}

async fn token(app: &Shared, user: &str) -> String {

    let me = db::accounts::account_by_name(&app.db, user).await.unwrap().unwrap();

    let (secret, hash) = oxygenhub::auth::new_token();

    db::accounts::create_token(&app.db, me.id, "laptop", &hash, "repo:read,repo:write", "", None).await.unwrap();

    secret

}

/// andy/demo with a client checkout at `work` and its first change pushed.

async fn demo(h: &Hub, work: &Path, tok: &str) -> Repo {

    let repo = RepoName::parse("andy/demo").unwrap();

    h.app.repos.create(&repo, None).unwrap();

    db::repos::create(&h.app.db, "andy", "demo", "", hub_api::Visibility::Public).await.unwrap();

    let mut client = Repo::init(work).unwrap();

    client.create_identity("Andy McCall", "[email protected]", oxy_core::repo::IdentityScope::User).unwrap();

    client.config.set("remote.origin.url", &format!("{}/andy/demo", h.base)).unwrap();

    client.config.set("remote.origin.token", tok).unwrap();

    client.save_config().unwrap();

    write(work, "README.md", b"one\ntwo\nthree\n");

    client.snapshot("t").unwrap();

    client.commit("base", &[]).unwrap();

    assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

    client

}

async fn row(app: &Shared) -> db::repos::RepoRow {

    db::repos::get(&app.db, "andy", "demo").await.unwrap().unwrap()

}

/// Commit `content` at `file` and push it with push protection off, as a

/// repository's history got its secrets before push protection existed.

async fn push_unprotected(app: &Shared, client: &Repo, work: &Path, file: &str, content: &str) -> ChangeHash {

    let id = row(app).await.id;

    db::push_protection::set_enabled(&app.db, id, false).await.unwrap();

    write(work, file, content.as_bytes());

    client.snapshot("t").unwrap();

    let h = client.commit(&format!("add {file}"), &[]).unwrap();

    assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

    db::push_protection::set_enabled(&app.db, id, true).await.unwrap();

    h

}

fn stripe() -> String {

    format!("{}_{}_{}", "sk", "live", random_hex(12))

}

fn github() -> String {

    format!("{}_{}", "ghp", random_hex(18))

}

/// The whole flow: found with change, file and line; allowed, resolved, an

/// allowed fingerprint removed; incremental after a push; scanned again

/// on request without duplicates.

#[tokio::test(flavor = "multi_thread")]

async fn a_secret_in_history_is_found_and_judged() {

    let h = hub(false).await;

    let app = &h.app;

    let andy = register(app, "andy").await;

    let tok = token(app, "andy").await;

    let work = tempfile::tempdir().unwrap();

    let client = demo(&h, work.path(), &tok).await;

    let key = stripe();

    let leak = push_unprotected(app, &client, work.path(), "deploy/pay.env", &format!("# payments\nSTRIPE={key}\n")).await;

    let budget = Budget::default();

    // The first pass scans the whole history and finds it, placed.

    let p = pass(app, &row(app).await, budget).await.unwrap();

    assert_eq!((p.found, p.left, p.idle), (1, 0, false), "{p:?}");

    assert!(p.changes >= 2, "{p:?}");

    let all = p.changes;

    let found = db::push_protection::history(&app.db, row(app).await.id, 50).await.unwrap();

    assert_eq!(found.len(), 1, "{found:?}");

    let f = &found[0];

    let fp = fingerprint("stripe-live-key", key.as_bytes());

    assert_eq!((f.path.as_str(), f.line, f.fingerprint.as_str()), ("deploy/pay.env", 2, fp.as_str()));

    assert_eq!((f.change_hash.as_str(), f.bookmark.as_str()), (leak.to_hex().as_str(), "main"));

    assert!(!f.excerpt.contains(&key), "{}", f.excerpt);

    // Nothing new: the next pass has nothing to do.

    let p = pass(app, &row(app).await, budget).await.unwrap();

    assert!(p.idle && p.changes == 0, "{p:?}");

    // Maintainers see it under Found in history, never the secret.

    let page = get(app, "/andy/demo/settings", Some(&andy)).await;

    assert!(page.body.contains("Found in history") && page.body.contains(&fp), "{}", page.body);

    assert!(page.body.contains(&format!("/andy/demo/change/{}", leak.to_hex())) && page.body.contains("deploy/pay.env:2"));

    assert!(!page.body.contains(&key), "the settings page shows no secret");

    // This hub runs no scan of its own: the page says so.

    assert!(page.body.contains("The history scan is off on this hub"), "{}", page.body);

    let csrf = csrf_of(&page.body);

    // A push is scanned incrementally: only its change.

    write(work.path(), "README.md", b"one\ntwo\nthree\nfour\n");

    client.snapshot("t").unwrap();

    client.commit("four", &[]).unwrap();

    assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

    let p = pass(app, &row(app).await, budget).await.unwrap();

    assert_eq!((p.changes, p.found, p.left), (1, 0, 0), "{p:?}");

    let ghp = github();

    let second = push_unprotected(app, &client, work.path(), "ci.env", &format!("GH_TOKEN={ghp}\n")).await;

    let p = pass(app, &row(app).await, budget).await.unwrap();

    assert_eq!((p.changes, p.found), (1, 1), "{p:?}");

    let gfp = fingerprint("github-token", ghp.as_bytes());

    let found = db::push_protection::history(&app.db, row(app).await.id, 50).await.unwrap();

    assert!(

        found.iter().any(|f| f.fingerprint == gfp && f.path == "ci.env" && f.line == 1 && f.change_hash == second.to_hex()),

        "{found:?}"

    );

    // A writer can neither resolve nor remove an allowed fingerprint.

    let bob = register(app, "bob").await;

    let r = post(app, "/andy/demo/settings/collaborators", Some(&andy), &format!("_csrf={csrf}&username=bob&role=write")).await;

    assert_eq!(r.status, StatusCode::OK, "{}", r.body);

    let bob_csrf = csrf_of(&get(app, "/andy/demo", Some(&bob)).await.body);

    let stripe_id = found.iter().find(|f| f.fingerprint == fp).unwrap().id;

    let r = post(app, "/andy/demo/settings/secret-resolve", Some(&bob), &format!("_csrf={bob_csrf}&finding={stripe_id}&reason=x")).await;

    assert_eq!(r.status, StatusCode::NOT_FOUND, "write is not maintain");

    let r = post(app, "/andy/demo/settings/secret-scan", Some(&bob), &format!("_csrf={bob_csrf}")).await;

    assert_eq!(r.status, StatusCode::NOT_FOUND, "write is not maintain");

    // Resolved: the key was revoked. A reason is needed; audited.

    let r = post(app, "/andy/demo/settings/secret-resolve", Some(&andy), &format!("_csrf={csrf}&finding={stripe_id}&reason=")).await;

    assert!(r.body.contains("Say how it was resolved"), "{}", r.body);

    let r = post(

        app,

        "/andy/demo/settings/secret-resolve",

        Some(&andy),

        &format!("_csrf={csrf}&finding={stripe_id}&reason=revoked+in+the+Stripe+dashboard"),

    )

    .await;

    assert_eq!(r.status, StatusCode::OK, "{}", r.body);

    assert!(r.body.contains("marked resolved") && r.body.contains("revoked in the Stripe dashboard"), "{}", r.body);

    let open = db::push_protection::history(&app.db, row(app).await.id, 50).await.unwrap();

    assert!(open.iter().all(|f| f.id != stripe_id), "{open:?}");

    let resolved = db::push_protection::resolved(&app.db, row(app).await.id, 50).await.unwrap();

    assert!(resolved.iter().any(|f| f.id == stripe_id && f.resolved_by == "andy" && f.resolved_ms.is_some()), "{resolved:?}");

    let r = post(app, "/andy/demo/settings/secret-resolve", Some(&andy), &format!("_csrf={csrf}&finding={stripe_id}&reason=again")).await;

    assert!(r.body.contains("not open here"), "only once");

    let audit = db::accounts::audit_rows(&app.db, 50).await.unwrap();

    assert!(

        audit.iter().any(|a| a.action == "secret.resolved" && a.detail.contains(&fp) && a.detail.contains("deploy/pay.env:2")),

        "{audit:?}"

    );

    // Allowed from its history row (a test token): no longer open.

    let r =

        post(app, "/andy/demo/settings/secret-bypass", Some(&andy), &format!("_csrf={csrf}&fingerprint={gfp}&reason=a+test+fixture")).await;

    assert_eq!(r.status, StatusCode::OK, "{}", r.body);

    assert!(db::push_protection::history(&app.db, row(app).await.id, 50).await.unwrap().is_empty());

    let audit = db::accounts::audit_rows(&app.db, 50).await.unwrap();

    assert!(audit.iter().any(|a| a.action == "secret.bypass" && a.detail.contains(&gfp) && a.detail.contains("github-token")), "{audit:?}");

    // The allowed fingerprint is removed again: a writer cannot; a

    // maintainer can, audited; the finding is open again and a push adding

    // that value is refused again.

    let r = post(app, "/andy/demo/settings/secret-bypass/remove", Some(&bob), &format!("_csrf={bob_csrf}&fingerprint={gfp}")).await;

    assert_eq!(r.status, StatusCode::NOT_FOUND, "write is not maintain");

    let r = post(app, "/andy/demo/settings/secret-bypass/remove", Some(&andy), &format!("_csrf={csrf}&fingerprint={gfp}")).await;

    assert_eq!(r.status, StatusCode::OK, "{}", r.body);

    assert!(r.body.contains(&format!("Fingerprint {gfp} is no longer allowed")), "{}", r.body);

    assert!(db::push_protection::allowed(&app.db, row(app).await.id).await.unwrap().is_empty());

    let audit = db::accounts::audit_rows(&app.db, 50).await.unwrap();

    assert!(

        audit.iter().any(|a| a.action == "secret.bypass_removed" && a.detail.contains(&gfp) && a.detail.contains("a test fixture")),

        "{audit:?}"

    );

    let open = db::push_protection::history(&app.db, row(app).await.id, 50).await.unwrap();

    assert!(open.iter().any(|f| f.fingerprint == gfp), "{open:?}");

    let r = post(app, "/andy/demo/settings/secret-bypass/remove", Some(&andy), &format!("_csrf={csrf}&fingerprint={gfp}")).await;

    assert!(r.body.contains("is not allowed here"), "{}", r.body);

    write(work.path(), "ci2.env", format!("GH_TOKEN={ghp}\n").as_bytes());

    client.snapshot("t").unwrap();

    let again = client.commit("again", &[]).unwrap();

    let err = client.push("origin", false).expect_err("refused again").to_string();

    assert!(err.contains(&format!("(fingerprint {gfp})")), "{err}");

    client.abandon(&again).unwrap();

    let _ = std::fs::remove_file(work.path().join("ci2.env"));

    client.snapshot("t").unwrap();

    // Scan history now: the whole history again, audited; nothing found

    // twice, and what was resolved stays resolved.

    let r = post(app, "/andy/demo/settings/secret-scan", Some(&andy), &format!("_csrf={csrf}")).await;

    assert_eq!(r.status, StatusCode::OK, "{}", r.body);

    assert!(r.body.contains("will be scanned again"), "{}", r.body);

    assert!(db::push_protection::scan_of(&app.db, row(app).await.id).await.unwrap().unwrap().requested_ms.is_some());

    let p = pass(app, &row(app).await, budget).await.unwrap();

    assert_eq!((p.changes, p.found, p.left), (all + 2, 0, 0), "{p:?}");

    let s = db::push_protection::scan_of(&app.db, row(app).await.id).await.unwrap().unwrap();

    assert!(s.requested_ms.is_none() && s.target_heads.is_empty() && s.finished_ms.is_some(), "{s:?}");

    assert_eq!(db::push_protection::resolved(&app.db, row(app).await.id, 50).await.unwrap().len(), 1);

    assert_eq!(db::push_protection::history(&app.db, row(app).await.id, 50).await.unwrap().len(), 1);

    let audit = db::accounts::audit_rows(&app.db, 50).await.unwrap();

    assert!(audit.iter().any(|a| a.action == "secret.history_scan"), "{audit:?}");

    // Findings of refused pushes stay apart from history's.

    let recent = db::push_protection::recent(&app.db, row(app).await.id, 50).await.unwrap();

    assert!(recent.iter().all(|f| f.fingerprint == gfp && f.path == "ci2.env"), "{recent:?}");

}

/// A pass stops at its budget and the next resumes where it stopped; a push

/// in the meantime waits for the scan under way, then is scanned alone.

#[tokio::test(flavor = "multi_thread")]

async fn the_scan_resumes_where_its_budget_stopped() {

    let h = hub(false).await;

    let app = &h.app;

    register(app, "andy").await;

    let tok = token(app, "andy").await;

    let work = tempfile::tempdir().unwrap();

    let client = demo(&h, work.path(), &tok).await;

    let key = stripe();

    for i in 0..5 {

        let text = if i == 3 { format!("n = {i}\nSTRIPE={key}\n") } else { format!("n = {i}\n") };

        push_unprotected(app, &client, work.path(), &format!("f{i}.txt"), &text).await;

    }

    // A byte: one change a pass.

    let budget = Budget { bytes: 1, ..Budget::default() };

    let first = pass(app, &row(app).await, budget).await.unwrap();

    assert_eq!(first.changes, 1, "{first:?}");

    let total = first.left + 1;

    assert!(total >= 6, "{first:?}");

    let s = db::push_protection::scan_of(&app.db, row(app).await.id).await.unwrap().unwrap();

    assert_eq!((s.position, s.pending), (1, total as i64), "{s:?}");

    assert!(!s.target_heads.is_empty() && s.done_heads.is_empty(), "{s:?}");

    let second = pass(app, &row(app).await, budget).await.unwrap();

    assert_eq!((second.changes, second.left), (1, total - 2), "{second:?}");

    // A push now does not restart the scan under way.

    write(work.path(), "late.txt", b"late\n");

    client.snapshot("t").unwrap();

    client.commit("late", &[]).unwrap();

    assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

    let mut found = first.found + second.found;

    let mut passes = 2;

    loop {

        let p = pass(app, &row(app).await, budget).await.unwrap();

        assert_eq!(p.changes, 1, "{p:?}");

        found += p.found;

        passes += 1;

        if p.left == 0 {

            break;

        }

        assert!(passes < 50, "the scan never ends");

    }

    assert_eq!((passes, found), (total, 1), "each change once, the secret once");

    // Then the push, alone.

    let p = pass(app, &row(app).await, budget).await.unwrap();

    assert_eq!((p.changes, p.left, p.found), (1, 0, 0), "{p:?}");

    assert!(pass(app, &row(app).await, budget).await.unwrap().idle);

    let found = db::push_protection::history(&app.db, row(app).await.id, 50).await.unwrap();

    assert_eq!(found.iter().map(|f| (f.path.as_str(), f.line)).collect::<Vec<_>>(), [("f3.txt", 2)]);

}

/// The worker scans every repository when it starts (as on the first

/// deploy) and a repository again after a push, by itself.

#[tokio::test(flavor = "multi_thread")]

async fn the_worker_scans_on_start_and_after_a_push() {

    let h = hub(true).await;

    let app = &h.app;

    let andy = register(app, "andy").await;

    let tok = token(app, "andy").await;

    let work = tempfile::tempdir().unwrap();

    let client = demo(&h, work.path(), &tok).await;

    let key = stripe();

    push_unprotected(app, &client, work.path(), "old.env", &format!("STRIPE={key}\n")).await;

    let id = row(app).await.id;

    let until = |n: usize| async move {

        for _ in 0..300 {

            if db::push_protection::history(&app.db, id, 50).await.unwrap().len() >= n {

                return true;

            }

            tokio::time::sleep(Duration::from_millis(100)).await;

        }

        false

    };

    assert!(until(1).await, "found on the first sweep");

    // A push with protection on that adds nothing is scanned by itself;

    // one let through (protection off for it) is found once it is on again.

    let ghp = github();

    push_unprotected(app, &client, work.path(), "new.env", &format!("GH={ghp}\n")).await;

    write(work.path(), "README.md", b"one\n");

    client.snapshot("t").unwrap();

    client.commit("touch", &[]).unwrap();

    assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

    assert!(until(2).await, "found after the push");

    let found = db::push_protection::history(&app.db, id, 50).await.unwrap();

    assert!(found.iter().any(|f| f.path == "new.env" && f.fingerprint == fingerprint("github-token", ghp.as_bytes())), "{found:?}");

    for _ in 0..100 {

        let s = db::push_protection::scan_of(&app.db, id).await.unwrap().unwrap();

        if s.target_heads.is_empty() && s.finished_ms.is_some() {

            break;

        }

        tokio::time::sleep(Duration::from_millis(100)).await;

    }

    let page = get(app, "/andy/demo/settings", Some(&andy)).await;

    assert!(page.body.contains("Scanned up to the bookmarks") && page.body.contains("SCAN HISTORY NOW"), "{}", page.body);

}

- **Secret scanning finds what is already in history (#277):** push protection's rules now also run over each repository's existing history in the background (every change reachable from its bookmarks: the lines it inserts and the text files it replaces), so a credential committed before push protection existed, or while it was off, is found too. Every repository's whole history is scanned once after this deploy, then only what each push adds, a budget at a time (`[limits] history_scan_limit`, 64 MiB, and `history_scan_time_ms`, 5 s, per repository per pass; `0` turns it off), at low priority, resuming where it stopped. Maintainers see the findings, with the change, bookmark, file and line, under Settings › Push protection › Found in history, and either allow one (not a secret) or mark it resolved with how it was revoked (audited as `secret.resolved`); "Scan history now" scans the whole history again. An allowed fingerprint can now be removed (audited as `secret.bypass_removed`), so pushes adding that value are refused again. Repositories with push protection off are not scanned.

        // GET Object), with its published signature. Its documented example

        // keys are put together at run time, so this file is not itself a

        // push protection or history scan finding (#277).

        let key_id = format!("{}{}", "AKIA", "IOSFODNN7EXAMPLE");

        let secret = format!("{}{}", "wJalrXUtnFEMI/K7MDENG/", "bPxRfiCYEXAMPLEKEY");

        let auth =

            sigv4("GET", "/test.txt", &[], &headers, empty, (key_id.as_str(), secret.as_str()), "us-east-1", "s3", "20130524T000000Z");

            format!(

                "AWS4-HMAC-SHA256 Credential={key_id}/20130524/us-east-1/s3/aws4_request, \

                 SignedHeaders=host;range;x-amz-content-sha256;x-amz-date, \

                 Signature=f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"

            )

    /// The history scan's budget (oxygenhub#277): what one pass over one

    /// repository may scan, and for how long, before it moves on to the

    /// next and resumes there later. A limit of 0 turns the scan off.

    pub history_scan_limit: ByteSize,

    pub history_scan_time_ms: u64,

        let history = crate::history_scan::Budget::default();

            history_scan_limit: ByteSize(history.bytes),

            history_scan_time_ms: history.time.as_millis() as u64,

//! a maintainer allowed, and the findings of recent refusals; and the

//! history scan's (migration 0086, oxygenhub#277): its findings, marked

//! resolved once the secret was revoked, and each repository's progress.

/// Refused pushes' findings kept per repository: the newest, for

/// maintainers to judge. The history scan's are all kept (one per value

/// and file).

/// A finding's source: a refused push, or the scan of history.

pub const PUSH: &str = "push";

pub const HISTORY: &str = "history";

    pub rule: String,

    pub description: String,

    pub path: String,

    pub line: i64,

    pub fingerprint: String,

    pub excerpt: String,

}

/// A finding of the history scan (#277): where it is, and whether a

/// maintainer marked it resolved.

#[derive(Clone, Debug, FromRow)]

pub struct HistoryRow {

    pub id: i64,

    pub at_ms: i64,

    pub change_hash: String,

    pub bookmark: String,

    pub resolved_ms: Option<i64>,

    pub resolved_by: String,

    pub resolved_reason: String,

        "DELETE FROM secret_findings WHERE repo_id = $1 AND source = 'push' \

         AND id NOT IN (SELECT id FROM secret_findings WHERE repo_id = $2 AND source = 'push' ORDER BY id DESC LIMIT $3)",

         WHERE repo_id = $1 AND source = 'push' \

         AND NOT EXISTS (SELECT 1 FROM secret_bypasses b WHERE b.repo_id = f.repo_id AND b.fingerprint = f.fingerprint) \

    Ok(())

}

/// Remove an allowed fingerprint (#277): pushes adding that value are

/// refused again. The row removed, if there was one.

pub async fn remove_bypass(pool: &Pool, repo_id: i64, fingerprint: &str) -> sqlx::Result<Option<Bypass>> {

    let row: Option<Bypass> = sqlx::query_as(

        "SELECT id, fingerprint, rule, reason, by_name, created_ms FROM secret_bypasses WHERE repo_id = $1 AND fingerprint = $2",

    )

    .bind(repo_id)

    .bind(fingerprint)

    .fetch_optional(pool)

    .await?;

    if let Some(b) = &row {

        sqlx::query("DELETE FROM secret_bypasses WHERE id = $1").bind(b.id).execute(pool).await?;

    }

    Ok(row)

}

/// A finding of the history scan to record.

pub struct NewHistoryFinding<'a> {

    pub finding: NewFinding<'a>,

    pub change_hash: String,

    pub bookmark: String,

}

/// Record the history scan's findings: one per value and file, so a value

/// found again (in a later change, or by a rescan) keeps its first row and

/// whether it was resolved. How many were new.

pub async fn record_history(pool: &Pool, repo_id: i64, findings: &[NewHistoryFinding<'_>]) -> sqlx::Result<usize> {

    let now = now_ms();

    let mut added = 0;

    for h in findings {

        let f = &h.finding;

        let r = sqlx::query(

            "INSERT INTO secret_findings (repo_id, at_ms, pusher, via, rule, description, path, line, fingerprint, excerpt, source, change_hash, bookmark) \

             VALUES ($1, $2, '', '', $3, $4, $5, $6, $7, $8, 'history', $9, $10) \

             ON CONFLICT (repo_id, fingerprint, path) WHERE source = 'history' DO NOTHING",

        )

        .bind(repo_id)

        .bind(now)

        .bind(f.rule)

        .bind(f.description)

        .bind(f.path)

        .bind(f.line)

        .bind(f.fingerprint)

        .bind(f.excerpt)

        .bind(&h.change_hash)

        .bind(&h.bookmark)

        .execute(pool)

        .await?;

        added += r.rows_affected() as usize;

    }

    Ok(added)

}

const HISTORY_COLUMNS: &str =

    "id, at_ms, rule, description, path, line, fingerprint, excerpt, change_hash, bookmark, resolved_ms, resolved_by, resolved_reason";

/// The history scan's findings still open: not resolved, their value not

/// allowed; oldest first, as history has them.

pub async fn history(pool: &Pool, repo_id: i64, limit: i64) -> sqlx::Result<Vec<HistoryRow>> {

    sqlx::query_as(&format!(

        "SELECT {HISTORY_COLUMNS} FROM secret_findings f WHERE repo_id = $1 AND source = 'history' AND resolved_ms IS NULL \

         AND NOT EXISTS (SELECT 1 FROM secret_bypasses b WHERE b.repo_id = f.repo_id AND b.fingerprint = f.fingerprint) \

         ORDER BY id LIMIT $2"

    ))

    .bind(repo_id)

    .bind(limit)

    .fetch_all(pool)

    .await

}

/// The history scan's findings marked resolved, newest mark first.

pub async fn resolved(pool: &Pool, repo_id: i64, limit: i64) -> sqlx::Result<Vec<HistoryRow>> {

    sqlx::query_as(&format!(

        "SELECT {HISTORY_COLUMNS} FROM secret_findings WHERE repo_id = $1 AND source = 'history' AND resolved_ms IS NOT NULL \

         ORDER BY resolved_ms DESC, id DESC LIMIT $2"

    ))

    .bind(repo_id)

    .bind(limit)

    .fetch_all(pool)

    .await

}

/// Mark history finding `id` resolved (the secret was revoked) with

/// `reason`; the finding, if it was an open one of this repository's.

pub async fn resolve(pool: &Pool, repo_id: i64, id: i64, by: &str, reason: &str) -> sqlx::Result<Option<HistoryRow>> {

    let done = sqlx::query(

        "UPDATE secret_findings SET resolved_ms = $1, resolved_by = $2, resolved_reason = $3 \

         WHERE id = $4 AND repo_id = $5 AND source = 'history' AND resolved_ms IS NULL",

    )

    .bind(now_ms())

    .bind(by)

    .bind(reason)

    .bind(id)

    .bind(repo_id)

    .execute(pool)

    .await?;

    if done.rows_affected() == 0 {

        return Ok(None);

    }

    sqlx::query_as(&format!("SELECT {HISTORY_COLUMNS} FROM secret_findings WHERE id = $1")).bind(id).fetch_optional(pool).await

}

/// A repository's history scan (#277): the rules it ran with, the heads

/// whose history is scanned (`done_heads`), the heads a scan under way

/// works toward (`target_heads`, empty when none is) and how far it got

/// (`position` of `pending` changes, in the scan's order), totals, and a

/// maintainer's request for a scan now.

#[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)]

pub struct ScanRow {

    pub repo_id: i64,

    pub rules_version: i64,

    pub done_heads: String,

    pub target_heads: String,

    pub position: i64,

    pub pending: i64,

    pub changes: i64,

    pub bytes: i64,

    pub requested_ms: Option<i64>,

    pub pass_ms: Option<i64>,

    pub finished_ms: Option<i64>,

    pub error: String,

}

const SCAN_COLUMNS: &str =

    "repo_id, rules_version, done_heads, target_heads, position, pending, changes, bytes, requested_ms, pass_ms, finished_ms, error";

pub async fn scan_of(pool: &Pool, repo_id: i64) -> sqlx::Result<Option<ScanRow>> {

    sqlx::query_as(&format!("SELECT {SCAN_COLUMNS} FROM secret_scans WHERE repo_id = $1")).bind(repo_id).fetch_optional(pool).await

}

/// Save a repository's scan progress.

pub async fn save_scan(pool: &Pool, s: &ScanRow) -> sqlx::Result<()> {

    sqlx::query(&format!(

        "INSERT INTO secret_scans ({SCAN_COLUMNS}) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \

         ON CONFLICT (repo_id) DO UPDATE SET rules_version = excluded.rules_version, done_heads = excluded.done_heads, \

         target_heads = excluded.target_heads, position = excluded.position, pending = excluded.pending, changes = excluded.changes, \

         bytes = excluded.bytes, requested_ms = excluded.requested_ms, pass_ms = excluded.pass_ms, finished_ms = excluded.finished_ms, \

         error = excluded.error"

    ))

    .bind(s.repo_id)

    .bind(s.rules_version)

    .bind(&s.done_heads)

    .bind(&s.target_heads)

    .bind(s.position)

    .bind(s.pending)

    .bind(s.changes)

    .bind(s.bytes)

    .bind(s.requested_ms)

    .bind(s.pass_ms)

    .bind(s.finished_ms)

    .bind(&s.error)

    .execute(pool)

    .await?;

}

/// A maintainer asked for the whole history to be scanned again now:

/// what was scanned is forgotten (findings and their marks are kept) and

/// the repository goes first.

pub async fn request_scan(pool: &Pool, repo_id: i64) -> sqlx::Result<()> {

    let now = now_ms();

    sqlx::query(

        "INSERT INTO secret_scans (repo_id, requested_ms) VALUES ($1, $2) \

         ON CONFLICT (repo_id) DO UPDATE SET requested_ms = excluded.requested_ms, done_heads = '', target_heads = '', position = 0, pending = 0",

    )

    .bind(repo_id)

    .bind(now)

    .execute(pool)

    .await?;

    Ok(())

}

/// Repositories whose maintainers asked for a scan, oldest request first.

pub async fn requested(pool: &Pool) -> sqlx::Result<Vec<i64>> {

    sqlx::query_scalar("SELECT repo_id FROM secret_scans WHERE requested_ms IS NOT NULL ORDER BY requested_ms").fetch_all(pool).await

pub mod history_scan;

    // Each repository's history scanned for secrets (#277), a budget at a