Draft: Implement mock_dns outbound

Cargo.toml

@@ -32,6 +32,7 @@ fastrand = "2.1"
 async-trait = "0.1"
 schemars = { version = "0.8", features = ["uuid1"] }
 bincode = "2.0.0-rc.3"
+reqwest = { version = "0.11", features = ["stream"] }
 
 [profile.release]
 opt-level = "s"

config.toml

@@ -34,3 +34,4 @@ match = [
  "172.64.0.0/13",
  "131.0.72.0/22"
 ]
+mock_udp = true

src/config.rs

@@ -34,6 +34,8 @@ pub struct Outbound {
  pub port: u16,
  #[serde(default)]
  pub uuid: Uuid,
+ #[serde(default)]
+ pub mock_udp: bool,
 }
 
 #[derive(Default, Clone, Serialize, Deserialize, JsonSchema)]

src/proxy/mock_udp/doh.rs

@@ -0,0 +1,22 @@
+use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, CONTENT_TYPE};
+use reqwest::Client;
+
+pub async fn doh(req_wireformat: &[u8]) -> reqwest::Result<Vec<u8>> {
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ CONTENT_TYPE,
+ HeaderValue::from_static("application/dns-message"),
+ );
+ headers.insert(ACCEPT, HeaderValue::from_static("application/dns-message"));
+ let client = Client::new();
+ let response = client
+ .post("https://1.1.1.1/dns-query")
+ .headers(headers)
+ .body(req_wireformat.to_vec())
+ .send()
+ .await?
+ .bytes()
+ .await?;
+
+ Ok(response.to_vec())
+}

src/proxy/mock_udp/mod.rs

@@ -0,0 +1,9 @@
+//! Mock some udp based protocols.
+//! - DNS: receive the dns request from in-bound but handle with DoH wireformat
+//! - QUIC:
+//! TODO: maybe we can receive the request and act like a sni-proxy?
+//! - WebRTC:
+//! TODO: maybe we can forge a response to force it to use TCP?
+
+pub mod doh;
+pub mod outbound;

src/proxy/mock_udp/outbound.rs

@@ -0,0 +1,94 @@
+use crate::proxy::Proxy;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use async_trait::async_trait;
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use worker::*;
+
+use std::sync::{Arc, Mutex};
+use std::task::Waker;
+
+use super::doh;
+
+struct MockUDPStreamInner {
+ response: Option<Vec<u8>>,
+ waker: Option<Waker>,
+}
+
+impl MockUDPStreamInner {
+ fn set_response(&mut self, response: Vec<u8>) {
+ self.response = Some(response);
+ if let Some(waker) = self.waker.take() {
+ waker.wake()
+ }
+ }
+}
+
+pub struct MockUDPStream {
+ inner: Arc<Mutex<MockUDPStreamInner>>,
+}
+
+impl MockUDPStream {
+ pub fn new() -> Self {
+ MockUDPStream {
+ inner: Arc::new(Mutex::new(MockUDPStreamInner {
+ response: None,
+ waker: None,
+ })),
+ }
+ }
+}
+
+#[async_trait]
+impl Proxy for MockUDPStream {
+ async fn process(&mut self) -> Result<()> {
+ Ok(())
+ }
+}
+
+impl AsyncRead for MockUDPStream {
+ fn poll_read(
+ self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ buf: &mut ReadBuf<'_>,
+ ) -> Poll<tokio::io::Result<()>> {
+ let mut inner = self.inner.lock().unwrap();
+ if let Some(response) = inner.response.take() {
+ let len = std::cmp::min(buf.remaining(), response.len());
+ buf.put_slice(&response[..len]);
+ Poll::Ready(Ok(()))
+ } else {
+ inner.waker = Some(cx.waker().clone());
+ Poll::Pending
+ }
+ }
+}
+
+impl AsyncWrite for MockUDPStream {
+ fn poll_write(
+ self: Pin<&mut Self>,
+ _cx: &mut Context<'_>,
+ buf: &[u8],
+ ) -> Poll<tokio::io::Result<usize>> {
+ let inner_ref = self.inner.clone();
+
+ let b = buf.to_vec();
+ wasm_bindgen_futures::spawn_local(async move {
+ match doh::doh(b.as_slice()).await {
+ Ok(x) => inner_ref.lock().unwrap().set_response(x),
+ Err(e) => console_error!("doh failed: {e:?}"),
+ };
+ });
+
+ Poll::Ready(Ok(buf.len()))
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
+ Poll::Ready(Ok(()))
+ }
+
+ fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<tokio::io::Result<()>> {
+ Poll::Ready(Ok(()))
+ }
+}

src/proxy/mod.rs

@@ -1,5 +1,6 @@
 pub mod bepass;
 pub mod blackhole;
+pub mod mock_udp;
 pub mod relay;
 pub mod trojan;
 pub mod vless;
@@ -109,22 +110,29 @@ async fn connect_outbound(ctx: RequestContext, outbound: Outbound) -> Result<Box
  outbound.protocol
  );
 
- let socket = Socket::builder().connect(addr, port)?;
-
- let mut stream: Box<dyn Proxy> = match outbound.protocol {
- Protocol::Vless => Box::new(vless::outbound::VlessStream::new(ctx, outbound, socket)),
- Protocol::RelayV1 => Box::new(relay::outbound::RelayStream::new(
- ctx,
- socket,
- relay::outbound::RelayVersion::V1,
- )),
- Protocol::RelayV2 => Box::new(relay::outbound::RelayStream::new(
- ctx,
- socket,
- relay::outbound::RelayVersion::V2,
- )),
- Protocol::Blackhole => Box::new(blackhole::outbound::BlackholeStream),
- _ => Box::new(socket),
+ let maybe_dns = outbound.mock_udp && port == 53 && matches!(ctx.network, Network::Udp);
+
+ let mut stream = if maybe_dns {
+ Box::new(mock_udp::outbound::MockUDPStream::new())
+ } else {
+ let socket = Socket::builder().connect(addr, port)?;
+
+ let stream: Box<dyn Proxy> = match outbound.protocol {
+ Protocol::Vless => Box::new(vless::outbound::VlessStream::new(ctx, outbound, socket)),
+ Protocol::RelayV1 => Box::new(relay::outbound::RelayStream::new(
+ ctx,
+ socket,
+ relay::outbound::RelayVersion::V1,
+ )),
+ Protocol::RelayV2 => Box::new(relay::outbound::RelayStream::new(
+ ctx,
+ socket,
+ relay::outbound::RelayVersion::V2,
+ )),
+ Protocol::Blackhole => Box::new(blackhole::outbound::BlackholeStream),
+ _ => Box::new(socket),
+ };
+ stream
  };
 
  stream.process().await?;