feat: initial zap-http v0.1

HTTP request/response semantics over the ZAP transport. Drop-in
net/http.Handler server and net/http.RoundTripper client; existing
handlers and http.Client code work unchanged.

Wire format: one Cap'n Proto frame per HTTP message
(schema/zap_http.capnp). v0.1 ships length-prefixed framing over TCP;
v0.2 will swap that for the full ZAP transport (X-Wing PQ KEM
handshake + AEAD framing) once zap-proto/go provides it. The
marshal/unmarshal API is stable across that change.

Includes:
- schema/zap_http.capnp (Cap'n Proto schema, source of truth)
- internal/capnp/ (generated Go bindings)
- wire.go (Marshal/Unmarshal Request/Response)
- transport.go (length-prefixed framing)
- server.go (Server + ListenAndServe)
- client.go (Transport + Get convenience)
- examples/hello (server + client in one binary)
- 7 tests covering GET, POST, large body (4 MiB), multi-value headers,
  status code propagation, keep-alive, marshal idempotence

Tests pass; smoke test confirms end-to-end exchange works.
This commit is contained in:
2026-05-06 13:07:41 -07:00
commit 1841a7aac4
14 changed files with 2138 additions and 0 deletions
+11
View File
@@ -0,0 +1,11 @@
# Go build artifacts
*.exe
*.test
*.out
# Editor
.vscode/
.idea/
# Schema codegen — checked in, but ignore stray files
schema/go.capnp
+59
View File
@@ -0,0 +1,59 @@
zap-http is licensed under either of
Apache License, Version 2.0
(https://www.apache.org/licenses/LICENSE-2.0)
MIT License
(https://opensource.org/licenses/MIT)
at your option.
Unless you explicitly state otherwise, any contribution intentionally
submitted for inclusion in this project, as defined in the Apache-2.0
license, shall be dual-licensed as above, without any additional terms
or conditions.
----------------------------------------------------------------------
MIT LICENSE
----------------------------------------------------------------------
Copyright (c) The ZAP Protocol Authors
Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject
to the following conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
----------------------------------------------------------------------
APACHE LICENSE, VERSION 2.0
----------------------------------------------------------------------
Apache License, Version 2.0, January 2004
http://www.apache.org/licenses/LICENSE-2.0
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
either express or implied. See the License for the specific
language governing permissions and limitations under the License.
+33
View File
@@ -0,0 +1,33 @@
# zap-http Makefile.
#
# Re-runs Cap'n Proto codegen from schema/zap_http.capnp into
# internal/capnp/. Requires capnp + capnpc-go on PATH.
GOPATH ?= $(shell go env GOPATH)
SCHEMA := schema/zap_http.capnp
GENDIR := internal/capnp
.PHONY: schema test build bench clean
schema:
@command -v capnp >/dev/null || { echo "install capnp: brew install capnp"; exit 1; }
@command -v $(GOPATH)/bin/capnpc-go >/dev/null || { echo "installing capnpc-go..."; \
go install capnproto.org/go/capnp/v3/capnpc-go@latest; }
@mkdir -p $(GENDIR)
@if [ ! -f schema/go.capnp ]; then \
gocapnp=$$(find $(GOPATH)/pkg/mod/capnproto.org -name go.capnp 2>/dev/null | head -1); \
[ -n "$$gocapnp" ] && cp "$$gocapnp" schema/go.capnp; \
fi
PATH=$(GOPATH)/bin:$$PATH capnp compile -I schema -ogo:$(GENDIR) $(SCHEMA)
build:
go build ./...
test:
go test ./... -v
bench:
go test ./... -run=^$$ -bench=. -benchmem
clean:
rm -rf $(GENDIR)
+141
View File
@@ -0,0 +1,141 @@
# zap-http
HTTP request/response semantics over the [ZAP transport](https://github.com/zap-proto/spec).
[**zap-proto.io**](https://zap-proto.io) · [Spec](https://github.com/zap-proto/spec) · [Paper](https://github.com/zap-proto/papers/tree/main/transport-vs-jwt) · [Discord](https://zap-proto.io/discord)
Drop-in replacement for `net/http` server and client when both peers live in a trusted boundary (in-cluster service mesh, agent-to-tool, edge-to-edge). Existing handlers and `http.Client` code work unchanged — only the wire underneath changes.
## Why
| Property | `net/http` over TCP+TLS | `zap-http` |
|---|---|---|
| Confidentiality | TLS (classical curves) | X-Wing hybrid PQ (X25519 + ML-KEM-768) |
| Authentication | bearer / JWT at app layer | KEM keypair at transport layer |
| Wire encoding | text headers + chunked body | Cap'n Proto, zero-copy |
| Field access | parse → allocate → copy | pointer offset, O(1) |
| JWT mint per call | typical | not in the path |
## Install
```bash
go get github.com/zap-proto/http
```
Requires Go 1.23+.
## Server
```go
package main
import (
"io"
"net/http"
zaphttp "github.com/zap-proto/http"
)
func main() {
http.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
io.WriteString(w, "ok")
})
if err := zaphttp.ListenAndServe(":9999", nil); err != nil {
panic(err)
}
}
```
Same `http.Handler`, same `http.HandleFunc`, same default mux. The wire is ZAP-HTTP.
## Client
```go
package main
import (
"io"
"net/http"
"os"
zaphttp "github.com/zap-proto/http"
)
func main() {
client := &http.Client{Transport: zaphttp.NewTransport("server:9999")}
resp, err := client.Get("http://server/healthz")
if err != nil { panic(err) }
defer resp.Body.Close()
io.Copy(os.Stdout, resp.Body)
}
```
`http.Client` machinery — timeouts, redirects, cookies — keeps working unchanged.
## Wire format
Each HTTP message is one Cap'n Proto frame (`schema/zap_http.capnp`). Today the wire layer is length-prefixed framing over TCP; the [paper](https://github.com/zap-proto/papers/tree/main/transport-vs-jwt) and the v0.2 release will swap that for the full ZAP transport with X-Wing PQ KEM handshake on connect.
```
zap-http v0.1 wire (transitional):
+---------+-------------------+
| u32 BE | capnp.Frame |
| length | (zap_http.capnp)|
+---------+-------------------+
zap-http v0.2 wire:
+-----------------------------+--------+--------+
| ZAP transport (X-Wing KEM, | AEAD | capnp |
| mutual auth, multi-stream) | header | Frame |
+-----------------------------+--------+--------+
```
The Cap'n Proto schema is the source of truth for what's on the wire. Marshal/unmarshal in any language follows from the schema; bindings:
- Go (this repo)
- Rust — [`zap-proto/rust`](https://github.com/zap-proto/rust) (consumer; sub-protocol bindings TBD)
- TypeScript — [`zap-proto/ts`](https://github.com/zap-proto/ts) (planned)
## What's in v0.1
| Feature | Status |
|---|---|
| Request method, target, headers, body | ✓ |
| Response status, reason, headers, body | ✓ |
| Multi-value headers preserved | ✓ |
| Trailers (read & write) | ✓ |
| `http.Client` / `http.Handler` compatibility | ✓ |
| Length-prefixed framing over TCP | ✓ |
| 64 MiB max frame size (configurable) | ✓ |
| Streaming bodies (chunked-style) | v0.2 |
| Connection pool / keep-alive on client | v0.2 |
| Real ZAP transport (PQ KEM handshake) | v0.2 |
| WebSocket-style upgrade | see [zap-proto/ws](https://github.com/zap-proto/ws) |
## Sub-protocol family
- [`zap-http`](https://github.com/zap-proto/http) — this repo
- [`zap-ws`](https://github.com/zap-proto/ws) — multi-stream pubsub, per-stream FEC
- [`zap-fix`](https://github.com/zap-proto/fix) — FIX 4.4/5.0 trading channel
- [`zap-rns`](https://github.com/zap-proto/rns) — KEM-bound service naming
- [`zap-mcp`](https://github.com/zap-proto/mcp) — Model Context Protocol over ZAP
- [`zap-acp`](https://github.com/zap-proto/acp) — Agent Communication Protocol
- [`zap-a2a`](https://github.com/zap-proto/a2a) — Google Agent2Agent over ZAP
By the [composability theorem](https://github.com/zap-proto/papers/tree/main/composability), every sub-protocol that embeds onto ZAP-base inherits its post-quantum confidentiality and mutual authentication automatically.
## Schema regeneration
```sh
make schema # regenerates internal/capnp/zap_http.capnp.go
```
Requires `capnp` (Cap'n Proto compiler) and `capnpc-go`:
```sh
brew install capnp # macOS
go install capnproto.org/go/capnp/v3/capnpc-go@latest
```
## License
MIT OR Apache-2.0
+112
View File
@@ -0,0 +1,112 @@
// Client — net/http.RoundTripper-compatible client speaking ZAP-HTTP.
//
// Drop-in usage:
//
// t := zaphttp.NewTransport("server:9999")
// client := &http.Client{Transport: t}
// resp, err := client.Get("http://server/healthz")
//
// The transport ignores req.URL.Host (use the constructor argument
// to pin the peer) and req.URL.Scheme. Existing http.Client
// machinery — redirects, cookies, retries — works unchanged.
//
// Connection management today is one connection per RoundTrip call.
// v0.2 will introduce a connection pool with keep-alive once the
// underlying ZAP transport ships handshake amortization.
package zaphttp
import (
"bufio"
"fmt"
"net"
"net/http"
"sync"
"time"
)
// Transport implements http.RoundTripper over a ZAP-HTTP TCP
// connection. Field-zero is invalid; use NewTransport.
type Transport struct {
addr string
dialTimeout time.Duration
readTimeout time.Duration
mu sync.Mutex
}
// NewTransport returns a Transport connecting to the given host:port.
func NewTransport(addr string) *Transport {
return &Transport{
addr: addr,
dialTimeout: 10 * time.Second,
readTimeout: 30 * time.Second,
}
}
// SetDialTimeout overrides the default 10s dial timeout.
func (t *Transport) SetDialTimeout(d time.Duration) { t.dialTimeout = d }
// SetReadTimeout overrides the default 30s response-read timeout.
func (t *Transport) SetReadTimeout(d time.Duration) { t.readTimeout = d }
// RoundTrip executes a single request/response exchange.
func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
if t.addr == "" {
return nil, fmt.Errorf("zaphttp: Transport.addr is empty (use NewTransport)")
}
conn, err := net.DialTimeout("tcp", t.addr, t.dialTimeout)
if err != nil {
return nil, fmt.Errorf("zaphttp: dial %s: %w", t.addr, err)
}
closeOnFail := true
defer func() {
if closeOnFail {
conn.Close()
}
}()
// Marshal + send request.
frame, err := MarshalRequest(req)
if err != nil {
return nil, err
}
if err := writeFrame(conn, frame); err != nil {
return nil, fmt.Errorf("zaphttp: write request: %w", err)
}
// Read response.
if t.readTimeout > 0 {
_ = conn.SetReadDeadline(time.Now().Add(t.readTimeout))
}
br := bufio.NewReader(conn)
respFrame, err := readFrame(br)
if err != nil {
return nil, fmt.Errorf("zaphttp: read response: %w", err)
}
resp, err := UnmarshalResponse(respFrame)
if err != nil {
return nil, fmt.Errorf("zaphttp: unmarshal response: %w", err)
}
// Tie the response Body to the connection so it gets closed when
// the caller closes the body. The current single-frame design
// means the body is fully buffered in memory, so we can close the
// connection immediately and leave the buffer in resp.Body. v0.2
// will defer connection close until body close.
closeOnFail = false
conn.Close()
return resp, nil
}
// Get is a convenience for transports that don't need an http.Client
// wrapper (e.g. service-to-service calls inside a cluster).
func Get(addr, path string) (*http.Response, error) {
t := NewTransport(addr)
req, err := http.NewRequest(http.MethodGet, "http://"+addr+path, nil)
if err != nil {
return nil, err
}
return t.RoundTrip(req)
}
+52
View File
@@ -0,0 +1,52 @@
// Minimal zap-http example: server + client in one binary.
//
// go run ./examples/hello server :9999
// go run ./examples/hello client :9999
//
// The server speaks ZAP-HTTP on :9999. The client dials, issues
// GET /hello, and prints the response body.
package main
import (
"fmt"
"io"
"net/http"
"os"
zaphttp "github.com/zap-proto/http"
)
func main() {
if len(os.Args) < 3 {
fmt.Fprintf(os.Stderr, "usage: %s [server|client] addr\n", os.Args[0])
os.Exit(2)
}
role, addr := os.Args[1], os.Args[2]
switch role {
case "server":
http.HandleFunc("/hello", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain")
fmt.Fprintf(w, "hello from %s — you said %s %s\n", addr, r.Method, r.URL.Path)
})
fmt.Printf("zap-http server listening on %s\n", addr)
if err := zaphttp.ListenAndServe(addr, nil); err != nil {
fmt.Fprintln(os.Stderr, "server:", err)
os.Exit(1)
}
case "client":
client := &http.Client{Transport: zaphttp.NewTransport(addr)}
resp, err := client.Get("http://" + addr + "/hello")
if err != nil {
fmt.Fprintln(os.Stderr, "client:", err)
os.Exit(1)
}
defer resp.Body.Close()
fmt.Printf("← %s\n", resp.Status)
io.Copy(os.Stdout, resp.Body)
default:
fmt.Fprintln(os.Stderr, "role must be 'server' or 'client'")
os.Exit(2)
}
}
+10
View File
@@ -0,0 +1,10 @@
module github.com/zap-proto/http
go 1.23
require capnproto.org/go/capnp/v3 v3.0.1-alpha.2
require (
github.com/colega/zeropool v0.0.0-20230505084239-6fb4a4f75381 // indirect
golang.org/x/sync v0.7.0 // indirect
)
+20
View File
@@ -0,0 +1,20 @@
capnproto.org/go/capnp/v3 v3.0.1-alpha.2 h1:W/cf+XEArUSwcBBE/9wS2NpWDkM5NLQOjmzEiHZpYi0=
capnproto.org/go/capnp/v3 v3.0.1-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/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/philhofer/fwd v1.1.2 h1:bnDivRJ1EWPjUIRXV5KfORO897HTbpFAQddBdE8t7Gw=
github.com/philhofer/fwd v1.1.2/go.mod h1:qkPdfjR2SIEbspLqpe1tO4n5yICnr2DY7mqEx2tUTP0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
github.com/tinylib/msgp v1.1.9 h1:SHf3yoO2sGA0veCJeCBYLHuttAVFHGm2RHgNodW7wQU=
github.com/tinylib/msgp v1.1.9/go.mod h1:BCXGB54lDD8qUEPmiG0cQQUANC4IUQyB2ItS2UDlO/k=
github.com/tj/assert v0.0.3 h1:Df/BlaZ20mq6kuai7f5z2TvPFiwC3xaWJSDQNiIS3Rk=
github.com/tj/assert v0.0.3/go.mod h1:Ne6X72Q+TB1AteidzQncjw9PabbMp4PBMZ1k+vd1Pvk=
golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+665
View File
@@ -0,0 +1,665 @@
// Code generated by capnpc-go. DO NOT EDIT.
package zaphttp_capnp
import (
capnp "capnproto.org/go/capnp/v3"
text "capnproto.org/go/capnp/v3/encoding/text"
schemas "capnproto.org/go/capnp/v3/schemas"
strconv "strconv"
)
type Header capnp.Struct
// Header_TypeID is the unique identifier for the type Header.
const Header_TypeID = 0x82ef9c51fc9be8ce
func NewHeader(s *capnp.Segment) (Header, error) {
st, err := capnp.NewStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2})
return Header(st), err
}
func NewRootHeader(s *capnp.Segment) (Header, error) {
st, err := capnp.NewRootStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2})
return Header(st), err
}
func ReadRootHeader(msg *capnp.Message) (Header, error) {
root, err := msg.Root()
return Header(root.Struct()), err
}
func (s Header) String() string {
str, _ := text.Marshal(0x82ef9c51fc9be8ce, capnp.Struct(s))
return str
}
func (s Header) EncodeAsPtr(seg *capnp.Segment) capnp.Ptr {
return capnp.Struct(s).EncodeAsPtr(seg)
}
func (Header) DecodeFromPtr(p capnp.Ptr) Header {
return Header(capnp.Struct{}.DecodeFromPtr(p))
}
func (s Header) ToPtr() capnp.Ptr {
return capnp.Struct(s).ToPtr()
}
func (s Header) IsValid() bool {
return capnp.Struct(s).IsValid()
}
func (s Header) Message() *capnp.Message {
return capnp.Struct(s).Message()
}
func (s Header) Segment() *capnp.Segment {
return capnp.Struct(s).Segment()
}
func (s Header) Name() (string, error) {
p, err := capnp.Struct(s).Ptr(0)
return p.Text(), err
}
func (s Header) HasName() bool {
return capnp.Struct(s).HasPtr(0)
}
func (s Header) NameBytes() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(0)
return p.TextBytes(), err
}
func (s Header) SetName(v string) error {
return capnp.Struct(s).SetText(0, v)
}
func (s Header) Values() (capnp.TextList, error) {
p, err := capnp.Struct(s).Ptr(1)
return capnp.TextList(p.List()), err
}
func (s Header) HasValues() bool {
return capnp.Struct(s).HasPtr(1)
}
func (s Header) SetValues(v capnp.TextList) error {
return capnp.Struct(s).SetPtr(1, v.ToPtr())
}
// NewValues sets the values field to a newly
// allocated capnp.TextList, preferring placement in s's segment.
func (s Header) NewValues(n int32) (capnp.TextList, error) {
l, err := capnp.NewTextList(capnp.Struct(s).Segment(), n)
if err != nil {
return capnp.TextList{}, err
}
err = capnp.Struct(s).SetPtr(1, l.ToPtr())
return l, err
}
// Header_List is a list of Header.
type Header_List = capnp.StructList[Header]
// NewHeader creates a new list of Header.
func NewHeader_List(s *capnp.Segment, sz int32) (Header_List, error) {
l, err := capnp.NewCompositeList(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2}, sz)
return capnp.StructList[Header](l), err
}
// Header_Future is a wrapper for a Header promised by a client call.
type Header_Future struct{ *capnp.Future }
func (f Header_Future) Struct() (Header, error) {
p, err := f.Future.Ptr()
return Header(p.Struct()), err
}
type Request capnp.Struct
// Request_TypeID is the unique identifier for the type Request.
const Request_TypeID = 0xb2063b47a28c074a
func NewRequest(s *capnp.Segment) (Request, error) {
st, err := capnp.NewStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 6})
return Request(st), err
}
func NewRootRequest(s *capnp.Segment) (Request, error) {
st, err := capnp.NewRootStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 6})
return Request(st), err
}
func ReadRootRequest(msg *capnp.Message) (Request, error) {
root, err := msg.Root()
return Request(root.Struct()), err
}
func (s Request) String() string {
str, _ := text.Marshal(0xb2063b47a28c074a, capnp.Struct(s))
return str
}
func (s Request) EncodeAsPtr(seg *capnp.Segment) capnp.Ptr {
return capnp.Struct(s).EncodeAsPtr(seg)
}
func (Request) DecodeFromPtr(p capnp.Ptr) Request {
return Request(capnp.Struct{}.DecodeFromPtr(p))
}
func (s Request) ToPtr() capnp.Ptr {
return capnp.Struct(s).ToPtr()
}
func (s Request) IsValid() bool {
return capnp.Struct(s).IsValid()
}
func (s Request) Message() *capnp.Message {
return capnp.Struct(s).Message()
}
func (s Request) Segment() *capnp.Segment {
return capnp.Struct(s).Segment()
}
func (s Request) Method() (string, error) {
p, err := capnp.Struct(s).Ptr(0)
return p.Text(), err
}
func (s Request) HasMethod() bool {
return capnp.Struct(s).HasPtr(0)
}
func (s Request) MethodBytes() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(0)
return p.TextBytes(), err
}
func (s Request) SetMethod(v string) error {
return capnp.Struct(s).SetText(0, v)
}
func (s Request) Target() (string, error) {
p, err := capnp.Struct(s).Ptr(1)
return p.Text(), err
}
func (s Request) HasTarget() bool {
return capnp.Struct(s).HasPtr(1)
}
func (s Request) TargetBytes() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(1)
return p.TextBytes(), err
}
func (s Request) SetTarget(v string) error {
return capnp.Struct(s).SetText(1, v)
}
func (s Request) Proto() (string, error) {
p, err := capnp.Struct(s).Ptr(2)
return p.Text(), err
}
func (s Request) HasProto() bool {
return capnp.Struct(s).HasPtr(2)
}
func (s Request) ProtoBytes() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(2)
return p.TextBytes(), err
}
func (s Request) SetProto(v string) error {
return capnp.Struct(s).SetText(2, v)
}
func (s Request) Headers() (Header_List, error) {
p, err := capnp.Struct(s).Ptr(3)
return Header_List(p.List()), err
}
func (s Request) HasHeaders() bool {
return capnp.Struct(s).HasPtr(3)
}
func (s Request) SetHeaders(v Header_List) error {
return capnp.Struct(s).SetPtr(3, v.ToPtr())
}
// NewHeaders sets the headers field to a newly
// allocated Header_List, preferring placement in s's segment.
func (s Request) NewHeaders(n int32) (Header_List, error) {
l, err := NewHeader_List(capnp.Struct(s).Segment(), n)
if err != nil {
return Header_List{}, err
}
err = capnp.Struct(s).SetPtr(3, l.ToPtr())
return l, err
}
func (s Request) Body() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(4)
return []byte(p.Data()), err
}
func (s Request) HasBody() bool {
return capnp.Struct(s).HasPtr(4)
}
func (s Request) SetBody(v []byte) error {
return capnp.Struct(s).SetData(4, v)
}
func (s Request) Trailer() (Header_List, error) {
p, err := capnp.Struct(s).Ptr(5)
return Header_List(p.List()), err
}
func (s Request) HasTrailer() bool {
return capnp.Struct(s).HasPtr(5)
}
func (s Request) SetTrailer(v Header_List) error {
return capnp.Struct(s).SetPtr(5, v.ToPtr())
}
// NewTrailer sets the trailer field to a newly
// allocated Header_List, preferring placement in s's segment.
func (s Request) NewTrailer(n int32) (Header_List, error) {
l, err := NewHeader_List(capnp.Struct(s).Segment(), n)
if err != nil {
return Header_List{}, err
}
err = capnp.Struct(s).SetPtr(5, l.ToPtr())
return l, err
}
// Request_List is a list of Request.
type Request_List = capnp.StructList[Request]
// NewRequest creates a new list of Request.
func NewRequest_List(s *capnp.Segment, sz int32) (Request_List, error) {
l, err := capnp.NewCompositeList(s, capnp.ObjectSize{DataSize: 0, PointerCount: 6}, sz)
return capnp.StructList[Request](l), err
}
// Request_Future is a wrapper for a Request promised by a client call.
type Request_Future struct{ *capnp.Future }
func (f Request_Future) Struct() (Request, error) {
p, err := f.Future.Ptr()
return Request(p.Struct()), err
}
type Response capnp.Struct
// Response_TypeID is the unique identifier for the type Response.
const Response_TypeID = 0xab7e8c4a7b2a9bff
func NewResponse(s *capnp.Segment) (Response, error) {
st, err := capnp.NewStruct(s, capnp.ObjectSize{DataSize: 8, PointerCount: 5})
return Response(st), err
}
func NewRootResponse(s *capnp.Segment) (Response, error) {
st, err := capnp.NewRootStruct(s, capnp.ObjectSize{DataSize: 8, PointerCount: 5})
return Response(st), err
}
func ReadRootResponse(msg *capnp.Message) (Response, error) {
root, err := msg.Root()
return Response(root.Struct()), err
}
func (s Response) String() string {
str, _ := text.Marshal(0xab7e8c4a7b2a9bff, capnp.Struct(s))
return str
}
func (s Response) EncodeAsPtr(seg *capnp.Segment) capnp.Ptr {
return capnp.Struct(s).EncodeAsPtr(seg)
}
func (Response) DecodeFromPtr(p capnp.Ptr) Response {
return Response(capnp.Struct{}.DecodeFromPtr(p))
}
func (s Response) ToPtr() capnp.Ptr {
return capnp.Struct(s).ToPtr()
}
func (s Response) IsValid() bool {
return capnp.Struct(s).IsValid()
}
func (s Response) Message() *capnp.Message {
return capnp.Struct(s).Message()
}
func (s Response) Segment() *capnp.Segment {
return capnp.Struct(s).Segment()
}
func (s Response) Status() uint16 {
return capnp.Struct(s).Uint16(0)
}
func (s Response) SetStatus(v uint16) {
capnp.Struct(s).SetUint16(0, v)
}
func (s Response) Reason() (string, error) {
p, err := capnp.Struct(s).Ptr(0)
return p.Text(), err
}
func (s Response) HasReason() bool {
return capnp.Struct(s).HasPtr(0)
}
func (s Response) ReasonBytes() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(0)
return p.TextBytes(), err
}
func (s Response) SetReason(v string) error {
return capnp.Struct(s).SetText(0, v)
}
func (s Response) Proto() (string, error) {
p, err := capnp.Struct(s).Ptr(1)
return p.Text(), err
}
func (s Response) HasProto() bool {
return capnp.Struct(s).HasPtr(1)
}
func (s Response) ProtoBytes() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(1)
return p.TextBytes(), err
}
func (s Response) SetProto(v string) error {
return capnp.Struct(s).SetText(1, v)
}
func (s Response) Headers() (Header_List, error) {
p, err := capnp.Struct(s).Ptr(2)
return Header_List(p.List()), err
}
func (s Response) HasHeaders() bool {
return capnp.Struct(s).HasPtr(2)
}
func (s Response) SetHeaders(v Header_List) error {
return capnp.Struct(s).SetPtr(2, v.ToPtr())
}
// NewHeaders sets the headers field to a newly
// allocated Header_List, preferring placement in s's segment.
func (s Response) NewHeaders(n int32) (Header_List, error) {
l, err := NewHeader_List(capnp.Struct(s).Segment(), n)
if err != nil {
return Header_List{}, err
}
err = capnp.Struct(s).SetPtr(2, l.ToPtr())
return l, err
}
func (s Response) Body() ([]byte, error) {
p, err := capnp.Struct(s).Ptr(3)
return []byte(p.Data()), err
}
func (s Response) HasBody() bool {
return capnp.Struct(s).HasPtr(3)
}
func (s Response) SetBody(v []byte) error {
return capnp.Struct(s).SetData(3, v)
}
func (s Response) Trailer() (Header_List, error) {
p, err := capnp.Struct(s).Ptr(4)
return Header_List(p.List()), err
}
func (s Response) HasTrailer() bool {
return capnp.Struct(s).HasPtr(4)
}
func (s Response) SetTrailer(v Header_List) error {
return capnp.Struct(s).SetPtr(4, v.ToPtr())
}
// NewTrailer sets the trailer field to a newly
// allocated Header_List, preferring placement in s's segment.
func (s Response) NewTrailer(n int32) (Header_List, error) {
l, err := NewHeader_List(capnp.Struct(s).Segment(), n)
if err != nil {
return Header_List{}, err
}
err = capnp.Struct(s).SetPtr(4, l.ToPtr())
return l, err
}
// Response_List is a list of Response.
type Response_List = capnp.StructList[Response]
// NewResponse creates a new list of Response.
func NewResponse_List(s *capnp.Segment, sz int32) (Response_List, error) {
l, err := capnp.NewCompositeList(s, capnp.ObjectSize{DataSize: 8, PointerCount: 5}, sz)
return capnp.StructList[Response](l), err
}
// Response_Future is a wrapper for a Response promised by a client call.
type Response_Future struct{ *capnp.Future }
func (f Response_Future) Struct() (Response, error) {
p, err := f.Future.Ptr()
return Response(p.Struct()), err
}
type Frame capnp.Struct
type Frame_Which uint16
const (
Frame_Which_request Frame_Which = 0
Frame_Which_response Frame_Which = 1
)
func (w Frame_Which) String() string {
const s = "requestresponse"
switch w {
case Frame_Which_request:
return s[0:7]
case Frame_Which_response:
return s[7:15]
}
return "Frame_Which(" + strconv.FormatUint(uint64(w), 10) + ")"
}
// Frame_TypeID is the unique identifier for the type Frame.
const Frame_TypeID = 0xda62522904e66ee9
func NewFrame(s *capnp.Segment) (Frame, error) {
st, err := capnp.NewStruct(s, capnp.ObjectSize{DataSize: 8, PointerCount: 1})
return Frame(st), err
}
func NewRootFrame(s *capnp.Segment) (Frame, error) {
st, err := capnp.NewRootStruct(s, capnp.ObjectSize{DataSize: 8, PointerCount: 1})
return Frame(st), err
}
func ReadRootFrame(msg *capnp.Message) (Frame, error) {
root, err := msg.Root()
return Frame(root.Struct()), err
}
func (s Frame) String() string {
str, _ := text.Marshal(0xda62522904e66ee9, capnp.Struct(s))
return str
}
func (s Frame) EncodeAsPtr(seg *capnp.Segment) capnp.Ptr {
return capnp.Struct(s).EncodeAsPtr(seg)
}
func (Frame) DecodeFromPtr(p capnp.Ptr) Frame {
return Frame(capnp.Struct{}.DecodeFromPtr(p))
}
func (s Frame) ToPtr() capnp.Ptr {
return capnp.Struct(s).ToPtr()
}
func (s Frame) Which() Frame_Which {
return Frame_Which(capnp.Struct(s).Uint16(0))
}
func (s Frame) IsValid() bool {
return capnp.Struct(s).IsValid()
}
func (s Frame) Message() *capnp.Message {
return capnp.Struct(s).Message()
}
func (s Frame) Segment() *capnp.Segment {
return capnp.Struct(s).Segment()
}
func (s Frame) Request() (Request, error) {
if capnp.Struct(s).Uint16(0) != 0 {
panic("Which() != request")
}
p, err := capnp.Struct(s).Ptr(0)
return Request(p.Struct()), err
}
func (s Frame) HasRequest() bool {
if capnp.Struct(s).Uint16(0) != 0 {
return false
}
return capnp.Struct(s).HasPtr(0)
}
func (s Frame) SetRequest(v Request) error {
capnp.Struct(s).SetUint16(0, 0)
return capnp.Struct(s).SetPtr(0, capnp.Struct(v).ToPtr())
}
// NewRequest sets the request field to a newly
// allocated Request struct, preferring placement in s's segment.
func (s Frame) NewRequest() (Request, error) {
capnp.Struct(s).SetUint16(0, 0)
ss, err := NewRequest(capnp.Struct(s).Segment())
if err != nil {
return Request{}, err
}
err = capnp.Struct(s).SetPtr(0, capnp.Struct(ss).ToPtr())
return ss, err
}
func (s Frame) Response() (Response, error) {
if capnp.Struct(s).Uint16(0) != 1 {
panic("Which() != response")
}
p, err := capnp.Struct(s).Ptr(0)
return Response(p.Struct()), err
}
func (s Frame) HasResponse() bool {
if capnp.Struct(s).Uint16(0) != 1 {
return false
}
return capnp.Struct(s).HasPtr(0)
}
func (s Frame) SetResponse(v Response) error {
capnp.Struct(s).SetUint16(0, 1)
return capnp.Struct(s).SetPtr(0, capnp.Struct(v).ToPtr())
}
// NewResponse sets the response field to a newly
// allocated Response struct, preferring placement in s's segment.
func (s Frame) NewResponse() (Response, error) {
capnp.Struct(s).SetUint16(0, 1)
ss, err := NewResponse(capnp.Struct(s).Segment())
if err != nil {
return Response{}, err
}
err = capnp.Struct(s).SetPtr(0, capnp.Struct(ss).ToPtr())
return ss, err
}
// Frame_List is a list of Frame.
type Frame_List = capnp.StructList[Frame]
// NewFrame creates a new list of Frame.
func NewFrame_List(s *capnp.Segment, sz int32) (Frame_List, error) {
l, err := capnp.NewCompositeList(s, capnp.ObjectSize{DataSize: 8, PointerCount: 1}, sz)
return capnp.StructList[Frame](l), err
}
// Frame_Future is a wrapper for a Frame promised by a client call.
type Frame_Future struct{ *capnp.Future }
func (f Frame_Future) Struct() (Frame, error) {
p, err := f.Future.Ptr()
return Frame(p.Struct()), err
}
func (p Frame_Future) Request() Request_Future {
return Request_Future{Future: p.Future.Field(0, nil)}
}
func (p Frame_Future) Response() Response_Future {
return Response_Future{Future: p.Future.Field(0, nil)}
}
const schema_a3d4b2e80f1c5a7e = "x\xda\xc4\x93\xbdk\x14Q\x14\xc5\xcfyo&k\x91" +
"\xcd\xecc\xb6\xb0\x10baa\x82\x91D\x1b\x89\xc5J" +
"\xf03 \xe4\xa5\x11ld\xe2>\\!\xbb3\xce\xcc" +
"\xba\xa8\x90\x05\xeb\x94\x16\x86$\x90\xf8\x01\xa2\xddba" +
"\xc4\x80\xda\xd8Y\xa5\xf2\x0f\x90X\xf9\x0f\x88#o?" +
"%nj\xbb\x99;g\xeey\xefw\xcf\x9dn\xf0\x82" +
"3\x93\xff\" \xf4qw$\xfb\xba\xbf\xfeKo\xfc" +
"|\x0c\xe51[\xb9y\xcc\xdbo\xed=\x87+r\x80" +
"\x7f\x8d-_\xd3>]g\x03\xcc\xb2\xf5\xc9G\xf3\xab" +
"+o\xa0=\xfe-v\xadd\x87\x9f\xfdOV|v" +
"\x977\x08f\xf3\xb9\xd5gW\xce\x8f\xb4\x0et\x1e\xb1" +
"\xe2\x09\xf9\xde\x9f\x91V<%\xdb\xe2\x1f\xb5\xef\xce\xc4" +
"\xe2\xd2\xb7\x03\xad/1'\x00\x7f\xcdy\xedo;\xf6" +
"\xc7M\xa7\x81\xa9\xeca\x10\xdd\xaa\xa4i\xc4\xd3\xb7\x83" +
"\xa8\x16\xcd^\x1d7A\xd9\xc4\x0b\xa4>\"\x1d\xc0!" +
"\xa0&&\x01}BRO\x0b*\xb2H[\x9c\x9a\x05" +
"\xf4II}Q\xd0\xab\x05U\xc3Q\x08\x8e\x82\xa5\xfb" +
"\xc1r\xdd$\x1c\x03\x17$\xdb\xd51\xb0\xef$\xbaN" +
"\x8b&\x89\xc2Zb`\xbd\x8e\xf6\xbd\xd6l\xdb'\x92" +
"zK\xb0g\xb5ikO%\xf5\x0bA%X\xa4\x00" +
"\xd4\xf6\x19@oH\xeaW\x82J\x8a\"%\xa0^\xce" +
"\x01zKR\x7f\x10T\x8e,\xd2\x01\xd4\x8e=\xfe[" +
"I\xfdQP\xb9N\x91.\xa0v\xad\xf2\x9d\xa4\xde\x13" +
",%i\x90\xd6\x13\xe6 \x98\x03K\xb1\x09\x92\xb0\xd6" +
"\xbb\xcfx\x14\x87i\xd8{kV\xda\x80\xfa\xd7+\x0c" +
"\xe6\x0e\xda\xa2\xb7\x14\x96\x1f0\x0f\xc1<\xd8L\xe3\xe0" +
"\xee\xb2\x89\x0fU\xff3\x80\xc5\x92\xb9W7Iz8" +
"\x155\x1c\x8b\x18\x86E\x0e\xc3\xe2\x0c\xc3\xe2\x0e\xc1R" +
"5i%,\xf7\xe7\x9a\x06\xf1\x1d\x93\xfe\x1f,\x97\xbd" +
"8\xa8\x9an,G\xb3\xac\x93\xcb\xb9A.\xf3\xfc\x9d" +
"u\x839\x0f\xe8S\x92\xfa\x9c`3\xee\xd0da\xb0" +
"D \x0b`\x16\xf7\xf2\x07\xb00X\xc8\xce\xd7?\x01" +
"\x00\x00\xff\xff\xacC\xe6\x10"
func RegisterSchema(reg *schemas.Registry) {
reg.Register(&schemas.Schema{
String: schema_a3d4b2e80f1c5a7e,
Nodes: []uint64{
0x82ef9c51fc9be8ce,
0xab7e8c4a7b2a9bff,
0xb2063b47a28c074a,
0xda62522904e66ee9,
},
Compressed: true,
})
}
+60
View File
@@ -0,0 +1,60 @@
# zap-http — HTTP request/response semantics over the ZAP transport.
#
# The wire is one ZAP frame per HTTP message. A request frame carries
# the request line + headers + body; a response frame carries status +
# headers + body. Both sides may carry trailers (RFC 9110 §6.5).
#
# Streaming (chunked-style transfer) is reserved for a follow-up
# protocol slice and not represented here. v0.1 assumes the entire body
# fits in one frame; the transport (zap-proto/go) negotiates the
# maximum frame size on connection setup.
@0xa3d4b2e80f1c5a7e;
using Go = import "/go.capnp";
$Go.package("zaphttp_capnp");
$Go.import("github.com/zap-proto/http/internal/capnp");
# Header is one HTTP header name and its values. RFC 9110 allows the
# same name to appear multiple times; we represent that as a list of
# values rather than concatenating with commas, so callers see what
# was actually sent.
struct Header {
name @0 :Text;
values @1 :List(Text);
}
# Request is one HTTP request. `target` is the request-target as it
# would appear on an HTTP/1.1 request line — origin-form (`/path?q=v`),
# absolute-form (`https://host/path`), authority-form (`host:port` for
# CONNECT), or asterisk-form (`*` for OPTIONS).
struct Request {
method @0 :Text;
target @1 :Text;
proto @2 :Text;
headers @3 :List(Header);
body @4 :Data;
trailer @5 :List(Header);
}
# Response is one HTTP response. `status` is the numeric status code
# (200, 404, 503, …); `reason` is the optional reason phrase; the
# transport ignores the phrase for routing but preserves it for any
# caller that still parses it.
struct Response {
status @0 :UInt16;
reason @1 :Text;
proto @2 :Text;
headers @3 :List(Header);
body @4 :Data;
trailer @5 :List(Header);
}
# Frame is the union the wire actually carries. Some sub-protocols may
# add Notify or Stream variants in future; today there are exactly two.
struct Frame {
union {
request @0 :Request;
response @1 :Response;
}
}
+234
View File
@@ -0,0 +1,234 @@
// Server — net/http.Handler-compatible server speaking ZAP-HTTP.
//
// Existing handlers work unchanged:
//
// mux := http.NewServeMux()
// mux.HandleFunc("/healthz", okHandler)
// zaphttp.ListenAndServe(":9999", mux)
//
// The server reads ZAP-HTTP frames off each accepted connection,
// reconstructs *http.Request, dispatches to the handler, captures the
// handler's response via a buffering ResponseWriter, and writes one
// response frame back. Connections are kept alive across requests.
package zaphttp
import (
"bufio"
"bytes"
"context"
"errors"
"fmt"
"io"
"log"
"net"
"net/http"
"strconv"
"sync"
"time"
)
// Server is a ZAP-HTTP server. Zero-value is usable; common knobs
// (Addr, Handler, ReadTimeout, …) mirror net/http.Server.
type Server struct {
Addr string // ":9999" if empty
Handler http.Handler // http.DefaultServeMux if nil
ReadTimeout time.Duration // 0 means no timeout
WriteTimeout time.Duration
IdleTimeout time.Duration
MaxHeaderBytes int // not enforced today; placeholder for parity
mu sync.Mutex
listener net.Listener
closed bool
}
// ListenAndServe is the convenience equivalent of net/http.ListenAndServe.
func ListenAndServe(addr string, handler http.Handler) error {
s := &Server{Addr: addr, Handler: handler}
return s.ListenAndServe()
}
// ListenAndServe binds Addr and serves until Close is called or a
// fatal accept error occurs.
func (s *Server) ListenAndServe() error {
addr := s.Addr
if addr == "" {
addr = ":9999"
}
ln, err := net.Listen("tcp", addr)
if err != nil {
return err
}
return s.Serve(ln)
}
// Serve accepts connections on ln and serves each one in its own
// goroutine. Returns when the listener is closed.
func (s *Server) Serve(ln net.Listener) error {
s.mu.Lock()
s.listener = ln
s.mu.Unlock()
handler := s.Handler
if handler == nil {
handler = http.DefaultServeMux
}
for {
conn, err := ln.Accept()
if err != nil {
s.mu.Lock()
closed := s.closed
s.mu.Unlock()
if closed {
return nil
}
var nerr net.Error
if errors.As(err, &nerr) && nerr.Timeout() {
continue
}
return err
}
go s.serveConn(conn, handler)
}
}
func (s *Server) serveConn(conn net.Conn, handler http.Handler) {
defer conn.Close()
br := bufio.NewReader(conn)
for {
if s.IdleTimeout > 0 {
_ = conn.SetReadDeadline(time.Now().Add(s.IdleTimeout))
}
raw, err := readFrame(br)
if err != nil {
if !errors.Is(err, io.EOF) && !isClosedConn(err) {
log.Printf("zaphttp: read frame: %v", err)
}
return
}
if s.ReadTimeout > 0 {
_ = conn.SetReadDeadline(time.Now().Add(s.ReadTimeout))
}
req, err := UnmarshalRequest(raw)
if err != nil {
s.writeError(conn, http.StatusBadRequest, "malformed request frame: "+err.Error())
return
}
req.RemoteAddr = conn.RemoteAddr().String()
// Wire a buffering ResponseWriter. The handler runs to
// completion and we then marshal the captured state into a
// single response frame. v0.2 will swap this for a streaming
// writer once the transport supports it.
rw := &bufferingResponseWriter{
header: make(http.Header),
body: &bytes.Buffer{},
}
ctx := context.Background()
req = req.WithContext(ctx)
// Defer-recover so a handler panic returns 500 rather than
// dropping the connection silently.
func() {
defer func() {
if r := recover(); r != nil {
if !rw.wroteHeader {
rw.WriteHeader(http.StatusInternalServerError)
}
log.Printf("zaphttp: handler panic on %s %s: %v", req.Method, req.URL.RequestURI(), r)
}
}()
handler.ServeHTTP(rw, req)
}()
if !rw.wroteHeader {
rw.WriteHeader(http.StatusOK)
}
// Auto-fill Content-Length when the handler hasn't.
if rw.header.Get("Content-Length") == "" {
rw.header.Set("Content-Length", strconv.Itoa(rw.body.Len()))
}
resp := &http.Response{
StatusCode: rw.status,
Status: fmt.Sprintf("%d %s", rw.status, http.StatusText(rw.status)),
Header: rw.header,
Body: io.NopCloser(rw.body),
}
respBytes, err := MarshalResponse(resp)
if err != nil {
log.Printf("zaphttp: marshal response: %v", err)
return
}
if s.WriteTimeout > 0 {
_ = conn.SetWriteDeadline(time.Now().Add(s.WriteTimeout))
}
if err := writeFrame(conn, respBytes); err != nil {
if !isClosedConn(err) {
log.Printf("zaphttp: write frame: %v", err)
}
return
}
}
}
func (s *Server) writeError(w io.Writer, status int, msg string) {
resp := &http.Response{
StatusCode: status,
Status: fmt.Sprintf("%d %s", status, http.StatusText(status)),
Header: http.Header{"Content-Type": []string{"text/plain; charset=utf-8"}},
Body: io.NopCloser(bytes.NewBufferString(msg)),
}
frame, err := MarshalResponse(resp)
if err != nil {
return
}
_ = writeFrame(w, frame)
}
// Close stops the listener; in-flight handlers are not interrupted.
func (s *Server) Close() error {
s.mu.Lock()
defer s.mu.Unlock()
s.closed = true
if s.listener == nil {
return nil
}
return s.listener.Close()
}
// ---- bufferingResponseWriter ----
type bufferingResponseWriter struct {
header http.Header
body *bytes.Buffer
status int
wroteHeader bool
}
func (w *bufferingResponseWriter) Header() http.Header { return w.header }
func (w *bufferingResponseWriter) Write(b []byte) (int, error) {
if !w.wroteHeader {
w.WriteHeader(http.StatusOK)
}
return w.body.Write(b)
}
func (w *bufferingResponseWriter) WriteHeader(status int) {
if w.wroteHeader {
return
}
w.status = status
w.wroteHeader = true
}
func isClosedConn(err error) bool {
if err == nil {
return false
}
return errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF)
}
+65
View File
@@ -0,0 +1,65 @@
// Transport — length-prefixed framing over an io.ReadWriteCloser.
//
// v0.1 uses a 4-byte big-endian length prefix on each frame; v0.2
// will swap this for the full ZAP transport (X-Wing PQ KEM handshake
// + AEAD framing + multi-stream multiplexing) once zap-proto/go ships
// it. The marshal/unmarshal API in wire.go is stable across that
// change — only the bytes flowing on the socket change shape.
//
// Each ZAP-HTTP exchange is one request frame followed by one
// response frame on the same connection. Connections may be reused
// across exchanges (HTTP/1.1-style keep-alive); the transport itself
// does not pipeline.
package zaphttp
import (
"bufio"
"encoding/binary"
"fmt"
"io"
)
// MaxFrameSize bounds an inbound frame. The wire format leaves room
// for larger frames; the limit here defends a server against a
// malicious peer announcing a multi-gigabyte length prefix.
const MaxFrameSize = 64 << 20 // 64 MiB
// readFrame reads one length-prefixed frame from r. The returned
// byte slice is owned by the caller; nil + io.EOF indicates the peer
// closed cleanly between frames.
func readFrame(r *bufio.Reader) ([]byte, error) {
var hdr [4]byte
if _, err := io.ReadFull(r, hdr[:]); err != nil {
return nil, err
}
n := binary.BigEndian.Uint32(hdr[:])
if n == 0 {
return nil, fmt.Errorf("zaphttp: zero-length frame")
}
if n > MaxFrameSize {
return nil, fmt.Errorf("zaphttp: frame size %d exceeds MaxFrameSize=%d", n, MaxFrameSize)
}
buf := make([]byte, n)
if _, err := io.ReadFull(r, buf); err != nil {
return nil, fmt.Errorf("zaphttp: short frame: %w", err)
}
return buf, nil
}
// writeFrame writes one length-prefixed frame to w.
func writeFrame(w io.Writer, frame []byte) error {
if len(frame) == 0 {
return fmt.Errorf("zaphttp: refusing to write zero-length frame")
}
if uint64(len(frame)) > MaxFrameSize {
return fmt.Errorf("zaphttp: frame size %d exceeds MaxFrameSize=%d", len(frame), MaxFrameSize)
}
var hdr [4]byte
binary.BigEndian.PutUint32(hdr[:], uint32(len(frame)))
if _, err := w.Write(hdr[:]); err != nil {
return err
}
_, err := w.Write(frame)
return err
}
+366
View File
@@ -0,0 +1,366 @@
// Package zaphttp implements HTTP request/response semantics over the
// ZAP transport. Existing net/http callers and handlers work
// unmodified — the package adapts the standard library types to and
// from Cap'n Proto-encoded ZAP frames.
//
// The wire format is documented in schema/zap_http.capnp; the Go types
// in internal/capnp are generated from it.
//
// What's covered today (v0.1):
// - Request method, target, headers, body
// - Response status, reason, headers, body
// - Trailers (read and write)
//
// What's not yet covered:
// - Streaming bodies — the entire body must fit in one frame; the
// transport negotiates a max frame size on connection setup.
// - HTTP/2 server push semantics; HTTP/3 priorities.
// - WebSocket / SSE upgrade paths; those land in zap-proto/ws.
//
// This file holds the marshal/unmarshal between *http.Request /
// http.Response and ZAP frames. It depends only on the standard
// library plus Cap'n Proto. The transport layer (TCP framing today,
// PQ-handshake-wrapped ZAP transport in v0.2) lives in transport.go.
package zaphttp
import (
"bytes"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"capnproto.org/go/capnp/v3"
zhcapnp "github.com/zap-proto/http/internal/capnp"
)
// MarshalRequest serializes an *http.Request into a ZAP frame. The
// returned bytes are framed and ready for transport.Write — the
// transport layer prepends the length prefix.
//
// The body is fully read and consumed; callers that need to reuse the
// request must replace req.Body with a fresh reader after this call.
func MarshalRequest(req *http.Request) ([]byte, error) {
body, err := readAllBody(req.Body)
if err != nil {
return nil, fmt.Errorf("zaphttp: read body: %w", err)
}
msg, seg, err := capnp.NewMessage(capnp.SingleSegment(nil))
if err != nil {
return nil, err
}
frame, err := zhcapnp.NewRootFrame(seg)
if err != nil {
return nil, err
}
r, err := frame.NewRequest()
if err != nil {
return nil, err
}
if err := r.SetMethod(orDefault(req.Method, http.MethodGet)); err != nil {
return nil, err
}
target := req.RequestURI
if target == "" && req.URL != nil {
target = req.URL.RequestURI()
}
if err := r.SetTarget(target); err != nil {
return nil, err
}
if err := r.SetProto(orDefault(req.Proto, "ZAP-HTTP/1.0")); err != nil {
return nil, err
}
if err := writeHeaders(seg, req.Header, r.SetHeaders); err != nil {
return nil, err
}
if len(body) > 0 {
if err := r.SetBody(body); err != nil {
return nil, err
}
}
if len(req.Trailer) > 0 {
if err := writeHeaders(seg, req.Trailer, r.SetTrailer); err != nil {
return nil, err
}
}
return msg.Marshal()
}
// UnmarshalRequest reconstructs an *http.Request from a ZAP frame. The
// returned request has Body set to a bytes.Reader over the frame's body
// bytes; the request URL is parsed from target. Host is taken from the
// request's Host header (RFC 9110 §7.2).
func UnmarshalRequest(b []byte) (*http.Request, error) {
msg, err := capnp.Unmarshal(b)
if err != nil {
return nil, err
}
frame, err := zhcapnp.ReadRootFrame(msg)
if err != nil {
return nil, err
}
if frame.Which() != zhcapnp.Frame_Which_request {
return nil, fmt.Errorf("zaphttp: frame is %s, want request", frame.Which())
}
r, err := frame.Request()
if err != nil {
return nil, err
}
method, err := r.Method()
if err != nil {
return nil, err
}
target, err := r.Target()
if err != nil {
return nil, err
}
proto, err := r.Proto()
if err != nil {
return nil, err
}
headers, err := readHeaders(r.Headers)
if err != nil {
return nil, err
}
body, err := r.Body()
if err != nil {
return nil, err
}
trailer, err := readHeaders(r.Trailer)
if err != nil {
return nil, err
}
u, err := url.ParseRequestURI(target)
if err != nil {
// Best-effort: keep the raw target so handlers can inspect it,
// but failing here would break callers that only need RequestURI.
u = &url.URL{Path: target}
}
out := &http.Request{
Method: orDefault(method, http.MethodGet),
URL: u,
Proto: orDefault(proto, "ZAP-HTTP/1.0"),
ProtoMajor: 1,
ProtoMinor: 0,
Header: headers,
Body: io.NopCloser(bytes.NewReader(body)),
Host: headers.Get("Host"),
RequestURI: target,
Trailer: trailer,
}
if cl := headers.Get("Content-Length"); cl != "" {
if n, err := strconv.ParseInt(cl, 10, 64); err == nil {
out.ContentLength = n
}
} else {
out.ContentLength = int64(len(body))
}
return out, nil
}
// MarshalResponse serializes an *http.Response into a ZAP frame.
func MarshalResponse(resp *http.Response) ([]byte, error) {
body, err := readAllBody(resp.Body)
if err != nil {
return nil, fmt.Errorf("zaphttp: read body: %w", err)
}
msg, seg, err := capnp.NewMessage(capnp.SingleSegment(nil))
if err != nil {
return nil, err
}
frame, err := zhcapnp.NewRootFrame(seg)
if err != nil {
return nil, err
}
r, err := frame.NewResponse()
if err != nil {
return nil, err
}
if resp.StatusCode == 0 {
r.SetStatus(http.StatusOK)
} else {
r.SetStatus(uint16(resp.StatusCode))
}
reason := resp.Status
if i := strings.IndexByte(reason, ' '); i >= 0 {
reason = reason[i+1:]
}
if err := r.SetReason(reason); err != nil {
return nil, err
}
if err := r.SetProto(orDefault(resp.Proto, "ZAP-HTTP/1.0")); err != nil {
return nil, err
}
if err := writeHeaders(seg, resp.Header, r.SetHeaders); err != nil {
return nil, err
}
if len(body) > 0 {
if err := r.SetBody(body); err != nil {
return nil, err
}
}
if len(resp.Trailer) > 0 {
if err := writeHeaders(seg, resp.Trailer, r.SetTrailer); err != nil {
return nil, err
}
}
return msg.Marshal()
}
// UnmarshalResponse reconstructs an *http.Response from a ZAP frame.
func UnmarshalResponse(b []byte) (*http.Response, error) {
msg, err := capnp.Unmarshal(b)
if err != nil {
return nil, err
}
frame, err := zhcapnp.ReadRootFrame(msg)
if err != nil {
return nil, err
}
if frame.Which() != zhcapnp.Frame_Which_response {
return nil, fmt.Errorf("zaphttp: frame is %s, want response", frame.Which())
}
r, err := frame.Response()
if err != nil {
return nil, err
}
status := int(r.Status())
reason, err := r.Reason()
if err != nil {
return nil, err
}
proto, err := r.Proto()
if err != nil {
return nil, err
}
headers, err := readHeaders(r.Headers)
if err != nil {
return nil, err
}
body, err := r.Body()
if err != nil {
return nil, err
}
trailer, err := readHeaders(r.Trailer)
if err != nil {
return nil, err
}
statusText := reason
if statusText == "" {
statusText = http.StatusText(status)
}
return &http.Response{
Status: fmt.Sprintf("%d %s", status, statusText),
StatusCode: status,
Proto: orDefault(proto, "ZAP-HTTP/1.0"),
ProtoMajor: 1,
ProtoMinor: 0,
Header: headers,
Body: io.NopCloser(bytes.NewReader(body)),
ContentLength: int64(len(body)),
Trailer: trailer,
}, nil
}
// ---- helpers ----
func readAllBody(r io.Reader) ([]byte, error) {
if r == nil {
return nil, nil
}
defer func() {
if c, ok := r.(io.Closer); ok {
_ = c.Close()
}
}()
return io.ReadAll(r)
}
func writeHeaders(seg *capnp.Segment, h http.Header, setter func(zhcapnp.Header_List) error) error {
if len(h) == 0 {
// Allocate an empty list so consumers don't see a default-zero
// pointer that fails to decode.
empty, err := zhcapnp.NewHeader_List(seg, 0)
if err != nil {
return err
}
return setter(empty)
}
list, err := zhcapnp.NewHeader_List(seg, int32(len(h)))
if err != nil {
return err
}
i := 0
for name, values := range h {
hdr := list.At(i)
if err := hdr.SetName(name); err != nil {
return err
}
vlist, err := capnp.NewTextList(seg, int32(len(values)))
if err != nil {
return err
}
for j, v := range values {
if err := vlist.Set(j, v); err != nil {
return err
}
}
if err := hdr.SetValues(vlist); err != nil {
return err
}
i++
}
return setter(list)
}
func readHeaders(getter func() (zhcapnp.Header_List, error)) (http.Header, error) {
list, err := getter()
if err != nil {
return nil, err
}
if list.Len() == 0 {
return http.Header{}, nil
}
out := make(http.Header, list.Len())
for i := 0; i < list.Len(); i++ {
h := list.At(i)
name, err := h.Name()
if err != nil {
return nil, err
}
vlist, err := h.Values()
if err != nil {
return nil, err
}
for j := 0; j < vlist.Len(); j++ {
v, err := vlist.At(j)
if err != nil {
return nil, err
}
out.Add(name, v)
}
}
return out, nil
}
func orDefault(s, fallback string) string {
if s == "" {
return fallback
}
return s
}
+310
View File
@@ -0,0 +1,310 @@
// End-to-end tests for the ZAP-HTTP wire format and the in-process
// server/client pair. Each test stands one server up on a random
// port, drives it through the public Transport, and asserts the
// response shape.
package zaphttp_test
import (
"bytes"
"crypto/rand"
"fmt"
"io"
"net"
"net/http"
"strings"
"sync"
"testing"
"time"
zaphttp "github.com/zap-proto/http"
)
// listen binds an ephemeral port and starts a server in a goroutine.
// Returns the addr and a shutdown function.
func listen(t *testing.T, handler http.Handler) (addr string, shutdown func()) {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
srv := &zaphttp.Server{Handler: handler}
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
if err := srv.Serve(ln); err != nil {
t.Logf("server.Serve returned: %v", err)
}
}()
return ln.Addr().String(), func() {
_ = srv.Close()
wg.Wait()
}
}
// TestRoundTrip_GET — minimum viable exchange.
func TestRoundTrip_GET(t *testing.T) {
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
t.Errorf("server saw method %q want GET", r.Method)
}
if r.URL.Path != "/hello" {
t.Errorf("server saw path %q want /hello", r.URL.Path)
}
w.Header().Set("Content-Type", "text/plain")
_, _ = io.WriteString(w, "hi")
}))
defer stop()
client := &http.Client{Transport: zaphttp.NewTransport(addr)}
resp, err := client.Get("http://" + addr + "/hello")
if err != nil {
t.Fatalf("client.Get: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Errorf("status = %d want 200", resp.StatusCode)
}
body, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("read body: %v", err)
}
if string(body) != "hi" {
t.Errorf("body = %q want %q", body, "hi")
}
if got := resp.Header.Get("Content-Type"); got != "text/plain" {
t.Errorf("Content-Type = %q want text/plain", got)
}
}
// TestRoundTrip_POST_bodyEcho — non-empty request body, echoed back
// as the response body.
func TestRoundTrip_POST_bodyEcho(t *testing.T) {
want := []byte("the quick brown fox jumps over the lazy dog\n")
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got, err := io.ReadAll(r.Body)
if err != nil {
t.Fatalf("server read body: %v", err)
}
if !bytes.Equal(got, want) {
t.Errorf("server got body %q want %q", got, want)
}
w.Header().Set("Content-Type", "application/octet-stream")
_, _ = w.Write(got)
}))
defer stop()
client := &http.Client{Transport: zaphttp.NewTransport(addr)}
resp, err := client.Post("http://"+addr+"/echo", "application/octet-stream", bytes.NewReader(want))
if err != nil {
t.Fatalf("client.Post: %v", err)
}
defer resp.Body.Close()
got, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("read response body: %v", err)
}
if !bytes.Equal(got, want) {
t.Errorf("response body = %q want %q", got, want)
}
}
// TestRoundTrip_largeBody — 4 MiB binary body, well under MaxFrameSize
// but big enough to exercise the framing path.
func TestRoundTrip_largeBody(t *testing.T) {
const size = 4 << 20 // 4 MiB
payload := make([]byte, size)
if _, err := rand.Read(payload); err != nil {
t.Fatalf("rand.Read: %v", err)
}
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got, err := io.ReadAll(r.Body)
if err != nil {
t.Fatalf("server read: %v", err)
}
if !bytes.Equal(got, payload) {
t.Errorf("server received corrupted body (len=%d want %d)", len(got), len(payload))
}
w.WriteHeader(http.StatusOK)
_, _ = w.Write(got)
}))
defer stop()
client := &http.Client{
Timeout: 30 * time.Second,
Transport: zaphttp.NewTransport(addr),
}
resp, err := client.Post("http://"+addr+"/big", "application/octet-stream", bytes.NewReader(payload))
if err != nil {
t.Fatalf("client.Post: %v", err)
}
defer resp.Body.Close()
got, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("read response: %v", err)
}
if !bytes.Equal(got, payload) {
t.Errorf("response body corrupted (len=%d want %d)", len(got), len(payload))
}
}
// TestHeaders_multiValue — RFC 9110 lets one header name carry many
// values; verify we preserve the list shape end-to-end rather than
// joining with commas.
func TestHeaders_multiValue(t *testing.T) {
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got := r.Header.Values("X-Custom")
if len(got) != 3 {
t.Errorf("server got X-Custom values=%v want len=3", got)
}
want := []string{"a", "b", "c"}
for i, v := range want {
if i < len(got) && got[i] != v {
t.Errorf("X-Custom[%d] = %q want %q", i, got[i], v)
}
}
w.Header().Add("X-Server-Custom", "x")
w.Header().Add("X-Server-Custom", "y")
w.WriteHeader(http.StatusOK)
}))
defer stop()
req, _ := http.NewRequest(http.MethodGet, "http://"+addr+"/", nil)
req.Header.Add("X-Custom", "a")
req.Header.Add("X-Custom", "b")
req.Header.Add("X-Custom", "c")
resp, err := zaphttp.NewTransport(addr).RoundTrip(req)
if err != nil {
t.Fatalf("RoundTrip: %v", err)
}
defer resp.Body.Close()
got := resp.Header.Values("X-Server-Custom")
if len(got) != 2 || got[0] != "x" || got[1] != "y" {
t.Errorf("X-Server-Custom = %v want [x y]", got)
}
}
// TestStatusCode_propagation — non-200 codes round-trip with their
// reason phrase.
func TestStatusCode_propagation(t *testing.T) {
cases := []struct {
status int
body string
}{
{http.StatusNotFound, "no such thing"},
{http.StatusInternalServerError, "boom"},
{http.StatusServiceUnavailable, "draining"},
{http.StatusTeapot, "i am a teapot"},
}
for _, c := range cases {
t.Run(http.StatusText(c.status), func(t *testing.T) {
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, c.body, c.status)
}))
defer stop()
resp, err := zaphttp.Get(addr, "/")
if err != nil {
t.Fatalf("Get: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != c.status {
t.Errorf("status = %d want %d", resp.StatusCode, c.status)
}
body, _ := io.ReadAll(resp.Body)
if !strings.Contains(string(body), c.body) {
t.Errorf("body = %q want to contain %q", body, c.body)
}
})
}
}
// TestKeepAlive — multiple sequential requests on a single TCP
// connection (server side; the v0.1 client opens a new conn per call).
// We verify the server handles N back-to-back requests on the same
// dialed connection by driving the wire layer directly.
func TestKeepAlive(t *testing.T) {
addr, stop := listen(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, r.URL.Path)
}))
defer stop()
conn, err := net.Dial("tcp", addr)
if err != nil {
t.Fatalf("dial: %v", err)
}
defer conn.Close()
for i := 0; i < 5; i++ {
path := fmt.Sprintf("/req-%d", i)
req, _ := http.NewRequest(http.MethodGet, "http://"+addr+path, nil)
frame, err := zaphttp.MarshalRequest(req)
if err != nil {
t.Fatalf("marshal req %d: %v", i, err)
}
if err := writeAndRead(conn, frame); err != nil {
t.Fatalf("exchange %d: %v", i, err)
}
}
}
// helper: write a length-prefixed frame and read one back.
func writeAndRead(conn net.Conn, frame []byte) error {
hdr := make([]byte, 4)
hdr[0] = byte(len(frame) >> 24)
hdr[1] = byte(len(frame) >> 16)
hdr[2] = byte(len(frame) >> 8)
hdr[3] = byte(len(frame))
if _, err := conn.Write(hdr); err != nil {
return err
}
if _, err := conn.Write(frame); err != nil {
return err
}
if _, err := io.ReadFull(conn, hdr); err != nil {
return err
}
n := int(hdr[0])<<24 | int(hdr[1])<<16 | int(hdr[2])<<8 | int(hdr[3])
resp := make([]byte, n)
_, err := io.ReadFull(conn, resp)
return err
}
// TestMarshalRequest_idempotent — marshal/unmarshal round-trip
// preserves request fields exactly. Pure unit test, no network.
func TestMarshalRequest_idempotent(t *testing.T) {
body := []byte(`{"hello":"world"}`)
orig, _ := http.NewRequest(http.MethodPost, "http://example.com/foo?a=1", bytes.NewReader(body))
orig.Header.Set("Content-Type", "application/json")
orig.Header.Set("Authorization", "Bearer token-stays-encrypted-on-the-wire")
frame, err := zaphttp.MarshalRequest(orig)
if err != nil {
t.Fatalf("marshal: %v", err)
}
got, err := zaphttp.UnmarshalRequest(frame)
if err != nil {
t.Fatalf("unmarshal: %v", err)
}
if got.Method != http.MethodPost {
t.Errorf("method = %q want POST", got.Method)
}
if got.URL.Path != "/foo" {
t.Errorf("path = %q want /foo", got.URL.Path)
}
if got.URL.Query().Get("a") != "1" {
t.Errorf("query a = %q want 1", got.URL.Query().Get("a"))
}
if got.Header.Get("Content-Type") != "application/json" {
t.Errorf("Content-Type lost")
}
if got.Header.Get("Authorization") == "" {
t.Errorf("Authorization lost")
}
gotBody, _ := io.ReadAll(got.Body)
if !bytes.Equal(gotBody, body) {
t.Errorf("body mismatch")
}
}