# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.
"""Transparently add scopes to URLs."""

import base64
import binascii
import datetime as dt
import re
from typing import TYPE_CHECKING

import django.http
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured, PermissionDenied
from django.utils import timezone
from django.utils.deprecation import MiddlewareMixin
from rest_framework.authentication import get_authorization_header

from debusine.db.permissioncontext import PermissionContext
from debusine.server.scopes import get_scope_urlconf

if TYPE_CHECKING:
    from debusine.db.models import Scope, Token

scope_prefix_re = re.compile(r'^/([^/]+)(/|$)')


class ScopeMiddleware(MiddlewareMixin):
    """
    Extract the current scope from the URL prefix.

    If used, it must be sequenced before
    django.middleware.common.CommonMiddleware, since it can make use of URL
    resolution.
    """

    def process_request(self, request: django.http.HttpRequest) -> None:
        """Process request in middleware."""
        from debusine.db.models.scopes import is_valid_scope_name

        scope_name: str | None = None

        if mo := scope_prefix_re.match(request.path_info):
            if mo.group(1) == "api":
                # /api/ gets special treatment, as scope can also be specified
                # in a header
                scope_name = request.headers.get("x-debusine-scope")
            elif is_valid_scope_name(mo.group(1)):
                scope_name = mo.group(1)

        if scope_name is None:
            scope_name = settings.DEBUSINE_DEFAULT_SCOPE

        request.debusine_scope = self.get_scope(scope_name)  # type: ignore[attr-defined]
        # VirtualHostMiddleware may have set this already; if so, leave it
        # alone, as the urlconfs it sets don't need to be per-scope.
        if getattr(request, "urlconf", None) is None:
            setattr(request, "urlconf", get_scope_urlconf(scope_name))

    def get_scope(self, name: str) -> "Scope":
        """Set the current scope to the given named one."""
        from django.shortcuts import get_object_or_404

        from debusine.db.models import Scope

        return get_object_or_404(Scope, name=name)


TOKEN_LAST_SEEN_AT_UPDATE_INTERVAL = dt.timedelta(minutes=5)


class AuthorizationMiddleware(MiddlewareMixin):
    """
    Check user access to the current scope.

    If used, it must be sequenced after
    django.contrib.auth.middleware.AuthenticationMiddleware, since it needs the
    current user, and after
    debusine.server.middlewares.token_last_seen_at.TokenLastSeenAtMiddleware
    to validate the access of the worker token.
    """

    def _basic_auth_allowed(self, request: django.http.HttpRequest) -> bool:
        """Check if we should attempt HTTP basic authentication."""
        return request.scheme == "https"

    def _get_token_from_basic_auth(
        self, request: django.http.HttpRequest
    ) -> "Token | None":
        """
        Get a Debusine token sent as basic auth credentials.

        This is used to allow apt to pass Debusine tokens for repositories in
        private workspaces for archive views.
        """
        from debusine.db.models import Token

        if not getattr(request, "is_archive_view", False):
            return None

        # Force a Django request to d-r-f's get_authorization_header, which we
        # can do as it only really checks request.META
        auth = get_authorization_header(
            request  # type: ignore[arg-type]
        ).split(None, 1)
        if not auth or auth[0].lower() != b'basic':
            return None

        if not self._basic_auth_allowed(request):
            raise PermissionDenied(
                "Authenticated access is only allowed using HTTPS"
            )

        if len(auth) == 1:
            raise PermissionDenied("No credentials provided")

        try:
            try:
                auth_decoded = base64.b64decode(auth[1]).decode('utf-8')
            except UnicodeDecodeError:
                auth_decoded = base64.b64decode(auth[1]).decode('latin-1')

            userid, password = auth_decoded.split(':', 1)
        except (TypeError, ValueError, UnicodeDecodeError, binascii.Error):
            raise PermissionDenied("Credentials string is malformed")

        # Ensure that a token, and only the right kind of token, is passed via
        # basic auth
        token = Token.objects.get_token_or_none(password)
        if token is not None:
            token_username = (
                token.user.username
                if token.user is not None
                else token.external_username
            )
            if token_username == userid:
                return token
        # Intentionally give the same error message for all cases to avoid
        # leaking information on failed attempts.
        raise PermissionDenied("Invalid token")

    def _get_token_from_header(
        self, request: django.http.HttpRequest
    ) -> "Token | None":
        """Get a Debusine token from a token header."""
        from debusine.db.models import Token

        if (token_key := request.headers.get("token")) is None:
            return None

        return Token.objects.get_token_or_none(token_key)

    def _validate_and_cache_token(
        self, request: django.http.HttpRequest
    ) -> None:
        """Validate and cache the token in the request."""
        from debusine.db.models import Token

        setattr(request, "_debusine_token", None)

        token: Token | None = self._get_token_from_header(request)
        if token is None:
            token = self._get_token_from_basic_auth(request)
        if token is None:
            return

        now = timezone.now()
        if (
            token.last_seen_at is None
            or token.last_seen_at < now - TOKEN_LAST_SEEN_AT_UPDATE_INTERVAL
        ):
            token.last_seen_at = now
            token.save(update_fields=["last_seen_at"])

        if not token.enabled:
            return

        setattr(request, "_debusine_token", token)

    def _get_permission_context(
        self, request: django.http.HttpRequest
    ) -> PermissionContext:
        """
        Create the permission context for a request.

        :returns: the permission context
        :raises PermissionDenied: in case of inconsistencies
        """
        from debusine.db.models import Token, WorkRequest

        if (token := getattr(request, "_debusine_token", None)) is None:
            if request.user.is_authenticated:
                return PermissionContext.for_ui_user(request.user)
            else:
                return PermissionContext.for_ui_anonymous()

        match token.token_type:
            case Token.Types.WORKER_ACTIVATION:
                return PermissionContext.for_worker_activation(token)
            case Token.Types.WORKER:
                return PermissionContext.for_worker(token)
            case Token.Types.WORK_REQUEST:
                assert token.work_request is not None
                if token.work_request.status != WorkRequest.Statuses.RUNNING:
                    raise PermissionDenied("work request is not running")
                return PermissionContext.for_work_request(token=token)
            case Token.Types.USER:
                if request.user.is_authenticated:
                    raise PermissionDenied(
                        "cannot use both Django and user token authentication"
                    )

                if not token.user.is_active:
                    raise PermissionDenied("user token has an inactive user")

                return PermissionContext.for_api_user(token)
            case Token.Types.EXTERNAL_USER:
                if request.user.is_authenticated:
                    raise PermissionDenied(
                        "cannot use both Django and user token authentication"
                    )

                return PermissionContext.for_api_external_user(token)
            case _ as unreachable:
                # Enforced by database constraint.
                raise AssertionError(f"Unexpected token type: {unreachable}")

    def process_request(
        self, request: django.http.HttpRequest
    ) -> django.http.HttpResponseForbidden | None:
        """Process request in middleware."""
        from debusine.db.context import ContextConsistencyError, context
        from debusine.db.models import Scope

        if not hasattr(request, "debusine_scope"):
            raise ImproperlyConfigured(
                "AuthorizationMiddleware needs to be called"
                " after ScopeMiddleware"
            )
        assert isinstance(request.debusine_scope, Scope)

        try:
            self._validate_and_cache_token(request)
        except PermissionDenied as e:
            return django.http.HttpResponseForbidden(str(e))

        # request.user may be a lazy object, e.g. as set up by
        # AuthenticationMiddleware.  In that case we must force it to be
        # evaluated here, as otherwise asgiref.sync.AsyncToSync may try to
        # evaluate it when restoring context and will raise a
        # SynchronousOnlyOperation exception.
        request.user.username

        try:
            pc = self._get_permission_context(request)
        except PermissionDenied as e:
            return django.http.HttpResponseForbidden(str(e))

        try:
            context.set_permission_context_scoped(request.debusine_scope, pc)
        except ContextConsistencyError as e:
            return django.http.HttpResponseForbidden(str(e))

        return None
