mirror/borg
1
0
Fork 0
mirror of https://github.com/borgbackup/borg.git synced 2024-12-22 15:57:15 +00:00
Code Issues Releases Wiki Activity

PR #173 - Merge branch 'rpc-whitelist' of https://github.com/dnnr/attic into merge

Browse source
This commit is contained in:
Thomas Waldmann 2015-03-06 22:55:53 +01:00
commit 8c4d290a93
2 changed files with 24 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
attic/remote.py
View file

@ -22,8 +22,23 @@ class ConnectionClosed(Error):
class PathNotAllowed(Error):
"""Repository path not allowed"""
class InvalidRPCMethod(Error):
"""RPC method is not valid"""
class RepositoryServer(object):
rpc_methods = (
'__len__',
'check',
'commit',
'delete',
'get',
'list',
'negotiate',
'open',
'put',
'repair',
'rollback',
)
def __init__(self, restrict_to_paths):
self.repository = None
@ -47,6 +62,8 @@ def serve(self):
for type, msgid, method, args in unpacker:
method = method.decode('ascii')
try:
if not method in self.rpc_methods:
raise InvalidRPCMethod(method)
try:
f = getattr(self, method)
except AttributeError:
@ -155,8 +172,10 @@ def fetch_from_cache(args):
raise IntegrityError(res)
elif error == b'PathNotAllowed':
raise PathNotAllowed(*res)
if error == b'ObjectNotFound':
elif error == b'ObjectNotFound':
raise Repository.ObjectNotFound(res[0], self.location.orig)
elif error == b'InvalidRPCMethod':
raise InvalidRPCMethod(*res)
raise self.RPCError(error)
else:
yield res

5
attic/testsuite/repository.py
View file

@ -4,7 +4,7 @@
from attic.testsuite.mock import patch
from attic.hashindex import NSIndex
from attic.helpers import Location, IntegrityError, UpgradableLock
from attic.remote import RemoteRepository
from attic.remote import RemoteRepository, InvalidRPCMethod
from attic.repository import Repository
from attic.testsuite import AtticTestCase
@ -319,6 +319,9 @@ class RemoteRepositoryTestCase(RepositoryTestCase):
def open(self, create=False):
return RemoteRepository(Location('__testsuite__:' + os.path.join(self.tmppath, 'repository')), create=create)
def test_invalid_rpc(self):
self.assert_raises(InvalidRPCMethod, lambda: self.repository.call('__init__', None))
class RemoteRepositoryCheckTestCase(RepositoryCheckTestCase):