Skip to content

EAI-2440: Check stream processor databases against databaseNames #181

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mongodben merged 4 commits into from
Oct 23, 2026
Merged

EAI-2440: Check stream processor databases against databaseNames #181

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1530e0b
EAI-2440: Check stream processor databases against databaseNames
cursoragent Oct 22, 2026
ec68b06
EAI-2440: Read only stream stage keys, not literal payloads or source…
cursoragent Oct 23, 2026
e601571
EAI-2440: Split database allowlist notes into separate points
cursoragent Oct 23, 2026
a272db3
EAI-2440: Reject database names MongoDB would reject
cursoragent Oct 23, 2026
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
17 changes: 11 additions & 6 deletions agents/atlas-explorer/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,8 @@ dataAccess: {
- `$out` and `$merge` target databases in `aggregate`, `aggregate_db`, and
`explain` pipelines.
- `namespaces` in `atlas_get_performance_advisor`.
- Processor databases (`$source`, `$lookup`, `$merge`, `$out`, and `dlq`)
in `atlas_streams_build` and `atlas_streams_manage`.
- It does not scan filters or documents.
- A rejected call returns an error to the agent, so it can correct itself.

Expand All @@ -312,12 +314,15 @@ roles.
- **`aggregate_db` reports on every database.** Cluster-level stages such as
`$currentOp` and `$queryStats` cover activity on every database, not only
the one named in the `database` argument.
- **Existing Atlas Stream Processing connections** (writes enabled only). A
connection is a saved link from a stream workspace to a data source, such
as a cluster. Creating one must name an allowed cluster. Updating or using
an existing one names only the connection, so if it points at a cluster
outside `dataAccess.projects`, the agent can still change it or start a processor
that reads or writes through it.
- **Existing Atlas Stream Processing connections and processors** (writes
enabled only). A connection is a saved link from a stream workspace to a
data source, such as a cluster. Creating one must name an allowed cluster.
Updating or using an existing one names only the connection, so if it
points at a cluster outside `dataAccess.projects`, the agent can still
change it or start a processor that reads or writes through it. Starting
an existing processor (`start-processor`) or creating a cluster-wide
`$source` without `db` likewise does not name a database in the call
arguments.

## MCP credential lifecycle

Expand Down
19 changes: 13 additions & 6 deletions agents/atlas-explorer/src/agent_atlas_explorer/config.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,12 +55,19 @@ import type {
* is the MCP access token, which can't be limited to specific databases
* today. What `dataAccess.databaseNames` covers:
*
* - Checked: each tool call's `database` argument, the target database of
* `$out` and `$merge` in the pipeline of `aggregate`, `aggregate_db`, and
* `explain`, and `namespaces` in `atlas_get_performance_advisor`.
* - Not checked: other parts of an aggregation pipeline.
* - Not checked: cluster-level `aggregate_db` stages such as `$currentOp`
* and `$queryStats`, which report activity on every database.
* - Checked:
* - Each tool call's `database` argument.
* - The target database of `$out` and `$merge` in `aggregate`,
* `aggregate_db`, and `explain` pipelines.
* - `namespaces` in `atlas_get_performance_advisor`.
* - Processor databases (`$source`, `$lookup`, `$merge`, `$out`, and
* `dlq`) in `atlas_streams_build` and `atlas_streams_manage`.
* - Not checked:
* - Other parts of an aggregation pipeline.
* - Starting an existing stream processor (`start-processor`).
* - Cluster-wide `$source` stages that omit `db`.
* - Cluster-level `aggregate_db` stages such as `$currentOp` and
* `$queryStats`, which report activity on every database.
* - Not filtered: tool results, so the agent can still see names of
* databases it may not use.
*
Expand Down
326 changes: 326 additions & 0 deletions packages/agent-engine-core/src/databaseAllowlist.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,332 @@ describe("guardDatabaseCall", () => {
});
});
});

describe("Stream Processing databases", () => {
const buildProcessor = (
pipeline: unknown,
dlq?: unknown,
) =>
call("mongodb__atlas_streams_build", {
projectId: "p",
workspaceName: "ws",
resource: "processor",
processorName: "proc",
pipeline,
...(dlq === undefined ? {} : { dlq }),
});

it.each([
[
"a workspace creation call",
call("mongodb__atlas_streams_build", {
projectId: "p",
resource: "workspace",
workspaceName: "ws",
}),
],
[
"a start-processor call",
call("mongodb__atlas_streams_manage", {
projectId: "p",
workspaceName: "ws",
action: "start-processor",
resourceName: "proc",
}),
],
[
"a Kafka source and emit pipeline",
buildProcessor([
{ $source: { connectionName: "kafka", topic: "events" } },
{ $emit: { connectionName: "kafka", topic: "out" } },
]),
],
[
"a cluster-wide change stream source without db",
buildProcessor([
{ $source: { connectionName: "cluster" } },
{ $emit: { connectionName: "kafka", topic: "out" } },
]),
],
[
"allowed $source, $lookup, $merge, and dlq databases",
buildProcessor(
[
{
$source: {
connectionName: "cluster",
db: "orders",
coll: "events",
},
},
{
$lookup: {
connectionName: "cluster",
from: { db: "orders", coll: "users" },
localField: "userId",
foreignField: "_id",
as: "user",
},
},
{
$merge: {
into: {
connectionName: "cluster",
db: "tenant_3",
coll: "archive",
},
},
},
],
{ connectionName: "cluster", db: "orders", coll: "dlq" },
),
],
[
"stringified pipeline and dlq",
buildProcessor(
JSON.stringify([
{
$merge: {
into: { connectionName: "cluster", db: "orders", coll: "out" },
},
},
]),
JSON.stringify({
connectionName: "cluster",
db: "orders",
coll: "dlq",
}),
),
],
[
"a stage key that names a database stage inside literal data",
buildProcessor([
{ $source: { connectionName: "kafka", topic: "events" } },
{ $set: { spec: { $literal: { $lookup: "template" } } } },
{ $emit: { connectionName: "kafka", topic: "out" } },
]),
],
[
"inline $source.documents holding stage-shaped data",
buildProcessor([
{
$source: {
documents: [
{ $merge: { into: { db: "admin", coll: "x" } } },
{ $lookup: "template" },
],
},
},
{ $emit: { connectionName: "kafka", topic: "out" } },
]),
],
[
"a $match filter on a field named pipeline",
buildProcessor([
{ $source: { connectionName: "kafka", topic: "events" } },
{ $match: { pipeline: [{ $merge: { into: { db: "admin" } } }] } },
{ $emit: { connectionName: "kafka", topic: "out" } },
]),
],
[
"a change stream source with an allowed pushdown pipeline",
buildProcessor([
{
$source: {
connectionName: "cluster",
db: "orders",
coll: "events",
pipeline: [{ $match: { operationType: "insert" } }],
},
},
{ $emit: { connectionName: "kafka", topic: "out" } },
]),
],
])("runs %s", async (_label, request) => {
expect((await run(request)).ran).toBe(true);
});

it.each([
[
"$source.db",
buildProcessor([
{ $source: { connectionName: "c", db: "admin", coll: "events" } },
]),
],
[
"$lookup.from.db",
buildProcessor([
{
$lookup: {
connectionName: "c",
from: { db: "admin", coll: "users" },
},
},
]),
],
[
"$merge.into.db",
buildProcessor([
{
$merge: {
into: { connectionName: "c", db: "admin", coll: "out" },
},
},
]),
],
[
"$out.db",
buildProcessor([{ $out: { db: "admin", coll: "out" } }]),
],
[
"dlq.db",
buildProcessor(
[{ $source: { connectionName: "k", topic: "t" } }],
{ connectionName: "c", db: "admin", coll: "dlq" },
),
],
[
"modify-processor with a disallowed $merge.into.db",
call("mongodb__atlas_streams_manage", {
projectId: "p",
workspaceName: "ws",
action: "modify-processor",
resourceName: "proc",
pipeline: [
{
$merge: {
into: { connectionName: "c", db: "admin", coll: "out" },
},
},
],
}),
],
[
"a disallowed database inside a window sub-pipeline",
buildProcessor([
{ $source: { connectionName: "kafka", topic: "events" } },
{
$tumblingWindow: {
interval: { size: 5, unit: "minute" },
pipeline: [
{
$merge: {
into: { connectionName: "c", db: "admin", coll: "out" },
},
},
],
},
},
]),
],
[
"a pushdown pipeline on a disallowed collection source",
buildProcessor([
{
$source: {
connectionName: "c",
db: "admin",
coll: "events",
pipeline: [{ $match: { operationType: "insert" } }],
},
},
]),
],
])("rejects a disallowed %s", async (_label, request) => {
const { result, ran } = await run(request);
expect(ran).toBe(false);
expect(result).toMatchObject({
content: expect.stringContaining(
"Database admin is not in the databaseNames allowlist.",
),
});
});

it("lists every disallowed database across pipeline stages and dlq in one error", async () => {
const { result, ran } = await run(
buildProcessor(
[
{ $source: { connectionName: "c", db: "secret", coll: "in" } },
{
$lookup: {
connectionName: "c",
from: { db: "admin", coll: "users" },
},
},
{
$merge: {
into: { connectionName: "c", db: "audit", coll: "out" },
},
},
],
{ connectionName: "c", db: "dlq_db", coll: "errors" },
),
);
expect(ran).toBe(false);
expect(result).toMatchObject({
content: expect.stringContaining(
"Databases dlq_db, secret, admin, audit are not in the databaseNames allowlist.",
),
});
});

it("rejects a stream stage database with surrounding whitespace", async () => {
const { ran } = await run(
buildProcessor([
{
$merge: {
into: { connectionName: "c", db: "orders ", coll: "out" },
},
},
]),
);
expect(ran).toBe(false);
});

it.each([
[
"a dlq without db",
buildProcessor(
[{ $source: { connectionName: "k", topic: "t" } }],
{ connectionName: "c", coll: "dlq" },
),
"dlq must include a database name",
],
[
"a non-array pipeline",
buildProcessor("{ $merge: {} }"),
"pipeline must be an array of stages",
],
[
"a collection $source without a string db",
buildProcessor([{ $source: { connectionName: "c", coll: "events" } }]),
"$source stage must include a string db",
],
[
"a $lookup without from.db",
buildProcessor([
{ $lookup: { connectionName: "c", from: "users" } },
]),
"$lookup stage must include a string from.db",
],
[
"a $merge without into.db",
buildProcessor([{ $merge: { into: { connectionName: "c", coll: "x" } } }]),
"$merge stage must include a string into.db",
],
[
"a $out without db",
buildProcessor([{ $out: "copy" }]),
"$out stage must be an object, so databaseNames",
],
])("rejects %s", async (_label, request, expectedMessage) => {
const { result, ran } = await run(request);
expect(ran).toBe(false);
expect(result).toMatchObject({
content: expect.stringContaining(expectedMessage),
});
});
});
});

describe("after an approval edit", () => {
Expand Down
Loading
Loading