"""Sign-in with Authelia and linking it to an existing account (accounts/oidc.py).""" import json import logging import secrets from django.conf import settings from django.contrib import messages from django.contrib.auth import login from django.contrib.auth.decorators import login_required from django.core import signing from django.http import JsonResponse from django.shortcuts import redirect, render from django.utils import timezone from django.utils.http import url_has_allowed_host_and_scheme from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from . import oidc from .models import ApiToken, OidcIdentity log = logging.getLogger(__name__) SESSION_KEY = 'oidc_flow' def _redirect_uri(request): return settings.OIDC_REDIRECT_URI or request.build_absolute_uri('/accounts/oidc/callback/') def _safe_next(request, value): if value and url_has_allowed_host_and_scheme(value, allowed_hosts={request.get_host()}, require_https=request.is_secure()): return value return '/' def oidc_context(request): """What the settings page shows about the provider.""" return { 'oidc_enabled': oidc.enabled(), 'oidc_provider': settings.OIDC_PROVIDER_NAME, 'oidc_identities': list(request.user.oidc_identities.all()) if request.user.is_authenticated else [], } @require_http_methods(['GET']) def oidc_start(request): """Off to the provider. `?link=1` (signed in) links the account instead of signing in.""" if not oidc.enabled(): messages.error(request, 'Die Anmeldung über Authelia ist auf diesem Server nicht eingerichtet.') return redirect('login') link = request.GET.get('link') == '1' if link and not request.user.is_authenticated: return redirect('login') flow = oidc.new_flow() flow['link'] = link flow['next'] = _safe_next(request, request.GET.get('next', '')) try: url = oidc.authorize_url(flow, _redirect_uri(request)) except oidc.OidcError as exc: log.warning('oidc start: %s', exc) messages.error(request, 'Authelia ist gerade nicht erreichbar.') return redirect('settings' if link else 'login') request.session[SESSION_KEY] = flow return redirect(url) @require_http_methods(['GET']) def oidc_callback(request): flow = request.session.pop(SESSION_KEY, None) fail_to = 'login' if flow and flow.get('link'): fail_to = 'settings' if not flow or request.GET.get('state') != flow.get('state'): messages.error(request, 'Die Anmeldung über Authelia ist abgelaufen. Bitte noch einmal.') return redirect(fail_to) if request.GET.get('error'): messages.error(request, 'Authelia hat die Anmeldung abgebrochen.') return redirect(fail_to) code = request.GET.get('code', '') try: claims = oidc.exchange_code(code, flow, _redirect_uri(request)) except oidc.OidcError as exc: log.warning('oidc callback: %s', exc) messages.error(request, 'Die Anmeldung über Authelia hat nicht geklappt.') return redirect(fail_to) issuer = settings.OIDC_ISSUER.rstrip('/') sub = claims['sub'] shown = { 'username': (claims.get('preferred_username') or '')[:150], 'email': (claims.get('email') or '')[:254], } identity = OidcIdentity.objects.filter(issuer=issuer, sub=sub).select_related('user').first() if flow.get('link'): if not request.user.is_authenticated: return redirect('login') if identity and identity.user_id != request.user.id: messages.error(request, 'Dieses Authelia-Konto ist schon mit einem anderen diora-Konto verknüpft.') return redirect('settings') if identity is None: OidcIdentity.objects.create(user=request.user, issuer=issuer, sub=sub, **shown) else: OidcIdentity.objects.filter(pk=identity.pk).update(**shown) messages.success(request, f'Authelia-Konto „{shown["username"] or sub}“ verknüpft. ' 'Du kannst dich jetzt auch darüber anmelden.') if oidc.key_enabled(): # leave the book key with Authelia's help right away return redirect('/accounts/oidc/key/?next=/accounts/settings/') return redirect('settings') if identity is None: messages.error(request, 'Dieses Authelia-Konto ist mit keinem diora-Konto verknüpft. ' 'Melde dich einmal mit Passwort an und verknüpfe es in den Einstellungen.') return redirect('login') if not identity.user.is_active: messages.error(request, 'Dieses diora-Konto ist gesperrt.') return redirect('login') OidcIdentity.objects.filter(pk=identity.pk).update(last_login_at=timezone.now(), **shown) login(request, identity.user, backend='django.contrib.auth.backends.ModelBackend') # A password login brings the book key along, this one cannot: the device fetches it # (oidc_key below), else the page asks (static/js/app/33-unlock.js). next_url = flow.get('next') or '/' if oidc.key_enabled(): from urllib.parse import urlencode return redirect('/accounts/oidc/key/?' + urlencode({'next': next_url})) return redirect(next_url) @login_required @require_http_methods(['POST']) def oidc_unlink(request, pk): OidcIdentity.objects.filter(user=request.user, pk=pk).delete() messages.success(request, 'Verknüpfung mit Authelia gelöst.') return redirect('settings') @login_required @require_http_methods(['GET']) def oidc_key(request): """The book key with the provider's help, without asking (docs/accounts.md). Runs entirely on the device: it signs in once more as the public client (OIDC_APP_CLIENT_ID, PKCE, prompt=none — the provider session from just before makes it invisible), reads its `diora_key` claim from userinfo and with it opens the book key the server keeps wrapped (vault "authelia-key") — or, when the device has the key, wraps and leaves it there. The server never sees the claim; this view only hands out the page. The same page is the redirect target (`/accounts/oidc/key/callback/`).""" next_url = _safe_next(request, request.GET.get('next', '')) if not oidc.key_enabled(): return redirect(next_url) try: disco = oidc.discovery() except oidc.OidcError as exc: log.warning('oidc key: %s', exc) return redirect(next_url) config = { 'authorize': disco['authorization_endpoint'], 'token': disco['token_endpoint'], 'userinfo': disco.get('userinfo_endpoint', ''), 'client_id': settings.OIDC_APP_CLIENT_ID, 'redirect_uri': request.build_absolute_uri('/accounts/oidc/key/callback/') if not settings.OIDC_REDIRECT_URI else settings.OIDC_REDIRECT_URI.replace('/oidc/callback/', '/oidc/key/callback/'), 'subs': [i.sub for i in request.user.oidc_identities.all()], 'user_id': request.user.id, 'next': next_url, } return render(request, 'accounts/oidc_key.html', {'config': config}) # --------------------------------------------------------------------------- # The apps (Android, desktop): sign in with Authelia as the public client themselves # --------------------------------------------------------------------------- # # 1. GET /accounts/oidc/app/ -> where Authelia is, the client id and a nonce # signed by this server (good for APP_NONCE_AGE). # 2. The app runs the code flow with PKCE in the system browser (redirect to its own # scheme or loopback), trades the code itself and reads `diora_key` from userinfo. # 3. POST /accounts/oidc/app/login/ {id_token} -> the account's API token, as the # password login (diora-android Auth.kt) hands it out. # The book key never passes here: the app opens the vault "authelia-key" itself. APP_NONCE_SALT = 'diora.oidc.app' APP_NONCE_AGE = 600 @require_http_methods(['GET']) def app_config(request): if not oidc.key_enabled(): return JsonResponse({'enabled': False}) try: disco = oidc.discovery() except oidc.OidcError as exc: log.warning('oidc app config: %s', exc) return JsonResponse({'enabled': False, 'error': 'provider unreachable'}, status=502) return JsonResponse({ 'enabled': True, 'provider': settings.OIDC_PROVIDER_NAME, 'issuer': settings.OIDC_ISSUER.rstrip('/'), 'authorize': disco['authorization_endpoint'], 'token': disco['token_endpoint'], 'userinfo': disco.get('userinfo_endpoint', ''), 'client_id': settings.OIDC_APP_CLIENT_ID, 'scope': 'openid profile diora_key', 'nonce': signing.dumps(secrets.token_urlsafe(16), salt=APP_NONCE_SALT), }) def _app_nonce_ok(value): try: signing.loads(value, salt=APP_NONCE_SALT, max_age=APP_NONCE_AGE) return True except signing.BadSignature: return False @csrf_exempt # a native client without cookies; the signed id_token is the proof @require_http_methods(['POST']) def app_login(request): if not oidc.key_enabled(): return JsonResponse({'error': 'not enabled'}, status=404) try: body = json.loads(request.body or b'{}') id_token = body['id_token'] if not isinstance(id_token, str): raise TypeError except (ValueError, KeyError, TypeError): return JsonResponse({'error': 'id_token required'}, status=400) try: claims = oidc.verify_id_token(id_token, _app_nonce_ok, client_id=settings.OIDC_APP_CLIENT_ID) except oidc.OidcError as exc: log.warning('oidc app login: %s', exc) return JsonResponse({'error': 'Die Anmeldung über Authelia war ungültig.'}, status=401) identity = OidcIdentity.objects.filter( issuer=settings.OIDC_ISSUER.rstrip('/'), sub=claims['sub']).select_related('user').first() if identity is None: return JsonResponse({'error': 'Dieses Authelia-Konto ist mit keinem diora-Konto verknüpft. ' 'Verknüpfe es einmal im Browser unter Einstellungen.'}, status=403) if not identity.user.is_active: return JsonResponse({'error': 'Dieses diora-Konto ist gesperrt.'}, status=403) OidcIdentity.objects.filter(pk=identity.pk).update(last_login_at=timezone.now()) token, _ = ApiToken.objects.get_or_create(user=identity.user) return JsonResponse({'token': token.token, 'user_id': identity.user.id, 'username': identity.user.username, 'sub': claims['sub']})