From f0db77ff7a70ff87e03ab259c8dcbcedce32c01c Mon Sep 17 00:00:00 2001 From: "Y. T. Chung" Date: Sun, 23 Oct 2016 22:03:02 +0800 Subject: [PATCH] fixed http bug --- src/relay/tcprelay/http.rs | 233 +++++++++++++++++++++++++++++++----- src/relay/tcprelay/local.rs | 119 ++++++++---------- 2 files changed, 249 insertions(+), 103 deletions(-) diff --git a/src/relay/tcprelay/http.rs b/src/relay/tcprelay/http.rs index 9caf9961..b4f97bf7 100644 --- a/src/relay/tcprelay/http.rs +++ b/src/relay/tcprelay/http.rs @@ -28,7 +28,7 @@ use std::str; use std::fmt; use hyper::uri::RequestUri; -use hyper::header::{Header, HeaderFormat, Headers}; +use hyper::header::{Header, HeaderFormat, Headers, ContentLength}; use hyper::status::StatusCode; use hyper::version::HttpVersion; use hyper::method::Method; @@ -182,16 +182,6 @@ pub struct HttpResponse { } impl HttpResponse { - /// Creates an empty Response - pub fn new() -> HttpResponse { - HttpResponse { - version: HttpVersion::Http11, - status: StatusCode::Ok, - message: None, - headers: Headers::new(), - } - } - pub fn from_raw<'headers, 'buf: 'headers>(rsp: &Response<'headers, 'buf>, headers: &'headers [httparse::Header]) -> hyper::Result { @@ -285,7 +275,7 @@ impl HeaderFormat for XForwardFor { } #[derive(Debug, Clone)] -pub struct XRealIp(pub Option); +pub struct XRealIp(pub IpAddr); impl Header for XRealIp { fn header_name() -> &'static str { @@ -310,41 +300,41 @@ impl Header for XRealIp { } } - Ok(XRealIp(ip)) + match ip { + Some(ip) => Ok(XRealIp(ip)), + None => Err(hyper::Error::Header), + } } } impl HeaderFormat for XRealIp { + #[inline] fn fmt_header(&self, f: &mut fmt::Formatter) -> fmt::Result { - if let Some(ref ip) = self.0 { - try!(write!(f, "{}", ip)); - } - - Ok(()) + write!(f, "{}", self.0) } } -/// HTTP Client -pub enum RequestReader +/// Future for reading HttpRequest +pub enum HttpRequestFut where R: Read { Pending { r: R, buf: Vec }, Empty, } -impl RequestReader +impl HttpRequestFut where R: Read { - pub fn new(r: R) -> RequestReader { - RequestReader::with_buf(r, Vec::new()) + pub fn new(r: R) -> HttpRequestFut { + HttpRequestFut::with_buf(r, Vec::new()) } - pub fn with_buf(r: R, buf: Vec) -> RequestReader { - RequestReader::Pending { r: r, buf: buf } + pub fn with_buf(r: R, buf: Vec) -> HttpRequestFut { + HttpRequestFut::Pending { r: r, buf: buf } } } -impl Future for RequestReader +impl Future for HttpRequestFut where R: Read { type Item = (R, HttpRequest, Vec); @@ -353,9 +343,10 @@ impl Future for RequestReader fn poll(&mut self) -> Poll { let mut lbuf = [0u8; 4096]; let (req, len) = match self { - &mut RequestReader::Pending { ref mut r, ref mut buf } => { - let mut http_req = None; - let mut total_len = 0; + &mut HttpRequestFut::Pending { ref mut r, ref mut buf } => { + // FIXME: Compiler force me to do this! + let http_req: Option; + let total_len: usize; loop { let n = try_nb!(r.read(&mut lbuf)); buf.extend_from_slice(&lbuf[..n]); @@ -398,12 +389,188 @@ impl Future for RequestReader (http_req.unwrap(), total_len) } - &mut RequestReader::Empty => panic!("poll a RequestReader after it's done"), + &mut HttpRequestFut::Empty => panic!("poll a HttpRequestFut after it's done"), }; - match mem::replace(self, RequestReader::Empty) { - RequestReader::Pending { r, buf } => Ok((r, req, buf[len..].to_vec()).into()), - RequestReader::Empty => unreachable!(), + match mem::replace(self, HttpRequestFut::Empty) { + HttpRequestFut::Pending { r, buf } => Ok((r, req, buf[len..].to_vec()).into()), + HttpRequestFut::Empty => unreachable!(), } } +} + +/// Future for reading HttpRequest +pub enum HttpResponseFut + where R: Read +{ + Pending { r: R, buf: Vec }, + Empty, +} + +impl HttpResponseFut + where R: Read +{ + pub fn new(r: R) -> HttpResponseFut { + HttpResponseFut::with_buf(r, Vec::new()) + } + + pub fn with_buf(r: R, buf: Vec) -> HttpResponseFut { + HttpResponseFut::Pending { r: r, buf: buf } + } +} + +impl Future for HttpResponseFut + where R: Read +{ + type Item = (R, HttpResponse, Vec); + type Error = io::Error; + + fn poll(&mut self) -> Poll { + let mut lbuf = [0u8; 4096]; + let (req, len) = match self { + &mut HttpResponseFut::Pending { ref mut r, ref mut buf } => { + // FIXME: Compiler force me to do this! + let http_req: Option; + let total_len: usize; + loop { + let n = try_nb!(r.read(&mut lbuf)); + buf.extend_from_slice(&lbuf[..n]); + + // Maximum 128 headers + let mut headers = [httparse::EMPTY_HEADER; 128]; + let headers_ptr = &headers as *const _; + let mut req = Response::new(&mut headers); + match req.parse(&mut buf[..]) { + Ok(httparse::Status::Partial) => { + if n == 0 { + // Already EOF! + let err = io::Error::new(io::ErrorKind::UnexpectedEof, "Unexpected Eof"); + return Err(err); + } + } + Ok(httparse::Status::Complete(len)) => { + total_len = len; + + // Make borrow checker happy + let headers_ref = unsafe { &*headers_ptr }; + let hreq = match HttpResponse::from_raw(&req, headers_ref) { + Ok(r) => r, + Err(err) => { + error!("HttpResponse::from_raw: {}", err); + let err = io::Error::new(io::ErrorKind::Other, "Hyper error"); + return Err(err); + } + }; + http_req = Some(hreq); + break; + } + Err(err) => { + error!("Request parse: {:?}", err); + let err = io::Error::new(io::ErrorKind::Other, "Hyper error"); + return Err(err); + } + } + } + + (http_req.unwrap(), total_len) + } + &mut HttpResponseFut::Empty => panic!("poll a HttpResponseFut after it's done"), + }; + + match mem::replace(self, HttpResponseFut::Empty) { + HttpResponseFut::Pending { r, buf } => Ok((r, req, buf[len..].to_vec()).into()), + HttpResponseFut::Empty => unreachable!(), + } + } +} + +fn socket_to_ip(addr: &SocketAddr) -> IpAddr { + match *addr { + SocketAddr::V4(ref v4) => IpAddr::V4(v4.ip().clone()), + SocketAddr::V6(ref v6) => IpAddr::V6(v6.ip().clone()), + } +} + +/// Proxy this HTTP Request to writer +pub fn proxy_request((r, w): (R, W), + client_addr: SocketAddr, + mut req: HttpRequest, + mut remains: Vec) + -> BoxFuture<(R, W, Vec), io::Error> + where R: Read + Send + 'static, + W: Write + Send + 'static +{ + let content_length = req.headers.get::().unwrap_or(&ContentLength(0)).0 as usize; + let client_ip = socket_to_ip(&client_addr); + + // Set proxy IP info + let xf = if let Some(fw) = req.headers.get_mut::() { + let mut flst = fw.0.clone(); + flst.push(client_ip.clone()); + flst + } else { + vec![client_ip.clone()] + }; + req.headers.set(XForwardFor(xf)); + + // Set real ip + req.headers.set(XRealIp(client_ip)); + + req.write_to(w) + .and_then(move |w| { + if content_length == 0 { + futures::finished((r, w, remains)).boxed() + } else if content_length <= remains.len() { + let after_that = remains.split_off(content_length); + write_all(w, remains).map(|(w, _)| (r, w, after_that)).boxed() + } else { + let missing_bytes = content_length - remains.len(); + write_all(w, remains) + .and_then(move |(w, _)| super::copy_exact(r, w, missing_bytes).map(|(r, w)| (r, w, vec![]))) + .boxed() + } + }) + .boxed() +} + +/// Proxy this HTTP Response to writer +pub fn proxy_response((r, w): (R, W), + server_addr: SocketAddr, + mut rsp: HttpResponse, + mut remains: Vec) + -> BoxFuture<(R, W, Vec), io::Error> + where R: Read + Send + 'static, + W: Write + Send + 'static +{ + let content_length = rsp.headers.get::().unwrap_or(&ContentLength(0)).0 as usize; + let server_ip = socket_to_ip(&server_addr); + + // Set proxy IP info + let xf = if let Some(fw) = rsp.headers.get_mut::() { + let mut flst = fw.0.clone(); + flst.push(server_ip.clone()); + flst + } else { + vec![server_ip.clone()] + }; + rsp.headers.set(XForwardFor(xf)); + + // Set real ip + rsp.headers.set(XRealIp(server_ip)); + + rsp.write_to(w) + .and_then(move |w| { + if content_length == 0 { + futures::finished((r, w, remains)).boxed() + } else if content_length <= remains.len() { + let after_that = remains.split_off(content_length); + write_all(w, remains).map(|(w, _)| (r, w, after_that)).boxed() + } else { + let missing_bytes = content_length - remains.len(); + write_all(w, remains) + .and_then(move |(w, _)| super::copy_exact(r, w, missing_bytes).map(|(r, w)| (r, w, vec![]))) + .boxed() + } + }) + .boxed() } \ No newline at end of file diff --git a/src/relay/tcprelay/local.rs b/src/relay/tcprelay/local.rs index 83dbd9e3..bb049b68 100644 --- a/src/relay/tcprelay/local.rs +++ b/src/relay/tcprelay/local.rs @@ -34,7 +34,6 @@ use tokio_core::io::Io; use tokio_core::io::{ReadHalf, WriteHalf}; use tokio_core::io::{flush, copy, write_all}; -use hyper::header::ContentLength; use hyper::method::Method; use config::{Config, ServerConfig}; @@ -44,7 +43,7 @@ use relay::socks5::{TcpRequestHeader, TcpResponseHeader}; use relay::loadbalancing::server::RoundRobin; use relay::loadbalancing::server::LoadBalancer; -use super::http::{self, RequestReader}; +use super::http::{self, HttpRequestFut, HttpResponseFut}; /// TCP relay local server pub struct TcpRelayLocal { @@ -235,6 +234,7 @@ impl HttpRelayServer { trace!("Sending HTTP tunnel handshake response"); write_all(w, handshake_resp.into_bytes()).and_then(|(w, _)| flush(w)).map(|w| (svr_r, svr_w, w)) }) + .and_then(move |(svr_r, svr_w, w)| write_all(svr_w, remains).map(|(svr_w, _)| (svr_r, svr_w, w))) .and_then(move |(svr_r, svr_w, w)| { let c2s = copy(r, svr_w); let s2c = copy(svr_r, w); @@ -250,99 +250,72 @@ impl HttpRelayServer { fn handle_http_again((r, w): (ReadHalf, WriteHalf), (svr_r, svr_w): (super::DecryptedHalf, super::EncryptedHalf), - remains: Vec) + client_addr: SocketAddr, + (req_remains, rsp_remains): (Vec, Vec), + svr_cfg: Arc) -> BoxFuture<(), io::Error> { - RequestReader::with_buf(r, remains) - .and_then(move |(r, mut req, remains)| { - trace!("Got HTTP Request, version: {}, method: {}, uri: {}", - req.version, - req.method, - req.request_uri); - - match req.get_address() { - Ok(..) => { - req.clear_request_uri_host(); - let content_length = req.headers.get::().unwrap_or(&ContentLength(0)).0; - req.write_to(svr_w) - .and_then(move |svr_w| { - if content_length == 0 { - // Do nothing because this request does not have a body - futures::finished((r, svr_w, remains)).boxed() - } else { - write_all(svr_w, remains) - .and_then(move |(svr_w, remains)| { - let remain_len = content_length - remains.len() as u64; - super::copy_exact(r, svr_w, remain_len as usize) - }) - .map(|(r, svr_w)| (r, svr_w, vec![])) - .boxed() - } + let client_addr_cloned = client_addr.clone(); + HttpRequestFut::with_buf(r, req_remains) + .and_then(move |(r, req, req_remains)| { + let svr_addr = svr_cfg.addr.clone(); + http::proxy_request((r, svr_w), client_addr, req, req_remains) + .and_then(move |(r, svr_w, req_remains)| { + HttpResponseFut::with_buf(svr_r, rsp_remains) + .and_then(move |(svr_r, rsp, rsp_remains)| { + http::proxy_response((svr_r, w), svr_addr, rsp, rsp_remains) }) - .map(move |(r, svr_w, remains)| { - HttpRelayServer::handle_http_again((r, w), (svr_r, svr_w), remains) - }) - .boxed() - } - Err(status_code) => { - http::write_response(w, req.version, status_code) - .then(|_| { - let err = io::Error::new(io::ErrorKind::Other, "Invalid Uri"); - Err(err) - }) - .boxed() - } - } - }) - .then(|res| { - if let Err(err) = res { - error!("HTTP again: {}", err); - } - Ok(()) + .map(move |(svr_r, w, rsp_remains)| (r, w, svr_r, svr_w, req_remains, rsp_remains)) + }) + .and_then(move |(r, w, svr_r, svr_w, req_remains, rsp_remains)| { + HttpRelayServer::handle_http_again((r, w), + (svr_r, svr_w), + client_addr_cloned, + (req_remains, rsp_remains), + svr_cfg) + }) }) .boxed() } fn handle_http_proxy(handle: Handle, (r, w): (ReadHalf, WriteHalf), + client_addr: SocketAddr, req: http::HttpRequest, addr: Address, remains: Vec, svr_cfg: Arc) -> BoxFuture<(), io::Error> { - let content_length = req.headers.get::().unwrap_or(&ContentLength(0)).0; + let client_addr_cloned = client_addr.clone(); - super::connect_proxy_server(&handle, svr_cfg, addr) + super::connect_proxy_server(&handle, svr_cfg.clone(), addr) .and_then(move |(svr_r, svr_w)| { trace!("Going to pass req to server: {:?}", req); - req.write_to(svr_w) - .and_then(move |svr_w| { - trace!("Going to relay request body, len: {}", content_length); - if content_length == 0 { - // Do nothing because this request does not have a body - futures::finished((r, svr_w, remains)).boxed() - } else { - write_all(svr_w, remains) - .and_then(move |(svr_w, remains)| { - let remain_len = content_length - remains.len() as u64; - super::copy_exact(r, svr_w, remain_len as usize) - }) - .map(|(r, svr_w)| (r, svr_w, vec![])) - .boxed() - } + let svr_addr = svr_cfg.addr.clone(); + http::proxy_request((r, svr_w), client_addr, req, remains) + .and_then(move |(r, svr_w, req_remains)| { + HttpResponseFut::new(svr_r) + .and_then(move |(svr_r, rsp, rsp_remains)| { + http::proxy_response((svr_r, w), svr_addr, rsp, rsp_remains) + }) + .map(move |(svr_r, w, rsp_remains)| (r, w, svr_r, svr_w, req_remains, rsp_remains)) }) - .map(move |(r, svr_w, remains)| { - HttpRelayServer::handle_http_again((r, w), (svr_r, svr_w), remains) + .and_then(move |(r, w, svr_r, svr_w, req_remains, rsp_remains)| { + HttpRelayServer::handle_http_again((r, w), + (svr_r, svr_w), + client_addr_cloned, + (req_remains, rsp_remains), + svr_cfg) }) }) - .map(|_| ()) .boxed() } fn handle_client(handle: &Handle, socket: TcpStream, _: SocketAddr, svr_cfg: Arc) -> io::Result<()> { let cloned_handle = handle.clone(); + let client_addr = try!(socket.peer_addr()); let fut = futures::lazy(|| Ok(socket.split())) .and_then(|(r, w)| { - RequestReader::new(r).and_then(move |(r, mut req, remains)| { + HttpRequestFut::new(r).and_then(move |(r, mut req, remains)| { trace!("Got HTTP Request, version: {}, method: {}, uri: {}", req.version, req.method, @@ -372,7 +345,13 @@ impl HttpRelayServer { } met => { info!("{} (Http) {}", met, addr); - HttpRelayServer::handle_http_proxy(cloned_handle, (r, w), req, addr, remains, svr_cfg) + HttpRelayServer::handle_http_proxy(cloned_handle, + (r, w), + client_addr, + req, + addr, + remains, + svr_cfg) } } });