"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
See the GNU General Public License for more details.


You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.

Copyright © 2019 Cloud Linux Software Inc.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import asyncio
import base64
import logging
import os
import re
import shutil
from collections import defaultdict
from pathlib import Path
from typing import Callable, Iterable, List, Optional, Set, Tuple

from defence360agent.contracts.config import ANTIVIRUS_MODE, Malware
from defence360agent.subsys import svcctl
from defence360agent.subsys.panels.hosting_panel import HostingPanel
from defence360agent.utils import check_run, is_cluster
from defence360agent.utils.common import LooseVersion
from imav.malwarelib.model import MalwareIgnorePath
from imav.malwarelib.scan.crontab import crontab_path
from imav.malwarelib.tenant_path import (
    GLOBAL_TENANT,
    split_marked,
    strip_tenant,
)
from imav.malwarelib.utils.user_list import registered_apps

logger = logging.getLogger(__name__)

# location of admin provided watched and ignored paths
_ADMIN_PATH = Path("/etc/sysconfig/imunify360/malware-filters-admin-conf")
# location of internal configs, shipped with imunify360-firewall
_INTERNAL_PATH = Path("/var/imunify360/files/realtime-av-conf/v1")
# location of processed configs
_PROCESSED_PATH = _ADMIN_PATH / "processed"

_PD_NAME = "pd-combined.txt"
_INTERNAL_NAME = "av-internal.txt"
_INTERNAL_OPTIMIZED_NAME = "av-internal-optimized.txt"
_ADMIN_NAME = "av-admin.txt"
_ADMIN_PATHS_NAME = "av-admin-paths.txt"
_IGNORED_SUB_DIR = "ignored"
_MAX_PATTERN_LENGTH = 64000

REALTIME_PACKAGE = "imunify-realtime-av"
REALTIME_SERVICE_NAME = "imunify-realtime-av"
_PD_PREPARE = "/usr/bin/i360-exclcomp"
_BIN_PATH = Path("/usr/sbin/imunify-realtime-av")

# K8s per-user ignore list file export (under _ADMIN_PATHS_NAME location)
_K8S_IGNORE_LISTS_PATH = _PROCESSED_PATH / _IGNORED_SUB_DIR / "user"


class PatternLengthError(Exception):
    """Raised when pattern's length is too big."""

    pass


def _save_basedirs(dir: Path, basedirs: Set[str]) -> None:
    """Save list of basedirs in a file inside dir."""
    with (dir / "basedirs-list.txt").open("w") as f:
        for basedir in sorted(basedirs):
            f.write(os.path.realpath(basedir) + "\n")


def _split_paths(paths: List[str]) -> Tuple[List[str], List[str]]:
    """Split paths into two lists: absolute and relative.

    Relative paths start with +. This + sign is removed from resulting path."""
    absolute, relative = [], []
    for path in paths:
        if path.startswith("+"):
            relative.append(path[1:])
        else:
            absolute.append(path)
    return absolute, relative


def _read_list(path: Path) -> List[str]:
    """Read file at path and return its lines as a list.

    Empty lines or lines starting with '#' symbol are skipped. Lines are
    stripped of leading and trailing whitespace. If the file does not exist,
    empty list is returned."""
    try:
        with path.open() as f:
            lines = [line.strip() for line in f]
            return [x for x in lines if len(x) > 0 and not x.startswith("#")]
    except FileNotFoundError:
        return []


class _Watched(list):
    """Holds a list of watched glob patterns ready to be saved."""

    def __init__(self, w: List[str], basedirs: Set[str]) -> None:
        super().__init__()
        absolute, relative = _split_paths(w)
        self.extend(
            os.path.realpath(p)
            for p in absolute + self._extend_relative(relative, basedirs)
            if self._is_valid(p)
        )

    @staticmethod
    def _is_valid(pattern: str) -> bool:
        """Return True if watched pattern is valid."""
        if not pattern.startswith("/"):
            logger.warning(
                "skipping watched path %s: not starts with /", pattern
            )
            return False
        return True

    @staticmethod
    def _extend_relative(paths: List[str], basedirs: Set[str]) -> List[str]:
        """Join basedirs with all paths and return resulting list."""
        extended = []
        for path in paths:
            for basedir in basedirs:
                extended.append(os.path.join(basedir, path))
        return extended

    def save(self, path: Path) -> None:
        """Save watched list at specified path."""
        with path.open("w") as f:
            f.write("\n".join(self))


class _Ignored(str):
    """Holds a list of ignored regexp patterns ready to be saved."""

    @staticmethod
    def _is_valid_relative(pattern: str) -> bool:
        """Return True if relative ignored pattern is valid."""
        if pattern.startswith("^"):
            logger.warning(
                "skipping relative ignored path %s: starts with ^", pattern
            )
            return False
        return True

    @staticmethod
    def _remove_leading_slash(pattern: str) -> str:
        """Remove leading slash from pattern, if present."""
        if pattern.startswith("/"):
            return pattern[1:]
        return pattern

    @staticmethod
    def _compiles(pattern: str) -> bool:
        """Return True if pattern successfully compiles as regexp."""
        try:
            re.compile(pattern)
            return True
        except Exception:
            logger.warning(
                "skipping ignored pattern %s: invalid regex", pattern
            )
            return False

    @classmethod
    def from_patterns(
        cls, patterns: List[str], basedirs: Set[str]
    ) -> "_Ignored":
        """Build single ignored regexp from given patterns and basedirs."""
        absolute, relative = _split_paths(patterns)
        absolute = [p for p in absolute if cls._compiles(p)]
        relative = [
            cls._remove_leading_slash(p)
            for p in relative
            if cls._is_valid_relative(p) and cls._compiles(p)
        ]
        if len(basedirs) > 0 and len(relative) > 0:
            relative_pattern = "^(?:{})/(?:{})".format(
                "|".join(basedirs), "|".join(relative)
            )
            absolute.append(relative_pattern)
        pat = "|".join(absolute)
        if pat == "":
            pat = "^$"
        return _Ignored(pat)

    def save(self, path: Path):
        """Save ignored list at specified path."""
        if len(self) > _MAX_PATTERN_LENGTH:
            raise PatternLengthError(
                "{} pattern is too long ({})".format(path, len(self))
            )
        with path.open("w") as f:
            f.write(self)


def _read_configs(panel: str, name: str) -> Tuple[List[str], List[str]]:
    """Read internal and admin lists from files with given name."""
    common_dir = _INTERNAL_PATH / "common"
    internal = _read_list(common_dir / name)

    panel_path = _INTERNAL_PATH / panel.lower()
    if panel_path.exists():
        internal.extend(_read_list(panel_path / name))

    return internal, _read_list(_ADMIN_PATH / name)


class _WatchedCtx:
    def __init__(self, internal: _Watched, admin: _Watched) -> None:
        self.internal = internal
        self.admin = admin

    def save(self, dir: Path) -> None:
        w = dir / "watched"
        w.mkdir(exist_ok=True)
        self.internal.save(w / _INTERNAL_NAME)
        self.admin.save(w / _ADMIN_NAME)


def _watched_context(
    panel_name: str, basedirs: Set[str], *, extra: Iterable[str]
) -> _WatchedCtx:
    internal_watched, admin_watched = _read_configs(panel_name, "watched.txt")
    internal_watched.extend(extra)
    return _WatchedCtx(
        _Watched(internal_watched, basedirs), _Watched(admin_watched, basedirs)
    )


class _IgnoredCtx:
    def __init__(
        self,
        internal: _Ignored,
        admin: _Ignored,
        pd: _Ignored,
        internal_optimized: Optional[_Ignored] = None,
    ) -> None:
        self.internal = internal
        self.admin = admin
        self.pd = pd
        self.internal_optimized = internal_optimized

    def save(self, dir: Path) -> None:
        w = dir / _IGNORED_SUB_DIR
        w.mkdir(exist_ok=True)
        self.internal.save(w / _INTERNAL_NAME)
        self.admin.save(w / _ADMIN_NAME)
        self.pd.save(w / _PD_NAME)
        if self.internal_optimized is not None:
            self.internal_optimized.save(w / _INTERNAL_OPTIMIZED_NAME)


def _ignored_context(panel_name: str, basedirs: Set[str]) -> _IgnoredCtx:
    internal_ignored, admin_ignored = _read_configs(panel_name, "ignored.txt")

    # Read optimized list if available (DEF-35124).
    # Built from optimized_ignored only (no admin), same as av-internal.txt
    # is built from internal_ignored only. Scanners read av-admin.txt
    # separately, so admin patterns must not be baked in here.
    optimized_ignored, _ = _read_configs(panel_name, "ignored-optimized.txt")
    internal_optimized = None
    if optimized_ignored:
        internal_optimized = _Ignored.from_patterns(
            optimized_ignored, basedirs
        )

    return _IgnoredCtx(
        _Ignored.from_patterns(internal_ignored, basedirs),
        _Ignored.from_patterns(admin_ignored, basedirs),
        _Ignored.from_patterns(internal_ignored + admin_ignored, basedirs),
        internal_optimized,
    )


def _admin_ignored_paths(dir: Path) -> None:
    ignored_paths = MalwareIgnorePath.path_list(resource_type="file")
    ignored_paths_base64 = b"".join(
        base64.b64encode(os.fsencode(strip_tenant(path))) + b"\n"
        for path in ignored_paths
    )
    target = dir / _IGNORED_SUB_DIR / _ADMIN_PATHS_NAME
    target.write_bytes(ignored_paths_base64)


def _contain_changes(dir1: Path, dir2: Path) -> bool:
    """Compare content of two folders if files in this directory are the
    same return False."""
    for file in dir1.iterdir():
        if file.is_dir():
            if _contain_changes(file, dir2 / file.name):
                return True
        if not file.is_file():
            continue
        other = dir2 / file.name
        if not other.exists():
            return True
        if file.read_bytes() != other.read_bytes():
            return True
    return False


def _save_configs(dir: Path, savers: List[Callable[[Path], None]]) -> bool:
    """Save configs in directory dir using saves callable.

    Each function in savers will be called with single dir argument."""
    temp = dir.with_suffix(".tmp")
    if temp.exists():
        shutil.rmtree(str(temp))
    temp.mkdir()

    for save in savers:
        save(temp)

    if dir.exists():
        backup = dir.with_name(".backup")
        if backup.exists():
            shutil.rmtree(str(backup))
        dir.rename(backup)
        try:
            temp.rename(dir)
        except Exception:
            backup.rename(dir)
            raise
        return _contain_changes(dir, backup)
    else:
        temp.rename(dir)
        return True


def _update_pd_symlink() -> None:
    target = _PROCESSED_PATH / _IGNORED_SUB_DIR / _PD_NAME
    source = _ADMIN_PATH / _PD_NAME
    try:
        # source.exists() returns False for broken symlink.
        # so call lstat() and if it throws exception, source does not exist.
        _ = source.lstat()
    except FileNotFoundError:
        source.symlink_to(target)
    else:
        if not (
            source.is_symlink() and os.readlink(str(source)) == str(target)
        ):
            source.unlink()
            source.symlink_to(target)


def _watches_user_content(watched: _WatchedCtx, basedirs: Set[str]) -> bool:
    if watched.admin:
        return True
    roots = [os.path.realpath(basedir).rstrip("/") for basedir in basedirs]
    return any(
        entry == root or entry.startswith(root + "/")
        for entry in watched.internal
        for root in roots
    )


def generate_configs() -> bool:
    """Generate new malware paths filters config."""
    panel = HostingPanel()
    basedirs = panel.basedirs()

    extra_watched = set()
    if Malware.CRONTABS_SCAN_ENABLED:
        extra_watched.add(str(crontab_path()))

    watched = _watched_context(panel.NAME, basedirs, extra=extra_watched)
    if should_be_running() and not _watches_user_content(watched, basedirs):
        logger.error(
            "realtime watch set contains no user-content paths, only"
            " system paths will be watched: panel=%s basedirs=%r",
            panel.NAME,
            basedirs,
        )
    changed = _save_configs(
        _PROCESSED_PATH,
        [
            lambda dir: _save_basedirs(dir, {*basedirs, *extra_watched}),
            watched.save,
            _ignored_context(panel.NAME, basedirs).save,
            _admin_ignored_paths,
        ],
    )
    _update_pd_symlink()
    return changed


def is_installed() -> bool:
    return _BIN_PATH.exists()


# the alternatives symlink resolves to .imrt2 or .legacy
_IMRT2_ALTERNATIVES_SUFFIX = ".imrt2"
# older imrt2 could absorb a config change landing during its in-process
# reload cooldown; only the agent-side restart masked that
_MIN_SELF_RELOAD_VERSION = "8.0.20"
_VERSION_RE = re.compile(rb"version\s+(\d+(?:\.\d+)+)")
_self_reload_cache: Tuple[Optional[Tuple[str, float]], bool] = (None, False)


async def _self_reload_capable() -> bool:
    """Whether the active realtime binary picks up config changes itself."""
    global _self_reload_cache
    try:
        real = _BIN_PATH.resolve(strict=True)
    except OSError:
        return False
    if not real.name.endswith(_IMRT2_ALTERNATIVES_SUFFIX):
        return False
    try:
        key = (str(real), real.stat().st_mtime)
    except OSError:
        return False
    cached_key, cached_value = _self_reload_cache
    if cached_key == key:
        return cached_value
    capable = False
    try:
        out = await check_run([str(_BIN_PATH), "-version"])
        match = _VERSION_RE.search(out)
        if match:
            capable = LooseVersion(match.group(1).decode()) >= LooseVersion(
                _MIN_SELF_RELOAD_VERSION
            )
    except asyncio.CancelledError:
        raise
    except Exception as e:
        logger.warning("realtime_av version probe failed: %s", e)
    _self_reload_cache = (key, capable)
    return capable


async def reload_services() -> None:
    """Apply freshly regenerated malware-filter configs."""
    # a self-reloading imrt2 picks the configs up in-process; the historical
    # restart also resurrected a stopped unit, so keep that via an idempotent
    # start-when-inactive; ANY probe failure falls back to the restart
    action: Optional[str] = None
    if not await _self_reload_capable():
        action = "restart"
    else:
        try:
            unit = svcctl.adaptor(REALTIME_SERVICE_NAME)
            if not await unit.is_active():
                # a failed unit is also inactive; clear its state (and any
                # start-rate-limit counter) so the plain start below cannot
                # be refused — no-op on a cleanly stopped unit
                await unit.reset_failed()
                action = "start"
        except Exception as e:
            logger.warning(
                "realtime_av.reload_services unit-state check failed: %s", e
            )
            action = "restart"
    tasks = []
    if action:
        tasks.append(check_run(["service", REALTIME_SERVICE_NAME, action]))
    tasks.append(check_run([_PD_PREPARE]))
    for t in tasks:
        try:
            await t
        except asyncio.CancelledError:
            raise
        except Exception as e:
            logger.warning("realtime_av.reload_services exception: %s", e)


def should_be_running() -> bool:
    return not ANTIVIRUS_MODE and Malware.INOTIFY_ENABLED


def _write_ignore_file(path: Path, paths: List[str]) -> None:
    """Write paths to ignore file, base64 encoded, one per line."""
    path.parent.mkdir(parents=True, exist_ok=True)
    content = b"".join(
        base64.b64encode(os.fsencode(p)) + b"\n" for p in sorted(paths)
    )
    path.write_bytes(content)


async def write_k8s_ignore_lists() -> None:
    """Write per-application ignore lists for the in-container watchers.

    One ``<app_id>/ignore.txt`` per application that has entries, holding
    the bare in-container paths — the tenant is the entry's path prefix
    here, but the consumer lives inside that tenant already.

    The tenant is trusted only when it is the wildcard or a registered
    application (``split_marked``), so a stored path whose first component
    is not a real tenant cannot direct a write outside the intended
    ``<app_id>/`` directory. Only file-type entries are exported; db
    entries never carry a container path.
    """
    if not is_cluster():
        return

    applications = await registered_apps()
    global_paths = []
    paths_by_app = defaultdict(list)
    for entry in MalwareIgnorePath.path_list(resource_type="file"):
        app_id, bare = split_marked(entry, applications)
        if app_id == GLOBAL_TENANT:
            global_paths.append(bare)
        elif app_id is not None:
            paths_by_app[app_id].append(bare)

    if _K8S_IGNORE_LISTS_PATH.exists():
        for user_dir in _K8S_IGNORE_LISTS_PATH.iterdir():
            if user_dir.is_dir() and user_dir.name not in paths_by_app:
                shutil.rmtree(str(user_dir))

    for app_id, paths in paths_by_app.items():
        _write_ignore_file(
            _K8S_IGNORE_LISTS_PATH / app_id / "ignore.txt",
            paths + global_paths,
        )