ovenemprex/management.py

import subprocess
from pathlib import Path
from urllib.parse import urlparse

import cherrypy
from mako.template import Template

import config
import ovenapi


class Management:
    def __init__(self):
        self.page_template = Path("template/management.mako").read_text(encoding="utf-8")
        self.redirect_template = Path("template/message.mako").read_text(encoding="utf-8")
        self.api = ovenapi.get_api_handle(username=config.API_USER, password=config.API_PASS)

    @staticmethod
    def __verify_same_domain() -> bool:
        """
        Verify that the requested domain and referer domain match.

        This is mainly intended to be used as a guard for "destructive" endpoints such as
        /management/restart. This safeguards against cross-site requests as well as accidental
        history completions such as intents to access /management but one's browser helpfully
        populates /management/restart.
        """
        referer = cherrypy.request.headers.get("Referer", "").lower()

        # For comparing request and referer domains we drop the port
        referer_domain = urlparse(referer).netloc.split(":")[0]
        request_domain = urlparse(cherrypy.request.base).netloc.split(":")[0]

        return referer_domain == request_domain

    @staticmethod
    def __restart_server() -> None:
        subprocess.call(["sudo", "/usr/bin/systemctl", "restart", "ovenmediaengine"])

    def __message_and_redirect(self, message: str) -> bytes | str:
        return Template(self.redirect_template).render(message=message)

    @cherrypy.expose
    def restart(self) -> bytes | str:
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"

        # Blank our stream list because we're about to DC everyone
        config.LAST_STREAM_LIST = []

        self.__restart_server()

        # Compile and dispatch our response
        return self.__message_and_redirect("Server restarted")

    @cherrypy.expose
    def disconnect(self, target):
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"
        vhost, app, stream = target.split(":")
        self.api.disconnect_key(vhost, app, stream)
        return self.__message_and_redirect(f"Disconnected {target}")

    @cherrypy.expose
    def ban(self, target):
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"
        vhost, app, stream = target.split(":")
        ip = self.api.get_stream_ip(vhost, app, stream)
        if ip:
            config.BLOCKED_IPS.append(ip)
            self.disconnect(target)
            return self.__message_and_redirect(f"Banned {ip}")
        return self.__message_and_redirect("No stream found at that location or other error")

    @cherrypy.expose
    def unban(self, target):
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"
        if target in config.BLOCKED_IPS:
            config.BLOCKED_IPS.remove(target)
            return self.__message_and_redirect(f"Unbanned {target}")
        return self.__message_and_redirect(f"{target} not in ban list")

    @cherrypy.expose
    def disable(self, target):
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"
        config.DISABLED_KEYS.append(target)
        self.disconnect(target)
        return self.__message_and_redirect(f"Disabled key {target}")

    @cherrypy.expose
    def enable(self, target):
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"
        if target in config.DISABLED_KEYS:
            config.DISABLED_KEYS.remove(target)
            return self.__message_and_redirect(f"Re-enabled {target}")
        return self.__message_and_redirect(f"{target} not in disabled key list")

    @cherrypy.expose
    def stop(self):
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"
        config.DISABLED = True
        self.api.disconnect_all()
        return self.__message_and_redirect("Server disabled")

    @cherrypy.expose
    def start(self):
        if not self.__verify_same_domain():
            cherrypy.response.status = 403
            return "Cross-site request detected. Please go back to /management and try again"
        config.DISABLED = False
        return self.__message_and_redirect("Server re-enabled")

    @cherrypy.expose
    def default(self) -> bytes | str:
        if not (config.API_USER and config.API_PASS):
            cherrypy.response.status = 503
            return "Remote management is disabled on this node."

        data = self.api.get_all_stream_info()

        return Template(self.page_template).render(
            DISABLED=config.DISABLED,
            BLOCKED_IPS=config.BLOCKED_IPS,
            DISABLED_KEYS=config.DISABLED_KEYS,
            data=data,
        )