mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-07-25 18:19:19 +00:00
* feat(webhooks): HTTP webhooks to trigger agents with HMAC auth and durable callbacks
Add multi-tenant HTTP webhook endpoints for agent triggering:
- /v1/webhooks/message: send messages to channels
- /v1/webhooks/llm: sync/async LLM prompts with HMAC-signed callbacks
- HMAC-256 + bearer token authentication
- Rate limiting and tenant isolation
- Durable callback worker with exponential backoff
- PG 000056 + SQLite schema v25 migrations
- Unit + integration tests, P0 tenant isolation invariants
- Channel media capability helpers for attachment routing
- Comprehensive webhook documentation and i18n strings
* fix(webhooks): address post-review findings (K1-K10)
Comprehensive post-merge fixes addressing 10 blocking code review issues
and 2 adversarial re-audit findings in webhook-agent-triggering feature:
K1: Fix auth middleware tenant context lookup sequencing — move
tenant context injection before authenticate() call to prevent
unscoped secret lookups.
K2: Canonicalize JSON payload format for jsonb compatibility across
PostgreSQL and SQLite — ensure consistent serialization without
whitespace variance to prevent hash mismatches.
K3: Add fail-closed JSON parsing in body hash extraction with explicit
error handling for malformed payloads before HMAC verification.
K4: Fix worker queue wedge by properly draining slot reservations
when delivery succeeds, preventing permanent slot occupancy.
K5: Implement lease-token optimistic concurrency control to prevent
duplicate webhook delivery under high concurrency or retry storms.
K6: Add AES-256-GCM encrypted secret storage at rest with fail-fast
skip-mount when GOCLAW_ENCRYPTION_KEY environment variable unset.
K7: Implement IP allowlist enforcement supporting both CIDR ranges
and exact IP matching with proper X-Forwarded-For parsing.
K8: Add HMAC replay nonce cache (5min expiry, non-blocking async flush)
to prevent request replay attacks on webhook handler.
K9: Fix invariant test schema selection — replace hardcoded assumption
with explicit schema name from config to support multi-schema testing.
K10: Consolidate rate limiters into single shared instance to prevent
per-endpoint limiter starvation and ensure fair rate limiting.
New database migrations:
- 000057: webhook_calls.lease_token for optimistic concurrency
- 000058: webhooks.encrypted_secret_key for AES-256-GCM encryption
New i18n keys: MsgWebhookIPDenied, MsgWebhookEncryptionUnavailable
(with English, Vietnamese, Chinese translations).
New modules:
- internal/http/webhooks_payload.go: JSON canonicalization + body hash
- internal/http/webhooks_nonce.go: Replay nonce cache implementation
- internal/http/webhooks_idempotency_test.go: Integration tests
Documentation updates:
- docs/webhooks.md: §13-14 security sections, encryption flow
- docs/00-architecture-overview.md: webhook subsystem security overview
- docs/codebase-summary.md: webhook security patterns
- docs/project-changelog.md: webhook fixes changelog
Test coverage: 53 webhook tests + 4 P0 invariant tests all passing.
No tenant isolation violations. All security gates enforced.
* docs(journals): webhook feature ship + fix cycle entries
* fix(webhooks): address Claude review findings
- webhooks_llm.go: remove misleading ptr() helper; use &completedAt
pattern for error-path audit rows (matches success path)
- webhooks_auth.go: wrap TouchLastUsed context in WithoutCancel so
background DB update isn't cancelled when HTTP response completes
- store GetByIDUnscoped (PG+SQLite): add NOT revoked / revoked = 0
filter for defense-in-depth parity with GetByHashUnscoped
- webhooks/sign.go: fix package doc — HMAC key is raw plaintext
secret bytes, not hex-decoded SHA-256
- webhooks_admin.go: check auth before encKey guard to avoid leaking
config state to unauthenticated callers
- webhooks_ratelimit.go: two-phase Load→LoadOrStore to avoid per-call
entry allocation on the hot path
* docs(webhooks): fix Sign() function doc to match actual key input
Function-level comment still referenced hex-decoded SecretHash after
the package-level doc was corrected. Align with actual caller usage
([]byte(rawSecret)).
* fix(webhooks): use WithoutCancel for worker execute DB updates
Terminal status writes in execute() ran through the worker main-loop
ctx, which is cancelled on graceful shutdown. If the outbound send
completed but the status update raced with shutdown, the row stayed
in 'running' and got re-delivered via reclaimStale. WithoutCancel
lets the DB write survive worker cancellation while preserving
propagated values (tenant ID, etc.).
* fix(webhooks): move tctx init before panic defer in worker execute
Panic recovery called updateRetry with raw ctx (no tenant ID), making
requireTenantID fail and the reset-to-retry DB write silently drop.
Row stayed 'running' until reclaimStale (~90s delay). Init tctx first
so defer closure captures tenant-scoped non-cancellable context.
* fix(webhooks): pass tenant-scoped tctx to invokeAgent in worker
execute() was passing the raw worker-loop ctx (no tenant ID) to
invokeAgent → router.Get → PGAgentStore.GetByID. GetByID reads
TenantIDFromContext which returned uuid.Nil, making every lookup
return 'agent not found'. Async LLM webhook calls silently failed
all retries. Pass tctx (already tenant-scoped + WithoutCancel) so
the router resolves the agent correctly.
* fix(tests): resolve integration test compile errors
- Remove duplicate contains() in mcp_grant_revoke_test.go (already
defined in tts_gemini_live_test.go)
- Update webhooks_admin_test.go RotateSecret call to match current
5-arg signature (newSecretHash, newPrefix, newEncryptedSecret)
* fix(webhooks): default nil scopes/ip_allowlist to empty slice in Create
PG columns are NOT NULL DEFAULT '{}'. Explicit NULL from pqStringArray(nil)
violated the constraint, breaking TestWebhookAdminCRUD/TenantIsolation.
Coerce nil slices to empty []string{} so the default applies at the DB layer.
* chore: trigger CI on digitopvn/goclaw fork
* ci: retrigger workflows
* fix(webhooks): renumber migrations to 000059-000061 for merge train
238 lines
6.7 KiB
Go
238 lines
6.7 KiB
Go
//go:build sqlite || sqliteonly
|
|
|
|
package sqlitestore
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
|
|
"github.com/nextlevelbuilder/goclaw/internal/store"
|
|
)
|
|
|
|
// compile-time interface assertion
|
|
var _ store.WebhookStore = (*SQLiteWebhookStore)(nil)
|
|
|
|
// SQLiteWebhookStore implements store.WebhookStore backed by SQLite.
|
|
type SQLiteWebhookStore struct {
|
|
db *sql.DB
|
|
}
|
|
|
|
// NewSQLiteWebhookStore creates a new SQLite-backed webhook store.
|
|
func NewSQLiteWebhookStore(db *sql.DB) *SQLiteWebhookStore {
|
|
return &SQLiteWebhookStore{db: db}
|
|
}
|
|
|
|
// scanSQLiteWebhookRow scans a single webhooks row from SQLite into WebhookData.
|
|
// scopes/ip_allowlist are stored as JSON TEXT; bool columns as INTEGER (0/1).
|
|
func scanSQLiteWebhookRow(row interface {
|
|
Scan(dest ...any) error
|
|
}) (*store.WebhookData, error) {
|
|
var w store.WebhookData
|
|
var agentID, channelID *uuid.UUID
|
|
// secret_prefix, created_by are nullable TEXT columns.
|
|
var secretPrefix, createdBy *string
|
|
var scopesRaw, ipAllowlistRaw []byte
|
|
var lastUsedAt nullSqliteTime
|
|
createdAt, updatedAt := scanTimePair()
|
|
|
|
err := row.Scan(
|
|
&w.ID, &w.TenantID, &agentID,
|
|
&w.Name, &w.Kind, &secretPrefix, &w.SecretHash, &w.EncryptedSecret,
|
|
&scopesRaw, &channelID, &w.RateLimitPerMin, &ipAllowlistRaw,
|
|
&w.RequireHMAC, &w.LocalhostOnly, &w.Revoked, &createdBy,
|
|
createdAt, updatedAt, &lastUsedAt,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
w.CreatedAt = createdAt.Time
|
|
w.UpdatedAt = updatedAt.Time
|
|
if lastUsedAt.Valid {
|
|
w.LastUsedAt = &lastUsedAt.Time
|
|
}
|
|
w.AgentID = agentID
|
|
w.ChannelID = channelID
|
|
if secretPrefix != nil {
|
|
w.SecretPrefix = *secretPrefix
|
|
}
|
|
if createdBy != nil {
|
|
w.CreatedBy = *createdBy
|
|
}
|
|
scanJSONStringArray(scopesRaw, &w.Scopes)
|
|
scanJSONStringArray(ipAllowlistRaw, &w.IPAllowlist)
|
|
return &w, nil
|
|
}
|
|
|
|
// sqliteWebhookSelectCols is the canonical SELECT column list for webhooks in SQLite.
|
|
const sqliteWebhookSelectCols = `id, tenant_id, agent_id, name, kind, secret_prefix, secret_hash, encrypted_secret,
|
|
scopes, channel_id, rate_limit_per_min, ip_allowlist,
|
|
require_hmac, localhost_only, revoked, created_by,
|
|
created_at, updated_at, last_used_at`
|
|
|
|
func (s *SQLiteWebhookStore) Create(ctx context.Context, w *store.WebhookData) error {
|
|
_, err := s.db.ExecContext(ctx,
|
|
`INSERT INTO webhooks
|
|
(id, tenant_id, agent_id, name, kind, secret_prefix, secret_hash, encrypted_secret,
|
|
scopes, channel_id, rate_limit_per_min, ip_allowlist,
|
|
require_hmac, localhost_only, revoked, created_by, created_at, updated_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
|
|
w.ID, w.TenantID, nilUUID(w.AgentID),
|
|
w.Name, w.Kind, nilStr(w.SecretPrefix), w.SecretHash, w.EncryptedSecret,
|
|
jsonStringArray(w.Scopes), nilUUID(w.ChannelID), w.RateLimitPerMin, jsonStringArray(w.IPAllowlist),
|
|
w.RequireHMAC, w.LocalhostOnly, w.Revoked,
|
|
nilStr(w.CreatedBy), w.CreatedAt, w.UpdatedAt,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (s *SQLiteWebhookStore) GetByID(ctx context.Context, id uuid.UUID) (*store.WebhookData, error) {
|
|
tid, err := requireTenantID(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
row := s.db.QueryRowContext(ctx,
|
|
`SELECT `+sqliteWebhookSelectCols+`
|
|
FROM webhooks
|
|
WHERE id = ? AND tenant_id = ?`,
|
|
id, tid,
|
|
)
|
|
return scanSQLiteWebhookRow(row)
|
|
}
|
|
|
|
func (s *SQLiteWebhookStore) GetByHash(ctx context.Context, secretHash string) (*store.WebhookData, error) {
|
|
tid, err := requireTenantID(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
row := s.db.QueryRowContext(ctx,
|
|
`SELECT `+sqliteWebhookSelectCols+`
|
|
FROM webhooks
|
|
WHERE secret_hash = ? AND tenant_id = ? AND revoked = 0`,
|
|
secretHash, tid,
|
|
)
|
|
return scanSQLiteWebhookRow(row)
|
|
}
|
|
|
|
// GetByHashUnscoped looks up a webhook by secret_hash without a tenant filter.
|
|
// Intended only for WebhookAuthMiddleware pre-auth resolution before tenant context
|
|
// has been established. Downstream queries must remain tenant-scoped.
|
|
func (s *SQLiteWebhookStore) GetByHashUnscoped(ctx context.Context, secretHash string) (*store.WebhookData, error) {
|
|
row := s.db.QueryRowContext(ctx,
|
|
`SELECT `+sqliteWebhookSelectCols+`
|
|
FROM webhooks
|
|
WHERE secret_hash = ? AND revoked = 0`,
|
|
secretHash,
|
|
)
|
|
return scanSQLiteWebhookRow(row)
|
|
}
|
|
|
|
// GetByIDUnscoped looks up a webhook by UUID without a tenant filter.
|
|
// Intended only for WebhookAuthMiddleware HMAC pre-auth resolution.
|
|
func (s *SQLiteWebhookStore) GetByIDUnscoped(ctx context.Context, id uuid.UUID) (*store.WebhookData, error) {
|
|
row := s.db.QueryRowContext(ctx,
|
|
`SELECT `+sqliteWebhookSelectCols+`
|
|
FROM webhooks
|
|
WHERE id = ? AND revoked = 0`,
|
|
id,
|
|
)
|
|
return scanSQLiteWebhookRow(row)
|
|
}
|
|
|
|
func (s *SQLiteWebhookStore) List(ctx context.Context, f store.WebhookListFilter) ([]store.WebhookData, error) {
|
|
tid, err := requireTenantID(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
q := `SELECT ` + sqliteWebhookSelectCols + ` FROM webhooks WHERE tenant_id = ?`
|
|
args := []any{tid}
|
|
|
|
if f.AgentID != nil {
|
|
q += ` AND agent_id = ?`
|
|
args = append(args, *f.AgentID)
|
|
}
|
|
q += ` ORDER BY created_at DESC`
|
|
|
|
limit := f.Limit
|
|
if limit <= 0 {
|
|
limit = 50
|
|
}
|
|
q += ` LIMIT ? OFFSET ?`
|
|
args = append(args, limit, f.Offset)
|
|
|
|
rows, err := s.db.QueryContext(ctx, q, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var out []store.WebhookData
|
|
for rows.Next() {
|
|
w, scanErr := scanSQLiteWebhookRow(rows)
|
|
if scanErr != nil {
|
|
return nil, scanErr
|
|
}
|
|
out = append(out, *w)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *SQLiteWebhookStore) Update(ctx context.Context, id uuid.UUID, updates map[string]any) error {
|
|
tid, err := requireTenantID(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return execMapUpdateWhereTenant(ctx, s.db, "webhooks", updates, id, tid)
|
|
}
|
|
|
|
func (s *SQLiteWebhookStore) RotateSecret(ctx context.Context, id uuid.UUID, newSecretHash, newPrefix, newEncryptedSecret string) error {
|
|
tid, err := requireTenantID(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
res, err := s.db.ExecContext(ctx,
|
|
`UPDATE webhooks SET secret_hash = ?, secret_prefix = ?, encrypted_secret = ?, updated_at = ?
|
|
WHERE id = ? AND tenant_id = ?`,
|
|
newSecretHash, newPrefix, newEncryptedSecret, time.Now(), id, tid,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n == 0 {
|
|
return sql.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *SQLiteWebhookStore) Revoke(ctx context.Context, id uuid.UUID) error {
|
|
tid, err := requireTenantID(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
res, err := s.db.ExecContext(ctx,
|
|
`UPDATE webhooks SET revoked = 1, updated_at = ?
|
|
WHERE id = ? AND tenant_id = ?`,
|
|
time.Now(), id, tid,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n == 0 {
|
|
return sql.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *SQLiteWebhookStore) TouchLastUsed(ctx context.Context, id uuid.UUID) error {
|
|
_, err := s.db.ExecContext(ctx,
|
|
`UPDATE webhooks SET last_used_at = ? WHERE id = ?`,
|
|
time.Now(), id,
|
|
)
|
|
return err
|
|
}
|