diora-web/accounts/oidc_views.py

"""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']})