はじめに
requestsライブラリを使って、PythonからWebサイトへリクエストを送る方法をすでに学びました。今回はその逆、つまりPythonでWebサーバーを作り、ブラウザからのリクエストを受け取る方法を学んでいきます。ここではPythonの標準ライブラリだけを使って、Webサーバーを立ち上げ、ブラウザにテキストを表示させるまでの流れを解説します。
Webサーバーの基本構成
Pythonには、Webアプリケーション開発のための強力なフレームワーク(DjangoやFlaskなど)が多数存在します。しかし、それらを学ぶ前に、PythonがどのようにHTTP通信を処理しているのかという根幹の部分を理解しておくことは非常に重要です。
標準ライブラリhttp.serverモジュール
Pythonの標準ライブラリに含まれるhttp.serverモジュールを使うと、複雑な設定なしに、Webサーバーの仕組みを学ぶのに最適な、シンプルなサーバーを構築できます。実際のプロダクション環境での使用は推奨されていませんが、学習目的にはうってつけです。

Pythonで簡易Webサーバーを立ち上げる際の、最も基本的な構成は以下のコードで実現できます。
from http.server import HTTPServer, BaseHTTPRequestHandler
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('Hello, World!'.encode('utf-8'))
def run(server_class=HTTPServer, handler_class=MyHandler):
server_address = ('', 8000) # 空文字は全てのインターフェース、ポートは8000
httpd = server_class(server_address, handler_class)
print(f'サーバーを起動しました: http://localhost:8000')
httpd.serve_forever()
if __name__ == '__main__':
run()
このコードには、HTTPServer と BaseHTTPRequestHandler という、2つの重要なクラスが登場します。今回のサンプルコードのMyHandlerクラスでは、do_GETメソッドを定義しています。これは、サーバーにGETリクエストが届いた場合に、このメソッドが自動的に呼び出されることを意味します。
レスポンスの手順
do_GETメソッドの中で、クライアントにレスポンスを返すためには、決められた手順を踏む必要があります。
- send_response(code) : HTTPステータスコード(例:200は成功)を送信します。
- send_header(keyword, value) : HTTPヘッダーを追加します。コンテンツのタイプを示す
Content-typeは特に重要です。 - end_headers() : ヘッダー部分の終わりを示す空白行を送信します。これでヘッダー送信は完了です。
- wfile.write(data) : レスポンスボディ(実際に表示させたい内容)を送信します。wfileはバイナリデータを書き込むためのストリームなので、文字列を送る場合はencode()メソッドでバイト列に変換する必要があります。
この例では、Content-typeを text/plain と指定しているので、ブラウザにはフォーマットされていないテキストが表示されます。text/html と指定すれば、HTMLとして解釈され、Webページを表示することも可能です。
HTTPServer クラス
HTTPServerは、実際にネットワーク上のポートをリッスン(待ち受け)し、クライアント(ブラウザ)からの接続を待つ役割を担います。
コンストラクタの第一引数server_addressは、タプル (サーバーのアドレス, ポート番号) で指定します。''(空文字)は、サーバーが動作するPCのすべてのIPアドレスで待ち受けることを意味し、’localhost’や’127.0.0.1’を指定すると、自分のPCからのアクセスのみに制限できます。ポート番号は 8000 などの空いている番号を指定します。
そして、httpd.serve_forever() を呼び出すことで、サーバーが起動し、リクエストが来るのを待ち続ける状態(イベントループ)に入ります。停止させるには Ctrl + C を押します。
BaseHTTPRequestHandlerクラス
BaseHTTPRequestHandlerは、実際に届いたリクエストの内容を解析し、それに応じた処理を行うためのクラスです。HTTPServerの第二引数として、このクラス(またはそのサブクラス)を渡します。重要なのは、このクラスは直接使うものではなく、必ず継承してサブクラスを作成するという点です。
BaseHTTPRequestHandlerを継承する
実際の処理は、BaseHTTPRequestHandlerを継承したクラスに記述します。HTTPリクエストのメソッド(GETやPOSTなど)に対応する、do_GETやdo_POSTという名前のメソッドを定義するのがルールです。do_GET と do_POST は、クライアントから送られてきたHTTPリクエストの種類によって自動的に呼び出されるメソッドです。
do_GET は データを取得するとき に使用します。例えば、Webページを表示したり、データベースから情報を読み取って表示したりする処理を記述します。一方、do_POST は データを送信するとき に使用します。例えば、フォームに入力した内容をサーバーへ送信したり、データベースへ新しいデータを登録・更新したりする処理を記述します。
一般的には、「表示や取得は GET」、「登録や更新は POST」というように使い分けます。
from http.server import BaseHTTPRequestHandler, HTTPServer
class MyHandler(BaseHTTPRequestHandler):
# GETリクエストを受け取ったとき
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/plain; charset=utf-8")
self.end_headers()
self.wfile.write("GETリクエストを受信しました".encode("utf-8"))
# POSTリクエストを受け取ったとき
def do_POST(self):
# 送信されたデータの長さを取得
content_length = int(self.headers["Content-Length"])
# データを読み込む
post_data = self.rfile.read(content_length)
self.send_response(200)
self.send_header("Content-type", "text/plain; charset=utf-8")
self.end_headers()
response = f"受信したデータ: {post_data.decode('utf-8')}"
self.wfile.write(response.encode("utf-8"))
server = HTTPServer(("localhost", 8000), MyHandler)
print("サーバー起動中...")
server.serve_forever()
この例では、ブラウザで http://localhost:8000 にアクセスすると do_GET() が実行され、「GETリクエストを受信しました」と表示されます。一方、POSTリクエストでデータを送信すると do_POST() が実行され、送信されたデータを読み取ってレスポンスとして返します。
ベストプラクティス
1. コードの準備
まず、以下の完全なコードを simple_server.py という名前で保存します。このコードはベストプラクティスに従い、エラーハンドリングやログ出力も含めた実用的な例です。
from http.server import HTTPServer, BaseHTTPRequestHandler
import datetime
import sys
class MyHandler(BaseHTTPRequestHandler):
"""HTTPリクエストを処理するハンドラークラス"""
def log_message(self, format, *args):
"""カスタムログ出力メソッド"""
timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f'[{timestamp}] {self.address_string()} - {format % args}')
def do_GET(self):
"""GETリクエストを処理する"""
try:
# リクエスト情報をログに出力
self.log_message(f'GETリクエスト: {self.path}')
# レスポンスボディを準備
response_body = 'Hello, World!'
# レスポンスを送信
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.send_header('Server', 'Python-HTTP-Server/1.0')
self.send_header('Content-Length', str(len(response_body.encode('utf-8'))))
self.end_headers()
self.wfile.write(response_body.encode('utf-8'))
except Exception as e:
# エラーハンドリング
self.log_message(f'エラー発生: {e}')
self.send_response(500)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('内部サーバーエラーが発生しました'.encode('utf-8'))
def do_HEAD(self):
"""HEADリクエストを処理する(ボディなし)"""
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.send_header('Server', 'Python-HTTP-Server/1.0')
self.end_headers()
def run(server_class=HTTPServer, handler_class=MyHandler, port=8000):
"""サーバーを起動する関数"""
server_address = ('', port)
httpd = server_class(server_address, handler_class)
print(f'🚀 サーバーを起動しました')
print(f'📡 アドレス: http://localhost:{port}')
print(f'🔄 待受中... (終了するには Ctrl+C を押してください)')
print('-' * 50)
try:
httpd.serve_forever()
except KeyboardInterrupt:
print('\n' + '-' * 50)
print('🛑 サーバーを停止しました')
httpd.server_close()
if __name__ == '__main__':
# ポート番号はコマンドライン引数で指定可能
port = 8000
if len(sys.argv) > 1:
try:
port = int(sys.argv[1])
except ValueError:
print(f'⚠️ 無効なポート番号です。デフォルトの8000を使用します。')
run(port=port)
2. サーバーの起動
ターミナル(コマンドプロンプト)を開き、simple_server.pyを保存したディレクトリに移動します。以下のコマンドを実行してサーバーを起動します。
python simple_server.py
または、ポート番号を指定して起動することもできます。
python simple_server.py 8080
正常に起動すると、以下のような出力が表示されます。
🚀 サーバーを起動しました
📡 アドレス: http://localhost:8000
🔄 待受中... (終了するには Ctrl+C を押してください)
--------------------------------------------------
3. ブラウザでアクセス
ブラウザを開き、アドレスバーに http://localhost:8000 と入力してアクセスします。
「Hello, World!」と表示されれば成功です。
4. ログの確認
ターミナルには、アクセスがあった際のログが表示されます。
[2026-06-29 15:30:45] 127.0.0.1 - GETリクエスト: /
[2026-06-29 15:30:45] 127.0.0.1 - "GET / HTTP/1.1" 200 -
5. 他のHTTPメソッドのテスト
curlコマンドを使用して、HEADリクエストを送信することもできます。
curl -I http://localhost:8000
以下のようなレスポンスヘッダーが表示されるはずです。
HTTP/1.0 200 OK
Server: Python-HTTP-Server/1.0
Date: Mon, 29 Jun 2026 06:30:45 GMT
Content-type: text/plain; charset=utf-8
Content-Length: 13
6. サーバーの停止
サーバーを停止するには、ターミナルで Ctrl + C を押します。
^C
--------------------------------------------------
🛑 サーバーを停止しました
上記のコードは以下のベストプラクティスに従っています。
- カスタムログ出力 : log_messageメソッドをオーバーライドして、タイムスタンプ付きの見やすいログを出力しています。
- エラーハンドリング : try-exceptブロックで例外を捕捉し、クライアントに適切なエラーレスポンス(500 Internal Server Error)を返します。
- HEADメソッドのサポート : do_HEADメソッドを実装して、ボディなしのレスポンスを返せるようにしています。
- コンテンツ長の指定 : Content-Lengthヘッダーを設定して、レスポンスのサイズを明示しています。
- サーバー情報の送信 : Serverヘッダーでサーバー情報を送信しています。
- クリーンなシャットダウン : KeyboardInterruptを捕捉して、サーバーソケットを適切にクローズしています。
- ポート指定の柔軟性 : コマンドライン引数でポート番号を指定できるようにしています。
まとめ
今回は、Pythonの標準ライブラリhttp.serverを使い、たった数行のコードでWebサーバーを立ち上げる方法を学びました。
- HTTPServerがリクエストを待ち受け、BaseHTTPRequestHandlerがリクエストを処理するという役割分担。
- 処理を記述するには、BaseHTTPRequestHandlerを継承し、do_GETなどのメソッドを定義する。
- レスポンスを返すには、send_response、send_header、end_headers、wfile.writeの順に呼び出す。
これで、PythonでHTTPのリクエストとレスポンスの基本を扱えるようになりました。次回は、この知識をさらに発展させ、ブラウザからの入力を受け付けるルーティングやHTMLファイルとの連携を実装し、より実用的なWebアプリケーションへと進化させていきます。
演習問題
初級問題(3問)
問題1: 基本的なGETリクエストの処理
ブラウザから http://localhost:8000 にアクセスした時に、以下の情報を表示するWebサーバーを作成してください。
- 現在の日時(例:2026年6月29日 15:30:45)
- サーバーが動作しているPythonのバージョン
- “これはPythonの簡易Webサーバーです”というメッセージ
実行結果(ブラウザ表示):
現在の日時: 2026年6月29日 15:30:45
Pythonバージョン: 3.11.4
これはPythonの簡易Webサーバーです
問題2: パスによる条件分岐
以下の要件を満たすWebサーバーを作成してください。
/にアクセスした場合:「ホームページへようこそ!」と表示/aboutにアクセスした場合:「このサーバーは学習用です」と表示/contactにアクセスした場合:「連絡先: example@example.com」と表示- 上記以外のパスにアクセスした場合:ステータスコード404と共に「ページが見つかりません」と表示
問題3: クエリパラメータの処理
ブラウザからのGETリクエストで、クエリパラメータを受け取るWebサーバーを作成してください。
http://localhost:8000/greet?name=田中にアクセスした場合:「こんにちは、田中さん!」と表示nameパラメータがない場合:「名前を入力してください」と表示http://localhost:8000/greet?name=鈴木&age=25のように複数パラメータがある場合:「こんにちは、鈴木さん(25歳)!」と表示
ヒント:urllib.parseモジュールのurlparseとparse_qs関数を使用します。
中級問題(6問)
問題4: JSONレスポンスの返却
リクエストに応じてJSONデータを返すWebサーバーを作成してください。
/api/timeにアクセスすると、現在時刻をJSON形式で返す/api/statusにアクセスすると、サーバーのステータス情報をJSONで返す- レスポンスの
Content-Typeはapplication/jsonに設定する
実行結果(/api/time アクセス時):
{
"status": "success",
"data": {
"datetime": "2026-06-29 15:30:45",
"timestamp": 1719657045
}
}
問題5: フォームデータの受け取り(GETメソッド)
GETメソッドを使用したフォーム送信を処理するWebサーバーを作成してください。
/formにアクセスすると、テキスト入力欄と送信ボタンがあるシンプルなHTMLフォームを表示- フォーム送信後(
/submit?message=入力内容)、入力内容を表示するページにリダイレクト - 入力内容がない場合はエラーメッセージを表示
問題6: リクエストヘッダーの表示
ブラウザから送信されたリクエストヘッダー情報を表示するWebサーバーを作成してください。
/headersにアクセスすると、すべてのリクエストヘッダーをリスト表示- 特に以下のヘッダーを強調表示:
- User-Agent
- Accept-Language
- Host
- ヘッダー情報は整形して見やすく表示
問題7: 静的ファイルの配信(テキストファイル)
指定されたテキストファイルを読み込んで表示するWebサーバーを作成してください。
/file?name=sample.txtにアクセスすると、sample.txtの内容を表示- ファイルが存在しない場合は「ファイルが見つかりません」と表示
- ファイルの内容はプレーンテキストとして表示
fileディレクトリ内のファイルのみアクセス可能にする(セキュリティ対策)
問題8: アクセスカウンターの実装
ページアクセス数をカウントするWebサーバーを作成してください。
- サーバー起動時にカウントを0に初期化
- ルートパス(
/)にアクセスされるたびにカウントを1増加 - アクセス時に「このページは X 回アクセスされました」と表示
- カウントはサーバーが動作している間だけ保持(再起動するとリセット)
問題9: リダイレクトの実装
以下のリダイレクト機能を持つWebサーバーを作成してください。
/old-pageにアクセスすると、/new-pageにリダイレクト(ステータスコード302)/new-pageでは「新しいページへようこそ!」と表示/externalにアクセスすると、外部サイト(例:https://www.python.org)にリダイレクト- リダイレクト先のURLをログに出力
上級問題(3問)
問題10: マルチスレッド対応サーバー
複数のクライアントからの同時アクセスを処理できるWebサーバーを作成してください。
ThreadingHTTPServerを使用してマルチスレッド化- 各リクエストの処理時間を計測してログ出力
- 同時に5つのリクエストを処理できることを確認するテスト用エンドポイント(
/slow)を実装(2秒待機してから応答) - リクエストごとにスレッドIDを表示
問題11: カスタムエラーページの実装
様々なHTTPエラーに対してカスタムエラーページを表示するWebサーバーを作成してください。
- 404 Not Found:オリジナルの404ページを表示
- 403 Forbidden:アクセス制限されたページにアクセスした場合に表示
- 500 Internal Server Error:サーバー内で例外が発生した場合に表示
- 各エラーページにはエラーコードと分かりやすいメッセージ、スタイルシートを適用
問題12: 簡易APIサーバーの構築
RESTfulなAPIエンドポイントを持つWebサーバーを作成してください。
GET /api/users:ユーザー一覧をJSONで返却GET /api/users/{id}:特定のユーザー情報をJSONで返却POST /api/users:新しいユーザーを追加(JSONボディを受け取る)DELETE /api/users/{id}:ユーザーを削除- ユーザーデータはメモリ上(リストや辞書)で管理
- 各エンドポイントで適切なHTTPステータスコードを返却
演習問題 解答例
初級問題 解答
問題1: 基本的なGETリクエストの処理
from http.server import HTTPServer, BaseHTTPRequestHandler
import datetime
import sys
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
# 現在の日時を取得
now = datetime.datetime.now()
date_str = now.strftime('%Y年%m月%d日 %H:%M:%S')
# Pythonバージョンを取得
python_version = sys.version.split()[0]
# レスポンスボディを作成
body = f"""現在の日時: {date_str}
Pythonバージョン: {python_version}
これはPythonの簡易Webサーバーです"""
# レスポンスを送信
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題2: パスによる条件分岐
from http.server import HTTPServer, BaseHTTPRequestHandler
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
# パスに基づいて処理を分岐
if self.path == '/':
body = 'ホームページへようこそ!'
status = 200
elif self.path == '/about':
body = 'このサーバーは学習用です'
status = 200
elif self.path == '/contact':
body = '連絡先: example@example.com'
status = 200
else:
body = 'ページが見つかりません'
status = 404
# レスポンスを送信
self.send_response(status)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題3: クエリパラメータの処理
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
# URLを解析
parsed_url = urlparse(self.path)
# パスが /greet の場合のみ処理
if parsed_url.path == '/greet':
# クエリパラメータを解析
params = parse_qs(parsed_url.query)
# nameパラメータを取得(存在しない場合はNone)
name = params.get('name', [''])[0]
age = params.get('age', [''])[0]
if not name:
body = '名前を入力してください'
elif age:
body = f'こんにちは、{name}さん({age}歳)!'
else:
body = f'こんにちは、{name}さん!'
status = 200
else:
body = 'ページが見つかりません'
status = 404
# レスポンスを送信
self.send_response(status)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
httpd.serve_forever()
if __name__ == '__main__':
run()
中級問題 解答
問題4: JSONレスポンスの返却
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse
import json
import datetime
import sys
import os
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/api/time':
# 現在時刻をJSONで返却
now = datetime.datetime.now()
data = {
'status': 'success',
'data': {
'datetime': now.strftime('%Y-%m-%d %H:%M:%S'),
'timestamp': int(now.timestamp())
}
}
status = 200
elif parsed_url.path == '/api/status':
# サーバーステータスをJSONで返却
data = {
'status': 'success',
'data': {
'server': 'Python HTTP Server',
'version': sys.version.split()[0],
'pid': os.getpid(),
'uptime': 'N/A' # 実際のuptimeは別途計測が必要
}
}
status = 200
else:
data = {'error': 'Not Found'}
status = 404
# JSONレスポンスを送信
self.send_response(status)
self.send_header('Content-type', 'application/json; charset=utf-8')
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題5: フォームデータの受け取り(GETメソッド)
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/form':
# フォームページを表示
html = """
<meta charset="UTF-8">
<title>フォーム</title>
<h1>メッセージを入力してください</h1>
<form action="/submit" method="GET">
<input type="text" name="message" placeholder="メッセージを入力">
<button type="submit">送信</button>
</form>
"""
self.send_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
elif parsed_url.path == '/submit':
# フォーム送信後の処理
params = parse_qs(parsed_url.query)
message = params.get('message', [''])[0]
if message:
body = f'入力されたメッセージ: {message}'
else:
body = 'エラー: メッセージが入力されていません'
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
else:
self.send_response(404)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ページが見つかりません'.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題6: リクエストヘッダーの表示
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/headers':
# ヘッダー情報を整形
html = """
<meta charset="UTF-8">
<title>リクエストヘッダー</title>
<style>
body { font-family: sans-serif; margin: 40px; }
.header-list { list-style: none; padding: 0; }
.header-item { padding: 8px; margin: 4px 0; background: #f5f5f5; border-radius: 4px; }
.header-key { font-weight: bold; color: #0066cc; }
.highlight { background: #fff3cd; border-left: 4px solid #ffc107; }
</style>
<h1>📋 リクエストヘッダー</h1>
<ul class="header-list">"""
# 強調表示するヘッダー
highlight_headers = ['User-Agent', 'Accept-Language', 'Host']
# すべてのヘッダーを表示
for key, value in self.headers.items():
is_highlight = key in highlight_headers
class_name = 'header-item highlight' if is_highlight else 'header-item'
html += f'<li class="{class_name}"><span class="header-key">{key}:</span> {value}</li>'
html += """</ul>
"""
self.send_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
else:
self.send_response(404)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ページが見つかりません'.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題7: 静的ファイルの配信(テキストファイル)
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
import os
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/file':
params = parse_qs(parsed_url.query)
filename = params.get('name', [''])[0]
if not filename:
self.send_response(400)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ファイル名が指定されていません'.encode('utf-8'))
return
# セキュリティ対策:ファイル名にパストラバーサル攻撃を防ぐ
if '..' in filename or '/' in filename or '\\' in filename:
self.send_response(403)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('アクセスが拒否されました'.encode('utf-8'))
return
# filesディレクトリ内のファイルのみアクセス可能
filepath = os.path.join('files', filename)
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(content.encode('utf-8'))
except FileNotFoundError:
self.send_response(404)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ファイルが見つかりません'.encode('utf-8'))
else:
self.send_response(404)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ページが見つかりません'.encode('utf-8'))
def run():
# filesディレクトリが存在しない場合は作成
if not os.path.exists('files'):
os.makedirs('files')
# サンプルファイルを作成
with open('files/sample.txt', 'w', encoding='utf-8') as f:
f.write('これはサンプルテキストファイルです。\nPythonのWebサーバーから配信されています。')
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
print('例: http://localhost:8000/file?name=sample.txt')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題8: アクセスカウンターの実装
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse
class MyHandler(BaseHTTPRequestHandler):
# クラス変数としてカウンターを保持
counter = 0
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/':
# アクセスカウントを増加
MyHandler.counter += 1
# HTMLを生成
html = f"""
<meta charset="UTF-8">
<title>アクセスカウンター</title>
<style>
body {{
font-family: sans-serif;
text-align: center;
margin-top: 100px;
}}
.counter {{
font-size: 48px;
color: #0066cc;
font-weight: bold;
}}
.message {{
font-size: 24px;
margin-bottom: 20px;
}}
</style>
<div class="message">このページは</div>
<div class="counter">{MyHandler.counter}</div>
<div class="message">回アクセスされました</div>
"""
self.send_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
else:
self.send_response(404)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ページが見つかりません'.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
print('アクセスカウンターを初期化しました(カウント: 0)')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題9: リダイレクトの実装
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse
class MyHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/old-page':
# リダイレクト(302 Found)
self.send_response(302)
self.send_header('Location', '/new-page')
self.end_headers()
print(f'リダイレクト: /old-page -> /new-page')
elif parsed_url.path == '/new-page':
# 新しいページ
body = '新しいページへようこそ!'
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
elif parsed_url.path == '/external':
# 外部サイトへのリダイレクト
external_url = 'https://www.python.org'
self.send_response(302)
self.send_header('Location', external_url)
self.end_headers()
print(f'リダイレクト: /external -> {external_url}')
else:
self.send_response(404)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ページが見つかりません'.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
print('利用可能なパス:')
print(' /old-page -> /new-page にリダイレクト')
print(' /new-page -> 新しいページを表示')
print(' /external -> Python公式サイトにリダイレクト')
httpd.serve_forever()
if __name__ == '__main__':
run()
上級問題 解答
問題10: マルチスレッド対応サーバー
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
from urllib.parse import urlparse
import threading
import time
import datetime
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
"""マルチスレッド対応のHTTPServer"""
pass
class MyHandler(BaseHTTPRequestHandler):
def log_request(self, code='-', size='-'):
# カスタムログ出力
thread_id = threading.current_thread().ident
print(f'[{datetime.datetime.now().strftime("%H:%M:%S")}] '
f'スレッドID:{thread_id} - {self.path} - ステータス:{code}')
def do_GET(self):
parsed_url = urlparse(self.path)
if parsed_url.path == '/slow':
# 2秒待機する遅い処理
thread_id = threading.current_thread().ident
print(f'スレッド {thread_id}: 遅延処理を開始')
time.sleep(2)
body = f'遅延処理が完了しました(スレッドID: {thread_id})'
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
elif parsed_url.path == '/':
thread_id = threading.current_thread().ident
body = f'通常レスポンス(スレッドID: {thread_id})'
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(body.encode('utf-8'))
else:
self.send_response(404)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write('ページが見つかりません'.encode('utf-8'))
def run():
server_address = ('', 8000)
httpd = ThreadingHTTPServer(server_address, MyHandler)
print('マルチスレッド対応サーバーを起動しました: http://localhost:8000')
print('テスト用の遅延エンドポイント: /slow (2秒待機)')
print('Ctrl+Cで停止')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題11: カスタムエラーページの実装
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse
class MyHandler(BaseHTTPRequestHandler):
def send_error_page(self, status_code, title, message):
"""カスタムエラーページを送信"""
html = f"""
<meta charset="UTF-8">
<title>{status_code} {title}</title>
<style>
body {{
font-family: sans-serif;
text-align: center;
padding: 50px;
background: #f5f5f5;
}}
.error-container {{
max-width: 600px;
margin: 0 auto;
background: white;
padding: 40px;
border-radius: 8px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}}
.error-code {{
font-size: 72px;
font-weight: bold;
color: #dc3545;
margin: 0;
}}
.error-title {{
font-size: 24px;
color: #333;
margin: 10px 0;
}}
.error-message {{
color: #666;
margin: 20px 0;
}}
.home-link {{
color: #0066cc;
text-decoration: none;
}}
.home-link:hover {{
text-decoration: underline;
}}
</style>
<div class="error-container">
<h1 class="error-code">{status_code}</h1>
<h2 class="error-title">{title}</h2>
<p class="error-message">{message}</p>
<a href="/" class="home-link">ホームに戻る</a>
</div>
"""
self.send_response(status_code)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
def do_GET(self):
parsed_url = urlparse(self.path)
try:
if parsed_url.path == '/':
# ホームページ
html = """
<meta charset="UTF-8">
<title>ホーム</title>
<h1>ホームページへようこそ</h1>
"""
self.send_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
elif parsed_url.path == '/admin':
# アクセス制限されたページ(403)
self.send_error_page(403, 'Forbidden', 'このページへのアクセスは許可されていません')
elif parsed_url.path == '/raise-error':
# 意図的にエラーを発生させる(500)
raise Exception('意図的なサーバーエラー')
else:
# 404 Not Found
self.send_error_page(404, 'Not Found', 'お探しのページは見つかりませんでした')
except Exception as e:
# 500 Internal Server Error
print(f'サーバーエラー: {e}')
self.send_error_page(500, 'Internal Server Error', 'サーバー内部でエラーが発生しました')
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('サーバーを起動しました: http://localhost:8000')
print('テスト用パス:')
print(' / - ホームページ')
print(' /admin - 403エラー')
print(' /raise-error - 500エラー')
print(' その他 - 404エラー')
httpd.serve_forever()
if __name__ == '__main__':
run()
問題12: 簡易APIサーバーの構築
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
import json
import re
# ユーザーデータ(メモリ上で管理)
users = [
{'id': 1, 'name': '田中太郎', 'email': 'tanaka@example.com'},
{'id': 2, 'name': '鈴木次郎', 'email': 'suzuki@example.com'},
{'id': 3, 'name': '佐藤花子', 'email': 'sato@example.com'}
]
next_id = 4
class MyHandler(BaseHTTPRequestHandler):
def send_json_response(self, data, status=200):
"""JSONレスポンスを送信"""
self.send_response(status)
self.send_header('Content-type', 'application/json; charset=utf-8')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
def do_GET(self):
parsed_url = urlparse(self.path)
# GET /api/users - ユーザー一覧
if parsed_url.path == '/api/users':
self.send_json_response({
'status': 'success',
'data': users
})
return
# GET /api/users/{id} - 特定ユーザー
match = re.match(r'^/api/users/(\d+)$', parsed_url.path)
if match:
user_id = int(match.group(1))
user = next((u for u in users if u['id'] == user_id), None)
if user:
self.send_json_response({
'status': 'success',
'data': user
})
else:
self.send_json_response({
'status': 'error',
'message': 'ユーザーが見つかりません'
}, 404)
return
# その他のパスは404
self.send_json_response({
'status': 'error',
'message': 'エンドポイントが見つかりません'
}, 404)
def do_POST(self):
parsed_url = urlparse(self.path)
# POST /api/users - ユーザー追加
if parsed_url.path == '/api/users':
global next_id
# リクエストボディを読み取り
content_length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_length).decode('utf-8')
try:
data = json.loads(body)
name = data.get('name')
email = data.get('email')
if not name or not email:
self.send_json_response({
'status': 'error',
'message': 'nameとemailは必須です'
}, 400)
return
# ユーザーを追加
new_user = {
'id': next_id,
'name': name,
'email': email
}
users.append(new_user)
next_id += 1
self.send_json_response({
'status': 'success',
'message': 'ユーザーを追加しました',
'data': new_user
}, 201)
except json.JSONDecodeError:
self.send_json_response({
'status': 'error',
'message': '無効なJSON形式です'
}, 400)
return
# その他のパスは404
self.send_json_response({
'status': 'error',
'message': 'エンドポイントが見つかりません'
}, 404)
def do_DELETE(self):
parsed_url = urlparse(self.path)
# DELETE /api/users/{id} - ユーザー削除
match = re.match(r'^/api/users/(\d+)$', parsed_url.path)
if match:
user_id = int(match.group(1))
global users
# ユーザーを検索
user_index = next((i for i, u in enumerate(users) if u['id'] == user_id), None)
if user_index is not None:
deleted_user = users.pop(user_index)
self.send_json_response({
'status': 'success',
'message': 'ユーザーを削除しました',
'data': deleted_user
})
else:
self.send_json_response({
'status': 'error',
'message': 'ユーザーが見つかりません'
}, 404)
return
# その他のパスは404
self.send_json_response({
'status': 'error',
'message': 'エンドポイントが見つかりません'
}, 404)
def run():
server_address = ('', 8000)
httpd = HTTPServer(server_address, MyHandler)
print('APIサーバーを起動しました: http://localhost:8000')
print('エンドポイント:')
print(' GET /api/users - ユーザー一覧')
print(' GET /api/users/{id} - 特定ユーザー')
print(' POST /api/users - ユーザー追加')
print(' DELETE /api/users/{id} - ユーザー削除')
print('ユーザーデータ:')
for user in users:
print(f' ID:{user["id"]} {user["name"]} ({user["email"]})')
httpd.serve_forever()
if __name__ == '__main__':
run()