validate id_token

Signature is validated, exp, aud and iis fields are checked.

Also add tests using tox and py.test. Proper validation of signature is verified
using jwcrypto.

getlasso.sh

@@ -0,0 +1,20 @@
+#!/bin/sh
+
+# Get venv site-packages path
+DSTDIR=`python -c 'from distutils.sysconfig import get_python_lib; print(get_python_lib())'`
+
+# Get not venv site-packages path
+# Remove first path (assuming that is the venv path)
+NONPATH=`echo $PATH | sed 's/^[^:]*://'`
+SRCDIR=`PATH=$NONPATH python -c 'from distutils.sysconfig import get_python_lib; print(get_python_lib())'`
+
+# Clean up
+rm -f $DSTDIR/lasso.*
+rm -f $DSTDIR/_lasso.*
+
+# Link
+ln -sv $SRCDIR/lasso.py $DSTDIR
+ln -sv $SRCDIR/_lasso.* $DSTDIR
+
+exit 0
+

src/authentic2_auth_fc/models.py

@@ -1,10 +1,18 @@
 import base64
 import json
+import hmac
+import hashlib
+import urlparse
 
 from django.db import models
 from django.utils.translation import ugettext_lazy as _
+from django.utils.timezone import now
 from django.conf import settings
 
+from authentic2_auth_oidc.utils import parse_timestamp
+
+from . import app_settings
+
 
 def base64url_decode(input):
     rem = len(input) % 4
@@ -13,10 +21,36 @@ def base64url_decode(input):
     return base64.urlsafe_b64decode(input)
 
 
-def parse_id_token(id_token):
-    payload = id_token.split('.')[1]
+def parse_id_token(id_token, client_id=None, client_secret=None):
+    try:
+        splitted = str(id_token).split('.')
+    except:
+        return None, 'invalid id_token'
+    if len(splitted) != 3:
+        return None, 'invalid id_token'
+    header, payload, signature = splitted
+    try:
+        signature = base64url_decode(signature)
+    except (ValueError, TypeError):
+        return None, 'invalid signature'
+    signed = '%s.%s' % (header, payload)
+    if client_secret is not None:
+        h = hmac.HMAC(key=client_secret, msg=signed, digestmod=hashlib.sha256)
+        if h.digest() != signature:
+            return None, 'hmac signature does not match'
     payload = base64url_decode(str(payload))
-    return json.loads(payload)
+    try:
+        payload = json.loads(payload)
+    except ValueError:
+        return None, 'invalid payload'
+    if client_id and ('aud' not in payload or payload['aud'] != client_id):
+        return None, 'invalid audience'
+    if 'exp' not in payload or parse_timestamp(payload['exp']) < now():
+        return None, 'id_token is expired'
+    parsed = urlparse.urlparse(app_settings.authorize_url)
+    if 'iss' not in payload or payload['iss'] != '%s://%s/' % (parsed.scheme, parsed.netloc):
+        return None, 'wrong issuer received, %r' % payload['iss']
+    return payload, None
 
 
 class FcAccount(models.Model):

src/authentic2_auth_fc/views.py

@@ -249,8 +249,15 @@ class FcOAuthSessionViewMixin(LoggerMixin):
                 self.logger.warning(msg)
                 messages.warning(request, _('Unable to connect to FranceConnect.'))
                 return self.redirect(request)
-
-            self.id_token = models.parse_id_token(self.token['id_token'])
+            key = app_settings.client_secret
+            if isinstance(key, unicode):
+                key = key.encode('utf-8')
+            self.id_token, error = models.parse_id_token(
+                self.token['id_token'], client_id=app_settings.client_id, client_secret=key)
+            if not self.id_token:
+                self.logger.warning(u'validation of id_token failed: %s', error)
+                messages.warning(request, _('Unable to connect to FranceConnect.'))
+                return self.redirect(request)
             nonce = self.id_token.get('nonce')
             states = request.session.get('fc-states', {})
             if not nonce or nonce not in states:

tests/conftest.py

@@ -0,0 +1,20 @@
+import pytest
+import django_webtest
+
+
+@pytest.fixture
+def app(request, db):
+    wtm = django_webtest.WebTestMixin()
+    wtm._patch_settings()
+    request.addfinalizer(wtm._unpatch_settings)
+    return django_webtest.DjangoTestApp(extra_environ={'HTTP_HOST': 'localhost'})
+
+
+@pytest.fixture
+def fc_settings(settings):
+    settings.A2_FC_ENABLE = True
+    settings.A2_FC_CLIENT_ID = 'xxx'
+    settings.A2_FC_CLIENT_SECRET = 'yyy'
+    return settings
+
+

tests/settings.py

@@ -0,0 +1,11 @@
+import os
+
+LANGUAGE_CODE = 'en'
+DATABASES = {
+    'default': {
+        'ENGINE': os.environ.get('DB_ENGINE', 'django.db.backends.sqlite3'),
+        'TEST': {
+            'NAME': 'a2-test',
+        },
+    }
+}

tests/test_auth_fc.py

@@ -0,0 +1,100 @@
+import pytest
+import urlparse
+import httmock
+import json
+import base64
+from jwcrypto import jwk, jwt
+import datetime
+
+from django.core.urlresolvers import reverse
+from django.contrib.auth import get_user_model
+from django.utils.timezone import now
+
+from authentic2.utils import timestamp_from_datetime
+
+
+User = get_user_model()
+
+
+def hmac_jwt(payload, key):
+    header = {'alg': 'HS256'}
+    k = jwk.JWK(kty='oct', k=base64.b64encode(key.encode('utf-8')))
+    t = jwt.JWT(header=header, claims=payload)
+    t.make_signed_token(k)
+    return t.serialize()
+
+
+def test_login_redirect(app, fc_settings):
+    url = reverse('fc-login-or-link')
+    response = app.get(url, status=302)
+    assert response['Location'].startswith('https://fcp.integ01')
+
+
+def check_authorization_url(url):
+    callback = reverse('fc-login-or-link')
+    assert url.startswith('https://fcp.integ01')
+    query_string = url.split('?')[1]
+    parsed = {x: y[0] for x, y in urlparse.parse_qs(query_string).items()}
+    assert 'redirect_uri' in parsed
+    assert callback in parsed['redirect_uri']
+    assert 'client_id' in parsed
+    assert parsed['client_id'] == 'xxx'
+    assert 'scope' in parsed
+    assert set(parsed['scope'].split()) == set(['openid', 'profile', 'birth', 'email'])
+    assert 'state' in parsed
+    assert 'nonce' in parsed
+    assert parsed['state'] == parsed['nonce']
+    assert 'response_type' in parsed
+    assert parsed['response_type'] == 'code'
+    return parsed['state']
+
+
+@pytest.mark.parametrize('exp', [timestamp_from_datetime(now() + datetime.timedelta(seconds=1000)),
+                                 timestamp_from_datetime(now() - datetime.timedelta(seconds=1000))])
+def test_login(app, fc_settings, caplog, exp):
+    callback = reverse('fc-login-or-link')
+    response = app.get(callback, status=302)
+    location = response['Location']
+    state = check_authorization_url(location)
+
+    @httmock.urlmatch(path=r'.*/token$')
+    def access_token_response(url, request):
+        parsed = {x: y[0] for x, y in urlparse.parse_qs(request.body).items()}
+        assert set(parsed.keys()) == set(['code', 'client_id', 'client_secret', 'redirect_uri',
+                                          'grant_type'])
+        assert parsed['code'] == 'zzz'
+        assert parsed['client_id'] == 'xxx'
+        assert parsed['client_secret'] == 'yyy'
+        assert parsed['grant_type'] == 'authorization_code'
+        assert callback in parsed['redirect_uri']
+        id_token = {
+            'sub': '1234',
+            'aud': 'xxx',
+            'nonce': state,
+            'exp': exp,
+            'iss': 'https://fcp.integ01.dev-franceconnect.fr/',
+        }
+        return json.dumps({
+            'access_token': 'uuu',
+            'id_token': hmac_jwt(id_token, 'yyy')
+        })
+
+    @httmock.urlmatch(path=r'.*userinfo$')
+    def user_info_response(url, request):
+        assert request.headers['Authorization'] == 'Bearer uuu'
+        return json.dumps({
+            'sub': '1234',
+            'family_name': 'Doe',
+            'given_name': 'John',
+        })
+
+    with httmock.HTTMock(access_token_response, user_info_response):
+        response = app.get(callback + '?code=zzz&state=%s' % state, status=302)
+    assert User.objects.count() == 0
+    fc_settings.A2_FC_CREATE = True
+    with httmock.HTTMock(access_token_response, user_info_response):
+        response = app.get(callback + '?code=zzz&state=%s' % state, status=302)
+    if exp < timestamp_from_datetime(now()):
+        assert User.objects.count() == 0
+    else:
+        assert User.objects.count() == 1

tox.ini

@@ -0,0 +1,46 @@
+# Tox (http://tox.testrun.org/) is a tool for running tests
+# in multiple virtualenvs. This configuration file will run the
+# test suite on all supported python versions. To use it, "pip install tox"
+# and then run "tox" from this directory.
+
+[tox]
+toxworkdir = {env:TMPDIR:/tmp}/tox-{env:USER}/authentic2-auth-fc/
+envlist = {coverage,nocoverage}-{dj18}-{pg,sqlite}
+
+[testenv]
+# django.contrib.auth is not tested it does not work with our templates
+whitelist_externals =
+  /bin/mv
+setenv =
+  AUTHENTIC2_SETTINGS_FILE=tests/settings.py
+  DJANGO_SETTINGS_MODULE=authentic2.settings
+  sqlite: DB_ENGINE=django.db.backends.sqlite3
+  pg: DB_ENGINE=django.db.backends.postgresql_psycopg2
+  coverage: COVERAGE=--junitxml=junit-{envname}.xml --cov-report xml --cov=src/ --cov-config .coveragerc
+  fast: FAST=--nomigrations
+usedevelop =
+  coverage: True
+  nocoverage: False
+deps =
+  dj18: django>1.8,<1.9
+  dj19: django>1.8,<1.9
+  pg: psycopg2<2.7
+  coverage
+  pytest-cov
+  pytest-django
+  mock
+  pytest
+  lxml
+  cssselect
+  pylint
+  pylint-django
+  django-webtest
+  WebTest
+  pyquery
+  httmock
+  pytest-catchlog
+  pytz
+  ../authentic2
+commands =
+  ./getlasso.sh
+  py.test {env:FAST:} {env:COVERAGE:} {posargs:tests/}