path: root/__main__.py

#!/usr/bin/python3
#####################################
## This is the Tkinter version
## It is mostly a TMW Vault Wrapper
##
## For lore, you need to use the app
#####################################
## (C) The Mana World Team, 2021-2024
## Published under the MIT License
#####################################

import requests, json, traceback, subprocess, hmac, struct, time, os, uuid, base64
import tkinter as tk
from tkinter.messagebox import showinfo, showerror, askyesno
from functools import partial

OBFS = uuid.getnode() % 256 # Get a number which *usually* doesn't change
print("Obfuscation key = %d" % OBFS) # <- Just to not leave it in plain text

## Internal wrapper for preferences
saveSettings = True
def _savePref():
    global saveSettings, pref
    if not saveSettings:
        return
    dump = json.dumps(pref)
    #print(repr(dump))
    dump = bytearray(ord(x) ^ int(OBFS) for x in dump)
    #print(repr(dump))
    dump = base64.b64encode(dump)
    dump = dump.decode('ascii')
    #print(repr(dump))
    with open("settings.b64", 'w') as raw:
        raw.write(str(dump))
    #print("Preferences updated...")
    return

## Try to load previously saved settings
try:
    with open("settings.b64") as raw:
        pref = base64.b64decode(raw.read())
        #print(repr(pref))
        #print(pref[0])
        #print(int(pref[0])) # <- XXX: use ord() if fail, and replace accordingly
        #print(int(pref[0]) ^ int(OBFS))
        pref = bytearray(int(x) ^ int(OBFS) for x in pref)
        #print(repr(pref))
        pref = pref.decode('utf8')
        pref = json.loads(str(pref))
        #print(repr(pref))
        print("User settings loaded...")
except:
    print("Failed to load user settings!")
    err = traceback.format_exc().split("\n")[-2]
    print("Error: %s" % err)
    ## Create new dummy prefs so it doesn't crash
    pref = {"user": "",
            "pass": "",
            "totp": "",
            "mana": False,
            "plus": False,
            "shell": True}
    ## If it is an error, give an option to not destroy the old settings
    if ("FileNotFoundError" in err):
        _savePref()
        print("New user settings created!")
    elif (askyesno("Mana Launcher", "Failed to load user settings!\n\nDo you want to create a new one?")):
        _savePref()
        print("New user settings created!")
    else:
        saveSettings = False

## Prepare some basic stuff
execute=subprocess.call
serverlist = []
class HoverButton(tk.Button):
    def enter(self, e):
        self["background"] = self["activebackground"]
    def exit(self, e):
        self["background"] = self.defaultBackground

    def __init__(self, **kw):
        tk.Button.__init__(self, **kw)
        self.defaultBackground = self["background"]
        self.bind("<Enter>", self.enter)
        self.bind("<Leave>", self.exit)

## TODO: Get from `sys` args parameters like world auto-selection
## TODO: Use Config and Local folders, not sure if relative path works

#####################################
#VAULT_HOST = "https://localhost:13370"
#SERVERLIST = "http://localhost/launcher"
VAULT_HOST = "https://api.themanaworld.org:13370"
SERVERLIST = "https://updates.tmw2.org/launcher"
# Vault SSL wrapper
vault=requests.Session()
if "localhost" in VAULT_HOST:
    vault.verify=False
#####################################

def stdout(message, bd=False):
    if bd:
        print("\033[1m%s\033[0m" % message)
    else:
        print(message)

def sdelay(delta=0.02):
    time.sleep(delta)
    return

# IF Then Else (IFTE)
def ifte(ifs, then, elses):
    if (ifs):
        return then
    else:
        return elses

# Sanitize a command (strip some flow control chars)
# While it covers all control operators and most metacharacters,
# it doesn't covers well the reserved words.
# ...Of course, it relies on this client not being compromised.
def san(cmd):
    return cmd.replace(";", "").replace("|", "").replace(">", "").replace("<", "").replace("&", "").replace("(", "").replace(")", "").replace("\n", "").replace("[[", "").replace("]]", "")

# Returns number of seconds since UNIX EPOCH
def now():
    return int(time.time())

# Search for array[?][key]==search in an array of dicts
# Returns the dictionary, or returns None
def dl_search(array, key, search):
    try:
        r=(item for item in array if item[key] == search).next()
    except:
        r=None
    if r is None:
        stdout("dlsearch: r is None")
    return r

# Search for array[?][key]==search in an array of dicts
# Returns the index, or returns -1
def dl_search_idx(array, key, search):
    try:
        r=next((i for i, item in enumerate(array) if item[key] == search), None)
    except:
        traceback.print_exc()
        r=-1
    return ifte(r is None, -1, r)

def nullp(arg):
    return

def validate_server(srv):
    name="Unknown Server"; ok=False
    try:
        name=srv["Name"]
        #stdout("Validating server \"%s\"..." % name)
        nullp("Host: %s" % srv["Host"])
        nullp("Port: %04d" % srv["Port"])
        nullp("Desc: %s" % srv["Desc"])
        nullp("Link: %s" % srv["Link"])
        nullp("News: %s" % srv["News"])
        nullp("Back: %s" % srv["Back"])
        nullp("UUID: %s" % srv["UUID"])
        nullp("Help: %s" % srv["Help"])
        nullp("Online List: %s" % srv["Online"])
        nullp("Policy: %s" % srv["Policy"])
        if not "Type" in srv.keys():
            srv["Type"]="evol2"
        stdout("Server \"%s\" loaded." % name)
        ok=True
    except:
        traceback.print_exc()
        stdout("Validation for server \"%s\" FAILED!" % name)

    return ok

def update_serverlist(hosti):
    global serverlist, host
    try:
        r=requests.get("%s/server_list.json" % hosti, timeout=10.0)
        if (r.status_code != 200):
            raise AssertionError("Mirror %s seems to be down!\nReturned error %03d\n" % (hosti.replace("https://", "").replace("http://", ""), r.status_code))

        j=json.loads(r.text)

        ## If we reached here, then everything is likely fine
        ## We can now build the persistent data
        print("Fetching server list and assets...")
        slist=len(j)
        for server in j:
            if (validate_server(server)):
                serverlist.append(server)

        ## If we were truly successful, save this host as our mirror
        host = hosti
        return True
    except:
        traceback.print_exc()
        return False

def launch_game(idx):
    ## Get/Set basic data
    HOST=serverlist[idx]["Host"]
    PORT=serverlist[idx]["Port"]
    CMD=""
    OPT=""
    PWD=""

    ## Get credentials
    auth = {"vaultId": vaultId,
            "token": vaultToken,
            "world": serverlist[idx]["UUID"]}

    try:
        r=vault.post(VAULT_HOST+"/world_pass", json=auth, timeout=15.0)
        ## We got the access credentials
        if r.status_code == 200:
            auth2=r.json()
            PWD=" -U %s -P %s" % (auth2["user"], auth2["pass"])
        ## We were refused by Vault
        elif r.status_code == 403:
            stdout("Warning: Connection was refused (Vault logout?)")
            return -1
        ## We are rate-limited, try again
        elif r.status_code == 429:
            stdout("Rate limited!")
            return -1
        ## Internal error, maybe?
        else:
            stdout("Get World Auth - Returned error code %d" % r.status_code)
            return -1
    except:
        traceback.print_exc()
        return -1

    ## Handle server type
    ## TODO: Do not use system-wide install unless specified so in prefs
    if not sys.platform.startswith('win'):
        CMD="manaplus"
    else:
        CMD="manaplus/Mana/manaplus.exe" # FIXME untested
    OPT="-s %s -y %s -p %s -S" % (HOST, serverlist[idx]["Type"], PORT)

    ## Execute the app
    if pref["shell"]:
        app=execute(san("%s %s%s" % (CMD, OPT, PWD)), shell=True) # nosec
    else:
        app=execute(san("%s %s%s" % (CMD, OPT, PWD)), shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # nosec

    return app

## Only meaningful on Linux
os.environ["APPIMAGELAUNCHER_DISABLE"]="1"

#################################################################################
## Construct session data
update_serverlist(SERVERLIST)
vaultId = -1
saveMail = None
savePass = None

#################################################################################
###### World Selection Screen
def world_select():
    global canva, serverlist
    canva = tk.Canvas(root, width=400, height=600, bg="#0c3251")
    canva.pack()
    label1 = tk.Label(root, text='World Selection', bg="#0c3251", fg="#fff")
    label1.config(font=('helvetica', 14))
    canva.create_window(200, 20, window=label1)

    # Not really necessary? Or just TODO?
    #scrollbar = tk.Scrollbar(canva, orient=tk.VERTICAL)
    #canva.create_window(400, 20, window=scrollbar)

    ## Create a list of all worlds
    ypos = 60
    for w in serverlist:
        ## TODO: Do not block main thread, launch this in a threading
        button = HoverButton(text=w["Name"], command=partial(launch_game, serverlist.index(w)), bg="#cc6600", fg="#fff", activebackground="#ee9933")
        ## TODO: Handle MLP (on the threading?)
        ## TODO: First login greeting?
        canva.create_window(200, ypos, window=button)
        ypos += 40

    return

#################################################################################
## Login function
def login():
    global canva, vaultId, vaultToken, saveMail, savePass
    email = entry1.get()
    passw = entry2.get()
    totps = entry3.get()

    ## Save email or delete saved copy
    if saveMail.get():
        pref["user"] = email
    else:
        pref["user"] = ""

    ## Save password or delete saved copy
    if savePass.get():
        pref["pass"] = passw
    else:
        pref["pass"] = ""

    ## Prepare the packet for Vault
    data = {"mail": email,
              "pass": passw,
              "totp": totps[:6]
             }
    try:
        r = vault.post(VAULT_HOST+"/user_auth", json=data)

        if (r.status_code != 200):
            add=""
            if r.status_code == 400:
                add="\nBad request.\n"
            elif r.status_code == 401:
                add="\nIncorrect username/password.\n"
            elif r.status_code == 429:
                add="\nYou are being rate-limited.\n"
            showerror(title="Mana Launcher", message="Vault returned error %d\n%s\nPlease try again later." % (r.status_code, add))
            exit(1)
    except SystemExit:
            exit(1)
    except:
        err = traceback.format_exc().split("\n")[-2]
        print("Error: %s" % err)
        showerror(title="Mana Launcher", message="Python error\n\nPlease try again later.")
        exit(1)

    auth2 = r.json()
    vaultId = auth2["vaultId"]
    vaultToken = auth2["token"]

    ## Only save settings if connection worked
    _savePref()

    ## Change the display
    print("Connected to vault!")
    canva.destroy()
    ## TODO: Do not jump to world selection if it is a new account
    world_select()
    return

#################################################################################
## Build Tkinter interface
root=tk.Tk()
root.title("Mana Launcher (tk)")
canva = tk.Canvas(root, width=400, height=600, bg="#0c3251")
canva.pack()

## Default variables
saveMail = tk.BooleanVar()
saveMail.set(pref["user"] != "")
savePass = tk.BooleanVar()
savePass.set(pref["pass"] != "")

## Email
label1 = tk.Label(root, text='Email:', bg="#0c3251", fg="#fff")
label1.config(font=('helvetica', 14))
canva.create_window(200, 40, window=label1)
entry1 = tk.Entry(root)
entry1.insert(0, pref["user"])
canva.create_window(200, 80, window=entry1)
c1 = tk.Checkbutton(root, text="Remember", variable=saveMail, bg="#0c3251", fg="#f70")
canva.create_window(350, 80, window=c1)


label2 = tk.Label(root, text='Password:', bg="#0c3251", fg="#fff")
label2.config(font=('helvetica', 14))
canva.create_window(200, 120, window=label2)
entry2 = tk.Entry(root, show="*")
entry2.insert(0, pref["pass"])
canva.create_window(200, 160, window=entry2)
c1 = tk.Checkbutton(root, text="Remember", variable=savePass, bg="#0c3251", fg="#f70")
canva.create_window(350, 160, window=c1)


label3 = tk.Label(root, text='TOTP:', bg="#0c3251", fg="#fff")
label3.config(font=('helvetica', 14))
canva.create_window(200, 200, window=label3)
entry3 = tk.Entry(root)
#entry3.insert(0, pref["totp"])
canva.create_window(200, 240, window=entry3)


button1 = HoverButton(text='Login', command=login, bg="#cc6600", fg="#fff", activebackground="#ee9933")
canva.create_window(200, 300, window=button1)
root.mainloop()

# Check if you're now logged in
if vaultId < 1:
    exit(1)

print("Thanks for playing!")

exit(0) # <- FIXME: Not necessary? Just delete stuff below

#################################################################################
try:
    while True:
        command=str(input("> "))
        # No command inserted? Do nothing
        if command == "":
            continue
        # We have a command, prepare it for processing
        stdout("CONSOLE: %s" % command)
        cmd=command.split(' ')[0].lower()
        com=" ".join(command.split(' ')[1:])
        # Parse the command
        # TODO: grant gems to an user
        if cmd in ["help", "h"]:
            print("help - Shows this help text")
            print("list - List servers")
            print("exit - Closes the program")
            print("connect <world> - Connects to <world>")
            print("")
            print("You need ManaPlus installed for Evol2 worlds.")
        elif cmd in ["list", "servers"]:
            for w in serverlist:
                print("\033[1m%s\033[0m" % w["Name"])
                print("Type: %s" % w["Type"])
                print("Help: %s" % w["Help"])
        elif cmd in ["exit", "quit", "close"]:
            break
        elif cmd in ["connect", "run", "world"]:
            target=dl_search_idx(serverlist, "Name", com)
            print("Target \"%s\" - %s" % (com, str(target)))
            try:
                idx=int(target)
            except:
                stdout("ERROR: Unknown world.")
                continue

            app=launch_game(idx)

            ## TODO: Handle MLP
            while app == 7:
                stdout("[CLIENT] Mirror Lake triggered.")
                r=vault.post(VAULT_HOST+"/getlake", json=auth, timeout=15.0)
                if r.status_code == 200:
                    goto=r.json()["world"]
                    stdout("MLP Target: %s" % str(goto))
                    if goto == "" or goto.lower() == "vault":
                        stdout("Mirror Lake False Positive")
                        break
                    try:
                        idx=int(dl_search_idx(serverlist, "UUID", goto))
                        app=launch_game(idx)
                    except:
                        traceback.print_exc()
                        break
        else:
            stdout("ERROR: Unknown command. Try \"help\".")
except:
    stdout("Abrupt error: Terminating!")
    traceback.print_exc()