mrwnslz
04548171bf
Android und Desktop melden sich selbst bei Authelia an. Mit dem geprüften Ergebnis bekommen sie von diora den Zugang des verknüpften Kontos, so wie bisher nach der Anmeldung mit Passwort. Den Bücher-Schlüssel öffnen sie selbst, der Server sieht ihn nicht. Details: docs/accounts.md Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
321 lines
16 KiB
Python
321 lines
16 KiB
Python
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 ApiToken, 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')
|
|
|
|
|
|
@override_settings(OIDC_ISSUER=ISSUER, OIDC_CLIENT_ID='diora', OIDC_CLIENT_SECRET='s3cret',
|
|
OIDC_APP_CLIENT_ID='diora-app')
|
|
class OidcAppLoginTests(TestCase):
|
|
def setUp(self):
|
|
oidc._cache.clear()
|
|
self.user = User.objects.create_user(username='crm', password='pw12345678')
|
|
OidcIdentity.objects.create(user=self.user, issuer=ISSUER, sub='uuid-1')
|
|
p = patch('accounts.oidc.requests.get', side_effect=fake_get)
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
def _nonce(self):
|
|
cfg = self.client.get('/accounts/oidc/app/').json()
|
|
self.assertEqual((cfg['client_id'], cfg['token']), ('diora-app', DISCOVERY['token_endpoint']))
|
|
return cfg['nonce']
|
|
|
|
def _login(self, **claims):
|
|
c = {'iss': ISSUER, 'aud': ['diora-app'], 'sub': 'uuid-1', 'exp': time.time() + 300,
|
|
'nonce': self._nonce()}
|
|
c.update(claims)
|
|
return self.client.post('/accounts/oidc/app/login/', json.dumps({'id_token': make_token(c)}),
|
|
content_type='application/json')
|
|
|
|
def test_linked_account_gets_its_existing_token(self):
|
|
existing = ApiToken.objects.create(user=self.user)
|
|
resp = self._login()
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertEqual(resp.json(), {'token': existing.token, 'user_id': self.user.id,
|
|
'username': 'crm', 'sub': 'uuid-1'})
|
|
|
|
def test_token_is_made_when_there_is_none(self):
|
|
resp = self._login()
|
|
self.assertEqual(resp.json()['token'], ApiToken.objects.get(user=self.user).token)
|
|
|
|
def test_refused(self):
|
|
for name, claims, code in [
|
|
('unlinked', {'sub': 'other'}, 403),
|
|
('web client token', {'aud': 'diora'}, 401),
|
|
('own nonce', {'nonce': 'made-up'}, 401),
|
|
('expired', {'exp': time.time() - 3600}, 401),
|
|
]:
|
|
with self.subTest(name):
|
|
self.assertEqual(self._login(**claims).status_code, code)
|
|
self.assertEqual(self.client.post('/accounts/oidc/app/login/', '{}',
|
|
content_type='application/json').status_code, 400)
|
|
|
|
def test_old_nonce_is_refused(self):
|
|
nonce = self._nonce()
|
|
with patch('django.core.signing.time.time', return_value=time.time() + 3600):
|
|
resp = self.client.post('/accounts/oidc/app/login/', json.dumps({'id_token': make_token(
|
|
{'iss': ISSUER, 'aud': 'diora-app', 'sub': 'uuid-1', 'exp': time.time() + 7200, 'nonce': nonce})}),
|
|
content_type='application/json')
|
|
self.assertEqual(resp.status_code, 401)
|
|
|
|
def test_off_without_app_client(self):
|
|
with override_settings(OIDC_APP_CLIENT_ID=''):
|
|
self.assertEqual(self.client.get('/accounts/oidc/app/').json(), {'enabled': False})
|
|
|
|
|
|
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)
|