M9: close the remaining open-hosting DoS paths
Second security review: - a broadcast counts once against the queued-bytes budget; each block of addresses gets at most an eighth; when the budget runs out the biggest consumer is shed, not the receiver - forced compactions at most once a minute per file - plugin downloads check the per-address share before the global budget and before reading - install.sh never replaces a Caddyfile with other sites (one import line, once; install-test.sh in packaging lint); the built binary runs only as the server's account - bytes being read count against per-block and global budgets, the biggest readers dropped first - block multipliers x2 (/56) and x4 (/48), connection caps at most a sixteenth/eighth of all; global publish and download budgets eight times a /48's share - problems capped at 200; writes refused below max(1 GiB, two files) free; the private-server file cap documented
This commit is contained in:
bflatastic
BFlat
parent
7edc61b8d2
commit
e0435d4eb4
|
|
@ -35,6 +35,8 @@ jobs:
|
|||
"$RUNNER_TEMP/actionlint" -color
|
||||
- name: shellcheck
|
||||
run: shellcheck packaging/*.sh packaging/*/*.sh packaging/linux/AppRun
|
||||
- name: Server install script (Caddyfile handling)
|
||||
run: bash packaging/server/install-test.sh
|
||||
- name: AppImage AppRun desktop integration
|
||||
run: bash packaging/linux/apprun-test.sh
|
||||
- name: PowerShell
|
||||
|
|
|
|||
|
|
@ -1645,6 +1645,7 @@ dependencies = [
|
|||
"figcraft-format",
|
||||
"figcraft-plugin",
|
||||
"log",
|
||||
"rustix 1.1.5",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
|
|
|
|||
|
|
@ -33,6 +33,10 @@ env_logger = { version = "0.11", default-features = false, features = ["humantim
|
|||
# Stopping gracefully on SIGTERM/SIGINT/SIGHUP (systemd, Docker, Ctrl+C) and Ctrl+C/close on Windows.
|
||||
ctrlc = { version = "3", features = ["termination"] }
|
||||
|
||||
# Free space on the data directory's disk (statvfs, without unsafe code of our own).
|
||||
[target.'cfg(unix)'.dependencies]
|
||||
rustix = { version = "1", default-features = false, features = ["std", "fs"] }
|
||||
|
||||
[dev-dependencies]
|
||||
# Crafting hostile archives (zip bombs) in tests.
|
||||
zip = { workspace = true }
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
//! 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 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
|
||||
//! /56 and /48 with 2 and 4 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).
|
||||
|
|
@ -19,12 +19,14 @@ 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;
|
||||
/// Bytes one address may download a second (team libraries, plugins), and at once: an IPv6 /48
|
||||
/// gets four times that, an eighth of everyone's budgets (32 MiB a second, 512 MiB at once).
|
||||
pub const DOWNLOAD_RATE: f64 = (1u64 << 20) as f64;
|
||||
pub const DOWNLOAD_BURST: f64 = (16u64 << 20) as f64;
|
||||
|
||||
/// Everyone together may publish without the key this many times what one address may.
|
||||
pub const PUBLISHES_ALL: f64 = 20.0;
|
||||
/// Everyone together may publish without the key this many times what one address may: eight
|
||||
/// times what the widest block (an IPv6 /48, four addresses' worth) may, so no block takes it all.
|
||||
pub const PUBLISHES_ALL: f64 = 32.0;
|
||||
|
||||
/// The rates (see the server's settings).
|
||||
#[derive(Clone, Copy, Debug, PartialEq)]
|
||||
|
|
|
|||
|
|
@ -247,6 +247,9 @@ impl Config {
|
|||
max_plugin_bytes: mib(self.max_plugin_mb),
|
||||
max_total_bytes: mib(self.max_plugins_mb),
|
||||
},
|
||||
// At least 1 GiB, and room for two of the largest files (a compaction writes one beside
|
||||
// the old one).
|
||||
min_free_bytes: mib(self.file_mb().saturating_mul(2)).max(1 << 30),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -279,11 +282,17 @@ impl Config {
|
|||
return base;
|
||||
}
|
||||
let max = self.open_message().min(base.max_client_message);
|
||||
// Shared budgets: one block of addresses (an IPv4 address, an IPv6 /48) gets at most an
|
||||
// eighth of each, and its biggest consumers are dropped first.
|
||||
let (queued, reading) = (256usize << 20, 192usize << 20);
|
||||
NetLimits {
|
||||
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_queued_bytes: max.saturating_add(1 << 20),
|
||||
max_queued_block: queued / 8,
|
||||
max_queued_total: queued,
|
||||
max_reading_block: max.saturating_add(1 << 20).min(reading / 8),
|
||||
max_reading_total: reading,
|
||||
max_in_flight: 128 << 20,
|
||||
reserve_for_sessions: self.max_connections / 8,
|
||||
request_idle: std::time::Duration::from_secs(20),
|
||||
|
|
|
|||
|
|
@ -37,6 +37,9 @@ const MAX_QUERY: usize = 200;
|
|||
const MAX_LISTING: usize = 32 << 20;
|
||||
/// What fetches may send, in MiB per second on average (each costs at least one).
|
||||
const SERVE_MIB_PER_SEC: f64 = 32.0;
|
||||
/// …and at once: eight times what one block of addresses may download at once, so no block can
|
||||
/// take it all (see `abuse`).
|
||||
const SERVE_BURST_MIB: f64 = 512.0;
|
||||
const MIB: f64 = (1 << 20) as f64;
|
||||
|
||||
/// What the directory enforces.
|
||||
|
|
@ -80,7 +83,7 @@ impl Directory {
|
|||
pub fn open(root: &Path, cfg: DirectoryConfig) -> (Directory, Vec<String>) {
|
||||
let dir = root.join(PLUGINS);
|
||||
let burst = (2 * cfg.max_plugin_bytes) as f64 / MIB;
|
||||
let mut d = Directory { dir, cfg, entries: Vec::new(), served: Bucket::new(SERVE_MIB_PER_SEC, burst.max(64.0)), broken: false };
|
||||
let mut d = Directory { dir, cfg, entries: Vec::new(), served: Bucket::new(SERVE_MIB_PER_SEC, burst.max(SERVE_BURST_MIB)), broken: false };
|
||||
let mut problems = Vec::new();
|
||||
let index = d.dir.join(INDEX);
|
||||
let _ = fs::remove_file(disk::temp_of(&index));
|
||||
|
|
@ -173,6 +176,15 @@ impl Directory {
|
|||
out
|
||||
}
|
||||
|
||||
/// The size of version `version` (or the latest) of plugin `id`'s bundle, without reading it.
|
||||
pub fn size_of(&self, id: &str, version: Option<&str>) -> Option<u64> {
|
||||
self.entries
|
||||
.iter()
|
||||
.filter(|e| e.id == id && version.is_none_or(|v| compare_versions(&e.version, v) == Some(Ordering::Equal)))
|
||||
.min_by(|a, b| by_version(a, b))
|
||||
.map(|e| e.bytes)
|
||||
}
|
||||
|
||||
/// Version `version` (or the latest) of plugin `id`, and its bundle, read back and checked
|
||||
/// against its hash.
|
||||
pub fn fetch(&mut self, id: &str, version: Option<&str>) -> Result<(PluginEntry, Vec<u8>), String> {
|
||||
|
|
|
|||
|
|
@ -15,11 +15,11 @@ use crate::directory::{DirectoryConfig, INDEX, PLUGINS};
|
|||
use crate::store::{Store, StoreConfig};
|
||||
use crate::test_util::{Raw, TempDir, config, key, session};
|
||||
|
||||
fn wasm() -> Vec<u8> {
|
||||
pub(crate) fn wasm() -> Vec<u8> {
|
||||
wat::parse_str(r#"(module (memory (export "memory") 1) (func (export "figcraft_run") (param i32) (result i32) i32.const 0))"#).unwrap()
|
||||
}
|
||||
|
||||
fn manifest(id: &str, version: &str) -> String {
|
||||
pub(crate) fn manifest(id: &str, version: &str) -> String {
|
||||
json!({"id": id, "name": "Tiny tool", "version": version, "author": "Ada Example", "description": "Does very little.", "main": "tiny.wasm",
|
||||
"permissions": {"access": "write", "commands": ["shape.rectangle"]}})
|
||||
.to_string()
|
||||
|
|
@ -125,7 +125,7 @@ fn bad_bundles_caps_and_keys() {
|
|||
assert!(matches!(d.handle(true, PluginReq::Publish { bundle: bundle(ID, "1.0.0") }), PluginReply::Published(_)));
|
||||
assert!(matches!(d.handle(false, PluginReq::List { query: "x".repeat(100_000) }), PluginReply::List(_)));
|
||||
// Fetches share a budget: a flood is turned away, not queued.
|
||||
let busy = (0..400).filter_map(|_| d.fetch(ID, None).err()).filter(|e| e.contains("busy")).count();
|
||||
let busy = (0..800).filter_map(|_| d.fetch(ID, None).err()).filter(|e| e.contains("busy")).count();
|
||||
assert!(busy > 0, "a flood of fetches hits the budget");
|
||||
drop(store);
|
||||
// Caps: versions kept, bytes kept, off.
|
||||
|
|
|
|||
|
|
@ -32,6 +32,19 @@ const SNAP_HEADER: usize = 4 + 1 + 8 + 4;
|
|||
pub const LOG_HEADER: usize = 4 + 1 + 8;
|
||||
const RECORD_HEADER: usize = 8;
|
||||
|
||||
/// Free space on the disk holding `path`, in bytes, where it can be told (Unix).
|
||||
pub fn free_bytes(path: &Path) -> Option<u64> {
|
||||
#[cfg(unix)]
|
||||
{
|
||||
rustix::fs::statvfs(path).ok().map(|s| s.f_bavail.saturating_mul(s.f_frsize))
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
{
|
||||
let _ = path;
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// `path` with `.tmp` added: where its next version is written first.
|
||||
pub fn temp_of(path: &Path) -> PathBuf {
|
||||
let mut name = path.file_name().map(|n| n.to_os_string()).unwrap_or_default();
|
||||
|
|
|
|||
|
|
@ -147,7 +147,7 @@ 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");
|
||||
assert_eq!(cap, 4 << 20, "the largest message and a margin");
|
||||
let view = {
|
||||
let mut store = Store::open(dir.path(), cfg.store()).unwrap();
|
||||
let (file, _, view) = store.publish("Library", &snapshot("Library")).unwrap();
|
||||
|
|
@ -390,20 +390,20 @@ fn one_ipv6_block_gets_rising_but_bounded_limits() {
|
|||
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 {
|
||||
// Two /64s of one /56 fit (2 × the cap), the third doesn't.
|
||||
for n in 1..=2 {
|
||||
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 {
|
||||
assert!(matches!(via_proxy(addr, "2001:db8:1:3::1").1, ServerMsg::Refused { reason } if reason.contains("too many connections")));
|
||||
// Four of one /48 fit (4 × the cap) — two of them taken above — the fifth doesn't.
|
||||
for n in 1..=2u16 {
|
||||
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:1:300::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();
|
||||
|
|
@ -413,14 +413,14 @@ fn one_ipv6_block_gets_rising_but_bounded_limits() {
|
|||
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.
|
||||
// One connection a minute per /64, two per /56, four 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);
|
||||
assert_eq!(in_one_56, 2);
|
||||
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);
|
||||
assert_eq!(in_one_48, 2, "the /48's four, two of them taken above");
|
||||
// Publishing: one an hour per address, 32 for everyone together.
|
||||
let published = (0..60u32).filter(|n| a.allow(IpAddr::from((0xC000_0200 + n).to_be_bytes()), Action::Publish, t)).count();
|
||||
assert_eq!(published, 32);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -529,7 +529,134 @@ fn downloads_have_a_share_per_address() {
|
|||
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!((3..=5).contains(&served), "{served} downloads of 4 MiB in a burst of 16 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 { .. }));
|
||||
}
|
||||
|
||||
// Second review: N2, L7, N3, M3, N5, N7 ------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn a_file_at_its_cap_is_compacted_at_most_once_a_minute() {
|
||||
let dir = TempDir::new("forced");
|
||||
let mut c = config(dir.path());
|
||||
c.max_file_mb = Some(1);
|
||||
let mut store = Store::open(dir.path(), c.store()).unwrap();
|
||||
let (info, ..) = store.publish("Full", &snapshot("Full")).unwrap();
|
||||
let mut server = Backend::open(&mut store, &info.id).unwrap();
|
||||
let mut client = join(&mut server);
|
||||
let before = store.compactions();
|
||||
// Alternate a small accepted batch with one that would take the file over its cap.
|
||||
for i in 0..50 {
|
||||
rename(&mut server, &mut client, &format!("page {i}"));
|
||||
Backend::changed(&mut store, &info.id, &mut server);
|
||||
assert!(Backend::room(&mut store, &info.id, &server, 2 << 20).is_err());
|
||||
}
|
||||
assert!(store.compactions() - before <= 1, "{} compactions for 50 over-cap batches", store.compactions() - before);
|
||||
}
|
||||
|
||||
/// A plugin directory with one plugin whose bundle is about `mib` MiB.
|
||||
fn plugin_store(dir: &std::path::Path, mib: usize) -> Store {
|
||||
let mut store = Store::open(dir, config(dir).store()).unwrap();
|
||||
let bundle = figcraft_plugin::bundle::Bundle {
|
||||
manifest: crate::directory_tests::manifest("org.example.big", "1.0.0"),
|
||||
wasm: crate::directory_tests::wasm(),
|
||||
assets: vec![figcraft_plugin::bundle::Asset { name: "data.bin".into(), data: vec![9; mib << 20] }],
|
||||
};
|
||||
store.directory().publish(&bundle.encode()).unwrap();
|
||||
store
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn one_address_cant_spend_everyones_plugin_downloads() {
|
||||
let dir = TempDir::new("plugshare");
|
||||
let mut store = plugin_store(dir.path(), 4);
|
||||
let fetch = || PluginReq::Fetch { id: "org.example.big".into(), version: None };
|
||||
let greedy = ip("198.51.100.1");
|
||||
// Its share runs out after a few; then 200 more tries cost nobody else anything.
|
||||
let served =
|
||||
(0..10).take_while(|_| matches!(store.plugins(greedy, false, fetch()), figcraft_collab::message::PluginReply::Bundle { .. })).count();
|
||||
assert!((2..=5).contains(&served), "{served}");
|
||||
for _ in 0..200 {
|
||||
let r = store.plugins(greedy, false, fetch());
|
||||
assert!(matches!(&r, figcraft_collab::message::PluginReply::Failed { reason } if reason == BUSY_DOWNLOADS));
|
||||
}
|
||||
assert!(
|
||||
matches!(store.plugins(ip("198.51.100.2"), false, fetch()), figcraft_collab::message::PluginReply::Bundle { .. }),
|
||||
"others still download"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn half_read_messages_count_against_a_reading_budget() {
|
||||
let dir = TempDir::new("reading");
|
||||
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_reading_block: 64 << 10, max_reading_total: 1 << 20, ..cfg.net() }).unwrap();
|
||||
let mut raw = Raw::connect(busy.addr());
|
||||
raw.send(&hosting());
|
||||
assert!(matches!(raw.next(), ServerMsg::Hosting(_)), "small messages are read");
|
||||
let mut big = Document::new("Big");
|
||||
let page = big.first_page().unwrap();
|
||||
big.create(page, NodeKind::Rectangle, &"r".repeat(200_000), 0.0, 0.0, 1.0, 1.0).unwrap();
|
||||
let _ = raw.try_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 reading")));
|
||||
until(10, "the reading budget to be given back", || busy.connections() == 0);
|
||||
// Given back: another address reads again.
|
||||
let mut raw = Raw::connect(busy.addr());
|
||||
raw.send(&hosting());
|
||||
assert!(matches!(raw.next(), ServerMsg::Hosting(_)));
|
||||
busy.stop();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn one_ipv6_block_takes_at_most_an_eighth_of_the_connections() {
|
||||
let dir = TempDir::new("eighth");
|
||||
let cfg = Config { max_connections: 64, 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();
|
||||
for n in 1..=8u16 {
|
||||
let (ws, r) = via_proxy(addr, &format!("2001:db8:9:{:x}::1", n << 8));
|
||||
assert!(matches!(r, ServerMsg::Hosting(_)), "connection {n}");
|
||||
held.push(ws);
|
||||
}
|
||||
assert!(matches!(via_proxy(addr, "2001:db8:9:900::1").1, ServerMsg::Refused { reason } if reason.contains("too many connections")), "64 / 8");
|
||||
assert!(matches!(via_proxy(addr, "2001:db8:a::1").1, ServerMsg::Hosting(_)), "another /48 still gets in");
|
||||
drop(held);
|
||||
server.stop();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn no_block_can_use_up_a_shared_budget() {
|
||||
let mut a = Abuse::new(AbuseConfig { open: true, connects_per_minute: 30, publishes_per_hour: 10, messages_per_second: 200 });
|
||||
let t = Instant::now();
|
||||
// A /48 publishes at most an eighth of what everyone may; everyone else keeps the rest.
|
||||
let from_48 = (0..200u16).filter(|n| a.allow(IpAddr::from([0x2001, 0xdb8, 5, n << 8, 0, 0, 0, 0]), Action::Publish, t)).count();
|
||||
assert_eq!(from_48, 40, "four addresses' worth");
|
||||
let others = (0..400u32).filter(|n| a.allow(IpAddr::from((0xC633_6400 + n).to_be_bytes()), Action::Publish, t)).count();
|
||||
assert_eq!(others, 320 - 40, "the rest of everyone's 320 an hour");
|
||||
// Downloads: a /48's burst is at most an eighth of everyone's (checked when compiling, in `store`).
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn problems_stay_bounded_and_a_full_disk_refuses_writes() {
|
||||
let dir = TempDir::new("problems");
|
||||
let mut store = Store::open(dir.path(), config(dir.path()).store()).unwrap();
|
||||
let (info, ..) = store.publish("Broken", &snapshot("Broken")).unwrap();
|
||||
std::fs::write(dir.path().join("files").join(&info.id).join("snapshot.bin"), b"junk").unwrap();
|
||||
for _ in 0..500 {
|
||||
assert!(Backend::open(&mut store, &info.id).is_err());
|
||||
}
|
||||
assert!(store.problems().len() <= 200, "{} problems kept", store.problems().len());
|
||||
drop(store);
|
||||
// Where the free space can be told, below the threshold nothing more is written.
|
||||
if crate::disk::free_bytes(dir.path()).is_some() {
|
||||
let mut c = config(dir.path()).store();
|
||||
c.min_free_bytes = u64::MAX;
|
||||
let mut store = Store::open(dir.path(), c).unwrap();
|
||||
let e = store.publish("More", &snapshot("More")).unwrap_err();
|
||||
assert!(e.contains("disk is nearly full"), "{e}");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -62,6 +62,11 @@ const MAX_REPORTS: usize = 1_000;
|
|||
const MAX_REASON: usize = 500;
|
||||
/// How often idle files are looked for (seconds), besides at startup.
|
||||
const SWEEP_EVERY: u64 = 600;
|
||||
/// A file over its cap is compacted to make room at most this often (seconds): a compaction
|
||||
/// rewrites the whole file on the hub thread.
|
||||
const FORCED_COMPACTION_EVERY: u64 = 60;
|
||||
/// Most problems kept for the operator (the oldest go first).
|
||||
const MAX_PROBLEMS: usize = 200;
|
||||
|
||||
/// 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";
|
||||
|
|
@ -84,7 +89,9 @@ const MAX_USAGE_FILE: u64 = 4 << 20;
|
|||
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;
|
||||
const LIBRARY_BURST: f64 = (512u64 << 20) as f64;
|
||||
// A /48 (four addresses' worth) may download at most an eighth of everyone's budgets.
|
||||
const _: () = assert!(crate::abuse::DOWNLOAD_RATE * 4.0 <= LIBRARY_RATE / 8.0 && crate::abuse::DOWNLOAD_BURST * 4.0 <= LIBRARY_BURST / 8.0);
|
||||
|
||||
/// What the store enforces.
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
|
|
@ -97,6 +104,9 @@ pub struct StoreConfig {
|
|||
pub core: figcraft_collab::Limits,
|
||||
pub directory: DirectoryConfig,
|
||||
pub hosting: HostingConfig,
|
||||
/// Below this much free space on the data directory's disk, changes and new files are
|
||||
/// refused (with the reason) rather than half written.
|
||||
pub min_free_bytes: u64,
|
||||
}
|
||||
|
||||
/// Hosting for the public (see the server's settings).
|
||||
|
|
@ -206,6 +216,8 @@ struct Entry {
|
|||
log: Option<Log>,
|
||||
/// Its snapshot's and log's bytes on disk (what was written; pending records not yet).
|
||||
bytes: u64,
|
||||
/// When it was last compacted to make room (seconds since 1970).
|
||||
forced: u64,
|
||||
}
|
||||
|
||||
/// A library's usage reports in memory, and whether (and when) they were written.
|
||||
|
|
@ -235,6 +247,10 @@ pub struct Store {
|
|||
reports: Vec<ReportInfo>,
|
||||
/// When idle files are looked for next (seconds since 1970).
|
||||
next_sweep: u64,
|
||||
/// Compactions so far (tests count them).
|
||||
compactions: u64,
|
||||
/// The free space last seen on the data directory's disk, and when.
|
||||
free: Option<(Instant, Option<u64>)>,
|
||||
_lock: File,
|
||||
}
|
||||
|
||||
|
|
@ -326,6 +342,8 @@ impl Store {
|
|||
blocked: BTreeSet::new(),
|
||||
reports: Vec::new(),
|
||||
next_sweep: 0,
|
||||
compactions: 0,
|
||||
free: None,
|
||||
_lock: lock,
|
||||
};
|
||||
store.scan(&dir);
|
||||
|
|
@ -400,7 +418,7 @@ impl Store {
|
|||
log::info!("file at {}: its log predates the last compaction; it is ignored", path.display());
|
||||
}
|
||||
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 })
|
||||
Ok(Entry { meta, links, log: None, bytes, forced: 0 })
|
||||
}
|
||||
|
||||
/// The server key (for managing files).
|
||||
|
|
@ -517,6 +535,9 @@ impl Store {
|
|||
if self.files.len() >= self.cfg.max_files {
|
||||
return Err(format!("the server keeps at most {} files", self.cfg.max_files));
|
||||
}
|
||||
if let Some(why) = self.low_disk() {
|
||||
return Err(why);
|
||||
}
|
||||
let replica = Replica::from_snapshot(snapshot).map_err(|e| format!("the file couldn't be read: {e}"))?;
|
||||
if replica.document().node_count() > self.cfg.core.max_nodes {
|
||||
return Err(format!("the file has more than {} layers", self.cfg.core.max_nodes));
|
||||
|
|
@ -558,7 +579,7 @@ 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, bytes: (bytes.len() + LOG_HEADER) as u64 };
|
||||
let entry = Entry { meta, links: vec![edit.clone(), view.clone()], log: None, bytes: (bytes.len() + LOG_HEADER) as u64, forced: 0 };
|
||||
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 {
|
||||
|
|
@ -894,9 +915,20 @@ impl Store {
|
|||
/// 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> {
|
||||
if let Some(why) = self.low_disk() {
|
||||
return Err(why);
|
||||
}
|
||||
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()) {
|
||||
// A compaction to make room at most once a minute per file (it rewrites the whole file).
|
||||
let t = now();
|
||||
let may_compact = self.files.get(file).is_some_and(|e| {
|
||||
t >= e.forced.saturating_add(FORCED_COMPACTION_EVERY) && e.log.as_ref().is_some_and(|l| l.entries > 0 || !l.pending.is_empty())
|
||||
});
|
||||
if over(self) && may_compact {
|
||||
if let Some(e) = self.files.get_mut(file) {
|
||||
e.forced = t;
|
||||
}
|
||||
self.compact(file, server);
|
||||
}
|
||||
if over(self) {
|
||||
|
|
@ -947,7 +979,7 @@ impl Store {
|
|||
}
|
||||
for i in issues {
|
||||
log::warn!("{i}");
|
||||
self.problems.push(i);
|
||||
self.note(i);
|
||||
}
|
||||
server.journaling(true);
|
||||
let file_handle = OpenOptions::new().append(true).open(dir.join(LOG));
|
||||
|
|
@ -986,8 +1018,39 @@ impl Store {
|
|||
Ok(Log { generation, file: f, pending: Vec::new(), entries: 0, bytes: LOG_HEADER as u64, dirty: false })
|
||||
}
|
||||
|
||||
/// Compactions so far.
|
||||
pub fn compactions(&self) -> u64 {
|
||||
self.compactions
|
||||
}
|
||||
|
||||
/// Why nothing more may be written when the data directory's disk is nearly full (less free
|
||||
/// than [`StoreConfig::min_free_bytes`]); looked up at most once a second. Where the free space
|
||||
/// can't be told, nothing is refused.
|
||||
fn low_disk(&mut self) -> Option<String> {
|
||||
let now = Instant::now();
|
||||
let free = match self.free {
|
||||
Some((at, free)) if now.saturating_duration_since(at) < std::time::Duration::from_secs(1) => free,
|
||||
_ => {
|
||||
let free = disk::free_bytes(&self.root);
|
||||
self.free = Some((now, free));
|
||||
free
|
||||
}
|
||||
};
|
||||
free.filter(|f| *f < self.cfg.min_free_bytes)
|
||||
.map(|_| "the server's disk is nearly full: it takes no changes or new files for now (tell its operator)".to_string())
|
||||
}
|
||||
|
||||
/// Note a problem for the operator (logged too); only the latest [`MAX_PROBLEMS`] are kept.
|
||||
fn note(&mut self, problem: String) {
|
||||
if self.problems.len() >= MAX_PROBLEMS {
|
||||
self.problems.remove(0);
|
||||
}
|
||||
self.problems.push(problem);
|
||||
}
|
||||
|
||||
/// Compact an open file: its state as the next snapshot, an empty log.
|
||||
fn compact(&mut self, file: &str, server: &Server) {
|
||||
self.compactions += 1;
|
||||
let Some(generation) = self.files.get(file).and_then(|e| e.log.as_ref()).map(|l| l.generation) else { return };
|
||||
match self.rewrite(file, generation + 1, server) {
|
||||
Ok(log) => {
|
||||
|
|
@ -1070,7 +1133,7 @@ impl Backend for Store {
|
|||
self.load(file).map_err(|e| {
|
||||
let why = format!("file {file} can't be opened: {e}");
|
||||
log::error!("{why}");
|
||||
self.problems.push(why);
|
||||
self.note(why);
|
||||
"this file can't be opened on the server (its data is damaged); tell the server's owner".to_string()
|
||||
})
|
||||
}
|
||||
|
|
@ -1266,14 +1329,14 @@ impl Backend for Store {
|
|||
}
|
||||
|
||||
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())
|
||||
// One address's share first (before everyone's budget is spent or anything is read).
|
||||
if let PluginReq::Fetch { id, version } = &req
|
||||
&& let Some(size) = self.directory.size_of(id, version.as_deref())
|
||||
&& !self.abuse.allow(ip, Action::Download(usize::try_from(size).unwrap_or(usize::MAX)), Instant::now())
|
||||
{
|
||||
return PluginReply::Failed { reason: BUSY_DOWNLOADS.into() };
|
||||
}
|
||||
reply
|
||||
self.directory.handle(publisher, req)
|
||||
}
|
||||
|
||||
fn closing(&self) -> Closing {
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ use tungstenite::{Bytes, Message};
|
|||
|
||||
use crate::inbox::Inbox;
|
||||
use crate::join::{ENDED, RESTARTING};
|
||||
use crate::net::{Bucket, Duplex, NetLimits, config, shutdown};
|
||||
use crate::net::{Bucket, Duplex, Meter, NetLimits, config, shutdown};
|
||||
use crate::throttle::{PRESENCE_INTERVAL, Throttle, earliest};
|
||||
|
||||
/// How long a closing connection waits for the peer to hang up.
|
||||
|
|
@ -216,51 +216,125 @@ enum Out {
|
|||
Close(CloseCode, String),
|
||||
}
|
||||
|
||||
/// Bytes waiting in a connection's queue, counted against its cap and the server's until dropped.
|
||||
/// One message's bytes held for one or more connections (a broadcast shares one buffer): counted
|
||||
/// once against the server's budget, until the last copy is sent or dropped.
|
||||
struct Held {
|
||||
n: usize,
|
||||
all: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
impl Drop for Held {
|
||||
fn drop(&mut self) {
|
||||
self.all.fetch_sub(self.n, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// Bytes waiting in a connection's queue, counted against its cap and its block's until dropped
|
||||
/// (and, through [`Held`], once against the server's).
|
||||
struct Charge {
|
||||
n: usize,
|
||||
mine: Arc<AtomicUsize>,
|
||||
all: Arc<AtomicUsize>,
|
||||
block: Arc<AtomicUsize>,
|
||||
_held: Arc<Held>,
|
||||
}
|
||||
|
||||
impl Drop for Charge {
|
||||
fn drop(&mut self) {
|
||||
self.mine.fetch_sub(self.n, Ordering::Relaxed);
|
||||
self.all.fetch_sub(self.n, Ordering::Relaxed);
|
||||
self.block.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.
|
||||
/// Why a message couldn't be queued.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
enum Push {
|
||||
/// The connection's or its block's queue is full: it can't keep up.
|
||||
Full,
|
||||
/// The server's budget for queued bytes is used up.
|
||||
Everyone,
|
||||
/// The connection is gone.
|
||||
Gone,
|
||||
}
|
||||
|
||||
/// One connection's way out: its writer's queue, capped in messages and in bytes (its own, its
|
||||
/// block of addresses', and the server's for everyone), 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>,
|
||||
block: Arc<AtomicUsize>,
|
||||
all: Arc<AtomicUsize>,
|
||||
max: usize,
|
||||
max_block: 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<()>> {
|
||||
/// Queue `b`. One message always fits an empty queue (a welcome as large as the largest file
|
||||
/// gets out). `held` is the message's count against the server's budget, made by its first
|
||||
/// recipient and shared by the others; `over_budget` lets it past that budget (the hub just
|
||||
/// dropped the biggest consumer to make room).
|
||||
fn push(&self, b: &Bytes, held: &mut Option<Arc<Held>>, over_budget: bool) -> Result<(), Push> {
|
||||
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(()));
|
||||
let block = self.block.load(Ordering::Relaxed);
|
||||
if (mine > 0 && mine.saturating_add(n) > self.max) || (block > 0 && block.saturating_add(n) > self.max_block) {
|
||||
return Err(Push::Full);
|
||||
}
|
||||
let held = match held {
|
||||
Some(h) => h.clone(),
|
||||
None => {
|
||||
if !over_budget && self.all.load(Ordering::Relaxed).saturating_add(n) > self.max_all {
|
||||
return Err(Push::Everyone);
|
||||
}
|
||||
self.all.fetch_add(n, Ordering::Relaxed);
|
||||
held.insert(Arc::new(Held { n, all: self.all.clone() })).clone()
|
||||
}
|
||||
};
|
||||
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(()),
|
||||
self.block.fetch_add(n, Ordering::Relaxed);
|
||||
let charge = Charge { n, mine: self.queued.clone(), block: self.block.clone(), _held: held };
|
||||
self.tx.try_send(Out::Bytes(b.clone(), Some(charge))).map_err(|e| match e {
|
||||
TrySendError::Full(_) => Push::Full,
|
||||
TrySendError::Disconnected(_) => Push::Gone,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// What one block of addresses (an IPv4 address, an IPv6 /48: [`wide_block`]) has queued and is
|
||||
/// having read, shared by its connections.
|
||||
#[derive(Clone, Default)]
|
||||
struct BlockUse {
|
||||
queued: Arc<AtomicUsize>,
|
||||
reading: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
/// The blocks with open connections; a block leaves when its last connection ends.
|
||||
#[derive(Clone, Default)]
|
||||
struct Blocks(Arc<Mutex<HashMap<IpAddr, BlockUse>>>);
|
||||
|
||||
impl Blocks {
|
||||
fn get(&self, key: IpAddr) -> BlockUse {
|
||||
self.0.lock().unwrap_or_else(PoisonError::into_inner).entry(key).or_default().clone()
|
||||
}
|
||||
|
||||
/// A connection of block `key` ended (its `held` handles are dropped by now or soon).
|
||||
fn leave(&self, key: IpAddr, held: BlockUse) {
|
||||
let mut t = self.0.lock().unwrap_or_else(PoisonError::into_inner);
|
||||
// The table's copy and this one: nobody else uses the block.
|
||||
if Arc::strong_count(&held.queued) <= 2 && Arc::strong_count(&held.reading) <= 2 {
|
||||
t.remove(&key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The widest block an address counts in for the server's shared budgets: an IPv4 address, or an
|
||||
/// IPv6 address's /48.
|
||||
pub fn wide_block(key: IpAddr) -> IpAddr {
|
||||
ip_blocks(key).last().map_or(key, |(b, _)| b.0)
|
||||
}
|
||||
|
||||
/// 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)]
|
||||
|
|
@ -359,7 +433,13 @@ 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 shared = Shared {
|
||||
per_address: PerAddress::default(),
|
||||
queued: Arc::default(),
|
||||
in_flight: in_flight.clone(),
|
||||
blocks: Blocks::default(),
|
||||
reading: Arc::default(),
|
||||
};
|
||||
let queued = shared.queued.clone();
|
||||
let accept_thread = {
|
||||
let (hub_tx, stop, open) = (hub_tx.clone(), stop.clone(), open.clone());
|
||||
|
|
@ -467,15 +547,15 @@ impl Drop for Open {
|
|||
pub type Block = (IpAddr, u8);
|
||||
|
||||
/// 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.
|
||||
/// gets: the address itself (an IPv4 /32 or an IPv6 /64, ×1) and, for IPv6, its /56 (×2) and /48
|
||||
/// (×4), 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),
|
||||
((IpAddr::V6(Ipv6Addr::new(a, b, c, d & 0xff00, 0, 0, 0, 0)), 56), 2),
|
||||
((IpAddr::V6(Ipv6Addr::new(a, b, c, 0, 0, 0, 0, 0)), 48), 4),
|
||||
]
|
||||
}
|
||||
|
||||
|
|
@ -505,11 +585,18 @@ impl PerAddress {
|
|||
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>> {
|
||||
/// Count a connection from `ip` in each of its blocks, unless one of them is full. A wider
|
||||
/// block's cap is its multiple of `cap`, but at most a sixteenth (/56) or an eighth (/48) of
|
||||
/// `all` connections, so no block takes every connection.
|
||||
fn enter_all(&self, ip: IpAddr, cap: usize, all: 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))?);
|
||||
for ((addr, prefix), times) in ip_blocks(ip) {
|
||||
let block_cap = match prefix {
|
||||
56 => cap.saturating_mul(times as usize).min(all / 16).max(1),
|
||||
48 => cap.saturating_mul(times as usize).min(all / 8).max(1),
|
||||
_ => cap,
|
||||
};
|
||||
held.push(self.enter((addr, prefix), if cap == 0 { 0 } else { block_cap })?);
|
||||
}
|
||||
Some(held)
|
||||
}
|
||||
|
|
@ -588,6 +675,9 @@ struct Shared {
|
|||
per_address: PerAddress,
|
||||
queued: Arc<AtomicUsize>,
|
||||
in_flight: InFlight,
|
||||
blocks: Blocks,
|
||||
/// Bytes being read into messages, all connections together.
|
||||
reading: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
fn accept(listener: &TcpListener, hub: &Sender<HubIn>, stop: &AtomicBool, open: &Arc<AtomicUsize>, limits: NetLimits, shared: Shared) {
|
||||
|
|
@ -654,12 +744,16 @@ fn next_message(ws: &mut WebSocket<Duplex>, max: usize) -> Result<Option<(Client
|
|||
loop {
|
||||
return match ws.read() {
|
||||
Ok(Message::Binary(b)) => {
|
||||
// Whole now: it counts as in flight from here, no longer as being read.
|
||||
ws.get_mut().read_done();
|
||||
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(too_large(max)),
|
||||
// The reading budget's refusal (see `Meter`).
|
||||
Err(tungstenite::Error::Io(e)) if e.kind() == io::ErrorKind::Other => Err(e.to_string()),
|
||||
Err(_) => Ok(None),
|
||||
};
|
||||
}
|
||||
|
|
@ -707,9 +801,20 @@ 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 = shared.per_address.enter_all(ip, limits.max_per_address);
|
||||
let counted = shared.per_address.enter_all(ip, limits.max_per_address, limits.max_connections);
|
||||
let wide = wide_block(ip);
|
||||
let block = shared.blocks.get(wide);
|
||||
ws.get_mut().meter(Meter::new(block.reading.clone(), shared.reading.clone(), limits.max_reading_block, limits.max_reading_total));
|
||||
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 out = Line {
|
||||
tx,
|
||||
queued: Arc::default(),
|
||||
block: block.queued.clone(),
|
||||
all: shared.queued.clone(),
|
||||
max: limits.max_queued_bytes,
|
||||
max_block: limits.max_queued_block,
|
||||
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,
|
||||
|
|
@ -765,6 +870,7 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit
|
|||
}
|
||||
}
|
||||
drop(counted);
|
||||
shared.blocks.leave(wide, block);
|
||||
linger(&mut ws);
|
||||
drop(out);
|
||||
// The writer closes the socket once it has sent what's queued (a refusal, a close).
|
||||
|
|
@ -1274,10 +1380,22 @@ 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.push(b).is_ok());
|
||||
if !sent {
|
||||
shutdown(&c.sock);
|
||||
if !self.conns.contains_key(&conn) {
|
||||
return;
|
||||
}
|
||||
let mut drop = Vec::new();
|
||||
if !encode(msg).is_some_and(|b| deliver(&self.conns, conn, &b, &mut None, &mut drop)) {
|
||||
drop.push(conn);
|
||||
}
|
||||
self.drop_conns(drop);
|
||||
}
|
||||
|
||||
/// Cut connections that can't keep up (or were shed) and let them go.
|
||||
fn drop_conns(&mut self, conns: Vec<u64>) {
|
||||
for conn in conns {
|
||||
if let Some(c) = self.conns.get(&conn) {
|
||||
shutdown(&c.sock);
|
||||
}
|
||||
self.gone(conn);
|
||||
}
|
||||
}
|
||||
|
|
@ -1369,30 +1487,52 @@ impl<B: Backend> Hub<B> {
|
|||
Outgoing::Others(x, m) => (room.server.peers().map(|p| p.client_id).filter(|id| *id != x).collect(), m),
|
||||
};
|
||||
let mut bytes = None;
|
||||
// One count against the server's budget for every copy of this message.
|
||||
let mut held = None;
|
||||
for id in to {
|
||||
if let Some((local, inbox)) = room.local.as_ref().filter(|(l, _)| *l == id) {
|
||||
to_local(inbox, *local, &msg);
|
||||
continue;
|
||||
}
|
||||
let Some((&conn, c)) = room.ids.get(&id).and_then(|conn| self.conns.get_key_value(conn)) else { continue };
|
||||
let Some(&conn) = room.ids.get(&id) else { continue };
|
||||
let Some(b) = bytes.get_or_insert_with(|| encode(&msg)).clone() else { continue };
|
||||
match c.out.push(b) {
|
||||
Ok(()) => {}
|
||||
Err(TrySendError::Full(_)) => {
|
||||
log::info!("live session: client {id} can't keep up; disconnecting it");
|
||||
stuck.push(conn);
|
||||
}
|
||||
Err(TrySendError::Disconnected(_)) => stuck.push(conn),
|
||||
if !deliver(&self.conns, conn, &b, &mut held, &mut stuck) {
|
||||
stuck.push(conn);
|
||||
}
|
||||
}
|
||||
}
|
||||
for conn in stuck {
|
||||
if let Some(c) = self.conns.get(&conn) {
|
||||
shutdown(&c.sock);
|
||||
self.drop_conns(stuck);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Queue `b` for connection `conn`: false when it must be dropped (it can't keep up, or it is
|
||||
/// gone). When the server's budget for queued bytes is used up, the connection holding the most
|
||||
/// queued bytes is shed (added to `shed`) to make room, unless that is `conn` itself: whoever
|
||||
/// hogs the budget goes, not the next client that happens to need a little of it.
|
||||
fn deliver(conns: &HashMap<u64, Conn>, conn: u64, b: &Bytes, held: &mut Option<Arc<Held>>, shed: &mut Vec<u64>) -> bool {
|
||||
let Some(c) = conns.get(&conn) else { return true };
|
||||
match c.out.push(b, held, false) {
|
||||
Ok(()) => true,
|
||||
Err(Push::Everyone) => {
|
||||
let biggest = conns.iter().filter(|(k, _)| !shed.contains(k)).max_by_key(|(_, c)| c.out.queued.load(Ordering::Relaxed));
|
||||
match biggest {
|
||||
Some((&big, bc)) if big != conn && bc.out.queued.load(Ordering::Relaxed) > 0 => {
|
||||
log::info!("event=shed conn={big} ip={} reason=queued_bytes", bc.ip);
|
||||
shed.push(big);
|
||||
c.out.push(b, held, true).is_ok()
|
||||
}
|
||||
_ => {
|
||||
log::info!("live session: connection {conn} holds the most queued bytes; disconnecting it");
|
||||
false
|
||||
}
|
||||
self.gone(conn);
|
||||
}
|
||||
}
|
||||
Err(Push::Full) => {
|
||||
log::info!("live session: connection {conn} can't keep up; disconnecting it");
|
||||
false
|
||||
}
|
||||
Err(Push::Gone) => false,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1407,3 +1547,7 @@ pub(crate) fn to_local(inbox: &Inbox, local: u32, msg: &ServerMsg) {
|
|||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "queue_tests.rs"]
|
||||
mod queue_tests;
|
||||
|
|
|
|||
|
|
@ -34,9 +34,15 @@ pub struct NetLimits {
|
|||
/// 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.
|
||||
/// queue), for one block of addresses (an IPv4 address, an IPv6 /48), and for every client
|
||||
/// together (a broadcast counted once; past it the biggest consumer is dropped).
|
||||
pub max_queued_bytes: usize,
|
||||
pub max_queued_block: usize,
|
||||
pub max_queued_total: usize,
|
||||
/// Bytes being read into messages (not whole yet) for one block, and for every connection:
|
||||
/// past them the biggest readers are dropped.
|
||||
pub max_reading_block: usize,
|
||||
pub max_reading_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,
|
||||
|
|
@ -66,7 +72,10 @@ impl Default for NetLimits {
|
|||
max_per_address: 0,
|
||||
trusted_proxy: None,
|
||||
max_queued_bytes: 160 << 20,
|
||||
max_queued_block: usize::MAX,
|
||||
max_queued_total: usize::MAX,
|
||||
max_reading_block: usize::MAX,
|
||||
max_reading_total: usize::MAX,
|
||||
max_in_flight: usize::MAX,
|
||||
max_request: 64 << 10,
|
||||
reserve_for_sessions: 0,
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@
|
|||
|
||||
use std::io::{self, Read, Write};
|
||||
use std::net::{Shutdown, TcpStream};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex, PoisonError};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
|
|
@ -29,16 +30,72 @@ pub struct Duplex {
|
|||
read: TcpStream,
|
||||
write: Arc<Mutex<TcpStream>>,
|
||||
tls: Option<Tls>,
|
||||
/// What the bytes read count against (a server's reader), if anything.
|
||||
meter: Option<Meter>,
|
||||
}
|
||||
|
||||
impl Duplex {
|
||||
/// Two handles on `stream` (for the reader and the writer) and one to shut it down with.
|
||||
pub fn split(stream: TcpStream, tls: Option<Tls>) -> io::Result<(Duplex, Duplex, TcpStream)> {
|
||||
let write = Arc::new(Mutex::new(stream.try_clone()?));
|
||||
let a = Duplex { read: stream.try_clone()?, write: write.clone(), tls: tls.clone() };
|
||||
let b = Duplex { read: stream.try_clone()?, write, tls };
|
||||
let a = Duplex { read: stream.try_clone()?, write: write.clone(), tls: tls.clone(), meter: None };
|
||||
let b = Duplex { read: stream.try_clone()?, write, tls, meter: None };
|
||||
Ok((a, b, stream))
|
||||
}
|
||||
|
||||
/// Count what this handle reads against `meter`.
|
||||
pub fn meter(&mut self, meter: Meter) {
|
||||
self.meter = Some(meter);
|
||||
}
|
||||
|
||||
/// What was read so far became a message (handed on): it no longer counts as being read.
|
||||
pub fn read_done(&mut self) {
|
||||
if let Some(m) = self.meter.as_mut() {
|
||||
m.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Bytes being read into messages (not whole yet), per connection, per block of addresses and
|
||||
/// for every connection together: a read that takes a block over its share, or takes everyone
|
||||
/// over the budget while its block holds a large part of it, fails, so the biggest readers are
|
||||
/// dropped and memory for half-read messages stays bounded.
|
||||
pub struct Meter {
|
||||
mine: usize,
|
||||
block: Arc<AtomicUsize>,
|
||||
all: Arc<AtomicUsize>,
|
||||
max_block: usize,
|
||||
max_all: usize,
|
||||
}
|
||||
|
||||
impl Meter {
|
||||
pub fn new(block: Arc<AtomicUsize>, all: Arc<AtomicUsize>, max_block: usize, max_all: usize) -> Self {
|
||||
Self { mine: 0, block, all, max_block, max_all }
|
||||
}
|
||||
|
||||
fn add(&mut self, n: usize) -> io::Result<()> {
|
||||
self.mine += n;
|
||||
let block = self.block.fetch_add(n, Ordering::Relaxed).saturating_add(n);
|
||||
let all = self.all.fetch_add(n, Ordering::Relaxed).saturating_add(n);
|
||||
// Over its block's share; or over everyone's budget while its block holds a quarter of it
|
||||
// (the big readers go first); or far over everyone's budget.
|
||||
if block > self.max_block || (all > self.max_all && block.saturating_mul(4) >= self.max_all) || all > self.max_all.saturating_mul(2) {
|
||||
return Err(io::Error::other("the server is busy reading from others: try again in a moment"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn release(&mut self) {
|
||||
self.block.fetch_sub(self.mine, Ordering::Relaxed);
|
||||
self.all.fetch_sub(self.mine, Ordering::Relaxed);
|
||||
self.mine = 0;
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Meter {
|
||||
fn drop(&mut self) {
|
||||
self.release();
|
||||
}
|
||||
}
|
||||
|
||||
impl Duplex {
|
||||
|
|
@ -51,10 +108,14 @@ impl Duplex {
|
|||
|
||||
impl Read for Duplex {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
match &self.tls {
|
||||
let n = match &self.tls {
|
||||
Some(t) => tls::read(t, &mut self.read, &self.write, buf),
|
||||
None => self.read.read(buf),
|
||||
}?;
|
||||
if let Some(m) = self.meter.as_mut() {
|
||||
m.add(n)?;
|
||||
}
|
||||
Ok(n)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,83 @@
|
|||
//! The hub's send queues (no writers: nothing is ever sent, so what is queued stays queued): a
|
||||
//! broadcast counts once against the server's budget however many copies are queued, a block of
|
||||
//! addresses can't take more than its share, and when the server's budget is used up the
|
||||
//! connection holding the most is shed, not the next client that needs a little.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::net::{IpAddr, TcpListener, TcpStream};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::mpsc::{Receiver, sync_channel};
|
||||
|
||||
use tungstenite::Bytes;
|
||||
|
||||
use super::{Conn, Kind, Line, Out, Push, deliver};
|
||||
|
||||
/// A line with its own queue counter, the given block counter and the server's.
|
||||
fn line(block: &Arc<AtomicUsize>, all: &Arc<AtomicUsize>, max_block: usize, max_all: usize) -> (Line, Receiver<Out>) {
|
||||
let (tx, rx) = sync_channel(64);
|
||||
(Line { tx, queued: Arc::default(), block: block.clone(), all: all.clone(), max: 1 << 20, max_block, max_all }, rx)
|
||||
}
|
||||
|
||||
fn socket() -> TcpStream {
|
||||
let l = TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
TcpStream::connect(l.local_addr().unwrap()).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_broadcast_counts_once_against_the_server() {
|
||||
let all = Arc::new(AtomicUsize::new(0));
|
||||
let lines: Vec<_> = (0..4).map(|_| line(&Arc::default(), &all, usize::MAX, 10_000)).collect();
|
||||
let b = Bytes::from(vec![0u8; 3_000]);
|
||||
let mut held = None;
|
||||
for (l, _) in &lines {
|
||||
l.push(&b, &mut held, false).unwrap();
|
||||
}
|
||||
assert_eq!(all.load(Ordering::Relaxed), 3_000, "four copies of one message, counted once");
|
||||
assert!(lines.iter().all(|(l, _)| l.queued.load(Ordering::Relaxed) == 3_000), "each queue counts its copy");
|
||||
drop(held);
|
||||
// Sent (or dropped): the counts go back as the copies go.
|
||||
for (_, rx) in &lines {
|
||||
drop(rx.try_recv().unwrap());
|
||||
}
|
||||
assert_eq!(all.load(Ordering::Relaxed), 0);
|
||||
assert!(lines.iter().all(|(l, _)| l.queued.load(Ordering::Relaxed) == 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn one_block_gets_only_its_share() {
|
||||
let (block, all) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
|
||||
let (a, _ra) = line(&block, &all, 1_000, usize::MAX);
|
||||
let (b, _rb) = line(&block, &all, 1_000, usize::MAX);
|
||||
let (c, _rc) = line(&Arc::default(), &all, 1_000, usize::MAX);
|
||||
let msg = Bytes::from(vec![0u8; 800]);
|
||||
a.push(&msg, &mut None, false).unwrap();
|
||||
assert_eq!(b.push(&msg, &mut None, false), Err(Push::Full), "the block's second connection is over its share");
|
||||
c.push(&msg, &mut None, false).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_biggest_consumer_is_shed_when_the_budget_runs_out() {
|
||||
let all = Arc::new(AtomicUsize::new(0));
|
||||
let ip: IpAddr = "203.0.113.1".parse().unwrap();
|
||||
let mut conns = HashMap::new();
|
||||
let mut receivers = Vec::new();
|
||||
for id in 1..=3u64 {
|
||||
let (l, rx) = line(&Arc::default(), &all, usize::MAX, 10_000);
|
||||
receivers.push(rx);
|
||||
conns.insert(id, Conn { out: l, sock: socket(), kind: Kind::Managing, ip });
|
||||
}
|
||||
// Connection 1 doesn't read: it holds most of the budget.
|
||||
let big = Bytes::from(vec![0u8; 9_000]);
|
||||
assert!(deliver(&conns, 1, &big, &mut None, &mut Vec::new()));
|
||||
// Connection 2 needs a little: the hog is shed and 2 still gets its message.
|
||||
let small = Bytes::from(vec![0u8; 2_000]);
|
||||
let mut shed = Vec::new();
|
||||
assert!(deliver(&conns, 2, &small, &mut None, &mut shed));
|
||||
assert_eq!(shed, vec![1]);
|
||||
assert_eq!(conns[&2].out.queued.load(Ordering::Relaxed), 2_000);
|
||||
// When the receiver is itself the biggest, it is the one to go.
|
||||
let mut shed = Vec::new();
|
||||
assert!(!deliver(&conns, 1, &small, &mut None, &mut shed), "the hog asking for more is dropped");
|
||||
assert!(shed.is_empty());
|
||||
}
|
||||
|
|
@ -311,9 +311,10 @@ keeps the key for everything else.
|
|||
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`, 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…",
|
||||
per file (also as it grows), `max_total_mb` and 32 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 2 and 4
|
||||
times one address's limits, and a /48 never more than an eighth of any budget everyone shares
|
||||
(connections, queued bytes, bytes being read, publishing, downloads). 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
|
||||
|
|
@ -371,22 +372,31 @@ revoke`; remove the file), and the operator reading files (no end-to-end encrypt
|
|||
### 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.
|
||||
memory and 2 GB of swap; the defaults are chosen to hold on 1 GB). The rule throughout: no single
|
||||
address or block of addresses can use up a budget everyone shares (each gets at most an eighth),
|
||||
and when a shared budget runs out the biggest consumer is dropped, not the next client that happens
|
||||
to need a little of it. These apply only with `open_publish = true` unless marked *always*.
|
||||
|
||||
**On private servers too** (a change from earlier versions): a batch that would take a file past
|
||||
`max_file_mb` (64 MiB by default there) is now refused with the reason, and the app rolls it back,
|
||||
where it used to be kept and only logged; and nothing is written while the disk has less than
|
||||
1 GiB (or two of the largest files) free. Raise `max_file_mb` (at most 4096) if your team's files
|
||||
are larger.
|
||||
|
||||
| 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` |
|
||||
| A file as it grows | Every batch is checked first (*always*): past `max_file_mb`, 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 long log is compacted to make room at most once a minute per file (a compaction rewrites the whole file). 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 queued for slow clients | The largest message and 1 MiB per connection, 32 MiB per block of addresses, 256 MiB for everyone (a message sent to many people counts once): a client that doesn't read its answers is dropped; when everyone's budget runs out, the connection holding the most is dropped |
|
||||
| Bytes being read | Messages not yet whole: the largest message and 1 MiB per block, 192 MiB for everyone; past them the biggest readers are dropped as busy |
|
||||
| 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 |
|
||||
| Connections | `max_connections` (512) in all (*always*); `max_connections_per_ip` (32) per address, ×2 per IPv6 /56 (at most a sixteenth of all), ×4 per /48 (at most an eighth of all), joins included, so no block can take every connection; 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 32 addresses' worth for everyone), `messages_per_second`; reports, ten an hour (*always*); downloads of libraries and plugins, 1 MiB a second with bursts of 16 MiB per address (*always*), checked before anything is read, within 32 MiB a second (512 MiB at once) for libraries and for plugins, everyone together |
|
||||
| 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 |
|
||||
| Disk writes | A batch is confirmed only once it is on disk; when its log can't be written, nobody hears it was accepted: the file closes, everyone in it reconnects and their apps send again what wasn't confirmed. Below 1 GiB (or two of the largest files) free, changes and new files are refused with "the server's disk is nearly full" instead (*always*, where the free space can be told: Unix). The server keeps the last 200 problems for the operator |
|
||||
| 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. Size `max_total_mb` to the disk: the files, a compaction's copy of the largest, the nightly backups (each about the data's size, seven kept unless you move them off) and 1 GiB to spare |
|
||||
| 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:
|
||||
|
|
@ -419,8 +429,10 @@ installs the systemd units and its Caddy site (`/etc/caddy/figcraft-server.caddy
|
|||
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
|
||||
unprivileged `figcraft-build` user, and the binary is only ever run as `figcraft` (never as root);
|
||||
a fresh clone builds the latest release tag. A Caddyfile with other sites is never replaced: it
|
||||
gains one `import figcraft-server.caddy` line, once (`packaging/server/install-test.sh` checks
|
||||
this). 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.
|
||||
|
|
@ -634,5 +646,6 @@ sudo systemctl start figcraft-server
|
|||
- **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.
|
||||
user, runs the binary only as the server's account, never replaces someone else's Caddyfile 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.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,43 @@
|
|||
#!/usr/bin/env bash
|
||||
# Checks install.sh's Caddyfile handling without installing anything (no root, no network):
|
||||
# a Caddyfile with other sites is never replaced, only gains one import line, and stays the same
|
||||
# however often install.sh runs. Run by the packaging lint workflow.
|
||||
#
|
||||
# Usage: packaging/server/install-test.sh
|
||||
set -euo pipefail
|
||||
HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
WORK="$(mktemp -d)"
|
||||
trap 'rm -rf "$WORK"' EXIT
|
||||
FAILED=0
|
||||
|
||||
main() { bash "$HERE/install.sh" --caddy-main "$1" ops@example.com; }
|
||||
fail() { printf 'FAIL: %s\n' "$*" >&2; FAILED=1; }
|
||||
|
||||
# Missing: ours.
|
||||
main "$WORK/none" >"$WORK/out"
|
||||
grep -qx 'import figcraft-server.caddy' "$WORK/out" || fail "a new Caddyfile imports our site"
|
||||
grep -q 'email ops@example.com' "$WORK/out" || fail "a new Caddyfile has the certificate email"
|
||||
|
||||
# The example site the package ships: replaced by ours.
|
||||
printf '# The Caddyfile is an easy way to configure your Caddy web server.\n:80 {\n\troot * /usr/share/caddy\n\tfile_server\n}\n' >"$WORK/default"
|
||||
main "$WORK/default" >"$WORK/out"
|
||||
grep -q 'usr/share/caddy' "$WORK/out" && fail "the packaged example site is replaced"
|
||||
|
||||
# Someone else's sites: kept, one import added, and the next run changes nothing.
|
||||
printf 'shop.example.com {\n\treverse_proxy 127.0.0.1:8080\n}\n' >"$WORK/theirs"
|
||||
cp "$WORK/theirs" "$WORK/run1"
|
||||
main "$WORK/run1" >"$WORK/run2"
|
||||
grep -q 'shop.example.com' "$WORK/run2" || fail "the operator's site is kept"
|
||||
[ "$(grep -c 'import figcraft-server.caddy' "$WORK/run2")" = 1 ] || fail "one import line is added"
|
||||
main "$WORK/run2" >"$WORK/run3"
|
||||
cmp -s "$WORK/run2" "$WORK/run3" || fail "running again leaves a Caddyfile with other sites as it is"
|
||||
main "$WORK/run3" >"$WORK/run4"
|
||||
cmp -s "$WORK/run3" "$WORK/run4" || fail "and again"
|
||||
|
||||
# Ours, run again: unchanged (it imports our site).
|
||||
main "$WORK/none" >"$WORK/ours1"
|
||||
main "$WORK/ours1" >"$WORK/ours2"
|
||||
cmp -s "$WORK/ours1" "$WORK/ours2" || fail "our own Caddyfile stays the same"
|
||||
|
||||
if [ "$FAILED" = 0 ]; then echo "install.sh Caddyfile handling: ok"; fi
|
||||
exit "$FAILED"
|
||||
|
|
@ -57,8 +57,26 @@ say() { printf '==> %s\n' "$*"; }
|
|||
warn() { printf 'install.sh: warning: %s\n' "$*" >&2; }
|
||||
die() { printf 'install.sh: %s\n' "$*" >&2; exit 1; }
|
||||
|
||||
MARK="# figcraft-server (install.sh)"
|
||||
|
||||
caddy_main() { # caddy_main FILE EMAIL: the main Caddyfile to install, on stdout
|
||||
local file="$1" email="$2"
|
||||
if [ -f "$file" ] && grep -qE '^[[:space:]]*import[[:space:]]+figcraft-server\.caddy[[:space:]]*$' "$file"; then
|
||||
cat "$file" # it imports our site already: kept exactly as it is
|
||||
elif [ ! -f "$file" ] || ! grep -qvE '^[[:space:]]*(#|$|:80|\{|\}|root|file_server)' "$file"; then
|
||||
# Missing, or only the example site the package ships: ours, with the certificate email.
|
||||
printf '%s\n{\n\temail %s\n}\n\nimport figcraft-server.caddy\n' "$MARK" "$email"
|
||||
else
|
||||
# Someone else's sites: kept, with one line added that imports ours.
|
||||
cat "$file"
|
||||
printf '\nimport figcraft-server.caddy\n'
|
||||
fi
|
||||
}
|
||||
|
||||
while [ $# -gt 0 ]; do
|
||||
case "$1" in
|
||||
# For tests (packaging/server/install-test.sh): print the Caddyfile this script would write.
|
||||
--caddy-main) caddy_main "${2:-}" "${3:-}"; exit 0 ;;
|
||||
--host) HOST="${2:-}"; shift 2 ;;
|
||||
--email) EMAIL="${2:-}"; shift 2 ;;
|
||||
--terms) TERMS="${2:-}"; shift 2 ;;
|
||||
|
|
@ -224,16 +242,19 @@ for f in figcraft-server.service Caddyfile figcraft-backup.service figcraft-back
|
|||
[ -f "$FILES/$f" ] || die "$FILES/$f is missing"
|
||||
done
|
||||
|
||||
# The server's own account runs the binary, from here on too (root never runs what was built).
|
||||
id -u figcraft >/dev/null 2>&1 || useradd --system --home-dir "$DATA" --shell /usr/sbin/nologin figcraft
|
||||
as_server() { runuser -u figcraft -- "$@"; }
|
||||
|
||||
CHANGED=false
|
||||
if ! cmp -s "$NEW_BIN" "$BIN"; then
|
||||
say "Installing $("$NEW_BIN" --version)"
|
||||
install -m 755 "$NEW_BIN" "$BIN.new"
|
||||
mv -f "$BIN.new" "$BIN"
|
||||
say "Installed $(as_server "$BIN" --version)"
|
||||
CHANGED=true
|
||||
fi
|
||||
|
||||
# --- User, data and settings ----------------------------------------------------------------
|
||||
id -u figcraft >/dev/null 2>&1 || useradd --system --home-dir "$DATA" --shell /usr/sbin/nologin figcraft
|
||||
# --- Data and settings ----------------------------------------------------------------------
|
||||
install -d -o figcraft -g figcraft -m 700 "$DATA"
|
||||
install -d -m 755 "$CONFIG_DIR"
|
||||
if [ ! -f "$CONFIG" ]; then
|
||||
|
|
@ -260,11 +281,11 @@ if [ ! -f "$CONFIG" ]; then
|
|||
} >"$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)"; }
|
||||
as_server "$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
|
||||
"$BIN" --config "$CONFIG" check >/dev/null ||
|
||||
as_server "$BIN" --config "$CONFIG" check >/dev/null ||
|
||||
die "$CONFIG isn't right for this figcraft-server (see above; a release older than open hosting needs --from-source)"
|
||||
|
||||
install_file() { # install_file SOURCE TARGET MODE: true when it changed
|
||||
|
|
@ -291,9 +312,8 @@ 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)"
|
||||
# Our site lives in a file of its own; the main Caddyfile only imports it (caddy_main). A
|
||||
# Caddyfile with other sites is never replaced: it only gains the import line, once.
|
||||
CADDYFILE=/etc/caddy/Caddyfile
|
||||
SITE_FILE=/etc/caddy/figcraft-server.caddy
|
||||
SITE="$(mktemp)"
|
||||
|
|
@ -302,15 +322,7 @@ MAIN="$(mktemp)"
|
|||
printf '%s: written by packaging/server/install.sh (run it again to update).\n' "$MARK"
|
||||
sed -e '/^#/d' -e "s/figcraft\.example\.com/$HOST/" "$FILES/Caddyfile" | sed -e '/./,$!d'
|
||||
} >"$SITE"
|
||||
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_main "$CADDYFILE" "$EMAIL" >"$MAIN"
|
||||
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
|
||||
|
|
|
|||
Loading…
Reference in New Issue