# app/ablo_source.py
import base64
import hashlib
import hmac
import json
import os
import time
from fastapi import APIRouter, HTTPException, Request
from app.services.tasks import get_task, list_tasks, apply_task_operations
router = APIRouter()
def verify_ablo_signature(request: Request, raw_body: bytes) -> None:
api_key = os.environ["ABLO_API_KEY"].encode()
message_id = request.headers.get("webhook-id")
timestamp = request.headers.get("webhook-timestamp")
signature_header = request.headers.get("webhook-signature", "")
if not message_id or not timestamp or not signature_header:
raise HTTPException(status_code=401, detail="missing signature")
signed_at = int(timestamp)
if abs(int(time.time()) - signed_at) > 5 * 60:
raise HTTPException(status_code=401, detail="expired signature")
payload = message_id.encode() + b"." + timestamp.encode() + b"." + raw_body
expected = base64.b64encode(
hmac.new(api_key, payload, hashlib.sha256).digest()
).decode()
presented = [
part.removeprefix("v1,")
for part in signature_header.split()
if part.startswith("v1,")
]
if not any(hmac.compare_digest(expected, value) for value in presented):
raise HTTPException(status_code=401, detail="invalid signature")
@router.post("/api/ablo/source")
async def ablo_source(request: Request):
raw_body = await request.body()
verify_ablo_signature(request, raw_body)
body = json.loads(raw_body)
if body["type"] == "load":
if body["model"] == "tasks":
return {"row": await get_task(body["id"])}
if body["type"] == "list":
if body["model"] == "tasks":
return {"rows": await list_tasks(body.get("query", {}))}
if body["type"] == "commit":
rows = await apply_task_operations(
operations=body["operations"],
client_tx_id=body.get("clientTxId"),
scope=body.get("scope", {}),
)
return {"rows": rows}
raise HTTPException(status_code=400, detail="unsupported request")