import bisect
import codecs
import errno
import inspect
import os
import platform
import shutil
import stat
import subprocess
import uuid
from collections import OrderedDict
from io import BytesIO, StringIO
from itertools import chain, takewhile
from tarfile import TarFile, TarInfo
from tempfile import NamedTemporaryFile, mkstemp
from jsmin import JavascriptMinify
import mozbuild.makeutil as makeutil
import mozpack.path as mozpath
from mozbuild.preprocessor import Preprocessor
from mozbuild.util import FileAvoidWrite, ensure_unicode, memoize
from mozpack.chrome.manifest import ManifestEntry, ManifestInterfaces
from mozpack.errors import ErrorMessage, errors
from mozpack.executables import elfhack, is_executable, may_elfhack, may_strip, strip
from mozpack.mozjar import JarReader
try:
import hglib
except ImportError:
hglib = None
if platform.system() != "Windows":
_copyfile = shutil.copyfile
else:
import ctypes
_kernel32 = ctypes.windll.kernel32
_CopyFileA = _kernel32.CopyFileA
_CopyFileW = _kernel32.CopyFileW
def _copyfile(src, dest):
if isinstance(src, str) and isinstance(dest, str):
_CopyFileW(src, dest, False)
elif isinstance(src, str) and isinstance(dest, str):
_CopyFileA(src, dest, False)
else:
raise TypeError("mismatched path types!")
def _open(path, mode="r"):
if "b" not in mode:
return open(path, mode, encoding="utf-8")
return open(path, mode)
class Dest:
def __init__(self, path):
self.file = None
self.mode = None
self.path = ensure_unicode(path)
@property
def name(self):
return self.path
def read(self, length=-1):
if self.mode != "r":
self.file = _open(self.path, mode="rb")
self.mode = "r"
return self.file.read(length)
def write(self, data):
if self.mode != "w":
self.file = _open(self.path, mode="wb")
self.mode = "w"
if isinstance(data, str):
data = data.encode()
return self.file.write(data)
def exists(self):
return os.path.exists(self.path)
def close(self):
if self.mode:
self.mode = None
self.file.close()
self.file = None
class BaseFile:
@staticmethod
def is_older(first, second):
return int(os.path.getmtime(first) * 1000) <= int(
os.path.getmtime(second) * 1000
)
@staticmethod
def any_newer(dest, inputs):
dest_mtime = int(os.path.getmtime(dest) * 1000)
for input in inputs:
try:
src_mtime = int(os.path.getmtime(input) * 1000)
except OSError as e:
if e.errno == errno.ENOENT:
return True
raise
if dest_mtime < src_mtime:
return True
return False
@staticmethod
def normalize_mode(mode):
ret = stat.S_IFMT(mode)
if mode & 0o0400:
ret |= 0o0444
if mode & 0o0100:
ret |= 0o0111
if mode & 0o0200:
ret |= 0o0200
return ret
def copy(self, dest, skip_if_older=True):
if isinstance(dest, str):
dest = Dest(dest)
else:
assert isinstance(dest, Dest)
can_skip_content_check = False
if not dest.exists():
can_skip_content_check = True
elif getattr(self, "path", None) and getattr(dest, "path", None):
if skip_if_older and BaseFile.is_older(self.path, dest.path):
return False
elif os.path.getsize(self.path) != os.path.getsize(dest.path):
can_skip_content_check = True
if can_skip_content_check:
if getattr(self, "path", None) and getattr(dest, "path", None):
destdir = os.path.dirname(dest.path)
os.makedirs(destdir, exist_ok=True)
_copyfile(self.path, dest.path)
shutil.copystat(self.path, dest.path)
else:
if not dest.exists():
dest.write(b"")
shutil.copyfileobj(self.open(), dest)
return True
src = self.open()
accumulated_src_content = []
while True:
dest_content = dest.read(32768)
src_content = src.read(32768)
accumulated_src_content.append(src_content)
if len(dest_content) == len(src_content) == 0:
break
if dest_content != src_content:
dest.write(b"".join(accumulated_src_content))
shutil.copyfileobj(src, dest)
break
if hasattr(self, "path") and hasattr(dest, "path"):
shutil.copystat(self.path, dest.path)
return True
def open(self):
assert self.path is not None
return open(self.path, "rb")
def read(self):
raise NotImplementedError("BaseFile.read() not implemented. Bug 1170329.")
def size(self):
return len(self.read())
@property
def mode(self):
return None
def inputs(self):
raise NotImplementedError("BaseFile.inputs() not implemented.")
class File(BaseFile):
def __init__(self, path):
self.path = ensure_unicode(path)
@property
def mode(self):
if platform.system() == "Windows":
return None
assert self.path is not None
mode = os.stat(self.path).st_mode
return self.normalize_mode(mode)
def read(self):
with open(self.path, "rb") as fh:
return fh.read()
def size(self):
return os.stat(self.path).st_size
def inputs(self):
return (self.path,)
class ExecutableFile(File):
def __init__(self, path):
File.__init__(self, path)
def copy(self, dest, skip_if_older=True):
real_dest = dest
if not isinstance(dest, str):
fd, dest = mkstemp()
os.close(fd)
os.remove(dest)
assert isinstance(dest, str)
if not File.copy(self, dest, skip_if_older) and os.path.getsize(
self.path
) > os.path.getsize(dest):
return False
try:
if may_strip(dest):
strip(dest)
if may_elfhack(dest):
elfhack(dest)
except ErrorMessage:
os.remove(dest)
raise
if real_dest != dest:
f = File(dest)
ret = f.copy(real_dest, skip_if_older)
os.remove(dest)
return ret
return True
class AbsoluteSymlinkFile(File):
def __init__(self, path):
if not os.path.isabs(path):
raise ValueError("Symlink target not absolute: %s" % path)
File.__init__(self, path)
def copy(self, dest, skip_if_older=True):
assert isinstance(dest, str)
if not hasattr(os, "symlink"):
return File.copy(self, dest, skip_if_older=skip_if_older)
if not os.path.exists(self.path):
errors.fatal("Symlink target path does not exist: %s" % self.path)
st = None
try:
st = os.lstat(dest)
except OSError as ose:
if ose.errno != errno.ENOENT:
raise
if st and stat.S_ISLNK(st.st_mode):
link = os.readlink(dest)
if link == self.path:
return False
os.remove(dest)
os.symlink(self.path, dest)
return True
if not st:
try:
os.symlink(self.path, dest)
return True
except OSError:
return File.copy(self, dest, skip_if_older=skip_if_older)
temp_dest = os.path.join(os.path.dirname(dest), str(uuid.uuid4()))
try:
os.symlink(self.path, temp_dest)
except OSError:
return File.copy(self, dest, skip_if_older=skip_if_older)
try:
os.remove(dest)
except OSError:
os.remove(temp_dest)
raise
os.rename(temp_dest, dest)
return True
class HardlinkFile(File):
def copy(self, dest, skip_if_older=True):
assert isinstance(dest, str)
if not hasattr(os, "link"):
return super(HardlinkFile, self).copy(dest, skip_if_older=skip_if_older)
try:
path_st = os.stat(self.path)
except OSError as e:
if e.errno == errno.ENOENT:
errors.fatal("Hard link target path does not exist: %s" % self.path)
else:
raise
st = None
try:
st = os.lstat(dest)
except OSError as e:
if e.errno != errno.ENOENT:
raise
if st:
if st.st_dev == path_st.st_dev and st.st_ino == path_st.st_ino:
return False
os.remove(dest)
try:
os.link(self.path, dest)
except OSError:
return super(HardlinkFile, self).copy(dest, skip_if_older=skip_if_older)
return True
class ExistingFile(BaseFile):
def __init__(self, required):
self.required = required
def copy(self, dest, skip_if_older=True):
if isinstance(dest, str):
dest = Dest(dest)
else:
assert isinstance(dest, Dest)
if not self.required:
return
if not dest.exists():
errors.fatal("Required existing file doesn't exist: %s" % dest.path)
def inputs(self):
return ()
class PreprocessedFile(BaseFile):
def __init__(
self,
path,
depfile_path,
marker,
defines,
extra_depends=None,
silence_missing_directive_warnings=False,
):
self.path = ensure_unicode(path)
self.depfile = ensure_unicode(depfile_path)
self.marker = marker
self.defines = defines
self.extra_depends = list(extra_depends or [])
self.silence_missing_directive_warnings = silence_missing_directive_warnings
def inputs(self):
pp = Preprocessor(defines=self.defines, marker=self.marker)
pp.setSilenceDirectiveWarnings(self.silence_missing_directive_warnings)
with _open(self.path, "r") as input:
with _open(os.devnull, "w") as output:
pp.processFile(input=input, output=output)
return pp.includes
def copy(self, dest, skip_if_older=True):
if isinstance(dest, str):
dest = Dest(dest)
else:
assert isinstance(dest, Dest)
if hasattr(os, "symlink") and os.path.islink(dest.path):
os.remove(dest.path)
pp_deps = set(self.extra_depends)
if self.depfile and os.path.exists(self.depfile):
target = mozpath.normpath(dest.name)
with _open(self.depfile, "rt") as fileobj:
for rule in makeutil.read_dep_makefile(fileobj):
if target in rule.targets():
pp_deps.update(rule.dependencies())
skip = False
if dest.exists() and skip_if_older:
if self.depfile and not os.path.exists(self.depfile):
skip = False
else:
skip = not BaseFile.any_newer(dest.path, pp_deps)
if skip:
return False
deps_out = None
if self.depfile:
deps_out = FileAvoidWrite(self.depfile)
pp = Preprocessor(defines=self.defines, marker=self.marker)
pp.setSilenceDirectiveWarnings(self.silence_missing_directive_warnings)
with _open(self.path, "r") as input:
pp.processFile(input=input, output=dest, depfile=deps_out)
dest.close()
if self.depfile:
deps_out.close()
return True
class GeneratedFile(BaseFile):
def __init__(self, content):
self._content = content
@property
def content(self):
if inspect.isfunction(self._content):
self._content = self._content()
if isinstance(self._content, str):
return self._content.encode()
return self._content
@content.setter
def content(self, content):
self._content = content
def open(self):
return BytesIO(self.content)
def read(self):
return self.content
def size(self):
return len(self.content)
def inputs(self):
return ()
class DeflatedFile(BaseFile):
def __init__(self, file):
from mozpack.mozjar import JarFileReader
assert isinstance(file, JarFileReader)
self.file = file
def open(self):
self.file.seek(0)
return self.file
class ExtractedTarFile(GeneratedFile):
def __init__(self, tar, info):
assert isinstance(info, TarInfo)
assert isinstance(tar, TarFile)
GeneratedFile.__init__(self, tar.extractfile(info).read())
self._unix_mode = self.normalize_mode(info.mode)
@property
def mode(self):
return self._unix_mode
def read(self):
return self.content
class ManifestFile(BaseFile):
def __init__(self, base, entries=None):
self._base = base
self._entries = []
self._interfaces = []
for e in entries or []:
self.add(e)
def add(self, entry):
assert isinstance(entry, ManifestEntry)
if isinstance(entry, ManifestInterfaces):
self._interfaces.append(entry)
else:
self._entries.append(entry)
def remove(self, entry):
assert isinstance(entry, ManifestEntry)
if isinstance(entry, ManifestInterfaces):
self._interfaces.remove(entry)
else:
self._entries.remove(entry)
def open(self):
content = "".join(
"%s\n" % e.rebase(self._base)
for e in chain(self._entries, self._interfaces)
)
return BytesIO(content.encode())
def __iter__(self):
return chain(self._entries, self._interfaces)
def isempty(self):
return len(self._entries) + len(self._interfaces) == 0
class MinifiedCommentStripped(BaseFile):
def __init__(self, file):
assert isinstance(file, BaseFile)
self._file = file
def open(self):
content = "".join(
l
for l in [s.decode() for s in self._file.open().readlines()]
if not l.startswith("#")
)
return BytesIO(content.encode())
class MinifiedJavaScript(BaseFile):
def __init__(self, file, verify_command=None):
assert isinstance(file, BaseFile)
self._file = file
self._verify_command = verify_command
def open(self):
output = StringIO()
minify = JavascriptMinify(
codecs.getreader("utf-8")(self._file.open()), output, quote_chars="'\"`"
)
minify.minify()
output.seek(0)
output_source = output.getvalue().encode()
output = BytesIO(output_source)
if not self._verify_command:
return output
input_source = self._file.open().read()
with NamedTemporaryFile("wb+") as fh1, NamedTemporaryFile("wb+") as fh2:
fh1.write(input_source)
fh2.write(output_source)
fh1.flush()
fh2.flush()
try:
args = list(self._verify_command)
args.extend([fh1.name, fh2.name])
subprocess.check_output(
args, stderr=subprocess.STDOUT, universal_newlines=True
)
except subprocess.CalledProcessError as e:
errors.warn(
"JS minification verification failed for %s:"
% (getattr(self._file, "path", "<unknown>"))
)
for line in e.output.splitlines():
errors.warn(line)
return self._file.open()
return output
class BaseFinder:
def __init__(
self, base, minify=False, minify_js=False, minify_js_verify_command=None
):
if minify_js and not minify:
raise ValueError("minify_js requires minify.")
self.base = mozpath.normsep(base)
self._minify = minify
self._minify_js = minify_js
self._minify_js_verify_command = minify_js_verify_command
def find(self, pattern):
while pattern.startswith("/"):
pattern = pattern[1:]
for p, f in self._find(pattern):
yield p, self._minify_file(p, f)
def get(self, path):
files = list(self.find(path))
if len(files) != 1:
return None
return files[0][1]
def __iter__(self):
return self.find("")
def __contains__(self, pattern):
raise RuntimeError(
"'in' operator forbidden for %s. Use contains()." % self.__class__.__name__
)
def contains(self, pattern):
return any(self.find(pattern))
def _minify_file(self, path, file):
if not self._minify or isinstance(file, ExecutableFile):
return file
if path.endswith((".ftl", ".properties")):
return MinifiedCommentStripped(file)
if self._minify_js and path.endswith((".js", ".jsm", ".mjs")):
return MinifiedJavaScript(file, self._minify_js_verify_command)
return file
def _find_helper(self, pattern, files, file_getter):
if "*" in pattern:
for p in files:
if mozpath.match(p, pattern):
yield p, file_getter(p)
elif pattern == "":
for p in files:
yield p, file_getter(p)
elif pattern in files:
yield pattern, file_getter(pattern)
else:
for p in files:
if mozpath.basedir(p, [pattern]) == pattern:
yield p, file_getter(p)
class FileFinder(BaseFinder):
def __init__(
self,
base,
find_executables=False,
ignore=(),
ignore_broken_symlinks=False,
find_dotfiles=False,
**kargs
):
BaseFinder.__init__(self, base, **kargs)
self.find_dotfiles = find_dotfiles
self.find_executables = find_executables
self.ignore = tuple(mozpath.normsep(path) for path in ignore)
self.ignore_broken_symlinks = ignore_broken_symlinks
def _find(self, pattern):
if "*" in pattern:
return self._find_glob("", mozpath.split(pattern))
elif os.path.isdir(os.path.join(self.base, pattern)):
return self._find_dir(pattern)
else:
f = self.get(pattern)
return ((pattern, f),) if f else ()
def _find_dir(self, path):
for p in self.ignore:
if mozpath.match(path, p):
return
for p in sorted(os.listdir(os.path.join(self.base, path))):
if p.startswith("."):
if p in (".", ".."):
continue
if not self.find_dotfiles:
continue
for p_, f in self._find(mozpath.join(path, p)):
yield p_, f
def get(self, path):
srcpath = os.path.join(self.base, path)
if not os.path.lexists(srcpath):
return None
if self.ignore_broken_symlinks and not os.path.exists(srcpath):
return None
for p in self.ignore:
if mozpath.match(path, p):
return None
if self.find_executables and is_executable(srcpath):
return ExecutableFile(srcpath)
else:
return File(srcpath)
def _find_glob(self, base, pattern):
if not pattern:
for p, f in self._find(base):
yield p, f
elif pattern[0] == "**":
for p, f in self._find(base):
if mozpath.match(p, mozpath.join(*pattern)):
yield p, f
elif "*" in pattern[0]:
if not os.path.exists(os.path.join(self.base, base)):
return
for p in self.ignore:
if mozpath.match(base, p):
return
for p in sorted(os.listdir(os.path.join(self.base, base))):
if p.startswith(".") and not pattern[0].startswith("."):
continue
if mozpath.match(p, pattern[0]):
for p_, f in self._find_glob(mozpath.join(base, p), pattern[1:]):
yield p_, f
else:
for p, f in self._find_glob(mozpath.join(base, pattern[0]), pattern[1:]):
yield p, f
class JarFinder(BaseFinder):
def __init__(self, base, reader, **kargs):
assert isinstance(reader, JarReader)
BaseFinder.__init__(self, base, **kargs)
self._files = OrderedDict((f.filename, f) for f in reader)
def _find(self, pattern):
return self._find_helper(
pattern, self._files, lambda x: DeflatedFile(self._files[x])
)
class TarFinder(BaseFinder):
def __init__(self, base, tar, **kargs):
assert isinstance(tar, TarFile)
self._tar = tar
BaseFinder.__init__(self, base, **kargs)
self._files = OrderedDict((f.name, f) for f in tar if f.isfile())
def _find(self, pattern):
return self._find_helper(
pattern, self._files, lambda x: ExtractedTarFile(self._tar, self._files[x])
)
class ComposedFinder(BaseFinder):
def __init__(self, finders):
from mozpack.copier import FileRegistry
self.files = FileRegistry()
for base, finder in sorted(finders.items()):
if self.files.contains(base):
self.files.remove(base)
for p, f in finder.find(""):
self.files.add(mozpath.join(base, p), f)
def find(self, pattern):
for p in self.files.match(pattern):
yield p, self.files[p]
class MercurialFile(BaseFile):
def __init__(self, client, rev, path):
self._content = client.cat([path.encode()], rev=rev.encode())
def open(self):
return BytesIO(self._content)
def read(self):
return self._content
class MercurialRevisionFinder(BaseFinder):
def __init__(self, repo, rev=".", recognize_repo_paths=False, **kwargs):
if not hglib:
raise Exception("hglib package not found")
super(MercurialRevisionFinder, self).__init__(base=repo, **kwargs)
self._root = mozpath.normpath(repo).rstrip("/")
self._recognize_repo_paths = recognize_repo_paths
oldcwd = os.getcwd()
os.chdir(self._root)
try:
self._client = hglib.open(path=repo, encoding=b"utf-8")
finally:
os.chdir(oldcwd)
self._rev = rev if rev is not None else "."
self._files = OrderedDict()
out = self._client.rawcommand(
[
b"files",
b"--rev",
self._rev.encode(),
]
)
for relpath in out.splitlines():
self._files[mozpath.normpath(relpath).decode()] = None
def _find(self, pattern):
if self._recognize_repo_paths:
raise NotImplementedError("cannot use find with recognize_repo_path")
return self._find_helper(pattern, self._files, self._get)
def get(self, path):
path = mozpath.normpath(path)
if self._recognize_repo_paths:
if not path.startswith(self._root):
raise ValueError(
"lookups in recognize_repo_paths mode must be "
"prefixed with repo path: %s" % path
)
path = path[len(self._root) + 1 :]
try:
return self._get(path)
except KeyError:
return None
def _get(self, path):
f = self._files[path]
if not f:
f = MercurialFile(self._client, self._rev, path)
self._files[path] = f
return f
class FileListFinder(BaseFinder):
def __init__(self, files):
self._files = sorted(files)
@memoize
def _match(self, pattern):
ret = []
components = pattern.split("/")
prefix = "/".join(takewhile(lambda s: "*" not in s, components))
start = bisect.bisect_left(self._files, prefix)
for i in range(start, len(self._files)):
f = self._files[i]
if not f.startswith(prefix):
break
if "/." in f[len(prefix) :]:
continue
if mozpath.match(f, pattern):
ret.append(f)
return ret
def find(self, pattern):
pattern = pattern.strip("/")
for path in self._match(pattern):
yield path, File(path)