Source code for evennia.server.portal.webclient

"""
Webclient based on websockets.

This implements a webclient with WebSockets (http://en.wikipedia.org/wiki/WebSocket)
by use of the autobahn-python package's implementation (https://github.com/crossbario/autobahn-python).
It is used together with evennia/web/media/javascript/evennia_websocket_webclient.js.

All data coming into the webclient is in the form of valid JSON on the form

`["inputfunc_name", [args], {kwarg}]`

which represents an "inputfunc" to be called on the Evennia side with *args, **kwargs.
The most common inputfunc is "text", which takes just the text input
from the command line and interprets it as an Evennia Command: `["text", ["look"], {}]`

"""

import html
import json
import re

from autobahn.exception import Disconnected
from autobahn.twisted.websocket import WebSocketServerProtocol
from django.conf import settings

from evennia.utils.ansi import parse_ansi
from evennia.utils.text2html import parse_html
from evennia.utils.utils import class_from_module, mod_import

_RE_SCREENREADER_REGEX = re.compile(
    r"%s" % settings.SCREENREADER_REGEX_STRIP, re.DOTALL + re.MULTILINE
)
_CLIENT_SESSIONS = mod_import(settings.SESSION_ENGINE).SessionStore
_UPSTREAM_IPS = settings.UPSTREAM_IPS

# Status Code 1000: Normal Closure
#   called when the connection was closed through JavaScript
CLOSE_NORMAL = WebSocketServerProtocol.CLOSE_STATUS_CODE_NORMAL

# Status Code 1001: Going Away
#   called when the browser is navigating away from the page
GOING_AWAY = WebSocketServerProtocol.CLOSE_STATUS_CODE_GOING_AWAY

_BASE_SESSION_CLASS = class_from_module(settings.BASE_SESSION_CLASS)


[docs]class WebSocketClient(WebSocketServerProtocol, _BASE_SESSION_CLASS): """ Implements the server-side of the Websocket connection. """ # nonce value, used to prevent the webclient from erasing the # webclient_authenticated_uid value of csession on disconnect nonce = 0
[docs] def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.protocol_key = "webclient/websocket" self.browserstr = ""
[docs] def get_client_session(self): """ Get the Client browser session (used for auto-login based on browser session) Returns: csession (ClientSession): This is a django-specific internal representation of the browser session. """ try: # client will connect with wsurl?csessid&page_id&browserid webarg = self.http_request_uri.split("?", 1)[1] except IndexError: # this may happen for custom webclients not caring for the # browser session. self.csessid = None return None except AttributeError: from evennia.utils import logger self.csessid = None logger.log_trace(str(self)) return None self.csessid, *cargs = webarg.split("&", 2) if len(cargs) == 1: self.browserstr = str(cargs[0]) elif len(cargs) == 2: self.page_id = str(cargs[0]) self.browserstr = str(cargs[1]) if self.csessid: return _CLIENT_SESSIONS(session_key=self.csessid)
[docs] def onOpen(self): """ This is called when the WebSocket connection is fully established. """ client_address = self.transport.client client_address = client_address[0] if client_address else None if client_address in _UPSTREAM_IPS and "x-forwarded-for" in self.http_headers: addresses = [x.strip() for x in self.http_headers["x-forwarded-for"].split(",")] addresses.reverse() for addr in addresses: if addr not in _UPSTREAM_IPS: client_address = addr break self.init_session("websocket", client_address, self.factory.sessionhandler) csession = self.get_client_session() # this sets self.csessid csessid = self.csessid uid = csession and csession.get("webclient_authenticated_uid", None) nonce = csession and csession.get("webclient_authenticated_nonce", 0) if uid: # the client session is already logged in. self.uid = uid self.nonce = nonce self.logged_in = True for old_session in self.sessionhandler.sessions_from_csessid(csessid): if ( hasattr(old_session, "websocket_close_code") and old_session.websocket_close_code != CLOSE_NORMAL ): # if we have old sessions with the same csession, they are remnants self.sessid = old_session.sessid self.sessionhandler.disconnect(old_session) browserstr = f":{self.browserstr}" if self.browserstr else "" self.protocol_flags["CLIENTNAME"] = f"Evennia Webclient (websocket{browserstr})" self.protocol_flags["UTF-8"] = True self.protocol_flags["OOB"] = True self.protocol_flags["TRUECOLOR"] = True self.protocol_flags["XTERM256"] = True self.protocol_flags["ANSI"] = True # watch for dead links self.transport.setTcpKeepAlive(1) # actually do the connection self.sessionhandler.connect(self)
[docs] def disconnect(self, reason=None): """ Generic hook for the engine to call in order to disconnect this protocol. Args: reason (str or None): Motivation for the disconnection. """ csession = self.get_client_session() if csession: # if the nonce is different, webclient_authenticated_uid has been # set *before* this disconnect (disconnect called after a new client # connects, which occurs in some 'fast' browsers like Google Chrome # and Mobile Safari) if csession.get("webclient_authenticated_nonce", 0) == self.nonce: csession["webclient_authenticated_uid"] = None csession["webclient_authenticated_nonce"] = 0 csession.save() self.logged_in = False self.sessionhandler.disconnect(self) # autobahn-python: # 1000 for a normal close, 1001 if the browser window is closed, # 3000-4999 for app. specific, # in case anyone wants to expose this functionality later. # # sendClose() under autobahn/websocket/interfaces.py self.sendClose(CLOSE_NORMAL, reason)
[docs] def onClose(self, wasClean, code=None, reason=None): """ This is executed when the connection is lost for whatever reason. it can also be called directly, from the disconnect method. Args: wasClean (bool): ``True`` if the WebSocket was closed cleanly. code (int or None): Close status as sent by the WebSocket peer. reason (str or None): Close reason as sent by the WebSocket peer. """ if code == CLOSE_NORMAL or code == GOING_AWAY: self.disconnect(reason) else: self.websocket_close_code = code
[docs] def onMessage(self, payload, isBinary): """ Callback fired when a complete WebSocket message was received. Args: payload (bytes): The WebSocket message received. isBinary (bool): Flag indicating whether payload is binary or UTF-8 encoded text. """ cmdarray = json.loads(str(payload, "utf-8")) if cmdarray: self.data_in(**{cmdarray[0]: [cmdarray[1], cmdarray[2]]})
[docs] def sendLine(self, line): """ Send data to client. Args: line (str): Text to send. """ try: return self.sendMessage(line.encode()) except Disconnected: # this can happen on an unclean close of certain browsers. # it means this link is actually already closed. self.disconnect(reason="Browser already closed.")
[docs] def at_login(self): csession = self.get_client_session() if csession: csession["webclient_authenticated_uid"] = self.uid csession.save()
[docs] def data_in(self, **kwargs): """ Data User > Evennia. Args: text (str): Incoming text. kwargs (any): Options from protocol. Notes: At initilization, the client will send the special 'csessid' command to identify its browser session hash with the Evennia side. The websocket client will also pass 'websocket_close' command to report that the client has been closed and that the session should be disconnected. Both those commands are parsed and extracted already at this point. """ if "websocket_close" in kwargs: self.disconnect() return self.sessionhandler.data_in(self, **kwargs)
[docs] def send_text(self, *args, **kwargs): """ Send text data. This will pre-process the text for color-replacement, conversion to html etc. Args: text (str): Text to send. Keyword Args: options (dict): Options-dict with the following keys understood: - raw (bool): No parsing at all (leave ansi-to-html markers unparsed). - nocolor (bool): Clean out all color. - screenreader (bool): Use Screenreader mode. - send_prompt (bool): Send a prompt with parsed html """ if args: args = list(args) text = args[0] if text is None: return else: return flags = self.protocol_flags options = kwargs.pop("options", {}) raw = options.get("raw", flags.get("RAW", False)) client_raw = options.get("client_raw", False) nocolor = options.get("nocolor", flags.get("NOCOLOR", False)) screenreader = options.get("screenreader", flags.get("SCREENREADER", False)) prompt = options.get("send_prompt", False) if screenreader: # screenreader mode cleans up output text = parse_ansi(text, strip_ansi=True, xterm256=False, mxp=False) text = _RE_SCREENREADER_REGEX.sub("", text) cmd = "prompt" if prompt else "text" if raw: if client_raw: args[0] = text else: args[0] = html.escape(text) # escape html! else: args[0] = parse_html(text, strip_ansi=nocolor) # send to client on required form [cmdname, args, kwargs] self.sendLine(json.dumps([cmd, args, kwargs]))
[docs] def send_prompt(self, *args, **kwargs): kwargs["options"].update({"send_prompt": True}) self.send_text(*args, **kwargs)
[docs] def send_default(self, cmdname, *args, **kwargs): """ Data Evennia -> User. Args: cmdname (str): The first argument will always be the oob cmd name. *args (any): Remaining args will be arguments for `cmd`. Keyword Args: options (dict): These are ignored for oob commands. Use command arguments (which can hold dicts) to send instructions to the client instead. """ if not cmdname == "options": self.sendLine(json.dumps([cmdname, args, kwargs]))