# This file is part of rhizi, a collaborative knowledge graph editor. # Copyright (C) 2014-2015 Rhizi # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published # by the Free Software Foundation, either version 3 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 Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . from collections import namedtuple from flask import request from functools import wraps from geventwebsocket.handler import WebSocketHandler import inspect import logging from socketio import socketio_manage from socketio.server import SocketIOHandler from socketio.server import SocketIOServer from model.graph import Topo_Diff from rz_api_rest import Req_Context from rz_api_websocket import WebSocket_Graph_NS from rz_req_handling import make_response__http__empty, \ HTTP_STATUS__500_INTERNAL_SERVER_ERROR, make_response__json, \ sock_addr_from_env_HTTP_headers, sock_addr_from_REMOTE_X_keys log = logging.getLogger('rhizi') class WS_Req_Env(object): def __init__(self): self.kernel = None class RZ_WebSocket_Server(SocketIOServer): """ Rhizi customized SocketIOServer: - allow response header injection on websocket connections """ class Req_Probe__sock_addr__proxy(object): def probe_client_socket_addr__ws_conn(self, ws_environ): ret = sock_addr_from_REMOTE_X_keys(ws_environ) # # relying on the presence of the 'X-Forwarded-For' is preferable, but # a bit flaky as it is not always present - see #496 # # TODO: evaluate proxy server's behavior on this # try: _, __ = sock_addr_from_env_HTTP_headers(ws_environ, key_name__addr='X-Forwarded-For') except Exception as e: log.warning('ws: client socket addr probe: %s, peer-addr ~: %s:%s' % (e.message, ret[0], ret[1])) return ret class Req_Probe__sock_addr__direct(object): def probe_client_socket_addr__ws_conn(self, ws_environ): return sock_addr_from_REMOTE_X_keys(ws_environ) class RZ_SocketIOHandler(SocketIOHandler): def __init__(self, config, *args, **kwargs): SocketIOHandler.__init__(self, config, *args, **kwargs) def start_response(self, status, headers, exc_info=None): # headers['Access-Control-Allow-Origin'] = '*' return SocketIOHandler.start_response(self, status, headers, exc_info) def handle_disconnect_request(self): SocketIOHandler.handle_disconnect_request(self) class RZ_WebSocketHandler(WebSocketHandler): def __init__(self, socket, address, server, rfile=None): WebSocketHandler.__init__(self, socket, address, server, rfile=rfile) def upgrade_connection(self): return WebSocketHandler.upgrade_connection(self) def upgrade_websocket(self): return WebSocketHandler.upgrade_websocket(self) def __init__ (self, rz_config, wsgi_app): self.wsgi_app = wsgi_app sock_addr = (rz_config.listen_address, rz_config.listen_port) SocketIOServer.__init__(self, sock_addr, wsgi_app, close_timeout=60, policy_server=False, heartbeat_interval=20, # should be less than the heartbeat_timeout heartbeat_timeout=40, handler_class=RZ_WebSocket_Server.RZ_SocketIOHandler, resource='socket.io', # URL prefix for socket.io requests ws_handler_class=RZ_WebSocket_Server.RZ_WebSocketHandler) self.req_probe__sock_addr = None def init_ws_interface(cfg, kernel, flask_webapp): """ Initialize websocket interface: - apply websocket route handlers @return: an initialized RZ_WebSocket_Server object """ def decorator__ws_multicast(ws_srv, f, f_multicast): """ Emit multicast on topo_diff, attr_diff @param f: [!] wrapped function, name used to derive socket message name """ def _prep_for_serialization(obj): # handle special serialization cases if isinstance(obj, Topo_Diff): return obj.to_json_dict() return obj @wraps(f) def wrapped_function(*args, **kw): f_ret = f(*args, **kw) assert type(f_ret) in [list, tuple] # expect (X_Diff, X_Diff.Commit_Result_Type) # # Identify f caller # # TODO: avoid stack inspection if possible # try: stack = inspect.stack() stack_frame = stack[1][0] obj_instance = stack_frame.f_locals.get('self') if None != obj_instance: # [!] no need emit broadcast if call originated from a websocket as # WebSocket_Graph_NS#on_diff_commit__xxx emit their own self-excluding multicast # we still asser call class == WebSocket_Graph_NS caller_class = obj_instance.__class__ assert WebSocket_Graph_NS == caller_class, 'decorator__ws_multicast: unknown callpath: not from WebSocket_Graph_NS' return f_ret except Exception as e: log.exception('decorator__ws_multicast: failed to detect REST/Websocket call via stack inspection') # exception derived from stack return # f caller unidentified, abort cast rzdoc = rzdoc_from_f_args_extractor(args) pkt_data = map(_prep_for_serialization, f_ret) msg_name = f.__name__ pkt = dict(type="event", name=msg_name, args=pkt_data, endpoint='/graph') ws_broadcast_to_rzdoc_readers(ws_srv, pkt, rzdoc) return f_ret return wrapped_function def rzdoc_from_f_args_extractor(f_args): # extract rzdoc from req ctx if len(f_args) < 2 or not isinstance(f_args[1], Req_Context): return req_ctx = f_args[1] return req_ctx.rzdoc def socketio_route_handler(url_path): # FIXME: rm if unused if None == request.environ.get('socketio'): # attempt ws upgrade process for non socketio clients header__upgrade = request.headers.get('Upgrade') if 'websocket' == header__upgrade: resp = make_response__http__empty(101) # 'switching protocols' HTTP status code resp.headers['Upgrade'] = 'websocket' resp.headers['Connection'] = 'Upgrade' rmt_addr, rmt_port = request.req_probe__sock_addr.probe_client_socket_addr__ws_conn(request.environ) log.debug('ws: \'Upgrade: websocket\' header detected, serving \'101\': remote-socket-addr: %s:%s' % (rmt_addr, rmt_port)) return resp else: raise Exception('ws: failed to obtain socketio object from WSGI environment') # init websocket-env ws_req_env = WS_Req_Env() ws_req_env.kernel = kernel ws_req_env.peer_sock_addr = ws_srv.req_probe__sock_addr.probe_client_socket_addr__ws_conn(request.environ) try: socketio_manage(request.environ, {'/graph': WebSocket_Graph_NS}, ws_req_env) # connect socketio manager except: log.exception("ws: exception while handling connection", exc_info=True) return make_response__json(status=HTTP_STATUS__500_INTERNAL_SERVER_ERROR) def ws_broadcast_to_rzdoc_readers(ws_srv, pkt, rzdoc): """ Cast update messege to subscribed readers """ r_assoc_set = kernel.rzdoc__reader_set_from_rzdoc(rzdoc) log.debug('ws: rzdoc cast: msg: \'%s\': rzdoc: %s, cast-size ~= %d' % (pkt.get('name'), rzdoc.name, len(r_assoc_set))) for r_assoc in r_assoc_set: try: r_assoc.socket.send_packet(pkt) except Exception as e: r_assoc.err_count__IO += 1 # connect socketio route route_dec = flask_webapp.route('/socket.io/') f = route_dec(socketio_route_handler) flask_webapp.f = f # init ws server ws_srv = RZ_WebSocket_Server(cfg, flask_webapp) if cfg.reverse_proxy_host is not None: # proxy mode ws_srv.req_probe__sock_addr = RZ_WebSocket_Server.Req_Probe__sock_addr__proxy() else: ws_srv.req_probe__sock_addr = RZ_WebSocket_Server.Req_Probe__sock_addr__direct() kernel.diff_commit__topo = decorator__ws_multicast(ws_srv, kernel.diff_commit__topo, f_multicast=WebSocket_Graph_NS.on_diff_commit__topo) kernel.diff_commit__attr = decorator__ws_multicast(ws_srv, kernel.diff_commit__attr, f_multicast=WebSocket_Graph_NS.on_diff_commit__attr) return ws_srv;