Accessing IP address headers and RPC HTTP connections within a registered ApplicationSession endpoint
I am using Autobahn 0.9.2 with Python 3.4 with asyncio.
Questions: Using WAMP, is it possible to access a peer acting as a caller's IP and HTTP connection header from an RPC endpoint? Is this information preserved across connection establishment? If not, how would I start expanding some factories to support this?
My goal is pretty simple: I want to have an RPC endpoint for the Geolocalize IP of the connected peer (Caller) and pass the extended data to Redis. I've read the source and know where the information is going (autobahn.websocket.protocol.WebSocketServerProtocol -> onConnect (request)), but I'm having trouble drilling it from the ApplicationSession RPC endpoint defined in the onJoin callback. I tried to traverse the transport / router / router session chain and couldn't get there. I am interested in both the Peer IP and the HTTP headers from the initial connection request.
Here's the distilled component:
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)
)
And server initialization:
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()
source to share
You can access additional session / transport information using the meta API wamp.session.get
in crossbar.io at a minimum:
@inlineCallbacks
def onJoin(self, ign):
@inlineCallbacks
def method(details):
session = yield self.call('wamp.session.get', details.caller)
peer = session['transport']['peer']
print "peer 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'),
)
source to share
There is no functionality yet as @oberstet pointed out, but thanks to the autobahn / WAMP stuff using the factory pattern, I was able to find a solution without changing the library code.
The subclass includes 3 components:
First, we add the ipAddress instance variable to the wamp.RouterSession subclass
class IncomingServerSession(wamp.RouterSession):
def __init__(self, routerFactory):
super().__init__(routerFactory)
self.ipAddress = None
Then we subclass wamp.RouterSessionFactory using IncomingServerSession
class IncomingServerSessionFactory(wamp.RouterSessionFactory):
session = IncomingServerSession
Finally, we subclass websocket.WampWebSocketServerProtocol and set the instance variable ipAddress. Since we are in the onOpen callback, we have access to the peer and HTTP headers. My server is reverse proxy, so I am looking for a custom HTTP header above the peer.
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)
Here we access the peer-to-peer IP address inside the RPC call:
@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))
Finally, we need to update our initialization code to use our subclasses:
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()
How do you do it until official support arrives!
source to share