##
## XMPP server
##
## Copyright (C) 2004 Alexey "Snake" Nezhdanov
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2, 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 General Public License for more details.
__version__="$Id"
"""
When your handler is called it is getting the session instance as the first argument.
This is the difference from xmpppy 0.1 where you got the "Client" instance.
With Session class you can have "multi-session" client instead of having
one client for each connection. Is is specifically important when you are
writing the server.
"""
from protocol import *
# Transport-level flags
SOCKET_UNCONNECTED =0
SOCKET_ALIVE =1
SOCKET_DEAD =2
# XML-level flags
STREAM__NOT_OPENED =1
STREAM__OPENED =2
STREAM__CLOSING =3
STREAM__CLOSED =4
# XMPP-session flags
SESSION_NOT_AUTHED =1
SESSION_AUTHED =2
SESSION_BOUND =3
SESSION_OPENED =4
SESSION_CLOSED =5
class Session:
"""
The Session class instance is used for storing all session-related info like
credentials, socket/xml stream/session state flags, roster items (in case of
client type connection) etc.
Session object have no means of discovering is any info is ready to be read.
Instead you should use poll() (recomended) or select() methods for this purpose.
Session can be one of two types: 'server' and 'client'. 'server' session handles
inbound connection and 'client' one used to create an outbound one.
Session instance have multitude of internal attributes. The most imporant is the 'peer' one.
It is set once the peer is authenticated (client).
"""
def __init__(self,socket,owner,xmlns=None,peer=None):
""" When the session is created it's type (client/server) is determined from the beginning.
socket argument is the pre-created socket-like object.
It must have the following methods: send, recv, fileno, close.
owner is the 'master' instance that have Dispatcher plugged into it and generally
will take care about all session events.
xmlns is the stream namespace that will be used. Client must set this argument
If server sets this argument than stream will be dropped if opened with some another namespace.
peer is the name of peer instance. This is the flag that differentiates client session from
server session. Client must set it to the name of the server that will be connected, server must
leave this argument alone.
"""
self.xmlns=xmlns
if peer:
self.TYP='client'
self.peer=peer
self._socket_state=SOCKET_UNCONNECTED
else:
self.TYP='server'
self.peer=None
self._socket_state=SOCKET_ALIVE
self._sock=socket
self._send=socket.send
self._recv=socket.recv
self.fileno=socket.fileno
self._registered=0
self.Dispatcher=owner.Dispatcher
self.DBG_LINE='session'
self.DEBUG=owner.Dispatcher.DEBUG
self._expected={}
self._owner=owner
if self.TYP=='server': self.ID=`random.random()`[2:]
else: self.ID=None
self.sendbuffer=''
self._stream_pos_queued=None
self._stream_pos_sent=0
self.deliver_key_queue=[]
self.deliver_queue_map={}
self.stanza_queue=[]
self._session_state=SESSION_NOT_AUTHED
self.waiting_features=[]
for feature in [NS_TLS,NS_SASL,NS_BIND,NS_SESSION]:
if feature in owner.features: self.waiting_features.append(feature)
self.features=[]
self.feature_in_process=None
self.slave_session=None
self.StartStream()
def StartStream(self):
""" This method is used to initialise the internal xml expat parser
and to send initial stream header (in case of client connection).
Should be used after initial connection and after every stream restart."""
self._stream_state=STREAM__NOT_OPENED
self.Stream=simplexml.NodeBuilder()
self.Stream._dispatch_depth=2
self.Stream.dispatch=self._dispatch
self.Parse=self.Stream.Parse
self.Stream.stream_footer_received=self._stream_close
if self.TYP=='client':
self.Stream.stream_header_received=self._catch_stream_id
self._stream_open()
else:
self.Stream.stream_header_received=self._stream_open
def receive(self):
""" Reads all pending incoming data.
Raises IOError on disconnection.
Blocks until at least one byte is read."""
try: received = self._recv(10240)
except: received = ''
if len(received): # length of 0 means disconnect
self.DEBUG(`self.fileno()`+' '+received,'got')
else:
self.DEBUG('Socket error while receiving data','error')
self.set_socket_state(SOCKET_DEAD)
raise IOError("Peer disconnected")
return received
def sendnow(self,chunk):
""" Put chunk into "immidiatedly send" queue.
Should only be used for auth/TLS stuff and like.
If you just want to shedule regular stanza for delivery use enqueue method.
"""
if isinstance(chunk,Node): chunk = chunk.__str__().encode('utf-8')
elif type(chunk)==type(u''): chunk = chunk.encode('utf-8')
self.enqueue(chunk)
def enqueue(self,stanza):
""" Takes Protocol instance as argument.
Puts stanza into "send" fifo queue. Items into the send queue are hold until
stream authenticated. After that this method is effectively the same as "sendnow" method."""
if isinstance(stanza,Protocol):
self.stanza_queue.append(stanza)
else: self.sendbuffer+=stanza
if self._socket_state>=SOCKET_ALIVE: self.push_queue()
def push_queue(self,failreason=ERR_RECIPIENT_UNAVAILABLE):
""" If stream is authenticated than move items from "send" queue to "immidiatedly send" queue.
Else if the stream is failed then return all queued stanzas with error passed as argument.
Otherwise do nothing."""
# If the stream authed - convert stanza_queue into sendbuffer and set the checkpoints
if self._stream_state>=STREAM__CLOSED or self._socket_state>=SOCKET_DEAD: # the stream failed. Return all stanzas that are still waiting for delivery.
self._owner.deactivatesession(self)
for key in self.deliver_key_queue: # Not sure. May be I
self._dispatch(Error(self.deliver_queue_map[key],failreason),trusted=1) # should simply re-dispatch it?
for stanza in self.stanza_queue: # But such action can invoke
self._dispatch(Error(stanza,failreason),trusted=1) # Infinite loops in case of S2S connection...
self.deliver_queue_map,self.deliver_key_queue,self.stanza_queue={},[],[]
return
elif self._session_state>=SESSION_AUTHED: # FIXME! - .
#### LOCK_QUEUE
for stanza in self.stanza_queue:
txt=stanza.__str__().encode('utf-8')
self.sendbuffer+=txt
self._stream_pos_queued+=len(txt) # should be re-evaluated for SSL connection.
self.deliver_queue_map[self._stream_pos_queued]=stanza # position of the stream when stanza will be successfully and fully sent
self.deliver_key_queue.append(self._stream_pos_queued)
self.stanza_queue=[]
#### UNLOCK_QUEUE
def flush_queue(self):
""" Put the "immidiatedly send" queue content on the wire. Blocks until at least one byte sent."""
if self.sendbuffer:
try:
# LOCK_QUEUE
sent=self._send(self.sendbuffer) # !
except:
# UNLOCK_QUEUE
self.set_socket_state(SOCKET_DEAD)
self.DEBUG("Socket error while sending data",'error')
return self.terminate_stream()
self.DEBUG(`self.fileno()`+' '+self.sendbuffer[:sent],'sent')
self._stream_pos_sent+=sent
self.sendbuffer=self.sendbuffer[sent:]
self._stream_pos_delivered=self._stream_pos_sent # Should be acquired from socket somehow. Take SSL into account.
while self.deliver_key_queue and self._stream_pos_delivered>self.deliver_key_queue[0]:
del self.deliver_queue_map[self.deliver_key_queue[0]]
self.deliver_key_queue.remove(self.deliver_key_queue[0])
# UNLOCK_QUEUE
def _dispatch(self,stanza,trusted=0):
""" This is callback that is used to pass the received stanza forth to owner's dispatcher
_if_ the stream is authorised. Otherwise the stanza is just dropped.
The 'trusted' argument is used to emulate stanza receive.
This method is used internally.
"""
self._owner.packets+=1
if self._stream_state==STREAM__OPENED or trusted: # if the server really should reject all stanzas after he is closed stream (himeself)?
self.DEBUG(stanza.__str__(),'dispatch')
stanza.trusted=trusted
return self.Dispatcher.dispatch(stanza,self)
def _catch_stream_id(self,ns=None,tag='stream',attrs={}):
""" This callback is used to detect the stream namespace of incoming stream. Used internally. """
if not attrs.has_key('id') or not attrs['id']:
return self.terminate_stream(STREAM_INVALID_XML)
self.ID=attrs['id']
if not attrs.has_key('version'): self._owner.Dialback(self)
def _stream_open(self,ns=None,tag='stream',attrs={}):
""" This callback is used to handle opening stream tag of the incoming stream.
In the case of client session it just make some validation.
Server session also sends server headers and if the stream valid the features node.
Used internally. """
text='<?xml version="1.0" encoding="utf-8"?>\n<stream:stream'
if self.TYP=='client':
text+=' to="%s"'%self.peer
else:
text+=' id="%s"'%self.ID
if not attrs.has_key('to'): text+=' from="%s"'%self._owner.servernames[0]
else: text+=' from="%s"'%attrs['to']
if attrs.has_key('xml:lang'): text+=' xml:lang="%s"'%attrs['xml:lang']
if self.xmlns: xmlns=self.xmlns
else: xmlns=NS_SERVER
text+=' xmlns:db="%s" xmlns:stream="%s" xmlns="%s"'%(NS_DIALBACK,NS_STREAMS,xmlns)
if attrs.has_key('version') or self.TYP=='client': text+=' version="1.0"'
self.sendnow(text+'>')
self.set_stream_state(STREAM__OPENED)
if self.TYP=='client': return
if tag<>'stream': return self.terminate_stream(STREAM_INVALID_XML)
if ns<>NS_STREAMS: return self.terminate_stream(STREAM_INVALID_NAMESPACE)
if self.Stream.xmlns<>self.xmlns: return self.terminate_stream(STREAM_BAD_NAMESPACE_PREFIX)
if not attrs.has_key('to'): return self.terminate_stream(STREAM_IMPROPER_ADDRESSING)
if attrs['to'] not in self._owner.servernames: return self.terminate_stream(STREAM_HOST_UNKNOWN)
self.ourname=attrs['to'].lower()
if self.TYP=='server' and attrs.has_key('version'):
# send features
features=Node('stream:features')
if NS_TLS in self.waiting_features:
features.NT.starttls.setNamespace(NS_TLS)
features.T.starttls.NT.required
if NS_SASL in self.waiting_features:
features.NT.mechanisms.setNamespace(NS_SASL)
for mec in self._owner.SASL.mechanisms:
features.T.mechanisms.NT.mechanism=mec
else:
if NS_BIND in self.waiting_features: features.NT.bind.setNamespace(NS_BIND)
if NS_SESSION in self.waiting_features: features.NT.session.setNamespace(NS_SESSION)
self.sendnow(features)
def feature(self,feature):
""" Declare some stream feature as activated one. """
if feature not in self.features: self.features.append(feature)
self.unfeature(feature)
def unfeature(self,feature):
""" Declare some feature as illegal. Illegal features can not be used.
Example: BIND feature becomes illegal after Non-SASL auth. """
if feature in self.waiting_features: self.waiting_features.remove(feature)
def _stream_close(self,unregister=1):
""" Write the closing stream tag and destroy the underlaying socket. Used internally. """
if self._stream_state>=STREAM__CLOSED: return
self.set_stream_state(STREAM__CLOSING)
self.sendnow('</stream:stream>')
self.set_stream_state(STREAM__CLOSED)
self.push_queue() # decompose queue really since STREAM__CLOSED
self._owner.flush_queues()
if unregister: self._owner.unregistersession(self)
self._destroy_socket()
def terminate_stream(self,error=None,unregister=1):
""" Notify the peer about stream closure.
Ensure that xmlstream is not brokes - i.e. if the stream isn't opened yet -
open it before closure.
If the error condition is specified than create a stream error and send it along with
closing stream tag.
Emulate receiving 'unavailable' type presence just before stream closure.
"""
if self._stream_state>=STREAM__CLOSING: return
if self._stream_state<STREAM__OPENED:
self.set_stream_state(STREAM__CLOSING)
self._stream_open()
else:
self.set_stream_state(STREAM__CLOSING)
p=Presence(typ='unavailable')
p.setNamespace(NS_CLIENT)
self._dispatch(p,trusted=1)
if error:
if isinstance(error,Node): self.sendnow(error)
else: self.sendnow(ErrorNode(error))
self._stream_close(unregister=unregister)
if self.slave_session:
self.slave_session.terminate_stream(STREAM_REMOTE_CONNECTION_FAILED)
def _destroy_socket(self):
""" Break cyclic dependancies to let python's GC free memory right now."""
self.Stream.dispatch=None
self.Stream.stream_footer_received=None
self.Stream.stream_header_received=None
self.Stream.destroy()
self._sock.close()
self.set_socket_state(SOCKET_DEAD)
def start_feature(self,f):
""" Declare some feature as "negotiating now" to prevent other features from start negotiating. """
if self.feature_in_process: raise "Starting feature %s over %s !"%(f,self.feature_in_process)
self.feature_in_process=f
def stop_feature(self,f):
""" Declare some feature as "negotiated" to allow other features start negotiating. """
if self.feature_in_process<>f: raise "Stopping feature %s instead of %s !"%(f,self.feature_in_process)
self.feature_in_process=None
def set_socket_state(self,newstate):
""" Change the underlaying socket state.
Socket starts with SOCKET_UNCONNECTED state
and then proceeds (possibly) to SOCKET_ALIVE
and then to SOCKET_DEAD """
if self._socket_state<newstate: self._socket_state=newstate
def set_session_state(self,newstate):
""" Change the session state.
Session starts with SESSION_NOT_AUTHED state
and then comes through
SESSION_AUTHED, SESSION_BOUND, SESSION_OPENED and SESSION_CLOSED states.
"""
if self._session_state<newstate:
if self._session_state<SESSION_AUTHED and \
newstate>=SESSION_AUTHED: self._stream_pos_queued=self._stream_pos_sent
self._session_state=newstate
def set_stream_state(self,newstate):
""" Change the underlaying XML stream state
Stream starts with STREAM__NOT_OPENED and then proceeds with
STREAM__OPENED, STREAM__CLOSING and STREAM__CLOSED states.
Note that some features (like TLS and SASL)
requires stream re-start so this state can have non-linear changes. """
if self._stream_state<newstate: self._stream_state=newstate
|