use crate::models::*; use futures_util::StreamExt; use serde::{Deserialize, Serialize}; use socket2::{Domain, Protocol, Socket, Type}; use std::{ net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket as StdUdpSocket}, sync::Arc, time::{Duration, Instant}, }; use tokio::{net::UdpSocket, sync::Semaphore}; const MAX_PROBE_BODY: usize = 8 * 1024; pub struct ProbeInfo { pub fp: String, pub name: String, pub trusts_you: bool, } // ───────────────────────── Интерфейсы ───────────────────────── fn is_virtual_if(name: &str) -> bool { let n = name.to_ascii_lowercase(); n.starts_with("docker") || n.starts_with("br-") || n.starts_with("veth") || n.starts_with("virbr") || n.starts_with("vmnet") || n.starts_with("vboxnet") } /// (ip, broadcast) всех локальных (private/link-local/CGNAT) IPv4-интерфейсов. pub fn lan_interfaces() -> Vec<(Ipv4Addr, Option)> { if_addrs::get_if_addrs() .unwrap_or_default() .into_iter() .filter_map(|i| match &i.addr { if_addrs::IfAddr::V4(v4) if !i.is_loopback() && is_lan_v4(v4.ip) => { let bc = v4.broadcast.or_else(|| { let mask = u32::from(v4.netmask); if mask != 0 && mask != u32::MAX { Some(Ipv4Addr::from(u32::from(v4.ip) | !mask)) } else { None } }); Some((v4.ip, bc)) } _ => None, }) .collect() } /// Адрес интерфейса маршрута по умолчанию (UDP connect не отправляет пакетов и не требует интернета). pub fn detect_lan_ip() -> Option { if let Ok(s) = StdUdpSocket::bind("0.0.0.0:0") { if s.connect("192.0.2.1:9").is_ok() { if let Ok(SocketAddr::V4(a)) = s.local_addr() { if is_lan_v4(*a.ip()) { return Some(*a.ip()); } } } } let ifs = if_addrs::get_if_addrs().unwrap_or_default(); ifs.iter() .filter(|i| !i.is_loopback() && !is_virtual_if(&i.name)) .find_map(|i| match &i.addr { if_addrs::IfAddr::V4(v4) if is_lan_v4(v4.ip) => Some(v4.ip), _ => None, }) .or_else(|| lan_interfaces().first().map(|(ip, _)| *ip)) } /// Сортирует список IP пира так, чтобы адреса из наших текущих подсетей (/24) проверялись первыми. pub fn prioritize_ips(ips: &[IpAddr]) -> Vec { let my_ips: Vec = lan_interfaces().into_iter().map(|(ip, _)| ip).collect(); let in_local_subnet = |ip: &IpAddr| match ip { IpAddr::V4(v4) => { let o = v4.octets(); my_ips.iter().any(|m| { let mo = m.octets(); mo[0] == o[0] && mo[1] == o[1] && mo[2] == o[2] }) } IpAddr::V6(_) => false, }; let mut out = ips.to_vec(); out.sort_by_key(|ip| !in_local_subnet(ip)); out } // ───────────────────────── Probe ───────────────────────── /// Рукопожатие с узлом. Отпечаток берём из РЕАЛЬНО предъявленного сертификата. async fn probe_with_hint( state: &AppState, ip: IpAddr, port: u16, hint_fp: Option<&str>, ) -> Result { let ip_s = ip.to_string(); let trust_hdr: Option = { let peers = state.peers.read(); let target = hint_fp .and_then(|fp| peers.get(fp)) .or_else(|| peers.values().find(|p| p.ips.iter().any(|x| x == &ip_s))); target.map(|p| if p.verified { p.id.clone() } else { "0".to_string() }) }; let url = format!("https://{}:{}/p2p/info", ip, port); let mut req = state .probe_client .get(&url) .header("x-node-name", pct_encode(&state.device_name)) .header("x-node-port", state.p2p_port.to_string()); if let Some(t) = &trust_hdr { req = req.header("x-node-trusts", t); } let mut resp = req.send().await.map_err(|_| "узел недоступен".to_string())?; if !resp.status().is_success() { return Err(format!("узел вернул код {}", resp.status())); } let fp = resp .extensions() .get::() .and_then(|t| t.peer_certificate()) .map(fingerprint) .ok_or_else(|| "не удалось получить сертификат узла".to_string())?; if resp.content_length().map_or(false, |l| l as usize > MAX_PROBE_BODY) { return Err("слишком большой ответ".into()); } let mut buf: Vec = Vec::with_capacity(256); while let Some(chunk) = resp.chunk().await.map_err(|_| "ошибка чтения ответа".to_string())? { buf.extend_from_slice(&chunk); if buf.len() > MAX_PROBE_BODY { return Err("слишком большой ответ".into()); } } let v: serde_json::Value = serde_json::from_slice(&buf).map_err(|_| "некорректный ответ".to_string())?; Ok(ProbeInfo { fp, name: clean_label(v["name"].as_str().unwrap_or("Node"), 48), trusts_you: v["trusts_you"].as_bool().unwrap_or(false), }) } pub async fn probe(state: &AppState, ip: IpAddr, port: u16) -> Result { probe_with_hint(state, ip, port, None).await } pub async fn probe_and_register(state: &AppState, ip: IpAddr, port: u16) -> Result { let ip_s = ip.to_string(); let had_ip = state .peers .read() .values() .any(|p| p.ips.iter().any(|x| x == &ip_s)); let info = probe_with_hint(state, ip, port, None).await?; if info.fp == state.my_fingerprint { return Err("это ваше собственное устройство".into()); } state.upsert_peer(&info.name, ip, port, &info.fp); state.set_remote_trust(&info.fp, info.trusts_you); // Если доверенный узел объявился на новом IP, сразу сообщаем ему наш статус доверия if !had_ip && state.trusted_name(&info.fp).is_some() { if let Ok(info2) = probe_with_hint(state, ip, port, Some(&info.fp)).await { state.set_remote_trust(&info.fp, info2.trusts_you); } } Ok(info) } // ───────────────────────── Ping и синхронизация доверия ───────────────────────── pub async fn ping_all_peers(state: &Arc) { let Ok(_guard) = state.ping_lock.try_lock() else { return; // опрос уже идёт }; if let Some(ip) = detect_lan_ip() { state.update_my_ip(ip); } let targets: Vec<(String, Vec, u16)> = state .peers .read() .values() .filter(|p| !p.ips.is_empty()) .map(|p| (p.id.clone(), p.ips.clone(), p.port)) .collect(); futures_util::stream::iter(targets) .for_each_concurrent(8, |(id, ips, port)| async move { ping_one(state, &id, ips, port).await; }) .await; } async fn ping_one(state: &AppState, id: &str, ips: Vec, port: u16) { let parsed: Vec = ips.iter().filter_map(|s| s.parse().ok()).collect(); for ip in prioritize_ips(&parsed) { if let Ok(info) = probe_with_hint(state, ip, port, Some(id)).await { if info.fp == id { state.upsert_peer(&info.name, ip, port, id); state.set_remote_trust(id, info.trusts_you); return; } // по этому адресу теперь другое устройство — игнорируем } } let mut peers = state.peers.write(); if let Some(p) = peers.get_mut(id) { if now_sec().saturating_sub(p.last_seen) > 15 { p.online = false; } } } /// Мгновенно отправляет обновлённый статус доверия пиру и обновляет его `trusts_me`. pub async fn push_trust(state: &AppState, id: &str) { let Some((ips, port)) = state .peers .read() .get(id) .map(|p| (p.ips.clone(), p.port)) else { return; }; ping_one(state, id, ips, port).await; } /// Возвращает рабочий IP пира (для потоковой отправки файлов при возможной смене подсети). pub async fn select_peer_ip(state: &AppState, id: &str) -> Option<(IpAddr, u16)> { let (ips, port, fresh) = { let peers = state.peers.read(); let p = peers.get(id)?; let parsed: Vec = p.ips.iter().filter_map(|s| s.parse().ok()).collect(); let fresh = p.online && now_sec().saturating_sub(p.last_seen) < 5; (parsed, p.port, fresh) }; let ordered = prioritize_ips(&ips); if fresh && ordered.first() == ips.first() { return ordered.first().copied().map(|ip| (ip, port)); } for ip in &ordered { if let Ok(info) = probe_with_hint(state, *ip, port, Some(id)).await { if info.fp == id { state.upsert_peer(&info.name, *ip, port, id); state.set_remote_trust(id, info.trusts_you); return Some((*ip, port)); } } } ordered.first().copied().map(|ip| (ip, port)) } // ───────────────────────── Сканер ───────────────────────── pub async fn run_scan(state: Arc, ips: Vec, port: u16) { let st = &state; futures_util::stream::iter(ips) .for_each_concurrent(48, |ip| async move { let addr = SocketAddr::new(IpAddr::V4(ip), port); let open = matches!( tokio::time::timeout(Duration::from_millis(500), tokio::net::TcpStream::connect(addr)).await, Ok(Ok(_)) ); let found = open && probe_and_register(st, IpAddr::V4(ip), port).await.is_ok(); let mut s = st.scan_status.write(); s.scanned += 1; if found { s.found += 1; } }) .await; } // ───────────────────────── UDP-обнаружение (без mDNS) ───────────────────────── /// Beacon — лишь подсказка «я здесь». Ему НЕ доверяют: пир регистрируется только /// после настоящего TLS-рукопожатия (probe), поэтому подделка UDP не отравляет список адресов. #[derive(Serialize, Deserialize)] struct Beacon { v: u8, port: u16, fp: String, } fn broadcast_targets() -> Vec { let mut targets: Vec = lan_interfaces() .into_iter() .map(|(ip, bc)| { bc.unwrap_or_else(|| { let o = ip.octets(); Ipv4Addr::new(o[0], o[1], o[2], 255) }) }) .collect(); targets.push(Ipv4Addr::BROADCAST); targets.sort(); targets.dedup(); targets } fn bind_discovery_socket() -> std::io::Result { let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?; socket.set_reuse_address(true)?; socket.set_broadcast(true)?; socket.set_nonblocking(true)?; let addr: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), DISCOVERY_PORT); socket.bind(&addr.into())?; UdpSocket::from_std(socket.into()) } pub fn start_discovery(state: Arc) { let serving = state.log_stdout; // Отправка маяков (в режиме serve — периодическая + отслеживание смены сети; в CLI — разовый запрос port=0) let tx = state.clone(); tokio::spawn(async move { let sock = match UdpSocket::bind("0.0.0.0:0").await { Ok(s) => s, Err(e) => { tx.add_log(format!("Ошибка UDP-сокета отправки: {}", e)); return; } }; let _ = sock.set_broadcast(true); let pkt = serde_json::to_vec(&Beacon { v: 1, port: if serving { tx.p2p_port } else { 0 }, fp: tx.my_fingerprint.clone(), }) .unwrap_or_default(); if !serving { for t in broadcast_targets() { let _ = sock.send_to(&pkt, SocketAddr::new(IpAddr::V4(t), DISCOVERY_PORT)).await; } return; } let mut last_ifs: Vec = Vec::new(); loop { let cur_ifs: Vec = lan_interfaces().into_iter().map(|(ip, _)| ip).collect(); if cur_ifs != last_ifs { if let Some(ip) = detect_lan_ip() { if tx.update_my_ip(ip) { tx.add_log(format!("Смена сети: активный адрес {}", ip)); } } tx.probe_cooldown.lock().clear(); tx.ping_notify.notify_one(); last_ifs = cur_ifs; } for t in broadcast_targets() { let _ = sock .send_to(&pkt, SocketAddr::new(IpAddr::V4(t), DISCOVERY_PORT)) .await; } tokio::time::sleep(Duration::from_secs(3)).await; } }); // Приём маяков tokio::spawn(async move { let socket = match bind_discovery_socket() { Ok(s) => s, Err(e) => { state.add_log(format!( "UDP-порт {} недоступен ({}). Автообнаружение отключено, используйте сканер.", DISCOVERY_PORT, e )); return; } }; let reply_pkt = serde_json::to_vec(&Beacon { v: 1, port: state.p2p_port, fp: state.my_fingerprint.clone(), }) .unwrap_or_default(); let mut last_reply = Instant::now() - Duration::from_secs(2); let slots = Arc::new(Semaphore::new(8)); let mut buf = [0u8; 512]; loop { let (len, remote) = match socket.recv_from(&mut buf).await { Ok(v) => v, Err(_) => { tokio::time::sleep(Duration::from_millis(100)).await; continue; } }; let IpAddr::V4(ip4) = remote.ip() else { continue }; if !is_lan_v4(ip4) { continue; } let Ok(b) = serde_json::from_slice::(&buf[..len]) else { continue }; if b.v != 1 || !is_valid_fp(&b.fp) || b.fp == state.my_fingerprint { continue; } // port == 0 — запрос от CLI или нового узла; отвечаем своим маяком без ожидания 3 с if b.port == 0 { if serving && last_reply.elapsed() >= Duration::from_millis(500) { last_reply = Instant::now(); for t in broadcast_targets() { let _ = socket .send_to(&reply_pkt, SocketAddr::new(IpAddr::V4(t), DISCOVERY_PORT)) .await; } } continue; } let ip = IpAddr::V4(ip4); if state.is_fresh(&b.fp, ip, b.port) || !state.probe_allowed(ip, b.port) { continue; } let Ok(permit) = slots.clone().try_acquire_owned() else { continue }; let st = state.clone(); let port = b.port; tokio::spawn(async move { let _permit = permit; let _ = probe_and_register(&st, ip, port).await; }); } }); }