1use crate::llm::actions::{
4 client_trait::{Client, ClientActionResult},
5 protocol_trait::Protocol,
6 ActionDefinition, ParameterDefinition, StartupExamples,
7};
8use crate::protocol::{ConnectContext, EventType};
9use crate::server::rsync::actions::{action, p};
10use crate::state::app_state::AppState;
11use anyhow::{bail, ensure, Context, Result};
12use serde_json::{json, Value};
13use std::sync::LazyLock;
14
15pub const DEFAULT_MAX_FETCH_BYTES: u64 = 4 * 1024 * 1024;
17pub const MAX_FETCH_BYTES_LIMIT: u64 = 64 * 1024 * 1024;
18pub const MAX_PUSH_FILES: usize = 1000;
20pub const MAX_PUSH_BYTES: usize = 8 * 1024 * 1024;
21
22#[derive(Default)]
23pub struct RsyncClientProtocol;
24impl RsyncClientProtocol {
25 pub fn new() -> Self {
26 Self
27 }
28}
29
30fn path_param() -> crate::llm::actions::Parameter {
31 p(
32 "path",
33 "string",
34 "Module and path, e.g. pub/ (the module's contents), pub/docs/ or pub/hello.txt",
35 true,
36 )
37}
38
39fn list_modules_action() -> ActionDefinition {
40 action(
41 "rsync_list_modules",
42 "Ask the daemon which modules it offers; the answer arrives as rsync_modules.",
43 vec![],
44 json!({"type":"rsync_list_modules"}),
45 )
46}
47
48fn list_action() -> ActionDefinition {
49 action(
50 "rsync_list",
51 "List a path (like rsync --list-only); the answer arrives as rsync_listing.",
52 vec![
53 path_param(),
54 p(
55 "recursive",
56 "boolean",
57 "List the whole subtree instead of one level",
58 false,
59 ),
60 ],
61 json!({"type":"rsync_list","path":"pub/","recursive":false}),
62 )
63}
64
65fn fetch_action() -> ActionDefinition {
66 action(
67 "rsync_fetch",
68 "Download a file, or a directory's files, and see their contents in rsync_fetched.",
69 vec![
70 path_param(),
71 p(
72 "recursive",
73 "boolean",
74 "Fetch a directory's whole subtree (default: just the named file or one level)",
75 false,
76 ),
77 ],
78 json!({"type":"rsync_fetch","path":"pub/hello.txt"}),
79 )
80}
81
82fn push_action() -> ActionDefinition {
83 action(
84 "rsync_push",
85 "Upload files whose contents you supply to a module path (like rsync -rl files rsync://host/module/path/). Directories they imply are created. The outcome arrives as rsync_pushed.",
86 vec![
87 p("path", "string", "Destination: module and directory, e.g. inbox/ or inbox/reports/", true),
88 p(
89 "files",
90 "array",
91 "Each {path (relative to the destination, e.g. a.txt or docs/b.md), type (file, dir or symlink; default file), content, encoding (utf8 or hex), target (symlinks), mode (octal string), mtime (unix seconds)}",
92 true,
93 ),
94 ],
95 json!({"type":"rsync_push","path":"inbox/","files":[{"path":"report.txt","content":"all good\n"}]}),
96 )
97}
98
99fn disconnect_action() -> ActionDefinition {
100 action(
101 "disconnect",
102 "Stop the client (no rsync connection stays open between operations).",
103 vec![],
104 json!({"type":"disconnect"}),
105 )
106}
107
108pub fn actions() -> Vec<ActionDefinition> {
109 vec![
110 list_modules_action(),
111 list_action(),
112 fetch_action(),
113 push_action(),
114 disconnect_action(),
115 ]
116}
117
118fn ev(id: &str, description: &str, params: Vec<crate::llm::actions::Parameter>) -> EventType {
119 EventType::new(id, description, list_action().example.clone())
120 .with_parameters(params)
121 .with_actions(actions())
122}
123
124fn entries_param() -> crate::llm::actions::Parameter {
125 p(
126 "entries",
127 "array",
128 "Each {path, type (file, dir, symlink), size, mode (octal), mtime, target (symlinks)}",
129 true,
130 )
131}
132
133pub static READY_EVENT: LazyLock<EventType> = LazyLock::new(|| {
134 ev(
135 "rsync_ready",
136 "The client is ready; every operation is one connection to the daemon.",
137 vec![p("daemon", "string", "The daemon's address and port", true)],
138 )
139});
140
141pub static MODULES_EVENT: LazyLock<EventType> = LazyLock::new(|| {
142 ev(
143 "rsync_modules",
144 "The daemon's module list.",
145 vec![
146 p("modules", "array", "Each {name, comment}", true),
147 p(
148 "motd",
149 "string",
150 "The daemon's message of the day, if any",
151 false,
152 ),
153 ],
154 )
155});
156
157pub static LISTING_EVENT: LazyLock<EventType> = LazyLock::new(|| {
158 ev(
159 "rsync_listing",
160 "A path's listing, in rsync's order.",
161 vec![
162 p("path", "string", "The path that was listed", true),
163 entries_param(),
164 ],
165 )
166});
167
168pub static FETCHED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
169 ev(
170 "rsync_fetched",
171 "Files downloaded from the daemon, each checked against rsync's whole-file MD4.",
172 vec![
173 p("path", "string", "The path that was fetched", true),
174 p(
175 "files",
176 "array",
177 "Each {path, size, content, encoding (utf8 or hex)}",
178 true,
179 ),
180 entries_param(),
181 p(
182 "skipped",
183 "array",
184 "Files not fetched because max_fetch_bytes was reached",
185 false,
186 ),
187 ],
188 )
189});
190
191pub static PUSHED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
192 ev(
193 "rsync_pushed",
194 "A push finished: what the daemon asked for and was sent (whole, with rsync's MD4), and what it said about the rest.",
195 vec![
196 p("path", "string", "The destination", true),
197 p("sent", "array", "Each file the daemon took, {path, size}", true),
198 p("not_requested", "array", "Files in the list the daemon never asked for", true),
199 p(
200 "errors",
201 "array",
202 "What the daemon reported (a refused file, a read-only module); empty when every file was taken",
203 true,
204 ),
205 p("complete", "boolean", "Every file was sent and the daemon reported nothing", true),
206 ],
207 )
208});
209
210pub static ERROR_EVENT: LazyLock<EventType> = LazyLock::new(|| {
211 ev(
212 "rsync_error",
213 "An operation failed: the daemon refused it, or a path did not exist.",
214 vec![
215 p(
216 "operation",
217 "string",
218 "rsync_list_modules, rsync_list, rsync_fetch or rsync_push",
219 true,
220 ),
221 p("path", "string", "The path it was about, if any", false),
222 p(
223 "message",
224 "string",
225 "What the daemon (or the client) said",
226 true,
227 ),
228 ],
229 )
230});
231
232pub fn check(v: &Value) -> Result<Option<(String, String)>> {
234 match v["type"].as_str().unwrap_or_default() {
235 "rsync_list_modules" => Ok(None),
236 "rsync_push" => {
237 let files = v["files"].as_array().context("files is an array")?;
238 ensure!(
239 !files.is_empty() && files.len() <= MAX_PUSH_FILES,
240 "a push carries 1 to {MAX_PUSH_FILES} files"
241 );
242 let mut bytes = 0usize;
243 for f in files {
244 bytes += crate::server::rsync::wire::entry_from_json(f, 0)?
245 .data
246 .len();
247 }
248 ensure!(
249 bytes <= MAX_PUSH_BYTES,
250 "a push carries at most {MAX_PUSH_BYTES} bytes"
251 );
252 let mut target = check(&json!({"type": "rsync_list", "path": v["path"]}))?
253 .context("a push needs a destination")?;
254 if !target.1.ends_with('/') {
255 target.1.push('/');
256 }
257 Ok(Some(target))
258 }
259 "rsync_list" | "rsync_fetch" => {
260 let path = v["path"]
261 .as_str()
262 .context("path is required")?
263 .trim_start_matches('/');
264 let module = path.split('/').next().unwrap_or_default();
265 ensure!(
266 !module.is_empty(),
267 "path starts with a module name, e.g. pub/"
268 );
269 ensure!(
270 !path.split('/').any(|c| c == ".."),
271 "path must not contain .."
272 );
273 ensure!(
274 !path.contains(['\n', '\r', '\0']) && path.len() <= 1024,
275 "path is one line of at most 1024 bytes"
276 );
277 let arg = if path.contains('/') {
278 path.to_string()
279 } else {
280 format!("{path}/")
281 };
282 Ok(Some((module.to_string(), arg)))
283 }
284 other => bail!("Unknown rsync client action {other:?}"),
285 }
286}
287
288impl Protocol for RsyncClientProtocol {
289 fn protocol_name(&self) -> &'static str {
290 "rsync"
291 }
292 fn stack_name(&self) -> &'static str {
293 "ETH>IP>TCP>rsync"
294 }
295 fn keywords(&self) -> Vec<&'static str> {
296 vec!["rsync", "rsync client", "rsync://", "mirror"]
297 }
298 fn description(&self) -> &'static str {
299 "rsync client for rsync:// daemons: lists modules and paths and downloads files (protocol 29)"
300 }
301 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
302 actions()
303 }
304 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
305 vec![]
306 }
307 fn get_event_types(&self) -> Vec<EventType> {
308 vec![
309 READY_EVENT.clone(),
310 MODULES_EVENT.clone(),
311 LISTING_EVENT.clone(),
312 FETCHED_EVENT.clone(),
313 ERROR_EVENT.clone(),
314 ]
315 }
316 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
317 vec![
318 ParameterDefinition {
319 name: "username".into(),
320 type_hint: "string".into(),
321 description:
322 "User for a module that asks for authentication (auth users in rsyncd.conf)"
323 .into(),
324 required: false,
325 example: json!("alice"),
326 default: None,
327 },
328 ParameterDefinition {
329 name: "password".into(),
330 type_hint: "string".into(),
331 description:
332 "Password for that user (the daemon's secrets file); never sent in clear".into(),
333 required: false,
334 example: json!("s3cret"),
335 default: None,
336 },
337 ParameterDefinition {
338 name: "max_fetch_bytes".into(),
339 type_hint: "number".into(),
340 description:
341 "Most file bytes one fetch brings back; files beyond it are listed as skipped"
342 .into(),
343 required: false,
344 example: json!(1048576),
345 default: Some(json!(DEFAULT_MAX_FETCH_BYTES)),
346 },
347 ]
348 }
349 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
350 use crate::protocol::metadata::*;
351 ProtocolMetadataV2::builder()
352 .state(DevelopmentState::Experimental)
353 .privilege_requirement(PrivilegeRequirement::None)
354 .implementation("The server's protocol-29 codec (src/server/rsync/wire.rs) as the receiver: @RSYNCD handshake with MOTD and MD4 challenge-response authentication, -ld/-lr requests, demultiplexed input, the file list sorted into rsync's order, whole-file requests written concurrently with the replies they produce, every file checked against MD4(seed ‖ data), and the end-of-run exchange")
355 .llm_control("Which modules and paths to list and which files to fetch, and what to do with what came back")
356 .e2e_testing("tests/client/rsync: a stock rsync 3.2.7 daemon (rsync --daemon) — module listing with MOTD, a recursive listing, a fetch whose contents the model acts on, a password-protected module, an unknown module, and rsync_push of a tree (text, hex, a multi-token file, a symlink, an empty directory) read back from the daemon's own directory, plus a push refused by a read-only module")
357 .notes("Files come back into the event, bounded by max_fetch_bytes; nothing is written to disk. rsync_push sends files whose contents the model supplies (whole, at most 1000 files and 8 MiB) and never deletes anything on the daemon. Devices and special files in a listing are skipped.")
358 .max_inbound_bytes(MAX_FETCH_BYTES_LIMIT as usize)
359 .build()
360 }
361 fn example_prompt(&self) -> &'static str {
362 "List the modules on rsync://mirror.example and fetch the README from the first one"
363 }
364 fn get_startup_examples(&self) -> StartupExamples {
365 let llm = json!({"type":"open_client","protocol":"rsync","remote_addr":"127.0.0.1:873",
366 "instruction":"List the modules, then fetch pub/hello.txt and summarise it"});
367 let mut static_example = llm.clone();
368 static_example["event_handlers"] = json!([
369 {"event_pattern":"rsync_ready","handler":{"type":"static","actions":[list_modules_action().example]}},
370 {"event_pattern":"*","handler":{"type":"static","actions":[]}}
371 ]);
372 let mut scripted = static_example.clone();
373 scripted["event_handlers"][1] = json!({"event_pattern":"rsync_modules","handler":{"type":"script","language":"python",
374 "code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'rsync_list','path':m['name']+'/'} for m in e['modules'][:1]]}))"}});
375 StartupExamples::new(llm, scripted, static_example)
376 }
377 fn group_name(&self) -> &'static str {
378 "File Transfer"
379 }
380}
381
382impl Client for RsyncClientProtocol {
383 fn connect(
384 &self,
385 ctx: ConnectContext,
386 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
387 {
388 Box::pin(super::connect(ctx))
389 }
390 fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
391 if v["type"] == "disconnect" {
392 return Ok(ClientActionResult::Disconnect);
393 }
394 check(&v)?;
395 Ok(ClientActionResult::Custom {
396 name: v["type"].as_str().unwrap_or_default().to_string(),
397 data: v,
398 })
399 }
400}