在ApplicationSession的注册端点内访问RPC调用者的IP和HTTP连接标头

Question

我在使用带有asyncio的Python 3.4的Autobahn 0.9.2。

问题：使用WAMP，是否可以从RPC端点内部访问作为调用方的IP和HTTP连接标头的对等方？建立连接时此信息是否持续？如果不是，我将如何开始扩展一些工厂来支持这个工厂？

我的目标很简单：我想拥有一个RPC端点来对本地连接的对等体（呼叫者）的IP进行地理定位，并将增强的数据转发给Redis。我已阅读源代码并知道信息通过的位置（autobahn.websocket.protocol.WebSocketServerProtocol - > onConnect（request）），但是无法从onJoin回调中定义的ApplicationSession的RPC端点深入到它。我尝试遍历传输/路由器/路由器会话链，并没有设法到达那里。我对来自初始连接请求的Peer IP和HTTP头感兴趣。

这里的蒸馏成分：

class IncomingComponent(ApplicationSession): 

def __init__(self, **params): 
    super().__init__() 
    self.redis = StrictRedis(host=config["redis"]["host"], port=config["redis"]["port"], db=config["redis"]["databases"]["ailytics"]) 

def onConnect(self): 
    self.join("abc") 

@asyncio.coroutine 
def onJoin(self, details): 

    def geolocalize_and_store_event(event, detail): 
     # Geolocalize here! Have access to caller ID through detail 
     self.redis.rpush("abc:events", json.dumps(event)) 

    yield from self.register(
     geolocalize_and_store_event, 
     "abc.geolocalize_and_store_event", 
     options=RegisterOptions(details_arg='detail', discloseCaller = True) 
    ) 

和服务器的初始化：

router_factory = wamp.RouterFactory() 

    session_factory = wamp.RouterSessionFactory(router_factory) 
    session_factory.add(IncomingComponent()) 

    transport_factory = websocket.WampWebSocketServerFactory(session_factory, debug=False, debug_wamp=False) 

    loop = asyncio.get_event_loop() 
    coro = loop.create_server(transport_factory, '0.0.0.0', 7788) 
    server = loop.run_until_complete(coro) 

    try: 
     loop.run_forever() 
    except KeyboardInterrupt: 
     pass 
    finally: 
     server.close() 
     loop.close() 

Answer 1

你至少可以通过wamp.session.get元API访问额外的会话/传输信息crossbar.io：

@inlineCallbacks 
def onJoin(self, ign): 

    @inlineCallbacks 
    def method(details): 
     session = yield self.call('wamp.session.get', details.caller) 
     peer = session['transport']['peer'] 
     print "peer's address", peer 

     headers = session['transport']['http_headers_received'] 
     print "headers:" 
     print '\n'.join(['{}: {}'.format(k, v) for (k, v) in headers.items()]) 

    yield self.register(
     method, 'some_method', 
     types.RegisterOptions(details_arg='details'), 
    ) 

Answer 2

的功能现在还没有按规定@oberstet，但得益于高速公路/ WAMP使用工厂模式的东西，我能够在不改变库代码的情况下提出解决方案。

有3个组成部分的子类：

首先，我们指定一个IP地址的实例变量添加到wamp.RouterSession的子类

class IncomingServerSession(wamp.RouterSession): 

    def __init__(self, routerFactory): 
     super().__init__(routerFactory) 
     self.ipAddress = None 

然后我们做wamp.RouterSessionFactory子类中使用IncomingServerSession

class IncomingServerSessionFactory(wamp.RouterSessionFactory): 
    session = IncomingServerSession 

最后，我们继承websocket.WampWebSocketServerProtocol并设置ipAddress实例变量。由于我们处于onOpen回调中，因此我们可以访问对等和HTTP标头。我的服务器是反向代理的，所以我正在寻找通过对等方的自定义HTTP头。

class IncomingServerProtocol(websocket.WampWebSocketServerProtocol): 
    def onOpen(self): 
     try: 
      self._session = self.factory._factory() 

      # Use your own header or just the peer if not reverse-proxied 
      self._session.ipAddress = (self.http_headers.get('x-real-ip') or self.peer) 

      self._session.onOpen(self) 

     except Exception as e: 
      if self.factory.debug_wamp: 
       traceback.print_exc() 
       # # Exceptions raised in onOpen are fatal .. 
       reason = "WAMP Internal Error ({})".format(e) 
       self._bailout(protocol.WebSocketProtocol.CLOSE_STATUS_CODE_INTERNAL_ERROR, reason=reason) 

下面是我们如何获取对端的IP RPC调用内：

@asyncio.coroutine 
def onJoin(self, details): 
    def event(e, details): 
     caller_session_id = details.caller 
     caller_session = self._transport._router._dealer._session_id_to_session[caller_session_id] 

     print(caller_session.ipAddress) 

    #discloseCaller needs to be True 
    yield from self.register(event, "abc.event", options=RegisterOptions(details_arg='details', discloseCaller=True)) 

最后，我们需要更新我们的初始化代码使用我们的子类：

router_factory = wamp.RouterFactory() 

session_factory = IncomingServerSessionFactory(router_factory) 
session_factory.add(IncomingComponent()) 

transport_factory = websocket.WampWebSocketServerFactory(session_factory, debug=False, debug_wamp=False) 
transport_factory.protocol = IncomingServerProtocol 

loop = asyncio.get_event_loop() 
coro = loop.create_server(transport_factory, '0.0.0.0', 7788) 
server = loop.run_until_complete(coro) 

try: 
    loop.run_forever() 
except KeyboardInterrupt: 
    pass 
finally: 
    server.close() 
    loop.close() 

这就是你如何去做，直到有官方支持！