add data structure and APIs for agents

.github/workflows/prod.yml

@@ -8,6 +8,7 @@ on:
   workflow_dispatch:
     inputs:
       release_version:
+        description: "The version of the release to be deployed."
         required: true
 
 # A workflow run is made up of one or more jobs that can run sequentially or in parallel
@@ -60,7 +61,7 @@ jobs:
           REGISTRY_ALIAS: p7q3h0z2
           IMAGE_TAG: kafkalatest
           IMAGE_TAG2: latest
-          IMAGE_TAG3: 1.42.1_local
+          IMAGE_TAG3: ${{ github.event.inputs.release_version }}_local
         run: |
           docker buildx create --use
           # Build a docker container and push it to DockerHub 
@@ -82,8 +83,21 @@ jobs:
           ECR_REPOSITORY: akto-api-security
           IMAGE_TAG: kafkalatest
           IMAGE_TAG2: latest
+          IMAGE_TAG3: ${{ github.event.inputs.release_version }}_local
         run: |
           docker buildx create --use
           # Build a docker container and push it to DockerHub 
           cd apps/database-abstractor
-          docker buildx build --platform linux/arm64/v8,linux/amd64 -t $ECR_REGISTRY/akto-api-security-database-abstractor:$IMAGE_TAG -t $ECR_REGISTRY/akto-api-security-database-abstractor:$IMAGE_TAG2 . --push
+          docker buildx build --platform linux/arm64/v8,linux/amd64 -t $ECR_REGISTRY/akto-api-security-database-abstractor:$IMAGE_TAG -t $ECR_REGISTRY/akto-api-security-database-abstractor:$IMAGE_TAG2 -t $ECR_REGISTRY/akto-api-security-database-abstractor:$IMAGE_TAG3 . --push
+
+      - name: Send Github release notification to Slack
+        id: slack
+        uses: slackapi/[email protected]
+        with:
+          payload: |
+            {
+              "text": "Database abstractor (codename: cyborg) version v${{ github.event.inputs.release_version }} released!"
+            }
+        env:
+          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
+          SLACK_WEBHOOK_TYPE: INCOMING_WEBHOOK

apps/dashboard/src/main/java/com/akto/action/agents/AgentAction.java

@@ -0,0 +1,262 @@
+package com.akto.action.agents;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.bson.conversions.Bson;
+
+import com.akto.action.UserAction;
+import com.akto.dao.agents.AgentRunDao;
+import com.akto.dao.agents.AgentSubProcessSingleAttemptDao;
+import com.akto.dao.context.Context;
+import com.akto.dto.agents.*;
+import com.mongodb.client.model.Filters;
+import com.mongodb.client.model.Updates;
+import com.opensymphony.xwork2.Action;
+
+public class AgentAction extends UserAction {
+
+    Map<String, Object> data;
+    String agent;
+
+    public String createAgentRun() {
+
+        if (data == null || data.isEmpty()) {
+            addActionError("Agent Init document is empty");
+            return Action.ERROR.toUpperCase();
+        }
+
+        if (agent == null || agent.isEmpty()) {
+            addActionError("Agent is empty");
+            return Action.ERROR.toUpperCase();
+        }
+
+        Agent agentModule;
+        try {
+            agentModule = Agent.valueOf(agent);
+        } catch (Exception e) {
+            addActionError("Invalid agent");
+            return Action.ERROR.toUpperCase();
+        }
+
+        AgentRun existingScheduledOrRunningRuns = AgentRunDao.instance.findOne(Filters.nin(
+                AgentRun._STATE, Arrays.asList(State.SCHEDULED, State.RUNNING)));
+
+        if (existingScheduledOrRunningRuns != null) {
+            addActionError("An existing agent run is running or scheduled. " +
+                    "Let it complete or stop it before starting another");
+            return Action.ERROR.toUpperCase();
+        }
+
+        String processId = UUID.randomUUID().toString();
+
+        agentRun = new AgentRun(processId, data, agentModule,
+                Context.now(), 0, 0, State.SCHEDULED);
+
+        AgentRunDao.instance.insertOne(agentRun);
+
+        return Action.SUCCESS.toUpperCase();
+    }
+
+    Agent[] agents;
+
+    AgentRun agentRun;
+
+    String processId;
+    int subProcessId;
+    int attemptId;
+    String state;
+
+    AgentSubProcessSingleAttempt subprocess;
+
+    public String getAgents() {
+        agents = Agent.values();
+        return Action.SUCCESS.toUpperCase();
+    }
+
+    List<AgentSubProcessSingleAttempt> subProcesses;
+
+    public String getAllSubProcesses() {
+        if (processId == null || processId.isEmpty()) {
+            addActionError("Process id is invalid");
+            return Action.ERROR.toUpperCase();
+        }
+        Bson processIdFilter = Filters.eq(AgentRun.PROCESS_ID, processId);
+
+        agentRun = AgentRunDao.instance.findOne(processIdFilter);
+
+        if (agentRun == null) {
+            addActionError("No process found");
+            return Action.ERROR.toUpperCase();
+        }
+
+        subProcesses = AgentSubProcessSingleAttemptDao.instance.findAll(processIdFilter);
+        return Action.SUCCESS.toUpperCase();
+    }
+
+    public String getSubProcess(){
+
+        if (processId == null || processId.isEmpty()) {
+            addActionError("Process id is invalid");
+            return Action.ERROR.toUpperCase();
+        }
+
+        Bson processIdFilter = Filters.eq(AgentRun.PROCESS_ID, processId);
+
+        agentRun = AgentRunDao.instance.findOne(processIdFilter);
+
+        if (agentRun == null) {
+            addActionError("No process found");
+            return Action.ERROR.toUpperCase();
+        }
+
+        Bson filter = AgentSubProcessSingleAttemptDao.instance.getFiltersForAgentSubProcess(processId, subProcessId, attemptId);
+        subprocess = AgentSubProcessSingleAttemptDao.instance.findOne(filter);
+
+        return Action.SUCCESS.toUpperCase();
+
+    }
+
+    public String updateAgentSubprocess() {
+
+        if (processId == null || processId.isEmpty()) {
+            addActionError("Process id is invalid");
+            return Action.ERROR.toUpperCase();
+        }
+
+        Bson processIdFilter = Filters.eq(AgentRun.PROCESS_ID, processId);
+
+        agentRun = AgentRunDao.instance.findOne(processIdFilter);
+
+        if (agentRun == null) {
+            addActionError("No process found");
+            return Action.ERROR.toUpperCase();
+        }
+
+        Bson filter = AgentSubProcessSingleAttemptDao.instance.getFiltersForAgentSubProcess(processId, subProcessId,
+                attemptId);
+
+        List<Bson> updates = new ArrayList<>();
+
+        updates.add(Updates.setOnInsert(AgentSubProcessSingleAttempt.CREATED_TIMESTAMP, Context.now()));
+        updates.add(Updates.setOnInsert(AgentSubProcessSingleAttempt._STATE, State.SCHEDULED));
+
+        State updatedState = null;
+        try {
+            updatedState = State.valueOf(state);
+        } catch (Exception e) {
+        }
+
+        if (updatedState != null) {
+            AgentSubProcessSingleAttempt subProcess = AgentSubProcessSingleAttemptDao.instance.findOne(filter);
+            if (subProcess == null) {
+                addActionError("No subprocess found");
+                return Action.ERROR.toUpperCase();
+            }
+
+            if (State.COMPLETED.equals(subProcess.getState())
+                    && (State.ACCEPTED.equals(updatedState) || State.DISCARDED.equals(updatedState))) {
+                updates.add(Updates.set(AgentSubProcessSingleAttempt._STATE, updatedState));
+            } else {
+                addActionError("Invalid state");
+                return Action.ERROR.toUpperCase();
+            }
+        }
+
+        /*
+         * For a new subprocess.
+         */
+        if (data != null) {
+            updates.add(Updates.set(AgentSubProcessSingleAttempt.USER_INPUT, data));
+        }
+
+        /*
+         * Upsert: true.
+         * Since state management is through dashboard,
+         * all subprocess' are created here and only modified using the agent-module
+         */
+        subprocess = AgentSubProcessSingleAttemptDao.instance.updateOne(filter, Updates.combine(updates));
+
+        return Action.SUCCESS.toUpperCase();
+    }
+
+    public Map<String, Object> getData() {
+        return data;
+    }
+
+    public void setData(Map<String, Object> data) {
+        this.data = data;
+    }
+
+    public String getAgent() {
+        return agent;
+    }
+
+    public void setAgent(String agent) {
+        this.agent = agent;
+    }
+
+    public void setAgents(Agent[] agents) {
+        this.agents = agents;
+    }
+
+    public AgentRun getAgentRun() {
+        return agentRun;
+    }
+
+    public void setAgentRun(AgentRun agentRun) {
+        this.agentRun = agentRun;
+    }
+
+    public String getProcessId() {
+        return processId;
+    }
+
+    public void setProcessId(String processId) {
+        this.processId = processId;
+    }
+
+    public int getSubProcessId() {
+        return subProcessId;
+    }
+
+    public void setSubProcessId(int subProcessId) {
+        this.subProcessId = subProcessId;
+    }
+
+    public int getAttemptId() {
+        return attemptId;
+    }
+
+    public void setAttemptId(int attemptId) {
+        this.attemptId = attemptId;
+    }
+
+    public String getState() {
+        return state;
+    }
+
+    public void setState(String state) {
+        this.state = state;
+    }
+
+    public AgentSubProcessSingleAttempt getSubprocess() {
+        return subprocess;
+    }
+
+    public void setSubprocess(AgentSubProcessSingleAttempt subprocess) {
+        this.subprocess = subprocess;
+    }
+
+    public List<AgentSubProcessSingleAttempt> getSubProcesses() {
+        return subProcesses;
+    }
+
+    public void setSubProcesses(List<AgentSubProcessSingleAttempt> subProcesses) {
+        this.subProcesses = subProcesses;
+    }
+
+}