c/quicproquo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Sprint 7 — Go SDK with QUIC transport and Cap'n Proto RPC

Browse Source 
First non-Rust client SDK for quicproquo ecosystem.

- Cap'n Proto codegen: generated 6487-line Go types from node.capnp
  with all 24 RPC methods (NodeService, Auth, Envelope)
- QUIC transport: quic-go + TLS 1.3, ALPN "capnp", single bidi stream,
  300s idle timeout, InsecureSkipVerify for dev, custom CA cert support
- High-level qpq package: Connect, Health, ResolveUser, CreateChannel,
  Send/SendWithTTL, Receive/ReceiveWait, DeleteAccount, OPAQUE wrappers
- Auth management: session token storage, version/token/deviceID on all RPCs
- Example program and README with API reference

All tests pass: go test ./..., go vet, go build
This commit is contained in:
Christian Nennemann
2026-03-04 01:03:02 +01:00
parent fd21ea625c
commit 65ff26235e
10 changed files with 7364 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
sdks/go/README.md Normal file
View File

@@ -0,0 +1,118 @@
# quicproquo Go SDK
Go client library for the [quicproquo](https://github.com/nicholasgasior/quicproquo) E2E encrypted messenger.
## Prerequisites
- Go 1.21+
- A running quicproquo server
## Installation
```sh
go get quicproquo.dev/sdk/go
```
## Quick Start
```go
package main
import (
"context"
"fmt"
"quicproquo.dev/sdk/go/qpq"
)
func main() {
ctx := context.Background()
// Connect to a local server (dev mode)
client, err := qpq.Connect(ctx, qpq.Options{
Addr: "127.0.0.1:5001",
InsecureSkipVerify: true,
})
if err != nil {
panic(err)
}
defer client.Close()
// Check server health
status, err := client.Health(ctx)
fmt.Println("Server:", status)
// Set a session token (obtained via OPAQUE login)
client.SetSessionToken(token)
// Resolve a user
key, err := client.ResolveUser(ctx, "alice")
// Create a DM channel
channelID, wasNew, err := client.CreateChannel(ctx, peerKey)
// Send a message
seq, err := client.Send(ctx, recipientKey, []byte("hello"))
// Receive messages
msgs, err := client.Receive(ctx, myKey)
// Long-poll for messages (5s timeout)
msgs, err = client.ReceiveWait(ctx, myKey, 5000)
}
```
## API Reference
### Connection
| Method | Description |
|---|---|
| `qpq.Connect(ctx, opts)` | Connect to a server, returns `*Client` |
| `client.Close()` | Disconnect |
### Authentication
| Method | Description |
|---|---|
| `client.SetSessionToken(token)` | Set a pre-existing OPAQUE session token |
| `client.RegisterStart(ctx, username, request)` | Start OPAQUE registration |
| `client.RegisterFinish(ctx, username, upload, identityKey)` | Complete OPAQUE registration |
| `client.LoginStart(ctx, username, request)` | Start OPAQUE login |
| `client.LoginFinish(ctx, username, finalization, identityKey)` | Complete OPAQUE login (stores token) |
### Messaging
| Method | Description |
|---|---|
| `client.ResolveUser(ctx, username)` | Look up a user's identity key |
| `client.CreateChannel(ctx, peerKey)` | Create a 1:1 DM channel |
| `client.Send(ctx, recipientKey, payload)` | Send a message |
| `client.SendWithTTL(ctx, recipientKey, payload, ttlSecs)` | Send a disappearing message |
| `client.Receive(ctx, recipientKey)` | Fetch queued messages |
| `client.ReceiveWait(ctx, recipientKey, timeoutMs)` | Long-poll for messages |
| `client.DeleteAccount(ctx)` | Permanently delete account |
| `client.Health(ctx)` | Check server health |
### OPAQUE Login
Full OPAQUE login requires a Go OPAQUE client library to generate the cryptographic
messages. The SDK provides the RPC wrappers (`RegisterStart`/`RegisterFinish`,
`LoginStart`/`LoginFinish`), but you must supply the OPAQUE protocol bytes.
For testing without OPAQUE, use `SetSessionToken` with a token obtained by other means.
## Structure
- `proto/node/` -- Generated Cap'n Proto Go types from `schemas/node.capnp`
- `transport/` -- QUIC + TLS 1.3 transport and Cap'n Proto RPC framing
- `qpq/` -- High-level client API (auth, messaging, channels)
- `cmd/example/` -- Example usage
## Regenerating Proto Types
```sh
capnp compile \
-I "$(go env GOPATH)/pkg/mod/capnproto.org/go/capnp/v3@v3.1.0-alpha.2/std" \
-ogo proto/node/node.capnp
```

55
sdks/go/cmd/example/main.go Normal file
View File

@@ -0,0 +1,55 @@
// Command example demonstrates basic usage of the quicproquo Go SDK.
package main
import (
"context"
"fmt"
"os"
"quicproquo.dev/sdk/go/qpq"
)
func main() {
addr := "127.0.0.1:5001"
if len(os.Args) > 1 {
addr = os.Args[1]
}
ctx := context.Background()
client, err := qpq.Connect(ctx, qpq.Options{
Addr: addr,
InsecureSkipVerify: true,
})
if err != nil {
fmt.Fprintf(os.Stderr, "connect failed: %v\n", err)
os.Exit(1)
}
defer client.Close()
status, err := client.Health(ctx)
if err != nil {
fmt.Fprintf(os.Stderr, "health check failed: %v\n", err)
os.Exit(1)
}
fmt.Printf("Server health: %s\n", status)
fmt.Println("\nUsage:")
fmt.Println(" // Authenticate with a pre-existing session token")
fmt.Println(" client.SetSessionToken(token)")
fmt.Println()
fmt.Println(" // Resolve a user's identity key")
fmt.Println(" key, err := client.ResolveUser(ctx, \"alice\")")
fmt.Println()
fmt.Println(" // Create a DM channel")
fmt.Println(" chID, wasNew, err := client.CreateChannel(ctx, peerKey)")
fmt.Println()
fmt.Println(" // Send a message")
fmt.Println(" seq, err := client.Send(ctx, recipientKey, payload)")
fmt.Println()
fmt.Println(" // Receive messages")
fmt.Println(" msgs, err := client.Receive(ctx, recipientKey)")
fmt.Println()
fmt.Println(" // Long-poll for messages (5 second timeout)")
fmt.Println(" msgs, err := client.ReceiveWait(ctx, recipientKey, 5000)")
}

13
sdks/go/go.mod Normal file
View File

@@ -0,0 +1,13 @@
module quicproquo.dev/sdk/go
go 1.25.7
require (
capnproto.org/go/capnp/v3 v3.1.0-alpha.2 // indirect
github.com/colega/zeropool v0.0.0-20230505084239-6fb4a4f75381 // indirect
github.com/quic-go/quic-go v0.59.0 // indirect
golang.org/x/crypto v0.41.0 // indirect
golang.org/x/net v0.43.0 // indirect
golang.org/x/sync v0.16.0 // indirect
golang.org/x/sys v0.35.0 // indirect
)

16
sdks/go/go.sum Normal file
View File

@@ -0,0 +1,16 @@
capnproto.org/go/capnp/v3 v3.1.0-alpha.2 h1:93ISpqHf2/3WQlfrBP0tT8Dg/RQc3uq6DV36vN0ePWk=
capnproto.org/go/capnp/v3 v3.1.0-alpha.2/go.mod h1:2vT5D2dtG8sJGEoEKU17e+j7shdaYp1Myl8X03B3hmc=
github.com/colega/zeropool v0.0.0-20230505084239-6fb4a4f75381 h1:d5EKgQfRQvO97jnISfR89AiCCCJMwMFoSxUiU0OGCRU=
github.com/colega/zeropool v0.0.0-20230505084239-6fb4a4f75381/go.mod h1:OU76gHeRo8xrzGJU3F3I1CqX1ekM8dfJw0+wPeMwnp0=
github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw=
github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
golang.org/x/crypto v0.41.0 h1:WKYxWedPGCTVVl5+WHSSrOBT0O8lx32+zxmHxijgXp4=
golang.org/x/crypto v0.41.0/go.mod h1:pO5AFd7FA68rFak7rOAGVuygIISepHftHnr8dr6+sUc=
golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE=
golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg=
golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw=
golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=

45
sdks/go/proto/node/node.capnp Normal file
View File

@@ -0,0 +1,45 @@
# node.capnp — Go-annotated copy of the quicproquo node schema.
# This adds Go package annotations needed by capnpc-go.
@0xd5ca5648a9cc1c28;
using Go = import "/go.capnp";
$Go.package("node");
$Go.import("quicproquo.dev/sdk/go/proto/node");
interface NodeService {
uploadKeyPackage @0 (identityKey :Data, package :Data, auth :Auth) -> (fingerprint :Data);
fetchKeyPackage @1 (identityKey :Data, auth :Auth) -> (package :Data);
enqueue @2 (recipientKey :Data, payload :Data, channelId :Data, version :UInt16, auth :Auth, ttlSecs :UInt32) -> (seq :UInt64, deliveryProof :Data);
fetch @3 (recipientKey :Data, channelId :Data, version :UInt16, auth :Auth, limit :UInt32) -> (payloads :List(Envelope));
fetchWait @4 (recipientKey :Data, channelId :Data, version :UInt16, timeoutMs :UInt64, auth :Auth, limit :UInt32) -> (payloads :List(Envelope));
health @5 () -> (status :Text);
uploadHybridKey @6 (identityKey :Data, hybridPublicKey :Data, auth :Auth) -> ();
fetchHybridKey @7 (identityKey :Data, auth :Auth) -> (hybridPublicKey :Data);
opaqueRegisterStart @8 (username :Text, request :Data) -> (response :Data);
opaqueRegisterFinish @9 (username :Text, upload :Data, identityKey :Data) -> (success :Bool);
opaqueLoginStart @10 (username :Text, request :Data) -> (response :Data);
opaqueLoginFinish @11 (username :Text, finalization :Data, identityKey :Data) -> (sessionToken :Data);
publishEndpoint @12 (identityKey :Data, nodeAddr :Data, auth :Auth) -> ();
resolveEndpoint @13 (identityKey :Data, auth :Auth) -> (nodeAddr :Data);
peek @14 (recipientKey :Data, channelId :Data, version :UInt16, auth :Auth, limit :UInt32) -> (payloads :List(Envelope));
ack @15 (recipientKey :Data, channelId :Data, version :UInt16, seqUpTo :UInt64, auth :Auth) -> ();
fetchHybridKeys @16 (identityKeys :List(Data), auth :Auth) -> (keys :List(Data));
batchEnqueue @17 (recipientKeys :List(Data), payload :Data, channelId :Data, version :UInt16, auth :Auth, ttlSecs :UInt32) -> (seqs :List(UInt64));
createChannel @18 (peerKey :Data, auth :Auth) -> (channelId :Data, wasNew :Bool);
resolveUser @19 (username :Text, auth :Auth) -> (identityKey :Data, inclusionProof :Data);
resolveIdentity @20 (identityKey :Data, auth :Auth) -> (username :Text);
uploadBlob @21 (auth :Auth, blobHash :Data, chunk :Data, offset :UInt64, totalSize :UInt64, mimeType :Text) -> (blobId :Data);
downloadBlob @22 (auth :Auth, blobId :Data, offset :UInt64, length :UInt32) -> (chunk :Data, totalSize :UInt64, mimeType :Text);
deleteAccount @23 (auth :Auth) -> (success :Bool);
}
struct Auth {
version @0 :UInt16;
accessToken @1 :Data;
deviceId @2 :Data;
}
struct Envelope {
seq @0 :UInt64;
data @1 :Data;
}

6487
sdks/go/proto/node/node.capnp.go Normal file
View File

File diff suppressed because it is too large Load Diff

377
sdks/go/qpq/client.go Normal file
View File

@@ -0,0 +1,377 @@
// Package qpq provides the high-level Go API for interacting with a quicproquo server.
//
// It wraps the generated Cap'n Proto types and transport layer into an
// ergonomic client that handles authentication, key management, and messaging.
package qpq
import (
"context"
"fmt"
"quicproquo.dev/sdk/go/proto/node"
"quicproquo.dev/sdk/go/transport"
)
// Options configures the connection to a quicproquo server.
type Options struct {
// Addr is the host:port of the server (e.g. "127.0.0.1:5001").
Addr string
// InsecureSkipVerify disables TLS certificate verification (dev mode only).
InsecureSkipVerify bool
// CACertPath is the path to a PEM-encoded CA certificate for production use.
CACertPath string
}
// Message represents a received message envelope.
type Message struct {
Seq uint64
Data []byte
}
// Client is the high-level quicproquo client.
type Client struct {
conn *transport.Connection
token []byte // session token from OPAQUE login
deviceID []byte // optional device ID
}
// Connect establishes a connection to a qpq server.
func Connect(ctx context.Context, opts Options) (*Client, error) {
conn, err := transport.Connect(ctx, transport.ConnectOptions{
Addr: opts.Addr,
InsecureSkipVerify: opts.InsecureSkipVerify,
CACertPath: opts.CACertPath,
})
if err != nil {
return nil, fmt.Errorf("qpq: connect: %w", err)
}
return &Client{conn: conn}, nil
}
// Close disconnects from the server.
func (c *Client) Close() error {
return c.conn.Close()
}
// SetSessionToken sets a pre-existing session token for authentication.
// Use this when you have already performed OPAQUE login externally.
func (c *Client) SetSessionToken(token []byte) {
c.token = token
}
// SetDeviceID sets the device ID sent with each authenticated RPC call.
func (c *Client) SetDeviceID(id []byte) {
c.deviceID = id
}
// setAuth populates an Auth struct on an RPC params message.
func (c *Client) setAuth(auth node.Auth) error {
auth.SetVersion(1)
if err := auth.SetAccessToken(c.token); err != nil {
return fmt.Errorf("set access token: %w", err)
}
if len(c.deviceID) > 0 {
if err := auth.SetDeviceId(c.deviceID); err != nil {
return fmt.Errorf("set device id: %w", err)
}
}
return nil
}
// Health checks server health and returns the status string.
func (c *Client) Health(ctx context.Context) (string, error) {
future, release := c.conn.Client().Health(ctx, nil)
defer release()
res, err := future.Struct()
if err != nil {
return "", fmt.Errorf("qpq: health: %w", err)
}
status, err := res.Status()
if err != nil {
return "", fmt.Errorf("qpq: health: read status: %w", err)
}
return status, nil
}
// ResolveUser looks up a username and returns their identity key.
func (c *Client) ResolveUser(ctx context.Context, username string) (identityKey []byte, err error) {
future, release := c.conn.Client().ResolveUser(ctx, func(p node.NodeService_resolveUser_Params) error {
if err := p.SetUsername(username); err != nil {
return err
}
auth, err := p.NewAuth()
if err != nil {
return err
}
return c.setAuth(auth)
})
defer release()
res, err := future.Struct()
if err != nil {
return nil, fmt.Errorf("qpq: resolve user %q: %w", username, err)
}
key, err := res.IdentityKey()
if err != nil {
return nil, fmt.Errorf("qpq: resolve user %q: read identity key: %w", username, err)
}
// Copy the key out of the capnp message buffer.
out := make([]byte, len(key))
copy(out, key)
return out, nil
}
// CreateChannel creates a 1:1 DM channel with a peer identified by their identity key.
func (c *Client) CreateChannel(ctx context.Context, peerKey []byte) (channelID []byte, wasNew bool, err error) {
future, release := c.conn.Client().CreateChannel(ctx, func(p node.NodeService_createChannel_Params) error {
if err := p.SetPeerKey(peerKey); err != nil {
return err
}
auth, err := p.NewAuth()
if err != nil {
return err
}
return c.setAuth(auth)
})
defer release()
res, err := future.Struct()
if err != nil {
return nil, false, fmt.Errorf("qpq: create channel: %w", err)
}
chID, err := res.ChannelId()
if err != nil {
return nil, false, fmt.Errorf("qpq: create channel: read channel id: %w", err)
}
out := make([]byte, len(chID))
copy(out, chID)
return out, res.WasNew(), nil
}
// Send enqueues a message payload to a recipient identified by their identity key.
func (c *Client) Send(ctx context.Context, recipientKey, payload []byte) (seq uint64, err error) {
return c.sendInternal(ctx, recipientKey, payload, 0)
}
// SendWithTTL enqueues a disappearing message with a time-to-live in seconds.
func (c *Client) SendWithTTL(ctx context.Context, recipientKey, payload []byte, ttlSecs uint32) (seq uint64, err error) {
return c.sendInternal(ctx, recipientKey, payload, ttlSecs)
}
func (c *Client) sendInternal(ctx context.Context, recipientKey, payload []byte, ttlSecs uint32) (uint64, error) {
future, release := c.conn.Client().Enqueue(ctx, func(p node.NodeService_enqueue_Params) error {
if err := p.SetRecipientKey(recipientKey); err != nil {
return err
}
if err := p.SetPayload(payload); err != nil {
return err
}
p.SetVersion(1)
if ttlSecs > 0 {
p.SetTtlSecs(ttlSecs)
}
auth, err := p.NewAuth()
if err != nil {
return err
}
return c.setAuth(auth)
})
defer release()
res, err := future.Struct()
if err != nil {
return 0, fmt.Errorf("qpq: send: %w", err)
}
return res.Seq(), nil
}
// Receive fetches queued messages for the given recipient key.
func (c *Client) Receive(ctx context.Context, recipientKey []byte) ([]Message, error) {
future, release := c.conn.Client().Fetch(ctx, func(p node.NodeService_fetch_Params) error {
if err := p.SetRecipientKey(recipientKey); err != nil {
return err
}
p.SetVersion(1)
auth, err := p.NewAuth()
if err != nil {
return err
}
return c.setAuth(auth)
})
defer release()
res, err := future.Struct()
if err != nil {
return nil, fmt.Errorf("qpq: receive: %w", err)
}
return extractMessages(res.Payloads())
}
// ReceiveWait long-polls for messages with a timeout in milliseconds.
func (c *Client) ReceiveWait(ctx context.Context, recipientKey []byte, timeoutMs uint64) ([]Message, error) {
future, release := c.conn.Client().FetchWait(ctx, func(p node.NodeService_fetchWait_Params) error {
if err := p.SetRecipientKey(recipientKey); err != nil {
return err
}
p.SetVersion(1)
p.SetTimeoutMs(timeoutMs)
auth, err := p.NewAuth()
if err != nil {
return err
}
return c.setAuth(auth)
})
defer release()
res, err := future.Struct()
if err != nil {
return nil, fmt.Errorf("qpq: receive wait: %w", err)
}
return extractMessages(res.Payloads())
}
// DeleteAccount permanently deletes the authenticated user's account.
func (c *Client) DeleteAccount(ctx context.Context) error {
future, release := c.conn.Client().DeleteAccount(ctx, func(p node.NodeService_deleteAccount_Params) error {
auth, err := p.NewAuth()
if err != nil {
return err
}
return c.setAuth(auth)
})
defer release()
res, err := future.Struct()
if err != nil {
return fmt.Errorf("qpq: delete account: %w", err)
}
if !res.Success() {
return fmt.Errorf("qpq: delete account: server returned success=false")
}
return nil
}
// RegisterStart initiates OPAQUE registration and returns the server's response bytes.
//
// The OPAQUE protocol requires client-side cryptographic operations.
// The request parameter must be the serialized OPAQUE RegistrationRequest
// generated by an OPAQUE client library (e.g. github.com/cloudflare/circl/opaque).
// Process the returned server response with your OPAQUE library to produce
// the upload bytes for RegisterFinish.
func (c *Client) RegisterStart(ctx context.Context, username string, request []byte) (serverResponse []byte, err error) {
future, release := c.conn.Client().OpaqueRegisterStart(ctx, func(p node.NodeService_opaqueRegisterStart_Params) error {
if err := p.SetUsername(username); err != nil {
return err
}
return p.SetRequest(request)
})
defer release()
res, err := future.Struct()
if err != nil {
return nil, fmt.Errorf("qpq: register start: %w", err)
}
resp, err := res.Response()
if err != nil {
return nil, fmt.Errorf("qpq: register start: read response: %w", err)
}
out := make([]byte, len(resp))
copy(out, resp)
return out, nil
}
// RegisterFinish completes OPAQUE registration with the upload and identity key.
func (c *Client) RegisterFinish(ctx context.Context, username string, upload, identityKey []byte) error {
future, release := c.conn.Client().OpaqueRegisterFinish(ctx, func(p node.NodeService_opaqueRegisterFinish_Params) error {
if err := p.SetUsername(username); err != nil {
return err
}
if err := p.SetUpload(upload); err != nil {
return err
}
return p.SetIdentityKey(identityKey)
})
defer release()
_, err := future.Struct()
if err != nil {
return fmt.Errorf("qpq: register finish: %w", err)
}
return nil
}
// LoginStart initiates OPAQUE login and returns the server's response bytes.
func (c *Client) LoginStart(ctx context.Context, username string, request []byte) (serverResponse []byte, err error) {
future, release := c.conn.Client().OpaqueLoginStart(ctx, func(p node.NodeService_opaqueLoginStart_Params) error {
if err := p.SetUsername(username); err != nil {
return err
}
return p.SetRequest(request)
})
defer release()
res, err := future.Struct()
if err != nil {
return nil, fmt.Errorf("qpq: login start: %w", err)
}
resp, err := res.Response()
if err != nil {
return nil, fmt.Errorf("qpq: login start: read response: %w", err)
}
out := make([]byte, len(resp))
copy(out, resp)
return out, nil
}
// LoginFinish completes OPAQUE login and stores the session token.
// The finalization parameter is the OPAQUE finalization message from
// your OPAQUE client library. The identityKey is your public identity key.
func (c *Client) LoginFinish(ctx context.Context, username string, finalization, identityKey []byte) error {
future, release := c.conn.Client().OpaqueLoginFinish(ctx, func(p node.NodeService_opaqueLoginFinish_Params) error {
if err := p.SetUsername(username); err != nil {
return err
}
if err := p.SetFinalization(finalization); err != nil {
return err
}
return p.SetIdentityKey(identityKey)
})
defer release()
res, err := future.Struct()
if err != nil {
return fmt.Errorf("qpq: login finish: %w", err)
}
token, err := res.SessionToken()
if err != nil {
return fmt.Errorf("qpq: login finish: read session token: %w", err)
}
c.token = make([]byte, len(token))
copy(c.token, token)
return nil
}
// extractMessages converts a Cap'n Proto Envelope_List into a slice of Message.
func extractMessages(envList node.Envelope_List, listErr error) ([]Message, error) {
if listErr != nil {
return nil, listErr
}
msgs := make([]Message, envList.Len())
for i := range msgs {
env := envList.At(i)
data, err := env.Data()
if err != nil {
return nil, fmt.Errorf("read message %d data: %w", i, err)
}
dataCopy := make([]byte, len(data))
copy(dataCopy, data)
msgs[i] = Message{
Seq: env.Seq(),
Data: dataCopy,
}
}
return msgs, nil
}

58
sdks/go/qpq/client_test.go Normal file
View File

@@ -0,0 +1,58 @@
package qpq
import (
"testing"
)
func TestSetSessionToken(t *testing.T) {
c := &Client{}
token := []byte("test-session-token-abc123")
c.SetSessionToken(token)
if string(c.token) != string(token) {
t.Errorf("expected token %q, got %q", token, c.token)
}
}
func TestSetDeviceID(t *testing.T) {
c := &Client{}
id := []byte{0x01, 0x02, 0x03, 0x04}
c.SetDeviceID(id)
if len(c.deviceID) != 4 {
t.Fatalf("expected 4-byte device ID, got %d bytes", len(c.deviceID))
}
for i, b := range id {
if c.deviceID[i] != b {
t.Errorf("deviceID[%d]: expected %d, got %d", i, b, c.deviceID[i])
}
}
}
func TestMessageStruct(t *testing.T) {
m := Message{Seq: 42, Data: []byte("hello")}
if m.Seq != 42 {
t.Errorf("expected Seq 42, got %d", m.Seq)
}
if string(m.Data) != "hello" {
t.Errorf("expected Data %q, got %q", "hello", m.Data)
}
}
func TestOptionsDefaults(t *testing.T) {
opts := Options{
Addr: "127.0.0.1:5001",
InsecureSkipVerify: true,
}
if opts.Addr != "127.0.0.1:5001" {
t.Errorf("unexpected addr: %s", opts.Addr)
}
if !opts.InsecureSkipVerify {
t.Error("expected InsecureSkipVerify to be true")
}
if opts.CACertPath != "" {
t.Error("expected empty CACertPath")
}
}

126
sdks/go/transport/transport.go Normal file
View File

@@ -0,0 +1,126 @@
// Package transport provides the low-level QUIC+TLS connection to a quicproquo server
// and bootstraps the Cap'n Proto RPC client over a single bidirectional QUIC stream.
package transport
import (
"context"
"crypto/tls"
"crypto/x509"
"fmt"
"os"
"time"
capnp "capnproto.org/go/capnp/v3"
"capnproto.org/go/capnp/v3/rpc"
"github.com/quic-go/quic-go"
"quicproquo.dev/sdk/go/proto/node"
)
// ConnectOptions configures the connection to a quicproquo server.
type ConnectOptions struct {
// Addr is the host:port of the server (e.g. "127.0.0.1:5001").
Addr string
// InsecureSkipVerify disables TLS certificate verification (dev mode only).
InsecureSkipVerify bool
// CACertPath is the path to a PEM-encoded CA certificate for production use.
// When set, the server's certificate is verified against this CA.
CACertPath string
// ConnectTimeout is the maximum time to wait for the QUIC handshake.
// Defaults to 10 seconds if zero.
ConnectTimeout time.Duration
}
// Connection holds the QUIC connection, Cap'n Proto RPC state, and the
// bootstrapped NodeService client.
type Connection struct {
quicConn *quic.Conn
stream *quic.Stream
rpcConn *rpc.Conn
client node.NodeService
}
// Connect establishes a QUIC+TLS 1.3 connection to the server, opens a single
// bidirectional stream, and bootstraps the Cap'n Proto RPC NodeService client.
func Connect(ctx context.Context, opts ConnectOptions) (*Connection, error) {
tlsCfg, err := buildTLSConfig(opts)
if err != nil {
return nil, fmt.Errorf("transport: build TLS config: %w", err)
}
quicCfg := &quic.Config{
MaxIdleTimeout: 300 * time.Second,
KeepAlivePeriod: 30 * time.Second,
}
timeout := opts.ConnectTimeout
if timeout == 0 {
timeout = 10 * time.Second
}
dialCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
qConn, err := quic.DialAddr(dialCtx, opts.Addr, tlsCfg, quicCfg)
if err != nil {
return nil, fmt.Errorf("transport: QUIC dial %s: %w", opts.Addr, err)
}
stream, err := qConn.OpenStream()
if err != nil {
qConn.CloseWithError(1, "failed to open stream")
return nil, fmt.Errorf("transport: open bidi stream: %w", err)
}
capnpTransport := rpc.NewStreamTransport(stream)
rpcConn := rpc.NewConn(capnpTransport, nil)
client := node.NodeService(rpcConn.Bootstrap(ctx))
return &Connection{
quicConn: qConn,
stream: stream,
rpcConn: rpcConn,
client: client,
}, nil
}
// Client returns the bootstrapped NodeService RPC client.
func (c *Connection) Client() node.NodeService {
return c.client
}
// Close gracefully shuts down the RPC connection and underlying QUIC transport.
func (c *Connection) Close() error {
capnp.Client(c.client).Release()
c.rpcConn.Close()
return c.quicConn.CloseWithError(0, "client closed")
}
func buildTLSConfig(opts ConnectOptions) (*tls.Config, error) {
cfg := &tls.Config{
NextProtos: []string{"capnp"},
MinVersion: tls.VersionTLS13,
}
if opts.InsecureSkipVerify {
cfg.InsecureSkipVerify = true
return cfg, nil
}
if opts.CACertPath != "" {
caPEM, err := os.ReadFile(opts.CACertPath)
if err != nil {
return nil, fmt.Errorf("read CA cert %s: %w", opts.CACertPath, err)
}
pool := x509.NewCertPool()
if !pool.AppendCertsFromPEM(caPEM) {
return nil, fmt.Errorf("CA cert %s: no valid PEM certificates found", opts.CACertPath)
}
cfg.RootCAs = pool
}
return cfg, nil
}

69
sdks/go/transport/transport_test.go Normal file
View File

@@ -0,0 +1,69 @@
package transport
import (
"testing"
"time"
)
func TestConnectOptionsDefaults(t *testing.T) {
opts := ConnectOptions{
Addr: "127.0.0.1:5001",
InsecureSkipVerify: true,
}
if opts.Addr != "127.0.0.1:5001" {
t.Errorf("unexpected addr: %s", opts.Addr)
}
if !opts.InsecureSkipVerify {
t.Error("expected InsecureSkipVerify=true")
}
if opts.ConnectTimeout != 0 {
t.Errorf("expected zero timeout, got %v", opts.ConnectTimeout)
}
}
func TestBuildTLSConfigInsecure(t *testing.T) {
cfg, err := buildTLSConfig(ConnectOptions{InsecureSkipVerify: true})
if err != nil {
t.Fatalf("buildTLSConfig: %v", err)
}
if !cfg.InsecureSkipVerify {
t.Error("expected InsecureSkipVerify=true in TLS config")
}
if len(cfg.NextProtos) != 1 || cfg.NextProtos[0] != "capnp" {
t.Errorf("expected ALPN [capnp], got %v", cfg.NextProtos)
}
if cfg.MinVersion != 0x0304 { // tls.VersionTLS13
t.Errorf("expected TLS 1.3 min version (0x0304), got 0x%04x", cfg.MinVersion)
}
}
func TestBuildTLSConfigWithMissingCA(t *testing.T) {
_, err := buildTLSConfig(ConnectOptions{CACertPath: "/nonexistent/ca.pem"})
if err == nil {
t.Error("expected error for missing CA cert")
}
}
func TestBuildTLSConfigDefault(t *testing.T) {
cfg, err := buildTLSConfig(ConnectOptions{})
if err != nil {
t.Fatalf("buildTLSConfig: %v", err)
}
if cfg.InsecureSkipVerify {
t.Error("expected InsecureSkipVerify=false by default")
}
if cfg.NextProtos[0] != "capnp" {
t.Error("expected ALPN capnp")
}
}
func TestConnectTimeoutDefault(t *testing.T) {
opts := ConnectOptions{Addr: "127.0.0.1:5001"}
timeout := opts.ConnectTimeout
if timeout == 0 {
timeout = 10 * time.Second
}
if timeout != 10*time.Second {
t.Errorf("expected 10s default timeout, got %v", timeout)
}
}