test_auth.py

from etcd.tests.integration.test_simple import EtcdIntegrationTest
from etcd import auth
import etcd


class TestEtcdAuthBase(EtcdIntegrationTest):
    cl_size = 1

    def setUp(self):
        # Sets up the root user, toggles auth
        u = auth.EtcdUser(self.client, "root")
        u.password = "testpass"
        u.write()
        self.client = etcd.Client(port=6001, username="root", password="testpass")
        self.unauth_client = etcd.Client(port=6001)
        a = auth.Auth(self.client)
        a.active = True

    def tearDown(self):
        u = auth.EtcdUser(self.client, "test_user")
        r = auth.EtcdRole(self.client, "test_role")
        try:
            u.delete()
        except:
            pass
        try:
            r.delete()
        except:
            pass
        a = auth.Auth(self.client)
        a.active = False


class EtcdUserTest(TestEtcdAuthBase):
    def test_names(self):
        u = auth.EtcdUser(self.client, "test_user")
        self.assertEqual(u.names, ["root"])

    def test_read(self):
        u = auth.EtcdUser(self.client, "root")
        # Reading an existing user succeeds
        try:
            u.read()
        except Exception:
            self.fail("reading the root user raised an exception")

        # roles for said user are fetched
        self.assertEqual(u.roles, set(["root"]))

        # The user is correctly rendered out
        self.assertEqual(u._to_net(), [{"user": "root", "password": None, "roles": ["root"]}])

        # An inexistent user raises the appropriate exception
        u = auth.EtcdUser(self.client, "user.does.not.exist")
        self.assertRaises(etcd.EtcdKeyNotFound, u.read)

        # Reading with an unauthenticated client raises an exception
        u = auth.EtcdUser(self.unauth_client, "root")
        self.assertRaises(etcd.EtcdInsufficientPermissions, u.read)

        # Generic errors are caught
        c = etcd.Client(port=9999)
        u = auth.EtcdUser(c, "root")
        self.assertRaises(etcd.EtcdException, u.read)

    def test_write_and_delete(self):
        # Create an user
        u = auth.EtcdUser(self.client, "test_user")
        u.roles.add("guest")
        u.roles.add("root")
        # directly from my suitcase
        u.password = "123456"
        try:
            u.write()
        except:
            self.fail("creating a user doesn't work")
        # Password gets wiped
        self.assertEqual(u.password, None)
        u.read()
        # Verify we can log in as this user and access the auth (it has the
        # root role)
        cl = etcd.Client(port=6001, username="test_user", password="123456")
        ul = auth.EtcdUser(cl, "root")
        try:
            ul.read()
        except etcd.EtcdInsufficientPermissions:
            self.fail("Reading auth with the new user is not possible")

        self.assertEqual(u.name, "test_user")
        self.assertEqual(u.roles, set(["guest", "root"]))
        # set roles as a list, it works!
        u.roles = ["guest", "test_group"]
        # We need this or the new API will return an internal error
        r = auth.EtcdRole(self.client, "test_group")
        r.acls = {"*": "R", "/test/*": "RW"}
        r.write()
        try:
            u.write()
        except:
            self.fail("updating a user you previously created fails")
        u.read()
        self.assertIn("test_group", u.roles)

        # Unauthorized access is properly handled
        ua = auth.EtcdUser(self.unauth_client, "test_user")
        self.assertRaises(etcd.EtcdInsufficientPermissions, ua.write)

        # now let's test deletion
        du = auth.EtcdUser(self.client, "user.does.not.exist")
        self.assertRaises(etcd.EtcdKeyNotFound, du.delete)

        # Delete test_user
        u.delete()
        self.assertRaises(etcd.EtcdKeyNotFound, u.read)
        # Permissions are properly handled
        self.assertRaises(etcd.EtcdInsufficientPermissions, ua.delete)


class EtcdRoleTest(TestEtcdAuthBase):
    def test_names(self):
        r = auth.EtcdRole(self.client, "guest")
        self.assertListEqual(r.names, ["guest", "root"])

    def test_read(self):
        r = auth.EtcdRole(self.client, "guest")
        try:
            r.read()
        except:
            self.fail("Reading an existing role failed")

        # XXX The ACL path result changed from '*' to '/*' at some point
        #     between etcd-2.2.2 and 2.2.5.  They're equivalent so allow
        #     for both.
        if "/*" in r.acls:
            self.assertEqual(r.acls, {"/*": "RW"})
        else:
            self.assertEqual(r.acls, {"*": "RW"})
        # We can actually skip most other read tests as they are common
        # with EtcdUser

    def test_write_and_delete(self):
        r = auth.EtcdRole(self.client, "test_role")
        r.acls = {"*": "R", "/test/*": "RW"}
        try:
            r.write()
        except:
            self.fail("Writing a simple groups should not fail")

        r1 = auth.EtcdRole(self.client, "test_role")
        r1.read()
        self.assertEqual(r1.acls, r.acls)
        r.revoke("/test/*", "W")
        r.write()
        r1.read()
        self.assertEqual(r1.acls, {"*": "R", "/test/*": "R"})
        r.grant("/pub/*", "RW")
        r.write()
        r1.read()
        self.assertEqual(r1.acls["/pub/*"], "RW")
        # All other exceptions are tested by the user tests
        r1.name = None
        self.assertRaises(etcd.EtcdException, r1.write)
        # ditto for delete
        try:
            r.delete()
        except:
            self.fail("A normal delete should not fail")
        self.assertRaises(etcd.EtcdKeyNotFound, r.read)