34
34
from bzrlib.tests import test_server
37
# Convert the pyftplib string version into a tuple to avoid traps in string
39
pyftplib_version = tuple(map(int, ftpserver.__ver__.split('.')))
37
42
class AnonymousWithWriteAccessAuthorizer(ftpserver.DummyAuthorizer):
39
44
def _check_permissions(self, username, perm):
92
97
ftpserver.FTPHandler.ftp_NLST(self, path)
94
def ftp_SITE_CHMOD(self, line):
96
mode, path = line.split(None, 1)
99
# We catch both malformed line and malformed mode with the same
101
self.respond("500 'SITE CHMOD %s': command not understood."
103
self.log('FAIL SITE CHMOD ' % line)
105
ftp_path = self.fs.fs2ftp(path)
107
self.run_as_current_user(self.fs.chmod, self.fs.ftp2fs(path), mode)
109
why = ftpserver._strerror(err)
110
self.log('FAIL SITE CHMOD 0%03o "%s". %s.' % (mode, ftp_path, why))
111
self.respond('550 %s.' % why)
113
self.log('OK SITE CHMOD 0%03o "%s".' % (mode, ftp_path))
114
self.respond('200 SITE CHMOD succesful.')
117
# pyftpdlib says to define SITE commands by declaring ftp_SITE_<CMD> methods,
118
# but fails to recognize them.
119
ftpserver.proto_cmds['SITE CHMOD'] = ftpserver._CommandProperty(
120
perm='w', # Best fit choice even if not exactly right (can be d, f or m too)
121
auth_needed=True, arg_needed=True, check_path=False,
122
help='Syntax: SITE CHMOD <SP> octal_mode_bits file-name (chmod file)',
124
# An empty password is valid, hence the arg is neither mandatory not forbidden
125
ftpserver.proto_cmds['PASS'].arg_needed = None
99
def log_cmd(self, cmd, arg, respcode, respstr):
100
# base class version choke on unicode, the alternative is to just
101
# provide an empty implementation and relies on the client to do
102
# the logging for debugging purposes. Not worth the trouble so far
104
if cmd in ("DELE", "RMD", "RNFR", "RNTO", "MKD"):
105
line = '"%s" %s' % (' '.join([cmd, unicode(arg)]).strip(), respcode)
109
# An empty password is valid, hence the arg is neither mandatory nor forbidden
110
ftpserver.proto_cmds['PASS']['arg'] = None
128
112
class ftp_server(ftpserver.FTPServer):
169
153
address = ('localhost', 0) # bind to a random port
170
154
authorizer = AnonymousWithWriteAccessAuthorizer()
171
authorizer.add_anonymous(self._root, perm='elradfmw')
155
authorizer.add_anonymous(self._root, perm='elradfmwM')
172
156
self._ftp_server = ftp_server(address, BzrConformingFTPHandler,
174
158
# This is hacky as hell, will not work if we need two servers working
204
188
% (self._ftpd_thread.ident,))
206
190
def _run_server(self):
207
"""Run the server until stop_server is called, shut it down properly then.
191
"""Run the server until stop_server is called.
193
Shut it down properly then.
209
195
self._ftpd_running = True
210
196
self._ftpd_starting.release()