import textwrap
from collections.abc import Mapping
from typing import Any, final

from asgiref.sync import sync_to_async
from django import forms
from django.core.handlers.wsgi import WSGIRequest
from django.http import HttpResponse
from django.shortcuts import render
from typing_extensions import override

from vuln_demo.apps.vulnerabilities.levels.abc import DockerConsumer, LevelABC
from vuln_demo.apps.vulnerabilities.models import CompletedLevel
from vuln_demo.util.auth import cast_user
from vuln_demo.util.pw_entries import ADMIN_EMAIL, get_cached_password_entries, reset_password_entries_cache


class PasswordForm(forms.Form):
    """Form class passed to level 7 view."""

    email = forms.EmailField(label="Email", initial=ADMIN_EMAIL)
    password = forms.CharField(label="Password", widget=forms.PasswordInput(attrs={"autofocus": "autofocus"}))


@final
class Level7DockerConsumer(DockerConsumer):
    """Docker WebSocket consumer for level 7."""

    CONSUMER_ID = "level7"
    DOCKER_IMAGE = "debian"
    # This will make the container run forever (otherwise it would exit immediately)
    DOCKER_COMMAND = "tail -f /dev/null"

    @override
    async def docker_prepare(self) -> bool:
        """
        Set up the docker container, creating a password file inside of it.

        The user is expected to obtain this password file, and extract the admin password
        from it, which will allow them to pass this level.
        """
        await super().docker_prepare()

        pw_entries, _ = await sync_to_async(get_cached_password_entries)(self.user, "level7")
        pw_file = "\n".join([f"{email}:{password}" for email, password in pw_entries.items()])

        result = await sync_to_async(self.container.exec_run)(["mkdir", "-p", "/cybervault"])
        if result.exit_code != 0:
            raise

        # We will write the password file line by line, since Docker SDK doesn't support
        # `docker cp` command, and since we only have a string, it's not trivial to escape
        # it properly to write it in a single command.
        for line in pw_file.splitlines():
            # This does a good enough job of escaping the line for our purposes
            # repr will also wrap the string in quotes, so it should be fine to pass to echo
            escaped_line = repr(line)
            result = await sync_to_async(self.container.exec_run)(
                [
                    "bash",
                    "-c",
                    f"echo {escaped_line} >> /cybervault/passwords.txt",
                ]
            )
            if result.exit_code != 0:
                raise

        return True

    @override
    async def process_message(self, data: Mapping[str, Any]) -> None:
        self.container.reload()
        if self.container.status != "running":
            await self.send_message("Container has stopped", "docker.error")
            await self.close()
            return

        if data["type"] == "tz-convert":
            timezone = data.get("timezone", None)
            time = data.get("time", None)
            if timezone is None or time is None:
                await self.send_message("Missing timezone or time", "tz-convert.error")
                return

            await self.send_message("Running command: date", "tz-convert.pending_exec")
            # NOTE: This is intentionally insecure, allowing RCE in the container
            # date --date="TZ=\"Asia/Taipei\" $TIME" +%F\ %H:%M
            result = await sync_to_async(self.container.exec_run)(
                [
                    "timeout",
                    "5s",
                    "bash",
                    "-c",
                    f'date --date="TZ=\\"{timezone}\\" {time}" +%F\\ %H:%M',
                ]
            )

            output = result.output.decode("utf-8")
            if result.exit_code != 0:
                await self.send_message(
                    f"Error running command, got code <code>{result.exit_code}</code>. "
                    "Please report this to CyberVault developers.",
                    "tz-convert.exec_error",
                    output=output,
                )
                return

            await self.send_message(output, "tz-convert.exec_result")
            return

        # We don't necessarily need to close the connection here, but the
        # fron-end implementation expects a close after getting consumer.error
        # we could implement a different type and handle it in the front-end
        # but it's not really that important for us now.
        await self.send_message("Unknown message type", "consumer.error")
        await self.close()


@final
class Level7(LevelABC):
    """Level 7 scenario, demonstrating Remote Code Execution vulnerability."""

    HINT_AMOUNT = 4

    @override
    @classmethod
    def _get_hint(cls, request: WSGIRequest, hint_num: int) -> str:
        if hint_num == 1:
            return textwrap.dedent(
                """
                <p>
                    This level is about the inner processing of the data you
                    send to the server. Think of how a time converter widget
                    like this might work.
                </p>
                """
            )
        if hint_num == 2:
            return textwrap.dedent(
                """
                <p>
                    A basic knowledge of UNIX/Linux commands can be very helpful.
                </p>
                """
            )
        if hint_num == 3:
            return textwrap.dedent(
                """
                <p>
                    What would happen if the server simply passed your input
                    into a shell command?
                </p>
                """
            )
        if hint_num == 4:
            return textwrap.dedent(
                """
                <p>
                    The admin password is stored in a file on the server. This
                    file is used to hold all of the user credentials. Can you
                    find a way to locate and read this file?
                </p>
                """
            )

        raise

    @override
    def get(self, request: WSGIRequest) -> HttpResponse:
        assert cast_user(request.user, cast=True)
        already_completed = CompletedLevel.check_completed(request.user, 7)

        form = PasswordForm()
        return render(
            request,
            "vulnerabilities/levels/level7.html",
            {
                "form": form,
                "completed": already_completed,
            },
        )

    def post(self, request: WSGIRequest) -> HttpResponse:
        """Handle password form submission."""
        assert cast_user(request.user, cast=True)
        already_completed = CompletedLevel.check_completed(request.user, 7)

        form = PasswordForm(request.POST)
        if not form.is_valid():
            return render(
                request,
                "vulnerabilities/levels/level7.html",
                {
                    "form": form,
                    "completed": already_completed,
                },
            )

        email = form.cleaned_data["email"]
        if email != ADMIN_EMAIL:
            return render(
                request,
                "vulnerabilities/levels/level7.html",
                {
                    "form": form,
                    "completed": already_completed,
                    "error_msg": "You're only interested in getting into the admin account.",
                },
            )

        pw = form.cleaned_data["password"]
        _, admin_password = get_cached_password_entries(request.user, "level7")
        if pw != admin_password:
            form.add_error("password", "Incorrect password")
            return render(
                request,
                "vulnerabilities/levels/level7.html",
                {"form": form, "completed": already_completed},
            )

        reset_password_entries_cache(request.user, "level7")
        return self.completed_redirect(request, 7, skip_db=already_completed)
