Merge lp:~thumper/launchpad/bzr-transport-branch-id-access into lp:launchpad

[Tim Penhey]

This branch provides support for the +branch-id alias over
the smartserver and sftp through our VFS.

All access to branches using the +branch-id alias are read
only.

[Graham Binns (gmb)]

This is good to land with a couple of additions:

124 + def test_translatePath_branch_id_alias_private_branch(self):

and

135 + def test_translatePath_branch_id_alias_private_branch_no_access(self):

Both need docstrings or explanatory comments so that I don't have to read them to understand what they test.

review: Approve (code)

=== modified file 'lib/lp/code/xmlrpc/codehosting.py'
--- lib/lp/code/xmlrpc/codehosting.py	2011-02-24 15:30:54 +0000
+++ lib/lp/code/xmlrpc/codehosting.py	2011-04-05 23:58:43 +0000
@@ -59,6 +59,7 @@
 from lp.code.interfaces.branchtarget import IBranchTarget
 from lp.code.interfaces.codehosting import (
     BRANCH_ALIAS_PREFIX,
+    BRANCH_ID_ALIAS_PREFIX,
     BRANCH_TRANSPORT,
     CONTROL_TRANSPORT,
     ICodehostingAPI,
@@ -287,7 +288,8 @@
 
         return run_with_login(login_id, branch_changed)
 
-    def _serializeBranch(self, requester, branch, trailing_path):
+    def _serializeBranch(self, requester, branch, trailing_path,
+                         force_readonly=False):
         if requester == LAUNCHPAD_SERVICES:
             branch = removeSecurityProxy(branch)
         try:
@@ -296,10 +298,13 @@
             raise faults.PermissionDenied()
         if branch.branch_type == BranchType.REMOTE:
             return None
+        if force_readonly:
+            writable = False
+        else:
+            writable = self._canWriteToBranch(requester, branch)
         return (
             BRANCH_TRANSPORT,
-            {'id': branch_id,
-             'writable': self._canWriteToBranch(requester, branch)},
+            {'id': branch_id, 'writable': writable},
             trailing_path)
 
     def _serializeControlDirectory(self, requester, product_path,
@@ -323,12 +328,34 @@
             {'default_stack_on': escape('/' + unique_name)},
             trailing_path)
 
+    def _translateBranchIdAlias(self, requester, path):
+        # If the path isn't a branch id alias, nothing more to do.
+        stripped_path = unescape(path.strip('/'))
+        if not stripped_path.startswith(BRANCH_ID_ALIAS_PREFIX + '/'):
+            return None
+        try:
+            parts = stripped_path.split('/', 2)
+            branch_id = int(parts[1])
+        except (ValueError, IndexError):
+            raise faults.PathTranslationError(path)
+        branch = getUtility(IBranchLookup).get(branch_id)
+        if branch is None:
+            raise faults.PathTranslationError(path)
+        try:
+            trailing = parts[2]
+        except IndexError:
+            trailing = ''
+        return self._serializeBranch(requester, branch, trailing, True)
+
     def translatePath(self, requester_id, path):
         """See `ICodehostingAPI`."""
         @return_fault
         def translate_path(requester):
             if not path.startswith('/'):
                 return faults.InvalidPath(path)
+            branch = self._translateBranchIdAlias(requester, path)
+            if branch is not None:
+                return branch
             stripped_path = path.strip('/')
             for first, second in iter_split(stripped_path, '/'):
                 first = unescape(first)
 
=== modified file 'lib/lp/code/xmlrpc/tests/test_codehosting.py'
--- lib/lp/code/xmlrpc/tests/test_codehosting.py	2010-10-13 03:56:46 +0000
+++ lib/lp/code/xmlrpc/tests/test_codehosting.py	2011-04-05 23:58:43 +0000
@@ -41,6 +41,7 @@
 from lp.code.interfaces.branchtarget import IBranchTarget
 from lp.code.interfaces.codehosting import (
     BRANCH_ALIAS_PREFIX,
+    BRANCH_ID_ALIAS_PREFIX,
     BRANCH_TRANSPORT,
     CONTROL_TRANSPORT,
     )
@@ -991,6 +992,63 @@
         requester = self.factory.makePerson()
         self.assertNotFound(requester, '/%s/.bzr' % BRANCH_ALIAS_PREFIX)
 
+    def test_translatePath_branch_id_alias_bzrdir_content(self):
+        # translatePath('/+branch-id/.bzr/.*') *must* return not found, otherwise
+        # bzr will look for it and we don't have a global bzr dir.
+        requester = self.factory.makePerson()
+        self.assertNotFound(
+            requester, '/%s/.bzr/branch-format' % BRANCH_ID_ALIAS_PREFIX)
+
+    def test_translatePath_branch_id_alias_bzrdir(self):
+        # translatePath('/+branch-id/.bzr') *must* return not found, otherwise
+        # bzr will look for it and we don't have a global bzr dir.
+        requester = self.factory.makePerson()
+        self.assertNotFound(requester, '/%s/.bzr' % BRANCH_ID_ALIAS_PREFIX)
+
+    def test_translatePath_branch_id_alias_trailing(self):
+        # Make sure the trailing path is returned.
+        requester = self.factory.makePerson()
+        branch = removeSecurityProxy(self.factory.makeAnyBranch())
+        path = escape(u'/%s/%s/foo/bar' % (BRANCH_ID_ALIAS_PREFIX, branch.id))
+        translation = self.codehosting_api.translatePath(requester.id, path)
+        self.assertEqual(
+            (BRANCH_TRANSPORT, {'id': branch.id, 'writable': False}, 'foo/bar'),
+            translation)
+
+    def test_translatePath_branch_id_alias_owned(self):
+        # Even if the the requester is the owner, the branch is read only.
+        requester = self.factory.makePerson()
+        branch = removeSecurityProxy(
+            self.factory.makeAnyBranch(
+                branch_type=BranchType.HOSTED, owner=requester))
+        path = escape(u'/%s/%s' % (BRANCH_ID_ALIAS_PREFIX, branch.id))
+        translation = self.codehosting_api.translatePath(requester.id, path)
+        self.assertEqual(
+            (BRANCH_TRANSPORT, {'id': branch.id, 'writable': False}, ''),
+            translation)
+
+    def test_translatePath_branch_id_alias_private_branch(self):
+        # Private branches are accessible but read-only even if you are the
+        # owner.
+        requester = self.factory.makePerson()
+        branch = removeSecurityProxy(
+            self.factory.makeAnyBranch(
+                branch_type=BranchType.HOSTED, private=True, owner=requester))
+        path = escape(u'/%s/%s' % (BRANCH_ID_ALIAS_PREFIX, branch.id))
+        translation = self.codehosting_api.translatePath(requester.id, path)
+        self.assertEqual(
+            (BRANCH_TRANSPORT, {'id': branch.id, 'writable': False}, ''),
+            translation)
+
+    def test_translatePath_branch_id_alias_private_branch_no_access(self):
+        # Private branches you don't have access to raise permission denied.
+        requester = self.factory.makePerson()
+        branch = removeSecurityProxy(
+            self.factory.makeAnyBranch(
+                branch_type=BranchType.HOSTED, private=True))
+        path = escape(u'/%s/%s' % (BRANCH_ID_ALIAS_PREFIX, branch.id))
+        self.assertPermissionDenied(requester, path)
+
     def assertTranslationIsControlDirectory(self, translation,
                                             default_stacked_on,
                                             trailing_path):
 
=== modified file 'lib/lp/codehosting/inmemory.py'
--- lib/lp/codehosting/inmemory.py	2011-02-24 15:30:54 +0000
+++ lib/lp/codehosting/inmemory.py	2011-04-05 23:58:43 +0000
@@ -37,6 +37,7 @@
 from lp.code.interfaces.branchtarget import IBranchTarget
 from lp.code.interfaces.codehosting import (
     BRANCH_ALIAS_PREFIX,
+    BRANCH_ID_ALIAS_PREFIX,
     BRANCH_TRANSPORT,
     CONTROL_TRANSPORT,
     LAUNCHPAD_ANONYMOUS,
@@ -806,21 +807,46 @@
             {'default_stack_on': escape('/' + default_branch.unique_name)},
             trailing_path)
 
-    def _serializeBranch(self, requester_id, branch, trailing_path):
+    def _serializeBranch(self, requester_id, branch, trailing_path,
+                         force_readonly=False):
         if not self._canRead(requester_id, branch):
             return faults.PermissionDenied()
         elif branch.branch_type == BranchType.REMOTE:
             return None
+        if force_readonly:
+            writable = False
         else:
-            return (
-                BRANCH_TRANSPORT,
-                {'id': branch.id,
-                 'writable': self._canWrite(requester_id, branch),
-                 }, trailing_path)
+            writable = self._canWrite(requester_id, branch)
+        return (
+            BRANCH_TRANSPORT,
+            {'id': branch.id, 'writable': writable},
+            trailing_path)
+
+    def _translateBranchIdAlias(self, requester, path):
+        # If the path isn't a branch id alias, nothing more to do.
+        stripped_path = unescape(path.strip('/'))
+        if not stripped_path.startswith(BRANCH_ID_ALIAS_PREFIX + '/'):
+            return None
+        try:
+            parts = stripped_path.split('/', 2)
+            branch_id = int(parts[1])
+        except (ValueError, IndexError):
+            return faults.PathTranslationError(path)
+        branch = self._branch_set.get(branch_id)
+        if branch is None:
+            return faults.PathTranslationError(path)
+        try:
+            trailing = parts[2]
+        except IndexError:
+            trailing = ''
+        return self._serializeBranch(requester, branch, trailing, True)
 
     def translatePath(self, requester_id, path):
         if not path.startswith('/'):
             return faults.InvalidPath(path)
+        branch = self._translateBranchIdAlias(requester_id, path)
+        if branch is not None:
+            return branch
         stripped_path = path.strip('/')
         for first, second in iter_split(stripped_path, '/'):
             first = unescape(first).encode('utf-8')