From e4b06609b1c365a0c8780a0a5800f5d42568bd15 Mon Sep 17 00:00:00 2001 From: Alex Date: Thu, 13 Aug 2026 10:05:17 +0100 Subject: [PATCH] fix: adopted agent image loss --- application/api/user/agents/routes.py | 13 ++- application/api/user/base.py | 55 +++++++++- application/utils.py | 5 +- tests/api/user/agents/test_routes_happy.py | 114 +++++++++++++++++++++ tests/test_utils.py | 61 +++++++++++ 5 files changed, 245 insertions(+), 3 deletions(-) diff --git a/application/api/user/agents/routes.py b/application/api/user/agents/routes.py index 938ea855..fc8f3753 100644 --- a/application/api/user/agents/routes.py +++ b/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] diff --git a/application/api/user/base.py b/application/api/user/base.py index f2fcfe39..c73aa692 100644 --- a/application/api/user/base.py +++ b/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]]: diff --git a/application/utils.py b/application/utils.py index 3c3e95dd..bd52bf41 100644 --- a/application/utils.py +++ b/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 diff --git a/tests/api/user/agents/test_routes_happy.py b/tests/api/user/agents/test_routes_happy.py index 94aaa559..12e6197e 100644 --- a/tests/api/user/agents/test_routes_happy.py +++ b/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 diff --git a/tests/test_utils.py b/tests/test_utils.py index b4c89de7..16ec3852 100644 --- a/tests/test_utils.py +++ b/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: