authentik/passbook/providers/oauth2/views/userinfo.py

"""passbook OAuth2 OpenID Userinfo views"""
from typing import Any, Dict, List

from django.http import HttpRequest, HttpResponse
from django.utils.translation import gettext_lazy as _
from django.views import View
from structlog import get_logger

from passbook.providers.oauth2.constants import (
    SCOPE_GITHUB_ORG_READ,
    SCOPE_GITHUB_USER,
    SCOPE_GITHUB_USER_EMAIL,
    SCOPE_GITHUB_USER_READ,
)
from passbook.providers.oauth2.models import RefreshToken, ScopeMapping
from passbook.providers.oauth2.utils import TokenResponse, cors_allow_any

LOGGER = get_logger()


class UserInfoView(View):
    """Create a dictionary with all the requested claims about the End-User.
    See: http://openid.net/specs/openid-connect-core-1_0.html#UserInfoResponse"""

    def get_scope_descriptions(self, scopes: List[str]) -> Dict[str, str]:
        """Get a list of all Scopes's descriptions"""
        scope_descriptions = {}
        for scope in ScopeMapping.objects.filter(scope_name__in=scopes).order_by(
            "scope_name"
        ):
            if scope.description != "":
                scope_descriptions[scope.scope_name] = scope.description
        # GitHub Compatibility Scopes are handeled differently, since they required custom paths
        # Hence they don't exist as Scope objects
        github_scope_map = {
            SCOPE_GITHUB_USER: _("GitHub Compatibility: Access your User Information"),
            SCOPE_GITHUB_USER_READ: _(
                "GitHub Compatibility: Access your User Information"
            ),
            SCOPE_GITHUB_USER_EMAIL: _(
                "GitHub Compatibility: Access you Email addresses"
            ),
            SCOPE_GITHUB_ORG_READ: _("GitHub Compatibility: Access your Groups"),
        }
        for scope in scopes:
            if scope in github_scope_map:
                scope_descriptions[scope] = github_scope_map[scope]
        return scope_descriptions

    def get_claims(self, token: RefreshToken) -> Dict[str, Any]:
        """Get a dictionary of claims from scopes that the token
        requires and are assigned to the provider."""

        scopes_from_client = token.scope
        final_claims = {}
        for scope in ScopeMapping.objects.filter(
            provider=token.provider, scope_name__in=scopes_from_client
        ).order_by("scope_name"):
            value = scope.evaluate(
                user=token.user,
                request=self.request,
                provider=token.provider,
                token=token,
            )
            if value is None:
                continue
            if not isinstance(value, dict):
                LOGGER.warning(
                    "Scope returned a non-dict value, ignoring",
                    scope=scope,
                    value=value,
                )
                continue
            LOGGER.debug("updated scope", scope=scope)
            final_claims.update(value)
        return final_claims

    def options(self, request: HttpRequest) -> HttpResponse:
        return cors_allow_any(self.request, TokenResponse({}))

    def get(self, request: HttpRequest, **kwargs) -> HttpResponse:
        """Handle GET Requests for UserInfo"""
        token: RefreshToken = kwargs["token"]
        claims = self.get_claims(token)
        claims["sub"] = token.id_token.sub
        response = TokenResponse(claims)
        cors_allow_any(self.request, response)
        return response

    def post(self, request: HttpRequest, **kwargs) -> HttpResponse:
        """POST Requests behave the same as GET Requests, so the get handler is called here"""
        return self.get(request, **kwargs)