Simple makecatalogs-based script to perform some level of Munki-specific linting on pkginfo files

munki_fuzzinator.py

#!/usr/bin/env python
# encoding: utf-8
#
# Copyright 2014 - The Regents of the University of Michigan.
#
# 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.
"""
fuzzinator

Created by Pepijn Bruienne on 2/19/2014.
Most code reused from 'makecatalogs', part of the Munki suite.

Recursively scans a directory, looking for installer item info files.
Performs various sanity checks on them and posts errors to a Hipchat room.

Assumes a pkgsinfo directory under repopath.
Requires 'python-simple-hipchat' module:
    pip install python-simple-hipchat
"""

import sys
import os
import optparse
import hipchat

try:
    from munkilib import FoundationPlist as plistlib
    LOCAL_PREFS_SUPPORT = True
except ImportError:
    try:
        import FoundationPlist as plistlib
        LOCAL_PREFS_SUPPORT = True
    except ImportError:
        # maybe we're not on an OS X machine...
        import plistlib
        LOCAL_PREFS_SUPPORT = False

try:
    from munkilib.munkicommon import listdir, get_version
except ImportError:
    # munkilib is not available
    def listdir(path):
        """OSX HFS+ string encoding safe listdir().

        Args:
            path: path to list contents of
        Returns:
            list of contents, items as str or unicode types
        """
        # if os.listdir() is supplied a unicode object for the path,
        # it will return unicode filenames instead of their raw fs-dependent
        # version, which is decomposed utf-8 on OSX.
        #
        # we use this to our advantage here and have Python do the decoding
        # work for us, instead of decoding each item in the output list.
        #
        # references:
        # http://docs.python.org/howto/unicode.html#unicode-filenames
        # http://developer.apple.com/library/mac/#qa/qa2001/qa1235.html
        # http://lists.zerezo.com/git/msg643117.html
        # http://unicode.org/reports/tr15/    section 1.2
        if type(path) is str:
            path = unicode(path, 'utf-8')
        elif type(path) is not unicode:
            path = unicode(path)
        return os.listdir(path)

    def get_version():
        '''Placeholder if munkilib is not available'''
        return 'UNKNOWN'


def hipchat_err_utf8(text, color):
    '''Print Unicode text to Hipchat room as UTF-8'''
    HIPLINT.message_room(HC_ROOM, HC_NICK, text.encode('UTF-8'), 'text', color)


def lintcatalogs(repopath, options):
    '''Assembles all pkginfo files into catalogs.
    Assumes a pkgsinfo directory under repopath.
    Sends any errors that were encountered to the Hipchat channel
    '''

    # Make sure the pkgsinfo directory exists
    pkgsinfopath = os.path.join(repopath, 'pkgsinfo')
    # make sure pkgsinfopath is Unicode so that os.walk later gives us
    # Unicode names back.
    if type(pkgsinfopath) is str:
        pkgsinfopath = unicode(pkgsinfopath, 'utf-8')
    elif type(pkgsinfopath) is not unicode:
        pkgsinfopath = unicode(pkgsinfopath)

    if not os.path.exists(pkgsinfopath):
        exit(-1)

    # Set a default exit code
    exitCode = 0

    errors = []
    catalogs = {}
    catalogs['all'] = []

    # Walk through the pkginfo files
    for dirpath, dirnames, filenames in os.walk(pkgsinfopath):
        for dirname in dirnames:
            # don't recurse into directories that start
            # with a period.
            if dirname.startswith('.'):
                dirnames.remove(dirname)
        for filename in filenames:
            if filename.startswith('.'):
                # skip files that start with a period as well
                continue
            elif filename.endswith('.sh'):
                # skip files that are executables (sh)
                continue

            filepath = os.path.join(dirpath, filename)

            # print 'Now checking: ' + filepath

            # Try to read the pkginfo file
            try:
                pkginfo = plistlib.readPlist(filepath)
                # don't copy admin notes to catalogs.
                if pkginfo.get('notes'):
                    del(pkginfo['notes'])
            except IOError, inst:
                errors.append("IO error for %s: %s" % (filepath, inst))
                exitCode = -1
                continue
            except Exception, inst:
                errors.append("Unexpected error for %s: %s" % (filepath, inst))
                exitCode = -1
                continue

            # simple sanity checking
            do_pkg_check = True
            installer_type = pkginfo.get('installer_type')
            if installer_type in ['nopkg', 'apple_update_metadata']:
                do_pkg_check = False
            if pkginfo.get('PackageCompleteURL'):
                do_pkg_check = False
            if pkginfo.get('PackageURL'):
                do_pkg_check = False

            if do_pkg_check:
                if not 'installer_item_location' in pkginfo:
                    errors.append(
                        "(unknown) file %s is missing installer_item_location"
                        % filepath[len(pkgsinfopath) + 1:])

                # Check for a well-formed 'update_for' array
                if type(pkginfo.get('update_for')) is str:
                    errors.append(
                        "(unknown) file %s has an incorrect update_for key (string, not array of strings)"
                        % filepath[len(pkgsinfopath) + 1:])

                # Check for a well-formed 'requires' array
                if type(pkginfo.get('requires')) is str:
                    errors.append(
                        "(unknown) file %s has an incorrect 'requires' key (string, not array of strings)"
                        % filepath[len(pkgsinfopath) + 1:])

                # Check for a well-formed 'blocking_applications' array
                if type(pkginfo.get('blocking_applications')) is str:
                    errors.append(
                        "(unknown) file %s has an incorrect 'blocking_applications' key (string, not array of strings)"
                        % filepath[len(pkgsinfopath) + 1:])

                # Check for a well-formed 'catalogs' array
                if type(pkginfo.get('catalogs')) is str:
                    errors.append(
                        "(unknown) file %s has an incorrect 'catalogs' key (string, not array of strings)"
                        % filepath[len(pkgsinfopath) + 1:])

                # Check for a well-formed 'supported_architectures' array
                if type(pkginfo.get('supported_architectures')) is str:
                    errors.append(
                        "(unknown) file %s has an incorrect 'supported_architectures' key (string, not array of strings)"
                        % filepath[len(pkgsinfopath) + 1:])

                # Try to form a path and fail if the
                # installer_item_location is not a valid type
                try:
                    installeritempath = os.path.join(repopath, "pkgs",
                                                     pkginfo['installer_item_location'])
                except TypeError:
                    errors.append("(unknown) invalid installer_item_location"
                                  " in info file %s" % filepath[len(pkgsinfopath) + 1:])
                    exitCode = -1
                    continue

                # Check if the installer item actually exists
                if not os.path.exists(installeritempath):
                    errors.append("(unknown) Info file %s refers to "
                                  "missing installer item: %s" %
                                  (filepath[len(pkgsinfopath) + 1:],
                                   pkginfo['installer_item_location']))

                # Check for trailing slash in items_to_copy keys
                if pkginfo.get('items_to_copy'):
                    for itemtocopy in pkginfo.get('items_to_copy'):
                        if itemtocopy.get('destination_path').endswith('/'):
                            errors.append("(unknown) File %s contains a "
                                          "items_to_copy key with a trailing "
                                          "slash (%s), please edit!"
                                          % (filepath[len(pkgsinfopath) + 1:],
                                             itemtocopy.get('destination_path')))

    if errors:
        # group all errors at the end for better visibility
        return errors
    else:
        return None


def pref(prefname):
    """Returns a preference for prefname"""
    if not LOCAL_PREFS_SUPPORT:
        return None
    try:
        _prefs = plistlib.readPlist(PREFSPATH)
    except Exception:
        return None
    if prefname in _prefs:
        return _prefs[prefname]
    else:
        return None


PREFSNAME = 'com.googlecode.munki.munkiimport.plist'
PREFSPATH = os.path.expanduser(os.path.join('~/Library/Preferences',
                                            PREFSNAME))
HC_TOKEN = 'YOUR HC TOKEN HERE'
HC_NICK = 'Fuzzinator'
HC_ROOM = 9999999999 # Your HC Room ID here

HIPLINT = hipchat.HipChat(token=HC_TOKEN)


def main():
    '''Main'''
    usage = "usage: %prog [options] [/path/to/repo_root]"
    p = optparse.OptionParser(usage=usage)
    p.add_option('--auto', '-a', action='store_true',
                 help='Auto-run mode, prints info line to Hipchat.')
    options, arguments = p.parse_args()

    # Make sure we have a path to work with
    repopath = None
    if len(arguments) == 0:
        repopath = pref('repo_path')
        if not repopath:
            exit(-1)
    else:
        repopath = arguments[0].rstrip("/")

    # Make sure the repo path exists
    if not os.path.exists(repopath):
        exit(-1)

    # Make the catalogs
    errors = lintcatalogs(repopath, options)

    if errors:
        color = 'red'
        if options.auto:
            hipchat_err_utf8(
                '(failed) Auto-run of Fuzzinator encountered the following issues that require attention', color)
        else:
            hipchat_err_utf8(
                '(failed) Manual run of Fuzzinator encountered the following issues that require attention', color)
        for error in errors:
            hipchat_err_utf8(error, color)
    else:
        color = 'green'
        if options.auto:
            hipchat_err_utf8(
                '(successful) And there was much rejoicing! Auto-run of Fuzzinator encountered no errors. (yey)', color)
        else:
            hipchat_err_utf8(
                '(successful) And there was much rejoicing! Manual run of Fuzzinator encountered no errors. (yey)', color)

if __name__ == '__main__':
    main()