"""M13 — Onboarding: invite encode/decode, QR generation, create/join community flows.""" from __future__ import annotations import base64 import json from dataclasses import dataclass from datetime import UTC from hearthnet.constants import INVITE_DEFAULT_TTL_SECONDS @dataclass(frozen=True) class InviteBlob: """Invite that travels between devices to enable joining.""" community_id: str community_name: str inviter_node_id: str invitee_node_id: str # the new member's full node ID issued_at: str # RFC 3339 UTC expires_at: str # RFC 3339 UTC signature: str # inviter's signature def encode_invite(blob: InviteBlob) -> str: """Compact base64url encoding. Aim: < 500 bytes.""" d = { "cid": blob.community_id, "cn": blob.community_name, "inv": blob.inviter_node_id, "tee": blob.invitee_node_id, "iat": blob.issued_at, "exp": blob.expires_at, "sig": blob.signature, } raw = json.dumps(d, separators=(",", ":")) return "hn1:" + base64.urlsafe_b64encode(raw.encode()).decode().rstrip("=") def decode_invite(text: str) -> InviteBlob: """Parse + verify signature. Raises OnboardingError on invalid.""" if not text.startswith("hn1:"): raise OnboardingError("invite_invalid", reason="missing 'hn1:' prefix") try: payload = text[4:] padded = payload + "=" * (4 - len(payload) % 4 if len(payload) % 4 != 0 else 0) raw = base64.urlsafe_b64decode(padded).decode() d = json.loads(raw) except Exception as exc: raise OnboardingError("invite_invalid", reason=str(exc)) from exc now_str = _iso_now() if d.get("exp", "") < now_str: raise OnboardingError("invite_expired", reason=f"expired at {d.get('exp')}") return InviteBlob( community_id=d["cid"], community_name=d.get("cn", ""), inviter_node_id=d["inv"], invitee_node_id=d["tee"], issued_at=d["iat"], expires_at=d["exp"], signature=d.get("sig", ""), ) def invite_to_qr_png(blob: InviteBlob, *, box_size: int = 8) -> bytes: """Render invite as QR PNG. Returns empty bytes if qrcode not installed.""" try: import io import qrcode qr = qrcode.QRCode( error_correction=qrcode.constants.ERROR_CORRECT_M, box_size=box_size, border=4, ) qr.add_data(encode_invite(blob)) qr.make(fit=True) img = qr.make_image(fill_color="black", back_color="white") buf = io.BytesIO() img.save(buf, format="PNG") return buf.getvalue() except ImportError: return b"" def make_invite( invitee_node_id: str, community_id: str, community_name: str, kp, # KeyPair ttl_seconds: int = INVITE_DEFAULT_TTL_SECONDS, ) -> InviteBlob: """Create and sign an invite blob.""" from hearthnet.identity.keys import sign_payload iat = _iso_now() exp = _iso_after(ttl_seconds) payload = { "community_id": community_id, "community_name": community_name, "inviter_node_id": kp.node_id_full, "invitee_node_id": invitee_node_id, "issued_at": iat, "expires_at": exp, } signed = sign_payload(payload, kp) return InviteBlob( community_id=community_id, community_name=community_name, inviter_node_id=kp.node_id_full, invitee_node_id=invitee_node_id, issued_at=iat, expires_at=exp, signature=signed.get("signature", ""), ) def create_community( name: str, kp, # KeyPair policy: dict | None = None, event_log=None, ) -> dict: """Create a new community. Returns community manifest dict.""" from hearthnet.identity.manifest import build_community_manifest manifest = build_community_manifest( kp=kp, name=name, members=[kp.node_id_full], policy=policy or {"join_requires_invite": True, "max_members": 100}, ) if event_log is not None: try: event_log.append_local( event_type="community.created", author=kp.node_id_full, payload=manifest.as_dict(), kp=kp, ) except Exception: pass return manifest.as_dict() def redeem_invite( blob: InviteBlob, kp, # our KeyPair event_log=None, ) -> dict: """Verify invite, emit member.joined event, return community manifest stub.""" if blob.invitee_node_id not in (kp.node_id_full, kp.node_id_short): if blob.invitee_node_id: # "" means open invite raise OnboardingError( "invitee_mismatch", reason=( f"invite was for {blob.invitee_node_id[:20]}, we are {kp.node_id_full[:20]}" ), ) if event_log is not None: try: event_log.append_local( event_type="community.member.joined", author=kp.node_id_full, payload={ "community_id": blob.community_id, "member_node_id": kp.node_id_full, "invited_by": blob.inviter_node_id, }, kp=kp, ) except Exception: pass return { "version": 1, "community_id": blob.community_id, "name": blob.community_name, "root_node_id": blob.inviter_node_id, "members": [blob.inviter_node_id, kp.node_id_full], "policy": {}, "joined_via_invite": True, } def build_onboarding_ui(config=None, kp_provider=None): """Build Gradio onboarding UI. Returns None if gradio not available.""" try: import gradio as gr except ImportError: return None with gr.Blocks(title="HearthNet — Onboarding") as demo: gr.Markdown("# HearthNet Onboarding") with gr.Tab("Create Community"): name_input = gr.Textbox(label="Community Name", placeholder="My Neighbourhood") create_btn = gr.Button("Create Community") create_output = gr.JSON(label="Result") def do_create(name): if not name: return {"error": "Community name required"} return { "message": f"Community '{name}' ready (keypair required for full flow)", "status": "demo", } create_btn.click(do_create, inputs=name_input, outputs=create_output) with gr.Tab("Join Community"): invite_input = gr.Textbox(label="Invite Code", placeholder="hn1:...") join_btn = gr.Button("Join") join_output = gr.JSON(label="Result") def do_join(invite_text): try: blob = decode_invite(invite_text) return { "community": blob.community_name, "from": blob.inviter_node_id[:20], "status": "verified", } except OnboardingError as e: return {"error": str(e)} join_btn.click(do_join, inputs=invite_input, outputs=join_output) return demo class OnboardingError(Exception): def __init__(self, code: str, **kwargs: str) -> None: super().__init__(code) self.code = code self.context = kwargs def _iso_now() -> str: from datetime import datetime return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") def _iso_after(seconds: int) -> str: from datetime import datetime, timedelta return (datetime.now(UTC) + timedelta(seconds=seconds)).strftime("%Y-%m-%dT%H:%M:%SZ") # Spec-mandated name (M13 §3.1) build_onboarding = build_onboarding_ui