buildbot_nix/__init__.py

@@ -310,18 +310,55 @@ class UpdateBuildOutput(steps.BuildStep):
         self.project = project
         self.path = path
 
+    def join_traversalsafe(self, root: Path, joined: Path) -> Path:
+        root = root.resolve()
+
+        for part in joined.parts:
+            new_root = (root / part).resolve()
+
+            if not new_root.is_relative_to(root):
+                msg = f"Joined path attempted to traverse upwards when processing {root} against {part} (gave {new_root})"
+                raise ValueError(msg)
+
+            root = new_root
+
+        return root
+
+    def join_all_traversalsafe(self, root: Path, *paths: Path) -> Path:
+        for path in paths:
+            root = self.join_traversalsafe(root, path)
+
+        return root
+
+    @defer.inlineCallbacks
     def run(self) -> Generator[Any, object, Any]:
         props = self.build.getProperties()
-        if props.getProperty("branch") != self.project.default_branch:
+
+        pr = props.getProperty("pr_number")
+
+        if not pr and props.getProperty("branch") != self.project.default_branch:
             return util.SKIPPED
 
-        attr = Path(props.getProperty("attr")).name
+        owner = Path(props.getProperty("owner"))
-        out_path = props.getProperty("out_path")
+        repo = Path(props.getProperty("repository_name"))
-        # XXX don't hardcode this
-        self.path.mkdir(parents=True, exist_ok=True)
-        (self.path / attr).write_text(out_path)
-        return util.SUCCESS
 
+        target = Path(props.getProperty("branch")) if not pr else Path(f"pulls/{pr}")
+
+        attr = Path(props.getProperty("attr"))
+
+        try:
+            file = self.join_all_traversalsafe(self.path, owner, repo, target, attr)
+        except ValueError as e:
+            error_log: StreamLog = yield self.addLog("path_error")
+            error_log.addStderr(f"Path traversal prevented ... skipping update: {e}")
+            return util.FAILURE
+
+        file.parent.mkdir(parents=True, exist_ok=True)
+
+        out_path = props.getProperty("out_path")
+        file.write_text(out_path)
+
+        return util.SUCCESS
 
 # GitHub somtimes fires the PR webhook before it has computed the merge commit
 # This is a workaround to fetch the merge commit and checkout the PR branch in CI
@@ -563,6 +600,7 @@ def nix_build_config(
 def nix_skipped_build_config(
     project: GitProject,
     worker_names: list[str],
+    outputs_path: Path | None = None,
 ) -> BuilderConfig:
     """Dummy builder that is triggered when a build is skipped."""
     factory = util.BuildFactory()
@@ -601,6 +639,14 @@ def nix_skipped_build_config(
             set_properties={"report_status": False},
         ),
     )
+    if outputs_path is not None:
+        factory.addStep(
+            UpdateBuildOutput(
+                project=project,
+                name="Update build output",
+                path=outputs_path,
+            ),
+        )
     return util.BuilderConfig(
         name=f"{project.name}/nix-skipped-build",
         project=project.name,
@@ -759,7 +805,7 @@ def config_for_project(
                 retries=build_retries,
                 post_build_steps=post_build_steps,
             ),
-            nix_skipped_build_config(project, [SKIPPED_BUILDER_NAME]),
+            nix_skipped_build_config(project, [SKIPPED_BUILDER_NAME], outputs_path),
             nix_register_gcroot_config(project, worker_names),
             nix_build_combined_config(project, worker_names),
         ],