rpcWebSocketConnections function

Stream<WebSocketChannel> rpcWebSocketConnections(HttpServer server, { Duration? pingInterval = const Duration(seconds: 30), dynamic protocolSelector(List<String> protocols)?, CompressionOptions compression = CompressionOptions.compressionOff, Set<String>? allowedOrigins, bool allowUpgrade(HttpRequest request)?, RpcSecurityPolicy? policy, LogScope? logger, Duration? frameProgressTimeout = const Duration(seconds: 60), })

Turns an HttpServer into the Stream<WebSocketChannel> that RpcWebSocketServer consumes, applying server-side keepalive.

Exists because the dart:io WebSocket is only reachable between the upgrade and the wrap — IOWebSocketChannel hides it.

pingInterval defaults to 30s, and an idle connection is pinged and closed if no pong comes. Without it a peer that completes the upgrade and goes silent holds its endpoint, and the contracts on it, forever: HttpServer.idleTimeout does not reach an upgraded socket. Pass null to disable; raise it where radio wake-ups matter.

protocolSelector is forwarded to WebSocketTransformer for subprotocol negotiation.

frameProgressTimeout bounds how long one frame may take to finish arriving, counted from its first byte and reset by every chunk (default 60 seconds, null disables). pingInterval cannot catch a peer that starts a frame and then drips bytes: dart:io pings only on total silence, so without this deadline such a peer holds its connection, and everything on it, indefinitely. Between frames there is no deadline — an idle connection is a legitimate one.

compression defaults to OFF where dart:io's default is ON. dart:io inflates each message with no output limit before rpc_dart sees it, so RpcSecurityPolicy.maxMessageLengthBytes cannot bound it and a few hundred KiB on the wire can peak at hundreds of MiB of RSS. Turn it on only between peers you control.

allowedOrigins refuses cross-origin handshakes. WebSocket is not subject to the same-origin policy, so a browser page will open a socket anywhere and attach ambient credentials; checking Origin at the handshake is the only protocol-level defence. Compared case-insensitively against the whole origin (scheme://host[:port]).

A request with NO Origin is ALLOWED even when this is set — every non-browser client sends none, and the attack being stopped is a browser riding cookies it cannot read. More than one Origin header is refused.

Configuring neither allowedOrigins nor allowUpgrade means ANY origin is upgraded, cross-site WebSocket hijacking included. That default is deliberate — non-browser clients send no Origin and the gate cannot know the application's origins — but it emits a one-time warning (to logger, or stderr when none is given) so an ungated deployment is a visible choice, not an oversight.

allowUpgrade runs after allowedOrigins; both must accept. Synchronous on purpose — it is in the accept path, where an await serializes every handshake. If it throws the connection is refused and the server lives: the accept loop is the ROOT ZONE, where an escaping error kills the isolate.

A refused request is answered 403 and never upgraded.

dart:io assembles a whole message before delivering it and has no ceiling of its own, so a peer that sends fragments and never a final one is buffered for as long as it writes. With compression off, each frame's header is read off the socket first, and a message past the largest frame the policy admits closes the connection before its payload is read, a peer is not read while more than 1 MiB written to it is unsent, and a frame that stalls mid-arrival past frameProgressTimeout closes the connection. With compression on, or with any other source of WebSocketChannels, none of those holds.

That policy is the RpcWebSocketServer's: the server hands it over when it starts, and a limit raised there raises this one. policy overrides it for a stream consumed by something else.

Hand the returned stream to the server as it is. A stream wrapped with .where or .map cannot take the server's policy, so a limit raised on the server alone does not reach the guard, and the server logs a warning. Filter connections with allowUpgrade and observe them with the server's onConnectionOpened instead, or pass the same policy here.

final http = await HttpServer.bind(host, port);
final server = RpcWebSocketServer(
  connections: rpcWebSocketConnections(
    http,
    allowedOrigins: {'https://app.example.com'},
  ),
  mode: RpcResponderMode.contracts([...]),
);

Implementation

Stream<WebSocketChannel> rpcWebSocketConnections(
  HttpServer server, {
  Duration? pingInterval = const Duration(seconds: 30),
  dynamic Function(List<String> protocols)? protocolSelector,
  CompressionOptions compression = CompressionOptions.compressionOff,
  Set<String>? allowedOrigins,
  bool Function(HttpRequest request)? allowUpgrade,
  RpcSecurityPolicy? policy,
  LogScope? logger,
  Duration? frameProgressTimeout = const Duration(seconds: 60),
}) {
  if (allowedOrigins == null && allowUpgrade == null) {
    _warnNoOriginGate(logger);
  }
  final ceiling = _MessageCeiling(policy);
  // Lower-cased ONCE. `_upgradeAllowed` ran `trim().toLowerCase()` over every
  // configured origin on every handshake, which allocates a string per entry per
  // connection to answer a question whose answer never changes.
  final permittedOrigins = allowedOrigins
      ?.map((origin) => origin.trim().toLowerCase())
      .toSet();

  // Filtered BEFORE the transformer rather than after: once WebSocketTransformer
  // has upgraded the request the response is already committed, and the only
  // thing left to do would be to close a socket the peer believes is open.
  //
  // Unconditional, where this used to hand `server` straight through when no gate
  // was configured. A non-upgrade request has to be caught here, and that is not
  // a gate the caller opted into — see below.
  final gated = server.where((request) {
    // The GATE comes first, and the shape check second. A forbidden origin is
    // told 403 whatever the request looked like -- that is the contract
    // `origin_guard_test` pins, and reversing the two answers a cross-origin
    // probe 400 instead, which says "wrong shape" about a request that was
    // refused on identity.
    //
    // This runs inside the accept loop's event handler, which is the ROOT
    // ZONE: anything thrown here is an unhandled async error and kills
    // the isolate -- unauthenticated, in one request. The origin read is
    // safe now, but [allowUpgrade] is USER code and cannot be, so failing
    // closed keeps a throwing predicate to a refused connection instead
    // of a dead server.
    if (permittedOrigins != null || allowUpgrade != null) {
      bool allowed;
      try {
        allowed = _upgradeAllowed(request, permittedOrigins, allowUpgrade);
      } catch (_) {
        allowed = false;
      }
      if (!allowed) {
        _answer(request, HttpStatus.forbidden, 'WebSocket upgrade refused');
        return false;
      }
    }

    // Answered HERE, because the transformer reports it as a STREAM ERROR.
    // `_upgrade` sends 400 and then completes with a `WebSocketException`, which
    // `bind` forwards to the output controller -- the server's `connections`
    // stream -- so every one reaches `onError`: an error-level record and an
    // `onConnectionError`, at whatever rate a load balancer probes. A health
    // check is not an error, and nothing downstream can tell it from one.
    //
    // A probe sends no `Origin`, so it passes the gate above and lands here.
    //
    // Guarded: dart:io reads these headers with `value()`, which THROWS on a
    // repeated one, so two Sec-WebSocket-Key headers made this predicate throw.
    // With compression on that error reached dart:io's transformer, which has
    // no error handler, and the process died.
    bool isUpgrade;
    try {
      isUpgrade = WebSocketTransformer.isUpgradeRequest(request);
    } catch (_) {
      isUpgrade = false;
    }
    if (!isUpgrade) {
      _answer(
        request,
        HttpStatus.badRequest,
        'Expected a WebSocket upgrade request',
      );
      return false;
    }
    return true;
  });

  final Stream<WebSocket> sockets = compression.enabled
      ? gated.transform(
          WebSocketTransformer(
            protocolSelector: protocolSelector,
            compression: compression,
          ),
        )
      : _upgradeEach(
          gated,
          maxMessageBytes: () => ceiling.bytes,
          protocolSelector: protocolSelector,
          frameProgressTimeout: frameProgressTimeout,
        );
  return _PolicyAwareConnections(
    sockets.map((socket) {
      // Set BEFORE wrapping: once inside IOWebSocketChannel the socket is no
      // longer reachable.
      socket.pingInterval = pingInterval;
      return IOWebSocketChannel(socket);
    }),
    ceiling,
  );
}