48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
from base58 import b58encode
|
|
|
|
from app.core._secrets import hot_seed, hot_pubkey
|
|
from app.core.crypto.x25519 import ed25519_to_x25519
|
|
from app.core.logger import make_log
|
|
from app.core.network.nodesig import sign_headers
|
|
|
|
|
|
async def request_key_from_peer(base_url: str, encrypted_cid: str) -> Optional[bytes]:
|
|
"""
|
|
Request a sealed key from peer and decrypt it using our X25519 private key.
|
|
Returns plaintext DEK bytes or None on failure.
|
|
"""
|
|
try:
|
|
sk_x, pk_x = ed25519_to_x25519(hot_seed)
|
|
node_id = b58encode(hot_pubkey).decode()
|
|
body = {
|
|
"encrypted_cid": encrypted_cid,
|
|
"requestor_node_id": node_id,
|
|
"recipient_box_pub": base64.b64encode(bytes(pk_x)).decode(),
|
|
}
|
|
path = "/api/v1/keys.request"
|
|
headers = sign_headers("POST", path, json.dumps(body).encode(), hot_seed, b58encode(hot_pubkey).decode())
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
r = await client.post(f"{base_url.rstrip('/')}{path}", json=body, headers=headers)
|
|
if r.status_code != 200:
|
|
make_log('key_client', f"{base_url} returned {r.status_code}: {r.text}", level='warning')
|
|
return None
|
|
j = r.json()
|
|
sealed_b64 = j.get('sealed_key_b64')
|
|
if not sealed_b64:
|
|
return None
|
|
sealed = base64.b64decode(sealed_b64)
|
|
from nacl.public import SealedBox
|
|
sb = SealedBox(sk_x)
|
|
dek = sb.decrypt(sealed)
|
|
return dek
|
|
except Exception as e:
|
|
make_log('key_client', f"request/decrypt failed: {e}", level='error')
|
|
return None
|