1# -*- coding: utf-8 -*-
2# Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
3# See https://llvm.org/LICENSE.txt for license information.
4# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
5""" This module is responsible to capture the compiler invocation of any
6build process. The result of that should be a compilation database.
7
8This implementation is using the LD_PRELOAD or DYLD_INSERT_LIBRARIES
9mechanisms provided by the dynamic linker. The related library is implemented
10in C language and can be found under 'libear' directory.
11
12The 'libear' library is capturing all child process creation and logging the
13relevant information about it into separate files in a specified directory.
14The parameter of this process is the output directory name, where the report
15files shall be placed. This parameter is passed as an environment variable.
16
17The module also implements compiler wrappers to intercept the compiler calls.
18
19The module implements the build command execution and the post-processing of
20the output files, which will condensates into a compilation database. """
21
22import sys
23import os
24import os.path
25import re
26import itertools
27import json
28import glob
29import logging
30from libear import build_libear, TemporaryDirectory
31from libscanbuild import (
32    command_entry_point,
33    compiler_wrapper,
34    wrapper_environment,
35    run_command,
36    run_build,
37)
38from libscanbuild import duplicate_check
39from libscanbuild.compilation import split_command
40from libscanbuild.arguments import parse_args_for_intercept_build
41from libscanbuild.shell import encode, decode
42
43__all__ = ["capture", "intercept_build", "intercept_compiler_wrapper"]
44
45GS = chr(0x1D)
46RS = chr(0x1E)
47US = chr(0x1F)
48
49COMPILER_WRAPPER_CC = "intercept-cc"
50COMPILER_WRAPPER_CXX = "intercept-c++"
51TRACE_FILE_EXTENSION = ".cmd"  # same as in ear.c
52WRAPPER_ONLY_PLATFORMS = frozenset({"win32", "cygwin"})
53
54
55@command_entry_point
56def intercept_build():
57    """Entry point for 'intercept-build' command."""
58
59    args = parse_args_for_intercept_build()
60    return capture(args)
61
62
63def capture(args):
64    """The entry point of build command interception."""
65
66    def post_processing(commands):
67        """To make a compilation database, it needs to filter out commands
68        which are not compiler calls. Needs to find the source file name
69        from the arguments. And do shell escaping on the command.
70
71        To support incremental builds, it is desired to read elements from
72        an existing compilation database from a previous run. These elements
73        shall be merged with the new elements."""
74
75        # create entries from the current run
76        current = itertools.chain.from_iterable(
77            # creates a sequence of entry generators from an exec,
78            format_entry(command)
79            for command in commands
80        )
81        # read entries from previous run
82        if "append" in args and args.append and os.path.isfile(args.cdb):
83            with open(args.cdb) as handle:
84                previous = iter(json.load(handle))
85        else:
86            previous = iter([])
87        # filter out duplicate entries from both
88        duplicate = duplicate_check(entry_hash)
89        return (
90            entry
91            for entry in itertools.chain(previous, current)
92            if os.path.exists(entry["file"]) and not duplicate(entry)
93        )
94
95    with TemporaryDirectory(prefix="intercept-") as tmp_dir:
96        # run the build command
97        environment = setup_environment(args, tmp_dir)
98        exit_code = run_build(args.build, env=environment)
99        # read the intercepted exec calls
100        exec_traces = itertools.chain.from_iterable(
101            parse_exec_trace(os.path.join(tmp_dir, filename))
102            for filename in sorted(glob.iglob(os.path.join(tmp_dir, "*.cmd")))
103        )
104        # do post processing
105        entries = post_processing(exec_traces)
106        # dump the compilation database
107        with open(args.cdb, "w+") as handle:
108            json.dump(list(entries), handle, sort_keys=True, indent=4)
109        return exit_code
110
111
112def setup_environment(args, destination):
113    """Sets up the environment for the build command.
114
115    It sets the required environment variables and execute the given command.
116    The exec calls will be logged by the 'libear' preloaded library or by the
117    'wrapper' programs."""
118
119    c_compiler = args.cc if "cc" in args else "cc"
120    cxx_compiler = args.cxx if "cxx" in args else "c++"
121
122    libear_path = (
123        None
124        if args.override_compiler or is_preload_disabled(sys.platform)
125        else build_libear(c_compiler, destination)
126    )
127
128    environment = dict(os.environ)
129    environment.update({"INTERCEPT_BUILD_TARGET_DIR": destination})
130
131    if not libear_path:
132        logging.debug("intercept gonna use compiler wrappers")
133        environment.update(wrapper_environment(args))
134        environment.update({"CC": COMPILER_WRAPPER_CC, "CXX": COMPILER_WRAPPER_CXX})
135    elif sys.platform == "darwin":
136        logging.debug("intercept gonna preload libear on OSX")
137        environment.update(
138            {"DYLD_INSERT_LIBRARIES": libear_path, "DYLD_FORCE_FLAT_NAMESPACE": "1"}
139        )
140    else:
141        logging.debug("intercept gonna preload libear on UNIX")
142        environment.update({"LD_PRELOAD": libear_path})
143
144    return environment
145
146
147@command_entry_point
148def intercept_compiler_wrapper():
149    """Entry point for `intercept-cc` and `intercept-c++`."""
150
151    return compiler_wrapper(intercept_compiler_wrapper_impl)
152
153
154def intercept_compiler_wrapper_impl(_, execution):
155    """Implement intercept compiler wrapper functionality.
156
157    It does generate execution report into target directory.
158    The target directory name is from environment variables."""
159
160    message_prefix = "execution report might be incomplete: %s"
161
162    target_dir = os.getenv("INTERCEPT_BUILD_TARGET_DIR")
163    if not target_dir:
164        logging.warning(message_prefix, "missing target directory")
165        return
166    # write current execution info to the pid file
167    try:
168        target_file_name = str(os.getpid()) + TRACE_FILE_EXTENSION
169        target_file = os.path.join(target_dir, target_file_name)
170        logging.debug("writing execution report to: %s", target_file)
171        write_exec_trace(target_file, execution)
172    except IOError:
173        logging.warning(message_prefix, "io problem")
174
175
176def write_exec_trace(filename, entry):
177    """Write execution report file.
178
179    This method shall be sync with the execution report writer in interception
180    library. The entry in the file is a JSON objects.
181
182    :param filename:    path to the output execution trace file,
183    :param entry:       the Execution object to append to that file."""
184
185    with open(filename, "ab") as handler:
186        pid = str(entry.pid)
187        command = US.join(entry.cmd) + US
188        content = RS.join([pid, pid, "wrapper", entry.cwd, command]) + GS
189        handler.write(content.encode("utf-8"))
190
191
192def parse_exec_trace(filename):
193    """Parse the file generated by the 'libear' preloaded library.
194
195    Given filename points to a file which contains the basic report
196    generated by the interception library or wrapper command. A single
197    report file _might_ contain multiple process creation info."""
198
199    logging.debug("parse exec trace file: %s", filename)
200    with open(filename, "r") as handler:
201        content = handler.read()
202        for group in filter(bool, content.split(GS)):
203            records = group.split(RS)
204            yield {
205                "pid": records[0],
206                "ppid": records[1],
207                "function": records[2],
208                "directory": records[3],
209                "command": records[4].split(US)[:-1],
210            }
211
212
213def format_entry(exec_trace):
214    """Generate the desired fields for compilation database entries."""
215
216    def abspath(cwd, name):
217        """Create normalized absolute path from input filename."""
218        fullname = name if os.path.isabs(name) else os.path.join(cwd, name)
219        return os.path.normpath(fullname)
220
221    logging.debug("format this command: %s", exec_trace["command"])
222    compilation = split_command(exec_trace["command"])
223    if compilation:
224        for source in compilation.files:
225            compiler = "c++" if compilation.compiler == "c++" else "cc"
226            command = [compiler, "-c"] + compilation.flags + [source]
227            logging.debug("formated as: %s", command)
228            yield {
229                "directory": exec_trace["directory"],
230                "command": encode(command),
231                "file": abspath(exec_trace["directory"], source),
232            }
233
234
235def is_preload_disabled(platform):
236    """Library-based interposition will fail silently if SIP is enabled,
237    so this should be detected. You can detect whether SIP is enabled on
238    Darwin by checking whether (1) there is a binary called 'csrutil' in
239    the path and, if so, (2) whether the output of executing 'csrutil status'
240    contains 'System Integrity Protection status: enabled'.
241
242    :param platform: name of the platform (returned by sys.platform),
243    :return: True if library preload will fail by the dynamic linker."""
244
245    if platform in WRAPPER_ONLY_PLATFORMS:
246        return True
247    elif platform == "darwin":
248        command = ["csrutil", "status"]
249        pattern = re.compile(r"System Integrity Protection status:\s+enabled")
250        try:
251            return any(pattern.match(line) for line in run_command(command))
252        except:
253            return False
254    else:
255        return False
256
257
258def entry_hash(entry):
259    """Implement unique hash method for compilation database entries."""
260
261    # For faster lookup in set filename is reverted
262    filename = entry["file"][::-1]
263    # For faster lookup in set directory is reverted
264    directory = entry["directory"][::-1]
265    # On OS X the 'cc' and 'c++' compilers are wrappers for
266    # 'clang' therefore both call would be logged. To avoid
267    # this the hash does not contain the first word of the
268    # command.
269    command = " ".join(decode(entry["command"])[1:])
270
271    return "<>".join([filename, directory, command])
272