Spaces:
Sleeping
Sleeping
File size: 2,858 Bytes
547bb9a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | """
Webhook Error Detector — token config, sync health, duplicate detection.
"""
from __future__ import annotations
import os
import json
from typing import List
from ..base import ErrorDetector, DetectionResult
SYNC_HEALTH_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
'..', '..', 'logs', 'sync_health.json')
class WebhookErrorDetector(ErrorDetector):
page = "webhook"
def run(self) -> List[DetectionResult]:
results = []
# 1. Webhook secret configured
secret = os.getenv("WEBHOOK_SECRET", "")
if not secret or secret in ("changeme", "your-secret", ""):
results.append(self._warn("webhook_secret",
"WEBHOOK_SECRET is default or unset",
"Webhook endpoint is insecure"))
else:
results.append(self._ok("webhook_secret", "WEBHOOK_SECRET is configured"))
# 2. PUBLIC_URL set (needed for Google Form webhook URL generation)
public_url = os.getenv("PUBLIC_URL", "")
space_id = os.getenv("SPACE_ID", "")
if space_id and not public_url:
results.append(self._warn("public_url",
"Running on Hugging Face but PUBLIC_URL is not set",
"Generated webhook URLs may be unreachable"))
elif public_url:
results.append(self._ok("public_url", f"PUBLIC_URL configured: {public_url}"))
else:
results.append(self._ok("public_url", "Not on HF Spaces — PUBLIC_URL not required"))
# 3. Sync health file
try:
health_path = os.path.normpath(SYNC_HEALTH_PATH)
if os.path.exists(health_path):
with open(health_path) as f:
health = json.load(f)
last_error = health.get("last_error")
if last_error:
results.append(self._warn("sync_health",
f"Last sync error: {last_error}",
"Check logs/sync_health.json for details"))
else:
results.append(self._ok("sync_health", "Sync health file shows no errors"))
else:
results.append(self._ok("sync_health", "No sync_health.json yet (normal on first run)"))
except Exception as e:
results.append(self._warn("sync_health", "Could not read sync_health.json", str(e)))
# 4. Admin secret configured
admin_secret = os.getenv("ADMIN_SECRET", "")
if not admin_secret or admin_secret in ("admin", "password", "changeme", ""):
results.append(self._warn("admin_secret",
"ADMIN_SECRET is weak or unset",
"Admin panel is insecure"))
else:
results.append(self._ok("admin_secret", "ADMIN_SECRET is configured"))
return results
|