Solved conflict on H2 upstream and Websocket, better gRPC ping

2026-10-23 06:01:35 +08:00 · 2026-07-24 12:27:12 +02:00
parent 430f867792
commit 3ae54185ad
6 changed files with 131 additions and 117 deletions
@@ -135,7 +135,11 @@ dependencies = [
 "axum",
 "base16ct 1.0.0",
 "base64",
+ "bytes",
 "dashmap",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
 "instant-acme",
 "jsonwebtoken",
 "libc",
@@ -164,7 +168,7 @@ dependencies = [
 "signal-hook",
 "subtle",
 "tokio",
- "tonic",
+ "tokio-native-tls",
 "tower-http 0.7.0",
 "urlencoding",
 "x509-parser",
@@ -1052,6 +1056,12 @@ dependencies = [
 "windows-sys 0.61.2",
 ]

+[[package]]
+name = "fastrand"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223"
+
 [[package]]
 name = "ff"
 version = "0.13.1"
@@ -1498,19 +1508,6 @@ dependencies = [
 "tower-service",
 ]

-[[package]]
-name = "hyper-timeout"
-version = "0.5.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
-dependencies = [
- "hyper",
- "hyper-util",
- "pin-project-lite",
- "tokio",
- "tower-service",
-]
-
 [[package]]
 name = "hyper-util"
 version = "0.1.20"
@@ -1909,6 +1906,12 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "linux-raw-sys"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
+
 [[package]]
 name = "litemap"
 version = "0.8.2"
@@ -2092,6 +2095,23 @@ dependencies = [
 "uuid",
 ]

+[[package]]
+name = "native-tls"
+version = "0.2.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "465500e14ea162429d264d44189adc38b199b62b1c21eea9f69e4b73cb03bbf2"
+dependencies = [
+ "libc",
+ "log",
+ "openssl",
+ "openssl-probe 0.2.1",
+ "openssl-sys",
+ "schannel",
+ "security-framework",
+ "security-framework-sys",
+ "tempfile",
+]
+
 [[package]]
 name = "neli"
 version = "0.7.4"
@@ -2460,26 +2480,6 @@ dependencies = [
 "indexmap 2.14.0",
 ]

-[[package]]
-name = "pin-project"
-version = "1.1.13"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924"
-dependencies = [
- "pin-project-internal",
-]
-
-[[package]]
-name = "pin-project-internal"
-version = "1.1.13"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.118",
-]
-
 [[package]]
 name = "pin-project-lite"
 version = "0.2.17"
@@ -3254,6 +3254,19 @@ dependencies = [
 "nom",
 ]

+[[package]]
+name = "rustix"
+version = "1.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
+dependencies = [
+ "bitflags 2.13.0",
+ "errno",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "rustls"
 version = "0.23.41"
@@ -3775,6 +3788,19 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"

+[[package]]
+name = "tempfile"
+version = "3.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
+dependencies = [
+ "fastrand",
+ "getrandom 0.4.3",
+ "once_cell",
+ "rustix",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "thiserror"
 version = "1.0.69"
@@ -3927,6 +3953,16 @@ dependencies = [
 "syn 2.0.118",
 ]

+[[package]]
+name = "tokio-native-tls"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbae76ab933c85776efabc971569dd6119c580d8f5d448769dec1764bf796ef2"
+dependencies = [
+ "native-tls",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-openssl"
 version = "0.6.5"
@@ -3983,35 +4019,6 @@ dependencies = [
 "tokio",
 ]

-[[package]]
-name = "tonic"
-version = "0.14.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
-dependencies = [
- "async-trait",
- "axum",
- "base64",
- "bytes",
- "h2",
- "http",
- "http-body",
- "http-body-util",
- "hyper",
- "hyper-timeout",
- "hyper-util",
- "percent-encoding",
- "pin-project",
- "socket2",
- "sync_wrapper",
- "tokio",
- "tokio-stream",
- "tower",
- "tower-layer",
- "tower-service",
- "tracing",
-]
-
 [[package]]
 name = "tower"
 version = "0.5.3"
@@ -4020,12 +4027,9 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
 dependencies = [
 "futures-core",
 "futures-util",
- "indexmap 2.14.0",
 "pin-project-lite",
- "slab",
 "sync_wrapper",
 "tokio",
- "tokio-util",
 "tower-layer",
 "tower-service",
 "tracing",
@@ -4094,21 +4098,9 @@ checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
 dependencies = [
 "log",
 "pin-project-lite",
- "tracing-attributes",
 "tracing-core",
 ]

-[[package]]
-name = "tracing-attributes"
-version = "0.1.31"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.118",
-]
-
 [[package]]
 name = "tracing-core"
 version = "0.1.36"
@@ -33,7 +33,6 @@ noyalib = { version = "0.0.15", features = ["compat-serde-yaml"] }
 rand = "0.10.2"
 base64 = "0.22.1"
 jsonwebtoken = { version = "10.4.0", default-features = false, features = ["use_pem", "rust_crypto"] }
-tonic = "0.14.6"
 sha2 = { version = "0.11.0-rc.5", default-features = false }
 base16ct = { version = "1.0.0", features = ["alloc"] }
 urlencoding = "2.1.3"
@@ -54,4 +53,9 @@ mimalloc = { version = "0.1.52", default-features = false }
 signal-hook = "0.4.4"
 sd-notify = "0.5.0"
 libc = "0.2.186"
+tokio-native-tls = "0.3.1"
+hyper = "1.10.1"
+hyper-util = "0.1.20"
+http-body-util = "0.1.3"
+bytes = "1.12.1"

@@ -2,6 +2,7 @@ pub mod auth;
 pub mod discovery;
 mod filewatch;
 pub mod fordebug;
+pub mod grpc;
 pub mod healthcheck;
 pub mod httpclient;
 pub mod jwt;
@@ -12,4 +13,3 @@ pub mod parceyaml;
 pub mod state;
 pub mod structs;
 pub mod tools;
-// pub mod watchksecret;
@@ -0,0 +1,23 @@
+use bytes::Bytes;
+use http_body_util::Full;
+use hyper::Request;
+use hyper_util::client::legacy::connect::HttpConnector;
+use hyper_util::client::legacy::Client;
+use hyper_util::rt::TokioExecutor;
+
+pub async fn ping_grpc(addr: &str) -> bool {
+    let connector = HttpConnector::new();
+    let client = Client::builder(TokioExecutor::new()).http2_only(true).build(connector);
+    let uri = format!("{}{}", addr, "/aralez.Probe/Ping");
+    let body = Full::new(Bytes::from(vec![0, 0, 0, 0, 0]));
+    let request = Request::post(uri).header("content-type", "application/grpc").header("te", "trailers").body(body).unwrap();
+    match client.request(request).await {
+        Ok(resp) => resp
+            .headers()
+            .get("content-type")
+            .and_then(|v| v.to_str().ok())
+            .map(|v| v.starts_with("application/grpc"))
+            .unwrap_or(false),
+        Err(_) => false,
+    }
+}
@@ -1,3 +1,4 @@
+use crate::utils::grpc::ping_grpc;
 use crate::utils::lazylock::REVERSE_STORE;
 use crate::utils::structs::{InnerMap, UpstreamsDashMap, UpstreamsIdMap};
 use crate::utils::tools::*;
@@ -7,8 +8,9 @@ use reqwest::{Client, Version};
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 use std::time::Duration;
+use tokio::net::TcpStream;
 use tokio::time::interval;
-use tonic::transport::Endpoint;
+use tokio_native_tls::native_tls::TlsConnector;

 pub async fn hc2(upslist: Arc<UpstreamsDashMap>, fullist: Arc<UpstreamsDashMap>, idlist: Arc<UpstreamsIdMap>, params: (&str, u64)) {
    let mut period = interval(Duration::from_secs(params.1));
@@ -28,16 +30,6 @@ pub async fn hc2(upslist: Arc<UpstreamsDashMap>, fullist: Arc<UpstreamsDashMap>,
    }
 }

-/*
-pub async fn populate_upstreams(upslist: &Arc<UpstreamsDashMap>, fullist: &Arc<UpstreamsDashMap>, idlist: &Arc<UpstreamsIdMap>, params: (&str, u64), client: &Client) {
-    let totest = build_upstreams(fullist, params.0, client).await;
-    if !compare_dashmaps(&totest, upslist) {
-        clone_dashmap_into(&totest, upslist);
-        clone_idmap_into(&totest, idlist);
-    }
-}
-*/
-
 pub async fn initiate_upstreams(fullist: UpstreamsDashMap) -> UpstreamsDashMap {
    let client = Client::builder().timeout(Duration::from_secs(2)).danger_accept_invalid_certs(true).build().unwrap();
    build_upstreams(&fullist, "HEAD", &client).await
@@ -60,14 +52,11 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str, client: &Clie
                } else {
                    (false, None)
                };
-
-                let is_h2 = matches!(tls.1, Some(Version::HTTP_2));
-
                let mut scheme = InnerMap {
                    address: upstream.address.clone(),
                    port: upstream.port,
                    is_ssl: tls.0,
-                    is_http2: is_h2,
+                    is_http2: matches!(tls.1, Some(Version::HTTP_2)),
                    to_https: upstream.to_https,
                    rate_limit: upstream.rate_limit,
                    x4xx_limit: upstream.x4xx_limit,
@@ -84,9 +73,10 @@ async fn build_upstreams(fullist: &UpstreamsDashMap, method: &str, client: &Clie
                    };

                    let resp = http_request(&link, method, "", client).await;
+
                    if resp.0 {
                        if resp.1 {
-                            scheme.is_http2 = is_h2; // could be adjusted further
+                            scheme.is_http2 = resp.1;
                        }
                        innervec.push(Arc::from(scheme));
                    } else {
@@ -126,28 +116,33 @@ async fn http_request(url: &str, method: &str, payload: &str, client: &Client) -
    }
 }

-pub async fn ping_grpc(addr: &str) -> bool {
-    let endpoint = match Endpoint::from_shared(addr.to_owned()) {
-        Ok(e) => e.timeout(Duration::from_secs(2)),
-        Err(_) => return false,
+async fn detect_tls(ip: &str, port: &u16, client: &Client) -> (bool, Option<Version>) {
+    let addr = format!("{}:{}", ip, port);
+    let Ok(stream) = TcpStream::connect(&addr).await else {
+        return (false, Some(Version::HTTP_11));
    };
-    tokio::time::timeout(Duration::from_secs(3), endpoint.connect()).await.ok().and_then(Result::ok).is_some()
+
+    let connector = tokio_native_tls::TlsConnector::from(
+        TlsConnector::builder()
+            .danger_accept_invalid_certs(true)
+            .danger_accept_invalid_hostnames(true)
+            .build()
+            .unwrap(),
+    );
+
+    let tls = connector.connect(ip, stream).await.is_ok();
+    if tls {
+        let vers = detect_version(ip, *port, true, client).await;
+        return (tls, vers);
+    }
+    (tls, Some(Version::HTTP_11))
 }

-async fn detect_tls(ip: &str, port: &u16, client: &Client) -> (bool, Option<Version>) {
-    let https_url = format!("https://{}:{}", ip, port);
-    if let Ok(response) = client.get(&https_url).send().await {
-        return (true, Some(response.version()));
-    }
-    let http_url = format!("http://{}:{}", ip, port);
-    match client.get(&http_url).send().await {
-        Ok(response) => (false, Some(response.version())),
-        Err(_) => {
-            if ping_grpc(&http_url).await {
-                (false, Some(Version::HTTP_2))
-            } else {
-                (false, None)
-            }
-        }
+async fn detect_version(ip: &str, port: u16, is_tls: bool, client: &Client) -> Option<Version> {
+    let scheme = if is_tls { "https" } else { "http" };
+    let url = format!("{}://{}:{}", scheme, ip, port);
+    match client.get(&url).send().await.ok().map(|r| r.version()) {
+        Some(version) => Some(version),
+        None => Some(Version::HTTP_11),
    }
 }
@@ -190,7 +190,7 @@ impl ProxyHttp for LB {
            Some(hostname) => match ctx.upstream_peer.as_ref() {
                Some(innermap) => {
                    let mut peer = Box::new(HttpPeer::new((&*innermap.address, innermap.port), innermap.is_ssl, hostname.to_string()));
-                    if innermap.is_http2 {
+                    if innermap.is_http2 && !session.is_upgrade_req() {
                        peer.options.alpn = ALPN::H2;
                    }
                    if innermap.is_ssl {