User: Add support for authentication via GPG.

This commit is contained in:
Valentin Lorentz 2013-01-20 00:09:37 +01:00
parent cd0bfe411a
commit 97bffbdef1
3 changed files with 169 additions and 6 deletions

View File

@ -47,4 +47,13 @@ User = conf.registerPlugin('User')
# conf.registerGlobalValue(User, 'someConfigVariableName', # conf.registerGlobalValue(User, 'someConfigVariableName',
# registry.Boolean(False, """Help for someConfigVariableName.""")) # registry.Boolean(False, """Help for someConfigVariableName."""))
conf.registerGroup(User, 'gpg')
conf.registerGlobalValue(User.gpg, 'enable',
registry.Boolean(True, """Determines whether or not users are
allowed to use GPG for authentication."""))
conf.registerGlobalValue(User.gpg, 'TokenTimeout',
registry.PositiveInteger(60*10, """Determines the lifetime of a GPG
authentication token (in seconds)."""))
# vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: # vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79:

View File

@ -28,10 +28,12 @@
### ###
import re import re
import uuid
import time
import fnmatch import fnmatch
import supybot.gpg as gpg
import supybot.conf as conf import supybot.conf as conf
import supybot.gpg as gpg
import supybot.utils as utils import supybot.utils as utils
import supybot.ircdb as ircdb import supybot.ircdb as ircdb
from supybot.commands import * from supybot.commands import *
@ -384,13 +386,22 @@ class User(callbacks.Plugin):
additional('something', '')]) additional('something', '')])
class gpg(callbacks.Commands): class gpg(callbacks.Commands):
def __init__(self, *args):
super(User.gpg, self).__init__(*args)
self._tokens = {}
def callCommand(self, command, irc, msg, *args, **kwargs): def callCommand(self, command, irc, msg, *args, **kwargs):
if gpg.available: if gpg.available and self.registryValue('gpg.enable'):
return super(gpg, self) \ return super(User.gpg, self) \
.callCommand(command, irc, msg, *args, **kwargs) .callCommand(command, irc, msg, *args, **kwargs)
else: else:
irc.error(_('GPG features are not enabled.')) irc.error(_('GPG features are not enabled.'))
def _expire_tokens(self):
now = time.time()
self._tokens = dict(filter(lambda (x,y): y[1]>now,
self._tokens.items()))
@internationalizeDocstring @internationalizeDocstring
def add(self, irc, msg, args, user, keyid, keyserver): def add(self, irc, msg, args, user, keyid, keyserver):
"""<key id> <key server> """<key id> <key server>
@ -432,6 +443,63 @@ class User(callbacks.Plugin):
irc.error(_('GPG key not associated with your account.')) irc.error(_('GPG key not associated with your account.'))
remove = wrap(remove, ['user', 'somethingWithoutSpaces']) remove = wrap(remove, ['user', 'somethingWithoutSpaces'])
@internationalizeDocstring
def gettoken(self, irc, msg, args):
"""takes no arguments
Send you a token that you'll have to sign with your key."""
self._expire_tokens()
token = '{%s}' % str(uuid.uuid4())
lifetime = conf.supybot.plugins.User.gpg.TokenTimeout()
self._tokens.update({token: (msg.prefix, time.time()+lifetime)})
irc.reply(_('Your token is: %s. Please sign it with your '
'GPG key, paste it somewhere, and call the \'auth\' '
'command with the URL to the (raw) file containing the '
'signature.') % token)
gettoken = wrap(gettoken, [])
_auth_re = re.compile(r'-----BEGIN PGP SIGNED MESSAGE-----\r?\n'
r'Hash: .*\r?\n\r?\n'
r'\s*({[0-9a-z-]+})\s*\r?\n'
r'-----BEGIN PGP SIGNATURE-----\r?\n.*'
r'\r?\n-----END PGP SIGNATURE-----',
re.S)
@internationalizeDocstring
def auth(self, irc, msg, args, url):
"""<url>
Check the GPG signature at the <url> and authenticates you if
the key used is associated to a user."""
self._expire_tokens()
match = self._auth_re.search(utils.web.getUrl(url))
if not match:
irc.error(_('Signature or token not found.'), Raise=True)
data = match.group(0)
token = match.group(1)
if token not in self._tokens:
irc.error(_('Unknown token. It may have expired before you '
'submit it.'), Raise=True)
if self._tokens[token][0] != msg.prefix:
irc.error(_('Your hostname/nick changed in the process. '
'Authentication aborted.'))
verified = gpg.keyring.verify(data)
if verified and verified.valid:
keyid = verified.key_id
prefix, expiry = self._tokens.pop(token)
found = False
for (id, user) in ircdb.users.items():
if keyid in map(lambda x:x[-len(keyid):], user.gpgkeys):
user.addAuth(msg.prefix)
ircdb.users.setUser(user, flush=False)
irc.reply(_('You are now authenticated as %s.') %
user.name)
return
irc.error(_('Unknown GPG key.'), Raise=True)
else:
irc.error(_('Signature could not be verified. Make sure '
'this is a valid GPG signature and the URL is valid.'))
auth = wrap(auth, ['url'])
@internationalizeDocstring @internationalizeDocstring
def capabilities(self, irc, msg, args, user): def capabilities(self, irc, msg, args, user):
"""[<name>] """[<name>]

View File

@ -27,11 +27,57 @@
# POSSIBILITY OF SUCH DAMAGE. # POSSIBILITY OF SUCH DAMAGE.
### ###
import supybot.gpg as gpg import re
from supybot.test import * from cStringIO import StringIO
import supybot.gpg as gpg
from supybot.test import PluginTestCase, network
import supybot.conf as conf
import supybot.world as world import supybot.world as world
import supybot.ircdb as ircdb import supybot.ircdb as ircdb
import supybot.utils as utils
PRIVATE_KEY = """
-----BEGIN PGP PRIVATE KEY BLOCK-----
Version: GnuPG v1.4.12 (GNU/Linux)
lQHYBFD7GxQBBACeu7bj/wgnnv5NkfHImZJVJLaq2cwKYc3rErv7pqLXpxXZbDOI
jP+5eSmTLhPUK67aRD6gG0wQ9iAhYR03weOmyjDGh0eF7kLYhu/4Il56Y/YbB8ll
Imz/pep/Hi72ShcW8AtifDup/KeHjaWa1yF2WThHbX/0N2ghSxbJnatpBwARAQAB
AAP6Arf7le7FD3ZhGZvIBkPr25qca6i0Qxb5XpOinV7jLcoycZriJ9Xofmhda9UO
xhNVppMvs/ofI/m0umnR4GLKtRKnJSc8Edxi4YKyqLehfBTF20R/kBYPZ772FkNW
Kzo5yCpP1jpOc0+QqBuU7OmrG4QhQzTLXIUgw4XheORncEECAMGkvR47PslJqzbY
VRIzWEv297r1Jxqy6qgcuCJn3RWYJbEZ/qdTYy+MgHGmaNFQ7yhfIzkBueq0RWZp
Z4PfJn8CANHZGj6AJZcvb+VclNtc5VNfnKjYD+qQOh2IS8NhE/0umGMKz3frH1TH
yCbh2LlPR89cqNcd4QvbHKA/UmzISXkB/37MbUnxXTpS9Y4HNpQCh/6SYlB0lucV
QN0cgjfhd6nBrb6uO6+u40nBzgynWcEpPMNfN0AtQeA4Dx+WrnK6kZqfd7QMU3Vw
eWJvdCB0ZXN0iLgEEwECACIFAlD7GxQCGwMGCwkIBwMCBhUIAgkKCwQWAgMBAh4B
AheAAAoJEMnTMjwgrwErV3AD/0kRq8UWPlkc6nyiIR6qiT3EoBNHKIi4cz68Wa1u
F2M6einrRR0HolrxonynTGsdr1u2f3egOS4fNfGhTNAowSefYR9q5kIYiYE2DL5G
YnjJKNfmnRxZM9YqmEnN50rgu2cifSRehp61fXdTtmOAR3js+9wb73dwbYzr3kIc
3WH1
=UBcd
-----END PGP PRIVATE KEY BLOCK-----
"""
WRONG_TOKEN_SIGNATURE = """
-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1
{a95dc112-780e-47f7-a83a-c6f3820d7dc3}
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.12 (GNU/Linux)
iJwEAQECAAYFAlD7Jb0ACgkQydMyPCCvASv9HgQAhQf/oFMWcKwGncH0hjXC3QYz
7ck3chgL3S1pPAvS69viz6i2bwYZYD8fhzHNJ/qtw/rx6thO6PwT4SpdhKerap+I
kdem3LjM4fAGHRunHZYP39obNKMn1xv+f26mEAAWxdv/W/BLAFqxi3RijJywRkXm
zo5GUl844kpnV+uk0Xk=
=z2Cz
-----END PGP SIGNATURE-----
"""
FINGERPRINT = '2CF3E41500218D30F0B654F5C9D3323C20AF012B'
class UserTestCase(PluginTestCase): class UserTestCase(PluginTestCase):
plugins = ('User',) plugins = ('User',)
@ -142,7 +188,7 @@ class UserTestCase(PluginTestCase):
self.assertNotError('load Seen') self.assertNotError('load Seen')
self.assertResponse('user list', 'Foo') self.assertResponse('user list', 'Foo')
if network: if gpg.available and network:
def testGpgAddRemove(self): def testGpgAddRemove(self):
self.assertNotError('register foo bar') self.assertNotError('register foo bar')
self.assertError('user gpg add 51E516F0B0C5CE6A pgp.mit.edu') self.assertError('user gpg add 51E516F0B0C5CE6A pgp.mit.edu')
@ -155,6 +201,46 @@ class UserTestCase(PluginTestCase):
self.assertResponse('user gpg add EB17F1E0CEB63930 pgp.mit.edu', self.assertResponse('user gpg add EB17F1E0CEB63930 pgp.mit.edu',
'Error: This key is already associated with your account.') 'Error: This key is already associated with your account.')
if gpg.available:
def testGpgAuth(self):
self.assertNotError('register spam egg')
gpg.keyring.import_keys(PRIVATE_KEY).__dict__
(id, user) = ircdb.users.items()[0]
user.gpgkeys.append(FINGERPRINT)
msg = self.getMsg('gpg gettoken').args[-1]
match = re.search('is: ({.*}).', msg)
assert match, repr(msg)
token = match.group(1)
def fakeGetUrlFd(*args, **kwargs):
return fd
(utils.web.getUrlFd, realGetUrlFd) = (fakeGetUrlFd, utils.web.getUrlFd)
fd = StringIO()
fd.write('foo')
fd.seek(0)
self.assertResponse('gpg auth http://foo.bar/baz.gpg',
'Error: Signature or token not found.')
fd = StringIO()
fd.write(token)
fd.seek(0)
self.assertResponse('gpg auth http://foo.bar/baz.gpg',
'Error: Signature or token not found.')
fd = StringIO()
fd.write(WRONG_TOKEN_SIGNATURE)
fd.seek(0)
self.assertRegexp('gpg auth http://foo.bar/baz.gpg',
'Error: Unknown token.*')
fd = StringIO()
fd.write(str(gpg.keyring.sign(token)))
fd.seek(0)
self.assertResponse('gpg auth http://foo.bar/baz.gpg',
'You are now authenticated as spam.')
utils.web.getUrlFd = realGetUrlFd
# vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: # vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: