Add unsupported headers (#1367)

* add unsupported headers

* add unsupported headers list in config-full.edn

* test unsupported headers in websocket upgrade request

* fixintegration test naming

* undo removal of previous test and add way to retrieve unsupported headers

* undo  https-redirect changes

* rename test to include ws

* add test for on the fly request with unsupported headers

* remove unecessary tests for x-waiter-maintenance in test-wrap-maintenance unit test

* fix annotation on test to be integration test

* remove unecessary temp function

* move header error when computing descriptor

* remove unecessary code

* remove unecessary changes to compute

* fix indentation

* set log level info, sort, appropriately name error-details

* change unit test to use set

Co-authored-by: Shams <shamsimam@users.noreply.github.com>

Co-authored-by: Shams <shamsimam@users.noreply.github.com>

waiter/config-full.edn

@@ -876,7 +876,11 @@
                                :queue-timeout-ms 300000
 
                                ;; Configures the idle timeout (milliseconds) in the response output stream:
-                               :streaming-timeout-ms 20000}
+                               :streaming-timeout-ms 20000
+
+                               ;; unsupported waiter headers which will cause requests to fail if these headers
+                               ;; are provided
+                               :unsupported-headers #{"x-waiter-maintenance"}}
 
  ; ---------- Load Balancing ----------
 

waiter/integration/waiter/basic_test.clj

@@ -1505,3 +1505,23 @@
                                 :x-waiter-queue-timeout 5000)
               response (make-kitchen-request waiter-url request-headers :path "/hello")]
           (assert-response-status response http-503-service-unavailable))))))
+
+(deftest  ^:parallel ^:integration-fast test-request-unsupported-headers
+  (testing-using-waiter-url
+    (let [settings (waiter-settings waiter-url)
+          waiter-unsupported-headers (get-in settings [:instance-request-properties :unsupported-headers])]
+
+      (testing "should give 400 if unsupported headers are specified"
+        (doseq [waiter-header waiter-unsupported-headers]
+          (let [{:keys [service-id body] :as response}
+                (make-request-with-debug-info {:x-waiter-name (rand-name)
+                                               waiter-header "some value"}
+                                              #(make-kitchen-request waiter-url % :path "/hello"))]
+            (assert-response-status response http-400-bad-request)
+            (is (nil? service-id))
+            (is (str/includes? body "Unsupported waiter headers found"))
+            (is (str/includes? body waiter-header))
+
+            ; cleanup in case a service was created
+            (when service-id
+              (delete-service waiter-url service-id))))))))

waiter/integration/waiter/websocket_integration_test.clj

@@ -216,6 +216,42 @@
         (finally
           (delete-token-and-assert waiter-url token))))))
 
+(deftest ^:parallel ^:integration-fast test-request-ws-unsupported-headers
+  (testing-using-waiter-url
+    (let [token (str "token-" (rand-name))
+          token-description (assoc (kitchen-request-headers :prefix "")
+                              :authentication "disabled"
+                              :metric-group "waiter_ws_test"
+                              :name (rand-name)
+                              :permitted-user "*"
+                              :run-as-user (retrieve-username)
+                              :token token)
+          waiter-headers {"x-waiter-token" token}
+          settings (waiter-settings waiter-url)
+          waiter-unsupported-headers (get-in settings [:instance-request-properties :unsupported-headers])]
+
+      (testing "should give 400 if unsupported headers are specified"
+        (doseq [waiter-header waiter-unsupported-headers]
+          (try
+            (let [token-response (post-token waiter-url token-description)]
+              (assert-response-status token-response http-200-ok)
+              (try
+                (let [response-promise (promise)
+                      connection (ws-client/connect!
+                                   (websocket-client-factory)
+                                   (ws-url waiter-url "/websocket-auth")
+                                   (fn [{:keys [out]}]
+                                     (deliver response-promise :success)
+                                     (async/close! out))
+                                   {:middleware (fn [_ ^UpgradeRequest request]
+                                                  (.setHeader request waiter-header "some value")
+                                                  (websocket/add-headers-to-upgrade-request! request waiter-headers))})
+                      [close-code error] (connection->ctrl-data connection)]
+                  (assert-websocket-upgrade-error-response close-code error  "Unexpected HTTP Response Status Code: 400 Bad Request")
+                  (is (not (realized? response-promise))))))
+            (finally
+              (delete-token-and-assert waiter-url token))))))))
+
 (deftest ^:parallel ^:integration-fast test-request-maintenance-mode
   (testing-using-waiter-url
     (let [token (str "token-" (rand-name))
@@ -245,27 +281,6 @@
                     [close-code error] (connection->ctrl-data connection)]
                 (assert-websocket-upgrade-error-response close-code error  "Unexpected HTTP Response Status Code: 503 Service Unavailable")
                 (is (not (realized? response-promise))))))
-          (finally
-            (delete-token-and-assert waiter-url token))))
-
-      (testing "should give 400 if x-waiter-maintenance header is specified"
-        (try
-          (let [token-response (post-token waiter-url token-description)]
-            (assert-response-status token-response http-200-ok)
-            (try
-              (let [response-promise (promise)
-                    connection (ws-client/connect!
-                                 (websocket-client-factory)
-                                 (ws-url waiter-url "/websocket-auth")
-                                 (fn [{:keys [out]}]
-                                   (deliver response-promise :success)
-                                   (async/close! out))
-                                 {:middleware (fn [_ ^UpgradeRequest request]
-                                                (.setHeader request "x-waiter-maintenance" "some value")
-                                                (websocket/add-headers-to-upgrade-request! request waiter-headers))})
-                    [close-code error] (connection->ctrl-data connection)]
-                (assert-websocket-upgrade-error-response close-code error  "Unexpected HTTP Response Status Code: 400 Bad Request")
-                (is (not (realized? response-promise))))))
           (finally
             (delete-token-and-assert waiter-url token)))))))
 

waiter/src/waiter/core.clj

@@ -972,11 +972,12 @@
                             (let [position-generator-atom (atom 0)]
                               (fn determine-priority-fn [waiter-headers]
                                 (pr/determine-priority position-generator-atom waiter-headers))))
-   :discover-service-parameters-fn (pc/fnk [[:state kv-store waiter-hostnames]
+   :discover-service-parameters-fn (pc/fnk [[:settings [:instance-request-properties unsupported-headers]]
+                                            [:state kv-store waiter-hostnames]
                                             attach-service-defaults-fn attach-token-defaults-fn]
                                      (fn discover-service-parameters-fn [headers]
                                        (sd/discover-service-parameters
-                                         kv-store attach-service-defaults-fn attach-token-defaults-fn waiter-hostnames headers)))
+                                         kv-store attach-service-defaults-fn attach-token-defaults-fn waiter-hostnames headers unsupported-headers)))
    :enable-work-stealing-support? (pc/fnk [[:settings [:work-stealing supported-distribution-schemes]]
                                            service-id->service-description-fn]
                                     (fn enable-work-stealing-support? [service-id]
@@ -1894,6 +1895,7 @@
                                                (fn waiter-request-interstitial-handler-fn [request]
                                                  (interstitial/display-interstitial-handler request))))
    :websocket-request-acceptor (pc/fnk [[:routines wrap-service-discovery-fn]
+                                        [:settings [:instance-request-properties unsupported-headers]]
                                         [:state server-name]
                                         websocket-secure-request-acceptor-fn]
                                  ; If adding new middleware for websocket upgrade requests, consider adding the same middleware to
@@ -1904,7 +1906,8 @@
                                                    pr/wrap-maintenance-mode-acceptor
                                                    handler/wrap-wss-redirect
                                                    ws/wrap-service-discovery-data
-                                                   wrap-service-discovery-fn)]
+                                                   wrap-service-discovery-fn
+                                                   ws/wrap-ws-acceptor-error-handling)]
                                    (ws/make-websocket-request-acceptor server-name handler)))
    :websocket-secure-request-acceptor-fn (pc/fnk [[:state passwords]]
                                            (fn websocket-secure-request-acceptor-fn

waiter/src/waiter/process_request.clj

@@ -17,6 +17,7 @@
   (:require [clj-time.core :as t]
             [clojure.core.async :as async]
             [clojure.core.async.impl.protocols :as protocols]
+            [clojure.set :as set]
             [clojure.string :as str]
             [clojure.tools.logging :as log]
             [full.async :as fa]
@@ -877,42 +878,30 @@
 
 (defn- make-maintenance-mode
   "Check if a service's token is in maintenance mode and pass a 503 response
-  with a custom message if specified to the on-maintenance-mode-error parameter.
-  Check if x-waiter-maintenance header is set and pass 400 response to the
-  on-header-error parameter."
-  [handler on-header-error on-maintenance-mode-error]
-  (fn maintenance-mode-handler [{{:keys [service-description-template token waiter-headers]
+  with a custom message if specified to the on-maintenance-mode-error parameter."
+  [handler on-maintenance-mode-error]
+  (fn maintenance-mode-handler [{{:keys [service-description-template token]
          {:strs [maintenance owner]} :token-metadata} :waiter-discovery
         :as request}]
     (let [response-map {:error-class error-class-maintenance
                         :name (get service-description-template "name")
                         :token token
                         :token-owner owner}]
-      (cond (contains? waiter-headers "x-waiter-maintenance")
-            (do
-              (log/info "x-waiter-maintenance is not supported as an on-the-fly header"
-                        {:service-description service-description-template :token token})
-              (on-header-error {:message "The maintenance parameter is not supported for on-the-fly requests"
-                                :status http-400-bad-request
-                                :waiter-headers waiter-headers}
-                               request))
-            (some? maintenance)
-            (do
-              (log/info (str "token " token " is in maintenance mode"))
-              (meters/mark! (metrics/waiter-meter "maintenance" "response-rate"))
-              (on-maintenance-mode-error {:details response-map
-                                          :message (get maintenance "message")
-                                          :status http-503-service-unavailable}
-                                         request))
-            :else (handler request)))))
+      (if (some? maintenance)
+        (do
+          (log/info (str "token " token " is in maintenance mode"))
+          (meters/mark! (metrics/waiter-meter "maintenance" "response-rate"))
+          (on-maintenance-mode-error {:details response-map
+                                      :message (get maintenance "message")
+                                      :status http-503-service-unavailable}
+                                     request))
+        (handler request)))))
 
 (defn wrap-maintenance-mode
   "Middleware to check for maintenance mode of a token and if improper maintenance mode header is provided"
   [handler]
   (make-maintenance-mode
     handler
-    (fn send-header-error [data-map request]
-      (utils/data->error-response data-map request))
     (fn send-maintenance-mode-error [data-map request]
       (utils/data->maintenance-mode-response data-map request))))
 
@@ -923,7 +912,7 @@
   (let [on-error (fn send-ws-error [{:keys [message status]} {^ServletUpgradeResponse upgrade-response :upgrade-response}]
                    (.sendError upgrade-response status message)
                    status)]
-    (make-maintenance-mode handler on-error on-error)))
+    (make-maintenance-mode handler on-error)))
 
 (defn wrap-too-many-requests
   "Check if a service has more pending requests than max-queue-length and immediately return a 503"

waiter/src/waiter/service_description.clj

@@ -1551,7 +1551,7 @@
   "Processing the request headers to identify the Waiter service parameters.
    Returns a map of the waiter and passthrough headers, the identified token, and
    the service description template created from the token."
-  [kv-store attach-service-defaults-fn attach-token-defaults-fn waiter-hostnames headers]
+  [kv-store attach-service-defaults-fn attach-token-defaults-fn waiter-hostnames headers unsupported-headers]
   (let [{:keys [passthrough-headers waiter-headers]} (headers/split-headers headers)
         {:keys [token]} (retrieve-token-from-service-description-or-hostname
                           waiter-headers passthrough-headers waiter-hostnames)
@@ -1571,7 +1571,15 @@
         token-metadata (when token
                          (-> (token->token-parameters kv-store token :error-on-missing false)
                            (attach-token-defaults-fn)
-                           (select-keys token-metadata-keys)))]
+                           (select-keys token-metadata-keys)))
+        unsupported-headers-in-request (set/intersection (set (keys waiter-headers)) unsupported-headers)]
+    (when (not-empty unsupported-headers-in-request)
+      (let [error-details {:unsupported-headers-in-request unsupported-headers-in-request}]
+        (log/info "Unsupported waiter headers found" error-details)
+        (throw (ex-info "Unsupported waiter headers found"
+                        {:details error-details
+                         :log-level :info
+                         :status http-400-bad-request}))))
     {:passthrough-headers passthrough-headers
      :service-description-template service-description-template
      :token token

waiter/src/waiter/settings.clj

@@ -72,7 +72,8 @@
                                                   (s/required-key :lingering-request-threshold-ms) schema/positive-int
                                                   (s/required-key :output-buffer-size) schema/positive-int
                                                   (s/required-key :queue-timeout-ms) schema/positive-int
-                                                  (s/required-key :streaming-timeout-ms) schema/positive-int}
+                                                  (s/required-key :streaming-timeout-ms) schema/positive-int
+                                                  (s/required-key :unsupported-headers) #{schema/non-empty-string}}
    (s/required-key :instance-tracker-config) {(s/required-key :instance-failure-handler) (s/constrained
                                                                                            {:kind s/Keyword
                                                                                             s/Keyword schema/require-symbol-factory-fn}
@@ -310,7 +311,8 @@
                                  :lingering-request-threshold-ms 60000 ; 1 minute
                                  :output-buffer-size 4096 ;; 4 KiB
                                  :queue-timeout-ms 300000
-                                 :streaming-timeout-ms 20000}
+                                 :streaming-timeout-ms 20000
+                                 :unsupported-headers #{"x-waiter-maintenance"}}
    :instance-tracker-config {:instance-failure-handler {:kind :default
                                                         :default {:factory-fn 'waiter.instance-tracker/create-instance-failure-event-handler
                                                                   :config {:recent-failed-instance-cache {:threshold 5000

waiter/src/waiter/util/utils.clj

@@ -444,9 +444,9 @@
       ex
       (ex-info (str "Internal error: " (.getMessage ex)) (assoc error-data :status http-500-internal-server-error) ex))))
 
-(defn exception->response
-  "Converts an exception into a ring response."
-  [^Exception ex request]
+(defn exception->response-metadata
+  "Converts an exception into response metadata."
+  [^Exception ex]
   (let [wrapped-ex (wrap-unhandled-exception ex)
         {:keys [friendly-error-message headers log-level message status] :as data} (ex-data wrapped-ex)
         response-msg (if (or message friendly-error-message)
@@ -457,11 +457,16 @@
       :info (log/info (.getMessage wrapped-ex))
       :warn (log/warn wrapped-ex response-msg)
       (log/error wrapped-ex response-msg))
-    (-> {:details data
-         :headers processed-headers
-         :message response-msg
-         :status status}
-        (data->error-response request))))
+    {:details data
+     :headers processed-headers
+     :message response-msg
+     :status status}))
+
+(defn exception->response
+  "Converts an exception into a ring response."
+  [^Exception ex request]
+  (-> (exception->response-metadata ex)
+      (data->error-response request)))
 
 (defmacro log-and-suppress-when-exception-thrown
   "Executes the body inside a try-catch block and suppresses any thrown exceptions."

waiter/src/waiter/websocket.clj

@@ -550,3 +550,13 @@
                             (when (ru/error-response? response)
                               (async/close! out))
                             response)))))
+
+(defn wrap-ws-acceptor-error-handling
+  "wraps a handler and catches any uncaught exceptions and sends an appropriate error response"
+  [handler]
+  (fn wrap-ws-error-handling-fn [{^ServletUpgradeResponse upgrade-response :upgrade-response :as request}]
+    (try
+      (handler request)
+      (catch Exception e
+        (let [{:keys [message status]} (utils/exception->response-metadata e)]
+          (.sendError upgrade-response status message))))))

waiter/test/waiter/auth/authenticator_test.clj

@@ -145,7 +145,7 @@
         handler-response (Object.)
         execute-request (fn execute-request-fn [{:keys [headers] :as in-request}]
                           (let [test-request (->> (sd/discover-service-parameters
-                                                    kv-store attach-service-defaults-fn attach-token-defaults-fn waiter-hostnames headers)
+                                                    kv-store attach-service-defaults-fn attach-token-defaults-fn waiter-hostnames headers #{})
                                                   (assoc in-request :waiter-discovery))
                                 request-handler-argument-atom (atom nil)
                                 test-request-handler (fn request-handler-fn [request]