Better 'not master' validation

db/instance.cpp

@@ -28,6 +28,7 @@
 #include "security.h"
 #include "json.h"
 #include "reccache.h"
+#include "replset.h"
 #include "../s/d_logic.h"
 
 namespace mongo {
@@ -302,6 +303,7 @@ namespace mongo {
         DbMessage d(m);
         const char *ns = d.getns();
         assert(*ns);
+        uassert( "not master", isMaster( ns ) );
         setClient(ns);
         //if( database->profile )
         ss << ns << ' ';
@@ -322,6 +324,7 @@ namespace mongo {
         DbMessage d(m);
         const char *ns = d.getns();
         assert(*ns);
+        uassert( "not master", isMaster( ns ) );
         setClient(ns);
         int flags = d.pullInt();
         bool justOne = flags & 1;
@@ -359,6 +362,9 @@ namespace mongo {
                 }
             }
 
+            // Check before setClient() to avoid creating ns unnecessarily.
+            uassert( "not master", isMasterNs( q.ns ) || (q.queryOptions & Option_SlaveOk) || strstr( q.ns, ".$cmd" ) );
+            
             setClient( q.ns );
             strncpy(currentOp.ns, q.ns, Namespace::MaxNsLen);
             msgdata = runQuery(m, ss ).release();
@@ -440,6 +446,7 @@ namespace mongo {
         DbMessage d(m);
 		const char *ns = d.getns();
 		assert(*ns);
+        uassert( "not master", isMaster( ns ) );
 		setClient(ns);
 		ss << ns;
 		

db/query.cpp

@@ -1161,8 +1161,8 @@ namespace mongo {
             AuthenticationInfo *ai = authInfo.get();
             uassert("unauthorized", ai->isAuthorized(database->name.c_str()));
             
-            uassert("not master", isMaster() || (queryOptions & Option_SlaveOk));
-            
+            uassert( "not master", isMaster() || (queryOptions & Option_SlaveOk) );
+
             BSONElement hint;
             bool explain = false;
             bool _gotquery = false;

db/replset.h

@@ -107,16 +107,24 @@ namespace mongo {
 
        See also CmdIsMaster.
 
+       If 'client' is not specified, the current client is used.
     */
-    inline bool isMaster() {
+    inline bool isMaster( const char *client = 0 ) {
+        if ( !client )
+            client = database->name.c_str();
         if ( replAllDead ) {
-            return database->name == "local";
+            return strcmp( client, "local" ) == 0;
         }
 
         if ( replPair == 0 || replPair->state == ReplPair::State_Master )
             return true;
 
-        return database->name == "local";
+        return strcmp( client, "local" ) == 0;
+    }
+    inline bool isMasterNs( const char *ns ) {
+        char cl[ 256 ];
+        nsToClient( ns, cl );
+        return isMaster( cl );
     }
 
     inline ReplPair::ReplPair(const char *remoteEnd, const char *arb) {

jstests/repl/pair1.js

@@ -3,7 +3,7 @@
 var baseName = "jstests_pair1test";
 
 ismaster = function( n ) {
-    im = n.getDB( "admin" ).runCommand( { "ismaster" : 1 } );
+    var im = n.getDB( "admin" ).runCommand( { "ismaster" : 1 } );
 //    print( "ismaster: " + tojson( im ) );
     assert( im );
     return im.ismaster;
@@ -32,6 +32,25 @@ checkWrite = function( m, s ) {
                 } );
 }
 
+// check that slave reads and writes are guarded
+checkSlaveGuard = function( s ) {
+    var t = s.getDB( baseName + "-temp" ).temp;
+    assert.throws( t.find().count, {}, "not master" );
+    assert.throws( t.find(), {}, "not master", "find did not assert" );
+    
+    checkError = function() {
+        assert.eq( "not master", s.getDB( "admin" ).getLastError() );
+        s.getDB( "admin" ).resetError();
+    }
+    s.getDB( "admin" ).resetError();
+    t.save( {x:1} );
+    checkError();
+    t.update( {}, {x:2}, true );
+    checkError();
+    t.remove( {x:0} );
+    checkError();
+}
+
 doTest = function( signal ) {
 
     // spec small oplog for fast startup on 64bit machines
@@ -51,8 +70,7 @@ doTest = function( signal ) {
                 return ( lm == 0 && rm == 1 );
                 } );
     
-    // Check that reading from slave fails
-    assert.throws( l.getDB( baseName + "-temp" ).temp.find().count, {}, "not master" );
+    checkSlaveGuard( l );
     
     checkWrite( r, l );