diff --git a/README.md b/README.md new file mode 100644 index 0000000..64fe9c8 --- /dev/null +++ b/README.md @@ -0,0 +1,18 @@ +# 咖喱烧烤 + +Rootless 开箱即用的 Project Sekai 用户数据抓包工具，支持上传到 Haruki。 + +## 使用 + +- 安装并打开应用 +- 如果需要上传，请在 [Haruki 工具箱](https://haruki.seiunx.com/upload-data) 中的 `iOS 模块` 方式里复制上传码并粘贴到应用中 +- 在右上角选择需要的配置 +- 可以取消勾选不需要的抓包步骤 +- 点击开始，如果跳出建立 VPN 的提示，请允许 +- 启动游戏，进入游戏触发需要抓的包，应用会自动抓取数据并自动结束，如果配置了上传会自动上传 + +## 触发位置 + +- Suite: 登录步骤后，进入游戏的加载阶段（下载资源之前） +- 烤森: 进入烤森 +- 烤森生日双叶: 我不知道 \ No newline at end of file diff --git a/app/build.gradle.kts b/app/build.gradle.kts index cf57ab4..41fa61c 100644 --- a/app/build.gradle.kts +++ b/app/build.gradle.kts @@ -1,8 +1,13 @@ +import java.util.Properties +import org.gradle.internal.os.OperatingSystem + plugins { id("com.android.application") id("org.jetbrains.kotlin.plugin.compose") } +val minSdk = 28 + android { namespace = "net.bluemangoo.indpjsk" compileSdk { @@ -11,10 +16,10 @@ android { defaultConfig { applicationId = "net.bluemangoo.indpjsk" - minSdk = 28 + minSdk = minSdk targetSdk = 37 versionCode = 1 - versionName = "1.0" + versionName = "1.1" } @@ -43,4 +48,128 @@ dependencies { implementation("androidx.compose.ui:ui-tooling-preview") implementation("androidx.compose.material3:material3") debugImplementation("androidx.compose.ui:ui-tooling") +} + +val cargoTarget = "aarch64-linux-android" +val cargoSoName = "libcore_vpn.so" +val cargoDir = rootProject.file("core_vpn") +val cargoSo = File(cargoDir, "target/$cargoTarget/release/$cargoSoName") +val jniLibsArm64 = file("src/main/jniLibs/arm64-v8a") +val cargoBin: String = providers.gradleProperty("cargoBin").getOrElse("cargo") + +fun resolveNdkDirectory(project: Project): File { + val localPropertiesFile = project.rootProject.file("local.properties") + val properties = Properties() + if (localPropertiesFile.exists()) { + localPropertiesFile.inputStream().use { properties.load(it) } + } + + properties.getProperty("ndk.dir")?.let { path -> + val file = File(path) + if (file.exists()) return file + } + + val sdkDirPath = properties.getProperty("sdk.dir") + ?: System.getenv("ANDROID_SDK_ROOT") + ?: System.getenv("ANDROID_HOME") + ?: throw GradleException("Failed to resolve Android SDK path") + + val ndkParentDir = File(sdkDirPath, "ndk") + if (!ndkParentDir.exists() || !ndkParentDir.isDirectory) { + throw GradleException("Failed to resolve NDK directory: ${ndkParentDir.absolutePath}") + } + + val installedVersions = ndkParentDir.listFiles { f -> f.isDirectory } + ?.sortedWith(compareBy { it.name }) + ?: emptyList() + + if (installedVersions.isEmpty()) { + throw GradleException("Failed to find any installed NDK versions in ${ndkParentDir.absolutePath}") + } + + // 取最新安装的版本 + return installedVersions.last() +} + +fun resolveLlvmBinDirectory(ndkDir: File): File { + val prebuiltDir = File(ndkDir, "toolchains/llvm/prebuilt") + val hostDir = prebuiltDir.listFiles { file -> file.isDirectory }?.firstOrNull() + ?: throw IllegalStateException("Ndk llvm not found: ${prebuiltDir.absolutePath}") + + val binDir = File(hostDir, "bin") + if (!binDir.exists()) { + throw IllegalStateException("Ndk llvm not found: ${binDir.absolutePath}") + } + return binDir +} + +val cargoBuildVpnCore = tasks.register("cargoBuildVpnCore") { + group = "build" + description = "Rust build core_vpn" + + workingDir = cargoDir + doFirst { + try { + val ndkDir = resolveNdkDirectory(project) + val ndkHomePath = ndkDir.absolutePath + val ndkBin = resolveLlvmBinDirectory(ndkDir) + val scriptExt = if (OperatingSystem.current().isWindows) ".cmd" else "" + val execExt = if (OperatingSystem.current().isWindows) ".exe" else "" + + environment("NDK_HOME", ndkHomePath) + environment("PATH", "$ndkHomePath:${System.getenv("PATH")}") + environment( + "CC_aarch64-linux-android", + ndkBin.resolve("aarch64-linux-android$minSdk-clang$scriptExt").absolutePath + ) + environment( + "CXX_aarch64-linux-android", + ndkBin.resolve("aarch64-linux-android$minSdk-clang++$scriptExt").absolutePath + ) + environment("AR_aarch64-linux-android", ndkBin.resolve("llvm-ar$execExt").absolutePath) + environment( + "CARGO_TARGET_AARCH64_LINUX_ANDROID_LINKER", + ndkBin.resolve("aarch64-linux-android$minSdk-clang$scriptExt").absolutePath + ) + } catch (e: Exception) { + logger.warn("[cargoBuildVpnCore] resolveNdkDirectory failed: ${e.message}") + } + } + commandLine(cargoBin, "build", "--release", "--target", cargoTarget) + + inputs.files( + File(cargoDir, "Cargo.toml"), + File(cargoDir, "Cargo.lock"), + File(cargoDir, "build.rs"), + ).withPropertyName("rustConfig") + inputs.dir(File(cargoDir, "src")).withPropertyName("rustSrc") + + outputs.file(cargoSo).withPropertyName("rustSo") +} + +val copyVpnSo = tasks.register("copyVpnSo") { + group = "build" + description = "Sync core_vpn.so" + + dependsOn(cargoBuildVpnCore) + + inputs.file(cargoSo).withPropertyName("rustSo") + outputs.dir(jniLibsArm64).withPropertyName("jniLibsArm64") + + doLast { + check(cargoSo.isFile) { + "File $cargoSo not exist" + } + if (jniLibsArm64.exists()) { + jniLibsArm64.deleteRecursively() + } + jniLibsArm64.mkdirs() + val dest = File(jniLibsArm64, cargoSoName) + cargoSo.copyTo(dest, overwrite = true) + logger.lifecycle("[copyVpnSo] ${cargoSo.absolutePath} -> ${dest.absolutePath}") + } +} + +tasks.named("preBuild") { + dependsOn(copyVpnSo) } \ No newline at end of file diff --git a/app/src/main/java/net/bluemangoo/indpjsk/data/PredefinedProfiles.kt b/app/src/main/java/net/bluemangoo/indpjsk/data/PredefinedProfiles.kt index 1b1637c..1e50983 100644 --- a/app/src/main/java/net/bluemangoo/indpjsk/data/PredefinedProfiles.kt +++ b/app/src/main/java/net/bluemangoo/indpjsk/data/PredefinedProfiles.kt @@ -31,6 +31,14 @@ object PredefinedProfiles { fileName = "mysekai" ) + private fun mysekaiBirthday(hosts: List) = CaptureProfileStep( + id = "mysekai_birthday", + name = "烤森生日双叶", + hosts = hosts, + path = "/api/user/(\\d+)/mysekai/birthday-party/(\\d+)/delivery", + fileName = "mysekai_birthday" + ) + private val CN_HOSTS = listOf("mkcn-prod-public-60001-1.dailygn.com") private val JP_HOSTS = listOf("production-game-api.sekai.colorfulpalette.org") private val EN_HOSTS = listOf("n-production-game-api.sekai-en.com") @@ -44,7 +52,7 @@ object PredefinedProfiles { ) val all: List = listOf( - // 国服不需要烤森 + // 国服不支持烤森 CaptureProfile( id = "cn", name = "国服", @@ -53,22 +61,22 @@ object PredefinedProfiles { CaptureProfile( id = "jp", name = "日服", - steps = listOf(suiteStep(JP_HOSTS), mysekaiStep(JP_HOSTS)) + steps = listOf(suiteStep(JP_HOSTS), mysekaiStep(JP_HOSTS), mysekaiBirthday(JP_HOSTS)) ), CaptureProfile( id = "en", name = "国际服", - steps = listOf(suiteStep(EN_HOSTS), mysekaiStep(EN_HOSTS)) + steps = listOf(suiteStep(EN_HOSTS), mysekaiStep(EN_HOSTS), mysekaiBirthday(EN_HOSTS)) ), CaptureProfile( id = "tw", name = "台服", - steps = listOf(suiteStep(TW_HOSTS), mysekaiStep(TW_HOSTS)) + steps = listOf(suiteStep(TW_HOSTS), mysekaiStep(TW_HOSTS), mysekaiBirthday(TW_HOSTS)) ), CaptureProfile( id = "kr", name = "韩服", - steps = listOf(suiteStep(KR_HOSTS), mysekaiStep(KR_HOSTS)) + steps = listOf(suiteStep(KR_HOSTS), mysekaiStep(KR_HOSTS), mysekaiBirthday(KR_HOSTS)) ), ) diff --git a/core_vpn/Cargo.lock b/core_vpn/Cargo.lock index cfcc050..72d96bc 100644 --- a/core_vpn/Cargo.lock +++ b/core_vpn/Cargo.lock @@ -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" diff --git a/core_vpn/Cargo.toml b/core_vpn/Cargo.toml index 48c68b6..3982887 100644 --- a/core_vpn/Cargo.toml +++ b/core_vpn/Cargo.toml @@ -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"] } diff --git a/core_vpn/src/cert.rs b/core_vpn/src/cert.rs index 18257e3..92c282e 100644 --- a/core_vpn/src/cert.rs +++ b/core_vpn/src/cert.rs @@ -18,8 +18,6 @@ pub fn build_tls_acceptor() -> Arc { .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) diff --git a/core_vpn/src/lib.rs b/core_vpn/src/lib.rs index 64fd68b..2b334d0 100644 --- a/core_vpn/src/lib.rs +++ b/core_vpn/src/lib.rs @@ -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 diff --git a/core_vpn/src/mitm.rs b/core_vpn/src/mitm.rs index 9b7b4c8..0f9538a 100644 --- a/core_vpn/src/mitm.rs +++ b/core_vpn/src/mitm.rs @@ -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> = 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 { ROOT_STORE @@ -54,7 +44,6 @@ fn root_store() -> Arc { 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 { .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( io: &mut RewindStream, ) -> io::Result, 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( 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| { + 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, + 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| { - 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, @@ -228,20 +218,14 @@ async fn forward_request( target_path: Regex, save_path: String, result_tx: Arc>>>, - cancel: CancellationToken, ) -> hyper::Response> { 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>, 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 { - 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::(); let result_tx = Arc::new(Mutex::new(Some(result_tx))); - info!("VPN capture started."); - let result: anyhow::Result> = 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(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( + 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( 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, } } } diff --git a/core_vpn/src/mut_qstring.rs b/core_vpn/src/mut_qstring.rs new file mode 100644 index 0000000..7e81d74 --- /dev/null +++ b/core_vpn/src/mut_qstring.rs @@ -0,0 +1,22 @@ +use qstring::QString; + +pub trait MutQString { + fn set, T: Into>(&mut self, key: S, val: T); +} + +impl MutQString for QString { + fn set, T: Into>(&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); + } +} \ No newline at end of file diff --git a/core_vpn/src/tun.rs b/core_vpn/src/tun.rs index 9eeb037..559271c 100644 --- a/core_vpn/src/tun.rs +++ b/core_vpn/src/tun.rs @@ -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, @@ -68,7 +67,6 @@ where } } -/// 抽象 trait，用于对接你的设备对象 pub trait DevicePoller { fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll>; fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll>; @@ -93,20 +91,11 @@ impl<'b> DevicePoller for tun_rs::BorrowedAsyncDevice<'b> { } } -/// 可回退（可重放）的流包装器。 -/// -/// 用于在不破坏字节流的前提下嗅探 TLS ClientHello： -/// * 读过的字节会被记录在 `history` 里； -/// * `rewind()` 之后，下次读取会从 `history` 开头重新吐出来； -/// * `history` 读完之后继续从底层流读取并追加。 -/// -/// 这样即使最终决定「不是目标，原样透传」，上游依然能收到完整的 ClientHello。 pub struct RewindStream { inner: S, history: Vec, - /// 已经交付给调用方的 `history` 字节数 + /// 已交付 pos: usize, - /// history 最多记录多少字节 max_record: usize, recording: bool, } @@ -122,13 +111,10 @@ impl RewindStream { } } - /// 回到流的起点，让后续读取重放已经嗅探过的字节 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(