# Copyright (c) TuxCare, Inc. All rights reserved.
# Licensed under TUXCARE LICENSE AGREEMENT
# https://tuxcare.com/legal/

import hashlib
import json
import os
import shutil
import struct
import tempfile

from . import errors, fetch, log_utils, platform_utils, utils

if False:  # pragma: no cover
    from typing import Optional, Tuple  # noqa: F401


LIBCARE_PLUGIN_VERSION_FILE = 'libcare-plugin.version'
LIBCARE_PLUGIN_BINARY_FILE = 'libcare-plugin.so'
LIBCARE_PLUGIN_STAGING_PATH = '/var/lib/libcare/' + LIBCARE_PLUGIN_BINARY_FILE + '.sig.staging'

LIBCARE_PLUGIN_LIBNAME = 'libcare-plugin'

# Where plugin_hash() looks for the release manifest that carries the
# expected digest, and for the signature that makes it trustworthy offline.
PLUGIN_VERIFY_MANIFEST = '/var/cache/kcare/release.content.json'
PLUGIN_VERIFY_MANIFEST_SIG = PLUGIN_VERIFY_MANIFEST + '.json-sig'


SHA256_HEX_LEN = 64


def _verify_manifest_bytes(content):  # type: (bytes) -> None
    """Check the signature of content itself, out of reach of any writer.

    Mirrors load_plugin() in kpatch_server_main.c: copy the bytes aside, unlink
    the copy, and hand the verifier a path that only this process can reach.
    """
    fd, path = tempfile.mkstemp(dir=os.path.dirname(PLUGIN_VERIFY_MANIFEST))
    try:
        os.write(fd, content)
    finally:
        os.close(fd)

    ro_fd = os.open(path, os.O_RDONLY)
    link = '/proc/{0}/fd/{1}'.format(os.getpid(), ro_fd)
    try:
        os.unlink(path)
        fetch.check_gpg_signature(link, PLUGIN_VERIFY_MANIFEST_SIG)  # type: ignore[no-untyped-call]
    finally:
        os.close(ro_fd)


def plugin_hash():  # type: () -> str
    """Return the SHA256 the signed manifest publishes for this plugin slot."""
    plugin_id = _libcare_plugin_id_for_distro(*platform_utils.get_os_release())
    if not plugin_id:
        raise errors.KcareError('Cannot determine the libcare plugin id for this distro')

    try:
        with open(PLUGIN_VERIFY_MANIFEST, 'rb') as f:
            content = f.read()
        _verify_manifest_bytes(content)
        files = json.loads(utils.nstr(content))['files']
        digest = files['{0}/{1}'.format(plugin_id, LIBCARE_PLUGIN_BINARY_FILE)]['sha256']
    except (IOError, OSError, ValueError, KeyError, TypeError) as err:
        raise errors.KcareError('Libcare plugin manifest lookup failed: {0}'.format(err))

    if len(digest) != SHA256_HEX_LEN:
        raise errors.KcareError('Malformed sha256 for {0} in the libcare plugin manifest'.format(LIBCARE_PLUGIN_BINARY_FILE))

    return digest  # type: ignore[no-any-return]


def stage_plugin(source_path):  # pragma: no cover unit
    # type: (str) -> None
    """Copy source_path into the staging slot."""
    staging_dir = os.path.dirname(LIBCARE_PLUGIN_STAGING_PATH)
    tmp_path = LIBCARE_PLUGIN_STAGING_PATH + '.tmp'
    try:
        if not os.path.isdir(staging_dir):
            os.makedirs(staging_dir)
        shutil.copy2(source_path, tmp_path)
        os.rename(tmp_path, LIBCARE_PLUGIN_STAGING_PATH)
    finally:
        if os.path.exists(tmp_path):
            os.unlink(tmp_path)


def install_manifest(manifest_path, signature):  # type: (str, bytes) -> None
    """Put the manifest and its signature where plugin_hash() looks for them.

    Both files are in place on disk before either is moved to its final name,
    and the pair being replaced is set aside for :func:`revert_manifest`.
    """
    target_dir = os.path.dirname(PLUGIN_VERIFY_MANIFEST)
    if not os.path.isdir(target_dir):
        os.makedirs(target_dir)

    tmp_manifest = PLUGIN_VERIFY_MANIFEST + '.tmp'
    tmp_sig = PLUGIN_VERIFY_MANIFEST_SIG + '.tmp'
    shutil.copy2(manifest_path, tmp_manifest)
    with open(tmp_sig, 'wb') as f:
        f.write(signature)

    if os.path.isfile(PLUGIN_VERIFY_MANIFEST) and os.path.isfile(PLUGIN_VERIFY_MANIFEST_SIG):
        os.rename(PLUGIN_VERIFY_MANIFEST, PLUGIN_VERIFY_MANIFEST + '.prev')
        os.rename(PLUGIN_VERIFY_MANIFEST_SIG, PLUGIN_VERIFY_MANIFEST_SIG + '.prev')
    os.rename(tmp_manifest, PLUGIN_VERIFY_MANIFEST)
    os.rename(tmp_sig, PLUGIN_VERIFY_MANIFEST_SIG)


def revert_manifest():  # type: () -> None
    """Undo install_manifest, so the manifest never outlives the swap."""
    for path in (PLUGIN_VERIFY_MANIFEST, PLUGIN_VERIFY_MANIFEST_SIG):
        try:
            if os.path.isfile(path + '.prev'):
                os.rename(path + '.prev', path)
            else:
                os.unlink(path)
        except OSError as err:
            log_utils.logwarn('Plugin manifest revert failed for {0}: {1}'.format(path, err))


def _libcare_plugin_id_for_distro(distro_id, version):  # type: (str, str) -> Optional[str]
    """Build patches03 build_id slot `<distro-id>-<version>-libcare-plugin`
    from /etc/os-release fields. Mirrors the publisher side in
    `scripts/makedeploy/libcare_distributives_mapping.py`. Version is
    full VERSION_ID for ubuntu, major-only int elsewhere. Returns
    ``None`` when distro_id or version is empty / unparseable.
    """
    distro_id = (distro_id or '').lower()
    version = str(version or '')
    if not distro_id or not version:
        return None
    if distro_id == 'ubuntu':
        ver = version
    else:
        try:
            ver = str(int(version.split('.', 1)[0]))
        except (TypeError, ValueError):
            return None
    raw = '{0}-{1}-libcare-plugin'.format(distro_id, ver)
    return hashlib.sha1(raw.encode('ascii')).hexdigest()  # nosec B324


def read_libcare_plugin_version_from_elf(path):  # type: (str) -> str
    """Read the .libcare_plugin_version section from an ELF64 file.

    Mirrors the C-side read_plugin_version in kpatch_server_main.c so the
    client can decide whether a locally-staged plugin is newer than the
    running one without having to load it.

    :param path: Absolute path to the ELF file.
    :return: Version string, or empty string on any failure.
    """
    try:
        with open(path, 'rb') as f:
            ehdr = f.read(64)
            if len(ehdr) < 64 or ehdr[:4] != b'\x7fELF' or ehdr[4:5] != b'\x02':
                return ''

            (e_shoff,) = struct.unpack('<Q', ehdr[40:48])
            (e_shentsize,) = struct.unpack('<H', ehdr[58:60])
            (e_shnum,) = struct.unpack('<H', ehdr[60:62])
            (e_shstrndx,) = struct.unpack('<H', ehdr[62:64])

            if e_shentsize != 64 or e_shnum == 0 or e_shstrndx >= e_shnum:
                return ''

            f.seek(e_shoff)
            shdrs = f.read(e_shentsize * e_shnum)
            if len(shdrs) < e_shentsize * e_shnum:
                return ''

            def _shdr(i):  # type: (int) -> Tuple[int, int, int]
                off = i * e_shentsize
                (sh_name,) = struct.unpack('<I', shdrs[off : off + 4])
                (sh_offset,) = struct.unpack('<Q', shdrs[off + 24 : off + 32])
                (sh_size,) = struct.unpack('<Q', shdrs[off + 32 : off + 40])
                return sh_name, sh_offset, sh_size

            _, shstr_off, shstr_sz = _shdr(e_shstrndx)
            f.seek(shstr_off)
            shstrtab = f.read(shstr_sz)

            for i in range(e_shnum):
                sh_name, sh_off, sh_sz = _shdr(i)
                if sh_name >= shstr_sz:
                    continue
                end = shstrtab.find(b'\0', sh_name)
                if end < 0:
                    continue
                if shstrtab[sh_name:end] != b'.libcare_plugin_version':
                    continue
                f.seek(sh_off)
                data = f.read(sh_sz)
                return data.rstrip(b'\0').decode('ascii', 'replace').strip()
    except (IOError, OSError, struct.error):
        return ''
    return ''


def compare_libcare_plugin_versions(v1, v2):  # type: (str, str) -> int
    """Compare two plugin version strings (YYYY-MM-DD-NUM[-s]).

    Strips the '-s' GPG suffix before comparison.
    The date prefix is compared lexicographically, the trailing
    NUM is compared numerically to handle multi-digit numbers.

    :return: <0 if v1 < v2, 0 if equal, >0 if v1 > v2
    """

    def normalize(v):  # type: (str) -> str
        if v.endswith('-s'):
            v = v[:-2]
        return v

    a = normalize(v1)
    b = normalize(v2)

    # Split at last '-' to get date prefix and NUM
    a_idx = a.rfind('-')
    b_idx = b.rfind('-')

    if a_idx > 0 and b_idx > 0:
        a_prefix, a_num = a[:a_idx], a[a_idx + 1 :]
        b_prefix, b_num = b[:b_idx], b[b_idx + 1 :]

        if len(a_prefix) == len(b_prefix):
            if a_prefix != b_prefix:
                return -1 if a_prefix < b_prefix else 1
            # Same date prefix - compare NUM numerically
            try:
                na, nb = int(a_num), int(b_num)
                return (na > nb) - (na < nb)
            except ValueError:
                pass

    # Fallback: lexicographic
    return -1 if a < b else (1 if a > b else 0)