Implement optional single-image intake, Signal ingestion, and multimodal vision analysis
- Add image intake service with format validation (JPEG, PNG, WebP) and EXIF/GPS stripping - Enforce strict single-image rule across Web and Signal attachment channels - Implement token-optimized vision downscaling and JPEG compression - Add IMAGE_CONTEXT pipeline stage with OmniRoute vision routing and resilient failover - Seed and manage versioned idea-image-interpreter prompt in catalog - Update Web UI with responsive image picker, preview chip, and Visual Context tab - Add comprehensive automated test suite in test_image_intake.py - Update README and Labyricorn devlog
This commit is contained in:
@@ -0,0 +1,338 @@
|
||||
"""
|
||||
Signal Gateway Integration API Router
|
||||
Receives, authenticates, and durably processes inbound webhooks and events from Signal Gateway.
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import secrets
|
||||
import base64
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any, List, Tuple
|
||||
from fastapi import APIRouter, Request, HTTPException, status, Header
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from ..config import config
|
||||
from ..database import get_db, next_sequence, get_utc_now
|
||||
from ..queue.worker import job_queue
|
||||
from ..services.image_handler import (
|
||||
validate_and_sanitize_image,
|
||||
save_image_artifact,
|
||||
ImageValidationError
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/api/integrations/signal", tags=["signal_integration"])
|
||||
|
||||
def extract_signal_image_candidate(attachment: Any) -> Optional[Tuple[bytes, str]]:
|
||||
"""
|
||||
Extracts raw bytes and filename from a Signal attachment item.
|
||||
Returns (raw_bytes, filename) or None.
|
||||
"""
|
||||
if isinstance(attachment, dict):
|
||||
# 1. Base64 data payload
|
||||
b64_data = attachment.get("data") or attachment.get("base64") or attachment.get("bytes")
|
||||
filename = attachment.get("filename") or attachment.get("name") or "signal_attachment"
|
||||
if b64_data and isinstance(b64_data, str):
|
||||
try:
|
||||
if "," in b64_data:
|
||||
b64_data = b64_data.split(",", 1)[1]
|
||||
return base64.b64decode(b64_data), filename
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 2. Filesystem path
|
||||
file_path_str = attachment.get("path") or attachment.get("file") or attachment.get("stored_filename")
|
||||
if file_path_str and isinstance(file_path_str, str):
|
||||
p = Path(file_path_str)
|
||||
if p.exists() and p.is_file():
|
||||
try:
|
||||
return p.read_bytes(), p.name
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
elif isinstance(attachment, str):
|
||||
# Could be path or base64
|
||||
p = Path(attachment)
|
||||
if p.exists() and p.is_file():
|
||||
try:
|
||||
return p.read_bytes(), p.name
|
||||
except Exception:
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
raw_str = attachment
|
||||
if "," in raw_str:
|
||||
raw_str = raw_str.split(",", 1)[1]
|
||||
return base64.b64decode(raw_str), "signal_attachment"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
elif isinstance(attachment, (bytes, bytearray)):
|
||||
return bytes(attachment), "signal_attachment"
|
||||
|
||||
return None
|
||||
|
||||
def get_configured_callback_secret() -> str:
|
||||
"""Retrieves configured callback bearer secret dynamically."""
|
||||
secret = config.services.signal_gateway_callback_secret or os.getenv("SIGNAL_GATEWAY_CALLBACK_SECRET", "")
|
||||
if secret:
|
||||
return secret
|
||||
# Try reloading .env in case it was updated after server startup
|
||||
from ..config import _load_env_file, BASE_DIR
|
||||
from pathlib import Path
|
||||
_load_env_file(Path("/root/.env"))
|
||||
_load_env_file(BASE_DIR / ".env")
|
||||
_load_env_file(BASE_DIR / "thinkstorm" / ".env")
|
||||
secret = os.getenv("SIGNAL_GATEWAY_CALLBACK_SECRET", "")
|
||||
if secret:
|
||||
config.services.signal_gateway_callback_secret = secret
|
||||
return secret
|
||||
try:
|
||||
with get_db() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT api_key_raw, config_json FROM service_configurations WHERE id = 'signal_gateway'"
|
||||
).fetchone()
|
||||
if row:
|
||||
conf = json.loads(row["config_json"] or "{}")
|
||||
if "callback_secret" in conf and conf["callback_secret"]:
|
||||
return conf["callback_secret"]
|
||||
except Exception:
|
||||
pass
|
||||
return ""
|
||||
|
||||
def verify_callback_auth(request: Request):
|
||||
"""
|
||||
Enforces constant-time authentication comparison on the Bearer credential.
|
||||
Returns 401 on missing or invalid authentication without leaking details.
|
||||
"""
|
||||
auth_header = request.headers.get("Authorization", "")
|
||||
if not auth_header or not auth_header.startswith("Bearer "):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Unauthorized"
|
||||
)
|
||||
|
||||
provided_token = auth_header[7:].strip()
|
||||
expected_secret = get_configured_callback_secret()
|
||||
|
||||
if not expected_secret or not provided_token:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Unauthorized"
|
||||
)
|
||||
|
||||
# Constant-time comparison to prevent timing attacks
|
||||
if not secrets.compare_digest(provided_token, expected_secret):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Unauthorized"
|
||||
)
|
||||
|
||||
@router.post("/events", status_code=status.HTTP_200_OK)
|
||||
async def handle_signal_events(
|
||||
request: Request,
|
||||
idempotency_key_header: Optional[str] = Header(None, alias="Idempotency-Key")
|
||||
):
|
||||
"""
|
||||
Inbound Signal Gateway Webhook Callback.
|
||||
Handles both test probes and actual normalized Signal messages.
|
||||
"""
|
||||
# 1. Authenticate Request
|
||||
verify_callback_auth(request)
|
||||
|
||||
# 2. Parse Raw JSON Body
|
||||
try:
|
||||
body = await request.json()
|
||||
except Exception:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Malformed JSON body."
|
||||
)
|
||||
|
||||
if not isinstance(body, dict):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid JSON structure."
|
||||
)
|
||||
|
||||
# Check schema_version
|
||||
schema_version = body.get("schema_version")
|
||||
if schema_version != 1:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Unsupported schema_version: {schema_version}. Expected 1."
|
||||
)
|
||||
|
||||
# 3. Handle Callback Test Request
|
||||
event_type = body.get("event_type")
|
||||
if event_type == "signal_gateway.callback_test":
|
||||
# Quick 2xx response without triggering message processing or database side effects
|
||||
return {
|
||||
"accepted": True,
|
||||
"event_type": "signal_gateway.callback_test"
|
||||
}
|
||||
|
||||
# 4. Handle Actual Signal Event
|
||||
event_id = body.get("event_id")
|
||||
if not event_id or not isinstance(event_id, str):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Missing or invalid event_id."
|
||||
)
|
||||
|
||||
# Validate Idempotency-Key header against JSON event_id
|
||||
if not idempotency_key_header or idempotency_key_header.strip() != event_id.strip():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Missing or mismatched Idempotency-Key header."
|
||||
)
|
||||
|
||||
channel = body.get("channel", "signal")
|
||||
if channel != "signal":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Unsupported channel: {channel}."
|
||||
)
|
||||
|
||||
sender_uuid = body.get("sender_uuid")
|
||||
if not sender_uuid or not isinstance(sender_uuid, str):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Missing or invalid sender_uuid."
|
||||
)
|
||||
|
||||
text = body.get("text")
|
||||
if text is None or not isinstance(text, str):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Missing or invalid text field."
|
||||
)
|
||||
|
||||
raw_text = text.strip()
|
||||
if not raw_text:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Message text cannot be empty."
|
||||
)
|
||||
|
||||
sender_number = body.get("sender_number")
|
||||
sender_name = body.get("sender_name")
|
||||
group_id = body.get("group_id")
|
||||
message_id = body.get("message_id") or str(int(request.state.time() if hasattr(request.state, "time") else 0))
|
||||
reply_to = body.get("reply_to")
|
||||
received_at = body.get("received_at") or get_utc_now()
|
||||
now_iso = get_utc_now()
|
||||
|
||||
# 5. Durable Idempotency & Persistence
|
||||
is_duplicate = False
|
||||
idea_id = None
|
||||
|
||||
with get_db() as conn:
|
||||
# Check if event already exists
|
||||
existing = conn.execute(
|
||||
"SELECT event_id, processing_status, idea_id FROM signal_inbound_events WHERE event_id = ?",
|
||||
(event_id,)
|
||||
).fetchone()
|
||||
|
||||
if existing:
|
||||
is_duplicate = True
|
||||
idea_id = existing["idea_id"]
|
||||
else:
|
||||
# Generate new Idea ID
|
||||
idea_id = next_sequence("idea")
|
||||
|
||||
# Process optional Signal attachments (One-Image Rule)
|
||||
accepted_image_meta = None
|
||||
attachments = body.get("attachments") or body.get("attachment") or []
|
||||
if not isinstance(attachments, list):
|
||||
attachments = [attachments]
|
||||
|
||||
for att in attachments:
|
||||
if not att:
|
||||
continue
|
||||
candidate = extract_signal_image_candidate(att)
|
||||
if not candidate:
|
||||
# Non-image or unreadable attachment -> ignore
|
||||
continue
|
||||
raw_img_bytes, filename = candidate
|
||||
if not raw_img_bytes:
|
||||
continue
|
||||
|
||||
try:
|
||||
sanitized_bytes, img_meta = validate_and_sanitize_image(
|
||||
raw_bytes=raw_img_bytes,
|
||||
original_filename=filename,
|
||||
source="signal",
|
||||
idea_id=idea_id
|
||||
)
|
||||
save_image_artifact(idea_id, sanitized_bytes, img_meta)
|
||||
accepted_image_meta = img_meta
|
||||
# Once first valid image is accepted, ignore all subsequent attachments!
|
||||
break
|
||||
except ImageValidationError:
|
||||
# Non-image or corrupted candidate -> ignore and check next attachment
|
||||
continue
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
img_json = json.dumps(accepted_image_meta) if accepted_image_meta else None
|
||||
|
||||
# Insert idea into ThinkStorm ideas table first (so foreign key in signal_inbound_events is satisfied)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO ideas (
|
||||
id, original_text, submitted_at, title, summary,
|
||||
lifecycle_state, processing_state, enrichment_level,
|
||||
source_channel, source_sender_uuid, source_group_id, source_event_id,
|
||||
submission_image, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, 'SUBMITTED', 'QUEUED', 0, 'signal', ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
idea_id,
|
||||
raw_text,
|
||||
received_at,
|
||||
"Processing New Idea (Signal)...",
|
||||
"Analyzing inbound Signal submission text...",
|
||||
sender_uuid,
|
||||
group_id,
|
||||
event_id,
|
||||
img_json,
|
||||
now_iso,
|
||||
now_iso
|
||||
)
|
||||
)
|
||||
|
||||
# Persist inbound event record referencing the new idea
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO signal_inbound_events (
|
||||
event_id, channel, sender_uuid, sender_number, sender_name,
|
||||
group_id, message_id, reply_to, text, received_at,
|
||||
processing_status, idea_id, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'QUEUED', ?, ?)
|
||||
""",
|
||||
(
|
||||
event_id, channel, sender_uuid, sender_number, sender_name,
|
||||
group_id, str(message_id), reply_to, raw_text, received_at,
|
||||
idea_id, now_iso
|
||||
)
|
||||
)
|
||||
|
||||
if is_duplicate:
|
||||
return {
|
||||
"accepted": True,
|
||||
"event_id": event_id,
|
||||
"idea_id": idea_id,
|
||||
"duplicate": True,
|
||||
"status": "already_processed"
|
||||
}
|
||||
|
||||
# 6. Enqueue Background Intake Pipeline
|
||||
await job_queue.enqueue_foreground("intake", idea_id)
|
||||
|
||||
return {
|
||||
"accepted": True,
|
||||
"event_id": event_id,
|
||||
"idea_id": idea_id,
|
||||
"status": "queued"
|
||||
}
|
||||
Reference in New Issue
Block a user