Spaces:
Running on Zero
Running on Zero
File size: 5,743 Bytes
4cd8837 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """M22 — Mobile invite helpers.
Generates mobile-targeted invite deep links (``hnapp://``) and QR codes
for the mobile native client (Flutter). Builds on top of the Phase 1
onboarding module (M13).
"""
from __future__ import annotations
import base64
import hashlib
import json
import time
from dataclasses import dataclass, field
from typing import Any
# ---------------------------------------------------------------------------
# Invite blob for mobile clients
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class MobileInviteBlob:
"""Compact invite that the mobile app can parse from a QR or deep link.
Wire format: ``hnapp://v1/<b64url(json_payload)>``
The payload is JSON with fields defined below.
"""
community_id: str
community_name: str
anchor_endpoints: list[str]
"""HTTP(S) or WebSocket URLs of community anchors the app can reach."""
invited_by: str
"""node_id of the inviting user (display hint only)."""
relay_url: str | None = None
"""Optional relay tier URL for NAT-traversal push delivery (M15)."""
invite_token: str | None = None
"""One-time capability token the app exchanges on first contact (M16)."""
created_at: float = field(default_factory=time.time)
expires_at: float | None = None
def is_expired(self, now: float | None = None) -> bool:
t = now if now is not None else time.time()
return self.expires_at is not None and t > self.expires_at
# ------------------------------------------------------------------
# Serialization
# ------------------------------------------------------------------
def to_dict(self) -> dict:
return {
"v": 1,
"community_id": self.community_id,
"community_name": self.community_name,
"anchor_endpoints": self.anchor_endpoints,
"invited_by": self.invited_by,
"relay_url": self.relay_url,
"invite_token": self.invite_token,
"created_at": self.created_at,
"expires_at": self.expires_at,
}
def to_deep_link(self) -> str:
"""Encode as ``hnapp://v1/<b64url>``."""
payload = json.dumps(self.to_dict(), separators=(",", ":"), sort_keys=True)
b64 = base64.urlsafe_b64encode(payload.encode()).rstrip(b"=").decode()
return f"hnapp://v1/{b64}"
def fingerprint(self) -> str:
"""SHA-256 (hex, 16 chars) of the JSON payload for logging."""
raw = json.dumps(self.to_dict(), separators=(",", ":"), sort_keys=True)
return hashlib.sha256(raw.encode()).hexdigest()[:16]
@classmethod
def from_deep_link(cls, deep_link: str) -> "MobileInviteBlob":
"""Parse a deep link produced by :meth:`to_deep_link`."""
if not deep_link.startswith("hnapp://v1/"):
raise ValueError(f"Not a valid hnapp:// deep link: {deep_link!r}")
b64 = deep_link[len("hnapp://v1/"):]
# Re-add padding
padding = 4 - len(b64) % 4
if padding != 4:
b64 += "=" * padding
payload = json.loads(base64.urlsafe_b64decode(b64).decode())
return cls(
community_id=payload["community_id"],
community_name=payload["community_name"],
anchor_endpoints=payload["anchor_endpoints"],
invited_by=payload["invited_by"],
relay_url=payload.get("relay_url"),
invite_token=payload.get("invite_token"),
created_at=payload.get("created_at", time.time()),
expires_at=payload.get("expires_at"),
)
# ---------------------------------------------------------------------------
# QR code rendering (qrcode optional)
# ---------------------------------------------------------------------------
def render_qr_svg(blob: MobileInviteBlob) -> str | None:
"""Return an SVG string for the invite QR code, or None if ``qrcode`` is
not installed. The SVG can be embedded directly in HTML."""
try:
import qrcode # type: ignore
import qrcode.image.svg # type: ignore
factory = qrcode.image.svg.SvgPathImage
qr = qrcode.make(blob.to_deep_link(), image_factory=factory)
import io
buf = io.BytesIO()
qr.save(buf)
return buf.getvalue().decode("utf-8")
except ImportError:
return None
def render_qr_terminal(blob: MobileInviteBlob) -> str:
"""Return the QR code as ASCII art (uses ``qrcode`` if available, else
falls back to the raw deep link)."""
try:
import qrcode # type: ignore
qr = qrcode.QRCode()
qr.add_data(blob.to_deep_link())
qr.make(fit=True)
import io
buf = io.StringIO()
qr.print_ascii(out=buf)
return buf.getvalue()
except ImportError:
return blob.to_deep_link()
# ---------------------------------------------------------------------------
# Factory helper
# ---------------------------------------------------------------------------
def build_mobile_invite(
community_id: str,
community_name: str,
anchor_endpoints: list[str],
invited_by: str,
relay_url: str | None = None,
invite_token: str | None = None,
ttl_seconds: float = 86_400 * 7, # 7 days default
) -> MobileInviteBlob:
"""Create a :class:`MobileInviteBlob` with a default 7-day TTL."""
now = time.time()
return MobileInviteBlob(
community_id=community_id,
community_name=community_name,
anchor_endpoints=anchor_endpoints,
invited_by=invited_by,
relay_url=relay_url,
invite_token=invite_token,
created_at=now,
expires_at=now + ttl_seconds,
)
|