diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index f6032dd2e3..5b6a219f1a 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -367,8 +367,21 @@ public class Config
      * "tell" a user that there's something really wrong with the UDF.
      * When you disable async UDF execution, users MUST pay attention to read-timeouts since these may indicate
      * UDFs that run too long or forever - and this can destabilize the cluster.
+     *
+     * This requires allow_insecure_udfs to be true
      */
     public boolean enable_user_defined_functions_threads = true;
+
+    /**
+     * Set this to true to allow running insecure UDFs.
+     */
+    public boolean allow_insecure_udfs = false;
+
+    /**
+     * Set this to allow UDFs accessing java.lang.System.* methods, which basically allows UDFs to execute any arbitrary code on the system.
+     */
+    public boolean allow_extra_insecure_udfs = false;
+
     /**
      * Time in milliseconds after a warning will be emitted to the log and to the client that a UDF runs too long.
      * (Only valid, if enable_user_defined_functions_threads==true)
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index a5a02ced98..d64d054d8f 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -712,6 +712,12 @@ public class DatabaseDescriptor
         if (conf.user_defined_function_fail_timeout < conf.user_defined_function_warn_timeout)
             throw new ConfigurationException("user_defined_function_warn_timeout must less than user_defined_function_fail_timeout", false);
 
+        if (!conf.allow_insecure_udfs && !conf.enable_user_defined_functions_threads)
+            throw new ConfigurationException("To be able to set enable_user_defined_functions_threads: false you need to set allow_insecure_udfs: true - this is an unsafe configuration and is not recommended.");
+
+        if (conf.allow_extra_insecure_udfs)
+            logger.warn("Allowing java.lang.System.* access in UDFs is dangerous and not recommended. Set allow_extra_insecure_udfs: false to disable.");
+
         if (conf.commitlog_segment_size_in_mb <= 0)
             throw new ConfigurationException("commitlog_segment_size_in_mb must be positive, but was "
                     + conf.commitlog_segment_size_in_mb, false);
@@ -2513,6 +2519,16 @@ public class DatabaseDescriptor
         conf.user_defined_function_warn_timeout = userDefinedFunctionWarnTimeout;
     }
 
+    public static boolean allowInsecureUDFs()
+    {
+        return conf.allow_insecure_udfs;
+    }
+
+    public static boolean allowExtraInsecureUDFs()
+    {
+        return conf.allow_extra_insecure_udfs;
+    }
+
     public static boolean getEnableMaterializedViews()
     {
         return conf.enable_materialized_views;
diff --git a/src/java/org/apache/cassandra/cql3/functions/UDFunction.java b/src/java/org/apache/cassandra/cql3/functions/UDFunction.java
index c06980d8e6..6dab536284 100644
--- a/src/java/org/apache/cassandra/cql3/functions/UDFunction.java
+++ b/src/java/org/apache/cassandra/cql3/functions/UDFunction.java
@@ -155,6 +155,11 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
     "java/util/zip/",
     };
 
+    private static final String[] disallowedPatternsSyncUDF =
+    {
+    "java/lang/System.class"
+    };
+
     static boolean secureResource(String resource)
     {
         while (resource.startsWith("/"))
@@ -163,14 +168,26 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
         for (String allowed : allowedPatterns)
             if (resource.startsWith(allowed))
             {
-
                 // resource is in allowedPatterns, let's see if it is not explicitly disallowed
                 for (String disallowed : disallowedPatterns)
+                {
                     if (resource.startsWith(disallowed))
                     {
                         logger.trace("access denied: resource {}", resource);
                         return false;
                     }
+                }
+                if (!DatabaseDescriptor.enableUserDefinedFunctionsThreads() && !DatabaseDescriptor.allowExtraInsecureUDFs())
+                {
+                    for (String disallowed : disallowedPatternsSyncUDF)
+                    {
+                        if (resource.startsWith(disallowed))
+                        {
+                            logger.trace("access denied: resource {}", resource);
+                            return false;
+                        }
+                    }
+                }
 
                 return true;
             }
diff --git a/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java b/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java
index c9402f153d..3d72559c3b 100644
--- a/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java
+++ b/src/java/org/apache/cassandra/security/ThreadAwareSecurityManager.java
@@ -32,6 +32,7 @@ import java.util.Enumeration;
 import io.netty.util.concurrent.FastThreadLocal;
 
 import org.apache.cassandra.utils.logging.LoggingSupportFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
 
 /**
  * Custom {@link SecurityManager} and {@link Policy} implementation that only performs access checks
@@ -65,6 +66,7 @@ public final class ThreadAwareSecurityManager extends SecurityManager
     private static final RuntimePermission CHECK_MEMBER_ACCESS_PERMISSION = new RuntimePermission("accessDeclaredMembers");
     private static final RuntimePermission MODIFY_THREAD_PERMISSION = new RuntimePermission("modifyThread");
     private static final RuntimePermission MODIFY_THREADGROUP_PERMISSION = new RuntimePermission("modifyThreadGroup");
+    private static final RuntimePermission SET_SECURITY_MANAGER_PERMISSION = new RuntimePermission("setSecurityManager");
 
     private static volatile boolean installed;
 
@@ -181,6 +183,9 @@ public final class ThreadAwareSecurityManager extends SecurityManager
 
     public void checkPermission(Permission perm)
     {
+        if (!DatabaseDescriptor.enableUserDefinedFunctionsThreads() && !DatabaseDescriptor.allowExtraInsecureUDFs() && SET_SECURITY_MANAGER_PERMISSION.equals(perm))
+            throw new AccessControlException("Access denied");
+
         if (!isSecuredThread())
             return;