Normalize linker path inputs to prevent absolute-path overrides

linker.py

@@ -25,6 +25,10 @@ def chmod_and_chown(path, *, uid=None, gid=None, mode=0o775):
     #if uid is not None and gid is not None:
     #    os.chown(path, uid, gid)
 
+def make_relative_path(path_value):
+    path = Path(path_value)
+    return Path(*path.parts[1:]) if path.is_absolute() else path
+
 @task(retries=2, retry_delay_seconds=10)
 def create_symlinks(ref, api_key=None, dry_run=False):
     """
@@ -56,13 +60,14 @@ def create_symlinks(ref, api_key=None, dry_run=False):
             if path_expr_alias := doc.get("experiment_alias_directory"):
                 path_proposal = Path(f"/nsls2/data/cms/proposals/{doc['cycle']}/{doc['data_session']}")
                 # stats = path_proposal.stat()
-                path_expr = path_proposal / "experiments"   # experiments directory
+                experiments_dir = make_relative_path(doc.get("experiments_directory", "experiments"))
+                path_expr = path_proposal / experiments_dir
                 if dry_run:
                     logger.info(f"Dry run: mkdir {path_expr}")
                 else:
                     path_expr.mkdir(exist_ok=True, parents=True)
                 #chmod_and_chown(path_expr, uid=stats.st_uid, gid=stats.st_gid)
-                path_expr_alias = path_expr / path_expr_alias
+                path_expr_alias = path_expr / make_relative_path(path_expr_alias)
                 if dry_run:
                     logger.info(f"Dry run: mkdir {path_expr_alias}")
                 else:

test_linker_paths.py

@@ -0,0 +1,52 @@
+import sys
+import types
+import unittest
+from pathlib import Path
+
+
+prefect_stub = types.ModuleType("prefect")
+
+def task(*args, **kwargs):
+    def decorator(func):
+        return func
+
+    return decorator
+
+
+def get_run_logger():
+    return None
+
+
+prefect_stub.task = task
+prefect_stub.get_run_logger = get_run_logger
+sys.modules.setdefault("prefect", prefect_stub)
+
+
+data_validation_stub = types.ModuleType("data_validation")
+data_validation_stub.get_run = lambda *args, **kwargs: None
+sys.modules.setdefault("data_validation", data_validation_stub)
+
+
+import linker  # noqa: E402
+
+
+class MakeRelativePathTests(unittest.TestCase):
+    def test_relative_path_is_unchanged(self):
+        self.assertEqual(linker.make_relative_path("experiments"), Path("experiments"))
+
+    def test_absolute_path_becomes_relative(self):
+        self.assertEqual(linker.make_relative_path("/experiments"), Path("experiments"))
+
+    def test_root_path_becomes_current_directory(self):
+        self.assertEqual(linker.make_relative_path("/"), Path("."))
+
+    def test_joined_path_never_overrides_proposal_path(self):
+        proposal = Path("/nsls2/data/cms/proposals/2026-1/pass-12345")
+        self.assertEqual(
+            proposal / linker.make_relative_path("/experiments"),
+            proposal / "experiments",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()