From ee13ae44fa00fdee2c38515ccccee7ffbf55bea1 Mon Sep 17 00:00:00 2001 From: Nils Pukropp Date: Tue, 20 Oct 2026 02:01:50 +0200 Subject: [PATCH] runtime: ranges, health, logs, supervised loops, strict config Media answers Range requests with 206 and Content-Range, so videos seek and Safari plays them. Sealed files too: the vault's opener now decrypts by record position (STREAM BE32 through StreamPrimitive, the same construction and the same files), so a range starts at the record that holds its first byte. Compression is limited to text, because a gzipped body no longer matches the byte ranges a player asks for. GET /api/health answers without a session once the database does, and the compose healthcheck asks it instead of only opening the port. Requests are logged at info with share tokens and keys left out of the path and no query strings. A request that is not answered in 30 seconds gets a 503, except uploads, imports, the export, re-sealing and backups. The backup, mail, sweep and SFTP loops run under a supervisor that logs a panic and starts the loop again. The backup loop starts without a Proton session and picks up a later proton-login without a restart. A lock file keeps backup-now and the server from mirroring at once, and an unreadable session key file turns backups off instead of failing boot. Numbers in the configuration are checked against bounds at startup, so a session length or an upload limit can no longer overflow later, and a switch that is neither on nor off is an error rather than off. Config and MailConfig print with the password, secret and SMTP URL redacted. A session epoch is signed into each cookie (migration 0014). :logout! advances it, which ends every session, and a changed WORKSPACE_PASSWORD advances it on the next start. Cookies issued before this keep working in the first epoch. --- Cargo.toml | 4 + README.md | 25 ++++- compose.yml | 5 +- migrations/0014_session_epoch.sql | 10 ++ src/auth.rs | 97 +++++++++++++++--- src/backup/mod.rs | 129 +++++++++++++++++------- src/config.rs | 159 +++++++++++++++++++++--------- src/main.rs | 125 +++++++++++++++++++++-- src/notify/mod.rs | 21 ++-- src/routes/media.rs | 154 ++++++++++++++++++++++++----- src/routes/mod.rs | 1 + src/routes/session.rs | 38 ++++++- src/routes/share.rs | 3 +- src/state.rs | 5 +- src/sweep.rs | 11 ++- src/tasks.rs | 63 ++++++++++++ src/vault.rs | 81 +++++++++++---- web/src/lib/api.ts | 4 +- web/src/lib/commands.ts | 4 + web/src/lib/store.svelte.ts | 4 +- 20 files changed, 771 insertions(+), 172 deletions(-) create mode 100644 migrations/0014_session_epoch.sql create mode 100644 src/tasks.rs diff --git a/Cargo.toml b/Cargo.toml index 5403e06..eaaee71 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,6 +62,10 @@ russh-sftp = "3" # speaks rand_core 0.10, one major version past the `rand` used everywhere else. rand10 = { package = "rand", version = "0.10" } +[dev-dependencies] +# A paused clock, so tests of timed loops do not wait in real time. +tokio = { version = "1", features = ["test-util"] } + [features] default = [] # Keep the Proton session in the OS keyring instead of a 0600 file. Needs a diff --git a/README.md b/README.md index 3206cac..84a5398 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,11 @@ spelling still accepted behind it, so a server configured before the rename keeps running untouched. `WORKSPACE_SECRET` signs the session cookie; generate one with -`openssl rand -base64 32`. Changing it logs every device out. The database and +`openssl rand -base64 32`. Changing it logs every device out, and so does +changing `WORKSPACE_PASSWORD` or `:logout!`, which ends every session but +needs no restart. A setting that is set but cannot be read — a number out of +range, a switch that is neither on nor off — stops the start with the reason, +rather than being taken for its default. The database and uploads live under `WORKSPACE_DATA_DIR` (`./data` by default) — that directory is the entire backup, and it is exactly what the Proton Drive mirror copies. @@ -445,11 +449,14 @@ both halves. ## API -Everything except the two share routes requires the session cookie. +Everything except the login, the health check and the two share routes +requires the session cookie. Media answers `Range` requests, so a video seeks, +sealed or not. | method | path | purpose | | --- | --- | --- | -| `POST` | `/api/login`, `/api/logout` | session | +| `GET` | `/api/health` | public: up, and the database answers — for a supervisor | +| `POST` | `/api/login`, `/api/logout` | session; `/api/logout?everywhere=true` ends every one | | `GET` | `/api/me` | current user | | `GET`/`POST` | `/api/spaces` | the spaces / create one | | `GET`/`POST` | `/api/nodes` | children (`?parent=`) or search (`?q=`, `?space=`) / create | @@ -494,7 +501,17 @@ so a proxy must pass the `Host` header through unchanged (traefik and nginx's `proxy_set_header Host $host` both do). Only the upload routes accept a body as large as `WORKSPACE_MAX_UPLOAD_MB`; everything else stops at 16 MiB, and the login at 16 KiB, before anyone has proved who they are. `SIGTERM` gets in-flight -requests eight seconds to finish, inside the ten `docker stop` waits. +requests eight seconds to finish, inside the ten `docker stop` waits. Any +other request that has not been answered after 30 seconds is given up on with +a `503`, except the ones that move files around: uploads, imports, the export, +re-sealing and a backup run. + +Every request is logged at `info`, path and status and time, never a query +string, and with a share link's token and key left out. A background loop that +panics — the mirror, the mail, the sweep, the SFTP server — is logged and +started again rather than lost until the next restart. If +`WORKSPACE_SESSION_KEY_FILE` cannot be read the server starts anyway with +backups off, and `:backup` says why. ### Docker diff --git a/compose.yml b/compose.yml index b54badf..fac9043 100644 --- a/compose.yml +++ b/compose.yml @@ -28,8 +28,9 @@ services: # rather than relying on the reverse proxy — see compose.override.yml.example. # - "127.0.0.1:2222:2222" healthcheck: - # No extra tooling in the image, so probe the port with bash directly. - test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/4242"] + # The image has bash and nothing like curl: a request by hand, and a + # 200 from /api/health, which also asks the database. + test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/4242 && printf 'GET /api/health HTTP/1.0\r\n\r\n' >&3 && head -1 <&3 | grep -q ' 200 '"] interval: 30s timeout: 3s start_period: 5s diff --git a/migrations/0014_session_epoch.sql b/migrations/0014_session_epoch.sql new file mode 100644 index 0000000..0c0a9f3 --- /dev/null +++ b/migrations/0014_session_epoch.sql @@ -0,0 +1,10 @@ +-- Sessions are signed cookies, not rows: nothing on the server lists them, so +-- nothing can revoke one. Signing a number into every cookie that has to match +-- this one gives a single lever instead — raise it, and every cookie issued +-- before is refused. It is raised by `:logout!` and whenever the login +-- password changes, which is recognised by a keyed hash of it kept here. +CREATE TABLE auth ( + id INTEGER PRIMARY KEY CHECK (id = 1), + epoch INTEGER NOT NULL, + password_hash TEXT NOT NULL +); diff --git a/src/auth.rs b/src/auth.rs index 5dede65..f019fe5 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -28,14 +28,16 @@ fn sign(secret: &[u8], payload: &str) -> String { /// A session token is `base64(expiry_unix_seconds).base64(hmac)`. There is no /// server-side session store: rotating `WORKSPACE_SECRET` invalidates every session. -pub fn issue_token(config: &Config) -> String { +/// A cookie is good until it expires and for as long as the epoch it was +/// issued under is the current one. +pub fn issue_token(config: &Config, epoch: i64) -> String { let expires = crate::now() + config.session_days * 86_400; - let payload = B64.encode(expires.to_string()); + let payload = B64.encode(format!("{expires}:{epoch}")); let signature = sign(&config.secret, &payload); format!("{payload}.{signature}") } -pub fn verify_token(config: &Config, token: &str) -> bool { +pub fn verify_token(config: &Config, token: &str, epoch: i64) -> bool { let Some((payload, signature)) = token.split_once('.') else { return false; }; @@ -46,10 +48,61 @@ pub fn verify_token(config: &Config, token: &str) -> bool { let Ok(decoded) = B64.decode(payload) else { return false; }; - let Ok(expires) = String::from_utf8_lossy(&decoded).parse::() else { + let decoded = String::from_utf8_lossy(&decoded); + // A cookie from before epochs existed carries only the expiry, and + // belongs to the first epoch. + let (expires, issued_in) = decoded.split_once(':').unwrap_or((&decoded, "0")); + let (Ok(expires), Ok(issued_in)) = (expires.parse::(), issued_in.parse::()) else { return false; }; - expires > crate::now() + expires > crate::now() && issued_in == epoch +} + +/// The session epoch: what every cookie must have been signed under. +pub struct Epoch; + +impl Epoch { + /// A keyed hash of the login password, so a change to it can be told + /// from a restart without the password being written down anywhere. + fn password_hash(config: &Config) -> String { + sign(&config.secret, &format!("password:{}", config.password)) + } + + /// The epoch to start with: the stored one, raised when the password + /// is not the one it was stored with. + pub async fn load(db: &sqlx::SqlitePool, config: &Config) -> anyhow::Result { + let hash = Self::password_hash(config); + let stored: Option<(i64, String)> = + sqlx::query_as("SELECT epoch, password_hash FROM auth WHERE id = 1") + .fetch_optional(db) + .await?; + let epoch = match stored { + None => 0, + Some((epoch, previous)) if previous == hash => return Ok(epoch), + Some((epoch, _)) => { + tracing::info!("the login password changed: every device is signed out"); + epoch + 1 + } + }; + sqlx::query( + "INSERT INTO auth (id, epoch, password_hash) VALUES (1, ?1, ?2) + ON CONFLICT (id) DO UPDATE SET epoch = excluded.epoch, password_hash = excluded.password_hash", + ) + .bind(epoch) + .bind(&hash) + .execute(db) + .await?; + Ok(epoch) + } + + /// Sign every device out: the epoch moves on, and no cookie has it yet. + pub async fn advance(state: &AppState) -> anyhow::Result { + let epoch: i64 = sqlx::query_scalar("UPDATE auth SET epoch = epoch + 1 WHERE id = 1 RETURNING epoch") + .fetch_one(&state.db) + .await?; + state.epoch.store(epoch, std::sync::atomic::Ordering::SeqCst); + Ok(epoch) + } } pub fn credentials_match(config: &Config, username: &str, password: &str) -> bool { @@ -94,7 +147,8 @@ impl FromRequestParts for Session { state: &AppState, ) -> Result { let token = cookie_value(parts, COOKIE_NAME).ok_or(AppError::Unauthorized)?; - if verify_token(&state.config, token) { + let epoch = state.epoch.load(std::sync::atomic::Ordering::SeqCst); + if verify_token(&state.config, token, epoch) { Ok(Session) } else { Err(AppError::Unauthorized) @@ -199,45 +253,45 @@ mod tests { #[test] fn a_fresh_token_verifies() { let config = config(30); - assert!(verify_token(&config, &issue_token(&config))); + assert!(verify_token(&config, &issue_token(&config, 0), 0)); } #[test] fn an_expired_token_does_not() { let config = config(-1); - assert!(!verify_token(&config, &issue_token(&config))); + assert!(!verify_token(&config, &issue_token(&config, 0), 0)); } #[test] fn rotating_the_secret_invalidates_every_session() { - let token = issue_token(&config(30)); + let token = issue_token(&config(30), 0); let mut rotated = config(30); rotated.secret = b"a-different-secret-entirely".to_vec(); - assert!(!verify_token(&rotated, &token)); + assert!(!verify_token(&rotated, &token, 0)); } #[test] fn a_tampered_token_does_not_verify() { let config = config(30); - let token = issue_token(&config); + let token = issue_token(&config, 0); let (payload, signature) = token.split_once('.').unwrap(); // A far-future expiry the holder signed themselves. let forged = base64::Engine::encode(&super::B64, "99999999999"); - assert!(!verify_token(&config, &format!("{forged}.{signature}"))); + assert!(!verify_token(&config, &format!("{forged}.{signature}"), 0)); // The real expiry with the last byte of the signature flipped. let mut bytes = signature.as_bytes().to_vec(); bytes[0] = if bytes[0] == b'A' { b'B' } else { b'A' }; let flipped = String::from_utf8(bytes).unwrap(); - assert!(!verify_token(&config, &format!("{payload}.{flipped}"))); + assert!(!verify_token(&config, &format!("{payload}.{flipped}"), 0)); } #[test] fn malformed_tokens_are_rejected_rather_than_panicking() { let config = config(30); for token in ["", ".", "no-dot", "!!!.!!!", "a.b.c", &".".repeat(100)] { - assert!(!verify_token(&config, token), "{token:?}"); + assert!(!verify_token(&config, token, 0), "{token:?}"); } } @@ -273,4 +327,19 @@ mod tests { // curl, the CLI, a script with its own credentials. assert!(is_same_origin(&headers(&[host]))); } + + #[test] + fn a_cookie_from_an_earlier_epoch_is_refused() { + let config = config(30); + let token = issue_token(&config, 3); + assert!(verify_token(&config, &token, 3)); + assert!(!verify_token(&config, &token, 4)); + + // Issued before epochs: only the expiry, and good in the first one. + let expires = crate::now() + 3600; + let payload = base64::Engine::encode(&super::B64, expires.to_string()); + let legacy = format!("{payload}.{}", super::sign(&config.secret, &payload)); + assert!(verify_token(&config, &legacy, 0)); + assert!(!verify_token(&config, &legacy, 1)); + } } diff --git a/src/backup/mod.rs b/src/backup/mod.rs index 2aa9fb7..1c0809f 100644 --- a/src/backup/mod.rs +++ b/src/backup/mod.rs @@ -62,16 +62,34 @@ pub struct Backup { /// Held for the duration of a run, so a manual backup and the timer cannot /// upload the same snapshot twice at once. running: tokio::sync::Mutex<()>, + /// Why backups cannot run at all — a session key file that is configured + /// but unreadable — when that is so. The server still starts; the mirror + /// says why it is off, every time it is asked. + disabled: Option, } impl Backup { pub fn new(db: SqlitePool, config: Arc) -> Result> { let store = SessionStore::open(&config)?; - let configured = matches!(store.load(), Ok(Some(_))); - Ok(Arc::new(Self { + Ok(Self::with_store(db, config, store, None)) + } + + /// A mirror that is off for `reason`, for a server that should start + /// anyway: backups are worth a loud error, not an outage of the workspace. + pub fn unavailable(db: SqlitePool, config: Arc, reason: String) -> Arc { + let store = SessionStore::new(config.proton_session_path()); + let backup = Self::with_store(db, config, store, Some(reason.clone())); + backup.set(|s| s.last_error = Some(reason)); + backup + } + + fn with_store(db: SqlitePool, config: Arc, store: SessionStore, disabled: Option) -> Arc { + let configured = disabled.is_none() && matches!(store.load(), Ok(Some(_))); + Arc::new(Self { db, config, store, + disabled, wake: Notify::new(), // The workspace may have been written to while this process was down, // and a restart is the cheapest moment to find out. @@ -81,11 +99,11 @@ impl Backup { ..Status::default() }), running: tokio::sync::Mutex::new(()), - })) + }) } pub fn status(&self) -> Status { - let mut status = self.status.lock().expect("backup status mutex poisoned").clone(); + let mut status = self.status.lock().unwrap_or_else(|e| e.into_inner()).clone(); status.pending = self.dirty.load(Ordering::Relaxed); status } @@ -100,7 +118,20 @@ impl Backup { /// session has been stored — asking an unconfigured backup to run is a /// no-op, not a failure. pub async fn run_now(&self) -> Result> { + if let Some(reason) = &self.disabled { + anyhow::bail!("backups are off: {reason}"); + } let _guard = self.running.lock().await; + // The same, across processes: `backup-now` beside a running server + // would otherwise share the snapshot file and the bookkeeping with it. + let _lock = match self.lock_data_dir() { + Ok(lock) => lock, + Err(e) => { + let message = format!("{e:#}"); + self.set(move |s| s.last_error = Some(message)); + return Err(e); + } + }; self.dirty.store(false, Ordering::Relaxed); self.set(|s| { s.running = true; @@ -145,6 +176,23 @@ impl Backup { outcome } + fn lock_data_dir(&self) -> Result { + let path = self.config.data_dir.join(".backup.lock"); + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&path) + .with_context(|| format!("could not open {}", path.display()))?; + match file.try_lock() { + Ok(()) => Ok(file), + Err(std::fs::TryLockError::WouldBlock) => { + anyhow::bail!("another backup of this data directory is running — `backup-now` and the server take turns") + } + Err(std::fs::TryLockError::Error(e)) => Err(e).context("could not lock the data directory for a backup"), + } + } + async fn mirror(&self) -> Result> { let Some(client) = proton::resume(&self.store).await? else { return Ok(None); @@ -285,50 +333,63 @@ impl Backup { } fn set(&self, edit: impl FnOnce(&mut Status)) { - let mut status = self.status.lock().expect("backup status mutex poisoned"); + let mut status = self.status.lock().unwrap_or_else(|e| e.into_inner()); edit(&mut status); } } -/// Run the mirror in the background for as long as the server lives. +/// Run the mirror in the background for as long as the server lives. With +/// no Proton session yet it waits for one: `proton-login` is a separate +/// process, and the server picks its session up without a restart. pub fn spawn(backup: Arc) { if backup.store.load().ok().flatten().is_none() { tracing::info!( "Proton Drive backups are not configured — run `narl-workspace proton-login` to enable them" ); + } + crate::tasks::supervise("the backup", move || mirror_forever(Arc::clone(&backup))); +} + +async fn mirror_forever(backup: Arc) { + if backup.disabled.is_some() { return; } + let interval = backup.config.backup.interval; + let debounce = backup.config.backup.debounce; - tokio::spawn(async move { - let interval = backup.config.backup.interval; - let debounce = backup.config.backup.debounce; - - loop { - // Nothing to do: wait for a change, or for the backstop timer. - while !backup.dirty.load(Ordering::Relaxed) { - match interval { - Some(interval) => { - tokio::select! { - _ = backup.wake.notified() => {} - _ = tokio::time::sleep(interval) => break, - } - } - None => backup.wake.notified().await, - } + loop { + if backup.store.load().ok().flatten().is_none() { + tokio::select! { + _ = backup.wake.notified() => {} + _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {} } - - // Let the workspace fall quiet. Every further change restarts the wait, - // so a writing session is mirrored once, when it is over. - loop { - tokio::select! { - _ = backup.wake.notified() => continue, - _ = tokio::time::sleep(debounce) => break, - } - } - - let _ = backup.run_now().await; + continue; } - }); + + // Nothing to do: wait for a change, or for the backstop timer. + while !backup.dirty.load(Ordering::Relaxed) { + match interval { + Some(interval) => { + tokio::select! { + _ = backup.wake.notified() => {} + _ = tokio::time::sleep(interval) => break, + } + } + None => backup.wake.notified().await, + } + } + + // Let the workspace fall quiet. Every further change restarts the wait, + // so a writing session is mirrored once, when it is over. + loop { + tokio::select! { + _ = backup.wake.notified() => continue, + _ = tokio::time::sleep(debounce) => break, + } + } + + let _ = backup.run_now().await; + } } /// One-shot mirror for `narl-workspace backup-now`, without starting a server. diff --git a/src/config.rs b/src/config.rs index e1674b2..49d105d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -6,7 +6,7 @@ use chrono_tz::Tz; /// Runtime configuration. Everything comes from the environment (`.env` is /// loaded on startup), because this is a deliberately single-user application. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct Config { pub username: String, pub password: String, @@ -23,6 +23,61 @@ pub struct Config { pub sftp: SftpConfig, } +/// Written by hand so that a `{:?}` anywhere — a log line, a panic — never +/// prints the password, the cookie secret or the relay's credentials. +impl std::fmt::Debug for Config { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Config") + .field("username", &self.username) + .field("password", &"") + .field("secret", &"") + .field("bind", &self.bind) + .field("data_dir", &self.data_dir) + .field("session_days", &self.session_days) + .field("max_upload_bytes", &self.max_upload_bytes) + .field("secure_cookie", &self.secure_cookie) + .field("vault_idle", &self.vault_idle) + .field("backup", &self.backup) + .field("mail", &self.mail) + .field("sftp", &self.sftp) + .finish() + } +} + +/// A whole number from the environment, `default` when it is not set, and an +/// error at startup — not an overflow later — when it is outside `allowed`. +fn number(key: &str, default: u64, allowed: std::ops::RangeInclusive) -> Result { + let Some(raw) = var(key) else { + return Ok(default); + }; + let value: u64 = raw + .trim() + .parse() + .with_context(|| format!("WORKSPACE_{key} must be a whole number, not {raw:?}"))?; + if !allowed.contains(&value) { + anyhow::bail!( + "WORKSPACE_{key} must be between {} and {}, not {value}", + allowed.start(), + allowed.end() + ); + } + Ok(value) +} + +/// A switch. Anything that is not clearly on or clearly off is an error: +/// `WORKSPACE_SECURE_COOKIE=yes` quietly meaning "no" is how a cookie ends up +/// going out over plain HTTP. +fn flag(key: &str) -> Result { + let Some(raw) = var(key) else { + return Ok(false); + }; + match raw.trim().to_ascii_lowercase().as_str() { + "1" | "true" | "yes" | "on" => Ok(true), + "0" | "false" | "no" | "off" => Ok(false), + _ => anyhow::bail!("WORKSPACE_{key} must be 1 or 0 (true/false, yes/no, on/off), not {raw:?}"), + } +} + /// The SFTP mount. Dormant until `WORKSPACE_SFTP_BIND` is set — same shape as /// the mail relay, because a workspace with no key registered yet should not /// refuse to start over a feature nobody has turned on. @@ -51,7 +106,7 @@ pub struct TimeOfDay { /// Notification mail. Dormant until `WORKSPACE_SMTP_URL` is set — the same /// dormant-until-configured shape the Proton mirror has, because a workspace /// that refuses to start over an unconfigured mail relay would be a bad trade. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct MailConfig { /// `smtp://host:587` or `smtps://user:pass@host:465`. `None` turns every /// notification off, including the queue that would otherwise pile up. @@ -71,6 +126,22 @@ pub struct MailConfig { pub kinds: Vec, } +impl std::fmt::Debug for MailConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("MailConfig") + // The URL carries the relay's user and password. + .field("url", &self.url.as_ref().map(|_| "")) + .field("from", &self.from) + .field("to", &self.to) + .field("timezone", &self.timezone) + .field("reminder_at", &self.reminder_at) + .field("card_due_at", &self.card_due_at) + .field("digest_at", &self.digest_at) + .field("kinds", &self.kinds) + .finish() + } +} + impl MailConfig { /// Whether mail is configured at all. pub fn enabled(&self) -> bool { @@ -190,32 +261,16 @@ impl Config { let data_dir = PathBuf::from(var("DATA_DIR").unwrap_or_else(|| "./data".into())); - let session_days = var("SESSION_DAYS") - .map(|v| v.parse()) - .transpose() - .context("WORKSPACE_SESSION_DAYS must be a number")? - .unwrap_or(30); - - let max_upload_bytes = var("MAX_UPLOAD_MB") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_MAX_UPLOAD_MB must be a number")? - .unwrap_or(1024) - * 1024 - * 1024; - - let vault_idle = std::time::Duration::from_secs( - var("VAULT_IDLE_MIN") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_VAULT_IDLE_MIN must be a number")? - .unwrap_or(15) - * 60, - ); - - let secure_cookie = var("SECURE_COOKIE") - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) - .unwrap_or(false); + // Ten years at most: a session is a cookie, and the expiry is + // arithmetic on it that must not overflow. + let session_days = number("SESSION_DAYS", 30, 1..=3650)? as i64; + // Up to a terabyte, in bytes. + let max_upload_bytes = usize::try_from(number("MAX_UPLOAD_MB", 1024, 1..=1024 * 1024)? * 1024 * 1024) + .context("WORKSPACE_MAX_UPLOAD_MB is more than this machine can address")?; + // A vault that locks the moment it opens could not be used; one that + // stays open for more than a day is not much of a vault. + let vault_idle = Duration::from_secs(number("VAULT_IDLE_MIN", 15, 1..=24 * 60)? * 60); + let secure_cookie = flag("SECURE_COOKIE")?; Ok(Self { username, @@ -311,24 +366,11 @@ impl BackupConfig { // second device and a fresh upload of everything in it. let device_name = var("PROTON_DEVICE").unwrap_or_else(|| "narl-diary".to_string()); - let minutes = var("BACKUP_INTERVAL_MIN") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_BACKUP_INTERVAL_MIN must be a number of minutes")? - .unwrap_or(60); + // 0 turns the backstop off; a week is the longest it waits. + let minutes = number("BACKUP_INTERVAL_MIN", 60, 0..=7 * 24 * 60)?; let interval = (minutes > 0).then(|| Duration::from_secs(minutes * 60)); - - let debounce = Duration::from_secs( - var("BACKUP_DEBOUNCE_SEC") - .map(|v| v.parse::()) - .transpose() - .context("WORKSPACE_BACKUP_DEBOUNCE_SEC must be a number of seconds")? - .unwrap_or(300), - ); - - let prune = var("BACKUP_PRUNE") - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) - .unwrap_or(false); + let debounce = Duration::from_secs(number("BACKUP_DEBOUNCE_SEC", 300, 0..=24 * 60 * 60)?); + let prune = flag("BACKUP_PRUNE")?; Ok(Self { device_name, @@ -342,9 +384,34 @@ impl BackupConfig { #[cfg(test)] mod tests { - use super::{parse_digest, parse_time, TimeOfDay}; + use super::{flag, number, parse_digest, parse_time, TimeOfDay}; use chrono::Weekday; + /// Each case under a name of its own: tests run in parallel and share + /// one environment. + fn with(key: &str, value: &str) -> String { + std::env::set_var(format!("WORKSPACE_{key}"), value); + key.to_string() + } + + #[test] + fn numbers_out_of_bounds_stop_the_start() { + assert_eq!(number("TEST_UNSET_NUMBER", 7, 1..=10).unwrap(), 7); + assert_eq!(number(&with("TEST_NUMBER_OK", " 9 "), 7, 1..=10).unwrap(), 9); + assert!(number(&with("TEST_NUMBER_ZERO", "0"), 7, 1..=10).is_err()); + assert!(number(&with("TEST_NUMBER_HUGE", "99999999999999999999"), 7, 1..=10).is_err()); + assert!(number(&with("TEST_NUMBER_NEGATIVE", "-1"), 7, 1..=10).is_err()); + } + + #[test] + fn a_switch_is_on_off_or_an_error() { + assert!(!flag("TEST_UNSET_FLAG").unwrap()); + assert!(flag(&with("TEST_FLAG_YES", "yes")).unwrap()); + assert!(flag(&with("TEST_FLAG_ONE", "1")).unwrap()); + assert!(!flag(&with("TEST_FLAG_OFF", "off")).unwrap()); + assert!(flag(&with("TEST_FLAG_TYPO", "ture")).is_err()); + } + #[test] fn a_time_of_day_is_read_as_written() { assert_eq!( diff --git a/src/main.rs b/src/main.rs index 3021da2..913cda6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,6 +11,7 @@ mod sftp; mod state; mod static_files; mod sweep; +mod tasks; mod throttle; mod vault; @@ -23,7 +24,9 @@ use sqlx::sqlite::{ }; use axum::http::{header, HeaderValue}; use tower_http::{ - compression::CompressionLayer, set_header::SetResponseHeaderLayer, trace::TraceLayer, + compression::CompressionLayer, + set_header::SetResponseHeaderLayer, + trace::{DefaultOnResponse, TraceLayer}, }; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter}; @@ -106,7 +109,14 @@ async fn main() -> Result<()> { let bind = config.bind; let max_upload = config.max_upload_bytes; let config = Arc::new(config); - let backup = backup::Backup::new(db.clone(), Arc::clone(&config))?; + let backup = match backup::Backup::new(db.clone(), Arc::clone(&config)) { + Ok(backup) => backup, + Err(e) => { + let reason = format!("{e:#}"); + tracing::error!("Proton Drive backups are off: {reason}"); + backup::Backup::unavailable(db.clone(), Arc::clone(&config), reason) + } + }; backup::spawn(Arc::clone(&backup)); let notify = notify::Notifier::new(db.clone(), Arc::clone(&config))?; @@ -114,6 +124,7 @@ async fn main() -> Result<()> { sweep::spawn(db.clone(), config.uploads_dir()); + let epoch = auth::Epoch::load(&db, &config).await?; let vault = vault::Vault::new(db.clone(), config.vault_idle); vault.spawn_idle_lock(); let state = AppState { @@ -124,13 +135,20 @@ async fn main() -> Result<()> { vault_throttle: Arc::default(), backup, notify, + epoch: Arc::new(std::sync::atomic::AtomicI64::new(epoch)), }; if let Some(sftp_bind) = config.sftp.bind { let sftp_state = state.clone(); - tokio::spawn(async move { - if let Err(e) = sftp::serve(sftp_state, sftp_bind).await { - tracing::error!("sftp server stopped: {e:#}"); + tasks::supervise("the sftp server", move || { + let state = sftp_state.clone(); + async move { + // A port that is taken now may be free in a minute; the web + // side carries on either way. + while let Err(e) = sftp::serve(state.clone(), sftp_bind).await { + tracing::error!("sftp server stopped: {e:#}; trying again in a minute"); + tokio::time::sleep(std::time::Duration::from_secs(60)).await; + } } }); } @@ -222,9 +240,21 @@ async fn serve(bind: std::net::SocketAddr, max_upload: usize, state: AppState) - .nest("/api", routes::api_router(max_upload)) .fallback(static_files::handler) .layer(axum::middleware::from_fn(auth::same_origin)) - .layer(CompressionLayer::new()) + .layer(axum::middleware::from_fn(deadline)) + .layer(CompressionLayer::new().compress_when(compressible)) .layer(security_headers()) - .layer(TraceLayer::new_for_http()) + .layer( + TraceLayer::new_for_http() + .make_span_with(|request: &axum::http::Request| { + tracing::info_span!( + "request", + method = %request.method(), + path = %loggable(request.uri().path()), + ) + }) + .on_request(()) + .on_response(DefaultOnResponse::new().level(tracing::Level::INFO)), + ) .with_state(state); let listener = tokio::net::TcpListener::bind(bind) @@ -250,9 +280,90 @@ async fn serve(bind: std::net::SocketAddr, max_upload: usize, state: AppState) - Ok(()) } +/// How long an ordinary request may take to be answered. A request that takes +/// this long is stuck — waiting on a lock it will not get — and is better +/// answered with an error than left hanging. +const REQUEST_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30); + +/// Requests that take as long as the bytes do: uploads, imports, re-sealing, +/// a backup run, the export. Only the time to the first byte of a response is +/// bounded anyway, so a long download of a file is never cut off. +fn is_slow(method: &axum::http::Method, path: &str) -> bool { + use axum::http::Method; + match *method { + Method::POST => { + path == "/api/media" + || path == "/api/import" + || path == "/api/backup" + || path.ends_with("/import") + || path.ends_with("/files") + || path.ends_with("/encrypt") + || path.ends_with("/decrypt") + } + Method::GET => path == "/api/export", + _ => false, + } +} + +async fn deadline(request: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response { + use axum::response::IntoResponse; + if is_slow(request.method(), request.uri().path()) { + return next.run(request).await; + } + let path = loggable(request.uri().path()).to_string(); + match tokio::time::timeout(REQUEST_DEADLINE, next.run(request)).await { + Ok(response) => response, + Err(_) => { + tracing::error!("{path} took longer than {REQUEST_DEADLINE:?} and was given up on"); + ( + axum::http::StatusCode::SERVICE_UNAVAILABLE, + axum::Json(serde_json::json!({ "error": "the server took too long — try again" })), + ) + .into_response() + } + } +} + /// Below the ten seconds `docker stop` waits before it sends SIGKILL. const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(8); +/// A request path as it may go into the log: a share link's token and key are +/// the whole of its secret, and the key is kept out of every log on purpose — +/// it travels in the URL fragment so that no server ever sees it, except this +/// one's API, which must not write it down either. Query strings, which carry +/// search terms, are never logged at all. +fn loggable(path: &str) -> &str { + if path.starts_with("/api/share/") { + "/api/share/…" + } else if path.starts_with("/s/") { + "/s/…" + } else { + path + } +} + +/// Text is worth compressing; uploads are not. Media is already compressed, +/// and a compressed body no longer matches the byte ranges a video player +/// asks for, so nothing that answers a range — or could — is touched. +fn compressible( + status: axum::http::StatusCode, + _: axum::http::Version, + headers: &axum::http::HeaderMap, + _: &axum::http::Extensions, +) -> bool { + if status == axum::http::StatusCode::PARTIAL_CONTENT || headers.contains_key(header::ACCEPT_RANGES) { + return false; + } + let kind = headers + .get(header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default(); + kind.starts_with("text/") + || kind.starts_with("application/json") + || kind.starts_with("application/javascript") + || kind.starts_with("image/svg+xml") +} + /// Content-Security-Policy for the app shell. `unsafe-inline` is granted to /// styles only, because CodeMirror injects its theme as a