Добавить опциональную авторизацию клиента и динамические лимиты трафика
Сервер получает флаги --require-auth/--backend-url (по умолчанию выключено, поведение уже развёрнутых нод не меняется). Auth-кадр хендшейка расширен с "session_id:leg_id" до "session_id:leg_id:token"; при включённом --require-auth сервер валидирует токен через новый AuthValidator (BackendClient к netrunner-backend, с кешем на 60с). Muxer теперь ведёт монотонный учёт трафика сессии (переживает реконнект ног — раньше total_bytes мог занижаться после переподключения) и раз в ~30с отчитывается бэкенду; при превышении лимита сессия разрывается. Токен клиента прокидывается через EngineConfig/TunnelConfig до Tauri-плагина (desktop.rs + Android Kotlin-плагин). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,37 @@
|
||||
//! Абстракция проверки клиентского токена и учёта расхода трафика на
|
||||
//! control-plane бэкенде. Ядро знает только этот трейт — конкретную реализацию
|
||||
//! (HTTP-клиент к `netrunner-backend`) даёт связывающий бинарь (`netrunner-server`),
|
||||
//! чтобы ядро не тянуло HTTP-клиент как обязательную зависимость.
|
||||
//!
|
||||
//! Включается/выключается на инстанс целиком через `--require-auth` (см.
|
||||
//! `server/src/main.rs`): если сервер запущен без флага, [`ServerHandler`]
|
||||
//! вообще не спрашивает валидатор, и поведение не отличается от того, что
|
||||
//! было до этой фичи.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use netrunner_logger::AppError;
|
||||
|
||||
/// Результат успешной проверки токена клиента.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct UserQuota {
|
||||
pub user_id: String,
|
||||
/// `None` — безлимит.
|
||||
pub limit_bytes: Option<u64>,
|
||||
pub used_bytes: u64,
|
||||
}
|
||||
|
||||
/// Ответ на отчёт о расходе трафика.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct UsageReport {
|
||||
pub used_bytes: u64,
|
||||
pub limit_bytes: Option<u64>,
|
||||
pub over_limit: bool,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait AuthValidator: Send + Sync {
|
||||
/// Проверяет Bearer-токен клиента (JWT, выданный бэкендом при логине).
|
||||
async fn validate(&self, token: &str) -> Result<UserQuota, AppError>;
|
||||
/// Отчитывается о переданных байтах и синхронно узнаёт, не превышен ли лимит.
|
||||
async fn report_usage(&self, user_id: &str, delta_bytes: u64) -> Result<UsageReport, AppError>;
|
||||
}
|
||||
@@ -254,13 +254,15 @@ impl ClientHandler {
|
||||
/// Шаги: послать поддельный `ClientHello` (профиль браузера по `profile`,
|
||||
/// SNI=`decoy_sni`) → дождаться `ServerHello` и вывести ключи → зарядить
|
||||
/// шифр и кодек → отправить первый зашифрованный auth-кадр `Heartbeat` с
|
||||
/// `"session_id:leg_id"`. Возвращает половинки сокета и готовые кодеки.
|
||||
/// `"session_id:leg_id:auth_token"` (третий сегмент может быть пустым).
|
||||
/// Возвращает половинки сокета и готовые кодеки.
|
||||
pub(crate) async fn perform_handshake(
|
||||
stream: tokio::net::TcpStream,
|
||||
session_id: &str,
|
||||
leg_id: u32,
|
||||
profile: &BrowserProfile,
|
||||
decoy_sni: &str,
|
||||
auth_token: &str,
|
||||
) -> Result<
|
||||
(
|
||||
OwnedReadHalf,
|
||||
@@ -346,7 +348,10 @@ impl ClientHandler {
|
||||
let codec = Codec::new(cipher, session_keys.get_auth_key());
|
||||
let (rx_codec, mut tx_codec) = codec.split();
|
||||
|
||||
let auth_payload = Bytes::from(format!("{}:{}", session_id, leg_id));
|
||||
// Третий сегмент — Bearer-токен клиента (пусто, если `--require-auth`
|
||||
// выключен на этом развёртывании или приложение ещё не залогинено).
|
||||
// Сервер игнорирует его целиком, если сам не запущен с `--require-auth`.
|
||||
let auth_payload = Bytes::from(format!("{}:{}:{}", session_id, leg_id, auth_token));
|
||||
let encrypted_auth = tx_codec
|
||||
.encode_frame(0, FrameType::Heartbeat, auth_payload)
|
||||
.map_err(|e| {
|
||||
@@ -383,6 +388,7 @@ impl ClientHandler {
|
||||
muxer: Arc<Muxer>,
|
||||
session_id: &str,
|
||||
decoy_sni: &Arc<str>,
|
||||
auth_token: &Arc<str>,
|
||||
) -> Result<(), AppError> {
|
||||
let leg_name = format!("TCP-Leg-{}", leg_id);
|
||||
|
||||
@@ -426,7 +432,8 @@ impl ClientHandler {
|
||||
|
||||
let profile = BrowserProfile::for_session(session_id);
|
||||
let (inbound, outbound, rx_codec, tx_codec) =
|
||||
Self::perform_handshake(stream, session_id, leg_id, profile, decoy_sni).await?;
|
||||
Self::perform_handshake(stream, session_id, leg_id, profile, decoy_sni, auth_token)
|
||||
.await?;
|
||||
|
||||
let cap = NetworkConfig::global().channel_capacity;
|
||||
let (control_tx, control_rx) = mpsc::channel::<MuxMessage>(cap);
|
||||
@@ -451,6 +458,7 @@ impl ClientHandler {
|
||||
session_id: session_id.to_string(),
|
||||
leg_status: crate::net::connection::engine::LegStatus::Active,
|
||||
decoy_sni: decoy_sni.clone(),
|
||||
auth_token: auth_token.clone(),
|
||||
};
|
||||
|
||||
let run_result = engine.run().await;
|
||||
@@ -487,10 +495,12 @@ impl ClientHandler {
|
||||
pub async fn connect(
|
||||
remote_proxy_addr: &str,
|
||||
decoy_sni: impl Into<Arc<str>>,
|
||||
auth_token: Option<String>,
|
||||
mut rx_from_engine: mpsc::Receiver<RawCastFrame>,
|
||||
tx_to_engine: mpsc::Sender<RawCastFrame>,
|
||||
) -> Result<Arc<Muxer>, AppError> {
|
||||
let decoy_sni: Arc<str> = decoy_sni.into();
|
||||
let auth_token: Arc<str> = auth_token.unwrap_or_default().into();
|
||||
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)>> =
|
||||
@@ -524,12 +534,14 @@ impl ClientHandler {
|
||||
let m = muxer.clone();
|
||||
let sid = session_id.clone();
|
||||
let decoy_sni = decoy_sni.clone();
|
||||
let auth_token = auth_token.clone();
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(LEG_STAGGER_DELAY * id).await;
|
||||
let mut attempt: u32 = 0;
|
||||
loop {
|
||||
if let Err(e) =
|
||||
Self::establish_leg(&addr, id, m.clone(), &sid, &decoy_sni).await
|
||||
Self::establish_leg(&addr, id, m.clone(), &sid, &decoy_sni, &auth_token)
|
||||
.await
|
||||
{
|
||||
attempt += 1;
|
||||
error!("Leg {} disconnected: {}. Reconnecting in 2s...", id, e);
|
||||
@@ -845,6 +857,10 @@ pub struct ServerHandler {
|
||||
/// Домен-декой для stealth-fallback (атрибут ноды, задаётся при старте
|
||||
/// сервера через `--decoy-host`; раньше был захардкожен на `ubuntu.com`).
|
||||
pub(crate) decoy_host: Arc<str>,
|
||||
/// `None` — авторизация выключена на этом инстансе (`--require-auth` не
|
||||
/// передан), поведение как до этой фичи. `Some` — токен клиента
|
||||
/// обязателен и проверяется бэкендом при установке первой ноги сессии.
|
||||
pub(crate) auth: Option<Arc<dyn crate::net::AuthValidator>>,
|
||||
}
|
||||
|
||||
impl ServerHandler {
|
||||
@@ -852,11 +868,13 @@ impl ServerHandler {
|
||||
connection: Connection,
|
||||
session_manager: Arc<SessionManager>,
|
||||
decoy_host: Arc<str>,
|
||||
auth: Option<Arc<dyn crate::net::AuthValidator>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
conn: connection,
|
||||
session_manager,
|
||||
decoy_host,
|
||||
auth,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1067,17 +1085,18 @@ impl TunnelHandler for ServerHandler {
|
||||
let codec = Codec::new(cipher, session_keys.get_auth_key());
|
||||
let (mut rx_codec, tx_codec) = codec.split();
|
||||
|
||||
let (session_id, leg_id) = loop {
|
||||
let (session_id, leg_id, auth_token) = loop {
|
||||
match rx_codec.decode_inbound(&mut read_buf) {
|
||||
Ok(Some(frame)) => {
|
||||
if frame.header.frame_type == FrameType::Heartbeat {
|
||||
let payload_str = std::str::from_utf8(&frame.payload).unwrap_or("");
|
||||
let parts: Vec<&str> = payload_str.split(':').collect();
|
||||
if parts.len() == 2 && parts[1].parse::<u32>().is_ok() {
|
||||
let parts: Vec<&str> = payload_str.splitn(3, ':').collect();
|
||||
if parts.len() == 3 && parts[1].parse::<u32>().is_ok() {
|
||||
let sid = parts[0].to_string();
|
||||
let lid: u32 = parts[1].parse().unwrap();
|
||||
let token = parts[2].to_string();
|
||||
info!("🤝 Secure Auth verified! Session: {}, Leg: {}", sid, lid);
|
||||
break (sid, lid);
|
||||
break (sid, lid, token);
|
||||
}
|
||||
}
|
||||
return Err(AppError::new(
|
||||
@@ -1122,6 +1141,19 @@ impl TunnelHandler for ServerHandler {
|
||||
};
|
||||
|
||||
let muxer = self.session_manager.get_or_create(&session_id);
|
||||
|
||||
// Проверка личности клиента у бэкенда — только если этот инстанс
|
||||
// запущен с `--require-auth`. До этой точки соединение прошло
|
||||
// Netrunner-хендшейк (не сканер/чужой TLS-клиент), поэтому отказ здесь
|
||||
// — обычный разрыв, а не stealth-fallback (светить уже нечего).
|
||||
if let Some(validator) = &self.auth {
|
||||
let quota = validator.validate(&auth_token).await.map_err(|e| {
|
||||
warn!("❌ Backend rejected client token: {}", e.internal_msg);
|
||||
AppError::new(ERR_AUTH_FAILED, "Доступ запрещен", e.internal_msg)
|
||||
})?;
|
||||
muxer.set_quota_user(quota.user_id);
|
||||
}
|
||||
|
||||
let cap = NetworkConfig::global().channel_capacity;
|
||||
let (control_tx, control_rx) = mpsc::channel::<MuxMessage>(cap);
|
||||
let (data_tx, data_rx) = mpsc::channel::<MuxMessage>(cap);
|
||||
@@ -1152,8 +1184,9 @@ impl TunnelHandler for ServerHandler {
|
||||
session_id,
|
||||
leg_status: crate::net::connection::engine::LegStatus::Active,
|
||||
// Сервер никогда не реконнектит (см. `attempt_reconnect`'s early
|
||||
// return on empty `remote_addr`), поэтому SNI здесь не используется.
|
||||
// return on empty `remote_addr`), поэтому SNI/токен здесь не используются.
|
||||
decoy_sni: Arc::from(""),
|
||||
auth_token: Arc::from(""),
|
||||
};
|
||||
|
||||
let res = engine.run().await;
|
||||
@@ -1326,6 +1359,7 @@ mod tests {
|
||||
conn,
|
||||
Arc::new(SessionManager::new()),
|
||||
Arc::from("example.com"),
|
||||
None,
|
||||
);
|
||||
// run() продолжает в muxer/engine после хендшейка и вернётся сам,
|
||||
// как только клиент закроет сокет (наш тест-клиент не шлёт
|
||||
@@ -1343,6 +1377,7 @@ mod tests {
|
||||
0,
|
||||
BrowserProfile::for_session(&session_id),
|
||||
&Arc::<str>::from("example.com"),
|
||||
"",
|
||||
),
|
||||
)
|
||||
.await
|
||||
@@ -1376,6 +1411,7 @@ mod tests {
|
||||
conn,
|
||||
Arc::new(SessionManager::new()),
|
||||
Arc::from("this-host-does-not-resolve.invalid"),
|
||||
None,
|
||||
);
|
||||
handler.run().await
|
||||
});
|
||||
|
||||
@@ -85,6 +85,9 @@ pub(crate) struct TunnelEngine {
|
||||
/// [`ClientHandler::connect`](crate::net::connection::ClientHandler::connect));
|
||||
/// нужен для внутреннего реконнекта в [`attempt_reconnect`](Self::attempt_reconnect).
|
||||
pub decoy_sni: Arc<str>,
|
||||
/// Bearer-токен клиента (пусто — авторизация выключена/не залогинен),
|
||||
/// нужен для того же внутреннего реконнекта, что и `decoy_sni` выше.
|
||||
pub auth_token: Arc<str>,
|
||||
}
|
||||
|
||||
impl TunnelEngine {
|
||||
@@ -132,6 +135,7 @@ impl TunnelEngine {
|
||||
self.leg_id,
|
||||
profile,
|
||||
&self.decoy_sni,
|
||||
&self.auth_token,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
@@ -299,6 +299,19 @@ pub struct Muxer {
|
||||
/// so a burst of new streams spreads across legs instead of all binding to
|
||||
/// the single current-best one (thundering herd).
|
||||
rr_counter: Arc<AtomicU32>,
|
||||
/// Байты ног, которые уже отцеплены (реконнект/эвикт) — без этого
|
||||
/// `total_bytes()` был бы не монотонным: у новой ноги счётчик стартует с
|
||||
/// нуля, и частые переподключения занижали бы расход трафика для лимитов
|
||||
/// (см. `total_bytes`/`fold_removed_leg`).
|
||||
cumulative_tx: Arc<AtomicU64>,
|
||||
cumulative_rx: Arc<AtomicU64>,
|
||||
/// Идентификатор юзера-владельца сессии для отчёта о расходе трафика
|
||||
/// прокси-серверу (`None` — авторизация выключена или это клиентская
|
||||
/// сторона муксера, отчёты о трафике шлёт только сервер).
|
||||
quota_user_id: Arc<ArcSwap<Option<String>>>,
|
||||
/// Сколько байт (tx+rx) уже было отчитано бэкенду по этой сессии —
|
||||
/// следующий тик репортит только дельту сверх этого значения.
|
||||
quota_reported_bytes: Arc<AtomicU64>,
|
||||
}
|
||||
|
||||
impl Muxer {
|
||||
@@ -313,11 +326,65 @@ impl Muxer {
|
||||
pending_pings: Arc::new(DashMap::new()),
|
||||
session_id: Arc::new(session_id),
|
||||
rr_counter: Arc::new(AtomicU32::new(0)),
|
||||
cumulative_tx: Arc::new(AtomicU64::new(0)),
|
||||
cumulative_rx: Arc::new(AtomicU64::new(0)),
|
||||
quota_user_id: Arc::new(ArcSwap::from_pointee(None)),
|
||||
quota_reported_bytes: Arc::new(AtomicU64::new(0)),
|
||||
};
|
||||
muxer.spawn_backlog_reaper();
|
||||
muxer
|
||||
}
|
||||
|
||||
/// Складывает байты ноги, которая уходит из `legs` (эвикт/реконнект), в
|
||||
/// сессионный кумулятивный счётчик — вызывать сразу после `DashMap::remove`.
|
||||
fn fold_removed_leg(&self, leg: &MuxLeg) {
|
||||
self.cumulative_tx
|
||||
.fetch_add(leg.stats.tx_bytes.load(Ordering::Relaxed), Ordering::Relaxed);
|
||||
self.cumulative_rx
|
||||
.fetch_add(leg.stats.rx_bytes.load(Ordering::Relaxed), Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Суммарный трафик сессии (все ноги, включая уже отцепленные) — источник
|
||||
/// истины для отчётов о расходе прокси-серверу бэкенду.
|
||||
pub fn total_bytes(&self) -> (u64, u64) {
|
||||
let mut tx = self.cumulative_tx.load(Ordering::Relaxed);
|
||||
let mut rx = self.cumulative_rx.load(Ordering::Relaxed);
|
||||
for leg in self.active_legs_cache.load_full().iter() {
|
||||
tx += leg.stats.tx_bytes.load(Ordering::Relaxed);
|
||||
rx += leg.stats.rx_bytes.load(Ordering::Relaxed);
|
||||
}
|
||||
(tx, rx)
|
||||
}
|
||||
|
||||
/// Привязывает сессию к юзеру бэкенда — вызывается один раз при успешной
|
||||
/// проверке auth-токена первой ноги сессии (см. `ServerHandler::run`).
|
||||
pub fn set_quota_user(&self, user_id: String) {
|
||||
self.quota_user_id.store(Arc::new(Some(user_id)));
|
||||
}
|
||||
|
||||
pub fn quota_user_id(&self) -> Option<String> {
|
||||
self.quota_user_id.load_full().as_ref().clone()
|
||||
}
|
||||
|
||||
/// Дельта трафика с прошлого репорта и (не блокирующий) сдвиг базовой
|
||||
/// точки — вызывающий обязан либо реально отправить дельту бэкенду, либо
|
||||
/// не звать этот метод (в отличие от `store`, здесь нет отмены на ошибку:
|
||||
/// невозможность связаться с бэкендом не должна накапливать неограниченно
|
||||
/// растущую "недоотчитанную" дельту).
|
||||
pub fn take_usage_delta(&self) -> u64 {
|
||||
let (tx, rx) = self.total_bytes();
|
||||
let total = tx + rx;
|
||||
let last = self.quota_reported_bytes.swap(total, Ordering::Relaxed);
|
||||
total.saturating_sub(last)
|
||||
}
|
||||
|
||||
/// Откатывает базовую точку назад на `delta` — вызывать, если репорт
|
||||
/// бэкенду не удался, чтобы не потерять дельту навсегда.
|
||||
pub fn rollback_usage_delta(&self, delta: u64) {
|
||||
self.quota_reported_bytes
|
||||
.fetch_sub(delta.min(self.quota_reported_bytes.load(Ordering::Relaxed)), Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Фоновый "ридер" бэклогов: единственное место, которое реально закрывает
|
||||
/// поток за зависший бэклог (см. докстринг [`StreamBacklog`]). Никогда не
|
||||
/// вызывается изнутри `dispatch_to_local` — работает по расписанию, вне
|
||||
@@ -443,7 +510,9 @@ impl Muxer {
|
||||
.get(&leg_id)
|
||||
.map_or(false, |leg| leg.control_tx.same_channel(tx));
|
||||
if should_remove {
|
||||
self.legs.remove(&leg_id);
|
||||
if let Some((_, leg)) = self.legs.remove(&leg_id) {
|
||||
self.fold_removed_leg(&leg);
|
||||
}
|
||||
// Unbind streams BEFORE refreshing the cache so a concurrent
|
||||
// select_leg never re-binds a stream to the leg we are evicting.
|
||||
self.clear_bindings_for_leg(leg_id);
|
||||
@@ -457,7 +526,8 @@ impl Muxer {
|
||||
|
||||
/// Безусловно удаляет ногу (без сверки канала) — при выходе её движка.
|
||||
pub fn force_remove_leg(&self, leg_id: u32) {
|
||||
if self.legs.remove(&leg_id).is_some() {
|
||||
if let Some((_, leg)) = self.legs.remove(&leg_id) {
|
||||
self.fold_removed_leg(&leg);
|
||||
self.clear_bindings_for_leg(leg_id);
|
||||
self.update_legs_cache();
|
||||
info!(leg_id, "MUXER: TCP leg force-removed on engine exit");
|
||||
@@ -466,6 +536,9 @@ impl Muxer {
|
||||
|
||||
/// Сбрасывает все ноги и привязки (полная остановка туннеля).
|
||||
pub fn remove_all_legs(&self) {
|
||||
for entry in self.legs.iter() {
|
||||
self.fold_removed_leg(entry.value());
|
||||
}
|
||||
self.legs.clear();
|
||||
self.stream_bindings.clear();
|
||||
self.update_legs_cache();
|
||||
@@ -1119,8 +1192,7 @@ impl Muxer {
|
||||
self.session_id
|
||||
));
|
||||
|
||||
let mut total_tx = 0;
|
||||
let mut total_rx = 0;
|
||||
let (total_tx, total_rx) = self.total_bytes();
|
||||
let mut legs_info = Vec::new();
|
||||
|
||||
let cached_legs = self.active_legs_cache.load_full();
|
||||
@@ -1128,8 +1200,6 @@ impl Muxer {
|
||||
let tx = leg.stats.tx_bytes.load(Ordering::Relaxed);
|
||||
let rx = leg.stats.rx_bytes.load(Ordering::Relaxed);
|
||||
let rtt = leg.stats.rtt_ms.load(Ordering::Relaxed);
|
||||
total_tx += tx;
|
||||
total_rx += rx;
|
||||
|
||||
let rtt_str = if rtt == 0 {
|
||||
"N/A".to_string()
|
||||
|
||||
@@ -35,11 +35,13 @@
|
||||
//! хендлеры/менеджер сессий (через `connection`) и [`Muxer`] с глобальной оценкой
|
||||
//! [`GLOBAL_MIN_RTT`], а также все константы.
|
||||
|
||||
mod auth;
|
||||
mod config;
|
||||
mod connection;
|
||||
mod constants;
|
||||
pub mod diagnostics;
|
||||
|
||||
pub use auth::{AuthValidator, UsageReport, UserQuota};
|
||||
pub use config::NetworkConfig;
|
||||
pub use connection::{
|
||||
ClientHandler, Connection, Muxer, ServerHandler, SessionManager, TunnelHandler, GLOBAL_MIN_RTT,
|
||||
|
||||
Reference in New Issue
Block a user