fix(mongodb): validate options and bound findOne queries
This commit is contained in:
1 parent
d098e3e970
commit
55f15c8e4b
14 files changed
+934
-5
No files matched your search
@@ -592,6 +592,61 @@ pub async fn find_documents(
|
||||
Ok(MongoDocumentResult { documents, total })
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct MongoFindOneOptions {
|
||||
sort: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
pub async fn find_one(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter: Option<&str>,
|
||||
projection: Option<&str>,
|
||||
options: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
let filter_doc = parse_optional_filter_document(filter)?.unwrap_or_default();
|
||||
let projection_doc = parse_optional_json_document(projection, "projection")?;
|
||||
let options = parse_find_one_options(options)?;
|
||||
let sort_doc = parse_optional_document(options.sort.as_ref(), "sort")?;
|
||||
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let mut action = col.find_one(filter_doc);
|
||||
if let Some(projection) = projection_doc {
|
||||
action = action.projection(projection);
|
||||
}
|
||||
if let Some(sort) = sort_doc {
|
||||
action = action.sort(sort);
|
||||
}
|
||||
|
||||
let result = action.await.map_err(|e| e.to_string())?;
|
||||
Ok(single_document_result(result))
|
||||
}
|
||||
|
||||
fn parse_optional_filter_document(value: Option<&str>) -> Result<Option<Document>, String> {
|
||||
let Some(raw) = value.map(str::trim).filter(|value| !value.is_empty()) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let json: serde_json::Value = serde_json::from_str(raw).map_err(|e| format!("Invalid filter JSON: {e}"))?;
|
||||
json_filter_to_document(&json).map(Some).map_err(|e| format!("Invalid filter: {e}"))
|
||||
}
|
||||
|
||||
fn parse_find_one_options(options: Option<&str>) -> Result<MongoFindOneOptions, String> {
|
||||
match options.map(str::trim).filter(|value| !value.is_empty()) {
|
||||
Some(raw) => serde_json::from_str(raw).map_err(|e| format!("Invalid findOne options: {e}")),
|
||||
None => Ok(MongoFindOneOptions::default()),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_optional_json_document(value: Option<&str>, label: &str) -> Result<Option<Document>, String> {
|
||||
let Some(raw) = value.map(str::trim).filter(|value| !value.is_empty()) else {
|
||||
return Ok(None);
|
||||
};
|
||||
let json: serde_json::Value = serde_json::from_str(raw).map_err(|e| format!("Invalid {label} JSON: {e}"))?;
|
||||
json_object_to_document(&json).map(Some).map_err(|e| format!("Invalid {label}: {e}"))
|
||||
}
|
||||
|
||||
pub async fn count_documents(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
@@ -979,6 +1034,180 @@ fn parse_update_array_filters(options_json: Option<&str>) -> Result<Option<Vec<D
|
||||
.transpose()
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct MongoFindOneAndUpdateOptions {
|
||||
return_document: Option<String>,
|
||||
return_new_document: Option<bool>,
|
||||
new: Option<bool>,
|
||||
upsert: Option<bool>,
|
||||
projection: Option<serde_json::Value>,
|
||||
sort: Option<serde_json::Value>,
|
||||
array_filters: Option<Vec<serde_json::Value>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct MongoFindOneAndReplaceOptions {
|
||||
return_document: Option<String>,
|
||||
return_new_document: Option<bool>,
|
||||
new: Option<bool>,
|
||||
upsert: Option<bool>,
|
||||
projection: Option<serde_json::Value>,
|
||||
sort: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
#[serde(rename_all = "camelCase", deny_unknown_fields)]
|
||||
struct MongoFindOneAndDeleteOptions {
|
||||
projection: Option<serde_json::Value>,
|
||||
sort: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
fn parse_find_and_modify_options<T>(options_json: Option<&str>, command: &str) -> Result<T, String>
|
||||
where
|
||||
T: Default + for<'de> Deserialize<'de>,
|
||||
{
|
||||
match options_json.map(str::trim).filter(|value| !value.is_empty()) {
|
||||
Some(raw) => serde_json::from_str(raw).map_err(|e| format!("Invalid {command} options: {e}")),
|
||||
None => Ok(T::default()),
|
||||
}
|
||||
}
|
||||
|
||||
fn find_and_modify_returns_after(
|
||||
return_document: Option<&str>,
|
||||
return_new_document: Option<bool>,
|
||||
new: Option<bool>,
|
||||
) -> Result<bool, String> {
|
||||
if let Some(return_document) = return_document {
|
||||
if return_document.eq_ignore_ascii_case("after") {
|
||||
return Ok(true);
|
||||
}
|
||||
if return_document.eq_ignore_ascii_case("before") {
|
||||
return Ok(false);
|
||||
}
|
||||
return Err("returnDocument must be either 'before' or 'after'".to_string());
|
||||
}
|
||||
Ok(return_new_document.or(new).unwrap_or(false))
|
||||
}
|
||||
|
||||
fn parse_optional_document(field: Option<&serde_json::Value>, label: &str) -> Result<Option<Document>, String> {
|
||||
field.map(|value| json_object_to_document(value).map_err(|e| format!("Invalid {label}: {e}"))).transpose()
|
||||
}
|
||||
|
||||
fn find_and_modify_array_filters(
|
||||
array_filters: Option<&Vec<serde_json::Value>>,
|
||||
) -> Result<Option<Vec<Document>>, String> {
|
||||
array_filters
|
||||
.map(|filters| {
|
||||
filters
|
||||
.iter()
|
||||
.map(json_filter_to_document)
|
||||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| format!("Invalid arrayFilters: {e}"))
|
||||
})
|
||||
.transpose()
|
||||
}
|
||||
|
||||
fn single_document_result(document: Option<Document>) -> MongoDocumentResult {
|
||||
match document {
|
||||
Some(document) => MongoDocumentResult { documents: vec![bson_to_json(&Bson::Document(document))], total: 1 },
|
||||
None => MongoDocumentResult { documents: Vec::new(), total: 0 },
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn find_one_and_update(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter_json: &str,
|
||||
update_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
let filter_value: serde_json::Value =
|
||||
serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?;
|
||||
let update_value: serde_json::Value =
|
||||
serde_json::from_str(update_json).map_err(|e| format!("Invalid update JSON: {e}"))?;
|
||||
let filter = json_filter_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?;
|
||||
let update = json_object_to_document(&update_value).map_err(|e| format!("Invalid update: {e}"))?;
|
||||
let options: MongoFindOneAndUpdateOptions = parse_find_and_modify_options(options_json, "findOneAndUpdate")?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let mut action = col.find_one_and_update(filter, update);
|
||||
if find_and_modify_returns_after(options.return_document.as_deref(), options.return_new_document, options.new)? {
|
||||
action = action.return_document(mongodb::options::ReturnDocument::After);
|
||||
}
|
||||
if let Some(upsert) = options.upsert {
|
||||
action = action.upsert(upsert);
|
||||
}
|
||||
if let Some(projection) = parse_optional_document(options.projection.as_ref(), "projection")? {
|
||||
action = action.projection(projection);
|
||||
}
|
||||
if let Some(sort) = parse_optional_document(options.sort.as_ref(), "sort")? {
|
||||
action = action.sort(sort);
|
||||
}
|
||||
if let Some(array_filters) = find_and_modify_array_filters(options.array_filters.as_ref())? {
|
||||
action = action.array_filters(array_filters);
|
||||
}
|
||||
let result = action.await.map_err(|e| e.to_string())?;
|
||||
Ok(single_document_result(result))
|
||||
}
|
||||
|
||||
pub async fn find_one_and_replace(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter_json: &str,
|
||||
replacement_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
let filter_value: serde_json::Value =
|
||||
serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?;
|
||||
let replacement_value: serde_json::Value =
|
||||
serde_json::from_str(replacement_json).map_err(|e| format!("Invalid replacement JSON: {e}"))?;
|
||||
let filter = json_filter_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?;
|
||||
let replacement = json_object_to_document(&replacement_value).map_err(|e| format!("Invalid replacement: {e}"))?;
|
||||
let options: MongoFindOneAndReplaceOptions = parse_find_and_modify_options(options_json, "findOneAndReplace")?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let mut action = col.find_one_and_replace(filter, replacement);
|
||||
if find_and_modify_returns_after(options.return_document.as_deref(), options.return_new_document, options.new)? {
|
||||
action = action.return_document(mongodb::options::ReturnDocument::After);
|
||||
}
|
||||
if let Some(upsert) = options.upsert {
|
||||
action = action.upsert(upsert);
|
||||
}
|
||||
if let Some(projection) = parse_optional_document(options.projection.as_ref(), "projection")? {
|
||||
action = action.projection(projection);
|
||||
}
|
||||
if let Some(sort) = parse_optional_document(options.sort.as_ref(), "sort")? {
|
||||
action = action.sort(sort);
|
||||
}
|
||||
let result = action.await.map_err(|e| e.to_string())?;
|
||||
Ok(single_document_result(result))
|
||||
}
|
||||
|
||||
pub async fn find_one_and_delete(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
let filter_value: serde_json::Value =
|
||||
serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?;
|
||||
let filter = json_filter_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?;
|
||||
let options: MongoFindOneAndDeleteOptions = parse_find_and_modify_options(options_json, "findOneAndDelete")?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let mut action = col.find_one_and_delete(filter);
|
||||
if let Some(projection) = parse_optional_document(options.projection.as_ref(), "projection")? {
|
||||
action = action.projection(projection);
|
||||
}
|
||||
if let Some(sort) = parse_optional_document(options.sort.as_ref(), "sort")? {
|
||||
action = action.sort(sort);
|
||||
}
|
||||
let result = action.await.map_err(|e| e.to_string())?;
|
||||
Ok(single_document_result(result))
|
||||
}
|
||||
|
||||
pub async fn delete_document(client: &Client, database: &str, collection: &str, id: &str) -> Result<u64, String> {
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
for filter in document_id_filters(id) {
|
||||
@@ -1910,6 +2139,57 @@ mod tests {
|
||||
assert!(!is_update_operator_document(&Document::new()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn find_one_options_accept_sort_and_reject_unimplemented_fields() {
|
||||
let options = parse_find_one_options(Some(r#"{"sort":{"createdAt":-1}}"#)).unwrap();
|
||||
assert_eq!(options.sort, Some(serde_json::json!({ "createdAt": -1 })));
|
||||
|
||||
let error = parse_find_one_options(Some(r#"{"hint":{"createdAt":1}}"#)).unwrap_err();
|
||||
assert!(error.contains("unknown field `hint`"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn find_and_modify_options_reject_fields_not_applied_by_each_command() {
|
||||
let update_error = parse_find_and_modify_options::<MongoFindOneAndUpdateOptions>(
|
||||
Some(r#"{"hint":{"name":1}}"#),
|
||||
"findOneAndUpdate",
|
||||
)
|
||||
.unwrap_err();
|
||||
assert!(update_error.contains("unknown field `hint`"));
|
||||
|
||||
let replace_error = parse_find_and_modify_options::<MongoFindOneAndReplaceOptions>(
|
||||
Some(r#"{"arrayFilters":[{"item.active":true}]}"#),
|
||||
"findOneAndReplace",
|
||||
)
|
||||
.unwrap_err();
|
||||
assert!(replace_error.contains("unknown field `arrayFilters`"));
|
||||
|
||||
let delete_error = parse_find_and_modify_options::<MongoFindOneAndDeleteOptions>(
|
||||
Some(r#"{"returnDocument":"after"}"#),
|
||||
"findOneAndDelete",
|
||||
)
|
||||
.unwrap_err();
|
||||
assert!(delete_error.contains("unknown field `returnDocument`"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn find_and_modify_return_document_rejects_invalid_values() {
|
||||
assert!(find_and_modify_returns_after(Some("after"), None, None).unwrap());
|
||||
assert!(!find_and_modify_returns_after(Some("before"), None, None).unwrap());
|
||||
assert!(find_and_modify_returns_after(Some("newest"), None, None).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn single_document_result_has_zero_or_one_metadata() {
|
||||
let empty = single_document_result(None);
|
||||
assert!(empty.documents.is_empty());
|
||||
assert_eq!(empty.total, 0);
|
||||
|
||||
let one = single_document_result(Some(doc! { "_id": 1, "name": "Ada" }));
|
||||
assert_eq!(one.documents.len(), 1);
|
||||
assert_eq!(one.total, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn json_object_to_document_preserves_unchanged_bson_date_fields() {
|
||||
let date = DateTime::parse_rfc3339_str("2026-06-10T13:59:31.287Z").unwrap();
|
||||
|
||||
@@ -121,6 +121,27 @@ pub async fn mongo_find_documents_core(
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn mongo_find_one_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter: Option<&str>,
|
||||
projection: Option<&str>,
|
||||
options: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::find_one(client, database, collection, filter, projection, options).await
|
||||
}
|
||||
// The legacy agent only exposes paginated find, which also performs a count.
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support the bounded findOne path".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mongo_count_documents_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
@@ -426,3 +447,71 @@ pub async fn mongo_delete_documents_core(
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mongo_find_one_and_update_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter_json: &str,
|
||||
update_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::find_one_and_update(client, database, collection, filter_json, update_json, options_json)
|
||||
.await
|
||||
}
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support findOneAndUpdate".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mongo_find_one_and_replace_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter_json: &str,
|
||||
replacement_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::find_one_and_replace(
|
||||
client,
|
||||
database,
|
||||
collection,
|
||||
filter_json,
|
||||
replacement_json,
|
||||
options_json,
|
||||
)
|
||||
.await
|
||||
}
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support findOneAndReplace".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn mongo_find_one_and_delete_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
filter_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
mongo_driver::find_one_and_delete(client, database, collection, filter_json, options_json).await
|
||||
}
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support findOneAndDelete".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user