Add support for Connect-Protocol-Version header

packages/connect-node-test/src/badweather/broken-input.spec.ts

@@ -46,6 +46,7 @@ describe("broken input", () => {
               ),
               method: "POST",
               ctype: "application/json",
+              headers: { "Connect-Protocol-Version": "1" },
             }).then((res) => {
               return {
                 status: res.status,
@@ -85,6 +86,7 @@ describe("broken input", () => {
               url: createMethodUrl(server.getUrl(), TestService, method),
               method: "POST",
               ctype: "application/connect+json",
+              headers: { "Connect-Protocol-Version": "1" },
             }).then((res) => ({
               status: res.status,
               endStream: endStreamFromJson(res.body.subarray(5)),

packages/connect-node-test/src/badweather/unsupported-content-encoding.spec.ts

@@ -44,6 +44,7 @@ describe("unsupported content encoding", () => {
             headers: {
               "content-type": "application/json",
               "content-encoding": "banana",
+              "connect-protocol-version": "1",
             },
             rejectUnauthorized,
           });
@@ -74,6 +75,7 @@ describe("unsupported content encoding", () => {
             headers: {
               "content-type": "application/connect+json",
               "connect-content-encoding": "banana",
+              "connect-protocol-version": "1",
             },
             rejectUnauthorized,
           });

packages/connect-node-test/src/helpers/testserver.ts

@@ -116,7 +116,10 @@ export function createTestServers() {
                 cert: certLocalhost.cert,
                 key: certLocalhost.key,
               },
-              connectNodeAdapter({ routes: testRoutes })
+              connectNodeAdapter({
+                routes: testRoutes,
+                requireConnectProtocolHeader: true,
+              })
             )
             .listen(0, resolve);
         });
@@ -145,7 +148,13 @@ export function createTestServers() {
       start() {
         return new Promise<void>((resolve) => {
           nodeH2cServer = http2
-            .createServer({}, connectNodeAdapter({ routes: testRoutes }))
+            .createServer(
+              {},
+              connectNodeAdapter({
+                routes: testRoutes,
+                requireConnectProtocolHeader: true,
+              })
+            )
             .listen(0, resolve);
         });
       },
@@ -201,7 +210,10 @@ export function createTestServers() {
             "Access-Control-Expose-Headers": corsExposeHeaders.join(", "),
             "Access-Control-Max-Age": 2 * 3600,
           };
-          const serviceHandler = connectNodeAdapter({ routes: testRoutes });
+          const serviceHandler = connectNodeAdapter({
+            routes: testRoutes,
+            requireConnectProtocolHeader: true,
+          });
           nodeHttpServer = http
             .createServer({}, (req, res) => {
               if (req.method === "OPTIONS") {
@@ -243,7 +255,10 @@ export function createTestServers() {
                 cert: certLocalhost.cert,
                 key: certLocalhost.key,
               },
-              connectNodeAdapter({ routes: testRoutes })
+              connectNodeAdapter({
+                routes: testRoutes,
+                requireConnectProtocolHeader: true,
+              })
             )
             .listen(0, resolve);
         });
@@ -280,6 +295,7 @@ export function createTestServers() {
         });
         await fastifyH2cServer.register(fastifyConnectPlugin, {
           routes: testRoutes,
+          requireConnectProtocolHeader: true,
         });
         await fastifyH2cServer.listen();
       },
@@ -306,6 +322,7 @@ export function createTestServers() {
         app.use(
           expressConnectMiddleware({
             routes: testRoutes,
+            requireConnectProtocolHeader: true,
           })
         );
         expressServer = http.createServer(app);

packages/connect/src/protocol-connect/handler-factory.spec.ts

@@ -0,0 +1,193 @@
+// Copyright 2021-2023 Buf Technologies, Inc.
+//
+// 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.
+
+import {
+  Int32Value,
+  Message,
+  MethodInfo,
+  MethodKind,
+  ServiceType,
+  StringValue,
+} from "@bufbuild/protobuf";
+import { createHandlerFactory } from "./handler-factory.js";
+import {
+  createMethodImplSpec,
+  HandlerContext,
+  MethodImpl,
+} from "../implementation.js";
+import type { UniversalHandlerOptions } from "../protocol/index.js";
+import { errorFromJsonBytes } from "./error-json.js";
+import { ConnectError } from "../connect-error.js";
+import { Code } from "../code.js";
+import { endStreamFromJson } from "./end-stream.js";
+import {
+  createAsyncIterableBytes,
+  readAllBytes,
+} from "../protocol/async-iterable-helper.spec.js";
+
+describe("createHandlerFactory()", function () {
+  const testService: ServiceType = {
+    typeName: "TestService",
+    methods: {
+      foo: {
+        name: "Foo",
+        I: Int32Value,
+        O: StringValue,
+        kind: MethodKind.Unary,
+      },
+      bar: {
+        name: "Bar",
+        I: Int32Value,
+        O: StringValue,
+        kind: MethodKind.ServerStreaming,
+      },
+    },
+  } as const;
+
+  function stub<M extends MethodInfo>(
+    opt: {
+      service?: ServiceType;
+      method?: M;
+      impl?: MethodImpl<M>;
+    } & Partial<UniversalHandlerOptions>
+  ) {
+    const method = opt.method ?? testService.methods.foo;
+    let implDefault: MethodImpl<M>;
+    switch (method.kind) {
+      case MethodKind.Unary:
+        // eslint-disable-next-line @typescript-eslint/require-await
+        implDefault = async function (req: Message, ctx: HandlerContext) {
+          ctx.responseHeader.set("stub-handler", "1");
+          return new ctx.method.O();
+        } as unknown as MethodImpl<M>;
+        break;
+      case MethodKind.ServerStreaming:
+        // eslint-disable-next-line @typescript-eslint/require-await
+        implDefault = async function* (req: Message, ctx: HandlerContext) {
+          ctx.responseHeader.set("stub-handler", "1");
+          yield new ctx.method.O();
+        } as unknown as MethodImpl<M>;
+        break;
+      default:
+        throw new Error("not implemented");
+    }
+    const spec = createMethodImplSpec(
+      opt.service ?? testService,
+      method,
+      opt.impl ?? implDefault
+    );
+    const f = createHandlerFactory(opt);
+    return f(spec);
+  }
+
+  describe("requireConnectProtocolHeader", function () {
+    describe("with unary RPC", function () {
+      const h = stub({ requireConnectProtocolHeader: true });
+      it("should raise error for missing header", async function () {
+        const res = await h({
+          httpVersion: "1.1",
+          method: "POST",
+          url: new URL("https://example.com"),
+          header: new Headers({ "Content-Type": "application/json" }),
+          body: 777,
+        });
+        expect(res.status).toBe(400);
+        expect(res.body).toBeInstanceOf(Uint8Array);
+        if (res.body instanceof Uint8Array) {
+          const err = errorFromJsonBytes(
+            res.body,
+            undefined,
+            new ConnectError("failed to parse connect err", Code.Internal)
+          );
+          expect(err.message).toBe(
+            '[invalid_argument] missing required header: set Connect-Protocol-Version to "1"'
+          );
+        }
+      });
+      it("should raise error for wrong header", async function () {
+        const res = await h({
+          httpVersion: "1.1",
+          method: "POST",
+          url: new URL("https://example.com"),
+          header: new Headers({
+            "Content-Type": "application/json",
+            "Connect-Protocol-Version": "UNEXPECTED",
+          }),
+          body: 777,
+        });
+        expect(res.status).toBe(400);
+        expect(res.body).toBeInstanceOf(Uint8Array);
+        if (res.body instanceof Uint8Array) {
+          const err = errorFromJsonBytes(
+            res.body,
+            undefined,
+            new ConnectError("failed to parse connect err", Code.Internal)
+          );
+          expect(err.message).toBe(
+            '[invalid_argument] Connect-Protocol-Version must be "1": got "UNEXPECTED"'
+          );
+        }
+      });
+    });
+    describe("with streaming RPC", function () {
+      const h = stub({
+        requireConnectProtocolHeader: true,
+        method: testService.methods.bar,
+      });
+      it("should raise error for missing header", async function () {
+        const res = await h({
+          httpVersion: "1.1",
+          method: "POST",
+          url: new URL("https://example.com"),
+          header: new Headers({ "Content-Type": "application/connect+json" }),
+          body: createAsyncIterableBytes(new Uint8Array()),
+        });
+        expect(res.status).toBe(200);
+        expect(res.body).not.toBeInstanceOf(Uint8Array);
+        expect(res.body).not.toBeUndefined();
+        if (res.body !== undefined && Symbol.asyncIterator in res.body) {
+          const end = endStreamFromJson(
+            (await readAllBytes(res.body)).slice(5)
+          );
+          expect(end.error?.message).toBe(
+            '[invalid_argument] missing required header: set Connect-Protocol-Version to "1"'
+          );
+        }
+      });
+      it("should raise error for wrong header", async function () {
+        const res = await h({
+          httpVersion: "1.1",
+          method: "POST",
+          url: new URL("https://example.com"),
+          header: new Headers({
+            "Content-Type": "application/connect+json",
+            "Connect-Protocol-Version": "UNEXPECTED",
+          }),
+          body: createAsyncIterableBytes(new Uint8Array()),
+        });
+        expect(res.status).toBe(200);
+        expect(res.body).not.toBeInstanceOf(Uint8Array);
+        expect(res.body).not.toBeUndefined();
+        if (res.body !== undefined && Symbol.asyncIterator in res.body) {
+          const end = endStreamFromJson(
+            (await readAllBytes(res.body)).slice(5)
+          );
+          expect(end.error?.message).toBe(
+            '[invalid_argument] Connect-Protocol-Version must be "1": got "UNEXPECTED"'
+          );
+        }
+      });
+    });
+  });
+});

packages/connect/src/protocol-connect/handler-factory.ts

@@ -73,6 +73,7 @@ import {
 import { codeToHttpStatus } from "./http-status.js";
 import { errorToJsonBytes } from "./error-json.js";
 import { trailerMux } from "./trailer-mux.js";
+import { requireProtocolVersion } from "./version.js";
 
 const protocolName = "connect";
 const methodPost = "POST";
@@ -153,6 +154,7 @@ function createUnaryHandler<I extends Message<I>, O extends Message<O>>(
     let status = uResponseOk.status;
     let body: Uint8Array;
     try {
+      if (opt.requireConnectProtocolHeader) requireProtocolVersion(req.header);
       // raise compression error to serialize it as a error response
       if (compression.error) {
         throw compression.error;
@@ -278,6 +280,8 @@ function createStreamHandler<I extends Message<I>, O extends Message<O>>(
     const outputIt = pipe(
       req.body,
       transformPrepend<Uint8Array>(() => {
+        if (opt.requireConnectProtocolHeader)
+          requireProtocolVersion(req.header);
         // raise compression error to serialize it as the end stream response
         if (compression.error) throw compression.error;
         return undefined;

packages/connect/src/protocol-connect/version.ts

@@ -12,7 +12,30 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+import { headerProtocolVersion } from "./headers.js";
+import { ConnectError } from "../connect-error.js";
+import { Code } from "../code.js";
+
 /**
  * The only know value for the header Connect-Protocol-Version.
  */
 export const protocolVersion = "1";
+
+/**
+ * Requires the Connect-Protocol-Version header to be present with the expected
+ * value. Raises a ConnectError with Code.InvalidArgument otherwise.
+ */
+export function requireProtocolVersion(requestHeader: Headers) {
+  const v = requestHeader.get(headerProtocolVersion);
+  if (v === null) {
+    throw new ConnectError(
+      `missing required header: set ${headerProtocolVersion} to "${protocolVersion}"`,
+      Code.InvalidArgument
+    );
+  } else if (v !== protocolVersion) {
+    throw new ConnectError(
+      `${headerProtocolVersion} must be "${protocolVersion}": got "${v}"`,
+      Code.InvalidArgument
+    );
+  }
+}

packages/connect/src/protocol-grpc-web/trailer.spec.ts

@@ -60,7 +60,7 @@ describe("trailerSerialize()", () => {
   });
 });
 
-describe("roundtrip", () => {
+describe("trailer roundtrip", () => {
   it("should work", () => {
     const a = new Headers({
       foo: "a, b",