launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #04529
[Merge] lp:~jelmer/launchpad/820453-safe-open-dup into lp:launchpad
Jelmer Vernooij has proposed merging lp:~jelmer/launchpad/820453-safe-open-dup into lp:launchpad.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
Related bugs:
Bug #820453 in Launchpad itself: "duplicate methods for safely opening branches"
https://bugs.launchpad.net/launchpad/+bug/820453
For more details, see:
https://code.launchpad.net/~jelmer/launchpad/820453-safe-open-dup/+merge/70898
Make lp.codehosting.bzrutils.safe_open a trivial wrapper around SafeBranchOpener rather than duplicating the code for safely opening a branch in two places.
They both do basically the same thing, SafeBranchOpener is a bit more advanced and was recently extracted from BranchMirrorer.
--
https://code.launchpad.net/~jelmer/launchpad/820453-safe-open-dup/+merge/70898
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~jelmer/launchpad/820453-safe-open-dup into lp:launchpad.
=== modified file 'lib/launchpad_loggerhead/app.py'
--- lib/launchpad_loggerhead/app.py 2011-03-10 18:27:08 +0000
+++ lib/launchpad_loggerhead/app.py 2011-08-09 14:59:28 +0000
@@ -32,7 +32,7 @@
from lp.code.interfaces.codehosting import (
BRANCH_TRANSPORT, LAUNCHPAD_ANONYMOUS)
from lp.codehosting.vfs import get_lp_server
-from lp.codehosting.bzrutils import safe_open
+from lp.codehosting.safe_open import safe_open
robots_txt = '''\
User-agent: *
=== modified file 'lib/lp/code/interfaces/branch.py'
--- lib/lp/code/interfaces/branch.py 2011-07-11 03:58:41 +0000
+++ lib/lp/code/interfaces/branch.py 2011-08-09 14:59:28 +0000
@@ -911,7 +911,7 @@
You can only call this if a server returned by `get_ro_server` or
`get_rw_server` is running.
- :raise lp.codehosting.bzrutils.UnsafeUrlSeen: If the branch is stacked
+ :raise lp.codehosting.safe_open.BadUrl: If the branch is stacked
on or a reference to an unacceptable URL.
"""
=== modified file 'lib/lp/code/model/branch.py'
--- lib/lp/code/model/branch.py 2011-06-24 18:08:18 +0000
+++ lib/lp/code/model/branch.py 2011-08-09 14:59:28 +0000
@@ -136,7 +136,7 @@
RevisionAuthor,
)
from lp.code.model.seriessourcepackagebranch import SeriesSourcePackageBranch
-from lp.codehosting.bzrutils import safe_open
+from lp.codehosting.safe_open import safe_open
from lp.registry.interfaces.person import (
validate_person,
validate_public_person,
=== modified file 'lib/lp/code/model/tests/test_branch.py'
--- lib/lp/code/model/tests/test_branch.py 2011-07-11 03:58:41 +0000
+++ lib/lp/code/model/tests/test_branch.py 2011-08-09 14:59:28 +0000
@@ -108,7 +108,7 @@
from lp.code.model.codereviewcomment import CodeReviewComment
from lp.code.model.revision import Revision
from lp.code.tests.helpers import add_revision_to_branch
-from lp.codehosting.bzrutils import UnsafeUrlSeen
+from lp.codehosting.safe_open import BadUrl
from lp.registry.interfaces.pocket import PackagePublishingPocket
from lp.registry.model.sourcepackage import SourcePackage
from lp.services.osutils import override_environ
@@ -2686,7 +2686,7 @@
class TestGetBzrBranch(TestCaseWithFactory):
- """Tests for `IBranch.safe_open`."""
+ """Tests for `IBranch.getBzrBranch`."""
layer = DatabaseFunctionalLayer
@@ -2722,7 +2722,7 @@
branch = BzrDir.create_branch_convenience('local')
db_stacked, stacked_tree = self.create_branch_and_tree()
stacked_tree.branch.set_stacked_on_url(branch.base)
- self.assertRaises(UnsafeUrlSeen, db_stacked.getBzrBranch)
+ self.assertRaises(BadUrl, db_stacked.getBzrBranch)
class TestMergeQueue(TestCaseWithFactory):
=== modified file 'lib/lp/codehosting/bzrutils.py'
--- lib/lp/codehosting/bzrutils.py 2011-05-25 19:57:07 +0000
+++ lib/lp/codehosting/bzrutils.py 2011-08-09 14:59:28 +0000
@@ -22,20 +22,16 @@
'read_locked',
'remove_exception_logging_hook',
'safe_open',
- 'UnsafeUrlSeen',
]
from contextlib import contextmanager
import os
import sys
-import threading
from bzrlib import (
config,
trace,
)
-from bzrlib.branch import Branch
-from bzrlib.bzrdir import BzrDir
from bzrlib.errors import (
NotStacked,
UnstackableBranchFormat,
@@ -292,76 +288,6 @@
get_vfs_format_classes(branch_two))
-checked_open_data = threading.local()
-
-
-def _install_checked_open_hook():
- """Install `_checked_open_pre_open_hook` as a ``pre_open`` hook.
-
- This is done at module import time, but _checked_open_pre_open_hook
- doesn't do anything unless the `checked_open_data` threading.Local object
- has a 'checked_opener' attribute in this thread.
-
- This is in a module-level function rather than performed at module level
- so that it can be called in setUp for testing `checked_open` as
- bzrlib.tests.TestCase.setUp clears hooks.
- """
- BzrDir.hooks.install_named_hook(
- 'pre_open', _checked_open_pre_open_hook, 'safe open')
-
-
-def _checked_open_pre_open_hook(transport):
- """If a checked_open validate function is present in this thread, call it.
- """
- if not getattr(checked_open_data, 'validate', False):
- return
- checked_open_data.validate(transport.base)
-
-
-_install_checked_open_hook()
-
-
-def checked_open(validation_function, url, possible_transports=None):
- """Open a branch, calling `validation_function` with any URL thus found.
-
- This is intended to be used to open a branch ensuring that it's not
- stacked or a reference to something unexpected.
- """
- if hasattr(checked_open_data, 'validate'):
- raise AssertionError("checked_open called recursively")
- checked_open_data.validate = validation_function
- try:
- return Branch.open(url, possible_transports=possible_transports)
- finally:
- del checked_open_data.validate
-
-
-class UnsafeUrlSeen(Exception):
- """`safe_open` found a URL that was not on the configured scheme."""
-
-
-def makeURLChecker(allowed_scheme):
- """Make a callable that rejects URLs not on the given scheme."""
-
- def checkURL(url):
- """Check that `url` is safe to open."""
- if URI(url).scheme != allowed_scheme:
- raise UnsafeUrlSeen(
- "Attempt to open %r which is not a %s URL" % (
- url, allowed_scheme))
- return checkURL
-
-
-def safe_open(allowed_scheme, url, possible_transports=None):
- """Open the branch at `url`, only accessing URLs on `allowed_scheme`.
-
- :raises UnsafeUrlSeen: An attempt was made to open a URL that was not on
- `allowed_scheme`.
- """
- return checked_open(
- makeURLChecker(allowed_scheme), url, possible_transports)
-
-
def get_stacked_on_url(branch):
"""Get the stacked-on URL for 'branch', or `None` if not stacked."""
try:
=== modified file 'lib/lp/codehosting/safe_open.py'
--- lib/lp/codehosting/safe_open.py 2011-08-06 18:45:40 +0000
+++ lib/lp/codehosting/safe_open.py 2011-08-09 14:59:28 +0000
@@ -9,6 +9,8 @@
from bzrlib.branch import Branch
from bzrlib.bzrdir import BzrDir
+from lazr.uri import URI
+
__all__ = [
'AcceptAnythingPolicy',
'BadUrl',
@@ -231,3 +233,30 @@
url = self.checkAndFollowBranchReference(url)
return self.runWithTransformFallbackLocationHookInstalled(
Branch.open, url)
+
+
+class URLChecker(BranchOpenPolicy):
+ """Branch open policy that rejects URLs not on the given scheme."""
+
+ def __init__(self, allowed_scheme):
+ self.allowed_scheme = allowed_scheme
+
+ def shouldFollowReferences(self):
+ return True
+
+ def transformFallbackLocation(self, branch, url):
+ return urlutils.join(branch.base, url), True
+
+ def checkOneURL(self, url):
+ """Check that `url` is safe to open."""
+ if URI(url).scheme != self.allowed_scheme:
+ raise BadUrl(url)
+
+
+def safe_open(allowed_scheme, url):
+ """Open the branch at `url`, only accessing URLs on `allowed_scheme`.
+
+ :raises BadUrl: An attempt was made to open a URL that was not on
+ `allowed_scheme`.
+ """
+ return SafeBranchOpener(URLChecker(allowed_scheme)).open(url)
=== modified file 'lib/lp/codehosting/tests/test_bzrutils.py'
--- lib/lp/codehosting/tests/test_bzrutils.py 2011-05-19 13:31:09 +0000
+++ lib/lp/codehosting/tests/test_bzrutils.py 2011-08-09 14:59:28 +0000
@@ -33,11 +33,8 @@
branch_scenarios,
TestCaseWithControlDir,
)
-from bzrlib.transport import chroot
-from lazr.uri import URI
from lp.codehosting.bzrutils import (
- _install_checked_open_hook,
add_exception_logging_hook,
checked_open,
DenyingServer,
@@ -45,8 +42,6 @@
get_vfs_format_classes,
is_branch_stackable,
remove_exception_logging_hook,
- safe_open,
- UnsafeUrlSeen,
)
from lp.codehosting.tests.helpers import TestResultWrapper
@@ -232,10 +227,6 @@
class TestCheckedOpen(TestCaseWithTransport):
"""Tests for `checked_open`."""
- def setUp(self):
- TestCaseWithTransport.setUp(self)
- _install_checked_open_hook()
-
def test_simple(self):
# Opening a branch with checked_open checks the branches url.
url = self.make_branch('branch').base
@@ -265,52 +256,6 @@
self.assertEqual([reference_url, target.base], seen_urls)
-class TestSafeOpen(TestCaseWithTransport):
- """Tests for `safe_open`."""
-
- def setUp(self):
- TestCaseWithTransport.setUp(self)
- _install_checked_open_hook()
-
- def get_chrooted_scheme(self, relpath):
- """Create a server that is chrooted to `relpath`.
-
- :return: ``(scheme, get_url)`` where ``scheme`` is the scheme of the
- chroot server and ``get_url`` returns URLs on said server.
- """
- transport = self.get_transport(relpath)
- chroot_server = chroot.ChrootServer(transport)
- chroot_server.start_server()
- self.addCleanup(chroot_server.stop_server)
- def get_url(relpath):
- return chroot_server.get_url() + relpath
- return URI(chroot_server.get_url()).scheme, get_url
-
- def test_stacked_within_scheme(self):
- # A branch that is stacked on a URL of the same scheme is safe to
- # open.
- self.get_transport().mkdir('inside')
- self.make_branch('inside/stacked')
- self.make_branch('inside/stacked-on')
- scheme, get_chrooted_url = self.get_chrooted_scheme('inside')
- Branch.open(get_chrooted_url('stacked')).set_stacked_on_url(
- get_chrooted_url('stacked-on'))
- safe_open(scheme, get_chrooted_url('stacked'))
-
- def test_stacked_outside_scheme(self):
- # A branch that is stacked on a URL that is not of the same scheme is
- # not safe to open.
- self.get_transport().mkdir('inside')
- self.get_transport().mkdir('outside')
- self.make_branch('inside/stacked')
- self.make_branch('outside/stacked-on')
- scheme, get_chrooted_url = self.get_chrooted_scheme('inside')
- Branch.open(get_chrooted_url('stacked')).set_stacked_on_url(
- self.get_url('outside/stacked-on'))
- self.assertRaises(
- UnsafeUrlSeen, safe_open, scheme, get_chrooted_url('stacked'))
-
-
def load_tests(basic_tests, module, loader):
"""Parametrize the tests of get_branch_stacked_on_url by branch format."""
result = loader.suiteClass()
@@ -325,7 +270,6 @@
result.addTests(loader.loadTestsFromTestCase(TestDenyingServer))
result.addTests(loader.loadTestsFromTestCase(TestExceptionLoggingHooks))
result.addTests(loader.loadTestsFromTestCase(TestGetVfsFormatClasses))
- result.addTests(loader.loadTestsFromTestCase(TestSafeOpen))
result.addTests(loader.loadTestsFromTestCase(TestCheckedOpen))
return result
