bitbot-3.11-fork/modules/git_webhooks/__init__.py

#--depends-on channel_access
#--depends-on check_mode
#--depends-on commands
#--depends-on shorturl

import itertools, json, re, urllib.parse
from src import ModuleManager, utils
from . import colors, gitea, github, gitlab

FORM_ENCODED = "application/x-www-form-urlencoded"

DEFAULT_EVENT_CATEGORIES = [
    "ping", "code", "pr", "issue", "repo"
]

@utils.export("channelset", utils.BoolSetting("git-prevent-highlight",
    "Enable/disable preventing highlights"))
@utils.export("channelset", utils.BoolSetting("git-hide-organisation",
    "Hide/show organisation in repository names"))
@utils.export("channelset", utils.BoolSetting("git-hide-prefix",
    "Hide/show command-like prefix on git webhook outputs"))
@utils.export("channelset", utils.BoolSetting("git-shorten-urls",
    "Weather or not git webhook URLs should be shortened"))
@utils.export("botset", utils.BoolSetting("git-show-private",
    "Whether or not to show git activity for private repositories"))
class Module(ModuleManager.BaseModule):
    _name = "Webhooks"

    def on_load(self):
        self._github = github.GitHub(self.log)
        self._gitea = gitea.Gitea()
        self._gitlab = gitlab.GitLab()

    @utils.hook("api.post.github")
    def _api_github_webhook(self, event):
        return self._webhook("github", "GitHub", self._github,
            event["data"], event["headers"], event["params"])

    @utils.hook("api.post.gitea")
    def _api_gitea_webhook(self, event):
        return self._webhook("gitea", "Gitea", self._gitea,
            event["data"], event["headers"], event["params"])

    @utils.hook("api.post.gitlab")
    def _api_gitlab_webhook(self, event):
        return self._webhook("gitlab", "GitLab", self._gitlab,
            event["data"], event["headers"], event["params"])

    def _webhook(self, webhook_type, webhook_name, handler, payload_str,
            headers, params):
        payload = payload_str.decode("utf8")
        if headers["Content-Type"] == FORM_ENCODED:
            payload = urllib.parse.unquote(urllib.parse.parse_qs(payload)[
                "payload"][0])
        data = json.loads(payload)

        if handler.is_private(data, headers) and not self.bot.get_setting(
                "git-show-private", False):
            return {"state": "success", "deliveries": 0}

        full_name, repo_username, repo_name, organisation = handler.names(
            data, headers)

        full_name_lower = (full_name or "").lower()
        repo_username_lower = (repo_username or "").lower()
        repo_name_lower = (repo_name or "").lower()
        organisation_lower = (organisation or "").lower()

        branch = handler.branch(data, headers)
        current_event, event_action = handler.event(data, headers)

        unfiltered_targets = []
        if "channels" in params:
            channels = params["channels"].split(",")
            for channel in params["channels"].split(","):
                server, _, channel_name = channel.partition(":")
                if server and channel_name:
                    server = self.bot.get_server_by_alias(server)

                    if server and channel_name in server.channels:
                        channel = server.channels.get(channel_name)
                        hooks = channel.get_setting("git-webhooks", {})

                        if hooks:
                            found_hook = self._find_hook(
                                full_name_lower, repo_username_lower,
                                organisation_lower, hooks)

                            if found_hook:
                                unfiltered_targets.append([
                                    server, channel, found_hook])
        else:
            unfiltered_targets = self._find_targets(full_name_lower,
                repo_username_lower, organisation_lower)

        repo_hooked = bool(unfiltered_targets)
        targets = []
        for server, channel, hook in unfiltered_targets:
            if (branch and
                    hook["branches"] and
                    not branch in hook["branches"]):
                continue

            events = []
            for hooked_event in hook["events"]:
                events.append(handler.event_categories(hooked_event))
            events = list(itertools.chain(*events))

            if (current_event in events or
                    (event_action and event_action in events)):
                targets.append([server, channel])

        if not targets:
            if not repo_hooked:
                return None
            else:
                return {"state": "success", "deliveries": 0}

        outputs = handler.webhook(full_name, current_event, data, headers)

        if outputs:
            for server, channel in targets:
                source = full_name or organisation
                hide_org = channel.get_setting("git-hide-organisation", False)
                if repo_name and hide_org:
                    source = repo_name

                for output, url in outputs:
                    output = "(%s) %s" % (
                        utils.irc.color(source, colors.COLOR_REPO), output)

                    if url:
                        if channel.get_setting("git-shorten-urls", False):
                            url = self.exports.get_one("shorturl")(server, url,
                                context=channel) or url
                        output = "%s - %s" % (output, url)

                    if channel.get_setting("git-prevent-highlight", False):
                        output = self._prevent_highlight(server, channel,
                            output)

                    hide_prefix = channel.get_setting("git-hide-prefix", False)
                    self.events.on("send.stdout").call(target=channel,
                        module_name=webhook_name, server=server, message=output,
                        hide_prefix=hide_prefix)

        return {"state": "success", "deliveries": len(targets)}

    def _prevent_highlight(self, server, channel, s):
        for user in channel.users:
            if len(user.nickname) == 1:
                # if we don't ignore 1-letter nicknames, the below while loop
                # will fire indefininitely.
                continue

            regex = re.compile(r"([0-9]|\W)(%s)(%s)" % (
                re.escape(user.nickname[0]), re.escape(user.nickname[1:])),
                re.I)
            s = regex.sub("\\1\\2\u200c\\3", s)

        return s

    def _find_targets(self, full_name_lower, repo_username_lower,
            organisation_lower):
        hooks = self.bot.database.channel_settings.find_by_setting(
            "git-webhooks")
        targets = []
        for server_id, channel_name, hooked_repos in hooks:
            found_hook = self._find_hook(full_name_lower, repo_username_lower,
                organisation_lower, hooked_repos)
            server = self.bot.get_server_by_id(server_id)
            if found_hook and server and channel_name in server.channels:
                channel = server.channels.get(channel_name)
                targets.append([server, channel, found_hook])

        return targets

    def _find_hook(self, full_name_lower, repo_username_lower,
            organisation_lower, hooks):
        hooked_repos_lower = {k.lower(): v for k, v in hooks.items()}
        if full_name_lower and full_name_lower in hooked_repos_lower:
            return hooked_repos_lower[full_name_lower]
        elif (repo_username_lower and
                repo_username_lower in hooked_repos_lower):
            return hooked_repos_lower[repo_username_lower]
        elif (organisation_lower and
                organisation_lower in hooked_repos_lower):
            return hooked_repos_lower[organisation_lower]


    @utils.hook("received.command.webhook", min_args=1, channel_only=True)
    def github_webhook(self, event):
        """
        :help: Add/remove/modify a git webhook
        :require_mode: high
        :require_access: git-webhook
        :permission: gitoverride
        :usage: list
        :usage: add <hook>
        :usage: remove <hook>
        :usage: events <hook> [category [category ...]]
        :usage: branches <hook> [branch [branch ...]]
        """
        all_hooks = event["target"].get_setting("git-webhooks", {})
        hook_name = None
        existing_hook = None
        if len(event["args_split"]) > 1:
            hook_name = event["args_split"][1]
            for existing_hook_name in all_hooks.keys():
                if existing_hook_name.lower() == hook_name.lower():
                    existing_hook = existing_hook_name
                    break

        success_message = None

        subcommand = event["args_split"][0].lower()
        if subcommand == "list":
            event["stdout"].write("Registered webhooks: %s" %
                ", ".join(all_hooks.keys()))
        elif subcommand == "add":
            if existing_hook:
                raise utils.EventError("There's already a hook for %s" %
                    hook_name)

            all_hooks[hook_name] = {
                "events": DEFAULT_EVENT_CATEGORIES.copy(),
                "branches": [],
            }
            success_message = "Added hook for %s" % hook_name

        elif subcommand == "remove":
            if not existing_hook:
                raise utils.EventError("No hook found for %s" % hook_name)

            del all_hooks[existing_hook]
            success_message = "Removed hook for %s" % hook_name

        elif subcommand == "events":
            if not existing_hook:
                raise utils.EventError("No hook found for %s" % hook_name)

            if len(event["args_split"]) < 3:
                event["stdout"].write("Events for hook %s: %s" %
                    (hook_name, " ".join(all_hooks[existing_hook]["events"])))
            else:
                new_events = [e.lower() for e in event["args_split"][2:]]
                all_hooks[existing_hook]["events"] = new_events
                success_message = "Updated events for hook %s" % hook_name
        elif subcommand == "branches":
            if not existing_hook:
                raise utils.EventError("No hook found for %s" % hook_name)

            if len(event["args_split"]) < 3:
                branches = ",".join(all_hooks[existing_hook]["branches"])
                event["stdout"].write("Branches shown for hook %s: %s" %
                    (hook_name, branches))
            else:
                all_hooks[existing_hook]["branches"] = event["args_split"][2:]
                success_message = "Updated branches for hook %s" % hook_name
        else:
            event["stderr"].write("Unknown command '%s'" %
                event["args_split"][0])

        if not success_message == None:
            if all_hooks:
                event["target"].set_setting("git-webhooks", all_hooks)
            else:
                event["target"].del_setting("git-webhooks")

            event["stdout"].write(success_message)