1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
#!/usr/bin/env python3
import argparse
import json
import time
from threading import Thread
from typing import Union
import dbus
import dbus.service
from session import Session
class RWAService(dbus.service.Object):
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)
self.update_service_running = False
self.sessions = {}
super().__init__(name, "/RWA")
@dbus.service.method("org.ArcticaProject.RWA", out_signature="s")
def start(self):
"""Start a new remote session."""
# Start session
session = Session(mockup_mode)
# Add session to sessions list
self.sessions[session.pid] = session
# Start session update service
self._ensure_update_service()
return json.dumps(session.client_meta)
@dbus.service.method("org.ArcticaProject.RWA", in_signature="i", out_signature="s")
def status(self, pid: int) -> str:
"""Get status information about a service."""
return self._get_status(pid)
@dbus.service.method("org.ArcticaProject.RWA", in_signature="i", out_signature="s")
def refresh_status(self, pid: int) -> str:
"""Get status information about a service and refresh status before."""
self._update_session(pid)
return self._get_status(pid)
@dbus.service.method("org.ArcticaProject.RWA", in_signature="i", out_signature="s")
def stop(self, pid: int):
"""Stop a remote session."""
try:
session = self.sessions[pid]
except KeyError:
return json.dumps({"pid": pid, "status": "stopped"}, sort_keys=True)
session.stop()
return json.dumps({"id": pid, "status": "stopped"}, sort_keys=True)
def _get_status(self, pid: int) -> str:
try:
session = self.sessions[pid]
except KeyError:
return json.dumps({"id": pid, "status": "dead"}, sort_keys=True)
return json.dumps(session.status)
def _ensure_update_service(self):
"""Start session update thread if it isn't already running."""
if not self.update_service_running:
self.update_thread = Thread(target=self._update_sessions)
self.update_thread.start()
def _update_session(self, pid: int):
"""Update the status of a session."""
try:
session = self.sessions[pid]
except KeyError:
print(f"Session #{pid}")
print("Session is dead.")
return
print(f"Session #{session.pid}")
# Check if VNC process is still running
running = session.vnc_process_running
if running:
print("Session is running")
elif session.status_text == "stopped" and session.pid in self.sessions:
del self.sessions[session.pid]
else:
print("Session is dead.")
session.stop()
del self.sessions[session.pid]
def _update_sessions(self):
"""Go through all running sessions and update their status using ``_update_session``."""
while len(self.sessions.values()) > 0:
for session in list(self.sessions.values()):
self._update_session(session.pid)
time.sleep(2)
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(mockup_mode)
loop.run()
|
