Migrate image endpoint to FastAPI, delete WSGI repository controller.

Move display_image_in_repository to FastAPI with hardened security:
realpath symlink validation, image-only MIME type restriction, stdlib
mimetypes fallback. Delete controllers/repository.py, remove image
route from buildapp, clean stale mypy/sphinx refs, drop orphaned
test_0140/test_1160 image tests. Add 5 new API tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
John Chilton
2026-03-17 08:21:24 -04:00
co-authored by Claude Opus 4.6
parent 6ac47fec4d
commit 9a4e3984bc
8 changed files with 103 additions and 177 deletions
@@ -33,14 +33,6 @@ tool\_shed.webapp.controllers.hg module
:undoc-members:
:show-inheritance:
tool\_shed.webapp.controllers.repository module
-----------------------------------------------
.. automodule:: tool_shed.webapp.controllers.repository
:members:
:undoc-members:
:show-inheritance:
tool\_shed.webapp.controllers.repository\_review module
-------------------------------------------------------
@@ -1,55 +0,0 @@
import logging
from ..base import common
from ..base.testcase import ShedTestCase
log = logging.getLogger(__name__)
repository_name = "htseq_count_0140"
repository_description = "Converter: BED to GFF"
repository_long_description = "Convert bed to gff"
category_name = "Test 0140 Tool Help Images"
category_description = "Test 0140 Tool Help Images"
"""
1) Create and populate the htseq_count_0140 repository.
2) Visit the manage_repository page.
3) Simulate clicking the htseq_count tool button.
4) On the resulting display tool page, look for the image string similar to the following string where the encoded repository_id is previously determined:
src="/repository/static/images/<id>/count_modes.png"
"""
class TestToolHelpImages(ShedTestCase):
"""Test features related to tool help images."""
def test_0000_initiate_users(self):
"""Create necessary user accounts."""
self.login(email=common.test_user_1_email, username=common.test_user_1_name)
self.login(email=common.admin_email, username=common.admin_username)
def test_0005_create_htseq_count_repository(self):
"""Create and populate htseq_count_0140.
We are at step 1 - Create and populate the htseq_count_0140 repository.
Create the htseq_count_0140 repository and upload the tarball.
"""
category = self.create_category(name=category_name, description=category_description)
self.login(email=common.test_user_1_email, username=common.test_user_1_name)
# Create a repository named htseq_count_0140 owned by user1.
repository = self.get_or_create_repository(
name=repository_name,
description=repository_description,
long_description=repository_long_description,
owner=common.test_user_1_name,
category=category,
strings_displayed=[],
)
# Upload htseq_count.tar to the repository.
self.commit_tar_to_repository(
repository,
"htseq_count/htseq_count.tar",
commit_message="Uploaded htseq_count.tar.",
)
@@ -1,63 +0,0 @@
import logging
from ..base import common
from ..base.testcase import ShedTestCase
log = logging.getLogger(__name__)
repository_name = "htseq_count_0140"
repository_description = "Converter: BED to GFF"
repository_long_description = "Convert bed to gff"
category_name = "Test 0140 Tool Help Images"
category_description = "Test 0140 Tool Help Images"
# 1) Create and populate the htseq_count_0140 repository.
# 2) Visit the manage_repository page, then the tool page, and look for the image string
# similar to the following string where the encoded repository_id is previously determined:
# src="/repository/static/images/<id>/count_modes.png"
class TestToolHelpImages(ShedTestCase):
"""Test features related to tool help images."""
requires_galaxy = True
def test_0000_initiate_users(self):
"""Create necessary user accounts."""
self.login(email=common.test_user_1_email, username=common.test_user_1_name)
self.login(email=common.admin_email, username=common.admin_username)
def test_0005_create_htseq_count_repository(self):
"""Create and populate htseq_count_0140.
We are at step 1 - Create and populate the htseq_count_0140 repository.
Create the htseq_count_0140 repository and upload the tarball.
"""
category = self.create_category(name=category_name, description=category_description)
self.login(email=common.test_user_1_email, username=common.test_user_1_name)
# Create a repository named htseq_count_0140 owned by user1.
repository = self.get_or_create_repository(
name=repository_name,
description=repository_description,
long_description=repository_long_description,
owner=common.test_user_1_name,
category=category,
strings_displayed=[],
)
if self.repository_is_new(repository):
# Upload htseq_count.tar to the repository if it hasn't already been populated.
self.commit_tar_to_repository(
repository,
"htseq_count/htseq_count.tar",
commit_message="Uploaded htseq_count.tar.",
)
def test_0010_load_tool_page(self):
"""Load the tool page and check for the image URL.
This is a duplicate of test method _0010 in test_0140_tool_help_images.
"""
# TODO: replace with API-based tool metadata check or Vue route.
# load_display_tool_page depends on deleted Mako route (/repository/display_tool).
pass
@@ -15,6 +15,7 @@ Endpoints tested:
GET /repository/updated_changeset_revisions
GET /repository/get_repository_type
GET /repository/get_tool_dependencies
GET /repository/static/images/{repository_id}/{image_file}
"""
import json
@@ -28,6 +29,7 @@ from galaxy.util.tool_shed.encoding_util import (
)
from galaxy_test.base import api_asserts
from ..base.api import ShedApiTestCase
from ..base.populators import TEST_DATA_REPO_FILES
class TestGalaxyInstallApis(ShedApiTestCase):
@@ -569,6 +571,58 @@ class TestGalaxyInstallApis(ShedApiTestCase):
# The next installable should be reachable from the update path
assert next_rev in updated_revisions
# ---- display_image_in_repository (static images) ----------------------
def _setup_htseq_count_repo(self):
"""Setup htseq_count repo which contains static/images/count_modes.png."""
populator = self.populator
category_id = populator.new_category(prefix="imagetest").id
repository = populator.new_repository(category_id, prefix="imagetest")
htseq_tar = TEST_DATA_REPO_FILES.joinpath("htseq_count/htseq_count.tar")
populator.upload_revision(repository, htseq_tar)
return repository
def test_display_image_in_repository(self):
"""Image endpoint should serve a PNG with correct MIME type."""
repository = self._setup_htseq_count_repo()
url = f"{self.url}/repository/static/images/{repository.id}/count_modes.png"
response = requests.get(url)
api_asserts.assert_status_code_is_ok(response)
assert len(response.content) > 0, "Expected non-empty image content"
content_type = response.headers.get("content-type", "")
assert "image/png" in content_type, \
f"Expected image/png content-type, got {content_type}"
def test_display_image_nonexistent_file(self):
"""Requesting a nonexistent image should return empty/error."""
repository = self._setup_htseq_count_repo()
url = f"{self.url}/repository/static/images/{repository.id}/nonexistent.png"
response = requests.get(url)
# Should either 404 or return empty
assert response.status_code >= 400 or len(response.content) == 0
def test_display_image_nonexistent_repo(self):
"""Requesting an image from a nonexistent repo should fail."""
url = f"{self.url}/repository/static/images/invalid_id_xyz/count_modes.png"
response = requests.get(url)
assert response.status_code >= 400
def test_display_image_path_traversal_rejected(self):
"""Path traversal attempts should not serve files outside the repo."""
repository = self._setup_htseq_count_repo()
url = f"{self.url}/repository/static/images/{repository.id}/..%2F..%2F..%2Fetc%2Fpasswd"
response = requests.get(url)
assert response.status_code >= 400 or len(response.content) == 0
def test_display_image_rejects_non_image_files(self):
"""Endpoint only serves image MIME types, rejects .xml etc."""
repository = self._setup_htseq_count_repo()
# htseq_count.tar contains htseq-count.xml — a non-image file
url = f"{self.url}/repository/static/images/{repository.id}/htseq-count.xml"
response = requests.get(url)
assert response.status_code >= 400, \
f"Expected rejection of non-image file, got {response.status_code}"
def test_all_endpoints_reject_nonexistent_repo(self):
"""All endpoints should fail gracefully for a nonexistent repository."""
bogus_params = dict(
+49 -3
View File
@@ -7,10 +7,17 @@ migrated here so the legacy WSGI controller can be deleted.
"""
import logging
import mimetypes
import os
from typing import Optional
from fastapi import Form
from starlette.responses import Response
from galaxy.exceptions import ObjectNotFound
from galaxy.tool_shed.util.repository_util import get_absolute_path_to_file_in_repository
from starlette.responses import (
FileResponse,
Response,
)
from tool_shed.context import SessionRequestContext
from tool_shed.managers.repositories import (
@@ -24,6 +31,7 @@ from tool_shed.managers.repositories import (
previous_changeset_revisions_str,
updated_changeset_revisions_str,
)
from tool_shed.util.repository_util import get_repository_in_tool_shed
from tool_shed.structured_app import ToolShedApp
from . import (
depends,
@@ -44,7 +52,6 @@ router = Router(tags=["legacy_install"])
class FastAPILegacyInstall:
app: ToolShedApp = depends(ToolShedApp)
# -- plain-text GET endpoints (Galaxy reads response.text) ---------
@router.get(
"/repository/get_ctx_rev",
@@ -137,7 +144,6 @@ class FastAPILegacyInstall:
result = get_tool_dependencies_for_changeset(self.app, name, owner, changeset_revision)
return Response(content=result, media_type="text/plain")
# -- JSON endpoints (Galaxy reads json.loads(response.text)) -------
@router.get(
"/repository/get_repository_dependencies",
@@ -173,3 +179,43 @@ class FastAPILegacyInstall:
encoded_str: Optional[str] = Form(default=None),
) -> dict:
return get_required_repo_info_dict_from_encoded(trans, encoded_str)
@router.get(
"/repository/static/images/{repository_id}/{image_file:path}",
operation_id="legacy_install__display_image",
tags=["legacy_install"],
)
def display_image_in_repository(
self,
repository_id: str,
image_file: str,
) -> FileResponse:
repository = get_repository_in_tool_shed(self.app, repository_id)
if not repository:
raise ObjectNotFound("Repository not found.")
repo_files_dir = repository.repo_path(self.app)
path_to_file = get_absolute_path_to_file_in_repository(repo_files_dir, image_file)
if not path_to_file or not os.path.exists(path_to_file):
raise ObjectNotFound("Image file not found.")
# Validate resolved path stays within repository directory (symlink protection)
resolved = os.path.realpath(path_to_file)
repo_dir_resolved = os.path.realpath(repo_files_dir)
if resolved != repo_dir_resolved and not resolved.startswith(repo_dir_resolved + os.sep):
raise ObjectNotFound("Image file not found.")
# Determine MIME type - try datatypes registry first, fall back to stdlib
media_type = None
file_name = os.path.basename(image_file)
try:
extension = file_name.rsplit(".", 1)[-1]
media_type = self.app.datatypes_registry.get_mimetype_by_extension(extension)
except Exception:
pass
if not media_type or media_type == "application/octet-stream":
guessed, _ = mimetypes.guess_type(file_name)
if guessed:
media_type = guessed
# Only serve known image types
if not media_type or not media_type.startswith("image/"):
raise ObjectNotFound("Image file not found.")
return FileResponse(path=resolved, media_type=media_type)
-8
View File
@@ -106,14 +106,6 @@ def app_pair(global_conf, load_app_kwds=None, **kwargs):
# Create the universe WSGI application
webapp = CommunityWebApplication(app, session_cookie="galaxycommunitysession", name="tool_shed")
add_ui_controllers(webapp, app)
# Handle displaying tool help images and README file images for tools contained in repositories.
webapp.add_route(
"/repository/static/images/{repository_id}/{image_file:.+?}",
controller="repository",
action="display_image_in_repository",
repository_id=None,
image_file=None,
)
# Enable 'hg clone' functionality on repos by letting hgwebapp handle the request
webapp.add_route("/repos/*path_info", controller="hg", action="handle_request", path_info="/")
webapp.finalize_config()
@@ -1,38 +0,0 @@
import logging
import os
from galaxy import web
from galaxy.webapps.base.controller import BaseUIController
from tool_shed.util import repository_util
log = logging.getLogger(__name__)
class RepositoryController(BaseUIController):
@web.expose
def display_image_in_repository(self, trans, **kwd):
"""
Open an image file that is contained in a repository for display.
Images can be referenced from README.rst or Galaxy tool help sections.
"""
repository_id = kwd.get("repository_id", None)
relative_path_to_image_file = kwd.get("image_file", None)
if repository_id and relative_path_to_image_file:
repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id)
if repository:
repo_files_dir = repository.repo_path(trans.app)
path_to_file = repository_util.get_absolute_path_to_file_in_repository(
repo_files_dir, relative_path_to_image_file
)
if os.path.exists(path_to_file):
file_name = os.path.basename(relative_path_to_image_file)
try:
extension = file_name.split(".")[-1]
except Exception:
extension = None
if extension:
mimetype = trans.app.datatypes_registry.get_mimetype_by_extension(extension)
if mimetype:
trans.response.set_content_type(mimetype)
return open(path_to_file, "rb")
-2
View File
@@ -635,8 +635,6 @@ check_untyped_defs = False
check_untyped_defs = False
[mypy-tool_shed.webapp.controllers.repository_review]
check_untyped_defs = False
[mypy-tool_shed.webapp.controllers.repository]
check_untyped_defs = False
[mypy-tool_shed.webapp.api.repository_revisions]
check_untyped_defs = False
[mypy-tool_shed.webapp.api.repositories]