# Copyright (C) 2006-2012, 2016 Canonical Ltd # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA """RemoteTransport client for the smart-server. This module shouldn't be accessed directly. The classes defined here should be imported from breezy.bzr.smart. """ __all__ = ['RemoteTransport', 'RemoteTCPTransport', 'RemoteSSHTransport'] from io import BytesIO from .. import ( config, debug, errors, trace, transport, urlutils, ) from ..bzr import ( remote, ) from ..bzr.smart import client, medium class _SmartStat(object): def __init__(self, size, mode): self.st_size = size self.st_mode = mode class RemoteTransport(transport.ConnectedTransport): """Connection to a smart server. The connection holds references to the medium that can be used to send requests to the server. The connection has a notion of the current directory to which it's connected; this is incorporated in filenames passed to the server. This supports some higher-level RPC operations and can also be treated like a Transport to do file-like operations. The connection can be made over a tcp socket, an ssh pipe or a series of http requests. There are concrete subclasses for each type: RemoteTCPTransport, etc. """ # When making a readv request, cap it at requesting 5MB of data _max_readv_bytes = 5 * 1024 * 1024 # IMPORTANT FOR IMPLEMENTORS: RemoteTransport MUST NOT be given encoding # responsibilities: Put those on SmartClient or similar. This is vital for # the ability to support multiple versions of the smart protocol over time: # RemoteTransport is an adapter from the Transport object model to the # SmartClient model, not an encoder. # FIXME: the medium parameter should be private, only the tests requires # it. It may be even clearer to define a TestRemoteTransport that handles # the specific cases of providing a _client and/or a _medium, and leave # RemoteTransport as an abstract class. def __init__(self, url, _from_transport=None, medium=None, _client=None): """Constructor. :param _from_transport: Another RemoteTransport instance that this one is being cloned from. Attributes such as the medium will be reused. :param medium: The medium to use for this RemoteTransport. If None, the medium from the _from_transport is shared. If both this and _from_transport are None, a new medium will be built. _from_transport and medium cannot both be specified. :param _client: Override the _SmartClient used by this transport. This should only be used for testing purposes; normally this is determined from the medium. """ super(RemoteTransport, self).__init__( url, _from_transport=_from_transport) # The medium is the connection, except when we need to share it with # other objects (RemoteBzrDir, RemoteRepository etc). In these cases # what we want to share is really the shared connection. if (_from_transport is not None and isinstance(_from_transport, RemoteTransport)): _client = _from_transport._client elif _from_transport is None: # If no _from_transport is specified, we need to intialize the # shared medium. credentials = None if medium is None: medium, credentials = self._build_medium() if 'hpss' in debug.debug_flags: trace.mutter('hpss: Built a new medium: %s', medium.__class__.__name__) self._shared_connection = transport._SharedConnection(medium, credentials, self.base) elif medium is None: # No medium was specified, so share the medium from the # _from_transport. medium = self._shared_connection.connection else: raise AssertionError( "Both _from_transport (%r) and medium (%r) passed to " "RemoteTransport.__init__, but these parameters are mutally " "exclusive." % (_from_transport, medium)) if _client is None: self._client = client._SmartClient(medium) else: self._client = _client def _build_medium(self): """Create the medium if _from_transport does not provide one. The medium is analogous to the connection for ConnectedTransport: it allows connection sharing. """ # No credentials return None, None def _report_activity(self, bytes, direction): """See Transport._report_activity. Does nothing; the smart medium will report activity triggered by a RemoteTransport. """ pass def is_readonly(self): """Smart server transport can do read/write file operations.""" try: resp = self._call2(b'Transport.is_readonly') except errors.UnknownSmartMethod: # XXX: nasty hack: servers before 0.16 don't have a # 'Transport.is_readonly' verb, so we do what clients before 0.16 # did: assume False. return False if resp == (b'yes', ): return True elif resp == (b'no', ): return False else: raise errors.UnexpectedSmartServerResponse(resp) def get_smart_client(self): return self._get_connection() def get_smart_medium(self): return self._get_connection() def _remote_path(self, relpath): """Returns the Unicode version of the absolute path for relpath.""" path = urlutils.URL._combine_paths(self._parsed_url.path, relpath) if not isinstance(path, bytes): path = path.encode() return path def _call(self, method, *args): resp = self._call2(method, *args) self._ensure_ok(resp) def _call2(self, method, *args): """Call a method on the remote server.""" try: return self._client.call(method, *args) except errors.ErrorFromSmartServer as err: # The first argument, if present, is always a path. if args: context = {'relpath': args[0].decode('utf-8')} else: context = {} self._translate_error(err, **context) def _call_with_body_bytes(self, method, args, body): """Call a method on the remote server with body bytes.""" try: return self._client.call_with_body_bytes(method, args, body) except errors.ErrorFromSmartServer as err: # The first argument, if present, is always a path. if args: context = {'relpath': args[0]} else: context = {} self._translate_error(err, **context) def has(self, relpath): """Indicate whether a remote file of the given name exists or not. :see: Transport.has() """ resp = self._call2(b'has', self._remote_path(relpath)) if resp == (b'yes', ): return True elif resp == (b'no', ): return False else: raise errors.UnexpectedSmartServerResponse(resp) def get(self, relpath): """Return file-like object reading the contents of a remote file. :see: Transport.get_bytes()/get_file() """ return BytesIO(self.get_bytes(relpath)) def get_bytes(self, relpath): remote = self._remote_path(relpath) try: resp, response_handler = self._client.call_expecting_body( b'get', remote) except errors.ErrorFromSmartServer as err: self._translate_error(err, relpath) if resp != (b'ok', ): response_handler.cancel_read_body() raise errors.UnexpectedSmartServerResponse(resp) return response_handler.read_body_bytes() def _serialise_optional_mode(self, mode): if mode is None: return b'' else: return ('%d' % mode).encode('ascii') def mkdir(self, relpath, mode=None): resp = self._call2(b'mkdir', self._remote_path(relpath), self._serialise_optional_mode(mode)) def open_write_stream(self, relpath, mode=None): """See Transport.open_write_stream.""" self.put_bytes(relpath, b"", mode) result = transport.AppendBasedFileStream(self, relpath) transport._file_streams[self.abspath(relpath)] = result return result def put_bytes(self, relpath, raw_bytes, mode=None): if not isinstance(raw_bytes, bytes): raise TypeError( 'raw_bytes must be bytes string, not %s' % type(raw_bytes)) resp = self._call_with_body_bytes( b'put', (self._remote_path(relpath), self._serialise_optional_mode(mode)), raw_bytes) self._ensure_ok(resp) return len(raw_bytes) def put_bytes_non_atomic(self, relpath, raw_bytes, mode=None, create_parent_dir=False, dir_mode=None): """See Transport.put_bytes_non_atomic.""" # FIXME: no encoding in the transport! create_parent_str = b'F' if create_parent_dir: create_parent_str = b'T' resp = self._call_with_body_bytes( b'put_non_atomic', (self._remote_path(relpath), self._serialise_optional_mode(mode), create_parent_str, self._serialise_optional_mode(dir_mode)), raw_bytes) self._ensure_ok(resp) def put_file(self, relpath, upload_file, mode=None): # its not ideal to seek back, but currently put_non_atomic_file depends # on transports not reading before failing - which is a faulty # assumption I think - RBC 20060915 pos = upload_file.tell() try: return self.put_bytes(relpath, upload_file.read(), mode) except: upload_file.seek(pos) raise def put_file_non_atomic(self, relpath, f, mode=None, create_parent_dir=False, dir_mode=None): return self.put_bytes_non_atomic(relpath, f.read(), mode=mode, create_parent_dir=create_parent_dir, dir_mode=dir_mode) def append_file(self, relpath, from_file, mode=None): return self.append_bytes(relpath, from_file.read(), mode) def append_bytes(self, relpath, bytes, mode=None): resp = self._call_with_body_bytes( b'append', (self._remote_path(relpath), self._serialise_optional_mode(mode)), bytes) if resp[0] == b'appended': return int(resp[1]) raise errors.UnexpectedSmartServerResponse(resp) def delete(self, relpath): resp = self._call2(b'delete', self._remote_path(relpath)) self._ensure_ok(resp) def external_url(self): """See breezy.transport.Transport.external_url.""" # the external path for RemoteTransports is the base return self.base def recommended_page_size(self): """Return the recommended page size for this transport.""" return 64 * 1024 def _readv(self, relpath, offsets): if not offsets: return offsets = list(offsets) sorted_offsets = sorted(offsets) coalesced = list(self._coalesce_offsets(sorted_offsets, limit=self._max_readv_combine, fudge_factor=self._bytes_to_read_before_seek, max_size=self._max_readv_bytes)) # now that we've coallesced things, avoid making enormous requests requests = [] cur_request = [] cur_len = 0 for c in coalesced: if c.length + cur_len > self._max_readv_bytes: requests.append(cur_request) cur_request = [c] cur_len = c.length continue cur_request.append(c) cur_len += c.length if cur_request: requests.append(cur_request) if 'hpss' in debug.debug_flags: trace.mutter('%s.readv %s offsets => %s coalesced' ' => %s requests (%s)', self.__class__.__name__, len(offsets), len(coalesced), len(requests), sum(map(len, requests))) # Cache the results, but only until they have been fulfilled data_map = {} # turn the list of offsets into a single stack to iterate offset_stack = iter(offsets) # using a list so it can be modified when passing down and coming back next_offset = [next(offset_stack)] for cur_request in requests: try: result = self._client.call_with_body_readv_array( (b'readv', self._remote_path(relpath),), [(c.start, c.length) for c in cur_request]) resp, response_handler = result except errors.ErrorFromSmartServer as err: self._translate_error(err, relpath) if resp[0] != b'readv': # This should raise an exception response_handler.cancel_read_body() raise errors.UnexpectedSmartServerResponse(resp) for res in self._handle_response(offset_stack, cur_request, response_handler, data_map, next_offset): yield res def _handle_response(self, offset_stack, coalesced, response_handler, data_map, next_offset): cur_offset_and_size = next_offset[0] # FIXME: this should know how many bytes are needed, for clarity. data = response_handler.read_body_bytes() data_offset = 0 for c_offset in coalesced: if len(data) < c_offset.length: raise errors.ShortReadvError(relpath, c_offset.start, c_offset.length, actual=len(data)) for suboffset, subsize in c_offset.ranges: key = (c_offset.start + suboffset, subsize) this_data = data[data_offset + suboffset: data_offset + suboffset + subsize] # Special case when the data is in-order, rather than packing # into a map and then back out again. Benchmarking shows that # this has 100% hit rate, but leave in the data_map work just # in case. # TODO: Could we get away with using buffer() to avoid the # memory copy? Callers would need to realize they may # not have a real string. if key == cur_offset_and_size: yield cur_offset_and_size[0], this_data try: cur_offset_and_size = next_offset[0] = next( offset_stack) except StopIteration: return else: data_map[key] = this_data data_offset += c_offset.length # Now that we've read some data, see if we can yield anything back while cur_offset_and_size in data_map: this_data = data_map.pop(cur_offset_and_size) yield cur_offset_and_size[0], this_data try: cur_offset_and_size = next_offset[0] = next(offset_stack) except StopIteration: return def rename(self, rel_from, rel_to): self._call(b'rename', self._remote_path(rel_from), self._remote_path(rel_to)) def move(self, rel_from, rel_to): self._call(b'move', self._remote_path(rel_from), self._remote_path(rel_to)) def rmdir(self, relpath): resp = self._call(b'rmdir', self._remote_path(relpath)) def _ensure_ok(self, resp): if resp[0] != b'ok': raise errors.UnexpectedSmartServerResponse(resp) def _translate_error(self, err, relpath=None): remote._translate_error(err, path=relpath) def disconnect(self): m = self.get_smart_medium() if m is not None: m.disconnect() def stat(self, relpath): resp = self._call2(b'stat', self._remote_path(relpath)) if resp[0] == b'stat': return _SmartStat(int(resp[1]), int(resp[2], 8)) raise errors.UnexpectedSmartServerResponse(resp) # def lock_read(self, relpath): # """Lock the given file for shared (read) access. # :return: A lock object, which should be passed to Transport.unlock() # """ # The old RemoteBranch ignore lock for reading, so we will # continue that tradition and return a bogus lock object. # class BogusLock(object): # def __init__(self, path): ## self.path = path # def unlock(self): # pass # return BogusLock(relpath) def listable(self): return True def list_dir(self, relpath): resp = self._call2(b'list_dir', self._remote_path(relpath)) if resp[0] == b'names': return [name.decode('utf-8') for name in resp[1:]] raise errors.UnexpectedSmartServerResponse(resp) def iter_files_recursive(self): resp = self._call2(b'iter_files_recursive', self._remote_path('')) if resp[0] == b'names': return [name.decode('utf-8') for name in resp[1:]] raise errors.UnexpectedSmartServerResponse(resp) class RemoteTCPTransport(RemoteTransport): """Connection to smart server over plain tcp. This is essentially just a factory to get 'RemoteTransport(url, SmartTCPClientMedium). """ def _build_medium(self): client_medium = medium.SmartTCPClientMedium( self._parsed_url.host, self._parsed_url.port, self.base) return client_medium, None class RemoteTCPTransportV2Only(RemoteTransport): """Connection to smart server over plain tcp with the client hard-coded to assume protocol v2 and remote server version <= 1.6. This should only be used for testing. """ def _build_medium(self): client_medium = medium.SmartTCPClientMedium( self._parsed_url.host, self._parsed_url.port, self.base) client_medium._protocol_version = 2 client_medium._remember_remote_is_before((1, 6)) return client_medium, None class RemoteSSHTransport(RemoteTransport): """Connection to smart server over SSH. This is essentially just a factory to get 'RemoteTransport(url, SmartSSHClientMedium). """ def _build_medium(self): location_config = config.LocationConfig(self.base) bzr_remote_path = location_config.get_bzr_remote_path() user = self._parsed_url.user if user is None: auth = config.AuthenticationConfig() user = auth.get_user('ssh', self._parsed_url.host, self._parsed_url.port) ssh_params = medium.SSHParams(self._parsed_url.host, self._parsed_url.port, user, self._parsed_url.password, bzr_remote_path) client_medium = medium.SmartSSHClientMedium(self.base, ssh_params) return client_medium, (user, self._parsed_url.password) class RemoteHTTPTransport(RemoteTransport): """Just a way to connect between a bzr+http:// url and http://. This connection operates slightly differently than the RemoteSSHTransport. It uses a plain http:// transport underneath, which defines what remote .bzr/smart URL we are connected to. From there, all paths that are sent are sent as relative paths, this way, the remote side can properly de-reference them, since it is likely doing rewrite rules to translate an HTTP path into a local path. """ def __init__(self, base, _from_transport=None, http_transport=None): if http_transport is None: # FIXME: the password may be lost here because it appears in the # url only for an intial construction (when the url came from the # command-line). http_url = base[len('bzr+'):] self._http_transport = transport.get_transport_from_url(http_url) else: self._http_transport = http_transport super(RemoteHTTPTransport, self).__init__( base, _from_transport=_from_transport) def _build_medium(self): # We let http_transport take care of the credentials return self._http_transport.get_smart_medium(), None def _remote_path(self, relpath): """After connecting, HTTP Transport only deals in relative URLs.""" # Adjust the relpath based on which URL this smart transport is # connected to. http_base = urlutils.normalize_url(self.get_smart_medium().base) url = urlutils.join(self.base[len('bzr+'):], relpath) url = urlutils.normalize_url(url) return urlutils.relative_url(http_base, url) def clone(self, relative_url): """Make a new RemoteHTTPTransport related to me. This is re-implemented rather than using the default RemoteTransport.clone() because we must be careful about the underlying http transport. Also, the cloned smart transport will POST to the same .bzr/smart location as this transport (although obviously the relative paths in the smart requests may be different). This is so that the server doesn't have to handle .bzr/smart requests at arbitrary places inside .bzr directories, just at the initial URL the user uses. """ if relative_url: abs_url = self.abspath(relative_url) else: abs_url = self.base return RemoteHTTPTransport(abs_url, _from_transport=self, http_transport=self._http_transport) def _redirected_to(self, source, target): """See transport._redirected_to""" redirected = self._http_transport._redirected_to(source, target) if (redirected is not None and isinstance(redirected, type(self._http_transport))): return RemoteHTTPTransport('bzr+' + redirected.external_url(), http_transport=redirected) else: # Either None or a transport for a different protocol return redirected class HintingSSHTransport(transport.Transport): """Simple transport that handles ssh:// and points out bzr+ssh:// and git+ssh://.""" # TODO(jelmer): Implement support for detecting whether the repository at the # other end is a git or bzr repository. def __init__(self, url): raise errors.UnsupportedProtocol( url, 'Use bzr+ssh for Bazaar operations over SSH, e.g. "bzr+%s". ' 'Use git+ssh for Git operations over SSH, e.g. "git+%s".' % (url, url)) def get_test_permutations(): """Return (transport, server) permutations for testing.""" # We may need a little more test framework support to construct an # appropriate RemoteTransport in the future. from ..tests import test_server return [(RemoteTCPTransport, test_server.SmartTCPServer_for_testing)]