app.py

"""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=[])