Very fast asynchronous FTP server library providing RFC-959 compliant FTP servers with advanced features including FTPS, IPv6, Unicode support, and flexible authentication systems
—
Cross-platform filesystem abstraction providing virtualized access to local and remote filesystems with security controls. pyftpdlib's filesystem interface enables secure, chrooted access to directories while maintaining compatibility across different operating systems and filesystem types.
Main filesystem interface providing cross-platform file operations with virtual chroot functionality and security controls.
class AbstractedFS:
def __init__(self, root, cmd_channel):
"""
Initialize filesystem interface.
Parameters:
- root: root directory path (user's home directory)
- cmd_channel: associated FTPHandler instance
"""
# Properties
@property
def root(self):
"""Get root directory path."""
@property
def cwd(self):
"""Get current working directory (relative to root)."""
# Path manipulation methods
def ftpnorm(self, ftppath):
"""
Normalize FTP path by resolving . and .. components.
Parameters:
- ftppath: FTP path string
Returns:
- Normalized FTP path
"""
def ftp2fs(self, ftppath):
"""
Convert FTP path to filesystem path.
Parameters:
- ftppath: FTP path (relative to user root)
Returns:
- Absolute filesystem path
"""
def fs2ftp(self, fspath):
"""
Convert filesystem path to FTP path.
Parameters:
- fspath: absolute filesystem path
Returns:
- FTP path relative to user root
"""
def validpath(self, path):
"""
Check if path is within user's accessible area.
Parameters:
- path: filesystem path to validate
Returns:
- True if path is valid and accessible
"""
# File operations
def open(self, filename, mode):
"""
Open file for reading or writing.
Parameters:
- filename: file path
- mode: file open mode ('r', 'w', 'a', 'rb', 'wb', etc.)
Returns:
- File object
"""
def mkstemp(self, suffix='', prefix='', dir=None, mode='wb'):
"""
Create temporary file.
Parameters:
- suffix: filename suffix
- prefix: filename prefix
- dir: directory for temp file (None = current directory)
- mode: file open mode
Returns:
- (fd, path) tuple of file descriptor and path
"""
def remove(self, path):
"""
Delete file.
Parameters:
- path: file path to delete
"""
def rename(self, src, dst):
"""
Rename/move file or directory.
Parameters:
- src: source path
- dst: destination path
"""
def chmod(self, path, mode):
"""
Change file/directory permissions.
Parameters:
- path: file/directory path
- mode: permission mode (octal integer)
"""
def stat(self, path):
"""
Get file/directory status information.
Parameters:
- path: file/directory path
Returns:
- os.stat_result object
"""
def lstat(self, path):
"""
Get file/directory status (don't follow symlinks).
Parameters:
- path: file/directory path
Returns:
- os.stat_result object
"""
def utime(self, path, timeval):
"""
Set file access and modification times.
Parameters:
- path: file path
- timeval: (atime, mtime) tuple or None for current time
"""
def readlink(self, path):
"""
Read symbolic link target.
Parameters:
- path: symlink path
Returns:
- Target path string
"""
# Directory operations
def chdir(self, path):
"""
Change current working directory.
Parameters:
- path: directory path (relative to root)
"""
def mkdir(self, path):
"""
Create directory.
Parameters:
- path: directory path to create
"""
def rmdir(self, path):
"""
Remove empty directory.
Parameters:
- path: directory path to remove
"""
def listdir(self, path):
"""
List directory contents.
Parameters:
- path: directory path
Returns:
- List of filenames
"""
def listdirinfo(self, path):
"""
List directory with file information.
Parameters:
- path: directory path
Returns:
- List of (filename, stat_result) tuples
"""
# File system queries
def isfile(self, path):
"""Check if path is a regular file."""
def isdir(self, path):
"""Check if path is a directory."""
def islink(self, path):
"""Check if path is a symbolic link."""
def getsize(self, path):
"""Get file size in bytes."""
def getmtime(self, path):
"""Get file modification time as timestamp."""
def realpath(self, path):
"""Get canonical path resolving all symlinks."""
def lexists(self, path):
"""Check if path exists (don't follow symlinks)."""
# User/group resolution
def get_user_by_uid(self, uid):
"""
Get username from UID.
Parameters:
- uid: user ID number
Returns:
- Username string or UID if not found
"""
def get_group_by_gid(self, gid):
"""
Get group name from GID.
Parameters:
- gid: group ID number
Returns:
- Group name string or GID if not found
"""
# Directory listing formatters
def format_list(self, basedir, listing, ignore_err=True):
"""
Format directory listing in Unix ls -l style.
Parameters:
- basedir: directory being listed
- listing: list of (filename, stat_result) tuples
- ignore_err: ignore files that cause stat errors
Returns:
- Iterator of formatted listing lines
"""
def format_mlsx(self, basedir, listing, perms, facts, ignore_err=True):
"""
Format directory listing in machine-readable MLSD format.
Parameters:
- basedir: directory being listed
- listing: list of (filename, stat_result) tuples
- perms: permission string for current user
- facts: requested fact names
- ignore_err: ignore files that cause stat errors
Returns:
- Iterator of formatted MLSD lines
"""Direct filesystem access without chroot restrictions, suitable for system-level FTP services on POSIX systems.
class UnixFilesystem(AbstractedFS): # POSIX only
def __init__(self, root, cmd_channel):
"""
Initialize Unix filesystem with direct access.
Parameters:
- root: root directory (not enforced as chroot)
- cmd_channel: associated FTPHandler instance
"""
def ftp2fs(self, ftppath):
"""
Convert FTP path directly to filesystem path.
No chroot restrictions applied.
"""
def fs2ftp(self, fspath):
"""Convert filesystem path directly to FTP path."""
def validpath(self, path):
"""Always returns True - no path restrictions."""class FilesystemError(Exception):
"""Custom exception for filesystem-related errors."""def _memoize(fun):
"""
Memoization decorator for expensive operations like user/group lookups.
Caches function results to improve performance.
"""from pyftpdlib.filesystems import AbstractedFS
# Create filesystem rooted at user's home directory
fs = AbstractedFS("/home/user", cmd_channel)
# Path operations
print(fs.cwd) # Current directory relative to root
fs.chdir("documents") # Change to documents subdirectory
real_path = fs.ftp2fs("file.txt") # Convert to actual filesystem path
# File operations
with fs.open("readme.txt", "r") as f:
content = f.read()
fs.mkdir("new_folder")
fs.remove("old_file.txt")
fs.rename("old_name.txt", "new_name.txt")# Get directory contents
files = fs.listdir(".")
print(files) # ['file1.txt', 'file2.txt', 'subfolder']
# Get detailed file information
detailed = fs.listdirinfo(".")
for filename, stat_info in detailed:
print(f"{filename}: {stat_info.st_size} bytes")
# Format as Unix ls -l listing
listing = fs.format_list(".", detailed)
for line in listing:
print(line)
# Output: -rw-r--r-- 1 user group 1024 Jan 01 12:00 file1.txt# Check file types
if fs.isfile("document.pdf"):
size = fs.getsize("document.pdf")
mtime = fs.getmtime("document.pdf")
print(f"File size: {size} bytes, modified: {mtime}")
if fs.isdir("folder"):
print("It's a directory")
if fs.islink("symlink"):
target = fs.readlink("symlink")
print(f"Symlink points to: {target}")class RestrictedFS(AbstractedFS):
def validpath(self, path):
"""Block access to hidden files and system directories."""
if not super().validpath(path):
return False
# Block hidden files
if "/.'" in path or path.startswith("."):
return False
# Block system directories
restricted = ["/etc", "/proc", "/sys", "/dev"]
for restricted_path in restricted:
if path.startswith(restricted_path):
return False
return True
def listdir(self, path):
"""Filter out hidden files from listings."""
files = super().listdir(path)
return [f for f in files if not f.startswith(".")]
# Use custom filesystem
class CustomFTPHandler(FTPHandler):
abstracted_fs = RestrictedFS
handler = CustomFTPHandler
handler.authorizer = authorizerfrom pyftpdlib.filesystems import UnixFilesystem
# Create filesystem with direct access (no chroot)
class SystemFTPHandler(FTPHandler):
abstracted_fs = UnixFilesystem
# WARNING: This allows access to entire filesystem
# Only use with trusted users and proper authorizer restrictions
handler = SystemFTPHandler
handler.authorizer = unix_authorizer # Should use UnixAuthorizerfrom pyftpdlib.handlers import FTPHandler
class CustomFTPHandler(FTPHandler):
# Use custom filesystem
abstracted_fs = RestrictedFS
def run_as_current_user(self, function, *args, **kwargs):
"""Execute filesystem operations with proper user context."""
return function(*args, **kwargs)
# Configure handler
handler = CustomFTPHandler
handler.authorizer = authorizer# Set file permissions (Unix/Linux)
fs.chmod("script.sh", 0o755) # rwxr-xr-x
# Set file modification time
import time
current_time = time.time()
fs.utime("file.txt", (current_time, current_time))
# User/group lookup
uid = 1000
username = fs.get_user_by_uid(uid)
print(f"UID {uid} belongs to user: {username}")validpath() before file operationsInstall with Tessl CLI
npx tessl i tessl/pypi-pyftpdlib