132 lines
4.4 KiB
Python
132 lines
4.4 KiB
Python
"""
|
|
FTP server module for FactsDB service
|
|
Provides secure FTP access with directory restrictions
|
|
"""
|
|
|
|
import os
|
|
import threading
|
|
from pyftpdlib.authorizers import DummyAuthorizer
|
|
from pyftpdlib.handlers import FTPHandler
|
|
from pyftpdlib.servers import FTPServer
|
|
from typing import List, Dict, Any
|
|
from .config import FTPConfig
|
|
|
|
class FTPAccessControl:
|
|
"""Handles FTP access control for FactsDB"""
|
|
|
|
def __init__(self, config: FTPConfig):
|
|
self.config = config
|
|
self.allowed_directories = set()
|
|
self._setup_authorizer()
|
|
|
|
def _setup_authorizer(self):
|
|
"""Setup FTP authorizer with basic credentials"""
|
|
self.authorizer = DummyAuthorizer()
|
|
self.authorizer.add_user(
|
|
self.config.username,
|
|
self.config.password,
|
|
self.config.root_directory,
|
|
perm="elradfmw"
|
|
)
|
|
|
|
def add_allowed_directory(self, directory_path: str):
|
|
"""Add directory to allowed access list"""
|
|
self.allowed_directories.add(os.path.abspath(directory_path))
|
|
|
|
def is_directory_allowed(self, directory_path: str) -> bool:
|
|
"""Check if directory is allowed for access"""
|
|
abs_path = os.path.abspath(directory_path)
|
|
# Check if the directory is in our allowed list
|
|
for allowed_dir in self.allowed_directories:
|
|
if abs_path.startswith(allowed_dir):
|
|
return True
|
|
return False
|
|
|
|
def get_allowed_directories(self) -> List[str]:
|
|
"""Get list of all allowed directories"""
|
|
return list(self.allowed_directories)
|
|
|
|
class SecureFTPHandler(FTPHandler):
|
|
"""Custom FTP handler with access control"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
# This will be set by the FTPServerManager
|
|
self.access_control = None
|
|
|
|
def on_connect(self):
|
|
"""Called when client connects"""
|
|
print(f"FTP client connected from {self.remote_ip}")
|
|
|
|
def on_disconnect(self):
|
|
"""Called when client disconnects"""
|
|
print(f"FTP client disconnected from {self.remote_ip}")
|
|
|
|
def on_file_received(self, file_path):
|
|
"""Called when file is received"""
|
|
print(f"File received: {file_path}")
|
|
|
|
def on_file_sent(self, file_path):
|
|
"""Called when file is sent"""
|
|
print(f"File sent: {file_path}")
|
|
|
|
class FTPServerManager:
|
|
"""Manages FTP server for FactsDB"""
|
|
|
|
def __init__(self, config: FTPConfig):
|
|
self.config = config
|
|
self.access_control = FTPAccessControl(config)
|
|
self.server = None
|
|
self.is_running = False
|
|
|
|
# Create root directory if it doesn't exist
|
|
os.makedirs(self.config.root_directory, exist_ok=True)
|
|
|
|
def start_server(self):
|
|
"""Start the FTP server"""
|
|
if self.is_running:
|
|
return
|
|
|
|
# Setup handler
|
|
handler = SecureFTPHandler
|
|
handler.authorizer = self.access_control.authorizer
|
|
handler.access_control = self.access_control
|
|
|
|
# Setup server
|
|
self.server = FTPServer((self.config.host, self.config.port), handler)
|
|
self.server.max_cons = 256
|
|
self.server.max_cons_per_ip = 5
|
|
|
|
# Start server in a separate thread
|
|
self.server_thread = threading.Thread(target=self._run_server)
|
|
self.server_thread.daemon = True
|
|
self.server_thread.start()
|
|
|
|
self.is_running = True
|
|
print(f"FTP Server started on {self.config.host}:{self.config.port}")
|
|
|
|
def _run_server(self):
|
|
"""Run the FTP server"""
|
|
try:
|
|
self.server.serve_forever()
|
|
except Exception as e:
|
|
print(f"FTP Server error: {e}")
|
|
|
|
def stop_server(self):
|
|
"""Stop the FTP server"""
|
|
if self.server and self.is_running:
|
|
self.server.close_all()
|
|
self.is_running = False
|
|
print("FTP Server stopped")
|
|
|
|
def add_allowed_directory(self, directory_path: str):
|
|
"""Add directory to allowed access list"""
|
|
self.access_control.add_allowed_directory(directory_path)
|
|
|
|
def is_directory_allowed(self, directory_path: str) -> bool:
|
|
"""Check if directory is allowed for access"""
|
|
return self.access_control.is_directory_allowed(directory_path)
|
|
|
|
def get_allowed_directories(self) -> List[str]:
|
|
"""Get list of all allowed directories"""
|
|
return self.access_control.get_allowed_directories() |