M9: harden open hosting against resource exhaustion

Fixes from the security review of the public server:
- team libraries only in key-published files on open servers; unpacking
  bounded by the file cap (zip bombs refused)
- byte-bounded send queues per connection and in all, a bound on bytes
  waiting for the hub, per-address and global download budgets
- batches checked against the file, per-address and total caps as files
  grow (refused with a Reject the app rolls back); the core's recent-batch
  memory bounded in bytes; nothing confirmed when a log can't be written
- usage reports kept in memory; per-kind message caps checked while read
- layered IPv6 limits (/64, /56, /48), a global publish rate, connections
  reserved for sessions, short idle timeout for one-shot requests
- revoked links and removed files cut library connections
- data directory 0700, server key 0600 from creation
- install.sh: SSH ports from sshd, ufw left off when other services
  listen, unprivileged builds of the latest tag, settings checked before
  use, own Caddy snippet; systemd memory and network caps; backups check
  free space and say they hold the key
- linger reads raw bytes after an oversized frame (tungstenite asserted)
This commit is contained in:
bflatastic 2026-10-12 11:07:25 +07:00 committed by BFlat
parent faf2e78c9d
commit 7edc61b8d2
26 changed files with 1434 additions and 243 deletions

1
Cargo.lock generated
View File

@ -1651,6 +1651,7 @@ dependencies = [
"toml",
"tungstenite",
"wat",
"zip",
]
[[package]]

View File

@ -257,6 +257,7 @@ Ranked; detail and milestones in [`docs/roadmap.md`](docs/roadmap.md).
| Date | Change | Summary |
|---|---|---|
| 2026-10-12 | minor | Public `figcraft-server` hardened against resource exhaustion after a security review: messages capped by what a connection is for (a file's size when open), byte-bounded send queues and a bound on bytes waiting for the hub, files checked as they grow (refused batches roll back with the reason; per-address and total caps count growth), the core's recent-batch memory bounded in bytes, no confirmation for changes that couldn't be written, team libraries only in key-published files on public servers with unpacking bounded by the file cap, usage reports kept in memory, layered IPv6 limits (/64, /56, /48), a global publish rate, connections reserved for sessions, per-address download shares, library connections cut on revoke and removal, a private data directory and key, and safer install.sh, systemd and backups. Percentages and hours unchanged |
| 2026-10-12 | minor | A public `figcraft-server` (open hosting): with `open_publish = true` anyone publishes a file without the server key and gets its edit and view links plus an owner token (128 bits, shown once, stored hashed) that lists, makes and revokes its links, renames it or removes it (`collab.manageFile`); limits per address (connections at once and a minute, messages a second, publishes an hour, files and MiB kept; IPv6 by /64; `X-Forwarded-For` believed only from `trusted_proxy`; tables capped at 10,000 addresses) and for everyone (`max_total_mb`), each refusal saying why; files published without the key expire after `idle_days` unopened (the date shown to everyone in the file); moderation (`files list --sort size|age|idle`, `block-ip`, `reports`, a Report button and `collab.report`); the operator's contact and terms shown in the Share dialog; `GET /healthz` on the same port; `packaging/server/install.sh` for a Debian/Ubuntu machine behind Caddy and a nightly backup timer. Protocol additions within version 2, gated like signals (older apps get clear refusals). Hosting infrastructure rather than a Figma feature: percentages and hours unchanged |
| 2026-10-12 | minor | Experimental `.fig` export (gap 1): `file.export {format: "fig"}`, File ▸ Export… ▸ .fig (experimental), the CLI and MCP write the whole document (layer tree, geometry, paints with image bytes, strokes, effects, text with styled runs, vector networks, auto layout, constraints, components, sets, instances with their overrides and property values, prototype interactions) with a kiwi encoder mirroring the decoder and the schema of the fields written; what's left out is a warning; verified only by round trips through our own importer. File formats 59% → 60%; import/export 84%/57% → 86%/58%; file exchange stays ×0.85; mainstream hours ~130–244 h → ~120–229 h; feature breadth (~84%), ready for real work (~53%), essentials (~57%), mainstream (~49%), beta (~155–280 h) and full parity (~265–523 h) unchanged at this rounding |
| 2026-10-12 | minor | Keyboard shortcut parity suite and bindings (gap 4): a table test with one row per shortcut in the public documentation, macOS and Windows/Linux keys each (129 rows: 126 bound, 110 before; 1 moved off a documented conflict; 2 missing; 2 kept by a platform), plus 48 pinned undocumented bindings so every default binding is in the table; newly bound: A (Frame), Shift+' (pixel grid), Alt+Cmd+\ (others' cursors, new `view.multiplayerCursors`), Ctrl+F6, Shift+Cmd+U (link), Cmd/Ctrl+] and [ (list levels while editing, new `text.listIndent` / `text.listOutdent`), Cmd/Ctrl+Backspace (ungroup), Alt+Space / Ctrl+Space (keyboard box selection, Cmd/Ctrl+arrows draw it out), suggest auto layout, go to main component, arrows that pan with nothing selected (new `view.pan`), Control suspending snapping on macOS, Space/N/C/F while presenting; symbols another layout types with Shift count without it; no default binding uses a system-reserved chord (tested). Menu breadth 207/207 (View ▸ Multiplayer cursors). UI/UX fidelity 58% and the interaction pass 40–60 h unchanged at this rounding |

View File

@ -34,6 +34,8 @@ env_logger = { version = "0.11", default-features = false, features = ["humantim
ctrlc = { version = "3", features = ["termination"] }
[dev-dependencies]
# Crafting hostile archives (zip bombs) in tests.
zip = { workspace = true }
wat = { workspace = true }
figcraft-engine = { workspace = true }
figcraft-doc = { workspace = true }

View File

@ -1,13 +1,16 @@
//! What one address may do how often: new connections, publishing without the key and messages
//! (only on a server open to the public), and reports (always). Each is a token bucket per address
//! ([`figcraft_collab_net::hub::ip_key`]: an IPv6 address counts by its /64). The table keeps at
//! most [`MAX_ADDRESSES`] addresses: when it is full the half used longest ago is forgotten, so a
//! flood of addresses costs bounded memory (and forgotten addresses start with full buckets).
//! (only on a server open to the public), and reports and downloads (always). Each is a token
//! bucket per block an address counts in ([`ip_blocks`]: an IPv4 address, or an IPv6 /64 and its
//! /56 and /48 with 4 and 16 times the rate), so a tunnel's /48 doesn't get 65,536 /64s' worth. Publishing has a budget for
//! everyone besides ([`PUBLISHES_ALL`] times one address's). The table keeps at most
//! [`MAX_ADDRESSES`] addresses: when it is full the half used longest ago is forgotten, so a flood
//! of addresses costs bounded memory (and forgotten addresses start with full buckets).
use std::collections::HashMap;
use std::net::IpAddr;
use std::time::Instant;
use figcraft_collab_net::hub::{Block, ip_blocks};
use figcraft_collab_net::net::Bucket;
/// Most addresses remembered.
@ -16,6 +19,13 @@ pub const MAX_ADDRESSES: usize = 10_000;
/// Reports one address may send an hour.
pub const REPORTS_PER_HOUR: f64 = 10.0;
/// Bytes one address may download a second (team libraries, plugins), and at once.
pub const DOWNLOAD_RATE: f64 = (4u64 << 20) as f64;
pub const DOWNLOAD_BURST: f64 = (64u64 << 20) as f64;
/// Everyone together may publish without the key this many times what one address may.
pub const PUBLISHES_ALL: f64 = 20.0;
/// The rates (see the server's settings).
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct AbuseConfig {
@ -33,6 +43,8 @@ pub enum Action {
Publish,
Message,
Report,
/// Downloading this many bytes.
Download(usize),
}
struct Address {
@ -40,13 +52,16 @@ struct Address {
publishes: Bucket,
messages: Bucket,
reports: Bucket,
downloads: Bucket,
used: Instant,
}
/// The buckets of the addresses seen lately.
pub struct Abuse {
cfg: AbuseConfig,
table: HashMap<IpAddr, Address>,
table: HashMap<Block, Address>,
/// Publishing without the key, everyone together.
publishes: Bucket,
}
/// `n` every `period` seconds on average, and all `n` at once at most.
@ -56,35 +71,69 @@ fn bucket(n: f64, period: f64) -> Bucket {
impl Abuse {
pub fn new(cfg: AbuseConfig) -> Self {
Self { cfg, table: HashMap::new() }
let publishes = bucket(f64::from(cfg.publishes_per_hour) * PUBLISHES_ALL, 3600.0);
Self { cfg, table: HashMap::new(), publishes }
}
/// Whether `ip` may do `what` now (counting it when it may).
/// Whether `ip` may do `what` now (counting it, in every block it is in, when it may).
pub fn allow(&mut self, ip: IpAddr, what: Action, now: Instant) -> bool {
if !self.cfg.open && what != Action::Report {
if !self.cfg.open && matches!(what, Action::Connect | Action::Publish | Action::Message) {
return true;
}
let keys: Vec<(Block, f64)> = ip_blocks(ip).into_iter().map(|(k, times)| (k, f64::from(times))).collect();
// Every block must have room before any is charged.
if !keys.iter().all(|&(k, times)| self.room(k, times, what, now)) {
return false;
}
if what == Action::Publish && !self.publishes.take(now) {
return false;
}
for &(k, times) in &keys {
let (b, n) = Self::pick(self.address(k, times, now), what);
b.take_n(now, n);
}
true
}
/// The address's buckets, made when it is new (forgetting the oldest half when full).
fn address(&mut self, ip: Block, times: f64, now: Instant) -> &mut Address {
if !self.table.contains_key(&ip) && self.table.len() >= MAX_ADDRESSES {
self.forget_oldest_half();
}
let c = self.cfg;
let a = self.table.entry(ip).or_insert_with(|| Address {
connects: bucket(f64::from(c.connects_per_minute), 60.0),
publishes: bucket(f64::from(c.publishes_per_hour), 3600.0),
connects: bucket(f64::from(c.connects_per_minute) * times, 60.0),
publishes: bucket(f64::from(c.publishes_per_hour) * times, 3600.0),
// A burst of four seconds' worth (a big paste, a quick drag).
messages: Bucket::new(f64::from(c.messages_per_second), f64::from(c.messages_per_second) * 4.0),
reports: bucket(REPORTS_PER_HOUR, 3600.0),
messages: Bucket::new(f64::from(c.messages_per_second) * times, f64::from(c.messages_per_second) * times * 4.0),
reports: bucket(REPORTS_PER_HOUR * times, 3600.0),
downloads: Bucket::new(DOWNLOAD_RATE * times, DOWNLOAD_BURST * times),
used: now,
});
a.used = now;
a
}
fn pick(a: &mut Address, what: Action) -> (&mut Bucket, f64) {
match what {
Action::Connect => a.connects.take(now),
Action::Publish => a.publishes.take(now),
Action::Message => a.messages.take(now),
Action::Report => a.reports.take(now),
Action::Connect => (&mut a.connects, 1.0),
Action::Publish => (&mut a.publishes, 1.0),
Action::Message => (&mut a.messages, 1.0),
Action::Report => (&mut a.reports, 1.0),
Action::Download(n) => (&mut a.downloads, n as f64),
}
}
/// The block has `what`'s tokens now (nothing is spent).
fn room(&mut self, ip: Block, times: f64, what: Action, now: Instant) -> bool {
let (b, n) = Self::pick(self.address(ip, times, now), what);
if b.take_n(now, n) {
b.give(n);
return true;
}
false
}
/// Addresses remembered.
pub fn len(&self) -> usize {
self.table.len()

View File

@ -241,7 +241,7 @@ impl Config {
max_links: self.max_links,
compact_after: self.compact_after,
compact_bytes: mib(self.compact_mb),
core: figcraft_collab::Limits { max_peers: self.max_people, ..figcraft_collab::Limits::default() },
core: self.core(),
directory: DirectoryConfig {
max_plugins: self.max_plugins,
max_plugin_bytes: mib(self.max_plugin_mb),
@ -250,13 +250,44 @@ impl Config {
}
}
/// The transport's part (connections per address are capped only when open to the public).
/// Largest message a client may send when open to the public: a whole file (or plugin bundle)
/// and a margin, not the core's 64 MiB.
fn open_message(&self) -> usize {
usize::try_from(self.file_mb().max(self.max_plugin_mb).saturating_add(1).saturating_mul(1 << 20)).unwrap_or(usize::MAX)
}
/// The collaboration core's part: when open, messages as large as a file at most, and the
/// batches kept for reconnecting clients (in memory, and in the file's snapshot) at most a
/// quarter of a file's bytes (256 KiB at least).
fn core(&self) -> figcraft_collab::Limits {
let base = figcraft_collab::Limits { max_peers: self.max_people, ..figcraft_collab::Limits::default() };
if !self.open_publish {
return base;
}
let max = self.open_message().min(base.max_message_bytes);
let file = usize::try_from(self.file_mb().saturating_mul(1 << 20)).unwrap_or(usize::MAX);
figcraft_collab::Limits { max_message_bytes: max, max_log_bytes: (file / 4).max(256 << 10), ..base }
}
/// The transport's part. A private server keeps the defaults; open to the public, connections
/// per address are capped, messages are as large as a file at most, the bytes queued for a
/// client and for everyone and those waiting for the hub are bounded, one-shot requests leave
/// an eighth of the connections to sessions and are dropped after 20 idle seconds.
pub fn net(&self) -> NetLimits {
let base = NetLimits { max_connections: self.max_connections, trusted_proxy: self.proxy().ok().flatten(), ..NetLimits::default() };
if !self.open_publish {
return base;
}
let max = self.open_message().min(base.max_client_message);
NetLimits {
max_connections: self.max_connections,
max_per_address: if self.open_publish { self.max_connections_per_ip } else { 0 },
trusted_proxy: self.proxy().ok().flatten(),
..NetLimits::default()
max_per_address: self.max_connections_per_ip,
max_client_message: max,
max_queued_bytes: max.saturating_mul(2),
max_queued_total: 256 << 20,
max_in_flight: 128 << 20,
reserve_for_sessions: self.max_connections / 8,
request_idle: std::time::Duration::from_secs(20),
..base
}
}
}

View File

@ -41,9 +41,27 @@ pub fn temp_of(path: &Path) -> PathBuf {
/// Replace `path` with `bytes` atomically: a reader sees the old or the new content, never a mix.
pub fn write_atomic(path: &Path, bytes: &[u8]) -> io::Result<()> {
write_atomic_as(path, bytes, false)
}
/// [`write_atomic`] for a secret: on Unix only the owner can read it, from the moment it exists.
pub fn write_atomic_private(path: &Path, bytes: &[u8]) -> io::Result<()> {
write_atomic_as(path, bytes, true)
}
fn write_atomic_as(path: &Path, bytes: &[u8], private: bool) -> io::Result<()> {
let tmp = temp_of(path);
{
let mut f = File::create(&tmp)?;
let mut o = fs::OpenOptions::new();
o.write(true).create(true).truncate(true);
#[cfg(unix)]
if private {
use std::os::unix::fs::OpenOptionsExt;
o.mode(0o600);
}
#[cfg(not(unix))]
let _ = private;
let mut f = o.open(&tmp)?;
f.write_all(bytes)?;
f.sync_all()?;
}

View File

@ -0,0 +1,535 @@
//! Open hosting against resource exhaustion: each test reproduces an attack from the security
//! review and checks what bounds it — a zip-bomb library (H1), a client that never reads its
//! replies (H2), a file growing past its cap through batches and a log that can't be written (H3),
//! usage reports that rewrote a file per request (M1), oversized messages and a busy hub (M2),
//! many addresses from one IPv6 block and one-shot requests taking every connection (M3), revoked
//! links on library connections (M4), data directory permissions (L6) and downloads per address
//! (L7).
use std::io::{Cursor, Write};
use std::net::{IpAddr, TcpListener};
use std::time::{Duration, Instant};
use figcraft_collab::message::{
Access, Admin, AdminReply, ClientMsg, LibraryReply, LibraryReq, OpenHosting, OpenLibrary, OpenPlugins, PROTOCOL, PluginReq, ServerMsg, Usage,
};
use figcraft_collab::sim::Rng;
use figcraft_collab::{Client, Limits, NoRecompute, Outgoing, Replica, Server};
use figcraft_collab_net::hub::{Backend, Hosting};
use figcraft_collab_net::{Endpoint, Invite, NetLimits, manage};
use figcraft_doc::{Document, NodeId, NodeKind};
use figcraft_engine::Session;
use serde_json::json;
use crate::abuse::{Abuse, AbuseConfig, Action, DOWNLOAD_BURST};
use crate::config::Config;
use crate::public_tests::via_proxy;
use crate::store::{BUSY_DOWNLOADS, NO_PUBLIC_LIBRARY, Store};
use crate::test_util::{Raw, TempDir, config};
fn open(dir: &std::path::Path) -> Config {
Config { open_publish: true, operator_contact: "abuse@example.com".into(), ..config(dir) }
}
fn snapshot(name: &str) -> Vec<figcraft_collab::Change> {
let mut d = Document::new(name);
let page = d.first_page().unwrap();
d.create(page, NodeKind::Rectangle, "Card", 0.0, 0.0, 100.0, 60.0).unwrap();
d.repair();
Replica::new(d).snapshot().unwrap()
}
fn ip(s: &str) -> IpAddr {
s.parse().unwrap()
}
/// A team library archive as the app makes it; with `bomb` bytes of spaces as its snapshot
/// (deflated: a few KiB) and `padding` incompressible bytes stored beside it.
fn archive(bomb: usize, padding: usize) -> Vec<u8> {
let mut s = Session::new();
let f = s.execute("frame.create", &json!({"x": 0, "y": 0, "width": 50, "height": 20, "name": "Button"})).unwrap()["id"].as_u64().unwrap();
s.execute("component.create", &json!({"ids": [f]})).unwrap();
s.execute("library.publish", &json!({})).unwrap();
let lib = s.doc().unwrap().published.clone().unwrap();
let bytes = figcraft_engine::library_server::archive(&lib).unwrap();
if bomb == 0 && padding == 0 {
return bytes;
}
let mut src = zip::ZipArchive::new(Cursor::new(bytes)).unwrap();
let mut out = zip::ZipWriter::new(Cursor::new(Vec::new()));
for i in 0..src.len() {
let f = src.by_index_raw(i).unwrap();
if bomb > 0 && f.name() == "library/document.json" {
continue;
}
out.raw_copy_file(f).unwrap();
}
let opts = |m| zip::write::SimpleFileOptions::default().compression_method(m);
if bomb > 0 {
out.start_file("library/document.json", opts(zip::CompressionMethod::Deflated)).unwrap();
let chunk = vec![b' '; 1 << 20];
for _ in 0..bomb >> 20 {
out.write_all(&chunk).unwrap();
}
}
if padding > 0 {
out.start_file("padding.bin", opts(zip::CompressionMethod::Stored)).unwrap();
let mut rng = Rng::new(3);
out.write_all(&(0..padding).map(|_| rng.next_u64() as u8).collect::<Vec<u8>>()).unwrap();
}
out.finish().unwrap().into_inner()
}
fn open_library(link: &Invite) -> ClientMsg {
ClientMsg::OpenLibrary(OpenLibrary { protocol: PROTOCOL, file: link.file.clone(), token: link.token.clone(), client_version: String::new() })
}
fn hosting() -> ClientMsg {
ClientMsg::OpenHosting(OpenHosting { protocol: PROTOCOL, client_version: String::new(), file: String::new(), owner: String::new() })
}
/// Poll `f` until it holds or `secs` pass (bounded: never a fixed sleep for the outcome).
fn until(secs: u64, what: &str, mut f: impl FnMut() -> bool) {
let end = Instant::now() + Duration::from_secs(secs);
while !f() {
assert!(Instant::now() < end, "timed out waiting for {what}");
std::thread::sleep(Duration::from_millis(10));
}
}
/// A client in `server` (as the hub would let it in).
fn join(server: &mut Server) -> Client {
let ClientMsg::Hello(h) = Client::hello("f", "", "Ada", "#FF8800") else { panic!() };
let (_, out) = server.join_as(h, Access::Edit).unwrap();
let Some(Outgoing::To(_, w)) = out.into_iter().next() else { panic!() };
Client::join(w).unwrap()
}
/// Rename the first page through `c`, as the server takes it.
fn rename(server: &mut Server, c: &mut Client, name: &str) {
let mut d = c.document().clone();
let page = d.first_page().unwrap();
d.node_mut(page).unwrap().name = name.into();
let m = c.edit(&d).unwrap().unwrap();
for o in server.receive(c.id(), m) {
if let Outgoing::All(m) | Outgoing::To(_, m) = o {
c.receive(m, &NoRecompute).unwrap();
}
}
}
// H1 -------------------------------------------------------------------------------------------
#[test]
fn a_zip_bomb_library_is_refused_before_it_unpacks() {
let dir = TempDir::new("bomb");
let mut c = config(dir.path());
c.max_file_mb = Some(1);
let mut store = Store::open(dir.path(), c.store()).unwrap();
let (keyed, ..) = store.publish("Library", &snapshot("Library")).unwrap();
let bomb = archive(64 << 20, 0);
assert!(bomb.len() < 1 << 20, "a small archive ({} bytes) that unpacks to 64 MiB", bomb.len());
let e = store.publish_library(&keyed.id, &bomb).unwrap_err();
assert!(e.contains("too large"), "{e}");
assert_eq!(store.publish_library(&keyed.id, &archive(0, 0)).unwrap(), 1, "a real library still goes in");
drop(store);
// Open to the public: a file published without the key holds no library at all.
let mut store = Store::open(dir.path(), open(dir.path()).store()).unwrap();
let (mine, ..) = store.publish_open(ip("203.0.113.5"), "Mine", &snapshot("Mine")).unwrap();
assert_eq!(store.publish_library(&mine.id, &archive(0, 0)).unwrap_err(), NO_PUBLIC_LIBRARY);
assert_ne!(store.publish_library(&keyed.id, &archive(0, 0)).unwrap_err(), NO_PUBLIC_LIBRARY, "the operator's files still can");
}
// H2 -------------------------------------------------------------------------------------------
#[test]
fn a_client_that_never_reads_can_queue_only_so_much() {
let dir = TempDir::new("noread");
let cfg = Config { max_file_mb: Some(2), max_plugin_mb: 1, ..open(dir.path()) };
let cap = cfg.net().max_queued_bytes;
assert_eq!(cap, 6 << 20, "twice the largest message");
let view = {
let mut store = Store::open(dir.path(), cfg.store()).unwrap();
let (file, _, view) = store.publish("Library", &snapshot("Library")).unwrap();
store.publish_library(&file.id, &archive(0, 1_500_000)).unwrap();
Invite { server: Endpoint::plain("x"), file: file.id, token: view.token }
};
let mut server = crate::start(&cfg).unwrap();
let mut raw = Raw::connect(server.addr());
raw.send(&open_library(&view));
assert!(matches!(raw.next(), ServerMsg::Library(r) if matches!(*r, LibraryReply::Ready { .. })));
// Ask for the 1.5 MB library over and over without reading an answer.
for _ in 0..100 {
if raw.try_send(&ClientMsg::Library(LibraryReq::Fetch { have: None, usage: None })).is_err() {
break;
}
}
let mut most = 0;
let end = Instant::now() + Duration::from_secs(3);
while Instant::now() < end {
most = most.max(server.queued_bytes());
std::thread::sleep(Duration::from_millis(5));
}
assert!(most <= cap, "{most} bytes were queued for one client (cap {cap})");
// Its queue is let go once it is dropped.
until(15, "the stuck client's queue to be freed", || server.queued_bytes() == 0);
server.stop();
}
// H3 -------------------------------------------------------------------------------------------
#[test]
fn a_file_cant_grow_past_its_cap_through_batches() {
let dir = TempDir::new("grow");
let cfg = Config { max_file_mb: Some(1), max_plugin_mb: 1, ..open(dir.path()) };
let max = 1u64 << 20;
let mut server = crate::start(&cfg).unwrap();
let addr = server.addr();
let (info, replies) = manage::hosting(
&Endpoint::plain(addr.to_string()),
"",
"",
vec![Admin::Publish { name: "Grow".into(), snapshot: snapshot("Grow") }],
&NetLimits::default(),
Duration::from_secs(5),
)
.unwrap();
assert!(info.open);
let Some(AdminReply::Hosted { file, edit, .. }) = replies.into_iter().next() else { panic!("not hosted") };
let mut raw = Raw::connect(addr);
raw.send(&Client::hello(&file.id, &edit.token, "Grower", ""));
let welcome = loop {
if let w @ ServerMsg::Welcome { .. } = raw.next() {
break w;
}
};
let mut client = Client::join(welcome).unwrap();
// Each batch adds a layer with a 200 KB name: the file grows until the server refuses.
let mut refused = None;
for i in 0..12 {
let mut d = client.document().clone();
let page = d.first_page().unwrap();
d.create(page, NodeKind::Rectangle, &format!("{i}{}", "x".repeat(200_000)), 0.0, 0.0, 10.0, 10.0).unwrap();
let Some(m) = client.edit(&d).unwrap() else { continue };
raw.send(&m);
let answer = loop {
if let m @ (ServerMsg::Changes(_) | ServerMsg::Reject { .. }) = raw.next() {
break m;
}
};
if let ServerMsg::Reject { reason, .. } = &answer {
refused = Some(reason.clone());
}
client.receive(answer, &NoRecompute).unwrap();
if refused.is_some() {
break;
}
}
let reason = refused.expect("the server refused a batch once the file was full");
assert!(reason.contains("reached the server's limit of 1 MiB"), "{reason}");
let on_disk: u64 = ["snapshot.bin", "log.bin"]
.iter()
.filter_map(|n| std::fs::metadata(dir.path().join("files").join(&file.id).join(n)).ok())
.map(|m| m.len())
.sum();
// Its snapshot holds the document and the recent batches kept for reconnecting clients (at
// most a quarter of the cap), and the last batch let in may finish what the cap allows.
assert!(on_disk <= max + max / 4 + 210_000, "{on_disk} bytes on disk");
server.stop();
}
#[test]
fn the_sessions_memory_of_batches_is_bounded_in_bytes() {
let mut server = Server::new(Document::new("Log"), "", Limits { max_log_bytes: 100_000, ..Limits::default() });
let mut c = join(&mut server);
for i in 0..20 {
rename(&mut server, &mut c, &format!("{i}{}", "y".repeat(30_000)));
assert!(server.log_bytes() <= 100_000, "{} bytes kept", server.log_bytes());
}
assert!(server.log_bytes() > 0, "recent batches are still kept");
}
#[test]
fn changes_that_cant_be_saved_are_never_confirmed() {
let dir = TempDir::new("unsaved");
let mut store = Store::open(dir.path(), config(dir.path()).store()).unwrap();
let (info, ..) = store.publish("Durable", &snapshot("Durable")).unwrap();
let mut server = Backend::open(&mut store, &info.id).unwrap();
let mut c = join(&mut server);
rename(&mut server, &mut c, "Kept");
Backend::changed(&mut store, &info.id, &mut server);
assert!(Backend::flush(&mut store).is_empty());
rename(&mut server, &mut c, "Lost");
Backend::changed(&mut store, &info.id, &mut server);
store.break_log(&info.id);
assert_eq!(Backend::flush(&mut store), vec![info.id.clone()], "the hub hears it couldn't be saved");
// The file opens again from what is on disk: the saved change, not the unsaved one.
drop(server);
let mut server = Backend::open(&mut store, &info.id).unwrap();
let c = join(&mut server);
let page = c.document().first_page().unwrap();
assert_eq!(c.document().node(page).unwrap().name, "Kept");
}
// M1 -------------------------------------------------------------------------------------------
#[test]
fn usage_reports_dont_rewrite_a_file_per_request() {
let dir = TempDir::new("usage");
let cfg = config(dir.path());
let view = {
let mut store = Store::open(dir.path(), cfg.store()).unwrap();
let (file, _, view) = store.publish("Library", &snapshot("Library")).unwrap();
store.publish_library(&file.id, &archive(0, 0)).unwrap();
Invite { server: Endpoint::plain("x"), file: file.id, token: view.token }
};
let usage_file = dir.path().join("files").join(&view.file).join("usage.json");
let mut server = crate::start(&cfg).unwrap();
let mut raw = Raw::connect(server.addr());
raw.send(&open_library(&view));
raw.next();
for i in 0..50u32 {
let usage = Usage { file: format!("User {}", i % 7), components: vec![(NodeId(2), i + 1)] };
raw.send(&ClientMsg::Library(LibraryReq::Fetch { have: Some(1), usage: Some(usage) }));
assert!(matches!(raw.next(), ServerMsg::Library(r) if matches!(*r, LibraryReply::Unchanged { .. })));
}
assert!(!usage_file.exists(), "kept in memory, not written 50 times");
raw.send(&ClientMsg::Library(LibraryReq::Usage));
let ServerMsg::Library(r) = raw.next() else { panic!() };
let LibraryReply::Usage(files) = *r else { panic!() };
assert_eq!(files.len(), 7);
server.stop();
assert!(usage_file.exists(), "written when the server stops");
}
// M2 -------------------------------------------------------------------------------------------
/// Panics in the server's connection threads (`live-…`) since the hook went in: they mustn't
/// happen, but a thread's panic doesn't fail a test by itself.
static CONNECTION_PANICS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
fn count_connection_panics() {
static HOOK: std::sync::Once = std::sync::Once::new();
HOOK.call_once(|| {
let before = std::panic::take_hook();
std::panic::set_hook(Box::new(move |info| {
if std::thread::current().name().is_some_and(|n| n.starts_with("live-")) {
CONNECTION_PANICS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
before(info);
}));
});
}
#[test]
fn messages_are_capped_by_what_the_connection_is_for() {
count_connection_panics();
let dir = TempDir::new("caps");
let mut server = crate::start(&open(dir.path())).unwrap();
let addr = server.addr();
// Browsing a plugin directory takes small requests only; the oversized frame is refused and
// the connection closed cleanly (its reader isn't asked for the rest of that frame).
let mut raw = Raw::connect(addr);
raw.send(&ClientMsg::OpenPlugins(OpenPlugins { protocol: PROTOCOL, key: String::new(), client_version: String::new() }));
raw.next();
raw.send(&ClientMsg::Plugins(PluginReq::List { query: "q".repeat(200_000) }));
assert!(matches!(raw.next(), ServerMsg::Refused { reason } if reason.contains("too large") && reason.contains("KiB")));
drop(raw);
// A joined client whose batch is bigger than a file may be is told the limit.
let (_, replies) = manage::hosting(
&Endpoint::plain(addr.to_string()),
"",
"",
vec![Admin::Publish { name: "Small".into(), snapshot: snapshot("Small") }],
&NetLimits::default(),
Duration::from_secs(5),
)
.unwrap();
let Some(AdminReply::Hosted { file, edit, .. }) = replies.into_iter().next() else { panic!("not hosted") };
let mut joined = Raw::connect(addr);
joined.send(&Client::hello(&file.id, &edit.token, "Big", ""));
while !matches!(joined.next(), ServerMsg::Welcome { .. }) {}
joined.send(&ClientMsg::Presence(figcraft_collab::Presence {
selection: Some((0..5_000_000u64).map(|i| NodeId(i + (1 << 40))).collect()),
..Default::default()
}));
let reason = loop {
if let ServerMsg::Refused { reason } = joined.next() {
break reason;
}
};
assert!(reason.contains("at most 17 MiB"), "{reason}");
drop(joined);
until(10, "the refused connections to close", || server.connections() == 0);
assert_eq!(CONNECTION_PANICS.load(std::sync::atomic::Ordering::SeqCst), 0, "a connection thread panicked");
// Open to the public, a message is at most a file's size (and a margin), not 64 MiB.
assert_eq!(open(dir.path()).net().max_client_message, 17 << 20);
server.stop();
// Past the bytes the hub may have waiting, a connection is dropped as busy.
let cfg = open(dir.path());
let store = Store::open(dir.path(), cfg.store()).unwrap();
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let mut busy = Hosting::serve(listener, store, NetLimits { max_in_flight: 10_000, ..cfg.net() }).unwrap();
let mut raw = Raw::connect(busy.addr());
raw.send(&hosting());
assert!(matches!(raw.next(), ServerMsg::Hosting(_)));
let mut big = Document::new("Big");
let page = big.first_page().unwrap();
big.create(page, NodeKind::Rectangle, &"z".repeat(30_000), 0.0, 0.0, 1.0, 1.0).unwrap();
raw.send(&ClientMsg::Admin(Admin::Publish { name: "Big".into(), snapshot: Replica::new(big).snapshot().unwrap() }));
assert!(matches!(raw.next(), ServerMsg::Refused { reason } if reason.contains("busy")));
busy.stop();
}
// M3 -------------------------------------------------------------------------------------------
#[test]
fn one_ipv6_block_gets_rising_but_bounded_limits() {
let dir = TempDir::new("blocks");
let cfg = Config { max_connections_per_ip: 1, trusted_proxy: Some("127.0.0.1".into()), ..open(dir.path()) };
let mut server = crate::start(&cfg).unwrap();
let addr = server.addr();
let mut held = Vec::new();
// Four /64s of one /56 fit (4 × the cap), the fifth doesn't.
for n in 1..=4 {
let (ws, r) = via_proxy(addr, &format!("2001:db8:1:{n}::1"));
assert!(matches!(r, ServerMsg::Hosting(_)), "/64 number {n}");
held.push(ws);
}
assert!(matches!(via_proxy(addr, "2001:db8:1:5::1").1, ServerMsg::Refused { reason } if reason.contains("too many connections")));
// Sixteen /56s of one /48 fit (16 × the cap) — four of them taken above — the seventeenth doesn't.
for n in 1..=12u16 {
let (ws, r) = via_proxy(addr, &format!("2001:db8:1:{:x}::1", n << 8));
assert!(matches!(r, ServerMsg::Hosting(_)), "/56 number {n}");
held.push(ws);
}
assert!(matches!(via_proxy(addr, "2001:db8:1:d00::1").1, ServerMsg::Refused { .. }), "the /48 is full");
assert!(matches!(via_proxy(addr, "2001:db8:2::1").1, ServerMsg::Hosting(_)), "another /48 isn't");
drop(held);
server.stop();
}
#[test]
fn rates_count_per_block_and_publishing_for_everyone() {
let mut a = Abuse::new(AbuseConfig { open: true, connects_per_minute: 1, publishes_per_hour: 1, messages_per_second: 1 });
let t = Instant::now();
// One connection a minute per /64, four per /56, sixteen per /48.
let in_one_56 = (0..64u16).filter(|n| a.allow(IpAddr::from([0x2001, 0xdb8, 7, *n, 0, 0, 0, 0]), Action::Connect, t)).count();
assert_eq!(in_one_56, 4);
let in_one_48 = (1..64u16).filter(|n| a.allow(IpAddr::from([0x2001, 0xdb8, 7, n << 8, 0, 0, 0, 0]), Action::Connect, t)).count();
assert_eq!(in_one_48, 12, "the /48's 16, four of them taken above");
// Publishing: one an hour per address, twenty for everyone together.
let published = (0..40u32).filter(|n| a.allow(IpAddr::from((0xC000_0200 + n).to_be_bytes()), Action::Publish, t)).count();
assert_eq!(published, 20);
}
#[test]
fn one_shot_requests_leave_connections_for_sessions() {
let dir = TempDir::new("reserve");
let cfg = Config { max_connections: 8, ..open(dir.path()) };
assert_eq!(cfg.net().reserve_for_sessions, 1);
let edit = {
let mut store = Store::open(dir.path(), cfg.store()).unwrap();
let (file, edit, _) = store.publish("Team", &snapshot("Team")).unwrap();
(file.id, edit.token)
};
let mut server = crate::start(&cfg).unwrap();
let addr = server.addr();
let mut held = Vec::new();
for _ in 0..7 {
let mut raw = Raw::connect(addr);
raw.send(&hosting());
assert!(matches!(raw.next(), ServerMsg::Hosting(_)));
held.push(raw);
}
let mut eighth = Raw::connect(addr);
eighth.send(&hosting());
assert!(matches!(eighth.next(), ServerMsg::Refused { reason } if reason.contains("busy")));
drop(eighth);
until(5, "the refused connection to close", || server.connections() <= 7);
let mut session = Raw::connect(addr);
session.send(&Client::hello(&edit.0, &edit.1, "Ada", ""));
assert!(matches!(session.next(), ServerMsg::Welcome { .. }), "a session still gets in");
server.stop();
}
// M4 -------------------------------------------------------------------------------------------
#[test]
fn revoking_or_removing_cuts_library_connections_too() {
let dir = TempDir::new("libkick");
let cfg = config(dir.path());
let (file, view) = {
let mut store = Store::open(dir.path(), cfg.store()).unwrap();
let (file, _, view) = store.publish("Library", &snapshot("Library")).unwrap();
store.publish_library(&file.id, &archive(0, 0)).unwrap();
(file.id, view)
};
let mut server = crate::start(&cfg).unwrap();
let addr = server.addr();
let invite = Invite { server: Endpoint::plain("x"), file: file.clone(), token: view.token.clone() };
let mut lib = Raw::connect(addr);
lib.send(&open_library(&invite));
lib.next();
let key = crate::test_util::key(dir.path());
manage::request(
&Endpoint::plain(addr.to_string()),
&key,
vec![Admin::Revoke { file: file.clone(), link: view.link.id.clone() }],
&NetLimits::default(),
Duration::from_secs(5),
)
.unwrap();
assert!(matches!(lib.next(), ServerMsg::Refused { reason } if reason.contains("revoked")));
// A library connection with another link is cut when the file is removed.
let made = manage::request(
&Endpoint::plain(addr.to_string()),
&key,
vec![Admin::CreateLink { file: file.clone(), access: Access::View, label: String::new() }],
&NetLimits::default(),
Duration::from_secs(5),
)
.unwrap();
let Some(AdminReply::Link(l)) = made.into_iter().next() else { panic!() };
let mut lib = Raw::connect(addr);
lib.send(&open_library(&Invite { token: l.token, ..invite }));
lib.next();
manage::request(&Endpoint::plain(addr.to_string()), &key, vec![Admin::Remove { file }], &NetLimits::default(), Duration::from_secs(5)).unwrap();
assert!(matches!(lib.next(), ServerMsg::Refused { reason } if reason.contains("removed")));
server.stop();
}
// L6, L7 ---------------------------------------------------------------------------------------
#[cfg(unix)]
#[test]
fn only_the_server_reads_its_data_directory_and_key() {
use std::os::unix::fs::PermissionsExt;
let dir = TempDir::new("perms");
let data = dir.path().join("data");
let store = Store::open(&data, config(&data).store()).unwrap();
drop(store);
assert_eq!(std::fs::metadata(&data).unwrap().permissions().mode() & 0o777, 0o700);
assert_eq!(std::fs::metadata(data.join("server.key")).unwrap().permissions().mode() & 0o777, 0o600);
}
#[test]
fn downloads_have_a_share_per_address() {
let mut a = Abuse::new(AbuseConfig { open: false, connects_per_minute: 1, publishes_per_hour: 1, messages_per_second: 1 });
let t = Instant::now();
let me = ip("198.51.100.1");
let chunk = 16 << 20;
let served = (0..10).filter(|_| a.allow(me, Action::Download(chunk), t)).count();
assert_eq!(served, (DOWNLOAD_BURST as usize) / chunk, "a burst, then wait");
assert!(a.allow(ip("198.51.100.2"), Action::Download(chunk), t), "another address has its own share");
// A team library is served within it, then refused, saying why; another address still gets it.
let dir = TempDir::new("libshare");
let mut store = Store::open(dir.path(), config(dir.path()).store()).unwrap();
let (file, ..) = store.publish("Library", &snapshot("Library")).unwrap();
store.publish_library(&file.id, &archive(0, 4 << 20)).unwrap();
let fetch = |s: &mut Store, who: &str| s.library(ip(who), &file.id, Access::View, LibraryReq::Fetch { have: None, usage: None });
let served = (0..40).take_while(|_| matches!(fetch(&mut store, "198.51.100.1"), LibraryReply::Library { .. })).count();
assert!((10..20).contains(&served), "{served} downloads of 4 MiB in a burst of 64 MiB");
assert!(matches!(fetch(&mut store, "198.51.100.1"), LibraryReply::Failed { reason } if reason == BUSY_DOWNLOADS));
assert!(matches!(fetch(&mut store, "198.51.100.9"), LibraryReply::Library { .. }));
}

View File

@ -39,6 +39,11 @@ impl Running {
self.hosting.connections()
}
/// Bytes waiting to be sent to clients, all connections together.
pub fn queued_bytes(&self) -> usize {
self.hosting.queued_bytes()
}
/// Stop (see the type's docs).
pub fn stop(&mut self) {
self.hosting.stop();
@ -82,6 +87,8 @@ mod cli_tests;
#[cfg(test)]
mod directory_tests;
#[cfg(test)]
mod hardening_tests;
#[cfg(test)]
mod library_tests;
#[cfg(test)]
mod public_tests;

View File

@ -130,7 +130,7 @@ fn publish_enable_update_and_usage() {
let LibraryReply::Usage(files) = answer(&mut raw) else { panic!("no usage") };
let spam = files.iter().find(|f| f.usage.file.starts_with("Spam")).unwrap();
assert!(spam.usage.file.chars().count() <= 256 && !spam.usage.file.contains('\n'), "names are cleaned and capped");
assert_eq!(spam.usage.components.len(), 2_000, "reports are capped");
assert_eq!(spam.usage.components.len(), 1_000, "reports are capped");
let mut raw = Raw::connect(addr);
raw.send(&open(&Invite { token: "nope".into(), ..edit.clone() }));
assert!(matches!(raw.next(), ServerMsg::Refused { .. }));
@ -141,7 +141,7 @@ fn publish_enable_update_and_usage() {
drop(server);
let mut store = Store::open(dir.path(), cfg.store()).unwrap();
assert!(matches!(
store.library(&edit.file, Access::View, LibraryReq::Fetch { have: None, usage: None }),
store.library(std::net::Ipv4Addr::LOCALHOST.into(), &edit.file, Access::View, LibraryReq::Fetch { have: None, usage: None }),
LibraryReply::Library { version: 2, .. }
));
}

View File

@ -276,7 +276,7 @@ fn participants_report_files_to_the_operator() {
}
/// Connect with `X-Forwarded-For: forwarded` and open a hosting connection: what the server said.
fn via_proxy(addr: SocketAddr, forwarded: &str) -> (tungstenite::WebSocket<TcpStream>, ServerMsg) {
pub(crate) fn via_proxy(addr: SocketAddr, forwarded: &str) -> (tungstenite::WebSocket<TcpStream>, ServerMsg) {
let s = TcpStream::connect(addr).unwrap();
s.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
let mut req = format!("ws://{addr}/").into_client_request().unwrap();
@ -410,6 +410,10 @@ fn open_mode_has_safe_defaults_and_warns_the_operator() {
assert_eq!(open.file_mb(), 16, "strangers' files stay small");
assert_eq!(open.store().hosting.idle_secs, 30 * 86_400, "idle files expire");
assert_eq!(open.net().max_per_address, 32);
assert_eq!(open.net().request_idle, Duration::from_secs(20), "idle one-shot requests go quickly");
assert_eq!(open.net().reserve_for_sessions, 512 / 8);
assert_eq!(open.net().max_client_message, 17 << 20);
assert_eq!(private.net().max_client_message, NetLimits::default().max_client_message, "unchanged when private");
let w = open.warnings().join("\n");
assert!(w.contains("operator_contact is empty") && w.contains("trusted_proxy"), "{w}");
let set = Config { operator_contact: "abuse@example.com".into(), trusted_proxy: Some("127.0.0.1".into()), max_file_mb: Some(8), ..open };

View File

@ -63,15 +63,28 @@ const MAX_REASON: usize = 500;
/// How often idle files are looked for (seconds), besides at startup.
const SWEEP_EVERY: u64 = 600;
/// Why a download is turned away (an address's share, or everyone's, is used up for now).
pub const BUSY_DOWNLOADS: &str = "too many downloads from your address or from everyone right now: try again in a minute";
/// Why publishing without the key is refused on a server that isn't open to the public.
pub const KEY_ONLY: &str = "this server takes files only from people who have its key: add ?key=… to the address (its operator has the key)";
/// Why a file published without the key can't hold a team library on a public server (a library
/// is a whole archive checked on the server; strangers' files don't get that).
pub const NO_PUBLIC_LIBRARY: &str = "on this public server only files published with the server key can hold a team library";
/// Most files whose usage of one library is kept (the least recently changed go first), and most
/// components one file's report keeps.
const MAX_USERS: usize = 500;
const MAX_REPORTED: usize = 2_000;
const MAX_USERS: usize = 200;
const MAX_REPORTED: usize = 1_000;
/// Largest `usage.json` read.
const MAX_USAGE_FILE: u64 = 32 << 20;
const MAX_USAGE_FILE: u64 = 4 << 20;
/// A library's usage reports are kept in memory and written at most this often (seconds), and
/// when the server stops.
const USAGE_WRITE_EVERY: u64 = 60;
/// Bytes of team libraries served a second, everyone together, and at once.
const LIBRARY_RATE: f64 = (32u64 << 20) as f64;
const LIBRARY_BURST: f64 = (256u64 << 20) as f64;
/// What the store enforces.
#[derive(Clone, Debug, PartialEq)]
@ -191,6 +204,15 @@ struct Entry {
links: Vec<Link>,
/// Open: its log.
log: Option<Log>,
/// Its snapshot's and log's bytes on disk (what was written; pending records not yet).
bytes: u64,
}
/// A library's usage reports in memory, and whether (and when) they were written.
struct UsageCache {
list: Vec<FileUsage>,
dirty: bool,
written: u64,
}
/// The data directory in use (it holds the directory's lock until dropped).
@ -205,6 +227,10 @@ pub struct Store {
directory: Directory,
/// What addresses may do how often, the ones blocked, and the reports kept.
abuse: Abuse,
/// Team library bytes served, everyone together.
served: figcraft_collab_net::net::Bucket,
/// Usage reports by library file.
usage: BTreeMap<String, UsageCache>,
blocked: BTreeSet<IpAddr>,
reports: Vec<ReportInfo>,
/// When idle files are looked for next (seconds since 1970).
@ -234,7 +260,12 @@ fn clean_name(s: &str) -> String {
/// Take the data directory's lock: `None` when someone else holds it.
pub fn lock(root: &Path) -> Result<Option<File>, String> {
fs::create_dir_all(root).map_err(|e| format!("can't create the data directory {}: {e}", root.display()))?;
// Only the server's account reads what it keeps (files, addresses, reports, the key).
let mut dirs = fs::DirBuilder::new();
dirs.recursive(true);
#[cfg(unix)]
std::os::unix::fs::DirBuilderExt::mode(&mut dirs, 0o700);
dirs.create(root).map_err(|e| format!("can't create the data directory {}: {e}", root.display()))?;
let f = OpenOptions::new()
.create(true)
.truncate(false)
@ -263,12 +294,7 @@ fn load_key(root: &Path) -> Result<String, String> {
fn write_key(root: &Path) -> Result<String, String> {
let key = token::secret()?;
let path = root.join(KEY);
disk::write_atomic(&path, format!("{key}\n").as_bytes()).map_err(|e| format!("can't write {}: {e}", path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let _ = fs::set_permissions(&path, fs::Permissions::from_mode(0o600));
}
disk::write_atomic_private(&path, format!("{key}\n").as_bytes()).map_err(|e| format!("can't write {}: {e}", path.display()))?;
Ok(key)
}
@ -295,6 +321,8 @@ impl Store {
problems,
directory,
abuse,
served: figcraft_collab_net::net::Bucket::new(LIBRARY_RATE, LIBRARY_BURST),
usage: BTreeMap::new(),
blocked: BTreeSet::new(),
reports: Vec::new(),
next_sweep: 0,
@ -371,7 +399,8 @@ impl Store {
{
log::info!("file at {}: its log predates the last compaction; it is ignored", path.display());
}
Ok(Entry { meta, links, log: None })
let bytes = [SNAPSHOT, LOG].iter().filter_map(|n| fs::metadata(path.join(n)).ok()).map(|m| m.len()).sum();
Ok(Entry { meta, links, log: None, bytes })
}
/// The server key (for managing files).
@ -413,10 +442,9 @@ impl Store {
self.files.get_mut(id).ok_or_else(|| "this file isn't on the server (it may have been removed)".to_string())
}
/// A file's bytes on disk (its snapshot and log).
/// A file's bytes on disk (its snapshot and log), with the log records not written yet.
fn bytes(&self, id: &str) -> u64 {
let dir = self.dir(id);
[SNAPSHOT, LOG].iter().filter_map(|n| fs::metadata(dir.join(n)).ok()).map(|m| m.len()).sum()
self.files.get(id).map_or(0, |e| e.bytes + e.log.as_ref().map_or(0, |l| l.pending.len() as u64))
}
/// When a file published without the key is deleted unless someone opens it.
@ -530,9 +558,9 @@ impl Store {
let _ = fs::remove_dir_all(&dir);
return Err(format!("couldn't keep the file: {e}"));
}
let entry = Entry { meta, links: vec![edit.clone(), view.clone()], log: None };
let info = self.info(&id, &entry, None);
let entry = Entry { meta, links: vec![edit.clone(), view.clone()], log: None, bytes: (bytes.len() + LOG_HEADER) as u64 };
self.files.insert(id.clone(), entry);
let info = self.files.get(&id).map(|e| self.info(&id, e, None)).ok_or("the file vanished")?;
match ip {
Some(ip) => log::info!("event=published file={id} bytes={} ip={ip} by=public", bytes.len()),
None => log::info!("event=published file={id} bytes={} by=key", bytes.len()),
@ -598,6 +626,17 @@ impl Store {
}
}
/// Make an open file's log refuse writes (tests: a full or failing disk).
#[cfg(test)]
pub fn break_log(&mut self, file: &str) {
let path = self.dir(file).join(LOG);
if let Some(log) = self.files.get_mut(file).and_then(|e| e.log.as_mut())
&& let Ok(f) = File::open(path)
{
log.file = f;
}
}
/// Note that a file was opened now (it is used: it doesn't expire).
fn touch(&mut self, file: &str) {
let path = self.dir(file).join(META);
@ -722,13 +761,19 @@ impl Store {
Ok(())
}
/// A file's team library, unless its version is `have`.
fn fetch_library(&mut self, file: &str, have: Option<u64>) -> Result<LibraryReply, String> {
/// A file's team library for `ip`, unless its version is `have`; within what one address and
/// everyone may download (the size is known before anything is read).
fn fetch_library(&mut self, ip: IpAddr, file: &str, have: Option<u64>) -> Result<LibraryReply, String> {
let (path, max) = (self.dir(file).join(LIBRARY), self.cfg.max_file_bytes);
let Some(lib) = self.entry(file)?.meta.library.clone() else { return Ok(LibraryReply::Empty) };
if have == Some(lib.version) {
return Ok(LibraryReply::Unchanged { version: lib.version });
}
let size = fs::metadata(&path).map_or(0, |m| m.len());
let now = Instant::now();
if !self.abuse.allow(ip, Action::Download(usize::try_from(size).unwrap_or(usize::MAX)), now) || !self.served.take_n(now, size as f64) {
return Err(BUSY_DOWNLOADS.into());
}
Ok(LibraryReply::Library { version: lib.version, archive: disk::read_capped(&path, max)? })
}
@ -738,10 +783,15 @@ impl Store {
// move it to a worker if large libraries make that noticeable.
pub fn publish_library(&mut self, file: &str, archive: &[u8]) -> Result<u64, String> {
let max = self.cfg.max_file_bytes;
if self.cfg.hosting.open && self.entry(file)?.meta.owner.is_some() {
return Err(NO_PUBLIC_LIBRARY.into());
}
if archive.len() as u64 > max {
return Err(format!("the library is {} MiB; this server keeps files up to {} MiB", archive.len() >> 20, max >> 20));
}
let lib = match figcraft_format::read_library(Cursor::new(archive)) {
// Everything it unpacks to counts against the same cap, so a small archive can't expand
// into a large library in memory.
let lib = match figcraft_format::read_library_within(Cursor::new(archive), max) {
Ok(Some(lib)) => lib,
Ok(None) => return Err("that isn't a published library".into()),
Err(e) => return Err(format!("the library can't be read: {e}")),
@ -772,40 +822,98 @@ impl Store {
Ok(m.version)
}
/// The usage reports kept for a file's library (none when it has no library).
pub fn library_usage(&mut self, file: &str) -> Result<Vec<FileUsage>, String> {
let path = self.dir(file).join(USAGE);
if self.entry(file)?.meta.library.is_none() || !path.exists() {
return Ok(Vec::new());
/// A library's usage reports in memory (read from disk the first time; a damaged list starts
/// over: it is only statistics). `None` when the file has no library.
fn usage_of(&mut self, file: &str) -> Result<Option<&mut UsageCache>, String> {
if self.entry(file)?.meta.library.is_none() {
return Ok(None);
}
serde_json::from_slice(&disk::read_capped(&path, MAX_USAGE_FILE)?).map_err(|e| format!("{USAGE} is damaged ({e})"))
if !self.usage.contains_key(file) {
let path = self.dir(file).join(USAGE);
let list = if path.exists() {
disk::read_capped(&path, MAX_USAGE_FILE).ok().and_then(|b| serde_json::from_slice(&b).ok()).unwrap_or_default()
} else {
Vec::new()
};
self.usage.insert(file.to_string(), UsageCache { list, dirty: false, written: now() });
}
Ok(self.usage.get_mut(file))
}
/// Keep how one file uses a file's library (written only when it changed).
/// The usage reports kept for a file's library (none when it has no library).
pub fn library_usage(&mut self, file: &str) -> Result<Vec<FileUsage>, String> {
Ok(self.usage_of(file)?.map(|u| u.list.clone()).unwrap_or_default())
}
/// Keep how one file uses a file's library (in memory; written when it changed, at most every
/// [`USAGE_WRITE_EVERY`] seconds).
pub fn report_usage(&mut self, file: &str, usage: Usage) -> Result<(), String> {
let name = clean_name(&usage.file);
if name.is_empty() || self.entry(file)?.meta.library.is_none() {
if name.is_empty() {
return Ok(());
}
let mut components: Vec<_> = usage.components.into_iter().filter(|(_, n)| *n > 0).collect();
let mut components: Vec<_> = usage.components.into_iter().filter(|(_, n)| *n > 0).take(MAX_REPORTED * 4).collect();
components.sort();
components.dedup_by_key(|c| c.0);
components.truncate(MAX_REPORTED);
// A damaged report list starts over (it is only statistics).
let mut all = self.library_usage(file).unwrap_or_default();
match all.iter_mut().find(|f| f.usage.file == name) {
let Some(u) = self.usage_of(file)? else { return Ok(()) };
match u.list.iter_mut().find(|f| f.usage.file == name) {
Some(f) if f.usage.components == components => return Ok(()),
Some(f) => {
f.usage.components = components;
f.updated = now();
}
None => all.push(FileUsage { usage: Usage { file: name, components }, updated: now() }),
None => u.list.push(FileUsage { usage: Usage { file: name, components }, updated: now() }),
}
if all.len() > MAX_USERS {
all.sort_by_key(|f| std::cmp::Reverse(f.updated));
all.truncate(MAX_USERS);
if u.list.len() > MAX_USERS {
u.list.sort_by_key(|f| std::cmp::Reverse(f.updated));
u.list.truncate(MAX_USERS);
}
write_json(&self.dir(file).join(USAGE), &all)
u.dirty = true;
if now() >= u.written.saturating_add(USAGE_WRITE_EVERY) {
self.write_usage(false);
}
Ok(())
}
/// Write the usage reports that changed (`all`: whenever they were last written).
fn write_usage(&mut self, all: bool) {
let t = now();
let root = self.root.join(FILES);
for (file, u) in &mut self.usage {
if u.dirty && (all || t >= u.written.saturating_add(USAGE_WRITE_EVERY)) {
if let Err(e) = write_json(&root.join(file).join(USAGE), &u.list) {
log::warn!("file {file}: library usage couldn't be kept: {e}");
}
(u.dirty, u.written) = (false, t);
}
}
}
/// Whether `file` has room for about `n` more bytes: its cap, and for a file published
/// without the key its address's share and everyone's. A log that took it over its cap is
/// compacted first (the state may be smaller than its history).
fn room_for(&mut self, file: &str, server: &Server, n: u64) -> Result<(), String> {
let max = self.cfg.max_file_bytes;
let over = |s: &Store| s.bytes(file).saturating_add(n) > max;
if over(self) && self.files.get(file).and_then(|e| e.log.as_ref()).is_some_and(|l| l.entries > 0 || !l.pending.is_empty()) {
self.compact(file, server);
}
if over(self) {
return Err(format!("this file has reached the server's limit of {} MiB: that change wasn't kept", max >> 20));
}
let h = &self.cfg.hosting;
let Some(ip) = self.files.get(file).filter(|e| e.meta.owner.is_some()).and_then(|e| e.meta.ip.clone()) else { return Ok(()) };
let (max_ip, max_all) = (h.max_bytes_per_ip, h.max_total_bytes);
let used_ip: u64 = self.files.iter().filter(|(_, e)| e.meta.ip.as_deref() == Some(ip.as_str())).map(|(id, _)| self.bytes(id)).sum();
if used_ip.saturating_add(n) > max_ip {
return Err(format!("the files published from this file's address have reached their {} MiB: that change wasn't kept", max_ip >> 20));
}
let used_all: u64 = self.files.keys().map(|id| self.bytes(id)).sum();
if used_all.saturating_add(n) > max_all {
return Err("this server is full: that change wasn't kept".into());
}
Ok(())
}
/// Load a file's session: its snapshot, then its log (a damaged end is dropped and reported).
@ -868,6 +976,7 @@ impl Store {
disk::write_atomic(&dir.join(LOG), &disk::log_header(generation)).map_err(|e| format!("couldn't start the log: {e}"))?;
let f = OpenOptions::new().append(true).open(dir.join(LOG)).map_err(|e| format!("couldn't open the log: {e}"))?;
if let Ok(e) = self.entry(file) {
e.bytes = (bytes.len() + LOG_HEADER) as u64;
e.meta.seq = server.seq();
let meta = e.meta.clone();
if let Err(err) = write_json(&dir.join(META), &meta) {
@ -892,27 +1001,43 @@ impl Store {
}
}
/// Write and sync what's pending in every open log.
fn flush_all(&mut self) {
/// Write and sync what's pending in every open log; the files whose log couldn't be written.
/// Those are closed here (their pending records dropped: nobody was told of them) and open
/// again from what is on disk.
fn flush_all(&mut self) -> Vec<String> {
let mut failed = Vec::new();
for (id, e) in &mut self.files {
if let Some(log) = e.log.as_mut() {
flush_log(id, log);
match flush_log(id, log) {
Ok(n) => e.bytes += n,
Err(()) => {
e.log = None;
failed.push(id.clone());
}
}
}
}
failed
}
}
/// Write and sync a log's pending records (kept for the next try when that fails).
fn flush_log(id: &str, log: &mut Log) {
/// Write and sync a log's pending records: the bytes written, or `Err` when that failed (the log
/// can't be trusted to hold them).
fn flush_log(id: &str, log: &mut Log) -> Result<u64, ()> {
if log.pending.is_empty() {
return;
return Ok(0);
}
match log.file.write_all(&log.pending).and_then(|()| log.file.sync_data()) {
Ok(()) => {
log.bytes += log.pending.len() as u64;
let n = log.pending.len() as u64;
log.bytes += n;
log.pending.clear();
Ok(n)
}
Err(e) => {
log::error!("file {id}: couldn't write its log: {e}");
Err(())
}
Err(e) => log::error!("file {id}: couldn't write its log: {e}"),
}
}
@ -973,13 +1098,17 @@ impl Backend for Store {
}
}
fn flush(&mut self) {
self.flush_all();
fn flush(&mut self) -> Vec<String> {
self.flush_all()
}
fn room(&mut self, file: &str, server: &Server, bytes: usize) -> Result<(), String> {
self.room_for(file, server, bytes as u64)
}
fn idle(&mut self, file: &str, server: &mut Server) -> bool {
let dirty = self.files.get_mut(file).and_then(|e| e.log.as_mut()).is_some_and(|log| {
flush_log(file, log);
let _ = flush_log(file, log);
log.dirty
});
if dirty {
@ -994,6 +1123,7 @@ impl Backend for Store {
}
fn stop(&mut self, open: Vec<(&str, &mut Server)>) {
self.write_usage(true);
self.flush_all();
for (file, server) in open {
if self.files.get(file).and_then(|e| e.log.as_ref()).is_some_and(|l| l.dirty) {
@ -1015,6 +1145,7 @@ impl Backend for Store {
fn connect(&mut self, ip: IpAddr) -> Result<(), String> {
if now() >= self.next_sweep {
self.sweep();
self.write_usage(false);
}
if self.blocked.contains(&ip) {
return Err("this server doesn't accept connections from your address".into());
@ -1113,7 +1244,7 @@ impl Backend for Store {
}
}
fn library(&mut self, file: &str, access: Access, req: LibraryReq) -> LibraryReply {
fn library(&mut self, ip: IpAddr, file: &str, access: Access, req: LibraryReq) -> LibraryReply {
let failed = |reason: String| LibraryReply::Failed { reason };
match req {
LibraryReq::Fetch { have, usage } => {
@ -1122,7 +1253,7 @@ impl Backend for Store {
{
log::warn!("file {file}: a library usage report couldn't be kept: {e}");
}
self.fetch_library(file, have).unwrap_or_else(failed)
self.fetch_library(ip, file, have).unwrap_or_else(failed)
}
LibraryReq::Publish { .. } if access != Access::Edit => {
failed("you can only view this file: publishing its library needs an edit link".into())
@ -1134,8 +1265,15 @@ impl Backend for Store {
}
}
fn plugins(&mut self, publisher: bool, req: PluginReq) -> PluginReply {
self.directory.handle(publisher, req)
fn plugins(&mut self, ip: IpAddr, publisher: bool, req: PluginReq) -> PluginReply {
let reply = self.directory.handle(publisher, req);
// One address's share of the directory's download budget (the bundle is dropped unsent).
if let PluginReply::Bundle { bundle, .. } = &reply
&& !self.abuse.allow(ip, Action::Download(bundle.len()), Instant::now())
{
return PluginReply::Failed { reason: BUSY_DOWNLOADS.into() };
}
reply
}
fn closing(&self) -> Closing {

View File

@ -123,6 +123,11 @@ impl Raw {
self.0.send(Message::Binary(Bytes::from(message::encode(m).unwrap()))).unwrap();
}
/// Send without waiting to be read (`Err` once the server has hung up).
pub fn try_send(&mut self, m: &ClientMsg) -> tungstenite::Result<()> {
self.0.send(Message::Binary(Bytes::from(message::encode(m).unwrap())))
}
pub fn next(&mut self) -> ServerMsg {
loop {
if let Message::Binary(b) = self.0.read().unwrap() {

View File

@ -129,8 +129,17 @@ pub trait Backend: Send + 'static {
fn changed(&mut self, _file: &str, _server: &mut Server) {}
/// Make what [`changed`](Self::changed) kept durable; the hub sends the servers' answers
/// after this.
fn flush(&mut self) {}
/// after this. The files whose changes couldn't be made durable: nobody hears of them, the
/// files are closed (the backend forgets their open state) and everyone in them reconnects.
fn flush(&mut self) -> Vec<String> {
Vec::new()
}
/// Whether `file` has room for a batch of about `bytes` (its size on the wire): why not, and
/// the batch is refused.
fn room(&mut self, _file: &str, _server: &Server, _bytes: usize) -> Result<(), String> {
Ok(())
}
/// Everyone left `file`: save it. True closes it (the next hello opens it again).
fn idle(&mut self, _file: &str, _server: &mut Server) -> bool {
@ -152,13 +161,13 @@ pub trait Backend: Send + 'static {
/// A request on a library connection to `file`, opened with a link of `access` (checked by
/// [`admit`](Self::admit)).
fn library(&mut self, _file: &str, _access: Access, _req: LibraryReq) -> LibraryReply {
fn library(&mut self, _ip: IpAddr, _file: &str, _access: Access, _req: LibraryReq) -> LibraryReply {
LibraryReply::Failed { reason: "this session is hosted by an app: team libraries live on a figcraft-server".into() }
}
/// A request on a plugin directory connection; `publisher`: it was opened with the right key
/// (checked by [`manage`](Self::manage)).
fn plugins(&mut self, _publisher: bool, _req: PluginReq) -> PluginReply {
fn plugins(&mut self, _ip: IpAddr, _publisher: bool, _req: PluginReq) -> PluginReply {
PluginReply::Failed { reason: "this session is hosted by an app: plugin directories live on a figcraft-server".into() }
}
@ -200,12 +209,77 @@ pub trait Backend: Send + 'static {
fn closing(&self) -> Closing;
}
/// A message for one connection's writer.
/// A message for one connection's writer: bytes (with what they count against, released when
/// they are sent or dropped), or the close.
enum Out {
Bytes(Bytes),
Bytes(Bytes, Option<Charge>),
Close(CloseCode, String),
}
/// Bytes waiting in a connection's queue, counted against its cap and the server's until dropped.
struct Charge {
n: usize,
mine: Arc<AtomicUsize>,
all: Arc<AtomicUsize>,
}
impl Drop for Charge {
fn drop(&mut self) {
self.mine.fetch_sub(self.n, Ordering::Relaxed);
self.all.fetch_sub(self.n, Ordering::Relaxed);
}
}
/// One connection's way out: its writer's queue, capped in messages and in bytes (its own and,
/// with every other connection's, the server's), so a client that doesn't read can't make the
/// server hold more than that for it.
#[derive(Clone)]
struct Line {
tx: SyncSender<Out>,
queued: Arc<AtomicUsize>,
all: Arc<AtomicUsize>,
max: usize,
max_all: usize,
}
impl Line {
/// Queue `b`; `Full` when it would go over a cap (one message always fits an empty queue, so a
/// welcome as large as the largest file gets out).
fn push(&self, b: Bytes) -> Result<(), TrySendError<()>> {
let n = b.len();
let mine = self.queued.load(Ordering::Relaxed);
if (mine > 0 && mine.saturating_add(n) > self.max) || self.all.load(Ordering::Relaxed).saturating_add(n) > self.max_all {
return Err(TrySendError::Full(()));
}
self.queued.fetch_add(n, Ordering::Relaxed);
self.all.fetch_add(n, Ordering::Relaxed);
let charge = Charge { n, mine: self.queued.clone(), all: self.all.clone() };
self.tx.try_send(Out::Bytes(b, Some(charge))).map_err(|e| match e {
TrySendError::Full(_) => TrySendError::Full(()),
TrySendError::Disconnected(_) => TrySendError::Disconnected(()),
})
}
}
/// Messages handed to the hub thread and not handled yet, in bytes: past the budget a reader
/// drops its connection instead of piling more on a busy hub.
#[derive(Clone)]
struct InFlight(Arc<AtomicUsize>);
impl InFlight {
fn take(&self, n: usize, max: usize) -> bool {
if self.0.fetch_add(n, Ordering::Relaxed).saturating_add(n) > max {
self.0.fetch_sub(n, Ordering::Relaxed);
return false;
}
true
}
fn give(&self, n: usize) {
self.0.fetch_sub(n, Ordering::Relaxed);
}
}
enum HubIn {
/// A connection's first message (`Hello`, `Manage`, `OpenLibrary`, `OpenPlugins` or
/// `OpenHosting`) and its client's address ([`ip_key`]).
@ -213,12 +287,15 @@ enum HubIn {
conn: u64,
ip: IpAddr,
first: ClientMsg,
out: SyncSender<Out>,
out: Line,
sock: TcpStream,
/// Its size (in flight until handled).
bytes: usize,
},
Msg {
conn: u64,
msg: ClientMsg,
bytes: usize,
},
Gone {
conn: u64,
@ -255,6 +332,9 @@ pub struct Hosting {
hub: Sender<HubIn>,
stop: Arc<AtomicBool>,
open: Arc<AtomicUsize>,
/// Bytes queued for every connection together, and in flight to the hub.
queued: Arc<AtomicUsize>,
in_flight: InFlight,
hub_thread: Option<JoinHandle<()>>,
accept_thread: Option<JoinHandle<()>>,
}
@ -267,7 +347,8 @@ impl Hosting {
pub(crate) fn start<B: Backend>(listener: TcpListener, backend: B, limits: NetLimits, local: Option<Local>) -> io::Result<Hosting> {
let addr = listener.local_addr()?;
let mut hub = Hub { backend, rooms: HashMap::new(), conns: HashMap::new(), local: None, outbox: Vec::new() };
let in_flight = InFlight(Arc::default());
let mut hub = Hub { backend, rooms: HashMap::new(), conns: HashMap::new(), local: None, outbox: Vec::new(), in_flight: in_flight.clone() };
if let Some(l) = local {
let mut room = Room::new(l.server);
room.local = Some((l.id, l.inbox));
@ -278,9 +359,11 @@ impl Hosting {
let stop = Arc::new(AtomicBool::new(false));
let open = Arc::new(AtomicUsize::new(0));
let hub_thread = spawn("live-hub", move || hub.run(&hub_rx))?;
let shared = Shared { per_address: PerAddress::default(), queued: Arc::default(), in_flight: in_flight.clone() };
let queued = shared.queued.clone();
let accept_thread = {
let (hub_tx, stop, open) = (hub_tx.clone(), stop.clone(), open.clone());
spawn("live-accept", move || accept(&listener, &hub_tx, &stop, &open, limits))
spawn("live-accept", move || accept(&listener, &hub_tx, &stop, &open, limits, shared))
};
let accept_thread = match accept_thread {
Ok(t) => t,
@ -291,7 +374,17 @@ impl Hosting {
}
};
log::info!("live session: listening on {addr}");
Ok(Hosting { addr, hub: hub_tx, stop, open, hub_thread: Some(hub_thread), accept_thread: Some(accept_thread) })
Ok(Hosting { addr, hub: hub_tx, stop, open, queued, in_flight, hub_thread: Some(hub_thread), accept_thread: Some(accept_thread) })
}
/// Bytes waiting to be sent to clients, all connections together.
pub fn queued_bytes(&self) -> usize {
self.queued.load(Ordering::Relaxed)
}
/// Bytes of messages read and not handled by the hub yet.
pub fn in_flight_bytes(&self) -> usize {
self.in_flight.0.load(Ordering::Relaxed)
}
/// Where it listens.
@ -369,28 +462,57 @@ impl Drop for Open {
}
}
/// Open connections per client address ([`ip_key`]); an address leaves when its last one closes,
/// so the table never holds more than the open connections.
#[derive(Clone, Default)]
struct PerAddress(Arc<Mutex<HashMap<IpAddr, usize>>>);
/// A block of addresses: its first address and its prefix length (2001:db8:: as a /64 and as a
/// /48 are different blocks).
pub type Block = (IpAddr, u8);
/// One connection counted for its address until dropped.
/// The blocks a client's address ([`ip_key`]) counts in, with how many times its own limits each
/// gets: the address itself (an IPv4 /32 or an IPv6 /64, ×1) and, for IPv6, its /56 (×4) and /48
/// (×16), so a tunnel's free /48 doesn't multiply the limits 65,536 times.
pub fn ip_blocks(key: IpAddr) -> Vec<(Block, u32)> {
let IpAddr::V6(v6) = key else { return vec![((key, 32), 1)] };
let [a, b, c, d, ..] = v6.segments();
vec![
((IpAddr::V6(Ipv6Addr::new(a, b, c, d, 0, 0, 0, 0)), 64), 1),
((IpAddr::V6(Ipv6Addr::new(a, b, c, d & 0xff00, 0, 0, 0, 0)), 56), 4),
((IpAddr::V6(Ipv6Addr::new(a, b, c, 0, 0, 0, 0, 0)), 48), 16),
]
}
/// Open connections per client address ([`ip_key`], and its wider IPv6 blocks); an address
/// leaves when its last one closes, so the table never holds more than the open connections.
#[derive(Clone, Default)]
struct PerAddress(Arc<Mutex<HashMap<Block, usize>>>);
/// One connection counted for a block until dropped.
struct Counted {
table: PerAddress,
ip: IpAddr,
ip: Block,
}
impl PerAddress {
/// Count a connection from `ip`, unless it has `cap` already (0: no cap).
fn enter(&self, ip: IpAddr, cap: usize) -> Option<Counted> {
/// Count a connection from block `ip`, unless it has `cap` already (0: no cap).
fn enter(&self, ip: Block, cap: usize) -> Option<Counted> {
let mut t = self.0.lock().unwrap_or_else(PoisonError::into_inner);
let n = t.entry(ip).or_insert(0);
if cap > 0 && *n >= cap {
if *n == 0 {
t.remove(&ip);
}
return None;
}
*n += 1;
Some(Counted { table: self.clone(), ip })
}
/// Count a connection from `ip` in each of its blocks, unless one of them is full.
fn enter_all(&self, ip: IpAddr, cap: usize) -> Option<Vec<Counted>> {
let mut held = Vec::new();
for (block, times) in ip_blocks(ip) {
held.push(self.enter(block, cap.saturating_mul(times as usize))?);
}
Some(held)
}
}
impl Drop for Counted {
@ -459,8 +581,16 @@ fn forwarded_for(req: &Request) -> Option<IpAddr> {
entry.parse::<IpAddr>().ok().or_else(|| entry.parse::<SocketAddr>().ok().map(|a| a.ip()))
}
fn accept(listener: &TcpListener, hub: &Sender<HubIn>, stop: &AtomicBool, open: &Arc<AtomicUsize>, limits: NetLimits) {
let per_address = PerAddress::default();
/// What every connection shares: the count per address, the bytes queued for all of them and the
/// bytes in flight to the hub.
#[derive(Clone)]
struct Shared {
per_address: PerAddress,
queued: Arc<AtomicUsize>,
in_flight: InFlight,
}
fn accept(listener: &TcpListener, hub: &Sender<HubIn>, stop: &AtomicBool, open: &Arc<AtomicUsize>, limits: NetLimits, shared: Shared) {
let mut next = 0u64;
for stream in listener.incoming() {
if stop.load(Ordering::SeqCst) {
@ -482,8 +612,8 @@ fn accept(listener: &TcpListener, hub: &Sender<HubIn>, stop: &AtomicBool, open:
open.fetch_add(1, Ordering::Relaxed);
let guard = Open(open.clone());
next += 1;
let (conn, hub, per_address) = (next, hub.clone(), per_address.clone());
if let Err(e) = spawn("live-conn", move || serve(stream, conn, &hub, limits, guard, &per_address)) {
let (conn, hub, shared) = (next, hub.clone(), shared.clone());
if let Err(e) = spawn("live-conn", move || serve(stream, conn, &hub, limits, guard, &shared)) {
log::warn!("live session: can't start a connection thread: {e}");
}
}
@ -508,30 +638,43 @@ fn encode(msg: &ServerMsg) -> Option<Bytes> {
}
}
/// Queue `Refused {reason}` and a close for one connection.
fn refuse(out: &SyncSender<Out>, why: &str) {
/// Queue `Refused {reason}` and a close for one connection (small: not counted against its caps).
fn refuse(out: &Line, why: &str) {
if let Some(b) = encode(&ServerMsg::Refused { reason: why.to_string() }) {
let _ = out.try_send(Out::Bytes(b));
let _ = out.tx.try_send(Out::Bytes(b, None));
}
let _ = out.try_send(Out::Close(CloseCode::Policy, reason(why)));
let _ = out.tx.try_send(Out::Close(CloseCode::Policy, reason(why)));
}
/// Read the next binary message (skipping pings), or why the connection should end.
fn next_message(ws: &mut WebSocket<Duplex>, max: usize) -> Result<Option<ClientMsg>, &'static str> {
/// What the server tells a client when it is too busy to take more from it.
const BUSY: &str = "the server is busy: try again in a moment";
/// Read the next binary message (skipping pings) and its size, or why the connection should end.
fn next_message(ws: &mut WebSocket<Duplex>, max: usize) -> Result<Option<(ClientMsg, usize)>, String> {
loop {
return match ws.read() {
Ok(Message::Binary(b)) => message::decode::<ClientMsg>(&b, max).map(Some).map_err(|_| "that message couldn't be read"),
Ok(Message::Text(_)) => Err("only binary messages are spoken here"),
Ok(Message::Binary(b)) => {
message::decode::<ClientMsg>(&b, max).map(|m| Some((m, b.len()))).map_err(|_| "that message couldn't be read".to_string())
}
Ok(Message::Text(_)) => Err("only binary messages are spoken here".into()),
Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_)) => continue,
Ok(Message::Close(_)) => Ok(None),
Err(tungstenite::Error::Capacity(_)) => Err("message too large"),
Err(tungstenite::Error::Capacity(_)) => Err(too_large(max)),
Err(_) => Ok(None),
};
}
}
/// Why a message over `max` bytes is refused.
fn too_large(max: usize) -> String {
match max >> 20 {
0 => format!("message too large: this server takes at most {} KiB at once here", max >> 10),
mib => format!("message too large: this server takes at most {mib} MiB at once (a change that big can't be kept here)"),
}
}
/// One connection's reader (the writer runs on its own thread).
fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimits, _open: Open, per_address: &PerAddress) {
fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimits, open: Open, shared: &Shared) {
let _ = stream.set_nodelay(true);
if stream.set_read_timeout(Some(limits.hello_timeout)).is_err() || stream.set_write_timeout(Some(limits.write_timeout)).is_err() {
return;
@ -564,8 +707,9 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit
log::debug!("live session: connection {conn} came through the proxy without X-Forwarded-For");
}
let ip = ip_key(forwarded.unwrap_or(peer));
let counted = per_address.enter(ip, limits.max_per_address);
let (out, rx) = sync_channel::<Out>(limits.queue.max(4));
let counted = shared.per_address.enter_all(ip, limits.max_per_address);
let (tx, rx) = sync_channel::<Out>(limits.queue.max(4));
let out = Line { tx, queued: Arc::default(), all: shared.queued.clone(), max: limits.max_queued_bytes, max_all: limits.max_queued_total };
let writer =
match sock.try_clone().map(|s| spawn("live-write", move || write_loop(WebSocket::from_raw_socket(wr, Role::Server, Some(cfg)), &rx, &s))) {
Ok(Ok(t)) => t,
@ -578,7 +722,7 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit
if let Some(why) = why {
refuse(&out, why);
} else {
let _ = out.try_send(Out::Close(CloseCode::Normal, String::new()));
let _ = out.tx.try_send(Out::Close(CloseCode::Normal, String::new()));
}
};
if counted.is_none() {
@ -586,23 +730,38 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit
end(Some("too many connections from your address: close some and try again"));
} else {
match next_message(&mut ws, limits.max_hello) {
Ok(Some(
Ok(Some((
first @ (ClientMsg::Hello(_)
| ClientMsg::Manage(_)
| ClientMsg::OpenLibrary(_)
| ClientMsg::OpenPlugins(_)
| ClientMsg::OpenHosting(_)),
)) => {
ws.set_config(|c| *c = cfg);
let opened = sock.try_clone().is_ok_and(|s| hub.send(HubIn::Open { conn, ip, first, out: out.clone(), sock: s }).is_ok());
if opened && ws.get_ref().set_read_timeout(Some(limits.idle_timeout)).is_ok() {
read_loop(&mut ws, conn, hub, limits, &end);
let _ = hub.send(HubIn::Gone { conn });
bytes,
))) => {
// Sessions and the operator may use the last connections; one-shot requests
// (hosting, libraries, browsing plugins) leave them free.
let session = matches!(first, ClientMsg::Hello(_) | ClientMsg::Manage(_));
let max = match &first {
ClientMsg::OpenPlugins(o) if o.key.is_empty() => limits.max_request,
_ => limits.max_client_message,
};
let idle = if session { limits.idle_timeout } else { limits.request_idle };
let reserved = !session && open.0.load(Ordering::Relaxed) > limits.max_connections.saturating_sub(limits.reserve_for_sessions);
// (In flight is only taken when the connection goes on: the hub gives it back.)
if reserved || !shared.in_flight.take(bytes, limits.max_in_flight) {
end(Some(BUSY));
} else {
ws.set_config(|c| *c = config(max));
let opened = sock.try_clone().is_ok_and(|s| hub.send(HubIn::Open { conn, ip, first, out: out.clone(), sock: s, bytes }).is_ok());
if opened && ws.get_ref().set_read_timeout(Some(idle)).is_ok() {
read_loop(&mut ws, conn, hub, limits, max, &shared.in_flight, &end);
let _ = hub.send(HubIn::Gone { conn });
}
}
}
Ok(Some(_)) => end(Some("say hello first")),
Ok(None) => {}
Err(why) => end(Some(why)),
Err(why) => end(Some(&why)),
}
}
drop(counted);
@ -613,12 +772,20 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit
shutdown(&sock);
}
fn read_loop(ws: &mut WebSocket<Duplex>, conn: u64, hub: &Sender<HubIn>, limits: NetLimits, end: &dyn Fn(Option<&str>)) {
fn read_loop(
ws: &mut WebSocket<Duplex>,
conn: u64,
hub: &Sender<HubIn>,
limits: NetLimits,
max: usize,
in_flight: &InFlight,
end: &dyn Fn(Option<&str>),
) {
let mut bucket = Bucket::new(limits.rate, limits.burst);
let mut signals = Bucket::new(SIGNAL_RATE, SIGNAL_BURST);
loop {
match next_message(ws, limits.max_client_message) {
Ok(Some(msg)) => {
match next_message(ws, max) {
Ok(Some((msg, bytes))) => {
let now = Instant::now();
if !bucket.take(now) {
log::info!("live session: connection {conn} sends too much; disconnecting it");
@ -629,12 +796,16 @@ fn read_loop(ws: &mut WebSocket<Duplex>, conn: u64, hub: &Sender<HubIn>, limits:
log::debug!("live session: connection {conn} sends signals too fast; dropping one");
continue;
}
if hub.send(HubIn::Msg { conn, msg }).is_err() {
if !in_flight.take(bytes, limits.max_in_flight) {
log::info!("live session: the hub is behind; dropping connection {conn}");
return end(Some(BUSY));
}
if hub.send(HubIn::Msg { conn, msg, bytes }).is_err() {
return;
}
}
Ok(None) => return end(None),
Err(why) => return end(Some(why)),
Err(why) => return end(Some(&why)),
}
}
}
@ -643,7 +814,7 @@ fn write_loop(mut ws: WebSocket<Duplex>, rx: &Receiver<Out>, sock: &TcpStream) {
let mut close = Some((CloseCode::Normal, String::new()));
while let Ok(o) = rx.recv() {
match o {
Out::Bytes(b) => {
Out::Bytes(b, _charge) => {
if ws.send(Message::Binary(b)).is_err() {
close = None;
break;
@ -672,14 +843,16 @@ fn write_loop(mut ws: WebSocket<Duplex>, rx: &Receiver<Out>, sock: &TcpStream) {
}
/// Read and drop what the peer still sends after the close, until it hangs up (at most
/// [`LINGER`]).
/// [`LINGER`]). Raw bytes, not frames: after a frame over the size limit the WebSocket's reader
/// is mid-frame and mustn't be asked for more.
fn linger(ws: &mut WebSocket<Duplex>) {
let until = Instant::now() + LINGER;
let mut buf = [0u8; 4096];
while Instant::now() < until && ws.get_ref().set_read_timeout(Some(Duration::from_millis(250))).is_ok() {
match ws.read() {
Ok(Message::Close(_)) => break,
match ws.get_mut().read(&mut buf) {
Ok(0) => break,
Ok(_) => {}
Err(tungstenite::Error::Io(e)) if matches!(e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut) => {}
Err(e) if matches!(e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut | io::ErrorKind::Interrupted) => {}
Err(_) => break,
}
}
@ -691,8 +864,8 @@ enum Kind {
Joined { file: String, id: u32, link: Option<String> },
/// Managing the server's files.
Managing,
/// Using `file`'s team library through a link of `access`.
Library { file: String, access: Access },
/// Using `file`'s team library through link `link` of `access` (revoking it disconnects).
Library { file: String, access: Access, link: Option<String> },
/// Using the plugin directory; `publisher`: with the server's key.
Plugins { publisher: bool },
/// Publishing without the key, or managing the file `owns` with its owner token.
@ -701,14 +874,14 @@ enum Kind {
/// A connection whose first message the hub is looking at.
struct Pending {
out: SyncSender<Out>,
out: Line,
sock: TcpStream,
ip: IpAddr,
}
/// A connection the hub took in.
struct Conn {
out: SyncSender<Out>,
out: Line,
sock: TcpStream,
kind: Kind,
/// Its client's address ([`ip_key`]).
@ -750,8 +923,12 @@ fn unsupported(protocol: u32) -> String {
format!("protocol {protocol} isn't supported (this server speaks {PROTOCOL})")
}
/// What everyone in a file is told when its changes couldn't be saved (they reconnect).
const UNSAVED: &str = "The server couldn't save the latest changes; reconnecting";
/// Owns the open files' servers and routes their messages.
struct Hub<B> {
in_flight: InFlight,
backend: B,
rooms: HashMap<String, Room>,
conns: HashMap<u64, Conn>,
@ -796,7 +973,7 @@ impl<B: Backend> Hub<B> {
Closing::Restarting => (CloseCode::Restart, RESTARTING),
};
for c in self.conns.values() {
if c.out.try_send(Out::Close(code, why.into())).is_err() {
if c.out.tx.try_send(Out::Close(code, why.into())).is_err() {
shutdown(&c.sock);
}
}
@ -804,7 +981,8 @@ impl<B: Backend> Hub<B> {
fn handle(&mut self, m: HubIn) {
match m {
HubIn::Open { conn, ip, first, out, sock } => {
HubIn::Open { conn, ip, first, out, sock, bytes } => {
self.in_flight.give(bytes);
// The operator (with the server key) gets in from anywhere; everyone else is
// checked (blocked addresses, connection rates).
let operator = matches!(&first, ClientMsg::Manage(m) if self.backend.manage(&m.key).is_ok());
@ -822,7 +1000,8 @@ impl<B: Backend> Hub<B> {
_ => refuse(&c.out, "say hello first"),
}
}
HubIn::Msg { conn, msg } => {
HubIn::Msg { conn, msg, bytes } => {
self.in_flight.give(bytes);
let Some(c) = self.conns.get(&conn) else { return };
let ip = c.ip;
// The operator (with the server key) isn't held to the limits on addresses.
@ -830,7 +1009,7 @@ impl<B: Backend> Hub<B> {
log::info!("event=refused ip={ip} reason=messages_per_address");
return self.kick(&[conn], "too many messages from your address: wait a minute and try again");
}
self.message(conn, ip, msg);
self.message(conn, ip, msg, bytes);
}
HubIn::Gone { conn } => self.gone(conn),
HubIn::Local(p) | HubIn::LocalMsg(ClientMsg::Presence(p)) => {
@ -840,7 +1019,7 @@ impl<B: Backend> Hub<B> {
}
HubIn::LocalMsg(msg) => {
if let Some((file, id)) = self.local_id() {
self.client_message(&file, id, msg);
self.client_message(&file, id, msg, None);
}
}
#[cfg(test)]
@ -854,7 +1033,7 @@ impl<B: Backend> Hub<B> {
}
/// A message on connection `conn` (from `ip`), by what the connection is for.
fn message(&mut self, conn: u64, ip: IpAddr, msg: ClientMsg) {
fn message(&mut self, conn: u64, ip: IpAddr, msg: ClientMsg, bytes: usize) {
match self.conns.get(&conn).map(|c| &c.kind) {
Some(Kind::Joined { file, link, .. }) if matches!(msg, ClientMsg::Report { .. }) => {
let (file, link) = (file.clone(), link.clone());
@ -864,7 +1043,7 @@ impl<B: Backend> Hub<B> {
}
Some(Kind::Joined { file, id, .. }) => {
let (file, id) = (file.clone(), *id);
self.client_message(&file, id, msg);
self.client_message(&file, id, msg, Some(bytes));
}
Some(Kind::Hosting { owns }) => match msg {
ClientMsg::Admin(req) => {
@ -882,10 +1061,10 @@ impl<B: Backend> Hub<B> {
ClientMsg::Ping => self.answer(conn, &ServerMsg::Pong),
_ => self.kick(&[conn], "only requests to manage files are spoken on this connection"),
},
Some(Kind::Library { file, access }) => match msg {
Some(Kind::Library { file, access, .. }) => match msg {
ClientMsg::Library(req) => {
let (file, access) = (file.clone(), *access);
let reply = self.backend.library(&file, access, req);
let reply = self.backend.library(ip, &file, access, req);
self.answer(conn, &ServerMsg::Library(Box::new(reply)));
}
ClientMsg::Ping => self.answer(conn, &ServerMsg::Pong),
@ -893,7 +1072,7 @@ impl<B: Backend> Hub<B> {
},
Some(Kind::Plugins { publisher }) => match msg {
ClientMsg::Plugins(req) => {
let reply = self.backend.plugins(*publisher, req);
let reply = self.backend.plugins(ip, *publisher, req);
self.answer(conn, &ServerMsg::Plugins(Box::new(reply)));
}
ClientMsg::Ping => self.answer(conn, &ServerMsg::Pong),
@ -911,10 +1090,20 @@ impl<B: Backend> Hub<B> {
}
/// A message from client `id` of `file`.
fn client_message(&mut self, file: &str, id: u32, msg: ClientMsg) {
/// A message from client `id` of `file`; `bytes`: its size on the wire (a socket's).
fn client_message(&mut self, file: &str, id: u32, msg: ClientMsg, bytes: Option<usize>) {
if let ClientMsg::Presence(p) = msg {
return self.presence(file, id, p);
}
// A batch the file has no room for is refused before the session takes it (the client
// rolls it back with the reason).
if let (ClientMsg::Changes { batch_id, .. }, Some(n)) = (&msg, bytes)
&& let Some(room) = self.rooms.get(file)
&& let Err(reason) = self.backend.room(file, &room.server, n)
{
let reject = ServerMsg::Reject { batch_id: *batch_id, reason, fixes: Vec::new() };
return self.queue(file, vec![Outgoing::To(id, reject)]);
}
let Some(room) = self.rooms.get_mut(file) else { return };
let outs = room.server.receive(id, msg);
self.backend.changed(file, &mut room.server);
@ -1009,11 +1198,11 @@ impl<B: Backend> Hub<B> {
if o.protocol != PROTOCOL {
return refuse(&c.out, &unsupported(o.protocol));
}
let access = match self.backend.admit(&o.file, &o.token) {
Ok(a) => a.access,
let (access, link) = match self.backend.admit(&o.file, &o.token) {
Ok(a) => (a.access, a.link),
Err(why) => return refuse(&c.out, &why),
};
self.take_in(conn, c, Kind::Library { file: o.file, access }, &ServerMsg::Library(Box::new(LibraryReply::Ready { access })));
self.take_in(conn, c, Kind::Library { file: o.file, access, link }, &ServerMsg::Library(Box::new(LibraryReply::Ready { access })));
}
/// A plugin directory connection: anyone may browse; a key, when given, must be the server's.
@ -1059,14 +1248,20 @@ impl<B: Backend> Hub<B> {
let users: Vec<u64> = self
.conns
.iter()
.filter(|(_, c)| matches!(&c.kind, Kind::Joined { file: f, link: Some(l), .. } if *f == file && *l == link))
.filter(|(_, c)| {
matches!(&c.kind, Kind::Joined { file: f, link: Some(l), .. } | Kind::Library { file: f, link: Some(l), .. } if *f == file && *l == link)
})
.map(|(k, _)| *k)
.collect();
self.kick(&users, "the link you joined with was revoked");
}
if let Some(file) = handled.removed {
let users: Vec<u64> =
self.conns.iter().filter(|(_, c)| matches!(&c.kind, Kind::Joined { file: f, .. } if *f == file)).map(|(k, _)| *k).collect();
let users: Vec<u64> = self
.conns
.iter()
.filter(|(_, c)| matches!(&c.kind, Kind::Joined { file: f, .. } | Kind::Library { file: f, .. } if *f == file))
.map(|(k, _)| *k)
.collect();
// Closed without saving: the file is gone.
self.rooms.remove(&file);
self.kick(&users, "this file was removed from the server");
@ -1080,7 +1275,7 @@ impl<B: Backend> Hub<B> {
/// Send one message to one connection now (managing connections' answers).
fn answer(&mut self, conn: u64, msg: &ServerMsg) {
let Some(c) = self.conns.get(&conn) else { return };
let sent = encode(msg).is_some_and(|b| c.out.try_send(Out::Bytes(b)).is_ok());
let sent = encode(msg).is_some_and(|b| c.out.push(b).is_ok());
if !sent {
shutdown(&c.sock);
self.gone(conn);
@ -1111,6 +1306,24 @@ impl<B: Backend> Hub<B> {
self.close_if_empty(&file);
}
/// What changed in `file` couldn't be made durable: nobody is told it was confirmed. The file
/// is closed without saving and everyone in it reconnects (its last durable state opens again
/// and their apps send again what wasn't confirmed).
fn unsaved(&mut self, file: &str) {
log::error!("file {file}: changes couldn't be saved; reconnecting everyone in it");
self.outbox.retain(|(f, _)| f != file);
self.rooms.remove(file);
let users: Vec<u64> =
self.conns.iter().filter(|(_, c)| matches!(&c.kind, Kind::Joined { file: f, .. } if f == file)).map(|(k, _)| *k).collect();
for conn in users {
if let Some(c) = self.conns.remove(&conn)
&& c.out.tx.try_send(Out::Close(CloseCode::Restart, reason(UNSAVED))).is_err()
{
shutdown(&c.sock);
}
}
}
/// Nobody is in `file`: the backend saves it and may close it.
fn close_if_empty(&mut self, file: &str) {
let Some(room) = self.rooms.get_mut(file).filter(|r| r.empty()) else { return };
@ -1143,7 +1356,9 @@ impl<B: Backend> Hub<B> {
/// Make what changed durable, then send what the servers said; connections that can't keep up
/// are dropped (which may say more: that they left).
fn send_out(&mut self) {
self.backend.flush();
for file in self.backend.flush() {
self.unsaved(&file);
}
while !self.outbox.is_empty() {
let mut stuck = Vec::new();
for (file, o) in std::mem::take(&mut self.outbox) {
@ -1161,7 +1376,7 @@ impl<B: Backend> Hub<B> {
}
let Some((&conn, c)) = room.ids.get(&id).and_then(|conn| self.conns.get_key_value(conn)) else { continue };
let Some(b) = bytes.get_or_insert_with(|| encode(&msg)).clone() else { continue };
match c.out.try_send(Out::Bytes(b)) {
match c.out.push(b) {
Ok(()) => {}
Err(TrySendError::Full(_)) => {
log::info!("live session: client {id} can't keep up; disconnecting it");

View File

@ -33,6 +33,20 @@ pub struct NetLimits {
/// The reverse proxy in front of a host, whose `X-Forwarded-For` names the client: only
/// connections from this address are believed (the header is anyone's to forge).
pub trusted_proxy: Option<IpAddr>,
/// Bytes queued for one client before it counts as stuck (one message always fits an empty
/// queue), and for every client together.
pub max_queued_bytes: usize,
pub max_queued_total: usize,
/// Bytes of messages handed to the hub and not handled yet, all connections together: past it
/// a connection is dropped as busy instead of piling more on.
pub max_in_flight: usize,
/// Largest message on a connection that only asks small things (browsing a plugin directory).
pub max_request: usize,
/// Connections only sessions (`Hello`) and the operator may take: one-shot requests leave them.
pub reserve_for_sessions: usize,
/// A one-shot request connection (hosting, a library, a plugin directory) with no traffic for
/// this long is dropped.
pub request_idle: Duration,
}
impl Default for NetLimits {
@ -51,6 +65,12 @@ impl Default for NetLimits {
write_timeout: Duration::from_secs(10),
max_per_address: 0,
trusted_proxy: None,
max_queued_bytes: 160 << 20,
max_queued_total: usize::MAX,
max_in_flight: usize::MAX,
max_request: 64 << 10,
reserve_for_sessions: 0,
request_idle: Duration::from_secs(45),
}
}
}

View File

@ -384,7 +384,7 @@ fn oversized_messages_end_the_connection() {
let mut raw = Raw::hello(addr, TOKEN);
let Some(ServerMsg::Welcome { client_id, .. }) = raw.next() else { panic!("no welcome") };
raw.send(&ClientMsg::Presence(Presence { selection: Some((0..5000).map(NodeId).collect()), ..Presence::default() }));
assert_eq!(raw.refusal().as_deref(), Some("message too large"));
assert!(raw.refusal().is_some_and(|r| r.starts_with("message too large")));
hp.wait("it left", |e| e.iter().any(|e| matches!(e, Event::PeerLeft(id) if *id == client_id)));
// The host still serves others.
let (_a, mut ap) = join(&invite, "Ada");

View File

@ -134,4 +134,9 @@ impl Bucket {
self.tokens -= n;
true
}
/// Give back `n` tokens (up to the burst).
pub fn give(&mut self, n: f64) {
self.tokens = (self.tokens + n).min(self.burst);
}
}

View File

@ -87,6 +87,8 @@ pub struct Limits {
pub max_log_batches: usize,
/// Most changes in those batches together.
pub max_log_changes: usize,
/// About how many bytes those batches may hold together ([`server::weight`]).
pub max_log_bytes: usize,
}
impl Default for Limits {
@ -99,6 +101,7 @@ impl Default for Limits {
max_name: 64,
max_log_batches: 10_000,
max_log_changes: 1_000_000,
max_log_bytes: usize::MAX,
}
}
}

View File

@ -63,8 +63,9 @@ pub struct Server {
issued: BTreeMap<u32, Processed>,
/// The latest accepted batches, oldest first (a reconnecting client gets what it missed).
log: VecDeque<Accepted>,
/// Changes in `log`.
/// Changes in `log`, and about how many bytes it holds ([`weight`]).
log_changes: usize,
log_bytes: usize,
/// The connected clients that may only view.
viewers: BTreeSet<u32>,
/// What changed the lasting state since it was last taken (`None`: not kept).
@ -117,6 +118,7 @@ impl Server {
issued: BTreeMap::new(),
log: VecDeque::new(),
log_changes: 0,
log_bytes: 0,
viewers: BTreeSet::new(),
journal: None,
signals: BTreeSet::new(),
@ -399,14 +401,25 @@ impl Server {
/// Keep an accepted batch in the log (dropping the oldest beyond the limits).
fn remember(&mut self, acc: Accepted) {
self.log_changes += acc.changes.len();
self.log_bytes = self.log_bytes.saturating_add(weight(&acc.changes));
self.log.push_back(acc);
while self.log.len() > self.limits.max_log_batches || (self.log_changes > self.limits.max_log_changes && !self.log.is_empty()) {
while !self.log.is_empty()
&& (self.log.len() > self.limits.max_log_batches
|| self.log_changes > self.limits.max_log_changes
|| self.log_bytes > self.limits.max_log_bytes)
{
if let Some(old) = self.log.pop_front() {
self.log_changes = self.log_changes.saturating_sub(old.changes.len());
self.log_bytes = self.log_bytes.saturating_sub(weight(&old.changes));
}
}
}
/// About how many bytes the batches kept for reconnecting clients hold.
pub fn log_bytes(&self) -> usize {
self.log_bytes
}
/// Validate and apply a batch; on success it has the next sequence number.
fn accept(&mut self, from: u32, base_seq: u64, changes: &[Change]) -> Result<(), Error> {
if changes.len() > self.limits.max_changes {
@ -559,3 +572,27 @@ fn reason(e: &Error) -> String {
e => e.to_string(),
}
}
/// About how many bytes `changes` take in memory: their values' text and a little per change.
pub fn weight(changes: &[Change]) -> usize {
fn json(v: &serde_json::Value) -> usize {
match v {
serde_json::Value::String(s) => s.len() + 2,
serde_json::Value::Array(a) => a.iter().map(json).fold(2, usize::saturating_add),
serde_json::Value::Object(o) => o.iter().map(|(k, v)| k.len() + 3 + json(v)).fold(2, usize::saturating_add),
_ => 8,
}
}
changes
.iter()
.map(|c| match c {
Change::Create { kind, position, props, .. } => {
props.iter().map(|(_, v)| 16 + json(&v.0)).fold(64 + json(&kind.0) + position.len(), usize::saturating_add)
}
Change::Set { value, .. } | Change::SetDoc { value, .. } => 32 + json(&value.0),
Change::Put { value, .. } => 64 + json(&value.0),
Change::Delete { .. } => 16,
Change::Remove { .. } => 64,
})
.fold(0, usize::saturating_add)
}

View File

@ -21,7 +21,7 @@ pub mod fig;
pub mod library;
pub mod rest;
pub use library::{is_valid_library_id, read_library};
pub use library::{is_valid_library_id, read_library, read_library_within};
use std::collections::{BTreeMap, BTreeSet};
use std::io::{Read, Seek, Write};
@ -470,7 +470,7 @@ pub fn load_bytes_with_versions(bytes: &[u8]) -> Result<Loaded, FormatError> {
let hash = name.trim_start_matches("images/").to_string();
images.insert(hash, image_data(data));
}
match library::read(&mut z, &images) {
match library::read(&mut z, &images, u64::MAX) {
Ok(lib) => doc.published = lib.map(Arc::new),
Err(e) => issues.push(format!("the published library was skipped: {e}")),
}

View File

@ -57,8 +57,13 @@ pub(crate) fn write<W: Write + Seek>(z: &mut zip::ZipWriter<W>, lib: &PublishedL
}
/// The published library in an archive (`None` when the file was never published). `own` are
/// the images already read from `images/`.
pub(crate) fn read<R: Read + Seek>(z: &mut zip::ZipArchive<R>, own: &BTreeMap<String, ImageData>) -> Result<Option<PublishedLibrary>, FormatError> {
/// the images already read from `images/`; `budget` caps the bytes its snapshot and images unpack
/// to, together (beyond the per-part caps).
pub(crate) fn read<R: Read + Seek>(
z: &mut zip::ZipArchive<R>,
own: &BTreeMap<String, ImageData>,
budget: u64,
) -> Result<Option<PublishedLibrary>, FormatError> {
if z.index_for_name(MANIFEST).is_none() {
return Ok(None);
}
@ -66,9 +71,9 @@ pub(crate) fn read<R: Read + Seek>(z: &mut zip::ZipArchive<R>, own: &BTreeMap<St
if !is_valid_library_id(&manifest.id) {
return Err(corrupt("the library has an invalid id"));
}
let text = String::from_utf8(read_member(z, SNAPSHOT, MAX_MEMBER)?).map_err(corrupt)?;
let text = String::from_utf8(read_member(z, SNAPSHOT, budget.min(MAX_MEMBER))?).map_err(corrupt)?;
let mut budget = budget.saturating_sub(text.len() as u64).min(MAX_IMAGE_BYTES);
let (mut doc, _) = from_json(&text)?;
let mut budget = MAX_IMAGE_BYTES;
for hash in used_images(&doc) {
if !is_valid_hash(&hash) {
continue;
@ -97,7 +102,15 @@ pub(crate) fn read<R: Read + Seek>(z: &mut zip::ZipArchive<R>, own: &BTreeMap<St
/// when the file was never published, an error when it isn't a FigCraft archive or is damaged.
pub fn read_library<R: Read + Seek>(r: R) -> Result<Option<PublishedLibrary>, FormatError> {
let mut z = zip::ZipArchive::new(r).map_err(corrupt)?;
read(&mut z, &BTreeMap::new())
read(&mut z, &BTreeMap::new(), u64::MAX)
}
/// [`read_library`] unpacking at most `max_bytes` in all (snapshot and images: an archive that
/// would unpack to more is refused, or loses the images past it), for libraries from people the
/// reader doesn't trust (a server).
pub fn read_library_within<R: Read + Seek>(r: R, max_bytes: u64) -> Result<Option<PublishedLibrary>, FormatError> {
let mut z = zip::ZipArchive::new(r).map_err(corrupt)?;
read(&mut z, &BTreeMap::new(), max_bytes)
}
#[cfg(test)]

View File

@ -610,6 +610,13 @@ installing: [`plugins.md`](plugins.md#plugin-directories)):
with the file, the link the reporter joined with and their address; blocking an address
disconnects its connections and is kept in `blocked.json`. Both are the operator's to read with
the server's command line.
- **Resources.** Open to the public, a message is at most a file's size (checked while it is read),
a batch a file has no room for is answered with a `Reject` carrying the reason (no fixes: the
client rolls it back), the bytes queued for a client and for all of them and the bytes waiting
for the hub thread are capped (past them a connection is dropped), one-shot request connections
leave an eighth of the connections to sessions, and when a file's log can't be written nothing
of it is confirmed: everyone in it reconnects with close code 1012 and sends again what wasn't
confirmed. Revoking a link or removing a file also cuts library connections that used it.
- **Uptime checks.** A plain `GET` of a path ending in `/healthz` on the session port is answered
`200 ok` (the request line is peeked at before the WebSocket handshake, which goes on untouched
otherwise).

View File

@ -26,6 +26,7 @@ cursors, selections, text carets, cursor chat and spotlights shown live ([`colla
| Date | Change | Summary |
|---|---|---|
| 2026-10-12 | minor | Public `figcraft-server` hardened against resource exhaustion after a security review: messages capped by what a connection is for (a file's size when open), byte-bounded send queues and a bound on bytes waiting for the hub, files checked as they grow (refused batches roll back with the reason; per-address and total caps count growth), the core's recent-batch memory bounded in bytes, no confirmation for changes that couldn't be written, team libraries only in key-published files on public servers with unpacking bounded by the file cap, usage reports kept in memory, layered IPv6 limits (/64, /56, /48), a global publish rate, connections reserved for sessions, per-address download shares, library connections cut on revoke and removal, a private data directory and key, and safer install.sh, systemd and backups. Percentages and hours unchanged |
| 2026-10-12 | minor | A public `figcraft-server` (open hosting): with `open_publish = true` anyone publishes a file without the server key and gets its edit and view links plus an owner token (128 bits, shown once, stored hashed) that lists, makes and revokes its links, renames it or removes it (`collab.manageFile`); limits per address (connections at once and a minute, messages a second, publishes an hour, files and MiB kept; IPv6 by /64; `X-Forwarded-For` believed only from `trusted_proxy`; tables capped at 10,000 addresses) and for everyone (`max_total_mb`), each refusal saying why; files published without the key expire after `idle_days` unopened (the date shown to everyone in the file); moderation (`files list --sort size|age|idle`, `block-ip`, `reports`, a Report button and `collab.report`); the operator's contact and terms shown in the Share dialog; `GET /healthz` on the same port; `packaging/server/install.sh` for a Debian/Ubuntu machine behind Caddy and a nightly backup timer. Protocol additions within version 2, gated like signals (older apps get clear refusals). Hosting infrastructure rather than a Figma feature: percentages and hours unchanged |
| 2026-10-12 | minor | Cursor chat and spotlight in live sessions: `/` opens a chat bubble at your pointer that everyone sees as you type and that fades a few seconds after the last change (Enter sends, Escape takes it back, 100 characters, control characters dropped); your avatar's menu (or `collab.spotlight`) asks everyone to follow your view and page, latecomers included, until they move their own view. Both travel as signals: ephemeral presence the host validates, rate-limits (20 a second per person) and passes only to clients that announce they read them, so older clients keep working; never in the document or the log. Live cursors / follow / spotlight row partial → done, 8–15 h → 2–4 h; cursor chat row no → partial (no audio or reactions), 4–8 h → 2–5 h; multiplayer 45–90 h → 42–84 h (it counted these rows only in part); remaining 55–110 h (stale: team libraries had already gone to 4–8 h) → 46–92 h; collaboration ~13% unchanged at this rounding |
| 2026-10-12 | minor | A branch's own version history and finer list merges (gap 17): versions saved or autosaved on a branch tab go into the branch (viewed, restored, renamed, deleted and duplicated there, saved in the file with it); merging adds a version to the main file's history and the merged or archived branch keeps its own; fills, strokes and effects merge item by item, conflicting only when the same item changed both ways, an item was removed on one side or both sides added different items. Comments, versions, find, misc breadth 85% → 87%, ready 60% → 61%, 2–5 h → 1–4 h; gap 17 1–3 h → 1–2 h; feature breadth (~82%) and ready for real work (~51%) unchanged at this rounding; full parity ~306–595 h → ~305–594 h |

View File

@ -252,9 +252,11 @@ the address it came from and when it was last opened (for expiry).
[`packaging/server/figcraft-backup.timer`](../packaging/server/figcraft-backup.timer) (installed by
`install.sh`; `sudo systemctl enable --now figcraft-backup.timer`) does the following every night:
it stops the server for the seconds the copy takes, keeps
`/var/backups/figcraft-server/figcraft-YYYY-MM-DD.tar.gz` for 14 days and starts it again. Copy that
folder off the machine too.
it checks there is twice the data directory's size free, stops the server for the seconds the copy
takes, keeps `/var/backups/figcraft-server/figcraft-YYYY-MM-DD.tar.gz` (readable by root only) for
7 days and starts it again. Copy that folder off the machine too. The archives hold every file,
the addresses files came from, the reports and the **server key**: keep them, and any copy, as
private as the server itself.
Copy the data directory (the plugin directory, `plugins/`, included) while the server is stopped,
or take an atomic snapshot of the filesystem it lives on (ZFS, Btrfs, LVM, a VSS shadow copy). A plain copy of a running server's
@ -308,9 +310,11 @@ keeps the key for everything else.
- **Limits per address** (only when open): connections at once (`max_connections_per_ip`), new
connections a minute (`connects_per_minute`), files published an hour (`publishes_per_hour`),
messages a second (`messages_per_second`), and files and MiB kept (`max_files_per_ip`,
`max_mb_per_ip`); for everyone: `max_files`, `max_file_mb` per file and `max_total_mb` for
publishing without the key. An address is an IPv4 address or an IPv6 /64. Each refusal says why
("too many connections from your address…", "this server is full…") and the app shows it.
`max_mb_per_ip`, counting what files grow to by edits); for everyone: `max_files`, `max_file_mb`
per file (also as it grows), `max_total_mb` and twenty addresses' worth of publishing an hour. An
address is an IPv4 address or an IPv6 /64; an IPv6 /56 and /48 also count together, with 4 and 16
times one address's limits. Each refusal says why ("too many connections from your address…",
"this server is full…") and the app shows it. See also [Resource limits](#resource-limits).
Behind a reverse proxy set `trusted_proxy` to its address: the server then counts the client its
`X-Forwarded-For` names (the last address in it, the one the proxy added) and believes the header
from no one else. The tables of addresses keep at most 10,000 addresses (the least recently seen
@ -320,6 +324,8 @@ keeps the key for everything else.
looks every ten minutes and at startup, never deletes a file someone is in, and tells everyone in
a file the date it would go (the Share popover shows it). Files published with the key never
expire.
- **Team libraries** on a public server live only in files published with the key: a file published
without it can't hold one (a library is a whole archive the server unpacks and checks).
- **Moderation.** `figcraft-server files list --sort size|age|idle` shows who published what
(the address) and how many reports each file has; `files remove FILE` takes a file down (whoever
is in it is disconnected); `block-ip` refuses an address and drops its connections (kept across
@ -348,16 +354,40 @@ What someone can do, by what they hold (the server is reached through TLS at the
|---|---|---|
| Anyone who reaches the server, with nothing | Ask about its hosting (open, operator, terms, limits); on an open server publish files within the limits per address and for everyone; list and download the plugin directory; ask `/healthz` | Join, read or list any file; see links, reports or blocked addresses; use or learn the server key or anyone's token (none are ever sent back, and all are compared in constant time against hashes); publish plugins; spoof their address through `X-Forwarded-For` (believed only from `trusted_proxy`); get past a block |
| Someone with a file's **view link** | See that file and everyone's changes and presence in it; read its team library and usage; report it | Change it (the server refuses a viewer's edits); reach any other file; manage links |
| Someone with a file's **edit link** | Change that file, publish its team library; report it | Reach any other file; list, make or revoke links; rename or remove it |
| Someone with a file's **edit link** | Change that file, within its size cap; publish its team library (on a public server only for files published with the key); report it | Reach any other file; grow it past `max_file_mb` or its publisher's share; list, make or revoke links; rename or remove it |
| Whoever holds a file's **owner token** (published without the key) | List, make and revoke that file's links (disconnecting whoever joined with a revoked one), rename it on the server, remove it | Touch any other file, even with its id; list files; read reports; block; anything of the plugin directory |
| The **operator** (server key, or the machine) | Everything above for every file, reports, blocks, the plugin directory; read every file on disk | Recover a lost link or owner token (only hashes are kept) |
Limits that hold against everyone: message and file sizes, connections in all and per address,
connection, message and publish rates per address, files and MiB per address and for everyone, and
idle expiry. Not defended against: someone with many addresses (each gets its own limits; the
global caps still hold), a link or owner token that leaks (revoke links with the owner token or
`figcraft-server links revoke`; remove the file), and the operator reading files (no end-to-end
encryption).
Limits that hold against everyone: message sizes by what a connection is for, file sizes as they
grow, connections in all and per address and block, connection, message, publish and download rates
per address and block, files and MiB per address and for everyone, the bytes queued for slow
clients and waiting for the server, and idle expiry ([Resource limits](#resource-limits)). Revoking
a link or removing a file cuts every connection that used it, library connections included. Not
defended against: someone with many unrelated addresses (each gets its own limits; the global caps
still hold, so they can fill the server up to them until you block them or the files expire), a
link or owner token that leaks (revoke links with the owner token or `figcraft-server links
revoke`; remove the file), and the operator reading files (no end-to-end encryption).
### Resource limits
What keeps one client (or a few) from exhausting a small machine (the official one has 2 GB of
memory and 2 GB of swap; the defaults are chosen to hold on 1 GB). All of these apply only with
`open_publish = true` unless marked *always*; a private server keeps its earlier behaviour.
| Resource | Limit |
|---|---|
| Message size | A file's size and 1 MiB (`max_file_mb`, 16 by default when open, or `max_plugin_mb` if larger): checked as the message is read, before anything is decoded; 64 KiB on a connection that only browses plugins (a client sending more is disconnected with the limit as the reason; the app keeps its own copy). The core's 64 MiB otherwise |
| A file as it grows | Every batch is checked first: past `max_file_mb` (after compacting a long log), past its publisher's `max_mb_per_ip` or past `max_total_mb`, it is refused with the reason and the app rolls it back. A file's snapshot also holds the recent batches kept for reconnecting clients, at most a quarter of `max_file_mb` |
| Recent batches in memory | A quarter of `max_file_mb` (256 KiB at least) per open file; older ones are dropped (a client that missed them catches up with the whole file) |
| Bytes queued for a slow client | Twice the largest message per connection, 256 MiB for all of them: a client that doesn't read its answers is dropped instead of making the server hold more |
| Bytes waiting for the server | 128 MiB of read messages not handled yet: past it a connection is dropped as busy |
| Connections | `max_connections` (512) in all (*always*); `max_connections_per_ip` (32) per address, ×4 per IPv6 /56, ×16 per /48; one-shot requests (asking about hosting, libraries, browsing plugins) can't take the last eighth, which stays for people joining files, and are dropped after 20 idle seconds |
| Rates | Per address and block: `connects_per_minute`, `publishes_per_hour` (and twenty addresses' worth for everyone), `messages_per_second`; reports, ten an hour (*always*); downloads of libraries and plugins, 4 MiB a second with bursts of 64 MiB (*always*), within 32 MiB a second for libraries and the plugin directory's own budget for everyone |
| Team libraries | Only files published with the key hold one on a public server; an archive is at most `max_file_mb` and everything it unpacks to counts against the same cap (*always*), so a zip bomb is refused before it expands |
| Usage reports | Kept in memory, written at most once a minute per library and when the server stops; at most 1,000 components per file and 200 files per library (*always*) |
| Disk writes | A batch is confirmed only once it is on disk; when its log can't be written (a full disk), nobody hears it was accepted: the file closes, everyone in it reconnects and their apps send again what wasn't confirmed (*always*) |
| Storage | `max_files`, `max_total_mb` (10240), `max_files_per_ip` (20), `max_mb_per_ip` (256), and idle expiry (`idle_days`, 30) for files published without the key |
| The service | systemd caps it at 60% (slows down) and 75% (stopped and restarted) of the machine's memory, and only lets it talk to this machine (the proxy) |
The settings for a public server, as `install.sh` writes them:
@ -385,9 +415,13 @@ source it sits in, or else downloads the latest release for the machine (checked
`SHA256SUMS.txt`) or builds a fresh clone (`--from-source`); a build adds a temporary 2 GB swap
file on a machine with less than 2 GB of memory, so it isn't killed. It makes the `figcraft` user, the
data directory and `/etc/figcraft-server/server.toml` (only when missing: your edits are kept),
installs the systemd unit, the nightly backup timer and the Caddy site, opens SSH, 80 and 443 in
`ufw`, starts it all and prints what to do next. Running it again upgrades the binary and restarts
the server only when something changed.
installs the systemd units and its Caddy site (`/etc/caddy/figcraft-server.caddy`, imported from
the main Caddyfile, whose other sites it leaves alone), opens the SSH ports `sshd` listens on, 80
and 443 in `ufw`, and turns `ufw` on unless other services listen on the machine (then it says
so: allow their ports and enable it yourself, or pass `--firewall-anyway`). Builds run as the
unprivileged `figcraft-build` user; a fresh clone builds the latest release tag. It checks new
settings before they replace anything, starts it all and prints what to do next. Running it again
upgrades the binary and restarts the server only when something changed.
1. Create the server (Ubuntu 24.04, x86_64 or arm64) with your SSH key, and note its address.
2. At your DNS provider, point `figcraft.example.com` (A, and AAAA for IPv6) at that address. Caddy
@ -475,7 +509,9 @@ EOF
**3. Run it with systemd.** [`figcraft-server.service`](../packaging/server/figcraft-server.service)
runs it as that user, restarts it if it fails, stops it gracefully, and locks it down: no new
privileges, a read-only system with only the data directory writable, no capabilities, private
`/tmp` and devices, and a system-call filter.
`/tmp` and devices, a system-call filter, memory caps (60% of the machine slows it down, 75%
stops and restarts it, so only this service is hit) and network only to this machine (the proxy;
remove `IPAddressAllow`/`IPAddressDeny` to listen on a network address without one).
```sh
sudo cp figcraft-server.service /etc/systemd/system/
@ -580,11 +616,11 @@ sudo systemctl start figcraft-server
- **Limits**: the transport's (message sizes, 100 messages a second per connection, idle and
handshake timeouts, `max_connections`) and the collaboration core's (layers, changes per batch,
`max_people`: the cap per file), plus `max_files`, `max_file_mb` and `max_links`, and on a public
server the limits per address. Every message is decoded defensively, and fuzzed in the tests
(including open hosting, owner tokens and reports). A library is read in full before it is kept (as files
using it read it: size limits on every part, at most `max_file_mb`), must be the file's library
at a newer version, and is checked on the server's main thread; usage reports keep at most 2,000
components per file and 500 files per library.
server the limits per address ([Resource limits](#resource-limits)). Every message is decoded
defensively, and fuzzed in the tests (including open hosting, owner tokens and reports). A
library is read in full before it is kept (everything it unpacks to within `max_file_mb`), must
be the file's library at a newer version, and is checked on the server's main thread; usage
reports keep at most 1,000 components per file and 200 files per library.
- **Team libraries**: any link of a file reads its library and its usage reports (file names and
counts); only an edit link publishes it.
- **Plugin directory**: anyone who reaches the server reads it; only the server key writes to it.
@ -595,6 +631,8 @@ sudo systemctl start figcraft-server
it installs anything.
- **Not end-to-end encrypted**: the server reads every file it keeps (it validates and stores them);
so can its operator.
- **Not there yet**: user accounts, per-person permissions, audit logs, end-to-end encryption, and
checking a file's size as it grows by edits (the core's caps bound each edit; `max_file_mb` is
checked when a file is published and logged when it is compacted).
- **The machine**: the data directory is made readable by the server's account only and the server
key is written owner-only from the start; the systemd unit caps memory and network
(`MemoryHigh`, `MemoryMax`, `IPAddressAllow=localhost`); `install.sh` builds as an unprivileged
user and never cuts off SSH or other services when it turns `ufw` on.
- **Not there yet**: user accounts, per-person permissions, audit logs, end-to-end encryption.

View File

@ -1,9 +1,11 @@
# Nightly backup of figcraft-server's data directory (an example; docs/server.md, "Backups").
# install.sh installs it with figcraft-backup.timer; turn it on with
# sudo systemctl enable --now figcraft-backup.timer
# It stops the server for the few seconds the copy takes (apps reconnect on their own), writes
# /var/backups/figcraft-server/figcraft-YYYY-MM-DD.tar.gz and keeps the last 14 days. Copy that
# folder off the machine too (rsync, your provider's snapshots, object storage).
# It checks there is room for the copy (twice the data directory's size free), stops the server
# for the few seconds the copy takes (apps reconnect on their own), writes
# /var/backups/figcraft-server/figcraft-YYYY-MM-DD.tar.gz (readable by root only) and keeps the last
# 7 days. The archives hold every file, the addresses files came from, the reports and the server
# key: keep them, and any copy off the machine, as private as the server itself.
[Unit]
Description=Back up figcraft-server's data directory
@ -13,5 +15,6 @@ After=figcraft-server.service
Type=oneshot
Nice=10
IOSchedulingClass=idle
UMask=0077
# `$$` and `%%` are a literal `$` and `%` for the shell (systemd expands the single ones).
ExecStart=/bin/sh -ec 'dir=/var/backups/figcraft-server; install -d -m 700 "$$dir"; systemctl stop figcraft-server; trap "systemctl start figcraft-server" EXIT; tar -C /var/lib -czf "$$dir/figcraft-$$(date +%%F).tar.gz" figcraft-server; find "$$dir" -name "figcraft-*.tar.gz" -mtime +14 -delete'
ExecStart=/bin/sh -ec 'dir=/var/backups/figcraft-server; install -d -m 700 "$$dir"; need=$$(du -sk /var/lib/figcraft-server | cut -f1); free=$$(df -k --output=avail "$$dir" | tail -n 1); if [ "$$free" -lt $$((need * 2)) ]; then echo "not enough free space for a backup ($$free KiB free, $$need KiB of data)" >&2; exit 1; fi; systemctl stop figcraft-server; trap "systemctl start figcraft-server" EXIT; tar -C /var/lib -czf "$$dir/figcraft-$$(date +%%F).tar.gz" figcraft-server; find "$$dir" -name "figcraft-*.tar.gz" -mtime +7 -delete'

View File

@ -23,6 +23,14 @@ RestartSec=2
# max_connections (512 by default) plus the files kept open.
LimitNOFILE=4096
UMask=0077
# Under memory pressure this service slows down, then it alone is stopped (and restarted), not the
# rest of the machine.
MemoryHigh=60%
MemoryMax=75%
# It talks only to the proxy on this machine (listen = "127.0.0.1:47900"). Listening on a network
# address without a proxy in front needs these two lines removed.
IPAddressAllow=localhost
IPAddressDeny=any
# Hardening: the only writable place is the data directory.
NoNewPrivileges=yes

View File

@ -4,13 +4,13 @@
#
# sudo ./install.sh --host figcraft.example.com --email you@example.com
#
# It installs the binary (from this folder when run from a release archive; built from the source
# tree around this script when it sits in one, e.g. a copy of the repository; else the latest
# release for this machine checked against its SHA256SUMS.txt, else built from a fresh clone; builds
# add a temporary swap file when the machine has under 2 GB of memory), makes the `figcraft` user, its
# data directory and /etc/figcraft-server/server.toml (only when missing: your edits are kept),
# installs the systemd unit and the Caddy site, opens SSH, 80 and 443 in ufw and starts it all.
# Safe to run again: it upgrades the binary and restarts the server only when something changed.
# The binary comes from this folder when it is an unpacked release archive; else it is built from
# the source tree around this script (a copy of the repository); else it is the latest release for
# this machine (checked against SHA256SUMS.txt); else it is built from a fresh clone of the latest
# release tag. Builds run as an unprivileged user, with a temporary swap file under 2 GB of memory.
# It makes the `figcraft` user, its data directory and /etc/figcraft-server/server.toml (only when
# missing: your edits are kept), installs the systemd units and its own Caddy site, opens SSH, 80
# and 443 in ufw and starts it all. Safe to run again: it upgrades and restarts only on change.
#
# Options:
# --host HOST the DNS name people use (its A/AAAA records must point at this machine)
@ -20,9 +20,10 @@
# --version X.Y.Z install this release (default: the latest)
# --from-source build from source even when a release archive exists
# --ref REF the git branch or tag to clone and build (default: vX.Y.Z with --version,
# else main; a source tree around this script is built as it is)
# else the latest release tag; a source tree around this script is built as is)
# --backups also enable the nightly backup timer (stops the server for a few seconds)
# --no-firewall leave ufw alone
# --firewall-anyway enable ufw even though other services listen on this machine
# -h, --help this text
set -euo pipefail
@ -32,6 +33,7 @@ CONFIG="$CONFIG_DIR/server.toml"
DATA=/var/lib/figcraft-server
BIN=/usr/local/bin/figcraft-server
BUILD=/var/cache/figcraft-build
BUILDER=figcraft-build
SERVICE=figcraft-server
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# The source tree this script sits in (packaging/server/ of a copy or checkout), if any.
@ -49,8 +51,10 @@ FROM_SOURCE=false
REF=""
BACKUPS=false
FIREWALL=true
FIREWALL_ANYWAY=false
say() { printf '==> %s\n' "$*"; }
warn() { printf 'install.sh: warning: %s\n' "$*" >&2; }
die() { printf 'install.sh: %s\n' "$*" >&2; exit 1; }
while [ $# -gt 0 ]; do
@ -64,7 +68,8 @@ while [ $# -gt 0 ]; do
--ref) REF="${2:-}"; shift 2 ;;
--backups) BACKUPS=true; shift ;;
--no-firewall) FIREWALL=false; shift ;;
-h | --help) sed -n '2,26p' "$0"; exit 0 ;;
--firewall-anyway) FIREWALL_ANYWAY=true; shift ;;
-h | --help) sed -n '2,27p' "$0"; exit 0 ;;
*) die "unknown option $1 (see --help)" ;;
esac
done
@ -77,7 +82,7 @@ case " ${ID:-} ${ID_LIKE:-} " in
*) die "this script is for Debian and Ubuntu (see docs/server.md for other systems)" ;;
esac
ask() { # ask VAR "question": read an answer from the terminal when the value wasn't given
ask() { # ask VAR "question" FLAG: read an answer from the terminal when the value wasn't given
local var="$1" question="$2" answer
if [ -z "${!var}" ]; then
[ -t 0 ] || die "give $3 (no terminal to ask on)"
@ -115,7 +120,7 @@ esac
# --- The binary -----------------------------------------------------------------------------
NEW_BIN=""
FILES="$HERE" # where the unit and the Caddyfile come from
FILES="$HERE" # where the units and the Caddyfile come from
from_release() { # download and check the release archive; sets NEW_BIN and FILES
local version="$VERSION" name dir
@ -145,34 +150,46 @@ cleanup() {
}
trap cleanup EXIT
as_builder() { # run a command as the unprivileged build user, with its own toolchain
local jobs=()
if [ -n "${CARGO_BUILD_JOBS:-}" ]; then jobs=("CARGO_BUILD_JOBS=$CARGO_BUILD_JOBS"); fi
runuser -u "$BUILDER" -- env HOME="$BUILD" RUSTUP_HOME="$BUILD/rustup" CARGO_HOME="$BUILD/cargo" \
CARGO_TARGET_DIR="$BUILD/target" "${jobs[@]}" "$@"
}
from_source() { # build the source tree around this script, else a fresh clone; sets NEW_BIN and FILES
local mem_kb src ref="$REF"
if [ -z "$ref" ]; then
if [ -n "$VERSION" ]; then ref="v$VERSION"; else ref=main; fi
fi
apt-get install -y -qq build-essential git pkg-config >/dev/null
mkdir -p "$BUILD"
export RUSTUP_HOME="$BUILD/rustup" CARGO_HOME="$BUILD/cargo" CARGO_TARGET_DIR="$BUILD/target"
if [ ! -x "$CARGO_HOME/bin/cargo" ]; then
curl --proto '=https' --tlsv1.2 -fsSL https://sh.rustup.rs | sh -s -- -y --profile minimal --default-toolchain stable --no-modify-path >/dev/null
id -u "$BUILDER" >/dev/null 2>&1 || useradd --system --home-dir "$BUILD" --shell /usr/sbin/nologin "$BUILDER"
install -d -o "$BUILDER" -g "$BUILDER" -m 755 "$BUILD"
if [ ! -x "$BUILD/cargo/bin/cargo" ]; then
as_builder sh -c "curl --proto '=https' --tlsv1.2 -fsSL https://sh.rustup.rs | sh -s -- -y --profile minimal --default-toolchain stable --no-modify-path" >/dev/null
else
"$CARGO_HOME/bin/rustup" update stable >/dev/null
as_builder "$BUILD/cargo/bin/rustup" update stable >/dev/null
fi
if [ -n "$LOCAL_SRC" ]; then
src="$LOCAL_SRC"
say "Building the source in $src: this takes a while"
# A copy the build user owns (the original may sit in root's home).
src="$BUILD/src-local"
say "Building the source in $LOCAL_SRC: this takes a while"
rm -rf "$src"
install -d -o "$BUILDER" -g "$BUILDER" -m 755 "$src"
tar -C "$LOCAL_SRC" --exclude=./target --exclude=./plugins/target -cf - . | as_builder tar -C "$src" -xf -
else
src="$BUILD/src"
say "Building from a clone ($ref): this takes a while"
if [ -d "$src/.git" ]; then
git -C "$src" fetch --quiet --tags origin
as_builder git -C "$src" fetch --quiet --tags origin
else
rm -rf "$src"
git clone --quiet "https://github.com/$REPO" "$src" || die "can't clone $REPO (a private repository: copy the source to this machine and run its packaging/server/install.sh)"
as_builder git clone --quiet "https://github.com/$REPO" "$src" ||
die "can't clone $REPO (a private repository: copy the source to this machine and run its packaging/server/install.sh)"
fi
if [ -z "$ref" ]; then
if [ -n "$VERSION" ]; then ref="v$VERSION"; else ref="$(as_builder git -C "$src" tag --list 'v*' --sort=-v:refname | head -n 1)"; fi
[ -n "$ref" ] || ref=main
fi
say "Building $ref from a clone: this takes a while"
# A branch as it is upstream now, or a tag.
git -C "$src" checkout --quiet --force --detach "origin/$ref" 2>/dev/null ||
git -C "$src" checkout --quiet --force --detach "$ref"
as_builder git -C "$src" checkout --quiet --force --detach "origin/$ref" 2>/dev/null ||
as_builder git -C "$src" checkout --quiet --force --detach "$ref"
fi
# The compiler needs about 2 GB: add a temporary swap file on smaller machines.
mem_kb="$(awk '/^MemTotal:/ {print $2}' /proc/meminfo)"
@ -185,10 +202,10 @@ from_source() { # build the source tree around this script, else a fresh clone;
swapon "$SWAP"
export CARGO_BUILD_JOBS=1
fi
(cd "$src" && "$CARGO_HOME/bin/cargo" build --release --locked -p figcraft-server)
(cd "$src" && as_builder "$BUILD/cargo/bin/cargo" build --release --locked -p figcraft-server)
cleanup
SWAP=""
NEW_BIN="$CARGO_TARGET_DIR/release/figcraft-server"
NEW_BIN="$BUILD/target/release/figcraft-server"
FILES="$src/packaging/server"
}
@ -201,7 +218,7 @@ elif [ "$FROM_SOURCE" = true ] || ! from_release; then
from_source
fi
[ -x "$NEW_BIN" ] || die "no figcraft-server binary to install"
# The unit files and the Caddyfile beside this script when it has them (a checkout or an archive).
# The units and the Caddyfile beside this script when it has them (a checkout or an archive).
[ -f "$HERE/figcraft-backup.timer" ] && FILES="$HERE"
for f in figcraft-server.service Caddyfile figcraft-backup.service figcraft-backup.timer; do
[ -f "$FILES/$f" ] || die "$FILES/$f is missing"
@ -242,6 +259,8 @@ if [ ! -f "$CONFIG" ]; then
printf 'max_mb_per_ip = 256\n'
} >"$CONFIG.new"
chmod 644 "$CONFIG.new"
# Checked before it replaces anything.
"$BIN" --config "$CONFIG.new" check >/dev/null || { rm -f "$CONFIG.new"; die "the new settings don't read (see above)"; }
mv -f "$CONFIG.new" "$CONFIG"
CHANGED=true
fi
@ -272,40 +291,70 @@ if [ "$BACKUPS" = true ]; then
fi
# --- Caddy ----------------------------------------------------------------------------------
# Our site lives in a file of its own; the main Caddyfile only imports it. A Caddyfile with other
# sites is left as it is apart from that one import line.
MARK="# figcraft-server (install.sh)"
CADDYFILE=/etc/caddy/Caddyfile
OWN=true
# Anything but the packaged example site (or ours) is someone else's: leave it, add a file of ours.
if [ -f "$CADDYFILE" ] && ! grep -qF "$MARK" "$CADDYFILE" && grep -qvE '^\s*(#|$|:80|\{|\}|root|file_server)' "$CADDYFILE"; then
CADDYFILE=/etc/caddy/figcraft-server.caddy
OWN=false
fi
SITE_FILE=/etc/caddy/figcraft-server.caddy
SITE="$(mktemp)"
MAIN="$(mktemp)"
{
printf '%s: written by packaging/server/install.sh (run it again to update).\n' "$MARK"
# Global options belong at the top of the main Caddyfile only.
[ "$OWN" = false ] || printf '{\n\temail %s\n}\n\n' "$EMAIL"
sed -e '/^#/d' -e "s/figcraft\.example\.com/$HOST/" "$FILES/Caddyfile" | sed -e '/./,$!d'
} >"$SITE"
[ "$OWN" = true ] || say "Caddy already serves other sites: add 'import figcraft-server.caddy' to /etc/caddy/Caddyfile"
if install_file "$SITE" "$CADDYFILE" 644; then
caddy validate --config /etc/caddy/Caddyfile --adapter caddyfile >/dev/null 2>&1 || die "Caddy doesn't accept /etc/caddy/Caddyfile"
if [ ! -f "$CADDYFILE" ] || grep -qF "$MARK" "$CADDYFILE" || ! grep -qvE '^\s*(#|$|:80|\{|\}|root|file_server)' "$CADDYFILE"; then
# Missing, ours, or the packaged example site: ours, with the certificate email.
printf '%s\n{\n\temail %s\n}\n\nimport figcraft-server.caddy\n' "$MARK" "$EMAIL" >"$MAIN"
elif grep -qE '^\s*import\s+figcraft-server\.caddy' "$CADDYFILE"; then
cp "$CADDYFILE" "$MAIN"
else
say "Caddy already serves other sites: adding 'import figcraft-server.caddy' to $CADDYFILE"
{ cat "$CADDYFILE"; printf '\n%s\nimport figcraft-server.caddy\n' "$MARK"; } >"$MAIN"
fi
CADDY_CHANGED=false
for f in "$CADDYFILE" "$SITE_FILE"; do
if [ -f "$f" ]; then cp -p "$f" "$f.before-figcraft"; else rm -f "$f.before-figcraft"; fi
done
install_file "$SITE" "$SITE_FILE" 644 && CADDY_CHANGED=true
install_file "$MAIN" "$CADDYFILE" 644 && CADDY_CHANGED=true
rm -f "$SITE" "$MAIN"
if [ "$CADDY_CHANGED" = true ]; then
if ! caddy validate --config "$CADDYFILE" --adapter caddyfile >/dev/null 2>&1; then
for f in "$CADDYFILE" "$SITE_FILE"; do
if [ -f "$f.before-figcraft" ]; then cp -p "$f.before-figcraft" "$f"; else rm -f "$f"; fi
done
die "Caddy doesn't accept the new configuration (the old one is back; see: caddy validate --config $CADDYFILE)"
fi
systemctl reload caddy || systemctl restart caddy
fi
rm -f "$SITE"
systemctl enable --now --quiet caddy
# --- Firewall -------------------------------------------------------------------------------
if [ "$FIREWALL" = true ]; then
say "Firewall: SSH, 80 and 443"
ufw allow 22/tcp >/dev/null
# The port this SSH session came in on, if it isn't 22.
if [ -n "${SSH_CONNECTION:-}" ]; then
ufw allow "${SSH_CONNECTION##* }/tcp" >/dev/null
fi
# The ports sshd really listens on (sudo doesn't pass SSH_CONNECTION on).
SSH_PORTS="$(sshd -T 2>/dev/null | awk '/^port / {print $2}' | sort -u)"
[ -n "$SSH_PORTS" ] || SSH_PORTS=22
say "Firewall: SSH ($(echo "$SSH_PORTS" | tr '\n' ' ' | sed 's/ $//')), 80 and 443"
for p in $SSH_PORTS; do ufw allow "$p/tcp" >/dev/null; done
ufw allow 80/tcp >/dev/null
ufw allow 443/tcp >/dev/null
ufw --force enable >/dev/null
if ufw status | grep -q '^Status: active'; then
:
else
# Other services reachable from outside would be cut off by enabling ufw.
OTHERS=""
for a in $(ss -Hltn | awk '{print $4}'); do
case "$a" in 127.* | "[::1]:"* | ::1:*) continue ;; esac
port="${a##*:}"
case " $(echo "$SSH_PORTS" | tr '\n' ' ') 80 443 " in *" $port "*) continue ;; esac
OTHERS="$OTHERS $port"
done
if [ -n "$OTHERS" ] && [ "$FIREWALL_ANYWAY" = false ]; then
warn "other services listen on port(s)$OTHERS: ufw was not enabled (rules added; allow those ports, then 'ufw enable', or run again with --firewall-anyway)"
else
ufw --force enable >/dev/null
fi
fi
fi
# --- Done -----------------------------------------------------------------------------------
@ -330,7 +379,8 @@ figcraft-server is running behind Caddy.
$ADMIN block-ip ADDRESS (unblock-ip, blocked)
Server key: $ADMIN key
Settings: $CONFIG (then: systemctl restart $SERVICE)
Backups: systemctl enable --now figcraft-backup.timer (nightly, to /var/backups/figcraft-server)
Backups: systemctl enable --now figcraft-backup.timer (nightly, to /var/backups/figcraft-server;
the archives hold every file and the server key: keep them as private as the server)
Upgrade: run this script again
You are responsible for what people publish here: keep the abuse contact and terms current.