645 views

【1】やりたいこと

Web上で公開している CGIプログラムの中には、処理が重たいものもある。

このプログラムに高頻度でリクエストが届くとサーバ負荷が上がり・・・
共用サーバであればアカウントがBANされる恐れがある。

よって、
同じIPアドレスからのリクエストを一定時間拒否する実装を加えたい。

【2】やってみる

IPアドレスによるブロック処理を実装し、これを CGIの先頭で実行する。
アクセス履歴の管理には、簡易DBである SQLite3 を使うことにした。

import os
import time
import sqlite3

class IPBlocker:
    def __init__(self, db_path='/home/hoge/ip_access_log.db', block_seconds=10):
        self.db_path = db_path
        self.block_seconds = block_seconds
        self._init_db()

    # データベースとテーブルを初期化
    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        cur = conn.cursor()
        cur.execute('CREATE TABLE IF NOT EXISTS ip_access (ip TEXT PRIMARY KEY, last_access INTEGER)')
        conn.commit()
        conn.close()

    def check_ip(self):
        remote_ip = os.environ.get('REMOTE_ADDR', '')
        now = int(time.time())

        conn = sqlite3.connect(self.db_path)
        cur = conn.cursor()

        # 古いレコード削除
        cutoff_time = now - self.block_seconds
        cur.execute('DELETE FROM ip_access WHERE last_access < ?', (cutoff_time,))

        # 現在のIPの最終アクセス確認
        cur.execute('SELECT last_access FROM ip_access WHERE ip = ?', (remote_ip,))
        row = cur.fetchone()

        # レコードが存在 → block_seconds以内にアクセスあり → アクセス拒否
        if row:
            conn.close()
            print("Status: 429 Too Many Requests")
            print("Content-Type: text/plain\n")
            print("Too many requests. Please wait a few seconds.")
            exit(0)

        # 今回のアクセスを記録（INSERT OR REPLACE）
        cur.execute('REPLACE INTO ip_access (ip, last_access) VALUES (?, ?)', (remote_ip, now))
        conn.commit()
        conn.close()

#!/usr/bin/python
# -*- coding: utf-8 -*-

#-------------------------------->>> IPアドレスによる頻度制限
from ip_blocker import IPBlocker
IPBlocker().check_ip()
#--------------------------------<<< IPアドレスによる頻度制限

print("Content-Type: text/html; charset=utf-8\n")
　：
以下に CGIプログラムを書く。
　：