Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def tostr(self, value: Union[MutableMapping[str, str], Any]) -> str:
if isinstance(value, MutableMapping) and value.get("class") in (
"File",
"Directory",
):
if "path" not in value:
raise WorkflowException(
u'%s object missing "path": %s' % (value["class"], value)
)
# Path adjust for windows file path when passing to docker, docker accepts unix like path only
(docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
if onWindows() and docker_req is not None:
# docker_req is none only when there is no dockerRequirement
# mentioned in hints and Requirement
path = docker_windows_path_adjust(value["path"])
return path
return value["path"]
else:
return str(value)
runtime,
any_path_okay=True,
secret_store=runtime_context.secret_store,
tmpdir_prefix=runtime_context.tmpdir_prefix,
)
if self.generatemapper is not None:
self.add_volumes(
self.generatemapper,
runtime,
any_path_okay=any_path_okay,
secret_store=runtime_context.secret_store,
tmpdir_prefix=runtime_context.tmpdir_prefix,
)
runtime.append("--pwd")
runtime.append("%s" % (docker_windows_path_adjust(self.builder.outdir)))
if runtime_context.custom_net:
raise UnsupportedRequirement(
"Singularity implementation does not support custom networking"
)
elif runtime_context.disable_net:
runtime.append("--net")
env["SINGULARITYENV_TMPDIR"] = tmpdir
env["SINGULARITYENV_HOME"] = self.builder.outdir
for name, value in self.environment.items():
env["SINGULARITYENV_{}".format(name)] = str(value)
return (runtime, None)
def create_runtime(self, env, rm_container=True, record_container_id=False, cidfile_dir="",
cidfile_prefix="", **kwargs):
# type: (MutableMapping[Text, Text], bool, bool, Text, Text, **Any) -> List
user_space_docker_cmd = kwargs.get("user_space_docker_cmd")
if user_space_docker_cmd:
runtime = [user_space_docker_cmd, u"run"]
else:
runtime = [u"docker", u"run", u"-i"]
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(os.path.realpath(self.outdir)),
self.builder.outdir))
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(os.path.realpath(self.tmpdir)), "/tmp"))
self.add_volumes(self.pathmapper, runtime, secret_store=kwargs.get("secret_store"))
if self.generatemapper:
self.add_volumes(self.generatemapper, runtime, secret_store=kwargs.get("secret_store"))
if user_space_docker_cmd:
runtime = [x.replace(":ro", "") for x in runtime]
runtime = [x.replace(":rw", "") for x in runtime]
runtime.append(u"--workdir=%s" % (
docker_windows_path_adjust(self.builder.outdir)))
if not user_space_docker_cmd:
def append_volume(
runtime: List[str], source: str, target: str, writable: bool = False
) -> None:
runtime.append("--bind")
runtime.append(
"{}:{}:{}".format(
docker_windows_path_adjust(source),
docker_windows_path_adjust(target),
"rw" if writable else "ro",
)
def add_file_or_directory_volume(
self, runtime: List[str], volume: MapperEnt, host_outdir_tgt: Optional[str]
) -> None:
"""Append volume a file/dir mapping to the runtime option list."""
if not volume.resolved.startswith("_:"):
_check_docker_machine_path(docker_windows_path_adjust(volume.resolved))
self.append_volume(runtime, volume.resolved, volume.target)
host_outdir = self.outdir
container_outdir = self.builder.outdir
for src, vol in pathmapper.items():
if not vol.staged:
continue
if vol.target.startswith(container_outdir+"/"):
host_outdir_tgt = os.path.join(
host_outdir, vol.target[len(container_outdir)+1:])
else:
host_outdir_tgt = None
if vol.type in ("File", "Directory"):
if not vol.resolved.startswith("_:"):
runtime.append(u"--volume=%s:%s:ro" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
elif vol.type == "WritableFile":
if self.inplace_update:
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
else:
shutil.copy(vol.resolved, host_outdir_tgt)
ensure_writable(host_outdir_tgt)
elif vol.type == "WritableDirectory":
if vol.resolved.startswith("_:"):
os.makedirs(host_outdir_tgt, 0o0755)
else:
if self.inplace_update:
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
def append_volume(
runtime: List[str], source: str, target: str, writable: bool = False
) -> None:
runtime.append("--bind")
runtime.append(
"{}:{}:{}".format(
docker_windows_path_adjust(source),
docker_windows_path_adjust(target),
"rw" if writable else "ro",
)
requirements: List[CWLObjectType],
outdir: Optional[str],
tmpdir: Optional[str],
resources: Dict[str, Union[float, int]],
context: Optional[CWLOutputType] = None,
timeout: float = default_timeout,
force_docker_pull: bool = False,
debug: bool = False,
js_console: bool = False,
strip_whitespace: bool = True,
cwlVersion: str = "",
) -> Optional[CWLOutputType]:
runtime = cast(MutableMapping[str, Union[int, str, None]], copy.deepcopy(resources))
runtime["tmpdir"] = docker_windows_path_adjust(tmpdir) if tmpdir else None
runtime["outdir"] = docker_windows_path_adjust(outdir) if outdir else None
rootvars = cast(
CWLObjectType,
bytes2str_in_dicts({"inputs": jobinput, "self": context, "runtime": runtime}),
)
if isinstance(ex, str) and needs_parsing(ex):
fullJS = False
jslib = ""
for r in reversed(requirements):
if r["class"] == "InlineJavascriptRequirement":
fullJS = True
jslib = jshead(cast(List[str], r.get("expressionLib", [])), rootvars)
break
try:
elif vol.type == "WritableFile":
if self.inplace_update:
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
else:
shutil.copy(vol.resolved, host_outdir_tgt)
ensure_writable(host_outdir_tgt)
elif vol.type == "WritableDirectory":
if vol.resolved.startswith("_:"):
os.makedirs(host_outdir_tgt, 0o0755)
else:
if self.inplace_update:
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
else:
shutil.copytree(vol.resolved, host_outdir_tgt)
ensure_writable(host_outdir_tgt)
elif vol.type == "CreateFile":
if secret_store:
contents = secret_store.retrieve(vol.resolved)
else:
contents = vol.resolved
if host_outdir_tgt:
with open(host_outdir_tgt, "wb") as f:
f.write(contents.encode("utf-8"))
else:
fd, createtmp = tempfile.mkstemp(dir=self.tmpdir)
with os.fdopen(fd, "wb") as f:
f.write(contents.encode("utf-8"))
runtime.append(u"--volume=%s:%s:rw" % (
if not vol.staged:
continue
if vol.target.startswith(container_outdir+"/"):
host_outdir_tgt = os.path.join(
host_outdir, vol.target[len(container_outdir)+1:])
else:
host_outdir_tgt = None
if vol.type in ("File", "Directory"):
if not vol.resolved.startswith("_:"):
runtime.append(u"--volume=%s:%s:ro" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
elif vol.type == "WritableFile":
if self.inplace_update:
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
else:
shutil.copy(vol.resolved, host_outdir_tgt)
ensure_writable(host_outdir_tgt)
elif vol.type == "WritableDirectory":
if vol.resolved.startswith("_:"):
os.makedirs(host_outdir_tgt, 0o0755)
else:
if self.inplace_update:
runtime.append(u"--volume=%s:%s:rw" % (
docker_windows_path_adjust(vol.resolved),
docker_windows_path_adjust(vol.target)))
else:
shutil.copytree(vol.resolved, host_outdir_tgt)
ensure_writable(host_outdir_tgt)
elif vol.type == "CreateFile":