Attachment 22824 Details for Bug 61241 – FTP-сервер для воспроизведения, v2: docroot без root (~/ftproot или argv[3])

FTP-сервер для воспроизведения, v2: docroot без root (~/ftproot или argv[3])

ftpsrv.py (text/plain), 7.31 KB, created by Alexander Makeenkov on 2026-10-28 09:55:17 MSK

(hide)

Description:

Filename:

MIME Type:

Creator: Alexander Makeenkov

Created: 2026-10-28 09:55:17 MSK

Size: 7.31 KB

patch

obsolete

>#!/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: ~/ftproot by default, or argv[3]
>Usage: python3 ftpsrv.py reject 2121 [/path/to/docroot]
>"""
>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 = os.path.abspath(os.path.expanduser(sys.argv[3])) if len(sys.argv) > 3 else os.path.expanduser("~/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()

Actions: View

Attachments on bug 61241: 22822 | 22823 | 22824