diff --git a/src/networking.c b/src/networking.c
index d9d300ac20..e79b18d70b 100644
--- a/src/networking.c
+++ b/src/networking.c
@@ -160,6 +160,7 @@ client *createClient(connection *conn) {
     c->bulklen = -1;
     c->sentlen = 0;
     c->flags = 0;
+    c->slot = -1;
     c->ctime = c->lastinteraction = server.unixtime;
     clientSetDefaultAuth(c);
     c->replstate = REPL_STATE_NONE;
@@ -2026,6 +2027,7 @@ void resetClient(client *c) {
     c->reqtype = 0;
     c->multibulklen = 0;
     c->bulklen = -1;
+    c->slot = -1;
 
     if (c->deferred_reply_errors)
         listRelease(c->deferred_reply_errors);
diff --git a/src/script.c b/src/script.c
index f3b7b4bb31..8216b47f54 100644
--- a/src/script.c
+++ b/src/script.c
@@ -36,6 +36,7 @@ scriptFlag scripts_flags_def[] = {
     {.flag = SCRIPT_FLAG_ALLOW_OOM, .str = "allow-oom"},
     {.flag = SCRIPT_FLAG_ALLOW_STALE, .str = "allow-stale"},
     {.flag = SCRIPT_FLAG_NO_CLUSTER, .str = "no-cluster"},
+    {.flag = SCRIPT_FLAG_ALLOW_CROSS_SLOT, .str = "allow-cross-slot-keys"},
     {.flag = 0, .str = NULL}, /* flags array end */
 };
 
@@ -218,6 +219,10 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
         run_ctx->flags |= SCRIPT_ALLOW_OOM;
     }
 
+    if ((script_flags & SCRIPT_FLAG_EVAL_COMPAT_MODE) || (script_flags & SCRIPT_FLAG_ALLOW_CROSS_SLOT)) {
+        run_ctx->flags |= SCRIPT_ALLOW_CROSS_SLOT;
+    }
+
     /* set the curr_run_ctx so we can use it to kill the script if needed */
     curr_run_ctx = run_ctx;
 
@@ -391,7 +396,7 @@ static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
     return C_OK;
 }
 
-static int scriptVerifyClusterState(client *c, client *original_c, sds *err) {
+static int scriptVerifyClusterState(scriptRunCtx *run_ctx, client *c, client *original_c, sds *err) {
     if (!server.cluster_enabled || mustObeyClient(original_c)) {
         return C_OK;
     }
@@ -402,7 +407,8 @@ static int scriptVerifyClusterState(client *c, client *original_c, sds *err) {
     /* Duplicate relevant flags in the script client. */
     c->flags &= ~(CLIENT_READONLY | CLIENT_ASKING);
     c->flags |= original_c->flags & (CLIENT_READONLY | CLIENT_ASKING);
-    if (getNodeByQuery(c, c->cmd, c->argv, c->argc, NULL, &error_code) != server.cluster->myself) {
+    int hashslot = -1;
+    if (getNodeByQuery(c, c->cmd, c->argv, c->argc, &hashslot, &error_code) != server.cluster->myself) {
         if (error_code == CLUSTER_REDIR_DOWN_RO_STATE) {
             *err = sdsnew(
                     "Script attempted to execute a write command while the "
@@ -416,6 +422,19 @@ static int scriptVerifyClusterState(client *c, client *original_c, sds *err) {
         }
         return C_ERR;
     }
+
+    /* If the script declared keys in advanced, the cross slot error would have
+     * already been thrown. This is only checking for cross slot keys being accessed
+     * that weren't pre-declared. */
+    if (hashslot != -1 && !(run_ctx->flags & SCRIPT_ALLOW_CROSS_SLOT)) {
+        if (original_c->slot == -1) {
+            original_c->slot = hashslot;
+        } else if (original_c->slot != hashslot) {
+            *err = sdsnew("Script attempted to access keys that do not hash to "
+                    "the same slot");
+            return C_ERR;
+        }
+    }
     return C_OK;
 }
 
@@ -520,7 +539,7 @@ void scriptCall(scriptRunCtx *run_ctx, robj* *argv, int argc, sds *err) {
         run_ctx->flags |= SCRIPT_WRITE_DIRTY;
     }
 
-    if (scriptVerifyClusterState(c, run_ctx->original_client, err) != C_OK) {
+    if (scriptVerifyClusterState(run_ctx, c, run_ctx->original_client, err) != C_OK) {
         goto error;
     }
 
diff --git a/src/script.h b/src/script.h
index 9785af0958..d855c80e22 100644
--- a/src/script.h
+++ b/src/script.h
@@ -64,6 +64,7 @@
 #define SCRIPT_READ_ONLY              (1ULL<<5) /* indicate that the current script should only perform read commands */
 #define SCRIPT_ALLOW_OOM              (1ULL<<6) /* indicate to allow any command even if OOM reached */
 #define SCRIPT_EVAL_MODE              (1ULL<<7) /* Indicate that the current script called from legacy Lua */
+#define SCRIPT_ALLOW_CROSS_SLOT       (1ULL<<8) /* Indicate that the current script may access keys from multiple slots */
 typedef struct scriptRunCtx scriptRunCtx;
 
 struct scriptRunCtx {
@@ -82,6 +83,7 @@ struct scriptRunCtx {
 #define SCRIPT_FLAG_ALLOW_STALE      (1ULL<<2)
 #define SCRIPT_FLAG_NO_CLUSTER       (1ULL<<3)
 #define SCRIPT_FLAG_EVAL_COMPAT_MODE (1ULL<<4) /* EVAL Script backwards compatible behavior, no shebang provided */
+#define SCRIPT_FLAG_ALLOW_CROSS_SLOT (1ULL<<5)
 
 /* Defines a script flags */
 typedef struct scriptFlag {
diff --git a/src/server.c b/src/server.c
index 47907a4bd6..c603af02eb 100644
--- a/src/server.c
+++ b/src/server.c
@@ -3676,17 +3676,16 @@ int processCommand(client *c) {
         !(!(c->cmd->flags&CMD_MOVABLE_KEYS) && c->cmd->key_specs_num == 0 &&
           c->cmd->proc != execCommand))
     {
-        int hashslot;
         int error_code;
         clusterNode *n = getNodeByQuery(c,c->cmd,c->argv,c->argc,
-                                        &hashslot,&error_code);
+                                        &c->slot,&error_code);
         if (n == NULL || n != server.cluster->myself) {
             if (c->cmd->proc == execCommand) {
                 discardTransaction(c);
             } else {
                 flagTransaction(c);
             }
-            clusterRedirectClient(c,n,hashslot,error_code);
+            clusterRedirectClient(c,n,c->slot,error_code);
             c->cmd->rejected_calls++;
             return C_OK;
         }
diff --git a/src/server.h b/src/server.h
index da4a1ebcdf..ef0215fe00 100644
--- a/src/server.h
+++ b/src/server.h
@@ -1107,6 +1107,7 @@ typedef struct client {
                                buffer or object being sent. */
     time_t ctime;           /* Client creation time. */
     long duration;          /* Current command duration. Used for measuring latency of blocking/non-blocking cmds */
+    int slot;               /* The slot the client is executing against. Set to -1 if no slot is being used */
     time_t lastinteraction; /* Time of the last interaction, used for timeout */
     time_t obuf_soft_limit_reached_time;
     uint64_t flags;         /* Client flags: CLIENT_* macros. */
diff --git a/tests/support/util.tcl b/tests/support/util.tcl
index a7972a8548..6741b719a1 100644
--- a/tests/support/util.tcl
+++ b/tests/support/util.tcl
@@ -77,6 +77,12 @@ proc getInfoProperty {infostr property} {
     }
 }
 
+proc cluster_info {r field} {
+    if {[regexp "^$field:(.*?)\r\n" [$r cluster info] _ value]} {
+        set _ $value
+    }
+}
+
 # Return value for INFO property
 proc status {r property} {
     set _ [getInfoProperty [{*}$r info] $property]
diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl
index 277fa38038..134c5c46b3 100644
--- a/tests/test_helper.tcl
+++ b/tests/test_helper.tcl
@@ -94,6 +94,7 @@ set ::all_tests {
     unit/client-eviction
     unit/violations
     unit/replybufsize
+    unit/cluster-scripting
 }
 # Index to the next test to run in the ::all_tests list.
 set ::next_test 0
@@ -274,6 +275,16 @@ proc s {args} {
     status [srv $level "client"] [lindex $args 0]
 }
 
+# Provide easy access to CLUSTER INFO properties. Same semantic as "proc s".
+proc csi {args} {
+    set level 0
+    if {[string is integer [lindex $args 0]]} {
+        set level [lindex $args 0]
+        set args [lrange $args 1 end]
+    }
+    cluster_info [srv $level "client"] [lindex $args 0]
+}
+
 # Test wrapped into run_solo are sent back from the client to the
 # test server, so that the test server will send them again to
 # clients once the clients are idle.
diff --git a/tests/unit/cluster-scripting.tcl b/tests/unit/cluster-scripting.tcl
new file mode 100644
index 0000000000..72fc028c39
--- /dev/null
+++ b/tests/unit/cluster-scripting.tcl
@@ -0,0 +1,64 @@
+# make sure the test infra won't use SELECT
+set old_singledb $::singledb
+set ::singledb 1
+
+start_server {overrides {cluster-enabled yes} tags {external:skip cluster}} {
+    r 0 cluster addslotsrange 0 16383
+    wait_for_condition 50 100 {
+        [csi 0 cluster_state] eq "ok"
+    } else {
+        fail "Cluster never became 'ok'"
+    }
+
+    test {Eval scripts with shebangs and functions default to no cross slots} {
+        # Test that scripts with shebang block cross slot operations
+        assert_error "ERR Script attempted to access keys that do not hash to the same slot*" {
+            r 0 eval {#!lua
+                redis.call('set', 'foo', 'bar')
+                redis.call('set', 'bar', 'foo')
+                return 'OK'
+            } 0}
+
+        # Test the functions by default block cross slot operations
+        r 0 function load REPLACE {#!lua name=crossslot
+            local function test_cross_slot(keys, args)
+                redis.call('set', 'foo', 'bar')
+                redis.call('set', 'bar', 'foo')
+                return 'OK'
+            end
+
+            redis.register_function('test_cross_slot', test_cross_slot)}
+        assert_error "ERR Script attempted to access keys that do not hash to the same slot*" {r FCALL test_cross_slot 0}
+    }
+
+    test {Cross slot commands are allowed by default for eval scripts and with allow-cross-slot-keys flag} {
+        # Old style lua scripts are allowed to access cross slot operations
+        r 0 eval "redis.call('set', 'foo', 'bar'); redis.call('set', 'bar', 'foo')" 0
+
+        # scripts with allow-cross-slot-keys flag are allowed
+        r 0 eval {#!lua flags=allow-cross-slot-keys
+            redis.call('set', 'foo', 'bar'); redis.call('set', 'bar', 'foo')
+        } 0
+
+        # Functions with allow-cross-slot-keys flag are allowed
+        r 0 function load REPLACE {#!lua name=crossslot
+            local function test_cross_slot(keys, args)
+                redis.call('set', 'foo', 'bar')
+                redis.call('set', 'bar', 'foo')
+                return 'OK'
+            end
+
+            redis.register_function{function_name='test_cross_slot', callback=test_cross_slot, flags={ 'allow-cross-slot-keys' }}}
+        r FCALL test_cross_slot 0
+    }
+    
+    test {Cross slot commands are also blocked if they disagree with pre-declared keys} {
+        assert_error "ERR Script attempted to access keys that do not hash to the same slot*" {
+            r 0 eval {#!lua
+                redis.call('set', 'foo', 'bar')
+                return 'OK'
+            } 1 bar}
+    }
+}
+
+set ::singledb $old_singledb
diff --git a/tests/unit/functions.tcl b/tests/unit/functions.tcl
index 5ea71e53e3..7b781c5881 100644
--- a/tests/unit/functions.tcl
+++ b/tests/unit/functions.tcl
@@ -999,7 +999,7 @@ start_server {tags {"scripting"}} {
 
         r config set maxmemory 1
 
-        assert_match {OK} [r fcall f1 1 k]
+        assert_match {OK} [r fcall f1 1 x]
         assert_match {1} [r get x]
 
         r config set maxmemory 0
diff --git a/tests/unit/moduleapi/cluster.tcl b/tests/unit/moduleapi/cluster.tcl
index f1238992de..f4ebaab1bd 100644
--- a/tests/unit/moduleapi/cluster.tcl
+++ b/tests/unit/moduleapi/cluster.tcl
@@ -2,22 +2,6 @@
 
 source tests/support/cli.tcl
 
-proc cluster_info {r field} {
-    if {[regexp "^$field:(.*?)\r\n" [$r cluster info] _ value]} {
-        set _ $value
-    }
-}
-
-# Provide easy access to CLUSTER INFO properties. Same semantic as "proc s".
-proc csi {args} {
-    set level 0
-    if {[string is integer [lindex $args 0]]} {
-        set level [lindex $args 0]
-        set args [lrange $args 1 end]
-    }
-    cluster_info [srv $level "client"] [lindex $args 0]
-}
-
 set testmodule [file normalize tests/modules/blockonkeys.so]
 set testmodule_nokey [file normalize tests/modules/blockonbackground.so]
 set testmodule_blockedclient [file normalize tests/modules/blockedclient.so]