Download the latest version of MS Teams and upload to Jamf

MSTeams.py

#!/usr/bin/env python3
#pylint: disable=invalid-name, attribute-defined-outside-init
#pylint: disable=too-many-statements, too-many-locals
"""
MsTeams.py v2.0
Tony Williams 2020-10-12

This is built to handle MS Teams
"""

from os import path
import subprocess
import plistlib
import xml.etree.ElementTree as ET
import datetime
import logging
import logging.handlers
from time import sleep
from urllib.request import urlretrieve
import requests

LOGFILE = '/usr/local/var/log/MSTeams.log'
LOGLEVEL = logging.DEBUG
# ID of Teams in patchsoftwaretitles in Jamf server - note it's a string to save
# a conversion call when we use it
PATCHID = '48'
APPNAME = 'MSTeams'

__all__ = [APPNAME]

#pylint: disable=unnecessary-pass
class TeamsError(Exception):
    """ An Error """
    pass


class MSTeams():
    """Checks to see if a new version of MS Teams exists and then downloads it if it does
    """

    description = __doc__

    def load_prefs(self):
        """ load the preferences from file """
        plist = path.expanduser("~/Library/Preferences/JPCImporter.plist")
        prefs = plistlib.load(open(plist, "rb"))
        self.url = prefs["url"]
        self.auth = (prefs["user"], prefs["password"])
        self.base = self.url + '/JSSResource'

    def setup_logging(self):
        """Defines a nicely formatted logger"""
        self.logger = logging.getLogger(APPNAME)
        self.logger.setLevel(LOGLEVEL)
        handler = logging.handlers.TimedRotatingFileHandler(
            LOGFILE, when="D", interval=1, backupCount=7
        )
        handler.setFormatter(
            logging.Formatter(
                "%(asctime)s %(levelname)s %(message)s",
                datefmt="%Y-%m-%d %H:%M:%S",
            )
        )
        self.logger.addHandler(handler)

    def latest(self):
        """Find out the latest version"""
        url = self.base + '/patchsoftwaretitles/id/' + PATCHID
        self.logger.debug("About to request %s", url)
        ret = requests.get(url, auth=self.auth)
        if ret.status_code != 200:
            raise TeamsError(
                "Patch title download failed: {} : {}".format(
                    ret.status_code, url))
        patch = ET.fromstring(ret.text)
        first = patch.findall('versions/version')[0]
        version = first.findtext('software_version')
        self.logger.debug("Got version %s", version)
        return version

    def check(self, version):
        """Check if a version is already uploaded"""
        url = self.base + '/packages/name/Microsoft_Teams-{}.pkg'.format(version)
        self.logger.debug("About to request %s", url)
        ret = requests.get(url, auth=self.auth)
        if ret.status_code == 200:
            self.logger.warning("Found package")
            return True
        return False

    def download(self, version):
        """downloads a version of Teams given the version string"""
        url = 'https://statics.teams.microsoft.com/production-osx/{}/Teams_osx.pkg'.format(version)
        self.logger.debug("About to download %s", url)
        dst = path.expanduser('~/Documents/') + 'Microsoft_Teams-{}.pkg'.format(version)
        # the below is going to take *some time*
        urlretrieve(url, dst)
        return dst

    def upload(self, pkg_path):
        """Upload the package `pkg_path` and returns the ID returned by JPC"""
        # pkg_path = 'Microsoft_Teams-{}.pkg'.format(version)
        self.logger.warning("Starting upload %s", pkg_path)
        pkg = path.basename(pkg_path)
        app = pkg.split('-')[0]
        category = 'Microsoft'
        date = datetime.datetime.now().strftime('%d-%b-%G')
        # use curl for the file upload as it seems to work nicer than requests
        # for this ridiculous workaround for file uploads.
        url = self.url + '/dbfileupload'
        auth = self.auth[0] + ":" + self.auth[1]
        command = ["curl", "-u", auth, '-s']
        command += ['-X', 'POST', url]
        command += ['--header', 'DESTINATION: 0']
        command += ['--header', 'OBJECT_ID: -1']
        command += ['--header', 'FILE_TYPE: 0']
        command += ['--header', 'FILE_NAME: {}'.format(pkg)]
        command += ['--upload-file', pkg_path]
        self.logger.debug("curl %s", command)
        ret = subprocess.check_output(command)
        self.logger.debug("Curl returned %s", ret)
        packid = ET.fromstring(ret).findtext('id')
        if packid == '':
            raise TeamsError(
                "curl failed for url :{}".format(url))
        self.logger.debug("Uploaded and got ID: %s", packid)
        data = "<package><id>{}</id>".format(packid)
        data += "<category>{}</category>".format(category)
        data += "<notes>Built by Autopkg. "
        data += "Uploaded {}</notes></package>".format(date)
        # we use requests for all the other API calls as it codes nicer
        # update the package details
        hdrs = {
            'Accept': 'application/xml',
            'Content-type': 'application/xml',
        }
        url = self.base + '/packages/id/{}'.format(packid)
        # we set up some retries as sometimes the server takes a minute to
        # settle with a new package upload (Can we have an API that allows
        # for an upload and setting this all in one go.)
        count = 0
        while True:
            count += 1
            ret = requests.put(url, auth=self.auth, headers=hdrs, data=data)
            self.logger.debug("package update attempt %s", count)
            self.logger.debug("Return: %s URL: %s", ret.status_code, url)
            if ret.status_code == 201:
                break
            if count > 10:
                raise TeamsError(
                    "Package update failed with code: %s" %
                    ret.status_code)
            sleep(20)
        policy_name = 'TEST-{}'.format(app)
        url = self.base + '/policies/name/{}'.format(policy_name)
        ret = requests.get(url, auth=self.auth)
        if ret.status_code != 200:
            raise TeamsError(
                "Policy %s get failed: %s" % (
                    url, ret.status_code))
        self.logger.debug("policy found")
        root = ET.fromstring(ret.text)
        root.find(
            'package_configuration/packages/package/id'
            ).text = str(packid)
        root.find('general/enabled').text = 'true'
        root.find(
            'package_configuration/packages/package/name').text = pkg
        idn = root.findtext('general/id')
        url = self.base + '/policies/id/{}'.format(idn)
        data = ET.tostring(root)
        self.logger.debug("About to put %s", url)
        ret = requests.put(url, auth=self.auth, data=data)
        if ret.status_code != 201:
            raise TeamsError(
                "Policy %s update failed: %s" % (
                    url, ret.status_code))
        pol_id = ET.fromstring(ret.text).findtext('id')
        self.logger.debug("got pol id: %s", pol_id)
        self.logger.info("Done Package: %s Pol: %s", pkg, pol_id)
        return pol_id

    def MSTeams(self):
        """The process"""
        self.setup_logging()
        self.load_prefs()
        self.logger.info("Starting run")
        version = self.latest()
        if not self.check(version):
            self.upload(self.download(version))

if __name__ == "__main__":
    MSTeams = MSTeams()
    MSTeams.MSTeams()