alex/ASE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pylint 2

Browse Source
This commit is contained in:
alex
2025-09-03 21:22:35 +02:00
parent 39dba8f54a
commit 54cb20b6af
5 changed files with 139 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
src/ftp_csv_receiver.py
View File

@@ -1,23 +1,28 @@
#!.venv/bin/python
"""This module implements an FTP server with custom commands for managing virtual users and handling CSV file uploads."""
"""
This module implements an FTP server with custom commands for
managing virtual users and handling CSV file uploads.
"""
import os
import logging
from hashlib import sha256
from pathlib import Path
from utils.config import loader_ftp_csv as setting
from utils.database.connection import connetti_db
from utils.connect import user_admin, file_management
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer
from pyftpdlib.authorizers import DummyAuthorizer, AuthenticationFailed
from utils.config import loader_ftp_csv as setting
from utils.database.connection import connetti_db
from utils.connect import user_admin, file_management
# Configure logging (moved inside main function)
logger = logging.getLogger(__name__)
class DummySha256Authorizer(DummyAuthorizer):
"""Custom authorizer that uses SHA256 for password hashing and manages users from a database."""
@@ -29,39 +34,44 @@ class DummySha256Authorizer(DummyAuthorizer):
"""
super().__init__()
self.add_user(
cfg.adminuser[0], cfg.adminuser[1], cfg.adminuser[2], perm=cfg.adminuser[3])
cfg.adminuser[0], cfg.adminuser[1], cfg.adminuser[2], perm=cfg.adminuser[3]
)
# Define the database connection
conn = connetti_db(cfg)
# Create a cursor
cur = conn.cursor()
cur.execute(f'SELECT ftpuser, hash, virtpath, perm FROM {cfg.dbname}.{cfg.dbusertable} WHERE disabled_at IS NULL')
cur.execute(
f"SELECT ftpuser, hash, virtpath, perm FROM {cfg.dbname}.{cfg.dbusertable} WHERE disabled_at IS NULL"
)
for ftpuser, hash, virtpath, perm in cur.fetchall():
self.add_user(ftpuser, hash, virtpath, perm)
"""
Create the user's directory if it does not exist.
"""
for ftpuser, user_hash, virtpath, perm in cur.fetchall():
self.add_user(ftpuser, user_hash, virtpath, perm)
# Create the user's directory if it does not exist.
try:
Path(cfg.virtpath + ftpuser).mkdir(parents=True, exist_ok=True)
except Exception as e:
self.responde(f'551 Error in create virtual user path: {e}')
except Exception as e: # pylint: disable=broad-except
self.responde(f"551 Error in create virtual user path: {e}")
def validate_authentication(self: object, username: str, password: str, handler: object) -> None:
# Validate the user's password against the stored hash
hash = sha256(password.encode("UTF-8")).hexdigest()
def validate_authentication(
self: object, username: str, password: str, handler: object
) -> None:
# Validate the user's password against the stored user_hash
user_hash = sha256(password.encode("UTF-8")).hexdigest()
try:
if self.user_table[username]["pwd"] != hash:
if self.user_table[username]["pwd"] != user_hash:
raise KeyError
except KeyError:
raise AuthenticationFailed
class ASEHandler(FTPHandler):
"""Custom FTP handler that extends FTPHandler with custom commands and file handling."""
def __init__(self: object, conn: object, server: object, ioloop:object=None) -> None:
def __init__(
self: object, conn: object, server: object, ioloop: object = None
) -> None:
"""Initializes the handler, adds custom commands, and sets up command permissions.
Args:
@@ -73,20 +83,44 @@ class ASEHandler(FTPHandler):
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).')}
{
"SITE ADDU": dict(
perm="M",
auth=True,
arg=True,
help="Syntax: SITE <SP> ADDU USERNAME PASSWORD (add virtual user).",
)
}
)
self.proto_cmds.update(
{'SITE DISU': dict(perm='M', auth=True, arg=True,
help='Syntax: SITE <SP> DISU USERNAME (disable virtual user).')}
{
"SITE DISU": dict(
perm="M",
auth=True,
arg=True,
help="Syntax: SITE <SP> DISU USERNAME (disable virtual user).",
)
}
)
self.proto_cmds.update(
{'SITE ENAU': dict(perm='M', auth=True, arg=True,
help='Syntax: SITE <SP> ENAU USERNAME (enable virtual user).')}
{
"SITE ENAU": dict(
perm="M",
auth=True,
arg=True,
help="Syntax: SITE <SP> ENAU USERNAME (enable virtual user).",
)
}
)
self.proto_cmds.update(
{'SITE LSTU': dict(perm='M', auth=True, arg=None,
help='Syntax: SITE <SP> LSTU (list virtual users).')}
{
"SITE LSTU": dict(
perm="M",
auth=True,
arg=None,
help="Syntax: SITE <SP> LSTU (list virtual users).",
)
}
)
def on_file_received(self: object, file: str) -> None:
@@ -111,6 +145,7 @@ class ASEHandler(FTPHandler):
def ftp_SITE_LSTU(self: object, line: str) -> None:
return user_admin.ftp_SITE_LSTU(self, line)
def main():
"""Main function to start the FTP server."""
# Load the configuration settings
@@ -140,9 +175,8 @@ def main():
server.serve_forever()
except Exception as e:
logger.error(
f"Exit with error: {e}."
)
logger.error("Exit with error: %s.", e)
if __name__ == "__main__":
main()
main()