Add Atlas Stream Processing client.#2388
Conversation
github-actions
Bot
added
the
documentation
![semgrep-code-mongodb[bot]](https://avatars.githubusercontent.com/u/45120?s=60&v=4){kind=small}
| // merged ClientOptions. | ||
| func applyStreamProcessingDefaults(opts *options.ClientOptions) { | ||
| if opts.TLSConfig == nil { | ||
| opts.TLSConfig = &tls.Config{} |
There was a problem hiding this comment.
Semgrep identified an issue in your code:
opts.TLSConfig is created without MinVersion in applyStreamProcessingDefaults, allowing TLS 1.2 for StreamProcessing connections. This enables downgrade to TLS 1.2 (e.g., via a TLS‑inspection proxy) and exposes request/response data to interception.
More details about this
applyStreamProcessingDefaults sets opts.TLSConfig = &tls.Config{} without a MinVersion, so ConnectStreamProcessing may negotiate TLS 1.2. This widens the protocol surface for all StreamProcessing connections instead of requiring TLS 1.3.
Concrete exploit scenario:
- An on‑path adversary (e.g., a corporate TLS‑inspection proxy) poisons DNS for atlas-stream-*.a.query.mongodb.net and routes traffic through a proxy that only offers TLS 1.2 but presents a trusted enterprise root-signed certificate. Because opts.TLSConfig lacks MinVersion, the client accepts a TLS 1.2 handshake.
- The proxy then terminates TLS 1.2 and re-encrypts upstream, allowing it to read or modify StreamProcessing requests and responses, including credentials and stream processor configurations handled by ConnectStreamProcessing and subsequent API calls via spc.client.
Code reference: In applyStreamProcessingDefaults, when opts.TLSConfig is nil it is replaced with &tls.Config{}; no MinVersion is set anywhere, so every StreamProcessingClient created by ConnectStreamProcessing inherits this permissive TLS policy.
To resolve this comment:
✨ Commit fix suggestion
- Update the TLS configuration to set the minimum version to TLS 1.3 by changing
opts.TLSConfig = &tls.Config{}toopts.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS13}. - This enforces the use of TLS 1.3 for connections, preventing clients from negotiating older, less secure protocols.
💬 Ignore this finding
Reply with Semgrep commands to ignore this finding.
/fp <comment>for false positive/ar <comment>for acceptable risk/other <comment>for all other reasons
Alternatively, triage in Semgrep AppSec Platform to ignore the finding created by missing-ssl-minversion.
🛟 Help? Slack #semgrep-help or go/semgrep-help.
Resolution Options:
- Fix the code
- Reply
/fp $reason(if security gap doesn’t exist) - Reply
/ar $reason(if gap is valid but intentional; add mitigations/monitoring) - Reply
/other $reason(e.g., test-only)
You can view more details about this finding in the Semgrep AppSec Platform.
🧪 Performance ResultsCommit SHA: 8a803fcThe following benchmark tests for version 6a05f5a2218175000743ad12 had statistically significant changes (i.e., |z-score| > 1.96):
For a comprehensive view of all microbenchmark results for this PR's commit, please check out the Evergreen perf task for this patch. |
API Change Report./v2/mongocompatible changesConnectStreamProcessing: added ./v2/mongo/optionscompatible changesCreateStreamProcessor: added ./v2/streamprocessingcompatible changespackage added ./v2/x/mongo/driver/operationcompatible changesCreateStreamProcessor: added |
1 participant
Summary
Adds first-class driver support for Atlas Stream Processing (ASP). Users
can now connect to a workspace endpoint with a dedicated
streamprocessing.Clientand manage stream processors through typedmethods, instead of dispatching commands by hand via
RunCommand.Implementation follows the Atlas Stream Processing driver spec:
streamprocessingpackage (go.mongodb.org/mongo-driver/v2/streamprocessing)with thin type aliases over the new types in the
mongopackage.x/mongo/driver/operation/for the ASPwire commands:
createStreamProcessor,startStreamProcessor,stopStreamProcessor,dropStreamProcessor,getStreamProcessor,getStreamProcessorStats,startSampleStreamProcessor,getMoreSampleStreamProcessor.mongo/options/streamprocessingoptions.go.examples/stream_processing/.The deferred commands listed in the spec (
modifyStreamProcessor,listStreamProcessors,listStreamConnections,processStreamProcessor,listWorkspaceDefaults) are intentionally out of scope here and will landin follow-up work.
Public API change
This change adds new public surface but does not modify or remove
anything existing. The new exported identifiers:
streamprocessing.{Client,StreamProcessors,StreamProcessor,Info,SamplesResult,Connect,IsWorkspaceHost}(type aliases + factory).mongo.{StreamProcessingClient,StreamProcessors,StreamProcessor,StreamProcessorInfo,GetStreamProcessorSamplesResult,ConnectStreamProcessing,IsStreamProcessingHost}.mongo/options.{CreateStreamProcessor*,StartStreamProcessor*,StreamProcessorFailover*,GetStreamProcessorStats*,GetStreamProcessorSamples*}(option types + builders).x/mongo/driver/operation/(importable but not semver-covered).task api-reportoutput:Spec compliance notes
A few requirements from the spec that are worth flagging during review:
streamprocessing.Connectsets a default*tls.Configif the caller didn't, per: "Drivers MUST enable TLS bydefault when connecting to a workspace endpoint and MUST NOT allow TLS
to be disabled."
authSourcedefaults toadminwhen credentials are suppliedwithout one, per: "Drivers MUST set authSource=admin as the default
authentication database for workspace connections."
create/start/stop/dropand both sample-cursor commands), per the spec's retryability table
and the rationale that "stream processor state transitions … are not
idempotent." Only
getStreamProcessorandgetStreamProcessorStatsare configured as retryable reads.
startAfteris not exposed — the spec reserves it: "RESERVED forfuture use; not yet accepted by the server. Drivers MUST NOT send this
field until a future revision of this spec enables it."
GetStreamProcessorSampleswith no
CursorIDissuesstartSampleStreamProcessorfollowed by animmediate
getMoreSampleStreamProcessor, so the caller gets documentson the first call, per the spec rationale.
StreamProcessorInfodecodes into a typed struct that ignores extras; stats responses are
returned as raw
bson.Rawfor the same reason. Server-internal fields(
tenantID,projectId,processorId, …) are never surfaced.stateis surfaced as a string, not mapped to a Go enum, sofuture server states pass through untouched.
Test plan
task fmtclean.task build(includescompilecheck-119).task test-shortpasses for the new packages (mongo/options,x/mongo/driver/operation,internal/integration/streamprocessing).Pre-existing spec-test failures are unrelated (require
task init-submodule).task lint.task api-report; output above.MONGODB_STREAM_PROCESSING_URI=<workspace-uri> go test ./internal/integration/streamprocessing/...exercises create → start → stats → sample → stop → drop end-to-end.
Out of scope / follow-ups
modifyStreamProcessor,listStreamProcessors,listStreamConnections,processStreamProcessor,listWorkspaceDefaults— deferred by spec.in a separate revision.
master(v2) only.Background & Motivation