fix(worker-auth): unify token checks and bind job environment
This commit is contained in:
@@ -4,6 +4,7 @@ import type { Doc, Id } from './_generated/dataModel';
|
||||
import type { MutationCtx } from './_generated/server';
|
||||
import { internalMutation, mutation, query } from './_generated/server';
|
||||
import { getOwnedSpoon, getRequiredUserId, optionalText } from './model';
|
||||
import { assertWorkerToken } from './workerAuth';
|
||||
|
||||
const jobStatus = v.union(
|
||||
v.literal('queued'),
|
||||
@@ -152,14 +153,6 @@ const defaultAgentSettings = {
|
||||
allowUserFileEditing: true,
|
||||
};
|
||||
|
||||
const getWorkerToken = () => process.env.SPOON_WORKER_TOKEN?.trim();
|
||||
|
||||
const requireWorkerToken = (workerToken: string) => {
|
||||
const expected = getWorkerToken();
|
||||
if (!expected) throw new ConvexError('SPOON_WORKER_TOKEN is not configured.');
|
||||
if (workerToken !== expected) throw new ConvexError('Invalid worker token.');
|
||||
};
|
||||
|
||||
const mergeMessageMetadata = (
|
||||
metadata: string | undefined,
|
||||
patch: Record<string, unknown>,
|
||||
@@ -1123,7 +1116,7 @@ export const updateStatus = mutation({
|
||||
summary: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1193,7 +1186,7 @@ export const markWorkspaceActive = mutation({
|
||||
workspaceExpiresAt: v.optional(v.number()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1223,7 +1216,7 @@ export const setRuntimeSession = mutation({
|
||||
containerId: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1247,7 +1240,7 @@ export const setCodexSessionId = mutation({
|
||||
codexSessionId: v.string(),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1275,7 +1268,7 @@ export const createInteractionRequest = mutation({
|
||||
metadata: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1327,7 +1320,7 @@ export const patchInteractionRequest = mutation({
|
||||
metadata: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const interaction = await ctx.db.get(args.interactionId);
|
||||
if (!interaction) throw new ConvexError('Interaction request not found.');
|
||||
const job = await ctx.db.get(interaction.jobId);
|
||||
@@ -1352,7 +1345,7 @@ export const markWorkspaceStopped = mutation({
|
||||
workspaceStatus: v.optional(workspaceStatus),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1373,7 +1366,7 @@ export const heartbeatWorkspace = mutation({
|
||||
jobId: v.id('agentJobs'),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1398,7 +1391,7 @@ export const completeWithDraftPr = mutation({
|
||||
summary: v.string(),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1445,7 +1438,7 @@ export const applyMaintenanceDecision = mutation({
|
||||
requiresUserApproval: v.boolean(),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1526,7 +1519,7 @@ export const appendEvent = mutation({
|
||||
metadata: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1555,7 +1548,7 @@ export const appendMessage = mutation({
|
||||
metadata: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1601,7 +1594,7 @@ export const updateMessage = mutation({
|
||||
metadata: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const message = await ctx.db.get(args.messageId);
|
||||
if (!message) throw new ConvexError('Agent message not found.');
|
||||
const job = await ctx.db.get(message.jobId);
|
||||
@@ -1656,7 +1649,7 @@ export const recordWorkspaceChange = mutation({
|
||||
diff: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
@@ -1685,7 +1678,7 @@ export const addArtifact = mutation({
|
||||
contentType: artifactContentType,
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const job = await ctx.db.get(args.jobId);
|
||||
if (job?.claimedBy !== args.workerId) {
|
||||
throw new ConvexError('Agent job not claimed by this worker.');
|
||||
|
||||
@@ -6,6 +6,7 @@ import type { Doc } from './_generated/dataModel';
|
||||
import { internal } from './_generated/api';
|
||||
import { action } from './_generated/server';
|
||||
import { decryptSecret } from './secretCrypto';
|
||||
import { assertWorkerToken } from './workerAuth';
|
||||
|
||||
type ClaimedJob = {
|
||||
job: Doc<'agentJobs'>;
|
||||
@@ -41,19 +42,13 @@ type WorkerClaim = {
|
||||
secrets: { name: string; value: string }[];
|
||||
};
|
||||
|
||||
const requireWorkerToken = (workerToken: string) => {
|
||||
const expected = process.env.SPOON_WORKER_TOKEN?.trim();
|
||||
if (!expected) throw new ConvexError('SPOON_WORKER_TOKEN is not configured.');
|
||||
if (workerToken !== expected) throw new ConvexError('Invalid worker token.');
|
||||
};
|
||||
|
||||
export const claimNextForWorker = action({
|
||||
args: {
|
||||
workerId: v.string(),
|
||||
workerToken: v.string(),
|
||||
},
|
||||
handler: async (ctx, args): Promise<WorkerClaim | null> => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const claimed = (await ctx.runMutation(
|
||||
internal.agentJobs.claimNextInternal,
|
||||
{
|
||||
|
||||
@@ -9,6 +9,7 @@ import { internal } from './_generated/api';
|
||||
import { action } from './_generated/server';
|
||||
import { normalizeDotfilePath } from './model';
|
||||
import { decryptSecret, encryptSecret } from './secretCrypto';
|
||||
import { assertWorkerToken } from './workerAuth';
|
||||
|
||||
const MAX_FILE_BYTES = 512 * 1024; // 512 KB per dotfile
|
||||
|
||||
@@ -18,12 +19,6 @@ const getRequiredUserId = async (ctx: ActionCtx): Promise<Id<'users'>> => {
|
||||
return userId;
|
||||
};
|
||||
|
||||
const requireWorkerToken = (workerToken: string) => {
|
||||
const expected = process.env.SPOON_WORKER_TOKEN;
|
||||
if (!expected) throw new ConvexError('Worker token is not configured.');
|
||||
if (workerToken !== expected) throw new ConvexError('Invalid worker token.');
|
||||
};
|
||||
|
||||
const putOne = async (
|
||||
ctx: ActionCtx,
|
||||
ownerId: Id<'users'>,
|
||||
@@ -114,7 +109,7 @@ type WorkerEnvironment = {
|
||||
export const getEnvironmentForJob = action({
|
||||
args: { workerToken: v.string(), jobId: v.id('agentJobs') },
|
||||
handler: async (ctx, args): Promise<WorkerEnvironment | null> => {
|
||||
requireWorkerToken(args.workerToken);
|
||||
assertWorkerToken(args.workerToken);
|
||||
const raw = await ctx.runQuery(
|
||||
internal.userEnvironment.getRawEnvironmentForJobInternal,
|
||||
{ jobId: args.jobId },
|
||||
|
||||
@@ -71,6 +71,7 @@ export const getRawEnvironmentForJobInternal = internalQuery({
|
||||
handler: async (ctx, { jobId }) => {
|
||||
const job = await ctx.db.get(jobId);
|
||||
if (!job) return null;
|
||||
if (!job.claimedBy) return null; // only a claimed job's env is exposed
|
||||
const ownerId = job.ownerId;
|
||||
const [user, settings, files] = await Promise.all([
|
||||
ctx.db.get(ownerId),
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
import { ConvexError } from 'convex/values';
|
||||
|
||||
export const assertWorkerToken = (provided: string): void => {
|
||||
const expected = process.env.SPOON_WORKER_TOKEN?.trim();
|
||||
if (!expected) {
|
||||
throw new ConvexError('SPOON_WORKER_TOKEN is not configured.');
|
||||
}
|
||||
if (provided.trim() !== expected) {
|
||||
throw new ConvexError('Invalid worker token.');
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1,106 @@
|
||||
import { convexTest } from 'convex-test';
|
||||
import { ConvexError } from 'convex/values';
|
||||
import { afterEach, describe, expect, test } from 'vitest';
|
||||
|
||||
import { internal } from '../../convex/_generated/api.js';
|
||||
import schema from '../../convex/schema';
|
||||
import { assertWorkerToken } from '../../convex/workerAuth';
|
||||
|
||||
const modules = import.meta.glob('../../convex/**/*.*s');
|
||||
|
||||
afterEach(() => {
|
||||
delete process.env.SPOON_WORKER_TOKEN;
|
||||
});
|
||||
|
||||
describe('assertWorkerToken', () => {
|
||||
test('accepts a token that matches after trimming', () => {
|
||||
process.env.SPOON_WORKER_TOKEN = 'secret\n';
|
||||
expect(() => assertWorkerToken('secret')).not.toThrow();
|
||||
expect(() => assertWorkerToken(' secret ')).not.toThrow();
|
||||
});
|
||||
|
||||
test('rejects a mismatched token', () => {
|
||||
process.env.SPOON_WORKER_TOKEN = 'secret';
|
||||
expect(() => assertWorkerToken('nope')).toThrow(ConvexError);
|
||||
});
|
||||
|
||||
test('throws when not configured', () => {
|
||||
expect(() => assertWorkerToken('x')).toThrow(ConvexError);
|
||||
});
|
||||
});
|
||||
|
||||
describe('getRawEnvironmentForJobInternal', () => {
|
||||
test('only exposes environment for a claimed job', async () => {
|
||||
const t = convexTest(schema, modules);
|
||||
const jobId = await t.mutation(async (ctx) => {
|
||||
const now = Date.now();
|
||||
const ownerId = await ctx.db.insert('users', {
|
||||
email: 'worker-auth@example.com',
|
||||
name: 'Worker Auth',
|
||||
});
|
||||
const spoonId = await ctx.db.insert('spoons', {
|
||||
ownerId,
|
||||
name: 'Worker Auth Spoon',
|
||||
provider: 'gitea',
|
||||
upstreamOwner: 'upstream',
|
||||
upstreamRepo: 'worker-auth',
|
||||
upstreamDefaultBranch: 'main',
|
||||
upstreamUrl: 'https://git.example.com/upstream/worker-auth',
|
||||
forkOwner: 'team',
|
||||
forkRepo: 'worker-auth-spoon',
|
||||
forkUrl: 'https://git.example.com/team/worker-auth-spoon',
|
||||
visibility: 'private',
|
||||
maintenanceMode: 'watch',
|
||||
syncCadence: 'daily',
|
||||
productionRefStrategy: 'default_branch',
|
||||
status: 'active',
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
});
|
||||
const requestId = await ctx.db.insert('agentRequests', {
|
||||
spoonId,
|
||||
ownerId,
|
||||
prompt: 'Test worker authorization',
|
||||
status: 'queued',
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
});
|
||||
return await ctx.db.insert('agentJobs', {
|
||||
spoonId,
|
||||
ownerId,
|
||||
agentRequestId: requestId,
|
||||
status: 'queued',
|
||||
prompt: 'Test worker authorization',
|
||||
runtime: 'opencode',
|
||||
baseBranch: 'main',
|
||||
workBranch: 'spoon/worker-auth',
|
||||
forkOwner: 'team',
|
||||
forkRepo: 'worker-auth-spoon',
|
||||
forkUrl: 'https://git.example.com/team/worker-auth-spoon',
|
||||
upstreamOwner: 'upstream',
|
||||
upstreamRepo: 'worker-auth',
|
||||
selectedSecretIds: [],
|
||||
model: '',
|
||||
reasoningEffort: 'medium',
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
});
|
||||
});
|
||||
|
||||
expect(
|
||||
await t.query(internal.userEnvironment.getRawEnvironmentForJobInternal, {
|
||||
jobId,
|
||||
}),
|
||||
).toBeNull();
|
||||
|
||||
await t.mutation(async (ctx) => {
|
||||
await ctx.db.patch(jobId, { claimedBy: 'w1' });
|
||||
});
|
||||
|
||||
expect(
|
||||
await t.query(internal.userEnvironment.getRawEnvironmentForJobInternal, {
|
||||
jobId,
|
||||
}),
|
||||
).not.toBeNull();
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user