/**
 * Copyright (C) 2009-2018 Lightbend Inc. <https://www.lightbend.com>
 */

package akka.cluster

import akka.cluster.MemberStatus.Up
import akka.testkit.{ AkkaSpec, LongRunningTest }
import com.typesafe.config.{ Config, ConfigFactory }

import scala.concurrent.duration._
import scala.collection.{ immutable ⇒ im }

object JoinConfigCompatCheckerSpec {

}

class JoinConfigCompatCheckerSpec extends AkkaSpec with ClusterTestKit {

  "A Joining Node" must {

    "be allowed to join a cluster when its configuration is compatible" taggedAs LongRunningTest in {

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(configWithChecker)
      val joiningNode = clusterTestUtil.newActorSystem(configWithChecker)
      clusterTestUtil.formCluster()

      try {
        awaitCond(Cluster(joiningNode).readView.status == Up, message = "awaiting joining node to be 'Up'")
      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "NOT be allowed to join a cluster when its configuration is incompatible" taggedAs LongRunningTest in {
      // this config is NOT compatible with the cluster config
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # this config is incompatible
              config-compat-test = "test2"
              configuration-compatibility-check {
                enforce-on-join = on
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                }
              }
            }
            """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(configWithChecker)
      val joiningNode = clusterTestUtil.newActorSystem(joinNodeConfig.withFallback(configWithChecker))

      clusterTestUtil.formCluster()

      try {

        // node will shutdown after unsuccessful join attempt
        within(5.seconds) {
          awaitCond(Cluster(joiningNode).readView.isTerminated)
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "NOT be allowed to join a cluster when one of its required properties are not available on cluster side" taggedAs LongRunningTest in {

      // this config is NOT compatible with the cluster config
      // because there is one missing required configuration property.
      // This test verifies that cluster config are being sent back and checked on joining node as well
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # this config is not available on cluster side
              akka.cluster.config-compat-test-extra = on

              configuration-compatibility-check {
                enforce-on-join = on
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                 akka-cluster-extra = "akka.cluster.JoinConfigCompatCheckerExtraTest"
                }
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(configWithChecker)
      val joiningNode = clusterTestUtil.newActorSystem(joinNodeConfig.withFallback(configWithChecker))

      clusterTestUtil.formCluster()

      try {
        // node will shutdown after unsuccessful join attempt
        within(5.seconds) {
          awaitCond(Cluster(joiningNode).readView.isTerminated)
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "NOT be allowed to join a cluster when one of the cluster required properties are not available on the joining side" taggedAs LongRunningTest in {

      // this config is NOT compatible with the cluster config
      // because there is one missing required configuration property.
      // This test verifies that cluster config are being sent back and checked on joining node as well
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # this config is required on cluster side
              # config-compat-test = "test"

              configuration-compatibility-check {
                enforce-on-join = on
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(configWithChecker)
      val joiningNode = clusterTestUtil.newActorSystem(joinNodeConfig.withFallback(baseConfig))

      clusterTestUtil.formCluster()

      try {
        // node will shutdown after unsuccessful join attempt
        within(5.seconds) {
          awaitCond(Cluster(joiningNode).readView.isTerminated)
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "be allowed to join a cluster when one of its required properties are not available on cluster side but it's configured to NOT enforce it" taggedAs LongRunningTest in {

      // this config is NOT compatible with the cluster config
      // because there is one missing required configuration property.
      // This test verifies that validation on joining side takes 'configuration-compatibility-check.enforce-on-join' in consideration
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # this config is not available on cluster side
              akka.cluster.config-compat-test-extra = on

              configuration-compatibility-check {
                enforce-on-join = off
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                 akka-cluster-extra = "akka.cluster.JoinConfigCompatCheckerExtraTest"
                }
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(configWithChecker)
      val joiningNode = clusterTestUtil.newActorSystem(joinNodeConfig.withFallback(configWithChecker))

      clusterTestUtil.formCluster()

      try {
        // join with compatible node
        awaitCond(Cluster(joiningNode).readView.status == Up, message = "awaiting joining node to be 'Up'")
      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "be allowed to join a cluster when its configuration is incompatible but it's configured to NOT enforce it" taggedAs LongRunningTest in {
      // this config is NOT compatible with the cluster config,
      // but node will ignore the the config check and join anyway
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              configuration-compatibility-check {
                # not enforcing config compat check
                enforce-on-join = off
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                }
              }
             # this config is incompatible
             config-compat-test = "test2"
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(configWithChecker)
      val joiningNode = clusterTestUtil.newActorSystem(joinNodeConfig.withFallback(configWithChecker))

      clusterTestUtil.formCluster()

      try {
        // join with compatible node
        awaitCond(Cluster(joiningNode).readView.status == Up, message = "awaiting joining node to be 'Up'")
      } finally {
        clusterTestUtil.shutdownAll()
      }
    }

    /** This test verifies the built-in JoinConfigCompatCheckerAkkaCluster */
    "NOT be allowed to join a cluster using a different value for akka.cluster.downing-provider-class" taggedAs LongRunningTest in {

      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # using explicit downing provider class
              downing-provider-class = "akka.cluster.AutoDowning"

              configuration-compatibility-check {
                enforce-on-join = on
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(baseConfig)
      val joiningNode = clusterTestUtil.newActorSystem(joinNodeConfig.withFallback(baseConfig))

      clusterTestUtil.formCluster()

      try {
        // node will shutdown after unsuccessful join attempt
        within(5.seconds) {
          awaitCond(Cluster(joiningNode).readView.isTerminated)
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }
  }

  "A First Node" must {

    "be allowed to re-join a cluster when its configuration is compatible" taggedAs LongRunningTest in {

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      val firstNode = clusterTestUtil.newActorSystem(configWithChecker)
      // second node
      val secondNode = clusterTestUtil.newActorSystem(configWithChecker)

      clusterTestUtil.formCluster()

      try {
        // we must wait second node to join the cluster before shutting down the first node
        awaitCond(Cluster(secondNode).readView.status == Up, message = "awaiting second node to be 'Up'")

        val restartedNode = clusterTestUtil.quitAndRestart(firstNode, configWithChecker)
        clusterTestUtil.joinCluster(restartedNode)

        within(20.seconds) {
          awaitCond(Cluster(restartedNode).readView.status == Up, message = "awaiting restarted first node to be 'Up'")
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "NOT be allowed to re-join a cluster when its configuration is incompatible" taggedAs LongRunningTest in {
      // this config is NOT compatible with the cluster config
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # this config is incompatible
              config-compat-test = "test2"
              configuration-compatibility-check {
                enforce-on-join = on
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                }
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      val firstNode = clusterTestUtil.newActorSystem(configWithChecker)
      // second node
      val secondNode = clusterTestUtil.newActorSystem(configWithChecker)

      clusterTestUtil.formCluster()

      try {
        // we must wait second node to join the cluster before shutting down the first node
        awaitCond(Cluster(secondNode).readView.status == Up, message = "awaiting second node to be 'Up'")

        val restartedNode = clusterTestUtil.quitAndRestart(firstNode, joinNodeConfig.withFallback(configWithChecker))
        clusterTestUtil.joinCluster(restartedNode)

        // node will shutdown after unsuccessful join attempt
        within(20.seconds) {
          awaitCond(Cluster(restartedNode).readView.isTerminated)
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "NOT be allowed to re-join a cluster when one of its required properties are not available on cluster side" taggedAs LongRunningTest in {

      // this config is NOT compatible with the cluster config
      // because there is one missing required configuration property.
      // This test verifies that cluster config are being sent back and checked on joining node as well
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # this config is not available on cluster side
              akka.cluster.config-compat-test-extra = on

              configuration-compatibility-check {
                enforce-on-join = on
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                 akka-cluster-extra = "akka.cluster.JoinConfigCompatCheckerExtraTest"
                }
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      val firstNode = clusterTestUtil.newActorSystem(configWithChecker)
      // second node
      val secondNode = clusterTestUtil.newActorSystem(configWithChecker)

      clusterTestUtil.formCluster()

      try {
        // we must wait second node to join the cluster before shutting down the first node
        awaitCond(Cluster(secondNode).readView.status == Up, message = "awaiting second node to be 'Up'")

        val restartedNode = clusterTestUtil.quitAndRestart(firstNode, joinNodeConfig.withFallback(configWithChecker))
        clusterTestUtil.joinCluster(restartedNode)

        // node will shutdown after unsuccessful join attempt
        within(20.seconds) {
          awaitCond(Cluster(restartedNode).readView.isTerminated)
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "NOT be allowed to re-join a cluster when one of the cluster required properties are not available on the joining side" taggedAs LongRunningTest in {

      // this config is NOT compatible with the cluster config
      // because there is one missing required configuration property.
      // This test verifies that cluster config are being sent back and checked on joining node as well
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
              akka.cluster {

                # this config is required on cluster side
                # config-compat-test = "test"

                configuration-compatibility-check {
                  enforce-on-join = on
                }
              }
            """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      val firstNode = clusterTestUtil.newActorSystem(configWithChecker)
      // second node
      val secondNode = clusterTestUtil.newActorSystem(configWithChecker)

      clusterTestUtil.formCluster()

      try {
        // we must wait second node to join the cluster before shutting down the first node
        awaitCond(Cluster(secondNode).readView.status == Up, message = "awaiting second node to be 'Up'")

        val restartedNode = clusterTestUtil.quitAndRestart(firstNode, joinNodeConfig.withFallback(baseConfig))
        clusterTestUtil.joinCluster(restartedNode)

        // node will shutdown after unsuccessful join attempt
        within(20.seconds) {
          awaitCond(Cluster(restartedNode).readView.isTerminated)
        }

      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "be allowed to re-join a cluster when one of its required properties are not available on cluster side but it's configured to NOT enforce it" taggedAs LongRunningTest in {

      // this config is NOT compatible with the cluster config
      // because there is one missing required configuration property.
      // This test verifies that validation on joining side takes 'configuration-compatibility-check.enforce-on-join' in consideration
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # this config is not available on cluster side
              akka.cluster.config-compat-test-extra = on

              configuration-compatibility-check {
                enforce-on-join = off
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                 akka-cluster-extra = "akka.cluster.JoinConfigCompatCheckerExtraTest"
                }
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      val firstNode = clusterTestUtil.newActorSystem(configWithChecker)
      // second node
      val secondNode = clusterTestUtil.newActorSystem(configWithChecker)

      clusterTestUtil.formCluster()

      try {
        // join with compatible node
        // we must wait second node to join the cluster before shutting down the first node
        awaitCond(Cluster(secondNode).readView.status == Up, message = "awaiting second node to be 'Up'")

        val restartedNode = clusterTestUtil.quitAndRestart(firstNode, joinNodeConfig.withFallback(configWithChecker))
        clusterTestUtil.joinCluster(restartedNode)

        // node will will have joined the cluster
        within(20.seconds) {
          awaitCond(Cluster(restartedNode).readView.status == Up, message = "awaiting restarted node to be 'Up'")
        }
      } finally {
        clusterTestUtil.shutdownAll()
      }

    }

    "be allowed to re-join a cluster when its configuration is incompatible but it's configured to NOT enforce it" taggedAs LongRunningTest in {
      // this config is NOT compatible with the cluster config,
      // but node will ignore the the config check and join anyway
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              configuration-compatibility-check {
                # not enforcing config compat check
                enforce-on-join = off
                checkers {
                 akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
                }
              }
             # this config is incompatible
             config-compat-test = "test2"
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      val firstNode = clusterTestUtil.newActorSystem(configWithChecker)
      // second node
      val secondNode = clusterTestUtil.newActorSystem(configWithChecker)

      clusterTestUtil.formCluster()

      try {
        // join with compatible node
        // we must wait second node to join the cluster before shutting down the first node
        awaitCond(Cluster(secondNode).readView.status == Up, message = "awaiting second node to be 'Up'")

        val restartedNode = clusterTestUtil.quitAndRestart(firstNode, joinNodeConfig.withFallback(configWithChecker))
        clusterTestUtil.joinCluster(restartedNode)

        // node will will have joined the cluster
        within(20.seconds) {
          awaitCond(Cluster(restartedNode).readView.status == Up, message = "awaiting restarted node to be 'Up'")
        }
      } finally {
        clusterTestUtil.shutdownAll()
      }
    }

  }

  "A Cluster" must {
    "NOT exchange sensitive config paths with joining node" taggedAs LongRunningTest in {

      // this config has sensitive properties that are not compatible with the cluster
      // the cluster will ignore them, because they are on the sensitive-config-path
      // the cluster won't let it be leaked back to the joining node neither which will fail the join attempt.
      val joinNodeConfig =
        ConfigFactory.parseString(
          """
            akka.cluster {

              # these config are compatible,
              # but won't be leaked back to joining node which will cause it to fail to join
              sensitive.properties {
                username = "abc"
                password = "def"
              }

              configuration-compatibility-check {
                enforce-on-join = on
                checkers {
                  # rogue checker to trick the cluster to leak sensitive data
                  rogue-checker = "akka.cluster.RogueJoinConfigCompatCheckerTest"
                }

                # unset sensitive config paths
                # this will allow the joining node to leak sensitive info and try
                # get back these same properties from the cluster
                sensitive-config-paths {
                  akka = []
                }
              }
            }
          """
        )

      val clusterTestUtil = new ClusterTestUtil(system.name)
      // first node
      clusterTestUtil.newActorSystem(configWithChecker)
      val joiningNode = clusterTestUtil.newActorSystem(joinNodeConfig.withFallback(configWithChecker))

      clusterTestUtil.formCluster()

      try {
        // node will shutdown after unsuccessful join attempt
        within(5.seconds) {
          awaitCond(Cluster(joiningNode).readView.isTerminated)
        }
      } finally {
        clusterTestUtil.shutdownAll()
      }
    }
  }

  val baseConfig: Config =
    ConfigFactory.parseString(
      """
     akka.actor.provider = "cluster"
     akka.coordinated-shutdown.terminate-actor-system = on
     akka.remote.netty.tcp.port = 0
     akka.remote.artery.canonical.port = 0
     """
    )

  val configWithChecker: Config =
    ConfigFactory.parseString(
      """
      akka.cluster {
        config-compat-test = "test"
        sensitive.properties {
          username = "abc"
          password = "def"
        }

        configuration-compatibility-check {
          enforce-on-join = on
          checkers {
           akka-cluster-test = "akka.cluster.JoinConfigCompatCheckerTest"
          }
          sensitive-config-paths {
            akka = [ "akka.cluster.sensitive.properties" ]
          }
        }
      }
    """
    ).withFallback(baseConfig)

}

class JoinConfigCompatCheckerTest extends JoinConfigCompatChecker {
  override def requiredKeys = im.Seq("akka.cluster.config-compat-test")

  override def check(toValidate: Config, actualConfig: Config): ConfigValidation =
    JoinConfigCompatChecker.fullMatch(requiredKeys, toValidate, actualConfig)
}

class JoinConfigCompatCheckerExtraTest extends JoinConfigCompatChecker {
  override def requiredKeys = im.Seq("akka.cluster.config-compat-test-extra")

  override def check(toValidate: Config, actualConfig: Config): ConfigValidation =
    JoinConfigCompatChecker.fullMatch(requiredKeys, toValidate, actualConfig)
}

/** Rogue checker that tries to leak sensitive information */
class RogueJoinConfigCompatCheckerTest extends JoinConfigCompatChecker {

  override def requiredKeys = im.Seq("akka.cluster.sensitive.properties.password", "akka.cluster.sensitive.properties.username")

  /** this check always returns Valid. The goal is to try to make the cluster leak those properties */
  override def check(toValidate: Config, actualConfig: Config): ConfigValidation =
    JoinConfigCompatChecker.fullMatch(requiredKeys, toValidate, actualConfig)
}