path: root/rwa/support/sessionservice/session.py

# This file is part of Remote Support Desktop
# https://gitlab.das-netzwerkteam.de/RemoteWebApp/rwa.support.sessionservice
# Copyright 2020, 2021 Jonathan Weth <dev@jonathanweth.de>
# Copyright 2020 Daniel Teichmann <daniel.teichmann@das-netzwerkteam.de>
# Copyright 2020 Mike Gabriel <mike.gabriel@das-netzwerkteam.de>
# SPDX-License-Identifier: GPL-2.0-or-later
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the
# Free Software Foundation, Inc.,
# 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

import os
import platform
import secrets
import signal
import string
import subprocess  # noqa
from typing import Dict, Union

import port_for
import psutil
import requests

from .lock import TEMP_DIR_PATH
from .log import logging
from .vnc import run_vnc, save_password


def random_digits(length: int):
    return "".join(secrets.choice(string.digits) for _ in range(length))


def get_desktop_dir():
    """Get desktop directory from xdg vars."""
    output = subprocess.check_output(["xdg-user-dir", "DESKTOP"]).decode()  # noqa
    return output.strip().replace("\n", "")


class Session:
    #: Session is running
    STATUS_RUNNING = "running"

    #: Remote has joined the session
    STATUS_JOINED = "active"

    def __init__(self, host: str, trigger_port: int, mockup_session: bool = False):
        self.host = host
        self.BASE_URL = self.host + "/app/rwasupport/api/"
        self.REGISTER_URL = self.BASE_URL + "register/"
        self.STOP_URL = self.BASE_URL + "stop/"
        self.STATUS_URL = self.BASE_URL + "status/"
        self.MARK_JOB_AS_DONE_URL = self.BASE_URL + "jobs/mark_as_done/"
        logging.info(f"Load API config: {self.host}")

        self.trigger_token = secrets.token_urlsafe(20)
        self.trigger_port = trigger_port
        self.done_jobs = []
        self.mockup_session = mockup_session
        self.desktop_dir = get_desktop_dir()
        self.desktop_dir = get_desktop_dir()
        self._generate_password()
        self._start_vnc()
        self._register_session()
        self.status_text = self.STATUS_RUNNING

    @property
    def pid(self) -> int:
        return self.vnc_pid

    @property
    def port(self) -> int:
        return self.ws_port

    @property
    def _api_headers(self) -> Dict[str, str]:
        return {"Authorization": f"Token {self.api_token}"}

    @property
    def _mock_lock_file_path(self) -> str:
        return os.path.join(
            TEMP_DIR_PATH, f"{self.ws_port}-{self.vnc_port}-{self.ws_pid}-{self.vnc_pid}.lock",
        )

    def _generate_password(self):
        """Generate password for x11vnc and save it."""
        self.password = secrets.token_urlsafe(20)

        # Don't actually save a password if we just pretend to be a session.
        if not self.mockup_session:
            self.pw_filename = save_password(self.password)

        logging.info("The password for the session has been generated.")

    def _start_vnc(self):
        """Start x11vnc server if not in mockup_session mode."""
        if not self.mockup_session:
            process_info = run_vnc(self.pw_filename)

            logging.info("The VNC server has been started.")

            self.vnc_pid = process_info["vnc"]["pid"]
            self.vnc_port = process_info["vnc"]["port"]
            self.ws_pid = process_info["ws"]["pid"]
            self.ws_port = process_info["ws"]["port"]
        else:
            self.ws_port = port_for.select_random()
            self.vnc_port = port_for.select_random()

            self.ws_pid = int(random_digits(5))
            self.vnc_pid = int(random_digits(5))

            logging.info("The lock file for mocking a VNC server has been created.")

            # Create a temporary file to indicate that this process is still 'Running'
            new_file = open(self._mock_lock_file_path, "w")
            new_file.write("this session is running")

    def _register_session(self):
        """Register session in RWA.Support.WebApp if not in mockup_session mode."""
        if not self.mockup_session:
            try:
                r = requests.post(
                    self.REGISTER_URL,
                    json={
                        "port": self.ws_port,
                        "pid": self.vnc_pid,
                        "trigger_port": self.trigger_port,
                        "hostname": str(platform.node()),
                    },
                )
            except requests.exceptions.ConnectionError:
                self.stop(triggered=True)
                raise ConnectionError()

            logging.info(
                "The session has been registered in RWA.Support.WebApp "
                f"with status code {r.status_code} and response {r.content.decode()}."
            )

            if r.status_code != 200:
                self.stop(triggered=True)
                raise ConnectionError()

            self.meta = r.json()
            self.session_id = self.meta["session_id"]
            self.web_url = self.meta["url"]
            self.api_token = self.meta["token"]
        else:
            logging.info("The session has pretended that he had created a session.")
            self.meta = {}
            self.session_id = int(random_digits(10))
            self.web_url = f"http://example.com:{random_digits(5)}/app/rwasupport/test/"
            self.api_token = secrets.token_urlsafe(10)
        self.pin = int(random_digits(4))

    def trigger(self, data: dict, method: str = "trigger") -> Union[dict, bool]:
        """Event triggered by Django."""
        if method == "trigger" and data.get("token", "") == self.trigger_token:
            self.pull()
            return True
        elif method == "authenticate" and data.get("pin", "") == self.pin:
            return {
                "password": self.password,
                "trigger_token": self.trigger_token,
            }

        return False

    def pull(self):
        """Update status: Get status from Django."""
        if not self.mockup_session:
            try:
                r = requests.get(
                    self.STATUS_URL, params={"id": self.session_id}, headers=self._api_headers
                )

                logging.info(
                    "The session has received its status from RWA.Support.WebApp "
                    f"with status code {r.status_code} and response {r.content.decode()}."
                )
            except requests.ConnectionError:
                pass

            if r.status_code in (401, 402, 403, 404, 405):
                # Session doesn't exist anymore, so stop it local
                self.stop(triggered=True)
            else:
                self.status_text = r.json()["status"]
                self.jobs = r.json()["jobs"]
                self._do_jobs()

    def _do_jobs(self):
        """Go through all jobs and execute undone ones."""
        for job in self.jobs:
            if not job["done"] or job["job_id"] in self.done_jobs:
                job_type = job["job_type"]
                if job_type == "file":
                    self._do_file_job(job)

    def _do_file_job(self, job):
        """Download a file from server to the user's desktop."""
        logging.info(f"The session has received a file job and is downloading it now ({job}):")
        subprocess.Popen(["wget", job["file"], "-P", self.desktop_dir])  # noqa
        self._mark_job_as_done(job)

    def _mark_job_as_done(self, job):
        """Mark a job as done (in this service and on the server)."""
        self.done_jobs.append(job["job_id"])
        try:
            r = requests.post(
                self.MARK_JOB_AS_DONE_URL, params={"id": job["job_id"]}, headers=self._api_headers,
            )
            logging.info(
                f"The session has marked the job {job} as done in RWA.Support.WebApp "
                f"with status code {r.status_code} and response {r.content.decode()}."
            )
        except requests.ConnectionError:
            pass

    def push(self):
        """Update status: Push status to Django."""
        pass

    def stop(self, triggered: bool = False):
        """Stop session and clean up."""
        if self.mockup_session:
            logging.info("Mock session has been stopped by deleting its lock file.")
            if os.path.isfile(self._mock_lock_file_path):
                os.remove(self._mock_lock_file_path)

            # Delete self
            del self
            return

        # Kill websockify
        if self.ws_pid in psutil.pids():
            os.kill(self.ws_pid, signal.SIGINT)
            logging.info("The websockify server has been terminated.")

        # Kill VNC
        if self.vnc_pid in psutil.pids():
            os.kill(self.vnc_pid, signal.SIGINT)
            logging.info("The VNC server has been terminated.")

        # Delete PW file
        if os.path.exists(self.pw_filename):
            os.remove(self.pw_filename)
            logging.info("The VNC server password file has been removed.")

        self.push()

        if not triggered:
            try:
                r = requests.post(
                    self.STOP_URL, params={"id": self.session_id}, headers=self._api_headers
                )
                logging.info(
                    "The stop action has been registered in RWA.Support.WebApp "
                    f"with status code {r.status_code} and response {r.content.decode()}."
                )
            except requests.ConnectionError:
                pass

        self.status_text = "stopped"

        # Delete self
        del self

    @property
    def vnc_process_running(self) -> bool:
        """Check if the VNC process is still running."""
        if self.mockup_session:
            return os.path.isfile(self._mock_lock_file_path)

        if self.vnc_pid in psutil.pids():
            p = psutil.Process(self.vnc_pid)
            if p.status() == "zombie":
                return False
            return True
        return False

    @property
    def client_meta(self) -> Dict[str, Union[str, int]]:
        return {"id": self.pid, "session_id": self.session_id, "url": self.web_url, "pin": self.pin}

    @property
    def status(self) -> Dict[str, Union[str, int]]:
        return {"id": self.pid, "status": self.status_text}