Files
ASE/ftpReceiver/FtpCsvReceiver.py
2024-11-17 15:09:37 +01:00

258 lines
9.0 KiB
Python
Executable File

#!/usr/bin/env python3
import sys
import os
import shutil
# import ssl
import re
import logging
import psycopg2
from hashlib import md5
from pathlib import Path
from datetime import datetime
from utils.time import timestamp_fmt as ts
from utils.time import date_refmt as df
from utils.config import set_config as setting
from pyftpdlib.handlers import FTPHandler, TLS_FTPHandler
from pyftpdlib.servers import FTPServer
from pyftpdlib.authorizers import DummyAuthorizer, AuthenticationFailed
class DummyMD5Authorizer(DummyAuthorizer):
def __init__(self, cfg):
# Initialize the DummyAuthorizer and add the admin user
super().__init__()
self.add_user(
cfg.adminuser[0], cfg.adminuser[1], cfg.adminuser[2], perm=cfg.adminuser[3])
# Definisci la connessione al database
conn = psycopg2.connect(
dbname=cfg.dbname,
user=cfg.dbuser,
password=cfg.dbpass,
host=cfg.dbhost,
port=cfg.dbport
)
# Crea un cursore
cur = conn.cursor()
cur.execute("SELECT ftpuser, hash, virtpath, perm FROM virtusers")
for ftpuser, hash, virtpath, perm in cur.fetchall():
self.add_user(ftpuser, hash, virtpath, perm)
try:
Path(cfg.virtpath + ftpuser).mkdir(parents=True, exist_ok=True)
except:
self.responde('551 Error in create virtual user path.')
def validate_authentication(self, username, password, handler):
# Validate the user's password against the stored hash
hash = md5(password.encode("UTF-8")).hexdigest()
try:
if self.user_table[username]["pwd"] != hash:
raise KeyError
except KeyError:
raise AuthenticationFailed
class ASEHandler(FTPHandler):
def __init__(self, conn, server, ioloop=None):
# Initialize the FTPHandler and add custom commands
super().__init__(conn, server, ioloop)
self.proto_cmds = FTPHandler.proto_cmds.copy()
# Add custom FTP commands for managing virtual users - command in lowercase
self.proto_cmds.update(
{'SITE ADDU': dict(perm='M', auth=True, arg=True,
help='Syntax: SITE <SP> ADDU USERNAME PASSWORD (add virtual user).')}
)
self.proto_cmds.update(
{'SITE DELU': dict(perm='M', auth=True, arg=True,
help='Syntax: SITE <SP> DELU USERNAME (remove virtual user).')}
)
self.proto_cmds.update(
{'SITE LSTU': dict(perm='M', auth=True, arg=None,
help='Syntax: SITE <SP> LSTU (list virtual users).')}
)
def on_file_received(self, file):
if not os.stat(file).st_size:
os.remove(file)
logging.info(f'File {file} was empty: removed.')
else:
cfg = self.cfg
path, filenameExt = os.path.split(file)
filename, fileExtension = os.path.splitext(filenameExt)
if (fileExtension.upper() in (cfg.fileext)):
with open(file, 'r') as csvfile:
lines = csvfile.readlines()
conn = psycopg2.connect(
dbname=cfg.dbname,
user=cfg.dbuser,
password=cfg.dbpass,
host=cfg.dbhost,
port=cfg.dbport
)
# Crea un cursore
print(file, lines)
cur = conn.cursor()
try:
cur.execute("INSERT INTO received (filename, content) VALUES (%s,%s)" , (filename, lines))
conn.commit()
conn.close()
except:
logging.error(f'File {file} not loaded. Held in user path.')
else:
os.remove(file)
logging.info(f'File {file} loaded: removed.')
def on_incomplete_file_received(self, file):
# Remove partially uploaded files
os.remove(file)
def ftp_SITE_ADDU(self, line):
"""
Add a virtual user and save the virtuser configuration file.
Create a directory for the virtual user in the specified virtpath.
"""
cfg = self.cfg
try:
parms = line.split()
user = os.path.basename(parms[0]) # Extract the username
password = parms[1] # Get the password
hash = md5(password.encode("UTF-8")).hexdigest() # Hash the password
except:
self.respond('501 SITE ADDU failed. Command needs 2 arguments')
else:
try:
# Create the user's directory
Path(cfg.virtpath + user).mkdir(parents=True, exist_ok=True)
except:
self.respond('551 Error in create virtual user path.')
else:
try:
# Add the user to the authorizer
self.authorizer.add_user(str(user),
hash, cfg.virtpath + "/" + user, perm="lmw")
# Save the user to the SQLite database
# Definisci la connessione al database
conn = psycopg2.connect(
dbname=cfg.dbname,
user=cfg.dbuser,
password=cfg.dbpass,
host=cfg.dbhost,
port=cfg.dbport
)
# Crea un cursore
cur = conn.cursor()
cur.execute("INSERT INTO virtusers (ftpuser, hash, virtpath, perm) VALUES (%s,%s,%s,%s)" , (user, hash, cfg.virtpath + user, 'elmw'))
conn.commit()
conn.close()
logging.info("User {} created.".format(user))
self.respond('200 SITE ADDU successful.')
except:
self.respond('501 SITE ADDU failed.')
def ftp_SITE_DELU(self, line):
"""
Remove a virtual user and save the virtuser configuration file.
"""
cfg = self.cfg
parms = line.split()
user = os.path.basename(parms[0]) # Extract the username
try:
# Remove the user from the authorizer
self.authorizer.remove_user(str(user))
# Delete the user from the SQLite database
conn = psycopg2.connect(
dbname=cfg.dbname,
user=cfg.dbuser,
password=cfg.dbpass,
host=cfg.dbhost,
port=cfg.dbport
)
# Crea un cursore
cur = conn.cursor()
cur.execute("DELETE FROM virtusers WHERE ftpuser = %s", (user, ))
conn.commit()
conn.close()
logging.info("User {} deleted.".format(user))
self.respond('200 SITE DELU successful.')
except:
self.respond('501 SITE DELU failed.')
def ftp_SITE_LSTU(self, line):
"""
List all virtual users.
"""
cfg = self.cfg
users_list = []
try:
# Connect to the SQLite database to fetch users
conn = psycopg2.connect(
dbname=cfg.dbname,
user=cfg.dbuser,
password=cfg.dbpass,
host=cfg.dbhost,
port=cfg.dbport
)
# Crea un cursore
cur = conn.cursor()
self.push("214-The following virtual users are defined:\r\n")
cur.execute("SELECT ftpuser, perm FROM virtusers")
[users_list.append(f'Username: {ftpuser}\tPerms: {perm}\r\n') for ftpuser, perm in cur.fetchall()]
self.push(''.join(users_list))
self.respond("214 LSTU SITE command successful.")
except:
self.respond('501 list users failed.')
def main():
# Load the configuration settings
cfg = setting.config()
try:
# Initialize the authorizer and handler
authorizer = DummyMD5Authorizer(cfg)
handler = ASEHandler
handler.cfg = cfg
handler.authorizer = authorizer
handler.masquerade_address = cfg.proxyaddr
# Set the range of passive ports for the FTP server
_range = list(range(cfg.firstport, cfg.firstport + cfg.portrangewidth))
handler.passive_ports = _range
# Configure logging
logging.basicConfig(
format="%(asctime)s %(message)s",
filename=cfg.logfilename,
level=logging.INFO,
)
# Create and start the FTP server
server = FTPServer(("0.0.0.0", 2121), handler)
server.serve_forever()
except KeyboardInterrupt:
logging.info(
"Info: {}.".format("Shutdown requested...exiting")
)
except Exception:
print(
"{} - PID {:>5} >> Error: {}.".format(
ts.timestamp("log"), os.getpid(), sys.exc_info()[1]
)
)
if __name__ == "__main__":
main()