Rooba
/
mitogen
mirror of https://github.com/mitogen-hq/mitogen.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

tests: file_service_test fixes 

Due to 6713b90acc
This commit is contained in:
David Wilson 2019-01-19 06:19:04 +00:00
parent 04755c3321
commit b2b7e7b06b
1 changed files with 56 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
tests/file_service_test.py
View File

@ -9,72 +9,100 @@ import testlib
class FetchTest(testlib.RouterMixin, testlib.TestCase):
klass = mitogen.service.FileService
def replyable_msg(self, **kwargs):
recv = mitogen.core.Receiver(self.router, persist=False)
msg = mitogen.core.Message(
src_id=mitogen.context_id,
reply_to=recv.handle,
**kwargs
)
msg.router = self.router
return recv, msg
def test_unauthorized(self):
service = self.klass(self.router)
e = self.assertRaises(mitogen.service.Error,
lambda: service.fetch(
path='/etc/shadow',
sender=None,
msg=mitogen.core.Message(),
)
recv, msg = self.replyable_msg()
service.fetch(
path='/etc/shadow',
sender=None,
msg=msg,
)
e = self.assertRaises(mitogen.core.CallError,
lambda: recv.get().unpickle())
expect = service.unregistered_msg % ('/etc/shadow',)
self.assertTrue(expect in e.args[0])
self.assertEquals(e.args[0], service.unregistered_msg)
def _validate_response(self, resp):
self.assertTrue(isinstance(resp, dict))
self.assertEquals('root', resp['owner'])
self.assertEquals('root', resp['group'])
self.assertTrue(isinstance(resp['mode'], int))
self.assertTrue(isinstance(resp['mtime'], float))
self.assertTrue(isinstance(resp['atime'], float))
self.assertTrue(isinstance(resp['size'], int))
def test_path_authorized(self):
recv = mitogen.core.Receiver(self.router)
service = self.klass(self.router)
service.register('/etc/passwd')
self.assertEquals(None, service.fetch(
recv, msg = self.replyable_msg()
service.fetch(
path='/etc/passwd',
sender=recv.to_sender(),
msg=mitogen.core.Message(),
))
msg=msg,
)
self._validate_response(recv.get().unpickle())
def test_root_authorized(self):
recv = mitogen.core.Receiver(self.router)
service = self.klass(self.router)
service.register_prefix('/')
self.assertEquals(None, service.fetch(
recv, msg = self.replyable_msg()
service.fetch(
path='/etc/passwd',
sender=recv.to_sender(),
msg=mitogen.core.Message(),
))
msg=msg,
)
self._validate_response(recv.get().unpickle())
def test_prefix_authorized(self):
recv = mitogen.core.Receiver(self.router)
service = self.klass(self.router)
service.register_prefix('/etc')
self.assertEquals(None, service.fetch(
recv, msg = self.replyable_msg()
service.fetch(
path='/etc/passwd',
sender=recv.to_sender(),
msg=mitogen.core.Message(),
))
msg=msg,
)
self._validate_response(recv.get().unpickle())
def test_prefix_authorized_abspath_bad(self):
recv = mitogen.core.Receiver(self.router)
service = self.klass(self.router)
service.register_prefix('/etc')
self.assertEquals(None, service.fetch(
recv, msg = self.replyable_msg()
service.fetch(
path='/etc/foo/bar/../../../passwd',
sender=recv.to_sender(),
msg=mitogen.core.Message(),
))
msg=msg,
)
self.assertEquals(None, recv.get().unpickle())
def test_prefix_authorized_abspath_bad(self):
recv = mitogen.core.Receiver(self.router)
service = self.klass(self.router)
service.register_prefix('/etc')
e = self.assertRaises(mitogen.service.Error,
lambda: service.fetch(
path='/etc/../shadow',
sender=recv.to_sender(),
msg=mitogen.core.Message(),
)
recv, msg = self.replyable_msg()
service.fetch(
path='/etc/../shadow',
sender=recv.to_sender(),
msg=msg,
)
self.assertEquals(e.args[0], service.unregistered_msg)
e = self.assertRaises(mitogen.core.CallError,
lambda: recv.get().unpickle())
expect = service.unregistered_msg % ('/etc/../shadow',)
self.assertTrue(expect in e.args[0])
if __name__ == '__main__':