From 655f8e441abb013236bb9c5842456771c334f632 Mon Sep 17 00:00:00 2001 From: Masaharu Hayashi Date: Fri, 30 Oct 2026 17:22:19 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat(api-inventory):=20=E5=84=AA=E5=85=88?= =?UTF-8?q?=E5=BA=A6=E3=81=AE=E5=88=A4=E5=AE=9A=E3=81=A7=E6=B0=B8=E7=B6=9A?= =?UTF-8?q?=E3=83=87=E3=83=BC=E3=82=BF=E3=81=AE=E5=A4=89=E6=9B=B4=E3=81=A8?= =?UTF-8?q?=E3=83=87=E3=82=B3=E3=83=AC=E3=83=BC=E3=82=BF=E4=BB=A5=E5=A4=96?= =?UTF-8?q?=E3=81=AE=E8=AA=8D=E8=A8=BC=E3=82=92=E8=80=83=E6=85=AE=E3=81=99?= =?UTF-8?q?=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P1 の状態変更系は、データベース/ファイルストレージを作成・更新・削除する ものに限る(POST で受ける取得やセッションだけの更新を含めない) - infra_auth: OAuth2 クライアント認証、nginx の location の制限、 Shibboleth SP の requireSession を判定する - credential_auth: ビュー内のパスワード・トークンの検証を判定する - 上記の認証がある行は auth_required が「不要」でも無認証として扱わない - docstring と README を実際の判定に合わせる Co-Authored-By: Claude Opus 5.5 --- tools/api-inventory/scripts/README.md | 46 ++- .../api-inventory/scripts/credential_auth.py | 146 +++++++++ tools/api-inventory/scripts/infra_auth.py | 290 ++++++++++++++++++ tools/api-inventory/scripts/prioritize.py | 116 +++++-- .../tests/test_credential_auth.py | 148 +++++++++ tools/api-inventory/tests/test_infra_auth.py | 186 +++++++++++ tools/api-inventory/tests/test_prioritize.py | 89 ++++++ 7 files changed, 984 insertions(+), 37 deletions(-) create mode 100644 tools/api-inventory/scripts/credential_auth.py create mode 100644 tools/api-inventory/scripts/infra_auth.py create mode 100644 tools/api-inventory/tests/test_credential_auth.py create mode 100644 tools/api-inventory/tests/test_infra_auth.py create mode 100644 tools/api-inventory/tests/test_prioritize.py diff --git a/tools/api-inventory/scripts/README.md b/tools/api-inventory/scripts/README.md index 91ea82f8c7..1ca48cafab 100644 --- a/tools/api-inventory/scripts/README.md +++ b/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,12 +1182,48 @@ 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`。 参照系でも、露出内容が **認証情報** または **非公開データの実体** であり認可が -緩い行は P1 に上げる。「読み取り系だから限定的」は露出物次第で成り立たないため。 +緩い行は P2 に上げる。「読み取り系だから限定的」は露出物次第で成り立たないため。 -`deprecated` の記述から **非利用** を判定し、認可上の判定が P2 以下なら +`deprecated` の記述から **非利用** を判定し、認可上の判定が P3 以下なら `整理対象` に置き換える(削除すれば認可の問題ごと消える)。 `reconcile_allow.json` に登録された URI(=実機の url_map に無いことを確認済み)は @@ -1195,7 +1231,7 @@ P2 まで引き上げるため、`test_coverage.py` を先に回すこと。 有効になるため。判定に `dynamic_verified` を使わないのは、実測欄の粒度がばらついていて 同じ機能群が別区分に割れるため(プラグイン群8件が対象外3件と整理対象5件に割れていた)。 -いずれも P0/P1 の行は優先度を落とさず、理由に事情を添えるだけにする。 +いずれも P1/P2 の行は優先度を落とさず、理由に事情を添えるだけにする。 末尾の派生列(priority / test_* / cleanup)の並びは `prioritize.py` が正規化するため、 実行順に依存しない。 diff --git a/tools/api-inventory/scripts/credential_auth.py b/tools/api-inventory/scripts/credential_auth.py new file mode 100644 index 0000000000..7b573aab27 --- /dev/null +++ b/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 '' diff --git a/tools/api-inventory/scripts/infra_auth.py b/tools/api-inventory/scripts/infra_auth.py new file mode 100644 index 0000000000..6a64ce5748 --- /dev/null +++ b/tools/api-inventory/scripts/infra_auth.py @@ -0,0 +1,290 @@ +# -*- coding: utf-8 -*- +"""アプリの外側で行われる認証を判定する。 + +台帳の auth_required / auth_mechanism はアプリのデコレータしか見ていない。 +次の認証はデコレータに現れないため、それだけだと「無認証」に見える。 + + OAuth2 クライアント認証 + @oauth2.token_handler / @oauth2.revoke_handler のビューは、oauthlib が + client_id/secret・認可コード・リフレッシュトークンを検証する。 + nginx + location に shib_request / deny all / auth_basic / auth_request、 + または `if ($var) { return 401|403; }` があるもの。 + Shibboleth SP + shib_request は SP の RequestMap で requireSession が有効なパスでだけ + セッションを要求する。無効なパスでは素通しになるので保護とみなさない。 + +nginx と SP の設定は **リポジトリ同梱のもの** を読む。サイトごとに設定を +変えている場合、その差は判定に入らない。 + + WEKO_ROOT リポジトリのルート(既定: このファイルから3階層上) + WEKO_NGINX_CONF nginx の設定(既定: /nginx/weko.conf) + WEKO_SHIB_CONF SP の設定(既定: /nginx/shibboleth2.xml) +""" +import ast +import functools +import os +import re +import warnings +import xml.etree.ElementTree as ET + +# クライアント認証を伴う Flask-OAuthlib のハンドラ +CLIENT_AUTH_HANDLERS = ('token_handler', 'revoke_handler') + +# 台帳の URI のパラメータ( 等)を置き換える値 +_PARAM_RE = re.compile(r'<[^>]*>') +# 他の変数(接続元アドレス等)には「任意の値」としてこれを入れる +_ARBITRARY = '\x01' + + +def repo_root(): + return os.environ.get('WEKO_ROOT') or os.path.abspath( + os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', '..')) + + +def _conf_path(env, rel): + return os.environ.get(env) or os.path.join(repo_root(), rel) + + +# --- OAuth2 ------------------------------------------------------------------ + +@functools.lru_cache(maxsize=None) +def _functions(path): + """ファイル内の関数を (開始行, 終了行, デコレータ名の一覧) で返す。""" + try: + with warnings.catch_warnings(): + # 解析対象のソースの不正なエスケープ等の警告は、ここでは関係ない + warnings.simplefilter('ignore', SyntaxWarning) + tree = ast.parse(open(path, encoding='utf-8', errors='replace').read()) + except (OSError, SyntaxError, ValueError): + return () + out = [] + for n in ast.walk(tree): + if not isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + names = [] + for d in n.decorator_list: + f = d.func if isinstance(d, ast.Call) else d + if isinstance(f, ast.Attribute): + names.append(f.attr) + elif isinstance(f, ast.Name): + names.append(f.id) + start = min([n.lineno] + [d.lineno for d in n.decorator_list]) + out.append((start, getattr(n, 'end_lineno', n.lineno), tuple(names))) + return tuple(out) + + +def oauth2_client_auth(impl_file, impl_line): + """実装関数が OAuth2 のクライアント認証を伴うハンドラなら説明を返す。""" + 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) + best = None + for start, end, names in _functions(path): + if start <= line <= end and (best is None or end - start < best[1] - best[0]): + best = (start, end, names) + if not best: + return '' + hit = [n for n in best[2] if n in CLIENT_AUTH_HANDLERS] + return f'oauth2-client(@{hit[0]})' if hit else '' + + +# --- nginx ------------------------------------------------------------------- + +_TOKEN_RE = re.compile(r'"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'|[{};]|[^\s{};]+') + + +def _tokens(text): + for line in text.splitlines(): + # 引用符の外の # 以降はコメント + out, quote = [], None + for ch in line: + if quote: + quote = None if ch == quote else quote + elif ch in '"\'': + quote = ch + elif ch == '#': + break + out.append(ch) + for t in _TOKEN_RE.findall(''.join(out)): + yield t + + +def parse_nginx(text): + """nginx の設定を [(名前, 引数, 子 or None), ...] の木にする。""" + stack = [[]] + words = [] + for t in _tokens(text): + if t == ';': + if words: + stack[-1].append((words[0], words[1:], None)) + words = [] + elif t == '{': + node = (words[0] if words else '', words[1:], []) + stack[-1].append(node) + stack.append(node[2]) + words = [] + elif t == '}': + if len(stack) > 1: + stack.pop() + words = [] + else: + words.append(t[1:-1] if t[:1] in '"\'' and t[-1:] == t[:1] else t) + return stack[0] + + +@functools.lru_cache(maxsize=None) +def _nginx(path): + try: + return parse_nginx(open(path, encoding='utf-8').read()) + except OSError: + return None + + +def _servers(tree): + servers = [n for n in tree if n[0] == 'server' and n[2] is not None] + tls = [s for s in servers + if any(d[0] == 'listen' and ('443' in ' '.join(d[1]) or 'ssl' in d[1]) + for d in s[2])] + return tls or servers + + +def _maps(tree): + """map ブロックを {変数名: (元の文字列, 既定値, [(キー, 値), ...])} にする。""" + maps = {} + for name, args, children in tree: + if name != 'map' or children is None or len(args) != 2: + continue + default, entries = '', [] + for k, v, _ in children: + val = v[0] if v else '' + if k == 'default': + default = val + elif k not in ('hostnames', 'volatile', 'include'): + entries.append((k, val)) + maps[args[1]] = (args[0], default, entries) + return maps + + +def _eval_map(m, method): + """map の値を、$request_method=method・他の変数=任意の値として求める。""" + source, default, entries = m + s = source.replace('$request_method', method) + s = re.sub(r'\$\{?\w+\}?', _ARBITRARY, s) + for k, v in entries: + if not k.startswith('~') and k == s: + return v + for k, v in entries: + if k.startswith('~*'): + if re.search(k[2:], s, re.I): + return v + elif k.startswith('~'): + if re.search(k[1:], s): + return v + return default + + +def match_location(server, path): + """nginx と同じ規則で path に一致する location ブロックを返す。""" + locs = [(a[0], a[1], c) if a and a[0] in ('=', '^~', '~', '~*') else ('', a[0] if a else '', c) + for n, a, c in server[2] if n == 'location' and c is not None] + for mod, pat, c in locs: + if mod == '=' and pat == path: + return f'= {pat}', c + prefix = None + for mod, pat, c in locs: + if mod in ('', '^~') and path.startswith(pat) and (prefix is None or len(pat) > len(prefix[1])): + prefix = (mod, pat, c) + if prefix and prefix[0] == '^~': + return prefix[1], prefix[2] + for mod, pat, c in locs: + if mod in ('~', '~*') and re.search(pat, path, re.I if mod == '~*' else 0): + return f'{mod} {pat}', c + return (prefix[1], prefix[2]) if prefix else (None, None) + + +def _requires_shib_session(path): + """SP の RequestMap で path に requireSession が効いているか。""" + p = _conf_path('WEKO_SHIB_CONF', 'nginx/shibboleth2.xml') + try: + root = ET.parse(p).getroot() + except (OSError, ET.ParseError): + return False + on = lambda e, inherited: (e.get('requireSession', inherited).lower() + in ('true', '1')) + for host in root.iter(): + if not host.tag.endswith('}Host') and host.tag != 'Host': + continue + req = on(host, 'false') + best = None + for sub in host.iter(): + if (sub.tag.endswith('}Path') or sub.tag == 'Path') and sub.get('name'): + prefix = '/' + sub.get('name').strip('/') + if (path == prefix or path.startswith(prefix + '/')) and ( + best is None or len(prefix) > len(best[0])): + best = (prefix, on(sub, 'true' if req else 'false')) + if (best[1] if best else req): + return True + return False + + +def _protection(block, maps, method, path): + """location ブロックの中で、method のリクエストを拒否/認証する仕組みを返す。""" + found = [] + for name, args, children in block: + if name == 'shib_request' and args and args[0] != 'off': + if _requires_shib_session(path): + found.append('shibboleth(shib_request+requireSession)') + elif name == 'deny' and args == ['all']: + found.append('nginx(allow/deny)') + elif name == 'auth_basic' and args and args[0] != 'off': + found.append('nginx(auth_basic)') + elif name == 'auth_request' and args and args[0] != 'off': + found.append('nginx(auth_request)') + elif name == 'limit_except' and children is not None \ + and method not in args \ + and not (method == 'HEAD' and 'GET' in args): + found += [p for p in _protection(children, maps, method, path)] + elif name == 'if' and children is not None: + denies = any(n == 'return' and a and a[0] in ('401', '403') + for n, a, _ in children) + cond = ' '.join(args).strip('()').strip() + if denies and re.fullmatch(r'\$\w+', cond) and cond in maps \ + and _eval_map(maps[cond], method) not in ('', '0'): + found.append(f'nginx(if {cond} → 403)') + return found + + +def nginx_auth(uri, methods): + """uri へのリクエストを nginx が(全 methods について)守っていれば説明を返す。""" + tree = _nginx(_conf_path('WEKO_NGINX_CONF', 'nginx/weko.conf')) + if not tree or not uri: + return '' + path = _PARAM_RE.sub('x', uri.strip()) + maps = _maps(tree) + labels = [] + for server in _servers(tree): + loc, block = match_location(server, path) + if block is None: + return '' + found = set() + for m in methods or ['GET']: + hit = _protection(block, maps, m, path) + if not hit: + return '' + found.update(hit) + labels += [f'{label} @location {loc}' for label in sorted(found)] + return '; '.join(labels) + + +def detect(uris, methods, impl_file, impl_line): + """アプリの外側の認証を説明する文字列を返す。無ければ ''。""" + found = oauth2_client_auth(impl_file, impl_line) + if found: + return found + hits = [nginx_auth(u, methods) for u in uris if u] + return hits[0] if hits and all(hits) else '' diff --git a/tools/api-inventory/scripts/prioritize.py b/tools/api-inventory/scripts/prioritize.py index 2a1f3dde1f..1f28e46be6 100644 --- a/tools/api-inventory/scripts/prioritize.py +++ b/tools/api-inventory/scripts/prioritize.py @@ -3,33 +3,52 @@ python3 prioritize.py # $WEKO_API_INVENTORY_DIR の台帳を更新 -判定基準(上から順に評価し、最初に合致したものを採用する): - - P1 無認証で既存のファイル実体を上書き/削除できる(データ破壊)、または - 状態変更系(POST/PUT/DELETE/PATCH)なのに認証・権限チェックが一切ない、 - または権限チェック機構はあるように見えるが実装上機能していない - P2 ログイン必須のみで所有者/ロール/スコープの限定がない状態変更系(IDOR疑い)、 - ゲストトークン等のバイパス、ロールチェックが実質不問、または「不明」。 - 認証が無い状態変更系でも新規作成のみで既存データを壊さないもの。 - 参照系でも露出内容が認証情報または非公開データの実体であるもの - P3 読み取り系(GET/HEAD)で認証・権限チェックが無い、または - ログイン必須のみでロール/所有者スコープなし +判定基準(classify が上から順に評価し、最初に合致したものを採用する): + + 「無認証」は auth_required が 不要/任意(匿名可)、または実測で未認証のまま到達 + したこと。ただし infra_auth がアプリの外側の認証(OAuth2 クライアント認証・ + nginx・Shibboleth SP)を、credential_auth がビュー内の本人認証(パスワード・ + トークンの検証)を見つけた行は、auth_required が 不要 でも無認証としない。 + ビュー内の本人認証がある行は、実測の「未認証で到達」も★実証が無ければ数えない + (ログイン等は未認証のリクエストがビューまで届くのが正常なため)。 + 「書き込み系」は method に POST/PUT/DELETE/PATCH を含むこと。 + 「新規作成のみ」は data_op に 作成 を含み 更新/削除 を含まないこと。 + 「永続データを変更」は data_store にデータベース(PostgreSQL)かファイルストレージ + があり、data_op に 作成/更新/削除 を含むこと。セッション・Redis・ES・ + 外部サービスだけを書き換えるものは含めない。 + + P1 (1) 無認証で、data_store がファイル実体かつ data_op が更新/削除 + (2) 書き込み系・無認証で、永続データを変更し、新規作成のみではない + (3) 書き込み系で指摘に「実効せず」があり、永続データを変更し、 + 新規作成のみではない + ※ POST で受けるだけの取得(ログイン・検索・エクスポート等)や + セッションだけを変えるもの(言語切替等)は P1 にならない + P2 書き込み系で、無認証/「実効せず」だが新規作成のみ、 + ゲストトークンのバイパス経路がある、ログインのみで到達するか + 所有者/ロールの限定が欠けている、または到達可否が未測定。 + 読み取り系でも、認可が緩く(無認証・ログインのみ・スコープ欠落) + 指摘か★実証があり、露出内容が認証情報または非公開データの実体のもの + 対象外 auth が admin-role-table/admin-access 相当または auth_required が + 要(管理)で、指摘も★実証も無い P4 意図的な公開設計(static配信・ヘルスチェック・robots.txt・OAI-PMH等)、 または deny_all 常時拒否 - P5 具体的な権限/所有者チェック機構が明記されており破綻が見えない - 整理対象 非利用(未使用/非推奨/呼出元なし)で、認可上の判定が P3 以下 - 環境依存 実機の url_map に存在しない(プラグイン未導入・config で無効・動的登録) - 対象外 認証必須かつ admin-access 相当の権限チェックがあり、指摘も★実証も無い + P3 読み取り系で、無認証、またはログインのみ/スコープ欠落の指摘がある、 + または指摘があるが到達可否が未測定 + P5 デコレータ以外の仕組みで認証している(infra_auth/credential_auth)、 + 具体的な権限/所有者チェック機構がある、または auth_required が 要 で + 破綻が見えない + P3 上のいずれにも合致しない(手動確認が必要) + + P1/P2 は対象外・P4 より先に評価する。管理画面や公開設計のエンドポイントでも、 + P1/P2 の条件に合えばそちらが優先される。 -テスト観点による引き上げ(上限 P3): - 正常値/異常値/境界値/例外処理 のチェックが確認できない行は、認可上の問題が - 無くても確認対象に上げる。ただし引き上げは P3 まで。 +テスト観点による引き上げ(bump、対象外/P4/P5 にのみ適用): + test_gap が「特定不能」、または 正常値/異常値/境界値/例外処理 の4観点が + 全て欠けている行は P3 に上げる。一部の欠落なら優先度は変えず理由に追記する。 -評価順について: 「対象外」と P3 は P2 より先に評価する。admin-access で保護された -管理画面や、意図的に公開している OAI-PMH を「認証なしの読み取り系」として -P2 に落とすと、実際に見るべき行が埋もれるため。ただし **指摘(security_finding)や -★実証がある行は「対象外」にしない**(保護されているように見えて破綻している行を -除外してしまうため)。 +classify の後に decide が次を重ねる(いずれも P1/P2 の行は優先度を変えない): + 環境依存 実機の url_map に存在しない(プラグイン未導入・config で無効・動的登録) + 整理対象 非利用(未使用/非推奨/実質未使用/呼出元なし/経路なし) """ import argparse import json @@ -39,9 +58,18 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from paths import data_path # noqa: E402 +import credential_auth # noqa: E402 +import infra_auth # noqa: E402 WRITE_METHODS = {'POST', 'PUT', 'DELETE', 'PATCH'} +# data_store の保存先の種別がデータベース/ファイルストレージであるもの。 +# data_store は「対象 / 種別:詳細; 種別:詳細」の形なので、区切りの直後に +# 種別として現れたものだけを拾う(「ファイルダウンロード統計」や +# 「認証情報ファイル:」のような対象名・修飾語には反応させない)。 +PERSISTENT_STORE_RE = re.compile( + r'(?:^|[/;；→])\s*(?:PostgreSQL|ファイル)(?=[:：(]|\s*(?:[/;；→]|$))') + # 露出内容が「認証情報」とみなせる語 CREDENTIAL_WORDS = ( '露出:秘密情報', '秘密:', 'client_secret', 'access_token', 'refresh_token', @@ -118,7 +146,7 @@ def classify(c, H): """(priority, reason, cleanup) を返す。 cleanup は「非利用につき整理対象」の根拠。非利用の行は、認可上の問題が - P2 以下なら **削除すれば済む** ので `整理対象` に置き換える。P0/P1 の行は + P3 以下なら **削除すれば済む** ので `整理対象` に置き換える。P1/P2 の行は 優先度を落とさず、理由に「削除が最短の対応」であることを添える。 """ method = field(c, H, 'method').upper() @@ -143,10 +171,10 @@ 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 @@ -161,10 +189,29 @@ def bump(pri, why): is_write_method = bool(methods & WRITE_METHODS) is_destructive = ('削除' in data_op) - is_mutating = any(k in data_op for k in ('作成', '更新', '削除')) + # 永続データ(データベース/ファイルストレージ)を作成・更新・削除するか。 + # セッション・Redis・ES・外部サービスだけを書き換えるもの(言語切替・CAPTCHA・ + # 統計・Handle 等)は含めない。保存先が書かれていない行(data_store が '-')も同様。 + persists = bool(PERSISTENT_STORE_RE.search(store)) + is_mutating = persists and any(k in data_op for k in ('作成', '更新', '削除')) no_auth = auth_req in ('不要', '任意(匿名可)') + # アプリの外側(OAuth2 プロバイダ・nginx・Shibboleth SP)で行う認証と、 + # ビューの中で行う本人認証(パスワード・トークンの検証)。 + # デコレータに現れないので auth_required は「不要」になっているが、無認証ではない。 + impl_file, impl_line = field(c, H, 'impl_file'), field(c, H, 'impl_line') + infra = infra_auth.detect( + [u.strip() for u in uri.split(';')], sorted(methods), impl_file, impl_line) + cred = '' if infra else credential_auth.credential_auth( + impl_file, impl_line, field(c, H, 'endpoint')) + if infra or cred: + no_auth = False unauth_reach = ('未認証で到達' in dyn) or ('未認証で' in dyn and '★' in dyn) proven = '★' in dyn + # ビューで本人認証をするもの(ログイン等)は、未認証のリクエストがビューまで + # 届くのが正常な動き。届いただけでは認証の欠落とみなさず、★実証(突破)が + # あるときだけ残す。 + if cred and not proven: + unauth_reach = False broken_authz = '実効せず' in finding login_only = ('ログインのみで到達' in dyn) or ('低権限ログインで到達' in dyn) scope_missing = any(k in finding for k in ( @@ -182,12 +229,15 @@ def bump(pri, why): return 'P1', f'無認証で既存のファイル実体を上書き/削除できる (data_op={data_op})' # --- P1: 状態変更系で認証・権限が無い/機能していない --- - # ただし新規作成しかせず既存データを壊さないものは P1 に下げる(下の P1 で拾う)。 + # 永続データを作成・更新・削除するものに限る。POST で受けるだけの取得や、 + # セッションだけを変えるものは P1 にしない。 + # ただし新規作成しかせず既存データを壊さないものは P2 に下げる(下の P2 で拾う)。 creates_only = ('作成' in data_op) and not ('更新' in data_op or '削除' in data_op) - if is_write_method and (no_auth or unauth_reach) and not creates_only: + if is_write_method and is_mutating and (no_auth or unauth_reach) \ + and not creates_only: why = '認証チェックが無い' if no_auth else '認証はあるが未認証で到達(実測)' return 'P1', f'状態変更系({method})で{why}' - if is_write_method and broken_authz and not creates_only: + if is_write_method and is_mutating and broken_authz and not creates_only: return 'P1', f'状態変更系({method})だが権限チェックが実装上機能していない' # --- P2 --- @@ -203,9 +253,9 @@ def bump(pri, why): if is_write_method and unknown: return 'P2', f'状態変更系({method})だが到達可否が未測定(不明)' - # --- 露出内容による引き上げ(参照系でも P1) --- + # --- 露出内容による引き上げ(参照系でも P2) --- # 「読み取り系だから情報漏洩リスクは限定的」は、露出するものが認証情報や - # 非公開データの実体である場合には成り立たない。認可が緩い行に限って P1 に上げる。 + # 非公開データの実体である場合には成り立たない。認可が緩い行に限って P2 に上げる。 # 「管理者に集約」のように適切に保護されている行は対象にしない。 # restricted_content / access_variance は「何が制限されているか」の説明であり、 # 適切に絞られている旨の記述にも「非公開」が出てくる。露出の根拠には使わない @@ -249,6 +299,8 @@ def bump(pri, why): return 'P3', f'読み取り系({method})に指摘があるが到達可否は未測定' # --- P5 --- + if infra or cred: + return bump('P5', f'デコレータ以外の仕組みで認証している({infra or cred})') hit = [k for k in CONCRETE_AUTHZ if k in auth] if hit: return bump('P5', f'具体的な権限チェック機構あり({", ".join(hit[:3])})') diff --git a/tools/api-inventory/tests/test_credential_auth.py b/tools/api-inventory/tests/test_credential_auth.py new file mode 100644 index 0000000000..70f075813b --- /dev/null +++ b/tools/api-inventory/tests/test_credential_auth.py @@ -0,0 +1,148 @@ +# -*- coding: utf-8 -*- +"""credential_auth のテスト。 + + python3 -m pytest tools/api-inventory/tests +""" +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), + '..', 'scripts')) +import credential_auth # noqa: E402 + +VIEWS = ''' +from .api import ShibUser + + +@blueprint.route('/confim/user', methods=['POST']) +def confirm_user(): + if request.form.get('csrf_random', '') != session['csrf_random']: + abort(400) + shib_user = ShibUser() + if not shib_user.check_weko_user(account, password): + abort(400) + return redirect('/') + + +@blueprint.route('/download', methods=['POST']) +def download_onetime(): + url_obj = convert_token_into_obj(token) + return send() + + +@blueprint.route('/lang', methods=['POST']) +def set_lang(): + session['lang'] = request.form['lang'] + return redirect('/') + + +@blueprint.route('/csrf', methods=['POST']) +def only_csrf(): + validate_csrf_token(request.form['csrf_token']) + check_captcha_token(request.form['captcha']) + return 'ok' + + +@blueprint.route('/issue', methods=['POST']) +def issue(): + return generate_onetime_token() + + +@blueprint.route('/ambiguous', methods=['POST']) +def ambiguous(): + return get() + + +class WekoLogin(ContentNegotiatedMethodView): + def post(self, **kwargs): + func_name = f"post_{kwargs.get('version')}" + return getattr(self, func_name)(**kwargs) + + def post_v1(self, **kwargs): + if not verify_password(password, user.password): + raise InvalidPasswordError() + login_user(user) +''' + +API = ''' +class ShibUser(object): + def check_weko_user(self, account, pwd): + return verify_password(pwd, self.user.password) + + +def get(): + return verify_password('a', 'b') + + +class Other(object): + def get(self): + return None +''' + + +@pytest.fixture(autouse=True) +def pkg(tmp_path, monkeypatch): + d = tmp_path / 'mod' / 'pkg' + d.mkdir(parents=True) + (d / 'views.py').write_text(VIEWS, encoding='utf-8') + (d / 'api.py').write_text(API, encoding='utf-8') + monkeypatch.setenv('WEKO_ROOT', str(tmp_path)) + credential_auth._parse.cache_clear() + credential_auth._package_index.cache_clear() + return d + + +def line_of(name): + for i, line in enumerate(VIEWS.splitlines(), 1): + if line.strip().startswith(f'def {name}('): + return i + raise AssertionError(name) + + +def check(name): + return credential_auth.credential_auth('mod/pkg/views.py', line_of(name)) + + +def test_password_in_callee_of_same_package(): + # confirm_user → ShibUser.check_weko_user → verify_password + assert check('confirm_user') == 'app-credential(password: verify_password)' + + +def test_token_verification(): + assert check('download_onetime') == 'app-credential(token: convert_token_into_obj)' + + +def test_class_view_dispatches_to_versioned_method(): + # post は getattr で post_v1 に振り分けるだけ + assert check('post') == 'app-credential(password: verify_password)' + + +@pytest.mark.parametrize('name', ['set_lang', 'only_csrf', 'issue']) +def test_not_identity_check(name): + # セッション更新・CSRF/CAPTCHA・トークンの発行は本人認証ではない + assert check(name) == '' + + +def test_ambiguous_callee_is_not_followed(): + # get は api.py に関数とメソッドの2つがあり、どちらが呼ばれるか決まらない + assert check('ambiguous') == '' + + +def test_framework_endpoint(): + assert credential_auth.credential_auth( + '(framework/自動生成)', '0', 'security.login') \ + == 'app-credential(password(Flask-Security login))' + assert credential_auth.credential_auth( + '(framework/自動生成)', '0', 'security.send_confirmation') == '' + + +@pytest.mark.parametrize('impl_file, impl_line', [ + ('(framework/自動生成)', '0'), + ('mod/pkg/views.py', '-'), + ('mod/pkg/missing.py', '1'), + ('', ''), +]) +def test_unknown_source(impl_file, impl_line): + assert credential_auth.credential_auth(impl_file, impl_line) == '' diff --git a/tools/api-inventory/tests/test_infra_auth.py b/tools/api-inventory/tests/test_infra_auth.py new file mode 100644 index 0000000000..c13bfbf757 --- /dev/null +++ b/tools/api-inventory/tests/test_infra_auth.py @@ -0,0 +1,186 @@ +# -*- coding: utf-8 -*- +"""infra_auth のテスト。 + + python3 -m pytest tools/api-inventory/tests +""" +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), + '..', 'scripts')) +import infra_auth # noqa: E402 + +NGINX = r''' +map "$request_method:$realip_remote_addr" $sp_denied { + default 1; + "~^(GET|HEAD):" 0; + "POST:127.0.0.1" 0; +} + +server { + listen 80; + return 301 https://$host$request_uri; +} + +server { + listen 443 ssl; + server_name weko3.example.org; + + location = /weko/shib/login { + if ($sp_denied) { + return 403; + } + uwsgi_pass app_server; + } + location /weko/shib { + uwsgi_pass app_server; + #deny all; + } + location ~ /secure/ { + shib_request /shibauthorizer; + } + location /nosession/ { + shib_request /shibauthorizer; + } + location /internal { + allow 127.0.0.1; + deny all; + } + location /basic { + auth_basic "restricted"; + } + location /readonly { + limit_except GET { + deny all; + } + } + location ^~ /static/secure { + root /var/www; + } + location / { + uwsgi_pass app_server; + } +} +''' + +SHIB = ''' + + + + + + + + + +''' + +VIEWS = ''' +@blueprint.route('/token', methods=['POST', ]) +@oauth2.token_handler +def access_token(): + return None + + +@blueprint.route('/authorize', methods=['GET', 'POST']) +@login_required +@oauth2.authorize_handler +def authorize(*args, **kwargs): + return True +''' + + +@pytest.fixture(autouse=True) +def confs(tmp_path, monkeypatch): + (tmp_path / 'nginx').mkdir() + (tmp_path / 'nginx' / 'weko.conf').write_text(NGINX, encoding='utf-8') + (tmp_path / 'nginx' / 'shibboleth2.xml').write_text(SHIB, encoding='utf-8') + (tmp_path / 'views.py').write_text(VIEWS, encoding='utf-8') + monkeypatch.setenv('WEKO_ROOT', str(tmp_path)) + monkeypatch.delenv('WEKO_NGINX_CONF', raising=False) + monkeypatch.delenv('WEKO_SHIB_CONF', raising=False) + infra_auth._nginx.cache_clear() + infra_auth._functions.cache_clear() + return tmp_path + + +def test_oauth2_token_handler(): + assert infra_auth.oauth2_client_auth('views.py', 4) \ + == 'oauth2-client(@token_handler)' + # デコレータの行を指していても同じ関数とみなす + assert infra_auth.oauth2_client_auth('views.py', 3) \ + == 'oauth2-client(@token_handler)' + + +def test_oauth2_authorize_handler_is_not_client_auth(): + assert infra_auth.oauth2_client_auth('views.py', 12) == '' + + +@pytest.mark.parametrize('impl_file, impl_line', [ + ('(framework/自動生成)', '0'), + ('views.py', '-'), + ('missing.py', '1'), + ('', ''), +]) +def test_oauth2_unknown_source(impl_file, impl_line): + assert infra_auth.oauth2_client_auth(impl_file, impl_line) == '' + + +def test_map_if_403_depends_on_method(): + # POST は 127.0.0.1 以外から 403 になるので守られている + assert infra_auth.nginx_auth('/weko/shib/login', ['POST']) \ + == 'nginx(if $sp_denied → 403) @location = /weko/shib/login' + # GET は map で 0 になるので素通し + assert infra_auth.nginx_auth('/weko/shib/login', ['GET']) == '' + # 1つでも素通しのメソッドがあれば守られていないとみなす + assert infra_auth.nginx_auth('/weko/shib/login', ['GET', 'POST']) == '' + + +def test_prefix_location_without_protection(): + assert infra_auth.nginx_auth('/weko/shib/logout', ['GET']) == '' + + +def test_shib_request_needs_require_session(): + assert infra_auth.nginx_auth('/secure/login.py', ['GET']) \ + == 'shibboleth(shib_request+requireSession) @location ~ /secure/' + # requireSession が無効なパスでは shib_request は素通し + assert infra_auth.nginx_auth('/nosession/x', ['GET']) == '' + + +def test_deny_all_and_auth_basic(): + assert infra_auth.nginx_auth('/internal/api', ['POST']) \ + == 'nginx(allow/deny) @location /internal' + assert infra_auth.nginx_auth('/basic/x', ['GET']) \ + == 'nginx(auth_basic) @location /basic' + + +def test_limit_except(): + assert infra_auth.nginx_auth('/readonly/x', ['GET']) == '' + assert infra_auth.nginx_auth('/readonly/x', ['POST']) \ + == 'nginx(allow/deny) @location /readonly' + + +def test_caret_tilde_prefix_wins_over_regex(): + # ^~ の前方一致は正規表現の location より優先される + assert infra_auth.nginx_auth('/static/secure/a.js', ['GET']) == '' + + +def test_uri_parameters_are_replaced(): + assert infra_auth.nginx_auth('/internal/', ['GET']) \ + == 'nginx(allow/deny) @location /internal' + + +def test_missing_nginx_conf(monkeypatch, tmp_path): + monkeypatch.setenv('WEKO_NGINX_CONF', str(tmp_path / 'none.conf')) + assert infra_auth.nginx_auth('/internal/api', ['POST']) == '' + + +def test_detect(): + assert infra_auth.detect(['/token'], ['POST'], 'views.py', '4') \ + == 'oauth2-client(@token_handler)' + assert infra_auth.detect(['/internal/a', '/internal/b'], ['GET'], '', '') \ + == 'nginx(allow/deny) @location /internal' + # 複数の URI のどれかが守られていなければ守られていないとみなす + assert infra_auth.detect(['/internal/a', '/weko/shib/x'], ['GET'], '', '') == '' diff --git a/tools/api-inventory/tests/test_prioritize.py b/tools/api-inventory/tests/test_prioritize.py new file mode 100644 index 0000000000..ad12b26547 --- /dev/null +++ b/tools/api-inventory/tests/test_prioritize.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- +"""prioritize.classify の P1 判定のテスト。 + + python3 -m pytest tools/api-inventory/tests +""" +import os +import sys + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), + '..', 'scripts')) +import credential_auth # noqa: E402 +import infra_auth # noqa: E402 +import prioritize # noqa: E402 + +COLS = ['method', 'uri', 'auth_required', 'auth_method', 'data_op', 'data_store', + 'dynamic_verified', 'impl_file', 'impl_line', 'endpoint', 'test_gap'] +H = {n: i for i, n in enumerate(COLS)} + +VIEWS = ''' +def login(): + if not verify_password(password, user.password): + abort(401) + login_user(user) + + +def update_item(): + item.update(request.json) +''' + + +@pytest.fixture(autouse=True) +def repo(tmp_path, monkeypatch): + (tmp_path / 'pkg').mkdir() + (tmp_path / 'pkg' / 'views.py').write_text(VIEWS, encoding='utf-8') + monkeypatch.setenv('WEKO_ROOT', str(tmp_path)) + # nginx の設定は置かない(アプリの外側の認証は無し) + monkeypatch.setenv('WEKO_NGINX_CONF', str(tmp_path / 'none.conf')) + for f in (infra_auth._nginx, infra_auth._functions, + credential_auth._parse, credential_auth._package_index): + f.cache_clear() + + +def row(**kw): + base = {'method': 'POST', 'uri': '/x', 'auth_required': '不要', + 'auth_method': 'none', 'data_op': '更新', + # 未測定('-')だと「到達可否が未測定」で P2 になるので測定済みにする + 'data_store': '対象 / PostgreSQL:t', + 'dynamic_verified': '[実測] anon=400(到達)', + 'impl_file': 'pkg/views.py', 'impl_line': '9', 'endpoint': '', + 'test_gap': '-'} + base.update(kw) + return [base[n] for n in COLS] + + +def test_unauthenticated_db_update_is_p1(): + assert prioritize.classify(row(), H)[0] == 'P1' + + +@pytest.mark.parametrize('kw', [ + {'data_op': '取得'}, # POST で受ける取得 + {'data_op': '更新(セッション言語のみ・永続DB無し)', 'data_store': '-'}, + {'data_store': '統計 / ES:stats'}, # ES だけ +]) +def test_not_persistent_change_is_not_p1(kw): + assert prioritize.classify(row(**kw), H)[0] != 'P1' + + +def test_view_with_password_check_is_not_p1(): + pri, why = prioritize.classify(row(impl_line='2'), H) + assert pri == 'P5' + assert 'app-credential(password: verify_password)' in why + + +def test_reach_without_proof_does_not_count_for_credential_view(): + # ログイン等は未認証のリクエストがビューまで届くのが正常 + dyn = '[実測] 未認証で到達 | anon=400(到達)' + assert prioritize.classify(row(impl_line='2', dynamic_verified=dyn), H)[0] == 'P5' + + +def test_proven_bypass_still_counts_for_credential_view(): + dyn = '[実測] ★未認証で到達 | anon=200(到達)' + assert prioritize.classify(row(impl_line='2', dynamic_verified=dyn), H)[0] == 'P1' + + +def test_framework_login_endpoint(): + r = row(impl_file='(framework/自動生成)', impl_line='0', endpoint='security.login') + assert prioritize.classify(r, H)[0] == 'P5' From 51972fccb133eceb84b80575ceaa841e694c5c89 Mon Sep 17 00:00:00 2001 From: Masaharu Hayashi Date: Fri, 30 Oct 2026 17:33:57 +0000 Subject: [PATCH 2/2] =?UTF-8?q?feat(api-inventory):=20=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E8=80=85=E5=90=91=E3=81=91=E3=81=AE=20API=20=E3=82=92=E3=80=8C?= =?UTF-8?q?=E5=AF=BE=E8=B1=A1=E5=A4=96=E3=80=8D=E3=81=AB=E3=81=9B=E3=81=9A?= =?UTF-8?q?=E3=83=AD=E3=83=BC=E3=83=AB=E3=81=A7=E5=84=AA=E5=85=88=E5=BA=A6?= =?UTF-8?q?=E3=82=92=E4=BB=98=E3=81=91=E3=82=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 「対象外」を廃止し、admin-access 相当の行も確認対象にする - bola_risk に★がある行は書き込み系を P2、読み取り系を P3 にする - リポジトリ/コミュニティ管理者も使える書き込み系は、管轄の限定を 確かめるため P3 にする - 上記以外(管理者向けの読み取り系・システム管理者のみ)は P5 にする Co-Authored-By: Claude Opus 5.5 --- tools/api-inventory/scripts/README.md | 12 ++++++ tools/api-inventory/scripts/prioritize.py | 43 +++++++++++++++----- tools/api-inventory/tests/test_prioritize.py | 36 +++++++++++++++- 3 files changed, 78 insertions(+), 13 deletions(-) diff --git a/tools/api-inventory/scripts/README.md b/tools/api-inventory/scripts/README.md index 1ca48cafab..29ec4b59f4 100644 --- a/tools/api-inventory/scripts/README.md +++ b/tools/api-inventory/scripts/README.md @@ -1220,6 +1220,18 @@ nginx と SP の設定はリポジトリ同梱の `nginx/weko.conf`・`nginx/shi テストは `python3 -m pytest tools/api-inventory/tests`。 +管理者向け(admin-access 相当)の行も確認対象から外さない(以前の「対象外」は廃止)。 +リポジトリ管理者・コミュニティ管理者は自分の管轄だけを操作できるはずなので、 +`roles` と `bola_risk` で次のように振り分ける。指摘や★実証がある行は P1/P2 の判定が先に拾う。 + +| 条件 | 優先度 | +|---|---| +| `bola_risk` に★(ID 直指定で他リソースを操作できる懸念)の書き込み系 | P2 | +| 同 読み取り系 | P3 | +| リポジトリ/コミュニティ管理者も使える書き込み系(管轄の限定を確かめる) | P3 | +| リポジトリ/コミュニティ管理者も使える読み取り系 | P5 | +| システム管理者のみ | P5 | + 参照系でも、露出内容が **認証情報** または **非公開データの実体** であり認可が 緩い行は P2 に上げる。「読み取り系だから限定的」は露出物次第で成り立たないため。 diff --git a/tools/api-inventory/scripts/prioritize.py b/tools/api-inventory/scripts/prioritize.py index 1f28e46be6..761f28e953 100644 --- a/tools/api-inventory/scripts/prioritize.py +++ b/tools/api-inventory/scripts/prioritize.py @@ -28,8 +28,14 @@ 所有者/ロールの限定が欠けている、または到達可否が未測定。 読み取り系でも、認可が緩く(無認証・ログインのみ・スコープ欠落) 指摘か★実証があり、露出内容が認証情報または非公開データの実体のもの - 対象外 auth が admin-role-table/admin-access 相当または auth_required が - 要(管理)で、指摘も★実証も無い + 管理者向け auth が admin-role-table/admin-access 相当または auth_required が + 要(管理)で、指摘も★実証も無いもの。確認対象から外さず次で振り分ける + bola_risk に★(ID 直指定で他リソースを操作できる懸念) + 書き込み系 → P2、読み取り系 → P3 + リポジトリ/コミュニティ管理者も使える書き込み系 → P3 + (自分の管轄だけを操作できるか確かめる) + リポジトリ/コミュニティ管理者も使える読み取り系 → P5 + システム管理者のみ → P5 P4 意図的な公開設計(static配信・ヘルスチェック・robots.txt・OAI-PMH等)、 または deny_all 常時拒否 P3 読み取り系で、無認証、またはログインのみ/スコープ欠落の指摘がある、 @@ -39,10 +45,10 @@ 破綻が見えない P3 上のいずれにも合致しない(手動確認が必要) - P1/P2 は対象外・P4 より先に評価する。管理画面や公開設計のエンドポイントでも、 + P1/P2 は管理者向け・P4 より先に評価する。管理画面や公開設計のエンドポイントでも、 P1/P2 の条件に合えばそちらが優先される。 -テスト観点による引き上げ(bump、対象外/P4/P5 にのみ適用): +テスト観点による引き上げ(bump、P4/P5 と管理者向けの P3/P5 に適用): test_gap が「特定不能」、または 正常値/異常値/境界値/例外処理 の4観点が 全て欠けている行は P3 に上げる。一部の欠落なら優先度は変えず理由に追記する。 @@ -100,7 +106,7 @@ def load_allow(): 実測欄(dynamic_verified)の粒度はばらついており、同じ機能群でも 「経路なし」と書かれた行と空欄の行が混在する。それを判定に使うと - 同一グループが 整理対象 と 対象外 に割れる。突き合わせで確認済みの + 同一グループが 整理対象 と別の区分に割れる。突き合わせで確認済みの allow リストを正とする。 """ p = data_path('reconcile_allow.json', required=False) @@ -121,6 +127,9 @@ def norm_uri(u): # 「非利用」とみなす根拠。deprecated 列の記述と、実機 url_map に無いこと。 UNUSED_WORDS = ('未使用', '非推奨', '実質未使用', '呼出元なし', '経路なし') +# 自分の管轄(リポジトリ/コミュニティ)だけを操作できるはずの管理者ロール +SCOPED_ADMIN_ROLES = ('Repository Administrator', 'Community Administrator') + # 具体的な権限/所有者チェック機構(P4 の根拠) CONCRETE_AUTHZ = [ 'page_permission_factory', 'check_created_id', 'roles_required', @@ -274,11 +283,23 @@ def bump(pri, why): if body: return 'P2', f'参照系だが露出内容が非公開データの実体({body[0]})で、認可が緩い' - # --- 対象外: admin-access 相当で保護され、指摘も実証も無い --- - if (not has_finding) and (not proven) and ( - 'admin-role-table' in auth or 'admin-access' in auth - or auth_req == '要(管理)'): - return bump('対象外', '認証必須+admin-access 相当の権限チェックあり、指摘なし') + # --- 管理者向け: admin-access 相当で保護され、指摘も実証も無い --- + # 管理者向けでも確認対象から外さない。リポジトリ管理者・コミュニティ管理者は + # 自分の管轄だけを操作できるはずなので、他の管轄に手が届かないかを確かめる。 + is_admin = ('admin-role-table' in auth or 'admin-access' in auth + or auth_req == '要(管理)') + if is_admin and (not has_finding) and (not proven): + roles = field(c, H, 'roles') + scoped_admin = any(r in roles for r in SCOPED_ADMIN_ROLES) + if '★' in field(c, H, 'bola_risk'): + if is_write_method: + return 'P2', f'状態変更系({method}): 管理者向けだが object-level 認可が無く、ID の直接指定で他のリソースを操作できる懸念' + return 'P3', f'読み取り系({method}): 管理者向けだが object-level 認可が無く、ID の直接指定で他のリソースを読める懸念' + if scoped_admin and is_write_method: + return bump('P3', f'状態変更系({method}): リポジトリ/コミュニティ管理者も実行でき、管轄の限定を確認できない') + if scoped_admin: + return bump('P5', '管理者向けの読み取り系(リポジトリ/コミュニティ管理者を含む)') + return bump('P5', 'システム管理者のみ') # --- P4: 意図的な公開設計 / deny_all --- for pat, label in PUBLIC_BY_DESIGN: @@ -389,7 +410,7 @@ def apply_to(path, out=None): return counts -ORDER = ['P1', 'P2', 'P3', 'P4', 'P5', '整理対象', '環境依存', '対象外'] +ORDER = ['P1', 'P2', 'P3', 'P4', 'P5', '整理対象', '環境依存'] def main(): diff --git a/tools/api-inventory/tests/test_prioritize.py b/tools/api-inventory/tests/test_prioritize.py index ad12b26547..1bffb13b6d 100644 --- a/tools/api-inventory/tests/test_prioritize.py +++ b/tools/api-inventory/tests/test_prioritize.py @@ -15,7 +15,8 @@ import prioritize # noqa: E402 COLS = ['method', 'uri', 'auth_required', 'auth_method', 'data_op', 'data_store', - 'dynamic_verified', 'impl_file', 'impl_line', 'endpoint', 'test_gap'] + 'dynamic_verified', 'impl_file', 'impl_line', 'endpoint', 'test_gap', + 'roles', 'bola_risk'] H = {n: i for i, n in enumerate(COLS)} VIEWS = ''' @@ -49,7 +50,7 @@ def row(**kw): 'data_store': '対象 / PostgreSQL:t', 'dynamic_verified': '[実測] anon=400(到達)', 'impl_file': 'pkg/views.py', 'impl_line': '9', 'endpoint': '', - 'test_gap': '-'} + 'test_gap': '-', 'roles': '', 'bola_risk': 'N/A(リソースID無し)'} base.update(kw) return [base[n] for n in COLS] @@ -87,3 +88,34 @@ def test_proven_bypass_still_counts_for_credential_view(): def test_framework_login_endpoint(): r = row(impl_file='(framework/自動生成)', impl_line='0', endpoint='security.login') assert prioritize.classify(r, H)[0] == 'P5' + + +SYS = 'System Administrator' +REPO = 'System Administrator; Repository Administrator' +COM = 'System Administrator; Repository Administrator; Community Administrator' + + +def admin_row(**kw): + base = {'auth_required': '要(管理)', 'auth_method': 'admin-role-table', + 'impl_file': '', 'impl_line': ''} + base.update(kw) + return row(**base) + + +@pytest.mark.parametrize('method, roles, bola, expected', [ + ('POST', REPO, 'N/A(リソースID無し)', 'P3'), # 管轄の限定を確かめる + ('POST', COM, 'N/A(リソースID無し)', 'P3'), + ('GET', COM, 'N/A(リソースID無し)', 'P5'), + ('POST', SYS, 'N/A(リソースID無し)', 'P5'), # システム管理者のみ + ('POST', REPO, '★object-level認可なし', 'P2'), + ('GET', REPO, '★object-level認可なし', 'P3'), + ('GET', SYS, '★object-level認可なし', 'P3'), +]) +def test_admin_rows_are_not_excluded(method, roles, bola, expected): + r = admin_row(method=method, roles=roles, bola_risk=bola) + assert prioritize.classify(r, H)[0] == expected + + +def test_admin_row_with_test_gap_is_raised_to_p3(): + r = admin_row(method='GET', roles=SYS, test_gap='特定不能') + assert prioritize.classify(r, H)[0] == 'P3'