path: root/modulemanager

#!/usr/bin/env python3
#
# InspIRCd -- Internet Relay Chat Daemon
#
#   Copyright (C) 2026 Sadie Powell <sadie@witchery.services>
#
# This file is part of InspIRCd.  InspIRCd 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, version 2.
#
# 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 <http://www.gnu.org/licenses/>.
#


import argparse
import configparser
import hashlib
import json
import pathlib
import re
import sys
import urllib.request

from src.python import common, console, module

CONTRIB_MODULES = module.CONTRIB_DIR / "installed-modules.json"
CONTRIB_SOURCES = module.CONTRIB_DIR / "module-sources.ini"


def fetch_http(url):
    try:
        with urllib.request.urlopen(url) as response:
            return response.read().decode("utf-8")

    except (urllib.error.HTTPError, urllib.error.URLError, OSError) as e:
        console.error(f"Unable to download {console.color(url, console.BOLD)}: {e}")


def fetch_modules(repository):
    config = configparser.ConfigParser()
    config.read(CONTRIB_SOURCES)

    modules = {}
    for section in config.sections():
        if repository and repository not in section:
            continue  # Not what we're looking for.

        repo = config[section]
        if "url" not in repo:
            console.warning(f"The {section} repository does not have an URL associated with it!")
            continue

        fetch_repository(section, repo, modules)
    return modules


def fetch_repository(name, repository, modules):
    repository_url = repository["url"]
    console.debug(f"Fetching modules for {name} from {repository_url}")

    data = fetch_http(repository_url)
    if data:
        module_config = None
        for line in data.splitlines():
            if match := re.fullmatch(r"\s*module (\S+) (\S+) (.+)", line, re.IGNORECASE):
                module_name = match.group(1)
                modules.setdefault(module_name, [])
                module_config = {
                    "author":          [],
                    "checksum":        "",
                    "conflicts":       {},
                    "depends":         {},
                    "description":     "",
                    "mask":            "",
                    "name":            module_name,
                    "repository":      name,
                    "repository-desc": repository["description"],
                    "url":             match.group(3),
                    "version":         module.version_parse(match.group(2))
                }
                modules[module_name].append(module_config)

            elif match := re.fullmatch(r"\s*author\s+(.+)", line, re.IGNORECASE):
                module_config["author"].append(match.group(1))

            elif match := re.fullmatch(r"\s*checksum\s+(\S+)", line, re.IGNORECASE):
                module_config["checksum"] = match.group(1)

            elif match := re.fullmatch(r"\s*conflicts\s+(\S+)(?:\s+(.+))", line, re.IGNORECASE):
                module_config["conflicts"][match.group(1)] = match.group(2)

            elif match := re.fullmatch(r"^\s*depends\s+(\S+)\s+(\d+(?:\.\d+)*)(?:-(\d+(?:\.\d+)*))?", line, re.IGNORECASE):
                oldest_version = module.version_parse(match.group(2))
                newest_version = module.version_parse(match.group(3)) if match.group(3) else oldest_version
                module_config["depends"][match.group(1)] = (oldest_version, newest_version)

            elif match := re.fullmatch(r"\s*description\s+(.+)", line, re.IGNORECASE):
                module_config["description"] = match.group(1)

            elif match := re.fullmatch(r"\s*mask\s+(.+)", line, re.IGNORECASE):
                module_config["mask"] = match.group(1)

            elif match := re.fullmatch(r"\s*source\s+(.+)", line, re.IGNORECASE):
                fetch_repository(repository, match.group(1), modules)

            elif line:
                console.debug(f"Unknown line: {line}")


def find_module(modules, name, repository):
    short_name = module.short_name(name)
    selected_module = None
    for module_data in modules.get(short_name, {}):
        if repository and repository != module_data["repository"]:
            continue

        version = module_data["version"]
        if not selected_module or module.version_compare(selected_module["version"], version) > 0:
            console.debug(f"Potential remote module match: {name} {module.version_string(version)}")
            selected_module = module_data

    if selected_module:
        console.debug(f"Remote dependency match: {name} {module.version_string(selected_module['version'])}")

    return selected_module


def resolve_conflicts(dependencies, module_to_install):
    module_to_install_name = module_to_install["name"]
    for conflict_module, conflict_reason in module_to_install["conflicts"].items():
        console.debug(f"The {module_to_install_name} conflicts with {conflict_module}: {conflict_reason}")
        if conflict_module in dependencies:
            console.error(f"The {module_to_install_name} module can not be installed whilst the {conflict_module} is installed: {conflict_reason}")


def resolve_dependencies(dependencies, installed_modules, remote_modules, module_to_install):
    module_to_install_name = module_to_install["name"]
    modules_to_install = []

    for required_module, required_version in module_to_install["depends"].items():
        requirement = required_module + " " + module.version_range_string(required_version)
        console.debug(f"The {module_to_install_name} depends on {requirement}")

        # Check for the dependency locally.
        dependency = dependencies.get(required_module)
        if dependency and module.version_in_range(dependency, required_version):
            console.debug(f"Local dependency match: {required_module} {module.version_string(dependency)}")
            continue

        # Check for the dependency in the remotes.
        remote_selection = None
        for remote_module in remote_modules.get(required_module, []):
            remote_version = remote_module["version"]
            if not module.version_in_range(remote_version, required_version):
                console.debug(f"Remote dependency not suitable: {required_module} {module.version_string(remote_version)}")
                continue

            # Check for a more recent potential dependency.
            if not remote_selection or module.version_compare(remote_selection["version"], remote_version) > 0:
                console.debug(f"Potential remote dependency match: {required_module} {module.version_string(remote_version)}")
                remote_selection = remote_module

        if not remote_selection:
            console.error(f"Unable to find a module matching {console.color(requirement, console.BOLD)} in any remotes.")

        console.debug(f"Remote dependency match: {required_module} {module.version_string(remote_selection['version'])}")
        modules_to_install.append(remote_selection)

    return modules_to_install


def install_module(dependencies, installed_modules, remote_modules, module_to_install, upgrade):
    module_to_install_name = module_to_install["name"]
    if module_to_install_name in dependencies and not upgrade:
        console.debug(f"The {module_to_install_name} module is already installed.")
        return

    module_to_install_version = module.version_string(module_to_install["version"])
    print(f"Installing version {console.color(module_to_install_version, console.BOLD)} of the {console.color(module_to_install_name, console.BOLD)} module ...")

    resolve_conflicts(dependencies, module_to_install)
    for module_dependency in resolve_dependencies(dependencies, installed_modules, remote_modules, module_to_install):
        install_module(dependencies, installed_modules, remote_modules, module_dependency, upgrade)

    try:
        contents = fetch_http(module_to_install["url"])

        actual_checksum = hashlib.sha256(contents.encode("utf-8")).hexdigest()
        if "checksum" in module_to_install:
            expected_checksum = module_to_install["checksum"]
            if actual_checksum != expected_checksum:
                console.error(
                    f"Checksum for the {console.color(module_to_install_name, console.BOLD)} module did not match, possibly corrupt module detected.",
                    f"Expected checksum: {expected_checksum}",
                    f"Actual checksum:   {actual_checksum}",
                )

        else:
            console.warning(f"The module source for {console.color(module_to_install_name, console.BOLD)} module did not specify a checksum, unable to detect corruption.")

        with open(module.CONTRIB_DIR / module.file_name(module_to_install_name), "w", encoding="utf-8") as fh:
            fh.write(contents)

        dependencies[module_to_install_name] = module_to_install["version"]
        installed_modules[module_to_install_name] = {
            "checksum": actual_checksum,
            "repository": module_to_install["repository"],
            "version": module_to_install_version,
        }

    except OSError as e:
        console.error(f"Unable to write module source: {e}")


def installed_dependencies(installed_modules):
    dependencies = {}

    # Core
    core_version = common.version()
    dependencies["core"] = (core_version["MAJOR"], core_version["MINOR"], core_version["PATCH"])

    # Official modules
    for long_module in module.DIR.iterdir():
        if long_module.suffix == ".cpp":
            short_name = module.short_name(long_module)
            dependencies[short_name] = dependencies["core"]

    # Installed modules
    for module_name, module_data in installed_modules.items():
        dependencies[module_name] = module.version_parse(module_data["version"])

    return dependencies


def load_contrib_manifest():
    try:
        console.debug(f"Reading the contrib module manifest from {CONTRIB_MODULES} ...")
        with open(CONTRIB_MODULES, "r", encoding="utf-8") as fh:
            return json.load(fh)

    except FileNotFoundError:
        return {}  # No contrib modules installed

    except (OSError, json.JSONDecodeError) as e:
        console.error(f"Unable to read contrib module manifest: {e}")


def save_contrib_manifest(installed_modules):
    try:
        console.debug(f"Writing the contrib module manifest to {CONTRIB_MODULES} ...")
        with open(CONTRIB_MODULES, "w", encoding="utf-8") as fh:
            json.dump(installed_modules, fh, indent=4)

    except OSError as e:
        console.error(f"Unable to write contrib module manifest: {e}")


def cmd_contrib(commands, args, start):
    return console.subcommand(args, 2, {
        "info": {
            "function": cmd_contrib_info,
            "help": "Install a module from a contrib repository",
            "options": {
                "--only-installed": "Only list modules that are installed",
                "--repository REPO": "The repository to show modules from",
            }
        },
        "install": {
            "function": cmd_contrib_install,
            "help": "Install a module from a contrib repository",
            "options": {
                "--repository REPO": "The repository to install modules from",
            }
        },
        "list": {
            "function": cmd_contrib_list,
            "help": "List available contrib modules",
            "options": {
                "--only-installed": "Only list modules that are installed",
                "--repository REPO": "The repository to list modules from",
                "--show-masked": "Show modules that have been masked",
            },
        },
        "uninstall": {
            "function": cmd_contrib_uninstall,
            "help": "Uninstall a module installed from a contrib repository",
        },
        "upgrade": {
            "function": cmd_contrib_upgrade,
            "help": "Update installed contrib modules",
        },

        # Helper aliases for common synonyms.
        "update": {
            "function": cmd_contrib_upgrade,
        },
    })


def cmd_contrib_info(commands, args, start):
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("--only-installed", help="Only list modules that are installed", action="store_true")
    parser.add_argument("--repository", help="The repository to list modules from", metavar="REPO")
    (options, modules_to_show) = parser.parse_known_args(args[start:])

    if not modules_to_show:
        console.error("You must specify one or more modules to show information about.")

    installed_modules = load_contrib_manifest()
    remote_modules = fetch_modules(options.repository)

    for module_name in modules_to_show:
        short_name = module.short_name(module_name)
        if short_name not in remote_modules:
            console.warning(f"The {console.color(short_name, console.BOLD)} module is not available.")
            continue  # Nothing to do

        print(f"Information about the {console.color(short_name, console.BOLD)} module:")
        print()

        for idx, remote_module in enumerate(remote_modules[short_name]):
            if idx:
                print("---")

            data = []
            data.append(["Description", remote_module["description"]])
            data.append(["Repository",  f"{remote_module['repository-desc']} ({remote_module['repository']})"])

            for author in remote_module["author"]:
                data.append(["Author", author])

            for conflict_module, conflict_reason in remote_module["conflicts"].items():
                data.append(["Conflicts with",  f"{conflict_module} ({conflict_reason})"])

            for required_module, required_version in remote_module["depends"].items():
                data.append(["Depends on", f"{required_module} {module.version_range_string(required_version)}"])

            if remote_module["mask"]:
                data.append(["Mask reason", remote_module['mask']])

            if short_name in installed_modules:
                installed_module = installed_modules[short_name]
                if installed_module["repository"] == remote_module["repository"]:
                    match module.version_compare(module.version_parse(installed_module["version"]), remote_module["version"]):
                        case -1:  # Outdated
                            data.append(["Status", "outdated"])
                        case 0:  # Up-to-date
                            data.append(["Status", "up to date"])
                        case 1:  # More recent than contrib
                            data.append(["Status", "more recent"])
                else:
                    data.append(["Status", "installed from different repository"])
            else:
                data.append(["Status", "not installed"])

            console.table(None, data, ":")


def cmd_contrib_install(commands, args, start):
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("--repository", help="The repository to list modules from", metavar="REPO")
    (options, modules_to_install) = parser.parse_known_args(args[start:])

    if not modules_to_install:
        console.error("You must specify one or more modules to install.")

    installed_modules = load_contrib_manifest()
    dependencies = installed_dependencies(installed_modules)
    remote_modules = fetch_modules(options.repository)

    modules_installed = False
    for module_name in modules_to_install:
        short_name = module.short_name(module_name)
        if short_name in installed_modules:
            console.warning(f"The {console.color(short_name, console.BOLD)} module is already installed.")
            continue  # Nothing to do

        remote_module = find_module(remote_modules, short_name, options.repository)
        if not remote_module:
            console.warning(f"The {console.color(short_name, console.BOLD)} module is not available in any repository.")
            continue  # Not available

        install_module(dependencies, installed_modules, remote_modules, remote_module, False)
        modules_installed = True

    if modules_installed:
        save_contrib_manifest(installed_modules)


def cmd_contrib_list(commands, args, start):
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument("--only-installed", help="Only list modules that are installed", action="store_true")
    parser.add_argument("--repository", help="The repository to list modules from", metavar="REPO")
    parser.add_argument("--show-masked", help="Show modules that have been masked", action="store_true")
    (options, _) = parser.parse_known_args(args[start:])

    installed_modules = load_contrib_manifest()
    remote_modules = fetch_modules(options.repository)

    module_table = []
    dependencies = installed_dependencies(installed_modules)

    for module_name, remote_variants in sorted(remote_modules.items()):
        for remote_module in remote_variants:
            available_core = dependencies.get("core")
            module_core = remote_module["depends"].get("core")
            if not module.version_in_range(available_core, module_core):
                continue  # Module is not for this core version

            installed_module = installed_modules.get(module_name)
            version_string = module.version_string(remote_module["version"])
            if installed_module:
                if installed_module["repository"] == remote_module["repository"]:
                    match module.version_compare(module.version_parse(installed_module["version"]), remote_module["version"]):
                        case -1:  # Outdated
                            version_string = console.color(version_string, console.BOLD_RED)
                        case 0:  # Up-to-date
                            version_string = console.color(version_string, console.BOLD_GREEN)
                        case 1:  # More recent than contrib
                            version_string = console.color(version_string, console.BOLD_YELLOW)

            elif options.only_installed:
                continue  # Not installed

            remote_mask = remote_module["mask"]
            if remote_mask:
                if options.show_masked:
                    module_name = console.color(module_name, console.FAINT)

                else:
                    console.debug(f"Module {module_name} {version_string} from {remote_module['repository']} is masked: {remote_mask}")
                    continue

            module_table.append([module_name, version_string, remote_module["description"]])

    if not module_table:
        console.error("No modules matching your requirements are available.")

    console.table(["Module", "Version", "Description"], module_table)


def cmd_contrib_uninstall(commands, args, start):
    if len(args) <= start:
        console.error("You must specify one or more modules to uninstall.")

    installed_modules = load_contrib_manifest()
    for module_name in args[start:]:
        short_name = module.short_name(module_name)
        long_module = module.CONTRIB_DIR / module.file_name(module_name)

        if not long_module.is_file():
            console.error(f"The {console.color(short_name, console.BOLD)} module is not installed.")

        elif short_name not in installed_modules:
            console.error(f"The {console.color(short_name, console.BOLD)} module is not a contrib module.")

        try:
            print(f"Removing the {console.color(short_name, console.BOLD)} module ...")
            long_module.unlink()
            del installed_modules[short_name]

        except OSError as e:
            console.error(f"Unable to unlink {long_module}: {e}")

    save_contrib_manifest(installed_modules)


def cmd_contrib_upgrade(commands, args, start):
    installed_modules = load_contrib_manifest()
    dependencies = installed_dependencies(installed_modules)
    remote_modules = fetch_modules(None)

    modules_to_upgrade = args[start:]
    if not modules_to_upgrade:
        modules_to_upgrade = list(installed_modules.keys())

    modules_upgraded = False
    for module_name in modules_to_upgrade:
        short_name = module.short_name(module_name)
        if short_name not in installed_modules:
            console.error(f"The {console.color(short_name, console.BOLD)} module is not installed.")

        installed_module = installed_modules[short_name]
        remote_repository = installed_module["repository"]
        remote_module = find_module(remote_modules, short_name, remote_repository)
        if not remote_module:
            console.warning(f"The {console.color(short_name, console.BOLD)} module is not available in the {remote_repository} repository.")
            continue

        installed_version = installed_module["version"]
        remote_version = remote_module["version"]
        if module.version_compare(module.version_parse(installed_version), remote_version) >= 0:
            print(f"The {console.color(short_name, console.BOLD)} module is up to date.")
            continue  # Module up to date

        try:
            with open(module.CONTRIB_DIR / module.file_name(module_name), "rb") as fh:
                file_checksum = hashlib.file_digest(fh, "sha256").hexdigest()
                installed_checksum = installed_module["checksum"]
                if installed_checksum != file_checksum:
                    console.warning(
                        f"Local checksum for the {console.color(short_name, console.BOLD)} module did not match; not updating.",
                        f"Expected checksum: {installed_checksum}",
                        f"Actual checksum:   {file_checksum}",
                    )
                    continue  # Locally modified

        except OSError as e:
            console.warning(f"Unable to checksum {short_name}: {e}")

        print(f"Upgrading the {console.color(short_name, console.BOLD)} module from {console.color(installed_version, console.BOLD)} ...")
        install_module(dependencies, installed_modules, remote_modules, remote_module, True)
        modules_upgraded = True

    if modules_upgraded:
        save_contrib_manifest(installed_modules)


def cmd_extra(commands, args, start):
    return console.subcommand(args, 2, {
        "disable": {
            "function": cmd_extra_disable,
            "help": "Disable one or more extra modules",
        },
        "enable": {
            "function": cmd_extra_enable,
            "help": "Enable one or more extra modules",
        },
        "interactive": {
            "function": cmd_extra_interactive,
            "help": "Interactively enable or disable extra modules",
        },
        "list": {
            "function": cmd_extra_list,
            "help": "List the status of extra modules",
        },
    })


def cmd_extra_disable(commands, args, start):
    if len(args) <= start:
        console.error("You must specify one or more modules to disable.")

    for module_name in args[start:]:
        short_name = module.short_name(module_name)
        long_module = module.DIR / module.file_name(module_name)
        long_extra = module.EXTRA_DIR / module.file_name(module_name)

        if not long_module.exists():
            if not long_extra.exists():
                console.error(f"The {console.color(short_name, console.BOLD)} module is not an extra module.")

            elif long_module.is_symlink() and long_extra.readlink() != long_extra:
                console.error(f"The {console.color(short_name, console.BOLD)} module is not a symlink to an extra module.")

            else:
                console.warning(f"The {console.color(short_name, console.BOLD)} module is not currently enabled.")
                continue  # Nothing to do.

        try:
            print(f"Disabling the {console.color(short_name, console.BOLD)} module ...")
            long_module.unlink(long_extra)

        except OSError as e:
            console.error(f"Unable to unlink {short_name} from {long_extra}: {e}")

    cmd_extra_list(commands, args, start)


def cmd_extra_enable(commands, args, start):
    if len(args) <= start:
        console.error("You must specify one or more modules to enable.")

    for module_name in args[start:]:
        short_name = module.short_name(module_name)
        long_extra = module.EXTRA_DIR / module.file_name(module_name)

        if not long_extra.exists():
            console.error(f"{console.color(short_name, console.BOLD)} is not an extra module!")

        long_module = module.DIR / module.file_name(module_name)
        if long_module.is_symlink():
            if long_module.readlink() == long_extra:
                console.warning(f"The {console.color(short_name, console.BOLD)} module is already enabled.")
                continue  # Nothing to do

            else:
                long_module.unlink(missing_ok=True)  # Remove the dead symlink.

        try:
            print(f"Enabling the {console.color(short_name, console.BOLD)} module ...")
            long_module.symlink_to(long_extra)

        except OSError as e:
            console.error(f"Unable to symlink {short_name} to {long_extra}: {e}")

    cmd_extra_list(commands, args, start)


def cmd_extra_interactive(commands, args, start):
    if not console.interactive():
        console.error("This command can not be used non-interactively!")

    modules_to_enable = []
    modules_to_disable = []
    for module_name in sorted(module.EXTRA_DIR.glob("*.cpp")):
        short_name = module.short_name(module_name)
        print(f"Would you like to enable the {console.color(short_name, console.BOLD)} module?")

        if console.prompt_boolean(False):
            modules_to_enable.append(short_name)
        else:
            modules_to_disable.append(short_name)

    for module_to_disable in modules_to_disable:
        long_module = module.DIR / module.file_name(module_to_disable)
        long_extra = module.EXTRA_DIR / module.file_name(module_to_disable)

        if long_module.is_symlink() and long_module.readlink() == long_extra:
            try:
                console.debug(f"Disabling the {module_to_disable} module ...")
                long_module.unlink()
            except OSError as e:
                console.warning(f"Unable to unlink {module_to_disable} from {long_extra}: {e}")

    for module_to_enable in modules_to_enable:
        long_module = module.DIR / module.file_name(module_to_enable)
        long_extra = module.EXTRA_DIR / module.file_name(module_to_enable)

        if not long_module.is_symlink() or long_module.readlink() != long_extra:
            try:
                console.debug(f"Enabling the {module_to_enable} module ...")
                long_module.symlink_to(long_extra)
            except OSError as e:
                console.warning(f"Unable to symlink {module_to_enable} to {long_extra}: {e}")


    cmd_extra_list(commands, args, start)


def cmd_extra_list(commands, args, start):
    modules = []
    for long_extra in sorted(module.EXTRA_DIR.iterdir()):
        if not long_extra.suffix == ".cpp":
            continue  # Not a module

        long_module = module.DIR / long_extra.name
        short_module = module.short_name(long_extra)
        if long_module.is_symlink():
            if long_module.readlink() == long_extra:
                modules.append([short_module, console.color("enabled", console.BOLD_GREEN)])

            else:
                modules.append([short_module, console.color("not an extra symlink", console.BOLD_YELLOW)])

        elif long_module.exists():
            modules.append([short_module, console.color("not a symlink", console.BOLD_RED)])

        else:
            modules.append([short_module, console.color("disabled", console.BOLD_YELLOW)])

    console.table(["Name", "Status"], modules)


sys.exit(console.subcommand(sys.argv, 1, {
    "contrib": {
        "function": cmd_contrib,
        "help": "Manipulate contrib modules",
    },
    "extra": {
        "function": cmd_extra,
        "help": "Manipulate extra modules",
    },

    # These commands exist for compatibility with v4 and earlier.
    "install": {
        "function": cmd_contrib_install,
    },
    "list": {
        "function": cmd_contrib_list,
    },
    "upgrade": {
        "function": cmd_contrib_upgrade,
    },
}))