diff --git a/src/spaceone/board/error/post.py b/src/spaceone/board/error/post.py index 61b9ec4..e8a3943 100644 --- a/src/spaceone/board/error/post.py +++ b/src/spaceone/board/error/post.py @@ -22,3 +22,7 @@ class ERROR_SMTP_CONNECTION_FAILED(ERROR_UNKNOWN): class ERROR_NOTICE_EMAIL_ALREADY_SENT(ERROR_UNKNOWN): _message = "Notice email has already been sent. please try again later. (post_id = {post_id})" + + +class ERROR_INVALID_CONTENTS(ERROR_INVALID_ARGUMENT): + _message = "Content is invalid. please check the content." diff --git a/src/spaceone/board/service/post_service.py b/src/spaceone/board/service/post_service.py index f958d5d..25eaee8 100644 --- a/src/spaceone/board/service/post_service.py +++ b/src/spaceone/board/service/post_service.py @@ -1,5 +1,7 @@ import copy +import html import logging +import re from typing import Tuple from spaceone.core import config, cache @@ -80,6 +82,9 @@ def create(self, params: dict) -> Post: _options.update({"is_popup": options["is_popup"]}) params["options"] = _options + if contents := params.get("contents"): + params["contents"] = self._check_contents(contents) + file_ids = params.get("files", []) self.file_mgr: FileManager = self.locator.get_manager(FileManager) self._check_files(file_ids) @@ -131,6 +136,9 @@ def update(self, params: dict) -> Post: params["options"] = _options + if contents := params.get("contents"): + params["contents"] = self._check_contents(contents) + if "files" in params: self.file_mgr: FileManager = self.locator.get_manager(FileManager) @@ -190,13 +198,15 @@ def send(self, params: dict) -> None: users_emails = self._get_verified_user_emails_from_domain(domain_id) verified_user_emails_info[language].extend(users_emails) + for language, verified_user_emails in verified_user_emails_info.items(): + verified_user_emails_info[language] = list(set(verified_user_emails)) + elif post_vo.resource_group == "DOMAIN": language = self._get_language_from_domain_config(domain_id) if language not in verified_user_emails_info: verified_user_emails_info[language] = [] users_emails = self._get_verified_user_emails_from_domain(domain_id) - users_emails = list(set(users_emails)) verified_user_emails_info[language].extend(users_emails) else: if post_vo.workspaces: @@ -440,14 +450,16 @@ def _get_verified_user_emails_from_domain(self, domain_id: str) -> list: } response = self.identity_mgr.list_users(domain_id, {"query": query_filter}) users_info = response.get("results", []) - total_count = response.get("total_count", 0) user_emails.extend([user["email"] for user in users_info]) + verified_user_emails = list(set(user_emails)) + total_count = len(verified_user_emails) + _LOGGER.debug( f"[_get_verified_user_emails_from_domain] user email count: {total_count} in domain_id: {domain_id}" ) - return user_emails + return verified_user_emails def _get_enabled_state_workspace_ids_from_post_vo_workspaces( self, domain_id: str, workspace_ids: list @@ -544,3 +556,42 @@ def _valid_options(options: dict) -> None: for key in options.keys(): if key not in supported_options_key: raise ERROR_INVALID_KEY_IN_OPTIONS(key=key) + + @staticmethod + def _check_contents(contents: str) -> str: + js_patterns = [ + r")<[^<]*)*<\/script>", #