1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

ポエム:claude code始めました。

1
Posted at

元々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 に適用。起動時)

  1. 所有者が root または LogSuke を実行しているユーザである。
  2. 「その他(other)」に読み・書きの権限がない(o-rw)。
  3. グループに書き込み権限がない(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 は、ローテートを跨いだときに一部を読めなかった可能性があることを示す(下記)。

ローテートの検知と読み取り手順

  1. カーソルの inode が path の現在の inode と同じであれば、通常どおり offset から読む。
    • サイズが offset より小さくなっていた場合は copytruncate 方式で切り詰められたと判断し、先頭から読んで rotated: true とする。
  2. inode が違う場合は、ローテートされたと判断する。
    • rotated_glob に一致するファイル(圧縮ファイルを除く)を inode で照合し、カーソルの inode と同じファイルを探す。ファイル名では判断しない。
    • 見つかった場合は、そのファイルの offset 以降を読み、それより新しいローテート済みファイル(更新日時順)を続けて読み、最後に現在の path の先頭から読む。rotated: true とする。
    • 見つからない場合(圧縮済み・削除済み・rotated_glob 未設定)は、現在の path の先頭から読み、rotated: true, missed: true とする。
  3. ファイルを開くのは、常に path と rotated_glob に一致したファイルだけとする。カーソルの inode を使ってファイルを直接開くことはしない。
  4. ファイルを開くときは、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. セキュリティ方針

  1. IDホワイトリスト方式:パス・ファイル名・ワイルドカードを受け付けない。

  2. 読み取り専用:専用ユーザで実行し、書き込み・削除APIは作らない。systemd で ProtectSystem=strict / NoNewPrivileges=yes を設定。

  3. 認証: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)。
  4. 量の制限:同時接続数・タイムアウト・走査量・行の長さ・応答サイズ・contains の個数と長さに上限を設ける(3章 limits)。上限を超えても、サーバは止まらずに打ち切る・4xx を返す・接続を切断するのいずれかで対応する。

  5. テスト環境専用:導入するのは開発環境と最終テスト環境(いずれも Linux)だけとし、利用者向けの本番サーバには導入しない。本番環境の構築手順書やイメージに LogSuke が混ざらないよう、INSTALL.md に明記する。

  6. アクセスログ:誰(API Key の名前)がどのIDを読んだかを LogSuke 自身の処理ログに記録する。キーそのものは記録しない。

  7. プロセス確認も同じ方式:確認できるのは processes に定義したIDだけにする。プロセス名などを端末から指定する検索はできない。

    • 他ユーザのプロセスを /proc で見えるようにするため、unit ファイルに ProtectProc=invisible は設定しない。
    • /proc が hidepid 付きでマウントされている場合は、gid= オプションで指定されたグループに logsuke を追加する。
  8. サーバ情報を出さない・応答ヘッダ:Server ヘッダは LogSuke だけにする(server_version = "LogSuke"、sys_version = "")。http.server 既定の HTML エラーページは使わず、send_error を上書きして必ず JSON のエラー応答を返す。全ての応答(エラーを含む)に次のヘッダを付ける。

    ヘッダ 値 目的
    Content-Type application/json; charset=utf-8 形式の明示
    Cache-Control no-store ログの内容を途中のキャッシュや端末に残さない
    X-Content-Type-Options nosniff ブラウザが JSON を別の形式と解釈しないようにする
    • CORS ヘッダ(Access-Control-Allow-*)は付けない。OPTIONS も 405 とする。これにより、テスト中のブラウザ上のページ(Webアプリや外部サイト)の JavaScript から LogSuke を呼び出せない。
  9. 処理ログ: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())でまとめて行う。
  10. ファイルの安全な開き方(シンボリックリンク対策):ログファイル(path と rotated_glob に一致したファイル)は次の手順でだけ開く。将来アプリのログを対象にすると、そのディレクトリにはアプリのユーザが書き込めることが多い。アプリが乗っ取られた場合に、app.log.1 -> /etc/logsuke/tls/server.key のようなリンクを置かれて、LogSuke が読めるファイルを読み出されるのを防ぐ。

    1. os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) で開く。最後の要素がシンボリックリンクなら開けずにエラーになる。O_NONBLOCK は FIFO を開いたときに止まらないようにするため。
    2. 開いたファイルを os.fstat() で調べ、通常のファイル(S_ISREG)以外は拒否する(ディレクトリ・FIFO・デバイスなど)。
    3. 開いたファイルの実体のパスを os.path.realpath(f"/proc/self/fd/{fd}") で取得し、設定したディレクトリ(path の親ディレクトリを realpath で解決したもの)の直下にあるかを確認する。途中のディレクトリがリンクに差し替えられた場合も、ここで検出できる。
    4. いずれかに該当した場合はファイルを閉じて読み取らず、そのファイルは「存在しない」として扱う(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. 段階的な進め方

  1. syslog のみで cursor / since / tail を実装し、開発環境に配置。
  2. Python クライアント(client/logsuke_client.py)を作成し、ログイン等1シナリオで動作確認。
  3. アプリのログファイルなど対象を config.json に追加(コード変更なし)。
  4. 必要に応じて 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 等を使うのは自由とする。
1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?