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
|
"""
webterm
~~~~~~~
An example showing how to use :mod:`pyte` to implement a basic
single-user web terminal.
Client-side ``webterm.js`` supports
* incremental rendering via :data:`~pyte.screens.DiffScreen.dirty`,
* most of the common keyboard events,
* pagination on Meta + P/Meta + A.
.. note:: This example requires at least Python 3.5 and a recent
version of ``aiohttp`` library.
.. seealso::
`The TTY demystified <http://www.linusakesson.net/programming/tty>`_
for an introduction to the inner workings of the TTY subsystem.
:copyright: (c) 2017 by pyte authors and contributors,
see AUTHORS for details.
:license: LGPL, see LICENSE for more details.
"""
import json
import os
import pty
import shlex
import signal
import webbrowser
from pathlib import Path
import aiohttp
import asyncio
from aiohttp import web
import pyte
class Terminal:
def __init__(self, columns, lines, p_in):
self.screen = pyte.HistoryScreen(columns, lines)
self.screen.set_mode(pyte.modes.LNM)
self.screen.write_process_input = \
lambda data: p_in.write(data.encode())
self.stream = pyte.ByteStream()
self.stream.attach(self.screen)
def feed(self, data):
self.stream.feed(data)
def dumps(self):
cursor = self.screen.cursor
lines = []
for y in self.screen.dirty:
line = self.screen.buffer[y]
data = [(char.data, char.reverse, char.fg, char.bg)
for char in (line[x] for x in range(self.screen.columns))]
lines.append((y, data))
self.screen.dirty.clear()
return json.dumps({"c": (cursor.x, cursor.y), "lines": lines})
def open_terminal(command="bash", columns=80, lines=24):
p_pid, master_fd = pty.fork()
if p_pid == 0: # Child.
argv = shlex.split(command)
env = dict(TERM="linux", LC_ALL="en_GB.UTF-8",
COLUMNS=str(columns), LINES=str(lines))
os.execvpe(argv[0], argv, env)
# File-like object for I/O with the child process aka command.
p_out = os.fdopen(master_fd, "w+b", 0)
return Terminal(columns, lines, p_out), p_pid, p_out
async def websocket_handler(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
request.app["websockets"].add(asyncio.Task.current_task())
terminal, p_pid, p_out = open_terminal()
ws.send_str(terminal.dumps())
def on_master_output():
terminal.feed(p_out.read(65536))
ws.send_str(terminal.dumps())
loop = asyncio.get_event_loop()
loop.add_reader(p_out, on_master_output)
try:
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
if msg.data == pyte.control.ESC + "N":
terminal.screen.next_page()
ws.send_str(terminal.dumps())
elif msg.data == pyte.control.ESC + "P":
terminal.screen.prev_page()
ws.send_str(terminal.dumps())
else:
p_out.write(msg.data.encode())
elif msg.type == aiohttp.WSMsgType.ERROR:
raise ws.exception()
except (asyncio.CancelledError,
OSError): # Process died?
pass
finally:
loop.remove_reader(p_out)
os.kill(p_pid, signal.SIGTERM)
p_out.close()
if not is_shutting_down:
request.app["websockets"].remove(asyncio.Task.current_task())
await ws.close()
return ws
is_shutting_down = False
async def on_shutdown(app):
"""Closes all WS connections on shutdown."""
global is_shutting_down
is_shutting_down = True
for task in app["websockets"]:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
app = web.Application()
app["websockets"] = set()
app.router.add_get("/ws", websocket_handler)
app.router.add_static("/", Path(__file__).parent / "static",
show_index=True)
app.on_shutdown.append(on_shutdown)
webbrowser.open_new_tab("http://localhost:8080/index.html")
web.run_app(app)
|
