Source code for plainbox.impl.secure.job

# This file is part of Checkbox.
#
# Copyright 2013 Canonical Ltd.
# Written by:
#   Sylvain Pineau <sylvain.pineau@canonical.com>
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.

#
# Checkbox 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 Checkbox.  If not, see <http://www.gnu.org/licenses/>.

"""
:mod:`plainbox.impl.secure.job` -- secure code for job definitions
==================================================================

.. warning::

    THIS MODULE DOES NOT HAVE STABLE PUBLIC API
"""

import collections
import hashlib
import json
import os
import re


[docs]class BaseJob: """ Base Job definition class. """ def __init__(self, data): self.__data = data self._checksum = None @property def _data(self): raise AttributeError("Hey, poking at _data is forbidden!")
[docs] def get_record_value(self, name, default=None): """ Obtain the value of the specified record attribute """ return self.__data.get(name, default)
@property
[docs] def plugin(self): return self.get_record_value('plugin')
@property
[docs] def command(self): return self.get_record_value('command')
@property
[docs] def environ(self): return self.get_record_value('environ')
@property
[docs] def user(self): return self.get_record_value('user')
@property
[docs] def checksum(self): """ Checksum of the job definition. This property can be used to compute the checksum of the canonical form of the job definition. The canonical form is the UTF-8 encoded JSON serialization of the data that makes up the full definition of the job (all keys and values). The JSON serialization uses no indent and minimal separators. The checksum is defined as the SHA256 hash of the canonical form. """ if self._checksum is None: self._checksum = self._compute_checksum() return self._checksum
def _compute_checksum(self): """ Compute the value for :attr:`checksum`. """ # Ideally we'd use simplejson.dumps() with sorted keys to get # predictable serialization but that's another dependency. To get # something simple that is equally reliable, just sort all the keys # manually and ask standard json to serialize that.. sorted_data = collections.OrderedDict(sorted(self.__data.items())) # Compute the canonical form which is arbitrarily defined as sorted # json text with default indent and separator settings. canonical_form = json.dumps( sorted_data, indent=None, separators=(',', ':')) # Compute the sha256 hash of the UTF-8 encoding of the canonical form # and return the hex digest as the checksum that can be displayed. return hashlib.sha256(canonical_form.encode('UTF-8')).hexdigest()
[docs] def get_environ_settings(self): """ Return a set of requested environment variables """ if self.environ is not None: return {variable for variable in re.split('[\s,]+', self.environ)} else: return set()