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

1 files changed, 478 insertions, 0 deletions

diff --git a/src/controller.rs b/src/controller.rs
new file mode 100644
index 0000000..1384ad8
--- /dev/null
+++ b/src/controller.rs
@@ -0,0 +1,478 @@
+use crate::{
+    model::*,
+    runtime::Runtime,
+    storage::Storage,
+    subscription::{SubscriptionSource, import, subscription_url},
+};
+use anyhow::{Result, anyhow, ensure};
+use serde::{Deserialize, Serialize};
+use std::path::Path;
+use uuid::Uuid;
+
+/// Constructed by the transport after resolving the real bus sender and checking polkit.
+pub struct Caller {
+    authorized: bool,
+}
+impl Caller {
+    pub fn root() -> Self {
+        Self { authorized: true }
+    }
+    pub fn denied() -> Self {
+        Self { authorized: false }
+    }
+    pub(crate) fn authorized() -> Self {
+        Self { authorized: true }
+    }
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+pub enum Operation {
+    AddSubscription { name: String, url: String },
+    EditSubscription { name: String, url: String },
+    DeleteSubscription { name: String },
+    Refresh { name: Option<String> },
+    Select { id: String },
+    Unset,
+    Enable,
+    Disable,
+    SaveSettings { settings: Settings },
+}
+
+pub struct Controller {
+    storage: Storage,
+    runtime: Box<dyn Runtime>,
+    source: Box<dyn SubscriptionSource>,
+    last_error: Option<String>,
+    observation: Option<crate::diagnostics::Observation>,
+    logs: std::collections::VecDeque<String>,
+    next_check: Option<tokio::time::Instant>,
+    measurements: Vec<crate::diagnostics::Observation>,
+}
+
+impl Controller {
+    pub fn open(
+        root: &Path,
+        runtime: Box<dyn Runtime>,
+        source: Box<dyn SubscriptionSource>,
+    ) -> Result<Self> {
+        Self::open_paths(&root.join("etc"), &root.join("data"), runtime, source)
+    }
+
+    pub fn open_paths(
+        config: &Path,
+        data: &Path,
+        runtime: Box<dyn Runtime>,
+        source: Box<dyn SubscriptionSource>,
+    ) -> Result<Self> {
+        Ok(Self {
+            storage: Storage::open(config, data)?,
+            runtime,
+            source,
+            last_error: None,
+            observation: None,
+            logs: std::collections::VecDeque::new(),
+            next_check: Some(tokio::time::Instant::now()),
+            measurements: vec![],
+        })
+    }
+
+    pub fn status(&self) -> Result<Status> {
+        let (requested_enabled, selected_configuration, generation) = self.storage.state()?;
+        Ok(Status {
+            version: 1,
+            requested_enabled,
+            selected_configuration,
+            generation,
+            routing: self.runtime.routing().into(),
+            health: self
+                .observation
+                .as_ref()
+                .filter(|o| o.generation == generation)
+                .map(|o| {
+                    if o.success {
+                        "reachable"
+                    } else {
+                        "check-failed"
+                    }
+                })
+                .unwrap_or("unknown")
+                .into(),
+            last_error: self.last_error.clone(),
+        })
+    }
+
+    pub fn subscriptions(&self) -> Vec<SubscriptionSummary> {
+        self.storage
+            .definitions
+            .subscriptions
+            .iter()
+            .map(|s| SubscriptionSummary {
+                id: s.id.clone(),
+                name: s.name.clone(),
+            })
+            .collect()
+    }
+
+    pub fn settings(&self) -> Settings {
+        self.storage.definitions.settings.clone()
+    }
+
+    pub fn configurations(&self, name: Option<&str>) -> Result<Vec<ConfigurationSummary>> {
+        let (_, selected, _) = self.storage.state()?;
+        if let Some(name) = name {
+            self.subscription(name)?;
+        }
+        let mut rows = vec![];
+        for subscription in &self.storage.definitions.subscriptions {
+            if name.is_some_and(|name| name != subscription.name) {
+                continue;
+            }
+            for entry in self.storage.entries(&subscription.id)? {
+                let ping = self
+                    .measurements
+                    .iter()
+                    .find(|o| o.configuration_id == entry.id && o.kind == "ping" && o.success);
+                let speed = self
+                    .measurements
+                    .iter()
+                    .find(|o| o.configuration_id == entry.id && o.kind == "speed" && o.success);
+                rows.push(ConfigurationSummary {
+                    ping_ms: ping.map(|o| o.elapsed_ms),
+                    speed_bytes_per_second: speed.and_then(|o| o.bytes_per_second),
+                    selected: Some(&entry.id) == selected.as_ref(),
+                    id: entry.id,
+                    subscription_id: subscription.id.clone(),
+                    name: entry.name,
+                    host: entry.connection.as_ref().map(|c| c.host.clone()),
+                    port: entry.connection.as_ref().map(|c| c.port),
+                    status: entry.status,
+                    explanation: entry.explanation,
+                });
+            }
+        }
+        Ok(rows)
+    }
+
+    pub async fn restore(&mut self) -> Result<()> {
+        if self.storage.state()?.0 {
+            self.runtime.arm().await?;
+            self.reconcile().await?;
+        } else {
+            self.runtime.disable().await?;
+        }
+        Ok(())
+    }
+
+    pub async fn execute(&mut self, caller: &Caller, operation: Operation) -> Result<()> {
+        ensure!(
+            caller.authorized,
+            "Administrator authorization required; no state changed"
+        );
+        self.last_error = None;
+        self.record("Authorized controller operation".into());
+        match operation {
+            Operation::AddSubscription { name, url } => {
+                validate_name(&name)?;
+                subscription_url(&url)?;
+                ensure!(
+                    !self
+                        .storage
+                        .definitions
+                        .subscriptions
+                        .iter()
+                        .any(|s| s.name == name),
+                    "A subscription with this name already exists"
+                );
+                let entries = import(&self.source.fetch(&url).await?)?;
+                let subscription = Subscription {
+                    id: Uuid::new_v4().to_string(),
+                    name,
+                    url,
+                };
+                // Orphan snapshots after a failed TOML write are removed on next startup.
+                self.storage.replace_entries(&subscription.id, &entries)?;
+                let mut definitions = self.storage.definitions.clone();
+                definitions.subscriptions.push(subscription);
+                self.storage.save_definitions(&definitions)?;
+                self.storage.definitions = definitions;
+            }
+            Operation::EditSubscription { name, url } => {
+                subscription_url(&url)?;
+                let id = self.subscription(&name)?.id.clone();
+                let mut definitions = self.storage.definitions.clone();
+                definitions
+                    .subscriptions
+                    .iter_mut()
+                    .find(|s| s.id == id)
+                    .unwrap()
+                    .url = url;
+                self.storage.save_definitions(&definitions)?;
+                self.storage.definitions = definitions;
+            }
+            Operation::DeleteSubscription { name } => {
+                let id = self.subscription(&name)?.id.clone();
+                let mut definitions = self.storage.definitions.clone();
+                definitions.subscriptions.retain(|s| s.id != id);
+                self.storage.save_definitions(&definitions)?;
+                self.storage.definitions = definitions;
+                self.storage.delete_entries(&id)?;
+                self.reconcile().await?;
+            }
+            Operation::Refresh { name } => {
+                let subscriptions = if let Some(name) = name {
+                    vec![self.subscription(&name)?.clone()]
+                } else {
+                    self.storage.definitions.subscriptions.clone()
+                };
+                let mut failures = 0;
+                for subscription in subscriptions {
+                    let result = self.refresh(&subscription).await;
+                    if let Err(error) = result {
+                        failures += 1;
+                        self.last_error = Some(error.to_string());
+                    }
+                }
+                ensure!(
+                    failures == 0,
+                    "{} subscription update(s) failed; {}",
+                    failures,
+                    self.last_error.as_deref().unwrap_or("retry update")
+                );
+            }
+            Operation::Select { id } => {
+                self.usable_entry(&id)?;
+                self.storage.select(Some(&id))?;
+                self.reconcile().await?;
+            }
+            Operation::Unset => {
+                self.storage.select(None)?;
+                self.reconcile().await?;
+            }
+            Operation::Enable => {
+                let (_, selected, _) = self.storage.state()?;
+                let id = selected
+                    .ok_or_else(|| anyhow!("Select a supported configuration before enabling"))?;
+                self.usable_entry(&id)?;
+                self.runtime.arm().await?;
+                self.storage.set_intent(true)?;
+                self.reconcile().await?;
+            }
+            Operation::Disable => {
+                self.storage.set_intent(false)?;
+                self.runtime.disable().await?;
+            }
+            Operation::SaveSettings { settings } => {
+                settings.validate()?;
+                let mut definitions = self.storage.definitions.clone();
+                definitions.settings = settings;
+                self.storage.save_definitions(&definitions)?;
+                self.storage.definitions = definitions;
+            }
+        }
+        Ok(())
+    }
+
+    fn record(&mut self, message: String) {
+        self.logs.push_back(message);
+        if self.logs.len() > 200 {
+            self.logs.pop_front();
+        }
+    }
+    pub fn report_runtime_error(&mut self, error: &anyhow::Error) {
+        self.last_error = Some(error.to_string());
+        self.record(error.to_string());
+    }
+    pub fn logs(&self) -> Vec<String> {
+        self.logs.iter().cloned().collect()
+    }
+
+    pub async fn diagnostic(
+        &mut self,
+        caller: &Caller,
+        id: Option<&str>,
+        speed: bool,
+    ) -> Result<crate::diagnostics::Observation> {
+        ensure!(
+            caller.authorized,
+            "Administrator authorization required; no state changed"
+        );
+        let (_, selected, generation) = self.storage.state()?;
+        let id = id
+            .or(selected.as_deref())
+            .ok_or_else(|| anyhow!("Select a configuration or supply its ID before testing"))?;
+        let entry = self.usable_entry(id)?;
+        let settings = self.settings();
+        let probe = crate::xray::Probe::start(
+            entry.connection.as_ref().unwrap(),
+            if speed {
+                settings.maximum_duration_seconds
+            } else {
+                5
+            },
+        )
+        .await?;
+        let result =
+            crate::diagnostics::measure(&settings, &probe.proxy, id.into(), generation, speed)
+                .await;
+        probe.stop().await;
+        match result {
+            Ok(observation) => {
+                if !speed && Some(id) == selected.as_deref() {
+                    let wait = observation.retry_after_seconds.unwrap_or(30).max(30);
+                    self.next_check = tokio::time::Instant::now()
+                        .checked_add(std::time::Duration::from_secs(wait));
+                    self.observation = Some(observation.clone());
+                }
+                self.measurements
+                    .retain(|o| o.configuration_id != id || o.kind != observation.kind);
+                self.measurements.push(observation.clone());
+                self.record(format!(
+                    "{} observation: {}",
+                    if speed { "Speed" } else { "Ping" },
+                    observation.stop_reason
+                ));
+                Ok(observation)
+            }
+            Err(error) => {
+                if !speed && Some(id) == selected.as_deref() {
+                    self.observation = Some(crate::diagnostics::Observation {
+                        configuration_id: id.into(),
+                        generation,
+                        kind: "ping".into(),
+                        success: false,
+                        elapsed_ms: 0,
+                        bytes: 0,
+                        bytes_per_second: None,
+                        stop_reason: error.to_string(),
+                        retry_after_seconds: None,
+                    });
+                }
+                self.record(error.to_string());
+                Err(error)
+            }
+        }
+    }
+
+    pub async fn scheduled_check(&mut self) -> Result<()> {
+        let (enabled, selected, generation) = self.storage.state()?;
+        if enabled {
+            let entry = selected
+                .as_deref()
+                .map(|id| self.usable_entry(id))
+                .transpose()?;
+            self.runtime
+                .maintain(
+                    entry.as_ref().and_then(|e| e.connection.as_ref()),
+                    generation,
+                )
+                .await?;
+        }
+        if selected.is_some()
+            && self
+                .next_check
+                .is_some_and(|deadline| tokio::time::Instant::now() >= deadline)
+        {
+            self.next_check =
+                Some(tokio::time::Instant::now() + std::time::Duration::from_secs(30));
+            self.diagnostic(&Caller::root(), None, false).await?;
+        }
+        Ok(())
+    }
+
+    async fn refresh(&mut self, subscription: &Subscription) -> Result<()> {
+        let mut entries = import(&self.source.fetch(&subscription.url).await?)?;
+        let old = self.storage.entries(&subscription.id)?;
+        // Match normalized content, then a unique endpoint/authentication tuple, never labels or positions.
+        let mut matched_old = std::collections::HashSet::new();
+        let mut matched_new = std::collections::HashSet::new();
+        for exact in [true, false] {
+            for i in 0..entries.len() {
+                if matched_new.contains(&i) {
+                    continue;
+                }
+                let Some(connection) = entries[i].connection.as_ref() else {
+                    continue;
+                };
+                let matches = |candidate: &Trojan| {
+                    if exact {
+                        candidate == connection
+                    } else {
+                        candidate.identity() == connection.identity()
+                    }
+                };
+                let candidates: Vec<_> = old
+                    .iter()
+                    .enumerate()
+                    .filter(|(j, old)| {
+                        !matched_old.contains(j) && old.connection.as_ref().is_some_and(matches)
+                    })
+                    .collect();
+                let new_count = entries
+                    .iter()
+                    .enumerate()
+                    .filter(|(j, e)| {
+                        !matched_new.contains(j) && e.connection.as_ref().is_some_and(matches)
+                    })
+                    .count();
+                if candidates.len() == 1 && new_count == 1 {
+                    entries[i].id = candidates[0].1.id.clone();
+                    matched_old.insert(candidates[0].0);
+                    matched_new.insert(i);
+                }
+            }
+        }
+        let affected: std::collections::HashSet<_> = old.iter().map(|e| e.id.as_str()).collect();
+        self.measurements
+            .retain(|o| !affected.contains(o.configuration_id.as_str()));
+        self.storage.replace_entries(&subscription.id, &entries)?;
+        self.reconcile().await
+    }
+
+    fn subscription(&self, name: &str) -> Result<&Subscription> {
+        self.storage
+            .definitions
+            .subscriptions
+            .iter()
+            .find(|s| s.name == name)
+            .ok_or_else(|| anyhow!("Subscription not found; run subscription list"))
+    }
+
+    pub fn usable_entry(&self, id: &str) -> Result<Entry> {
+        let entry = self
+            .storage
+            .find_entry(id)?
+            .ok_or_else(|| anyhow!("Configuration not found; run configuration list"))?;
+        ensure!(
+            entry.connection.is_some(),
+            "Configuration is invalid or unsupported; inspect its explanation"
+        );
+        Ok(entry)
+    }
+
+    async fn reconcile(&mut self) -> Result<()> {
+        let (enabled, selected, generation) = self.storage.state()?;
+        if enabled {
+            let entry = selected.map(|id| self.usable_entry(&id)).transpose()?;
+            if let Err(error) = self
+                .runtime
+                .apply(
+                    entry.as_ref().and_then(|e| e.connection.as_ref()),
+                    generation,
+                )
+                .await
+            {
+                self.last_error = Some(error.to_string());
+                return Err(error);
+            }
+        }
+        Ok(())
+    }
+}
+
+fn validate_name(name: &str) -> Result<()> {
+    ensure!(
+        !name.trim().is_empty() && name.len() <= 128 && !name.chars().any(char::is_control),
+        "Subscription name must be 1-128 bytes without control characters"
+    );
+    Ok(())
+}