1.0 能用

2026-10-24 14:48:59 +08:00
commit eea818f5f4
48 changed files with 4802 additions and 0 deletions
@@ -0,0 +1,587 @@
+use crate::cert::build_tls_acceptor;
+use crate::tun::{BorrowedTunIo, RewindStream};
+use crate::{error, info, warn};
+use bytes::Bytes;
+use http_body_util::{BodyExt, Full};
+use hyper::body::Incoming;
+use hyper_util::rt::TokioIo;
+use ipstack::IpStackStream;
+use regex::Regex;
+use rustls::RootCertStore;
+use std::io;
+use std::net::SocketAddr;
+use std::os::fd::RawFd;
+use std::sync::{Arc, OnceLock};
+use std::time::Duration;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::net::UdpSocket;
+use tokio::sync::Mutex;
+use tokio::task::JoinSet;
+use tokio_rustls::StartHandshake;
+use tokio_util::sync::CancellationToken;
+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",
+];
+
+fn root_store() -> Arc<RootCertStore> {
+    ROOT_STORE
+        .get_or_init(|| {
+            let mut root_store = RootCertStore::empty();
+            let native_certs = rustls_native_certs::load_native_certs();
+            for err in native_certs.errors {
+                warn!("加载证书警告: {err}");
+            }
+            for cert in native_certs.certs {
+                let _ = root_store.add(cert);
+            }
+
+            load_system_ca_dirs(&mut root_store);
+
+            // 兜底：一个根证书都没有时用内置 Mozilla 列表，否则上游必然握手失败
+            if root_store.is_empty() {
+                root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
+            }
+            Arc::new(root_store)
+        })
+        .clone()
+}
+
+/// 读取系统 CA 目录（Android 的路径，其他平台不存在就走空）
+fn load_system_ca_dirs(store: &mut RootCertStore) {
+    use std::io::BufReader;
+
+    const DIRS: &[&str] = &[
+        "/apex/com.android.conscrypt/cacerts", // Android 14+
+        "/system/etc/security/cacerts",        // 旧版 Android
+        "/data/misc/user/0/cacerts-added",     // 用户自己装的 CA
+    ];
+
+    for dir in DIRS {
+        let Ok(entries) = std::fs::read_dir(dir) else {
+            continue;
+        };
+        for entry in entries.flatten() {
+            let Ok(file) = std::fs::File::open(entry.path()) else {
+                continue;
+            };
+            for cert in rustls_pemfile::certs(&mut BufReader::new(file)).flatten() {
+                let _ = store.add(cert);
+            }
+        }
+    }
+}
+
+/// 嗅探 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)>>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    let mut acceptor = rustls::server::Acceptor::default();
+    let mut buf = vec![0u8; 16 * 1024];
+
+    loop {
+        let n = match tokio::time::timeout(SNIFF_TIMEOUT, io.read(&mut buf)).await {
+            Ok(Ok(n)) => n,
+            Ok(Err(e)) => return Err(e),
+            Err(_) => return Ok(None), // 对端迟迟不发数据，当作非 TLS
+        };
+        if n == 0 {
+            return Ok(None); // 还没收到 ClientHello 就 EOF
+        }
+
+        // 只喂给 acceptor 一份拷贝，原始字节仍留在 RewindStream 的 history 里
+        if acceptor.read_tls(&mut &buf[..n]).is_err() {
+            return Ok(None);
+        }
+        match acceptor.accept() {
+            Ok(Some(accepted)) => {
+                let sni = accepted.client_hello().server_name().map(|s| s.to_string());
+                return Ok(Some((sni, accepted)));
+            }
+            Ok(None) => continue,      // 还没凑齐
+            Err(_) => return Ok(None), // 不是合法 TLS
+        }
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn handle_tls_connection<S>(
+    tcp: S,
+    dst: SocketAddr,
+    target_hosts: Vec<String>,
+    target_path: Regex,
+    save_path: String,
+    server_config: Arc<rustls::ServerConfig>,
+    cancel: CancellationToken,
+    result_tx: Arc<Mutex<Option<tokio::sync::oneshot::Sender<String>>>>,
+) -> anyhow::Result<()>
+where
+    S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+{
+    let cancel_listen = cancel.clone();
+
+    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
+    }
+}
+
+/// 转发单个请求到真实服务器，命中 target_path 时把响应体落盘。
+///
+/// 这里用「一个请求一条上游连接」的方式：目标是抓一个 API 响应，
+/// 简单、不会踩 keep-alive / 多路复用状态机的坑。
+#[allow(clippy::too_many_arguments)]
+async fn forward_request(
+    req: hyper::Request<Incoming>,
+    dst: SocketAddr,
+    url_host: String,
+    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()
+                );
+                let path = std::path::Path::new(&save_path);
+                if !path.is_absolute() {
+                    error!(
+                        "save_path 不是绝对路径（{save_path}）"
+                    );
+                }
+                if let Some(parent) = path.parent() {
+                    let _ = tokio::fs::create_dir_all(parent).await;
+                }
+                match tokio::fs::write(path, &body).await {
+                    Ok(()) => {
+                        if let Some(tx) = result_tx.lock().await.take() {
+                            let _ = tx.send(url);
+                        }
+                    }
+                    Err(e) => error!("write {save_path} failed: {e:?}"),
+                }
+                cancel.cancel();
+            }
+            response
+        }
+        Err(e) => {
+            error!("forward to {dst} failed: {e:?}");
+            hyper::Response::builder()
+                .status(hyper::StatusCode::BAD_GATEWAY)
+                .header("content-type", "text/plain; charset=utf-8")
+                .body(Full::new(Bytes::from(format!("mitm forward error: {e}"))))
+                .unwrap()
+        }
+    }
+}
+
+async fn forward_request_inner(
+    req: hyper::Request<Incoming>,
+    dst: SocketAddr,
+    url_host: &str,
+    target_path: &Regex,
+) -> 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",
+        "proxy-connection",
+        "transfer-encoding",
+        "upgrade",
+        "te",
+        "trailer",
+        "content-length",
+    ] {
+        parts.headers.remove(h);
+    }
+
+    let mut client_config = rustls::ClientConfig::builder()
+        .with_root_certificates(root_store())
+        .with_no_client_auth();
+    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())?;
+    let tcp_to_server = tokio::net::TcpStream::connect(dst).await?;
+    let server_tls = connector.connect(server_name, tcp_to_server).await?;
+
+    let (mut sender, conn) =
+        hyper::client::conn::http1::handshake::<_, Full<Bytes>>(TokioIo::new(server_tls)).await?;
+    tokio::spawn(async move {
+        if let Err(e) = conn.await {
+            warn!("upstream http1 connection error: {e:?}");
+        }
+    });
+
+    let upstream_req = hyper::Request::from_parts(parts, Full::new(body_bytes));
+    let resp = sender.send_request(upstream_req).await?;
+
+    let (mut res_parts, res_body) = resp.into_parts();
+    let res_bytes = res_body.collect().await?.to_bytes();
+    for h in [
+        "connection",
+        "keep-alive",
+        "proxy-connection",
+        "transfer-encoding",
+        "upgrade",
+        "te",
+        "trailer",
+    ] {
+        res_parts.headers.remove(h);
+    }
+
+    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 {
+        Some((url, res_bytes.clone()))
+    } else {
+        None
+    };
+    Ok((
+        hyper::Response::from_parts(res_parts, Full::new(res_bytes)),
+        save,
+    ))
+}
+
+/// 拼出 "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();
+    }
+    let pq = uri.path_and_query().map(|p| p.as_str()).unwrap_or("/");
+    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;
+    }
+    http::Uri::try_from(format!("{head}True{tail}").as_str()).ok()
+}
+
+pub async fn run_capture_async(
+    tun_fd: i32,
+    target_hosts: Vec<String>,
+    target_path: Regex,
+    save_path: String,
+    cancel_token: CancellationToken,
+) -> anyhow::Result<Option<String>> {
+    let device = unsafe { BorrowedAsyncDevice::borrow_raw(tun_fd as RawFd)? };
+
+    let mut ipstack_config = ipstack::IpStackConfig::default();
+    ipstack_config.mtu(1500)?;
+    let device_ref: &BorrowedAsyncDevice<'static> = unsafe { std::mem::transmute(&device) };
+    let tun_io = BorrowedTunIo::new(device_ref);
+    let mut ip_stack = ipstack::IpStack::new(ipstack_config, tun_io);
+    let mut bridge_tasks = JoinSet::new();
+
+    let tls_acceptor = build_tls_acceptor();
+
+    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(); // 确保所有任务退出
+                break Ok(Some(url));
+            }
+            // 定期清理已经结束的搬运任务，防止内存泄漏
+            _ = bridge_tasks.join_next(), if !bridge_tasks.is_empty() => {}
+
+            res = ip_stack.accept() => {
+                let stream = match res {
+                    Ok(s) => s,
+                    Err(e) => {
+                        error!("ip_stack accept error (device might be dropped): {e:?}");
+                        break Err(anyhow::Error::from(e));
+                    }
+                };
+
+                match stream {
+                    IpStackStream::Udp(udp) => {
+                        let dst = udp.peer_addr();
+                        if dst.port() == 443 {
+                            // 丢弃 UDP 443，使客户端的 QUIC 快速回退到 TCP
+                            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:?}");
+                                }
+                            });
+                        }
+                    }
+                    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 启动读写泵
+                            bridge_tasks.spawn(async move {
+                                if let Err(e) =
+                                    tokio::io::copy_bidirectional(&mut tun_io, &mut tcp).await
+                                {
+                                    // 连接被任意一端关闭时属正常现象
+                                    let _ = e;
+                                }
+                            });
+
+                            // 独立 SPAWN 运行 TLS 业务逻辑（无借用约束）
+                            let hosts = target_hosts.clone();
+                            let path = target_path.clone();
+                            let save = save_path.clone();
+                            let server_config = tls_acceptor.clone();
+                            let token = cancel_token.clone();
+                            let result_tx = result_tx.clone();
+
+                            tokio::spawn(async move {
+                                match handle_tls_connection(
+                                    app_io,
+                                    dst,
+                                    hosts,
+                                    path,
+                                    save,
+                                    server_config,
+                                    token,
+                                    result_tx
+                                ).await {
+                                    Ok(()) => { /* 命中时已通过 result_tx 把 URL 送回主循环 */ }
+                                    Err(e) => error!("handle tls conn error: {e:?}"),
+                                }
+                            });
+                        }else{
+                            bridge_tasks.spawn(async move {
+                                let _ = tcp_passthrough(tcp, dst).await;
+                            });
+                        }
+                    }
+                    _ => {}
+                }
+            }
+        }
+    };
+
+    // 关键：必须在 runtime 还活着的时候，把所有 bridge task（它们持有 ipstack 的
+    // IpStackTcpStream）彻底 drop 掉。
+    // 原因：ipstack 的 Drop 实现会 `block_in_place(...block_on(task_handle))` 等它自己的
+    // TCP 任务退出。如果拖到 runtime 析构阶段才 drop，worker 已经不跑了，就会永远等下去，
+    // 表现出来就是「按了停止没反应」。这里主动 abort + join，让 drop 发生在正常调度中。
+    bridge_tasks.abort_all();
+    while bridge_tasks.join_next().await.is_some() {}
+
+    result
+}
+
+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 udp_passthrough<IO>(
+    client: IO,
+    dst: SocketAddr,
+    cancel: CancellationToken, // 把 cancel 传进来
+) -> 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 {
+        "[::]:0".parse()?
+    };
+
+    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);
+
+    let (mut cr, mut cw) = tokio::io::split(client);
+
+    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
+            res = tokio::time::timeout(idle_timeout, cr.read(&mut client_buf)) => {
+                match res {
+                    Ok(Ok(0)) | Err(_) => break,          // EOF 或超时
+                    Ok(Ok(n)) => {
+                        if socket.send(&client_buf[..n]).await.is_err() {
+                            break;
+                        }
+                    }
+                    Ok(Err(_)) => break,
+                }
+            }
+
+            // server → client
+            res = tokio::time::timeout(idle_timeout, socket.recv(&mut server_buf)) => {
+                match res {
+                    Ok(Ok(n)) if n > 0 => {
+                        if cw.write_all(&server_buf[..n]).await.is_err() {
+                            break;
+                        }
+                    }
+                    _ => break,   // 超时 / 错误 / 空包
+                }
+            }
+        }
+    }
+
+    Ok(())
+}