Bump version to 0.6.0

From: https://github.com/ekzhang/bore/pull/162#discussion_r2042195219

.github/FUNDING.yml

@@ -0,0 +1 @@
+github: [ekzhang]

.github/workflows/ci.yml

@@ -1,13 +1,17 @@
 name: CI
 
-on: [push, pull_request]
+on:
+  push:
+    branches:
+      - main
+  pull_request:
 
 jobs:
   rust:
     name: Build and Test
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
       - uses: actions-rs/toolchain@v1
         with:
@@ -22,7 +26,7 @@ jobs:
     name: Rustfmt
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
       - uses: actions-rs/toolchain@v1
         with:
@@ -36,7 +40,7 @@ jobs:
     name: Clippy
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
       - uses: actions-rs/toolchain@v1
         with:
@@ -44,4 +48,4 @@ jobs:
           toolchain: stable
           components: clippy
 
-      - run: cargo clippy
+      - run: cargo clippy -- -D warnings

.github/workflows/docker.yml

@@ -11,11 +11,11 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
 
       - name: Docker meta
         id: meta
-        uses: docker/metadata-action@v3
+        uses: docker/metadata-action@v4
         with:
           images: ekzhang/bore
           tags: |
@@ -23,22 +23,22 @@ jobs:
             type=semver,pattern={{major}}.{{minor}}
 
       - name: Set up QEMU
-        uses: docker/setup-qemu-action@v1
+        uses: docker/setup-qemu-action@v2
         with:
           platforms: arm64
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v1
+        uses: docker/setup-buildx-action@v2
 
       - name: Login to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v2
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
           password: ${{ secrets.DOCKERHUB_TOKEN }}
 
       - name: Build and push
         id: docker_build
-        uses: docker/build-push-action@v2
+        uses: docker/build-push-action@v3
         with:
           platforms: linux/amd64,linux/arm64
           push: true

.github/workflows/mean_bean_ci.yml

@@ -0,0 +1,130 @@
+name: Mean Bean CI
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+jobs:
+  install-cross:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 50
+
+      - uses: XAMPPRocky/get-github-release@v1
+        id: cross
+        with:
+          owner: rust-embedded
+          repo: cross
+          matches: ${{ matrix.platform }}
+          token: ${{ secrets.GITHUB_TOKEN }}
+
+      - uses: actions/upload-artifact@v4
+        with:
+          name: cross-${{ matrix.platform }}
+          path: ${{ steps.cross.outputs.install_path }}
+    strategy:
+      matrix:
+        platform: [linux-musl]
+
+  macos:
+    runs-on: macos-latest
+    strategy:
+      fail-fast: true
+      matrix:
+        channel: [stable]
+        target:
+          - aarch64-apple-darwin
+
+    steps:
+      - name: Setup | Checkout
+        uses: actions/checkout@v4
+      - name: Setup | Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          override: true
+          profile: minimal
+          target: ${{ matrix.target }}
+      - run: ci/set_rust_version.bash ${{ matrix.channel }} ${{ matrix.target }}
+      - name: Test
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --target ${{ matrix.target }}
+          use-cross: false
+
+  linux:
+    runs-on: ubuntu-latest
+    needs: install-cross
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 50
+
+      - name: Download Cross
+        uses: actions/download-artifact@v4
+        with:
+          name: cross-linux-musl
+          path: /tmp/
+      - run: chmod +x /tmp/cross
+      - run: ci/set_rust_version.bash ${{ matrix.channel }} ${{ matrix.target }}
+      - run: ci/build.bash /tmp/cross ${{ matrix.target }}
+        # These targets have issues with being tested so they are disabled
+        # by default. You can try disabling to see if they work for
+        # your project.
+      - run: ci/test.bash /tmp/cross ${{ matrix.target }}
+        if: |
+          !contains(matrix.target, 'android') &&
+          !contains(matrix.target, 'bsd') &&
+          !contains(matrix.target, 'solaris') &&
+          matrix.target != 'armv5te-unknown-linux-musleabi' &&
+          matrix.target != 'sparc64-unknown-linux-gnu'
+
+    strategy:
+      fail-fast: true
+      matrix:
+        channel: [stable]
+        target:
+          - aarch64-unknown-linux-musl
+          - arm-unknown-linux-musleabi
+          - arm-unknown-linux-gnueabi
+          - armv7-unknown-linux-gnueabihf
+          - armv7-unknown-linux-musleabihf
+          - i686-unknown-linux-musl
+          - x86_64-unknown-linux-musl
+
+  windows:
+    runs-on: windows-latest
+    # Windows technically doesn't need this, but if we don't block windows on it
+    # some of the windows jobs could fill up the concurrent job queue before
+    # one of the install-cross jobs has started, so this makes sure all
+    # artifacts are downloaded first.
+    needs: install-cross
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 50
+      - run: ci/set_rust_version.bash ${{ matrix.channel }} ${{ matrix.target }}
+        shell: bash
+      - run: ci/build.bash cargo ${{ matrix.target }}
+        shell: bash
+      - run: ci/test.bash cargo ${{ matrix.target }}
+        shell: bash
+
+    strategy:
+      fail-fast: true
+      matrix:
+        channel: [stable]
+        target:
+          # MSVC
+          - i686-pc-windows-msvc
+          - x86_64-pc-windows-msvc
+          # GNU: You typically only need to test Windows GNU if you're
+          # specifically targetting it, and it can cause issues with some
+          # dependencies if you're not so it's disabled by self.
+          # - i686-pc-windows-gnu
+          # - x86_64-pc-windows-gnu

.github/workflows/mean_bean_deploy.yml

@@ -0,0 +1,195 @@
+on:
+  push:
+    # # Sequence of patterns matched against refs/tags
+    tags:
+      - "v*" # Push events to matching v*, i.e. v1.0, v20.15.10
+
+name: Mean Bean Deploy
+env:
+  BIN: bore
+
+jobs:
+  # This job downloads and stores `cross` as an artifact, so that it can be
+  # redownloaded across all of the jobs. Currently this copied pasted between
+  # `mean_bean_ci.yml` and `mean_bean_deploy.yml`. Make sure to update both places when making
+  # changes.
+  install-cross:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 50
+
+      - uses: XAMPPRocky/get-github-release@v1
+        id: cross
+        with:
+          owner: rust-embedded
+          repo: cross
+          matches: ${{ matrix.platform }}
+          token: ${{ secrets.GITHUB_TOKEN }}
+
+      - uses: actions/upload-artifact@v4
+        with:
+          name: cross-${{ matrix.platform }}
+          path: ${{ steps.cross.outputs.install_path }}
+    strategy:
+      matrix:
+        platform: [linux-musl]
+
+  macos:
+    runs-on: macos-latest
+    strategy:
+      matrix:
+        target:
+          # macOS
+          - x86_64-apple-darwin
+          - aarch64-apple-darwin
+    steps:
+      - name: Get tag
+        id: tag
+        uses: dawidd6/action-get-tag@v1
+
+      - name: Setup | Checkout
+        uses: actions/checkout@v4
+
+      # Cache files between builds
+      - name: Setup | Cache Cargo
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+          key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}-${{ matrix.target }}
+
+      - name: Setup | Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          override: true
+          profile: minimal
+          target: ${{ matrix.target }}
+
+      - name: Build | Build
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --release --target ${{ matrix.target }}
+
+      - run: tar -czvf ${{ env.BIN }}.tar.gz --directory=target/${{ matrix.target }}/release ${{ env.BIN }}
+      - uses: XAMPPRocky/create-release@v1.0.2
+        id: create_release
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          tag_name: ${{ github.ref }}
+          release_name: ${{ github.ref }}
+          draft: false
+          prerelease: false
+
+      - uses: actions/upload-release-asset@v1
+        id: upload-release-asset
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }}
+          asset_path: ${{ env.BIN }}.tar.gz
+          asset_name: ${{ env.BIN }}-${{steps.tag.outputs.tag}}-${{ matrix.target }}.tar.gz
+          asset_content_type: application/gzip
+
+  linux:
+    runs-on: ubuntu-latest
+    needs: install-cross
+    strategy:
+      matrix:
+        target:
+          - aarch64-unknown-linux-musl
+          - arm-unknown-linux-musleabi
+          - arm-unknown-linux-gnueabi
+          - armv7-unknown-linux-gnueabihf
+          - armv7-unknown-linux-musleabihf
+          - i686-unknown-linux-musl
+          - x86_64-unknown-linux-musl
+    steps:
+      - name: Get tag
+        id: tag
+        uses: dawidd6/action-get-tag@v1
+
+      - uses: actions/checkout@v4
+      - uses: actions/download-artifact@v4
+        with:
+          name: cross-linux-musl
+          path: /tmp/
+      - run: chmod +x /tmp/cross
+
+      - run: ci/set_rust_version.bash stable ${{ matrix.target }}
+      - run: ci/build.bash /tmp/cross ${{ matrix.target }} RELEASE
+      - run: tar -czvf ${{ env.BIN }}.tar.gz --directory=target/${{ matrix.target }}/release ${{ env.BIN }}
+
+      - uses: XAMPPRocky/create-release@v1.0.2
+        id: create_release
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          tag_name: ${{ github.ref }}
+          release_name: ${{ github.ref }}
+          draft: false
+          prerelease: false
+
+      - name: Upload Release Asset
+        id: upload-release-asset
+        uses: actions/upload-release-asset@v1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }}
+          asset_path: ${{ env.BIN }}.tar.gz
+          asset_name: ${{ env.BIN }}-${{steps.tag.outputs.tag}}-${{ matrix.target }}.tar.gz
+          asset_content_type: application/gzip
+
+  windows:
+    runs-on: windows-latest
+    needs: install-cross
+    strategy:
+      matrix:
+        target:
+          # MSVC
+          - i686-pc-windows-msvc
+          - x86_64-pc-windows-msvc
+          # GNU
+          # - i686-pc-windows-gnu
+          # - x86_64-pc-windows-gnu
+    steps:
+      - name: Get tag
+        id: tag
+        uses: dawidd6/action-get-tag@v1
+
+      - uses: actions/checkout@v4
+      - run: bash ci/set_rust_version.bash stable ${{ matrix.target }}
+      - run: bash ci/build.bash cargo ${{ matrix.target }} RELEASE
+      - run: |
+          cd ./target/${{ matrix.target }}/release/
+          7z a "${{ env.BIN }}.zip" "${{ env.BIN }}.exe"
+          mv "${{ env.BIN }}.zip" $GITHUB_WORKSPACE
+        shell: bash
+        # We're using using a fork of `actions/create-release` that detects
+        # whether a release is already available or not first.
+      - uses: XAMPPRocky/create-release@v1.0.2
+        id: create_release
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          tag_name: ${{ github.ref }}
+          release_name: ${{ github.ref }}
+          # Draft should **always** be false. GitHub doesn't provide a way to
+          # get draft releases from its API, so there's no point using it.
+          draft: false
+          prerelease: false
+      - uses: actions/upload-release-asset@v1
+        id: upload-release-asset
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }}
+          asset_path: ${{ env.BIN }}.zip
+          asset_name: ${{ env.BIN }}-${{steps.tag.outputs.tag}}-${{ matrix.target }}.zip
+          asset_content_type: application/zip

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "bore-cli"
-version = "0.3.0"
+version = "0.6.0"
 authors = ["Eric Zhang <ekzhang1@gmail.com>"]
 license = "MIT"
 description = "A modern, simple TCP tunnel in Rust that exposes local ports to a remote server, bypassing standard NAT connection firewalls."
@@ -17,19 +17,22 @@ path = "src/main.rs"
 
 [dependencies]
 anyhow = { version = "1.0.56", features = ["backtrace"] }
-clap = { version = "3.1.8", features = ["derive", "env"] }
+clap = { version = "4.0.22", features = ["derive", "env"] }
 dashmap = "5.2.0"
+fastrand = "1.9.0"
+futures-util = { version = "0.3.21", features = ["sink"] }
 hex = "0.4.3"
 hmac = "0.12.1"
 serde = { version = "1.0.136", features = ["derive"] }
 serde_json = "1.0.79"
 sha2 = "0.10.2"
 tokio = { version = "1.17.0", features = ["rt-multi-thread", "io-util", "macros", "net", "time"] }
+tokio-util = { version = "0.7.1", features = ["codec"] }
 tracing = "0.1.32"
-tracing-subscriber = "0.3.10"
-uuid = { version = "0.8.2", features = ["serde", "v4"] }
+tracing-subscriber = "0.3.18"
+uuid = { version = "1.2.1", features = ["serde", "v4"] }
 
 [dev-dependencies]
 lazy_static = "1.4.0"
-rstest = "0.12.0"
+rstest = "0.15.0"
 tokio = { version = "1.17.0", features = ["sync"] }

Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:alpine as builder
+FROM rust:alpine AS builder
 WORKDIR /home/rust/src
 RUN apk --no-cache add musl-dev
 COPY . .

README.md

@@ -1,6 +1,6 @@
 # bore
 
-[![Build status](https://img.shields.io/github/workflow/status/ekzhang/bore/CI)](https://github.com/ekzhang/bore/actions)
+[![Build status](https://img.shields.io/github/actions/workflow/status/ekzhang/bore/ci.yml)](https://github.com/ekzhang/bore/actions)
 [![Crates.io](https://img.shields.io/crates/v/bore-cli.svg)](https://crates.io/crates/bore-cli)
 
 A modern, simple TCP tunnel in Rust that exposes local ports to a remote server, bypassing standard NAT connection firewalls. **That's all it does: no more, and no less.**
@@ -8,7 +8,7 @@ A modern, simple TCP tunnel in Rust that exposes local ports to a remote server,
 ![Video demo](https://i.imgur.com/vDeGsmx.gif)
 
 ```shell
-# Installation (requires Rust)
+# Installation (requires Rust, see alternatives below)
 cargo install bore-cli
 
 # On your local machine
@@ -19,17 +19,53 @@ This will expose your local port at `localhost:8000` to the public internet at `
 
 Similar to [localtunnel](https://github.com/localtunnel/localtunnel) and [ngrok](https://ngrok.io/), except `bore` is intended to be a highly efficient, unopinionated tool for forwarding TCP traffic that is simple to install and easy to self-host, with no frills attached.
 
-(`bore` totals less than 400 lines of safe, async Rust code and is trivial to set up — just run a single binary for the client and server.)
+(`bore` totals about 400 lines of safe, async Rust code and is trivial to set up — just run a single binary for the client and server.)
 
 ## Installation
 
-You can build the `bore` CLI command from source using [Cargo](https://doc.rust-lang.org/cargo/), the Rust package manager. This command installs the `bore` binary at a user-accessible path.
+### macOS
+
+`bore` is packaged as a Homebrew core formula.
+
+```shell
+brew install bore-cli
+```
+
+### Linux
+
+#### Arch Linux
+
+`bore` is available in the AUR as `bore`.
+
+```shell
+yay -S bore # or your favorite AUR helper
+```
+
+#### Gentoo Linux
+
+`bore` is available in the [gentoo-zh](https://github.com/microcai/gentoo-zh) overlay.
+
+```shell
+sudo eselect repository enable gentoo-zh
+sudo emerge --sync gentoo-zh
+sudo emerge net-proxy/bore
+```
+
+### Binary Distribution
+
+Otherwise, the easiest way to install bore is from prebuilt binaries. These are available on the [releases page](https://github.com/ekzhang/bore/releases) for macOS, Windows, and Linux. Just unzip the appropriate file for your platform and move the `bore` executable into a folder on your PATH.
+
+### Cargo
+
+You also can build `bore` from source using [Cargo](https://doc.rust-lang.org/cargo/), the Rust package manager. This command installs the `bore` binary at a user-accessible path.
 
 ```shell
 cargo install bore-cli
 ```
 
-We also publish versioned Docker images for each release. Each image is built for AMD 64-bit and Arm 64-bit architectures. They're tagged with the specific version and allow you to run the statically-linked `bore` binary from a minimal "scratch" container.
+### Docker
+
+We also publish versioned Docker images for each release. The image is built for an AMD 64-bit architecture. They're tagged with the specific version and allow you to run the statically-linked `bore` binary from a minimal "scratch" container.
 
 ```shell
 docker run -it --init --rm --network host ekzhang/bore <ARGS>
@@ -52,22 +88,19 @@ You can optionally pass in a `--port` option to pick a specific port on the remo
 The full options are shown below.
 
 ```shell
-bore-local 0.3.0
 Starts a local proxy to the remote server
 
-USAGE:
-    bore local [OPTIONS] --to <TO> <LOCAL_PORT>
+Usage: bore local [OPTIONS] --to <TO> <LOCAL_PORT>
 
-ARGS:
-    <LOCAL_PORT>    The local port to expose
+Arguments:
+  <LOCAL_PORT>  The local port to expose [env: BORE_LOCAL_PORT=]
 
-OPTIONS:
-    -h, --help                 Print help information
-    -l, --local-host <HOST>    The local host to expose [default: localhost]
-    -p, --port <PORT>          Optional port on the remote server to select [default: 0]
-    -s, --secret <SECRET>      Optional secret for authentication [env: BORE_SECRET]
-    -t, --to <TO>              Address of the remote server to expose local ports to
-    -V, --version              Print version information
+Options:
+  -l, --local-host <HOST>  The local host to expose [default: localhost]
+  -t, --to <TO>            Address of the remote server to expose local ports to [env: BORE_SERVER=]
+  -p, --port <PORT>        Optional port on the remote server to select [default: 0]
+  -s, --secret <SECRET>    Optional secret for authentication [env: BORE_SECRET]
+  -h, --help               Print help
 ```
 
 ### Self-Hosting
@@ -80,20 +113,22 @@ bore server
 
 That's all it takes! After the server starts running at a given address, you can then update the `bore local` command with option `--to <ADDRESS>` to forward a local port to this remote server.
 
+It's possible to specify different IP addresses for the control server and for the tunnels. This setup is useful for cases where you might want the control server to be on a private network while allowing tunnel connections over a public interface, or vice versa.
+
 The full options for the `bore server` command are shown below.
 
 ```shell
-bore-server 0.3.0
 Runs the remote proxy server
 
-USAGE:
-    bore server [OPTIONS]
+Usage: bore server [OPTIONS]
 
-OPTIONS:
-    -h, --help                   Print help information
-        --min-port <MIN_PORT>    Minimum TCP port number to accept [default: 1024]
-    -s, --secret <SECRET>        Optional secret for authentication [env: BORE_SECRET]
-    -V, --version                Print version information
+Options:
+      --min-port <MIN_PORT>          Minimum accepted TCP port number [env: BORE_MIN_PORT=] [default: 1024]
+      --max-port <MAX_PORT>          Maximum accepted TCP port number [env: BORE_MAX_PORT=] [default: 65535]
+  -s, --secret <SECRET>              Optional secret for authentication [env: BORE_SECRET]
+      --bind-addr <BIND_ADDR>        IP address to bind to, clients must reach this [default: 0.0.0.0]
+      --bind-tunnels <BIND_TUNNELS>  IP address where tunnels will listen on, defaults to --bind-addr
+  -h, --help                         Print help
 ```
 
 ## Protocol

ci/build.bash

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+# Script for building your rust projects.
+set -e
+
+source ci/common.bash
+
+# $1 {path} = Path to cross/cargo executable
+CROSS=$1
+# $1 {string} = <Target Triple> e.g. x86_64-pc-windows-msvc
+TARGET_TRIPLE=$2
+# $3 {boolean} = Are we building for deployment? 
+RELEASE_BUILD=$3
+
+required_arg $CROSS 'CROSS'
+required_arg $TARGET_TRIPLE '<Target Triple>'
+
+if [ -z "$RELEASE_BUILD" ]; then
+    $CROSS build --target $TARGET_TRIPLE
+    $CROSS build --target $TARGET_TRIPLE --all-features
+else
+    $CROSS build --target $TARGET_TRIPLE --all-features --release
+fi

ci/common.bash

@@ -0,0 +1,6 @@
+required_arg() {
+    if [ -z "$1" ]; then
+        echo "Required argument $2 missing"
+        exit 1
+    fi
+}

ci/set_rust_version.bash

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -e
+rustup default $1
+rustup target add $2

ci/test.bash

@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+# Script for building your rust projects.
+set -e
+
+source ci/common.bash
+
+# $1 {path} = Path to cross/cargo executable
+CROSS=$1
+# $1 {string} = <Target Triple>
+TARGET_TRIPLE=$2
+
+required_arg $CROSS 'CROSS'
+required_arg $TARGET_TRIPLE '<Target Triple>'
+
+max_attempts=3
+count=0
+
+while [ $count -lt $max_attempts ]; do
+    $CROSS test --target $TARGET_TRIPLE
+    status=$?
+    if [ $status -eq 0 ]; then
+        echo "Test passed"
+        break
+    else
+        echo "Test failed, attempt $(($count + 1))"
+    fi
+    count=$(($count + 1))
+done
+
+if [ $status -ne 0 ]; then
+    echo "Test failed after $max_attempts attempts"
+fi

src/auth.rs

@@ -3,10 +3,10 @@
 use anyhow::{bail, ensure, Result};
 use hmac::{Hmac, Mac};
 use sha2::{Digest, Sha256};
-use tokio::io::{AsyncBufRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite};
 use uuid::Uuid;
 
-use crate::shared::{recv_json_timeout, send_json, ClientMessage, ServerMessage};
+use crate::shared::{ClientMessage, Delimited, ServerMessage};
 
 /// Wrapper around a MAC used for authenticating clients that have a secret.
 pub struct Authenticator(Hmac<Sha256>);
@@ -48,13 +48,13 @@ impl Authenticator {
     }
 
     /// As the server, send a challenge to the client and validate their response.
-    pub async fn server_handshake(
+    pub async fn server_handshake<T: AsyncRead + AsyncWrite + Unpin>(
         &self,
-        stream: &mut (impl AsyncBufRead + AsyncWrite + Unpin),
+        stream: &mut Delimited<T>,
     ) -> Result<()> {
         let challenge = Uuid::new_v4();
-        send_json(stream, ServerMessage::Challenge(challenge)).await?;
-        match recv_json_timeout(stream).await? {
+        stream.send(ServerMessage::Challenge(challenge)).await?;
+        match stream.recv_timeout().await? {
             Some(ClientMessage::Authenticate(tag)) => {
                 ensure!(self.validate(&challenge, &tag), "invalid secret");
                 Ok(())
@@ -64,16 +64,16 @@ impl Authenticator {
     }
 
     /// As the client, answer a challenge to attempt to authenticate with the server.
-    pub async fn client_handshake(
+    pub async fn client_handshake<T: AsyncRead + AsyncWrite + Unpin>(
         &self,
-        stream: &mut (impl AsyncBufRead + AsyncWrite + Unpin),
+        stream: &mut Delimited<T>,
     ) -> Result<()> {
-        let challenge = match recv_json_timeout(stream).await? {
+        let challenge = match stream.recv_timeout().await? {
             Some(ServerMessage::Challenge(challenge)) => challenge,
             _ => bail!("expected authentication challenge, but no secret was required"),
         };
         let tag = self.answer(&challenge);
-        send_json(stream, ClientMessage::Authenticate(tag)).await?;
+        stream.send(ClientMessage::Authenticate(tag)).await?;
         Ok(())
     }
 }

src/client.rs

@@ -3,20 +3,17 @@
 use std::sync::Arc;
 
 use anyhow::{bail, Context, Result};
-use tokio::{io::BufReader, net::TcpStream, time::timeout};
+use tokio::{io::AsyncWriteExt, net::TcpStream, time::timeout};
 use tracing::{error, info, info_span, warn, Instrument};
 use uuid::Uuid;
 
 use crate::auth::Authenticator;
-use crate::shared::{
-    proxy, recv_json, recv_json_timeout, send_json, ClientMessage, ServerMessage, CONTROL_PORT,
-    NETWORK_TIMEOUT,
-};
+use crate::shared::{ClientMessage, Delimited, ServerMessage, CONTROL_PORT, NETWORK_TIMEOUT};
 
 /// State structure for the client.
 pub struct Client {
     /// Control connection to the server.
-    conn: Option<BufReader<TcpStream>>,
+    conn: Option<Delimited<TcpStream>>,
 
     /// Destination address of the server.
     to: String,
@@ -43,15 +40,14 @@ impl Client {
         port: u16,
         secret: Option<&str>,
     ) -> Result<Self> {
-        let mut stream = BufReader::new(connect_with_timeout(to, CONTROL_PORT).await?);
-
+        let mut stream = Delimited::new(connect_with_timeout(to, CONTROL_PORT).await?);
         let auth = secret.map(Authenticator::new);
         if let Some(auth) = &auth {
             auth.client_handshake(&mut stream).await?;
         }
 
-        send_json(&mut stream, ClientMessage::Hello(port)).await?;
-        let remote_port = match recv_json_timeout(&mut stream).await? {
+        stream.send(ClientMessage::Hello(port)).await?;
+        let remote_port = match stream.recv_timeout().await? {
             Some(ServerMessage::Hello(remote_port)) => remote_port,
             Some(ServerMessage::Error(message)) => bail!("server error: {message}"),
             Some(ServerMessage::Challenge(_)) => {
@@ -82,10 +78,8 @@ impl Client {
     pub async fn listen(mut self) -> Result<()> {
         let mut conn = self.conn.take().unwrap();
         let this = Arc::new(self);
-        let mut buf = Vec::new();
         loop {
-            let msg = recv_json(&mut conn, &mut buf).await?;
-            match msg {
+            match conn.recv().await? {
                 Some(ServerMessage::Hello(_)) => warn!("unexpected hello"),
                 Some(ServerMessage::Challenge(_)) => warn!("unexpected challenge"),
                 Some(ServerMessage::Heartbeat) => (),
@@ -110,14 +104,16 @@ impl Client {
 
     async fn handle_connection(&self, id: Uuid) -> Result<()> {
         let mut remote_conn =
-            BufReader::new(connect_with_timeout(&self.to[..], CONTROL_PORT).await?);
+            Delimited::new(connect_with_timeout(&self.to[..], CONTROL_PORT).await?);
         if let Some(auth) = &self.auth {
             auth.client_handshake(&mut remote_conn).await?;
         }
-        send_json(&mut remote_conn, ClientMessage::Accept(id)).await?;
-
-        let local_conn = connect_with_timeout(&self.local_host, self.local_port).await?;
-        proxy(local_conn, remote_conn).await?;
+        remote_conn.send(ClientMessage::Accept(id)).await?;
+        let mut local_conn = connect_with_timeout(&self.local_host, self.local_port).await?;
+        let mut parts = remote_conn.into_parts();
+        debug_assert!(parts.write_buf.is_empty(), "framed write buffer not empty");
+        local_conn.write_all(&parts.read_buf).await?; // mostly of the cases, this will be empty
+        tokio::io::copy_bidirectional(&mut local_conn, &mut parts.io).await?;
         Ok(())
     }
 }

src/main.rs

@@ -1,10 +1,11 @@
+use std::net::IpAddr;
+
 use anyhow::Result;
 use bore_cli::{client::Client, server::Server};
-use clap::{Parser, Subcommand};
+use clap::{error::ErrorKind, CommandFactory, Parser, Subcommand};
 
 #[derive(Parser, Debug)]
 #[clap(author, version, about)]
-#[clap(propagate_version = true)]
 struct Args {
     #[clap(subcommand)]
     command: Command,
@@ -15,6 +16,7 @@ enum Command {
     /// Starts a local proxy to the remote server.
     Local {
         /// The local port to expose.
+        #[clap(env = "BORE_LOCAL_PORT")]
         local_port: u16,
 
         /// The local host to expose.
@@ -22,7 +24,7 @@ enum Command {
         local_host: String,
 
         /// Address of the remote server to expose local ports to.
-        #[clap(short, long)]
+        #[clap(short, long, env = "BORE_SERVER")]
         to: String,
 
         /// Optional port on the remote server to select.
@@ -36,13 +38,25 @@ enum Command {
 
     /// Runs the remote proxy server.
     Server {
-        /// Minimum TCP port number to accept.
-        #[clap(long, default_value_t = 1024)]
+        /// Minimum accepted TCP port number.
+        #[clap(long, default_value_t = 1024, env = "BORE_MIN_PORT")]
         min_port: u16,
 
+        /// Maximum accepted TCP port number.
+        #[clap(long, default_value_t = 65535, env = "BORE_MAX_PORT")]
+        max_port: u16,
+
         /// Optional secret for authentication.
         #[clap(short, long, env = "BORE_SECRET", hide_env_values = true)]
         secret: Option<String>,
+
+        /// IP address to bind to, clients must reach this.
+        #[clap(long, default_value = "0.0.0.0")]
+        bind_addr: IpAddr,
+
+        /// IP address where tunnels will listen on, defaults to --bind-addr.
+        #[clap(long)]
+        bind_tunnels: Option<IpAddr>,
     },
 }
 
@@ -59,8 +73,23 @@ async fn run(command: Command) -> Result<()> {
             let client = Client::new(&local_host, local_port, &to, port, secret.as_deref()).await?;
             client.listen().await?;
         }
-        Command::Server { min_port, secret } => {
-            Server::new(min_port, secret.as_deref()).listen().await?;
+        Command::Server {
+            min_port,
+            max_port,
+            secret,
+            bind_addr,
+            bind_tunnels,
+        } => {
+            let port_range = min_port..=max_port;
+            if port_range.is_empty() {
+                Args::command()
+                    .error(ErrorKind::InvalidValue, "port range is empty")
+                    .exit();
+            }
+            let mut server = Server::new(port_range, secret.as_deref());
+            server.set_bind_addr(bind_addr);
+            server.set_bind_tunnels(bind_tunnels.unwrap_or(bind_addr));
+            server.listen().await?;
         }
     }
 

src/server.rs

@@ -1,50 +1,65 @@
 //! Server implementation for the `bore` service.
 
-use std::net::SocketAddr;
-use std::sync::Arc;
-use std::time::Duration;
+use std::net::{IpAddr, Ipv4Addr};
+use std::{io, ops::RangeInclusive, sync::Arc, time::Duration};
 
 use anyhow::Result;
 use dashmap::DashMap;
-use tokio::io::BufReader;
+use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpListener, TcpStream};
 use tokio::time::{sleep, timeout};
 use tracing::{info, info_span, warn, Instrument};
 use uuid::Uuid;
 
 use crate::auth::Authenticator;
-use crate::shared::{
-    proxy, recv_json_timeout, send_json, ClientMessage, ServerMessage, CONTROL_PORT,
-};
+use crate::shared::{ClientMessage, Delimited, ServerMessage, CONTROL_PORT};
 
 /// State structure for the server.
 pub struct Server {
-    /// The minimum TCP port that can be forwarded.
-    min_port: u16,
+    /// Range of TCP ports that can be forwarded.
+    port_range: RangeInclusive<u16>,
 
     /// Optional secret used to authenticate clients.
     auth: Option<Authenticator>,
 
     /// Concurrent map of IDs to incoming connections.
     conns: Arc<DashMap<Uuid, TcpStream>>,
+
+    /// IP address where the control server will bind to.
+    bind_addr: IpAddr,
+
+    /// IP address where tunnels will listen on.
+    bind_tunnels: IpAddr,
 }
 
 impl Server {
     /// Create a new server with a specified minimum port number.
-    pub fn new(min_port: u16, secret: Option<&str>) -> Self {
+    pub fn new(port_range: RangeInclusive<u16>, secret: Option<&str>) -> Self {
+        assert!(!port_range.is_empty(), "must provide at least one port");
         Server {
-            min_port,
+            port_range,
             conns: Arc::new(DashMap::new()),
             auth: secret.map(Authenticator::new),
+            bind_addr: IpAddr::V4(Ipv4Addr::UNSPECIFIED),
+            bind_tunnels: IpAddr::V4(Ipv4Addr::UNSPECIFIED),
         }
     }
 
+    /// Set the IP address where tunnels will listen on.
+    pub fn set_bind_addr(&mut self, bind_addr: IpAddr) {
+        self.bind_addr = bind_addr;
+    }
+
+    /// Set the IP address where the control server will bind to.
+    pub fn set_bind_tunnels(&mut self, bind_tunnels: IpAddr) {
+        self.bind_tunnels = bind_tunnels;
+    }
+
     /// Start the server, listening for new connections.
     pub async fn listen(self) -> Result<()> {
         let this = Arc::new(self);
-        let addr = SocketAddr::from(([0, 0, 0, 0], CONTROL_PORT));
-        let listener = TcpListener::bind(&addr).await?;
-        info!(?addr, "server listening");
+        let listener = TcpListener::bind((this.bind_addr, CONTROL_PORT)).await?;
+        info!(addr = ?this.bind_addr, "server listening");
 
         loop {
             let (stream, addr) = listener.accept().await?;
@@ -63,47 +78,73 @@ impl Server {
         }
     }
 
+    async fn create_listener(&self, port: u16) -> Result<TcpListener, &'static str> {
+        let try_bind = |port: u16| async move {
+            TcpListener::bind((self.bind_tunnels, port))
+                .await
+                .map_err(|err| match err.kind() {
+                    io::ErrorKind::AddrInUse => "port already in use",
+                    io::ErrorKind::PermissionDenied => "permission denied",
+                    _ => "failed to bind to port",
+                })
+        };
+        if port > 0 {
+            // Client requests a specific port number.
+            if !self.port_range.contains(&port) {
+                return Err("client port number not in allowed range");
+            }
+            try_bind(port).await
+        } else {
+            // Client requests any available port in range.
+            //
+            // In this case, we bind to 150 random port numbers. We choose this value because in
+            // order to find a free port with probability at least 1-δ, when ε proportion of the
+            // ports are currently available, it suffices to check approximately -2 ln(δ) / ε
+            // independently and uniformly chosen ports (up to a second-order term in ε).
+            //
+            // Checking 150 times gives us 99.999% success at utilizing 85% of ports under these
+            // conditions, when ε=0.15 and δ=0.00001.
+            for _ in 0..150 {
+                let port = fastrand::u16(self.port_range.clone());
+                match try_bind(port).await {
+                    Ok(listener) => return Ok(listener),
+                    Err(_) => continue,
+                }
+            }
+            Err("failed to find an available port")
+        }
+    }
+
     async fn handle_connection(&self, stream: TcpStream) -> Result<()> {
-        let mut stream = BufReader::new(stream);
+        let mut stream = Delimited::new(stream);
         if let Some(auth) = &self.auth {
             if let Err(err) = auth.server_handshake(&mut stream).await {
                 warn!(%err, "server handshake failed");
-                send_json(&mut stream, ServerMessage::Error(err.to_string())).await?;
+                stream.send(ServerMessage::Error(err.to_string())).await?;
                 return Ok(());
             }
         }
 
-        match recv_json_timeout(&mut stream).await? {
+        match stream.recv_timeout().await? {
             Some(ClientMessage::Authenticate(_)) => {
                 warn!("unexpected authenticate");
                 Ok(())
             }
             Some(ClientMessage::Hello(port)) => {
-                if port != 0 && port < self.min_port {
-                    warn!(?port, "client port number too low");
-                    return Ok(());
-                }
-                info!(?port, "new client");
-                let listener = match TcpListener::bind(("::", port)).await {
+                let listener = match self.create_listener(port).await {
                     Ok(listener) => listener,
-                    Err(_) => {
-                        warn!(?port, "could not bind to local port");
-                        send_json(
-                            &mut stream,
-                            ServerMessage::Error("port already in use".into()),
-                        )
-                        .await?;
+                    Err(err) => {
+                        stream.send(ServerMessage::Error(err.into())).await?;
                         return Ok(());
                     }
                 };
+                let host = listener.local_addr()?.ip();
                 let port = listener.local_addr()?.port();
-                send_json(&mut stream, ServerMessage::Hello(port)).await?;
+                info!(?host, ?port, "new client");
+                stream.send(ServerMessage::Hello(port)).await?;
 
                 loop {
-                    if send_json(&mut stream, ServerMessage::Heartbeat)
-                        .await
-                        .is_err()
-                    {
+                    if stream.send(ServerMessage::Heartbeat).await.is_err() {
                         // Assume that the TCP connection has been dropped.
                         return Ok(());
                     }
@@ -123,28 +164,24 @@ impl Server {
                                 warn!(%id, "removed stale connection");
                             }
                         });
-                        send_json(&mut stream, ServerMessage::Connection(id)).await?;
+                        stream.send(ServerMessage::Connection(id)).await?;
                     }
                 }
             }
             Some(ClientMessage::Accept(id)) => {
                 info!(%id, "forwarding connection");
                 match self.conns.remove(&id) {
-                    Some((_, stream2)) => proxy(stream, stream2).await?,
+                    Some((_, mut stream2)) => {
+                        let mut parts = stream.into_parts();
+                        debug_assert!(parts.write_buf.is_empty(), "framed write buffer not empty");
+                        stream2.write_all(&parts.read_buf).await?;
+                        tokio::io::copy_bidirectional(&mut parts.io, &mut stream2).await?;
+                    }
                     None => warn!(%id, "missing connection"),
                 }
                 Ok(())
             }
-            None => {
-                warn!("unexpected EOF");
-                Ok(())
-            }
+            None => Ok(()),
         }
     }
 }
-
-impl Default for Server {
-    fn default() -> Self {
-        Server::new(1024, None)
-    }
-}

src/shared.rs

@@ -3,16 +3,20 @@
 use std::time::Duration;
 
 use anyhow::{Context, Result};
-use serde::de::DeserializeOwned;
-use serde::{Deserialize, Serialize};
-use tokio::io::{self, AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt};
+use futures_util::{SinkExt, StreamExt};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::time::timeout;
+use tokio_util::codec::{AnyDelimiterCodec, Framed, FramedParts};
 use tracing::trace;
 use uuid::Uuid;
 
 /// TCP port used for control connections with the server.
 pub const CONTROL_PORT: u16 = 7835;
 
+/// Maximum byte length for a JSON frame in the stream.
+pub const MAX_FRAME_LENGTH: usize = 256;
+
 /// Timeout for network connections and initial protocol messages.
 pub const NETWORK_TIMEOUT: Duration = Duration::from_secs(3);
 
@@ -48,55 +52,48 @@ pub enum ServerMessage {
     Error(String),
 }
 
-/// Copy data mutually between two read/write streams.
-pub async fn proxy<S1, S2>(stream1: S1, stream2: S2) -> io::Result<()>
-where
-    S1: AsyncRead + AsyncWrite + Unpin,
-    S2: AsyncRead + AsyncWrite + Unpin,
-{
-    let (mut s1_read, mut s1_write) = io::split(stream1);
-    let (mut s2_read, mut s2_write) = io::split(stream2);
-    tokio::select! {
-        res = io::copy(&mut s1_read, &mut s2_write) => res,
-        res = io::copy(&mut s2_read, &mut s1_write) => res,
-    }?;
-    Ok(())
-}
+/// Transport stream with JSON frames delimited by null characters.
+pub struct Delimited<U>(Framed<U, AnyDelimiterCodec>);
 
-/// Read the next null-delimited JSON instruction from a stream.
-pub async fn recv_json<T: DeserializeOwned>(
-    reader: &mut (impl AsyncBufRead + Unpin),
-    buf: &mut Vec<u8>,
-) -> Result<Option<T>> {
-    trace!("waiting to receive json message");
-    buf.clear();
-    reader.read_until(0, buf).await?;
-    if buf.is_empty() {
-        return Ok(None);
+impl<U: AsyncRead + AsyncWrite + Unpin> Delimited<U> {
+    /// Construct a new delimited stream.
+    pub fn new(stream: U) -> Self {
+        let codec = AnyDelimiterCodec::new_with_max_length(vec![0], vec![0], MAX_FRAME_LENGTH);
+        Self(Framed::new(stream, codec))
     }
-    if buf.last() == Some(&0) {
-        buf.pop();
+
+    /// Read the next null-delimited JSON instruction from a stream.
+    pub async fn recv<T: DeserializeOwned>(&mut self) -> Result<Option<T>> {
+        trace!("waiting to receive json message");
+        if let Some(next_message) = self.0.next().await {
+            let byte_message = next_message.context("frame error, invalid byte length")?;
+            let serialized_obj =
+                serde_json::from_slice(&byte_message).context("unable to parse message")?;
+            Ok(serialized_obj)
+        } else {
+            Ok(None)
+        }
     }
-    Ok(serde_json::from_slice(buf).context("failed to parse JSON")?)
-}
 
-/// Read the next null-delimited JSON instruction, with a default timeout.
-///
-/// This is useful for parsing the initial message of a stream for handshake or
-/// other protocol purposes, where we do not want to wait indefinitely.
-pub async fn recv_json_timeout<T: DeserializeOwned>(
-    reader: &mut (impl AsyncBufRead + Unpin),
-) -> Result<Option<T>> {
-    timeout(NETWORK_TIMEOUT, recv_json(reader, &mut Vec::new()))
-        .await
-        .context("timed out waiting for initial message")?
-}
+    /// Read the next null-delimited JSON instruction, with a default timeout.
+    ///
+    /// This is useful for parsing the initial message of a stream for handshake or
+    /// other protocol purposes, where we do not want to wait indefinitely.
+    pub async fn recv_timeout<T: DeserializeOwned>(&mut self) -> Result<Option<T>> {
+        timeout(NETWORK_TIMEOUT, self.recv())
+            .await
+            .context("timed out waiting for initial message")?
+    }
 
-/// Send a null-terminated JSON instruction on a stream.
-pub async fn send_json<T: Serialize>(writer: &mut (impl AsyncWrite + Unpin), msg: T) -> Result<()> {
-    trace!("sending json message");
-    let msg = serde_json::to_vec(&msg)?;
-    writer.write_all(&msg).await?;
-    writer.write_all(&[0]).await?;
-    Ok(())
+    /// Send a null-terminated JSON instruction on a stream.
+    pub async fn send<T: Serialize>(&mut self, msg: T) -> Result<()> {
+        trace!("sending json message");
+        self.0.send(serde_json::to_string(&msg)?).await?;
+        Ok(())
+    }
+
+    /// Consume this object, returning current buffers and the inner transport.
+    pub fn into_parts(self) -> FramedParts<U, AnyDelimiterCodec> {
+        self.0.into_parts()
+    }
 }

tests/auth_test.rs

@@ -1,14 +1,14 @@
 use anyhow::Result;
-use bore_cli::auth::Authenticator;
-use tokio::io::{self, BufReader};
+use bore_cli::{auth::Authenticator, shared::Delimited};
+use tokio::io::{self};
 
 #[tokio::test]
 async fn auth_handshake() -> Result<()> {
     let auth = Authenticator::new("some secret string");
 
     let (client, server) = io::duplex(8); // Ensure correctness with limited capacity.
-    let mut client = BufReader::new(client);
-    let mut server = BufReader::new(server);
+    let mut client = Delimited::new(client);
+    let mut server = Delimited::new(server);
 
     tokio::try_join!(
         auth.client_handshake(&mut client),
@@ -24,8 +24,8 @@ async fn auth_handshake_fail() {
     let auth2 = Authenticator::new("different server secret");
 
     let (client, server) = io::duplex(8); // Ensure correctness with limited capacity.
-    let mut client = BufReader::new(client);
-    let mut server = BufReader::new(server);
+    let mut client = Delimited::new(client);
+    let mut server = Delimited::new(server);
 
     let result = tokio::try_join!(
         auth.client_handshake(&mut client),

tests/e2e_test.rs

@@ -2,7 +2,7 @@ use std::net::SocketAddr;
 use std::time::Duration;
 
 use anyhow::{anyhow, Result};
-use bore_cli::{client::Client, server::Server};
+use bore_cli::{client::Client, server::Server, shared::CONTROL_PORT};
 use lazy_static::lazy_static;
 use rstest::*;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -17,7 +17,7 @@ lazy_static! {
 
 /// Spawn the server, giving some time for the control port TcpListener to start.
 async fn spawn_server(secret: Option<&str>) {
-    tokio::spawn(Server::new(1024, secret).listen());
+    tokio::spawn(Server::new(1024..=65535, secret).listen());
     time::sleep(Duration::from_millis(50)).await;
 }
 
@@ -26,7 +26,7 @@ async fn spawn_client(secret: Option<&str>) -> Result<(TcpListener, SocketAddr)>
     let listener = TcpListener::bind("localhost:0").await?;
     let local_port = listener.local_addr()?.port();
     let client = Client::new("localhost", local_port, "localhost", 0, secret).await?;
-    let remote_addr = ([0, 0, 0, 0], client.remote_port()).into();
+    let remote_addr = ([127, 0, 0, 1], client.remote_port()).into();
     tokio::spawn(client.listen());
     Ok((listener, remote_addr))
 }
@@ -84,7 +84,7 @@ async fn mismatched_secret(
 async fn invalid_address() -> Result<()> {
     // We don't need the serial guard for this test because it doesn't create a server.
     async fn check_address(to: &str, use_secret: bool) -> Result<()> {
-        match Client::new("localhost", 5000, to, 0, use_secret.then(|| "a secret")).await {
+        match Client::new("localhost", 5000, to, 0, use_secret.then_some("a secret")).await {
             Ok(_) => Err(anyhow!("expected error for {to}, use_secret={use_secret}")),
             Err(_) => Ok(()),
         }
@@ -99,3 +99,66 @@ async fn invalid_address() -> Result<()> {
     )?;
     Ok(())
 }
+
+#[tokio::test]
+async fn very_long_frame() -> Result<()> {
+    let _guard = SERIAL_GUARD.lock().await;
+
+    spawn_server(None).await;
+    let mut attacker = TcpStream::connect(("localhost", CONTROL_PORT)).await?;
+
+    // Slowly send a very long frame.
+    for _ in 0..10 {
+        let result = attacker.write_all(&[42u8; 100000]).await;
+        if result.is_err() {
+            return Ok(());
+        }
+        time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("did not exit after a 1 MB frame");
+}
+
+#[test]
+#[should_panic]
+fn empty_port_range() {
+    let min_port = 5000;
+    let max_port = 3000;
+    let _ = Server::new(min_port..=max_port, None);
+}
+
+#[tokio::test]
+async fn half_closed_tcp_stream() -> Result<()> {
+    // Check that "half-closed" TCP streams will not result in spontaneous hangups.
+    let _guard = SERIAL_GUARD.lock().await;
+
+    spawn_server(None).await;
+    let (listener, addr) = spawn_client(None).await?;
+
+    let (mut cli, (mut srv, _)) = tokio::try_join!(TcpStream::connect(addr), listener.accept())?;
+
+    // Send data before half-closing one of the streams.
+    let mut buf = b"message before shutdown".to_vec();
+    cli.write_all(&buf).await?;
+
+    // Only close the write half of the stream. This is a half-closed stream. In the
+    // TCP protocol, it is represented as a FIN packet on one end. The entire stream
+    // is only closed after two FINs are exchanged and ACKed by the other end.
+    cli.shutdown().await?;
+
+    srv.read_exact(&mut buf).await?;
+    assert_eq!(buf, b"message before shutdown");
+    assert_eq!(srv.read(&mut buf).await?, 0); // EOF
+
+    // Now make sure that the other stream can still send data, despite
+    // half-shutdown on client->server side.
+    let mut buf = b"hello from the other side!".to_vec();
+    srv.write_all(&buf).await?;
+    cli.read_exact(&mut buf).await?;
+    assert_eq!(buf, b"hello from the other side!");
+
+    // We don't have to think about CLOSE_RD handling because that's not really
+    // part of the TCP protocol, just the POSIX streams API. It is implemented by
+    // the OS ignoring future packets received on that stream.
+
+    Ok(())
+}