src/allmydata/frontends/auth.py

   1 import os
   2 from zope.interface import implements
   3 from twisted.web.client import getPage
   4 from twisted.internet import defer
   5 from twisted.cred import error, checkers, credentials
   6 from twisted.conch import error as conch_error
   7 from twisted.conch.ssh import keys
   8
   9 from allmydata.util import base32
  10
  11 class NeedRootcapLookupScheme(Exception):
  12     """Accountname+Password-based access schemes require some kind of
  13     mechanism to translate name+passwd pairs into a rootcap, either a file of
  14     name/passwd/rootcap tuples, or a server to do the translation."""
  15
  16 class FTPAvatarID:
  17     def __init__(self, username, rootcap):
  18         self.username = username
  19         self.rootcap = rootcap
  20
  21 class AccountFileChecker:
  22     implements(checkers.ICredentialsChecker)
  23     credentialInterfaces = (credentials.IUsernamePassword,
  24                             credentials.IUsernameHashedPassword,
  25                             credentials.ISSHPrivateKey)
  26     def __init__(self, client, accountfile):
  27         self.client = client
  28         self.passwords = {}
  29         self.pubkeys = {}
  30         self.rootcaps = {}
  31         for line in open(os.path.expanduser(accountfile), "r"):
  32             line = line.strip()
  33             if line.startswith("#") or not line:
  34                 continue
  35             name, passwd, rest = line.split(None, 2)
  36             if passwd.startswith("ssh-"):
  37                 bits = rest.split()
  38                 keystring = " ".join([passwd] + bits[:-1])
  39                 rootcap = bits[-1]
  40                 self.pubkeys[name] = keystring
  41             else:
  42                 self.passwords[name] = passwd
  43                 rootcap = rest
  44             self.rootcaps[name] = rootcap
  45
  46     def _avatarId(self, username):
  47         return FTPAvatarID(username, self.rootcaps[username])
  48
  49     def _cbPasswordMatch(self, matched, username):
  50         if matched:
  51             return self._avatarId(username)
  52         raise error.UnauthorizedLogin
  53
  54     def requestAvatarId(self, creds):
  55         if credentials.ISSHPrivateKey.providedBy(creds):
  56             # Re-using twisted.conch.checkers.SSHPublicKeyChecker here, rather
  57             # than re-implementing all of the ISSHPrivateKey checking logic,
  58             # would be better.  That would require Twisted 14.1.0 or newer,
  59             # though.
  60             return self._checkKey(creds)
  61         elif credentials.IUsernameHashedPassword.providedBy(creds):
  62             return self._checkPassword(creds)
  63         elif credentials.IUsernamePassword.providedBy(creds):
  64             return self._checkPassword(creds)
  65         else:
  66             raise NotImplementedError()
  67
  68     def _checkPassword(self, creds):
  69         """
  70         Determine whether the password in the given credentials matches the
  71         password in the account file.
  72
  73         Returns a Deferred that fires with the username if the password matches
  74         or with an UnauthorizedLogin failure otherwise.
  75         """
  76         try:
  77             correct = self.passwords[creds.username]
  78         except KeyError:
  79             return defer.fail(error.UnauthorizedLogin())
  80
  81         d = defer.maybeDeferred(creds.checkPassword, correct)
  82         d.addCallback(self._cbPasswordMatch, str(creds.username))
  83         return d
  84
  85     def _checkKey(self, creds):
  86         """
  87         Determine whether some key-based credentials correctly authenticates a
  88         user.
  89
  90         Returns a Deferred that fires with the username if so or with an
  91         UnauthorizedLogin failure otherwise.
  92         """
  93
  94         # Is the public key indicated by the given credentials allowed to
  95         # authenticate the username in those credentials?
  96         if creds.blob == self.pubkeys.get(creds.username):
  97             if creds.signature is None:
  98                 return defer.fail(conch_error.ValidPublicKey())
  99
 100             # Is the signature in the given credentials the correct
 101             # signature for the data in those credentials?
 102             key = keys.Key.fromString(creds.blob)
 103             if key.verify(creds.signature, creds.sigData):
 104                 return defer.succeed(self._avatarId(creds.username))
 105
 106         return defer.fail(error.UnauthorizedLogin())
 107
 108 class AccountURLChecker:
 109     implements(checkers.ICredentialsChecker)
 110     credentialInterfaces = (credentials.IUsernamePassword,)
 111
 112     def __init__(self, client, auth_url):
 113         self.client = client
 114         self.auth_url = auth_url
 115
 116     def _cbPasswordMatch(self, rootcap, username):
 117         return FTPAvatarID(username, rootcap)
 118
 119     def post_form(self, username, password):
 120         sepbase = base32.b2a(os.urandom(4))
 121         sep = "--" + sepbase
 122         form = []
 123         form.append(sep)
 124         fields = {"action": "authenticate",
 125                   "email": username,
 126                   "passwd": password,
 127                   }
 128         for name, value in fields.iteritems():
 129             form.append('Content-Disposition: form-data; name="%s"' % name)
 130             form.append('')
 131             assert isinstance(value, str)
 132             form.append(value)
 133             form.append(sep)
 134         form[-1] += "--"
 135         body = "\r\n".join(form) + "\r\n"
 136         headers = {"content-type": "multipart/form-data; boundary=%s" % sepbase,
 137                    }
 138         return getPage(self.auth_url, method="POST",
 139                        postdata=body, headers=headers,
 140                        followRedirect=True, timeout=30)
 141
 142     def _parse_response(self, res):
 143         rootcap = res.strip()
 144         if rootcap == "0":
 145             raise error.UnauthorizedLogin
 146         return rootcap
 147
 148     def requestAvatarId(self, credentials):
 149         # construct a POST to the login form. While this could theoretically
 150         # be done with something like the stdlib 'email' package, I can't
 151         # figure out how, so we just slam together a form manually.
 152         d = self.post_form(credentials.username, credentials.password)
 153         d.addCallback(self._parse_response)
 154         d.addCallback(self._cbPasswordMatch, str(credentials.username))
 155         return d
 156