CH-J Toolsgestion de serveur sur SSH
Menu
Source publiée

CH-J Tools

Parcourez les répertoires et les fichiers pour une version spécifique de l'application.

Télécharger la source ZIP
CH-J / Server Manager 1704 fichiers
0.1.2-beta Ajouté: · Construit:
service.rs Ajouté: Date non enregistrée 22,7 KB · 489 lignes
Code Télécharger le fichier
1//! Diagnostics jobs (port of `diagnosticsService.js`): one running job, two tools at a time,
2//! at most two IP families, five jobs kept, 500 progress events, time limit 5 min (or the
3//! ping budget, max 30 min), cancellation of the whole job or of a single tool.
5use crate::core::diagnostics::common::{error_result, normalize_options, now_ms, parse_target, DiagError, Options, PING_DURATION_LIMIT_MS};
6use crate::core::diagnostics::history::{export_report, scrubbed, DiagnosticsHistory};
7use crate::core::diagnostics::http::HttpDiagnostics;
8use crate::core::diagnostics::{dns, process, socket};
9use parking_lot::Mutex;
10use serde_json::{json, Map, Value};
11use std::collections::HashMap;
12use std::net::IpAddr;
13use std::path::{Path, PathBuf};
14use std::sync::Arc;
15use std::time::Duration;
16use tokio_util::sync::CancellationToken;
18pub type ProgressSink = Arc<dyn Fn(Value) + Send + Sync>;
20struct Job {
21 report: Value,
22 cancel: CancellationToken,
23 tools: HashMap<String, CancellationToken>,
24 cancelled_tools: Vec<String>,
27pub struct DiagnosticsService {
28 pub history: DiagnosticsHistory,
29 http: HttpDiagnostics,
30 jobs: Mutex<Vec<(String, Arc<Mutex<Job>>)>>,
31 resolvers: Mutex<Vec<(u64, CancellationToken)>>,
32 resolver_seq: std::sync::atomic::AtomicU64,
33 sink: ProgressSink,
36struct ResolverGuard<'a> {
37 service: &'a DiagnosticsService,
38 ticket: u64,
39 cancel: CancellationToken,
42impl Drop for ResolverGuard<'_> {
43 fn drop(&mut self) {
44 self.cancel.cancel();
45 self.service.resolvers.lock().retain(|(ticket, _)| *ticket != self.ticket);
46 }
49pub fn job_time_limit(options: &Options) -> u64 {
50 if options.continuous {
51 return PING_DURATION_LIMIT_MS;
52 }
53 let ping = if options.tools.iter().any(|t| t == "ping") { options.count * options.repetitions * options.interval_ms + options.ping_timeout_ms.unwrap_or(options.timeout_ms) + 60_000 } else { 0 };
54 PING_DURATION_LIMIT_MS.min((5 * 60 * 1000u64).max(ping))
57fn target_for_address(options: &Options, address: &str) -> Result<Options, DiagError> {
58 let mut url = url::Url::parse(&options.target.url).map_err(|_| DiagError::new("INVALID_URL"))?;
59 let host = if address.parse::<IpAddr>().map(|ip| ip.is_ipv6()).unwrap_or(false) { format!("[{address}]") } else { address.to_string() };
60 url.set_host(Some(&host)).map_err(|_| DiagError::new("INVALID_URL"))?;
61 Ok(Options { target: parse_target(url.as_str())?, ..options.clone() })
64fn same_ip(a: &str, b: &str) -> bool {
65 match (a.parse::<IpAddr>(), b.parse::<IpAddr>()) {
66 (Ok(x), Ok(y)) => x == y,
67 _ => false,
68 }
71fn merge(mut base: Value, extra: &Value) -> Value {
72 if let (Some(target), Some(fields)) = (base.as_object_mut(), extra.as_object()) {
73 for (k, v) in fields {
74 target.insert(k.clone(), v.clone());
75 }
76 }
77 base
80impl DiagnosticsService {
81 pub fn new(storage_root: &Path, resources: PathBuf, sink: ProgressSink) -> Self {
82 Self { history: DiagnosticsHistory::new(Some(storage_root)), http: HttpDiagnostics { resources, ca_path: None }, jobs: Mutex::new(Vec::new()), resolvers: Mutex::new(Vec::new()), resolver_seq: Default::default(), sink }
83 }
85 fn reserve_resolver(&self) -> Result<ResolverGuard<'_>, DiagError> {
86 let mut resolvers = self.resolvers.lock();
87 if resolvers.len() >= 2 {
88 return Err(DiagError::new("DIAGNOSTICS_BUSY"));
89 }
90 let ticket = self.resolver_seq.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
91 let cancel = CancellationToken::new();
92 resolvers.push((ticket, cancel.clone()));
93 Ok(ResolverGuard { service: self, ticket, cancel })
94 }
96 pub async fn resolve(&self, source: &Value) -> Result<Value, DiagError> {
97 let mut input = source.clone();
98 input["tools"] = json!(["dns"]);
99 let options = normalize_options(&input)?;
100 let reservation = self.reserve_resolver()?;
101 let cancel = reservation.cancel.clone();
102 let result = async {
103 if let Some(manual) = &options.manual_addresses {
104 let mut target = options.target.to_json();
105 target["type"] = json!("ipv4+ipv6");
106 return Ok(json!({ "target": target, "addresses": manual.iter().map(|(a, f)| json!({ "address": a, "family": f })).collect::<Vec<_>>() }));
107 }
108 let addresses = tokio::time::timeout(Duration::from_secs(30), dns::resolve(&options.target.host, &options, &cancel)).await.map_err(|_| DiagError::new("TIMEOUT"))??;
109 Ok(json!({ "target": options.target.to_json(), "addresses": addresses }))
110 }
111 .await;
112 drop(reservation);
113 result
114 }
116 pub fn start(self: &Arc<Self>, source: &Value) -> Result<Value, DiagError> {
117 let options = normalize_options(source)?;
118 let mut id_bytes = [0u8; 16];
119 rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut id_bytes);
120 let id = hex::encode(id_bytes);
121 let report = json!({ "id": id, "createdAt": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true), "status": "running", "options": options.json, "results": {}, "progress": [], "addresses": [] });
122 let job = Arc::new(Mutex::new(Job { report: report.clone(), cancel: CancellationToken::new(), tools: HashMap::new(), cancelled_tools: Vec::new() }));
123 {
124 let mut jobs = self.jobs.lock();
125 if jobs.iter().any(|(_, job)| job.lock().report["status"] == "running") {
126 return Err(DiagError::new("DIAGNOSTICS_BUSY"));
127 }
128 jobs.push((id.clone(), job.clone()));
129 while jobs.len() > 5 {
130 jobs.remove(0);
131 }
132 }
133 let service = self.clone();
134 tokio::spawn(async move {
135 let outcome = service.run(&job, options).await;
136 if let Err(error) = outcome {
137 service.finish(&job, Some(error_result(&error)));
138 }
139 });
140 Ok(json!({ "id": id, "report": scrubbed(&report) }))
141 }
143 pub fn cancel(&self, id: &str, tool: Option<&str>) -> Result<Value, DiagError> {
144 let job = self.jobs.lock().iter().find(|(jid, _)| jid == id).map(|(_, j)| j.clone()).ok_or_else(|| DiagError::new("JOB_NOT_FOUND"))?;
145 let mut guard = job.lock();
146 match tool.filter(|t| !t.is_empty()) {
147 Some(tool) => {
148 if !guard.report["options"]["tools"].as_array().map(|t| t.iter().any(|x| x == tool)).unwrap_or(false) {
149 return Err(DiagError::new("INVALID_TOOLS"));
150 }
151 guard.cancelled_tools.push(tool.to_string());
152 if let Some(token) = guard.tools.get(tool) {
153 token.cancel();
154 }
155 }
156 None => guard.cancel.cancel(),
157 }
158 Ok(json!({ "cancelling": true }))
159 }
161 pub fn stop_all(&self) {
162 for (_, job) in self.jobs.lock().iter() {
163 let guard = job.lock();
164 if guard.report["status"] == "running" {
165 guard.cancel.cancel();
166 }
167 }
168 for (_, token) in self.resolvers.lock().iter() {
169 token.cancel();
170 }
171 }
173 pub fn get(&self, id: &str) -> Result<Value, DiagError> {
174 if let Some(job) = self.jobs.lock().iter().find(|(jid, _)| jid == id).map(|(_, j)| j.clone()) {
175 return Ok(scrubbed(&job.lock().report));
176 }
177 self.history.get(id)
178 }
180 pub fn export(&self, id: &str, format: &str) -> Result<String, DiagError> {
181 export_report(&self.get(id)?, format)
182 }
184 fn emit(&self, job: &Arc<Mutex<Job>>, event: Value) {
185 let mut guard = job.lock();
186 let id = guard.report["id"].clone();
187 let kind = event["kind"].as_str().unwrap_or("").to_string();
188 if kind != "snapshot" && kind != "complete" {
189 if let Some(progress) = guard.report["progress"].as_array_mut() {
190 progress.push(scrubbed(&event));
191 if progress.len() > 500 {
192 progress.remove(0);
193 }
194 }
195 }
196 drop(guard);
197 (self.sink)(merge(json!({ "jobId": id }), &event));
198 }
200 fn finish(&self, job: &Arc<Mutex<Job>>, failure: Option<Value>) {
201 let report = {
202 let mut guard = job.lock();
203 if guard.report["status"] != "running" {
204 return;
205 }
206 let aborted = guard.cancel.is_cancelled();
207 guard.report["status"] = json!(if aborted { "cancelled" } else if failure.is_some() { "failed" } else { "completed" });
208 guard.report["finishedAt"] = json!(chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true));
209 if let Some(failure) = failure {
210 guard.report["error"] = failure;
211 }
212 if let Some(reason) = guard.report.get("_abortReason").cloned() {
213 guard.report["error"] = reason;
214 if let Some(o) = guard.report.as_object_mut() {
215 o.remove("_abortReason");
216 }
217 }
218 guard.report.clone()
219 };
220 if let Err(error) = self.history.add(&report) {
221 job.lock().report["historyError"] = json!(error.code);
222 }
223 let snapshot = scrubbed(&job.lock().report);
224 self.emit(job, json!({ "kind": "complete", "report": snapshot }));
225 }
227 async fn address_run(&self, job: &Arc<Mutex<Job>>, tool: &str, item: &Value, options: &Options, cancel: &CancellationToken, progress: ProgressSink) -> Value {
228 let address = item["address"].as_str().unwrap_or("").to_string();
229 let options = if options.manual_addresses.is_some() { target_for_address(options, &address).unwrap_or_else(|_| options.clone()) } else { options.clone() };
230 let outcome: Result<Value, DiagError> = async {
231 if cancel.is_cancelled() {
232 return Err(DiagError::new("CANCELLED"));
233 }
234 let mut result = match tool {
235 "http" => self.http.run(&options.target, &address, &options, cancel, progress.clone()).await?,
236 "compression" => self.http.compression(&options.target, &address, &options, cancel, progress.clone()).await?,
237 "tls" => socket::run_tls(&options.target, &address, &options, cancel, progress.clone(), None).await?,
238 "websocket" => socket::websocket_probe(&options.target, &address, &options, cancel, None).await?,
239 "ping" => process::ping(&address, &options, cancel, progress.clone()).await?,
240 "traceroute" => process::traceroute(&address, &options, cancel, progress.clone()).await?,
241 "tcp" => socket::run_tcp(&address, &options, cancel, progress.clone()).await?,
242 _ => return Err(DiagError::new("INVALID_TOOLS")),
243 };
244 if tool == "traceroute" {
245 let mut hops: Vec<String> = Vec::new();
246 for route in result["routes"].as_array().cloned().unwrap_or_default() {
247 for hop in route["hops"].as_array().cloned().unwrap_or_default() {
248 for a in hop["addresses"].as_array().cloned().unwrap_or_default() {
249 if let Some(a) = a.as_str() {
250 if !hops.contains(&a.to_string()) {
251 hops.push(a.to_string());
252 }
253 }
254 }
255 }
256 }
257 hops.truncate(options.max_hops as usize);
258 let mut reverse = Map::new();
259 for hop in hops {
260 if cancel.is_cancelled() {
261 return Err(DiagError::new("CANCELLED"));
262 }
263 let names = match dns::reverse_name(&hop) {
264 Ok(name) => match dns::query(&name, "PTR", &options, cancel).await {
265 Ok(answer) => answer["records"].as_array().cloned().unwrap_or_default().into_iter().filter(|r| r["type"] == "PTR").map(|r| r["value"].clone()).collect(),
266 Err(error) => {
267 if cancel.is_cancelled() {
268 return Err(error);
269 }
270 Vec::new()
271 }
272 },
273 Err(_) => Vec::new(),
274 };
275 reverse.insert(hop, Value::Array(names));
276 }
277 result["reverseDns"] = Value::Object(reverse);
278 }
279 Ok(result)
280 }
281 .await;
282 let base = json!({ "address": address, "family": item["family"] });
283 match outcome {
284 Ok(result) => merge(base, &result),
285 Err(error) => {
286 let mut partial = merge(base, &error_result(&error));
287 let report = job.lock().report.clone();
288 let progress_list = report["progress"].as_array().cloned().unwrap_or_default();
289 match tool {
290 "ping" => {
291 if let Some(latest) = report["livePing"].get(&address) {
292 for key in ["sent", "received", "lossPercent", "statistics"] {
293 partial[key] = latest[key].clone();
294 }
295 }
296 partial["packets"] = Value::Array(progress_list.into_iter().filter(|e| e["kind"] == "ping-packet" && e["address"] == address.as_str()).collect());
297 }
298 "traceroute" => {
299 partial["routes"] = json!([{ "hops": progress_list.into_iter().filter(|e| e["kind"] == "trace-hop" && e["address"] == address.as_str()).collect::<Vec<_>>() }]);
300 }
301 "http" => {
302 partial["comparisons"] = Value::Array(
303 options
304 .protocols
305 .iter()
306 .map(|p| json!({ "protocol": p, "address": address, "samples": progress_list.iter().filter(|e| e["kind"] == "http-sample" && e["address"] == address.as_str() && e["protocol"] == p.as_str()).cloned().collect::<Vec<_>>() }))
307 .collect(),
308 );
309 }
310 _ => {}
311 }
312 partial
313 }
314 }
315 }
317 async fn run(self: &Arc<Self>, job: &Arc<Mutex<Job>>, options: Options) -> Result<(), DiagError> {
318 let cancel = job.lock().cancel.clone();
319 let deadline_cancel = cancel.clone();
320 let deadline_job = job.clone();
321 let limit = job_time_limit(&options);
322 let deadline = tokio::spawn(async move {
323 tokio::time::sleep(Duration::from_millis(limit)).await;
324 deadline_job.lock().report["_abortReason"] = error_result(&DiagError::new("JOB_TIME_LIMIT"));
325 deadline_cancel.cancel();
326 });
327 let outcome = async {
328 if cancel.is_cancelled() {
329 return Err(DiagError::new("CANCELLED"));
330 }
331 let started = now_ms();
332 let mut addresses: Vec<Value> = match &options.manual_addresses {
333 Some(manual) => manual.iter().map(|(a, f)| json!({ "address": a, "family": f, "source": "manual" })).collect(),
334 None => match dns::resolve(&options.target.host, &options, &cancel).await {
335 Ok(list) => list,
336 Err(error) => {
337 if cancel.is_cancelled() {
338 return Err(error);
339 }
340 job.lock().report["resolutionError"] = error_result(&error);
341 Vec::new()
342 }
343 },
344 };
345 let resolved = addresses.clone();
346 let resolution_ms = now_ms() - started;
347 if !options.selected_ip.is_empty() {
348 addresses.retain(|a| same_ip(a["address"].as_str().unwrap_or(""), &options.selected_ip));
349 if addresses.is_empty() {
350 return Err(DiagError::new("SELECTED_IP_NOT_RESOLVED"));
351 }
352 }
353 let family = |a: &Value| a["family"].as_u64().unwrap_or(0);
354 match options.mode.as_str() {
355 "ipv4" => addresses.retain(|a| family(a) == 4),
356 "ipv6" => addresses.retain(|a| family(a) == 6),
357 "auto" if options.manual_addresses.is_none() => {
358 addresses = addresses.iter().find(|a| family(a) == 4).or(addresses.first()).cloned().into_iter().collect();
359 }
360 "both" => {
361 let v4 = addresses.iter().find(|a| family(a) == 4).cloned();
362 let v6 = addresses.iter().find(|a| family(a) == 6).cloned();
363 addresses = v4.into_iter().chain(v6).collect();
364 }
365 _ => {}
366 }
367 let resolution_error = {
368 let mut guard = job.lock();
369 guard.report["addresses"] = json!(addresses);
370 guard.report["resolvedAddresses"] = json!(resolved);
371 guard.report["resolutionMs"] = json!(resolution_ms);
372 guard.report["resolutionError"].clone()
373 };
374 self.emit(job, json!({ "kind": "resolved", "addresses": addresses, "resolvedAddresses": resolved, "resolutionMs": resolution_ms, "error": resolution_error }));
375 let queue = Arc::new(Mutex::new(std::collections::VecDeque::from(options.tools.clone())));
376 let workers = (0..options.tools.len().min(2)).map(|_| {
377 let queue = queue.clone();
378 let service = self.clone();
379 let job = job.clone();
380 let options = options.clone();
381 let cancel = cancel.clone();
382 let addresses = addresses.clone();
383 async move {
384 loop {
385 if cancel.is_cancelled() {
386 break;
387 }
388 let Some(tool) = queue.lock().pop_front() else { break };
389 let sub = cancel.child_token();
390 {
391 let mut guard = job.lock();
392 if guard.cancelled_tools.contains(&tool) {
393 sub.cancel();
394 }
395 guard.tools.insert(tool.clone(), sub.clone());
396 }
397 service.emit(&job, json!({ "kind": "tool-start", "tool": tool }));
398 let progress_job = job.clone();
399 let progress_service = service.clone();
400 let progress_tool = tool.clone();
401 let progress: ProgressSink = Arc::new(move |event: Value| {
402 if event["kind"] == "ping-packet" {
403 let address = event["address"].as_str().unwrap_or("").to_string();
404 let mut guard = progress_job.lock();
405 if !guard.report["livePing"].is_object() {
406 guard.report["livePing"] = json!({});
407 }
408 guard.report["livePing"][address] = event.clone();
409 }
410 progress_service.emit(&progress_job, merge(json!({ "tool": progress_tool }), &event));
411 });
412 let result = if sub.is_cancelled() {
413 json!([error_result(&DiagError::new("CANCELLED"))])
414 } else if tool == "dns" {
415 match dns::run(&options.target.host, &options, &sub, &*progress).await {
416 Ok(result) => result,
417 Err(error) => json!([error_result(&error)]),
418 }
419 } else if addresses.is_empty() {
420 json!([{ "status": "not-tested", "code": "NO_ADDRESS" }])
421 } else {
422 let futures = addresses.iter().map(|item| service.address_run(&job, &tool, item, &options, &sub, progress.clone()));
423 Value::Array(futures::future::join_all(futures).await)
424 };
425 job.lock().report["results"][&tool] = result.clone();
426 service.emit(&job, json!({ "kind": "tool-complete", "tool": tool, "result": scrubbed(&result) }));
427 job.lock().tools.remove(&tool);
428 }
429 }
430 });
431 futures::future::join_all(workers).await;
432 Ok(())
433 }
434 .await;
435 deadline.abort();
436 match outcome {
437 Ok(()) => self.finish(job, None),
438 Err(error) => self.finish(job, Some(error_result(&error))),
439 }
440 Ok(())
441 }
444#[cfg(test)]
445mod tests {
446 use super::*;
447 use std::sync::Barrier;
449 #[test]
450 fn concurrent_resolvers_reserve_two_slots_and_release_on_drop() {
451 let root = tempfile::tempdir().unwrap();
452 let service = DiagnosticsService::new(root.path(), PathBuf::new(), Arc::new(|_| {}));
453 let barrier = Barrier::new(16);
454 std::thread::scope(|scope| {
455 let handles: Vec<_> = (0..16).map(|_| scope.spawn(|| {
456 let reservation = service.reserve_resolver();
457 barrier.wait();
458 let accepted = reservation.is_ok();
459 barrier.wait();
460 accepted
461 })).collect();
462 assert_eq!(handles.into_iter().map(|h| h.join().unwrap()).filter(|accepted| *accepted).count(), 2);
463 });
464 assert!(service.resolvers.lock().is_empty());
465 assert!(service.reserve_resolver().is_ok());
466 }
468 #[tokio::test]
469 async fn concurrent_starts_reserve_one_job() {
470 // The current-thread runtime does not run spawned jobs until all starters return.
471 let root = tempfile::tempdir().unwrap();
472 let service = Arc::new(DiagnosticsService::new(root.path(), PathBuf::new(), Arc::new(|_| {})));
473 let runtime = tokio::runtime::Handle::current();
474 let barrier = Barrier::new(16);
475 std::thread::scope(|scope| {
476 let handles: Vec<_> = (0..16).map(|_| scope.spawn(|| {
477 let _entered = runtime.enter();
478 barrier.wait();
479 service.start(&json!({ "target": "example.com", "tools": ["dns"] })).is_ok()
480 })).collect();
481 assert_eq!(handles.into_iter().map(|h| h.join().unwrap()).filter(|accepted| *accepted).count(), 1);
482 });
483 service.stop_all();
484 }
487#[cfg(test)]
488#[path = "../../../tests/support/diagnostics_safety.rs"]
489mod safety_tests;
SHA-256690329e516628585272e2cf9f59a4d2e4e6736bcf6a71b9c6c3e16ce6402bac5

CH-J Proprietary Software License 1.14 · La source est fournie sous CH-J Proprietary Software License 1.14. Sa disponibilité ne modifie pas les termes de la licence ni n'accorde de droits supplémentaires.

Archive SHA-256ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f
Fichiers