Fuente publicada

CH-J Tools

Busque directorios y archivos para obtener una versión de aplicación específica.

Descargar fuente ZIP

CH-J / Server Manager 1704 archivos

0.1.2-beta Añadido: 8 nov 2026, 2:53 · Construido: 7 nov 2026, 19:32

monitor.rs Añadido: Fecha no registrada 37,9 KB · 881 líneas

Código Descargar archivo

1//! Small TCP availability monitor. Consumers must show unknown/stale after `validUntil`: the completed round's

2//! `updatedAt` plus three intervals, but at least `MIN_VALIDITY_SECS`, because macOS redraws widgets only every few

3//! minutes; a shorter validity made a running helper look stale between redraws (2026-11-06).

4use serde_json::{json, Value};

5use std::collections::{hash_map::DefaultHasher, BTreeMap};

6use std::fs::{self, OpenOptions};

7use std::hash::{Hash, Hasher};

8use std::io::{self, Read, Write};

9use std::net::{SocketAddr, TcpStream, ToSocketAddrs};

10use std::path::{Path, PathBuf};

11use std::sync::atomic::{AtomicUsize, Ordering};

12use std::sync::mpsc;

13use std::thread;

14use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

16const INPUT: &str = "standby-servers.json";

17const OUTPUT: &str = "standby-status.json";

18const MIN_VALIDITY_SECS: i64 = 15 * 60;

19/// Check types the app may request; only `ssh` is verified beyond the TCP connect (banner), the rest mean "port open".

20const PROTOCOLS: [&str; 8] = ["ssh", "https", "http", "smb", "ftp", "dns", "rdp", "tcp"];

21const WORKERS: usize = 6;

22const STACK_SIZE: usize = 128 * 1024;

23const PROBE_LIMIT: Duration = Duration::from_secs(6);

24static RESOLVERS: AtomicUsize = AtomicUsize::new(0);

25static TEMP_ID: AtomicUsize = AtomicUsize::new(0);

27#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]

28struct Endpoint {

29 id: String,

30 host: String,

31 port: u16,

32 protocol: String,

33}

35impl Endpoint {

36 // A non-secret identity token lets one-shot history match hidden addresses without

37 // copying them into the status file. This is not an authentication/security hash.

38 fn token(&self) -> String {

39 let mut hash = DefaultHasher::new();

40 self.hash(&mut hash);

41 format!("{:016x}", hash.finish())

42 }

43}

45#[derive(Clone, Debug)]

46struct Server {

47 endpoint: Endpoint,

48 name: String,

49}

51#[derive(Default)]

52struct Config {

53 enabled: bool,

54 interval: u64,

55 show_addresses: bool,

56 servers: Vec<Server>,

57}

59fn same_config(a: &Config, b: &Config) -> bool {

60 a.enabled == b.enabled

61 && a.interval == b.interval

62 && a.show_addresses == b.show_addresses

63 && a.servers.len() == b.servers.len()

64 && a.servers.iter().zip(&b.servers).all(|(x, y)| x.endpoint == y.endpoint && x.name == y.name)

65}

67/// Replaces both status copies with the "off" document (helper switched off: no names or addresses may stay).

68pub fn publish_off(data: &Path) {

69 let value = json!({"version": 1, "enabled": false, "updatedAt": timestamp(now()), "servers": []});

70 for path in extra_status_paths() {

71 let _ = write_status(&path, &value);

72 }

73 let _ = write_status(&data.join(OUTPUT), &value);

74}

76fn parse_config(value: &Value) -> Config {

77 let interval = match value["intervalSecs"].as_u64() {

78 Some(n @ (30 | 60 | 300)) => n,

79 _ => 60,

80 };

81 let mut config = Config {

82 enabled: value["version"] == 1 && value["enabled"] == true,

83 interval,

84 show_addresses: value["showAddresses"] == true,

85 servers: Vec::new(),

86 };

87 if !config.enabled {

88 return config;

89 }

90 for entry in value["servers"].as_array().into_iter().flatten().take(200) {

91 let parse = || {

92 let id = entry["id"].as_str()?;

93 let name = entry["name"].as_str()?;

94 let host = entry["host"].as_str()?;

95 let port = entry["port"].as_u64().filter(|p| (1..=65535).contains(p))? as u16;

96 let protocol = entry["protocol"].as_str().filter(|p| PROTOCOLS.contains(p))?;

97 if id.is_empty() || id.chars().count() > 64 || name.is_empty() || name.chars().count() > 128

98 || host.is_empty() || host.chars().count() > 253

99 || host.chars().any(|c| c.is_whitespace() || c.is_control() || c == '/')

100 {

101 return None;

102 }

103 Some(Server {

104 endpoint: Endpoint { id: id.into(), host: host.into(), port, protocol: protocol.into() },

105 name: name.into(),

106 })

107 };

108 if let Some(server) = parse() {

109 config.servers.push(server);

110 }

111 }

112 config

113}

114

115fn read_json(path: &Path) -> Option<Value> {

116 // Bounded read: the app writes at most 200 short entries; a huge file must not blow the RAM budget.

117 serde_json::from_reader(io::BufReader::new(io::Read::take(fs::File::open(path).ok()?, 1 << 20))).ok()

118}

119

120fn load(data: &Path) -> Config {

121 parse_config(&read_json(&data.join(INPUT)).unwrap_or(Value::Null))

122}

123

124#[derive(Clone, Debug)]

125struct History {

126 status: String,

127 service: Option<String>,

128 latency: Option<u64>,

129 failed: bool,

130}

131

132/// In-memory history for consecutive failures. A fresh state seeds its first round

133/// from the previous status file, so separate `monitor` commands also debounce failures.

134#[derive(Default)]

135pub struct State {

136 history: BTreeMap<Endpoint, History>,

137 initialized: bool,

138 /// Widget redraw requests: last request time and the statuses it showed (macOS limits redraws per day).

139 widget_reload: Option<(std::time::Instant, String)>,

140}

141

142/// Redraw again without a status change at most this often.

143const WIDGET_RELOAD_EVERY: std::time::Duration = std::time::Duration::from_secs(300);

144

145/// Asks WidgetKit to redraw after new data reached the widget: right away when a status changed, otherwise at most

146/// every `WIDGET_RELOAD_EVERY`. The request must come from the containing app, so the helper starts

147/// `chj-widget-reload` from the app's `Contents/MacOS` (detached; macOS decides when it actually redraws).

148fn request_widget_reload(state: &mut State, value: &Value) {

149

    let statuses: String = value["servers"].as_array().into_iter().flatten().map(|s| format!("{}={};", s["id"], s["status"])).collect();

150 let due = match &state.widget_reload {

151 Some((at, last)) => *last != statuses || at.elapsed() >= WIDGET_RELOAD_EVERY,

152 None => true,

153 };

154 if !due {

155 return;

156 }

157 let tool = crate::app_binary().with_file_name("chj-widget-reload");

158 if tool.is_file() {

159

        if let Ok(mut child) = std::process::Command::new(&tool).stdin(std::process::Stdio::null()).stdout(std::process::Stdio::null()).stderr(std::process::Stdio::null()).spawn() {

160 std::thread::spawn(move || {

161 let _ = child.wait();

162 });

163 }

164 }

165 state.widget_reload = Some((std::time::Instant::now(), statuses));

166}

167

168fn previous(data: &Path, servers: &[Server]) -> BTreeMap<Endpoint, History> {

169 let value = read_json(&data.join(OUTPUT)).unwrap_or(Value::Null);

170 let mut history = BTreeMap::new();

171 if value["version"] != 1 || value["enabled"] != true { return history; }

172 for server in servers {

173 let key = &server.endpoint;

174 let old = value["servers"].as_array().into_iter().flatten().find(|v| {

175 v["id"] == key.id && v["port"] == key.port && v["protocol"] == key.protocol

176 && match v["host"].as_str() {

177 Some(host) => host == key.host,

178 None => v["endpointKey"].as_str() == Some(key.token().as_str()),

179 }

180 });

181 if let Some(old) = old {

182

            let status = old["status"].as_str().filter(|s| matches!(*s, "online" | "offline" | "unknown")).unwrap_or("unknown");

183 let service = if status == "online" {

184 Some(if old["service"] == "ssh" { "ssh" } else { "unverified" }.into())

185 } else { None };

186 history.insert(key.clone(), History {

187 status: status.into(), service, latency: old["latencyMs"].as_u64(),

188 failed: is_reason(&old["reason"]) || is_reason(&old["failureReason"]),

189 });

190 }

191 }

192 history

193}

194

195fn is_reason(value: &Value) -> bool {

196

    matches!(value.as_str(), Some("DNS_FAILED" | "DNS_TIMEOUT" | "CONNECTION_REFUSED" | "TIMEOUT" | "NETWORK_UNREACHABLE" | "OTHER"))

197}

198

199struct ResolverSlot;

200impl Drop for ResolverSlot {

201 fn drop(&mut self) { RESOLVERS.fetch_sub(1, Ordering::Relaxed); }

202}

203

204fn resolve(endpoint: &Endpoint, deadline: Instant) -> Result<Vec<SocketAddr>, &'static str> {

205 // std cannot cancel a blocked system resolver. Cap even abandoned DNS threads

206 // across rounds, so an unhealthy resolver cannot grow the helper indefinitely.

207 let dns_deadline = deadline.min(Instant::now() + Duration::from_secs(3));

208 loop {

209 let active = RESOLVERS.load(Ordering::Relaxed);

210

        if active < WORKERS && RESOLVERS.compare_exchange_weak(active, active + 1, Ordering::Relaxed, Ordering::Relaxed).is_ok() {

211 break;

212 }

213 let left = dns_deadline.saturating_duration_since(Instant::now());

214 if left.is_zero() { return Err("DNS_TIMEOUT"); }

215 thread::sleep(left.min(Duration::from_millis(5)));

216 }

217 let slot = ResolverSlot;

218 let host = endpoint.host.clone();

219 let port = endpoint.port;

220 let (send, receive) = mpsc::sync_channel(1);

221 thread::Builder::new().name("monitor-dns".into()).stack_size(STACK_SIZE).spawn(move || {

222 let _slot = slot;

223 let addresses = (host.as_str(), port).to_socket_addrs()

224 .map(|addresses| addresses.take(3).collect::<Vec<_>>());

225 let _ = send.send(addresses);

226 }).map_err(|_| "OTHER")?;

227 match receive.recv_timeout(dns_deadline.saturating_duration_since(Instant::now())) {

228 Ok(Ok(addresses)) if !addresses.is_empty() => Ok(addresses),

229 Ok(_) => Err("DNS_FAILED"),

230 Err(mpsc::RecvTimeoutError::Timeout) => Err("DNS_TIMEOUT"),

231 Err(_) => Err("OTHER"),

232 }

233}

234

235fn connection_reason(error: &io::Error) -> &'static str {

236 match error.kind() {

237 io::ErrorKind::ConnectionRefused => "CONNECTION_REFUSED",

238 io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "TIMEOUT",

239 io::ErrorKind::NetworkUnreachable | io::ErrorKind::HostUnreachable => "NETWORK_UNREACHABLE",

240 _ => "OTHER",

241 }

242}

243

244fn ssh_banner(stream: &mut TcpStream, deadline: Instant) -> bool {

245 let deadline = deadline.min(Instant::now() + Duration::from_millis(1500));

246 let mut bytes = [0; 255];

247 let mut used = 0;

248 while used < bytes.len() {

249 let left = deadline.saturating_duration_since(Instant::now());

250 if left.is_zero() || stream.set_read_timeout(Some(left)).is_err() { break; }

251 match stream.read(&mut bytes[used..]) {

252 Ok(0) => break,

253 Ok(n) => {

254 used += n;

255 if used >= 4 { return bytes[..used].starts_with(b"SSH-"); }

256 if !b"SSH-".starts_with(&bytes[..used]) { break; }

257 }

258 Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,

259 Err(_) => break,

260 }

261 }

262 false

263}

264

265type Probe = Result<(&'static str, u64), &'static str>;

266

267fn probe(endpoint: &Endpoint) -> Probe {

268 let deadline = Instant::now() + PROBE_LIMIT;

269 let addresses = resolve(endpoint, deadline)?;

270 let mut reason = "OTHER";

271 for address in addresses {

272 let left = deadline.saturating_duration_since(Instant::now());

273 if left.is_zero() { return Err("TIMEOUT"); }

274 let start = Instant::now();

275 match TcpStream::connect_timeout(&address, left.min(Duration::from_millis(1500))) {

276 Ok(mut stream) => {

277 let latency = start.elapsed().as_millis() as u64;

278 let service = if endpoint.protocol == "ssh" && ssh_banner(&mut stream, deadline) {

279 "ssh"

280 } else {

281 // TODO: verify TLS for https; a TCP connection alone proves no service.

282 "unverified"

283 };

284 return Ok((service, latency));

285 }

286 Err(error) => reason = connection_reason(&error),

287 }

288 }

289 Err(reason)

290}

291

292fn probes(servers: &[Server]) -> Vec<Probe> {

293 let mut results = Vec::with_capacity(servers.len());

294 // Batches keep concurrency bounded without a queue, runtime, or persistent pool.

295 for batch in servers.chunks(WORKERS) {

296 thread::scope(|scope| {

297 let jobs: Vec<_> = batch.iter().map(|server| {

298 thread::Builder::new().name("monitor-probe".into()).stack_size(STACK_SIZE)

299 .spawn_scoped(scope, || probe(&server.endpoint))

300 }).collect();

301 for job in jobs {

302 results.push(match job {

303 Ok(job) => job.join().unwrap_or(Err("OTHER")),

304 Err(_) => Err("OTHER"),

305 });

306 }

307 });

308 }

309 results

310}

311

312/// RFC 3339 UTC, using the Gregorian civil-from-days algorithm (including century rules).

313fn timestamp(seconds: i64) -> String {

314 let days = seconds.div_euclid(86400);

315 let time = seconds.rem_euclid(86400);

316 let z = days + 719468;

317 let era = z.div_euclid(146097);

318 let doe = z - era * 146097;

319 let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;

320 let mut year = yoe + era * 400;

321 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);

322 let mp = (5 * doy + 2) / 153;

323 let day = doy - (153 * mp + 2) / 5 + 1;

324 let month = mp + if mp < 10 { 3 } else { -9 };

325 year += i64::from(month <= 2);

326 format!("{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z", time / 3600, time / 60 % 60, time % 60)

327}

328

329fn now() -> i64 {

330 SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs() as i64

331}

332

333/// Future platform copies (e.g. macOS App Group or Windows widget data).

334/// macOS App Group shared with the WidgetKit extension (`macos-widget/`, `Contents/PlugIns/CH-J-StatusWidget.appex`).

335/// The release signs the helper with the same `application-groups` entitlement, so writing needs no privacy prompt.

336pub const MAC_APP_GROUP: &str = "2943B9C4Y5.de.ch-j.standby";

337

338/// Copies of the status for platform widgets. Only the signed macOS release helper (inside

339/// `CH-J-Standby-Helper.app`) writes into the App Group; an unsigned development helper would trigger macOS's

340/// "access data from other apps" prompt.

341pub fn extra_status_paths() -> Vec<PathBuf> {

342 #[cfg(target_os = "macos")]

343 {

344

        let in_bundle = std::env::current_exe().ok().is_some_and(|exe| exe.to_string_lossy().contains("/CH-J-Standby-Helper.app/Contents/MacOS/"));

345 // The container must come from Foundation: it checks the entitlement and grants access; writing to the path

346 // `~/Library/Group Containers/<group>` directly is refused with PermissionDenied (seen 2026-11-06).

347 if in_bundle {

348 let manager = objc2_foundation::NSFileManager::defaultManager();

349 let group = objc2_foundation::NSString::from_str(MAC_APP_GROUP);

350

            if let Some(path) = manager.containerURLForSecurityApplicationGroupIdentifier(&group).and_then(|url| url.path()) {

351 return vec![PathBuf::from(path.to_string()).join(OUTPUT)];

352 }

353 }

354 }

355 Vec::new()

356}

357

358fn write_status(path: &Path, value: &Value) -> io::Result<()> {

359 let parent = path.parent().ok_or_else(|| io::Error::from(io::ErrorKind::InvalidInput))?;

360 fs::create_dir_all(parent)?;

361 // create_new prevents following a pre-existing temp symlink and gives concurrent

362 // daemon / one-shot writers distinct files in the destination filesystem.

363 let (temporary, mut file) = loop {

364 let id = TEMP_ID.fetch_add(1, Ordering::Relaxed);

365 let temporary = parent.join(format!(".standby-status-{}-{id}.tmp", std::process::id()));

366 let mut options = OpenOptions::new();

367 options.write(true).create_new(true);

368 #[cfg(unix)]

369 {

370 use std::os::unix::fs::OpenOptionsExt;

371 options.mode(0o600);

372 }

373 match options.open(&temporary) {

374 Ok(file) => break (temporary, file),

375 Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue,

376 Err(e) => return Err(e),

377 }

378 };

379 let result = (|| {

380 #[cfg(unix)]

381 {

382 use std::os::unix::fs::PermissionsExt;

383 file.set_permissions(fs::Permissions::from_mode(0o600))?;

384 }

385 serde_json::to_writer(&mut file, value)?;

386 file.write_all(b"\n")?;

387 file.sync_all()?;

388 drop(file);

389 fs::rename(&temporary, path)

390 })();

391 if result.is_err() { let _ = fs::remove_file(&temporary); }

392 result

393}

394

395/// Probe once, persist atomically, and return the same status JSON. Write failures

396/// are best effort; consumers of the last successful file still expire it normally.

397pub fn round(data: &Path, state: &mut State) -> Value {

398 let config = load(data);

399 let value = if !config.enabled {

400 state.history.clear();

401 state.initialized = true;

402 json!({"version": 1, "enabled": false, "updatedAt": timestamp(now()), "servers": []})

403 } else {

404 if !state.initialized {

405 state.history = previous(data, &config.servers);

406 state.initialized = true;

407 }

408 let results = probes(&config.servers);

409 let stamp = now();

410 let mut next = BTreeMap::new();

411 let mut servers = Vec::with_capacity(config.servers.len());

412 for (server, result) in config.servers.iter().zip(results) {

413 let key = &server.endpoint;

414 let old = state.history.get(key);

415 let (history, reason) = match result {

416 Ok((service, latency)) => (History {

417 status: "online".into(), service: Some(service.into()), latency: Some(latency), failed: false,

418 }, None),

419 Err(reason) => {

420 let offline = old.is_some_and(|h| h.failed);

421 let status = if offline { "offline" } else { old.map(|h| h.status.as_str()).unwrap_or("unknown") };

422 (History {

423 status: status.into(),

424 service: if status == "online" { old.and_then(|h| h.service.clone()) } else { None },

425 latency: if status == "online" { old.and_then(|h| h.latency) } else { None },

426 failed: true,

427 }, Some(reason))

428 }

429 };

430 let mut entry = json!({

431 "id": key.id, "name": server.name, "port": key.port, "protocol": key.protocol,

432 "status": history.status, "service": history.service, "latencyMs": history.latency,

433 "checkedAt": timestamp(stamp),

434 "reason": if history.status == "online" { None } else { reason },

435 });

436 if config.show_addresses {

437 entry["host"] = json!(key.host);

438 } else {

439 entry["endpointKey"] = json!(key.token());

440 }

441 // Preserve reason:null for a retained online status, but record the first

442 // failure so a later one-shot invocation can recognize consecutive failures.

443 if history.status == "online" && history.failed {

444 entry["failureReason"] = json!(reason);

445 }

446 next.insert(key.clone(), history);

447 servers.push(entry);

448 }

449 state.history = next;

450 json!({

451 "version": 1, "enabled": true, "updatedAt": timestamp(stamp),

452

            "validUntil": timestamp(stamp + (3 * config.interval as i64).max(MIN_VALIDITY_SECS)), "intervalSecs": config.interval,

453 "helper": format!("ch-j-standby-helper {}", crate::VERSION), "servers": servers,

454 })

455 };

456 // Widget copies first, so the app-side status file records whether they worked (the App Group container is

457 // not readable for diagnostics from outside). Only the error kind is stored, never the path.

458 // The configuration may have changed while the probes ran (monitoring off, addresses hidden, servers removed):

459 // then this result is obsolete and must not overwrite what the newer configuration publishes.

460 if config.enabled && !same_config(&config, &load(data)) {

461 return value;

462 }

463 let copies: Vec<Value> = extra_status_paths()

464 .iter()

465 .map(|path| match write_status(path, &value) {

466 Ok(()) => json!({"ok": true}),

467 Err(error) => json!({"ok": false, "error": format!("{:?}", error.kind())}),

468 })

469 .collect();

470 let mut local = value.clone();

471 if !copies.is_empty() {

472 local["widgetCopy"] = json!(copies);

473 }

474 let _ = write_status(&data.join(OUTPUT), &local);

475 if copies.iter().any(|c| c["ok"] == true) {

476 request_widget_reload(state, &local);

477 }

478 local

479}

480

481fn revision(data: &Path) -> Option<(SystemTime, u64)> {

482 let metadata = fs::metadata(data.join(INPUT)).ok()?;

483 Some((metadata.modified().ok()?, metadata.len()))

484}

485

486/// Endless daemon loop. Disabled configurations publish once, then only watch for

487/// changes; active configurations also wake on their interval. No network while off.

488pub fn run_loop(data: &Path) {

489 // The daemon uses only its own round history; one-shot commands seed from disk.

490 let mut state = State { initialized: true, ..State::default() };

491 loop {

492 let before = revision(data);

493 let start = Instant::now();

494 let status = round(data, &mut state);

495 let deadline = status["intervalSecs"].as_u64().map(|n| start + Duration::from_secs(n));

496 loop {

497 let sleep = deadline.map(|d| d.saturating_duration_since(Instant::now()).min(Duration::from_secs(5)))

498 .unwrap_or(Duration::from_secs(5));

499 if sleep.is_zero() { break; }

500 thread::sleep(sleep);

501 if revision(data) != before || deadline.is_some_and(|d| Instant::now() >= d) { break; }

502 }

503 }

504}

505

506#[cfg(test)]

507mod tests {

508 use super::*;

509 use std::net::TcpListener;

510

511 /// Windows answers a closed loopback port only after retrying the SYN (> the 1.5 s connect timeout).

512 fn refused(reason: &Value) -> bool {

513 reason == "CONNECTION_REFUSED" || (cfg!(windows) && reason == "TIMEOUT")

514 }

515

516 fn directory() -> tempfile::TempDir {

517 tempfile::tempdir().unwrap()

518 }

519

520 fn server(port: u16, protocol: &str) -> Value {

521 json!({"id": "p1", "name": "Test", "host": "127.0.0.1", "port": port, "protocol": protocol})

522 }

523

524 fn config(servers: Vec<Value>) -> Value {

525 json!({"version": 1, "enabled": true, "intervalSecs": 60, "showAddresses": true, "servers": servers})

526 }

527

528 fn input(data: &Path, value: &Value) {

529 fs::write(data.join(INPUT), serde_json::to_vec(value).unwrap()).unwrap();

530 }

531

532 fn listener() -> TcpListener {

533 TcpListener::bind(("127.0.0.1", 0)).expect("loopback socket required (see REPORT.md if sandbox blocks it)")

534 }

535

536 fn closed_port() -> u16 {

537 listener().local_addr().unwrap().port()

538 }

539

540 #[test]

541 fn open_tcp_and_https_are_online_unverified_with_latency() {

542 let data = directory();

543 let listener = listener();

544 let port = listener.local_addr().unwrap().port();

545 input(data.path(), &config(vec![server(port, "tcp"), server(port, "https")]));

546 let status = round(data.path(), &mut State::default());

547 for entry in status["servers"].as_array().unwrap() {

548 assert_eq!(entry["status"], "online");

549 assert_eq!(entry["service"], "unverified");

550 assert!(entry["latencyMs"].is_u64());

551 assert_eq!(entry["reason"], Value::Null);

552 assert_eq!(entry["host"], "127.0.0.1");

553 assert!(entry.get("failureReason").is_none());

554 }

555 assert_eq!(status["helper"], format!("ch-j-standby-helper {}", crate::VERSION));

556 assert_eq!(status["intervalSecs"], 60);

557 let updated = status["updatedAt"].as_str().unwrap();

558 assert_eq!(updated.len(), 20);

559 assert!(updated.ends_with('Z'));

560 // The returned object is exactly the persisted object.

561 assert_eq!(read_json(&data.path().join(OUTPUT)).unwrap(), status);

562 }

563

564 #[test]

565 fn ssh_banner_is_verified_even_when_split_between_reads() {

566 let data = directory();

567 let listener = listener();

568 let port = listener.local_addr().unwrap().port();

569 let writer = thread::spawn(move || {

570 let (mut stream, _) = listener.accept().unwrap();

571 stream.write_all(b"SS").unwrap();

572 thread::sleep(Duration::from_millis(15));

573 stream.write_all(b"H-2.0-test\r\n").unwrap();

574 });

575 input(data.path(), &config(vec![server(port, "ssh")]));

576 let status = round(data.path(), &mut State::default());

577 writer.join().unwrap();

578 assert_eq!(status["servers"][0]["status"], "online");

579 assert_eq!(status["servers"][0]["service"], "ssh");

580 }

581

582 #[test]

583 fn ssh_without_a_banner_stays_online_unverified_with_a_bounded_read() {

584 let data = directory();

585 let listener = listener();

586 let port = listener.local_addr().unwrap().port();

587 let (send, receive) = mpsc::channel();

588 let writer = thread::spawn(move || {

589 let (stream, _) = listener.accept().unwrap();

590 let _ = receive.recv_timeout(Duration::from_secs(4));

591 drop(stream);

592 });

593 input(data.path(), &config(vec![server(port, "ssh")]));

594 let start = Instant::now();

595 let status = round(data.path(), &mut State::default());

596 let _ = send.send(());

597 writer.join().unwrap();

598 assert!(start.elapsed() < Duration::from_secs(4));

599 assert_eq!(status["servers"][0]["status"], "online");

600 assert_eq!(status["servers"][0]["service"], "unverified");

601 }

602

603 #[test]

604 fn rounds_use_at_most_six_concurrent_connections() {

605 let data = directory();

606 let listener = listener();

607 let port = listener.local_addr().unwrap().port();

608 listener.set_nonblocking(true).unwrap();

609 let (stop, receive) = mpsc::channel();

610 let observer = thread::spawn(move || {

611 let mut streams: Vec<TcpStream> = Vec::new();

612 let mut peak = 0;

613 let mut accepted = 0;

614 let deadline = Instant::now() + Duration::from_secs(15);

615 loop {

616 while let Ok((stream, _)) = listener.accept() {

617 stream.set_nonblocking(true).unwrap();

618 streams.push(stream);

619 accepted += 1;

620 }

621 streams.retain(|stream| match stream.peek(&mut [0]) {

622 Ok(0) => false,

623 Err(e) => e.kind() == io::ErrorKind::WouldBlock,

624 _ => true,

625 });

626 peak = peak.max(streams.len());

627 if receive.try_recv().is_ok() || Instant::now() >= deadline { break; }

628 thread::sleep(Duration::from_millis(2));

629 }

630 (peak, accepted)

631 });

632 let servers = (0..13).map(|i| {

633 let mut entry = server(port, "ssh");

634 entry["id"] = json!(format!("p{i}"));

635 entry

636 }).collect();

637 input(data.path(), &config(servers));

638 let start = Instant::now();

639 let status = round(data.path(), &mut State::default());

640 let _ = stop.send(());

641 let (peak, accepted) = observer.join().unwrap();

642 assert_eq!(accepted, 13);

643 assert!((2..=6).contains(&peak), "observed concurrency: {peak}");

644 assert!(start.elapsed() < Duration::from_secs(12));

645

        assert!(status["servers"].as_array().unwrap().iter().all(|v| v["status"] == "online" && v["service"] == "unverified"));

646 }

647

648 #[test]

649 fn closed_port_is_unknown_then_offline_and_removal_clears_history() {

650 let data = directory();

651 let configuration = config(vec![server(closed_port(), "tcp")]);

652 input(data.path(), &configuration);

653 let mut state = State::default();

654 let first = round(data.path(), &mut state);

655 assert_eq!(first["servers"][0]["status"], "unknown");

656 assert!(refused(&first["servers"][0]["reason"]), "{first}");

657 assert_eq!(first["servers"][0]["service"], Value::Null);

658 assert_eq!(first["servers"][0]["latencyMs"], Value::Null);

659 let second = round(data.path(), &mut state);

660 assert_eq!(second["servers"][0]["status"], "offline");

661 assert!(refused(&second["servers"][0]["reason"]), "{second}");

662 input(data.path(), &config(vec![]));

663 round(data.path(), &mut state);

664 input(data.path(), &configuration);

665 assert_eq!(round(data.path(), &mut state)["servers"][0]["status"], "unknown");

666 }

667

668 #[test]

669 fn one_shot_second_failure_is_offline_with_visible_or_hidden_addresses() {

670 for show in [true, false] {

671 let data = directory();

672 let mut config = config(vec![server(closed_port(), "tcp")]);

673 config["showAddresses"] = json!(show);

674 input(data.path(), &config);

675 let first = round(data.path(), &mut State::default());

676 assert_eq!(first["servers"][0]["status"], "unknown");

677 assert_eq!(first["servers"][0].get("host").is_some(), show);

678 if !show {

679 assert!(!serde_json::to_string(&first).unwrap().contains("127.0.0.1"));

680 }

681 let second = round(data.path(), &mut State::default());

682 assert_eq!(second["servers"][0]["status"], "offline");

683 assert!(refused(&second["servers"][0]["reason"]), "{second}");

684 }

685 }

686

687 #[test]

688 fn first_failure_retains_online_and_second_one_shot_failure_goes_offline() {

689 let data = directory();

690 let listener = listener();

691 let port = listener.local_addr().unwrap().port();

692 input(data.path(), &config(vec![server(port, "tcp")]));

693 let mut state = State::default();

694 assert_eq!(round(data.path(), &mut state)["servers"][0]["status"], "online");

695 drop(listener);

696 let first = round(data.path(), &mut state);

697 assert_eq!(first["servers"][0]["status"], "online");

698 assert_eq!(first["servers"][0]["reason"], Value::Null);

699 assert!(refused(&first["servers"][0]["failureReason"]), "{first}");

700 assert_eq!(first["servers"][0]["service"], "unverified");

701 let second = round(data.path(), &mut State::default());

702 assert_eq!(second["servers"][0]["status"], "offline");

703 assert_eq!(second["servers"][0]["service"], Value::Null);

704 assert!(refused(&second["servers"][0]["reason"]), "{second}");

705 }

706

707 #[test]

708 fn successful_round_resets_consecutive_failures() {

709 let data = directory();

710 let listener = listener();

711 let port = listener.local_addr().unwrap().port();

712 let config = config(vec![server(port, "tcp")]);

713 input(data.path(), &config);

714 let key = parse_config(&config).servers[0].endpoint.clone();

715 let mut state = State { initialized: true, ..State::default() };

716 state.history.insert(key, History { status: "offline".into(), service: None, latency: None, failed: true });

717 let recovered = round(data.path(), &mut state);

718 assert_eq!(recovered["servers"][0]["status"], "online");

719 assert!(recovered["servers"][0].get("failureReason").is_none());

720 drop(listener);

721 assert_eq!(round(data.path(), &mut state)["servers"][0]["status"], "online");

722 assert_eq!(round(data.path(), &mut state)["servers"][0]["status"], "offline");

723 }

724

725 #[test]

726 fn previous_failure_requires_the_same_id_host_port_and_protocol() {

727 let data = directory();

728 let original = server(12345, "tcp");

729 for show in [true, false] {

730 let configuration = config(vec![original.clone()]);

731 let key = parse_config(&configuration).servers[0].endpoint.clone();

732 let mut old = original.clone();

733 old["status"] = json!("unknown");

734 old["reason"] = json!("TIMEOUT");

735 if !show {

736 old.as_object_mut().unwrap().remove("host");

737 old["endpointKey"] = json!(key.token());

738 }

739 write_status(&data.path().join(OUTPUT), &json!({"version": 1, "enabled": true, "servers": [old]})).unwrap();

740 assert!(previous(data.path(), &parse_config(&configuration).servers)[&key].failed);

741

            for (field, changed) in [("id", json!("p2")), ("host", json!("127.0.0.2")), ("port", json!(12346)), ("protocol", json!("ssh"))] {

742 let mut changed_server = original.clone();

743 changed_server[field] = changed;

744 let configuration = config(vec![changed_server]);

745 assert!(previous(data.path(), &parse_config(&configuration).servers).is_empty());

746 }

747 }

748 }

749

750 #[test]

751 fn invalid_entries_are_dropped_and_only_first_200_are_considered() {

752 let valid = server(22, "ssh");

753 let mut entries = vec![valid.clone()];

754 for (field, value) in [

755 ("id", json!("")), ("id", json!("i".repeat(65))), ("id", json!(1)),

756 ("name", json!("")), ("name", json!("n".repeat(129))), ("name", Value::Null),

757 ("host", json!("")), ("host", json!("h".repeat(254))), ("host", json!("with space")),

758 ("host", json!("with\nnewline")), ("host", json!("with\u{0001}control")),

759 ("host", json!("with/path")), ("host", json!("with\u{00a0}space")),

760 ("port", json!(0)), ("port", json!(65536)), ("port", json!(-1)), ("port", json!(22.5)),

761 ("protocol", json!("gopher")), ("protocol", Value::Null),

762 ] {

763 let mut bad = valid.clone();

764 bad[field] = value;

765 entries.push(bad);

766 }

767 assert_eq!(parse_config(&config(entries)).servers.len(), 1);

768 let mut boundary = valid.clone();

769 boundary["id"] = json!("é".repeat(64));

770 boundary["name"] = json!("é".repeat(128));

771 boundary["host"] = json!("h".repeat(253));

772 boundary["port"] = json!(65535);

773 assert_eq!(parse_config(&config(vec![boundary])).servers.len(), 1);

774 let entries: Vec<_> = (0..205).map(|i| {

775 let mut entry = valid.clone(); entry["id"] = json!(format!("p{i}")); entry

776 }).collect();

777 let parsed = parse_config(&config(entries));

778 assert_eq!(parsed.servers.len(), 200);

779 assert_eq!(parsed.servers[199].endpoint.id, "p199");

780 }

781

782 #[test]

783 fn interval_normalization_and_unknown_versions() {

784 for (raw, expected) in [

785 (json!(30), 30), (json!(60), 60), (json!(300), 300), (json!(0), 60),

786 (json!(31), 60), (json!(-30), 60), (json!(30.0), 60), (json!("30"), 60), (Value::Null, 60),

787 ] {

788 let mut config = config(vec![]);

789 config["intervalSecs"] = raw;

790 assert_eq!(parse_config(&config).interval, expected);

791 }

792 let mut config = config(vec![server(22, "ssh")]);

793 config["version"] = json!(2);

794 assert!(!parse_config(&config).enabled);

795 }

796

797 #[test]

798 fn disabled_missing_and_broken_input_publish_off_without_probing() {

799 let data = directory();

800 let listener = listener();

801 listener.set_nonblocking(true).unwrap();

802 let mut state = State::default();

803 let expected = |value: Value| {

804 assert_eq!(value["enabled"], false);

805 assert_eq!(value["servers"], json!([]));

806 assert!(value.get("validUntil").is_none());

807 };

808 expected(round(data.path(), &mut state));

809 let mut config = config(vec![server(listener.local_addr().unwrap().port(), "tcp")]);

810 config["enabled"] = json!(false);

811 input(data.path(), &config);

812 expected(round(data.path(), &mut state));

813 fs::write(data.path().join(INPUT), b"broken json").unwrap();

814 expected(round(data.path(), &mut state));

815 assert!(state.history.is_empty());

816 assert_eq!(listener.accept().unwrap_err().kind(), io::ErrorKind::WouldBlock);

817 }

818

819 #[test]

820 fn status_is_atomic_private_and_leaves_no_temporary_file_even_on_error() {

821 let data = directory();

822 for i in 0..3 {

823 let value = json!({"version": 1, "sequence": i});

824 write_status(&data.path().join(OUTPUT), &value).unwrap();

825 assert_eq!(read_json(&data.path().join(OUTPUT)), Some(value));

826 #[cfg(unix)]

827 {

828 use std::os::unix::fs::PermissionsExt;

829 assert_eq!(fs::metadata(data.path().join(OUTPUT)).unwrap().permissions().mode() & 0o777, 0o600);

830 }

831 }

832 let bad_target = data.path().join("directory");

833 fs::create_dir(&bad_target).unwrap();

834 assert!(write_status(&bad_target, &json!({})).is_err());

835 assert!(!fs::read_dir(data.path()).unwrap().any(|p| p.unwrap().path().extension().is_some_and(|e| e == "tmp")));

836 }

837

838 #[test]

839 fn utc_timestamps_cover_epoch_leap_years_and_non_leap_centuries() {

840 for (seconds, expected) in [

841 (0, "1970-01-01T00:00:00Z"), (-1, "1969-12-31T23:59:59Z"),

842 (951782400, "2000-02-29T00:00:00Z"),

843 (1709164800, "2024-02-29T00:00:00Z"),

844 (1709251199, "2024-02-29T23:59:59Z"),

845 (1709251200, "2024-03-01T00:00:00Z"),

846 (4107542399, "2100-02-28T23:59:59Z"),

847 (4107542400, "2100-03-01T00:00:00Z"),

848 ] {

849 assert_eq!(timestamp(seconds), expected);

850 }

851 let data = directory();

852 let mut config = config(vec![]);

853 config["intervalSecs"] = json!(300);

854 input(data.path(), &config);

855 let start = now();

856 let status = round(data.path(), &mut State::default());

857

        assert!((start..=now()).any(|n| status["updatedAt"] == timestamp(n) && status["validUntil"] == timestamp(n + 900)));

858 }

859

860 #[test]

861 fn connection_errors_use_only_fixed_reason_codes() {

862 for (kind, expected) in [

863 (io::ErrorKind::ConnectionRefused, "CONNECTION_REFUSED"),

864 (io::ErrorKind::TimedOut, "TIMEOUT"), (io::ErrorKind::WouldBlock, "TIMEOUT"),

865 (io::ErrorKind::NetworkUnreachable, "NETWORK_UNREACHABLE"),

866 (io::ErrorKind::HostUnreachable, "NETWORK_UNREACHABLE"), (io::ErrorKind::Other, "OTHER"),

867 ] {

868 assert_eq!(connection_reason(&io::Error::from(kind)), expected);

869 }

870 }

871

872 #[test]

873 fn switching_the_helper_off_leaves_no_names_or_addresses_in_the_status() {

874 let dir = directory();

875

        std::fs::write(dir.path().join(OUTPUT), r#"{"version":1,"enabled":true,"servers":[{"id":"a","name":"secret-name","host":"10.1.2.3"}]}"#).unwrap();

876 publish_off(dir.path());

877 let text = std::fs::read_to_string(dir.path().join(OUTPUT)).unwrap();

878

        assert!(text.contains("\"enabled\":false") && !text.contains("secret-name") && !text.contains("10.1.2.3"), "{text}");

879 }

880

881}

882

SHA-256

f17ababb54d0dac4a37f0de17c299b2d750d7bb553cb395c75edfe7150dba4c3

CH-J Proprietary Software License 1.14 · La fuente se proporciona en el CH-J Proprietary Software License 1.14. Su disponibilidad no cambia los términos de la licencia ni otorga derechos adicionales.

Archivo SHA-256

ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f