Update app.py

app.py

@@ -1,166 +1,220 @@
 import os
 import json
 import time
+import hmac
 import hashlib
-import threading
+from collections import deque
 from typing import Any, Dict, Optional
 
-import requests
-from fastapi import FastAPI, Request, HTTPException
+from fastapi import FastAPI, Request, Header, HTTPException
 from fastapi.responses import JSONResponse
+import gradio as gr
 
-# ----------------------------
-# Config
-# ----------------------------
-WAKEUP_URL = os.getenv("WAKEUP_URL", "http://163.245.192.149:8080/wakeup").strip()
-TIMEOUT = int(os.getenv("RELAY_TIMEOUT", "30"))
-
-# Flood guard: keep this at 1 until you're 100% sure you're safe
-MAX_INFLIGHT = int(os.getenv("MAX_INFLIGHT", "1"))
-
-# Dedupe cache TTL (seconds). Keep reasonable so it doesn't grow forever.
-SEEN_TTL_SECONDS = int(os.getenv("SEEN_TTL_SECONDS", "86400"))  # 24h default
-
-# How many chars of downstream response to return to caller
-FORWARD_BODY_LIMIT = int(os.getenv("FORWARD_BODY_LIMIT", "2000"))
-
-app = FastAPI(title="Inserio Relay", version="1.0.0")
-
-# ----------------------------
-# Helpers
-# ----------------------------
-def norm(v: Any) -> Any:
-    if v is None:
-        return ""
-    if isinstance(v, str):
-        s = v.strip()
-        return "" if s.lower() in ("null", "none") else s
-    return v
-
-def normalize_url(url: str) -> str:
-    """
-    Avoid trailing slash issues. Uvicorn/FastAPI often 307 redirect when trailing
-    slash mismatches the route. We'll send to the canonical version.
-    """
-    return url.rstrip("/")
-
-def compute_fingerprint(payload: Dict[str, Any]) -> str:
-    """
-    MVP fingerprint: stable hash from fields that should represent the same upload.
-    NOTE: This is NOT file-bytes fingerprint. That must be done downstream (163 server)
-    after downloading the attachment bytes from QuickBase.
-    """
-    parts = {
-        "rid": norm(payload.get("rid")),
-        "file_type": norm(payload.get("file_type")),
-        "case_vendor_rid": norm(payload.get("case_vendor_rid")),
-        "case_rid": norm(payload.get("case_rid")),
-        "date_modified": norm(payload.get("date_modified")),
+# -----------------------------------------------------------------------------
+# Configuration (set these as HF Space "Secrets" or "Variables")
+# -----------------------------------------------------------------------------
+WEBHOOK_TOKEN = os.getenv("WEBHOOK_TOKEN", "").strip()
+WEBHOOK_HMAC_SECRET = os.getenv("WEBHOOK_HMAC_SECRET", "").strip()
+WEBHOOK_TOKEN_HEADER = os.getenv("WEBHOOK_TOKEN_HEADER", "X-SA-Token").strip()
+
+MAX_MESSAGES = int(os.getenv("MAX_MESSAGES", "200"))
+
+PERSIST_TO_FILE = os.getenv("PERSIST_TO_FILE", "0").strip() == "1"
+PERSIST_PATH = os.getenv("PERSIST_PATH", "webhook_messages.jsonl").strip()
+
+# -----------------------------------------------------------------------------
+# In-memory state
+# -----------------------------------------------------------------------------
+MESSAGES = deque(maxlen=MAX_MESSAGES)
+TOTAL_COUNT = 0
+LAST_RECEIVED_TS: Optional[float] = None
+
+
+def _now_ts() -> float:
+    return time.time()
+
+
+def _ts_to_str(ts: Optional[float]) -> str:
+    if not ts:
+        return "n/a"
+    return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
+
+
+def _safe_json_dumps(obj: Any) -> str:
+    try:
+        return json.dumps(obj, indent=2, sort_keys=True, ensure_ascii=False)
+    except Exception:
+        return str(obj)
+
+
+def _append_message(payload: Dict[str, Any], meta: Dict[str, Any]) -> None:
+    global TOTAL_COUNT, LAST_RECEIVED_TS
+    TOTAL_COUNT += 1
+    LAST_RECEIVED_TS = _now_ts()
+
+    entry = {
+        "received_at_ts": LAST_RECEIVED_TS,
+        "received_at": _ts_to_str(LAST_RECEIVED_TS),
+        "meta": meta,
+        "payload": payload,
     }
-    raw = json.dumps(parts, sort_keys=True, separators=(",", ":")).encode("utf-8")
-    return hashlib.sha256(raw).hexdigest()
-
-# ----------------------------
-# In-memory dedupe store with TTL
-# fp -> last_seen_epoch
-# ----------------------------
-SEEN: Dict[str, float] = {}
-SEEN_LOCK = threading.Lock()
-
-def seen_prune(now: float) -> None:
-    cutoff = now - SEEN_TTL_SECONDS
-    dead = [fp for fp, ts in SEEN.items() if ts < cutoff]
-    for fp in dead:
-        SEEN.pop(fp, None)
-
-def seen_check_and_set(fp: str) -> bool:
-    """
-    Returns True if already seen (duplicate), else records and returns False.
-    """
-    now = time.time()
-    with SEEN_LOCK:
-        seen_prune(now)
-        if fp in SEEN:
-            SEEN[fp] = now
-            return True
-        SEEN[fp] = now
-        return False
-
-# ----------------------------
-# Flood guard semaphore
-# ----------------------------
-INFLIGHT = threading.BoundedSemaphore(value=max(1, MAX_INFLIGHT))
-
-# ----------------------------
-# Routes
-# ----------------------------
-@app.get("/")
-def root():
+
+    MESSAGES.appendleft(entry)
+
+    if PERSIST_TO_FILE:
+        try:
+            with open(PERSIST_PATH, "a", encoding="utf-8") as f:
+                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
+        except Exception:
+            pass
+
+
+def _verify_token(provided_token: Optional[str]) -> None:
+    if not WEBHOOK_TOKEN:
+        return
+    if not provided_token or provided_token.strip() != WEBHOOK_TOKEN:
+        raise HTTPException(status_code=401, detail="Invalid or missing webhook token.")
+
+
+def _verify_hmac_signature(raw_body: bytes, provided_sig: Optional[str]) -> None:
+    if not WEBHOOK_HMAC_SECRET:
+        return
+    if not provided_sig:
+        raise HTTPException(status_code=401, detail="Missing signature header.")
+    mac = hmac.new(WEBHOOK_HMAC_SECRET.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(mac, provided_sig.strip()):
+        raise HTTPException(status_code=401, detail="Invalid signature.")
+
+
+# -----------------------------------------------------------------------------
+# FastAPI app (webhook listener)
+# -----------------------------------------------------------------------------
+app = FastAPI(title="Webhook Listener", version="1.0.0")
+
+
+@app.get("/health")
+def health():
     return {
         "ok": True,
-        "service": "inserio-relay",
-        "post": ["/webhook", "/wakeup"],
-        "forward_to": normalize_url(WAKEUP_URL),
-        "max_inflight": MAX_INFLIGHT,
-        "seen_ttl_seconds": SEEN_TTL_SECONDS,
+        "total_count": TOTAL_COUNT,
+        "last_received_at": _ts_to_str(LAST_RECEIVED_TS),
     }
 
-@app.get("/health")
-def health():
-    return {"ok": True}
 
+# IMPORTANT: define both with and without trailing slash
+@app.post("/wakeup")
+@app.post("/wakeup/")
 @app.post("/webhook")
-async def webhook(req: Request):
-    # Validate JSON
+@app.post("/webhook/")
+async def webhook(
+    request: Request,
+    x_sa_signature: Optional[str] = Header(default=None),
+    x_sa_token: Optional[str] = Header(default=None),
+):
+    raw_body = await request.body()
+
+    # Token auth (optional)
+    if WEBHOOK_TOKEN:
+        token_value = request.headers.get(WEBHOOK_TOKEN_HEADER) or x_sa_token
+        _verify_token(token_value)
+
+    # HMAC auth (optional)
+    if WEBHOOK_HMAC_SECRET:
+        sig_value = request.headers.get("X-SA-Signature") or x_sa_signature
+        _verify_hmac_signature(raw_body, sig_value)
+
+    # Parse JSON
     try:
-        payload = await req.json()
+        payload = await request.json()
+        if not isinstance(payload, dict):
+            payload = {"_non_object_payload": payload}
     except Exception:
-        raise HTTPException(status_code=400, detail="invalid_json")
+        payload = {"_raw_body": raw_body.decode("utf-8", errors="replace")}
+
+    meta = {
+        "client": request.client.host if request.client else None,
+        "path": str(request.url.path),
+        "headers": {
+            "user-agent": request.headers.get("user-agent"),
+            "content-type": request.headers.get("content-type"),
+            WEBHOOK_TOKEN_HEADER: "***" if request.headers.get(WEBHOOK_TOKEN_HEADER) else None,
+            "x-sa-signature": "***" if request.headers.get("x-sa-signature") else None,
+        },
+    }
+
+    _append_message(payload=payload, meta=meta)
 
-    if not isinstance(payload, dict):
-        raise HTTPException(status_code=400, detail="payload_must_be_object")
+    return JSONResponse(
+        status_code=200,
+        content={
+            "ok": True,
+            "message": "Webhook received.",
+            "total_count": TOTAL_COUNT,
+            "last_received_at": _ts_to_str(LAST_RECEIVED_TS),
+            "echo": payload,
+        },
+    )
+
+
+# -----------------------------------------------------------------------------
+# Gradio UI (simple dashboard)
+# -----------------------------------------------------------------------------
+def ui_status() -> str:
+    return _safe_json_dumps(
+        {
+            "total_count": TOTAL_COUNT,
+            "last_received_at": _ts_to_str(LAST_RECEIVED_TS),
+            "buffer_size": len(MESSAGES),
+            "max_buffer": MAX_MESSAGES,
+            "auth": {
+                "token_enabled": bool(WEBHOOK_TOKEN),
+                "token_header": WEBHOOK_TOKEN_HEADER,
+                "hmac_enabled": bool(WEBHOOK_HMAC_SECRET),
+            },
+            "endpoints": {
+                "health": "/health",
+                "webhook_post": "/wakeup",
+                "webhook_post_alias": "/webhook",
+            },
+        }
+    )
 
-    # 1) processing state default
-    state = norm(payload.get("sa_processing_state"))
-    if not state:
-        payload["sa_processing_state"] = "new"
 
-    # 2) fingerprint + dedupe
-    fp = norm(payload.get("sa_fingerprint"))
-    if not fp:
-        fp = compute_fingerprint(payload)
-        payload["sa_fingerprint"] = fp
+def ui_recent(limit: int = 25) -> str:
+    limit = max(1, min(int(limit), 200))
+    items = list(MESSAGES)[:limit]
+    return _safe_json_dumps(items)
 
-    if seen_check_and_set(fp):
-        # Do not forward duplicates (safe default)
-        payload["sa_processing_state"] = "duplicate"
-        return {"ok": True, "duplicate": True, "fingerprint": fp}
 
-    # 3) forward/relay with flood guard
-    if not INFLIGHT.acquire(blocking=False):
-        # You are getting flooded; do NOT forward.
-        # This is the safety brake Mark warned about.
-        raise HTTPException(status_code=429, detail="too_many_inflight_requests")
+def ui_clear() -> str:
+    global TOTAL_COUNT, LAST_RECEIVED_TS
+    MESSAGES.clear()
+    TOTAL_COUNT = 0
+    LAST_RECEIVED_TS = None
+    return "Cleared in-memory messages."
 
-    try:
-        forward_url = normalize_url(WAKEUP_URL)
 
-        r = requests.post(forward_url, json=payload, timeout=TIMEOUT)
+with gr.Blocks(title="Webhook Listener Dashboard") as demo:
+    gr.Markdown("# Webhook Listener Dashboard")
+    gr.Markdown(
+        "Send a POST request with JSON to `/wakeup` or `/webhook` and it will appear below.\n\n"
+        "This is for payload debugging / confirming delivery."
+    )
 
-        return {
-            "ok": True,
-            "duplicate": False,
-            "fingerprint": fp,
-            "forward_url": forward_url,
-            "forward_status": r.status_code,
-            "forward_body": (r.text or "")[:FORWARD_BODY_LIMIT],
-        }
-    except requests.RequestException as e:
-        raise HTTPException(status_code=502, detail=f"relay_failed: {e}")
-    finally:
-        INFLIGHT.release()
+    with gr.Row():
+        status_box = gr.Code(label="Status", value=ui_status(), language="json")
+        clear_btn = gr.Button("Clear messages", variant="stop")
 
-# Alias route because people keep trying /wakeup on HF
-@app.post("/wakeup")
-async def wakeup_alias(req: Request):
-    return await webhook(req)
+    with gr.Row():
+        limit = gr.Slider(1, 200, value=25, step=1, label="Recent message limit")
+
+    recent_box = gr.Code(label="Recent payloads (newest first)", value=ui_recent(25), language="json")
+
+    def refresh(limit_val: int):
+        return ui_status(), ui_recent(limit_val)
+
+    demo.load(refresh, inputs=[limit], outputs=[status_box, recent_box], every=2)
+    limit.change(refresh, inputs=[limit], outputs=[status_box, recent_box])
+    clear_btn.click(lambda: (ui_clear(), ui_status(), ui_recent(25)), outputs=[recent_box, status_box, recent_box])
+
+app = gr.mount_gradio_app(app, demo, path="/")