use anyhow::{bail, ensure, Context, Result};
use md4::{Digest, Md4};
use serde_json::{json, Value};
use tokio::io::{AsyncRead, AsyncReadExt};
pub const PROTOCOL: i32 = 29;
pub const GREETING: &str = "@RSYNCD: 29.0\n";
pub const NDX_DONE: i32 = -1;
const MPLEX_BASE: u8 = 7;
pub const MSG_DATA: u8 = 0;
pub const MSG_ERROR_XFER: u8 = 1;
pub const MSG_INFO: u8 = 2;
pub const MSG_ERROR: u8 = 3;
pub const MSG_WARNING: u8 = 4;
pub const MSG_IO_ERROR: u8 = 22;
pub const MSG_NOOP: u8 = 42;
pub const MSG_ERROR_EXIT: u8 = 86;
pub const ITEM_TRANSFER: u16 = 0x8000;
pub const ITEM_IS_NEW: u16 = 0x2000;
pub const ITEM_XNAME_FOLLOWS: u16 = 0x1000;
pub const ITEM_BASIS_TYPE_FOLLOWS: u16 = 0x0800;
const XMIT_TOP_DIR: u16 = 0x01;
const XMIT_SAME_MODE: u16 = 0x02;
const XMIT_EXTENDED_FLAGS: u16 = 0x04;
const XMIT_SAME_UID: u16 = 0x08;
const XMIT_SAME_GID: u16 = 0x10;
const XMIT_SAME_NAME: u16 = 0x20;
const XMIT_LONG_NAME: u16 = 0x40;
const XMIT_SAME_TIME: u16 = 0x80;
pub const CHUNK: usize = 32 * 1024;
pub const MAX_LINE: usize = 4096;
pub const MAX_ARGS: usize = 64;
pub const MAX_NAME: usize = 4096;
pub const MAX_ENTRIES: usize = 100_000;
pub const MAX_UPLOAD_LIST_ENTRIES: usize = 2 * super::MAX_UPLOAD_FILES;
pub const MAX_UPLOAD_LIST_NAME_BYTES: usize = 1024 * 1024;
pub const MAX_FILTER_BYTES: usize = 64 * 1024;
pub const MAX_SUM_COUNT: i32 = 1 << 20;
pub const S_IFMT: u32 = 0o170000;
pub const S_IFREG: u32 = 0o100000;
pub const S_IFDIR: u32 = 0o040000;
pub const S_IFLNK: u32 = 0o120000;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Kind {
File,
Dir,
Symlink,
}
impl Kind {
pub fn name(self) -> &'static str {
match self {
Kind::File => "file",
Kind::Dir => "dir",
Kind::Symlink => "symlink",
}
}
pub fn of_mode(mode: u32) -> Option<Self> {
match mode & S_IFMT {
S_IFREG => Some(Kind::File),
S_IFDIR => Some(Kind::Dir),
S_IFLNK => Some(Kind::Symlink),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Entry {
pub path: String,
pub kind: Kind,
pub data: Vec<u8>,
pub mode: u32,
pub mtime: u32,
pub uid: u32,
pub gid: u32,
pub size: u64,
}
impl Entry {
pub fn is_dir(&self) -> bool {
self.kind == Kind::Dir
}
}
pub fn le32(x: i32) -> [u8; 4] {
x.to_le_bytes()
}
pub fn longint(x: u64) -> Vec<u8> {
if x <= 0x7fff_ffff {
(x as i32).to_le_bytes().to_vec()
} else {
let mut v = vec![0xff; 4];
v.extend_from_slice(&(x as i64).to_le_bytes());
v
}
}
pub fn md4(parts: &[&[u8]]) -> [u8; 16] {
let mut h = Md4::new();
for p in parts {
h.update(p);
}
h.finalize().into()
}
pub fn file_sum(seed: i32, data: &[u8]) -> [u8; 16] {
md4(&[&seed.to_le_bytes(), data])
}
pub fn list_sum(data: &[u8]) -> [u8; 16] {
md4(&[data])
}
pub fn auth_response(password: &str, challenge: &str) -> String {
use base64::Engine;
let digest = md4(&[&[0u8; 4], password.as_bytes(), challenge.as_bytes()]);
base64::engine::general_purpose::STANDARD_NO_PAD.encode(digest)
}
pub fn frame(code: u8, payload: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(4 + payload.len());
let header = ((u32::from(MPLEX_BASE + code)) << 24) | payload.len() as u32;
out.extend_from_slice(&header.to_le_bytes());
out.extend_from_slice(payload);
out
}
pub fn data_frames(bytes: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(bytes.len() + 4 * (bytes.len() / 0xffff + 1));
for chunk in bytes.chunks(0xffff) {
out.extend_from_slice(&frame(MSG_DATA, chunk));
}
out
}
pub fn sort_key(path: &str, is_dir: bool) -> Vec<(u8, Vec<u8>)> {
if path == "." {
return vec![];
}
let comps: Vec<&[u8]> = path.as_bytes().split(|&b| b == b'/').collect();
let mut key = Vec::with_capacity(comps.len());
for c in &comps[..comps.len() - 1] {
key.push((1, [*c, b"/"].concat()));
}
let last = comps[comps.len() - 1];
key.push(if is_dir {
(1, [last, b"/"].concat())
} else {
(0, last.to_vec())
});
key
}
pub fn sort(entries: &mut [Entry]) {
entries.sort_by_cached_key(|e| sort_key(&e.path, e.is_dir()));
}
#[derive(Debug, Default, Clone)]
pub struct Request {
pub sender: bool,
pub recursive: bool,
pub dirs: bool,
pub links: bool,
pub owner: bool,
pub group: bool,
pub dry_run: bool,
pub checksum: bool,
pub numeric_ids: bool,
pub list_only: bool,
pub devices: bool,
pub hard_links: bool,
pub checksum_seed: Option<i32>,
pub paths: Vec<String>,
pub refusal: Option<String>,
}
pub fn parse_args(args: &[String], module: &str) -> Request {
let mut r = Request::default();
let mut paths = false;
for a in args {
if paths {
let p = if a == module {
""
} else if let Some(rest) = a.strip_prefix(&format!("{module}/")) {
rest
} else {
r.refusal
.get_or_insert(format!("path {a:?} is outside module {module}"));
continue;
};
r.paths.push(p.to_string());
continue;
}
match a.as_str() {
"." => paths = true,
"--server" => {}
"--sender" => r.sender = true,
"--numeric-ids" => r.numeric_ids = true,
"--list-only" => r.list_only = true,
"--devices" | "--specials" => r.devices = true,
s if s.starts_with("--delete") || s == "--del" => {
r.refusal
.get_or_insert(format!("{s} is not supported by this daemon"));
}
s if s.starts_with("--checksum-seed=") => {
r.checksum_seed = s["--checksum-seed=".len()..].parse().ok();
}
s if s.starts_with("--files-from") || s == "--from0" || s == "--secluded-args" => {
r.refusal
.get_or_insert(format!("{s} is not supported by this daemon"));
}
s if s.starts_with("--") => {}
s if s.starts_with('-') => {
for c in s[1..].chars() {
match c {
'e' => break, 'r' => r.recursive = true,
'd' => r.dirs = true,
'l' => r.links = true,
'o' => r.owner = true,
'g' => r.group = true,
'n' => r.dry_run = true,
'c' => r.checksum = true,
'D' => r.devices = true,
'H' => r.hard_links = true,
'z' | 'U' | 'N' | 'R' | 's' => {
r.refusal
.get_or_insert(format!("-{c} is not supported by this daemon"));
}
_ => {}
}
}
}
_ => {}
}
}
if !r.sender && r.paths.len() > 1 {
r.refusal
.get_or_insert("an upload names one destination".into());
}
if !r.sender && r.hard_links {
r.refusal
.get_or_insert("-H is not supported for uploads by this daemon".into());
}
r
}
#[derive(Default)]
pub struct FlistWriter {
last_name: Vec<u8>,
mode: Option<u32>,
mtime: Option<u32>,
uid: Option<u32>,
gid: Option<u32>,
}
impl FlistWriter {
pub fn entry(&mut self, e: &Entry, top: bool, r: &Request, out: &mut Vec<u8>) {
let name = e.path.as_bytes();
let first = self.mode.is_none();
let mut x: u16 = 0;
if e.is_dir() && top {
x |= XMIT_TOP_DIR;
}
if self.mode == Some(e.mode) {
x |= XMIT_SAME_MODE;
}
if !r.owner || (!first && self.uid == Some(e.uid)) {
x |= XMIT_SAME_UID;
}
if !r.group || (!first && self.gid == Some(e.gid)) {
x |= XMIT_SAME_GID;
}
if self.mtime == Some(e.mtime) {
x |= XMIT_SAME_TIME;
}
let l1 = self
.last_name
.iter()
.zip(name)
.take(255)
.take_while(|(a, b)| a == b)
.count();
let l2 = name.len() - l1;
if l1 > 0 {
x |= XMIT_SAME_NAME;
}
if l2 > 255 {
x |= XMIT_LONG_NAME;
}
if x == 0 && !e.is_dir() {
x |= XMIT_TOP_DIR; }
if x & 0xff00 != 0 || x == 0 {
x |= XMIT_EXTENDED_FLAGS;
out.extend_from_slice(&x.to_le_bytes());
} else {
out.push(x as u8);
}
if x & XMIT_SAME_NAME != 0 {
out.push(l1 as u8);
}
if x & XMIT_LONG_NAME != 0 {
out.extend_from_slice(&le32(l2 as i32));
} else {
out.push(l2 as u8);
}
out.extend_from_slice(&name[l1..]);
out.extend_from_slice(&longint(e.size));
if x & XMIT_SAME_TIME == 0 {
out.extend_from_slice(&e.mtime.to_le_bytes());
}
if x & XMIT_SAME_MODE == 0 {
out.extend_from_slice(&e.mode.to_le_bytes());
}
if r.owner && x & XMIT_SAME_UID == 0 {
out.extend_from_slice(&e.uid.to_le_bytes());
}
if r.group && x & XMIT_SAME_GID == 0 {
out.extend_from_slice(&e.gid.to_le_bytes());
}
if r.links && e.kind == Kind::Symlink {
out.extend_from_slice(&le32(e.data.len() as i32));
out.extend_from_slice(&e.data);
}
if r.checksum && e.kind == Kind::File {
out.extend_from_slice(&list_sum(&e.data));
}
self.last_name = name.to_vec();
self.mode = Some(e.mode);
self.mtime = Some(e.mtime);
self.uid = Some(e.uid);
self.gid = Some(e.gid);
}
pub fn finish(r: &Request, io_error: i32, out: &mut Vec<u8>) {
out.push(0);
if r.owner && !r.numeric_ids {
out.extend_from_slice(&le32(0));
}
if r.group && !r.numeric_ids {
out.extend_from_slice(&le32(0));
}
out.extend_from_slice(&le32(io_error));
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FlistLimits {
pub max_entries: usize,
pub max_name_bytes: usize,
}
impl FlistLimits {
pub const DOWNLOAD: Self = Self {
max_entries: MAX_ENTRIES,
max_name_bytes: usize::MAX,
};
pub const UPLOAD: Self = Self {
max_entries: MAX_UPLOAD_LIST_ENTRIES,
max_name_bytes: MAX_UPLOAD_LIST_NAME_BYTES,
};
}
pub struct MuxReader<R> {
inner: R,
left: usize,
mux: bool,
pub consumed: u64,
pub messages: Vec<(u8, String)>,
pub io_error: i32,
}
impl<R: AsyncRead + Unpin> MuxReader<R> {
pub fn new(inner: R) -> Self {
Self {
inner,
left: 0,
mux: true,
consumed: 0,
messages: vec![],
io_error: 0,
}
}
pub fn raw(inner: R) -> Self {
Self {
mux: false,
..Self::new(inner)
}
}
pub async fn read(&mut self, n: usize) -> Result<Vec<u8>> {
let mut out = vec![0u8; n];
if !self.mux {
self.inner
.read_exact(&mut out)
.await
.context("the peer closed the connection")?;
self.consumed += n as u64;
return Ok(out);
}
let mut have = 0;
while have < n {
while self.left == 0 {
let mut h = [0u8; 4];
self.inner
.read_exact(&mut h)
.await
.context("the daemon closed the connection")?;
self.consumed += 4;
let h = u32::from_le_bytes(h);
let tag = ((h >> 24) as u8).wrapping_sub(MPLEX_BASE);
let len = (h & 0x00ff_ffff) as usize;
if tag == MSG_DATA {
self.left = len;
continue;
}
ensure!(len <= 8192, "a {len}-byte rsync message is too long");
let mut body = vec![0u8; len];
self.inner.read_exact(&mut body).await?;
self.consumed += len as u64;
match tag {
MSG_ERROR_XFER | MSG_INFO | MSG_ERROR | MSG_WARNING => self
.messages
.push((tag, String::from_utf8_lossy(&body).trim_end().to_string())),
MSG_IO_ERROR if len == 4 => {
self.io_error |= i32::from_le_bytes([body[0], body[1], body[2], body[3]])
}
MSG_NOOP => {}
MSG_ERROR_EXIT => bail!("the daemon exited with an error"),
other => bail!("unexpected rsync message tag {other}"),
}
}
let k = (n - have).min(self.left);
self.inner.read_exact(&mut out[have..have + k]).await?;
self.consumed += k as u64;
have += k;
self.left -= k;
}
Ok(out)
}
pub async fn byte(&mut self) -> Result<u8> {
Ok(self.read(1).await?[0])
}
pub async fn int(&mut self) -> Result<i32> {
let b = self.read(4).await?;
Ok(i32::from_le_bytes([b[0], b[1], b[2], b[3]]))
}
pub async fn short(&mut self) -> Result<u16> {
let b = self.read(2).await?;
Ok(u16::from_le_bytes([b[0], b[1]]))
}
pub async fn longint(&mut self) -> Result<u64> {
let v = self.int().await?;
if v != -1 {
return Ok(v as u32 as u64);
}
let b = self.read(8).await?;
Ok(i64::from_le_bytes(b.try_into().unwrap_or([0; 8])) as u64)
}
pub async fn file_list(&mut self, r: &Request) -> Result<(Vec<Entry>, i32)> {
self.file_list_bounded(r, FlistLimits::DOWNLOAD).await
}
pub async fn file_list_bounded(
&mut self,
r: &Request,
limits: FlistLimits,
) -> Result<(Vec<Entry>, i32)> {
let mut entries = Vec::new();
let (mut last, mut mode, mut mtime, mut uid, mut gid) =
(Vec::new(), 0u32, 0u32, 0u32, 0u32);
let mut listed = 0usize;
let mut name_bytes = 0usize;
loop {
let mut x = u16::from(self.byte().await?);
if x == 0 {
break;
}
listed += 1;
ensure!(
listed <= limits.max_entries,
"more than {} file-list entries",
limits.max_entries
);
if x & XMIT_EXTENDED_FLAGS != 0 {
x |= u16::from(self.byte().await?) << 8;
}
let l1 = if x & XMIT_SAME_NAME != 0 {
self.byte().await? as usize
} else {
0
};
let l2 = if x & XMIT_LONG_NAME != 0 {
self.int().await?.max(0) as usize
} else {
self.byte().await? as usize
};
ensure!(
l1 <= last.len() && l1 + l2 <= MAX_NAME,
"a file-list name is malformed"
);
name_bytes += l1 + l2;
ensure!(
name_bytes <= limits.max_name_bytes,
"more than {} bytes of file-list names",
limits.max_name_bytes
);
let mut name = last[..l1].to_vec();
name.extend(self.read(l2).await?);
last = name.clone();
let size = self.longint().await?;
if x & XMIT_SAME_TIME == 0 {
mtime = self.int().await? as u32;
}
if x & XMIT_SAME_MODE == 0 {
mode = self.int().await? as u32;
}
if r.owner && x & XMIT_SAME_UID == 0 {
uid = self.int().await? as u32;
}
if r.group && x & XMIT_SAME_GID == 0 {
gid = self.int().await? as u32;
}
let kind = Kind::of_mode(mode);
let mut data = Vec::new();
if r.links && kind == Some(Kind::Symlink) {
let n = self.int().await?;
ensure!(
(0..=MAX_NAME as i32).contains(&n),
"a symlink target is too long"
);
name_bytes += n as usize;
ensure!(
name_bytes <= limits.max_name_bytes,
"more than {} bytes of file-list names",
limits.max_name_bytes
);
data = self.read(n as usize).await?;
}
ensure!(
kind.is_some() || !r.devices,
"a device or special file in the list; -D is not supported for uploads"
);
if r.checksum && kind == Some(Kind::File) {
self.read(16).await?;
}
let path = String::from_utf8_lossy(&name).to_string();
ensure!(
!path.starts_with('/') && !path.split('/').any(|c| c == ".."),
"unsafe file-list name {path:?}"
);
if let Some(kind) = kind {
entries.push(Entry {
path,
kind,
data,
mode,
mtime,
uid,
gid,
size,
});
}
}
for wanted in [r.owner, r.group] {
if wanted && !r.numeric_ids {
loop {
let id = self.int().await?;
if id == 0 {
break;
}
let n = self.byte().await?;
self.read(n as usize).await?;
}
}
}
let io_error = self.int().await?;
Ok((entries, io_error))
}
}
pub fn entry_json(e: &Entry) -> Value {
let mut v = json!({
"path": e.path,
"type": e.kind.name(),
"size": e.size,
"mode": format!("{:o}", e.mode & 0o7777),
"mtime": e.mtime,
});
if e.kind == Kind::Symlink && !e.data.is_empty() {
v["target"] = json!(String::from_utf8_lossy(&e.data));
}
v
}
pub fn content_json(data: &[u8]) -> (String, &'static str) {
match std::str::from_utf8(data) {
Ok(s)
if s.chars()
.all(|c| !c.is_control() || matches!(c, '\n' | '\r' | '\t')) =>
{
(s.to_string(), "utf8")
}
_ => (hex::encode(data), "hex"),
}
}
pub fn entry_from_json(v: &Value, default_mtime: u32) -> Result<Entry> {
let path = v["path"]
.as_str()
.context("each entry needs a path")?
.trim_matches('/')
.to_string();
ensure!(
!path.is_empty() && path.len() <= MAX_NAME,
"an entry path is empty or too long"
);
ensure!(
!path
.split('/')
.any(|c| c.is_empty() || c == "." || c == ".."),
"entry path {path:?} must be relative, without . or .. components"
);
let kind = match v["type"].as_str().unwrap_or("file") {
"file" => Kind::File,
"dir" | "directory" => Kind::Dir,
"symlink" | "link" => Kind::Symlink,
other => bail!("entry type {other:?} is not file, dir or symlink"),
};
let data = match kind {
Kind::File => {
let content = v["content"].as_str().unwrap_or_default();
match v["encoding"].as_str().unwrap_or("utf8") {
"utf8" => content.as_bytes().to_vec(),
"hex" => hex::decode(content).context("content is not hex")?,
other => bail!("encoding {other:?} is not utf8 or hex"),
}
}
Kind::Symlink => v["target"]
.as_str()
.context("a symlink needs a target")?
.as_bytes()
.to_vec(),
Kind::Dir => vec![],
};
let type_bits = match kind {
Kind::File => S_IFREG,
Kind::Dir => S_IFDIR,
Kind::Symlink => S_IFLNK,
};
let perms = match v.get("mode") {
None | Some(Value::Null) => match kind {
Kind::File => 0o644,
Kind::Dir => 0o755,
Kind::Symlink => 0o777,
},
Some(Value::String(s)) => u32::from_str_radix(s.trim_start_matches("0o"), 8)
.with_context(|| format!("mode {s:?} is not octal"))?,
Some(n) => n
.as_u64()
.context("mode is an octal string such as \"644\"")? as u32,
};
ensure!(
perms <= 0o7777,
"mode {perms:o} has bits beyond permissions"
);
let mtime = v["mtime"]
.as_u64()
.map(|t| t as u32)
.unwrap_or(default_mtime);
let size = match kind {
Kind::Dir => 4096,
_ => data.len() as u64,
};
Ok(Entry {
path,
kind,
data,
mode: type_bits | perms,
mtime,
uid: v["uid"].as_u64().unwrap_or(0) as u32,
gid: v["gid"].as_u64().unwrap_or(0) as u32,
size,
})
}