Support user consent flow (#1293)

* Introduce user consent flow
* Add permissions to confirmation management
* rename from consent to confirmation
* Reformat code. Remove unused imports. Change and add permission checks when configuring auto-confirmation.
* Do not include null values for DDI confirmation base endpoint
* fix confirmation required checkbox id
* Remove unused import. Fix consume/produce type of new API's.
* Change term processing to proceeding when activating user consent flow
* Align formatting and extend integration test cases for DMF and DDI.
* Extend DMF test cases to consider auto-confirmation
* Refactor action management to fix problem of handling action status updates on closed actions.
* remove unsupported validation
* use new confirmation api for DMF. Extend test cases.,
* Remove unnecessary fields.
* Extend API documentation for DDI and MGMT API.
* adapt ddi api docs adoc file
* Fixed the duplicate migration version for db files
* fix method to support confirmation
* Fixed PR comments
* Addressed PR comments
* Fixed after merge compilation issue
* Fixed after merge compilation issue
* Fix failing tests in MgmtRolloutResourceTest
* Fixed the permissions issue reflected by integration tests
* Added back the missing line of code lost during merge
* Fix the failing test on Jenkins

Signed-off-by: Stanislav Trailov <stanislav.trailov@bosch.io>
Signed-off-by: Dimitar Shterev <dimitar.shterev@bosch.io>
Signed-off-by: Michael Herdt <Michael.Herdt@bosch.io>
Signed-off-by: Shruthi Manavalli Ramanna <shruthimanavalli.ramanna@bosch-si.com>
Co-authored-by: Shruthi Manavalli Ramanna <shruthimanavalli.ramanna@bosch-si.com>

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpConfiguration.java

@@ -14,6 +14,7 @@ import org.eclipse.hawkbit.api.HostnameResolver;
 import org.eclipse.hawkbit.cache.DownloadIdCache;
 import org.eclipse.hawkbit.dmf.amqp.api.AmqpSettings;
 import org.eclipse.hawkbit.repository.ArtifactManagement;
+import org.eclipse.hawkbit.repository.ConfirmationManagement;
 import org.eclipse.hawkbit.repository.ControllerManagement;
 import org.eclipse.hawkbit.repository.DeploymentManagement;
 import org.eclipse.hawkbit.repository.DistributionSetManagement;
@@ -275,9 +276,10 @@ public class AmqpConfiguration {
             final AmqpMessageDispatcherService amqpMessageDispatcherService,
             final ControllerManagement controllerManagement, final EntityFactory entityFactory,
             final SystemSecurityContext systemSecurityContext,
-            final TenantConfigurationManagement tenantConfigurationManagement) {
+            final TenantConfigurationManagement tenantConfigurationManagement,
+            final ConfirmationManagement confirmationManagement) {
         return new AmqpMessageHandlerService(rabbitTemplate, amqpMessageDispatcherService, controllerManagement,
-                entityFactory, systemSecurityContext, tenantConfigurationManagement);
+                entityFactory, systemSecurityContext, tenantConfigurationManagement, confirmationManagement);
     }
 
     /**

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherService.java

@@ -36,6 +36,7 @@ import org.eclipse.hawkbit.dmf.json.model.DmfActionRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfArtifact;
 import org.eclipse.hawkbit.dmf.json.model.DmfArtifactHash;
 import org.eclipse.hawkbit.dmf.json.model.DmfBatchDownloadAndUpdateRequest;
+import org.eclipse.hawkbit.dmf.json.model.DmfConfirmRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfMetadata;
 import org.eclipse.hawkbit.dmf.json.model.DmfMultiActionRequest;
@@ -273,6 +274,8 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
             final Map<SoftwareModule, List<SoftwareModuleMetadata>> softwareModules) {
         if (action.isCancelingOrCanceled()) {
             return createPlainActionRequest(action);
+        } else if (action.isWaitingConfirmation()) {
+            return createConfirmRequest(target, action.getId(), softwareModules);
         }
         return createDownloadAndUpdateRequest(target, action.getId(), softwareModules);
     }
@@ -310,6 +313,9 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
      * @return {@link EventTopic} to use for message.
      */
     private static EventTopic getEventTypeForTarget(final ActionProperties action) {
+        if (action.isWaitingConfirmation()) {
+            return EventTopic.CONFIRM;
+        }
         return (Action.ActionType.DOWNLOAD_ONLY == action.getActionType() || !action.isMaintenanceWindowAvailable())
                 ? EventTopic.DOWNLOAD
                 : EventTopic.DOWNLOAD_AND_INSTALL;
@@ -415,9 +421,16 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
             return;
         }
 
-        final DmfDownloadAndUpdateRequest downloadAndUpdateRequest = createDownloadAndUpdateRequest(target,
-                action.getId(), modules);
-        final Message message = getMessageConverter().toMessage(downloadAndUpdateRequest,
+        DmfActionRequest request;
+        if (action.isWaitingConfirmation()) {
+            // For the moment the confirmation request is the same as download and update request.
+            // It can be modified not to expose all the software modules in the future.
+            request = createConfirmRequest(target, action.getId(), modules);
+        } else {
+            request = createDownloadAndUpdateRequest(target, action.getId(), modules);
+        }
+
+        final Message message = getMessageConverter().toMessage(request,
                 createConnectorMessagePropertiesEvent(tenant, target.getControllerId(), getEventTypeForTarget(action)));
         amqpSenderService.sendMessage(message, targetAddress);
     }
@@ -624,4 +637,25 @@ public class AmqpMessageDispatcherService extends BaseAmqpService {
                 ? EventTopic.BATCH_DOWNLOAD
                 : EventTopic.BATCH_DOWNLOAD_AND_INSTALL;
     }
+
+    /**
+     * Creates a Confirmation request.
+     * @param target the target
+     * @param actionId the actionId
+     * @param softwareModules the software modules
+     * @return
+     */
+    protected DmfConfirmRequest createConfirmRequest(final Target target, final Long actionId, final Map<SoftwareModule,
+            List<SoftwareModuleMetadata>> softwareModules) {
+        final DmfConfirmRequest request = new DmfConfirmRequest();
+        request.setActionId(actionId);
+        request.setTargetSecurityToken(systemSecurityContext.runAsSystem(target::getSecurityToken));
+
+        //Software modules can be filtered in the future exposing only the needed.
+        if (softwareModules != null) {
+            softwareModules.entrySet()
+                    .forEach(entry -> request.addSoftwareModule(convertToAmqpSoftwareModule(target, entry)));
+        }
+        return request;
+    }
 }

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerService.java

@@ -24,12 +24,15 @@ import java.util.stream.Collectors;
 import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
 import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
 import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
+import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionUpdateStatus;
 import org.eclipse.hawkbit.dmf.json.model.DmfAttributeUpdate;
+import org.eclipse.hawkbit.dmf.json.model.DmfAutoConfirmation;
 import org.eclipse.hawkbit.dmf.json.model.DmfCreateThing;
 import org.eclipse.hawkbit.dmf.json.model.DmfUpdateMode;
 import org.eclipse.hawkbit.im.authentication.SpPermission.SpringEvalExpressions;
 import org.eclipse.hawkbit.im.authentication.TenantAwareAuthenticationDetails;
+import org.eclipse.hawkbit.repository.ConfirmationManagement;
 import org.eclipse.hawkbit.repository.ControllerManagement;
 import org.eclipse.hawkbit.repository.EntityFactory;
 import org.eclipse.hawkbit.repository.RepositoryConstants;
@@ -75,6 +78,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
     private final AmqpMessageDispatcherService amqpMessageDispatcherService;
 
     private ControllerManagement controllerManagement;
+    private ConfirmationManagement confirmationManagement;
 
     private final EntityFactory entityFactory;
 
@@ -101,18 +105,21 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
      *            the system Security Context
      * @param tenantConfigurationManagement
      *            the tenant configuration Management
+     * @param confirmationManagement
+     *            the confirmation management
      */
     public AmqpMessageHandlerService(final RabbitTemplate rabbitTemplate,
             final AmqpMessageDispatcherService amqpMessageDispatcherService,
             final ControllerManagement controllerManagement, final EntityFactory entityFactory,
             final SystemSecurityContext systemSecurityContext,
-            final TenantConfigurationManagement tenantConfigurationManagement) {
+            final TenantConfigurationManagement tenantConfigurationManagement, final ConfirmationManagement confirmationManagement) {
         super(rabbitTemplate);
         this.amqpMessageDispatcherService = amqpMessageDispatcherService;
         this.controllerManagement = controllerManagement;
         this.entityFactory = entityFactory;
         this.systemSecurityContext = systemSecurityContext;
         this.tenantConfigurationManagement = tenantConfigurationManagement;
+        this.confirmationManagement = confirmationManagement;
     }
 
     /**
@@ -314,6 +321,9 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
         case UPDATE_ATTRIBUTES:
             updateAttributes(message);
             break;
+        case UPDATE_AUTO_CONFIRM:
+            setAutoConfirmationState(message);
+            break;
         default:
             logAndThrowMessageError(message, "Got event without appropriate topic.");
             break;
@@ -331,7 +341,23 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
         final String thingId = getStringHeaderKey(message, MessageHeaderKey.THING_ID, THING_ID_NULL);
 
         controllerManagement.updateControllerAttributes(thingId, attributeUpdate.getAttributes(),
-                getUpdateMode(attributeUpdate));
+              getUpdateMode(attributeUpdate));
+    }
+
+    private void setAutoConfirmationState(final Message message) {
+        final DmfAutoConfirmation autoConfirmation = convertMessage(message, DmfAutoConfirmation.class);
+        final String thingId = getStringHeaderKey(message, MessageHeaderKey.THING_ID, THING_ID_NULL);
+        if (autoConfirmation.isEnabled()) {
+            LOG.debug("Activate auto-confirmation for device {} using DMF. Initiator: {}. Remark: {}", thingId,
+                    autoConfirmation.getInitiator(), autoConfirmation.getRemark());
+            final String remark = autoConfirmation.getRemark() == null
+                    ? "Activated using Device Management Federation API."
+                    : autoConfirmation.getRemark();
+            controllerManagement.activateAutoConfirmation(thingId, autoConfirmation.getInitiator(), remark);
+        } else {
+            LOG.debug("Deactivate auto-confirmation for device {} using DMF.", thingId);
+            controllerManagement.deactivateAutoConfirmation(thingId);
+        }
     }
 
     /**
@@ -352,19 +378,27 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
         }
 
         final Status status = mapStatus(message, actionUpdateStatus, action);
-        final ActionStatusCreate actionStatus = entityFactory.actionStatus().create(action.getId()).status(status)
-                .messages(messages);
 
-        actionUpdateStatus.getCode().ifPresent(code -> {
-            actionStatus.code(code);
-            actionStatus.message("Device reported status code: " + code);
-        });
+        final Action updatedAction;
 
-        final Action updatedAction = (Status.CANCELED == status)
-                ? controllerManagement.addCancelActionStatus(actionStatus)
-                : controllerManagement.addUpdateActionStatus(actionStatus);
+        if (actionUpdateStatus.getActionStatus() == DmfActionStatus.CONFIRMED) {
+            updatedAction = confirmationManagement.confirmAction(action.getId(),
+                    actionUpdateStatus.getCode().orElse(null), messages);
+        } else if (actionUpdateStatus.getActionStatus() == DmfActionStatus.DENIED) {
+            updatedAction = confirmationManagement.denyAction(action.getId(), actionUpdateStatus.getCode().orElse(null),
+                    messages);
+        } else {
+            final ActionStatusCreate actionStatus = entityFactory.actionStatus().create(action.getId()).status(status)
+                    .messages(messages);
+            actionUpdateStatus.getCode().ifPresent(code -> {
+                actionStatus.code(code);
+                actionStatus.message("Device reported status code: " + code);
+            });
+            updatedAction = (Status.CANCELED == status) ? controllerManagement.addCancelActionStatus(actionStatus)
+                    : controllerManagement.addUpdateActionStatus(actionStatus);
+        }
 
-        if (shouldTargetProceed(updatedAction)) {
+        if (shouldTargetProceed(updatedAction) || actionUpdateStatus.getActionStatus() == DmfActionStatus.CONFIRMED) {
             sendUpdateCommandToTarget(action.getTarget());
         }
     }
@@ -391,6 +425,7 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
             status = Status.RETRIEVED;
             break;
         case RUNNING:
+        case CONFIRMED:
             status = Status.RUNNING;
             break;
         case CANCELED:
@@ -411,6 +446,9 @@ public class AmqpMessageHandlerService extends BaseAmqpService {
         case CANCEL_REJECTED:
             status = handleCancelRejectedState(message, action);
             break;
+        case DENIED:
+            status = Status.WAIT_FOR_CONFIRMATION;
+            break;
         default:
             logAndThrowMessageError(message, "Status for action does not exisit.");
         }

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpControllerAuthenticationTest.java

@@ -27,6 +27,7 @@ import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
 import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
 import org.eclipse.hawkbit.dmf.json.model.DmfDownloadResponse;
 import org.eclipse.hawkbit.repository.ArtifactManagement;
+import org.eclipse.hawkbit.repository.ConfirmationManagement;
 import org.eclipse.hawkbit.repository.ControllerManagement;
 import org.eclipse.hawkbit.repository.SystemManagement;
 import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
@@ -117,6 +118,9 @@ public class AmqpControllerAuthenticationTest {
     @Mock
     private ControllerManagement controllerManagement;
 
+    @Mock
+    private ConfirmationManagement confirmationManagement;
+
     @Mock
     private DdiSecurityProperties securityProperties;
 
@@ -156,7 +160,7 @@ public class AmqpControllerAuthenticationTest {
 
         amqpMessageHandlerService = new AmqpMessageHandlerService(rabbitTemplate,
                 mock(AmqpMessageDispatcherService.class), controllerManagement, new JpaEntityFactory(),
-                systemSecurityContext, tenantConfigurationManagementMock);
+                systemSecurityContext, tenantConfigurationManagementMock, confirmationManagement);
 
         amqpAuthenticationMessageHandlerService = new AmqpAuthenticationMessageHandler(rabbitTemplate,
                 authenticationManager, artifactManagementMock, cacheMock, hostnameResolverMock, controllerManagement,

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerServiceTest.java

@@ -36,16 +36,19 @@ import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionUpdateStatus;
 import org.eclipse.hawkbit.dmf.json.model.DmfAttributeUpdate;
+import org.eclipse.hawkbit.dmf.json.model.DmfAutoConfirmation;
 import org.eclipse.hawkbit.dmf.json.model.DmfCreateThing;
 import org.eclipse.hawkbit.dmf.json.model.DmfDownloadResponse;
 import org.eclipse.hawkbit.dmf.json.model.DmfUpdateMode;
 import org.eclipse.hawkbit.repository.ArtifactManagement;
+import org.eclipse.hawkbit.repository.ConfirmationManagement;
 import org.eclipse.hawkbit.repository.ControllerManagement;
 import org.eclipse.hawkbit.repository.EntityFactory;
 import org.eclipse.hawkbit.repository.TenantConfigurationManagement;
 import org.eclipse.hawkbit.repository.UpdateMode;
 import org.eclipse.hawkbit.repository.builder.ActionStatusBuilder;
 import org.eclipse.hawkbit.repository.builder.ActionStatusCreate;
+import org.eclipse.hawkbit.repository.jpa.model.JpaAction;
 import org.eclipse.hawkbit.repository.jpa.builder.JpaActionStatusBuilder;
 import org.eclipse.hawkbit.repository.jpa.model.JpaActionStatus;
 import org.eclipse.hawkbit.repository.jpa.model.helper.SecurityTokenGeneratorHolder;
@@ -110,6 +113,9 @@ public class AmqpMessageHandlerServiceTest {
     @Mock
     private ControllerManagement controllerManagementMock;
 
+    @Mock
+    private ConfirmationManagement confirmationManagementMock;
+
     @Mock
     private EntityFactory entityFactoryMock;
 
@@ -146,6 +152,12 @@ public class AmqpMessageHandlerServiceTest {
     @Captor
     private ArgumentCaptor<String> targetIdCaptor;
 
+    @Captor
+    private ArgumentCaptor<String> initiatorCaptor;
+
+    @Captor
+    private ArgumentCaptor<String> remarkCaptor;
+
     @Captor
     private ArgumentCaptor<String> targetNameCaptor;
 
@@ -169,7 +181,8 @@ public class AmqpMessageHandlerServiceTest {
         final SystemSecurityContext systemSecurityContext = new SystemSecurityContext(tenantAware);
 
         amqpMessageHandlerService = new AmqpMessageHandlerService(rabbitTemplate, amqpMessageDispatcherServiceMock,
-                controllerManagementMock, entityFactoryMock, systemSecurityContext, tenantConfigurationManagement);
+                controllerManagementMock, entityFactoryMock, systemSecurityContext, tenantConfigurationManagement,
+                confirmationManagementMock);
         amqpAuthenticationMessageHandlerService = new AmqpAuthenticationMessageHandler(rabbitTemplate,
                 authenticationManagerMock, artifactManagementMock, downloadIdCache, hostnameResolverMock,
                 controllerManagementMock, tenantAwareMock);
@@ -230,6 +243,16 @@ public class AmqpMessageHandlerServiceTest {
         assertThat(uriCaptor.getValue()).as("Uri is not right").hasToString("amqp://" + VIRTUAL_HOST + "/" + replyTo);
     }
 
+    @Step
+    private void assertInitiatorCapturedField(final String initiator) {
+        assertThat(initiatorCaptor.getValue()).as("Initiator is wrong").isEqualTo(initiator);
+    }
+
+    @Step
+    private void assertRemarkCapturedField(final String remark) {
+        assertThat(remarkCaptor.getValue()).as("Remark is wrong").isEqualTo(remark);
+    }
+
     @Test
     @Description("Tests the creation of a target/thing with specified name by calling the same method that incoming RabbitMQ messages would access.")
     public void createThingWithName() {
@@ -556,6 +579,7 @@ public class AmqpMessageHandlerServiceTest {
         final ActionStatusCreate create = mock(ActionStatusCreate.class);
         when(builder.create(22L)).thenReturn(create);
         when(create.status(any())).thenReturn(create);
+        when(create.messages(any())).thenReturn(create);
         when(entityFactoryMock.actionStatus()).thenReturn(builder);
         // for the test the same action can be used
         when(controllerManagementMock.findActiveActionWithHighestWeight(any())).thenReturn(Optional.of(action));
@@ -572,7 +596,7 @@ public class AmqpMessageHandlerServiceTest {
         final ArgumentCaptor<Target> targetCaptor = ArgumentCaptor.forClass(Target.class);
 
         verify(amqpMessageDispatcherServiceMock, times(1)).sendUpdateMessageToTarget(actionPropertiesCaptor.capture(),
-              targetCaptor.capture(), any(Map.class));
+                targetCaptor.capture(), any(Map.class));
         final ActionProperties actionProperties = actionPropertiesCaptor.getValue();
         assertThat(actionProperties).isNotNull();
         assertThat(actionProperties.getTenant()).as("event has tenant").isEqualTo("DEFAULT");
@@ -616,6 +640,56 @@ public class AmqpMessageHandlerServiceTest {
                 .contains("Device reported status code: 12");
     }
 
+    @Test
+    @Description("Tests activating auto-confirmation on a target.")
+    void setAutoConfirmationStateActive() {
+        final String knownThingId = "1";
+        final String initiator = "iAmTheInitiator";
+        final String remark = "remarkForTesting";
+
+        final MessageProperties messageProperties = createMessageProperties(MessageType.EVENT);
+        messageProperties.setHeader(MessageHeaderKey.THING_ID, knownThingId);
+        messageProperties.setHeader(MessageHeaderKey.TOPIC, "UPDATE_AUTO_CONFIRM");
+        final DmfAutoConfirmation autoConfirmation = new DmfAutoConfirmation();
+        autoConfirmation.setEnabled(true);
+        autoConfirmation.setInitiator(initiator);
+        autoConfirmation.setRemark(remark);
+
+        final Message message = createMessage(autoConfirmation, messageProperties);
+
+        when(controllerManagementMock.activateAutoConfirmation(targetIdCaptor.capture(), initiatorCaptor.capture(),
+              remarkCaptor.capture())).thenReturn(null);
+
+        amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, VIRTUAL_HOST);
+
+        verify(controllerManagementMock, times(0)).deactivateAutoConfirmation(anyString());
+
+        assertThingIdCapturedField(knownThingId);
+        assertInitiatorCapturedField(initiator);
+        assertRemarkCapturedField(remark);
+    }
+
+
+    @Test
+    @Description("Tests deactivating auto-confirmation on a target.")
+    void setAutoConfirmationStateDeactivated() {
+        final String knownThingId = "1";
+
+        final MessageProperties messageProperties = createMessageProperties(MessageType.EVENT);
+        messageProperties.setHeader(MessageHeaderKey.THING_ID, knownThingId);
+        messageProperties.setHeader(MessageHeaderKey.TOPIC, "UPDATE_AUTO_CONFIRM");
+        final DmfAutoConfirmation autoConfirmation = new DmfAutoConfirmation();
+
+        final Message message = createMessage(autoConfirmation, messageProperties);
+
+        amqpMessageHandlerService.onMessage(message, MessageType.EVENT.name(), TENANT, VIRTUAL_HOST);
+
+        verify(controllerManagementMock).deactivateAutoConfirmation(targetIdCaptor.capture());
+        verify(controllerManagementMock, times(0)).activateAutoConfirmation(anyString(), anyString(), anyString());
+
+        assertThingIdCapturedField(knownThingId);
+    }
+
     private DmfActionUpdateStatus createActionUpdateStatus(final DmfActionStatus status) {
         return createActionUpdateStatus(status, 2L);
     }

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AbstractAmqpServiceIntegrationTest.java

@@ -12,7 +12,6 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -31,6 +30,7 @@ import org.eclipse.hawkbit.dmf.json.model.DmfActionRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionUpdateStatus;
 import org.eclipse.hawkbit.dmf.json.model.DmfAttributeUpdate;
+import org.eclipse.hawkbit.dmf.json.model.DmfConfirmRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfCreateThing;
 import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfMetadata;
@@ -129,7 +129,13 @@ public abstract class AbstractAmqpServiceIntegrationTest extends AbstractAmqpInt
         testdataFactory.addSoftwareModuleMetadata(distributionSet);
 
         return registerTargetAndAssignDistributionSet(distributionSet.getId(), TargetUpdateStatus.REGISTERED,
-                distributionSet.getModules(), controllerId);
+              distributionSet.getModules(), controllerId);
+    }
+
+    protected DistributionSetAssignmentResult prepareDistributionSetAndAssign(final String controllerId) {
+        distributionSet = testdataFactory.createDistributionSet(UUID.randomUUID().toString());
+        testdataFactory.addSoftwareModuleMetadata(distributionSet);
+        return assignDistributionSet(distributionSet.getId(), controllerId);
     }
 
     protected DistributionSetAssignmentResult registerTargetAndAssignDistributionSet(final Long assignDs,
@@ -138,7 +144,11 @@ public abstract class AbstractAmqpServiceIntegrationTest extends AbstractAmqpInt
         registerAndAssertTargetWithExistingTenant(controllerId, 1, expectedStatus, CREATED_BY);
 
         final DistributionSetAssignmentResult assignmentResult = assignDistributionSet(assignDs, controllerId);
-        assertDownloadAndInstallMessage(expectedSoftwareModulesInMessage, controllerId);
+        if (isConfirmationFlowEnabled()) {
+            assertConfirmMessage(expectedSoftwareModulesInMessage, controllerId);
+        } else {
+            assertDownloadAndInstallMessage(expectedSoftwareModulesInMessage, controllerId);
+        }
         return assignmentResult;
     }
 
@@ -452,4 +462,24 @@ public abstract class AbstractAmqpServiceIntegrationTest extends AbstractAmqpInt
         softwareModules.forEach(dmfModule -> assertThat(dmfModule.getMetadata()).containsExactly(
                 new DmfMetadata(TestdataFactory.VISIBLE_SM_MD_KEY, TestdataFactory.VISIBLE_SM_MD_VALUE)));
     }
+
+    protected void assertConfirmMessage(final Set<SoftwareModule> dsModules, final String controllerId) {
+
+        final Message replyMessage = assertReplyMessageHeader(EventTopic.CONFIRM, controllerId);
+        assertAllTargetsCount(1);
+
+        final DmfConfirmRequest confirmRequest = (DmfConfirmRequest) getDmfClient()
+                .getMessageConverter().fromMessage(replyMessage);
+
+        assertConfirmRequest(confirmRequest, dsModules, controllerId);
+    }
+
+    protected void assertConfirmRequest(final DmfConfirmRequest request, final Set<SoftwareModule> softwareModules,
+            final String controllerId) {
+        assertSoftwareModules(softwareModules, request.getSoftwareModules());
+        final Target updatedTarget = waitUntilIsPresent(() -> targetManagement.getByControllerID(controllerId));
+        assertThat(updatedTarget).isNotNull();
+        assertThat(updatedTarget.getSecurityToken()).isEqualTo(request.getTargetSecurityToken());
+    }
+
 }

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageDispatcherServiceIntegrationTest.java

@@ -466,7 +466,7 @@ public class AmqpMessageDispatcherServiceIntegrationTest extends AbstractAmqpSer
 
     private Rollout createAndStartRollout(final DistributionSet ds, final String filterQuery, final Integer weight) {
         final Rollout rollout = testdataFactory.createRolloutByVariables(UUID.randomUUID().toString(), "", 1,
-                filterQuery, ds, "50", "5", ActionType.FORCED, weight);
+                filterQuery, ds, "50", "5", ActionType.FORCED, weight, false);
         rolloutManagement.start(rollout.getId());
         rolloutManagement.handleRollouts();
         return rollout;
@@ -695,4 +695,28 @@ public class AmqpMessageDispatcherServiceIntegrationTest extends AbstractAmqpSer
                     assertThat(tokens.contains(requestTarget.getTargetSecurityToken()));
                 });
     }
+
+    @Test
+    @Description("Verify that a distribution assignment send a confirm message.")
+    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
+            @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+            @Expect(type = ActionCreatedEvent.class, count = 1),
+            @Expect(type = SoftwareModuleCreatedEvent.class, count = 3),
+            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+            @Expect(type = DistributionSetCreatedEvent.class, count = 1),
+            @Expect(type = TargetUpdatedEvent.class, count = 1),
+            @Expect(type = TargetPollEvent.class, count = 1),
+            @Expect(type = TenantConfigurationCreatedEvent.class, count = 1) })
+    void sendConfirmStatus() {
+        final String controllerId = TARGET_PREFIX + "sendConfirmStatus";
+        enableConfirmationFlow();
+        registerTargetAndAssignDistributionSet(controllerId);
+
+        waitUntilTargetHasStatus(controllerId, TargetUpdateStatus.PENDING);
+        assertConfirmMessage(getDistributionSet().getModules(), controllerId);
+        assertEventMessageNotPresent(EventTopic.DOWNLOAD_AND_INSTALL);
+    }
+    protected void assertEventMessageNotPresent(final EventTopic eventTopic) {
+        assertThat(replyToListener.getLatestEventMessage(eventTopic)).isNull();
+    }
 }

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/integration/AmqpMessageHandlerServiceIntegrationTest.java

@@ -47,6 +47,8 @@ import org.eclipse.hawkbit.repository.event.remote.entity.SoftwareModuleCreatedE
 import org.eclipse.hawkbit.repository.event.remote.entity.SoftwareModuleUpdatedEvent;
 import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent;
 import org.eclipse.hawkbit.repository.event.remote.entity.TargetUpdatedEvent;
+import org.eclipse.hawkbit.repository.event.remote.entity.TenantConfigurationCreatedEvent;
+import org.eclipse.hawkbit.repository.event.remote.entity.TenantConfigurationUpdatedEvent;
 import org.eclipse.hawkbit.repository.exception.EntityAlreadyExistsException;
 import org.eclipse.hawkbit.repository.jpa.model.JpaDistributionSet;
 import org.eclipse.hawkbit.repository.jpa.model.JpaTarget;
@@ -966,6 +968,26 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr
         getDmfClient().send(updateMessage);
     }
 
+    private void assertActionStatusList(final Long actionId, final int statusListCount,
+            final Status... expectedActionStates) {
+        createConditionFactory().await().untilAsserted(() -> {
+            try {
+                WithSpringAuthorityRule.runAsPrivileged(() -> {
+                    final List<ActionStatus> actionStatusList = deploymentManagement
+                            .findActionStatusByAction(PAGE, actionId).getContent();
+                    assertThat(actionStatusList).hasSize(statusListCount);
+
+                    final List<Status> status = actionStatusList.stream().map(ActionStatus::getStatus)
+                            .collect(Collectors.toList());
+                    assertThat(status).containsOnly(expectedActionStates);
+
+                    return null;
+                });
+            } catch (final Exception e) {
+                throw new RuntimeException(e);
+            }
+        });
+    }
     private int getAuthenticationMessageCount() {
         return Integer
                 .parseInt(Objects.requireNonNull(getRabbitAdmin().getQueueProperties(amqpProperties.getReceiverQueue()))
@@ -993,4 +1015,142 @@ class AmqpMessageHandlerServiceIntegrationTest extends AbstractAmqpServiceIntegr
         assertThat(node.has("actionId")).isTrue();
         return node.get("actionId").asText();
     }
+
+    @Test
+    @Description("Register a target and send a update action status (confirmed). Verify if the updated action status is correct.")
+    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
+            @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+            @Expect(type = ActionUpdatedEvent.class, count = 1),
+            @Expect(type = ActionCreatedEvent.class, count = 1),
+            @Expect(type = SoftwareModuleCreatedEvent.class, count = 3),
+            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+            @Expect(type = DistributionSetCreatedEvent.class, count = 1),
+            @Expect(type = TargetUpdatedEvent.class, count = 1),
+            @Expect(type = TargetPollEvent.class, count = 1),
+            @Expect(type = TenantConfigurationCreatedEvent.class, count = 1) })
+    void confirmedActionStatus() {
+        enableConfirmationFlow();
+        final String controllerId = TARGET_PREFIX + "confirmedActionStatus";
+
+        final DistributionSetAssignmentResult assignmentResult = registerTargetAndAssignDistributionSet(controllerId);
+        final Long actionId = getFirstAssignedActionId(assignmentResult);
+        createAndSendActionStatusUpdateMessage(controllerId, actionId, DmfActionStatus.CONFIRMED);
+        assertActionStatusList(actionId, 2, Status.WAIT_FOR_CONFIRMATION, Status.RUNNING);
+
+        // assert download and install message
+        waitUntilEventMessagesAreDispatchedToTarget(EventTopic.CONFIRM, EventTopic.DOWNLOAD_AND_INSTALL);
+        assertDownloadAndInstallMessage(assignmentResult.getDistributionSet().getModules(), controllerId);
+    }
+
+    @Test
+    @Description("Verify the DMF confirmed feedback can be provided if confirmation flow is disabled")
+    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
+            @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+            @Expect(type = ActionUpdatedEvent.class, count = 1),
+            @Expect(type = ActionCreatedEvent.class, count = 1),
+            @Expect(type = SoftwareModuleCreatedEvent.class, count = 3),
+            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+            @Expect(type = DistributionSetCreatedEvent.class, count = 1),
+            @Expect(type = TargetUpdatedEvent.class, count = 1),
+            @Expect(type = TargetPollEvent.class, count = 1),
+            @Expect(type = TenantConfigurationCreatedEvent.class, count = 1),
+            @Expect(type = TenantConfigurationUpdatedEvent.class, count = 1) })
+    void verifyActionCanBeConfirmedOnDisabledConfirmationFlow() {
+        enableConfirmationFlow();
+        final String controllerId = TARGET_PREFIX + "confirmedActionStatus";
+
+        final DistributionSetAssignmentResult assignmentResult = registerTargetAndAssignDistributionSet(controllerId);
+        final Long actionId = getFirstAssignedActionId(assignmentResult);
+        // verify action status is in WAIT_FOR_CONFIRMATION
+        assertActionStatusList(actionId, 1, Status.WAIT_FOR_CONFIRMATION);
+
+        disableConfirmationFlow();
+
+        createAndSendActionStatusUpdateMessage(controllerId, actionId, DmfActionStatus.CONFIRMED);
+        assertActionStatusList(actionId, 2, Status.WAIT_FOR_CONFIRMATION, Status.RUNNING);
+
+        // assert download and install message
+        waitUntilEventMessagesAreDispatchedToTarget(EventTopic.CONFIRM, EventTopic.DOWNLOAD_AND_INSTALL);
+        assertDownloadAndInstallMessage(assignmentResult.getDistributionSet().getModules(), controllerId);
+    }
+
+
+    @Test
+    @Description("Verify the DMF confirmed feedback can be provided if confirmation flow is disabled")
+    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
+          @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+          @Expect(type = ActionUpdatedEvent.class, count = 0),
+          @Expect(type = ActionCreatedEvent.class, count = 1),
+          @Expect(type = SoftwareModuleCreatedEvent.class, count = 3),
+          @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+          @Expect(type = DistributionSetCreatedEvent.class, count = 1),
+          @Expect(type = TargetUpdatedEvent.class, count = 1),
+          @Expect(type = TargetPollEvent.class, count = 1),
+          @Expect(type = TenantConfigurationCreatedEvent.class, count = 1),
+          @Expect(type = TenantConfigurationUpdatedEvent.class, count = 1)})
+    void verifyActionCanBeDeniedOnDisabledConfirmationFlow() {
+        enableConfirmationFlow();
+        final String controllerId = TARGET_PREFIX + "confirmedActionStatus";
+
+        final DistributionSetAssignmentResult assignmentResult = registerTargetAndAssignDistributionSet(controllerId);
+        final Long actionId = getFirstAssignedActionId(assignmentResult);
+        // verify action status is in WAIT_FOR_CONFIRMATION
+        assertActionStatusList(actionId, 1, Status.WAIT_FOR_CONFIRMATION);
+
+        disableConfirmationFlow();
+
+        createAndSendActionStatusUpdateMessage(controllerId, actionId, DmfActionStatus.DENIED);
+        assertActionStatusList(actionId, 2, Status.WAIT_FOR_CONFIRMATION);
+    }
+
+    @Test
+    @Description("Verify the DMF download and install message is send directly if auto-confirmation is active")
+    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
+            @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+            @Expect(type = ActionUpdatedEvent.class, count = 0),
+            @Expect(type = ActionCreatedEvent.class, count = 1),
+            @Expect(type = SoftwareModuleCreatedEvent.class, count = 3),
+            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+            @Expect(type = DistributionSetCreatedEvent.class, count = 1),
+            @Expect(type = TargetUpdatedEvent.class, count = 2),
+            @Expect(type = TargetPollEvent.class, count = 1),
+            @Expect(type = TenantConfigurationCreatedEvent.class, count = 1) })
+    void verifyDownloadAndInstallDirectlySendOnAutoConfirmationEnabled() {
+        enableConfirmationFlow();
+        final String controllerId = TARGET_PREFIX + "confirmedActionStatus";
+
+        registerAndAssertTargetWithExistingTenant(controllerId);
+
+        confirmationManagement.activateAutoConfirmation(controllerId, null, null);
+
+        final DistributionSetAssignmentResult assignmentResult = prepareDistributionSetAndAssign(controllerId);
+        final Long actionId = getFirstAssignedActionId(assignmentResult);
+        // verify action status is in WAIT_FOR_CONFIRMATION
+        assertActionStatusList(actionId, 1, Status.WAIT_FOR_CONFIRMATION);
+
+        // assert download and install message
+        waitUntilEventMessagesAreDispatchedToTarget(EventTopic.DOWNLOAD_AND_INSTALL);
+        assertDownloadAndInstallMessage(assignmentResult.getDistributionSet().getModules(), controllerId);
+    }
+
+    @Test
+    @Description("Register a target and send a update action status (denied). Verify if the updated action status is correct.")
+    @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 1),
+            @Expect(type = TargetAssignDistributionSetEvent.class, count = 1),
+            @Expect(type = ActionUpdatedEvent.class), @Expect(type = ActionCreatedEvent.class, count = 1),
+            @Expect(type = SoftwareModuleCreatedEvent.class, count = 3),
+            @Expect(type = SoftwareModuleUpdatedEvent.class, count = 6),
+            @Expect(type = DistributionSetCreatedEvent.class, count = 1),
+            @Expect(type = TargetUpdatedEvent.class, count = 1),
+            @Expect(type = TargetPollEvent.class, count = 1),
+            @Expect(type = TenantConfigurationCreatedEvent.class, count = 1) })
+    void deniedActionStatus() {
+        enableConfirmationFlow();
+        final String controllerId = TARGET_PREFIX + "deniedActionStatus";
+        final DistributionSetAssignmentResult assignmentResult = registerTargetAndAssignDistributionSet(controllerId);
+        final Long actionId = getFirstAssignedActionId(assignmentResult);
+        assertActionStatusList(actionId, 1, Status.WAIT_FOR_CONFIRMATION);
+        createAndSendActionStatusUpdateMessage(controllerId, actionId, DmfActionStatus.DENIED);
+        assertActionStatusList(actionId, 2, Status.WAIT_FOR_CONFIRMATION);
+    }
 }