diff options
Diffstat (limited to 'ivt.py')
| -rw-r--r-- | ivt.py | 425 |
1 files changed, 425 insertions, 0 deletions
@@ -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()) |