~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2007 Canonical Ltd

#

# 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 of the License, 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.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

"""

FTP test server.

Based on medusa: http://www.amk.ca/python/code/medusa.html

"""

import asyncore

import errno

import os

import select

import stat

import threading

import medusa

import medusa.filesys

import medusa.ftp_server

from bzrlib import (

    tests,

    trace,

    transport,

    )

class test_filesystem(medusa.filesys.os_filesystem):

    """A custom filesystem wrapper to add missing functionalities."""

    def chmod(self, path, mode):

        p = self.normalize(self.path_module.join (self.wd, path))

        return os.chmod(self.translate(p), mode)

class test_authorizer(object):

    """A custom Authorizer object for running the test suite.

    The reason we cannot use dummy_authorizer, is because it sets the

    channel to readonly, which we don't always want to do.

    """

    def __init__(self, root):

        self.root = root

        # If secured_user is set secured_password will be checked

        self.secured_user = None

        self.secured_password = None

    def authorize(self, channel, username, password):

        """Return (success, reply_string, filesystem)"""

        channel.persona = -1, -1

        if username == 'anonymous':

            channel.read_only = 1

        else:

            channel.read_only = 0

        # Check secured_user if set

        if (self.secured_user is not None

            and username == self.secured_user

            and password != self.secured_password):

            return 0, 'Password invalid.', None

        else:

            return 1, 'OK.', test_filesystem(self.root)

class ftp_channel(medusa.ftp_server.ftp_channel):

    """Customized ftp channel"""

    def log(self, message):

        """Redirect logging requests."""

        trace.mutter('ftp_channel: %s', message)

    def log_info(self, message, type='info'):

        """Redirect logging requests."""

        trace.mutter('ftp_channel %s: %s', type, message)

    def cmd_rnfr(self, line):

        """Prepare for renaming a file."""

        self._renaming = line[1]

        self.respond('350 Ready for RNTO')

        # TODO: jam 20060516 in testing, the ftp server seems to

        #       check that the file already exists, or it sends

        #       550 RNFR command failed

    def cmd_rnto(self, line):

        """Rename a file based on the target given.

        rnto must be called after calling rnfr.

        """

        if not self._renaming:

            self.respond('503 RNFR required first.')

        pfrom = self.filesystem.translate(self._renaming)

        self._renaming = None

        pto = self.filesystem.translate(line[1])

        if os.path.exists(pto):

            self.respond('550 RNTO failed: file exists')

            return

        try:

            os.rename(pfrom, pto)

        except (IOError, OSError), e:

            # TODO: jam 20060516 return custom responses based on

            #       why the command failed

            # (bialix 20070418) str(e) on Python 2.5 @ Windows

            # sometimes don't provide expected error message;

            # so we obtain such message via os.strerror()

            self.respond('550 RNTO failed: %s' % os.strerror(e.errno))

        except:

            self.respond('550 RNTO failed')

            # For a test server, we will go ahead and just die

            raise

        else:

            self.respond('250 Rename successful.')

    def cmd_size(self, line):

        """Return the size of a file

        This is overloaded to help the test suite determine if the

        target is a directory.

        """

        filename = line[1]

        if not self.filesystem.isfile(filename):

            if self.filesystem.isdir(filename):

                self.respond('550 "%s" is a directory' % (filename,))

            else:

                self.respond('550 "%s" is not a file' % (filename,))

        else:

            self.respond('213 %d'

                % (self.filesystem.stat(filename)[stat.ST_SIZE]),)

    def cmd_mkd(self, line):

        """Create a directory.

        Overloaded because default implementation does not distinguish

        *why* it cannot make a directory.

        """

        if len (line) != 2:

            self.command_not_understood(''.join(line))

        else:

            path = line[1]

            try:

                self.filesystem.mkdir (path)

                self.respond ('257 MKD command successful.')

            except (IOError, OSError), e:

                # (bialix 20070418) str(e) on Python 2.5 @ Windows

                # sometimes don't provide expected error message;

                # so we obtain such message via os.strerror()

                self.respond ('550 error creating directory: %s' %

                              os.strerror(e.errno))

            except:

                self.respond ('550 error creating directory.')

    def cmd_site(self, line):

        """Site specific commands."""

        command, args = line[1].split(' ', 1)

        if command.lower() == 'chmod':

            try:

                mode, path = args.split()

                mode = int(mode, 8)

            except ValueError:

                # We catch both malformed line and malformed mode with the same

                # ValueError.

                self.command_not_understood(' '.join(line))

                return

            try:

                # Yes path and mode are reversed

                self.filesystem.chmod(path, mode)

                self.respond('200 SITE CHMOD command successful')

            except AttributeError:

                # The chmod method is not available in read-only and will raise

                # AttributeError since a different filesystem is used in that

                # case

                self.command_not_authorized(' '.join(line))

        else:

            # Another site specific command was requested. We don't know that

            # one

            self.command_not_understood(' '.join(line))

class ftp_server(medusa.ftp_server.ftp_server):

    """Customize the behavior of the Medusa ftp_server.

    There are a few warts on the ftp_server, based on how it expects

    to be used.

    """

    _renaming = None

    ftp_channel_class = ftp_channel

    def __init__(self, *args, **kwargs):

        trace.mutter('Initializing ftp_server: %r, %r', args, kwargs)

        medusa.ftp_server.ftp_server.__init__(self, *args, **kwargs)

    def log(self, message):

        """Redirect logging requests."""

        trace.mutter('ftp_server: %s', message)

    def log_info(self, message, type='info'):

        """Override the asyncore.log_info so we don't stipple the screen."""

        trace.mutter('ftp_server %s: %s', type, message)

class FTPServer(transport.Server):

    """Common code for FTP server facilities."""

    def __init__(self):

        self._root = None

        self._ftp_server = None

        self._port = None

        self._async_thread = None

        # ftp server logs

        self.logs = []

    def get_url(self):

        """Calculate an ftp url to this server."""

        return 'ftp://foo:bar@localhost:%d/' % (self._port)

    def get_bogus_url(self):

        """Return a URL which cannot be connected to."""

        return 'ftp://127.0.0.1:1'

    def log(self, message):

        """This is used by medusa.ftp_server to log connections, etc."""

        self.logs.append(message)

    def setUp(self, vfs_server=None):

        from bzrlib.transport.local import LocalURLServer

        if not (vfs_server is None or isinstance(vfs_server, LocalURLServer)):

            raise AssertionError(

                "FTPServer currently assumes local transport, got %s" % vfs_server)

        self._root = os.getcwdu()

        self._ftp_server = ftp_server(

            authorizer=test_authorizer(root=self._root),

            ip='localhost',

            port=0, # bind to a random port

            resolver=None,

            logger_object=self # Use FTPServer.log() for messages

            )

        self._port = self._ftp_server.getsockname()[1]

        # Don't let it loop forever, or handle an infinite number of requests.

        # In this case it will run for 1000s, or 10000 requests

        self._async_thread = threading.Thread(

                target=FTPServer._asyncore_loop_ignore_EBADF,

                kwargs={'timeout':0.1, 'count':10000})

        self._async_thread.setDaemon(True)

        self._async_thread.start()

    def tearDown(self):

        """See bzrlib.transport.Server.tearDown."""

        self._ftp_server.close()

        asyncore.close_all()

        self._async_thread.join()

    @staticmethod

    def _asyncore_loop_ignore_EBADF(*args, **kwargs):

        """Ignore EBADF during server shutdown.

        We close the socket to get the server to shutdown, but this causes

        select.select() to raise EBADF.

        """

        try:

            asyncore.loop(*args, **kwargs)

            # FIXME: If we reach that point, we should raise an exception

            # explaining that the 'count' parameter in setUp is too low or

            # testers may wonder why their test just sits there waiting for a

            # server that is already dead. Note that if the tester waits too

            # long under pdb the server will also die.

        except select.error, e:

            if e.args[0] != errno.EBADF:

                raise