Python で強力なポートスキャナーを構築する

はじめに

このプロジェクトでは、Python を使用してターゲットサーバー上のオープンポートを検出するサーバーポートスキャナーを開発します。このツールは、セキュリティポリシーを検証するシステム管理者にとっても、ホスト上で動作しているネットワークサービスを特定しようとする攻撃者にとっても極めて重要です。このプロジェクトでは、ポートスキャンの手法やその影響など、基本的な側面を網羅します。Python の機能を活用して、効率とパフォーマンスを向上させるためのマルチスレッドアプローチに焦点を当て、シンプルながらも効果的なポートスキャナーの作成に深く踏み込みます。

ポートスキャンとは、一連のサーバーポートに対してリクエストを送信し、どのポートが開いているかを確認するプロセスです。このステップは、セキュリティの維持だけでなく、攻撃者が脆弱なサービスを見つけるためにも重要です。まずは、ポートスキャンの基本とその意味について探求することから始めましょう。

主なコンセプト：

ポートスキャナーは、サーバーやホスト上のオープンポートを検出するのに役立ちます。
セキュリティ評価に使用されるほか、攻撃者が脆弱なサービスを特定するためにも使用されます。
最も単純な形式のポートスキャンは、特定の範囲のポートに対して TCP 接続を試行することです。

👀 プレビュー

これから構築するポートスキャナーツールの動作イメージは以下の通りです：

python port_scanner.py 127.0.0.1 5000-9000

出力結果：

Opened Port: 8081
Scanning completed.

🎯 タスク

このプロジェクトでは、以下の内容を学びます：

Python のソケットプログラミング（socket programming）機能を活用してネットワークポートと通信する方法。
ネットワークスキャンタスクの効率を高めるために、Python でマルチスレッド（multi-threaded）アプローチを実装する方法。
柔軟なポートスキャンのためにユーザー入力を受け取るコマンドラインツールを Python で開発する方法。

🏆 達成事項

このプロジェクトを完了すると、以下のことができるようになります：

Python の socket ライブラリを使用して、ネットワーク接続の作成、ポートの可用性テスト、およびネットワーク例外の処理を行う。
Python におけるマルチスレッドを理解して適用し、並行タスクを実行することで、ネットワーク負荷の高い操作のパフォーマンスを大幅に向上させる。
実用的なコマンドラインポートスキャナーツールを構築し、Python スクリプト作成スキルとコマンドライン引数の解析に関する理解を深める。

TCP 接続テストの確立

最初のステップでは、ターゲット IP 上で TCP ポートが開いているかどうかをテストする関数を作成します。この関数は、指定されたポートへの接続を試み、そのステータスを判定します。

/home/labex/project ディレクトリにある port_scanner.py という名前のファイルを開き、以下の Python 関数を追加してください。

import argparse
from queue import Queue
from socket import AF_INET, gethostbyname, socket, SOCK_STREAM
import threading

def tcp_test(port: int, target_ip: str) -> None:
    with socket(AF_INET, SOCK_STREAM) as sock:
        sock.settimeout(1)
        result = sock.connect_ex((target_ip, port))
        if result == 0:
            print(f"Opened Port: {port}")

このコードは、ターゲット IP アドレスの指定されたポートに対して TCP 接続の確立を試みる tcp_test 関数を定義しています。各ポートでの長時間の待機を避けるためにタイムアウトを設定し、connect_ex を使用して接続を試行します。

connect_ex が 0 を返した場合、接続が成功したこと（つまりポートが開いていること）を示し、関数はオープンポートを示すメッセージを表示します。

並行ポートチェックの設定

次に、各スレッドがキューからポートを取り出して処理し、tcp_test 関数を使用して各ポートの状態を確認するための worker 関数を作成します。

port_scanner.py に以下のコードを追加してください。

def worker(target_ip: str, queue: Queue) -> None:
    while not queue.empty():
        port = queue.get()
        tcp_test(port, target_ip)
        queue.task_done()

worker 関数は、複数のスレッドによって実行されるように設計されています。キューからポート番号を継続的に取得し、定義済みの tcp_test 関数を使用して、ターゲット IP 上でそれらのポートが開いているかどうかを確認します。

ポートのテストが終わると、キュー内のタスクを完了としてマークします。この関数により、ポートの並行スキャンが可能になり、プロセスが大幅にスピードアップします。

スキャンプロセスのオーケストレーション

main 関数は、キューのセットアップ、ワーカー（worker）スレッドの初期化、および指定されたポート範囲にわたるスキャン操作の管理を行うことで、スキャンプロセス全体を統括します。

port_scanner.py に main 関数を追加して続行しましょう。

def main(host: str, start_port: int, end_port: int) -> None:
    target_ip = gethostbyname(host)
    queue = Queue()
    for port in range(start_port, end_port + 1):
        queue.put(port)
    for _ in range(100):  ## 必要に応じてスレッド数を調整してください
        t = threading.Thread(target=worker, args=(target_ip, queue,))
        t.daemon = True
        t.start()
    queue.join()
    print("Scanning completed.")

main 関数は、ポートスキャンプロセス全体を制御します。

まず、ターゲットのホスト名を IP アドレスに解決し、スキャンするポート番号用のキューを作成します。指定された範囲内のすべてのポートをキューに入れ、複数のワーカースレッド（この例では 100 個）を生成して、各ポートをチェックすることでキューを並行して処理します。関数はすべてのポートの処理が完了するのを待ち（queue.join()）、スキャンが完了したことを示すメッセージを表示します。

ポートスキャナーの実行

いよいよ、ポートスキャナーを実行する時が来ました。ツールをテストするために、ローカルホスト（localhost）のポート範囲をスキャンします。

スクリプトを実行可能にするために、port_scanner.py の最後に以下の行を追加してください。

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='TCP Port Scanner')
    parser.add_argument('host', help='Host to scan')
    parser.add_argument('ports', help='Port range to scan, formatted as start-end')
    args = parser.parse_args()

    start_port, end_port = map(int, args.ports.split('-'))
    main(args.host, start_port, end_port)

このコードは、ポートスキャナースクリプトにコマンドライン引数の解析機能を追加し、より汎用性が高く使いやすいものにします。

argparse ライブラリを使用して、host と ports という 2 つの必須引数を定義しています。

host 引数は、スキャンしたいターゲットホストのアドレスまたはホスト名を指定するためのものです。
ports 引数は、スキャンするポートの範囲を "開始 - 終了"（例："5000-8000"）の形式で定義する文字列を受け取ります。スクリプトは - 区切り文字を使用してこれを start_port と end_port に分割し、整数に変換します。これにより、ユーザーはコマンドラインからスクリプトを実行する際に、ターゲットとポート範囲を簡単に指定できるようになります。

それでは、ターミナルで次のコマンドを実行してスキャナーを起動しましょう。

python port_scanner.py 127.0.0.1 5000-9000

出力結果：

Opened Port: 8081
Scanning completed.

ポートスキャナーを使用してオープンポート（8081）を特定できました。そこではウェブサーバーが稼働しているため、どのような内容がホストされているかを確認できます。

これを行うには、まず追加ボタンをクリックし、"Web Service" オプションを選択します。

Web Service オプションの選択

次に、ポート番号 8081 を入力し、"Access" ボタンをクリックします。

ポートを入力してウェブサーバーにアクセス

システム内でポート 8081 をリッスンして動作しているウェブサーバーが表示されます。

ポート 8081 で動作中のウェブサーバー

まとめ

このプロジェクトでは、基本的な tcp_test 関数から始まり、マルチスレッドによるスキャンアプローチまで、Python を使用して基本的なポートスキャナーを構築しました。プロセスを管理しやすいステップに分解し、最終的にターゲットサーバー上のオープンポートを特定できる機能的なツールを完成させました。この実践的な経験を通じて、ネットワークセキュリティへの理解を深めただけでなく、Python プログラミングと並行処理管理のスキルも磨くことができました。

Python によるポートスキャナーの構築