Merge lp:~cjwatson/txpkgupload/drop-distro-files into lp:~lazr-developers/txpkgupload/trunk

[Colin Watson]

Institute a temporary directory on the same filesystem as the upload root; stop creating obsolete .distro files; remove locking, since we now move directories into place atomically.

Institute a temporary directory on the same filesystem as the upload root; stop creating obsolete .distro files; remove locking, since we now move directories into place atomically.

Before deploying this, we need to make sure that the tmp-incoming directory is created with appropriate permissions, and it wouldn't hurt to land an lp-production-configs change making the new directory explicit.

[William Grant (wgrant)]

review: Approve (code)

=== modified file 'etc/txpkgupload.yaml'
--- etc/txpkgupload.yaml	2015-01-12 13:14:55 +0000
+++ etc/txpkgupload.yaml	2015-01-21 13:10:45 +0000
@@ -41,7 +41,11 @@
 # idle_timeout: 3600
 
 ## Where on the filesystem do uploads live?
-# fsroot: "/var/tmp/txpkgupload"
+# fsroot: "/var/tmp/txpkgupload/incoming"
+
+## Where do we write temporary files during uploads?  This must be on the
+## same filesystem as fsroot.
+# temp_dir: "/var/tmp/txpkgupload/tmp-incoming"
 
 # If true, enable additional debug logging.
 # debug: false

=== removed file 'src/txpkgupload/glock.py'
--- src/txpkgupload/glock.py	2015-01-07 14:20:13 +0000
+++ src/txpkgupload/glock.py	1970-01-01 00:00:00 +0000
@@ -1,316 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: latin1 -*-
-#----------------------------------------------------------------------------
-# glock.py:                 Global mutex
-#
-# See __doc__ string below.
-#
-# Requires:
-#    - Python 1.5.2 or newer (www.python.org)
-#    - On windows: win32 extensions installed
-#           (http://www.python.org/windows/win32all/win32all.exe)
-#    - OS: Unix, Windows.
-#
-# $Id: //depot/rgutils/rgutils/glock.py#5 $
-#----------------------------------------------------------------------------
-'''
-This module defines the class GlobalLock that implements a global
-(inter-process) mutex on Windows and Unix, using file-locking on
-Unix.
-
-@see: class L{GlobalLock} for more details.
-'''
-__version__ = '0.2.' + '$Revision: #5 $'[12:-2]
-__author__ = 'Richard Gruet', '[e-mail]'
-__date__    = '$Date: 2005/06/19 $'[7:-2], '$Author: rgruet $'[9:-2]
-__since__ = '2000-01-22'
-__doc__ += '\n@author: %s (U{%s})\n@version: %s' % (__author__[0],
-                                            __author__[1], __version__)
-__all__ = ['GlobalLock', 'GlobalLockError', 'LockAlreadyAcquired', 'NotOwner']
-
-# Imports:
-import sys, string, os, errno, re
-
-# System-dependent imports for locking implementation:
-_windows = (sys.platform == 'win32')
-
-if _windows:
-    try:
-        import win32event, win32api, pywintypes
-    except ImportError:
-        sys.stderr.write('The win32 extensions need to be installed!')
-    try:
-        import ctypes
-    except ImportError:
-        ctypes = None
-else:   # assume Unix
-    try:
-        import fcntl
-    except ImportError:
-        sys.stderr.write("On what kind of OS am I ? (Mac?) I should be on "
-                         "Unix but can't import fcntl.\n")
-        raise
-    import threading
-
-# Exceptions :
-# ----------
-class GlobalLockError(Exception):
-    ''' Error raised by the glock module.
-    '''
-    pass
-
-class NotOwner(GlobalLockError):
-    ''' Attempt to release somebody else's lock.
-    '''
-    pass
-
-class LockAlreadyAcquired(GlobalLockError):
-    ''' Non-blocking acquire but lock already seized.
-    '''
-    pass
-
-
-# Constants
-# ---------:
-if sys.version[:3] < '2.2':
-    True, False = 1, 0  # built-in in Python 2.2+
-
-#----------------------------------------------------------------------------
-class GlobalLock:
-#----------------------------------------------------------------------------
-    ''' A global mutex.
-
-        B{Specification}
-
-         - The lock must act as a global mutex, ie block between different
-           candidate processus, but ALSO between different candidate
-           threads of the same process.
-
-         - It must NOT block in case of reentrant lock request issued by
-           the SAME thread.
-         - Extraneous unlocks should be ideally harmless.
-
-        B{Implementation}
-
-        In Python there is no portable global lock AFAIK. There is only a
-        LOCAL/ in-process Lock mechanism (threading.RLock), so we have to
-        implement our own solution:
-
-         - Unix: use fcntl.flock(). Recursive calls OK. Different process OK.
-           But <> threads, same process don't block so we have to use an extra
-           threading.RLock to fix that point.
-         - Windows: We use WIN32 mutex from Python Win32 extensions. Can't use
-           std module msvcrt.locking(), because global lock is OK, but
-           blocks also for 2 calls from the same thread!
-    '''
-    RE_ERROR_MSG = re.compile ("^\[Errno ([0-9]+)\]")
-
-    def __init__(self, fpath, lockInitially=False, logger=None):
-        ''' Creates (or opens) a global lock.
-
-            @param fpath: Path of the file used as lock target. This is also
-                          the global id of the lock. The file will be created
-                          if non existent.
-            @param lockInitially: if True locks initially.
-            @param logger: an optional logger object.
-        '''
-        self.logger = logger
-        self.fpath = fpath
-        if os.path.exists(fpath):
-            self.previous_lockfile_present = True
-        else:
-            self.previous_lockfile_present = False
-        if _windows:
-            self.name = string.replace(fpath, '\\', '_')
-            self.mutex = win32event.CreateMutex(None, lockInitially, self.name)
-        else: # Unix
-            self.name = fpath
-            self.flock = open(fpath, 'w')
-            self.fdlock = self.flock.fileno()
-            self.threadLock = threading.RLock()
-        if lockInitially:
-            self.acquire()
-
-    def __del__(self):
-        #print '__del__ called' ##
-        try: self.release()
-        except: pass
-        if _windows:
-            win32api.CloseHandle(self.mutex)
-        else:
-            try: self.flock.close()
-            except: pass
-
-    def __repr__(self):
-        return '<Global lock @ %s>' % self.name
-
-    def acquire(self, blocking=False):
-        """ Locks. Attemps to acquire a lock.
-
-            @param blocking: If True, suspends caller until done. Otherwise,
-            LockAlreadyAcquired is raised if the lock cannot be acquired immediately.
-
-            On windows an IOError is always raised after ~10 sec if the lock
-            can't be acquired.
-            @exception GlobalLockError: if lock can't be acquired (timeout)
-            @exception LockAlreadyAcquired: someone already has the lock and
-                       the caller decided not to block.
-        """
-        if self.logger:
-            self.logger.info('Creating lockfile: %s', self.fpath)
-        if _windows:
-            if blocking:
-                timeout = win32event.INFINITE
-            else:
-                timeout = 0
-            r = win32event.WaitForSingleObject(self.mutex, timeout)
-            if r == win32event.WAIT_FAILED:
-                raise GlobalLockError("Can't acquire mutex: error")
-            if not blocking and r == win32event.WAIT_TIMEOUT:
-                raise LockAlreadyAcquired('Lock %s already acquired by '
-                                          'someone else' % self.name)
-        else:
-            # First, acquire the global (inter-process) lock:
-            if blocking:
-                options = fcntl.LOCK_EX
-            else:
-                options = fcntl.LOCK_EX|fcntl.LOCK_NB
-            try:
-                fcntl.flock(self.fdlock, options)
-            except IOError as message: #(errno 13: perm. denied,
-                            #       36: Resource deadlock avoided)
-                if not blocking and self._errnoOf (message) == errno.EWOULDBLOCK:
-                    raise LockAlreadyAcquired('Lock %s already acquired by '
-                                              'someone else' % self.name)
-                else:
-                    raise GlobalLockError('Cannot acquire lock on "file" '
-                                          '%s: %s\n' % (self.name, message))
-            #print 'got file lock.' ##
-
-            # Then acquire the local (inter-thread) lock:
-            if not self.threadLock.acquire(blocking):
-                fcntl.flock(self.fdlock, fcntl.LOCK_UN) # release global lock
-                raise LockAlreadyAcquired('Lock %s already acquired by '
-                                          'someone else' % self.name)
-            if self.previous_lockfile_present and self.logger:
-                self.logger.warn("Stale lockfile detected and claimed.")
-            #print 'got thread lock.' ##
-
-        self.is_locked = True
-
-    def release(self, skip_delete=False):
-        ''' Unlocks. (caller must own the lock!)
-
-            @param skip_delete: don't try to delete the file. This can
-                be used when the original filename has changed; for
-                instance, if the lockfile is erased out-of-band, or if
-                the directory it contains has been renamed.
-
-            @return: The lock count.
-            @exception IOError: if file lock can't be released
-            @exception NotOwner: Attempt to release somebody else's lock.
-        '''
-        if not self.is_locked:
-            return
-        if not skip_delete:
-            if self.logger:
-                self.logger.debug('Removing lock file: %s', self.fpath)
-            os.unlink(self.fpath)
-        elif self.logger:
-            # At certain times the lockfile will have been removed or
-            # moved away before we call release(); log a message because
-            # this is unusual and could be an error.
-            self.logger.debug('Oops, my lock file disappeared: %s', self.fpath)
-        if _windows:
-            if ctypes:
-                result = ctypes.windll.kernel32.ReleaseMutex(self.mutex.handle)
-                if not result:
-                   raise NotOwner("Attempt to release somebody else's lock")
-            else:
-                try:
-                    win32event.ReleaseMutex(self.mutex)
-                    #print "released mutex"
-                except pywintypes.error as e:
-                    errCode, fctName, errMsg =  e.args
-                    if errCode == 288:
-                        raise NotOwner("Attempt to release somebody else's lock")
-                    else:
-                        raise GlobalLockError('%s: err#%d: %s' % (fctName, errCode,
-                                                                  errMsg))
-        else:
-            # First release the local (inter-thread) lock:
-            try:
-                self.threadLock.release()
-            except AssertionError:
-                raise NotOwner("Attempt to release somebody else's lock")
-
-            # Then release the global (inter-process) lock:
-            try:
-                fcntl.flock(self.fdlock, fcntl.LOCK_UN)
-            except IOError: # (errno 13: permission denied)
-                raise GlobalLockError('Unlock of file "%s" failed\n' %
-                                                            self.name)
-        self.is_locked = False
-
-    def _errnoOf (self, message):
-        match = self.RE_ERROR_MSG.search(str(message))
-        if match:
-            return int(match.group(1))
-        else:
-            raise Exception ('Malformed error message "%s"' % message)
-
-#----------------------------------------------------------------------------
-def test():
-#----------------------------------------------------------------------------
-    ##TODO: a more serious test with distinct processes !
-
-    print 'Testing glock.py...' 
-
-    # unfortunately can't test inter-process lock here!
-    lockName = 'myFirstLock'
-    l = GlobalLock(lockName)
-    if not _windows:
-        assert os.path.exists(lockName)
-    l.acquire()
-    l.acquire() # reentrant lock, must not block
-    l.release()
-    l.release()
-
-    try: l.release()
-    except NotOwner: pass
-    else: raise Exception('should have raised a NotOwner exception')
-
-    # Check that <> threads of same process do block:
-    import threading, time
-    thread = threading.Thread(target=threadMain, args=(l,))
-    print 'main: locking...',
-    l.acquire()
-    print ' done.'
-    thread.start()
-    time.sleep(3)
-    print '\nmain: unlocking...',
-    l.release()
-    print ' done.'
-    time.sleep(0.1)
-
-    print '=> Test of glock.py passed.'
-    return l
-
-def threadMain(lock):
-    print 'thread started(%s).' % lock
-    try: lock.acquire(blocking=False)
-    except LockAlreadyAcquired: pass
-    else: raise Exception('should have raised LockAlreadyAcquired')
-    print 'thread: locking (should stay blocked for ~ 3 sec)...',
-    lock.acquire()
-    print 'thread: locking done.'
-    print 'thread: unlocking...',
-    lock.release()
-    print ' done.'
-    print 'thread ended.'
-
-#----------------------------------------------------------------------------
-#       M A I N
-#----------------------------------------------------------------------------
-if __name__ == "__main__":
-    l = test()

=== modified file 'src/txpkgupload/hooks.py'
--- src/txpkgupload/hooks.py	2015-01-09 15:55:08 +0000
+++ src/txpkgupload/hooks.py	2015-01-21 13:10:45 +0000
@@ -10,14 +10,10 @@
 
 
 import os
-import shutil
-import stat
 import time
 
 from twisted.python import log
 
-from txpkgupload.glock import GlobalLock
-
 
 class InterfaceFailure(Exception):
     pass
@@ -29,11 +25,8 @@
     LOG_MAGIC = "Post-processing finished"
     _targetcount = 0
 
-    def __init__(self, targetpath, allow_user, cmd=None,
-                 targetstart=0, perms=None, prefix=''):
+    def __init__(self, targetpath, targetstart=0, perms=None, prefix=''):
         self.targetpath = targetpath
-        self.cmd = cmd
-        self.allow_user = allow_user
         self.perms = perms
         self.prefix = prefix
 
@@ -53,9 +46,7 @@
         log.msg("Session from %s:%s" % (host, port), debug=True)
 
     def client_done_hook(self, fsroot, host, port):
-        """A client has completed. If it authenticated then it stands a chance
-        of having uploaded a file to the set. If not; then it is simply an
-        aborted transaction and we remove the fsroot."""
+        """A client has completed."""
 
         if fsroot not in self.clients:
             raise InterfaceFailure("Unable to find fsroot in client set")
@@ -63,103 +54,29 @@
         log.msg("Processing session complete in %s" % fsroot, debug=True)
 
         client = self.clients[fsroot]
-        if "distro" not in client:
-            # Login username defines the distribution context of the upload.
-            # So abort unauthenticated sessions by removing its contents
-            shutil.rmtree(fsroot)
-            return
-
-        # Protect from race condition between creating the directory
-        # and creating the distro file, and also in cases where the
-        # temporary directory and the upload directory are not in the
-        # same filesystem (non-atomic "rename").
-        lockfile_path = os.path.join(self.targetpath, ".lock")
-        self.lock = GlobalLock(lockfile_path)
-
-        # XXX cprov 20071024 bug=156795: We try to acquire the lock as soon
-        # as possible after creating the lockfile but are still open to
-        # a race.
-        self.lock.acquire(blocking=True)
-        mode = stat.S_IMODE(os.stat(lockfile_path).st_mode)
-
-        # XXX cprov 20081024 bug=185731: The lockfile permission can only be
-        # changed by its owner. Since we can't predict which process will
-        # create it in production systems we simply ignore errors when trying
-        # to grant the right permission. At least, one of the process will
-        # be able to do so.
-        try:
-            os.chmod(lockfile_path, mode | stat.S_IWGRP)
-        except OSError:
-            pass
-
-        try:
-            timestamp = time.strftime("%Y%m%d-%H%M%S")
-            path = "upload%s-%s-%06d" % (
-                self.prefix, timestamp, self.targetcount)
-            target_fsroot = os.path.join(self.targetpath, path)
-
-            # Create file to store the distro used.
-            log.msg(
-                "Upload was targetted at %s" % client["distro"], debug=True)
-            distro_filename = target_fsroot + ".distro"
-            distro_file = open(distro_filename, "w")
-            distro_file.write(client["distro"])
-            distro_file.close()
-
-            # Move the session directory to the target directory.
-            if os.path.exists(target_fsroot):
-                log.msg("Targeted upload already present: %s" % path)
-                log.msg("System clock skewed ?")
-            else:
-                try:
-                    shutil.move(fsroot, target_fsroot)
-                except (OSError, IOError):
-                    if not os.path.exists(target_fsroot):
-                        raise
-
-            # XXX cprov 20071024: We should replace os.system call by os.chmod
-            # and fix the default permission value accordingly in txpkgupload
-            if self.perms is not None:
-                os.system("chmod %s -R %s" % (self.perms, target_fsroot))
-
-            # Invoke processing script, if provided.
-            if self.cmd:
-                cmd = self.cmd
-                cmd = cmd.replace("@fsroot@", target_fsroot)
-                cmd = cmd.replace("@distro@", client["distro"])
-                log.msg("Running upload handler: %s" % cmd, debug=True)
-                os.system(cmd)
-        finally:
-            # We never delete the lockfile, this way the inode will be
-            # constant while the machine is up. See comment on 'acquire'
-            self.lock.release(skip_delete=True)
+
+        timestamp = time.strftime("%Y%m%d-%H%M%S")
+        path = "upload%s-%s-%06d" % (
+            self.prefix, timestamp, self.targetcount)
+        target_fsroot = os.path.join(self.targetpath, path)
+
+        # Move the session directory to the target directory.
+        if os.path.exists(target_fsroot):
+            log.msg("Targeted upload already present: %s" % path)
+            log.msg("System clock skewed?")
+        else:
+            try:
+                os.rename(fsroot, target_fsroot)
+            except (OSError, IOError):
+                if not os.path.exists(target_fsroot):
+                    raise
+
+        # XXX cprov 20071024: We should replace os.system call by os.chmod
+        # and fix the default permission value accordingly in txpkgupload
+        if self.perms is not None:
+            os.system("chmod %s -R %s" % (self.perms, target_fsroot))
 
         self.clients.pop(fsroot)
         # This is mainly done so that tests know when the
         # post-processing hook has finished.
         log.msg(self.LOG_MAGIC)
-
-    def auth_verify_hook(self, fsroot, user, password):
-        """Verify that the username matches a distribution we care about.
-
-        The password is irrelevant to auth, as is the fsroot"""
-        if fsroot not in self.clients:
-            raise InterfaceFailure("Unable to find fsroot in client set")
-
-        # local authentication
-        self.clients[fsroot]["distro"] = self.allow_user
-        return True
-
-        # When we get on with the txpkgupload path stuff, the below may be
-        # useful and is thus left in rather than being removed.
-
-        #try:
-        #    d = Distribution.byName(user)
-        #    if d:
-        #        log.msg("Accepting login for %s" % user, debug=True)
-        #        self.clients[fsroot]["distro"] = user
-        #        return True
-        #except object as e:
-        #    print e
-        #return False
-

=== modified file 'src/txpkgupload/plugin.py'
--- src/txpkgupload/plugin.py	2015-01-12 18:50:14 +0000
+++ src/txpkgupload/plugin.py	2015-01-21 13:10:45 +0000
@@ -7,6 +7,7 @@
     ]
 
 from functools import partial
+import os
 
 from formencode import Schema
 from formencode.api import set_stdtranslation
@@ -119,6 +120,10 @@
     # Where on the filesystem do uploads live?
     fsroot = String(if_missing=None)
 
+    # Where do we write temporary files during uploads?  This must be on the
+    # same filesystem as fsroot.
+    temp_dir = String(if_missing=None)
+
     @classmethod
     def parse(cls, stream):
         """Load a YAML configuration from `stream` and validate."""
@@ -143,19 +148,22 @@
     """An SSH avatar specific to txpkgupload.
 
     :ivar fs_root: The file system root for this session.
+    :ivar temp_dir: The temporary directory for this session.
     """
 
-    def __init__(self, user_dict, fs_root):
+    def __init__(self, user_dict, fs_root, temp_dir):
         LaunchpadAvatar.__init__(self, user_dict)
         self.fs_root = fs_root
+        self.temp_dir = temp_dir
 
 
 class Realm:
     implements(IRealm)
 
-    def __init__(self, authentication_proxy, fs_root):
+    def __init__(self, authentication_proxy, fs_root, temp_dir):
         self.authentication_proxy = authentication_proxy
         self.fs_root = fs_root
+        self.temp_dir = temp_dir
 
     def requestAvatar(self, avatar_id, mind, *interfaces):
         # Fetch the user's details from the authserver
@@ -164,20 +172,20 @@
 
         # Once all those details are retrieved, we can construct the avatar.
         def got_user_dict(user_dict):
-            avatar = PkgUploadAvatar(user_dict, self.fs_root)
+            avatar = PkgUploadAvatar(user_dict, self.fs_root, self.temp_dir)
             return interfaces[0], avatar, avatar.logout
 
         return deferred.addCallback(got_user_dict)
 
 
-def make_portal(authentication_endpoint, fs_root):
+def make_portal(authentication_endpoint, fs_root, temp_dir):
     """Create and return a `Portal` for the SSH service.
 
     This portal accepts SSH credentials and returns our customized SSH
     avatars (see `LaunchpadAvatar`).
     """
     authentication_proxy = Proxy(authentication_endpoint)
-    portal = Portal(Realm(authentication_proxy, fs_root))
+    portal = Portal(Realm(authentication_proxy, fs_root, temp_dir))
     portal.registerChecker(
         PublicKeyFromLaunchpadChecker(authentication_proxy))
     return portal
@@ -212,18 +220,26 @@
             oops_dir, oops_reporter, server_argv=server_argv)
 
         root = get_txpkgupload_root(config["fsroot"])
+        temp_dir = config["temp_dir"]
+        if temp_dir is None:
+            temp_dir = os.path.abspath(os.path.join(
+                root, os.pardir, "tmp-incoming"))
+        if not os.path.exists(temp_dir):
+            os.makedirs(temp_dir, 0775)
 
         ftp_config = config["ftp"]
         ftp_service = FTPServiceFactory.makeFTPService(
             port=ftp_config["port"],
             root=root,
+            temp_dir=temp_dir,
             idle_timeout=config["idle_timeout"])
         ftp_service.name = "ftp"
         ftp_service.setServiceParent(services)
 
         sftp_config = config["sftp"]
         sftp_service = SSHService(
-            portal=make_portal(sftp_config["authentication_endpoint"], root),
+            portal=make_portal(
+                sftp_config["authentication_endpoint"], root, temp_dir),
             private_key_path=sftp_config["host_key_private"],
             public_key_path=sftp_config["host_key_public"],
             main_log='txpkgupload',

=== modified file 'src/txpkgupload/tests/test_plugin.py'
--- src/txpkgupload/tests/test_plugin.py	2015-01-12 13:14:55 +0000
+++ src/txpkgupload/tests/test_plugin.py	2015-01-21 13:10:45 +0000
@@ -115,6 +115,7 @@
                 "host_key_public": None,
                 "port": "tcp:5022",
                 },
+            "temp_dir": None,
             }
         observed = Config.to_python({})
         self.assertEqual(expected, observed)
@@ -262,6 +263,8 @@
 
     def __init__(self, root_dir):
         self.root_dir = root_dir
+        self.fsroot = os.path.join(self.root_dir, "incoming")
+        self.temp_dir = os.path.join(self.root_dir, "tmp-incoming")
         top = os.path.join(
             os.path.dirname(__file__), os.pardir, os.pardir, os.pardir)
         with open(os.path.join(top, "etc", "txpkgupload.yaml")) as stream:
@@ -270,8 +273,11 @@
 
     def setUp(self):
         super(FTPServer, self).setUp()
-        self.pkgupload = self.useFixture(PkgUploadFixture(
-            "fsroot: %s" % self.root_dir))
+        os.mkdir(self.fsroot)
+        os.mkdir(self.temp_dir)
+        self.pkgupload = self.useFixture(PkgUploadFixture(dedent("""\
+            fsroot: %s
+            temp_dir: %s""") % (self.fsroot, self.temp_dir)))
 
     def getAnonClient(self):
         creator = ClientCreator(
@@ -368,6 +374,8 @@
 
     def __init__(self, root_dir):
         self.root_dir = root_dir
+        self.fsroot = os.path.join(self.root_dir, "incoming")
+        self.temp_dir = os.path.join(self.root_dir, "tmp-incoming")
         #self._factory = factory
         top = os.path.join(
             os.path.dirname(__file__), os.pardir, os.pardir, os.pardir)
@@ -401,10 +409,14 @@
         self.authserver_url = b"http://localhost:%d/" % self.authserver_port
         self.addCleanup(self.authserver_listener.stopListening)
         self.setUpUser('joe')
+        os.mkdir(self.fsroot)
+        os.mkdir(self.temp_dir)
         self.pkgupload = self.useFixture(PkgUploadFixture(dedent("""\
             sftp:
               authentication_endpoint: %s
-            fsroot: %s""") % (self.authserver_url, self.root_dir)))
+            fsroot: %s
+            temp_dir: %s""") %
+            (self.authserver_url, self.fsroot, self.temp_dir)))
 
     @defer.inlineCallbacks
     def getClient(self):
@@ -460,8 +472,8 @@
     def setUp(self):
         """Set up txpkgupload in a temp dir."""
         super(TestPkgUploadServiceMakerMixin, self).setUp()
-        self.root_dir = self.useFixture(TempDir()).path
-        self.server = self.server_factory(self.root_dir)
+        root_dir = self.useFixture(TempDir()).path
+        self.server = self.server_factory(root_dir)
         self.useFixture(self.server)
 
     def test_init(self):
@@ -484,9 +496,9 @@
 
         Only works for a single upload (txpkgupload transaction).
         """
-        contents = sorted(os.listdir(self.root_dir))
-        upload_dir = contents[1]
-        return os.path.join(self.root_dir, upload_dir, path)
+        contents = sorted(os.listdir(self.server.fsroot))
+        upload_dir = contents[0]
+        return os.path.join(self.server.fsroot, upload_dir, path)
 
     @defer.inlineCallbacks
     def test_mkdir(self):
@@ -611,16 +623,15 @@
         yield self.server.waitForClose(4)
 
         # Build a list of directories representing the 4 sessions.
-        upload_dirs = [leaf for leaf in sorted(os.listdir(self.root_dir))
-                       if not leaf.startswith(".") and
-                       not leaf.endswith(".distro")]
+        upload_dirs = [leaf for leaf in sorted(os.listdir(self.server.fsroot))
+                       if not leaf.startswith(".")]
         self.assertEqual(len(upload_dirs), 4)
 
         # Check the contents of files on each session.
         expected_contents = ['ONE', 'TWO', 'THREE', 'FOUR']
         for index in range(4):
             content = open(os.path.join(
-                self.root_dir, upload_dirs[index], "test")).read()
+                self.server.fsroot, upload_dirs[index], "test")).read()
             self.assertEqual(content, expected_contents[index])
 
 

=== modified file 'src/txpkgupload/tests/test_twistedsftp.py'
--- src/txpkgupload/tests/test_twistedsftp.py	2015-01-21 12:20:52 +0000
+++ src/txpkgupload/tests/test_twistedsftp.py	2015-01-21 13:10:45 +0000
@@ -16,18 +16,20 @@
 
 class MockAvatar:
 
-    def __init__(self, fs_root):
+    def __init__(self, fs_root, temp_dir):
         self.fs_root = fs_root
+        self.temp_dir = temp_dir
 
 
 class TestSFTPServer(testtools.TestCase):
 
     def setUp(self):
-        tempdir = fixtures.TempDir()
-        self.useFixture(tempdir)
-        self.useFixture(fixtures.MonkeyPatch("tempfile.tempdir", tempdir.path))
-        self.fs_root = self.useFixture(fixtures.TempDir()).path
-        self.sftp_server = SFTPServer(MockAvatar(self.fs_root))
+        temp_dir = self.useFixture(fixtures.TempDir()).path
+        fs_root = os.path.join(temp_dir, "incoming")
+        temp_dir = os.path.join(temp_dir, "tmp-incoming")
+        os.mkdir(fs_root)
+        os.mkdir(temp_dir)
+        self.sftp_server = SFTPServer(MockAvatar(fs_root, temp_dir))
         super(TestSFTPServer, self).setUp()
 
     def assertPermissions(self, expected, file_name):

=== modified file 'src/txpkgupload/twistedftp.py'
--- src/txpkgupload/twistedftp.py	2015-01-09 15:55:08 +0000
+++ src/txpkgupload/twistedftp.py	2015-01-21 13:10:45 +0000
@@ -52,15 +52,14 @@
     Roughly equivalent to the SFTPServer in the sftp side of things.
     """
 
-    def __init__(self, fsroot):
+    def __init__(self, fsroot, temp_dir):
         self._fs_root = fsroot
-        self.uploadfilesystem = UploadFileSystem(tempfile.mkdtemp())
+        self.uploadfilesystem = UploadFileSystem(
+            tempfile.mkdtemp(dir=temp_dir))
         self._current_upload = self.uploadfilesystem.rootpath
         os.chmod(self._current_upload, 0770)
-        self.hook = Hooks(
-            self._fs_root, "ubuntu", perms='g+rws', prefix='-ftp')
+        self.hook = Hooks(self._fs_root, perms='g+rws', prefix='-ftp')
         self.hook.new_client_hook(self._current_upload, 0, 0)
-        self.hook.auth_verify_hook(self._current_upload, None, None)
         super(AnonymousShell, self).__init__(
             filepath.FilePath(self._current_upload))
 
@@ -113,8 +112,9 @@
     """FTP Realm that lets anyone in."""
     implements(IRealm)
 
-    def __init__(self, root):
+    def __init__(self, root, temp_dir):
         self.root = root
+        self.temp_dir = temp_dir
 
     def requestAvatar(self, avatarId, mind, *interfaces):
         """Return a txpkgupload avatar - that is, an "authorisation".
@@ -124,7 +124,7 @@
         """
         for iface in interfaces:
             if iface is ftp.IFTPShell:
-                avatar = AnonymousShell(self.root)
+                avatar = AnonymousShell(self.root, self.temp_dir)
                 return ftp.IFTPShell, avatar, getattr(
                     avatar, 'logout', lambda: None)
         raise NotImplementedError(
@@ -134,8 +134,8 @@
 class FTPServiceFactory(service.Service):
     """A factory that makes an `FTPService`"""
 
-    def __init__(self, port, root, idle_timeout):
-        realm = FTPRealm(root)
+    def __init__(self, port, root, temp_dir, idle_timeout):
+        realm = FTPRealm(root, temp_dir)
         portal = Portal(realm)
         portal.registerChecker(AccessCheck())
         factory = ftp.FTPFactory(portal)
@@ -149,7 +149,7 @@
         self.portno = port
 
     @staticmethod
-    def makeFTPService(port, root, idle_timeout):
+    def makeFTPService(port, root, temp_dir, idle_timeout):
         strport = "tcp:%s" % port
-        factory = FTPServiceFactory(port, root, idle_timeout)
+        factory = FTPServiceFactory(port, root, temp_dir, idle_timeout)
         return strports.service(strport, factory.ftpfactory)

=== modified file 'src/txpkgupload/twistedsftp.py'
--- src/txpkgupload/twistedsftp.py	2015-01-09 15:55:08 +0000
+++ src/txpkgupload/twistedsftp.py	2015-01-21 13:10:45 +0000
@@ -38,13 +38,12 @@
         provideHandler(self.connectionClosed)
         self._avatar = avatar
         self._fs_root = avatar.fs_root
-        self.uploadfilesystem = UploadFileSystem(tempfile.mkdtemp())
+        self.uploadfilesystem = UploadFileSystem(
+            tempfile.mkdtemp(dir=avatar.temp_dir))
         self._current_upload = self.uploadfilesystem.rootpath
         os.chmod(self._current_upload, 0770)
-        self.hook = Hooks(
-            self._fs_root, "ubuntu", perms='g+rws', prefix='-sftp')
+        self.hook = Hooks(self._fs_root, perms='g+rws', prefix='-sftp')
         self.hook.new_client_hook(self._current_upload, 0, 0)
-        self.hook.auth_verify_hook(self._current_upload, None, None)
 
     def gotVersion(self, other_version, ext_data):
         return {}