use std::io::{Read as _, Write as _};
use std::time::Duration;
use flate2::read::GzDecoder;
use http::StatusCode;
use knot_types::RepoDid;
use crate::cgi::Request;
use crate::cob_write::WriteError;
use crate::config::ValidatedConfig;
pub const MAX_INFLATED_BYTES: u64 = 16 * 1024 * 1024;
pub const TIMEOUT_SECS: u64 = 300;
pub(super) fn repo(request: &Request) -> Result<RepoDid, WriteError> {
let not_a_fetch = || WriteError::invalid_request("only a gzipped git-upload-pack is inflated");
if request.method != "POST" {
return Err(WriteError::unsupported(
StatusCode::METHOD_NOT_ALLOWED,
"InvalidRequest",
"a fetch is a POST request",
));
}
if !request.content_encoding.eq_ignore_ascii_case("gzip") {
return Err(not_a_fetch());
}
let did = request
.path()
.strip_prefix('/')
.and_then(|rest| rest.strip_suffix("/git-upload-pack"))
.ok_or_else(not_a_fetch)?;
RepoDid::new(did).map_err(|_| not_a_fetch())
}
fn loopback(listen: &str) -> String {
let port = listen.rsplit(':').next().unwrap_or(listen);
format!("127.0.0.1:{port}")
}
pub async fn serve(config: &ValidatedConfig, request: &Request) -> Result<(), WriteError> {
let did = repo(request)?;
let body = inflated(std::io::stdin().lock())?;
let unavailable = |error: reqwest::Error| WriteError::upstream_unavailable(error.to_string());
let timeout = Duration::from_secs(TIMEOUT_SECS);
let mut call = reqwest::Client::builder()
.no_proxy()
.connect_timeout(Duration::from_secs(5))
.read_timeout(timeout)
.build()
.map_err(unavailable)?
.post(format!(
"http://{}/{did}/git-upload-pack",
loopback(&config.activate.listen)
))
.header("Content-Type", &request.content_type);
if let Some(protocol) = &request.git_protocol {
call = call.header("Git-Protocol", protocol);
}
// no X-Forwarded-For, nginx only takes a resolve from loopback and counts on handback.
let answer = call
.header("X-Ks-Resolve", "1")
.body(body)
.send()
.await
.map_err(unavailable)?;
let pack = answer
.headers()
.get("x-ks-pack")
.and_then(|value| value.to_str().ok());
if let Some(pack) = handed_back(answer.status(), pack)? {
let mut stdout = std::io::stdout().lock();
let _ = write!(stdout, "X-Accel-Redirect: {}\r\n\r\n", pack.as_str());
let _ = stdout.flush();
return Ok(());
}
let status = answer.status().as_u16();
let content_type = answer
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok())
.unwrap_or("application/octet-stream")
.to_string();
let body = answer.bytes().await.map_err(unavailable)?;
let mut stdout = std::io::stdout().lock();
let _ = write!(
stdout,
"Status: {status}\r\nContent-Type: {content_type}\r\nCache-Control: no-store\r\n\r\n",
);
let _ = stdout.write_all(&body);
let _ = stdout.flush();
Ok(())
}
pub(super) fn inflated(mut stdin: impl std::io::Read) -> Result<Vec<u8>, WriteError> {
let mut body = Vec::new();
let inflated = GzDecoder::new(&mut stdin)
.take(MAX_INFLATED_BYTES + 1)
.read_to_end(&mut body);
// drain it first, fcgiwrap answers an unread body with its own 502.
let _ = std::io::copy(&mut stdin, &mut std::io::sink());
inflated.map_err(|_| WriteError::invalid_request("the request body isn't gzip"))?;
if body.len() as u64 > MAX_INFLATED_BYTES {
return Err(WriteError::request_too_large(format!(
"a fetch request is at most {MAX_INFLATED_BYTES} bytes"
)));
}
Ok(body)
}
/// a handback is a 204 with `x-ks-pack`, a 2xx without one means nginx didn't resolve.
pub(super) fn handed_back(
status: StatusCode,
pack: Option<&str>,
) -> Result<Option<crate::nginx::PackRedirect>, WriteError> {
if !status.is_success() {
return Ok(None);
}
match (status, pack.and_then(crate::nginx::PackRedirect::parse)) {
(StatusCode::NO_CONTENT, Some(pack)) => Ok(Some(pack)),
_ => Err(WriteError::upstream_unavailable(format!(
"nginx answered a resolve with {status} and no reply to hand back"
))),
}
}
#[cfg(test)]
mod tests {
use super::*;
fn request(method: &str, target: &str, encoding: &str) -> Request {
Request {
method: method.into(),
target: target.into(),
authorization: None,
content_length: None,
garbled_length: false,
content_type: String::new(),
content_encoding: encoding.into(),
git_protocol: None,
inflating: true,
}
}
#[test]
fn only_a_resolved_reply_is_handed_back_and_only_an_error_relayed() {
let pack = "/_ks_pack/gen-1/abc_9/have-0-done";
let handed = handed_back(StatusCode::NO_CONTENT, Some(pack))
.unwrap()
.unwrap();
assert_eq!(handed.as_str(), pack);
for status in [
StatusCode::BAD_REQUEST,
StatusCode::NOT_FOUND,
StatusCode::BAD_GATEWAY,
] {
assert!(handed_back(status, None).unwrap().is_none(), "{status}");
}
for (status, pack) in [
(StatusCode::NO_CONTENT, None),
(StatusCode::NO_CONTENT, Some("/_ks_git/gen-1/abc/have-0")),
(StatusCode::NO_CONTENT, Some("/_ks_pack/gen-1/abc/../x")),
(StatusCode::NO_CONTENT, Some("/_ks_pack/gen-1/abc/Have-0")),
(StatusCode::NO_CONTENT, Some("/_ks_pack/gen-1/a.c/have-0")),
(StatusCode::NO_CONTENT, Some("/_ks_pack/gen-1/abc/have-0/x")),
(StatusCode::NO_CONTENT, Some("/_ks_pack//abc/have-0")),
(StatusCode::OK, Some(pack)),
(StatusCode::OK, None),
] {
assert!(handed_back(status, pack).is_err(), "{status} {pack:?}");
}
}
#[test]
fn only_a_gzipped_upload_pack_post_to_a_repo_did_is_inflated() {
let did = "did:plc:abcdefghijklmnopqrstuvwx";
let fetch = format!("/{did}/git-upload-pack");
assert_eq!(
repo(&request("POST", &fetch, "GZip")).unwrap().as_str(),
did
);
for (method, target, encoding) in [
("GET", fetch.as_str(), "gzip"),
("POST", fetch.as_str(), "br"),
("POST", "/xrpc/com.atproto.repo.createRecord", "gzip"),
("POST", "/../x/git-upload-pack", "gzip"),
("POST", "/a/b/git-upload-pack", "gzip"),
(
"POST",
&format!("/xrpc/createRecord#/{did}/git-upload-pack"),
"gzip",
),
] {
assert!(
repo(&request(method, target, encoding)).is_err(),
"{method} {target} {encoding}"
);
}
}
}