Merge pull request apache#61 from derekd/derekd-fix-nimbus-topoconf-a…

…utho

Correct authorization check in nimbus methods

storm-core/src/clj/backtype/storm/daemon/nimbus.clj

@@ -881,15 +881,15 @@
     (throw (InvalidTopologyException.
             (str "Topology name cannot contain any of the following: " (pr-str DISALLOWED-TOPOLOGY-NAME-STRS))))))
 
-(defn- try-read-storm-conf [conf storm-id]
+(defn try-read-storm-conf [conf storm-id]
   (try-cause
     (read-storm-conf conf storm-id)
     (catch FileNotFoundException e
        (throw (NotAliveException. storm-id)))
   )
 )
 
-(defn- try-read-storm-topology [conf storm-id]
+(defn try-read-storm-topology [conf storm-id]
   (try-cause
     (read-storm-topology conf storm-id)
     (catch FileNotFoundException e
@@ -1110,22 +1110,22 @@
         (to-json (:conf nimbus)))
 
       (^String getTopologyConf [this ^String id]
-        (check-authorization! nimbus nil nil "getTopologyConf")
         (let [topology-conf (try-read-storm-conf conf id)
               storm-name (topology-conf TOPOLOGY-NAME)]
-              (to-json conf)))
+              (check-authorization! nimbus storm-name topology-conf "getTopologyConf")
+              (to-json topology-conf)))
 
       (^StormTopology getTopology [this ^String id]
-        (check-authorization! nimbus nil nil "getTopology")
         (let [topology-conf (try-read-storm-conf conf id)
               storm-name (topology-conf TOPOLOGY-NAME)]
-              (system-topology! conf (try-read-storm-topology conf id))))
+              (check-authorization! nimbus storm-name topology-conf "getTopology")
+              (system-topology! topology-conf (try-read-storm-topology conf id))))
 
       (^StormTopology getUserTopology [this ^String id]
-        (check-authorization! nimbus nil nil "getUserTopology")
         (let [topology-conf (try-read-storm-conf conf id)
               storm-name (topology-conf TOPOLOGY-NAME)]
-              (try-read-storm-topology conf id)))
+              (check-authorization! nimbus storm-name topology-conf "getUserTopology")
+              (try-read-storm-topology topology-conf id)))
 
       (^ClusterSummary getClusterInfo [this]
         (check-authorization! nimbus nil nil "getClusterInfo")

storm-core/test/clj/backtype/storm/nimbus_test.clj

@@ -1,7 +1,7 @@
 (ns backtype.storm.nimbus-test
   (:use [clojure test])
+  (:require [backtype.storm [util :as util]])
   (:require [backtype.storm.daemon [nimbus :as nimbus]])
-
   (:import [backtype.storm.testing TestWordCounter TestWordSpout TestGlobalCount TestAggregatesCounter])
   (:import [backtype.storm.scheduler INimbus])
   (:use [backtype.storm bootstrap testing])
@@ -875,6 +875,65 @@
   )
 )
 
+(deftest test-nimbus-check-authorization-params
+  (with-local-cluster [cluster
+                       :daemon-conf {NIMBUS-AUTHORIZER "backtype.storm.security.auth.authorizer.NoopAuthorizer"}]
+    (let [nimbus (:nimbus cluster)
+          topology-name "test-nimbus-check-autho-params"
+          topology (thrift/mk-topology {} {})]
+      ; Fake good authorization as part of setup.
+      (mocking [nimbus/check-authorization!]
+          (submit-local-topology-with-opts nimbus topology-name {} topology
+              (SubmitOptions. TopologyInitialStatus/INACTIVE)))
+      (let [expected-name topology-name
+            expected-conf {TOPOLOGY-NAME expected-name
+                           :foo :bar}]
+
+        (testing "getTopologyConf calls check-authorization! with the correct parameters."
+          (let [expected-operation "getTopologyConf"]
+            (stubbing [nimbus/check-authorization! nil
+                       nimbus/try-read-storm-conf expected-conf
+                       util/to-json nil]
+              (try
+                (.getTopologyConf nimbus "fake-id")
+                (catch NotAliveException e)
+                (finally
+                  (verify-first-call-args-for-indices
+                    nimbus/check-authorization!
+                      [1 2 3] expected-name expected-conf expected-operation)
+                  (verify-first-call-args-for util/to-json expected-conf))))))
+
+        (testing "getTopology calls check-authorization! with the correct parameters."
+          (let [expected-operation "getTopology"]
+            (stubbing [nimbus/check-authorization! nil
+                       nimbus/try-read-storm-conf expected-conf
+                       nimbus/try-read-storm-topology nil
+                       system-topology! nil]
+              (try
+                (.getTopology nimbus "fake-id")
+                (catch NotAliveException e)
+                (finally
+                  (verify-first-call-args-for-indices
+                    nimbus/check-authorization!
+                      [1 2 3] expected-name expected-conf expected-operation)
+                  (verify-first-call-args-for-indices
+                    system-topology! [0] expected-conf))))))
+
+        (testing "getUserTopology calls check-authorization with the correct parameters."
+          (let [expected-operation "getUserTopology"]
+            (stubbing [nimbus/check-authorization! nil
+                       nimbus/try-read-storm-conf expected-conf
+                       nimbus/try-read-storm-topology nil]
+              (try
+                (.getUserTopology nimbus "fake-id")
+                (catch NotAliveException e)
+                (finally
+                  (verify-first-call-args-for-indices
+                    nimbus/check-authorization!
+                      [1 2 3] expected-name expected-conf expected-operation)
+                  (verify-first-call-args-for-indices
+                    nimbus/try-read-storm-topology [0] expected-conf))))))))))
+
 (deftest test-nimbus-iface-getTopology-methods-throw-correctly
   (with-local-cluster [cluster]
     (let [