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

from asgiref.sync import sync_to_async
from django import forms
from django.core.cache import cache
from django.core.handlers.wsgi import WSGIRequest
from django.http import HttpResponse
from django.shortcuts import render
from docker.models.containers import ExecResult
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 8 view."""

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


class HiddenSubmitTokenForm(forms.Form):
    """Form class used to submit completion token to the level."""

    token = forms.CharField(widget=forms.HiddenInput())


class DockerSQLError(Exception):
    """Error raised when SQL command fails in Docker container."""

    def __init__(self, message: str, command: str, result: ExecResult) -> None:
        self.result = result
        self.command = command
        super().__init__(message)

    @override
    def __str__(self) -> str:
        return (
            f"{super().__str__()}: [\n\tcommand={self.command!r}, \n\texit_code={self.result.exit_code}, \n\t"
            f"output={self.result.output.decode('utf-8')!r}\n]"
        )

    @override
    def __repr__(self) -> str:
        return f"DockerSQLError(message={self.args[0]!r}, command={self.command!r}, result={self.result!r})"


@final
class Level8DockerConsumer(DockerConsumer):
    """Docker WebSocket consumer for level 8."""

    _MYSQL_PASSWORD = "verysecret"  # noqa: S105
    CONSUMER_ID = "level8"
    DOCKER_IMAGE = "mysql"
    DOCKER_ENVIRONMENT: ClassVar[Mapping[str, str]] = {  # pyright: ignore[reportIncompatibleVariableOverride]
        "MYSQL_ROOT_PASSWORD": _MYSQL_PASSWORD,
    }
    DOCKER_HEALTHCHECK: ClassVar[Mapping[str, Any]] = {  # pyright: ignore[reportIncompatibleVariableOverride]
        "test": [
            "CMD",
            "mysql",
            "-h",
            "127.0.0.1",
            "-P",
            "3306",
            "-u",
            "root",
            f"--password={_MYSQL_PASSWORD}",
            "-e",
            "status",
        ],
        "start_period": 60_000_000_000,  # don't count failures during first 60 seconds
        "interval": 1_000_000_000,  # check every 1 second
        "timeout": 5_000_000_000,  # command execution can take at most 5 seconds
        "retries": 3,
    }

    async def run_sql(self, sql: str, without_database: bool = False, auto_error: bool = True) -> ExecResult:
        """Run SQL query in the MySQL database."""
        command = ["mysql", "-h", "127.0.0.1", "-P", "3306", "--user=root"]
        if not without_database:
            command.append("--database=vuln_demo")
        command.append("-e")
        command.append(sql)

        result = await sync_to_async(self.container.exec_run)(command, environment={"MYSQL_PWD": self._MYSQL_PASSWORD})

        if auto_error and result.exit_code != 0:
            raise DockerSQLError("Error running SQL command", sql, result)

        return result

    @override
    async def docker_prepare(self) -> bool:
        """Set up the docker container, creating basic database structure and populating it."""
        await super().docker_prepare()  # will wait until healthy

        await self.send_message("Creating user database...", "docker.waiting")
        await self.run_sql("CREATE DATABASE vuln_demo;", without_database=True)
        await self.run_sql("CREATE TABLE users (email VARCHAR(255) NOT NULL, password VARCHAR(255) NOT NULL);")

        entries, _ = get_cached_password_entries(self.user, "level8")
        for username, password in entries.items():
            await self.run_sql(f"INSERT INTO users (email, password) VALUES ('{username}', '{password}');")  # noqa: S608

        await self.send_message("Attempting connector connection...", "docker.waiting")

        return True

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

        if data["type"] != "login-data":
            # 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()
            return

        email = data.get("email", None)
        password = data.get("password", None)
        if email is None or password is None:
            await self.send_message("Missing username or password", "login.error")
            return

        query = f"SELECT * FROM users WHERE email='{email}' AND password='{password}'"  # noqa: S608 # this is indeed an SQL injection
        result = await self.run_sql(query, auto_error=False)

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

        # Many prorgrams just check that the result from select query is not empty
        # if it isn't, it means that the user exists and the password is correct
        if len(output) == 0:
            await self.send_message("Wrong email or password", "login.invalid_credentials")
            return

        # This will send a POST request to the server, which is then interpreted
        # from the Level8 class below.
        # To avoid spoofing this, we create a temporary flag stored in the cache
        # that will be checked in the Level8 class.
        token_cache_key = f"level8_token-{self.user.username}"
        token = str(secrets.randbits(32))
        cache.set(token_cache_key, token, timeout=60)  # A minute is more than enough
        await self.send_message("Login Successful", "login.success", token=token)


@final
class Level8(LevelABC):
    """Level 8 scenario, demonstrating SQL Injection vulnerability."""

    HINT_AMOUNT = 3

    @override
    @classmethod
    def _get_hint(cls, request: WSGIRequest, hint_num: int) -> str:
        if hint_num == 1:
            return textwrap.dedent(
                """
                <p>
                    In order to find whether a user exists, the server needs to check
                    with the database.
                </p>
                """
            )
        if hint_num == 2:
            return textwrap.dedent(
                """
                <p>
                    A basic knowlede of SQL commands can be very helpful.
                </p>
                """
            )
        if hint_num == 3:
            return textwrap.dedent(
                """
                <p>
                    Many servers simply check, if the result from a select query is not empty.
                </p>
                """
            )
        raise

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

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

    def post(self, request: WSGIRequest) -> HttpResponse:
        """
        Handle token form submission.

        Unlike the other levels, this one is a bit different. Pretty much all of the logic
        is handled via the DockerConsumer (websockets). This includes the submission form
        which only submissions reports back to the docker consumer.

        For that reason, this method will be used with a special hidden level submission
        form, which will be used to submit the completion token. This token is generated
        by the DockerConsumer and stored in the cache. This is done to avoid spoofing
        completions. If the submitted token matches the on in cache, the level is completed.
        """
        assert cast_user(request.user, cast=True)
        already_completed = CompletedLevel.check_completed(request.user, 8)

        form = HiddenSubmitTokenForm(request.POST)
        if not form.is_valid():
            return render(
                request,
                "vulnerabilities/levels/level8.html",
                {
                    "form": PasswordForm(),
                    "completed": already_completed,
                    "error_msg": "Submission form not valid. (Nice try)",
                },
            )

        token_cache_key = f"level8_token-{request.user.username}"
        expected_output = cache.get(token_cache_key)

        if expected_output is None:
            return render(
                request,
                "vulnerabilities/levels/level8.html",
                {
                    "form": PasswordForm(),
                    "completed": already_completed,
                    "error_msg": "Submission token not found in cache. (Nice try)",
                },
            )

        received_token = form.cleaned_data["token"]
        if received_token != expected_output:
            return render(
                request,
                "vulnerabilities/levels/level8.html",
                {
                    "form": PasswordForm(),
                    "completed": already_completed,
                    "error_msg": "Invliad submission token. (Nice try)",
                },
            )

        reset_password_entries_cache(request.user, "level8")
        return self.completed_redirect(request, 8, skip_db=already_completed)
