fix(api): validate source connection ownership

.changeset/validate-source-connections.md

@@ -0,0 +1,6 @@
+---
+'@hyperdx/api': patch
+---
+
+Reject source writes that reference malformed, missing, or another team's
+connection.

packages/api/src/controllers/connection.ts

@@ -1,4 +1,39 @@
+import type { ObjectId } from '@/models';
 import Connection, { IConnection } from '@/models/connection';
+import { objectIdSchema } from '@/utils/zod';
+
+export type ConnectionValidation =
+  | { ok: true }
+  | { ok: false; status: 400 | 403; message: string };
+
+export async function validateConnectionId(
+  connection: unknown,
+  teamId: string | ObjectId | undefined,
+): Promise<ConnectionValidation> {
+  const parsed = objectIdSchema.safeParse(connection);
+  if (!parsed.success) {
+    return {
+      ok: false,
+      status: 400,
+      message: 'connection must be a valid connection id',
+    };
+  }
+  if (teamId == null) {
+    return { ok: false, status: 403, message: 'Forbidden' };
+  }
+  const connectionExists = await Connection.exists({
+    _id: parsed.data,
+    team: teamId,
+  });
+  if (connectionExists == null) {
+    return {
+      ok: false,
+      status: 400,
+      message: 'connection must be an existing connection id',
+    };
+  }
+  return { ok: true };
+}
 
 // Returns all connections across all teams. Only intended for instance-level
 // operations (e.g. startup auto-provisioning); user-facing routes must use

packages/api/src/mcp/tools/sources/saveSource.ts

@@ -2,14 +2,14 @@ import { SourceSchemaNoId } from '@hyperdx/common-utils/dist/types';
 import mongoose from 'mongoose';
 
 import * as config from '@/config';
+import { validateConnectionId } from '@/controllers/connection';
 import { createSource, getSource, updateSource } from '@/controllers/sources';
 import type { ToolRegistrar } from '@/mcp/tools/types';
 import {
   mcpServerError,
   mcpUserError,
   validateObjectId,
 } from '@/mcp/utils/errors';
-import { validateConnectionId } from '@/routers/external-api/v2/sources';
 import { isDuplicateKeyError } from '@/utils/errors';
 
 import { buildSourceInput, mcpSaveSourceSchema } from './schemas';

packages/api/src/routers/api/__tests__/sources.int.test.ts

@@ -7,8 +7,12 @@ import express from 'express';
 import { Types } from 'mongoose';
 import request from 'supertest';
 
-import { getLoggedInAgent, getServer } from '@/fixtures';
+import {
+  getLoggedInAgent as getFixtureLoggedInAgent,
+  getServer,
+} from '@/fixtures';
 import { appErrorHandler } from '@/middleware/error';
+import Connection from '@/models/connection';
 import { Source } from '@/models/source';
 import sourcesRouter from '@/routers/api/sources';
 
@@ -43,6 +47,27 @@ const MOCK_METRIC_SOURCE: Omit<Extract<TSource, { kind: 'metric' }>, 'id'> = {
   },
 };
 
+const createTestConnection = (team: Types.ObjectId, id: string) =>
+  Connection.create({
+    _id: id,
+    team,
+    name: 'Test Connection',
+    host: 'http://localhost:8123',
+    username: 'default',
+    password: 'password',
+  });
+
+const getLoggedInAgent = async (server: ReturnType<typeof getServer>) => {
+  const result = await getFixtureLoggedInAgent(server);
+
+  await Promise.all([
+    createTestConnection(result.team._id, MOCK_SOURCE.connection),
+    createTestConnection(result.team._id, MOCK_METRIC_SOURCE.connection),
+  ]);
+
+  return result;
+};
+
 describe('sources router', () => {
   const server = getServer();
 
@@ -103,6 +128,87 @@ describe('sources router', () => {
     expect(sources).toHaveLength(1);
   });
 
+  describe('connection validation', () => {
+    it('POST / - returns 400 for a malformed connection id', async () => {
+      const { agent } = await getLoggedInAgent(server);
+
+      await agent
+        .post('/sources')
+        .send({ ...MOCK_SOURCE, connection: 'not-an-object-id' })
+        .expect(400);
+    });
+
+    it('POST / - returns 400 for a nonexistent connection id', async () => {
+      const { agent } = await getLoggedInAgent(server);
+
+      await agent
+        .post('/sources')
+        .send({
+          ...MOCK_SOURCE,
+          connection: new Types.ObjectId().toString(),
+        })
+        .expect(400);
+    });
+
+    it('POST / - returns 400 for another team connection', async () => {
+      const { agent } = await getLoggedInAgent(server);
+      const otherConnection = await createTestConnection(
+        new Types.ObjectId(),
+        new Types.ObjectId().toString(),
+      );
+
+      await agent
+        .post('/sources')
+        .send({
+          ...MOCK_SOURCE,
+          connection: otherConnection._id.toString(),
+        })
+        .expect(400);
+    });
+
+    it('PUT /:id - rejects an inaccessible connection without changing the source', async () => {
+      const { agent, team } = await getLoggedInAgent(server);
+      const source = await Source.create({
+        ...MOCK_SOURCE,
+        team: team._id,
+      });
+      const otherConnection = await createTestConnection(
+        new Types.ObjectId(),
+        new Types.ObjectId().toString(),
+      );
+
+      await agent
+        .put(`/sources/${source._id}`)
+        .send({
+          ...MOCK_SOURCE,
+          id: source._id.toString(),
+          connection: otherConnection._id.toString(),
+        })
+        .expect(400);
+
+      const unchanged = await Source.findById(source._id);
+      expect(unchanged?.name).toBe(MOCK_SOURCE.name);
+      expect(unchanged?.connection.toString()).toBe(MOCK_SOURCE.connection);
+    });
+
+    it('PUT /:id - returns 400 for a nonexistent connection id', async () => {
+      const { agent, team } = await getLoggedInAgent(server);
+      const source = await Source.create({
+        ...MOCK_SOURCE,
+        team: team._id,
+      });
+
+      await agent
+        .put(`/sources/${source._id}`)
+        .send({
+          ...MOCK_SOURCE,
+          id: source._id.toString(),
+          connection: new Types.ObjectId().toString(),
+        })
+        .expect(400);
+    });
+  });
+
   it('POST / - returns 400 when request body is invalid', async () => {
     const { agent } = await getLoggedInAgent(server);
 
@@ -284,7 +390,7 @@ describe('sources router', () => {
     const metricSource = await Source.create({
       kind: SourceKind.Metric,
       name: 'Test Metric Source',
-      connection: new Types.ObjectId().toString(),
+      connection: MOCK_METRIC_SOURCE.connection,
       from: {
         databaseName: 'test_db',
         tableName: 'otel_metrics',
@@ -346,7 +452,7 @@ describe('sources router', () => {
     const metricSource = await Source.create({
       kind: SourceKind.Metric,
       name: 'Test Metric Source',
-      connection: new Types.ObjectId().toString(),
+      connection: MOCK_METRIC_SOURCE.connection,
       from: {
         databaseName: 'test_db',
         tableName: 'otel_metrics',
@@ -693,7 +799,7 @@ describe('sources router', () => {
     it('successfully updates a legacy Session source when timestampValueExpression is provided', async () => {
       const { agent, team } = await getLoggedInAgent(server);
 
-      const connectionId = new Types.ObjectId();
+      const connectionId = new Types.ObjectId(MOCK_SOURCE.connection);
       const result = await Source.collection.insertOne({
         kind: SourceKind.Session,
         name: 'Legacy Session',
@@ -841,6 +947,10 @@ describe('sources router', () => {
 
     it('POST / - creates a source when team id is a string', async () => {
       const app = getLocalAppModeApp();
+      await createTestConnection(
+        new Types.ObjectId('_local_team_'),
+        MOCK_SOURCE.connection,
+      );
 
       const response = await request(app)
         .post('/sources')
@@ -858,6 +968,10 @@ describe('sources router', () => {
 
     it('PUT /:id - updates a source when team id is a string', async () => {
       const app = getLocalAppModeApp();
+      await createTestConnection(
+        new Types.ObjectId('_local_team_'),
+        MOCK_SOURCE.connection,
+      );
 
       const source = await Source.create({
         ...MOCK_SOURCE,
@@ -991,7 +1105,7 @@ describe('sources router', () => {
       const traceSource: Omit<Extract<TSource, { kind: 'trace' }>, 'id'> = {
         kind: SourceKind.Trace,
         name: 'Trace with text index pref',
-        connection: new Types.ObjectId().toString(),
+        connection: MOCK_SOURCE.connection,
         from: { databaseName: 'test_db', tableName: 'otel_traces' },
         timestampValueExpression: 'Timestamp',
         defaultTableSelectExpression: '*',

packages/api/src/routers/api/sources.ts

@@ -6,6 +6,7 @@ import express from 'express';
 import { z } from 'zod';
 import { validateRequest } from 'zod-express-middleware';
 
+import { validateConnectionId } from '@/controllers/connection';
 import {
   createSource,
   deleteSource,
@@ -43,6 +44,16 @@ router.post(
     try {
       const { teamId } = getNonNullUserWithTeam(req);
 
+      const connectionCheck = await validateConnectionId(
+        req.body.connection,
+        teamId,
+      );
+      if (!connectionCheck.ok) {
+        return res
+          .status(connectionCheck.status)
+          .json({ message: connectionCheck.message });
+      }
+
       const source = await createSource(teamId.toString(), {
         ...req.body,
         team: teamId.toString(),
@@ -67,6 +78,16 @@ router.put(
     try {
       const { teamId } = getNonNullUserWithTeam(req);
 
+      const connectionCheck = await validateConnectionId(
+        req.body.connection,
+        teamId,
+      );
+      if (!connectionCheck.ok) {
+        return res
+          .status(connectionCheck.status)
+          .json({ message: connectionCheck.message });
+      }
+
       const source = await updateSource(teamId.toString(), req.params.id, {
         ...req.body,
         team: teamId.toString(),

packages/api/src/routers/external-api/__tests__/sources-crud.int.test.ts

@@ -3,6 +3,7 @@ import mongoose from 'mongoose';
 import request, { SuperAgentTest } from 'supertest';
 
 import * as config from '@/config';
+import { validateConnectionId } from '@/controllers/connection';
 import {
   DEFAULT_DATABASE,
   DEFAULT_LOGS_TABLE,
@@ -13,7 +14,6 @@ import Connection, { IConnection } from '@/models/connection';
 import { LogSource, Source, TraceSource } from '@/models/source';
 import { ITeam } from '@/models/team';
 import { IUser } from '@/models/user';
-import { validateConnectionId } from '@/routers/external-api/v2/sources';
 
 describe('External API v2 Sources CRUD', () => {
   const server = getServer();

packages/api/src/routers/external-api/v2/sources.ts

@@ -7,14 +7,14 @@ import {
 import express from 'express';
 import { z } from 'zod';
 
+import { validateConnectionId } from '@/controllers/connection';
 import {
   createSource,
   deleteSource,
   getSource,
   getSources,
   updateSource,
 } from '@/controllers/sources';
-import Connection from '@/models/connection';
 import { SourceDocument } from '@/models/source';
 import { processRequestWithEnhancedErrors as validateRequest } from '@/utils/enhancedErrors';
 import logger from '@/utils/logger';
@@ -91,46 +91,6 @@ function mapRequestGranularitiesToInternalFormat(
   next();
 }
 
-type ConnectionValidation =
-  | { ok: true }
-  | { ok: false; status: 400 | 403; message: string };
-
-// Validates that `connection` is a valid ObjectId referencing a connection
-// owned by the given team. SourceSchemaNoId only requires `connection` to be a
-// non-empty string, but the Mongoose model declares it as an ObjectId ref — a
-// non-ObjectId value would surface as a 500 CastError instead of a 400. The
-// team-scoped existence check also ensures a source can never reference another
-// team's ClickHouse credentials. Kept separate from the middleware so it's
-// readable and unit-testable on its own.
-export async function validateConnectionId(
-  connection: unknown,
-  teamId: Express.User['team'] | undefined,
-): Promise<ConnectionValidation> {
-  const parsed = objectIdSchema.safeParse(connection);
-  if (!parsed.success) {
-    return {
-      ok: false,
-      status: 400,
-      message: 'connection must be a valid connection id',
-    };
-  }
-  if (teamId == null) {
-    return { ok: false, status: 403, message: 'Forbidden' };
-  }
-  const connectionExists = await Connection.exists({
-    _id: parsed.data,
-    team: teamId,
-  });
-  if (connectionExists == null) {
-    return {
-      ok: false,
-      status: 400,
-      message: 'connection must be an existing connection id',
-    };
-  }
-  return { ok: true };
-}
-
 // Runs after body validation so req.body is the parsed shape.
 async function requireValidConnectionId(
   req: express.Request,