ExternalVendorCode/onefuzz
1
0
Fork 0
mirror of https://github.com/microsoft/onefuzz.git synced 2025-06-17 20:38:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add debug command to download a task’s containers (#2359)

Browse Source
This commit is contained in:
George Pollard
2022-09-09 12:45:15 +12:00
committed by GitHub
parent 511eafca73
commit ef7b557599
2 changed files with 45 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
src/cli/onefuzz/api.py
View File

@ -460,6 +460,44 @@ class Containers(Endpoint):
self.logger.debug("list containers")
return self._req_model_list("GET", responses.ContainerInfoBase)
def download_job(
self, job_id: UUID_EXPANSION, *, output: Optional[primitives.Directory] = None
) -> None:
tasks = self.onefuzz.tasks.list(job_id=job_id, state=None)
if not tasks:
raise Exception("no tasks with job_id:%s" % job_id)
self._download_tasks(tasks, output)
def download_task(
self, task_id: UUID_EXPANSION, *, output: Optional[primitives.Directory] = None
) -> None:
self._download_tasks([self.onefuzz.tasks.get(task_id=task_id)], output)
def _download_tasks(
self, tasks: List[models.Task], output: Optional[primitives.Directory]
) -> None:
to_download: Dict[str, str] = {}
for task in tasks:
for container in task.config.containers:
info = self.onefuzz.containers.get(container.name)
name = os.path.join(container.type.name, container.name)
to_download[name] = info.sas_url
if output is None:
output = primitives.Directory(os.getcwd())
for name in to_download:
outdir = os.path.join(output, name)
if not os.path.exists(outdir):
os.makedirs(outdir)
self.logger.info("downloading: %s", name)
# security note: the src for azcopy comes from the server which is
# trusted in this context, while the destination is provided by the
# user
azcopy_sync(to_download[name], outdir)
class Repro(Endpoint):
"""Interact with Reproduction VMs"""
@ -1005,33 +1043,6 @@ class JobContainers(Endpoint):
results[container] = self.onefuzz.containers.files.list(container).files
return results
def download(
self, job_id: UUID_EXPANSION, *, output: Optional[primitives.Directory] = None
) -> None:
to_download = {}
tasks = self.onefuzz.tasks.list(job_id=job_id, state=None)
if not tasks:
raise Exception("no tasks with job_id:%s" % job_id)
for task in tasks:
for container in task.config.containers:
info = self.onefuzz.containers.get(container.name)
name = os.path.join(container.type.name, container.name)
to_download[name] = info.sas_url
if output is None:
output = primitives.Directory(os.getcwd())
for name in to_download:
outdir = os.path.join(output, name)
if not os.path.exists(outdir):
os.makedirs(outdir)
self.logger.info("downloading: %s", name)
# security note: the src for azcopy comes from the server which is
# trusted in this context, while the destination is provided by the
# user
azcopy_sync(to_download[name], outdir)
def delete(
self,
job_id: UUID_EXPANSION,