Package xmpp :: Module session
[hide private]
[frames] | no frames]

Source Code for Module xmpp.session

  1  ## 
  2  ##   XMPP server 
  3  ## 
  4  ##   Copyright (C) 2004 Alexey "Snake" Nezhdanov 
  5  ## 
  6  ##   This program is free software; you can redistribute it and/or modify 
  7  ##   it under the terms of the GNU General Public License as published by 
  8  ##   the Free Software Foundation; either version 2, or (at your option) 
  9  ##   any later version. 
 10  ## 
 11  ##   This program is distributed in the hope that it will be useful, 
 12  ##   but WITHOUT ANY WARRANTY; without even the implied warranty of 
 13  ##   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 
 14  ##   GNU General Public License for more details. 
 15   
 16  __version__="$Id" 
 17   
 18  """ 
 19  When your handler is called it is getting the session instance as the first argument. 
 20  This is the difference from xmpppy 0.1 where you got the "Client" instance. 
 21  With Session class you can have "multi-session" client instead of having 
 22  one client for each connection. Is is specifically important when you are 
 23  writing the server. 
 24  """ 
 25   
 26  from protocol import * 
 27   
 28  # Transport-level flags 
 29  SOCKET_UNCONNECTED  =0 
 30  SOCKET_ALIVE        =1 
 31  SOCKET_DEAD         =2 
 32  # XML-level flags 
 33  STREAM__NOT_OPENED =1 
 34  STREAM__OPENED     =2 
 35  STREAM__CLOSING    =3 
 36  STREAM__CLOSED     =4 
 37  # XMPP-session flags 
 38  SESSION_NOT_AUTHED =1 
 39  SESSION_AUTHED     =2 
 40  SESSION_BOUND      =3 
 41  SESSION_OPENED     =4 
 42  SESSION_CLOSED     =5 
 43   
44 -class Session:
45 """ 46 The Session class instance is used for storing all session-related info like 47 credentials, socket/xml stream/session state flags, roster items (in case of 48 client type connection) etc. 49 Session object have no means of discovering is any info is ready to be read. 50 Instead you should use poll() (recomended) or select() methods for this purpose. 51 Session can be one of two types: 'server' and 'client'. 'server' session handles 52 inbound connection and 'client' one used to create an outbound one. 53 Session instance have multitude of internal attributes. The most imporant is the 'peer' one. 54 It is set once the peer is authenticated (client). 55 """
56 - def __init__(self,socket,owner,xmlns=None,peer=None):
57 """ When the session is created it's type (client/server) is determined from the beginning. 58 socket argument is the pre-created socket-like object. 59 It must have the following methods: send, recv, fileno, close. 60 owner is the 'master' instance that have Dispatcher plugged into it and generally 61 will take care about all session events. 62 xmlns is the stream namespace that will be used. Client must set this argument 63 If server sets this argument than stream will be dropped if opened with some another namespace. 64 peer is the name of peer instance. This is the flag that differentiates client session from 65 server session. Client must set it to the name of the server that will be connected, server must 66 leave this argument alone. 67 """ 68 self.xmlns=xmlns 69 if peer: 70 self.TYP='client' 71 self.peer=peer 72 self._socket_state=SOCKET_UNCONNECTED 73 else: 74 self.TYP='server' 75 self.peer=None 76 self._socket_state=SOCKET_ALIVE 77 self._sock=socket 78 self._send=socket.send 79 self._recv=socket.recv 80 self.fileno=socket.fileno 81 self._registered=0 82 83 self.Dispatcher=owner.Dispatcher 84 self.DBG_LINE='session' 85 self.DEBUG=owner.Dispatcher.DEBUG 86 self._expected={} 87 self._owner=owner 88 if self.TYP=='server': self.ID=`random.random()`[2:] 89 else: self.ID=None 90 91 self.sendbuffer='' 92 self._stream_pos_queued=None 93 self._stream_pos_sent=0 94 self.deliver_key_queue=[] 95 self.deliver_queue_map={} 96 self.stanza_queue=[] 97 98 self._session_state=SESSION_NOT_AUTHED 99 self.waiting_features=[] 100 for feature in [NS_TLS,NS_SASL,NS_BIND,NS_SESSION]: 101 if feature in owner.features: self.waiting_features.append(feature) 102 self.features=[] 103 self.feature_in_process=None 104 self.slave_session=None 105 self.StartStream()
106
107 - def StartStream(self):
108 """ This method is used to initialise the internal xml expat parser 109 and to send initial stream header (in case of client connection). 110 Should be used after initial connection and after every stream restart.""" 111 self._stream_state=STREAM__NOT_OPENED 112 self.Stream=simplexml.NodeBuilder() 113 self.Stream._dispatch_depth=2 114 self.Stream.dispatch=self._dispatch 115 self.Parse=self.Stream.Parse 116 self.Stream.stream_footer_received=self._stream_close 117 if self.TYP=='client': 118 self.Stream.stream_header_received=self._catch_stream_id 119 self._stream_open() 120 else: 121 self.Stream.stream_header_received=self._stream_open
122
123 - def receive(self):
124 """ Reads all pending incoming data. 125 Raises IOError on disconnection. 126 Blocks until at least one byte is read.""" 127 try: received = self._recv(10240) 128 except: received = '' 129 130 if len(received): # length of 0 means disconnect 131 self.DEBUG(`self.fileno()`+' '+received,'got') 132 else: 133 self.DEBUG('Socket error while receiving data','error') 134 self.set_socket_state(SOCKET_DEAD) 135 raise IOError("Peer disconnected") 136 return received
137
138 - def sendnow(self,chunk):
139 """ Put chunk into "immidiatedly send" queue. 140 Should only be used for auth/TLS stuff and like. 141 If you just want to shedule regular stanza for delivery use enqueue method. 142 """ 143 if isinstance(chunk,Node): chunk = chunk.__str__().encode('utf-8') 144 elif type(chunk)==type(u''): chunk = chunk.encode('utf-8') 145 self.enqueue(chunk)
146
147 - def enqueue(self,stanza):
148 """ Takes Protocol instance as argument. 149 Puts stanza into "send" fifo queue. Items into the send queue are hold until 150 stream authenticated. After that this method is effectively the same as "sendnow" method.""" 151 if isinstance(stanza,Protocol): 152 self.stanza_queue.append(stanza) 153 else: self.sendbuffer+=stanza 154 if self._socket_state>=SOCKET_ALIVE: self.push_queue()
155
156 - def push_queue(self,failreason=ERR_RECIPIENT_UNAVAILABLE):
157 """ If stream is authenticated than move items from "send" queue to "immidiatedly send" queue. 158 Else if the stream is failed then return all queued stanzas with error passed as argument. 159 Otherwise do nothing.""" 160 # If the stream authed - convert stanza_queue into sendbuffer and set the checkpoints 161 162 if self._stream_state>=STREAM__CLOSED or self._socket_state>=SOCKET_DEAD: # the stream failed. Return all stanzas that are still waiting for delivery. 163 self._owner.deactivatesession(self) 164 for key in self.deliver_key_queue: # Not sure. May be I 165 self._dispatch(Error(self.deliver_queue_map[key],failreason),trusted=1) # should simply re-dispatch it? 166 for stanza in self.stanza_queue: # But such action can invoke 167 self._dispatch(Error(stanza,failreason),trusted=1) # Infinite loops in case of S2S connection... 168 self.deliver_queue_map,self.deliver_key_queue,self.stanza_queue={},[],[] 169 return 170 elif self._session_state>=SESSION_AUTHED: # FIXME! äÏÌÖÅÎ ÂÙÔØ ËÁËÏÊ-ÔÏ ÄÒÕÇÏÊ ÆÌÁÇ. 171 #### LOCK_QUEUE 172 for stanza in self.stanza_queue: 173 txt=stanza.__str__().encode('utf-8') 174 self.sendbuffer+=txt 175 self._stream_pos_queued+=len(txt) # should be re-evaluated for SSL connection. 176 self.deliver_queue_map[self._stream_pos_queued]=stanza # position of the stream when stanza will be successfully and fully sent 177 self.deliver_key_queue.append(self._stream_pos_queued) 178 self.stanza_queue=[]
179 #### UNLOCK_QUEUE 180
181 - def flush_queue(self):
182 """ Put the "immidiatedly send" queue content on the wire. Blocks until at least one byte sent.""" 183 if self.sendbuffer: 184 try: 185 # LOCK_QUEUE 186 sent=self._send(self.sendbuffer) # âÌÏËÉÒÕÀÝÁÑ ÛÔÕÞËÁ! 187 except: 188 # UNLOCK_QUEUE 189 self.set_socket_state(SOCKET_DEAD) 190 self.DEBUG("Socket error while sending data",'error') 191 return self.terminate_stream() 192 self.DEBUG(`self.fileno()`+' '+self.sendbuffer[:sent],'sent') 193 self._stream_pos_sent+=sent 194 self.sendbuffer=self.sendbuffer[sent:] 195 self._stream_pos_delivered=self._stream_pos_sent # Should be acquired from socket somehow. Take SSL into account. 196 while self.deliver_key_queue and self._stream_pos_delivered>self.deliver_key_queue[0]: 197 del self.deliver_queue_map[self.deliver_key_queue[0]] 198 self.deliver_key_queue.remove(self.deliver_key_queue[0])
199 # UNLOCK_QUEUE 200
201 - def _dispatch(self,stanza,trusted=0):
202 """ This is callback that is used to pass the received stanza forth to owner's dispatcher 203 _if_ the stream is authorised. Otherwise the stanza is just dropped. 204 The 'trusted' argument is used to emulate stanza receive. 205 This method is used internally. 206 """ 207 self._owner.packets+=1 208 print self._owner.packets 209 if self._stream_state==STREAM__OPENED or trusted: # if the server really should reject all stanzas after he is closed stream (himeself)? 210 self.DEBUG(stanza.__str__(),'dispatch') 211 stanza.trusted=trusted 212 return self.Dispatcher.dispatch(stanza,self)
213
214 - def _catch_stream_id(self,ns=None,tag='stream',attrs={}):
215 """ This callback is used to detect the stream namespace of incoming stream. Used internally. """ 216 if not attrs.has_key('id') or not attrs['id']: 217 return self.terminate_stream(STREAM_INVALID_XML) 218 self.ID=attrs['id'] 219 if not attrs.has_key('version'): self._owner.Dialback(self)
220
221 - def _stream_open(self,ns=None,tag='stream',attrs={}):
222 """ This callback is used to handle opening stream tag of the incoming stream. 223 In the case of client session it just make some validation. 224 Server session also sends server headers and if the stream valid the features node. 225 Used internally. """ 226 text='<?xml version="1.0" encoding="utf-8"?>\n<stream:stream' 227 if self.TYP=='client': 228 text+=' to="%s"'%self.peer 229 else: 230 text+=' id="%s"'%self.ID 231 if not attrs.has_key('to'): text+=' from="%s"'%self._owner.servernames[0] 232 else: text+=' from="%s"'%attrs['to'] 233 if attrs.has_key('xml:lang'): text+=' xml:lang="%s"'%attrs['xml:lang'] 234 if self.xmlns: xmlns=self.xmlns 235 else: xmlns=NS_SERVER 236 text+=' xmlns:db="%s" xmlns:stream="%s" xmlns="%s"'%(NS_DIALBACK,NS_STREAMS,xmlns) 237 if attrs.has_key('version') or self.TYP=='client': text+=' version="1.0"' 238 self.sendnow(text+'>') 239 self.set_stream_state(STREAM__OPENED) 240 if self.TYP=='client': return 241 if tag<>'stream': return self.terminate_stream(STREAM_INVALID_XML) 242 if ns<>NS_STREAMS: return self.terminate_stream(STREAM_INVALID_NAMESPACE) 243 if self.Stream.xmlns<>self.xmlns: return self.terminate_stream(STREAM_BAD_NAMESPACE_PREFIX) 244 if not attrs.has_key('to'): return self.terminate_stream(STREAM_IMPROPER_ADDRESSING) 245 if attrs['to'] not in self._owner.servernames: return self.terminate_stream(STREAM_HOST_UNKNOWN) 246 self.ourname=attrs['to'].lower() 247 if self.TYP=='server' and attrs.has_key('version'): 248 # send features 249 features=Node('stream:features') 250 if NS_TLS in self.waiting_features: 251 features.NT.starttls.setNamespace(NS_TLS) 252 features.T.starttls.NT.required 253 if NS_SASL in self.waiting_features: 254 features.NT.mechanisms.setNamespace(NS_SASL) 255 for mec in self._owner.SASL.mechanisms: 256 features.T.mechanisms.NT.mechanism=mec 257 else: 258 if NS_BIND in self.waiting_features: features.NT.bind.setNamespace(NS_BIND) 259 if NS_SESSION in self.waiting_features: features.NT.session.setNamespace(NS_SESSION) 260 self.sendnow(features)
261
262 - def feature(self,feature):
263 """ Declare some stream feature as activated one. """ 264 if feature not in self.features: self.features.append(feature) 265 self.unfeature(feature)
266
267 - def unfeature(self,feature):
268 """ Declare some feature as illegal. Illegal features can not be used. 269 Example: BIND feature becomes illegal after Non-SASL auth. """ 270 if feature in self.waiting_features: self.waiting_features.remove(feature)
271
272 - def _stream_close(self,unregister=1):
273 """ Write the closing stream tag and destroy the underlaying socket. Used internally. """ 274 if self._stream_state>=STREAM__CLOSED: return 275 self.set_stream_state(STREAM__CLOSING) 276 self.sendnow('</stream:stream>') 277 self.set_stream_state(STREAM__CLOSED) 278 self.push_queue() # decompose queue really since STREAM__CLOSED 279 self._owner.flush_queues() 280 if unregister: self._owner.unregistersession(self) 281 self._destroy_socket()
282
283 - def terminate_stream(self,error=None,unregister=1):
284 """ Notify the peer about stream closure. 285 Ensure that xmlstream is not brokes - i.e. if the stream isn't opened yet - 286 open it before closure. 287 If the error condition is specified than create a stream error and send it along with 288 closing stream tag. 289 Emulate receiving 'unavailable' type presence just before stream closure. 290 """ 291 if self._stream_state>=STREAM__CLOSING: return 292 if self._stream_state<STREAM__OPENED: 293 self.set_stream_state(STREAM__CLOSING) 294 self._stream_open() 295 else: 296 self.set_stream_state(STREAM__CLOSING) 297 p=Presence(typ='unavailable') 298 p.setNamespace(NS_CLIENT) 299 self._dispatch(p,trusted=1) 300 if error: 301 if isinstance(error,Node): self.sendnow(error) 302 else: self.sendnow(ErrorNode(error)) 303 self._stream_close(unregister=unregister) 304 if self.slave_session: 305 self.slave_session.terminate_stream(STREAM_REMOTE_CONNECTION_FAILED)
306
307 - def _destroy_socket(self):
308 """ Break cyclic dependancies to let python's GC free memory right now.""" 309 self.Stream.dispatch=None 310 self.Stream.stream_footer_received=None 311 self.Stream.stream_header_received=None 312 self.Stream.destroy() 313 self._sock.close() 314 self.set_socket_state(SOCKET_DEAD)
315
316 - def start_feature(self,f):
317 """ Declare some feature as "negotiating now" to prevent other features from start negotiating. """ 318 if self.feature_in_process: raise "Starting feature %s over %s !"%(f,self.feature_in_process) 319 self.feature_in_process=f
320
321 - def stop_feature(self,f):
322 """ Declare some feature as "negotiated" to allow other features start negotiating. """ 323 if self.feature_in_process<>f: raise "Stopping feature %s instead of %s !"%(f,self.feature_in_process) 324 self.feature_in_process=None
325
326 - def set_socket_state(self,newstate):
327 """ Change the underlaying socket state. 328 Socket starts with SOCKET_UNCONNECTED state 329 and then proceeds (possibly) to SOCKET_ALIVE 330 and then to SOCKET_DEAD """ 331 if self._socket_state<newstate: self._socket_state=newstate
332
333 - def set_session_state(self,newstate):
334 """ Change the session state. 335 Session starts with SESSION_NOT_AUTHED state 336 and then comes through 337 SESSION_AUTHED, SESSION_BOUND, SESSION_OPENED and SESSION_CLOSED states. 338 """ 339 if self._session_state<newstate: 340 if self._session_state<SESSION_AUTHED and \ 341 newstate>=SESSION_AUTHED: self._stream_pos_queued=self._stream_pos_sent 342 self._session_state=newstate
343
344 - def set_stream_state(self,newstate):
345 """ Change the underlaying XML stream state 346 Stream starts with STREAM__NOT_OPENED and then proceeds with 347 STREAM__OPENED, STREAM__CLOSING and STREAM__CLOSED states. 348 Note that some features (like TLS and SASL) 349 requires stream re-start so this state can have non-linear changes. """ 350 if self._stream_state<newstate: self._stream_state=newstate
351