diff --git a/docs/cli-reference.md b/docs/cli-reference.md
index 0cf838d..b877816 100644
--- a/docs/cli-reference.md
+++ b/docs/cli-reference.md
@@ -427,6 +427,58 @@ Use an existing prepared UUID in the local owner's POSIX ledger. Status remains
Watch is read-only: no intents, waits, receipts, ledger writes, submission or wake. It adds no host/device/Windows route or Codex ACK. Limits are 16 KiB per frame and 256 KiB total with a 2 KiB end reserve. Real-fd stdout uses one owned isolated output-only Python helper that sees only bounded public serialized bytes, never changes shared descriptor flags and reads no ledger/native data. The parent enforces the original total budget including IPC, checks it before new ledger read/validation stages, and reserves 100 ms within it for final output and cleanup, so sampling may end earlier. It terminates and reaps its helper on timeout, SIGINT or error; uninterruptible OS operations prevent a hard-real-time guarantee. Partial or unacknowledged writes prohibit subsequent/end frames; consumers must reject truncated final NDJSON lines. Helper completion means OS write completion, not recipient consumption. Snapshots are not a lossless event stream, and local UTC observations do not prove ACK arrival order. See the [handoff runtime guide](handoff-runtime.md) for privacy, retained submission/ACK/wait meanings, read-only collector queries and the existing quarantine gate. Watch launches no native message process or collector. This candidate is not a published release. No payload/capability digests are exposed; public targetGeneration keeps its existing contract. The read-only ledger releases its shared lock after reading immutable snapshot bytes, before parsing and validation. Interrupted final output and cleanup are capped at 100 ms within the remaining total budget. Overall success requires a complete successful end frame and exit 0: cleanup may fail after that frame. A nonzero exit means overall failure, without promoting ACK, discarding independently valid submission facts or permitting retry.
+### Trusted local SSH configuration (v1.3 candidate, not published)
+
+On Linux and macOS, an explicit local CLI invocation of `list`, `send`, `doctor`,
+or `update` with `--host` can select an existing OpenSSH configuration by pairing
+`--ssh-config ABSOLUTE_PATH` with `--trust-ssh-config`. Both flags are required.
+For example, using a placeholder path (quoted spaces are supported):
+
+```bash
+session-peer list --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config
+session-peer doctor --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config --check-return-route
+```
+
+This grants local code-execution authority to the selected configuration:
+`Match exec` can run even during `ssh -G` identity discovery or `send --dry-run`.
+The operator must also trust `Include` files, `ProxyCommand`, and other external
+dependencies. Validation is not a configuration sandbox and does not recursively
+validate those dependencies. No existing profile or configuration file is edited.
+
+The path must be absolute, with no `..` components, control characters, or
+symlinks in any component. The selected file must be a regular file owned by the
+current effective user, have exactly one hard link, be at most 64 KiB, and have
+neither group/other write permissions nor setuid/setgid/sticky bits. Ancestor
+directories must be owned by root or the current effective user and must not be
+group/other writable, except for root-owned sticky directories. Linux rejects
+extended access ACLs and directory default ACLs; macOS rejects allow ACL entries
+and volumes with ownership disabled. ACL or ownership checks that cannot be
+verified refuse the selection. These are conservative checks, not a full
+effective-permissions interpreter.
+
+One selection is reused throughout that CLI call for local `ssh -G` discovery,
+doctor, remote listing/version checks, update probes/install/verification, and
+send/handoff SSH calls. Before each SSH spawn it revalidates file identity,
+metadata, content hash, and storage safety. Detected changes refuse the operation
+without falling back to default configuration. OpenSSH opens the original path;
+this is not an immutable byte snapshot and a same-user change between validation
+and OpenSSH's read remains possible.
+
+This mode forces `StrictHostKeyChecking=yes`, `UpdateHostKeys=no`,
+`ForwardAgent=no`, `PermitLocalCommand=no`, and `BatchMode=yes`, and disables
+connection multiplexing. Existing explicit host-key pins retain priority. The
+`--ssh-opt` allowlist is unchanged: `-F` and `ProxyCommand` remain forbidden there.
+The selected local path is not forwarded to the remote command, reply route, or
+public JSON. Reverse doctor checks use the remote machine's own configuration;
+they never inherit this path. Forward success does not prove reverse
+authentication, message consumption, or an ACK.
+
+Environment variables, MCP requests, ReplyTo addresses, and stored presets cannot
+select this mode. Windows, `setup`, paired-device routes, preset combinations,
+reply-URI destinations, and invocations without `--host` are rejected before
+reading the configuration or starting SSH. Omitting the pair preserves the
+existing default SSH behavior.
+
### Updating
For package-managed installs, use the same manager that installed the command:
diff --git a/docs/i18n-manifest.json b/docs/i18n-manifest.json
index 990e0bb..9dcecb1 100644
--- a/docs/i18n-manifest.json
+++ b/docs/i18n-manifest.json
@@ -15,7 +15,7 @@
"docs/ai-assistant-guide.md": "9624ec96b92fb688fae3be3c94cb3564c77985091be451fbf1f056a1af9b9d9c",
"docs/antigravity.md": "f2b5122a0ce6e6fb5bd4d8ed50ce2c8bd2dc9c98ec8ce328405cf7c2cd24229d",
"docs/architecture/agent-transports.md": "01a4e3942e87acb476595859f2e37f9653ea1e3bc19ebc68dced1adf20a6966f",
- "docs/cli-reference.md": "af1d3524bad9020dd4cb9ecb6b23db7207686ee956ae2215c21d01d79551ede1",
+ "docs/cli-reference.md": "5d4a52e5eda0e3287e45fd1d8bd852acee76af9917e759431efd84d247a73b0e",
"docs/compatibility-v1.md": "7e551aad37a369faff7e370728376763efb54c2c9e9cfd8556a0c19aabc3d103",
"docs/contracts/handoff-v1.md": "46a953010c49214bd8f969412f26713384fbe09a0ba78ebf0fef75da40d0d889",
"docs/diagnostics.md": "0949e53612eeab2cd7c7bbda498e14d71e9cb1b1189aa7722030cb8702cd8481",
diff --git a/docs/ja/cli-reference.md b/docs/ja/cli-reference.md
index 066e589..9c58d0f 100644
--- a/docs/ja/cli-reference.md
+++ b/docs/ja/cli-reference.md
@@ -415,6 +415,25 @@ session-peer handoff watch --correlation-id PREPARED_UUID --watch-timeout 30 --i
監視は読み取り専用で、intent・待機・受領確認の作成、記録の書き込み、提出や起動を行いません。ホスト・機器・Windows 経路や Codex ACK は追加しません。フレームは 16 KiB、総量は 256 KiB に制限し、終了用に 2 KiB を予約します。実ファイル記述子の標準出力には専用の分離された Python 出力補助プロセスを一つ使用します。サイズ制限付きの直列化された公開バイトだけを受け取り、共有記述子フラグを変えず、記録やネイティブデータも読みません。親は IPC を含め元の総予算を適用し、新しい記録読み取り・検証段階の前にも期限を確認します。総時間内に最終出力と終了処理用の 100 ms を確保するため、照会は早めに終わる場合があります。タイムアウト、SIGINT、エラー時には補助プロセスを終了・回収しますが、中断不能な OS 操作の厳密リアルタイム保証ではありません。部分書き込みや完了応答がない場合は終了を含む後続フレームを禁止し、利用側は切れた最後の NDJSON 行を拒否する必要があります。補助プロセスの完了は OS 書き込み完了であり、受信者の消費ではありません。スナップショットは全イベントを欠落なく伝えるストリームではなく、ローカル UTC 観測は ACK 到着順序を証明しません。プライバシー、保持する提出・ACK・待機の意味、読み取り専用コレクター照会、従来の隔離の制約は [handoff ランタイムガイド](handoff-runtime.md)を参照してください。監視はネイティブメッセージプロセスやコレクターを起動しません。この候補は公開リリースではありません。 ペイロード・権限のダイジェストは公開せず、公開 targetGeneration の既存契約は維持します。 読み取り専用記録は不変のスナップショットバイトを読み終えた後、解析・検証の前に共有ロックを解放します。 中断時の最終出力と終了処理は、残りの総予算内で最大 100 ms に制限します。 全体の成功には完全な成功終了フレームと終了コード 0 の両方が必要です。そのフレームの後に終了処理が失敗する場合があります。非ゼロの終了は全体の失敗であり、ACK を格上げしたり、独立して有効な提出事実を捨てたり、再試行を許可したりするものではありません。
+### 信頼するローカル SSH 設定 (v1.3 候補、未公開)
+
+Linux と macOS では、`--host` を指定したローカル CLI の `list`、`send`、`doctor`、`update` で、`--ssh-config ABSOLUTE_PATH` と `--trust-ssh-config` を併用して既存の OpenSSH 設定を選択できます。両方のフラグが必要です。次のパスは例であり、引用符で囲んだ空白も利用できます。
+
+```bash
+session-peer list --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config
+session-peer doctor --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config --check-return-route
+```
+
+この選択は設定にローカルコード実行権限を与えます。`Match exec` は `ssh -G` によるユーザー確認や `send --dry-run` 中でも実行され得ます。運用者は `Include` ファイル、`ProxyCommand`、その他の外部依存先も信頼する必要があります。検証は設定のサンドボックスではなく、依存先を再帰的に検証しません。既存のプロファイルや設定ファイルは編集しません。
+
+絶対パスが必要で、`..` 要素、制御文字、パス内のシンボリックリンクは許可しません。選択したファイルは現在の実効ユーザーが所有する通常ファイルで、ハードリンク数が 1、最大 64 KiB である必要があります。グループ・その他の書き込み権限、および setuid/setgid/sticky ビットも拒否します。祖先ディレクトリは root または現在の実効ユーザーの所有で、グループ・その他から書き込み不可である必要があります。ただし root 所有の sticky ディレクトリは例外です。Linux では拡張アクセス ACL とディレクトリのデフォルト ACL、macOS では allow ACL エントリと所有権が無効のボリュームを拒否します。ACL や所有権を確認できない場合も拒否します。これは保守的な検査であり、実効権限の完全な解釈ではありません。
+
+同じ CLI 呼び出し内で、ローカル `ssh -G` 照会、doctor、リモート一覧・バージョン確認、更新の調査・インストール・検証、send/handoff の SSH に同じ選択を使います。各 SSH 起動前にファイルの識別情報、メタデータ、内容ハッシュ、保存場所の安全性を再検証します。変更を検出すると、既定設定にフォールバックせず拒否します。OpenSSH は元のパスを開きます。不変のバイトスナップショットではなく、検証から OpenSSH の読み取りまでの間に同じユーザーが変更する競合は残ります。
+
+このモードでは `StrictHostKeyChecking=yes`、`UpdateHostKeys=no`、`ForwardAgent=no`、`PermitLocalCommand=no`、`BatchMode=yes` を強制し、接続の多重化を無効にします。既存の明示的なホスト鍵固定が優先されます。`--ssh-opt` の許可リストは変更せず、そこでの `-F` と `ProxyCommand` は引き続き禁止します。選択したローカルパスはリモートコマンド、返信経路、公開 JSON に渡しません。逆方向の doctor はリモート側自身の設定を使用し、このパスを継承しません。順方向の成功は逆方向の認証、メッセージ消費、ACK を証明しません。
+
+環境変数、MCP リクエスト、ReplyTo アドレス、保存済みプリセットからは選択できません。Windows、`setup`、ペアリング済みデバイス経路、プリセットとの併用、返信 URI 宛先、`--host` のない呼び出しは、設定を読む前および SSH 起動前に拒否します。両フラグを省略した場合は従来の既定 SSH 動作を維持します。
+
### 更新
diff --git a/docs/ko/cli-reference.md b/docs/ko/cli-reference.md
index 4abacb8..1616da5 100644
--- a/docs/ko/cli-reference.md
+++ b/docs/ko/cli-reference.md
@@ -414,6 +414,25 @@ session-peer handoff watch --correlation-id PREPARED_UUID --watch-timeout 30 --i
감시는 읽기 전용이며 의도·대기·수신 확인 생성, 기록 쓰기, 제출·깨우기를 수행하지 않습니다. 호스트·기기·Windows 경로나 Codex ACK를 추가하지 않습니다. 프레임은 16 KiB, 전체는 256 KiB로 제한하고 종료용 2 KiB를 남깁니다. 실제 파일 디스크립터 표준 출력에는 전용 격리 Python 출력 보조 프로세스 하나를 사용합니다. 크기가 제한된 직렬화된 공개 바이트만 받고 공유 디스크립터 플래그를 바꾸거나 기록·네이티브 데이터를 읽지 않습니다. 부모는 IPC까지 원래 총예산으로 제한하고 새 기록 읽기·검증 단계 전에도 기한을 확인합니다. 총 시간 안에서 최종 출력·정리에 100 ms를 남기므로 조회는 더 일찍 끝날 수 있습니다. 시간 초과·SIGINT·오류 시 보조 프로세스를 종료·회수합니다. 다만 중단할 수 없는 OS 작업이 있어 엄격한 실시간 보장을 제공하지는 못합니다. 부분 쓰기나 완료 응답 부재 시 종료를 포함한 후속 프레임을 금지하며 소비자는 잘린 마지막 NDJSON 줄을 거부해야 합니다. 보조 프로세스 완료는 OS 쓰기 완료이며 수신자 소비의 증거가 아닙니다. 스냅샷은 모든 이벤트를 빠짐없이 전달하는 스트림이 아니며 로컬 UTC 관측은 ACK 도착 순서의 증거가 아닙니다. 개인정보, 유지되는 제출·ACK·대기의 의미, 읽기 전용 수집기 조회와 기존 격리 조건은 [handoff 런타임 가이드](handoff-runtime.md)를 참고하세요. 감시는 네이티브 메시지 프로세스나 수집기를 시작하지 않습니다. 이 후보는 배포된 릴리스가 아닙니다. 페이로드·권한 다이제스트를 노출하지 않으며 공개 targetGeneration의 기존 계약은 유지합니다. 읽기 전용 기록은 불변 스냅샷 바이트를 읽은 뒤 파싱·검증 전에 공유 잠금을 해제합니다. 중단 시 최종 출력·정리는 남은 총예산 안에서 최대 100 ms로 제한합니다. 전체 성공에는 완전한 성공 종료 프레임과 종료 코드 0이 모두 필요합니다. 해당 프레임 뒤에도 정리가 실패할 수 있습니다. 0이 아닌 종료는 전체 실패이며 ACK를 상향 해석하거나 독립적으로 유효한 제출 사실을 버리거나 재시도해도 된다는 뜻은 아닙니다.
+### 신뢰하는 로컬 SSH 설정 (v1.3 후보, 미출시)
+
+Linux와 macOS에서는 `--host`를 지정한 로컬 CLI의 `list`, `send`, `doctor`, `update` 호출에서 `--ssh-config ABSOLUTE_PATH`와 `--trust-ssh-config`를 함께 사용해 기존 OpenSSH 설정을 선택할 수 있습니다. 두 플래그가 모두 필요합니다. 다음은 예시 경로이며, 따옴표로 감싼 공백도 지원합니다.
+
+```bash
+session-peer list --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config
+session-peer doctor --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config --check-return-route
+```
+
+이 선택은 해당 설정에 로컬 코드 실행 권한을 부여합니다. `Match exec`는 `ssh -G` 사용자 확인이나 `send --dry-run` 중에도 실행될 수 있습니다. 운영자는 `Include` 파일, `ProxyCommand`, 기타 외부 의존성도 신뢰해야 합니다. 검증은 설정 샌드박스가 아니며 의존성을 재귀적으로 검증하지 않습니다. 기존 프로필이나 설정 파일을 수정하지 않습니다.
+
+경로는 절대 경로여야 하며 `..` 구성 요소, 제어 문자, 경로 내 심볼릭 링크를 허용하지 않습니다. 선택한 파일은 현재 유효 사용자가 소유한 일반 파일이어야 하고, 하드 링크 수가 정확히 1개이며, 최대 64 KiB여야 합니다. 그룹·다른 사용자 쓰기 권한과 setuid/setgid/sticky 비트도 거부합니다. 상위 디렉터리는 root 또는 현재 유효 사용자 소유여야 하며 그룹·다른 사용자 쓰기가 금지되어야 합니다. 단, root 소유의 sticky 디렉터리는 예외입니다. Linux에서는 확장 접근 ACL과 디렉터리 기본 ACL을 거부하고, macOS에서는 allow ACL 항목과 소유권이 비활성화된 볼륨을 거부합니다. ACL이나 소유권을 확인할 수 없어도 거부합니다. 이는 보수적 검사이며 전체 유효 권한 해석기는 아닙니다.
+
+하나의 CLI 호출에서 같은 선택을 로컬 `ssh -G` 조회, doctor, 원격 목록·버전 확인, 업데이트 탐색·설치·검증, send/handoff SSH 호출에 재사용합니다. 매 SSH 실행 전에 파일 식별 정보·메타데이터·내용 해시와 저장 경로 안전성을 다시 확인합니다. 변경을 발견하면 기본 설정으로 대체하지 않고 거부합니다. OpenSSH는 원래 경로를 열기 때문에 불변 바이트 스냅샷이 아니며, 검증과 OpenSSH 읽기 사이에 같은 사용자가 변경하는 경쟁 조건은 남습니다.
+
+이 모드는 `StrictHostKeyChecking=yes`, `UpdateHostKeys=no`, `ForwardAgent=no`, `PermitLocalCommand=no`, `BatchMode=yes`를 강제하고 연결 다중화를 비활성화합니다. 기존 명시적 호스트 키 고정이 우선합니다. `--ssh-opt` 허용 목록은 그대로이며, 여기서 `-F`와 `ProxyCommand`는 계속 금지됩니다. 선택한 로컬 경로는 원격 명령·응답 경로·공개 JSON으로 전달되지 않습니다. 역방향 doctor 검사는 원격 장비 자체 설정을 사용하고 이 경로를 상속하지 않습니다. 정방향 성공은 역방향 인증, 메시지 소비, ACK를 입증하지 않습니다.
+
+환경 변수, MCP 요청, ReplyTo 주소, 저장된 프리셋에서는 이 모드를 선택할 수 없습니다. Windows, `setup`, 페어링된 기기 경로, 프리셋 조합, 응답 URI 대상, `--host` 없는 호출은 설정을 읽거나 SSH를 시작하기 전에 거부됩니다. 두 플래그를 생략하면 기존 기본 SSH 동작을 유지합니다.
+
### 업데이트
diff --git a/docs/zh-CN/cli-reference.md b/docs/zh-CN/cli-reference.md
index cd75c74..656b219 100644
--- a/docs/zh-CN/cli-reference.md
+++ b/docs/zh-CN/cli-reference.md
@@ -299,6 +299,25 @@ session-peer handoff watch --correlation-id PREPARED_UUID --watch-timeout 30 --i
观察为只读操作，不创建意图、等待、收据，不写入记录、提交或唤醒。不增加主机、设备、Windows 路径或 Codex ACK。每帧上限为 16 KiB，总量为 256 KiB，并预留 2 KiB 用于结束帧。真实文件描述符标准输出使用一个专属、隔离的 Python 输出辅助进程；它仅接收有界的已序列化公开字节，不更改共享描述符标志，也不读取记录或原生数据。父进程将 IPC 纳入原始总预算，开始新的记录读取或验证阶段前也检查期限；总时间内预留 100 ms 用于最终输出和清理，因此采样可能提前结束。超时、SIGINT 或错误时终止并回收辅助进程，但不可中断的 OS 操作不具备严格实时保证。部分写入或无完成回复后禁止后续帧、结束帧；消费方必须拒绝被截断的最后一行 NDJSON。辅助进程完成仅表示 OS 写入完成，不证明接收者消费。快照并非无损事件流，本地 UTC 观察不证明 ACK 到达顺序。有关隐私、保留的提交/ACK/等待含义、只读收集器查询及现有隔离限制，参见 [handoff 运行时指南](handoff-runtime.md)。观察不启动原生消息进程或收集器。此候选版本尚未发布。 不公开载荷或权限摘要，公开 targetGeneration 保持既有约定。 只读记录在读取不可变快照字节后、解析和验证前释放共享锁。 中断后的最终输出和清理在剩余总预算内最多使用 100 ms。 整体成功需要完整的成功结束帧和退出码 0：该帧之后清理仍可能失败。非零退出表示整体失败，不表示 ACK 升级、丢弃独立有效的提交事实或允许重试。
+### 可信本地 SSH 配置（v1.3 候选，尚未发布）
+
+在 Linux 和 macOS 上，显式指定 `--host` 的本地 CLI `list`、`send`、`doctor` 或 `update` 调用可以同时使用 `--ssh-config ABSOLUTE_PATH` 和 `--trust-ssh-config` 选择现有 OpenSSH 配置。两个标志缺一不可。以下为示例路径，支持用引号包裹的空格：
+
+```bash
+session-peer list --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config
+session-peer doctor --host worker --ssh-config '/home/alice/SSH Profiles/work.conf' --trust-ssh-config --check-return-route
+```
+
+此选择赋予配置本地代码执行权限：即使在 `ssh -G` 用户查询或 `send --dry-run` 中，`Match exec` 也可能执行。操作者还必须信任 `Include` 文件、`ProxyCommand` 及其他外部依赖。验证不是配置沙箱，也不会递归验证这些依赖。不会编辑任何现有配置档案或配置文件。
+
+路径必须是绝对路径，不得包含 `..` 路径段、控制字符或任何路径组件中的符号链接。所选文件必须是当前有效用户拥有的普通文件，硬链接数恰好为 1，大小不超过 64 KiB，且不得具有组或其他用户写权限以及 setuid/setgid/sticky 位。祖先目录须由 root 或当前有效用户拥有，且不得允许组或其他用户写入；root 拥有的 sticky 目录除外。Linux 拒绝扩展访问 ACL 和目录默认 ACL；macOS 拒绝 allow ACL 条目以及禁用所有权的卷。无法验证 ACL 或所有权时也会拒绝。这是保守检查，并非完整的有效权限解释器。
+
+同一次 CLI 调用会在本地 `ssh -G` 查询、doctor、远程列表与版本检查、更新探测/安装/验证，以及 send/handoff SSH 调用中复用同一选择。每次启动 SSH 前都会重新验证文件身份、元数据、内容哈希及存储安全性。检测到变更时拒绝操作，不回退至默认配置。OpenSSH 打开原始路径；这不是不可变字节快照，验证与 OpenSSH 读取之间仍可能发生同一用户修改文件的竞态。
+
+此模式强制 `StrictHostKeyChecking=yes`、`UpdateHostKeys=no`、`ForwardAgent=no`、`PermitLocalCommand=no` 和 `BatchMode=yes`，并禁用连接复用。现有显式主机密钥固定仍具有优先级。`--ssh-opt` 允许列表不变，其中的 `-F` 和 `ProxyCommand` 仍被禁止。所选本地路径不会传给远程命令、回复路由或公开 JSON。反向 doctor 使用远程机器自身的配置，绝不继承此路径。正向成功不能证明反向认证、消息消费或 ACK。
+
+环境变量、MCP 请求、ReplyTo 地址和已存储预设不能选择此模式。Windows、`setup`、配对设备路由、预设组合、回复 URI 目标以及未指定 `--host` 的调用，会在读取配置或启动 SSH 前被拒绝。省略这对标志时保留现有默认 SSH 行为。
+
### 更新
diff --git a/session_peer.py b/session_peer.py
index 4d66840..c6d457b 100644
--- a/session_peer.py
+++ b/session_peer.py
@@ -3204,6 +3204,348 @@ def reply_line(explicit_host: str | None) -> str | None:
return _reply_from_identity(sender_identity(explicit_host))
+"""Opt-in local SSH configuration validation; never a remote identity proof."""
+
+import hashlib
+from contextvars import ContextVar
+
+SSH_CONFIG_MAX_BYTES = 65536
+SSH_CONFIG_SELECTION = ContextVar("ssh_config_selection", default=None)
+
+def _ssh_config_error(reason):
+ return CcPeerError("trusted SSH configuration refused: " + reason,
+ {"reason": reason, "retryAllowed": False})
+
+
+def _ssh_config_cleanup(action, *args, preserve=False, **kwargs):
+ """A secondary ordinary cleanup error must not replace an active failure.
+
+ New KeyboardInterrupt/SystemExit are never swallowed. On a normal return,
+ cleanup failure is still reported, without reflecting raw diagnostics.
+ """
+ try:
+ return action(*args, **kwargs)
+ except Exception as error:
+ if not preserve:
+ if isinstance(error, CcPeerError):
+ raise
+ raise _ssh_config_error("ssh_config_cleanup_failed") from None
+
+
+@contextlib.contextmanager
+def _ssh_config_cleanup_scope(action, *args, **kwargs):
+ """Derive preservation from this body, never an ambient caller exception."""
+ try:
+ yield
+ except BaseException:
+ # A caller may already supply this policy for a nested cleanup. The
+ # active body's failure takes precedence without a duplicate keyword.
+ _ssh_config_cleanup(action, *args, **{**kwargs, "preserve": True})
+ raise
+ else:
+ _ssh_config_cleanup(action, *args, **kwargs)
+
+
+def _ssh_config_platform():
+ if IS_WINDOWS or fcntl is None or sys.platform not in ("darwin", "linux"):
+ raise _ssh_config_error("ssh_config_platform_unsupported")
+
+
+def _ssh_config_darwin_mount_flags(fd):
+ """Use the explicit statfs64 ABI, never the ambiguous legacy statfs ABI."""
+ import ctypes
+ try:
+ class Statfs64(ctypes.Structure):
+ # sys/mount.h __DARWIN_STRUCT_STATFS64. The reserved tail remains
+ # 32 bytes across the f_flags_ext addition; no string is inspected.
+ _fields_ = [
+ ("f_bsize", ctypes.c_uint32), ("f_iosize", ctypes.c_int32),
+ ("f_blocks", ctypes.c_uint64), ("f_bfree", ctypes.c_uint64),
+ ("f_bavail", ctypes.c_uint64), ("f_files", ctypes.c_uint64),
+ ("f_ffree", ctypes.c_uint64), ("f_fsid", ctypes.c_int32 * 2),
+ ("f_owner", ctypes.c_uint32), ("f_type", ctypes.c_uint32),
+ ("f_flags", ctypes.c_uint32), ("f_fssubtype", ctypes.c_uint32),
+ ("f_fstypename", ctypes.c_char * 16),
+ ("f_mntonname", ctypes.c_char * 1024),
+ ("f_mntfromname", ctypes.c_char * 1024),
+ ("f_reserved", ctypes.c_uint32 * 8),
+ ]
+ if (ctypes.sizeof(ctypes.c_void_p) != 8 or ctypes.sizeof(Statfs64) != 2168
+ or Statfs64.f_flags.offset != 64):
+ raise _ssh_config_error("ssh_config_mount_unverifiable")
+ api = ctypes.CDLL("/usr/lib/libSystem.B.dylib", use_errno=True)
+ api.fstatfs64.argtypes = [ctypes.c_int, ctypes.POINTER(Statfs64)]
+ api.fstatfs64.restype = ctypes.c_int
+ value = Statfs64()
+ if api.fstatfs64(fd, ctypes.byref(value)) != 0:
+ raise _ssh_config_error("ssh_config_mount_unverifiable")
+ return value.f_flags
+ except (OSError, AttributeError, TypeError, ValueError, OverflowError):
+ raise _ssh_config_error("ssh_config_mount_unverifiable") from None
+
+
+def _ssh_config_check_ownership(fd):
+ if sys.platform == "darwin" and _ssh_config_darwin_mount_flags(fd) & 0x00200000:
+ # MNT_IGNORE_OWNERSHIP can synthesize owner/mode metadata. Do not call
+ # such metadata an owner-only store, even if its ACL is empty.
+ raise _ssh_config_error("ssh_config_ownership_unenforced")
+
+
+def _ssh_config_check_fd(fd, *, private, directory=False):
+ _ssh_config_check_ownership(fd)
+ _ssh_config_check_acl(fd, private=private, directory=directory)
+
+
+def _ssh_config_darwin_acl_tags(fd):
+ """Read only Darwin's extended ACL, without paths or a child process.
+
+ Darwin returns 0 for an entry, -1/EINVAL at exhaustion (unlike Linux).
+ Validate the independent ACL handle first; release it on every path.
+ Constants and prototypes are from the macOS SDK sys/acl.h.
+ """
+ import ctypes
+ try:
+ api = ctypes.CDLL("/usr/lib/libSystem.B.dylib", use_errno=True)
+ api.acl_get_fd_np.argtypes = [ctypes.c_int, ctypes.c_int]
+ api.acl_get_fd_np.restype = ctypes.c_void_p
+ api.acl_valid.argtypes = [ctypes.c_void_p]
+ api.acl_valid.restype = ctypes.c_int
+ api.acl_get_entry.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(ctypes.c_void_p)]
+ api.acl_get_entry.restype = ctypes.c_int
+ api.acl_get_tag_type.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_int)]
+ api.acl_get_tag_type.restype = ctypes.c_int
+ api.acl_free.argtypes = [ctypes.c_void_p]
+ api.acl_free.restype = ctypes.c_int
+ ctypes.set_errno(0)
+ acl = api.acl_get_fd_np(fd, 0x100) # ACL_TYPE_EXTENDED
+ if not acl:
+ # Darwin's fd getter returns NULL/ENOENT when FILESEC_ACL is absent
+ # (Libc posix1e/acl_file.c and gen/filesec.c), not an empty object.
+ if ctypes.get_errno() == errno.ENOENT:
+ return ()
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ acl_failed = False
+ try:
+ if api.acl_valid(acl) != 0:
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ tags = []
+ for index in range(129):
+ entry = ctypes.c_void_p()
+ ctypes.set_errno(0)
+ result = api.acl_get_entry(acl, 0 if index == 0 else -1, ctypes.byref(entry))
+ if result == -1 and ctypes.get_errno() == errno.EINVAL:
+ return tuple(tags)
+ if result != 0 or not entry.value or index == 128:
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ tag = ctypes.c_int()
+ if api.acl_get_tag_type(entry, ctypes.byref(tag)) != 0 or tag.value not in (1, 2):
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ tags.append(tag.value) # ACL_EXTENDED_ALLOW=1, DENY=2
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ except BaseException:
+ acl_failed = True
+ raise
+ finally:
+ def release_acl():
+ if api.acl_free(acl) != 0:
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ _ssh_config_cleanup(release_acl, preserve=acl_failed)
+ except (OSError, AttributeError, TypeError, ValueError, OverflowError):
+ raise _ssh_config_error("ssh_config_acl_unverifiable") from None
+
+
+def _ssh_config_check_acl(fd, *, private, directory=False):
+ """Conservative ACL policy, not an effective-permissions interpreter."""
+ if sys.platform == "darwin":
+ tags = _ssh_config_darwin_acl_tags(fd)
+ if any(tag not in (1, 2) for tag in tags):
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ if (private and tags) or (not private and 1 in tags):
+ raise _ssh_config_error("ssh_config_unsafe_acl")
+ return
+ if sys.platform == "linux":
+ try:
+ for attribute in (("system.posix_acl_access", "system.posix_acl_default")
+ if directory else ("system.posix_acl_access",)):
+ try:
+ os.getxattr(fd, attribute)
+ except OSError as error:
+ if error.errno == errno.ENODATA:
+ continue
+ raise _ssh_config_error("ssh_config_acl_unverifiable") from None
+ raise _ssh_config_error("ssh_config_unsafe_acl")
+ except (AttributeError, TypeError, ValueError):
+ raise _ssh_config_error("ssh_config_acl_unverifiable") from None
+ return
+ raise _ssh_config_error("ssh_config_platform_unsupported")
+
+
+
+def _ssh_config_path(path):
+ if (not isinstance(path, str) or not path or len(path) > 4096
+ or not path.startswith("/") or path.startswith("//")
+ or ".." in path.split("/")
+ or any(ord(c) < 32 or 127 <= ord(c) <= 159
+ or 0xD800 <= ord(c) <= 0xDFFF for c in path)):
+ raise _ssh_config_error("ssh_config_invalid_path")
+ return path
+
+
+def _ssh_config_overflow_uid():
+ if sys.platform != "linux":
+ return None
+ try:
+ with open("/proc/sys/kernel/overflowuid", "rb") as stream:
+ raw = stream.read(32)
+ if not raw.strip().isdigit() or len(raw) == 32:
+ raise ValueError()
+ return int(raw)
+ except (OSError, ValueError):
+ raise _ssh_config_error("ssh_config_ownership_unverifiable") from None
+
+
+def _ssh_config_identity(info):
+ return (info.st_dev, info.st_ino, info.st_mode, info.st_uid,
+ info.st_gid, info.st_nlink, info.st_size,
+ info.st_mtime_ns, info.st_ctime_ns)
+
+
+def _ssh_config_metadata(info, overflow_uid, *, directory=False):
+ if overflow_uid is not None and info.st_uid == overflow_uid:
+ raise _ssh_config_error("ssh_config_ownership_unverifiable")
+ if directory:
+ if (not stat.S_ISDIR(info.st_mode) or info.st_uid not in (0, os.geteuid())
+ or (info.st_mode & 0o022
+ and not (info.st_uid == 0 and info.st_mode & stat.S_ISVTX))):
+ raise _ssh_config_error("ssh_config_unsafe_storage")
+ elif (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid()
+ or info.st_nlink != 1 or info.st_mode & 0o7022):
+ raise _ssh_config_error("ssh_config_unsafe_storage")
+
+
+def _ssh_config_snapshot(path):
+ _ssh_config_platform()
+ _ssh_config_path(path)
+ overflow_uid = _ssh_config_overflow_uid()
+ directory = fd = None
+ ancestors = []
+ failed = False
+ try:
+ flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK
+ directory = os.open("/", flags | os.O_DIRECTORY)
+ parts = path.split("/")[1:]
+ # Reject trailing slash / empty filename, but harmless internal repeated
+ # separators do not change the original path passed to OpenSSH.
+ if not parts[-1]:
+ raise _ssh_config_error("ssh_config_invalid_path")
+ for part in parts[:-1]:
+ info = os.fstat(directory)
+ _ssh_config_metadata(info, overflow_uid, directory=True)
+ _ssh_config_check_fd(directory, private=False, directory=True)
+ if not part:
+ continue
+ ancestors.append((info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid))
+ child = os.open(part, flags | os.O_DIRECTORY, dir_fd=directory)
+ previous, directory = directory, child
+ _ssh_config_cleanup(os.close, previous)
+ info = os.fstat(directory)
+ _ssh_config_metadata(info, overflow_uid, directory=True)
+ ancestors.append((info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid))
+ _ssh_config_check_fd(directory, private=False, directory=True)
+ fd = os.open(parts[-1], flags, dir_fd=directory)
+ before = os.fstat(fd)
+ _ssh_config_metadata(before, overflow_uid)
+ _ssh_config_check_fd(fd, private=False)
+ if before.st_size > SSH_CONFIG_MAX_BYTES:
+ raise _ssh_config_error("ssh_config_capacity")
+ raw = bytearray()
+ while len(raw) <= SSH_CONFIG_MAX_BYTES:
+ chunk = os.read(fd, min(8192, SSH_CONFIG_MAX_BYTES + 1 - len(raw)))
+ if not chunk:
+ break
+ raw.extend(chunk)
+ if len(raw) > SSH_CONFIG_MAX_BYTES:
+ raise _ssh_config_error("ssh_config_capacity")
+ after = os.fstat(fd)
+ _ssh_config_metadata(after, overflow_uid)
+ _ssh_config_check_fd(fd, private=False)
+ current = os.stat(parts[-1], dir_fd=directory, follow_symlinks=False)
+ if (_ssh_config_identity(before) != _ssh_config_identity(after)
+ or _ssh_config_identity(after) != _ssh_config_identity(current)):
+ raise _ssh_config_error("ssh_config_storage_changed")
+ return tuple(ancestors), _ssh_config_identity(after), hashlib.sha256(raw).digest()
+ except OSError:
+ failed = True
+ raise _ssh_config_error("ssh_config_unsafe_storage") from None
+ except BaseException:
+ failed = True
+ raise
+ finally:
+ def close_directory():
+ if directory is not None:
+ os.close(directory)
+ with _ssh_config_cleanup_scope(
+ _ssh_config_cleanup, close_directory, preserve=failed):
+ if fd is not None:
+ _ssh_config_cleanup(os.close, fd, preserve=failed)
+
+
+class _TrustedSshConfig:
+ # Source-streamed consumers may execute without a registered module entry;
+ # dataclass annotation resolution on supported older Python cannot do so.
+ __slots__ = ("_path", "_snapshot")
+
+ def __init__(self, path, snapshot):
+ object.__setattr__(self, "_path", path)
+ object.__setattr__(self, "_snapshot", snapshot)
+
+ @property
+ def path(self):
+ return self._path
+
+ def __setattr__(self, name, value):
+ raise AttributeError("trusted SSH selection is immutable")
+
+ def __delattr__(self, name):
+ raise AttributeError("trusted SSH selection is immutable")
+
+ def __repr__(self):
+ return ""
+
+ def revalidate(self):
+ if _ssh_config_snapshot(self.path) != self._snapshot:
+ raise _ssh_config_error("ssh_config_storage_changed")
+
+
+def select_trusted_ssh_config(path):
+ """Validate without executing config, resolving Includes or making copies.
+
+ Revalidation detects changes since selection. OpenSSH subsequently reopens
+ the original path: same-user races and executable config remain user trust.
+ """
+ return _TrustedSshConfig(path, _ssh_config_snapshot(path))
+
+
+def trusted_ssh_command(ssh_opts, host, *remote, prefix=(), identity_options=()):
+ selected = SSH_CONFIG_SELECTION.get()
+ safety = []
+ if selected is not None:
+ selected.revalidate()
+ safety = [
+ "-oStrictHostKeyChecking=yes", "-oUpdateHostKeys=no",
+ "-oForwardAgent=no", "-oPermitLocalCommand=no", "-oBatchMode=yes",
+ "-oControlMaster=no", "-oControlPath=none", "-oControlPersist=no",
+ "-F", selected.path,
+ ]
+ return ["ssh", *prefix, *identity_options, *safety, *ssh_opts, host, *remote]
+
+
+def redact_ssh_config_path(text):
+ """Mask the selected local path; not a general secret/content sanitizer."""
+ selected = SSH_CONFIG_SELECTION.get()
+ return text.replace(selected.path, "") if selected is not None else text
+
+
# --------------------------------------------------------------------------
# Remote dispatch. Ships this file over SSH and runs it there, so the remote
# machine needs nothing installed beyond python3.
@@ -3308,7 +3650,7 @@ def ssh_user_metadata(host: str, ssh_opts: list[str], *, handoff_budget=None) ->
try:
completed = subprocess.run(
- ["ssh", "-G", *ssh_opts, host], capture_output=True,
+ trusted_ssh_command(ssh_opts, host, prefix=("-G",)), capture_output=True,
encoding="utf-8", errors="replace", timeout=DETECT_TIMEOUT,
)
except (OSError, subprocess.SubprocessError):
@@ -3364,7 +3706,7 @@ def ssh_failure_error(host: str, ssh_info: dict, failure: str,
elif failure == "timeout":
message = f"SSH connection to {target} timed out"
else:
- suffix = f": {detail.strip()[:2000]}" if detail and detail.strip() else ""
+ suffix = f": {redact_ssh_config_path(detail.strip())[:2000]}" if detail and detail.strip() else ""
message = f"SSH transport failed for {target}{suffix}"
return CcPeerError(message, {**ssh_info, "sshFailure": failure})
@@ -3521,7 +3863,7 @@ def ssh_handoff_configuration(host, ssh_opts, handoff_budget):
completed = None
try:
probe_cutoff = min(cutoff, handoff_now() + DETECT_TIMEOUT)
- completed = handoff_stream_child(["ssh", "-G", *ssh_opts, host], b"", probe_cutoff,
+ completed = handoff_stream_child(trusted_ssh_command(ssh_opts, host, prefix=("-G",)), b"", probe_cutoff,
min(total, probe_cutoff + 5), handoff_now)
valid = (completed.returncode in (0, -signal.SIGKILL) and completed.reason is None
and not completed.interrupted and not completed.stdout_overflow
@@ -3584,7 +3926,7 @@ def _run_remote_handoff_dispatch(host, argv, ssh_opts, identity_options, budget,
raise CcPeerError("Cannot read source for private SSH handoff",
{"reason": "ssh_handoff_source_unavailable", "retryAllowed": False, "spawned": False}) from exc
ssh_info = ssh_user_metadata(host, ssh_opts, handoff_budget=budget)
- completed = handoff_stream_child(["ssh", *identity_options, *ssh_opts, host, remote],
+ completed = handoff_stream_child(trusted_ssh_command(ssh_opts, host, remote, identity_options=identity_options),
source, *budget, handoff_now)
if context["phase"] == "probe" and completed.interrupted:
# Probe success is not effect evidence. Operator cancellation must
@@ -3641,7 +3983,7 @@ def _run_remote_dispatch(host: str, argv: list[str], ssh_opts: list[str], *, ide
# result to the remote *shell*, so an argv list is not the protection it
# looks like: a metacharacter in any element executes over there. Build
# the remote command as one already-quoted string instead.
- command = ["ssh", *identity_options, *ssh_opts, host, remote]
+ command = trusted_ssh_command(ssh_opts, host, remote, identity_options=identity_options)
def incomplete_response_error(message: str, details: dict | None = None) -> CcPeerError:
metadata = {**ssh_info, **(details or {})}
@@ -3695,7 +4037,7 @@ def incomplete_response_error(message: str, details: dict | None = None) -> CcPe
if isinstance(stderr, bytes):
stderr = stderr.decode("utf-8", errors="replace")
stderr = stderr if isinstance(stderr, str) else ""
- detail = stderr.strip() or f"ssh exited {completed.returncode}"
+ detail = redact_ssh_config_path(stderr.strip()) or f"ssh exited {completed.returncode}"
# A complete response proves the remote command ran, even if the SSH
# process later exits 255. Login-shell stderr must not hide submission
# evidence or turn a remote partial failure into an invitation to resend.
@@ -3726,7 +4068,7 @@ def incomplete_response_error(message: str, details: dict | None = None) -> CcPe
if not parsed_result or (isinstance(result, dict)
and "schemaVersion" in result and not valid_result) or (
argv and argv[0] == "send" and not valid_result and not legacy_send_result):
- raise incomplete_response_error(f"{host}: unexpected output: {stdout[:200]}")
+ raise incomplete_response_error(f"{host}: unexpected output: {redact_ssh_config_path(stdout)[:200]}")
if isinstance(result, dict):
result.update(ssh_info)
@@ -3768,7 +4110,7 @@ def incomplete_response_error(message: str, details: dict | None = None) -> CcPe
elif "codexHomeResolution" in result:
details["codexHomeResolution"] = result["codexHomeResolution"]
raise CcPeerError(
- f"{host}: {result.get('error', 'remote command failed')}", details
+ f"{host}: {redact_ssh_config_path(str(result.get('error', 'remote command failed')))}", details
)
return result
@@ -4020,7 +4362,7 @@ def owns_publication():
f'python3 -c {shlex.quote(publisher)} "$T/session_peer.py" '
f'"$D/session_peer.py" "$B/session-peer" {shlex.quote(expected_line)}'
)
- command = ["ssh", *ssh_opts, host, remote_script]
+ command = trusted_ssh_command(ssh_opts, host, remote_script)
uncertain = {**ssh_info, "commitStatus": "unknown", "retryAllowed": False}
def push_failure(failure: str, detail: str | None = None,
@@ -4063,7 +4405,7 @@ def reject_receipt(output: str | bytes | None) -> None:
failure = classify_ssh_failure(detail, completed.returncode)
if failure:
raise push_failure(failure, detail)
- raise CcPeerError(f"{host}: {detail}", uncertain)
+ raise CcPeerError(f"{host}: {redact_ssh_config_path(detail)}", uncertain)
raise CcPeerError(f"{host}: installed version did not match {__version__}", uncertain)
@@ -4130,7 +4472,7 @@ def ssh_identity_configuration(host: str, ssh_opts: list[str], *, handoff_budget
try:
completed = (ssh_handoff_configuration(host, ssh_opts, handoff_budget)
if handoff_budget is not None else subprocess.run(
- ["ssh", "-G", *ssh_opts, host], capture_output=True, timeout=DETECT_TIMEOUT))
+ trusted_ssh_command(ssh_opts, host, prefix=("-G",)), capture_output=True, timeout=DETECT_TIMEOUT))
output = completed.stdout
if isinstance(output, bytes):
if len(output) > 1024 * 1024:
@@ -4612,6 +4954,16 @@ def diagnose_codex(args: argparse.Namespace) -> dict:
def probe_return_route(destination: str) -> dict:
+ # A reverse probe belongs to the machine running this doctor. A caller's
+ # forward selection is never inherited, even by an in-process invocation.
+ token = SSH_CONFIG_SELECTION.set(None)
+ try:
+ return _probe_return_route(destination)
+ finally:
+ SSH_CONFIG_SELECTION.reset(token)
+
+
+def _probe_return_route(destination: str) -> dict:
"""Test reverse SSH without prompts, key enrollment, or config mutation."""
check_ssh_argument(destination, "--host")
if is_self_ssh_destination(destination):
@@ -5980,7 +6332,7 @@ def remote_installed_version(host: str, ssh_opts: list[str],
)
try:
done = subprocess.run(
- ["ssh", *ssh_opts, host, probe],
+ trusted_ssh_command(ssh_opts, host, probe),
capture_output=True, timeout=30,
)
except FileNotFoundError as exc:
@@ -9970,6 +10322,38 @@ def error(self, message):
super().error(human_text(message))
+class SshConfigArgument(argparse.Action):
+ def __call__(self, parser, namespace, values, option_string=None):
+ if getattr(namespace, self.dest, None) not in (None, False):
+ parser.error(self.option_strings[0] + " may be specified only once")
+ setattr(namespace, self.dest, self.const if self.nargs == 0 else values)
+
+
+def prepare_trusted_ssh_config(parser, args):
+ path = getattr(args, "ssh_config", None)
+ trust = getattr(args, "trust_ssh_config", False)
+ if path is None and not trust:
+ return
+ if path is None or not trust:
+ parser.error("--ssh-config and --trust-ssh-config must be supplied together")
+ if args.command not in ("list", "send", "doctor", "update") or not getattr(args, "host", None):
+ parser.error("--ssh-config requires list/send/doctor/update with explicit --host")
+ if getattr(args, "device", None) or getattr(args, "preset", None) is not None:
+ parser.error("--ssh-config cannot be combined with paired devices or route presets")
+ if str(getattr(args, "to", "")).lower().startswith("session-peer:"):
+ parser.error("--ssh-config cannot be selected through a Reply-To invocation")
+ for host in args.host:
+ check_ssh_argument(host, "--host")
+ check_ssh_options(getattr(args, "ssh_opt", []))
+ if IS_WINDOWS:
+ raise _ssh_config_error("ssh_config_platform_unsupported")
+ if args.command == "send" and handoff_requested(args):
+ # Configuration setup belongs to the ORIGINAL handoff budget; no
+ # metadata/effect child receives a renewed deadline or this local path.
+ args._handoff_fixed_total = handoff_now() + args.wait_timeout
+ SSH_CONFIG_SELECTION.set(select_trusted_ssh_config(path))
+
+
def build_parser() -> argparse.ArgumentParser:
parser = HumanArgumentParser(
prog="session-peer",
@@ -10122,6 +10506,13 @@ def add_common(sub: argparse.ArgumentParser) -> None:
)
updating.set_defaults(func=cmd_update)
+ # Deliberately not add_common: setup must not gain executable config input.
+ for sub in (listing, sending, doctor, updating):
+ sub.add_argument("--ssh-config", action=SshConfigArgument, metavar="ABS_PATH",
+ help="operator-trusted local executable SSH configuration; requires --trust-ssh-config")
+ sub.add_argument("--trust-ssh-config", action=SshConfigArgument, nargs=0, const=True, default=False,
+ help="authorize this configuration's local commands, including during ssh -G and dry-run")
+
for sub in (listing, sending, doctor):
sub.add_argument('--antigravity-home', help='filter registered Antigravity home on the destination')
sending.add_argument('--antigravity-generation', help='require this registered bridge generation')
@@ -10187,6 +10578,16 @@ def json_error_result(args: argparse.Namespace, payload: dict) -> dict | list[di
def main(argv: list[str] | None = None) -> int:
+ # Invocation-local authority, never an environment setting, MCP default or
+ # leaked state from a previous/reentrant call in the same interpreter.
+ token = SSH_CONFIG_SELECTION.set(None)
+ try:
+ return _main(argv)
+ finally:
+ SSH_CONFIG_SELECTION.reset(token)
+
+
+def _main(argv: list[str] | None = None) -> int:
global _CLIENT_UPDATE_NOTICE, _SKILL_UPDATE_NOTICES
cli_invocation = argv is None
raw_argv = list(sys.argv[1:] if argv is None else argv)
@@ -10228,6 +10629,16 @@ def main(argv: list[str] | None = None) -> int:
if args.json and output_format == "text":
parser.error("--json conflicts with --output-format text")
args.json = args.json or output_format == "json"
+ try:
+ prepare_trusted_ssh_config(parser, args)
+ except CcPeerError as exc:
+ if args.json:
+ print(json.dumps(json_error_result(args, {"error": str(exc), **exc.details}), ensure_ascii=False))
+ else:
+ print(human_text(f"session-peer: {exc}"), file=sys.stderr)
+ return EXIT_ERROR
+ except KeyboardInterrupt:
+ return 130
try:
_CLIENT_UPDATE_NOTICE = prepare_client_update(args) if cli_invocation else None
except Exception:
@@ -10243,7 +10654,7 @@ def main(argv: list[str] | None = None) -> int:
validate_ssh_identity_options(args)
exit_code = args.func(args)
except CcPeerError as exc:
- message = str(exc)
+ message = redact_ssh_config_path(str(exc))
if args.json:
payload = json_error_result(args, {"error": message, **exc.details})
print(json.dumps(with_client_update(payload), ensure_ascii=False))
@@ -10258,7 +10669,7 @@ def main(argv: list[str] | None = None) -> int:
show_human_notice = False
exit_code = 130
except Exception as exc:
- message = f"{type(exc).__name__}: {exc}"
+ message = redact_ssh_config_path(f"{type(exc).__name__}: {exc}")
if getattr(args, "json", False):
payload = json_error_result(args, {"error": message})
print(json.dumps(with_client_update(payload), ensure_ascii=False))
diff --git a/session_peer_core/cli.py b/session_peer_core/cli.py
index 91d53d7..88a1253 100644
--- a/session_peer_core/cli.py
+++ b/session_peer_core/cli.py
@@ -17,6 +17,38 @@ def error(self, message):
super().error(human_text(message))
+class SshConfigArgument(argparse.Action):
+ def __call__(self, parser, namespace, values, option_string=None):
+ if getattr(namespace, self.dest, None) not in (None, False):
+ parser.error(self.option_strings[0] + " may be specified only once")
+ setattr(namespace, self.dest, self.const if self.nargs == 0 else values)
+
+
+def prepare_trusted_ssh_config(parser, args):
+ path = getattr(args, "ssh_config", None)
+ trust = getattr(args, "trust_ssh_config", False)
+ if path is None and not trust:
+ return
+ if path is None or not trust:
+ parser.error("--ssh-config and --trust-ssh-config must be supplied together")
+ if args.command not in ("list", "send", "doctor", "update") or not getattr(args, "host", None):
+ parser.error("--ssh-config requires list/send/doctor/update with explicit --host")
+ if getattr(args, "device", None) or getattr(args, "preset", None) is not None:
+ parser.error("--ssh-config cannot be combined with paired devices or route presets")
+ if str(getattr(args, "to", "")).lower().startswith("session-peer:"):
+ parser.error("--ssh-config cannot be selected through a Reply-To invocation")
+ for host in args.host:
+ check_ssh_argument(host, "--host")
+ check_ssh_options(getattr(args, "ssh_opt", []))
+ if IS_WINDOWS:
+ raise _ssh_config_error("ssh_config_platform_unsupported")
+ if args.command == "send" and handoff_requested(args):
+ # Configuration setup belongs to the ORIGINAL handoff budget; no
+ # metadata/effect child receives a renewed deadline or this local path.
+ args._handoff_fixed_total = handoff_now() + args.wait_timeout
+ SSH_CONFIG_SELECTION.set(select_trusted_ssh_config(path))
+
+
def build_parser() -> argparse.ArgumentParser:
parser = HumanArgumentParser(
prog="session-peer",
@@ -169,6 +201,13 @@ def add_common(sub: argparse.ArgumentParser) -> None:
)
updating.set_defaults(func=cmd_update)
+ # Deliberately not add_common: setup must not gain executable config input.
+ for sub in (listing, sending, doctor, updating):
+ sub.add_argument("--ssh-config", action=SshConfigArgument, metavar="ABS_PATH",
+ help="operator-trusted local executable SSH configuration; requires --trust-ssh-config")
+ sub.add_argument("--trust-ssh-config", action=SshConfigArgument, nargs=0, const=True, default=False,
+ help="authorize this configuration's local commands, including during ssh -G and dry-run")
+
for sub in (listing, sending, doctor):
sub.add_argument('--antigravity-home', help='filter registered Antigravity home on the destination')
sending.add_argument('--antigravity-generation', help='require this registered bridge generation')
@@ -234,6 +273,16 @@ def json_error_result(args: argparse.Namespace, payload: dict) -> dict | list[di
def main(argv: list[str] | None = None) -> int:
+ # Invocation-local authority, never an environment setting, MCP default or
+ # leaked state from a previous/reentrant call in the same interpreter.
+ token = SSH_CONFIG_SELECTION.set(None)
+ try:
+ return _main(argv)
+ finally:
+ SSH_CONFIG_SELECTION.reset(token)
+
+
+def _main(argv: list[str] | None = None) -> int:
global _CLIENT_UPDATE_NOTICE, _SKILL_UPDATE_NOTICES
cli_invocation = argv is None
raw_argv = list(sys.argv[1:] if argv is None else argv)
@@ -275,6 +324,16 @@ def main(argv: list[str] | None = None) -> int:
if args.json and output_format == "text":
parser.error("--json conflicts with --output-format text")
args.json = args.json or output_format == "json"
+ try:
+ prepare_trusted_ssh_config(parser, args)
+ except CcPeerError as exc:
+ if args.json:
+ print(json.dumps(json_error_result(args, {"error": str(exc), **exc.details}), ensure_ascii=False))
+ else:
+ print(human_text(f"session-peer: {exc}"), file=sys.stderr)
+ return EXIT_ERROR
+ except KeyboardInterrupt:
+ return 130
try:
_CLIENT_UPDATE_NOTICE = prepare_client_update(args) if cli_invocation else None
except Exception:
@@ -290,7 +349,7 @@ def main(argv: list[str] | None = None) -> int:
validate_ssh_identity_options(args)
exit_code = args.func(args)
except CcPeerError as exc:
- message = str(exc)
+ message = redact_ssh_config_path(str(exc))
if args.json:
payload = json_error_result(args, {"error": message, **exc.details})
print(json.dumps(with_client_update(payload), ensure_ascii=False))
@@ -305,7 +364,7 @@ def main(argv: list[str] | None = None) -> int:
show_human_notice = False
exit_code = 130
except Exception as exc:
- message = f"{type(exc).__name__}: {exc}"
+ message = redact_ssh_config_path(f"{type(exc).__name__}: {exc}")
if getattr(args, "json", False):
payload = json_error_result(args, {"error": message})
print(json.dumps(with_client_update(payload), ensure_ascii=False))
diff --git a/session_peer_core/diagnostics.py b/session_peer_core/diagnostics.py
index 2c3f386..70bc822 100644
--- a/session_peer_core/diagnostics.py
+++ b/session_peer_core/diagnostics.py
@@ -185,6 +185,16 @@ def diagnose_codex(args: argparse.Namespace) -> dict:
def probe_return_route(destination: str) -> dict:
+ # A reverse probe belongs to the machine running this doctor. A caller's
+ # forward selection is never inherited, even by an in-process invocation.
+ token = SSH_CONFIG_SELECTION.set(None)
+ try:
+ return _probe_return_route(destination)
+ finally:
+ SSH_CONFIG_SELECTION.reset(token)
+
+
+def _probe_return_route(destination: str) -> dict:
"""Test reverse SSH without prompts, key enrollment, or config mutation."""
check_ssh_argument(destination, "--host")
if is_self_ssh_destination(destination):
diff --git a/session_peer_core/ssh.py b/session_peer_core/ssh.py
index 28ff630..af24959 100644
--- a/session_peer_core/ssh.py
+++ b/session_peer_core/ssh.py
@@ -102,7 +102,7 @@ def ssh_user_metadata(host: str, ssh_opts: list[str], *, handoff_budget=None) ->
try:
completed = subprocess.run(
- ["ssh", "-G", *ssh_opts, host], capture_output=True,
+ trusted_ssh_command(ssh_opts, host, prefix=("-G",)), capture_output=True,
encoding="utf-8", errors="replace", timeout=DETECT_TIMEOUT,
)
except (OSError, subprocess.SubprocessError):
@@ -158,7 +158,7 @@ def ssh_failure_error(host: str, ssh_info: dict, failure: str,
elif failure == "timeout":
message = f"SSH connection to {target} timed out"
else:
- suffix = f": {detail.strip()[:2000]}" if detail and detail.strip() else ""
+ suffix = f": {redact_ssh_config_path(detail.strip())[:2000]}" if detail and detail.strip() else ""
message = f"SSH transport failed for {target}{suffix}"
return CcPeerError(message, {**ssh_info, "sshFailure": failure})
@@ -315,7 +315,7 @@ def ssh_handoff_configuration(host, ssh_opts, handoff_budget):
completed = None
try:
probe_cutoff = min(cutoff, handoff_now() + DETECT_TIMEOUT)
- completed = handoff_stream_child(["ssh", "-G", *ssh_opts, host], b"", probe_cutoff,
+ completed = handoff_stream_child(trusted_ssh_command(ssh_opts, host, prefix=("-G",)), b"", probe_cutoff,
min(total, probe_cutoff + 5), handoff_now)
valid = (completed.returncode in (0, -signal.SIGKILL) and completed.reason is None
and not completed.interrupted and not completed.stdout_overflow
@@ -378,7 +378,7 @@ def _run_remote_handoff_dispatch(host, argv, ssh_opts, identity_options, budget,
raise CcPeerError("Cannot read source for private SSH handoff",
{"reason": "ssh_handoff_source_unavailable", "retryAllowed": False, "spawned": False}) from exc
ssh_info = ssh_user_metadata(host, ssh_opts, handoff_budget=budget)
- completed = handoff_stream_child(["ssh", *identity_options, *ssh_opts, host, remote],
+ completed = handoff_stream_child(trusted_ssh_command(ssh_opts, host, remote, identity_options=identity_options),
source, *budget, handoff_now)
if context["phase"] == "probe" and completed.interrupted:
# Probe success is not effect evidence. Operator cancellation must
@@ -435,7 +435,7 @@ def _run_remote_dispatch(host: str, argv: list[str], ssh_opts: list[str], *, ide
# result to the remote *shell*, so an argv list is not the protection it
# looks like: a metacharacter in any element executes over there. Build
# the remote command as one already-quoted string instead.
- command = ["ssh", *identity_options, *ssh_opts, host, remote]
+ command = trusted_ssh_command(ssh_opts, host, remote, identity_options=identity_options)
def incomplete_response_error(message: str, details: dict | None = None) -> CcPeerError:
metadata = {**ssh_info, **(details or {})}
@@ -489,7 +489,7 @@ def incomplete_response_error(message: str, details: dict | None = None) -> CcPe
if isinstance(stderr, bytes):
stderr = stderr.decode("utf-8", errors="replace")
stderr = stderr if isinstance(stderr, str) else ""
- detail = stderr.strip() or f"ssh exited {completed.returncode}"
+ detail = redact_ssh_config_path(stderr.strip()) or f"ssh exited {completed.returncode}"
# A complete response proves the remote command ran, even if the SSH
# process later exits 255. Login-shell stderr must not hide submission
# evidence or turn a remote partial failure into an invitation to resend.
@@ -520,7 +520,7 @@ def incomplete_response_error(message: str, details: dict | None = None) -> CcPe
if not parsed_result or (isinstance(result, dict)
and "schemaVersion" in result and not valid_result) or (
argv and argv[0] == "send" and not valid_result and not legacy_send_result):
- raise incomplete_response_error(f"{host}: unexpected output: {stdout[:200]}")
+ raise incomplete_response_error(f"{host}: unexpected output: {redact_ssh_config_path(stdout)[:200]}")
if isinstance(result, dict):
result.update(ssh_info)
@@ -562,7 +562,7 @@ def incomplete_response_error(message: str, details: dict | None = None) -> CcPe
elif "codexHomeResolution" in result:
details["codexHomeResolution"] = result["codexHomeResolution"]
raise CcPeerError(
- f"{host}: {result.get('error', 'remote command failed')}", details
+ f"{host}: {redact_ssh_config_path(str(result.get('error', 'remote command failed')))}", details
)
return result
@@ -814,7 +814,7 @@ def owns_publication():
f'python3 -c {shlex.quote(publisher)} "$T/session_peer.py" '
f'"$D/session_peer.py" "$B/session-peer" {shlex.quote(expected_line)}'
)
- command = ["ssh", *ssh_opts, host, remote_script]
+ command = trusted_ssh_command(ssh_opts, host, remote_script)
uncertain = {**ssh_info, "commitStatus": "unknown", "retryAllowed": False}
def push_failure(failure: str, detail: str | None = None,
@@ -857,6 +857,6 @@ def reject_receipt(output: str | bytes | None) -> None:
failure = classify_ssh_failure(detail, completed.returncode)
if failure:
raise push_failure(failure, detail)
- raise CcPeerError(f"{host}: {detail}", uncertain)
+ raise CcPeerError(f"{host}: {redact_ssh_config_path(detail)}", uncertain)
raise CcPeerError(f"{host}: installed version did not match {__version__}", uncertain)
diff --git a/session_peer_core/ssh_config.py b/session_peer_core/ssh_config.py
new file mode 100644
index 0000000..317caec
--- /dev/null
+++ b/session_peer_core/ssh_config.py
@@ -0,0 +1,340 @@
+"""Opt-in local SSH configuration validation; never a remote identity proof."""
+
+import hashlib
+from contextvars import ContextVar
+
+SSH_CONFIG_MAX_BYTES = 65536
+SSH_CONFIG_SELECTION = ContextVar("ssh_config_selection", default=None)
+
+def _ssh_config_error(reason):
+ return CcPeerError("trusted SSH configuration refused: " + reason,
+ {"reason": reason, "retryAllowed": False})
+
+
+def _ssh_config_cleanup(action, *args, preserve=False, **kwargs):
+ """A secondary ordinary cleanup error must not replace an active failure.
+
+ New KeyboardInterrupt/SystemExit are never swallowed. On a normal return,
+ cleanup failure is still reported, without reflecting raw diagnostics.
+ """
+ try:
+ return action(*args, **kwargs)
+ except Exception as error:
+ if not preserve:
+ if isinstance(error, CcPeerError):
+ raise
+ raise _ssh_config_error("ssh_config_cleanup_failed") from None
+
+
+@contextlib.contextmanager
+def _ssh_config_cleanup_scope(action, *args, **kwargs):
+ """Derive preservation from this body, never an ambient caller exception."""
+ try:
+ yield
+ except BaseException:
+ # A caller may already supply this policy for a nested cleanup. The
+ # active body's failure takes precedence without a duplicate keyword.
+ _ssh_config_cleanup(action, *args, **{**kwargs, "preserve": True})
+ raise
+ else:
+ _ssh_config_cleanup(action, *args, **kwargs)
+
+
+def _ssh_config_platform():
+ if IS_WINDOWS or fcntl is None or sys.platform not in ("darwin", "linux"):
+ raise _ssh_config_error("ssh_config_platform_unsupported")
+
+
+def _ssh_config_darwin_mount_flags(fd):
+ """Use the explicit statfs64 ABI, never the ambiguous legacy statfs ABI."""
+ import ctypes
+ try:
+ class Statfs64(ctypes.Structure):
+ # sys/mount.h __DARWIN_STRUCT_STATFS64. The reserved tail remains
+ # 32 bytes across the f_flags_ext addition; no string is inspected.
+ _fields_ = [
+ ("f_bsize", ctypes.c_uint32), ("f_iosize", ctypes.c_int32),
+ ("f_blocks", ctypes.c_uint64), ("f_bfree", ctypes.c_uint64),
+ ("f_bavail", ctypes.c_uint64), ("f_files", ctypes.c_uint64),
+ ("f_ffree", ctypes.c_uint64), ("f_fsid", ctypes.c_int32 * 2),
+ ("f_owner", ctypes.c_uint32), ("f_type", ctypes.c_uint32),
+ ("f_flags", ctypes.c_uint32), ("f_fssubtype", ctypes.c_uint32),
+ ("f_fstypename", ctypes.c_char * 16),
+ ("f_mntonname", ctypes.c_char * 1024),
+ ("f_mntfromname", ctypes.c_char * 1024),
+ ("f_reserved", ctypes.c_uint32 * 8),
+ ]
+ if (ctypes.sizeof(ctypes.c_void_p) != 8 or ctypes.sizeof(Statfs64) != 2168
+ or Statfs64.f_flags.offset != 64):
+ raise _ssh_config_error("ssh_config_mount_unverifiable")
+ api = ctypes.CDLL("/usr/lib/libSystem.B.dylib", use_errno=True)
+ api.fstatfs64.argtypes = [ctypes.c_int, ctypes.POINTER(Statfs64)]
+ api.fstatfs64.restype = ctypes.c_int
+ value = Statfs64()
+ if api.fstatfs64(fd, ctypes.byref(value)) != 0:
+ raise _ssh_config_error("ssh_config_mount_unverifiable")
+ return value.f_flags
+ except (OSError, AttributeError, TypeError, ValueError, OverflowError):
+ raise _ssh_config_error("ssh_config_mount_unverifiable") from None
+
+
+def _ssh_config_check_ownership(fd):
+ if sys.platform == "darwin" and _ssh_config_darwin_mount_flags(fd) & 0x00200000:
+ # MNT_IGNORE_OWNERSHIP can synthesize owner/mode metadata. Do not call
+ # such metadata an owner-only store, even if its ACL is empty.
+ raise _ssh_config_error("ssh_config_ownership_unenforced")
+
+
+def _ssh_config_check_fd(fd, *, private, directory=False):
+ _ssh_config_check_ownership(fd)
+ _ssh_config_check_acl(fd, private=private, directory=directory)
+
+
+def _ssh_config_darwin_acl_tags(fd):
+ """Read only Darwin's extended ACL, without paths or a child process.
+
+ Darwin returns 0 for an entry, -1/EINVAL at exhaustion (unlike Linux).
+ Validate the independent ACL handle first; release it on every path.
+ Constants and prototypes are from the macOS SDK sys/acl.h.
+ """
+ import ctypes
+ try:
+ api = ctypes.CDLL("/usr/lib/libSystem.B.dylib", use_errno=True)
+ api.acl_get_fd_np.argtypes = [ctypes.c_int, ctypes.c_int]
+ api.acl_get_fd_np.restype = ctypes.c_void_p
+ api.acl_valid.argtypes = [ctypes.c_void_p]
+ api.acl_valid.restype = ctypes.c_int
+ api.acl_get_entry.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.POINTER(ctypes.c_void_p)]
+ api.acl_get_entry.restype = ctypes.c_int
+ api.acl_get_tag_type.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_int)]
+ api.acl_get_tag_type.restype = ctypes.c_int
+ api.acl_free.argtypes = [ctypes.c_void_p]
+ api.acl_free.restype = ctypes.c_int
+ ctypes.set_errno(0)
+ acl = api.acl_get_fd_np(fd, 0x100) # ACL_TYPE_EXTENDED
+ if not acl:
+ # Darwin's fd getter returns NULL/ENOENT when FILESEC_ACL is absent
+ # (Libc posix1e/acl_file.c and gen/filesec.c), not an empty object.
+ if ctypes.get_errno() == errno.ENOENT:
+ return ()
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ acl_failed = False
+ try:
+ if api.acl_valid(acl) != 0:
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ tags = []
+ for index in range(129):
+ entry = ctypes.c_void_p()
+ ctypes.set_errno(0)
+ result = api.acl_get_entry(acl, 0 if index == 0 else -1, ctypes.byref(entry))
+ if result == -1 and ctypes.get_errno() == errno.EINVAL:
+ return tuple(tags)
+ if result != 0 or not entry.value or index == 128:
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ tag = ctypes.c_int()
+ if api.acl_get_tag_type(entry, ctypes.byref(tag)) != 0 or tag.value not in (1, 2):
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ tags.append(tag.value) # ACL_EXTENDED_ALLOW=1, DENY=2
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ except BaseException:
+ acl_failed = True
+ raise
+ finally:
+ def release_acl():
+ if api.acl_free(acl) != 0:
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ _ssh_config_cleanup(release_acl, preserve=acl_failed)
+ except (OSError, AttributeError, TypeError, ValueError, OverflowError):
+ raise _ssh_config_error("ssh_config_acl_unverifiable") from None
+
+
+def _ssh_config_check_acl(fd, *, private, directory=False):
+ """Conservative ACL policy, not an effective-permissions interpreter."""
+ if sys.platform == "darwin":
+ tags = _ssh_config_darwin_acl_tags(fd)
+ if any(tag not in (1, 2) for tag in tags):
+ raise _ssh_config_error("ssh_config_acl_unverifiable")
+ if (private and tags) or (not private and 1 in tags):
+ raise _ssh_config_error("ssh_config_unsafe_acl")
+ return
+ if sys.platform == "linux":
+ try:
+ for attribute in (("system.posix_acl_access", "system.posix_acl_default")
+ if directory else ("system.posix_acl_access",)):
+ try:
+ os.getxattr(fd, attribute)
+ except OSError as error:
+ if error.errno == errno.ENODATA:
+ continue
+ raise _ssh_config_error("ssh_config_acl_unverifiable") from None
+ raise _ssh_config_error("ssh_config_unsafe_acl")
+ except (AttributeError, TypeError, ValueError):
+ raise _ssh_config_error("ssh_config_acl_unverifiable") from None
+ return
+ raise _ssh_config_error("ssh_config_platform_unsupported")
+
+
+
+def _ssh_config_path(path):
+ if (not isinstance(path, str) or not path or len(path) > 4096
+ or not path.startswith("/") or path.startswith("//")
+ or ".." in path.split("/")
+ or any(ord(c) < 32 or 127 <= ord(c) <= 159
+ or 0xD800 <= ord(c) <= 0xDFFF for c in path)):
+ raise _ssh_config_error("ssh_config_invalid_path")
+ return path
+
+
+def _ssh_config_overflow_uid():
+ if sys.platform != "linux":
+ return None
+ try:
+ with open("/proc/sys/kernel/overflowuid", "rb") as stream:
+ raw = stream.read(32)
+ if not raw.strip().isdigit() or len(raw) == 32:
+ raise ValueError()
+ return int(raw)
+ except (OSError, ValueError):
+ raise _ssh_config_error("ssh_config_ownership_unverifiable") from None
+
+
+def _ssh_config_identity(info):
+ return (info.st_dev, info.st_ino, info.st_mode, info.st_uid,
+ info.st_gid, info.st_nlink, info.st_size,
+ info.st_mtime_ns, info.st_ctime_ns)
+
+
+def _ssh_config_metadata(info, overflow_uid, *, directory=False):
+ if overflow_uid is not None and info.st_uid == overflow_uid:
+ raise _ssh_config_error("ssh_config_ownership_unverifiable")
+ if directory:
+ if (not stat.S_ISDIR(info.st_mode) or info.st_uid not in (0, os.geteuid())
+ or (info.st_mode & 0o022
+ and not (info.st_uid == 0 and info.st_mode & stat.S_ISVTX))):
+ raise _ssh_config_error("ssh_config_unsafe_storage")
+ elif (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid()
+ or info.st_nlink != 1 or info.st_mode & 0o7022):
+ raise _ssh_config_error("ssh_config_unsafe_storage")
+
+
+def _ssh_config_snapshot(path):
+ _ssh_config_platform()
+ _ssh_config_path(path)
+ overflow_uid = _ssh_config_overflow_uid()
+ directory = fd = None
+ ancestors = []
+ failed = False
+ try:
+ flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK
+ directory = os.open("/", flags | os.O_DIRECTORY)
+ parts = path.split("/")[1:]
+ # Reject trailing slash / empty filename, but harmless internal repeated
+ # separators do not change the original path passed to OpenSSH.
+ if not parts[-1]:
+ raise _ssh_config_error("ssh_config_invalid_path")
+ for part in parts[:-1]:
+ info = os.fstat(directory)
+ _ssh_config_metadata(info, overflow_uid, directory=True)
+ _ssh_config_check_fd(directory, private=False, directory=True)
+ if not part:
+ continue
+ ancestors.append((info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid))
+ child = os.open(part, flags | os.O_DIRECTORY, dir_fd=directory)
+ previous, directory = directory, child
+ _ssh_config_cleanup(os.close, previous)
+ info = os.fstat(directory)
+ _ssh_config_metadata(info, overflow_uid, directory=True)
+ ancestors.append((info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid))
+ _ssh_config_check_fd(directory, private=False, directory=True)
+ fd = os.open(parts[-1], flags, dir_fd=directory)
+ before = os.fstat(fd)
+ _ssh_config_metadata(before, overflow_uid)
+ _ssh_config_check_fd(fd, private=False)
+ if before.st_size > SSH_CONFIG_MAX_BYTES:
+ raise _ssh_config_error("ssh_config_capacity")
+ raw = bytearray()
+ while len(raw) <= SSH_CONFIG_MAX_BYTES:
+ chunk = os.read(fd, min(8192, SSH_CONFIG_MAX_BYTES + 1 - len(raw)))
+ if not chunk:
+ break
+ raw.extend(chunk)
+ if len(raw) > SSH_CONFIG_MAX_BYTES:
+ raise _ssh_config_error("ssh_config_capacity")
+ after = os.fstat(fd)
+ _ssh_config_metadata(after, overflow_uid)
+ _ssh_config_check_fd(fd, private=False)
+ current = os.stat(parts[-1], dir_fd=directory, follow_symlinks=False)
+ if (_ssh_config_identity(before) != _ssh_config_identity(after)
+ or _ssh_config_identity(after) != _ssh_config_identity(current)):
+ raise _ssh_config_error("ssh_config_storage_changed")
+ return tuple(ancestors), _ssh_config_identity(after), hashlib.sha256(raw).digest()
+ except OSError:
+ failed = True
+ raise _ssh_config_error("ssh_config_unsafe_storage") from None
+ except BaseException:
+ failed = True
+ raise
+ finally:
+ def close_directory():
+ if directory is not None:
+ os.close(directory)
+ with _ssh_config_cleanup_scope(
+ _ssh_config_cleanup, close_directory, preserve=failed):
+ if fd is not None:
+ _ssh_config_cleanup(os.close, fd, preserve=failed)
+
+
+class _TrustedSshConfig:
+ # Source-streamed consumers may execute without a registered module entry;
+ # dataclass annotation resolution on supported older Python cannot do so.
+ __slots__ = ("_path", "_snapshot")
+
+ def __init__(self, path, snapshot):
+ object.__setattr__(self, "_path", path)
+ object.__setattr__(self, "_snapshot", snapshot)
+
+ @property
+ def path(self):
+ return self._path
+
+ def __setattr__(self, name, value):
+ raise AttributeError("trusted SSH selection is immutable")
+
+ def __delattr__(self, name):
+ raise AttributeError("trusted SSH selection is immutable")
+
+ def __repr__(self):
+ return ""
+
+ def revalidate(self):
+ if _ssh_config_snapshot(self.path) != self._snapshot:
+ raise _ssh_config_error("ssh_config_storage_changed")
+
+
+def select_trusted_ssh_config(path):
+ """Validate without executing config, resolving Includes or making copies.
+
+ Revalidation detects changes since selection. OpenSSH subsequently reopens
+ the original path: same-user races and executable config remain user trust.
+ """
+ return _TrustedSshConfig(path, _ssh_config_snapshot(path))
+
+
+def trusted_ssh_command(ssh_opts, host, *remote, prefix=(), identity_options=()):
+ selected = SSH_CONFIG_SELECTION.get()
+ safety = []
+ if selected is not None:
+ selected.revalidate()
+ safety = [
+ "-oStrictHostKeyChecking=yes", "-oUpdateHostKeys=no",
+ "-oForwardAgent=no", "-oPermitLocalCommand=no", "-oBatchMode=yes",
+ "-oControlMaster=no", "-oControlPath=none", "-oControlPersist=no",
+ "-F", selected.path,
+ ]
+ return ["ssh", *prefix, *identity_options, *safety, *ssh_opts, host, *remote]
+
+
+def redact_ssh_config_path(text):
+ """Mask the selected local path; not a general secret/content sanitizer."""
+ selected = SSH_CONFIG_SELECTION.get()
+ return text.replace(selected.path, "") if selected is not None else text
diff --git a/session_peer_core/ssh_identity.py b/session_peer_core/ssh_identity.py
index 32b15c2..22cb9fb 100644
--- a/session_peer_core/ssh_identity.py
+++ b/session_peer_core/ssh_identity.py
@@ -60,7 +60,7 @@ def ssh_identity_configuration(host: str, ssh_opts: list[str], *, handoff_budget
try:
completed = (ssh_handoff_configuration(host, ssh_opts, handoff_budget)
if handoff_budget is not None else subprocess.run(
- ["ssh", "-G", *ssh_opts, host], capture_output=True, timeout=DETECT_TIMEOUT))
+ trusted_ssh_command(ssh_opts, host, prefix=("-G",)), capture_output=True, timeout=DETECT_TIMEOUT))
output = completed.stdout
if isinstance(output, bytes):
if len(output) > 1024 * 1024:
diff --git a/session_peer_core/updates.py b/session_peer_core/updates.py
index e31756e..23c0d01 100644
--- a/session_peer_core/updates.py
+++ b/session_peer_core/updates.py
@@ -17,7 +17,7 @@ def remote_installed_version(host: str, ssh_opts: list[str],
)
try:
done = subprocess.run(
- ["ssh", *ssh_opts, host, probe],
+ trusted_ssh_command(ssh_opts, host, probe),
capture_output=True, timeout=30,
)
except FileNotFoundError as exc:
diff --git a/tests/core/test_handoff_watch.py b/tests/core/test_handoff_watch.py
index bfd2e4f..146d8c3 100644
--- a/tests/core/test_handoff_watch.py
+++ b/tests/core/test_handoff_watch.py
@@ -91,6 +91,45 @@ def snapshot(self):
return {path.name: (path.read_bytes(), path.stat().st_ino, path.stat().st_mtime_ns)
for path in self.ledger.root.iterdir() if path.is_file()}
+ def write_real_output_frames(self, stream):
+ # These positive controls test real descriptor delivery, not whether
+ # an OS scheduler always resumes within the watch's final 75ms. The
+ # command's original deadline and exhausted reserve are tested below
+ # with a deterministic clock, without changing the helper's clock.
+ writer = peer.HandoffWatchOutput(stream, peer.handoff_now() + 3)
+ try:
+ snapshot = {"operation": "watch", "event": "snapshot", "sequence": 1,
+ "historyPhase": self.record["phase"],
+ "handoff": peer.handoff_public(self.epoch, self.record, None)}
+ end = {"operation": "watch", "event": "end", "reason": "duration_elapsed",
+ "snapshotsEmitted": 1, "samples": 1}
+ self.assertIsNone(writer.write(peer.handoff_watch_frame(self.args, snapshot, ok=True)))
+ self.assertIsNone(writer.write(peer.handoff_watch_frame(self.args, end, ok=True), end=True))
+ finally:
+ closed = writer.close()
+ if writer.process.poll() is None:
+ # Cleanup does not turn a missed production deadline into a
+ # passing assertion; do not leave the test's child behind.
+ writer.process.kill()
+ writer.process.wait(timeout=1)
+ self.assertTrue(closed, "output helper was not reaped within its original budget")
+ self.assertIsNotNone(writer.process.poll(), "output helper remains running")
+
+ def read_real_output_frames(self, descriptor):
+ deadline = time.monotonic() + 3
+ raw = bytearray()
+ while raw.count(b"\n") < 2 and len(raw) <= 2 * peer.HANDOFF_WATCH_FRAME_BYTES:
+ remaining = deadline - time.monotonic()
+ self.assertGreater(remaining, 0, "incomplete output frames")
+ ready, _, _ = select.select([descriptor], [], [], remaining)
+ self.assertTrue(ready, "output frames were not readable")
+ chunk = os.read(descriptor, 8192)
+ self.assertTrue(chunk, "output ended before both frames")
+ raw.extend(chunk)
+ self.assertLessEqual(len(raw), 2 * peer.HANDOFF_WATCH_FRAME_BYTES)
+ self.assertTrue(raw.endswith(b"\n"), "truncated output frame")
+ return [json.loads(line) for line in raw.decode().splitlines()]
+
def test_ascii_parsers(self):
for value in ("1", "30", "60"):
self.assertEqual(peer.handoff_watch_timeout(value), int(value))
@@ -379,14 +418,12 @@ def test_tty_output_completes_without_changing_inherited_flags(self):
os.write(slave, b"x")
os.read(master, 1)
before = peer.fcntl.fcntl(slave, peer.fcntl.F_GETFL)
- self.args.watch_timeout = 1
with mock.patch.object(peer.fcntl, "fcntl", side_effect=AssertionError("inherited flag mutation")):
- code, _ = self.invoke(stream=stream, real_clock=True)
- self.assertEqual(code, 0, self.output_diagnostics)
- rows = [json.loads(line) for line in os.read(master, 8192).decode().splitlines()]
+ self.write_real_output_frames(stream)
+ rows = self.read_real_output_frames(master)
self.assertEqual([row["event"] for row in rows], ["snapshot", "end"])
+ self.assertTrue(all(row["ok"] for row in rows))
self.assertEqual(peer.fcntl.fcntl(slave, peer.fcntl.F_GETFL), before)
- self.assertTrue(all(process.poll() is not None for process in self.output_processes))
finally:
stream.close()
os.close(master)
@@ -618,13 +655,12 @@ def test_inherited_nonblocking_stdout_waits_for_drain_and_succeeds(self):
before = peer.fcntl.fcntl(write_fd, peer.fcntl.F_GETFL)
drainer = threading.Timer(0.1, lambda: os.read(read_fd, 1024 * 1024))
drainer.start()
- self.args.watch_timeout = 1
with mock.patch.object(peer.fcntl, "fcntl", side_effect=AssertionError("inherited flag mutation")):
- code, unused = self.invoke(stream=stream, real_clock=True)
+ self.write_real_output_frames(stream)
drainer.join()
- self.assertEqual(code, 0, self.output_diagnostics)
- rows = [json.loads(line) for line in os.read(read_fd, 8192).decode().splitlines()]
+ rows = self.read_real_output_frames(read_fd)
self.assertEqual([row["event"] for row in rows], ["snapshot", "end"])
+ self.assertTrue(all(row["ok"] for row in rows))
self.assertEqual(peer.fcntl.fcntl(write_fd, peer.fcntl.F_GETFL), before)
finally:
if drainer is not None:
@@ -677,6 +713,54 @@ def coalesced_sleep(seconds):
self.assertEqual([row["event"] for row in frames], ["snapshot", "end"])
self.assertLess(self.clock.now, 1 - peer.HANDOFF_WATCH_REAP_SECONDS)
+ def test_additive_scheduler_delay_exhausts_end_reserve_without_late_output(self):
+ self.args.watch_timeout = 1
+ before = self.snapshot()
+ # Cover both the final-write cutoff (before the total deadline) and
+ # the already-expired total budget; neither permits a late end frame.
+ for wake_at in (1 - peer.HANDOFF_WATCH_REAP_SECONDS + 0.001, 1.05):
+ with self.subTest(wake_at=wake_at):
+ self.clock.now = 0
+ def delayed_sleep(seconds):
+ self.clock.now = wake_at
+ with mock.patch.object(self.clock, "sleep", side_effect=delayed_sleep), \
+ mock.patch.object(peer.HandoffLedger, "record", wraps=self.ledger.record) as lookup:
+ code, frames = self.invoke()
+ lookup.assert_called_once()
+ self.assertEqual(code, 1)
+ self.assertEqual([row["event"] for row in frames], ["snapshot"])
+ self.assertEqual([(row["end"], row["result"]) for row in self.output_diagnostics],
+ [(False, None), (True, "output_unavailable")])
+ self.assertEqual(self.output_diagnostics[-1]["result"], "output_unavailable")
+ self.assertTrue(self.output_diagnostics[-1]["end"])
+ self.assertLess(self.output_diagnostics[-1]["remainingMs"],
+ peer.HANDOFF_WATCH_REAP_SECONDS * 1000)
+ self.assertEqual(self.clock.now, wake_at, "watch reset or extended its original budget")
+ self.assertTrue(before == self.snapshot(), "watch mutated history after output expiry")
+
+ def test_command_and_real_writer_complete_at_controlled_sample_limit(self):
+ read_fd, write_fd = os.pipe()
+ stream = os.fdopen(write_fd, "w")
+ self.args.watch_timeout = 10
+ before = self.snapshot()
+ try:
+ # Only pacing is synthetic. The command, clock, output helper,
+ # deadline, frame ACKs and cleanup remain real. The unchanged
+ # 61-sample bound ends the loop away from the last 75ms window.
+ with mock.patch.object(peer, "handoff_watch_pause", return_value=None):
+ code, _ = self.invoke(stream=stream, real_clock=True)
+ self.assertEqual(code, 0, self.output_diagnostics)
+ rows = self.read_real_output_frames(read_fd)
+ self.assertEqual([row["event"] for row in rows], ["snapshot", "end"])
+ self.assertTrue(all(row["ok"] for row in rows))
+ self.assertEqual(rows[-1]["samples"], 61)
+ self.assertEqual(rows[-1]["reason"], "duration_elapsed")
+ self.assertTrue(all(process.poll() is not None for process in self.output_processes))
+ self.assertTrue(before == self.snapshot(), "command output mutated history")
+ finally:
+ stream.close()
+ os.close(read_fd)
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/core/test_ssh_config_cli.py b/tests/core/test_ssh_config_cli.py
new file mode 100644
index 0000000..8cbc193
--- /dev/null
+++ b/tests/core/test_ssh_config_cli.py
@@ -0,0 +1,317 @@
+"""Trusted config selection is invocation-local and reaches every SSH spawn.
+
+All config selection and process boundaries are mocked: these tests never load
+user SSH configuration, execute hooks, connect, or submit a native message.
+"""
+import contextlib
+from dataclasses import dataclass
+import io
+import json
+import subprocess
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+import session_peer as peer
+
+
+CONFIG = '/private/fixture/secret-ssh-config'
+FLAGS = ['--ssh-config', CONFIG, '--trust-ssh-config']
+SAFETY = ['-oStrictHostKeyChecking=yes', '-oUpdateHostKeys=no',
+ '-oForwardAgent=no', '-oPermitLocalCommand=no', '-oBatchMode=yes',
+ '-oControlMaster=no', '-oControlPath=none', '-oControlPersist=no']
+
+
+@dataclass(frozen=True)
+class SelectedConfig:
+ path: str
+ revalidate: object
+
+
+class UnsupportedTrustedConfigCli(unittest.TestCase):
+ def test_windows_refuses_before_selection(self):
+ output, error = io.StringIO(), io.StringIO()
+ with mock.patch.object(peer, 'IS_WINDOWS', True), \
+ mock.patch.object(peer, 'select_trusted_ssh_config') as selector, \
+ mock.patch.object(peer.subprocess, 'run') as run, \
+ contextlib.redirect_stdout(output), contextlib.redirect_stderr(error):
+ code = peer.main(['list', '--host', 'fixture', *FLAGS, '--json'])
+ self.assertEqual(code, 1)
+ self.assertEqual(json.loads(output.getvalue())['reason'], 'ssh_config_platform_unsupported')
+ selector.assert_not_called()
+ run.assert_not_called()
+
+
+@unittest.skipIf(peer.IS_WINDOWS, 'selected-config and handoff/identity execution are POSIX-only')
+class TrustedConfigCli(unittest.TestCase):
+ def setUp(self):
+ self.config = SelectedConfig(CONFIG, mock.Mock())
+ self.selector = mock.patch.object(peer, 'select_trusted_ssh_config', return_value=self.config).start()
+ self.addCleanup(mock.patch.stopall)
+ mock.patch.object(peer, 'tailscale_status', return_value={}).start()
+ mock.patch.object(peer, 'resolve_ssh_destination', side_effect=lambda host, status: host).start()
+
+ def invoke(self, *argv, standalone=False):
+ output, error = io.StringIO(), io.StringIO()
+ with contextlib.redirect_stdout(output), contextlib.redirect_stderr(error), \
+ mock.patch.object(peer.sys, 'argv', ['session-peer', *argv]):
+ try:
+ code = peer.main(None if standalone else list(argv))
+ except SystemExit as exc:
+ code = exc.code
+ return code, output.getvalue(), error.getvalue()
+
+ @contextlib.contextmanager
+ def selected(self):
+ token = peer.SSH_CONFIG_SELECTION.set(self.config)
+ try:
+ yield
+ finally:
+ peer.SSH_CONFIG_SELECTION.reset(token)
+
+ def assert_selected(self, argv):
+ index = argv.index('-F')
+ self.assertEqual(argv[index + 1], CONFIG)
+ self.assertEqual(argv[index - len(SAFETY):index], SAFETY)
+ self.assertEqual(argv.count('-F'), 1)
+
+ def test_all_host_commands_select_once_and_restore_context(self):
+ for command in ('list', 'send', 'doctor', 'update'):
+ seen = []
+ def dispatch(args):
+ seen.append(peer.SSH_CONFIG_SELECTION.get())
+ return 0
+ extra = ['--to', 'worker', '-m', 'hello'] if command == 'send' else []
+ outer = object()
+ token = peer.SSH_CONFIG_SELECTION.set(outer)
+ try:
+ with self.subTest(command=command), mock.patch.object(peer, 'cmd_' + command, side_effect=dispatch):
+ self.selector.reset_mock()
+ code, _, _ = self.invoke(command, '--host', 'fixture', *FLAGS, *extra)
+ self.assertEqual(code, 0)
+ self.assertEqual(seen, [self.config])
+ self.selector.assert_called_once_with(CONFIG)
+ self.assertIs(peer.SSH_CONFIG_SELECTION.get(), outer)
+ finally:
+ peer.SSH_CONFIG_SELECTION.reset(token)
+
+ def test_opt_out_clears_inherited_context_and_restores_after_error(self):
+ with self.selected(), mock.patch.object(peer, 'cmd_list') as dispatch:
+ def fail(args):
+ self.assertIsNone(peer.SSH_CONFIG_SELECTION.get())
+ raise peer.CcPeerError('fixture refusal')
+ dispatch.side_effect = fail
+ code, _, _ = self.invoke('list', '--host', 'fixture', '--json')
+ self.assertEqual(code, 1)
+ self.assertIs(peer.SSH_CONFIG_SELECTION.get(), self.config)
+ self.selector.assert_not_called()
+ self.config.revalidate.assert_not_called()
+
+ def test_invalid_combinations_fail_before_io_advisories_or_stdin(self):
+ cases = [
+ ['list', '--host', 'fixture', '--ssh-config', CONFIG],
+ ['list', '--host', 'fixture', '--trust-ssh-config'],
+ ['list', '--host', 'fixture', *FLAGS, '--ssh-config', CONFIG],
+ ['list', '--host', 'fixture', *FLAGS, '--trust-ssh-config'],
+ ['list', *FLAGS],
+ ['setup', *FLAGS],
+ ['send', '--host', 'fixture', '--device', 'a' * 64, '--to', 'worker', *FLAGS],
+ ['send', '--host', 'fixture', '--to', 'session-peer://v1/reply/fixture', *FLAGS],
+ ]
+ for argv in cases:
+ with self.subTest(argv=argv), mock.patch.object(peer.subprocess, 'run') as run, \
+ mock.patch.object(peer.sys, 'stdin') as stdin, \
+ mock.patch.object(peer, 'prepare_client_update') as client, \
+ mock.patch.object(peer, 'prepare_skill_updates') as skills:
+ self.selector.reset_mock()
+ code, _, _ = self.invoke(*argv, standalone=True)
+ self.assertNotEqual(code, 0)
+ self.selector.assert_not_called()
+ run.assert_not_called()
+ stdin.read.assert_not_called()
+ client.assert_not_called()
+ skills.assert_not_called()
+
+ def test_opt_out_helper_preserves_exact_argv_without_config_io(self):
+ token = peer.SSH_CONFIG_SELECTION.set(None)
+ try:
+ self.assertEqual(peer.trusted_ssh_command(['-p', '2222'], 'fixture', 'remote',
+ prefix=('-G',), identity_options=('-oKnownHostsCommand=helper',)),
+ ['ssh', '-G', '-oKnownHostsCommand=helper', '-p', '2222', 'fixture', 'remote'])
+ finally:
+ peer.SSH_CONFIG_SELECTION.reset(token)
+ self.selector.assert_not_called()
+ self.config.revalidate.assert_not_called()
+
+ def test_identity_pinning_and_safety_precede_selected_config(self):
+ identity = ['-oKnownHostsCommand=pin-helper', '-oStrictHostKeyChecking=yes']
+ with self.selected():
+ result = peer.trusted_ssh_command(['-p', '2222'], 'fixture', 'remote',
+ prefix=('-G',), identity_options=identity)
+ self.assertEqual(result, ['ssh', '-G', *identity, *SAFETY, '-F', CONFIG,
+ '-p', '2222', 'fixture', 'remote'])
+ self.config.revalidate.assert_called_once_with()
+
+ def test_raw_config_flags_remain_rejected_even_when_trusted(self):
+ with self.selected():
+ for opts in (['-F', CONFIG], ['-F' + CONFIG], ['-oProxyCommand=arbitrary']):
+ with self.subTest(opts=opts), self.assertRaises(peer.CcPeerError):
+ peer.check_ssh_options(opts)
+
+ def test_metadata_identity_and_version_spawns_use_selected_config(self):
+ cases = [
+ (lambda: peer.ssh_user_metadata('fixture', []), 'user deploy\n'),
+ (lambda: peer.ssh_identity_configuration('fixture', []), 'hostname fixture\nport 22\n'),
+ (lambda: peer.remote_installed_version('deploy@fixture', [], {}), 'session-peer 1.2.1\n'),
+ ]
+ for call, stdout in cases:
+ with self.subTest(stdout=stdout), self.selected(), \
+ mock.patch.object(peer.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, stdout, '')) as run:
+ self.config.revalidate.reset_mock()
+ call()
+ run.assert_called_once()
+ self.assert_selected(run.call_args.args[0])
+ self.config.revalidate.assert_called_once_with()
+
+ def test_metadata_then_effect_revalidates_and_refuses_changed_config(self):
+ self.config.revalidate.side_effect = [None, peer.CcPeerError('Trusted SSH config changed')]
+ with self.selected(), mock.patch.object(peer.Path, 'read_text', return_value='source'), \
+ mock.patch.object(peer.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, 'user deploy\n', '')) as run, \
+ self.assertRaises(peer.CcPeerError):
+ peer.run_remote('fixture', ['send', '--to', 'worker', '--b64', 'eA=='], [])
+ run.assert_called_once()
+ self.assertIn('-G', run.call_args.args[0])
+ self.assertEqual(self.config.revalidate.call_count, 2)
+
+ def test_bounded_metadata_spawn_uses_same_selection(self):
+ completed = SimpleNamespace(returncode=0, stdout=b'user deploy\n', reason=None,
+ interrupted=False, stdout_overflow=False,
+ stderr_overflow=False, cleanup_failed=False)
+ budget = (peer.handoff_now() + 3, peer.handoff_now() + 8)
+ with self.selected(), mock.patch.object(peer, 'handoff_stream_child', return_value=completed) as child:
+ peer.ssh_handoff_configuration('fixture', [], budget)
+ child.assert_called_once()
+ self.assert_selected(child.call_args.args[0])
+ self.assertIn('-G', child.call_args.args[0])
+ self.config.revalidate.assert_called_once_with()
+
+ def test_reverse_probe_does_not_inherit_forward_config(self):
+ with self.selected(), mock.patch.object(peer, 'is_self_ssh_destination', return_value=False), \
+ mock.patch.object(peer.shutil, 'which', return_value='/fixture/ssh'), \
+ mock.patch.object(peer.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, 'user deploy\n', '')) as run:
+ result = peer.probe_return_route('return-fixture')
+ self.assertIs(peer.SSH_CONFIG_SELECTION.get(), self.config)
+ self.assertEqual(result['status'], 'verified')
+ self.assertEqual(run.call_count, 2)
+ for call in run.call_args_list:
+ self.assertNotIn('-F', call.args[0])
+ self.assertNotIn(CONFIG, call.args[0])
+ self.config.revalidate.assert_not_called()
+
+ def test_bounded_effect_spawn_uses_same_selection_and_identity_options(self):
+ payload = {'schemaVersion': 1, 'command': 'send', 'ok': True, 'submitted': True}
+ completed = SimpleNamespace(returncode=0, stdout=json.dumps(payload).encode(),
+ reason=None, interrupted=False, stdout_overflow=False,
+ stderr_overflow=False, cleanup_failed=False, spawned=True)
+ budget = (peer.handoff_now() + 3, peer.handoff_now() + 8)
+ with self.selected(), mock.patch.object(peer, 'handoff_stream_child', return_value=completed) as child, \
+ mock.patch.object(peer, 'ssh_handoff_request'), \
+ mock.patch.object(peer, 'handoff_validate_remote_argv'), \
+ mock.patch.object(peer, 'handoff_validate_remote_response', side_effect=lambda result, context: result):
+ result = peer._run_remote_handoff_dispatch('deploy@fixture', ['send'], [],
+ ['-oKnownHostsCommand=pin-helper'], budget, {'phase': 'effect'})
+ self.assertTrue(result['submitted'])
+ child.assert_called_once()
+ command = child.call_args.args[0]
+ self.assert_selected(command)
+ self.assertLess(command.index('-oKnownHostsCommand=pin-helper'), command.index('-F'))
+ self.assertEqual(child.call_args.args[2:4], budget)
+
+ def test_identity_pin_reaches_actual_dispatch_before_trusted_config(self):
+ fingerprint = 'SHA256:' + 'A' * 43
+ payload = {'schemaVersion': 1, 'command': 'send', 'ok': True, 'submitted': True}
+ outcomes = [subprocess.CompletedProcess([], 0, b'hostname fixture\nport 22\n', b''),
+ subprocess.CompletedProcess([], 0, json.dumps(payload).encode(), b'')]
+ proof = {'decision': 'observed', 'fingerprint': fingerprint, 'algorithm': 'ssh-ed25519'}
+ with self.selected(), mock.patch.object(peer.subprocess, 'run', side_effect=outcomes) as run, \
+ mock.patch.object(peer.tempfile, 'TemporaryDirectory') as temporary, \
+ mock.patch.object(peer.Path, 'chmod'), mock.patch.object(peer.os, 'open', return_value=123), \
+ mock.patch.object(peer.os, 'fdopen'), mock.patch.object(peer.Path, 'read_text', return_value='source'), \
+ mock.patch.object(peer, 'ssh_key_receipt', return_value=proof):
+ temporary.return_value.__enter__.return_value = '/fixture/identity'
+ result = peer.run_remote_with_identity('deploy@fixture', ['send'], [], expected=fingerprint)
+ self.assertTrue(result['submitted'])
+ self.assertEqual(result['sshIdentity']['status'], 'verified')
+ self.assertEqual(run.call_count, 2)
+ for call in run.call_args_list:
+ self.assert_selected(call.args[0])
+ effect = run.call_args.args[0]
+ for option in ('-oStrictHostKeyChecking=yes', '-oUserKnownHostsFile=none',
+ '-oGlobalKnownHostsFile=none', '-oHostKeyAlias=session-peer-pinned'):
+ self.assertLess(effect.index(option), effect.index('-F'))
+ self.assertTrue(any(option.startswith('-oKnownHostsCommand=') for option in effect[:effect.index('-F')]))
+
+ def test_complete_send_evidence_survives_transport_failure(self):
+ payload = {'schemaVersion': 1, 'command': 'send', 'ok': False, 'submitted': True,
+ 'status': 'queued', 'queueId': 'fixture-queue', 'consumptionConfirmed': False}
+ with self.selected(), mock.patch.object(peer.Path, 'read_text', return_value='source'), \
+ mock.patch.object(peer.subprocess, 'run', return_value=subprocess.CompletedProcess(
+ [], 255, json.dumps(payload), 'Permission denied')) as run:
+ result = peer.run_remote('deploy@fixture', ['send'], [])
+ self.assertEqual(result, {**payload, 'sshUser': 'deploy', 'sshUserSource': 'explicit'})
+ self.assert_selected(run.call_args.args[0])
+
+ def test_selected_path_is_redacted_from_transport_diagnostics(self):
+ calls = [lambda: peer.run_remote('deploy@fixture', ['list'], []),
+ lambda: peer.remote_installed_version('deploy@fixture', [], {}),
+ lambda: peer.push_to_remote('deploy@fixture', [], {})]
+ for call in calls:
+ with self.subTest(call=call), self.selected(), \
+ mock.patch.object(peer.Path, 'read_text', return_value='source'), \
+ mock.patch.object(peer.Path, 'read_bytes', return_value=b'source'), \
+ mock.patch.object(peer.subprocess, 'run', return_value=subprocess.CompletedProcess(
+ [], 255, b'', ('cannot open ' + CONFIG).encode())), \
+ self.assertRaises(peer.CcPeerError) as caught:
+ call()
+ self.assertNotIn(CONFIG, str(caught.exception))
+ self.assertNotIn(CONFIG, json.dumps(caught.exception.details))
+
+ def test_cli_config_is_not_forwarded_to_remote_arguments(self):
+ with mock.patch.object(peer, 'run_remote', return_value={'ok': True, 'target': {'name': 'worker'}}) as remote:
+ code, _, _ = self.invoke('send', '--host', 'deploy@fixture', '--to', 'worker',
+ '-m', 'hello', '--no-from', '--no-reply-to', '--json', *FLAGS)
+ self.assertEqual(code, 0)
+ args = remote.call_args.args[1]
+ for value in (CONFIG, '--ssh-config', '--trust-ssh-config'):
+ self.assertNotIn(value, args)
+
+ def test_optional_list_probe_change_preserves_successful_discovery(self):
+ listing = {'ok': True, 'sessions': [{'name': 'worker', 'agent': 'claude',
+ 'pid': 7, 'alive': True, 'reachable': True}]}
+ self.config.revalidate.side_effect = peer.CcPeerError('Trusted SSH config changed')
+ with mock.patch.object(peer.SshTransport, 'execute', return_value=listing), \
+ mock.patch.object(peer.subprocess, 'run') as run:
+ code, output, _ = self.invoke('list', '--host', 'deploy@fixture', '--json', *FLAGS)
+ self.assertEqual(code, 0)
+ result = json.loads(output)
+ self.assertTrue(result['ok'])
+ self.assertEqual(result['sessions'], listing['sessions'])
+ self.assertEqual(result['remoteVersionProbe']['status'], 'unknown')
+ run.assert_not_called()
+
+ def test_update_fresh_verification_change_preserves_commit_evidence(self):
+ self.config.revalidate.side_effect = [None, None, peer.CcPeerError('Trusted SSH config changed')]
+ outcomes = [subprocess.CompletedProcess([], 0, 'session-peer 0.1.0\n', ''),
+ subprocess.CompletedProcess([], 0, 'session-peer ' + peer.__version__ + '\n', '')]
+ with mock.patch.object(peer.Path, 'read_bytes', return_value=b'fixture source'), \
+ mock.patch.object(peer.subprocess, 'run', side_effect=outcomes) as run:
+ code, output, _ = self.invoke('update', '--host', 'deploy@fixture', '--json', *FLAGS)
+ self.assertEqual(code, 1)
+ result = json.loads(output)
+ self.assertTrue(result['committed'])
+ self.assertEqual(result['commitStatus'], 'committed')
+ self.assertEqual(result['verificationStatus'], 'unknown')
+ self.assertFalse(result['retryAllowed'])
+ self.assertEqual(run.call_count, 2)
+ for call in run.call_args_list:
+ self.assert_selected(call.args[0])
diff --git a/tests/core/test_ssh_config_storage.py b/tests/core/test_ssh_config_storage.py
new file mode 100644
index 0000000..ba40224
--- /dev/null
+++ b/tests/core/test_ssh_config_storage.py
@@ -0,0 +1,307 @@
+"""Offline trusted-config storage tests; external-volume fixtures are synthetic.
+
+The fixture bypasses ownership-enforcement mount detection. A test-only
+SESSION_PEER_TEST_CONFIG_UNSAFE_ANCESTOR can explicitly select one ancestor
+whose group-write bit is normalized. Other native ACL, inode, owner and mode
+checks still execute. This does not prove safety of that mount; a separate
+native descriptor test requires ignore-ownership mounts be refused.
+"""
+
+import errno
+import os
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+import session_peer as peer
+
+
+@unittest.skipUnless(sys.platform == "linux" and shutil.which("ssh"), "native Linux OpenSSH config-only check")
+class NativeTrustedSshConfig(unittest.TestCase):
+ def test_native_config_parse_uses_selected_file_and_forced_host_key_policy(self):
+ # No connection, Match exec, credentials, or default user config. This
+ # native acceptance check uses real ownership/ACL checks, without mocks.
+ with tempfile.TemporaryDirectory(prefix="codex-ssh-config-") as temp:
+ path = Path(temp).resolve() / "selected config"
+ path.write_text("Host fixture-host\n HostName fixture.invalid\n User fixture-user\n"
+ " StrictHostKeyChecking no\n ControlMaster auto\n"
+ " IdentityFile none\n IdentityAgent none\n", encoding="utf-8")
+ path.chmod(0o600)
+ selected = peer.select_trusted_ssh_config(str(path))
+ token = peer.SSH_CONFIG_SELECTION.set(selected)
+ try:
+ done = subprocess.run(peer.trusted_ssh_command([], "fixture-host", prefix=("-G",)),
+ capture_output=True, text=True, timeout=5)
+ finally:
+ peer.SSH_CONFIG_SELECTION.reset(token)
+ self.assertEqual(done.returncode, 0, "native config-only parse failed")
+ values = dict(line.split(" ", 1) for line in done.stdout.splitlines() if " " in line)
+ self.assertEqual(values["user"], "fixture-user")
+ self.assertEqual(values["hostname"], "fixture.invalid")
+ self.assertEqual(values["stricthostkeychecking"], "true")
+ self.assertEqual(values["controlmaster"], "false")
+
+
+@unittest.skipUnless(os.name == "posix", "POSIX storage contract")
+class TrustedSshConfigStorage(unittest.TestCase):
+ def setUp(self):
+ self.temp = tempfile.TemporaryDirectory(prefix="ssh-config-storage-")
+ self.addCleanup(self.temp.cleanup)
+ self.root = Path(self.temp.name).resolve()
+ self.path = self.root / "ssh.conf"
+ self.path.write_bytes(b"Host example\n HostName example.invalid\n")
+ self.path.chmod(0o600)
+ self.mount = mock.patch.object(peer, "_ssh_config_check_ownership")
+ self.mount.start()
+ self.addCleanup(self.mount.stop)
+ fixture_ancestor = os.environ.get("SESSION_PEER_TEST_CONFIG_UNSAFE_ANCESTOR")
+ if fixture_ancestor:
+ volume = Path(fixture_ancestor).resolve()
+ self.assertIn(volume, self.root.parents, "synthetic ancestor is outside the test path")
+ volume_info = volume.stat()
+ volume_id = (volume_info.st_dev, volume_info.st_ino)
+ real_metadata = peer._ssh_config_metadata
+ def synthetic_volume_metadata(info, overflow_uid, *, directory=False):
+ if directory and (getattr(info, "st_dev", None),
+ getattr(info, "st_ino", None)) == volume_id:
+ info = SimpleNamespace(st_uid=info.st_uid,
+ st_mode=info.st_mode & ~0o020)
+ return real_metadata(info, overflow_uid, directory=directory)
+ patch = mock.patch.object(peer, "_ssh_config_metadata",
+ side_effect=synthetic_volume_metadata)
+ patch.start()
+ self.addCleanup(patch.stop)
+
+ def select(self):
+ return peer.select_trusted_ssh_config(str(self.path))
+
+ def assertRefused(self, action, reason=None):
+ with self.assertRaises(peer.CcPeerError) as caught:
+ action()
+ self.assertNotIn(str(self.path), str(caught.exception))
+ if reason:
+ self.assertIn(reason, str(caught.exception))
+ return caught.exception
+
+ def test_accepts_readable_file_and_immutable_selection(self):
+ self.path.chmod(0o644)
+ selected = self.select()
+ self.assertEqual(selected.path, str(self.path))
+ selected.revalidate()
+ self.assertNotIn(str(self.path), repr(selected))
+ with self.assertRaises(AttributeError):
+ selected.path = "/other"
+
+ def test_invalid_paths(self):
+ for path in ("relative", "/", "//tmp/config", "/a/../config",
+ "/a/\nconfig", "/a/\x7fconfig", "/a/\x85config", "/a/\ud800"):
+ with self.subTest(path=repr(path)):
+ self.assertRefused(lambda: peer.select_trusted_ssh_config(path))
+
+ def test_modes_and_hardlinks(self):
+ for mode in (0o620, 0o602, 0o4600, 0o2600, 0o1600):
+ self.path.chmod(mode)
+ self.assertRefused(self.select)
+ self.path.chmod(0o600)
+ os.link(self.path, self.root / "link")
+ self.assertRefused(self.select)
+
+ def test_symlink_file_and_component(self):
+ target = self.root / "target"
+ self.path.rename(target)
+ self.path.symlink_to(target)
+ self.assertRefused(self.select)
+ alias = self.root / "alias"
+ alias.symlink_to(self.root, target_is_directory=True)
+ self.assertRefused(lambda: peer.select_trusted_ssh_config(str(alias / "target")))
+
+ def test_fifo_and_directory_refused_without_blocking(self):
+ self.path.unlink()
+ os.mkfifo(self.path, 0o600)
+ self.assertRefused(self.select)
+ self.path.unlink()
+ self.path.mkdir()
+ self.assertRefused(self.select)
+
+ def test_size_boundary(self):
+ self.path.write_bytes(b"#" * 65536)
+ self.select()
+ self.path.write_bytes(b"#" * 65537)
+ self.assertRefused(self.select, "ssh_config_capacity")
+
+ def test_revalidation_content_replacement_and_mode(self):
+ for mutation in ("content", "replace", "mode"):
+ with self.subTest(mutation=mutation):
+ self.path.chmod(0o600)
+ self.path.write_bytes(b"Host alpha\n")
+ selected = self.select()
+ if mutation == "content":
+ self.path.write_bytes(b"Host bravo\n")
+ elif mutation == "replace":
+ replacement = self.root / "replacement"
+ replacement.write_bytes(self.path.read_bytes())
+ replacement.chmod(0o600)
+ replacement.replace(self.path)
+ else:
+ self.path.chmod(0o640)
+ self.assertRefused(selected.revalidate, "ssh_config_storage_changed")
+
+ def test_unsafe_parent(self):
+ self.root.chmod(0o777)
+ self.addCleanup(self.root.chmod, 0o700)
+ self.assertRefused(self.select)
+
+ def test_owner_and_overflow_uid_metadata(self):
+ real = self.path.stat()
+ fields = {name: getattr(real, name) for name in
+ ("st_mode", "st_uid", "st_nlink")}
+ fields["st_uid"] = os.geteuid() + 1
+ self.assertRefused(lambda: peer._ssh_config_metadata(
+ SimpleNamespace(**fields), None))
+ fields["st_uid"] = os.geteuid()
+ self.assertRefused(lambda: peer._ssh_config_metadata(
+ SimpleNamespace(**fields), os.geteuid()), "ssh_config_ownership_unverifiable")
+
+ def test_root_sticky_parent_and_unknown_owner(self):
+ safe = SimpleNamespace(st_uid=0, st_mode=stat.S_IFDIR | 0o1777)
+ peer._ssh_config_metadata(safe, 65534, directory=True)
+ for uid, mode in ((os.geteuid() + 1, 0o700), (0, 0o777), (65534, 0o700)):
+ with self.subTest(uid=uid, mode=mode):
+ self.assertRefused(lambda: peer._ssh_config_metadata(
+ SimpleNamespace(st_uid=uid, st_mode=stat.S_IFDIR | mode),
+ 65534, directory=True))
+
+ def test_linux_overflow_ownership_proof_unavailable(self):
+ with mock.patch.object(peer.sys, "platform", "linux"):
+ with mock.patch("builtins.open", mock.mock_open(read_data=b"65534\n")):
+ self.assertEqual(peer._ssh_config_overflow_uid(), 65534)
+ with mock.patch("builtins.open", side_effect=OSError("private path")):
+ self.assertRefused(peer._ssh_config_overflow_uid,
+ "ssh_config_ownership_unverifiable")
+
+ def test_hash_change_detected_even_when_identity_matches(self):
+ with mock.patch.object(peer, "_ssh_config_identity", return_value=("synthetic",)):
+ selected = self.select()
+ self.path.write_bytes(b"different content")
+ self.assertRefused(selected.revalidate, "ssh_config_storage_changed")
+
+ def test_changed_during_read(self):
+ real_read = os.read
+ def mutate(fd, size):
+ result = real_read(fd, size)
+ if result:
+ self.path.write_bytes(b"changed")
+ return result
+ with mock.patch.object(peer.os, "read", side_effect=mutate):
+ self.assertRefused(self.select, "ssh_config_storage_changed")
+
+ def test_acl_policy_mocks(self):
+ with mock.patch.object(peer.sys, "platform", "darwin"):
+ for tags in ((1,), (1, 2)):
+ with mock.patch.object(peer, "_ssh_config_darwin_acl_tags", return_value=tags):
+ self.assertRefused(lambda: peer._ssh_config_check_acl(42, private=False),
+ "ssh_config_unsafe_acl")
+ with mock.patch.object(peer, "_ssh_config_darwin_acl_tags", return_value=(2,)):
+ peer._ssh_config_check_acl(42, private=False)
+ with mock.patch.object(peer.sys, "platform", "linux"):
+ with mock.patch.object(peer.os, "getxattr", create=True, return_value=b"acl"):
+ self.assertRefused(lambda: peer._ssh_config_check_acl(42, private=False))
+ with mock.patch.object(peer.os, "getxattr", create=True,
+ side_effect=OSError(errno.ENOTSUP, "not supported")):
+ self.assertRefused(lambda: peer._ssh_config_check_acl(42, private=False),
+ "ssh_config_acl_unverifiable")
+
+ def test_mount_ignores_ownership_refused(self):
+ self.mount.stop()
+ with mock.patch.object(peer.sys, "platform", "darwin"):
+ with mock.patch.object(peer, "_ssh_config_darwin_mount_flags", return_value=0x00200000):
+ self.assertRefused(lambda: peer._ssh_config_check_ownership(42),
+ "ssh_config_ownership_unenforced")
+
+ @unittest.skipUnless(peer.sys.platform == "darwin", "Darwin native mount API")
+ def test_native_mount_policy(self):
+ self.mount.stop()
+ fd = os.open(self.path, os.O_RDONLY)
+ try:
+ flags = peer._ssh_config_darwin_mount_flags(fd)
+ if flags & 0x00200000:
+ self.assertRefused(lambda: peer._ssh_config_check_ownership(fd),
+ "ssh_config_ownership_unenforced")
+ else:
+ peer._ssh_config_check_ownership(fd)
+ finally:
+ os.close(fd)
+
+ def test_primary_failure_survives_close_failure(self):
+ with mock.patch.object(peer.os, "close", side_effect=OSError("private diagnostic")):
+ with self.assertRaisesRegex(peer.CcPeerError, "original"):
+ with peer._ssh_config_cleanup_scope(peer.os.close, 42):
+ raise peer._ssh_config_error("original")
+
+ def test_file_close_failure_still_closes_directory(self):
+ real_open, real_close = os.open, os.close
+ active = set()
+ failed_file = [False]
+ parent_closed = [False]
+ parent_inode = self.root.stat().st_ino
+
+ def opened(*args, **kwargs):
+ fd = real_open(*args, **kwargs)
+ active.add(fd)
+ return fd
+
+ def closed(fd):
+ info = os.fstat(fd)
+ real_close(fd)
+ active.discard(fd)
+ if stat.S_ISREG(info.st_mode):
+ failed_file[0] = True
+ raise OSError("private file-close diagnostic")
+ if failed_file[0] and info.st_ino == parent_inode:
+ parent_closed[0] = True
+
+ try:
+ with mock.patch.object(peer.os, "open", side_effect=opened), \
+ mock.patch.object(peer.os, "close", side_effect=closed):
+ error = self.assertRefused(self.select, "ssh_config_cleanup_failed")
+ self.assertNotIn("private file-close diagnostic", str(error))
+ self.assertTrue(parent_closed[0], "directory was not closed after file cleanup failed")
+ self.assertFalse(active, "validation left owned descriptors open")
+ finally:
+ # Also clean the original buggy implementation during regression
+ # reproduction, without weakening the leak assertion above.
+ for fd in active:
+ real_close(fd)
+
+
+class TrustedSshConfigCommand(unittest.TestCase):
+ def test_opt_out_exact_argv_no_io(self):
+ token = peer.SSH_CONFIG_SELECTION.set(None)
+ try:
+ with mock.patch.object(peer.os, "open", side_effect=AssertionError("file I/O")):
+ self.assertEqual(peer.trusted_ssh_command(
+ ["-p", "22"], "host", "command", prefix=("-T",), identity_options=("-v",)),
+ ["ssh", "-T", "-v", "-p", "22", "host", "command"])
+ finally:
+ peer.SSH_CONFIG_SELECTION.reset(token)
+ def test_selected_safety_order_revalidation_and_diagnostic(self):
+ selected = mock.Mock(path="/private/config")
+ token = peer.SSH_CONFIG_SELECTION.set(selected)
+ try:
+ argv = peer.trusted_ssh_command(["-oBatchMode=no"], "host", "cmd", prefix=("-G",))
+ selected.revalidate.assert_called_once_with()
+ self.assertEqual(argv[:3], ["ssh", "-G", "-oStrictHostKeyChecking=yes"])
+ self.assertLess(argv.index("-oBatchMode=yes"), argv.index("-oBatchMode=no"))
+ self.assertEqual(argv[-5:], ["-F", "/private/config", "-oBatchMode=no", "host", "cmd"])
+ self.assertEqual(peer.redact_ssh_config_path("bad /private/config"), "bad ")
+ selected.revalidate.side_effect = peer._ssh_config_error("changed")
+ with self.assertRaises(peer.CcPeerError):
+ peer.trusted_ssh_command([], "host")
+ finally:
+ peer.SSH_CONFIG_SELECTION.reset(token)
diff --git a/tools/generate_session_peer.py b/tools/generate_session_peer.py
index 9bf7d3d..4817035 100644
--- a/tools/generate_session_peer.py
+++ b/tools/generate_session_peer.py
@@ -20,6 +20,7 @@
"claude.py",
"target_generation.py",
"replies.py",
+ "ssh_config.py",
"ssh.py",
"ssh_identity.py",
"output.py",