MunkiImporterSlacker.py

#!/usr/local/autopkg/python
"""See docstring for MunkiImporterSlacker class

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Takes elements from:
- https://github.com/autopkg/grahampugh-recipes/blob/main/JamfUploaderProcessors/JamfUploaderSlacker.py
- https://gist.github.com/devStepsize/b1b795309a217d24566dcc0ad136f784
- https://github.com/autopkg/nmcspadden-recipes/blob/master/PostProcessors/Yo.py
"""

import json

from time import sleep
from autopkglib import URLGetter, ProcessorError

__all__ = ["MunkiImporterSlacker"]


class MunkiImporterSlacker(URLGetter):
    """A postprocessor for AutoPkg that will send details about a recipe run
    to a Slack webhook based on the output of MunkiImporter."""
    description = __doc__
    input_variables = {
        "munki_info": {
            "description": "The pkginfo property list. Empty if item not imported.",
            "required": False
        },
        "slack_channel": {
            "description": "Slack channel (for overriding the default).",
            "required": False
        },
        "slack_icon_emoji": {
            "description": "Slack display emoji markup.",
            "required": False
        },
        "slack_icon_url": {
            "default": "",
            "description": "Slack display icon URL.",
            "required": False
        },
        "slack_username": {
            "default": "AutoPkg",
            "description": "Slack message display name.",
            "required": False
        },
        "slack_webhook_url": {
            "description": "Slack webhook.",
            "required": True
        }
    }
    output_variables = {}

    def slack_status_check(self, proc_stdout):
        """Return a message dependent on the HTTP response"""
        if proc_stdout == "ok":
            self.output("Slack webhook sent successfully")
            return "break"
        else:
            self.output("WARNING: Slack webhook failed to send")
            self.output(proc_stdout, verbose_level=2)

    def main(self):
        pkginfo = self.env.get("munki_info")
        if not pkginfo:
            self.output("Nothing to report to Slack")
            return

        catalogs = ",".join(pkginfo["catalogs"])
        name = pkginfo["name"]
        version = pkginfo["version"]

        slack_channel = self.env.get("slack_channel") or ""
        slack_icon_emoji = self.env.get("slack_icon_emoji") or ""
        slack_icon_url = self.env.get("slack_icon_url") or ""
        slack_username = self.env.get("slack_username")
        slack_webhook_url = self.env.get("slack_webhook_url")

        self.output(f"Name: {name}")
        self.output(f"Version: {version}")
        self.output(f"Catalogs: {catalogs}")

        slack_text = (
            "*New Item imported to Munki:*\n"
            + f"Name: *{name}*\n"
            + f"Version: *{version}*\n"
            + f"Catalogs: *{catalogs}*"
        )

        slack_data = {
            "text": slack_text,
            "username": slack_username,
        }
        if slack_channel:
            slack_data["channel"] = slack_channel
        if slack_icon_emoji:
            slack_data["icon_emoji"] = slack_icon_emoji
        if slack_icon_url:
            slack_data["icon_url"] = slack_icon_url

        slack_json = json.dumps(slack_data)

        count = 0
        while True:
            count += 1
            self.output(
                "Slack webhook post attempt {}".format(count),
                verbose_level=2
            )
            curl_cmd = [
                self.curl_binary(),
                "--data",
                slack_json,
                "--header",
                "Content-Type: application/json",
                "--request",
                "POST",
                "--show-error",
                "--silent",
                slack_webhook_url
            ]
            proc_stdout, proc_stderr, = self.execute_curl(curl_cmd)
            # check HTTP response
            if self.slack_status_check(proc_stdout) == "break":
                break
            if count >= 5:
                self.output(
                    "Slack webhook send did not succeed after 5 attempts")
                self.output(f"HTTP Response: {proc_stderr}")
                raise ProcessorError("ERROR: Slack webhook failed to send")
            sleep(10)


if __name__ == "__main__":
    PROCESSOR = MunkiImporterSlacker()
    PROCESSOR.execute_shell()