launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #03195
[Merge] lp:~thumper/launchpad/bzr-transport-branch-id-access into lp:launchpad
Tim Penhey has proposed merging lp:~thumper/launchpad/bzr-transport-branch-id-access into lp:launchpad with lp:~thumper/launchpad/anon-http-branch-id-access as a prerequisite.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~thumper/launchpad/bzr-transport-branch-id-access/+merge/56288
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.
--
https://code.launchpad.net/~thumper/launchpad/bzr-transport-branch-id-access/+merge/56288
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~thumper/launchpad/bzr-transport-branch-id-access into lp:launchpad.
=== 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 04:48:02 +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 04:48:02 +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,60 @@
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):
+ 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):
+ 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 04:48:02 +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')
