External/agentic-soc-platform
1
0
Fork 0
mirror of https://github.com/FunnyWolf/agentic-soc-platform.git synced 2026-08-22 13:12:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

1017

Browse Source
This commit is contained in:
funnywolf
2025-10-17 22:09:58 +08:00
parent 05c09ee1ae
commit 5e78b50590
10 changed files with 84 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ASF/urls.py
+3 -3
View File
@@ -15,6 +15,6 @@ router.register(r'api/v1/automation/playbook', PlaybookView, basename="PlaybookV
urlpatterns = [
re_path(r'^', include(router.urls)),
]
from Lib.montior import MainMonitor
MainMonitor().start()
# from Lib.montior import MainMonitor
#
# MainMonitor().start()
Automation/Handle/playbook.py
+2 -2
View File
@@ -20,7 +20,7 @@ class Playbook(object):
class_intent = importlib.import_module(load_path)
playbook_intent: BasePlaybook = class_intent.Module(params=params)
except Exception as E:
logger.warning(E)
logger.exception(E)
context = data_return(305, {}, Playbook_MSG_ZH.get(305), Playbook_MSG_EN.get(305))
return context
@@ -29,6 +29,6 @@ class Playbook(object):
context = data_return(201, check_result, Playbook_MSG_ZH.get(201), Playbook_MSG_EN.get(201))
return context
except Exception as E:
logger.warning(E)
logger.exception(E)
context = data_return(301, {}, Playbook_MSG_ZH.get(301), Playbook_MSG_EN.get(301))
return context
Automation/views.py
-1
View File
@@ -13,7 +13,6 @@ class PlaybookView(BaseView):
try:
playbook = request.data.get('playbook')
context = Playbook.create(playbook, params=request.data)
return Response(context)
except Exception as E:
logger.exception(E)
context = data_return(500, {}, CODE_MSG_ZH.get(500), CODE_MSG_EN.get(500))
Lib/External/nocolyapi.py
Vendored
+38 -26
View File
@@ -2,7 +2,7 @@ from typing import TypedDict, Literal, List, Union, Any, Dict, Optional
import requests
from CONFIG import NOCOLY_URL, AISOAR_APPKEY, AISOAR_SIGN
from CONFIG import SIRP_URL, SIRP_APPKEY, SIRP_SIGN
from Lib.api import get_current_time_string, string_to_timestamp
from Lib.ruledefinition import RuleDefinition
@@ -65,10 +65,10 @@ class Worksheet(object):
pass
@staticmethod
def get_fields(worksheet_id: str) -> List[FieldType]:
headers = {"HAP-Appkey": AISOAR_APPKEY,
"HAP-Sign": AISOAR_SIGN}
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}"
def get_fields(worksheet_id: str) -> Dict[str, FieldType]:
headers = {"HAP-Appkey": SIRP_APPKEY,
"HAP-Sign": SIRP_SIGN}
url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}"
response = requests.get(
url,
@@ -79,7 +79,11 @@ class Worksheet(object):
response_data = response.json()
if response_data.get("success"):
return response_data.get("data").get("fields")
fields_list: List[FieldType] = response_data.get("data").get("fields")
fields_dict = {}
for field in fields_list:
fields_dict[field["id"]] = field
return fields_dict
else:
raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}")
@@ -90,10 +94,10 @@ class WorksheetRow(object):
@staticmethod
def get(worksheet_id: str, row_id: str):
headers = {"HAP-Appkey": AISOAR_APPKEY,
"HAP-Sign": AISOAR_SIGN}
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}"
headers = {"HAP-Appkey": SIRP_APPKEY,
"HAP-Sign": SIRP_SIGN}
url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}"
fields = Worksheet.get_fields(worksheet_id)
try:
response = requests.get(
url,
@@ -104,7 +108,15 @@ class WorksheetRow(object):
response_data = response.json()
if response_data.get("success"):
return response_data.get("data")
data = response_data.get("data")
data_new = {}
for key in data:
if key.startswith("_") or key == "rowId":
data_new[key] = data[key]
else:
alias = fields.get(key).get('alias')
data_new[alias] = data[key]
return data_new
else:
raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}")
except Exception as e:
@@ -112,9 +124,9 @@ class WorksheetRow(object):
@staticmethod
def list(worksheet_id: str, filter: dict, fields: list = None):
headers = {"HAP-Appkey": AISOAR_APPKEY,
"HAP-Sign": AISOAR_SIGN}
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}/rows/list"
headers = {"HAP-Appkey": SIRP_APPKEY,
"HAP-Sign": SIRP_SIGN}
url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/list"
data = {
"fields": fields,
"filter": filter,
@@ -145,9 +157,9 @@ class WorksheetRow(object):
@staticmethod
def create(worksheet_id: str, fields: list):
headers = {"HAP-Appkey": AISOAR_APPKEY,
"HAP-Sign": AISOAR_SIGN}
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}/rows"
headers = {"HAP-Appkey": SIRP_APPKEY,
"HAP-Sign": SIRP_SIGN}
url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows"
data = {
"triggerWorkflow": True,
@@ -170,9 +182,9 @@ class WorksheetRow(object):
@staticmethod
def update(worksheet_id: str, row_id: str, fields: list):
headers = {"HAP-Appkey": AISOAR_APPKEY,
"HAP-Sign": AISOAR_SIGN}
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}"
headers = {"HAP-Appkey": SIRP_APPKEY,
"HAP-Sign": SIRP_SIGN}
url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/{row_id}"
data = {
"triggerWorkflow": True,
@@ -195,9 +207,9 @@ class WorksheetRow(object):
@staticmethod
def delete(worksheet_id: str, row_ids: list):
headers = {"HAP-Appkey": AISOAR_APPKEY,
"HAP-Sign": AISOAR_SIGN}
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}/rows/batch"
headers = {"HAP-Appkey": SIRP_APPKEY,
"HAP-Sign": SIRP_SIGN}
url = f"{SIRP_URL}/api/v3/app/worksheets/{worksheet_id}/rows/batch"
data = {
"rowIds": row_ids,
@@ -225,9 +237,9 @@ class OptionSet(object):
@staticmethod
def list():
headers = {"HAP-Appkey": AISOAR_APPKEY,
"HAP-Sign": AISOAR_SIGN}
url = f"{NOCOLY_URL}/api/v3/app/optionsets"
headers = {"HAP-Appkey": SIRP_APPKEY,
"HAP-Sign": SIRP_SIGN}
url = f"{SIRP_URL}/api/v3/app/optionsets"
response = requests.get(url,
headers=headers)
Lib/baseplaybook.py
+4 -2
View File
@@ -1,7 +1,9 @@
from abc import ABC
from Lib.log import logger
class BasePlaybook(object):
class BasePlaybook(ABC):
def __init__(self, params):
super().__init__()
self._params = params
@@ -11,4 +13,4 @@ class BasePlaybook(object):
return self._params.get(key, default)
def run(self):
raise NotImplementedError
pass
MODULES/ES-Rule-21-Phishing_user_report_mail.py
+2 -2
View File
@@ -8,8 +8,8 @@ from langgraph.graph import StateGraph
from langgraph.graph.state import CompiledStateGraph
from pydantic import BaseModel, Field
from Lib.External.llmapi import LLMAPI
from Lib.External.nocolyapi import InputAlert, common_handler
from Lib.External.opanaiapi import OpenAIAPI
from Lib.External.thehiveclient import TheHiveClient
from Lib.api import string_to_string_time, get_current_time_string
from Lib.basemodule import LanggraphModule
@@ -129,7 +129,7 @@ class Module(LanggraphModule):
]
# 运行
openai_api = OpenAIAPI()
openai_api = LLMAPI()
llm = openai_api.get_model()
llm = llm.with_structured_output(AnalyzeResult)
PLAYBOOK/TI_artifact_query_mock.py
+32
View File
@@ -0,0 +1,32 @@
import json
from Lib.External.nocolyapi import WorksheetRow
from Lib.baseplaybook import BasePlaybook
class Module(BasePlaybook):
def __init__(self, params):
super().__init__(params) # do not delete this code
def run(self):
worksheet = self.param("worksheet")
rowid = self.param("rowid")
artifact = WorksheetRow.get(worksheet, rowid)
self.logger.info(f"Querying threat intelligence for : {artifact}")
# 模拟查询威胁情报数据库
# 在实际应用中，这里应该调用外部API或数据库进行查询
if artifact.get("type") not in ["ip", "domain", "hash", "vm_ip"]:
ti_result = {"error": "Unsupported type. Please use 'ip', 'domain', or 'hash'."}
else:
ti_result = {"malicious": True, "score": 85, "description": "This IP is associated with known malicious activities.", "source": "ThreatIntelDB",
"last_seen": "2024-10-01T12:34:56Z"}
ti_result = {"enrichment": json.dumps(ti_result)}
return ti_result
if __name__ == "__main__":
params = {'playbook': 'TI_artifact_query_mock', 'rowid': 'c6caa5eb-b4af-44df-aa10-b94829243fd4', 'worksheet': 'Artifact'}
module = Module(params)
result = module.run()
print(result)
PLAYBOOK/Threat_Intelligence_Query.py
-30
View File
@@ -1,30 +0,0 @@
from Lib.baseplaybook import BasePlaybook
class Module(BasePlaybook):
def __init__(self, params):
super().__init__(params) # do not delete this code
def run(self):
type = self.param("type")
value = self.param("value")
self.logger.info(f"Querying threat intelligence for type: {type}, value: {value}")
# 模拟查询威胁情报数据库
# 在实际应用中，这里应该调用外部API或数据库进行查询
if type not in ["ip", "domain", "hash"]:
result = {"error": "Unsupported type. Please use 'ip', 'domain', or 'hash'."}
else:
result = {"malicious": True, "score": 85, "description": "This IP is associated with known malicious activities.", "source": "ThreatIntelDB",
"last_seen": "2024-10-01T12:34:56Z"}
result = {"enrichment": json.dumps(result)}
return result
if __name__ == "__main__":
import json
params = {'playbook': 'Threat_Intelligence_Query', 'rowid': 'bce78786-9a99-42f1-9c3a-8eb7ffc76bfa', 'worksheet': 'Artifact'}
module = Module(params)
result = module.run()
print(json.dumps(result))
PLAYBOOK/__init__.py
View File
requirements.txt
+3 -1
View File
@@ -24,4 +24,6 @@ pluginbase
rsa
pluginbase
lxml
uvicorn
openpyxl
langchain-ollama
uvicorn