1
# Copyright (C) 2006 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Implementation of Transport that prevents access to locations above a set
20
from urlparse import urlparse
22
from bzrlib import errors, urlutils
23
from bzrlib.transport import (
31
from bzrlib.transport.decorator import TransportDecorator, DecoratorServer
32
from bzrlib.transport.memory import MemoryTransport
35
class ChrootServer(Server):
36
"""User space 'chroot' facility.
38
The server's get_url returns the url for a chroot transport mapped to the
39
backing transport. The url is of the form chroot-xxx:/// so parent
40
directories of the backing transport are not visible. The chroot url will
41
not allow '..' sequences to result in requests to the chroot affecting
42
directories outside the backing transport.
45
def __init__(self, backing_transport):
46
self.backing_transport = backing_transport
48
def _factory(self, url):
49
assert url.startswith(self.scheme)
50
return ChrootTransport(self, url)
56
self.scheme = 'chroot-%d:///' % id(self)
57
register_transport(self.scheme, self._factory)
60
unregister_transport(self.scheme, self._factory)
63
class ChrootTransport(Transport):
66
Please see ChrootServer for details.
69
def __init__(self, server, base):
71
if not base.endswith('/'):
73
Transport.__init__(self, base)
74
self.base_path = self.base[len(self.server.scheme)-1:]
75
self.scheme = self.server.scheme
77
def _call(self, methodname, relpath, *args):
78
method = getattr(self.server.backing_transport, methodname)
79
return method(self._safe_relpath(relpath), *args)
81
def _safe_relpath(self, relpath):
82
safe_relpath = self._combine_paths(self.base_path, relpath)
83
assert safe_relpath.startswith('/')
84
return safe_relpath[1:]
87
def abspath(self, relpath):
88
return self.scheme + self._safe_relpath(relpath)
90
def append_file(self, relpath, f, mode=None):
91
return self._call('append_file', relpath, f, mode)
93
def clone(self, relpath):
94
return ChrootTransport(self.server, self.abspath(relpath))
96
def delete(self, relpath):
97
return self._call('delete', relpath)
99
def delete_tree(self, relpath):
100
return self._call('delete_tree', relpath)
102
def get(self, relpath):
103
return self._call('get', relpath)
105
def has(self, relpath):
106
return self._call('has', relpath)
108
def iter_files_recursive(self):
109
backing_transport = self.server.backing_transport.clone(
110
self._safe_relpath('.'))
111
return backing_transport.iter_files_recursive()
114
return self.server.backing_transport.listable()
116
def list_dir(self, relpath):
117
return self._call('list_dir', relpath)
119
def lock_read(self, relpath):
120
return self._call('lock_read', relpath)
122
def lock_write(self, relpath):
123
return self._call('lock_write', relpath)
125
def mkdir(self, relpath, mode=None):
126
return self._call('mkdir', relpath, mode)
128
def put_file(self, relpath, f, mode=None):
129
return self._call('put_file', relpath, f, mode)
131
def rename(self, rel_from, rel_to):
132
return self._call('rename', rel_from, self._safe_relpath(rel_to))
134
def rmdir(self, relpath):
135
return self._call('rmdir', relpath)
137
def stat(self, relpath):
138
return self._call('stat', relpath)
141
class TestingChrootServer(ChrootServer):
144
"""TestingChrootServer is not usable until setUp is called."""
146
def setUp(self, backing_server=None):
147
"""Setup the Chroot on backing_server."""
148
if backing_server is not None:
149
self.backing_transport = get_transport(backing_server.get_url())
151
self.backing_transport = get_transport('.')
152
ChrootServer.setUp(self)
155
def get_test_permutations():
156
"""Return the permutations to be used in testing."""
157
return [(ChrootTransport, TestingChrootServer),
added
removed
bzrlib/transport/chroot.py
