summaryrefslogtreecommitdiff
path: root/tests/diagnostics_acceptance.rs
diff options
context:
space:
mode:
Diffstat (limited to 'tests/diagnostics_acceptance.rs')
-rw-r--r--tests/diagnostics_acceptance.rs605
1 files changed, 605 insertions, 0 deletions
diff --git a/tests/diagnostics_acceptance.rs b/tests/diagnostics_acceptance.rs
new file mode 100644
index 0000000..f1db2de
--- /dev/null
+++ b/tests/diagnostics_acceptance.rs
@@ -0,0 +1,605 @@
+use anyhow::{Result, bail};
+use async_trait::async_trait;
+use roentgen::{
+ controller::{Caller, Controller, Operation},
+ model::Trojan,
+ runtime::{ProbeConnection, ProbeSession, Runtime},
+ subscription::SubscriptionSource,
+ xray::{ProbeListener, probe_configuration},
+};
+use std::{path::PathBuf, process::Stdio, time::Duration};
+use tempfile::TempDir;
+
+struct Source {
+ port: u16,
+}
+#[async_trait]
+impl SubscriptionSource for Source {
+ async fn fetch(&self, _: &str) -> Result<Vec<u8>> {
+ Ok(format!(
+ "trojan://fixture-password@127.0.0.1:{}?sni=localhost#Fixture\n",
+ self.port
+ )
+ .into_bytes())
+ }
+}
+
+#[derive(Clone)]
+enum ProbeMode {
+ Available,
+ Unavailable,
+ Impostor,
+}
+struct FixtureRuntime {
+ mode: std::sync::Arc<std::sync::Mutex<ProbeMode>>,
+ impostor: String,
+}
+struct FixtureProbe {
+ child: Option<tokio::process::Child>,
+ _directory: TempDir,
+ proxy: String,
+ certificate: String,
+}
+#[async_trait]
+impl ProbeSession for FixtureProbe {
+ fn connection(&self) -> ProbeConnection<'_> {
+ ProbeConnection {
+ proxy_url: &self.proxy,
+ certificate: &self.certificate,
+ }
+ }
+ async fn stop(mut self: Box<Self>) {
+ if let Some(child) = &mut self.child {
+ let _ = child.kill().await;
+ let _ = child.wait().await;
+ }
+ }
+}
+#[async_trait]
+impl Runtime for FixtureRuntime {
+ async fn arm(&mut self) -> Result<()> {
+ bail!("Host activation is outside this diagnostic fixture")
+ }
+ async fn apply(&mut self, _: Option<&Trojan>, _: u64) -> Result<()> {
+ bail!("No TUN in diagnostic fixture")
+ }
+ async fn disable(&mut self) -> Result<()> {
+ Ok(())
+ }
+ fn routing(&self) -> &str {
+ "direct"
+ }
+ async fn probe(&mut self, connection: &Trojan, _: u64) -> Result<Box<dyn ProbeSession>> {
+ let mode = self.mode.lock().unwrap().clone();
+ if matches!(mode, ProbeMode::Unavailable) {
+ bail!("Fixture probe could not start");
+ }
+ let socket = std::net::TcpListener::bind("127.0.0.1:0")?;
+ let port = socket.local_addr()?.port();
+ let certificate = rcgen::generate_simple_self_signed(vec!["127.0.0.1".into()])?;
+ let certificate_pem = certificate.cert.pem();
+ let key = certificate.key_pair.serialize_pem();
+ let directory = tempfile::tempdir()?;
+ let path = directory.path().join("probe.json");
+ if matches!(mode, ProbeMode::Impostor) {
+ return Ok(Box::new(FixtureProbe {
+ child: None,
+ _directory: directory,
+ proxy: self.impostor.clone(),
+ certificate: certificate_pem,
+ }));
+ }
+ let config = probe_configuration(
+ connection,
+ &ProbeListener {
+ port,
+ password: "private-probe-password",
+ certificate: &certificate_pem,
+ private_key: &key,
+ },
+ );
+ std::fs::write(&path, serde_json::to_vec(&config)?)?;
+ drop(socket);
+ // The older distribution core is used only as a non-TUN fixture. Product
+ // host activation retains its independent compatibility requirement.
+ let child = tokio::process::Command::new("/usr/bin/xray")
+ .args(["run", "-config"])
+ .arg(path)
+ .kill_on_drop(true)
+ .stdin(Stdio::null())
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()?;
+ let probe = FixtureProbe {
+ child: Some(child),
+ _directory: directory,
+ proxy: format!("https://roentgen:private-probe-password@127.0.0.1:{port}"),
+ certificate: certificate_pem,
+ };
+ for _ in 0..100 {
+ if tokio::net::TcpStream::connect(("127.0.0.1", port))
+ .await
+ .is_ok()
+ {
+ return Ok(Box::new(probe));
+ }
+ tokio::time::sleep(Duration::from_millis(20)).await;
+ }
+ bail!("Fixture Xray proxy did not start")
+ }
+}
+
+#[derive(serde::Deserialize)]
+struct Fixture {
+ origin: String,
+ bad_origin: String,
+ trojan_port: u16,
+ access_log: PathBuf,
+ requests_log: PathBuf,
+}
+
+struct Context {
+ _home: TempDir,
+ controller: Controller,
+ fixture: Fixture,
+ id: String,
+ mode: std::sync::Arc<std::sync::Mutex<ProbeMode>>,
+}
+impl Context {
+ async fn open() -> Self {
+ let hostname = std::fs::read_to_string("/etc/hostname").unwrap();
+ assert_eq!(
+ hostname.trim(),
+ "altctlrs",
+ "Run diagnostic acceptance only on the authorized VM"
+ );
+ let path = std::env::var("ROENTGEN_DIAGNOSTIC_FIXTURE")
+ .expect("Use tests/vm_diagnostics_acceptance.py on the supplied VM");
+ let fixture: Fixture = serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap();
+ let home = tempfile::tempdir().unwrap();
+ let mode = std::sync::Arc::new(std::sync::Mutex::new(ProbeMode::Available));
+ let mut impostor = url::Url::parse(&fixture.bad_origin).unwrap();
+ impostor.set_username("roentgen").unwrap();
+ impostor
+ .set_password(Some("private-probe-password"))
+ .unwrap();
+ let runtime = FixtureRuntime {
+ mode: mode.clone(),
+ impostor: impostor.to_string(),
+ };
+ let mut controller = Controller::open(
+ home.path(),
+ Box::new(runtime),
+ Box::new(Source {
+ port: fixture.trojan_port,
+ }),
+ )
+ .unwrap();
+ let root = Caller::root();
+ controller
+ .execute(
+ &root,
+ Operation::AddSubscription {
+ name: "Fixture".into(),
+ url: "https://subscription.invalid/fixture".into(),
+ },
+ )
+ .await
+ .unwrap();
+ let id = controller.configurations(None).unwrap()[0].id.clone();
+ controller
+ .execute(&root, Operation::Select { id: id.clone() })
+ .await
+ .unwrap();
+ let mut result = Self {
+ _home: home,
+ controller,
+ fixture,
+ id,
+ mode,
+ };
+ result.settings("/empty", "/large", 1, 1).await;
+ result
+ }
+ async fn settings(&mut self, ping: &str, speed: &str, mib: u64, seconds: u64) {
+ let mut settings = self.controller.settings();
+ settings.ping_url = format!("{}{}", self.fixture.origin, ping);
+ settings.speed_url = format!("{}{}", self.fixture.origin, speed);
+ settings.maximum_download_mib = mib;
+ settings.maximum_duration_seconds = seconds;
+ self.controller
+ .execute(&Caller::root(), Operation::SaveSettings { settings })
+ .await
+ .unwrap();
+ }
+ fn requests(&self) -> Vec<serde_json::Value> {
+ std::fs::read_to_string(&self.fixture.requests_log)
+ .unwrap()
+ .lines()
+ .map(|line| serde_json::from_str(line).unwrap())
+ .collect()
+ }
+}
+
+#[tokio::test]
+async fn controller_ping_uses_the_selected_trojan_path_with_verified_tls_without_enabling() {
+ let mut context = Context::open().await;
+ let accepted_before = std::fs::read_to_string(&context.fixture.access_log)
+ .unwrap_or_default()
+ .lines()
+ .filter(|line| line.contains("accepted"))
+ .count();
+ let observation = context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .unwrap();
+ assert!(observation.success);
+ assert_eq!(observation.bytes, 0);
+ assert_eq!(observation.stop_reason, "complete");
+ assert_eq!(observation.configuration_id, context.id);
+ assert!(!context.controller.status().unwrap().requested_enabled);
+ assert_eq!(context.controller.status().unwrap().routing, "direct");
+ let accepted_after = std::fs::read_to_string(&context.fixture.access_log)
+ .unwrap()
+ .lines()
+ .filter(|line| line.contains("accepted"))
+ .count();
+ assert!(
+ accepted_after > accepted_before,
+ "The fixture Trojan endpoint must carry this diagnostic connection"
+ );
+ assert!(context.requests().iter().all(|request| {
+ request["user_agent"]
+ .as_str()
+ .is_some_and(|value| value.starts_with("Roentgen/"))
+ }));
+ assert!(
+ context
+ .requests()
+ .iter()
+ .all(|request| request["proxy_authorization"].is_null()),
+ "Local proxy credentials must not reach the target"
+ );
+}
+
+#[tokio::test]
+async fn controller_speed_stops_at_the_first_byte_or_duration_limit() {
+ let mut context = Context::open().await;
+ let bounded = context
+ .controller
+ .diagnostic(&Caller::root(), None, true)
+ .await
+ .unwrap();
+ assert!(bounded.success);
+ assert_eq!(bounded.bytes, 1_048_576);
+ assert_eq!(bounded.stop_reason, "byte-limit");
+ assert!(bounded.bytes_per_second.unwrap() > 0.0);
+ context.settings("/empty", "/slow", 1, 1).await;
+ let timed = context
+ .controller
+ .diagnostic(&Caller::root(), None, true)
+ .await
+ .unwrap();
+ assert!(timed.success);
+ assert_eq!(timed.bytes, 16384);
+ assert_eq!(timed.stop_reason, "duration-limit");
+ assert!(timed.elapsed_ms >= 900 && timed.elapsed_ms < 4000);
+ assert!(!context.controller.status().unwrap().requested_enabled);
+}
+
+async fn private_bus(controller: Controller) -> (tokio::process::Child, zbus::Connection, String) {
+ use tokio::io::{AsyncBufReadExt, BufReader};
+ let mut bus = tokio::process::Command::new("/bin/dbus-daemon")
+ .args(["--session", "--nofork", "--print-address=1"])
+ .stdout(Stdio::piped())
+ .stderr(Stdio::null())
+ .kill_on_drop(true)
+ .spawn()
+ .unwrap();
+ let mut address = String::new();
+ BufReader::new(bus.stdout.take().unwrap())
+ .read_line(&mut address)
+ .await
+ .unwrap();
+ let address = address.trim().to_owned();
+ let connection = zbus::connection::Builder::address(address.as_str())
+ .unwrap()
+ .name(roentgen::bus::NAME)
+ .unwrap()
+ .serve_at(
+ roentgen::bus::PATH,
+ roentgen::bus::Service {
+ controller: std::sync::Arc::new(tokio::sync::Mutex::new(controller)),
+ cancellation: std::sync::Arc::new(tokio::sync::Notify::new()),
+ },
+ )
+ .unwrap()
+ .build()
+ .await
+ .unwrap();
+ (bus, connection, address)
+}
+
+#[tokio::test]
+async fn cli_exits_unsuccessfully_when_the_provider_rate_limits_a_diagnostic() {
+ let mut context = Context::open().await;
+ context.settings("/rate", "/large", 1, 1).await;
+ let (_bus, _service, address) = private_bus(context.controller).await;
+ let binary = std::env::var("ROENTGEN_TEST_BINARY")
+ .expect("Set ROENTGEN_TEST_BINARY to the VM candidate executable");
+ let output = tokio::process::Command::new(binary)
+ .arg("ping")
+ .env("DBUS_SYSTEM_BUS_ADDRESS", address)
+ .stdin(Stdio::null())
+ .output()
+ .await
+ .unwrap();
+ assert!(
+ !output.status.success(),
+ "A failed diagnostic must not produce a successful CLI exit: {}",
+ String::from_utf8_lossy(&output.stdout)
+ );
+ let error = String::from_utf8_lossy(&output.stderr);
+ assert!(
+ error.contains("provider-rate-limit") && error.contains("60"),
+ "The failure should report the provider delay: {error}"
+ );
+}
+
+#[tokio::test]
+async fn controller_records_a_failed_check_when_a_previously_working_probe_cannot_start() {
+ let mut context = Context::open().await;
+ context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .unwrap();
+ assert_eq!(context.controller.status().unwrap().health, "reachable");
+ *context.mode.lock().unwrap() = ProbeMode::Unavailable;
+ assert!(
+ context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .is_err()
+ );
+ assert_eq!(context.controller.status().unwrap().health, "check-failed");
+ assert!(
+ context.controller.configurations(None).unwrap()[0]
+ .ping_ms
+ .is_none()
+ );
+ assert!(!context.controller.status().unwrap().requested_enabled);
+}
+
+#[tokio::test]
+async fn controller_rejects_redirects_oversized_ping_and_error_downloads() {
+ let mut context = Context::open().await;
+ context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .unwrap();
+ for (path, speed) in [
+ ("/html", true),
+ ("/partial", true),
+ ("/empty", true),
+ ("/oversize", false),
+ ("/redirect", false),
+ ("/error", true),
+ ("/json-error", true),
+ ] {
+ context.settings(path, path, 1, 1).await;
+ let before = context.requests();
+ assert!(
+ context
+ .controller
+ .diagnostic(&Caller::root(), None, speed)
+ .await
+ .is_err(),
+ "Diagnostic unexpectedly accepted {path}"
+ );
+ let after = context.requests();
+ assert!(
+ after[before.len()..]
+ .iter()
+ .any(|request| request["path"] == path),
+ "The response fixture was not reached for {path}"
+ );
+ if path == "/redirect" {
+ assert!(
+ !after[before.len()..]
+ .iter()
+ .any(|request| request["path"] == "/empty"),
+ "Redirects must not be followed"
+ );
+ }
+ assert!(!context.controller.status().unwrap().requested_enabled);
+ }
+}
+
+#[tokio::test]
+async fn controller_rejects_an_untrusted_target_or_local_proxy_without_direct_fallback() {
+ let mut context = Context::open().await;
+ context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .unwrap();
+ let before = context.requests().len();
+ let mut settings = context.controller.settings();
+ settings.ping_url = format!("{}/empty", context.fixture.bad_origin);
+ context
+ .controller
+ .execute(&Caller::root(), Operation::SaveSettings { settings })
+ .await
+ .unwrap();
+ assert!(
+ context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .is_err()
+ );
+ assert_eq!(
+ context.requests().len(),
+ before,
+ "An untrusted target must not receive an HTTP request"
+ );
+ context.settings("/empty", "/large", 1, 1).await;
+ *context.mode.lock().unwrap() = ProbeMode::Impostor;
+ assert!(
+ context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .is_err()
+ );
+ assert_eq!(
+ context.requests().len(),
+ before,
+ "An impostor proxy must not receive credentials or cause direct target fallback"
+ );
+ assert!(!context.controller.status().unwrap().requested_enabled);
+}
+
+#[tokio::test]
+async fn provider_retry_after_blocks_manual_ping_and_speed_for_both_header_forms() {
+ for path in ["/rate", "/rate-date"] {
+ let mut context = Context::open().await;
+ context.settings(path, "/large", 1, 1).await;
+ let observation = context
+ .controller
+ .diagnostic(&Caller::root(), None, false)
+ .await
+ .unwrap();
+ assert!(!observation.success);
+ assert_eq!(observation.stop_reason, "provider-rate-limit");
+ assert!(observation.retry_after_seconds.unwrap() >= 60);
+ let before = context.requests().len();
+ for speed in [false, true] {
+ let error = context
+ .controller
+ .diagnostic(&Caller::root(), None, speed)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("Retry-After"));
+ }
+ assert_eq!(
+ context.requests().len(),
+ before,
+ "Cooldown must prevent all further provider requests"
+ );
+ }
+}
+
+#[tokio::test]
+async fn scheduled_checks_use_only_selection_never_run_speed_and_do_not_overlap() {
+ let mut context = Context::open().await;
+ context
+ .controller
+ .execute(
+ &Caller::root(),
+ Operation::AddSubscription {
+ name: "Other".into(),
+ url: "https://subscription.invalid/other".into(),
+ },
+ )
+ .await
+ .unwrap();
+ let before = context.requests().len();
+ context.controller.scheduled_check().await.unwrap();
+ context.controller.scheduled_check().await.unwrap();
+ let after = context.requests();
+ assert_eq!(after.len(), before + 1, "Only one selected Ping is due");
+ assert_eq!(after.last().unwrap()["path"], "/empty");
+ let entries = context.controller.configurations(None).unwrap();
+ assert!(
+ entries
+ .iter()
+ .find(|entry| entry.id == context.id)
+ .unwrap()
+ .ping_ms
+ .is_some()
+ );
+ assert!(
+ entries
+ .iter()
+ .filter(|entry| entry.id != context.id)
+ .all(|entry| entry.ping_ms.is_none())
+ );
+ assert!(
+ entries
+ .iter()
+ .all(|entry| entry.speed_bytes_per_second.is_none())
+ );
+ assert!(!context.controller.status().unwrap().requested_enabled);
+}
+
+#[tokio::test]
+async fn cancelling_a_bus_diagnostic_stops_it_without_changing_proxy_intent() {
+ let mut context = Context::open().await;
+ context.settings("/empty", "/slow", 1, 10).await;
+ let before = context.requests().len();
+ let requests = context.fixture.requests_log.clone();
+ let (_bus, _service, address) = private_bus(context.controller).await;
+ let connection = zbus::connection::Builder::address(address.as_str())
+ .unwrap()
+ .build()
+ .await
+ .unwrap();
+ let request_connection = connection.clone();
+ let request = tokio::spawn(async move {
+ let proxy = zbus::Proxy::new(
+ &request_connection,
+ roentgen::bus::NAME,
+ roentgen::bus::PATH,
+ roentgen::bus::INTERFACE,
+ )
+ .await
+ .unwrap();
+ proxy
+ .call::<_, _, String>("Diagnostic", &(String::new(), true, false))
+ .await
+ });
+ let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
+ loop {
+ let count = std::fs::read_to_string(&requests).unwrap().lines().count();
+ if count > before {
+ break;
+ }
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "Diagnostic did not reach its fixture"
+ );
+ tokio::time::sleep(Duration::from_millis(20)).await;
+ }
+ let proxy = zbus::Proxy::new(
+ &connection,
+ roentgen::bus::NAME,
+ roentgen::bus::PATH,
+ roentgen::bus::INTERFACE,
+ )
+ .await
+ .unwrap();
+ let _: () = proxy.call("CancelDiagnostic", &(false,)).await.unwrap();
+ let result = tokio::time::timeout(Duration::from_secs(2), request)
+ .await
+ .unwrap()
+ .unwrap();
+ assert!(result.unwrap_err().to_string().contains("cancelled"));
+ let status: String = proxy.call("Status", &()).await.unwrap();
+ let status: roentgen::model::Status = serde_json::from_str(&status).unwrap();
+ assert!(!status.requested_enabled);
+ assert_eq!(status.selected_configuration, Some(context.id));
+ let rows: String = proxy
+ .call("Configurations", &(String::new(),))
+ .await
+ .unwrap();
+ let rows: Vec<roentgen::model::ConfigurationSummary> = serde_json::from_str(&rows).unwrap();
+ assert!(
+ rows.iter()
+ .all(|entry| entry.speed_bytes_per_second.is_none())
+ );
+}
generated by cgit v1.2.3 (git 2.25.1) at 2026-10-30 11:57:05 +0000