X3DH 鍵合意プロトコルをRubyで実装する

X3DH 鍵合意プロトコル（X3DH Key Agreement Protocol）をRubyで実装してみる。

このプロトコルはSignalプロトコルの主要な構成要素の1つであり、Signalの公式Webサイトに[技術的なホワイトペーパー](https://signal.org/docs/specifications/x3dh/)がある。今回はこのドキュメントにできるだけ沿った形での実装を目指す。

動くコードは https://github.com/ts-3156/x3dh-ruby/ に置いています。

## X3DH 鍵合意プロトコルで何ができるか

X3DH 鍵合意プロトコルを正しく実装できれば、Signalアプリのような匿名メッセンジャーアプリを作ることができます。

現代的で堅牢な暗号通信を実装するにはX3DH 鍵合意プロトコルはほぼ必須の要素です。

暗号通信において、X3DH 鍵合意プロトコルは「メッセージ送受信の開始時」に使用されます。


## X3DH 鍵合意プロトコルとは

X3DH 鍵合意プロトコル（X3DH Key Agreement Protocol）は、安全なエンドツーエンド暗号通信のための鍵合意プロトコルです。事前の共有鍵なしで、2つの当事者間の共有秘密鍵を確立します。長期キー（IDキー）、署名済みキー、一時キー、ワンタイムキーを組み合わせ、前方秘匿性と暗号化の否認可能性を提供します。Signalプロトコルなどで使用されます。

X3DHは非同期で動作するように設計されており、相手がオフラインであっても将来の通信用に共有秘密鍵の確立とメッセージ送信ができます。


:::note alert
暗号化アルゴリズムや暗号化プロトコルにはたくさんの種類があり、誤用しやすく、潜在的な落とし穴がたくさんあります。

安全なセキュリティシステムの設計を正しく行うことは非常に難しく、通常はセキュリティ専門家の領域となります。

設計や実装にほんの小さな間違いがあるだけでセキュリティが完全に無効になる可能性があります。

この記事の作者はセキュリティ専門家ではありません。できるだけ正しい情報を書くよう心掛けていますが、書かれた情報の正しさはご自身で検証してください。
:::

## 使用するキーの種類

X3DH 鍵合意プロトコルでは以下の4つの鍵を用いる。

| 鍵の種類 |	説明 |
|---|---|
| Identity Key (IK) |	ユーザーを識別するための長期的な鍵 |
| Ephemeral Key (EK) |	セッションごとに生成する一時鍵 |
| Signed Prekey (SPK) |	定期的に更新する準長期鍵 |
| One-time Prekeys (OPK) |	使い捨ての鍵（複数用意する） |

アリスからボブにメッセージを送信するには以下の鍵が必要になる。それぞれ単一の鍵ではなく鍵ペアなので、公開鍵と対応する秘密鍵がある。

| 鍵の名前 |	役割 |
|---|---|
| IKA |	アリスのIDキー |
| EKA |​	アリスの一時的な鍵 |
| IKB |​	ボブのIDキー |
| SKB |​	ボブの署名用のキー。公式ドキュメントには無いが実装上は必要 |
| SPKB |​	ボブの署名済みプレキー |
| OPKB |​	ボブのワンタイムプレキー |

X3DH 鍵合意プロトコルでは、X25519またはX448のどちらかの曲線を1つだけ選んで使用する。今回はX25519形式の公開鍵暗号を選択する。

Rubyでの実装にはlibsodiumのRubyバインディングであるRbNaCl gemを用いる。そうするとそれぞれの鍵は以下のように生成できる。

```ruby
require 'rbnacl'

# アリスのIDキー
ika = RbNaCl::PrivateKey.generate
ika_pub = ika.public_key

# アリスの一時的なキー
eka = RbNaCl::PrivateKey.generate
eka_pub = eka.public_key

# ボブのIDキー
ikb = RbNaCl::PrivateKey.generate
ikb_pub = ikb.public_key

# ボブの署名用のキー
skb = RbNaCl::SigningKey.generate
skb_pub = skb.verify_key

# 曲線の種類（X25519またはX448）を表す1バイトの定数とRFC7748で規定されている
# u座標のリトルエンディアンエンコードが推奨されている。今回はこれだけよいはず
def encode_u_coordinate(key)
  # X25519は1、X448は2とする
  '1' + key.to_bytes
end

# ボブの署名済みプレキーと署名
spkb = RbNaCl::PrivateKey.generate
spkb_pub = spkb.public_key
spkb_signature = skb.sign(encode_u_coordinate(spkb_pub))

# ボブのワンタイムプレキー（複数）
opkb_set = 100.times.map do
  RbNaCl::PrivateKey.generate
end
opkb_pub_set = opkb_set.map(&:public_key)

opkb = opkb_set[0]
opkb_pub = opkb_pub_set[0]
```

暗号化用の鍵にはX25519形式を使用している。`RbNaCl::PrivateKey`は`Boxes::Curve25519XSalsa20Poly1305::PrivateKey`のエイリアスである。

署名用の鍵にはEd25519形式を使用している。`RbNaCl::SigningKey`は`Signatures::Ed25519::SigningKey`のエイリアスである。

#### 暗号化鍵と署名鍵を分けた理由

今回の実装ではIDキーと署名用のキーを別に用意している。分けた理由は、X3DH 鍵合意プロトコルで指定されているXEdDSA署名を直接的に扱う方法がlibsodiumに実装されていないため。
これらの鍵を1つにしたい場合は、`RbNaCl::SigningKey#to_curve25519_private_key`で署名用のキーをX25519形式のキーに変換することができる。
一般論として、1つの鍵を複数の用途（暗号化鍵と署名鍵）に利用することは推奨されていない。ただし、Signalプロトコルでは意図的にこれを研究してきており、その場合はなんの問題もない。

#### 公開鍵の比較は他の方法で行う必要がある

IDキーの公開鍵は他の方法（例：QRコードの読み込み）で比較を行う必要がある。そうしないと、誰と通信しているのかについて暗号化による保証を受けることができない。



## キーの公開



ボブはIDキーとプレキーを事前にサーバーで公開する。

```ruby
require 'base64'

def encode64(data)
  Base64.strict_encode64(data)
end

# IDキー（と署名用のキー）のアップロードは今回のプロトコル開始よりも前に一度だけ
server.upload(
  ikb_pub: encode64(ikb_pub.to_bytes),
  skb_pub: encode64(skb_pub.to_bytes)
)

# 署名済みプレキーとプレキー署名は一定の間隔（週に1回または月に1回）でアップロードする
server.upload(
  spkb_pub: encode64(spkb_pub.to_bytes),
  spkb_signature: encode64(spkb_signature)
)

# ワンタイムプレキーは他のタイミング（ワンタイムプレキーの数が少なくなってきたと
# サーバーから通知があった時など）でアップロードする
server.upload(opkb_pub_set: opkb_pub_set.map { |key| encode64(key.to_bytes) })
```

#### IDキーの取り扱い

IDキーは長期的に使用する。原則としてユーザー登録の時に一度だけアップロードする。

#### 署名済みプレキーの取り扱い（サーバー側）

署名済みプレキーは中期的に使用する。週に1回、または月に1回アップロードする。新しい署名済みプレキーとプレキー署名をアップロードしたら、サーバー内の古いキーは削除する必要がある。

#### 署名済みプレキーの取り扱い（署名した側）

新しい署名済みプレキーのアップロード後、ボブが持っている「以前の署名済みプレキーに対応する秘密鍵」は一定期間保持し、その後に削除する。そうすることで配送中に遅延したメッセージを処理することができる。削除をしないと前方秘匿性が低下することになる。

#### ワンタイムプレキーの取り扱い

ワンタイムプレキーの公開鍵は一度使用したら削除する。秘密鍵はそれを使用したメッセージを受信したら削除する。

## 最初のメッセージの送信

ボブとのX3DH 鍵合意を実行するために、アリスはサーバーに接続し、次の値を含むプレキーバンドルを取得する。

```ruby
def decode64(data)
  Base64.strict_decode64(data)
end

res = server.download

prekey_bundle = {
  ikb_pub: decode64(res[:ikb_pub]),
  skb_pub: decode64(res[:skb_pub]),
  spkb_pub: decode64(res[:spkb_pub]),
  spkb_signature: decode64(res[:spkb_signature]),
  opkb_id: 'xxx-yyy-ddd', # どのプレキーを使用したかを示すID
  opkb_pub: decode64(res[:opkb_pub]) # プロトコル上は必須ではない
}
```

サーバーは、ボブのワンタイムプレキーが存在する場合はその1つを提供し、その後に削除する。ボブのワンタイムプレキーがすべて削除されている場合はプレキーバンドルにワンタイムプレキーを含めなくてもよい。

ただし、ここでワンタイムプレキーを含めないと前方秘匿性が少し低下するため、実装する上では必須にした方がよいかもしれない。

`opkb_id`は使用したワンタイムプレキーがどれなのかを示すID。これがあることでメッセージの送信時にアリスがどのプレキーを使用したかをボブに伝えることができる。

----

プレキーバンドルを取得した後、アリスはボブのプレキー署名を検証する。検証に失敗した場合はプロトコルを中止する。成功した場合は一時的なキーを生成する。

```ruby
skb_pub = RbNaCl::VerifyKey.new(prekey_bundle[:skb_pub])
spkb_signature = prekey_bundle[:spkb_signature]
spkb_pub = RbNaCl::PublicKey.new(prekey_bundle[:spkb_pub])

begin
  skb_pub.verify(spkb_signature, encode_u_coordinate(spkb_pub))
rescue RbNaCl::BadSignatureError => e
  exit
end

# アリスの一時的な鍵はここで初めて生成する
eka = RbNaCl::PrivateKey.generate
eka_pub = eka.public_key
```

これまでに得たキーから、アリスは共通鍵を導出することができる。

```ruby
# アリスのIDキーとボブの署名済みプレキー
dh1 = RbNaCl::GroupElement.new(spkb_pub).mult(ika)

# アリスの一時的なキーとボブのIDキー
dh2 = RbNaCl::GroupElement.new(ikb_pub).mult(eka)

# アリスの一時的なキーとボブの署名済みプレキー
dh3 = RbNaCl::GroupElement.new(spkb_pub).mult(eka)

# アリスの一時的なキーとボブのワンタイムプレキー
dh4 = RbNaCl::GroupElement.new(opkb_pub).mult(eka)

require 'openssl'

def hkdf_sha256(km, n)
  ikm = ['ff' * 32].pack('H*') + km
  opt = {
    salt: ['00' * 32].pack('H*'),
    info: "MyProtocol key#{n + 1}".unpack1('H*'),
    length: 32,
    hash: 'SHA256'
  }
  OpenSSL::KDF.hkdf(ikm, **opt)
end

# 導出した共通鍵からサブキーを10個派生させている
shared_keys = 10.times.map do |n|
  hkdf_sha256([dh1, dh2, dh3, dh4].map(&:to_bytes).join, n)
end

# 使用済みの値を削除する
eka = dh1 = dh2 = dh3 = dh4 = nil
```

鍵交換のためにX25519形式の乗算を直接使用している。`RbNaCl::GroupElement`は`GroupElements::Curve25519`のエイリアスである。

dh1からdh4はそれぞれ暗号的な意味合いが異なる。dh1とdh2で相互認証を提供し、dh3とdh4で前方秘匿性を提供している。

ここまでできたら、アリスはボブに以下の形式で最初のメッセージを送信する。

```ruby
ad = encode_u_coordinate(ika_pub) + encode_u_coordinate(ikb_pub)

def encrypt_message(msg, key, ad)
  cipher = RbNaCl::AEAD::XChaCha20Poly1305IETF.new(key)
  nonce = RbNaCl::Random.random_bytes(cipher.nonce_bytes)
  [cipher.encrypt(nonce, msg, ad), nonce]
end

encrypted_message, nonce = encrypt_message('Hello!', shared_keys[0], ad)

payload = {
  ika_pub: encode64(ika_pub),
  eka_pub: encode64(eka_pub),
  opkb_id: prekey_bundle[:opkb_id],
  message: encode64(encrypted_message),
  nonce: encode64(nonce)
}

alice.send_message(payload, to: 'Bob')
```

`ad`はAEAD暗号化方式におけるAssociated dataであり、ユーザー名、証明書、その他の識別情報などの追加情報を含めることもできる。このデータは暗号化されないため秘匿情報を含めてはいけない。

今回使用した共通鍵はセキュリティ上の考慮事項に従った上で引き続き使用できる。共通鍵を派生させて使用する場合は使用した共通鍵を識別するIDも必要になるが、このサンプルコードでは送信するデータに含めていない。

AEAD暗号化方式として今回はXChaCha20-Poly1305を使用する。AEAD暗号化方式としてはTLSで採用されているAES256-GCMが有名だが今回の用途には適さない。

#### AES256-GCMが今回の用途に適さない理由

AES256-GCMはnonceが96ビットと小さく、衝突の危険性がある。また、同じ鍵で暗号化したデータが350GBに達するまでに鍵を作り直すことが推奨されている。



## 最初のメッセージの受信

ボブはアリスからのメッセージの受信後、アリスが使用したDH法と同じ計算を対応する鍵で行い、同じ共通鍵を導出する。

```ruby
res = bob.receive_message(from: 'Alice')

ika_pub = decode64(res[:ika_pub])
eka_pub = decode64(res[:eka_pub])

# 自分の秘密鍵を読み込む（ファイルである必要はない）
ikb = RbNaCl::PrivateKey.new(File.read('...'))
spkb = RbNaCl::PrivateKey.new(File.read('...'))

# ワンタイムプレキーは複数あるため、使用したものを選ぶ必要がある
opkb = OnetimePrekey.find_by(id: res[:opkb_id])

# ボブの署名済みプレキーとアリスのIDキー
dh1 = RbNaCl::GroupElement.new(ika_pub).mult(spkb)

# ボブのIDキーとアリスの一時的なキー
dh2 = RbNaCl::GroupElement.new(eka_pub).mult(ikb)

# ボブの署名済みプレキーとアリスの一時的なキー
dh3 = RbNaCl::GroupElement.new(eka_pub).mult(spkb)

# ボブのワンタイムプレキーとアリスの一時的なキー
dh4 = RbNaCl::GroupElement.new(eka_pub).mult(opkb)

# アリスが導出したshared_keys[0]と同じものになる
shared_key =
  hkdf_sha256([dh1, dh2, dh3, dh4].map(&:to_bytes).join, 0)

ikb_pub = RbNaCl::PublicKey.new(File.read('...'))
ad = encode_u_coordinate(ika_pub) + encode_u_coordinate(ikb_pub)

def decrypt_message(ciphertext, key, nonce, ad)
  cipher = RbNaCl::AEAD::XChaCha20Poly1305IETF.new(key)
  cipher.decrypt(nonce, ciphertext, ad)
end

received_message = decode64(res[:message])
nonce = decode64(res[:nonce])

begin
  decrypted_message = decrypt_message(received_message, shared_key, nonce, ad)
  # => 'Hello!'
rescue RbNaCl::CryptoError => e
  shared_key = nil
  exit
end

# 使用済みの値を削除する
dh1 = dh2 = dh3 = dh4 = nil
OnetimePrekey.where(id: res[:opkb_id]).destroy_all
```

ここまでの手順で共通鍵の共有ができたので、これ以降は今回生成した共通鍵（またはその派生鍵）を用いて暗号通信を行うことができる。

SignalプロトコルではX3DH 鍵合意プロトコル以降のプロトコルについても規定されており、Signalプロトコルに沿った形で通信するのであればこれ以降はダブルラチェットアルゴリズムで暗号通信を行う必要がある。


#### サーバーの役割は公開鍵の保存のみ

X3DH 鍵合意プロトコルにおいて、サーバーの役割は公開鍵の保存のみであり、メッセージ本文をサーバーから覗き見ることはできない。また、サーバーとの通信が発生するのはプレキーバンドル取得時の一度きりである。このことから、サーバーはアリスとボブの通信内容だけでなく、通信を行ったかどうかすら知ることができない。




## セキュリティ上の考慮事項

[公式ドキュメント](https://signal.org/docs/specifications/x3dh/#security-considerations) をご確認ください。


## 参考にした情報

https://signal.org/docs/specifications/x3dh/

https://github.com/RubyCrypto/rbnacl

https://doc.libsodium.org/

https://github.com/ts-3156/x3dh-ruby/