#!/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, getpass
import base64
import tkinter as tk
from tkinter.messagebox import showinfo, showerror, askyesno
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 FileNotFoundError: # <-- TODO
except:
print("Failed to load user settings!")
err = traceback.format_exc().split("\n")[-2]
print("Error: %s" % err)
# TODO: Move this to a function of its own, dont ask if FileNotFoundError
# Create new dummy prefs
pref = {"user": "",
"pass": "",
"totp": "",
"mana": False,
"plus": False,
"shell": True}
## Do you want us to create a settings file?
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
execute=subprocess.call
serverlist = []
## 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
CMD="manaplus"
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
###### TODO: World Selection Screen
def world_select():
global canva
canva = tk.Canvas(root, width=400, height=600)
canva.pack()
label1 = tk.Label(root, text='You are a winner')
label1.config(font=('helvetica', 14))
canva.create_window(200, 40, window=label1)
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()
world_select()
return
#################################################################################
## Build Tkinter interface
root=tk.Tk()
canva = tk.Canvas(root, width=400, height=600, bg="#0c3251") # TODO: Coloring
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="#fff")
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="#fff")
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 = tk.Button(text='Login', command=login, bg="#cc6600", fg="#fff")
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)
## 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()