#!/usr/bin/env fuchsia-vendored-python
"""Validates file system accesses of a subprocess command.

This uses a traced exection wrapper (fsatrace) to invoke a command,
captures a trace of file system {read,write} operations, and validates
those access against constraints such as declared inputs and outputs.
"""
# Copyright 2020 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import argparse
import dataclasses
import enum
import os
import re
import shlex
import subprocess
import sys
from typing import (
    AbstractSet,
    Any,
    Callable,
    Collection,
    FrozenSet,
    Iterable,
    Optional,
    Sequence,
    TextIO,
    Tuple,
)

trailing_white_spaces = re.compile("\\\s+\r?\n")


def _partition(
    iterable: Iterable[Any], predicate: Callable[[Any], bool]
) -> Tuple[Sequence[Any], Sequence[Any]]:
    """Splits sequence into two sequences based on predicate function."""
    trues = []
    falses = []
    for item in iterable:
        if predicate(item):
            trues.append(item)
        else:
            falses.append(item)
    return trues, falses


class FileAccessType(enum.Enum):
    READ = enum.auto()
    WRITE = enum.auto()
    DELETE = enum.auto()


@dataclasses.dataclass
class MatchConditions(object):
    prefixes: FrozenSet[str] = dataclasses.field(default_factory=set)
    suffixes: FrozenSet[str] = dataclasses.field(default_factory=set)
    components: FrozenSet[str] = dataclasses.field(default_factory=set)

    def matches(self, path: str) -> bool:
        """Returns true if path matches any of the conditions."""
        if any(path.startswith(prefix) for prefix in self.prefixes):
            return True
        if any(path.endswith(suffix) for suffix in self.suffixes):
            return True
        if set(path.split(os.path.sep)).intersection(self.components):
            return True
        return False


def _find_first_index(
    iterable: Iterable[Any], pred: Callable[[Any], bool]
) -> int:
    """Returns the index of the first element that satisfies the predicate, or -1."""
    for n, item in enumerate(iterable):
        if pred(item):
            return n
    return -1


@dataclasses.dataclass
class ToolCommand(object):
    """This models a command for invoking a tool in a shell.

    A command consists of the following components in order:
      * an optional list of environment variable overrides like 'HOME=/path'
      * one tool (script or binary in PATH)
      * an optional list of flags and arguments passed to the tool
    """

    tokens: Sequence[str] = dataclasses.field(default_factory=list)

    @property
    def _tool_index(self) -> int:
        # The first token that isn't 'X=Y' is the tool/script.
        return _find_first_index(self.tokens, lambda x: "=" not in x)

    @property
    def env_tokens(self) -> Sequence[str]:
        """Returns the environment overrides (X=Y) of a shell command."""
        return self.tokens[: self._tool_index]

    @property
    def tool(self) -> str:
        """Returns the tool/script of the command."""
        return self.tokens[self._tool_index]

    @property
    def args(self) -> Sequence[str]:
        """Returns all options and arguments after the tool of the command."""
        return self.tokens[self._tool_index + 1 :]

    @property
    def _end_opts_index(self) -> int:
        # Find the position of '--', which is conventionally used to stop option
        # processing.
        return _find_first_index(self.tokens, lambda x: x == "--")

    def unwrap(self) -> "ToolCommand":
        # Assuming that '--' separates a wrapper from a command, this unwraps
        # a command one level.  (For example, this script is such a wrapper,
        # with the original command following '--' in trailing position.)
        end_opts_index = self._end_opts_index
        if end_opts_index == -1:
            # Deduce that the command is not wrapped.
            return self
        return ToolCommand(tokens=self.tokens[end_opts_index + 1 :])


@dataclasses.dataclass
class FSAccess(object):
    """Represents a single file system access."""

    # One of: "read", "write" (covers touch), "delete" (covers move-from)
    op: FileAccessType
    # The path accessed
    path: str

    # TODO(fangism): for diagnostic purposes, we may want a copy of the fsatrace
    # line from which this access came.

    def __repr__(self) -> str:
        return f"({self.op} {self.path})"

    def should_check(
        self, ignore_conditions: MatchConditions, required_path_prefix: str = ""
    ) -> bool:
        """Predicate function use to filter out FSAccesses.

        Args:
          required_path_prefix: Accesses outside of this path prefix are not checked.
            An empty string means: check this access.
          ignore_conditions: prefixes, suffixes, components to ignore.

        Returns:
          true if this access should be checked.
        """
        if not self.path.startswith(required_path_prefix):
            return False

        return not ignore_conditions.matches(self.path)

    def allowed(
        self, allowed_reads: FrozenSet[str], allowed_writes: FrozenSet[str]
    ) -> bool:
        """Validates a file system access against a set of allowed accesses.

        Args:
          allowed_reads: set of allowed read paths.
          allowed_writes: set of allowed write paths.

        Returns:
          True if this access is allowed.
        """
        if self.op == FileAccessType.READ:
            return self.path in allowed_reads
        elif self.op == FileAccessType.WRITE:
            return self.path in allowed_writes
        elif self.op == FileAccessType.DELETE:
            # TODO(fangism): separate out forbidded_deletes
            return self.path in allowed_writes
        raise ValueError(f"Unknown operation: {self.op}")


# Factory functions for making FSAccess objects.
def Read(path: str) -> FSAccess:
    return FSAccess(FileAccessType.READ, path)


def Write(path: str) -> FSAccess:
    return FSAccess(FileAccessType.WRITE, path)


def Delete(path: str) -> FSAccess:
    return FSAccess(FileAccessType.DELETE, path)


def _parse_fsatrace_line(fsatrace_line: str) -> Iterable[FSAccess]:
    """Parses an output line from fsatrace into a stream of FSAccesses.

    See: https://github.com/jacereda/fsatrace#output-format
    Moves are split into two operations: delete source, write destination

    Args:
      fsatrace_line: one line of trace from fsatrace

    Yields:
      0 to 2 FSAccess objects.
    """
    # ignore any lines that do not parse
    op, sep, path = fsatrace_line.partition("|")
    if sep != "|":
        return

    # op: operation code in [rwdtm]
    if op == "r":
        yield Read(path)
    elif op in {"w", "t"}:
        yield Write(path)
    elif op == "d":
        yield Delete(path)
    elif op == "m":
        # path: "destination|source"
        # The source is deleted, and the destination is written.
        dest, sep, source = path.partition("|")
        if sep != "|":
            raise ValueError("Malformed move line: " + fsatrace_line)
        yield Delete(source)
        yield Write(dest)


def parse_fsatrace_output(fsatrace_lines: Iterable[str]) -> Iterable[FSAccess]:
    """Returns a stream of FSAccess objects."""
    ret = []
    for line in fsatrace_lines:
        try:
            ret.extend(_parse_fsatrace_line(line))
        except Exception as e:
            print("\n".join(fsatrace_lines))
            raise e
    return ret


def _abspaths(container: Iterable[str]) -> AbstractSet[str]:
    return {os.path.abspath(f) for f in container}


@dataclasses.dataclass
class AccessConstraints(object):
    """Set of file system accesses constraints."""

    allowed_reads: FrozenSet[str] = dataclasses.field(default_factory=set)
    allowed_writes: FrozenSet[str] = dataclasses.field(default_factory=set)
    required_writes: FrozenSet[str] = dataclasses.field(default_factory=set)
    # TODO(fangism): forbidden_deletes should probably include declared inputs

    @property
    def inputs(self) -> FrozenSet[str]:
        # allowed_reads includes allowed_writes (and required_writes), so consider
        # "inputs" as their set-difference.
        return self.allowed_reads - self.allowed_writes - self.required_writes


@dataclasses.dataclass
class DepEdges(object):
    ins: FrozenSet[str] = dataclasses.field(default_factory=set)
    outs: FrozenSet[str] = dataclasses.field(default_factory=set)

    def abspaths(self) -> "DepEdges":
        return DepEdges(ins=_abspaths(self.ins), outs=_abspaths(self.outs))


def parse_dep_edges(depfile_line: str) -> DepEdges:
    """Parse a single line of a depfile.

    This assumes that all depfile entries are formatted onto a single line.
    TODO(fangism): support more generalized forms of input, e.g. multi-line.
      See https://github.com/ninja-build/ninja/blob/master/src/depfile_parser_test.cc

    Args:
      depfile_line: has the form "OUTPUT1 [OUTPUT2 ...]: INPUT [INPUT ...]"

    Returns:
      A DepEdges object represending a dependency between inputs and outputs.

    Raises:
      ValueError if unable to parse dependency entry.
    """
    outs, sep, ins = depfile_line.strip().partition(":")
    if not sep:
        raise ValueError("Failed to parse depfile entry:\n" + depfile_line)
    return DepEdges(ins=set(shlex.split(ins)), outs=set(shlex.split(outs)))


@dataclasses.dataclass
class DepFile(object):
    """DepFile represents a collection of dependency edges."""

    deps: Collection[DepEdges] = dataclasses.field(default_factory=list)

    @property
    def all_ins(self) -> AbstractSet[str]:
        """Returns a set of all dependency inputs."""
        return {f for dep in self.deps for f in dep.ins}

    @property
    def all_outs(self) -> AbstractSet[str]:
        """Returns a set of all dependency outputs."""
        return {f for dep in self.deps for f in dep.outs}


def parse_depfile(depfile_lines: Iterable[str]) -> DepFile:
    """Parses a depfile into a set of inputs and outputs.

    See https://github.com/ninja-build/ninja/blob/master/src/depfile_parser_test.cc
    for examples of format using Ninja syntax.

    Args:
      depfile_lines: lines from a depfile

    Returns:
      DepFile object, collection of dependencies.
    """

    # Go through all lines and join continuations. Doing this manually to avoid
    # copies as much as possible.
    lines = []
    current_line = ""
    for line in depfile_lines:
        # Ignore empty line
        if not line.strip():
            if current_line:
                raise ValueError(
                    "Line continuation followed by empty line in depfile line "
                    + line
                )
            continue
        # Ignore comments
        if line.strip().startswith("#"):
            if current_line:
                raise ValueError(
                    "Line continuation followed by comment in depfile line "
                    + line
                )
            continue
        # We currently don't allow consecutive backslashes in filenames to
        # simplify depfile parsing. Support can be added if use cases come up.
        #
        # Ninja's implementation:
        # https://github.com/ninja-build/ninja/blob/5993141c0977f563de5e064fbbe617f9dc34bb8d/src/depfile_parser.cc#L39
        if r"\\" in line:
            raise ValueError(
                f'Consecutive backslashes found in depfile line "{line}", this is not supported by action tracer'
            )
        # We currently don't have any use cases with trailing whitespaces in
        # file names, so treat them as errors when they show up at the end of a
        # line, because users usually want a line continuation. We can
        # reconsider this check when use cases come up.
        if trailing_white_spaces.match(line):
            raise ValueError(
                f'Backslash followed by trailing whitespaces at end of line "{line}", remove whitespaces for proper line continuation'
            )

        if line.endswith(("\\\n", "\\\r\n")):
            current_line += line.rstrip("\\\r\n")
            continue
        current_line += line
        lines.append(current_line)
        current_line = ""

    if current_line:
        raise ValueError("Line continuation found at end of file")

    return DepFile(deps=[parse_dep_edges(line) for line in lines])


def abspaths_from_depfile(
    depfile: DepFile, allowed_abspaths: FrozenSet[str]
) -> Collection[str]:
    return [
        f
        for f in (depfile.all_ins | depfile.all_outs)
        if f not in allowed_abspaths and os.path.isabs(f)
    ]


@dataclasses.dataclass
class Action(object):
    """Represents a set of parameters of a single build action."""

    inputs: Sequence[str] = dataclasses.field(default_factory=list)
    outputs: Collection[str] = dataclasses.field(default_factory=list)
    depfile: Optional[str] = None
    hermetic_inputs: Optional[Collection[str]] = None
    parsed_depfile: Optional[DepFile] = None

    def access_constraints(
        self, writeable_depfile_inputs: bool = False
    ) -> AccessConstraints:
        """Build AccessConstraints from action attributes."""
        # Action is required to write outputs and depfile, if provided.
        required_writes = {path for path in self.outputs}

        # Paths that the action is allowed to write.
        # Actions may touch files other than their listed outputs.
        allowed_writes = required_writes.copy()

        allowed_reads = set(self.inputs)

        if self.hermetic_inputs:
            allowed_reads.update(self.hermetic_inputs)
        elif self.depfile:
            # Writing the depfile is not required (yet), but allowed.
            allowed_writes.add(self.depfile)
            if os.path.exists(self.depfile):
                with open(self.depfile, "r") as f:
                    self.parsed_depfile = parse_depfile(f)

                if writeable_depfile_inputs:
                    allowed_writes.update(self.parsed_depfile.all_ins)
                else:
                    allowed_reads.update(self.parsed_depfile.all_ins)
                allowed_writes.update(self.parsed_depfile.all_outs)

        # Everything writeable is readable.
        allowed_reads.update(allowed_writes)

        return AccessConstraints(
            # Follow links in all inputs because fsatrace will log access to link
            # destination instead of the link.
            allowed_reads=_abspaths(
                os.path.realpath(path) for path in allowed_reads
            ),
            # TODO(https://fxbug.dev/42148140): Should we follow links of outputs as well?
            # What's our stance on writing to soft links?
            allowed_writes=_abspaths(allowed_writes),
            required_writes=_abspaths(required_writes),
        )


def _sorted_join(elements: Iterable[str], joiner: str) -> str:
    return joiner.join(sorted(elements))


@dataclasses.dataclass
class FSAccessSet(object):
    reads: FrozenSet[str] = dataclasses.field(default_factory=set)
    writes: FrozenSet[str] = dataclasses.field(default_factory=set)
    deletes: FrozenSet[str] = dataclasses.field(default_factory=set)

    @property
    def all_accesses(self) -> FrozenSet[str]:
        return self.reads | self.writes | self.deletes

    def __str__(self) -> str:
        if not self.all_accesses:
            return "[empty accesses]"
        text = ""
        if self.reads:
            text += "\nReads:\n  " + _sorted_join(self.reads, "\n  ")
        if self.writes:
            text += "\nWrites:\n  " + _sorted_join(self.writes, "\n  ")
        if self.deletes:
            text += "\nDeletes:\n  " + _sorted_join(self.deletes, "\n  ")
        # trim first newline if there is one
        return text.lstrip("\n")


def finalize_filesystem_accesses(accesses: Iterable[FSAccess]) -> FSAccessSet:
    """Converts a sequence of filesystem accesses into sets of accesses.

    This tracks deleted files, assuming that a file that is written and
    then deleted is only a temporary, and is not counted as a final write.
    Reads of temporary files are allowed and not recorded.
    Deletes of files not written by this sequence of accesses are recorded.
    Converting from a stream to set(s) loses access sequence information.

    Args:
      accesses: stream of file-system accesses.

    Returns:
      Sets of read, written, and deleted files that should be verified
      elsewhere (excluding inferred temporaries).
    """
    reads = set()
    writes = set()
    deletes = set()
    for access in accesses:
        if access.op == FileAccessType.READ:
            # Reading a file that we've written is not interesting.
            # Omit those, but add all others.
            if access.path not in writes:
                reads.add(access.path)
        elif access.op == FileAccessType.WRITE:
            writes.add(access.path)
            deletes.discard(access.path)
        elif access.op == FileAccessType.DELETE:
            if access.path in writes:
                # Infer that this is a temporary file.
                writes.discard(access.path)
                # Allow and ignore reads to written files.
                reads.discard(access.path)
                # Do not record this as a deleted file.
            else:
                # All other deletes require scrutiny.
                deletes.add(access.path)

    # writes contains the set of files that were not deleted
    return FSAccessSet(reads=reads, writes=writes, deletes=deletes)


def check_access_permissions(
    accesses: FSAccessSet, constraints: AccessConstraints
) -> FSAccessSet:
    """Checks a sequence of accesses against permission constraints.

    Args:
      accesses: sets of file-system read/write accesses.
      constraints: permitted accesses.
        .allowed_reads: set of files that are allowed to be read.
        .allowed_writes: set of files that are allowed to be written.

    Returns:
      Subset of not-permitted file accesses.
    """
    # Suppress diagnostics on reading files that are written,
    # regardless of whether or not those writes were allowed.
    # For example, temporarily written files (not declared as outputs)
    # should be allowed to be read without issue.
    allowed_reads = constraints.allowed_reads | accesses.writes
    unexpected_reads = accesses.reads - allowed_reads
    unexpected_writes = accesses.writes - constraints.allowed_writes
    return FSAccessSet(reads=unexpected_reads, writes=unexpected_writes)


def check_missing_writes(
    accesses: Iterable[FSAccess], required_writes: FrozenSet[str]
) -> AbstractSet[str]:
    """Tracks sequence of access to verify that required files are written.

    Args:
      accesses: file-system accesses.
      required_writes: paths that are expected to be written.

    Returns:
      Subset of required_writes that were not fulfilled.
    """
    missing_writes = required_writes.copy()
    for access in accesses:
        if access.op == FileAccessType.WRITE and access.path in missing_writes:
            missing_writes.remove(access.path)
        elif (
            access.op == FileAccessType.DELETE
            and access.path in required_writes
        ):
            missing_writes.add(access.path)

    return missing_writes


def actually_read_files(accesses: Iterable[FSAccess]) -> AbstractSet[str]:
    """Returns subset of files that were actually used/read."""
    return {
        access.path for access in accesses if access.op == FileAccessType.READ
    }


def _verbose_path(path: str) -> str:
    """When any symlinks are followed, show this."""
    realpath = os.path.realpath(path)
    if path != realpath:
        return path + " -> " + realpath
    return path


@dataclasses.dataclass
class StalenessDiagnostics(object):
    """Just a structure to capture results of diagnosing outputs."""

    required_writes: FrozenSet[str] = dataclasses.field(default_factory=set)
    nonexistent_outputs: FrozenSet[str] = dataclasses.field(default_factory=set)
    # If there are stale_outputs, then it must have been compared against a
    # newest_input.
    newest_input: Optional[str] = None
    stale_outputs: FrozenSet[str] = dataclasses.field(default_factory=set)

    @property
    def has_findings(self) -> FrozenSet[str]:
        return self.nonexistent_outputs or self.stale_outputs

    def print_findings(self, stream: TextIO) -> None:
        """Prints human-readable diagnostics.

        Args:
          stream: a file stream, like sys.stderr.
        """
        required_writes_formatted = "\n".join(
            _verbose_path(f) for f in self.required_writes
        )
        print(
            f"""
Required writes:
{required_writes_formatted}
""",
            file=stream,
        )
        if self.nonexistent_outputs:
            nonexistent_outputs_formatted = "\n".join(
                _verbose_path(f) for f in self.nonexistent_outputs
            )
            print(
                f"""
Missing outputs:
{nonexistent_outputs_formatted}
""",
                file=stream,
            )

        if self.stale_outputs:
            stale_outputs_formatted = "\n".join(
                _verbose_path(f) for f in self.stale_outputs
            )
            print(
                f"""
Stale outputs: (older than newest input: {self.newest_input})
{stale_outputs_formatted}
""",
                file=stream,
            )


def realpath_ctime(path: str) -> int:
    """Follow symlinks before getting ctime.

    This reflects Ninja's behavior of using `stat()` instead of `lstat()`
    on symlinks.

    Args:
      path: file or symlink

    Returns:
      ctime of the realpath of path.
    """
    return os.path.getctime(os.path.realpath(path))


def diagnose_stale_outputs(
    accesses: Iterable[FSAccess], access_constraints: AccessConstraints
) -> StalenessDiagnostics:
    """Analyzes access stream for missing writes.

    Also compares timestamps of inputs relative to outputs
    to determine staleness.

    Args:
      accesses: trace of file system accesses.
      access_constraints: access that may/must[not] occur.

    Returns:
      Structure of findings, including missing/stale outputs.
    """
    # Verify that outputs are written as promised.
    missing_writes = check_missing_writes(
        accesses, access_constraints.required_writes
    )

    # Distinguish stale from nonexistent output files.
    untouched_outputs, nonexistent_outputs = _partition(
        missing_writes, os.path.exists
    )

    # Check that timestamps relative to inputs (allowed_reads) are newer,
    # in which case, not-writing outputs is acceptable.
    # Determines file use based on the `accesses` trace,
    # not the stat() filesystem function.
    read_files = actually_read_files(accesses)
    # Ignore allowed-but-unused inputs.
    # Outputs are readable, but should not be considered as inputs.
    used_inputs = access_constraints.inputs.intersection(read_files)

    # Compare timestamps vs. newest input to find stale outputs.
    stale_outputs = set()
    newest_input = None
    if used_inputs and untouched_outputs:
        # All links in inputs are followed to their destinations already in
        # previous steps, so realpath_ctime is unnecessary on them.
        newest_input = max(used_inputs, key=os.path.getctime)
        # Filter out untouched outputs that are still newer than used inputs.
        input_timestamp = os.path.getctime(newest_input)
        stale_outputs = {
            out
            for out in untouched_outputs
            if realpath_ctime(out) < input_timestamp
        }
    return StalenessDiagnostics(
        required_writes=access_constraints.required_writes,
        nonexistent_outputs=set(nonexistent_outputs),
        newest_input=newest_input,
        stale_outputs=stale_outputs,
    )


def main_arg_parser() -> argparse.ArgumentParser:
    """Construct the argument parser, called by main()."""
    parser = argparse.ArgumentParser(
        description="Traces a GN action and enforces strict inputs/outputs",
        argument_default=[],
    )
    parser.add_argument(
        "--fsatrace-path",
        default="fsatrace",
        help="Path to fsatrace binary.  If omitted, it will search for one in PATH.",
    )
    parser.add_argument(
        "--label", required=True, help="The wrapped target's label"
    )

    parser.add_argument(
        "--trace-output", required=True, help="Where to store the trace"
    )
    parser.add_argument(
        "--keep-raw-trace",
        action=argparse.BooleanOptionalAction,
        default=False,
        help="Whether to keep trace output after the checks are successful",
    )
    parser.add_argument(
        "--target-type",
        choices=["action", "action_foreach"],
        default="action",
        help="Type of target being wrapped",
    )
    parser.add_argument("--inputs", nargs="*", help="action#inputs")
    parser.add_argument("--outputs", nargs="*", help="action#outputs")
    parser.add_argument("--depfile", help="action#depfile")
    parser.add_argument(
        "--hermetic-inputs-file",
        help="Path to file listing extra inputs for this command",
    )
    parser.add_argument(
        "--failed-check-status",
        type=int,
        default=1,
        help="On failing tracing checks, exit with this code.  Use 0 to report findings without failing.",
    )
    parser.add_argument(
        "--ignore-prefix",
        nargs="*",
        default=[],
        help="Extra file-path prefix that should be ignored.",
    )
    parser.add_argument(
        "--check-access-permissions",
        action=argparse.BooleanOptionalAction,
        default=True,
        help="Check permissions on file reads and writes",
    )

    # This affects the set of files that are allowed to be written.
    # TODO(fangism): remove this flag entirely, disallowing writes to inputs
    parser.add_argument(
        "--writeable-depfile-inputs",
        action=argparse.BooleanOptionalAction,
        default=False,
        help="Allow writes to inputs found in depfiles.  Only effective with --check-access-permissions.",
    )

    # TODO(fangism): This check is blocked on *.py being in the ignored set.
    parser.add_argument(
        "--check-inputs-not-in-ignored-set",
        action="store_true",
        default=False,  # Goal: always True (remove this flag)
        help="Check that inputs do not belong to the set of ignored files",
    )

    # Positional args are the command (tool+args) to run and trace.
    parser.add_argument("command", nargs="*", help="action#command")
    return parser


def _tool_is_python(tool: str) -> bool:
    base = os.path.basename(tool)
    return base == "python" or base.startswith("python3")


def get_intended_tool(command: ToolCommand) -> str:
    if python_script := get_python_script(command):
        return python_script
    if os.path.basename(command.tool) == "gn_run_binary.sh":
        return command.args[1]
    return command.tool


def get_python_script(command: ToolCommand) -> Optional[str]:
    """If the script being invoked is python, return the relevant .py file"""
    # Cover both cases when the tool:
    #
    # 1. is executed directly, for example: ./build.py
    if command.tool.endswith((".py", ".pyz")):
        return command.tool
    # 2. is explicitly executed by an interpreter
    #    for example: path/to/prebuilt/python3 build.py
    elif _tool_is_python(command.tool):
        script_index = _find_first_index(
            command.args, lambda x: x.endswith((".py", ".pyz"))
        )
        assert (
            script_index != -1
        ), f"Expected to find Python script after interpreter: {command.args}"
        return command.args[script_index]
    return None


def is_known_wrapper(command: ToolCommand) -> bool:
    """Is this a command-wrapping script?

    Returns:
      True if the command is one of the known wrapper scripts that encapsulates
        another command in tail position after '--'.
    """
    if python_script := get_python_script(command):
        return os.path.basename(python_script) in {
            "action_tracer.py",
            "output_cacher.py",
        }
    return False


def all_parent_dirs(abspath: str) -> AbstractSet[str]:
    """Identifies all parent dirs of a given path.

    Args:
      abspath: absolute path to a file or dir, which need not exist.

    Returns:
      Set of all ancestor directories of `abspath`.
    """
    dirs = set()
    temp_path = os.path.dirname(abspath)
    while temp_path and temp_path != os.path.sep:
        dirs.add(temp_path)
        temp_path = os.path.dirname(temp_path)
    return dirs


def detect_all_dirs(paths: Iterable[str]) -> AbstractSet[str]:
    """Evaluates the set of directories seen from a collection of paths.

    Args:
      paths: sequence of file or directory paths.

    Returns:
      Set union of all ancestor directories of paths.
    """
    dirs = set()
    for path in paths:
        dirs.update(all_parent_dirs(path))
    return dirs


def main() -> int:
    parser = main_arg_parser()
    args = parser.parse_args()

    command = ToolCommand(tokens=args.command)

    # Unwrap certain command wrapper scripts.
    while is_known_wrapper(command):
        command = command.unwrap()

    # Ensure trace_output directory exists
    trace_output_dir = os.path.dirname(args.trace_output)
    os.makedirs(trace_output_dir, exist_ok=True)

    retval = subprocess.call(
        [
            args.fsatrace_path,
            "rwmdt",
            args.trace_output,
            "--",
        ]
        + command.tokens
    )

    hermetic_inputs = []
    implicit_inputs = []
    depfile = args.depfile
    if args.hermetic_inputs_file:
        assert args.depfile, "--hermetic-inputs-file requires --depfile!"
        with open(args.hermetic_inputs_file) as f:
            for l in f:
                line = l.strip()
                hermetic_inputs.append(os.path.abspath(line))
                implicit_inputs.append(line)

        # Generate the depfile here. Take care of creating the output directory
        # if needed.
        depfile_dir = os.path.dirname(args.depfile)
        if not os.path.exists(depfile_dir):
            os.makedirs(depfile_dir)
        with open(args.depfile, "w") as f:
            f.write(
                "%s: %s\n"
                % (
                    " ".join(sorted(args.outputs)),
                    " ".join(sorted(implicit_inputs)),
                )
            )

    # Compute constraints from action properties (from args).
    action = Action(
        inputs=args.inputs,
        outputs=args.outputs,
        depfile=depfile,
        hermetic_inputs=hermetic_inputs,
    )
    access_constraints = action.access_constraints(
        writeable_depfile_inputs=args.writeable_depfile_inputs
    )

    # Limit most access checks to files under src_root.
    src_root = os.path.dirname(os.path.dirname(os.getcwd()))

    # Paths that are ignored
    ignored_prefixes = {
        ### Build
        # compile_commands.json is generated by out GN but GN itself doesn't see
        # this as a dependency so it can't be listed as an input.
        # https://bugs.chromium.org/p/gn/issues/detail?id=313
        os.path.join(os.getcwd(), "compile_commands.json"),
        ### C/C++ toolchain-related
        # Specify toolchain/sysroot related exceptions using explicit
        # --ignore-prefix (hermetic_action_ignored_prefixes in GN),
        # instead of hard-coding their paths here.
        ### Python
        # Python scripts access Python prebuilts for the interpreter,
        # standard library, and other things that are not strict inputs
        # to Python scripts.  This also covers dirs named python3.x.
        os.path.join(src_root, "prebuilt", "third_party", "python3"),
        ### Dart
        # Dart provides prebuilt libs and their snapshots for its standard libraries
        # that are not strict inputs to Dart programs.
        os.path.join(src_root, "prebuilt", "third_party", "dart/"),
        # The Dart format and analyzer want to write to $HOME/.dart/...
        # but there is no HOME defined when running `fx build`, so they end
        # up writing to the output directory instead. Ignore these since this
        # is harmless. This is favored to setting a fake HOME value in
        # the `hermetic-env` script (used by `fx build`), because it allows
        # catching other tools trying to write to $HOME in the future.
        # This only affects local builds.
        os.path.join(os.getcwd(), ".dart/"),
        ### Ninja
        # There is no way to specify this file as an input to a GN action
        # since this file is not known to GN.
        os.path.join(os.getcwd(), "build.ninja.d"),
    }
    # Ignored prefixes are to be given relative to the root_build_dir (since
    # that is the only rebase_path() option from GN that can be used.
    for prefix in args.ignore_prefix:
        ignored_prefixes.add(
            os.path.normpath(os.path.join(os.getcwd(), prefix))
        )
    ignored_suffixes = {
        # TODO(jayzhuang): Figure out whether `.dart_tool/package_config.json`
        # should be included in inputs.
        "/.dart_tool/package_config.json",
        # Allow global access to remote action download file locks,
        # which are not consumed by any other builds actions.
        ".dl-lock",
        # Downloads stubs can be kept as backups after downloading
        # the real artifact.  No build action consumes these backups.
        ".dl-stub",
        # Allow actions to read .fx-build-dir to figure out the current build
        # directory.
        #
        # This only affects local builds.
        os.path.join(src_root, ".fx-build-dir"),
    }
    ignored_path_parts = {
        # Python creates these directories with bytecode caches
        "__pycache__",
        # https://fxbug.dev/42147415: some actions are known to generate implicit outputs in
        # these directories that are unknown before the metadata collection phase.
        # It was decided to tolerate this behavior.
        "__untraced_shebangs__",
        # This temporary directory is only used to find nonterministic outputs.
        ".tmp-repro",
        # Dart kernel actions write intermediate outputs in order to generate
        # the final `.dil` file. These intermediate outputs cannot be determined
        # at GN gen time, so write them to `__untraced_dart_kernel__` and ignore
        # accesses to them.
        "__untraced_dart_kernel__",
    }
    # It's ok to access */.git/* if your action is sensitive to .git contents.
    for args_input in args.inputs:
        if ".git" in args_input.split(os.sep):
            ignored_path_parts.add(".git")
            break
    # TODO(fangism): for suffixes that we always ignore for writing, such as
    # safe or intended side-effect byproducts, make sure no declared inputs ever
    # match them.

    raw_trace = ""
    with open(args.trace_output, "r") as trace:
        raw_trace = trace.read()

    # Parse trace file.
    all_accesses = parse_fsatrace_output(raw_trace.splitlines())

    # Record all directories ever seen, so we know to ignore them.
    all_dirs = detect_all_dirs(access.path for access in all_accesses)

    # Files' contents are what matters for reproducibilty.
    # Ignore directory accesses, including symlinked dirs.
    # Also filter out temporary directories that no longer exist.
    file_accesses = [
        access
        for access in all_accesses
        if not (
            os.path.isdir(os.path.realpath(access.path))
            or access.path in all_dirs
        )
    ]

    # Filter out accesses we don't want to track.
    ignore_conditions = MatchConditions(
        prefixes=ignored_prefixes,
        suffixes=ignored_suffixes,
        components=ignored_path_parts,
    )

    exit_code = 0

    # Make sure no declared inputs match ignored patterns.
    # Ignored files should never be depended on by other actions.
    declared_ignored_inputs = {
        path for path in action.inputs if ignore_conditions.matches(path)
    }
    if args.check_inputs_not_in_ignored_set and declared_ignored_inputs:
        ignored_inputs_formatted = "\n  ".join(declared_ignored_inputs)
        print(
            f"""
The following inputs of {args.label} are ignored by action tracing, and thus,
should not be declared as dependencies.
  {ignored_inputs_formatted}

""",
            file=sys.stderr,
        )
        exit_code = 1

    # Filter out access we don't want to track.
    filtered_accesses = [
        access
        for access in file_accesses
        if access.should_check(
            ignore_conditions=ignore_conditions,
            # Ignore accesses that fall outside of the source root.
            required_path_prefix=src_root,
        )
    ]

    file_access_sets = finalize_filesystem_accesses(filtered_accesses)

    # Check for overall correctness, print diagnostics,
    # and exit with the right code.
    if args.check_access_permissions and retval == 0:
        # Verify the filesystem access trace.
        unexpected_accesses = check_access_permissions(
            accesses=file_access_sets, constraints=access_constraints
        )

        if unexpected_accesses.all_accesses:
            unexpected_accesses_formatted = str(unexpected_accesses)
            print(
                f"""
Unexpected file accesses building {args.label}:
{unexpected_accesses_formatted}

Full access trace in build directory: {args.trace_output}

See: https://fuchsia.dev/fuchsia-src/development/build/hermetic_actions

""",
                file=sys.stderr,
            )
            exit_code = args.failed_check_status
        elif not args.keep_raw_trace:
            os.remove(args.trace_output)

    if action.parsed_depfile:
        allowed_abspaths = {"/usr/bin/env"}
        abspaths = abspaths_from_depfile(
            action.parsed_depfile, allowed_abspaths
        )

        if abspaths:
            exit_code = args.failed_check_status
            one_path_per_line = "\n".join(sorted(abspaths))
            print(
                f"""
Found the following files with absolute paths in depfile {action.depfile} for {args.label}:

{one_path_per_line}

See: https://fuchsia.dev/fuchsia-src/development/build/hermetic_actions#depfiles
""",
                file=sys.stderr,
            )

    if retval != 0:
        # Always forward the action's non-zero exit code, regardless of tracer findings.
        return retval

    return exit_code


if __name__ == "__main__":
    sys.exit(main())