cli/lib/selinux/policy.py

#
# Copyright (C) 2016 The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#


"""Functions for generating SELinux policy files."""


import glob
import os
import shutil
import subprocess
import tempfile

import error
from core import tool


class Error(error.Error):
    """General SELinux-related failure."""


class SELinuxCommandError(Error):
    """Raised when an SELinux-related command fails."""
    description = 'SELinux command error'


POLICYVERS = 29

MLS_SENS = 1
MLS_CATS = 1024

SEPOLICY_BUILD_FILES = [
    'security_classes',
    'initial_sids',
    'access_vectors',
    'global_macros',
    'neverallow_macros',
    'mls_macros',
    'mls',
    'policy_capabilities',
    'te_macros',
    'attributes',
    'ioctl_defines',
    'ioctl_macros',
    '*.te',
    'roles',
    'users',
    'initial_sid_contexts',
    'fs_use',
    'genfs_contexts',
    'port_contexts'
]

FC = 'file_contexts'
SEPOLICY_DIRS_FILE = 'sepolicy_dirs.txt'


def _ExpandSepolicyPaths(base, dirs, files):
    """Generates all the possible combinations of '/base/dir/file'."""

    policy_files = []
    for filename in files:
        for d in dirs:
            policy_files.extend(glob.glob(os.path.join(base, d, filename)))

    return policy_files


def _RunM4(inputs, output, opts=None):
    # TODO(jorgelo): Use HostToolWrapper once it's easier to redirect output.
    subprocess.check_call('m4 %s -s %s > %s' % (' '.join(opts) if opts else '',
                                                ' '.join(inputs),
                                                output),
                          shell=True)


def _RunSepolicyAnalyze(build_out, _input_file, _output_file):
    # Currently unused, keep for completeness. Will be included in the build
    # soon.
    # pylint: disable=unused-variable
    sepolicy_analize = tool.HostToolWrapper('sepolicy-analyze', build_out)


def GetBoardComboSepolicyDirs(platform):
    """Gets the BOARD_SEPOLICY_DIRS variable from the build."""
    cmds = ['cd "{}"'.format(platform.os.root),
            '. build/envsetup.sh',
            'lunch "{}-{}"'.format(platform.device.name, platform.build_type),
            'get_build_var BOARD_SEPOLICY_DIRS']

    # Link all commands with && to exit immediately if one fails.
    cmdline = ' && '.join(cmds)

    try:
        output = subprocess.check_output(cmdline, shell=True,
                                         executable='bash',
                                         stderr=subprocess.STDOUT)
    except Exception as e:
        raise SELinuxCommandError(
            'Failed to get BOARD_SEPOLICY_DIRS: {}'.format(e))

    lines = output.splitlines()
    if len(lines) < 1:
        raise SELinuxCommandError(
            'Failed to get BOARD_SEPOLICY_DIRS: Output was empty')

    return lines[-1].strip().split()


def SaveSepolicyDirsCache(platform_out, sepolicy_dirs):
    """Saves the BOARD_SEPOLICY_DIRS cache file."""
    try:
        dirs_file = os.path.join(platform_out, SEPOLICY_DIRS_FILE)
        with open(dirs_file, 'w') as fw:
            fw.write('\n'.join(sepolicy_dirs))
    except IOError as e:
        raise SELinuxCommandError(
            'Failed to save BOARD_SEPOLICY_DIRS cache: {}'.format(e))


def LoadCachedSepolicyDirs(platform):
    """Loads and parses the BOARD_SEPOLICY_DIRS cache file."""
    cache_path = os.path.join(platform.build_cache, SEPOLICY_DIRS_FILE)

    board_sepolicy_dirs = None
    if os.access(cache_path, os.R_OK):
        # Cache exists.
        with open(cache_path, 'r') as fr:
            board_sepolicy_dirs = [line.strip() for line in fr.readlines()]
            if len(board_sepolicy_dirs) == 0:
                # Cache file was empty.
                board_sepolicy_dirs = None

    if not board_sepolicy_dirs:
        # Cache did not exist, or was empty.
        board_sepolicy_dirs = GetBoardComboSepolicyDirs(platform)
        SaveSepolicyDirsCache(platform.build_cache, board_sepolicy_dirs)

    return board_sepolicy_dirs


def BuildSepolicy(platform, output_path):
    """Builds the main 'sepolicy' SELinux policy file.

    This needs to be built before attempting to build
    the 'file_contexts.bin' file.
    """
    interm_dir = tempfile.mkdtemp()
    runner = tool.HostToolRunner(platform)
    # TODO: platformify
    try:
        os_path = platform.os.root
        main_sepolicy_dir = os.path.join(os_path, 'system', 'sepolicy')

        all_paths = [main_sepolicy_dir]
        all_paths.extend(LoadCachedSepolicyDirs(platform))

        policy_inputs = _ExpandSepolicyPaths(os_path, all_paths,
                                             SEPOLICY_BUILD_FILES)
        policy_conf = os.path.join(interm_dir, 'policy.conf')
        m4_opts = ['-D mls_num_sens=%s' % MLS_SENS,
                   '-D mls_num_cats=%s' % MLS_CATS,
                   '-D target_build_variant=%s' % platform.build_type]
        _RunM4(policy_inputs, policy_conf, m4_opts)

        runner.run('checkpolicy',
                   ('-M -c %d -o %s %s' % (POLICYVERS, output_path,
                                           policy_conf)).split())
    finally:
        shutil.rmtree(interm_dir, ignore_errors=True)


def BuildFileContexts(platform, sepolicy_path, output_path,
                      additional_context_dir=None):
    """Builds the file_contexts bin SELinux policy file."""

    interm_dir = tempfile.mkdtemp()
    runner = tool.HostToolRunner(platform)
    try:
        os_path = platform.os.root

        fc_local = os.path.join(interm_dir, 'file_contexts.local.tmp')
        fc_device = os.path.join(interm_dir, 'file_contexts.device.tmp')
        fc_device_sorted = os.path.join(interm_dir,
                                        'file_contexts.device.sorted.tmp')
        fc_concat = os.path.join(interm_dir, 'file_contexts.concat.tmp')

        main_sepolicy_dir = os.path.join(os_path, 'system', 'sepolicy')
        main_fc = os.path.join(main_sepolicy_dir, FC)

        _RunM4([main_fc], fc_local)
        board_sepolicy_dirs = LoadCachedSepolicyDirs(platform)

        if additional_context_dir:
            board_sepolicy_dirs.append(additional_context_dir)

        _RunM4(_ExpandSepolicyPaths(os_path, board_sepolicy_dirs, [FC]),
               fc_device)
        runner.run('checkfc', ['-e', sepolicy_path, fc_device])
        runner.run('fc_sort', [fc_device, fc_device_sorted])

        _RunM4([fc_local, fc_device_sorted], fc_concat)
        runner.run('checkfc', ['-e', sepolicy_path, fc_concat])
        runner.run('sefcontext_compile', ['-o', output_path, fc_concat])
    finally:
        shutil.rmtree(interm_dir, ignore_errors=True)