第18章 セキュリティ

入力検証の基本例

ユーザーからの数字の入力を受け取り、範囲内にあるかどうかをチェックするシンプルな例です。

def get_number_input():
    try:
        number = int(input("数字を入力してください(0-100): "))
        if 0 <= number <= 100:
            return number
        else:
            print("0から100の間の数字を入力してください。")
            return None
    except ValueError:
        print("有効な数字を入力してください。")
        return None

number = get_number_input()
if number is not None:
    print(f"入力された数字は {number} です。")


文字列のフィルタリング例

ユーザーからの文字列入力を受け取り、特定のキーワードが含まれていないかをチェックする例です。

def get_string_input():
    user_input = input("メッセージを入力してください: ")

    forbidden_keywords = ["badword1", "badword2", "badword3"]
    for keyword in forbidden_keywords:
        if keyword in user_input:
            print(f"'{keyword}'は禁止されているワードです。")
            return None

    return user_input

message = get_string_input()
if message is not None:
    print(f"入力されたメッセージ: {message}")


SQLインジェクションの防止例

安全なコードの例を以下に示します。

import sqlite3

# ユーザー入力を受け取る
user_id = input("Enter user id: ")

# プレースホルダを使用してSQLを実行
query = "SELECT * FROM users WHERE id = ?"

conn = sqlite3.connect('my_database.db')
cursor = conn.cursor()
cursor.execute(query, (user_id,))

results = cursor.fetchall()
print(results)


セッションの基本的な使用方法

セッションIDの生成

セッションIDは十分な長さと複雑さを持つランダムな文字列である必要があります。

secretsモジュールを使用して、十分な強度のあるセッションIDを生成します。

import secrets

def generate_session_id():
    return secrets.token_hex(16)

セッションの保存

セッションデータはサーバ側に安全に保存する必要があります。このデモでは、単純な辞書を使用しますが、実際にはデータベースやキャッシュシステムを使用することが一般的です。

sessions = {}

def create_session(user_id):
    session_id = generate_session_id()
    sessions[session_id] = user_id
    return session_id

セッションの有効期限

セッションは無期限に持続するべきではありません。セッションに有効期限を設定して、古いセッションを削除します。

import datetime

sessions = {}

def create_session(user_id):
    session_id = generate_session_id()
    sessions[session_id] = {'user_id': user_id, 'expiry': datetime.datetime.now() + datetime.timedelta(hours=1)}
    return session_id

def is_session_valid(session_id):
    if session_id in sessions:
        if sessions[session_id]['expiry'] > datetime.datetime.now():
            return True
        else:
            del sessions[session_id]
    return False


パスワードセキュリティ

パスワードのハッシュ化

パスワードをデータベースに保存する場合、生のパスワードを直接保存するのではなく、ハッシュ化された形で保存します。これにより、データベースが漏洩しても、攻撃者が直接パスワードを知ることはできません。

import hashlib
import os

def hash_password(password):
    # 16バイトのランダムな塩を生成
    salt = os.urandom(16)
    hasher = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, 100000)
    return (salt + hasher).hex()

hashed_pw = hash_password("my_password")

パスワードの検証

ユーザからの入力と保存されたハッシュを比較するための関数を提供します。

def verify_password(stored_hash, input_password):
    salt = bytes.fromhex(stored_hash[:32])
    stored_pw_hash = stored_hash[32:]
    hasher = hashlib.pbkdf2_hmac('sha256', input_password.encode('utf-8'), salt, 100000)
    return hasher.hex() == stored_pw_hash

is_valid = verify_password(hashed_pw, "my_password")

パスワードの強度の検証

ユーザが強力なパスワードを選択することを確認するための基本的なチェックを提供します。

import re

def is_strong_password(password):
    if len(password) < 8:
        return False
    if not re.search(r"[a-z]", password):
        return False
    if not re.search(r"[A-Z]", password):
        return False
    if not re.search(r"[0-9]", password):
        return False
    return True

print(is_strong_password("P@ssw0rd"))  # True


CSRF対策

トークンベースの防御(トークンの生成)

各セッションまたはフォームごとに一意のトークンを生成します。

import os

def generate_csrf_token() -> str:
    return os.urandom(16).hex()

トークンベースの防御(トークンの埋め込み)

フォームに隠しフィールドとしてトークンを埋め込みます。例:

<form action="/submit" method="post">
    <input type="hidden" name="csrf_token" value="">
    <!-- other form fields -->
    <input type="submit" value="Submit">
</form>

トークンベースの防御(トークンの検証)

サーバ側で、受け取ったトークンがセッションで保存されているトークンと一致するかを確認します。

def is_valid_csrf_token(session_token: str, form_token: str) -> bool:
    return session_token == form_token


Refererヘッダの検証

Refererヘッダを検証して、リクエストが正当なソースから来ていることを確認します。

def is_valid_referer(referer: str) -> bool:
    # この例では、正当なリクエストのRefererヘッダは https://example.com から始まるとします。
    return referer.startswith('https://example.com')


ディレクトリ・トラバーサル攻撃(パス・トラバーサル攻撃)

ディレクトリ・トラバーサル攻撃(パス・トラバーサル攻撃)を防ぐための基本的な方法を示すコード例

import os

def safe_join(base_directory, user_input_path):
    # 絶対パスに変換して、意図しないディレクトリへのアクセスを防ぐ
    safe_path = os.path.abspath(os.path.join(base_directory, user_input_path))
    
    # 安全なベースディレクトリの中にあることを確認する
    if not safe_path.startswith(os.path.abspath(base_directory)):
        raise ValueError("Unauthorized access attempt to a restricted area")

    return safe_path

def get_file_contents(base_directory, user_input_path):
    # 安全な結合関数を使用してファイルパスを取得
    file_path = safe_join(base_directory, user_input_path)
    
    # ファイルの内容を読み込んで返す
    with open(file_path, 'r') as file:
        return file.read()

# 安全なベースディレクトリ
base_directory = '/var/www/myapp/data'

# ユーザーからの入力(ファイル名やパス)
user_input_path = '../etc/passwd'

try:
    # ユーザーの入力に基づいてファイルの内容を取得
    contents = get_file_contents(base_directory, user_input_path)
except ValueError as e:
    print(e)  # 不正なアクセスの場合、エラーメッセージを表示


クリックジャッキング対策

X-Frame-Optionsヘッダを設定する単純なHTTPサーバーの例です。

from http.server import HTTPServer, BaseHTTPRequestHandler

class HTTPRequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # クリックジャッキングを防ぐためのヘッダを設定
        self.send_response(200)
        self.send_header('Content-type', 'text/html')
        self.send_header('X-Frame-Options', 'DENY')  # iframe内での表示を完全に拒否
        # 他にも'SAMEORIGIN'オプションを使用でき、同じドメイン内の表示のみを許可します
        # self.send_header('X-Frame-Options', 'SAMEORIGIN')
        self.end_headers()

        # 簡単なHTMLコンテンツを返す
        self.wfile.write(b"<html><body><h1>Hello, World!</h1></body></html>")

# サーバーを起動
httpd = HTTPServer(('localhost', 8000), HTTPRequestHandler)
httpd.serve_forever()


XSS(クロスサイトスクリプティング)対策

Pythonの標準ライブラリを使ったXSS対策の例を示します。

from http.server import HTTPServer, BaseHTTPRequestHandler
import html

class XSSPreventionHTTPRequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # サーバーからクライアントへの応答
        self.send_response(200)
        self.send_header('Content-type', 'text/html')
        self.end_headers()

        # URLのクエリパラメータを取得(実際のコードでは適切なパースが必要)
        query = self.path.split('?', 1)[-1]
        # ユーザー入力(ここではクエリパラメータ)をエスケープ
        safe_input = html.escape(query)

        # 安全な入力を含むHTMLコンテンツを生成
        html_content = f"<html><body><h1>Your input was: {safe_input}</h1></body></html>"
        # 応答としてHTMLコンテンツを送信
        self.wfile.write(html_content.encode())

# サーバーを起動
httpd = HTTPServer(('localhost', 8000), XSSPreventionHTTPRequestHandler)
httpd.serve_forever()


ファイルアップロードのセキュリティ

Pythonのhttp.serverを用いてファイルアップロードのセキュリティを扱う簡単な例です。

import os
from http.server import HTTPServer, BaseHTTPRequestHandler
from cgi import FieldStorage
from urllib.parse import parse_header

class SecureFileUploadHTTPRequestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        # コンテンツタイプのチェック
        ctype, pdict = parse_header(self.headers['content-type'])
        if ctype != 'multipart/form-data':
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"Content-Type is not multipart/form-data")
            return

        # アップロードされたファイルを取得
        form = FieldStorage(
            fp=self.rfile,
            headers=self.headers,
            environ={'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': self.headers['Content-Type'],}
        )

        # formからファイルデータを取得
        file_item = form['file']

        # ファイル名のチェック(ディレクトリトラバーサル攻撃対策)
        file_name = os.path.basename(file_item.filename)
        
        # 安全でないファイル名のチェック
        if not file_name:
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"No filename provided.")
            return

        # ファイルサイズのチェック
        file_size = os.fstat(file_item.file.fileno()).st_size
        if file_size > 1024 * 1024 * 5:  # 5MB以上のファイルは拒否
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"File too large.")
            return
        
        # ファイルタイプのチェック(例:.exeファイルを拒否)
        if file_name.endswith('.exe'):
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"Executable files are not allowed.")
            return

        # ファイルの保存
        with open(f'uploads/{file_name}', 'wb') as f:
            f.write(file_item.file.read())

        # レスポンスを返す
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"File uploaded successfully.")

# サーバーを起動
httpd = HTTPServer(('localhost', 8000), SecureFileUploadHTTPRequestHandler)
httpd.serve_forever()


練習問題1.

ユーザーからパスワードを受け取り、安全なハッシュを生成するプログラムを作成してください。さらに、ハッシュを検証する関数も実装してください。

# 必要なライブラリ: hashlib, os
import hashlib
import os

def generate_salt():
    # ここにソルト生成のコードを記述してください。

def hash_password(password, salt):
    # ここにパスワードをハッシュ化するコードを記述してください。

def verify_password(stored_password_hash, input_password, salt):
    # ここに入力されたパスワードが正しいかどうかを検証するコードを記述してください。

# 実装した関数の動作テストコードも記述してください。


練習問題2.

与えられた文字列を暗号化し、その暗号化された文字列を元に戻す(復号化する)関数を作成してください。

# 必要なライブラリ: cryptography
from cryptography.fernet import Fernet

def generate_key():
    # ここに鍵生成のコードを記述してください。

def encrypt_message(message, key):
    # ここにメッセージを暗号化するコードを記述してください。

def decrypt_message(encrypted_message, key):
    # ここに暗号化されたメッセージを復号化するコードを記述してください。

# 実装した関数の動作テストコードも記述してください。


練習問題3.

ユーザーからの入力を受けてデータベースに問い合わせを行うプログラムを作成してください。SQLインジェクション攻撃を防ぐための対策を施してください。

# 必要なライブラリ: sqlite3
import sqlite3

def search_user(username):
    # ここにユーザー名を使ってデータベースを検索する安全なコードを記述してください。

# 実装した関数の動作テストコードも記述してください。攻撃シナリオも考慮してください。