nineap/netrunner-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

loggin update

Browse Source
This commit is contained in:
nineAp
2026-05-02 18:55:37 +07:00
parent ebb123f4ea
commit 738fa8ca9b
18 changed files with 417 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/net/connection/connection.rs
+104 -34
View File
@@ -18,7 +18,9 @@ use crate::{
};
use bytes::{Bytes, BytesMut};
use dashmap::DashMap;
use netrunner_logger::{debug, error, info, warn};
use netrunner_logger::{
debug, error, info, warn, AppError, ERR_AUTH_FAILED, ERR_INFRA_TIMEOUT, ERR_NET_TLS_TAMPER,
};
use rand::Rng;
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
@@ -77,7 +79,7 @@ impl SessionManager {
#[async_trait::async_trait]
pub trait TunnelHandler {
async fn run(self) -> Result<(), String>;
async fn run(self) -> Result<(), AppError>;
}
pub struct Connection {
@@ -110,26 +112,38 @@ impl ClientHandler {
leg_id: u32,
muxer: Arc<Muxer>,
session_id: &str,
) -> Result<(), String> {
) -> Result<(), AppError> {
let leg_name = format!("TCP-Leg-{}", leg_id);
let addrs_future = tokio::net::lookup_host(remote_proxy_addr);
let mut addrs = tokio::time::timeout(std::time::Duration::from_secs(3), addrs_future)
.await
.map_err(|_| "DNS Lookup Timeout".to_string())?
.map_err(|e| format!("DNS resolution failed: {}", e))?;
.map_err(|_| {
AppError::new(ERR_INFRA_TIMEOUT, "Сервер недоступен", "DNS Lookup Timeout")
})?
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Ошибка DNS", e.to_string()))?;
let addr = addrs
.next()
.ok_or_else(|| format!("No IPs found for {}", remote_proxy_addr))?;
let addr = addrs.next().ok_or_else(|| {
AppError::new(
ERR_INFRA_TIMEOUT,
"Ошибка сети",
format!("No IPs found for {}", remote_proxy_addr),
)
})?;
let stream = tokio::time::timeout(
std::time::Duration::from_secs(5),
tokio::net::TcpStream::connect(addr),
)
.await
.map_err(|_| "Connection timeout".to_string())?
.map_err(|e| e.to_string())?;
.map_err(|_| {
AppError::new(
ERR_INFRA_TIMEOUT,
"Таймаут подключения",
"Connection timeout",
)
})?
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сокета", e.to_string()))?;
stream.set_nodelay(true).unwrap_or_default();
@@ -141,14 +155,12 @@ impl ClientHandler {
conn.outbound
.write_all(&ch)
.await
.map_err(|e| e.to_string())?;
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой сети", e.to_string()))?;
loop {
match TlsBridge::unpack_handshake(&mut conn.read_buf) {
Ok(Some(msg)) => {
session_keys
.update_keys(msg.random(), msg.extensions(), false)
.map_err(|e| format!("Keys update error: {}", e))?;
session_keys.update_keys(msg.random(), msg.extensions(), false)?;
break;
}
Ok(None) => {
@@ -157,15 +169,38 @@ impl ClientHandler {
conn.inbound.read_buf(&mut conn.read_buf),
)
.await;
match res {
Ok(Ok(0)) => return Err(format!("EOF on {}", leg_name)),
Ok(Ok(0)) => {
return Err(AppError::new(
ERR_INFRA_TIMEOUT,
"Разрыв соединения",
format!("EOF on {}", leg_name),
))
}
Ok(Ok(_)) => continue,
Ok(Err(e)) => return Err(format!("Read error: {}", e)),
Err(_) => return Err("Handshake read timeout".into()),
Ok(Err(e)) => {
return Err(AppError::new(
ERR_INFRA_TIMEOUT,
"Ошибка чтения",
e.to_string(),
))
}
Err(_) => {
return Err(AppError::new(
ERR_INFRA_TIMEOUT,
"Таймаут handshake",
"Handshake read timeout",
))
}
}
}
Err(e) => return Err(format!("TLS error on {}: {:?}", leg_name, e)),
Err(e) => {
return Err(AppError::new(
ERR_NET_TLS_TAMPER,
"Ошибка TLS",
format!("TLS error on {}: {:?}", leg_name, e.stage),
))
}
}
}
@@ -178,12 +213,18 @@ impl ClientHandler {
let auth_payload = Bytes::from(format!("{}:{}", session_id, leg_id));
let encrypted_auth = tx_codec
.encode_frame(0, FrameType::Heartbeat, auth_payload)
.map_err(|e| format!("Failed to encrypt Auth: {:?}", e))?;
.map_err(|e| {
AppError::new(
ERR_NET_TLS_TAMPER,
"Сбой шифрования",
format!("Failed to encrypt Auth: {:?}", e),
)
})?;
conn.outbound
.write_all(&encrypted_auth)
.await
.map_err(|e| e.to_string())?;
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Сбой отправки", e.to_string()))?;
let cap = NetworkConfig::global().channel_capacity;
let (control_tx, control_rx) = mpsc::channel::<MuxMessage>(cap);
@@ -207,18 +248,21 @@ impl ClientHandler {
};
let run_result = engine.run().await;
muxer.remove_leg(leg_id, &control_tx_clone);
run_result.map_err(|e| e.to_string())?;
Err(format!("{} Engine stopped", leg_name))
run_result?;
Err(AppError::new(
ERR_INFRA_TIMEOUT,
"Движок остановлен",
format!("{} Engine stopped", leg_name),
))
}
pub async fn connect(
remote_proxy_addr: &str,
mut rx_from_engine: mpsc::Receiver<RawCastFrame>,
tx_to_engine: mpsc::Sender<RawCastFrame>,
) -> Result<(), String> {
) -> Result<(), AppError> {
let session_id = SessionManager::generate_id();
let muxer = Arc::new(Muxer::new(true, session_id.clone()));
let registry: Arc<DashMap<u32, (u64, Ipv4Addr, u16, LocalProtocol)>> =
@@ -428,7 +472,7 @@ impl ServerHandler {
#[async_trait::async_trait]
impl TunnelHandler for ServerHandler {
async fn run(self) -> Result<(), String> {
async fn run(self) -> Result<(), AppError> {
info!("Acting as TLS Server with Stealth Fallback");
let Connection {
@@ -464,7 +508,13 @@ impl TunnelHandler for ServerHandler {
tokio::time::timeout(TLS_HELLO_TIMEOUT, inbound.read_buf(&mut read_buf))
.await;
match res {
Ok(Ok(0)) => return Err("Client closed".into()),
Ok(Ok(0)) => {
return Err(AppError::new(
ERR_INFRA_TIMEOUT,
"Клиент отключился",
"Client closed connection",
))
}
Ok(Ok(_)) => continue,
_ => {
warn!("⏰ TLS_HELLO_TIMEOUT reached. Triggering fallback...");
@@ -484,7 +534,7 @@ impl TunnelHandler for ServerHandler {
outbound
.write_all(&hello)
.await
.map_err(|e| e.to_string())?;
.map_err(|e| AppError::new(ERR_INFRA_TIMEOUT, "Ошибка отправки", e.to_string()))?;
let (tx_key, tx_iv, rx_key, rx_iv) = session_keys.get_aead_parameters();
let mut cipher = ChaChaCipher::new();
@@ -506,7 +556,11 @@ impl TunnelHandler for ServerHandler {
break (sid, lid);
}
}
return Err("Expected Auth payload in first Heartbeat frame".into());
return Err(AppError::new(
ERR_AUTH_FAILED,
"Ошибка авторизации",
"Expected Auth payload in first Heartbeat frame",
));
}
Ok(None) => {
let n = tokio::time::timeout(
@@ -514,15 +568,31 @@ impl TunnelHandler for ServerHandler {
inbound.read_buf(&mut read_buf),
)
.await
.map_err(|_| "Timeout waiting for Auth")?
.map_err(|e| e.to_string())?;
.map_err(|_| {
AppError::new(
ERR_INFRA_TIMEOUT,
"Таймаут авторизации",
"Timeout waiting for Auth",
)
})?
.map_err(|e| {
AppError::new(ERR_INFRA_TIMEOUT, "Ошибка сокета", e.to_string())
})?;
if n == 0 {
return Err("Client closed connection before Auth".into());
return Err(AppError::new(
ERR_AUTH_FAILED,
"Отказ",
"Client closed connection before Auth",
));
}
}
Err(e) => {
error!("❌ Secure Auth Failed: {:?}", e);
return Err("Dropped by security strategy (Auth Phase)".into());
error!("❌ Secure Auth Failed: {:?}", e.stage);
return Err(AppError::new(
ERR_AUTH_FAILED,
"Доступ запрещен",
"Dropped by security strategy (Auth Phase)",
));
}
}
};
core/src/net/connection/engine.rs
+56 -16
View File
@@ -1,13 +1,14 @@
use std::sync::Arc;
use bytes::{Bytes, BytesMut};
use netrunner_logger::{debug, error, info};
use netrunner_logger::{debug, error, info, AppError, ERR_INFRA_TIMEOUT, ERR_NET_TLS_TAMPER};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::tcp::{OwnedReadHalf, OwnedWriteHalf},
sync::mpsc::Receiver,
};
use tokio_util::sync::CancellationToken;
use tracing::instrument;
use crate::{
net::{
@@ -31,7 +32,8 @@ pub(crate) struct TunnelEngine {
}
impl TunnelEngine {
pub async fn run(self) -> Result<(), String> {
#[instrument(skip_all, fields(leg_id = self.leg_id))]
pub async fn run(self) -> Result<(), AppError> {
let inbound = self.inbound;
let outbound = self.outbound;
let read_buf = self.read_buf;
@@ -67,7 +69,13 @@ impl TunnelEngine {
break;
}
res = inbound.read_buf(&mut read_buf) => {
let n = res.map_err(|e| e.to_string())?;
let n = res.map_err(|e| {
netrunner_logger::AppError::new(
netrunner_logger::ERR_INFRA_TIMEOUT,
"Сбой сети",
e.to_string(),
)
})?;
if n == 0 {
if read_buf.is_empty() {
@@ -75,7 +83,11 @@ impl TunnelEngine {
} else {
error!("Connection abruptly closed by peer (Incomplete frame: {} bytes left)", read_buf.len());
}
return Err::<(), String>("EOF".into());
return Err(netrunner_logger::AppError::new(
netrunner_logger::ERR_INFRA_TIMEOUT,
"Соединение разорвано",
"EOF occurred during read",
));
}
muxer.record_leg_rx(leg_id, n as u64);
@@ -90,10 +102,18 @@ impl TunnelEngine {
if e.action == ErrorAction::Wait { break; }
if e.action == ErrorAction::Drop {
error!("CRITICAL: Crypto tampering or sync lost. Hard dropping tunnel!");
return Err("Crypto drop".into());
return Err(netrunner_logger::AppError::new(
netrunner_logger::ERR_NET_TLS_TAMPER,
"Критическая ошибка шифрования",
"Crypto drop",
));
}
error!(error = ?e, "Codec inbound failed");
return Err(format!("Codec error: {:?}", e));
error!(error = ?e, "Codec inbound failed");
return Err(netrunner_logger::AppError::new(
netrunner_logger::ERR_NET_TLS_TAMPER,
"Ошибка декодирования",
format!("Codec error: {:?}", e),
));
}
}
}
@@ -110,7 +130,7 @@ impl TunnelEngine {
}
}
}
Ok::<(), String>(())
Ok::<(), AppError>(())
});
let writer_handle = tokio::spawn(async move {
@@ -138,12 +158,20 @@ impl TunnelEngine {
}
}
}
Ok::<(), String>(())
Ok::<(), AppError>(())
});
let res = tokio::select! {
res = reader_handle => res.unwrap_or_else(|e| Err(format!("Reader panic: {}", e))),
res = writer_handle => res.unwrap_or_else(|e| Err(format!("Writer panic: {}", e))),
let res: Result<(), netrunner_logger::AppError> = tokio::select! {
res = reader_handle => res.unwrap_or_else(|e| Err(netrunner_logger::AppError::new(
netrunner_logger::ERR_SYS_PANIC,
"Сбой",
format!("Reader panic: {}", e)
))),
res = writer_handle => res.unwrap_or_else(|e| Err(netrunner_logger::AppError::new(
netrunner_logger::ERR_SYS_PANIC,
"Сбой",
format!("Writer panic: {}", e)
))),
};
token.cancel();
@@ -158,7 +186,7 @@ impl TunnelEngine {
outbound: &mut OwnedWriteHalf,
tx_codec: &mut TxCodec,
msg: MuxMessage,
) -> Result<(), String> {
) -> Result<(), AppError> {
let mut data = msg.data;
let stream_id = msg.stream_id;
let frame_type = msg.frame_type;
@@ -173,7 +201,11 @@ impl TunnelEngine {
Ok(pkt) => packets.push(pkt),
Err(e) => {
error!(stream_id, error = ?e, "Encryption failed for TCP chunk");
return Err(format!("Encryption error: {:?}", e));
return Err(netrunner_logger::AppError::new(
netrunner_logger::ERR_NET_TLS_TAMPER,
"Ошибка шифрования пакета",
format!("Encryption error: {:?}", e),
));
}
}
}
@@ -182,7 +214,11 @@ impl TunnelEngine {
Ok(pkt) => packets.push(pkt),
Err(e) => {
error!(stream_id, error = ?e, "Encryption failed for control/udp frame");
return Err(format!("Encryption error: {:?}", e));
return Err(netrunner_logger::AppError::new(
netrunner_logger::ERR_NET_TLS_TAMPER,
"Ошибка шифрования пакета",
format!("Encryption error: {:?}", e),
));
}
}
}
@@ -193,7 +229,11 @@ impl TunnelEngine {
tokio::time::timeout(std::time::Duration::from_secs(10), write_future).await
{
error!(stream_id, "🔥 Physical leg STUCK on write. Killing leg.");
return Err("Leg write timeout".into());
return Err(AppError::new(
ERR_INFRA_TIMEOUT,
"Таймаут отправки",
"Physical leg STUCK on write",
));
}
}
Ok(())
core/src/net/connection/muxer.rs
+12 -6
View File
@@ -1,6 +1,6 @@
use bytes::Bytes;
use dashmap::DashMap;
use netrunner_logger::{info, trace, warn};
use netrunner_logger::{AppError, ERR_INFRA_TIMEOUT, info, instrument, trace, warn};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
@@ -201,19 +201,24 @@ impl Muxer {
}
}
pub async fn send_to_network(&self, message: MuxMessage) -> Result<(), String> {
#[instrument(skip(self, message), fields(
session_id = %self.session_id,
stream_id = message.stream_id,
frame = ?message.frame_type
))]
pub async fn send_to_network(&self, message: MuxMessage) -> Result<(), AppError>{
let size = message.data.len() as u64;
let mut attempts = 0;
loop {
if attempts >= 3 {
return Err("MUXER: All attempts to send failed".to_string());
return Err(AppError::new(ERR_INFRA_TIMEOUT, "Сбой передачи данных", "MUXER: All attempts to send failed"));
}
attempts += 1;
let (leg_id, leg) = match self.select_leg(&message.frame_type, message.stream_id) {
Some(l) => l,
None => return Err("MUXER: No active physical legs available".to_string()),
None => return Err(AppError::new(ERR_INFRA_TIMEOUT, "Нет связи с сервером", "MUXER: No active physical legs available")),
};
let target_tx = match message.frame_type {
@@ -262,7 +267,7 @@ impl Muxer {
stream_id: u32,
data: Bytes,
is_udp: bool,
) -> Result<(), String> {
) -> Result<(), AppError> {
let frame_type = if is_udp {
FrameType::UdpData
} else {
@@ -281,7 +286,7 @@ impl Muxer {
stream_id: u32,
f_type: FrameType,
data: Bytes,
) -> Result<(), String> {
) -> Result<(), AppError>{
self.send_to_network(MuxMessage {
stream_id,
frame_type: f_type,
@@ -300,6 +305,7 @@ impl Muxer {
self.stream_bindings.remove(&stream_id);
}
#[instrument(skip(self, data), fields(session_id = %self.session_id, stream_id = stream_id))]
pub async fn dispatch_to_local(&self, stream_id: u32, data: Bytes) {
let stream_opt = self
.streams