diff options
| -rwxr-xr-x | service.py | 36 | ||||
| -rw-r--r-- | session.py | 92 | ||||
| -rwxr-xr-x | test_client.py | 2 |
3 files changed, 100 insertions, 30 deletions
@@ -1,4 +1,6 @@ #!/usr/bin/env python3 +import argparse + import json import time from threading import Thread @@ -8,9 +10,12 @@ import dbus.service from session import Session +from typing import Union class RWAService(dbus.service.Object): - def __init__(self): + def __init__(self, mockup_mode: bool): + self.mockup_mode = mockup_mode + self.bus = dbus.SessionBus() name = dbus.service.BusName("org.ArcticaProject.RWA", bus=self.bus) @@ -22,7 +27,7 @@ class RWAService(dbus.service.Object): def start(self): """Start a new remote session.""" # Start session - session = Session() + session = Session(mockup_mode) # Add session to sessions list self.sessions[session.pid] = session @@ -85,13 +90,38 @@ class RWAService(dbus.service.Object): self.update_service_running = False # TODO Probably kill daemon here (quit main loop) +def str2bool(v: Union[str, bool, int]) -> bool: + """Return true or false if the given string can be interpreted as a boolean otherwise raise an exception.""" + if isinstance(v, bool): + return v + if v.lower() in ('yes', 'true', 't', 'y', '1', 1): + return True + elif v.lower() in ('no', 'false', 'f', 'n', '0', 0): + return False + else: + raise argparse.ArgumentTypeError('Boolean value expected.') if __name__ == "__main__": + parser = argparse.ArgumentParser(description='DBus session service for ' + + 'ArcticaProject\'s ' + + 'Remote Web App') + parser.add_argument("-m", "--mockup-mode", type=str2bool, nargs='?', + const=True, default=False, + help="Activate mockup mode. Act like the session " + + "service but don\'t do changes or call other " + + "parts of RWA.") + + args = parser.parse_args() + mockup_mode = args.mockup_mode + + if mockup_mode: + print("All API responses are faked and should NOT BE USED IN " + + "PRODUCTION!") import dbus.mainloop.glib from gi.repository import GLib dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) loop = GLib.MainLoop() - object = RWAService() + object = RWAService(mockup_mode) loop.run() @@ -3,6 +3,8 @@ import secrets import signal from multiprocessing import Process +import random, string + import psutil import requests @@ -21,7 +23,8 @@ class Session: #: Remote has joined the session STATUS_JOINED = "joined" - def __init__(self): + def __init__(self, mockup_session: bool): + self.mockup_session = mockup_session self._generate_password() self._start_vnc() self._start_trigger_service() @@ -38,35 +41,61 @@ class Session: def _generate_password(self): """Generate password for x11vnc and save it.""" self.password = secrets.token_urlsafe(20) - self.pw_filename = save_password(self.password) + + # 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) def _start_vnc(self): - """Start x11vnc server.""" - process_info = run_vnc(self.pw_filename) + """Start x11vnc server if not in mockup_session mode.""" + if not self.mockup_session: + process_info = run_vnc(self.pw_filename) + + 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() + + # Use negative values to ensure we don't do something harmful + # to random processes + self.ws_pid = int('-' + ''.join(random.choice(string.digits) for _ in range(5))) + self.vnc_pid = int('-' + ''.join(random.choice(string.digits) for _ in range(5))) + + # Create a temporary file to indicate that this process is still 'Running' + filename = f"/tmp/rwa/{str(self.ws_port) + str(self.vnc_port) + str(self.ws_pid) + str(self.vnc_pid)}.lock" + new_file = open(filename, "w"); + new_file.write('this session is running') - 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"] def _register_session(self): - """Register session in RWA.""" - r = requests.post( - REGISTER_URL, - json={ - "port": self.ws_port, - "password": self.password, - "pid": self.vnc_pid, - "trigger_port": self.trigger_port, - "trigger_token": self.trigger_token, - }, - ) - print(r) - self.meta = r.json() - self.session_id = self.meta["session_id"] - self.web_url = self.meta["url"] - self.api_token = self.meta["token"] - self.pin = self.meta["pin"] + """Register session in RWA if not in mockup_session mode.""" + if not self.mockup_session: + r = requests.post( + REGISTER_URL, + json={ + "port": self.ws_port, + "password": self.password, + "pid": self.vnc_pid, + "trigger_port": self.trigger_port, + "trigger_token": self.trigger_token, + }, + ) + print(r) + self.meta = r.json() + self.session_id = self.meta["session_id"] + self.web_url = self.meta["url"] + self.api_token = self.meta["token"] + self.pin = self.meta["pin"] + else: + print("\"Registered\" in RWA") + self.meta = {} + self.session_id = int(''.join(random.choice(string.digits) for _ in range(10))) + self.web_url = "testhostname:" + ''.join(random.choice(string.digits) for _ in range(5)) + "/RWA/test/" + self.api_token = ''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in range(10)) + self.pin = int(''.join(random.choice(string.digits) for _ in range(5))) def _start_trigger_service(self): self.trigger_port = port_for.select_random() @@ -102,6 +131,15 @@ class Session: pass def stop(self): + if self.mockup_session: + filename = f"/tmp/rwa/{str(self.ws_port) + str(self.vnc_port) + str(self.ws_pid) + str(self.vnc_pid)}.lock" + if os.path.isfile(filename): + os.remove(filename); + + # Delete self + del self + return + """Stop session and clean up.""" # Kill VNC if self.vnc_pid in psutil.pids(): @@ -129,6 +167,10 @@ class Session: @property def vnc_process_running(self): + if self.mockup_session: + filename = f"/tmp/rwa/{str(self.ws_port) + str(self.vnc_port) + str(self.ws_pid) + str(self.vnc_pid)}.lock" + return os.path.isfile(filename) + """Check if the VNC process is still running.""" if self.vnc_pid in psutil.pids(): p = psutil.Process(self.vnc_pid) diff --git a/test_client.py b/test_client.py index 9cc6b68..4e60c29 100755 --- a/test_client.py +++ b/test_client.py @@ -3,9 +3,7 @@ import dbus bus = dbus.SessionBus() - time = bus.get_object("org.ArcticaProject.RWA", "/RWA") - curr = time.start() print("Your VNC session is", curr) |