# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""A test verifying Address Space Layout Randomization

Uses system calls to get important pids and then gets information about
the pids in /proc/<pid>/maps. Restarts the tested processes and reads
information about them again. If ASLR is enabled, memory mappings should
change.
"""

from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import upstart

import logging
import time
import pprint
import re

def _pidof(exe_name):
    """Returns the PID of the first process with the given name."""
    pid = utils.system_output('pidof %s' % exe_name, ignore_status=True).strip()
    if len(pid.split()) > 1:
        pid = pid.split()[0]

    return pid


class Process(object):
    """Holds information about a process.

    Stores basic information about a process. This class is a base for
    UpstartProcess and SystemdProcess declared below.

    Attributes:
        _name: String name of process.
        _service_name: Name of the service corresponding to the process.
        _parent: String name of process's parent. Defaults to None.
    """

    _START_POLL_INTERVAL_SECONDS = 1
    _START_TIMEOUT = 30

    def __init__(self, name, service_name, parent=None):
        self._name = name
        self._service_name = service_name
        self._parent = parent

    def get_name(self):
        return self._name

    def get_pid(self):
        """Gets pid of process, waiting for it if not found.

        Raises:
            error.TestFail: corresponding process is not found.
        """
        retries = 0
        ps_results = ""
        while retries < self._START_TIMEOUT:
            if self._parent is None:
                ps_results = _pidof(self._name)
            else:
                ppid = _pidof(self._parent)
                get_pid_command = ('ps -C %s -o pid,ppid | grep " %s$"'
                    ' | awk \'{print $1}\'') % (self._name, ppid)
                ps_results = utils.system_output(get_pid_command).strip()

            if ps_results != "":
                return ps_results

            # The process could not be found. We then sleep, hoping the
            # process is just slow to initially start.
            time.sleep(self._START_POLL_INTERVAL_SECONDS)
            retries += 1

        # We never saw the process, so abort with details on who was missing.
        raise error.TestFail('Never saw a pid for "%s"' % (self._name))


class UpstartProcess(Process):
    """Represents an Upstart service."""

    def exists(self):
        """Checks if the service is present in Upstart configuration."""
        return upstart.has_service(self._service_name)

    def restart(self):
        """Restarts the process via initctl."""
        utils.system('initctl restart %s' % self._service_name)

class SystemdProcess(Process):
    """Represents an systemd service."""

    def exists(self):
        """Checks if the service is present in systemd configuration."""
        cmd = 'systemctl show -p ActiveState %s.service' % self._service_name
        output = utils.system_output(cmd, ignore_status=True).strip()
        return output == 'ActiveState=active'

    def restart(self):
        """Restarts the process via systemctl."""
        utils.system('systemctl restart %s' % self._service_name)


class Mapping(object):
    """Holds information about a process's address mapping.

    Stores information about one memory mapping for a process.

    Attributes:
        _name: String name of process/memory occupying the location.
        _start: String containing memory address range start.
    """
    def __init__(self, name, start):
        self._start = start
        self._name = name

    def set_start(self, new_value):
        self._start = new_value

    def get_start(self):
        return self._start

    def __repr__(self):
        return "<mapping %s %s>" % (self._name, self._start)


class security_ASLR(test.test):
    """Runs ASLR tests

    See top document comments for more information.

    Attributes:
        version: Current version of the test.
    """
    version = 1

    _TEST_ITERATION_COUNT = 5

    _ASAN_SYMBOL = "__asan_init"

    # 'update_engine' should at least be present on all boards.
    _PROCESS_LIST = [UpstartProcess('chrome', 'ui', parent='session_manager'),
                     UpstartProcess('debugd', 'debugd'),
                     UpstartProcess('update_engine', 'update-engine'),
                     SystemdProcess('update_engine', 'update-engine'),
                     SystemdProcess('systemd-journald', 'systemd-journald'),]


    def get_processes_to_test(self):
        """Gets processes to test for main function.

        Called by run_once to get processes for this program to test.
        Filters binaries that actually exist on the system.
        This has to be a method because it constructs process objects.

        Returns:
            A list of process objects to be tested (see below for
            definition of process class).
        """
        return [p for p in self._PROCESS_LIST if p.exists()]


    def running_on_asan(self):
        """Returns whether we're running on ASan."""
        # -q, --quiet         * Only output 'bad' things
        # -F, --format <arg>  * Use specified format for output
        # -g, --gmatch        * Use regex rather than string compare (with -s)
        # -s, --symbol <arg>  * Find a specified symbol
        scanelf_command = "scanelf -qF'%s#F'"
        scanelf_command += " -gs %s `which debugd`" % self._ASAN_SYMBOL
        symbol = utils.system_output(scanelf_command)
        logging.debug("running_on_asan(): symbol: '%s', _ASAN_SYMBOL: '%s'",
                      symbol, self._ASAN_SYMBOL)
        return symbol != ""


    def test_randomization(self, process):
        """Tests ASLR of a single process.

        This is the main test function for the program. It creates data
        structures out of useful information from sampling /proc/<pid>/maps
        after restarting the process and then compares address starting
        locations of all executable, stack, and heap memory from each iteration.

        @param process: a process object representing the process to be tested.

        Returns:
            A dict containing a Boolean for whether or not the test passed
            and a list of string messages about passing/failing cases.
        """
        test_result = dict([('pass', True), ('results', []), ('cases', dict())])
        name = process.get_name()
        mappings = list()
        pid = -1
        for i in range(self._TEST_ITERATION_COUNT):
            new_pid = process.get_pid()
            if pid == new_pid:
                raise error.TestFail(
                    'Service "%s" retained PID %d after restart.' % (name, pid))
            pid = new_pid
            mappings.append(self.map(pid))
            process.restart()
        logging.debug('Complete mappings dump for process %s:\n%s',
                      name, pprint.pformat(mappings, 4))

        initial_map = mappings[0]
        for i, mapping in enumerate(mappings[1:]):
            logging.debug('Iteration %d', i)
            for key in mapping.iterkeys():
                # Set default case result to fail, pass when an address change
                # occurs.
                if not test_result['cases'].has_key(key):
                    test_result['cases'][key] = dict([('pass', False),
                            ('number', 0),
                            ('total', self._TEST_ITERATION_COUNT)])
                was_same = (initial_map.has_key(key) and
                        initial_map[key].get_start() ==
                        mapping[key].get_start())
                if was_same:
                    logging.debug("Bad: %s address didn't change", key)
                else:
                    logging.debug('Good: %s address changed', key)
                    test_result['cases'][key]['number'] += 1
                    test_result['cases'][key]['pass'] = True
        for case, result in test_result['cases'].iteritems():
            if result['pass']:
                test_result['results'].append( '[PASS] Address for %s '
                        'successfully changed' % case)
            else:
                test_result['results'].append('[FAIL] Address for %s had '
                        'deterministic value: %s' % (case,
                        mappings[0][case].get_start()))
            test_result['pass'] = test_result['pass'] and result['pass']
        return test_result


    def map(self, pid):
        """Creates data structure from table in /proc/<pid>/maps.

        Gets all data from /proc/<pid>/maps, parses each entry, and saves
        entries corresponding to executable, stack, or heap memory into
        a dictionary.

        @param pid: a string containing the pid to be tested.

        Returns:
            A dict mapping names to mapping objects (see above for mapping
            definition).
        """
        memory_map = dict()
        maps_file = open("/proc/%s/maps" % pid)
        for maps_line in maps_file:
            result = self.parse_result(maps_line)
            if result is None:
                continue
            name = result['name']
            start = result['start']
            perms = result['perms']
            is_memory = name == '[heap]' or name == '[stack]'
            is_useful = re.search('x', perms) is not None or is_memory
            if not is_useful:
                continue
            if not name in memory_map:
                memory_map[name] = Mapping(name, start)
            elif memory_map[name].get_start() < start:
                memory_map[name].set_start(start)
        return memory_map


    def parse_result(self, result):
        """Builds dictionary from columns of a line of /proc/<pid>/maps

        Uses regular expressions to determine column separations. Puts
        column data into a dict mapping column names to their string values.

        @param result: one line of /proc/<pid>/maps as a string, for any <pid>.

        Returns:
            None if the regular expression wasn't matched. Otherwise:
            A dict of string column names mapped to their string values.
            For example:

        {'start': '9e981700000', 'end': '9e981800000', 'perms': 'rwxp',
            'something': '00000000', 'major': '00', 'minor': '00', 'inode':
            '00'}
        """
        # Build regex to parse one line of proc maps table.
        memory = r'(?P<start>\w+)-(?P<end>\w+)'
        perms = r'(?P<perms>(r|-)(w|-)(x|-)(s|p))'
        something = r'(?P<something>\w+)'
        devices = r'(?P<major>\w+):(?P<minor>\w+)'
        inode = r'(?P<inode>[0-9]+)'
        name = r'(?P<name>([a-zA-Z0-9/]+|\[heap\]|\[stack\]))'
        regex = r'%s +%s +%s +%s +%s +%s' % (memory, perms, something,
            devices, inode, name)
        found_match = re.match(regex, result)
        if found_match is None:
            return None
        parsed_result = found_match.groupdict()
        return parsed_result


    def run_once(self):
        """Main function.

        Called when test is run. Gets processes to test and calls test on
        them.

        Raises:
            error.TestFail if any processes' memory mapping addresses are the
            same after restarting.
        """

        if self.running_on_asan():
            logging.warning("security_ASLR is not available on ASan.")
            return

        processes = self.get_processes_to_test()
        # If we don't find any of the processes we wanted to test, we fail.
        if len(processes) == 0:
            proc_names = ", ".join([p.get_name() for p in self._PROCESS_LIST])
            raise error.TestFail(
                'Could not find any of "%s" processes to test' % proc_names)

        aslr_enabled = True
        full_results = dict()
        for process in processes:
            test_results = self.test_randomization(process)
            full_results[process.get_name()] = test_results['results']
            if not test_results['pass']:
                aslr_enabled = False

        logging.debug('SUMMARY:')
        for process_name, results in full_results.iteritems():
            logging.debug('Results for %s:', process_name)
            for result in results:
                logging.debug(result)

        if not aslr_enabled:
            raise error.TestFail('One or more processes had deterministic '
                    'memory mappings')