Public Sequencer: Add automation to revoke ParticipantSynchronizerPermission based on ValidatorUnpermision. Extend integration test to to check both temporary revocation and permanent revocation

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/PermissionedSynchronizerIntegrationTest.scala

@@ -1,12 +1,20 @@
 package org.lfdecentralizedtrust.splice.integration.tests
 
 import com.digitalasset.canton.HasExecutionContext
+import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.DsoRules_UnpermissionValidator
+import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.actionrequiringconfirmation.ARC_DsoRules
+import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.dsorules_actionrequiringconfirmation.SRARC_UnpermissionValidator
 import org.lfdecentralizedtrust.splice.config.ConfigTransforms
 import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
 import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.IntegrationTest
 import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient.SynchronizerPermissionState
 import org.lfdecentralizedtrust.splice.util.*
 
+import java.time.Instant
+import java.util.Optional
+import com.digitalasset.canton.data.CantonTimestamp
+import com.digitalasset.canton.logging.SuppressionRule
+
 class PermissionedSynchronizerIntegrationTest
     extends IntegrationTest
     with HasExecutionContext
@@ -120,5 +128,91 @@ class PermissionedSynchronizerIntegrationTest
       bobValidatorBackend.startSync()
       bobValidatorBackend.onboardUser("TestUserBob")
     }
+
+    val bobParticipantId = bobValidatorBackend.participantClient.id.toProtoPrimitive
+    val suspendTime = env.environment.clock.now.plus(java.time.Duration.ofHours(1)).toInstant
+
+    loggerFactory.suppress(
+      SuppressionRule.Level(
+        org.slf4j.event.Level.WARN // because unpermissioning Bob leads to many warnings from sequencer
+      )
+    ) {
+      clue("SVs vote to temporarily suspend Bob") {
+        manuallyUnpermissionValidator(bobParticipantId, Some(suspendTime), revoked = false)
+      }
+
+      clue("Verify Bob's ParticipantSynchronizerPermission is updated with loginAfter") {
+        eventually() {
+          sv1ScanBackend.getParticipantSynchronizerPermission(
+            decentralizedSynchronizerId.toProtoPrimitive,
+            bobParticipantId,
+          ) shouldBe Some(
+            SynchronizerPermissionState(Some(CantonTimestamp.assertFromInstant(suspendTime)))
+          )
+        }
+      }
+      clue("SVs vote to permanently revoke Bob") {
+        manuallyUnpermissionValidator(bobParticipantId, None, revoked = true)
+      }
+
+      clue("Verify Bob's ParticipantSynchronizerPermission is completely removed") {
+        eventually() {
+          sv1ScanBackend.getParticipantSynchronizerPermission(
+            decentralizedSynchronizerId.toProtoPrimitive,
+            bobParticipantId,
+          ) shouldBe None
+        }
+      }
+
+      bobValidatorBackend.stop() // to avoid logs in canton_before_shutdown.clog
+
+    }
+
+    def manuallyUnpermissionValidator(
+        participantId: String,
+        loginAfter: Option[Instant],
+        revoked: Boolean,
+    ): Unit = {
+      val action = new ARC_DsoRules(
+        new SRARC_UnpermissionValidator(
+          new DsoRules_UnpermissionValidator(
+            participantId,
+            loginAfter.map(Optional.of(_)).getOrElse(Optional.empty()),
+            java.lang.Boolean.valueOf(revoked),
+          )
+        )
+      )
+
+      val (_, voteRequest) = actAndCheck(
+        s"SV1 creates vote request to unpermission $participantId (revoked=$revoked)",
+        eventuallySucceeds() {
+          sv1Backend.createVoteRequest(
+            sv1Backend.getDsoInfo().svParty.toProtoPrimitive,
+            action,
+            "url",
+            "description",
+            sv1Backend.getDsoInfo().dsoRules.payload.config.voteRequestTimeout,
+            None,
+          )
+        },
+      )(
+        "vote request has been created",
+        _ => sv1Backend.listVoteRequests().filter(_.payload.action == action).head,
+      )
+
+      Seq(sv2Backend, sv3Backend).foreach { sv =>
+        clue(s"${sv.participantClient.name} accepts the vote request") {
+          eventuallySucceeds() {
+            sv.castVote(
+              voteRequest.contractId,
+              isAccepted = true,
+              "url",
+              "description",
+            )
+          }
+        }
+      }
+    }
+
   }
 }

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/environment/TopologyAdminConnection.scala

@@ -198,10 +198,19 @@ abstract class TopologyAdminConnection(
       participantId: ParticipantId,
       permission: ParticipantPermission,
       retryFor: RetryFor,
+      limits: Option[ParticipantSynchronizerLimits] = None,
+      loginAfter: Option[CantonTimestamp] = None,
   )(implicit
       tc: TraceContext,
       ec: ExecutionContext,
   ): Future[TopologyResult[ParticipantSynchronizerPermission]] = {
+    val expectedMapping = ParticipantSynchronizerPermission(
+      synchronizerId = synchronizerId,
+      participantId = participantId,
+      permission = permission,
+      limits = limits,
+      loginAfter = loginAfter,
+    )
     ensureTopologyMappingO(
       TopologyStoreId.Synchronizer(synchronizerId),
       s"ParticipantSynchronizerPermission with $permission for $participantId",
@@ -216,28 +225,37 @@ abstract class TopologyAdminConnection(
           )
           .subflatMap { results =>
             results.headOption match {
-              case Some(result) if result.mapping.permission == permission =>
+              case Some(result) if result.mapping == expectedMapping =>
                 Right(result)
               case other =>
                 Left(other)
             }
           },
       update = { _ =>
         Right(
-          ParticipantSynchronizerPermission(
-            synchronizerId = synchronizerId,
-            participantId = participantId,
-            permission = permission,
-            limits = None,
-            loginAfter = None,
-          )
+          expectedMapping
         )
       },
       isProposal = true,
       retryFor = retryFor,
     )
   }
 
+  def ensureParticipantSynchronizerPermissionRemoved(
+      synchronizerId: SynchronizerId,
+      participantId: ParticipantId,
+  )(implicit tc: TraceContext, ec: ExecutionContext): Future[Unit] = {
+    ensureTopologyMappingRemoved(
+      s"Remove ParticipantSynchronizerPermission for $participantId on $synchronizerId",
+      synchronizerId,
+      listParticipantSynchronizerPermission(
+        synchronizerId,
+        participantId.filterString,
+      ).map(_.headOption),
+      proposal = true,
+    )
+  }
+
   def listPartyToParticipant(
       store: Option[TopologyStoreId] = None,
       // list only active (non-removed) mappings by default; this matches the Canton console defaults

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala

@@ -297,6 +297,13 @@ class SvDsoAutomationService(
           connection(SpliceLedgerConnectionPriority.High),
         )
       )
+      registerTrigger(
+        new ValidatorUnpermissionTrigger(
+          triggerContext,
+          dsoStore,
+          participantAdminConnection,
+        )
+      )
     }
     registerTrigger(
       new SvOnboardingRequestTrigger(
@@ -792,5 +799,6 @@ object SvDsoAutomationService extends AutomationServiceCompanion {
       aTrigger[ReconcileSequencingParametersTrigger],
       aTrigger[GrantValidatorPermissionTrigger],
       aTrigger[ValidatorLicenseRequestTrigger],
+      aTrigger[ValidatorUnpermissionTrigger],
     )
 }

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/ValidatorUnpermissionTrigger.scala

@@ -0,0 +1,99 @@
+// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package org.lfdecentralizedtrust.splice.sv.automation
+
+import com.digitalasset.canton.data.CantonTimestamp
+import com.digitalasset.canton.topology.{ParticipantId, SynchronizerId}
+import com.digitalasset.canton.topology.transaction.ParticipantPermission.Submission
+import com.digitalasset.canton.tracing.TraceContext
+import io.opentelemetry.api.trace.Tracer
+import org.apache.pekko.stream.Materializer
+import org.lfdecentralizedtrust.splice.automation.{
+  OnAssignedContractTrigger,
+  TaskOutcome,
+  TaskSuccess,
+  TriggerContext,
+}
+import org.lfdecentralizedtrust.splice.codegen.java.splice.validatorunpermission.ValidatorUnpermission
+import org.lfdecentralizedtrust.splice.environment.{ParticipantAdminConnection, RetryFor}
+import org.lfdecentralizedtrust.splice.sv.store.SvDsoStore
+import org.lfdecentralizedtrust.splice.util.AssignedContract
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.jdk.OptionConverters.*
+
+class ValidatorUnpermissionTrigger(
+    override protected val context: TriggerContext,
+    store: SvDsoStore,
+    participantAdminConnection: ParticipantAdminConnection,
+)(implicit
+    override val ec: ExecutionContext,
+    mat: Materializer,
+    tracer: Tracer,
+) extends OnAssignedContractTrigger.Template[
+      ValidatorUnpermission.ContractId,
+      ValidatorUnpermission,
+    ](
+      store,
+      ValidatorUnpermission.COMPANION,
+    ) {
+
+  override protected def completeTask(
+      unpermission: AssignedContract[ValidatorUnpermission.ContractId, ValidatorUnpermission]
+  )(implicit tc: TraceContext): Future[TaskOutcome] = {
+    val payload = unpermission.payload
+
+    ParticipantId
+      .fromProtoPrimitive(payload.participantId, "participantId")
+      .fold(
+        err =>
+          Future.successful(
+            TaskSuccess(s"Skipping ValidatorUnpermission with invalid participantId: $err")
+          ),
+        participantId => {
+          for {
+            dsoRules <- store.getDsoRules()
+            synchronizerId = SynchronizerId.tryFromString(
+              dsoRules.payload.config.decentralizedSynchronizer.activeSynchronizerId
+            )
+
+            outcome <-
+              if (payload.revoked) {
+                participantAdminConnection
+                  .ensureParticipantSynchronizerPermissionRemoved(
+                    synchronizerId,
+                    participantId,
+                  )
+                  .map { _ =>
+                    TaskSuccess(
+                      s"Permanently revoked ParticipantSynchronizerPermission for participant $participantId"
+                    )
+                  }
+              } else {
+                for {
+                  existingMappings <- participantAdminConnection
+                    .listParticipantSynchronizerPermission(
+                      synchronizerId,
+                      participantId.filterString,
+                    )
+
+                  _ <- participantAdminConnection.ensureParticipantSynchronizerPermission(
+                    synchronizerId = synchronizerId,
+                    participantId = participantId,
+                    permission = Submission,
+                    retryFor = RetryFor.Automation,
+                    limits = existingMappings.headOption.flatMap(_.mapping.limits),
+                    loginAfter = payload.loginAfter.toScala
+                      .map(t => CantonTimestamp.assertFromInstant(t)),
+                  )
+                } yield TaskSuccess(
+                  s"Temporarily revoked ParticipantSynchronizerPermission for participant $participantId (loginAfter: ${payload.loginAfter.toScala
+                      .map(t => CantonTimestamp.assertFromInstant(t))})"
+                )
+              }
+          } yield outcome
+        },
+      )
+  }
+}