src/allmydata/test/test_auth.py

   1 from twisted.trial import unittest
   2 from twisted.python import filepath
   3 from twisted.cred import error, credentials
   4 from twisted.conch import error as conch_error
   5 from twisted.conch.ssh import keys
   6
   7 from allmydata.frontends import auth
   8 from allmydata.util.fileutil import abspath_expanduser_unicode
   9
  10
  11 DUMMY_KEY = keys.Key.fromString("""\
  12 -----BEGIN RSA PRIVATE KEY-----
  13 MIICXQIBAAKBgQDEP3DYiukOu+NrUlBZeLL9JoHkK5nSvINYfeOQWYVW9J5NG485
  14 pZFVUQKzvvht34Ihj4ucrrvj7vOp+FFvzxI+zHKBpDxyJwV96dvWDAZMjxTxL7iV
  15 8HcO7hqgtQ/Xk1Kjde5lH3EOEDs3IhFHA+sox9y6i4A5NUr2AJZSHiOEVwIDAQAB
  16 AoGASrrNwefDr7SkeS2zIx7vKa8ML1LbFIBsk7n8ee9c8yvbTAl+lLkTiqV6ne/O
  17 sig2aYk75MI1Eirf5o2ElUsI6u36i6AeKL2u/W7tLBVijmBB8dTiWZ5gMOARWt8w
  18 daF2An2826YdcU+iNZ7Yi0q4xtlxHQn3JcNNWxicphLvt0ECQQDtajJ/bK+Nqd9j
  19 /WGvqYcMzkkorQq/0+MQYhcIwDlpf2Xoi45tP4HeoBubeJmU5+jXpXmdP5epWpBv
  20 k3ZCwV7pAkEA05xBP2HTdwRFTJov5I/w7uKOrn7mj7DCvSjQFCufyPOoCJJMeBSq
  21 tfCQlHFtwlkyNfiSbhtgZ0Pp6ovL+1RBPwJBAOlFRBKxrpgpxcXQK5BWqMwrT/S4
  22 eWxb+6mYR3ugq4h91Zq0rJ+pG6irdhS/XV/SsZRZEXIxDoom4u3OXQ9gQikCQErM
  23 ywuaiuNhMRXY0uEaOHJYx1LLLLjSJKQ0zwiyOvMPnfAZtsojlAxoEtNGHSQ731HQ
  24 ogIlzzfxe7ga3mni6IUCQQCwNK9zwARovcQ8nByqotGQzohpl+1b568+iw8GXP2u
  25 dBSD8940XU3YW+oeq8e+p3yQ2GinHfeJ3BYQyNQLuMAJ
  26 -----END RSA PRIVATE KEY-----
  27 """)
  28
  29 DUMMY_ACCOUNTS = u"""\
  30 alice password URI:DIR2:aaaaaaaaaaaaaaaaaaaaaaaaaa:1111111111111111111111111111111111111111111111111111
  31 bob sekrit URI:DIR2:bbbbbbbbbbbbbbbbbbbbbbbbbb:2222222222222222222222222222222222222222222222222222
  32 carol {key} URI:DIR2:cccccccccccccccccccccccccc:3333333333333333333333333333333333333333333333333333
  33 """.format(key=DUMMY_KEY.public().toString("openssh")).encode("ascii")
  34
  35 class AccountFileCheckerKeyTests(unittest.TestCase):
  36     """
  37     Tests for key handling done by allmydata.frontends.auth.AccountFileChecker.
  38     """
  39     def setUp(self):
  40         self.account_file = filepath.FilePath(self.mktemp())
  41         self.account_file.setContent(DUMMY_ACCOUNTS)
  42         abspath = abspath_expanduser_unicode(unicode(self.account_file.path))
  43         self.checker = auth.AccountFileChecker(None, abspath)
  44
  45     def test_unknown_user(self):
  46         """
  47         AccountFileChecker.requestAvatarId returns a Deferred that fires with
  48         UnauthorizedLogin if called with an SSHPrivateKey object with a
  49         username not present in the account file.
  50         """
  51         key_credentials = credentials.SSHPrivateKey(
  52             b"dennis", b"md5", None, None, None)
  53         avatarId = self.checker.requestAvatarId(key_credentials)
  54         return self.assertFailure(avatarId, error.UnauthorizedLogin)
  55
  56     def test_password_auth_user(self):
  57         """
  58         AccountFileChecker.requestAvatarId returns a Deferred that fires with
  59         UnauthorizedLogin if called with an SSHPrivateKey object for a username
  60         only associated with a password in the account file.
  61         """
  62         key_credentials = credentials.SSHPrivateKey(
  63             b"alice", b"md5", None, None, None)
  64         avatarId = self.checker.requestAvatarId(key_credentials)
  65         return self.assertFailure(avatarId, error.UnauthorizedLogin)
  66
  67     def test_unrecognized_key(self):
  68         """
  69         AccountFileChecker.requestAvatarId returns a Deferred that fires with
  70         UnauthorizedLogin if called with an SSHPrivateKey object with a public
  71         key other than the one indicated in the account file for the indicated
  72         user.
  73         """
  74         wrong_key_blob = b"""\
  75 ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAYQDJGMWlPXh2M3pYzTiamjcBIMqctt4VvLVW2QZgEFc86XhGjPXq5QAiRTKv9yVZJR9HW70CfBI7GHun8+v4Wb6aicWBoxgI3OB5NN+OUywdme2HSaif5yenFdQr0ME71Xs=
  76 """
  77         key_credentials = credentials.SSHPrivateKey(
  78             b"carol", b"md5", wrong_key_blob, None, None)
  79         avatarId = self.checker.requestAvatarId(key_credentials)
  80         return self.assertFailure(avatarId, error.UnauthorizedLogin)
  81
  82     def test_missing_signature(self):
  83         """
  84         AccountFileChecker.requestAvatarId returns a Deferred that fires with
  85         ValidPublicKey if called with an SSHPrivateKey object with an
  86         authorized key for the indicated user but with no signature.
  87         """
  88         right_key_blob = DUMMY_KEY.public().toString("openssh")
  89         key_credentials = credentials.SSHPrivateKey(
  90             b"carol", b"md5", right_key_blob, None, None)
  91         avatarId = self.checker.requestAvatarId(key_credentials)
  92         return self.assertFailure(avatarId, conch_error.ValidPublicKey)
  93
  94     def test_wrong_signature(self):
  95         """
  96         AccountFileChecker.requestAvatarId returns a Deferred that fires with
  97         UnauthorizedLogin if called with an SSHPrivateKey object with a public
  98         key matching that on the user's line in the account file but with the
  99         wrong signature.
 100         """
 101         right_key_blob = DUMMY_KEY.public().toString("openssh")
 102         key_credentials = credentials.SSHPrivateKey(
 103             b"carol", b"md5", right_key_blob, b"signed data", b"wrong sig")
 104         avatarId = self.checker.requestAvatarId(key_credentials)
 105         return self.assertFailure(avatarId, error.UnauthorizedLogin)
 106
 107     def test_authenticated(self):
 108         """
 109         If called with an SSHPrivateKey object with a username and public key
 110         found in the account file and a signature that proves possession of the
 111         corresponding private key, AccountFileChecker.requestAvatarId returns a
 112         Deferred that fires with an FTPAvatarID giving the username and root
 113         capability for that user.
 114         """
 115         username = b"carol"
 116         signed_data = b"signed data"
 117         signature = DUMMY_KEY.sign(signed_data)
 118         right_key_blob = DUMMY_KEY.public().toString("openssh")
 119         key_credentials = credentials.SSHPrivateKey(
 120             username, b"md5", right_key_blob, signed_data, signature)
 121         avatarId = self.checker.requestAvatarId(key_credentials)
 122         def authenticated(avatarId):
 123             self.assertEqual(
 124                 (username,
 125                  b"URI:DIR2:cccccccccccccccccccccccccc:3333333333333333333333333333333333333333333333333333"),
 126                 (avatarId.username, avatarId.rootcap))
 127         avatarId.addCallback(authenticated)
 128         return avatarId