Shuffle singul (#460)
* Enhance organization handling: add detailed response model, update service methods, and introduce utility to retrieve organization ID * precommit-fixes * testing for opencti * Refactor alert creation logic to handle existing assets and notifications for Velociraptor Sigma alerts * precommit-fixes
taylor_socfortress committed
Jun 19, 2025 at 08:31 UTC
294438417bbd4a5b98abba3ab237bd12266e4431
7 files changed
+241
-46
backend/app/connectors/shuffle/routes/organizations.py
+8
-3
@@ -5,6 +5,7 @@ from fastapi import Query
5
from loguru import logger
6
7
from app.auth.utils import AuthHandler
8
+from app.connectors.shuffle.schema.organizations import DetailedOrganizationResponse
9
from app.connectors.shuffle.schema.organizations import OrganizationResponse
10
from app.connectors.shuffle.schema.organizations import OrganizationsListResponse
11
from app.connectors.shuffle.services.organizations import OrganizationsService
@@ -47,7 +48,7 @@ async def list_organizations(connector_name: str = Query("Shuffle", description=
48
49
@shuffle_organizations_router.get(
50
"/organizations/{org_id}",
50
- response_model=OrganizationResponse,
51
+ response_model=DetailedOrganizationResponse,
52
description="Retrieve a specific organization by ID",
53
dependencies=[Depends(auth_handler.require_any_scope("admin", "analyst"))],
54
)
@@ -60,13 +61,17 @@ async def get_organization_by_id(org_id: str, connector_name: str = Query("Shuff
61
connector_name (str): Name of the Shuffle connector to use.
62
63
Returns:
63
- OrganizationResponse: The organization data.
64
+ DetailedOrganizationResponse: The detailed organization data.
65
"""
66
logger.info(f"Request to get organization with ID: {org_id} using connector: {connector_name}")
67
68
try:
69
organization = await OrganizationsService.get_organization_by_id(org_id, connector_name)
69
- return OrganizationResponse(success=True, message=f"Successfully retrieved organization: {organization.name}", data=organization)
70
+ return DetailedOrganizationResponse(
71
+ success=True,
72
+ message=f"Successfully retrieved organization: {organization.name}",
73
+ data=organization,
74
+ )
75
except HTTPException:
76
raise
77
except Exception as e:
backend/app/connectors/shuffle/schema/organizations.py
+46
-5
@@ -1,3 +1,4 @@
1
+from datetime import datetime
2
from typing import Any
3
from typing import Dict
4
from typing import List
@@ -27,14 +28,20 @@ class SSOConfig(BaseModel):
28
openid_token: str = ""
29
30
31
+class OrgAuth(BaseModel):
32
+ token: str = ""
33
+ expires: Optional[datetime] = None
34
+
35
+
36
class Organization(BaseModel):
37
name: str
32
- description: str
38
+ description: Optional[str] = None
39
company_type: str = ""
34
- image: str = ""
40
+ # image: str = ""
41
id: str
36
- org: str
37
- users: List[Any] = []
42
+ org: Optional[str] = None
43
+ org_auth: OrgAuth = Field(default_factory=OrgAuth)
44
+ users: Optional[List[str]] = Field(default_factory=list, description="List of user IDs associated with the organization")
45
role: str = ""
46
roles: List[str] = []
47
active_apps: List[str] = []
@@ -43,7 +50,7 @@ class Organization(BaseModel):
50
sync_config: SyncConfig = Field(default_factory=SyncConfig)
51
sync_features: Dict[str, Any] = Field(default_factory=dict)
52
invites: Optional[Any] = None
46
- child_orgs: List[str] = []
53
+ child_orgs: Optional[List[str]] = None
54
manager_orgs: Optional[List[str]] = None
55
creator_org: Optional[str] = None
56
disabled: bool = False
@@ -66,3 +73,37 @@ class OrganizationResponse(BaseModel):
73
success: bool
74
message: str
75
data: Optional[Organization] = None
76
+
77
+
78
+class DetailedOrganization(BaseModel):
79
+ name: str
80
+ description: str = ""
81
+ company_type: str = ""
82
+ image: str = ""
83
+ id: str
84
+ org: str = ""
85
+ org_auth: OrgAuth = Field(default_factory=OrgAuth)
86
+ users: List[Any] = []
87
+ role: str = ""
88
+ roles: List[str] = []
89
+ active_apps: List[str] = []
90
+ cloud_sync: bool = False
91
+ cloud_sync_active: bool = True
92
+ sync_config: SyncConfig = Field(default_factory=SyncConfig)
93
+ sync_features: Dict[str, Any] = Field(default_factory=dict)
94
+ invites: Optional[Any] = None
95
+ manager_orgs: Optional[List[str]] = None
96
+ creator_org: Optional[str] = None
97
+ disabled: bool = False
98
+ partner_info: PartnerInfo = Field(default_factory=PartnerInfo)
99
+ sso_config: SSOConfig = Field(default_factory=SSOConfig)
100
+ main_priority: str = ""
101
+ region: str = ""
102
+ region_url: str = ""
103
+ tutorials: List[Any] = []
104
+
105
+
106
+class DetailedOrganizationResponse(BaseModel):
107
+ success: bool
108
+ message: str
109
+ data: Optional[DetailedOrganization] = None
backend/app/connectors/shuffle/schema/singul.py
-1
@@ -4,4 +4,3 @@ from pydantic import Field
4
5
class SingulRequest(BaseModel):
6
app: str = Field(..., description="The name of the application", example="outlook_office365")
7
- org_id: str = Field(..., description="The organization ID", example="org_12345")
backend/app/connectors/shuffle/services/organizations.py
+34
-11
@@ -1,6 +1,7 @@
1
from fastapi import HTTPException
2
from loguru import logger
3
4
+from app.connectors.shuffle.schema.organizations import DetailedOrganization
5
from app.connectors.shuffle.schema.organizations import Organization
6
from app.connectors.shuffle.schema.organizations import OrganizationsListResponse
7
from app.connectors.shuffle.utils.universal import send_get_request
@@ -64,29 +65,51 @@ class OrganizationsService:
65
raise HTTPException(status_code=500, detail=f"Unexpected error while fetching organizations: {str(e)}")
66
67
@staticmethod
67
- async def get_organization_by_id(org_id: str, connector_name: str = "Shuffle") -> Organization:
68
+ async def get_organization_by_id(org_id: str, connector_name: str = "Shuffle") -> DetailedOrganization:
69
"""
69
- Get a specific organization by ID.
70
+ Get a specific organization by ID using direct API call.
71
72
Args:
73
org_id (str): The organization ID to retrieve.
74
connector_name (str, optional): The name of the connector to use. Defaults to "Shuffle".
75
76
Returns:
76
- Organization: The organization data.
77
+ DetailedOrganization: The detailed organization data.
78
"""
79
logger.info(f"Fetching organization with ID: {org_id}")
80
80
- # Get all organizations and filter by ID
81
- organizations_response = await OrganizationsService.list_organizations(connector_name)
81
+ try:
82
+ # Send GET request to Shuffle API for specific organization
83
+ response = await send_get_request(endpoint=f"/api/v1/orgs/{org_id}", connector_name=connector_name)
84
83
- for org in organizations_response.data:
84
- if org.id == org_id:
85
- logger.info(f"Found organization: {org.name}")
86
- return org
85
+ if not response.get("success", False):
86
+ logger.error(f"Failed to fetch organization {org_id}: {response.get('message', 'Unknown error')}")
87
+ raise HTTPException(
88
+ status_code=500,
89
+ detail=f"Failed to fetch organization {org_id}: {response.get('message', 'Unknown error')}",
90
+ )
91
+
92
+ # Parse the response data
93
+ organization_data = response.get("data", {})
94
+
95
+ if not organization_data:
96
+ logger.error(f"Organization with ID {org_id} not found")
97
+ raise HTTPException(status_code=404, detail=f"Organization with ID {org_id} not found")
98
88
- logger.error(f"Organization with ID {org_id} not found")
89
- raise HTTPException(status_code=404, detail=f"Organization with ID {org_id} not found")
99
+ try:
100
+ organization = DetailedOrganization(**organization_data)
101
+ logger.info(f"Found organization: {organization.name}")
102
+ return organization
103
+ except Exception as e:
104
+ logger.error(f"Failed to parse organization data for ID {org_id}: {e}")
105
+ raise HTTPException(status_code=500, detail=f"Failed to parse organization data: {str(e)}")
106
+
107
+ except HTTPException:
108
+ # Re-raise HTTPExceptions as-is
109
+ raise
110
+ except Exception as e:
111
+ logger.error(f"Unexpected error while fetching organization {org_id}: {e}")
112
+ raise HTTPException(status_code=500, detail=f"Unexpected error while fetching organization {org_id}: {str(e)}")
113
114
@staticmethod
115
async def get_organization_by_name(org_name: str, connector_name: str = "Shuffle") -> Organization:
backend/app/connectors/shuffle/services/singul.py
+43
-8
@@ -1,8 +1,47 @@
1
from loguru import logger
2
3
from app.connectors.shuffle.schema.singul import SingulRequest
4
+from app.connectors.shuffle.utils.universal import get_shuffle_org_id
5
from app.connectors.shuffle.utils.universal import get_singul_client
6
7
+# async def execute_singul(
8
+# request: SingulRequest,
9
+# ) -> dict:
10
+# """
11
+# Execute a Singul integration.
12
+
13
+# Args:
14
+# request (SingulRequest): The request object containing the workflow ID.
15
+
16
+# Returns:
17
+# dict: The response containing the execution ID.
18
+# """
19
+# logger.info("Executing Singul integration")
20
+
21
+# # Get Singul client from database credentials
22
+# singul = await get_singul_client()
23
+
24
+# try:
25
+# response = singul.communication.send_message(
26
+# app=request.app,
27
+# org_id=await get_shuffle_org_id(),
28
+# fields=[
29
+# {"key": "to", "value": "walton.taylor23@gmail.com"},
30
+# {"key": "subject", "value": "Test Email from Singul"},
31
+# {"key": "body", "value": "This is a test email sent from Singul."},
32
+# ],
33
+# )
34
+# logger.info(f"Singul response: {response}")
35
+# logger.info(f"Singul response success: {response.get('success', 'unknown')}")
36
+
37
+# return {
38
+# "executionId": response.get("id", "unknown"),
39
+# "message": "Singul integration executed successfully",
40
+# }
41
+# except Exception as e:
42
+# logger.error(f"Failed to execute Singul integration: {e}")
43
+# return {"executionId": "unknown", "message": f"Singul integration failed: {e}", "success": False}
44
+
45
46
async def execute_singul(
47
request: SingulRequest,
@@ -22,14 +61,10 @@ async def execute_singul(
61
singul = await get_singul_client()
62
63
try:
25
- response = singul.communication.send_message(
26
- app=request.app,
27
- org_id=request.org_id,
28
- fields=[
29
- {"key": "to", "value": "walton.taylor23@gmail.com"},
30
- {"key": "subject", "value": "Test Email from Singul"},
31
- {"key": "body", "value": "This is a test email sent from Singul."},
32
- ],
64
+ response = singul.intel.get_ioc(
65
+ app="opencti_dcon",
66
+ org_id=await get_shuffle_org_id(),
67
+ fields=[{"key": "ip", "value": "8.8.8.8"}],
68
)
69
logger.info(f"Singul response: {response}")
70
logger.info(f"Singul response success: {response.get('success', 'unknown')}")
backend/app/connectors/shuffle/utils/universal.py
+17
@@ -69,6 +69,23 @@ async def verify_shuffle_connection(connector_name: str) -> str:
69
return await verify_shuffle_credentials(attributes)
70
71
72
+async def get_shuffle_org_id() -> Optional[str]:
73
+ """
74
+ Retrieves the organization ID from the Shuffle service.
75
+
76
+ Returns:
77
+ Optional[str]: The organization ID if found, otherwise None.
78
+ """
79
+ logger.info("Retrieving Shuffle organization ID")
80
+ async with get_db_session() as session: # This will correctly enter the context manager
81
+ attributes = await get_connector_info_from_db("Shuffle", session)
82
+ if attributes is None:
83
+ logger.error("No Shuffle connector found in the database")
84
+ return None
85
+
86
+ return attributes.get("connector_extra_data", None)
87
+
88
+
89
async def send_get_request(
90
endpoint: str,
91
params: Optional[Dict[str, Any]] = None,
backend/app/incidents/services/incident_alert.py
+93
-18
@@ -554,11 +554,13 @@ async def handle_customer_notifications(
554
)
555
556
557
+# ! OLD FUNCTION ! #
558
# async def create_alert_full(
559
# alert_payload: CreatedAlertPayload,
560
# customer_code: str,
561
# session: AsyncSession,
562
# threshold_alert: bool = False,
563
+# velo_sigma_alert: bool = False,
564
# ) -> Alert:
565
# """
566
# Create an alert in CoPilot.
@@ -567,6 +569,8 @@ async def handle_customer_notifications(
569
# alert_payload (dict): The alert payload.
570
# customer_code (str): The customer code.
571
# session (AsyncSession): The database session.
572
+# threshold_alert (bool, optional): Whether this is a threshold alert. Defaults to False.
573
+# velo_sigma_alert (bool, optional): Whether this is a Velociraptor Sigma alert. Defaults to False.
574
575
# Returns:
576
# CreateAlertResponse: The response object containing the created alert details.
@@ -574,6 +578,66 @@ async def handle_customer_notifications(
578
# Raises:
579
# HTTPException: If there is an error creating the alert.
580
# """
581
+# # For velo_sigma_alert, check if an open alert with the same title already exists
582
+# if velo_sigma_alert:
583
+# existing_alert_id = await open_alert_exists(alert_payload, customer_code, session)
584
+# if existing_alert_id:
585
+# logger.info(
586
+# f"Found existing open alert ID {existing_alert_id} for Velociraptor Sigma alert with title {alert_payload.alert_title_payload}",
587
+# )
588
+
589
+# # Add the asset to the existing alert if it doesn't already exist
590
+# asset_exists = await does_assit_exist(alert_payload, existing_alert_id, session)
591
+# if not asset_exists and alert_payload.asset_payload:
592
+# logger.info(f"Adding asset {alert_payload.asset_payload} to existing alert ID {existing_alert_id}")
593
+# await add_asset_to_copilot_alert(
594
+# alert_payload=alert_payload,
595
+# alert_id=existing_alert_id,
596
+# customer_code=customer_code,
597
+# session=session,
598
+# )
599
+
600
+# # Add IOC if present and doesn't already exist
601
+# if alert_payload.ioc_payload is not None:
602
+# ioc_exists = await does_ioc_exist(alert_payload, existing_alert_id, session)
603
+# if not ioc_exists:
604
+# logger.info(f"Adding IOC {alert_payload.ioc_payload['ioc_value']} to existing alert ID {existing_alert_id}")
605
+# await add_ioc_to_copilot_alert(
606
+# alert_payload=alert_payload,
607
+# alert_id=existing_alert_id,
608
+# customer_code=customer_code,
609
+# session=session,
610
+# )
611
+
612
+# # Update the document reference if needed
613
+# if alert_payload.index_name and alert_payload.index_id:
614
+# await add_alert_to_document(
615
+# CreateAlertRequest(index_name=alert_payload.index_name, alert_id=alert_payload.index_id),
616
+# existing_alert_id,
617
+# )
618
+
619
+# # Set alert ID for notifications
620
+# alert_payload.alert_id = existing_alert_id
621
+
622
+# # Handle customer notifications
623
+# if alert_payload.asset_payload:
624
+# await handle_customer_notifications(
625
+# customer_code=customer_code,
626
+# asset_name=alert_payload.asset_payload,
627
+# alert_payload=alert_payload,
628
+# session=session,
629
+# )
630
+# else:
631
+# await handle_customer_notifications(
632
+# customer_code=customer_code,
633
+# asset_name="No asset found",
634
+# alert_payload=alert_payload,
635
+# session=session,
636
+# )
637
+
638
+# return existing_alert_id
639
+
640
+# # If not velo_sigma_alert or no existing alert found, proceed with normal alert creation
641
# alert_id = (await create_alert_in_copilot(alert_payload=alert_payload, customer_code=customer_code, session=session)).id
642
# alert_context_id = (
643
# await create_alert_context_payload(source=alert_payload.source, alert_payload=alert_payload.alert_context_payload, session=session)
@@ -618,8 +682,10 @@ async def handle_customer_notifications(
682
# session=session,
683
# )
684
621
-# if threshold_alert is True:
622
-# logger.info(f"Threshold alert created for customer code {customer_code} with alert ID {alert_id}")
685
+# if threshold_alert is True or velo_sigma_alert is True:
686
+# logger.info(
687
+# f"{'Threshold' if threshold_alert else 'Velociraptor Sigma'} alert created for customer code {customer_code} with alert ID {alert_id}",
688
+# )
689
# return alert_id
690
691
# await add_alert_to_document(CreateAlertRequest(index_name=alert_payload.index_name, alert_id=alert_payload.index_id), alert_id)
@@ -627,6 +693,7 @@ async def handle_customer_notifications(
693
# return alert_id
694
695
696
+# ! NEW FUNCTION ! #
697
async def create_alert_full(
698
alert_payload: CreatedAlertPayload,
699
customer_code: str,
@@ -658,9 +725,13 @@ async def create_alert_full(
725
f"Found existing open alert ID {existing_alert_id} for Velociraptor Sigma alert with title {alert_payload.alert_title_payload}",
726
)
727
728
+ # Check if the asset already exists for this alert (to determine if we should skip notifications)
729
+ asset_already_exists = False
730
+ if alert_payload.asset_payload:
731
+ asset_already_exists = await does_assit_exist(alert_payload, existing_alert_id, session)
732
+
733
# Add the asset to the existing alert if it doesn't already exist
662
- asset_exists = await does_assit_exist(alert_payload, existing_alert_id, session)
663
- if not asset_exists and alert_payload.asset_payload:
734
+ if not asset_already_exists and alert_payload.asset_payload:
735
logger.info(f"Adding asset {alert_payload.asset_payload} to existing alert ID {existing_alert_id}")
736
await add_asset_to_copilot_alert(
737
alert_payload=alert_payload,
@@ -691,21 +762,25 @@ async def create_alert_full(
762
# Set alert ID for notifications
763
alert_payload.alert_id = existing_alert_id
764
694
- # Handle customer notifications
695
- if alert_payload.asset_payload:
696
- await handle_customer_notifications(
697
- customer_code=customer_code,
698
- asset_name=alert_payload.asset_payload,
699
- alert_payload=alert_payload,
700
- session=session,
701
- )
765
+ # Handle customer notifications only if the asset didn't already exist
766
+ if not asset_already_exists:
767
+ logger.info(f"Sending notifications for new asset {alert_payload.asset_payload} in existing alert {existing_alert_id}")
768
+ if alert_payload.asset_payload:
769
+ await handle_customer_notifications(
770
+ customer_code=customer_code,
771
+ asset_name=alert_payload.asset_payload,
772
+ alert_payload=alert_payload,
773
+ session=session,
774
+ )
775
+ else:
776
+ await handle_customer_notifications(
777
+ customer_code=customer_code,
778
+ asset_name="No asset found",
779
+ alert_payload=alert_payload,
780
+ session=session,
781
+ )
782
else:
703
- await handle_customer_notifications(
704
- customer_code=customer_code,
705
- asset_name="No asset found",
706
- alert_payload=alert_payload,
707
- session=session,
708
- )
783
+ logger.info(f"Skipping notifications for existing asset {alert_payload.asset_payload} in alert {existing_alert_id}")
784
785
return existing_alert_id
786