import asyncio
import json
import logging
import re
import secrets
from ipaddress import ip_address, ip_network
from operator import attrgetter
from time import time
from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, List, Literal, Optional, Set

from sentry_sdk import capture_event
from sentry_sdk.utils import event_from_exception, exc_info_from_error
from starlette.applications import Starlette
from starlette.datastructures import URL
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import RedirectResponse, Response
from starlette.routing import get_name as get_endpoint_name

from . import glove
from .exceptions import UnexpectedResponse
from .utils import get_ip

logger = logging.getLogger('foxglove.middleware')
request_logger = logging.getLogger('foxglove.bad_requests')
CallNext = Callable[[Request], Awaitable[Response]]

__all__ = (
    'ErrorMiddleware',
    'CsrfMiddleware',
    'HostRedirectMiddleware',
    'CloudflareCheckMiddleware',
    'request_log_extra',
    'get_session_id',
    'update_session_id',
)


class ErrorMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: Starlette,
        should_warn: Callable[[Response], bool] = None,
        get_user: Callable[[Request], Awaitable[Dict[str, Any]]] = None,
    ):
        super().__init__(app)
        self.custom_should_warn = should_warn
        self.get_user = get_user

        from .main import glove

        self.glove = glove

    async def dispatch(self, request: Request, call_next: CallNext) -> Response:
        try:
            request.state.start_time = get_request_start(request)

            try:
                response = await call_next(request)
            except Exception as exc:
                await self.log(request, exc=exc)
                return Response('Internal Server Error', media_type='text/plain', status_code=500)
            else:
                if self.should_warn(response):
                    await self.log(request, response=response)
                return response
        except Exception:  # pragma: no cover
            # not sure if this is required, but better to keep it
            logger.critical('unhandled error in ErrorMiddleware', exc_info=True)
            raise

    async def log(
        self, request: Request, *, exc: Optional[Exception] = None, response: Optional[Response] = None
    ) -> None:
        event_data = await request_log_extra(request, exc, response)
        event_data['user'] = await self.user_info(request)
        view_ref = event_data['transaction']

        if exc:
            level = 'error'
            message = f'"{line_one(request)}", {exc!r}'
            fingerprint = view_ref, request.method, repr(exc)
            request_logger.exception(message, extra=event_data)
        else:
            assert response is not None
            level = 'warning'
            message = f'"{line_one(request)}", unexpected response: {response.status_code}'
            request_logger.warning(message, extra=event_data)
            fingerprint = view_ref, request.method, str(response.status_code)

        if glove.settings.sentry_dsn:
            hint = None
            if exc:
                exc_data, hint = event_from_exception(exc_info_from_error(exc))
                event_data.update(exc_data)

            event_data.update(message=message, level=level, logger='foxglove.request_errors', fingerprint=fingerprint)
            if not capture_event(event_data, hint):
                logger.critical(
                    'sentry not configured correctly, not sending message: %s',
                    message,
                    extra={'event_data': event_data},
                )

    def should_warn(self, response: Response) -> bool:
        if self.custom_should_warn:
            return self.custom_should_warn(response)
        else:
            return response.status_code > 310

    async def user_info(self, request: Request) -> Dict[str, Any]:
        user = dict(ip_address=get_ip(request))
        if get_user := self.get_user:
            try:
                user.update(await get_user(request))
            except Exception:
                logger.exception('error getting user for middleware logging')
        return user

    @staticmethod
    async def response_body(response: Response) -> bytes:
        if hasattr(response, 'body'):
            return response.body
        else:
            body_chunks = []
            async for chunk in response.body_iterator:
                if not isinstance(chunk, bytes):
                    chunk = chunk.encode(response.charset)
                body_chunks.append(chunk)

            response.body_iterator = async_gen_list(body_chunks)
            return b''.join(body_chunks)


async def request_log_extra(
    request: Request, exc: Optional[Exception] = None, response: Optional[Response] = None
) -> Dict[str, Any]:
    extra = dict(query=dict(request.query_params))

    if start_time := getattr(request.state, 'start_time', None):
        extra['duration'] = f'{(time() - start_time) * 1000:0.2f}ms'

    if endpoint := request.scope.get('endpoint'):
        extra.update(route_endpoint=get_endpoint_name(endpoint), path_params=dict(request.path_params))

    if exc:
        extra['exception_extra'] = exc_extra(exc)
    elif response:
        extra.update(
            response_status=response.status_code,
            response_headers=dict(response.headers),
            response_body=lenient_json(await get_response_body(response)),
        )

    return dict(
        extra=extra,
        user=dict(ip_address=get_ip(request)),
        transaction=re.sub(r'/\d{2,}/', '/{number}/', str(request.url.path)),
        request=dict(
            url=str(request.url),
            query_string=request.url.query,
            cookies=dict(request.cookies),
            headers=dict(request.headers),
            method=request.method,
            data=lenient_json(request.scope.get('_body')),
            inferred_content_type=request.headers.get('Content-Type'),
        ),
    )


async def get_response_body(response: Response) -> bytes:
    if hasattr(response, 'body'):
        return response.body
    else:
        body_chunks = []
        async for chunk in response.body_iterator:
            if not isinstance(chunk, bytes):
                chunk = chunk.encode(response.charset)
            body_chunks.append(chunk)

        response.body_iterator = async_gen_list(body_chunks)
        return b''.join(body_chunks)


async def async_gen_list(list_: List[bytes]) -> AsyncGenerator[bytes, None]:
    for c in list_:
        yield c


def line_one(request: Request) -> str:
    line = f'{request.method} {request.url.path}'
    if q := request.url.query:
        line += f'?{q}'
    return line


def get_request_start(request):
    try:
        return float(request.headers.get('X-Request-Start', '.')) / 1000
    except ValueError:
        return time()


def exc_extra(exc: Exception):
    exception_extra = getattr(exc, 'extra', None)
    if exception_extra:
        try:
            v = exception_extra()
        except Exception:
            pass
        else:
            return lenient_json(v)


def lenient_json(v: Any) -> Any:
    if isinstance(v, (str, bytes)):
        try:
            return json.loads(v)
        except (ValueError, TypeError):
            pass
    return v


session_id_key = 'session_id'
no_cookie_response = """\
{
  "message": "Permission Denied, no session set, updates not permitted"
}
"""
header_error_response = """\
{
  "message": "Permission Denied, %s"
}
"""


def get_session_id(request: Request) -> str:
    return request.session[session_id_key]


def update_session_id(request: Request) -> str:
    request.session[session_id_key] = new_session_id = secrets.token_urlsafe()
    return new_session_id


class CsrfMiddleware(BaseHTTPMiddleware):
    """
    Ensures a GET request has been made before post requests and that session_id is set in the session.

    This prevents CSRF especially if the cookie has same_site strict
    """

    def __init__(
        self,
        app: Starlette,
        *,
        should_check: Callable[[Request], bool] = None,
        enable_header_check: bool = None,
        allows_origins: Set[str] = None,
    ):
        super().__init__(app)
        self.should_check = should_check
        if enable_header_check is None:
            self.enable_header_check = not (glove.settings.dev_mode or glove.settings.test_mode)
        else:
            self.enable_header_check = enable_header_check
        self.allows_origins = glove.settings.origin if allows_origins is None else allows_origins

    async def dispatch(self, request: Request, call_next: CallNext) -> Response:
        if self.should_check and not self.should_check(request):
            return await call_next(request)

        session_id = request.session.get(session_id_key)
        benign_request = request.method in {'HEAD', 'GET', 'OPTIONS'}
        if not benign_request:
            if session_id is None:
                return Response(no_cookie_response, media_type='application/json', status_code=403)
            if error := self.header_check(request):
                return Response(header_error_response % error, media_type='application/json', status_code=403)

        response = await call_next(request)

        # set the session id for any valid GET request
        if benign_request and response.status_code == 200 and session_id is None:
            update_session_id(request)
        return response

    def header_check(self, request: Request) -> Optional[str]:
        """
        Origin and Referrer checks for CSRF, see
        https://cheatsheetseries.owasp.org/cheatsheets/
        Cross-Site_Request_Forgery_Prevention_Cheat_Sheet.html#verifying-origin-with-standard-headers
        """
        if not self.enable_header_check:
            return

        # origin header is optional (e.g. is omitted from firefox uploads)
        origin_ok = False
        if origin := request.headers.get('origin'):
            if origin not in self.allows_origins:
                return 'Incorrect Origin header'
            origin_ok = True

        if referrer := request.headers.get('referer'):
            referrer_url = URL(referrer)
            if f'{referrer_url.scheme}://{referrer_url.hostname}' not in self.allows_origins:
                return 'Incorrect Referrer header'
        elif not origin_ok:
            return 'Missing Origin and Referrer headers'


class HostRedirectMiddleware(BaseHTTPMiddleware):
    def __init__(self, app: Starlette, host: str = None):
        super().__init__(app)
        self.host = host or glove.settings.domain
        assert self.host, 'host must not be None in HostRedirectMiddleware'

    async def dispatch(self, request: Request, call_next: CallNext) -> Response:
        if request.url.hostname == self.host:
            return await call_next(request)
        else:
            return RedirectResponse(request.url.replace(hostname=self.host), status_code=301)


class IPRangeCounter:
    __slots__ = 'range', 'counter'

    def __init__(self, network_range: str):
        self.range = ip_network(network_range)
        self.counter = 0

    def __repr__(self):
        return f'IPRangeCounter({self.range}, {self.counter})'


class CloudflareCheckMiddleware(BaseHTTPMiddleware):
    default_response_body = b'Request incorrectly routed, this looks like a problem with your DNS or Proxy.'

    def __init__(self, app: Starlette, response_text: str = None):
        super().__init__(app)
        self.response_body = response_text.encode() if response_text else self.default_response_body
        self.ip_ranges: Optional[List[IPRangeCounter]] = None

    async def dispatch(self, request: Request, call_next: CallNext) -> Response:
        """
        On Heroku (and any properly configured system) we can trust the last entry in X-Forwarded-For,
        hence using that before trying client

        https://stackoverflow.com/a/37061471/949890
        """
        ip = None
        x_forwarded_for = request.headers.get('x-forwarded-for')
        if x_forwarded_for:
            ip = x_forwarded_for.rsplit(',', 1)[-1].strip()
        else:
            client = request.scope.get('client')
            if client:
                ip = client[0]

        if ip and await self.is_cloudflare_ip(ip):
            return await call_next(request)

        extra = {'extra': await request_log_extra(request), 'cf_ip_ranges': self.ip_ranges}
        logger.warning('Request not routed through CloudFlare ip=%s url="%s"', ip, request.url, extra=extra)
        return Response(self.response_body, status_code=400)

    async def is_cloudflare_ip(self, ip: str) -> bool:
        try:
            ip = ip_address(ip.strip())
        except ValueError:
            pass
        else:
            if self.ip_ranges is None:
                self.ip_ranges = ip_ranges = await get_cloudflare_ips()
            else:
                ip_ranges = self.ip_ranges

            for r in ip_ranges:
                if ip in r.range:
                    r.counter += 1
                    self.ip_ranges.sort(key=attrgetter('counter'), reverse=True)
                    return True
        return False


async def get_cloudflare_ips() -> List[IPRangeCounter]:
    """
    Get a list of cloudflare IPs from https://www.cloudflare.com/ips-v4 and https://www.cloudflare.com/ips-v6,
    see https://www.cloudflare.com/en-gb/ips/ for details.
    """

    async def get_ips(v: Literal[4, 6]) -> List[IPRangeCounter]:
        r = await glove.http.get(f'https://www.cloudflare.com/ips-v{v}', follow_redirects=True)
        UnexpectedResponse.check(r)
        return [IPRangeCounter(ip) for ip in r.text.strip().split('\n')]

    v4_ips, v6_ips = await asyncio.gather(get_ips(4), get_ips(6))
    ips = v4_ips + v6_ips
    logger.info('downloaded %d IPs from CloudFlare to check requests against', len(ips))
    return ips
