Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 45 additions & 3 deletions modules/weko-search-ui/weko_search_ui/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,11 @@

import time
import traceback
from functools import wraps
from xml.etree import ElementTree

from blinker import Namespace
from flask import Blueprint, current_app, flash, jsonify, render_template, request
from flask import Blueprint, abort, current_app, flash, jsonify, render_template, request
from flask_babelex import gettext as _
from flask_login import login_required
from flask_security import current_user
Expand All @@ -41,7 +42,10 @@
from weko_admin.utils import get_search_setting
from weko_index_tree.api import Indexes
from weko_index_tree.models import IndexStyle
from weko_index_tree.utils import get_index_link_list
from weko_index_tree.utils import (
filter_index_list_by_role,
get_index_link_list
)
from weko_records.api import ItemLink, FeedbackMailList
from weko_records_ui.ipaddr import check_site_license_permission
from weko_workflow.utils import (
Expand Down Expand Up @@ -77,6 +81,43 @@
static_folder="static",
)

def check_index_permission(view):
"""Require access to indexes supplied as ``path_str`` or ``index_id``."""

@wraps(view)
def decorated_view(*args, **kwargs):
path_str = kwargs.get("path_str")
index_id = kwargs.get("index_id")
if path_str is not None:
index_ids = path_str.split("_")
elif index_id is not None:
index_ids = [str(index_id)]
else:
abort(404)

index_list = []
for index_id in index_ids:
if not index_id.isdigit():
abort(404)

index = Indexes.get_index(index_id=index_id)
if index is None:
abort(404)

index_list.append(index)

allowed_index_list = filter_index_list_by_role(index_list)
if not allowed_index_list:
abort(403)

if path_str is not None:
kwargs["path_str"] = "_".join(str(index.id) for index in allowed_index_list)
else:
kwargs["index_id"] = allowed_index_list[0].id

return view(*args, **kwargs)

return decorated_view

@blueprint.route("/search/index")
@check_index_access_permissions
Expand Down Expand Up @@ -351,6 +392,7 @@ def opensearch_description():


@blueprint.route("/journal_info/<int:index_id>", methods=["GET"])
@check_index_permission

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (testing): check_index_permission reads only kwargs for index_id and path_str, so direct positional calls such as the existing journal_detail(33) and get_path_name_dict('33_44') tests have neither value and abort with 404 instead of invoking the view.

Triggers: When these decorated views are called positionally, as in the existing unit tests or any non-routing caller.

Suggested fix: Accept the corresponding positional arguments or update the decorator to bind arguments using the wrapped function signature before checking them.

def journal_detail(index_id=0):
"""Render a check view."""
result = get_journal_info(index_id)
Expand All @@ -373,8 +415,8 @@ def get_child_list(index_id=0):
"""Get child id list to index list display."""
return jsonify(Indexes.get_child_id_list(index_id))


@blueprint.route("/get_path_name_dict/<string:path_str>", methods=["GET"])
@check_index_permission
def get_path_name_dict(path_str=""):
"""Get path and name."""
path_name_dict = {}
Expand Down
3 changes: 2 additions & 1 deletion modules/weko-workflow/weko_workflow/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -1358,7 +1358,7 @@ def _get_shared_user_ids_from_list(shared_user_ids_list):
# if exist shared_user_ids or owner allow to access
if int(cur_user) == int(activity_owner):
return 0

if proxy_posting:
# If current user is in activity_user_ids or temp_user_ids
if int(cur_user) in activity_user_ids + temp_user_ids:
Expand Down Expand Up @@ -2802,6 +2802,7 @@ def save_item_application(activity_id='0', action_id='0'):
@workflow_blueprint.route('/get_feedback_maillist/<string:activity_id>',
methods=['GET'])
@login_required
@check_authority

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): Applying check_authority before get_feedback_maillist dereferences activity_detail.action_order without checking whether get_activity_by_id(activity_id) returned None, so a nonexistent activity ID raises AttributeError and returns a 500 response instead of reaching the view's argument/error handling.

Triggers: When an authenticated client requests /workflow/get_feedback_maillist/<activity_id> for an activity ID that does not exist.

Suggested fix: Return a 404 or the endpoint's existing error response when activity_detail is None before accessing action_order.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): Unauthorized users are rejected by check_authority with jsonify(code=403, msg=...) but no HTTP status argument, so the endpoint returns HTTP 200 with a body containing code: 403; API clients and middleware that rely on the HTTP status still treat the unauthorized request as successful.

Triggers: When a logged-in user is denied by the workflow action authority check.

Suggested fix: Return the JSON response with HTTP status 403, for example return jsonify(code=403, msg=error_msg), 403.

def get_feedback_maillist(activity_id='0'):
"""アクティビティに設定されているフィードバックメール送信先の情報を取得して返す

Expand Down
Loading