//! Minimal RouterOS API client for graceful MikroTik power-off. //! //! Protocol per the official MikroTik "Python3 Example" template //! (help.mikrotik.com, updated 2026-02): plain TCP (default port 8728), //! variable-length word framing (no flags, no CRC encoding), sentences //! terminated by an empty word. Login sends `/login =name= =password=`; //! modern RouterOS answers `!done` outright, pre-6.43 answers //! `!done =ret=` and expects an MD5 challenge response. //! //! We only need one command: `/system/shutdown`. use anyhow::{Context, Result}; use std::time::Duration; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::net::TcpStream; use tracing::info; use crate::config::MikrotikDevice; /// Liveness probe for `health`: connect + login, nothing else. /// `Ok` = the router is up AND the configured credentials work; /// any failure (unreachable, timeout, rejected login) = the entity is off, /// exactly like an HA switch that does not answer. pub async fn ping_device(dev: &MikrotikDevice) -> Result<()> { let target = format!("{}:{}", dev.address, dev.api_port); let mut s = tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(&target)) .await .with_context(|| format!("connecting to RouterOS at {}", target))? .with_context(|| format!("connect to RouterOS at {}", target))?; s.set_nodelay(true)?; tokio::time::timeout(Duration::from_secs(10), login_stream(&mut s, dev)) .await .map_err(|_| anyhow::anyhow!("RouterOS login timed out ({})", dev.name))? } /// Connect + login, then shut the router down via `/system/shutdown`. /// One connection per call — the router is about to die anyway. pub async fn shutdown_device(dev: &MikrotikDevice) -> Result<()> { let target = format!("{}:{}", dev.address, dev.api_port); let mut s = tokio::time::timeout(Duration::from_secs(10), TcpStream::connect(&target)) .await .with_context(|| format!("connecting to RouterOS at {}", target))? .with_context(|| format!("connect to RouterOS at {}", target))?; s.set_nodelay(true)?; login_stream(&mut s, dev).await?; shutdown_stream(&mut s, dev).await } /// The login exchange on an authenticated-or-not stream: `/login` with /// username + password, then `!done` (modern) or the legacy MD5 /// challenge round-trip (pre-6.43). async fn login_stream(s: &mut S, dev: &MikrotikDevice) -> Result<()> where S: AsyncRead + AsyncWrite + Unpin, { send_sentence( s, &[ "/login", &format!("=name={}", dev.username), &format!("=password={}", dev.password), ], ) .await .context("sending /login")?; loop { let mut words = read_sentence(s).await.context("reading /login reply")?; if words.first().is_some_and(|w| w == "#routeros") { words.remove(0); // tolerate the legacy banner word } let rest: Vec<&str> = words.iter().skip(1).map(|w| w.as_str()).collect(); match words.first().map(|s| s.as_str()) { Some("!trap") => { anyhow::bail!("RouterOS login rejected ({}): {}", dev.name, rest.join(" ")) } Some("!done") => { let Some(hex) = rest.iter().find_map(|w| w.strip_prefix("=ret=")) else { return Ok(()); // no challenge — logged in }; let challenge = hex_decode(hex) .map_err(|e| anyhow::anyhow!("login challenge is not hex: {hex} ({e})"))?; let mut pre = vec![0u8]; pre.extend_from_slice(dev.password.as_bytes()); pre.extend_from_slice(&challenge); let md = md5::compute(&pre); send_sentence( s, &[ "/login", &format!("=name={}", dev.username), &format!("=response=00{}", hex_encode(md.as_slice())), ], ) .await .context("sending login challenge response")?; } _ => anyhow::bail!( "unexpected RouterOS reply during login ({}): {}", dev.name, words.join(" ") ), } } } /// `/system/shutdown` on a logged-in stream; read replies until /// `!done` (`!trap` = the router refused, e.g. policy). async fn shutdown_stream(s: &mut S, dev: &MikrotikDevice) -> Result<()> where S: AsyncRead + AsyncWrite + Unpin, { info!("RouterOS graceful shutdown: {}", dev.name); let r = tokio::time::timeout(Duration::from_secs(10), async { send_sentence(s, &["/system/shutdown"]) .await .context("sending /system/shutdown")?; loop { let mut words = read_sentence(s).await.context("reading shutdown reply")?; if words.first().is_some_and(|w| w == "#routeros") { words.remove(0); } match words.first().map(|s| s.as_str()) { Some("!trap") => { anyhow::bail!( "RouterOS {} refused /system/shutdown: {}", dev.name, words .iter() .skip(1) .map(|w| w.as_str()) .collect::>() .join(" ") ); } Some("!done") => break, _ => {} // interim `.*` progress lines } } Ok::<(), anyhow::Error>(()) }) .await; r.context("RouterOS shutdown timed out (10 s)")? } // --- Framing ------------------------------------------------------------- // // Word = variable-length length prefix + raw UTF-8 bytes (no flags). // <0x80 1-byte length // 0x80..0xC0 (len | 0x8000), 2 bytes total // 0xC0..0xE0 (len | 0xC00000), 3 bytes total // 0xE0..0xF0 (len | 0xE0000000), 4 bytes total // 0xF0.. 5 bytes total, first is a flag only async fn write_word(w: &mut W, s: &str) -> std::io::Result<()> { let data = s.as_bytes(); let len = data.len(); let mut hdr = [0u8; 5]; let n = if len < 0x80 { hdr[0] = len as u8; 1 } else if len < 0x4000 { let v = (len | 0x8000) as u16; hdr[0..2].copy_from_slice(&v.to_be_bytes()); 2 } else if len < 0x200000 { let v = (len | 0xC00000) as u32; hdr[0..3].copy_from_slice(&v.to_be_bytes()[1..]); 3 } else if len < 0x10000000 { let v = (len | 0xE0000000) as u32; hdr[0..4].copy_from_slice(&v.to_be_bytes()); 4 } else { hdr[0] = 0xF0; hdr[1..5].copy_from_slice(&len.to_be_bytes()); 5 }; w.write_all(&hdr[..n]).await?; w.write_all(data).await?; Ok(()) } async fn read_u8(r: &mut R) -> std::io::Result { let mut x = [0u8; 1]; r.read_exact(&mut x).await?; Ok(u32::from(x[0])) } async fn read_word_len(r: &mut R) -> std::io::Result { let mut c = [0u8; 1]; r.read_exact(&mut c).await?; Ok(if c[0] < 0x80 { u32::from(c[0]) } else if c[0] < 0xC0 { ((c[0] & 0x3F) as u32) << 8 | read_u8(r).await? } else if c[0] < 0xE0 { ((c[0] & 0x1F) as u32) << 16 | read_u8(r).await? << 8 | read_u8(r).await? } else if c[0] < 0xF0 { ((c[0] & 0x0F) as u32) << 24 | read_u8(r).await? << 16 | read_u8(r).await? << 8 | read_u8(r).await? } else { read_u8(r).await? << 24 | read_u8(r).await? << 16 | read_u8(r).await? << 8 | read_u8(r).await? }) } async fn read_word(r: &mut R) -> std::io::Result> { let len = read_word_len(r).await?; let mut buf = vec![0u8; len as usize]; r.read_exact(&mut buf).await?; Ok(buf) } /// One sentence = words + a terminating empty word. async fn send_sentence(w: &mut W, words: &[&str]) -> std::io::Result<()> { for word in words { write_word(w, word).await?; } write_word(w, "").await } async fn read_sentence(r: &mut R) -> std::io::Result> { let mut words = Vec::new(); loop { let raw = read_word(r).await?; if raw.is_empty() { return Ok(words); } words.push(String::from_utf8_lossy(&raw).into_owned()); } } fn hex_decode(s: &str) -> std::result::Result, String> { let bytes = s.as_bytes(); if !bytes.len().is_multiple_of(2) { return Err("odd number of hex digits".to_string()); } let val = |c: u8| -> std::result::Result { match c { b'0'..=b'9' => Ok(c - b'0'), b'a'..=b'f' => Ok(c - b'a' + 10), b'A'..=b'F' => Ok(c - b'A' + 10), _ => Err("not a hex digit".to_string()), } }; (0..bytes.len()) .step_by(2) .map(|i| Ok(val(bytes[i])? * 16 + val(bytes[i + 1])?)) .collect() } fn hex_encode(bytes: &[u8]) -> String { let mut s = String::with_capacity(bytes.len() * 2); for b in bytes { s.push_str(&format!("{:02x}", b)); } s } #[cfg(test)] mod tests { use super::*; use tokio::io::duplex; fn device() -> MikrotikDevice { MikrotikDevice { name: "t".into(), address: "127.0.0.1".into(), username: "admin".into(), password: "secret".into(), api_port: 8728, on_entity: None, } } #[tokio::test] async fn framing_roundtrip_all_bands() { for n in [0usize, 1, 127, 128, 16383, 16384, 2_097_151] { // duplex is a bounded pipe — write and read must run concurrently let (mut a, mut b) = duplex(64 * 1024); let s = "x".repeat(n); let s2 = s.clone(); let writer = tokio::spawn(async move { write_word(&mut a, &s2).await }); let reader = tokio::spawn(async move { read_word(&mut b).await }); let got = reader.await.unwrap().unwrap(); writer.await.unwrap().unwrap(); assert_eq!(got.len(), n, "size {n}"); assert_eq!(String::from_utf8(got).unwrap(), s); } } #[tokio::test] async fn framing_band5_roundtrip() { // 2 MiB+ payload exercises the 5-byte header band. let (mut a, mut b) = duplex(64 * 1024); let s = "y".repeat(2_100_000); let s2 = s.clone(); let writer = tokio::spawn(async move { write_word(&mut a, &s2).await }); let reader = tokio::spawn(async move { read_word(&mut b).await }); let got = reader.await.unwrap().unwrap(); writer.await.unwrap().unwrap(); assert_eq!(got.len(), 2_100_000); } #[tokio::test] async fn sentence_terminates_on_empty_word() { let (mut a, mut b) = duplex(64 * 1024); tokio::spawn(async move { send_sentence(&mut a, &["/login", "=name=admin"]) .await .unwrap(); }); let words = read_sentence(&mut b).await.unwrap(); assert_eq!(words, vec!["/login", "=name=admin"]); } /// Client login exchange against a mock RouterOS that answers modern /// style: `!done` with no challenge. #[tokio::test] async fn login_modern_ok() { let (mut c, mut s) = duplex(64 * 1024); tokio::spawn(async move { let sent = read_sentence(&mut s).await.unwrap(); assert_eq!(sent, vec!["/login", "=name=admin", "=password=secret"]); send_sentence(&mut s, &["!done"]).await.unwrap(); }); login_stream(&mut c, &device()).await.unwrap(); } /// Legacy challenge: `!done =ret=` → client must answer with the /// MD5 response, then `!done`. #[tokio::test] async fn login_legacy_challenge() { let (mut c, mut s) = duplex(64 * 1024); tokio::spawn(async move { let sent = read_sentence(&mut s).await.unwrap(); assert_eq!(sent[0], "/login"); send_sentence(&mut s, &["!done", "=ret=deadbeef"]) .await .unwrap(); let resp = read_sentence(&mut s).await.unwrap(); // expected: md5(0x00 || "secret" || 0xde 0xad 0xbe 0xef) let mut pre = vec![0u8]; pre.extend_from_slice(b"secret"); pre.extend_from_slice(&[0xde, 0xad, 0xbe, 0xef]); let md = md5::compute(&pre); let want = format!("=response=00{}", hex_encode(md.as_slice())); assert_eq!( resp, vec!["/login".to_string(), "=name=admin".to_string(), want], "challenge response" ); send_sentence(&mut s, &["!done"]).await.unwrap(); }); login_stream(&mut c, &device()).await.unwrap(); } #[tokio::test] async fn login_trap_fails() { let (mut c, mut s) = duplex(64 * 1024); tokio::spawn(async move { let _ = read_sentence(&mut s).await; send_sentence(&mut s, &["!trap", "=message=wrong pass"]) .await .unwrap(); }); let e = login_stream(&mut c, &device()).await.unwrap_err(); assert!(e.to_string().contains("wrong pass"), "{}", e); } #[tokio::test] async fn shutdown_done() { let (mut c, mut s) = duplex(64 * 1024); tokio::spawn(async move { let sent = read_sentence(&mut s).await.unwrap(); assert_eq!(sent, vec!["/system/shutdown"]); send_sentence(&mut s, &["!done"]).await.unwrap(); }); shutdown_stream(&mut c, &device()).await.unwrap(); } #[tokio::test] async fn shutdown_trap_fails() { let (mut c, mut s) = duplex(64 * 1024); tokio::spawn(async move { let _ = read_sentence(&mut s).await; send_sentence(&mut s, &["!trap", "=message=denied"]) .await .unwrap(); }); let e = shutdown_stream(&mut c, &device()).await.unwrap_err(); assert!(e.to_string().contains("denied"), "{}", e); } #[test] fn hex_roundtrip() { let bytes: Vec = vec![0xde, 0xad, 0x00, 0xff, 0x01]; assert_eq!(hex_decode(&hex_encode(&bytes)).unwrap(), bytes); assert!(hex_decode("abc").is_err()); assert!(hex_decode("zz").is_err()); } }