元々GitHub Copilotでコーディングしていたのですが、6月の料金体系の変更でちょっと不便になってしまい、ついに「Claude Code」に手をだしてみました!!
お値段的には少し高くなりましたが、Claude Proでも十分に賢いし、トークン使用量を抑えるコツ(tips)もいろいろ用意されているので、十分な量が使えますね。もっと使い込んでいくとトークンが足りなくなるかもしれませんが、今のところは満足です。
Claude Codeだと、事前に設計を相談しながら進める「仕様駆動開発」がおすすめされていますが、これがかなり勉強になります。小さいツールとかを作るときに、できた仕様書にセキュリティ上の懸念がないかチェックさせると、すごく鋭い指摘が返ってきます。そのときに「なぜ危険なのか」の解説までしてくれるのですが、攻撃者の方はいろいろなことを考えるなと感心してしまいます。
そんな感じで、練習で作ったツールの仕様を試しに貼ってみます。もっと使いこなせたら、より良いものになるのかなと。
LogSuke 設計書(ログ参照API)
1. 目的と前提
- Playwright(テスト端末)から、Webアプリサーバ上のログをHTTP経由で取得して検証する。
- Webアプリはパッケージ製品で改修できないため、アプリとは別の独立プロセス(サイドカー) としてサーバに常駐させる。
- 初期対象は Linux の syslog。今後対象ログが増えることを前提に、ログ定義を設定ファイルで追加するだけで拡張できる構成にする。
- 端末からは サーバ側で定義したログIDしか指定できない(ファイルパスは一切受け付けない)。
[テスト端末] [Webアプリサーバ]
Playwright ──HTTP / HTTPS──▶ LogSuke (読み取り専用ユーザ)
(API Key、設定で切替)
│ 設定: /etc/logsuke/config.json
├─▶ /var/log/syslog (id: syslog)
├─▶ /opt/app/logs/app.log (id: app) ※将来追加
└─▶ journald (id: ...) ※将来追加
2. 技術選定
| 項目 | 選定 | 理由 |
|---|---|---|
| 言語 | Python 3.9 以上(標準ライブラリのみ、pip 不要) | サーバへの導入はファイル配置のみで済む |
| HTTP / HTTPS |
http.server.ThreadingHTTPServer + ssl
|
小規模用途では十分。標準は HTTP、設定で HTTPS に切り替え可能(3章 server.tls 参照) |
| 設定形式 | JSON(json) |
YAML は PyYAML が必要になるため使わない |
| 認証・IP制限 |
hmac.compare_digest / ipaddress
|
標準ライブラリで実装できる |
| 常駐 | systemd サービス | 自動起動・再起動・権限制御が容易 |
| 実行ユーザ | 専用ユーザ logsuke(adm グループ所属) |
syslog の読み取りのみ可能。書き込み権限なし |
3. ログ定義(サーバ側設定)
/etc/logsuke/config.json(root所有、logsuke は読み取りのみ。systemd を使わずに手動で起動する場合は任意の場所でよい。8.2)
{
"server": {
"bind": "0.0.0.0",
"port": 8765,
"api_keys_file": "/etc/logsuke/api_keys",
"allow_ips": ["192.168.0.0/24"],
"tls": {
"enabled": false,
"cert_file": "/etc/logsuke/tls/server.crt",
"key_file": "/etc/logsuke/tls/server.key"
}
},
"limits": {
"max_connections": 16,
"request_timeout_sec": 30,
"max_scan_bytes": 52428800,
"max_line_bytes": 16384,
"max_response_bytes": 5242880,
"max_contains": 5,
"max_contains_length": 200,
"max_tail_lines": 1000
},
"logging": {
"level": "INFO",
"file": null
},
"logs": {
"syslog": {
"type": "file",
"path": "/var/log/syslog",
"rotated_glob": "/var/log/syslog.[0-9]*",
"description": "OS syslog",
"max_lines": 5000
}
},
"processes": {
"httpd": {
"description": "Apache",
"comm": "httpd",
"min_count": 1
},
"appserver": {
"description": "パッケージアプリ本体",
"comm": "java",
"cmdline_contains": "com.example.AppMain",
"user": "appuser",
"min_count": 1,
"max_count": 1
}
}
}
processes の各項目(監視対象プロセスの定義)
| キー | 必須 | 内容 |
|---|---|---|
| comm | ○ | 実行ファイル名(/proc/<pid>/comm と完全一致で比較) |
| cmdline_contains | - | コマンドラインに含まれる文字列。java のように同じ名前のプロセスが複数あるときに絞り込むために使う |
| user | - | 実行ユーザ名 |
| min_count / max_count | - | 正常とみなすプロセス数の範囲(既定は 1 以上、上限なし) |
logs の各項目
| キー | 必須 | 内容 |
|---|---|---|
| type | ○ |
file(将来 journal 等を追加) |
| path | ○ | 現在書き込まれているログファイル。環境によっては /var/log/messages など |
| rotated_glob | - | ローテート後のファイルを探すパターン(glob)。ローテートを跨いで読むときに使う |
| description | - | 一覧に表示する説明 |
| max_lines | - | 1リクエストで返す行数の上限 |
rotated_glob の設定例(ローテート後のファイル名は環境の logrotate の設定によって異なる)
| logrotate の方式 | ローテート後の名前 | rotated_glob |
|---|---|---|
番号付き(delaycompress) |
syslog.1、syslog.2.gz … |
/var/log/syslog.[0-9]* |
日付付き(dateext、圧縮なし) |
messages-20261021 … |
/var/log/messages-* |
- パターンに一致したファイルのうち、圧縮ファイル(
.gz/.xz/.bz2/.zst)は対象外にする。 -
rotated_globに一致するファイルもpathと同じディレクトリにあることを起動時に検証し、パターンは管理者だけが設定できる(端末からは指定できない)。
server の各項目
| キー | 必須 | 既定値 | 内容 |
|---|---|---|---|
| bind | ○ | - | 待ち受けアドレス |
| port | ○ | - | 待ち受けポート |
| api_keys_file | ○ | - | API Key ファイル(下記) |
| allow_ips | - |
["0.0.0.0/0", "::/0"](全て許可) |
接続を許可する送信元(CIDR)。省略した場合は全て許可とし、起動時に WARNING を出す。明示的に 0.0.0.0/0 / ::/0 を指定した場合も同じ WARNING を出す。空のリスト [] は設定ミスとみなして起動時エラー |
| tls | - | 無効 | HTTPS 設定(下記) |
-
allow_ipsを省略しても API Key による認証は必ず行う。可能な限りテスト端末のネットワークに絞ることを推奨する(FW・systemd のIPAddressAllowでも制限できる。8章)。
api_keys_file(API Key ファイル)の書式
# 名前:キー(1行に1つ。# で始まる行と空行は無視)
pc-tanaka:2cB0q9...(32文字以上のランダム文字列)
ci-runner:Xr7T1m...
-
名前:英数字・
_・-で1〜32文字。アクセスログには名前だけを記録し、キーそのものは記録しない。 -
キー:32文字以上。
python3 -c "import secrets; print(secrets.token_urlsafe(32))"で生成する。総当たりで当てられない長さにすることで、認証失敗の回数制限は設けない。 - 名前の重複、キーの重複、32文字未満のキー、
:のない行がある場合は起動時にエラーにする。 - ファイルは
root:logsuke 0640とする(手動起動の場合は実行ユーザ所有の0600)。権限は下の「秘密ファイルの権限チェック」で起動時に確認する。 - キーの入れ替え手順:新しい行を追加 → 再起動 → 端末側の
LOGSUKE_API_KEYを差し替え → 古い行を削除 → 再起動。名前ごとに独立しているので、1人分だけ止めることもできる。
limits の各項目(リソース保護。省略時は上の値が既定値)
| キー | 既定値 | 内容 | 超えたときの動作 |
|---|---|---|---|
| max_connections | 16 | 同時に処理する接続(スレッド)数 | 接続を受け付けた時点で判定し、上限以上ならスレッドを作らずにすぐ切断する(応答は返さない) |
| request_timeout_sec | 30 | ソケットの読み書きタイムアウト。遅いクライアントに接続を占有されないようにする | 接続を切断 |
| max_scan_bytes | 50MB | 1回の since で読むファイルの量 |
そこで打ち切り truncated: true。next_cursor から続きを読める |
| max_line_bytes | 16KB | 1行の長さ | 行の末尾を切り捨てて …[truncated] を付ける |
| max_response_bytes | 5MB | 応答本文の大きさ | 収まる行までで打ち切り truncated: true
|
| max_contains | 5 |
contains の個数 |
400 |
| max_contains_length | 200 |
contains 1つの長さ |
400 |
| max_tail_lines | 1000 |
tail の lines の上限 |
上限値に丸める |
- この他、HTTP の仕様として URL は 4096 バイト以下とし、超えた場合は
414を返す。リクエスト本文は受け付けない(GET のみ)。 -
max_connectionsの判定はThreadingHTTPServer.process_request()を上書きして行う。稼働中のスレッド数を数え、上限以上ならソケットを閉じるだけでスレッドを作らない。ハンドラの中で判定する方式と違い、大量の接続を受けてもスレッドが増え続けない。 - HTTP/1.0(keep-alive なし)で応答し、1接続1リクエストとする。待機中の接続に枠を占有されないようにするため。
- 切断された場合、クライアント(6.1)は短い間隔で数回再試行する。
- 値は設定で変更できるが、極端な値(例:
max_connectionsが 0 以下や 256 超)は起動時にエラーにする。
logging の各項目(LogSuke 自身の処理ログ。5章の 9 参照)
| キー | 既定値 | 内容 |
|---|---|---|
| level | INFO |
DEBUG / INFO / WARNING / ERROR
|
| file | null |
null の場合は標準エラー出力に出す(systemd 経由で journald に入る)。パスを指定した場合はそのファイルに出し、RotatingFileHandler で 10MB × 5 世代にローテートする |
server.tls の各項目(HTTPS 設定)
| キー | 必須 | 内容 |
|---|---|---|
| enabled | - |
true で HTTPS、false(既定)で HTTP |
| cert_file | enabled 時 ○ | サーバ証明書(PEM)。開発環境・最終テスト環境では自己署名証明書を想定 |
| key_file | enabled 時 ○ | 秘密鍵(PEM、パスフレーズなし)。root:logsuke 0640(手動起動の場合は 0600)。下の「秘密ファイルの権限チェック」で確認する |
-
tlsを省略した場合は HTTP で起動する。
秘密ファイルの権限チェック(api_keys_file・tls.key_file に適用。起動時)
- 所有者が root または LogSuke を実行しているユーザである。
- 「その他(other)」に読み・書きの権限がない(
o-rw)。 - グループに書き込み権限がない(
g-w)。
- systemd で動かす場合の
root:logsuke 0640も、手動起動の場合の<実行ユーザ>:<グループ> 0600も、このチェックを通る。 - 設定ファイル(
config.json)は秘密情報を含まないため、2 の「読み」は許可する。ただし、その他・グループに書き込み権限がある場合はエラーにする。 - HTTP で起動したときは「API Key とログが平文で流れる」旨の警告を起動ログに1回出す。
HTTPS の運用(自己署名証明書)
標準は HTTP とし、server.tls.enabled を true にすると HTTPS で起動する。開発環境・最終テスト環境では自己署名証明書(いわゆるオレオレ証明書)を使う想定とする。
-
ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)で待ち受けソケットをラップする。 - TLS 1.2 以上に限定する。暗号スイートは Python と OpenSSL の既定値を使う。
- 証明書・秘密鍵が読めない、ペアが一致しない、秘密鍵の権限が緩い場合は、起動時にエラーで停止する。
- 証明書の更新はファイルを差し替えて
systemctl restart logsukeで反映する(実行中の再読み込みはしない)。
証明書の作成例(INSTALL.md に記載。サーバ上で root で実行)
mkdir -p /etc/logsuke/tls && cd /etc/logsuke/tls
openssl req -x509 -newkey rsa:2048 -nodes -days 825 \
-keyout server.key -out server.crt \
-subj "/CN=web01.test.local" \
-addext "subjectAltName=DNS:web01.test.local,IP:192.168.0.50"
chown root:logsuke server.key server.crt
chmod 0640 server.key && chmod 0644 server.crt
-
subjectAltNameは必須。Python の HTTPS クライアントは CN ではなく SAN でホスト名を照合するため、テスト端末から接続するときのホスト名か IP アドレスを必ず入れる。 -
-addextは OpenSSL 1.1.1 以降で使える。現在の主要な Linux ディストリビューションは対応済み。 -
作成した
server.crt(秘密鍵ではない)をテスト端末にコピーし、LOGSUKE_CA_FILEに指定する。 -
端末が指定できるのは
syslogなどの ID のみ。 -
未定義IDは
404。パスは設定ファイル以外から決して組み立てない。 -
typeを分けておくことで、journald 等も読み取りクラスを1つ追加するだけで対応可能にする。
4. API仕様
全API共通:ヘッダ X-API-Key 必須。不一致は 401、許可外IPは 403。
エラー応答の形式(全API共通)
{"error": "not_found", "message": "unknown log id", "request_id": "6f1c2a9e"}
-
messageは定型文だけとし、スタックトレース・ファイルパス・例外メッセージは返さない。 - 予期しない例外は
500("error": "internal_error")とし、詳細は LogSuke 自身の処理ログにだけ出す(5章の 9)。 -
request_idは処理ログと突き合わせるための番号。テストが失敗したときは、この値で処理ログを検索する。
| ステータス | error | 主な原因 |
|---|---|---|
| 400 | bad_request |
不正な cursor、数値でない limit、contains の個数・長さ超過 |
| 401 | unauthorized |
API Key なし・不一致 |
| 403 | forbidden |
許可されていない IP |
| 404 | not_found |
未定義のID・パス |
| 405 | method_not_allowed |
GET 以外 |
| 414 | uri_too_long |
URL が長すぎる |
| 500 | internal_error |
予期しない例外 |
| 503 | log_unavailable |
ログファイルが読めない |
4.1 ログ一覧
GET /logs → 参照可能なログIDの一覧(パスは返さない)
[{"id": "syslog", "description": "OS syslog"}]
4.2 カーソル取得(テスト開始時の「しおり」)
GET /logs/{id}/cursor
{"id": "syslog", "cursor": "inode:1234567:offset:884213"}
アプリにリクエストIDを埋め込めないため、「操作前にカーソルを取り、操作後にその差分だけ読む」 方式でテストごとのログを切り出す。
4.3 差分取得(主に使うAPI)
GET /logs/{id}/since?cursor=...&contains=...&limit=...
| パラメータ | 必須 | 内容 |
|---|---|---|
| cursor | ○ | 4.2 で取得した値 |
| contains | - | 部分一致フィルタ(複数指定でAND)。正規表現は ReDoS 回避のため初期は不可 |
| limit | - | 返却行数(max_lines が上限) |
{
"id": "syslog",
"lines": ["Oct 21 20:41:02 web01 app[123]: login ok user=test01"],
"next_cursor": "inode:1234567:offset:884390",
"truncated": false,
"rotated": false,
"missed": false,
"truncated_reason": null
}
-
truncated: trueのとき、truncated_reasonにmax_lines/max_scan_bytes/max_response_bytesのどれで打ち切ったかを入れる。next_cursorは「最後に読んだ位置の直後」を指すので、同じnext_cursorでsinceを呼び直せば続きを読める。 -
missed: trueは、ローテートを跨いだときに一部を読めなかった可能性があることを示す(下記)。
ローテートの検知と読み取り手順
- カーソルの inode が
pathの現在の inode と同じであれば、通常どおり offset から読む。- サイズが offset より小さくなっていた場合は copytruncate 方式で切り詰められたと判断し、先頭から読んで
rotated: trueとする。
- サイズが offset より小さくなっていた場合は copytruncate 方式で切り詰められたと判断し、先頭から読んで
- inode が違う場合は、ローテートされたと判断する。
-
rotated_globに一致するファイル(圧縮ファイルを除く)を inode で照合し、カーソルの inode と同じファイルを探す。ファイル名では判断しない。 - 見つかった場合は、そのファイルの offset 以降を読み、それより新しいローテート済みファイル(更新日時順)を続けて読み、最後に現在の
pathの先頭から読む。rotated: trueとする。 - 見つからない場合(圧縮済み・削除済み・
rotated_glob未設定)は、現在のpathの先頭から読み、rotated: true, missed: trueとする。
-
- ファイルを開くのは、常に
pathとrotated_globに一致したファイルだけとする。カーソルの inode を使ってファイルを直接開くことはしない。 - ファイルを開くときは、5章の 10「ファイルの安全な開き方」に従う。inode の照合も、安全に開けたファイルに対してだけ行う。
- この方式であれば、番号付き方式でも日付付き方式でも、同じ処理で対応できる。
- テストは数秒〜数分で終わるため、通常は「1回分のローテートを跨ぐ」ことしか起きない。複数回跨いでも上の手順で読める。
4.4 末尾取得(調査用)
GET /logs/{id}/tail?lines=100 — 失敗時の添付用。
4.5 プロセス一覧
GET /processes → 確認できるプロセスIDの一覧(判定条件は返さない)
[{"id": "httpd", "description": "Apache"}]
4.6 プロセス状態の確認
GET /processes/{id}
{
"id": "appserver",
"ok": true,
"count": 1,
"expected": {"min": 1, "max": 1},
"processes": [
{"pid": 4321, "user": "appuser", "started_at": "2026-10-21T20:30:12+09:00",
"uptime_sec": 1830, "rss_kb": 512340, "state": "S"}
]
}
-
okはcountがmin_count以上max_count以下のときにtrueになる。プロセスが0件でも、条件どおりであれば HTTP は200で返す(判定結果はokで表す)。 -
psコマンドは呼ばず、/procを直接読んで判定する(標準ライブラリのみで実装でき、コマンドを外部から実行される余地がない。出力形式が環境によって変わる心配もない)。 - コマンドラインは返さない。パスワードなどが引数に含まれている可能性があるため、判定にだけ使う。
- 再起動を検知するには
pidとstarted_atを比較する。例:テストの前後で値が変わっていれば、途中でプロセスが落ちて再起動されたと判断できる。 - 一覧取得(
GET /processes?all=1)と、プロセスの起動・停止を行う API は作らない(参照のみ)。
4.7 ヘルスチェック
GET /health
テストセッションの開始時に1回呼び、「接続できるか」「認証が通るか」「ログが読めるか」をまとめて確認する。
ほかのAPIと同じく、API Key と IP 制限の対象とする(認証の確認も兼ねるため)。
{
"status": "ok",
"version": "1.0.0",
"server_time": "2026-10-21T20:41:02+09:00",
"tls": false,
"key_name": "pc-tanaka",
"logs": [{"id": "syslog", "readable": true}],
"processes": [{"id": "httpd"}, {"id": "appserver"}]
}
| 項目 | 内容 |
|---|---|
| status | 全ログが読めれば ok、1つでも読めなければ degraded。どちらの場合も HTTP は 200
|
| version | LogSuke のバージョン。クライアントとの組み合わせ確認用 |
| server_time | サーバの現在時刻(タイムゾーン付き)。テスト端末との時刻のずれの調査用 |
| tls | HTTPS で動作しているか |
| key_name | 認証に使われた API Key の名前。端末側の設定ミスの確認用 |
| logs[].readable | ログファイル(path)を開けるか。ローテート後に ACL が外れた場合などを早く検知する |
| processes | 定義されているプロセスID。プロセスの状態は返さない(/processes/{id} で確認する) |
- ファイルパス、ホスト名、OS や Python のバージョン、
readable: falseの理由(例外メッセージ)は返さない。理由は処理ログに WARNING で出す。 -
readableはファイルを開けるかの確認だけで、中身は読まない(負荷をかけないため)。
5. セキュリティ方針
-
IDホワイトリスト方式:パス・ファイル名・ワイルドカードを受け付けない。
-
読み取り専用:専用ユーザで実行し、書き込み・削除APIは作らない。systemd で
ProtectSystem=strict/NoNewPrivileges=yesを設定。 -
認証:API Key(名前付き、定数時間比較)+送信元IP制限。可能ならFWでテスト端末のみ許可。
-
IP制限は、接続元ソケットのアドレス(
client_address[0])だけで判定する。X-Forwarded-For/X-Real-IP/Forwardedなどのヘッダは一切参照しない(端末が自由に書き換えられるため)。 - そのため、リバースプロキシ経由の構成はサポートしない。テスト端末から LogSuke に直接接続する。
- IPv4射影IPv6アドレス(
::ffff:192.168.0.10)は IPv4 アドレスに直してから判定する。 - 標準は HTTP のため、API Key とログ内容は平文で流れる。テスト用の閉じたネットワーク以外を通る場合は、設定で HTTPS を有効にする(3章
server.tls)。
-
IP制限は、接続元ソケットのアドレス(
-
量の制限:同時接続数・タイムアウト・走査量・行の長さ・応答サイズ・
containsの個数と長さに上限を設ける(3章limits)。上限を超えても、サーバは止まらずに打ち切る・4xxを返す・接続を切断するのいずれかで対応する。 -
テスト環境専用:導入するのは開発環境と最終テスト環境(いずれも Linux)だけとし、利用者向けの本番サーバには導入しない。本番環境の構築手順書やイメージに LogSuke が混ざらないよう、INSTALL.md に明記する。
-
アクセスログ:誰(API Key の名前)がどのIDを読んだかを LogSuke 自身の処理ログに記録する。キーそのものは記録しない。
-
プロセス確認も同じ方式:確認できるのは
processesに定義したIDだけにする。プロセス名などを端末から指定する検索はできない。- 他ユーザのプロセスを
/procで見えるようにするため、unit ファイルにProtectProc=invisibleは設定しない。 -
/procがhidepid付きでマウントされている場合は、gid=オプションで指定されたグループにlogsukeを追加する。
- 他ユーザのプロセスを
-
サーバ情報を出さない・応答ヘッダ:
ServerヘッダはLogSukeだけにする(server_version = "LogSuke"、sys_version = "")。http.server既定の HTML エラーページは使わず、send_errorを上書きして必ず JSON のエラー応答を返す。全ての応答(エラーを含む)に次のヘッダを付ける。ヘッダ 値 目的 Content-Typeapplication/json; charset=utf-8形式の明示 Cache-Controlno-storeログの内容を途中のキャッシュや端末に残さない X-Content-Type-Optionsnosniffブラウザが JSON を別の形式と解釈しないようにする -
CORS ヘッダ(
Access-Control-Allow-*)は付けない。OPTIONSも405とする。これにより、テスト中のブラウザ上のページ(Webアプリや外部サイト)の JavaScript から LogSuke を呼び出せない。
-
CORS ヘッダ(
-
処理ログ:LogSuke 自身の動作を Python 標準の
loggingで出力する。種類 レベル 内容 起動・停止 INFO バージョン、待ち受けアドレス、HTTP/HTTPS、読み込んだログID・プロセスIDの一覧、HTTP 時の平文警告 アクセス INFO 時刻、 request_id、接続元IP、API Key の名前、メソッド、パス、ステータス、応答バイト数、処理時間認証失敗・拒否 WARNING 接続元IP、理由(キー不一致・IP拒否・上限超過)。送られてきたキーの値は出さない 予期しない例外 ERROR request_idとスタックトレース(logger.exception)設定エラー ERROR 起動を中止した理由(どのキーが不正か) - アクセスログの
containsはそのまま記録する。テストで使う検索文字列なので、秘密情報を入れない運用とする。 - 出力先は 3章の
logging.fileで変える。既定は標準エラー出力で、journalctl -u logsukeで確認できる。 -
http.server既定のアクセスログ(log_message)は上書きして、この形式にそろえる。 -
ログインジェクション対策:パス、クエリ(
containsなど)、ヘッダ由来の値など、端末から送られた値を処理ログに書くときは、制御文字(改行\n・\r、タブ、ESC など0x00〜0x1fと0x7f)を\nや\x1bの形にエスケープする。これにより、1リクエストのログは必ず1行になり、偽の行を書き込まれない。また1項目あたり 512 文字で切り詰める。エスケープはlogging.Formatterではなく、値を埋め込む関数(safe_for_log())でまとめて行う。
- アクセスログの
-
ファイルの安全な開き方(シンボリックリンク対策):ログファイル(
pathとrotated_globに一致したファイル)は次の手順でだけ開く。将来アプリのログを対象にすると、そのディレクトリにはアプリのユーザが書き込めることが多い。アプリが乗っ取られた場合に、app.log.1 -> /etc/logsuke/tls/server.keyのようなリンクを置かれて、LogSuke が読めるファイルを読み出されるのを防ぐ。-
os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)で開く。最後の要素がシンボリックリンクなら開けずにエラーになる。O_NONBLOCKは FIFO を開いたときに止まらないようにするため。 - 開いたファイルを
os.fstat()で調べ、通常のファイル(S_ISREG)以外は拒否する(ディレクトリ・FIFO・デバイスなど)。 - 開いたファイルの実体のパスを
os.path.realpath(f"/proc/self/fd/{fd}")で取得し、設定したディレクトリ(pathの親ディレクトリをrealpathで解決したもの)の直下にあるかを確認する。途中のディレクトリがリンクに差し替えられた場合も、ここで検出できる。 - いずれかに該当した場合はファイルを閉じて読み取らず、そのファイルは「存在しない」として扱う(
sinceはmissed: true、path自体なら503 log_unavailable)。処理ログには WARNING で理由とパスを出す(応答には出さない)。
- 起動時には、
api_keys_file・tls.cert_file・tls.key_file・設定ファイルが、どのログのディレクトリとも別の場所にあることを確認し、同じディレクトリなら起動時エラーにする。 -
rotated_globの検索結果も、上の 1〜3 を通ったファイルだけを照合の対象にする。
-
6. Playwright 側(Python)のクライアント
Playwright は Python 版(playwright / pytest-playwright)で動かす前提とする。
LogSuke を呼び出すクライアントは client/logsuke_client.py の1ファイルにまとめ、標準ライブラリ(urllib)だけで実装する。Playwright 自体には依存しないため、pytest 以外からも使える。
6.1 クライアント仕様
LogSukeClient(base_url, api_key, timeout=10.0, ca_file=None, insecure=False)
- 引数を省略した場合は、環境変数から読む。API Key をテストコードに直接書かないようにするため。
| 引数 | 環境変数 | 内容 |
|---|---|---|
| base_url | LOGSUKE_URL |
http://... または https://...。スキームで HTTP / HTTPS が決まる |
| api_key | LOGSUKE_API_KEY |
API Key |
| ca_file | LOGSUKE_CA_FILE |
HTTPS 時に信頼する証明書(自己署名証明書をそのまま指定する) |
| insecure | LOGSUKE_INSECURE=1 |
証明書の検証をしない。一時的な切り分け専用。使用時は毎回警告を出す |
| メソッド | 対応API | 戻り値 |
|---|---|---|
health() |
GET /health |
dict(4.7 の内容) |
assert_healthy() |
health を呼んで判定 |
status が ok でなければ AssertionError(読めないログIDをメッセージに含める) |
list_logs() |
GET /logs |
list[dict] |
cursor(log_id) |
GET /logs/{id}/cursor |
str(カーソル) |
since(log_id, cursor, contains=None, limit=None) |
GET /logs/{id}/since |
dict(lines, next_cursor, truncated, rotated, missed) |
tail(log_id, lines=100) |
GET /logs/{id}/tail |
list[str] |
wait_for_log(log_id, cursor, contains, timeout=10, interval=0.5) |
since を繰り返す |
一致した行の list[str]。期限までに見つからなければ LogWaitTimeout
|
assert_no_log(log_id, cursor, contains, wait=3.0) |
wait 秒待ってから since
|
一致する行がなければ何も返さない。あれば AssertionError(一致した行を先頭 20 行までメッセージに含める) |
list_processes() |
GET /processes |
list[dict] |
process(proc_id) |
GET /processes/{id} |
dict(ok, count, processes) |
assert_process_ok(proc_id) |
process を呼んで判定 |
ok が false なら AssertionError(メッセージに件数と期待範囲を含める) |
-
containsは文字列でもリストでもよい。リストの場合は AND 条件で絞り込む。 - エラー時の例外
-
LogSukeError:HTTP ステータスとサーバのメッセージを持つ。 -
LogSukeAuthError:401 / 403 のとき。設定ミスにすぐ気付けるよう分けておく。 -
LogWaitTimeout:wait_for_logが期限切れになったとき。
-
- 通信の安全対策
-
リダイレクトには従わない(
urllibは既定でリダイレクト先にもX-API-Keyを送るため)。3xx はLogSukeErrorにする。 -
プロキシは既定で使わない(環境変数
HTTP_PROXY等を無視する)。社内プロキシのログに API Key が残るのを防ぐため。必要な場合のみLOGSUKE_USE_PROXY=1で有効にする。 - サーバに接続を切断された(同時接続数の上限)場合は、0.2 秒・0.5 秒・1 秒の間隔で最大 3 回再試行する。
-
リダイレクトには従わない(
- Python 版 Playwright には TypeScript 版の
expect.poll相当がないため、ログの書き込み遅延はwait_for_logで吸収する。 - 「出ていないこと」の確認は、ログの書き込み遅延があるため、すぐに
sinceを呼ぶと見逃す。assert_no_logはwait秒待ってから確認する。waitはアプリの書き込み遅延に合わせて調整する。
6.2 使い方(pytest-playwright)
conftest.py
import pytest
from logsuke_client import LogSukeClient
@pytest.fixture(scope="session")
def logsuke():
client = LogSukeClient() # LOGSUKE_URL / LOGSUKE_API_KEY を使用
client.assert_healthy() # 接続・認証・ログの読み取り権限を最初に1回確認
return client
@pytest.fixture
def syslog_cursor(logsuke):
return logsuke.cursor("syslog") # テストごとに操作前のカーソルを取る
テスト
def test_login(page, logsuke, syslog_cursor):
before = logsuke.process("appserver")
page.goto("https://web01.example/login")
page.get_by_role("button", name="ログイン").click()
# アプリ側のログに出力されたことを確認(書き込み遅延はリトライで吸収)
logsuke.wait_for_log("syslog", syslog_cursor, contains="login ok")
# 異常が出ていないこと(書き込み遅延を考慮して少し待ってから確認)
logsuke.assert_no_log("syslog", syslog_cursor, contains="ERROR")
# プロセスが落ちていないこと(再起動されていないこと)
after = logsuke.process("appserver")
assert after["ok"]
assert [p["pid"] for p in after["processes"]] == [p["pid"] for p in before["processes"]]
-
pytest-xdistで並列実行すると他のテストのログが混ざる。そのため、ログを検証するテストは並列化しない(マーカーで分けて直列実行する)か、ユーザIDなどテスト固有の値でcontainsを使って絞り込む。
6.3 HTTPS の利用(クライアント側)
サーバ側の設定と証明書の作成は 3章「HTTPS の運用」を参照。
-
ca_fileを指定すると、その証明書だけを信頼して検証する。ホスト名の照合も行う。 - OS の証明書ストアへの登録は不要で、テスト端末の他の通信には影響しない。
-
insecure=Trueは検証を無効にするため、証明書の切り分けのときだけ使う。常用しない。
6.4 pytest プラグイン(失敗時のログ添付)
client/logsuke_pytest.py(150行程度)。テストが失敗したときに、そのテストの実行中に出たログ(テスト開始時のカーソル以降)を pytest のレポートに自動で添付する。pytest のプラグインのため pytest を import するが、テスト端末には pytest が既にあるので追加の pip は不要。
有効化(テスト側の conftest.py)
pytest_plugins = ["logsuke_pytest"]
設定(pytest.ini / pyproject.toml)
[pytest]
logsuke_attach_logs = syslog # 添付するログID(カンマ区切り。空なら添付しない)
logsuke_attach_max_lines = 200 # 1ログあたりの最大行数(末尾を残す)
コマンドラインの --logsuke-attach=syslog,app でも上書きできる。
提供する fixture とフック
| 名前 | 種類 | 内容 |
|---|---|---|
logsuke |
fixture(session) |
LogSukeClient を作り、assert_healthy を1回行って返す |
logsuke_cursors |
fixture(function、autouse) |
テスト開始時に、logsuke_attach_logs の各ログのカーソルを取得して保持する。テストからも logsuke_cursors["syslog"] で使える |
pytest_runtest_makereport |
フック | テスト本体(call)またはセットアップで失敗したとき、保持したカーソル以降のログを since で取得し、report.sections に ("logsuke: syslog", 内容) として追加する |
- 追加したセクションは、pytest の失敗表示(
-rAなど)と pytest-html の詳細欄にそのまま表示される。ファイルにも残したい場合は、--logsuke-attach-dir=<dir>を指定すると<テスト名>_<ログID>.logで保存する。 - 行数が
logsuke_attach_max_linesを超える場合は末尾を残し、先頭に「… N 行省略」と書く。truncated/missedの場合もその旨を書く。 - 添付の取得に失敗しても(LogSuke 停止など)、テストの結果は変えない。セクションに「ログを取得できませんでした: 理由」と書くだけにする。
- 成功したテストでは添付しない(レポートを大きくしないため)。
6.2 の conftest.py の例は、このプラグインを使うと次のように短くなる。
pytest_plugins = ["logsuke_pytest"]
def test_login(page, logsuke, logsuke_cursors):
page.goto("https://web01.example/login")
page.get_by_role("button", name="ログイン").click()
logsuke.wait_for_log("syslog", logsuke_cursors["syslog"], contains="login ok")
logsuke.assert_no_log("syslog", logsuke_cursors["syslog"], contains="ERROR")
7. ディレクトリ構成(予定)
小規模ツールのため、Python モジュールは 3ファイル に限定します。目安として、1ファイル 500行以下、1関数 50行以下にします。コードの書き方は 11章「コーディング規約」に従います。
LogSuke/
├── logsuke.py # 本体:設定・認証・HTTP・処理ログ・起動(想定 300〜350行)
├── log_service.py # ログ読み取り処理(想定 150〜200行)
├── process_service.py # プロセス確認処理(想定 100〜150行)
├── config.sample.json # 設定サンプル
├── logsuke.service # systemd unit(Python モジュールではない)
├── client/logsuke_client.py # Playwright(Python) 用クライアント(200行程度、標準ライブラリのみ)
├── client/logsuke_pytest.py # pytest プラグイン:fixture と失敗時のログ添付(150行程度)
└── tests/ # unittest(標準ライブラリ)。詳細は「10. テスト設計」
各モジュールの役割
| モジュール | 主な関数・クラス | 役割 |
|---|---|---|
logsuke.py |
load_config(), validate_config()
|
JSON の読み込み、必須項目と型の検証 |
load_api_keys(), check_api_key(), check_ip()
|
名前付き API Key の読み込みと検証、定数時間比較(一致した名前を返す)、接続元ソケットのIPによる判定 | |
Handler(BaseHTTPRequestHandler)、do_GET からルーティング |
共通チェック(同時接続数・認証・URL 長)を行い、各 service を呼び出して JSON で応答。send_error / log_message を上書きする |
|
setup_logging(), safe_for_log()
|
処理ログの出力先とレベルを設定する。端末から来た値の制御文字をエスケープする | |
create_server(), build_ssl_context(), LimitedThreadingHTTPServer
|
サーバの生成。tls.enabled のときだけ SSL コンテキストでソケットをラップする。process_request() を上書きして同時接続数を制限する |
|
main()(argparse で --config を受け取る) |
サーバの起動とシグナル処理 | |
log_service.py |
encode_cursor(), decode_cursor()
|
inode:offset 形式の文字列と値の相互変換 |
FileReader(cursor / since / tail) |
type: file の実装。将来は同じメソッドを持つ JournalReader を追加し、READERS = {"file": FileReader} に登録する |
|
list_logs(), get_reader()
|
設定の logs からIDで読み取りクラスを取り出す |
|
check_readable() |
ヘルスチェック用に、各ログファイルを開けるかだけを確認する | |
open_log_file() |
5章の 10 の手順でファイルを安全に開く(全ての読み取りはこの関数を通す) | |
process_service.py |
scan_processes(), read_proc_info()
|
/proc を走査し、各プロセスの情報を読む |
list_processes(), check_process()
|
設定の processes の条件で集計し、ok を判定する |
依存関係は logsuke.py → log_service.py / process_service.py の一方向だけにします。2つの service は HTTP を知らず、辞書を受け取って辞書を返すだけにするため、単体でテストできます。
8. 対象環境
LogSuke を導入するのはテスト用の環境(Linux)だけで、利用者向けの本番サーバは対象外とする。
- 開発環境(Linux):LogSuke とテストシナリオの作成に使う。
- 最終テスト環境(Linux):本番と同じ構成のテスト環境。Playwright によるテストは最終的にここで実行する。
環境(Linux の種類や設定)によって異なる主な項目
| 項目 | 内容 |
|---|---|
| Python | 3.9 以上。標準の Python が 3.9 の環境もあるため、3.9 の文法の範囲で実装する |
| syslog | rsyslog が既定で導入されていない環境では、事前に導入が必要 |
| syslog パス・権限 |
/var/log/syslog(root:adm 0640)、/var/log/messages(root:root 0600)など、環境によって異なる |
| 読み取り権限の付与 | グループで読める場合は logsuke をそのグループ(adm 等)に追加する。root だけが読める場合は ACL で付与し、ローテート時に付け直す(下記) |
| SELinux | 有効(Enforcing)な環境では、ポートの許可が必要 |
| FW | 有効な環境では、テスト端末からのポート(既定 8765/tcp)を許可する |
- 環境ごとの差は
config.jsonと導入手順だけで吸収し、logsuke.pyとlogsuke.serviceは共通にする。 - ACL・SELinux が必要な環境での権限付与(例):
setfacl -m u:logsuke:r /var/log/messages # logrotate の rsyslog 用設定の postrotate に同じコマンドを追記する semanage port -a -t http_port_t -p tcp 8765 restorecon -Rv /etc/logsuke # 証明書・鍵を配置した後に SELinux のラベルを戻す - ファイアウォールが有効な環境では、使用しているファイアウォールの手順でポート 8765/tcp を許可する。
-
logsuke.pyを/optに置くと SELinux に拒否される場合がある。その場合は/usr/local/binに置くか、restoreconで対応する。 - 環境ごとの導入手順の差は
INSTALL.mdに1ファイルでまとめる。
8.1 systemd unit(logsuke.service)
アプリの外側で二重に守るため、systemd のサンドボックス機能を使う。systemd 252 以降で使える設定だけにする。
[Unit]
Description=LogSuke log viewer API (test environment only)
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=logsuke
Group=logsuke
SupplementaryGroups=adm
ExecStart=/usr/bin/python3 /opt/logsuke/logsuke.py --config /etc/logsuke/config.json
Restart=on-failure
RestartSec=5
UMask=0077
# 権限
NoNewPrivileges=yes
CapabilityBoundingSet=
AmbientCapabilities=
# ファイルシステム
ProtectSystem=strict
ProtectHome=yes
PrivateTmp=yes
PrivateDevices=yes
# カーネル・システム
ProtectKernelTunables=yes
ProtectKernelModules=yes
ProtectKernelLogs=yes
ProtectControlGroups=yes
ProtectClock=yes
ProtectHostname=yes
RestrictNamespaces=yes
RestrictRealtime=yes
RestrictSUIDSGID=yes
LockPersonality=yes
MemoryDenyWriteExecute=yes
SystemCallArchitectures=native
SystemCallFilter=@system-service
SystemCallFilter=~@privileged
RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
# 資源の上限(コードの不具合があってもサーバ全体を巻き込まない)
MemoryMax=256M
TasksMax=64
CPUQuota=50%
# ネットワーク(任意。allow_ips と同じ範囲を書く)
#IPAddressDeny=any
#IPAddressAllow=192.168.0.0/24 localhost
[Install]
WantedBy=multi-user.target
| 設定 | 目的 |
|---|---|
CapabilityBoundingSet=(空) |
root の特権を一切持たない |
ProtectSystem=strict ほか |
OS 全体を読み取り専用にし、/home・デバイス・カーネル設定を見せない |
SystemCallFilter / RestrictAddressFamilies
|
通常のサービスに必要なシステムコールと通信方式だけを許可する |
MemoryMax / TasksMax / CPUQuota
|
メモリ・スレッド数・CPU の上限。TasksMax は max_connections に余裕を持たせた値にする(起動時に max_connections + 16 > TasksMax なら WARNING) |
IPAddressAllow / IPAddressDeny
|
OS(cgroup)側での IP 制限。コードの allow_ips とは独立して効く。allow_ips を省略して全許可にした場合も、ここで絞れる |
-
ProtectProc=invisibleは設定しない(他ユーザのプロセスを確認できなくなるため。5章の 7)。 -
ProtectSystem=strictでもログの読み取りは可能。処理ログをファイルに出す場合だけLogsDirectory=logsukeを追加する(下記)。 -
設定の効き具合は
systemd-analyze security logsukeで確認できる(INSTALL.md のチェック項目に入れる)。 -
unit ファイルを変えた場合は
systemctl daemon-reloadの後に再起動する。 -
処理ログを
logging.fileでファイルに出す場合は、ProtectSystem=strictのもとで書き込めるよう、unit ファイルにLogsDirectory=logsuke(/var/log/logsukeを自動作成)を設定する。既定の標準エラー出力(journald)なら追加設定は不要。
8.2 systemd を使わない手動起動(SSH ログインして実行)
LogSuke は python3 logsuke.py --config <設定ファイル> だけで動くため、systemd に登録しなくても、SSH でログインしたユーザが手動で起動できる。開発環境での試行や、最終テスト環境で root 権限による導入ができない場合に使う。
前提
- SSH ユーザが、対象のログファイルを読める(
admグループ所属、または ACL で読み取り権限がある)。 - ポートは 1024 以上を使う(root 以外は 1024 未満で待ち受けできないため。既定の 8765 で問題ない)。
- テスト端末からポートへの通信が FW で許可されている(ファイアウォールの設定には root 権限が必要なため、事前に管理者へ依頼する)。
配置と設定(root 不要。ホームディレクトリ等に置く)
mkdir -p ~/logsuke && cd ~/logsuke
# logsuke.py / log_service.py / process_service.py / config.sample.json を配置する
cp config.sample.json config.json # logs・processes・server を編集する
python3 -c "import secrets; print('pc-tanaka:' + secrets.token_urlsafe(32))" > api_keys
chmod 600 api_keys # 権限チェック(3章)を通すため
chmod 644 config.json
-
config.jsonのapi_keys_fileは/home/<ユーザ>/logsuke/api_keysのように絶対パスで書く。相対パスで書いた場合は、設定ファイルのあるディレクトリを基準に解決する(起動したときのカレントディレクトリには依存しない)。 - 設定が正しいかは、起動して
/healthを呼ぶか、起動時のエラーメッセージで確認する。
起動と停止
| 方法 | 起動 | 停止 | 向いている場面 |
|---|---|---|---|
| 前面で実行 | python3 logsuke.py --config config.json |
Ctrl+C |
動作確認。SSH を切ると止まる |
| tmux / screen |
tmux new -s logsuke の中で上のコマンドを実行し、Ctrl+B → D で離脱 |
tmux attach -t logsuke → Ctrl+C
|
テスト期間中に動かし続ける場合(推奨。ログを画面で確認できる) |
| nohup |
nohup python3 logsuke.py --config config.json >> logsuke.out 2>&1 &echo $! > logsuke.pid
|
kill $(cat logsuke.pid) |
tmux / screen が入っていない場合 |
-
Ctrl+C(SIGINT)とkill(SIGTERM)のどちらでも、処理中のリクエストを終えてから正常に終了する。 - 処理ログは標準エラー出力に出る。nohup の場合は
logsuke.outに入る。ファイルに分けたい場合はlogging.fileに絶対パスを書く(10MB×5世代でローテートされる)。 - SSH ログアウト後も動かし続ける必要がある場合は、tmux / screen か nohup を使う。なお、systemd の設定(
KillUserProcesses=yes)によっては、ログアウトするとユーザのプロセスが全て終了されることがある。その場合は管理者にloginctl enable-linger <ユーザ>を依頼する。
systemd で動かす場合との違い(注意)
| 項目 | systemd | 手動起動 |
|---|---|---|
| 実行ユーザ | 専用ユーザ logsuke
|
SSH ユーザ本人。そのユーザが読める全てのファイルを読める権限で動く(LogSuke が開くのは設定したログだけだが、万一の不具合時の影響範囲が広い) |
| サンドボックス・資源の上限(8.1) | 有効 |
無効。LogSuke 自体の limits(3章)は手動起動でも有効 |
| 自動起動・異常時の再起動 | あり | なし。サーバ再起動や異常終了の後は手動で起動し直す |
| SELinux(有効な環境のみ) |
semanage port が必要な場合がある |
SSH ユーザのプロセス(unconfined_t)として動くため、通常はポートの登録は不要 |
| 処理ログ | journald(journalctl -u logsuke) |
標準エラー出力(画面・logsuke.out)または logging.file
|
-
root で起動しない。起動時に実行ユーザが root(UID 0)の場合は警告を出して終了する(systemd で動かす場合も
User=logsukeのため root にはならない)。 - テスト期間が終わったら停止する(常駐させたままにしない)。
- INSTALL.md に「systemd で導入する手順」と「手動で起動する手順」の両方を記載する。
9. 段階的な進め方
- syslog のみで
cursor/since/tailを実装し、開発環境に配置。 - Python クライアント(
client/logsuke_client.py)を作成し、ログイン等1シナリオで動作確認。 - アプリのログファイルなど対象を
config.jsonに追加(コード変更なし)。 - 必要に応じて journald リーダーを追加。
10. テスト設計
標準ライブラリ unittest のみ使用。開発環境・最終テスト環境のどちらの Python(3.9 以上)でも同じテストが通ることを目標とする。
10.1 ディレクトリ構成
LogSuke/
└── tests/
├── __init__.py
├── helpers.py # 共通部品(一時ディレクトリ、設定生成、偽 /proc、テストサーバ起動)
├── test_config.py # logsuke.py:設定の読み込み・検証
├── test_main.py # logsuke.py:コマンドラインからの起動・停止(サブプロセス)
├── test_health.py # 結合:ヘルスチェック
├── test_code_rules.py # 静的検査:Python 3.9 互換・行数ルール・標準ライブラリのみ・コーディング規約
├── test_auth.py # logsuke.py:API Key / IP 制限
├── test_cursor.py # log_service.py:カーソルの変換
├── test_file_reader.py # log_service.py:cursor / since / tail
├── test_log_rotation.py # log_service.py:ローテート跨ぎ
├── test_process_service.py # process_service.py:偽 /proc を使った判定
├── test_http_api.py # 結合:実サーバを起動して HTTP で確認
├── test_client.py # 結合:logsuke_client.py を実サーバに対して実行
├── test_pytest_plugin.py # logsuke_pytest.py(pytest がない環境ではスキップ)
├── test_symlink.py # log_service.py:シンボリックリンク・特殊ファイル対策
├── test_security.py # 結合:不正リクエストの拒否
├── test_limits.py # 結合:リソース上限(走査量・行長・応答サイズ・同時接続・タイムアウト)
├── test_logging.py # 処理ログ(アクセスログ・スタックトレース・キー非出力)
└── test_tls.py # 結合:HTTPS で起動・接続(openssl コマンドがない環境ではスキップ)
実行方法:python3 -m unittest discover -s tests -v(LogSuke/ 直下で実行)
10.2 テスト容易性のための実装側の約束
-
パスの差し替え:
process_serviceの関数はproc_root="/proc"引数を持ち、テストでは偽 /proc ディレクトリを渡す。 -
時刻の差し替え:稼働秒数の計算は
nowを引数で受け取れるようにする(既定は現在時刻)。 -
設定は辞書で渡す:service はファイルを読まず、
logsuke.pyが読んだ設定辞書を受け取る。 -
サーバの起動を関数化:
create_server(config) -> ThreadingHTTPServerを用意し、テストではポート 0(空きポート)で別スレッド起動する。 - 本物の
/var/log/syslogや root 権限には依存しない(一時ファイルで代替)。
10.3 helpers.py の共通部品
| 部品 | 内容 |
|---|---|
TempDirTestCase |
setUp で tempfile.TemporaryDirectory を作り、tearDown で消す基底クラス |
make_config(tmp, **override) |
一時ログファイルと API Key ファイルを指す最小構成の設定辞書を返す |
write_log(path, lines) / append_log(path, lines)
|
ログファイルの作成・追記 |
make_fake_proc(root, procs) |
<pid>/stat, comm, cmdline, status と /proc/stat(btime), uptime を生成 |
ServerThread |
create_server で起動・停止するコンテキストマネージャ。base_url を持つ。TLS 設定を渡すと https:// で起動する |
make_self_signed_cert(tmp) |
openssl コマンドで localhost / 127.0.0.1 を SAN に持つ自己署名証明書を一時ディレクトリに作る。テスト時だけ使い、証明書をリポジトリに置かない |
http_get(url, api_key=None) |
urllib.request で GET し、(status, json) を返す |
10.4 テストケース一覧
test_config.py
| ケース | 期待結果 |
|---|---|
| 正しい設定を読み込める | 辞書が返る |
| JSON 文法エラー | 分かりやすいエラーで起動失敗 |
logs の必須キー(type, path)欠落 |
エラー |
未知の type
|
エラー |
processes の comm 欠落 / min_count > max_count
|
エラー |
ログIDに / や .. を含む |
エラー(IDは英数字・_・- のみ) |
max_lines 省略 |
既定値が入る |
limits / logging / tls を省略 |
既定値が入る(HTTP、既定の上限、標準エラー出力) |
limits の値が範囲外、logging.level が不正 |
エラー |
tls.enabled: true で cert_file / key_file 欠落 |
エラー |
rotated_glob が path と別のディレクトリ |
エラー |
未知のキー(綴り間違い。例:max_line) |
エラー(設定ミスに気付けるように) |
allow_ips に不正な CIDR |
エラー |
| 設定ファイルの権限が他ユーザ書き込み可能 | エラー |
test_auth.py
| ケース | 期待結果 |
|---|---|
| 正しい API Key | 許可され、キーの名前が返る |
| 誤り / 空 / ヘッダなし | 拒否 |
| API Key ファイルの空行・コメント行 | 無視される |
: のない行、名前の重複、キーの重複、32文字未満のキー、名前に使えない文字 |
起動時エラー |
| API Key ファイルの権限が 0644 / 0660(グループ書き込み)/ 他人の所有 | 起動時エラー |
| API Key ファイルが実行ユーザ所有の 0600(手動起動を想定) | 起動できる |
| 許可範囲内の IP / 範囲外 / IPv6 | 許可 / 拒否 / 設定どおり判定 |
allow_ips を省略 |
全て許可(API Key は必要)、起動時に WARNING |
allow_ips に 0.0.0.0/0 を明示 |
全て許可、起動時に WARNING |
allow_ips が空のリスト []
|
起動時エラー |
IPv4射影IPv6(::ffff:192.168.0.10) |
IPv4 として判定 |
範囲外 IP から X-Forwarded-For: 192.168.0.10 付きで接続 |
拒否(ヘッダを参照しない) |
test_cursor.py
| ケース | 期待結果 |
|---|---|
| encode → decode の往復 | 元の値に戻る |
| 不正形式(文字列、負の数、欠落) | ValueError |
test_file_reader.py
| ケース | 期待結果 |
|---|---|
| cursor 取得後に追記 → since | 追記分だけ返る |
| 追記なし | 空リスト、next_cursor は変わらない |
contains 1つ / 複数(AND) |
一致行だけ返る |
limit 超過、max_lines 超過 |
上限で切られ truncated: true
|
| 行の途中まで書かれた状態 | 改行のない最終行は返さず次回に回す |
| UTF-8 以外のバイト列 | 例外にならず置換文字で返る |
| tail(n) | 末尾 n 行 |
| ログファイルが存在しない | 専用エラー(HTTP では 404 ではなく 503) |
| ログファイルの読み取り権限がない | 専用エラー(HTTP では 503 log_unavailable) |
| 空のファイル | cursor は offset 0、since は空リスト |
| CRLF の改行 | 行末の \r を取り除いて返す |
| 別のログIDのカーソルを渡す(inode 不一致かつ照合先なし) | 先頭から返し missed: true(他のファイルは開かない) |
| 同じファイルへの同時 since(複数スレッド) | 互いに干渉せず、それぞれ正しい結果 |
test_log_rotation.py
| ケース | 期待結果 |
|---|---|
番号付き方式:cursor 取得 → 追記 → syslog.1 に rename → 新ファイルに追記 → since |
旧ファイル残り+新ファイルが順に返り rotated: true, missed: false
|
日付付き方式:messages-20261021 に rename して同様に確認 |
同上 |
2回ローテートを跨ぐ(.1 → .2、新たに .1 を作成) |
古い順にすべて返る |
| 一致するファイルが複数あるが、inode が一致するのは1つだけ | inode が一致したファイルから読む(ファイル名に依存しない) |
ローテート後のファイルが圧縮済み(.gz)で inode が一致しない |
新ファイル先頭から返し missed: true
|
| copytruncate 方式(同じ inode でサイズが縮む) | 先頭から読み直し rotated: true
|
rotated_glob 未設定でローテート |
新ファイル先頭から返し rotated: true, missed: true
|
rotated_glob が path と別のディレクトリを指す設定 |
起動時にエラー(test_config.py) |
test_process_service.py(偽 /proc 使用)
| ケース | 期待結果 |
|---|---|
comm 一致 1件 |
ok: true, count: 1 |
| 0件 |
ok: false(例外にはしない) |
max_count 超過 |
ok: false |
cmdline_contains / user で絞り込み |
条件に合うものだけ数える |
| 起動時刻・稼働秒数の計算 | btime と starttime から算出した値と一致 |
| 走査中にプロセスが消える(ファイルなし) | そのプロセスを無視して続行 |
| 応答に cmdline が含まれない | キーが存在しない |
| (Linux のみ)実 /proc で自分自身の python3 を検出 | count >= 1 |
test_symlink.py(一時ディレクトリでリンクや特殊ファイルを作成)
| ケース | 期待結果 |
|---|---|
path 自体がシンボリックリンク |
開かない。503 log_unavailable、処理ログに WARNING |
rotated_glob に一致するファイルが、ディレクトリ外のファイル(API Key ファイルの代わりのダミー)へのリンク |
照合の対象にならず、ダミーの内容は返らない(カーソルの inode をダミーに合わせても返らない) |
| 同じディレクトリ内のファイルへのリンク | 同じく開かない(リンクは一律拒否) |
path の親ディレクトリの途中がリンクに差し替えられている |
実体のパスの確認で拒否 |
rotated_glob に一致するのが FIFO / ディレクトリ |
拒否される。FIFO を開いても処理が止まらない |
| ハードリンク(同じディレクトリ内) | 通常のファイルとして扱う(ディレクトリの外へのハードリンクは同じファイルシステム上でしか作れず、/var/log 等に一般ユーザは作れないため対象外とする) |
api_keys_file をログと同じディレクトリに置いた設定 |
起動時エラー(test_config.py でも確認) |
test_http_api.py(実サーバ起動)
| ケース | 期待結果 |
|---|---|
GET /logs |
ID と description のみ(path を含まない) |
GET /logs/{id}/cursor → 追記 → /since
|
追記行が返る |
GET /logs/{id}/tail?lines= |
末尾行 |
GET /processes, /processes/{id}
|
200 と仕様どおりの JSON |
| 応答ヘッダ | Content-Type: application/json; charset=utf-8 |
| 日本語を含むログ行 | JSON で文字化けせずに返る |
truncated のとき next_cursor で続けて呼ぶ |
取りこぼし・重複なくすべての行が取れる |
クエリパラメータの URL エンコード(contains に空白・&・日本語) |
正しく解釈される |
test_client.py(実サーバ起動、logsuke_client.py の確認)
| ケース | 期待結果 |
|---|---|
| 各メソッドが対応する API を呼び、正しい型で返す | 仕様どおり |
| 環境変数からの URL / API Key 読み込み | 引数省略時に使われる |
| API Key 誤り | LogSukeAuthError |
| 未定義ID |
LogSukeError(status 404) |
wait_for_log:途中で別スレッドから追記 |
追記後に一致行を返す |
wait_for_log:一致なし |
timeout 経過後に LogWaitTimeout
|
assert_process_ok:条件外 |
AssertionError と分かりやすいメッセージ |
wait_for_log の待機中にログがローテートされる |
ローテート後に出力された一致行も返る |
wait_for_log の contains がリスト |
AND で一致した行だけ返る |
| サーバが停止している | 接続エラーが LogSukeError として分かりやすいメッセージで出る(タイムアウトまで固まらない) |
| サーバに接続を切断される(同時接続数の上限) | 最大3回再試行し、それでも失敗したら LogSukeError
|
assert_no_log:一致なし |
wait 秒後に正常終了 |
assert_no_log:wait の間に一致する行が追記される |
AssertionError、一致した行がメッセージに含まれる |
| サーバが 302 を返す(テスト用サーバで再現) | リダイレクト先にアクセスせず LogSukeError(API Key が他へ送られない) |
環境変数 HTTP_PROXY が設定されている |
プロキシを使わずに直接接続する。LOGSUKE_USE_PROXY=1 のときだけ使う |
assert_healthy:degraded
|
AssertionError、読めないログIDがメッセージに含まれる |
| 例外のメッセージに API Key の値が含まれない | 含まれない |
test_security.py(実サーバ起動)
| ケース | 期待結果 |
|---|---|
| API Key なし / 誤り | 401 |
| 未定義のログID / プロセスID | 404 |
../etc/passwd、URL エンコードした %2e%2e%2f、絶対パス |
404(ファイルを開かない) |
不正な cursor / 数値でない limit
|
400 |
| GET 以外(POST / PUT / DELETE) | 405 |
| 未定義パス | 404 |
| 応答本文・エラーメッセージにサーバのファイルパスが含まれない | 含まれない |
| 全ての応答(正常・エラー) |
Cache-Control: no-store、X-Content-Type-Options: nosniff が付く |
Origin ヘッダ付きのリクエスト / OPTIONS
|
Access-Control-Allow-* が付かない / 405
|
| 内部で例外を起こす(テスト用に service を差し替え) |
500、本文は定型文と request_id のみ(スタックトレースなし) |
Server ヘッダ |
LogSuke のみ(Python のバージョンを含まない) |
| 404 / 405 などのエラー応答 | HTML ではなく JSON |
test_limits.py(実サーバ起動。上限値を小さくした設定で確認)
| ケース | 期待結果 |
|---|---|
走査量が max_scan_bytes を超える |
打ち切られ truncated_reason: max_scan_bytes。next_cursor で続きを読むと残りが取れる |
1行が max_line_bytes を超える |
切り捨てられ …[truncated] が付く |
応答が max_response_bytes を超える |
truncated_reason: max_response_bytes |
contains の個数・長さ超過 |
400 |
tail の lines が上限超過 |
上限値で返る |
| 同時接続数を超える(遅いハンドラを差し込んで占有) | 超えた接続はすぐに切断され、スレッド数が max_connections を超えない(threading.active_count() で確認) |
| 上限の接続が終わった後 | 再び接続を受け付ける |
| 接続後に何も送らない |
request_timeout_sec 後に切断され、他のリクエストは処理され続ける |
| 4096 バイトを超える URL | 414 |
limits に不正な値(0 以下など) |
起動時エラー(test_config.py でも確認) |
test_logging.py(assertLogs で処理ログを検査)
| ケース | 期待結果 |
|---|---|
| 正常なリクエスト | INFO のアクセスログに request_id・IP・キーの名前・パス・ステータスが出る |
| 認証失敗 | WARNING。送られてきたキーの値はログに出ない |
| どのログにも API Key の値が出ない | 全レベルのログを検査して含まれない |
| 内部で例外 | ERROR にスタックトレースが出て、応答の request_id と一致する |
| 起動時 | 待ち受けアドレス、ログID一覧が出る。HTTP の場合は平文警告が出る |
logging.file 指定 |
指定ファイルに出力される |
contains に改行・CR・ESC を含むリクエスト(%0A など) |
アクセスログは1行のままで、\n / \r / \x1b とエスケープされている |
| 512 文字を超える値 | 切り詰められて記録される |
safe_for_log() の単体テスト(全制御文字、日本語、空文字) |
制御文字だけがエスケープされ、日本語はそのまま |
test_pytest_plugin.py(pytest の pytester を使用。pytest がなければ skip)
| ケース | 期待結果 |
|---|---|
| 失敗するテスト | レポートに logsuke: syslog セクションが追加され、テスト中に追記した行だけが入っている |
| 成功するテスト | セクションが追加されない |
| セットアップ(fixture)で失敗 | セクションが追加される |
logsuke_attach_max_lines を超える行 |
末尾だけ残り「… N 行省略」が付く |
| LogSuke が停止している | テスト結果は変わらず、「ログを取得できませんでした」と記載される |
--logsuke-attach-dir 指定 |
ファイルが保存される |
logsuke_attach_logs が空 |
何もしない |
test_health.py(実サーバ起動)
| ケース | 期待結果 |
|---|---|
| 全ログが読める |
status: ok、key_name が使ったキーの名前、tls: false
|
| 1つのログファイルを削除 / 権限を外す |
status: degraded、該当ログだけ readable: false、HTTP は 200 |
| 処理ログ |
readable: false の理由が WARNING で出る |
| API Key なし / 許可外 IP |
401 / 403(health も認証対象) |
| 応答にファイルパス・Python バージョンが含まれない | 含まれない |
server_time |
タイムゾーン付きの ISO 8601 形式 |
| HTTPS で起動した場合 |
tls: true(test_tls.py で確認) |
test_main.py(python3 logsuke.py --config ... をサブプロセスで起動)
| ケース | 期待結果 |
|---|---|
正しい設定で起動 → /health に接続 → SIGTERM |
応答後、正常終了(終了コード 0)し、停止ログが出る |
同様に起動して SIGINT(Ctrl+C 相当) |
正常終了(終了コード 0)。スタックトレース(KeyboardInterrupt)を出さない |
別のカレントディレクトリから、相対パスの api_keys_file を含む設定で起動 |
設定ファイルのディレクトリを基準に解決されて起動する |
root(UID 0)で起動(os.geteuid を差し替えて確認) |
警告を出して終了コード 1 |
| 設定ファイルがない / 不正 | すぐに終了コード 1 で終了し、標準エラー出力に理由が出る(スタックトレースではなく1行の説明) |
| ポートが使用中 | 終了コード 1 と分かりやすいメッセージ |
--help |
使い方が表示される |
test_code_rules.py(ソースコードの静的検査。サーバを起動しない)
| ケース | 期待結果 |
|---|---|
全 .py を ast.parse(..., feature_version=(3, 9)) で解析 |
開発環境の Python が 3.10 以降でも、3.9 で使えない文法(match 文など)を検出できる |
| 各モジュールの行数 | 500行以下 |
| 各関数・メソッドの行数 | 50行以下 |
全ファイルの先頭(docstring の後)に from __future__ import annotations がある |
ある |
全関数・メソッドの引数(self / cls 以外)と戻り値に型ヒントがある |
ある |
typing から List / Dict / Tuple / Set / Optional / Union を import していない |
していない |
型ヒント以外の場所(モジュール直下の代入・isinstance の引数)で型に | を使っていない |
使っていない(3.9 で実行時エラーになるため) |
| 全関数・メソッド・クラスに docstring がある | ある |
docstring に Args: がある場合、引数名と一致し、型を書いていない(name (str): の形でない) |
一致し、型なし |
例外を raise する関数の docstring に Raises: がある |
ある |
戻り値が None の関数の docstring に Returns: がない |
ない |
| 関数名・変数名が snake_case、クラス名が PascalCase、モジュール直下の定数が UPPER_SNAKE_CASE | 規則どおり |
import しているモジュール |
標準ライブラリと LogSuke 自身のモジュールだけ(sys.stdlib_module_names は 3.10 以降のため、許可リストで判定する) |
subprocess / os.system / eval / exec の使用 |
サーバ側モジュール(logsuke.py・log_service.py・process_service.py)にない |
| ログファイルを開く処理 |
log_service.py で open() / os.open() を呼んでいるのが open_log_file() の中だけ(安全な開き方を迂回していない) |
| 標準ライブラリ以外の import の例外 |
client/logsuke_pytest.py の pytest だけを許可する |
test_tls.py(実サーバ起動、openssl がなければ skip)
| ケース | 期待結果 |
|---|---|
tls.enabled: false(既定) |
HTTP で応答し、起動ログに平文警告が出る |
tls.enabled: true + ca_file 指定でクライアント接続 |
HTTPS で 200 |
ca_file 未指定(自己署名証明書を信頼していない) |
証明書エラーで LogSukeError
|
| SAN にないホスト名で接続 | ホスト名不一致エラー |
insecure=True |
接続できるが警告が出る |
HTTPS サーバに http:// で接続 |
応答を得られずエラーになる(平文で処理されない) |
| TLS 1.1 以下で接続 | ハンドシェイク失敗 |
| 証明書・鍵のファイルがない、ペア不一致、鍵の権限が 0644 | 起動時エラー(test_config.py でも確認) |
10.4.1 要件とテストの対応
| 要件 | 主なテスト |
|---|---|
| IDで定義したログ・プロセスしか見えない | test_security, test_http_api, test_config |
| ローテートを跨いでも取りこぼさない(番号付き / 日付付き) | test_log_rotation, test_client |
| 認証(名前付きキー)・IP制限(ソケットのIPのみ、省略時は全許可+警告) | test_auth, test_security, test_health |
| シンボリックリンク・特殊ファイル対策 | test_symlink, test_code_rules |
| 応答ヘッダ・CORS なし | test_security |
| ログインジェクション対策 | test_logging |
失敗時のログ添付・assert_no_log
|
test_pytest_plugin, test_client |
| リソースの上限(接続時点でのスレッド数制限を含む) | test_limits |
| 情報を漏らさない(パス・スタックトレース・Server ヘッダ・キー) | test_security, test_logging, test_health, test_client |
| 処理ログ・スタックトレース | test_logging, test_main |
| HTTPS(自己署名証明書) | test_tls |
| Python 3.9 互換・行数ルール・pip 不要・コーディング規約(11章) | test_code_rules |
10.5 対象外(手動確認とする)
- Python 3.9 での実行:test_code_rules で文法は確認するが、標準ライブラリの細かな差は Python 3.9 の Linux 環境上で
python3 -m unittest discover -s testsを実行して確認する。 - systemd unit の起動、SELinux、ファイアウォール、ACL:INSTALL.md のチェックリストで最終テスト環境の実機確認。
systemd-analyze security logsukeの結果と、MemoryMax等のサンドボックス設定下で全APIが動くことも確認する。 - 手動起動:SSH ログインしたユーザで tmux / nohup により起動し、ログアウト後もテスト端末から接続できることを確認する。
- Playwright との組み合わせ:開発環境と最終テスト環境で1シナリオを pytest-playwright で実際に流して確認。
10.6 見積り
テストファイル16個、各 50〜250行程度(いずれも500行以下)、helpers.py 150〜200行程度。合計 約150ケース。全件の実行時間は30秒以内を目安とする(タイムアウト系のテストは設定値を1〜2秒に縮めて実行する)。
11. コーディング規約
本規約に記載のない事項は、PEP 8 と Google Python Style Guide に従う。
対象は logsuke.py・log_service.py・process_service.py・client/ 配下・tests/ 配下の全ての Python ファイルとする。
11.1 命名規則
| 対象 | 形式 | 例 |
|---|---|---|
| 変数・関数・引数 | snake_case |
next_cursor, load_api_keys, proc_root
|
| クラス | PascalCase |
FileReader, LogSukeClient
|
| 定数(モジュール直下) | UPPER_SNAKE_CASE |
DEFAULT_LIMITS, COMPRESSED_SUFFIXES
|
11.2 型ヒント
- すべての関数の引数と戻り値に型ヒントを付ける。
- 各ファイルの先頭に
from __future__ import annotationsを書く。 - 組み込みの型と
|を使う(list[str]、dict[str, str]、FileInfo | None)。typing.List、typing.Dict、Optional、Unionは使わない。 - 戻り値の型は、実際に返す値の型と一致させる。
- コールバックの型は
Callable[[引数の型], 戻り値の型]と書く(collections.abcから import する)。
Python 3.9 での注意
from __future__ import annotations があると、型ヒントは文字列として扱われ実行されないため、X | None も 3.9 で使える。ただし、型ヒント以外の場所で | を型に使うと 3.9 ではエラーになるので、次は書かない。
# NG: 3.9 では実行時に TypeError(型の | は 3.10 から)
Cursor = tuple[int, int] | None # モジュール直下の型エイリアス
if isinstance(value, int | str): ... # isinstance の引数
# OK
def decode_cursor(text: str) -> tuple[int, int] | None: ... # 型ヒントの中は可
if isinstance(value, (int, str)): ...
- 型エイリアスが必要な場合は、
|を使わない形にするか、型ヒントの中に直接書く。 -
typing.get_type_hints()など、型ヒントを実行時に評価する処理は使わない。
11.3 docstring
関数には、日本語の Google スタイル docstring を書く。
def is_sub_path(base_path: str, full_path: str) -> bool:
"""full_path が base_path のサブパスであるかを判定する。
base_path と full_path が同一の場合も True を返す。
Args:
base_path: 基準となるパス。
full_path: 判定対象の完全なパス。
Returns:
サブパスであれば True、そうでなければ False。
"""
- 1行目は
"""と同じ行に、「〜する。」の形で要約を書く。補足は空行を1行挟んで書く。 -
Args:には型を書かない(root (str):ではなくroot:)。 - 戻り値が
Noneの関数はReturns:を書かない。 - 例外を送出する関数は
Raises:を書く。
11.4 コメント
- コメントは日本語で書く。
- コメントには、処理の内容ではなく、その処理をする理由を書く。
- 保留事項は
# TODO:、注意事項は# NOTE:で書き始める。 - クラス内の区切りには
# ====の見出しコメントを使う。
11.5 本プロジェクトでの補足
-
テストコード:テストメソッド(
test_*)とヘルパーにも型ヒント(戻り値は-> None)と docstring を付ける。ただしテストメソッドの docstring は「何を確認するか」の1行だけでよい(Args:等は不要)。setUp/tearDownなどunittestの決まったメソッドは1行の docstring でよい。 - サイズの目安:1ファイル 500 行以下、1関数 50 行以下(7章)。docstring とコメントも行数に含める。
-
標準ライブラリのみ:サーバ側とクライアント(
logsuke_client.py)は標準ライブラリだけを使う。例外はlogsuke_pytest.pyのpytestのみ。 -
自動チェック:規約のうち機械的に確認できるものは
tests/test_code_rules.pyで検査する(10.4)。PEP 8 の細かな書式(行の長さ等)は、pip 不要の方針のため自動チェックの対象外とし、レビューで確認する。開発者の手元でflake8等を使うのは自由とする。