from typing import Dict, List, Optional
"""VerIQ — FastAPI application (v1 MVP).
Standalone public IQ test with signed, verifiable results.
Run: uvicorn backend.main:app --host 127.0.0.1 --port 8801
"""
import json
import os
import re
import secrets
import time
from pathlib import Path
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse
from pydantic import BaseModel, Field
from . import db, oauth, og, scoring, signing
app = FastAPI(title="VerIQ", docs_url=None, redoc_url=None)
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
# --- in-memory test sessions (single-process MVP) ---------------------------
# session_token -> {task_ids, started_at, ip}
SESSIONS: Dict[str, dict] = {}
SESSION_TTL_S = 60 * 60 # 1 hour to finish after start
# --- simple per-IP rate limiting --------------------------------------------
_RATE: Dict[str, List[float]] = {}
RATE_LIMIT_STARTS_PER_HOUR = 10
def _rate_ok(ip: str) -> bool:
now = time.time()
window = _RATE.setdefault(ip, [])
window[:] = [t for t in window if now - t < 3600]
if len(window) >= RATE_LIMIT_STARTS_PER_HOUR:
return False
window.append(now)
return True
def _client_ip(request: Request) -> str:
fwd = request.headers.get("x-forwarded-for")
if fwd:
return fwd.split(",")[0].strip()
return request.client.host if request.client else "unknown"
def _inject_meta(html: str, title: str, description: str,
image_url: Optional[str] = None) -> str:
"""Server-side OG meta injection for social scrapers."""
esc = lambda s: s.replace("&", "&").replace('"', """).replace("<", "<")
meta = f"<title>{esc(title)}</title>\n"
meta += f'<meta property="og:title" content="{esc(title)}">\n'
meta += f'<meta property="og:description" content="{esc(description)}">\n'
if image_url:
meta += f'<meta property="og:image" content="{esc(image_url)}">\n'
meta += '<meta name="twitter:card" content="summary_large_image">\n'
return re.sub(r"<title>.*?</title>", meta, html, count=1, flags=re.S)
# --- models ------------------------------------------------------------------
class StartResponse(BaseModel):
session_token: str
tasks: List[dict]
time_limit_s: int
class FinishRequest(BaseModel):
session_token: str
answers: Dict[str, int] = Field(default_factory=dict)
total_time_s: float = 0.0
class FinishResponse(BaseModel):
result_id: str
score: dict
# --- static pages -------------------------------------------------------------
@app.get("/")
def index():
return FileResponse(FRONTEND_DIR / "index.html")
def _identity_from(request: Request) -> Optional[dict]:
return oauth.verify_identity(request.cookies.get(oauth.IDENTITY_COOKIE, ""))
@app.get("/login")
def login_page():
"""Explicit provider choice: X or Instagram (button disabled until configured)."""
return FileResponse(FRONTEND_DIR / "login.html")
@app.get("/test")
def test_page(request: Request):
"""Auth-before-test: no valid identity cookie -> explicit provider choice."""
if _identity_from(request) is None:
return RedirectResponse("/login", status_code=303)
return FileResponse(FRONTEND_DIR / "test.html")
def _result_context(result_id: str):
row = db.get_result(result_id)
if row is None:
return None, None
valid = signing.verify_payload(row["payload"], row["signature"])
return row, valid
@app.get("/result/{result_id}")
def result_page(result_id: str):
row, valid = _result_context(result_id)
if row is None:
raise HTTPException(404, "Result not found")
html = (FRONTEND_DIR / "result.html").read_text(encoding="utf-8")
title = f"My IQ is {row['score']} — VerIQ"
sig = "valid" if valid else "INVALID"
desc = (f"Higher than {row['percentile']}% of test takers. "
f"Signature {sig}. Verify it yourself.")
return HTMLResponse(_inject_meta(html, title, desc, f"/og/{result_id}.png"))
@app.get("/v/{result_id}")
def verify_page(result_id: str):
row, valid = _result_context(result_id)
if row is None:
raise HTTPException(404, "Result not found")
html = (FRONTEND_DIR / "verify.html").read_text(encoding="utf-8")
title = f"Verified IQ {row['score']} — VerIQ"
sig = "valid" if valid else "INVALID"
desc = (f"Publicly verified result: IQ {row['score']}, higher than "
f"{row['percentile']}% of test takers. Signature {sig}.")
return HTMLResponse(_inject_meta(html, title, desc, f"/og/{result_id}.png"))
@app.get("/og/{result_id}.png")
def og_image(result_id: str):
row, valid = _result_context(result_id)
if row is None:
raise HTTPException(404, "Result not found")
path = og.generate_card(row["score"], row["percentile"], valid,
bool(row["flagged"]), result_id)
return FileResponse(path, media_type="image/png")
@app.get("/assets/{filename}")
def assets(filename: str):
# simple safe-join guard
if "/" in filename or ".." in filename:
raise HTTPException(404)
path = FRONTEND_DIR / "assets" / filename
if not path.exists():
raise HTTPException(404)
return FileResponse(path)
@app.get("/.well-known/veriq-pubkey.txt")
def pubkey():
return PlainTextResponse(signing.public_key_b64() + "\n")
# --- API ----------------------------------------------------------------------
@app.post("/api/start", response_model=StartResponse)
def api_start(request: Request):
ip = _client_ip(request)
if not _rate_ok(ip):
raise HTTPException(429, "Too many attempts, try later")
bank = scoring.load_bank()
all_tasks = bank["tasks"]
# v1: use all tasks, shuffled deterministically per session
ids = [t["id"] for t in all_tasks]
secrets.SystemRandom().shuffle(ids)
tasks = [scoring.task_by_id(tid) for tid in ids]
token = secrets.token_urlsafe(24)
SESSIONS[token] = {
"task_ids": ids,
"started_at": time.time(),
"ip": ip,
}
# prune expired sessions
now = time.time()
for k in [k for k, v in SESSIONS.items() if now - v["started_at"] > SESSION_TTL_S]:
SESSIONS.pop(k, None)
db.log_event("start", {"ip_hash": secrets.token_hex(4)})
# strip answers before sending
public_tasks = [
{k: t[k] for k in ("id", "category", "difficulty", "time_limit_s",
"prompt", "choices")}
for t in tasks
]
total_limit = sum(t.get("time_limit_s", 60) for t in tasks)
return StartResponse(session_token=token, tasks=public_tasks,
time_limit_s=total_limit)
@app.post("/api/finish", response_model=FinishResponse)
def api_finish(req: FinishRequest, request: Request):
sess = SESSIONS.get(req.session_token)
if sess is None:
raise HTTPException(404, "Session not found or expired")
if time.time() - sess["started_at"] > SESSION_TTL_S:
SESSIONS.pop(req.session_token, None)
raise HTTPException(410, "Session expired")
SESSIONS.pop(req.session_token, None)
task_ids = sess["task_ids"]
# ignore answers for unknown tasks
answers = {k: v for k, v in req.answers.items() if scoring.task_by_id(k)}
flagged = scoring.sanity_check(answers, task_ids, req.total_time_s)
score = scoring.score_test(answers, task_ids, req.total_time_s, flagged)
# identity comes from the login cookie (auth-before-test); a new result
# supersedes all previous active results of the same identity
ident = _identity_from(request)
identity = None
if ident:
identity = ident["provider"] + ":@" + ident["username"]
db.supersede_identity(identity)
result_id = db.new_result_id()
completed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
payload = signing.canonical_payload(
result_id=result_id,
score=score["iq"],
percentile=score["percentile"],
bank_version=scoring.bank_version(),
identity=identity,
completed_at=completed_at,
)
signature = signing.sign_payload(payload)
db.save_result(
result_id=result_id,
score=score["iq"],
percentile=score["percentile"],
bank_version=scoring.bank_version(),
identity=identity,
completed_at=completed_at,
payload=payload,
signature=signature,
answers=answers,
flagged=flagged,
score_json=json.dumps(score),
)
db.log_event("finish", {"flagged": flagged, "iq": score["iq"]})
return FinishResponse(result_id=result_id, score=score)
@app.get("/api/result/{result_id}")
def api_result(result_id: str):
row = db.get_result(result_id)
if row is None:
raise HTTPException(404, "Result not found")
valid = signing.verify_payload(row["payload"], row["signature"])
payload = json.loads(row["payload"])
snapshot = {}
if row.get("score_json"):
try:
snapshot = json.loads(row["score_json"])
except Exception:
snapshot = {}
return JSONResponse({
"result_id": result_id,
"iq": row["score"],
"percentile": row["percentile"],
"completed_at": row["completed_at"],
"identity": row["identity"],
"bank_version": row["bank_version"],
"signature_valid": valid,
"signature": row["signature"],
"payload": payload,
"flagged": bool(row["flagged"]),
"superseded": bool(row.get("superseded", 0)),
"categories": snapshot.get("categories", {}),
"calibrated": snapshot.get("calibrated"),
"total_time_s": snapshot.get("total_time_s"),
})
@app.get("/stats")
def stats_page():
return FileResponse(FRONTEND_DIR / "stats.html")
@app.get("/how")
def how_page():
return FileResponse(FRONTEND_DIR / "how.html")
@app.get("/api/me")
def api_me(request: Request):
ident = _identity_from(request)
if not ident:
return JSONResponse({"identity": None})
return JSONResponse({
"identity": ident["provider"] + ":@" + ident["username"],
"username": ident["username"],
"provider": ident["provider"],
})
@app.get("/api/stats")
def api_stats():
return JSONResponse(db.stats_overview())
# --- identity binding (phase 2 scaffolding) -----------------------------------
def _login_error_page(provider: str) -> HTMLResponse:
"""Standalone error page: login failed, no identity cookie is set."""
body = (
'<!doctype html><html><head><meta charset="utf-8">'
'<title>VerIQ - login failed</title></head>'
'<body style="font-family:system-ui,sans-serif;background:#111;color:#eee;'
'margin:0;display:flex;align-items:center;justify-content:center;min-height:100vh">'
'<div style="max-width:520px;padding:28px;border:1px solid #444;border-radius:10px">'
'<h2 style="margin-top:0">Login failed</h2>'
'<p>' + provider + ' did not return a usable profile. No result was '
'affected — you simply are not logged in. You can try again.</p>'
'<p><a href="/login" style="color:#7ab">Back to login</a> · '
'<a href="/" style="color:#7ab">Back to home</a></p>'
'</div></body></html>'
)
return HTMLResponse(body, status_code=502)
@app.get("/auth/x/start")
def auth_x_start():
"""Begin X OAuth login. 503 until apps registered."""
if not oauth.x_configured():
raise HTTPException(503, "X binding not configured yet")
url = oauth.x_authorize_url()
if url is None:
raise HTTPException(503, "X binding not configured yet")
return RedirectResponse(url[0], status_code=303)
@app.get("/auth/x/callback")
def auth_x_callback(code: str = None, state: str = None):
if not oauth.x_configured():
raise HTTPException(503, "X binding not configured yet")
if not code or not state:
raise HTTPException(400, "missing code/state")
flow = oauth.pop_flow(state)
if flow is None or flow["provider"] != "x":
raise HTTPException(400, "invalid or expired state")
tok = oauth.x_exchange(code, flow["verifier"])
user = oauth.x_userinfo(tok.get("access_token")) if tok else None
if not user or not user.get("username"):
return _login_error_page("X")
token = oauth.sign_identity("x", str(user.get("id", "")), user["username"])
db.log_event("login", {"provider": "x", "username": user["username"]})
resp = RedirectResponse("/test", status_code=303)
resp.set_cookie(oauth.IDENTITY_COOKIE, token, max_age=oauth.IDENTITY_TTL_S,
httponly=True, samesite="lax", secure=True)
return resp
@app.get("/auth/ig/start")
def auth_ig_start():
"""Begin Instagram OAuth login (same identity-cookie flow as X)."""
if not oauth.ig_configured():
raise HTTPException(503, "Instagram binding not configured yet")
url = oauth.ig_authorize_url("")
if url is None:
raise HTTPException(503, "Instagram binding not configured yet")
return RedirectResponse(url[0], status_code=303)
@app.get("/auth/ig/callback")
def auth_ig_callback(code: str = None, state: str = None):
if not oauth.ig_configured():
raise HTTPException(503, "Instagram binding not configured yet")
if not code or not state:
raise HTTPException(400, "missing code/state")
flow = oauth.pop_flow(state)
if flow is None or flow["provider"] != "ig":
raise HTTPException(400, "invalid or expired state")
tok = oauth.ig_exchange(code)
user = oauth.ig_userinfo(tok.get("access_token")) if tok else None
if not user or not user.get("username"):
return _login_error_page("Instagram")
token = oauth.sign_identity("ig", str(user.get("id", "")), user["username"])
db.log_event("login", {"provider": "ig", "username": user["username"]})
resp = RedirectResponse("/test", status_code=303)
resp.set_cookie(oauth.IDENTITY_COOKIE, token, max_age=oauth.IDENTITY_TTL_S,
httponly=True, samesite="lax", secure=True)
return resp
# --- startup --------------------------------------------------------------------
@app.on_event("startup")
def startup():
db.init_db()
# fail fast if bank is missing/broken
scoring.load_bank()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=int(
os.environ.get("VERIQ_PORT", "8801")))
