a.0.0.5
This commit is contained in:
netadmin committed
1 parent
e88dc07696
commit
5d01dfb63a
12 files changed
+2438
-1891
No files matched your search
@@ -1,433 +1,467 @@
|
||||
use crate::models::{
|
||||
now_sec, parse_ip_range, sanitize_filename, url_decode, AppState, IncomingTextPayload,
|
||||
MediaItem, PeerDevice, ScanProgress, TextItem, TransferProgress, UiManualPeerRequest,
|
||||
UiScanRangeRequest, UiSendTextRequest, UiTrustRequest, MEDIA_DIR,
|
||||
};
|
||||
use crate::{models::*, net, storage};
|
||||
use axum::{
|
||||
body::Bytes,
|
||||
extract::{Path, State},
|
||||
http::{header, HeaderMap, StatusCode},
|
||||
response::{Html, IntoResponse},
|
||||
body::{Body, Bytes},
|
||||
extract::{Path, Request, State},
|
||||
http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode},
|
||||
middleware::Next,
|
||||
response::{Html, IntoResponse, Response},
|
||||
Json,
|
||||
};
|
||||
use std::{net::SocketAddr, path::PathBuf, sync::Arc, time::Duration};
|
||||
use tokio::sync::Semaphore;
|
||||
use futures_util::StreamExt;
|
||||
use std::{
|
||||
io,
|
||||
net::IpAddr,
|
||||
sync::Arc,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
use tokio_stream::wrappers::ReceiverStream;
|
||||
use tokio_util::io::ReaderStream;
|
||||
|
||||
const INDEX_HTML: &str = include_str!("static/index.html");
|
||||
const APP_JS: &str = include_str!("static/app.js");
|
||||
|
||||
const CSP: &str = "default-src 'none'; script-src 'self'; style-src 'unsafe-inline'; img-src 'self'; \
|
||||
connect-src 'self'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'";
|
||||
|
||||
// ───────────────────────── Защита UI (CSRF / DNS-rebinding) ─────────────────────────
|
||||
pub struct UiGuard {
|
||||
pub token: String,
|
||||
pub cookie_name: String,
|
||||
pub hosts: Vec<String>,
|
||||
}
|
||||
|
||||
fn ct_eq(a: &[u8], b: &[u8]) -> bool {
|
||||
a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
|
||||
}
|
||||
|
||||
fn cookie_ok(headers: &HeaderMap, g: &UiGuard) -> bool {
|
||||
headers
|
||||
.get_all(header::COOKIE)
|
||||
.iter()
|
||||
.filter_map(|v| v.to_str().ok())
|
||||
.flat_map(|s| s.split(';'))
|
||||
.any(|kv| {
|
||||
kv.trim()
|
||||
.strip_prefix(g.cookie_name.as_str())
|
||||
.and_then(|r| r.strip_prefix('='))
|
||||
.map_or(false, |v| ct_eq(v.as_bytes(), g.token.as_bytes()))
|
||||
})
|
||||
}
|
||||
|
||||
fn with_security_headers(mut resp: Response) -> Response {
|
||||
let items: [(HeaderName, &'static str); 5] = [
|
||||
(header::CONTENT_SECURITY_POLICY, CSP),
|
||||
(HeaderName::from_static("x-content-type-options"), "nosniff"),
|
||||
(header::X_FRAME_OPTIONS, "DENY"),
|
||||
(header::REFERRER_POLICY, "no-referrer"),
|
||||
(header::CACHE_CONTROL, "no-store"),
|
||||
];
|
||||
let h = resp.headers_mut();
|
||||
for (k, v) in items {
|
||||
h.entry(k).or_insert(HeaderValue::from_static(v)); // не перетираем заголовки обработчика
|
||||
}
|
||||
resp
|
||||
}
|
||||
|
||||
/// 1) Host только 127.0.0.1/localhost:порт (анти-rebinding)
|
||||
/// 2) POST только с Origin нашего же хоста (анти-CSRF)
|
||||
/// 3) Доступ по токену из консоли: /?t=… выставляет HttpOnly SameSite=Strict cookie
|
||||
pub async fn ui_guard(State(g): State<Arc<UiGuard>>, req: Request, next: Next) -> Response {
|
||||
let headers = req.headers();
|
||||
|
||||
let host_ok = headers
|
||||
.get(header::HOST)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map_or(false, |h| g.hosts.iter().any(|a| a.eq_ignore_ascii_case(h)));
|
||||
if !host_ok {
|
||||
return StatusCode::FORBIDDEN.into_response();
|
||||
}
|
||||
|
||||
if !matches!(*req.method(), Method::GET | Method::HEAD) {
|
||||
let origin_ok = headers
|
||||
.get(header::ORIGIN)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|o| o.strip_prefix("http://"))
|
||||
.map_or(false, |o| g.hosts.iter().any(|a| a.eq_ignore_ascii_case(o)));
|
||||
if !origin_ok {
|
||||
return StatusCode::FORBIDDEN.into_response();
|
||||
}
|
||||
}
|
||||
|
||||
if req.uri().path() == "/" {
|
||||
if let Some(q) = req.uri().query() {
|
||||
let token_ok = q
|
||||
.split('&')
|
||||
.filter_map(|kv| kv.strip_prefix("t="))
|
||||
.any(|t| ct_eq(t.as_bytes(), g.token.as_bytes()));
|
||||
if token_ok {
|
||||
let mut resp = StatusCode::SEE_OTHER.into_response();
|
||||
resp.headers_mut().insert(header::LOCATION, HeaderValue::from_static("/"));
|
||||
if let Ok(c) = HeaderValue::from_str(&format!(
|
||||
"{}={}; Path=/; HttpOnly; SameSite=Strict",
|
||||
g.cookie_name, g.token
|
||||
)) {
|
||||
resp.headers_mut().insert(header::SET_COOKIE, c);
|
||||
}
|
||||
return with_security_headers(resp);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !cookie_ok(headers, &g) {
|
||||
return with_security_headers(
|
||||
(StatusCode::FORBIDDEN, "Откройте адрес из консоли (с токеном ?t=…)").into_response(),
|
||||
);
|
||||
}
|
||||
with_security_headers(next.run(req).await)
|
||||
}
|
||||
|
||||
// ───────────────────────── Статика и состояние ─────────────────────────
|
||||
pub async fn ui_index() -> Html<&'static str> {
|
||||
Html(INDEX_HTML)
|
||||
}
|
||||
|
||||
pub async fn ui_get_state(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
|
||||
let mut peers_list: Vec<PeerDevice> =
|
||||
state.peers.read().unwrap().values().cloned().collect();
|
||||
peers_list.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
pub async fn ui_js() -> impl IntoResponse {
|
||||
([(header::CONTENT_TYPE, "text/javascript; charset=utf-8")], APP_JS)
|
||||
}
|
||||
|
||||
let texts: Vec<TextItem> = state.ram_texts.read().unwrap().iter().cloned().collect();
|
||||
let media: Vec<MediaItem> = state.disk_media.read().unwrap().iter().cloned().collect();
|
||||
let transfers: Vec<TransferProgress> = state.transfers.read().unwrap().clone();
|
||||
let scan = state.scan_status.read().unwrap().clone();
|
||||
let logs: Vec<String> = state.logs.read().unwrap().iter().cloned().collect();
|
||||
pub async fn ui_get_state(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
|
||||
let mut peers: Vec<PeerDevice> = state.peers.read().values().cloned().collect();
|
||||
peers.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
let texts: Vec<TextItem> = state.ram_texts.read().iter().cloned().collect();
|
||||
let media: Vec<MediaItem> = state.disk_media.read().iter().cloned().collect();
|
||||
let transfers: Vec<TransferProgress> = state.transfers.read().clone();
|
||||
let scan = state.scan_status.read().clone();
|
||||
|
||||
Json(serde_json::json!({
|
||||
"device_name": state.device_name,
|
||||
"my_ip": state.my_ip,
|
||||
"p2p_port": state.p2p_port,
|
||||
"my_fingerprint": state.my_fingerprint,
|
||||
"my_pin": state.my_fingerprint[..8].to_uppercase(),
|
||||
"peers": peers_list,
|
||||
"my_pin": short_id(&state.my_fingerprint),
|
||||
"media_dir": state.media_dir.display().to_string(),
|
||||
"peers": peers,
|
||||
"texts": texts,
|
||||
"media": media,
|
||||
"transfers": transfers,
|
||||
"scan": scan,
|
||||
"logs": logs
|
||||
}))
|
||||
}
|
||||
|
||||
// ───────────────────────── Устройства ─────────────────────────
|
||||
pub async fn ui_add_manual_peer(
|
||||
State(state): State<Arc<AppState>>,
|
||||
Json(req): Json<UiManualPeerRequest>,
|
||||
) -> Result<StatusCode, (StatusCode, String)> {
|
||||
let ip = req.ip.trim().to_string();
|
||||
let port = req.port.unwrap_or(state.p2p_port);
|
||||
|
||||
let url = format!("https://{}:{}/p2p/info", ip, port);
|
||||
let resp = state
|
||||
.probe_client
|
||||
.get(&url)
|
||||
.header("X-Node-Fp", &state.my_fingerprint)
|
||||
.header("X-Node-Name", &state.device_name)
|
||||
.header("X-Node-Ip", &state.my_ip)
|
||||
.header("X-Node-Port", state.p2p_port.to_string())
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| (StatusCode::BAD_REQUEST, format!("Узел {} недоступен: {}", ip, e)))?;
|
||||
|
||||
if !resp.status().is_success() {
|
||||
return Err((StatusCode::BAD_GATEWAY, "Удаленный узел вернул ошибку".into()));
|
||||
) -> Result<StatusCode, AppError> {
|
||||
let ip: IpAddr = req
|
||||
.ip
|
||||
.trim()
|
||||
.parse()
|
||||
.map_err(|_| AppError::new(StatusCode::BAD_REQUEST, "Некорректный IP-адрес"))?;
|
||||
match ip {
|
||||
IpAddr::V4(v4) if is_lan_v4(v4) => {}
|
||||
_ => return Err(AppError::new(StatusCode::BAD_REQUEST, "Разрешены только локальные IPv4-адреса")),
|
||||
}
|
||||
|
||||
let info: serde_json::Value = resp
|
||||
.json()
|
||||
let port = req.port.filter(|p| *p != 0).unwrap_or(state.p2p_port);
|
||||
net::probe_and_register(&state, ip, port)
|
||||
.await
|
||||
.map_err(|_| (StatusCode::BAD_GATEWAY, "Некорректный JSON от узла".into()))?;
|
||||
|
||||
let name = info["name"].as_str().unwrap_or("Remote-Node").to_string();
|
||||
let fingerprint = info["fingerprint"].as_str().unwrap_or("").to_string();
|
||||
let remote_port = info["port"].as_u64().unwrap_or(port as u64) as u16;
|
||||
|
||||
if fingerprint.len() < 8 {
|
||||
return Err((StatusCode::BAD_GATEWAY, "Недействительный отпечаток сертификата".into()));
|
||||
}
|
||||
|
||||
state.upsert_peer(name, ip, remote_port, fingerprint);
|
||||
.map_err(|e| AppError::new(StatusCode::BAD_GATEWAY, format!("Узел {}: {}", ip, e)))?;
|
||||
Ok(StatusCode::OK)
|
||||
}
|
||||
|
||||
/// Асинхронное сканирование всего или части диапазона IP-адресов
|
||||
pub async fn ui_scan_range(
|
||||
State(state): State<Arc<AppState>>,
|
||||
Json(req): Json<UiScanRangeRequest>,
|
||||
) -> Result<StatusCode, (StatusCode, String)> {
|
||||
{
|
||||
let scan = state.scan_status.read().unwrap();
|
||||
if scan.active {
|
||||
return Err((StatusCode::CONFLICT, "Сканирование уже выполняется".into()));
|
||||
}
|
||||
}
|
||||
|
||||
let port = req.port.unwrap_or(state.p2p_port);
|
||||
let (range_label, ips) =
|
||||
parse_ip_range(&req.range, &state.my_ip).map_err(|e| (StatusCode::BAD_REQUEST, e))?;
|
||||
|
||||
) -> Result<StatusCode, AppError> {
|
||||
let port = req.port.filter(|p| *p != 0).unwrap_or(state.p2p_port);
|
||||
let (label, ips) =
|
||||
parse_ip_range(&req.range, &state.my_ip).map_err(|e| AppError::new(StatusCode::BAD_REQUEST, e))?;
|
||||
let total = ips.len();
|
||||
|
||||
{
|
||||
let mut scan = state.scan_status.write().unwrap();
|
||||
*scan = ScanProgress {
|
||||
// проверка и установка флага — атомарно, под одним write-замком
|
||||
let mut s = state.scan_status.write();
|
||||
if s.active {
|
||||
return Err(AppError::new(StatusCode::CONFLICT, "Сканирование уже выполняется"));
|
||||
}
|
||||
*s = ScanProgress {
|
||||
active: true,
|
||||
range: range_label.clone(),
|
||||
range: label.clone(),
|
||||
scanned: 0,
|
||||
total,
|
||||
found: 0,
|
||||
};
|
||||
}
|
||||
state.add_log(format!("Запущено сканирование {} ({} адресов, порт {})", label, total, port));
|
||||
|
||||
state.add_log(format!(
|
||||
"Запущено сканирование диапазона {} ({} адресов, порт {})",
|
||||
range_label, total, port
|
||||
));
|
||||
|
||||
let state_bg = state.clone();
|
||||
let st = state.clone();
|
||||
let worker = tokio::spawn(net::run_scan(state.clone(), ips, port));
|
||||
tokio::spawn(async move {
|
||||
let semaphore = Arc::new(Semaphore::new(48));
|
||||
let mut handles = Vec::with_capacity(total);
|
||||
|
||||
for ip in ips {
|
||||
let sem = semaphore.clone();
|
||||
let st = state_bg.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
let _permit = sem.acquire().await.unwrap();
|
||||
|
||||
// Быстрая проверка открытости TCP-порта (350 мс) перед HTTPS-рукопожатием
|
||||
let sock_addr: Option<SocketAddr> = format!("{}:{}", ip, port).parse().ok();
|
||||
let mut found_node = false;
|
||||
|
||||
if let Some(addr) = sock_addr {
|
||||
if tokio::time::timeout(
|
||||
Duration::from_millis(350),
|
||||
tokio::net::TcpStream::connect(addr),
|
||||
)
|
||||
.await
|
||||
.map(|r| r.is_ok())
|
||||
.unwrap_or(false)
|
||||
{
|
||||
let url = format!("https://{}:{}/p2p/info", ip, port);
|
||||
if let Ok(resp) = st
|
||||
.probe_client
|
||||
.get(&url)
|
||||
.header("X-Node-Fp", &st.my_fingerprint)
|
||||
.header("X-Node-Name", &st.device_name)
|
||||
.header("X-Node-Ip", &st.my_ip)
|
||||
.header("X-Node-Port", st.p2p_port.to_string())
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
if let Ok(info) = resp.json::<serde_json::Value>().await {
|
||||
let name = info["name"].as_str().unwrap_or("Node").to_string();
|
||||
let fp = info["fingerprint"].as_str().unwrap_or("").to_string();
|
||||
let r_port = info["port"].as_u64().unwrap_or(port as u64) as u16;
|
||||
if fp.len() >= 8 {
|
||||
st.upsert_peer(name, ip, r_port, fp);
|
||||
found_node = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut scan = st.scan_status.write().unwrap();
|
||||
scan.scanned += 1;
|
||||
if found_node {
|
||||
scan.found += 1;
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
for h in handles {
|
||||
let _ = h.await;
|
||||
}
|
||||
|
||||
let found_total = {
|
||||
let mut scan = state_bg.scan_status.write().unwrap();
|
||||
scan.active = false;
|
||||
scan.found
|
||||
let _ = worker.await; // даже при панике воркера флаг active будет сброшен
|
||||
let found = {
|
||||
let mut s = st.scan_status.write();
|
||||
s.active = false;
|
||||
s.found
|
||||
};
|
||||
|
||||
state_bg.add_log(format!(
|
||||
"Сканирование {} завершено. Найдено устройств: {}",
|
||||
range_label, found_total
|
||||
));
|
||||
st.add_log(format!("Сканирование {} завершено. Найдено устройств: {}", label, found));
|
||||
});
|
||||
|
||||
Ok(StatusCode::OK)
|
||||
}
|
||||
|
||||
pub async fn ui_ping_peers(State(state): State<Arc<AppState>>) -> StatusCode {
|
||||
let state_clone = state.clone();
|
||||
tokio::spawn(async move {
|
||||
crate::ping_all_peers(&state_clone).await;
|
||||
});
|
||||
tokio::spawn(async move { net::ping_all_peers(&state).await });
|
||||
StatusCode::OK
|
||||
}
|
||||
|
||||
pub async fn ui_trust_peer(
|
||||
State(state): State<Arc<AppState>>,
|
||||
Json(req): Json<UiTrustRequest>,
|
||||
) -> Result<StatusCode, (StatusCode, String)> {
|
||||
let mut peers = state.peers.write().unwrap();
|
||||
let peer = peers
|
||||
.get_mut(&req.peer_id)
|
||||
.ok_or((StatusCode::NOT_FOUND, "Устройство не найдено".to_string()))?;
|
||||
|
||||
peer.verified = req.trust;
|
||||
let msg = if req.trust {
|
||||
format!("Сертификат '{}' [{}] ЗАВЕРЕН", peer.name, peer.pin_code)
|
||||
} else {
|
||||
format!("Доверие устройству '{}' [{}] ОТОЗВАНО", peer.name, peer.pin_code)
|
||||
) -> Result<StatusCode, AppError> {
|
||||
let msg = {
|
||||
let mut peers = state.peers.write();
|
||||
let p = peers
|
||||
.get_mut(&req.peer_id)
|
||||
.ok_or_else(|| AppError::new(StatusCode::NOT_FOUND, "Устройство не найдено"))?;
|
||||
p.verified = req.trust;
|
||||
if req.trust {
|
||||
format!("Устройство '{}' [{}] ЗАВЕРЕНО", p.name, p.pin_code)
|
||||
} else {
|
||||
format!("Доверие устройству '{}' [{}] ОТОЗВАНО", p.name, p.pin_code)
|
||||
}
|
||||
};
|
||||
drop(peers);
|
||||
state.persist_trust();
|
||||
state.add_log(msg);
|
||||
|
||||
Ok(StatusCode::OK)
|
||||
}
|
||||
|
||||
pub async fn ui_clear_ram(State(state): State<Arc<AppState>>) -> StatusCode {
|
||||
state.ram_texts.write().unwrap().clear();
|
||||
state.ram_texts.write().clear();
|
||||
state.add_log("Текстовый буфер в ОЗУ очищен");
|
||||
StatusCode::OK
|
||||
}
|
||||
|
||||
// ───────────────────────── Отправка ─────────────────────────
|
||||
fn resolve_target(state: &AppState, id: &str) -> Result<(Vec<IpAddr>, u16, String), AppError> {
|
||||
let peers = state.peers.read();
|
||||
let p = peers
|
||||
.get(id)
|
||||
.ok_or_else(|| AppError::new(StatusCode::NOT_FOUND, "Получатель не найден"))?;
|
||||
if !p.verified {
|
||||
return Err(AppError::new(
|
||||
StatusCode::FORBIDDEN,
|
||||
format!("Сначала заверьте устройство '{}' (код сверки {})", p.name, p.pin_code),
|
||||
));
|
||||
}
|
||||
let ips: Vec<IpAddr> = p.ips.iter().filter_map(|s| s.parse().ok()).collect();
|
||||
if ips.is_empty() {
|
||||
return Err(AppError::new(
|
||||
StatusCode::BAD_REQUEST,
|
||||
"Адрес получателя неизвестен. Выполните опрос или сканирование.",
|
||||
));
|
||||
}
|
||||
Ok((ips, p.port, p.name.clone()))
|
||||
}
|
||||
|
||||
pub async fn ui_send_text(
|
||||
State(state): State<Arc<AppState>>,
|
||||
Json(req): Json<UiSendTextRequest>,
|
||||
) -> Result<StatusCode, (StatusCode, String)> {
|
||||
let (target_ip, port, peer_name) = {
|
||||
let peers = state.peers.read().unwrap();
|
||||
let peer = peers
|
||||
.get(&req.peer_id)
|
||||
.ok_or((StatusCode::NOT_FOUND, "Получатель не найден".to_string()))?;
|
||||
if !peer.verified {
|
||||
return Err((
|
||||
StatusCode::FORBIDDEN,
|
||||
format!(
|
||||
"Сначала заверьте сертификат устройства '{}' [PIN: {}]!",
|
||||
peer.name, peer.pin_code
|
||||
),
|
||||
));
|
||||
}
|
||||
let ip = peer
|
||||
.ips
|
||||
.first()
|
||||
.cloned()
|
||||
.ok_or((StatusCode::BAD_REQUEST, "У получателя нет известного IP".to_string()))?;
|
||||
(ip, peer.port, peer.name.clone())
|
||||
};
|
||||
|
||||
let url = format!("https://{}:{}/p2p/text", target_ip, port);
|
||||
let payload = IncomingTextPayload {
|
||||
sender: state.device_name.clone(),
|
||||
sender_fp: state.my_fingerprint.clone(),
|
||||
text: req.text,
|
||||
};
|
||||
|
||||
let resp = state
|
||||
.http_client
|
||||
.post(&url)
|
||||
.json(&payload)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| (StatusCode::BAD_GATEWAY, format!("Ошибка TLS/Сети: {}", e)))?;
|
||||
|
||||
if resp.status().is_success() {
|
||||
state.add_log(format!("Текст доставлен в ОЗУ устройства '{}'", peer_name));
|
||||
Ok(StatusCode::OK)
|
||||
} else {
|
||||
Err((resp.status(), "Удаленный узел отклонил запрос".to_string()))
|
||||
) -> Result<StatusCode, AppError> {
|
||||
if req.text.trim().is_empty() {
|
||||
return Err(AppError::new(StatusCode::BAD_REQUEST, "Пустой текст"));
|
||||
}
|
||||
if req.text.len() > MAX_TEXT_BYTES {
|
||||
return Err(AppError::new(StatusCode::PAYLOAD_TOO_LARGE, "Текст больше 64 КБ"));
|
||||
}
|
||||
let (ips, port, peer_name) = resolve_target(&state, &req.peer_id)?;
|
||||
let client = state.client_for_peer(&req.peer_id, false)?;
|
||||
let body = serde_json::to_vec(&IncomingTextPayload { text: req.text })
|
||||
.map_err(|e| AppError::internal("json", e))?;
|
||||
|
||||
for ip in ips {
|
||||
let url = format!("https://{}:{}/p2p/text", ip, port);
|
||||
match client
|
||||
.post(&url)
|
||||
.header(header::CONTENT_TYPE, "application/json")
|
||||
.body(body.clone())
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(r) if r.status().is_success() => {
|
||||
state.add_log(format!("Текст доставлен в ОЗУ устройства '{}'", peer_name));
|
||||
return Ok(StatusCode::OK);
|
||||
}
|
||||
Ok(r) if r.status() == StatusCode::FORBIDDEN => {
|
||||
return Err(AppError::new(
|
||||
StatusCode::FORBIDDEN,
|
||||
"Получатель ещё не заверил ваше устройство (доверие должно быть взаимным)",
|
||||
))
|
||||
}
|
||||
Ok(r) => {
|
||||
return Err(AppError::new(
|
||||
StatusCode::BAD_GATEWAY,
|
||||
format!("Получатель отклонил запрос ({})", r.status()),
|
||||
))
|
||||
}
|
||||
Err(_) => continue, // пробуем следующий адрес
|
||||
}
|
||||
}
|
||||
Err(AppError::new(
|
||||
StatusCode::BAD_GATEWAY,
|
||||
"Не удалось установить защищённое соединение (узел недоступен или сертификат не совпал)",
|
||||
))
|
||||
}
|
||||
|
||||
/// Почанковая отправка медиафайла с реальным отслеживанием прогресса по сети
|
||||
/// Потоковая отправка: браузер → (канал с backpressure) → TLS → получатель. Память O(1).
|
||||
pub async fn ui_send_media(
|
||||
State(state): State<Arc<AppState>>,
|
||||
headers: HeaderMap,
|
||||
body: Bytes,
|
||||
) -> Result<StatusCode, (StatusCode, String)> {
|
||||
let peer_id = headers
|
||||
.get("X-Target-Id")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.ok_or((StatusCode::BAD_REQUEST, "Не указан ID получателя".to_string()))?;
|
||||
|
||||
let raw_filename = headers
|
||||
.get("X-Filename")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("file.bin")
|
||||
body: Body,
|
||||
) -> Result<StatusCode, AppError> {
|
||||
let peer_id = hdr(&headers, "x-target-id")
|
||||
.ok_or_else(|| AppError::new(StatusCode::BAD_REQUEST, "Не указан получатель"))?
|
||||
.to_string();
|
||||
let display_name = sanitize_filename(&pct_decode(hdr(&headers, "x-filename").unwrap_or("file.bin")));
|
||||
let total: u64 = hdr(&headers, "content-length")
|
||||
.and_then(|v| v.parse().ok())
|
||||
.ok_or_else(|| AppError::new(StatusCode::LENGTH_REQUIRED, "Не указан размер файла"))?;
|
||||
if total == 0 {
|
||||
return Err(AppError::new(StatusCode::BAD_REQUEST, "Файл пустой"));
|
||||
}
|
||||
if total > MAX_FILE_SIZE {
|
||||
return Err(AppError::new(StatusCode::PAYLOAD_TOO_LARGE, "Файл больше 2 ГБ"));
|
||||
}
|
||||
|
||||
let display_name = sanitize_filename(&url_decode(&raw_filename));
|
||||
let (ips, port, peer_name) = resolve_target(&state, &peer_id)?;
|
||||
let client = state.client_for_peer(&peer_id, true)?;
|
||||
let url = format!("https://{}:{}/p2p/media", ips[0], port);
|
||||
|
||||
let mime = headers
|
||||
.get(header::CONTENT_TYPE)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("application/octet-stream")
|
||||
.to_string();
|
||||
|
||||
let (target_ip, port, peer_name) = {
|
||||
let peers = state.peers.read().unwrap();
|
||||
let peer = peers
|
||||
.get(peer_id)
|
||||
.ok_or((StatusCode::NOT_FOUND, "Получатель не найден".to_string()))?;
|
||||
if !peer.verified {
|
||||
return Err((
|
||||
StatusCode::FORBIDDEN,
|
||||
format!(
|
||||
"Сначала заверьте сертификат устройства '{}' [PIN: {}]!",
|
||||
peer.name, peer.pin_code
|
||||
),
|
||||
));
|
||||
}
|
||||
let ip = peer
|
||||
.ips
|
||||
.first()
|
||||
.cloned()
|
||||
.ok_or((StatusCode::BAD_REQUEST, "У получателя нет IP".to_string()))?;
|
||||
(ip, peer.port, peer.name.clone())
|
||||
};
|
||||
|
||||
let total_size = body.len();
|
||||
let transfer_id = format!("tx_{}_{}", now_sec(), total_size);
|
||||
let url = format!("https://{}:{}/p2p/media", target_ip, port);
|
||||
|
||||
// Адаптивный размер чанка для плавного обновления прогресс-бара на любых файлах
|
||||
let chunk_size = if total_size < 1024 * 1024 {
|
||||
32 * 1024 // 32 КБ для файлов до 1 МБ
|
||||
} else if total_size < 15 * 1024 * 1024 {
|
||||
128 * 1024 // 128 КБ для файлов до 15 МБ
|
||||
} else {
|
||||
512 * 1024 // 512 КБ для крупных файлов
|
||||
};
|
||||
|
||||
state.upsert_transfer(TransferProgress {
|
||||
id: transfer_id.clone(),
|
||||
let tid = format!("tx_{}", random_hex(6));
|
||||
state.add_transfer(TransferProgress {
|
||||
id: tid.clone(),
|
||||
filename: display_name.clone(),
|
||||
peer_name: peer_name.clone(),
|
||||
direction: "out".to_string(),
|
||||
direction: Direction::Out,
|
||||
transferred_bytes: 0,
|
||||
total_bytes: total_size,
|
||||
total_bytes: total,
|
||||
percent: 0,
|
||||
done: false,
|
||||
error: None,
|
||||
updated_at: now_sec(),
|
||||
});
|
||||
|
||||
let mut offset = 0usize;
|
||||
loop {
|
||||
let end = (offset + chunk_size).min(total_size);
|
||||
let chunk = body.slice(offset..end);
|
||||
|
||||
let send_res = state
|
||||
.http_client
|
||||
.post(&url)
|
||||
.header("X-Sender", &state.device_name)
|
||||
.header("X-Filename", &raw_filename)
|
||||
.header("X-Transfer-Id", &transfer_id)
|
||||
.header("X-Offset", offset.to_string())
|
||||
.header("X-Total-Size", total_size.to_string())
|
||||
.header(header::CONTENT_TYPE, &mime)
|
||||
.body(chunk)
|
||||
.send()
|
||||
.await;
|
||||
|
||||
match send_res {
|
||||
Ok(resp) if resp.status().is_success() => {
|
||||
offset = end;
|
||||
let is_done = offset >= total_size;
|
||||
let percent = if total_size == 0 {
|
||||
100
|
||||
} else {
|
||||
((offset as f64 / total_size as f64) * 100.0).min(100.0) as u8
|
||||
};
|
||||
|
||||
state.upsert_transfer(TransferProgress {
|
||||
id: transfer_id.clone(),
|
||||
filename: display_name.clone(),
|
||||
peer_name: peer_name.clone(),
|
||||
direction: "out".to_string(),
|
||||
transferred_bytes: offset,
|
||||
total_bytes: total_size,
|
||||
percent,
|
||||
done: is_done,
|
||||
error: None,
|
||||
updated_at: now_sec(),
|
||||
});
|
||||
|
||||
if is_done {
|
||||
break;
|
||||
let (tx, rx) = tokio::sync::mpsc::channel::<Result<Bytes, io::Error>>(4);
|
||||
let fwd_state = state.clone();
|
||||
let fwd_id = tid.clone();
|
||||
let fwd = tokio::spawn(async move {
|
||||
let mut stream = std::pin::pin!(body.into_data_stream());
|
||||
let mut sent: u64 = 0;
|
||||
let mut last = Instant::now();
|
||||
while let Some(item) = stream.next().await {
|
||||
let chunk = match item {
|
||||
Ok(c) => c,
|
||||
Err(_) => {
|
||||
let _ = tx.send(Err(io::Error::new(io::ErrorKind::Other, "client aborted"))).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
sent += chunk.len() as u64;
|
||||
if sent > total {
|
||||
let _ = tx.send(Err(io::Error::new(io::ErrorKind::Other, "body too long"))).await;
|
||||
return;
|
||||
}
|
||||
Ok(resp) => {
|
||||
let err_msg = format!("Узел отклонил чанк (код {})", resp.status());
|
||||
state.remove_transfer(&transfer_id);
|
||||
return Err((resp.status(), err_msg));
|
||||
// если получатель не читает 30 с — закрываем поток, запрос завершится ошибкой
|
||||
match tokio::time::timeout(Duration::from_secs(30), tx.send(Ok(chunk))).await {
|
||||
Ok(Ok(())) => {}
|
||||
_ => return,
|
||||
}
|
||||
Err(e) => {
|
||||
let err_msg = format!("Ошибка сети при передаче файла: {}", e);
|
||||
state.remove_transfer(&transfer_id);
|
||||
return Err((StatusCode::BAD_GATEWAY, err_msg));
|
||||
if last.elapsed() >= Duration::from_millis(100) {
|
||||
fwd_state.update_transfer_bytes(&fwd_id, sent);
|
||||
last = Instant::now();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
state.add_log(format!(
|
||||
"Медиафайл '{}' ({} байт) успешно доставлен на '{}'",
|
||||
display_name, total_size, peer_name
|
||||
));
|
||||
|
||||
let st_cleanup = state.clone();
|
||||
let tid_cleanup = transfer_id.clone();
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(Duration::from_secs(4)).await;
|
||||
st_cleanup.remove_transfer(&tid_cleanup);
|
||||
});
|
||||
|
||||
Ok(StatusCode::OK)
|
||||
}
|
||||
let result = client
|
||||
.post(&url)
|
||||
.header("x-filename", pct_encode(&display_name))
|
||||
.header("x-total-size", total.to_string())
|
||||
.header(header::CONTENT_TYPE, "application/octet-stream")
|
||||
.header(header::CONTENT_LENGTH, total.to_string())
|
||||
.body(reqwest::Body::wrap_stream(ReceiverStream::new(rx)))
|
||||
.send()
|
||||
.await;
|
||||
fwd.abort();
|
||||
|
||||
pub async fn ui_serve_saved_file(Path(filename): Path<String>) -> impl IntoResponse {
|
||||
let safe_name = sanitize_filename(&filename);
|
||||
let path = PathBuf::from(MEDIA_DIR).join(&safe_name);
|
||||
match tokio::fs::read(&path).await {
|
||||
Ok(bytes) => (StatusCode::OK, bytes).into_response(),
|
||||
Err(_) => (StatusCode::NOT_FOUND, "Файл не найден на диске").into_response(),
|
||||
let fail = |msg: &str, status: StatusCode| {
|
||||
state.fail_transfer(&tid, msg.to_string());
|
||||
remove_transfer_later(&state, tid.clone(), 10);
|
||||
AppError::new(status, msg)
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok(r) if r.status().is_success() => {
|
||||
state.finish_transfer(&tid);
|
||||
remove_transfer_later(&state, tid.clone(), 4);
|
||||
state.add_log(format!(
|
||||
"Файл '{}' ({} байт) доставлен на '{}'",
|
||||
display_name, total, peer_name
|
||||
));
|
||||
Ok(StatusCode::OK)
|
||||
}
|
||||
Ok(r) if r.status() == StatusCode::FORBIDDEN => Err(fail(
|
||||
"Получатель ещё не заверил ваше устройство",
|
||||
StatusCode::FORBIDDEN,
|
||||
)),
|
||||
Ok(r) => Err(fail(
|
||||
&format!("Получатель отклонил передачу (код {})", r.status()),
|
||||
StatusCode::BAD_GATEWAY,
|
||||
)),
|
||||
Err(_) => Err(fail(
|
||||
"Ошибка сети/TLS при передаче (узел недоступен, сертификат не совпал или обрыв)",
|
||||
StatusCode::BAD_GATEWAY,
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
// ───────────────────────── Раздача принятых файлов ─────────────────────────
|
||||
pub async fn ui_serve_saved_file(State(state): State<Arc<AppState>>, Path(filename): Path<String>) -> Response {
|
||||
let not_found = || (StatusCode::NOT_FOUND, "Файл не найден").into_response();
|
||||
if filename.is_empty()
|
||||
|| filename.starts_with('.')
|
||||
|| filename.contains(|c: char| matches!(c, '/' | '\\' | '\0'))
|
||||
{
|
||||
return not_found();
|
||||
}
|
||||
let path = state.media_dir.join(&filename);
|
||||
let Ok(meta) = tokio::fs::metadata(&path).await else { return not_found() };
|
||||
if !meta.is_file() {
|
||||
return not_found();
|
||||
}
|
||||
let Ok(file) = tokio::fs::File::open(&path).await else { return not_found() };
|
||||
|
||||
let inline = storage::is_previewable(&filename);
|
||||
// Content-Type определяем САМИ по расширению, а не по заявлению отправителя
|
||||
let ctype = if inline { storage::mime_from_name(&filename) } else { "application/octet-stream" };
|
||||
let disposition = format!(
|
||||
"{}; filename*=UTF-8''{}",
|
||||
if inline { "inline" } else { "attachment" },
|
||||
pct_encode(&filename)
|
||||
);
|
||||
|
||||
let mut resp = Response::new(Body::from_stream(ReaderStream::new(file)));
|
||||
let h = resp.headers_mut();
|
||||
h.insert(header::CONTENT_TYPE, HeaderValue::from_static(ctype));
|
||||
h.insert(header::CONTENT_LENGTH, HeaderValue::from(meta.len()));
|
||||
if let Ok(v) = HeaderValue::from_str(&disposition) {
|
||||
h.insert(header::CONTENT_DISPOSITION, v);
|
||||
}
|
||||
h.insert(
|
||||
header::CONTENT_SECURITY_POLICY,
|
||||
HeaderValue::from_static("default-src 'none'; sandbox"),
|
||||
);
|
||||
h.insert(
|
||||
HeaderName::from_static("cross-origin-resource-policy"),
|
||||
HeaderValue::from_static("same-origin"),
|
||||
);
|
||||
resp
|
||||
}
|
||||
Reference in new issue
Block a user