mirror of
https://github.com/FunnyWolf/agentic-soc-platform.git
synced 2026-08-22 13:12:56 +02:00
add nocolyapi v0.1
This commit is contained in:
funnywolf
Binary file not shown.
Vendored
+349
@@ -0,0 +1,349 @@
|
||||
from typing import TypedDict, Literal, List, Union, Any
|
||||
|
||||
import requests
|
||||
|
||||
from CONFIG import NOCOLY_URL, AISOAR_APPKEY, AISOAR_SIGN
|
||||
|
||||
|
||||
class Field(TypedDict):
|
||||
id: str
|
||||
name: str
|
||||
alias: str
|
||||
value: str
|
||||
desc: str
|
||||
type: str
|
||||
required: bool
|
||||
isHidden: bool
|
||||
isReadOnly: bool
|
||||
isHiddenOnCreate: bool
|
||||
isUnique: bool
|
||||
isTitle: bool
|
||||
remark: str
|
||||
|
||||
|
||||
class Option(TypedDict):
|
||||
key: str
|
||||
value: str
|
||||
index: int
|
||||
score: float
|
||||
|
||||
|
||||
# Define the recursive types first
|
||||
class Condition(TypedDict):
|
||||
type: Literal["condition"]
|
||||
field: str
|
||||
operator: str
|
||||
value: Any
|
||||
|
||||
|
||||
class Group(TypedDict):
|
||||
type: Literal["group"]
|
||||
logic: Literal["AND", "OR"]
|
||||
children: List[Union["Group", Condition]]
|
||||
|
||||
|
||||
class Worksheet(object):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def get_fields(worksheet_id: str) -> List[Field]:
|
||||
headers = {"HAP-Appkey": AISOAR_APPKEY,
|
||||
"HAP-Sign": AISOAR_SIGN}
|
||||
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}"
|
||||
|
||||
response = requests.get(
|
||||
url,
|
||||
params={"includeSystemFields": True},
|
||||
headers=headers
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
response_data = response.json()
|
||||
if response_data.get("success"):
|
||||
return response_data.get("data").get("fields")
|
||||
else:
|
||||
raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}")
|
||||
|
||||
|
||||
class WorksheetRow(object):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@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}"
|
||||
|
||||
try:
|
||||
response = requests.get(
|
||||
url,
|
||||
params={"includeSystemFields": True},
|
||||
headers=headers
|
||||
)
|
||||
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')}")
|
||||
except Exception as e:
|
||||
raise
|
||||
|
||||
@staticmethod
|
||||
def list(worksheet_id: str, filter: dict):
|
||||
headers = {"HAP-Appkey": AISOAR_APPKEY,
|
||||
"HAP-Sign": AISOAR_SIGN}
|
||||
url = f"{NOCOLY_URL}/api/v3/app/worksheets/{worksheet_id}/rows/list"
|
||||
data = {
|
||||
"filter": filter,
|
||||
"sorts": [
|
||||
{
|
||||
"field": "utime",
|
||||
"direction": "desc"
|
||||
}
|
||||
],
|
||||
"includeTotalCount": True,
|
||||
"includeSystemFields": True,
|
||||
"useFieldIdAsKey": False,
|
||||
}
|
||||
try:
|
||||
response = requests.post(url,
|
||||
headers=headers,
|
||||
json=data)
|
||||
response.raise_for_status()
|
||||
|
||||
response_data = response.json()
|
||||
if response_data.get("success"):
|
||||
return response_data.get("data").get("rows")
|
||||
else:
|
||||
raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}")
|
||||
except Exception as e:
|
||||
raise
|
||||
|
||||
@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"
|
||||
|
||||
data = {
|
||||
"triggerWorkflow": True,
|
||||
"fields": fields
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.post(url,
|
||||
headers=headers,
|
||||
json=data)
|
||||
response.raise_for_status()
|
||||
|
||||
response_data = response.json()
|
||||
if response_data.get("success"):
|
||||
return response_data.get("data").get("id")
|
||||
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
|
||||
|
||||
@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}"
|
||||
|
||||
data = {
|
||||
"triggerWorkflow": True,
|
||||
"fields": fields
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.patch(url,
|
||||
headers=headers,
|
||||
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')}")
|
||||
except Exception as e:
|
||||
raise
|
||||
|
||||
|
||||
class OptionSet(object):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def list():
|
||||
headers = {"HAP-Appkey": AISOAR_APPKEY,
|
||||
"HAP-Sign": AISOAR_SIGN}
|
||||
url = f"{NOCOLY_URL}/api/v3/app/optionsets"
|
||||
|
||||
response = requests.get(url,
|
||||
headers=headers)
|
||||
response.raise_for_status()
|
||||
|
||||
response_data = response.json()
|
||||
if response_data.get("success"):
|
||||
return response_data.get("data").get("optionsets")
|
||||
else:
|
||||
raise Exception(f"error_code: {response_data.get('error_code')} error_msg: {response_data.get('error_msg')}")
|
||||
|
||||
@staticmethod
|
||||
def get(name):
|
||||
optionsets = OptionSet.list()
|
||||
for optionset in optionsets:
|
||||
if optionset["name"] == name:
|
||||
return optionset
|
||||
raise Exception(f"optionset {name} not found")
|
||||
|
||||
@staticmethod
|
||||
def get_option_by_name_and_value(name, value) -> Option:
|
||||
optionsets = OptionSet.list()
|
||||
for optionset in optionsets:
|
||||
if optionset["name"] == name:
|
||||
options = optionset.get("options", [])
|
||||
for option in options:
|
||||
if option["value"] == value:
|
||||
return option
|
||||
raise Exception(f"optionset {name} {value} not found")
|
||||
|
||||
@staticmethod
|
||||
def get_option_key_by_name_and_value(name, value):
|
||||
optionsets = OptionSet.list()
|
||||
for optionset in optionsets:
|
||||
if optionset["name"] == name:
|
||||
options = optionset.get("options", [])
|
||||
for option in options:
|
||||
if option["value"] == value:
|
||||
return option["key"]
|
||||
raise Exception(f"optionset {name} {value} not found")
|
||||
|
||||
|
||||
class Artifact(object):
|
||||
WORKSHEET_ID = "artifact"
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def list(filter: dict):
|
||||
result = WorksheetRow.list(Artifact.WORKSHEET_ID, filter)
|
||||
return result
|
||||
|
||||
@staticmethod
|
||||
def update(rowid, fields: list):
|
||||
row_id = WorksheetRow.update(Artifact.WORKSHEET_ID, rowid, fields)
|
||||
return row_id
|
||||
|
||||
@staticmethod
|
||||
def create(fields: list):
|
||||
row_id = WorksheetRow.create(Artifact.WORKSHEET_ID, fields)
|
||||
return row_id
|
||||
|
||||
@staticmethod
|
||||
def update_by_type_and_value(data: dict):
|
||||
# 第一层必须是group
|
||||
filter = {
|
||||
"type": "group",
|
||||
"logic": "AND",
|
||||
"children": [
|
||||
{
|
||||
"type": "condition",
|
||||
"field": "type",
|
||||
"operator": "eq",
|
||||
"value": data["type"]
|
||||
},
|
||||
{
|
||||
"type": "condition",
|
||||
"field": "value",
|
||||
"operator": "eq",
|
||||
"value": data["value"]
|
||||
}
|
||||
]
|
||||
}
|
||||
rows = Artifact.list(filter)
|
||||
if rows:
|
||||
for row in rows:
|
||||
rowid = row['rowId']
|
||||
fields = [
|
||||
{"id": "enrichment", "value": data["enrichment"]},
|
||||
]
|
||||
rowid_updated = Artifact.update(rowid, fields)
|
||||
return rowid_updated
|
||||
else:
|
||||
fields = [
|
||||
{"id": "type", "value": data["type"], "type": 2},
|
||||
{"id": "value", "value": data["value"]},
|
||||
{"id": "enrichment", "value": data["enrichment"]},
|
||||
]
|
||||
rowid_created = Artifact.create(fields)
|
||||
return rowid_created
|
||||
|
||||
|
||||
class Alert(object):
|
||||
WORKSHEET_ID = "alert"
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def create(fields: list):
|
||||
row_id = WorksheetRow.create(Alert.WORKSHEET_ID, fields)
|
||||
return row_id
|
||||
|
||||
|
||||
class Case(object):
|
||||
WORKSHEET_ID = "case"
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def create(fields: list):
|
||||
row_id = WorksheetRow.create(Case.WORKSHEET_ID, fields)
|
||||
return row_id
|
||||
|
||||
@staticmethod
|
||||
def update(row_id, fields: list):
|
||||
row_id = WorksheetRow.update(Case.WORKSHEET_ID, row_id, fields)
|
||||
return row_id
|
||||
|
||||
@staticmethod
|
||||
def get_by_deduplication_key(deduplication_key: str):
|
||||
filter = {
|
||||
"type": "group",
|
||||
"logic": "AND",
|
||||
"children": [
|
||||
{
|
||||
"type": "condition",
|
||||
"field": "deduplication_key",
|
||||
"operator": "eq",
|
||||
"value": deduplication_key
|
||||
},
|
||||
]
|
||||
}
|
||||
rows = WorksheetRow.list(Case.WORKSHEET_ID, filter)
|
||||
if rows:
|
||||
if len(rows) > 1:
|
||||
raise Exception(f"found multiple cases with deduplication_key {deduplication_key}")
|
||||
return rows[0]
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
class OptionAPI(object):
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def to_value_list(options: list):
|
||||
value_list = []
|
||||
for option in options:
|
||||
value_list.append(option.get("value"))
|
||||
return value_list
|
||||
+62
@@ -0,0 +1,62 @@
|
||||
import datetime
|
||||
import time
|
||||
|
||||
|
||||
def timestamp_to_string(timestamp, format_str: str = "%Y-%m-%d %H:%M:%S") -> str:
|
||||
"""
|
||||
current_timestamp = 1672531200 # 对应 2023-01-01 00:00:00
|
||||
|
||||
# 转换为默认格式的时间字符串
|
||||
time_string_default = timestamp_to_string(current_timestamp)
|
||||
print(f"默认格式: {time_string_default}")
|
||||
|
||||
# 转换为带毫秒的格式
|
||||
time_string_with_ms = timestamp_to_string(current_timestamp, "%Y-%m-%d %H:%M:%S.%f")
|
||||
print(f"带毫秒格式: {time_string_with_ms}")
|
||||
|
||||
# 转换为只包含日期和时区的格式
|
||||
time_string_custom = timestamp_to_string(current_timestamp, "%Y/%m/%d %Z")
|
||||
print(f"自定义格式: {time_string_custom}")
|
||||
"""
|
||||
dt_object = datetime.datetime.fromtimestamp(timestamp)
|
||||
return dt_object.strftime(format_str)
|
||||
|
||||
|
||||
def string_to_timestamp(time_string: str, format_str: str = "%Y-%m-%dT%H:%M:%S") -> int:
|
||||
"""
|
||||
time_string = "2023-01-01 00:00:00"
|
||||
|
||||
timestamp_result = string_to_timestamp(time_string)
|
||||
print(f"时间戳结果: {timestamp_result}")
|
||||
|
||||
time_string_custom = "2023/12/25 10:30:00"
|
||||
timestamp_custom = string_to_timestamp(time_string_custom, "%Y/%m/%d %H:%M:%S")
|
||||
|
||||
time_string = "2025-09-18T14:51:30Z"
|
||||
timestamp_result = string_to_timestamp(time_string, "%Y-%m-%dT%H:%M:%SZ")
|
||||
"""
|
||||
dt_object = datetime.datetime.strptime(time_string, format_str)
|
||||
|
||||
return int(dt_object.timestamp())
|
||||
|
||||
|
||||
def get_current_timestamp() -> int:
|
||||
"""
|
||||
current_ts = get_current_timestamp()
|
||||
print(f"当前时间戳: {current_ts}")
|
||||
"""
|
||||
return int(time.time())
|
||||
|
||||
|
||||
def get_current_time_string(format_str: str = "%Y-%m-%dT%H:%M:%SZ") -> str:
|
||||
"""
|
||||
# 示例
|
||||
# 默认格式
|
||||
current_time_str = get_current_time_string()
|
||||
print(f"当前时间字符串（默认格式）: {current_time_str}")
|
||||
|
||||
# 自定义格式：年-月-日
|
||||
current_date_str = get_current_time_string("%Y-%m-%d")
|
||||
print(f"当前时间字符串（自定义格式）: {current_date_str}")
|
||||
"""
|
||||
return datetime.datetime.now().strftime(format_str)
|
||||
@@ -0,0 +1,75 @@
|
||||
from datetime import datetime, timezone
|
||||
from typing import List, Dict, Any, Optional, Union
|
||||
|
||||
|
||||
class RuleDefinition:
|
||||
"""
|
||||
优化版规则定义。
|
||||
指纹生成逻辑可以处理一个可选的外部时间戳。
|
||||
"""
|
||||
|
||||
def __init__(self,
|
||||
rule_id: str,
|
||||
rule_name: str,
|
||||
deduplication_fields: List[str],
|
||||
case_title_template: str,
|
||||
deduplication_window: str,
|
||||
source: str,
|
||||
):
|
||||
|
||||
self.rule_id = rule_id
|
||||
self.rule_name = rule_name
|
||||
self.deduplication_fields = deduplication_fields
|
||||
self.case_title_template = case_title_template
|
||||
self.source = source
|
||||
valid_windows = ['10m', '30m', '1h', '8h', '12h', '24h']
|
||||
if deduplication_window not in valid_windows:
|
||||
raise ValueError(f"'{deduplication_window}' 不是一个有效的时间窗口选项。请从 {valid_windows} 中选择。")
|
||||
self.deduplication_window = deduplication_window
|
||||
|
||||
@staticmethod
|
||||
def _get_time_bucket(dt_object: datetime, window: str) -> datetime:
|
||||
if window.endswith('m'):
|
||||
minutes = int(window[:-1])
|
||||
new_minute = (dt_object.minute // minutes) * minutes
|
||||
return dt_object.replace(minute=new_minute, second=0, microsecond=0)
|
||||
elif window.endswith('h'):
|
||||
hours = int(window[:-1])
|
||||
if hours == 24:
|
||||
return dt_object.replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
else:
|
||||
new_hour = (dt_object.hour // hours) * hours
|
||||
return dt_object.replace(hour=new_hour, minute=0, second=0, microsecond=0)
|
||||
return dt_object
|
||||
|
||||
def generate_deduplication_key(self,
|
||||
artifacts: List[Dict[str, Any]],
|
||||
timestamp: Optional[Union[int, float]] = None) -> str:
|
||||
"""
|
||||
生成包含“时间桶”的去重指纹。
|
||||
|
||||
:param artifacts: 事件中的凭据列表
|
||||
:param timestamp: (可选) 事件的UTC Unix时间戳 (整数或浮点数)。如果为None，则使用当前系统时间。
|
||||
:return: 包含时间桶的去重指纹
|
||||
"""
|
||||
if timestamp is not None:
|
||||
processing_dt = datetime.fromtimestamp(timestamp, tz=timezone.utc)
|
||||
else:
|
||||
|
||||
processing_dt = datetime.now(timezone.utc)
|
||||
|
||||
time_bucket_dt = self._get_time_bucket(processing_dt, self.deduplication_window)
|
||||
time_bucket_str = time_bucket_dt.strftime('%Y-%m-%dT%H:%M:%S')
|
||||
|
||||
key_parts = [self.rule_id, time_bucket_str]
|
||||
artifacts_map = {art['type']: art['value'] for art in artifacts}
|
||||
for field in sorted(self.deduplication_fields):
|
||||
key_parts.append(artifacts_map.get(field, 'N/A'))
|
||||
|
||||
return "-".join(key_parts)
|
||||
|
||||
def generate_case_title(self, artifacts: List[Dict[str, Any]]) -> str:
|
||||
template_values = {"rule_name": self.rule_name}
|
||||
for art in artifacts:
|
||||
template_values[art['type']] = art['value']
|
||||
return self.case_title_template.format_map(template_values)
|
||||
Reference in New Issue
Block a user