117 lines
4 KiB
Python
117 lines
4 KiB
Python
"""Network Error Bundle + system bounce report.
|
|
|
|
ct_sender: only the sender can read the failure status.
|
|
ct_system: system public key material for logging/recovery.
|
|
Never include the intended recipient's message plaintext.
|
|
Sender identifies the original with lookup_id only.
|
|
"""
|
|
|
|
from __future__ import annotations # annotations
|
|
|
|
import hashlib # lab box
|
|
import json # encode reports
|
|
import os # nonce
|
|
from dataclasses import dataclass # bundle
|
|
from typing import Any, Dict # maps
|
|
|
|
|
|
def _xor(key: bytes, data: bytes) -> bytes:
|
|
"""Lab box: XOR with SHA-256(key||nonce) — replaced by X25519 in production."""
|
|
out = bytearray(len(data)) # buffer
|
|
for i, b in enumerate(data): # each byte
|
|
out[i] = b ^ key[i % len(key)] # xor
|
|
return bytes(out) # done
|
|
|
|
|
|
def _lab_box(pubkey_material: bytes, nonce: bytes, obj: Dict[str, Any]) -> str:
|
|
"""Encrypt obj to a lab 'public' blob (hash of material+nonce as key)."""
|
|
raw = json.dumps(obj, sort_keys=True, separators=(",", ":")).encode() # bytes
|
|
key = hashlib.sha256(pubkey_material + nonce).digest() # 32 bytes
|
|
return _xor(key, raw).hex() # hex ct
|
|
|
|
|
|
def _lab_unbox(pubkey_material: bytes, nonce: bytes, ct_hex: str) -> Dict[str, Any]:
|
|
"""Decrypt lab box (same keying as _lab_box)."""
|
|
key = hashlib.sha256(pubkey_material + nonce).digest() # same kdf
|
|
pt = _xor(key, bytes.fromhex(ct_hex)) # decrypt
|
|
return json.loads(pt.decode()) # json
|
|
|
|
|
|
@dataclass
|
|
class NetworkErrorBundle:
|
|
"""Two ciphertexts: sender-only status + system bounce report."""
|
|
|
|
lookup_id: str # how sender finds the failed send
|
|
error_code: str # machine code (e.g. SM-DEAD-1)
|
|
dest_class: str # mailbox | service (not a mailbox id)
|
|
nonce: str # hex
|
|
ct_sender: str # hex, sender-only
|
|
ct_system: str # hex, system-key bounce report
|
|
|
|
def header(self) -> Dict[str, str]:
|
|
"""Logging header: codes only."""
|
|
return { # never put ct in logs
|
|
"lookup_id": self.lookup_id,
|
|
"error_code": self.error_code,
|
|
"dest_class": self.dest_class,
|
|
}
|
|
|
|
def wire(self) -> Dict[str, Any]:
|
|
"""NATS body for verae.sm.dead / verae.sm.error."""
|
|
return { # both ciphertexts
|
|
"v": 1,
|
|
"type": "network-error-bundle",
|
|
"lookup_id": self.lookup_id,
|
|
"error_code": self.error_code,
|
|
"dest_class": self.dest_class,
|
|
"nonce": self.nonce,
|
|
"ct_sender": self.ct_sender,
|
|
"ct_system": self.ct_system,
|
|
}
|
|
|
|
|
|
def build_bundle(
|
|
*,
|
|
lookup_id: str,
|
|
error_code: str,
|
|
dest_class: str,
|
|
sender_pub: bytes,
|
|
system_pub: bytes,
|
|
detail: str,
|
|
) -> NetworkErrorBundle:
|
|
"""Build bundle. detail is a status phrase, never recipient plaintext."""
|
|
nonce = os.urandom(16) # fresh
|
|
sender_obj = { # only sender opens this
|
|
"lookup_id": lookup_id,
|
|
"error_code": error_code,
|
|
"detail": detail,
|
|
}
|
|
system_obj = { # bounce report — no mail body
|
|
"lookup_id": lookup_id,
|
|
"error_code": error_code,
|
|
"dest_class": dest_class,
|
|
"respond": True,
|
|
}
|
|
return NetworkErrorBundle( # pack
|
|
lookup_id=lookup_id,
|
|
error_code=error_code,
|
|
dest_class=dest_class,
|
|
nonce=nonce.hex(),
|
|
ct_sender=_lab_box(sender_pub, nonce, sender_obj),
|
|
ct_system=_lab_box(system_pub, nonce, system_obj),
|
|
)
|
|
|
|
|
|
def open_sender(bundle: NetworkErrorBundle, sender_pub: bytes) -> Dict[str, Any]:
|
|
"""Sender reads their status ciphertext."""
|
|
return _lab_unbox(sender_pub, bytes.fromhex(bundle.nonce), bundle.ct_sender) # sender only
|
|
|
|
|
|
def open_system(bundle: NetworkErrorBundle, system_pub: bytes) -> Dict[str, Any]:
|
|
"""Ops logger reads bounce metadata (still no mail body)."""
|
|
return _lab_unbox(system_pub, bytes.fromhex(bundle.nonce), bundle.ct_system) # system only
|
|
|
|
|
|
def log_summary(bundle: NetworkErrorBundle) -> Dict[str, str]:
|
|
"""Central log line: codes only."""
|
|
return bundle.header() # no ciphertext, no body
|