# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import grp
import json
import logging
import os
import pwd
import socket
import struct
from functools import wraps
from http import HTTPStatus
from typing import Callable, Optional, Tuple, Any

from clcommon.cpapi import get_main_username_by_uid
from clcommon.lib.cledition import is_cl_solo_edition, is_cl_shared_pro_edition

from xray import gettext as _
from .constants import user_tasks_count, fpm_reload_timeout
from .exceptions import XRayAPIError, XRayError, XRayMissingDomain
from .fpm_utils import FPMReloadController
from .nginx_utils import NginxUserCache

# --------- GLOBALS ---------

logger = logging.getLogger('user_plugin_utils')
_format = '>I'


# --------- FUNCTIONS ---------


def pack_request(_input: Any) -> bytes:
    """
    Pack input for sending with length-prefix framing
    """
    data = json.dumps(_input).encode()
    return struct.pack(_format, len(data)) + data


def unpack_request(byte_command: bytes) -> Any:
    """
    Unpack incoming command
    """
    parsed = json.loads(byte_command.decode())
    logger.info('Command requested => %s', parsed.get('command'))
    return parsed


def pack_response(msg: bytes) -> bytes:
    """
    Prefix message with a 4-byte length
    """
    logger.debug('Packing message of %i length', len(msg))
    return struct.pack(_format, len(msg)) + msg


def unpack_response(sock_object: socket.socket) -> bytes:
    """
    Read length-prefixed amount of data from socket
    """
    chunk = 4096
    max_msglen = 10 * 1024 * 1024
    msg = bytes()
    raw_msglen = sock_object.recv(4)
    if not raw_msglen:
        return sock_object.recv(chunk)
    msglen = struct.unpack(_format, raw_msglen)[0]

    if msglen > max_msglen:
        raise ConnectionError('Response message too large: %d bytes' % msglen)

    while len(msg) < msglen:
        part = sock_object.recv(chunk)
        if not part:
            raise ConnectionError('Connection closed before full message received')
        msg += part

    return msg


def extract_creds(sock_object: socket.socket) -> Tuple[Any, Any, Any]:
    """
    Retrieve credentials from SO_PEERCRED option
    """
    _format = '3i'
    creds = sock_object.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize(_format))
    _pid, _uid, _gid = struct.unpack(_format, creds)
    try:
        user, group = pwd.getpwuid(_uid).pw_name, grp.getgrgid(_gid).gr_name
    except KeyError:
        logger.info('Connected by proc %i of %i:%i', _pid, _uid, _gid)
    else:
        logger.info('Connected by proc %i of %i:%i (%s:%s)', _pid, _uid, _gid, user, group)
    return _pid, _uid, _gid


def check_for_root(_uid: int = None) -> bool:
    """
    Check for execution as root | command from root
    """
    if _uid is None:
        _uid = os.geteuid()
    return _uid == 0


def get_xray_exec_user() -> Optional[str]:
    """
    Retrieve the value of XRAYEXEC_UID env and resolve it to username
    """
    proxyuid = os.getenv('XRAYEXEC_UID')
    if proxyuid is not None:
        _proxyuser = get_main_username_by_uid(int(proxyuid))
        logger.info('Got XRAYEXEC_UID: %s (%s), working in USER_MODE', proxyuid, _proxyuser)
        return _proxyuser


def error_response(msg: str) -> str:
    """
    Construct an appropriate formatted response in case of error
    """
    return json.dumps({'result': msg}, ensure_ascii=False)


def nginx_user_cache() -> Optional[bool]:
    """
    Check nginx cache status for current user
    """
    proxyuser = get_xray_exec_user()
    if proxyuser is not None:
        return NginxUserCache(proxyuser).is_enabled


def root_execution_only_check() -> None:
    """
    Check if utility is executed as root and throw error in case if no
    """
    if not check_for_root():
        raise SystemExit(error_response(_('Only root is allowed to execute this utility')))


# --------- DECORATORS ---------


def user_mode_verification(func: Callable) -> Callable:
    """
    Decorator aimed to verify domain owner in X-Ray Manager user mode
    Applies to get_domain_info method
    """

    @wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        proxyuser = get_xray_exec_user()
        if proxyuser is None:
            return func(*args, **kwargs)
        try:
            info = func(*args, **kwargs)
        except XRayMissingDomain as e:
            # Convert to a generic XRayError so callers cannot distinguish
            # "domain does not exist" from "domain belongs to another user"
            # (information-oracle defense). The `from None` suppresses the
            # original exception context so traceback printing also keeps
            # the two cases indistinguishable.
            #
            # Internal-only marker: lets the original caller (e.g.
            # base.py:stop()) re-detect the "missing domain" case to
            # preserve graceful cleanup of orphaned task data / cron jobs
            # for an already-deleted domain. The marker is an INSTANCE
            # attribute on a plain XRayError; it is NOT serialized by
            # XRayError.__str__ (which only emits self.reason and
            # self.context), so it is not observable to user-mode
            # callers via the user_agent socket. Subclassing would
            # have changed type(err) and broken the uniform-class
            # invariant — see tests in user_mode_verification_tests.py.
            err = XRayError(_('%s cannot be found') % f"Domain {e.domain_name}")
            err._internal_missing_domain = True
            raise err from None
        if info.user != proxyuser:
            logger.warning('%s does not belong to user %s', info, proxyuser)
            # No marker here — the "not your domain" path must not be
            # able to trigger the graceful cleanup branch in callers.
            raise XRayError(_('%s cannot be found') % str(info))
        return info

    return wrapper


def verify_user_owns_domain(domain_name: str, domain_owner: str) -> None:
    """
    Enforce caller-owns-domain BEFORE the phpinfo probe's side effects (the
    temp PHP file-write into the target docroot + HTTP fetch) run. It is called
    from phpinfo_utils.get_php_configuration — the single sink every panel
    manager funnels into — so one guard covers Plesk, cPanel, DirectAdmin and
    custom panels. user_mode_verification only compares ownership AFTER the
    decorated get_domain_info body has returned, so without this pre-check a
    user-mode tenant can drive a cross-tenant side effect against another
    user's domain.

    Uses the SAME ownership primitive as user_mode_verification
    (get_xray_exec_user) and raises the SAME uniform error so a foreign domain
    is indistinguishable from a missing one. For root / non-user-mode callers
    (proxyuser is None) this is a no-op, preserving existing behaviour.
    """
    proxyuser = get_xray_exec_user()
    if proxyuser is not None and domain_owner != proxyuser:
        logger.warning('Domain %s does not belong to user %s', domain_name, proxyuser)
        raise XRayError(_('%s cannot be found') % f"Domain {domain_name}")


def user_mode_restricted(func: Callable) -> Callable:
    """
    Decorator aimed to check if user is not hitting limit of running tasks,
    set in X-Ray Manager user mode.
    Applies to start and continue methods.
    Limiting of user's running tasks is applied to Shared PRO only.
    """

    def check(*args):
        """
        If XRAYEXEC_UID exists, check if user does not exceed
        limit of running tasks
        """
        # TODO: [unification] ensure is_cl_shared_pro_edition really needed here
        # https://cloudlinux.atlassian.net/browse/XRAY-244 - (seems yes)
        if not is_cl_shared_pro_edition(skip_jwt_check=True):
            return

        proxyuser = get_xray_exec_user()
        if proxyuser is not None:
            ui_api_cli_instanse = args[0].ui_api_client
            resp = ui_api_cli_instanse.get_task_list()
            list_of_tasks = resp.get('result')
            if list_of_tasks is not None:
                running_count = len([item for item in list_of_tasks if item.get('status') == 'running'])
                if running_count >= user_tasks_count:
                    raise XRayError(
                        _(
                            'Limit of running tasks is {}. You already have {} running task'.format(
                                str(user_tasks_count), str(user_tasks_count)
                            )
                        )
                    )

    @wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        check(*args)
        return func(*args, **kwargs)

    return wrapper


def with_fpm_reload_restricted(func: Callable) -> Callable:
    """
    Decorator aimed to restrict frequent reloads of FPM service
    Applies to get_domain_info method.

    Decorator ordering invariant: this decorator MUST sit OUTSIDE
    user_mode_verification (i.e. ``@with_fpm_reload_restricted`` above
    ``@user_mode_verification``) on every get_domain_info implementation.
    The wrapper below runs ``check`` only AFTER ``func`` returns, so any
    ownership-failure XRayError raised by the inner user_mode_verification
    wrapper short-circuits the throttle probe. Reversing the order would
    let the throttle check inspect another user's FPM service name and
    raise a distinguishable XRayError(flag='warning') before the outer
    ownership check could uniformise the response, leaking a side channel
    on whether the queried domain exists and whose service is currently
    throttled (see XRAY-244 / CLPRO-3079 information-oracle defense).
    """

    def check(*args, data):
        """ """
        # TODO: [unification] ensure is_cl_solo_edition really needed here
        # https://cloudlinux.atlassian.net/browse/XRAY-244 (seems yes)
        if is_cl_solo_edition(skip_jwt_check=True):
            return

        proxyuser = get_xray_exec_user()
        if proxyuser is not None and data.panel_fpm:
            _fpm_service = args[0].fpm_service_name(data)
            if FPMReloadController(_fpm_service).restrict():
                raise XRayError(
                    _(
                        'The X-Ray User service is currently busy. Operation is temporarily not permitted. '
                        'Try again in %s minute'
                    )
                    % str(fpm_reload_timeout),
                    flag='warning',
                )

    @wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        info = func(*args, **kwargs)
        check(*args, data=info)
        return info

    return wrapper


def username_verification(func: Callable) -> Callable:
    def validate(username: str):
        """
        If exists, check XRAYEXEC_UID against user passed param
        """
        proxyuser = get_xray_exec_user()
        if proxyuser is not None and username != proxyuser:
            raise XRayError(_('Incorrect user for request'))

    @wraps(func)
    def wrapper(*args, **kwargs):
        username = kwargs['username']
        validate(username)
        return func(*args, **kwargs)

    return wrapper


def user_mode_advice_verification(func: Callable) -> Callable:
    """
    Decorator aimed to verify user in X-Ray Smart Advice user mode
    Applies to get_detailed_advice method, which takes part in
    advice_details and advice_apply methods
    """

    def verify(data: dict) -> None:
        """
        If exists, check XRAYEXEC_UID against user in metadata of an advice
        """
        proxyuser = get_xray_exec_user()
        try:
            username = data['metadata']['username']
        except KeyError:
            raise XRayError(_('Requested advice cannot be verified'))
        if proxyuser is not None and username != proxyuser:
            raise XRayError(_('Requested advice does not exist'))

    @wraps(func)
    def wrapper(*args, **kwargs):
        """
        Wraps func
        """
        try:
            advice_info, advice_obj = func(*args, **kwargs)
        except XRayAPIError as e:
            # In user mode, normalise an API "not found" (404) for a missing
            # advice_id to the SAME generic error as the ownership-mismatch
            # branch in verify(), so a co-tenant cannot tell "advice absent"
            # from "advice owned by another user" and enumerate valid ids.
            # Scoped to 404 only: 5xx/connectivity/auth errors keep their own
            # distinct message so real failures stay debuggable.
            if get_xray_exec_user() is not None and e.errno == HTTPStatus.NOT_FOUND:
                raise XRayError(_('Requested advice does not exist')) from None
            raise
        verify(advice_info)
        return advice_info, advice_obj

    return wrapper