Fileverity API docs
Sample code
Complete Node.js and Python programs for the Fileverity API: check an image, check a large PDF, detect AI-written text and verify webhook signatures.
This page has four complete programs for the Fileverity API, each in Node.js 18 or later and in Python 3.9 or later: check an image and wait for the result, check a large PDF, detect AI-written text, and receive webhooks with signature verification. Copy one, set FILEVERITY_API_KEY and run it.
The programs call the HTTPS API directly, with the fetch built into Node.js and the requests package in Python, so all you need is an HTTP client. The calls, fields and statuses are the ones described in Files, Analyses, Text detections and Webhooks.
| Program | What it does | Calls |
|---|---|---|
| Check an image | Uploads an image, waits for the result and prints the summary and the report URL. | POST /files, POST /analyses, GET /analyses/{id} |
| Check a large PDF | Uploads a document with a pre-signed URL, polls until the check finishes and saves the PDF report. | POST /files, GET /files/{id}, POST /analyses, GET /analyses/{id}, GET /analyses/{id}/report.pdf |
| Detect AI-written text | Sends a text file and prints the verdict, the score and each AI-written passage. | POST /text/detections |
| Receive webhooks | Verifies the signature of each delivery, ignores duplicates and handles the events. | Your endpoint, called with analysis.completed, analysis.failed and balance.low |
Before you run
- Create a Test key under API keys in the Console. A test key returns fixed sample results and is never charged, so every program runs at no cost. To check real files, create a Live key. The programs do not change.
- Node.js: save each program as an
.mjsfile, which lets it useimportand top-levelawait, and run it withnode. Nothing to install. - Python: run
pip install requests. The webhook receiver also needspip install flask.
export FILEVERITY_API_KEY=fv_test_...$env:FILEVERITY_API_KEY = "fv_test_..."How the programs handle keys, errors and retries
The three programs that call the API share the same request and api functions. Copy them into your own code as they are.
| Concern | What the programs do |
|---|---|
| Key | Read from FILEVERITY_API_KEY and never written into the code. Call the API from your server, not from browser code. |
| Idempotency | Every POST sends an Idempotency-Key, a new UUID per request, and reuses it on every retry, so a retry never runs or bills a check twice. To make running the whole script again safe, derive the key from your own record ID instead. See Idempotency. |
| Errors | A response that is not 2xx carries error.type, error.code, error.message, error.param and error.request_id. The programs stop with code: message (request request_id). Branch on code and quote the request ID to support. See Errors. |
| Retries | A network failure, a 5xx status (internal_error, analysis_failed, service_unavailable), rate_limited (429) and idempotency_in_progress (409) are retried, up to 5 attempts. The wait is the retry-after header when the response has one, otherwise 2, 4, 8 and 16 seconds. Every other 4xx stops at once. pending_files_limit is also a 429 but has no retry-after, so it is not retried. |
| Rate limit | When a response reports x-ratelimit-remaining-requests: 0, the program waits x-ratelimit-reset-requests seconds before the next call, so it does not run into a 429. See Rate limits and headers. |
| Timeouts | The Python programs time out after 90 seconds per request, longer than the 60 seconds that Prefer: wait can hold a request. |
Check an image and wait for the result
The program uploads the image as multipart, creates an analysis with Prefer: wait=60 so the request is held until the result is ready, and prints the summary and the report URL. If the wait ends first, the API answers 202 with status: processing, and the program polls GET /analyses/{id} every 3 seconds. See Wait, poll or use webhooks.
Save it as check-image.mjs or check_image.py and run node check-image.mjs photo.jpg or python check_image.py photo.jpg. Multipart takes images up to 4 MB. For a larger image, up to 10 MB, use the pre-signed upload in the next program with the content type of the image, for example image/jpeg.
import { randomUUID } from "node:crypto";import { readFile } from "node:fs/promises";import { basename } from "node:path"; const base = "https://api.fileverity.com/v1";const key = process.env.FILEVERITY_API_KEY;const path = process.argv[2];if (!key || !path) throw new Error("Usage: FILEVERITY_API_KEY=... node check-image.mjs photo.jpg"); const retryable = ["rate_limited", "idempotency_in_progress"];const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000)); async function request(url, init) { for (let attempt = 1; ; attempt++) { let response = null; let problem; try { response = await fetch(url, init); if (response.ok) { const remaining = response.headers.get("x-ratelimit-remaining-requests"); const reset = Number(response.headers.get("x-ratelimit-reset-requests")); if (remaining === "0") await sleep(reset); return response; } problem = (await response.json().catch(() => ({}))).error ?? {}; } catch (error) { problem = { code: "network_error", message: error.message }; } const status = response?.status ?? 0; const retry = status === 0 || status >= 500 || retryable.includes(problem.code); if (!retry || attempt === 5) { const { code = status, message = "Request failed", request_id: id = "unknown" } = problem; throw new Error(`${code}: ${message} (request ${id})`); } await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt); }} async function api(method, route, { headers, body } = {}) { const auth = { Authorization: `Bearer ${key}` }; const response = await request(base + route, { method, headers: { ...auth, ...headers }, body }); return response.json();} const form = new FormData();form.append("file", new Blob([await readFile(path)]), basename(path));const file = await api("POST", "/files", { headers: { "Idempotency-Key": randomUUID() }, body: form,}); let analysis = await api("POST", "/analyses", { headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID(), Prefer: "wait=60", }, body: JSON.stringify({ model: "fvimg-2", file: file.id }),});while (analysis.status === "processing") { await sleep(3); analysis = await api("GET", `/analyses/${analysis.id}`);} if (analysis.status !== "succeeded") { throw new Error(`The check ${analysis.status}: ${analysis.error?.message}`);}const { level, action, assessment, ai, edit } = analysis.summary;console.log(`${analysis.id} ${analysis.status}`);console.log(`level ${level}, action ${action}, assessment ${assessment}`);console.log(`AI generation: ${ai}, editing: ${edit}`);console.log(`Report: ${analysis.report_url}`);import osimport sysimport timeimport uuid import requests BASE = "https://api.fileverity.com/v1"KEY = os.environ.get("FILEVERITY_API_KEY")if not KEY or len(sys.argv) < 2: sys.exit("Usage: FILEVERITY_API_KEY=... python check_image.py photo.jpg") RETRYABLE = ("rate_limited", "idempotency_in_progress") def request(method, url, **kwargs): for attempt in range(1, 6): try: response = requests.request(method, url, timeout=90, **kwargs) except requests.RequestException as error: response, problem = None, {"code": "network_error", "message": str(error)} else: if response.ok: if response.headers.get("x-ratelimit-remaining-requests") == "0": time.sleep(int(response.headers["x-ratelimit-reset-requests"])) return response try: problem = response.json()["error"] except (ValueError, KeyError, TypeError): problem = {} status = response.status_code if response is not None else 0 retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE if not retry or attempt == 5: code = problem.get("code", status) message = problem.get("message", "Request failed") request_id = problem.get("request_id", "unknown") raise RuntimeError(f"{code}: {message} (request {request_id})") wait = response.headers.get("retry-after") if response is not None else None time.sleep(int(wait or 2**attempt)) def api(method, route, headers=None, **kwargs): headers = {"Authorization": f"Bearer {KEY}", **(headers or {})} return request(method, BASE + route, headers=headers, **kwargs).json() with open(sys.argv[1], "rb") as source: data = source.read()upload = {"file": (os.path.basename(sys.argv[1]), data)}file = api("POST", "/files", headers={"Idempotency-Key": str(uuid.uuid4())}, files=upload) analysis = api( "POST", "/analyses", headers={"Idempotency-Key": str(uuid.uuid4()), "Prefer": "wait=60"}, json={"model": "fvimg-2", "file": file["id"]},)while analysis["status"] == "processing": time.sleep(3) analysis = api("GET", f"/analyses/{analysis['id']}") if analysis["status"] != "succeeded": problem = analysis["error"] or {} sys.exit(f"The check {analysis['status']}: {problem.get('message')}")summary = analysis["summary"]print(f"{analysis['id']} {analysis['status']}")print(f"level {summary['level']}, action {summary['action']}, assessment {summary['assessment']}")print(f"AI generation: {summary['ai']}, editing: {summary['edit']}")print(f"Report: {analysis['report_url']}")an_8Hk2pQ4vN9sL1xR7cW3mY6zB succeededlevel L2, action accept, assessment authentic_untouchedAI generation: very_low, editing: noneReport: https://fileverity.com/verify/FV-7K2M-9QXD-4T8Rsummary is the stable, machine-readable verdict of an image check: level, action, assessment, ai and edit, described in Image summary. report_url is the public verification page of the report. With a test key the result is a fixed sample, and its report URL has no page behind it.
Check a large PDF
Documents up to 50 MB are uploaded in three steps: ask for an upload URL with POST /files and a JSON body, send the bytes to that URL with PUT, and wait until the file shows uploaded. The program then creates the analysis and polls every 5 seconds, because documents can be queued and a check can take a while. It prints the progress, the summary and the report URL, and saves the PDF report next to the script.
Save it as check-pdf.mjs or check_pdf.py and run node check-pdf.mjs contract.pdf or python check_pdf.py contract.pdf. For another document type, change content_type to one of the accepted content types. For an encrypted document, add options: { password } to the analysis request.
import { randomUUID } from "node:crypto";import { readFile, writeFile } from "node:fs/promises";import { basename } from "node:path"; const base = "https://api.fileverity.com/v1";const key = process.env.FILEVERITY_API_KEY;const path = process.argv[2];if (!key || !path) throw new Error("Usage: FILEVERITY_API_KEY=... node check-pdf.mjs contract.pdf"); const retryable = ["rate_limited", "idempotency_in_progress"];const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000)); async function request(url, init) { for (let attempt = 1; ; attempt++) { let response = null; let problem; try { response = await fetch(url, init); if (response.ok) { const remaining = response.headers.get("x-ratelimit-remaining-requests"); const reset = Number(response.headers.get("x-ratelimit-reset-requests")); if (remaining === "0") await sleep(reset); return response; } problem = (await response.json().catch(() => ({}))).error ?? {}; } catch (error) { problem = { code: "network_error", message: error.message }; } const status = response?.status ?? 0; const retry = status === 0 || status >= 500 || retryable.includes(problem.code); if (!retry || attempt === 5) { const { code = status, message = "Request failed", request_id: id = "unknown" } = problem; throw new Error(`${code}: ${message} (request ${id})`); } await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt); }} async function api(method, route, { headers, body } = {}) { const auth = { Authorization: `Bearer ${key}` }; const response = await request(base + route, { method, headers: { ...auth, ...headers }, body }); return response.json();} const bytes = await readFile(path);const file = await api("POST", "/files", { headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() }, body: JSON.stringify({ filename: basename(path), content_type: "application/pdf", bytes: bytes.length, }),}); if (file.status === "pending") { const { url, method, headers } = file.upload; await request(url, { method, headers, body: bytes }); let ready = false; for (let attempt = 0; attempt < 15 && !ready; attempt++) { ready = (await api("GET", `/files/${file.id}`)).status === "uploaded"; if (!ready) await sleep(2); } if (!ready) throw new Error("The upload was not confirmed.");}console.log(`${file.id} uploaded`); let analysis = await api("POST", "/analyses", { headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() }, body: JSON.stringify({ model: "fvdoc-3", file: file.id }),});const deadline = Date.now() + 15 * 60 * 1000;while (analysis.status === "processing") { if (Date.now() > deadline) { throw new Error(`Still processing. Fetch it later with GET /analyses/${analysis.id}`); } console.log(`${analysis.id} processing: ${analysis.progress?.stage ?? "starting"}`); await sleep(5); analysis = await api("GET", `/analyses/${analysis.id}`);} if (analysis.status !== "succeeded") { throw new Error(`The check ${analysis.status}: ${analysis.error?.message}`);}const { level, action, reason, pages, images, text, findings } = analysis.summary;console.log(`${analysis.id} ${analysis.status}`);console.log(`level ${level}, action ${action}, reason ${reason ?? "none"}`);console.log(`${pages} pages, ${images} photos examined, text: ${text ?? "not evaluated"}`);console.log(`findings: ${findings.join(", ") || "none"}`);console.log(`Report: ${analysis.report_url}`); if (analysis.livemode) { const pdf = await request(`${base}/analyses/${analysis.id}/report.pdf`, { headers: { Authorization: `Bearer ${key}` }, }); const name = `Fileverity-${analysis.report_id}.pdf`; await writeFile(name, Buffer.from(await pdf.arrayBuffer())); console.log(`Saved ${name}`);}import osimport sysimport timeimport uuid import requests BASE = "https://api.fileverity.com/v1"KEY = os.environ.get("FILEVERITY_API_KEY")if not KEY or len(sys.argv) < 2: sys.exit("Usage: FILEVERITY_API_KEY=... python check_pdf.py contract.pdf") RETRYABLE = ("rate_limited", "idempotency_in_progress") def request(method, url, **kwargs): for attempt in range(1, 6): try: response = requests.request(method, url, timeout=90, **kwargs) except requests.RequestException as error: response, problem = None, {"code": "network_error", "message": str(error)} else: if response.ok: if response.headers.get("x-ratelimit-remaining-requests") == "0": time.sleep(int(response.headers["x-ratelimit-reset-requests"])) return response try: problem = response.json()["error"] except (ValueError, KeyError, TypeError): problem = {} status = response.status_code if response is not None else 0 retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE if not retry or attempt == 5: code = problem.get("code", status) message = problem.get("message", "Request failed") request_id = problem.get("request_id", "unknown") raise RuntimeError(f"{code}: {message} (request {request_id})") wait = response.headers.get("retry-after") if response is not None else None time.sleep(int(wait or 2**attempt)) def api(method, route, headers=None, **kwargs): headers = {"Authorization": f"Bearer {KEY}", **(headers or {})} return request(method, BASE + route, headers=headers, **kwargs).json() with open(sys.argv[1], "rb") as source: data = source.read()file = api( "POST", "/files", headers={"Idempotency-Key": str(uuid.uuid4())}, json={ "filename": os.path.basename(sys.argv[1]), "content_type": "application/pdf", "bytes": len(data), },) if file["status"] == "pending": request("PUT", file["upload"]["url"], headers=file["upload"]["headers"], data=data) for _ in range(15): if api("GET", f"/files/{file['id']}")["status"] == "uploaded": break time.sleep(2) else: sys.exit("The upload was not confirmed.")print(f"{file['id']} uploaded") analysis = api( "POST", "/analyses", headers={"Idempotency-Key": str(uuid.uuid4())}, json={"model": "fvdoc-3", "file": file["id"]},)deadline = time.time() + 15 * 60while analysis["status"] == "processing": if time.time() > deadline: sys.exit(f"Still processing. Fetch it later with GET /analyses/{analysis['id']}") stage = (analysis["progress"] or {}).get("stage", "starting") print(f"{analysis['id']} processing: {stage}") time.sleep(5) analysis = api("GET", f"/analyses/{analysis['id']}") if analysis["status"] != "succeeded": problem = analysis["error"] or {} sys.exit(f"The check {analysis['status']}: {problem.get('message')}")summary = analysis["summary"]reason = summary["reason"] or "none"text = summary["text"] or "not evaluated"findings = ", ".join(summary["findings"]) or "none"print(f"{analysis['id']} {analysis['status']}")print(f"level {summary['level']}, action {summary['action']}, reason {reason}")print(f"{summary['pages']} pages, {summary['images']} photos examined, text: {text}")print(f"findings: {findings}")print(f"Report: {analysis['report_url']}") if analysis["livemode"]: url = f"{BASE}/analyses/{analysis['id']}/report.pdf" pdf = request("GET", url, headers={"Authorization": f"Bearer {KEY}"}) name = f"Fileverity-{analysis['report_id']}.pdf" with open(name, "wb") as target: target.write(pdf.content) print(f"Saved {name}")file_Pw3mY6zB8Hk2pQ4vN9sL1xR7 uploadedan_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: queuedan_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: analyzingan_Zc4vB7nP1sK8tL3wR6yQm2Xa processing: reportingan_Zc4vB7nP1sK8tL3wR6yQm2Xa succeededlevel L2, action accept, reason clean12 pages, 3 photos examined, text: humanfindings: noneReport: https://fileverity.com/verify/FV-3PXH-6WQ2-8NDCSaved Fileverity-FV-3PXH-6WQ2-8NDC.pdf- The
PUTgoes straight to storage. It takes noAuthorizationheader, and the headers inupload.headersare sent exactly as returned. See Pre-signed upload. - With a test key the file comes back
uploadedand there is nothing to send, so the program skips the upload. The analysis returns a fixed sample at once, and there is no sample PDF, so the program saves the report only for a live key. - Download the PDF as soon as the check finishes. The page images and text excerpts of a document are kept for 3 hours, and a PDF downloaded later leaves them out. See PDF report.
- A check that ends with
failedorcanceledwas not charged. The program stops with the error message of the analysis.
Detect AI-written text
Text detection takes a single synchronous call. The program reads a text file, sends it to POST /text/detections and prints the verdict, the score, the share of AI-written text and every AI-written passage with its offsets. AI-written text detection supports English, French, German, Spanish and Chinese (en, fr, de, es, zh). The language is detected automatically; pass a code as the second argument to set it. The text must have at least 100 words and at most 100,000 characters.
Save it as detect-text.mjs or detect_text.py and run node detect-text.mjs essay.txt or python detect_text.py essay.txt fr. The Quickstart has a command that writes a sample essay.txt. With a test key the response is a fixed result for human-written text, so the program prints no passage.
import { randomUUID } from "node:crypto";import { readFile } from "node:fs/promises"; const base = "https://api.fileverity.com/v1";const key = process.env.FILEVERITY_API_KEY;const [, , path, language] = process.argv;if (!key || !path) { throw new Error("Usage: FILEVERITY_API_KEY=... node detect-text.mjs essay.txt [en|fr|de|es|zh]");} const retryable = ["rate_limited", "idempotency_in_progress"];const sleep = (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000)); async function request(url, init) { for (let attempt = 1; ; attempt++) { let response = null; let problem; try { response = await fetch(url, init); if (response.ok) { const remaining = response.headers.get("x-ratelimit-remaining-requests"); const reset = Number(response.headers.get("x-ratelimit-reset-requests")); if (remaining === "0") await sleep(reset); return response; } problem = (await response.json().catch(() => ({}))).error ?? {}; } catch (error) { problem = { code: "network_error", message: error.message }; } const status = response?.status ?? 0; const retry = status === 0 || status >= 500 || retryable.includes(problem.code); if (!retry || attempt === 5) { const { code = status, message = "Request failed", request_id: id = "unknown" } = problem; throw new Error(`${code}: ${message} (request ${id})`); } await sleep(Number(response?.headers.get("retry-after")) || 2 ** attempt); }} async function api(method, route, { headers, body } = {}) { const auth = { Authorization: `Bearer ${key}` }; const response = await request(base + route, { method, headers: { ...auth, ...headers }, body }); return response.json();} const text = await readFile(path, "utf8");const detection = await api("POST", "/text/detections", { headers: { "Content-Type": "application/json", "Idempotency-Key": randomUUID() }, body: JSON.stringify({ text, language }),}); const { verdict, score, ai_fraction: share, usage, segments } = detection;console.log(`verdict: ${verdict} (score ${score.toFixed(2)})`);console.log(`AI-written share: ${Math.round(share * 100)}% of ${usage.characters} characters`);console.log(`language: ${detection.language}, billed units: ${usage.billed_units}`);const points = Array.from(text);for (const { start, end, score: passage } of segments) { const excerpt = points.slice(start, Math.min(end, start + 80)).join(""); console.log(`${start}-${end} score ${passage.toFixed(2)}: ${excerpt}`);}import osimport sysimport timeimport uuid import requests BASE = "https://api.fileverity.com/v1"KEY = os.environ.get("FILEVERITY_API_KEY")if not KEY or len(sys.argv) < 2: sys.exit("Usage: FILEVERITY_API_KEY=... python detect_text.py essay.txt [en|fr|de|es|zh]")sys.stdout.reconfigure(encoding="utf-8") RETRYABLE = ("rate_limited", "idempotency_in_progress") def request(method, url, **kwargs): for attempt in range(1, 6): try: response = requests.request(method, url, timeout=90, **kwargs) except requests.RequestException as error: response, problem = None, {"code": "network_error", "message": str(error)} else: if response.ok: if response.headers.get("x-ratelimit-remaining-requests") == "0": time.sleep(int(response.headers["x-ratelimit-reset-requests"])) return response try: problem = response.json()["error"] except (ValueError, KeyError, TypeError): problem = {} status = response.status_code if response is not None else 0 retry = status == 0 or status >= 500 or problem.get("code") in RETRYABLE if not retry or attempt == 5: code = problem.get("code", status) message = problem.get("message", "Request failed") request_id = problem.get("request_id", "unknown") raise RuntimeError(f"{code}: {message} (request {request_id})") wait = response.headers.get("retry-after") if response is not None else None time.sleep(int(wait or 2**attempt)) def api(method, route, headers=None, **kwargs): headers = {"Authorization": f"Bearer {KEY}", **(headers or {})} return request(method, BASE + route, headers=headers, **kwargs).json() with open(sys.argv[1], encoding="utf-8") as source: text = source.read()language = sys.argv[2] if len(sys.argv) > 2 else Nonedetection = api( "POST", "/text/detections", headers={"Idempotency-Key": str(uuid.uuid4())}, json={"text": text, "language": language},) usage = detection["usage"]share = round(detection["ai_fraction"] * 100)print(f"verdict: {detection['verdict']} (score {detection['score']:.2f})")print(f"AI-written share: {share}% of {usage['characters']} characters")print(f"language: {detection['language']}, billed units: {usage['billed_units']}")for segment in detection["segments"]: start, end = segment["start"], segment["end"] excerpt = text[start : min(end, start + 80)] print(f"{start}-{end} score {segment['score']:.2f}: {excerpt}")verdict: mixed (score 0.97)AI-written share: 50% of 889 characterslanguage: en, billed units: 1267-495 score 0.97: In today's fast-paced logistics landscape, it is essential for businesses to pri590-809 score 0.93: Moving forward, it is crucial to implement robust quality assurance measures andstart and end count Unicode code points, not bytes and not UTF-16 units. Python strings already index by code point; the Node.js program splits the text with Array.from first. See Segment offsets.
Text detection returns a statistical estimate, not proof of authorship. Use it as one input to a human review, and do not use it alone to make decisions that affect a person.
Receive webhooks and verify signatures
A webhook tells your server when a check finishes, so you do not have to poll. This receiver is a minimal HTTP server: the http module of Node.js, and Flask for Python. It verifies the signature of every delivery, ignores a delivery it has already handled and prints the event. It follows the Webhooks page.
- Make port 3000 reachable at a public HTTPS address, for example with a tunnel to your machine. Endpoint URLs must use HTTPS on the default port 443 and point to a public address.
- In the Console, open Webhooks, add an endpoint with the URL
https://your-host/webhooks/fileverityand choose its events. Copy the signing secret, which starts withwhsec_. It is shown once. - Save the program as
webhook-receiver.mjsorwebhook_receiver.pyand start it with the secret:FILEVERITY_WEBHOOK_SECRET=whsec_... node webhook-receiver.mjsorFILEVERITY_WEBHOOK_SECRET=whsec_... python webhook_receiver.py. - Choose Send test on the endpoint. Test keys never send events; Send test delivers a signed sample event instead.
import { createHmac, timingSafeEqual } from "node:crypto";import { createServer } from "node:http"; const secret = process.env.FILEVERITY_WEBHOOK_SECRET;if (!secret) throw new Error("Set FILEVERITY_WEBHOOK_SECRET to the endpoint's signing secret.");const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");const port = Number(process.env.PORT) || 3000;const seen = new Set(); function verify(headers, body) { const id = headers["webhook-id"]; const timestamp = headers["webhook-timestamp"]; const signatures = headers["webhook-signature"]; if (!id || !timestamp || !signatures) return false; if (!(Math.abs(Date.now() / 1000 - Number(timestamp)) <= 300)) return false; const expected = createHmac("sha256", key).update(`${id}.${timestamp}.`).update(body).digest(); return signatures.split(" ").some((entry) => { const [version, value = ""] = entry.split(","); const given = Buffer.from(value, "base64"); return version === "v1" && given.length === expected.length && timingSafeEqual(given, expected); });} function handleEvent(event) { const { type, data } = event; if (data.test) { console.log(`Test event ${event.id}`); } else if (type === "analysis.completed") { const { level, action } = data.summary; console.log(`${data.id} succeeded: level ${level}, action ${action}, ${data.report_url}`); } else if (type === "analysis.failed") { console.log(`${data.id} ${data.status}: ${data.error.code}`); } else if (type === "balance.low") { const money = (cents) => `US$${(cents / 100).toFixed(2)}`; console.log(`Balance ${money(data.balance)} is below your ${money(data.threshold)} alert`); }} async function receive(req, res) { if (req.method !== "POST" || req.url !== "/webhooks/fileverity") return res.writeHead(404).end(); const chunks = []; let size = 0; for await (const chunk of req) { size += chunk.length; if (size > 1_000_000) return res.writeHead(413).end(); chunks.push(chunk); } const body = Buffer.concat(chunks); if (!verify(req.headers, body)) return res.writeHead(400).end(); const event = JSON.parse(body.toString("utf8")); try { if (!seen.has(event.id)) { handleEvent(event); seen.add(event.id); } } catch (error) { console.error(`Could not handle ${event.id}`, error); return res.writeHead(500).end(); } res.writeHead(200).end();} createServer((req, res) => receive(req, res).catch(() => res.destroy())).listen(port, () => console.log(`Listening on port ${port}`));import base64import hashlibimport hmacimport jsonimport osimport threadingimport time from flask import Flask, request SECRET = os.environ.get("FILEVERITY_WEBHOOK_SECRET")if not SECRET: raise SystemExit("Set FILEVERITY_WEBHOOK_SECRET to the endpoint's signing secret.")KEY = base64.b64decode(SECRET.removeprefix("whsec_"))seen = set()lock = threading.Lock()app = Flask(__name__)app.config["MAX_CONTENT_LENGTH"] = 1_000_000 def verify(headers, body): message_id = headers.get("webhook-id") timestamp = headers.get("webhook-timestamp", "") signatures = headers.get("webhook-signature", "") if not message_id or not timestamp.isdecimal() or abs(time.time() - int(timestamp)) > 300: return False signed = f"{message_id}.{timestamp}.".encode() + body expected = hmac.new(KEY, signed, hashlib.sha256).digest() for entry in signatures.split(" "): version, _, value = entry.partition(",") try: given = base64.b64decode(value, validate=True) except ValueError: continue if version == "v1" and hmac.compare_digest(given, expected): return True return False def handle_event(event): data = event["data"] if data.get("test"): print(f"Test event {event['id']}") elif event["type"] == "analysis.completed": level, action = data["summary"]["level"], data["summary"]["action"] print(f"{data['id']} succeeded: level {level}, action {action}, {data['report_url']}") elif event["type"] == "analysis.failed": print(f"{data['id']} {data['status']}: {data['error']['code']}") elif event["type"] == "balance.low": balance, threshold = data["balance"] / 100, data["threshold"] / 100 print(f"Balance US${balance:.2f} is below your US${threshold:.2f} alert") @app.post("/webhooks/fileverity")def receive(): body = request.get_data() if not verify(request.headers, body): return "", 400 event = json.loads(body) with lock: try: if event["id"] not in seen: handle_event(event) seen.add(event["id"]) except Exception: app.logger.exception("Could not handle %s", event["id"]) return "", 500 return "", 200 if __name__ == "__main__": port = int(os.environ.get("PORT", "3000")) print(f"Listening on port {port}") app.run(port=port)Listening on port 3000Test event evt_Qm4vB7nP1sK8tL3wR6yZ2cXaan_8Hk2pQ4vN9sL1xR7cW3mY6zB succeeded: level L2, action accept, https://fileverity.com/verify/FV-7K2M-9QXD-4T8R- Raw body. The signature covers the exact bytes that were sent. The programs read the body as bytes, compute an HMAC-SHA256 over
webhook-id, a dot,webhook-timestamp, a dot and those bytes with the base64-decoded secret as the key, and parse the JSON only after the signature is valid. - Constant-time comparison.
timingSafeEqualin Node.js andhmac.compare_digestin Python compare the result with everyv1signature in thewebhook-signatureheader. - Timestamp tolerance. A
webhook-timestampmore than five minutes from the clock, or one that is not a number, is rejected. - Fast answers. A valid delivery gets
200as soon as the event is handled. A bad signature gets400. If handling fails, the program answers500, and Fileverity retries later with the same event ID. See Delivery and retries. - Duplicates. Delivery is at least once, and a retry keeps the event ID. The programs remember the IDs they have handled and answer
200without handling an event twice. In production, replace the in-memory set with a unique constraint on the event ID in your database, and put the work on a queue so the answer stays fast. - Event data.
analysis.completedandanalysis.failedcarry the analysis withoutresult; callGET /analyses/{id}when you need the full report.balanceandthresholdinbalance.loware in US cents.
Where to go next
- Run a whole intake flow with human review in Document checks in an intake workflow.
- Look up any field in the API reference, or load the OpenAPI document into your own tools.
- Go live with a Live key, a monthly spend limit and a
balance.lowwebhook. See Pricing and billing. - Need help? Quote the
x-request-idheader when you contact support.