Skip to content

Commit

Permalink
Add support for Connect-Protocol-Version header
Browse files Browse the repository at this point in the history
  • Loading branch information
timostamm committed Mar 8, 2023
1 parent a453c23 commit f4ff382
Show file tree
Hide file tree
Showing 12 changed files with 579 additions and 36 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ describe("broken input", () => {
),
method: "POST",
ctype: "application/json",
headers: { "Connect-Protocol-Version": "1" },
}).then((res) => {
return {
status: res.status,
Expand Down Expand Up @@ -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)),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ describe("unsupported content encoding", () => {
headers: {
"content-type": "application/json",
"content-encoding": "banana",
"connect-protocol-version": "1",
},
rejectUnauthorized,
});
Expand Down Expand Up @@ -74,6 +75,7 @@ describe("unsupported content encoding", () => {
headers: {
"content-type": "application/connect+json",
"connect-content-encoding": "banana",
"connect-protocol-version": "1",
},
rejectUnauthorized,
});
Expand Down
25 changes: 21 additions & 4 deletions packages/connect-node-test/src/helpers/testserver.ts
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,10 @@ export function createTestServers() {
cert: certLocalhost.cert,
key: certLocalhost.key,
},
connectNodeAdapter({ routes: testRoutes })
connectNodeAdapter({
routes: testRoutes,
requireConnectProtocolHeader: true,
})
)
.listen(0, resolve);
});
Expand Down Expand Up @@ -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);
});
},
Expand Down Expand Up @@ -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") {
Expand Down Expand Up @@ -243,7 +255,10 @@ export function createTestServers() {
cert: certLocalhost.cert,
key: certLocalhost.key,
},
connectNodeAdapter({ routes: testRoutes })
connectNodeAdapter({
routes: testRoutes,
requireConnectProtocolHeader: true,
})
)
.listen(0, resolve);
});
Expand Down Expand Up @@ -280,6 +295,7 @@ export function createTestServers() {
});
await fastifyH2cServer.register(fastifyConnectPlugin, {
routes: testRoutes,
requireConnectProtocolHeader: true,
});
await fastifyH2cServer.listen();
},
Expand All @@ -306,6 +322,7 @@ export function createTestServers() {
app.use(
expressConnectMiddleware({
routes: testRoutes,
requireConnectProtocolHeader: true,
})
);
expressServer = http.createServer(app);
Expand Down
File renamed without changes.
193 changes: 193 additions & 0 deletions packages/connect/src/protocol-connect/handler-factory.spec.ts
Original file line number Diff line number Diff line change
@@ -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"'
);
}
});
});
});
});
4 changes: 4 additions & 0 deletions packages/connect/src/protocol-connect/handler-factory.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
23 changes: 23 additions & 0 deletions packages/connect/src/protocol-connect/version.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
);
}
}
2 changes: 1 addition & 1 deletion packages/connect/src/protocol-grpc-web/trailer.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ describe("trailerSerialize()", () => {
});
});

describe("roundtrip", () => {
describe("trailer roundtrip", () => {
it("should work", () => {
const a = new Headers({
foo: "a, b",
Expand Down
Loading

0 comments on commit f4ff382

Please sign in to comment.