Replaced admin-view, with per-user ownership model

Signed-off-by: A-makarim <syedmakarim0.2@gmail.com>

Author: Thun78

ai_platform_engineering/autonomous_agents/src/autonomous_agents/routes/tasks.py

@@ -51,7 +51,17 @@ def _assert_task_access(task: TaskDefinition, caller_email: str | None, is_admin
     """Raise 403 if caller does not own the task and is not an admin.
 
     Tasks without an owner_id (created before this feature) are treated as
-    admin-only to prevent accidental cross-user exposure.
+    admin-only to prevent accidental cross-user exposure. This orphaned-task
+    branch is the **only** path that produces a 403 for an admin-eligible
+    caller: once `is_admin` is true above, every other ownership check is
+    short-circuited. Backfilling `owner_id` for pre-feature tasks is the
+    out-of-band remediation; we deliberately do not auto-assign here so the
+    audit story stays clean (admin acted, not "system silently re-owned").
+
+    Audit signal for cross-user admin actions is NOT emitted here — it is
+    emitted at the verb call sites (`update_task` / `delete_task` /
+    `trigger_task_manually`) so the log line carries the action verb and
+    the task's human-readable name without re-fetching from the store.
     """
     if is_admin:
         return
@@ -243,6 +253,15 @@ async def update_task(task_id: str, task: TaskDefinition, request: Request) -> d
     if existing is not None:
         caller_email, is_admin = _get_caller(request)
         _assert_task_access(existing, caller_email, is_admin)
+        # Admin acting on someone else's task -- emit an audit log line at
+        # the verb call site (per plan section 4.4) so log scanners see the
+        # action verb and the human-readable task name without joining
+        # against the store.
+        if is_admin and existing.owner_id and existing.owner_id != caller_email:
+            logger.info(
+                "Admin %s acted on task %s (%r) owned by %s (action=%s)",
+                caller_email, task_id, existing.name, existing.owner_id, "update",
+            )
         # Preserve owner_id from the original task — callers cannot reassign ownership.
         task = task.model_copy(update={"owner_id": existing.owner_id})
 
@@ -310,6 +329,11 @@ async def delete_task(task_id: str, request: Request) -> None:
         raise HTTPException(status_code=404, detail=f"Task '{task_id}' not found")
     caller_email, is_admin = _get_caller(request)
     _assert_task_access(task, caller_email, is_admin)
+    if is_admin and task.owner_id and task.owner_id != caller_email:
+        logger.info(
+            "Admin %s acted on task %s (%r) owned by %s (action=%s)",
+            caller_email, task_id, task.name, task.owner_id, "delete",
+        )
     try:
         await store.delete(task_id)
     except TaskNotFoundError as exc:
@@ -340,6 +364,11 @@ async def trigger_task_manually(task_id: str, request: Request) -> dict:
         raise HTTPException(status_code=404, detail=f"Task '{task_id}' not found")
     caller_email, is_admin = _get_caller(request)
     _assert_task_access(task, caller_email, is_admin)
+    if is_admin and task.owner_id and task.owner_id != caller_email:
+        logger.info(
+            "Admin %s acted on task %s (%r) owned by %s (action=%s)",
+            caller_email, task_id, task.name, task.owner_id, "trigger",
+        )
 
     # Fire-and-forget -- the run is recorded in the store as it
     # progresses so the UI can poll /tasks/{id}/runs to see the result.

ai_platform_engineering/autonomous_agents/tests/test_tasks_route.py

@@ -762,3 +762,185 @@ def test_owner_can_delete_own_task(self, client: TestClient):
         client.post("/api/v1/tasks", json=_cron_task("t1"), headers=_user_headers("alice@example.com"))
         resp = client.delete("/api/v1/tasks/t1", headers=_user_headers("alice@example.com"))
         assert resp.status_code == 204
+
+    def test_non_admin_cannot_update_another_users_task(self, client: TestClient):
+        """PUT /tasks/{id} returns 403 when task belongs to a different user."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("alice@example.com"),
+        )
+        updated_payload = _cron_task("t1")
+        updated_payload["name"] = "renamed by bob"
+        resp = client.put(
+            "/api/v1/tasks/t1",
+            json=updated_payload,
+            headers=_user_headers("bob@example.com"),
+        )
+        assert resp.status_code == 403
+
+    def test_non_admin_cannot_trigger_another_users_task(self, client: TestClient):
+        """POST /tasks/{id}/run returns 403 when task belongs to a different user."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("alice@example.com"),
+        )
+        resp = client.post(
+            "/api/v1/tasks/t1/run",
+            headers=_user_headers("bob@example.com"),
+        )
+        assert resp.status_code == 403
+
+    def test_admin_can_update_another_users_task(self, client: TestClient):
+        """An admin can PUT another user's task."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("bob@example.com"),
+        )
+        updated_payload = _cron_task("t1")
+        updated_payload["name"] = "renamed by admin"
+        resp = client.put(
+            "/api/v1/tasks/t1",
+            json=updated_payload,
+            headers=_admin_headers(),
+        )
+        assert resp.status_code == 200
+        # Ownership is preserved across an admin-initiated update.
+        listed = client.get("/api/v1/tasks", headers=_admin_headers()).json()
+        owners = {t["id"]: t["owner_id"] for t in listed}
+        assert owners["t1"] == "bob@example.com"
+
+    def test_admin_can_delete_another_users_task(self, client: TestClient):
+        """An admin can DELETE another user's task."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("bob@example.com"),
+        )
+        resp = client.delete("/api/v1/tasks/t1", headers=_admin_headers())
+        assert resp.status_code == 204
+
+
+class TestAdminAuditLogging:
+    """Section 4.4: admin actions on someone else's task emit a log line."""
+
+    def test_admin_update_emits_audit_log(
+        self, client: TestClient, caplog: pytest.LogCaptureFixture
+    ):
+        """Admin PUT on another user's task emits a log line with both emails + action."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("bob@example.com"),
+        )
+        updated_payload = _cron_task("t1")
+        updated_payload["name"] = "renamed by admin"
+        caplog.clear()
+        with caplog.at_level("INFO", logger="autonomous_agents"):
+            resp = client.put(
+                "/api/v1/tasks/t1",
+                json=updated_payload,
+                headers={
+                    "X-Authenticated-User-Email": "admin@example.com",
+                    "X-Authenticated-User-Is-Admin": "true",
+                },
+            )
+            assert resp.status_code == 200
+        admin_log_lines = [
+            r.getMessage()
+            for r in caplog.records
+            if "Admin " in r.getMessage() and "acted on task" in r.getMessage()
+        ]
+        assert admin_log_lines, "expected an admin audit log line on PUT"
+        msg = admin_log_lines[0]
+        assert "admin@example.com" in msg
+        assert "bob@example.com" in msg
+        assert "t1" in msg
+        assert "update" in msg
+
+    def test_admin_delete_emits_audit_log(
+        self, client: TestClient, caplog: pytest.LogCaptureFixture
+    ):
+        """Admin DELETE on another user's task emits a log line with both emails + action."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("bob@example.com"),
+        )
+        caplog.clear()
+        with caplog.at_level("INFO", logger="autonomous_agents"):
+            resp = client.delete(
+                "/api/v1/tasks/t1",
+                headers={
+                    "X-Authenticated-User-Email": "admin@example.com",
+                    "X-Authenticated-User-Is-Admin": "true",
+                },
+            )
+            assert resp.status_code == 204
+        admin_log_lines = [
+            r.getMessage()
+            for r in caplog.records
+            if "Admin " in r.getMessage() and "acted on task" in r.getMessage()
+        ]
+        assert admin_log_lines, "expected an admin audit log line on DELETE"
+        msg = admin_log_lines[0]
+        assert "admin@example.com" in msg
+        assert "bob@example.com" in msg
+        assert "delete" in msg
+
+    def test_admin_trigger_emits_audit_log(
+        self, client: TestClient, caplog: pytest.LogCaptureFixture
+    ):
+        """Admin manual trigger on another user's task emits a log line."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("bob@example.com"),
+        )
+        caplog.clear()
+        with caplog.at_level("INFO", logger="autonomous_agents"):
+            resp = client.post(
+                "/api/v1/tasks/t1/run",
+                headers={
+                    "X-Authenticated-User-Email": "admin@example.com",
+                    "X-Authenticated-User-Is-Admin": "true",
+                },
+            )
+            assert resp.status_code == 200
+        admin_log_lines = [
+            r.getMessage()
+            for r in caplog.records
+            if "Admin " in r.getMessage() and "acted on task" in r.getMessage()
+        ]
+        assert admin_log_lines, "expected an admin audit log line on manual trigger"
+        msg = admin_log_lines[0]
+        assert "admin@example.com" in msg
+        assert "bob@example.com" in msg
+        assert "trigger" in msg
+
+    def test_owner_action_does_not_emit_admin_audit_log(
+        self, client: TestClient, caplog: pytest.LogCaptureFixture
+    ):
+        """Owner acting on their own task must not emit the cross-user admin audit log line."""
+        client.post(
+            "/api/v1/tasks",
+            json=_cron_task("t1"),
+            headers=_user_headers("bob@example.com"),
+        )
+        caplog.clear()
+        with caplog.at_level("INFO", logger="autonomous_agents"):
+            resp = client.delete(
+                "/api/v1/tasks/t1",
+                headers=_user_headers("bob@example.com"),
+            )
+            assert resp.status_code == 204
+        admin_log_lines = [
+            r.getMessage()
+            for r in caplog.records
+            if "Admin " in r.getMessage() and "acted on task" in r.getMessage()
+        ]
+        assert admin_log_lines == [], (
+            "owner deleting their own task should not emit the admin audit line"
+        )