path: root/ivt.py

# 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, mkdir, link
from os.path import isdir
from uuid import uuid4, UUID
from hmac import compare_digest
from colorsys import rgb_to_hsv
from shutil import rmtree

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]

        if not isdir("static/dl"):
            mkdir("static/dl")

    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.")
            rmtree("static/dl")

    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}...")
        songuuid = uuid4()
        link(f"music/{album}/{song}.mp3", f"static/dl/{songuuid}.mp3")
        await self.send_music(songuuid)

    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_music(self, uuid: UUID) -> None:
        await self.socket.send(dumps({"act": "music", "uuid": str(uuid)}))

    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())