Merge pull request #425 from pschord/remote-admin

Create PsAdmin framework

build.sbt

@@ -25,6 +25,7 @@ lazy val commonSettings = Seq(
     "-sourcepath", baseDirectory.value.getAbsolutePath // needed for scaladoc relative source paths
     )
   },
+  classLoaderLayeringStrategy := ClassLoaderLayeringStrategy.Flat,
   resolvers += "Sonatype OSS Snapshots" at "https://oss.sonatype.org/content/repositories/snapshots",
   libraryDependencies ++= Seq(
     "com.typesafe.akka"          %% "akka-actor"    % "2.4.4",
@@ -46,9 +47,9 @@ lazy val commonSettings = Seq(
     "org.ini4j"                  % "ini4j"         % "0.5.4",
     "org.scala-graph"            %% "graph-core"    % "1.12.5",
     "io.kamon"                   %% "kamon-bundle" % "2.1.0",
-    "io.kamon"                   %% "kamon-apm-reporter" % "2.1.0"
+    "io.kamon"                   %% "kamon-apm-reporter" % "2.1.0",
+    "org.json4s"                 %% "json4s-native" % "3.6.8",
   ),
-  classLoaderLayeringStrategy := ClassLoaderLayeringStrategy.Flat
 )
 
 lazy val pscryptoSettings = Seq(

common/src/main/scala/net/psforever/config/ConfigParser.scala

@@ -190,6 +190,10 @@ trait ConfigParser {
     config_map = map
   }
 
+  def GetRawConfig : Map[String, Any] = {
+    config_map
+  }
+
   def FormatErrors(invalidResult : Invalid) : Seq[String] = {
     var count = 0;
 
@@ -209,7 +213,7 @@ trait ConfigParser {
   }
 
   protected def parseSection(sectionIni : org.ini4j.Profile.Section, entry : ConfigEntry, map : Map[String, Any]) : ValidationResult = {
-    var rawValue = sectionIni.get(entry.key)
+    var rawValue = sectionIni.get(entry.key, 0)
     val full_key : String = sectionIni.getName + "." + entry.key
 
     val value = if (rawValue == null) {

common/src/main/scala/net/psforever/objects/zones/InterstellarCluster.scala

@@ -64,6 +64,16 @@ class InterstellarCluster(zones : List[Zone]) extends Actor {
         case None => //zone_number does not exist
           sender ! Zone.Lattice.NoValidSpawnPoint(zone_number, None)
       }
+    case InterstellarCluster.ListPlayers() =>
+      var players : List[String] = List()
+
+      for(zone <- zones) {
+        val zonePlayers = zone.Players
+        for (player <- zonePlayers) {
+          players ::= player.name
+        }
+      }
+      sender ! InterstellarCluster.PlayerList(players)
 
     case msg @ Zone.Lattice.RequestSpecificSpawnPoint(zone_number, _, _, _) =>
       recursiveFindWorldInCluster(zones.iterator, _.Number == zone_number) match {
@@ -188,6 +198,9 @@ object InterstellarCluster {
     */
   final case class GiveWorld(zoneId : String, zone : Zone)
 
+  final case class ListPlayers()
+  final case class PlayerList(players : List[String])
+
   /**
     * Signal to the cluster that a new client needs to be initialized for all listed `Zone` destinations.
     * @see `Zone`

config/worldserver.ini.dist

@@ -119,7 +119,7 @@ ServerType = Released
 #        Important:   Must be different from the worldserver.ListeningPort. Ports below 1024 are
 #                     privileged on Linux and may require root.
 #        Range:       [1, 65535] - (UDP port 1, UDP port 65535)
-#        Default:     51000      - (Listen on UDP port 5100)
+#        Default:     51000      - (Listen on UDP port 51000)
 
 ListeningPort = 51000
 
@@ -134,6 +134,22 @@ ListeningPort = 51000
 
 CreateMissingAccounts = yes
 
+###################################################################################################
+# PSADMIN SETTINGS
+###################################################################################################
+
+[psadmin]
+
+#    ListeningPort (int)
+#        Description: The TCP listening port for the server admin interface.
+#        Important:   Must be different from the worldserver and loginserver ListeningPort.
+#                     Ports below 1024 are privileged on Linux and may require root.
+#                     NEVER EXPOSE THIS PORT TO THE INTERNET! CHECK YOUR FIREWALL CONFIG.
+#        Range:       [1, 65535] - (TCP port 1, TCP port 65535)
+#        Default:     51002      - (Listen on TCP port 51002)
+
+ListeningPort = 51002
+
 ###################################################################################################
 # NETWORK SETTINGS
 ###################################################################################################

pslogin/src/main/scala/Database.scala

@@ -1,4 +1,5 @@
 // Copyright (c) 2017 PSForever
+import net.psforever.WorldConfig
 import com.github.mauricio.async.db.postgresql.PostgreSQLConnection
 import com.github.mauricio.async.db.{Configuration, QueryResult, RowData, SSLConfiguration}
 import scala.util.{Try,Success,Failure}

pslogin/src/main/scala/LoginSessionActor.scala

@@ -14,6 +14,7 @@ import com.github.mauricio.async.db.{Connection, QueryResult}
 import net.psforever.objects.Account
 import net.psforever.objects.DefaultCancellable
 import net.psforever.types.PlanetSideEmpire
+import net.psforever.WorldConfig
 import services.ServiceManager
 import services.ServiceManager.Lookup
 import services.account.{ReceiveIPAddress, RetrieveIPAddress, StoreAccountData}

pslogin/src/main/scala/PsLogin.scala

@@ -15,6 +15,8 @@ import net.psforever.config.{Invalid, Valid}
 import net.psforever.crypto.CryptoInterface
 import net.psforever.objects.zones._
 import net.psforever.objects.guid.TaskResolver
+import net.psforever.psadmin.PsAdminActor
+import net.psforever.WorldConfig
 import org.slf4j
 import org.fusesource.jansi.Ansi._
 import org.fusesource.jansi.Ansi.Color._
@@ -260,6 +262,7 @@ object PsLogin {
 
     val loginServerPort = WorldConfig.Get[Int]("loginserver.ListeningPort")
     val worldServerPort = WorldConfig.Get[Int]("worldserver.ListeningPort")
+    val psAdminPort = WorldConfig.Get[Int]("psadmin.ListeningPort")
 
     val netSim : Option[NetworkSimulatorParameters] = WorldConfig.Get[Boolean]("developer.NetSim.Active") match {
       case true =>
@@ -295,6 +298,8 @@ object PsLogin {
     loginListener = system.actorOf(Props(new UdpListener(loginRouter, "login-session-router", LoginConfig.serverIpAddress, loginServerPort, netSim)), "login-udp-endpoint")
     worldListener = system.actorOf(Props(new UdpListener(worldRouter, "world-session-router", LoginConfig.serverIpAddress, worldServerPort, netSim)), "world-udp-endpoint")
 
+    val adminListener = system.actorOf(Props(new TcpListener(classOf[PsAdminActor], "psadmin-client-", InetAddress.getLoopbackAddress, psAdminPort)), "psadmin-tcp-endpoint")
+
     logger.info(s"NOTE: Set client.ini to point to ${LoginConfig.serverIpAddress.getHostAddress}:$loginServerPort")
 
     // Add our shutdown hook (this works for Control+C as well, but not in Cygwin)
@@ -321,8 +326,5 @@ object PsLogin {
     Locale.setDefault(Locale.US); // to have floats with dots, not comma...
     this.args = args
     run()
-
-    // Wait forever until the actor system shuts down
-    Await.result(system.whenTerminated, Duration.Inf)
   }
 }

pslogin/src/main/scala/SessionRouter.scala

@@ -9,6 +9,7 @@ import scala.collection.mutable
 import akka.actor.SupervisorStrategy.Stop
 import net.psforever.packet.PacketCoding
 import net.psforever.packet.control.ConnectionClose
+import net.psforever.WorldConfig
 import services.ServiceManager
 import services.ServiceManager.Lookup
 import services.account.{IPAddress, StoreIPAddress}

pslogin/src/main/scala/TcpListener.scala

@@ -0,0 +1,53 @@
+// Copyright (c) 2020 PSForever
+import java.net.{InetAddress, InetSocketAddress}
+
+import akka.actor.SupervisorStrategy.Stop
+import akka.actor.{Actor, ActorRef, OneForOneStrategy, Props, Terminated}
+import akka.io._
+import scodec.bits._
+import scodec.interop.akka._
+import akka.util.ByteString
+
+class TcpListener[T <: Actor](actorClass : Class[T],
+                  nextActorName : String,
+                  listenAddress : InetAddress,
+                  port : Int) extends Actor {
+  private val log = org.log4s.getLogger(self.path.name)
+
+  override def supervisorStrategy = OneForOneStrategy() {
+    case _ => Stop
+  }
+
+  import context.system
+
+  IO(Tcp) ! Tcp.Bind(self, new InetSocketAddress(listenAddress, port))
+
+  var sessionId = 0L
+  var bytesRecevied = 0L
+  var bytesSent = 0L
+  var nextActor : ActorRef = Actor.noSender
+
+  def receive = {
+    case Tcp.Bound(local) =>
+      log.info(s"Now listening on TCP:$local")
+
+      context.become(ready(sender()))
+    case Tcp.CommandFailed(Tcp.Bind(_, address, _, _, _)) =>
+      log.error("Failed to bind to the network interface: " + address)
+      context.system.terminate()
+    case default =>
+      log.error(s"Unexpected message $default")
+  }
+
+  def ready(socket: ActorRef): Receive = {
+    case Tcp.Connected(remote, local) =>
+      val connection = sender()
+      val session = sessionId
+      val handler = context.actorOf(Props(actorClass, remote, connection), nextActorName + session)
+      connection ! Tcp.Register(handler)
+      sessionId += 1
+    case Tcp.Unbind  => socket ! Tcp.Unbind
+    case Tcp.Unbound => context.stop(self)
+    case default => log.error(s"Unhandled message: $default")
+  }
+}

pslogin/src/main/scala/WorldConfig.scala

@@ -1,4 +1,6 @@
 // Copyright (c) 2019 PSForever
+package net.psforever
+
 import scala.util.matching.Regex
 import net.psforever.config._
 import scala.concurrent.duration._
@@ -31,6 +33,9 @@ object WorldConfig extends ConfigParser {
       ConfigEntryTime("Session.InboundGraceTime", 1 minute, Constraints.min(10 seconds)),
       ConfigEntryTime("Session.OutboundGraceTime", 1 minute, Constraints.min(10 seconds))
     ),
+    ConfigSection("psadmin",
+      ConfigEntryInt("ListeningPort", 51002, Constraints.min(1), Constraints.max(65535))
+    ),
     ConfigSection("developer",
       ConfigEntryBool ("NetSim.Active", false),
       ConfigEntryFloat("NetSim.Loss", 0.02f, Constraints.min(0.0f), Constraints.max(1.0f)),

pslogin/src/main/scala/psadmin/CmdInternal.scala

@@ -0,0 +1,30 @@
+package net.psforever.psadmin
+
+import net.psforever.WorldConfig
+import scala.collection.mutable.Map
+
+object CmdInternal {
+
+  def cmdDumpConfig(args : Array[String]) = {
+    val config = WorldConfig.GetRawConfig
+
+    CommandGoodResponse(s"Dump of WorldConfig", config)
+  }
+
+  def cmdThreadDump(args : Array[String]) = {
+    import scala.collection.JavaConverters._
+
+    var data = Map[String,Any]()
+    val traces = Thread.getAllStackTraces().asScala
+    var traces_fmt = List[String]()
+
+    for ((thread, trace) <- traces) {
+      val info = s"Thread ${thread.getId} - ${thread.getName}\n"
+      traces_fmt = traces_fmt ++ List(info + trace.mkString("\n"))
+    }
+
+    data{"trace"} = traces_fmt
+
+    CommandGoodResponse(s"Dump of ${traces.size} threads", data)
+  }
+}

pslogin/src/main/scala/psadmin/CmdListPlayers.scala

@@ -0,0 +1,41 @@
+// Copyright (c) 2020 PSForever
+package net.psforever.psadmin
+
+import java.net.InetAddress
+import java.net.InetSocketAddress
+import akka.actor.{ActorRef, ActorSystem, Props}
+import akka.actor.{Actor, Stash}
+import akka.io.Tcp
+import scodec.bits._
+import scodec.interop.akka._
+import scala.collection.mutable.Map
+import akka.util.ByteString
+import com.typesafe.config.ConfigFactory
+import scala.collection.JavaConverters._
+import net.psforever.objects.zones.InterstellarCluster
+
+import services.ServiceManager.Lookup
+import services._
+
+class CmdListPlayers(args : Array[String], services : Map[String,ActorRef]) extends Actor {
+  private [this] val log = org.log4s.getLogger(self.path.name)
+
+  override def preStart = {
+    services{"cluster"} ! InterstellarCluster.ListPlayers()
+  }
+
+  override def receive = {
+    case InterstellarCluster.PlayerList(players) =>
+      val data = Map[String,Any]()
+      data{"player_count"} = players.size
+      data{"player_list"} = Array[String]()
+
+      if (players.isEmpty) {
+        context.parent ! CommandGoodResponse("No players currently online!", data)
+      } else {
+        data{"player_list"} = players
+        context.parent ! CommandGoodResponse(s"${players.length} players online\n", data)
+      }
+    case default => log.error(s"Unexpected message $default")
+  }
+}

pslogin/src/main/scala/psadmin/CmdShutdown.scala

@@ -0,0 +1,17 @@
+// Copyright (c) 2020 PSForever
+package net.psforever.psadmin
+
+import akka.actor.{Actor,ActorRef}
+import scala.collection.mutable.Map
+
+class CmdShutdown(args : Array[String], services : Map[String,ActorRef]) extends Actor {
+  override def preStart = {
+    var data = Map[String,Any]()
+    context.parent ! CommandGoodResponse("Shutting down", data)
+    context.system.terminate()
+  }
+
+  override def receive = {
+    case default =>
+  }
+}

pslogin/src/main/scala/psadmin/PsAdminActor.scala

@@ -0,0 +1,195 @@
+// Copyright (c) 2020 PSForever
+package net.psforever.psadmin
+
+import java.net.InetAddress
+import java.net.InetSocketAddress
+import akka.actor.{ActorRef, ActorSystem, Props}
+import akka.actor.{Actor, Stash}
+import akka.io.Tcp
+import scodec.bits._
+import scodec.interop.akka._
+import scala.collection.mutable.Map
+import akka.util.ByteString
+import com.typesafe.config.ConfigFactory
+import scala.collection.JavaConverters._
+import net.psforever.objects.zones.InterstellarCluster
+
+import org.json4s._
+import org.json4s.Formats._
+import org.json4s.native.Serialization.write
+
+import services.ServiceManager.Lookup
+import services._
+
+object PsAdminActor {
+  val whiteSpaceRegex = """\s+""".r
+}
+
+class PsAdminActor(peerAddress : InetSocketAddress, connection : ActorRef) extends Actor with Stash {
+  private [this] val log = org.log4s.getLogger(self.path.name)
+
+  val services = Map[String,ActorRef]()
+  val servicesToResolve = Array("cluster")
+  var buffer = ByteString()
+
+  implicit val formats = DefaultFormats // for JSON serialization
+
+  case class CommandCall(operation : String, args : Array[String])
+
+  override def preStart() = {
+    log.trace(s"PsAdmin connection started $peerAddress")
+
+    for (service <- servicesToResolve) {
+      ServiceManager.serviceManager ! Lookup(service)
+    }
+  }
+
+  override def receive = ServiceLookup
+
+  def ServiceLookup : Receive = {
+    case ServiceManager.LookupResult(service, endpoint) =>
+      services{service} = endpoint
+
+      if (services.size == servicesToResolve.size) {
+        unstashAll()
+        context.become(ReceiveCommand)
+      }
+
+    case default => stash()
+  }
+
+  def ReceiveCommand : Receive = {
+    case Tcp.Received(data) =>
+      buffer ++= data
+
+      var pos = -1;
+      var amount = 0
+      do {
+        pos = buffer.indexOf('\n')
+        if (pos != -1) {
+          val (cmd, rest) = buffer.splitAt(pos)
+          buffer = rest.drop(1); // drop the newline
+
+          // make sure the CN cant crash us
+          val line = cmd.decodeString("utf-8").trim
+
+          if (line != "") {
+            val tokens = PsAdminActor.whiteSpaceRegex.split(line)
+            val cmd = tokens.head
+            val args = tokens.tail
+
+
+            amount += 1
+            self ! CommandCall(cmd, args)
+          }
+        }
+      } while (pos != -1)
+
+      if (amount > 0)
+        context.become(ProcessCommands)
+
+    case Tcp.PeerClosed =>
+      context.stop(self)
+
+    case default =>
+      log.error(s"Unexpected message $default")
+  }
+
+  /// Process all buffered commands and stash other ones
+  def ProcessCommands : Receive = {
+    case c : CommandCall =>
+      stash()
+      unstashAll()
+      context.become(ProcessCommand)
+
+    case default =>
+      stash()
+      unstashAll()
+      context.become(ReceiveCommand)
+  }
+
+  /// Process a single command
+  def ProcessCommand : Receive = {
+    case CommandCall(cmd, args) =>
+      val data = Map[String,Any]()
+
+      if (cmd == "help" || cmd == "?") {
+        if (args.size == 0) {
+          var resp = "PsAdmin command usage\n"
+
+          for ((command, info) <- PsAdminCommands.commands) {
+            resp += s"${command} - ${info.usage}\n"
+          }
+
+          data{"message"} = resp
+        } else {
+          if (PsAdminCommands.commands.contains(args(0))) {
+            val info = PsAdminCommands.commands{args(0)}
+
+            data{"message"} = s"${args(0)} - ${info.usage}"
+          } else {
+            data{"message"} = s"Unknown command ${args(0)}"
+            data{"error"} = true
+          }
+        }
+
+        sendLine(write(data.toMap))
+      } else if (PsAdminCommands.commands.contains(cmd)) {
+        val cmd_template = PsAdminCommands.commands{cmd}
+
+        cmd_template match {
+          case PsAdminCommands.Command(usage, handler) =>
+            context.actorOf(Props(handler, args, services))
+
+          case PsAdminCommands.CommandInternal(usage, handler) =>
+            val resp = handler(args)
+
+            resp match {
+              case CommandGoodResponse(msg, data) =>
+                data{"message"} = msg
+                sendLine(write(data.toMap))
+
+              case CommandErrorResponse(msg, data) =>
+                data{"message"} = msg
+                data{"error"} = true
+                sendLine(write(data.toMap))
+            }
+
+            context.become(ProcessCommands)
+        }
+      } else {
+        data{"message"} = "Unknown command"
+        data{"error"} = true
+        sendLine(write(data.toMap))
+        context.become(ProcessCommands)
+      }
+
+    case resp : CommandResponse =>
+      resp match {
+        case CommandGoodResponse(msg, data) =>
+          data{"message"} = msg
+          sendLine(write(data.toMap))
+
+        case CommandErrorResponse(msg, data) =>
+          data{"message"} = msg
+          data{"error"} = true
+          sendLine(write(data.toMap))
+      }
+
+      context.become(ProcessCommands)
+      context.stop(sender())
+    case default =>
+      stash()
+      unstashAll()
+      context.become(ProcessCommands)
+  }
+
+  def sendLine(line : String) = {
+    ByteVector.encodeUtf8(line + "\n") match {
+      case Left(e) =>
+        log.error(s"Message encoding failure: $e")
+      case Right(bv) =>
+        connection ! Tcp.Write(bv.toByteString)
+    }
+  }
+}

pslogin/src/main/scala/psadmin/PsAdminCommands.scala

@@ -0,0 +1,31 @@
+// Copyright (c) 2020 PSForever
+package net.psforever.psadmin
+
+import scala.collection.mutable.Map
+
+sealed trait CommandResponse
+case class CommandGoodResponse(message : String, data : Map[String,Any]) extends CommandResponse
+case class CommandErrorResponse(message : String, data : Map[String,Any]) extends CommandResponse
+
+object PsAdminCommands {
+  import CmdInternal._
+
+  val commands : Map[String,CommandInfo] = Map(
+    "list_players" -> Command("""Return a list of players connected to the interstellar cluster.""", classOf[CmdListPlayers]),
+    "dump_config" -> CommandInternal("""Dumps entire running config.""", cmdDumpConfig),
+    "shutdown" -> Command("""Shuts down the server forcefully.""", classOf[CmdShutdown]),
+    "thread_dump" -> CommandInternal("""Returns all thread's stack traces.""", cmdThreadDump)
+  )
+
+  sealed trait CommandInfo {
+    def usage: String
+  }
+
+  /// A command with full access to the ActorSystem and WorldServer services.
+  /// Spawns an Actor to handle the request and the service queries
+  case class Command[T](usage : String, handler : Class[T]) extends CommandInfo
+
+  /// A command without access to the ActorSystem or any services
+  case class CommandInternal(usage : String, handler : ((Array[String]) => CommandResponse)) extends CommandInfo
+}
+

pslogin/src/test/scala/ConfigTest.scala

@@ -3,6 +3,7 @@ import java.io._
 import scala.io.Source
 import org.specs2.mutable._
 import net.psforever.config._
+import net.psforever.WorldConfig
 import scala.concurrent.duration._
 
 class ConfigTest extends Specification {