AIアシスタントがSSHターミナル上でNginxの502エラーを調べてくれるなら、ログをコピーして別のウィンドウへ貼る手間は減るでしょう。ただし、ターミナルに表示されるのはエラーだけではありません。環境変数、社内ホスト名、設定ファイル、誤って出力されたアクセストークンが混ざることもあります。

ファイルを読めてコマンドも実行できるアシスタントに対しては、回答の正確さ以外にも確認すべきことがあります。ターミナルの内容は誰に送られ、どの操作が再確認なしで進むのか。

Hatoba は、その境界をソースコードで検討しやすい例です。Show HNの投稿で紹介されたこのオープンソースSSHクライアントには、AIアシスタント、ローカルVault、自分のCloudflareアカウント上のWorker + D1によるE2EE同期があります。ただし「同期内容が暗号化される」ことと「AIに関わる処理全体が保護される」ことは同義ではありません。

最初に4つの結論です。

  • E2EEが守るのはVaultの内容であり、すべてのクラウド情報ではありません。 Workerは認証情報やitem ID、revision、時刻などの同期メタデータを扱います。
  • 読み取り専用ツールでも、内容はモデルへ送られ得ます。 端末画面、ログ、添付ファイル、read_fileの結果がモデルのコンテキストに入る経路は、Cloudflareの暗号化同期とは別です。
  • Manualモードはすべての読み取りを逐次承認するものではありません。 読み取り、書き込み、Auto、Bypass、MCPで条件が異なります。
  • 認証済みの暗号文が最新とは限りません。 AES-GCMのタグだけでは、古い正当な暗号文の再送を検出できません。
リスク越える信頼境界固定バージョンの実装から分かること導入前の確認
クラウド上のVault流出ローカルVault → Worker/D1内容はクライアント側で暗号化。同期メタデータは一部可視鍵、メタデータ、復元、バックアップ
本番ログや認証情報のモデル送信SSH画面/ファイル → モデル提供者画面、添付、ツールの結果は同期とは別に送信され得るコンテキスト設定、マスキング、提供者
AIによる危険なコマンド実行AIツール → SSHホスト読み取りと変更/実行では承認条件が違う最小権限ユーザー、Manualでの確認
古い同期データの再送Worker/D1 → ローカルVault認証成功だけではバージョンの新しさを保証しない信頼できるバージョン状態と復元方法

対象は Hatoba v0.3.0の固定コミット 5e9b35302 です。Rustの暗号処理、WorkerのDBスキーマ、フロントエンドの承認判断を確認し、暗号の性質は別途作成したPythonの小さなモデルで試しました。Hatoba本体のインストールやSSH実機テスト、ペネトレーションテストではありません。

E2EEの内側に残るデータと、D1から見えるメタデータ

まず鍵の構成を確認します。Rustのcrypto.rsでは、マスターパスワードからArgon2idでmaster_keyを導出し、HKDF-SHA256で用途の異なる鍵を分離しています。重要なラベルは次の2つです。

enc_key: hkdf_sha256(master_key, "hatoba/enc/v1"),
auth_key: hkdf_sha256(master_key, "hatoba/auth/v1"),

enc_keyはランダムに生成されたvault_keyをローカルで包むために使われ、同期する各itemの内容はvault_keyを用いてAES-256-GCMで暗号化されます。AADはhatoba/item/v1/{item_id}に結び付けられるため、あるitemの暗号文を別のitem IDのものとして差し替えると認証に失敗します。

一方のauth_keyはログイン用です。Workerの認証ルートを見ると、クライアントが送った派生認証キーのSHA-256を保存済みauth_hashと比較し、成功するとセッショントークンを発行します。ログイン用の秘密情報とVaultの復号鍵は同一ではないのが設計上の要点です。それでも認証キー、リカバリーコード、有効なセッションは厳重な管理が必要です。

次にD1のitemsテーブルです。コメントとインデックスを省いて主要列だけを示します。

CREATE TABLE items (
  id         TEXT PRIMARY KEY,
  envelope   TEXT,
  revision   INTEGER NOT NULL,
  seq        INTEGER NOT NULL,
  deleted    INTEGER NOT NULL DEFAULT 0,
  updated_at INTEGER NOT NULL
);

envelopeには暗号化したitemの内容が入ります。しかしid、revision、seq、deleted、updated_atは同期や競合処理のため、サーバーから扱える列です。item数、暗号文のサイズ、更新時刻や利用状況も、暗号化された業務データそのものとは別の情報です。公式セキュリティ仕様の4.5節にもサーバーから見える範囲が説明されています。

つまり、E2EEは内容の機密性を守る仕組みであって、利用した形跡をすべて隠す仕組みではありません。 正しく暗号化されたSSH秘密鍵はD1のコピーだけで直接読めない設計ですが、同期の頻度やセッションの問題まで消えるわけではありません。

さらにデータの新しさという別の問題があります。Hatobaの脅威モデルは、悪意のあるWorkerによるデータ削除、サービス拒否、古いデータへの巻き戻しを認めています。実装状況表でもrollback detectionは未完了のP2項目です。暗号の強度とロールバック対策は、別々に評価する必要があります。

読み取りツールからモデルへ出ていく別経路

本番のNginxログに社内ホスト名や認証情報の一部が出ていたとします。SSH接続が暗号化され、Vault同期がE2EEでも、その文字列がモデルへのリクエストに入れば、今度はモデル提供者が受信者になります。

HatobaのREADMEには、メッセージ送信時にターミナル画面を含められ、デフォルトの添付動作は設定で変更できることが書かれています。ユーザーの文章、添付ファイル、ツールの戻り値もモデルへ送られる可能性があります。これはVault全体を自動アップロードするという意味ではなく、各リクエストにどの情報がコンテキストとして載るかが問題です。クラウド同期を切ってもモデルとの通信まで止まるとは限りません。

承認条件のコードはさらに具体的です。フロントエンドtools.tsのneedsApproval()には、通常の読み取りツールは列挙されていません。

export function needsApproval(kind: ToolKind, mode: AiPermissionMode): boolean {
  if (mode === "bypass") return false;
  return kind === "run_command" || kind === "send_input" ||
    kind === "edit_file" || kind === "write_file" ||
    kind === "fetch_url" || kind === "mcp";
}

Manualではrun_command、send_input、edit_file、write_file、fetch_urlなどが通常の承認対象です。対してread_terminal、read_file、grep、globなどは、必ずしも毎回ダイアログを出しません。既に保存された許可によって次の確認が省略される場合もあります。MCPツールは別の条件があり、同じファイルのmustAsk()はalways_askとalways_allowも判別します。

「読み取りのみ」=「データはローカルのまま」ではありません。 読める権限と、その内容をモデルへ送ってよいかは違う判断です。本番環境の機密情報が関わるなら、Manualの承認ダイアログを見る前に、画面コンテキストとモデルの受信先を調べるべきです。

Autoには、一部のツール呼び出しで承認を省略できるか判定するjudgeが入ります。公式状況表の単体/モック/限定的なテストから、悪意ある出力によるprompt injectionをすべて防げるとは言えません。Bypassは一般的な承認を省略しますが、MCPのalways_askには例外が残ります。いずれもSSHアカウントとOSの最小権限が独立した防御になります。

改ざんされていない暗号文も古いことがある:独立Pythonテスト6件

区別したかったのは、改ざんされたAES-GCM暗号文なら認証に失敗する一方、以前に正しく生成された暗号文は、新しい版が存在しても認証できるということです。

公開ソースのArgon2id、HKDF、AES-GCM、item-ID AADを参考に、独立したPythonモデルを作成しました。HatobaのRustやWorkerコードは呼び出していません。保存された実行環境はPython 3.13.5、argon2-cffi 25.1.0、cryptography 46.0.4。結果は 6件成功、失敗0件、終了コード0 でした。

独立モデルの検査記録された結果
enc_keyとauth_keyの分離異なる32バイトの鍵が得られた
暗号化・復号と平文の検査ダミーデータを復元でき、envelopeには平文がない
AES-GCM暗号文の一部を変更InvalidTagで復号に失敗
別のitem_idとして復号AAD不一致でInvalidTag
同一itemの古い版と新しい版を生成どちらの正しい暗号文も復号できた
同期行のメタデータを模擬revision、updated_atが見え、itemの平文は見えない

5番目は、実際に実行したスクリプトの次のアサーションで確認しました。

old = seal_item(self.vault_key, self.item_id, {"generation": 1})
new = seal_item(self.vault_key, self.item_id, {"generation": 2})

self.assertEqual(open_item(self.vault_key, self.item_id, new), {"generation": 2})
self.assertEqual(open_item(self.vault_key, self.item_id, old), {"generation": 1})

これはAES-GCMの欠陥ではありません。新旧どちらの暗号文も改ざんされていないから認証は成功します。認証タグだけでは、クライアントが過去に受け入れた最大の世代を覚えられません。悪意あるサーバーが古い正当なitemを返すのを検出するには、同時に巻き戻せない信頼できる状態やバージョンアンカーが必要です。

有効な暗号文 ≠ 最新のデータ。 Hatobaがrollback detectionを別項目にしている背景を、この独立モデルは説明できます。ただし、Hatoba本体の同期プロトコルに対する実際の攻撃再現ではありません。

保存されたunittest出力の末尾は以下です。

----------------------------------------------------------------------
Ran 6 tests in 0.100s

OK

この短い実行時間は合成テストの値で、Hatobaの暗号化速度やAI応答遅延ではありません。スクリプト全体を記事末尾に掲載します。

AIに本番SSHを許可する前の確認手順

HatobaにはSSH、SFTP、Vault、AIによるファイル操作、クラウド同期という実装があります。ただし「実装された」「モックで確認した」「実サービスで検証した」は違います。v0.3.0のdocs/status.mdを見ると、その差がわかります。

機能当時の公式な検証範囲
Worker同期実装と関連テストあり。本記事に実Cloudflareデプロイの証拠はない
D1 DirectSQLiteを使ったREST模擬で検査。実D1は未検証
AIプロバイダー、ChatGPTログインプロトコル/モックテストあり。実アカウントとの組み合わせは要確認
AIファイル操作、承認UIスクリプト、OpenSSH統合またはモックのテスト。実モデルのデスクトップ操作は未完了
macOSApple Silicon向け実験版、ad-hoc署名で公証なし。実機の検証が残る

チームで試すなら、いきなり本番の管理者アカウントを渡さず、次の順で範囲を広げます。

  1. ホストと権限を分離する。 機密データを持たない検証用マシンを選び、非root・最小権限のSSHユーザーで開始。不要な同期を無効にし、限定的な読み取り診断から試す。
  2. 外向きのデータを確認する。 ターミナル画面、ログ、添付、ツール結果のどれが送信されるか、モデル提供者と保持方針を調べる。顧客データや本番の資格情報は分類・マスキング後に扱い、条件を満たさなければ外部モデルに原文を送らない。
  3. Manualで承認を確かめる。 検証マシンで無害なファイル書き込みを提案し、確認ダイアログ、保存済みの「常に許可」、復元方法を確かめる。Auto、Bypass、MCP、外部Skillは別に評価する。
  4. 復元とバックアップを点検する。 マスターパスワードとリカバリーコードを安全に分けて保管し、端末紛失時のセッション失効、利用可能なエクスポート/復元、オフラインバックアップを確認。両方の認証情報を失えば、Hatobaの設計上、既存Vaultは復号できない。

書き込みが承認制でも、リモートファイルや端末出力にモデルを誘導する不正な指示が含まれることはあります。承認UIとprompt injection対策を混同しないことが大切です。

AI付きSSHツールの利点は、診断と次の操作の間にあるコピー作業を減らせることです。ただし導入の判断は「E2EE」という表示ではなく、誰がデータを受け取り、誰がコマンドを実行でき、失敗時にどこまで復元できるかで決めたいところです。上の4分類表は、ほかのAIターミナル製品の評価にも使えます。

AIへSSH権限を直接渡すのではなく、生成された操作を別のデプロイ経路で審査したい場合には、ChatGPT Web + GitHub Self-hosted Runnerで異なる設計を取り上げています。

独立したPython暗号モデルの完全なソースを表示

合成データだけを使うサンプルです。Hatobaや外部サービスには接続しません。隔離したPython環境にargon2-cffiとcryptographyを入れ、hatoba_crypto_boundary_fixture.pyとして保存した後、python3 -m unittest -v hatoba_crypto_boundary_fixture.pyで確認できます。

"""Isolated, source-adapted crypto boundary demonstration; NOT Hatoba execution.

Models only the publicly documented v0.3.0 key labels, algorithms and item AAD.
No Hatoba source is imported or run, no network, accounts, SSH or real data.
"""
import base64
import json
import os
import unittest
from argon2.low_level import Type, hash_secret_raw
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF


def derive_keys(password: bytes, salt: bytes):
    master_key = hash_secret_raw(
        password, salt, time_cost=3, memory_cost=64 * 1024,
        parallelism=4, hash_len=32, type=Type.ID,
    )
    def derive(info):
        return HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=info).derive(master_key)
    return derive(b"hatoba/enc/v1"), derive(b"hatoba/auth/v1")


def seal_item(vault_key: bytes, item_id: str, content: dict) -> dict:
    nonce = os.urandom(12)
    aad = ("hatoba/item/v1/" + item_id).encode()
    ciphertext = AESGCM(vault_key).encrypt(
        nonce, json.dumps(content, sort_keys=True).encode(), aad,
    )
    return {"v": 1, "n": base64.b64encode(nonce).decode(),
            "c": base64.b64encode(ciphertext).decode()}


def open_item(vault_key: bytes, item_id: str, envelope: dict) -> dict:
    nonce = base64.b64decode(envelope["n"])
    ciphertext = base64.b64decode(envelope["c"])
    aad = ("hatoba/item/v1/" + item_id).encode()
    return json.loads(AESGCM(vault_key).decrypt(nonce, ciphertext, aad))


class BoundaryFixture(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.enc_key, cls.auth_key = derive_keys(b"test-only-long-password", b"16-byte-testsalt")
        cls.vault_key = os.urandom(32)
        cls.item_id = "test-item-1"
        cls.content = {"type": "host", "hostname": "fixture.example.invalid", "secret": "synthetic-only"}

    def test_distinct_auth_and_encryption_keys(self):
        self.assertEqual(len(self.enc_key), 32)
        self.assertEqual(len(self.auth_key), 32)
        self.assertNotEqual(self.enc_key, self.auth_key)

    def test_roundtrip_and_no_plaintext_in_envelope(self):
        envelope = seal_item(self.vault_key, self.item_id, self.content)
        self.assertNotIn("fixture.example.invalid", json.dumps(envelope))
        self.assertNotIn("synthetic-only", json.dumps(envelope))
        self.assertEqual(open_item(self.vault_key, self.item_id, envelope), self.content)

    def test_tampered_ciphertext_rejected(self):
        envelope = seal_item(self.vault_key, self.item_id, self.content)
        tampered = dict(envelope)
        c = bytearray(base64.b64decode(envelope["c"]))
        c[-1] ^= 0x01
        tampered["c"] = base64.b64encode(c).decode()
        with self.assertRaises(InvalidTag):
            open_item(self.vault_key, self.item_id, tampered)

    def test_other_item_id_rejected_by_aad(self):
        envelope = seal_item(self.vault_key, self.item_id, self.content)
        with self.assertRaises(InvalidTag):
            open_item(self.vault_key, "another-item", envelope)

    def test_stale_but_valid_envelope_still_decrypts(self):
        # An authenticated envelope is not a freshness/revision check.
        old = seal_item(self.vault_key, self.item_id, {"generation": 1})
        new = seal_item(self.vault_key, self.item_id, {"generation": 2})
        self.assertEqual(open_item(self.vault_key, self.item_id, new), {"generation": 2})
        self.assertEqual(open_item(self.vault_key, self.item_id, old), {"generation": 1})

    def test_server_row_exposes_sync_metadata_not_item_plaintext(self):
        envelope = seal_item(self.vault_key, self.item_id, self.content)
        row = {"id": self.item_id, "envelope": json.dumps(envelope),
               "revision": 2, "seq": 6, "deleted": False, "updated_at": 42}
        serialized = json.dumps(row)
        self.assertIn('"revision": 2', serialized)
        self.assertIn('"updated_at": 42', serialized)
        self.assertNotIn("fixture.example.invalid", serialized)
        self.assertEqual(json.loads(row["envelope"]), envelope)


if __name__ == "__main__":
    unittest.main()