From c580d9c173162b0c632a70f94e3c1ca422ab5cce Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Thu, 10 Nov 2016 08:55:19 +0100 Subject: [PATCH 01/11] version: Add version parsing and formating --- src/borg/version.py | 49 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 src/borg/version.py diff --git a/src/borg/version.py b/src/borg/version.py new file mode 100644 index 000000000..4eb0c77d5 --- /dev/null +++ b/src/borg/version.py @@ -0,0 +1,49 @@ +import re + + +def parse_version(version): + """ + simplistic parser for setuptools_scm versions + + supports final versions and alpha ('a'), beta ('b') and rc versions. It just discards commits since last tag + and git revision hash. + + Output is a version tuple containing integers. It ends with one or two elements that ensure that relational + operators yield correct relations for alpha, beta and rc versions too. For final versions the last element + is a -1, for prerelease versions the last two elements are a smaller negative number and the number of e.g. + the beta. + + Note, this sorts version 1.0 before 1.0.0. + + This version format is part of the remote protocol, don‘t change in breaking ways. + """ + + parts = version.split('+')[0].split('.') + if parts[-1].startswith('dev'): + del parts[-1] + version = [int(segment) for segment in parts[:-1]] + + prerelease = re.fullmatch('([0-9]+)(a|b|rc)([0-9]+)', parts[-1]) + if prerelease: + version_type = {'a': -4, 'b': -3, 'rc': -2}[prerelease.group(2)] + version += [int(prerelease.group(1)), version_type, int(prerelease.group(3))] + else: + version += [int(parts[-1]), -1] + + return tuple(version) + + +def format_version(version): + """a reverse for parse_version (obviously without the dropped information)""" + f = [] + it = iter(version) + while True: + part = next(it) + if part >= 0: + f += str(part) + elif part == -1: + break + else: + f[-1] = f[-1] + {-2: 'rc', -3: 'b', -4: 'a'}[part] + str(next(it)) + break + return '.'.join(f) From 0da913f8ed9a3e4feea91c6894e25aa93e61a4c2 Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Thu, 10 Nov 2016 09:44:00 +0100 Subject: [PATCH 02/11] remote: Decode method name as utf-8 instead of ascii for consistency. --- src/borg/remote.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index 6264241c2..d94354ec3 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -112,7 +112,7 @@ class RepositoryServer: # pragma: no cover self.repository.close() raise UnexpectedRPCDataFormatFromClient(__version__) type, msgid, method, args = unpacked - method = method.decode('ascii') + method = method.decode() try: if method not in self.rpc_methods: raise InvalidRPCMethod(method) From d25e9aa4f0ce9724e409e8ee4d525998d6cb5511 Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Thu, 10 Nov 2016 09:41:33 +0100 Subject: [PATCH 03/11] remote: Use single quotes --- src/borg/remote.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index d94354ec3..d326ce923 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -102,7 +102,7 @@ class RepositoryServer: # pragma: no cover if self.repository is not None: self.repository.close() else: - os.write(stderr_fd, "Borg {}: Got connection close before repository was opened.\n" + os.write(stderr_fd, 'Borg {}: Got connection close before repository was opened.\n' .format(__version__).encode()) return unpacker.feed(data) @@ -137,7 +137,7 @@ class RepositoryServer: # pragma: no cover tb = '%s\n%s' % (traceback.format_exc(), sysinfo()) logging.error(msg) logging.log(tb_log_level, tb) - exc = "Remote Exception (see remote log for the traceback)" + exc = 'Remote Exception (see remote log for the traceback)' os.write(stdout_fd, msgpack.packb((1, msgid, e.__class__.__name__, exc))) else: os.write(stdout_fd, msgpack.packb((1, msgid, None, res))) @@ -272,7 +272,7 @@ This TypeError is a cosmetic side effect of the compatibility code borg clients >= 1.0.7 have to support older borg servers. This problem will go away as soon as the server has been upgraded to 1.0.7+. """ - # emit this msg in the same way as the "Remote: ..." lines that show the remote TypeError + # emit this msg in the same way as the 'Remote: ...' lines that show the remote TypeError sys.stderr.write(msg) if append_only: raise self.NoAppendOnlyOnServer() @@ -283,10 +283,10 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. def __del__(self): if len(self.responses): - logging.debug("still %d cached responses left in RemoteRepository" % (len(self.responses),)) + logging.debug('still %d cached responses left in RemoteRepository' % (len(self.responses),)) if self.p: self.close() - assert False, "cleanup happened in Repository.__del__" + assert False, 'cleanup happened in Repository.__del__' def __repr__(self): return '<%s %s>' % (self.__class__.__name__, self.location.canonical_path()) @@ -310,7 +310,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. def borg_cmd(self, args, testing): """return a borg serve command line""" - # give some args/options to "borg serve" process as they were given to us + # give some args/options to 'borg serve' process as they were given to us opts = [] if args is not None: opts.append('--umask=%03o' % args.umask) @@ -430,7 +430,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. while not self.to_send and (calls or self.preload_ids) and len(waiting_for) < MAX_INFLIGHT: if calls: if is_preloaded: - assert cmd == "get", "is_preload is only supported for 'get'" + assert cmd == 'get', "is_preload is only supported for 'get'" if calls[0][0] in self.chunkid_to_msgids: waiting_for.append(pop_preload_msgid(calls.pop(0)[0])) else: @@ -531,7 +531,7 @@ def handle_remote_line(line): logname, msg = msg.split(' ', 1) logging.getLogger(logname).log(level, msg.rstrip()) else: - sys.stderr.write("Remote: " + line) + sys.stderr.write('Remote: ' + line) class RepositoryNoCache: From 6c1b337ce2ac0c0b058608f252f33165fb23fdfc Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Thu, 10 Nov 2016 10:04:15 +0100 Subject: [PATCH 04/11] remote: Replace broken exception argument restoration with code that uses a fixed value. --- src/borg/remote.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index d326ce923..b071a07b2 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -371,13 +371,13 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. elif error == 'CheckNeeded': raise Repository.CheckNeeded(self.location.orig) elif error == 'IntegrityError': - raise IntegrityError(res) + raise IntegrityError('(not available)') elif error == 'PathNotAllowed': - raise PathNotAllowed(*res) + raise PathNotAllowed() elif error == 'ObjectNotFound': - raise Repository.ObjectNotFound(res[0], self.location.orig) + raise Repository.ObjectNotFound('(not available)', self.location.orig) elif error == 'InvalidRPCMethod': - raise InvalidRPCMethod(*res) + raise InvalidRPCMethod('(not available)') else: raise self.RPCError(res.decode('utf-8'), error) From ba553ec628b84cfa2f2f92898627b711244ea1d6 Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Thu, 10 Nov 2016 09:56:18 +0100 Subject: [PATCH 05/11] remote: Introduce rpc protocol with named parameters. --- src/borg/remote.py | 286 +++++++++++++++++++++++++++++++++++++-------- 1 file changed, 236 insertions(+), 50 deletions(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index b071a07b2..b4e1b3d2d 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -1,5 +1,7 @@ import errno import fcntl +import functools +import inspect import logging import os import select @@ -20,8 +22,11 @@ from .helpers import bin_to_hex from .helpers import replace_placeholders from .helpers import yes from .repository import Repository +from .version import parse_version, format_version RPC_PROTOCOL_VERSION = 2 +BORG_VERSION = parse_version(__version__) +MSGID, MSG, ARGS, RESULT = b'i', b'm', b'a', b'r' BUFSIZE = 10 * 1024 * 1024 @@ -54,6 +59,51 @@ class UnexpectedRPCDataFormatFromServer(Error): """Got unexpected RPC data format from server.""" +# Protocol compatibility: +# In general the server is responsible for rejecting too old clients and the client it responsible for rejecting +# too old servers. This ensures that the knowledge what is compatible is always held by the newer component. +# +# The server can do checks for the client version in RepositoryServer.negotiate. If the client_data is 2 then +# client is in the version range [0.29.0, 1.0.x] inclusive. For newer clients client_data is a dict which contains +# client_version. +# +# For the client the return of the negotiate method is either 2 if the server is in the version range [0.29.0, 1.0.x] +# inclusive, or it is a dict which includes the server version. +# +# All method calls on the remote repository object must be whitelisted in RepositoryServer.rpc_methods and have api +# stubs in RemoteRepository. The @api decorator on these stubs is used to set server version requirements. +# +# Method parameters are identified only by name and never by position. Unknown parameters are ignored by the server side. +# If a new parameter is important and may not be ignored, on the client a parameter specific version requirement needs +# to be added. +# When parameters are removed, they need to be preserved as defaulted parameters on the client stubs so that older +# servers still get compatible input. + + +compatMap = { + 'check': ('repair', 'save_space', ), + 'commit': ('save_space', ), + 'rollback': (), + 'destroy': (), + '__len__': (), + 'list': ('limit', 'marker', ), + 'put': ('id', 'data', ), + 'get': ('id_', ), + 'delete': ('id', ), + 'save_key': ('keydata', ), + 'load_key': (), + 'break_lock': (), + 'negotiate': ('client_data', ), + 'open': ('path', 'create', 'lock_wait', 'lock', 'exclusive', 'append_only', ), + 'get_free_nonce': (), + 'commit_nonce_reservation': ('next_unreserved', 'start_nonce', ), +} + + +def decode_keys(d): + return {k.decode(): d[k] for k in d} + + class RepositoryServer: # pragma: no cover rpc_methods = ( '__len__', @@ -79,6 +129,16 @@ class RepositoryServer: # pragma: no cover self.repository = None self.restrict_to_paths = restrict_to_paths self.append_only = append_only + self.client_version = parse_version('1.0.8') # fallback version if client is too old to send version information + + def positional_to_named(self, method, argv): + """Translate from positional protocol to named protocol.""" + return {name: argv[pos] for pos, name in enumerate(compatMap[method])} + + def filter_args(self, f, kwargs): + """Remove unknown named parameters from call, because client did (implicitly) say it's ok.""" + known = set(inspect.signature(f).parameters) + return {name: kwargs[name] for name in kwargs if name in known} def serve(self): stdin_fd = sys.stdin.fileno() @@ -107,12 +167,20 @@ class RepositoryServer: # pragma: no cover return unpacker.feed(data) for unpacked in unpacker: - if not (isinstance(unpacked, tuple) and len(unpacked) == 4): + if isinstance(unpacked, dict): + dictFormat = True + msgid = unpacked[MSGID] + method = unpacked[MSG].decode() + args = decode_keys(unpacked[ARGS]) + elif isinstance(unpacked, tuple) and len(unpacked) == 4: + dictFormat = False + type, msgid, method, args = unpacked + method = method.decode() + args = self.positional_to_named(method, args) + else: if self.repository is not None: self.repository.close() raise UnexpectedRPCDataFormatFromClient(__version__) - type, msgid, method, args = unpacked - method = method.decode() try: if method not in self.rpc_methods: raise InvalidRPCMethod(method) @@ -120,7 +188,8 @@ class RepositoryServer: # pragma: no cover f = getattr(self, method) except AttributeError: f = getattr(self.repository, method) - res = f(*args) + args = self.filter_args(f, args) + res = f(**args) except BaseException as e: if isinstance(e, (Repository.DoesNotExist, Repository.AlreadyExists, PathNotAllowed)): # These exceptions are reconstructed on the client end in RemoteRepository.call_many(), @@ -138,18 +207,35 @@ class RepositoryServer: # pragma: no cover logging.error(msg) logging.log(tb_log_level, tb) exc = 'Remote Exception (see remote log for the traceback)' - os.write(stdout_fd, msgpack.packb((1, msgid, e.__class__.__name__, exc))) + if dictFormat: + os.write(stdout_fd, msgpack.packb({MSGID: msgid, b'exception_class': e.__class__.__name__})) + else: + os.write(stdout_fd, msgpack.packb((1, msgid, e.__class__.__name__, exc))) else: - os.write(stdout_fd, msgpack.packb((1, msgid, None, res))) + if dictFormat: + os.write(stdout_fd, msgpack.packb({MSGID: msgid, RESULT: res})) + else: + os.write(stdout_fd, msgpack.packb((1, msgid, None, res))) if es: self.repository.close() return - def negotiate(self, versions): - return RPC_PROTOCOL_VERSION + def negotiate(self, client_data): + # old format used in 1.0.x + if client_data == RPC_PROTOCOL_VERSION: + return RPC_PROTOCOL_VERSION + # clients since 1.1.0b3 use a dict as client_data + if isinstance(client_data, dict): + self.client_version = client_data[b'client_version'] + else: + self.client_version = BORG_VERSION # seems to be newer than current version (no known old format) + + # not a known old format, send newest negotiate this version knows + return {'server_version': BORG_VERSION} def open(self, path, create=False, lock_wait=None, lock=True, exclusive=None, append_only=False): - path = os.fsdecode(path) + if isinstance(path, bytes): + path = os.fsdecode(path) if path.startswith('/~'): # /~/x = path x relative to home dir, /~username/x = relative to "user" home dir path = os.path.join(get_home_dir(), path[2:]) # XXX check this (see also 1.0-maint), is it correct for ~u? elif path.startswith('/./'): # /./x = path x relative to cwd @@ -203,6 +289,54 @@ class SleepingBandwidthLimiter: return written +def api(*, since, **kwargs_decorator): + """Check version requirements and use self.call to do the remote method call. + + specifies the version in which borg introduced this method, + calling this method when connected to an older version will fail without transmiting + anything to the server. + + Further kwargs can be used to encode version specific restrictions. + If a previous hardcoded behaviour is parameterized in a version, this allows calls that + use the previously hardcoded behaviour to pass through and generates an error if another + behaviour is requested by the client. + + e.g. when 'append_only' was introduced in 1.0.7 the previous behaviour was what now is append_only=False. + Thus @api(..., append_only={'since': parse_version('1.0.7'), 'previously': False}) allows calls + with append_only=False for all version but rejects calls using append_only=True on versions older than 1.0.7. + """ + def decorator(f): + @functools.wraps(f) + def do_rpc(self, *args, **kwargs): + sig = inspect.signature(f) + bound_args = sig.bind(self, *args, **kwargs) + named = {} + for name, param in sig.parameters.items(): + if name == 'self': + continue + if name in bound_args.arguments: + named[name] = bound_args.arguments[name] + else: + if param.default is not param.empty: + named[name] = param.default + + if self.server_version < since: + raise self.RPCServerOutdated(f.__name__, format_version(since)) + + for name, restriction in kwargs_decorator.items(): + if restriction['since'] <= self.server_version: + continue + if 'previously' in restriction and named[name] == restriction['previously']: + continue + + raise self.RPCServerOutdated("{0} {1}={2!s}".format(f.__name__, name, named[name]), + format_version(restriction['since'])) + + return self.call(f.__name__, named) + return do_rpc + return decorator + + class RemoteRepository: extra_test_args = [] @@ -214,6 +348,17 @@ class RemoteRepository: class NoAppendOnlyOnServer(Error): """Server does not support --append-only.""" + class RPCServerOutdated(Error): + """Borg server is too old for {}. Required version {}""" + + @property + def method(self): + return self.args[0] + + @property + def required_version(self): + return self.args[1] + def __init__(self, location, create=False, exclusive=False, lock_wait=None, lock=True, append_only=False, args=None): self.location = self._location = location self.preload_ids = [] @@ -225,6 +370,8 @@ class RemoteRepository: self.ratelimit = SleepingBandwidthLimiter(args.remote_ratelimit * 1024 if args and args.remote_ratelimit else 0) self.unpacker = msgpack.Unpacker(use_list=False) + self.dictFormat = False + self.server_version = parse_version('1.0.8') # fallback version if server is too old to send version information self.p = None testing = location.host == '__testsuite__' borg_cmd = self.borg_cmd(args, testing) @@ -254,15 +401,22 @@ class RemoteRepository: try: try: - version = self.call('negotiate', RPC_PROTOCOL_VERSION) + version = self.call('negotiate', {'client_data': {b'client_version': BORG_VERSION}}) except ConnectionClosed: raise ConnectionClosedWithHint('Is borg working on the server?') from None - if version != RPC_PROTOCOL_VERSION: - raise Exception('Server insisted on using unsupported protocol version %d' % version) + if version == RPC_PROTOCOL_VERSION: + self.dictFormat = False + elif isinstance(version, dict) and b'server_version' in version: + self.dictFormat = True + self.server_version = version[b'server_version'] + else: + raise Exception('Server insisted on using unsupported protocol version %s' % version) + try: - self.id = self.call('open', self.location.path, create, lock_wait, lock, exclusive, append_only) + self.id = self.call('open', {'path': self.location.path, 'create': create, 'lock_wait': lock_wait, + 'lock': lock, 'exclusive': exclusive, 'append_only': append_only}) except self.RPCError as err: - if err.remote_type != 'TypeError': + if self.dictFormat or err.remote_type != 'TypeError': raise msg = """\ Please note: @@ -276,7 +430,9 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. sys.stderr.write(msg) if append_only: raise self.NoAppendOnlyOnServer() - self.id = self.call('open', self.location.path, create, lock_wait, lock) + compatMap['open'] = ('path', 'create', 'lock_wait', 'lock', ) + self.id = self.call('open', {'path': self.location.path, 'create': create, 'lock_wait': lock_wait, + 'lock': lock, 'exclusive': exclusive, 'append_only': append_only}) except Exception: self.close() raise @@ -348,7 +504,10 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. args.append('%s' % location.host) return args - def call(self, cmd, *args, **kw): + def named_to_positional(self, method, kwargs): + return [kwargs[name] for name in compatMap[method]] + + def call(self, cmd, args, **kw): for resp in self.call_many(cmd, [args], **kw): return resp @@ -386,12 +545,12 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. while wait or calls: while waiting_for: try: - error, res = self.responses.pop(waiting_for[0]) + unpacked = self.responses.pop(waiting_for[0]) waiting_for.pop(0) - if error: - handle_error(error, res) + if b'exception_class' in unpacked: + handle_error(unpacked[b'exception_class'], None) else: - yield res + yield unpacked[RESULT] if not waiting_for and not calls: return except KeyError: @@ -410,15 +569,22 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. raise ConnectionClosed() self.unpacker.feed(data) for unpacked in self.unpacker: - if not (isinstance(unpacked, tuple) and len(unpacked) == 4): + if isinstance(unpacked, dict): + msgid = unpacked[MSGID] + elif isinstance(unpacked, tuple) and len(unpacked) == 4: + type, msgid, error, res = unpacked + if error: + unpacked = {MSGID: msgid, b'exception_class': error} + else: + unpacked = {MSGID: msgid, RESULT: res} + else: raise UnexpectedRPCDataFormatFromServer() - type, msgid, error, res = unpacked if msgid in self.ignore_responses: self.ignore_responses.remove(msgid) - if error: - handle_error(error, res) + if b'exception_class' in unpacked: + handle_error(unpacked[b'exception_class'], None) else: - self.responses[msgid] = error, res + self.responses[msgid] = unpacked elif fd is self.stderr_fd: data = os.read(fd, 32768) if not data: @@ -431,22 +597,28 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. if calls: if is_preloaded: assert cmd == 'get', "is_preload is only supported for 'get'" - if calls[0][0] in self.chunkid_to_msgids: - waiting_for.append(pop_preload_msgid(calls.pop(0)[0])) + if calls[0]['id_'] in self.chunkid_to_msgids: + waiting_for.append(pop_preload_msgid(calls.pop(0)['id_'])) else: args = calls.pop(0) - if cmd == 'get' and args[0] in self.chunkid_to_msgids: - waiting_for.append(pop_preload_msgid(args[0])) + if cmd == 'get' and args['id_'] in self.chunkid_to_msgids: + waiting_for.append(pop_preload_msgid(args['id_'])) else: self.msgid += 1 waiting_for.append(self.msgid) - self.to_send = msgpack.packb((1, self.msgid, cmd, args)) + if self.dictFormat: + self.to_send = msgpack.packb({MSGID: self.msgid, MSG: cmd, ARGS: args}) + else: + self.to_send = msgpack.packb((1, self.msgid, cmd, self.named_to_positional(cmd, args))) if not self.to_send and self.preload_ids: chunk_id = self.preload_ids.pop(0) - args = (chunk_id,) + args = {'id_': chunk_id} self.msgid += 1 self.chunkid_to_msgids.setdefault(chunk_id, []).append(self.msgid) - self.to_send = msgpack.packb((1, self.msgid, 'get', args)) + if self.dictFormat: + self.to_send = msgpack.packb({MSGID: self.msgid, MSG: 'get', ARGS: args}) + else: + self.to_send = msgpack.packb((1, self.msgid, 'get', self.named_to_positional(cmd, args))) if self.to_send: try: @@ -458,55 +630,69 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. raise self.ignore_responses |= set(waiting_for) + @api(since=parse_version('1.0.0')) def check(self, repair=False, save_space=False): - return self.call('check', repair, save_space) + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def commit(self, save_space=False): - return self.call('commit', save_space) + """actual remoting is done via self.call in the @api decorator""" - def rollback(self, *args): - return self.call('rollback') + @api(since=parse_version('1.0.0')) + def rollback(self): + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def destroy(self): - return self.call('destroy') + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def __len__(self): - return self.call('__len__') + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def list(self, limit=None, marker=None): - return self.call('list', limit, marker) + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.1.0b3')) def scan(self, limit=None, marker=None): - return self.call('scan', limit, marker) + """actual remoting is done via self.call in the @api decorator""" def get(self, id_): for resp in self.get_many([id_]): return resp def get_many(self, ids, is_preloaded=False): - for resp in self.call_many('get', [(id_,) for id_ in ids], is_preloaded=is_preloaded): + for resp in self.call_many('get', [{'id_': id_} for id_ in ids], is_preloaded=is_preloaded): yield resp - def put(self, id_, data, wait=True): - return self.call('put', id_, data, wait=wait) + @api(since=parse_version('1.0.0')) + def put(self, id, data, wait=True): + """actual remoting is done via self.call in the @api decorator""" - def delete(self, id_, wait=True): - return self.call('delete', id_, wait=wait) + @api(since=parse_version('1.0.0')) + def delete(self, id, wait=True): + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def save_key(self, keydata): - return self.call('save_key', keydata) + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def load_key(self): - return self.call('load_key') + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def get_free_nonce(self): - return self.call('get_free_nonce') + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def commit_nonce_reservation(self, next_unreserved, start_nonce): - return self.call('commit_nonce_reservation', next_unreserved, start_nonce) + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def break_lock(self): - return self.call('break_lock') + """actual remoting is done via self.call in the @api decorator""" def close(self): if self.p: From 4854fcef2e80ae3e5d1e012facb222ae358f94fc Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Sun, 31 Jul 2016 21:19:30 +0200 Subject: [PATCH 06/11] remote: Move open to a normal api stub. --- src/borg/remote.py | 42 +++++++++++++++++++++++++----------------- 1 file changed, 25 insertions(+), 17 deletions(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index b4e1b3d2d..5e536935e 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -345,9 +345,6 @@ class RemoteRepository: self.name = name self.remote_type = remote_type - class NoAppendOnlyOnServer(Error): - """Server does not support --append-only.""" - class RPCServerOutdated(Error): """Borg server is too old for {}. Required version {}""" @@ -412,13 +409,20 @@ class RemoteRepository: else: raise Exception('Server insisted on using unsupported protocol version %s' % version) - try: - self.id = self.call('open', {'path': self.location.path, 'create': create, 'lock_wait': lock_wait, - 'lock': lock, 'exclusive': exclusive, 'append_only': append_only}) - except self.RPCError as err: - if self.dictFormat or err.remote_type != 'TypeError': - raise - msg = """\ + def do_open(): + self.id = self.open(path=self.location.path, create=create, lock_wait=lock_wait, + lock=lock, exclusive=exclusive, append_only=append_only) + + if self.dictFormat: + do_open() + else: + # Ugly detection of versions prior to 1.0.7: If open throws it has to be 1.0.6 or lower + try: + do_open() + except self.RPCError as err: + if err.remote_type != 'TypeError': + raise + msg = """\ Please note: If you see a TypeError complaining about the number of positional arguments given to open(), you can ignore it if it comes from a borg version < 1.0.7. @@ -426,13 +430,12 @@ This TypeError is a cosmetic side effect of the compatibility code borg clients >= 1.0.7 have to support older borg servers. This problem will go away as soon as the server has been upgraded to 1.0.7+. """ - # emit this msg in the same way as the 'Remote: ...' lines that show the remote TypeError - sys.stderr.write(msg) - if append_only: - raise self.NoAppendOnlyOnServer() - compatMap['open'] = ('path', 'create', 'lock_wait', 'lock', ) - self.id = self.call('open', {'path': self.location.path, 'create': create, 'lock_wait': lock_wait, - 'lock': lock, 'exclusive': exclusive, 'append_only': append_only}) + # emit this msg in the same way as the 'Remote: ...' lines that show the remote TypeError + sys.stderr.write(msg) + self.server_version = parse_version('1.0.6') + compatMap['open'] = ('path', 'create', 'lock_wait', 'lock', ), + # try again with corrected version and compatMap + do_open() except Exception: self.close() raise @@ -630,6 +633,11 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. raise self.ignore_responses |= set(waiting_for) + @api(since=parse_version('1.0.0'), + append_only={'since': parse_version('1.0.7'), 'previously': False}) + def open(self, path, create=False, lock_wait=None, lock=True, exclusive=False, append_only=False): + """actual remoting is done via self.call in the @api decorator""" + @api(since=parse_version('1.0.0')) def check(self, repair=False, save_space=False): """actual remoting is done via self.call in the @api decorator""" From e14406fdbfdb3074e021f58933ff3f6c0be38eb8 Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Tue, 8 Nov 2016 23:18:18 +0100 Subject: [PATCH 07/11] remote: Redo exception handling --- src/borg/archiver.py | 10 ++-- src/borg/remote.py | 126 ++++++++++++++++++++++++++++++++----------- 2 files changed, 103 insertions(+), 33 deletions(-) diff --git a/src/borg/archiver.py b/src/borg/archiver.py index 6df3284c3..59f4e17bb 100644 --- a/src/borg/archiver.py +++ b/src/borg/archiver.py @@ -2753,10 +2753,14 @@ def main(): # pragma: no cover tb = "%s\n%s" % (traceback.format_exc(), sysinfo()) exit_code = e.exit_code except RemoteRepository.RPCError as e: - msg = "%s %s" % (e.remote_type, e.name) - important = e.remote_type not in ('LockTimeout', ) + important = e.exception_class not in ('LockTimeout', ) tb_log_level = logging.ERROR if important else logging.DEBUG - tb = sysinfo() + if important: + msg = e.exception_full + else: + msg = e.get_message() + tb = '\n'.join('Borg server: ' + l for l in e.sysinfo.splitlines()) + tb += "\n" + sysinfo() exit_code = EXIT_ERROR except Exception: msg = 'Local Exception' diff --git a/src/borg/remote.py b/src/borg/remote.py index 5e536935e..7d1a11b44 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -191,25 +191,53 @@ class RepositoryServer: # pragma: no cover args = self.filter_args(f, args) res = f(**args) except BaseException as e: - if isinstance(e, (Repository.DoesNotExist, Repository.AlreadyExists, PathNotAllowed)): - # These exceptions are reconstructed on the client end in RemoteRepository.call_many(), - # and will be handled just like locally raised exceptions. Suppress the remote traceback - # for these, except ErrorWithTraceback, which should always display a traceback. - pass - else: - if isinstance(e, Error): - tb_log_level = logging.ERROR if e.traceback else logging.DEBUG - msg = e.get_message() - else: - tb_log_level = logging.ERROR - msg = '%s Exception in RPC call' % e.__class__.__name__ - tb = '%s\n%s' % (traceback.format_exc(), sysinfo()) - logging.error(msg) - logging.log(tb_log_level, tb) - exc = 'Remote Exception (see remote log for the traceback)' if dictFormat: - os.write(stdout_fd, msgpack.packb({MSGID: msgid, b'exception_class': e.__class__.__name__})) + ex_short = traceback.format_exception_only(e.__class__, e) + ex_full = traceback.format_exception(*sys.exc_info()) + if isinstance(e, Error): + ex_short = e.get_message() + if isinstance(e, (Repository.DoesNotExist, Repository.AlreadyExists, PathNotAllowed)): + # These exceptions are reconstructed on the client end in RemoteRepository.call_many(), + # and will be handled just like locally raised exceptions. Suppress the remote traceback + # for these, except ErrorWithTraceback, which should always display a traceback. + pass + else: + logging.debug('\n'.join(ex_full)) + + try: + msg = msgpack.packb({MSGID: msgid, + b'exception_class': e.__class__.__name__, + b'exception_args': e.args, + b'exception_full': ex_full, + b'exception_short': ex_short, + b'sysinfo': sysinfo()}) + except TypeError: + msg = msgpack.packb({MSGID: msgid, + b'exception_class': e.__class__.__name__, + b'exception_args': [x if isinstance(x, (str, bytes, int)) else None + for x in e.args], + b'exception_full': ex_full, + b'exception_short': ex_short, + b'sysinfo': sysinfo()}) + + os.write(stdout_fd, msg) else: + if isinstance(e, (Repository.DoesNotExist, Repository.AlreadyExists, PathNotAllowed)): + # These exceptions are reconstructed on the client end in RemoteRepository.call_many(), + # and will be handled just like locally raised exceptions. Suppress the remote traceback + # for these, except ErrorWithTraceback, which should always display a traceback. + pass + else: + if isinstance(e, Error): + tb_log_level = logging.ERROR if e.traceback else logging.DEBUG + msg = e.get_message() + else: + tb_log_level = logging.ERROR + msg = '%s Exception in RPC call' % e.__class__.__name__ + tb = '%s\n%s' % (traceback.format_exc(), sysinfo()) + logging.error(msg) + logging.log(tb_log_level, tb) + exc = 'Remote Exception (see remote log for the traceback)' os.write(stdout_fd, msgpack.packb((1, msgid, e.__class__.__name__, exc))) else: if dictFormat: @@ -341,9 +369,34 @@ class RemoteRepository: extra_test_args = [] class RPCError(Exception): - def __init__(self, name, remote_type): - self.name = name - self.remote_type = remote_type + def __init__(self, unpacked): + # for borg < 1.1: unpacked only has b'exception_class' as key + # for borg 1.1+: unpacked has keys: b'exception_args', b'exception_full', b'exception_short', b'sysinfo' + self.unpacked = unpacked + + def get_message(self): + if b'exception_short' in self.unpacked: + return b'\n'.join(self.unpacked[b'exception_short']).decode() + else: + return self.exception_class + + @property + def exception_class(self): + return self.unpacked[b'exception_class'].decode() + + @property + def exception_full(self): + if b'exception_full' in self.unpacked: + return b'\n'.join(self.unpacked[b'exception_full']).decode() + else: + return self.get_message() + '\nRemote Exception (see remote log for the traceback)' + + @property + def sysinfo(self): + if b'sysinfo' in self.unpacked: + return self.unpacked[b'sysinfo'].decode() + else: + return '' class RPCServerOutdated(Error): """Borg server is too old for {}. Required version {}""" @@ -411,7 +464,7 @@ class RemoteRepository: def do_open(): self.id = self.open(path=self.location.path, create=create, lock_wait=lock_wait, - lock=lock, exclusive=exclusive, append_only=append_only) + lock=lock, exclusive=exclusive, append_only=append_only) if self.dictFormat: do_open() @@ -420,7 +473,7 @@ class RemoteRepository: try: do_open() except self.RPCError as err: - if err.remote_type != 'TypeError': + if err.exception_class != 'TypeError': raise msg = """\ Please note: @@ -524,8 +577,11 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. del self.chunkid_to_msgids[chunkid] return msgid - def handle_error(error, res): - error = error.decode('utf-8') + def handle_error(unpacked): + error = unpacked[b'exception_class'].decode() + old_server = b'exception_args' not in unpacked + args = unpacked.get(b'exception_args') + if error == 'DoesNotExist': raise Repository.DoesNotExist(self.location.orig) elif error == 'AlreadyExists': @@ -533,15 +589,24 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. elif error == 'CheckNeeded': raise Repository.CheckNeeded(self.location.orig) elif error == 'IntegrityError': - raise IntegrityError('(not available)') + if old_server: + raise IntegrityError('(not available)') + else: + raise IntegrityError(args[0].decode()) elif error == 'PathNotAllowed': raise PathNotAllowed() elif error == 'ObjectNotFound': - raise Repository.ObjectNotFound('(not available)', self.location.orig) + if old_server: + raise Repository.ObjectNotFound('(not available)', self.location.orig) + else: + raise Repository.ObjectNotFound(args[0].decode(), self.location.orig) elif error == 'InvalidRPCMethod': - raise InvalidRPCMethod('(not available)') + if old_server: + raise InvalidRPCMethod('(not available)') + else: + raise InvalidRPCMethod(args[0].decode()) else: - raise self.RPCError(res.decode('utf-8'), error) + raise self.RPCError(unpacked) calls = list(calls) waiting_for = [] @@ -551,7 +616,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. unpacked = self.responses.pop(waiting_for[0]) waiting_for.pop(0) if b'exception_class' in unpacked: - handle_error(unpacked[b'exception_class'], None) + handle_error(unpacked) else: yield unpacked[RESULT] if not waiting_for and not calls: @@ -577,6 +642,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. elif isinstance(unpacked, tuple) and len(unpacked) == 4: type, msgid, error, res = unpacked if error: + # ignore res, because it is only a fixed string anyway. unpacked = {MSGID: msgid, b'exception_class': error} else: unpacked = {MSGID: msgid, RESULT: res} @@ -585,7 +651,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. if msgid in self.ignore_responses: self.ignore_responses.remove(msgid) if b'exception_class' in unpacked: - handle_error(unpacked[b'exception_class'], None) + handle_error(unpacked) else: self.responses[msgid] = unpacked elif fd is self.stderr_fd: From bd3a4a2f9258a3354a2d73f5db1776f1c985f49c Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Tue, 8 Nov 2016 23:19:06 +0100 Subject: [PATCH 08/11] Add testing for exception transport. --- src/borg/remote.py | 24 +++++++++++++- src/borg/testsuite/repository.py | 57 ++++++++++++++++++++++++++++++-- 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index 7d1a11b44..face64c5f 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -122,7 +122,8 @@ class RepositoryServer: # pragma: no cover 'load_key', 'break_lock', 'get_free_nonce', - 'commit_nonce_reservation' + 'commit_nonce_reservation', + 'inject_exception', ) def __init__(self, restrict_to_paths, append_only): @@ -286,6 +287,27 @@ class RepositoryServer: # pragma: no cover self.repository.__enter__() # clean exit handled by serve() method return self.repository.id + def inject_exception(self, kind): + kind = kind.decode() + s1 = 'test string' + s2 = 'test string2' + if kind == 'DoesNotExist': + raise Repository.DoesNotExist(s1) + elif kind == 'AlreadyExists': + raise Repository.AlreadyExists(s1) + elif kind == 'CheckNeeded': + raise Repository.CheckNeeded(s1) + elif kind == 'IntegrityError': + raise IntegrityError(s1) + elif kind == 'PathNotAllowed': + raise PathNotAllowed() + elif kind == 'ObjectNotFound': + raise Repository.ObjectNotFound(s1, s2) + elif kind == 'InvalidRPCMethod': + raise InvalidRPCMethod(s1) + elif kind == 'divide': + 0 // 0 + class SleepingBandwidthLimiter: def __init__(self, limit): diff --git a/src/borg/testsuite/repository.py b/src/borg/testsuite/repository.py index 6d6f00a71..6b66fb736 100644 --- a/src/borg/testsuite/repository.py +++ b/src/borg/testsuite/repository.py @@ -12,7 +12,7 @@ from ..hashindex import NSIndex from ..helpers import Location from ..helpers import IntegrityError from ..locking import Lock, LockFailed -from ..remote import RemoteRepository, InvalidRPCMethod, ConnectionClosedWithHint, handle_remote_line +from ..remote import RemoteRepository, InvalidRPCMethod, PathNotAllowed, ConnectionClosedWithHint, handle_remote_line from ..repository import Repository, LoggedIO, MAGIC, MAX_DATA_SIZE, TAG_DELETE from . import BaseTestCase from .hashindex import H @@ -647,7 +647,60 @@ class RemoteRepositoryTestCase(RepositoryTestCase): exclusive=True, create=create) def test_invalid_rpc(self): - self.assert_raises(InvalidRPCMethod, lambda: self.repository.call('__init__', None)) + self.assert_raises(InvalidRPCMethod, lambda: self.repository.call('__init__', {})) + + def test_rpc_exception_transport(self): + s1 = 'test string' + + try: + self.repository.call('inject_exception', {'kind': 'DoesNotExist'}) + except Repository.DoesNotExist as e: + assert len(e.args) == 1 + assert e.args[0] == self.repository.location.orig + + try: + self.repository.call('inject_exception', {'kind': 'AlreadyExists'}) + except Repository.AlreadyExists as e: + assert len(e.args) == 1 + assert e.args[0] == self.repository.location.orig + + try: + self.repository.call('inject_exception', {'kind': 'CheckNeeded'}) + except Repository.CheckNeeded as e: + assert len(e.args) == 1 + assert e.args[0] == self.repository.location.orig + + try: + self.repository.call('inject_exception', {'kind': 'IntegrityError'}) + except IntegrityError as e: + assert len(e.args) == 1 + assert e.args[0] == s1 + + try: + self.repository.call('inject_exception', {'kind': 'PathNotAllowed'}) + except PathNotAllowed as e: + assert len(e.args) == 0 + + try: + self.repository.call('inject_exception', {'kind': 'ObjectNotFound'}) + except Repository.ObjectNotFound as e: + assert len(e.args) == 2 + assert e.args[0] == s1 + assert e.args[1] == self.repository.location.orig + + try: + self.repository.call('inject_exception', {'kind': 'InvalidRPCMethod'}) + except InvalidRPCMethod as e: + assert len(e.args) == 1 + assert e.args[0] == s1 + + try: + self.repository.call('inject_exception', {'kind': 'divide'}) + except RemoteRepository.RPCError as e: + assert e.unpacked + assert e.get_message() == 'ZeroDivisionError: integer division or modulo by zero\n' + assert e.exception_class == 'ZeroDivisionError' + assert len(e.exception_full) > 0 def test_ssh_cmd(self): assert self.repository.ssh_cmd(Location('example.com:foo')) == ['ssh', 'example.com'] From 8955d8bb2ae56d66050f130ae26f4d10cc597438 Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Thu, 10 Nov 2016 12:12:32 +0100 Subject: [PATCH 09/11] remote: Test that the legacy free rpc bootstrap works. --- src/borg/remote.py | 4 +++- src/borg/testsuite/repository.py | 25 +++++++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index face64c5f..0b15428da 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -431,6 +431,9 @@ class RemoteRepository: def required_version(self): return self.args[1] + # If compatibility with 1.0.x is not longer needed, replace all checks of this with True and simplify the code + dictFormat = False # outside of __init__ for testing of legacy free protocol + def __init__(self, location, create=False, exclusive=False, lock_wait=None, lock=True, append_only=False, args=None): self.location = self._location = location self.preload_ids = [] @@ -442,7 +445,6 @@ class RemoteRepository: self.ratelimit = SleepingBandwidthLimiter(args.remote_ratelimit * 1024 if args and args.remote_ratelimit else 0) self.unpacker = msgpack.Unpacker(use_list=False) - self.dictFormat = False self.server_version = parse_version('1.0.8') # fallback version if server is too old to send version information self.p = None testing = location.host == '__testsuite__' diff --git a/src/borg/testsuite/repository.py b/src/borg/testsuite/repository.py index 6b66fb736..8d77e1205 100644 --- a/src/borg/testsuite/repository.py +++ b/src/borg/testsuite/repository.py @@ -723,6 +723,31 @@ class RemoteRepositoryTestCase(RepositoryTestCase): assert self.repository.borg_cmd(args, testing=False) == ['borg-0.28.2', 'serve', '--umask=077', '--info'] +class RemoteLegacyFree(RepositoryTestCaseBase): + # Keep testing this so we can someday safely remove the legacy tuple format. + + def open(self, create=False): + with patch.object(RemoteRepository, 'dictFormat', True): + return RemoteRepository(Location('__testsuite__:' + os.path.join(self.tmppath, 'repository')), + exclusive=True, create=create) + + def test_legacy_free(self): + # put + self.repository.put(H(0), b'foo') + self.repository.commit() + self.repository.close() + # replace + self.repository = self.open() + with self.repository: + self.repository.put(H(0), b'bar') + self.repository.commit() + # delete + self.repository = self.open() + with self.repository: + self.repository.delete(H(0)) + self.repository.commit() + + @pytest.mark.skipif(sys.platform == 'cygwin', reason='remote is broken on cygwin and hangs') class RemoteRepositoryCheckTestCase(RepositoryCheckTestCase): From 1edff44b3d032826e8cda983f3fdf464ddab1679 Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Thu, 10 Nov 2016 11:08:45 +0100 Subject: [PATCH 10/11] Repository,remote: Rename argument of get to 'id'. --- src/borg/remote.py | 18 +++++++++--------- src/borg/repository.py | 8 ++++---- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index 0b15428da..6894a729a 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -88,7 +88,7 @@ compatMap = { '__len__': (), 'list': ('limit', 'marker', ), 'put': ('id', 'data', ), - 'get': ('id_', ), + 'get': ('id', ), 'delete': ('id', ), 'save_key': ('keydata', ), 'load_key': (), @@ -690,12 +690,12 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. if calls: if is_preloaded: assert cmd == 'get', "is_preload is only supported for 'get'" - if calls[0]['id_'] in self.chunkid_to_msgids: - waiting_for.append(pop_preload_msgid(calls.pop(0)['id_'])) + if calls[0]['id'] in self.chunkid_to_msgids: + waiting_for.append(pop_preload_msgid(calls.pop(0)['id'])) else: args = calls.pop(0) - if cmd == 'get' and args['id_'] in self.chunkid_to_msgids: - waiting_for.append(pop_preload_msgid(args['id_'])) + if cmd == 'get' and args['id'] in self.chunkid_to_msgids: + waiting_for.append(pop_preload_msgid(args['id'])) else: self.msgid += 1 waiting_for.append(self.msgid) @@ -705,7 +705,7 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. self.to_send = msgpack.packb((1, self.msgid, cmd, self.named_to_positional(cmd, args))) if not self.to_send and self.preload_ids: chunk_id = self.preload_ids.pop(0) - args = {'id_': chunk_id} + args = {'id': chunk_id} self.msgid += 1 self.chunkid_to_msgids.setdefault(chunk_id, []).append(self.msgid) if self.dictFormat: @@ -756,12 +756,12 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. def scan(self, limit=None, marker=None): """actual remoting is done via self.call in the @api decorator""" - def get(self, id_): - for resp in self.get_many([id_]): + def get(self, id): + for resp in self.get_many([id]): return resp def get_many(self, ids, is_preloaded=False): - for resp in self.call_many('get', [{'id_': id_} for id_ in ids], is_preloaded=is_preloaded): + for resp in self.call_many('get', [{'id': id} for id in ids], is_preloaded=is_preloaded): yield resp @api(since=parse_version('1.0.0')) diff --git a/src/borg/repository.py b/src/borg/repository.py index 7d1bf8294..fda085a63 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -825,14 +825,14 @@ class Repository: return result return result - def get(self, id_): + def get(self, id): if not self.index: self.index = self.open_index(self.get_transaction_id()) try: - segment, offset = self.index[id_] - return self.io.read(segment, offset, id_) + segment, offset = self.index[id] + return self.io.read(segment, offset, id) except KeyError: - raise self.ObjectNotFound(id_, self.path) from None + raise self.ObjectNotFound(id, self.path) from None def get_many(self, ids, is_preloaded=False): for id_ in ids: From f37109848f884b7e1cc7b0e9a2f8dd871b3be3ed Mon Sep 17 00:00:00 2001 From: Martin Hostettler Date: Tue, 15 Nov 2016 23:18:21 +0100 Subject: [PATCH 11/11] remote: Remove unused type variable. --- src/borg/remote.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/borg/remote.py b/src/borg/remote.py index 6894a729a..9a5221bfb 100644 --- a/src/borg/remote.py +++ b/src/borg/remote.py @@ -175,7 +175,8 @@ class RepositoryServer: # pragma: no cover args = decode_keys(unpacked[ARGS]) elif isinstance(unpacked, tuple) and len(unpacked) == 4: dictFormat = False - type, msgid, method, args = unpacked + # The first field 'type' was always 1 and has always been ignored + _, msgid, method, args = unpacked method = method.decode() args = self.positional_to_named(method, args) else: @@ -664,7 +665,8 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+. if isinstance(unpacked, dict): msgid = unpacked[MSGID] elif isinstance(unpacked, tuple) and len(unpacked) == 4: - type, msgid, error, res = unpacked + # The first field 'type' was always 1 and has always been ignored + _, msgid, error, res = unpacked if error: # ignore res, because it is only a fixed string anyway. unpacked = {MSGID: msgid, b'exception_class': error}