import base64 import json import time from unittest.mock import patch from urllib.parse import parse_qs, urlsplit from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding, rsa from django.contrib.auth.models import User from django.test import TestCase, override_settings from . import oidc from .models import KeyShare, OidcIdentity ISSUER = 'https://auth.example.org' PRIV = rsa.generate_private_key(public_exponent=65537, key_size=2048) OTHER_PRIV = rsa.generate_private_key(public_exponent=65537, key_size=2048) def _b64url(b): return base64.urlsafe_b64encode(b).rstrip(b'=').decode() def _int_b64(i): return _b64url(i.to_bytes((i.bit_length() + 7) // 8, 'big')) def make_token(claims, key=PRIV, alg='RS256', kid='k1'): header = _b64url(json.dumps({'alg': alg, 'kid': kid, 'typ': 'JWT'}).encode()) payload = _b64url(json.dumps(claims).encode()) sig = key.sign(f'{header}.{payload}'.encode(), padding.PKCS1v15(), hashes.SHA256()) return f'{header}.{payload}.{_b64url(sig)}' DISCOVERY = { 'issuer': ISSUER, 'authorization_endpoint': ISSUER + '/api/oidc/authorization', 'token_endpoint': ISSUER + '/api/oidc/token', 'jwks_uri': ISSUER + '/jwks.json', } pub = PRIV.public_key().public_numbers() JWKS = {'keys': [{'kty': 'RSA', 'kid': 'k1', 'n': _int_b64(pub.n), 'e': _int_b64(pub.e)}]} class FakeResp: def __init__(self, data, status=200): self._data, self.status_code = data, status def json(self): return self._data def raise_for_status(self): if self.status_code >= 400: raise oidc.requests.HTTPError(str(self.status_code)) def fake_get(url, timeout=None): return FakeResp(DISCOVERY if url.endswith('openid-configuration') else JWKS) @override_settings(OIDC_ISSUER=ISSUER, OIDC_CLIENT_ID='diora', OIDC_CLIENT_SECRET='s3cret', OIDC_REDIRECT_URI='https://diora.test/accounts/oidc/callback/') class OidcFlowTests(TestCase): def setUp(self): oidc._cache.clear() self.user = User.objects.create_user(username='crm', password='pw12345678') self.get = patch('accounts.oidc.requests.get', side_effect=fake_get) self.get.start() self.addCleanup(self.get.stop) def _start(self, query=''): resp = self.client.get('/accounts/oidc/start/' + query) self.assertEqual(resp.status_code, 302) q = parse_qs(urlsplit(resp['Location']).query) self.assertEqual(q['code_challenge_method'], ['S256']) self.assertEqual(q['client_id'], ['diora']) return q['state'][0], q['nonce'][0] def _callback(self, state, claims, token_status=200): token = make_token(claims) with patch('accounts.oidc.requests.post', return_value=FakeResp({'id_token': token}, token_status)) as post: resp = self.client.get('/accounts/oidc/callback/', {'state': state, 'code': 'c0de'}) return resp, post def _claims(self, nonce, **extra): c = {'iss': ISSUER, 'aud': ['diora'], 'sub': 'uuid-1', 'exp': time.time() + 300, 'nonce': nonce, 'preferred_username': 'marwin', 'email': 'm@example.org'} c.update(extra) return c def test_unlinked_identity_is_turned_away(self): state, nonce = self._start() resp, _ = self._callback(state, self._claims(nonce)) self.assertEqual(resp['Location'], '/accounts/login/') self.assertNotIn('_auth_user_id', self.client.session) self.assertFalse(User.objects.exclude(pk=self.user.pk).exists()) def test_link_then_login(self): self.client.force_login(self.user) state, nonce = self._start('?link=1') resp, post = self._callback(state, self._claims(nonce)) self.assertEqual(resp['Location'], '/accounts/settings/') ident = OidcIdentity.objects.get() self.assertEqual((ident.user, ident.sub, ident.username), (self.user, 'uuid-1', 'marwin')) # client_secret_basic + PKCE verifier self.assertEqual(post.call_args.kwargs['auth'], ('diora', 's3cret')) self.assertIn('code_verifier', post.call_args.kwargs['data']) self.client.logout() state, nonce = self._start('?next=/books/') resp, _ = self._callback(state, self._claims(nonce)) self.assertEqual(resp['Location'], '/books/') self.assertEqual(int(self.client.session['_auth_user_id']), self.user.pk) ident.refresh_from_db() self.assertIsNotNone(ident.last_login_at) def test_next_to_other_host_is_dropped(self): OidcIdentity.objects.create(user=self.user, issuer=ISSUER, sub='uuid-1') state, nonce = self._start('?next=https://evil.example/') resp, _ = self._callback(state, self._claims(nonce)) self.assertEqual(resp['Location'], '/') def test_identity_of_other_account_is_not_moved(self): other = User.objects.create_user(username='bob', password='pw12345678') OidcIdentity.objects.create(user=other, issuer=ISSUER, sub='uuid-1') self.client.force_login(self.user) state, nonce = self._start('?link=1') self._callback(state, self._claims(nonce)) self.assertEqual(OidcIdentity.objects.get().user, other) def test_wrong_state_nonce_audience_or_signature_fail(self): OidcIdentity.objects.create(user=self.user, issuer=ISSUER, sub='uuid-1') cases = [ ('state', {}), ('nonce', {'nonce': 'other'}), ('aud', {'aud': 'someone-else'}), ('iss', {'iss': 'https://evil.example'}), ('exp', {'exp': time.time() - 3600}), ] for name, extra in cases: with self.subTest(name): state, nonce = self._start() claims = {**self._claims(nonce), **extra} resp, _ = self._callback('wrong' if name == 'state' else state, claims) self.assertEqual(resp['Location'], '/accounts/login/') self.assertNotIn('_auth_user_id', self.client.session) state, nonce = self._start() with patch('accounts.oidc.requests.post', return_value=FakeResp({'id_token': make_token(self._claims(nonce), key=OTHER_PRIV)})): self.client.get('/accounts/oidc/callback/', {'state': state, 'code': 'x'}) self.assertNotIn('_auth_user_id', self.client.session) def test_state_is_single_use(self): OidcIdentity.objects.create(user=self.user, issuer=ISSUER, sub='uuid-1') state, nonce = self._start() self._callback(state, self._claims(nonce)) self.client.logout() resp, _ = self._callback(state, self._claims(nonce)) self.assertNotIn('_auth_user_id', self.client.session) def test_unlink(self): ident = OidcIdentity.objects.create(user=self.user, issuer=ISSUER, sub='uuid-1') other = User.objects.create_user(username='bob', password='pw12345678') foreign = OidcIdentity.objects.create(user=other, issuer=ISSUER, sub='uuid-2') self.client.force_login(self.user) self.client.post(f'/accounts/oidc/{foreign.pk}/unlink/') self.client.post(f'/accounts/oidc/{ident.pk}/unlink/') self.assertEqual(list(OidcIdentity.objects.all()), [foreign]) def test_with_app_client_login_and_link_go_by_the_key_page(self): OidcIdentity.objects.create(user=self.user, issuer=ISSUER, sub='uuid-1') with override_settings(OIDC_APP_CLIENT_ID='diora-app'): state, nonce = self._start('?next=/books/') resp, _ = self._callback(state, self._claims(nonce)) self.assertEqual(resp['Location'], '/accounts/oidc/key/?next=%2Fbooks%2F') page = self.client.get(resp['Location']) self.assertContains(page, 'oidc-key-config') cfg = page.context['config'] self.assertEqual(cfg['client_id'], 'diora-app') self.assertEqual(cfg['redirect_uri'], 'https://diora.test/accounts/oidc/key/callback/') self.assertEqual(cfg['subs'], ['uuid-1']) self.assertEqual(cfg['next'], '/books/') self.assertEqual(cfg['token'], DISCOVERY['token_endpoint']) self.assertEqual(self.client.get('/accounts/oidc/key/callback/?code=x&state=y').status_code, 200) state, nonce = self._start('?link=1') resp, _ = self._callback(state, self._claims(nonce)) self.assertEqual(resp['Location'], '/accounts/oidc/key/?next=/accounts/settings/') def test_key_page_without_app_client_goes_straight_on(self): self.client.force_login(self.user) resp = self.client.get('/accounts/oidc/key/?next=/books/') self.assertEqual(resp['Location'], '/books/') def test_login_page_shows_button_only_when_configured(self): self.assertContains(self.client.get('/accounts/login/'), 'Mit Authelia anmelden') with override_settings(OIDC_CLIENT_SECRET=''): self.assertNotContains(self.client.get('/accounts/login/'), 'Mit Authelia anmelden') PUB = base64.b64encode(b'\x04' + b'\x01' * 64).decode() class KeyShareTests(TestCase): def setUp(self): self.user = User.objects.create_user(username='crm', password='pw12345678') self.client.force_login(self.user) def _post(self, url, body): return self.client.post(url, json.dumps(body), content_type='application/json') def test_round_trip_hands_answer_out_once(self): token = self._post('/accounts/keyshare/', {'pub': PUB, 'device': 'Firefox auf Linux'}).json()['token'] self.assertEqual(self.client.get(f'/accounts/keyshare/{token}/').json(), {'state': 'pending'}) pending = self.client.get('/accounts/keyshare/pending/').json()['requests'] self.assertEqual([(r['token'], r['pub'], r['device']) for r in pending], [(token, PUB, 'Firefox auf Linux')]) answer = {'pub': PUB, 'iv': base64.b64encode(b'i' * 12).decode(), 'ct': base64.b64encode(b'c' * 48).decode()} self.assertEqual(self._post(f'/accounts/keyshare/{token}/answer/', answer).status_code, 200) self.assertEqual(self._post(f'/accounts/keyshare/{token}/answer/', answer).status_code, 409) self.assertEqual(self.client.get('/accounts/keyshare/pending/').json()['requests'], []) got = self.client.get(f'/accounts/keyshare/{token}/').json() self.assertEqual(got, {'state': 'answered', **answer}) self.assertEqual(self.client.get(f'/accounts/keyshare/{token}/').json(), {'state': 'expired'}) def test_requests_of_devices_that_stopped_asking_are_not_offered(self): token = self._post('/accounts/keyshare/', {'pub': PUB}).json()['token'] KeyShare.objects.update(polled_at=KeyShare.objects.get().created_at.replace(year=2000)) self.assertEqual(self.client.get('/accounts/keyshare/pending/').json()['requests'], []) self.client.get(f'/accounts/keyshare/{token}/') # asks again self.assertEqual(len(self.client.get('/accounts/keyshare/pending/').json()['requests']), 1) def test_decline(self): token = self._post('/accounts/keyshare/', {'pub': PUB}).json()['token'] self._post(f'/accounts/keyshare/{token}/answer/', {'decline': True}) self.assertEqual(self.client.get(f'/accounts/keyshare/{token}/').json(), {'state': 'declined'}) def test_bad_input_is_refused(self): self.assertEqual(self._post('/accounts/keyshare/', {'pub': 'nope'}).status_code, 400) token = self._post('/accounts/keyshare/', {'pub': PUB}).json()['token'] bad = {'pub': PUB, 'iv': base64.b64encode(b'i' * 12).decode(), 'ct': base64.b64encode(b'c' * 10).decode()} self.assertEqual(self._post(f'/accounts/keyshare/{token}/answer/', bad).status_code, 400) def test_other_accounts_see_nothing(self): token = self._post('/accounts/keyshare/', {'pub': PUB}).json()['token'] other = User.objects.create_user(username='bob', password='pw12345678') self.client.force_login(other) self.assertEqual(self.client.get('/accounts/keyshare/pending/').json()['requests'], []) self.assertEqual(self._post(f'/accounts/keyshare/{token}/answer/', {'decline': True}).status_code, 404) self.client.logout() self.assertEqual(self.client.get('/accounts/keyshare/pending/').status_code, 401) def test_open_requests_are_limited_and_run_out(self): tokens = [self._post('/accounts/keyshare/', {'pub': PUB}).json()['token'] for _ in range(7)] self.assertEqual(KeyShare.objects.count(), 5) self.assertEqual(self.client.get(f'/accounts/keyshare/{tokens[0]}/').json(), {'state': 'expired'}) self.assertEqual(self.client.get(f'/accounts/keyshare/{tokens[-1]}/').json(), {'state': 'pending'}) KeyShare.objects.update(created_at=KeyShare.objects.first().created_at.replace(year=2000)) self.assertEqual(self.client.get('/accounts/keyshare/pending/').json()['requests'], []) self.assertEqual(self._post('/accounts/keyshare/', {'pub': PUB}).status_code, 200)