Implement Linux proxy controller, subscription state, and CLI/TUI

1 files changed, 264 insertions, 0 deletions

diff --git a/src/storage.rs b/src/storage.rs
new file mode 100644
index 0000000..cb7fcd9
--- /dev/null
+++ b/src/storage.rs
@@ -0,0 +1,264 @@
+use crate::model::{Definitions, Entry};
+use anyhow::{Context, Result, ensure};
+use rusqlite::{Connection, OptionalExtension, params};
+use std::{
+    fs,
+    io::Write,
+    os::unix::fs::{MetadataExt, PermissionsExt},
+    path::{Path, PathBuf},
+};
+
+pub struct Storage {
+    db: Connection,
+    pub definitions: Definitions,
+    config: PathBuf,
+}
+
+pub fn private_directory(path: &Path) -> Result<()> {
+    fs::create_dir_all(path).context("Create Roentgen directory")?;
+    let metadata = fs::symlink_metadata(path)?;
+    ensure!(
+        metadata.is_dir() && !metadata.file_type().is_symlink(),
+        "Roentgen directory must not be a symlink"
+    );
+    ensure!(
+        metadata.uid() == unsafe { libc::geteuid() },
+        "Roentgen directory must belong to the service user"
+    );
+    fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
+    Ok(())
+}
+
+pub fn atomic_private_write(path: &Path, data: &[u8]) -> Result<()> {
+    let parent = path.parent().context("Missing parent directory")?;
+    let mut file = tempfile::NamedTempFile::new_in(parent)?;
+    file.as_file()
+        .set_permissions(fs::Permissions::from_mode(0o600))?;
+    file.write_all(data)?;
+    file.as_file().sync_all()?;
+    file.persist(path).map_err(|e| e.error)?;
+    fs::File::open(parent)?.sync_all()?;
+    Ok(())
+}
+
+impl Storage {
+    pub fn open(config_dir: &Path, data_dir: &Path) -> Result<Self> {
+        private_directory(config_dir)?;
+        private_directory(data_dir)?;
+        let config = config_dir.join("config.toml");
+        let definitions: Definitions = if config.exists() {
+            ensure!(
+                fs::symlink_metadata(&config)?.is_file(),
+                "Editable definitions must be a regular file"
+            );
+            toml::from_str(&fs::read_to_string(&config)?)
+                .context("Invalid Roentgen TOML; repair it before starting")?
+        } else {
+            Definitions::default()
+        };
+        ensure!(definitions.version == 1, "Unsupported TOML version");
+        definitions.settings.validate()?;
+        let mut ids = std::collections::HashSet::new();
+        let mut names = std::collections::HashSet::new();
+        for subscription in &definitions.subscriptions {
+            ensure!(
+                !subscription.id.is_empty()
+                    && ids.insert(subscription.id.clone())
+                    && !subscription.name.is_empty()
+                    && names.insert(subscription.name.clone()),
+                "Subscription IDs and names must be unique and nonempty"
+            );
+            crate::subscription::subscription_url(&subscription.url)?;
+        }
+        let path = data_dir.join("state.sqlite3");
+        if path.exists() {
+            ensure!(
+                fs::symlink_metadata(&path)?.is_file(),
+                "State database must be a regular file"
+            );
+        }
+        let db = Connection::open(&path)?;
+        fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?;
+        db.busy_timeout(std::time::Duration::from_secs(5))?;
+        db.execute_batch("PRAGMA synchronous=FULL; PRAGMA foreign_keys=ON;")?;
+        let version: u32 = db.query_row("PRAGMA user_version", [], |r| r.get(0))?;
+        ensure!(version <= 1, "Unsupported database version");
+        if version == 0 {
+            db.execute_batch("BEGIN IMMEDIATE;
+                CREATE TABLE entries (subscription_id TEXT NOT NULL, ordinal INTEGER NOT NULL, id TEXT PRIMARY KEY, data TEXT NOT NULL);
+                CREATE TABLE state (singleton INTEGER PRIMARY KEY CHECK(singleton=1), enabled INTEGER NOT NULL CHECK(enabled IN (0,1)), selected TEXT, generation INTEGER NOT NULL);
+                INSERT INTO state VALUES (1,0,NULL,0);
+                PRAGMA user_version=1; COMMIT;")?;
+        }
+        let store = Self {
+            db,
+            definitions,
+            config,
+        };
+        // Removed TOML definitions never authorize retained snapshots or a stale selection.
+        let known: std::collections::HashSet<_> = store
+            .definitions
+            .subscriptions
+            .iter()
+            .map(|s| s.id.as_str())
+            .collect();
+        for subscription_id in store
+            .db
+            .prepare("SELECT DISTINCT subscription_id FROM entries")?
+            .query_map([], |r| r.get::<_, String>(0))?
+            .collect::<rusqlite::Result<Vec<_>>>()?
+        {
+            if !known.contains(subscription_id.as_str()) {
+                store.delete_entries(&subscription_id)?;
+            }
+        }
+        if !store.config.exists() {
+            store.save_definitions(&store.definitions)?;
+        }
+        Ok(store)
+    }
+
+    pub fn save_definitions(&self, definitions: &Definitions) -> Result<()> {
+        atomic_private_write(
+            &self.config,
+            toml::to_string_pretty(definitions)?.as_bytes(),
+        )
+    }
+
+    pub fn state(&self) -> Result<(bool, Option<String>, u64)> {
+        Ok(self.db.query_row(
+            "SELECT enabled,selected,generation FROM state WHERE singleton=1",
+            [],
+            |r| Ok((r.get(0)?, r.get(1)?, r.get::<_, i64>(2)? as u64)),
+        )?)
+    }
+
+    pub fn set_intent(&self, enabled: bool) -> Result<()> {
+        self.db.execute(
+            "UPDATE state SET enabled=?1,generation=generation+1 WHERE singleton=1",
+            [enabled],
+        )?;
+        Ok(())
+    }
+
+    pub fn select(&self, id: Option<&str>) -> Result<()> {
+        self.db.execute(
+            "UPDATE state SET selected=?1,generation=generation+1 WHERE singleton=1",
+            [id],
+        )?;
+        Ok(())
+    }
+
+    pub fn entries(&self, subscription_id: &str) -> Result<Vec<Entry>> {
+        let mut query = self
+            .db
+            .prepare("SELECT data FROM entries WHERE subscription_id=?1 ORDER BY ordinal")?;
+        let values = query
+            .query_map([subscription_id], |r| r.get::<_, String>(0))?
+            .collect::<rusqlite::Result<Vec<_>>>()?;
+        values
+            .iter()
+            .map(|v| {
+                serde_json::from_str(v)
+                    .context("Corrupt saved configuration; explicit recovery is required")
+            })
+            .collect()
+    }
+
+    pub fn find_entry(&self, id: &str) -> Result<Option<Entry>> {
+        self.db
+            .query_row("SELECT data FROM entries WHERE id=?1", [id], |r| {
+                r.get::<_, String>(0)
+            })
+            .optional()?
+            .map(|v| serde_json::from_str(&v).context("Corrupt saved configuration"))
+            .transpose()
+    }
+
+    pub fn replace_entries(&mut self, subscription_id: &str, entries: &[Entry]) -> Result<()> {
+        let transaction = self.db.transaction()?;
+        let selected: Option<String> =
+            transaction.query_row("SELECT selected FROM state WHERE singleton=1", [], |r| {
+                r.get(0)
+            })?;
+        let owns_selection = match selected.as_ref() {
+            Some(id) => contains_id(&transaction, subscription_id, id)?,
+            None => false,
+        };
+        transaction.execute(
+            "DELETE FROM entries WHERE subscription_id=?1",
+            [subscription_id],
+        )?;
+        for (ordinal, entry) in entries.iter().enumerate() {
+            transaction.execute(
+                "INSERT INTO entries VALUES (?1,?2,?3,?4)",
+                params![
+                    subscription_id,
+                    ordinal as i64,
+                    entry.id,
+                    serde_json::to_string(entry)?
+                ],
+            )?;
+        }
+        if owns_selection
+            && !entries
+                .iter()
+                .any(|e| Some(&e.id) == selected.as_ref() && e.connection.is_some())
+        {
+            transaction.execute("UPDATE state SET selected=NULL WHERE singleton=1", [])?;
+        }
+        transaction.execute(
+            "UPDATE state SET generation=generation+1 WHERE singleton=1",
+            [],
+        )?;
+        transaction.commit()?;
+        Ok(())
+    }
+
+    pub fn delete_entries(&self, subscription_id: &str) -> Result<()> {
+        self.db.execute_batch("BEGIN IMMEDIATE")?;
+        let result = (|| -> Result<()> {
+            self.db.execute("UPDATE state SET selected=NULL,generation=generation+1 WHERE singleton=1 AND selected IN (SELECT id FROM entries WHERE subscription_id=?1)", [subscription_id])?;
+            self.db.execute(
+                "DELETE FROM entries WHERE subscription_id=?1",
+                [subscription_id],
+            )?;
+            self.db.execute_batch("COMMIT")?;
+            Ok(())
+        })();
+        if result.is_err() {
+            let _ = self.db.execute_batch("ROLLBACK");
+        }
+        result
+    }
+}
+
+fn contains_id(db: &Connection, subscription_id: &str, id: &str) -> Result<bool> {
+    Ok(db.query_row(
+        "SELECT EXISTS(SELECT 1 FROM entries WHERE subscription_id=?1 AND id=?2)",
+        params![subscription_id, id],
+        |r| r.get(0),
+    )?)
+}
+
+/// Recovery deliberately needs no valid editable definitions or imported entries.
+pub fn persist_recovery_disabled(data_dir: &Path) -> Result<()> {
+    private_directory(data_dir)?;
+    let path = data_dir.join("state.sqlite3");
+    if path.exists() {
+        ensure!(
+            fs::symlink_metadata(&path)?.is_file(),
+            "Recovery database must be a regular file"
+        );
+    }
+    let db = Connection::open(&path)?;
+    fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?;
+    db.busy_timeout(std::time::Duration::from_secs(5))?;
+    db.execute_batch("PRAGMA synchronous=FULL; BEGIN EXCLUSIVE;")?;
+    db.execute(
+        "UPDATE state SET enabled=0,generation=generation+1 WHERE singleton=1",
+        [],
+    )?;
+    db.execute_batch("COMMIT")?;
+    Ok(())
+}