"""
FALCONS.AI "Verify Anything" — Hugging Face Space (Gradio transport)
Refactor of the FastAPI Space to the `gradio` SDK. Look and behaviour are
preserved: the product's dark theme tokens and stylesheet, the sample cards,
the drop grid, the Hub-repo form, a cancel control, and the verdict stack
(five newest, each with the engine's receipt). The verification logic is
NOT here — it lives, byte-identical to the FastAPI original, in
`engine_wrapper.py`, and still invokes the pinned, unmodified
`engine/verify_attestation.py` as a subprocess.
Standing rules (BUILD SPEC section 1), how each survives the transport:
1. Engine as subprocess only -> engine_wrapper._run_engine_sync
2. No model-parsing code -> none here; JSON metadata only
3. Nothing retained -> per-request temp dirs deleted in
`finally`; Gradio's own upload copy
is deleted in the same `finally`;
analytics off; Gradio launches
uvicorn at log_level="warning" (no
access log); nothing logged.
4. Missing capability != tampering -> engine_wrapper (unchanged)
5. Weights never re-served -> the verdict stack shows text only;
no file output component exists.
# Verify a Model Surgeon package
Drop in a signed package and this will tell you whether it is intact, what it
contains, and where it came from. **Free, permanently. No account, no sign-up,
nothing stored.**
## What it checks
- **Integrity** — whether a single byte has changed since the package was signed.
- **Lineage** — the chain of operations performed on the model, and by whom.
A model pulled, edited and re-pushed keeps its earlier attestation, so the
chain holds across multiple hands.
- **Scope** — attestations name precisely the files they cover, and this reports
exactly that. No more, no less.
## What it does not claim
Verification proves that **this file is the file that was signed, and records
what was done to it.** It does not establish that a model is original, or that
it is not derived from an open-weight base. That is a different question and
this tool does not pretend to answer it.
## Why it is free
Because a verification tool you have to buy from the party being verified is
not verification. The person who needs convincing should never have to trust
us, hold an account, or pay anything.
The same verifier is open source under Apache-2.0, and **a copy ships inside
every package** — so verification works offline, in an air-gapped environment,
and continues to work whether or not we are still here.
- **Source (CLI + spec):** https://github.com/Falcons-ai/surgeon-verify
- **Where the packages come from:** https://surgeon.falcons.ai/?utm_source=huggingface
## One result worth knowing
During launch testing, this verifier was handed a package signed by a build
**twenty-five releases earlier**. It honoured the old signature correctly. A
single byte was then altered — one byte out of roughly fourteen million — and
the verifier caught the tamper and **named the exact file**.
That is the whole point of the thing.
"""
import base64
import html
import json
import os
import shutil
import tempfile
import threading
import zipfile
from pathlib import Path
import gradio as gr
# ZeroGPU runtime (RealFalconsAI/verifier runs on the free ZeroGPU tier — the
# CPU tier is not selectable for this Space): the runtime kills any app that
# declares no @spaces.GPU function at startup ("No @spaces.GPU function
# detected"). The verifier needs no GPU, so the declared function is a no-op
# that no control ever calls — every verification runs on the CPU path below
# and consumes no GPU quota. Guarded so `python app.py` runs anywhere.
try:
import spaces as _spaces
except Exception: # not on Spaces, or the package is absent
_spaces = None
if _spaces is not None:
@_spaces.GPU(duration=1)
def _zerogpu_placeholder():
"""Declared for the ZeroGPU runtime only. Never invoked."""
return None
from engine_wrapper import (
BUILD_STAMP, LOCAL_CMD, MANIFEST, MAX_UPLOAD_BYTES, MAX_UPLOAD_MB,
MAX_REPO_MB, NORMALIZED_NOTE, NOT_A_PACKAGE_COPY, ROOT, SAMPLES_DIR,
_REPO_ID_RE, _normalize_container, _resolve_and_verify_repo_sync,
_result, _run_engine_sync, _zip_declared_size,
)
MAX_CONCURRENCY = 4
STACK_MAX = 5
print(f"falcons-verify build {BUILD_STAMP}")
# ------------------------------------------------------------- concurrency
# The FastAPI version counted in-flight requests under an asyncio lock and
# answered 429 when full. Gradio runs handlers in worker threads, so the same
# rule is a non-blocking semaphore: full -> the same "at capacity" verdict.
_slots = threading.Semaphore(MAX_CONCURRENCY)
def _busy_result():
return _result(
"error", None, "",
detail=("The verifier is at capacity right now — give it a few seconds and "
"try again. " + LOCAL_CMD),
)
# ---------------------------------------------------------------- verdicts
# pushVerdict() from static/app.js, ported line for line. Same classes, same
# headings, same rules (SCAN F3: a TAMPERED card shows the engine's ✖ line and
# its signed/actual digest pair — never the reassuring signature line).
COPY = {
"not_a_package": NOT_A_PACKAGE_COPY,
"repo_not_attested": (
"This repo doesn't carry a Surgeon attestation, so there's nothing to "
"verify against — that's a statement about provenance, not quality. "
"Repos published through Model Surgeon's attested push verify here."
),
}
def _e(s):
return html.escape(str(s), quote=True)
def _first_line_matching(text, needle_lower):
for ln in (text or "").split("\n"):
if needle_lower in ln.lower():
return ln.strip()
return None
def _first_non_empty(text):
for ln in (text or "").split("\n"):
if ln.strip():
return ln.strip()
return None
def render_card(res, source_label):
v = res.get("verdict")
if v == "verified":
cls, head = "v-verified", "✔ VERIFIED"
elif v == "tampered":
cls, head = "v-tampered", "✘ TAMPERED"
elif v == "not_a_package":
cls, head = "v-amber", "NOT A SURGEON PACKAGE"
elif v == "repo_not_attested":
cls, head = "v-amber", "REPO NOT ATTESTED"
else:
cls, head = "v-error", "LIMIT / ERROR"
parts = [f'{_e(head)}
']
if source_label:
parts.append(f'{_e(source_label)}
')
stdout = res.get("stdout") or ""
if v == "verified":
if res.get("keyid"):
parts.append(
''
+ _e(res["keyid"])
+ 'pin this to know who signed
'
)
fc = res.get("files_checked")
if fc is not None:
parts.append(f'{_e(fc)} files checked
')
att = _first_line_matching(stdout, "attestation")
if att:
parts.append(f'{_e(att)}
')
if v == "tampered":
fail = [ln.strip() for ln in stdout.split("\n")
if ln.lstrip().startswith(("✖", "signed ", "actual "))]
if not fail:
fail = [_first_non_empty(stdout)]
for ln in fail:
if ln:
parts.append(f'{_e(ln)}
')
detail = res.get("detail") or COPY.get(v, "")
if detail:
parts.append(f'{_e(detail)}
')
if stdout.strip():
parts.append(
'show the receipt
'
+ _e(stdout) + "' + "".join(stack) + "
"
def _push(stack, res, source_label):
stack = [render_card(res, source_label)] + list(stack or [])
return stack[:STACK_MAX]
# ---------------------------------------------------------------- handlers
# Each returns (new_stack_state, rendered_html). Every path releases its slot
# and deletes its temp dir in `finally` (SPEC 1.3).
def _samples():
return json.loads(MANIFEST.read_text()) if MANIFEST.exists() else []
def verify_sample(sample_id, stack, progress=gr.Progress()):
if not _slots.acquire(blocking=False):
s = _push(stack, _busy_result(), sample_id)
return s, render_stack(s)
try:
progress(0, desc="recomputing digests…")
entry = next((e for e in _samples() if e.get("id") == sample_id), None)
if entry is None:
res = _result("error", None, "", detail="Unknown sample id.")
else:
sample_path = SAMPLES_DIR / Path(entry["filename"]).name
if not sample_path.exists():
res = _result("error", None, "", detail=(
"This sample package hasn't been installed on the Space yet. "
+ LOCAL_CMD))
else:
# Server-side copy only — the file never round-trips through
# the browser.
res = _run_engine_sync(sample_path)
s = _push(stack, res, entry["label"] if entry else sample_id)
return s, render_stack(s)
finally:
_slots.release()
def verify_upload(file_path, stack, progress=gr.Progress()):
if not file_path:
return stack, render_stack(stack)
label = Path(file_path).name
if not _slots.acquire(blocking=False):
s = _push(stack, _busy_result(), label)
return s, render_stack(s)
tmp = tempfile.mkdtemp(prefix="verify-")
try:
progress(0, desc="recomputing digests…")
# Gradio has already streamed the upload to its cache (bounded by
# launch(max_file_size=…)); it is moved into our per-request dir so
# exactly one deletion covers everything.
target = Path(tmp) / "upload.bin"
shutil.move(file_path, target)
received = target.stat().st_size
if received > MAX_UPLOAD_BYTES:
res = _result("error", None, "", detail=(
"That file is over this verifier's " + str(MAX_UPLOAD_MB)
+ " MB upload cap. " + LOCAL_CMD))
elif received == 0:
res = _result("error", None, "", detail="No file was received. " + LOCAL_CMD)
else:
res = None
# Zip-bomb guard: declared uncompressed size, before any extraction.
if zipfile.is_zipfile(target):
try:
if _zip_declared_size(target) > 2 * MAX_UPLOAD_BYTES:
res = _result("error", None, "", detail=(
"This archive declares an uncompressed size more than "
"twice the " + str(MAX_UPLOAD_MB)
+ " MB cap, so it won't be opened here. " + LOCAL_CMD))
except zipfile.BadZipFile:
pass # let the engine speak for itself
if res is None:
# Any file type is accepted; the engine's own output routes
# non-zips and attestation-less zips to the polite
# not-a-package card. Windows/macOS-made zips get their
# container repaired first.
target, normalized = _normalize_container(target, tmp)
res = _run_engine_sync(target)
if normalized and isinstance(res, dict):
res["detail"] = NORMALIZED_NOTE + (res.get("detail") or "")
s = _push(stack, res, label)
return s, render_stack(s)
finally:
shutil.rmtree(tmp, ignore_errors=True) # SPEC 1.3 — nothing retained
try:
Path(file_path).unlink(missing_ok=True) # Gradio's cached copy, if any remains
except Exception:
pass
_slots.release()
def verify_repo(repo_id, stack, progress=gr.Progress()):
repo_id = (repo_id or "").strip()
if not repo_id:
return stack, render_stack(stack)
if not _slots.acquire(blocking=False):
s = _push(stack, _busy_result(), repo_id)
return s, render_stack(s)
tmp = tempfile.mkdtemp(prefix="verify-repo-")
try:
progress(0, desc="fetching from the Hub…")
if not _REPO_ID_RE.match(repo_id):
res = _result("error", None, "", detail=(
"That doesn't look like a Hugging Face repo id — the format is "
"owner/name. " + LOCAL_CMD))
else:
# Wall-clock: the download itself is bounded inside the resolver's
# hf_hub_download calls; the ✕ cancel button aborts the event.
res = _resolve_and_verify_repo_sync(repo_id, Path(tmp))
s = _push(stack, res, repo_id)
return s, render_stack(s)
finally:
shutil.rmtree(tmp, ignore_errors=True) # SPEC 1.3 — nothing retained
_slots.release()
# ---------------------------------------------------------------- the page
def _data_uri(path, mime):
return f"data:{mime};base64," + base64.b64encode(Path(path).read_bytes()).decode()
LOGO = _data_uri(ROOT / "static" / "logo.png", "image/png")
CSS = (ROOT / "static" / "style.css").read_text(encoding="utf-8") + (ROOT / "static" / "gradio.css").read_text(encoding="utf-8")
MASTHEAD = f"""
"""
FOOTER = """
Verify a Model Surgeon package
Drop in a signed package and this will tell you whether it is intact, what it contains, and where it came from. Free, permanently. No account, no sign-up, nothing stored.
What it checks
- Integrity — whether a single byte has changed since the package was signed.
- Lineage — the chain of operations performed on the model, and by whom. A model pulled, edited and re-pushed keeps its earlier attestation, so the chain holds across multiple hands.
- Scope — attestations name precisely the files they cover, and this reports exactly that. No more, no less.
What it does not claim
Verification proves that this file is the file that was signed, and records what was done to it. It does not establish that a model is original, or that it is not derived from an open-weight base. That is a different question and this tool does not pretend to answer it.
Why it is free
Because a verification tool you have to buy from the party being verified is not verification. The person who needs convincing should never have to trust us, hold an account, or pay anything.
The same verifier is open source under Apache-2.0, and a copy ships inside every package — so verification works offline, in an air-gapped environment, and continues to work whether or not we are still here.
One result worth knowing
During launch testing, this verifier was handed a package signed by a build twenty-five releases earlier. It honoured the old signature correctly. A single byte was then altered — one byte out of roughly fourteen million — and the verifier caught the tamper and named the exact file.
That is the whole point of the thing.
"""
# Force the dark theme regardless of the visitor's OS setting (the product is
# dark only, SPEC 3) — Gradio otherwise follows prefers-color-scheme.
FORCE_DARK_JS = """
() => {
const u = new URL(window.location);
if (u.searchParams.get('__theme') !== 'dark') {
u.searchParams.set('__theme', 'dark');
window.location.replace(u.toString());
}
}
"""
def build():
samples = _samples()
with gr.Blocks(title="Provenance Verifier · FALCONS.AI Model Surgeon",
css=CSS, js=FORCE_DARK_JS, analytics_enabled=False,
delete_cache=(60, 60), # Gradio's cache: sweep every minute, nothing older than a minute
theme=gr.themes.Base(font=gr.themes.GoogleFont("IBM Plex Mono"),
font_mono=gr.themes.GoogleFont("IBM Plex Mono"))) as ui:
stack = gr.State([])
gr.HTML(MASTHEAD)
with gr.Row(elem_classes="workbench"):
with gr.Column(scale=1, min_width=260, elem_classes="samples"):
gr.HTML('SAMPLE PACKAGES
')
sample_buttons = []
for entry in samples:
b = gr.Button(f"▤ {entry['label']}", elem_classes="sample-card")
gr.HTML(f'{_e(entry["blurb"])}
')
sample_buttons.append((b, entry["id"]))
gr.HTML('tap a sample to verify it
')
with gr.Column(scale=2, elem_classes="verify"):
gr.HTML('THE GRID
')
grid = gr.File(label="drop any .zip of your own here — or click to choose a file",
file_count="single", type="filepath",
elem_id="grid", elem_classes="grid")
with gr.Row(elem_classes="alt-inputs"):
repo = gr.Textbox(show_label=False, placeholder="owner/name — e.g. an attested Surgeon push",
elem_classes="repoinput", scale=3, max_lines=1)
repo_btn = gr.Button("VERIFY", elem_classes="repobtn", scale=1)
cancel_btn = gr.Button("✕ cancel", elem_classes="cancel", scale=1)
verdicts = gr.HTML(render_stack([]), elem_classes="verdicts")
gr.HTML(FOOTER)
gr.HTML(f'build {_e(BUILD_STAMP)} · upload cap {MAX_UPLOAD_MB} MB · repo cap {MAX_REPO_MB} MB
')
events = []
for b, sid in sample_buttons:
events.append(b.click(verify_sample, inputs=[gr.State(sid), stack], outputs=[stack, verdicts],
concurrency_limit=MAX_CONCURRENCY))
events.append(grid.upload(verify_upload, inputs=[grid, stack], outputs=[stack, verdicts],
concurrency_limit=MAX_CONCURRENCY))
events.append(repo.submit(verify_repo, inputs=[repo, stack], outputs=[stack, verdicts],
concurrency_limit=MAX_CONCURRENCY))
events.append(repo_btn.click(verify_repo, inputs=[repo, stack], outputs=[stack, verdicts],
concurrency_limit=MAX_CONCURRENCY))
# Same behaviour as the AbortController in the old front end: the
# in-flight verification is dropped; nothing is recorded for it.
cancel_btn.click(None, None, None, cancels=events)
return ui
ui = build()
if __name__ == "__main__":
# `sdk: gradio` on Spaces runs this file and owns the port through
# GRADIO_SERVER_NAME / GRADIO_SERVER_PORT — launch through Gradio, never a
# hand-rolled server (a second uvicorn fought the runtime for the port).
# SPEC 1.3 holds: Gradio starts uvicorn at log_level="warning", so no
# access log is written; this module never logs user input.
ui.queue(default_concurrency_limit=MAX_CONCURRENCY)
ui.launch(
server_name=os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0"),
server_port=int(os.environ.get("GRADIO_SERVER_PORT", os.environ.get("PORT", "7860"))),
max_file_size=f"{MAX_UPLOAD_MB}mb", # the streamed upload cap (SPEC 4.2)
show_api=False,
favicon_path=str(ROOT / "static" / "favicon.png"),
show_error=False,
ssr_mode=False, # experimental Node-side rendering off: nothing here needs it, and a CPU Space shouldn't run a Node sidecar
)