fix: adopted agent image loss

application/api/user/agents/routes.py

@@ -11,6 +11,7 @@ from pydantic import ValidationError as PydanticValidationError
 from application.api import api
 from application.guardrails.config import AgentConfig
 from application.api.user.base import (
+    copy_agent_image_for_user,
     handle_image_upload,
     resolve_prompt_name,
     resolve_source_details,
@@ -1664,7 +1665,7 @@ class AdoptAgent(Resource):
                 # and must get its own values (slug stays NULL until exported).
                 create_kwargs: dict = {}
                 for col in (
-                    "description", "agent_type", "image", "retriever",
+                    "description", "agent_type", "retriever",
                     "default_model_id",
                     "source_id", "prompt_id", "folder_id", "workflow_id",
                     "extra_source_ids",
@@ -1672,6 +1673,16 @@ class AdoptAgent(Resource):
                     val = template.get(col)
                     if val not in (None, ""):
                         create_kwargs[col] = val
+
+                # The avatar must live under the adopter's own upload
+                # directory: image paths are validated against their owner,
+                # so a copied reference to the template owner's blob would
+                # fail closed and render blank.
+                adopted_image = copy_agent_image_for_user(
+                    template.get("image"), user, storage
+                )
+                if adopted_image:
+                    create_kwargs["image"] = adopted_image
                 for col in ("tools", "json_schema", "models", "shared_metadata", "config"):
                     if template.get(col) is not None:
                         create_kwargs[col] = template[col]

application/api/user/base.py

@@ -20,7 +20,12 @@ from application.storage.db.base_repository import looks_like_uuid, row_to_dict
 from application.storage.db.repositories.users import UsersRepository
 from application.storage.db.session import db_readonly, db_session
 from application.storage.storage_creator import StorageCreator
-from application.utils import AGENT_IMAGE_FORMATS, safe_user_storage_component
+from application.utils import (
+    AGENT_IMAGE_FORMATS,
+    get_agent_image_content_type,
+    is_external_image_url,
+    safe_user_storage_component,
+)
 from application.vectorstore.vector_creator import VectorCreator
 
 
@@ -294,6 +299,54 @@ def _safe_agent_image_filename(filename: str) -> str:
     return f"{safe_stem}{extension}"
 
 
+def copy_agent_image_for_user(image_path: object, user: str, storage) -> str:
+    """Copy an internal agent avatar into ``user``'s own attachments directory.
+
+    Avatar paths are validated against their owner's upload directory, so an
+    agent copied to a new owner needs its own blob rather than a reference to
+    the original owner's.
+
+    Args:
+        image_path: Source avatar storage path, or an external image URL.
+        user: Identity that will own the copy.
+        storage: Storage backend used for both the read and the write.
+
+    Returns:
+        The newly owned storage path, an unchanged external URL, or an empty
+        string when the source is unusable.
+    """
+    if not isinstance(image_path, str) or not image_path:
+        return ""
+    if is_external_image_url(image_path):
+        return image_path
+    if get_agent_image_content_type(image_path) is None:
+        return ""
+
+    source_name = PurePosixPath(image_path).name
+    prefix, separator, remainder = source_name.partition("_")
+    if separator and remainder and looks_like_uuid(prefix):
+        source_name = remainder
+    try:
+        filename = _safe_agent_image_filename(source_name)
+    except ValueError:
+        return ""
+
+    owner_component = safe_user_storage_component(user)
+    destination = (
+        f"{settings.UPLOAD_FOLDER.rstrip('/')}/{owner_component}/"
+        f"attachments/{uuid.uuid4()}_{filename}"
+    )
+    try:
+        with storage.get_file(image_path) as source_file:
+            storage.save_file(source_file, destination, storage_class="STANDARD")
+    except Exception as e:
+        current_app.logger.warning(
+            "Could not copy agent image %s for %s: %s", image_path, user, e
+        )
+        return ""
+    return destination
+
+
 def handle_image_upload(
     request, existing_url: str, user: str, storage
 ) -> Tuple[Optional[str], Optional[Response]]:

application/utils.py

@@ -277,7 +277,10 @@ def is_safe_agent_image_path(image_path: object, user_id: object) -> bool:
 
     candidate = PurePosixPath(image_path)
     upload_root = PurePosixPath(str(settings.UPLOAD_FOLDER).rstrip("/"))
-    if candidate.is_absolute() or upload_root.is_absolute() or ".." in candidate.parts:
+    # An absolute UPLOAD_FOLDER yields absolute stored paths, so the two must
+    # agree; mismatched anchors mean the path did not come from this root. The
+    # exact parent match below is what actually contains the path.
+    if candidate.is_absolute() != upload_root.is_absolute() or ".." in candidate.parts:
         return False
     if get_agent_image_content_type(image_path) is None:
         return False

tests/api/user/agents/test_routes_happy.py

@@ -1243,6 +1243,120 @@ class TestAdoptAgentMore:
         mine = repo.list_for_user("u-adopter")
         assert any(a["name"] == "Template X" for a in mine)
 
+    def test_adopt_copies_image_into_adopter_storage(self, app, pg_conn):
+        import io
+        from unittest.mock import MagicMock
+
+        from application.api.user.agents.routes import AdoptAgent
+        from application.core.settings import settings
+        from application.storage.db.repositories.agents import AgentsRepository
+        from application.utils import (
+            is_safe_agent_image_path,
+            safe_user_storage_component,
+        )
+
+        repo = AgentsRepository(pg_conn)
+        template = repo.create(
+            "__system__",
+            "Template Img",
+            "template",
+            image="inputs/__system__/attachments/abc_logo.png",
+        )
+
+        storage = MagicMock()
+        storage.get_file.return_value = io.BytesIO(b"png-bytes")
+        storage.save_file.side_effect = lambda data, path, **kw: {"path": path}
+
+        with _patch_db(pg_conn), patch(
+            "application.api.user.agents.routes.storage", storage
+        ), app.test_request_context(
+            f"/api/adopt_agent?id={template['id']}", method="POST"
+        ):
+            from flask import request
+            request.decoded_token = {"sub": "u-adopter"}
+            response = AdoptAgent().post()
+
+        assert response.status_code == 200
+        adopted = next(
+            a for a in repo.list_for_user("u-adopter") if a["name"] == "Template Img"
+        )
+        owner = safe_user_storage_component("u-adopter")
+        expected_prefix = f"{settings.UPLOAD_FOLDER}/{owner}/attachments/"
+        assert adopted["image"].startswith(expected_prefix)
+        assert adopted["image"].endswith("_logo.png")
+        assert is_safe_agent_image_path(adopted["image"], "u-adopter")
+        storage.get_file.assert_called_once_with(
+            "inputs/__system__/attachments/abc_logo.png"
+        )
+
+    def test_adopt_drops_image_when_copy_fails(self, app, pg_conn):
+        from unittest.mock import MagicMock
+
+        from application.api.user.agents.routes import AdoptAgent
+        from application.storage.db.repositories.agents import AgentsRepository
+
+        repo = AgentsRepository(pg_conn)
+        template = repo.create(
+            "__system__",
+            "Template Broken Img",
+            "template",
+            image="inputs/__system__/attachments/gone_logo.png",
+        )
+
+        storage = MagicMock()
+        storage.get_file.side_effect = FileNotFoundError("missing")
+
+        with _patch_db(pg_conn), patch(
+            "application.api.user.agents.routes.storage", storage
+        ), app.test_request_context(
+            f"/api/adopt_agent?id={template['id']}", method="POST"
+        ):
+            from flask import request
+            request.decoded_token = {"sub": "u-adopter2"}
+            response = AdoptAgent().post()
+
+        assert response.status_code == 200
+        adopted = next(
+            a
+            for a in repo.list_for_user("u-adopter2")
+            if a["name"] == "Template Broken Img"
+        )
+        assert not adopted.get("image")
+
+    def test_adopt_keeps_external_image_url(self, app, pg_conn):
+        from unittest.mock import MagicMock
+
+        from application.api.user.agents.routes import AdoptAgent
+        from application.storage.db.repositories.agents import AgentsRepository
+
+        repo = AgentsRepository(pg_conn)
+        template = repo.create(
+            "__system__",
+            "Template Ext Img",
+            "template",
+            image="https://cdn.example.com/logo.png",
+        )
+
+        storage = MagicMock()
+
+        with _patch_db(pg_conn), patch(
+            "application.api.user.agents.routes.storage", storage
+        ), app.test_request_context(
+            f"/api/adopt_agent?id={template['id']}", method="POST"
+        ):
+            from flask import request
+            request.decoded_token = {"sub": "u-adopter3"}
+            response = AdoptAgent().post()
+
+        assert response.status_code == 200
+        adopted = next(
+            a
+            for a in repo.list_for_user("u-adopter3")
+            if a["name"] == "Template Ext Img"
+        )
+        assert adopted["image"] == "https://cdn.example.com/logo.png"
+        storage.get_file.assert_not_called()
+
     def test_adopt_template_missing_returns_404(self, app, pg_conn):
         from application.api.user.agents.routes import AdoptAgent
 

tests/test_utils.py

@@ -16,6 +16,7 @@ from application.utils import (
     get_hash,
     get_missing_fields,
     generate_image_url,
+    is_safe_agent_image_path,
     limit_chat_history,
     num_tokens_from_object_or_list,
     num_tokens_from_string,
@@ -366,6 +367,66 @@ class TestGenerateImageUrl:
             )
             assert result == ""
 
+    @pytest.mark.unit
+    def test_absolute_upload_folder_still_serves_owned_images(self):
+        with patch("application.utils.settings") as s:
+            s.JWT_SECRET_KEY = "test-image-secret"
+            s.UPLOAD_FOLDER = "/data/inputs"
+            s.API_URL = "https://api.example.com"
+            result = generate_image_url(
+                "/data/inputs/user-1/attachments/avatar.png",
+                "00000000-0000-0000-0000-000000000001",
+                "user-1",
+            )
+            assert result.startswith(
+                "https://api.example.com/api/images/"
+                "00000000-0000-0000-0000-000000000001/"
+            )
+
+
+class TestIsSafeAgentImagePath:
+
+    @pytest.mark.unit
+    def test_absolute_upload_folder_accepts_owned_path(self):
+        with patch("application.utils.settings") as s:
+            s.UPLOAD_FOLDER = "/data/inputs"
+            assert is_safe_agent_image_path(
+                "/data/inputs/user-1/attachments/avatar.png", "user-1"
+            )
+
+    @pytest.mark.unit
+    def test_absolute_upload_folder_rejects_other_owner(self):
+        with patch("application.utils.settings") as s:
+            s.UPLOAD_FOLDER = "/data/inputs"
+            assert not is_safe_agent_image_path(
+                "/data/inputs/user-2/attachments/avatar.png", "user-1"
+            )
+
+    @pytest.mark.unit
+    def test_absolute_upload_folder_rejects_escape(self):
+        with patch("application.utils.settings") as s:
+            s.UPLOAD_FOLDER = "/data/inputs"
+            assert not is_safe_agent_image_path(
+                "/data/inputs/user-1/attachments/../../../etc/passwd.png", "user-1"
+            )
+            assert not is_safe_agent_image_path("/etc/passwd.png", "user-1")
+
+    @pytest.mark.unit
+    def test_relative_upload_folder_rejects_absolute_path(self):
+        with patch("application.utils.settings") as s:
+            s.UPLOAD_FOLDER = "inputs"
+            assert not is_safe_agent_image_path(
+                "/inputs/user-1/attachments/avatar.png", "user-1"
+            )
+
+    @pytest.mark.unit
+    def test_relative_upload_folder_accepts_owned_path(self):
+        with patch("application.utils.settings") as s:
+            s.UPLOAD_FOLDER = "inputs"
+            assert is_safe_agent_image_path(
+                "inputs/user-1/attachments/avatar.png", "user-1"
+            )
+
 
 class TestCalculateCompressionThreshold: