2024-08-21T0227Z

.gitignore

@@ -6,6 +6,7 @@
 
 *.spec
 *.sqlite
+GameConfig.toml
 Source/ssl/?*.*
 
 

Source/assets/__init__.py

@@ -147,6 +147,10 @@ def _load_redir_asset(self, asset_id: int | str, redirect: structs.asset_redirec
                     extract.process_command_line,
                 ),
             )
+        elif redirect.raw_data is not None:
+            return returns.construct(
+                data=redirect.raw_data,
+            )
         else:
             return returns.construct()
 

Source/config/structure.py

@@ -3,6 +3,7 @@
 from . import allocateable
 import util.versions
 import util.resource
+import textwrap
 
 
 class config_type(allocateable.obj_type):
@@ -42,37 +43,93 @@ class persistence(allocateable.obj_type):
         icon_path: wrappers.path_str = ''  # type:ignore
 
     class server_core(allocateable.obj_type):
-        chat_style: structs.chat_style
-
-        retrieve_default_user_code: callable[[float], str]
-
-        check_user_allowed: callable[[int, str], bool] = \
-            'function() return true end'  # type: ignore
-
-        check_user_has_admin: callable[[int, str], bool] = \
-            'function() return false end'  # type: ignore
-
-        retrieve_username: callable[[str], str]
-
-        retrieve_user_id: callable[[str], int]
-
-        retrieve_avatar_type: callable[[int, str], structs.avatar_type]
-
-        retrieve_avatar_items: callable[[int, str], list[int]]
-
-        retrieve_avatar_scales: callable[[int, str], structs.avatar_scales]
-
-        retrieve_avatar_colors: callable[[int, str], structs.avatar_colors]
-
-        retrieve_groups: callable[[int, str], dict[str, int]] = \
-            'function() return {} end'  # type: ignore
-
-        retrieve_account_age: callable[[int, str], int]
-        retrieve_default_funds: callable[[int, str], int]
-        filter_text: callable[[str, int, str], str]
+        chat_style: structs.chat_style = structs.chat_style.CLASSIC_CHAT
+
+        retrieve_default_user_code: callable[[float], str] = textwrap.dedent('''\
+        def f(tick):
+            return 'Player%d' % tick
+        ''')  # type: ignore
+
+        check_user_allowed: callable[[int, str], bool] = textwrap.dedent('''\
+        def f(*a):
+            return True
+        ''')  # type: ignore
+
+        check_user_has_admin: callable[[int, str], bool] = textwrap.dedent('''\
+        def f(*a):
+            return False
+        ''')  # type: ignore
+
+        retrieve_username: callable[[str], str] = textwrap.dedent('''\
+        def f(t, *a):
+            return t
+        ''')  # type: ignore
+
+        retrieve_user_id: callable[[str], int] = textwrap.dedent('''\
+        count = 0
+        def f(*a):
+            count += 1
+            return count
+        ''')  # type: ignore
+
+        retrieve_avatar_type: callable[[int, str], structs.avatar_type] = textwrap.dedent('''\
+        def f():
+            return 'R6'
+        ''')  # type: ignore
+
+        retrieve_avatar_items: callable[[int, str], list[int]] = textwrap.dedent('''\
+        def f():
+            return []
+        ''')  # type: ignore
+
+        retrieve_avatar_scales: callable[[int, str], structs.avatar_scales] = textwrap.dedent('''\
+        def f():
+            return {
+                "height": 1,
+                "width": 1,
+                "head": 1,
+                "depth": 1,
+                "proportion": 0,
+                "body_type": 0,
+            }
+        ''')  # type: ignore
+
+        retrieve_avatar_colors: callable[[int, str], structs.avatar_colors] = textwrap.dedent('''\
+        def f():
+            return {
+                "head": 1,
+                "left_arm": 1,
+                "left_leg": 1,
+                "right_arm": 1,
+                "right_leg": 1,
+                "torso": 1,
+            }
+        ''')  # type: ignore
+
+        retrieve_groups: callable[[int, str], dict[str, int]] = textwrap.dedent('''\
+        def f(*a):
+            return []
+        ''')  # type: ignore
+
+        retrieve_account_age: callable[[int, str], int] = textwrap.dedent('''\
+        def f(*a):
+            return 0
+        ''')  # type: ignore
+
+        retrieve_default_funds: callable[[int, str], int] = textwrap.dedent('''\
+        def f(*a):
+            return 0
+        ''')  # type: ignore
+
+        filter_text: callable[[str, int, str], str] = textwrap.dedent('''\
+        def f(t, *a):
+            return t
+        ''')  # type: ignore
 
     class remote_data(allocateable.obj_type):
         gamepasses: structs.gamepasses = []  # type: ignore
-        asset_redirects: callable[[int | str], structs.asset_redirect | None] = \
-            'function() return nil end'  # type: ignore
+        asset_redirects: callable[[int | str], structs.asset_redirect | None] = textwrap.dedent('''\
+        def f(*a):
+            return None
+        ''')  # type: ignore
         badges: structs.badges = []  # type: ignore

Source/config/types/callable.py

@@ -1,5 +1,5 @@
-from re import S
 from typing import ParamSpec, TypeVar, Generic
+import textwrap
 import enum
 
 
@@ -56,26 +56,34 @@ def assume_call_mode(self) -> call_mode_enum:
     def get_call(self):
         match self.call_mode:
             case call_mode_enum.lua:
-                def call(*args):
+                def call_lua(*args):
                     return self.config.data_transferer.call(
                         self.field_path, self.config, *args,
                     )
-                return call
+                return call_lua
             case call_mode_enum.python:
                 assert isinstance(self.rep, str)
                 local_vars = {}
+                modded_rep = (
+                    textwrap.dedent("""\
+                    def func():
+                    %s
+                        return next(
+                            v
+                            for v in reversed(locals().values())
+                            if callable(v)
+                        )
+                    """) %
+                    (textwrap.indent(self.rep, ' ' * 4),)
+                )
                 exec(
-                    self.rep,  # source
+                    modded_rep,  # source
                     {},  # globals
                     local_vars,  # locals
                 )
-                return next(
-                    v
-                    for v in reversed(local_vars.values())
-                    if callable(v)
-                )
+                return local_vars['func']()
             case call_mode_enum.dicted:
-                def call(*args):
+                def call_dicted(*args):
                     assert isinstance(self.rep, dict)
                     arg_strs = [str(a) for a in args]
                     candidate_keys = [
@@ -94,7 +102,7 @@ def call(*args):
                         ),
                         None,
                     )
-                return call
+                return call_dicted
             case _:
                 raise Exception(
                     "Config option `%s` doesn't seem to be in a valid format." %

Source/config/types/structs.py

@@ -23,13 +23,15 @@ class asset_redirect:
     def __post_init__(self):
         if sum([
             self.cmd_line is not None,
+            self.raw_data is not None,
             self.uri is not None,
         ]) > 1:
             raise Exception(
                 'Entries for `asset_redirects` should not have '
-                'both a `uri` and a pipeable `cmd_line`.'
+                'more than one of a `uri`, a pipeable `cmd_line`, or a `raw_data` chunk.'
             )
     uri: wrappers.uri_obj | None = None
+    raw_data: bytes | None = None
     cmd_line: str | None = None
 
 

Source/launcher/downloader/__init__.py

@@ -1,10 +1,12 @@
+import py7zr.exceptions
+from tqdm import tqdm
 import urllib.request
 import util.resource
 import util.versions
 import util.const
 import py7zr
 import io
-from tqdm import tqdm
+
 
 def get_remote_link(rōblox_version: util.versions.rōblox, bin_type: util.resource.bin_subtype) -> str:
     return util.const.GIT_LINK_FORMAT % (
@@ -13,37 +15,44 @@ def get_remote_link(rōblox_version: util.versions.rōblox, bin_type: util.resou
         bin_type.value,
     )
 
-def download_binary(rōblox_version: util.versions.rōblox, bin_type: util.resource.bin_subtype) -> None:
+
+def download(link: str) -> io.BytesIO:
+    with urllib.request.urlopen(link) as response:
+        if response.status != 200:
+            raise Exception(
+                "Failed to download: HTTP Status %d" %
+                (response.status),
+            )
+
+        total_size = int(response.info().get('Content-Length').strip())
+        response = io.BytesIO()
+
+        with tqdm(
+            total=total_size,
+            unit='B',
+            unit_scale=True,
+            unit_divisor=1024,
+            desc="Downloading",
+        ) as bar:
+            while True:
+                chunk = response.read(1024)
+                if not chunk:
+                    break
+                response.write(chunk)
+                bar.update(len(chunk))
+
+    response.seek(0)
+    return response
+
+
+def bootstrap_binary(rōblox_version: util.versions.rōblox, bin_type: util.resource.bin_subtype) -> None:
     link = get_remote_link(rōblox_version, bin_type)
-
-    try:
-        with urllib.request.urlopen(link) as response:
-            if response.status != 200:
-                raise Exception(f"Failed to download: HTTP Status {response.status}")
-
-            total_size = int(response.info().get('Content-Length').strip())
-            res = io.BytesIO()
-
-            with tqdm(total=total_size, unit='B', unit_scale=True, unit_divisor=1024, desc="Downloading") as bar:
-                while True:
-                    chunk = response.read(1024)
-                    if not chunk:
-                        break
-                    res.write(chunk)
-                    bar.update(len(chunk))
-
-        res.seek(0)
-
-        full_dir = util.resource.retr_rōblox_full_path(rōblox_version, bin_type)
-
-        print(f"Extracting to {full_dir}...")
-        try:
-            py7zr.unpack_7zarchive(res, full_dir)
-            print("Extraction completed successfully.")
-        except py7zr.exceptions.Bad7zFile as e:
-            raise Exception("Downloaded file is not a valid 7z archive") from e
-        except py7zr.exceptions.CrcError as e:
-            raise Exception(f"CRC error while extracting file. Error details: {e}") from e
-
-    except Exception as e:
-        raise Exception(f"An error occurred: {e}") from e
+    response = download(link)
+
+    full_dir = util.resource.retr_rōblox_full_path(
+        rōblox_version, bin_type,
+    )
+
+    print(f"Extracting to {full_dir}...")
+    py7zr.unpack_7zarchive(response, full_dir)
+    print("Done.")

Source/launcher/routines/_logic.py

@@ -212,7 +212,7 @@ def maybe_download_binary(self) -> None:
                 'Downloading zipped "%s" for Rōblox version %s...' %
                 (self.BIN_SUBTYPE.name, self.rōblox_version.get_number())
             )
-            downloader.download_binary(self.rōblox_version, self.BIN_SUBTYPE)
+            downloader.bootstrap_binary(self.rōblox_version, self.BIN_SUBTYPE)
             print('Download completed!')
         else:
             raise Exception(

Source/launcher/routines/download.py

@@ -22,7 +22,7 @@ def __init__(self, rōblox_version: util.versions.rōblox, bin_subtype: str):
         self.bin_subtype = bin_subtype
 
     def process(self) -> None:
-        downloader.download_binary(
+        downloader.bootstrap_binary(
             self.rōblox_version,
             self.BIN_SUBTYPE,
         )

Source/launcher/startup_scripts/rcc_server.py

@@ -1,7 +1,9 @@
 import data_transfer
+import textwrap
 import config
 
-BASE_SCRIPT_FORMAT = """%(rcc_snippet)s
+BASE_SCRIPT_FORMAT = """\
+%(rcc_snippet)s
 local BaseUrl = game:GetService("ContentProvider").BaseUrl:lower()
 local HttpRbxApiService = game:GetService("HttpRbxApiService")
 local HttpService = game:GetService("HttpService")