1use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
2use tokio::net::TcpStream;
3use tokio::time::{Duration, interval, timeout};
4use tracing::{debug, info, warn};
5
6#[derive(Debug, Clone)]
8pub struct FtpOptions {
9 pub connect_timeout: Duration,
10 pub read_timeout: Duration,
11 pub passive_mode: bool,
12 pub username: String,
13 pub password: String,
14 pub keepalive_interval: Option<Duration>,
16 pub max_retries: u32,
18}
19
20impl Default for FtpOptions {
21 fn default() -> Self {
22 Self {
23 connect_timeout: Duration::from_secs(30),
24 read_timeout: Duration::from_secs(30),
25 passive_mode: true,
26 username: "anonymous".to_string(),
27 password: "aria2@".to_string(),
28 keepalive_interval: Some(Duration::from_secs(60)),
29 max_retries: 3,
30 }
31 }
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub enum FtpResponseClass {
37 PositivePreliminary,
39 PositiveCompletion,
41 PositiveIntermediate,
43 TransientNegative,
45 PermanentNegative,
47 Unknown,
49}
50
51impl FtpResponseClass {
52 pub fn from_code(code: u16) -> Self {
54 match code {
55 100..=199 => FtpResponseClass::PositivePreliminary,
56 200..=299 => FtpResponseClass::PositiveCompletion,
57 300..=399 => FtpResponseClass::PositiveIntermediate,
58 400..=499 => FtpResponseClass::TransientNegative,
59 500..=599 => FtpResponseClass::PermanentNegative,
60 _ => FtpResponseClass::Unknown,
61 }
62 }
63
64 pub fn is_success(&self) -> bool {
66 matches!(
67 self,
68 FtpResponseClass::PositivePreliminary
69 | FtpResponseClass::PositiveCompletion
70 | FtpResponseClass::PositiveIntermediate
71 )
72 }
73
74 pub fn is_transient(&self) -> bool {
76 *self == FtpResponseClass::TransientNegative
77 }
78
79 pub fn is_permanent(&self) -> bool {
81 *self == FtpResponseClass::PermanentNegative
82 }
83}
84
85#[derive(Debug, Clone)]
87pub struct FtpResponse {
88 pub code: u16,
89 pub message: String,
90}
91
92impl FtpResponse {
93 pub fn is_success(&self) -> bool {
94 (100..400).contains(&self.code)
95 }
96
97 pub fn is_intermediate(&self) -> bool {
98 (100..200).contains(&self.code)
99 }
100
101 pub fn is_positive_completion(&self) -> bool {
102 (200..300).contains(&self.code)
103 }
104
105 pub fn is_positive_preliminary(&self) -> bool {
106 (100..200).contains(&self.code)
107 }
108
109 pub fn class(&self) -> FtpResponseClass {
111 FtpResponseClass::from_code(self.code)
112 }
113
114 pub fn is_transient_error(&self) -> bool {
116 self.class().is_transient()
117 }
118
119 pub fn is_permanent_error(&self) -> bool {
121 self.class().is_permanent()
122 }
123}
124
125pub struct FtpConnection {
126 pub stream: BufReader<TcpStream>,
127 pub options: FtpOptions,
128 #[allow(dead_code)] pub host: String,
130 #[allow(dead_code)] pub port: u16,
132}
133
134impl FtpConnection {
135 pub async fn connect(
136 host: &str,
137 port: u16,
138 options: Option<FtpOptions>,
139 ) -> Result<Self, String> {
140 let options = options.unwrap_or_default();
141 info!("FTP connecting: {}:{}", host, port);
142
143 let stream = timeout(options.connect_timeout, TcpStream::connect((host, port)))
144 .await
145 .map_err(|_| {
146 format!(
147 "FTP connection timeout ({}s)",
148 options.connect_timeout.as_secs()
149 )
150 })?
151 .map_err(|e| format!("FTP connection failed: {}", e))?;
152
153 let mut conn = Self {
154 stream: BufReader::new(stream),
155 options,
156 host: host.to_string(),
157 port,
158 };
159
160 let welcome = conn.read_response().await?;
161 if !welcome.is_positive_completion() && !welcome.is_positive_preliminary() {
162 return Err(format!(
163 "FTP server refused connection: {} {}",
164 welcome.code, welcome.message
165 ));
166 }
167 debug!("FTP connected: {}", welcome.message);
168
169 Ok(conn)
170 }
171
172 pub async fn login(&mut self) -> Result<(), String> {
173 debug!("Sending USER command: {}", self.options.username);
174 self.send_command(&format!("USER {}", self.options.username))
175 .await?;
176 let resp = self.read_response().await?;
177
178 if resp.code == 331 || resp.code == 332 {
179 debug!("Password required, sending PASS command");
180 self.send_command(&format!("PASS {}", self.options.password))
181 .await?;
182 let pass_resp = self.read_response().await?;
183 if !pass_resp.is_positive_completion() {
184 return Err(format!(
185 "FTP login failed: {} {}",
186 pass_resp.code, pass_resp.message
187 ));
188 }
189 info!("FTP login successful");
190 } else if !resp.is_positive_completion() {
191 return Err(format!("FTP login failed: {} {}", resp.code, resp.message));
192 } else {
193 info!("FTP login successful (no password required)");
194 }
195
196 Ok(())
197 }
198
199 pub async fn cwd(&mut self, path: &str) -> Result<(), String> {
200 debug!("Changing directory: {}", path);
201 self.send_command(&format!("CWD {}", path)).await?;
202 let resp = self.read_response().await?;
203 if !resp.is_positive_completion() {
204 return Err(format!("CWD failed: {} {}", resp.code, resp.message));
205 }
206 Ok(())
207 }
208
209 pub async fn size(&mut self, filename: &str) -> Result<u64, String> {
210 debug!("Querying file size: {}", filename);
211 self.send_command(&format!("SIZE {}", filename)).await?;
212 let resp = self.read_response().await?;
213 if resp.code == 213 {
214 let size_str = resp.message.trim();
215 size_str
216 .parse::<u64>()
217 .map_err(|e| format!("Failed to parse file size: {} ({})", e, size_str))
218 } else {
219 Err(format!(
220 "SIZE command failed: {} {}",
221 resp.code, resp.message
222 ))
223 }
224 }
225
226 pub async fn mdtm(&mut self, filename: &str) -> Result<String, String> {
227 debug!("Querying modification time: {}", filename);
228 self.send_command(&format!("MDTM {}", filename)).await?;
229 let resp = self.read_response().await?;
230 if resp.code == 213 {
231 Ok(resp.message.trim().to_string())
232 } else {
233 Err(format!(
234 "MDTM command failed: {} {}",
235 resp.code, resp.message
236 ))
237 }
238 }
239
240 pub async fn pasv(&mut self) -> Result<(String, u16), String> {
241 debug!("Requesting passive mode data connection");
242 self.send_command("PASV").await?;
243 let resp = self.read_response().await?;
244 if resp.code != 227 {
245 return Err(format!("PASV failed: {} {}", resp.code, resp.message));
246 }
247
248 match Self::parse_pasv_response(&resp.message) {
249 Some(addr) => {
250 debug!("PASV data channel: {}:{}", addr.0, addr.1);
251 Ok(addr)
252 }
253 None => Err("Failed to parse PASV response".to_string()),
254 }
255 }
256
257 pub async fn epsv(&mut self) -> Result<u16, String> {
258 debug!("Requesting extended passive mode");
259 self.send_command("EPSV").await?;
260 let resp = self.read_response().await?;
261 if resp.code == 229 {
262 if let Some(port) = Self::parse_epsv_response(&resp.message) {
263 debug!("EPSV port: {}", port);
264 return Ok(port);
265 }
266 } else if resp.code == 590 || resp.code == 500 || resp.code == 501 {
267 warn!("Server does not support EPSV, falling back to PASV mode");
268 let (host, port) = self.pasv().await?;
269 let _ = host;
270 return Ok(port);
271 }
272 Err(format!("EPSV failed: {} {}", resp.code, resp.message))
273 }
274
275 pub async fn port_active(&mut self) -> Result<u16, String> {
278 debug!("Requesting active mode data connection (IPv4)");
279
280 let listener = tokio::net::TcpListener::bind("0.0.0.0:0")
282 .await
283 .map_err(|e| format!("Failed to bind local port: {}", e))?;
284
285 let local_addr = listener
286 .local_addr()
287 .map_err(|e| format!("Failed to get local address: {}", e))?;
288
289 let ip = local_addr.ip();
290 let port = local_addr.port();
291
292 let octets = match ip {
294 std::net::IpAddr::V4(v4) => v4.octets(),
295 std::net::IpAddr::V6(_) => {
296 return Err(
297 "IPv6 address not supported for PORT command, use EPRT instead".to_string(),
298 );
299 }
300 };
301
302 let p1 = port / 256;
303 let p2 = port % 256;
304 let port_cmd = format!(
305 "PORT {},{},{},{},{},{}",
306 octets[0], octets[1], octets[2], octets[3], p1, p2
307 );
308
309 debug!("Sending PORT command: {}", port_cmd);
310 self.send_command(&port_cmd).await?;
311 let resp = self.read_response().await?;
312 if !resp.is_positive_completion() {
313 return Err(format!("PORT failed: {} {}", resp.code, resp.message));
314 }
315
316 debug!("Waiting for active mode data connection on port {}", port);
319
320 drop(listener); Ok(port)
324 }
325
326 pub async fn eprt_active(&mut self) -> Result<(String, u16), String> {
329 debug!("Requesting extended active mode data connection");
330
331 let listener = tokio::net::TcpListener::bind("::0")
333 .await
334 .map_err(|e| format!("Failed to bind local port: {}", e))?;
335
336 let local_addr = listener
337 .local_addr()
338 .map_err(|e| format!("Failed to get local address: {}", e))?;
339
340 let ip = local_addr.ip();
341 let port = local_addr.port();
342
343 let (proto, addr_str) = match ip {
346 std::net::IpAddr::V4(v4) => ("1", v4.to_string()),
347 std::net::IpAddr::V6(v6) => ("2", v6.to_string()),
348 };
349
350 let eprt_cmd = format!("EPRT |{}|{}|{}", proto, addr_str, port);
351
352 debug!("Sending EPRT command: {}", eprt_cmd);
353 self.send_command(&eprt_cmd).await?;
354 let resp = self.read_response().await?;
355 if !resp.is_positive_completion() {
356 return Err(format!("EPRT failed: {} {}", resp.code, resp.message));
357 }
358
359 debug!("EPRT successful, listening {}:{}", addr_str, port);
360 drop(listener);
361
362 Ok((addr_str, port))
363 }
364
365 pub async fn rest(&mut self, offset: u64) -> Result<(), String> {
366 debug!("Setting resume offset: {}", offset);
367 self.send_command(&format!("REST {}", offset)).await?;
368 let resp = self.read_response().await?;
369 if resp.code != 350 {
370 return Err(format!("REST failed: {} {}", resp.code, resp.message));
371 }
372 Ok(())
373 }
374
375 pub async fn retr(&mut self, filename: &str) -> Result<(), String> {
376 debug!("Preparing to download file: {}", filename);
377 self.send_command(&format!("RETR {}", filename)).await?;
378 let resp = self.read_response().await?;
379 if !resp.is_positive_preliminary() {
380 return Err(format!("RETR failed: {} {}", resp.code, resp.message));
381 }
382 Ok(())
383 }
384
385 pub async fn type_image(&mut self) -> Result<(), String> {
386 debug!("Setting transfer type to binary (I)");
387 self.send_command("TYPE I").await?;
388 let resp = self.read_response().await?;
389 if !resp.is_positive_completion() {
390 return Err(format!("TYPE I failed: {} {}", resp.code, resp.message));
391 }
392 Ok(())
393 }
394
395 pub async fn quit(mut self) -> Result<(), String> {
396 debug!("Sending QUIT command");
397 self.send_command("QUIT").await?;
398 let resp = self.read_response().await?;
399 info!("FTP disconnected: {}", resp.message);
400 Ok(())
401 }
402
403 pub async fn abor(&mut self) -> Result<(), String> {
405 debug!("Sending ABOR command to abort transfer");
406 self.stream
408 .get_mut()
409 .write_all(b"\xff\xf4") .await
411 .map_err(|e| format!("Failed to send Telnet IP: {}", e))?;
412 self.stream
413 .get_mut()
414 .flush()
415 .await
416 .map_err(|e| format!("Failed to flush buffer: {}", e))?;
417
418 tokio::time::sleep(Duration::from_millis(100)).await;
419
420 self.send_command("ABOR").await?;
421 let resp = self.read_response().await?;
422 if resp.code == 226 || resp.code == 225 {
424 debug!("Transfer successfully aborted: {}", resp.message);
425 Ok(())
426 } else {
427 warn!("ABOR response unusual: {} {}", resp.code, resp.message);
428 Ok(())
429 }
430 }
431
432 pub async fn list(&mut self, path: Option<&str>) -> Result<FtpResponse, String> {
434 match path {
435 Some(p) => {
436 debug!("Listing directory details: {}", p);
437 self.send_command(&format!("LIST {}", p)).await?
438 }
439 None => {
440 debug!("Listing current directory details");
441 self.send_command("LIST").await?
442 }
443 };
444 let resp = self.read_response().await?;
445 Ok(resp)
446 }
447
448 pub async fn nlst(&mut self, path: Option<&str>) -> Result<FtpResponse, String> {
450 match path {
451 Some(p) => {
452 debug!("Listing directory names: {}", p);
453 self.send_command(&format!("NLST {}", p)).await?
454 }
455 None => {
456 debug!("Listing current directory names");
457 self.send_command("NLST").await?
458 }
459 };
460 let resp = self.read_response().await?;
461 Ok(resp)
462 }
463
464 pub async fn type_ascii(&mut self) -> Result<(), String> {
466 debug!("Setting transfer type to ASCII (A)");
467 self.send_command("TYPE A").await?;
468 let resp = self.read_response().await?;
469 if !resp.is_positive_completion() {
470 return Err(format!("TYPE A failed: {} {}", resp.code, resp.message));
471 }
472 Ok(())
473 }
474
475 pub async fn noop(&mut self) -> Result<(), String> {
477 debug!("Sending NOOP keep-alive command");
478 self.send_command("NOOP").await?;
479 let resp = self.read_response().await?;
480 if resp.code == 200 {
481 debug!("NOOP keep-alive successful");
482 Ok(())
483 } else {
484 Err(format!("NOOP failed: {} {}", resp.code, resp.message))
485 }
486 }
487
488 pub fn start_keepalive(&self) -> Option<tokio::task::JoinHandle<()>> {
491 let keepalive_duration = self.options.keepalive_interval?;
492
493 let handle = tokio::spawn(async move {
496 let mut ticker = interval(keepalive_duration);
497 loop {
498 ticker.tick().await;
499 debug!("FTP keep-alive tick");
501 }
502 });
503
504 Some(handle)
505 }
506
507 pub fn get_data_stream(&mut self) -> &mut BufReader<TcpStream> {
508 &mut self.stream
509 }
510
511 async fn send_command(&mut self, command: &str) -> Result<(), String> {
512 debug!("FTP command: {}", command.trim());
513 self.stream
514 .write_all(command.as_bytes())
515 .await
516 .map_err(|e| format!("Failed to send FTP command: {}", e))?;
517 self.stream
518 .write_all(b"\r\n")
519 .await
520 .map_err(|e| format!("Failed to send newline: {}", e))?;
521 self.stream
522 .flush()
523 .await
524 .map_err(|e| format!("Failed to flush buffer: {}", e))?;
525 Ok(())
526 }
527
528 pub async fn read_response(&mut self) -> Result<FtpResponse, String> {
529 let mut line = String::new();
530 let mut code: Option<u16> = None;
531 let mut message = String::new();
532 let mut is_multiline = false;
533
534 loop {
535 line.clear();
536 let bytes_read = timeout(self.options.read_timeout, self.stream.read_line(&mut line))
537 .await
538 .map_err(|_| "FTP read timeout".to_string())?
539 .map_err(|e| format!("Failed to read FTP response: {}", e))?;
540
541 if bytes_read == 0 {
542 break;
543 }
544
545 let trimmed = line.trim_end();
546 if trimmed.len() < 4 {
547 continue;
548 }
549
550 let response_code: u16 = trimmed[..3].parse().unwrap_or(0);
551
552 if code.is_none() {
553 code = Some(response_code);
554 }
555
556 let separator = trimmed.as_bytes()[3];
557 if separator == b'-' && !is_multiline {
558 is_multiline = true;
559 message.push_str(&trimmed[4..]);
560 message.push('\n');
561 } else if separator == b' '
562 || (is_multiline && trimmed.starts_with(&format!("{:3} ", code.unwrap_or(0))))
563 {
564 message.push_str(&trimmed[4..]);
565 break;
566 } else if is_multiline {
567 message.push_str(&trimmed[4..]);
568 message.push('\n');
569 }
570 }
571
572 let code_val = code.unwrap_or(0);
573 debug!("FTP response: {} {}", code_val, message.trim());
574 Ok(FtpResponse {
575 code: code_val,
576 message,
577 })
578 }
579
580 fn parse_pasv_response(message: &str) -> Option<(String, u16)> {
581 let start = message.find('(')?;
582 let end = message.find(')')?;
583 let inner = &message[start + 1..end];
584
585 let parts: Vec<&str> = inner.split(',').collect();
586 if parts.len() != 6 {
587 return None;
588 }
589
590 let h1: u8 = parts[0].trim().parse().ok()?;
591 let h2: u8 = parts[1].trim().parse().ok()?;
592 let h3: u8 = parts[2].trim().parse().ok()?;
593 let h4: u8 = parts[3].trim().parse().ok()?;
594 let p1: u16 = parts[4].trim().parse().ok()?;
595 let p2: u16 = parts[5].trim().parse().ok()?;
596
597 let host = format!("{}.{}.{}.{}", h1, h2, h3, h4);
598 let port = p1 * 256 + p2;
599
600 Some((host, port))
601 }
602
603 fn parse_epsv_response(message: &str) -> Option<u16> {
604 let start = message.rfind('|')?;
605 let prev_pipe = message[..start].rfind('|')?;
606 let port_str = &message[prev_pipe + 1..start];
607 port_str.parse::<u16>().ok()
608 }
609}
610
611#[cfg(test)]
612mod tests {
613 use super::*;
614
615 #[test]
616 fn test_ftp_response_checks() {
617 let ok = FtpResponse {
618 code: 226,
619 message: "Transfer complete".into(),
620 };
621 assert!(ok.is_success());
622 assert!(ok.is_positive_completion());
623 assert!(!ok.is_permanent_error());
624 assert!(!ok.is_transient_error());
625
626 let intermediate = FtpResponse {
627 code: 150,
628 message: "Opening data channel".into(),
629 };
630 assert!(intermediate.is_success());
631 assert!(intermediate.is_positive_preliminary());
632
633 let error = FtpResponse {
634 code: 550,
635 message: "File not found".into(),
636 };
637 assert!(!error.is_success());
638 assert!(error.is_permanent_error());
639 assert!(!error.is_transient_error());
640
641 let transient = FtpResponse {
642 code: 425,
643 message: "Can't open data connection".into(),
644 };
645 assert!(transient.is_transient_error());
646 assert!(!transient.is_permanent_error());
647 }
648
649 #[test]
650 fn test_response_class_classification() {
651 assert_eq!(
652 FtpResponseClass::from_code(150),
653 FtpResponseClass::PositivePreliminary
654 );
655 assert_eq!(
656 FtpResponseClass::from_code(226),
657 FtpResponseClass::PositiveCompletion
658 );
659 assert_eq!(
660 FtpResponseClass::from_code(331),
661 FtpResponseClass::PositiveIntermediate
662 );
663 assert_eq!(
664 FtpResponseClass::from_code(425),
665 FtpResponseClass::TransientNegative
666 );
667 assert_eq!(
668 FtpResponseClass::from_code(550),
669 FtpResponseClass::PermanentNegative
670 );
671 assert_eq!(FtpResponseClass::from_code(999), FtpResponseClass::Unknown);
672 }
673
674 #[test]
675 fn test_response_class_methods() {
676 let prelim = FtpResponseClass::PositivePreliminary;
677 assert!(prelim.is_success());
678 assert!(!prelim.is_transient());
679 assert!(!prelim.is_permanent());
680
681 let transient = FtpResponseClass::TransientNegative;
682 assert!(!transient.is_success());
683 assert!(transient.is_transient());
684 assert!(!transient.is_permanent());
685
686 let permanent = FtpResponseClass::PermanentNegative;
687 assert!(!permanent.is_success());
688 assert!(!permanent.is_transient());
689 assert!(permanent.is_permanent());
690 }
691
692 #[test]
693 fn test_parse_pasv() {
694 let msg = "Entering Passive Mode (192,168,1,1,195,123)";
695 let result = FtpConnection::parse_pasv_response(msg);
696 assert!(result.is_some());
697 let (host, port) = result.unwrap();
698 assert_eq!(host, "192.168.1.1");
699 assert_eq!(port, 195 * 256 + 123);
700 }
701
702 #[test]
703 fn test_parse_pasv_various_formats() {
704 let msg1 = "227 Entering Passive Mode (192,168,1,100,200,10)";
706 let result1 = FtpConnection::parse_pasv_response(msg1);
707 assert_eq!(result1, Some(("192.168.1.100".to_string(), 200 * 256 + 10)));
708
709 let msg2 = "(10,0,0,1,0,21)";
711 let result2 = FtpConnection::parse_pasv_response(msg2);
712 assert_eq!(result2, Some(("10.0.0.1".to_string(), 21)));
713
714 let msg3 = "192,168,1,1,195,123";
716 let result3 = FtpConnection::parse_pasv_response(msg3);
717 assert!(result3.is_none());
718
719 let msg4 = "(192,168,1,1,195)";
721 let result4 = FtpConnection::parse_pasv_response(msg4);
722 assert!(result4.is_none());
723 }
724
725 #[test]
726 fn test_parse_epsv() {
727 let msg = "Entering Extended Passive Mode (|||50001|)";
728 let result = FtpConnection::parse_epsv_response(msg);
729 assert_eq!(result, Some(50001));
730 }
731
732 #[test]
733 fn test_parse_epsv_various_formats() {
734 let msg1 = "229 |||50001|";
736 let result1 = FtpConnection::parse_epsv_response(msg1);
737 assert_eq!(result1, Some(50001));
738
739 let msg2 = "Entering Extended Passive Mode (|||60000|)";
741 let result2 = FtpConnection::parse_epsv_response(msg2);
742 assert_eq!(result2, Some(60000));
743
744 let msg3 = "50001";
746 let result3 = FtpConnection::parse_epsv_response(msg3);
747 assert!(result3.is_none());
748 }
749
750 #[test]
751 fn test_ftp_options_default() {
752 let opts = FtpOptions::default();
753 assert_eq!(opts.username, "anonymous");
754 assert_eq!(opts.password, "aria2@");
755 assert!(opts.passive_mode);
756 assert_eq!(opts.connect_timeout, Duration::from_secs(30));
757 assert_eq!(opts.read_timeout, Duration::from_secs(30));
758 assert!(opts.keepalive_interval.is_some());
759 assert_eq!(opts.keepalive_interval.unwrap(), Duration::from_secs(60));
760 assert_eq!(opts.max_retries, 3);
761 }
762
763 #[test]
764 fn test_build_port_command_ipv4() {
765 let octets: [u8; 4] = [192, 168, 1, 100];
767 let port: u16 = 50000;
768 let p1 = port / 256;
769 let p2 = port % 256;
770 let port_cmd = format!(
771 "PORT {},{},{},{},{},{}",
772 octets[0], octets[1], octets[2], octets[3], p1, p2
773 );
774 assert_eq!(port_cmd, "PORT 192,168,1,100,195,80");
775 }
776
777 #[test]
778 fn test_build_eprt_command() {
779 let addr = "192.168.1.100";
781 let port: u16 = 50001;
782 let eprt_cmd = format!("EPRT |1|{}|{}", addr, port);
783 assert_eq!(eprt_cmd, "EPRT |1|192.168.1.100|50001");
784
785 let addr_v6 = "::1";
787 let eprt_cmd_v6 = format!("EPRT |2|{}|{}", addr_v6, port);
788 assert_eq!(eprt_cmd_v6, "EPRT |2|::1|50001");
789 }
790}