diff --git a/apps/server/src/cli/connect.ts b/apps/server/src/cli/connect.ts index 9de28623c318..f5c41e4272bf 100644 --- a/apps/server/src/cli/connect.ts +++ b/apps/server/src/cli/connect.ts @@ -6,6 +6,7 @@ import { } from "@t3tools/contracts"; import { RelayOkResponse } from "@t3tools/contracts/relay"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import { describeClockSkew } from "@t3tools/shared/clockSkew"; import * as RelayClient from "@t3tools/shared/relayClient"; import { withRelayClientTracing } from "@t3tools/shared/relayTracing"; import * as Cause from "effect/Cause"; @@ -29,7 +30,8 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as BootService from "../cloud/bootService.ts"; import * as CliState from "../cloud/CliState.ts"; import * as CliTokenManager from "../cloud/CliTokenManager.ts"; -import { filterRelayResponse } from "../cloud/relayResponse.ts"; +import { checkRelayClock } from "../cloud/relayClock.ts"; +import { filterRelayClient } from "../cloud/relayResponse.ts"; import { CLOUD_LINK_REVOKED, CLOUD_LINKED_USER_ID, @@ -143,6 +145,7 @@ interface CloudCliStatus { readonly relayUrl: string | null; readonly publishAgentActivity: boolean; readonly relayClient: RelayClient.RelayClientStatus; + readonly clockSkewSeconds: number | null; } function formatRelayClientStatus(executable: RelayClient.RelayClientStatus): ReadonlyArray { @@ -170,8 +173,11 @@ function formatRelayClientStatus(executable: RelayClient.RelayClientStatus): Rea } } -function formatCloudStatus(status: CloudCliStatus, options?: { readonly json?: boolean }): string { - if (options?.json) { +function formatCloudStatus( + status: CloudCliStatus, + options: { readonly json?: boolean; readonly platform: NodeJS.Platform }, +): string { + if (options.json) { return JSON.stringify(status, null, 2); } @@ -200,6 +206,9 @@ function formatCloudStatus(status: CloudCliStatus, options?: { readonly json?: b ` Relay: ${status.relayUrl ?? "not provisioned"}`, ` Publish agent activity: ${status.publishAgentActivity ? "enabled" : "disabled"}`, ...formatRelayClientStatus(status.relayClient), + ...(status.clockSkewSeconds === null + ? [] + : ["", describeClockSkew(status.clockSkewSeconds, options.platform)]), "", "This is saved setup, not a live connection check. Check the background service with `t3 service status`.", ...(nextStep ? ["", `Next: ${nextStep}`] : []), @@ -334,13 +343,12 @@ const unlinkRelayEnvironment = Effect.fn("cloud.cli.unlink_relay_environment")(f const environment = yield* ServerEnvironment.ServerEnvironmentIdentity; const environmentId = yield* environment.getEnvironmentId; const relayUrl = yield* relayUrlConfig; - const httpClient = yield* HttpClient.HttpClient; + const httpClient = filterRelayClient(yield* HttpClient.HttpClient); const response = yield* HttpClientRequest.delete( `${relayUrl}/v1/client/environment-links/${encodeURIComponent(environmentId)}`, ).pipe( HttpClientRequest.bearerToken(token.value.accessToken), httpClient.execute, - Effect.flatMap(filterRelayResponse), Effect.flatMap(HttpClientResponse.schemaBodyJson(RelayOkResponse)), withRelayClientTracing, ); @@ -580,6 +588,16 @@ const connectStatusCommand = Command.make("status", { ], { concurrency: "unbounded" }, ); + // Before the first link succeeds only the configured relay is known, + // and a wrong clock is one reason that link keeps failing. + const clockRelayUrl = Option.isSome(relayUrl) + ? Option.some(bytesToString(relayUrl.value)) + : desired + ? yield* relayUrlConfig.pipe(Effect.option) + : Option.none(); + const clockSkewSeconds = Option.isSome(clockRelayUrl) + ? yield* checkRelayClock(clockRelayUrl.value) + : null; const status: CloudCliStatus = { desired, authenticated, @@ -591,8 +609,11 @@ const connectStatusCommand = Command.make("status", { Option.isSome(publishAgentActivity) ? bytesToString(publishAgentActivity.value) : null, ), relayClient: executable, + clockSkewSeconds, }; - yield* Console.log(formatCloudStatus(status, { json: flags.json })); + yield* Console.log( + formatCloudStatus(status, { json: flags.json, platform: yield* HostProcess.Platform }), + ); }), { quietLogs: flags.json, diff --git a/apps/server/src/cloud/CloudLink.lifecycle.test.ts b/apps/server/src/cloud/CloudLink.lifecycle.test.ts index faa11d9c1253..214abc0acd3d 100644 --- a/apps/server/src/cloud/CloudLink.lifecycle.test.ts +++ b/apps/server/src/cloud/CloudLink.lifecycle.test.ts @@ -1,6 +1,8 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { describe, expect, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; +import * as Clock from "effect/Clock"; +import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -10,11 +12,15 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as PlatformError from "effect/PlatformError"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as TestClock from "effect/testing/TestClock"; import * as Stream from "effect/Stream"; import { HttpClient, HttpClientResponse, HttpServer, type HttpClientRequest } from "effect/http"; import * as NetAddress from "effect/net/NetAddress"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { decodeRelayJwt } from "@t3tools/shared/relayJwt"; + import { DESKTOP_UPDATE_RESTART_MARKER_FILE, EnvironmentId } from "@t3tools/contracts"; import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; @@ -30,7 +36,10 @@ import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts"; import { CLOUD_CLI_DESIRED_LINK_SECRET } from "./CliState.ts"; import * as CliTokenManager from "./CliTokenManager.ts"; -import { RelayManagedEndpointRecoveryRegistrationRequest } from "@t3tools/contracts/relay"; +import { + RelayManagedEndpointRecoveryRegistrationRequest, + RelayManagedEndpointRecoveryRequest, +} from "@t3tools/contracts/relay"; import { CLOUD_ENDPOINT_CONFIRMED_ORIGIN, CLOUD_ENDPOINT_RUNTIME_CONFIG, @@ -79,6 +88,9 @@ const idleHttpServer = HttpServer.HttpServer.of({ const decodeManagedTunnelRecoveryRegistration = Schema.decodeUnknownEffect( Schema.fromJsonString(RelayManagedEndpointRecoveryRegistrationRequest), ); +const decodeManagedTunnelRecovery = Schema.decodeSync( + Schema.fromJsonString(RelayManagedEndpointRecoveryRequest), +); function makeSecretStore( create: ServerSecretStore.ServerSecretStore["Service"]["create"], @@ -269,9 +281,9 @@ describe("releaseManagedTunnelOnShutdown", () => { return Option.getOrThrow(mtime).getTime(); }); - const provideReleaseHarness = - (harness: ReleaseHarness) => - (effect: Effect.Effect) => + const provideReleaseHarness = (harness: ReleaseHarness) => { + const linkStateLock = Semaphore.makeUnsafe(1); + return (effect: Effect.Effect) => effect.pipe( Effect.provide(CloudLink.layer), Effect.provideService(HttpServer.HttpServer, idleHttpServer), @@ -303,7 +315,7 @@ describe("releaseManagedTunnelOnShutdown", () => { recoveryRequests: Stream.empty, requestRecovery: () => Effect.void, tunnelConnected: Stream.empty, - withLinkStateLock: (effect) => effect, + withLinkStateLock: linkStateLock.withPermits(1), }), ), Effect.provideService( @@ -343,6 +355,7 @@ describe("releaseManagedTunnelOnShutdown", () => { ), Effect.scoped, ); + }; const managedLinkSecrets = [ [CLOUD_ENDPOINT_RUNTIME_CONFIG, "runtime-config"], @@ -684,13 +697,29 @@ describe("releaseManagedTunnelOnShutdown", () => { return Effect.gen(function* () { const link = yield* CloudLink.CloudLink; + expect( + yield* link.startManagedTunnelIfOriginConfirmed("http://127.0.0.1:3773", { + requireConfirmedOrigin: false, + }), + ).toBe(true); expect(yield* link.registerManagedTunnelRecovery("http://127.0.0.1:3773")).toEqual({ status: "revoked", }); expect((yield* link.linkState()).linkRevoked).toBe(true); - - expect(requests).toHaveLength(1); - expect(applyConfigCalls).toEqual([]); + expect( + yield* link.startManagedTunnelIfOriginConfirmed("http://127.0.0.1:3773", { + requireConfirmedOrigin: false, + }), + ).toBe(false); + expect(requests.filter((request) => request.method !== "HEAD")).toHaveLength(1); + expect(applyConfigCalls).toEqual([ + { + providerKind: "cloudflare_tunnel", + connectorToken: "existing-token", + tunnelId: "existing-tunnel", + }, + null, + ]); expect(values.has(RELAY_ENVIRONMENT_CREDENTIAL_SECRET)).toBe(true); expect(values.has(CLOUD_ENDPOINT_RUNTIME_CONFIG)).toBe(true); }).pipe( @@ -712,6 +741,220 @@ describe("releaseManagedTunnelOnShutdown", () => { ); }); + it.effect("does not let a delayed relay success undo a newer revocation", () => + Effect.gen(function* () { + const outcomes = yield* Effect.forEach( + ["registration", "recovery"] as const, + (acceptedRequest) => + Effect.gen(function* () { + const config = { + providerKind: "cloudflare_tunnel", + connectorToken: "existing-token", + tunnelId: "existing-tunnel", + } as const; + const { store, values } = makeMemorySecretStore([ + [CLOUD_ENDPOINT_RUNTIME_CONFIG, JSON.stringify(config)], + [RELAY_URL_SECRET, "https://relay.example.test"], + [CLOUD_LINKED_USER_ID, "user-123"], + [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "environment-credential"], + ]); + const applyConfigCalls: Array = []; + const requests: Array = []; + const requestStarted = yield* Deferred.make(); + const response = yield* Deferred.make(); + + return yield* Effect.gen(function* () { + const link = yield* CloudLink.CloudLink; + const localOrigin = "http://127.0.0.1:3773"; + expect( + yield* link.startManagedTunnelIfOriginConfirmed(localOrigin, { + requireConfirmedOrigin: false, + }), + ).toBe(true); + const pending = yield* ( + acceptedRequest === "registration" + ? link.registerManagedTunnelRecovery(localOrigin).pipe(Effect.asVoid) + : link.recoverManagedTunnel(localOrigin, config).pipe(Effect.asVoid) + ).pipe(Effect.forkChild()); + yield* Deferred.await(requestStarted); + + if (acceptedRequest === "registration") { + const error = yield* Effect.flip(link.recoverManagedTunnel(localOrigin, config)); + expect(error).toMatchObject({ + _tag: "RelayRequestError", + rejection: "unauthorized", + }); + expect(CloudLink.shouldRetryCloudLink(error)).toBe(false); + } else { + expect(yield* link.registerManagedTunnelRecovery(localOrigin)).toEqual({ + status: "revoked", + }); + } + expect(values.has(CLOUD_LINK_REVOKED)).toBe(true); + expect(applyConfigCalls).toEqual([config, null]); + expect(requests.map((request) => request.headers.authorization)).toEqual([ + "Bearer environment-credential", + "Bearer environment-credential", + ]); + const stoppedAt = applyConfigCalls.length; + + yield* Deferred.succeed( + response, + Response.json( + acceptedRequest === "registration" + ? { status: "ready" } + : { + endpoint: { + httpBaseUrl: "https://environment.example.test/", + wsBaseUrl: "wss://environment.example.test/ws", + providerKind: "cloudflare_tunnel", + }, + endpointRuntime: { + ...config, + connectorToken: "new-token", + tunnelId: "new-tunnel", + }, + }, + ), + ); + yield* Fiber.join(pending); + + return { + acceptedRequest, + revoked: values.has(CLOUD_LINK_REVOKED), + configsAfterStop: applyConfigCalls + .slice(stoppedAt) + .filter((config) => config !== null), + }; + }).pipe( + provideReleaseHarness({ + store, + applyConfigCalls, + requests, + onRequest: () => Deferred.succeed(requestStarted, undefined), + respondEffect: Effect.suspend(() => + requests.length === 1 + ? Deferred.await(response) + : Effect.succeed( + Response.json( + { + _tag: "RelayAuthInvalidError", + code: "auth_invalid", + reason: "not_authorized", + traceId: "trace-revoked", + }, + { status: 401, headers: { date: "Thu, 01 Jan 1970 00:00:00 GMT" } }, + ), + ), + ), + }), + ); + }), + ); + + expect(outcomes).toEqual([ + { acceptedRequest: "registration", revoked: true, configsAfterStop: [] }, + { acceptedRequest: "recovery", revoked: true, configsAfterStop: [] }, + ]); + }), + ); + + it.effect("keeps registering a skewed host and reconnects after its clock is corrected", () => { + const config = { + providerKind: "cloudflare_tunnel" as const, + connectorToken: "existing-token", + tunnelId: "existing-tunnel", + }; + const { store, values } = makeMemorySecretStore([ + [CLOUD_ENDPOINT_RUNTIME_CONFIG, JSON.stringify(config)], + [RELAY_URL_SECRET, "https://relay.example.test"], + [CLOUD_LINKED_USER_ID, "user-123"], + [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "environment-credential"], + [CLOUD_CLI_DESIRED_LINK_SECRET, "managed"], + ]); + const applyConfigCalls: Array = []; + const requests: Array = []; + let clockCorrected = false; + + return Effect.gen(function* () { + const link = yield* CloudLink.CloudLink; + const firstFailure = yield* Deferred.make(); + const retryWindowExhausted = yield* Deferred.make(); + const registration = yield* retryManagedTunnelRegistration( + link + .registerManagedTunnelRecovery("http://127.0.0.1:3773") + .pipe(Effect.tapError(() => Deferred.succeed(firstFailure, undefined))), + CloudLink.shouldRetryCloudLink, + Deferred.succeed(retryWindowExhausted, undefined), + ).pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.raceFirst(Deferred.await(firstFailure), Fiber.await(registration)); + const state = yield* link.linkState(); + expect(state).toMatchObject({ + clockSkewSeconds: -420, + linkRevoked: false, + hostPlatform: yield* HostProcess.Platform, + }); + expect(values.has(CLOUD_LINK_REVOKED)).toBe(false); + + // Clock synchronization can take longer than the normal startup retry window. + yield* TestClock.adjust("15 minutes"); + yield* Deferred.await(retryWindowExhausted); + const attempts = requests.filter((request) => request.method === "POST").length; + expect(attempts).toBeGreaterThan(1); + expect(values.has(CLOUD_LINK_REVOKED)).toBe(false); + expect(applyConfigCalls).toEqual([]); + + clockCorrected = true; + yield* TestClock.setTime((yield* Clock.currentTimeMillis) + 420_000); + yield* TestClock.adjust("30 seconds"); + expect(yield* Fiber.join(registration)).toMatchObject({ status: "ready" }); + expect(requests.filter((request) => request.method === "POST").length).toBeGreaterThan( + attempts, + ); + expect(applyConfigCalls).toEqual([config]); + const recoveredState = yield* link.linkState(); + expect(recoveredState).toMatchObject({ linked: true, linkRevoked: false }); + expect(recoveredState.clockSkewSeconds).toBeUndefined(); + expect(new TextDecoder().decode(values.get(RELAY_ENVIRONMENT_CREDENTIAL_SECRET))).toBe( + "environment-credential", + ); + expect( + Option.getOrThrow( + decodeConfirmedOrigin( + new TextDecoder().decode(values.get(CLOUD_ENDPOINT_CONFIRMED_ORIGIN)), + ), + ), + ).toEqual({ config, origin: { localHttpHost: "127.0.0.1", localHttpPort: 3773 } }); + }).pipe( + provideReleaseHarness({ + store, + applyConfigCalls, + requests, + respondEffect: Effect.gen(function* () { + const now = yield* Clock.currentTimeMillis; + return Response.json( + clockCorrected + ? { status: "ready" } + : { + _tag: "RelayAuthInvalidError", + code: "auth_invalid", + reason: "not_authorized", + traceId: "trace-clock", + }, + { + status: clockCorrected ? 200 : 401, + headers: { + date: DateTime.toDateUtc( + DateTime.makeUnsafe(now + (clockCorrected ? 0 : 420_000)), + ).toUTCString(), + }, + }, + ); + }), + }), + ); + }); + it.effect("does not mark a link that was replaced while its rejection was in flight", () => { const { store, values } = makeMemorySecretStore([ [ @@ -1088,13 +1331,21 @@ describe("releaseManagedTunnelOnShutdown", () => { [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "environment-credential"], ]); const applyConfigCalls: Array = []; + let clockCorrected = false; return Effect.gen(function* () { const link = yield* CloudLink.CloudLink; + const skewed = yield* Effect.flip(link.recoverManagedTunnel("http://127.0.0.1:3773")); + expect(CloudLink.shouldRetryCloudLink(skewed)).toBe(true); + expect((yield* link.linkState()).linkRevoked).toBe(false); + expect(values.has(CLOUD_LINK_REVOKED)).toBe(false); + + clockCorrected = true; const error = yield* Effect.flip(link.recoverManagedTunnel("http://127.0.0.1:3773")); + expect(CloudLink.shouldRetryCloudLink(error)).toBe(false); expect(error).toMatchObject({ _tag: "RelayRequestError", rejection: "unauthorized" }); expect((yield* link.linkState()).linkRevoked).toBe(true); - expect(applyConfigCalls).toEqual([]); + expect(applyConfigCalls).toEqual([null]); expect(values.has(RELAY_ENVIRONMENT_CREDENTIAL_SECRET)).toBe(true); }).pipe( provideReleaseHarness({ @@ -1109,12 +1360,186 @@ describe("releaseManagedTunnelOnShutdown", () => { reason: "not_authorized", traceId: "trace-recovery-revoked", }, - { status: 401 }, + { + status: 401, + headers: { + date: clockCorrected + ? "Thu, 01 Jan 1970 00:00:00 GMT" + : "Thu, 01 Jan 1970 00:07:00 GMT", + }, + }, ), }), ); }); + it.effect( + "revokes a genuine recovery refusal delayed in transit with an accurate host clock", + () => + Effect.gen(function* () { + const { store, values } = makeMemorySecretStore([ + [ + CLOUD_ENDPOINT_RUNTIME_CONFIG, + '{"providerKind":"cloudflare_tunnel","connectorToken":"old-token","tunnelId":"old-tunnel"}', + ], + [RELAY_URL_SECRET, "https://relay.example.test"], + [CLOUD_LINKED_USER_ID, "user-123"], + [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "environment-credential"], + ]); + const applyConfigCalls: Array = []; + const requests: Array = []; + const requestStarted = yield* Deferred.make(); + const response = yield* Deferred.make(); + + yield* Effect.gen(function* () { + const link = yield* CloudLink.CloudLink; + const recovery = yield* Effect.flip( + link.recoverManagedTunnel("http://127.0.0.1:3773"), + ).pipe(Effect.forkChild({ startImmediately: true })); + const sentAt = yield* Deferred.await(requestStarted); + const refusal = Response.json( + { + _tag: "RelayAuthInvalidError", + code: "auth_invalid", + reason: "not_authorized", + traceId: "trace-delayed-recovery-refusal", + }, + { + status: 401, + headers: { date: DateTime.toDateUtc(DateTime.makeUnsafe(sentAt)).toUTCString() }, + }, + ); + + // The Date header predates transport delay, not a change to the host's clock. + yield* TestClock.adjust("75 seconds"); + yield* Deferred.succeed(response, refusal); + const error = yield* Fiber.join(recovery); + + expect(error).toMatchObject({ _tag: "RelayRequestError", rejection: "unauthorized" }); + expect(requests).toHaveLength(1); + expect({ + retryable: CloudLink.shouldRetryCloudLink(error), + revoked: values.has(CLOUD_LINK_REVOKED), + applyConfigCalls, + }).toEqual({ retryable: false, revoked: true, applyConfigCalls: [null] }); + }).pipe( + provideReleaseHarness({ + store, + applyConfigCalls, + requests, + onRequest: () => + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => Deferred.succeed(requestStarted, now)), + ), + respondEffect: Deferred.await(response), + }), + ); + }), + ); + + it.effect( + "retries recovery when the host clock corrects while its rejected proof is in flight", + () => + Effect.gen(function* () { + const { store, values } = makeMemorySecretStore([ + [ + CLOUD_ENDPOINT_RUNTIME_CONFIG, + '{"providerKind":"cloudflare_tunnel","connectorToken":"old-token","tunnelId":"old-tunnel"}', + ], + [RELAY_URL_SECRET, "https://relay.example.test"], + [CLOUD_LINKED_USER_ID, "user-123"], + [RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "environment-credential"], + ]); + const nextConfig = { + providerKind: "cloudflare_tunnel" as const, + connectorToken: "new-token", + tunnelId: "new-tunnel", + }; + const applyConfigCalls: Array = []; + const requests: Array = []; + const requestStarted = yield* Deferred.make(); + const response = yield* Deferred.make(); + const clock = yield* Clock.Clock; + let wallOffset = -420_000; + const wallMillis = () => clock.currentTimeMillisUnsafe() + wallOffset; + const hostClock = { + ...clock, + currentTimeMillisUnsafe: wallMillis, + currentTimeMillis: Effect.sync(wallMillis), + currentTimeNanosUnsafe: () => BigInt(wallMillis()) * 1_000_000n, + currentTimeNanos: Effect.sync(() => BigInt(wallMillis()) * 1_000_000n), + }; + yield* TestClock.setTime(420_000); + + yield* Effect.gen(function* () { + const link = yield* CloudLink.CloudLink; + const recovery = yield* Effect.flip( + link.recoverManagedTunnel("http://127.0.0.1:3773"), + ).pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(requestStarted); + const refusal = Response.json( + { + _tag: "RelayAuthInvalidError", + code: "auth_invalid", + reason: "not_authorized", + traceId: "trace-inflight-clock-correction", + }, + { status: 401, headers: { date: "Thu, 01 Jan 1970 00:07:00 GMT" } }, + ); + + // Time sync changes the wall clock, not the elapsed request duration. + yield* TestClock.adjust("2 seconds"); + wallOffset = 0; + yield* Deferred.succeed(response, refusal); + const error = yield* Fiber.join(recovery); + expect(error).toMatchObject({ _tag: "RelayRequestError" }); + expect({ + retryable: CloudLink.shouldRetryCloudLink(error), + revoked: values.has(CLOUD_LINK_REVOKED), + applyConfigCalls, + }).toEqual({ retryable: true, revoked: false, applyConfigCalls: [] }); + + expect(yield* link.recoverManagedTunnel("http://127.0.0.1:3773")).toBe(true); + expect(applyConfigCalls).toEqual([nextConfig]); + expect(values.has(CLOUD_LINK_REVOKED)).toBe(false); + expect(requests).toHaveLength(2); + expect( + requests.map(({ body }) => { + if (body._tag !== "Uint8Array") throw new Error("Expected a JSON recovery request"); + const { proof } = decodeManagedTunnelRecovery(new TextDecoder().decode(body.body)); + const { iat, exp } = decodeRelayJwt(proof); + return { iat, exp }; + }), + ).toEqual([ + { iat: 0, exp: 60 }, + { iat: 422, exp: 482 }, + ]); + }).pipe( + provideReleaseHarness({ + store, + applyConfigCalls, + requests, + onRequest: () => Deferred.succeed(requestStarted, undefined), + respondEffect: Effect.suspend(() => + requests.length === 1 + ? Deferred.await(response) + : Effect.succeed( + Response.json({ + endpoint: { + httpBaseUrl: "https://environment.example.test/", + wsBaseUrl: "wss://environment.example.test/ws", + providerKind: "cloudflare_tunnel", + }, + endpointRuntime: nextConfig, + }), + ), + ), + }), + Effect.provideService(Clock.Clock, hostClock), + ); + }), + ); + it.effect("allows managed tunnel provisioning to take longer than ten seconds", () => Effect.gen(function* () { const oldConfig = @@ -1243,7 +1668,7 @@ describe("releaseManagedTunnelOnShutdown", () => { expect(error).toMatchObject({ _tag: "RelayRequestError", rejection }); expect(CloudLink.shouldRetryCloudLink(error)).toBe(false); expect(requests).toHaveLength(1); - expect(applyConfigCalls).toEqual([]); + expect(applyConfigCalls).toEqual(status === 401 ? [null] : []); }).pipe( provideReleaseHarness({ store, diff --git a/apps/server/src/cloud/CloudLink.ts b/apps/server/src/cloud/CloudLink.ts index 3270f93baca9..f04e5a854091 100644 --- a/apps/server/src/cloud/CloudLink.ts +++ b/apps/server/src/cloud/CloudLink.ts @@ -43,6 +43,7 @@ import { import { isSecureRelayUrl } from "@t3tools/shared/relayUrl"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -96,8 +97,9 @@ import { import { getOrCreateEnvironmentKeyPairFromSecretStore } from "./environmentKeys.ts"; import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts"; import { relayUrlConfig } from "./publicConfig.ts"; +import { checkRelayClock } from "./relayClock.ts"; import { - filterRelayResponse, + filterRelayClient, relayRequestError, type RelayRequestError, shouldRetryRelayRequest, @@ -439,6 +441,7 @@ export class CloudLink extends Context.Service< | EnvironmentAuth.ServerAuthInternalError | RelayRequestError >; + /** The saved link, plus how far this machine's clock is off when that breaks it. */ readonly linkState: () => Effect.Effect< EnvironmentCloudLinkStateResult, CloudLinkInternalError @@ -524,12 +527,15 @@ const make = Effect.gen(function* () { const environmentAuth = yield* EnvironmentAuth.EnvironmentAuth; const cliTokenManager = yield* CliTokenManager.CloudCliTokenManager; const httpClient = yield* HttpClient.HttpClient; + const relayHttpClient = filterRelayClient(httpClient); const httpServer = yield* HttpServer.HttpServer; const awarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; const crypto = yield* Crypto.Crypto; const config = yield* ServerConfig.ServerConfig; const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; + // Accepted responses may heal old markers, but cannot undo a newer refusal. + let revocationRevision = 0; const withSecrets = Effect.provideService(ServerSecretStore.ServerSecretStore, secrets); const withRuntimeFiles = ( @@ -637,19 +643,36 @@ const make = Effect.gen(function* () { Effect.catchTags({ PlatformError: internalError("generate-link-proof") }), ); + const isCurrentManagedTunnel = Effect.fnUntraced(function* (input: { + readonly configJson: string; + readonly credential: string; + readonly revocationRevision: number; + }) { + const [currentConfig, currentCredential] = yield* Effect.all([ + secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG), + secrets.get(RELAY_ENVIRONMENT_CREDENTIAL_SECRET), + ]); + return ( + revocationRevision === input.revocationRevision && + Option.isSome(currentConfig) && + bytesToString(currentConfig.value) === input.configJson && + Option.isSome(currentCredential) && + bytesToString(currentCredential.value) === input.credential + ); + }); + const activateManagedTunnel = Effect.fn("environment.cloud.activateManagedTunnel")( function* (input: { readonly config: RelayManagedEndpointRuntimeConfig; readonly configJson: string; readonly origin: RelayManagedEndpointOrigin; + readonly credential: string; + readonly revocationRevision: number; }) { return yield* endpointRuntime.withLinkStateLock( Effect.gen(function* () { - const currentConfig = yield* secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG); - if ( - Option.isNone(currentConfig) || - bytesToString(currentConfig.value) !== input.configJson - ) { + const revoked = yield* secrets.get(CLOUD_LINK_REVOKED); + if (Option.isSome(revoked) || !(yield* isCurrentManagedTunnel(input))) { return null; } const status = yield* endpointRuntime.applyConfig(input.config); @@ -675,6 +698,8 @@ const make = Effect.gen(function* () { readonly config: RelayManagedEndpointRuntimeConfig; readonly configJson: string; readonly origin: RelayManagedEndpointOrigin; + readonly credential: string; + readonly revocationRevision: number; }, retryRuntimeFailures: boolean, ) => { @@ -708,11 +733,12 @@ const make = Effect.gen(function* () { }); return yield* endpointRuntime.withLinkStateLock( Effect.gen(function* () { - const [runtimeBytes, markerBytes] = yield* Effect.all([ + const [runtimeBytes, markerBytes, revoked] = yield* Effect.all([ secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG), secrets.get(CLOUD_ENDPOINT_CONFIRMED_ORIGIN), + secrets.get(CLOUD_LINK_REVOKED), ]); - if (Option.isNone(runtimeBytes)) return false; + if (Option.isNone(runtimeBytes) || Option.isSome(revoked)) return false; const config = Option.getOrNull(decodeRuntimeConfig(bytesToString(runtimeBytes.value))); if (config === null || config.providerKind !== "cloudflare_tunnel") return false; // With the marker required, only a config the relay already confirmed on @@ -828,8 +854,7 @@ const make = Effect.gen(function* () { HttpClientRequest.post(input.url).pipe( HttpClientRequest.bearerToken(input.token), HttpClientRequest.bodyJson(input.payload), - Effect.flatMap(httpClient.execute), - Effect.flatMap(filterRelayResponse), + Effect.flatMap(relayHttpClient.execute), Effect.flatMap(HttpClientResponse.schemaBodyJson(input.schema)), Effect.timeout(input.timeout ?? "10 seconds"), Effect.mapError(relayRequestError), @@ -967,12 +992,8 @@ const make = Effect.gen(function* () { ); }); - /** - * Nothing stored is removed: the relay also answers 401 for a skewed clock, - * and the next accepted registration clears the mark. Returns false, without - * marking, when a relink replaced the credential while the rejected request - * was in flight. - */ + // Check the credential under the same lock as connector startup so an old + // rejection cannot stop a replacement link. const markLinkRevoked = (credential: string, cause: RelayRequestError) => endpointRuntime.withLinkStateLock( Effect.gen(function* () { @@ -980,18 +1001,26 @@ const make = Effect.gen(function* () { if (Option.isNone(current) || bytesToString(current.value) !== credential) { return false; } + revocationRevision++; yield* Effect.logWarning( "T3 Connect no longer accepts this environment's link. Link it again to restore remote access.", { cause }, ); yield* secrets.set(CLOUD_LINK_REVOKED, stringToBytes("true")); + yield* endpointRuntime.applyConfig(null); return true; }), ); + // The relay also answers 401 while this machine's clock is wrong. That + // refusal is retried instead, since fixing the clock is enough to recover. + const isLinkRevocation = (error: RelayRequestError) => + error.rejection === "unauthorized" && error.clockSkewSeconds === undefined; + const registerManagedTunnelRecovery = Effect.fn( "environment.cloud.registerManagedCloudTunnelRecovery", )(function* (localOrigin: string, options?: { readonly retryRuntimeFailures?: boolean }) { + const requestRevocationRevision = revocationRevision; const [runtimeConfig, relayUrl, cloudUserId, environmentCredential] = yield* Effect.all([ secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG), secrets.get(RELAY_URL_SECRET), @@ -1045,7 +1074,7 @@ const make = Effect.gen(function* () { }).pipe( Effect.catchTags({ RelayRequestError: (error) => - error.rejection === "unauthorized" + isLinkRevocation(error) ? markLinkRevoked(credential, error).pipe( Effect.map((marked) => (marked ? ("revoked" as const) : ("superseded" as const))), ) @@ -1055,16 +1084,26 @@ const make = Effect.gen(function* () { if (registered === "revoked" || registered === "superseded") { return { status: registered }; } - yield* secrets.remove(CLOUD_LINK_REVOKED); + const activation = { + config, + configJson: bytesToString(runtimeConfig.value), + origin, + credential, + revocationRevision: requestRevocationRevision, + }; + const accepted = yield* endpointRuntime.withLinkStateLock( + Effect.gen(function* () { + if (!(yield* isCurrentManagedTunnel(activation))) return false; + yield* secrets.remove(CLOUD_LINK_REVOKED); + return true; + }), + ); + if (!accepted) return { status: "superseded" as const }; if (registered.status === "recovery_required") { return { status: registered.status, config }; } const endpointRuntimeStatus = yield* activateManagedTunnelWithRetry( - { - config, - configJson: bytesToString(runtimeConfig.value), - origin, - }, + activation, options?.retryRuntimeFailures === true, ); return endpointRuntimeStatus === null @@ -1077,6 +1116,7 @@ const make = Effect.gen(function* () { expectedConfig?: RelayManagedEndpointRuntimeConfig, options?: { readonly retryRuntimeFailures?: boolean }, ) { + const requestRevocationRevision = revocationRevision; const [runtimeConfig, relayUrl, cloudUserId, environmentCredential] = yield* Effect.all([ secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG), secrets.get(RELAY_URL_SECRET), @@ -1133,7 +1173,7 @@ const make = Effect.gen(function* () { timeout: MANAGED_ENDPOINT_PROVISION_REQUEST_TIMEOUT, }).pipe( Effect.tapError((error) => - error.rejection === "unauthorized" ? markLinkRevoked(credential, error) : Effect.void, + isLinkRevocation(error) ? markLinkRevoked(credential, error) : Effect.void, ), ); if (recovered.endpointRuntime.providerKind !== "cloudflare_tunnel") { @@ -1147,10 +1187,12 @@ const make = Effect.gen(function* () { ); const stored = yield* endpointRuntime.withLinkStateLock( Effect.gen(function* () { - const currentConfig = yield* secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG); if ( - Option.isNone(currentConfig) || - bytesToString(currentConfig.value) !== bytesToString(runtimeConfig.value) + !(yield* isCurrentManagedTunnel({ + configJson: bytesToString(runtimeConfig.value), + credential, + revocationRevision: requestRevocationRevision, + })) ) { return false; } @@ -1166,6 +1208,8 @@ const make = Effect.gen(function* () { config: recovered.endpointRuntime, configJson: encoded, origin, + credential, + revocationRevision: requestRevocationRevision, }, options?.retryRuntimeFailures === true, ); @@ -1263,8 +1307,7 @@ const make = Effect.gen(function* () { `${bytesToString(relayUrl.value)}/v1/client/environment-links/${encodeURIComponent(environmentId)}/tunnel`, ).pipe( HttpClientRequest.bearerToken(token.value.accessToken), - httpClient.execute, - Effect.flatMap(filterRelayResponse), + relayHttpClient.execute, Effect.flatMap(HttpClientResponse.schemaBodyJson(RelayOkResponse)), withRelayClientTracing, ); @@ -1328,7 +1371,19 @@ const make = Effect.gen(function* () { const linkState = Effect.fn("environment.cloud.linkState")( function* () { - return yield* readCloudLinkState(); + const state = yield* readCloudLinkState(); + const relayUrl = + state.relayUrl ?? + ((yield* readCliDesiredCloudLink.pipe(withSecrets)) + ? Option.getOrNull(yield* relayUrlConfig.pipe(Effect.option)) + : null); + if (relayUrl === null) return state; + const clockSkewSeconds = yield* checkRelayClock(relayUrl).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + ); + return clockSkewSeconds === null + ? state + : { ...state, clockSkewSeconds, hostPlatform: yield* HostProcess.Platform }; }, Effect.catchIf(ServerSecretStore.isSecretStoreError, internalError("read-relay-config")), ); diff --git a/apps/server/src/cloud/managedTunnelStartup.test.ts b/apps/server/src/cloud/managedTunnelStartup.test.ts index e87f3e8e363e..c69becb5cded 100644 --- a/apps/server/src/cloud/managedTunnelStartup.test.ts +++ b/apps/server/src/cloud/managedTunnelStartup.test.ts @@ -1,4 +1,6 @@ import { describe, expect, it } from "@effect/vitest"; +import * as Clock from "effect/Clock"; +import { RelayRequestError, shouldRetryRelayRequest } from "./relayResponse.ts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -6,6 +8,8 @@ import * as TestClock from "effect/testing/TestClock"; import { managedTunnelStartupAction, + makeManagedTunnelRecovery, + retryDesiredManagedTunnelLink, retryManagedTunnelRegistration, } from "./managedTunnelStartup.ts"; @@ -129,3 +133,112 @@ describe("retryManagedTunnelRegistration", () => { }), ); }); + +it.effect( + "retries the first desired link beyond ten minutes until clock correction, but stops on a permanent refusal", + () => + Effect.gen(function* () { + let corrected = false; + let attempts = 0; + const firstAttempt = yield* Deferred.make(); + const reconcile = Effect.suspend(() => { + attempts += 1; + return Deferred.succeed(firstAttempt, undefined).pipe( + Effect.andThen( + corrected + ? Effect.succeed("managed" as const) + : Effect.fail( + new RelayRequestError({ + rejection: "unauthorized", + description: "clock skew", + clockSkewSeconds: -420, + }), + ), + ), + ); + }); + const fiber = yield* Effect.forkChild( + retryDesiredManagedTunnelLink(reconcile, shouldRetryRelayRequest), + { startImmediately: true }, + ); + yield* Deferred.await(firstAttempt); + yield* TestClock.adjust("15 minutes"); + expect(attempts).toBeGreaterThan(20); + corrected = true; + yield* TestClock.adjust("30 seconds"); + expect(yield* Fiber.join(fiber)).toBe("managed"); + let rejectedAttempts = 0; + const error = yield* Effect.flip( + retryDesiredManagedTunnelLink( + Effect.suspend(() => { + rejectedAttempts += 1; + return Effect.fail( + new RelayRequestError({ rejection: "unauthorized", description: "revoked" }), + ); + }), + shouldRetryRelayRequest, + ), + ); + expect(error.rejection).toBe("unauthorized"); + expect(rejectedAttempts).toBe(1); + }), +); + +it.effect("keeps recovery cooldowns through forward and backward wall-clock corrections", () => + Effect.gen(function* () { + const clock = yield* Clock.Clock; + let wallOffset = 3_600_000; + const wallMillis = () => clock.currentTimeMillisUnsafe() + wallOffset; + const hostClock = { + ...clock, + currentTimeMillisUnsafe: wallMillis, + currentTimeMillis: Effect.sync(wallMillis), + currentTimeNanosUnsafe: () => BigInt(wallMillis()) * 1_000_000n, + currentTimeNanos: Effect.sync(() => BigInt(wallMillis()) * 1_000_000n), + }; + const config = { + providerKind: "cloudflare_tunnel" as const, + connectorToken: "token", + tunnelId: "tunnel", + }; + let attempts = 0; + const firstAttempt = yield* Deferred.make(); + const recover = yield* makeManagedTunnelRecovery( + () => + Effect.gen(function* () { + attempts += 1; + if (attempts === 1) { + yield* Deferred.succeed(firstAttempt, undefined); + return yield* new RelayRequestError({ + rejection: "unauthorized", + description: "clock skew", + clockSkewSeconds: 3600, + }); + } + return true; + }), + shouldRetryRelayRequest, + ); + const fiber = yield* Effect.forkChild( + recover(config).pipe(Effect.provideService(Clock.Clock, hostClock)), + { startImmediately: true }, + ); + yield* Deferred.await(firstAttempt); + wallOffset = 0; + yield* TestClock.adjust("119 seconds"); + expect(attempts).toBe(1); + yield* TestClock.adjust("1 second"); + expect(attempts).toBe(2); + expect(yield* Fiber.join(fiber)).toBe(true); + wallOffset = 3_600_000; + const next = yield* Effect.forkChild( + recover(config).pipe(Effect.provideService(Clock.Clock, hostClock)), + { startImmediately: true }, + ); + yield* TestClock.adjust("119 seconds"); + expect(attempts).toBe(2); + yield* TestClock.adjust("1 second"); + expect(attempts).toBe(3); + expect(yield* Fiber.join(next)).toBe(true); + }), +); diff --git a/apps/server/src/cloud/managedTunnelStartup.ts b/apps/server/src/cloud/managedTunnelStartup.ts index bf442e848ad7..baede590c4bf 100644 --- a/apps/server/src/cloud/managedTunnelStartup.ts +++ b/apps/server/src/cloud/managedTunnelStartup.ts @@ -1,8 +1,12 @@ import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay"; +import * as Clock from "effect/Clock"; +import * as Semaphore from "effect/Semaphore"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Schedule from "effect/Schedule"; +import { isRelayClockSkewError } from "./relayResponse.ts"; + // "revoked" leads to no action: relinking a deregistered environment is the // user's call, so startup neither retries nor relinks it. export type ManagedTunnelRegistrationResult = @@ -69,9 +73,58 @@ export const retryManagedTunnelRegistration = ( // Every managed host shares one relay, so a host stuck in a bad loop must not // turn into a fleet-wide request storm. -export const MANAGED_TUNNEL_RECOVERY_COOLDOWN = Duration.minutes(2); +const MANAGED_TUNNEL_RECOVERY_COOLDOWN = Duration.minutes(2); // Existing hosts register on their first boot after an upgrade, and desktop // auto-update delivers that boot to many hosts at once. Spread the first // registration so the relay and Cloudflare see a ramp instead of a spike. export const MANAGED_TUNNEL_FIRST_REGISTRATION_JITTER = Duration.seconds(30); + +export const retryDesiredManagedTunnelLink = ( + reconcile: Effect.Effect, + isRetryable: (error: E) => boolean, +) => + reconcile.pipe( + Effect.retry({ while: isRelayClockSkewError, schedule: Schedule.spaced("30 seconds") }), + Effect.retry({ + while: isRetryable, + schedule: Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), + Schedule.upTo({ duration: "10 minutes" }), + ), + }), + ); + +export const makeManagedTunnelRecovery = Effect.fnUntraced(function* ( + recover: (config: RelayManagedEndpointRuntimeConfig) => Effect.Effect, + isRetryable: (error: E) => boolean, +) { + const lock = yield* Semaphore.make(1); + let lastAttemptNanos: bigint | null = null; + return (config: RelayManagedEndpointRuntimeConfig) => + lock.withPermits(1)( + Effect.gen(function* () { + // Clock corrections must neither bypass nor extend the relay cooldown. + const now = yield* Clock.monotonicTimeNanos; + if (lastAttemptNanos !== null) { + const elapsedMillis = Number(now - lastAttemptNanos) / 1_000_000; + const wait = Duration.toMillis(MANAGED_TUNNEL_RECOVERY_COOLDOWN) - elapsedMillis; + if (wait > 0) yield* Effect.sleep(Duration.millis(wait)); + } + lastAttemptNanos = yield* Clock.monotonicTimeNanos; + return yield* recover(config); + }).pipe( + Effect.retry({ + while: isRetryable, + schedule: Schedule.exponential("1 second").pipe( + Schedule.modifyDelay(({ duration }) => + Effect.succeed(Duration.min(duration, Duration.seconds(30))), + ), + Schedule.jittered, + ), + }), + ), + ); +}); diff --git a/apps/server/src/cloud/relayClock.ts b/apps/server/src/cloud/relayClock.ts new file mode 100644 index 000000000000..170acc73a87f --- /dev/null +++ b/apps/server/src/cloud/relayClock.ts @@ -0,0 +1,22 @@ +import { clockSkewFromDateHeader } from "@t3tools/shared/clockSkew"; +import * as Clock from "effect/Clock"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import { HttpClient, HttpClientRequest } from "effect/http"; + +// Status also works before authorization succeeds, so this probe needs no credentials. +export const checkRelayClock = Effect.fn("cloud.checkRelayClock")( + function* (relayUrl: string) { + const httpClient = yield* HttpClient.HttpClient; + const [duration, response] = yield* httpClient + .execute(HttpClientRequest.head(`${relayUrl}/`)) + .pipe(Effect.timed); + return clockSkewFromDateHeader( + response.headers.date, + yield* Clock.currentTimeMillis, + Duration.toMillis(duration), + ); + }, + Effect.timeout("2 seconds"), + Effect.orElseSucceed(() => null), +); diff --git a/apps/server/src/cloud/relayResponse.test.ts b/apps/server/src/cloud/relayResponse.test.ts index 59f341aa3c47..96bcd1b18eb1 100644 --- a/apps/server/src/cloud/relayResponse.test.ts +++ b/apps/server/src/cloud/relayResponse.test.ts @@ -1,5 +1,7 @@ import { expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as TestClock from "effect/testing/TestClock"; import { HttpClientRequest, HttpClientResponse } from "effect/http"; import * as HttpClientError from "effect/http/HttpClientError"; @@ -43,18 +45,24 @@ it("reports a transport failure category without exposing request or cause detai it.effect("reports the tunnel limit and relay trace instead of a generic 403", () => Effect.gen(function* () { const error = yield* filterRelayResponse( - response(403, { - _tag: "RelayEnvironmentLinkLimitExceededError", - code: "environment_link_limit_exceeded", - maxTunnels: 3, - traceId: "trace-limit", - }), + response( + 403, + { + _tag: "RelayEnvironmentLinkLimitExceededError", + code: "environment_link_limit_exceeded", + maxTunnels: 3, + traceId: "trace-limit", + }, + { date: "Thu, 01 Jan 1970 00:07:00 GMT" }, + ), ).pipe(Effect.mapError(relayRequestError), Effect.flip); expect(error).toMatchObject({ _tag: "RelayRequestError", rejection: "forbidden" }); expect(error.message).toContain("at most 3 tunnels"); expect(error.message).toContain("Unlink an unused environment"); expect(error.message).toContain("Trace ID: trace-limit"); + expect(shouldRetryRelayRequest(error)).toBe(false); + expect(error.clockSkewSeconds).toBeUndefined(); }), ); @@ -76,6 +84,39 @@ it.effect("makes revoked authorization actionable and non-retryable", () => }), ); +it.effect("keeps a rejected proof retryable until the host clock is within 60 seconds", () => + Effect.gen(function* () { + const rejectedProof = () => + filterRelayResponse( + response( + 400, + { + _tag: "RelayEnvironmentLinkProofInvalidError", + code: "environment_link_proof_invalid", + reason: "invalid_signature_or_scope", + traceId: "trace-clock", + }, + { date: "Thu, 01 Jan 1970 00:07:00 GMT" }, + ), + ).pipe(Effect.flip); + + const skewed = yield* rejectedProof(); + expect(skewed.clockSkewSeconds).toBe(-420); + expect(shouldRetryRelayRequest(skewed)).toBe(true); + expect(skewed.message).toContain("clock is 7 minutes behind"); + expect(skewed.message).toContain('"Sync now"'); + expect(skewed.message).toContain("reconnects by itself"); + expect(skewed.message).toContain("Trace ID: trace-clock"); + + yield* TestClock.setTime(359_000); + expect(shouldRetryRelayRequest(yield* rejectedProof())).toBe(true); + yield* TestClock.setTime(360_000); + const withinTolerance = yield* rejectedProof(); + expect(withinTolerance.clockSkewSeconds).toBeUndefined(); + expect(shouldRetryRelayRequest(withinTolerance)).toBe(false); + }).pipe(Effect.provideService(HostProcess.Platform, "win32")), +); + it.effect("retries an expired link proof but not a revoked bearer, both 401", () => Effect.gen(function* () { const expired = yield* filterRelayResponse( diff --git a/apps/server/src/cloud/relayResponse.ts b/apps/server/src/cloud/relayResponse.ts index f86cc5719401..e9a40b4bc63b 100644 --- a/apps/server/src/cloud/relayResponse.ts +++ b/apps/server/src/cloud/relayResponse.ts @@ -1,20 +1,22 @@ import { RelayProtectedError } from "@t3tools/contracts/relay"; +import { clockSkewFromDateHeader, describeClockSkew } from "@t3tools/shared/clockSkew"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Clock from "effect/Clock"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as HttpClient from "effect/http/HttpClient"; import * as HttpClientResponse from "effect/http/HttpClientResponse"; import { isHttpClientError } from "effect/http/HttpClientError"; -/** - * `unavailable` means it may succeed on retry; the relay refused the other - * kinds. The description is safe to show: it carries the relay's own - * explanation and trace ID, never request details. - */ +// Descriptions carry relay diagnostics and trace IDs, never request credentials. export class RelayRequestError extends Schema.TaggedError()( "RelayRequestError", { rejection: Schema.Literals(["unauthorized", "forbidden", "rejected", "unavailable"]), description: Schema.String, + clockSkewSeconds: Schema.optional(Schema.Number), }, ) { override get message(): string { @@ -34,7 +36,13 @@ export function relayRequestError(cause: unknown): RelayRequestError { } export const shouldRetryRelayRequest = (error: unknown): boolean => - !isRelayRequestError(error) || error.rejection === "unavailable"; + !isRelayRequestError(error) || + error.rejection === "unavailable" || + error.clockSkewSeconds !== undefined; + +/** A relay refusal that came while this machine's clock was too far off to trust. */ +export const isRelayClockSkewError = (error: unknown): error is RelayRequestError => + isRelayRequestError(error) && error.clockSkewSeconds !== undefined; function recoveryHint(error: RelayProtectedError): string { switch (error._tag) { @@ -44,7 +52,7 @@ function recoveryHint(error: RelayProtectedError): string { return "Run `t3 connect login` to check this machine's authorization. If the stored credential was revoked, sign out with `t3 connect logout`, then run `t3 connect` again. Restart T3 Code after signing in."; case "RelayEnvironmentLinkProofExpiredError": case "RelayEnvironmentLinkProofInvalidError": - return "Check this machine's date and time, update T3 Code, then restart it."; + return "Check this machine's date and time. T3 Connect retries expired proofs automatically; update T3 Code if this continues."; default: return "Retry when the relay is available. If this continues, include the trace ID when reporting it."; } @@ -52,8 +60,16 @@ function recoveryHint(error: RelayProtectedError): string { export const filterRelayResponse = Effect.fn("cloud.filter_relay_response")(function* ( response: HttpClientResponse.HttpClientResponse, + requestDurationMillis = 0, + requestStartedAtMillis?: number, ) { if (response.status >= 200 && response.status < 300) return response; + const nowMillis = yield* Clock.currentTimeMillis; + const clockSkewSeconds = clockSkewFromDateHeader( + response.headers.date, + nowMillis, + requestDurationMillis, + ); const decoded = yield* HttpClientResponse.schemaBodyJson(RelayProtectedError)(response).pipe( Effect.option, ); @@ -62,23 +78,59 @@ export const filterRelayResponse = Effect.fn("cloud.filter_relay_response")(func const description = Option.isSome(decoded) ? `T3 Connect: ${decoded.value.message}. ${recoveryHint(decoded.value)} Trace ID: ${decoded.value.traceId}.` : `T3 Connect relay returned HTTP ${response.status} without a recognized error response. Check relay access and any proxy or firewall restrictions, then restart T3 Code.${requestId}`; + // The proof predates any in-flight clock correction. Project the starting + // wall clock forward with monotonic time to check the clock that signed it. + if ( + (response.status === 400 || response.status === 401) && + clockSkewSeconds === null && + requestStartedAtMillis !== undefined && + clockSkewFromDateHeader( + response.headers.date, + requestStartedAtMillis + requestDurationMillis, + requestDurationMillis, + ) !== null + ) { + return yield* new RelayRequestError({ + rejection: "unavailable", + description: `The host computer's clock changed during the T3 Connect relay request. T3 Connect will retry with a fresh proof. Relay response: ${ + Option.isSome(decoded) + ? `${decoded.value.message}. Trace ID: ${decoded.value.traceId}.` + : `HTTP ${response.status}.${requestId}` + }`, + }); + } + // Timestamp rejections can clear after clock sync; unrelated HTTP statuses stay permanent. + const clockSuspected = + clockSkewSeconds !== null && (response.status === 400 || response.status === 401); + const platform = yield* HostProcess.Platform; + const refusal = (rejection: RelayRequestError["rejection"]) => + !clockSuspected + ? new RelayRequestError({ rejection, description }) + : new RelayRequestError({ + rejection, + // The relay's own recovery advice does not apply until the clock is + // fixed, so keep only what it said and its trace. + description: `${describeClockSkew(clockSkewSeconds, platform)} T3 Connect reconnects by itself once the clock is correct. Relay response: ${ + Option.isSome(decoded) + ? `${decoded.value.message}. Trace ID: ${decoded.value.traceId}.` + : `HTTP ${response.status}.${requestId}` + }`, + clockSkewSeconds, + }); - // The only 401 that clears itself: a proof signed with a skewed clock (a host - // that booted before NTP) expires on arrival, and the next attempt signs a new - // one. It is recognized only from a schema-decoded relay body. Every other - // 401, including a revoked bearer or an undecodable body, stays permanent. + // A decoded expired proof remains retryable even without a usable Date header. if ( response.status === 401 && Option.isSome(decoded) && decoded.value._tag === "RelayEnvironmentLinkProofExpiredError" ) { - return yield* new RelayRequestError({ rejection: "unavailable", description }); + return yield* refusal("unavailable"); } if (response.status === 401) { - return yield* new RelayRequestError({ rejection: "unauthorized", description }); + return yield* refusal("unauthorized"); } if (response.status === 403) { - return yield* new RelayRequestError({ rejection: "forbidden", description }); + return yield* refusal("forbidden"); } if ( response.status >= 400 && @@ -86,7 +138,16 @@ export const filterRelayResponse = Effect.fn("cloud.filter_relay_response")(func response.status !== 408 && response.status !== 429 ) { - return yield* new RelayRequestError({ rejection: "rejected", description }); + return yield* refusal("rejected"); } return yield* new RelayRequestError({ rejection: "unavailable", description }); }); + +export const filterRelayClient = (client: HttpClient.HttpClient) => + HttpClient.transformResponse(client, (response) => + Effect.gen(function* () { + const startedAtMillis = yield* Clock.currentTimeMillis; + const [duration, result] = yield* response.pipe(Effect.timed); + return yield* filterRelayResponse(result, Duration.toMillis(duration), startedAtMillis); + }), + ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 59daedc15107..c16893fac411 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -1,8 +1,5 @@ import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers"; -import type { RelayManagedEndpointRuntimeConfig } from "@t3tools/contracts/relay"; -import * as Clock from "effect/Clock"; import * as Random from "effect/Random"; -import * as Semaphore from "effect/Semaphore"; import * as DatabaseStorage from "./persistence/DatabaseStorage.ts"; import * as StorageCleanup from "./storageCleanup.ts"; import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts"; @@ -21,7 +18,6 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; -import * as Schedule from "effect/Schedule"; import { FetchHttpClient, HttpRouter, HttpServer } from "effect/http"; import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder"; @@ -140,11 +136,13 @@ import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as CloudHttp from "./cloud/http.ts"; import * as CloudLink from "./cloud/CloudLink.ts"; import { pendingServiceUpdateExists } from "./cloud/updateHandoff.ts"; +import { isRelayClockSkewError } from "./cloud/relayResponse.ts"; import * as RelayTracing from "./cloud/relayTracing.ts"; import * as CloudManagedEndpointRuntime from "./cloud/ManagedEndpointRuntime.ts"; import { MANAGED_TUNNEL_FIRST_REGISTRATION_JITTER, - MANAGED_TUNNEL_RECOVERY_COOLDOWN, + makeManagedTunnelRecovery, + retryDesiredManagedTunnelLink, managedTunnelStartupAction, retryManagedTunnelRegistration, } from "./cloud/managedTunnelStartup.ts"; @@ -849,32 +847,31 @@ const layerMakeServer = Layer.unwrap( if (typeof address === "string" || !("port" in address)) return; const localOrigin = `http://127.0.0.1:${address.port}`; const endpointRuntime = yield* CloudManagedEndpointRuntime.CloudManagedEndpointRuntime; - const recoveryLock = yield* Semaphore.make(1); - let lastRecoveryAtMillis = 0; - const recoverManagedTunnel = (config: RelayManagedEndpointRuntimeConfig) => - recoveryLock.withPermits(1)( - Effect.gen(function* () { - const elapsed = (yield* Clock.currentTimeMillis) - lastRecoveryAtMillis; - const wait = Duration.toMillis(MANAGED_TUNNEL_RECOVERY_COOLDOWN) - elapsed; - if (wait > 0) yield* Effect.sleep(Duration.millis(wait)); - lastRecoveryAtMillis = yield* Clock.currentTimeMillis; - }).pipe( - Effect.andThen( - cloudLink.recoverManagedTunnel(localOrigin, config, { - retryRuntimeFailures: true, - }), - ), - Effect.retry({ - while: (error) => - CloudLink.shouldRetryCloudLink(error) && - error._tag !== "CloudLinkEndpointUnavailableError", - schedule: Schedule.exponential("1 second").pipe( - Schedule.modifyDelay(({ duration }) => - Effect.succeed(Duration.min(duration, Duration.seconds(30))), - ), - Schedule.jittered, + let clockSkewReported = false; + const reportClockSkew = (error: unknown) => + isRelayClockSkewError(error) && !clockSkewReported + ? Effect.logWarning(error.message).pipe( + Effect.andThen( + Effect.sync(() => { + clockSkewReported = true; + }), ), - }), + ) + : Effect.void; + const recoverManagedTunnel = yield* makeManagedTunnelRecovery( + (config) => + cloudLink + .recoverManagedTunnel(localOrigin, config, { + retryRuntimeFailures: true, + }) + .pipe(Effect.tapError(reportClockSkew)), + (error) => + CloudLink.shouldRetryCloudLink(error) && + error._tag !== "CloudLinkEndpointUnavailableError", + ); + yield* endpointRuntime.recoveryRequests.pipe( + Stream.runForEach((config) => + recoverManagedTunnel(config).pipe( Effect.tap((recovered) => recovered ? Effect.logInfo("T3 Connect managed tunnel recovered") : Effect.void, ), @@ -886,9 +883,7 @@ const layerMakeServer = Layer.unwrap( }), ), ), - ); - yield* endpointRuntime.recoveryRequests.pipe( - Stream.runForEach(recoverManagedTunnel), + ), Effect.forkScoped, ); const wantsCliLink = hasCloudPublicConfig @@ -941,9 +936,11 @@ const layerMakeServer = Layer.unwrap( Effect.asVoid, ); const registerManagedTunnel = retryManagedTunnelRegistration( - cloudLink.registerManagedTunnelRecovery(localOrigin, { - retryRuntimeFailures: true, - }), + cloudLink + .registerManagedTunnelRecovery(localOrigin, { + retryRuntimeFailures: true, + }) + .pipe(Effect.tapError(reportClockSkew)), (error) => CloudLink.shouldRetryCloudLink(error) && error._tag !== "CloudLinkEndpointUnavailableError", @@ -983,29 +980,23 @@ const layerMakeServer = Layer.unwrap( yield* endpointRuntime.requestRecovery(startupAction.config); } if (startupAction.action === "reconcile_link") { - const reconciledMode = yield* cloudLink - .reconcileDesiredLinkIfStillDesired(localOrigin) - .pipe( - Effect.retry({ - while: CloudLink.shouldRetryCloudLink, - schedule: Schedule.exponential("1 second").pipe( - Schedule.modifyDelay(({ duration }) => - Effect.succeed(Duration.min(duration, Duration.seconds(30))), - ), - Schedule.upTo({ duration: "10 minutes" }), - ), - }), - Effect.tap((mode) => - mode === null - ? Effect.void - : Effect.logInfo("T3 Connect desired link reconciled on startup"), - ), - Effect.catch((cause) => - Effect.logWarning("Failed to reconcile T3 Connect desired link on startup", { - cause, - }).pipe(Effect.as(null)), - ), - ); + const reconciledMode = yield* retryDesiredManagedTunnelLink( + cloudLink + .reconcileDesiredLinkIfStillDesired(localOrigin) + .pipe(Effect.tapError(reportClockSkew)), + CloudLink.shouldRetryCloudLink, + ).pipe( + Effect.tap((mode) => + mode === null + ? Effect.void + : Effect.logInfo("T3 Connect desired link reconciled on startup"), + ), + Effect.catch((cause) => + Effect.logWarning("Failed to reconcile T3 Connect desired link on startup", { + cause, + }).pipe(Effect.as(null)), + ), + ); if (reconciledMode === "managed") { const afterReconcile = yield* registerManagedTunnel; if (afterReconcile.status === "recovery_required") { diff --git a/apps/web/src/components/settings/ConnectionsSettings.tsx b/apps/web/src/components/settings/ConnectionsSettings.tsx index b465b39cd2e9..ce21ee146c79 100644 --- a/apps/web/src/components/settings/ConnectionsSettings.tsx +++ b/apps/web/src/components/settings/ConnectionsSettings.tsx @@ -1,5 +1,6 @@ import { SessionPermissions } from "./SessionPermissions"; import { AUTH_SCOPE_OPTIONS as PAIRING_SCOPE_OPTIONS } from "@t3tools/shared/authScopeOptions"; +import { describeClockSkew } from "@t3tools/shared/clockSkew"; import { ChevronRightIcon, ChevronsLeftRightEllipsisIcon, @@ -1806,6 +1807,18 @@ function ConfiguredCloudLinkRow({ canManageRelay }: { readonly canManageRelay: b ? "Your session does not have permission to manage T3 Connect access." : null; const isBusy = isUpdating || isUpdatingPreference; + const clockSkewSeconds = primaryCloudLinkState.data?.clockSkewSeconds; + const clockSkewWarning = + clockSkewSeconds === undefined + ? null + : describeClockSkew(clockSkewSeconds, primaryCloudLinkState.data?.hostPlatform ?? ""); + // The fix happens outside the app, so check again when the user comes back. + const refreshLinkState = primaryCloudLinkState.refresh; + useEffect(() => { + if (clockSkewWarning === null) return; + window.addEventListener("focus", refreshLinkState); + return () => window.removeEventListener("focus", refreshLinkState); + }, [clockSkewWarning, refreshLinkState]); const updateManagedTunnel = async (enabled: boolean) => { if ( @@ -1886,7 +1899,7 @@ function ConfiguredCloudLinkRow({ canManageRelay }: { readonly canManageRelay: b ? "T3 Connect is out of sync. This environment is linked only to publish agent activity, so other devices cannot connect. Turn T3 Connect on to repair it." : "Make this environment available to your other devices through T3 Connect." } - status={operationError ?? primaryCloudLinkState.error} + status={operationError ?? clockSkewWarning ?? primaryCloudLinkState.error} control={ } /> - ) : linkRevoked ? ( + ) : linkRevoked || clockSkewWarning !== null ? ( ) : null} { expect(relayProtectedErrorMessage(error)).toBe("Relay rejected the cloud session token."); }); + it("points clients to the host clock only when a link proof could have a time failure", () => { + const expired = new RelayEnvironmentLinkProofExpiredError({ + code: "environment_link_proof_expired", + traceId: "trace-expired", + }); + const invalid = new RelayEnvironmentLinkProofInvalidError({ + code: "environment_link_proof_invalid", + reason: "invalid_signature_or_scope", + traceId: "trace-invalid", + }); + const mismatch = new RelayEnvironmentLinkProofInvalidError({ + code: "environment_link_proof_invalid", + reason: "descriptor_mismatch", + traceId: "trace-mismatch", + }); + + expect(relayProtectedErrorMessage(expired)).toContain( + "computer running T3 Code sets its date and time automatically", + ); + expect(relayProtectedErrorMessage(invalid)).toContain( + "computer running T3 Code sets its date and time automatically", + ); + expect(relayProtectedErrorMessage(mismatch)).toBe( + "Relay rejected the environment link proof (descriptor_mismatch).", + ); + }); }); diff --git a/packages/client-runtime/src/relay/errorPresentation.ts b/packages/client-runtime/src/relay/errorPresentation.ts index d6df6985019a..9175c92d4f2a 100644 --- a/packages/client-runtime/src/relay/errorPresentation.ts +++ b/packages/client-runtime/src/relay/errorPresentation.ts @@ -11,6 +11,10 @@ export const DPOP_UNKNOWN_HINT = export const DPOP_RETRY_HINT = "Hint: Try again. If the problem continues, copy the trace ID."; +/** The relay reports a host clock outside its window as an expired or unverifiable link proof. */ +const LINK_PROOF_CLOCK_HINT = + "Hint: Check that the computer running T3 Code sets its date and time automatically, then try again."; + function dpopFailureHint(reason: DpopFailureReason | undefined): string { if (reason === "time_window") return DPOP_CLOCK_HINT; if (reason === undefined) return DPOP_UNKNOWN_HINT; @@ -34,9 +38,11 @@ export function relayProtectedErrorMessage(error: RelayProtectedError): string { return "Relay rejected the authenticated request."; } case "RelayEnvironmentLinkProofExpiredError": - return "Relay rejected an expired environment link proof."; + return `Relay rejected an expired environment link proof. ${LINK_PROOF_CLOCK_HINT}`; case "RelayEnvironmentLinkProofInvalidError": - return `Relay rejected the environment link proof (${error.reason}).`; + return error.reason === "invalid_signature_or_scope" + ? `Relay rejected the environment link proof (${error.reason}). ${LINK_PROOF_CLOCK_HINT}` + : `Relay rejected the environment link proof (${error.reason}).`; case "RelayEnvironmentConnectNotAuthorizedError": // "Not authorized" covers non-auth causes too; surface the reason so a // missing link does not read as a credential problem. diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 31b69997f43c..fa278f34b5b8 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -417,6 +417,10 @@ export const EnvironmentCloudLinkStateResult = Schema.Struct({ publishAgentActivity: Schema.Boolean, holdWebhooksWhileOffline: Schema.optional(Schema.Boolean), linkRevoked: Schema.optional(Schema.Boolean), + // Seconds this environment's clock is ahead of (positive) or behind the + // relay's, present only when the gap is large enough to break T3 Connect. + clockSkewSeconds: Schema.optional(Schema.Number), + hostPlatform: Schema.optional(Schema.String), }); export type EnvironmentCloudLinkStateResult = typeof EnvironmentCloudLinkStateResult.Type; diff --git a/packages/shared/package.json b/packages/shared/package.json index 22f240aa163a..0406080cb1f5 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -195,6 +195,10 @@ "types": "./src/relayJwt.ts", "import": "./src/relayJwt.ts" }, + "./clockSkew": { + "types": "./src/clockSkew.ts", + "import": "./src/clockSkew.ts" + }, "./oauthScope": { "types": "./src/oauthScope.ts", "import": "./src/oauthScope.ts" diff --git a/packages/shared/src/clockSkew.ts b/packages/shared/src/clockSkew.ts new file mode 100644 index 000000000000..b036074f1b7f --- /dev/null +++ b/packages/shared/src/clockSkew.ts @@ -0,0 +1,63 @@ +// The relay permits proofs issued up to 60 seconds in its future. +const CLOCK_SKEW_LIMIT_SECONDS = 60; + +/** + * Seconds this machine's clock is ahead of (positive) or behind (negative) the + * relay, judged from the HTTP `Date` header of a relay response received at + * `nowMillis`, allowing for the request's monotonic elapsed time. `null` when + * the header is missing, unparseable, or does not establish skew beyond the limit. + */ +export function clockSkewFromDateHeader( + dateHeader: string | undefined, + nowMillis: number, + requestDurationMillis = 0, +): number | null { + if (dateHeader === undefined) return null; + const relayMillis = Date.parse(dateHeader); + if (Number.isNaN(relayMillis)) return null; + // `Date` has whole-second precision, so the relay's true time is up to a + // second later than the header says. + const skewSeconds = Math.round((nowMillis - relayMillis - 500) / 1_000); + // Date may originate anywhere between request start and receipt. Transport + // delay alone cannot establish that the host clock is ahead. + const minimumSkewSeconds = Math.round( + (nowMillis - relayMillis - 500 - requestDurationMillis) / 1_000, + ); + return minimumSkewSeconds > CLOCK_SKEW_LIMIT_SECONDS + ? minimumSkewSeconds + : skewSeconds < -CLOCK_SKEW_LIMIT_SECONDS + ? skewSeconds + : null; +} + +function skewAmount(seconds: number): string { + const magnitude = Math.abs(seconds); + const minutes = Math.round(magnitude / 60); + const hours = Math.round(magnitude / (60 * 60)); + const [value, unit] = + magnitude < 90 + ? [magnitude, "second"] + : minutes < 60 + ? [minutes, "minute"] + : hours < 36 + ? [hours, "hour"] + : [Math.round(magnitude / (24 * 60 * 60)), "day"]; + return `${value} ${unit}${value === 1 ? "" : "s"}`; +} + +function clockFixInstructions(platform: string): string { + if (platform === "win32") { + return 'Open Settings > Time & language > Date & time, turn on "Set time automatically", then press "Sync now".'; + } + if (platform === "darwin") { + return 'Open System Settings > General > Date & Time and turn on "Set time and date automatically".'; + } + return platform === "linux" + ? "Turn on network time sync, for example with `timedatectl set-ntp true`." + : "Turn on automatic date and time on the computer running T3 Code."; +} + +export function describeClockSkew(seconds: number, platform: string): string { + const direction = seconds > 0 ? "ahead" : "behind"; + return `The host computer's clock is ${skewAmount(seconds)} ${direction}, and T3 Connect needs it to be right. ${clockFixInstructions(platform)}`; +}