import base64 import hashlib import json from datetime import timedelta from django.core.cache import cache from django.test import TestCase from django.urls import reverse from django.utils import timezone from users.auth_codes import CODE_TTL_SECONDS, hash_code from users.forms import LOCKOUT_ATTEMPTS from users.models import CustomUser, DesktopAuthCode VERIFIER = 'a-verifier-long-enough-to-be-a-real-one-0123456789' CHALLENGE = base64.urlsafe_b64encode( hashlib.sha256(VERIFIER.encode()).digest() ).rstrip(b'=').decode() PASSWORD = 'pw-for-tests-123' class DesktopAuthFlowTests(TestCase): def setUp(self): cache.clear() self.user = CustomUser.objects.create_user('signin@example.com', PASSWORD) self.start = reverse('auth-browser:desktop-sign-in') self.done = reverse('auth-browser:desktop-done') self.exchange = reverse('users:desktop-token') # --- the browser half ------------------------------------------------ def _begin(self, state='state-value', ip=None): return self.client.get( f'{self.start}?state={state}&code_challenge={CHALLENGE}&code_challenge_method=S256', headers={'x-forwarded-for': ip} if ip else {}, ) def test_sign_in_page_needs_state_and_challenge(self): """A bare visit is not a sign-in request, and must not 500.""" self.assertEqual(self.client.get(self.start).status_code, 400) def test_sign_in_page_rejects_a_plain_challenge_method(self): """S256 only. Accepting `plain` would make PKCE decorative.""" response = self.client.get( f'{self.start}?state=s&code_challenge={CHALLENGE}&code_challenge_method=plain' ) self.assertEqual(response.status_code, 400) def test_sign_in_mints_a_code_and_shows_it(self): self._begin() response = self.client.post( self.start, {'email': self.user.email, 'password': PASSWORD}, follow=True ) self.assertEqual(response.status_code, 200) entry = DesktopAuthCode.objects.get() self.assertEqual(entry.user, self.user) self.assertEqual(entry.code_challenge, CHALLENGE) # The plaintext is never stored, only shown once. self.assertNotIn(entry.code_hash, response.content.decode()) self.assertContains(response, 'lifey://auth/callback') def test_sign_in_is_case_insensitive_on_the_email(self): self._begin() response = self.client.post( self.start, {'email': 'SignIn@Example.COM', 'password': PASSWORD} ) self.assertEqual(response.status_code, 302) def test_a_wrong_password_does_not_mint_anything(self): self._begin() response = self.client.post(self.start, {'email': self.user.email, 'password': 'wrong'}) self.assertEqual(response.status_code, 200) self.assertFalse(DesktopAuthCode.objects.exists()) def test_an_unknown_address_reads_the_same_as_a_wrong_password(self): """Or the form becomes a way to ask who has an account.""" self._begin() unknown = self.client.post(self.start, {'email': 'nobody@example.com', 'password': 'wrong'}) self._begin() wrong = self.client.post(self.start, {'email': self.user.email, 'password': 'wrong'}) self.assertContains(unknown, 'Email or password is incorrect.') self.assertContains(wrong, 'Email or password is incorrect.') def test_repeated_failures_lock_the_address_out_across_ips(self): """ EVERY ATTEMPT COMES FROM A DIFFERENT IP, which is the whole point. core.middleware's limiter keys on the client address, so an attacker who rotates it gets a fresh bucket every time and that limiter never fires. The per-email counter in users/forms.py is keyed on the address being attacked, which is the part that cannot be rotated — so the right password is refused at the end even though no single IP ever made more than one attempt. """ for attempt in range(LOCKOUT_ATTEMPTS): ip = f'198.51.100.{attempt + 1}' self._begin(ip=ip) self.client.post( self.start, {'email': self.user.email, 'password': 'wrong'}, headers={'x-forwarded-for': ip}, ) fresh = '198.51.100.200' self._begin(ip=fresh) response = self.client.post( self.start, {'email': self.user.email, 'password': PASSWORD}, headers={'x-forwarded-for': fresh}, ) self.assertContains(response, 'Too many attempts') self.assertFalse(DesktopAuthCode.objects.exists()) def test_the_done_page_has_nothing_to_show_on_its_own(self): self.assertEqual(self.client.get(self.done).status_code, 400) # --- the exchange ---------------------------------------------------- def _mint(self): self._begin() self.client.post(self.start, {'email': self.user.email, 'password': PASSWORD}) page = self.client.get(self.done) return page.context['code'] def _post(self, **body): return self.client.post(self.exchange, data=json.dumps(body), content_type='application/json') def test_exchange_returns_both_tokens(self): response = self._post(code=self._mint(), code_verifier=VERIFIER) self.assertEqual(response.status_code, 200) self.assertIn('access', response.json()) self.assertIn('refresh', response.json()) def test_a_code_works_exactly_once(self): code = self._mint() self.assertEqual(self._post(code=code, code_verifier=VERIFIER).status_code, 200) self.assertEqual(self._post(code=code, code_verifier=VERIFIER).status_code, 400) def test_a_wrong_verifier_burns_the_code(self): """ The attacker holds the code and not the verifier. Leaving the row live would give them a second guess, which is the whole point of single-use. """ code = self._mint() self.assertEqual(self._post(code=code, code_verifier='not-it').status_code, 400) self.assertEqual(self._post(code=code, code_verifier=VERIFIER).status_code, 400) def test_an_expired_code_is_refused(self): code = self._mint() entry = DesktopAuthCode.objects.get() entry.created_at = timezone.now() - timedelta(seconds=CODE_TTL_SECONDS + 1) entry.save(update_fields=['created_at']) self.assertEqual(self._post(code=code, code_verifier=VERIFIER).status_code, 400) def test_every_failure_reads_the_same(self): """Unknown, expired, used and wrong-verifier must not be tellable apart.""" code = self._mint() used = self._post(code=code, code_verifier=VERIFIER) and \ self._post(code=code, code_verifier=VERIFIER) unknown = self._post(code='ZZZZZZZZ', code_verifier=VERIFIER) malformed = self._post(code='', code_verifier='') self.assertEqual(used.json(), unknown.json()) self.assertEqual(used.json(), malformed.json()) def test_the_code_is_stored_hashed(self): code = self._mint() entry = DesktopAuthCode.objects.get() self.assertNotEqual(entry.code_hash, code) self.assertEqual(entry.code_hash, hash_code(code)) def test_minting_sweeps_old_codes(self): stale = DesktopAuthCode.objects.create( code_hash='x' * 64, state='s', code_challenge=CHALLENGE, user=self.user, ) DesktopAuthCode.objects.filter(pk=stale.pk).update( created_at=timezone.now() - timedelta(hours=1) ) self._mint() self.assertFalse(DesktopAuthCode.objects.filter(pk=stale.pk).exists()) class AuthenticatedEndpointTests(TestCase): def setUp(self): self.user = CustomUser.objects.create_user('me@example.com', PASSWORD) self.ping = reverse('lifey_api:ping') self.me = reverse('users:me') def _bearer(self): from rest_framework_simplejwt.tokens import RefreshToken return f'Bearer {RefreshToken.for_user(self.user).access_token}' def test_ping_is_401_without_a_token(self): self.assertEqual(self.client.get(self.ping).status_code, 401) def test_ping_is_ok_with_a_token(self): response = self.client.get(self.ping, headers={'authorization': self._bearer()}) self.assertEqual(response.status_code, 200) self.assertEqual(response.json(), {'ok': True}) def test_me_returns_the_caller(self): response = self.client.get(self.me, headers={'authorization': self._bearer()}) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()['email'], 'me@example.com') def test_me_will_not_change_the_email(self): """Changing the address is changing the login; it needs its own flow.""" self.client.patch( self.me, data=json.dumps({'email': 'someone-else@example.com', 'first_name': 'Sam'}), content_type='application/json', headers={'authorization': self._bearer()}, ) self.user.refresh_from_db() self.assertEqual(self.user.email, 'me@example.com') self.assertEqual(self.user.first_name, 'Sam') def test_logout_blacklists_the_refresh_token(self): from rest_framework_simplejwt.tokens import RefreshToken refresh = RefreshToken.for_user(self.user) response = self.client.post( reverse('users:logout'), data=json.dumps({'refresh': str(refresh)}), content_type='application/json', headers={'authorization': f'Bearer {refresh.access_token}'}, ) self.assertEqual(response.status_code, 205) reused = self.client.post( reverse('token-refresh'), data=json.dumps({'refresh': str(refresh)}), content_type='application/json', ) self.assertEqual(reused.status_code, 401)