launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #23730
[Merge] lp:~cjwatson/launchpad/use-bzr-policy-open into lp:launchpad
Colin Watson has proposed merging lp:~cjwatson/launchpad/use-bzr-policy-open into lp:launchpad.
Commit message:
Use new bzrlib.url_policy_open API for safely opening branches.
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/use-bzr-policy-open/+merge/368761
Launchpad's safe_open API was fairly generic and has been moved into bzrlib as bzrlib.url_policy_open.
This branch changes Launchpad to use the copy of safe_open that is present in bzrlib, and removes it from Launchpad itself.
This is a resurrection of https://code.launchpad.net/~jelmer/launchpad/use-bzr-policy-open/+merge/112529, updated to current Launchpad and with some bugs fixed.
--
Your team Launchpad code reviewers is requested to review the proposed merge of lp:~cjwatson/launchpad/use-bzr-policy-open into lp:launchpad.
=== modified file 'constraints.txt'
--- constraints.txt 2019-06-07 09:52:52 +0000
+++ constraints.txt 2019-06-13 10:09:50 +0000
@@ -234,7 +234,8 @@
beautifulsoup4[lxml]==4.7.1
billiard==3.5.0.5
bson==0.3.3
-bzr==2.6.0.lp.3
+# lp:~launchpad/bzr/lp
+bzr==2.6.0.lp.4
celery==4.1.1
cffi==1.11.2
Chameleon==2.11
=== modified file 'lib/launchpad_loggerhead/app.py'
--- lib/launchpad_loggerhead/app.py 2018-05-16 14:38:22 +0000
+++ lib/launchpad_loggerhead/app.py 2019-06-13 10:09:50 +0000
@@ -18,6 +18,7 @@
favicon_app,
static_app,
)
+from bzrlib.url_policy_open import open_only_scheme
from loggerhead.apps.branch import BranchWSGIApp
import oops_wsgi
from openid.consumer.consumer import (
@@ -47,7 +48,6 @@
LAUNCHPAD_ANONYMOUS,
LAUNCHPAD_SERVICES,
)
-from lp.codehosting.safe_open import safe_open
from lp.codehosting.vfs import get_lp_server
from lp.services.config import config
from lp.services.webapp.errorlog import ErrorReportingUtility
@@ -275,7 +275,7 @@
self.log.info("Branch is public")
try:
- bzr_branch = safe_open(
+ bzr_branch = open_only_scheme(
lp_server.get_url().strip(':/'), branch_url)
except errors.NotBranchError as err:
self.log.warning('Not a branch: %s', err)
=== modified file 'lib/lp/code/interfaces/branch.py'
--- lib/lp/code/interfaces/branch.py 2019-04-17 12:03:22 +0000
+++ lib/lp/code/interfaces/branch.py 2019-06-13 10:09:50 +0000
@@ -1012,7 +1012,7 @@
You can only call this if a server returned by `get_ro_server` or
`get_rw_server` is running.
- :raise lp.codehosting.safe_open.BadUrl: If the branch is stacked
+ :raise bzrlib.url_policy_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 2019-04-17 10:34:13 +0000
+++ lib/lp/code/model/branch.py 2019-06-13 10:09:50 +0000
@@ -16,6 +16,7 @@
from bzrlib import urlutils
from bzrlib.revision import NULL_REVISION
+from bzrlib.url_policy_open import open_only_scheme
from lazr.lifecycle.event import ObjectCreatedEvent
import pytz
import six
@@ -138,7 +139,6 @@
RevisionAuthor,
)
from lp.code.model.seriessourcepackagebranch import SeriesSourcePackageBranch
-from lp.codehosting.safe_open import safe_open
from lp.registry.enums import PersonVisibility
from lp.registry.errors import CannotChangeInformationType
from lp.registry.interfaces.accesspolicy import (
@@ -733,11 +733,11 @@
def getInternalBzrUrl(self):
"""See `IBranch`."""
- return 'lp-internal:///' + self.unique_name
+ return six.ensure_str('lp-internal:///' + self.unique_name)
def getBzrBranch(self):
"""See `IBranch`."""
- return safe_open('lp-internal', self.getInternalBzrUrl())
+ return open_only_scheme('lp-internal', self.getInternalBzrUrl())
@property
def display_name(self):
=== modified file 'lib/lp/code/model/tests/test_branch.py'
--- lib/lp/code/model/tests/test_branch.py 2019-05-22 14:57:45 +0000
+++ lib/lp/code/model/tests/test_branch.py 2019-06-13 10:09:50 +0000
@@ -16,6 +16,7 @@
from bzrlib.branch import Branch
from bzrlib.bzrdir import BzrDir
from bzrlib.revision import NULL_REVISION
+from bzrlib.url_policy_open import BadUrl
from pytz import UTC
from sqlobject import SQLObjectNotFound
from storm.exceptions import LostObjectError
@@ -116,7 +117,6 @@
add_revision_to_branch,
BranchHostingFixture,
)
-from lp.codehosting.safe_open import BadUrl
from lp.codehosting.vfs.branchfs import get_real_branch_path
from lp.registry.enums import (
BranchSharingPolicy,
@@ -3271,7 +3271,7 @@
self.useBzrBranches(direct_database=True)
def test_simple(self):
- # safe_open returns the underlying bzr branch of a database branch in
+ # open_only_scheme returns the underlying bzr branch of a database branch in
# the simple, unstacked, case.
db_branch, tree = self.create_branch_and_tree()
# XXX: AaronBentley 2010-08-06 bug=614404: a bzr username is
@@ -3283,7 +3283,7 @@
def test_acceptable_stacking(self):
# If the underlying bzr branch of a database branch is stacked on
- # another launchpad branch safe_open returns it.
+ # another launchpad branch open_only_scheme returns it.
db_stacked_on, stacked_on_tree = self.create_branch_and_tree()
db_stacked, stacked_tree = self.create_branch_and_tree()
stacked_tree.branch.set_stacked_on_url(
=== modified file 'lib/lp/codehosting/codeimport/tests/test_worker.py'
--- lib/lp/codehosting/codeimport/tests/test_worker.py 2019-05-22 14:57:45 +0000
+++ lib/lp/codehosting/codeimport/tests/test_worker.py 2019-06-13 10:09:50 +0000
@@ -40,6 +40,13 @@
join as urljoin,
local_path_from_url,
)
+from bzrlib.url_policy_open import (
+ _BlacklistPolicy,
+ AcceptAnythingPolicy,
+ BadUrl,
+ BranchOpenPolicy,
+ BranchOpener,
+ )
from CVS import (
Repository,
tree as CVSTree,
@@ -76,13 +83,6 @@
ImportDataStore,
ToBzrImportWorker,
)
-from lp.codehosting.safe_open import (
- AcceptAnythingPolicy,
- BadUrl,
- BlacklistPolicy,
- BranchOpenPolicy,
- SafeBranchOpener,
- )
from lp.codehosting.tests.helpers import create_branch_with_one_revision
from lp.services.config import config
from lp.services.log.logger import BufferLogger
@@ -129,7 +129,7 @@
def setUp(self):
super(WorkerTest, self).setUp()
self.disable_directory_isolation()
- SafeBranchOpener.install_hook()
+ BranchOpener.install_hook()
def assertDirectoryTreesEqual(self, directory1, directory2):
"""Assert that `directory1` has the same structure as `directory2`.
@@ -1242,7 +1242,7 @@
cache_dir = get_cache(uuid).create_cache_dir()
cache_dir_contents = os.listdir(cache_dir)
self.assertNotEqual([], cache_dir_contents)
- opener = SafeBranchOpener(worker._opener_policy, worker.probers)
+ opener = BranchOpener(worker._opener_policy, worker.probers)
remote_branch = opener.open(worker.source_details.url)
worker.pushBazaarBranch(
self.make_branch('.'), remote_branch=remote_branch)
@@ -1331,7 +1331,7 @@
reference_url, target_url = self.createBranchReference()
source_details = self.factory.makeCodeImportSourceDetails(
url=reference_url, rcstype='bzr')
- policy = BlacklistPolicy(True, [target_url])
+ policy = _BlacklistPolicy(True, [target_url])
worker = self.makeImportWorker(source_details, opener_policy=policy)
self.assertEqual(
CodeImportWorkerExitCode.FAILURE_FORBIDDEN, worker.run())
@@ -1363,15 +1363,15 @@
self.policy = CodeImportBranchOpenPolicy("bzr", "bzr")
def test_follows_references(self):
- self.assertEqual(True, self.policy.shouldFollowReferences())
+ self.assertEqual(True, self.policy.should_follow_references())
def assertBadUrl(self, url):
- self.assertRaises(BadUrl, self.policy.checkOneURL, url)
+ self.assertRaises(BadUrl, self.policy.check_one_url, url)
def assertGoodUrl(self, url):
- self.policy.checkOneURL(url)
+ self.policy.check_one_url(url)
- def test_checkOneURL(self):
+ def test_check_one_url(self):
self.assertBadUrl("sftp://somehost/";)
self.assertBadUrl("/etc/passwd")
self.assertBadUrl("file:///etc/passwd")
@@ -1384,7 +1384,7 @@
self.assertGoodUrl("bzr://bzr.example.com/somebzrurl/")
self.assertBadUrl("bzr://bazaar.launchpad.test/example")
- def test_checkOneURL_git_to_bzr(self):
+ def test_check_one_url_git_to_bzr(self):
self.policy = CodeImportBranchOpenPolicy("git", "bzr")
self.assertBadUrl("/etc/passwd")
self.assertBadUrl("file:///etc/passwd")
@@ -1392,7 +1392,7 @@
self.assertGoodUrl("git://git.example.com/repo")
self.assertGoodUrl("git://git.launchpad.test/example")
- def test_checkOneURL_git_to_git(self):
+ def test_check_one_url_git_to_git(self):
self.policy = CodeImportBranchOpenPolicy("git", "git")
self.assertBadUrl("/etc/passwd")
self.assertBadUrl("file:///etc/passwd")
@@ -1408,7 +1408,7 @@
def setUp(self):
http_utils.TestCaseWithRedirectedWebserver.setUp(self)
self.disable_directory_isolation()
- SafeBranchOpener.install_hook()
+ BranchOpener.install_hook()
tree = self.make_branch_and_tree('.')
self.revid = tree.commit("A commit")
self.bazaar_store = BazaarBranchStore(
@@ -1438,14 +1438,14 @@
def __init__(self, new_url):
self.new_url = new_url
- def shouldFollowReferences(self):
+ def should_follow_references(self):
return True
- def checkOneURL(self, url):
+ def check_one_url(self, url):
if url.startswith(self.new_url):
raise BadUrl(url)
- def transformFallbackLocation(self, branch, url):
+ def transform_fallback_location(self, branch, url):
return urlutils.join(branch.base, url), False
policy = NewUrlBlacklistPolicy(self.get_new_url())
=== modified file 'lib/lp/codehosting/codeimport/worker.py'
--- lib/lp/codehosting/codeimport/worker.py 2019-03-07 10:37:59 +0000
+++ lib/lp/codehosting/codeimport/worker.py 2019-06-13 10:09:50 +0000
@@ -59,6 +59,12 @@
join as urljoin,
local_path_from_url,
)
+from bzrlib.url_policy_open import (
+ BadUrl,
+ BranchOpenPolicy,
+ BranchOpener,
+ )
+
import cscvs
from cscvs.cmds import totla
import CVS
@@ -81,11 +87,6 @@
extract_tarball,
)
from lp.codehosting.codeimport.uifactory import LoggingUIFactory
-from lp.codehosting.safe_open import (
- BadUrl,
- BranchOpenPolicy,
- SafeBranchOpener,
- )
from lp.services.config import config
from lp.services.propertycache import cachedproperty
from lp.services.timeout import urlfetch
@@ -108,8 +109,8 @@
self.rcstype = rcstype
self.target_rcstype = target_rcstype
- def shouldFollowReferences(self):
- """See `BranchOpenPolicy.shouldFollowReferences`.
+ def should_follow_references(self):
+ """See `BranchOpenPolicy.should_follow_references`.
We traverse branch references for MIRRORED branches because they
provide a useful redirection mechanism and we want to be consistent
@@ -117,16 +118,16 @@
"""
return True
- def transformFallbackLocation(self, branch, url):
- """See `BranchOpenPolicy.transformFallbackLocation`.
+ def transform_fallback_location(self, branch, url):
+ """See `BranchOpenPolicy.transform_fallback_location`.
For mirrored branches, we stack on whatever the remote branch claims
to stack on, but this URL still needs to be checked.
"""
return urljoin(branch.base, url), True
- def checkOneURL(self, url):
- """See `BranchOpenPolicy.checkOneURL`.
+ def check_one_url(self, url):
+ """See `BranchOpenPolicy.check_one_url`.
We refuse to mirror Bazaar branches from Launchpad, or any branches
from a ssh-like or file URL.
@@ -726,7 +727,7 @@
def _doImport(self):
self._logger.info("Starting job.")
saved_factory = bzrlib.ui.ui_factory
- opener = SafeBranchOpener(self._opener_policy, self.probers)
+ opener = BranchOpener(self._opener_policy, self.probers)
bzrlib.ui.ui_factory = LoggingUIFactory(logger=self._logger)
try:
self._logger.info(
@@ -1046,7 +1047,7 @@
def _doImport(self):
self._logger.info("Starting job.")
try:
- self._opener_policy.checkOneURL(self.source_details.url)
+ self._opener_policy.check_one_url(self.source_details.url)
except BadUrl as e:
self._logger.info("Invalid URL: %s" % e)
return CodeImportWorkerExitCode.FAILURE_FORBIDDEN
=== modified file 'lib/lp/codehosting/puller/tests/__init__.py'
--- lib/lp/codehosting/puller/tests/__init__.py 2012-04-16 23:02:44 +0000
+++ lib/lp/codehosting/puller/tests/__init__.py 2019-06-13 10:09:50 +0000
@@ -16,6 +16,7 @@
TestingHTTPServer,
TestingThreadingHTTPServer,
)
+from bzrlib.url_policy_open import AcceptAnythingPolicy
from lp.codehosting.puller.worker import (
BranchMirrorer,
@@ -23,7 +24,6 @@
PullerWorker,
PullerWorkerProtocol,
)
-from lp.codehosting.safe_open import AcceptAnythingPolicy
from lp.codehosting.tests.helpers import LoomTestMixin
from lp.testing import TestCaseWithFactory
=== modified file 'lib/lp/codehosting/puller/tests/test_errors.py'
--- lib/lp/codehosting/puller/tests/test_errors.py 2019-05-22 14:57:45 +0000
+++ lib/lp/codehosting/puller/tests/test_errors.py 2019-06-13 10:09:50 +0000
@@ -18,6 +18,10 @@
UnknownFormatError,
UnsupportedFormatError,
)
+from bzrlib.url_policy_open import (
+ BranchLoopError,
+ BranchReferenceForbidden,
+ )
from lazr.uri import InvalidURIError
from lp.code.enums import BranchType
@@ -29,10 +33,6 @@
PullerWorker,
PullerWorkerProtocol,
)
-from lp.codehosting.safe_open import (
- BranchLoopError,
- BranchReferenceForbidden,
- )
from lp.testing import TestCase
=== modified file 'lib/lp/codehosting/puller/tests/test_worker.py'
--- lib/lp/codehosting/puller/tests/test_worker.py 2019-05-22 14:57:45 +0000
+++ lib/lp/codehosting/puller/tests/test_worker.py 2019-06-13 10:09:50 +0000
@@ -22,6 +22,12 @@
TestCaseWithTransport,
)
from bzrlib.transport import get_transport
+from bzrlib.url_policy_open import (
+ AcceptAnythingPolicy,
+ BadUrl,
+ BranchOpenPolicy,
+ BranchOpener,
+ )
from lp.code.enums import BranchType
from lp.codehosting.puller.tests import (
@@ -37,14 +43,8 @@
install_worker_ui_factory,
MirroredBranchPolicy,
PullerWorkerProtocol,
- SafeBranchOpener,
WORKER_ACTIVITY_NETWORK,
)
-from lp.codehosting.safe_open import (
- AcceptAnythingPolicy,
- BadUrl,
- BranchOpenPolicy,
- )
from lp.testing import TestCase
from lp.testing.factory import (
LaunchpadObjectFactory,
@@ -86,7 +86,7 @@
def setUp(self):
super(TestPullerWorker, self).setUp()
- SafeBranchOpener.install_hook()
+ BranchOpener.install_hook()
def test_mirror_opener_with_stacked_on_url(self):
# A PullerWorker for a mirrored branch gets a MirroredBranchPolicy as
@@ -277,7 +277,7 @@
def setUp(self):
super(TestReferenceOpener, self).setUp()
- SafeBranchOpener.install_hook()
+ BranchOpener.install_hook()
def createBranchReference(self, url):
"""Create a pure branch reference that points to the specified URL.
@@ -318,20 +318,20 @@
self.assertEqual(opened_branch.base, target_branch.base)
def testFollowReferenceValue(self):
- # SafeBranchOpener.followReference gives the reference value for
+ # BranchOpener.follow_reference gives the reference value for
# a branch reference.
- opener = SafeBranchOpener(BranchOpenPolicy())
+ opener = BranchOpener(BranchOpenPolicy())
reference_value = 'http://example.com/branch'
reference_url = self.createBranchReference(reference_value)
self.assertEqual(
- reference_value, opener.followReference(reference_url))
+ reference_value, opener.follow_reference(reference_url))
def testFollowReferenceNone(self):
- # SafeBranchOpener.followReference gives None for a normal branch.
+ # BranchOpener.follow_reference gives None for a normal branch.
self.make_branch('repo')
branch_url = self.get_url('repo')
- opener = SafeBranchOpener(BranchOpenPolicy())
- self.assertIs(None, opener.followReference(branch_url))
+ opener = BranchOpener(BranchOpenPolicy())
+ self.assertIs(None, opener.follow_reference(branch_url))
class TestMirroredBranchPolicy(TestCase):
@@ -344,44 +344,44 @@
def testNoFileURL(self):
policy = MirroredBranchPolicy()
self.assertRaises(
- BadUrlScheme, policy.checkOneURL,
+ BadUrlScheme, policy.check_one_url,
self.factory.getUniqueURL(scheme='file'))
def testNoUnknownSchemeURLs(self):
policy = MirroredBranchPolicy()
self.assertRaises(
- BadUrlScheme, policy.checkOneURL,
+ BadUrlScheme, policy.check_one_url,
self.factory.getUniqueURL(scheme='decorator+scheme'))
def testNoSSHURL(self):
policy = MirroredBranchPolicy()
self.assertRaises(
- BadUrlSsh, policy.checkOneURL,
+ BadUrlSsh, policy.check_one_url,
self.factory.getUniqueURL(scheme='bzr+ssh'))
def testNoSftpURL(self):
policy = MirroredBranchPolicy()
self.assertRaises(
- BadUrlSsh, policy.checkOneURL,
+ BadUrlSsh, policy.check_one_url,
self.factory.getUniqueURL(scheme='sftp'))
def testNoLaunchpadURL(self):
policy = MirroredBranchPolicy()
self.assertRaises(
- BadUrlLaunchpad, policy.checkOneURL,
+ BadUrlLaunchpad, policy.check_one_url,
self.factory.getUniqueURL(host='bazaar.launchpad.test'))
def testNoHTTPSLaunchpadURL(self):
policy = MirroredBranchPolicy()
self.assertRaises(
- BadUrlLaunchpad, policy.checkOneURL,
+ BadUrlLaunchpad, policy.check_one_url,
self.factory.getUniqueURL(
host='bazaar.launchpad.test', scheme='https'))
def testNoOtherHostLaunchpadURL(self):
policy = MirroredBranchPolicy()
self.assertRaises(
- BadUrlLaunchpad, policy.checkOneURL,
+ BadUrlLaunchpad, policy.check_one_url,
self.factory.getUniqueURL(host='code.launchpad.test'))
def testLocalhost(self):
@@ -389,9 +389,9 @@
'codehosting', blacklisted_hostnames='localhost,127.0.0.1')
policy = MirroredBranchPolicy()
localhost_url = self.factory.getUniqueURL(host='localhost')
- self.assertRaises(BadUrl, policy.checkOneURL, localhost_url)
+ self.assertRaises(BadUrl, policy.check_one_url, localhost_url)
localhost_url = self.factory.getUniqueURL(host='127.0.0.1')
- self.assertRaises(BadUrl, policy.checkOneURL, localhost_url)
+ self.assertRaises(BadUrl, policy.check_one_url, localhost_url)
def test_no_stacked_on_url(self):
# By default, a MirroredBranchPolicy does not stack branches.
@@ -495,7 +495,7 @@
def setUp(self):
super(TestWorkerProgressReporting, self).setUp()
- SafeBranchOpener.install_hook()
+ BranchOpener.install_hook()
self.saved_factory = bzrlib.ui.ui_factory
self.disable_directory_isolation()
self.addCleanup(setattr, bzrlib.ui, 'ui_factory', self.saved_factory)
=== modified file 'lib/lp/codehosting/puller/tests/test_worker_formats.py'
--- lib/lp/codehosting/puller/tests/test_worker_formats.py 2015-09-28 17:38:45 +0000
+++ lib/lp/codehosting/puller/tests/test_worker_formats.py 2019-06-13 10:09:50 +0000
@@ -15,10 +15,10 @@
from bzrlib.repofmt.knitpack_repo import RepositoryFormatKnitPack5
from bzrlib.repofmt.knitrepo import RepositoryFormatKnit1
from bzrlib.tests.per_repository import TestCaseWithRepository
+from bzrlib.url_policy_open import BranchOpener
import lp.codehosting # For bzr plugins.
from lp.codehosting.puller.tests import PullerWorkerMixin
-from lp.codehosting.safe_open import SafeBranchOpener
from lp.codehosting.tests.helpers import LoomTestMixin
@@ -29,7 +29,7 @@
TestCaseWithRepository.setUp(self)
# make_bzrdir relies on this being a relative filesystem path.
self._source_branch_path = 'source-branch'
- SafeBranchOpener.install_hook()
+ BranchOpener.install_hook()
self.worker = self.makePullerWorker(
self.get_url(self._source_branch_path),
self.get_url('dest-path'))
=== modified file 'lib/lp/codehosting/puller/worker.py'
--- lib/lp/codehosting/puller/worker.py 2012-06-29 08:40:05 +0000
+++ lib/lp/codehosting/puller/worker.py 2019-06-13 10:09:50 +0000
@@ -27,6 +27,14 @@
from bzrlib.transport import get_transport
import bzrlib.ui
from bzrlib.ui import SilentUIFactory
+from bzrlib.url_policy_open import (
+ BadUrl,
+ BranchLoopError,
+ BranchOpenPolicy,
+ BranchReferenceForbidden,
+ BranchOpener,
+ )
+
from lazr.uri import (
InvalidURIError,
URI,
@@ -39,13 +47,6 @@
from lp.code.enums import BranchType
from lp.codehosting.bzrutils import identical_formats
from lp.codehosting.puller import get_lock_id_for_branch_id
-from lp.codehosting.safe_open import (
- BadUrl,
- BranchLoopError,
- BranchOpenPolicy,
- BranchReferenceForbidden,
- SafeBranchOpener,
- )
from lp.services.config import config
from lp.services.webapp import errorlog
@@ -75,7 +76,7 @@
"""Found a URL with an untrusted scheme."""
def __init__(self, scheme, url):
- BadUrl.__init__(self, scheme, url)
+ BadUrl.__init__(self, url)
self.scheme = scheme
@@ -195,7 +196,7 @@
"""
self.policy = policy
self.protocol = protocol
- self.opener = SafeBranchOpener(policy)
+ self.opener = BranchOpener(policy)
if log is not None:
self.log = log
else:
@@ -213,7 +214,7 @@
URL must point to a writable location.
:return: The destination branch.
"""
- return self.opener.runWithTransformFallbackLocationHookInstalled(
+ return self.opener.run_with_transform_fallback_location_hook_installed(
self.policy.createDestinationBranch, source_branch,
destination_url)
@@ -552,8 +553,8 @@
return None
return self.stacked_on_url
- def shouldFollowReferences(self):
- """See `BranchOpenPolicy.shouldFollowReferences`.
+ def should_follow_references(self):
+ """See `BranchOpenPolicy.should_follow_references`.
We traverse branch references for MIRRORED branches because they
provide a useful redirection mechanism and we want to be consistent
@@ -561,16 +562,16 @@
"""
return True
- def transformFallbackLocation(self, branch, url):
- """See `BranchOpenPolicy.transformFallbackLocation`.
+ def transform_fallback_location(self, branch, url):
+ """See `BranchOpenPolicy.transform_fallback_location`.
For mirrored branches, we stack on whatever the remote branch claims
to stack on, but this URL still needs to be checked.
"""
return urlutils.join(branch.base, url), True
- def checkOneURL(self, url):
- """See `BranchOpenPolicy.checkOneURL`.
+ def check_one_url(self, url):
+ """See `BranchOpenPolicy.check_one_url`.
We refuse to mirror from Launchpad or a ssh-like or file URL.
"""
@@ -619,23 +620,23 @@
break
return Branch.open_from_transport(dest_transport)
- def shouldFollowReferences(self):
- """See `BranchOpenerPolicy.shouldFollowReferences`.
+ def should_follow_references(self):
+ """See `BranchOpenerPolicy.should_follow_references`.
We do not traverse references for IMPORTED branches because the
code-import system should never produce branch references.
"""
return False
- def transformFallbackLocation(self, branch, url):
- """See `BranchOpenerPolicy.transformFallbackLocation`.
+ def transform_fallback_location(self, branch, url):
+ """See `BranchOpenerPolicy.transform_fallback_location`.
Import branches should not be stacked, ever.
"""
raise AssertionError("Import branch unexpectedly stacked!")
- def checkOneURL(self, url):
- """See `BranchOpenerPolicy.checkOneURL`.
+ def check_one_url(self, url):
+ """See `BranchOpenerPolicy.check_one_url`.
If the URL we are mirroring from does not start how we expect the pull
URLs of import branches to start, something has gone badly wrong, so
=== removed file 'lib/lp/codehosting/safe_open.py'
--- lib/lp/codehosting/safe_open.py 2012-08-21 17:34:42 +0000
+++ lib/lp/codehosting/safe_open.py 1970-01-01 00:00:00 +0000
@@ -1,354 +0,0 @@
-# Copyright 2009-2011 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Safe branch opening."""
-
-__metaclass__ = type
-
-import threading
-
-from bzrlib import (
- errors,
- trace,
- urlutils,
- )
-from bzrlib.branch import Branch
-from bzrlib.bzrdir import (
- BzrProber,
- RemoteBzrProber,
- )
-from bzrlib.transport import (
- do_catching_redirections,
- get_transport,
- )
-from lazr.uri import URI
-
-
-__all__ = [
- 'AcceptAnythingPolicy',
- 'BadUrl',
- 'BlacklistPolicy',
- 'BranchLoopError',
- 'BranchOpenPolicy',
- 'BranchReferenceForbidden',
- 'SafeBranchOpener',
- 'WhitelistPolicy',
- 'safe_open',
- ]
-
-
-# TODO JelmerVernooij 2011-08-06: This module is generic enough to be
-# in bzrlib, and may be of use to others. bug=850843
-
-# These are the default probers that SafeBranchOpener will try to use,
-# unless a different set was specified.
-
-DEFAULT_PROBERS = [BzrProber, RemoteBzrProber]
-
-
-class BadUrl(Exception):
- """Tried to access a branch from a bad URL."""
-
-
-class BranchReferenceForbidden(Exception):
- """Trying to mirror a branch reference and the branch type does not allow
- references.
- """
-
-
-class BranchLoopError(Exception):
- """Encountered a branch cycle.
-
- A URL may point to a branch reference or it may point to a stacked branch.
- In either case, it's possible for there to be a cycle in these references,
- and this exception is raised when we detect such a cycle.
- """
-
-
-class BranchOpenPolicy:
- """Policy on how to open branches.
-
- In particular, a policy determines which branches are safe to open by
- checking their URLs and deciding whether or not to follow branch
- references.
- """
-
- def shouldFollowReferences(self):
- """Whether we traverse references when mirroring.
-
- Subclasses must override this method.
-
- If we encounter a branch reference and this returns false, an error is
- raised.
-
- :returns: A boolean to indicate whether to follow a branch reference.
- """
- raise NotImplementedError(self.shouldFollowReferences)
-
- def transformFallbackLocation(self, branch, url):
- """Validate, maybe modify, 'url' to be used as a stacked-on location.
-
- :param branch: The branch that is being opened.
- :param url: The URL that the branch provides for its stacked-on
- location.
- :return: (new_url, check) where 'new_url' is the URL of the branch to
- actually open and 'check' is true if 'new_url' needs to be
- validated by checkAndFollowBranchReference.
- """
- raise NotImplementedError(self.transformFallbackLocation)
-
- def checkOneURL(self, url):
- """Check the safety of the source URL.
-
- Subclasses must override this method.
-
- :param url: The source URL to check.
- :raise BadUrl: subclasses are expected to raise this or a subclass
- when it finds a URL it deems to be unsafe.
- """
- raise NotImplementedError(self.checkOneURL)
-
-
-class BlacklistPolicy(BranchOpenPolicy):
- """Branch policy that forbids certain URLs."""
-
- def __init__(self, should_follow_references, unsafe_urls=None):
- if unsafe_urls is None:
- unsafe_urls = set()
- self._unsafe_urls = unsafe_urls
- self._should_follow_references = should_follow_references
-
- def shouldFollowReferences(self):
- return self._should_follow_references
-
- def checkOneURL(self, url):
- if url in self._unsafe_urls:
- raise BadUrl(url)
-
- def transformFallbackLocation(self, branch, url):
- """See `BranchOpenPolicy.transformFallbackLocation`.
-
- This class is not used for testing our smarter stacking features so we
- just do the simplest thing: return the URL that would be used anyway
- and don't check it.
- """
- return urlutils.join(branch.base, url), False
-
-
-class AcceptAnythingPolicy(BlacklistPolicy):
- """Accept anything, to make testing easier."""
-
- def __init__(self):
- super(AcceptAnythingPolicy, self).__init__(True, set())
-
-
-class WhitelistPolicy(BranchOpenPolicy):
- """Branch policy that only allows certain URLs."""
-
- def __init__(self, should_follow_references, allowed_urls=None,
- check=False):
- if allowed_urls is None:
- allowed_urls = []
- self.allowed_urls = set(url.rstrip('/') for url in allowed_urls)
- self.check = check
-
- def shouldFollowReferences(self):
- return self._should_follow_references
-
- def checkOneURL(self, url):
- if url.rstrip('/') not in self.allowed_urls:
- raise BadUrl(url)
-
- def transformFallbackLocation(self, branch, url):
- """See `BranchOpenPolicy.transformFallbackLocation`.
-
- Here we return the URL that would be used anyway and optionally check
- it.
- """
- return urlutils.join(branch.base, url), self.check
-
-
-class SingleSchemePolicy(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)
-
-
-class SafeBranchOpener(object):
- """Safe branch opener.
-
- All locations that are opened (stacked-on branches, references) are
- checked against a policy object.
-
- The policy object is expected to have the following methods:
- * checkOneURL
- * shouldFollowReferences
- * transformFallbackLocation
- """
-
- _threading_data = threading.local()
-
- def __init__(self, policy, probers=None):
- """Create a new SafeBranchOpener.
-
- :param policy: The opener policy to use.
- :param probers: Optional list of probers to allow.
- Defaults to local and remote bzr probers.
- """
- self.policy = policy
- self._seen_urls = set()
- if probers is None:
- self.probers = list(DEFAULT_PROBERS)
- else:
- self.probers = probers
-
- @classmethod
- def install_hook(cls):
- """Install the ``transformFallbackLocation`` hook.
-
- This is done at module import time, but transformFallbackLocationHook
- doesn't do anything unless the `_active_openers` threading.Local
- object has a '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 `SafeBranchOpener`
- as bzrlib.tests.TestCase.setUp clears hooks.
- """
- Branch.hooks.install_named_hook(
- 'transform_fallback_location',
- cls.transformFallbackLocationHook,
- 'SafeBranchOpener.transformFallbackLocationHook')
-
- def checkAndFollowBranchReference(self, url):
- """Check URL (and possibly the referenced URL) for safety.
-
- This method checks that `url` passes the policy's `checkOneURL`
- method, and if `url` refers to a branch reference, it checks whether
- references are allowed and whether the reference's URL passes muster
- also -- recursively, until a real branch is found.
-
- :param url: URL to check
- :raise BranchLoopError: If the branch references form a loop.
- :raise BranchReferenceForbidden: If this opener forbids branch
- references.
- """
- while True:
- if url in self._seen_urls:
- raise BranchLoopError()
- self._seen_urls.add(url)
- self.policy.checkOneURL(url)
- next_url = self.followReference(url)
- if next_url is None:
- return url
- url = next_url
- if not self.policy.shouldFollowReferences():
- raise BranchReferenceForbidden(url)
-
- @classmethod
- def transformFallbackLocationHook(cls, branch, url):
- """Installed as the 'transform_fallback_location' Branch hook.
-
- This method calls `transformFallbackLocation` on the policy object and
- either returns the url it provides or passes it back to
- checkAndFollowBranchReference.
- """
- try:
- opener = getattr(cls._threading_data, "opener")
- except AttributeError:
- return url
- new_url, check = opener.policy.transformFallbackLocation(branch, url)
- if check:
- return opener.checkAndFollowBranchReference(new_url)
- else:
- return new_url
-
- def runWithTransformFallbackLocationHookInstalled(
- self, callable, *args, **kw):
- assert (self.transformFallbackLocationHook in
- Branch.hooks['transform_fallback_location'])
- self._threading_data.opener = self
- try:
- return callable(*args, **kw)
- finally:
- del self._threading_data.opener
- # We reset _seen_urls here to avoid multiple calls to open giving
- # spurious loop exceptions.
- self._seen_urls = set()
-
- def followReference(self, url):
- """Get the branch-reference value at the specified url.
-
- This exists as a separate method only to be overridden in unit tests.
- """
- bzrdir = self._open_dir(url)
- return bzrdir.get_branch_reference()
-
- def _open_dir(self, url):
- """Simple BzrDir.open clone that only uses specific probers.
-
- :param url: URL to open
- :return: ControlDir instance
- """
- def redirected(transport, e, redirection_notice):
- self.policy.checkOneURL(e.target)
- redirected_transport = transport._redirected_to(
- e.source, e.target)
- if redirected_transport is None:
- raise errors.NotBranchError(e.source)
- trace.note('%s is%s redirected to %s',
- transport.base, e.permanently, redirected_transport.base)
- return redirected_transport
-
- def find_format(transport):
- last_error = errors.NotBranchError(transport.base)
- for prober_kls in self.probers:
- prober = prober_kls()
- try:
- return transport, prober.probe_transport(transport)
- except errors.NotBranchError as e:
- last_error = e
- else:
- raise last_error
- transport = get_transport(url)
- transport, format = do_catching_redirections(find_format, transport,
- redirected)
- return format.open(transport)
-
- def open(self, url, ignore_fallbacks=False):
- """Open the Bazaar branch at url, first checking for safety.
-
- What safety means is defined by a subclasses `followReference` and
- `checkOneURL` methods.
- """
- url = self.checkAndFollowBranchReference(url)
-
- def open_branch(url, ignore_fallbacks):
- dir = self._open_dir(url)
- return dir.open_branch(ignore_fallbacks=ignore_fallbacks)
- return self.runWithTransformFallbackLocationHookInstalled(
- open_branch, url, ignore_fallbacks)
-
-
-def safe_open(allowed_scheme, url, ignore_fallbacks=False):
- """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(SingleSchemePolicy(allowed_scheme)).open(url,
- ignore_fallbacks=ignore_fallbacks)
-
-
-SafeBranchOpener.install_hook()
=== modified file 'lib/lp/codehosting/scanner/tests/test_bzrsync.py'
--- lib/lp/codehosting/scanner/tests/test_bzrsync.py 2018-01-02 16:10:26 +0000
+++ lib/lp/codehosting/scanner/tests/test_bzrsync.py 2019-06-13 10:09:50 +0000
@@ -14,6 +14,7 @@
)
from bzrlib.tests import TestCaseWithTransport
from bzrlib.uncommit import uncommit
+from bzrlib.url_policy_open import BranchOpener
from fixtures import TempDir
import pytz
from storm.locals import Store
@@ -45,7 +46,6 @@
read_locked,
write_locked,
)
-from lp.codehosting.safe_open import SafeBranchOpener
from lp.codehosting.scanner.bzrsync import BzrSync
from lp.services.config import config
from lp.services.database.interfaces import IStore
@@ -86,7 +86,7 @@
def setUp(self):
super(BzrSyncTestCase, self).setUp()
- SafeBranchOpener.install_hook()
+ BranchOpener.install_hook()
self.disable_directory_isolation()
self.useBzrBranches(direct_database=True)
self.makeFixtures()
=== removed file 'lib/lp/codehosting/tests/test_safe_open.py'
--- lib/lp/codehosting/tests/test_safe_open.py 2018-01-02 16:10:26 +0000
+++ lib/lp/codehosting/tests/test_safe_open.py 1970-01-01 00:00:00 +0000
@@ -1,378 +0,0 @@
-# Copyright 2009-2011 Canonical Ltd. This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""Tests for the safe branch open code."""
-
-
-__metaclass__ = type
-
-from bzrlib.branch import (
- Branch,
- BranchReferenceFormat,
- BzrBranchFormat7,
- )
-from bzrlib.bzrdir import (
- BzrDir,
- BzrDirMetaFormat1,
- BzrProber,
- )
-from bzrlib.controldir import ControlDirFormat
-from bzrlib.errors import NotBranchError
-from bzrlib.repofmt.knitpack_repo import RepositoryFormatKnitPack1
-from bzrlib.tests import TestCaseWithTransport
-from bzrlib.transport import chroot
-from lazr.uri import URI
-
-from lp.codehosting.safe_open import (
- BadUrl,
- BlacklistPolicy,
- BranchLoopError,
- BranchReferenceForbidden,
- safe_open,
- SafeBranchOpener,
- WhitelistPolicy,
- )
-from lp.codehosting.tests.helpers import force_stacked_on_url
-from lp.testing import TestCase
-
-
-class TestSafeBranchOpenerCheckAndFollowBranchReference(TestCase):
- """Unit tests for `SafeBranchOpener.checkAndFollowBranchReference`."""
-
- def setUp(self):
- super(TestSafeBranchOpenerCheckAndFollowBranchReference, self).setUp()
- SafeBranchOpener.install_hook()
-
- class StubbedSafeBranchOpener(SafeBranchOpener):
- """SafeBranchOpener that provides canned answers.
-
- We implement the methods we need to to be able to control all the
- inputs to the `BranchMirrorer.checkSource` method, which is what is
- being tested in this class.
- """
-
- def __init__(self, references, policy):
- parent_cls = TestSafeBranchOpenerCheckAndFollowBranchReference
- super(parent_cls.StubbedSafeBranchOpener, self).__init__(policy)
- self._reference_values = {}
- for i in range(len(references) - 1):
- self._reference_values[references[i]] = references[i + 1]
- self.follow_reference_calls = []
-
- def followReference(self, url):
- self.follow_reference_calls.append(url)
- return self._reference_values[url]
-
- def makeBranchOpener(self, should_follow_references, references,
- unsafe_urls=None):
- policy = BlacklistPolicy(should_follow_references, unsafe_urls)
- opener = self.StubbedSafeBranchOpener(references, policy)
- return opener
-
- def testCheckInitialURL(self):
- # checkSource rejects all URLs that are not allowed.
- opener = self.makeBranchOpener(None, [], set(['a']))
- self.assertRaises(
- BadUrl, opener.checkAndFollowBranchReference, 'a')
-
- def testNotReference(self):
- # When branch references are forbidden, checkAndFollowBranchReference
- # does not raise on non-references.
- opener = self.makeBranchOpener(False, ['a', None])
- self.assertEqual('a', opener.checkAndFollowBranchReference('a'))
- self.assertEqual(['a'], opener.follow_reference_calls)
-
- def testBranchReferenceForbidden(self):
- # checkAndFollowBranchReference raises BranchReferenceForbidden if
- # branch references are forbidden and the source URL points to a
- # branch reference.
- opener = self.makeBranchOpener(False, ['a', 'b'])
- self.assertRaises(
- BranchReferenceForbidden,
- opener.checkAndFollowBranchReference, 'a')
- self.assertEqual(['a'], opener.follow_reference_calls)
-
- def testAllowedReference(self):
- # checkAndFollowBranchReference does not raise if following references
- # is allowed and the source URL points to a branch reference to a
- # permitted location.
- opener = self.makeBranchOpener(True, ['a', 'b', None])
- self.assertEqual('b', opener.checkAndFollowBranchReference('a'))
- self.assertEqual(['a', 'b'], opener.follow_reference_calls)
-
- def testCheckReferencedURLs(self):
- # checkAndFollowBranchReference checks if the URL a reference points
- # to is safe.
- opener = self.makeBranchOpener(
- True, ['a', 'b', None], unsafe_urls=set('b'))
- self.assertRaises(
- BadUrl, opener.checkAndFollowBranchReference, 'a')
- self.assertEqual(['a'], opener.follow_reference_calls)
-
- def testSelfReferencingBranch(self):
- # checkAndFollowBranchReference raises BranchReferenceLoopError if
- # following references is allowed and the source url points to a
- # self-referencing branch reference.
- opener = self.makeBranchOpener(True, ['a', 'a'])
- self.assertRaises(
- BranchLoopError, opener.checkAndFollowBranchReference, 'a')
- self.assertEqual(['a'], opener.follow_reference_calls)
-
- def testBranchReferenceLoop(self):
- # checkAndFollowBranchReference raises BranchReferenceLoopError if
- # following references is allowed and the source url points to a loop
- # of branch references.
- references = ['a', 'b', 'a']
- opener = self.makeBranchOpener(True, references)
- self.assertRaises(
- BranchLoopError, opener.checkAndFollowBranchReference, 'a')
- self.assertEqual(['a', 'b'], opener.follow_reference_calls)
-
-
-class TrackingProber(BzrProber):
- """Subclass of BzrProber which tracks URLs it has been asked to open."""
-
- seen_urls = []
-
- @classmethod
- def probe_transport(klass, transport):
- klass.seen_urls.append(transport.base)
- return BzrProber.probe_transport(transport)
-
-
-class TestSafeBranchOpenerStacking(TestCaseWithTransport):
-
- def setUp(self):
- super(TestSafeBranchOpenerStacking, self).setUp()
- SafeBranchOpener.install_hook()
-
- def makeBranchOpener(self, allowed_urls, probers=None):
- policy = WhitelistPolicy(True, allowed_urls, True)
- return SafeBranchOpener(policy, probers)
-
- def makeBranch(self, path, branch_format, repository_format):
- """Make a Bazaar branch at 'path' with the given formats."""
- bzrdir_format = BzrDirMetaFormat1()
- bzrdir_format.set_branch_format(branch_format)
- bzrdir = self.make_bzrdir(path, format=bzrdir_format)
- repository_format.initialize(bzrdir)
- return bzrdir.create_branch()
-
- def testProbers(self):
- # Only the specified probers should be used
- b = self.make_branch('branch')
- opener = self.makeBranchOpener([b.base], probers=[])
- self.assertRaises(NotBranchError, opener.open, b.base)
- opener = self.makeBranchOpener([b.base], probers=[BzrProber])
- self.assertEqual(b.base, opener.open(b.base).base)
-
- def testDefaultProbers(self):
- # If no probers are specified to the constructor
- # of SafeBranchOpener, then a safe set will be used,
- # rather than all probers registered in bzr.
- self.addCleanup(ControlDirFormat.unregister_prober, TrackingProber)
- ControlDirFormat.register_prober(TrackingProber)
- # Open a location without any branches, so that all probers are
- # tried.
- # First, check that the TrackingProber tracks correctly.
- TrackingProber.seen_urls = []
- opener = self.makeBranchOpener(["."], probers=[TrackingProber])
- self.assertRaises(NotBranchError, opener.open, ".")
- self.assertEqual(1, len(TrackingProber.seen_urls))
- TrackingProber.seen_urls = []
- # And make sure it's registered in such a way that BzrDir.open would
- # use it.
- self.assertRaises(NotBranchError, BzrDir.open, ".")
- self.assertEqual(1, len(TrackingProber.seen_urls))
- TrackingProber.seen_urls = []
- # Make sure that SafeBranchOpener doesn't use it if no
- # probers were specified
- opener = self.makeBranchOpener(["."])
- self.assertRaises(NotBranchError, opener.open, ".")
- self.assertEqual(0, len(TrackingProber.seen_urls))
-
- def testAllowedURL(self):
- # checkSource does not raise an exception for branches stacked on
- # branches with allowed URLs.
- stacked_on_branch = self.make_branch('base-branch')
- stacked_branch = self.make_branch('stacked-branch')
- stacked_branch.set_stacked_on_url(stacked_on_branch.base)
- opener = self.makeBranchOpener(
- [stacked_branch.base, stacked_on_branch.base])
- # This doesn't raise an exception.
- opener.open(stacked_branch.base)
-
- def testUnstackableRepository(self):
- # checkSource treats branches with UnstackableRepositoryFormats as
- # being not stacked.
- branch = self.makeBranch(
- 'unstacked', BzrBranchFormat7(), RepositoryFormatKnitPack1())
- opener = self.makeBranchOpener([branch.base])
- # This doesn't raise an exception.
- opener.open(branch.base)
-
- def testAllowedRelativeURL(self):
- # checkSource passes on absolute urls to checkOneURL, even if the
- # value of stacked_on_location in the config is set to a relative URL.
- stacked_on_branch = self.make_branch('base-branch')
- stacked_branch = self.make_branch('stacked-branch')
- stacked_branch.set_stacked_on_url('../base-branch')
- opener = self.makeBranchOpener(
- [stacked_branch.base, stacked_on_branch.base])
- # Note that stacked_on_branch.base is not '../base-branch', it's an
- # absolute URL.
- self.assertNotEqual('../base-branch', stacked_on_branch.base)
- # This doesn't raise an exception.
- opener.open(stacked_branch.base)
-
- def testAllowedRelativeNested(self):
- # Relative URLs are resolved relative to the stacked branch.
- self.get_transport().mkdir('subdir')
- a = self.make_branch('subdir/a')
- b = self.make_branch('b')
- b.set_stacked_on_url('../subdir/a')
- c = self.make_branch('subdir/c')
- c.set_stacked_on_url('../../b')
- opener = self.makeBranchOpener([c.base, b.base, a.base])
- # This doesn't raise an exception.
- opener.open(c.base)
-
- def testForbiddenURL(self):
- # checkSource raises a BadUrl exception if a branch is stacked on a
- # branch with a forbidden URL.
- stacked_on_branch = self.make_branch('base-branch')
- stacked_branch = self.make_branch('stacked-branch')
- stacked_branch.set_stacked_on_url(stacked_on_branch.base)
- opener = self.makeBranchOpener([stacked_branch.base])
- self.assertRaises(BadUrl, opener.open, stacked_branch.base)
-
- def testForbiddenURLNested(self):
- # checkSource raises a BadUrl exception if a branch is stacked on a
- # branch that is in turn stacked on a branch with a forbidden URL.
- a = self.make_branch('a')
- b = self.make_branch('b')
- b.set_stacked_on_url(a.base)
- c = self.make_branch('c')
- c.set_stacked_on_url(b.base)
- opener = self.makeBranchOpener([c.base, b.base])
- self.assertRaises(BadUrl, opener.open, c.base)
-
- def testSelfStackedBranch(self):
- # checkSource raises StackingLoopError if a branch is stacked on
- # itself. This avoids infinite recursion errors.
- a = self.make_branch('a')
- force_stacked_on_url(a, a.base)
- opener = self.makeBranchOpener([a.base])
- self.assertRaises(BranchLoopError, opener.open, a.base)
-
- def testLoopStackedBranch(self):
- # checkSource raises StackingLoopError if a branch is stacked in such
- # a way so that it is ultimately stacked on itself. e.g. a stacked on
- # b stacked on a.
- a = self.make_branch('a')
- b = self.make_branch('b')
- a.set_stacked_on_url(b.base)
- b.set_stacked_on_url(a.base)
- opener = self.makeBranchOpener([a.base, b.base])
- self.assertRaises(BranchLoopError, opener.open, a.base)
- self.assertRaises(BranchLoopError, opener.open, b.base)
-
- def testCustomOpener(self):
- # A custom function for opening a control dir can be specified.
- a = self.make_branch('a')
- b = self.make_branch('b')
- b.set_stacked_on_url(a.base)
-
- TrackingProber.seen_urls = []
- opener = self.makeBranchOpener(
- [a.base, b.base], probers=[TrackingProber])
- opener.open(b.base)
- self.assertEqual(set(TrackingProber.seen_urls), set([b.base, a.base]))
-
- def testCustomOpenerWithBranchReference(self):
- # A custom function for opening a control dir can be specified.
- a = self.make_branch('a')
- b_dir = self.make_bzrdir('b')
- b = BranchReferenceFormat().initialize(b_dir, target_branch=a)
- TrackingProber.seen_urls = []
- opener = self.makeBranchOpener(
- [a.base, b.base], probers=[TrackingProber])
- opener.open(b.base)
- self.assertEqual(set(TrackingProber.seen_urls), set([b.base, a.base]))
-
- def test_ignore_fallbacks(self):
- """"Cross-format stacking doesn't error with ignore_fallbacks."""
- stacked, stacked_on = make_cross_format_stacked(self)
- opener = self.makeBranchOpener([stacked.base, stacked_on.base])
- opener.open(stacked.base, ignore_fallbacks=True)
-
-
-def make_cross_format_stacked(test_case):
- test_case.get_transport().mkdir('inside')
- stacked = test_case.make_branch('inside/stacked', format='1.6')
- stacked_on = test_case.make_branch('inside/stacked-on', format='2a')
- force_stacked_on_url(stacked, stacked_on.base)
- return stacked, stacked_on
-
-
-class TestSafeOpen(TestCaseWithTransport):
- """Tests for `safe_open`."""
-
- def setUp(self):
- super(TestSafeOpen, self).setUp()
- SafeBranchOpener.install_hook()
-
- def test_hook_does_not_interfere(self):
- # The transform_fallback_location hook does not interfere with regular
- # stacked branch access outside of safe_open.
- self.make_branch('stacked')
- self.make_branch('stacked-on')
- Branch.open('stacked').set_stacked_on_url('../stacked-on')
- Branch.open('stacked')
-
- 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(
- BadUrl, safe_open, scheme, get_chrooted_url('stacked'))
-
- def test_ignore_fallbacks(self):
- """"Cross-format stacking doesn't error with ignore_fallbacks."""
- scheme, get_chrooted_url = self.get_chrooted_scheme('inside')
- stacked, stacked_on = make_cross_format_stacked(self)
- force_stacked_on_url(stacked, get_chrooted_url('stacked-on'))
- safe_open(scheme, get_chrooted_url('stacked'), ignore_fallbacks=True)
=== modified file 'lib/lp/codehosting/upgrade.py'
--- lib/lp/codehosting/upgrade.py 2014-01-30 15:04:06 +0000
+++ lib/lp/codehosting/upgrade.py 2019-06-13 10:09:50 +0000
@@ -29,6 +29,10 @@
)
from bzrlib.repofmt.groupcompress_repo import RepositoryFormat2aSubtree
from bzrlib.upgrade import upgrade
+from bzrlib.url_policy_open import (
+ BranchOpener,
+ SingleSchemePolicy,
+ )
from lp.code.bzr import (
branch_changed,
@@ -36,7 +40,6 @@
)
from lp.code.model.branch import Branch
from lp.codehosting.bzrutils import read_locked
-from lp.codehosting.safe_open import safe_open
from lp.codehosting.vfs.branchfs import get_real_branch_path
from lp.services.database.interfaces import IStore
@@ -52,9 +55,9 @@
self.branch = branch
self.bzr_branch = bzr_branch
if self.bzr_branch is None:
- self.bzr_branch = safe_open('lp-internal',
- self.branch.getInternalBzrUrl(),
- ignore_fallbacks=True)
+ opener = BranchOpener(SingleSchemePolicy('lp-internal'))
+ self.bzr_branch = opener.open(
+ self.branch.getInternalBzrUrl(), ignore_fallbacks=True)
self.target_dir = target_dir
self.target_subdir = os.path.join(
self.target_dir, str(self.branch.id))
=== modified file 'scripts/code-import-worker.py'
--- scripts/code-import-worker.py 2016-11-03 17:31:43 +0000
+++ scripts/code-import-worker.py 2019-06-13 10:09:50 +0000
@@ -21,6 +21,7 @@
import sys
from bzrlib.transport import get_transport
+from bzrlib.url_policy_open import AcceptAnythingPolicy
from lp.codehosting.codeimport.worker import (
BzrImportWorker,
@@ -32,7 +33,6 @@
GitImportWorker,
GitToGitImportWorker,
)
-from lp.codehosting.safe_open import AcceptAnythingPolicy
from lp.services import scripts
from lp.services.config import config
from lp.services.timeout import set_default_timeout_function
Follow ups
