use super::super::http::{
HTTP_TIMEOUT, HttpSyncTransport, RawHttpClient, RawHttpRequest, RawHttpResponse,
SYNC_TRANSPORT_ERROR_CODE, response_too_large_limit,
};
use crate::LixError;
use crate::authority_client::{
ProtocolByteStream, ProtocolHttp, ProtocolHttpRequest, ProtocolHttpResponse, ProtocolHttpStream,
};
use crate::sync::SyncTransportFuture;
use bytes::Bytes;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
#[derive(Clone, Debug)]
pub(crate) struct AuthorityHttp(reqwest::Client);
pub(crate) fn authority_http(headers: &[(String, String)]) -> Result<AuthorityHttp, LixError> {
Ok(AuthorityHttp(build_client(headers)?))
}
impl AuthorityHttp {
pub(crate) async fn upload(
&self,
request: ProtocolHttpRequest,
body: Option<ProtocolByteStream>,
) -> Result<ProtocolHttpResponse, LixError> {
let method = request.method.parse::<reqwest::Method>().map_err(|error| {
LixError::new(
LixError::CODE_INVALID_PARAM,
format!("invalid HTTP method: {error}"),
)
})?;
let mut builder = self.0.request(method, request.url);
for (name, value) in request.headers {
builder = builder.header(name, value);
}
if let Some(body) = body {
builder = builder.body(reqwest::Body::wrap_stream(body));
}
let response = builder
.send()
.await
.map_err(|error| transport_error("upload repository", error))?;
let status = response.status().as_u16();
let headers = response
.headers()
.iter()
.map(|(name, value)| {
(
name.as_str().to_owned(),
value.to_str().unwrap_or_default().to_owned(),
)
})
.collect();
let mut response = response;
let mut bytes = Vec::new();
while let Some(chunk) = response
.chunk()
.await
.map_err(|error| transport_error("read creation response", error))?
{
if bytes.len() + chunk.len() > 64 * 1024 {
return Err(LixError::new(
"LIX_SERVER_PROTOCOL_ERROR",
"creation response exceeds 64 KiB",
));
}
bytes.extend_from_slice(&chunk);
}
Ok(ProtocolHttpResponse {
status,
headers,
body: Bytes::from(bytes),
})
}
}
impl HttpSyncTransport<reqwest::Client> {
pub(crate) async fn connect(
repository_url: &str,
headers: &[(String, String)],
) -> Result<Self, LixError> {
let client = build_client(headers)?;
Self::connect_with(client, repository_url).await
}
}
fn build_client(headers: &[(String, String)]) -> Result<reqwest::Client, LixError> {
let mut default_headers = reqwest::header::HeaderMap::new();
for (name, value) in headers {
if HttpSyncTransport::<reqwest::Client>::is_reserved_header(name) {
continue;
}
let name = reqwest::header::HeaderName::from_bytes(name.as_bytes()).map_err(|error| {
LixError::new(
LixError::CODE_INVALID_PARAM,
format!("invalid sync HTTP header name: {error}"),
)
})?;
let value = reqwest::header::HeaderValue::from_str(value).map_err(|error| {
LixError::new(
LixError::CODE_INVALID_PARAM,
format!("invalid sync HTTP header value: {error}"),
)
})?;
default_headers.append(name, value);
}
reqwest::Client::builder()
.default_headers(default_headers)
.redirect(reqwest::redirect::Policy::none())
.timeout(HTTP_TIMEOUT)
.build()
.map_err(|error| transport_error("configure sync transport", error))
}
impl ProtocolHttp for AuthorityHttp {
async fn request(
&self,
request: ProtocolHttpRequest,
) -> Result<ProtocolHttpResponse, LixError> {
let mut builder = self.0.request(
request.method.parse::<reqwest::Method>().map_err(|error| {
LixError::new(
LixError::CODE_INVALID_PARAM,
format!("invalid HTTP method: {error}"),
)
})?,
request.url,
);
for (name, value) in request.headers {
builder = builder.header(name, value);
}
if let Some(body) = request.body {
builder = builder.body(body);
}
let response = builder
.send()
.await
.map_err(|error| transport_error("authority request", error))?;
let status = response.status().as_u16();
let headers = response
.headers()
.iter()
.map(|(name, value)| {
(
name.as_str().to_owned(),
value.to_str().unwrap_or_default().to_owned(),
)
})
.collect();
let body = response
.bytes()
.await
.map_err(|error| transport_error("read authority response", error))?;
Ok(ProtocolHttpResponse {
status,
headers,
body,
})
}
async fn request_stream(
&self,
request: ProtocolHttpRequest,
) -> Result<ProtocolHttpStream, LixError> {
let mut builder = self.0.request(
request.method.parse::<reqwest::Method>().map_err(|error| {
LixError::new(
LixError::CODE_INVALID_PARAM,
format!("invalid HTTP method: {error}"),
)
})?,
request.url,
);
for (name, value) in request.headers {
builder = builder.header(name, value);
}
if let Some(body) = request.body {
builder = builder.body(body);
}
let response = builder
.send()
.await
.map_err(|error| transport_error("authority stream", error))?;
let status = response.status().as_u16();
let headers = response
.headers()
.iter()
.map(|(name, value)| {
(
name.as_str().to_owned(),
value.to_str().unwrap_or_default().to_owned(),
)
})
.collect();
let body: ProtocolByteStream = Box::pin(async_stream::try_stream! {
let mut response = response;
while let Some(chunk) = response.chunk().await.map_err(|error| transport_error("read authority stream", error))? {
yield Bytes::from(chunk);
}
});
let cancel: Arc<dyn Fn() + Send + Sync> = Arc::new(|| {});
Ok(ProtocolHttpStream {
status,
headers,
body,
cancel,
})
}
async fn sleep(&self, duration: Duration) {
tokio::time::sleep(duration).await;
}
fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
tokio::spawn(future);
}
}
impl RawHttpClient for reqwest::Client {
fn send(&self, request: RawHttpRequest) -> SyncTransportFuture<'_, RawHttpResponse> {
Box::pin(async move {
let mut builder = self.request(request.method, &request.url);
for (name, value) in &request.headers {
builder = builder.header(name, value);
}
if let Some(body) = request.body {
builder = builder.body(body);
}
let mut response = builder
.send()
.await
.map_err(|error| transport_error(request.operation, error))?;
let status = response.status();
let status_text = status
.canonical_reason()
.map(str::to_owned)
.unwrap_or_default();
if response
.content_length()
.is_some_and(|length| length > request.response_limit as u64)
{
return Err(response_too_large_limit(
request.operation,
request.response_limit,
));
}
let mut body = Vec::new();
while let Some(chunk) = response
.chunk()
.await
.map_err(|error| transport_error(request.operation, error))?
{
if body.len().saturating_add(chunk.len()) > request.response_limit {
return Err(response_too_large_limit(
request.operation,
request.response_limit,
));
}
body.extend_from_slice(&chunk);
}
Ok(RawHttpResponse {
status: status.as_u16(),
status_text,
body,
})
})
}
}
fn transport_error(operation: &str, error: impl std::fmt::Display) -> LixError {
LixError::new(SYNC_TRANSPORT_ERROR_CODE, format!("{operation}: {error}"))
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::{Read, Write};
#[tokio::test]
async fn native_receive_limit_rejects_chunked_body_without_content_length() {
for (limit, succeeds) in [(4, false), (6, true)] {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let (mut socket, _) = listener.accept().unwrap();
socket.set_read_timeout(Some(HTTP_TIMEOUT)).unwrap();
let mut received = Vec::new();
let mut buffer = [0; 1024];
while !received.windows(4).any(|bytes| bytes == b"\r\n\r\n") {
let count = socket.read(&mut buffer).unwrap();
assert!(count > 0, "client closed before request headers");
received.extend_from_slice(&buffer[..count]);
}
socket.write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n3\r\nabc\r\n3\r\ndef\r\n0\r\n\r\n").unwrap();
});
let client = build_client(&[]).unwrap();
let result = RawHttpClient::send(
&client,
RawHttpRequest {
method: http::Method::GET,
url: format!("http://{address}/native-limit"),
headers: Vec::new(),
body: None,
cache_immutable: false,
response_limit: limit,
operation: "native receive limit fixture",
},
)
.await;
server.join().unwrap();
if succeeds {
assert_eq!(result.unwrap().body, b"abcdef");
} else {
let error = result.unwrap_err();
assert_eq!(error.code, LixError::CODE_INVALID_PARAM);
assert!(error.message.contains("exceeds 4 bytes"));
}
}
}
}