v0.1.1

.gitignore

@@ -0,0 +1,3 @@
+/target
+/server.key
+/server.crt

Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "proxy_checker"
+version = "0.1.1"
+edition = "2021"
+
+[dependencies]
+tokio = { version = "1.39.2", features = ["full"] }
+rustls = { version = "0.23.12"}
+tracing = "0.1.40"
+libc = "0.2.155"
+anyhow = "1.0.86"
+tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
+tokio-rustls = "0.26.0"
+rustls-pemfile = "2.1.3"
+http = "1.1.0"
+h2 = "0.4.5"
+bytes = "1.7.1"
+serde = "1.0.208"
+serde_json = "1.0.125"
+clap = { version = "4.5.15", features = ["derive"] }

src/args.rs

@@ -0,0 +1,16 @@
+use clap::Parser;
+
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+pub struct Args {
+    #[arg(short, long, default_value_t = String::from("0.0.0.0:8443"))]
+    pub addr: String,
+    #[arg(long, default_value_t = String::from("./server.crt"))]
+    pub cert: String,
+    #[arg(long, default_value_t = String::from("./server.key"))]
+    pub key: String,
+    #[arg(short, long, default_value_t = String::from("info"))]
+    pub log_level: String,
+    #[arg(short, long, default_value_t = 10)]
+    pub duration: i32,
+}

src/http2.rs

@@ -0,0 +1,102 @@
+use crate::tcp::get_tcp_rtt;
+use crate::ARGS;
+use anyhow::Result;
+use bytes::Bytes;
+use h2::server::SendResponse;
+use h2::{Ping, PingPong, RecvStream};
+use http::header::{ACCESS_CONTROL_ALLOW_ORIGIN, CONTENT_TYPE};
+use http::Request;
+use serde_json::json;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tracing::{debug, error, info};
+
+#[tracing::instrument(fields(uri = %request.uri().path(), ip = %addr.ip()), skip_all)]
+pub async fn handle(
+    request: &mut Request<RecvStream>,
+    respond: &mut SendResponse<Bytes>,
+    raw_fd: i32,
+    addr: SocketAddr,
+    ping_pong: Arc<Mutex<PingPong>>,
+) -> Result<()> {
+    if request.uri().path() != "/" {
+        error!(?request, "not found");
+        let response = http::Response::builder()
+            .status(http::StatusCode::NOT_FOUND)
+            .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*")
+            .body(())?;
+        let _ = respond.send_response(response, true)?;
+        return Ok(());
+    }
+
+    let header_frame = http::Response::builder()
+        .status(http::StatusCode::OK)
+        .header(CONTENT_TYPE, "application/json")
+        .header(ACCESS_CONTROL_ALLOW_ORIGIN, "*")
+        .body(())?;
+    let mut send = respond.send_response(header_frame, false)?;
+
+    let tls_rtts = tls_rtt(ping_pong).await?;
+    let (tls_rtt, _std_dev, anomalies) = detect_anomalies(&tls_rtts, 0.5);
+    let tcp_rtt = get_tcp_rtt(raw_fd);
+
+    // TODO: type convert
+    let is_proxy = tls_rtt - tcp_rtt as f64 > ARGS.duration as f64 * 1000.0;
+
+    info!(?is_proxy, ?tcp_rtt, ?tls_rtt, ?anomalies, "result");
+
+    Ok(send.send_data(
+        json!({
+            "is_proxy": is_proxy,
+            "ip": addr.ip().to_string(),
+        })
+        .to_string()
+        .into(),
+        true,
+    )?)
+}
+
+#[tracing::instrument(skip_all)]
+async fn tls_rtt(ping_pong: Arc<Mutex<PingPong>>) -> Result<Vec<i32>> {
+    let mut ping = ping_pong.lock().await;
+    let mut tls_rtts = vec![];
+    for index in 0..10 {
+        let instant = tokio::time::Instant::now();
+        ping.ping(Ping::opaque()).await?;
+        let duration = instant.elapsed().as_micros();
+        debug!(?duration, %index, "ping");
+        tls_rtts.push(duration as i32);
+    }
+    Ok(tls_rtts)
+}
+
+#[tracing::instrument]
+fn detect_anomalies(data: &[i32], threshold: f64) -> (f64, f64, Vec<i32>) {
+    let sum: i32 = data.iter().sum();
+    let mean = sum as f64 / data.len() as f64;
+
+    let variance = calculate_variance(data, mean);
+    let std_dev = variance.sqrt();
+
+    let anomalies: Vec<i32> = data
+        .iter()
+        .filter(|&&x| (x as f64 - mean).abs() > threshold * std_dev)
+        .copied()
+        .collect();
+
+    let non_anomalies_count = data.len() - anomalies.len();
+    let adjusted_mean = if non_anomalies_count > 0 {
+        (sum - anomalies.iter().sum::<i32>()) as f64 / non_anomalies_count as f64
+    } else {
+        mean
+    };
+
+    info!(?data, ?adjusted_mean, ?std_dev, ?anomalies);
+
+    (adjusted_mean, std_dev, anomalies)
+}
+
+fn calculate_variance(data: &[i32], mean: f64) -> f64 {
+    data.iter().map(|&x| (x as f64 - mean).powi(2)).sum::<f64>() / data.len() as f64
+}

src/main.rs

@@ -0,0 +1,94 @@
+use anyhow::Result;
+use clap::Parser;
+use rustls::pki_types::{CertificateDer, PrivateKeyDer};
+use rustls::ServerConfig;
+use std::io::BufReader;
+use std::sync::{Arc, LazyLock};
+use tokio::net::TcpListener;
+use tracing::{debug, info, trace};
+use tracing_subscriber::layer::SubscriberExt;
+use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::EnvFilter;
+
+mod args;
+mod http2;
+mod tcp;
+mod tls;
+
+pub static ARGS: LazyLock<args::Args> = LazyLock::new(args::Args::parse);
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    tracing_subscriber::Registry::default()
+        .with(
+            EnvFilter::builder()
+                .with_default_directive(ARGS.log_level.parse().unwrap())
+                .from_env_lossy(),
+        )
+        .with(tracing_subscriber::fmt::layer().pretty())
+        .init();
+
+    let listener = TcpListener::bind(&ARGS.addr).await?;
+    info!("listening on {}", ARGS.addr);
+    debug!("listening on {}", ARGS.addr);
+    trace!("listening on {}", ARGS.addr);
+    println!("listening on {}", ARGS.addr);
+
+    let server_config = load_tls().expect(
+        r#"
+load tls certificate fail,
+run:
+openssl req -new -newkey rsa:2048 -days 365 -nodes -x509 -keyout server.key -out server.crt
+
+generate one
+        "#,
+    );
+    let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
+
+    loop {
+        if let Ok((socket, addr)) = listener.accept().await {
+            info!(name = "Accept TCP connection from", addr = %addr.ip());
+
+            let tls_acceptor = tls_acceptor.clone();
+            tokio::spawn(tcp::handle(socket, tls_acceptor, addr));
+        }
+    }
+}
+
+fn load_tls() -> Result<ServerConfig> {
+    info!("load key and cert");
+    let cert = load_certs(&ARGS.cert)?;
+    let key = load_private_key(&ARGS.key)?;
+    let mut config = rustls::ServerConfig::builder()
+        .with_no_client_auth()
+        .with_single_cert(cert, key)?;
+    config.alpn_protocols = vec![b"h2".to_vec()];
+    Ok(config)
+}
+
+fn load_certs(filename: &str) -> Result<Vec<CertificateDer<'static>>> {
+    let certfile = std::fs::File::open(filename)?;
+    let mut reader = BufReader::new(certfile);
+    Ok(rustls_pemfile::certs(&mut reader)
+        .map(|result| result.expect("parse cert error"))
+        .collect())
+}
+
+fn load_private_key(filename: &str) -> Result<PrivateKeyDer<'static>> {
+    let keyfile = std::fs::File::open(filename).expect("cannot open private key file");
+    let mut reader = BufReader::new(keyfile);
+
+    Ok(
+        match rustls_pemfile::read_one(&mut reader).expect("cannot parse private key .pem file") {
+            Some(rustls_pemfile::Item::Pkcs1Key(key)) => key.into(),
+            Some(rustls_pemfile::Item::Pkcs8Key(key)) => key.into(),
+            Some(rustls_pemfile::Item::Sec1Key(key)) => key.into(),
+            _ => {
+                panic!(
+                    "no keys found in {:?} (encrypted keys not supported)",
+                    filename
+                )
+            }
+        },
+    )
+}

src/tcp.rs

@@ -0,0 +1,119 @@
+use libc::{self, __u16, __u32, __u64, __u8};
+use std::net::SocketAddr;
+use tokio::net::TcpStream;
+use tokio_rustls::TlsAcceptor;
+use tracing::{error, info};
+
+#[tracing::instrument(skip_all)]
+pub async fn handle(conn: TcpStream, acceptor: TlsAcceptor, addr: SocketAddr) {
+    let socket = match acceptor.accept(conn).await {
+        Ok(v) => v,
+        Err(err) => {
+            error!(?err, "Failed to perform TLS handshake");
+            return;
+        }
+    };
+
+    info!(name = "Accept TLS connection", addr = %addr.ip());
+
+    if let Err(err) = super::tls::handle(socket, addr).await {
+        error!(error = ?err, "error while handle tls connection");
+    }
+}
+
+#[repr(C)]
+#[derive(Default)]
+pub struct TcpInfo {
+    pub tcpi_state: __u8,
+    pub tcpi_ca_state: __u8,
+    pub tcpi_retransmits: __u8,
+    pub tcpi_probes: __u8,
+    pub tcpi_backoff: __u8,
+    pub tcpi_options: __u8,
+    pub tcpi_snd_wscale_rcv_wscale: __u8,
+    pub tcpi_delivery_rate_app_limited_fastopen_client_fail: __u8,
+
+    pub tcpi_rto: __u32,
+    pub tcpi_ato: __u32,
+    pub tcpi_snd_mss: __u32,
+    pub tcpi_rcv_mss: __u32,
+
+    pub tcpi_unacked: __u32,
+    pub tcpi_sacked: __u32,
+    pub tcpi_lost: __u32,
+    pub tcpi_retrans: __u32,
+    pub tcpi_fackets: __u32,
+
+    pub tcpi_last_data_sent: __u32,
+    pub tcpi_last_ack_sent: __u32,
+    pub tcpi_last_data_recv: __u32,
+    pub tcpi_last_ack_recv: __u32,
+
+    pub tcpi_pmtu: __u32,
+    pub tcpi_rcv_ssthresh: __u32,
+    pub tcpi_rtt: __u32,
+    pub tcpi_rttvar: __u32,
+    pub tcpi_snd_ssthresh: __u32,
+    pub tcpi_snd_cwnd: __u32,
+    pub tcpi_advmss: __u32,
+    pub tcpi_reordering: __u32,
+
+    pub tcpi_rcv_rtt: __u32,
+    pub tcpi_rcv_space: __u32,
+
+    pub tcpi_total_retrans: __u32,
+
+    pub tcpi_pacing_rate: __u64,
+    pub tcpi_max_pacing_rate: __u64,
+    pub tcpi_bytes_acked: __u64,
+    pub tcpi_bytes_received: __u64,
+    pub tcpi_segs_out: __u32,
+    pub tcpi_segs_in: __u32,
+
+    pub tcpi_notsent_bytes: __u32,
+    pub tcpi_min_rtt: __u32,
+    pub tcpi_data_segs_in: __u32,
+    pub tcpi_data_segs_out: __u32,
+
+    pub tcpi_delivery_rate: __u64,
+
+    pub tcpi_busy_time: __u64,
+    pub tcpi_rwnd_limited: __u64,
+    pub tcpi_sndbuf_limited: __u64,
+
+    pub tcpi_delivered: __u32,
+    pub tcpi_delivered_ce: __u32,
+
+    pub tcpi_bytes_sent: __u64,
+    pub tcpi_bytes_retrans: __u64,
+    pub tcpi_dsack_dups: __u32,
+    pub tcpi_reord_seen: __u32,
+
+    pub tcpi_rcv_ooopack: __u32,
+
+    pub tcpi_snd_wnd: __u32,
+    pub tcpi_rcv_wnd: __u32,
+
+    pub tcpi_rehash: __u32,
+
+    pub tcpi_total_rto: __u16,
+    pub tcpi_total_rto_recoveries: __u16,
+    pub tcpi_total_rto_time: __u32,
+}
+
+pub fn get_tcp_rtt(raw_fd: i32) -> u32 {
+    let mut tcp_info = TcpInfo::default();
+    let mut tcp_info_len = std::mem::size_of::<TcpInfo>() as u32;
+
+    let _ret = unsafe {
+        libc::getsockopt(
+            raw_fd,
+            libc::IPPROTO_TCP,
+            libc::TCP_INFO,
+            &mut tcp_info as *mut _ as *mut libc::c_void,
+            &mut tcp_info_len,
+        )
+    };
+
+    tcp_info.tcpi_rtt
+}

src/tls.rs

@@ -0,0 +1,37 @@
+use anyhow::{Context, Result};
+use std::net::SocketAddr;
+use std::os::fd::AsRawFd;
+use std::sync::Arc;
+use tokio::net::TcpStream;
+use tokio::sync::Mutex;
+use tokio_rustls::server::TlsStream;
+use tracing::{error, info, Instrument};
+
+#[tracing::instrument(skip_all)]
+pub async fn handle(conn: TlsStream<TcpStream>, addr: SocketAddr) -> Result<()> {
+    let raw_fd = conn.get_ref().0.as_raw_fd();
+    let mut connection = h2::server::handshake(conn).await?;
+    info!("H2 connection bound");
+    let ping_pong = Arc::new(Mutex::new(
+        connection.ping_pong().context("ping pong error")?,
+    ));
+
+    while let Some(result) = connection.accept().await {
+        let (mut request, mut respond) = result?;
+        let ping_pong = ping_pong.clone();
+        // TODO: spawn a ping task
+        tokio::spawn(
+            async move {
+                if let Err(e) =
+                    super::http2::handle(&mut request, &mut respond, raw_fd, addr, ping_pong).await
+                {
+                    error!("error while handling request: {}", e);
+                }
+            }
+            .in_current_span(),
+        );
+    }
+
+    connection.graceful_shutdown();
+    Ok(())
+}