89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
"""Ende-zu-Ende-Verschluesselung. Muss mit client-android/.../Crypto.kt uebereinstimmen.
|
|
|
|
Ableitung: master = PBKDF2-HMAC-SHA256(token, "netclip-v1", 200000, 32)
|
|
room_id = hex(HMAC-SHA256(master, "room-id")) -> sieht der Server
|
|
key = HMAC-SHA256(master, "enc-key") -> bleibt beim Client
|
|
Meta: base64( nonce[12] || AES-256-GCM(key, nonce, json, aad="meta") )
|
|
Blob: "NCB1" || prefix[8] || Chunks
|
|
Chunk = flag[1] || len[4, BE] || AES-256-GCM(key, prefix||counter[4, BE], data, aad=flag)
|
|
flag = 1 beim letzten Chunk (verhindert Abschneiden), sonst 0
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import struct
|
|
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
SALT = b"netclip-v1"
|
|
ITERATIONS = 200_000
|
|
MAGIC = b"NCB1"
|
|
CHUNK = 1024 * 1024
|
|
|
|
|
|
class CryptoError(Exception):
|
|
pass
|
|
|
|
|
|
class Keys:
|
|
def __init__(self, token: str):
|
|
master = hashlib.pbkdf2_hmac("sha256", token.encode("utf-8"), SALT, ITERATIONS, 32)
|
|
self.room_id = hmac.new(master, b"room-id", hashlib.sha256).hexdigest()
|
|
self.key = hmac.new(master, b"enc-key", hashlib.sha256).digest()
|
|
self.aead = AESGCM(self.key)
|
|
|
|
def encrypt_meta(self, meta: dict) -> str:
|
|
nonce = os.urandom(12)
|
|
ct = self.aead.encrypt(nonce, json.dumps(meta).encode("utf-8"), b"meta")
|
|
return base64.b64encode(nonce + ct).decode("ascii")
|
|
|
|
def decrypt_meta(self, data: str) -> dict:
|
|
try:
|
|
raw = base64.b64decode(data)
|
|
return json.loads(self.aead.decrypt(raw[:12], raw[12:], b"meta"))
|
|
except Exception as e: # falscher Token oder manipuliert
|
|
raise CryptoError(str(e)) from e
|
|
|
|
def encrypt_label(self, text: str) -> str:
|
|
return self.encrypt_meta({"name": text})
|
|
|
|
def encrypt_stream(self, src, dst):
|
|
"""Liest Klartext aus src (Datei-Objekt), schreibt Blob nach dst."""
|
|
prefix = os.urandom(8)
|
|
dst.write(MAGIC + prefix)
|
|
counter = 0
|
|
buf = src.read(CHUNK)
|
|
while True:
|
|
nxt = src.read(CHUNK) if buf else b""
|
|
last = not nxt
|
|
flag = b"\x01" if last else b"\x00"
|
|
ct = self.aead.encrypt(prefix + struct.pack(">I", counter), buf, flag)
|
|
dst.write(flag + struct.pack(">I", len(ct)) + ct)
|
|
if last:
|
|
return
|
|
counter += 1
|
|
buf = nxt
|
|
|
|
def decrypt_stream(self, src, dst):
|
|
if src.read(4) != MAGIC:
|
|
raise CryptoError("kein NetClip-Blob")
|
|
prefix = src.read(8)
|
|
counter = 0
|
|
while True:
|
|
head = src.read(5)
|
|
if len(head) != 5:
|
|
raise CryptoError("Blob abgeschnitten")
|
|
flag, length = head[:1], struct.unpack(">I", head[1:])[0]
|
|
if length > CHUNK + 16:
|
|
raise CryptoError("Chunk zu gross")
|
|
ct = src.read(length)
|
|
try:
|
|
dst.write(self.aead.decrypt(prefix + struct.pack(">I", counter), ct, flag))
|
|
except Exception as e:
|
|
raise CryptoError("Entschluesselung fehlgeschlagen") from e
|
|
if flag == b"\x01":
|
|
return
|
|
counter += 1
|