pushflo-go

A Go SDK for PushFlo, plus an end-to-end encryption envelope layer so payloads stay opaque to the relay.

This first pass ships two things:

• pushflo — a subscriber client (WebSocket, auto-reconnect, heartbeat) mirroring the ergonomics of the official @pushflodev/sdk client.
• pushflo/envelope — a transport-agnostic package that seals a payload to a recipient's key and signs it, so PushFlo (or any pub/sub relay) only ever moves ciphertext it cannot read, forge, or replay.

Status: v0. The wire protocol (WebSocket frames, auth, heartbeat, REST publish) is confirmed against the official @pushflodev/sdk v1.1.0 source and verified against the live service by the integration test suite. The public API may still evolve before v1.

Install

Requires Go 1.25 or newer.

go get github.com/PushFlo/pushflo-go

Subscriber client

ctx := context.Background()
client, err := pushflo.NewClient(ctx, pushflo.ClientOptions{
    PublishKey: "pub_xxxxxxxxxxxxx",
})
if err != nil { log.Fatal(err) }
defer client.Destroy()

client.OnConnectionChange(func(s pushflo.ConnectionState) {
    log.Println("connection:", s)
})

if err := client.Connect(ctx); err != nil { log.Fatal(err) }

sub, _ := client.Subscribe("orders-user-123", pushflo.SubscriptionHandlers{
    OnMessage: func(m pushflo.Message) {
        log.Printf("[%s] %s", m.EventType, string(m.Content))
    },
})
defer sub.Unsubscribe()

Options mirror the TS SDK: BaseURL, AutoConnect, Debug, ConnectionTimeout, HeartbeatInterval, AutoReconnect, MaxReconnectAttempts (0 = infinite), ReconnectDelay, MaxReconnectDelay. Channel-slug helpers (IsValidChannelSlug, ToChannelSlug, ValidateChannelSlug) and typed errors (ConnectionError, AuthenticationError, NetworkError, ValidationError) are included.

Encryption envelope

The relay is treated as untrusted transport. Each payload is sealed to the recipient's X25519 public key (NaCl sealed box) and signed with the sender's Ed25519 key. The recipient verifies the signature, checks freshness and replay, then decrypts.

 cloud (sender)                         PushFlo (relay)            executor (recipient)
 ─────────────                          ───────────────           ────────────────────
 Seal(payload) ──► {mid,ts,ch,ct,sig} ──► opaque bytes ──► Open() ─► payload
   sign Ed25519                            can't read /              verify sig
   seal to X25519 pub                      forge / replay            check ts + replay
                                                                     decrypt X25519

What each side holds after examples/keygen:

• Cloud: executor's RECIPIENT_PUBLIC + its own SIGNER_PRIVATE.
• Executor: its own RECIPIENT_PRIVATE + cloud's SIGNER_PUBLIC.

No secret is ever shared with PushFlo.

Sender:

sealer, _ := envelope.NewSealer(envelope.SealerConfig{
    RecipientPublic: executorPub,       // [32]byte
    SignerPrivate:   cloudSignPriv,     // ed25519.PrivateKey
    SignerKeyID:     "cloud-1",
})
env, _ := sealer.SealJSON(job, envelope.Meta{Channel: "executor-1"})
wire, _ := env.Marshal()                // publish `wire` as the channel message

Recipient:

opener, _ := envelope.NewOpener(envelope.OpenerConfig{
    RecipientPublic:  executorPub,
    RecipientPrivate: executorPriv,
    Signers:          map[string]ed25519.PublicKey{"cloud-1": cloudSignPub},
    MaxClockSkew:     2 * time.Minute,
    Replay:           envelope.NewMemoryReplayCache(10 * time.Minute),
})

env, _ := envelope.Parse(message.Content)
var job Job
if err := opener.OpenJSON(env, &job); err != nil {
    // ErrBadSignature / ErrReplay / ErrExpired / ErrDecrypt / ErrUnknownSigner
    return
}

What the envelope guarantees

• Confidentiality — only the recipient's private key can open the sealed box.
• Authenticity & integrity — every field (message id, timestamp, channel, content type) is bound into the Ed25519 signature, so the relay cannot alter routing metadata or the ciphertext undetected. This matters because the executor runs HTTP calls on command: an unauthenticated job would be an SSRF-with-your-credentials hole.
• Replay protection — message id + timestamp; ReplayCache rejects duplicates. Set the cache TTL ≥ MaxClockSkew. For a fleet or across restarts, back it with Redis/DB by implementing the ReplayCache interface.

What it does not hide

Message sizes, timing, and which channels are active remain visible to the relay. Pad or batch if that metadata matters.

Key rotation

KeyID (recipient epoch) and SignKeyID travel in the clear so you can rotate: give the Opener a Signers map with multiple ids, publish under the new id, then retire the old key.

Publisher

Publishing uses a secret (sec_) key. PublishSealed is the symmetric counterpart to the envelope Opener — it encrypts, signs, and publishes in one call, so the relay only sees ciphertext.

pub, _ := pushflo.NewPublisher(pushflo.PublisherOptions{SecretKey: "sec_xxx"})

// plaintext publish
pub.Publish(ctx, "orders", map[string]any{"status": "shipped"},
    pushflo.WithEventType("order.shipped"))

// encrypted publish (e.g. an executor returning a result to the cloud)
pub.PublishSealed(ctx, "results-executor-1", resultSealer, result, envelope.Meta{})

The publisher retries retryable failures (network errors, 429, 5xx) with jittered backoff; 4xx is returned immediately, and 401/403 become an AuthenticationError. API error bodies ({"error", "code"}) are surfaced in the error message, and the PublishResult carries the full server response (id, channel, event type, publisher client id, delivered count, timestamp).

Privileged channel/project operations use a mgmt_ key against the Console API (console.pushflo.dev) and live in a separate Management type (management.go), deliberately kept out of the publish path. It is currently a stub returning ErrNotImplemented: the method set and privilege boundary exist, but the REST calls are TODO until something needs to drive them programmatically.

Reference executor

examples/executor is the v1 agent in miniature: subscribe → decrypt job → run the HTTP request(s) locally with full timing capture (DNS/connect/TLS/TTFB via httptrace), redirects not auto-followed, per-request TLS-verify and timeout controls, structured error classification (dns/connect/tls/ timeout/http), and a size-capped base64 body. Egress to loopback, private, and link-local addresses is blocked by default (checked after DNS resolution, so rebinding can't bypass it); set EXECUTOR_ALLOW_PRIVATE=1 to probe internal targets deliberately.

go run ./examples/keygen                 # print key material
PUSHFLO_PUBLISH_KEY=pub_... \
CONTROL_CHANNEL=executor-1 \
RECIPIENT_PUBLIC=... RECIPIENT_PRIVATE=... SIGNER_PUBLIC=... \
  go run ./examples/executor

Wire protocol

The protocol lives isolated in protocol.go (WebSocket) and the top of publisher.go (REST) and is confirmed against the official @pushflodev/sdk v1.1.0 source:

• Auth — the API key travels as a ?token= query parameter on the WebSocket upgrade URL, and as Authorization: Bearer for REST. Note that URLs can end up in proxy/server logs; pub_ keys are low-privilege and revocable, but rotate them if a log leak is suspected.
• Connect — the client only counts as connected after the server's connected frame (which carries the client id), not on socket open.
• Heartbeat — an app-level {"type":"ping"} every interval, answered by {"type":"pong"}. A connection silent for 2x the interval is treated as dead and torn down (triggering auto-reconnect), so half-open TCP connections are detected promptly.

Integration tests

Unit tests run hermetically. To additionally verify against the live service:

PUSHFLO_PUBLISH_KEY=pub_... PUSHFLO_SECRET_KEY=sec_... \
  go test -run Integration -v .

They cover connect/auth, invalid-key rejection, a publish→subscribe round trip, and a sealed end-to-end envelope through the real relay (PUSHFLO_TEST_CHANNEL overrides the default go-sdk-integration channel).

Security

See SECURITY.md for the vulnerability disclosure policy, the envelope threat model, and key-handling guidance.

License

MIT — matching the upstream SDK.