SatiAPI/SatiAPI.py

#!/usr/bin/env python3
"""
https://github.com/wolveix/satisfactory-server/wiki/Official-API-Docs
https://satisfactory.wiki.gg/wiki/Dedicated_servers/HTTPS_API
"""

import requests
import logging
import json
import yaml
from os.path import exists

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class SatiAPI:
    """A API wrapper for Satisfactory dedicated server"""
    __AUTHOR__      = 'anima'
    __VERSION__     = '0.9.0'

    def __init__(self, host: str = None, port: int = 7777, token: str = None, conffile: str = 'conf.yml', logfile: str = 'SatiAPI.log', loglevel: int = 20) -> None:
        """create a wrapper for satisfactory dedicated server
        
        :param host: host ip or dns
        :type host: string
        :param port: port of host
        :type port: int | str
        :pram token: token for auth on server (can be get with `get_token`)
        :type token: string
        :param conffile: path to config file in yaml format
        :type conffile: string
        :param logfile: path to logfile
        :type logfile: string
        :param loglevel: level of logging (10 => debug, 20 => info (default), 30 => warning, 40 => error)

        all server specific settings can be set via config file
        """
        self._log                       = logging.getLogger(__name__)
        logHandler                      = logging.FileHandler(logfile)
        logPrinter                      = logging.StreamHandler()
        logFormat                       = logging.Formatter('%(asctime)s %(levelname)s \t %(name)s : %(message)s')
        logHandler.setFormatter(logFormat)   # Format for File
        logPrinter.setFormatter(logFormat)   # Format for Console
        self._log.addHandler(logHandler)
        self._log.addHandler(logPrinter)
        self._log.setLevel(loglevel)

        self.host = host
        self.port = port
        self.token = token
        self.__map_config(conffile)

    ## propertys
    @property
    def host(self):
        return self.__host
    
    @host.setter
    def host(self, value) -> bool:
        """setter method for host. check if value valid. 

        :param value: host ip or dns to dedicated server
        :type value: string
        :return: true if set successfull
        :rtype: bool
        """
        if isinstance(value, str) or value is None:
            self.__host = value
            return True
        else:
            self._log.debug(f'wrong {value=} ({type(value)}) for host')
            return False

    @property
    def port(self):
        return self.__port
    
    @port.setter
    def port(self, value) -> bool:
        """setter method for port. check if value a valid network port

        :param value: port of dedicated server instance
        :type value: int | str
        :return: true if set successfull
        :rtype: bool
        """
        if not isinstance(value, int):
            if isinstance(value, str):
                try:
                    value = int(value)
                except ValueError:
                    self._log.debug(f'wrong {value=} ({type(value)}) for port')
                    return False
            else:
                self._log.debug(f'wrong {value=} ({type(value)}) for port')
                return False
        
        if value > 0 and value < 65535:
            self.__port = value
            return True
        else:
            self._log.debug(f'out of port range {value=}')
            return False

    @property
    def token(self):
        return self.__token
    
    @token.setter
    def token(self, value) -> bool:
        """setter method for token. check if value a valid token
        
        :param value: token for auth on server
        :type value: string
        :return: true if set successfull
        :rtype: bool
        """
        if isinstance(value, str) or value is None:
            self.__token = value
            return True
        else:
            self._log.debug(f'wrong {value=} ({type(value)}) for token')

    ## methods
    def __map_config(self, conffile) -> None:
        """map config settings with multible usage in this class override only default values not changed.
        
        :param conffile: path to config file in yaml format
        :type conffile: string
        """
        if exists(conffile):
            with open(conffile, 'r') as f:
                self._config = yaml.safe_load(f.read())

            if 'host' in self._config and self.host is None:
                self.host = self._config['host']
            
            if 'port' in self._config and self.port == 7777:
                self.port = self._config['port']

            if 'auth' in self._config:
                if 'Administrator' in self._config['auth']:
                    if 'token' in self._config['auth']['Administrator'] and self.token is None:
                        self.token = self._config['auth']['Administrator']['token']

    def _get_config_password(self, privilegeLevel: str = 'Administrator') -> str | None:
        """internal helper method to get password from config file
        
        :param privilegeLevel: level of needed privilege
        :type privilegeLevel: string
        :return: password if exists
        :rtype: string | None
        """
        if 'auth' in self._config:
            if privilegeLevel in self._config['auth']:
                if 'password' in self._config['auth'][privilegeLevel]:
                    return self._config['auth'][privilegeLevel]['password']

    def __query(self, query: str, data: dict = None, auth: bool = False) -> dict | bool | None:
        """run a query to satisfacoty dedicated server
        
        :param query: name of querytype
        :type query: string
        :param data: (maybe optional) data to be send as dict
        :type data: dict
        :param auth: set true if authentication requiered
        :type auth: bool
        :return: response from server
        :rtype: dict | None | bool
        """
        payload = dict()
        payload['function'] = query
        
        if data is not None:
            payload['data'] = data

        headers = dict()
        headers['Content-Type'] = 'application/json'
        if auth:
            if self.token is None:
                self.get_token()
                
            if self.token is not None:
                headers['Authorization'] = f'Bearer {self.token}'
            else:
                self._log.error('no data to authencitate')
                return False

        response = requests.post(f'https://{self.host}:{self.port}/api/v1', headers=headers, json=payload, verify=False)
        if response.status_code == 200:
            json_response = json.loads(response.text)
            if 'data' in json_response:
                return json_response
            else:
                # if json_response['errorMessage'] == '':
                        # self._log.debug(f'token expired')
                        # self.get_token()
                self._log.error(f'wrong response data {json_response=}')

        else:
            self._log.error(f'non successfull response {response.content} [{response.status_code}]')
            self._log.debug(f'{response.request.url=}')
            self._log.debug(f'{response.request.headers=}')
            self._log.debug(f'{response.request.body=}')
            return False

    def _save_token(self, response: dict) -> bool:
        """internal helper method to save token from response
        
        :param response: response after query
        :type response: dict
        :return: true if token successfull saved
        :rtype: bool
        """
        if response:
            if 'authenticationToken' in response['data']:
                self.token = response['data']['authenticationToken']
                return True
        return False


    def _passwordless_login(self, privilegeLevel: str = 'InitialAdmin') -> bool:
        """Attempts to perform a passwordless login to the Dedicated Server as a player. Passwordless login is possible if the Dedicated Server is not claimed, or if Client Protection Password is not set for the Dedicated Server. This function requires no Authentication.
        
        :param privilegeLevel: Minimum privilege level to attempt to acquire by logging in. See Privilege Level enum for possible values
        :type privilegeLevel: string
        :return: true if successfull
        :rtype: bool
        """
        data = dict()
        data['MinimumPrivilegeLevel'] = privilegeLevel

        response = self.__query('PasswordlessLogin', data)
        return self._save_token(response)

    def claim_server(self, servername: str, password: str = None) -> bool:
        """Claims this Dedicated Server if it is not claimed. Requires InitialAdmin privilege level, which can only be acquired by attempting passwordless login while the server does not have an Admin Password set, e.g. it is not claimed yet. Function does not return any data in case of success, and the server is claimed. The client should drop InitialAdmin privileges after that and use returned AuthenticationToken instead, and update it's cached server game state by calling QueryServerState.
        
        :param servername: New name of the Dedicated Server
        :type servername: string
        :param password: Admin Password to set on the Dedicated Server, in plaintext. If empty the password will cread from conf file
        :type password: string
        :return: true if claim successfull
        :rtype: bool

        after claim it get a new token automaticly
        """
        if self._passwordless_login():
            data = dict()
            if password is None: 
                password = self._get_config_password()
                self._log.debug('password used from conf file')

            if password is None: 
                self._log.error('no password for server claim set')
                return False
            data['ServerName']      = servername
            data['AdminPassword']   = password

            response = self.__query('ClaimServer', data, True)
            return self._save_token(response)
        else: 
            self._log.error('server can not be clamed because no authentication possable')
            return False

    def set_admin_password(self, password: str = None):
        """Updates the currently set Admin Password. This will invalidate all previously issued Client and Admin authentication tokens. Requires Admin privileges. Function does not return any data on success.
        
        :param password: the new passwort to set will be read from config file it not set
        :type password: string
        :return: true if new password successfull set
        :rtype: bool
        """
        if password is None: 
            password = self._get_config_password()

        if password is None:
            self._log.error('can not set new admin password because no passwort is set')
            return False
        
        data = dict()
        data['Password'] = password
        response = self.__query('SetAdminPassword', data, True)
        return self._save_token(response)

    def set_client_password(self, password: str = None):
        """Updates the currently set Client Protection Password. This will invalidate all previously issued Client authentication tokens. Pass empty string to remove the password, and let anyone join the server as Client. Requres Admin privileges. Function does not return any data on success.

        :param password: the new passwort to set will be read from config file it not set
        :type password: string
        :return: true if new password successfull set
        :rtype: bool
        """
        if password is None: 
            password = self._get_config_password('Client')

        if password is None:
            self._log.error('can not set new client password because no passwort is set')
            return False
        
        data = dict()
        data['Password'] = password
        response = self.__query('SetClientPassword', data, True)
        return True

    def get_token(self, password: str = None, privilegeLevel: str = 'Administrator') -> bool:
        """get token from satisfacory dedicated server, password is needed!
        Attempts to log in to the Dedicated Server as a player using either Admin Password or Client Protection Password. This function requires no Authentication.
        
        :param password: password of admin level
        :type password: string
        :param privilegeLevel: level of authentication
        :type privilegeLevel: string
        :return: true if successfull get a token
        :rtype: bool
        """
        if self.token is not None: 
            self._log.warning('you have already a token')
        if isinstance(password, str):
            self._log.debug('password used from parameter')
        else:
            password = self._get_config_password()
            if password is not None:
                self._log.debug('password used from conf file')
            else:
                self._log.error(f'no password for auth given!')
                return False

        data = dict()
        data['password'] = password
        data['MinimumPrivilegeLevel'] = privilegeLevel

        response = self.__query('PasswordLogin', data)
        return self._save_token(response)


    def get_status(self) -> dict:
        """Retrieves the current state of the Dedicated Server. Does not require any input parameters.
        
        :return: full return from server
        :rtpye: dict
        """
        return self.__query('QueryServerState', auth=True)

    def get_health(self, ClientCustomData: str = '') -> dict:
        """get health check
        
        :param ClientCustomData: Custom Data passed from the Game Client or Third Party service. Not used by vanilla Dedicated Servers
        :type ClientCustomData: string
        :return: full return from server
        :rtpye: dict
        """
        data = dict()
        data['ClientCustomData'] = ClientCustomData
        
        return self.__query('HealthCheck', data)

    def get_server_options(self) -> dict:
        """Retrieves currently applied server options and server options that are still pending application (because of needing session or server restart) Does not require input parameters.
        
        :return: full return from server
        :rtpye: dict
        """
        return self.__query('GetServerOptions')

    def get_advanced_game_settings(self):
        """Retrieves currently applied advanced game settings. Does not require input parameters.
        
        :return: full return from server
        :rtpye: dict
        """
        return self.__query('GetAdvancedGameSettings')
    
    
    
    ##  todo
    #RenameServer
    #SetClientPassword
    #SetAutoLoadSessionName
    #ApplyServerOptions
    #ApplyAdvancedGameSettings
    #RunCommand
    #Shutdown
    #CreateNewGame
    #SaveGame
    #DeleteSaveFile
    #DeleteSaveSession
    #EnumerateSessions
    #LoadGame
    #UploadSaveGame
    #DownloadSaveGame

if __name__ == "__main__":
    sati = SatiAPI(loglevel=10)
    # print('Verbundene Spieler: ', sati.get_status()['serverGameState']['numConnectedPlayers'])
    print(sati.get_status())
    print(sati.token)
    print(sati.set_client_password())