use super::*;
impl<T: AsyncRead + AsyncWrite + Unpin + Send> AsyncReadWrite for T {}
impl<T: Read + Write> ReadWrite for T {}
impl HttpRequest {
pub fn send(&mut self) -> RequestResult {
self.send_sync()
}
pub async fn send_async(&mut self) -> RequestResult {
self.send_async_impl().await
}
pub(crate) fn parse_url(&self) -> Result<HttpUrlComponents, RequestError> {
let (scheme, remainder): (&str, &str) = self
.get_url()
.split_once(SCHEME_SEPARATOR)
.ok_or(RequestError::InvalidUrlScheme(HttpStatus::BadRequest))?;
if !scheme.eq_ignore_ascii_case(HTTP_SCHEME)
&& !scheme.eq_ignore_ascii_case(HTTPS_LOWERCASE)
{
return Err(RequestError::InvalidUrlScheme(HttpStatus::BadRequest));
}
let authority_end: usize = remainder
.find([URL_SLASH, URL_QUERY, URL_FRAGMENT])
.unwrap_or(remainder.len());
let authority: &str = &remainder[..authority_end];
if authority.is_empty()
|| authority.contains(URL_BACKSLASH)
|| authority
.bytes()
.any(|byte: u8| byte.is_ascii_control() || byte == SPACE_U8)
{
return Err(RequestError::InvalidUrlHost(HttpStatus::BadRequest));
}
let target: &str = remainder[authority_end..]
.split(URL_FRAGMENT)
.next()
.unwrap_or_default();
let (path, query): (&str, &str) = target
.split_once(URL_QUERY)
.unwrap_or((target, Default::default()));
if path
.bytes()
.any(|byte: u8| byte.is_ascii_control() || byte == SPACE_U8)
{
return Err(RequestError::InvalidUrlPath(HttpStatus::BadRequest));
}
if query
.bytes()
.any(|byte: u8| byte.is_ascii_control() || byte == SPACE_U8)
{
return Err(RequestError::InvalidUrlQuery(HttpStatus::BadRequest));
}
let components: HttpUrlComponents = match HttpUrlComponents::parse(self.get_url()) {
Ok(value) => value,
Err(_) => return Err(RequestError::InvalidUrlHost(HttpStatus::BadRequest)),
};
let host: String = components.get_host().unwrap_or_default();
if host.is_empty()
|| host
.bytes()
.any(|byte: u8| byte.is_ascii_control() || byte == SPACE_U8)
{
return Err(RequestError::InvalidUrlHost(HttpStatus::BadRequest));
}
Ok(components)
}
pub(crate) fn full_path(&self) -> Result<String, RequestError> {
let url_obj: HttpUrlComponents = self.parse_url()?;
let query: String = url_obj.get_query().unwrap_or_default();
let path: String = url_obj.get_path().unwrap_or_default();
if query.is_empty() {
Ok(path)
} else {
Ok(format!("{}{}{}", path, QUERY, query))
}
}
pub(crate) fn protocol_lower(config: &RequestConfig) -> String {
config.get_http_version().to_string().to_ascii_lowercase()
}
pub(crate) fn header_bytes(&self, body_length: usize) -> Result<Vec<u8>, RequestError> {
let mut header: HashMap<String, String> = self.get_headers().clone();
let host_value: String = self.parse_url()?.get_host().unwrap_or_default();
let content_length_value: String = body_length.to_string();
if !Self::header_has_key(&header, HOST) {
header.insert(HOST.to_ascii_lowercase(), host_value);
}
if !Self::header_has_key(&header, CONTENT_LENGTH) {
header.insert(CONTENT_LENGTH.to_ascii_lowercase(), content_length_value);
}
if !Self::header_has_key(&header, ACCEPT) {
header.insert(ACCEPT.to_ascii_lowercase(), ACCEPT_ANY.to_owned());
}
if !Self::header_has_key(&header, USER_AGENT) {
header.insert(USER_AGENT.to_ascii_lowercase(), APP_NAME.to_owned());
}
validate_request_headers(&header)?;
let estimated_size: usize = header
.iter()
.map(|(k, v): (&String, &String)| k.len() + v.len() + 4)
.sum();
let mut out: Vec<u8> = Vec::with_capacity(estimated_size);
for (key, value) in &header {
out.extend_from_slice(key.as_bytes());
out.extend_from_slice(HEADER_VALUE_SEPARATOR);
out.extend_from_slice(value.as_bytes());
out.extend_from_slice(HTTP_BR_BYTES);
}
Ok(out)
}
fn header_has_key(header: &HashMap<String, String>, target_key: &str) -> bool {
header
.keys()
.any(|key: &String| key.eq_ignore_ascii_case(target_key))
}
pub(crate) fn body_bytes(&self) -> Vec<u8> {
let ct: Option<String> = self
.headers
.iter()
.find(|(k, _): &(&String, &String)| k.eq_ignore_ascii_case(CONTENT_TYPE))
.map(|(_, v): (&String, &String)| v.clone());
match ct {
Some(value) => value
.to_lowercase()
.parse::<ContentType>()
.unwrap_or_default()
.get_body_string(&String::from_utf8_lossy(self.get_body().as_slice()))
.into_bytes(),
None => Vec::new(),
}
}
pub(crate) fn send_sync(&mut self) -> RequestResult {
let url_obj: HttpUrlComponents = self.parse_url()?;
validate_request_headers(self.get_headers())?;
let host: String = url_obj.get_host().clone().unwrap_or_default();
let port: u16 = self.resolve_port(url_obj.get_port().unwrap_or_default());
let mut stream: BoxReadWrite = self.open_sync_stream(host, port)?;
let method: Method = self.get_method();
if method.is_get() {
self.send_get_request_sync(&mut stream)
} else if method.is_post() {
self.send_post_request_sync(&mut stream)
} else {
Err(RequestError::Request(METHOD_NOT_ALLOWED.to_string()))
}
}
async fn send_async_impl(&mut self) -> RequestResult {
let url_obj: HttpUrlComponents = self.parse_url()?;
validate_request_headers(self.get_headers())?;
let host: String = url_obj.get_host().clone().unwrap_or_default();
let port: u16 = self.resolve_port(url_obj.get_port().unwrap_or_default());
let mut stream: BoxAsyncReadWrite = self.open_async_stream(host, port).await?;
let method: Method = self.get_method();
if method.is_get() {
self.send_get_request_async(&mut stream).await
} else if method.is_post() {
self.send_post_request_async(&mut stream).await
} else {
Err(RequestError::Request(METHOD_NOT_ALLOWED.to_string()))
}
}
fn resolve_port(&self, port: u16) -> u16 {
if port != 0 {
return port;
}
let protocol: String = Self::protocol_lower(self.get_config());
Protocol::get_port(&protocol)
}
fn is_https(&self) -> bool {
Self::protocol_lower(self.get_config()) == HTTPS_LOWERCASE
}
fn open_sync_stream(&self, host: String, port: u16) -> Result<BoxReadWrite, RequestError> {
if let Some(proxy) = &self.get_config().get_proxy() {
return self.open_sync_proxy_stream(host, port, proxy);
}
let timeout: Duration = Duration::from_millis(self.get_config().get_timeout());
let tcp: TcpStream = TcpStream::connect((host.clone(), port))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
tcp.set_read_timeout(Some(timeout))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
tcp.set_write_timeout(Some(timeout))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if self.is_https() {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let dns: ServerName<'static> = ServerName::try_from(host.clone())
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let session: ClientConnection = ClientConnection::new(Arc::new(tls_cfg), dns)
.map_err(|e: rustls::Error| RequestError::Request(e.to_string()))?;
Ok(Box::new(StreamOwned::new(session, tcp)))
} else {
Ok(Box::new(tcp))
}
}
fn send_get_request_sync(
&mut self,
stream: &mut BoxReadWrite,
) -> Result<HttpResponse, RequestError> {
let path: String = self.full_path()?;
let header_bytes: Vec<u8> = self.header_bytes(0)?;
let version: String = self.get_config().get_http_version().to_string();
let request: Vec<u8> = build_http_request("GET", path, header_bytes, None, version);
stream
.write_all(&request)
.and_then(|_: ()| stream.flush())
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
self.read_response_sync(stream)
}
fn send_post_request_sync(
&mut self,
stream: &mut BoxReadWrite,
) -> Result<HttpResponse, RequestError> {
let body_bytes: Vec<u8> = self.body_bytes();
let path: String = self.full_path()?;
let header_bytes: Vec<u8> = self.header_bytes(body_bytes.len())?;
let version: String = self.get_config().get_http_version().to_string();
let request: Vec<u8> = build_http_request(
HTTP_METHOD_POST,
path,
header_bytes,
Some(body_bytes),
version,
);
stream
.write_all(&request)
.and_then(|_: ()| stream.flush())
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
self.read_response_sync(stream)
}
fn read_response_sync(
&mut self,
stream: &mut BoxReadWrite,
) -> Result<HttpResponse, RequestError> {
let buffer_size: usize = self.get_config().buffer_size;
let mut buffer: Vec<u8> = vec![0u8; buffer_size];
let mut response_bytes: Vec<u8> = Vec::with_capacity(buffer_size.max(8192));
let mut headers_done: bool = false;
let mut content_length: usize = 0usize;
let mut redirect_url: Option<Vec<u8>> = None;
let mut headers_end_pos: usize = 0usize;
let mut is_chunked: bool = false;
let version_bytes: Vec<u8> = self
.config
.get_http_version()
.to_string()
.to_ascii_lowercase()
.into_bytes();
let location_key: Vec<u8> = format!("{}:", LOCATION.to_ascii_lowercase()).into_bytes();
loop {
let n: usize = stream
.read(&mut buffer)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if n == 0 {
break;
}
let new_cap: usize =
calculate_buffer_capacity(&response_bytes, n, response_bytes.capacity());
if new_cap > 0 {
response_bytes.reserve(new_cap - response_bytes.capacity());
}
let old_len: usize = response_bytes.len();
response_bytes.extend_from_slice(&buffer[..n]);
if !headers_done {
let search_start: usize = old_len.saturating_sub(3);
if let Some(pos) = find_double_crlf(&response_bytes, search_start) {
headers_done = true;
headers_end_pos = pos + 4;
parse_response_headers(
&response_bytes[..headers_end_pos],
&version_bytes,
&location_key,
&mut content_length,
&mut redirect_url,
&mut is_chunked,
)?;
}
}
if headers_done {
if is_chunked {
if Self::is_chunked_response_complete(&response_bytes[headers_end_pos..])? {
break;
}
} else if response_bytes.len() >= headers_end_pos + content_length {
response_bytes.truncate(headers_end_pos + content_length);
break;
}
}
}
if !headers_done {
return Err(RequestError::HttpRequestPartsInsufficient(
HttpStatus::BadRequest,
));
}
if !is_chunked && response_bytes.len() < headers_end_pos + content_length {
return Err(RequestError::InvalidContentLength(HttpStatus::BadRequest));
}
if is_chunked {
let body_bytes: Vec<u8> = response_bytes[headers_end_pos..].to_vec();
let decoded: Vec<u8> = parse_chunked_body(&body_bytes)?;
response_bytes.truncate(headers_end_pos);
response_bytes.extend_from_slice(&decoded);
}
let mut response: HttpResponse = HttpResponse::from_bytes(&response_bytes);
if !self.get_config().get_redirect() || redirect_url.is_none() {
if self.get_config().get_decode() {
response = response.decode(self.get_config().buffer_size);
}
return Ok(response);
}
let url_bytes: Vec<u8> = redirect_url
.take()
.ok_or_else(|| RequestError::Request(MISSING_REDIRECT_URL.to_string()))?;
let url: String = String::from_utf8(url_bytes)
.map_err(|e: FromUtf8Error| RequestError::Request(e.to_string()))?;
self.handle_redirect(url)
}
fn handle_redirect(&mut self, url: String) -> Result<HttpResponse, RequestError> {
if !self.get_config().get_redirect() {
return Err(RequestError::Request(REDIRECT_NOT_ENABLED.to_string()));
}
if self.get_tmp().get_visit_url().contains(&url) {
return Err(RequestError::Request(REDIRECT_URL_DEAD_LOOP.to_string()));
}
self.get_mut_tmp().get_mut_visit_url().insert(url.clone());
if self.get_config().get_max_redirect_times() == 0 {
return Err(RequestError::Request(
MAX_REDIRECT_TIMES_EXCEEDED.to_string(),
));
}
let remaining: usize = self.get_config().get_max_redirect_times();
self.get_mut_config()
.set_max_redirect_times(remaining.saturating_sub(1));
self.set_url(url);
self.send_sync()
}
fn is_chunked_response_complete(body_bytes: &[u8]) -> Result<bool, RequestError> {
chunked_body_end(body_bytes).map(|end: Option<usize>| end.is_some())
}
async fn open_async_stream(
&self,
host: String,
port: u16,
) -> Result<BoxAsyncReadWrite, RequestError> {
if let Some(proxy) = &self.get_config().get_proxy() {
return self.open_async_proxy_stream(host, port, proxy).await;
}
let tcp: http_type::tokio::net::TcpStream =
http_type::tokio::net::TcpStream::connect((host.clone(), port))
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if self.is_https() {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let connector: TlsConnector = TlsConnector::from(Arc::new(tls_cfg));
let dns: ServerName<'static> = ServerName::try_from(host.clone())
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let tls: tokio_rustls::client::TlsStream<http_type::tokio::net::TcpStream> = connector
.connect(dns, tcp)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
Ok(Box::new(tls))
} else {
Ok(Box::new(tcp))
}
}
async fn send_get_request_async(
&mut self,
stream: &mut BoxAsyncReadWrite,
) -> Result<HttpResponse, RequestError> {
let path: String = self.full_path()?;
let header_bytes: Vec<u8> = self.header_bytes(0)?;
let version: String = self.get_config().get_http_version().to_string();
let request: Vec<u8> = build_http_request("GET", path, header_bytes, None, version);
stream
.write_all(&request)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
stream
.flush()
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
self.read_response_async(stream).await
}
async fn send_post_request_async(
&mut self,
stream: &mut BoxAsyncReadWrite,
) -> Result<HttpResponse, RequestError> {
let body_bytes: Vec<u8> = self.body_bytes();
let path: String = self.full_path()?;
let header_bytes: Vec<u8> = self.header_bytes(body_bytes.len())?;
let version: String = self.get_config().get_http_version().to_string();
let request: Vec<u8> = build_http_request(
HTTP_METHOD_POST,
path,
header_bytes,
Some(body_bytes),
version,
);
stream
.write_all(&request)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
stream
.flush()
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
self.read_response_async(stream).await
}
async fn read_response_async(
&mut self,
stream: &mut BoxAsyncReadWrite,
) -> Result<HttpResponse, RequestError> {
let buffer_size: usize = self.get_config().buffer_size;
let mut buffer: Vec<u8> = vec![0u8; buffer_size];
let mut response_bytes: Vec<u8> = Vec::with_capacity(buffer_size.max(8192));
let mut headers_done: bool = false;
let mut content_length: usize = 0usize;
let mut redirect_url: Option<Vec<u8>> = None;
let mut headers_end_pos: usize = 0usize;
let mut is_chunked: bool = false;
let version_bytes: Vec<u8> = self
.config
.get_http_version()
.to_string()
.to_ascii_lowercase()
.into_bytes();
let location_key: Vec<u8> = format!("{}:", LOCATION.to_ascii_lowercase()).into_bytes();
loop {
let n: usize = stream
.read(&mut buffer)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if n == 0 {
break;
}
let new_cap: usize =
calculate_buffer_capacity(&response_bytes, n, response_bytes.capacity());
if new_cap > 0 {
response_bytes.reserve(new_cap - response_bytes.capacity());
}
let old_len: usize = response_bytes.len();
response_bytes.extend_from_slice(&buffer[..n]);
if !headers_done {
let search_start: usize = old_len.saturating_sub(3);
if let Some(pos) = find_double_crlf(&response_bytes, search_start) {
headers_done = true;
headers_end_pos = pos + 4;
parse_response_headers(
&response_bytes[..headers_end_pos],
&version_bytes,
&location_key,
&mut content_length,
&mut redirect_url,
&mut is_chunked,
)?;
}
}
if headers_done {
if is_chunked {
if Self::is_chunked_response_complete(&response_bytes[headers_end_pos..])? {
break;
}
} else if response_bytes.len() >= headers_end_pos + content_length {
response_bytes.truncate(headers_end_pos + content_length);
break;
}
}
}
if !headers_done {
return Err(RequestError::HttpRequestPartsInsufficient(
HttpStatus::BadRequest,
));
}
if !is_chunked && response_bytes.len() < headers_end_pos + content_length {
return Err(RequestError::InvalidContentLength(HttpStatus::BadRequest));
}
if is_chunked {
let body_bytes: Vec<u8> = response_bytes[headers_end_pos..].to_vec();
let decoded: Vec<u8> = parse_chunked_body(&body_bytes)?;
response_bytes.truncate(headers_end_pos);
response_bytes.extend_from_slice(&decoded);
}
let mut response: HttpResponse = HttpResponse::from_bytes(&response_bytes);
if !self.get_config().get_redirect() || redirect_url.is_none() {
if self.get_config().get_decode() {
response = response.decode(self.get_config().buffer_size);
}
return Ok(response);
}
let url_bytes: Vec<u8> = redirect_url
.take()
.ok_or_else(|| RequestError::Request(MISSING_REDIRECT_URL.to_string()))?;
let url: String = String::from_utf8(url_bytes)
.map_err(|e: FromUtf8Error| RequestError::Request(e.to_string()))?;
self.handle_redirect_async(url).await
}
async fn handle_redirect_async(&mut self, url: String) -> Result<HttpResponse, RequestError> {
if !self.get_config().get_redirect() {
return Err(RequestError::Request(REDIRECT_NOT_ENABLED.to_string()));
}
if self.get_tmp().get_visit_url().contains(&url) {
return Err(RequestError::Request(REDIRECT_URL_DEAD_LOOP.to_string()));
}
self.get_mut_tmp().get_mut_visit_url().insert(url.clone());
if self.get_config().get_max_redirect_times() == 0 {
return Err(RequestError::Request(
MAX_REDIRECT_TIMES_EXCEEDED.to_string(),
));
}
let remaining: usize = self.get_config().get_max_redirect_times();
self.get_mut_config()
.set_max_redirect_times(remaining.saturating_sub(1));
self.set_url(url);
Box::pin(self.send_async()).await
}
fn open_sync_proxy_stream(
&self,
target_host: String,
target_port: u16,
proxy: &Proxy,
) -> Result<BoxReadWrite, RequestError> {
match *proxy.get_proxy_type() {
ProxyType::Http | ProxyType::Https => {
self.open_sync_http_proxy(target_host, target_port, proxy)
}
ProxyType::Socks5 => self.open_sync_socks5_proxy(target_host, target_port, proxy),
}
}
fn open_sync_http_proxy(
&self,
target_host: String,
target_port: u16,
proxy: &Proxy,
) -> Result<BoxReadWrite, RequestError> {
let timeout: Duration = Duration::from_millis(self.get_config().get_timeout());
let tcp: TcpStream = TcpStream::connect((proxy.get_host().clone(), proxy.get_port()))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
tcp.set_read_timeout(Some(timeout))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
tcp.set_write_timeout(Some(timeout))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut proxy_stream: BoxReadWrite = if proxy.get_proxy_type() == &ProxyType::Https {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let dns: ServerName<'static> = ServerName::try_from(proxy.get_host().clone())
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let session: ClientConnection = ClientConnection::new(Arc::new(tls_cfg), dns)
.map_err(|e: rustls::Error| RequestError::Request(e.to_string()))?;
Box::new(StreamOwned::new(session, tcp))
} else {
Box::new(tcp)
};
let connect_request: String = if let (Some(u), Some(p)) =
(proxy.get_username(), proxy.get_password())
{
let auth: String = format!("{u}:{p}");
let encoded: String = crate::utils::base64_encode(auth.as_bytes());
format!(
"CONNECT {target_host}:{target_port} HTTP/1.1\r\nHost: {target_host}:{target_port}\r\nProxy-Authorization: Basic {encoded}\r\n\r\n"
)
} else {
format!(
"CONNECT {target_host}:{target_port} HTTP/1.1\r\nHost: {target_host}:{target_port}\r\n\r\n"
)
};
proxy_stream
.write_all(connect_request.as_bytes())
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
proxy_stream
.flush()
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut buf: [u8; 1024] = [0u8; 1024];
let n: usize = proxy_stream
.read(&mut buf)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let s: &str = from_utf8(&buf[..n]).unwrap_or("");
let pre_read: Vec<u8> = if let Some(pos) = s.find(HEADER_TERMINATOR) {
let header_part: &str = &s[..pos];
if !header_part.starts_with(HTTP_1_1_OK_PREFIX)
&& !header_part.starts_with(HTTP_1_0_OK_PREFIX)
{
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
buf[pos + 4..n].to_vec()
} else {
if !s.starts_with(HTTP_1_1_OK_PREFIX) && !s.starts_with(HTTP_1_0_OK_PREFIX) {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
Vec::new()
};
if self.is_https() {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let dns: ServerName<'static> = ServerName::try_from(target_host.clone())
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let session: ClientConnection = ClientConnection::new(Arc::new(tls_cfg), dns)
.map_err(|e: rustls::Error| RequestError::Request(e.to_string()))?;
let tunnel: SyncProxyTunnelStream = SyncProxyTunnelStream::new(proxy_stream, pre_read);
return Ok(Box::new(StreamOwned::new(session, tunnel)));
}
Ok(Box::new(SyncProxyTunnelStream::new(proxy_stream, pre_read)))
}
fn open_sync_socks5_proxy(
&self,
target_host: String,
target_port: u16,
proxy: &Proxy,
) -> Result<BoxReadWrite, RequestError> {
let timeout: Duration = Duration::from_millis(self.get_config().get_timeout());
let mut tcp: TcpStream = TcpStream::connect((proxy.get_host().clone(), proxy.get_port()))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
tcp.set_read_timeout(Some(timeout))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
tcp.set_write_timeout(Some(timeout))
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let auth_methods: Vec<u8> =
if proxy.get_username().is_some() && proxy.get_password().is_some() {
vec![0x05, 0x02, 0x00, 0x02]
} else {
vec![0x05, 0x01, 0x00]
};
tcp.write_all(&auth_methods)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut resp: [u8; 2] = [0u8; 2];
tcp.read_exact(&mut resp)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if resp[0] != 0x05 {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
match resp[1] {
0x00 => {}
0x02 => {
let (Some(u), Some(p)) = (proxy.get_username(), proxy.get_password()) else {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
};
let mut auth_req: Vec<u8> = vec![0x01];
auth_req.push(u.len() as u8);
auth_req.extend_from_slice(u.as_bytes());
auth_req.push(p.len() as u8);
auth_req.extend_from_slice(p.as_bytes());
tcp.write_all(&auth_req)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut auth_resp: [u8; 2] = [0u8; 2];
tcp.read_exact(&mut auth_resp)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if auth_resp[1] != 0x00 {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
}
_ => return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string())),
}
let mut connect_req: Vec<u8> = vec![0x05, 0x01, 0x00];
if let Ok(ip) = target_host.parse::<Ipv4Addr>() {
connect_req.push(0x01);
connect_req.extend_from_slice(&ip.octets());
} else if let Ok(ip) = target_host.parse::<Ipv6Addr>() {
connect_req.push(0x04);
connect_req.extend_from_slice(&ip.octets());
} else {
connect_req.push(0x03);
connect_req.push(target_host.len() as u8);
connect_req.extend_from_slice(target_host.as_bytes());
}
connect_req.extend_from_slice(&target_port.to_be_bytes());
tcp.write_all(&connect_req)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut connect_resp: [u8; 4] = [0u8; 4];
tcp.read_exact(&mut connect_resp)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if connect_resp[0] != 0x05 || connect_resp[1] != 0x00 {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
match connect_resp[3] {
0x01 => {
let mut skip: [u8; 6] = [0u8; 6];
tcp.read_exact(&mut skip)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
}
0x03 => {
let mut len: [u8; 1] = [0u8; 1];
tcp.read_exact(&mut len)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut skip: Vec<u8> = vec![0u8; len[0] as usize + 2];
tcp.read_exact(&mut skip)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
}
0x04 => {
let mut skip: [u8; 18] = [0u8; 18];
tcp.read_exact(&mut skip)
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
}
_ => return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string())),
}
if self.is_https() {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let dns: ServerName<'static> = ServerName::try_from(target_host)
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let session: ClientConnection = ClientConnection::new(Arc::new(tls_cfg), dns)
.map_err(|e: rustls::Error| RequestError::Request(e.to_string()))?;
let proxy_box: BoxReadWrite = Box::new(tcp);
let tunnel: SyncProxyTunnelStream = SyncProxyTunnelStream::new(proxy_box, Vec::new());
return Ok(Box::new(StreamOwned::new(session, tunnel)));
}
Ok(Box::new(tcp))
}
async fn open_async_proxy_stream(
&self,
target_host: String,
target_port: u16,
proxy: &Proxy,
) -> Result<BoxAsyncReadWrite, RequestError> {
match *proxy.get_proxy_type() {
ProxyType::Http | ProxyType::Https => {
self.open_async_http_proxy(target_host, target_port, proxy)
.await
}
ProxyType::Socks5 => {
self.open_async_socks5_proxy(target_host, target_port, proxy)
.await
}
}
}
async fn open_async_http_proxy(
&self,
target_host: String,
target_port: u16,
proxy: &Proxy,
) -> Result<BoxAsyncReadWrite, RequestError> {
let tcp: http_type::tokio::net::TcpStream =
http_type::tokio::net::TcpStream::connect((proxy.get_host().clone(), proxy.get_port()))
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut proxy_stream: BoxAsyncReadWrite = if proxy.get_proxy_type() == &ProxyType::Https {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let connector: TlsConnector = TlsConnector::from(Arc::new(tls_cfg));
let dns: ServerName<'static> = ServerName::try_from(proxy.get_host().clone())
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let tls: tokio_rustls::client::TlsStream<http_type::tokio::net::TcpStream> = connector
.connect(dns, tcp)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
Box::new(tls)
} else {
Box::new(tcp)
};
let connect_request: String = if let (Some(u), Some(p)) =
(proxy.get_username(), proxy.get_password())
{
let auth: String = format!("{u}:{p}");
let encoded: String = crate::utils::base64_encode(auth.as_bytes());
format!(
"CONNECT {target_host}:{target_port} HTTP/1.1\r\nHost: {target_host}:{target_port}\r\nProxy-Authorization: Basic {encoded}\r\n\r\n"
)
} else {
format!(
"CONNECT {target_host}:{target_port} HTTP/1.1\r\nHost: {target_host}:{target_port}\r\n\r\n"
)
};
proxy_stream
.write_all(connect_request.as_bytes())
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
proxy_stream
.flush()
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut buf: [u8; 1024] = [0u8; 1024];
let n: usize = proxy_stream
.read(&mut buf)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let s: &str = from_utf8(&buf[..n]).unwrap_or("");
let pre_read: Vec<u8> = if let Some(pos) = s.find(HEADER_TERMINATOR) {
let header_part: &str = &s[..pos];
if !header_part.starts_with(HTTP_1_1_OK_PREFIX)
&& !header_part.starts_with(HTTP_1_0_OK_PREFIX)
{
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
buf[pos + 4..n].to_vec()
} else {
if !s.starts_with(HTTP_1_1_OK_PREFIX) && !s.starts_with(HTTP_1_0_OK_PREFIX) {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
Vec::new()
};
if self.is_https() {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let connector: TlsConnector = TlsConnector::from(Arc::new(tls_cfg));
let dns: ServerName<'static> = ServerName::try_from(target_host.clone())
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let tunnel: ProxyTunnelStream = ProxyTunnelStream::new(proxy_stream, pre_read);
let tls: tokio_rustls::client::TlsStream<ProxyTunnelStream> = connector
.connect(dns, tunnel)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
Ok(Box::new(tls))
} else {
Ok(Box::new(ProxyTunnelStream::new(proxy_stream, pre_read)))
}
}
async fn open_async_socks5_proxy(
&self,
target_host: String,
target_port: u16,
proxy: &Proxy,
) -> Result<BoxAsyncReadWrite, RequestError> {
let mut tcp: http_type::tokio::net::TcpStream =
http_type::tokio::net::TcpStream::connect((proxy.get_host().clone(), proxy.get_port()))
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let auth_methods: Vec<u8> =
if proxy.get_username().is_some() && proxy.get_password().is_some() {
vec![0x05, 0x02, 0x00, 0x02]
} else {
vec![0x05, 0x01, 0x00]
};
tcp.write_all(&auth_methods)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut resp: [u8; 2] = [0u8; 2];
tcp.read_exact(&mut resp)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if resp[0] != 0x05 {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
match resp[1] {
0x00 => {}
0x02 => {
let (Some(u), Some(p)) = (proxy.get_username(), proxy.get_password()) else {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
};
let mut auth_req: Vec<u8> = vec![0x01u8];
auth_req.push(u.len() as u8);
auth_req.extend_from_slice(u.as_bytes());
auth_req.push(p.len() as u8);
auth_req.extend_from_slice(p.as_bytes());
tcp.write_all(&auth_req)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut auth_resp: [u8; 2] = [0u8; 2];
tcp.read_exact(&mut auth_resp)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if auth_resp[1] != 0x00 {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
}
_ => return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string())),
}
let mut connect_req: Vec<u8> = vec![0x05, 0x01, 0x00];
if let Ok(ip) = target_host.parse::<Ipv4Addr>() {
connect_req.push(0x01);
connect_req.extend_from_slice(&ip.octets());
} else if let Ok(ip) = target_host.parse::<Ipv6Addr>() {
connect_req.push(0x04);
connect_req.extend_from_slice(&ip.octets());
} else {
connect_req.push(0x03);
connect_req.push(target_host.len() as u8);
connect_req.extend_from_slice(target_host.as_bytes());
}
connect_req.extend_from_slice(&target_port.to_be_bytes());
tcp.write_all(&connect_req)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut connect_resp: [u8; 4] = [0u8; 4];
tcp.read_exact(&mut connect_resp)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
if connect_resp[0] != 0x05 || connect_resp[1] != 0x00 {
return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string()));
}
match connect_resp[3] {
0x01 => {
let mut skip: [u8; 6] = [0u8; 6];
tcp.read_exact(&mut skip)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
}
0x03 => {
let mut len: [u8; 1] = [0u8; 1];
tcp.read_exact(&mut len)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
let mut skip: Vec<u8> = vec![0u8; len[0] as usize + 2];
tcp.read_exact(&mut skip)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
}
0x04 => {
let mut skip: [u8; 18] = [0u8; 18];
tcp.read_exact(&mut skip)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
}
_ => return Err(RequestError::Request(INTERNAL_SERVER_ERROR.to_string())),
}
if self.is_https() {
let roots: RootCertStore = self.get_tmp().get_root_cert().clone();
let tls_cfg: ClientConfig = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
let connector: TlsConnector = TlsConnector::from(Arc::new(tls_cfg));
let dns: ServerName<'static> = ServerName::try_from(target_host)
.map_err(|e: InvalidDnsNameError| RequestError::Request(e.to_string()))?;
let proxy_box: BoxAsyncReadWrite = Box::new(tcp);
let tunnel: ProxyTunnelStream = ProxyTunnelStream::new(proxy_box, Vec::new());
let tls: tokio_rustls::client::TlsStream<ProxyTunnelStream> = connector
.connect(dns, tunnel)
.await
.map_err(|e: io::Error| RequestError::Request(e.to_string()))?;
Ok(Box::new(tls))
} else {
Ok(Box::new(tcp))
}
}
}
impl HttpRequest {
pub fn get<T: Into<String>>(url: T) -> Self {
Self {
method: Method::Get,
url: url.into(),
headers: HashMap::new(),
body: Body::default(),
config: RequestConfig::default(),
tmp: Tmp::default(),
}
}
pub fn post<T: Into<String>>(url: T) -> Self {
Self {
method: Method::Post,
url: url.into(),
headers: HashMap::new(),
body: Body::default(),
config: RequestConfig::default(),
tmp: Tmp::default(),
}
}
pub fn set_method(&mut self, method: Method) -> &mut Self {
self.method = method;
self
}
pub fn set_url<T: Into<String>>(&mut self, url: T) -> &mut Self {
self.url = url.into();
self
}
pub fn set_header<K: AsRef<str>, V: AsRef<str>>(&mut self, key: K, value: V) -> &mut Self {
let normalized: String = Self::normalize_header_key(key.as_ref());
self.headers.insert(normalized, value.as_ref().to_owned());
self
}
pub fn remove_header<K: AsRef<str>>(&mut self, key: K) -> &mut Self {
let normalized: String = Self::normalize_header_key(key.as_ref());
self.get_mut_headers().remove(&normalized);
self
}
pub fn clear_headers(&mut self) -> &mut Self {
self.get_mut_headers().clear();
self
}
pub fn set_body(&mut self, body: Body) -> &mut Self {
self.body = body;
self
}
pub fn set_config(&mut self, config: RequestConfig) -> &mut Self {
self.config = config;
self
}
fn normalize_header_key(key: &str) -> String {
key.to_ascii_lowercase()
}
}