Hold a held-back download open so allowing it lets it through

core/Cargo.toml

@@ -90,6 +90,10 @@ path = "tests/tpm.rs"
 name = "managers"
 path = "tests/managers.rs"
 
+[[test]]
+name = "blocks"
+path = "tests/blocks.rs"
+
 [features]
 # Builds the machine-wide binaries: the Windows service and tray, and the macOS uninstaller.
 machine-wide = []

core/src/blocks.rs

@@ -1,14 +1,25 @@
+use crate::config::{self, AllowedPackage};
 use crate::policies::Package;
+use crate::Firewall;
+use rama::error::BoxError;
 use serde::Serialize;
+use std::path::Path;
 use std::sync::Mutex;
+use std::time::{Duration, Instant};
+use tokio::sync::watch;
 
-/// Bounded, so blocks pile up harmlessly while nothing collects them.
+/// So blocks can't pile up while no app collects them.
 const KEPT: usize = 50;
+/// Under pip's, pipenv's and pdm's 15 s read timeout, the shortest of the supported managers.
+const HOLD: Duration = Duration::from_secs(10);
+/// Only hold while an app is asking for blocks, so someone can answer.
+const WATCHED: Duration = Duration::from_secs(2);
+/// For a minute, so the rest of that install can fetch it too.
+const ONCE: Duration = Duration::from_secs(60);
 
-static RECENT: Mutex<Vec<BlockedPackage>> = Mutex::new(Vec::new());
-
-#[derive(Serialize)]
+#[derive(Clone, Serialize)]
 pub struct BlockedPackage {
+    pub id: u64,
     pub eco: &'static str,
     pub name: String,
     pub version: Option<String>,
@@ -16,21 +27,125 @@ pub struct BlockedPackage {
     pub reason: String,
 }
 
-pub fn record(eco: &'static str, pkg: &Package, filter: &'static str, reason: &str) {
-    let mut recent = RECENT.lock().unwrap_or_else(|e| e.into_inner());
-    if recent.len() == KEPT {
-        recent.remove(0);
+struct Blocks {
+    last_id: u64,
+    unsent: Vec<BlockedPackage>,
+    /// Age holds, the only blocks that can be allowed, each with what wakes its waiting downloads.
+    held: Vec<(BlockedPackage, watch::Sender<bool>)>,
+    polled: Option<Instant>,
+}
+
+static BLOCKS: Mutex<Blocks> =
+    Mutex::new(Blocks { last_id: 0, unsent: Vec::new(), held: Vec::new(), polled: None });
+
+/// Records a block and holds a new package's download for up to `HOLD`. Returns whether it was allowed.
+pub async fn block(eco: &'static str, pkg: &Package, filter: &'static str, reason: &str) -> bool {
+    let Some(mut answer) = record(eco, pkg, filter, reason) else {
+        return false;
+    };
+    let allowed = tokio::time::timeout(HOLD, answer.wait_for(|allowed| *allowed)).await;
+    matches!(allowed, Ok(Ok(_)))
+}
+
+/// The app's requests: `blocks`, `allow <id>` or `allow-once <id>`.
+/// An allow replies `late` if the download had stopped waiting.
+pub fn answer(firewall: &Firewall, config: &Path, request: &str) -> String {
+    if request == "blocks" {
+        return serde_json::to_string(&take()).expect("blocks serialize");
     }
-    recent.push(BlockedPackage {
+    let result = match request.split_once(' ').map(|(verb, id)| (verb, id.parse())) {
+        Some(("allow", Ok(id))) => trust(firewall, config, id),
+        Some(("allow-once", Ok(id))) => allow_once(firewall, id),
+        _ => Err(format!("unknown message {request}").into()),
+    };
+    match result {
+        Ok(true) => "ok".to_owned(),
+        Ok(false) => "late".to_owned(),
+        Err(error) => format!("error: {error}"),
+    }
+}
+
+/// Hands each block over once.
+fn take() -> Vec<BlockedPackage> {
+    let mut blocks = lock();
+    blocks.polled = Some(Instant::now());
+    std::mem::take(&mut blocks.unsent)
+}
+
+fn record(
+    eco: &'static str,
+    pkg: &Package,
+    filter: &'static str,
+    reason: &str,
+) -> Option<watch::Receiver<bool>> {
+    let mut blocks = lock();
+    let watched = blocks.polled.is_some_and(|polled| polled.elapsed() < WATCHED);
+    let holds = filter == "age" && watched;
+    // A repeat fetch of a held package waits on the same answer.
+    let waiting = blocks.held.iter().find(|(held, wake)| {
+        wake.receiver_count() > 0 && held.eco == eco && held.name == pkg.name && held.version == pkg.version
+    });
+    if let (true, Some((_, wake))) = (holds, waiting) {
+        return Some(wake.subscribe());
+    }
+    blocks.last_id += 1;
+    let block = BlockedPackage {
+        id: blocks.last_id,
         eco,
         name: pkg.name.clone(),
         version: pkg.version.clone(),
         filter,
         reason: reason.to_owned(),
-    });
+    };
+    push(&mut blocks.unsent, block.clone());
+    if filter != "age" {
+        return None;
+    }
+    let (wake, allowed) = watch::channel(false);
+    push(&mut blocks.held, (block, wake));
+    holds.then_some(allowed)
 }
 
-/// Hands each block over once.
-pub fn take() -> Vec<BlockedPackage> {
-    std::mem::take(&mut *RECENT.lock().unwrap_or_else(|e| e.into_inner()))
+/// Also saved to the config, so it survives a restart.
+fn trust(firewall: &Firewall, config: &Path, id: u64) -> Result<bool, BoxError> {
+    let block = held(id)?;
+    let version = block.version.ok_or("only a version can be trusted")?;
+    config::allow(config, block.eco, &block.name, &version)?;
+    let package = AllowedPackage { name: block.name, version: Some(version) };
+    firewall.lift_age_check(block.eco, package, None);
+    Ok(release(id))
+}
+
+fn allow_once(firewall: &Firewall, id: u64) -> Result<bool, BoxError> {
+    let block = held(id)?;
+    let package = AllowedPackage { name: block.name, version: block.version };
+    firewall.lift_age_check(block.eco, package, Some(Instant::now() + ONCE));
+    Ok(release(id))
+}
+
+fn held(id: u64) -> Result<BlockedPackage, BoxError> {
+    let blocks = lock();
+    let held = blocks.held.iter().find(|(held, _)| held.id == id);
+    Ok(held.ok_or(format!("no new package was held back as {id}"))?.0.clone())
+}
+
+/// Returns whether a download was still waiting.
+fn release(id: u64) -> bool {
+    let blocks = lock();
+    let Some((_, wake)) = blocks.held.iter().find(|(held, _)| held.id == id) else {
+        return false;
+    };
+    wake.send_replace(true);
+    wake.receiver_count() > 0
+}
+
+fn push<T>(list: &mut Vec<T>, item: T) {
+    if list.len() == KEPT {
+        list.remove(0);
+    }
+    list.push(item);
+}
+
+fn lock() -> std::sync::MutexGuard<'static, Blocks> {
+    BLOCKS.lock().unwrap_or_else(|e| e.into_inner())
 }

core/src/config/mod.rs

@@ -121,7 +121,7 @@ fn merge(base: &mut Map<String, Value>, top: Map<String, Value>) {
 }
 
 /// Adds an `allow` entry to the file, leaving the rest of it as written.
-pub fn allow(path: &Path, eco: &str, name: &str, version: Option<&str>) -> std::io::Result<()> {
+pub fn allow(path: &Path, eco: &str, name: &str, version: &str) -> std::io::Result<()> {
     let invalid = |e: String| Error::new(ErrorKind::InvalidData, format!("{}: {e}", path.display()));
     let mut config = match std::fs::read_to_string(path) {
         Ok(text) => serde_json::from_str(&text).map_err(|e| invalid(e.to_string()))?,

core/src/firewall.rs

@@ -1,4 +1,4 @@
-use crate::config::FirewallConfig;
+use crate::config::{AllowedPackage, FirewallConfig};
 use crate::policies::filters::age::Age;
 use crate::policies::filters::denylist::Denylist;
 use crate::policies::filters::osv::Osv;
@@ -6,12 +6,16 @@ use crate::policies::filters::Filter;
 use crate::policies::registries::*;
 use regex::Regex;
 use std::collections::HashMap;
+use std::sync::Mutex;
+use std::time::Instant;
 
 pub struct Firewall {
     pub config: FirewallConfig,
     registries: Vec<&'static Registry>,
     patterns: HashMap<&'static str, Vec<Regex>>,
     filters: Vec<Box<dyn Filter>>,
+    /// Each lift expires at its time, or lasts until the firewall stops. Never saved.
+    age_lifted: Mutex<Vec<(String, AllowedPackage, Option<Instant>)>>,
 }
 
 impl Default for Firewall {
@@ -34,7 +38,7 @@ impl Firewall {
         if config.min_age_hours > 0 {
             filters.push(Box::new(Age::new(config.min_age_hours)));
         }
-        Firewall { config, registries, patterns, filters }
+        Firewall { config, registries, patterns, filters, age_lifted: Mutex::default() }
     }
 
     pub fn match_on<'a>(
@@ -72,6 +76,19 @@ impl Firewall {
     pub fn filters(&self) -> &[Box<dyn Filter>] {
         &self.filters
     }
+
+    pub fn lift_age_check(&self, eco: &str, package: AllowedPackage, until: Option<Instant>) {
+        let mut lifted = self.age_lifted.lock().unwrap_or_else(|e| e.into_inner());
+        lifted.retain(|(_, _, until)| until.is_none_or(|until| until > Instant::now()));
+        lifted.push((eco.to_owned(), package, until));
+    }
+
+    pub fn age_check_lifted(&self, eco: &str, matches: impl Fn(&AllowedPackage) -> bool) -> bool {
+        let lifted = self.age_lifted.lock().unwrap_or_else(|e| e.into_inner());
+        lifted.iter().any(|(e, package, until)| {
+            e == eco && until.is_none_or(|until| until > Instant::now()) && matches(package)
+        })
+    }
 }
 
 /// The built-ins, then a copy of one per mirror. Leaked, as a `Registry` is `'static` everywhere.

core/src/os/macos/mod.rs

@@ -4,7 +4,7 @@ mod uninstall;
 
 pub use uninstall::uninstall;
 
-use crate::config::{self, FirewallConfig};
+use crate::config::FirewallConfig;
 use crate::{blocks, mitm, telemetry};
 use crate::Firewall;
 use rama::bytes::Bytes;
@@ -22,7 +22,6 @@ use rama::net::apple::networkextension::{transparent_proxy_ffi, NwTcpStream, Tcp
 use rama::rt::Executor;
 use rama::telemetry::tracing::apple::oslog::{OsLogLayer, Privacy};
 use rama::{Layer, Service};
-use serde::Deserialize;
 use std::convert::Infallible;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -40,6 +39,7 @@ const CONFIG_FILE: &str = "/Library/Application Support/Firewall/config.json";
 struct Handler {
     ca: Arc<mitm::CertificateAuthority>,
     hosts: Vec<&'static str>,
+    firewall: Arc<Firewall>,
     service: mitm::MitmService,
 }
 
@@ -59,12 +59,7 @@ impl TransparentProxyHandler for Handler {
         let message = String::from_utf8_lossy(&message).into_owned();
         Some(Bytes::from(match message.as_str() {
             "ca" => self.ca.pem.clone(),
-            "blocks" => serde_json::to_string(&blocks::take()).expect("blocks serialize"),
-            other => match other.strip_prefix("allow ").map(allow) {
-                Some(Ok(())) => "ok".to_owned(),
-                Some(Err(error)) => format!("error: {error}"),
-                None => format!("error: unknown message {other}"),
-            },
+            other => blocks::answer(&self.firewall, CONFIG_FILE.as_ref(), other),
         }))
     }
 
@@ -119,22 +114,8 @@ fn handler(exec: Executor) -> Result<Handler, BoxError> {
     let signer: mitm::CaSigner = Arc::new(key);
     let ca = install::certificate_authority(&hosts, &*signer);
     let ca = Arc::new(ca.inspect_err(|_| telemetry::error("ca"))?);
-    let service = mitm::mitm_service(firewall, &ca, signer, None, exec);
-    Ok(Handler { ca, hosts, service })
-}
-
-#[derive(Deserialize)]
-struct AllowRequest {
-    eco: String,
-    name: String,
-    version: Option<String>,
-}
-
-/// Takes effect when the provider restarts, which rebuilds it from the config.
-fn allow(request: &str) -> Result<(), BoxError> {
-    let request: AllowRequest = serde_json::from_str(request)?;
-    config::allow(CONFIG_FILE.as_ref(), &request.eco, &request.name, request.version.as_deref())?;
-    Ok(())
+    let service = mitm::mitm_service(firewall.clone(), &ca, signer, None, exec);
+    Ok(Handler { ca, hosts, firewall, service })
 }
 
 /// stderr is invisible in a system extension, so events go to the unified log:

core/src/policies/filters/mod.rs

@@ -3,6 +3,7 @@ pub mod denylist;
 pub mod osv;
 
 use super::registries::{Package, Registry};
+use crate::config::AllowedPackage;
 use crate::Firewall;
 use std::future::Future;
 use std::pin::Pin;
@@ -31,10 +32,11 @@ pub async fn consult(
     registry: &'static Registry,
     pkg: &Package,
 ) -> Option<(&'static str, Verdict)> {
-    let allowed = allowed(firewall, registry, pkg);
+    let lifted = allowed(firewall, registry, pkg)
+        || firewall.age_check_lifted(registry.eco, |entry| matches(registry, pkg, entry));
     let mut unsure = None;
     for filter in firewall.filters() {
-        if allowed && filter.name() == "age" {
+        if lifted && filter.name() == "age" {
             continue;
         }
         match filter.check(registry, pkg).await {
@@ -51,13 +53,15 @@ fn allowed(firewall: &Firewall, registry: &Registry, pkg: &Package) -> bool {
     let Some(entries) = firewall.config.allow.get(registry.eco) else {
         return false;
     };
-    entries.iter().any(|entry| {
-        let name_matches = match registry.normalize {
-            Some(normalize) => normalize(&entry.name) == pkg.name,
-            None => entry.name == pkg.name,
-        };
-        // Go writes a leading v that the parsed version drops, and PEP 440 ignores one.
-        let version = entry.version.as_deref().map(|v| v.strip_prefix('v').unwrap_or(v));
-        name_matches && (version.is_none() || version == pkg.version.as_deref())
-    })
+    entries.iter().any(|entry| matches(registry, pkg, entry))
+}
+
+fn matches(registry: &Registry, pkg: &Package, entry: &AllowedPackage) -> bool {
+    let name_matches = match registry.normalize {
+        Some(normalize) => normalize(&entry.name) == pkg.name,
+        None => entry.name == pkg.name,
+    };
+    // Go writes a leading v that the parsed version drops, and PEP 440 ignores one.
+    let version = entry.version.as_deref().map(|v| v.strip_prefix('v').unwrap_or(v));
+    name_matches && (version.is_none() || version == pkg.version.as_deref())
 }

core/src/policies/mod.rs

@@ -70,7 +70,10 @@ pub async fn screen(url: &str, firewall: &Firewall) -> Option<String> {
     info!(eco = registry.eco, pkg = %pkg.label(), %url, "screening");
     match consult(firewall, registry, &pkg).await {
         Some((filter, Verdict::Deny(reason))) => {
-            crate::blocks::record(registry.eco, &pkg, filter, &reason);
+            if crate::blocks::block(registry.eco, &pkg, filter, &reason).await {
+                info!(eco = registry.eco, pkg = %pkg.label(), "allowed while held");
+                return None;
+            }
             Some(format!("{} {reason}", pkg.label()))
         }
         Some((filter, Verdict::Unknown(e))) if !firewall.config.fail_open => {

core/tests/blocks.rs

@@ -0,0 +1,27 @@
+use firewall::blocks::{answer, block};
+use firewall::policies::Package;
+use firewall::Firewall;
+use std::path::Path;
+use std::time::Duration;
+
+#[tokio::test]
+async fn allowing_once_releases_the_held_download() {
+    let firewall = Firewall::default();
+    let config = Path::new("/nonexistent/config.json");
+    let pkg = Package { name: "left-pad".into(), version: Some("1.3.0".into()) };
+    assert!(!block("npm", &pkg, "age", "is new").await, "held while nothing was asking");
+    answer(&firewall, config, "blocks");
+
+    let held = tokio::spawn(async move { block("npm", &pkg, "age", "is new").await });
+    let id = loop {
+        let blocks: serde_json::Value = serde_json::from_str(&answer(&firewall, config, "blocks")).unwrap();
+        if let Some(id) = blocks[0]["id"].as_u64() {
+            break id;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    };
+    assert_eq!(answer(&firewall, config, &format!("allow-once {id}")), "ok");
+    assert!(held.await.unwrap(), "the waiting download goes through");
+    assert!(firewall.age_check_lifted("npm", |p| p.name == "left-pad"), "and the rest of the install");
+    assert!(!block("npm", &Package { name: "evil".into(), version: None }, "denylist", "is malware").await);
+}

core/tests/config.rs

@@ -1,4 +1,4 @@
-use firewall::config::path;
+use firewall::config::{path, AllowedPackage};
 use firewall::policies::filters::consult;
 use firewall::policies::{screen, Package};
 use firewall::{Firewall, FirewallConfig};
@@ -95,12 +95,23 @@ async fn failing_closed_blocks_what_could_not_be_checked() {
     assert!(blocked.contains("could not be checked"), "{blocked:?}");
 }
 
+#[tokio::test]
+async fn lifting_the_age_check_still_checks_for_malware() {
+    let config = r#"{"denylist": {"base": "http://127.0.0.1:1"}}"#;
+    let firewall = Firewall::new(serde_json::from_str(config).expect("parses"));
+    let (registry, _) = firewall.match_on("/go/golang.org/x/net/@v/list", |r| r.prefix).expect("go");
+    let pkg = Package { name: "golang.org/x/net".into(), version: Some("0.20.0".into()) };
+    firewall.lift_age_check("go", AllowedPackage { name: pkg.name.clone(), version: None }, None);
+    let (filter, _) = consult(&firewall, registry, &pkg).await.expect("the denylist still runs");
+    assert_eq!(filter, "denylist");
+}
+
 #[test]
 fn allowing_keeps_the_rest_of_the_file() {
     let path = std::env::temp_dir().join(format!("firewall-allow-{}.json", std::process::id()));
     std::fs::write(&path, r#"{"fail_open": false}"#).expect("write");
     for _ in 0..2 {
-        firewall::config::allow(&path, "npm", "left-pad", Some("1.3.0")).expect("allow");
+        firewall::config::allow(&path, "npm", "left-pad", "1.3.0").expect("allow");
     }
     let text = std::fs::read_to_string(&path).expect("read");
     std::fs::remove_file(&path).ok();