added scripts feature

Signed-off-by: RonniSkansing <rskansing@gmail.com>
This commit is contained in:
RonniSkansing committed 2026-09-19 11:31:42 +02:00
1 parent a32d2ef4fb
commit f7f73dd531
37 files changed
+4319 -82

No files matched your search

+299
View File
@@ -0,0 +1,299 @@
package script
import (
"bytes"
"compress/flate"
"compress/gzip"
"crypto/hmac"
"crypto/md5"
"crypto/rand"
"crypto/sha1"
"crypto/sha256"
"crypto/sha512"
"encoding/base32"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"hash"
"html"
"io"
"net/url"
"strings"
"github.com/dop251/goja"
"github.com/google/uuid"
)
// registerCodec installs the encode/decode/hash/hmac/jwt/random helpers on the
// VM. These are the data-transform toolkit for scripts: base and binary
// encodings, hashing and keyed HMAC signing, JWT inspection, and secure random
// (nonces, PKCE, OAuth state). Output encoding for hash/hmac/random is selectable
// as "hex" (default), "base64", "base64url" or "base32".
func registerCodec(vm *goja.Runtime) {
vm.Set("encode", map[string]interface{}{
// text and url
"base64": func(s string) string { return base64.StdEncoding.EncodeToString([]byte(s)) },
"base64url": func(s string) string { return base64.RawURLEncoding.EncodeToString([]byte(s)) },
"base32": func(s string) string { return base32.StdEncoding.EncodeToString([]byte(s)) },
"hex": func(s string) string { return hex.EncodeToString([]byte(s)) },
"url": func(s string) string { return url.QueryEscape(s) },
"urlPath": func(s string) string { return url.PathEscape(s) },
"html": func(s string) string { return html.EscapeString(s) },
"json": func(call goja.FunctionCall) goja.Value {
b, err := json.Marshal(call.Argument(0).Export())
if err != nil {
panic(vm.NewGoError(err))
}
return vm.ToValue(string(b))
},
// object -> application/x-www-form-urlencoded
"form": func(call goja.FunctionCall) goja.Value {
return vm.ToValue(formEncode(call.Argument(0)))
},
// compression, output as base64 so it survives as a JS string
"gzip": func(s string) (string, error) { return deflateEncode(s, true) },
"deflate": func(s string) (string, error) {
return deflateEncode(s, false)
},
})
vm.Set("decode", map[string]interface{}{
"base64": func(s string) (string, error) {
b, err := base64.StdEncoding.DecodeString(s)
return string(b), err
},
"base64url": func(s string) (string, error) {
b, err := decodeBase64URL(s)
return string(b), err
},
"base32": func(s string) (string, error) {
b, err := base32.StdEncoding.DecodeString(s)
return string(b), err
},
"hex": func(s string) (string, error) {
b, err := hex.DecodeString(s)
return string(b), err
},
"url": func(s string) (string, error) { return url.QueryUnescape(s) },
"urlPath": func(s string) (string, error) { return url.PathUnescape(s) },
"html": func(s string) string { return html.UnescapeString(s) },
"json": func(s string) (interface{}, error) {
var v interface{}
err := json.Unmarshal([]byte(s), &v)
return v, err
},
// application/x-www-form-urlencoded -> object
"form": func(s string) (map[string]interface{}, error) { return formDecode(s) },
"gzip": func(s string) (string, error) { return inflateDecode(s, true) },
"deflate": func(s string) (string, error) {
return inflateDecode(s, false)
},
})
// hashing: hash.sha256(input, enc?) -> string
vm.Set("hash", map[string]interface{}{
"md5": hashFn(vm, func() hash.Hash { return md5.New() }),
"sha1": hashFn(vm, func() hash.Hash { return sha1.New() }),
"sha256": hashFn(vm, func() hash.Hash { return sha256.New() }),
"sha384": hashFn(vm, func() hash.Hash { return sha512.New384() }),
"sha512": hashFn(vm, func() hash.Hash { return sha512.New() }),
})
// keyed HMAC: hmac.sha256(key, message, enc?) -> string
vm.Set("hmac", map[string]interface{}{
"sha1": hmacFn(vm, func() hash.Hash { return sha1.New() }),
"sha256": hmacFn(vm, func() hash.Hash { return sha256.New() }),
"sha384": hmacFn(vm, func() hash.Hash { return sha512.New384() }),
"sha512": hmacFn(vm, func() hash.Hash { return sha512.New() }),
})
// jwt.decode(token) -> { header, payload, signature } (NOT verified)
vm.Set("jwt", map[string]interface{}{
"decode": func(token string) (map[string]interface{}, error) { return jwtDecode(token) },
})
// random: nonces, PKCE verifiers, OAuth state
vm.Set("random", map[string]interface{}{
"bytes": func(call goja.FunctionCall) goja.Value {
n := int(call.Argument(0).ToInteger())
if n <= 0 || n > 4096 {
panic(vm.NewTypeError("random.bytes: length must be 1..4096"))
}
b := make([]byte, n)
if _, err := rand.Read(b); err != nil {
panic(vm.NewGoError(err))
}
return vm.ToValue(codecOutput(b, argString(call, 1, "hex")))
},
"uuid": func() string { return uuid.NewString() },
})
}
// codecOutput renders bytes in the requested encoding, defaulting to hex.
func codecOutput(b []byte, enc string) string {
switch strings.ToLower(enc) {
case "base64":
return base64.StdEncoding.EncodeToString(b)
case "base64url":
return base64.RawURLEncoding.EncodeToString(b)
case "base32":
return base32.StdEncoding.EncodeToString(b)
default:
return hex.EncodeToString(b)
}
}
// argString reads an optional string argument, falling back to def.
func argString(call goja.FunctionCall, i int, def string) string {
if len(call.Arguments) > i {
v := call.Argument(i)
if !goja.IsUndefined(v) && !goja.IsNull(v) {
return v.String()
}
}
return def
}
func hashFn(vm *goja.Runtime, newH func() hash.Hash) func(goja.FunctionCall) goja.Value {
return func(call goja.FunctionCall) goja.Value {
h := newH()
h.Write([]byte(call.Argument(0).String()))
return vm.ToValue(codecOutput(h.Sum(nil), argString(call, 1, "hex")))
}
}
func hmacFn(vm *goja.Runtime, newH func() hash.Hash) func(goja.FunctionCall) goja.Value {
return func(call goja.FunctionCall) goja.Value {
m := hmac.New(newH, []byte(call.Argument(0).String()))
m.Write([]byte(call.Argument(1).String()))
return vm.ToValue(codecOutput(m.Sum(nil), argString(call, 2, "hex")))
}
}
// decodeBase64URL decodes url-safe base64 with or without padding.
func decodeBase64URL(s string) ([]byte, error) {
s = strings.TrimRight(s, "=")
return base64.RawURLEncoding.DecodeString(s)
}
func deflateEncode(s string, gz bool) (string, error) {
var buf bytes.Buffer
var w io.WriteCloser
var err error
if gz {
w = gzip.NewWriter(&buf)
} else {
w, err = flate.NewWriter(&buf, flate.DefaultCompression)
if err != nil {
return "", err
}
}
if _, err := w.Write([]byte(s)); err != nil {
return "", err
}
if err := w.Close(); err != nil {
return "", err
}
return base64.StdEncoding.EncodeToString(buf.Bytes()), nil
}
func inflateDecode(b64 string, gz bool) (string, error) {
data, err := base64.StdEncoding.DecodeString(b64)
if err != nil {
return "", err
}
var r io.ReadCloser
if gz {
r, err = gzip.NewReader(bytes.NewReader(data))
if err != nil {
return "", err
}
} else {
r = flate.NewReader(bytes.NewReader(data))
}
defer r.Close()
out, err := io.ReadAll(r)
if err != nil {
return "", err
}
return string(out), nil
}
// formEncode turns an object into an application/x-www-form-urlencoded string.
func formEncode(val goja.Value) string {
values := url.Values{}
if m, ok := val.Export().(map[string]interface{}); ok {
for k, v := range m {
switch vv := v.(type) {
case []interface{}:
for _, item := range vv {
values.Add(k, fmt.Sprint(item))
}
default:
values.Set(k, fmt.Sprint(v))
}
}
}
return values.Encode()
}
// formDecode parses a form-encoded string into an object. A key with one value
// becomes a string, a repeated key becomes an array.
func formDecode(s string) (map[string]interface{}, error) {
vals, err := url.ParseQuery(s)
if err != nil {
return nil, err
}
out := map[string]interface{}{}
for k, v := range vals {
if len(v) == 1 {
out[k] = v[0]
continue
}
arr := make([]interface{}, len(v))
for i, x := range v {
arr[i] = x
}
out[k] = arr
}
return out, nil
}
// jwtDecode splits a JWT and returns its header and payload as objects plus the
// raw signature. It does NOT verify the signature: it is for inspecting tokens.
func jwtDecode(token string) (map[string]interface{}, error) {
parts := strings.Split(token, ".")
if len(parts) < 2 {
return nil, fmt.Errorf("invalid jwt: expected header.payload.signature")
}
header, err := decodeB64URLJSON(parts[0])
if err != nil {
return nil, fmt.Errorf("jwt header: %w", err)
}
payload, err := decodeB64URLJSON(parts[1])
if err != nil {
return nil, fmt.Errorf("jwt payload: %w", err)
}
sig := ""
if len(parts) >= 3 {
sig = parts[2]
}
return map[string]interface{}{
"header": header,
"payload": payload,
"signature": sig,
}, nil
}
func decodeB64URLJSON(s string) (interface{}, error) {
b, err := decodeBase64URL(s)
if err != nil {
return nil, err
}
var v interface{}
if err := json.Unmarshal(b, &v); err != nil {
return nil, err
}
return v, nil
}
+579
View File
@@ -0,0 +1,579 @@
// Package script runs admin authored JavaScript when a subscribed campaign
// event fires. It is the scripting counterpart to webhooks: a script receives
// the event payload, can call out over HTTP, transform data, and create a new
// campaign event in the same context.
//
// Trust model: scripts run only when the feature is enabled at the server level
// (config.Script.Enabled), which is the operator acknowledgement that every
// admin is trusted as a server admin. The engine is a goja VM with a small,
// explicit binding set (no require, no filesystem, no process access). Scripts
// are bounded by a wall clock timeout, run one per fresh VM, and execute on a
// bounded worker pool so a burst of events cannot exhaust memory or goroutines.
//
// All bindings are synchronous: http.fetch blocks and returns the response.
// Each run owns its VM on its own worker goroutine, so a blocking call stalls
// only that run.
package script
import (
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"sync"
"time"
"github.com/dop251/goja"
"github.com/phishingclub/phishingclub/data"
"go.uber.org/zap"
"golang.org/x/net/proxy"
)
const (
// DefaultTimeout is the wall clock budget for a single script run.
DefaultTimeout = 10 * time.Second
// DefaultWorkers is the number of scripts that can run at once.
DefaultWorkers = 4
// DefaultQueueSize is how many pending jobs are buffered before new ones
// are dropped.
DefaultQueueSize = 256
// maxResponseBytes caps a fetch response body so a large download cannot
// exhaust memory.
maxResponseBytes = 5 << 20 // 5 MB
// defaultFetchTimeout and maxFetchTimeout bound a single outbound request.
defaultFetchTimeout = 10 * time.Second
maxFetchTimeout = 30 * time.Second
// maxCallStackSize bounds JS recursion depth so a runaway recursive script
// throws a StackOverflowError instead of exhausting the Go stack.
maxCallStackSize = 2000
)
// scriptStopError is thrown by the stop() binding to end a run cleanly. It is
// detected by identity via errors.As (which traverses goja.Exception.Unwrap to
// the wrapped Go error), so a script that catches stop() and then throws a real
// error later is never mistaken for a clean stop.
type scriptStopError struct{}
func (scriptStopError) Error() string { return "script stopped" }
// emittableEvents are the only events a script may create via emitEvent: data the
// script itself authored. Server-detected outcome events (message delivery,
// opens, clicks, page visits, reports, training) are deliberately excluded so a
// script cannot fabricate a campaign's statistics. info() emits the info event
// through its own binding, not emitEvent.
var emittableEvents = map[string]bool{
data.EVENT_CAMPAIGN_RECIPIENT_SUBMITTED_DATA: true,
data.EVENT_CAMPAIGN_RECIPIENT_INFO: true,
}
// EventContext is the payload handed to a script. The caller fills it after
// applying the anonymization guard and the none/basic/full data level, so a
// script never sees more than its configuration allows.
type EventContext struct {
CampaignID string
RecipientID string
Event string
CampaignName string
Email string
Data map[string]interface{}
}
// EmitFunc lets a script create a new campaign event in the same context.
// The caller implements it so the write goes through the native event
// chokepoint (SaveSubmittedData plus anonymization) and does not re-trigger
// scripts.
type EmitFunc func(eventName string, data map[string]interface{}) error
// Job is a single script run.
type Job struct {
ScriptID string
Script string
Event EventContext
Emit EmitFunc
// test, when set, puts the run in capture mode: log/info/emitEvent are
// recorded into it instead of applied, and errors are captured. Set only by
// RunTest.
test *TestResult
}
// Runner executes one job in a fresh goja VM.
type Runner struct {
Logger *zap.SugaredLogger
HTTPClient *http.Client
Timeout time.Duration
}
// run executes a single job. It never returns an error to the caller: a script
// failure is logged, not propagated, because scripts are out of band.
func (r *Runner) run(job Job) {
// a broken script or a panic in a native binding must never take down the
// server, so catch anything that escapes the VM.
defer func() {
if rec := recover(); rec != nil {
msg := fmt.Sprintf("%v", rec)
// a test run captures failures in its own run log; keep the server log
// quiet so the test output is the single source of truth
if job.test == nil {
r.Logger.Errorw("script panicked",
"scriptID", job.ScriptID,
"recover", msg,
)
}
r.reportError(job, "panic", msg)
}
}()
timeout := r.Timeout
if timeout <= 0 {
timeout = DefaultTimeout
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
vm := goja.New()
// bound recursion so a deeply recursive script throws a JS StackOverflowError
// instead of growing the Go stack until the process dies.
vm.SetMaxCallStackSize(maxCallStackSize)
// interrupt the VM when the run times out. Guard with recover so this third
// goroutine can never take down the process, upholding the crash isolation
// invariant even if a future change does more work here.
go func() {
defer func() { _ = recover() }()
<-ctx.Done()
vm.Interrupt(ctx.Err())
}()
r.registerBindings(vm, job, ctx)
// wrap in an IIFE so the script can use return, matching the remote browser
_, err := vm.RunString("(function(){\n" + job.Script + "\n})()")
if err != nil {
// a clean stop() exit is detected by identity: errors.As traverses the
// goja exception to the wrapped Go error. A caught and rethrown error is
// therefore never mistaken for a stop.
var stopErr scriptStopError
if errors.As(err, &stopErr) {
return
}
if _, ok := err.(*goja.InterruptedError); ok {
if job.test == nil {
r.Logger.Warnw("script timed out", "scriptID", job.ScriptID)
}
r.reportError(job, "timeout", "script exceeded its time budget")
return
}
// a native call cancelled by the run budget (for example http.fetch
// blocked when the deadline passed) surfaces as a thrown exception, not
// an InterruptedError; classify it as a timeout too
if ctx.Err() == context.DeadlineExceeded {
if job.test == nil {
r.Logger.Warnw("script timed out", "scriptID", job.ScriptID)
}
r.reportError(job, "timeout", "script exceeded its time budget")
return
}
// a test run surfaces the error in its own run log; don't also spam the
// server log
if job.test == nil {
r.Logger.Errorw("script error",
"scriptID", job.ScriptID,
"error", err.Error(),
)
}
r.reportError(job, "exception", err.Error())
}
}
// reportError records an uncaught script failure as a campaign info event so
// it is visible beyond the server logs. It goes through the same event funnel as
// emitEvent, so the detail follows the campaign's data-retention and anonymity
// rules (the full message is always in the server logs). Best effort: a failure
// to record is only logged.
func (r *Runner) reportError(job Job, phase, message string) {
if job.test != nil {
job.test.setError(phase, message)
return
}
if job.Emit == nil {
return
}
// reportError is called from the deferred panic recover; a panic in Emit (a DB
// write plus webhook fan out) would escape that recover and kill the worker, so
// isolate it here.
defer func() {
if rec := recover(); rec != nil {
r.Logger.Errorw("panic recording script error event",
"scriptID", job.ScriptID,
"recover", fmt.Sprintf("%v", rec),
)
}
}()
err := job.Emit(data.EVENT_CAMPAIGN_RECIPIENT_INFO, map[string]interface{}{
"source": "script",
"level": "error",
"scriptId": job.ScriptID,
"phase": phase, // "exception" | "timeout" | "panic"
"error": message,
})
if err != nil {
r.Logger.Errorw("failed to record script error event",
"scriptID", job.ScriptID,
"error", err,
)
}
}
// registerBindings installs the script API on the VM. This is the entire
// capability surface: an event payload, outbound http, encode/decode helpers,
// log, emitEvent and stop. No require, no filesystem, no process access.
func (r *Runner) registerBindings(vm *goja.Runtime, job Job, ctx context.Context) {
// event payload, already filtered by the caller
vm.Set("event", map[string]interface{}{
"name": job.Event.Event,
"campaignId": job.Event.CampaignID,
"recipientId": job.Event.RecipientID,
"campaignName": job.Event.CampaignName,
"email": job.Event.Email,
"data": job.Event.Data,
})
vm.Set("stop", func(call goja.FunctionCall) goja.Value {
panic(vm.NewGoError(scriptStopError{}))
})
vm.Set("log", func(call goja.FunctionCall) goja.Value {
msg := call.Argument(0).String()
var extra interface{}
if len(call.Arguments) > 1 && !goja.IsUndefined(call.Argument(1)) && !goja.IsNull(call.Argument(1)) {
extra = call.Argument(1).Export()
}
if job.test != nil {
job.test.addLog(msg, extra)
return goja.Undefined()
}
if extra != nil {
r.Logger.Infow("script log", "scriptID", job.ScriptID, "message", msg, "data", extra)
} else {
r.Logger.Infow("script log", "scriptID", job.ScriptID, "message", msg)
}
return goja.Undefined()
})
// info records a campaign_recipient_info event, visible in the campaign
// timeline. Unlike log (server logs only) this is observable in the app; the
// detail follows the campaign's data-retention and anonymity rules.
vm.Set("info", func(call goja.FunctionCall) goja.Value {
msg := call.Argument(0).String()
// the optional second argument is extra structured data
var extra map[string]interface{}
if len(call.Arguments) > 1 {
if m, ok := call.Argument(1).Export().(map[string]interface{}); ok {
extra = m
}
}
if job.test != nil {
job.test.addInfo(msg, extra)
return goja.Undefined()
}
if job.Emit == nil {
return goja.Undefined()
}
payload := map[string]interface{}{
"source": "script",
"level": "info",
"message": msg,
}
for k, v := range extra {
payload[k] = v
}
if err := job.Emit(data.EVENT_CAMPAIGN_RECIPIENT_INFO, payload); err != nil {
panic(vm.NewGoError(err))
}
return goja.Undefined()
})
vm.Set("http", map[string]interface{}{
"fetch": r.makeFetch(vm, ctx, job),
})
// encode/decode/hash/hmac/jwt/random data transform toolkit
registerCodec(vm)
vm.Set("emitEvent", func(call goja.FunctionCall) goja.Value {
name := call.Argument(0).String()
// a script may only author its own data events; it must not be able to
// fabricate server-detected outcomes (opens, clicks, reports, delivery,
// training) and skew a campaign's statistics.
if !emittableEvents[name] {
panic(vm.NewTypeError(fmt.Sprintf(
"emitEvent: %q cannot be created by a script (allowed: %s, %s)",
name,
data.EVENT_CAMPAIGN_RECIPIENT_SUBMITTED_DATA,
data.EVENT_CAMPAIGN_RECIPIENT_INFO,
)))
}
var d map[string]interface{}
if exp := call.Argument(1).Export(); exp != nil {
if m, ok := exp.(map[string]interface{}); ok {
d = m
}
}
if job.test != nil {
job.test.addEvent(name, d)
return goja.Undefined()
}
if job.Emit == nil {
panic(vm.NewTypeError("emitEvent is not available for this event"))
}
if err := job.Emit(name, d); err != nil {
panic(vm.NewGoError(err))
}
return goja.Undefined()
})
}
// makeFetch builds the synchronous http.fetch binding.
func (r *Runner) makeFetch(vm *goja.Runtime, ctx context.Context, job Job) func(goja.FunctionCall) goja.Value {
return func(call goja.FunctionCall) goja.Value {
urlStr := call.Argument(0).String()
method := "GET"
var bodyReader io.Reader
headers := map[string]string{}
fetchTimeout := defaultFetchTimeout
proxyStr := ""
if exp := call.Argument(1).Export(); exp != nil {
if opts, ok := exp.(map[string]interface{}); ok {
if v, ok := opts["method"].(string); ok && v != "" {
method = strings.ToUpper(v)
}
if v, ok := opts["body"].(string); ok {
bodyReader = strings.NewReader(v)
}
if v, ok := opts["proxy"].(string); ok {
proxyStr = v
}
if h, ok := opts["headers"].(map[string]interface{}); ok {
for k, val := range h {
headers[k] = fmt.Sprint(val)
}
}
if ms := coerceMillis(opts["timeoutMs"]); ms > 0 {
fetchTimeout = time.Duration(ms) * time.Millisecond
if fetchTimeout > maxFetchTimeout {
fetchTimeout = maxFetchTimeout
}
}
}
}
reqCtx, cancel := context.WithTimeout(ctx, fetchTimeout)
defer cancel()
req, err := http.NewRequestWithContext(reqCtx, method, urlStr, bodyReader)
if err != nil {
panic(vm.NewTypeError(err.Error()))
}
for k, v := range headers {
req.Header.Set(k, v)
}
// route through a proxy for this request when the script asks for one
client := r.HTTPClient
if proxyStr != "" {
pc, perr := proxyClient(proxyStr)
if perr != nil {
panic(vm.NewTypeError("http.fetch: " + perr.Error()))
}
client = pc
}
resp, err := client.Do(req)
if err != nil {
if job.test != nil {
job.test.addFetchErr(method, urlStr, err.Error())
}
panic(vm.NewGoError(err))
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
if err != nil {
if job.test != nil {
job.test.addFetchErr(method, urlStr, err.Error())
}
panic(vm.NewGoError(err))
}
if job.test != nil {
job.test.addFetchOK(method, urlStr, resp.StatusCode)
}
respHeaders := map[string]interface{}{}
for k := range resp.Header {
respHeaders[k] = resp.Header.Get(k)
}
return vm.ToValue(map[string]interface{}{
"status": resp.StatusCode,
"headers": respHeaders,
"body": string(body),
})
}
}
// proxyClient builds an http.Client that routes a request through the given
// proxy. Supports http/https and socks5. Keep-alives are disabled so a per
// request proxy client does not accumulate idle connections.
func proxyClient(proxyStr string) (*http.Client, error) {
u, err := url.Parse(proxyStr)
if err != nil {
return nil, fmt.Errorf("invalid proxy url: %w", err)
}
switch strings.ToLower(u.Scheme) {
case "http", "https":
return &http.Client{
Timeout: maxFetchTimeout,
Transport: &http.Transport{
Proxy: http.ProxyURL(u),
DisableKeepAlives: true,
},
}, nil
case "socks5", "socks5h":
dialer, err := proxy.FromURL(u, proxy.Direct)
if err != nil {
return nil, err
}
tr := &http.Transport{DisableKeepAlives: true}
if cd, ok := dialer.(proxy.ContextDialer); ok {
tr.DialContext = cd.DialContext
} else {
tr.DialContext = func(_ context.Context, network, addr string) (net.Conn, error) {
return dialer.Dial(network, addr)
}
}
return &http.Client{Timeout: maxFetchTimeout, Transport: tr}, nil
default:
return nil, fmt.Errorf("unsupported proxy scheme %q (use http, https or socks5)", u.Scheme)
}
}
// coerceMillis reads a JS number that goja may export as int64 or float64.
func coerceMillis(v interface{}) int64 {
switch n := v.(type) {
case int64:
return n
case float64:
return int64(n)
default:
return 0
}
}
// Dispatcher runs jobs on a bounded worker pool.
type Dispatcher struct {
jobs chan Job
runner *Runner
logger *zap.SugaredLogger
workers int
wg sync.WaitGroup
stopOnce sync.Once
// mu guards closed so Enqueue never sends on a channel Stop has closed.
mu sync.RWMutex
closed bool
}
// NewDispatcher builds a dispatcher. Zero values fall back to the defaults.
func NewDispatcher(
logger *zap.SugaredLogger,
workers int,
queueSize int,
timeout time.Duration,
) *Dispatcher {
if workers <= 0 {
workers = DefaultWorkers
}
if queueSize <= 0 {
queueSize = DefaultQueueSize
}
if timeout <= 0 {
timeout = DefaultTimeout
}
return &Dispatcher{
jobs: make(chan Job, queueSize),
runner: &Runner{
Logger: logger,
HTTPClient: &http.Client{Timeout: maxFetchTimeout},
Timeout: timeout,
},
logger: logger,
workers: workers,
}
}
// Start launches the worker goroutines.
func (d *Dispatcher) Start() {
for i := 0; i < d.workers; i++ {
d.wg.Add(1)
go func() {
defer d.wg.Done()
for job := range d.jobs {
d.runOne(job)
}
}()
}
}
// runOne isolates a single job. run() has its own recover, but the error
// reporting path runs inside that recover, so a panic there could still escape.
// This last line of defence guarantees one bad job can never crash the worker
// goroutine (and with it the process).
func (d *Dispatcher) runOne(job Job) {
defer func() {
if rec := recover(); rec != nil {
d.logger.Errorw("script worker recovered from panic",
"scriptID", job.ScriptID,
"recover", fmt.Sprintf("%v", rec),
)
}
}()
d.runner.run(job)
}
// Enqueue submits a job. It returns false when the queue is full, in which case
// the job is dropped rather than blocking the caller on the event capture path.
func (d *Dispatcher) Enqueue(job Job) bool {
// hold the read lock across the send so Stop cannot close the channel
// between the closed check and the send
d.mu.RLock()
defer d.mu.RUnlock()
if d.closed {
return false
}
select {
case d.jobs <- job:
return true
default:
return false
}
}
// Stop closes the queue and waits for in flight jobs to finish.
func (d *Dispatcher) Stop() {
d.stopOnce.Do(func() {
// take the write lock so no Enqueue is mid send when the channel closes
d.mu.Lock()
d.closed = true
close(d.jobs)
d.mu.Unlock()
})
d.wg.Wait()
}
+431
View File
@@ -0,0 +1,431 @@
package script
import (
"fmt"
"io"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"go.uber.org/zap"
)
// newTestRunner builds a Runner with a no-op logger for tests.
func newTestRunner() *Runner {
return &Runner{
Logger: zap.NewNop().Sugar(),
HTTPClient: &http.Client{Timeout: maxFetchTimeout},
Timeout: 5 * time.Second,
}
}
// TestScriptFetchEncodeEmit proves the engine can read the event payload, call
// out over http.fetch, use encode/decode, and write back through emitEvent.
func TestScriptFetchEncodeEmit(t *testing.T) {
// a server that echoes the request body back
var gotBody string
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
buf, _ := io.ReadAll(r.Body)
gotBody = string(buf)
w.WriteHeader(200)
_, _ = w.Write([]byte(`{"ok":true,"token":"abc123"}`))
}))
defer srv.Close()
var mu sync.Mutex
emitted := map[string]map[string]interface{}{}
emit := func(name string, data map[string]interface{}) error {
mu.Lock()
defer mu.Unlock()
emitted[name] = data
return nil
}
script := `
log('starting', { event: event.name });
// send the campaign name to the echo server, base64 encoded
var res = http.fetch('` + srv.URL + `', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: encode.json({ name: event.campaignName, b: encode.base64('hi') })
});
if (res.status !== 200) { throw new Error('bad status ' + res.status); }
var parsed = decode.json(res.body);
// write the returned token back as a submit event
emitEvent('campaign_recipient_submitted_data', { token: parsed.token });
`
r := newTestRunner()
r.run(Job{
ScriptID: "test-1",
Script: script,
Event: EventContext{
CampaignID: "c1",
Event: "campaign_recipient_page_visited",
CampaignName: "Q3 Phish",
},
Emit: emit,
})
mu.Lock()
defer mu.Unlock()
// the server must have received the base64 of "hi" = "aGk="
if gotBody == "" {
t.Fatalf("server received no body")
}
if want := "aGk="; !contains(gotBody, want) {
t.Fatalf("request body %q did not contain %q", gotBody, want)
}
// emitEvent must have fired with the decoded token
ev, ok := emitted["campaign_recipient_submitted_data"]
if !ok {
t.Fatalf("emitEvent was not called; emitted=%v", emitted)
}
if ev["token"] != "abc123" {
t.Fatalf("emit token = %v, want abc123", ev["token"])
}
}
// TestCodec exercises the encode/decode/hash/hmac/jwt/random toolkit.
func TestCodec(t *testing.T) {
var mu sync.Mutex
var got map[string]interface{}
emit := func(name string, data map[string]interface{}) error {
mu.Lock()
defer mu.Unlock()
got = data
return nil
}
script := `
var r = {};
r.b64url = encode.base64url('hi');
r.b64urlRound = decode.base64url(encode.base64url('héllo'));
r.sha256abc = hash.sha256('abc');
r.hmacLen = hmac.sha256('key', 'msg').length;
r.hmacDet = hmac.sha256('key', 'msg') === hmac.sha256('key', 'msg');
r.hmacB64Len = hmac.sha256('key', 'msg', 'base64').length > 0;
r.gzipRound = decode.gzip(encode.gzip('the quick brown fox'));
r.deflateRound = decode.deflate(encode.deflate('deflate me'));
r.form = encode.form({ a: '1', b: 'two' });
var f = decode.form('a=1&b=two&b=three');
r.formA = f.a;
r.formBArr = Array.isArray(f.b) ? f.b.length : 0;
var tok = encode.base64url('{"alg":"HS256"}') + '.' + encode.base64url('{"sub":"123"}') + '.xxx';
var jd = jwt.decode(tok);
r.jwtSub = jd.payload.sub;
r.uuidLen = random.uuid().length;
r.randHexLen = random.bytes(16, 'hex').length;
r.b32 = encode.base32('foo');
r.htmlRound = decode.html(encode.html('<b>&"x"</b>'));
emitEvent('campaign_recipient_submitted_data', r);
`
r := newTestRunner()
r.run(Job{ScriptID: "codec", Script: script, Emit: emit})
mu.Lock()
defer mu.Unlock()
if got == nil {
t.Fatalf("codec script emitted nothing")
}
want := map[string]string{
"b64url": "aGk",
"b64urlRound": "héllo",
"sha256abc": "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
"hmacLen": "64",
"hmacDet": "true",
"hmacB64Len": "true",
"gzipRound": "the quick brown fox",
"deflateRound": "deflate me",
"form": "a=1&b=two",
"formA": "1",
"formBArr": "2",
"jwtSub": "123",
"uuidLen": "36",
"randHexLen": "32",
"b32": "MZXW6===",
"htmlRound": `<b>&"x"</b>`,
}
for k, w := range want {
if g := fmt.Sprint(got[k]); g != w {
t.Errorf("codec %s = %q, want %q", k, g, w)
}
}
}
// TestUncaughtErrorRecordsEvent proves an uncaught exception is captured as an
// info event (level=error) through the Emit funnel, not just the server logs.
func TestUncaughtErrorRecordsEvent(t *testing.T) {
var mu sync.Mutex
var name string
var payload map[string]interface{}
emit := func(n string, d map[string]interface{}) error {
mu.Lock()
defer mu.Unlock()
name = n
payload = d
return nil
}
r := newTestRunner()
r.run(Job{ScriptID: "boom", Script: `throw new Error('kaboom');`, Emit: emit})
mu.Lock()
defer mu.Unlock()
if name != "campaign_recipient_info" {
t.Fatalf("error event name = %q, want campaign_recipient_info", name)
}
if payload["level"] != "error" || payload["phase"] != "exception" {
t.Fatalf("error payload = %v", payload)
}
if s, _ := payload["error"].(string); s == "" || !contains(s, "kaboom") {
t.Fatalf("error detail did not include the message: %v", payload["error"])
}
}
// TestInfoBinding proves info() records a visible info event distinct from log().
func TestInfoBinding(t *testing.T) {
var mu sync.Mutex
var name string
var payload map[string]interface{}
emit := func(n string, d map[string]interface{}) error {
mu.Lock()
defer mu.Unlock()
name = n
payload = d
return nil
}
r := newTestRunner()
r.run(Job{ScriptID: "info", Script: `info('hello', { step: 3 });`, Emit: emit})
mu.Lock()
defer mu.Unlock()
if name != "campaign_recipient_info" {
t.Fatalf("info event name = %q", name)
}
if payload["message"] != "hello" || payload["level"] != "info" {
t.Fatalf("info payload = %v", payload)
}
if fmt.Sprint(payload["step"]) != "3" {
t.Fatalf("info extra field lost: %v", payload["step"])
}
}
// TestRunTestCapture proves RunTest records log/info/emitEvent instead of
// applying them, and captures an uncaught error.
func TestRunTestCapture(t *testing.T) {
r := newTestRunner()
res := r.RunTest(`
log('a log', { x: 1 });
info('an info');
emitEvent('campaign_recipient_submitted_data', { token: hmac.sha256('k', 'm') });
`, EventContext{Event: "campaign_recipient_submitted_data", CampaignName: "Demo"})
if !res.OK {
t.Fatalf("expected ok, got error %q", res.Error)
}
// ordered run log: log, info, event(emitEvent), done
types := []string{}
for _, e := range res.Entries {
types = append(types, e.Type)
}
if len(res.Entries) != 4 ||
res.Entries[0].Type != "log" || res.Entries[0].Message != "a log" ||
res.Entries[1].Type != "info" ||
res.Entries[2].Type != "event" || res.Entries[2].Key != "campaign_recipient_submitted_data" ||
res.Entries[3].Type != "done" {
t.Fatalf("entries = %+v (types %v)", res.Entries, types)
}
if v, _ := res.Entries[2].Value.(map[string]interface{}); v["token"] == "" || v["token"] == nil {
t.Fatalf("emitEvent data not captured: %+v", res.Entries[2].Value)
}
// an uncaught throw is captured, not panicked
bad := r.RunTest(`throw new Error('nope');`, EventContext{})
if bad.OK || bad.ErrorPhase != "exception" || bad.Error == "" {
t.Fatalf("expected captured exception, got %+v", bad)
}
}
// TestEmitPanicDoesNotCrash proves that a panic in the Emit callback (e.g. a DB
// failure) during the error-reporting path is contained, not propagated. A
// script throws (triggering reportError), and Emit panics; the run must return.
func TestEmitPanicDoesNotCrash(t *testing.T) {
panicEmit := func(string, map[string]interface{}) error {
panic("emit blew up")
}
r := newTestRunner()
done := make(chan struct{})
go func() {
// throw -> reportError -> Emit panics; must not escape run()
r.run(Job{ScriptID: "emit-panic", Script: `throw new Error('x');`, Emit: panicEmit})
close(done)
}()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatalf("run() did not return after a panicking Emit")
}
}
// TestWorkerSurvivesPanickingJob proves the dispatcher keeps working after a job
// whose Emit panics: a later job still runs.
func TestWorkerSurvivesPanickingJob(t *testing.T) {
d := NewDispatcher(zap.NewNop().Sugar(), 1, 8, time.Second)
d.Start()
defer d.Stop()
panicEmit := func(string, map[string]interface{}) error { panic("boom") }
d.Enqueue(Job{ScriptID: "bad", Script: `throw new Error('x');`, Emit: panicEmit})
ran := make(chan struct{}, 1)
okEmit := func(string, map[string]interface{}) error { ran <- struct{}{}; return nil }
d.Enqueue(Job{ScriptID: "good", Script: `emitEvent('x', {});`, Emit: okEmit})
select {
case <-ran:
case <-time.After(3 * time.Second):
t.Fatalf("worker died after a panicking job; later job never ran")
}
}
// TestDeepRecursionBounded proves runaway recursion throws instead of crashing.
func TestDeepRecursionBounded(t *testing.T) {
r := newTestRunner()
done := make(chan struct{})
go func() {
r.run(Job{ScriptID: "recurse", Script: `function f(){ return f(); } f();`})
close(done)
}()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatalf("deep recursion was not bounded")
}
}
// TestTestCaptureCapped proves a looping test script cannot grow the run log
// without bound.
func TestTestCaptureCapped(t *testing.T) {
r := newTestRunner()
res := r.RunTest(`for (var i = 0; i < 100000; i++) { log('x'); }`, EventContext{})
if len(res.Entries) > maxTestEntries+2 {
t.Fatalf("captured %d entries, want <= %d", len(res.Entries), maxTestEntries+2)
}
}
// TestEmitEventAllowlist proves a script may only emit its own data events and
// cannot fabricate a server-detected outcome (e.g. a report).
func TestEmitEventAllowlist(t *testing.T) {
bad := newTestRunner().RunTest(`emitEvent('campaign_recipient_reported', {});`, EventContext{})
if bad.OK {
t.Fatalf("emitEvent('campaign_recipient_reported') should be rejected")
}
good := newTestRunner().RunTest(`emitEvent('campaign_recipient_submitted_data', { a: 1 });`, EventContext{})
if !good.OK {
t.Fatalf("emitEvent('campaign_recipient_submitted_data') should be allowed: %s", good.Error)
}
}
// TestScriptTimeout proves a runaway script is interrupted, not hung.
func TestScriptTimeout(t *testing.T) {
r := newTestRunner()
r.Timeout = 300 * time.Millisecond
done := make(chan struct{})
go func() {
r.run(Job{ScriptID: "loop", Script: `while (true) {}`})
close(done)
}()
select {
case <-done:
// interrupted and returned
case <-time.After(3 * time.Second):
t.Fatalf("runaway script was not interrupted")
}
}
// TestScriptStop proves stop() exits cleanly and code after it does not run.
func TestScriptStop(t *testing.T) {
var called bool
emit := func(name string, data map[string]interface{}) error {
called = true
return nil
}
r := newTestRunner()
r.run(Job{
ScriptID: "stop",
Script: `stop(); emitEvent('should_not_fire', {});`,
Emit: emit,
})
if called {
t.Fatalf("code after stop() ran")
}
}
// TestScriptStopCaughtDoesNotSwallowLaterError proves stop() is detected by
// identity, not a latched flag: a script that catches stop() and then throws a
// real error still has that error reported instead of being treated as a clean
// stop.
func TestScriptStopCaughtDoesNotSwallowLaterError(t *testing.T) {
var reported bool
var phase string
emit := func(name string, data map[string]interface{}) error {
if lvl, _ := data["level"].(string); lvl == "error" {
reported = true
phase, _ = data["phase"].(string)
}
return nil
}
r := newTestRunner()
r.run(Job{
ScriptID: "caught-stop",
Script: `try { stop(); } catch (e) {} throw new Error('boom');`,
Emit: emit,
})
if !reported {
t.Fatalf("a real error after a caught stop() was swallowed")
}
if phase != "exception" {
t.Fatalf("expected phase 'exception', got %q", phase)
}
}
// TestDispatcherDropsWhenFull proves the pool bounds work: a full queue drops.
func TestDispatcherDropsWhenFull(t *testing.T) {
// one worker, tiny queue, a slow script so the queue fills
d := NewDispatcher(zap.NewNop().Sugar(), 1, 1, time.Second)
d.Start()
defer d.Stop()
block := make(chan struct{})
// fill the single worker with a job that blocks via a slow emit
slow := func(string, map[string]interface{}) error {
<-block
return nil
}
// occupy the worker
d.Enqueue(Job{Script: `emitEvent('x', {});`, Emit: slow})
time.Sleep(50 * time.Millisecond)
// fill the queue (depth 1)
d.Enqueue(Job{Script: `1;`})
// next enqueue should be dropped
dropped := !d.Enqueue(Job{Script: `1;`})
close(block)
if !dropped {
t.Fatalf("expected a full queue to drop the job")
}
}
func contains(s, sub string) bool {
for i := 0; i+len(sub) <= len(s); i++ {
if s[i:i+len(sub)] == sub {
return true
}
}
return false
}
+102
View File
@@ -0,0 +1,102 @@
package script
import (
"fmt"
"sync"
"time"
)
// TestEntry is one line in the run log. It mirrors the remote browser editor's
// runLog entry shape so the editor renders both the same way.
// - type "log": Message (+ optional Data)
// - type "info": Message
// - type "event": Key (event name) + Value (event data) — an emitEvent call
// - type "error": Message
// - type "done": end marker
type TestEntry struct {
Type string `json:"type"`
Time string `json:"time"`
Message string `json:"message,omitempty"`
Data interface{} `json:"data,omitempty"`
Key string `json:"key,omitempty"`
Value interface{} `json:"value,omitempty"`
}
// TestResult is the captured output of a test run: an ordered run log plus the
// terminal error, if any.
type TestResult struct {
OK bool `json:"ok"`
DurationMs int64 `json:"durationMs"`
Entries []TestEntry `json:"entries"`
Error string `json:"error,omitempty"`
ErrorPhase string `json:"errorPhase,omitempty"`
mu sync.Mutex
}
// maxTestEntries caps the captured run log so a script that logs in a tight loop
// cannot grow the response unboundedly and exhaust memory in the request handler.
const maxTestEntries = 2000
func (t *TestResult) add(e TestEntry) {
e.Time = time.Now().UTC().Format(time.RFC3339Nano)
t.mu.Lock()
defer t.mu.Unlock()
if len(t.Entries) >= maxTestEntries {
if len(t.Entries) == maxTestEntries {
t.Entries = append(t.Entries, TestEntry{
Type: "log",
Time: e.Time,
Message: "… output truncated (too many entries)",
})
}
return
}
t.Entries = append(t.Entries, e)
}
func (t *TestResult) addLog(msg string, data interface{}) {
t.add(TestEntry{Type: "log", Message: msg, Data: data})
}
func (t *TestResult) addInfo(msg string, data interface{}) {
t.add(TestEntry{Type: "info", Message: msg, Data: data})
}
func (t *TestResult) addEvent(name string, data map[string]interface{}) {
t.add(TestEntry{Type: "event", Key: name, Value: data})
}
func (t *TestResult) addFetchOK(method, url string, status int) {
t.add(TestEntry{Type: "log", Message: fmt.Sprintf("%s %s → %d", method, url, status)})
}
func (t *TestResult) addFetchErr(method, url, errMsg string) {
t.add(TestEntry{Type: "log", Message: fmt.Sprintf("%s %s ✗ %s", method, url, errMsg)})
}
func (t *TestResult) setError(phase, msg string) {
t.mu.Lock()
if t.Error == "" {
t.Error = msg
t.ErrorPhase = phase
}
t.mu.Unlock()
t.add(TestEntry{Type: "error", Message: msg})
}
// RunTest runs a script against a simulated event and captures what it does as
// an ordered run log, without touching the campaign: log, info and emitEvent are
// recorded rather than applied, and no webhooks fire. http.fetch runs for real.
// The Runner's Timeout bounds the run.
func (r *Runner) RunTest(script string, event EventContext) *TestResult {
res := &TestResult{Entries: []TestEntry{}}
start := time.Now()
r.run(Job{ScriptID: "test", Script: script, Event: event, test: res})
res.DurationMs = time.Since(start).Milliseconds()
res.OK = res.Error == ""
if res.OK {
res.add(TestEntry{Type: "done"})
}
return res
}