code.delx.au - pymsnt/blob - src/ft.py

   1 # Copyright 2005-2006 James Bunton <james@delx.cjb.net>
   2 # Licensed for distribution under the GPL version 2, check COPYING for details
   3
   4 from throttle import Throttler
   5 from twisted.internet import protocol
   6 from twisted.words.xish.domish import Element
   7
   8 import disco
   9 import lang
  10 from debug import LogEvent, INFO, WARN, ERROR
  11 import config
  12 import utils
  13
  14 import random
  15 import sys
  16
  17
  18 def doRateLimit(setConsumer, consumer):
  19         try:
  20                 rateLimit = int(config.ftRateLimit)
  21         except ValueError:
  22                 rateLimit = 0
  23         if rateLimit > 0:
  24                 throttler = Throttler(consumer, rateLimit)
  25                 setConsumer(throttler)
  26         else:
  27                 setConsumer(consumer)
  28
  29 def checkSizeOk(size):
  30         try:
  31                 size = int(size)
  32                 limit = int(config.ftSizeLimit)
  33         except ValueError:
  34                 return False
  35         if limit == 0:
  36                 return True
  37         return limit > size
  38
  39 ###########
  40 # Sending #
  41 ###########
  42
  43 class FTSend:
  44         """ For file transfers going from Jabber to MSN. """
  45         def __init__(self, session, to, startTransfer, cancelTransfer, filename, filesize):
  46                 self.startTransfer = startTransfer
  47                 self.cancelTransfer = cancelTransfer
  48                 self.filename = filename
  49                 self.filesize = filesize
  50                 if not checkSizeOk(self.filesize):
  51                         LogEvent(INFO, session.jabberID, "File too large.")
  52                         text = lang.get(session.lang).msnFtSizeRejected % (self.filename, config.ftSizeLimit, config.website)
  53                         session.legacycon.sendMessage(to, "", text, True)
  54                         session.sendMessage(to=session.jabberID, fro=to, body=text)
  55                         self.reject()
  56                         return
  57
  58                 session.legacycon.sendFile(to, self)
  59
  60         def accept(self, legacyFileSend):
  61                 doRateLimit(self.startTransfer, legacyFileSend)
  62                 self.cleanup()
  63
  64         def reject(self):
  65                 self.cancelTransfer()
  66                 self.cleanup()
  67
  68         def cleanup(self):
  69                 del self.startTransfer, self.cancelTransfer
  70
  71
  72 try:
  73         from twisted.web import http
  74 except ImportError:
  75         try:
  76                 from twisted.protocols import http
  77         except ImportError:
  78                 print "Couldn't find http.HTTPClient. If you're using Twisted 2.0, make sure that you've installed twisted.web"
  79                 raise
  80
  81
  82 class OOBHeaderHelper(http.HTTPClient):
  83         """ Makes a HEAD request and grabs the length """
  84         def connectionMade(self):
  85                 self.sendCommand("HEAD", self.factory.path.encode("utf-8"))
  86                 self.sendHeader("Host", (self.factory.host + ":" + str(self.factory.port)).encode("utf-8"))
  87                 self.endHeaders()
  88
  89         def handleEndHeaders(self):
  90                 self.factory.gotLength(self.length)
  91
  92         def handleResponse(self, data):
  93                 pass
  94
  95
  96 class OOBSendConnector(http.HTTPClient):
  97         def connectionMade(self):
  98                 self.sendCommand("GET", self.factory.path.encode("utf-8"))
  99                 self.sendHeader("Host", (self.factory.host + ":" + str(self.factory.port)).encode("utf-8"))
 100                 self.endHeaders()
 101                 self.first = True
 102
 103         def handleResponsePart(self, data):
 104                 self.factory.consumer.write(data)
 105
 106         def handleResponseEnd(self):
 107                 # This is called once before writing is finished, and once when the
 108                 # connection closes. We only consumer.close() on the second.
 109                 if self.first:
 110                         self.first = False
 111                 else:
 112                         self.factory.consumer.close()
 113                         self.factory.consumer = None
 114                         self.factory.finished()
 115
 116
 117
 118
 119
 120 #############
 121 # Receiving #
 122 #############
 123
 124 class FTReceive:
 125         """ For file transfers going from MSN to Jabber. """
 126
 127         """
 128         Plan of action for this class:
 129         * Determine the FT support of the Jabber client.
 130         * If we find a common protocol, then send the invitation.
 131         * Tell the legacyftp object the result of the invitation.
 132         * If it was accepted, then start the transfer.
 133
 134         """
 135
 136         def __init__(self, session, senderJID, legacyftp):
 137                 if not checkSizeOk(legacyftp.filesize):
 138                         LogEvent(INFO, session.jabberID, "File too large.")
 139                         legacyftp.reject()
 140                         text = lang.get(session.lang).msnFtSizeRejected % (legacyftp.filename, config.ftSizeLimit, config.website)
 141                         session.legacycon.sendMessage(senderJID, "", text, False)
 142                         session.sendMessage(to=session.jabberID, fro=senderJID, body=text)
 143                         return
 144                 self.session = session
 145                 self.toJID = self.session.jabberID + "/" + self.session.highestResource()
 146                 self.senderJID = senderJID
 147                 self.ident = (self.toJID, self.senderJID)
 148                 self.legacyftp = legacyftp
 149                 LogEvent(INFO, session.jabberID)
 150                 self.checkSupport()
 151
 152         def checkSupport(self):
 153                 def discoDone(features):
 154                         LogEvent(INFO, self.ident)
 155                         enabledS5B = hasattr(self.session.pytrans, "ftSOCKS5Receive")
 156                         enabledOOB = hasattr(self.session.pytrans, "ftOOBReceive")
 157                         hasFT  = features.count(disco.FT)
 158                         hasS5B = features.count(disco.S5B)
 159                         hasOOB = features.count(disco.IQOOB)
 160                         LogEvent(INFO, self.ident, "Choosing transfer mode.")
 161                         if hasFT > 0 and hasS5B > 0 and enabledS5B:
 162                                 self.socksMode()
 163                         elif hasOOB > 0 and enabledOOB:
 164                                 self.oobMode()
 165                         elif enabledOOB:
 166                                 self.messageOobMode()
 167                         else:
 168                                 # No support
 169                                 self.legacyftp.reject()
 170                                 del self.legacyftp
 171
 172                 def discoFail(err=None):
 173                         LogEvent(INFO, self.ident, str(err))
 174                         if hasattr(self.session.pytrans, "ftOOBReceive"):
 175                                 self.messageOobMode()
 176                         else:
 177                                 # No support
 178                                 self.legacyftp.reject()
 179                                 del self.legacyftp
 180
 181                 d = disco.DiscoRequest(self.session.pytrans, self.toJID).doDisco()
 182                 d.addCallbacks(discoDone, discoFail)
 183
 184         def socksMode(self):
 185                 def ftReply(el):
 186                         if el.getAttribute("type") != "result":
 187                                 ftDeclined()
 188                                 return
 189                         self.session.pytrans.ftSOCKS5Receive.addConnection(utils.socks5Hash(self.sid, self.senderJID, self.toJID), self.legacyftp)
 190                         LogEvent(INFO, self.ident)
 191                         iq = Element((None, "iq"))
 192                         iq.attributes["type"] = "set"
 193                         iq.attributes["to"] = self.toJID
 194                         iq.attributes["from"] = self.senderJID
 195                         query = iq.addElement("query")
 196                         query.attributes["xmlns"] = disco.S5B
 197                         query.attributes["sid"] = self.sid
 198                         query.attributes["mode"] = "tcp"
 199                         streamhost = query.addElement("streamhost")
 200                         streamhost.attributes["jid"] = self.senderJID
 201                         streamhost.attributes["host"] = config.host
 202                         streamhost.attributes["port"] = config.ftJabberPort
 203                         d = self.session.pytrans.discovery.sendIq(iq)
 204                         d.addErrback(ftDeclined) # Timeout
 205
 206                 def ftDeclined(el):
 207                         self.legacyftp.reject()
 208                         del self.legacyftp
 209
 210                 LogEvent(INFO, self.ident)
 211                 self.sid = str(random.randint(1000, sys.maxint))
 212                 iq = Element((None, "iq"))
 213                 iq.attributes["type"] = "set"
 214                 iq.attributes["to"] = self.toJID
 215                 iq.attributes["from"] = self.senderJID
 216                 si = iq.addElement("si")
 217                 si.attributes["xmlns"] = disco.SI
 218                 si.attributes["profile"] = disco.FT
 219                 si.attributes["id"] = self.sid
 220                 file = si.addElement("file")
 221                 file.attributes["xmlns"] = disco.FT
 222                 file.attributes["size"] = str(self.legacyftp.filesize)
 223                 file.attributes["name"] = self.legacyftp.filename
 224                 # Feature negotiation
 225                 feature = si.addElement("feature")
 226                 feature.attributes["xmlns"] = disco.FEATURE_NEG
 227                 x = feature.addElement("x")
 228                 x.attributes["xmlns"] = disco.XDATA
 229                 x.attributes["type"] = "form"
 230                 field = x.addElement("field")
 231                 field.attributes["type"] = "list-single"
 232                 field.attributes["var"] = "stream-method"
 233                 option = field.addElement("option")
 234                 value = option.addElement("value")
 235                 value.addContent(disco.S5B)
 236                 d = self.session.pytrans.discovery.sendIq(iq, 60*3)
 237                 d.addCallback(ftReply)
 238                 d.addErrback(ftDeclined)
 239
 240         def oobMode(self):
 241                 def cb(el):
 242                         if el.getAttribute("type") != "result":
 243                                 self.legacyftp.reject()
 244                         del self.legacyftp
 245                         self.session.pytrans.ftOOBReceive.remFile(filename)
 246
 247                 def ecb(ignored=None):
 248                         self.legacyftp.reject()
 249                         del self.legacyftp
 250
 251                 LogEvent(INFO, self.ident)
 252                 filename = self.session.pytrans.ftOOBReceive.putFile(self, self.legacyftp.filename)
 253                 iq = Element((None, "iq"))
 254                 iq.attributes["to"] = self.toJID
 255                 iq.attributes["from"] = self.senderJID
 256                 query = m.addElement("query")
 257                 query.attributes["xmlns"] = disco.IQOOB
 258                 query.addElement("url").addContent(config.ftOOBRoot + "/" + filename)
 259                 d = self.session.send(iq)
 260                 d.addCallbacks(cb, ecb)
 261
 262         def messageOobMode(self):
 263                 LogEvent(INFO, self.ident)
 264                 filename = self.session.pytrans.ftOOBReceive.putFile(self, self.legacyftp.filename)
 265                 m = Element((None, "message"))
 266                 m.attributes["to"] = self.session.jabberID
 267                 m.attributes["from"] = self.senderJID
 268                 m.addElement("body").addContent(config.ftOOBRoot + "/" + filename)
 269                 x = m.addElement("x")
 270                 x.attributes["xmlns"] = disco.XOOB
 271                 x.addElement("url").addContent(config.ftOOBRoot + "/" + filename)
 272                 self.session.pytrans.send(m)
 273
 274         def error(self, ignored=None):
 275                 # FIXME
 276                 LogEvent(WARN)
 277
 278
 279
 280 # SOCKS5
 281
 282 import socks5
 283 import struct
 284
 285 class JEP65ConnectionSend(protocol.Protocol):
 286 # TODO, clean up and move this to socks5
 287         STATE_INITIAL = 1
 288         STATE_WAIT_AUTHOK = 2
 289         STATE_WAIT_CONNECTOK = 3
 290         STATE_READY = 4
 291
 292         def __init__(self):
 293                 self.state = self.STATE_INITIAL
 294                 self.buf = ""
 295
 296         def connectionMade(self):
 297                 self.transport.write(struct.pack("!BBB", 5, 1, 0))
 298                 self.state = self.STATE_WAIT_AUTHOK
 299
 300         def connectionLost(self, reason):
 301                 if self.state == self.STATE_READY:
 302                         self.factory.consumer.close()
 303                 else:
 304                         self.factory.consumer.error()
 305
 306         def _waitAuthOk(self):
 307                 ver, method = struct.unpack("!BB", self.buf[:2])
 308                 if ver != 5 or method != 0:
 309                         self.transport.loseConnection()
 310                         return
 311                 self.buf = self.buf[2:] # chop
 312
 313                 # Send CONNECT request
 314                 length = len(self.factory.hash)
 315                 self.transport.write(struct.pack("!BBBBB", 5, 1, 0, 3, length))
 316                 self.transport.write("".join([struct.pack("!B" , ord(x))[0] for x in self.factory.hash]))
 317                 self.transport.write(struct.pack("!H", 0))
 318                 self.state = self.STATE_WAIT_CONNECTOK
 319
 320         def _waitConnectOk(self):
 321                 ver, rep, rsv, atyp = struct.unpack("!BBBB", self.buf[:4])
 322                 if not (ver == 5 and rep == 0):
 323                         self.transport.loseConnection()
 324                         return
 325
 326                 self.state = self.STATE_READY
 327                 self.factory.madeConnection(self.transport.addr[0])
 328
 329         def dataReceived(self, buf):
 330                 if self.state == self.STATE_READY:
 331                         self.factory.consumer.write(buf)
 332
 333                 self.buf += buf
 334                 if self.state == self.STATE_WAIT_AUTHOK:
 335                         self._waitAuthOk()
 336                 elif self.state == self.STATE_WAIT_CONNECTOK:
 337                         self._waitConnectOk()
 338
 339
 340 class JEP65ConnectionReceive(socks5.SOCKSv5):
 341         def __init__(self, listener):
 342                 socks5.SOCKSv5.__init__(self)
 343                 self.listener = listener
 344                 self.supportedAuthMechs = [socks5.AUTHMECH_ANON]
 345                 self.supportedAddrs = [socks5.ADDR_DOMAINNAME]
 346                 self.enabledCommands = [socks5.CMD_CONNECT]
 347                 self.addr = ""
 348
 349         def connectRequested(self, addr, port):
 350                 # So that the legacyftp can close the connection
 351                 self.transport.close = self.transport.loseConnection
 352
 353                 # Check for special connect to the namespace -- this signifies that
 354                 # the client is just checking that it can connect to the streamhost
 355                 if addr == disco.S5B:
 356                         self.connectCompleted(addr, 0)
 357                         self.transport.loseConnection()
 358                         return
 359
 360                 self.addr = addr
 361
 362                 if self.listener.isActive(addr):
 363                         self.sendErrorReply(socks5.REPLY_CONN_NOT_ALLOWED)
 364                         return
 365
 366                 if self.listener.addConnection(addr, self):
 367                         self.connectCompleted(addr, 0)
 368                 else:
 369                         self.sendErrorReply(socks5.REPLY_CONN_REFUSED)
 370
 371         def connectionLost(self, reason):
 372                 if self.state == socks5.STATE_CONNECT_PENDING:
 373                         self.listener.removePendingConnection(self.addr, self)
 374                 else:
 375                         self.transport.unregisterProducer()
 376                         if self.peersock != None:
 377                                 self.peersock.peersock = None
 378                                 self.peersock.transport.unregisterProducer()
 379                                 self.peersock = None
 380                                 self.listener.removeActiveConnection(self.addr)
 381
 382 class Proxy65(protocol.Factory):
 383         def __init__(self, port):
 384                 LogEvent(INFO)
 385                 reactor.listenTCP(port, self)
 386                 self.pendingConns = {}
 387                 self.activeConns = {}
 388
 389         def buildProtocol(self, addr):
 390                 return JEP65ConnectionReceive(self)
 391
 392         def isActive(self, address):
 393                 return address in self.activeConns
 394
 395         def activateStream(self, address):
 396                 if address in self.pendingConns:
 397                         olist = self.pendingConns[address]
 398                         if len(olist) != 2:
 399                                 LogEvent(WARN, '', "Not exactly two!")
 400                                 return
 401
 402                         assert address not in self.activeConns
 403                         self.activeConns[address] = None
 404
 405                         if not isinstance(olist[0], JEP65ConnectionReceive):
 406                                 legacyftp = olist[0]
 407                                 connection = olist[1]
 408                         elif not isinstance(olist[1], JEP65ConnectionReceive):
 409                                 legacyftp = olist[1]
 410                                 connection = olist[0]
 411                         else:
 412                                 LogEvent(WARN, '', "No JEP65Connection")
 413                                 return
 414
 415                         doRateLimit(legacyftp.accept, connection.transport)
 416                 else:
 417                         LogEvent(WARN, '', "No pending connection.")
 418
 419         def addConnection(self, address, connection):
 420                 olist = self.pendingConns.get(address, [])
 421                 if len(olist) <= 1:
 422                         olist.append(connection)
 423                         self.pendingConns[address] = olist
 424                         if len(olist) == 2:
 425                                 self.activateStream(address)
 426                         return True
 427                 else:
 428                         return False
 429
 430         def removePendingConnection(self, address, connection):
 431                 olist = self.pendingConns[address]
 432                 if len(olist) == 1:
 433                         del self.pendingConns[address]
 434                 else:
 435                         olist.remove(connection)
 436
 437         def removeActiveConnection(self, address):
 438                 del self.activeConns[address]
 439
 440
 441 # OOB download server
 442
 443 from twisted.web import server, resource, error
 444 from twisted.internet import reactor
 445
 446 from debug import LogEvent, INFO, WARN, ERROR
 447
 448 class OOBReceiveConnector:
 449         def __init__(self, ftReceive, ftHttpPush):
 450                 self.ftReceive, self.ftHttpPush = ftReceive, ftHttpPush
 451                 doRateLimit(self.ftReceive.legacyftp.accept, self)
 452
 453         def write(self, data):
 454                 self.ftHttpPush.write(data)
 455
 456         def close(self):
 457                 self.ftHttpPush.finish()
 458
 459         def error(self):
 460                 self.ftHttpPush.finish()
 461                 self.ftReceive.error()
 462
 463 class FileTransferOOBReceive(resource.Resource):
 464         def __init__(self, port):
 465                 LogEvent(INFO)
 466                 self.isLeaf = True
 467                 self.files = {}
 468                 self.oobSite = server.Site(self)
 469                 reactor.listenTCP(port, self.oobSite)
 470
 471         def putFile(self, file, filename):
 472                 path = str(random.randint(100000000, 999999999))
 473                 filename = (path + "/" + filename).replace("//", "/")
 474                 self.files[filename] = file
 475                 return filename
 476
 477         def remFile(self, filename):
 478                 if self.files.has_key(filename):
 479                         del self.files[filename]
 480
 481         def render_GET(self, request):
 482                 filename = request.path[1:] # Remove the leading /
 483                 if self.files.has_key(filename):
 484                         file = self.files[filename]
 485                         request.setHeader("Content-Length", str(file.legacyftp.filesize))
 486                         request.setHeader("Content-Disposition", "attachment; filename=\"%s\"" % file.legacyftp.filename.encode("utf-8"))
 487                         OOBReceiveConnector(file, request)
 488                         del self.files[filename]
 489                         return server.NOT_DONE_YET
 490                 else:
 491                         page = error.NoResource(message="404 File Not Found")
 492                         return page.render(request)
 493
 494         def render_HEAD(self, request):
 495                 filename = request.path[1:] # Remove the leading /
 496                 if self.files.has_key(filename):
 497                         file = self.files[filename]
 498                         request.setHeader("Content-Length", str(file.legacyftp.filesize))
 499                         request.setHeader("Content-Disposition", "attachment; filename=\"%s\"" % file.legacyftp.filename.encode("utf-8"))
 500                         return ""
 501                 else:
 502                         page = error.NoResource(message="404 File Not Found")
 503                         return page.render(request)
 504
 505