fix(api): validate source connection ownership (#2801)

## Summary

The internal source API accepts any non-empty `connection` value when creating or updating a source. Unlike the external API v2 and MCP paths, it does not verify that the connection exists or belongs to the authenticated team.

That can leave sources pointing at missing connections or another team's ClickHouse credentials.

## What changed

- Moved the existing connection validation into the connection controller.
- Applied it to internal `POST /sources` and `PUT /sources/:id`.
- Kept the external API v2 and MCP paths on the same shared validator.
- Preserved the existing 400 responses for malformed, missing, and inaccessible connection IDs.
- Kept local app mode compatible with its string team ID.

## Tests

- Added internal API coverage for malformed connection IDs.
- Added coverage for missing and foreign-team connections.
- Verified rejected updates leave the existing source unchanged.
- Preserved coverage for successful source creation and updates.
- Targeted ESLint passes.

Author: Minh Vu

.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,