Bluemangoo/IndPjsk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

cleanup

Browse Source
This commit is contained in:
Bluemangoo
2026-10-25 17:52:52 +08:00
parent eea818f5f4
commit 5ec7815a13
10 changed files with 361 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core_vpn/Cargo.lock
Generated
+16
View File
@@ -271,6 +271,7 @@ dependencies = [
"hyper-util",
"ipstack",
"jni",
"qstring",
"rcgen",
"regex",
"rustls",
@@ -973,6 +974,12 @@ dependencies = [
"serde_core",
]
[[package]]
name = "percent-encoding"
version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "pin-project-lite"
version = "0.2.17"
@@ -1011,6 +1018,15 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "qstring"
version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d464fae65fff2680baf48019211ce37aaec0c78e9264c84a3e484717f965104e"
dependencies = [
"percent-encoding",
]
[[package]]
name = "quote"
version = "1.0.47"
core_vpn/Cargo.toml
+1
View File
@@ -25,6 +25,7 @@ hyper = { version = "1.0", features = ["server", "client", "http1"] }
hyper-util = { version = "0.1", features = ["tokio", "server", "client"] }
http-body-util = "0.1"
http = "1.5.0"
qstring = "0.7.2"
[build-dependencies]
rcgen = { version = "0.14", features = ["x509-parser"] }
core_vpn/src/cert.rs
-2
View File
@@ -18,8 +18,6 @@ pub fn build_tls_acceptor() -> Arc<rustls::ServerConfig> {
.with_single_cert(cert_chain, key)
.unwrap();
// 真实服务器（dailygn / 拼锁 API）只支持 HTTP/1.1（ALPN 实测不会选 h2），
// 所以这里也只声明 http/1.1，让 App 用 HTTP/1.1 跟我们握手，避免跨协议转换。
config.alpn_protocols = vec![b"http/1.1".to_vec()];
Arc::new(config)
core_vpn/src/lib.rs
+1 -5
View File
@@ -1,6 +1,7 @@
mod cert;
mod mitm;
mod tun;
mod mut_qstring;
use crate::mitm::run_capture_async;
use jni::objects::{Global, JClass, JObject, JString};
@@ -46,9 +47,7 @@ fn log_with_env(env: &mut Env, label: &str, msg: &str) {
return;
};
// jni 0.21 中，GlobalRef 可能需要转成 JClass：
let jclass: &JClass = unsafe { &*(class_ref.as_obj() as *const JObject as *const JClass) };
// 或者更安全：let jclass = unsafe { JClass::from_raw(class_ref.as_obj().as_raw()) };
let _ = env.call_static_method(
jclass,
@@ -128,11 +127,8 @@ pub extern "system" fn Java_net_bluemangoo_indpjsk_vpn_VpnBridge_runCapture(
rt.block_on(async move { run_capture_async(tun_fd, hosts, re, out, cancel_token_clone).await });
cancel_token.cancel();
// 不要直接 drop runtime：万一有阻塞任务没退干净，drop 会无限等下去，
// JNI 调用就永远不返回（表现为按停止没反应）。设上限，超时就放弃。
rt.shutdown_timeout(std::time::Duration::from_secs(2));
// 抓到目标请求 -> 返回原始完整 URL（https://...?query）；否则 null
let url = result.unwrap_or_else(|e| {
error!("capture failed: {e:?}");
None
core_vpn/src/mitm.rs
+158 -168
View File
@@ -1,4 +1,5 @@
use crate::cert::build_tls_acceptor;
use crate::mut_qstring::MutQString;
use crate::tun::{BorrowedTunIo, RewindStream};
use crate::{error, info, warn};
use bytes::Bytes;
@@ -23,22 +24,11 @@ use tun_rs::BorrowedAsyncDevice;
static ROOT_STORE: OnceLock<Arc<RootCertStore>> = OnceLock::new();
/// 嗅探 ClientHello 的最长等待时间。超时后按「非目标」原样透传，
/// 这样「服务器先说话」的协议（客户端连上后不发数据）也不会被卡住。
const SNIFF_TIMEOUT: Duration = Duration::from_secs(5);
const DOH_BLACKLIST: &[&str] = &[
"dns.google",
"dns.google.com",
"cloudflare-dns.com",
"mozilla.cloudflare-dns.com",
"dns.quad9.net",
"dns.adguard.com",
"doh.opendns.com",
"dns.alidns.com", // 阿里 DoH
"doh.pub", // 腾讯 DoH
"dns.alidns.com",
];
const TEARDOWN_DRAIN_TIMEOUT: Duration = Duration::from_secs(3);
const TEARDOWN_SETTLE: Duration = Duration::from_millis(50);
const TEARDOWN_FIN_TIMEOUT: Duration = Duration::from_millis(300);
const TEARDOWN_GRACE: Duration = Duration::from_secs(4);
fn root_store() -> Arc<RootCertStore> {
ROOT_STORE
@@ -54,7 +44,6 @@ fn root_store() -> Arc<RootCertStore> {
load_system_ca_dirs(&mut root_store);
// 兜底：一个根证书都没有时用内置 Mozilla 列表，否则上游必然握手失败
if root_store.is_empty() {
root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
}
@@ -63,14 +52,12 @@ fn root_store() -> Arc<RootCertStore> {
.clone()
}
/// 读取系统 CA 目录（Android 的路径，其他平台不存在就走空）
fn load_system_ca_dirs(store: &mut RootCertStore) {
use std::io::BufReader;
const DIRS: &[&str] = &[
"/system/etc/security/cacerts", // Android 13-
"/apex/com.android.conscrypt/cacerts", // Android 14+
"/system/etc/security/cacerts", // 旧版 Android
"/data/misc/user/0/cacerts-added", // 用户自己装的 CA
];
for dir in DIRS {
@@ -88,10 +75,6 @@ fn load_system_ca_dirs(store: &mut RootCertStore) {
}
}
/// 嗅探 TLS ClientHello（不破坏字节流）。
///
/// 返回 `None` 表示对端不是 TLS（明文 HTTP、游戏自定义协议、超时……），
/// 调用方应当调用 `io.rewind()` 之后原样透传，而**不是**断开连接。
async fn sniff_client_hello<S>(
io: &mut RewindStream<S>,
) -> io::Result<Option<(Option<String>, rustls::server::Accepted)>>
@@ -108,10 +91,9 @@ where
Err(_) => return Ok(None), // 对端迟迟不发数据，当作非 TLS
};
if n == 0 {
return Ok(None); // 还没收到 ClientHello 就 EOF
return Ok(None);
}
// 只喂给 acceptor 一份拷贝，原始字节仍留在 RewindStream 的 history 里
if acceptor.read_tls(&mut &buf[..n]).is_err() {
return Ok(None);
}
@@ -140,86 +122,94 @@ async fn handle_tls_connection<S>(
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
let cancel_listen = cancel.clone();
let mut io = RewindStream::new(tcp);
let sniffed = match sniff_client_hello(&mut io).await {
Ok(v) => v,
Err(e) => {
warn!("client hello sniff error on {dst:?}: {e:?}");
return Ok(());
}
};
let Some((sni, accepted)) = sniffed else {
io.rewind();
io.stop_recording();
return passthrough_cancel_wrapped(io, dst, &cancel).await;
};
let host_matched = sni
.as_deref()
.map(|s| target_hosts.iter().any(|h| s.eq_ignore_ascii_case(h)))
.unwrap_or(false);
if !host_matched {
io.rewind();
io.stop_recording();
let r = passthrough_cancel_wrapped(io, dst, &cancel).await;
if r.is_err() {
warn!("tcp_passthrough error on {sni:?} {dst:?}");
}
return r;
}
info!("Encountered target host.");
io.stop_recording();
let client_tls = StartHandshake::from_parts(accepted, io)
.into_stream(server_config)
.await?;
let svc_host = sni.clone().unwrap_or_default();
let svc = hyper::service::service_fn(move |req: hyper::Request<Incoming>| {
let host = svc_host.clone();
let path = target_path.clone();
let save = save_path.clone();
let result_tx = result_tx.clone();
async move {
Ok::<_, std::convert::Infallible>(
forward_request(req, dst, host, path, save, result_tx).await,
)
}
});
let conn =
hyper::server::conn::http1::Builder::new().serve_connection(TokioIo::new(client_tls), svc);
tokio::pin!(conn);
let mut graceful = false;
loop {
tokio::select! {
res = conn.as_mut() => {
match res {
Ok(()) => {}
Err(e) => warn!("downstream http1 connection error: {e:?}"),
}
break;
}
_ = cancel.cancelled(), if !graceful => {
info!("teardown: graceful shutdown of downstream connection");
conn.as_mut().graceful_shutdown();
graceful = true;
}
}
}
Ok(())
}
async fn passthrough_cancel_wrapped<IO>(
io: IO,
dst: SocketAddr,
cancel: &CancellationToken,
) -> anyhow::Result<()>
where
IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
tokio::select! {
_ = cancel_listen.cancelled() => Ok(()),
res = async {
// 1. 先嗅探 ClientHello 的 SNI（可重放，不丢字节）
let mut io = RewindStream::new(tcp);
let sniffed = match sniff_client_hello(&mut io).await {
Ok(v) => v,
Err(e) => {
warn!("client hello sniff error on {dst:?}: {e:?}");
return Ok(());
}
};
// 不是 TLS：明文 HTTP / 自定义协议 / 服务器先说话，一律原样透传
let Some((sni, accepted)) = sniffed else {
io.rewind();
io.stop_recording();
return tcp_passthrough(io, dst).await.map(|_| ());
};
if let Some(ref sni_str) = sni
&& DOH_BLACKLIST.iter().any(|d| sni_str.eq_ignore_ascii_case(d)) {
return Ok(());
}
let host_matched = sni
.as_deref()
.map(|s| target_hosts.iter().any(|h| s.eq_ignore_ascii_case(h)))
.unwrap_or(false);
// 不是目标 host，原样透传（必须把已嗅探的 ClientHello 重放给上游）
if !host_matched {
io.rewind();
io.stop_recording();
let r=tcp_passthrough(io, dst).await.map(|_| ());
if r.is_err() {
warn!("tcp_passthrough error on {sni:?} {dst:?}");
}
return r;
}
info!(
"Encountered target host."
);
// 2. 与客户端完成 TLS（accepted 里已经缓存了 ClientHello，不能再重放）
io.stop_recording();
let client_tls = StartHandshake::from_parts(accepted, io)
.into_stream(server_config)
.await?;
// 3. 用 hyper 的 HTTP/1.1 服务端服务这条连接，
// 每个请求单独起一条上游 TLS 连接转发（HKDF 每请求）
let svc_host = sni.clone().unwrap_or_default();
let svc = hyper::service::service_fn(move |req: hyper::Request<Incoming>| {
let host = svc_host.clone();
let path = target_path.clone();
let save = save_path.clone();
let result_tx = result_tx.clone();
let cancel = cancel.clone();
async move {
Ok::<_, std::convert::Infallible>(
forward_request(req, dst, host, path, save, result_tx, cancel).await,
)
}
});
hyper::server::conn::http1::Builder::new()
.serve_connection(TokioIo::new(client_tls), svc)
.await?;
Ok(())
} => res
_ = cancel.cancelled() => Ok(()),
r = tcp_passthrough(io, dst) => r,
}
}
/// 转发单个请求到真实服务器，命中 target_path 时把响应体落盘。
///
/// 这里用「一个请求一条上游连接」的方式：目标是抓一个 API 响应，
/// 简单、不会踩 keep-alive / 多路复用状态机的坑。
#[allow(clippy::too_many_arguments)]
async fn forward_request(
req: hyper::Request<Incoming>,
@@ -228,20 +218,14 @@ async fn forward_request(
target_path: Regex,
save_path: String,
result_tx: Arc<Mutex<Option<tokio::sync::oneshot::Sender<String>>>>,
cancel: CancellationToken,
) -> hyper::Response<Full<Bytes>> {
match forward_request_inner(req, dst, &url_host, &target_path).await {
Ok((response, matched_body)) => {
if let Some((url, body)) = matched_body {
info!(
"Target matched! Downloading {} bytes",
body.len()
);
info!("Target matched! Downloading {} bytes", body.len());
let path = std::path::Path::new(&save_path);
if !path.is_absolute() {
error!(
"save_path 不是绝对路径（{save_path}）"
);
error!("Failed to resolve path: {save_path}");
}
if let Some(parent) = path.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
@@ -254,7 +238,6 @@ async fn forward_request(
}
Err(e) => error!("write {save_path} failed: {e:?}"),
}
cancel.cancel();
}
response
}
@@ -277,22 +260,19 @@ async fn forward_request_inner(
) -> anyhow::Result<(hyper::Response<Full<Bytes>>, Option<(String, Bytes)>)> {
let (mut parts, body) = req.into_parts();
// 和「模块」的 URL Rewrite 行为一致：把 MySekai 的增量刷新标志强制改成 True，
// 让服务端下发全量数据（游戏自己有时会发 False）。改完再匹配路径 / 转发 / 记 URL。
if let Some(rewritten) = force_full_mysekai(&parts.uri) {
warn!("rewrite mysekai flag: {} -> {}", parts.uri, rewritten);
parts.uri = rewritten;
}
let parts_uri = parts.uri.clone();
// 目标正则可能带 query（例如 (?\?isLogin=true)?$），所以要拿 path + query 去匹配
let path_and_query = parts_uri
.path_and_query()
.map(|p| p.as_str().to_string())
.unwrap_or_else(|| "/".to_string());
let body_bytes = body.collect().await?.to_bytes();
// 去掉逐跳头，避免把下游的 chunked/keep-alive 语义原样塞给上游
// 拒绝长链接，反正抓到就跑。
for h in [
"connection",
"keep-alive",
@@ -309,6 +289,7 @@ async fn forward_request_inner(
let mut client_config = rustls::ClientConfig::builder()
.with_root_certificates(root_store())
.with_no_client_auth();
// 问就是有服务器不接受 H2
client_config.alpn_protocols = vec![b"http/1.1".to_vec()];
let connector = tokio_rustls::TlsConnector::from(Arc::new(client_config));
let server_name = rustls::pki_types::ServerName::try_from(url_host.to_string())?;
@@ -341,8 +322,6 @@ async fn forward_request_inner(
}
let is_target = target_path.is_match(&path_and_query);
// 原始完整 URL（https://host + path + ?query）
let url = build_url(url_host, &parts_uri);
let save = if is_target {
@@ -356,7 +335,6 @@ async fn forward_request_inner(
))
}
/// 拼出 "https://host/path?query" 形式的完整 URL
fn build_url(host: &str, uri: &http::Uri) -> String {
if uri.scheme().is_some() && uri.authority().is_some() {
return uri.to_string();
@@ -365,21 +343,15 @@ fn build_url(host: &str, uri: &http::Uri) -> String {
format!("https://{host}{pq}")
}
/// `?isForceAllReloadOnlyMysekai=False` -> `=True`
///
/// 游戏在对 MySekai 做增量刷新时会带 False；模块里把它改写成 True 来触发全量下发。
/// 值已经是 True（或没这个参数）时返回 None，不做改动。
fn force_full_mysekai(uri: &http::Uri) -> Option<http::Uri> {
const KEY: &str = "isForceAllReloadOnlyMysekai=";
let pq = uri.path_and_query()?.as_str();
let start = pq.find(KEY)? + KEY.len();
let (head, rest) = pq.split_at(start);
let end = rest.find('&').unwrap_or(rest.len());
let (value, tail) = rest.split_at(end);
if !value.eq_ignore_ascii_case("false") {
return None;
if let q = Some(uri.query().is_some()) {
let mut query = qstring::QString::from(q);
if query.has("isForceAllReloadOnlyMysekai") {
query.set("isForceAllReloadOnlyMysekai", "true");
return Some(http::Uri::try_from(query.to_string().as_str()).ok()?);
}
}
http::Uri::try_from(format!("{head}True{tail}").as_str()).ok()
None
}
pub async fn run_capture_async(
@@ -403,21 +375,17 @@ pub async fn run_capture_async(
let (result_tx, mut result_rx) = tokio::sync::oneshot::channel::<String>();
let result_tx = Arc::new(Mutex::new(Some(result_tx)));
info!("VPN capture started.");
let result: anyhow::Result<Option<String>> = loop {
let token = cancel_token.clone();
tokio::select! {
_ = cancel_token.cancelled() => {
// 取消时如果刚好已经抓到了 URL，也带回去
let final_result = result_rx.try_recv().ok();
break Ok(final_result);
}
Ok(url) = &mut result_rx => {
cancel_token.cancel(); // 确保所有任务退出
cancel_token.cancel();
break Ok(Some(url));
}
// 定期清理已经结束的搬运任务，防止内存泄漏
_ = bridge_tasks.join_next(), if !bridge_tasks.is_empty() => {}
res = ip_stack.accept() => {
@@ -433,11 +401,10 @@ pub async fn run_capture_async(
IpStackStream::Udp(udp) => {
let dst = udp.peer_addr();
if dst.port() == 443 {
// 丢弃 UDP 443，使客户端的 QUIC 快速回退到 TCP
// drop quic roughly
drop(udp);
} else{
bridge_tasks.spawn(async move {
// info!("UDP passthrough: {:?}", dst);
if let Err(e) = udp_passthrough(udp, dst, token).await {
error!("udp passthrough error: {e:?}");
}
@@ -447,24 +414,28 @@ pub async fn run_capture_async(
IpStackStream::Tcp(mut tcp) => {
let dst = tcp.peer_addr();
if dst.port() == 443 {
// info!("Intercepting TCP 443: {:?}", dst);
//
// 用 duplex 替代手写的 mpsc 通道：duplex 自带正确的
// 反压与 waker 语义（旧实现 try_send 满了直接返回 Pending
// 却不注册 waker，连接会永久卡死，最后被协议栈 RST）。
let (app_io, mut tun_io) = tokio::io::duplex(256 * 1024);
// 在受借用保护的 join_set 启动读写泵
let pump_token = cancel_token.clone();
bridge_tasks.spawn(async move {
if let Err(e) =
tokio::io::copy_bidirectional(&mut tun_io, &mut tcp).await
{
// 连接被任意一端关闭时属正常现象
let _ = e;
tokio::select! {
_ = pump_token.cancelled() => {
let _ = tokio::time::timeout(
TEARDOWN_DRAIN_TIMEOUT,
tokio::io::copy(&mut tun_io, &mut tcp),
)
.await;
}
_ = tokio::io::copy_bidirectional(&mut tun_io, &mut tcp) => {}
}
tokio::time::sleep(TEARDOWN_SETTLE).await;
let _ = tokio::time::timeout(
TEARDOWN_FIN_TIMEOUT,
tcp.shutdown(),
)
.await;
});
// 独立 SPAWN 运行 TLS 业务逻辑（无借用约束）
let hosts = target_hosts.clone();
let path = target_path.clone();
let save = save_path.clone();
@@ -483,13 +454,14 @@ pub async fn run_capture_async(
token,
result_tx
).await {
Ok(()) => { /* 命中时已通过 result_tx 把 URL 送回主循环 */ }
Ok(()) => { }
Err(e) => error!("handle tls conn error: {e:?}"),
}
});
}else{
let pump_token = cancel_token.clone();
bridge_tasks.spawn(async move {
let _ = tcp_passthrough(tcp, dst).await;
let _ = tcp_passthrough_cancellable(tcp, dst, pump_token).await;
});
}
}
@@ -499,11 +471,10 @@ pub async fn run_capture_async(
}
};
// 关键：必须在 runtime 还活着的时候，把所有 bridge task（它们持有 ipstack 的
// IpStackTcpStream）彻底 drop 掉。
// 原因：ipstack 的 Drop 实现会 `block_in_place(...block_on(task_handle))` 等它自己的
// TCP 任务退出。如果拖到 runtime 析构阶段才 drop，worker 已经不跑了，就会永远等下去，
// 表现出来就是「按了停止没反应」。这里主动 abort + join，让 drop 发生在正常调度中。
let _ = tokio::time::timeout(TEARDOWN_GRACE, async {
while bridge_tasks.join_next().await.is_some() {}
})
.await;
bridge_tasks.abort_all();
while bridge_tasks.join_next().await.is_some() {}
@@ -514,20 +485,45 @@ async fn tcp_passthrough<IO>(mut client: IO, dst: SocketAddr) -> anyhow::Result<
where
IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
// info!("TCP passthrough to {dst}");
let mut server = tokio::net::TcpStream::connect(dst).await?;
tokio::io::copy_bidirectional(&mut client, &mut server).await?;
Ok(())
}
async fn tcp_passthrough_cancellable<IO>(
mut client: IO,
dst: SocketAddr,
cancel: CancellationToken,
) -> anyhow::Result<()>
where
IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
let mut server = tokio::net::TcpStream::connect(dst).await?;
tokio::select! {
_ = cancel.cancelled() => {
let _ = tokio::time::timeout(
TEARDOWN_DRAIN_TIMEOUT,
tokio::io::copy(&mut client, &mut server),
)
.await;
}
res = tokio::io::copy_bidirectional(&mut client, &mut server) => {
res?;
}
}
tokio::time::sleep(TEARDOWN_SETTLE).await;
let _ = tokio::time::timeout(TEARDOWN_FIN_TIMEOUT, client.shutdown()).await;
let _ = tokio::time::timeout(TEARDOWN_FIN_TIMEOUT, server.shutdown()).await;
Ok(())
}
async fn udp_passthrough<IO>(
client: IO,
dst: SocketAddr,
cancel: CancellationToken, // 把 cancel 传进来
cancel: CancellationToken,
) -> anyhow::Result<()>
where
IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
// 1. 绑定本地 socket
let bind_addr: SocketAddr = if dst.is_ipv4() {
"0.0.0.0:0".parse()?
} else {
@@ -536,10 +532,6 @@ where
let socket = UdpSocket::bind(bind_addr).await?;
// 2. Android 关键：一定要 protect，否则流量会重新进 TUN
// 如果你有 VpnService 的 protect 方法，在这里调用
// protect(socket.as_raw_fd());
socket.connect(dst).await?;
let socket = Arc::new(socket);
@@ -548,18 +540,16 @@ where
let mut client_buf = vec![0u8; 65535];
let mut server_buf = vec![0u8; 65535];
// 空闲超时（DNS 建议 8~15 秒，其他 UDP 可以更短）
let idle_timeout = Duration::from_secs(12);
loop {
tokio::select! {
// 外部取消
_ = cancel.cancelled() => break,
// client → server
// c2s
res = tokio::time::timeout(idle_timeout, cr.read(&mut client_buf)) => {
match res {
Ok(Ok(0)) | Err(_) => break, // EOF 或超时
Ok(Ok(0)) | Err(_) => break,
Ok(Ok(n)) => {
if socket.send(&client_buf[..n]).await.is_err() {
break;
@@ -569,7 +559,7 @@ where
}
}
// server → client
// s2c
res = tokio::time::timeout(idle_timeout, socket.recv(&mut server_buf)) => {
match res {
Ok(Ok(n)) if n > 0 => {
@@ -577,7 +567,7 @@ where
break;
}
}
_ => break, // 超时 / 错误 / 空包
_ => break,
}
}
}
core_vpn/src/mut_qstring.rs
+22
View File
@@ -0,0 +1,22 @@
use qstring::QString;
pub trait MutQString {
fn set<S: Into<String>, T: Into<String>>(&mut self, key: S, val: T);
}
impl MutQString for QString {
fn set<S: Into<String>, T: Into<String>>(&mut self, key: S, val: T) {
let key = key.into();
let val = val.into();
let mut pairs = std::mem::take(self).into_pairs();
if let Some((_, v)) = pairs.iter_mut().find(|(k, _)| k == &key) {
*v = val;
} else {
pairs.push((key, val));
}
*self = QString::new(pairs);
}
}
core_vpn/src/tun.rs
+1 -17
View File
@@ -4,7 +4,6 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
/// 包装你的 Borrowed 设备（无论类型是 BorrowedAsyncDevice 还是对其借用）
pub struct BorrowedTunIo<'a, D> {
device: &'a D,
packet_buf: Vec<u8>,
@@ -68,7 +67,6 @@ where
}
}
/// 抽象 trait，用于对接你的设备对象
pub trait DevicePoller {
fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>>;
fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>>;
@@ -93,20 +91,11 @@ impl<'b> DevicePoller for tun_rs::BorrowedAsyncDevice<'b> {
}
}
/// 可回退（可重放）的流包装器。
///
/// 用于在不破坏字节流的前提下嗅探 TLS ClientHello：
/// * 读过的字节会被记录在 `history` 里；
/// * `rewind()` 之后，下次读取会从 `history` 开头重新吐出来；
/// * `history` 读完之后继续从底层流读取并追加。
///
/// 这样即使最终决定「不是目标，原样透传」，上游依然能收到完整的 ClientHello。
pub struct RewindStream<S> {
inner: S,
history: Vec<u8>,
/// 已经交付给调用方的 `history` 字节数
/// 已交付
pos: usize,
/// history 最多记录多少字节
max_record: usize,
recording: bool,
}
@@ -122,13 +111,10 @@ impl<S> RewindStream<S> {
}
}
/// 回到流的起点，让后续读取重放已经嗅探过的字节
pub fn rewind(&mut self) {
self.pos = 0;
}
/// 决策完成后调用：不再缓存新字节（老字节仍会正常重放）。
/// 必须在确定「不会再 rewind」之后调用，否则会丢数据。
pub fn stop_recording(&mut self) {
self.recording = false;
self.history.shrink_to_fit();
@@ -157,8 +143,6 @@ where
return Pin::new(&mut this.inner).poll_read(cx, buf);
}
// 记录容量用尽：绝不能再从底层读，否则 rewind 会丢字节。
// 直接报错让上层放弃解析（此时的流量本来就不像 TLS）。
let room = this.max_record - this.history.len();
if buf.remaining() > room {
return Poll::Ready(Err(io::Error::new(