import uos import socket import gc import asyncio class FTPServer: def __init__(self, ip_address=None, port=21, timeout=None): if not ip_address: raise ValueError("La dirección IP es obligatoria.") self.ip_address = ip_address self.port = port self.timeout = timeout async def run(self): puerto_datos = 13333 ftpsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) datasocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) ftpsocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) datasocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) ftpsocket.bind((self.ip_address, self.port)) datasocket.bind((self.ip_address, puerto_datos)) ftpsocket.listen(1) ftpsocket.settimeout(self.timeout) datasocket.listen(1) datasocket.settimeout(None) msg_250_OK = '250 OK\r\n' msg_550_fail = '550 Fallido\r\n' print(f"FTP servidor se inició en {self.ip_address}:{self.port}") try: datoscliente = None direccion_remota = None corre = True while corre: cl, remote_addr = ftpsocket.accept() cl.settimeout(300) cwd = '/' try: cl.sendall("220 Hola, Esta es ESP32-S3.\r\n") while True: gc.collect() data = cl.readline().decode("utf-8").rstrip("\r\n") if len(data) <= 0: print("Cliente desaparecido") corre = False break comando = data.split(" ")[0].upper() datos = data[len(comando):].lstrip() ruta = obtener_ruta_absoluta(cwd, datos) print("Comando={}, Datos={}".format(comando, datos)) if comando == "USER": cl.sendall("230 Iniciado sesión.\r\n") elif comando == "SYST": cl.sendall("215 UNIX Type: L8\r\n") elif comando == "NOOP": cl.sendall("200 OK\r\n") elif comando == "FEAT": cl.sendall("211 sin funciones\r\n") elif comando == "PWD" or comando == "XPWD": cl.sendall('257 "{}"\r\n'.format(cwd)) elif comando == "CWD" or comando == "XCWD": try: uos.listdir(ruta) cwd = ruta cl.sendall(msg_250_OK) except: cl.sendall(msg_550_fail) elif comando == "CDUP": cwd = obtener_ruta_absoluta(cwd, "..") cl.sendall(msg_250_OK) elif comando == "TYPE": cl.sendall('200 Modo de transferencia establecido\r\n') elif comando == "SIZE": try: size = uos.stat(ruta)[6] cl.sendall('213 {}\r\n'.format(size)) except: cl.sendall(msg_550_fail) elif comando == "QUIT": cl.sendall('221 Adios.\r\n') corre = False break elif comando == "PASV": cl.sendall('227 Entrar en modo pasivo ({},{},{}).\r\n'.format( self.ip_address.replace('.', ','), puerto_datos >> 8, puerto_datos % 256)) datoscliente, data_addr = datasocket.accept() print("FTP Conexión de datos desde:", data_addr) puerto_datos = 13333 elif comando == "PORT": items = datos.split(",") if len(items) >= 6: data_addr = '.'.join(items[:4]) if data_addr == "127.0.1.1": data_addr = remote_addr puerto_datos = int(items[4]) * 256 + int(items[5]) datoscliente = socket.socket(socket.AF_INET, socket.SOCK_STREAM) datoscliente.settimeout(10) datoscliente.connect((data_addr, puerto_datos)) print("FTP Conexión de datos con:", data_addr) cl.sendall('200 OK\r\n') else: cl.sendall('504 Fallido\r\n') elif comando in ["LIST", "NLST"]: lugar = ruta if not datos.startswith("-") else cwd try: cl.sendall("150 Aquí viene el listado del directorio.\r\n") enviar_listado(lugar, datoscliente, comando == "LIST" or datos == "-l") cl.sendall("226 Listado.\r\n") except: cl.sendall(msg_550_fail) if datoscliente: datoscliente.close() datoscliente = None elif comando == "RETR": try: cl.sendall("150 Apertura de la conexión de datos.\r\n") enviar_archivo(ruta, datoscliente) cl.sendall("226 Transferencia completa.\r\n") except: cl.sendall(msg_550_fail) if datoscliente: datoscliente.close() datoscliente = None elif comando == "STOR": try: cl.sendall("150 Ok para enviar datos.\r\n") guardar_archivo(ruta, datoscliente) cl.sendall("226 Transfer complete.\r\n") except: cl.sendall(msg_550_fail) if datoscliente: datoscliente.close() datoscliente = None elif comando == "DELE": try: uos.remove(ruta) cl.sendall(msg_250_OK) except: cl.sendall(msg_550_fail) elif comando in ["RMD", "XRMD"]: try: uos.rmdir(ruta) cl.sendall(msg_250_OK) except: cl.sendall(msg_550_fail) elif comando in ["MKD", "XMKD"]: try: uos.mkdir(ruta) cl.sendall(msg_250_OK) except: cl.sendall(msg_550_fail) elif comando == "RNFR": direccion_remota = ruta cl.sendall("350 Rename from\r\n") elif comando == "RNTO": if direccion_remota is not None: try: uos.rename(direccion_remota, ruta) cl.sendall(msg_250_OK) except: cl.sendall(msg_550_fail) else: cl.sendall(msg_550_fail) direccion_remota = None elif comando == "MDTM": try: tm = uos.stat(ruta)[8] cl.sendall('213 {:04d}{:02d}{:02d}{:02d}{:02d}{:02d}\r\n'.format(*tm)) except: cl.sendall('550 Fail\r\n') elif comando == "STAT": if datos == "": cl.sendall(f"211-Conectado a ({remote_addr[0]})\r\n" f" Data address ({self.ip_address})\r\n" "211 TYPE: Binary STRU: File MODE: Stream\r\n") else: cl.sendall("213-Directorio listado:\r\n") enviar_listado(ruta, cl, True) cl.sendall("213 Hecho.\r\n") else: cl.sendall("502 Comando no compatible.\r\n") print(f"Comando no compatible {comando} con datos {datos}") except Exception as err: print(err) finally: cl.close() except Exception as e: print(e) finally: datasocket.close() ftpsocket.close() if datoscliente: datoscliente.close() def enviar_listado(ruta, datoscliente, full): try: for fichero in sorted(uos.listdir(ruta), key=str.lower): datoscliente.sendall(crear_descripcion(ruta, fichero, full)) except: patron = ruta.split("/")[-1] ruta = ruta[:-(len(patron) + 1)] if ruta == "": ruta = "/" for fichero in sorted(uos.listdir(ruta), key=str.lower): if compara_patron(fichero, patron): datoscliente.sendall(crear_descripcion(ruta, fichero, full)) def crear_descripcion(ruta, fichero, full): if full: estado_archivo = uos.stat(obtener_ruta_absoluta(ruta, fichero)) permisos_archivo = ("drwxr-xr-x" if (estado_archivo[0] & 0o170000 == 0o040000) else "-rw-r--r--") tamano_archivo = estado_archivo[6] descripcion = "{} 1 owner group {:>10} Jan 1 2000 {}\r\n".format( permisos_archivo, tamano_archivo, fichero) else: descripcion = fichero + "\r\n" return descripcion def enviar_archivo(ruta, datoscliente): with open(ruta, "rb") as archivo: fragmento = archivo.read(512) while len(fragmento) > 0: datoscliente.sendall(fragmento) fragmento = archivo.read(512) def guardar_archivo(ruta, datoscliente): with open(ruta, "wb") as archivo: fragmento = datoscliente.recv(512) while len(fragmento) > 0: archivo.write(fragmento) fragmento = datoscliente.recv(512) def obtener_ruta_absoluta(cwd, datos): if datos.startswith('/'): cwd = "/" for token in datos.split("/"): if token == '..': if cwd != '/': cwd = '/'.join(cwd.split('/')[:-1]) if cwd == '': cwd = '/' elif token != '.' and token != '': if cwd == '/': cwd += token else: cwd = cwd + '/' + token return cwd def compara_patron(fichero, patron): pi = 0 si = 0 while pi < len(patron) and si < len(fichero): if (fichero[si] == patron[pi]) or (patron[pi] == '?'): si += 1 pi += 1 else: if patron[pi] == '*': if (pi + 1) == len(patron): return True while si < len(fichero): if compara_patron(fichero[si:], patron[pi+1:]): return True else: si += 1 return False else: return False if pi == len(patron.rstrip("*")) and si == len(fichero): return True else: return False