Spaces:
Running
Running
| """Light Rider Quantum Entropy: Hugging Face Space demo. | |
| Draws random bytes from a named quantum source with the lightrider SDK and | |
| shows the signed receipt, verified with Ed25519. | |
| Secret: | |
| LR_API_KEY Light Rider API key that pays for the demo's draws | |
| Optional variables: | |
| DEMO_PER_VISITOR_DAILY draws per visitor per UTC day (default 5) | |
| DEMO_GLOBAL_DAILY draws for the whole Space per UTC day (default 200) | |
| If a source can't serve, the visitor sees why. No other source is used in its | |
| place, and this app never generates randomness itself. | |
| """ | |
| from __future__ import annotations | |
| import datetime as dt | |
| import html | |
| import json | |
| import os | |
| import threading | |
| import gradio as gr | |
| # The Light Rider SDK's entropy client (pip install lightrider>=1.6.0). | |
| from lightrider.entropy import api as lre | |
| MIN_BYTES, MAX_BYTES = 16, 512 | |
| PER_VISITOR_DAILY = int(os.environ.get("DEMO_PER_VISITOR_DAILY", "5")) | |
| GLOBAL_DAILY = int(os.environ.get("DEMO_GLOBAL_DAILY", "200")) | |
| PUBLIC_SITE = "https://entropy.lightriderinc.com" | |
| BUTTON_RED = "#dc2626" # brand red, deep enough for white text (4.8:1) | |
| # Reads LR_API_KEY from the environment. | |
| client = lre.EntropyAPI(max_retries=1, timeout=60) | |
| # --------------------------------------------------------------------- demo wallet limits | |
| class DailyLimits: | |
| """In-memory per-visitor and global draw caps, reset at 00:00 UTC. | |
| A slot is reserved before the draw and given back if the draw fails (a | |
| failed draw is never charged), so concurrent visitors can't overshoot the | |
| caps. Counters live in memory: a Space restart resets them, which is fine | |
| for a demo. Visitor ids (IP-derived) are kept in memory only, never logged. | |
| """ | |
| def __init__(self, per_visitor: int, global_cap: int): | |
| self.per_visitor = per_visitor | |
| self.global_cap = global_cap | |
| self._lock = threading.Lock() | |
| self._day = "" | |
| self._visitors: dict[str, int] = {} | |
| self._total = 0 | |
| def _roll(self) -> None: | |
| today = dt.datetime.now(dt.timezone.utc).date().isoformat() | |
| if today != self._day: | |
| self._day, self._visitors, self._total = today, {}, 0 | |
| def reserve(self, visitor: str) -> str | None: | |
| """Takes one draw slot; returns a message instead when a cap is reached.""" | |
| with self._lock: | |
| self._roll() | |
| if self._total >= self.global_cap: | |
| return ( | |
| "The demo has reached its daily limit for everyone. Please come back tomorrow (00:00 UTC), " | |
| "or use your own key from your code (see **Use in your code**)." | |
| ) | |
| if self._visitors.get(visitor, 0) >= self.per_visitor: | |
| return ( | |
| f"You've used your {self.per_visitor} demo draws for today. Come back tomorrow (00:00 UTC), " | |
| "or draw as much as you like with your own key (see **Use in your code**)." | |
| ) | |
| self._visitors[visitor] = self._visitors.get(visitor, 0) + 1 | |
| self._total += 1 | |
| return None | |
| def release(self, visitor: str) -> None: | |
| with self._lock: | |
| if self._visitors.get(visitor, 0) > 0: | |
| self._visitors[visitor] -= 1 | |
| self._total -= 1 | |
| def remaining(self, visitor: str) -> int: | |
| with self._lock: | |
| self._roll() | |
| return max(0, min(self.per_visitor - self._visitors.get(visitor, 0), self.global_cap - self._total)) | |
| limits = DailyLimits(PER_VISITOR_DAILY, GLOBAL_DAILY) | |
| def visitor_id(request: gr.Request | None) -> str: | |
| """The visitor's address as seen by the Space (first X-Forwarded-For hop).""" | |
| if request is None: | |
| return "local" | |
| forwarded = request.headers.get("x-forwarded-for", "") if request.headers else "" | |
| if forwarded: | |
| return forwarded.split(",")[0].strip() | |
| return getattr(request.client, "host", None) or request.session_hash or "unknown" | |
| # --------------------------------------------------------------------- sources | |
| def _label(s: lre.Source) -> str: | |
| state = f"{s.bytes_available:,} B on hand" if s.ready else "out of entropy, refill pending" | |
| name = f"{s.name} ({s.alias})" if s.alias else s.name | |
| return f"{name} · {s.type} · {state}" | |
| def load_sources(): | |
| """Fill the pickers from /api/v1/sources (Available sources only).""" | |
| try: | |
| sources = client.sources() | |
| except lre.EntropyAPIError as err: | |
| msg = f"⚠️ Couldn't load sources right now ({_short(err)}). Try **Refresh sources** in a moment." | |
| return gr.update(choices=[], value=None), gr.update(choices=[], value=[]), msg | |
| single = [(_label(s), s.id) for s in sources] | |
| first_ready = next((s.id for s in sources if s.ready and s.max_single_draw_bytes >= MIN_BYTES), None) | |
| blendable = [(_label(s), s.id) for s in sources if s.custom_pool_selectable] | |
| ready_n = sum(1 for s in sources if s.ready) | |
| status = f"{ready_n} of {len(sources)} sources ready right now." if sources else "No sources are available right now." | |
| return ( | |
| gr.update(choices=single, value=first_ready), | |
| gr.update(choices=blendable, value=[]), | |
| status, | |
| ) | |
| # --------------------------------------------------------------------- generate | |
| def _short(err: Exception) -> str: | |
| return str(err).strip().rstrip(".") or type(err).__name__ | |
| def _friendly(err: lre.EntropyAPIError, what: str) -> str: | |
| if isinstance(err, lre.SourceEmptyError): | |
| return ( | |
| f"⏳ **{what} is out of entropy right now.** It refills on its own; try another source or try again later. " | |
| "Nothing was charged and nothing was substituted." | |
| ) | |
| if isinstance(err, lre.SourceUnavailableError): | |
| return f"⏸️ **{what} isn't serving right now.** Try another source. Nothing was charged and nothing was substituted." | |
| if isinstance(err, lre.RateLimitError): | |
| return f"🐢 **Too many requests just now.** Please try again in about {int(err.retry_after) or 1} seconds." | |
| if isinstance(err, lre.InsufficientCreditsError): | |
| return "💳 **The demo's credits have run out for now.** Please try again later, or use your own key." | |
| if isinstance(err, lre.AuthenticationError): | |
| return "🔑 **The demo isn't configured correctly** (its API key was refused). Please try again later." | |
| if isinstance(err, lre.InvalidRequestError): | |
| return f"⚠️ {_short(err)}." | |
| return "⚠️ **The entropy service is temporarily unavailable.** Nothing was charged. Please try again shortly." | |
| def _badge(ok: bool) -> str: | |
| if ok: | |
| return '<div class="lr-badge lr-ok">✔ Verified: signed by Light Rider and unchanged.</div>' | |
| return '<div class="lr-badge lr-bad">✘ Not verified: the signature didn\'t match Light Rider\'s public key.</div>' | |
| # --------------------------------------------------------------------- receipt view | |
| MODE_LABELS = {"single_source": "Single source", "custom_pool": "Custom pool", "multi_source": "Custom pool"} | |
| def _local_time(ns, offset_min, fmt: str = "%Y-%m-%d %H:%M:%S") -> str: | |
| """A receipt time in the visitor's local time. `offset_min` is the | |
| browser's getTimezoneOffset() (minutes behind UTC); UTC if unknown.""" | |
| try: | |
| t = dt.datetime.fromtimestamp(int(ns) / 1e9, dt.timezone.utc) | |
| except (TypeError, ValueError, OverflowError, OSError): | |
| return "unknown" | |
| try: | |
| off = int(offset_min) | |
| except (TypeError, ValueError): | |
| off = 0 | |
| local = t.astimezone(dt.timezone(dt.timedelta(minutes=-off))) | |
| if off == 0: | |
| zone = "UTC" | |
| else: | |
| h, m = divmod(abs(off), 60) | |
| zone = f"UTC{'+' if off < 0 else '-'}{h}" + (f":{m:02d}" if m else "") | |
| return f"{local.strftime(fmt)} ({zone})" | |
| def receipt_view(receipt: dict, verified: bool, offset_min) -> str: | |
| """Status line and a readable summary of a receipt (the signed JSON is shown below it).""" | |
| e = html.escape | |
| def mono(v) -> str: | |
| return f"<code>{e(str(v))}</code>" | |
| policy = str(receipt.get("policy", "")) | |
| sig = f"{'✔ Verified' if verified else '✘ Not verified'} ({e(str(receipt.get('signature_alg', '')))})" | |
| fields = [ | |
| ("Mode", e(MODE_LABELS.get(policy, policy.replace("_", " ").capitalize()))), | |
| ("Pool", mono(receipt.get("pool_id", ""))), | |
| ("Contributing sources", ", ".join(mono(s) for s in receipt.get("contributing_sources", []))), | |
| ("Bytes", e(str(receipt.get("output_bytes", "")))), | |
| ("Input min-entropy", e(f"{receipt.get('input_min_entropy_bits', '')} bits")), | |
| ("Quality score", e(str(receipt.get("quality_score", "")))), | |
| ("Extractor", e(str(receipt.get("extractor_alg", "")))), | |
| ("DRBG", e(str(receipt.get("drbg_alg", "")))), | |
| ("Issued at", e(_local_time(receipt.get("timestamp_unix_ns"), offset_min))), | |
| ("Request ID", mono(receipt.get("request_id", ""))), | |
| ("Signature", sig), | |
| ] | |
| grid = "".join(f'<div><span class="lr-label">{k}</span><span class="lr-value">{v}</span></div>' for k, v in fields) | |
| return f'{_badge(verified)}<div class="lr-card"><div class="lr-grid">{grid}</div></div>' | |
| # --------------------------------------------------------------------- your draws (this browser session) | |
| # | |
| # Kept in Gradio session state only, and only what the receipt already says: | |
| # never the bytes. It ends with the session. | |
| HISTORY_HEADERS = ["Time", "Source", "Bytes", "Request ID"] | |
| def history_rows(history: list[dict], offset_min) -> list[list]: | |
| return [ | |
| [_local_time(d["receipt"].get("timestamp_unix_ns"), offset_min, "%H:%M:%S"), | |
| ", ".join(d["receipt"].get("contributing_sources", [])), | |
| d["receipt"].get("output_bytes"), | |
| d["receipt"].get("request_id")] | |
| for d in history | |
| ] | |
| def show_draw(history: list[dict], offset_min, evt: gr.SelectData): | |
| """Clicking a row in "Your draws" shows that draw's receipt.""" | |
| row = evt.index[0] if isinstance(evt.index, (list, tuple)) else evt.index | |
| if not history or row is None or not (0 <= row < len(history)): | |
| return gr.update(), gr.update(), gr.update(), gr.update() | |
| d = history[row] | |
| note = "Showing an earlier draw. Its bytes were shown once and aren't kept; this is its receipt." | |
| return "", receipt_view(d["receipt"], d["verified"], offset_min), d["receipt_text"], note | |
| def generate(mode: str, source_id: str | None, custom_ids: list[str] | None, n_bytes: float, | |
| history: list[dict], offset_min, request: gr.Request | None = None): | |
| """One draw. Returns (hex, receipt view, receipt JSON, message, usage note, history, history rows).""" | |
| visitor = visitor_id(request) | |
| n = int(n_bytes) | |
| history = list(history or []) | |
| def fail(message: str): | |
| return "", gr.update(), gr.update(), message, _usage(visitor), history, gr.update() | |
| if mode == "Custom pool": | |
| ids = list(custom_ids or []) | |
| if not (2 <= len(ids) <= 8): | |
| return fail("Pick **2 to 8** sources for a custom pool.") | |
| what, kwargs = "One of the picked sources", {"sources": ids} | |
| else: | |
| if not source_id: | |
| return fail("Pick a source first (use **Refresh sources** if the list is empty).") | |
| what, kwargs = source_id, {"source": source_id} | |
| if not (MIN_BYTES <= n <= MAX_BYTES): | |
| return fail(f"Choose {MIN_BYTES} to {MAX_BYTES} bytes.") | |
| refusal = limits.reserve(visitor) | |
| if refusal: | |
| return fail(refusal) | |
| try: | |
| result = client.get(n, **kwargs) | |
| except lre.EntropyAPIError as err: | |
| limits.release(visitor) # a failed draw is never charged | |
| return fail(_friendly(err, what)) | |
| try: | |
| verified = client.verify(result.receipt) | |
| except lre.EntropyAPIError: | |
| verified = False | |
| receipt = result.receipt | |
| # Exact text, built in Python: a browser JSON viewer would round the | |
| # 64-bit timestamp, and a rounded receipt no longer verifies. | |
| receipt_text = json.dumps(receipt, indent=2, ensure_ascii=False) | |
| history.insert(0, {"receipt": receipt, "receipt_text": receipt_text, "verified": verified}) | |
| message = f"**{len(result.data)} bytes** from **{', '.join(result.sources)}**." | |
| return (result.hex, receipt_view(receipt, verified, offset_min), receipt_text, message, _usage(visitor), | |
| history, history_rows(history, offset_min)) | |
| def _usage(visitor: str) -> str: | |
| left = limits.remaining(visitor) | |
| return f"Demo draws left for you today: **{left}** of {PER_VISITOR_DAILY}." | |
| def toggle_mode(mode: str): | |
| single = mode != "Custom pool" | |
| return gr.update(visible=single), gr.update(visible=not single) | |
| # --------------------------------------------------------------------- UI | |
| INTRO = f""" | |
| # Light Rider Quantum Entropy | |
| Draw random bytes from a named quantum source: a quantum random number generator or a quantum processor. | |
| Every draw comes with a receipt, signed by Light Rider, that records which source produced the bytes and when. | |
| The app checks the signature for you. | |
| It's an alternative entropy source you can verify, not a claim that quantum randomness beats a good classical | |
| generator. If a source can't serve, you'll see why; no other source is used in its place. | |
| <small>Up to {PER_VISITOR_DAILY} draws per visitor per day. Your bytes are shown once and never stored.</small> | |
| """ | |
| # Same commands as https://entropy.lightriderinc.com/developers (tested per platform). | |
| CODE_TABS = { | |
| 'macOS / Linux': '**For this terminal**\n\n```bash\nexport LR_API_KEY="lr_live_YOUR_KEY"\n```\n\n**32 bytes from one source**\n\n```bash\ncurl -s https://entropy.lightriderinc.com/api/v1/entropy \\\n -H "Authorization: Bearer $LR_API_KEY" \\\n -H "Content-Type: application/json" \\\n -H "Idempotency-Key: $(uuidgen 2>/dev/null || cat /proc/sys/kernel/random/uuid)" \\\n -d \'{"bytes": 32, "mode": "source", "id": "anu"}\'\n```\n', | |
| 'Windows PowerShell': '**For this window**\n\n```powershell\n$env:LR_API_KEY = "lr_live_YOUR_KEY"\n```\n\n**32 bytes from one source**\n\n```powershell\n$headers = @{ Authorization = "Bearer $env:LR_API_KEY"; "Idempotency-Key" = [guid]::NewGuid().ToString() }\n$body = @{ bytes = 32; mode = "source"; id = "anu" } | ConvertTo-Json\n$r = Invoke-RestMethod -Method Post -Uri https://entropy.lightriderinc.com/api/v1/entropy -Headers $headers -ContentType "application/json" -Body $body\n$r.bytes_hex\n$r.receipt.contributing_sources\n```\n', | |
| 'Python': '**Install (in a terminal)**\n\n```bash\npip install -U lightrider\n```\n\n**entropy_demo.py**\n\n```python\nimport os\nfrom getpass import getpass\n\nif not os.environ.get("LR_API_KEY"):\n os.environ["LR_API_KEY"] = getpass("Light Rider API key: ")\n\nimport lightrider as lr\n\nfor s in lr.entropy.sources():\n print(s.alias or s.id, s.state, s.bytes_available)\n\nr = lr.entropy.get(32, source="anu")\nprint(r.hex)\nprint(r.sources, r.request_id)\n\npool = lr.entropy.get(64, sources=["anu", "cisco"])\nprint(pool.sources)\n\nassert lr.entropy.verify(r.receipt)\nassert lr.entropy.verify_commitment(r)\nprint("Receipt verified")\n```\n\n**Run it (in a terminal)**\n\n```bash\npython entropy_demo.py\n```\n', | |
| 'Google Colab': '**Cell 1: install**\n\n```python\n!pip install -q -U lightrider\n```\n\n**Cell 2: your key, from Colab Secrets (key icon, name LR_API_KEY, notebook access on)**\n\n```python\nimport os\nfrom google.colab import userdata\n\nos.environ["LR_API_KEY"] = userdata.get("LR_API_KEY")\n```\n\n**Cell 3: draw and verify**\n\n```python\nimport lightrider as lr\n\nfor s in lr.entropy.sources():\n print(s.alias or s.id, s.state, s.bytes_available)\n\nr = lr.entropy.get(32, source="anu")\nprint(r.hex)\nprint(r.sources, r.request_id)\n\npool = lr.entropy.get(64, sources=["anu", "cisco"])\nprint(pool.sources)\n\nassert lr.entropy.verify(r.receipt)\nassert lr.entropy.verify_commitment(r)\nprint("Receipt verified")\n```\n', | |
| } | |
| CODE_FOOTER = 'Create your API key at [entropy.lightriderinc.com/developers](https://entropy.lightriderinc.com/developers). It gives full access to your account, so keep it out of public code. Sources: `anu`, `cisco`, `qispace`, `rigetti`, `iqm`, `iqm-qec`. Price: 1 credit ($0.01) per 256 bytes.' | |
| # Theme variables and translucent tints only, so every custom block reads in | |
| # both the light and the dark theme. | |
| CSS = f""" | |
| .lr-badge {{ padding: 10px 14px; border-radius: 8px; font-weight: 600; border: 1px solid; | |
| color: var(--body-text-color); margin-bottom: 10px; }} | |
| .lr-ok {{ background: rgba(34, 197, 94, 0.14); border-color: rgba(34, 197, 94, 0.6); }} | |
| .lr-bad {{ background: rgba(239, 68, 68, 0.14); border-color: rgba(239, 68, 68, 0.6); }} | |
| .lr-card {{ border: 1px solid var(--border-color-primary); border-radius: 8px; padding: 12px 14px; | |
| background: var(--background-fill-secondary); }} | |
| .lr-grid {{ display: grid; grid-template-columns: repeat(auto-fill, minmax(190px, 1fr)); gap: 10px 18px; }} | |
| .lr-label {{ display: block; font-size: 12px; color: var(--body-text-color); opacity: 0.75; }} | |
| .lr-value {{ color: var(--body-text-color); overflow-wrap: anywhere; }} | |
| .lr-value code {{ font-family: var(--font-mono); background: none; padding: 0; color: inherit; }} | |
| #lr-generate {{ background: {BUTTON_RED}; color: white; }} | |
| """ | |
| def build() -> gr.Blocks: | |
| with gr.Blocks(title="Light Rider Quantum Entropy") as demo: | |
| gr.Markdown(INTRO) | |
| history = gr.State([]) | |
| # Browser-side value (a State lives on the server, out of reach of the page's JS). | |
| tz_offset = gr.Number(value=0, visible=False) | |
| with gr.Tab("Generate"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| mode = gr.Radio(["Single source", "Custom pool"], value="Single source", label="Mode") | |
| source = gr.Dropdown(label="Source", choices=[], interactive=True) | |
| custom = gr.CheckboxGroup(label="Sources to blend (2 to 8)", choices=[], visible=False) | |
| n_bytes = gr.Slider(MIN_BYTES, MAX_BYTES, value=32, step=16, label="Bytes") | |
| with gr.Row(): | |
| refresh = gr.Button("Refresh sources", variant="secondary") | |
| go = gr.Button("Generate", variant="primary", elem_id="lr-generate") | |
| status = gr.Markdown() | |
| usage = gr.Markdown(f"Demo draws left for you today: **{PER_VISITOR_DAILY}** of {PER_VISITOR_DAILY}.") | |
| with gr.Column(scale=2): | |
| message = gr.Markdown() | |
| out_hex = gr.Textbox(label="Entropy (hex)", lines=4, buttons=["copy"], interactive=False) | |
| view = gr.HTML() | |
| receipt = gr.Code(language="json", interactive=False, label="Signed receipt (JSON)", | |
| buttons=["copy", "download"]) | |
| gr.Markdown("### Your draws\nThis browser session only. Click a draw to see its receipt.") | |
| history_table = gr.Dataframe(headers=HISTORY_HEADERS, value=[], interactive=False, show_label=False, | |
| wrap=True) | |
| with gr.Tab("Use in your code"): | |
| with gr.Tabs(): | |
| for label, text in CODE_TABS.items(): | |
| with gr.Tab(label): | |
| gr.Markdown(text) | |
| gr.Markdown(CODE_FOOTER) | |
| mode.change(toggle_mode, mode, [source, custom], api_visibility="private") | |
| refresh.click(load_sources, None, [source, custom, status], api_visibility="private") | |
| demo.load(load_sources, None, [source, custom, status], api_visibility="private") | |
| # The visitor's UTC offset, so receipt times show in their local time. | |
| demo.load(None, None, tz_offset, js="() => new Date().getTimezoneOffset()", api_visibility="private") | |
| go.click( | |
| generate, | |
| [mode, source, custom, n_bytes, history, tz_offset], | |
| [out_hex, view, receipt, message, usage, history, history_table], | |
| api_visibility="private", | |
| concurrency_limit=4, | |
| ) | |
| history_table.select(show_draw, [history, tz_offset], [out_hex, view, receipt, message], | |
| api_visibility="private") | |
| return demo | |
| demo = build() | |
| if __name__ == "__main__": | |
| demo.launch(css=CSS, theme=gr.themes.Soft(primary_hue="red"), footer_links=[]) | |