+def _ParseAsn1Generalizedtime(value):
+ """Parses an ASN1 GENERALIZEDTIME timestamp as used by pyOpenSSL.
+
+ @type value: string
+ @param value: ASN1 GENERALIZEDTIME timestamp
+
+ """
+ m = re.match(r"^(\d+)([-+]\d\d)(\d\d)$", value)
+ if m:
+ # We have an offset
+ asn1time = m.group(1)
+ hours = int(m.group(2))
+ minutes = int(m.group(3))
+ utcoffset = (60 * hours) + minutes
+ else:
+ if not value.endswith("Z"):
+ raise ValueError("Missing timezone")
+ asn1time = value[:-1]
+ utcoffset = 0
+
+ parsed = time.strptime(asn1time, "%Y%m%d%H%M%S")
+
+ tt = datetime.datetime(*(parsed[:7])) - datetime.timedelta(minutes=utcoffset)
+
+ return calendar.timegm(tt.utctimetuple())
+
+
+def GetX509CertValidity(cert):
+ """Returns the validity period of the certificate.
+
+ @type cert: OpenSSL.crypto.X509
+ @param cert: X509 certificate object
+
+ """
+ # The get_notBefore and get_notAfter functions are only supported in
+ # pyOpenSSL 0.7 and above.
+ try:
+ get_notbefore_fn = cert.get_notBefore
+ except AttributeError:
+ not_before = None
+ else:
+ not_before_asn1 = get_notbefore_fn()
+
+ if not_before_asn1 is None:
+ not_before = None
+ else:
+ not_before = _ParseAsn1Generalizedtime(not_before_asn1)
+
+ try:
+ get_notafter_fn = cert.get_notAfter
+ except AttributeError:
+ not_after = None
+ else:
+ not_after_asn1 = get_notafter_fn()
+
+ if not_after_asn1 is None:
+ not_after = None
+ else:
+ not_after = _ParseAsn1Generalizedtime(not_after_asn1)
+
+ return (not_before, not_after)
+
+
+def SafeEncode(text):
+ """Return a 'safe' version of a source string.
+
+ This function mangles the input string and returns a version that
+ should be safe to display/encode as ASCII. To this end, we first
+ convert it to ASCII using the 'backslashreplace' encoding which
+ should get rid of any non-ASCII chars, and then we process it
+ through a loop copied from the string repr sources in the python; we
+ don't use string_escape anymore since that escape single quotes and
+ backslashes too, and that is too much; and that escaping is not
+ stable, i.e. string_escape(string_escape(x)) != string_escape(x).
+
+ @type text: str or unicode
+ @param text: input data
+ @rtype: str
+ @return: a safe version of text
+
+ """
+ if isinstance(text, unicode):
+ # only if unicode; if str already, we handle it below
+ text = text.encode('ascii', 'backslashreplace')
+ resu = ""
+ for char in text:
+ c = ord(char)
+ if char == '\t':
+ resu += r'\t'
+ elif char == '\n':
+ resu += r'\n'
+ elif char == '\r':
+ resu += r'\'r'
+ elif c < 32 or c >= 127: # non-printable
+ resu += "\\x%02x" % (c & 0xff)
+ else:
+ resu += char
+ return resu
+
+
+def UnescapeAndSplit(text, sep=","):
+ """Split and unescape a string based on a given separator.
+
+ This function splits a string based on a separator where the
+ separator itself can be escape in order to be an element of the
+ elements. The escaping rules are (assuming coma being the
+ separator):
+ - a plain , separates the elements
+ - a sequence \\\\, (double backslash plus comma) is handled as a
+ backslash plus a separator comma
+ - a sequence \, (backslash plus comma) is handled as a
+ non-separator comma
+
+ @type text: string
+ @param text: the string to split
+ @type sep: string
+ @param text: the separator
+ @rtype: string
+ @return: a list of strings
+
+ """
+ # we split the list by sep (with no escaping at this stage)
+ slist = text.split(sep)
+ # next, we revisit the elements and if any of them ended with an odd
+ # number of backslashes, then we join it with the next
+ rlist = []
+ while slist:
+ e1 = slist.pop(0)
+ if e1.endswith("\\"):
+ num_b = len(e1) - len(e1.rstrip("\\"))
+ if num_b % 2 == 1:
+ e2 = slist.pop(0)
+ # here the backslashes remain (all), and will be reduced in
+ # the next step
+ rlist.append(e1 + sep + e2)
+ continue
+ rlist.append(e1)
+ # finally, replace backslash-something with something
+ rlist = [re.sub(r"\\(.)", r"\1", v) for v in rlist]
+ return rlist
+
+
+def CommaJoin(names):
+ """Nicely join a set of identifiers.
+
+ @param names: set, list or tuple
+ @return: a string with the formatted results
+
+ """
+ return ", ".join([str(val) for val in names])
+
+
+def BytesToMebibyte(value):
+ """Converts bytes to mebibytes.
+
+ @type value: int
+ @param value: Value in bytes
+ @rtype: int
+ @return: Value in mebibytes
+
+ """
+ return int(round(value / (1024.0 * 1024.0), 0))
+
+
+def CalculateDirectorySize(path):
+ """Calculates the size of a directory recursively.
+
+ @type path: string
+ @param path: Path to directory
+ @rtype: int
+ @return: Size in mebibytes
+
+ """
+ size = 0
+
+ for (curpath, _, files) in os.walk(path):
+ for filename in files:
+ st = os.lstat(PathJoin(curpath, filename))
+ size += st.st_size
+
+ return BytesToMebibyte(size)
+
+
+def GetFilesystemStats(path):
+ """Returns the total and free space on a filesystem.
+
+ @type path: string
+ @param path: Path on filesystem to be examined
+ @rtype: int
+ @return: tuple of (Total space, Free space) in mebibytes
+
+ """
+ st = os.statvfs(path)
+
+ fsize = BytesToMebibyte(st.f_bavail * st.f_frsize)
+ tsize = BytesToMebibyte(st.f_blocks * st.f_frsize)
+ return (tsize, fsize)
+
+
+def RunInSeparateProcess(fn, *args):
+ """Runs a function in a separate process.
+
+ Note: Only boolean return values are supported.
+
+ @type fn: callable
+ @param fn: Function to be called
+ @rtype: bool
+ @return: Function's result
+
+ """
+ pid = os.fork()
+ if pid == 0:
+ # Child process
+ try:
+ # In case the function uses temporary files
+ ResetTempfileModule()
+
+ # Call function
+ result = int(bool(fn(*args)))
+ assert result in (0, 1)
+ except: # pylint: disable-msg=W0702
+ logging.exception("Error while calling function in separate process")
+ # 0 and 1 are reserved for the return value
+ result = 33
+
+ os._exit(result) # pylint: disable-msg=W0212
+
+ # Parent process
+
+ # Avoid zombies and check exit code
+ (_, status) = os.waitpid(pid, 0)
+
+ if os.WIFSIGNALED(status):
+ exitcode = None
+ signum = os.WTERMSIG(status)
+ else:
+ exitcode = os.WEXITSTATUS(status)
+ signum = None
+
+ if not (exitcode in (0, 1) and signum is None):
+ raise errors.GenericError("Child program failed (code=%s, signal=%s)" %
+ (exitcode, signum))
+
+ return bool(exitcode)
+
+