summaryrefslogtreecommitdiff
path: root/src/controller.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/controller.rs')
-rw-r--r--src/controller.rs478
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(())
+}
generated by cgit v1.2.3 (git 2.25.1) at 2026-10-30 11:10:06 +0000