swarmctl and the authelia bridge serialise users.yml writes with a shared flock

Cargo.lock

@@ -4779,6 +4779,7 @@ dependencies = [
  "axum",
  "hive-types",
  "reqwest",
+ "rustix",
  "serde",
  "serde_json",
  "serde_norway",

swarm-authelia-bridge/Cargo.toml

@@ -13,6 +13,8 @@ hive-types.workspace = true
 anyhow.workspace = true
 axum.workspace = true
 reqwest.workspace = true
+# `flock` on `users.yml`'s lock file, shared with `swarmctl`.
+rustix.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_norway = "0.9.42"

swarm-authelia-bridge/README.md

@@ -51,8 +51,14 @@ which is exactly what a hive with existing users hit.
 
 What still has to hold, since two processes share the file:
 
-- **Read before write.** Both do, so neither can drop a user the other
+- **One lock around read → write.** Both take an exclusive `flock` on
-  added between operations.
+  `users.yml.lock`, a sibling file in the same directory, before loading and
+  hold it until the rename lands, waiting at most 10 seconds. Reading the
+  file first is not enough: both writers spend about a second minting a
+  password between load and write, and the second rename would drop the
+  first writer's user. The lock file is opened `O_NOFOLLOW`, so neither side
+  can be pointed at another file through it. `swarmctl` is root and hands
+  the file it creates to the directory's owner, so the bridge can open it.
 - **Unknown keys survive.** Both carry unmodelled top-level and per-user
   fields through a round-trip, or whichever writes second would silently
   delete what the first added.

swarm-authelia-bridge/src/main.rs

@@ -98,7 +98,8 @@ struct AppState {
     /// whichever `publish`es second silently drops the first agent's entry —
     /// the store is a plain file, not a database with its own concurrency
     /// control. `tokio::sync::Mutex`, not `std`'s: held across the `.await`s
-    /// in `generate_password` and `publish`.
+    /// in `generate_password` and `publish`. It covers this process only;
+    /// `store::lock_users_file` is the lock against `swarmctl`.
     write_lock: tokio::sync::Mutex<()>,
 }
 
@@ -316,6 +317,10 @@ async fn handle(
     // doc comment on `AppState::write_lock` for why this is reachable in
     // practice, not just theoretically.
     let _write_guard = write_lock.lock().await;
+    // The in-process lock does not cover `swarmctl`, which rewrites the same
+    // file from the host: the `flock` does, and is held to the end of the
+    // function like the guard above.
+    let _file_lock = store::lock_users_file(&cfg.users_file, store::USERS_LOCK_WAIT).await?;
 
     let mut user_store = store::load_store(&cfg.users_file)?;
     if user_store.users.contains_key(&name) {
@@ -443,7 +448,7 @@ async fn generate_password(bin: &std::path::Path) -> Result<String> {
 #[cfg(test)]
 mod tests {
     use super::{
-        BridgeRequest, Config, Refusal, StatusCode, classify, introspect::Verdict, respond,
+        BridgeRequest, Config, Refusal, StatusCode, classify, handle, introspect::Verdict, respond,
     };
 
     const CONTROLLER: &str = "swarm-controller";
@@ -586,4 +591,59 @@ mod tests {
         );
         let _ = std::fs::remove_dir_all(config.users_file.parent().unwrap());
     }
+
+    /// `handle` waits for the `flock` `swarmctl` takes, not only for the
+    /// in-process lock: it is held here while `handle` runs with a lock of
+    /// its own that nothing contends.
+    #[tokio::test]
+    async fn an_identity_request_waits_for_the_users_file_lock() {
+        use std::os::unix::fs::PermissionsExt as _;
+        use std::time::Duration;
+
+        let mut config = config_with_one_operator();
+        let dir = config.users_file.parent().unwrap().to_owned();
+        let authelia = dir.join("authelia");
+        std::fs::write(
+            &authelia,
+            "#!/bin/sh\nprintf 'Random Password: pw\\nDigest: $argon2id$x\\n'\n",
+        )
+        .unwrap();
+        std::fs::set_permissions(&authelia, std::fs::Permissions::from_mode(0o755)).unwrap();
+        config.authelia_bin = authelia;
+
+        let held = crate::store::lock_users_file(&config.users_file, Duration::from_secs(1))
+            .await
+            .unwrap();
+        let write_lock = std::sync::Arc::new(tokio::sync::Mutex::new(()));
+        let request = {
+            let config = std::sync::Arc::new(config);
+            let write_lock = std::sync::Arc::clone(&write_lock);
+            let users_file = config.users_file.clone();
+            (
+                tokio::spawn(async move {
+                    handle(&config, &write_lock, "atlas".to_owned())
+                        .await
+                        .map(|_| ())
+                }),
+                users_file,
+            )
+        };
+        let (mut request, users_file) = request;
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(300), &mut request)
+                .await
+                .is_err(),
+            "the request finished while another process held the lock"
+        );
+        drop(held);
+        request.await.unwrap().expect("request after release");
+        assert!(
+            crate::store::load_store(&users_file)
+                .unwrap()
+                .users
+                .contains_key("atlas")
+        );
+        let _ = std::fs::remove_dir_all(users_file.parent().unwrap());
+    }
 }

swarm-authelia-bridge/src/store.rs

@@ -19,8 +19,9 @@
 use std::collections::BTreeMap;
 use std::fs::{self, File, Permissions};
 use std::io::Write as _;
-use std::os::unix::fs::PermissionsExt as _;
+use std::os::unix::fs::{OpenOptionsExt as _, PermissionsExt as _};
 use std::path::Path;
+use std::time::Duration;
 
 use anyhow::{Context, Result, bail};
 use serde::{Deserialize, Serialize};
@@ -277,6 +278,72 @@ fn fill_missing_emails(store: &mut UserStore) {
     }
 }
 
+/// How long a writer waits for the other process's `users.yml` update.
+pub const USERS_LOCK_WAIT: Duration = Duration::from_secs(10);
+
+/// Take the exclusive `flock` on `<users_file>.lock`, held until the
+/// returned file drops.
+///
+/// `swarmctl` rewrites the same file from the host and takes the same lock on
+/// the same inode. `handle` holds it across load → modify → [`publish`]:
+/// the in-process `write_lock` does not cover that other process, and the
+/// rename alone would not stop two writers loading the same snapshot.
+///
+/// The lock file is a sibling of `users_file`, not `users_file` itself,
+/// because `write_atomic` replaces that inode on every write. It is opened
+/// `O_NOFOLLOW` and `O_NONBLOCK` (a planted symlink or FIFO cannot redirect or
+/// hang it) and must be a regular file; no ownership is touched by path.
+pub async fn lock_users_file(users_file: &Path, wait: Duration) -> Result<File> {
+    let dir = users_file
+        .parent()
+        .with_context(|| format!("{} has no parent directory", users_file.display()))?;
+    fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
+    let name = users_file
+        .file_name()
+        .with_context(|| format!("{} has no file name", users_file.display()))?;
+    let mut lock_name = name.to_os_string();
+    lock_name.push(".lock");
+    let lock_path = dir.join(lock_name);
+
+    let flags = rustix::fs::OFlags::NOFOLLOW | rustix::fs::OFlags::NONBLOCK;
+    let file = fs::OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .mode(0o600)
+        .custom_flags(flags.bits().cast_signed())
+        .open(&lock_path)
+        .with_context(|| format!("opening {}", lock_path.display()))?;
+    if !file
+        .metadata()
+        .with_context(|| format!("stat {}", lock_path.display()))?
+        .is_file()
+    {
+        bail!("{} is not a regular file", lock_path.display());
+    }
+
+    let deadline = tokio::time::Instant::now() + wait;
+    loop {
+        match rustix::fs::flock(&file, rustix::fs::FlockOperation::NonBlockingLockExclusive) {
+            Ok(()) => return Ok(file),
+            Err(e)
+                if e == rustix::io::Errno::WOULDBLOCK && tokio::time::Instant::now() < deadline =>
+            {
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+            Err(e) if e == rustix::io::Errno::WOULDBLOCK => bail!(
+                "timed out after {wait:?} waiting for {}: another process (`swarmctl user`) \
+                 is updating {}",
+                lock_path.display(),
+                users_file.display()
+            ),
+            Err(e) => {
+                return Err(e).with_context(|| format!("locking {}", lock_path.display()));
+            }
+        }
+    }
+}
+
 /// Replace `path`'s contents atomically. Unlike `swarmctl::write_atomic`,
 /// this does **not** need to preserve a foreign owner via `chown` — this
 /// process runs as `users_file`'s own owning user (see the module doc:
@@ -637,6 +704,70 @@ some_future_top_level_key: 7
         fs::remove_dir_all(&dir).ok();
     }
 
+    #[tokio::test]
+    async fn the_users_lock_excludes_a_second_holder_until_it_is_released() {
+        let dir = tempdir();
+        let users_file = dir.join("users.yml");
+        let held = lock_users_file(&users_file, Duration::from_secs(1))
+            .await
+            .expect("first lock");
+
+        let err = lock_users_file(&users_file, Duration::from_millis(200))
+            .await
+            .expect_err("a held lock must not be granted twice");
+        assert!(err.to_string().contains("timed out"), "{err}");
+
+        drop(held);
+        lock_users_file(&users_file, Duration::from_millis(200))
+            .await
+            .expect("a released lock is granted");
+
+        fs::remove_dir_all(&dir).ok();
+    }
+
+    #[tokio::test]
+    async fn the_users_lock_refuses_a_symlink_planted_at_the_lock_path() {
+        let dir = tempdir();
+        let users_file = dir.join("users.yml");
+        let target = dir.join("victim");
+        fs::write(&target, "keep").unwrap();
+        std::os::unix::fs::symlink(&target, dir.join("users.yml.lock")).unwrap();
+
+        lock_users_file(&users_file, Duration::from_secs(1))
+            .await
+            .expect_err("a symlinked lock path must be refused");
+        assert_eq!(fs::read_to_string(&target).unwrap(), "keep");
+
+        fs::remove_dir_all(&dir).ok();
+    }
+
+    /// Two read-modify-writes that each sit between load and publish long
+    /// enough for the other to load: without the lock the second `publish`
+    /// drops the first user.
+    #[tokio::test]
+    async fn interleaved_read_modify_writes_both_land() {
+        let dir = tempdir();
+        let users_file = dir.join("users.yml");
+        let add = |name: &'static str| {
+            let users_file = users_file.clone();
+            tokio::spawn(async move {
+                let _lock = lock_users_file(&users_file, Duration::from_secs(5)).await?;
+                let mut store = load_store(&users_file)?;
+                tokio::time::sleep(Duration::from_millis(300)).await;
+                store.users.insert(name.to_owned(), user("$argon2id$x"));
+                publish(&users_file, &mut store)
+            })
+        };
+        let (a, b) = (add("alice"), add("bob"));
+        a.await.unwrap().expect("alice");
+        b.await.unwrap().expect("bob");
+
+        let names: Vec<_> = load_store(&users_file).unwrap().users.into_keys().collect();
+        assert_eq!(names, ["alice", "bob"]);
+
+        fs::remove_dir_all(&dir).ok();
+    }
+
     /// Test-only temp dir under the OS temp root — disposable scratch for
     /// one test's lifetime, not durable state (see `state-not-tmp`: this
     /// is exactly the legitimate use, not a place we persist anything).

swarmctl/README.md

@@ -31,6 +31,8 @@ controller can queue (see below). Nothing here becomes reachable by that.
 
 `swarmctl` reads and writes `users.yml` — authelia's own users database —
 directly. `swarm-authelia-bridge` writes agent subjects into the same file.
+Every verb that writes holds an exclusive `flock` on `users.yml.lock` from
+before the load until the rename, as the bridge does.
 
 ⚠️ The file is round-tripped, so **comments and hand-formatting do not
 survive a write**. Values do, and so do keys this binary does not model.

swarmctl/src/main.rs

@@ -39,6 +39,7 @@ use std::io::{self, Write as _};
 use std::os::unix::fs::{MetadataExt as _, OpenOptionsExt as _, PermissionsExt as _};
 use std::path::{Path, PathBuf};
 use std::process::Command;
+use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result, bail};
 use clap::{Args, Parser, Subcommand};
@@ -758,6 +759,7 @@ fn schedule(remote: &RemoteArgs, command: ScheduleVerb) -> Result<()> {
 fn user_add(paths: &Paths, args: AddArgs) -> Result<()> {
     users::validate_username(&args.username)?;
 
+    let _lock = lock_users_file(&paths.users_file)?;
     let mut store = load_store(&paths.users_file)?;
     if store.users.contains_key(&args.username) {
         bail!(
@@ -789,6 +791,7 @@ fn user_add(paths: &Paths, args: AddArgs) -> Result<()> {
 }
 
 fn user_reset_password(paths: &Paths, args: &ResetPasswordArgs) -> Result<()> {
+    let _lock = lock_users_file(&paths.users_file)?;
     let mut store = load_store(&paths.users_file)?;
     let Some(user) = store.users.get_mut(&args.username) else {
         bail!(
@@ -814,6 +817,7 @@ fn user_reset_password(paths: &Paths, args: &ResetPasswordArgs) -> Result<()> {
 }
 
 fn user_update(paths: &Paths, args: UpdateArgs) -> Result<()> {
+    let _lock = lock_users_file(&paths.users_file)?;
     let mut store = load_store(&paths.users_file)?;
     let Some(user) = store.users.get_mut(&args.username) else {
         bail!(
@@ -981,6 +985,81 @@ fn parse_field(stdout: &str, marker: &str) -> Option<String> {
         .filter(|value| !value.is_empty())
 }
 
+/// How long a writer waits for the other process's `users.yml` update.
+const USERS_LOCK_WAIT: Duration = Duration::from_secs(10);
+
+/// Take the exclusive `flock` on `<users_file>.lock`, held until the
+/// returned file drops.
+///
+/// `swarm-authelia-bridge` rewrites the same file from inside the
+/// `swarm-authelia` container and takes the same lock on the same inode.
+/// Every verb holds it across load → modify → [`publish`]: the rename alone
+/// would not stop two writers loading the same snapshot, and the second
+/// rename would drop the first one's user.
+///
+/// The lock file is a sibling of `users_file`, not `users_file` itself,
+/// because `write_atomic` replaces that inode on every write. The directory
+/// belongs to a less-privileged uid, so the lock file is opened `O_NOFOLLOW`
+/// (a planted symlink fails the open) and `O_NONBLOCK` (a planted FIFO cannot
+/// hang it), and must be a regular file. A file created here is handed to the
+/// directory's owner by descriptor so that uid can open it too.
+fn lock_users_file(users_file: &Path) -> Result<File> {
+    lock_users_file_within(users_file, USERS_LOCK_WAIT)
+}
+
+fn lock_users_file_within(users_file: &Path, wait: Duration) -> Result<File> {
+    let dir = users_file
+        .parent()
+        .with_context(|| format!("{} has no parent directory", users_file.display()))?;
+    fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
+    let name = users_file
+        .file_name()
+        .with_context(|| format!("{} has no file name", users_file.display()))?;
+    let mut lock_name = name.to_os_string();
+    lock_name.push(".lock");
+    let lock_path = dir.join(lock_name);
+
+    let flags = rustix::fs::OFlags::NOFOLLOW | rustix::fs::OFlags::NONBLOCK;
+    let file = fs::OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .mode(0o600)
+        .custom_flags(flags.bits().cast_signed())
+        .open(&lock_path)
+        .with_context(|| format!("opening {}", lock_path.display()))?;
+    let meta = file
+        .metadata()
+        .with_context(|| format!("stat {}", lock_path.display()))?;
+    if !meta.is_file() {
+        bail!("{} is not a regular file", lock_path.display());
+    }
+    let dir_meta = fs::metadata(dir).with_context(|| format!("stat {}", dir.display()))?;
+    if (meta.uid(), meta.gid()) != (dir_meta.uid(), dir_meta.gid()) {
+        std::os::unix::fs::fchown(&file, Some(dir_meta.uid()), Some(dir_meta.gid()))
+            .with_context(|| format!("setting owner on {}", lock_path.display()))?;
+    }
+
+    let deadline = Instant::now() + wait;
+    loop {
+        match rustix::fs::flock(&file, rustix::fs::FlockOperation::NonBlockingLockExclusive) {
+            Ok(()) => return Ok(file),
+            Err(e) if e == rustix::io::Errno::WOULDBLOCK && Instant::now() < deadline => {
+                std::thread::sleep(Duration::from_millis(50));
+            }
+            Err(e) if e == rustix::io::Errno::WOULDBLOCK => bail!(
+                "timed out after {wait:?} waiting for {}: another process (the \
+                 authelia bridge or another `swarmctl user` verb) is updating {}",
+                lock_path.display(),
+                users_file.display()
+            ),
+            Err(e) => {
+                return Err(e).with_context(|| format!("locking {}", lock_path.display()));
+            }
+        }
+    }
+}
+
 /// Replace `path`'s contents atomically, preserving the existing owner
 /// and mode.
 ///
@@ -1388,6 +1467,156 @@ mod tests {
         fs::remove_dir_all(&dir).ok();
     }
 
+    /// A scratch directory holding a `users.yml` with one user, `carol`, and
+    /// a stand-in `authelia` that takes `delay` seconds to mint a password.
+    fn locked_users_fixture(tag: &str, delay: &str) -> Paths {
+        let dir = std::env::temp_dir().join(format!("swarmctl-lock-{tag}-{}", std::process::id()));
+        let _ = fs::remove_dir_all(&dir);
+        fs::create_dir_all(&dir).expect("temp dir");
+        let authelia_bin = dir.join("authelia");
+        fs::write(
+            &authelia_bin,
+            format!(
+                "#!/bin/sh\nsleep {delay}\nprintf 'Random Password: pw\\nDigest: $argon2id$x\\n'\n"
+            ),
+        )
+        .expect("fake authelia");
+        fs::set_permissions(&authelia_bin, Permissions::from_mode(0o755)).expect("chmod");
+        let users_file = dir.join("users.yml");
+        fs::write(
+            &users_file,
+            "users:\n  carol:\n    displayname: carol\n    password: x\n    groups: []\n",
+        )
+        .expect("seed");
+        Paths {
+            authelia_bin,
+            users_file,
+        }
+    }
+
+    fn add_args(username: &str) -> AddArgs {
+        AddArgs {
+            username: username.to_owned(),
+            display_name: None,
+            email: None,
+            groups: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn the_users_lock_excludes_a_second_holder_until_it_is_released() {
+        let paths = locked_users_fixture("exclusive", "0");
+        let held = lock_users_file(&paths.users_file).expect("first lock");
+
+        let err = lock_users_file_within(&paths.users_file, Duration::from_millis(200))
+            .expect_err("a held lock must not be granted twice");
+        assert!(err.to_string().contains("timed out"), "{err}");
+
+        drop(held);
+        lock_users_file_within(&paths.users_file, Duration::from_millis(200))
+            .expect("a released lock is granted");
+
+        fs::remove_dir_all(paths.users_file.parent().unwrap()).ok();
+    }
+
+    #[test]
+    fn the_users_lock_refuses_a_symlink_planted_at_the_lock_path() {
+        let paths = locked_users_fixture("symlink", "0");
+        let dir = paths.users_file.parent().unwrap();
+        let target = dir.join("victim");
+        fs::write(&target, "keep").expect("victim");
+        std::os::unix::fs::symlink(&target, dir.join("users.yml.lock")).expect("plant");
+
+        lock_users_file(&paths.users_file).expect_err("a symlinked lock path must be refused");
+        assert_eq!(fs::read_to_string(&target).expect("read"), "keep");
+
+        fs::remove_dir_all(dir).ok();
+    }
+
+    /// Both adds load the file before either publishes (each spends 0.5s in
+    /// authelia), so without the lock the second rename drops the first user.
+    #[test]
+    fn two_concurrent_user_adds_both_land() {
+        let paths = std::sync::Arc::new(locked_users_fixture("adds", "0.5"));
+        let handles: Vec<_> = ["alice", "bob"]
+            .into_iter()
+            .map(|name| {
+                let paths = std::sync::Arc::clone(&paths);
+                std::thread::spawn(move || user_add(&paths, add_args(name)))
+            })
+            .collect();
+        for handle in handles {
+            handle.join().expect("thread").expect("user add");
+        }
+
+        let store = load_store(&paths.users_file).expect("reload");
+        let names: Vec<_> = store.users.keys().map(String::as_str).collect();
+        assert_eq!(names, ["alice", "bob", "carol"]);
+
+        fs::remove_dir_all(paths.users_file.parent().unwrap()).ok();
+    }
+
+    #[test]
+    fn every_user_verb_waits_for_the_users_lock() {
+        let paths = std::sync::Arc::new(locked_users_fixture("verbs", "0"));
+        let held = lock_users_file(&paths.users_file).expect("lock");
+
+        let (add_tx, add_rx) = std::sync::mpsc::channel();
+        let (reset_tx, reset_rx) = std::sync::mpsc::channel();
+        let (update_tx, update_rx) = std::sync::mpsc::channel();
+        let p = std::sync::Arc::clone(&paths);
+        std::thread::spawn(move || add_tx.send(user_add(&p, add_args("dave"))));
+        let p = std::sync::Arc::clone(&paths);
+        std::thread::spawn(move || {
+            let args = ResetPasswordArgs {
+                username: "carol".to_owned(),
+            };
+            reset_tx.send(user_reset_password(&p, &args))
+        });
+        let p = std::sync::Arc::clone(&paths);
+        std::thread::spawn(move || {
+            let args = UpdateArgs {
+                username: "carol".to_owned(),
+                display_name: None,
+                email: None,
+                add_groups: vec!["ops".to_owned()],
+                remove_groups: Vec::new(),
+            };
+            update_tx.send(user_update(&p, args))
+        });
+
+        let wait = Duration::from_millis(300);
+        assert!(add_rx.recv_timeout(wait).is_err(), "user add ran unlocked");
+        assert!(
+            reset_rx.recv_timeout(wait).is_err(),
+            "user reset-password ran unlocked"
+        );
+        assert!(
+            update_rx.recv_timeout(wait).is_err(),
+            "user update ran unlocked"
+        );
+
+        drop(held);
+        let done = Duration::from_secs(5);
+        add_rx
+            .recv_timeout(done)
+            .expect("add finishes")
+            .expect("add");
+        reset_rx
+            .recv_timeout(done)
+            .expect("reset finishes")
+            .expect("reset");
+        update_rx
+            .recv_timeout(done)
+            .expect("update finishes")
+            .expect("update");
+        let store = load_store(&paths.users_file).expect("reload");
+        assert!(store.users.contains_key("dave"));
+        assert_eq!(store.users["carol"].groups, ["ops"]);
+
+        fs::remove_dir_all(paths.users_file.parent().unwrap()).ok();
+    }
+
     #[test]
     fn parses_authelia_hash_output() {
         let out = "Random Password: hunter2\nDigest: $argon2id$v=19$m=65536$abc\n";