1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
"""
MeshBay cryptographic primitives.
Validated in Spike 1 and Spike 6 of the POC.
All operations use PyCA cryptography (OpenSSL-backed, hardware-accelerated).
"""
import os
import base64
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey, X25519PublicKey
from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import hashes, serialization
import blake3
# ── Key serialisation helpers ─────────────────────────────────────────────────
def sk_to_raw(sk: Ed25519PrivateKey | X25519PrivateKey) -> bytes:
return sk.private_bytes(
serialization.Encoding.Raw,
serialization.PrivateFormat.Raw,
serialization.NoEncryption(),
)
def pk_to_raw(pk: Ed25519PublicKey | X25519PublicKey) -> bytes:
return pk.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
def pk_to_b64(pk: Ed25519PublicKey | X25519PublicKey) -> str:
return base64.b64encode(pk_to_raw(pk)).decode()
def sk_to_b64(sk: Ed25519PrivateKey | X25519PrivateKey) -> str:
return base64.b64encode(sk_to_raw(sk)).decode()
# ── GEK (Group Encryption Key) ────────────────────────────────────────────────
def generate_gek() -> bytes:
"""Generate a fresh 256-bit Group Encryption Key."""
return ChaCha20Poly1305.generate_key()
def chunk_key(gek: bytes, file_hash: bytes, chunk_index: int) -> bytes:
"""Derive a per-chunk encryption key from the GEK (deterministic)."""
return HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=None,
info=b"file:" + file_hash + b":chunk:" + chunk_index.to_bytes(4, "big"),
).derive(gek)
def encrypt_chunk(key: bytes, plaintext: bytes) -> tuple[bytes, bytes]:
"""Encrypt plaintext with ChaCha20-Poly1305. Returns (nonce, ciphertext)."""
nonce = os.urandom(12)
ct = ChaCha20Poly1305(key).encrypt(nonce, plaintext, None)
return nonce, ct
def decrypt_chunk(key: bytes, nonce: bytes, ciphertext: bytes) -> bytes:
"""Decrypt ciphertext. Raises InvalidTag on authentication failure."""
return ChaCha20Poly1305(key).decrypt(nonce, ciphertext, None)
def file_hash(path_or_bytes) -> bytes:
"""Compute blake3 hash of a file (bytes or path-like)."""
if isinstance(path_or_bytes, (str, bytes)) and not isinstance(path_or_bytes, bytes):
import pathlib
data = pathlib.Path(path_or_bytes).read_bytes()
elif isinstance(path_or_bytes, bytes):
data = path_or_bytes
else:
data = path_or_bytes.read_bytes()
return blake3.blake3(data).digest()
# ── GEK wrapping (ECIES-like) ─────────────────────────────────────────────────
GEK_WRAP_INFO = b"meshbay:gek_wrap:v1"
def wrap_gek(gek: bytes, pk_recipient: bytes) -> dict:
"""
Wrap a GEK for a recipient using ephemeral X25519 + HKDF + ChaCha20-Poly1305.
Protocol:
1. Generate ephemeral (sk_eph, pk_eph)
2. shared = X25519(sk_eph, pk_recipient)
3. wrap_key = HKDF(shared, salt=pk_eph, info=GEK_WRAP_INFO)
4. wrapped = ChaCha20-Poly1305(wrap_key).encrypt(nonce, gek, aad=pk_recipient)
The hub stores {pk_eph, nonce, wrapped} — opaque, cannot decrypt.
"""
sk_eph = X25519PrivateKey.generate()
pk_eph_raw = pk_to_raw(sk_eph.public_key())
shared = sk_eph.exchange(X25519PublicKey.from_public_bytes(pk_recipient))
wrap_key = HKDF(
algorithm=hashes.SHA256(), length=32,
salt=pk_eph_raw, info=GEK_WRAP_INFO,
).derive(shared)
nonce = os.urandom(12)
wrapped = ChaCha20Poly1305(wrap_key).encrypt(nonce, gek, pk_recipient)
return {
"pk_eph_b64": base64.b64encode(pk_eph_raw).decode(),
"nonce_b64": base64.b64encode(nonce).decode(),
"wrapped_b64": base64.b64encode(wrapped).decode(),
}
def unwrap_gek(bundle: dict, sk_recipient: bytes, pk_recipient: bytes) -> bytes:
"""
Unwrap a GEK bundle using the recipient's X25519 private key.
Raises InvalidTag if the key is wrong or the bundle was tampered.
"""
pk_eph_raw = base64.b64decode(bundle["pk_eph_b64"])
nonce = base64.b64decode(bundle["nonce_b64"])
wrapped = base64.b64decode(bundle["wrapped_b64"])
shared = X25519PrivateKey.from_private_bytes(sk_recipient).exchange(
X25519PublicKey.from_public_bytes(pk_eph_raw)
)
wrap_key = HKDF(
algorithm=hashes.SHA256(), length=32,
salt=pk_eph_raw, info=GEK_WRAP_INFO,
).derive(shared)
return ChaCha20Poly1305(wrap_key).decrypt(nonce, wrapped, pk_recipient)
# ── Keystore (local key storage) ──────────────────────────────────────────────
# Argon2id parameters — calibrate to ~500ms on target hardware before production.
# POC measured 78ms with these; increase memory_cost to 262144 (256MB) for prod.
ARGON2_ITERATIONS = 3
ARGON2_MEMORY_COST = 65536 # 64 MB — increase to 262144 for production
ARGON2_LANES = 4
ARGON2_KEY_LENGTH = 32
def derive_keystore_key(password: str, salt: bytes) -> bytes:
"""Derive AES-256 key from password using Argon2id."""
return Argon2id(
salt=salt,
length=ARGON2_KEY_LENGTH,
iterations=ARGON2_ITERATIONS,
lanes=ARGON2_LANES,
memory_cost=ARGON2_MEMORY_COST,
).derive(password.encode())
def encrypt_keystore(plaintext: bytes, key: bytes) -> tuple[bytes, bytes, bytes]:
"""Encrypt keystore blob with AES-256-GCM. Returns (iv, ciphertext, tag)."""
iv = os.urandom(12)
enc = Cipher(algorithms.AES(key), modes.GCM(iv)).encryptor()
ct = enc.update(plaintext) + enc.finalize()
return iv, ct, enc.tag
def decrypt_keystore(iv: bytes, ciphertext: bytes, tag: bytes, key: bytes) -> bytes:
"""Decrypt keystore blob. Raises on authentication failure."""
dec = Cipher(algorithms.AES(key), modes.GCM(iv, tag)).decryptor()
return dec.update(ciphertext) + dec.finalize()
# ── Chunk signing ─────────────────────────────────────────────────────────────
def sign_chunk(sk_node: Ed25519PrivateKey, chunk_index: int,
nonce: bytes, ct_hash: bytes) -> bytes:
"""Sign chunk metadata. Payload: chunk_index || nonce || ct_hash."""
payload = chunk_index.to_bytes(4, "big") + nonce + ct_hash
return sk_node.sign(payload)
def verify_chunk_signature(pk_node: Ed25519PublicKey, chunk_index: int,
nonce: bytes, ct_hash: bytes, signature: bytes) -> None:
"""Verify chunk signature. Raises InvalidSignature on failure."""
payload = chunk_index.to_bytes(4, "big") + nonce + ct_hash
pk_node.verify(signature, payload)
|