use std::io::{BufRead, Read, Write};
use std::os::fd::{AsFd, AsRawFd as _};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use anyhow::Context as _;
use knot_types::RepoDid;
use crate::config::ValidatedConfig;
use crate::manifest::GenerationManifest;
const MAX_REQUEST_BYTES: usize = 16 * 1024 * 1024;
pub const REQUEST_DEADLINE: Duration = Duration::from_secs(60);
const MAX_COMMANDS: usize = 64;
pub fn serve<R: Read + AsFd>(
config: &ValidatedConfig,
repo_did: &RepoDid,
protocol: Option<&str>,
input: Deadline<R>,
mut output: impl Write,
) -> anyhow::Result<()> {
if !protocol.is_some_and(|protocol| protocol.split(':').any(|item| item == "version=2")) {
return refuse(
&mut output,
"fetches over ssh need git protocol v2: git -c protocol.version=2 ...",
);
}
let Some((generation, mut manifest)) = active(&config.output_root)? else {
return refuse(&mut output, "this knot hasn't published anything yet");
};
// rows of other repos can't match, and each one costs a regex compile.
manifest
.repos
.retain(|repo| repo.repo_did == repo_did.as_str());
if manifest.repos.is_empty() {
return refuse(&mut output, "this repo isn't published yet");
}
let maps = crate::nginx::git_maps(&manifest)?;
let smart = generation
.join("www/_git")
.join(crate::plan::repo_token(repo_did.as_str()))
.join("smart");
send(&smart.join(maps.advertisement("version=2")?), &mut output)?;
session(input, &mut output, |request| {
Ok(maps
.reply(repo_did.as_str(), request)?
.map(|name| smart.join(name)))
})
}
fn session<R: Read + AsFd>(
input: Deadline<R>,
output: &mut impl Write,
mut answer: impl FnMut(&[u8]) -> anyhow::Result<Option<PathBuf>>,
) -> anyhow::Result<()> {
let mut input = std::io::BufReader::new(input);
for _ in 0..MAX_COMMANDS {
input.get_mut().restart();
let request = match read_request(&mut input) {
Ok(Some(request)) => request,
Ok(None) => return Ok(()),
Err(error) => return refuse(output, &format!("{error:#}")),
};
match answer(&request)? {
Some(file) => send(&file, output)?,
None => return refuse(output, "unsupported git request"),
}
}
refuse(
output,
&format!("one connection is answered at most {MAX_COMMANDS} commands"),
)
}
pub struct Deadline<R> {
inner: R,
budget: Duration,
until: Instant,
}
impl<R: Read + AsFd> Deadline<R> {
pub fn new(inner: R, budget: Duration) -> Self {
Self {
inner,
budget,
until: Instant::now() + budget,
}
}
fn restart(&mut self) {
self.until = Instant::now() + self.budget;
}
}
impl<R: Read + AsFd> Read for Deadline<R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let left = self.until.saturating_duration_since(Instant::now());
let mut fd = libc::pollfd {
fd: self.inner.as_fd().as_raw_fd(),
events: libc::POLLIN,
revents: 0,
};
let ms = libc::c_int::try_from(left.as_millis()).unwrap_or(libc::c_int::MAX);
// SAFETY: one pollfd this call owns, for an fd `inner` keeps open
match unsafe { libc::poll(&mut fd, 1, ms) } {
0 => Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
format!("a command takes at most {:?} to send", self.budget),
)),
n if n < 0 => Err(std::io::Error::last_os_error()),
_ => self.inner.read(buf),
}
}
}
fn active(output_root: &Path) -> anyhow::Result<Option<(PathBuf, GenerationManifest)>> {
let Some(id) = crate::publish::current_generation(output_root) else {
return Ok(None);
};
let generation = output_root.join("generations").join(id);
let manifest = GenerationManifest::load(&generation)
.with_context(|| format!("load {}", generation.display()))?;
Ok(Some((generation, manifest)))
}
fn read_request(input: &mut impl BufRead) -> anyhow::Result<Option<Vec<u8>>> {
let mut request = Vec::new();
loop {
let mut head = [0; 4];
match input.read_exact(&mut head) {
Err(error)
if error.kind() == std::io::ErrorKind::UnexpectedEof && request.is_empty() =>
{
return Ok(None);
}
other => other.context("read a pkt-line")?,
}
// from_str_radix alone would also take a sign, as in `+fff`
let len = std::str::from_utf8(&head)
.ok()
.filter(|head| head.bytes().all(|b| b.is_ascii_hexdigit()))
.and_then(|head| usize::from_str_radix(head, 16).ok())
.with_context(|| format!("a pkt-line length {head:?}"))?;
request.extend_from_slice(&head);
match len {
0 => return Ok((request.len() > 4).then_some(request)),
// 0001 is the delim between a command's capabilities and its arguments.
1 => {}
2 => anyhow::bail!("a request can't hold a response-end packet"),
3 => anyhow::bail!("0003 isn't a pkt-line"),
_ => {
let start = request.len();
request.resize(start + len - 4, 0);
input
.read_exact(&mut request[start..])
.context("read a pkt-line")?;
}
}
anyhow::ensure!(
request.len() <= MAX_REQUEST_BYTES,
"a fetch request is at most {MAX_REQUEST_BYTES} bytes"
);
}
}
fn send(file: &Path, output: &mut impl Write) -> anyhow::Result<()> {
let mut reply =
std::fs::File::open(file).with_context(|| format!("open {}", file.display()))?;
std::io::copy(&mut reply, output).with_context(|| format!("send {}", file.display()))?;
output.flush().context("flush the reply")
}
pub fn refuse(output: &mut impl Write, message: &str) -> anyhow::Result<()> {
output.write_all(&crate::git_export::pkt_line(
format!("ERR {message}\n").as_bytes(),
))?;
output.flush().context("send the refusal")
}
#[cfg(test)]
mod tests {
use std::os::unix::net::UnixStream;
use std::time::Duration;
use super::*;
fn pkt(text: &str) -> Vec<u8> {
crate::git_export::pkt_line(text.as_bytes())
}
fn ls_refs() -> Vec<u8> {
[
pkt("command=ls-refs\n"),
b"0001".to_vec(),
pkt("peel\n"),
b"0000".to_vec(),
]
.concat()
}
fn answered(sent: &[u8], budget: Duration) -> (String, Vec<Vec<u8>>) {
let (mut client, server) = UnixStream::pair().unwrap();
// a big request outgrows the socket buffer, so it can't be written up front.
let sent = sent.to_vec();
let writer = std::thread::spawn(move || {
let _ = client.write_all(&sent);
let _ = client.shutdown(std::net::Shutdown::Write);
});
let reply = tempfile::NamedTempFile::new().unwrap();
std::fs::write(reply.path(), "R").unwrap();
let mut requests = Vec::new();
let mut output = Vec::new();
session(Deadline::new(server, budget), &mut output, |request| {
requests.push(request.to_vec());
Ok(Some(reply.path().to_path_buf()))
})
.unwrap();
writer.join().unwrap();
(String::from_utf8(output).unwrap(), requests)
}
#[test]
fn each_command_is_read_through_its_flush_and_answered() {
let (output, requests) = answered(&[ls_refs(), ls_refs()].concat(), REQUEST_DEADLINE);
assert_eq!(output, "RR");
assert_eq!(requests, [ls_refs(), ls_refs()]);
}
#[test]
fn a_response_end_or_an_invalid_packet_is_refused() {
for bad in ["0002", "0003"] {
let (output, requests) =
answered(&[bad.as_bytes(), &ls_refs()].concat(), REQUEST_DEADLINE);
assert!(output.contains("ERR "), "{bad}: {output}");
assert!(requests.is_empty(), "{bad}");
}
}
#[test]
fn a_connection_is_answered_at_most_its_commands() {
let sent = ls_refs().repeat(MAX_COMMANDS + 1);
let (output, requests) = answered(&sent, REQUEST_DEADLINE);
assert_eq!(requests.len(), MAX_COMMANDS);
assert!(
output.ends_with(&format!("at most {MAX_COMMANDS} commands\n")),
"{output}"
);
}
#[test]
fn a_command_not_sent_within_the_deadline_is_refused() {
let (mut client, server) = UnixStream::pair().unwrap();
client.write_all(&pkt("command=ls-refs\n")).unwrap();
let (done, finished) = std::sync::mpsc::channel();
std::thread::spawn(move || {
let mut output = Vec::new();
session(
Deadline::new(server, Duration::from_millis(200)),
&mut output,
|_| Ok(None),
)
.unwrap();
done.send(String::from_utf8(output).unwrap()).unwrap();
});
let output = finished
.recv_timeout(Duration::from_secs(10))
.expect("the session waited past its deadline");
assert!(output.contains("a command takes at most 200ms"), "{output}");
drop(client);
}
/// the pkt-lines one request holds, checked as git would frame them.
fn packets(request: &[u8]) -> Vec<&[u8]> {
let mut at = 0;
let mut packets = Vec::new();
while at < request.len() {
let head = std::str::from_utf8(&request[at..at + 4]).unwrap();
assert!(head.bytes().all(|b| b.is_ascii_hexdigit()), "{head:?}");
let len = usize::from_str_radix(head, 16).unwrap();
let end = at + len.max(4);
packets.push(&request[at..end]);
at = end;
}
assert_eq!(at, request.len());
packets
}
#[test]
fn a_length_that_isnt_four_hex_digits_is_refused() {
for bad in [
"+00a", "-001", " 00a", "0x0a", "00g0", "\0\0\0\0", "000", "ÿÿ",
] {
// a lax parse of `+00a` would take `abcdef` as its packet and answer the flush
let sent = [bad.as_bytes(), b"abcdef0000"].concat();
let (output, requests) = answered(&sent, REQUEST_DEADLINE);
assert!(output.contains("ERR "), "{bad:?}: {output}");
assert!(requests.is_empty(), "{bad:?}: {requests:?}");
}
}
#[test]
fn a_request_past_its_cap_is_refused() {
let big = pkt(&"x".repeat(0xfff0 - 4));
let delims = b"0001".repeat(MAX_REQUEST_BYTES / 4 + 1);
let empties = b"0004".repeat(MAX_REQUEST_BYTES / 4 + 1);
for (what, sent) in [
(
"full packets",
big.repeat(MAX_REQUEST_BYTES / big.len() + 2),
),
("delims", delims),
("empty packets", empties),
] {
let (output, requests) = answered(&sent, REQUEST_DEADLINE);
assert!(
output.contains(&format!("at most {MAX_REQUEST_BYTES} bytes")),
"{what}: {output}"
);
assert!(requests.is_empty(), "{what}");
}
}
#[test]
fn a_packet_cut_short_is_refused() {
for cut in ["0010command", "0005"] {
let sent = [ls_refs(), cut.as_bytes().to_vec()].concat();
let (output, requests) = answered(&sent, REQUEST_DEADLINE);
assert_eq!(requests, [ls_refs()], "{cut}");
assert!(
output.starts_with("R") && output.contains("ERR "),
"{cut}: {output}"
);
}
}
fn noise(rng: &mut crate::test_rng::Rng) -> Vec<u8> {
match rng.below(8) {
0 => b"0000".to_vec(),
1 => b"0001".to_vec(),
2 => rng
.pick(&[&b"0002"[..], b"0003", b"+00a", b"zzzz"])
.to_vec(),
3 => {
let len = rng.range(0, 300);
let body: Vec<u8> = (0..len).map(|_| rng.next() as u8).collect();
let mut packet = format!("{:04x}", len + 4).into_bytes();
packet.extend(&body);
// a lying length cuts the stream mid-packet
if rng.odds(10) {
packet.truncate(rng.below(packet.len()));
}
packet
}
4 => ls_refs(),
5 => pkt(&format!("want {:040x}\n", rng.next())),
_ => (0..rng.range(1, 6)).map(|_| rng.next() as u8).collect(),
}
}
#[test]
fn session_fuzz_answers_framed_requests_or_refuses_and_always_ends() {
for seed in 0..3000u64 {
let rng = &mut crate::test_rng::Rng(seed);
let sent: Vec<u8> = (0..rng.range(0, 90)).flat_map(|_| noise(rng)).collect();
let (output, requests) = answered(&sent, Duration::from_secs(5));
assert!(requests.len() <= MAX_COMMANDS, "seed {seed}");
let answers = output.bytes().take_while(|&b| b == b'R').count();
assert_eq!(answers, requests.len(), "seed {seed}: {output}");
let rest = &output[answers..];
assert!(
rest.is_empty()
|| (rest.starts_with(|c: char| c.is_ascii_hexdigit())
&& rest[4..].starts_with("ERR ")),
"seed {seed}: {rest:?}"
);
for request in &requests {
assert!(request.len() <= MAX_REQUEST_BYTES + 0xffff, "seed {seed}");
let packets = packets(request);
assert_eq!(packets.last(), Some(&&b"0000"[..]), "seed {seed}");
assert!(packets.len() > 1, "seed {seed}: a bare flush was answered");
}
}
}
}