WebSockets

Router.WS registers a WebSocket route. The protocol is implemented in the framework rather than delegated to a library: RFC 6455 framing, masking, UTF-8 validation and the close handshake, with every rule the specification lays down enforced and every violation answered with the status it calls for.

// A WebSocket route is declared like any other: the input is bound from the
// handshake, the dependency resolves before the upgrade, and the handler owns
// the connection until it returns.
r.WS("/items/{item_id}/ws", handlers.ItemSocket,
    muzak.Summary("Talk to an item over a WebSocket"),
    muzak.Needs(core.GetCaller),
    muzak.WithWebSocket(muzak.WSOptions{
        ReadLimit:   64 << 10,
        MaxLifetime: 8 * time.Hour,
        MessageLimits: []muzak.Quota{
            {Name: "ws-messages", Window: time.Second, Limit: 10},
        },
    }))

// WSItemIn is the input of the item's WebSocket route.
//
// A handshake carries no body, so every field comes from the path, the query
// string, a header or a cookie. The pointer makes the query parameter optional:
// it stays nil when the client did not send one, which is how an absent value is
// told from a zero one.
type WSItemIn struct {
    ItemID string `path:"item_id" doc:"The item being talked about"`
    Q      *int   `query:"q" doc:"An optional number echoed back with each reply"`

    // Caller is filled by the route's provider before the handshake
    // completes; it is never read from the request.
    Caller muzak.Dep[core.Caller]
}

// ItemSocket answers a WebSocket conversation about one item.
//
// The handshake has already succeeded by the time this runs: the input is
// bound, the guards have passed and the dependency is resolved, so what is left
// is the conversation itself. Returning ends it, and Muzak closes the
// connection; returning nil closes it normally.
func ItemSocket(ctx *muzak.Context, in schemas.WSItemIn, conn *muzak.WSConn) error {
    caller := in.Caller.Get()

    for {
        message, err := conn.ReadText(ctx.Context())
        if err != nil {
            // The peer closed, or the connection was lost. Either way there is
            // nothing left to say.
            return nil
        }
        if err := conn.WriteText(ctx.Context(), "hello, "+caller.Username); err != nil {
            return err
        }
        if in.Q != nil {
            if err := conn.WriteText(ctx.Context(), fmt.Sprintf("q is %d", *in.Q)); err != nil {
                return err
            }
        }
        if err := conn.WriteText(ctx.Context(),
            fmt.Sprintf("you said %q, about item %s", message, in.ItemID)); err != nil {
            return err
        }
    }
}

The handshake is an ordinary GET

Everything that applies to a route applies here. Middleware runs, guards run, dependencies resolve, and the input struct is bound and validated before a single byte is upgraded.

A request that fails any of that is answered with the usual JSON error and never becomes a connection at all, which is the difference between a rejection a client can read and a socket that closes a moment after it opened.

One thing is judged before the guards: a request that asks to upgrade has its origin and the connection limits checked first. A cross-origin handshake, or one arriving at a full server, used to be answered 403 or 503 only after the session dependency had run with the visitor's cookie, so a refused handshake still cost a store lookup and whatever side effect it has. A plain GET that does not ask to upgrade still reaches the guards first, so a client without credentials still gets 401. So does an HTTP/2 GET, which is answered 426 after the guards: a WebSocket handshake is not carried over HTTP/2.

A room is reserved for the connection under the register's lock before the 101 is written. It used to be recorded after the response had told the peer it was connected, so a client that dialled again as soon as it read the response could be admitted past the connection limits.

curl -i 'http://localhost:8080/items/plumbus/ws'

{
  "error": {
    "code": "unauthorized",
    "message": "sign in, or pass a token query parameter",
    "status": 401
  },
  "request_id": "0611f4b2-2f0a-4b57-9c1a-6e6a2e2f9b31"
}

Because a handshake carries no body, an input type with a body field is a registration error rather than a request that mysteriously never arrives.

Where a browser can put a credential

A browser cannot set headers on a WebSocket handshake, so the two places a credential can arrive are the session cookie the browser attaches itself and a query parameter the page puts in the URL. A provider covering both is what makes the route usable from either.

// Caller is who is on the other end of a WebSocket route.
type Caller struct {
    Username  string
    FromToken bool
}

// GetCaller resolves the caller from the session a sign-in stored, or from a
// token query parameter.
func GetCaller(ctx *muzak.Context) (Caller, error) {
    if user, ok := muzak.SessionGet[string](ctx.Session(), "user"); ok {
        return Caller{Username: user}, nil
    }
    if token := ctx.Query("token"); token != "" {
        user, err := tokens.Verify(ctx.Context(), token)
        if err != nil {
            return Caller{}, muzak.Unauthorized("the token is not valid")
        }
        return Caller{Username: user, FromToken: true}, nil
    }
    return Caller{}, muzak.Unauthorized("sign in, or pass a token query parameter")
}

ctx.Session() needs AppOptions.Sessions, and reads the visitor's session during the handshake, as it would on any route. A WebSocket route only reads it: the 101 has gone out before the handler runs, so no cookie can follow, and a change made to the session on such a route is not saved and is logged as a warning. See Sessions and Dependencies.

A cookie is ambient authority a browser attaches without being asked, which is exactly why the origin check below cannot be turned off casually.

Reading and writing

Reading and writing are message oriented. A message split across several frames is reassembled and delivered once, whole. Control frames that arrive in between are handled without interrupting it: a ping is answered automatically, and a close is answered and then reported as a *WSCloseError.

type Command struct {
    Action string `json:"action"`
    Target string `json:"target"`
}

func Control(ctx *muzak.Context, in ControlIn, conn *muzak.WSConn) error {
    for {
        var command Command
        if err := conn.ReadJSON(ctx.Context(), &command); err != nil {
            return nil
        }
        if err := conn.WriteJSON(ctx.Context(), handle(command)); err != nil {
            return err
        }
    }
}

ReadJSON decodes with the same rules a request body is decoded with: unknown members, duplicate members and invalid UTF-8 are all rejected. A message that does not decode closes the connection with 1007, because a peer sending malformed JSON on a JSON connection is not going to be understood by carrying on.

The slice Read returns belongs to the caller and is not reused, so it may be kept. The payload passed to Write is not retained, so the caller may reuse it as soon as Write returns.

Concurrency

Writes are serialized, so any number of goroutines may write to one connection and each message goes out intact. A broadcast from many goroutines cannot interleave two messages.

Reads are serialized too, but a second reader is rarely what is wanted: the messages of one connection arrive in order and one loop should consume them.

func Room(ctx *muzak.Context, in RoomIn, conn *muzak.WSConn) error {
    hub := muzak.From[*core.Hub](ctx)

    updates := hub.Subscribe(ctx.Context(), in.Room)
    go func() {
        for update := range updates {
            // Safe from another goroutine: writes are serialized.
            _ = conn.WriteJSON(ctx.Context(), update)
        }
    }()

    for {
        message, err := conn.ReadText(ctx.Context())
        if err != nil {
            return nil
        }
        hub.Publish(in.Room, message)
    }
}

A deadline on a read

ReadTimeout bounds one message once it has begun, and deliberately not the wait between messages. A handler that wants its peer to say something within a given time gives each read a deadline of its own:

func Conversation(ctx *muzak.Context, in ConversationIn, conn *muzak.WSConn) error {
    for {
        // A peer that says nothing for five minutes is done with.
        readCtx, cancel := context.WithTimeout(ctx.Context(), 5*time.Minute)
        message, err := conn.ReadText(readCtx)
        cancel()
        if err != nil {
            return err
        }
        if err := conn.WriteText(ctx.Context(), "you said "+message); err != nil {
            return err
        }
    }
}

A read that its context ends ends the connection, and the peer is told why. A deadline that passed closes with 1008 and the reason "no message arrived within the time allowed", the status the read timeout and keepalive use, and a cancellation closes with 1001. The read returns the context's error once the close frame is out and the peer has answered it, or after CloseGracePeriod (250 ms by default) at the most; set it negative if a read must return the moment its deadline passes. Returning that error, as above, is not treated as a failure.

A write whose context had already ended when it got the connection sends nothing of the message and closes with 1001. One that its context interrupts part way through a frame ends the connection with no close frame, so the peer sees 1006, because a close frame would be read as more of the frame. On a WSDial connection a cancelled read closes the transport, and no close frame can follow.

Ending a connection

The connection is closed when the handler returns, so a handler owns its connection for as long as it runs and never has to arrange the teardown itself.

if !allowed(caller, in.Room) {
    return &muzak.WSCloseError{Status: muzak.WSStatusPolicyViolation, Reason: "not a member of this room"}
}

Calling Close yourself is only necessary to choose a status other than normal closure or to end the connection from another goroutine. It is safe to call more than once and from more than one goroutine; only the first call sends anything. The reason is truncated to the 123 bytes a close frame can carry, on a rune boundary.

A value the route's providers acquired with Acquire lives as long as the connection. Its release runs as soon as the handler returns, before the connection is closed, and is handed what the handler returned, so a Transaction held for the conversation commits only when the handler returned nil. A release that fails a connection which had not failed closes it with 1011, as a handler's error would. See Dependencies.

With tracing on, the request's span lasts as long as the connection and ends when it does, and a handler that fails (the 1011 row above) marks it an error, while a connection its context ended does not. AppOptions.Observer is told about the connection once, after it has ended. See Observability.

Telling a goodbye from a violation

Every read and every write returns a *WSCloseError once the connection is finished, which is why the loops above end on any error.

A peer that is still sending when the server refuses the connection reads the close status instead of a reset. A message over the read limit is answered with a close frame 1009, but a client that kept sending, a message still streaming in, often saw the connection reset (1006) and never read it: the server closed the socket with unread data in its receive buffer, which makes the kernel send a reset and drop the close frame in flight. The same happened after every close the server started on finding a fault, 1002, 1003, 1007 and 1008 included. Once the close frame is out, the server now stops sending (a half-close, or the close_notify alert on TLS), keeps reading and dropping what arrives until the peer's close frame, the end of its stream, CloseGracePeriod or 1 MiB, whichever is first, and only then closes the transport. Nothing read is kept, a peer that answers the close frame promptly is not made to wait, and one that never stops is cut off at the grace period as before. A close the handler starts with Close gets the same half-close. In a test that sends a message over the limit and keeps sending, the 1009 reached the client in 0 of 50 runs before and 50 of 50 after.

The status reaches the peer even when two closers race. A second Close (a shutdown, a keepalive timeout, a returning handler) used to tear the transport down under the first one's close frame, so the peer saw 1006. It now waits for that frame, and a peer's own close is echoed even when another goroutine is writing at that moment. A close reason that is not valid UTF-8, one an application forwarded from elsewhere, is repaired before it is sent, with invalid bytes written as U+FFFD, so the peer reads the status instead of failing the connection.

message, err := conn.ReadText(ctx.Context())
if err != nil {
    if status, ok := muzak.WSCloseStatus(err); ok && status == muzak.WSStatusNormalClosure {
        return nil
    }
    return err
}

WSCloseError.Unwrap exposes the transport failure behind an abnormal closure, so errors.Is can test for io.EOF or a network error.

The codes RFC 6455 defines are exported as muzak.WSStatus constants, from WSStatusNormalClosure (1000) to WSStatusTLSHandshake (1015). Three of them describe a local observation rather than something a peer said: WSStatusNoStatusReceived, WSStatusAbnormalClosure and WSStatusTLSHandshake are reported by Muzak but never written to the wire, and passing one to Close closes without a status code. Codes from 4000 to 4999 are free for an application to define.

What a hostile peer cannot do

A WebSocket is the longest-lived thing an unauthenticated stranger can ask a server for, so every direction a peer controls is bounded, and each bound stops something specific.

Nothing a peer sends is echoed into a response header: only a subprotocol the route itself offered can be answered with, and a route offering one that is not a token is refused when the application is built.

None of the defaults bounds how long a connection lasts. A peer that answers every ping and sends nothing is alive as far as a keepalive can tell, and with MaxLifetime unset nothing ends it. MaxConnections divided by MaxConnectionsPerIP is how many addresses can hold every slot: 1024 / 64 = 16 with the defaults, which is sixteen IPv4 addresses, or a single IPv6 /52 at the default /56 grouping. Every other handshake is then refused with 503 until they let go. A route that clients you do not control can open is worth more than the defaults:

• Put it behind a guard that authenticates the client where it can be, which ties a connection to someone rather than to an address.
• Set MaxLifetime, to minutes rather than hours, so that every slot turns over. A client that reconnects on going away, as a browser application usually does, carries on.
• Count IPv6 clients by their /48 with ClientIPOptions.ConnectionIPv6Prefix: 48, which makes each of those addresses cost a whole site's allocation.

Server-Sent Events works through the same arithmetic for event streams, which share the per-address grouping.

Options

app.Include(chat, muzak.WithWebSocket(muzak.WSOptions{
    ReadLimit:      64 << 10,
    PingInterval:   30 * time.Second,
    AllowedOrigins: []string{"https://app.example.com"},
}))

Options can be set application-wide through AppOptions.WebSocket and narrowed for a router or a route with WithWebSocket. Layering works field by field, so a route that raises only the read limit keeps the application's origin policy.

MaxConnections and MaxConnectionsPerIP may only be set on the application: the resource they protect is the process, not a route, so a router or a route that sets either is refused when the application is built.

Keepalive

Keepalive is on by default: a ping every 30 seconds (DefaultWSPingInterval), and the connection is closed when nothing has arrived from the peer within 10 (DefaultWSPongTimeout). A peer that connects and goes silent, or whose network dropped without a word, used to hold a connection slot for as long as it liked; only the per-address and global caps bounded it. A browser answers pings without being asked, so nothing needs to change for one. A negative PingInterval turns keepalive off, and a negative PongTimeout means the default, since it used to be read as zero and the first ping closed every connection whose peer could not answer instantly.

muzak.WSOptions{PingInterval: -1} // no keepalive on this route

A pong is consumed by a read like any other frame, so a connection is closed for an unanswered ping only when the handler was reading during it. The rule is what makes the default safe: a handler that only ever writes, or one busy with a message, is left alone, which also stops keepalive closing a healthy peer of a handler that does not read. It also means a handler that only writes is never closed by keepalive, and should ping by hand with Ping if it wants that. The pong check uses the monotonic clock, so a wall-clock step cannot close a healthy connection.

Anything that arrives from the peer after a ping answers it as well as a pong would, since what is being asked is whether the peer is still there. A peer part way through one large frame cannot send its pong until the frame is finished, and used to be closed with 1008 while its bytes were still coming in: with the defaults, any 1 MiB message sent at under about 100 KB/s was closed ten seconds in, although ReadTimeout allowed thirty. Every frame header and every read that brings payload now counts as hearing from the peer, so only a peer from which nothing at all has arrived for a pong timeout is closed. One that stalls part way through a frame is still closed by the keepalive, and by ReadTimeout in any case.

Ending a connection on purpose

Keepalive closes a peer that stops answering. It cannot close one that answers every ping and never says anything, which costs that peer a TCP connection and holds a slot, a goroutine and a file descriptor here until the server stops. MaxLifetime is the bound for that. It is unset by default, which leaves connections unbounded, and worth setting on any route a client you do not control can open:

muzak.WSOptions{MaxLifetime: 2 * time.Hour}

When the lifetime has passed the connection is closed with 1001 Going Away, the status of a server that is done with a connection rather than one that failed. The handler sees it as the close it reads, and the connection's slot is released once the transport is closed. A client that reconnects on going away, as a browser application usually does, carries on. A negative value removes a bound a wider scope set.

The origin check

A cross-origin handshake is refused by default, and no CORS policy changes that.

A WebSocket handshake is not subject to the same-origin policy and is never preflighted, which is what makes cross-site hijacking possible in the first place: without the check, any page on the internet could open an authenticated connection to your server from a visitor's browser, cookies and all. AppOptions.CORS has no bearing on it, for exactly that reason. Nor does AppOptions.CrossOriginProtection: a handshake is a GET, which it never refuses, so this check is what keeps another site's page from opening a connection with your visitor's cookies. See Cross-Origin Requests.

muzak.WSOptions{AllowedOrigins: []string{"https://app.example.com"}}

The server's own origin is always allowed. The single entry "*" allows any origin.

An origin is read strictly: a scheme and a host with nothing after it, compared as a browser serializes it. http://attacker@host, http://host/path, http://host?x and //host used to pass as the server's own origin, because the host was taken out of any value a lenient URL parse could find one in. A value carrying credentials, a path, a query or a fragment is not the server's own origin however its host reads, and a handshake with more than one Origin header is refused with 400. Values listed in AllowedOrigins, and the ones AllowOriginFunc sees, are compared as they arrived, without regard to case. A browser sends neither shape; a non-browser client that sent a decorated Origin must send a bare scheme://host[:port] or none at all.

The lists themselves are checked when the application is built, the same way the CORS list is. An AllowedOrigins entry that can never equal what a browser sends, such as https://app.example.com/, https://app.example.com/ws, https://app.example.com:443 or https://*.example.com, is a build error naming the route and the spelling to use, and so is "null": it is the origin a sandboxed iframe sends, so any page can present it, and listing it would let every page open a connection with the visitor's cookies. The single entry "*" still allows any origin, and wildcard subdomains belong in AllowOriginFunc. An AllowedHosts entry with a scheme, a path, credentials or a pattern can never equal a Host header and is refused too.

DNS rebinding

The server's own origin is any whose host is the request's Host, which is whatever the client wrote there. That is enough for a browser that reached the server by its name, and it is not for one that was pointed at it by somebody else's: in a DNS-rebinding attack a page served from the attacker's name, made to resolve to your server, sends that name as both Origin and Host, and reads as same-origin. AllowedHosts lists the names the route answers to:

muzak.WSOptions{AllowedHosts: []string{"app.example.com", "app.example.com:8443"}}

When it is set, a handshake counts as same-origin only if its Host is listed, compared without regard to case and with any port written as the client wrote it. Any other Host is refused with 403 unless AllowedOrigins or AllowOriginFunc names the origin. A handshake with no Origin is not affected, since only a browser sends one. Unset, the rule is what it was, so nothing changes until it is set; it is worth setting for a server on a loopback or private address, where such a page is a real threat.

The option bounds the WebSocket handshake only. A plain HTTP route is not protected against rebinding by it. AppOptions.AllowedHosts holds every request to the application's own names, handshakes included: a request for any other host is answered 421 before a route, a guard or the origin check runs. See Hosts and HTTPS.

muzak.WSOptions{
    AllowOriginFunc: func(r *http.Request, origin string) bool {
        return strings.HasSuffix(origin, ".example.com")
    },
}

It runs on every handshake an earlier rule did not already allow, so it must be cheap and free of side effects.

Options layer field by field, so an application's AllowOriginFunc stays in force for a route that only lists origins of its own, and a scope that says nothing keeps a wider scope's InsecureSkipOriginCheck. A router or route that does need the check, such as one authenticated by a cookie in an application that otherwise authenticates by token, sets EnforceOriginCheck: true. A scope beneath it can set InsecureSkipOriginCheck again, and the narrowest scope that set either wins. A scope that sets both keeps the check.

// Every route takes a token the client presents explicitly...
app := muzak.New(muzak.AppOptions{
    WebSocket: muzak.WSOptions{InsecureSkipOriginCheck: true},
})

// ...except this one, which a browser authenticates with its session cookie.
r.WS("/account/ws", handlers.AccountSocket,
    muzak.WithWebSocket(muzak.WSOptions{EnforceOriginCheck: true}))

InsecureSkipOriginCheck accepts a handshake from anywhere. It is safe only for a connection that carries no ambient authority: one authenticated by a token the client has to present explicitly, never by a cookie, since a browser attaches cookies to a cross-origin handshake without being asked.

Subprotocols

muzak.WSOptions{Subprotocols: []string{"graphql-transport-ws", "graphql-ws"}}

The client's own list is in preference order, so the first of its choices that appears here is the one negotiated. A client asking for something else is answered without the header, which tells it to give up. Read the result with conn.Subprotocol().

Shutdown

Open connections are tracked, so a graceful shutdown tells every peer it is going away with 1001 and waits for the handlers, and for connections that are still closing. net/http cannot do that on its own: a hijacked connection is no longer one it knows about. The handler's context ends with its connection, so a handler waiting on it does not outlive its peer or the server's close at shutdown.

The goodbye comes after readiness has turned unavailable and ServerOptions.DrainDelay has passed, during which connections stay open and new handshakes are still served. It is sent as the listeners close, alongside the drain of ordinary requests, and the background tasks and lifecycle components are dealt with only once the handlers have returned. Shutdown shares one deadline with everything else; see Lifecycle.

Documentation

A WebSocket route appears in the OpenAPI document with its parameters and a 101 Switching Protocols response, because that is what OpenAPI can say about a handshake. The conversation itself continues off the document. What a handshake is refused with is listed too: 426 and 422, 400 for a malformed handshake, 403 for a foreign origin (left out for a route that skips the check) and 503 for a draining or full server. A status a route declared with WithResponseDoc keeps its own description.

Testing

muzak.WSDial is the client half of the same engine, which is what lets a route be tested over a real connection rather than against a second implementation. The test client wraps it:

func TestItemSocket(t *testing.T) {
    client := testclient.New(t, buildApp())

    // A token tokens.Verify accepts for the user jessica.
    conn := client.WS("/items/plumbus/ws", testclient.Query("token", jessicasToken))

    if err := conn.WriteText(t.Context(), "hello"); err != nil {
        t.Fatalf("WriteText = %v", err)
    }
    reply, err := conn.ReadText(t.Context())
    if err != nil {
        t.Fatalf("ReadText = %v", err)
    }
    if reply != "hello, jessica" {
        t.Errorf("reply = %q", reply)
    }
}

func TestItemSocketNeedsCredential(t *testing.T) {
    client := testclient.New(t, buildApp())

    _, response := client.TryWS("/items/plumbus/ws")

    response.AssertStatus(http.StatusUnauthorized)
}

Outside a test, dial directly:

conn, _, err := muzak.WSDial(ctx, "ws://"+app.Addr()+"/items/plumbus/ws", muzak.WSDialOptions{})
if err != nil {
    return err
}
defer conn.Close(muzak.WSStatusNormalClosure, "")

The URL may use ws, wss, http or https. The response is returned alongside the connection so a caller can read the handshake headers, and on failure the status and body the server refused with. WSDial never follows a redirect, because following one would send the handshake's headers to whatever host the answer named.

A 101 that names an extension is refused, because the client never offers one and RFC 6455 section 4.1 says a response that negotiates one it did not ask for fails the connection. It used to read on with frames whose meaning the server had changed. WSDial connections apply ReadTimeout and WriteTimeout, which they did not before: the transport net/http hands back is not a net.Conn, so no deadline was ever set and a server that stopped reading held the client forever. The frame-size check that let a message slip past the read limit is fixed on the client too.

A WSDial connection waits, up to CloseGracePeriod, for the server to close TCP first, as RFC 6455 section 7.1.1 asks, so that the server rather than every client holds the TIME_WAIT state a closed connection leaves behind. It reads on to the end of the stream, so Close returns once the server has hung up, or after the grace period at the most. It used to hang up the moment its close frame was out, and WSDialOptions.CloseGracePeriod did nothing.

Where to go next

Server-Sent Events covers the simpler half of what a WebSocket is usually reached for, and Rate Limiting covers the quotas MessageLimits reuses.