48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
"""Shared helpers for the demo scripts: logging + the key store file.
|
|
|
|
Credential story (a real deployment would do this properly):
|
|
- demo/psk holds the out-of-band registration secret, created by
|
|
whichever script runs first (both scripts must see the same file).
|
|
- demo/<role>.keys holds the key store (JSON: peer id -> hex
|
|
credential). Initially it maps "peer" to the registration secret;
|
|
the handshake protocol itself replaces entries as credentials
|
|
rotate, get flagged verified, or are burned. The scripts persist
|
|
the store after every connection attempt, success or failure.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).parent
|
|
|
|
|
|
def setup_logging(role: str) -> logging.Logger:
|
|
logging.basicConfig(
|
|
level=logging.DEBUG,
|
|
format=f"%(asctime)s {role} %(levelname)-5s %(name)s: %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
)
|
|
return logging.getLogger(role)
|
|
|
|
|
|
def psk() -> bytes:
|
|
f = HERE / "psk"
|
|
if not f.exists():
|
|
from quantumpipe import generate_registration_secret
|
|
|
|
f.write_bytes(generate_registration_secret())
|
|
return f.read_bytes()
|
|
|
|
|
|
def key_store(role: str) -> dict:
|
|
f = HERE / f"{role}.keys"
|
|
if f.exists():
|
|
return {k: bytes.fromhex(v) for k, v in json.loads(f.read_text()).items()}
|
|
return {"peer": psk()}
|
|
|
|
|
|
def save_store(role: str, store: dict) -> None:
|
|
f = HERE / f"{role}.keys"
|
|
f.write_text(json.dumps({k: v.hex() for k, v in store.items()}, indent=1) + "\n")
|