技術

pyftpdlibでカスタムエラーレスポンスを返すFTPサーバー構築をしてみる

はじめに

Pythonのpyftpdlibライブラリを使用して、カスタムエラーレスポンスを返すことができるFTPサーバーの構築方法を紹介する。

FTPクライアントの開発や検証を行っている際、サーバー側からのエラーレスポンスをシミュレートしたい場合がある。
特に、ファイル転送完了時に通常の成功コード（226）ではなく、エラーコード（4XX系など）が返ってきた場合のクライアント側の挙動（リトライ処理など）を確認したい場合に、意図的にエラーを返すサーバーがあると便利である。

今回は、Pythonのpyftpdlibを使用して、そのようなテスト用FTPサーバーを構築してみたので美貌としてまとめておく。

環境

Windows 11 Professional
WSL2 Ubuntu 24.04 LTS
Python 3.12

インストール

pip install pyftpdlib

最小構成のFTPサーバー

まずは、基本的なFTPサーバーを構築して動作確認を行う。
以下のコードは、指定したディレクトリをルートとしてFTPサーバーを起動する最小構成の例である。

simple_ftp_server.py
import os
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler
from pyftpdlib.servers import FTPServer

# FTPデータディレクトリの作成
ftp_directory = "./ftp_data"
if not os.path.exists(ftp_directory):
    os.makedirs(ftp_directory)
    print(f"FTPディレクトリを作成しました: {ftp_directory}")

# 認証設定
authorizer = DummyAuthorizer()
authorizer.add_user("testuser", "testpass", ftp_directory, perm="elradfmw")

# ハンドラー設定
handler = FTPHandler
handler.authorizer = authorizer
handler.permit_foreign_addresses = True
handler.masquerade_address = '127.0.0.1'
handler.passive_ports = range(60000, 65535)

# サーバー起動
server = FTPServer(("127.0.0.1", 2121), handler)
print("FTPサーバー起動: ftp://localhost:2121")
print(f"ユーザー名: testuser")
print(f"パスワード: testpass")
print(f"データディレクトリ: {os.path.abspath(ftp_directory)}")
server.serve_forever()

これを実行し、FTPクライアントから接続できることを確認する。

サーバ側プログラム起動

python3 simple_ftp_server.py
ftp localhost 2121
# ユーザー名: testuser
# パスワード: testpass

サーバ側ではログインのログが出力されている。

ftp localhost 2121

Connected to localhost.
220 pyftpdlib 2.1.0 ready.
Name (localhost:kbushi): testuser
331 Username ok, send password.
Password:
230 Login successful.
Remote system type is UNIX.
Using binary mode to transfer files.

[I 2025-11-30 20:08:34] 127.0.0.1:46110-[] FTP session opened (connect)
[I 2025-11-30 20:08:34] 127.0.0.1:46110-[testuser] USER 'testuser' logged in.

カスタムエラーレスポンスの実装

次に、ファイル転送完了時に任意のステータスコードを返せるようにサーバーを拡張する。
pyftpdlibのFTPHandlerとDTPHandler（データ転送プロトコルハンドラー）を継承し、handle_closeメソッドをオーバーライドすることで実現する。

実装コード

simple_ftp_server.py
import os
import logging
from pyftpdlib.authorizers import DummyAuthorizer
from pyftpdlib.handlers import FTPHandler, DTPHandler
from pyftpdlib.servers import FTPServer

# ログ設定
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

# FTPデータディレクトリの作成
ftp_directory = "./ftp_data"
if not os.path.exists(ftp_directory):
    os.makedirs(ftp_directory)
    print(f"FTPディレクトリを作成しました: {ftp_directory}")

# 認証設定
authorizer = DummyAuthorizer()
authorizer.add_user("testuser", "testpass", ftp_directory, perm="elradfmw")


# カスタムDTPハンドラー（データ転送プロトコル）
class CustomDTPHandler(DTPHandler):
    """226レスポンスをカスタマイズできるDTPハンドラー"""
    
    def handle_close(self):
        """データコネクションクローズ時の処理（226レスポンスを送信するタイミング）"""
        # 親ハンドラーのカスタムエラー設定を確認
        if hasattr(self.cmd_channel, 'custom_errors'):
            error_config = self.cmd_channel.custom_errors.get('FILE_RECEIVED')
            if error_config and self.receive:  # STORコマンドの場合
                code = error_config.get('code', 451)
                message = error_config.get('message', 'Transfer aborted: local error in processing.')
                self.cmd_channel.respond(f"{code} {message}")
                logging.info(f"カスタムレスポンス{code}を返しました: DTP handle_close")
                # データチャンネルをクリーンアップ
                DTPHandler.handle_close(self)
                return
            
            error_config_sent = self.cmd_channel.custom_errors.get('FILE_SENT')
            if error_config_sent and not self.receive:  # RETRコマンドの場合
                code = error_config_sent.get('code', 426)
                message = error_config_sent.get('message', 'Connection closed; transfer aborted.')
                self.cmd_channel.respond(f"{code} {message}")
                logging.info(f"カスタムレスポンス{code}を返しました: DTP handle_close (RETR)")
                # データチャンネルをクリーンアップ
                DTPHandler.handle_close(self)
                return
        
        # 通常の処理（226 Transfer complete）
        super().handle_close()


# カスタムFTPハンドラー
class CustomFTPHandler(FTPHandler):
    """カスタムエラーレスポンスを返せるFTPハンドラー"""
    
    # カスタムDTPハンドラーを使用
    dtp_handler = CustomDTPHandler
    
    # カスタムエラー設定（必要に応じて変更）
    # None: 通常処理, {'code': XXX, 'message': 'エラーメッセージ'}: カスタムエラーを返す
    custom_errors = {
        # ファイル受信完了時のレスポンス（226の代わりに返すエラーコード）
        # 425: Can't open data connection
        # 426: Connection closed; transfer aborted
        # 450: Requested file action not taken
        # 451: Requested action aborted: local error in processing
        # 452: Requested action not taken: insufficient storage
        'FILE_RECEIVED': {'code': 426, 'message': 'Connection closed; transfer aborted.'},
        'FILE_SENT': None,  # ファイル送信完了時（RETRコマンド後）
    }


# ハンドラー設定
handler = CustomFTPHandler
handler.authorizer = authorizer
handler.permit_foreign_addresses = True
handler.masquerade_address = '127.0.0.1'
handler.passive_ports = range(60000, 65535)

# サーバー起動
server = FTPServer(("127.0.0.1", 2121), handler)
print("FTPサーバー起動: ftp://localhost:2121")
print(f"ユーザー名: testuser")
print(f"パスワード: testpass")
print(f"データディレクトリ: {os.path.abspath(ftp_directory)}")
server.serve_forever()

仕組み

  1. CustomDTPHandler
    データ転送終了時（handle_close）に、親ハンドラー（CustomFTPHandler）に設定されたcustom_errorsを確認する。

  2. エラーレスポンス送信
    設定があれば、通常の226 Transfer completeの代わりに指定されたエラーコード（例: 426）を返す。

  3. CustomFTPHandler
    dtp_handlerとしてCustomDTPHandlerを指定し、custom_errors辞書で返すエラーコードを定義する。

動作確認

サーバーを起動し、FTPクライアントからファイルをアップロードしてみる。

python simple_ftp_server.py
ftp localhost 2121
put test.txt

サーバー側のログには以下のように出力され、クライアントには426エラーが返されている。
ただし、サーバー側ではファイルは正常に保存されている（データ転送自体は完了しているため）。

ftp> put hash.txt
local: hash.txt remote: hash.txt
229 Entering extended passive mode (|||61999|).
125 Data connection already open. Transfer starting.
100% |***************************************************************|    33      125.88 KiB/s    00:00 ETA
426 Connection closed; transfer aborted.
33 bytes sent in 00:00 (83.92 KiB/s)

これにより、クライアント側で「転送はできたがサーバーからエラーが返ってきた」という状況を再現でき、例外処理やリトライロジックのテストが可能となっている。

おわりに

pyftpdlibは拡張性が高く、今回のようにハンドラーをオーバーライドすることで柔軟な挙動を実現できる。
異常系のテスト環境構築に非常に役立つツールだ。

© 2020 - 2026 kbushi
Hugo で構築されています。
テーマ Stack は Jimmy によって設計されています。