fix 61534

modules/weko-admin/tests/test_utils.py

@@ -60,7 +60,8 @@
     get_title_facets,
     is_exits_facet,
     overwrite_the_memory_config_with_db,
-    get_detail_search_list
+    get_detail_search_list,
+    sanitize_html_string
 )
 
 from tests.helpers import json_data
@@ -2616,3 +2617,22 @@ def test_get_detail_search_list(i18n_app, users):
     with patch("flask_login.utils._get_user", return_value=users[3]['obj']):
         result =  get_detail_search_list()
         assert result
+
+# def sanitize_html_string(html_string, allow_tags=[], allow_attributes={}, strip=True):
+# .tox/c1/bin/pytest --cov=weko_admin tests/test_utils.py::test_sanitize_html_string -vv -s --cov-branch --cov-report=term --basetemp=/code/modules/weko-admin/.tox/c1/tmp
+def test_sanitize_html_string():
+    test_str = "<div title=\"test\" onclick=\"alert(0)\">a<script>alert(1)</script>b</div>"
+    result = sanitize_html_string(test_str)
+    assert result == "aalert(1)b"
+
+    result = sanitize_html_string(test_str, allow_tags=["div"])
+    assert result == "<div>aalert(1)b</div>"
+
+    result = sanitize_html_string(test_str, allow_tags=["div"], allow_attributes={"div": ["onclick"]})
+    assert result == "<div onclick=\"alert(0)\">aalert(1)b</div>"
+
+    result = sanitize_html_string(test_str, allow_tags=["div"], allow_attributes={"div": ["onclick"]}, strip=True)
+    assert result == "<div onclick=\"alert(0)\">aalert(1)b</div>"
+
+    result = sanitize_html_string(12345)
+    assert result == 12345

modules/weko-admin/weko_admin/utils.py

@@ -30,6 +30,7 @@
 from io import BytesIO, StringIO
 from typing import Dict, Optional, Tuple, Union
 
+import bleach
 import requests
 from flask import current_app, request
 from flask_babelex import gettext as __
@@ -2835,3 +2836,30 @@ def _elasticsearch_remake_item_index(index_name):
     current_app.logger.info(' END elasticsearch import from records_metadata')
 
     return returnlist
+
+def sanitize_html_string(
+        value,
+        allow_tags=[],
+        allow_attributes={},
+        strip=True
+    ):
+    """bleachを使用してHTMLをサニタイズする
+
+    Args:
+        value (str): HTML文字列
+        allow_tags (list): 許可するHTMLタグのリスト
+        allow_attributes (dict): 許可するHTML属性の辞書
+        strip (bool): サニタイズ後にタグを削除するかどうか
+
+    Returns:
+        str: サニタイズされたHTML文字列
+    """
+
+    if isinstance(value, str):
+        value = bleach.clean(
+            value,
+            tags=allow_tags,
+            attributes=allow_attributes,
+            strip=strip
+        ).strip()
+    return value

modules/weko-search-ui/tests/conftest.py

@@ -295,7 +295,7 @@ def base_app(instance_path, search_class, request):
         WEKO_INDEX_TREE_STATE_PREFIX="index_tree_expand_state",
         REDIS_PORT="6379",
         DEPOSIT_DEFAULT_JSONSCHEMA=DEPOSIT_DEFAULT_JSONSCHEMA,
-        SERVER_NAME="TEST_SERVER",
+        SERVER_NAME="test_server",
         LOGIN_DISABLED=False,
         INDEXER_DEFAULT_DOCTYPE="item-v1.0.0",
         WEKO_SCHEMA_JPCOAR_V1_SCHEMA_NAME = 'jpcoar_v1_mapping',
@@ -791,6 +791,15 @@ def db(app):
     if not database_exists(str(db_.engine.url)):
         create_database(str(db_.engine.url))
     db_.create_all()
+    _now = datetime.now()
+    _p_start = _now.date().replace(day=1)
+    _p_end = (_p_start + timedelta(days=31)).replace(day=1)
+    _p_name = "user_activity_logs_{}_{:02d}".format(_now.year, _now.month)
+    db_.session.execute(
+        "CREATE TABLE IF NOT EXISTS {name} PARTITION OF user_activity_logs "
+        "FOR VALUES FROM ('{start}') TO ('{end}');".format(
+            name=_p_name, start=_p_start, end=_p_end))
+    db_.session.commit()
     yield db_
     db_.session.remove()
     db_.drop_all()
@@ -878,6 +887,123 @@ def client_request_args(app, file_instance_mock):
             )
         yield r
 
+@pytest.yield_fixture()
+def client_request_args2(app, file_instance_mock):
+    app.register_blueprint(
+        create_blueprint(app, app.config["WEKO_SEARCH_REST_ENDPOINTS"])
+    )
+
+    file_path = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        "data",
+        "sample_file",
+        "sample_file.txt",
+    )
+
+    # files = {'upload_file': open(file_path,'rb')}
+    # values = {'DB': 'photcat', 'OUT': 'txt', 'SHORT': 'short'}
+
+    # r = requests.post(url, files=files, data=values)
+
+    with app.test_client() as client:
+        with patch("flask.templating._render", return_value=""):
+            r = client.get(
+                "/",
+                query_string={
+                    "index_id": "33",
+                    "page": 1,
+                    "count": 20,
+                    "term": 14,
+                    "lang": "en",
+                    "parent_id": 33,
+                    "index_info": {},
+                    "community": "comm1",
+                    "item_link": "1",
+                    "is_search": 1,
+                    "search_type": WEKO_SEARCH_TYPE_DICT["INDEX"],
+                    "is_change_identifier": True,
+                    "remote_addr": "0.0.0.0",
+                    "referrer": "test",
+                    "host": "127.0.0.1",
+                    "q": "q"
+                },
+            )
+        yield r
+
+@pytest.yield_fixture()
+def client_request_args3(app, file_instance_mock):
+    app.register_blueprint(
+        create_blueprint(app, app.config["WEKO_SEARCH_REST_ENDPOINTS"])
+    )
+
+    file_path = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        "data",
+        "sample_file",
+        "sample_file.txt",
+    )
+
+    with app.test_client() as client:
+        with patch("flask.templating._render", return_value=""):
+            r = client.get(
+                "/",
+                query_string={
+                    "index_id": "33",
+                    "page": 1,
+                    "count": 20,
+                    "term": 14,
+                    "lang": "en",
+                    "parent_id": 33,
+                    "index_info": {},
+                    "community": "comm1",
+                    "item_link": "1",
+                    "is_search": 0,
+                    "search_type": WEKO_SEARCH_TYPE_DICT["INDEX"],
+                    "is_change_identifier": True,
+                    "remote_addr": "0.0.0.0",
+                    "referrer": "test",
+                    "host": "127.0.0.1",
+                },
+            )
+        yield r
+
+@pytest.yield_fixture()
+def client_request_args4(app, file_instance_mock):
+    app.register_blueprint(
+        create_blueprint(app, app.config["WEKO_SEARCH_REST_ENDPOINTS"])
+    )
+
+    file_path = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        "data",
+        "sample_file",
+        "sample_file.txt",
+    )
+
+    with app.test_client() as client:
+        with patch("flask.templating._render", return_value=""):
+            r = client.get(
+                "/",
+                query_string={
+                    "index_id": "33",
+                    "page": 2,
+                    "count": 20,
+                    "term": 14,
+                    "lang": "en",
+                    "parent_id": 33,
+                    "index_info": {},
+                    "community": "comm1",
+                    "item_link": "1",
+                    "is_search": 1,
+                    "search_type": WEKO_SEARCH_TYPE_DICT["INDEX"],
+                    "is_change_identifier": True,
+                    "remote_addr": "0.0.0.0",
+                    "referrer": "test",
+                    "host": "127.0.0.1",
+                },
+            )
+        yield r
+
 @pytest.yield_fixture()
 def client_request_args_FULL_TEXT(app, file_instance_mock):
     app.register_blueprint(