diff --git a/tools/api-inventory/.gitignore b/tools/api-inventory/.gitignore index 31e52f5828..aa0bf73b32 100644 --- a/tools/api-inventory/.gitignore +++ b/tools/api-inventory/.gitignore @@ -6,6 +6,7 @@ weko3_api_auth_findings.md api_snapshot*.json reconcile_allow.json reconcile_report.md +detect_allow.json probe*.json drift*.md # fixtures.py が生成する。秘密は入らない(パスワードは fixtures.py の定数、 @@ -14,3 +15,4 @@ drift*.md fixtures.json __pycache__/ *.pyc +.pytest_cache/ diff --git a/tools/api-inventory/ci/README.md b/tools/api-inventory/ci/README.md index c6c512d602..b49ac926af 100644 --- a/tools/api-inventory/ci/README.md +++ b/tools/api-inventory/ci/README.md @@ -9,7 +9,7 @@ | 置き場所 | 内容 | |---|---| | **本リポジトリ `tools/api-inventory/`** | **ツールのみ**(scripts / ci)。データは1件も置かない | -| **`RCOSDP/weko-secret`**(private) | 台帳TSV(57列/24列)、列定義README、`api_snapshot.json`、`reconcile_allow.json`、`reconcile_report.md`、調査記録 | +| **`RCOSDP/weko-secret`**(private) | 台帳TSV(62列/32列)、列定義README、`api_snapshot.json`、`reconcile_allow.json`、`detect_allow.json`、`reconcile_report.md`、調査記録、台帳の検査テスト | 本書では `RCOSDP/weko-secret`(private)を単に**プライベートリポジトリ**と呼ぶ。 スクリプトは環境変数 `WEKO_API_INVENTORY_DIR` でその場所を指す。未設定なら理由を添えて中断する。 @@ -22,6 +22,19 @@ python3 tools/api-inventory/scripts/reconcile.py --gate CI の出力は **`--summary-only` で件数のみ**。URI や endpoint 名は出さない。 +## ワークフローは2本 + +| ワークフロー | 見るもの | 要るもの | 所要 | +|---|---|---|---| +| `api-inventory-tests.yml` | **台帳を作る側**(スクリプト・手順書)が壊れていないか | なし | 数秒 | +| `api-inventory-drift.yml` | **台帳の中身**が実機・ソースとずれていないか | Secret + Docker | 60分枠 | + +ツールが壊れたまま drift だけ回すと、検知器が黙って死んでいても緑で通る。 +**先に tests を通すこと。** + +台帳の中身そのものの検査(列数・語彙・派生列の再現・突き合わせゲート)は、 +データのある**プライベートリポジトリ側の `tests/`** が持つ。 + ## 1. 移設するファイル WEKO3 リポジトリに `tools/api-inventory/` を作り、weko-document の @@ -34,27 +47,34 @@ weko/tools/api-inventory/ ← public。ツールのみ │ ├── paths.py $WEKO_API_INVENTORY_DIR の解決 │ ├── extract_routes.py … Phase 1-2: 静的抽出・観点付与 │ ├── probe.py / asuser.sh Phase 3: 実機Docker実測(参考実装) -│ ├── build_checklist.py Phase 5: 57列 → 24列の再生成 +│ ├── schema.py 列定義の唯一の正(62列 / 32列) +│ ├── build_checklist.py Phase 5: 62列 → 32列の再生成 │ ├── snapshot.py Phase 6: 実機url_map → スナップショット │ ├── diff_snapshot.py Phase 6: スナップショット間の差分 + ゲート │ ├── reconcile.py Phase 6: スナップショット ↔ 台帳の突き合わせ +│ ├── detect_routes.py ソース(AST)↔ 台帳の突き合わせ。実機不要 │ ├── changed_rows.py Phase 6: git差分 → 再レビュー対象行 │ ├── fixtures.py Phase 7: 到達可否測定用の最小コーパス投入 │ ├── probe_ci.py Phase 7: フィクスチャ駆動の到達可否測定(CI が直接呼ぶ) │ └── measure.sh 手作業で実測するときの唯一の入口(上記を固定順で回す) +├── tests/ ツールの単体テスト(pytest。データ不要) +├── pytest.ini ├── ci/ -│ ├── api-inventory-drift.yml +│ ├── api-inventory-drift.yml 実機を起こして突き合わせる(60分枠) +│ ├── api-inventory-tests.yml ツールの単体テスト(数秒。Secret 不要) │ └── README.md このファイル └── .gitignore データ類を誤ってコミットしないための保険 $WEKO_API_INVENTORY_DIR/ ← プライベートリポジトリ。public リポジトリには置かない -├── weko3_api_list_full.tsv 台帳(57列・所見と実証結果つき) -├── weko3_api_list.tsv 台帳(24列) -├── weko3_api_list_README.md 24列の列定義・運用手順 -├── weko3_api_list_full_README.md 57列の列定義 +├── weko3_api_list_full.tsv 台帳(62列・所見と実証結果つき) +├── weko3_api_list.tsv 台帳(32列) +├── weko3_api_list_README.md 32列の列定義・運用手順 +├── weko3_api_list_full_README.md 62列の列定義 ├── api_snapshot.json 経路のベースライン -├── reconcile_allow.json 実機に無い行の許可リスト +├── reconcile_allow.json 実機に無いが台帳に残す行の許可リスト +├── detect_allow.json ソースにあるが経路にならないものの許可リスト ├── reconcile_report.md 突き合わせ結果 +├── tests/ 台帳そのものの検査(pytest。実機不要) └── weko3_api_auth_findings.md 調査記録 ``` diff --git a/tools/api-inventory/ci/api-inventory-drift.yml b/tools/api-inventory/ci/api-inventory-drift.yml index f21290378a..50eff4db09 100644 --- a/tools/api-inventory/ci/api-inventory-drift.yml +++ b/tools/api-inventory/ci/api-inventory-drift.yml @@ -17,6 +17,13 @@ # 個人アカウントに紐づかないため(PAT より事故時の影響が小さい)。 # 未設定なら、このジョブは何もせずスキップする(fork からの PR でも安全)。 # +# 網羅性は二段で見る: +# reconcile.py 実機 url_map ↔ 台帳(この環境で登録されている経路) +# detect_routes.py ソース(AST) ↔ 台帳(config で無効な経路まで含む) +# 前者だけだと、config で無効・プラグイン未導入の経路が台帳から落ちても気付けない。 +# +# ツールそのものの単体テストは api-inventory-tests.yml(Secret も Docker も不要)。 +# # 設置手順: tools/api-inventory/ci/README.md name: API Inventory Drift @@ -147,6 +154,13 @@ jobs: --snapshot /tmp/api_snapshot.new.json \ --summary-only --gate --out /tmp/reconcile.md + # 実機 url_map は「この環境で登録された経路」しか映さない。config で無効・ + # プラグイン未導入・設定値が真のときだけ登録される経路は、API として + # 存在するのに reconcile では見えない。ソースからの検知で二段目を張る。 + python3 $T/detect_routes.py \ + --weko-root "$PWD" --cross-check \ + --summary-only --gate --out /tmp/detect.md + - name: Probe changed endpoints if: always() && steps.cfg.outputs.enabled == 'true' env: @@ -175,6 +189,7 @@ jobs: path: | /tmp/drift.md /tmp/reconcile.md + /tmp/detect.md - name: Comment on PR (counts only) if: always() && steps.cfg.outputs.enabled == 'true' && github.event_name == 'pull_request' @@ -206,6 +221,7 @@ jobs: + '該当箇所はプライベートリポジトリ側の台帳・レポートで確認してください。'; body += read('/tmp/drift.md', 'ベースラインとの差分'); body += read('/tmp/reconcile.md', '台帳との突き合わせ'); + body += read('/tmp/detect.md', 'ソース由来の経路検知'); await github.rest.issues.createComment({ issue_number: context.issue.number, owner: context.repo.owner, diff --git a/tools/api-inventory/ci/api-inventory-tests.yml b/tools/api-inventory/ci/api-inventory-tests.yml new file mode 100644 index 0000000000..4894f457da --- /dev/null +++ b/tools/api-inventory/ci/api-inventory-tests.yml @@ -0,0 +1,57 @@ +# WEKO3 リポジトリ(RCOSDP/weko)の .github/workflows/ に配置する。 +# +# 台帳ツールの単体テスト。**Docker も実機も台帳も要らない**ので数秒で終わる。 +# api-inventory-drift.yml(実機を起こして突き合わせる。60分枠)とは役割が違う: +# +# このワークフロー … 台帳を作る側(スクリプト・手順書)が壊れていないか +# drift ワークフロー … 台帳の中身が実機とずれていないか +# +# ツールが壊れたまま drift だけ回すと、検知器が黙って死んでいても緑で通る。 +# 先にこちらを通すこと。Secret も不要なので fork からの PR でも動く。 + +name: API Inventory Tests + +on: + pull_request: + paths: + - 'tools/api-inventory/**' + - '.github/workflows/api-inventory-tests.yml' + push: + branches: ['**'] + paths: + - 'tools/api-inventory/**' + workflow_dispatch: + +jobs: + unit: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install pytest + run: python3 -m pip install --disable-pip-version-check pytest + + - name: Run unit tests + working-directory: tools/api-inventory + run: python3 -m pytest -q + + # 台帳が無くても、ソースからの経路検知そのものは動く。 + # 検知件数が 0 に落ちていれば、検知器が壊れている。 + - name: Smoke check the static detector + run: | + set -o pipefail + python3 tools/api-inventory/scripts/detect_routes.py \ + --weko-root "$PWD" --summary-only | tee /tmp/detect.md + python3 - <<'PY' + import re, sys + text = open('/tmp/detect.md', encoding='utf-8').read() + total = int(re.search(r'\*\*計\*\* \| \*\*(\d+)\*\*', text).group(1)) + print(f'detections={total}') + # 経路が数百ある前提のリポジトリ。2桁に落ちたら検知器の故障を疑う。 + sys.exit(0 if total >= 300 else 1) + PY diff --git a/tools/api-inventory/pytest.ini b/tools/api-inventory/pytest.ini new file mode 100644 index 0000000000..41da5ba660 --- /dev/null +++ b/tools/api-inventory/pytest.ini @@ -0,0 +1,11 @@ +# 台帳ツールの単体テスト。 +# +# cd tools/api-inventory && python3 -m pytest +# +# `scripts/test_coverage.py` は名前が test_ で始まるが**テストではない** +# (台帳にテスト観点を付与する本体スクリプト)。testpaths で tests/ に限定して +# 誤収集を防ぐ。 +[pytest] +testpaths = tests +python_files = test_*.py +addopts = -q diff --git a/tools/api-inventory/scripts/README.md b/tools/api-inventory/scripts/README.md index 91ea82f8c7..8480880f4a 100644 --- a/tools/api-inventory/scripts/README.md +++ b/tools/api-inventory/scripts/README.md @@ -22,7 +22,7 @@ > 成果物TSV/MDは一つ上の階層(`../weko3_api_list.tsv` 等)にある。 -`weko3_api_list.tsv`(24列・チェックリスト版)と `weko3_api_list_full.tsv`(57列・詳細版)を +`weko3_api_list.tsv`(32列・チェックリスト版)と `weko3_api_list_full.tsv`(62列・詳細版)を **バージョンアップのたびに再生成**するための手順とスクリプト一式。 # 台帳の更新手順(まずここを読む) @@ -39,13 +39,20 @@ cd /path/to/weko # ツールは WEKO3 リポジトリ側にある ## 大原則 -- **`weko3_api_list.tsv`(24列版)は直接編集しない。** `weko3_api_list_full.tsv` から +- **`weko3_api_list.tsv`(32列版)は直接編集しない。** `weko3_api_list_full.tsv` から `build_checklist.py` が丸ごと生成する派生物で、手を入れても次の生成で消える。 - **派生列も手編集しない。** `priority` / `priority_reason` / `test_normal`〜`test_gap` / `cleanup` はスクリプトが毎回上書きする。直したいときは判定の入力側 (`security_finding` / `dynamic_verified` / `data_op` / `deprecated` 等)を直すか、 `prioritize.py` のルールを変える。 - **実行順がある。** `prioritize.py` は `test_gap` を参照するので `test_coverage.py` が先。 +- **git 由来の列は放っておくと古びる。** `impl_line` と + `last_commit` / `last_commit_date` / `last_commit_subject` / `release_tag` は + ソースが変われば実態とずれるが、上の3本では更新されない。 + **実装に手が入ったら `refresh_impl.py --write` → `enrich_git.py --write` を回すこと** + (v2.0.3 → v2.0.4 では、この2本が手順に無かったために台帳の release_tag が + v2.0.3 生成時のまま据え置かれ、issue62569 で認可を足した30行が + 「v0.1.0b1 で最後に変更」と表示され続けた)。 ## ケース1: 派生列を再計算するだけ(最も多い) @@ -54,12 +61,12 @@ cd /path/to/weko # ツールは WEKO3 リポジトリ側にある ```bash python3 tools/api-inventory/scripts/test_coverage.py # テスト4観点を判定 python3 tools/api-inventory/scripts/prioritize.py # 優先度・整理対象を付与 -python3 tools/api-inventory/scripts/build_checklist.py # 24列版を再生成 +python3 tools/api-inventory/scripts/build_checklist.py # 32列版を再生成 ``` ## ケース2: 台帳に行を追加する -`reconcile.py` が「A. インベントリ未収載」を出したとき。57列を手で並べる必要はない。 +`reconcile.py` が「A. インベントリ未収載」を出したとき。62列を手で並べる必要はない。 ```bash # 1) 何が未収載かを確認する @@ -77,10 +84,10 @@ python3 tools/api-inventory/scripts/add_row.py --endpoint api:weko_admin.foo --a vi "$WEKO_API_INVENTORY_DIR/weko3_api_list_full.tsv" # 5) 列数の検算 -awk -F'\t' 'NR>1 && NF!=65{print "行"NR" 列数="NF}' \ +awk -F'\t' 'NR>1 && NF!=62{print "行"NR" 列数="NF}' \ "$WEKO_API_INVENTORY_DIR/weko3_api_list_full.tsv" -# 6) 派生列を再計算 → 24列版を再生成 → 突き合わせ +# 6) 派生列を再計算 → 32列版を再生成 → 突き合わせ python3 tools/api-inventory/scripts/test_coverage.py python3 tools/api-inventory/scripts/prioritize.py python3 tools/api-inventory/scripts/build_checklist.py @@ -97,24 +104,24 @@ python3 tools/api-inventory/scripts/add_cols.py # csrf_protection / inp # audit_logged / triggers_task / resource_limit python3 tools/api-inventory/scripts/add_ssrf_redirect.py # redirect_target / ssrf_surface python3 tools/api-inventory/scripts/add_idempotency.py # idempotency -python3 tools/api-inventory/scripts/add_dataop4.py # data_op_detail +python3 tools/api-inventory/scripts/add_dataop4.py # data_op(操作4区分) python3 tools/api-inventory/scripts/add_authmech.py # auth_mechanism / bola_risk ``` **これらは空欄/`TODO` のセルだけを埋める。既存値は上書きしない。** 台帳の既存値は機械出力そのままではなく後から精査されており、一括再生成すると劣化する -(実測: `bola_risk` の判定が逆転、`data_op_detail` の論理削除/物理削除の区別が失われる、 +(実測: `bola_risk` の判定が逆転、`data_op` の論理削除/物理削除の区別が失われる、 `csrf_protection` の指摘が消える)。意図して作り直すときだけ `WEKO_INVENTORY_OVERWRITE=1` を付ける。 ### add_row.py が埋める列 / 埋めない列 -`api_snapshot.json`(実機 url_map)と git から**機械的に決まる27列**を埋め、 -調査が要る31列に `TODO` を入れる。 +`api_snapshot.json`(実機 url_map)と git から**機械的に決まる26列**を埋め、 +調査が要る28列に `TODO` を入れる(残り8列は派生列。手順6で自動的に付く)。 -| 自動(27列) | no / module / api_type / app / method / uri / path_params / blueprint / endpoint / impl_func / impl_file / impl_line / auth_required / auth_method / auth_mechanism / api_version / last_commit系4列 ほか | +| 自動(26列) | no / module / api_type / app / method / uri / path_params / query_params / body_params / request_content_type / blueprint / endpoint / impl_func / impl_file / impl_line / auth_required / auth_method / oauth_scope / cache_ratelimit / api_version / deprecated / auth_mechanism / last_commit系4列 | |---|---| -| **`TODO`(31列)** | **summary / response / status_codes / exceptions / roles / auth_response_variance / restricted_content / data_op / data_target / data_store / side_effects / config_deps / test_file / category_tags / notes / sec_* / dynamic_verified / csrf_protection / input_validation / audit_logged / triggers_task / resource_limit / redirect_target / ssrf_surface / idempotency / data_op_detail / bola_risk** | +| **`TODO`(28列)** | **summary / response / response_content_type / status_codes / exceptions / roles / access_variance / data_op / data_store / side_effects / config_deps / test_file / category_tags / notes / sec_pattern / sec_detail / sec_exposed / sec_evidence / dynamic_verified / csrf_protection / input_validation / audit_logged / triggers_task / resource_limit / redirect_target / ssrf_surface / idempotency / bola_risk** | `TODO` は **ソースを読まないと書けない列**。Phase 2(静的解析)と Phase 3(実機実測)で やっていることを、その1行について行う。埋め方は列定義 README(プライベートリポジトリ側の @@ -127,16 +134,60 @@ python3 tools/api-inventory/scripts/add_authmech.py # auth_mechanism / bola 調査が終わるまでは、少なくとも `data_op` / `auth_required` / `dynamic_verified` を 埋めること。 +## ケース1b: 実機に映らない経路まで含めて漏れを見る + +`reconcile.py` は **実機 url_map** が正。今このコンテナで登録されている経路しか映さない。 +config で無効・プラグイン未導入・設定値が真のときだけ登録される経路は、 +API として存在するのに実機からは見えず、台帳から落ちても誰も気付けない。 + +`detect_routes.py` はソースだけを読んで、6系統(`route` / `expose` / `add_url_rule` / +`rest_config` / `modelview` / `entry_point`)から「あるべき経路」を検知し、台帳と +突き合わせる。**Docker も実機も要らない。** + +```bash +export WEKO_ROOT=/path/to/weko +python3 tools/api-inventory/scripts/detect_routes.py --cross-check # 一覧 +python3 tools/api-inventory/scripts/detect_routes.py --cross-check --gate # 差分0を強制 +``` + +検知したのに台帳に無いものが出たら、次のどちらかを必ず行う。 + +1. 台帳に行を足す(`add_row.py` → TODO を埋める) +2. 経路にならない正当な理由を `$WEKO_API_INVENTORY_DIR/detect_allow.json` に**理由付きで**書く + +```json +{ + "modules/invenio-deposit/invenio_deposit/config.py::DEPOSIT_REST_ENDPOINTS:list_route": + "invenio-deposit の既定値。weko-deposit/config.py が同名で再定義して上書きするため登録されない" +} +``` + +許可リストのキーは `ファイル::識別子` で、**行番号を含めない**。行がずれるたびに +書き直す運用は続かないため。 + +網羅性は「実機(`reconcile.py`)+ 静的(`detect_routes.py`)」の二段で担保する。 +どちらか一方でしか見えない経路があるので、両方を通すこと。 + ## ケース2b: 既存行を修正する ```bash vi "$WEKO_API_INVENTORY_DIR/weko3_api_list_full.tsv" # 本体列(1-57)だけを直す + +# 実装(modules/*.py)にも手が入っているなら、先にこの2本 ★順序が重要 +python3 tools/api-inventory/scripts/refresh_impl.py --write # impl_line を引き直す +python3 tools/api-inventory/scripts/enrich_git.py --write # last_commit / release_tag + python3 tools/api-inventory/scripts/test_coverage.py python3 tools/api-inventory/scripts/prioritize.py python3 tools/api-inventory/scripts/build_checklist.py ``` -派生列(58-65)は手で直しても次の実行で消える。優先度を変えたいときは、 +`enrich_git.py` は `impl_line` の指す関数のコミットを引くので、`impl_line` がずれたまま +回すと**手前の関数のコミットを拾う**(no.480 `publish` は行がずれた状態だと +直前の `get_version` を見て 2019 年のコミットを返した)。必ず `refresh_impl.py` が先。 +台帳だけを直して実装は触っていない(注記の追加など)なら、この2本は不要。 + +派生列(55-62)は手で直しても次の実行で消える。優先度を変えたいときは、 判定の入力側(`security_finding` / `dynamic_verified` / `data_op` / `deprecated`)を 直すか、`prioritize.py` のルールを変える。 @@ -335,7 +386,7 @@ git log --oneline <前回タグ>..HEAD -- '*/alembic/*' # 追加リビジョ python3 .../diff_snapshot.py "$WEKO_API_INVENTORY_DIR/api_snapshot.json" /tmp/snap_new.json cp /tmp/snap_new.json "$WEKO_API_INVENTORY_DIR/api_snapshot.json" python3 .../reconcile.py # A(未収載) を洗い出す -python3 .../add_row.py --append --no <新規のendpoint> # 自動27列だけ埋まる +python3 .../add_row.py --append --no <新規のendpoint> # 自動26列だけ埋まる python3 .../reconcile.py --gate # 0件になるまで繰り返す ``` @@ -494,12 +545,18 @@ python3 .../changed_rows.py <前回タグ> HEAD --out /tmp/rerun.txt ### 7. 再計算してゲートを通す ```bash +python3 .../refresh_impl.py --write # impl_line を新バージョンのソースへ追随させる +python3 .../enrich_git.py --write # last_commit / date / subject / release_tag python3 .../test_coverage.py python3 .../prioritize.py python3 .../build_checklist.py python3 .../reconcile.py --gate # exit 0 を確認 ``` +バージョンアップでは行番号が必ずずれるので、先頭2本を飛ばすと台帳の +`release_tag` が前バージョンのまま残る。**`release_tag` に今回のタグが +1行も出てこなかったら、この2本を回し忘れている。** + > 実績(v2.1.0 / 931行): 特定617 特定不能314 / > P1=82 P2=150 P3=450 P4=4 P5=64 整理対象=20 環境依存=11 対象外=150 / reconcile ✅ 0件。 @@ -534,20 +591,26 @@ git push origin main --follow-tags |---|---|---| | `snapshot.py` | 実機 url_map + ソース | `api_snapshot.json` | | `reconcile.py` | snapshot + full.tsv | 何も書かない(差分を報告するだけ) | +| `detect_routes.py` | ソース(AST)+ full.tsv | 何も書かない(ソース由来の経路と台帳の差を報告するだけ) | | `refresh_impl.py` | full.tsv + 実装ソース(AST) | full.tsv の `impl_line`(`--write` 時のみ) | +| `enrich_git.py` | full.tsv + `git log -L` / `git tag --contains` | full.tsv の `last_commit` / `last_commit_date` / `last_commit_subject` / `release_tag`(`--write` 時のみ)。**`refresh_impl.py` の後に回す** | | `changed_rows.py` | git diff + full.tsv | 再確認対象の `no` 一覧 + 変更ヘルパ関数の報告 | -| `test_coverage.py` | full.tsv + テストコード | full.tsv の 60-64列 | -| `prioritize.py` | full.tsv | full.tsv の 58-59, 65列 + 末尾列順の正規化 | +| `test_coverage.py` | full.tsv + テストコード | full.tsv の 57-61列 | +| `prioritize.py` | full.tsv | full.tsv の 55-56, 62列 + 末尾列順の正規化 | | `build_checklist.py` | full.tsv | **`weko3_api_list.tsv` を全体再生成** | | `add_row.py` | `api_snapshot.json` + git | full.tsv に新規行の雛形を追記(`--append`) | | `apply_probe_results.py` | probe.json | full.tsv の `dynamic_verified`(空欄のみ / `--overwrite` で差し替え、`--keep-history` で旧値を ` ‖ 旧: ` として残す) | | `measure.sh` | `measure_profile.json` | 実測の唯一の入口。上記を固定順で回し `measure_report.md` を書く | | `_ensure_profile.py` / `_read_profile.py` / `_targets.py` / `_report.py` | — | `measure.sh` の内部ヘルパ | +| `schema.py` | — | 列定義の唯一の正。他から import されるだけ | +| `add_reqinfo.py` | full.tsv + 実装ソース | full.tsv の `query_params` / `body_params` / `request_content_type` / `oauth_scope` の空欄 | +| `apply2.py` / `check_reachable.py` / `dump_modelviews.py` | — | Phase 1-3 の使い捨て。パスが決め打ちなので、そのままでは回らない。参考として残してある | | `remeasure.sh` | — | 非推奨。`measure.sh` に統合(案内のみ) | | `add_cols.py` / `add_ssrf_redirect.py` / `add_idempotency.py` / `add_dataop4.py` / `add_authmech.py` | full.tsv + 実装ソース | full.tsv の**空欄/TODO セルのみ**を機械付与 | `test_coverage.py` → `prioritize.py` → `build_checklist.py` は**何度流しても結果が変わらない** -(冪等)。24列版は full.tsv から完全に再現できることを確認済み。 +(冪等)。32列版は full.tsv から完全に再現できることを確認済み。 +`refresh_impl.py` → `enrich_git.py` も、解析対象リビジョンが同じなら冪等。 --- @@ -618,13 +681,13 @@ git describe --tags # タグ ### 1-1. blueprint route を AST 抽出 ```bash -python3 tools/api-inventory/extract_routes.py routes.json +python3 tools/api-inventory/scripts/extract_routes.py routes.json ``` `@blueprint.route` / `add_url_rule` を全 modules から収集。357件程度。 ### 1-2. config駆動 REST エンドポイントを抽出 ```bash -python3 tools/api-inventory/extract_endpoints.py endpoints.json +python3 tools/api-inventory/scripts/extract_endpoints.py endpoints.json ``` `*_REST_ENDPOINTS` config の route 文字列(`//...`)を展開。 @@ -658,7 +721,7 @@ docker exec weko-web-1 bash -lc 'source ~/.virtualenvs/invenio/bin/activate; cd | `add_cols.py` | csrf_protection, input_validation, audit_logged, triggers_task, resource_limit | | `add_ssrf_redirect.py` | redirect_target(オープンリダイレクト), ssrf_surface | | `add_idempotency.py` | idempotency(冪等性) | -| `add_dataop4.py` | data_op_detail(取得/作成/更新/**論理削除/物理削除**) | +| `add_dataop4.py` | data_op(取得/作成/更新/**論理削除/物理削除**。旧 data_op_detail を統合済み) | | `add_authmech.py` | auth_mechanism(decorator/config-factory/modelview), bola_risk | ### 認証・認可の参照辞書(手動で維持) @@ -669,10 +732,18 @@ docker exec weko-web-1 bash -lc 'source ~/.virtualenvs/invenio/bin/activate; cd ### git情報の付与 ```bash -python3 tools/api-inventory/enrich_git.py body.tsv body_enriched.tsv +python3 tools/api-inventory/scripts/refresh_impl.py --write # 先に impl_line +python3 tools/api-inventory/scripts/enrich_git.py # 差分の確認だけ +python3 tools/api-inventory/scripts/enrich_git.py --write # 台帳へ書き戻す +python3 tools/api-inventory/scripts/enrich_git.py --tsv body.tsv --out body_enriched.tsv ``` `git log -L <開始>,<終了>:` で**実装関数の行範囲**の最終コミットを取得(ファイル単位より正確)。 -`git tag --sort=creatordate --contains ` で導入リリースタグ。 +`git tag --sort=creatordate --contains ` で導入リリースタグ。どのタグにも入っていなければ +`(未リリース)`、`impl_file` が実ファイルでない行(Flask-Admin ModelView の総称表記 / +framework 自動生成 / site-packages)は `-`。 + +対象は列名で引く(`last_commit` / `last_commit_date` / `last_commit_subject` / `release_tag`)。 +解析対象リポジトリは `WEKO_ROOT`、台帳は `WEKO_API_INVENTORY_DIR`。 ## Phase 3: 動的検証(実測で裏取り) ★静的だけでは不正確 @@ -715,9 +786,9 @@ python3 tools/api-inventory/probe.py probe_results.json # 未認証+各ロー python3 tools/api-inventory/merge.py out/ merged.tsv # 分割TSVを結合・重複排除・採番 ``` -## Phase 5: チェックリスト版(24列)を生成 +## Phase 5: チェックリスト版(32列)を生成 ```bash -python3 tools/api-inventory/build_checklist.py # 57列 full → 24列 に統合 +python3 tools/api-inventory/scripts/build_checklist.py # 62列 full → 32列 に統合 ``` 派生列を統合: impl(func+file+line), auth(required+method+mechanism), security_flags(CSRF/BOLA/SSRF等8観点を該当のみ), last_change(commit系4列) 等。 @@ -1118,7 +1189,7 @@ python3 scripts/probe_ci.py --only rerun_nos.txt --allow-writes --gate --out pro `probe_ci.py` は `fixtures.json` からプレースホルダを解決するため、まっさらな環境で動く。 - 測定 identity: anon / general / contributor / comadmin / repoadmin / sysadmin -- 測定対象は `--only` で渡した `no` に限定する(全926行を毎PR測ると時間がかかりすぎる) +- 測定対象は `--only` で渡した `no` に限定する(全1048行を毎PR測ると時間がかかりすぎる) - **安全装置**: GET/HEAD 以外は既定でスキップ。`--allow-writes` を明示したときだけ測る (CI のコンテナは使い捨てなので許可してよいが、実環境では既定のままにすること) @@ -1159,7 +1230,7 @@ CI では `changed_rows.py` が出す `rerun_nos.txt`(変更が触れた行)だ ```bash python3 scripts/test_coverage.py # 4観点(正常値/異常値/境界値/例外処理)を判定 python3 scripts/prioritize.py # 優先度を付与(テスト観点を参照するので後に実行) -python3 scripts/build_checklist.py # 24列版(=32列)へ引き継ぐ +python3 scripts/build_checklist.py # 32列版へ引き継ぐ ``` **実行順が重要**: `prioritize.py` は `test_gap` を参照して「テスト観点が確認できない行」を @@ -1176,8 +1247,9 @@ P2 まで引き上げるため、`test_coverage.py` を先に回すこと。 ## prioritize.py -`security_finding` / `security_flags` / `dynamic_verified` / `data_op` / `data_target` / -`method` / `auth` / `test_gap` から、対応優先度を機械判定して台帳に書き戻す。 +`sec_pattern` / `dynamic_verified` / `data_op` / `data_store` / `method` / `auth_required` / +`auth_method` / `deprecated` / `test_gap` から、対応優先度を機械判定して台帳に書き戻す +(チェックリスト版を読ませたときは `security_finding` / `auth` / `data_store` の統合列でも引ける)。 判定基準・凡例・限界はプライベートリポジトリ側の `weko3_api_list_README.md`「priority の凡例」に記載。 「データ破壊」は **既存の実データを不可逆に壊すこと** と定義している。メタデータの diff --git a/tools/api-inventory/scripts/build_checklist.py b/tools/api-inventory/scripts/build_checklist.py index e0a3513378..e1e1bf7a69 100644 --- a/tools/api-inventory/scripts/build_checklist.py +++ b/tools/api-inventory/scripts/build_checklist.py @@ -1,8 +1,12 @@ # -*- coding: utf-8 -*- -"""57列詳細版 → 24列チェックリスト版に統合""" +"""詳細版(62列) → チェックリスト版(32列) に統合する。 + +出力列は schema.CHECKLIST_COLUMNS。列定義は schema.py を直す。 +""" import os, sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from paths import data_path +from schema import CHECKLIST_COLUMNS SRC = sys.argv[1] if len(sys.argv) > 1 else data_path("weko3_api_list_full.tsv") DST = sys.argv[2] if len(sys.argv) > 2 else data_path("weko3_api_list.tsv") def load(p): return [l.rstrip("\n").split("\t") for l in open(p,encoding="utf-8") if l.rstrip("\n")] @@ -11,14 +15,9 @@ def load(p): return [l.rstrip("\n").split("\t") for l in open(p,encoding="utf-8" def g(c,name): i=H[name]; return c[i] if len(c)>i and c[i] not in("","-","不明") else "" -# 24列チェックリスト設計 -NEW=["no","module","api_type","method","uri","impl","summary", - "auth","roles_scope","access_variance","data_op","data_store","side_effects", - "security_finding","security_flags","dynamic_verified", - "api_version","deprecated","test_file","last_change","tags","notes","config_deps","response", - # 末尾に追加する。既存列の位置を動かすと README の awk 例が全て壊れるため。 - "priority","priority_reason", - "test_normal","test_abnormal","test_boundary","test_exception","test_gap","cleanup"] +# 出力列は schema.py が唯一の正。ここに直接並べると README・テスト・台帳の +# どれかと必ずずれる(実測: 「24列」と書かれたまま実体は32列になっていた)。 +NEW = CHECKLIST_COLUMNS out=[NEW] for c in data: diff --git a/tools/api-inventory/scripts/detect_routes.py b/tools/api-inventory/scripts/detect_routes.py new file mode 100644 index 0000000000..5ae754dee5 --- /dev/null +++ b/tools/api-inventory/scripts/detect_routes.py @@ -0,0 +1,600 @@ +# -*- coding: utf-8 -*- +"""ソースだけから「あるべき経路」を検知し、台帳と突き合わせる。 + + python3 detect_routes.py # 検知結果のサマリ + python3 detect_routes.py --json out.json # 明細を出す + python3 detect_routes.py --cross-check # 台帳と突き合わせる + python3 detect_routes.py --cross-check --gate # 未収載があれば exit 1 + python3 detect_routes.py --cross-check --summary-only # 件数のみ(public CI 用) + +## なぜ実機スナップショットだけでは足りないか + +`reconcile.py` は **実機 url_map** を正として突き合わせる。これは「今このコンテナで +登録されている経路」しか見ない。したがって次を構造的に取りこぼす: + + - プラグイン未導入・config で無効になっている経路(`/plugins`, `/api/admin/indexjournal`) + - `suggesters` のように **設定値が真のときだけ**登録される経路(`/api/records/_suggest`) + - 起動後に動的登録される経路(`WidgetDesignPage.url`) + - 別サイト・別設定では有効になる経路 + +これらは「この環境に無い」だけで、**API としては存在する**。台帳から漏れれば +そのまま監査の穴になる。本スクリプトは実機を一切使わず、ソースと設定だけから +経路を検知して台帳と突き合わせる。実機検知(`reconcile.py`)との二段構えにより、 +どちらか一方でしか見えない経路も拾える。 + +## 検知源(すべて AST。実機・Docker 不要) + +| source | 拾うもの | +|---------------|---------| +| `route` | `@bp.route(...)` / `@app.route(...)` | +| `expose` | Flask-Admin の `@expose(...)`(BaseView 派生。url_map には出るが AST 抽出では従来落ちていた) | +| `add_url_rule`| `bp.add_url_rule(...)`。rule が式(config 由来)の場合も view_func で追う | +| `rest_config` | `config.py` の `*ENDPOINTS` 辞書にある `*route` 値(config 駆動 REST) | +| `modelview` | `class X(ModelView)` と `invenio_admin.views` entry point の登録先 URL | +| `entry_point` | `setup.py` の `invenio_base.{apps,blueprints,api_blueprints}` | + +## 突き合わせの規則 + +検知 1件につき、台帳に対応行があるかを次の順で見る。 + + 1. 実装一致 … (impl_file, 関数名) または (impl_file, `Class.method`) + 2. URI 一致 … 正規化 URI。先頭 `/api` の有無は吸収する + 3. 登録名一致 … blueprint / endpoint 名(entry_point・modelview 用) + +どれにも当たらなければ「台帳未収載の疑い」。正当な理由があるものは +`$WEKO_API_INVENTORY_DIR/detect_allow.json` に**理由を書いて**登録する +(`reconcile_allow.json` と同じ思想。黙って消さない)。 + +出力は `--summary-only` を付けると件数だけになる。public リポジトリの CI ログ・ +artifact・PR コメントは誰でも読めるため、経路名を出したくない場面で使う。 +""" +import argparse +import ast +import collections +import json +import os +import re +import sys +import warnings + +warnings.filterwarnings('ignore', category=SyntaxWarning) + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from paths import data_path # noqa: E402 +from snapshot import default_weko_root # noqa: E402 + +SKIP_DIRS = ('/tests', '/examples', '/.tox', '/node_modules', '/cookiecutter', + '/docs/', '/build/', '/.git/') + +# route を持つ辞書キー。invenio 系は route / item_route / list_route / *_route。 +ROUTE_KEY = re.compile(r'(^|_)route$') + +# 台帳の impl_file が実ファイルを指さない行の総称表記。 +# これらは AST では裏取りできない(pip パッケージ・framework 自動生成)。 +NON_SOURCE_IMPL = re.compile(r'^\((provider|site-packages|framework)|^Flask-Admin ModelView') + +SOURCES = ('route', 'expose', 'add_url_rule', 'rest_config', 'modelview', 'entry_point') + + +# -------------------------------------------------------------------------- +# 収集 +# -------------------------------------------------------------------------- + +def iter_py(root, sub='modules'): + base = os.path.join(root, sub) + for dp, dn, fn in os.walk(base): + if any(s in dp.replace(os.sep, '/') + '/' for s in SKIP_DIRS): + dn[:] = [] + continue + for f in sorted(fn): + if f.endswith('.py'): + yield os.path.join(dp, f) + + +def lit(node): + try: + return ast.literal_eval(node) + except Exception: + return None + + +def dotted(node): + """Attribute/Name を 'a.b.c' に戻す。""" + parts = [] + while isinstance(node, ast.Attribute): + parts.append(node.attr) + node = node.value + if isinstance(node, ast.Name): + parts.append(node.id) + return '.'.join(reversed(parts)) + + +def as_view_class(node): + """`X.as_view(...)` から X(クラス名)を取り出す。それ以外は None。""" + if isinstance(node, ast.Call): + node = node.func + if isinstance(node, ast.Attribute) and node.attr == 'as_view': + return dotted(node.value).rsplit('.', 1)[-1] or None + return None + + +def dec_call_name(d): + c = d.func if isinstance(d, ast.Call) else d + if isinstance(c, ast.Attribute): + return c.attr + return getattr(c, 'id', '') + + +def methods_of(call): + if not isinstance(call, ast.Call): + return ['GET'] + for k in call.keywords: + if k.arg == 'methods': + v = lit(k.value) + if v: + return sorted({str(m).upper() for m in v}) + return ['GET'] + + +def _parse(path): + try: + return ast.parse(open(path, encoding='utf-8', errors='replace').read()) + except Exception: + return None + + +def collect_module(root, path): + """1ファイルから route / expose / add_url_rule / modelview を拾う。""" + tree = _parse(path) + if tree is None: + return [] + rel = os.path.relpath(path, root) + out = [] + + # クラス配下の関数 -> 所属クラス名 + owner = {} + bases_of = {} + for cls in [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]: + bases_of[cls.name] = [dotted(b).rsplit('.', 1)[-1] for b in cls.bases] + for m in cls.body: + if isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef)): + owner[id(m)] = cls.name + + # `view_func = SomeResource.as_view(...)` の束縛を追う。 + # config 駆動の create_blueprint() は例外なくこの形で、view_func だけを見ると + # 変数名 'view_func' しか取れず、どのクラスの経路かが分からなくなる。 + asview = collections.defaultdict(list) # 変数名 -> [(lineno, クラス名)] + for n in ast.walk(tree): + if not isinstance(n, ast.Assign): + continue + cls_name = as_view_class(n.value) + if not cls_name: + continue + for t in n.targets: + if isinstance(t, ast.Name): + asview[t.id].append((n.lineno, cls_name)) + + for n in ast.walk(tree): + if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)): + cls = owner.get(id(n)) + for d in n.decorator_list: + name = dec_call_name(d) + if name not in ('route', 'expose'): + continue + rule = lit(d.args[0]) if isinstance(d, ast.Call) and d.args else None + out.append({ + 'source': 'route' if name == 'route' else 'expose', + 'file': rel, 'line': n.lineno, + 'cls': cls, 'func': n.name, + 'qual': f'{cls}.{n.name}' if cls else n.name, + 'rule': rule, + 'rule_expr': (ast.unparse(d.args[0]) + if isinstance(d, ast.Call) and d.args and rule is None + else None), + 'methods': methods_of(d), + 'holder': (dotted(d.func.value) + if isinstance(d, ast.Call) and isinstance(d.func, ast.Attribute) + else None), + }) + + if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) \ + and n.func.attr == 'add_url_rule': + kw = {k.arg: k.value for k in n.keywords} + rnode = n.args[0] if n.args else kw.get('rule') + vf = kw.get('view_func') + if vf is None and len(n.args) > 2: + vf = n.args[2] + vname = dotted(vf).rsplit('.', 1)[-1] if vf is not None else None + vcls = as_view_class(vf) if vf is not None else None + if not vcls and vname: + # 直前に束縛された `X.as_view(...)` に遡る + cands = [(ln, c) for ln, c in asview.get(vname, []) if ln <= n.lineno] + if cands: + vcls = max(cands)[1] + ep = lit(kw['endpoint']) if 'endpoint' in kw else ( + lit(n.args[1]) if len(n.args) > 1 else None) + # `add_url_rule(**rule)` のような一括登録は、rule も view_func も + # 静的には取り出せない。個々の経路は config 側(rest_config)で検知するので、 + # ここでは「config 駆動の一括登録がある」事実だけを記録して門番からは外す。 + dispatch = (rnode is None and vf is None) + out.append({ + 'source': 'add_url_rule', + 'file': rel, 'line': n.lineno, + 'cls': vcls, + 'func': vname, + 'qual': vcls or vname or (ep if isinstance(ep, str) else None), + 'endpoint': ep if isinstance(ep, str) else None, + 'rule': lit(rnode) if rnode is not None else None, + 'rule_expr': (ast.unparse(rnode) + if rnode is not None and lit(rnode) is None else None), + 'methods': methods_of(n), + 'holder': dotted(n.func.value), + 'dispatch': dispatch, + }) + + # ModelView 派生クラス(/admin// が自動生成される) + for cls_name, bases in bases_of.items(): + if any(b.endswith('ModelView') for b in bases): + out.append({'source': 'modelview', 'file': rel, 'line': 0, + 'cls': cls_name, 'func': None, 'qual': cls_name, + 'rule': None, 'rule_expr': None, 'methods': ['GET'], + 'holder': None}) + return out + + +def collect_rest_config(root, path): + """config.py の `*ENDPOINTS` 辞書から route 値を拾う。""" + tree = _parse(path) + if tree is None: + return [] + rel = os.path.relpath(path, root) + out = [] + for n in tree.body: + if not isinstance(n, ast.Assign) or not isinstance(n.targets[0], ast.Name): + continue + var = n.targets[0].id + if 'ENDPOINTS' not in var: + continue + for sub in ast.walk(n.value): + if not isinstance(sub, ast.Dict): + continue + for k, v in zip(sub.keys, sub.values): + if not (isinstance(k, ast.Constant) and isinstance(k.value, str)): + continue + if not ROUTE_KEY.search(k.value): + continue + val = lit(v) + if not isinstance(val, str) or not val.startswith('/'): + continue + out.append({'source': 'rest_config', 'file': rel, + 'line': getattr(v, 'lineno', n.lineno), + 'cls': None, 'func': None, + 'qual': f'{var}:{k.value}', + 'rule': val, 'rule_expr': None, + 'methods': ['GET'], 'holder': var}) + return out + + +# 経路を生む登録だけを見る。`invenio_base.apps` / `api_apps` は拡張(Flask extension)の +# 登録で、それ自体は経路を作らない。混ぜると常時 13件の偽陽性になり、ゲートが形骸化する。 +EP_GROUPS = ('invenio_base.blueprints', 'invenio_base.api_blueprints', + 'invenio_admin.views') + + +def collect_adminview_dicts(root, path): + """`xxx_adminview = {'view_class': FooView, ...}` を集める。 + + `invenio_admin.views` entry point は `module:xxx_adminview` を指すだけなので、 + その辞書が指すクラス名まで辿らないと Flask-Admin の登録名が分からない + (`session_adminview` -> `SessionActivityView` -> 登録名 `sessionactivity`)。 + """ + tree = _parse(path) + if tree is None: + return {} + rel = os.path.relpath(path, root) + mod = rel[:-3].replace('/', '.').replace(os.sep, '.') + mod = mod.split('.', 2)[-1] if mod.startswith('modules.') else mod + out = {} + for n in tree.body: + if not isinstance(n, ast.Assign) or not isinstance(n.targets[0], ast.Name): + continue + var = n.targets[0].id + if 'adminview' not in var and not var.endswith('_view'): + continue + names = [x.id for x in ast.walk(n.value) if isinstance(x, ast.Name)] + if names: + out[f'{mod}:{var}'] = names + return out + + +def collect_entry_points(root, path): + """setup.py の entry_points から blueprint / admin view の登録を拾う。""" + tree = _parse(path) + if tree is None: + return [] + rel = os.path.relpath(path, root) + out = [] + for n in ast.walk(tree): + if not isinstance(n, ast.Dict): + continue + for k, v in zip(n.keys, n.values): + if not (isinstance(k, ast.Constant) and k.value in EP_GROUPS): + continue + for item in (lit(v) or []): + if not isinstance(item, str) or '=' not in item: + continue + name, target = (x.strip() for x in item.split('=', 1)) + out.append({'source': 'entry_point', 'file': rel, + 'line': getattr(v, 'lineno', n.lineno), + 'cls': None, 'func': target.rsplit(':', 1)[-1], + 'qual': name, 'target': target, + 'rule': None, 'rule_expr': None, + 'methods': ['GET'], 'holder': k.value}) + return out + + +def detect(root): + """全検知源を回して検知一覧を返す。""" + found = [] + adminviews = {} + for p in iter_py(root): + found += collect_module(root, p) + if os.path.basename(p) == 'config.py': + found += collect_rest_config(root, p) + if os.path.basename(p) in ('admin.py', 'views.py'): + adminviews.update(collect_adminview_dicts(root, p)) + for dp, dn, fn in os.walk(os.path.join(root, 'modules')): + if any(s in dp.replace(os.sep, '/') + '/' for s in SKIP_DIRS): + dn[:] = [] + continue + if 'setup.py' in fn: + found += collect_entry_points(root, os.path.join(dp, 'setup.py')) + # entry point が指す `*_adminview` を、その辞書が参照するクラス名まで解決する + for d in found: + if d['source'] == 'entry_point' and d.get('target') in adminviews: + d['via'] = adminviews[d['target']] + return found + + +# -------------------------------------------------------------------------- +# 台帳との突き合わせ +# -------------------------------------------------------------------------- + +def norm_uri(u): + u = u.strip() + if len(u) > 1 and u.endswith('/'): + u = u[:-1] + return u + + +def uri_variants(u): + """先頭 `/api` の有無を吸収した比較キー。""" + u = norm_uri(u) + out = {u} + if u.startswith('/api'): + out.add(norm_uri(u[4:]) or '/') + else: + out.add(norm_uri('/api' + u)) + return {x for x in out if x} + + +def load_ledger(path): + rows = [l.rstrip('\n').split('\t') for l in open(path, encoding='utf-8') if l.strip()] + hdr = rows[0] + H = {n: i for i, n in enumerate(hdr)} + data = rows[1:] + + by_impl = collections.defaultdict(list) + by_uri = collections.defaultdict(list) + by_name = collections.defaultdict(list) + by_ep_prefix = collections.defaultdict(list) + for r in data: + no = r[H['no']] + f = r[H['impl_file']] + fn = r[H['impl_func']] + # impl_func は `A→B`(委譲) / `A/B`(別名) / `f(...)`(内訳) の表記を取りうる + for part in re.split(r'[/→]', fn.split('(')[0]): + part = part.strip() + if not part: + continue + by_impl[(f, part)].append(no) + by_impl[(f, part.rsplit('.', 1)[-1])].append(no) + if '.' in part: + # `Class.method` はクラス名だけでも引けるようにする。 + # add_url_rule は `Class.as_view(...)` を渡すので、台帳側の + # メソッド名までは分からない。 + by_impl[(f, part.split('.', 1)[0])].append(no) + for u in r[H['uri']].split(';'): + for v in uri_variants(u): + by_uri[v].append(no) + for col in ('blueprint', 'endpoint'): + v = r[H[col]].strip() + if v and v not in ('-', 'TODO'): + by_name[v].append(no) + by_name[v.rsplit('.', 1)[-1]].append(no) + # Flask-Admin は `role.index_view` のように「登録名.ビュー名」になる。 + # ModelView クラスや *_adminview からはビュー名まで分からないので、 + # 登録名だけでも引けるようにする。 + if '.' in v: + by_ep_prefix[v.split('.', 1)[0]].append(no) + return {'hdr': hdr, 'H': H, 'rows': data, 'by_impl': by_impl, + 'by_uri': by_uri, 'by_name': by_name, 'by_ep_prefix': by_ep_prefix} + + +ADMIN_SUFFIX = re.compile(r'(ModelView|AdminView|View|_adminview|_view)$') + + +def admin_prefixes(d): + """ModelView クラス名 / `*_adminview` から Flask-Admin の登録名候補を作る。 + + `RoleView` -> `role`、`SessionActivityView` -> `sessionactivity`、 + `user_adminview` -> `user`。登録名は台帳の endpoint 列の `.` の手前に出る。 + """ + if d['source'] not in ('modelview', 'entry_point'): + return [] + out = [] + for raw in [d.get('cls'), d.get('func'), d.get('qual')] + list(d.get('via') or []): + if not raw: + continue + base = ADMIN_SUFFIX.sub('', raw) + if base: + out.append(base.lower()) + return out + + +def match(d, L): + """検知1件を台帳に当てる。当たれば (規則, [no]) を返す。""" + f, q, fn = d['file'], d.get('qual'), d.get('func') + for key in (q, fn): + if key and (f, key) in L['by_impl']: + return 'impl', L['by_impl'][(f, key)] + if d.get('rule'): + for v in uri_variants(d['rule']): + if v in L['by_uri']: + return 'uri', L['by_uri'][v] + for key in (d.get('qual'), d.get('func'), d.get('cls'), d.get('endpoint')): + if key and key in L['by_name']: + return 'name', L['by_name'][key] + for key in admin_prefixes(d): + if key in L['by_ep_prefix']: + return 'admin', L['by_ep_prefix'][key] + # entry_point / modelview は「登録名」でしか追えないことがある。 + # 同じファイルの行が台帳にあれば、その登録は台帳に届いているとみなす。 + if d['source'] in ('entry_point', 'modelview'): + mod = d['file'].split('/')[1] if '/' in d['file'] else '' + for (ff, _), nos in L['by_impl'].items(): + if mod and ff.startswith(f'modules/{mod}/'): + return 'module', nos + return None, [] + + +def load_allow(): + p = data_path('detect_allow.json', required=False) + if not p or not os.path.isfile(p): + return {} + try: + a = json.load(open(p, encoding='utf-8')) + except Exception: + return {} + return {k: v for k, v in a.items() if not k.startswith('_')} + + +def allow_key(d): + """許可リストのキー。ファイル+識別子で、行番号の移動に影響されない形にする。""" + return f"{d['file']}::{d.get('qual') or d.get('rule') or '?'}" + + +# -------------------------------------------------------------------------- + +def main(): + p = argparse.ArgumentParser( + description='ソースだけから経路を検知し、台帳と突き合わせる') + p.add_argument('--weko-root', default=None, help='既定: $WEKO_ROOT') + p.add_argument('--tsv', default=None, + help='既定: $WEKO_API_INVENTORY_DIR/weko3_api_list_full.tsv') + p.add_argument('--json', help='検知明細の書き出し先') + p.add_argument('--cross-check', action='store_true', help='台帳と突き合わせる') + p.add_argument('--gate', action='store_true', help='未収載があれば exit 1') + p.add_argument('--summary-only', action='store_true', + help='件数のみ出力する(public な CI ログに経路名を出さない)') + p.add_argument('--out', help='Markdown 出力先') + a = p.parse_args() + + root = a.weko_root or default_weko_root() + found = detect(root) + + by_src = collections.Counter(d['source'] for d in found) + L = ['# ソース由来の経路検知', '', f'- 解析対象: `{root}`', ''] + L += ['| 検知源 | 件数 |', '|---|---:|'] + for s in SOURCES: + L.append(f'| `{s}` | {by_src.get(s, 0)} |') + L += [f'| **計** | **{len(found)}** |', ''] + + unexplained = [] + if a.cross_check: + tsv = a.tsv or data_path('weko3_api_list_full.tsv') + led = load_ledger(tsv) + allow = load_allow() + miss, known, dispatched, hit = [], [], [], collections.Counter() + matched_nos = set() + for d in found: + rule, nos = match(d, led) + if nos: + hit[d['source']] += 1 + matched_nos.update(nos) + continue + if d.get('dispatch'): + dispatched.append(d) + continue + k = allow_key(d) + if k in allow: + d = dict(d, reason=allow[k]) + known.append(d) + else: + miss.append(d) + unexplained = miss + + L += [f"## 判定: {'❌ 台帳未収載の疑いあり' if miss else '✅ 全検知が台帳に対応'}" + f' ({len(miss)}件)', ''] + L += ['| 検知源 | 検知 | 台帳に対応 | 未収載 | 既知・許容 |', '|---|---:|---:|---:|---:|'] + for s in SOURCES: + n = by_src.get(s, 0) + m = sum(1 for x in miss if x['source'] == s) + kn = sum(1 for x in known if x['source'] == s) + L.append(f'| `{s}` | {n} | {hit.get(s, 0)} | {m} | {kn} |') + L.append('') + if dispatched: + L += [f'> `add_url_rule(**rule)` 形式の config 駆動一括登録が ' + f'{len(dispatched)} 箇所。個々の経路は `rest_config` 側で検知する。', + ''] + + if miss and not a.summary_only: + L += ['## 台帳未収載の疑い — 行の追加、または理由付きで ' + '`detect_allow.json` へ登録が必要', ''] + for d in miss: + where = f"{d['file']}:{d['line']}" + what = d.get('rule') or d.get('rule_expr') or d.get('qual') or '?' + L.append(f"- `{d['source']}` `{what}` — {where} " + f"({d.get('qual') or ''} {','.join(d['methods'])})") + L.append('') + if known and not a.summary_only: + L += ['## 既知・許容(`detect_allow.json`)', ''] + for d in known: + L.append(f"- `{d['source']}` `{allow_key(d)}` — {d['reason']}") + L.append('') + + # 静的に裏取りできなかった台帳行。ModelView・framework・pip 由来は + # ソースに定義が無いので当然入る。数が急に動いたら台帳側の異常を疑う。 + unbacked = [r for r in led['rows'] if r[led['H']['no']] not in matched_nos] + real = [r for r in unbacked + if not NON_SOURCE_IMPL.match(r[led['H']['impl_file']])] + L += ['## 参考: 静的検知と結びつかなかった台帳行', '', + f'- 全体: {len(unbacked)} / {len(led["rows"])} 行', + f'- うち実ファイルを持つ行: {len(real)}' + '(pip・framework・ModelView 総称表記を除いた数)', ''] + if real and not a.summary_only: + for r in real[:40]: + L.append(f"- no={r[led['H']['no']]} `{r[led['H']['uri']][:60]}` " + f"— {r[led['H']['impl_file']]}") + if len(real) > 40: + L.append(f'- … 他 {len(real) - 40} 行') + L.append('') + + md = '\n'.join(L) + if a.out: + open(a.out, 'w', encoding='utf-8').write(md + '\n') + print(f'{a.out} を書き出しました') + else: + print(md) + if a.json: + json.dump({'meta': {'weko_root': root, 'counts': dict(by_src)}, + 'detections': found}, + open(a.json, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) + print(f'{a.json} を書き出しました') + + if a.gate and unexplained: + sys.exit(1) + + +if __name__ == '__main__': + main() diff --git a/tools/api-inventory/scripts/enrich_git.py b/tools/api-inventory/scripts/enrich_git.py index 6845ca69cb..a873f9bfc2 100644 --- a/tools/api-inventory/scripts/enrich_git.py +++ b/tools/api-inventory/scripts/enrich_git.py @@ -1,20 +1,45 @@ # -*- coding: utf-8 -*- -"""TSV の 36-39列 (last_commit / date / subject / release_tag) を git から埋める。 +"""台帳の git 由来4列を引き直す。 -使い方: python3 enrich_git.py -- 14列目 impl_file (repo相対), 15列目 impl_line を見て、その行を含む - def/class の行範囲を AST で特定し `git log -1 -L a,b:file` で最終コミットを取る。 -- release_tag は `git tag --sort=creatordate --contains ` の先頭 (最初に入ったリリース)。 + last_commit / last_commit_date / last_commit_subject / release_tag + + python3 enrich_git.py # 差分を表示するだけ + python3 enrich_git.py --write # 台帳に書き戻す + python3 enrich_git.py --tsv in.tsv --out out.tsv # 別ファイルへ出す(初回生成向け) + +`impl_file`(リポジトリ相対) と `impl_line` が指す def/class の行範囲を AST で特定し、 +`git log -1 -L <開始>,<終了>:` でその範囲を最後に変更したコミットを取る +(ファイル単位で見るより正確)。`release_tag` は +`git tag --sort=creatordate --contains ` の先頭 = 最初に入ったリリース。 +コミットがどのタグにも入っていなければ `(未リリース)`。 + +`impl_file` が実ファイルでない行(Flask-Admin ModelView の総称表記 / framework 自動生成 / +site-packages)は git で追えないので4列とも `-` にする。 + +★ `impl_line` がずれていると手前の関数のコミットを拾う。**必ず `refresh_impl.py --write` + を先に回すこと。** バージョンアップでデコレータが増えると行番号は簡単にずれる。 + +解析対象リポジトリは `WEKO_ROOT`、台帳は `WEKO_API_INVENTORY_DIR` で指す。 """ -import ast, os, subprocess, sys, functools +import argparse +import ast +import functools +import os +import subprocess +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from paths import data_path # noqa: E402 +from changed_rows import default_weko_root # noqa: E402 + +COLS = ('last_commit', 'last_commit_date', 'last_commit_subject', 'release_tag') +EMPTY = ('-', '-', '-', '-') -ROOT = '/home/mhaya/wekov2' -NCOL = 41 @functools.lru_cache(maxsize=None) -def def_ranges(path): - """ファイル内の全 def/class の (start, end) をリストで返す。""" - fp = os.path.join(ROOT, path) +def def_ranges(root, path): + """ファイル内の全 def/class の (開始, 終了) を返す。開始はデコレータ行を含む。""" + fp = os.path.join(root, path) if not os.path.isfile(fp): return () try: @@ -28,69 +53,103 @@ def def_ranges(path): out.append((s, getattr(n, 'end_lineno', n.lineno))) return tuple(out) -def enclosing(path, line): + +def enclosing(root, path, line): """line を含む最小の def/class 範囲。無ければ (line, line)。""" best = None - for s, e in def_ranges(path): - if s <= line <= e: - if best is None or (e - s) < (best[1] - best[0]): - best = (s, e) + for s, e in def_ranges(root, path): + if s <= line <= e and (best is None or (e - s) < (best[1] - best[0])): + best = (s, e) return best or (line, line) + @functools.lru_cache(maxsize=None) -def git_last(path, start, end): +def git_last(root, path, start, end): try: r = subprocess.run( - ['git', '-C', ROOT, 'log', '-1', '--format=%h\x1f%ad\x1f%s', '--date=short', + ['git', '-C', root, 'log', '-1', '--format=%h\x1f%ad\x1f%s', '--date=short', '-L', f'{start},{end}:{path}'], - capture_output=True, text=True, timeout=60) + capture_output=True, text=True, timeout=120) except Exception: return ('', '', '') line = r.stdout.split('\n', 1)[0] if r.stdout else '' parts = line.split('\x1f') if len(parts) != 3: return ('', '', '') - subj = parts[2].replace('\t', ' ').strip() - return (parts[0], parts[1], subj[:120]) + return (parts[0], parts[1], parts[2].replace('\t', ' ').strip()[:120]) + @functools.lru_cache(maxsize=None) -def git_tag(sha): +def git_tag(root, sha): if not sha: return '' - r = subprocess.run(['git', '-C', ROOT, 'tag', '--sort=creatordate', '--contains', sha], - capture_output=True, text=True, timeout=60) - tags = [t for t in r.stdout.split('\n') if t.strip()] + try: + r = subprocess.run(['git', '-C', root, 'tag', '--sort=creatordate', '--contains', sha], + capture_output=True, text=True, timeout=120) + except Exception: + return '' + tags = [t.strip() for t in r.stdout.split('\n') if t.strip()] return tags[0] if tags else '(未リリース)' -def main(src, dst): - out = [] - bad = 0 - for i, raw in enumerate(open(src, encoding='utf-8'), 1): - raw = raw.rstrip('\n') - if not raw.strip(): - continue - c = raw.split('\t') - if len(c) < NCOL: - c += [''] * (NCOL - len(c)) - elif len(c) > NCOL: - sys.stderr.write(f'WARN line {i}: {len(c)} cols (>41), truncating tail into notes\n') - c = c[:NCOL - 1] + [' | '.join(c[NCOL - 1:])] - bad += 1 - path, ln = c[13].strip(), c[14].strip() - sha = date = subj = tag = '' + +def main(): + p = argparse.ArgumentParser() + p.add_argument('--tsv', default=None, help='入力の台帳(既定: $WEKO_API_INVENTORY_DIR の57列版)') + p.add_argument('--out', default=None, help='出力先(既定: --tsv と同じ = 上書き)') + p.add_argument('--write', action='store_true', help='書き戻す(付けないと差分表示のみ)') + a = p.parse_args() + + tsv = a.tsv or data_path('weko3_api_list_full.tsv') + dst = a.out or tsv + root = default_weko_root() + + rows = [l.rstrip('\n').split('\t') for l in open(tsv, encoding='utf-8') if l.rstrip('\n')] + head = {n: i for i, n in enumerate(rows[0])} + for c in ('impl_file', 'impl_line') + COLS: + if c not in head: + sys.exit(f'列が無い: {c}') + i_file, i_line = head['impl_file'], head['impl_line'] + idx = [head[c] for c in COLS] + + changed, same, nofile = [], 0, 0 + for r in rows[1:]: + if len(r) < len(rows[0]): + r += [''] * (len(rows[0]) - len(r)) + path, ln = r[i_file].strip(), r[i_line].strip() try: line = int(ln) except ValueError: line = None - if path and line and os.path.isfile(os.path.join(ROOT, path)): - s, e = enclosing(path, line) - sha, date, subj = git_last(path, s, e) - tag = git_tag(sha) - c[35], c[36], c[37], c[38] = sha or '-', date or '-', subj or '-', tag or '-' - out.append('\t'.join(x.replace('\t', ' ') for x in c)) - with open(dst, 'w', encoding='utf-8') as f: - f.write('\n'.join(out) + '\n') - print(f'rows={len(out)} col_fixups={bad}') + if path and line and os.path.isfile(os.path.join(root, path)): + s, e = enclosing(root, path, line) + sha, date, subj = git_last(root, path, s, e) + new = (sha or '-', date or '-', subj or '-', git_tag(root, sha) or '-') + else: + nofile += 1 + new = EMPTY + old = tuple(r[i] for i in idx) + if old != new: + changed.append((r[0], path, ln, old, new)) + for i, v in zip(idx, new): + r[i] = v + else: + same += 1 + + print(f'{tsv} (root={root})') + print(f' 更新 {len(changed)} / 変化なし {same} / 追えない行 {nofile}') + for no, path, ln, old, new in changed[:40]: + print(f' no={no:<5} {path}:{ln}') + print(f' {old[0]} / {old[3]} -> {new[0]} ({new[1]}) / {new[3]}') + if len(changed) > 40: + print(f' ... 他 {len(changed) - 40} 件') + + if a.write: + with open(dst, 'w', encoding='utf-8') as f: + f.write('\n'.join('\t'.join(x.replace('\t', ' ') for x in r) for r in rows) + '\n') + print(f' → {dst} に書き戻した') + else: + print(' (--write を付けると書き戻す)') + if __name__ == '__main__': - main(sys.argv[1], sys.argv[2]) + main() diff --git a/tools/api-inventory/scripts/fixtures.py b/tools/api-inventory/scripts/fixtures.py index 1309710c90..50c4ade365 100644 --- a/tools/api-inventory/scripts/fixtures.py +++ b/tools/api-inventory/scripts/fixtures.py @@ -1289,8 +1289,9 @@ def main(): p = argparse.ArgumentParser(description='動的検証用フィクスチャを投入する') p.add_argument('--out', default=os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'fixtures.json')) - p.add_argument('--container', default='', - help='投入先コンテナ(省略時は compose ラベルから自動検出)') + p.add_argument('--container', default=os.environ.get('WEKO_WEB_CONTAINER', ''), + help='投入先コンテナ。既定は $WEKO_WEB_CONTAINER、' + 'それも無ければ compose ラベルから自動検出') p.add_argument('--password', default=PASSWORD) p.add_argument('--scale', type=int, default=0, help='デモ用アイテムの件数。0(既定)はテストに必要な最低限のみ。' diff --git a/tools/api-inventory/scripts/prioritize.py b/tools/api-inventory/scripts/prioritize.py index 2a1f3dde1f..ff3792df34 100644 --- a/tools/api-inventory/scripts/prioritize.py +++ b/tools/api-inventory/scripts/prioritize.py @@ -143,10 +143,11 @@ def classify(c, H): unused_src = '経路なし(実機 url_map に未登録)' def bump(pri, why): - """テスト観点が確認できない行を P2 まで引き上げる。 + """テスト観点が確認できない行を P3 まで引き上げる。 認可上の問題が無くても「4観点のチェックが確認できない」なら確認対象に - 上げる。ただし認可の欠陥と同列にはしないため **上限は P2**。 + 上げる。ただし認可の欠陥と同列にはしないため **上限は P3** + (モジュール冒頭の判定基準と揃えること)。 """ if not gap or gap == '-': return pri, why diff --git a/tools/api-inventory/scripts/schema.py b/tools/api-inventory/scripts/schema.py new file mode 100644 index 0000000000..f771eda9a2 --- /dev/null +++ b/tools/api-inventory/scripts/schema.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +"""台帳の列定義。**列に関する唯一の正**。 + +列名・列数はツール・README・awk の例・CI の検算がそれぞれ持っていて、これまでは +どれかを直しても他が古びるだけだった(v2.0.4 時点で README は 57列/24列/926行 と +書いたまま、実ファイルは 62列/32列/1048行 になっていた)。 + +ここを直せば、次がまとめて追随する: + + - `build_checklist.py` … 24列版(実際は32列)の出力列 + - 公開側 `tests/` … README の記述との整合検査 + - 非公開側 `tests/` … 実台帳のヘッダ検査 + +列名だけなので public リポジトリに置いてよい(所見・実証結果は含まない)。 +""" + +# 詳細版 weko3_api_list_full.tsv の 62列。 +FULL_COLUMNS = [ + # 経路の同定 (1-15) + 'no', 'module', 'api_type', 'app', 'method', 'uri', 'path_params', + 'query_params', 'body_params', 'request_content_type', 'blueprint', + 'endpoint', 'impl_func', 'impl_file', 'impl_line', + # 入出力 (16-20) + 'summary', 'response', 'response_content_type', 'status_codes', 'exceptions', + # 認証・認可 (21-25) + 'auth_required', 'auth_method', 'oauth_scope', 'roles', 'access_variance', + # データ操作・運用 (26-33) + 'data_op', 'data_store', 'side_effects', 'cache_ratelimit', 'config_deps', + 'api_version', 'deprecated', 'test_file', + # git 由来 (34-37) — enrich_git.py が上書きする + 'last_commit', 'last_commit_date', 'last_commit_subject', 'release_tag', + # 分類・備考 (38-39) + 'category_tags', 'notes', + # セキュリティ所見と実測 (40-44) + 'sec_pattern', 'sec_detail', 'sec_exposed', 'sec_evidence', 'dynamic_verified', + # 攻撃観点 (45-54) + 'csrf_protection', 'input_validation', 'audit_logged', 'triggers_task', + 'resource_limit', 'redirect_target', 'ssrf_surface', 'idempotency', + 'auth_mechanism', 'bola_risk', + # 優先度 (55-56) — prioritize.py が上書きする + 'priority', 'priority_reason', + # テスト観点と整理 (57-62) — test_coverage.py / prioritize.py が上書きする + 'test_normal', 'test_abnormal', 'test_boundary', 'test_exception', + 'test_gap', 'cleanup', +] + +# チェックリスト版 weko3_api_list.tsv の 32列。build_checklist.py が生成する。 +# **末尾に足す。** 既存列の位置を動かすと README の awk 例(`$20` など)が全部壊れる。 +CHECKLIST_COLUMNS = [ + 'no', 'module', 'api_type', 'method', 'uri', 'impl', 'summary', + 'auth', 'roles_scope', 'access_variance', 'data_op', 'data_store', + 'side_effects', 'security_finding', 'security_flags', 'dynamic_verified', + 'api_version', 'deprecated', 'test_file', 'last_change', 'tags', 'notes', + 'config_deps', 'response', + 'priority', 'priority_reason', + 'test_normal', 'test_abnormal', 'test_boundary', 'test_exception', + 'test_gap', 'cleanup', +] + +# スクリプトが毎回上書きする派生列。手編集しても次の実行で消える。 +DERIVED_COLUMNS = [ + 'priority', 'priority_reason', + 'test_normal', 'test_abnormal', 'test_boundary', 'test_exception', + 'test_gap', 'cleanup', +] + +# 値の語彙。台帳側で新しい値が現れたら、まずここに足すか、書き間違いを疑う。 +APPS = ['UIアプリ', 'APIアプリ(/api)', '両方'] +API_TYPES = [ + 'REST API', 'AJAX', '画面ビュー', '管理画面', '管理画面(ModelView自動生成)', + 'ファイル配信', 'フレームワーク', 'OAI-PMH', 'SWORD', 'ResourceSync', + 'RSS/Sitemap', '認証', +] +HTTP_METHODS = ['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS'] +AUTH_REQUIRED = ['要', '要(管理)', '要(設計上)', '不要', '任意(匿名可)'] +PRIORITIES = ['P1', 'P2', 'P3', 'P4', 'P5', '整理対象', '環境依存', '対象外'] +TEST_MARKS = ['○', '-', '?'] +TEST_ASPECTS = [('test_normal', '正常値'), ('test_abnormal', '異常値'), + ('test_boundary', '境界値'), ('test_exception', '例外処理')] + +assert len(FULL_COLUMNS) == 62 +assert len(CHECKLIST_COLUMNS) == 32 +assert len(set(FULL_COLUMNS)) == len(FULL_COLUMNS) +assert len(set(CHECKLIST_COLUMNS)) == len(CHECKLIST_COLUMNS) +assert FULL_COLUMNS[-8:] == DERIVED_COLUMNS diff --git a/tools/api-inventory/scripts/snapshot.py b/tools/api-inventory/scripts/snapshot.py index 3e58a29835..dface0b14f 100644 --- a/tools/api-inventory/scripts/snapshot.py +++ b/tools/api-inventory/scripts/snapshot.py @@ -4,10 +4,15 @@ python3 snapshot.py --out api_snapshot.json なぜ実機 url_map が正か: - AST で `@bp.route` / `add_url_rule` を全部拾っても 357件。実機は 903ルート。static 配信ルートも収録する(is_static で識別)。 - 差の 52% は Flask-Admin の自動生成(223) / `@expose`(約100) / config駆動 REST(約30) / + AST で `@bp.route` / `add_url_rule` を拾っても 357件。実機は 903ルート。static 配信ルートも収録する(is_static で識別)。 + 差は Flask-Admin の自動生成 / `@expose` / config駆動 REST / modules配下に無い pip パッケージ / route が式の add_url_rule / framework 由来。 + ただし **実機 url_map はこの環境で登録された経路しか映さない**。config で無効・ + プラグイン未導入・設定値が真のときだけ登録される経路は、API として存在するのに + ここには出ない。その穴は `detect_routes.py`(ソースだけから 6系統で検知)が埋める。 + 台帳の網羅性は「実機(reconcile.py) + 静的(detect_routes.py)」の二段で担保する。 + 出力構造: meta … 生成条件(リビジョン・プロファイル・件数) endpoints … 経路ごとの属性 + auth_hash/body_hash @@ -189,8 +194,9 @@ def resolve_container(name): sys.exit('web コンテナが見つかりません。スタックを起動してください。\n' ' 例: ./install.sh / docker compose -p weko up -d web\n' ' 起動済みなら --container <名前> を明示してください。') - sys.exit('web コンテナが複数あります。--container で指定してください:\n ' - + '\n '.join(cands)) + sys.exit('web コンテナが複数あります。--container か $WEKO_WEB_CONTAINER で' + '指定してください:\n ' + '\n '.join(cands) + + '\n (compose の service=web ラベルは WEKO3 以外のスタックも持ちうる)') def live_dump(container, workdir): @@ -541,8 +547,9 @@ def main(): p = argparse.ArgumentParser(description='API スナップショットを生成する') p.add_argument('--out', default='api_snapshot.json') p.add_argument('--weko-root', default=default_weko_root()) - p.add_argument('--container', default='', - help='実機ダンプ元のコンテナ名(省略時は compose ラベルから自動検出)') + p.add_argument('--container', default=os.environ.get('WEKO_WEB_CONTAINER', ''), + help='実機ダンプ元のコンテナ名。既定は $WEKO_WEB_CONTAINER、' + 'それも無ければ compose ラベルから自動検出') p.add_argument('--dump', help='ダンプ済み JSON を使う(コンテナ起動不要)') p.add_argument('--profile', default='default', help='設定プロファイル名(条件付き登録の差を区別する)') p.add_argument('--workdir', help='中間ファイル置き場') diff --git a/tools/api-inventory/tests/README.md b/tools/api-inventory/tests/README.md new file mode 100644 index 0000000000..85c6ef5442 --- /dev/null +++ b/tools/api-inventory/tests/README.md @@ -0,0 +1,37 @@ +# 台帳ツールの単体テスト + +```bash +cd tools/api-inventory +python3 -m pytest # 1秒程度。Docker も実機も台帳も要らない +``` + +## 何を守っているか + +台帳づくりの失敗は**静かに起きる**。列名を変えてもスクリプトは例外を出さずに +空欄を書き、検知器が1系統死んでも件数が減るだけで、ゲートは緑のまま通る。 +ここでは「壊れたことが分かる」ための最低線を固定している。 + +| ファイル | 守るもの | +|---|---| +| `test_reconcile.py` | 実機 url_map との突き合わせ。A〜E の各検出が**本当に鳴る**こと | +| `test_detect_routes.py` | ソース由来の経路検知。6系統それぞれが拾えること、許可リストが効くこと | +| `test_build_checklist.py` | チェックリスト版の生成。参照している列名が実在すること | +| `test_prioritize.py` | 優先度判定の分岐。見るべき行が埋もれないこと | +| `test_test_coverage.py` | テスト4観点の判定。緩む方向に壊れていないこと | +| `test_merge.py` | Phase 1 の合流・採番・列数の正規化 | +| `test_docs.py` | 手順書(`scripts/README.md`)が実装とずれていないこと | + +## 方針 + +- **データを使わない。** このリポジトリは public。台帳・スナップショット・所見は + 一切置かない。テストは合成した最小のリポジトリと最小の台帳をその場で組み立てる。 + 実台帳そのものの検査はプライベートリポジトリ側の `tests/` が持つ。 +- **秘匿の担保もテストする。** `--summary-only` が経路名を出さないことを、 + reconcile と detect_routes の両方で確かめている。public な CI のログ・artifact・ + PR コメントは誰でも読めるため。 +- **テスト名は日本語。** 何が壊れたのかを失敗行だけで判断できるようにする。 + +## 列定義を変えるとき + +`scripts/schema.py` が列の唯一の正。ここを直せば `build_checklist.py`・ +この単体テスト・非公開側の台帳検査・`README.md` の整合検査がまとめて追随する。 diff --git a/tools/api-inventory/tests/conftest.py b/tools/api-inventory/tests/conftest.py new file mode 100644 index 0000000000..2c8b18ebbb --- /dev/null +++ b/tools/api-inventory/tests/conftest.py @@ -0,0 +1,124 @@ +# -*- coding: utf-8 -*- +"""台帳ツールの単体テスト用の足場。 + +テストは**データを一切必要としない**。合成した最小のリポジトリと最小の台帳を +その場で組み立て、スクリプトの判定だけを確かめる。台帳そのものの検査は +プライベートリポジトリ側の `tests/` が受け持つ(公開リポジトリに台帳は置けない)。 +""" +import json +import os +import subprocess +import sys + +import pytest + +HERE = os.path.dirname(os.path.abspath(__file__)) +SCRIPTS = os.path.normpath(os.path.join(HERE, '..', 'scripts')) +sys.path.insert(0, SCRIPTS) + + +def run(script, *args, env=None, expect=None): + """スクリプトを別プロセスで回す。(returncode, stdout, stderr) を返す。 + + `build_checklist.py` のようにモジュール直下で処理を走らせるものがあるので、 + import ではなく実行で確かめる。 + """ + e = dict(os.environ) + e.pop('WEKO_API_INVENTORY_DIR', None) # 実データを踏まないようにする + e.update(env or {}) + p = subprocess.run([sys.executable, os.path.join(SCRIPTS, script), *map(str, args)], + capture_output=True, text=True, env=e) + if expect is not None: + assert p.returncode == expect, \ + f'{script} の終了コードが {p.returncode}(期待 {expect})\n' \ + f'--- stdout ---\n{p.stdout}\n--- stderr ---\n{p.stderr}' + return p + + +# --- 台帳(full)の合成 ----------------------------------------------------- + +from schema import FULL_COLUMNS + +# 台帳(詳細版)のヘッダ。定義は scripts/schema.py が持つ。 +FULL_HEADER = FULL_COLUMNS + + +def make_row(**over): + """既定値で埋めた1行を作る。変えたい列だけキーワードで渡す。""" + r = {n: '-' for n in FULL_HEADER} + r.update({ + 'no': '1', 'module': 'weko-demo', 'api_type': 'REST API', + 'app': 'UIアプリ', 'method': 'GET', 'uri': '/demo', + 'blueprint': 'demo', 'endpoint': 'demo.index', + 'impl_func': 'index', 'impl_file': 'modules/weko-demo/weko_demo/views.py', + 'impl_line': '10', 'auth_required': '要', 'auth_method': 'login_required', + 'data_op': '取得', 'dynamic_verified': '-', + }) + r.update(over) + return r + + +def write_full(path, rows): + with open(path, 'w', encoding='utf-8') as f: + f.write('\t'.join(FULL_HEADER) + '\n') + for r in rows: + f.write('\t'.join(r.get(n, '-') for n in FULL_HEADER) + '\n') + return str(path) + + +@pytest.fixture +def full_tsv(tmp_path): + """行を渡すと台帳(62列)を書き出して、そのパスを返す関数。""" + def _make(rows, name='weko3_api_list_full.tsv'): + return write_full(tmp_path / name, rows) + return _make + + +# --- スナップショット(実機 url_map)の合成 -------------------------------- + +def snap_entry(app, endpoint, rule, methods=('GET',), **extra): + d = {'app': app, 'endpoint': endpoint, + 'routes': [{'rule': rule, 'methods': list(methods)}], + 'provider': None, 'attrs': 'ast'} + d.update(extra) + return d + + +@pytest.fixture +def snapshot(tmp_path): + """endpoints を渡すとスナップショット JSON を書き出して、そのパスを返す関数。""" + def _make(endpoints, revision='deadbee', tag='v0.0.0', name='api_snapshot.json'): + p = tmp_path / name + json.dump({'meta': {'revision': revision, 'tag': tag}, 'endpoints': endpoints}, + open(p, 'w', encoding='utf-8'), ensure_ascii=False) + return str(p) + return _make + + +@pytest.fixture +def allow_json(tmp_path): + def _make(not_registered=None, not_a_route=None, name='reconcile_allow.json'): + p = tmp_path / name + json.dump({'not_registered': not_registered or {}, + 'not_a_route': not_a_route or []}, + open(p, 'w', encoding='utf-8'), ensure_ascii=False) + return str(p) + return _make + + +# --- 合成リポジトリ -------------------------------------------------------- + +@pytest.fixture +def fake_repo(tmp_path): + """`modules///` にソースを置く最小リポジトリを作る。""" + root = tmp_path / 'repo' + + def _write(relpath, text): + p = root / relpath + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text, encoding='utf-8') + return str(p) + + _write('modules/.keep', '') + _write.root = str(root) + return _write diff --git a/tools/api-inventory/tests/test_build_checklist.py b/tools/api-inventory/tests/test_build_checklist.py new file mode 100644 index 0000000000..d6d8fc83eb --- /dev/null +++ b/tools/api-inventory/tests/test_build_checklist.py @@ -0,0 +1,111 @@ +# -*- coding: utf-8 -*- +"""build_checklist.py — 62列の詳細版から32列のチェックリスト版を丸ごと生成する。 + +24列版は派生物で、手編集は次の生成で消える。ここで守るのは2点。 + + 1. **参照している列名が実在すること。** `g(c, "存在しない列")` は例外にならず + 空文字を返す。列をリネームすると、その列だけが黙って空になったチェックリストが + できあがる。 + 2. **統合の規則が変わっていないこと。** impl の組み立て、auth の連結、 + security_flags の拾い方は、列を読む側の awk 例と README の凡例に直結する。 +""" +import ast +import os +import re + +import pytest + +import schema +from conftest import SCRIPTS, FULL_HEADER, make_row, write_full, run + +SRC = open(os.path.join(SCRIPTS, 'build_checklist.py'), encoding='utf-8').read() + + +def test_出力列を自前で並べ直していない(): + """列定義は schema.py が唯一の正。ここで並べ直すと必ず台帳とずれる。""" + assert 'NEW = CHECKLIST_COLUMNS' in SRC + + +def _referenced_columns(): + """`g(c, "xxx")` で参照している列名を全部拾う。""" + return set(re.findall(r'g\(\s*c\s*,\s*"([^"]+)"\s*\)', SRC)) + + +def test_参照している列が全て台帳のヘッダに実在する(): + missing = sorted(_referenced_columns() - set(FULL_HEADER)) + assert not missing, ( + f'build_checklist.py が存在しない列を読んでいる: {missing}\n' + 'g() は存在しない列名でも例外にならず空文字を返すため、' + '該当列だけが黙って空のチェックリストが出来上がる。') + + +def test_出力列は32列で列名が重複しない(): + new = schema.CHECKLIST_COLUMNS + assert len(new) == 32 + assert len(set(new)) == len(new) + + +def test_派生列は末尾に置く(): + """既存列の位置を動かすと README の awk 例(`$20` など)が全て壊れる。""" + assert schema.CHECKLIST_COLUMNS[24:] == schema.DERIVED_COLUMNS + + +# --- 生成そのもの --------------------------------------------------------- + +def _build(tmp_path, rows): + src = write_full(tmp_path / 'full.tsv', rows) + dst = str(tmp_path / 'chk.tsv') + run('build_checklist.py', src, dst, expect=0) + out = [l.rstrip('\n').split('\t') for l in open(dst, encoding='utf-8')] + return out[0], out[1:] + + +def test_行数と列数が揃う(tmp_path): + hdr, rows = _build(tmp_path, [make_row(no='1'), make_row(no='2')]) + assert len(hdr) == 32 + assert len(rows) == 2 + assert all(len(r) == 32 for r in rows) + + +def test_implは関数とファイルと行を組み立てる(tmp_path): + hdr, [r] = _build(tmp_path, [make_row( + impl_func='show', impl_file='modules/weko-demo/views.py', impl_line='42')]) + assert r[hdr.index('impl')] == 'show @modules/weko-demo/views.py:42' + + +def test_impl_lineが0なら行番号を付けない(tmp_path): + """0 は「行が取れなかった」印。`views.py:0` と書くと実在の位置に見えてしまう。""" + hdr, [r] = _build(tmp_path, [make_row( + impl_func='show', impl_file='modules/weko-demo/views.py', impl_line='0')]) + assert r[hdr.index('impl')] == 'show @modules/weko-demo/views.py' + + +def test_authは要否と方式と仕組みを連結する(tmp_path): + hdr, [r] = _build(tmp_path, [make_row( + auth_required='要(管理)', auth_method='roles_required', + auth_mechanism='admin-role-table(WEKO_ADMIN_ACCESS_TABLE)')]) + assert r[hdr.index('auth')] == '要(管理) | roles_required | [admin-role-table]' + + +def test_security_flagsは該当する観点だけを集める(tmp_path): + hdr, [r] = _build(tmp_path, [make_row( + csrf_protection='なし(状態変更なのに未保護)', + input_validation='あり(スキーマ検証)', + bola_risk='★所有者チェックなし')]) + flags = r[hdr.index('security_flags')] + assert 'CSRF:' in flags and 'BOLA:' in flags + assert 'INPUT:' not in flags # 「あり」は指摘ではない + + +def test_空欄と不明はハイフンに寄せる(tmp_path): + """読む側が「空欄」「-」「不明」を区別しなくて済むようにする。""" + hdr, [r] = _build(tmp_path, [make_row(summary='', roles='不明')]) + assert r[hdr.index('summary')] == '-' + assert r[hdr.index('roles_scope')] == '-' + + +def test_タブと改行はセルに残さない(tmp_path): + """1行1レコードの TSV が壊れると、以降の全行の列がずれる。""" + hdr, rows = _build(tmp_path, [make_row(summary='一行目\t二行目')]) + assert len(rows) == 1 and len(rows[0]) == 32 + assert '\t' not in rows[0][hdr.index('summary')] diff --git a/tools/api-inventory/tests/test_detect_routes.py b/tools/api-inventory/tests/test_detect_routes.py new file mode 100644 index 0000000000..77d34cf204 --- /dev/null +++ b/tools/api-inventory/tests/test_detect_routes.py @@ -0,0 +1,252 @@ +# -*- coding: utf-8 -*- +"""detect_routes.py — ソースだけから経路を検知する。 + +実機 url_map は「今この環境で登録されている経路」しか映さない。config で無効な経路、 +プラグイン未導入の経路、設定値が真のときだけ登録される経路は、実機からは見えないのに +API としては存在する。この検知器はそこを埋めるためのもので、**検知源が1つ黙って +死んでも件数が減るだけで気付けない**。だから検知源ごとに「拾えること」を固定する。 +""" +import os + +import pytest + +import detect_routes as dr +from conftest import make_row, write_full + + +# -------------------------------------------------------------------------- +# 検知源ごとの回帰 +# -------------------------------------------------------------------------- + +def _detect(fake_repo, relpath, src): + fake_repo(relpath, src) + return dr.detect(fake_repo.root) + + +def _sources(found, kind): + return [d for d in found if d['source'] == kind] + + +def test_route_デコレータを拾う(fake_repo): + found = _detect(fake_repo, 'modules/weko-demo/weko_demo/views.py', ''' +from flask import Blueprint +bp = Blueprint('demo', __name__) + +@bp.route('/demo/', methods=['GET', 'POST']) +def show(pk): + return '' +''') + [d] = _sources(found, 'route') + assert d['rule'] == '/demo/' + assert d['methods'] == ['GET', 'POST'] + assert d['func'] == 'show' + + +def test_expose_を拾う(fake_repo): + """Flask-Admin の `@expose`。従来の AST 抽出は route しか見ておらず、 + 205件の管理画面ビューがまるごと静的検知から漏れていた。""" + found = _detect(fake_repo, 'modules/weko-demo/weko_demo/admin.py', ''' +from flask_admin import BaseView, expose + +class SettingView(BaseView): + @expose('/', methods=['GET']) + def index(self): + return '' + + @expose('/save', methods=['POST']) + def save(self): + return '' +''') + got = {(d['qual'], d['rule']) for d in _sources(found, 'expose')} + assert got == {('SettingView.index', '/'), ('SettingView.save', '/save')} + + +def test_add_url_rule_のas_viewを解決する(fake_repo): + """config 駆動の登録は `view_func = X.as_view(...)` を挟む。変数名だけ見ると + どのクラスの経路か分からなくなる(実測: 70件が照合不能になった)。""" + found = _detect(fake_repo, 'modules/weko-demo/weko_demo/rest.py', ''' +def create_blueprint(endpoints): + for endpoint, options in endpoints.items(): + view_func = DemoResource.as_view(DemoResource.view_name) + blueprint.add_url_rule(options.get('route'), view_func=view_func, + methods=['POST']) + return blueprint +''') + [d] = _sources(found, 'add_url_rule') + assert d['cls'] == 'DemoResource' + assert d['qual'] == 'DemoResource' + assert d['rule_expr'] == "options.get('route')" + assert d['methods'] == ['POST'] + + +def test_add_url_rule_の一括登録は門番から外れる(fake_repo): + """`add_url_rule(**rule)` は rule も view_func も静的に取れない。 + 個々の経路は rest_config 側で拾うので、ここで落とすと常時赤になる。""" + found = _detect(fake_repo, 'modules/weko-demo/weko_demo/rest.py', ''' +def create_blueprint(endpoints): + for endpoint, options in endpoints.items(): + for rule in build(endpoint, **options): + blueprint.add_url_rule(**rule) +''') + [d] = _sources(found, 'add_url_rule') + assert d['dispatch'] is True + + +def test_rest_config_の経路定義を拾う(fake_repo): + found = _detect(fake_repo, 'modules/weko-demo/weko_demo/config.py', ''' +DEMO_REST_ENDPOINTS = { + 'demo': { + 'list_route': '/demo/items', + 'item_route': '/demo/items/', + 'record_class': 'weko_demo.api:Demo', + }, +} +''') + got = {d['rule'] for d in _sources(found, 'rest_config')} + assert got == {'/demo/items', '/demo/items/'} + + +def test_modelview_のクラスを拾う(fake_repo): + found = _detect(fake_repo, 'modules/weko-demo/weko_demo/admin.py', ''' +from flask_admin.contrib.sqla import ModelView + +class WidgetView(ModelView): + can_delete = True +''') + assert [d['cls'] for d in _sources(found, 'modelview')] == ['WidgetView'] + + +def test_entry_point_は経路を生む群だけを見る(fake_repo): + """`invenio_base.apps` は拡張の登録で、それ自体は経路を作らない。 + 混ぜると恒常的な偽陽性になってゲートが形骸化する。""" + fake_repo('modules/weko-demo/setup.py', ''' +setup(entry_points={ + 'invenio_base.blueprints': ['weko_demo = weko_demo.views:blueprint'], + 'invenio_base.apps': ['weko_demo_ext = weko_demo:WekoDemo'], + 'invenio_admin.views': ['weko_demo_widget = weko_demo.admin:widget_adminview'], +}) +''') + found = dr.detect(fake_repo.root) + got = {d['qual'] for d in _sources(found, 'entry_point')} + assert got == {'weko_demo', 'weko_demo_widget'} + + +def test_adminview辞書を経由してビュークラスまで辿る(fake_repo): + """entry point は `module:xxx_adminview` を指すだけ。辞書の中身まで辿らないと + Flask-Admin の登録名(台帳の endpoint の `.` の手前)が分からない。""" + fake_repo('modules/weko-demo/weko_demo/admin.py', ''' +class SessionActivityView(ModelView): + pass + +session_adminview = {'model': SessionActivity, 'modelview': SessionActivityView} +''') + fake_repo('modules/weko-demo/setup.py', ''' +setup(entry_points={ + 'invenio_admin.views': ['demo_session = weko_demo.admin:session_adminview'], +}) +''') + found = dr.detect(fake_repo.root) + [ep] = _sources(found, 'entry_point') + assert 'SessionActivityView' in ep.get('via', []) + assert 'sessionactivity' in dr.admin_prefixes(ep) + + +def test_テストコードは検知対象から外す(fake_repo): + fake_repo('modules/weko-demo/tests/test_views.py', ''' +@bp.route('/only-in-tests') +def x(): + return '' +''') + assert dr.detect(fake_repo.root) == [] + + +# -------------------------------------------------------------------------- +# 台帳との突き合わせ +# -------------------------------------------------------------------------- + +@pytest.mark.parametrize('a,b', [ + ('/api/demo', '/demo'), ('/demo/', '/demo'), ('/demo', '/api/demo')]) +def test_uri比較はapi前置と末尾スラッシュを吸収する(a, b): + assert dr.uri_variants(a) & dr.uri_variants(b) + + +def test_実装一致で照合する(tmp_path, fake_repo): + src = 'modules/weko-demo/weko_demo/views.py' + fake_repo(src, "@bp.route('/demo')\ndef show():\n return ''\n") + tsv = write_full(tmp_path / 'full.tsv', + [make_row(impl_file=src, impl_func='show', uri='/other')]) + led = dr.load_ledger(tsv) + [d] = dr.detect(fake_repo.root) + assert dr.match(d, led)[0] == 'impl' + + +def test_URIだけが一致する場合も照合する(tmp_path, fake_repo): + """委譲やラッパで impl_func 名が変わることがある。URI でも当てられること。""" + fake_repo('modules/weko-demo/weko_demo/views.py', + "@bp.route('/demo')\ndef show():\n return ''\n") + tsv = write_full(tmp_path / 'full.tsv', + [make_row(impl_file='modules/other/x.py', + impl_func='wrapper', uri='/api/demo')]) + led = dr.load_ledger(tsv) + [d] = dr.detect(fake_repo.root) + assert dr.match(d, led)[0] == 'uri' + + +def test_台帳に無い経路は照合できない(tmp_path, fake_repo): + fake_repo('modules/weko-demo/weko_demo/views.py', + "@bp.route('/undocumented')\ndef leak():\n return ''\n") + tsv = write_full(tmp_path / 'full.tsv', [make_row(uri='/demo')]) + led = dr.load_ledger(tsv) + [d] = dr.detect(fake_repo.root) + assert dr.match(d, led)[1] == [] + + +def test_許可リストのキーは行番号に依存しない(tmp_path, fake_repo): + """行がずれるたびに許可リストを書き直すことになると、いずれ運用されなくなる。""" + src = 'modules/weko-demo/weko_demo/views.py' + fake_repo(src, "@bp.route('/x')\ndef f():\n return ''\n") + a = dr.detect(fake_repo.root)[0] + fake_repo(src, "\n\n\n@bp.route('/x')\ndef f():\n return ''\n") + b = dr.detect(fake_repo.root)[0] + assert a['line'] != b['line'] + assert dr.allow_key(a) == dr.allow_key(b) + + +# -------------------------------------------------------------------------- +# ゲートと出力 +# -------------------------------------------------------------------------- + +def _cross(fake_repo, tsv, *extra, expect=None, allow=None): + from conftest import run + env = {'WEKO_API_INVENTORY_DIR': os.path.dirname(tsv)} if allow else {} + return run('detect_routes.py', '--weko-root', fake_repo.root, '--tsv', tsv, + '--cross-check', *extra, env=env, expect=expect) + + +def test_未収載があればゲートで落ちる(tmp_path, fake_repo): + fake_repo('modules/weko-demo/weko_demo/views.py', + "@bp.route('/undocumented')\ndef leak():\n return ''\n") + tsv = write_full(tmp_path / 'full.tsv', [make_row(uri='/demo')]) + p = _cross(fake_repo, tsv, '--gate', expect=1) + assert '/undocumented' in p.stdout + + +def test_許可リストに理由を書けばゲートを通る(tmp_path, fake_repo): + src = 'modules/weko-demo/weko_demo/views.py' + fake_repo(src, "@bp.route('/undocumented')\ndef leak():\n return ''\n") + tsv = write_full(tmp_path / 'full.tsv', [make_row(uri='/demo')]) + import json + json.dump({f'{src}::leak': 'このサイトでは config で無効'}, + open(tmp_path / 'detect_allow.json', 'w', encoding='utf-8'), + ensure_ascii=False) + p = _cross(fake_repo, tsv, '--gate', allow=True, expect=0) + assert 'このサイトでは config で無効' in p.stdout + + +def test_summary_onlyは経路名を出さない(tmp_path, fake_repo): + fake_repo('modules/weko-demo/weko_demo/views.py', + "@bp.route('/secret/leak/me')\ndef leak():\n return ''\n") + tsv = write_full(tmp_path / 'full.tsv', [make_row(uri='/demo')]) + p = _cross(fake_repo, tsv, '--summary-only') + assert '/secret/leak/me' not in p.stdout + assert 'leak' not in p.stdout diff --git a/tools/api-inventory/tests/test_docs.py b/tools/api-inventory/tests/test_docs.py new file mode 100644 index 0000000000..19359fcc56 --- /dev/null +++ b/tools/api-inventory/tests/test_docs.py @@ -0,0 +1,147 @@ +# -*- coding: utf-8 -*- +"""手順書(scripts/README.md)が実装とずれていないかを検査する。 + +手順書は**壊れても誰も落ちない**ので、いちばん静かに腐る。実際 v2.0.4 時点で +README は「57列 / 24列 / 926行 / `NF!=65`」と書いたまま、実体は +「62列 / 32列 / 1048行」になっていた。列数の検算例が間違っていると、 +検算をすり抜けた壊れた行がそのまま台帳に入る。 + +ここで見るのは3点。 + + 1. 手順に出てくるスクリプトが実在すること + 2. 列数・列名の記述が `schema.py` と一致すること + 3. 手順に書かれた実行順が、スクリプトの依存関係と矛盾しないこと +""" +import os +import re + +import pytest + +import schema +from conftest import SCRIPTS + +DOC = os.path.join(SCRIPTS, 'README.md') +TEXT = open(DOC, encoding='utf-8').read() + +# 台帳の列名として出てくるが、実際には24列版・中間生成物の名前であるもの。 +# schema.FULL_COLUMNS に無くても誤りではない。 +NOT_FULL_COLUMNS = set(schema.CHECKLIST_COLUMNS) | { + 'impl', 'auth', 'roles_scope', 'last_change', 'tags', 'security_finding', + 'security_flags', +} + +# 列名に見えるが列ではない語。entry point 群の名前など。 +NOT_A_COLUMN = {'api_apps', 'api_blueprints', 'data_dir', 'api_route', + 'api_route_item', 'access_token', 'refresh_token', 'api_key'} + + +def test_手順に出てくるスクリプトが実在する(): + """`python3 .../xxx.py` の形で案内しているものだけを見る + (解析対象側の views.py / admin.py などの言及と混ぜない)。""" + named = set(re.findall(r'python3\s+(?:[^\s$]*/)?([a-z_0-9]+\.py)', TEXT)) + named |= set(re.findall(r'`([a-z_0-9]+\.py)`', TEXT)) & set(os.listdir(SCRIPTS)) + missing = sorted(n for n in named if not os.path.isfile(os.path.join(SCRIPTS, n))) + assert not missing, f'README が存在しないスクリプトを案内している: {missing}' + + +def test_全スクリプトが手順書のどこかで説明されている(): + """入口が README しかない。載っていないスクリプトは、いずれ誰も回さなくなる。""" + files = {f for f in os.listdir(SCRIPTS) + if f.endswith('.py') and not f.startswith('_')} + files -= {'schema.py', 'paths.py'} # 他から読まれるだけの土台 + undocumented = sorted(f for f in files if f not in TEXT) + assert not undocumented, f'README に説明が無いスクリプト: {undocumented}' + + +# 台帳そのものの列数を名指ししている書き方。ここが古びると検算が意味を失う。 +FULL_CLAIM = re.compile(r'(?:weko3_api_list_full\.tsv`?\(|詳細版\()(\d+)列') +CHECKLIST_CLAIM = re.compile( + r'(?:weko3_api_list\.tsv`?\(|チェックリスト版\()(\d+)列|(\d+)列版') + + +def test_台帳の列数の記述がschemaと一致する(): + """README は台帳の列数を何度も書く。1か所でも古いと検算がすり抜ける + (実測: 「57列 / 24列」と書いたまま実体は 62列 / 32列 になっていた)。""" + full = {int(x) for x in FULL_CLAIM.findall(TEXT)} + chk = {int(x or y) for x, y in CHECKLIST_CLAIM.findall(TEXT)} + assert full, 'README から詳細版の列数の記述が消えている' + assert chk, 'README からチェックリスト版の列数の記述が消えている' + assert full == {len(schema.FULL_COLUMNS)}, \ + f'詳細版の列数 {sorted(full)} が実際の {len(schema.FULL_COLUMNS)} と合わない' + assert chk <= {len(schema.CHECKLIST_COLUMNS), len(schema.FULL_COLUMNS)}, \ + f'チェックリスト版の列数 {sorted(chk)} が実際の {len(schema.CHECKLIST_COLUMNS)} と合わない' + + +def test_列数の検算例が正しい列数を使っている(): + """`awk NF!=N` は行追加のたびに回す検算。N がずれると常に無言で通る。""" + got = re.findall(r'NF\s*!=\s*(\d+)', TEXT) + assert got, 'README から列数の検算例が消えている' + assert set(got) == {str(len(schema.FULL_COLUMNS))}, \ + f'検算例の列数 {set(got)} が実際の {len(schema.FULL_COLUMNS)} と合わない' + + +def test_README_が触れる列名が実在する(): + """列を統合・改名したのに README が旧名で残ると、その手順は実行できない + (実測: `auth_response_variance` / `data_target` / `data_op_detail` が該当した)。""" + named = _column_like() - NOT_FULL_COLUMNS - NOT_A_COLUMN + missing = sorted(n for n in named if n not in schema.FULL_COLUMNS) + assert not missing, f'README が実在しない列名を使っている: {missing}' + + +def _column_like(): + """列名らしい語だけに絞る。関数名や設定キーを巻き込まないための当たり表。""" + prefixes = ('sec_', 'test_', 'auth_', 'data_', 'impl_', 'last_commit', + 'input_', 'audit_', 'csrf_', 'ssrf_', 'redirect_', 'resource_', + 'triggers_', 'bola_', 'api_', 'path_', 'query_', 'body_', + 'request_', 'response_', 'oauth_', 'cache_', 'config_', + 'category_', 'release_', 'priority', 'dynamic_', 'access_', + 'restricted_', 'idempotency', 'deprecated', 'side_effects') + return {w for w in re.findall(r'`([a-z][a-z_0-9]{3,})`', TEXT) + if w.startswith(prefixes)} + + +def test_派生列が手編集禁止として説明されている(): + """「手編集しても消える列」の説明。抜けがあると、消える列を人が直し続ける。 + 連番の列は `test_normal`〜`test_gap` のような範囲表記でもよい。""" + for col in schema.DERIVED_COLUMNS: + assert col in TEXT or f'`{schema.DERIVED_COLUMNS[2]}`〜`{schema.DERIVED_COLUMNS[-2]}`' in TEXT, \ + f'派生列 {col} が README で説明されていない' + assert '手編集しない' in TEXT or '直接編集しない' in TEXT + + +def _order_in_procedure(*names): + """『ケース1』のコードブロックに現れる順を返す。""" + start = TEXT.index('## ケース1:') + block = TEXT[start:TEXT.index('## ケース1b')] + return [block.index(n) for n in names] + + +def test_ケース1の実行順が依存関係どおり(): + """`prioritize.py` は `test_gap` を読むので `test_coverage.py` が先。 + 逆順に書かれていると、1回目の実行で優先度が1世代古い値になる。""" + tc, pr, bc = _order_in_procedure( + 'test_coverage.py', 'prioritize.py', 'build_checklist.py') + assert tc < pr < bc, \ + 'ケース1 の実行順が test_coverage → prioritize → build_checklist になっていない' + + +def test_静的検知の手順が案内されている(): + """実機 url_map だけでは、config で無効な経路の漏れを検出できない。""" + assert 'detect_routes.py' in TEXT + assert 'detect_allow.json' in TEXT + + +def test_実装を触ったときの順序が明記されている(): + """`enrich_git.py` は `impl_line` の指す関数のコミットを引く。 + `refresh_impl.py` を先に回さないと手前の関数のコミットを拾う。""" + ri = TEXT.index('refresh_impl.py') + eg = TEXT.index('enrich_git.py') + assert ri < eg + assert '★順序が重要' in TEXT or '必ず `refresh_impl.py` が先' in TEXT + + +def test_公開してはいけないものの注意が残っている(): + """このリポジトリは public。台帳を置かない前提が消えたら手順ごと危険になる。""" + assert 'public' in TEXT + assert 'WEKO_API_INVENTORY_DIR' in TEXT + assert '--summary-only' in TEXT diff --git a/tools/api-inventory/tests/test_merge.py b/tools/api-inventory/tests/test_merge.py new file mode 100644 index 0000000000..09d9798ec5 --- /dev/null +++ b/tools/api-inventory/tests/test_merge.py @@ -0,0 +1,66 @@ +# -*- coding: utf-8 -*- +"""merge.py — Phase 1 の out/*.tsv を1本にまとめて採番する。 + +台帳の初回生成でしか使わないが、ここが崩れると以降の全 Phase の入力が崩れる。 +""" +import os + +import merge +from conftest import run + + +def _merge(tmp_path, files): + outdir = tmp_path / 'out' + outdir.mkdir() + for name, lines in files.items(): + (outdir / name).write_text('\n'.join(lines) + '\n', encoding='utf-8') + dst = tmp_path / 'merged.tsv' + run('merge.py', str(outdir), str(dst), expect=0) + rows = [l.rstrip('\n').split('\t') for l in open(dst, encoding='utf-8')] + return rows[0], rows[1:] + + +def row(uri, method='GET', file='a.py', line='1', module='m'): + c = [''] * merge.NCOL + c[1], c[4], c[5], c[13], c[14] = module, method, uri, file, line + return '\t'.join(c) + + +def test_ヘッダは定義どおりの列数(tmp_path): + hdr, _ = _merge(tmp_path, {'a.tsv': [row('/a')]}) + assert hdr == merge.HEADER + assert len(hdr) == merge.NCOL + + +def test_連番を振り直す(tmp_path): + _, rows = _merge(tmp_path, {'a.tsv': [row('/b'), row('/a')]}) + assert [r[0] for r in rows] == ['1', '2'] + + +def test_同じ経路の重複を落とす(tmp_path): + """uri+method+file+line が同じなら同一行。Phase 1 は複数の抽出を合流させる。""" + _, rows = _merge(tmp_path, {'a.tsv': [row('/a')], 'b.tsv': [row('/a')]}) + assert len(rows) == 1 + + +def test_列が足りない行は埋める(tmp_path): + _, [r] = _merge(tmp_path, {'a.tsv': ['x\ty\tz']}) + assert len(r) == merge.NCOL + + +def test_列が多い行は末尾にまとめる(tmp_path): + """切り捨てると備考が黙って消える。最終列に連結して残す。""" + _, [r] = _merge(tmp_path, {'a.tsv': ['\t'.join(['v'] * (merge.NCOL + 2))]}) + assert len(r) == merge.NCOL + assert ' | ' in r[-1] + + +def test_誤って混ざったヘッダ行を落とす(tmp_path): + _, rows = _merge(tmp_path, {'a.tsv': ['\t'.join(merge.HEADER), row('/a')]}) + assert len(rows) == 1 + + +def test_セルの前後の空白を落とす(tmp_path): + """抽出元によって空白の付き方が違う。突き合わせは文字列一致なので揃える。""" + _, [r] = _merge(tmp_path, {'a.tsv': [row(' /a ')]}) + assert r[5] == '/a' diff --git a/tools/api-inventory/tests/test_prioritize.py b/tools/api-inventory/tests/test_prioritize.py new file mode 100644 index 0000000000..cbf986218c --- /dev/null +++ b/tools/api-inventory/tests/test_prioritize.py @@ -0,0 +1,198 @@ +# -*- coding: utf-8 -*- +"""prioritize.py — 台帳に対応優先度を付ける。 + +優先度は「どの行から手を付けるか」を決める唯一の指標なので、判定が静かに変わると +**見るべき行が埋もれる**。ルールの分岐そのものを固定する。 + +判定の入力は台帳の本体列(security_finding / dynamic_verified / data_op / deprecated 等)。 +派生列は毎回上書きされるので、ここを直しても意味がない。 +""" +import pytest + +import prioritize +from conftest import FULL_HEADER, make_row, write_full + +H = {n: i for i, n in enumerate(FULL_HEADER)} + + +def cls(**over): + """1行を作って classify に掛け、(優先度, 理由) を返す。""" + r = make_row(**over) + return prioritize.classify([r[n] for n in FULL_HEADER], H) + + +def dec(allow=(frozenset(), frozenset()), **over): + r = make_row(**over) + return prioritize.decide([r[n] for n in FULL_HEADER], H, allow) + + +# --- P1: データ破壊と、認可の無い状態変更 -------------------------------- + +def test_無認証で既存ファイル実体を壊せる行はP1(): + p, why = cls(method='POST', auth_required='不要', data_op='更新', + data_store='ファイル実体(FileInstance)') + assert p == 'P1' and 'ファイル実体' in why + + +def test_認証の無い状態変更系はP1(): + p, why = cls(method='POST', auth_required='不要', data_op='更新') + assert p == 'P1' and '認証チェックが無い' in why + + +def test_権限チェックが機能していない状態変更系はP1(): + p, why = cls(method='DELETE', sec_pattern='ロールチェックが実効せず', + data_op='物理削除') + assert p == 'P1' and '機能していない' in why + + +def test_未認証で到達したという実測はP1に上げる(): + """静的には login_required が付いていても、実測で通っていれば実態が正。""" + p, _ = cls(method='POST', auth_required='要', data_op='更新', + dynamic_verified='[実測] 未認証で到達') + assert p == 'P1' + + +# --- P2: 壊さない、または限定が足りない ----------------------------------- + +def test_新規作成しかしない無認証の書き込みはP2(): + """既存データを壊さない。P1(データ破壊)と同列には置かない。""" + p, why = cls(method='POST', auth_required='不要', data_op='作成') + assert p == 'P2' and '新規作成のみ' in why + + +def test_ログインのみで所有者限定が無い状態変更系はP2(): + p, why = cls(method='PUT', auth_required='要', data_op='更新', + dynamic_verified='[実測] ログインのみで到達') + assert p == 'P2' and 'IDOR' in why + + +def test_到達可否が未測定の状態変更系はP2(): + """「分からない」を安全側に倒さない。測っていない書き込みは確認対象。""" + p, why = cls(method='POST', auth_required='要', data_op='更新', + dynamic_verified='-') + assert p == 'P2' and '未測定' in why + + +def test_参照系でも露出が認証情報で認可が緩ければP2(): + p, why = cls(method='GET', auth_required='不要', data_op='取得', + sec_pattern='認証不要で参照可', sec_exposed='client_secret') + assert p == 'P2' and '認証情報' in why + + +def test_露出の記述だけでは引き上げない(): + """指摘も実証も無い行を露出語だけで上げると、適切に絞られている行まで赤くなる。""" + p, _ = cls(method='GET', auth_required='要', data_op='取得', + access_variance='非公開アイテムは除外される') + assert p != 'P2' + + +# --- P3 / P4 / P5 / 対象外 ------------------------------------------------ + +def test_認証の無い読み取り系はP3(): + p, why = cls(method='GET', auth_required='不要', data_op='取得') + assert p == 'P3' and '読み取り系' in why + + +@pytest.mark.parametrize('uri,label', [ + ('/ping', 'ヘルスチェック'), ('/robots.txt', 'robots.txt'), + ('/api/oai', 'OAI-PMH'), ('/static/x.js', '静的ファイル配信')]) +def test_意図的な公開設計はP4(uri, label): + p, why = cls(method='GET', uri=uri, auth_required='不要', data_op='取得') + assert p == 'P4' and label in why + + +def test_具体的な権限チェック機構があればP5(): + p, why = cls(method='GET', auth_required='要', + auth_method='need_record_permission', data_op='取得') + assert p == 'P5' and 'need_record_permission' in why + + +def test_admin保護され指摘も実証も無ければ対象外(): + p, _ = cls(method='GET', auth_required='要(管理)', + auth_method='roles_required', data_op='取得') + assert p == '対象外' + + +def test_指摘がある行は対象外にしない(): + """保護されているように見えて破綻している行を除外してしまうため。""" + p, _ = cls(method='GET', auth_required='要(管理)', + auth_method='roles_required', data_op='取得', + sec_pattern='管理画面だが権限表に載っていない') + assert p != '対象外' + + +# --- テスト観点による引き上げ --------------------------------------------- + +def test_テスト観点が全く確認できない行はP3まで上げる(): + p, why = cls(method='GET', auth_required='要(管理)', + auth_method='roles_required', data_op='取得', + test_gap='正常値,異常値,境界値,例外処理') + assert p == 'P3' and '4観点' in why + + +def test_テスト関数を特定できない行もP3まで上げる(): + p, why = cls(method='GET', auth_required='要(管理)', + auth_method='roles_required', data_op='取得', + test_gap='特定不能') + assert p == 'P3' and '特定できず' in why + + +def test_観点が一部欠けるだけなら優先度は変えず理由に残す(): + p, why = cls(method='GET', auth_required='要(管理)', + auth_method='roles_required', data_op='取得', + test_gap='例外処理') + assert p == '対象外' and '例外処理' in why + + +# --- 非利用・環境依存の重ね合わせ ------------------------------------------ + +def test_非利用で認可も軽ければ整理対象(): + p, why, cleanup = dec(method='GET', auth_required='不要', data_op='取得', + deprecated='未使用(呼出元なし)') + assert p == '整理対象' and cleanup == '未使用(呼出元なし)' + + +def test_非利用でもP1は優先度を落とさない(): + """消せば済むが、消すまでは穴が空いたまま。優先度を下げると見落とす。""" + p, why, _ = dec(method='POST', auth_required='不要', data_op='更新', + deprecated='未使用(呼出元なし)') + assert p == 'P1' and '削除が最短' in why + + +def test_実機に無い行は環境依存にするが削除候補にしない(): + """別の設定・別サイトでは有効になる。台帳からは消さない。""" + p, why, cleanup = dec(allow=(frozenset({'/demo'}), frozenset()), + method='GET', uri='/demo', + auth_required='不要', data_op='取得') + assert p == '環境依存' and cleanup == '-' + assert '認可上の判定は P3' in why + + +def test_実測の履歴を現在値として読まない(): + """apply_probe_results.py --keep-history が旧測定を同じセルに残す。 + 旧測定の『未認証で到達』を今の値として読むと、直した行が赤いままになる。""" + now = '[実測·2026-09-02] 管理者で到達' + old = '[実測·2026-08-26] 未認証で到達' + p, _ = cls(method='POST', auth_required='要', data_op='更新', + dynamic_verified=f'{now}{prioritize.HISTORY_SEP}{old}') + assert p != 'P1' + + +# --- 列順の正規化 ---------------------------------------------------------- + +def test_派生列は実行順に依存しない位置に揃える(tmp_path): + """test_coverage.py は test_* を末尾に付け直す。prioritize.py が並びを + 正規化しないと、同じ内容でも実行順で列順が変わって差分が出続ける。""" + p = write_full(tmp_path / 'full.tsv', [make_row()]) + prioritize.apply_to(p) + hdr = open(p, encoding='utf-8').readline().rstrip('\n').split('\t') + assert hdr[-8:] == prioritize.TAIL + assert hdr == FULL_HEADER + + +def test_二度流しても結果が変わらない(tmp_path): + p = write_full(tmp_path / 'full.tsv', [make_row(no='1'), make_row(no='2')]) + prioritize.apply_to(p) + once = open(p, encoding='utf-8').read() + prioritize.apply_to(p) + assert open(p, encoding='utf-8').read() == once diff --git a/tools/api-inventory/tests/test_reconcile.py b/tools/api-inventory/tests/test_reconcile.py new file mode 100644 index 0000000000..94874437ba --- /dev/null +++ b/tools/api-inventory/tests/test_reconcile.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +"""reconcile.py — 実機 url_map と台帳の突き合わせ。 + +**検出器そのものの回帰テスト**。ここが黙って壊れると、台帳から経路が漏れていても +ゲートは緑のまま通る。A〜E の各検出が「本当に鳴る」ことを毎回確かめる。 +""" +import json + +import pytest + +import reconcile +from conftest import make_row, snap_entry, run + + +def _run(snapshot, tsv, allow, *extra, expect=None): + return run('reconcile.py', '--snapshot', snapshot, '--tsv', tsv, + '--allow', allow, *extra, expect=expect) + + +# --- 一致する状態が本当に緑になるか -------------------------------------- + +def test_一致していればゲートを通る(full_tsv, snapshot, allow_json): + tsv = full_tsv([make_row(uri='/demo', method='GET', endpoint='demo.index')]) + snap = snapshot({'ui:demo.index': snap_entry('ui', 'demo.index', '/demo')}) + p = _run(snap, tsv, allow_json(), '--gate', expect=0) + assert '✅ 一致' in p.stdout + + +# --- A: 台帳の抽出漏れ ---------------------------------------------------- + +def test_A_実機にあって台帳に無い経路を検出する(full_tsv, snapshot, allow_json): + tsv = full_tsv([make_row(uri='/demo', endpoint='demo.index')]) + snap = snapshot({ + 'ui:demo.index': snap_entry('ui', 'demo.index', '/demo'), + 'ui:demo.hidden': snap_entry('ui', 'demo.hidden', '/demo/hidden'), + }) + p = _run(snap, tsv, allow_json(), '--gate', expect=1) + assert '/demo/hidden' in p.stdout + assert 'A. インベントリ未収載(抽出漏れ) | 1' in p.stdout + + +# --- B: 台帳にあって実機に無い ------------------------------------------ + +def test_B_実機に無い行を検出し許可リストで既知にできる(full_tsv, snapshot, allow_json): + rows = [make_row(uri='/demo', endpoint='demo.index'), + make_row(no='2', uri='/gone', endpoint='demo.gone')] + tsv = full_tsv(rows) + snap = snapshot({'ui:demo.index': snap_entry('ui', 'demo.index', '/demo')}) + + p = _run(snap, tsv, allow_json(), '--gate', expect=1) + assert "B. 実機に無い(未説明) | 1" in p.stdout + + # 理由を書いて許可リストに載せれば既知(B')に移り、ゲートは通る。 + # URI を許可すると、その行の endpoint も E' 側で黙認される。 + ok = allow_json(not_registered={'/gone': 'config で無効'}) + p = _run(snap, tsv, ok, '--gate', expect=0) + assert "B'. 実機に無い(既知・許容) | 1" in p.stdout + assert 'B. 実機に無い(未説明) | 0' in p.stdout + assert 'config で無効' in p.stdout # 理由が必ず出力に残る + + +# --- C: メソッド不一致 ---------------------------------------------------- + +def test_C_メソッドの食い違いを検出する(full_tsv, snapshot, allow_json): + tsv = full_tsv([make_row(uri='/demo', method='GET', endpoint='demo.index')]) + snap = snapshot({'ui:demo.index': + snap_entry('ui', 'demo.index', '/demo', ('GET', 'POST'))}) + p = _run(snap, tsv, allow_json(), '--gate', expect=1) + assert 'C. メソッド不一致 | 1' in p.stdout + + +def test_C_HEADとOPTIONSは差分に数えない(full_tsv, snapshot, allow_json): + """werkzeug が GET に自動付与するだけなので、比較対象から外れていること。""" + tsv = full_tsv([make_row(uri='/demo', method='GET,HEAD,OPTIONS', + endpoint='demo.index')]) + snap = snapshot({'ui:demo.index': snap_entry('ui', 'demo.index', '/demo')}) + _run(snap, tsv, allow_json(), '--gate', expect=0) + + +# --- D: app 列の不一致 ---------------------------------------------------- + +def test_D_登録先アプリの記載誤りを検出する(full_tsv, snapshot, allow_json): + tsv = full_tsv([make_row(uri='/api/demo', app='UIアプリ', endpoint='demo.index')]) + snap = snapshot({'api:demo.index': snap_entry('api', 'demo.index', '/demo')}) + p = _run(snap, tsv, allow_json(), '--gate', expect=1) + assert 'D. app列の不一致 | 1' in p.stdout + + +# --- E: endpoint 単位の取りこぼし ---------------------------------------- + +def test_E_同一URIに複数endpointがある取りこぼしを検出する(full_tsv, snapshot, + allow_json): + """URI 単位の A では拾えない。台帳は endpoint 単位で行を持つ方針。""" + tsv = full_tsv([make_row(uri='/demo', endpoint='demo.index')]) + snap = snapshot({ + 'ui:demo.index': snap_entry('ui', 'demo.index', '/demo'), + 'ui:other.index': snap_entry('ui', 'other.index', '/demo'), + }) + p = _run(snap, tsv, allow_json(), '--gate', expect=1) + assert 'A. インベントリ未収載(抽出漏れ) | 0' in p.stdout # URI は一致している + assert 'E. endpoint 未収載 | 1' in p.stdout + + +# --- 出力の秘匿 ----------------------------------------------------------- + +def test_summary_onlyは経路名を出さない(full_tsv, snapshot, allow_json): + """public な CI ログ・artifact・PR コメントは誰でも読める。""" + tsv = full_tsv([make_row(uri='/demo', endpoint='demo.index')]) + snap = snapshot({ + 'ui:demo.index': snap_entry('ui', 'demo.index', '/demo'), + 'ui:demo.secret': snap_entry('ui', 'demo.secret', '/secret/leak/me'), + }) + p = _run(snap, tsv, allow_json(), '--summary-only') + assert '/secret/leak/me' not in p.stdout + assert 'demo.secret' not in p.stdout + assert 'A. インベントリ未収載(抽出漏れ) | 1' in p.stdout + + +# --- 正規化規則 ----------------------------------------------------------- + +@pytest.mark.parametrize('a,b', [('/demo/', '/demo'), ('/demo', '/demo'), ('/', '/')]) +def test_末尾スラッシュは同一視する(a, b): + assert reconcile.norm(a) == reconcile.norm(b) + + +def test_APIアプリの経路にはapiが前置される(snapshot): + """API アプリは DispatcherMiddleware で /api にマウントされ、url_map 側には出ない。""" + snap = snapshot({'api:x': snap_entry('api', 'x', '/records')}) + _, S = reconcile.load_snapshot(snap) + assert '/api/records' in S + + +@pytest.mark.parametrize('apps,expected', [ + ({'ui'}, 'UIアプリ'), ({'api'}, 'APIアプリ(/api)'), ({'ui', 'api'}, '両方')]) +def test_app列の期待値(apps, expected): + assert reconcile.app_expected(apps) == expected diff --git a/tools/api-inventory/tests/test_test_coverage.py b/tools/api-inventory/tests/test_test_coverage.py new file mode 100644 index 0000000000..7b50771c4c --- /dev/null +++ b/tools/api-inventory/tests/test_test_coverage.py @@ -0,0 +1,143 @@ +# -*- coding: utf-8 -*- +"""test_coverage.py — 各行のテストが4観点を押さえているかを静的に判定する。 + +これは**キーワード判定であり、テストの十分性は見ていない**。「観点が全く見当たらない」 +ことの検出にだけ使える。だからこそ、判定が緩む方向に壊れると +「テストがある」と誤って言い切る台帳が出来上がる。 +""" +import os + +import pytest + +import test_coverage as tc +from conftest import make_row, write_full, run + + +def analyse(src): + return tc.analyse({'t.py::test_x': src}) + + +# --- 4観点の判定 ----------------------------------------------------------- + +@pytest.mark.parametrize('code', [ + 'assert res.status_code == 200', + 'assert res.status_code == 201', + 'assert res.status_code in (200, 302)', +]) +def test_正常値は2xxの検証で立つ(code): + assert analyse(code)['normal'] + + +@pytest.mark.parametrize('code', [ + 'assert res.status_code == 403', + 'assert res.status_code == 500', + 'assert res.status_code in (400, 422)', +]) +def test_異常値は4xx5xxの検証で立つ(code): + assert analyse(code)['abnormal'] + + +def test_2xxだけなら異常値は立たない(): + r = analyse('assert res.status_code == 200') + assert r['normal'] and not r['abnormal'] + + +@pytest.mark.parametrize('code', [ + 'with pytest.raises(ValueError):\n f()', + 'self.assertRaises(KeyError, f)', +]) +def test_例外処理は例外検証で立つ(code): + assert analyse(code)['exception'] + + +def test_境界値はparametrizeで立つ(): + assert analyse('@pytest.mark.parametrize("v", [0, 1])\ndef test_x(v): pass')['boundary'] + + +def test_境界値は関数名からも立つ(): + """本体に現れなくても、名前が境界を狙っていると分かるものは拾う。""" + assert tc.analyse({'t.py::test_empty_title': 'assert True'})['boundary'] + + +def test_観点が何も無ければ全て偽(): + r = analyse('assert res is not None') + assert not any(r.values()) + + +# --- 対応するテスト関数の特定 ---------------------------------------------- + +def test_URIから検索に使う静的部分を取り出す(): + assert tc.norm_static('/api/records//files') == 'files' + assert tc.norm_static('/admin/community/new/') == 'new' + assert tc.norm_static('/') == '' + + +def _run_on(tmp_path, rows, test_src=None, test_rel='modules/demo/tests/test_x.py'): + root = tmp_path / 'repo' + if test_src is not None: + p = root / test_rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(test_src, encoding='utf-8') + full = write_full(tmp_path / 'full.tsv', rows) + run('test_coverage.py', '--full', full, '--weko-root', str(root), expect=0) + out = [l.rstrip('\n').split('\t') for l in open(full, encoding='utf-8')] + return out[0], out[1:] + + +def test_同じファイル内の別APIのテストを自分のものにしない(tmp_path): + """ファイル単位で見ると、隣の API のアサーションを自分の観点として数えてしまう。""" + src = ''' +def test_other_api(client): + res = client.get('/other') + assert res.status_code == 200 + +def test_mine(client): + res = client.get('/mine') + assert something(res) +''' + hdr, [r] = _run_on(tmp_path, [make_row( + uri='/mine', impl_func='mine_view', test_file='modules/demo/tests/test_x.py')], + test_src=src) + assert r[hdr.index('test_normal')] == '-' + + +def test_関係するテストが見つかれば観点を判定する(tmp_path): + src = ''' +def test_show_view_ok(client): + res = client.get('/show') + assert res.status_code == 200 +''' + hdr, [r] = _run_on(tmp_path, [make_row( + uri='/show', impl_func='show_view', test_file='modules/demo/tests/test_x.py')], + test_src=src) + assert r[hdr.index('test_normal')] == '○' + assert r[hdr.index('test_gap')] == '異常値,境界値,例外処理' + + +def test_名前の判定は部分一致なので過検出しうる(tmp_path): + """`min` は `mine` にも当たる。境界値の `○` は「それらしい名前がある」以上の + 意味を持たない。緩む方向の癖として明示的に固定しておく。""" + assert tc.analyse({'t.py::test_mine_ok': 'assert True'})['boundary'] + + +def test_特定不能とテスト無しを同じ記号にしない(tmp_path): + """どちらも '-' にすると、テストが本当に無い行と区別できなくなる。""" + hdr, [r] = _run_on(tmp_path, [make_row(test_file='-')]) + assert r[hdr.index('test_normal')] == '?' + assert r[hdr.index('test_gap')] == '特定不能' + + +def test_列を増やさず上書きする(tmp_path): + """毎回付け足すと実行のたびに列が増える。""" + rows = [make_row()] + hdr, _ = _run_on(tmp_path, rows) + from conftest import FULL_HEADER + assert sorted(hdr) == sorted(FULL_HEADER) + + +def test_dry_runは台帳を書き換えない(tmp_path): + full = write_full(tmp_path / 'full.tsv', [make_row()]) + before = open(full, encoding='utf-8').read() + run('test_coverage.py', '--full', full, '--weko-root', str(tmp_path), + '--dry-run', expect=0) + assert open(full, encoding='utf-8').read() == before