import re
import select
import socket
import ssl
from http.server import BaseHTTPRequestHandler
from subprocess import CalledProcessError, check_output
from threading import Lock
from typing import Any
from urllib.parse import urlparse

from botocore.awsrequest import AWSPreparedRequest

from moto.backend_index import backend_url_patterns
from moto.backends import get_backend
from moto.core import DEFAULT_ACCOUNT_ID
from moto.core.base_backend import BackendDict
from moto.core.exceptions import RESTError
from moto.core.utils import get_equivalent_url_in_aws_domain
from moto.moto_api._internal.models import moto_api_backend

from . import debug, error, info, with_color
from .certificate_creator import CertificateCreator
from .utils import get_body_from_form_data

# Adapted from https://github.com/xxlv/proxy3


class MotoRequestHandler:
    def __init__(self, port: int):
        self.lock = Lock()
        self.port = port

    def get_backend_for_host(self, host: str) -> Any:
        if host == f"http://localhost:{self.port}":
            return "moto_api"

        # Handle non-standard AWS endpoint hostnames from ISO regions or custom S3 endpoints.
        parsed_url, _ = get_equivalent_url_in_aws_domain(host)
        # Remove the querystring from the URL, as we'll never match on that
        clean_url = f"{parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}"

        for backend, pattern in backend_url_patterns:
            if pattern.match(clean_url):
                return backend

    def get_handler_for_host(self, host: str, path: str) -> Any:
        # We do not match against URL parameters
        path = path.split("?")[0]
        backend_name = self.get_backend_for_host(host)
        if not backend_name:
            return None
        backend_dict = get_backend(backend_name)

        # Get an instance of this backend.
        # We'll only use this backend to resolve the URL's, so the exact region/account_id is irrelevant
        if isinstance(backend_dict, BackendDict):
            if "us-east-1" in backend_dict[DEFAULT_ACCOUNT_ID]:
                backend = backend_dict[DEFAULT_ACCOUNT_ID]["us-east-1"]
            else:
                backend = backend_dict[DEFAULT_ACCOUNT_ID]["aws"]
        else:
            backend = backend_dict["global"]

        for url_path, handler in backend.url_paths.items():
            if re.match(url_path, path):
                return handler

        return None

    def parse_request(
        self,
        method: str,
        host: str,
        path: str,
        headers: Any,
        body: bytes,
        form_data: dict[str, Any],
    ) -> Any:
        handler = self.get_handler_for_host(host=host, path=path)
        if handler is None:
            return 404, {}, b"AWS Service not recognized or supported"
        full_url = host + path
        request = AWSPreparedRequest(
            method, full_url, headers, body, stream_output=False
        )
        request.form_data = form_data  # type: ignore[attr-defined]
        return handler(request, full_url, headers)


class ProxyRequestHandler(BaseHTTPRequestHandler):
    timeout = 5

    def __init__(self, *args: Any, **kwargs: Any):
        sock = [a for a in args if isinstance(a, socket.socket)][0]
        _, port = sock.getsockname()
        self.protocol_version = "HTTP/1.1"
        self.moto_request_handler = MotoRequestHandler(port)
        self.cert_creator = CertificateCreator()
        BaseHTTPRequestHandler.__init__(self, *args, **kwargs)

    @staticmethod
    def validate() -> None:
        debug("Starting initial validation...")
        CertificateCreator().validate()
        # Validate the openssl command is available
        try:
            debug("Verifying SSL version...")
            svn_output = check_output(["openssl", "version"])
            debug(svn_output)
        except CalledProcessError as e:
            info(e.output)
            raise

    def do_CONNECT(self) -> None:
        address = self.path.split(":")
        port = int(address[1]) or 443
        if address[0] in moto_api_backend.proxy_hosts_to_passthrough:
            self.connect_relay((address[0], port))
            return

        certpath = self.cert_creator.create(self.path)
        self.wfile.write(
            f"{self.protocol_version} 200 Connection Established\r\n".encode()
        )
        self.send_header("k", "v")
        self.end_headers()

        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ssl_context.load_cert_chain(
            keyfile=CertificateCreator.certkey,
            certfile=certpath,
        )
        ssl_context.check_hostname = False
        self.connection = ssl_context.wrap_socket(
            self.connection,
            server_side=True,
        )
        self.rfile = self.connection.makefile("rb", self.rbufsize)  # type: ignore
        self.wfile = self.connection.makefile("wb", self.wbufsize)  # type: ignore

        conntype = self.headers.get("Proxy-Connection", "")
        if self.protocol_version == "HTTP/1.1" and conntype.lower() != "close":
            self.close_connection = 0  # type: ignore
        else:
            self.close_connection = 1  # type: ignore

    def do_GET(self) -> None:
        req = self
        host, path = self._get_host_and_path(req)

        if f"{host}{path}" in moto_api_backend.proxy_urls_to_passthrough:
            parsed = urlparse(host)
            target_host, target_port = parsed.netloc, "80"
            if ":" in target_host:
                target_host, target_port = target_host.split(":")
            self.passthrough_http((target_host, int(target_port)))
            return

        req_body = b""
        if "chunked" in self.headers.get("Transfer-Encoding", ""):
            # https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding
            req_body = self.read_chunked_body(self.rfile)
        elif "Content-Length" in req.headers:
            content_length = int(req.headers["Content-Length"])
            req_body = self.rfile.read(content_length)
        if self.headers.get("Content-Type", "").startswith("multipart/form-data"):
            boundary = self.headers["Content-Type"].split("boundary=")[-1]
            req_body, form_data = get_body_from_form_data(req_body, boundary)  # type: ignore
            for key, val in form_data.items():
                self.headers[key] = [val]  # type: ignore
        else:
            form_data = {}

        req_body = self.decode_request_body(req.headers, req_body)  # type: ignore

        try:
            info(f"{with_color(33, req.command.upper())} {host}{path}")  # noqa
            if req_body is not None:
                debug("\tbody\t" + with_color(31, text=req_body))
            debug(f"\theaders\t{with_color(31, text=dict(req.headers))}")
            response = self.moto_request_handler.parse_request(
                method=req.command,
                host=host,
                path=path,
                headers=req.headers,
                body=req_body,
                form_data=form_data,
            )
            debug("\t=====RESPONSE========")
            debug("\t" + with_color(color=33, text=response))
            debug("\n")

            if isinstance(response, tuple):
                res_status, res_headers, res_body = response
            else:
                res_status, res_headers, res_body = (200, {}, response)

        except RESTError as e:
            if isinstance(e.get_headers(), list):
                res_headers = dict(e.get_headers())
            else:
                res_headers = e.get_headers()
            res_status = e.code
            res_body = e.get_body()

        except Exception as e:
            error(e)
            self.send_error(502)
            return

        res_reason = "OK"
        if isinstance(res_body, str):
            res_body = res_body.encode("utf-8")

        if "content-length" not in res_headers and res_body:
            res_headers["Content-Length"] = str(len(res_body))

        self.wfile.write(
            f"{self.protocol_version} {res_status} {res_reason}\r\n".encode()
        )
        if res_headers:
            for k, v in res_headers.items():
                if isinstance(v, bytes):
                    self.send_header(k, v.decode("utf-8"))
                else:
                    self.send_header(k, v)
            self.end_headers()
        if res_body:
            self.wfile.write(res_body)
        self.close_connection = True

    def _get_host_and_path(self, req: Any) -> tuple[str, str]:
        if isinstance(self.connection, ssl.SSLSocket):
            host = "https://" + req.headers["Host"]
        else:
            host = "http://" + req.headers["Host"]
        path = req.path
        if path.startswith(host):
            path = path[len(host) :]
        return host, path

    def passthrough_http(self, address: tuple[str, int]) -> None:
        s = socket.create_connection(address, timeout=self.timeout)
        s.send(self.raw_requestline)  # type: ignore[attr-defined]
        for key, val in self.headers.items():
            s.send(f"{key}: {val}\r\n".encode())
        s.send(b"\r\n")
        while True:
            data = s.recv(1024)
            if not data:
                break
            self.wfile.write(data)

    def connect_relay(self, address: tuple[str, int]) -> None:
        try:
            s = socket.create_connection(address, timeout=self.timeout)
        except Exception:
            self.send_error(502)
            return
        self.send_response(200, "Connection Established")
        self.end_headers()

        conns = [self.connection, s]
        self.close_connection = False
        while not self.close_connection:
            rlist, wlist, xlist = select.select(conns, [], conns, self.timeout)
            if xlist or not rlist:
                break
            for r in rlist:
                other = conns[1] if r is conns[0] else conns[0]
                data = r.recv(8192)
                if not data:
                    self.close_connection = True
                    break
                other.sendall(data)

    def read_chunked_body(self, reader: Any) -> bytes:
        chunked_body = b""
        while True:
            line = reader.readline().strip()
            chunk_length = int(line, 16)
            if chunk_length != 0:
                chunked_body += reader.read(chunk_length)

            # Each chunk is followed by an additional empty newline
            reader.readline()

            # a chunk size of 0 is an end indication
            if chunk_length == 0:
                # AWS does send additional (checksum-)headers, but we can ignore them
                break
        return chunked_body

    def decode_request_body(self, headers: dict[str, str], body: Any) -> Any:
        if body is None:
            return body
        if headers.get("Content-Type", "") in [
            "application/x-amz-json-1.1",
            "application/x-www-form-urlencoded; charset=utf-8",
        ]:
            return body.decode("utf-8")
        return body

    do_HEAD = do_GET
    do_POST = do_GET
    do_PUT = do_GET
    do_PATCH = do_GET
    do_DELETE = do_GET
    do_OPTIONS = do_GET