Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 61 additions & 0 deletions packages/client-runtime/src/authorization/layer.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import {
AuthStandardClientScopes,
EnvironmentId,
ORCHESTRATION_PROTOCOL_VERSION,
type ExecutionEnvironmentDescriptor,
} from "@t3tools/contracts";
import { type RelayEnvironmentConnectResponse } from "@t3tools/contracts/relay";
import { describe, expect, it } from "@effect/vitest";
Expand Down Expand Up @@ -317,6 +318,66 @@ describe("RemoteEnvironmentAuthorization", () => {
}),
);

it.effect("reuses a matching descriptor a check just read, once, while fresh, over HTTPS", () =>
Effect.gen(function* () {
const descriptor = {
...DESCRIPTOR,
orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
} as ExecutionEnvironmentDescriptor;
const harness = yield* makeHarness({
responses: [
websocketTicket("reused"),
...Array.from({ length: 5 }, () => [
Response.json(descriptor),
websocketTicket("fresh"),
]).flat(),
],
});

yield* Effect.gen(function* () {
const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization;
const authorizeAt = (httpBaseUrl: string) =>
remote.authorizeBearer({
expectedEnvironmentId: ENVIRONMENT_ID,
httpBaseUrl,
wsBaseUrl: httpBaseUrl.replace(/^http/, "ws"),
bearerToken: "bearer-token",
connectionMethod: "direct",
});
const authorize = authorizeAt(ENDPOINT.httpBaseUrl);
const remember = (value: Option.Option<ExecutionEnvironmentDescriptor>) =>
remote.rememberCheckedDescriptor(ENDPOINT.httpBaseUrl, value);

yield* remember(Option.some(descriptor));
yield* authorize;
// Used once.
yield* authorize;
// Stale.
yield* remember(Option.some(descriptor));
yield* TestClock.adjust("6 seconds");
yield* authorize;
// A mismatch reads again, so a host back on the right environment is not blocked.
yield* remember(Option.some({ ...descriptor, environmentId: "other" as never }));
yield* authorize;
// A failed check drops an older match.
yield* remember(Option.some(descriptor));
yield* remember(Option.none());
yield* authorize;
// After a network change, a plain LAN address can belong to another machine.
yield* remote.rememberCheckedDescriptor("http://192.168.1.2:3773", Option.some(descriptor));
yield* authorizeAt("http://192.168.1.2:3773");
}).pipe(Effect.provide(harness.layer));

expect(harness.fetch.calls.map(([url]) => new URL(String(url)).pathname)).toEqual([
"/api/auth/websocket-ticket/rpc",
...Array.from({ length: 5 }, () => [
"/.well-known/t3/environment",
"/api/auth/websocket-ticket/rpc",
]).flat(),
]);
}),
);

it.effect("blocks a warm direct bearer prepare when the environment changes identity", () =>
Effect.gen(function* () {
const harness = yield* makeHarness({
Expand Down
40 changes: 37 additions & 3 deletions packages/client-runtime/src/authorization/service.ts
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,11 @@ export class RemoteEnvironmentAuthorization extends Context.Service<
readonly expectedEnvironmentId: EnvironmentId;
readonly rejectedAccessToken?: string;
}) => Effect.Effect<AuthorizedRemoteHttpEnvironment, ConnectionAttemptError>;
/** Lets the next direct authorization of this address skip its own descriptor read. */
readonly rememberCheckedDescriptor: (
httpBaseUrl: string,
descriptor: Option.Option<ExecutionEnvironmentDescriptor>,
) => Effect.Effect<void>;
/**
* Renews a live socket's token before it expires, so the server moves the
* socket to the new session instead of closing it. Runs until interrupted.
Expand All @@ -95,6 +100,7 @@ export class RemoteEnvironmentAuthorization extends Context.Service<

const CACHED_ENDPOINT_SOCKET_TIMEOUT_MS = 3_000;
const DPOP_AUTHORIZATION_TIMEOUT_MS = 30_000;
const CHECKED_DESCRIPTOR_TTL_NANOS = 5_000_000_000n;

function mapDpopSocketError(error: RemoteEnvironmentAuthError | ConnectionAttemptError) {
return error._tag === "ConnectionTransientError" || error._tag === "ConnectionBlockedError"
Expand Down Expand Up @@ -135,6 +141,14 @@ export const make = Effect.gen(function* () {
}
>();

// A route or host check hands its descriptor to the connect it leads to, once,
// so the connect skips a round trip. Only over HTTPS: a plain LAN address can
// belong to another machine after a network change.
const checkedDescriptors = new Map<
string,
{ readonly descriptor: ExecutionEnvironmentDescriptor; readonly at: bigint }
>();

/**
* A saved LAN address can belong to another machine on a different network.
*/
Expand All @@ -144,9 +158,18 @@ export const make = Effect.gen(function* () {
readonly httpBaseUrl: string;
readonly connectionMethod: ClientConnectionMethod;
}) {
const descriptor = yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe(
Effect.provideService(HttpClient.HttpClient, httpClient),
);
const checked = checkedDescriptors.get(input.httpBaseUrl);
checkedDescriptors.delete(input.httpBaseUrl);
// A remembered mismatch still reads again, so a host that answers for the
// right environment a moment later is not blocked.
const descriptor =
checked !== undefined &&
checked.descriptor.environmentId === input.expectedEnvironmentId &&
(yield* Clock.monotonicTimeNanos) - checked.at <= CHECKED_DESCRIPTOR_TTL_NANOS
? checked.descriptor
: yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe(
Effect.provideService(HttpClient.HttpClient, httpClient),
);
if (descriptor.environmentId !== input.expectedEnvironmentId) {
return yield* environmentMismatchError({
expected: input.expectedEnvironmentId,
Expand Down Expand Up @@ -527,6 +550,17 @@ export const make = Effect.gen(function* () {

return RemoteEnvironmentAuthorization.of({
authorizeBearer,
// Untraced: it runs on every route and host check.
rememberCheckedDescriptor: Effect.fnUntraced(function* (httpBaseUrl, descriptor) {
if (Option.isNone(descriptor) || !httpBaseUrl.startsWith("https:")) {
checkedDescriptors.delete(httpBaseUrl);
return;
}
checkedDescriptors.set(httpBaseUrl, {
descriptor: descriptor.value,
at: yield* Clock.monotonicTimeNanos,
});
}),
authorizeDpop: Effect.fn("environment.authorization")(function (input) {
return authorizeDpop(input);
}),
Expand Down
10 changes: 8 additions & 2 deletions packages/client-runtime/src/connection/driver.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import * as Context from "effect/Context";
import * as Effect from "effect/Effect";
import * as Option from "effect/Option";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Layer from "effect/Layer";
Expand Down Expand Up @@ -152,10 +153,15 @@ export const make = Effect.gen(function* () {
httpBaseUrl,
timeoutMs: ROUTE_CHECK_TIMEOUT_MS,
}).pipe(
Effect.option,
// The connect this check leads to skips its own read. Every answer, even a
// failure, replaces what an older check remembered.
Effect.tap((descriptor) => remote.rememberCheckedDescriptor(httpBaseUrl, descriptor)),
Effect.map((descriptor): RouteCheck =>
descriptor.environmentId === environmentId ? "answered" : "silent",
Option.exists(descriptor, (value) => value.environmentId === environmentId)
? "answered"
: "silent",
),
Effect.orElseSucceed((): RouteCheck => "silent"),
// A cached descriptor would answer for a host that cannot be reached.
Effect.provideService(FetchHttpClient.RequestInit, { cache: "no-store" }),
Effect.provideService(HttpClient.HttpClient, httpClient),
Expand Down
1 change: 1 addition & 0 deletions packages/client-runtime/src/connection/resolver.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@ const makeDependencies = Effect.fn("TestConnectionResolver.makeDependencies")((o
},
})),
authorizeDpopHttp: () => Effect.die("unused"),
rememberCheckedDescriptor: () => Effect.void,
renewDpopBeforeExpiry: () => Effect.die("unused"),
});
const ssh = ClientCapabilities.SshEnvironmentGateway.of({
Expand Down
12 changes: 7 additions & 5 deletions packages/client-runtime/src/rpc/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,12 @@ export class EnvironmentRpcSubscriptionObserver extends Context.Reference<{
export const isRpcClientError = Schema.is(RpcClientError.RpcClientError);
const isEnvironmentAuthorizationError = Schema.is(EnvironmentAuthorizationError);

/** A subscription never retries an authorization failure on its own. */
export const isAuthorizationFailure = (cause: Cause.Cause<unknown>) =>
cause.reasons.some(
(reason) => reason._tag === "Fail" && isEnvironmentAuthorizationError(reason.error),
);

const MAX_EXPECTED_FAILURE_RETRY_DELAY_MS = 30_000;

export type EnvironmentRpcInput<TTag extends EnvironmentRpcTag> = Parameters<RpcMethod<TTag>>[0];
Expand Down Expand Up @@ -359,13 +365,9 @@ function subscribeDynamicMapped<TTag extends EnvironmentSubscriptionRpcTag, A>(
const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe(
Stream.drain,
);
const isAuthorizationFailure = cause.reasons.some(
(reason) =>
reason._tag === "Fail" && isEnvironmentAuthorizationError(reason.error),
);
if (
options.retryExpectedFailureAfter === undefined ||
isAuthorizationFailure
isAuthorizationFailure(cause)
) {
return handled;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,7 @@ function makeHarness(reply: (requestNumber: number) => Response | Promise<Respon
},
};
}),
rememberCheckedDescriptor: () => Effect.void,
renewDpopBeforeExpiry: () => Effect.die("HTTP requests must not renew a socket's token."),
});
const signer = ManagedRelay.ManagedRelayDpopSigner.of({
Expand Down
Loading
Loading