/usr/lib/python3/dist-packages/provisioningserver/utils/shell.py is in python3-maas-provisioningserver 2.4.0~beta2-6865-gec43e47e6-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | # Copyright 2014-2016 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Utilities for executing external commands."""
__all__ = [
'call_and_check',
'ExternalProcessError',
'get_env_with_bytes_locale',
'get_env_with_locale',
]
import os
from pipes import quote
from string import printable
from subprocess import (
CalledProcessError,
PIPE,
Popen,
)
# A table suitable for use with str.translate() to replace each
# non-printable and non-ASCII character in a byte string with a question
# mark, mimicking the "replace" strategy when encoding and decoding.
non_printable_replace_table = "".join(
chr(i) if chr(i) in printable else "?"
for i in range(0xff + 0x01)).encode("ascii")
class ExternalProcessError(CalledProcessError):
"""Raised when there's a problem calling an external command.
Unlike `CalledProcessError`:
- `__str__()` returns a string containing the output of the failed
external process, if available, and tries to keep in valid Unicode
characters from the error message.
"""
@classmethod
def upgrade(cls, error):
"""Upgrade the given error to an instance of this class.
If `error` is an instance of :py:class:`CalledProcessError`, this will
change its class, in-place, to :py:class:`ExternalProcessError`.
There are two ways we could have done this:
1. Change the class of `error` in-place.
2. Capture ``exc_info``, create a new exception based on `error`, then
re-raise with the 3-argument version of ``raise``.
#1 seems a lot simpler so that's what this method does. The caller
needs then only use a naked ``raise`` to get the utility of this class
without losing the traceback.
"""
if type(error) is CalledProcessError:
error.__class__ = cls
@staticmethod
def _to_unicode(string):
if isinstance(string, bytes):
return string.decode("ascii", "replace")
else:
return str(string)
@staticmethod
def _to_ascii(string, table=non_printable_replace_table):
if isinstance(string, bytes):
return string.translate(table)
elif isinstance(string, str):
return string.encode("ascii", "replace").translate(table)
else:
return str(string).encode("ascii", "replace").translate(table)
def __str__(self):
cmd = " ".join(quote(self._to_unicode(part)) for part in self.cmd)
output = self._to_unicode(self.output)
return "Command `%s` returned non-zero exit status %d:\n%s" % (
cmd, self.returncode, output)
@property
def output_as_ascii(self):
"""The command's output as printable ASCII.
Non-printable and non-ASCII characters are filtered out.
"""
return self._to_ascii(self.output)
@property
def output_as_unicode(self):
"""The command's output as Unicode text.
Invalid Unicode characters are filtered out.
"""
return self._to_unicode(self.output)
def call_and_check(command, *args, **kwargs):
"""Execute a command, similar to `subprocess.check_call()`.
:param command: Command line, as a list of strings.
:return: The command's output from standard output.
:raise ExternalProcessError: If the command returns nonzero.
"""
process = Popen(command, *args, stdout=PIPE, stderr=PIPE, **kwargs)
stdout, stderr = process.communicate()
stderr = stderr.strip()
if process.returncode != 0:
raise ExternalProcessError(process.returncode, command, output=stderr)
return stdout
def has_command_available(command):
"""Return True if `command` is available on the system."""
try:
call_and_check(["which", command])
except ExternalProcessError:
return False
return True
def get_env_with_locale(environ=os.environ, locale='C.UTF-8'):
"""Return an environment dict with locale vars set (to C.UTF-8 by default).
C.UTF-8 is the new en_US.UTF-8, i.e. it's the new default locale when no
other locale makes sense.
This function takes a starting environment, by default that of the current
process, strips away all locale and language settings (i.e. LC_* and LANG)
and selects the specified locale in their place.
:param environ: A base environment to start from. By default this is
``os.environ``. It will not be modified.
:param locale: The locale to set in the environment, 'C.UTF-8' by default.
"""
environ = {
name: value for name, value in environ.items()
if not name.startswith('LC_')
}
environ.update({
'LC_ALL': locale,
'LANG': locale,
'LANGUAGE': locale,
})
return environ
def get_env_with_bytes_locale(environ=os.environb, locale=b'C.UTF-8'):
"""Return an environment dict with locale vars set (to C.UTF-8 by default).
C.UTF-8 is the new en_US.UTF-8, i.e. it's the new default locale when no
other locale makes sense.
This function takes a starting environment, by default that of the current
process, strips away all locale and language settings (i.e. LC_* and LANG)
and selects C.UTF-8 in their place.
:param environ: A base environment to start from. By default this is
``os.environb``. It will not be modified.
:param locale: The locale to set in the environment, 'C.UTF-8' by default.
"""
environ = {
name: value for name, value in environ.items()
if not name.startswith(b'LC_')
}
environ.update({
b'LC_ALL': locale,
b'LANG': locale,
b'LANGUAGE': locale,
})
return environ
|