AstroTuxLauncher.py

#!/usr/bin/python3

import os
from os import path
import argparse
import json
import tomli, tomli_w
import dataclasses
from dataclasses import dataclass, field
from dataclasses_json import dataclass_json, config
from typing import Optional, List
from utils.misc import ExcludeIfNone, read_build_version, LAUNCHER_VERSION, CONTROL_CODES_SUPPORTED
from utils.termutils import set_window_title
from enum import Enum
from pansi import ansi
import utils.interface as interface
import logging
import sys
from queue import Queue
import shutil
from utils import steam
from utils.net import get_request
from packaging import version
import astro.playfab as playfab
from astro.dedicatedserver import AstroDedicatedServer, ServerStatus
import utils.net as net
import signal
import subprocess
import time
import traceback


"""
Code based on https://github.com/ricky-davis/AstroLauncher
"""

LOGGER = logging.getLogger("Launcher")

BANNER_LOGO = f"""{ansi.weight.bold}
    {ansi.BLUE}___         __           {ansi.YELLOW}______          
   {ansi.BLUE}/   |  _____/ /__________{ansi.YELLOW}/_  __/_  ___  __
  {ansi.BLUE}/ /| | / ___/ __/ ___/ __ \\{ansi.YELLOW}/ / / / / / |/_/
 {ansi.BLUE}/ ___ |(__  ) /_/ /  / /_/ {ansi.YELLOW}/ / / /_/ />  <  
{ansi.BLUE}/_/  |_/____/\\__/_/   \\____{ansi.YELLOW}/_/  \\__,_/_/|_|  {ansi.reset}
"""
BANNER_SUBTITLE = "L a u n c h e r".center(45)

BANNER_TEXT="Unofficial Astroneer Dedicated Server Launcher for Linux"


#
#   Constants
#

NAME = "AstroTuxLauncher"

HELP_COMMAND = f"""What {NAME} should do

    - install: Installs the Astroneer Dedicated Server using steamcmd
    - start: Starts the installed dedicated server
    - update: Updates the Astroneer Dedicated Server using steamcmd
"""

DEPOTDL_PATH = "libs/depotdownloader"
DS_EXECUTABLE = "AstroServer.exe"

ASTRO_SERVER_STATS_URL = "https://astroservercheck.joejoetv.de/api/stats"

class LauncherCommand(Enum):
    """ Represents the command passed to the launcher """
    
    START = "start"
    INSTALL = "install"
    UPDATE = "update"



#
#   Configuration classes
#

class NotificationMethod(Enum):
    """ Represents, which notification method should be used """
    
    NONE = ""
    NTFY = "ntfy"
    DISCORD = "discord"

@dataclass
class DiscordConfig:
    webhookURL: str = None

@dataclass
class NTFYConfig:
    topic: str = None
    serverURL: str = "https://ntfy.sh"

@dataclass
class NotificationConfig:
    method: NotificationMethod = NotificationMethod.NONE
    name: str = "Astroneer Dedicated Server"
    EventWhitelist: List[interface.EventType] = field(default_factory=lambda: [e for e in interface.EventType])
    
    discord: Optional[DiscordConfig] = field(metadata=config(exclude=ExcludeIfNone), default=None)
    ntfy: Optional[NTFYConfig] = field(metadata=config(exclude=ExcludeIfNone), default=None)

@dataclass
class StatusConfig:
    SendStatus: bool = False    # Wether to send status updates
    Interval: int = 120         # Interval in which to send status updates
    EndpointURL: str = ""       # URL to send status updates as GET requests to

@dataclass_json
@dataclass
class LauncherConfig:
    AutoUpdateServer: bool = True   # Wether to automatically install/update the Astroneer DS at start if update is available
    
    CheckNetwork: bool = True       # Wether to perform a network check before starting the Astroneer DS
    OverwritePublicIP: bool = False # Wether to overwrite the PublicIP DS config option with the fetched public IP
    
    # Settings related to notifications
    notifications: NotificationConfig = field(default_factory=NotificationConfig)    # Configuration for notifications
    
    # Settings related to sending status updates
    status: StatusConfig = field(default_factory=StatusConfig)
    
    LogDebugMessages: bool = False  # Wether the the console and log file should include log messages with level logging.DEBUG
    
    AstroServerPath: str = "AstroneerServer"    # The path, where the Astroneer DS installation should reside
    OverrideWinePath: Optional[str] = field(metadata=config(exclude=ExcludeIfNone), default=None)   # Path to wine executable, only used, if set
    WinePrefixPath: str = "winepfx"             # The path, where the Wine prefix should be stored
    LogPath: str = "logs"                       # The path where logs should be saved
    
    PlayfabAPIInterval: int = 2                 # Time to wait between Playfab API requests
    ServerStatusInterval: float = 3             # Time to wait between Server Status checks
    
    DisableEncryption: bool = True  # Wether to disable encryption for the Astroneer DS. CURRENTLY REQUIRED TO BE "True" FOR HOSTING ON LINUX
        
    @staticmethod
    def ensure_toml_config(config_path):
        """
            Reads the launcher configuration and fist creates the config file if not present, populated with the default values
        """
        
        config = None
        
        if path.exists(config_path):
            # If config file exists, read it into a config object
            if not path.isfile(config_path):
                raise ValueError("Specified config path doesn't point to a file!")
            
            with open(config_path, "rb") as tf:
                toml_dict = tomli.load(tf)
            
            # If no "launcher" section is present in the file, create it as empty
            if not ("launcher" in toml_dict.keys()):
                toml_dict = {"launcher": {}}
            
            config = LauncherConfig.from_dict(toml_dict["launcher"])

        else:
            # If config file is not present, create directories and default config
            if not path.exists(path.dirname(config_path)):
                os.makedirs(path.dirname(config_path))
            
            config = LauncherConfig()
        
        # Write config back to file to add missing entried and remove superflous ones
        # In the case of the file not existing prior, it will be created
        config_dict = {"launcher": config.to_dict(encode_json=True)}
        
        with open(config_path, "wb") as tf:
            tomli_w.dump(config_dict, tf)
        
        return config

class AstroTuxLauncher():
    
    def __init__(self, config_path, astro_path, depotdl_exec, force_debug_log=False):
        self.dedicatedserver = None
        self.status_thread = None
        
        # Setup basic logging
        interface.LauncherLogging.prepare()
        interface.LauncherLogging.setup_console()
        
        try:
            self.config_path = path.abspath(config_path)
            
            LOGGER.info(f"Configuration file path: {self.config_path}")
            
            self.config = LauncherConfig.ensure_toml_config(self.config_path)
        except Exception as e:
            LOGGER.error(f"Error while loading config file ({type(e).__name__}): {str(e)}")
            LOGGER.error(f"Please check the config path parameter and/or config file")
            self.exit()
        
        # If cli parameter is specified, it overrides the config value
        if not (astro_path is None):
            self.config = dataclasses.replace(self.config, AstroServerPath=astro_path)
        
        # If flag was passed, overrule config option
        if force_debug_log:
            self.config.LogDebugMessages = True
        
        # Make sure we use absolute paths
        self.config.AstroServerPath = path.abspath(self.config.AstroServerPath)
        self.config.WinePrefixPath = path.abspath(self.config.WinePrefixPath)
        self.config.LogPath = path.abspath(self.config.LogPath)
        
        # Apply wine path override if possible and check that is exists
        self.wineexec = shutil.which("wine")
        self.wineserverexec = shutil.which("wineserver")
        
        if self.config.OverrideWinePath is not None and path.isfile(self.config.OverrideWinePath):
            self.wineexec = path.abspath(self.config.OverrideWinePath)
            self.wineserverexec = path.join(path.dirname(self.wineexec), "wineserver")
        
        if (self.wineexec is None) or (self.wineserverexec is None):
            LOGGER.error("Wine (or Wineserver) executable not found!")
            LOGGER.error("Make sure that you have wine installed and accessible")
            LOGGER.error("or set 'OverrideWinePath' config option to the path of the wine executable")
            self.exit()
        
        # Finish setting up logging
        interface.LauncherLogging.set_log_debug(self.config.LogDebugMessages)
        interface.LauncherLogging.setup_logfile(self.config.LogPath)
        
        self.launcherPath = os.getcwd()
        
        self.depotdl_path = None
        
        # If argument is given, file has to exist
        if depotdl_exec:
            # If {depotdl_exec} is a command, get full path
            wpath = shutil.which(depotdl_exec)
            if wpath is not None:
                depotdl_exec = wpath
            
            if path.isfile(depotdl_exec):
                self.depotdl_path = path.abspath(depotdl_exec)
                LOGGER.info(f"DepotDownloader path overridden: {self.depotdl_path}")
            else:
                LOGGER.warning("The given DepotDownloader path doesn't point to a file, using default path")
        
        # If argument is not given, default path is used and may not exists yet, so create directories
        if self.depotdl_path is None:
            self.depotdl_path = path.abspath(DEPOTDL_PATH)
            os.makedirs(path.dirname(self.depotdl_path), exist_ok=True)
        
        # Log some information about loaded paths, configs, etc.
        LOGGER.info(f"Working directory: {self.launcherPath}")
        LOGGER.debug(f"Launcher configuration (including overrides):\n{json.dumps(self.config.to_dict(encode_json=True), indent=4)}")
        
        # Initialize console command parser
        self.console_parser = interface.ConsoleParser()
        self.cmd_queue = Queue()
        
        # Initialize Input Thread to handle console input later. Don't start thread just yet
        self.input_thread = interface.KeyboardThread(self.on_input, True)
        
        # Initialize thread for sending status updates to endpoint
        self.status_thread = interface.StatusUpdaterThread(self.config.status.EndpointURL, timeout=self.config.status.Interval, status=False)
        
        # Initialize notification objects
        self.notifications = interface.NotificationManager()
        
        self.notifications.add_handler(interface.LoggingNotificationHandler())
        
        if self.config.notifications.method == NotificationMethod.DISCORD:
            if self.config.notifications.discord.webhookURL:
                self.notifications.add_handler(interface.DiscordNotificationHandler(self.config.notifications.discord.webhookURL, name=self.config.notifications.name, event_whitelist=set(self.config.notifications.EventWhitelist)))
            else:
                LOGGER.warning("Discord Webhook URL is not set in config, not sending Discord notifications")
        elif self.config.notifications.method == NotificationMethod.NTFY:
            if self.config.notifications.ntfy.topic:
                self.notifications.add_handler(interface.NTFYNotificationHandler(self.config.notifications.ntfy.topic, ntfy_url=self.config.notifications.ntfy.serverURL, name=self.config.notifications.name, event_whitelist=set(self.config.notifications.EventWhitelist)))
            else:
                LOGGER.warning("ntfy topic is not set in config, not sending ntfy notifications")
        
        # Create Dedicated Server object
        self.dedicatedserver = AstroDedicatedServer(self)
    
    def check_ds_executable(self):
        """ Checks is Astroneer DS executable exists and is a file """
        
        execpath = os.path.join(self.config.AstroServerPath, DS_EXECUTABLE)
        
        return os.path.exists(execpath) and os.path.isfile(execpath)

    def on_input(self, input_string):
        """ Callback method to handle console input """
        
        # Parse console input
        success, result = self.console_parser.parse_input(input_string)
        
        if success:
            if result["cmd"] == interface.ConsoleParser.Command.HELP:
                # If it's a help command, we don't need to add it to the command queue as there is nothing to be done
                LOGGER.info(result["message"])
            else:
                # Add any other command to the command queue to be processed later
                self.cmd_queue.put(result)
        else:
            # If an error occured, {result} is just a message, so log it to console
            # We send event for command first, when it's processed
            LOGGER.warning(result)

    def update_wine_prefix(self):
        """
            Creates/updated the WINE prefix
        """
        
        LOGGER.debug("Ensuring WINE prefix is setup...")
        
        cmd = [self.wineexec, "wineboot"]
        env = os.environ.copy()
        
        # Remove DISPLAY environment variable to stop wine from creating a window
        if "DISPLAY" in env:
            del env["DISPLAY"]
        
        env["WINEPREFIX"] = self.config.WinePrefixPath
        env["WINEDEBUG"] = "-all"
        
        LOGGER.debug(f"Executing command '{' '.join(cmd)}' in WINE prefix '{self.config.WinePrefixPath}'...")
        
        try:
            wineprocess = subprocess.Popen(cmd, env=env, cwd=self.config.AstroServerPath, stderr=subprocess.PIPE, stdout=subprocess.PIPE, close_fds=True)
            code = wineprocess.wait(timeout=30)
        except TimeoutError:
            LOGGER.debug("Wine process took longer than 30 seconds, aborting")
            return False
        except Exception as e:
            LOGGER.error(f"Error occured during updating of wine prefix: {str(e)}")
            return False
        
        return code == 0    
    
    def check_network_config(self):
        if not self.dedicatedserver:
            raise ValueError("Dedcated Server has to be created first")
        
        LOGGER.info("Checking Network Configuration...")
        
        # Check if server port is reachable from local network over UDP
        server_local_reachable = net.net_test_local(self.dedicatedserver.ds_config.PublicIP, self.dedicatedserver.engine_config.Port, False)
        
        # Check if server post is reachable from internet over UDP
        server_nonlocal_reachable = net.net_test_nonlocal(self.dedicatedserver.ds_config.PublicIP, self.dedicatedserver.engine_config.Port)
        
        test_res = (server_local_reachable, server_nonlocal_reachable)
        
        if test_res == (True, True):
            LOGGER.info("Network configuration looks good")
        elif test_res == (False, True):
            LOGGER.warning("The Server is not accessible from the local network")
            LOGGER.warning("This usually indicates an issue with NAT Loopback")
        elif test_res == (True, False):
            LOGGER.warning("The server can be reached locally, but not from outside of the local network")
            LOGGER.warning(f"Make sure the Server Port ({self.dedicatedserver.engine_config.Port}) is forwarded for UDP traffic")
        elif test_res == (False, False):
            LOGGER.warning("The Server is completely unreachable")
            LOGGER.warning(f"Make sure the Server Port ({self.dedicatedserver.engine_config.Port}) is forwarded for UDP traffic and check firewall settings")
        
        rcon_local_blocked = not net.net_test_local(self.dedicatedserver.ds_config.PublicIP, self.dedicatedserver.ds_config.ConsolePort, True)
        
        if rcon_local_blocked:
            LOGGER.info("RCON network configuration looks good")
        else:
            LOGGER.warning(f"SECURITY ALERT: The RCON Port ({self.dedicatedserver.ds_config.ConsolePort}) is accessible from outside")
            LOGGER.warning("SECURITY ALERT: This potentially allows access to the Remote Console from outside your network")
            LOGGER.warning("SECURITY ALERT: Disable this ASAP to prevent issues")
            
            # kept from AstroLauncher
            time.sleep(5)
    
    def update_server(self):
        """
            Installs/Updates the Astroneer Dedicated Server.
            Also ensures that DepotDownloader is present
        """
        
        # If DepotDownloader executable doesn't exists yet, download it
        if not path.exists(self.depotdl_path):
            LOGGER.info("DepotDownloader not found, downloading...")
            steam.dl_depotdownloader(path.dirname(self.depotdl_path), path.basename(self.depotdl_path))
        
        LOGGER.info("Updating Astroneer Dedicated Server app from Steam...")
        success = steam.update_app(exec_path=self.depotdl_path, app="728470", os="windows", directory=self.config.AstroServerPath)
        
        self.buildversion = read_build_version(self.config.AstroServerPath)
        
        if success and (self.buildversion is not None):
            LOGGER.info(f"Sucessfully updated Astroneer Dedicated Server to version {self.buildversion}")
        else:
            LOGGER.error("Error while updating Astroneer Dedicated Server")
    
    def check_server_update(self, force_update=False):
        """
            Checks if an update for the Astroneer Dedicated Server is available or if it needs to be installed.
            Also performs update if set in config or {force_update} is set to True
        """
        
        oldversion = read_build_version(self.config.AstroServerPath)
        
        do_update = False
        installed = True
        
        if (oldversion is None) or not self.check_ds_executable():
            # No version is present yet or executable not present, we need an update/installation
            LOGGER.warning("Astroneer Dedicated Server is not installed yet")
            do_update = True
            installed = False
        else:
            # Get current server version from Spycibot endpoint
            try:
                data = json.load(get_request(ASTRO_SERVER_STATS_URL))
                newversion = data["stats"]["latestVersion"]
                
                if version.parse(newversion) > version.parse(oldversion):
                    LOGGER.warning(f"Astroneer Dedicated Server update available ({oldversion} -> {newversion})")
                    do_update = True
            except Exception as e:
                LOGGER.error(f"Error occured while checking for newest version: {str(e)}")
                LOGGER.warning(f"Trying server update blindly...")
                do_update = True

        if do_update:
            if self.config.AutoUpdateServer:
                if installed:
                    LOGGER.info("Automatically updating Astroneer Dedicated Server...")
                else:
                    LOGGER.info("Automatically installing Astroneer Dedicated Server...")
            
            if self.config.AutoUpdateServer or force_update:
                self.update_server()
            else:
                LOGGER.info("Not installing/updating automatically")
        else:
            if force_update:
                LOGGER.info("Noting to do")
            else:
                LOGGER.info("No update available, the Astroneer Dedicated Server is on the newest version")
        
    def start_server(self):
        """
            Starts the Astroneer Dedicated Server after setting up environment
        """
        
        # Check for and install DS update if wanted
        self.check_server_update()
        
        # If Playfab API can't be reached, we can't continue
        if not playfab.check_api_health():
            LOGGER.error("Playfab API is unavailable. Are you connected to the internet?")
            self.exit(reason="Playfab API unavailable")
        
        # Make sure wine prefix is ready
        if not self.update_wine_prefix():
            self.exit(reason="Error while updating WINE prefix")
        
        # Check that ports are available for the Server and RCON
        if not self.dedicatedserver.check_ports_free():
            self.exit(reason="Port not available")
        
        # Check netowrk configuration
        if self.config.CheckNetwork:
            self.check_network_config()
        
        LOGGER.debug("Starting input thread...")
        self.input_thread.start()
        
        # Prepare and start dedicated server
        try:
            if not self.dedicatedserver.start():
                return
        except Exception as e:
            LOGGER.error(f"There as an error while starting the Dedicated Server: {str(e)}")
            self.exit(reason="Error while starting Dedicated Server")
        
        LOGGER.info("Enter 'help' to get help about command usage")
        
        self.status_thread.update_status(status=True, message="Server is running")
        
        # If sending of status updates is enabled, start thread
        if self.config.status.SendStatus:
            LOGGER.info("Sending of status updates is enabled")
            self.status_thread.start()
        
        # Run Server Loop
        LOGGER.debug("Starting server loop...")
        self.dedicatedserver.server_loop()
    
    
    def user_exit(self, signal, frame):
        """ Callback for when user requests to exit the application """
        self.exit(graceful=True, reason="Received SIGINT signal")
    
    def exit(self, graceful=False, reason=None):
        if graceful:
            if reason:
                LOGGER.info(f"Quitting gracefully... (Reason: {reason})")
            else:
                LOGGER.info("Quitting gracefully...")
            
            if self.dedicatedserver and self.dedicatedserver.status in [ServerStatus.RUNNING, ServerStatus.STARTING]:
                # If no RCON is connected while running or starting, simply kill server
                if not self.dedicatedserver.rcon.connected:
                    self.dedicatedserver.kill()
                    return
                
                # If server is running, simply shut it down and return to let it finish normally
                LOGGER.debug("Shutting down Dedicated Server before quitting...")
                self.dedicatedserver.shutdown()
                return
            else:
                # If no server is running, exit directly
                LOGGER.info("Goodbye!")
                LOGGER.debug("Quitting with exit code 0...")
                sys.exit(0)
        else:
            if reason:
                LOGGER.info(f"Quitting... (Reason: {reason})")
            else:
                LOGGER.info("Quitting...")
            
            # Kill server if it's running or not
            if self.dedicatedserver:
                self.dedicatedserver.kill()
            
            # Give short time to send status update
            if self.status_thread:
                self.status_thread.update_status(status=False, message="Server was forcibly closed")
            
            time.sleep(0.1)
            
            sys.exit(1)

if __name__ == "__main__":
    # Exit directly, if python version below 3.9 is discovered
    if (sys.version_info.major < 3) or ((sys.version_info.major == 3) and (sys.version_info.minor < 9)):
        print()
        print("ERROR:   AstroTuxLauncher needs at least Python 3.9 to run properly!")
        print(f"        You are currently running version {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}.")
        print()
        sys.exit(1)
    
    # Parse command line arguments
    parser = argparse.ArgumentParser()
    parser.add_argument("command", type=LauncherCommand, action=interface.EnumStoreAction, help=HELP_COMMAND)
    parser.add_argument("-c", "--config_path", help="The location of the configuration file (default: %(default)s)", type=str, dest="config_path", default="launcher.toml")
    parser.add_argument("-p", "--astro_path", help="The path of the Astroneer Dedicated Server installation (default: %(default)s)", dest="astro_path", default=None)
    parser.add_argument("-d", "--depotdl_exec", help="The path to anm existing depotdownloader executable (default: %(default)s)", dest="depotdl_exec", default=None)
    parser.add_argument("-l", "--log_debug", help="Also log debug messages (Overrules config option)", action='store_true', dest="log_debug", default=False)
    
    args = parser.parse_args()
    
    
    # Set terminal window title
    if CONTROL_CODES_SUPPORTED is None:
        set_window_title(f"{NAME} - Unofficial Astroneer Dedicated Server Launcher for Linux")
    
    # Print Banner
    print(BANNER_LOGO, end="")
    print(BANNER_SUBTITLE)
    print("")
    print(BANNER_TEXT)
    print(f"v{LAUNCHER_VERSION}")
    print("")
    
    try:
        launcher = AstroTuxLauncher(args.config_path, args.astro_path, args.depotdl_exec, force_debug_log=args.log_debug)
    except KeyboardInterrupt:
        print("Quitting... (requested by user)")
        sys.exit(0)
    except Exception as e:
        print(f"Error while initializing launcher on line {sys.exc_info()[-1].tb_lineno}: {type(e).__name__}: {e}")
        print(traceback.format_exc())
        print("Quitting...")
        sys.exit(1)
    
    signal.signal(signal.SIGINT, launcher.user_exit)
    
    if CONTROL_CODES_SUPPORTED == False:
        LOGGER.debug("ANSI escape codes except color codes are disabled")
    
    LOGGER.debug(f"CLI Command: {args.command.value}")
    
    if args.command == LauncherCommand.INSTALL:
        LOGGER.info("Installing Astroneer Dedicated Server...")
        try:
            launcher.update_server()
        except Exception as e:
            LOGGER.critical(f"Error while installing server on line {sys.exc_info()[-1].tb_lineno}: {type(e).__name__}: {e}")
            LOGGER.error(traceback.format_exc())
            sys.exit(1)
    elif args.command == LauncherCommand.UPDATE:
        LOGGER.info("Checking for available updates to the Astroneer Dedicated Server...")
        
        try:
            launcher.check_server_update(force_update=True)
        except Exception as e:
            LOGGER.critical(f"Error while updating server on line {sys.exc_info()[-1].tb_lineno}: {type(e).__name__}: {e}")
            LOGGER.error(traceback.format_exc())
            sys.exit(1)
    elif args.command == LauncherCommand.START:
        try:
            launcher.start_server()
        except Exception as e:
            if launcher.dedicatedserver:
                launcher.dedicatedserver.kill()
            
            raise
    
    LOGGER.info("Goodbye!")