Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@pytest.mark.skipif(onWindows(), reason="udocker is Linux/macOS only") # type: ignore
@pytest.mark.parametrize("factor", test_factors) # type: ignore
def test_bad_userspace_runtime(factor: str) -> None:
test_file = "tests/wf/wc-tool.cwl"
job_file = "tests/wf/wc-job.json"
commands = factor.split()
commands.extend(
[
"--user-space-docker-cmd=quaquioN",
"--default-container=debian",
get_data(test_file),
get_data(job_file),
]
)
error_code, stdout, stderr = get_main_output(commands)
assert "or quaquioN is missing or broken" in stderr, stderr
assert error_code == 1
[ "${trueExe#-}" = "$trueExe" ] || trueExe=${trueExe#-}
[ "${trueExe#/}" != "$trueExe" ] || trueExe=$([ -n "$ZSH_VERSION" ] && which -p "$trueExe" || which "$trueExe")
while nextTarget=$(readlink "$trueExe"); do trueExe=$nextTarget; done
printf '%s\n' "$(basename "$trueExe")"
} ; getTrueShellExeName""",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=None,
)
sh_name_b, sh_name_err = process.communicate()
sh_name = sh_name_b.decode("utf-8").strip()
assert "completed success" in stderr, (error_code, stdout, stderr)
assert error_code == 0, (error_code, stdout, stderr)
if onWindows():
target = 5
elif sh_name == "dash":
target = 4
else: # bash adds "SHLVL" and "_" environment variables
target = 6
result = json.loads(stdout)["env_count"]
details = ""
if result != target:
_, details, _ = get_main_output(["--quiet", get_data("tests/env2.cwl")])
print(sh_name)
print(sh_name_err)
print(details)
assert result == target, (error_code, sh_name, sh_name_err, details, stdout, stderr)
if "CWLTOOL_OPTIONS" in os.environ:
addl = os.environ["CWLTOOL_OPTIONS"].split(" ")
args = arg_parser().parse_args(addl + argsl)
if args.record_container_id:
if not args.cidfile_dir:
args.cidfile_dir = os.getcwd()
del args.record_container_id
if runtimeContext is None:
runtimeContext = RuntimeContext(vars(args))
else:
runtimeContext = runtimeContext.copy()
# If on Windows platform, a default Docker Container is used if not
# explicitely provided by user
if onWindows() and not runtimeContext.default_container:
# This docker image is a minimal alpine image with bash installed
# (size 6 mb). source: https://github.com/frol/docker-alpine-bash
runtimeContext.default_container = windows_default_container_id
# If caller parsed its own arguments, it may not include every
# cwltool option, so fill in defaults to avoid crashing when
# dereferencing them in args.
for key, val in get_default_args().items():
if not hasattr(args, key):
setattr(args, key, val)
configure_logging(args, stderr_handler, runtimeContext)
if args.version:
print(versionfunc())
return 0
def _whoami() -> Tuple[str, str]:
"""Return the current operating system account as (username, fullname)."""
username = getuser()
try:
if onWindows():
get_user_name = ctypes.windll.secur32.GetUserNameExW # type: ignore
size = ctypes.pointer(ctypes.c_ulong(0))
get_user_name(3, None, size)
name_buffer = ctypes.create_unicode_buffer(size.contents.value)
get_user_name(3, name_buffer, size)
fullname = str(name_buffer.value)
else:
fullname = pwd.getpwuid(os.getuid())[4].split(",")[0]
except (KeyError, IndexError):
fullname = username
return (username, fullname)
_logger.warning(
"Running with support for javascript console in expressions (DO NOT USE IN PRODUCTION)"
)
elif context is not None:
js_engine = "cwlNodeEngineWithContext.js"
else:
js_engine = "cwlNodeEngine.js"
created_new_process = False
if context is not None:
nodejs = localdata.procs.get((js_engine, context))
else:
nodejs = localdata.procs.get(js_engine)
if nodejs is None or nodejs.poll() is not None or onWindows():
res = resource_stream(__name__, js_engine)
js_engine_code = res.read().decode("utf-8")
created_new_process = True
new_proc = new_js_proc(js_engine_code, force_docker_pull=force_docker_pull)
if context is None:
localdata.procs[js_engine] = new_proc
nodejs = new_proc
else:
localdata.procs[(js_engine, context)] = new_proc
nodejs = new_proc
killed = []
if not vol.target.startswith(container_outdir):
# this is an input file written outside of the working
# directory, so therefor ineligable for being an output file.
# Thus, none of our business
continue
host_outdir_tgt = os.path.join(
host_outdir, vol.target[len(container_outdir) + 1 :]
)
if os.path.islink(host_outdir_tgt) or os.path.isfile(host_outdir_tgt):
try:
os.remove(host_outdir_tgt)
except PermissionError:
pass
elif os.path.isdir(host_outdir_tgt) and not vol.resolved.startswith("_:"):
shutil.rmtree(host_outdir_tgt)
if onWindows():
# If this becomes a big issue for someone then we could
# refactor the code to process output from a running container
# and avoid all the extra IO below
if vol.type in ("File", "WritableFile"):
shutil.copy(vol.resolved, host_outdir_tgt)
elif vol.type in ("Directory", "WritableDirectory"):
copytree_with_merge(vol.resolved, host_outdir_tgt)
elif not vol.resolved.startswith("_:"):
try:
os.symlink(vol.resolved, host_outdir_tgt)
except FileExistsError:
pass
def docker_compatible_realpath(self, path): # type: (str) -> str
if onWindows():
if path.startswith("/"):
return path
return "/" + path
return self.realpath(path)
if not user_space_docker_cmd:
if not runtimeContext.no_read_only:
runtime.append("--read-only=true")
if self.networkaccess:
if runtimeContext.custom_net:
runtime.append("--net={0}".format(runtimeContext.custom_net))
else:
runtime.append("--net=none")
if self.stdout is not None:
runtime.append("--log-driver=none")
euid, egid = docker_vm_id()
if not onWindows():
# MS Windows does not have getuid() or geteuid() functions
euid, egid = euid or os.geteuid(), egid or os.getgid()
if runtimeContext.no_match_user is False and (
euid is not None and egid is not None
):
runtime.append("--user=%d:%d" % (euid, egid))
if runtimeContext.rm_container:
runtime.append("--rm")
runtime.append("--env=TMPDIR=/tmp")
# spec currently says "HOME must be set to the designated output
# directory." but spec might change to designated temp directory.
# runtime.append("--env=HOME=/tmp")
if default_container is not None:
dockerReq = {
"class": "DockerRequirement",
"dockerPull": default_container,
}
if mpiRequired:
self.hints.insert(0, dockerReq)
dockerRequired = False
else:
self.requirements.insert(0, dockerReq)
dockerRequired = True
if (
default_container == windows_default_container_id
and runtimeContext.use_container
and onWindows()
):
_logger.warning(
DEFAULT_CONTAINER_MSG,
windows_default_container_id,
windows_default_container_id,
)
if dockerReq is not None and runtimeContext.use_container:
if mpiReq is not None:
_logger.warning("MPIRequirement with containers is a beta feature")
if runtimeContext.singularity:
return SingularityCommandLineJob
elif runtimeContext.user_space_docker_cmd:
return UDockerCommandLineJob
if mpiReq is not None:
if mpiRequired:
vars_to_preserve = runtimeContext.preserve_environment
if runtimeContext.preserve_entire_environment is not False:
vars_to_preserve = os.environ
if vars_to_preserve:
for key, value in os.environ.items():
if key in vars_to_preserve and key not in env:
# On Windows, subprocess env can't handle unicode.
env[key] = str(value) if onWindows() else value
env["HOME"] = str(self.outdir) if onWindows() else self.outdir
env["TMPDIR"] = str(self.tmpdir) if onWindows() else self.tmpdir
if "PATH" not in env:
env["PATH"] = str(os.environ["PATH"]) if onWindows() else os.environ["PATH"]
if "SYSTEMROOT" not in env and "SYSTEMROOT" in os.environ:
env["SYSTEMROOT"] = (
str(os.environ["SYSTEMROOT"])
if onWindows()
else os.environ["SYSTEMROOT"]
)
stage_files(
self.pathmapper,
ignore_writable=True,
symlink=True,
secret_store=runtimeContext.secret_store,
)
if self.generatemapper is not None:
stage_files(
self.generatemapper,
ignore_writable=self.inplace_update,
symlink=True,
secret_store=runtimeContext.secret_store,
)