#!/usr/bin/env python3 """Minimal FTP server for KIO ftp ioslave testing (bug: LIST misparsed as MLSD). Modes: mlsd - FEAT advertises MLST+MDTM, MLSD works nofeat - FEAT rejected (500), only LIST works reject - FEAT advertises MLST, but MLSD is rejected (550) -> LIST fallback Docroot: /srv/ftproot Usage: python3 ftpsrv.py reject 2121 """ import socket, socketserver, sys, threading, os, time, stat MODE = sys.argv[1] if len(sys.argv) > 1 else "mlsd" PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 2121 ROOT = "/srv/ftproot" def list_entry_mlsd(path, name): st = os.lstat(path) typ = "dir" if stat.S_ISDIR(st.st_mode) else ("OS.unix=slink" if stat.S_ISLNK(st.st_mode) else "file") modify = time.strftime("%Y%m%d%H%M%S", time.gmtime(st.st_mtime)) facts = f"Type={typ};Size={st.st_size};Modify={modify};UNIX.mode=0{oct(st.st_mode)[-3:]}; {name}" return facts def list_entry_list(path, name): st = os.lstat(path) import pwd, grp perm = stat.filemode(st.st_mode) try: uname = pwd.getpwuid(st.st_uid).pw_name except KeyError: uname = str(st.st_uid) try: gname = grp.getgrgid(st.st_gid).gr_name except KeyError: gname = str(st.st_gid) t = time.strftime("%b %d %H:%M", time.gmtime(st.st_mtime)) tgt = "" if stat.S_ISLNK(st.st_mode): tgt = " -> " + os.readlink(path) return f"{perm} 1 {uname} {gname} {st.st_size:>8} {t} {name}{tgt}" class Handler(socketserver.BaseRequestHandler): def setup(self): self.cwd = "/" self.datasock = None self.datasrv = None self.buf = b"" def reply(self, line): self.request.sendall((line + "\r\n").encode()) def readline(self): while b"\n" not in self.buf: data = self.request.recv(4096) if not data: return None self.buf += data line, self.buf = self.buf.split(b"\n", 1) return line.decode(errors="replace").strip() def open_data(self): if self.datasrv: conn, _ = self.datasrv.accept() return conn return None def real(self, p): p = os.path.normpath(os.path.join(self.cwd, p)) if not p.startswith("/") else os.path.normpath(p) if not p.startswith("/"): p = "/" return p, os.path.join(ROOT, p.lstrip("/")) def handle(self): self.reply("220 test ftpd ready") while True: line = self.readline() if line is None: break parts = line.split(" ", 1) cmd = parts[0].upper() arg = parts[1] if len(parts) > 1 else "" if cmd == "USER": self.reply("331 password please") elif cmd == "PASS": self.reply("230 logged in") elif cmd == "SYST": self.reply("215 UNIX Type: L8") elif cmd == "PWD": self.reply(f'257 "{self.cwd}"') elif cmd == "TYPE": self.reply("200 type set") elif cmd == "FEAT": if MODE == "nofeat": self.reply("500 FEAT not supported") else: self.request.sendall(b"211-Features:\r\n MDTM\r\n") if MODE in ("mlsd", "reject"): self.request.sendall(b" MLST Type*;Size*;Modify*;UNIX.mode*;\r\n") self.request.sendall(b" UTF8\r\n211 End\r\n") elif cmd == "MDTM": v, rp = self.real(arg) try: st = os.lstat(rp) self.reply("213 " + time.strftime("%Y%m%d%H%M%S", time.gmtime(st.st_mtime))) except OSError: self.reply("550 no such file") elif cmd == "SIZE": v, rp = self.real(arg) try: self.reply(f"213 {os.lstat(rp).st_size}") except OSError: self.reply("550 no such file") elif cmd == "PASV": if self.datasrv: self.datasrv.close() self.datasrv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.datasrv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.datasrv.bind(("0.0.0.0", 0)) self.datasrv.listen(1) port = self.datasrv.getsockname()[1] self.reply(f"227 Entering Passive Mode (127,0,0,1,{port//256},{port%256})") elif cmd == "EPSV": if self.datasrv: self.datasrv.close() self.datasrv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.datasrv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.datasrv.bind(("0.0.0.0", 0)) self.datasrv.listen(1) port = self.datasrv.getsockname()[1] self.reply(f"229 Entering Extended Passive Mode (|||{port}|)") elif cmd in ("MLSD", "LIST", "NLST"): if cmd == "MLSD" and MODE == "reject": self.reply("550 MLSD rejected by policy") continue if cmd == "MLSD" and MODE == "nofeat": self.reply("500 unknown command") continue v, rp = self.real(arg or ".") if not os.path.isdir(rp): # MLSD of a file: single fact line if os.path.exists(rp) and cmd == "MLSD": self.reply("150 opening") c = self.open_data() c.sendall((list_entry_mlsd(rp, os.path.basename(rp)) + "\r\n").encode()) c.close() self.reply("226 done") else: self.reply("550 not a directory") continue self.reply("150 here comes the listing") c = self.open_data() out = [] for name in sorted(os.listdir(rp)): fp = os.path.join(rp, name) if cmd == "MLSD": out.append(list_entry_mlsd(fp, name)) elif cmd == "LIST": out.append(list_entry_list(fp, name)) else: out.append(name) c.sendall(("\r\n".join(out) + "\r\n").encode()) c.close() self.reply("226 listing done") elif cmd == "CWD": v, rp = self.real(arg) if os.path.isdir(rp): self.cwd = v self.reply("250 cwd ok") else: self.reply("550 no such dir") elif cmd == "RETR": v, rp = self.real(arg) try: data = open(rp, "rb").read() self.reply("150 sending file") c = self.open_data() c.sendall(data) c.close() self.reply("226 transfer complete") except OSError: self.reply("550 no such file") elif cmd == "QUIT": self.reply("221 bye"); break else: self.reply(f"502 {cmd} not implemented") class Server(socketserver.ThreadingTCPServer): allow_reuse_address = True os.makedirs(ROOT, exist_ok=True) srv = Server(("0.0.0.0", PORT), Handler) print(f"ftpsrv mode={MODE} port={PORT} root={ROOT}", flush=True) srv.serve_forever()