Job queue with tests (#14)

* Job queue with tests

* Commented the code; bit of cleanup

* max-total-requests configurable

* Renamed job-status prefix for redis key to validation

* Renamed enqueue-validate-endpoint to enqueue-validation

* Fixed three minor points

* Use middleware for goose; expiry-seconds configurable; start worker in main

* Cleaned up make-status-call

* More fixes

* Fixed middleware

* Cleaner fix for middleware

.github/workflows/test.yml

@@ -28,6 +28,11 @@ jobs:
     - name: Install clj runtime
       run: .github/workflows/install-binaries.sh
 
+    - name: Start Redis
+      uses: supercharge/[email protected]
+      with:
+        redis-version: 6.2
+
     - name: Run tests
       env:
         GATEWAY_URL: https://gateway.test.surfeduhub.nl/

.nvd-suppressions.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<suppressions xmlns="https://jeremylong.github.io/DependencyCheck/dependency-suppression.1.2.xsd"
+              xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+              xsi:schemaLocation="https://jeremylong.github.io/DependencyCheck/dependency-suppression.1.2.xsd
+                                  https://jeremylong.github.io/DependencyCheck/dependency-suppression.1.2.xsd">
+
+    <suppress>
+        <cvssBelow>5.0</cvssBelow>
+        <cve>CVE-2023-46120</cve>
+        <!-- Related to RabbitMQ. We don't use RabbitMQ, nor do we let untrusted users add random jobs. -->
+    </suppress>
+
+</suppressions>

deps.edn

@@ -4,6 +4,8 @@
            ch.qos.logback/logback-classic              {:mvn/version "1.5.8"}
            cheshire/cheshire                           {:mvn/version "5.13.0"}
            com.fasterxml.jackson.core/jackson-databind {:mvn/version "2.17.2"}
+           com.nilenso/goose                           {:mvn/version "0.5.3"}
+           com.taoensso/carmine                        {:mvn/version "3.4.1"}
            compojure/compojure                         {:mvn/version "1.7.1"}
            nl.jomco/clj-http-status-codes              {:mvn/version "0.1"}
            nl.jomco/envopts                            {:mvn/version "0.0.4"}

readme.md

@@ -25,8 +25,10 @@ SURF_CONEXT_CLIENT_ID               SurfCONEXT client id for validation service
 SURF_CONEXT_CLIENT_SECRET           SurfCONEXT client secret for validation service
 SURF_CONEXT_INTROSPECTION_ENDPOINT  SurfCONEXT introspection endpoint
 ALLOWED_CLIENT_IDS                  Comma separated list of allowed SurfCONEXT client ids. 
+MAX_TOTAL_REQUESTS                  Maximum number of requests that validator is allowed to make before raising an error
 OOAPI_VERSION                       Ooapi version to pass through to gateway
 SERVER_PORT                         Starts the app server on this port
+JOB_STATUS_EXPIRY_SECONDS           Number of seconds before job status in Redis expires
 ```
 
 ## Build

src/nl/surf/eduhub/validator/service/api.clj

@@ -0,0 +1,70 @@
+;; This file is part of eduhub-validator-service
+;;
+;; Copyright (C) 2024 SURFnet B.V.
+;;
+;; This program is free software: you can redistribute it and/or
+;; modify it under the terms of the GNU Affero General Public License
+;; as published by the Free Software Foundation, either version 3 of
+;; the License, or (at your option) any later version.
+;;
+;; This program is distributed in the hope that it will be useful, but
+;; WITHOUT ANY WARRANTY; without even the implied warranty of
+;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+;; Affero General Public License for more details.
+;;
+;; You should have received a copy of the GNU Affero General Public
+;; License along with this program.  If not, see
+;; <https://www.gnu.org/licenses/>.
+
+(ns nl.surf.eduhub.validator.service.api
+  (:require [clojure.string :as str]
+            [compojure.core :refer [GET POST defroutes]]
+            [compojure.route :as route]
+            [nl.jomco.http-status-codes :as http-status]
+            [nl.surf.eduhub.validator.service.authentication :as auth]
+            [nl.surf.eduhub.validator.service.checker :as checker]
+            [nl.surf.eduhub.validator.service.jobs.client :as jobs-client]
+            [nl.surf.eduhub.validator.service.jobs.status :as status]
+            [ring.middleware.defaults :refer [api-defaults wrap-defaults]]
+            [ring.middleware.json :refer [wrap-json-response]]))
+
+(defroutes app-routes
+           (GET "/status/:uuid" [uuid]
+             {:load-status true, :uuid uuid})
+           (POST "/endpoints/:endpoint-id/config" [endpoint-id]
+             {:checker true :endpoint-id endpoint-id})
+           (POST "/endpoints/:endpoint-id/paths" [endpoint-id profile]
+             {:validator true :endpoint-id endpoint-id :profile profile})
+           (route/not-found "Not Found"))
+
+;; Many response handlers have the same structure - with this function they can be written inline.
+;; `activate-handler?` is a function that takes a request and returns a boolean which determines if
+;; the current handler should be activated (or skipped).
+;; `response-handler` takes an intermediate response and processes it into the next step.
+(defn wrap-response-handler [app activate-handler? response-handler]
+  (fn [req]
+    (let [resp (app req)]
+      (if (activate-handler? resp)
+        (response-handler resp)
+        resp))))
+
+;; Turn the contents of a job status (stored in redis) into an http response.
+(defn- job-status-handler [{:keys [redis-conn] :as _config}]
+  (fn handle-job-status [resp]
+    (let [job-status (status/load-status redis-conn (:uuid resp))]
+      (if (empty? job-status)
+        {:status http-status/not-found}
+        {:status http-status/ok :body job-status}))))
+
+;; Compose the app from the routes and the wrappers. Authentication can be disabled for testing purposes.
+(defn compose-app [{:keys [introspection-endpoint-url introspection-basic-auth allowed-client-ids] :as config} auth-enabled]
+  (let [allowed-client-id-set  (set (str/split allowed-client-ids #","))
+        auth-opts              {:auth-enabled (boolean auth-enabled)}]
+    (-> app-routes
+        (wrap-response-handler :checker    #(checker/check-endpoint (:endpoint-id %) config))
+        (wrap-response-handler :validator  #(jobs-client/enqueue-validation (:endpoint-id %) (:profile %) config))
+        (wrap-response-handler :load-status (job-status-handler config))
+        (auth/wrap-allowed-clients-checker allowed-client-id-set auth-opts)
+        (auth/wrap-authentication introspection-endpoint-url introspection-basic-auth auth-opts)
+        wrap-json-response
+        (wrap-defaults api-defaults))))

src/nl/surf/eduhub/validator/service/authentication.clj

@@ -110,17 +110,20 @@
 
   If no bearer token is provided, the request is executed without a client-id."
   ; auth looks like {:user client-id :pass client-secret}
-  [f introspection-endpoint auth]
+  [f introspection-endpoint auth {:keys [auth-enabled]}]
   (let [authenticator (memo/ttl (make-token-authenticator introspection-endpoint auth) :ttl/threshold 60000)] ; 1 minute
     (fn authentication [request]
-      (if-let [token (bearer-token request)]
-        (handle-request-with-token request f (authenticator token))
+      (if auth-enabled
+        (if-let [token (bearer-token request)]
+          (handle-request-with-token request f (authenticator token))
+          (f request))
         (f request)))))
 
-(defn wrap-allowed-clients-checker [f allowed-client-id-set]
+(defn wrap-allowed-clients-checker [f allowed-client-id-set {:keys [auth-enabled]}]
   {:pre [(set? allowed-client-id-set)]}
   (fn allowed-clients-checker [{:keys [client-id] :as request}]
-    (if (and client-id (allowed-client-id-set client-id))
+    (if (or (not auth-enabled)
+            (and client-id (allowed-client-id-set client-id)))
       (f request)
       {:body (if client-id "Unknown client id" "No client-id found")
        :status http-status/forbidden})))

src/nl/surf/eduhub/validator/service/checker.clj

@@ -0,0 +1,47 @@
+(ns nl.surf.eduhub.validator.service.checker
+  (:require [babashka.json :as json]
+            [clojure.tools.logging :as log]
+            [nl.jomco.http-status-codes :as http-status]
+            [nl.surf.eduhub.validator.service.validate :as validate]))
+
+(defn- handle-check-endpoint-response [status body endpoint-id]
+  (condp = status
+    ;; If the validator doesn't have the right credentials for the gateway, manifested by a 401 response,
+    ;; we'll return a 502 error and log it.
+    http-status/unauthorized
+    {:status http-status/bad-gateway :body {:valid   false
+                                            :message "Incorrect credentials for gateway"}}
+
+    ;; If the gateway returns OK we assume we've gotten a json envelope response and check the response status
+    ;; of the endpoint.
+    http-status/ok
+    (let [envelope        (json/read-str body)
+          envelope-status (get-in envelope [:gateway :endpoints (keyword endpoint-id) :responseCode])]
+      {:status http-status/ok
+       :body   (if (= "200" (str envelope-status))
+                 {:valid true}
+                 ;; If the endpoint denied the gateway's request or otherwise returned a response outside the 200 range
+                 ;; we return 200 ok and return the status of the endpoint and the message of the gateway in our response
+                 {:valid false
+                  :message (str "Endpoint validation failed with status: " envelope-status)})})
+
+    ;; If the gateway returns something else than a 200 or a 401, treat it similar to an error
+    (let [error-msg (str "Unexpected response status received from gateway: " status)]
+      (log/error error-msg)
+      {:status http-status/internal-server-error
+       :body   {:valid   false
+                :message error-msg}})))
+
+;; The endpoint checker from phase 1. This connects to an endpoint via the gateway and checks if it receives
+;; a valid response.
+(defn check-endpoint [endpoint-id config]
+  (try
+    (let [{:keys [status body]} (validate/check-endpoint endpoint-id config)]
+      ;; If the client credentials for the validator are incorrect, the wrap-allowed-clients-checker
+      ;; middleware has already returned 401 forbidden and execution doesn't get here.
+      (handle-check-endpoint-response status body endpoint-id))
+    (catch Throwable e
+      (log/error e "Internal error in validator-service")
+      {:status http-status/internal-server-error
+       :body   {:valid false
+                :message "Internal error in validator-service"}})))

src/nl/surf/eduhub/validator/service/config.clj

@@ -30,13 +30,22 @@
                                         :in [:gateway-basic-auth :pass]]
    :allowed-client-ids                 ["Comma separated list of allowed SurfCONEXT client ids." :str
                                         :in [:allowed-client-ids]]
+   :max-total-requests                 ["Maximum number of requests that validator is allowed to make before raising an error" :int
+                                        :default 10000
+                                        :in [:max-total-requests]]
    :surf-conext-client-id              ["SurfCONEXT client id for validation service" :str
                                         :in [:introspection-basic-auth :user]]
    :surf-conext-client-secret          ["SurfCONEXT client secret for validation service" :str
                                         :in [:introspection-basic-auth :pass]]
    :surf-conext-introspection-endpoint ["SurfCONEXT introspection endpoint" :str
                                         :in [:introspection-endpoint-url]]
+   :redis-uri                          ["URI to redis" :str
+                                        :default "redis://localhost"
+                                        :in [:redis-conn :spec :uri]]
    :server-port                        ["Starts the app server on this port" :int]
+   :job-status-expiry-seconds          ["Number of seconds before job status in Redis expires" :int
+                                        :default (* 3600 24 14)
+                                        :in [:expiry-seconds]]
    :ooapi-version                      ["Ooapi version to pass through to gateway" :str
                                         :in [:ooapi-version]]})
 
@@ -68,3 +77,13 @@
 (defn load-config-from-env [env-map]
   (-> (reduce file-secret-loader-reducer env-map env-keys-with-alternate-file-secret)
       (envopts/opts opt-specs)))
+
+(defn validate-and-load-config [env]
+  (let [[config errs] (load-config-from-env env)]
+    (when errs
+      (.println *err* "Error in environment configuration")
+      (.println *err* (envopts/errs-description errs))
+      (.println *err* "Available environment vars:")
+      (.println *err* (envopts/specs-description opt-specs))
+      (System/exit 1))
+    config))

src/nl/surf/eduhub/validator/service/jobs/client.clj

@@ -0,0 +1,29 @@
+(ns nl.surf.eduhub.validator.service.jobs.client
+  (:require [clojure.tools.logging :as log]
+            [goose.brokers.redis.broker :as broker]
+            [goose.client :as c]
+            [goose.retry :as retry]
+            [nl.surf.eduhub.validator.service.jobs.status :as status]
+            [nl.surf.eduhub.validator.service.jobs.worker :as worker])
+  (:import [java.util UUID]))
+
+(defn job-error-handler [_cfg _job ex]
+  (log/error ex "Error in job"))
+
+(def client-opts (assoc c/default-opts
+                   :broker (broker/new-producer broker/default-opts)
+                   :retry-opts (assoc retry/default-opts :error-handler-fn-sym `job-error-handler)))
+
+;; Enqueue the validate-endpoint call in the worker queue.
+(defn enqueue-validation
+  [endpoint-id profile {:keys [redis-conn gateway-basic-auth gateway-url ooapi-version max-total-requests] :as _config}]
+  (let [uuid (str (UUID/randomUUID))
+        prof (or profile "rio")
+        opts {:basic-auth         gateway-basic-auth
+              :base-url           gateway-url
+              :max-total-requests max-total-requests
+              :ooapi-version      ooapi-version
+              :profile            prof}]
+    (status/set-status-fields redis-conn uuid "pending" {:endpoint-id endpoint-id, :profile prof} nil)
+    (c/perform-async client-opts `worker/validate-endpoint endpoint-id uuid opts)
+    {:status 200 :body {:job-status "pending" :uuid uuid}}))

src/nl/surf/eduhub/validator/service/jobs/status.clj

@@ -0,0 +1,28 @@
+(ns nl.surf.eduhub.validator.service.jobs.status
+  (:require [taoensso.carmine :as car])
+  (:import [java.time Instant]))
+
+(defn status-key [uuid] (str "validation:" uuid))
+
+(def job-status "job-status")
+
+;; Updates the status of the job status entry of job `uuid` to `new-status`.
+;; Also sets a timestamp named after the new status (e.g. finished-at)
+;; Also sets any values in `key-value-map` in the redis hash.
+;; If `expires-in-seconds` is set, (re)sets the expiry.
+(defn set-status-fields [redis-conn uuid new-status key-value-map expires-in-seconds]
+  (let [v (assoc key-value-map
+            job-status new-status
+            (str new-status "-at") (-> (System/currentTimeMillis)
+                                       Instant/ofEpochMilli
+                                       str))
+        key (status-key uuid)]
+    (car/wcar redis-conn (car/hmset* key v))
+    (when expires-in-seconds
+      (car/wcar redis-conn (car/expire key expires-in-seconds)))))
+
+;; Loads the job status as a clojure map for the job with given uuid.
+(defn load-status [redis-conn uuid]
+  (let [result (car/wcar redis-conn (car/hgetall (status-key uuid)))]
+    (when-not (empty? result)
+      (into {} (apply hash-map result)))))

src/nl/surf/eduhub/validator/service/jobs/worker.clj

@@ -0,0 +1,20 @@
+(ns nl.surf.eduhub.validator.service.jobs.worker
+  (:require [clojure.tools.logging :as log]
+            [nl.surf.eduhub.validator.service.jobs.status :as status]
+            [nl.surf.eduhub.validator.service.validate :as validate]))
+
+;; A worker thread running in the background
+;; Called by the workers. Runs the validate-endpoint function
+;; and updates the values in the job status.
+;; opts should contain: basic-auth ooapi-version base-url profile
+(defn validate-endpoint [endpoint-id uuid {:keys [config] :as opts}]
+  (let [{:keys [redis-conn expiry-seconds]} config]
+    (assert redis-conn)
+    (try
+      (let [html (validate/validate-endpoint endpoint-id opts)]
+        ;; assuming everything went ok, save html in status, update status and set expiry to value configured in ENV
+        (status/set-status-fields redis-conn uuid "finished" {"html-report" html} expiry-seconds))
+      (catch Throwable ex
+        ;; otherwise set status to error, include error message and also set expiry
+        (log/error ex "Validate endpoint threw an exception")
+        (status/set-status-fields redis-conn uuid "failed" {"error" (str ex)} expiry-seconds)))))