From b23f803f8ffed0b0c3776c00a538e561561da536 Mon Sep 17 00:00:00 2001 From: rootkit Date: Fri, 23 Jan 2026 00:58:19 +0800 Subject: [PATCH] add api --- PLAYBOOKS/Artifact_TI_Enrichment_By_Mock.py | 6 +- PLUGINS/Mock/SIRP/delete_all.py | 2 +- PLUGINS/SIRP/nocolyapi.py | 161 ++++++++++++++++---- PLUGINS/SIRP/sirpbase.py | 6 +- 4 files changed, 140 insertions(+), 35 deletions(-) diff --git a/PLAYBOOKS/Artifact_TI_Enrichment_By_Mock.py b/PLAYBOOKS/Artifact_TI_Enrichment_By_Mock.py index 374e62c..8f9d31d 100644 --- a/PLAYBOOKS/Artifact_TI_Enrichment_By_Mock.py +++ b/PLAYBOOKS/Artifact_TI_Enrichment_By_Mock.py @@ -1,5 +1,4 @@ import json -import time from Lib.baseplaybook import BasePlaybook from PLUGINS.SIRP.sirpapi import Artifact @@ -17,8 +16,6 @@ class Playbook(BasePlaybook): artifact = Artifact.get(self.param_source_rowid) # Simulate querying a threat intelligence database. In a real application, this should call an external API or database. - time.sleep(5) - if artifact.type not in ["IP Address", "Hash"]: self.update_playbook_status(PlaybookJobStatus.FAILED, "Unsupported type. Please use 'IP Address', 'Hash'.") return @@ -41,6 +38,7 @@ class Playbook(BasePlaybook): if __name__ == "__main__": - PlaybookModel(source_worksheet='Artifact', source_rowid='a966036e-b29e-4449-be48-23293bacac5d') + model = PlaybookModel(source_worksheet='Artifact', source_rowid='a966036e-b29e-4449-be48-23293bacac5d') module = Playbook() + module._playbook_model = model module.run() diff --git a/PLUGINS/Mock/SIRP/delete_all.py b/PLUGINS/Mock/SIRP/delete_all.py index dc08cb3..41c1748 100644 --- a/PLUGINS/Mock/SIRP/delete_all.py +++ b/PLUGINS/Mock/SIRP/delete_all.py @@ -26,4 +26,4 @@ if __name__ == "__main__": for row in rows: row_ids.append(row["rowid"]) if row_ids: - WorksheetRow.delete(worksheet_id=worksheet_id, row_ids=row_ids) + WorksheetRow.batch_delete(worksheet_id=worksheet_id, row_ids=row_ids) diff --git a/PLUGINS/SIRP/nocolyapi.py b/PLUGINS/SIRP/nocolyapi.py index dd15ec4..8805dd1 100644 --- a/PLUGINS/SIRP/nocolyapi.py +++ b/PLUGINS/SIRP/nocolyapi.py @@ -67,25 +67,6 @@ class WorksheetRow(object): def __init__(self): pass - @staticmethod - def get(worksheet_id: str, row_id: str, include_system_fields=True) -> dict: - url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}" - fields = Worksheet.get_fields(worksheet_id) - response = HTTP_SESSION.get( - url, - timeout=SIRP_REQUEST_TIMEOUT, - params={"includeSystemFields": include_system_fields} - ) - response.raise_for_status() - - response_data = response.json() - if response_data.get("success"): - row = response_data.get("data") - data_new = WorksheetRow._format_input_row(row, fields, include_system_fields) - return data_new - else: - raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") - @staticmethod def _format_input_row(row, fields, include_system_fields=True) -> dict: data_new = {} @@ -175,6 +156,25 @@ class WorksheetRow(object): value_to_key = {opt["value"]: opt["key"] for opt in target_field["options"]} filter_data["value"] = [value_to_key.get(v, v) for v in filter_data["value"]] + @staticmethod + def get(worksheet_id: str, row_id: str, include_system_fields=True) -> dict: + url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}" + fields = Worksheet.get_fields(worksheet_id) + response = HTTP_SESSION.get( + url, + timeout=SIRP_REQUEST_TIMEOUT, + params={"includeSystemFields": include_system_fields} + ) + response.raise_for_status() + + response_data = response.json() + if response_data.get("success"): + row = response_data.get("data") + data_new = WorksheetRow._format_input_row(row, fields, include_system_fields) + return data_new + else: + raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") + @staticmethod def list(worksheet_id: str, filter: dict, include_system_fields=True) -> List: url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/list" @@ -227,9 +227,6 @@ class WorksheetRow(object): @staticmethod def create(worksheet_id: str, fields: List, trigger_workflow: bool = True): - # fields = [ - # field for field in fields if field.get("value") is not None and field.get("value") != "" - # ] url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows" fields_config = Worksheet.get_fields(worksheet_id) @@ -256,10 +253,6 @@ class WorksheetRow(object): @staticmethod def update(worksheet_id: str, row_id: str, fields: List, trigger_workflow: bool = True): - # fields = [ - # field for field in fields if field.get("value") is not None - # ] - fields_config = Worksheet.get_fields(worksheet_id) fields = WorksheetRow._format_output_value(fields_config, fields) url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}" @@ -280,7 +273,90 @@ class WorksheetRow(object): raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") @staticmethod - def delete(worksheet_id: str, row_ids: List, trigger_workflow: bool = True): + def batch_create(worksheet_id: str, + rows: List[List[Dict]], + trigger_workflow: bool = True) -> Dict: + + url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/batch" + fields_config = Worksheet.get_fields(worksheet_id) + + formatted_rows = [] + for row_fields in rows: + formatted_fields = WorksheetRow._format_output_value(fields_config, row_fields) + formatted_rows.append({"fields": formatted_fields}) + + data = { + "rows": formatted_rows, + "triggerWorkflow": trigger_workflow + } + + try: + response = HTTP_SESSION.post(url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) + response.raise_for_status() + + response_data = response.json() + if response_data.get("success"): + return response_data.get("data") # {"rowIds":[rowid1,rowid2]} + else: + raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')} data: {response_data.get('data')}") + except Exception as e: + raise e + + @staticmethod + def batch_update(worksheet_id: str, + row_ids: List[str], + fields: List[Dict], + trigger_workflow: bool = True) -> Dict: + + url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/batch" + fields_config = Worksheet.get_fields(worksheet_id) + + fields = WorksheetRow._format_output_value(fields_config, fields) + + data = { + "rowIds": row_ids, + "fields": fields, + "triggerWorkflow": trigger_workflow + } + + try: + response = HTTP_SESSION.patch(url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) + response.raise_for_status() + + response_data = response.json() + if response_data.get("success"): + return response_data.get("data") # {"failedRowIds":[rowid1,rowid2], "successfulRowIds":[rowid3,rowid4]} + else: + raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')} data: {response_data.get('data')}") + except Exception as e: + raise e + + @staticmethod + def delete(worksheet_id: str, rowid: List, trigger_workflow: bool = True): + + url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{rowid}" + + data = { + "triggerWorkflow": trigger_workflow, + } + + response = HTTP_SESSION.delete(url, + timeout=SIRP_REQUEST_TIMEOUT, + json=data) + response.raise_for_status() + + response_data = response.json() + if response_data.get("success"): + return response_data.get("data") + else: + raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") + + @staticmethod + def batch_delete(worksheet_id: str, row_ids: List, trigger_workflow: bool = True): url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/batch" @@ -300,6 +376,37 @@ class WorksheetRow(object): else: raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") + @staticmethod + def get_discussions(worksheet_id: str, row_id: str) -> List[Dict]: + url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}/discussions" + + response = HTTP_SESSION.get(url, + timeout=SIRP_REQUEST_TIMEOUT) + response.raise_for_status() + + response_data = response.json() + if response_data.get("success"): + # [ + # { + # "id": "讨论ID", + # "message": "讨论内容", + # "_createdAt": "创建时间", + # "_createdBy": {"id", "name", "avatar", "isPortal", "status"}, + # "replyToAuthor": {"id", "name", "avatar", "isPortal", "status"}, + # "replyId": "回复ID", + # "projectId": "项目ID", + # "replyToId": [], + # "mentions": [], + # "attachments": [ + # {"originalFilename", "ext", "filesize", "downloadUrl"} + # ] + # } + # ] + discussions = response_data.get("data", {}).get("discussions", []) + return discussions + else: + raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}") + @staticmethod def relations(worksheet_id: str, row_id: str, field: str, relation_worksheet_id: str, include_system_fields: bool = True, page_size: int = 1000, page_index: int = 1): diff --git a/PLUGINS/SIRP/sirpbase.py b/PLUGINS/SIRP/sirpbase.py index b6bcb75..c169260 100644 --- a/PLUGINS/SIRP/sirpbase.py +++ b/PLUGINS/SIRP/sirpbase.py @@ -65,21 +65,21 @@ class BaseWorksheetEntity(ABC, Generic[T]): @classmethod def list( cls, - model: Group, + filter_model: Group, include_system_fields: bool = True, lazy_load: bool = False ) -> List[T]: """按过滤条件列表查询 Args: - model: 过滤条件Group对象 + filter_model: 过滤条件Group对象 include_system_fields: 是否包含系统字段 lazy_load: 是否延迟加载关联数据（True时不加载关联） Returns: 模型实例列表 """ - filter_dict = model.model_dump() + filter_dict = filter_model.model_dump() result = WorksheetRow.list( cls.WORKSHEET_ID, filter_dict,