Apache アイドルタイムアウト 検証 (python)

1.目的

・apacheのアイドルタイムアウト値（timeout）を検証する
※timeout値の設定でアイドルセッションが切れてしまうのか？検証

2.環境

OS：Red Hat Enterprise Linux release 8.5 (Ootpa)
Python：Python 3.6.8
apche：2.4.37
ファイル名：apache_timeout_idle_test.py

3.実行方法

python3△apache_timeout_idle_test.py

4.実行結果

(1)timeout内

ApacheサーバーのURLを入力してください (デフォルト: http://192.168.56.132/):
入力がありませんでした。デフォルト値 'http://192.168.56.132/' を使用します。
アイドルタイムアウトをテストする秒数を入力してください (デフォルト: 120): 30

テスト開始時刻: 2024-09-28 22:19:26
最初のリクエストを送信しています: http://192.168.56.132/
接続が成功しました。サーバー応答: 200 OK
最初のリクエストで使用されたポート番号: 44004
30秒間アイドル状態になります...
待機時間: 00:00 残り
再度リクエストを送信して、接続が維持されているか確認します...
Success: 30秒後も接続が維持されています。サーバー応答: 200 OK
再度リクエストで使用されたポート番号: 44004
ポート番号が一致しています。接続は維持されています。
テスト終了時刻: 2024-09-28 22:19:56
テスト成功: 接続はアイドル状態後も維持されました。

(2)timeoutオーバ時

ApacheサーバーのURLを入力してください (デフォルト: http://192.168.56.132/):
入力がありませんでした。デフォルト値 'http://192.168.56.132/' を使用します。
アイドルタイムアウトをテストする秒数を入力してください (デフォルト: 120): 45

テスト開始時刻: 2024-09-28 22:05:07
最初のリクエストを送信しています: http://192.168.56.132/
接続が成功しました。サーバー応答: 200 OK
最初のリクエストで使用されたポート番号: 39294
45秒間アイドル状態になります...
待機時間: 00:00 残り
再度リクエストを送信して、接続が維持されているか確認します...
再度のリクエストでエラーが発生しました: Remote end closed connection without response
再度リクエストで使用されたポート番号: 不明
ポート番号が一致しません。接続が切断された可能性があります。
最初のリクエスト: 39294, 再リクエスト: 不明
テスト終了時刻: 2024-09-28 22:05:52
テスト失敗: 接続が維持されませんでした。

5.プログラム

import time
import http.client
import sys
from urllib.parse import urlparse
from datetime import datetime

def get_local_port_from_socket(conn):
    """
    ソケット接続からローカルポート番号を取得します。
    """
    try:
        return conn.sock.getsockname()[1]
    except AttributeError:
        # ソケットが存在しない場合（接続が閉じられている場合）
        return "不明"

def display_wait_time(idle_duration):
    """
    待機時間をリアルタイムで表示します。

    Args:
        idle_duration (int): アイドル状態の秒数
    """
    for remaining in range(idle_duration, 0, -1):
        minutes, seconds = divmod(remaining, 60)
        time_format = f"{minutes:02d}:{seconds:02d}"
        sys.stdout.write(f"\r待機時間: {time_format} 残り")
        sys.stdout.flush()
        time.sleep(1)
    sys.stdout.write("\r待機時間: 00:00 残り\n")
    sys.stdout.flush()

def check_idle_timeout(url, idle_duration):
    """
    Apacheのアイドルタイムアウトを検証するプログラム

    Args:
        url (str): ApacheサーバーのURL
        idle_duration (int): アイドル状態の秒数

    Returns:
        bool: テストが成功したかどうか
    """
    try:
        # URLを解析し、ホストとポートを取得
        parsed_url = urlparse(url)
        hostname = parsed_url.hostname
        port = parsed_url.port or (443 if parsed_url.scheme == 'https' else 80)

        # 接続を管理するためのHTTPConnectionを作成
        if parsed_url.scheme == 'https':
            conn = http.client.HTTPSConnection(hostname, port)
        else:
            conn = http.client.HTTPConnection(hostname, port)

        # 最初のリクエストを送信して、接続を確立
        print(f"最初のリクエストを送信しています: {url}")
        conn.request("GET", parsed_url.path or "/")
        response = conn.getresponse()

        if response.status == 200:
            print("接続が成功しました。サーバー応答: 200 OK")
        else:
            print(f"接続が成功しましたが、ステータスコード: {response.status}")

        # 使用されたポート番号を取得して表示
        used_port = get_local_port_from_socket(conn)
        print(f"最初のリクエストで使用されたポート番号: {used_port}")

        # 応答ストリームを少しだけ消費する（全データを読み込まない）
        response.read(1024)  # 少しだけデータを読み込んで接続を維持

        # アイドル状態を作る（指定された時間待機）
        print(f"{idle_duration}秒間アイドル状態になります...")
        display_wait_time(idle_duration)

        # 同じ接続を使って再度リクエストを送信して、接続が維持されているか確認
        print("再度リクエストを送信して、接続が維持されているか確認します...")
        try:
            conn.request("GET", parsed_url.path or "/")
            response = conn.getresponse()

            if response.status == 200:
                print(f"Success: {idle_duration}秒後も接続が維持されています。サーバー応答: 200 OK")
            else:
                print(f"Warning: 接続が維持されていますが、ステータスコード: {response.status}")

            # 再度リクエスト時のポート番号を取得して表示
            used_port_after_idle = get_local_port_from_socket(conn)
            print(f"再度リクエストで使用されたポート番号: {used_port_after_idle}")

            # ポート番号の比較
            if used_port == used_port_after_idle:
                print(f"ポート番号が一致しています。接続は維持されています。")
                success = True
            else:
                print(f"Error: ポート番号が一致しません。接続が切断された可能性があります。")
                print(f"最初のリクエスト: {used_port}, 再リクエスト: {used_port_after_idle}")
                success = False  # ポート番号が異なる場合は失敗とみなす

        except Exception as e:
            # 再度リクエストが失敗した場合でもポート番号を表示
            print(f"再度のリクエストでエラーが発生しました: {e}")
            used_port_after_idle = get_local_port_from_socket(conn)
            print(f"再度リクエストで使用されたポート番号: {used_port_after_idle}")
            print(f"ポート番号が一致しません。接続が切断された可能性があります。")
            print(f"最初のリクエスト: {used_port}, 再リクエスト: {used_port_after_idle}")
            success = False

        conn.close()  # 接続を閉じる

    except Exception as e:
        # 例外処理（接続エラーなど）
        print(f"Error: {e}")
        success = False

    return success

def get_input_with_default(prompt, default, is_integer=False):
    """
    ユーザーからの入力を取得し、入力がない場合はデフォルト値を返します。

    Args:
        prompt (str): ユーザーに表示するプロンプトメッセージ
        default (str or int): 入力がない場合に使用するデフォルト値
        is_integer (bool): 入力が整数であるかどうかを指定

    Returns:
        str or int: ユーザーの入力、またはデフォルト値
    """
    user_input = input(prompt).strip()
    if not user_input:
        print(f"入力がありませんでした。デフォルト値 '{default}' を使用します。")
        return default
    if is_integer:
        if user_input.isdigit():
            return int(user_input)
        else:
            print(f"無効な入力 '{user_input}' がありました。デフォルト値 '{default}' を使用します。")
            return default
    return user_input

if __name__ == "__main__":
    # デフォルト値の設定
    DEFAULT_URL = "http://localhost/"
    DEFAULT_TIMEOUT = 120

    # ユーザーからURLを入力（デフォルト値を設定）
    target_url = get_input_with_default(
        f"ApacheサーバーのURLを入力してください (デフォルト: {DEFAULT_URL}): ",
        DEFAULT_URL,
        is_integer=False
    )

    # ユーザーからアイドルタイムアウト秒数を入力（デフォルト値を設定）
    idle_timeout_seconds = get_input_with_default(
        f"アイドルタイムアウトをテストする秒数を入力してください (デフォルト: {DEFAULT_TIMEOUT}): ",
        DEFAULT_TIMEOUT,
        is_integer=True
    )

    # テストの開始時間を記録
    start_time = datetime.now()
    print(f"テスト開始時刻: {start_time.strftime('%Y-%m-%d %H:%M:%S')}")

    # アイドルタイムアウト検証を実行
    success = check_idle_timeout(target_url, idle_timeout_seconds)

    # テストの終了時間を記録
    end_time = datetime.now()
    print(f"テスト終了時刻: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")

    # テスト結果を表示
    if success:
        print("テスト成功: 接続はアイドル状態後も維持されました。")
    else:
        print("テスト失敗: 接続が維持されませんでした。")