From 3058502e8746d9fed6a0b236f722625bae0e3b5d Mon Sep 17 00:00:00 2001
From: Manohar Vanam <manohar.crazy09@gmail.com>
Date: Wed, 13 Jun 2018 19:38:04 +0530
Subject: [PATCH] [STORM-3094] : Added topology name validation at client side

---
 .../src/jvm/org/apache/storm/StormSubmitter.java  |  4 +++-
 .../jvm/org/apache/storm/blobstore/BlobStore.java |  6 ++----
 .../src/jvm/org/apache/storm/utils/Utils.java     | 15 +++++++++++++++
 3 files changed, 20 insertions(+), 5 deletions(-)

diff --git a/storm-core/src/jvm/org/apache/storm/StormSubmitter.java b/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
index 86ec6c2cab4..4827dbe07f9 100644
--- a/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
+++ b/storm-core/src/jvm/org/apache/storm/StormSubmitter.java
@@ -236,7 +236,9 @@ public static void submitTopologyAs(String name, Map stormConf, StormTopology to
                     if (topologyNameExists(name, client)) {
                         throw new RuntimeException("Topology with name `" + name + "` already exists on cluster");
                     }
-
+                    if (!Utils.isValidKey(name)) {
+                        throw new IllegalArgumentException(name + " does not appear to be a valid topology name.");
+                    }
                     // Dependency uploading only makes sense for distributed mode
                     List<String> jarsBlobKeys = Collections.emptyList();
                     List<String> artifactsBlobKeys;
diff --git a/storm-core/src/jvm/org/apache/storm/blobstore/BlobStore.java b/storm-core/src/jvm/org/apache/storm/blobstore/BlobStore.java
index 92f32ec31ec..b9ab21a154b 100644
--- a/storm-core/src/jvm/org/apache/storm/blobstore/BlobStore.java
+++ b/storm-core/src/jvm/org/apache/storm/blobstore/BlobStore.java
@@ -31,7 +31,7 @@
 import javax.security.auth.Subject;
 
 import org.apache.storm.nimbus.NimbusInfo;
-import org.apache.commons.lang.StringUtils;
+import org.apache.storm.utils.Utils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -61,7 +61,6 @@
  */
 public abstract class BlobStore implements Shutdownable {
     private static final Logger LOG = LoggerFactory.getLogger(BlobStore.class);
-    private static final Pattern KEY_PATTERN = Pattern.compile("^[\\w \\t\\.:_-]+$", Pattern.UNICODE_CHARACTER_CLASS);
     protected static final String BASE_BLOBS_DIR_NAME = "blobs";
 
     /**
@@ -194,8 +193,7 @@ public <R> Set<R> filterAndListKeys(KeyFilter<R> filter) {
      * @param key Key for the blob.
      */
     public static final void validateKey(String key) throws AuthorizationException {
-        if (StringUtils.isEmpty(key) || "..".equals(key) || ".".equals(key) || !KEY_PATTERN.matcher(key).matches()) {
-            LOG.error("'{}' does not appear to be valid {}", key, KEY_PATTERN);
+        if (!Utils.isValidKey(key)) {
             throw new AuthorizationException(key+" does not appear to be a valid blob key");
         }
     }
diff --git a/storm-core/src/jvm/org/apache/storm/utils/Utils.java b/storm-core/src/jvm/org/apache/storm/utils/Utils.java
index 8892ba93365..c5b6f390d44 100644
--- a/storm-core/src/jvm/org/apache/storm/utils/Utils.java
+++ b/storm-core/src/jvm/org/apache/storm/utils/Utils.java
@@ -170,6 +170,7 @@ public static Utils setInstance(Utils u) {
     public static final boolean IS_ON_WINDOWS = "Windows_NT".equals(System.getenv("OS"));
     public static final String FILE_PATH_SEPARATOR = System.getProperty("file.separator");
     public static final String CLASS_PATH_SEPARATOR = System.getProperty("path.separator");
+    private static final Pattern TOPOLOGY_KEY_PATTERN = Pattern.compile("^[\\w \\t\\.:_-]+$", Pattern.UNICODE_CHARACTER_CLASS);
 
     public static final int SIGKILL = 9;
     public static final int SIGTERM = 15;
@@ -2320,4 +2321,18 @@ public static StormTopology addVersions(StormTopology topology) {
         }
         return topology;
     }
+
+    /**
+     * Validates topology name / blob key.
+     *
+     * @param key topology name / Key for the blob.
+     */
+    public static boolean isValidKey(String key) {
+        if (StringUtils.isEmpty(key) || "..".equals(key) || ".".equals(key) || !TOPOLOGY_KEY_PATTERN.matcher(key).matches()) {
+            LOG.error("'{}' does not appear to be valid. It must match {}. And it can't be \".\", \"..\", null or empty string.", key,
+                    TOPOLOGY_KEY_PATTERN);
+            return false;
+        }
+        return true;
+    }
 }