Feature/activemq camel stomp

cwms-data-api/build.gradle

@@ -13,7 +13,7 @@ configurations {
 }
 
 configurations.implementation {
-    exclude group: 'com.oracle.database.jdbc'
+    exclude group: 'com.oracle.database.jdbc', module: 'ojdbc'
 }
 
 dependencies {
@@ -114,6 +114,21 @@ dependencies {
 
     implementation(libs.bundles.jackson)
 
+    implementation(libs.aqapi)
+    implementation(libs.jmscommon)
+    //For some reason the caffeine transitive dependency makes gradle angry
+    implementation(libs.activemq.artemis.server) {
+        exclude group: "com.github.ben-manes.caffeine", module: "caffeine"
+    }
+    implementation(libs.activemq.artemis.client) {
+        exclude group: "com.github.ben-manes.caffeine", module: "caffeine"
+    }
+    implementation(libs.activemq.artemis.stomp) {
+        exclude group: "com.github.ben-manes.caffeine", module: "caffeine"
+    }
+    implementation(libs.camel.core)
+    implementation(libs.camel.jms)
+
     testImplementation(libs.bundles.junit)
     testRuntimeOnly(libs.junit.jupiter.engine)
     testImplementation(libs.mockito.core)
@@ -228,6 +243,7 @@ task run(type: JavaExec) {
     mainClass = "fixtures.TomcatServer"
     systemProperties += project.properties.findAll { k, v -> k.startsWith("RADAR") }
     systemProperties += project.properties.findAll { k, v -> k.startsWith("CDA") }
+    systemProperties += project.properties.findAll { k, v -> k.startsWith("cwms") }
 
     def context = project.findProperty("cda.war.context") ?: "spk-data"
 

cwms-data-api/src/main/java/cwms/cda/ApiServlet.java

@@ -25,12 +25,6 @@
 package cwms.cda;
 
 import static cwms.cda.api.Controllers.NAME;
-import cwms.cda.api.DownstreamLocationsGetController;
-import cwms.cda.api.LookupTypeController;
-import cwms.cda.api.StreamController;
-import cwms.cda.api.StreamLocationController;
-import cwms.cda.api.StreamReachController;
-import cwms.cda.api.UpstreamLocationsGetController;
 import static io.javalin.apibuilder.ApiBuilder.crud;
 import static io.javalin.apibuilder.ApiBuilder.delete;
 import static io.javalin.apibuilder.ApiBuilder.get;
@@ -54,6 +48,7 @@
 import cwms.cda.api.ClobController;
 import cwms.cda.api.Controllers;
 import cwms.cda.api.CountyController;
+import cwms.cda.api.DownstreamLocationsGetController;
 import cwms.cda.api.EmbankmentController;
 import cwms.cda.api.ForecastFileController;
 import cwms.cda.api.ForecastInstanceController;
@@ -63,6 +58,7 @@
 import cwms.cda.api.LocationCategoryController;
 import cwms.cda.api.LocationController;
 import cwms.cda.api.LocationGroupController;
+import cwms.cda.api.LookupTypeController;
 import cwms.cda.api.OfficeController;
 import cwms.cda.api.ParametersController;
 import cwms.cda.api.PoolController;
@@ -75,6 +71,9 @@
 import cwms.cda.api.SpecifiedLevelController;
 import cwms.cda.api.StandardTextController;
 import cwms.cda.api.StateController;
+import cwms.cda.api.StreamController;
+import cwms.cda.api.StreamLocationController;
+import cwms.cda.api.StreamReachController;
 import cwms.cda.api.TextTimeSeriesController;
 import cwms.cda.api.TextTimeSeriesValueController;
 import cwms.cda.api.TimeSeriesCategoryController;
@@ -88,6 +87,7 @@
 import cwms.cda.api.TurbineChangesPostController;
 import cwms.cda.api.TurbineController;
 import cwms.cda.api.UnitsController;
+import cwms.cda.api.UpstreamLocationsGetController;
 import cwms.cda.api.auth.ApiKeyController;
 import cwms.cda.api.enums.UnitSystem;
 import cwms.cda.api.errors.AlreadyExists;
@@ -98,6 +98,7 @@
 import cwms.cda.api.errors.JsonFieldsException;
 import cwms.cda.api.errors.NotFoundException;
 import cwms.cda.api.errors.RequiredQueryParameterException;
+import cwms.cda.api.messaging.CdaTopicHandler;
 import cwms.cda.data.dao.JooqDao;
 import cwms.cda.formatters.Formats;
 import cwms.cda.formatters.FormattingException;
@@ -148,6 +149,7 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import javax.sql.DataSource;
+
 import org.apache.http.entity.ContentType;
 import org.jetbrains.annotations.NotNull;
 import org.owasp.html.HtmlPolicyBuilder;
@@ -186,7 +188,7 @@
         "/projects/*",
         "/properties/*",
         "/lookup-types/*",
-        "/embankments/*"
+        "/cda-topics"
 })
 public class ApiServlet extends HttpServlet {
 
@@ -218,12 +220,13 @@ public class ApiServlet extends HttpServlet {
 
     @Resource(name = "jdbc/CWMS3")
     DataSource cwms;
-
+    private CdaTopicHandler cdaTopicHandler;
 
 
     @Override
     public void destroy() {
         javalin.destroy();
+        cdaTopicHandler.shutdown();
     }
 
     @Override
@@ -237,7 +240,7 @@ public void init(ServletConfig config) throws ServletException {
 
     @SuppressWarnings({"java:S125","java:S2095"}) // closed in destroy handler
     @Override
-    public void init() {
+    public void init() throws ServletException {
         JavalinValidation.register(UnitSystem.class, UnitSystem::systemFor);
         JavalinValidation.register(JooqDao.DeleteMethod.class, Controllers::getDeleteMethod);
 
@@ -513,6 +516,12 @@ protected void configureRoutes() {
                 new PropertyController(metrics), requiredRoles,1, TimeUnit.DAYS);
         cdaCrudCache(format("/lookup-types/{%s}", Controllers.NAME),
                 new LookupTypeController(metrics), requiredRoles,1, TimeUnit.DAYS);
+        if(Boolean.getBoolean("cwms.data.api.messaging.enabled")) {
+            //TODO: setup separate data source for persistent connections to Oracle AQ
+            cdaTopicHandler = new CdaTopicHandler(cwms, metrics);
+            get("/cda-topics", cdaTopicHandler);
+            addCacheControl("/cda-topics", 1, TimeUnit.DAYS);
+        }
     }
 
     /**

cwms-data-api/src/main/java/cwms/cda/api/messaging/ArtemisSecurityManager.java

@@ -0,0 +1,83 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 Hydrologic Engineering Center
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package cwms.cda.api.messaging;
+
+import static cwms.cda.ApiServlet.CWMS_USERS_ROLE;
+
+import com.google.common.flogger.FluentLogger;
+import cwms.cda.data.dao.AuthDao;
+import cwms.cda.security.CwmsAuthException;
+import cwms.cda.security.DataApiPrincipal;
+import java.util.Set;
+import javax.sql.DataSource;
+import org.apache.activemq.artemis.core.security.CheckType;
+import org.apache.activemq.artemis.core.security.Role;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
+import org.jooq.SQLDialect;
+import org.jooq.impl.DSL;
+
+final class ArtemisSecurityManager implements ActiveMQSecurityManager {
+    private static final FluentLogger LOGGER = FluentLogger.forEnclosingClass();
+    private final DataSource dataSource;
+    private final String cdaUser;
+
+    ArtemisSecurityManager(DataSource dataSource) {
+        this.dataSource = dataSource;
+        cdaUser = DSL.using(dataSource, SQLDialect.ORACLE18C)
+            .connectionResult(c -> c.getMetaData().getUserName());
+    }
+
+    @Override
+    public boolean validateUser(String user, String password) {
+        return validate(user, password);
+    }
+
+    @Override
+    public boolean validateUserAndRole(String user, String password, Set<Role> roles, CheckType checkType) {
+        //CDA User is allowed to send and manage messages for the invm acceptor.
+        //Other users are not allowed to send messages.
+        if (!cdaUser.equalsIgnoreCase(user) && (checkType == CheckType.SEND || checkType == CheckType.MANAGE)) {
+            LOGGER.atWarning().log("User: " + user
+                + " attempting to access Artemis Server with check type: " + checkType
+                + " Only message consumption is supported.");
+            return false;
+        }
+        return validate(user, password);
+    }
+
+    private boolean validate(String user, String password) {
+        AuthDao instance = AuthDao.getInstance(DSL.using(dataSource, SQLDialect.ORACLE18C));
+        boolean retval = false;
+        try {
+            DataApiPrincipal principal = instance.getByApiKey(password);
+            retval = principal.getName().equalsIgnoreCase(user)
+                && principal.getRoles().contains(new cwms.cda.security.Role(CWMS_USERS_ROLE));
+        } catch (CwmsAuthException ex) {
+            LOGGER.atWarning().withCause(ex).log("Unauthenticated user: " + user
+                + " attempting to access Artemis Server");
+        }
+        return retval;
+    }
+}