Initial Commit

1 files changed, 425 insertions, 0 deletions

diff --git a/ivt.py b/ivt.py
new file mode 100644
index 0000000..a2683bd
--- /dev/null
+++ b/ivt.py
@@ -0,0 +1,425 @@
+# Index Virtual Terminal (IVT) Webserver
+# Copyright (C) 2026 Morgana <morgana@icolotl.com>
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import asyncio
+from random import choice, randint
+from string import ascii_uppercase
+from json import dumps, loads, JSONDecodeError
+from argparse import ArgumentParser, ArgumentError, Namespace, _SubParsersAction
+from typing import Literal
+import sqlite3
+from hashlib import scrypt
+from os import urandom, listdir
+from hmac import compare_digest
+from colorsys import rgb_to_hsv
+
+import requests
+from websockets.asyncio.server import serve, ServerConnection
+
+
+class IVTException(Exception): ...
+
+
+HEX_CHARS = set("0123456789ABCDEF")
+SCRYPT_PARAMS = {"n": 16384, "r": 8, "p": 1}
+
+
+class IVTModel:
+    """Represents a Index Virtual Terminal Worldstate. Only one instance of this class is created per server."""
+
+    def __init__(self) -> None:
+        self.users: dict[str, IVTSession] = {}
+        self.subparsers: dict[str, ArgumentParser] = {}
+        self.parser: ArgumentParser = self.makeParser()
+
+        self.db = sqlite3.connect("data/akasha")
+
+        cur = self.db.cursor()
+        cur.execute("SELECT value FROM config WHERE key = 'light_url'")
+        self.light_url = cur.fetchone()[0]
+
+    async def newSession(self, websocket: ServerConnection) -> None:
+        name = self.genNewAnonName()
+        session = IVTSession(self, name, websocket)
+        self.users[name] = session
+        await session.mainloop()
+
+    def genNewAnonName(self) -> str:
+        name = "GUEST" + str(randint(0, 99)).zfill(2) + choice(ascii_uppercase)
+        if name in self.users:
+            return self.genNewAnonName()
+        return name
+
+    def changeUsername(self, oldname: str, newname: str) -> None:
+        if oldname not in self.users:
+            raise IVTException(f"unable to find user '{oldname}'")
+        if newname in self.users:
+            raise IVTException(f"user '{newname}' already exists")
+        if oldname == newname:
+            return
+        self.users[newname] = self.users[oldname]
+        self.users[newname].username = newname
+        del self.users[oldname]
+
+    def addSubparser(
+        self, subparsers: _SubParsersAction[ArgumentParser], name: str
+    ) -> ArgumentParser:
+        subparser = subparsers.add_parser(
+            name,
+            suggest_on_error=False,
+            exit_on_error=False,
+            add_help=False,
+            color=False,
+        )
+        self.subparsers[name] = subparser
+        return subparser
+
+    async def main(self) -> None:
+        print("Starting INDEX Server...")
+        try:
+            async with serve(self.newSession, "localhost", 5417) as server:
+                await server.serve_forever()
+        except (InterruptedError, asyncio.exceptions.CancelledError):
+            print("Exiting.")
+
+    def makeParser(self) -> ArgumentParser:
+        parser = ArgumentParser(
+            "ivt",
+            suggest_on_error=False,
+            exit_on_error=False,
+            add_help=False,
+            color=False,
+        )
+        subparsers = parser.add_subparsers(required=True, dest="cmd")
+
+        self.addSubparser(subparsers, "help")
+        self.addSubparser(subparsers, "list")
+
+        parser_usage = self.addSubparser(subparsers, "usage")
+        parser_usage.add_argument("subparser", type=str, nargs="?")
+
+        self.addSubparser(subparsers, "clear")
+
+        parser_login = self.addSubparser(subparsers, "login")
+        parser_login.add_argument("username", type=str)
+        parser_login.add_argument("-r", "--register", action="store_true")
+
+        parser_lighting = self.addSubparser(subparsers, "lighting")
+        parser_lighting.add_argument("state", type=str, choices=["on", "off"])
+        parser_lighting.add_argument("-l", "--light", type=int, default=5)
+        parser_lighting.add_argument(
+            "-c", "--color", type=str, nargs="?", const="#FF00AA", default="NIL"
+        )
+
+        parser_music = self.addSubparser(subparsers, "music")
+        parser_music.add_argument("album", type=str, nargs="?")
+        parser_music.add_argument("song", type=str, nargs="?")
+
+        return parser
+
+    def parse(self, line: str) -> ArgumentError | Namespace:
+        try:
+            return self.parser.parse_args(line.split())
+        except ArgumentError as error:
+            return error
+
+    def checkPerms(self, username: str, command: str) -> bool:
+        cur = self.db.cursor()
+        res = cur.execute("SELECT public FROM commands WHERE command = ?", (command,))
+        cmddata = res.fetchone()
+        if not cmddata:
+            return False
+        if cmddata[0]:
+            return True
+        res = cur.execute(
+            "SELECT 1 FROM perms WHERE username = ? AND command = ?",
+            (username, command),
+        )
+        return res.fetchone() is not None
+
+    def tool_disable_light(self, light: int):
+        requests.put(
+            f"{self.light_url}/lights/{light}/state",
+            json={"on": False},
+            timeout=5,
+        )
+
+    def tool_enable_light(self, light: int):
+        requests.put(
+            f"{self.light_url}/lights/{light}/state",
+            json={"on": True, "ct": 366},
+            timeout=5,
+        )
+
+    def tool_color_light(self, light: int, color: str):
+        if color[0] == "#":
+            color = color[1:]
+        color = color.upper()
+        if len(color) != 6 or not all(c in HEX_CHARS for c in color):
+            color = "FF00AA"
+        rgb = [int(color[i : i + 2], 16) / 255 for i in range(0, 6, 2)]
+        hsv = rgb_to_hsv(*rgb)
+        requests.put(
+            f"{self.light_url}/lights/{light}/state",
+            json={
+                "on": True,
+                "hue": int(hsv[0] * 65535),
+                "sat": int(hsv[1] * 254),
+                "bri": int(hsv[2] * 254),
+            },
+            timeout=5,
+        )
+
+
+class IVTSession:
+    """Represents a Index Virtual Terminal Session. A new instance of this class is created for each connection."""
+
+    def __init__(
+        self, parent: IVTModel, username: str, websocket: ServerConnection
+    ) -> None:
+        self.parent = parent
+        self.db = self.parent.db
+        self.username = username
+        self.socket = websocket
+        self.pwdreturn: str | None = None
+        self.pwddata: Namespace | None = None
+
+    def changeUsername(self, newname: str) -> None:
+        self.parent.changeUsername(self.username, newname)
+
+    def checkPerms(self, command: str) -> bool:
+        return self.parent.checkPerms(self.username, command)
+
+    async def mainloop(self) -> None:
+        await self.socket.send(dumps({"act": "activate", "username": self.username}))
+        await self.send_text(f"Connection established; Logged in as {self.username}.")
+        await self.send_display(
+            [
+                {"txt": "For more information, please see "},
+                {"tag": "button", "txt": "help", "term": "help"},
+                {"txt": ", "},
+                {"tag": "button", "txt": "list", "term": "list"},
+                {"txt": ", or "},
+                {"tag": "button", "txt": "usage", "term": "usage"},
+                {"txt": "."},
+            ]
+        )
+        async for message in self.socket:
+            if not isinstance(message, str):
+                await self.send_error("incorrect data type")
+                continue
+            await self.process(message)
+
+    async def process(self, line: str) -> None:
+        if self.pwdreturn and self.pwddata:
+            match self.pwdreturn:
+                case "register":
+                    await self.register_user(self.pwddata.username.upper(), line)
+                case "login":
+                    await self.login_user(self.pwddata.username.upper(), line)
+            self.pwdreturn = None
+            self.pwddata = None
+            return
+        result = self.parent.parse(line)
+        if isinstance(result, ArgumentError):
+            await self.send_error(result.message)
+            return
+        if not self.checkPerms(result.cmd):
+            await self.send_error("permission denied")
+            return
+        match result.cmd:
+            case "login":
+                await self.login_prompt(result)
+            case "usage":
+                await self.display_usage(result)
+            case "lighting":
+                await self.set_lighting(result)
+            case "music":
+                await self.play_music(result)
+            case "help":
+                await self.display_help()
+            case "list":
+                await self.display_list()
+            case "clear":
+                await self.send_action("clear")
+            case unknown:
+                await self.send_error(f"command not found: {unknown}")
+
+    async def login_prompt(self, result: Namespace):
+        if result.register:
+            await self.send_password_request(
+                "register",
+                result,
+                f"REGISTER/{result.username.upper()}>",
+                f"*** requesting password for {result.username.upper()} ***",
+            )
+        else:
+            await self.send_password_request(
+                "login",
+                result,
+                f"LOGIN/{result.username.upper()}>",
+                f"*** requesting password for {result.username.upper()} ***",
+            )
+
+    async def register_user(self, username: str, password: str) -> None:
+        cur = self.db.cursor()
+        res = cur.execute("SELECT 1 FROM users WHERE username = ?", (username,))
+        if res.fetchone():
+            await self.send_error("user already exists")
+            return
+        salt = urandom(32)
+        hash = scrypt(password.encode("utf-8"), salt=salt, **SCRYPT_PARAMS)
+        cur.execute("INSERT INTO users VALUES (?, ?, ?)", (username, hash, salt))
+        self.db.commit()
+        await self.send_text(f"User {username} registered.")
+
+    async def login_user(self, username: str, password: str) -> None:
+        cur = self.db.cursor()
+        res = cur.execute(
+            "SELECT hash, salt FROM users WHERE username = ?", (username,)
+        )
+        userdata = res.fetchone()
+        if not userdata:
+            await self.send_error("invalid login credentials")
+            return
+        salt = userdata[1]
+        expected_hash = userdata[0]
+        actual_hash = scrypt(password.encode("utf-8"), salt=salt, **SCRYPT_PARAMS)
+        if not compare_digest(expected_hash, actual_hash):
+            await self.send_error("invalid login credentials")
+            return
+        self.username = username
+        await self.socket.send(dumps({"act": "userchange", "username": username}))
+        await self.send_text(f"Login successful; Switched to user {self.username}.")
+
+    async def display_usage(self, result: Namespace) -> None:
+        if result.subparser and result.subparser in self.parent.subparsers:
+            await self.send_text(
+                "\n" + self.parent.subparsers[result.subparser].format_help() + "\n"
+            )
+        else:
+            await self.send_text("\n" + self.parent.parser.format_help() + "\n")
+
+    async def display_list(self) -> None:
+        await self.send_text("*** ivt command list ***")
+        for cmd in self.parent.subparsers:
+            await self.send_display(
+                [
+                    {"txt": " - "},
+                    {"tag": "button", "txt": cmd, "term": cmd},
+                    {"txt": " ("},
+                    {"tag": "button", "txt": "usage", "term": f"usage {cmd}"},
+                    {"txt": ")"},
+                ]
+            )
+
+    async def display_help(self) -> None:
+        await self.send_text(
+            "The Index Virtual Terminal (IVT) is a web-based terminal emulator that interfaces with other Index applications, enabling their control from a centralized location."
+        )
+        await self.send_display(
+            [
+                {
+                    "txt": "Many commands here do require logging in with an authorized Index account, which is possible with the "
+                },
+                {"tag": "button", "txt": "login <username>", "term": "usage login"},
+                {
+                    "txt": " command. For a list of all valid commands (plus shortcuts to their usage page), please see "
+                },
+                {"tag": "button", "txt": "list", "term": "list"},
+                {
+                    "txt": ". Lastly, do not forget that usage for any command can be seen with the "
+                },
+                {"tag": "button", "txt": "usage <command>", "term": "usage usage"},
+                {"txt": " command."},
+            ]
+        )
+
+    async def set_lighting(self, result: Namespace) -> None:
+        if result.state == "on":
+            if result.color and result.color != "NIL":
+                self.parent.tool_color_light(result.light, result.color)
+            else:
+                self.parent.tool_enable_light(result.light)
+        else:
+            self.parent.tool_disable_light(result.light)
+
+    async def play_music(self, result: Namespace) -> None:
+        album = result.album
+        song = result.song
+        if not album:
+            await self.send_text("*** ivt album list ***")
+            for other in listdir("music"):
+                await self.send_display(
+                    [
+                        {"txt": " - "},
+                        {"tag": "button", "txt": other, "term": f"music {other}"},
+                    ]
+                )
+            return
+        if album not in listdir("music"):
+            await self.send_error("album not found")
+            return
+        if not song:
+            await self.send_text(f"*** songs on {album} ***")
+            for other in listdir(f"music/{album}"):
+                await self.send_display(
+                    [
+                        {"txt": " - "},
+                        {
+                            "tag": "button",
+                            "txt": other.removesuffix(".mp3"),
+                            "term": f"music {album} {other.removesuffix(".mp3")}",
+                        },
+                    ]
+                )
+            return
+        if f"{song}.mp3" not in listdir(f"music/{album}"):
+            await self.send_error("song not found")
+            return
+        await self.send_text(f"Playing {album}/{song}...")
+        with open(f"music/{album}/{song}.mp3", "rb") as musicfile:
+            await self.send_prelude("music")
+            await self.socket.send(musicfile.read())
+
+    async def send_password_request(
+        self, pwdreturn: str, pwddata: Namespace, prompt: str, text: str
+    ) -> None:
+        self.pwdreturn = pwdreturn
+        self.pwddata = pwddata
+        await self.socket.send(
+            dumps({"act": "password", "prompt": prompt, "text": text})
+        )
+
+    async def send_error(self, text: str) -> None:
+        await self.socket.send(dumps({"act": "text", "text": f"error: {text}"}))
+
+    async def send_text(self, text: str) -> None:
+        await self.socket.send(dumps({"act": "text", "text": text}))
+
+    async def send_action(self, action: str) -> None:
+        await self.socket.send(dumps({"act": "action", "action": action}))
+
+    async def send_prelude(self, action: str) -> None:
+        await self.socket.send(dumps({"act": "prelude", "action": action}))
+
+    async def send_display(self, msg) -> None:
+        await self.socket.send(dumps({"act": "display", "msg": msg}))
+
+
+if __name__ == "__main__":
+    MAIN_MODEL = IVTModel()
+    asyncio.run(MAIN_MODEL.main())