netget/server/rsync/

actions.rs

//! What the model decides for an rsync daemon: which modules exist, what a module holds at the
//! path a client asked for, and which uploaded files to take. The model supplies every byte it
//! serves and is handed every byte it takes; nothing is read from or written to disk.
use crate::llm::actions::{
    protocol_trait::{ActionResult, Protocol, Server},
    ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
};
use crate::protocol::{log_template::LogTemplate, EventType, SpawnContext};
use crate::state::app_state::AppState;
use anyhow::{bail, ensure, Context, Result};
use serde_json::{json, Value};
use std::sync::LazyLock;

pub const MODULES: &str = "rsync_modules";
pub const ENTRIES: &str = "rsync_send_entries";
pub const REFUSE: &str = "rsync_refuse";
pub const ACCEPT_UPLOAD: &str = "rsync_accept_upload";
/// Modules one listing may name.
pub const MAX_MODULES: usize = 256;

#[derive(Default, Clone)]
pub struct RsyncProtocol;

impl RsyncProtocol {
    pub fn new() -> Self {
        Self
    }
}

pub fn p(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
    Parameter {
        name: name.into(),
        type_hint: type_hint.into(),
        description: description.into(),
        required,
    }
}

pub fn action(
    name: &str,
    description: &str,
    parameters: Vec<Parameter>,
    example: Value,
) -> ActionDefinition {
    let info = match name {
        MODULES => "-> rsync module list".to_string(),
        ENTRIES => "-> rsync file list".to_string(),
        REFUSE => "-> rsync refusal: {message}".to_string(),
        ACCEPT_UPLOAD => "-> rsync upload accepted".to_string(),
        other => format!("-> rsync {}", other.trim_start_matches("rsync_")),
    };
    ActionDefinition {
        name: name.into(),
        description: description.into(),
        parameters,
        example,
        log_template: Some(LogTemplate::new().with_info(info)),
    }
}

pub fn modules_action() -> ActionDefinition {
    action(
        MODULES,
        "Answer the module listing (what `rsync rsync://host/` prints): each module's name and comment.",
        vec![p("modules", "array", "Modules, each {name, comment}; name is one word", true)],
        json!({"type":MODULES,"modules":[{"name":"pub","comment":"public files"}]}),
    )
}

pub fn entries_action() -> ActionDefinition {
    action(
        ENTRIES,
        "Answer with what the module holds at and under the requested path. The daemon sends the right part of it, in rsync's order.",
        vec![p(
            "entries",
            "array",
            "Each {path (relative to the module root, e.g. docs/a.txt), type (file, dir or symlink), content (files), encoding (utf8 or hex), target (symlinks), mode (octal string, e.g. \"644\"), mtime (unix seconds)}",
            true,
        )],
        json!({"type":ENTRIES,"entries":[
            {"path":"hello.txt","type":"file","content":"hello\n"},
            {"path":"docs","type":"dir"},
            {"path":"docs/readme.md","type":"file","content":"# Readme\n","mode":"644"}]}),
    )
}

pub fn refuse_action() -> ActionDefinition {
    action(
        REFUSE,
        "Refuse the request; the client prints the message and exits with an error.",
        vec![p(
            "message",
            "string",
            "Why, e.g. Unknown module 'x' or permission denied",
            true,
        )],
        json!({"type":REFUSE,"message":"Unknown module 'secret'"}),
    )
}

pub fn accept_upload_action() -> ActionDefinition {
    action(
        ACCEPT_UPLOAD,
        "Accept an upload. Give `paths` to take only those files (the rest are refused, each with an error the client prints); leave it out to take every file. The files' contents arrive afterwards in rsync_upload_complete. Nothing is written to disk.",
        vec![p(
            "paths",
            "array",
            "Optional: the files to take, as module-relative paths from the event's files list",
            false,
        )],
        json!({"type":ACCEPT_UPLOAD,"paths":["incoming/report.txt"]}),
    )
}

pub static UPLOAD_REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
    EventType::new(
        "rsync_upload_request",
        "A client is uploading (rsync files rsync://host/module/path/). Its file list is here, without contents. Answer rsync_accept_upload to take all or some of the files, or rsync_refuse.",
        accept_upload_action().example.clone(),
    )
    .with_parameters(vec![
        p("module", "string", "The module named in the request", true),
        p("path", "string", "The destination inside the module (empty for its root)", true),
        p("files", "array", "Each {path (module-relative), type (file, dir or symlink), size, mode (octal string), target (symlinks)}", true),
        p("file_count", "number", "Regular files in the list", true),
        p("total_bytes", "number", "Their sizes added up", true),
        p("client", "string", "The client's address and port", true),
    ])
    .with_actions(vec![accept_upload_action(), refuse_action()])
});

pub static UPLOAD_COMPLETE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
    EventType::new(
        "rsync_upload_complete",
        "An upload finished: the files that were taken, with their contents (each checked against the client's MD4), and the ones that were not. Nothing was written to disk; record what matters in memory if it should be served later.",
        json!({"type":"set_memory","value":"received incoming/report.txt"}),
    )
    .with_parameters(vec![
        p("module", "string", "The module uploaded to", true),
        p("path", "string", "The destination inside the module", true),
        p("files", "array", "Each received file {path, size, content, encoding (utf8 or hex)}; directories and symlinks the list named are included without content", true),
        p("refused", "array", "Files not taken, each {path, reason}", true),
        p("client", "string", "The client's address and port", true),
    ])
    .with_no_actions()
});

pub static LIST_MODULES_EVENT: LazyLock<EventType> = LazyLock::new(|| {
    EventType::new(
        "rsync_list_modules",
        "A client asked which modules this daemon offers (rsync rsync://host/). Answer rsync_modules.",
        modules_action().example.clone(),
    )
    .with_parameters(vec![p("client", "string", "The client's address and port", true)])
    .with_actions(vec![modules_action()])
});

pub static REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
    EventType::new(
        "rsync_request",
        "A client asked for a path in a module, to list it or download it. Answer rsync_send_entries with what is there, or rsync_refuse.",
        entries_action().example.clone(),
    )
    .with_parameters(vec![
        p("module", "string", "The module named in the request", true),
        p("path", "string", "The path inside the module (empty for its root; a trailing / means the directory's contents)", true),
        p("recursive", "boolean", "Whether the whole subtree is wanted (-r)", true),
        p("list_only", "boolean", "Whether the client said --list-only (it also lists when it names no destination)", true),
        p("client", "string", "The client's address and port", true),
    ])
    .with_actions(vec![entries_action(), refuse_action()])
});

/// Check an action's shape (the daemon applies it to the request it answers).
pub fn check(v: &Value) -> Result<()> {
    match v["type"].as_str().unwrap_or_default() {
        MODULES => {
            let list = v["modules"].as_array().context("modules is an array")?;
            ensure!(list.len() <= MAX_MODULES, "more than {MAX_MODULES} modules");
            for m in list {
                let name = m["name"].as_str().context("each module needs a name")?;
                ensure!(
                    !name.is_empty()
                        && name.len() <= 64
                        && !name.contains(['/', ' ', '\t', '\n', '\r']),
                    "module name {name:?} must be one word without /"
                );
            }
        }
        ENTRIES => {
            let list = v["entries"].as_array().context("entries is an array")?;
            ensure!(list.len() <= super::wire::MAX_ENTRIES, "too many entries");
            for e in list {
                super::wire::entry_from_json(e, 0)?;
            }
        }
        REFUSE => {
            v["message"]
                .as_str()
                .context("rsync_refuse needs a message")?;
        }
        ACCEPT_UPLOAD => {
            if let Some(paths) = v.get("paths") {
                let list = paths.as_array().context("paths is an array")?;
                ensure!(list.len() <= super::wire::MAX_ENTRIES, "too many paths");
                ensure!(list.iter().all(|p| p.is_string()), "every path is a string");
            }
        }
        other => bail!("Unknown rsync action {other:?}"),
    }
    Ok(())
}

impl Protocol for RsyncProtocol {
    fn protocol_name(&self) -> &'static str {
        "rsync"
    }
    fn stack_name(&self) -> &'static str {
        "ETH>IP>TCP>rsync"
    }
    fn keywords(&self) -> Vec<&'static str> {
        vec![
            "rsync",
            "rsyncd",
            "rsync daemon",
            "rsync://",
            "873",
            "file sync",
        ]
    }
    fn description(&self) -> &'static str {
        "rsync daemon (rsync://, protocol 29): modules whose listings and file contents the model supplies, for stock rsync clients to list and download, and uploads whose files the model chooses to take"
    }
    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
        Vec::new()
    }
    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
        vec![
            modules_action(),
            entries_action(),
            refuse_action(),
            accept_upload_action(),
        ]
    }
    fn get_event_types(&self) -> Vec<EventType> {
        vec![
            LIST_MODULES_EVENT.clone(),
            REQUEST_EVENT.clone(),
            UPLOAD_REQUEST_EVENT.clone(),
            UPLOAD_COMPLETE_EVENT.clone(),
        ]
    }
    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
        vec![
            ParameterDefinition {
                name: "motd".into(),
                type_hint: "string".into(),
                description: "Message of the day, sent after the greeting (rsync prints it)".into(),
                required: false,
                example: json!("Welcome to the NetGet mirror"),
                default: None,
            },
            ParameterDefinition {
                name: "idle_timeout_secs".into(),
                type_hint: "number".into(),
                description: "Seconds to wait for the client's next line or request".into(),
                required: false,
                example: json!(60),
                default: Some(json!(super::IDLE_TIMEOUT.as_secs())),
            },
        ]
    }
    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
        use crate::protocol::metadata::*;
        ProtocolMetadataV2::builder()
            .state(DevelopmentState::Experimental)
            .privilege_requirement(PrivilegeRequirement::PrivilegedPort(873))
            .well_known_port(873)
            .answers_on_failure()
            .request_only("rsync carries one request per connection on a strictly sequenced binary stream; a message injected into it would desynchronise the client")
            .implementation("Hand-rolled rsync daemon protocol pinned at 29 (src/server/rsync/wire.rs): @RSYNCD handshake and MOTD, module listing, newline-terminated arguments, the raw checksum seed, multiplexed output, the protocol-29 file list in rsync's f_name_cmp order, whole-file transfers as literal tokens with MD4(seed ‖ data), dry runs, -c list checksums, and the three-NDX_DONE end of run with stats")
            .llm_control("Which modules exist, and the files, directories and symlinks a module holds (with their contents) at each requested path; refusals")
            .e2e_testing("tests/server/rsync: the stock rsync 3.2.7 client — module listing, --list-only, -a and -r downloads compared byte for byte, a dry run, a symlink, a refusal, and an upload refused as read-only; upload_test.rs: pushes of a tree and of one file with -a, the model taking part of an upload with the rest refused per file, contents compared byte for byte, and uploads refused (the model, --delete, -H, the per-file size bound); upload_bounds_test.rs: a hand-written protocol-29 client sending more than a file's listed size, more file data than an upload may carry, an oversized file list, a refused upload, and a multi-byte path across the message cut")
            .notes("Uploads are offered to the model, which takes all or some files; their contents reach it in rsync_upload_complete and nothing is written to disk. An upload names one destination; --delete and -H are refused for it. Bounds: 1 MiB per uploaded file and no more than its listed size (a file sending more is refused), 8 MiB of declared sizes and of file data actually received and 1000 files per upload, and an upload's file list at most 2000 entries and 1 MiB of names and symlink targets. A refused upload raises no rsync_upload_complete. -z, -R, -U, -N, -s and --files-from are refused rather than half-supported. No authentication (every module is open). Block matching against an existing destination file is not done: every file is sent whole, which is correct, just not minimal.")
            .max_inbound_bytes(super::MAX_INBOUND_BYTES)
            .build()
    }
    fn example_prompt(&self) -> &'static str {
        "Run an rsync daemon with a module 'pub' holding a README and a docs directory"
    }
    fn get_startup_examples(&self) -> StartupExamples {
        let base = json!({"type":"open_server","base_stack":"rsync","port":0,
            "instruction":"Offer one module, pub, holding hello.txt (hello) and docs/readme.md"});
        let mut static_example = base.clone();
        static_example["event_handlers"] = json!([
            {"event_pattern":"rsync_list_modules","handler":{"type":"static","actions":[modules_action().example]}},
            {"event_pattern":"rsync_request","handler":{"type":"static","actions":[entries_action().example]}}
        ]);
        let mut scripted = base.clone();
        scripted["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"script","language":"python",
            "code":"import json,sys\ni=json.load(sys.stdin); t=i['event_type_id']; e=i['event']\nif t=='rsync_list_modules': a=[{'type':'rsync_modules','modules':[{'name':'pub','comment':'public'}]}]\nelif e['module']!='pub': a=[{'type':'rsync_refuse','message':\"Unknown module '%s'\" % e['module']}]\nelse: a=[{'type':'rsync_send_entries','entries':[{'path':'hello.txt','type':'file','content':'hello\\n'}]}]\nprint(json.dumps({'actions':a}))"}}]);
        StartupExamples::new(base, scripted, static_example)
    }
    fn group_name(&self) -> &'static str {
        "File Transfer"
    }
}

impl Server for RsyncProtocol {
    fn spawn(
        &self,
        ctx: SpawnContext,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
    {
        Box::pin(super::spawn(ctx))
    }
    fn execute_action(&self, action: Value) -> Result<ActionResult> {
        check(&action)?;
        // The connection applies it to the request it answers.
        Ok(ActionResult::Custom {
            name: action["type"].as_str().unwrap_or_default().to_string(),
            data: action,
        })
    }
}