细化+上游代理+1.1

core_vpn/src/mitm.rs

@@ -14,6 +14,7 @@ use std::net::SocketAddr;
 use std::os::fd::RawFd;
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
+use http::Uri;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tokio::net::UdpSocket;
 use tokio::sync::Mutex;
@@ -112,6 +113,7 @@ where
 async fn handle_tls_connection<S>(
     tcp: S,
     dst: SocketAddr,
+    upstream: Upstream,
     target_hosts: Vec<String>,
     target_path: Regex,
     save_path: String,
@@ -134,7 +136,8 @@ where
     let Some((sni, accepted)) = sniffed else {
         io.rewind();
         io.stop_recording();
-        return passthrough_cancel_wrapped(io, dst, &cancel).await;
+        // 非 TLS：拿不到 SNI，走代理时只能按 IP CONNECT
+        return passthrough_cancel_wrapped(io, dst, None, upstream, &cancel).await;
     };
 
     let host_matched = sni
@@ -145,7 +148,7 @@ where
     if !host_matched {
         io.rewind();
         io.stop_recording();
-        let r = passthrough_cancel_wrapped(io, dst, &cancel).await;
+        let r = passthrough_cancel_wrapped(io, dst, sni.as_deref(), upstream, &cancel).await;
         if r.is_err() {
             warn!("tcp_passthrough error on {sni:?} {dst:?}");
         }
@@ -159,14 +162,16 @@ where
         .await?;
 
     let svc_host = sni.clone().unwrap_or_default();
+    let svc_upstream = upstream;
     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 upstream = svc_upstream.clone();
         async move {
             Ok::<_, std::convert::Infallible>(
-                forward_request(req, dst, host, path, save, result_tx).await,
+                forward_request(req, dst, host, path, save, result_tx, upstream).await,
             )
         }
     });
@@ -199,6 +204,8 @@ where
 async fn passthrough_cancel_wrapped<IO>(
     io: IO,
     dst: SocketAddr,
+    sni: Option<&str>,
+    upstream: Upstream,
     cancel: &CancellationToken,
 ) -> anyhow::Result<()>
 where
@@ -206,7 +213,7 @@ where
 {
     tokio::select! {
         _ = cancel.cancelled() => Ok(()),
-        r = tcp_passthrough(io, dst) => r,
+        r = tcp_passthrough(io, dst, sni, upstream) => r,
     }
 }
 
@@ -218,8 +225,9 @@ async fn forward_request(
     target_path: Regex,
     save_path: String,
     result_tx: Arc<Mutex<Option<tokio::sync::oneshot::Sender<String>>>>,
+    upstream: Upstream,
 ) -> hyper::Response<Full<Bytes>> {
-    match forward_request_inner(req, dst, &url_host, &target_path).await {
+    match forward_request_inner(req, dst, &url_host, &target_path, &upstream).await {
         Ok((response, matched_body)) => {
             if let Some((url, body)) = matched_body {
                 info!("Target matched! Downloading {} bytes", body.len());
@@ -257,6 +265,7 @@ async fn forward_request_inner(
     dst: SocketAddr,
     url_host: &str,
     target_path: &Regex,
+    upstream: &Upstream,
 ) -> anyhow::Result<(hyper::Response<Full<Bytes>>, Option<(String, Bytes)>)> {
     let (mut parts, body) = req.into_parts();
 
@@ -293,7 +302,7 @@ async fn forward_request_inner(
     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 tcp_to_server = upstream.connect(dst, Some(url_host)).await?;
     let server_tls = connector.connect(server_name, tcp_to_server).await?;
 
     let (mut sender, conn) =
@@ -335,7 +344,7 @@ async fn forward_request_inner(
     ))
 }
 
-fn build_url(host: &str, uri: &http::Uri) -> String {
+fn build_url(host: &str, uri: &Uri) -> String {
     if uri.scheme().is_some() && uri.authority().is_some() {
         return uri.to_string();
     }
@@ -343,15 +352,155 @@ fn build_url(host: &str, uri: &http::Uri) -> String {
     format!("https://{host}{pq}")
 }
 
-fn force_full_mysekai(uri: &http::Uri) -> Option<http::Uri> {
-    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()?);
+/// 把 MySekai 的增量刷新标志强制改成 True（和模块的 URL Rewrite 一致），
+/// 其它 query 参数原样保留。没有这个参数、或已经是 True 时返回 None。
+fn force_full_mysekai(uri: &Uri) -> Option<Uri> {
+    const KEY: &str = "isForceAllReloadOnlyMysekai";
+    let mut query = qstring::QString::from(uri.query()?);
+    if !query.has(KEY) || query.get(KEY) == Some("True") {
+        return None;
+    }
+    query.set(KEY, "True");
+    Uri::try_from(format!("{}?{}", uri.path(), query).as_str()).ok()
+}
+
+
+#[derive(Clone, Debug)]
+pub struct ProxyConfig {
+    pub tls: bool,
+    pub host: String,
+    pub port: u16,
+    pub auth: Option<String>,
+}
+
+impl ProxyConfig {
+    pub fn parse(input: &str) -> anyhow::Result<Option<Self>> {
+        let raw = input.trim();
+        if raw.is_empty() {
+            return Ok(None);
+        }
+        let url = url::Url::parse(raw)
+            .map_err(|e| anyhow::anyhow!("代理地址不是合法 URL: {raw} ({e})"))?;
+        let tls = match url.scheme() {
+            "http" => false,
+            "https" => true,
+            other => anyhow::bail!("代理只支持 http/https，当前是 {other}://"),
+        };
+        let host = url
+            .host_str()
+            .filter(|h| !h.is_empty())
+            .ok_or_else(|| anyhow::anyhow!("代理地址缺少 host: {raw}"))?
+            .to_string();
+        let port = url
+            .port_or_known_default()
+            .ok_or_else(|| anyhow::anyhow!("代理地址缺少端口: {raw}"))?;
+        let auth = if url.username().is_empty() {
+            None
+        } else {
+            let pass = url.password().unwrap_or("");
+            Some(base64_encode(format!("{}:{}", url.username(), pass).as_bytes()))
+        };
+        Ok(Some(Self { tls, host, port, auth }))
+    }
+}
+
+fn base64_encode(data: &[u8]) -> String {
+    use base64::Engine as _;
+    base64::engine::general_purpose::STANDARD.encode(data)
+}
+
+type UpstreamIo = Box<dyn AsyncReadWrite + Send + Unpin>;
+
+trait AsyncReadWrite: AsyncRead + AsyncWrite {}
+impl<T: AsyncRead + AsyncWrite + ?Sized> AsyncReadWrite for T {}
+
+#[derive(Clone)]
+struct Upstream {
+    proxy: Option<Arc<ProxyConfig>>,
+}
+
+impl Upstream {
+    fn new(proxy: Option<ProxyConfig>) -> Self {
+        Self { proxy: proxy.map(Arc::new) }
+    }
+
+    async fn connect(&self, dst: SocketAddr, host: Option<&str>) -> anyhow::Result<UpstreamIo> {
+        match &self.proxy {
+            None => {
+                let tcp = tokio::net::TcpStream::connect(dst).await?;
+                tcp.set_nodelay(true).ok();
+                Ok(Box::new(tcp))
+            }
+            Some(proxy) => connect_via_proxy(proxy, dst, host).await,
         }
     }
-    None
+}
+
+async fn connect_via_proxy(
+    proxy: &ProxyConfig,
+    dst: SocketAddr,
+    host: Option<&str>,
+) -> anyhow::Result<UpstreamIo> {
+    let tcp = tokio::net::TcpStream::connect((proxy.host.as_str(), proxy.port))
+        .await
+        .map_err(|e| anyhow::anyhow!("连接代理 {}:{} 失败: {e}", proxy.host, proxy.port))?;
+    tcp.set_nodelay(true).ok();
+
+    let mut io: UpstreamIo = if proxy.tls {
+        let mut cfg = rustls::ClientConfig::builder()
+            .with_root_certificates(root_store())
+            .with_no_client_auth();
+        cfg.alpn_protocols = vec![b"http/1.1".to_vec()];
+        let connector = tokio_rustls::TlsConnector::from(Arc::new(cfg));
+        let name = rustls::pki_types::ServerName::try_from(proxy.host.clone())?;
+        Box::new(
+            connector
+                .connect(name, tcp)
+                .await
+                .map_err(|e| anyhow::anyhow!("与代理建立 TLS 失败: {e}"))?,
+        )
+    } else {
+        Box::new(tcp)
+    };
+
+    let authority = match host {
+        Some(h) if !h.is_empty() => format!("{h}:{}", dst.port()),
+        _ => dst.to_string(),
+    };
+    let mut req = format!("CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\n");
+    if let Some(auth) = &proxy.auth {
+        req.push_str(&format!("Proxy-Authorization: Basic {auth}\r\n"));
+    }
+    req.push_str("\r\n");
+    io.write_all(req.as_bytes()).await?;
+    io.flush().await?;
+
+    let mut head = Vec::with_capacity(256);
+    let mut byte = [0u8; 1];
+    loop {
+        let n = io.read(&mut byte).await?;
+        if n == 0 {
+            anyhow::bail!("代理在 CONNECT 阶段直接断开");
+        }
+        head.push(byte[0]);
+        if head.ends_with(b"\r\n\r\n") {
+            break;
+        }
+        if head.len() > 16 * 1024 {
+            anyhow::bail!("代理 CONNECT 响应头过长");
+        }
+    }
+    let head_str = String::from_utf8_lossy(&head);
+    let status_line = head_str.lines().next().unwrap_or("").trim().to_string();
+    let ok = status_line
+        .split_whitespace()
+        .nth(1)
+        .map(|code| code == "200")
+        .unwrap_or(false);
+    if !ok {
+        anyhow::bail!("代理 CONNECT 失败: {status_line}");
+    }
+    Ok(io)
 }
 
 pub async fn run_capture_async(
@@ -359,6 +508,7 @@ pub async fn run_capture_async(
     target_hosts: Vec<String>,
     target_path: Regex,
     save_path: String,
+    proxy: Option<ProxyConfig>,
     cancel_token: CancellationToken,
 ) -> anyhow::Result<Option<String>> {
     let device = unsafe { BorrowedAsyncDevice::borrow_raw(tun_fd as RawFd)? };
@@ -371,6 +521,7 @@ pub async fn run_capture_async(
     let mut bridge_tasks = JoinSet::new();
 
     let tls_acceptor = build_tls_acceptor();
+    let upstream = Upstream::new(proxy);
 
     let (result_tx, mut result_rx) = tokio::sync::oneshot::channel::<String>();
     let result_tx = Arc::new(Mutex::new(Some(result_tx)));
@@ -442,11 +593,13 @@ pub async fn run_capture_async(
                             let server_config = tls_acceptor.clone();
                             let token = cancel_token.clone();
                             let result_tx = result_tx.clone();
+                            let upstream = upstream.clone();
 
                             tokio::spawn(async move {
                                 match handle_tls_connection(
                                     app_io,
                                     dst,
+                                    upstream,
                                     hosts,
                                     path,
                                     save,
@@ -460,8 +613,11 @@ pub async fn run_capture_async(
                             });
                         }else{
                             let pump_token = cancel_token.clone();
+                            let upstream = upstream.clone();
                             bridge_tasks.spawn(async move {
-                                let _ = tcp_passthrough_cancellable(tcp, dst, pump_token).await;
+                                let _ =
+                                    tcp_passthrough_cancellable(tcp, dst, None, upstream, pump_token)
+                                        .await;
                             });
                         }
                     }
@@ -481,11 +637,16 @@ pub async fn run_capture_async(
     result
 }
 
-async fn tcp_passthrough<IO>(mut client: IO, dst: SocketAddr) -> anyhow::Result<()>
+async fn tcp_passthrough<IO>(
+    mut client: IO,
+    dst: SocketAddr,
+    sni: Option<&str>,
+    upstream: Upstream,
+) -> anyhow::Result<()>
 where
     IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
 {
-    let mut server = tokio::net::TcpStream::connect(dst).await?;
+    let mut server = upstream.connect(dst, sni).await?;
     tokio::io::copy_bidirectional(&mut client, &mut server).await?;
     Ok(())
 }
@@ -493,12 +654,14 @@ where
 async fn tcp_passthrough_cancellable<IO>(
     mut client: IO,
     dst: SocketAddr,
+    sni: Option<&str>,
+    upstream: Upstream,
     cancel: CancellationToken,
 ) -> anyhow::Result<()>
 where
     IO: AsyncRead + AsyncWrite + Unpin + Send + 'static,
 {
-    let mut server = tokio::net::TcpStream::connect(dst).await?;
+    let mut server = upstream.connect(dst, sni).await?;
     tokio::select! {
         _ = cancel.cancelled() => {
             let _ = tokio::time::timeout(