#!/usr/bin/env python3 """ODG for Linux. Installs and updates the Operator Development Group pack for OPERATOR on Linux, where Steam runs the game through Proton. It does the job ODG Ops does on Windows: the same releases, the same checksums and the same stamp files, so a game folder that has seen both stays consistent. python3 odg-linux.py install ODG, or update it python3 odg-linux.py help every command Kept in step with the Windows launcher (odg-launcher/src): update.rs for the release rules, melonsetup.rs for the loader, stale.rs and uninstall.rs for what may be deleted, access.rs for accounts, terms.rs for the terms. """ import errno import hashlib import http.client import json import os import re import shutil import stat import sys import time import urllib.error import urllib.parse import urllib.request import zipfile from pathlib import Path ODG_LINUX_TOOL = True # THE LINUX VERSION is the public release this copy went out with: ODG for Linux 2.2.3 shipped # beside ODG 2.2.3. A Linux-only fix between releases adds a fourth number (2.2.3.1). Every public # release moves this to the new number, then stage-site.py and a site deploy publish it; copies # already installed replace themselves on their next run (compare() reads "1" < "2.2.3"). TOOL_VERSION = "2.2.3.1" APPID = "1913370" PUBLIC_API = "https://api.github.com/repos/ArchdukePierre/operator-development-group/releases/latest" SITE = "https://operatordevgroup.com" TOOL_KEY = "odg-linux" DEFAULT_SERVER = "https://account.operatordevgroup.com" USER_AGENT = "ODGModManager" GITHUB = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"} LAUNCH_OPTION = 'WINEDLLOVERRIDES="version=n,b" %command%' GRACE_DAYS = 10 OFF = ".odg-off" HELD = "version.dll.odg-hold" STAMP = "Mods/.installed_release" MAP_ROOT = "Mods/MapLoader_Assets" MAP_STAMP = "Mods/MapLoader_Assets/.installed_mapdata" SHIPPED_LIST = "Mods/.odg-shipped-files.txt" OBSOLETE_LIST = "Mods/.odg-obsolete.txt" PREV_LIST = "Mods/.odg-shipped-files.prev" FORGOTTEN = "Mods/.installed_release.forgotten" DEV_HOLD = "Mods/.odg-dev-hold" STAGING = "UserData/ODGModManager/cache" CUSTOM = "Mods/MapLoader_Assets/Custom" # What "the mod loader is installed" means: the proxy, the loader, its runtime # config, and the private .NET 6 it starts. Under Proton that runtime is what # MelonLoader uses, so nothing has to be installed into the Proton prefix. LOADER_FILES = [ "version.dll", "MelonLoader/net6/MelonLoader.dll", "MelonLoader/net6/MelonLoader.runtimeconfig.json", "MelonLoader/Dependencies/dotnet/dotnet.exe", ] # The launcher's names for ODG's mods (mods.rs). A reason marks a mod that stays on. KNOWN = { "maploader.mod.dll": ("ODG Map Loader", "Carries the ODG Ops tab and every custom map. Off, the operations are gone."), "ammolock.mod.dll": ("ODG AmmoLock", "Keeps ammunition fair for everyone in an ODG lobby, so it stays on."), "diagnostics.mod.dll": ("ODG Diagnostics", "Records what happened when something breaks. Off, a crash report says nothing."), "hsr.mod.dll": ("M200 Intervention", None), "m107.mod.dll": ("Barrett M107A1", None), "vks.mod.dll": ("VKS", None), "fireteamroles.mod.dll": ("Fireteam Roles", None), "airborneinfil.mod.dll": ("Airborne Infiltration", None), "newslots.mod.dll": ("New Slots", None), "loadoutpresets.mod.dll": ("Loadout Presets", None), "quartermaster.mod.dll": ("Quartermaster", None), "thirdperson.mod.dll": ("Third Person", None), "operatorcompass.mod.dll": ("Operator Compass", None), "spectatorname.mod.dll": ("Spectator Name", None), "nofatigue.mod.dll": ("No Fatigue", None), "hostpin.mod.dll": ("Host Pin", None), "infilmarker.mod.dll": ("Infil Markers", None), "dlss5.mod.dll": ("DLSS 5", None), } # Retired files earlier packs left behind (cleanup.rs). Exact names in Mods only. RETIRED_FILES = ["ODGGuard.Mod.dll", "ODGGuard.Mod.dll.odg-off", "manifest.odgsig", "Weather.Mod.dll", "Weather.Mod.dll.odg-off", "Weather.log"] RETIRED_DIRS = ["Weather_Assets"] # Files the game or the mods write into the pack's folders (stale.rs). Never leftovers. ALLOWLISTED = {".installed_release", ".installed_mapdata", ".modpack_version", "door_placements.json", "ai_skeleton.json", "thumbs.db", "desktop.ini"} UNINSTALL_MARKERS = [ "Mods/.odg-shipped-files.txt", "Mods/.odg-shipped-files.prev", "Mods/.odg-obsolete.txt", "Mods/.installed_release", "Mods/.installed_release.forgotten", "Mods/.modpack_version", "Mods/MapLoader_Assets/.installed_mapdata", ] CUSTOM_README = ( "Put your own maps in here. One folder per map.\n" "\n" "A map is the folder the build tool wrote, copied in whole:\n" "it has manifest.bin and gameplay.bin at the top, and meshes and textures\n" "folders next to them. If manifest.bin is not there it is not a map yet.\n" "\n" "Do not put maps next to this folder. Anything up there that is not part of\n" "the pack gets removed on the next update. In here is safe.\n" ) # THE TERMS, word for word as the Windows launcher shows them (terms.rs). The # version is the contract: change a word there, change it here, and move both. TERMS_VERSION = "2026-09-28" TERMS_URL = "https://operatordevgroup.com/terms" TERMS = [ ("WE ARE NOT THE GAME", "Operator Development Group is a group of players. We are not connected to the people who make or publish OPERATOR, and nothing here is official. If the game itself breaks, that is theirs. If something we made breaks, tell us and we will fix it."), ("WHAT IT DOES TO YOUR GAME FOLDER", "It installs the mod pack into your OPERATOR folder, and it renames mod files in place to turn them on and off. It installs the mod loader if you do not have one. It removes files that an older pack of ours left behind, and it asks first. When you move between an early build and the public one, it removes what only the other build had. If you press INSTALL DLSS 5, it runs the pack's installer, which downloads DLSS 5's files from the people who make them. SETTINGS > UNINSTALL takes all of it off again, the launcher included. It never touches your saves, and it never deletes anything it did not put there. Your own mods and your own maps are left alone, always."), ("WHAT LEAVES YOUR PC", "For the free release: nothing. It downloads from GitHub and tells us nothing else. For a subscriber: your launcher key and the launcher version, to our server, so it can answer whether your subscription is live and which build your tier is on. No telemetry, no game files, no saves, no hardware. A crash report leaves only when you press SEND on the REPORT page, and only from an account signed in with Steam. It carries the logs the game and the mods wrote, your mod settings, the names of the files in your game folder, a description of your PC (processor, graphics card and driver, memory, motherboard), what Windows recorded about the game crashing, your account name, the Steam account you signed in with, which Steam account your PC is on, your Discord name if you add it, and the crash dump if you leave it in. Your Windows user name and PC name are blanked out first. A message you send on a report goes the same way. SAVE THE ZIP INSTEAD shows you exactly what would go. While you are signed in with Steam, the launcher also tells our server which Steam account owns OPERATOR on your PC, so a launcher passed to somebody else shows up."), ("IF YOU SIGN IN WITH STEAM", "You sign in on Steam's own page, and we never see your Steam password. Steam tells us the number of the account you signed in with. From that account's public profile we read its name, and whether it is a limited account or has a VAC ban on record, and we keep those with your account. Sending a crash report needs it."), ("IF YOU SIGN IN WITH PATREON", "Patreon tells us your member id, the name on your account, whether you are an active patron, what you are entitled to, and whether your last payment went through. We keep those so the launcher knows a subscription is live and paid. We never see your card, and we do not ask Patreon for your address. If you would rather send us none of that, use an invite code instead; it tells us only the name and tier on the code."), ("WHAT WE KEEP ABOUT YOU", "One row: a name, your tier, whether access is live, a one-way hash of your key, your Patreon member id if you used that, the pledge amount, when your launcher last checked in, its version, which version of these terms you accepted, and your GitHub username if you asked for the developer repository, and the Steam account you signed in with. No email address. No payment details. A crash report you send is kept on our server and in a private GitHub repository only ODG's developers can open: the zip for 90 days, or 30 if it holds a crash dump, and the ticket, your note and the messages on it until you ask, or until we clear out closed reports. Ask and we will delete the row or any report."), ("THE DEVELOPER REPOSITORY", "Two tiers reach the private repository. That is a real GitHub invite, so it needs your GitHub username, and a person confirms it by hand before anything is sent. Do not pass what is in there to people who did not pay for it. That is the one thing that ends access on the spot."), ("ACCESS CAN END", "Your subscription ends when you cancel, when a payment fails, or when we end it for sharing access or for conduct that spoils the channels. When it ends this launcher stops starting the game and stops treating the pack as installed. It then offers you three things: resubscribe, remove the pack, or take the free launcher and carry on with the public release. Nothing is deleted from your PC unless you press delete yourself."), ("THE FREE RELEASE STAYS FREE", "Every mod, map and fix reaches every player for nothing, a little later. Subscribing buys time and a voice, never a file that free players never get. A paid tier gets a build early only when we push one to it; the rest of the time it gets the public release like everyone. If that ever changed it would change on the website before it changed here."), ("NO WARRANTY", "This is unofficial software made by volunteers and it comes as it is. We test it and we use it ourselves, and we still cannot promise it will not break a save, a mod or an afternoon. Back up anything you would be sorry to lose. Using it is your decision."), ("SHARING, AND WHAT IS NOT OURS TO GIVE", "Our own work is yours to pass on. Do not sell it, do not put it behind a paywall, and do not present it as yours. Some of what the pack carries came from elsewhere and is not ours to licence, so sharing the pack is not permission to lift the pieces out of it."), ] # --------------------------------------------------------------------------- # output # --------------------------------------------------------------------------- TTY = sys.stdout.isatty() def say(msg=""): print(msg, flush=True) def mb(n): n = max(int(n or 0), 0) if n >= 1048576: return "{:,.0f} MB".format(n / 1048576) return "{:,.0f} KB".format(max(n / 1024, 1)) class Problem(Exception): """Something went wrong, in words meant for the player.""" class NoAnswer(Exception): """The server could not be reached, or what came back was not an answer.""" class Refused(Exception): """The server answered with an HTTP error.""" def __init__(self, status, body=""): Exception.__init__(self, "HTTP {}".format(status)) self.status = status self.body = body class Bar: """One progress line, redrawn in place on a terminal.""" def __init__(self, label, total): self.label = label self.total = max(int(total or 0), 0) self.last = 0.0 self.shown = -1 def update(self, got): if TTY: now = time.monotonic() if now - self.last < 0.2 and got < self.total: return self.last = now if self.total: line = " {}: {} of {} ({}%)".format(self.label, mb(got), mb(self.total), min(100, 100 * got // self.total)) else: line = " {}: {}".format(self.label, mb(got)) sys.stdout.write("\r" + line.ljust(72)) sys.stdout.flush() elif self.total: step = min(100, 100 * got // self.total) // 25 * 25 if step != self.shown: self.shown = step say(" {}: {}%".format(self.label, step)) def done(self): if TTY: sys.stdout.write("\n") sys.stdout.flush() # --------------------------------------------------------------------------- # small helpers # --------------------------------------------------------------------------- def now_unix(): return int(time.time()) def parts_of(rel): return [p for p in str(rel).replace("\\", "/").split("/") if p not in ("", ".")] def norm(rel): s = str(rel).strip().replace("\\", "/") while s.startswith("./"): s = s[2:] return s.lstrip("/") def read_list(text): out = [] for line in text.splitlines(): line = line.strip() if not line or line.startswith("#"): continue line = norm(line) if line: out.append(line) return out def safe_rel(rel): """Under Mods, no '..', no drive letter, either separator (stale.rs).""" lower = rel.lower() trimmed = rel.rstrip("/\\") return (lower.startswith("mods/") and lower != "mods/" and ":" not in rel and all(seg not in ("", ".", "..") for seg in re.split(r"[/\\]", trimmed))) def is_custom(rel_lower): return "/maploader_assets/custom/" in rel_lower or rel_lower.endswith("/maploader_assets/custom") def remove_quietly(path): try: os.remove(str(path)) except OSError: pass def sha256_file(path): h = hashlib.sha256() with open(str(path), "rb") as f: while True: chunk = f.read(1 << 20) if not chunk: break h.update(chunk) return h.hexdigest() def write_atomic(path, text): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(path.name + ".tmp") with open(str(tmp), "w", encoding="utf-8", newline="\n") as f: f.write(text) os.replace(str(tmp), str(path)) def ask(question, default=False): """Yes or no. With no terminal to ask, the default stands.""" if not sys.stdin or not sys.stdin.isatty(): return default hint = "[Y/n]" if default else "[y/N]" try: answer = input("{} {} ".format(question, hint)).strip().lower() except EOFError: return default if not answer: return default return answer in ("y", "yes") # --------------------------------------------------------------------------- # versions (version.rs) # --------------------------------------------------------------------------- def parse_version(s): s = str(s or "").strip().lstrip("vV") if not s: return None m = re.search(r"[-+_ ]", s) if m: nums, label = s[:m.start()], s[m.start() + 1:].strip().lower() else: nums, label = s, "" parts = [] for piece in nums.split("."): piece = piece.strip() if not piece: continue if not piece.isdigit(): return None parts.append(int(piece)) if not parts: return None return parts, label def compare(mine, theirs): """'behind', 'same', 'ahead', or 'unclear' when either does not parse.""" a, b = parse_version(mine), parse_version(theirs) if a is None or b is None: return "unclear" n = max(len(a[0]), len(b[0])) for i in range(n): x = a[0][i] if i < len(a[0]) else 0 y = b[0][i] if i < len(b[0]) else 0 if x != y: return "behind" if x < y else "ahead" la, lb = a[1], b[1] if la == lb: return "same" if not la: return "ahead" if not lb: return "behind" return "behind" if la < lb else "ahead" def lineage(tag): """'v1.0.1' is the public line, 'v109' the developer one (update.rs).""" t = re.split(r"[-+]", str(tag or "").strip())[0] if not t: return "" return "public" if "." in t else "dev" # --------------------------------------------------------------------------- # http # --------------------------------------------------------------------------- RETRY_STATUS = (408, 429, 500, 502, 503, 504) def _local_override(var): """Tests point at a server on this PC. Never anywhere else (update.rs, access.rs).""" v = os.environ.get(var, "").strip() if v.startswith("http://127.0.0.1:") or v.startswith("http://localhost:"): return v.rstrip("/") return None def public_api(): return _local_override("ODG_TEST_PUBLIC_API") or PUBLIC_API def server(): return _local_override("ODG_SERVER") or DEFAULT_SERVER def site(): return _local_override("ODG_TEST_SITE") or SITE def _request(url, key=None, headers=None, data=None, method=None): req = urllib.request.Request(url, data=data, method=method) req.add_header("User-Agent", USER_AGENT) for k, v in (headers or {}).items(): req.add_header(k, v) if key: # Never carried onto a redirect. The account server answers a download # with a 302 to GitHub's storage, and the member key stops at our server. req.add_unredirected_header("Authorization", "Bearer " + key) return req def http_open(url, key=None, headers=None, data=None, method=None, timeout=30, attempts=5): last = None for i in range(attempts): try: return urllib.request.urlopen(_request(url, key, headers, data, method), timeout=timeout) except urllib.error.HTTPError as e: body = "" try: body = e.read().decode("utf-8", "replace") except Exception: pass if e.code in RETRY_STATUS and i + 1 < attempts: last = e time.sleep(2) continue raise Refused(e.code, body) except (urllib.error.URLError, OSError, http.client.HTTPException) as e: last = e if i + 1 < attempts: time.sleep(2) continue raise NoAnswer(str(getattr(e, "reason", e))) raise NoAnswer(str(last)) def http_json(url, key=None, headers=None, timeout=60, attempts=5): r = http_open(url, key, headers, timeout=timeout, attempts=attempts) try: raw = r.read() except (OSError, http.client.HTTPException) as e: raise NoAnswer(str(e)) finally: r.close() try: return json.loads(raw.decode("utf-8")) except ValueError: raise NoAnswer("the answer was not a release listing") def account_headers(): return {"Accept": "application/json", "X-ODG-Launcher": "linux " + TOOL_VERSION} def account_call(path, key=None, body=None, extra=None, attempts=1): """A call to the account server. Its answers carry an 'error' field rather than relying on the status code, so any JSON body is an answer.""" headers = account_headers() headers.update(extra or {}) data = None if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" try: r = http_open(server() + path, key, headers, data=data, method="POST" if data is not None else "GET", timeout=20, attempts=attempts) try: raw = r.read() finally: r.close() except Refused as e: raw = (e.body or "").encode("utf-8") if not raw.strip(): raise NoAnswer("the account server answered {}".format(e.status)) except (OSError, http.client.HTTPException) as e: raise NoAnswer(str(e)) try: doc = json.loads(raw.decode("utf-8")) except ValueError: raise NoAnswer("the account server's answer could not be read") if not isinstance(doc, dict): raise NoAnswer("the account server's answer could not be read") return doc def check_space(folder, need): """Refuse only when failure is certain: the archive alone will not fit (preflight.rs).""" if not need: return try: free = shutil.disk_usage(str(folder)).free except OSError: return margin = 64 * 1024 * 1024 if free < need + margin: raise Problem("not enough disk space: {} free, {} needed".format(mb(free), mb(need + margin))) def download(url, dest, size, label, key=None): """Stream url to dest. A dropped connection picks up where it stopped.""" dest = Path(dest) remove_quietly(dest) bar = Bar(label, size) got = 0 tries = 0 while True: tries += 1 headers = {"Accept": "application/octet-stream"} if got: headers["Range"] = "bytes={}-".format(got) try: r = http_open(url, key, headers, timeout=60, attempts=1) except Refused as e: if e.status not in RETRY_STATUS or tries >= 5: bar.done() raise Problem("the download was refused (HTTP {})".format(e.status)) time.sleep(2) continue except NoAnswer as e: if tries >= 5: bar.done() raise Problem("the download failed: {}".format(e)) time.sleep(2) continue try: if got and getattr(r, "status", 200) != 206: got = 0 with open(str(dest), "ab" if got else "wb") as f: while True: chunk = r.read(1 << 20) if not chunk: break f.write(chunk) got += len(chunk) bar.update(got) except OSError as e: if getattr(e, "errno", None) == errno.ENOSPC: bar.done() remove_quietly(dest) raise Problem("the disk filled up during the download") if tries >= 5: bar.done() raise Problem("the download failed: {}".format(e)) time.sleep(2) continue except http.client.HTTPException as e: if tries >= 5: bar.done() raise Problem("the download failed: {}".format(e)) time.sleep(2) continue finally: r.close() if size and got < size: if tries >= 5: bar.done() raise Problem("the download stopped early ({} of {})".format(mb(got), mb(size))) continue break bar.done() if got == 0: raise Problem("downloaded nothing") def verify(path, size, digest): """The size and sha256 GitHub publishes with every asset (sha256.rs).""" got = os.path.getsize(str(path)) if size and got != size: raise Problem("the download is {} bytes and the release says {}".format(got, size)) digest = str(digest or "").strip() if digest.lower().startswith("sha256:"): if sha256_file(path).lower() != digest[7:].strip().lower(): raise Problem("the download does not match the release's checksum") # --------------------------------------------------------------------------- # the game folder, without regard to case # --------------------------------------------------------------------------- class Disk: """The game folder, matched the way Windows and Proton match it. Proton finds Mods\\MapLoader_Assets whatever case it is written in. A Linux disk does not, so every path here is matched against what is on disk, and a new folder takes the pack's spelling only when no spelling of it exists yet. That keeps one Mods folder, never a Mods and a mods. """ def __init__(self, root): self.root = Path(root) self._ls = {} def _names(self, d): k = str(d) m = self._ls.get(k) if m is None: m = {} try: for n in os.listdir(k): m.setdefault(n.lower(), []).append(n) except OSError: pass self._ls[k] = m return m def forget(self, d=None): if d is None: self._ls.clear() else: self._ls.pop(str(d), None) def _step(self, cur, part): hits = self._names(cur).get(part.lower()) if not hits: return None return cur / (part if part in hits else sorted(hits)[0]) def find(self, rel): cur = self.root for part in parts_of(rel): nxt = self._step(cur, part) if nxt is None: return None cur = nxt return cur def note(self, path): names = self._names(path.parent) spellings = names.setdefault(path.name.lower(), []) if path.name not in spellings: spellings.append(path.name) def _folder(self, cur, part): """cur/part as a folder, made if needed. Some Linux unzip tools write a pack's folder entries as empty files, so a game folder that had the pack copied in by hand can hold an empty MapLoader_Assets FILE. An empty file where the pack needs a folder is replaced. A spelling that is already a folder wins over one that is not. """ spellings = sorted(self._names(cur).get(part.lower(), []), key=lambda n: (n != part, n)) for name in spellings: if (cur / name).is_dir(): return cur / name for name in spellings: q = cur / name try: st = os.lstat(str(q)) except OSError: continue if stat.S_ISREG(st.st_mode) and st.st_size == 0: os.remove(str(q)) self.forget(cur) say("Replaced an empty file named {} with a folder.".format(self.rel(q))) spellings = [] break raise Problem("{} is a file where the pack needs a folder. Move it out of the way and run this again." .format(self.rel(q))) q = cur / part try: q.mkdir() except FileExistsError: pass except OSError as e: raise Problem("could not make {}: {}".format(self.rel(q), e)) if not q.is_dir(): raise Problem("could not make {}".format(self.rel(q))) self.note(q) return q def place(self, rel): """Where rel is, or will be. Folders on the way are made as needed.""" cur = self.root ps = parts_of(rel) for i, part in enumerate(ps): if i < len(ps) - 1: cur = self._folder(cur, part) else: cur = self._step(cur, part) or cur / part return cur def make_dir(self, rel): ps = parts_of(rel) parent = self.place("/".join(ps[:-1]) + "/x").parent if len(ps) > 1 else self.root return self._folder(parent, ps[-1]) def rel(self, path): try: return Path(path).relative_to(self.root).as_posix() except ValueError: return str(path) def isfile(self, rel): p = self.find(rel) return p is not None and p.is_file() def read(self, rel, default=""): p = self.find(rel) if p is None: return default try: return p.read_text(encoding="utf-8", errors="replace") except OSError: return default def write(self, rel, text): p = self.place(rel) write_atomic(p, text) self.note(p) def remove(self, rel): p = self.find(rel) if p is not None: remove_quietly(p) self.forget(p.parent) def extract(zip_path, disk, label): """Unpack over the game folder. The archive root holds Mods/... and the rest, so every file lands where it belongs.""" try: z = zipfile.ZipFile(str(zip_path)) except (zipfile.BadZipFile, OSError) as e: raise Problem("{} is not a readable zip ({})".format(label, e)) with z: infos = z.infolist() bar = Bar("unpacking " + label, sum(i.file_size for i in infos)) done = 0 try: for info in infos: ps = parts_of(info.filename) if not ps or ".." in ps or ":" in ps[0]: continue rel = "/".join(ps) if info.filename.endswith("/"): disk.make_dir(rel) continue dest = disk.place(rel) if dest.is_dir(): raise Problem("{}: a folder is where a file goes".format(rel)) with z.open(info) as src: try: f = open(str(dest), "wb") except PermissionError: os.chmod(str(dest), stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH) f = open(str(dest), "wb") with f: shutil.copyfileobj(src, f, 1 << 20) disk.note(dest) done += info.file_size bar.update(done) except zipfile.BadZipFile as e: bar.done() raise Problem("{} is damaged ({})".format(label, e)) except OSError as e: bar.done() if getattr(e, "errno", None) == errno.ENOSPC: raise Problem("the disk filled up while unpacking {}. Free some space and run this again".format(label)) raise Problem("could not write a file from {}: {}".format(label, e)) bar.done() # --------------------------------------------------------------------------- # steam: where the game is # --------------------------------------------------------------------------- class _Open: pass class _Close: pass OPEN, CLOSE = _Open(), _Close() def vdf_tokens(text): out = [] i, n = 0, len(text) esc = {"n": "\n", "t": "\t", "\\": "\\", '"': '"'} while i < n: c = text[i] if c in " \t\r\n": i += 1 elif c == "/" and text.startswith("//", i): j = text.find("\n", i) i = n if j < 0 else j + 1 elif c == "{": out.append(OPEN) i += 1 elif c == "}": out.append(CLOSE) i += 1 elif c == '"': i += 1 buf = [] while i < n and text[i] != '"': if text[i] == "\\" and i + 1 < n: buf.append(esc.get(text[i + 1], "\\" + text[i + 1])) i += 2 continue buf.append(text[i]) i += 1 i += 1 out.append("".join(buf)) else: j = i while j < n and text[j] not in ' \t\r\n{}"': j += 1 out.append(text[i:j]) i = j return out def vdf_parse(text): """Steam's text KeyValues as nested dicts. Keys are lower-cased; Steam has written both cases of the same key over the years.""" toks = vdf_tokens(text) pos = [0] def block(): d = {} while pos[0] < len(toks): t = toks[pos[0]] if t is CLOSE: pos[0] += 1 return d if t is OPEN: pos[0] += 1 block() continue pos[0] += 1 if pos[0] >= len(toks): break v = toks[pos[0]] if v is OPEN: pos[0] += 1 val = block() elif v is CLOSE: break else: val = v pos[0] += 1 d.setdefault(t.lower(), val) return d return block() def vdf_file(path): try: return vdf_parse(Path(path).read_text(encoding="utf-8", errors="replace")) except OSError: return {} def dig(doc, *keys): for k in keys: if not isinstance(doc, dict): return None doc = doc.get(k.lower()) return doc def ci_child(folder, name): """folder/name whatever its case on disk, or None.""" if folder is None: return None folder = Path(folder) p = folder / name if os.path.lexists(str(p)): return p try: for n in os.listdir(str(folder)): if n.lower() == name.lower(): return folder / n except OSError: pass return None def steam_roots(): home = Path.home() cands = [] if os.environ.get("ODG_STEAM_DIR"): cands.append(Path(os.environ["ODG_STEAM_DIR"])) cands += [ home / ".steam" / "steam", home / ".steam" / "root", home / ".local" / "share" / "Steam", home / ".var" / "app" / "com.valvesoftware.Steam" / ".local" / "share" / "Steam", home / ".var" / "app" / "com.valvesoftware.Steam" / ".steam" / "steam", home / "snap" / "steam" / "common" / ".local" / "share" / "Steam", home / ".steam" / "debian-installation", ] seen, out = set(), [] for c in cands: if ci_child(c, "steamapps") is None: continue try: key = str(c.resolve()) except OSError: continue if key not in seen: seen.add(key) out.append(c) return out def library_dirs(root): libs = [Path(root)] sa = ci_child(root, "steamapps") for vdf in (sa / "libraryfolders.vdf" if sa else None, Path(root) / "config" / "libraryfolders.vdf"): if vdf is None or not vdf.is_file(): continue lf = vdf_file(vdf).get("libraryfolders") if not isinstance(lf, dict): continue for k, v in lf.items(): p = v.get("path") if isinstance(v, dict) else (v if k.isdigit() else None) if isinstance(p, str) and p: libs.append(Path(p)) seen, out = set(), [] for lib in libs: try: key = str(lib.resolve()) except OSError: continue if key not in seen: seen.add(key) out.append(lib) return out class Install: def __init__(self, game, steamapps=None, roots=None): self.game = Path(game) self.steamapps = steamapps self.roots = roots or [] self.prefix = self._prefix() def _prefix(self): places = [] if self.steamapps is not None: places.append(self.steamapps) for r in self.roots: places.append(ci_child(r, "steamapps")) for sa in places: pfx = ci_child(ci_child(ci_child(sa, "compatdata"), APPID), "pfx") if pfx is not None and pfx.is_dir(): return pfx return None def manifest(self): if self.steamapps is not None: m = self.steamapps / "appmanifest_{}.acf".format(APPID) if m.is_file(): return m for r in self.roots: for lib in library_dirs(r): m = ci_child(lib, "steamapps") m = m / "appmanifest_{}.acf".format(APPID) if m else None if m is not None and m.is_file(): return m return None def roaming(self): """%APPDATA% as the game sees it inside Proton.""" if self.prefix is None: return None return self.prefix / "drive_c" / "users" / "steamuser" / "AppData" / "Roaming" def is_game_dir(p): return p is not None and Path(p).is_dir() and ci_child(p, "OPERATOR.exe") is not None def find_installs(): found, seen = [], set() roots = steam_roots() for root in roots: for lib in library_dirs(root): sa = ci_child(lib, "steamapps") if sa is None: continue installdir = "OPERATOR" man = sa / "appmanifest_{}.acf".format(APPID) if man.is_file(): d = dig(vdf_file(man), "AppState", "installdir") if isinstance(d, str) and d: installdir = d game = ci_child(ci_child(sa, "common"), installdir) if not is_game_dir(game): continue try: key = str(game.resolve()) except OSError: continue if key not in seen: seen.add(key) found.append(Install(game, sa, roots)) return found def resolve_install(opts, cfg): given = opts.get("game") remembered = cfg.get("game") if not given and remembered and is_game_dir(Path(remembered)): given = remembered if given: game = Path(os.path.expanduser(given)) if not is_game_dir(game): raise Problem("{} is not the OPERATOR folder: OPERATOR.exe is not in it.".format(game)) game = game.resolve() sa = game.parent.parent if game.parent.name.lower() == "common" else None return Install(game, sa, steam_roots()) found = find_installs() if not found: raise Problem( "OPERATOR was not found. Install it in Steam and start it once, then run this again.\n" "If it is installed somewhere unusual, give the folder with --game:\n" " python3 odg-linux.py --game \"/path/to/steamapps/common/OPERATOR\"\n" "In Steam: right-click OPERATOR > Manage > Browse local files shows that folder.") if len(found) > 1: say("OPERATOR is installed in more than one place. Using the first:") for f in found: say(" " + str(f.game)) say("Pick another with --game.") return found[0] def game_running(): """Is OPERATOR running? Under Proton the game's processes carry OPERATOR.exe in their command line.""" proc = Path("/proc") if not proc.is_dir(): return False me = str(os.getpid()) for d in proc.iterdir(): if not d.name.isdigit() or d.name == me: continue try: cmd = (d / "cmdline").read_bytes() except OSError: continue if b"operator.exe" in cmd.lower(): return True return False # Steam accounts (steam.rs). Only the number and the display name are read. STEAM_BASE = 76561197960265728 def valid_steam_id(s): s = str(s or "") return len(s) == 17 and s.startswith("7656119") and s.isdigit() def operator_owner(inst): """The Steam account the installed copy belongs to (appmanifest LastOwner).""" m = inst.manifest() if m is None: return None owner = dig(vdf_file(m), "AppState", "LastOwner") return owner if valid_steam_id(owner) else None def steam_users(roots): users = {} for r in roots: doc = vdf_file(Path(r) / "config" / "loginusers.vdf") for sid, u in (doc.get("users") or {}).items(): if isinstance(u, dict) and valid_steam_id(sid): users[sid] = { "name": u.get("personaname", "") or "", "recent": str(u.get("mostrecent", "0")).strip() == "1", "ts": int(u.get("timestamp", "0") or 0) if str(u.get("timestamp", "0")).isdigit() else 0, } return users def steam_pc_account(roots): """The Steam account signed in on this PC now, else the last one to be.""" users = steam_users(roots) home = Path.home() regs = [home / ".steam" / "registry.vdf", home / ".var" / "app" / "com.valvesoftware.Steam" / ".steam" / "registry.vdf"] regs += [Path(r) / "registry.vdf" for r in roots] for reg in regs: active = dig(vdf_file(reg), "Registry", "HKCU", "Software", "Valve", "Steam", "ActiveProcess", "ActiveUser") try: n = int(str(active or "0"), 0) except ValueError: n = 0 if n: sid = str(STEAM_BASE + n) if valid_steam_id(sid): return {"id": sid, "name": users.get(sid, {}).get("name", "")} if users: sid = max(users, key=lambda k: (users[k]["recent"], users[k]["ts"])) return {"id": sid, "name": users[sid]["name"]} return None # --------------------------------------------------------------------------- # config and account # --------------------------------------------------------------------------- def config_dir(): if os.environ.get("ODG_CONFIG_DIR"): return Path(os.environ["ODG_CONFIG_DIR"]) base = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config") return Path(base) / "odg" def load_json(path): try: with open(str(path), encoding="utf-8") as f: doc = json.load(f) return doc if isinstance(doc, dict) else {} except (OSError, ValueError): return {} def load_config(): return load_json(config_dir() / "config.json") def save_config(cfg): write_atomic(config_dir() / "config.json", json.dumps(cfg, indent=2) + "\n") TIERS = ["free", "release", "beta", "alpha", "bug-hunter", "dev-partner", "developer"] TIER_STAMP = {"free": "FREE", "release": "RELEASE", "beta": "BETA", "alpha": "ALPHA", "bug-hunter": "BUG HUNTER", "dev-partner": "DEV PARTNER", "developer": "DEVELOPER"} def tier_of(s): t = str(s or "").strip().lower().replace(" ", "-").replace("_", "-") return {"bughunter": "bug-hunter", "devpartner": "dev-partner", "dev": "developer"}.get( t, t if t in TIERS else "free") def status_of(s): t = str(s or "").strip().lower() if t == "active": return "active" if t in ("cancelled", "canceled", "revoked", "expired"): return "cancelled" if t == "none": return "none" return "unknown" def blank_account(): return {"key": "", "name": "", "tier": "free", "status": "none", "checked": 0, "message": "", "dev_repo": False, "dev_state": "none", "github": "", "steam_id": "", "steam_name": "", "patreon_linked": False} def clean_account(doc): a = blank_account() if not isinstance(doc, dict): return a a["key"] = str(doc.get("key") or "").strip() a["name"] = str(doc.get("name") or "") a["tier"] = tier_of(doc.get("tier")) a["status"] = status_of(doc.get("status")) if "status" in doc else "none" a["checked"] = doc.get("checked") if isinstance(doc.get("checked"), int) else 0 a["message"] = str(doc.get("message") or "") a["dev_repo"] = bool(doc.get("dev_repo")) a["dev_state"] = str(doc.get("dev_state") or "none") a["github"] = str(doc.get("github") or "") a["steam_id"] = str(doc.get("steam_id") or "") a["steam_name"] = str(doc.get("steam_name") or "") a["patreon_linked"] = bool(doc.get("patreon_linked")) return a def from_answer(doc, key, fallback_name, now): steam = doc.get("steam") if isinstance(doc.get("steam"), dict) else {} name = str(doc.get("name") or "") or fallback_name return {"key": key or "", "name": name, "tier": tier_of(doc.get("tier")), "status": status_of(doc.get("status")), "checked": now, "message": str(doc.get("message") or ""), "dev_repo": bool(doc.get("dev_repo")), "dev_state": str(doc.get("dev_state") or "none"), "github": str(doc.get("github") or ""), "steam_id": str(steam.get("id") or ""), "steam_name": str(steam.get("name") or ""), "patreon_linked": bool(doc.get("patreon"))} def prefix_account_path(inst): roaming = inst.roaming() if inst is not None else None return roaming / "OperatorMods" / "account.json" if roaming is not None else None def load_account(inst=None): path = config_dir() / "account.json" if path.is_file(): return clean_account(load_json(path)) # A sign-in made by the Windows launcher, carried over in the prefix. pa = prefix_account_path(inst) if pa is not None and pa.is_file(): return clean_account(load_json(pa)) return blank_account() def save_account(a, inst=None): """The game reads this file too: NewSlots, the Lounge and the Watch gates take tier, status and steam_id from %APPDATA%\\OperatorMods\\account.json, which under Proton is inside the prefix.""" text = json.dumps(a, indent=2) write_atomic(config_dir() / "account.json", text) pa = prefix_account_path(inst) if pa is not None: try: write_atomic(pa, text) except OSError as e: say("Could not write the sign-in into the game's Proton folder: {}".format(e)) def is_member(a): return bool(a["key"]) and a["status"] == "active" and a["tier"] != "free" def stamp_of(a): if a["status"] == "active": return TIER_STAMP.get(a["tier"], "FREE") return "FREE" def refresh_account(a, inst): """Ask the server about this key (access.rs refresh). Unreachable is never a cancellation; an active answer is honoured for GRACE_DAYS without a check.""" if not a["key"]: a = dict(a) a["status"] = "none" return a owner = operator_owner(inst) if inst is not None else None extra = {"X-ODG-Steam-Owner": owner} if owner else {} now = now_unix() try: doc = account_call("/api/launcher/status", key=a["key"], extra=extra, attempts=2) if not str(doc.get("status") or "").strip(): raise NoAnswer(str(doc.get("error") or "the account server gave no status")) return from_answer(doc, a["key"], a["name"], now) except NoAnswer as e: b = dict(a) b["message"] = str(e) if b["status"] == "active" and now - int(b["checked"] or 0) > GRACE_DAYS * 86400: b["status"] = "unknown" return b def record_consent(key): try: account_call("/api/launcher/consent", key=key, body={"version": TERMS_VERSION}) except NoAnswer: pass # --------------------------------------------------------------------------- # the mod loader (melonsetup.rs) # --------------------------------------------------------------------------- def loader_state(disk): """('ready'|'absent'|'incomplete'|'held', missing).""" missing = [r for r in LOADER_FILES if not disk.isfile(r)] if missing == ["version.dll"] and disk.isfile(HELD): return "held", [] if not missing: return "ready", [] if len(missing) == len(LOADER_FILES): return "absent", missing return "incomplete", missing def is_melonloader(path): try: with open(str(path), "rb") as f: return b"MelonLoader" in f.read() except OSError: return False # --------------------------------------------------------------------------- # mods on and off (applier.rs) # --------------------------------------------------------------------------- def display_name(file): k = KNOWN.get(file.lower()) if k: return k[0] stem = file[:-4] if file.lower().endswith(".dll") else file return stem[:-4] if stem.lower().endswith(".mod") else stem def requirement(file): k = KNOWN.get(file.lower()) return k[1] if k else None def mod_files(disk): """[(file, on_name, off_name)] for every DLL in Mods, either state.""" mods = disk.find("Mods") if mods is None or not mods.is_dir(): return [] found = {} for n in sorted(os.listdir(str(mods))): low = n.lower() if low.endswith(".dll" + OFF): canon = n[:-len(OFF)] found.setdefault(canon.lower(), [canon, None, None])[2] = n elif low.endswith(".dll"): found.setdefault(low, [n, None, None])[1] = n out = [tuple(v) for v in found.values()] out.sort(key=lambda m: (requirement(m[0]) is None, KNOWN.get(m[0].lower()) is None, display_name(m[0]).lower())) return out def switched_off(disk): return {f.lower() for f, on, off in mod_files(disk) if off and not on} def set_mod(disk, file, on_name, off_name, enable): mods = disk.find("Mods") if enable: if on_name: return False os.replace(str(mods / off_name), str(mods / file)) else: if not on_name: return False os.replace(str(mods / on_name), str(mods / (on_name + OFF))) disk.forget(mods) return True def reapply_off(disk, off_before): """An update unpacks a fresh X.dll beside a switched-off X.dll.odg-off. Mods that were off stay off, on the new copy.""" for file, on, off in mod_files(disk): if file.lower() in off_before and on and not requirement(file): set_mod(disk, file, on, off, False) def required_on(disk): """Load-bearing mods are held on, as the launcher holds them.""" mods = disk.find("Mods") for file, on, off in mod_files(disk): if not requirement(file) or not off: continue if on: remove_quietly(mods / off) else: os.replace(str(mods / off), str(mods / file)) say("{} is required and was switched back on.".format(display_name(file))) disk.forget(mods) # --------------------------------------------------------------------------- # leftovers (stale.rs) and deletes # --------------------------------------------------------------------------- def allowlisted(name_lower): return (name_lower in ALLOWLISTED or name_lower.endswith(".log") or name_lower.endswith(OFF) or name_lower.endswith(".tmp") or ".bak" in name_lower) def walk_files(disk, folder, out): try: entries = list(os.scandir(str(folder))) except OSError: return for e in entries: try: if e.is_symlink(): continue if e.is_dir(follow_symlinks=False): walk_files(disk, e.path, out) elif e.is_file(follow_symlinks=False): out.append((disk.rel(e.path), e.stat(follow_symlinks=False).st_size)) except OSError: continue def dir_bytes(folder): total = 0 for dirpath, _dirs, files in os.walk(str(folder)): for f in files: try: total += os.lstat(os.path.join(dirpath, f)).st_size except OSError: pass return total def stale_scan(disk): """Files earlier packs left that the current one does not ship, inside the folders the pack owns, plus whatever the obsolete list names.""" out, seen = [], set() mods = disk.find("Mods") if mods is None or not mods.is_dir(): return out shipped = read_list(disk.read(SHIPPED_LIST)) if shipped: shipped_lower = {s.lower() for s in shipped} owned, owned_assets = set(), set() for s in shipped_lower: if not s.startswith("mods/"): continue rest = s[5:] if "/" not in rest: continue top, more = rest.split("/", 1) owned.add(top) if top == "maploader_assets" and "/" in more: owned_assets.add(more.split("/", 1)[0]) owned_assets.discard("custom") files = [] for e in os.scandir(str(mods)): try: if e.is_symlink() or not e.is_dir(follow_symlinks=False): continue except OSError: continue if e.name.lower() in owned: walk_files(disk, e.path, files) for rel, size in files: lower = rel.lower() if lower in shipped_lower or allowlisted(lower.rsplit("/", 1)[-1]): continue if lower.startswith("mods/maploader_assets/"): rest = lower[len("mods/maploader_assets/"):] if "/" in rest: sub = rest.split("/", 1)[0] if sub == "custom" or sub not in owned_assets: continue if lower not in seen: seen.add(lower) out.append((rel, size, False)) for rel in read_list(disk.read(OBSOLETE_LIST)): is_dir = rel.endswith("/") clean = rel.rstrip("/") if not safe_rel(clean): continue p = disk.find(clean) if p is None: continue try: st = os.lstat(str(p)) except OSError: continue if stat.S_ISLNK(st.st_mode) or is_dir != stat.S_ISDIR(st.st_mode): continue lower = clean.lower() if lower not in seen: seen.add(lower) out.append((disk.rel(p), dir_bytes(p) if is_dir else st.st_size, is_dir)) out.sort(key=lambda e: e[0].lower()) return out def through_link(path, stop): d = Path(path) stop = Path(stop) while d != stop and stop in d.parents: if os.path.islink(str(d)): return True d = d.parent return False def delete_entries(disk, entries): """Remove exactly these, each re-checked: under Mods, no '..', not a link.""" removed, failed = [], [] mods = disk.find("Mods") for rel, _size, _is_dir in entries: r = rel.rstrip("/") if mods is None or not safe_rel(r): failed.append((rel, "refused: outside Mods")) continue p = disk.find(r) if p is None: removed.append(rel) continue if mods not in p.parents: failed.append((rel, "refused: outside Mods")) continue try: st = os.lstat(str(p)) except OSError: removed.append(rel) continue if stat.S_ISLNK(st.st_mode) or through_link(p.parent, mods): failed.append((rel, "refused: a link")) continue try: if stat.S_ISDIR(st.st_mode): shutil.rmtree(str(p)) else: os.remove(str(p)) removed.append(rel) except OSError as e: failed.append((rel, str(e))) disk.forget(p.parent) return removed, failed # --------------------------------------------------------------------------- # install and update (update.rs) # --------------------------------------------------------------------------- def asset_named(assets, name): for a in assets: if str(a.get("name") or "").lower() == name.lower(): return a return None def asset_size(a): s = a.get("size") return s if isinstance(s, int) and s > 0 else 0 class Updater: def __init__(self, inst): self.inst = inst self.disk = Disk(inst.game) self.failed = [] self.changed = False self.src = "public" self.key = None self.tag = "" self.hold = None self.running = False def fail(self, msg): self.failed.append(msg) say(msg) def staging(self): try: cache = self.disk.make_dir(STAGING) except Problem: import tempfile return Path(tempfile.gettempdir()) day = 24 * 3600 try: for e in os.scandir(str(cache)): n = e.name.lower() if n.startswith("odg_") and n.endswith(".zip") and time.time() - e.stat().st_mtime > day: remove_quietly(e.path) except OSError: pass return cache def asset_url(self, a): if self.src == "broker": if not isinstance(a.get("id"), int): return None return "{}/api/launcher/dev/asset?id={}".format(server(), a["id"]) u = a.get("browser_download_url") return u if isinstance(u, str) and u else None def fetch(self, a, dest, label): url = self.asset_url(a) if not url: raise Problem("the release gives no address for it") size = asset_size(a) check_space(Path(dest).parent, size) download(url, dest, size, label, key=self.key if self.src == "broker" else None) verify(dest, size, a.get("digest")) # ---- which release ---------------------------------------------------- def listing(self, acct, public_doc): """The release this PC should be on, or None when a member's pack stays because the account server did not answer.""" have = self.disk.read(STAMP).strip() if is_member(acct): self.src, self.key = "broker", acct["key"] say("Checking your channel") doc = None try: doc = http_json(server() + "/api/launcher/dev/release", key=self.key, headers=account_headers(), timeout=30, attempts=2) except Refused as e: if e.status in (401, 403): say("No early build on this account, so this is the public release.") self.src = "public" except NoAnswer: pass if self.src == "broker": tag = doc.get("tag_name") if isinstance(doc, dict) else None if not isinstance(tag, str): if have: say("The account server did not answer, so {} stays.".format(have)) return None say("The account server could not be reached, so this is the public release.") self.src = "public" elif not tag: say("Nothing is pushed to your channel, so this is the public release.") self.src = "public" else: brief = str(doc.get("brief") or "").strip() say("Your channel ({}) is on {}.{}".format(doc.get("channel") or "?", tag, " " + brief if brief else "")) return doc self.src, self.key = "public", None return public_doc if public_doc is not None else public_listing() # ---- the run ----------------------------------------------------------- def run(self, acct, public_doc): disk = self.disk doc = self.listing(acct, public_doc) if doc is None: return self.tag = tag = str(doc.get("tag_name") or "") assets = [a for a in (doc.get("assets") or []) if isinstance(a, dict)] exact = self.src == "broker" and bool(doc.get("exact")) self.hold = hold = "held by .odg-dev-hold" if disk.find(DEV_HOLD) is not None else None self.running = running = hold is None and game_running() staging = self.staging() # ---- mods.zip ---- have = disk.read(STAMP).strip() forgotten = disk.read(FORGOTTEN).strip() kept_newer = False pack = asset_named(assets, "mods.zip") if pack is None: self.fail("This release carries no mod pack.") else: ahead = (not exact) and lineage(have) == lineage(tag) and compare(have, tag) == "ahead" if hold or ahead: kept_newer = ahead say("Pack {} kept ({}, the release is {}).".format(have, hold or "ahead of the release", tag)) elif have == tag and tag: if not running: self.finish_leftovers(tag) say("The pack is up to date ({}).".format(tag)) elif running: self.fail("OPERATOR is running, so the update to {} waits until it is closed.".format(tag)) else: self.install_pack(pack, tag, have, forgotten, staging) # ---- map data, each map once ---- maps = [a for a in assets if str(a.get("name") or "").lower().startswith("mapdata_") and str(a.get("name") or "").lower().endswith(".zip")] shipped_maps = [] history = None for idx, a in enumerate(maps, 1): name = str(a.get("name")) lower = name.lower() stem = lower[len("mapdata_"):-len(".zip")] shipped_maps.append(stem) lines = [l.strip() for l in disk.read(MAP_STAMP).splitlines() if l.strip()] on_disk = self.map_on_disk(stem) digest = str(a.get("digest") or "") digest = digest if digest.startswith("sha256:") else "" updated_at = str(a.get("updated_at") or "") want = digest or updated_at or str(a.get("size") or "") stamped = None for l in lines: n, _, marker = l.partition("\t") if n.strip().lower() == lower: stamped = marker.strip() break if stamped is not None: current = bool(stamped) and (stamped == want or (updated_at and stamped == updated_at)) if not current and on_disk and digest and stamped and not stamped.startswith("sha256:"): if history is None: history = self.digest_history() current = any(n == lower and at == stamped and d == digest for n, at, d in history) if current and on_disk: if stamped != want and hold is None: self.record_map(lines, name, want) continue if hold: say("Map data {} kept ({}).".format(idx, hold)) continue if running: say("Map data {} waits until OPERATOR is closed.".format(idx)) continue say("Map data {} of {}: {}, one time only".format(idx, len(maps), mb(asset_size(a)))) tmp = staging / "odg_mapdata_{}.zip".format(os.getpid()) try: self.fetch(a, tmp, "map data") extract(tmp, disk, "map data") except Problem as e: self.fail("Map data {} skipped: {}".format(idx, e)) continue finally: remove_quietly(tmp) disk.forget() lines = [l.strip() for l in disk.read(MAP_STAMP).splitlines() if l.strip()] self.record_map(lines, name, want) self.changed = True # ---- the mod loader ---- state, _missing = loader_state(disk) if hold is None and not running and state in ("absent", "incomplete"): a = asset_named(assets, "melonloader.zip") if a is None: self.fail("The mod loader is missing and this release does not carry one.") else: say("Installing the mod loader ({})".format(mb(asset_size(a)))) tmp = staging / "odg_melonloader_{}.zip".format(os.getpid()) try: self.fetch(a, tmp, "mod loader") extract(tmp, disk, "the mod loader") disk.forget() state, missing = loader_state(disk) if state != "ready": raise Problem("still not complete after unpacking: " + ", ".join(missing)) say("Mod loader installed.") self.changed = True except Problem as e: self.fail("The mod loader could not be installed: {}".format(e)) finally: remove_quietly(tmp) # ---- maps the release no longer ships ---- if hold: say("Maps kept ({}).".format(hold)) elif not kept_newer and not running: self.prune_maps(shipped_maps) # ---- the pack ---------------------------------------------------------- def install_pack(self, a, tag, have, forgotten, staging): disk = self.disk was = have or forgotten leaving = bool(was) and (lineage(was) != lineage(tag) or compare(was, tag) == "ahead") before = self.remember_old_pack(tag, was, leaving) off_before = switched_off(disk) say("Downloading the pack {} ({})".format(tag, mb(asset_size(a)))) tmp = staging / "odg_mods_{}.zip".format(os.getpid()) try: self.fetch(a, tmp, "pack") extract(tmp, disk, "the pack") except Problem as e: self.fail("Pack update skipped: {}".format(e)) return finally: remove_quietly(tmp) disk.forget() if leaving: self.take_back_leftovers(before, was, tag) else: disk.remove(PREV_LIST) reapply_off(disk, off_before) # Written last: until it is, the next run installs this build again. disk.write(STAMP, tag) disk.remove(FORGOTTEN) self.changed = True say("Pack {} installed.".format(tag)) def shipped(self): return read_list(self.disk.read(SHIPPED_LIST)) def read_prev(self): text = self.disk.read(PREV_LIST) lines = text.splitlines() if not lines or not lines[0].strip().startswith("# for "): return None head = lines[0].strip()[len("# for "):].split("\t") target = head[0].strip() if head else "" was = head[1].strip() if len(head) > 1 else "" leaving = len(head) > 2 and head[2].strip() == "1" return target, was, leaving, read_list("\n".join(lines[1:])) def remember_old_pack(self, target, was, leaving): now = self.shipped() prev = self.read_prev() if prev and prev[0] == target: return prev[3] if prev: lst = list(prev[3]) low = {x.lower() for x in lst} for l in now: if l.lower() not in low: lst.append(l) low.add(l.lower()) else: lst = now self.disk.write(PREV_LIST, "# for {}\t{}\t{}\n{}\n".format(target, was, 1 if leaving else 0, "\n".join(lst))) return lst def switch_leftovers(self, old, new_lower): out, seen = [], set() for rel in old: lower = rel.lower() if not lower.startswith("mods/") or "/maploader_assets/custom/" in lower or lower in new_lower: continue names = [rel] + ([rel + OFF] if lower.endswith(".dll") else []) for n in names: if n.lower() in seen: continue seen.add(n.lower()) p = self.disk.find(n) if p is None: continue try: st = os.lstat(str(p)) except OSError: continue if stat.S_ISREG(st.st_mode): out.append((self.disk.rel(p), st.st_size, False)) return out def take_back_leftovers(self, before, was, tag): new_lower = {s.lower() for s in self.shipped()} if not new_lower: self.disk.remove(PREV_LIST) return gone = self.switch_leftovers(before, new_lower) failed = [] if gone: removed, failed = delete_entries(self.disk, gone) if lineage(was) != lineage(tag): which = "early" if lineage(was) == "dev" else "public" else: which = "newer" say("Removed {} file{} only the {} build had.".format(len(removed), "" if len(removed) == 1 else "s", which)) if not failed: self.disk.remove(PREV_LIST) def finish_leftovers(self, tag): prev = self.read_prev() if prev is None: return target, was, leaving, lst = prev if target == tag and leaving: self.take_back_leftovers(lst, was, tag) else: self.disk.remove(PREV_LIST) # ---- maps -------------------------------------------------------------- def map_on_disk(self, stem): root = self.disk.find(MAP_ROOT) if root is None: return False for actual in self.disk._names(root).get(stem, []): if self.disk.isfile("{}/{}/manifest.bin".format(MAP_ROOT, actual)): return True return False def record_map(self, lines, name, marker): kept = [l for l in lines if l.partition("\t")[0].strip().lower() != name.lower()] kept.append("{}\t{}".format(name, marker)) self.disk.write(MAP_STAMP, "\n".join(kept) + "\n") def digest_history(self): """Every asset's (name, upload time, digest) in recent public releases, to recognise a map stamped before stamps carried digests.""" if self.src == "broker": return [] base = public_api() if not base.endswith("/latest"): return [] try: doc = http_json(base[:-len("/latest")] + "?per_page=40", headers=GITHUB, attempts=2) except (NoAnswer, Refused): return [] out = [] for rel in doc if isinstance(doc, list) else []: for a in rel.get("assets") or [] if isinstance(rel, dict) else []: if not isinstance(a, dict): continue n = str(a.get("name") or "").lower() at = str(a.get("updated_at") or "") d = str(a.get("digest") or "") if n and at and d.startswith("sha256:"): out.append((n, at, d)) return out def prune_maps(self, shipped): """Only a map this tool or the launcher installed, never the player's own, and never anything when the release listed no maps at all.""" if not shipped: return disk = self.disk root = disk.find(MAP_ROOT) if root is None: return installed = set() for l in disk.read(MAP_STAMP).splitlines(): n = l.split()[0].lower() if l.split() else "" if n.startswith("mapdata_") and n.endswith(".zip") and len(n) > len("mapdata_.zip"): installed.add(n[len("mapdata_"):-len(".zip")]) removed = [] for e in list(os.scandir(str(root))): try: if e.is_symlink() or not e.is_dir(follow_symlinks=False): continue except OSError: continue low = e.name.lower() if low == "custom" or low not in installed or low in shipped: continue if ci_child(e.path, "manifest.bin") is None: continue try: shutil.rmtree(e.path) removed.append(low) say("Removed a map the release no longer ships ({}).".format(e.name)) except OSError as err: self.fail("Could not remove {}: {}".format(e.name, err)) disk.forget(root) if removed: lines = [l.strip() for l in disk.read(MAP_STAMP).splitlines() if l.strip()] keep = [] for l in lines: n = l.split()[0].lower() stem = n[len("mapdata_"):] if n.startswith("mapdata_") else n stem = stem[:-len(".zip")] if stem.endswith(".zip") else stem if stem not in removed: keep.append(l) disk.write(MAP_STAMP, "\n".join(keep) + "\n") # ---- tidy up ----------------------------------------------------------- def tidy(self, ask_leftovers): disk = self.disk mods = disk.find("Mods") if mods is None: return for name in RETIRED_FILES: p = ci_child(mods, name) if p is not None and os.path.isfile(str(p)) and not os.path.islink(str(p)): remove_quietly(p) say("Removed a retired file ({}).".format(name)) for name in RETIRED_DIRS: p = ci_child(mods, name) if p is not None and os.path.isdir(str(p)) and not os.path.islink(str(p)): shutil.rmtree(str(p), ignore_errors=True) say("Removed a retired folder ({}).".format(name)) disk.forget() if disk.find(MAP_ROOT) is not None: if disk.find(CUSTOM) is None: disk.make_dir(CUSTOM) if disk.find(CUSTOM + "/README.txt") is None: disk.write(CUSTOM + "/README.txt", CUSTOM_README) required_on(disk) if ask_leftovers and self.hold is None and not self.running: stale = stale_scan(disk) if stale: total = sum(s for _r, s, _d in stale) say("") say("{} file{} an older pack left behind ({}), for example {}.".format( len(stale), "" if len(stale) == 1 else "s", mb(total), stale[0][0][len("Mods/"):])) if OPTS.get("yes") or ask("Remove them?", default=False): removed, failed = delete_entries(disk, stale) say("Removed {}.{}".format(len(removed), " {} could not be removed.".format(len(failed)) if failed else "")) else: say("Left as they are.") def public_listing(): try: doc = http_json(public_api(), headers=GITHUB) except Refused as e: raise Problem("GitHub refused the release listing (HTTP {}). Try again in a few minutes.".format(e.status)) except NoAnswer as e: raise Problem("Could not reach GitHub: {}. Check your internet connection.".format(e)) if not isinstance(doc, dict) or "tag_name" not in doc: raise Problem("GitHub's answer was not a release listing. Try again in a few minutes.") return doc # --------------------------------------------------------------------------- # this tool's own updates # --------------------------------------------------------------------------- def self_update(): """Newer copies of this file are published on the site. A newer one replaces this file and runs in its place; an older or equal one never does.""" if OPTS.get("no_self_update") or os.environ.get("ODG_NO_SELF_UPDATE") or os.environ.get("ODG_SELF_UPDATED"): return try: me = Path(os.path.abspath(sys.argv[0])) except Exception: return if not me.is_file() or me.suffix.lower() != ".py" or not os.access(str(me), os.W_OK): return try: doc = http_json(site() + "/api/releases", headers={"Accept": "application/json"}, timeout=15, attempts=2) except (NoAnswer, Refused): return entry = None for a in doc.get("assets") or [] if isinstance(doc, dict) else []: if isinstance(a, dict) and a.get("key") == TOOL_KEY: entry = a break if not entry or compare(TOOL_VERSION, str(entry.get("version") or "")) != "behind": return want = str(entry.get("sha256") or "").lower() if len(want) != 64: return new = me.with_name(me.name + ".new") try: download(site() + "/dl/" + TOOL_KEY, new, int(entry.get("size") or 0), "ODG for Linux") verify(new, int(entry.get("size") or 0), "sha256:" + want) text = new.read_text(encoding="utf-8") m = re.search(r'^TOOL_VERSION = "([^"]+)"', text, re.M) if "ODG_LINUX_TOOL = True" not in text or not m or compare(TOOL_VERSION, m.group(1)) != "behind": raise Problem("the published copy is not a newer version of this tool") os.chmod(str(new), stat.S_IMODE(os.stat(str(me)).st_mode)) os.replace(str(new), str(me)) except (Problem, OSError) as e: remove_quietly(new) say("This tool could not update itself ({}). Carrying on with version {}.".format(e, TOOL_VERSION)) return say("Updated this tool to version {}. Starting it again.".format(m.group(1))) env = dict(os.environ) env["ODG_SELF_UPDATED"] = "1" os.execve(sys.executable, [sys.executable, str(me)] + sys.argv[1:], env) # --------------------------------------------------------------------------- # steam's launch option # --------------------------------------------------------------------------- OVERRIDE_RE = re.compile(r"""WINEDLLOVERRIDES\s*=\s*("([^"]*)"|'([^']*)'|(\S*))""", re.I) def _override_value(m): for g in (2, 3, 4): if m.group(g) is not None: return m.group(g) return "" def override_ok(opts): for m in OVERRIDE_RE.finditer(opts or ""): for entry in _override_value(m).split(";"): if "=" not in entry: continue dlls, mode = entry.split("=", 1) names = [d.strip().lower() for d in dlls.split(",")] if ("version" in names or "version.dll" in names) and mode.strip().lower().startswith("n"): return True return False def suggested_option(opts): opts = (opts or "").strip() if not opts: return LAUNCH_OPTION m = OVERRIDE_RE.search(opts) if m: val = _override_value(m).strip().rstrip(";") new = 'WINEDLLOVERRIDES="{}"'.format(val + ";version=n,b" if val else "version=n,b") return opts[:m.start()] + new + opts[m.end():] if "%command%" in opts: return 'WINEDLLOVERRIDES="version=n,b" ' + opts return LAUNCH_OPTION + " " + opts def launch_options(roots): """[(account name, launch options, most recent)] for each Steam account on this PC that has OPERATOR in its saved settings.""" out = [] users = steam_users(roots) for r in roots: ud = ci_child(r, "userdata") if ud is None: continue try: ids = [d for d in os.listdir(str(ud)) if d.isdigit()] except OSError: continue for d in ids: lc = Path(str(ud)) / d / "config" / "localconfig.vdf" try: text = lc.read_text(encoding="utf-8", errors="replace") except OSError: continue if '"{}"'.format(APPID) not in text: continue doc = vdf_parse(text) apps = dig(doc, "UserLocalConfigStore", "Software", "Valve", "Steam", "apps") app = apps.get(APPID) if isinstance(apps, dict) else None if not isinstance(app, dict): continue sid = str(STEAM_BASE + int(d)) u = users.get(sid, {}) out.append((u.get("name") or "Steam account " + d, str(app.get("launchoptions") or ""), bool(u.get("recent")))) return out def launch_option_report(inst, verbose=True): """True when the launch option is seen. Steam saves it when it closes, so a missing one is reported as not seen yet, not as wrong.""" found = launch_options(inst.roots) if not found: if verbose: say("Launch option: not seen in Steam's saved settings yet.") return False good = True for name, opts, _recent in found: if override_ok(opts): if verbose: say("Launch option: set ({}).".format(name)) else: good = False if verbose: say("Launch option: missing for {}.".format(name)) if opts.strip(): say(" Yours now: " + opts.strip()) say(" Change it to: " + suggested_option(opts)) return good def print_launch_steps(current=""): say("") say("One step in Steam, once:") say(" 1. In your Library, right-click OPERATOR and pick Properties.") say(" 2. Under General, find Launch Options and paste this in:") say("") say(" " + suggested_option(current)) say("") say(" 3. Close the window. Start the game from Steam as usual.") say("Without it the game still starts, but no mods load.") # --------------------------------------------------------------------------- # what happened last time the game ran # --------------------------------------------------------------------------- def last_runs(inst): disk = Disk(inst.game) ml = disk.find("MelonLoader/Latest.log") ml_time = os.path.getmtime(str(ml)) if ml is not None else None player = None if inst.prefix is not None: base = inst.prefix / "drive_c" / "users" / "steamuser" / "AppData" player = ci_child(ci_child(ci_child(ci_child(base, "LocalLow"), "VECTOR INTERACTIVE"), "OPERATOR"), "Player.log") game_time = os.path.getmtime(str(player)) if player is not None and player.is_file() else None loaded, problem = None, None if ml is not None: try: text = ml.read_text(encoding="utf-8", errors="replace") except OSError: text = "" m = None for m in re.finditer(r"(\d+) Mods? loaded", text): pass loaded = int(m.group(1)) if m else None if "Failed to load Hostfxr" in text: problem = "the mod loader could not start its runtime (Failed to load Hostfxr)" return game_time, ml_time, loaded, problem def when(t): return time.strftime("%Y-%m-%d %H:%M", time.localtime(t)) if t else "never" # --------------------------------------------------------------------------- # commands # --------------------------------------------------------------------------- OPTS = {} def require_not_root(): if hasattr(os, "geteuid") and os.geteuid() == 0: raise Problem("Run this as yourself, without sudo. Files made by root lock Steam out of the game folder.") def accept_terms(cfg): if cfg.get("terms") == TERMS_VERSION: return if OPTS.get("accept_terms"): cfg["terms"] = TERMS_VERSION save_config(cfg) return if not sys.stdin or not sys.stdin.isatty(): raise Problem("The terms have not been accepted. Run this in a terminal, or add --accept-terms " "after reading them at " + TERMS_URL) say("BEFORE YOU START") say("") say("This tool installs mods into your copy of OPERATOR. These are the same terms") say("the Windows launcher shows. The full text: {} (version {}).".format(TERMS_URL, TERMS_VERSION)) for head, body in TERMS: say("") say(head) say(wrap(body, 78)) say("") try: answer = input("Type yes to accept, or anything else to stop: ").strip().lower() except EOFError: answer = "" if answer != "yes": raise Problem("Nothing was changed.") cfg["terms"] = TERMS_VERSION save_config(cfg) a = load_account() if a["key"]: record_consent(a["key"]) say("") def wrap(text, width): lines, line = [], "" for word in text.split(): if line and len(line) + 1 + len(word) > width: lines.append(line) line = word else: line = (line + " " + word).strip() if line: lines.append(line) return "\n".join(lines) def header(inst): say("ODG for Linux {}".format(TOOL_VERSION)) say("Game: {}".format(inst.game)) if not os.access(str(inst.game), os.W_OK): raise Problem("This user cannot write to the game folder. If you ever ran this with sudo, " "fix the folder's owner, then run this again.") def cmd_install(_args): require_not_root() cfg = load_config() accept_terms(cfg) self_update() inst = resolve_install(OPTS, cfg) if OPTS.get("game"): cfg["game"] = str(inst.game) save_config(cfg) header(inst) acct = load_account(inst) if acct["key"]: acct = refresh_account(acct, inst) save_account(acct, inst) say("Account: {} ({})".format(acct["name"] or "signed in", stamp_of(acct))) say("Checking for updates") public_doc = None if is_member(acct) else public_listing() up = Updater(inst) up.run(acct, public_doc) up.tidy(ask_leftovers=True) disk = up.disk state, missing = loader_state(disk) have = disk.read(STAMP).strip() say("") if up.failed: say("Finished with problems. Run this again to retry what was skipped.") elif have: say("ODG {} is installed.".format(have)) if state == "held": say("The mod loader is switched off (vanilla). Run this with 'modded' to turn it back on.") elif state != "ready" and not up.running: say("The mod loader is not complete: missing {}.".format(", ".join(missing))) if not launch_option_report(inst, verbose=False): found = launch_options(inst.roots) print_launch_steps(found[0][1] if found else "") say("Steam saves launch options when it closes. If you have already added it, ignore this.") if inst.prefix is None: say("") say("Start the game from Steam. The first start takes a few minutes while the mod loader") say("sets itself up.") if acct["key"]: say("After that first start, run this again so your membership reaches the game.") elif up.changed: say("") say("Start the game from Steam. The first start after an install takes a few minutes.") return 1 if up.failed else 0 def cmd_status(_args): cfg = load_config() inst = resolve_install(OPTS, cfg) disk = Disk(inst.game) say("ODG for Linux {}".format(TOOL_VERSION)) say("Game: {}".format(inst.game)) say("Proton: {}".format("the game has run on this PC" if inst.prefix else "not set up yet (start the game once)")) have = disk.read(STAMP).strip() latest = "" try: latest = str(public_listing().get("tag_name") or "") except Problem: pass acct = load_account(inst) line = have or "not installed" if latest and have and not is_member(acct): line += " (up to date)" if have == latest else " (the public release is {})".format(latest) say("Pack: " + line) maps = [l for l in disk.read(MAP_STAMP).splitlines() if l.strip()] say("Map data: {} installed".format(len(maps))) state, missing = loader_state(disk) say("Mod loader: " + { "ready": "ready", "held": "switched off (vanilla)", "absent": "not installed", "incomplete": "damaged, missing " + ", ".join(missing), }[state]) mods = mod_files(disk) on = sum(1 for m in mods if m[1]) say("Mods: {} on, {} off".format(on, len(mods) - on)) say("Account: {}".format("{} ({})".format(acct["name"] or "signed in", stamp_of(acct)) if acct["key"] else "not signed in (free)")) say("") ok = launch_option_report(inst) game_t, ml_t, loaded, problem = last_runs(inst) say("Game last ran: " + when(game_t)) say("Mod loader last ran: " + when(ml_t) + (", {} mods loaded".format(loaded) if loaded is not None else "")) if problem: say("Last run problem: " + problem) if game_t and (ml_t is None or game_t - ml_t > 120): say("") say("The game ran without the mod loader last time. The launch option is missing or mistyped.") found = launch_options(inst.roots) print_launch_steps(found[0][1] if found else "") elif not ok and state == "ready": print_launch_steps() return 0 def squash(s): return re.sub(r"[^a-z0-9]", "", s.lower()) def mod_keys(file): """What a player might type for a mod: 'Third Person', 'thirdperson', 'Map Loader'.""" f = file.lower() stem = f[:-4] if f.endswith(".dll") else f short = stem[:-4] if stem.endswith(".mod") else stem name = squash(display_name(file)) keys = {squash(f), squash(stem), squash(short), name} if name.startswith("odg"): keys.add(name[3:]) return keys def find_mods(disk, names): mods = mod_files(disk) picked = [] for want in names: w = squash(want) if w == "all": picked.extend(mods) continue if not w: raise Problem("Name the mod. 'mods' lists them.") keys = mod_keys exact = [m for m in mods if w in keys(m[0])] if not exact: exact = [m for m in mods if any(k.startswith(w) for k in keys(m[0]))] if not exact: raise Problem("No mod called '{}'. 'mods' lists them.".format(want)) if len(exact) > 1: raise Problem("'{}' matches {}. Be more exact.".format(want, ", ".join(display_name(m[0]) for m in exact))) picked.append(exact[0]) seen, out = set(), [] for m in picked: if m[0].lower() not in seen: seen.add(m[0].lower()) out.append(m) return out def cmd_mods(_args): inst = resolve_install(OPTS, load_config()) disk = Disk(inst.game) mods = mod_files(disk) if not mods: say("No mods are installed. Run this tool with no command to install ODG.") return 0 width = max(len(display_name(m[0])) for m in mods) for file, on, _off in mods: tag = "REQUIRED" if requirement(file) else ("on" if on else "off") say(" {} {}".format(display_name(file).ljust(width), tag)) say("") say("Switch one off with: python3 odg-linux.py off \"Third Person\"") return 0 def cmd_switch(args, enable): if not args: raise Problem("Name the mod, for example: python3 odg-linux.py {} \"Third Person\"".format("on" if enable else "off")) inst = resolve_install(OPTS, load_config()) disk = Disk(inst.game) for file, on, off in find_mods(disk, args): name = display_name(file) if not enable and requirement(file): if len(args) == 1 and args[0].lower() != "all": say("{} stays on. {}".format(name, requirement(file))) continue if set_mod(disk, file, on, off, enable): say("{} is {}.".format(name, "on" if enable else "off")) else: say("{} was already {}.".format(name, "on" if enable else "off")) if game_running(): say("The game is running. The change counts from its next start.") return 0 def cmd_vanilla(_args, on): inst = resolve_install(OPTS, load_config()) disk = Disk(inst.game) proxy, held = disk.find("version.dll"), disk.find(HELD) if not on: if proxy is None: say("The mod loader is already off." if held else "The mod loader is not installed.") return 0 os.replace(str(proxy), str(inst.game / HELD)) say("The mod loader is off. The game starts without mods. Turn it back on with: python3 odg-linux.py modded") else: if held is None: say("The mod loader is already on." if proxy else "The mod loader is not installed. Run this tool with no command.") return 0 if proxy is None: os.replace(str(held), str(inst.game / "version.dll")) else: remove_quietly(held) say("The mod loader is on again.") return 0 def open_browser(url): if os.environ.get("ODG_NO_BROWSER"): return False try: import webbrowser return webbrowser.open(url) except Exception: return False def cmd_signin(args): require_not_root() cfg = load_config() accept_terms(cfg) inst = None try: inst = resolve_install(OPTS, cfg) except Problem: pass acct = load_account(inst) how = args[0].strip() if args else "" if not how: say("Sign in with Patreon, with Steam, or with an invite code from us.") try: how = input("Type patreon, steam, or your code: ").strip() except EOFError: how = "" if not how: raise Problem("Nothing was changed.") now = now_unix() if how.lower() not in ("patreon", "steam"): code = how.upper() if len(code) < 6: raise Problem("That code is too short.") try: doc = account_call("/api/launcher/redeem", body={"code": code}) except NoAnswer as e: raise Problem("Could not reach the account server: {}".format(e)) if doc.get("error"): raise Problem(str(doc["error"])) key = str(doc.get("key") or "") if not key: raise Problem("The server did not issue a key.") a = from_answer(doc, key, "", now) else: key = acct["key"] or None try: if how.lower() == "steam": owner = operator_owner(inst) if inst else None pc = steam_pc_account(inst.roots if inst else steam_roots()) doc = account_call("/api/launcher/steam/start", key=key, body={ "owner": owner or "", "pc": pc["id"] if pc else "", "pc_name": pc["name"] if pc else ""}) else: doc = account_call("/api/launcher/link/start", key=key, body={}) except NoAnswer as e: raise Problem("Could not reach the account server: {}".format(e)) if doc.get("error"): raise Problem(str(doc["error"])) code, secret, url = str(doc.get("code") or ""), str(doc.get("secret") or ""), str(doc.get("url") or "") if not (code and secret and url): raise Problem("The server did not start a sign-in.") say("") say("Open this address in your browser and sign in there:") say(" " + url) say("Your code: " + code) open_browser(url) say("") say("Waiting for the sign-in to finish. Ctrl+C stops waiting.") a = None deadline = time.time() + 15 * 60 while time.time() < deadline: time.sleep(2) try: doc = account_call("/api/launcher/link/poll?code={}&secret={}".format( urllib.parse.quote(code), urllib.parse.quote(secret))) except NoAnswer: continue if doc.get("error"): raise Problem(str(doc["error"])) if doc.get("pending") or not doc.get("key"): continue a = from_answer(doc, str(doc["key"]), "", now_unix()) break if a is None: raise Problem("That sign-in expired. Start again.") save_account(a, inst) record_consent(a["key"]) say("Signed in as {} ({}).".format(a["name"] or "a member", stamp_of(a))) if inst is not None and inst.prefix is None: say("Start the game once, then run this tool again, so your membership reaches the game.") else: say("Run this tool with no command to get the build your tier is on.") return 0 def cmd_signout(_args): inst = None try: inst = resolve_install(OPTS, load_config()) except Problem: pass save_account(blank_account(), inst) say("Signed out. This PC gets the public release from now on.") return 0 def cmd_uninstall(_args): require_not_root() cfg = load_config() inst = resolve_install(OPTS, cfg) disk = Disk(inst.game) if game_running(): raise Problem("Close OPERATOR first.") if disk.find(DEV_HOLD) is not None: raise Problem("This folder is held by .odg-dev-hold. Nothing was removed.") mods = disk.find("Mods") entries, seen = [], set() def add(rel): lower = rel.lower() if not safe_rel(rel) or is_custom(lower) or lower in seen: return seen.add(lower) p = disk.find(rel) if p is None: return try: st = os.lstat(str(p)) except OSError: return if stat.S_ISLNK(st.st_mode): return entries.append((disk.rel(p), 0 if stat.S_ISDIR(st.st_mode) else st.st_size, stat.S_ISDIR(st.st_mode))) stems = [] for rel in read_list(disk.read(SHIPPED_LIST)): add(rel) if rel.lower().endswith(".dll"): add(rel + OFF) if rel.startswith("Mods/") and "/" not in rel[5:] and rel.lower().endswith(".dll"): stems.append(rel[5:].split(".")[0]) for file in KNOWN: real = ci_child(mods, file) if mods else None name = real.name if real is not None else file add("Mods/" + name) add("Mods/" + name + OFF) stems.append(name.split(".")[0]) for m in UNINSTALL_MARKERS: add(m) for rel, _s, _d in stale_scan(disk): add(rel.rstrip("/")) stems = [s.lower() for s in stems if len(s) >= 3] others = [] if mods is not None: for n in os.listdir(str(mods)): low = n.lower() if (low.endswith(".log") or low.endswith(".txt")) and os.path.isfile(str(mods / n)): if any(len(low) > len(s) and low.startswith(s) and low[len(s)] in "._-" for s in stems): add("Mods/" + n) ours = {e[0].lower() for e in entries} for n in os.listdir(str(mods)): low = n.lower() if (low.endswith(".dll") or low.endswith(".dll" + OFF)) and os.path.isfile(str(mods / n)) \ and "mods/" + low not in ours: others.append(n[:-len(OFF)] if low.endswith(OFF) else n) proxy = disk.find("version.dll") or disk.find(HELD) loader = proxy is not None and is_melonloader(proxy) size = sum(s for _r, s, _d in entries) say("ODG for Linux {}".format(TOOL_VERSION)) say("Game: {}".format(inst.game)) say("") say("This removes the ODG pack: {} files and folders, {}.".format(len(entries), mb(size))) say("It keeps OPERATOR, your saves, your mod settings and outfits, and your own maps.") take_loader = False if loader: if others: say("Other mods use the mod loader: {}.".format(", ".join(sorted(set(others))))) take_loader = ask("Remove the mod loader too?", default=False) else: say("The mod loader goes too.") take_loader = True if not (OPTS.get("yes") or ask("Remove ODG now?", default=False)): raise Problem("Nothing was changed.") removed, failed = delete_entries(disk, entries) # Folders the pack owned and left empty, deepest first. Never Custom. dirs = set() for rel, _s, _d in entries: p = inst.game / rel while p.parent != inst.game and mods is not None and mods in p.parents: p = p.parent if p == mods: break if "/maploader_assets/custom" not in disk.rel(p).lower(): dirs.add(p) for d in sorted(dirs, key=lambda x: len(x.parts), reverse=True): try: d.rmdir() except OSError: pass diag = ci_child(ci_child(inst.game, "UserData"), "ODG Diagnostics") if diag is not None and diag.is_dir() and not diag.is_symlink(): shutil.rmtree(str(diag), ignore_errors=True) if take_loader: for name in ("version.dll", HELD): p = ci_child(inst.game, name) if p is not None and is_melonloader(p): try: os.remove(str(p)) except OSError as e: failed.append((name, str(e))) if ci_child(inst.game, "version.dll") is None and ci_child(inst.game, HELD) is None: ml = ci_child(inst.game, "MelonLoader") if ml is not None and ml.is_dir() and not ml.is_symlink(): shutil.rmtree(str(ml), ignore_errors=True) for d in ("Plugins", "UserLibs", "Mods"): p = ci_child(inst.game, d) if p is not None: try: p.rmdir() except OSError: pass data = ci_child(ci_child(inst.game, "UserData"), "ODGModManager") if data is not None and data.is_dir() and not data.is_symlink(): shutil.rmtree(str(data), ignore_errors=True) pa = prefix_account_path(inst) if pa is not None: remove_quietly(pa) remove_quietly(config_dir() / "account.json") say("Removed {} item{}.".format(len(removed), "" if len(removed) == 1 else "s")) for rel, why in failed[:10]: say(" could not remove {}: {}".format(rel, why)) say("") say("Two things left for you:") say(" 1. In Steam, clear OPERATOR's Launch Options (Properties > General).") say(" 2. In Steam, right-click OPERATOR > Properties > Installed Files > Verify integrity") say(" of game files. That puts back any game file the pack replaced.") say("This file, odg-linux.py, can be deleted.") return 1 if failed else 0 HELP = """ODG for Linux {v} python3 odg-linux.py install ODG, or update it python3 odg-linux.py status what is installed, and whether Steam is set up python3 odg-linux.py mods list the mods python3 odg-linux.py off NAME switch a mod off ("on NAME" switches it back) python3 odg-linux.py vanilla start the game without mods ("modded" undoes it) python3 odg-linux.py signin sign in with Patreon, Steam, or an invite code python3 odg-linux.py signout forget the sign-in on this PC python3 odg-linux.py uninstall take ODG off this PC Options: --game DIR your OPERATOR folder, when it is not found by itself --yes answer yes to every question except the terms --accept-terms accept the terms without being asked ({url}) """ def parse_args(argv): opts, rest = {}, [] i = 0 while i < len(argv): a = argv[i] if a in ("-h", "--help"): rest.insert(0, "help") elif a in ("-y", "--yes"): opts["yes"] = True elif a == "--accept-terms": opts["accept_terms"] = True elif a == "--no-self-update": opts["no_self_update"] = True elif a == "--game" or a.startswith("--game="): if a == "--game": i += 1 if i >= len(argv): raise Problem("--game needs the folder after it.") opts["game"] = argv[i] else: opts["game"] = a.split("=", 1)[1] elif a.startswith("-") and len(a) > 1: raise Problem("Unknown option {}. 'help' lists them.".format(a)) else: rest.append(a) i += 1 cmd = rest[0].lower() if rest else "install" return cmd, rest[1:], opts def main(argv=None): if sys.version_info < (3, 8): print("ODG for Linux needs Python 3.8 or newer.", file=sys.stderr) return 2 try: cmd, args, opts = parse_args(sys.argv[1:] if argv is None else argv) OPTS.clear() OPTS.update(opts) commands = { "install": cmd_install, "update": cmd_install, "status": cmd_status, "mods": cmd_mods, "list": cmd_mods, "on": lambda a: cmd_switch(a, True), "off": lambda a: cmd_switch(a, False), "vanilla": lambda a: cmd_vanilla(a, False), "modded": lambda a: cmd_vanilla(a, True), "signin": cmd_signin, "login": cmd_signin, "signout": cmd_signout, "logout": cmd_signout, "uninstall": cmd_uninstall, } if cmd == "help": print(HELP.format(v=TOOL_VERSION, url=TERMS_URL)) return 0 if cmd not in commands: raise Problem("Unknown command '{}'. 'help' lists them.".format(cmd)) lock = _lock() try: return commands[cmd](args) or 0 finally: if lock is not None: lock.close() except Problem as e: print("", file=sys.stderr) print(str(e), file=sys.stderr) return 1 except KeyboardInterrupt: print("\nStopped. Run this again to finish.", file=sys.stderr) return 130 def _lock(): """One copy at a time: two unpacking over one game folder would mix packs.""" try: import fcntl except ImportError: return None d = config_dir() try: d.mkdir(parents=True, exist_ok=True) f = open(str(d / "lock"), "w") except OSError: return None try: fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: f.close() raise Problem("ODG for Linux is already running in another window.") return f if __name__ == "__main__": sys.exit(main())