Merge lp:~canonical-isd-hackers/canonical-identity-provider/yubi-internal-validation into lp:~canonical-isd-hackers/canonical-identity-provider/add-yubikey-support

[Łukasz Czyżykowski]

Added support for internal OTP validation.

Overview
========
This branch adds support for internal validation of OTPs.

Details
=======
Added new model ``identityprovider.models.yubi.YubiKey`` which
contains ``validate_otp(otp)`` class method. This will return boolean
if the OTP is valid for the keys registered in the server.

Also added full test suite, including code to programatically
generating OTPs.

[Michael Foord (mfoord)]

Nice work.

review: Approve

=== modified file 'identityprovider/admin.py'
--- identityprovider/admin.py	2010-11-11 16:14:06 +0000
+++ identityprovider/admin.py	2011-05-12 08:45:28 +0000
@@ -8,7 +8,7 @@
 from identityprovider.const import SREG_LABELS
 from identityprovider.fields import CommaSeparatedField
 from identityprovider.models import (Account, AccountPassword, EmailAddress,
-                                     OpenIDRPConfig, APIUser)
+                                     OpenIDRPConfig, APIUser, YubiKey)
 from identityprovider.models.const import AccountStatus, EmailStatus
 from identityprovider.utils import encrypt_launchpad_password
 from identityprovider.widgets import (StatusWidget, ReadOnlyDateTimeWidget,
@@ -215,3 +215,11 @@
 admin.site.register(OpenIDRPConfig, OpenIDRPConfigAdmin)
 admin.site.register(APIUser, APIUserAdmin)
 admin.site.register(EmailAddress, EmailAddressAdmin)
+
+
+class YubiKeyAdmin(admin.ModelAdmin):
+    list_display = ("publicname", "created", "active")
+    search_fields = ("publicname",)
+    list_filter = ("active", "created")
+
+admin.site.register(YubiKey, YubiKeyAdmin)

=== modified file 'identityprovider/models/__init__.py'
--- identityprovider/models/__init__.py	2011-05-10 14:32:26 +0000
+++ identityprovider/models/__init__.py	2011-05-12 08:45:28 +0000
@@ -10,3 +10,4 @@
 from openidmodels import *
 from team import *
 from api import *
+from yubi import *

=== added file 'identityprovider/models/yubi.py'
--- identityprovider/models/yubi.py	1970-01-01 00:00:00 +0000
+++ identityprovider/models/yubi.py	2011-05-12 08:45:28 +0000
@@ -0,0 +1,121 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+import re
+
+from Crypto.Cipher import AES
+from string import maketrans
+from django.db import models
+
+
+class OTPValidator(object):
+
+    otp_re = re.compile(
+        r'''
+            (?P<userid>[cbdefghijklnrtuv]{0,16})
+            (?P<token>[cbdefghijklnrtuv]{32})
+        ''',
+        re.VERBOSE
+    )
+
+    def validate(self, otp):
+        if len(otp) > 48 or len(otp) < 32:
+            return False
+        match = self.otp_re.match(otp)
+        if not match:
+            return False
+
+        userid = match.groupdict()['userid']
+        token = self.modhex2hex(match.groupdict()['token'])
+
+        yks = YubiKey.objects.filter(publicname=userid, active=True)
+        if yks.count() != 1:
+            return False
+
+        yk = yks[0]
+
+        plaintext = self.aes128ecb_decrypt(yk.aeskey, token)
+        uid = plaintext[:12]
+
+        if yk.internalname != uid:
+            return False
+
+        if not self.crc_is_valid(plaintext):
+            return False
+
+        internal_counter = self.hexdec(
+            plaintext[14:16] + plaintext[12:14])
+        timestamp = self.hexdec(
+            plaintext[20:22] + plaintext[18:20] + plaintext[16:18])
+
+        if yk.counter >= internal_counter:
+            return False
+
+        if yk.time >= timestamp:
+            if (yk.counter >> 8) == (internal_counter >> 8):
+                return False
+
+        yk.counter = internal_counter
+        yk.time = timestamp
+        yk.save()
+
+        return True
+
+    def modhex2hex(self, string):
+        return string.translate(
+            maketrans("cbdefghijklnrtuv", "0123456789abcdef"))
+
+    def hexdec(self, hexstr):
+        return int(hexstr, 16)
+
+    def crc(self, s):
+        crc = 0xffff
+        for i in range(0, 16):
+            b = self.hexdec(s[i * 2] + s[(i * 2) + 1])
+            crc = crc ^ (b & 0xff)
+            for j in range(0, 8):
+                n = crc & 1
+                crc = crc >> 1
+                if n != 0:
+                    crc = crc ^ 0x8408
+        return crc
+
+    def crc_is_valid(self, token):
+        crc = self.crc(token)
+        return crc == 0xf0b8
+
+    def aes128ecb_decrypt(self, aeskey, aesdata):
+        aes = AES.new(aeskey.decode('hex'), AES.MODE_ECB)
+        return aes.decrypt(aesdata.decode('hex')).encode('hex')
+
+
+
+class YubiKey(models.Model):
+
+    publicname = models.CharField(
+        max_length=12, unique=True,
+        help_text="The key public identity (the beginning of the OTPs)")
+    created = models.DateTimeField(auto_now_add=True, editable=False)
+    internalname = models.CharField(
+        max_length=12,
+        help_text="The private identity of the OTP")
+    aeskey = models.CharField(max_length=32, help_text="AES Key")
+    active = models.BooleanField(default=True)
+    counter = models.IntegerField(default=1, editable=False)
+    time = models.IntegerField(default=1, editable=False)
+
+    class Meta:
+        app_label = 'identityprovider'
+        db_table = u'yubikey'
+
+    @classmethod
+    def sanitize_otp(cls, otp):
+        if re.match(r"[jxe.uidchtnbpygk]+", otp):
+            return otp.translate(
+                maketrans("jxe.uidchtnbpygk", "cbdefghijklnrtuv"))
+        return otp
+
+    @classmethod
+    def validate_otp(cls, otp):
+        otp = cls.sanitize_otp(otp)
+        return OTPValidator().validate(otp)

=== added file 'identityprovider/tests/test_models_yubi.py'
--- identityprovider/tests/test_models_yubi.py	1970-01-01 00:00:00 +0000
+++ identityprovider/tests/test_models_yubi.py	2011-05-12 08:45:28 +0000
@@ -0,0 +1,156 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+from random import randint
+from Crypto.Cipher import AES
+from string import maketrans
+
+from django.test import TestCase
+
+from identityprovider.models.yubi import OTPValidator, YubiKey
+
+
+
+class OTPValidatorTestCase(TestCase):
+
+    def setUp(self):
+        self.validator = OTPValidator()
+
+    def test_validate_otp_fails_too_short(self):
+        self.assertFalse(self.validator.validate('cbdefghijklnrtuv'))
+
+    def test_validate_otp_fails_too_long(self):
+        self.assertFalse(self.validator.validate('cbdefghijklnrtuv' * 10))
+
+    def test_validate_otp_fails_match_regexp(self):
+        self.assertFalse(self.validator.validate('a' * 44))
+
+    def test_modhex2hex(self):
+        self.assertEquals(self.validator.modhex2hex("fibb"), "4711")
+
+    def test_hexdec(self):
+        self.assertEquals(self.validator.hexdec("ff"), 255)
+        self.assertEquals(self.validator.hexdec("0a"), 10)
+
+
+
+class YubiKeyTestCase(TestCase):
+
+    def compute_crc(self, plaintext):
+        crc = 0xffff
+        for v in plaintext:
+            crc ^= ord(v)
+            for i in range(0, 8):
+                n = crc & 1
+                crc >>= 1
+                if n != 0:
+                    crc ^= 0x8408
+        crc = ~crc & 0xffff
+        return "%02x%02x" % (crc & 0xff, crc >> 8)
+
+    def create_otp(self, key, counter=None, time=None, session=1):
+        counter = (key.counter + 1) if counter is None else counter
+        time = (key.time + 1) if time is None else time
+
+        counter = counter
+        plaintext = "".join([
+            "%12s" % key.internalname,
+            "%02x" % (counter & 0xff), # 12:14
+            "%02x" % ((counter >> 8) & 0xff), # 14:16
+            "%02x" % (time & 0xff), # 16:18
+            "%02x" % ((time & 0xff00) >> 8), # 18:20
+            "%02x" % ((time & 0xff0000) >> 16), # 20:22
+            "%02x" % (session & 0xff), # 22:24
+            "%04x" % randint(0, 0xFFFF) # 24:28
+        ])
+
+        crc = self.compute_crc(plaintext.decode("hex"))
+        plaintext = plaintext + crc
+
+        assert len(plaintext) == 32
+
+        aes = AES.new(key.aeskey.decode("hex"), AES.MODE_ECB)
+        token = aes.encrypt(plaintext.decode("hex")).encode("hex")
+
+        assert len(token) == 32
+
+        token = token.translate(
+            maketrans("0123456789abcdef", "cbdefghijklnrtuv"))
+        otp = key.publicname + token
+
+        assert len(otp) == 44
+
+        return otp
+
+    def create_key(self, publicname=None, internalname=None, aeskey=None,
+                   **kwargs):
+        defaults = {}
+        if publicname is None:
+            defaults['publicname'] = "vvkdtkjureru"
+
+        if internalname is None:
+            defaults['internalname'] = "980a8608b307"
+
+        if aeskey is None:
+            defaults['aeskey'] = "f1dc9c6585d600d06f9aae1abea2969e"
+
+        defaults.update(kwargs)
+        return YubiKey.objects.create(**defaults)
+
+    def test_otp_validates_properly(self):
+        self.create_key()
+
+        # The otp below is generated by yubikey dongle
+        self.assertTrue(YubiKey.validate_otp(
+            "vvkdtkjurerunkdcnkinjvgidtttggvegjdekdvhugvf"))
+
+    def test_otp_fails_because_key_is_not_in_the_db(self):
+        self.assertFalse(YubiKey.validate_otp(
+            "vvkdtkjurerunkdcnkinjvgidtttggvegjdekdvhugvf"))
+
+    def test_otp_fails_uid_doesnt_match(self):
+        self.create_key(internalname="980a8608b307")
+
+        self.assertFalse(YubiKey.validate_otp(
+            "vvkdtkjurerunkdcnkinjvgidtttggvegjdekdvhugvf"))
+
+    def test_otp_fails_on_crc_validation(self):
+        key = self.create_key()
+        otp = self.create_otp(key)
+        # Setting obviously invalid CRC bit
+        otp = otp[:-4] + "cccc"
+        self.assertFalse(YubiKey.validate_otp(otp))
+
+    def test_otp_fails_if_counter_is_too_late(self):
+        key = self.create_key(counter=5001)
+        otp = self.create_otp(key, counter=5000)
+
+        self.assertFalse(YubiKey.validate_otp(otp))
+
+    def test_otp_fails_if_timestamp_is_too_late(self):
+        key = self.create_key(time=10)
+        otp = self.create_otp(key, time=9)
+
+        self.assertFalse(YubiKey.validate_otp(otp))
+
+    def test_otp_fails_if_used_twice(self):
+        key = self.create_key()
+        otp = self.create_otp(key)
+
+        self.assertTrue(YubiKey.validate_otp(otp))
+        self.assertFalse(YubiKey.validate_otp(otp))
+
+    def test_otp_validates_when_entered_with_dvorak(self):
+        self.create_key()
+        otp = 'kkteythgp.pgpedkucydtbxkpkdkpgxejubytiuubntb'
+        self.assertTrue(YubiKey.validate_otp(otp))
+
+    def test_sanitize_otp_leaves_alone_normal_otp(self):
+        otp = "vvkdtkjurerunkdcnkinjvgidtttggvegjdekdvhugvf"
+        self.assertEquals(YubiKey.sanitize_otp(otp), otp)
+
+    def test_sanitize_otp_properly_handles_dvorak_keyboard(self):
+        otp = 'kkteythgp.pgpedkucydtbxkpkdkpgxejubytiuubntb'
+        self.assertEquals(
+            YubiKey.sanitize_otp(otp),
+            'vvkdtkjurerurdhvfithknbvrvhvrubdcfntkgffnlkn')

=== modified file 'requirements.txt'
--- requirements.txt	2011-03-04 16:49:25 +0000
+++ requirements.txt	2011-05-12 08:45:28 +0000
@@ -44,6 +44,7 @@
 zope.tales==3.5.0
 zope.testing==3.8.3
 zope.traversing==3.12.0
+pycrypto==2.3
 
 # project dependencies
 -f http://launchpad.net/lazr.restfulclient/trunk/0.9.11/+download/lazr.restfulclient-0.9.11.tar.gz#egg=lazr.restfulclient-0.9.11