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
10 changes: 8 additions & 2 deletions apps/cluster/src/Chat/Egress/Client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,18 +54,24 @@ export const makeEgressContext = (
answers.set(key, asked)
return asked
}
/** A credential the workspace couldn't hand out: the request goes on without one, and says so. */
const withoutCredential = (op: EgressQuery["op"]) => (error: LookupFailed) =>
Effect.logWarning(
"Chat egress couldn't get a credential from the workspace",
error.reason,
).pipe(Effect.annotateLogs({ op }), Effect.as(undefined))
return {
sessionId: props.sessionId,
turnId: props.turnId,
// A workspace that can't be reached hands out no token: the request goes without one.
multicaToken: ask({ op: "multica" }).pipe(
Effect.map((token) => (token as string | null) ?? undefined),
Effect.orElseSucceed(() => undefined),
Effect.catch(withoutCredential("multica")),
),
githubTokens: (owner) =>
ask({ op: "tokens", owner }).pipe(
Effect.map((tokens) => (tokens as GitHubTokens | null) ?? undefined),
Effect.orElseSucceed(() => undefined),
Effect.catch(withoutCredential("tokens")),
),
github: {
repository: (repository) =>
Expand Down
51 changes: 49 additions & 2 deletions apps/cluster/src/Chat/Egress/Egress.ts
Original file line number Diff line number Diff line change
Expand Up @@ -621,6 +621,47 @@ const apiCredential = (
: none
})

/** Upstream answers worth chasing: auth failures, rate limits and server errors. */
const isUpstreamError = (status: number) =>
status === 401 || status === 403 || status === 429 || status >= 500

/** The only response headers an upstream error's log line carries, under its annotation names. */
const UPSTREAM_ERROR_HEADERS = [
["requestId", "x-github-request-id"],
["retryAfter", "retry-after"],
["rateLimitRemaining", "x-ratelimit-remaining"],
["rateLimitReset", "x-ratelimit-reset"],
["rateLimitResource", "x-ratelimit-resource"],
] as const

/**
* Logs an upstream error on a GitHub or Multica request with the credential
* that was actually sent. Never the token, the body or the request headers,
* and no query string.
*/
const logUpstreamError = (
request: Request,
url: URL,
credential: "read" | "write" | "multica" | "none",
response: Response,
) => {
if (!isUpstreamError(response.status)) return Effect.void
const annotations: Record<string, unknown> = {
method: request.method,
host: url.host,
path: url.pathname,
credential,
status: response.status,
}
for (const [name, header] of UPSTREAM_ERROR_HEADERS) {
const value = response.headers.get(header)
if (value !== null) annotations[name] = value
}
return Effect.logWarning("Chat egress got an error from upstream").pipe(
Effect.annotateLogs(annotations),
)
}

const refusal = (request: Request, url: URL, reason: string) =>
new Response(
`janitor's egress proxy refused ${request.method} ${url.host}${url.pathname}: ${reason}\n`,
Expand Down Expand Up @@ -654,7 +695,9 @@ export const handleChatEgress = (request: Request, egress: ChatEgress): Effect.E
if (url.protocol === "https:" && url.hostname === MULTICA_HOST) {
const token = yield* egress.multicaToken
if (token !== undefined) headers.set("authorization", `Bearer ${token}`)
return yield* forward()
const response = yield* forward()
yield* logUpstreamError(request, url, token === undefined ? "none" : "multica", response)
return response
}
if (!isGitHubRequest(url)) return yield* forward()

Expand All @@ -672,9 +715,11 @@ export const handleChatEgress = (request: Request, egress: ChatEgress): Effect.E
)
if (decided instanceof Refused) return refusal(request, url, decided.reason)
const { credential, body } = decided
let sent: "read" | "write" | "none" = "none"
if (credential !== undefined) {
const tokens = yield* egress.githubTokens(credential.owner)
if (tokens !== undefined) {
sent = credential.access
const token = credential.access === "read" ? tokens.read : tokens.write
headers.set(
"authorization",
Expand All @@ -684,5 +729,7 @@ export const handleChatEgress = (request: Request, egress: ChatEgress): Effect.E
)
}
}
return yield* forward(body)
const response = yield* forward(body)
yield* logUpstreamError(request, url, sent, response)
return response
}).pipe(Effect.annotateLogs({ sessionId: egress.sessionId, turnId: egress.turnId }))
77 changes: 74 additions & 3 deletions apps/cluster/test/Chat/Egress/Egress.test.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { assert, describe, it } from "@effect/vitest"
import * as Effect from "effect/Effect"
import * as Logger from "effect/Logger"
import {
handleChatEgress,
LookupFailed,
Expand Down Expand Up @@ -47,9 +48,12 @@ const key = (repository: Repository) => `${repository.owner}/${repository.name}`

/**
* A sandbox's egress against the GitHub above. Every forwarded request is
* recorded and answered 200.
* recorded and answered 200, unless `answer` says otherwise.
*/
const makeEgress = (forwarded: Array<Request>): ChatEgress => ({
const makeEgress = (
forwarded: Array<Request>,
answer: () => Response = () => new Response("upstream"),
): ChatEgress => ({
sessionId: "slack/T1/C1/1730000000.000100",
turnId: "turn-1",
githubTokens: (owner) => Effect.succeed(TOKENS[owner.toLowerCase()]),
Expand Down Expand Up @@ -77,7 +81,7 @@ const makeEgress = (forwarded: Array<Request>): ChatEgress => ({
upstream: (request) =>
Effect.sync(() => {
forwarded.push(request)
return new Response("upstream")
return answer()
}),
})

Expand Down Expand Up @@ -668,4 +672,71 @@ describe("chat egress proxy", () => {
),
]),
)

it.effect(
"an upstream error on a push is logged with what it takes to chase it, never the token",
() =>
Effect.gen(function* () {
const logs: Array<string> = []
const capture = Logger.make((options) => {
logs.push(Logger.formatJson.log(options))
})
const limited = () =>
new Response("Too Many Requests", {
status: 429,
headers: {
"retry-after": "60",
"x-github-request-id": "ABCD:1234:5678",
"x-ratelimit-remaining": "0",
"x-ratelimit-reset": "1730000060",
"x-ratelimit-resource": "git",
},
})
const request = () =>
new Request("https://github.com/Effect-TS/effect.git/git-receive-pack", {
method: "POST",
body: receivePack([update("refs/heads/feature")]),
})
const forwarded: Array<Request> = []
yield* handleChatEgress(request(), makeEgress(forwarded, limited)).pipe(
Effect.andThen(handleChatEgress(request(), makeEgress(forwarded))),
Effect.provide(Logger.layer([capture])),
)

const entries = logs.map(
(line) => JSON.parse(line) as { level: string; annotations: Record<string, unknown> },
)
assert.deepStrictEqual(
{
forwarded: forwarded.length,
logged: entries.map(({ level, annotations }) => ({ level, annotations })),
leaked: logs.some((line) => line.includes(TOKENS["effect-ts"]!.write)),
},
{
forwarded: 2,
// The 429 is logged once; the 200 that followed is not.
logged: [
{
level: "WARN",
annotations: {
sessionId: "slack/T1/C1/1730000000.000100",
turnId: "turn-1",
method: "POST",
host: "github.com",
path: "/Effect-TS/effect.git/git-receive-pack",
credential: "write",
status: 429,
requestId: "ABCD:1234:5678",
retryAfter: "60",
rateLimitRemaining: "0",
rateLimitReset: "1730000060",
rateLimitResource: "git",
},
},
],
leaked: false,
},
)
}),
)
})
54 changes: 54 additions & 0 deletions apps/cluster/test/Chat/Egress/Lifecycle.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ import * as Deferred from "effect/Deferred"
import * as Effect from "effect/Effect"
import * as Fiber from "effect/Fiber"
import * as Layer from "effect/Layer"
import * as Logger from "effect/Logger"
import { BotAppError, ChatCredentials, type BotApp } from "../../../src/Chat/Egress/BotApp.ts"
import { makeEgressContext } from "../../../src/Chat/Egress/Client.ts"
import { handleChatEgress } from "../../../src/Chat/Egress/Egress.ts"
Expand Down Expand Up @@ -168,4 +169,57 @@ describe("chat egress credential lifecycle", () => {
assert.strictEqual(authorization, null)
}),
)

it.effect("a workspace that can't be asked hands out no credential, and the proxy says so", () =>
Effect.gen(function* () {
const logs: Array<string> = []
const capture = Logger.make((options) => {
logs.push(Logger.formatJson.log(options))
})
const context = makeEgressContext(
{ sessionId: acquireInput.sessionId, generation: 1, turnId: "turn-1" },
{ fetch: () => Promise.reject(new Error("the object is unreachable")) },
)
let authorization: string | null = null
yield* handleChatEgress(new Request(GITHUB), {
...context,
upstream: (request) =>
Effect.sync(() => {
authorization = request.headers.get("authorization")
return new Response("upstream")
}),
}).pipe(Effect.provide(Logger.layer([capture])))

const warnings = logs
.map(
(line) =>
JSON.parse(line) as {
level: string
message: unknown
annotations: Record<string, unknown>
},
)
.filter((entry) => entry.level === "WARN")
assert.deepStrictEqual(
{
authorization,
warnings: warnings.map((entry) => ({
annotations: entry.annotations,
saysWhy: JSON.stringify(entry.message).includes(
"the sandbox's workspace couldn't be reached",
),
})),
},
{
authorization: null,
warnings: [
{
annotations: { sessionId: acquireInput.sessionId, turnId: "turn-1", op: "tokens" },
saysWhy: true,
},
],
},
)
}),
)
})