From 21493d2fa3f5dfb71495dc94e753e8c06b83d52c Mon Sep 17 00:00:00 2001 From: alex Date: Sat, 16 Nov 2024 16:57:30 +0100 Subject: [PATCH] add doc --- ftpReceiver/FtpCsvReceiver.py | 157 ++++++++++------------------------ 1 file changed, 46 insertions(+), 111 deletions(-) diff --git a/ftpReceiver/FtpCsvReceiver.py b/ftpReceiver/FtpCsvReceiver.py index 7e9d76c..39f1e72 100755 --- a/ftpReceiver/FtpCsvReceiver.py +++ b/ftpReceiver/FtpCsvReceiver.py @@ -4,7 +4,7 @@ import sys import os import shutil # import ssl -import pika + import re import logging import sqlite3 @@ -13,9 +13,6 @@ from hashlib import md5 from pathlib import Path from datetime import datetime -from smtplib import SMTP_SSL as SMTP, SMTPException, SMTPAuthenticationError -from email.mime.text import MIMEText - from utils.time import timestamp_fmt as ts from utils.time import date_refmt as df from utils.config import set_config as setting @@ -24,92 +21,30 @@ from pyftpdlib.handlers import FTPHandler, TLS_FTPHandler from pyftpdlib.servers import FTPServer from pyftpdlib.authorizers import DummyAuthorizer, AuthenticationFailed - -def send_mail(sev, msg, cfg): - msg = MIMEText(cfg.message + "\n" + msg) - msg["Subject"] = cfg.subject + " " + sev - msg["From"] = cfg.sender - msg["To"] = cfg.receivers - conn = SMTP( - host=cfg.smtphost, - port=cfg.smtpport, - local_hostname=None, - timeout=5, - source_address=None, - ) - conn.set_debuglevel(cfg.debuglevel) - try: - conn.login(cfg.sender, cfg.password) - conn.sendmail(cfg.sender, cfg.receivers, msg.as_string()) - except SMTPAuthenticationError: - logging.error( - "Mail failed: {}.".format("SMTP authentication error") - ) - except: - logging.info( - "Mail failed: {}.".format("CUSTOM_ERROR") - ) - finally: - conn.quit() - - -class mq: - def __init__(self, cfg): - parameters = pika.URLParameters( - "amqp://" - + cfg.mquser - + ":" - + cfg.mqpass - + "@" - + cfg.mqhost - + ":" - + cfg.mqport - + "/%2F" - ) - connection = pika.BlockingConnection(parameters) - self.channel = connection.channel() - self.channel.queue_declare(queue=cfg.csv_queue, durable=True) - - def write(self, msg, cfg): - try: - props = pika.BasicProperties( - delivery_mode=2, - content_encoding='utf-8', - timestamp=msg["timestamp"],) - self.channel.basic_publish( - exchange="", - routing_key=cfg.csv_queue, - body=msg["payload"], - properties=props - ) - logging.info( - "Write message {} in queue".format(msg)) - except: - logging.error( - "Error write message {} in queue".format(msg)) - - def close(self): - self.channel.close() - - 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]) + # Connect to the SQLite database for virtual users con = sqlite3.connect(cfg.virtusersdb) cur = con.cursor() + # Create the virtusers table if it doesn't exist cur.execute( '''CREATE TABLE IF NOT EXISTS virtusers (user text, hash text, virtpath text, perm text)''') + # Create an index on the user column for faster lookups cur.execute( '''CREATE INDEX IF NOT EXISTS user_idx on virtusers(user)''') + # Load existing virtual users from the database for row in cur.execute('SELECT * FROM virtusers'): self.add_user(row[0], row[1], row[2], perm=row[3]) con.close() 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: @@ -121,8 +56,10 @@ class DummyMD5Authorizer(DummyAuthorizer): 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 ADDU USERNAME PASSWORD (add virtual user).')} @@ -137,6 +74,7 @@ class ASEHandler(FTPHandler): ) def on_file_received(self, file): + # Handle the event when a file is received unitType = "" unitName = "" toolName = "" @@ -144,6 +82,7 @@ class ASEHandler(FTPHandler): fileDate = "" fileTime = "" queue = "" + # Check if the file is empty and remove it if so if not os.stat(file).st_size: os.remove(file) logging.info( @@ -152,7 +91,9 @@ class ASEHandler(FTPHandler): cfg = self.cfg path, filenameExt = os.path.split(file) filename, fileExtension = os.path.splitext(filenameExt) + # Process the file if it has a valid extension if (fileExtension.upper() in (cfg.fileext)): + # Match the filename against a specific pattern if m := re.match( r"^(G\d\d\d|GFLOW)_(ID\d\d\d\d)_(DT\d\d\d\d)_(\d\d)(\d\d)(\d\d\d\d|\d\d)(\d\d)(\d\d)(\d\d)$", filename, @@ -165,12 +106,15 @@ class ASEHandler(FTPHandler): fileDate = m.group(6) + "/" + m.group(5) + "/" + m.group(4) fileTime = m.group(7) + ":" + m.group(8) + ":" + m.group(9) + # Match against another pattern for different file formats elif re.match( r"^(\d\d_\d\d\d\d|)(DT\d\d\d\d|LOC\d\d\d\d|GD\d\d\d\d)$", filename, re.I ): with open(file, "r") as fileCsv: try: + # Read the CSV file line by line for i, line in enumerate(fileCsv.readlines(4096), 1): + # Extract creation date and time if m1 := re.match( r"^(File Creation Date:\s)?(\d*\/\d*\/\d*)\s(\d*:\d*:\d*)\;*\n?$", line, @@ -179,6 +123,7 @@ class ASEHandler(FTPHandler): fileDate = m1.group(2) fileTime = m1.group(3) + # Extract unit type and name elif m2 := re.match( r"^(\w+\d+)\s(\w+\d+)\;*\n?$", line, @@ -187,6 +132,7 @@ class ASEHandler(FTPHandler): unitType = m2.group(1).upper() unitName = m2.group(2).upper() + # Extract tool type and name from the path elif m3 := re.match( r"^SD path: a:\/\w+\/(\w+)(?:\.\w+)?\/*(\w*)(?:\.\w+)?\;*\n?$", line, @@ -207,6 +153,7 @@ class ASEHandler(FTPHandler): "Error: {}.".format(sys.exc_info()[1])) fileCsv.close + # Log the extracted information logging.info( "{} - {} - {} - {} - {} {}.".format( unitType, @@ -217,6 +164,7 @@ class ASEHandler(FTPHandler): fileTime, ) ) + # Prepare the new path for the received file newPath = cfg.csvfs + "/" + self.username + "/received/" + \ unitName.upper() + "/" newFilename = ( @@ -224,70 +172,50 @@ class ASEHandler(FTPHandler): str(ts.timestamp("tms") + fileExtension) ) fileRenamed = file + "_" + str(ts.timestamp("tms")) + # Rename the original file to avoid conflicts os.rename(file, fileRenamed) try: + # Create the new directory for the user if it doesn't exist os.makedirs(newPath) logging.info("Path {} created.".format(newPath)) except FileExistsError: logging.info("Path {} already exists.".format(newPath)) try: + # Move the renamed file to the new location shutil.move(fileRenamed, newFilename) logging.info("{} moved into {}.".format( filenameExt, newFilename)) except OSError: logging.error("Error to move {} into {}.".format( filenameExt, newFilename)) - send_mail( - "Error", "OS error move " + filenameExt + " to " + newFilename, cfg - ) now = datetime.now() - mq_message = {"payload": "{};{};{};{};{};{};{}".format( - unitType, - unitName, - toolName, - toolType, - df.dateFmt(fileDate), - fileTime, - newFilename), - "timestamp": int(datetime.timestamp(now)*1000000) - } - try: - queue = mq(cfg) - queue.write(mq_message, cfg) - logging.info("Queue message: {}.".format(mq_message)) - except: - logging.error( - "Error to put message in queue: {}.".format(mq_message)) - send_mail( - "Error", "Error to put message " + mq_message + " in queue.", cfg - ) - finally: - queue.close() - def on_incomplete_file_received(self, file): - # remove partially uploaded files + # Remove partially uploaded files os.remove(file) def ftp_SITE_ADDU(self, line): """ - add virtual user and save virtuser cfg file - create virtuser dir in virtpath cfg path + Add a virtual user and save the virtuser configuration file. + Create a directory for the virtual user in the specified virtpath. """ cfg = self.cfg parms = line.split() - user = os.path.basename(parms[0]) - password = parms[1] - hash = md5(password.encode("UTF-8")).hexdigest() + 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 try: + # Create the user's directory Path(cfg.virtpath + user).mkdir(parents=True, exist_ok=True) except: - self.responde('551 Error in create virtual user path.') + 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 con = sqlite3.connect(cfg.virtusersdb) cur = con.cursor() cur.execute("INSERT INTO virtusers VALUES (?,?,?,?)", @@ -301,20 +229,21 @@ class ASEHandler(FTPHandler): def ftp_SITE_DELU(self, line): """ - remove virtual user and save virtuser cfg file + Remove a virtual user and save the virtuser configuration file. """ cfg = self.cfg parms = line.split() - user = os.path.basename(parms[0]) + 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 con = sqlite3.connect(cfg.virtusersdb) cur = con.cursor() cur.execute("DELETE FROM virtusers WHERE user = ?", (user,)) con.commit() con.close() logging.info("User {} deleted.".format(user)) - # self.push(' The user path has not been removed!\r\n') self.respond('200 SITE DELU successful.') except: @@ -322,14 +251,16 @@ class ASEHandler(FTPHandler): def ftp_SITE_LSTU(self, line): """ - list virtual user + List all virtual users. """ cfg = self.cfg users_list = [] try: + # Connect to the SQLite database to fetch users con = sqlite3.connect(cfg.virtusersdb) cur = con.cursor() self.push("214-The following virtual users are defined:\r\n") + # Fetch and list all virtual users for row in cur.execute("SELECT * FROM virtusers").fetchall(): users_list.append( " Username: " + row[0] + "\tPerms: " + row[3] + "\r\n") @@ -342,30 +273,34 @@ class ASEHandler(FTPHandler): 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") - )F + ) except Exception: print( "{} - PID {:>5} >> Error: {}.".format(