diff --git a/.gitignore b/.gitignore index 8da54a50..2327d594 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,9 @@ projects/* .env.sources.yml .cache/ +# 暗号化した機密の保存先 (PLAN35)。暗号文とはいえリポジトリには載せない。 +secrets/ + # クロスレビュー (ndf:cross-review) の作業生成物 .cross_review/ gh-payload.json diff --git a/etc/_devbase b/etc/_devbase index ebb590bc..74464f9f 100644 --- a/etc/_devbase +++ b/etc/_devbase @@ -105,6 +105,7 @@ _devbase() { 'project:Setup project-specific variables' 'export:Export .env files as an encrypted bundle (age)' 'import:Import .env bundle (age decrypt + merge)' + 'keygen:Generate the devbase age key used by the secret store' ) plugin_subcommands=( @@ -278,6 +279,11 @@ _devbase() { '--backup-dir[Override backup directory]:dir:_files -/' \ '--keep-last[Keep only the last N backup directories]:n:' ;; + keygen) + _arguments \ + '--force[Overwrite an existing key]' \ + '--yes[Skip the confirmation prompt for --force]' '-y[Skip the confirmation prompt for --force]' + ;; *) _describe -t env-commands 'env command' env_subcommands ;; diff --git a/etc/devbase-completion.bash b/etc/devbase-completion.bash index 5bb07fc4..f3f9c180 100644 --- a/etc/devbase-completion.bash +++ b/etc/devbase-completion.bash @@ -35,7 +35,7 @@ _devbase_completions() { # project / container は同じサブコマンド群 (container は非推奨だが補完は維持)。 local project_subcommands="up down ps login logs scale build rebuild list" local container_subcommands="up down ps login logs scale build rebuild" - local env_subcommands="init sync list set get delete edit project export import" + local env_subcommands="init sync list set get delete edit project export import keygen" local plugin_subcommands="list install uninstall update info sync repo" local repo_subcommands="add remove list refresh" local snapshot_subcommands="create list restore copy delete rotate" diff --git a/lib/devbase/cli.py b/lib/devbase/cli.py index 734c7ce1..c36b81be 100644 --- a/lib/devbase/cli.py +++ b/lib/devbase/cli.py @@ -52,7 +52,8 @@ SUBCMD_MAP = { ('project',): ['up', 'down', 'ps', 'login', 'logs', 'scale', 'build', 'rebuild', 'list'], ('container', 'ct'): ['up', 'down', 'ps', 'login', 'logs', 'scale', 'build', 'rebuild'], - ('env',): ['init', 'sync', 'list', 'set', 'get', 'delete', 'edit', 'project', 'export', 'import'], + ('env',): ['init', 'sync', 'list', 'set', 'get', 'delete', 'edit', 'project', 'keygen', + 'export', 'import'], ('plugin', 'pl'): ['list', 'install', 'uninstall', 'update', 'info', 'sync', 'repo', 'migrate'], ('snapshot', 'ss'): ['create', 'list', 'restore', 'copy', 'delete', 'rotate'], } @@ -280,6 +281,23 @@ def _add_env_parser(subparsers): env_sub.add_parser('edit', help='Open .env in editor') env_sub.add_parser('project', help='Setup project-specific variables') + # 生成先を選ぶオプションは置かない。復号側は $DEVBASE_AGE_KEY_FILE か既定パスしか + # 探索しないため、任意のパスへ生成できると「その鍵で保存した機密を復号できない」 + # 状態を作れてしまう。場所を変えたい場合は環境変数を設定してから実行してもらう。 + # 説明中の環境変数名は devbase.env.agekeys.KEY_FILE_ENV と対。agekeys は pyrage を + # 引き込むため、parser 構築時に import せず文字列で持つ (暗号機能を使わない + # コマンドまで pyrage のロード失敗に巻き込まないため)。 + env_keygen = env_sub.add_parser( + 'keygen', + help='Generate the devbase age key used by the secret store ' + '(written to $DEVBASE_AGE_KEY_FILE or ~/.config/devbase/age/keys.txt; ' + 'set $DEVBASE_AGE_KEY_FILE before running to use another location)') + env_keygen.add_argument('--force', action='store_true', + help='Overwrite an existing key (previously encrypted ' + 'secrets may become unrecoverable)') + env_keygen.add_argument('--yes', '-y', action='store_true', dest='assume_yes', + help='Skip the confirmation prompt for --force') + _add_env_export_parser(env_sub) _add_env_import_parser(env_sub) diff --git a/lib/devbase/commands/env.py b/lib/devbase/commands/env.py index 9d5c0393..40e12dca 100644 --- a/lib/devbase/commands/env.py +++ b/lib/devbase/commands/env.py @@ -35,6 +35,9 @@ def cmd_env(devbase_root: Path, args) -> int: 'project': lambda: cmd_env_project(devbase_root), 'export': lambda: cmd_env_export(devbase_root, args), 'import': lambda: cmd_env_import(devbase_root, args), + 'keygen': lambda: cmd_env_keygen(devbase_root, + force=getattr(args, 'force', False), + assume_yes=getattr(args, 'assume_yes', False)), } handler = handlers.get(subcmd) @@ -458,6 +461,124 @@ def cmd_env_import(devbase_root: Path, args) -> int: return import_bundle(devbase_root, opts) +def _has_encrypted_secrets(devbase_root: Path) -> bool: + """暗号化済みの機密が 1 つでも存在するか""" + from devbase.env.secret_store import SecretStore, SecretRef + + store = SecretStore(devbase_root) + if store.age.exists(SecretRef.for_global()): + return True + return bool(store.project_names()) + + +def _print_key_backup_notice(path, public: str) -> None: + print() + print("=" * 60) + print("鍵のバックアップを必ず取ってください") + print("=" * 60) + print(f" 鍵ファイル: {path}") + print(f" 公開鍵 : {public}") + print() + print(" この鍵を失うと、暗号化した機密は誰にも復号できません。") + print(" パスワード管理ツールなど、端末とは別の場所へ複製を保管してください。") + print("=" * 60) + + +def cmd_env_keygen(devbase_root: Path, force: bool = False, + assume_yes: bool = False) -> int: + """devbase 専用の age 鍵を生成する + + 生成先は必ず ``agekeys.key_file_path()`` (= ``DEVBASE_AGE_KEY_FILE`` があれば + それ、無ければ ``~/.config/devbase/age/keys.txt``) にする。生成先を CLI 引数で + 自由に選べるようにすると、復号側の ``agekeys.resolve_identities()`` はそのパスを + 探索しないため「生成した鍵で保存した機密を復号できない」状態を作れてしまう。 + 場所を変えたい場合は ``DEVBASE_AGE_KEY_FILE`` を設定してから実行してもらい、 + 生成先と探索先が構造的に一致する契約を保つ。 + """ + from devbase.env import agekeys + from devbase.errors import DevbaseError + + path = agekeys.key_file_path() + + if path.exists() and not force: + try: + public = agekeys.read_public_key(path) + except DevbaseError as e: + logger.error("%s", e) + return 1 + print(f"鍵は既に存在します: {path}") + print(f" 公開鍵: {public}") + print(" 作り直す場合: devbase env keygen --force") + return 0 + + # keygen はワークスペース固有の受信者リスト (secrets/recipients.txt) を触らない。 + # 鍵はグローバル (~/.config/devbase/age/keys.txt) なのに受信者リストは + # ワークスペースごとに存在するため、ここで書き込むと別ワークスペースには旧公開鍵が + # 取り残され、既に失われた秘密鍵に対応する公開鍵で暗号化してしまう。 + # agekeys.resolve_recipients() は recipients.txt が無ければ鍵ファイルの公開鍵へ + # フォールバックするので、単独利用ではリストを作る必要がない。チーム運用で明示的に + # 受信者を足す経路 (rekey) だけが recipients.txt を作る。 + # + # 書き込みの原子性は agekeys.generate_key_file → + # io_common.write_secure_bytes_atomic (一時ファイル + fsync + os.replace) が + # 担保しており、生成が途中で失敗しても既存の鍵ファイルは元のまま残る。 + # したがってこの層で「内容をメモリへ退避して書き戻す」手動ロールバックは重ねない。 + # 重ねてもリストア自体が失敗しうるぶん壊れ方の種類が増えるだけで、守れるものが + # 増えないため。将来ここへロールバックを足したくなったら、まず io 層の原子性が + # 破れていないかを疑うこと。 + # + # 一方で「既存鍵が在るのに読めない」ときに中止するガードは、原子性とは別の目的で + # 残す。読めないだけなら権限を直せば回収できる可能性があるのに、生成が成功すると + # 旧鍵は上書きで確実に消えるため。判定は確認プロンプトより前に置き、 + # 「同意させてから中止する」空振りを避ける。 + if path.exists() and not os.access(path, os.R_OK): + logger.error( + "既存の鍵ファイルを読めないため、上書きを中止しました: %s", path) + logger.error( + "権限を確認するか、不要と判断できる場合は手動で退避してから" + "再実行してください") + return 1 + + # ここへ来るのは「鍵が無い」か「--force で作り直す」場合だけ。後者は既存鍵を + # 捨てる操作なので、常に明示的な同意を取る。 + # + # 鍵は ~/.config/devbase/age/keys.txt = 全ワークスペース共通のグローバル資産 + # なのに対し、暗号化された機密はワークスペースごとに散らばっている。同意の要否を + # カレントの DEVBASE_ROOT に機密があるか (_has_encrypted_secrets) で決めると、 + # まだ機密の無い別プロジェクトで --force した瞬間に無警告で鍵が消え、他プロジェクトの + # 機密が復旧不能になる。カレントの状況は「文言をどれだけ強くするか」にだけ使う。 + if path.exists() and not assume_yes: + print("鍵ファイルを作り直します。この鍵は全プロジェクト共通です。") + print(f" 鍵ファイル: {path}") + if _has_encrypted_secrets(devbase_root): + print(" このワークスペースには暗号化済みの機密があり、" + "旧鍵でしか復号できないものは失われます。") + print(" 他のワークスペースで暗号化した機密も、" + "旧鍵を失うと復号できなくなります。") + print(" 続行前に旧鍵のバックアップがあるか確認してください。") + answer = safe_input("続行しますか? (yes と入力): ") + if answer != 'yes': + print("中止しました") + return 1 + + # force はコマンドの --force をそのまま渡す。ここで無条件に force=True に + # すると、上の path.exists() 判定から実際の書き込みまでの隙間に他プロセスが + # 鍵を作っていた場合、利用者が上書きを要求していないのにその鍵を消してしまう + # (TOCTOU)。force=False なら agekeys 側が O_CREAT|O_EXCL で作るため、隙間に + # 現れた鍵は上書きされずエラーで止まる。 + try: + path, public = agekeys.generate_key_file(path, force=force) + except (DevbaseError, OSError) as e: + # 新規生成は排他作成、--force の差し替えは atomic なので、いずれの失敗でも + # 既に在る鍵はそのまま残っている。 + logger.error("%s", e) + return 1 + + logger.info("鍵を生成しました: %s", path) + _print_key_backup_notice(path, public) + return 0 + + def _update_source_metadata(devbase_root: Path, env_file: EnvFile) -> None: """ソースメタデータを更新する""" sources = SourcesManager(devbase_root) diff --git a/lib/devbase/env/agekeys.py b/lib/devbase/env/agekeys.py new file mode 100644 index 00000000..0728bcf7 --- /dev/null +++ b/lib/devbase/env/agekeys.py @@ -0,0 +1,379 @@ +"""devbase 専用 age 鍵と受信者リストの管理 + +``devbase env export`` / ``import`` が使う ``~/.ssh`` の鍵とは別に、devbase が +機密の保存に使う専用鍵を扱う。署名用の SSH 鍵とは失効・保管・バックアップの +扱いが異なるため、鍵を分けて管理する (plan35 §5.1)。 + +鍵ファイルの場所は ``DEVBASE_AGE_KEY_FILE`` で上書きでき、既定は +``~/.config/devbase/age/keys.txt`` (``XDG_CONFIG_HOME`` があればそれを尊重)。 +""" + +from __future__ import annotations + +import os +import stat +from datetime import datetime, timezone +from pathlib import Path +from typing import List, Optional, Tuple + +import pyrage + +from devbase.env import cipher as _cipher +from devbase.env import io_common as _io_common +from devbase.errors import DevbaseError +from devbase.log import get_logger + +logger = get_logger(__name__) + + +class AgeKeyError(DevbaseError): + """鍵ファイル / 受信者リストの操作エラー""" + + +#: 鍵ファイルの場所を明示するための環境変数。OS ごとの既定位置の違いを +#: 利用者が 1 箇所で吸収できるようにする (plan35 §5.1)。 +KEY_FILE_ENV = 'DEVBASE_AGE_KEY_FILE' + +#: 受信者リストのファイル名 (``$DEVBASE_ROOT/secrets/`` 配下)。 +RECIPIENTS_FILENAME = 'recipients.txt' + +_AGE_SECRET_PREFIX = 'AGE-SECRET-KEY-1' + + +# --------------------------------------------------------------------------- +# パス解決 +# --------------------------------------------------------------------------- + +def default_key_dir() -> Path: + """既定の鍵ディレクトリ ``$XDG_CONFIG_HOME/devbase/age`` を返す。 + + ``XDG_CONFIG_HOME`` が未設定なら ``~/.config`` を使う。 + """ + base = os.environ.get('XDG_CONFIG_HOME') + root = Path(base).expanduser() if base else Path.home() / '.config' + return root / 'devbase' / 'age' + + +def key_file_path() -> Path: + """使用する鍵ファイルのパス (環境変数の指定を優先)""" + override = os.environ.get(KEY_FILE_ENV) + if override: + return Path(override).expanduser() + return default_key_dir() / 'keys.txt' + + +def recipients_file(devbase_root: Path) -> Path: + """受信者リストのパス ``$DEVBASE_ROOT/secrets/recipients.txt``""" + return Path(devbase_root) / 'secrets' / RECIPIENTS_FILENAME + + +# --------------------------------------------------------------------------- +# 鍵の生成・読み取り +# --------------------------------------------------------------------------- + +def _ensure_private_dir(path: Path) -> None: + """ディレクトリを用意し、**自分が新規作成した階層だけ** ``0700`` にする。 + + 既存ディレクトリまで chmod すると、``DEVBASE_AGE_KEY_FILE=/tmp/devbase-key`` + のように共有ディレクトリを鍵の置き場に指定されたとき、その共有ディレクトリ + ごと他ユーザーやサービスから読めなくしてしまう。devbase が作っていない + ディレクトリの権限はその所有者の管轄なので触らず、緩い場合は警告に留める。 + + ``mkdir(parents=True)`` で一括作成してから chmod すると、作成から chmod まで + の間だけ umask 依存の緩い権限 (例 0755) が見えてしまう。その隙に開いた fd は + 後から chmod しても閉じないため、**作成前に** 未存在の階層を控えておき、親→子 + の順に ``mkdir(mode=0o700)`` で 1 階層ずつ作る。こうすれば最初から 0700 で、 + 緩い権限が一瞬も露出しない。 + + ``mode`` は umask でビットが削られることはあっても広がることはなく、``0o700`` + には group / other ビットが無いので umask の影響を受けない。「umask で緩く + なるのでは」と後追いの chmod を足す必要は無い。 + """ + path = Path(path) + missing: List[Path] = [] + probe = path + while not probe.exists(): + missing.append(probe) + parent = probe.parent + if parent == probe: # ルートまで到達 (通常は起こらない) + break + probe = parent + + if not missing: + _warn_if_world_accessible(path) + return + + # missing は子→親の順に積んであるので、逆順 (親→子) に作る + for target in reversed(missing): + try: + target.mkdir(mode=0o700) + except FileExistsError: + # 並行して他プロセスが先に作った場合。既存ディレクトリは + # 所有者の管轄として権限を触らない方針に合わせ、chmod しない。 + continue + + +def _warn_if_world_accessible(path: Path) -> None: + """既存ディレクトリの権限が緩ければ警告する (権限は変更しない)""" + try: + mode = stat.S_IMODE(path.stat().st_mode) + except OSError: + return + if mode & 0o077: + logger.warning( + "%s は他ユーザーからアクセスできます (mode %04o)。" + "devbase が作成したディレクトリではないため権限は変更しません。" + "機密を置く場所なら chmod 700 を検討してください", + path, mode, + ) + + +def _key_exists_error(path: Path) -> AgeKeyError: + """「既に鍵がある」エラー。事前チェックと排他生成の両方から使う""" + return AgeKeyError( + f"鍵ファイルが既に存在します: {path}\n" + "上書きすると既存の暗号化ファイルを復号できなくなります。" + "意図的に作り直す場合のみ --force を指定してください" + ) + + +def _create_key_file_exclusive(path: Path, data: bytes) -> None: + """新規鍵を ``O_CREAT|O_EXCL`` で **排他的に** 作成する。 + + 「存在チェック → 生成」を別々に行うと、その隙間に他プロセスが同じ判定を + 通り抜けられる。両者が生成へ進むと後発の書き込みが先発の鍵を消し、先発鍵で + 暗号化した機密がその瞬間から復号不能になる (TOCTOU)。``O_EXCL`` は + 「存在しなければ作る」をカーネル側で不可分に行うため、この隙間が原理的に + 消える。既存ファイルが無い状況では守るべき旧内容も無いので、一時ファイル + + ``os.replace`` は不要なだけでなく有害 — ``os.replace`` は既存を無条件に + 置き換えてしまい、まさに塞ぎたい上書きを許すため。 + + 書き込み途中で失敗したら、中途半端な鍵ファイルを残さないよう自分で作った + ファイルを消す。半端な鍵が残ると以後の生成が「既に存在します」で止まり、 + しかもその鍵では何も復号できない。 + """ + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError as e: + raise _key_exists_error(path) from e + try: + with os.fdopen(fd, 'wb') as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + except BaseException: + try: + os.close(fd) + except OSError: + pass + try: + path.unlink() + except OSError: + pass + raise + # mode 引数が無視される環境 (Windows 等) に備えて明示的に揃える + try: + os.chmod(path, 0o600) + except OSError: + pass + + +def generate_key_file(path: Optional[Path] = None, *, + force: bool = False) -> Tuple[Path, str]: + """devbase 専用の age 鍵を生成して ``0600`` で保存する。 + + 書き込み方法は ``force`` で変える。守るべき旧内容の有無が違うため。 + + - ``force=False`` (新規生成): ``O_CREAT|O_EXCL`` で直接排他作成する。 + 判定と作成の隙間を閉じ、並行実行しても先に作った側の鍵が生き残る。 + - ``force=True`` (作り直し): 同一ディレクトリの一時ファイルへ書いて fsync + してから atomic に差し替える。直接 ``O_TRUNC`` で上書きすると、書き込み + 途中の失敗 (ディスク枯渇・強制終了など) で旧鍵だけが失われ、既存の暗号文を + 誰も復号できなくなるため。差し替えに成功するまで旧鍵はそのまま残る。 + + ``--force`` 同士の並行実行にはロックを掛けない。どちらも利用者が「既存鍵を + 捨てて作り直す」と明示的に要求した操作であり、後勝ちで最後の鍵が残ること自体が + 要求どおりの結果だから。ロックで直列化しても「先の鍵が消える」事実は変わらず、 + グローバルな鍵ファイルにロックの残骸 (stale lock) という別の詰まり方を持ち込む + ぶん損になる。塞ぐべきだったのは「誰も上書きを要求していないのに上書きされる」 + 新規生成側だけで、そこは ``O_EXCL`` で閉じている。 + + Returns: + ``(鍵ファイルのパス, 公開鍵文字列)`` + + Raises: + AgeKeyError: 既存の鍵があり ``force`` が偽のとき + """ + path = Path(path) if path is not None else key_file_path() + # 早期に弾いて無駄な鍵生成を避けるための事前チェック。ここを通り抜けた + # 並行プロセスは下の O_EXCL で確実に止まるので、この判定は最適化にすぎない。 + if path.exists() and not force: + raise _key_exists_error(path) + + identity = pyrage.x25519.Identity.generate() + public = str(identity.to_public()) + created = datetime.now(timezone.utc).isoformat(timespec='seconds') + content = ( + "# devbase age key file\n" + f"# created: {created}\n" + f"# public key: {public}\n" + "# この鍵を失うと暗号化した機密は復旧できません。\n" + "# パスワード管理ツール等へ必ず複製を保管してください。\n" + f"{identity}\n" + ) + + _ensure_private_dir(path.parent) + if force: + _io_common.write_secure_bytes_atomic(path, content.encode('utf-8')) + else: + _create_key_file_exclusive(path, content.encode('utf-8')) + return path, public + + +def read_public_key(path: Optional[Path] = None) -> str: + """鍵ファイルから公開鍵を導出する。 + + ファイル中のコメント (``# public key:``) は信頼せず、秘密鍵行から都度導出する。 + コメントは手で書き換えられうるため、そこを信じると「登録した受信者と実際の + 鍵が食い違ったまま暗号化してしまう」事故が起きる。 + """ + path = Path(path) if path is not None else key_file_path() + if not path.exists(): + raise AgeKeyError( + f"鍵ファイルが見つかりません: {path}\n" + "`devbase env keygen` で生成してください" + ) + try: + text = path.read_text(encoding='utf-8') + except (OSError, UnicodeDecodeError) as e: + raise AgeKeyError(f"鍵ファイルを読み込めませんでした ({path}): {e}") from e + + for line in text.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith('#'): + continue + if not stripped.startswith(_AGE_SECRET_PREFIX): + break + try: + return str(pyrage.x25519.Identity.from_str(stripped).to_public()) + except Exception as e: + raise AgeKeyError(f"age 秘密鍵の解釈に失敗しました ({path}): {e}") from e + + raise AgeKeyError( + f"age 秘密鍵 ({_AGE_SECRET_PREFIX}...) が含まれていません: {path}\n" + "OpenSSH 鍵など age 形式以外を使う場合は、対応する公開鍵を " + "`devbase env keygen` ではなく受信者リストへ直接登録してください" + ) + + +# --------------------------------------------------------------------------- +# 受信者リスト +# --------------------------------------------------------------------------- + +def load_recipients(devbase_root: Path) -> List[str]: + """受信者リストの有効行 (コメント・空行を除く) を返す""" + path = recipients_file(devbase_root) + if not path.exists(): + return [] + try: + text = path.read_text(encoding='utf-8') + except (OSError, UnicodeDecodeError) as e: + raise AgeKeyError(f"受信者リストを読み込めませんでした ({path}): {e}") from e + return [ + line.strip() for line in text.splitlines() + if line.strip() and not line.strip().startswith('#') + ] + + +def save_recipients(devbase_root: Path, recipients: List[str]) -> Path: + """受信者リストを書き出す。 + + 公開鍵そのものは秘密ではないが、ファイルは ``0600`` で保護する。第三者が + 自分の公開鍵をここへ追記できると、以後の暗号化がその相手にも復号可能に + なるため、機密性ではなく**改竄防止**のために権限を絞る。 + + 書き込みは鍵ファイルと同じく atomic に行う。途中失敗で受信者が欠けたリストが + 残ると、以後の暗号化から一部の受信者が黙って外れてしまうため。 + + 置き場 (``secrets/``) の扱いも鍵ファイルと揃えて ``_ensure_private_dir`` に + 任せる。既に存在する ``secrets/`` — 例えば git clone 直後の 0755 — を勝手に + 0700 へ落とすと、ワークスペースを共有している他ユーザーの参照を壊すため。 + """ + path = recipients_file(devbase_root) + _ensure_private_dir(path.parent) + header = ( + "# devbase secret store recipients\n" + "# 1 行に 1 つの公開鍵 (age1... / ssh-ed25519 ... / ssh-rsa ...)。\n" + "# ここに列挙した全員が機密を復号できる。\n" + "# 編集しても既存の暗号化ファイルは変わらないため、変更後は再暗号化すること。\n" + ) + body = ''.join(f"{r}\n" for r in recipients) + _io_common.write_secure_bytes_atomic(path, (header + body).encode('utf-8')) + return path + + +def add_recipient(devbase_root: Path, spec: str) -> bool: + """受信者を追加する。既に登録済みなら ``False`` を返して何もしない。""" + spec = spec.strip() + if not spec: + raise AgeKeyError("受信者が空です") + # 形式不正をここで弾いておく。登録後に初めて暗号化で落ちるより早い。 + _cipher.validate_recipient(spec) + + current = load_recipients(devbase_root) + if spec in current: + return False + save_recipients(devbase_root, current + [spec]) + return True + + +def remove_recipient(devbase_root: Path, spec: str) -> bool: + """受信者を削除する。登録が無ければ ``False`` を返す。""" + spec = spec.strip() + current = load_recipients(devbase_root) + if spec not in current: + return False + save_recipients(devbase_root, [r for r in current if r != spec]) + return True + + +# --------------------------------------------------------------------------- +# 暗号化・復号に渡す鍵の解決 +# --------------------------------------------------------------------------- + +def resolve_recipients(devbase_root: Path) -> List[str]: + """暗号化に使う受信者を解決する。 + + 受信者リストに登録があればそれを使い、無ければ専用鍵の公開鍵を使う。 + どちらも無ければ ``AgeKeyError``。 + """ + registered = load_recipients(devbase_root) + if registered: + return registered + + key_file = key_file_path() + if key_file.exists(): + return [read_public_key(key_file)] + + raise AgeKeyError( + "暗号化に使う公開鍵がありません。\n" + " `devbase env keygen` で devbase 専用鍵を生成するか、\n" + f" {recipients_file(devbase_root)} へ公開鍵を登録してください" + ) + + +def resolve_identities() -> List[str]: + """復号に使う秘密鍵の候補を返す。 + + 専用鍵を先頭に置き、続けて ``~/.ssh`` の既定鍵を候補に加える。``pyrage`` は + 複数 identity を受け取り一致したものだけを使うため、旧来 ``~/.ssh`` の鍵で + 暗号化したファイルも移行期間中そのまま復号できる。 + """ + found: List[str] = [] + key_file = key_file_path() + if key_file.exists(): + found.append(str(key_file)) + for path in _cipher.default_identity_paths(): + if path.exists() and str(path) not in found: + found.append(str(path)) + return found diff --git a/lib/devbase/env/cipher.py b/lib/devbase/env/cipher.py index e3c3f812..1cdd2e0f 100644 --- a/lib/devbase/env/cipher.py +++ b/lib/devbase/env/cipher.py @@ -8,6 +8,9 @@ import pyrage from devbase.errors import DevbaseError +from devbase.log import get_logger + +logger = get_logger(__name__) class CipherError(DevbaseError): @@ -150,6 +153,14 @@ def _resolve_identity(path_spec: str): ) from e +def validate_recipient(spec: str) -> None: + """recipient 仕様文字列が解釈可能かを検証する (不正なら CipherError)。 + + 受信者リストへの登録時など、実際に暗号化する前に形式不正を弾くために使う。 + """ + _resolve_recipient(spec) + + def encrypt(data: bytes, recipients: Sequence[str] = (), passphrase: Optional[str] = None) -> bytes: @@ -198,7 +209,36 @@ def decrypt(data: bytes, if not identities: raise CipherError("identity または passphrase を指定してください") - resolved = [_resolve_identity(p) for p in identities] + # identities は「devbase 専用鍵 → ~/.ssh の既定鍵」のように複数候補を並べて + # 渡される (agekeys.resolve_identities)。ここで 1 つでも解決に失敗した時点で + # 例外にすると、壊れた / 読めない鍵ファイルが 1 つ混ざっているだけで後続の + # 有効な鍵を試せず、旧来 ~/.ssh の鍵で暗号化した暗号文を移行期間中に復号 + # できるという意図が壊れる。そこで候補ごとに解決を試し、失敗した候補は + # 理由を warning に残したうえで読み飛ばす。黙って捨てると「鍵を指定したのに + # 復号できない」原因を利用者が追えなくなるため、ログは必須。 + resolved = [] + failures: List[str] = [] + for spec in identities: + try: + resolved.append(_resolve_identity(spec)) + except Exception as e: + # 想定外の例外もここで握り潰さず失敗理由として蓄積する。全滅時には + # 下で CipherError に含めて送出するので、情報は失われない。 + failures.append(f"{spec}: {e}") + logger.warning( + "identity を解決できなかったため復号候補から除外します (%s): %s", + spec, e, + ) + + # 全候補が解決できなかったときだけ失敗させる。どの候補がなぜ駄目だったかを + # 並べて示し、鍵の置き場所・権限・形式のどれが原因かを切り分けられるようにする。 + if not resolved: + detail = '\n'.join(f" - {f}" for f in failures) + raise CipherError( + "復号に使える identity がありません " + f"(候補 {len(failures)} 件をいずれも解決できませんでした):\n{detail}" + ) + try: return pyrage.decrypt(data, resolved) except Exception as e: diff --git a/lib/devbase/env/io_common.py b/lib/devbase/env/io_common.py index a0b27daa..b6e16a35 100644 --- a/lib/devbase/env/io_common.py +++ b/lib/devbase/env/io_common.py @@ -10,6 +10,7 @@ import getpass import os import sys +import tempfile from pathlib import Path from typing import List, Optional, Sequence, Type @@ -120,3 +121,62 @@ def write_secure_bytes(path: Path, data: bytes, *, mode: int = 0o600) -> None: os.chmod(path, mode) except OSError: pass + + +def _fsync_dir(directory: Path) -> None: + """ディレクトリエントリを fsync する (対応しない環境では黙って諦める)。 + + ``os.replace`` 自体は atomic でも、rename の記録がディスクへ届く前に電源断 + すると差し替えが失われうる。ディレクトリを fsync して rename を永続化する。 + Windows などディレクトリを開けない環境では何もしない。 + """ + try: + fd = os.open(str(directory), os.O_RDONLY) + except OSError: + return + try: + os.fsync(fd) + except OSError: + pass + finally: + os.close(fd) + + +def write_secure_bytes_atomic(path: Path, data: bytes, *, mode: int = 0o600) -> None: + """``path`` の中身を ``data`` へ **atomic に** 差し替える (``mode`` を強制)。 + + ``write_secure_bytes`` は既存ファイルを ``O_TRUNC`` で直接上書きするため、 + ディスク枯渇やプロセス中断が起きると「旧内容は消えたが新内容も揃っていない」 + 中途半端なファイルが残る。age 鍵のように失うと復旧不能なファイルでは、 + + - 同一ディレクトリの一時ファイルへ ``0600`` で書く (別 FS だと rename が + atomic にならないため、必ず同じディレクトリに作る) + - ``fsync`` して中身をディスクへ確定させる + - ``os.replace`` で差し替え、ディレクトリも ``fsync`` する + + という順序にして、途中のどこで失敗しても旧内容がそのまま残るようにする。 + 失敗時は一時ファイルを掃除してから例外を送出する。 + """ + path.parent.mkdir(parents=True, exist_ok=True) + # mkstemp は 0600 で作成するため、作成時点から権限が広がらない。 + fd, tmp_name = tempfile.mkstemp( + prefix=f'.{path.name}.', suffix='.tmp', dir=str(path.parent)) + tmp = Path(tmp_name) + try: + with os.fdopen(fd, 'wb') as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + # mkstemp の mode が無視される環境 (Windows 等) に備えて明示的に揃える + try: + os.chmod(tmp, mode) + except OSError: + pass + os.replace(tmp, path) + except BaseException: + try: + tmp.unlink() + except OSError: + pass + raise + _fsync_dir(path.parent) diff --git a/lib/devbase/env/secret_store.py b/lib/devbase/env/secret_store.py new file mode 100644 index 00000000..7de534e8 --- /dev/null +++ b/lib/devbase/env/secret_store.py @@ -0,0 +1,320 @@ +"""機密の保存先を抽象化する層 (平文 / age) + +``devbase`` が扱う機密は、これまで平文の ``.env`` に直接置かれていた。本モジュールは +「どこに」「どの形式で」保存するかを 1 箇所に閉じ込め、上位の設定操作コマンドからは +``load`` / ``save`` だけを見えるようにする (plan35 §3.1)。 + +保存先の対応: + +=================== ================================== ========================================== +参照 平文 (従来) age (暗号化) +=================== ================================== ========================================== +共通 ``$DEVBASE_ROOT/.env`` ``$DEVBASE_ROOT/secrets/global.env.age`` +プロジェクト ``projects//.env`` ``secrets/projects/.env.age`` +=================== ================================== ========================================== + +どちらを使うかは**ファイルの存在で自動判定**する。暗号化ファイルがあればそれを使い、 +無ければ平文を使う。同じ参照に対して両方が存在する状態は、どちらが正なのか判断できない +ため明示的なエラーにして利用者に解消させる (plan35 §9)。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Dict, List, Optional, Protocol, Sequence + +from devbase.env import agekeys +from devbase.env import cipher as _cipher +from devbase.env import io_common as _io_common +from devbase.env.store import EnvFile +from devbase.errors import DevbaseError +from devbase.log import get_logger + +logger = get_logger(__name__) + + +class SecretStoreError(DevbaseError): + """秘密ストアの操作エラー""" + + +#: 暗号化された機密を置くディレクトリ名 (``$DEVBASE_ROOT`` 相対) +SECRETS_DIRNAME = 'secrets' + +GLOBAL_ENCRYPTED_FILENAME = 'global.env.age' + +MODE_AGE = 'age' +MODE_PLAINTEXT = 'plaintext' +MODE_ABSENT = 'absent' + + +def _validate_project_name(name: str) -> str: + """プロジェクト名がパスを跨がないことを確認する。 + + 参照名はそのままファイル名に使われるため、``..`` や区切り文字を許すと + ``secrets/`` の外側へ書き出せてしまう。 + """ + if not name: + raise SecretStoreError("プロジェクト名が空です") + if name != Path(name).name or name in ('.', '..'): + raise SecretStoreError( + f"プロジェクト名にパス区切りは使えません: {name!r}" + ) + return name + + +@dataclass(frozen=True) +class SecretRef: + """機密の参照 (共通 / プロジェクト)""" + kind: str # 'global' | 'project' + name: Optional[str] = None + + @staticmethod + def for_global() -> 'SecretRef': + return SecretRef(kind='global') + + @staticmethod + def for_project(name: str) -> 'SecretRef': + return SecretRef(kind='project', name=_validate_project_name(name)) + + def label(self) -> str: + return 'グローバル' if self.kind == 'global' else f"プロジェクト '{self.name}'" + + +class SecretBackend(Protocol): + name: str + + def path(self, ref: SecretRef) -> Path: ... + def exists(self, ref: SecretRef) -> bool: ... + def load(self, ref: SecretRef) -> Dict[str, str]: ... + def save(self, ref: SecretRef, data: Dict[str, str]) -> Path: ... + def remove(self, ref: SecretRef) -> bool: ... + + +class PlaintextBackend: + """従来どおり平文の ``.env`` を読み書きする""" + + name = MODE_PLAINTEXT + + def __init__(self, devbase_root: Path): + self._root = Path(devbase_root) + + def path(self, ref: SecretRef) -> Path: + if ref.kind == 'global': + return self._root / '.env' + return self._root / 'projects' / _validate_project_name(ref.name or '') / '.env' + + def exists(self, ref: SecretRef) -> bool: + return self.path(ref).is_file() + + def load(self, ref: SecretRef) -> Dict[str, str]: + path = self.path(ref) + if not path.is_file(): + return {} + try: + return EnvFile.parse_bytes(path.read_bytes()) + except OSError as e: + raise SecretStoreError(f"読み込みに失敗しました ({path}): {e}") from e + except UnicodeDecodeError as e: + raise SecretStoreError( + f"{path} を UTF-8 として読めませんでした: {e}\n" + "暗号化済みファイルを平文として読もうとしていないか確認してください" + ) from e + + def save(self, ref: SecretRef, data: Dict[str, str]) -> Path: + path = self.path(ref) + try: + # 平文とはいえ機密の入れ物なので、暗号化側と同じく atomic に差し替える。 + # 直接 O_TRUNC すると書き込み途中の失敗で旧値も新値も失った空ファイルが + # 残り、その状態で暗号化すると中身の無い機密を保存してしまう。 + _io_common.write_secure_bytes_atomic(path, EnvFile.dump_bytes(data)) + except OSError as e: + raise SecretStoreError(f"書き込みに失敗しました ({path}): {e}") from e + return path + + def remove(self, ref: SecretRef) -> bool: + path = self.path(ref) + if not path.exists(): + return False + try: + path.unlink() + except OSError as e: + raise SecretStoreError(f"削除に失敗しました ({path}): {e}") from e + return True + + +class AgeBackend: + """age で暗号化したファイルを読み書きする""" + + name = MODE_AGE + + def __init__(self, devbase_root: Path, *, + recipients: Optional[Sequence[str]] = None, + identities: Optional[Sequence[str]] = None): + self._root = Path(devbase_root) + self._recipients = list(recipients) if recipients is not None else None + self._identities = list(identities) if identities is not None else None + + # -- 鍵の解決 ----------------------------------------------------------- + + def recipients(self) -> List[str]: + if self._recipients is not None: + return self._recipients + return agekeys.resolve_recipients(self._root) + + def identities(self) -> List[str]: + if self._identities is not None: + return self._identities + found = agekeys.resolve_identities() + if not found: + raise SecretStoreError( + "復号に使える秘密鍵が見つかりません。\n" + f" `devbase env keygen` で生成するか、{agekeys.KEY_FILE_ENV} " + "で鍵ファイルの場所を指定してください" + ) + return found + + # -- 保存先 ------------------------------------------------------------- + + def path(self, ref: SecretRef) -> Path: + base = self._root / SECRETS_DIRNAME + if ref.kind == 'global': + return base / GLOBAL_ENCRYPTED_FILENAME + name = _validate_project_name(ref.name or '') + return base / 'projects' / f'{name}.env.age' + + def exists(self, ref: SecretRef) -> bool: + return self.path(ref).is_file() + + # -- 読み書き ----------------------------------------------------------- + + def load(self, ref: SecretRef) -> Dict[str, str]: + path = self.path(ref) + if not path.is_file(): + return {} + try: + blob = path.read_bytes() + except OSError as e: + raise SecretStoreError(f"読み込みに失敗しました ({path}): {e}") from e + try: + plain = _cipher.decrypt(blob, identities=self.identities()) + except _cipher.CipherError as e: + raise SecretStoreError( + f"{ref.label()}の機密を復号できませんでした ({path}): {e}" + ) from e + try: + return EnvFile.parse_bytes(plain) + except UnicodeDecodeError as e: + # 復号は成功したのに中身が UTF-8 でない = 元々 .env ではない + # バイナリを暗号化していた、というケース。PlaintextBackend.load と + # 同じく SecretStoreError へ包み、呼び出し側が扱う例外を 1 種類に保つ。 + raise SecretStoreError( + f"{ref.label()}の機密を復号しましたが、UTF-8 として読めませんでした " + f"({path}): {e}\n" + "KEY=VALUE 形式以外のファイルを暗号化していないか確認してください" + ) from e + + def save(self, ref: SecretRef, data: Dict[str, str]) -> Path: + path = self.path(ref) + try: + blob = _cipher.encrypt(EnvFile.dump_bytes(data), + recipients=self.recipients()) + except _cipher.CipherError as e: + raise SecretStoreError( + f"{ref.label()}の機密を暗号化できませんでした: {e}" + ) from e + try: + # 暗号文は失うと復旧不能なので、既存ファイルを直接 O_TRUNC せず + # 一時ファイル → fsync → os.replace で差し替える。ディスク枯渇や中断が + # 起きても旧 ciphertext はそのまま残り、書きかけの一時ファイルも消える。 + _io_common.write_secure_bytes_atomic(path, blob) + except OSError as e: + raise SecretStoreError(f"書き込みに失敗しました ({path}): {e}") from e + return path + + def remove(self, ref: SecretRef) -> bool: + path = self.path(ref) + if not path.exists(): + return False + try: + path.unlink() + except OSError as e: + raise SecretStoreError(f"削除に失敗しました ({path}): {e}") from e + return True + + +class SecretStore: + """保存先を自動判定して機密を読み書きする窓口""" + + def __init__(self, devbase_root: Path, *, + recipients: Optional[Sequence[str]] = None, + identities: Optional[Sequence[str]] = None): + self.root = Path(devbase_root) + self.plaintext = PlaintextBackend(self.root) + self.age = AgeBackend(self.root, recipients=recipients, + identities=identities) + + # -- 判定 --------------------------------------------------------------- + + def backend_for(self, ref: SecretRef) -> SecretBackend: + """参照に対して使うべき backend を返す。 + + 暗号化ファイルと平文ファイルが同時に存在する場合は、どちらが最新なのか + devbase 側では判断できない。黙って一方を採用すると「編集したはずの値が + 反映されない」形で事故になるため、明示的に停止して利用者に解消させる。 + """ + age_exists = self.age.exists(ref) + plain_exists = self.plaintext.exists(ref) + if age_exists and plain_exists: + raise SecretStoreError( + f"{ref.label()}の機密が暗号化・平文の両方に存在します:\n" + f" 暗号化: {self.age.path(ref)}\n" + f" 平文: {self.plaintext.path(ref)}\n" + "どちらが正しいか判断できないため中止しました。" + "不要な方を削除 (または退避) してから再実行してください" + ) + return self.age if age_exists else self.plaintext + + def mode(self, ref: SecretRef) -> str: + """``'age'`` / ``'plaintext'`` / ``'absent'`` のいずれかを返す""" + if self.age.exists(ref): + if self.plaintext.exists(ref): + # backend_for と同じ理由でここでも停止させる + self.backend_for(ref) + return MODE_AGE + if self.plaintext.exists(ref): + return MODE_PLAINTEXT + return MODE_ABSENT + + def is_encrypted(self, ref: SecretRef) -> bool: + return self.mode(ref) == MODE_AGE + + # -- 読み書き ----------------------------------------------------------- + + def exists(self, ref: SecretRef) -> bool: + return self.mode(ref) != MODE_ABSENT + + def path(self, ref: SecretRef) -> Path: + return self.backend_for(ref).path(ref) + + def load(self, ref: SecretRef) -> Dict[str, str]: + return self.backend_for(ref).load(ref) + + def save(self, ref: SecretRef, data: Dict[str, str]) -> Path: + """既存の保存形式を維持したまま保存する。 + + まだ何も無い参照は平文に落とす。暗号化へ移すのは ``devbase env encrypt`` + の役目であり、``set`` や ``sync`` が暗黙に形式を変えるべきではない。 + """ + return self.backend_for(ref).save(ref, data) + + def project_names(self) -> List[str]: + """暗号化済みの機密を持つプロジェクト名を返す""" + base = self.root / SECRETS_DIRNAME / 'projects' + if not base.is_dir(): + return [] + return sorted( + p.name[: -len('.env.age')] + for p in base.iterdir() + if p.is_file() and p.name.endswith('.env.age') + ) diff --git a/tests/cli/test_prefix_resolution.py b/tests/cli/test_prefix_resolution.py index 06f849ac..0acc12c6 100644 --- a/tests/cli/test_prefix_resolution.py +++ b/tests/cli/test_prefix_resolution.py @@ -69,3 +69,22 @@ def test_expand_argv_env_in_resolves_to_init(monkeypatch): monkeypatch.setattr(sys, "argv", ["devbase", "env", "in"]) cli._expand_argv() assert sys.argv == ["devbase", "env", "init"] + + +def test_expand_argv_env_k_resolves_to_keygen(monkeypatch): + """`devbase env k` は唯一の候補 (`keygen`) に解決される (PLAN35)""" + monkeypatch.setattr(sys, "argv", ["devbase", "env", "k"]) + cli._expand_argv() + assert sys.argv == ["devbase", "env", "keygen"] + + +def test_env_subcmd_map_covers_all_registered_subcommands(): + """SUBCMD_MAP['env'] が parser 登録済みサブコマンドを漏れなく含む。 + + `keygen` のように parser にだけ追加して SUBCMD_MAP を更新し忘れると、 + prefix 展開の対象から外れて短縮形が invalid choice で落ちる。 + """ + parser = cli._create_parser() + env_parser = parser._subparsers._group_actions[0].choices["env"] + registered = set(env_parser._subparsers._group_actions[0].choices) + assert registered == set(cli.SUBCMD_MAP[("env",)]) diff --git a/tests/commands/test_env_keygen.py b/tests/commands/test_env_keygen.py new file mode 100644 index 00000000..95e3d640 --- /dev/null +++ b/tests/commands/test_env_keygen.py @@ -0,0 +1,344 @@ +"""cmd_env_keygen: 生成先の契約と、鍵ローテーションの原子性""" + +from __future__ import annotations + +import stat +from pathlib import Path + +import pyrage +import pytest + +from devbase.commands import env as env_cmd +from devbase.env import agekeys +from devbase.env.secret_store import SecretRef, SecretStore +from devbase.errors import DevbaseError + + +@pytest.fixture +def devbase_root(tmp_path, monkeypatch): + """鍵を tmp_path 配下へ閉じ込める。 + + keygen は生成先を CLI で選べず必ず ``agekeys.key_file_path()`` へ書くため、 + テストからは ``DEVBASE_AGE_KEY_FILE`` を差し替えて既定パスごと tmp へ向ける。 + 実運用で別の場所へ置きたい利用者と同じ経路を通ることになる。 + """ + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(tmp_path / 'keys' / 'keys.txt')) + root = tmp_path / 'devbase' + root.mkdir() + return root + + +def _keygen(root, **kwargs): + return env_cmd.cmd_env_keygen(root, assume_yes=True, **kwargs) + + +# --------------------------------------------------------------------------- +# 生成先の契約 +# --------------------------------------------------------------------------- + +def test_keygen_writes_to_the_resolved_key_file_path(devbase_root, tmp_path): + """生成先は常に agekeys.key_file_path() = 復号側が探索する場所""" + assert _keygen(devbase_root) == 0 + + key_path = agekeys.key_file_path() + assert key_path == tmp_path / 'keys' / 'keys.txt' + assert key_path.exists() + assert stat.S_IMODE(key_path.stat().st_mode) == 0o600 + + +def test_generated_key_is_discoverable_for_decryption(devbase_root): + """生成した鍵が resolve_identities() / resolve_recipients() の双方から見える。 + + 生成先と探索先がずれると「保存はできるが復号できない」機密ができてしまうため、 + keygen 直後に暗号化・復号の両側が同じ鍵へ到達することを固定する。 + """ + assert _keygen(devbase_root) == 0 + + key_path = agekeys.key_file_path() + assert str(key_path) in agekeys.resolve_identities() + assert agekeys.resolve_recipients(devbase_root) == \ + [agekeys.read_public_key(key_path)] + + +def test_keygen_does_not_write_recipients_file(devbase_root): + """keygen はワークスペース固有の recipients.txt を作らない。 + + 鍵はグローバルなのに受信者リストはワークスペースごとに存在するため、ここで + 書き込むと別ワークスペースへ古い公開鍵が取り残され、失われた秘密鍵に対応する + 公開鍵で暗号化してしまう。 + """ + assert _keygen(devbase_root) == 0 + assert not agekeys.recipients_file(devbase_root).exists() + + assert _keygen(devbase_root, force=True) == 0 + assert not agekeys.recipients_file(devbase_root).exists() + + +def test_keygen_leaves_an_existing_recipients_file_untouched(devbase_root): + """明示的に登録済みの受信者リストは keygen が書き換えない (チーム運用の保全)""" + other = str(pyrage.x25519.Identity.generate().to_public()) + agekeys.save_recipients(devbase_root, [other]) + before = agekeys.recipients_file(devbase_root).read_bytes() + + assert _keygen(devbase_root, force=True) == 0 + assert agekeys.recipients_file(devbase_root).read_bytes() == before + + +# --------------------------------------------------------------------------- +# 鍵の保全 +# --------------------------------------------------------------------------- + +def test_keygen_without_force_keeps_existing_key(devbase_root): + assert _keygen(devbase_root) == 0 + before = agekeys.key_file_path().read_bytes() + + assert _keygen(devbase_root) == 0 + assert agekeys.key_file_path().read_bytes() == before + + +def test_keygen_without_force_does_not_request_an_overwrite(devbase_root, + monkeypatch): + """非 --force 実行は generate_key_file(force=True) を呼ばない。 + + 無条件に force=True を渡すと、コマンド側の存在チェックから実際の書き込みまで + の隙間に他プロセスが作った鍵を、利用者が要求していないのに消してしまう。 + """ + seen = [] + real = agekeys.generate_key_file + + def spy(path, *, force=False): + seen.append(force) + return real(path, force=force) + + monkeypatch.setattr(agekeys, 'generate_key_file', spy) + + assert _keygen(devbase_root) == 0 + assert seen == [False] + + +def test_keygen_aborts_when_a_key_appears_after_the_check(devbase_root, + monkeypatch): + """事前チェック後に鍵が現れたら、上書きせずエラー終了する (TOCTOU)。 + + ``_ensure_private_dir`` の直後に鍵を差し込んで、判定と書き込みの隙間で + 並行プロセスが先に生成した状況を再現する。 + """ + key_path = agekeys.key_file_path() + real_ensure = agekeys._ensure_private_dir + rival = b'AGE-SECRET-KEY-1RIVAL\n' + + def ensure_then_race(parent): + real_ensure(parent) + if not key_path.exists(): + key_path.write_bytes(rival) + + monkeypatch.setattr(agekeys, '_ensure_private_dir', ensure_then_race) + + assert _keygen(devbase_root) == 1 + assert key_path.read_bytes() == rival + + +def test_keygen_force_replaces_the_key(devbase_root): + assert _keygen(devbase_root) == 0 + old_public = agekeys.read_public_key(agekeys.key_file_path()) + + assert _keygen(devbase_root, force=True) == 0 + new_public = agekeys.read_public_key(agekeys.key_file_path()) + + assert new_public != old_public + + +# --------------------------------------------------------------------------- +# --force の確認プロンプト +# +# 鍵はグローバル (全ワークスペース共通) なので、確認の要否をカレントの +# DEVBASE_ROOT に機密があるかで決めてはいけない。機密がまだ無いプロジェクトから +# --force しても、他プロジェクトの機密は旧鍵でしか復号できないため。 +# --------------------------------------------------------------------------- + +def _answers(monkeypatch, *values): + """safe_input の応答をスクリプト化し、実際に聞かれた回数を返す""" + asked = [] + + def fake_input(prompt, default=''): + asked.append(prompt) + return values[len(asked) - 1] if len(asked) <= len(values) else default + + monkeypatch.setattr(env_cmd, 'safe_input', fake_input) + return asked + + +def test_keygen_force_prompts_even_without_encrypted_secrets(devbase_root, + monkeypatch): + """機密がまだ無いワークスペースでも --force は必ず確認する""" + assert _keygen(devbase_root) == 0 + before = agekeys.key_file_path().read_bytes() + + asked = _answers(monkeypatch, 'yes') + assert env_cmd.cmd_env_keygen(devbase_root, force=True) == 0 + + assert asked, "確認プロンプトが出ていない" + assert agekeys.key_file_path().read_bytes() != before + + +def test_keygen_force_prompt_mentions_other_workspaces(devbase_root, + monkeypatch, capsys): + """鍵がグローバルで他ワークスペースにも影響する旨をプロンプトで明示する""" + assert _keygen(devbase_root) == 0 + + _answers(monkeypatch, 'yes') + assert env_cmd.cmd_env_keygen(devbase_root, force=True) == 0 + + out = capsys.readouterr().out + assert '全プロジェクト共通' in out + assert 'ワークスペース' in out + + +def test_keygen_force_aborts_and_keeps_the_key_when_not_confirmed(devbase_root, + monkeypatch): + """yes 以外を入力したら中止し、鍵は 1 バイトも変えない""" + assert _keygen(devbase_root) == 0 + before = agekeys.key_file_path().read_bytes() + + _answers(monkeypatch, 'y') + assert env_cmd.cmd_env_keygen(devbase_root, force=True) == 1 + + assert agekeys.key_file_path().read_bytes() == before + + +def test_keygen_force_aborts_on_empty_answer(devbase_root, monkeypatch): + """非対話 (EOF → 空文字) でも黙って上書きせず中止する""" + assert _keygen(devbase_root) == 0 + before = agekeys.key_file_path().read_bytes() + + _answers(monkeypatch, '') + assert env_cmd.cmd_env_keygen(devbase_root, force=True) == 1 + + assert agekeys.key_file_path().read_bytes() == before + + +def test_keygen_force_skips_the_prompt_with_assume_yes(devbase_root, monkeypatch): + """--yes / -y でのみ確認を飛ばせる""" + assert _keygen(devbase_root) == 0 + before = agekeys.key_file_path().read_bytes() + + asked = _answers(monkeypatch) + assert env_cmd.cmd_env_keygen(devbase_root, force=True, assume_yes=True) == 0 + + assert asked == [] + assert agekeys.key_file_path().read_bytes() != before + + +def test_keygen_does_not_prompt_when_no_key_exists(devbase_root, monkeypatch): + """初回生成は失うものが無いので確認しない""" + asked = _answers(monkeypatch) + assert env_cmd.cmd_env_keygen(devbase_root, force=True) == 0 + assert asked == [] + + +def test_keygen_force_prompt_is_stronger_when_local_secrets_exist(devbase_root, + monkeypatch, + capsys): + """カレントに機密があるときは、その旨も併せて警告する""" + assert _keygen(devbase_root) == 0 + store = SecretStore(devbase_root) + store.age.save(SecretRef.for_global(), {'TOKEN': 'x'}) + + _answers(monkeypatch, 'yes') + assert env_cmd.cmd_env_keygen(devbase_root, force=True) == 0 + + out = capsys.readouterr().out + assert 'このワークスペースには暗号化済みの機密があり' in out + + +def test_keygen_keeps_the_old_key_when_generation_fails(devbase_root, monkeypatch): + """鍵生成が落ちても旧鍵はそのまま残る。 + + keygen 側に手動ロールバックは無く、原子性は + ``agekeys.generate_key_file`` → ``io_common.write_secure_bytes_atomic`` + (一時ファイル + fsync + os.replace) が担保する。ここではその契約が + コマンド層から見て守られていることだけを確認する。 + """ + assert _keygen(devbase_root) == 0 + key_path = agekeys.key_file_path() + key_before = key_path.read_bytes() + + def boom(*args, **kwargs): + raise DevbaseError('鍵を書けませんでした') + + monkeypatch.setattr(agekeys, 'generate_key_file', boom) + + assert _keygen(devbase_root, force=True) == 1 + assert key_path.read_bytes() == key_before + assert stat.S_IMODE(key_path.stat().st_mode) == 0o600 + + +def test_keygen_keeps_the_old_key_on_oserror(devbase_root, monkeypatch): + """OSError (ディスク枯渇など) でも旧鍵は無傷のまま残る""" + assert _keygen(devbase_root) == 0 + key_path = agekeys.key_file_path() + key_before = key_path.read_bytes() + + def boom(*args, **kwargs): + raise OSError(28, 'No space left on device') + + monkeypatch.setattr(agekeys, 'generate_key_file', boom) + + assert _keygen(devbase_root, force=True) == 1 + assert key_path.read_bytes() == key_before + + +# --------------------------------------------------------------------------- +# 読み取り不能な既存鍵 +# +# 原子性とは別の保護。読めないだけの鍵は権限を直せば回収できる可能性があるのに、 +# 生成が成功すると上書きで確実に消えるため、その前に中止する。 +# --------------------------------------------------------------------------- + +def _unreadable(monkeypatch, target: Path): + """``target`` にだけ「読み取り権限が無い」と見せる。 + + chmod 000 は root 実行だと読めてしまい、コンテナ内 CI とローカルで結果が + 変わる。読めない状態は権限ではなく ``os.access`` の差し替えで作る。 + """ + real_access = env_cmd.os.access + + def fake_access(path, mode, **kwargs): + if Path(path) == target and mode & env_cmd.os.R_OK: + return False + return real_access(path, mode, **kwargs) + + monkeypatch.setattr(env_cmd.os, 'access', fake_access) + + +def test_keygen_force_aborts_when_the_existing_key_is_unreadable(devbase_root, + monkeypatch, + caplog): + """読めない既存鍵は上書きせず中止する (削除も生成もしない)""" + assert _keygen(devbase_root) == 0 + key_path = agekeys.key_file_path() + key_before = key_path.read_bytes() + + generated = [] + monkeypatch.setattr(agekeys, 'generate_key_file', + lambda *a, **kw: generated.append(a)) + _unreadable(monkeypatch, key_path) + + assert _keygen(devbase_root, force=True) == 1 + + assert generated == [], "読めない鍵を上書きしようとしている" + assert key_path.exists(), "読めなかっただけの鍵を削除している" + assert key_path.read_bytes() == key_before + assert '上書きを中止' in caplog.text + + +def test_keygen_aborts_before_asking_for_confirmation_when_unreadable( + devbase_root, monkeypatch): + """中止が確定しているなら同意を求めない (空振りの確認を出さない)""" + assert _keygen(devbase_root) == 0 + _unreadable(monkeypatch, agekeys.key_file_path()) + + asked = _answers(monkeypatch, 'yes') + assert env_cmd.cmd_env_keygen(devbase_root, force=True) == 1 + + assert asked == [] diff --git a/tests/env/test_agekeys.py b/tests/env/test_agekeys.py new file mode 100644 index 00000000..94c72d61 --- /dev/null +++ b/tests/env/test_agekeys.py @@ -0,0 +1,513 @@ +"""agekeys.py: devbase 専用 age 鍵と受信者リストの管理""" + +from __future__ import annotations + +import os +import stat + +import pyrage +import pytest + +from devbase.env import agekeys + + +@pytest.fixture +def isolated_home(tmp_path, monkeypatch): + """鍵の既定パスを tmp_path 配下へ閉じ込める""" + monkeypatch.setenv('XDG_CONFIG_HOME', str(tmp_path / 'config')) + monkeypatch.delenv(agekeys.KEY_FILE_ENV, raising=False) + monkeypatch.setattr(agekeys.Path, 'home', staticmethod(lambda: tmp_path / 'home')) + return tmp_path + + +# --------------------------------------------------------------------------- +# パス解決 +# --------------------------------------------------------------------------- + +def test_key_file_path_uses_xdg_config_home(isolated_home): + assert agekeys.key_file_path() == isolated_home / 'config' / 'devbase' / 'age' / 'keys.txt' + + +def test_key_file_path_env_override_wins(isolated_home, monkeypatch): + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(isolated_home / 'custom' / 'k.txt')) + assert agekeys.key_file_path() == isolated_home / 'custom' / 'k.txt' + + +def test_key_file_path_falls_back_to_home_config(isolated_home, monkeypatch): + monkeypatch.delenv('XDG_CONFIG_HOME', raising=False) + assert agekeys.key_file_path() == isolated_home / 'home' / '.config' / 'devbase' / 'age' / 'keys.txt' + + +# --------------------------------------------------------------------------- +# 鍵の生成 +# --------------------------------------------------------------------------- + +def test_generate_key_file_writes_private_key_with_0600(isolated_home): + path, public = agekeys.generate_key_file() + + assert path.exists() + assert public.startswith('age1') + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + assert 'AGE-SECRET-KEY-1' in path.read_text() + + +def test_generate_key_file_creates_dir_with_0700(isolated_home): + path, _ = agekeys.generate_key_file() + assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700 + + +def test_generate_key_file_creates_every_missing_level_with_0700(isolated_home, + monkeypatch): + """親を複数階層まとめて作る場合、作った階層はすべて 0700 になる""" + key_path = isolated_home / 'a' / 'b' / 'c' / 'keys.txt' + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(key_path)) + + agekeys.generate_key_file() + + for level in (key_path.parent, key_path.parent.parent, + key_path.parent.parent.parent): + assert stat.S_IMODE(level.stat().st_mode) == 0o700 + + +def test_created_dirs_are_0700_from_the_moment_of_creation(isolated_home, + monkeypatch): + """umask 0 でも各階層は「作成した瞬間から」0700。 + + 一括作成してから chmod する実装だと、作成〜chmod の間だけ umask 依存の + 緩い権限が露出する。作成直後の mode を記録して、その隙が無いことを見る。 + """ + key_path = isolated_home / 'u1' / 'u2' / 'keys.txt' + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(key_path)) + + real_mkdir = agekeys.Path.mkdir + modes_at_creation = {} + + def recording_mkdir(self, *args, **kwargs): + result = real_mkdir(self, *args, **kwargs) + modes_at_creation[self] = stat.S_IMODE(self.stat().st_mode) + return result + + monkeypatch.setattr(agekeys.Path, 'mkdir', recording_mkdir) + + old = os.umask(0) + try: + agekeys.generate_key_file() + finally: + os.umask(old) + + levels = (key_path.parent, key_path.parent.parent) + for level in levels: + assert modes_at_creation[level] == 0o700, f'{level} が作成時点で緩い' + assert stat.S_IMODE(level.stat().st_mode) == 0o700 + + +def test_generate_key_file_survives_a_concurrently_created_level(isolated_home, + monkeypatch): + """途中の階層を別プロセスが先に作っていても失敗しない。 + + 先に作られた階層は「既存ディレクトリ」なので、権限は触らず素通りする + (既存ディレクトリを chmod しないという方針と一貫させる)。 + """ + key_path = isolated_home / 'x' / 'y' / 'keys.txt' + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(key_path)) + + racy = isolated_home / 'x' + real_mkdir = agekeys.Path.mkdir + + def racing_mkdir(self, *args, **kwargs): + if self == racy and not self.exists(): + # 別プロセスが一足先に作った状況を再現する + real_mkdir(self) + os.chmod(self, 0o755) + raise FileExistsError(17, 'File exists', str(self)) + return real_mkdir(self, *args, **kwargs) + + monkeypatch.setattr(agekeys.Path, 'mkdir', racing_mkdir) + + path, _ = agekeys.generate_key_file() + + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + # 他プロセスが作った階層の権限は変えない + assert stat.S_IMODE(racy.stat().st_mode) == 0o755 + # 自分で作った階層は 0700 + assert stat.S_IMODE(key_path.parent.stat().st_mode) == 0o700 + + +def test_generate_key_file_does_not_chmod_an_existing_dir(isolated_home, + monkeypatch): + """既存の共有ディレクトリを鍵の置き場に指定しても、その権限を変えない。 + + ``DEVBASE_AGE_KEY_FILE=/tmp/devbase-key`` のように既に在る共有ディレクトリを + 指されたとき、そこを 0700 に落とすと他ユーザーやサービスのアクセスを壊す。 + devbase が作っていないディレクトリは devbase の管轄外として触らない。 + """ + shared = isolated_home / 'shared' + shared.mkdir() + os.chmod(shared, 0o755) + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(shared / 'devbase-key')) + + path, _ = agekeys.generate_key_file() + + assert stat.S_IMODE(shared.stat().st_mode) == 0o755 + # ディレクトリを緩いままにする代わり、鍵ファイル自体は 0600 で守る + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_generate_key_file_warns_about_a_permissive_existing_dir(isolated_home, + monkeypatch, + caplog): + """権限を変えない代わりに、緩い既存ディレクトリは警告で知らせる""" + shared = isolated_home / 'shared' + shared.mkdir() + os.chmod(shared, 0o777) + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(shared / 'devbase-key')) + + with caplog.at_level('WARNING'): + agekeys.generate_key_file() + + assert any('shared' in r.getMessage() for r in caplog.records) + + +def test_generate_key_file_keeps_quiet_for_an_already_tight_existing_dir( + isolated_home, monkeypatch, caplog): + """既存でも 0700 なら警告しない (毎回鳴ると本当の警告が埋もれる)""" + tight = isolated_home / 'tight' + tight.mkdir() + os.chmod(tight, 0o700) + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(tight / 'devbase-key')) + + with caplog.at_level('WARNING'): + agekeys.generate_key_file() + + assert caplog.records == [] + + +def test_save_recipients_does_not_chmod_an_existing_secrets_dir(tmp_path): + """受信者リスト側も既存ディレクトリの権限を変えない (鍵ファイルと一貫)""" + secrets = tmp_path / 'secrets' + secrets.mkdir(parents=True) + os.chmod(secrets, 0o755) + + path = agekeys.save_recipients( + tmp_path, [str(pyrage.x25519.Identity.generate().to_public())]) + + assert stat.S_IMODE(secrets.stat().st_mode) == 0o755 + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_save_recipients_creates_the_secrets_dir_with_0700(tmp_path): + """自分で作った secrets/ は 0700 にする""" + agekeys.save_recipients( + tmp_path, [str(pyrage.x25519.Identity.generate().to_public())]) + assert stat.S_IMODE((tmp_path / 'secrets').stat().st_mode) == 0o700 + + +def test_generate_key_file_refuses_overwrite_without_force(isolated_home): + path, _ = agekeys.generate_key_file() + before = path.read_bytes() + + with pytest.raises(agekeys.AgeKeyError, match='既に存在'): + agekeys.generate_key_file() + + assert path.read_bytes() == before + + +# --------------------------------------------------------------------------- +# 新規生成の排他性 (TOCTOU) +# +# 「存在チェック → 生成」の隙間に他プロセスが鍵を作れると、後発が先発の鍵を +# 消し、先発鍵で暗号化した機密がその瞬間から復号不能になる。新規生成は +# O_CREAT|O_EXCL で不可分に作り、隙間そのものを無くす。 +# --------------------------------------------------------------------------- + +def test_generate_key_file_creates_a_new_key_exclusively(isolated_home, monkeypatch): + """新規生成は O_EXCL 付きで open する (os.replace で置き換えない)""" + seen = [] + real_open = os.open + + def spy_open(target, flags, *args, **kwargs): + seen.append((str(target), flags)) + return real_open(target, flags, *args, **kwargs) + + monkeypatch.setattr(agekeys.os, 'open', spy_open) + + path, _ = agekeys.generate_key_file() + + key_flags = [flags for target, flags in seen if target == str(path)] + assert key_flags, "鍵ファイルが os.open 経由で作られていない" + assert all(flags & os.O_EXCL for flags in key_flags), \ + "新規生成に O_EXCL が付いていない (判定と作成の隙間が残る)" + + +def test_generate_key_file_does_not_clobber_a_key_created_after_the_check( + isolated_home, monkeypatch): + """事前チェック通過後に他プロセスが鍵を作っても、その鍵を上書きしない。 + + ``_ensure_private_dir`` の直後に鍵を差し込んで、判定と書き込みの隙間で + 並行プロセスが先に生成した状況を再現する。排他生成なら後発 (このテストの + 呼び出し) が負けて、先発の鍵が 1 バイトも変わらずに残る。 + """ + path = agekeys.key_file_path() + real_ensure = agekeys._ensure_private_dir + rival = b'AGE-SECRET-KEY-1RIVAL\n' + + def ensure_then_race(parent): + real_ensure(parent) + if not path.exists(): + path.write_bytes(rival) + + monkeypatch.setattr(agekeys, '_ensure_private_dir', ensure_then_race) + + with pytest.raises(agekeys.AgeKeyError, match='既に存在'): + agekeys.generate_key_file() + + assert path.read_bytes() == rival + + +def test_generate_key_file_leaves_no_temp_file_on_first_generation(isolated_home): + """新規生成は一時ファイルを経由しない (鍵ディレクトリに残骸を残さない)""" + path, _ = agekeys.generate_key_file() + assert [p.name for p in path.parent.iterdir()] == [path.name] + + +def test_generate_key_file_removes_a_half_written_key_on_failure(isolated_home, + monkeypatch): + """書き込み途中で落ちたら中途半端な鍵を残さない。 + + 半端な鍵が残ると、以後の生成が「既に存在します」で止まるうえ、その鍵では + 何も復号できないという最悪の状態になる。 + """ + path = agekeys.key_file_path() + + def boom(fd): + raise OSError(28, 'No space left on device') + + monkeypatch.setattr(agekeys.os, 'fsync', boom) + + with pytest.raises(OSError): + agekeys.generate_key_file() + + assert not path.exists() + + +def test_generate_key_file_force_replaces_key(isolated_home): + path, first = agekeys.generate_key_file() + _, second = agekeys.generate_key_file(force=True) + assert first != second + assert agekeys.read_public_key(path) == second + + +def test_generate_key_file_force_keeps_0600(isolated_home): + """一時ファイル経由の差し替えでも権限が広がらない""" + path, _ = agekeys.generate_key_file() + agekeys.generate_key_file(force=True) + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_generate_key_file_force_leaves_no_temp_file(isolated_home): + """差し替え用の一時ファイルが鍵ディレクトリに残らない""" + path, _ = agekeys.generate_key_file() + agekeys.generate_key_file(force=True) + assert [p.name for p in path.parent.iterdir()] == [path.name] + + +def test_generate_key_file_force_keeps_old_key_when_replace_fails(isolated_home, + monkeypatch): + """差し替えに失敗しても旧鍵は無傷のまま残る (O_TRUNC 直書きなら失われる)""" + path, first = agekeys.generate_key_file() + before = path.read_bytes() + + def boom(src, dst): + raise OSError(28, 'No space left on device') + + monkeypatch.setattr(agekeys._io_common.os, 'replace', boom) + + with pytest.raises(OSError): + agekeys.generate_key_file(force=True) + + assert path.read_bytes() == before + assert agekeys.read_public_key(path) == first + # 書きかけの一時ファイルも掃除されている + assert [p.name for p in path.parent.iterdir()] == [path.name] + + +def test_save_recipients_keeps_old_list_when_replace_fails(tmp_path, monkeypatch): + """受信者リストも差し替え失敗時に旧内容を保つ""" + pubs = [str(pyrage.x25519.Identity.generate().to_public()) for _ in range(2)] + agekeys.save_recipients(tmp_path, pubs) + path = agekeys.recipients_file(tmp_path) + before = path.read_bytes() + + def boom(src, dst): + raise OSError(28, 'No space left on device') + + monkeypatch.setattr(agekeys._io_common.os, 'replace', boom) + + with pytest.raises(OSError): + agekeys.save_recipients(tmp_path, pubs[:1]) + + assert path.read_bytes() == before + assert agekeys.load_recipients(tmp_path) == pubs + + +def test_generated_key_can_decrypt_what_its_public_key_encrypted(isolated_home): + from devbase.env import cipher + + path, public = agekeys.generate_key_file() + blob = cipher.encrypt(b'payload', recipients=[public]) + assert cipher.decrypt(blob, identities=[str(path)]) == b'payload' + + +# --------------------------------------------------------------------------- +# 公開鍵の読み取り +# --------------------------------------------------------------------------- + +def test_read_public_key_derives_from_secret_not_comment(isolated_home): + """コメント行が嘘でも、秘密鍵から導出した公開鍵を返す""" + path, public = agekeys.generate_key_file() + tampered = path.read_text().replace(f'# public key: {public}', + '# public key: age1deadbeef') + path.write_text(tampered) + + assert agekeys.read_public_key(path) == public + + +def test_read_public_key_missing_file(isolated_home): + with pytest.raises(agekeys.AgeKeyError, match='見つかりません'): + agekeys.read_public_key(isolated_home / 'nope.txt') + + +def test_read_public_key_rejects_non_age_key(isolated_home): + path = isolated_home / 'ssh_like.txt' + path.write_text('-----BEGIN OPENSSH PRIVATE KEY-----\nzzz\n') + with pytest.raises(agekeys.AgeKeyError, match='AGE-SECRET-KEY-1'): + agekeys.read_public_key(path) + + +# --------------------------------------------------------------------------- +# 受信者リスト +# --------------------------------------------------------------------------- + +def test_recipients_roundtrip(tmp_path): + pub = str(pyrage.x25519.Identity.generate().to_public()) + + assert agekeys.load_recipients(tmp_path) == [] + assert agekeys.add_recipient(tmp_path, pub) is True + assert agekeys.load_recipients(tmp_path) == [pub] + + # 重複登録は no-op + assert agekeys.add_recipient(tmp_path, pub) is False + assert agekeys.load_recipients(tmp_path) == [pub] + + assert agekeys.remove_recipient(tmp_path, pub) is True + assert agekeys.load_recipients(tmp_path) == [] + assert agekeys.remove_recipient(tmp_path, pub) is False + + +def test_recipients_file_is_0600(tmp_path): + pub = str(pyrage.x25519.Identity.generate().to_public()) + agekeys.add_recipient(tmp_path, pub) + path = agekeys.recipients_file(tmp_path) + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_add_recipient_rejects_malformed_key(tmp_path): + from devbase.env.cipher import CipherError + + with pytest.raises(CipherError): + agekeys.add_recipient(tmp_path, 'not-a-key') + assert not agekeys.recipients_file(tmp_path).exists() + + +def test_load_recipients_skips_comments_and_blanks(tmp_path): + pub = str(pyrage.x25519.Identity.generate().to_public()) + path = agekeys.recipients_file(tmp_path) + path.parent.mkdir(parents=True) + path.write_text(f"# header\n\n{pub}\n \n") + assert agekeys.load_recipients(tmp_path) == [pub] + + +# --------------------------------------------------------------------------- +# 鍵の解決 +# --------------------------------------------------------------------------- + +def test_resolve_recipients_prefers_registered_list(isolated_home, tmp_path): + _, own = agekeys.generate_key_file() + other = str(pyrage.x25519.Identity.generate().to_public()) + agekeys.add_recipient(tmp_path, other) + + assert agekeys.resolve_recipients(tmp_path) == [other] + assert own not in agekeys.resolve_recipients(tmp_path) + + +def test_resolve_recipients_falls_back_to_own_public_key(isolated_home, tmp_path): + _, own = agekeys.generate_key_file() + assert agekeys.resolve_recipients(tmp_path) == [own] + + +def test_resolve_recipients_without_any_key_raises(isolated_home, tmp_path): + with pytest.raises(agekeys.AgeKeyError, match='公開鍵がありません'): + agekeys.resolve_recipients(tmp_path) + + +def test_resolve_identities_puts_devbase_key_first(isolated_home, monkeypatch): + ssh_key = isolated_home / 'id_ed25519' + ssh_key.write_text('dummy') + monkeypatch.setattr(agekeys._cipher, 'default_identity_paths', + lambda: [ssh_key]) + + path, _ = agekeys.generate_key_file() + assert agekeys.resolve_identities() == [str(path), str(ssh_key)] + + +def test_resolve_identities_empty_when_nothing_exists(isolated_home, monkeypatch): + monkeypatch.setattr(agekeys._cipher, 'default_identity_paths', lambda: []) + assert agekeys.resolve_identities() == [] + + +def test_save_recipients_is_idempotent_for_content(tmp_path): + pubs = [str(pyrage.x25519.Identity.generate().to_public()) for _ in range(2)] + agekeys.save_recipients(tmp_path, pubs) + first = agekeys.recipients_file(tmp_path).read_text() + agekeys.save_recipients(tmp_path, pubs) + assert agekeys.recipients_file(tmp_path).read_text() == first + assert agekeys.load_recipients(tmp_path) == pubs + + +def test_umask_does_not_widen_key_permissions(isolated_home): + """umask 0 でも鍵が 0600 で作られる (作成時点から権限を絞る)""" + old = os.umask(0) + try: + path, _ = agekeys.generate_key_file() + finally: + os.umask(old) + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_resolve_identities_still_decrypts_ssh_ciphertext_when_devbase_key_broken( + isolated_home, monkeypatch): + """専用鍵ファイルが壊れていても、旧来 ``~/.ssh`` の鍵で暗号化した暗号文を + ``resolve_identities()`` 経由で復号できる (移行互換性そのものの検証)。 + + ``resolve_identities`` は専用鍵を先頭に置くため、専用鍵の解決失敗でそこで + 止まってしまうと ``~/.ssh`` の鍵を試せない (PR #91 codex 指摘)。 + """ + from devbase.env import cipher + + ssh_identity = pyrage.x25519.Identity.generate() + ssh_key = isolated_home / 'id_ed25519' + ssh_key.write_text(str(ssh_identity)) + monkeypatch.setattr(agekeys._cipher, 'default_identity_paths', + lambda: [ssh_key]) + + # 専用鍵ファイルは存在するが中身が壊れている状態を作る + key_file = agekeys.key_file_path() + key_file.parent.mkdir(parents=True, exist_ok=True) + key_file.write_text('broken key material\n') + + identities = agekeys.resolve_identities() + assert identities == [str(key_file), str(ssh_key)] + + blob = cipher.encrypt(b'legacy-secret', + recipients=[str(ssh_identity.to_public())]) + assert cipher.decrypt(blob, identities=identities) == b'legacy-secret' diff --git a/tests/env/test_cipher.py b/tests/env/test_cipher.py index 65f1ce35..16b9de8b 100644 --- a/tests/env/test_cipher.py +++ b/tests/env/test_cipher.py @@ -2,6 +2,8 @@ from __future__ import annotations +import logging + import pyrage import pytest @@ -215,3 +217,62 @@ def test_resolve_identity_accepts_age_keygen_output_with_comments( blob = cipher.encrypt(b"payload", recipients=[pub]) assert cipher.decrypt(blob, identities=[str(id_path)]) == b"payload" + + +def test_decrypt_skips_unresolvable_identity_and_uses_valid_one( + tmp_path, x25519_keypair, caplog): + """解決できない identity が先頭にあっても、後続の有効な identity で復号できる。 + + ``agekeys.resolve_identities`` は「devbase 専用鍵 → ``~/.ssh`` の既定鍵」の順に + 候補を並べる。専用鍵ファイルが壊れているだけで復号が止まると、旧来 ``~/.ssh`` + の鍵で暗号化した暗号文を移行期間中に復号できるという意図が失われるため、 + 解決できない候補は警告を出して読み飛ばす (PR #91 codex 指摘)。 + """ + pub, priv_str = x25519_keypair + broken = tmp_path / "broken.key" + broken.write_text("not a key at all\n") + valid = tmp_path / "valid.key" + valid.write_text(priv_str) + + blob = cipher.encrypt(b"migrated", recipients=[pub]) + + with caplog.at_level(logging.WARNING, logger="devbase.env.cipher"): + plain = cipher.decrypt(blob, identities=[str(broken), str(valid)]) + + assert plain == b"migrated" + # 黙って読み飛ばすと原因が追えないので、理由が warning に残ること + warnings = [r.getMessage() for r in caplog.records + if r.levelno >= logging.WARNING] + assert any(str(broken) in m for m in warnings) + assert not any(str(valid) in m for m in warnings) + + +def test_decrypt_reports_every_failure_when_no_identity_resolves( + tmp_path, x25519_keypair): + """候補が全滅した場合は従来どおり CipherError。各候補の失敗理由を含む""" + pub, _ = x25519_keypair + missing = tmp_path / "missing.key" + broken = tmp_path / "broken.key" + broken.write_text("not a key at all\n") + + blob = cipher.encrypt(b"x", recipients=[pub]) + + with pytest.raises(cipher.CipherError) as excinfo: + cipher.decrypt(blob, identities=[str(missing), str(broken)]) + + message = str(excinfo.value) + assert str(missing) in message and "見つかりません" in message + assert str(broken) in message and "秘密鍵の解釈に失敗" in message + + +def test_decrypt_keeps_message_when_identity_resolves_but_mismatches( + tmp_path, x25519_keypair): + """解決には成功したが鍵が一致しない場合のメッセージは従来どおり""" + pub, _ = x25519_keypair + other = tmp_path / "other.key" + other.write_text(str(pyrage.x25519.Identity.generate())) + + blob = cipher.encrypt(b"x", recipients=[pub]) + + with pytest.raises(cipher.CipherError, match="復号に失敗しました"): + cipher.decrypt(blob, identities=[str(other)]) diff --git a/tests/env/test_io_common.py b/tests/env/test_io_common.py index c65392d5..e47bdccb 100644 --- a/tests/env/test_io_common.py +++ b/tests/env/test_io_common.py @@ -120,3 +120,52 @@ def test_decrypt_uses_correct_identity_from_multiple_defaults(tmp_path, fake_hom # 両 identity を渡して復号 → pyrage が正しい鍵 (id2) を選んで復号する plain = cipher.decrypt(blob, identities=identities) assert plain == b"team-secret" + + +# --------------------------------------------------------------------------- +# write_secure_bytes_atomic +# --------------------------------------------------------------------------- + +def test_write_secure_bytes_atomic_creates_file_with_0600(tmp_path): + import os + import stat + + path = tmp_path / "nested" / "secret.bin" + old = os.umask(0) + try: + io_common.write_secure_bytes_atomic(path, b"payload") + finally: + os.umask(old) + + assert path.read_bytes() == b"payload" + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_write_secure_bytes_atomic_replaces_and_leaves_no_temp(tmp_path): + path = tmp_path / "secret.bin" + io_common.write_secure_bytes_atomic(path, b"old") + io_common.write_secure_bytes_atomic(path, b"new") + + assert path.read_bytes() == b"new" + assert [p.name for p in tmp_path.iterdir()] == [path.name] + + +def test_write_secure_bytes_atomic_keeps_old_content_on_failure(tmp_path, + monkeypatch): + """差し替えに失敗しても旧内容は残る (直接上書きなら失われる差分)""" + import os + + path = tmp_path / "secret.bin" + io_common.write_secure_bytes_atomic(path, b"old") + + def boom(src, dst): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(os, "replace", boom) + + with pytest.raises(OSError): + io_common.write_secure_bytes_atomic(path, b"new") + + assert path.read_bytes() == b"old" + # 書きかけの一時ファイルも残さない + assert [p.name for p in tmp_path.iterdir()] == [path.name] diff --git a/tests/env/test_secret_store.py b/tests/env/test_secret_store.py new file mode 100644 index 00000000..fd13c86d --- /dev/null +++ b/tests/env/test_secret_store.py @@ -0,0 +1,331 @@ +"""secret_store.py: 平文 / age の保存先抽象と自動判定""" + +from __future__ import annotations + +import stat + +import pyrage +import pytest + +from devbase.env.secret_store import ( + MODE_ABSENT, + MODE_AGE, + MODE_PLAINTEXT, + AgeBackend, + SecretRef, + SecretStore, + SecretStoreError, +) + + +@pytest.fixture +def keypair(): + identity = pyrage.x25519.Identity.generate() + return str(identity.to_public()), str(identity) + + +@pytest.fixture +def store(tmp_path, keypair): + """明示的な鍵を渡した SecretStore (ホームの鍵に依存しない)""" + public, secret = keypair + id_path = tmp_path / 'identity.key' + id_path.write_text(secret) + (tmp_path / 'projects').mkdir() + return SecretStore(tmp_path, recipients=[public], identities=[str(id_path)]) + + +GLOBAL = SecretRef.for_global() +SAMPLE = {'ANTHROPIC_API_KEY': 'sk-test', 'AWS_SECRET_ACCESS_KEY': 'secret value'} + + +# --------------------------------------------------------------------------- +# 参照 +# --------------------------------------------------------------------------- + +def test_project_ref_rejects_path_traversal(): + for bad in ('../evil', 'a/b', '.', '..'): + with pytest.raises(SecretStoreError): + SecretRef.for_project(bad) + + +def test_project_ref_rejects_empty_name(): + with pytest.raises(SecretStoreError): + SecretRef.for_project('') + + +# --------------------------------------------------------------------------- +# 保存先パス +# --------------------------------------------------------------------------- + +def test_paths_follow_the_documented_layout(tmp_path, store): + proj = SecretRef.for_project('web') + + assert store.plaintext.path(GLOBAL) == tmp_path / '.env' + assert store.plaintext.path(proj) == tmp_path / 'projects' / 'web' / '.env' + assert store.age.path(GLOBAL) == tmp_path / 'secrets' / 'global.env.age' + assert store.age.path(proj) == tmp_path / 'secrets' / 'projects' / 'web.env.age' + + +# --------------------------------------------------------------------------- +# ラウンドトリップ +# --------------------------------------------------------------------------- + +def test_age_backend_roundtrip(store): + path = store.age.save(GLOBAL, SAMPLE) + + assert path.exists() + assert b'sk-test' not in path.read_bytes() # 平文が残っていない + assert store.age.load(GLOBAL) == SAMPLE + + +def test_age_backend_file_is_0600(store): + path = store.age.save(GLOBAL, SAMPLE) + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_plaintext_backend_roundtrip(store): + store.plaintext.save(GLOBAL, SAMPLE) + assert store.plaintext.load(GLOBAL) == SAMPLE + + +def test_project_secrets_roundtrip(store): + proj = SecretRef.for_project('web') + store.age.save(proj, {'DB_PASSWORD': 'p@ss word'}) + assert store.age.load(proj) == {'DB_PASSWORD': 'p@ss word'} + + +def test_load_of_missing_file_is_empty(store): + assert store.age.load(GLOBAL) == {} + assert store.plaintext.load(GLOBAL) == {} + + +def test_age_load_with_wrong_identity_raises(tmp_path, keypair): + public, _ = keypair + other = tmp_path / 'other.key' + other.write_text(str(pyrage.x25519.Identity.generate())) + + writer = AgeBackend(tmp_path, recipients=[public]) + writer.save(GLOBAL, SAMPLE) + + reader = AgeBackend(tmp_path, identities=[str(other)]) + with pytest.raises(SecretStoreError, match='復号'): + reader.load(GLOBAL) + + +def test_age_load_wraps_invalid_utf8_plaintext(tmp_path, keypair): + """復号は通ったが中身が UTF-8 でない場合も SecretStoreError にする。 + + 素の UnicodeDecodeError が漏れると、呼び出し側は DevbaseError だけを捕まえて + いるためトレースバックのまま落ちる。PlaintextBackend.load と例外を揃える。 + """ + from devbase.env import cipher as _cipher + + public, secret = keypair + id_path = tmp_path / 'identity.key' + id_path.write_text(secret) + + backend = AgeBackend(tmp_path, recipients=[public], + identities=[str(id_path)]) + path = backend.path(GLOBAL) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(_cipher.encrypt(b'\xff\xfe not utf-8', recipients=[public])) + + with pytest.raises(SecretStoreError, match='UTF-8'): + backend.load(GLOBAL) + + +# --------------------------------------------------------------------------- +# 自動判定 +# --------------------------------------------------------------------------- + +def test_mode_is_absent_when_nothing_exists(store): + assert store.mode(GLOBAL) == MODE_ABSENT + assert store.exists(GLOBAL) is False + + +def test_mode_is_plaintext_when_only_plain_exists(store): + store.plaintext.save(GLOBAL, SAMPLE) + assert store.mode(GLOBAL) == MODE_PLAINTEXT + assert store.load(GLOBAL) == SAMPLE + + +def test_mode_is_age_when_only_encrypted_exists(store): + store.age.save(GLOBAL, SAMPLE) + assert store.mode(GLOBAL) == MODE_AGE + assert store.is_encrypted(GLOBAL) is True + assert store.load(GLOBAL) == SAMPLE + + +def test_both_present_is_an_error(store): + store.plaintext.save(GLOBAL, SAMPLE) + store.age.save(GLOBAL, SAMPLE) + + with pytest.raises(SecretStoreError, match='両方に存在'): + store.load(GLOBAL) + with pytest.raises(SecretStoreError, match='両方に存在'): + store.mode(GLOBAL) + + +def test_both_present_error_names_both_paths(store): + store.plaintext.save(GLOBAL, SAMPLE) + store.age.save(GLOBAL, SAMPLE) + with pytest.raises(SecretStoreError) as exc: + store.path(GLOBAL) + message = str(exc.value) + assert str(store.age.path(GLOBAL)) in message + assert str(store.plaintext.path(GLOBAL)) in message + + +def test_save_keeps_the_existing_format(store): + """set / sync 相当の保存が形式を勝手に変えない""" + store.age.save(GLOBAL, SAMPLE) + store.save(GLOBAL, {**SAMPLE, 'NEW': '1'}) + + assert store.mode(GLOBAL) == MODE_AGE + assert store.plaintext.path(GLOBAL).exists() is False + assert store.load(GLOBAL)['NEW'] == '1' + + +def test_save_defaults_to_plaintext_for_new_refs(store): + store.save(GLOBAL, SAMPLE) + assert store.mode(GLOBAL) == MODE_PLAINTEXT + + +# --------------------------------------------------------------------------- +# 削除・一覧 +# --------------------------------------------------------------------------- + +def test_remove_reports_whether_a_file_was_deleted(store): + store.age.save(GLOBAL, SAMPLE) + assert store.age.remove(GLOBAL) is True + assert store.age.remove(GLOBAL) is False + + +def test_project_names_lists_encrypted_projects_only(store): + store.age.save(SecretRef.for_project('web'), {'A': '1'}) + store.age.save(SecretRef.for_project('api'), {'B': '2'}) + store.plaintext.save(SecretRef.for_project('legacy'), {'C': '3'}) + + assert store.project_names() == ['api', 'web'] + + +def test_project_names_empty_without_secrets_dir(store): + assert store.project_names() == [] + + +# --------------------------------------------------------------------------- +# 鍵未整備時のエラー +# --------------------------------------------------------------------------- + +def test_age_save_without_recipients_raises(tmp_path, monkeypatch): + from devbase.env import agekeys + + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(tmp_path / 'absent' / 'keys.txt')) + backend = AgeBackend(tmp_path) + with pytest.raises(agekeys.AgeKeyError, match='公開鍵がありません'): + backend.save(GLOBAL, SAMPLE) + + +def test_age_load_without_identities_raises(tmp_path, keypair, monkeypatch): + from devbase.env import agekeys + + public, _ = keypair + AgeBackend(tmp_path, recipients=[public]).save(GLOBAL, SAMPLE) + + monkeypatch.setenv(agekeys.KEY_FILE_ENV, str(tmp_path / 'absent' / 'keys.txt')) + monkeypatch.setattr(agekeys._cipher, 'default_identity_paths', lambda: []) + with pytest.raises(SecretStoreError, match='秘密鍵が見つかりません'): + AgeBackend(tmp_path).load(GLOBAL) + + +def test_plaintext_load_of_binary_reports_a_useful_error(store): + path = store.plaintext.path(GLOBAL) + path.write_bytes(b'\xff\xfe\x00binary') + with pytest.raises(SecretStoreError, match='UTF-8'): + store.plaintext.load(GLOBAL) + + +# --------------------------------------------------------------------------- +# 保存の原子性 +# +# 暗号文を失うと機密は復旧できない。既存ファイルを直接 truncate せず、一時ファイル +# → os.replace で差し替えているので、書き込みの途中で落ちても旧内容が残る。 +# --------------------------------------------------------------------------- + +def _fail_replace(monkeypatch, exc): + """``os.replace`` だけを失敗させる (差し替え直前までは正常に進む)""" + from devbase.env import io_common + + def boom(src, dst): + raise exc + + monkeypatch.setattr(io_common.os, 'replace', boom) + + +def test_age_save_keeps_the_old_ciphertext_when_replace_fails(store, monkeypatch): + store.age.save(GLOBAL, SAMPLE) + path = store.age.path(GLOBAL) + before = path.read_bytes() + + _fail_replace(monkeypatch, OSError(28, 'No space left on device')) + + with pytest.raises(SecretStoreError, match='書き込みに失敗'): + store.age.save(GLOBAL, {**SAMPLE, 'NEW': '1'}) + + # 旧 ciphertext が無傷 = まだ旧内容を復号できる + assert path.read_bytes() == before + monkeypatch.undo() + assert store.age.load(GLOBAL) == SAMPLE + + +def test_age_save_leaves_no_temp_file_when_replace_fails(store, monkeypatch): + store.age.save(GLOBAL, SAMPLE) + path = store.age.path(GLOBAL) + + _fail_replace(monkeypatch, OSError(28, 'No space left on device')) + + with pytest.raises(SecretStoreError): + store.age.save(GLOBAL, {**SAMPLE, 'NEW': '1'}) + + # 書きかけの一時ファイル (中身は新しい暗号文) を放置しない + assert sorted(p.name for p in path.parent.iterdir()) == [path.name] + + +def test_age_save_keeps_the_old_ciphertext_when_interrupted(store, monkeypatch): + """KeyboardInterrupt のような BaseException でも旧内容と後始末は変わらない""" + store.age.save(GLOBAL, SAMPLE) + path = store.age.path(GLOBAL) + before = path.read_bytes() + + _fail_replace(monkeypatch, KeyboardInterrupt()) + + with pytest.raises(KeyboardInterrupt): + store.age.save(GLOBAL, {**SAMPLE, 'NEW': '1'}) + + assert path.read_bytes() == before + assert sorted(p.name for p in path.parent.iterdir()) == [path.name] + + +def test_plaintext_save_keeps_the_old_content_when_replace_fails(store, + monkeypatch): + store.plaintext.save(GLOBAL, SAMPLE) + path = store.plaintext.path(GLOBAL) + before = path.read_bytes() + + _fail_replace(monkeypatch, OSError(28, 'No space left on device')) + + with pytest.raises(SecretStoreError, match='書き込みに失敗'): + store.plaintext.save(GLOBAL, {**SAMPLE, 'NEW': '1'}) + + assert path.read_bytes() == before + + +def test_age_save_is_atomic_across_updates(store): + """通常経路では差し替えが成功し、一時ファイルも残らない""" + store.age.save(GLOBAL, SAMPLE) + store.age.save(GLOBAL, {**SAMPLE, 'NEW': '1'}) + + path = store.age.path(GLOBAL) + assert sorted(p.name for p in path.parent.iterdir()) == [path.name] + assert store.age.load(GLOBAL)['NEW'] == '1' + assert stat.S_IMODE(path.stat().st_mode) == 0o600