28270112145

Committed 26 Jun 2026 11:03PM UTC coverage: 95.859% (-0.005%) from 95.864%

Build # 28270112145

Build Type

Pull #895

github

Committed by

web-flow

Commit Message

Merge 030a1b5a2 into dc6787fd9

Pull Request Pull Request #895: Fix input path mounting on retry

Coverage Stats

14 of 14 new or added lines in 2 files covered. (100.0%)

21 existing lines in 4 files now uncovered.

8797 of 9177 relevant lines covered (95.86%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.76

/WDL/Zip.py

"""
Routines for packaging a WDL source file, with all imported source files, into a ZIP file.

*New in v1.5.0*
"""

import io
import os
import json
import glob
import pathlib
import shutil
import logging
import tarfile
import tempfile
import contextlib
import zipfile
from typing import List, Dict, Optional, Any, Iterator, NamedTuple, Tuple, Set

from . import Tree, Error
from ._util import path_really_within


def build(
    top_doc: Tree.Document,
    archive: str,
    logger: logging.Logger,
    inputs: Optional[Dict[str, Any]] = None,
    meta: Optional[Dict[str, Any]] = None,
    archive_format: str = "zip",
    additional_files: Optional[List[str]] = None,
):
    """
    Generate zip archive of the WDL document, all its imports, optional default inputs, and a
    generated manifest JSON.

    If imports are drawn from outside the main WDL's directory (or by URI), they'll be stored in a
    special subdirectory and import statements will be rewritten to match.
    """

    with contextlib.ExitStack() as cleanup:
        # write WDL source code to temp directory
        dir_to_zip, zip_paths, wdls = _build_source_dir(cleanup, top_doc, logger)

        # add MANIFEST.json; schema roughly following Amazon Genomics CLI's:
        #  https://aws.github.io/amazon-genomics-cli/docs/concepts/workflows/#multi-file-workflows
        manifest: Dict[str, Any] = {"mainWorkflowURL": os.path.basename(top_doc.pos.abspath)}
        if meta:
            manifest["meta"] = meta
        if inputs:
            manifest["inputFileURLs"] = ["default_input.json"]
            with open(os.path.join(dir_to_zip, "default_input.json"), "w") as inputs_file:
                json.dump(inputs, inputs_file, indent=2)
        with open(os.path.join(dir_to_zip, "MANIFEST.json"), "w") as manifest_file:
            json.dump(manifest, manifest_file, indent=2)
        logger.debug("manifest = " + json.dumps(manifest))
        if additional_files:
            add_additional_files(dir_to_zip, additional_files, zip_paths, wdls, logger)

        # zip the temp directory (into another temp directory)
        spool_dir = cleanup.enter_context(tempfile.TemporaryDirectory(prefix="miniwdl_zip_"))
        spool_zip = os.path.join(spool_dir, os.path.basename(archive))
        logger.info(f"Prepare archive {spool_zip} from directory {dir_to_zip}")
        create_reproducible_archive(dir_to_zip, spool_zip, archive_format)

        # move into final location (hopefully atomic)
        dirname = os.path.dirname(archive)
        if dirname:
            os.makedirs(dirname, exist_ok=True)
        logger.info(f"Move archive to destination {archive}")
        shutil.move(spool_zip, archive)


def build_source_dir(
    cleanup: contextlib.ExitStack, top_doc: Tree.Document, logger: logging.Logger
) -> str:
    zip_dir, _zip_paths, _wdls = _build_source_dir(cleanup, top_doc, logger)
    return zip_dir


def _build_source_dir(
    cleanup: contextlib.ExitStack, top_doc: Tree.Document, logger: logging.Logger
) -> Tuple[str, Dict[str, str], Dict[str, Tree.Document]]:
    """
    Stage rewritten WDL source files and return the path mapping used to do it.

    ``build_source_dir()`` historically returned only the staging directory. The additional-file
    path logic needs the same original-source to archive-path mapping, so this internal variant
    returns all three pieces while preserving the public helper above.
    """
    # directory of main WDL file (possibly URI)
    main_dir = os.path.dirname(top_doc.pos.abspath).rstrip("/") + "/"

    # collect all WDL docs keyed by abspath
    wdls = {}
    queue = [top_doc]
    while queue:
        a_doc = queue.pop()
        for imported_doc in a_doc.imports:
            assert imported_doc.doc
            queue.append(imported_doc.doc)
        wdls[a_doc.pos.abspath] = a_doc

    # derive archive paths
    zip_paths = build_zip_paths(main_dir, wdls, logger)
    assert sorted(list(zip_paths.keys())) == sorted(list(wdls.keys()))
    assert zip_paths[top_doc.pos.abspath] == os.path.basename(top_doc.pos.abspath)

    # write source files into temp directory (rewriting imports as needed)
    zip_dir = cleanup.enter_context(tempfile.TemporaryDirectory(prefix="miniwdl_zip_"))
    for abspath, a_doc in wdls.items():
        source_lines = rewrite_imports(a_doc, zip_paths, logger)
        fn = os.path.join(zip_dir, zip_paths[abspath])
        os.makedirs(os.path.dirname(fn), exist_ok=True)
        with open(fn, "w") as outfile:
            for line in source_lines:
                print(line, file=outfile)

    return zip_dir, zip_paths, wdls


def build_zip_paths(
    main_dir: str, wdls: Dict[str, Tree.Document], logger: logging.Logger
) -> Dict[str, str]:
    # compute the path inside the archive at which to store each document

    ans = {}
    outside_warn = False
    for abspath in wdls.keys():
        if abspath.startswith(main_dir):
            ans[abspath] = os.path.relpath(abspath, main_dir)
        else:
            # place outside import under __outside_wdl, vaguely reproducing directory structure
            abspath2 = abspath.replace("://", "_")
            prefix = os.path.commonprefix([abspath2, main_dir.replace("://", "_")])
            if prefix and not prefix.endswith("/"):
                prefix = os.path.dirname(prefix) + "/"
            ans[abspath] = "__outside_wdl/" + abspath2[len(prefix) :]
            outside_warn = True
        logger.info(f"{ans[abspath]} <= {abspath}")

    if outside_warn:
        logger.warning(
            "One or more source files are imported from outside the top-level WDL's directory."
            " The source archive will store them under __outside_wdl/"
            " and WDL import statements will be rewritten to match."
        )

    return ans


def rewrite_imports(
    doc: Tree.Document, zip_paths: Dict[str, str], logger: logging.Logger
) -> List[str]:
    # rewrite doc source_lines, changing import statements to refer to relative path in zip
    source_lines = doc.source_lines.copy()

    for imp in doc.imports:
        assert imp.doc
        lo = imp.pos.line - 1
        hi = imp.pos.end_line
        found = False
        for lineno in range(lo, hi):
            line = source_lines[lineno]
            old_uri = imp.uri
            new_uri = os.path.relpath(
                zip_paths[imp.doc.pos.abspath], os.path.dirname(zip_paths[doc.pos.abspath])
            )
            for quot in ('"', "'"):
                old_uri_pattern = f"{quot}{old_uri}{quot}"
                if old_uri_pattern in line:
                    assert quot not in new_uri
                    found = True
                    line2 = line.replace(old_uri_pattern, f"{quot}{new_uri}{quot}")
                    if line != line2:
                        logger.debug(doc.pos.abspath)
                        logger.debug("  " + line)
                        logger.debug("  => " + line2)
                        source_lines[lineno] = line2
        assert found

    return source_lines


def add_additional_files(
    zip_dir: str,
    additional_files: List[str],
    zip_paths: Dict[str, str],
    wdls: Dict[str, Tree.Document],
    logger: logging.Logger,
) -> None:
    """
    Add files/directories to an archive staging directory, preserving paths relative to WDL source
    directories represented in ``zip_paths``.

    Examples:

    * ``/proj/data/ref.fa`` -> ``data/ref.fa`` (when ``/proj/main.wdl`` is archived as
      ``main.wdl``)
    * ``/shared/lib/data/ref.fa`` -> ``__outside_wdl/lib/data/ref.fa`` (when
      ``/shared/lib/tasks.wdl`` is rewritten under ``__outside_wdl/lib/tasks.wdl``)
    """
    logger.debug(f"Additional files: {additional_files}")
    source_dirs = _additional_source_dirs(zip_paths, wdls)
    if not source_dirs:
        raise Error.InputError("Additional files require local WDL source files")

    copied: Dict[str, str] = {}
    for pattern in additional_files:
        # Leave non-glob paths literal so we can report "not found" rather than "matched nothing".
        matches = glob.glob(pattern, recursive=True) if glob.has_magic(pattern) else [pattern]
        matches = sorted(matches)
        if not matches:
            raise Error.InputError("Additional file pattern matched nothing: " + pattern)
        for match in matches:
            if not os.path.lexists(match):
                raise Error.InputError("Additional file not found: " + match)
            for src, dest in _additional_files_from_path(match, source_dirs):
                if dest in copied:
                    # Overlapping globs may name the same source file twice; that's harmless.
                    # A different realpath landing at the same archive path is a real collision.
                    if copied[dest] == os.path.realpath(src):
                        continue
                    raise Error.InputError("Additional file overwrites existing path: " + dest)
                dest_path = os.path.join(zip_dir, dest)
                if os.path.exists(dest_path) or os.path.lexists(dest_path):
                    raise Error.InputError("Additional file overwrites existing path: " + dest)
                if not path_really_within(dest_path, zip_dir):
                    raise Error.InputError("Invalid additional file destination: " + dest)
                os.makedirs(os.path.dirname(dest_path), exist_ok=True)
                shutil.copyfile(src, dest_path)
                copied[dest] = os.path.realpath(src)
                logger.info(f"{dest} <= {src}")


def _additional_source_dirs(
    zip_paths: Dict[str, str], wdls: Dict[str, Tree.Document]
) -> List[Tuple[str, str]]:
    """
    Pair each local WDL source directory with its corresponding archive directory.

    Additional files are accepted only when they resolve under one of these directories. Sorting by
    longest source directory first makes nested imports win over their parents, which is needed to
    preserve the source-relative meaning of WDL 1.2 paths after import rewrites.

    Examples:

    * ``/proj/tasks/t.wdl`` -> ``tasks`` (nested local import directory wins over ``/proj``)
    * ``/proj/main.wdl`` -> ``""`` (top-level source directory maps to the archive root)
    * ``/shared/lib/t.wdl`` -> ``__outside_wdl/lib`` (outside import keeps its rewritten prefix)
    """
    ans: Dict[str, str] = {}
    for abspath, zip_path in zip_paths.items():
        if abspath in wdls:
            source_dir = wdls[abspath].source_dir.rstrip("/")
            if not source_dir or not os.path.isfile(abspath):
                continue
            archive_dir = os.path.dirname(zip_path)
            # Different archive paths for one source directory would make it impossible to know
            # where non-WDL neighbors from that directory should go.
            if source_dir in ans and ans[source_dir] != archive_dir:
                raise Error.InputError(
                    "Cannot place additional files relative to ambiguous WDL source directory: "
                    + source_dir
                )
            ans[source_dir] = archive_dir
    return sorted(ans.items(), key=lambda item: len(item[0]), reverse=True)


def _additional_files_from_path(
    path: str, source_dirs: List[Tuple[str, str]]
) -> Iterator[Tuple[str, str]]:
    """
    Expand one matched additional path into concrete file/archive-path pairs.

    Directories are walked recursively, following symlinks only after each directory entry passes
    ``_additional_dest()``. ``visited_dirs`` prevents symlink loops from cycling indefinitely while
    still allowing safe symlinks to contribute the file contents they point to.

    Examples:

    * ``/proj/data/a.txt`` -> ``data/a.txt`` (file found while walking ``/proj/data``)
    * ``/proj/data/sub/b.txt`` -> ``data/sub/b.txt`` (recursive directory contents are kept)
    * ``/proj/data/latest.txt`` -> ``data/latest.txt`` (safe symlink content is copied)
    """
    path = os.path.abspath(path)
    _additional_dest(path, source_dirs)  # validate the requested path, including symlink target
    if os.path.isdir(path):
        visited_dirs: Set[str] = set()
        for dirpath, dirnames, filenames in os.walk(path, followlinks=True):
            # Re-check every walked directory because os.walk follows symlinks when asked.
            _additional_dest(dirpath, source_dirs)
            real_dirpath = os.path.realpath(dirpath)
            if real_dirpath in visited_dirs:
                dirnames[:] = []
                continue
            visited_dirs.add(real_dirpath)
            for dirname in list(dirnames):
                dirname_path = os.path.join(dirpath, dirname)
                try:
                    # Reject a directory symlink before os.walk descends into it.
                    _additional_dest(dirname_path, source_dirs)
                except Error.InputError:
                    raise Error.InputError(
                        "Additional directory contains unsafe symlink or path: " + dirname_path
                    ) from None
                if os.path.realpath(dirname_path) in visited_dirs:
                    # Avoid loops like data/back -> .., even when the target is otherwise safe.
                    dirnames.remove(dirname)
            for filename in sorted(filenames):
                filename_path = os.path.join(dirpath, filename)
                yield filename_path, _additional_dest(filename_path, source_dirs)
    else:
        yield path, _additional_dest(path, source_dirs)


def _additional_dest(path: str, source_dirs: List[Tuple[str, str]]) -> str:
    """
    Compute the archive path for one additional file or directory after safety checks.

    The path and its realpath target must both remain inside the selected source directory. This
    permits ordinary in-tree symlinks while rejecting symlinks that would package files from outside
    the WDL source tree.

    Examples:

    * ``/proj/data/a.txt`` -> ``data/a.txt`` (source dir ``/proj`` maps to archive root)
    * ``/shared/lib/data/a.txt`` -> ``__outside_wdl/lib/data/a.txt`` (source dir
      ``/shared/lib`` maps to ``__outside_wdl/lib``)
    * ``/proj/data/secret -> /etc/passwd`` -> rejected (symlink target escapes ``/proj``)
    """
    if not (os.path.isfile(path) or os.path.isdir(path)):
        raise Error.InputError("Additional path is neither a file nor a directory: " + path)
    # Resolve symlinks in the parent directory (e.g. macOS /tmp -> /private/tmp) so the containment
    # check lines up with the realpath-canonicalized source directories, while keeping the leaf name
    # so safe in-tree symlinks retain their own archive path.
    real_path = os.path.join(os.path.realpath(os.path.dirname(path)), os.path.basename(path))
    for source_dir, archive_dir in source_dirs:
        relpath = os.path.relpath(real_path, os.path.realpath(source_dir))
        if relpath.startswith(".." + os.sep) or relpath == "..":
            continue
        # path_really_within additionally rejects symlinks whose target escapes the source tree.
        if path_really_within(path, source_dir):
            return os.path.normpath(os.path.join(archive_dir, relpath))
    raise Error.InputError("Additional path must reside within a WDL source directory: " + path)


def create_reproducible_archive(zip_dir: str, output_path: str, format: str):
    # write zip/tar archive with internal filenames lexicographically-ordered and all timestamps
    # set to an arbitrary constant
    src_dest_list = [
        (path, path.relative_to(zip_dir))
        for path in pathlib.Path(zip_dir).glob("**/*")  # Finds all files recursively
        if path.is_file()
        or path.is_symlink()  # Symlinks will be included in the zip as normal files
    ]
    # Sort paths by destination
    src_dest_list.sort(key=lambda x: x[1])
    if format == "zip":
        _write_no_mtime_zip(output_path, src_dest_list)
    elif format == "tar":
        _write_no_mtime_tar(output_path, src_dest_list)
    else:
        raise ValueError(f"Unknown format: {format}")
    return output_path


def _write_no_mtime_zip(zip_archive: str, src_dest_list: List[Tuple[pathlib.Path, pathlib.Path]]):
    with zipfile.ZipFile(zip_archive, "w") as archive:
        for src, dest in src_dest_list:
            # This always sets the mod time at 1980-1-1
            dest_info = zipfile.ZipInfo(str(dest))
            with archive.open(dest_info, "w") as archive_file:
                with open(src, "rb") as in_file:
                    while True:
                        block = in_file.read(io.DEFAULT_BUFFER_SIZE)
                        if not block:
                            break
                        archive_file.write(block)


def _write_no_mtime_tar(tar_archive: str, src_dest_list: List[Tuple[pathlib.Path, pathlib.Path]]):
    with tarfile.TarFile(tar_archive, "w") as archive:
        for src, dest in src_dest_list:
            dest_info = tarfile.TarInfo(str(dest))  # Mtime by default at 0
            dest_info.size = os.stat(src).st_size
            with open(src, "rb") as in_file:
                archive.addfile(dest_info, in_file)


UnpackedZip = NamedTuple(
    "UnpackedZip", [("dir", str), ("main_wdl", str), ("input_file", Optional[str])]
)
"""
Contextual value of `WDL.Zip.unpack()`: absolute paths of source directory, main WDL, and default
input JSON file (if any). The source directory prefixes the latter paths.
"""


@contextlib.contextmanager
def unpack(archive_fn: str, tempdir_parent: Optional[str] = None) -> Iterator[UnpackedZip]:
    """
    Open a context with the WDL source archive unpacked into a temp directory, yielding
    `UnpackedZip`. The temp directory will be deleted on context exit.

    A path to the MANIFEST.json of an already-unpacked source archive may also be used, or a
    directory containing one. In this case, it is NOT deleted on context exit. ::

        with WDL.Zip.unpack("/path/to/source.zip") as unpacked:
            doc = WDL.load(unpacked.main_wdl)
            ...
    """
    with contextlib.ExitStack() as cleanup:
        # extract zip if needed (also allowing use of already-extracted manifest/dir)
        if os.path.isdir(archive_fn):
            archive_fn = os.path.join(archive_fn, "MANIFEST.json")
        if os.path.basename(archive_fn) == "MANIFEST.json":
            manifest_fn = archive_fn
        else:
            try:
                dn = cleanup.enter_context(
                    tempfile.TemporaryDirectory(prefix="miniwdl_run_zip_", dir=tempdir_parent)
                )
            except OSError as exn:
                msg = "Unable to create temporary directory for unpacking source archive"
                if tempdir_parent:
                    msg += (
                        " under "
                        + tempdir_parent
                        + "; set TMPDIR to a writable directory under file_io.root"
                    )
                raise Error.InputError(msg) from exn
            try:
                shutil.unpack_archive(archive_fn, dn)
            except Exception:
                raise Error.InputError("Unreadable source archive " + archive_fn)
            manifest_fn = os.path.join(dn, "MANIFEST.json")

        try:
            with open(manifest_fn) as infile:
                manifest = json.load(infile)
            assert isinstance(manifest, dict) and isinstance(
                manifest.get("mainWorkflowURL", None), str
            )
        except Exception:
            raise Error.InputError("Missing or invalid MANIFEST.json in " + archive_fn)

        dn = os.path.abspath(os.path.dirname(manifest_fn))
        main_wdl = manifest["mainWorkflowURL"]

        input_file = None
        if (
            isinstance(manifest.get("inputFileURLs", None), list)
            and manifest["inputFileURLs"]
            and isinstance(manifest["inputFileURLs"][0], str)
        ):
            input_file = manifest["inputFileURLs"][0]

        # sanity check
        main_wdl_abs = os.path.join(dn, main_wdl)
        input_file_abs = os.path.join(dn, input_file) if input_file else None
        if not (os.path.isfile(main_wdl_abs) and path_really_within(main_wdl_abs, dn)) or (
            input_file_abs
            and not (os.path.isfile(input_file_abs) and path_really_within(input_file_abs, dn))
        ):
            raise Error.InputError(
                "MANIFEST.json refers to missing or invalid files in " + archive_fn
            )

        yield UnpackedZip(dn, main_wdl_abs, input_file_abs)

1	"""
2	Routines for packaging a WDL source file, with all imported source files, into a ZIP file.
3
4	New in v1.5.0
5	"""
6
7	import io	1✔
8	import os	1✔
9	import json	1✔
10	import glob	1✔
11	import pathlib	1✔
12	import shutil	1✔
13	import logging	1✔
14	import tarfile	1✔
15	import tempfile	1✔
16	import contextlib	1✔
17	import zipfile	1✔
18	from typing import List, Dict, Optional, Any, Iterator, NamedTuple, Tuple, Set	1✔
19
20	from . import Tree, Error	1✔
21	from ._util import path_really_within	1✔
22
23
24	def build(	1✔
25	top_doc: Tree.Document,
26	archive: str,
27	logger: logging.Logger,
28	inputs: Optional[Dict[str, Any]] = None,
29	meta: Optional[Dict[str, Any]] = None,
30	archive_format: str = "zip",
31	additional_files: Optional[List[str]] = None,
32	):
33	"""
34	Generate zip archive of the WDL document, all its imports, optional default inputs, and a
35	generated manifest JSON.
36
37	If imports are drawn from outside the main WDL's directory (or by URI), they'll be stored in a
38	special subdirectory and import statements will be rewritten to match.
39	"""
40
41	with contextlib.ExitStack() as cleanup:	1✔
42	# write WDL source code to temp directory
43	dir_to_zip, zip_paths, wdls = _build_source_dir(cleanup, top_doc, logger)	1✔
44
45	# add MANIFEST.json; schema roughly following Amazon Genomics CLI's:
46	# https://aws.github.io/amazon-genomics-cli/docs/concepts/workflows/#multi-file-workflows
47	manifest: Dict[str, Any] = {"mainWorkflowURL": os.path.basename(top_doc.pos.abspath)}	1✔
48	if meta:	1✔
49	manifest["meta"] = meta	1✔
50	if inputs:	1✔
51	manifest["inputFileURLs"] = ["default_input.json"]	1✔
52	with open(os.path.join(dir_to_zip, "default_input.json"), "w") as inputs_file:	1✔
53	json.dump(inputs, inputs_file, indent=2)	1✔
54	with open(os.path.join(dir_to_zip, "MANIFEST.json"), "w") as manifest_file:	1✔
55	json.dump(manifest, manifest_file, indent=2)	1✔
56	logger.debug("manifest = " + json.dumps(manifest))	1✔
57	if additional_files:	1✔
58	add_additional_files(dir_to_zip, additional_files, zip_paths, wdls, logger)	1✔
59
60	# zip the temp directory (into another temp directory)
61	spool_dir = cleanup.enter_context(tempfile.TemporaryDirectory(prefix="miniwdl_zip_"))	1✔
62	spool_zip = os.path.join(spool_dir, os.path.basename(archive))	1✔
63	logger.info(f"Prepare archive {spool_zip} from directory {dir_to_zip}")	1✔
64	create_reproducible_archive(dir_to_zip, spool_zip, archive_format)	1✔
65
66	# move into final location (hopefully atomic)
67	dirname = os.path.dirname(archive)	1✔
68	if dirname:	1✔
69	os.makedirs(dirname, exist_ok=True)	1✔
70	logger.info(f"Move archive to destination {archive}")	1✔
71	shutil.move(spool_zip, archive)	1✔
72
73
74	def build_source_dir(	1✔
75	cleanup: contextlib.ExitStack, top_doc: Tree.Document, logger: logging.Logger
76	) -> str:
77	zip_dir, _zip_paths, _wdls = _build_source_dir(cleanup, top_doc, logger)	1✔
78	return zip_dir	1✔
79
80
81	def _build_source_dir(	1✔
82	cleanup: contextlib.ExitStack, top_doc: Tree.Document, logger: logging.Logger
83	) -> Tuple[str, Dict[str, str], Dict[str, Tree.Document]]:
84	"""
85	Stage rewritten WDL source files and return the path mapping used to do it.
86
87	``build_source_dir()`` historically returned only the staging directory. The additional-file
88	path logic needs the same original-source to archive-path mapping, so this internal variant
89	returns all three pieces while preserving the public helper above.
90	"""
91	# directory of main WDL file (possibly URI)
92	main_dir = os.path.dirname(top_doc.pos.abspath).rstrip("/") + "/"	1✔
93
94	# collect all WDL docs keyed by abspath
95	wdls = {}	1✔
96	queue = [top_doc]	1✔
97	while queue:	1✔
98	a_doc = queue.pop()	1✔
99	for imported_doc in a_doc.imports:	1✔
100	assert imported_doc.doc	1✔
101	queue.append(imported_doc.doc)	1✔
102	wdls[a_doc.pos.abspath] = a_doc	1✔
103
104	# derive archive paths
105	zip_paths = build_zip_paths(main_dir, wdls, logger)	1✔
106	assert sorted(list(zip_paths.keys())) == sorted(list(wdls.keys()))	1✔
107	assert zip_paths[top_doc.pos.abspath] == os.path.basename(top_doc.pos.abspath)	1✔
108
109	# write source files into temp directory (rewriting imports as needed)
110	zip_dir = cleanup.enter_context(tempfile.TemporaryDirectory(prefix="miniwdl_zip_"))	1✔
111	for abspath, a_doc in wdls.items():	1✔
112	source_lines = rewrite_imports(a_doc, zip_paths, logger)	1✔
113	fn = os.path.join(zip_dir, zip_paths[abspath])	1✔
114	os.makedirs(os.path.dirname(fn), exist_ok=True)	1✔
115	with open(fn, "w") as outfile:	1✔
116	for line in source_lines:	1✔
117	print(line, file=outfile)	1✔
118
119	return zip_dir, zip_paths, wdls	1✔
120
121
122	def build_zip_paths(	1✔
123	main_dir: str, wdls: Dict[str, Tree.Document], logger: logging.Logger
124	) -> Dict[str, str]:
125	# compute the path inside the archive at which to store each document
126
127	ans = {}	1✔
128	outside_warn = False	1✔
129	for abspath in wdls.keys():	1✔
130	if abspath.startswith(main_dir):	1✔
131	ans[abspath] = os.path.relpath(abspath, main_dir)	1✔
132	else:
133	# place outside import under __outside_wdl, vaguely reproducing directory structure
134	abspath2 = abspath.replace("://", "_")	1✔
135	prefix = os.path.commonprefix([abspath2, main_dir.replace("://", "_")])	1✔
136	if prefix and not prefix.endswith("/"):	1✔
137	prefix = os.path.dirname(prefix) + "/"	1✔
138	ans[abspath] = "__outside_wdl/" + abspath2[len(prefix) :]	1✔
139	outside_warn = True	1✔
140	logger.info(f"{ans[abspath]} <= {abspath}")	1✔
141
142	if outside_warn:	1✔
143	logger.warning(	1✔
144	"One or more source files are imported from outside the top-level WDL's directory."
145	" The source archive will store them under __outside_wdl/"
146	" and WDL import statements will be rewritten to match."
147	)
148
149	return ans	1✔
150
151
152	def rewrite_imports(	1✔
153	doc: Tree.Document, zip_paths: Dict[str, str], logger: logging.Logger
154	) -> List[str]:
155	# rewrite doc source_lines, changing import statements to refer to relative path in zip
156	source_lines = doc.source_lines.copy()	1✔
157
158	for imp in doc.imports:	1✔
159	assert imp.doc	1✔
160	lo = imp.pos.line - 1	1✔
161	hi = imp.pos.end_line	1✔
162	found = False	1✔
163	for lineno in range(lo, hi):	1✔
164	line = source_lines[lineno]	1✔
165	old_uri = imp.uri	1✔
166	new_uri = os.path.relpath(	1✔
167	zip_paths[imp.doc.pos.abspath], os.path.dirname(zip_paths[doc.pos.abspath])
168	)
169	for quot in ('"', "'"):	1✔
170	old_uri_pattern = f"{quot}{old_uri}{quot}"	1✔
171	if old_uri_pattern in line:	1✔
172	assert quot not in new_uri	1✔
173	found = True	1✔
174	line2 = line.replace(old_uri_pattern, f"{quot}{new_uri}{quot}")	1✔
175	if line != line2:	1✔
176	logger.debug(doc.pos.abspath)	1✔
177	logger.debug(" " + line)	1✔
178	logger.debug(" => " + line2)	1✔
179	source_lines[lineno] = line2	1✔
180	assert found	1✔
181
182	return source_lines	1✔
183
184
185	def add_additional_files(	1✔
186	zip_dir: str,
187	additional_files: List[str],
188	zip_paths: Dict[str, str],
189	wdls: Dict[str, Tree.Document],
190	logger: logging.Logger,
191	) -> None:
192	"""
193	Add files/directories to an archive staging directory, preserving paths relative to WDL source
194	directories represented in ``zip_paths``.
195
196	Examples:
197
198	* ``/proj/data/ref.fa`` -> ``data/ref.fa`` (when ``/proj/main.wdl`` is archived as
199	``main.wdl``)
200	* ``/shared/lib/data/ref.fa`` -> ``__outside_wdl/lib/data/ref.fa`` (when
201	``/shared/lib/tasks.wdl`` is rewritten under ``__outside_wdl/lib/tasks.wdl``)
202	"""
203	logger.debug(f"Additional files: {additional_files}")	1✔
204	source_dirs = _additional_source_dirs(zip_paths, wdls)	1✔
205	if not source_dirs:	1✔
206	raise Error.InputError("Additional files require local WDL source files")	1✔
207
208	copied: Dict[str, str] = {}	1✔
209	for pattern in additional_files:	1✔
210	# Leave non-glob paths literal so we can report "not found" rather than "matched nothing".
211	matches = glob.glob(pattern, recursive=True) if glob.has_magic(pattern) else [pattern]	1✔
212	matches = sorted(matches)	1✔
213	if not matches:	1✔
214	raise Error.InputError("Additional file pattern matched nothing: " + pattern)	1✔
215	for match in matches:	1✔
216	if not os.path.lexists(match):	1✔
217	raise Error.InputError("Additional file not found: " + match)	1✔
218	for src, dest in _additional_files_from_path(match, source_dirs):	1✔
219	if dest in copied:	1✔
220	# Overlapping globs may name the same source file twice; that's harmless.
221	# A different realpath landing at the same archive path is a real collision.
222	if copied[dest] == os.path.realpath(src):	1✔
223	continue	1✔
224	raise Error.InputError("Additional file overwrites existing path: " + dest)	1✔
225	dest_path = os.path.join(zip_dir, dest)	1✔
226	if os.path.exists(dest_path) or os.path.lexists(dest_path):	1✔
227	raise Error.InputError("Additional file overwrites existing path: " + dest)	1✔
228	if not path_really_within(dest_path, zip_dir):	1✔
229	raise Error.InputError("Invalid additional file destination: " + dest)	1✔
230	os.makedirs(os.path.dirname(dest_path), exist_ok=True)	1✔
231	shutil.copyfile(src, dest_path)	1✔
232	copied[dest] = os.path.realpath(src)	1✔
233	logger.info(f"{dest} <= {src}")	1✔
234
235
236	def _additional_source_dirs(	1✔
237	zip_paths: Dict[str, str], wdls: Dict[str, Tree.Document]
238	) -> List[Tuple[str, str]]:
239	"""
240	Pair each local WDL source directory with its corresponding archive directory.
241
242	Additional files are accepted only when they resolve under one of these directories. Sorting by
243	longest source directory first makes nested imports win over their parents, which is needed to
244	preserve the source-relative meaning of WDL 1.2 paths after import rewrites.
245
246	Examples:
247
248	* ``/proj/tasks/t.wdl`` -> ``tasks`` (nested local import directory wins over ``/proj``)
249	* ``/proj/main.wdl`` -> ``""`` (top-level source directory maps to the archive root)
250	* ``/shared/lib/t.wdl`` -> ``__outside_wdl/lib`` (outside import keeps its rewritten prefix)
251	"""
252	ans: Dict[str, str] = {}	1✔
253	for abspath, zip_path in zip_paths.items():	1✔
254	if abspath in wdls:	1✔
255	source_dir = wdls[abspath].source_dir.rstrip("/")	1✔
256	if not source_dir or not os.path.isfile(abspath):	1✔
257	continue	1✔
258	archive_dir = os.path.dirname(zip_path)	1✔
259	# Different archive paths for one source directory would make it impossible to know
260	# where non-WDL neighbors from that directory should go.
261	if source_dir in ans and ans[source_dir] != archive_dir:	1✔
262	raise Error.InputError(	1✔
263	"Cannot place additional files relative to ambiguous WDL source directory: "
264	+ source_dir
265	)
266	ans[source_dir] = archive_dir	1✔
267	return sorted(ans.items(), key=lambda item: len(item[0]), reverse=True)	1✔
268
269
270	def _additional_files_from_path(	1✔
271	path: str, source_dirs: List[Tuple[str, str]]
272	) -> Iterator[Tuple[str, str]]:
273	"""
274	Expand one matched additional path into concrete file/archive-path pairs.
275
276	Directories are walked recursively, following symlinks only after each directory entry passes
277	``_additional_dest()``. ``visited_dirs`` prevents symlink loops from cycling indefinitely while
278	still allowing safe symlinks to contribute the file contents they point to.
279
280	Examples:
281
282	* ``/proj/data/a.txt`` -> ``data/a.txt`` (file found while walking ``/proj/data``)
283	* ``/proj/data/sub/b.txt`` -> ``data/sub/b.txt`` (recursive directory contents are kept)
284	* ``/proj/data/latest.txt`` -> ``data/latest.txt`` (safe symlink content is copied)
285	"""
286	path = os.path.abspath(path)	1✔
287	_additional_dest(path, source_dirs) # validate the requested path, including symlink target	1✔
288	if os.path.isdir(path):	1✔
289	visited_dirs: Set[str] = set()	1✔
290	for dirpath, dirnames, filenames in os.walk(path, followlinks=True):	1✔
291	# Re-check every walked directory because os.walk follows symlinks when asked.
292	_additional_dest(dirpath, source_dirs)	1✔
293	real_dirpath = os.path.realpath(dirpath)	1✔
294	if real_dirpath in visited_dirs:	1✔
295	dirnames[:] = []	1✔
296	continue	1✔
297	visited_dirs.add(real_dirpath)	1✔
298	for dirname in list(dirnames):	1✔
299	dirname_path = os.path.join(dirpath, dirname)	1✔
300	try:	1✔
301	# Reject a directory symlink before os.walk descends into it.
302	_additional_dest(dirname_path, source_dirs)	1✔
303	except Error.InputError:	1✔
304	raise Error.InputError(	1✔
305	"Additional directory contains unsafe symlink or path: " + dirname_path
306	) from None
307	if os.path.realpath(dirname_path) in visited_dirs:	1✔
308	# Avoid loops like data/back -> .., even when the target is otherwise safe.
309	dirnames.remove(dirname)	1✔
310	for filename in sorted(filenames):	1✔
311	filename_path = os.path.join(dirpath, filename)	1✔
312	yield filename_path, _additional_dest(filename_path, source_dirs)	1✔
313	else:
314	yield path, _additional_dest(path, source_dirs)	1✔
315
316
317	def _additional_dest(path: str, source_dirs: List[Tuple[str, str]]) -> str:	1✔
318	"""
319	Compute the archive path for one additional file or directory after safety checks.
320
321	The path and its realpath target must both remain inside the selected source directory. This
322	permits ordinary in-tree symlinks while rejecting symlinks that would package files from outside
323	the WDL source tree.
324
325	Examples:
326
327	* ``/proj/data/a.txt`` -> ``data/a.txt`` (source dir ``/proj`` maps to archive root)
328	* ``/shared/lib/data/a.txt`` -> ``__outside_wdl/lib/data/a.txt`` (source dir
329	``/shared/lib`` maps to ``__outside_wdl/lib``)
330	* ``/proj/data/secret -> /etc/passwd`` -> rejected (symlink target escapes ``/proj``)
331	"""
332	if not (os.path.isfile(path) or os.path.isdir(path)):	1✔
333	raise Error.InputError("Additional path is neither a file nor a directory: " + path)	1✔
334	# Resolve symlinks in the parent directory (e.g. macOS /tmp -> /private/tmp) so the containment
335	# check lines up with the realpath-canonicalized source directories, while keeping the leaf name
336	# so safe in-tree symlinks retain their own archive path.
337	real_path = os.path.join(os.path.realpath(os.path.dirname(path)), os.path.basename(path))	1✔
338	for source_dir, archive_dir in source_dirs:	1✔
339	relpath = os.path.relpath(real_path, os.path.realpath(source_dir))	1✔
340	if relpath.startswith(".." + os.sep) or relpath == "..":	1✔
UNCOV 341	continue	×
342	# path_really_within additionally rejects symlinks whose target escapes the source tree.
343	if path_really_within(path, source_dir):	1✔
344	return os.path.normpath(os.path.join(archive_dir, relpath))	1✔
345	raise Error.InputError("Additional path must reside within a WDL source directory: " + path)	1✔
346
347
348	def create_reproducible_archive(zip_dir: str, output_path: str, format: str):	1✔
349	# write zip/tar archive with internal filenames lexicographically-ordered and all timestamps
350	# set to an arbitrary constant
351	src_dest_list = [	1✔
352	(path, path.relative_to(zip_dir))
353	for path in pathlib.Path(zip_dir).glob("*/") # Finds all files recursively
354	if path.is_file()
355	or path.is_symlink() # Symlinks will be included in the zip as normal files
356	]
357	# Sort paths by destination
358	src_dest_list.sort(key=lambda x: x[1])	1✔
359	if format == "zip":	1✔
360	_write_no_mtime_zip(output_path, src_dest_list)	1✔
361	elif format == "tar":	1✔
362	_write_no_mtime_tar(output_path, src_dest_list)	1✔
363	else:
364	raise ValueError(f"Unknown format: {format}")	1✔
365	return output_path	1✔
366
367
368	def _write_no_mtime_zip(zip_archive: str, src_dest_list: List[Tuple[pathlib.Path, pathlib.Path]]):	1✔
369	with zipfile.ZipFile(zip_archive, "w") as archive:	1✔
370	for src, dest in src_dest_list:	1✔
371	# This always sets the mod time at 1980-1-1
372	dest_info = zipfile.ZipInfo(str(dest))	1✔
373	with archive.open(dest_info, "w") as archive_file:	1✔
374	with open(src, "rb") as in_file:	1✔
UNCOV 375	while True:	×
376	block = in_file.read(io.DEFAULT_BUFFER_SIZE)	1✔
377	if not block:	1✔
378	break	1✔
379	archive_file.write(block)	1✔
380
381
382	def _write_no_mtime_tar(tar_archive: str, src_dest_list: List[Tuple[pathlib.Path, pathlib.Path]]):	1✔
383	with tarfile.TarFile(tar_archive, "w") as archive:	1✔
384	for src, dest in src_dest_list:	1✔
385	dest_info = tarfile.TarInfo(str(dest)) # Mtime by default at 0	1✔
386	dest_info.size = os.stat(src).st_size	1✔
387	with open(src, "rb") as in_file:	1✔
388	archive.addfile(dest_info, in_file)	1✔
389
390
391	UnpackedZip = NamedTuple(	1✔
392	"UnpackedZip", [("dir", str), ("main_wdl", str), ("input_file", Optional[str])]
393	)
UNCOV 394	"""	×
395	Contextual value of `WDL.Zip.unpack()`: absolute paths of source directory, main WDL, and default
396	input JSON file (if any). The source directory prefixes the latter paths.
397	"""
398
399
400	@contextlib.contextmanager	1✔
401	def unpack(archive_fn: str, tempdir_parent: Optional[str] = None) -> Iterator[UnpackedZip]:	1✔
402	"""
403	Open a context with the WDL source archive unpacked into a temp directory, yielding
404	`UnpackedZip`. The temp directory will be deleted on context exit.
405
406	A path to the MANIFEST.json of an already-unpacked source archive may also be used, or a
407	directory containing one. In this case, it is NOT deleted on context exit. ::
408
409	with WDL.Zip.unpack("/path/to/source.zip") as unpacked:
410	doc = WDL.load(unpacked.main_wdl)
411	...
412	"""
413	with contextlib.ExitStack() as cleanup:	1✔
414	# extract zip if needed (also allowing use of already-extracted manifest/dir)
415	if os.path.isdir(archive_fn):	1✔
416	archive_fn = os.path.join(archive_fn, "MANIFEST.json")	1✔
417	if os.path.basename(archive_fn) == "MANIFEST.json":	1✔
418	manifest_fn = archive_fn	1✔
419	else:
420	try:	1✔
421	dn = cleanup.enter_context(	1✔
422	tempfile.TemporaryDirectory(prefix="miniwdl_run_zip_", dir=tempdir_parent)
423	)
424	except OSError as exn:	1✔
425	msg = "Unable to create temporary directory for unpacking source archive"	1✔
426	if tempdir_parent:	1✔
427	msg += (	1✔
428	" under "
429	+ tempdir_parent
430	+ "; set TMPDIR to a writable directory under file_io.root"
431	)
432	raise Error.InputError(msg) from exn	1✔
433	try:	1✔
434	shutil.unpack_archive(archive_fn, dn)	1✔
435	except Exception:	1✔
436	raise Error.InputError("Unreadable source archive " + archive_fn)	1✔
437	manifest_fn = os.path.join(dn, "MANIFEST.json")	1✔
438
439	try:	1✔
440	with open(manifest_fn) as infile:	1✔
441	manifest = json.load(infile)	1✔
442	assert isinstance(manifest, dict) and isinstance(	1✔
443	manifest.get("mainWorkflowURL", None), str
444	)
445	except Exception:	1✔
446	raise Error.InputError("Missing or invalid MANIFEST.json in " + archive_fn)	1✔
447
448	dn = os.path.abspath(os.path.dirname(manifest_fn))	1✔
449	main_wdl = manifest["mainWorkflowURL"]	1✔
450
451	input_file = None	1✔
452	if (	1✔
453	isinstance(manifest.get("inputFileURLs", None), list)
454	and manifest["inputFileURLs"]
455	and isinstance(manifest["inputFileURLs"][0], str)
456	):
457	input_file = manifest["inputFileURLs"][0]	1✔
458
459	# sanity check
460	main_wdl_abs = os.path.join(dn, main_wdl)	1✔
461	input_file_abs = os.path.join(dn, input_file) if input_file else None	1✔
462	if not (os.path.isfile(main_wdl_abs) and path_really_within(main_wdl_abs, dn)) or (	1✔
463	input_file_abs
464	and not (os.path.isfile(input_file_abs) and path_really_within(input_file_abs, dn))
465	):
466	raise Error.InputError(	1✔
467	"MANIFEST.json refers to missing or invalid files in " + archive_fn
468	)
469
470	yield UnpackedZip(dn, main_wdl_abs, input_file_abs)	1✔

chanzuckerberg / miniwdl / 28270112145

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous