[CERTTF-303] Introduce support for file attachments

agent/testflinger_agent/agent.py

@@ -15,14 +15,39 @@
 import json
 import logging
 import os
+from pathlib import Path
 import shutil
+import tarfile
+import tempfile
 
 from testflinger_agent.job import TestflingerJob
 from testflinger_agent.errors import TFServerError
+from testflinger_agent.config import ATTACHMENTS_DIR
 
 logger = logging.getLogger(__name__)
 
 
+def secure_filter(member, path):
+    """Combine the `data` filter with custom attachment filtering
+
+    Makes sure that the starting folder for all attachments coincides
+    with one of the supported phases, i.e. that the attachment archive
+    has been created properly and no attachment will be extracted to an
+    unexpected location.
+    """
+    try:
+        resolved = Path(member.name).resolve().relative_to(Path.cwd())
+    except ValueError as error:
+        # essentially trying to extract higher than the attachments folder
+        raise tarfile.OutsideDestinationError(member, path) from error
+    if not str(resolved).startswith(
+        ("provision/", "firmware_update/", "test/")
+    ):
+        # trying to extract in an invalid folder, under the attachments folder
+        raise tarfile.OutsideDestinationError(member, path)
+    return tarfile.data_filter(member, path)
+
+
 class TestflingerAgent:
     def __init__(self, client):
         self.client = client
@@ -111,6 +136,18 @@ def mark_device_offline(self):
         # Create the offline file, this should work even if it exists
         open(self.get_offline_files()[0], "w").close()
 
+    def unpack_attachments(self, job_data: dict, cwd: Path):
+        """Download and unpack the attachments associated with a job"""
+        job_id = job_data["job_id"]
+
+        with tempfile.NamedTemporaryFile(suffix="tar.gz") as archive_tmp:
+            archive_path = Path(archive_tmp.name)
+            # download attachment archive
+            self.client.get_attachments(job_id, path=archive_path)
+            # extract archive into the attachments folder
+            with tarfile.open(archive_path, "r:gz") as tar:
+                tar.extractall(cwd / ATTACHMENTS_DIR, filter=secure_filter)
+
     def process_jobs(self):
         """Coordinate checking for new jobs and handling them if they exists"""
         TEST_PHASES = [
@@ -136,7 +173,9 @@ def process_jobs(self):
                     self.client.config.get("execution_basedir"), job.job_id
                 )
                 os.makedirs(rundir)
+
                 self.client.post_agent_data({"job_id": job.job_id})
+
                 # Dump the job data to testflinger.json in our execution dir
                 with open(os.path.join(rundir, "testflinger.json"), "w") as f:
                     json.dump(job_data, f)
@@ -146,14 +185,19 @@ def process_jobs(self):
                 ) as f:
                     json.dump({}, f)
 
+                # handle job attachments, if any
+                # (always after creating "testflinger.json", for reporting
+                # in case of an unpacking error)
+                if job_data.get("attachments_status") == "complete":
+                    self.unpack_attachments(job_data, cwd=Path(rundir))
+
                 for phase in TEST_PHASES:
                     # First make sure the job hasn't been cancelled
                     if self.client.check_job_state(job.job_id) == "cancelled":
                         logger.info("Job cancellation was requested, exiting.")
                         break
                     self.client.post_job_state(job.job_id, phase)
                     self.set_agent_state(phase)
-
                     exitcode = job.run_test_phase(phase, rundir)
 
                     self.client.post_influx(phase, exitcode)

agent/testflinger_agent/client.py

@@ -15,6 +15,7 @@
 import logging
 import json
 import os
+from pathlib import Path
 import requests
 import shutil
 import tempfile
@@ -105,6 +106,26 @@ def check_jobs(self):
             # Wait a little extra before trying again
             time.sleep(60)
 
+    def get_attachments(self, job_id: str, path: Path):
+        """Download the attachment archive associated with a job
+
+        :param job_id:
+            Id for the job
+        :param path:
+            Where to save the attachment archive
+        """
+        uri = urljoin(self.server, f"/v1/job/{job_id}/attachments")
+        with requests.get(uri, stream=True, timeout=600) as response:
+            if response.status_code != 200:
+                logger.error(
+                    f"Unable to retrieve attachments for job {job_id} "
+                    f"(error: {response.status_code})"
+                )
+                raise TFServerError(response.status_code)
+            with open(path, "wb") as attachments:
+                for chunk in response.iter_content(chunk_size=4096):
+                    attachments.write(chunk)
+
     def check_job_state(self, job_id):
         job_data = self.get_result(job_id)
         if job_data:

agent/testflinger_agent/config.py

@@ -0,0 +1,17 @@
+# Copyright (C) 2024 Canonical
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Configuration constants for the Testflinger agent"""
+
+ATTACHMENTS_DIR = "attachments"