feat: API 台帳の優先度判定で永続データの変更とデコレータ以外の認証を考慮する

tools/api-inventory/scripts/README.md

@@ -1163,7 +1163,7 @@ python3 scripts/build_checklist.py     # 24列版(=32列)へ引き継ぐ
 ```
 
 **実行順が重要**: `prioritize.py` は `test_gap` を参照して「テスト観点が確認できない行」を
-P2 まで引き上げるため、`test_coverage.py` を先に回すこと。
+P3 まで引き上げるため、`test_coverage.py` を先に回すこと。
 
 ## test_coverage.py
 
@@ -1182,20 +1182,68 @@ P2 まで引き上げるため、`test_coverage.py` を先に回すこと。
 
 「データ破壊」は **既存の実データを不可逆に壊すこと** と定義している。メタデータの
 更新や新規作成は含めない。新規作成しかせず既存データを壊さないものは、認証が
-無くても P0 ではなく P1 に置く。
+無くても P1 ではなく P2 に置く。
+
+状態変更系の P1 は、**永続データ(データベース/ファイルストレージ)を作成・更新・
+削除するもの** に限る。`data_store` の保存先種別が PostgreSQL かファイルである
+ことで判定する。POST で受けるだけの取得(ログイン・検索・エクスポート等)や、
+セッション・Redis・ES・外部サービスだけを書き換えるもの(言語切替・CAPTCHA・
+統計・Handle 等)は P1 にしない。
+
+アプリの外側で行う認証は `infra_auth.py` で判定し、見つかった行は `auth_required` が
+`不要` でも無認証として扱わない(P5)。
+
+- OAuth2 クライアント認証: 実装関数に `@oauth2.token_handler` / `@oauth2.revoke_handler`
+- nginx: URI に一致する location(nginx と同じ優先順位で選ぶ)に `deny all`・
+  `auth_basic`・`auth_request`・`limit_except`、または `if ($var) { return 403; }`
+  があるもの。`$var` が `map` なら `$request_method` を当てはめ、他の変数(接続元等)は
+  任意の値として評価する。行の全メソッドで拒否される場合だけ守られているとみなす
+- Shibboleth SP: `shib_request` のある location で、`shibboleth2.xml` の RequestMap が
+  そのパスに `requireSession` を効かせているもの
+
+nginx と SP の設定はリポジトリ同梱の `nginx/weko.conf`・`nginx/shibboleth2.xml` を読む
+(`WEKO_NGINX_CONF`・`WEKO_SHIB_CONF` で差し替え可)。サイトごとに設定を変えている
+場合、その差は判定に入らない。
+
+ビューの中で行う本人認証は `credential_auth.py` で判定し、同じく P5 にする。
+
+- パスワード: `verify_password` / `check_password` / `verify_and_update_password`
+- トークン: 関数名が「検証の動詞 + token」のもの(`convert_token_into_obj`、
+  `validate_*_token`、`*_token_status` 等)。CSRF トークンと CAPTCHA は本人確認では
+  ないので含めない
+- 実装関数と、そこから呼ぶ同じパッケージ内の関数を1段までたどる(同名の定義が複数
+  あるものはたどらない)。`post` → `post_v1` のような振り分け先も含める
+- ソースの無い Flask-Security の行は endpoint 名(`security.login` 等)で判定する
+
+本人認証をするビュー(ログイン等)は未認証のリクエストがビューまで届くのが正常なので、
+実測の「未認証で到達」は★実証(突破)が無ければ認証の欠落として数えない。
+
+テストは `python3 -m pytest tools/api-inventory/tests`。
+
+管理者向け(admin-access 相当)の行も確認対象から外さない(以前の「対象外」は廃止)。
+リポジトリ管理者・コミュニティ管理者は自分の管轄だけを操作できるはずなので、
+`roles` と `bola_risk` で次のように振り分ける。指摘や★実証がある行は P1/P2 の判定が先に拾う。
+
+| 条件 | 優先度 |
+|---|---|
+| `bola_risk` に★(ID 直指定で他リソースを操作できる懸念)の書き込み系 | P2 |
+| 同 読み取り系 | P3 |
+| リポジトリ/コミュニティ管理者も使える書き込み系(管轄の限定を確かめる) | P3 |
+| リポジトリ/コミュニティ管理者も使える読み取り系 | P5 |
+| システム管理者のみ | P5 |
 
 参照系でも、露出内容が **認証情報** または **非公開データの実体** であり認可が
-緩い行は P1 に上げる。「読み取り系だから限定的」は露出物次第で成り立たないため。
+緩い行は P2 に上げる。「読み取り系だから限定的」は露出物次第で成り立たないため。
 
-`deprecated` の記述から **非利用** を判定し、認可上の判定が P2 以下なら
+`deprecated` の記述から **非利用** を判定し、認可上の判定が P3 以下なら
 `整理対象` に置き換える(削除すれば認可の問題ごと消える)。
 
 `reconcile_allow.json` に登録された URI(=実機の url_map に無いことを確認済み)は
 `環境依存` とする。**`整理対象` とは別物で、削除してはいけない**。別の設定・別サイトでは
 有効になるため。判定に `dynamic_verified` を使わないのは、実測欄の粒度がばらついていて
 同じ機能群が別区分に割れるため(プラグイン群8件が対象外3件と整理対象5件に割れていた)。
 
-いずれも P0/P1 の行は優先度を落とさず、理由に事情を添えるだけにする。
+いずれも P1/P2 の行は優先度を落とさず、理由に事情を添えるだけにする。
 
 末尾の派生列(priority / test_* / cleanup)の並びは `prioritize.py` が正規化するため、
 実行順に依存しない。

tools/api-inventory/scripts/credential_auth.py

@@ -0,0 +1,146 @@
+# -*- coding: utf-8 -*-
+"""ビューの中で行う本人認証(パスワード・トークンの検証)を判定する。
+
+ログインやメール確認、ワンタイムダウンロードのように、ログインセッションの
+代わりにリクエストで渡された認証情報を確かめるビューは、デコレータを持たない
+ため台帳では auth_required が「不要」になる。実装を読んで次を探す。
+
+  パスワード   verify_password / check_password / verify_and_update_password
+  トークン     関数名が「検証の動詞 + token」のもの(validate_*_token、
+               convert_token_into_obj、confirm_email_token_status 等)
+
+対象は実装関数と、そこから呼ぶ **同じパッケージ内の関数を1段** まで。
+同名の定義が複数ある関数(get / post 等)はどれが呼ばれるか決まらないのでたどらない。
+クラスベースのビューで post → post_v1 のように getattr で振り分けている
+場合に備え、同じクラスの `<メソッド名>_*` も含める。
+
+CSRF トークンと CAPTCHA は本人確認ではないので含めない。
+
+ソースの無いフレームワークの行(Flask-Security 等)は endpoint 名で判定する。
+"""
+import ast
+import functools
+import os
+import re
+import warnings
+
+from infra_auth import repo_root
+
+PASSWORD_CALLS = ('verify_password', 'check_password', 'verify_and_update_password')
+TOKEN_CALL_RE = re.compile(
+    r'(?:verify|validate|check|confirm|convert)\w*token|token\w*_(?:status|valid)',
+    re.I)
+NOT_IDENTITY_RE = re.compile(r'csrf|captcha', re.I)
+
+# ソースが台帳に無いフレームワークのエンドポイント
+FRAMEWORK = {
+    'security.login': 'password(Flask-Security login)',
+    'security.confirm_email': 'token(Flask-Security confirm_email)',
+    'security.reset_password': 'token(Flask-Security reset_password)',
+}
+
+HTTP_METHODS = ('get', 'post', 'put', 'patch', 'delete')
+
+
+def _calls(node):
+    names = set()
+    for n in ast.walk(node):
+        if isinstance(n, ast.Call):
+            f = n.func
+            if isinstance(f, ast.Attribute):
+                names.add(f.attr)
+            elif isinstance(f, ast.Name):
+                names.add(f.id)
+    return names
+
+
+@functools.lru_cache(maxsize=None)
+def _parse(path):
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore', SyntaxWarning)
+            return ast.parse(open(path, encoding='utf-8', errors='replace').read())
+    except (OSError, SyntaxError, ValueError):
+        return None
+
+
+@functools.lru_cache(maxsize=None)
+def _package_index(pkg_dir):
+    """パッケージ内の関数名 → [呼び出している関数名の集合, ...](定義ごと)。"""
+    index = {}
+    for base, _, files in os.walk(pkg_dir):
+        for f in files:
+            if not f.endswith('.py'):
+                continue
+            tree = _parse(os.path.join(base, f))
+            if tree is None:
+                continue
+            for n in ast.walk(tree):
+                if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                    index.setdefault(n.name, []).append(_calls(n))
+    return index
+
+
+def _targets(tree, line):
+    """impl_line を含む関数と、post → post_v1 のような振り分け先を返す。"""
+    best = None
+    parent = {}
+    for n in ast.walk(tree):
+        for ch in ast.iter_child_nodes(n):
+            parent[ch] = n
+        if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            start = min([n.lineno] + [d.lineno for d in n.decorator_list])
+            end = getattr(n, 'end_lineno', n.lineno)
+            if start <= line <= end and (best is None or end - start < best[1] - best[0]):
+                best = (start, end, n)
+    if not best:
+        return []
+    fn = best[2]
+    out = [fn]
+    cls = parent.get(fn)
+    if isinstance(cls, ast.ClassDef) and fn.name in HTTP_METHODS:
+        out += [m for m in cls.body
+                if isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef))
+                and m.name.startswith(fn.name + '_')]
+    return out
+
+
+def _kinds(names):
+    pw = sorted(n for n in names if n in PASSWORD_CALLS)
+    tk = sorted(n for n in names
+                if TOKEN_CALL_RE.search(n) and not NOT_IDENTITY_RE.search(n))
+    out = []
+    if pw:
+        out.append(f'password: {", ".join(pw)}')
+    if tk:
+        out.append(f'token: {", ".join(tk)}')
+    return out
+
+
+def credential_auth(impl_file, impl_line, endpoint=''):
+    """実装がパスワード/トークンで本人を確かめていれば説明を返す。無ければ ''。"""
+    if endpoint in FRAMEWORK:
+        return f'app-credential({FRAMEWORK[endpoint]})'
+    try:
+        line = int(impl_line)
+    except (TypeError, ValueError):
+        return ''
+    if not impl_file or impl_file.startswith('('):
+        return ''
+    path = os.path.join(repo_root(), impl_file)
+    tree = _parse(path)
+    if tree is None:
+        return ''
+    direct = set()
+    for fn in _targets(tree, line):
+        direct |= _calls(fn)
+    index = _package_index(os.path.dirname(path))
+    reached = set(direct)
+    for name in direct:
+        # 同名の定義が複数ある(get / post 等)ものはどれが呼ばれるか決まらないので
+        # たどらない。無関係な関数の検証処理を拾ってしまうため。
+        defs = index.get(name, [])
+        if len(defs) == 1:
+            reached |= defs[0]
+    kinds = _kinds(reached)
+    return f'app-credential({"; ".join(kinds)})' if kinds else ''