SERVER-70921 Do not drop off tenant information when running find, count and distinct commands against views

10 files changed, 99 insertions, 36 deletions

diff --git a/buildscripts/resmokeconfig/suites/native_tenant_data_isolation_with_security_token_jscore_passthrough.yml b/buildscripts/resmokeconfig/suites/native_tenant_data_isolation_with_security_token_jscore_passthrough.yml
index de347945315..17d1deaad5a 100644
--- a/buildscripts/resmokeconfig/suites/native_tenant_data_isolation_with_security_token_jscore_passthrough.yml
+++ b/buildscripts/resmokeconfig/suites/native_tenant_data_isolation_with_security_token_jscore_passthrough.yml
@@ -63,17 +63,6 @@ selector:
   # In a multitenancy environment the catalog will always return tenant-prefixed entries, and the
   # override we use in this suite checks for the absence of a prefix breaking the list_catalog tests.
   - jstests/core/list_catalog.js
-  # TODO SERVER-70921: these tests cannot work as the tenant information is missed when running find, count and distinct command against views.
-  - jstests/core/timeseries/timeseries_index_partial.js
-  - jstests/core/txns/view_reads_in_transaction.js
-  - jstests/core/views/views_aggregation.js
-  - jstests/core/views/views_collation.js
-  - jstests/core/views/views_count.js
-  - jstests/core/views/views_distinct_with_arrays.js
-  - jstests/core/views/views_find.js
-  - jstests/core/views/views_drop.js
-  - jstests/core/views/views_distinct.js
-  - jstests/core/views/dbref_projection.js
   # TODO SERVER-71470: this test cannot work as aggregate with $lookup query on multiple collections drops off the tenant informaton.
   - jstests/core/txns/aggregation_in_transaction.js
 
diff --git a/src/mongo/db/auth/validated_tenancy_scope.h b/src/mongo/db/auth/validated_tenancy_scope.h
index 07d69d179e0..98b5c06c302 100644
--- a/src/mongo/db/auth/validated_tenancy_scope.h
+++ b/src/mongo/db/auth/validated_tenancy_scope.h
@@ -115,11 +115,11 @@ public:
         : _tenantOrUser(std::move(tenant)) {}
 
     /**
-     * Backdoor API for use by FLE Query Analysis to setup a validated tenant without a security
-     * context.
+     * Backdoor API to setup a validated tenant. For use only when a security context is not
+     * available.
      */
-    struct TrustedFLEQueryAnalysisTag {};
-    explicit ValidatedTenancyScope(TenantId tenant, TrustedFLEQueryAnalysisTag)
+    struct TrustedForInnerOpMsgRequestTag {};
+    explicit ValidatedTenancyScope(TenantId tenant, TrustedForInnerOpMsgRequestTag)
         : _tenantOrUser(std::move(tenant)) {}
 
 private:
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index 5596a6a788d..9b33d9b0a1a 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -176,7 +176,9 @@ public:
             }
 
             auto viewAggCmd =
-                OpMsgRequest::fromDBAndBody(nss.db(), viewAggregation.getValue()).body;
+                OpMsgRequestBuilder::createWithValidatedTenancyScope(
+                    nss.dbName(), opMsgRequest.validatedTenancyScope, viewAggregation.getValue())
+                    .body;
             auto viewAggRequest = aggregation_request_helper::parseFromBSON(
                 opCtx,
                 nss,
@@ -267,10 +269,15 @@ public:
             ctx.reset();
 
             uassertStatusOK(viewAggregation.getStatus());
+            using VTS = auth::ValidatedTenancyScope;
+            boost::optional<VTS> vts = boost::none;
+            if (dbName.tenantId()) {
+                vts = VTS(dbName.tenantId().value(), VTS::TrustedForInnerOpMsgRequestTag{});
+            }
+            auto aggRequest = OpMsgRequestBuilder::createWithValidatedTenancyScope(
+                dbName, vts, std::move(viewAggregation.getValue()));
 
-            BSONObj aggResult = CommandHelpers::runCommandDirectly(
-                opCtx,
-                OpMsgRequest::fromDBAndBody(dbName.db(), std::move(viewAggregation.getValue())));
+            BSONObj aggResult = CommandHelpers::runCommandDirectly(opCtx, aggRequest);
 
             uassertStatusOK(ViewResponseFormatter(aggResult).appendAsCountResponse(&result));
             return true;
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index eb599424001..f61ab426299 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -176,7 +176,9 @@ public:
             }
 
             auto viewAggCmd =
-                OpMsgRequest::fromDBAndBody(nss.db(), viewAggregation.getValue()).body;
+                OpMsgRequestBuilder::createWithValidatedTenancyScope(
+                    nss.dbName(), request.validatedTenancyScope, viewAggregation.getValue())
+                    .body;
             auto viewAggRequest = aggregation_request_helper::parseFromBSON(
                 opCtx,
                 nss,
@@ -262,9 +264,15 @@ public:
             auto viewAggregation = parsedDistinct.asAggregationCommand();
             uassertStatusOK(viewAggregation.getStatus());
 
-            BSONObj aggResult = CommandHelpers::runCommandDirectly(
-                opCtx,
-                OpMsgRequest::fromDBAndBody(dbName.db(), std::move(viewAggregation.getValue())));
+            using VTS = auth::ValidatedTenancyScope;
+            boost::optional<VTS> vts = boost::none;
+            if (dbName.tenantId()) {
+                vts = VTS(dbName.tenantId().value(), VTS::TrustedForInnerOpMsgRequestTag{});
+            }
+            auto aggRequest = OpMsgRequestBuilder::createWithValidatedTenancyScope(
+                dbName, vts, std::move(viewAggregation.getValue()));
+
+            BSONObj aggResult = CommandHelpers::runCommandDirectly(opCtx, aggRequest);
             uassertStatusOK(ViewResponseFormatter(aggResult).appendAsDistinctResponse(&result));
             return true;
         }
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index 6034f67f54a..94a301b9d3f 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -311,7 +311,9 @@ public:
                     uassertStatusOK(query_request_helper::asAggregationCommand(findCommand));
 
                 auto viewAggCmd =
-                    OpMsgRequest::fromDBAndBody(_dbName.db(), viewAggregationCommand).body;
+                    OpMsgRequestBuilder::createWithValidatedTenancyScope(
+                        _dbName, _request.validatedTenancyScope, viewAggregationCommand)
+                        .body;
 
                 // Create the agg request equivalent of the find operation, with the explain
                 // verbosity included.
@@ -534,9 +536,8 @@ public:
                 const auto& findCommand = cq->getFindCommandRequest();
                 auto viewAggregationCommand =
                     uassertStatusOK(query_request_helper::asAggregationCommand(findCommand));
-                auto aggRequest =
-                    OpMsgRequestBuilder::create(_dbName, std::move(viewAggregationCommand));
-                aggRequest.validatedTenancyScope = _request.validatedTenancyScope;
+                auto aggRequest = OpMsgRequestBuilder::createWithValidatedTenancyScope(
+                    _dbName, _request.validatedTenancyScope, std::move(viewAggregationCommand));
                 BSONObj aggResult = CommandHelpers::runCommandDirectly(opCtx, aggRequest);
                 auto status = getStatusFromCommandResult(aggResult);
                 if (status.code() == ErrorCodes::InvalidPipelineOperator) {
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 225896a6b5f..df277843e3f 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -372,7 +372,8 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
                 // collection name references a view on the aggregation request's database. Note
                 // that the inverse scenario (mistaking a view for a collection) is not an issue
                 // because $merge/$out cannot target a view.
-                auto nssToCheck = NamespaceString(request.getNamespace().db(), involvedNs.coll());
+                auto nssToCheck =
+                    NamespaceString(request.getNamespace().dbName(), involvedNs.coll());
                 if (catalog->lookupView(opCtx, nssToCheck)) {
                     auto status = resolveViewDefinition(nssToCheck);
                     if (!status.isOK()) {
diff --git a/src/mongo/db/dbdirectclient.cpp b/src/mongo/db/dbdirectclient.cpp
index f217f12001d..571a10f263e 100644
--- a/src/mongo/db/dbdirectclient.cpp
+++ b/src/mongo/db/dbdirectclient.cpp
@@ -201,16 +201,18 @@ long long DBDirectClient::count(const NamespaceStringOrUUID nsOrUuid,
                                 boost::optional<BSONObj> readConcernObj,
                                 const boost::optional<TenantId>& dollarTenant) {
     invariant(!readConcernObj, "passing readConcern to DBDirectClient functions is not supported");
-    DirectClientScope directClientScope(_opCtx);
     BSONObj cmdObj = _countCmd(nsOrUuid, query, options, limit, skip, boost::none, dollarTenant);
 
     auto& dbName = (nsOrUuid.uuid() ? nsOrUuid.dbName().value() : (*nsOrUuid.nss()).dbName());
-
     auto request = OpMsgRequestBuilder::create(dbName, cmdObj);
-    auto result = CommandHelpers::runCommandDirectly(_opCtx, request);
+
+    // Calls runCommand instead of runCommandDirectly to ensure the tenant inforamtion of this
+    // command gets validated and is used for parsing the command request.
+    auto response = runCommand(request);
+    auto& result = response->getCommandReply();
 
     uassertStatusOK(getStatusFromCommandResult(result));
-    return static_cast<unsigned long long>(result["n"].numberLong());
+    return result["n"].numberLong();
 }
 
 }  // namespace mongo
diff --git a/src/mongo/rpc/op_msg.cpp b/src/mongo/rpc/op_msg.cpp
index 5f8a87438a0..8e055db9efc 100644
--- a/src/mongo/rpc/op_msg.cpp
+++ b/src/mongo/rpc/op_msg.cpp
@@ -253,7 +253,7 @@ OpMsg OpMsg::parse(const Message& message, Client* client) try {
 }
 
 OpMsgRequest OpMsgRequest::fromDBAndBody(StringData db, BSONObj body, const BSONObj& extraFields) {
-    return OpMsgRequestBuilder::create(db, std::move(body), extraFields);
+    return OpMsgRequestBuilder::create({boost::none, db}, std::move(body), extraFields);
 }
 
 boost::optional<TenantId> parseDollarTenant(const BSONObj body) {
@@ -291,8 +291,31 @@ void OpMsgRequest::setDollarTenant(const TenantId& tenant) {
     body = bodyBuilder.obj();
 }
 
-OpMsgRequest OpMsgRequestBuilder::create(StringData db, BSONObj body, const BSONObj& extraFields) {
-    return create({boost::none, db}, std::move(body), extraFields);
+OpMsgRequest OpMsgRequestBuilder::createWithValidatedTenancyScope(
+    const DatabaseName& dbName,
+    boost::optional<auth::ValidatedTenancyScope> validatedTenancyScope,
+    BSONObj body,
+    const BSONObj& extraFields) {
+    auto dollarTenant = parseDollarTenant(body);
+    OpMsgRequest request;
+    request.body = ([&] {
+        BSONObjBuilder bodyBuilder(std::move(body));
+        bodyBuilder.appendElements(extraFields);
+        if (dollarTenant) {
+            // If already having $tenant, never include the prefix in $db.
+            bodyBuilder.append("$db", dbName.db());
+        } else if (validatedTenancyScope && !validatedTenancyScope->hasAuthenticatedUser()) {
+            // Add $tenant into the body if the validated tenant id comes from $tenant.
+            bodyBuilder.append("$db", dbName.db());
+            appendDollarTenant(bodyBuilder, validatedTenancyScope->tenantId());
+        } else {
+            bodyBuilder.append("$db", DatabaseNameUtil::serialize(dbName));
+        }
+        return bodyBuilder.obj();
+    }());
+
+    request.validatedTenancyScope = validatedTenancyScope;
+    return request;
 }
 
 OpMsgRequest OpMsgRequestBuilder::create(const DatabaseName& dbName,
diff --git a/src/mongo/rpc/op_msg.h b/src/mongo/rpc/op_msg.h
index a1333446265..6563bbe7473 100644
--- a/src/mongo/rpc/op_msg.h
+++ b/src/mongo/rpc/op_msg.h
@@ -413,10 +413,26 @@ private:
  */
 struct OpMsgRequestBuilder {
 public:
-    static OpMsgRequest create(StringData db, BSONObj body, const BSONObj& extraFields = {});
+    /**
+     * Creates an OpMsgRequest object.
+     * If tenant id exists in db name, then a "$tenant" will be appended to the OpMsgRequest
+     * object's body.
+     * Do not use if creating an OpMsgRequest in order to run a command directly (i.e.
+     * CommandHelpers::runCommandDirectly). This function does not set a ValidatedTenancyScope on
+     * the request itself, which will lead to the request being parsed incorrectly.
+     */
     static OpMsgRequest create(const DatabaseName& dbName,
                                BSONObj body,
                                const BSONObj& extraFields = {});
+
+    /**
+     * Creates an OpMsgRequest object and directly sets a validated tenancy scope on it.
+     */
+    static OpMsgRequest createWithValidatedTenancyScope(
+        const DatabaseName& dbName,
+        boost::optional<auth::ValidatedTenancyScope> validatedTenancyScope,
+        BSONObj body,
+        const BSONObj& extraFields = {});
 };
 
 }  // namespace mongo
diff --git a/src/mongo/rpc/op_msg_test.cpp b/src/mongo/rpc/op_msg_test.cpp
index 5c03218e8b2..95da2298ec4 100644
--- a/src/mongo/rpc/op_msg_test.cpp
+++ b/src/mongo/rpc/op_msg_test.cpp
@@ -953,6 +953,22 @@ TEST(OpMsgRequestBuilder, WithSameTenantInBody) {
     ASSERT_EQ(TenantId::parseFromBSON(msg.body.getField("$tenant")), tenantId);
 }
 
+TEST(OpMsgRequestBuilder, WithVTS) {
+    const TenantId tenantId(OID::gen());
+    auto const body = fromjson("{ping: 1}");
+
+    using VTS = auth::ValidatedTenancyScope;
+    VTS vts = VTS(tenantId, VTS::TenantForTestingTag{});
+    OpMsgRequest msg =
+        OpMsgRequestBuilder::createWithValidatedTenancyScope({tenantId, "testDb"}, vts, body);
+    ASSERT(msg.validatedTenancyScope);
+    ASSERT_EQ(msg.validatedTenancyScope->tenantId(), tenantId);
+    // Verify $tenant is added to the msg body, as the vts does not come from security token.
+    ASSERT_EQ(msg.body.getField("$tenant").eoo(), false);
+    ASSERT_EQ(TenantId::parseFromBSON(msg.body.getField("$tenant")), tenantId);
+    ASSERT_EQ(msg.getDatabase(), "testDb");
+}
+
 TEST(OpMsgRequestBuilder, FailWithDiffTenantInBody) {
     const TenantId tenantId(OID::gen());
     const TenantId otherTenantId(OID::gen());