CVE-2024-56142

@@ -511,6 +511,12 @@ def _validate_target_path(pg_data_directory: str, target_path: str) -> None:
         if not xlog_file.parent.is_dir():
             raise HttpResponse(f"Invalid xlog file path {target_path}, parent directory should exist", status=409)
 
+    @staticmethod
+    def _is_valid_xlog_path(xlog_path_str: str) -> bool:
+        xlog_path = Path(xlog_path_str)
+        return xlog_path.is_file() and not xlog_path.is_symlink()
+
+
     def get_wal_or_timeline_file(self, site: str, filename: str, filetype: str) -> None:
         target_path = self.headers.get("x-pghoard-target-path")
         if not target_path:
@@ -614,8 +620,8 @@ def handle_archival_request(self, site, filename, filetype):
         xlog_dir = get_pg_wal_directory(site_config)
         xlog_path = os.path.join(xlog_dir, filename)
         self.server.log.debug("Got request to archive: %r %r %r, %r", site, filetype, filename, xlog_path)
-        if not os.path.exists(xlog_path):
-            self.server.log.debug("xlog_path: %r did not exist, cannot archive, returning 404", xlog_path)
+        if not self._is_valid_xlog_path(xlog_path):
+            self.server.log.debug("xlog_path: %r did not exist or contains symlinks, cannot archive, returning 404", xlog_path)
             raise HttpResponse("N/A", status=404)
 
         if filetype == "xlog":

@@ -467,6 +467,17 @@ def test_get_invalid(self, pghoard, tmpdir):
         status = conn.getresponse().status
         assert status == 409
 
+    def test_put_invalid_timeline_fails(self, pghoard, tmpdir):
+        wal_dir = get_pg_wal_directory(pghoard.config["backup_sites"][pghoard.test_site])
+        symlink_timeline = os.path.join(str(wal_dir), "00000001.history")
+        secret_file = os.path.join(str(tmpdir), "config.json")
+        os.symlink(secret_file, symlink_timeline)
+        symlink_timeline_request = "/{}/timeline/00000001.history".format(pghoard.test_site)
+        conn = HTTPConnection(host="127.0.0.1", port=pghoard.config["http_port"])
+        conn.request("PUT", symlink_timeline_request)
+        status = conn.getresponse().status
+        assert status == 404
+
     def test_get_invalid_retry(self, pghoard_no_mp):
         # inject a failure by making a static function fail
         failures = [0, ""]