Add per-user job ownership for cancel authorization

Record the Slack user ID in request.json when submitting transfer jobs
and enforce ownership checks in cancel_job so only the submitting user
can cancel their own jobs. Legacy jobs without a submitted_by field
remain cancellable by anyone for backwards compatibility.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: fluidnumerics-joe

src/xfer/slackbot/app.py

@@ -177,6 +177,7 @@ def handle_mention(event: dict[str, Any], say: Any) -> None:
                 channel_id=channel,
                 thread_ts=thread_ts,
                 conversation_history=history.copy() if history else None,
+                user_id=user,
             )
 
             # Store the exchange
@@ -228,6 +229,7 @@ def handle_message(event: dict[str, Any], say: Any) -> None:
                     channel_id=channel,
                     thread_ts=thread_ts,
                     conversation_history=history.copy() if history else None,
+                    user_id=user,
                 )
 
                 conversations.append(
@@ -291,6 +293,7 @@ def handle_message(event: dict[str, Any], say: Any) -> None:
                     channel_id=channel,
                     thread_ts=thread_ts,
                     conversation_history=history.copy() if history else None,
+                    user_id=user,
                 )
 
                 conversations.append(

src/xfer/slackbot/claude_agent.py

@@ -405,6 +405,7 @@ def execute_tool(
         tool_input: dict[str, Any],
         channel_id: str,
         thread_ts: str,
+        user_id: str = "",
     ) -> str:
         """Execute a tool and return the result as a string."""
         if tool_name == "submit_transfer":
@@ -426,6 +427,7 @@ def execute_tool(
                 time_limit=tool_input.get("time_limit"),
                 job_name=tool_input.get("job_name"),
                 rclone_flags=tool_input.get("rclone_flags"),
+                user_id=user_id,
             )
             return json.dumps(result.__dict__, default=str)
 
@@ -472,6 +474,7 @@ def execute_tool(
                 tool_input["job_id"],
                 channel_id,
                 thread_ts,
+                user_id=user_id,
             )
             return json.dumps({"success": success, "message": message})
 
@@ -716,6 +719,7 @@ def process_message(
         channel_id: str,
         thread_ts: str,
         conversation_history: list[dict] | None = None,
+        user_id: str = "",
     ) -> str:
         """
         Process a user message and return Claude's response.
@@ -748,6 +752,7 @@ def process_message(
                             block.input,
                             channel_id,
                             thread_ts,
+                            user_id=user_id,
                         )
                         tool_results.append(
                             {

src/xfer/slackbot/slurm_tools.py

@@ -310,6 +310,7 @@ def submit_transfer(
     time_limit: Optional[str] = None,
     job_name: Optional[str] = None,
     rclone_flags: Optional[str] = None,
+    user_id: str = "",
 ) -> TransferResult:
     """
     Submit a data transfer job via xfer.
@@ -393,6 +394,7 @@ def submit_transfer(
             "prepare_job_id": job_id,
             "num_shards": num_shards,
             "submitted_at": datetime.utcnow().isoformat() + "Z",
+            "submitted_by": user_id,
         }
         (run_dir / "request.json").write_text(json.dumps(request_meta, indent=2))
 
@@ -630,11 +632,14 @@ def get_transfer_progress_by_job(job_id: str) -> Optional[dict]:
     return progress
 
 
-def cancel_job(job_id: str, channel_id: str, thread_ts: str) -> tuple[bool, str]:
+def cancel_job(
+    job_id: str, channel_id: str, thread_ts: str, *, user_id: str = ""
+) -> tuple[bool, str]:
     """
     Cancel a Slurm job.
 
-    Validates that the job belongs to the requesting thread via comment.
+    Validates that the job belongs to the requesting thread via comment,
+    and that the requesting user is the one who submitted the job.
     """
     # First, verify the job belongs to this thread
     job_info = get_job_status(job_id)
@@ -645,6 +650,18 @@ def cancel_job(job_id: str, channel_id: str, thread_ts: str) -> tuple[bool, str]
     if expected_comment not in job_info.comment:
         return False, f"Job {job_id} does not belong to this thread. Cannot cancel."
 
+    # Check user ownership via request.json in the job's work_dir
+    if user_id and job_info.work_dir:
+        request_file = Path(job_info.work_dir) / "request.json"
+        if request_file.exists():
+            try:
+                meta = json.loads(request_file.read_text())
+                submitted_by = meta.get("submitted_by", "")
+                if submitted_by and submitted_by != user_id:
+                    return False, "Only the user who submitted this job can cancel it."
+            except (json.JSONDecodeError, IOError):
+                pass  # Allow cancel if request.json is unreadable
+
     # Cancel the job
     try:
         run_cmd(["scancel", job_id], check=True)

tests/test_slackbot_dryrun.py

@@ -25,9 +25,11 @@
     slack_comment,
 )
 from xfer.slackbot.slurm_tools import (
+    JobInfo,
     TransferResult,
     _parse_job_from_sacct_json,
     _write_prepare_script,
+    cancel_job,
     get_allowed_backends,
     get_transfer_progress,
     validate_backend,
@@ -506,6 +508,110 @@ def test_full_flow_simulation():
         print("\n✓ Full flow simulation passed")
 
 
+def test_cancel_job_by_submitter():
+    """Test that the user who submitted a job can cancel it."""
+    print("\n=== Testing cancel job by submitter ===")
+
+    channel = "C07ABC123"
+    thread_ts = "1234567890.123456"
+    comment = slack_comment(channel, thread_ts)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        run_dir = Path(tmpdir)
+        request_meta = {"submitted_by": "U_ALICE"}
+        (run_dir / "request.json").write_text(json.dumps(request_meta))
+
+        job_info = JobInfo(
+            job_id="123",
+            array_job_id=None,
+            state="RUNNING",
+            name="xfer-slack",
+            comment=comment,
+            work_dir=str(run_dir),
+            submit_time=None,
+            start_time=None,
+            end_time=None,
+            partition="transfer",
+        )
+
+        with patch("xfer.slackbot.slurm_tools.get_job_status", return_value=job_info), \
+             patch("xfer.slackbot.slurm_tools.run_cmd"):
+            success, message = cancel_job("123", channel, thread_ts, user_id="U_ALICE")
+            assert success, f"Expected success, got: {message}"
+            assert "cancelled" in message
+
+    print("✓ Cancel job by submitter test passed")
+
+
+def test_cancel_job_by_other_user():
+    """Test that a different user cannot cancel someone else's job."""
+    print("\n=== Testing cancel job by other user ===")
+
+    channel = "C07ABC123"
+    thread_ts = "1234567890.123456"
+    comment = slack_comment(channel, thread_ts)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        run_dir = Path(tmpdir)
+        request_meta = {"submitted_by": "U_ALICE"}
+        (run_dir / "request.json").write_text(json.dumps(request_meta))
+
+        job_info = JobInfo(
+            job_id="123",
+            array_job_id=None,
+            state="RUNNING",
+            name="xfer-slack",
+            comment=comment,
+            work_dir=str(run_dir),
+            submit_time=None,
+            start_time=None,
+            end_time=None,
+            partition="transfer",
+        )
+
+        with patch("xfer.slackbot.slurm_tools.get_job_status", return_value=job_info):
+            success, message = cancel_job("123", channel, thread_ts, user_id="U_BOB")
+            assert not success, f"Expected failure, got success: {message}"
+            assert "Only the user who submitted" in message
+
+    print("✓ Cancel job by other user test passed")
+
+
+def test_cancel_job_legacy_no_submitter():
+    """Test that jobs without submitted_by field can be cancelled by anyone (backwards compat)."""
+    print("\n=== Testing cancel job legacy (no submitter) ===")
+
+    channel = "C07ABC123"
+    thread_ts = "1234567890.123456"
+    comment = slack_comment(channel, thread_ts)
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        run_dir = Path(tmpdir)
+        request_meta = {"source": "s3src:bucket/data", "dest": "s3dst:archive/data"}
+        (run_dir / "request.json").write_text(json.dumps(request_meta))
+
+        job_info = JobInfo(
+            job_id="123",
+            array_job_id=None,
+            state="RUNNING",
+            name="xfer-slack",
+            comment=comment,
+            work_dir=str(run_dir),
+            submit_time=None,
+            start_time=None,
+            end_time=None,
+            partition="transfer",
+        )
+
+        with patch("xfer.slackbot.slurm_tools.get_job_status", return_value=job_info), \
+             patch("xfer.slackbot.slurm_tools.run_cmd"):
+            success, message = cancel_job("123", channel, thread_ts, user_id="U_ANYONE")
+            assert success, f"Expected success for legacy job, got: {message}"
+            assert "cancelled" in message
+
+    print("✓ Cancel job legacy (no submitter) test passed")
+
+
 def main():
     """Run all dry-run tests."""
     print("=" * 60)
@@ -524,6 +630,9 @@ def main():
     test_triage_skip()
     test_triage_error_defaults_to_respond()
     test_full_flow_simulation()
+    test_cancel_job_by_submitter()
+    test_cancel_job_by_other_user()
+    test_cancel_job_legacy_no_submitter()
 
     print("\n" + "=" * 60)
     print("All dry-run tests passed! ✓")