| 1 | import os |
| 2 | import tempfile |
| 3 | import xml.etree.ElementTree as ET |
| 4 | from typing import Optional |
| 5 | from typing import Tuple |
| 6 | from urllib.parse import urlparse |
| 7 | |
| 8 | import aiohttp |
| 9 | from fastapi import HTTPException |
| 10 | from loguru import logger |
| 11 | from sqlalchemy.ext.asyncio import AsyncSession |
| 12 | |
| 13 | from app.active_response.schema.sysmon_config import SysmonConfigDeploymentResult |
| 14 | from app.data_store.data_store_operations import download_sysmon_config |
| 15 | from app.utils import get_connector_attribute |
| 16 | |
| 17 | |
| 18 | async def validate_sysmon_config(xml_content: str) -> None: |
| 19 | """Validate a Sysmon configuration XML string.""" |
| 20 | try: |
| 21 | root = ET.fromstring(xml_content) |
| 22 | |
| 23 | # Verify root element |
| 24 | if root.tag != "Sysmon": |
| 25 | raise HTTPException(status_code=400, detail="Root element must be 'Sysmon'") |
| 26 | |
| 27 | # Verify EventFiltering element |
| 28 | event_filtering = root.find("EventFiltering") |
| 29 | if event_filtering is None: |
| 30 | raise HTTPException(status_code=400, detail="Missing required 'EventFiltering' element") |
| 31 | |
| 32 | # Check for direct text content in EventFiltering |
| 33 | if event_filtering.text and event_filtering.text.strip(): |
| 34 | raise HTTPException(status_code=400, detail="Invalid text content found directly in EventFiltering element") |
| 35 | |
| 36 | # Check for RuleGroup elements |
| 37 | rule_groups = event_filtering.findall("RuleGroup") |
| 38 | if not rule_groups: |
| 39 | raise HTTPException(status_code=400, detail="EventFiltering must contain at least one RuleGroup") |
| 40 | |
| 41 | # Check that each RuleGroup has required attributes |
| 42 | for rule_group in rule_groups: |
| 43 | if not rule_group.attrib.get("groupRelation"): |
| 44 | raise HTTPException(status_code=400, detail="RuleGroup must have 'groupRelation' attribute") |
| 45 | |
| 46 | except ET.ParseError as e: |
| 47 | raise HTTPException(status_code=400, detail=f"Invalid XML syntax: {str(e)}") |
| 48 | |
| 49 | |
| 50 | async def check_config_exists(customer_code: str) -> bool: |
| 51 | """ |
| 52 | Check if a sysmon config exists for the given customer. |
| 53 | Creates the bucket if it doesn't exist. |
| 54 | """ |
| 55 | from app.data_store.data_store_session import create_session |
| 56 | |
| 57 | bucket_name = "sysmon-configs" |
| 58 | object_name = f"{customer_code}/sysmon_config.xml" |
| 59 | |
| 60 | try: |
| 61 | # Get MinIO client |
| 62 | client = await create_session() |
| 63 | |
| 64 | # Create bucket if it doesn't exist |
| 65 | if not await client.bucket_exists(bucket_name): |
| 66 | logger.info(f"Bucket {bucket_name} doesn't exist. Creating it.") |
| 67 | await client.make_bucket(bucket_name) |
| 68 | return False |
| 69 | |
| 70 | # Check if object exists directly using stat_object |
| 71 | try: |
| 72 | await client.stat_object(bucket_name, object_name) |
| 73 | logger.info(f"Found existing sysmon config for customer {customer_code}") |
| 74 | return True |
| 75 | except Exception as e: |
| 76 | # If stat_object fails, the file doesn't exist |
| 77 | logger.info(f"No existing sysmon config found for customer {customer_code} - {str(e)}") |
| 78 | return False |
| 79 | |
| 80 | except Exception as e: |
| 81 | logger.error(f"Error checking if config exists for customer {customer_code}: {str(e)}") |
| 82 | raise HTTPException(status_code=500, detail=f"Error checking config existence: {str(e)}") |
| 83 | |
| 84 | |
| 85 | async def fetch_sysmon_config(customer_code: str) -> bytes: |
| 86 | """Fetch a customer's sysmon config from MinIO storage.""" |
| 87 | try: |
| 88 | config_data = await download_sysmon_config(customer_code) |
| 89 | logger.info(f"Successfully fetched Sysmon config for {customer_code} from storage") |
| 90 | return config_data |
| 91 | except Exception as e: |
| 92 | logger.error(f"Failed to fetch Sysmon config for {customer_code}: {str(e)}") |
| 93 | raise |
| 94 | |
| 95 | |
| 96 | async def save_to_temp_file(data: bytes) -> str: |
| 97 | """Save binary data to a temporary file and return the path.""" |
| 98 | temp_fd, temp_path = tempfile.mkstemp(suffix=".xml") |
| 99 | try: |
| 100 | with os.fdopen(temp_fd, "wb") as temp_file: |
| 101 | temp_file.write(data) |
| 102 | logger.info(f"Saved data to temporary file: {temp_path}") |
| 103 | return temp_path |
| 104 | except Exception as e: |
| 105 | # Clean up if we fail to write |
| 106 | if os.path.exists(temp_path): |
| 107 | os.remove(temp_path) |
| 108 | logger.error(f"Failed to save to temporary file: {str(e)}") |
| 109 | raise |
| 110 | |
| 111 | |
| 112 | async def get_wazuh_endpoint(session: AsyncSession) -> str: |
| 113 | """Get the Wazuh API endpoint for deploying sysmon configs.""" |
| 114 | # Get the base URL from connector |
| 115 | wazuh_api_base_url = await get_connector_attribute(column_name="connector_url", connector_name="Wazuh-Manager", session=session) |
| 116 | |
| 117 | # Parse the URL to extract hostname |
| 118 | parsed_url = urlparse(wazuh_api_base_url) |
| 119 | hostname = parsed_url.netloc.split(":")[0] |
| 120 | |
| 121 | # Create endpoint with correct port and path |
| 122 | endpoint = f"http://{hostname}:5003/provision_worker/sysmon-config" |
| 123 | logger.info(f"Constructed Wazuh API endpoint: {endpoint}") |
| 124 | |
| 125 | return endpoint |
| 126 | |
| 127 | |
| 128 | async def upload_to_wazuh(endpoint: str, customer_code: str, file_path: str) -> Tuple[bool, dict, Optional[str]]: |
| 129 | """Upload a sysmon config file to the Wazuh master.""" |
| 130 | # Replace spaces with underscores |
| 131 | wazuh_customer_code = customer_code.replace(" ", "_") |
| 132 | |
| 133 | logger.info(f"Uploading Sysmon config for customer {wazuh_customer_code} to {endpoint}") |
| 134 | |
| 135 | async with aiohttp.ClientSession() as http_session: |
| 136 | # Create form data |
| 137 | form_data = aiohttp.FormData() |
| 138 | form_data.add_field("customer_code", wazuh_customer_code) |
| 139 | |
| 140 | # Add the file |
| 141 | with open(file_path, "rb") as file_to_upload: |
| 142 | form_data.add_field("sysmon_config", file_to_upload, filename="sysmon_config.xml", content_type="application/xml") |
| 143 | |
| 144 | # Send the POST request |
| 145 | async with http_session.post(endpoint, data=form_data) as response: |
| 146 | if response.status == 200: |
| 147 | response_data = await response.json() |
| 148 | return True, response_data, None |
| 149 | else: |
| 150 | error_text = await response.text() |
| 151 | return False, {}, error_text |
| 152 | |
| 153 | |
| 154 | async def deploy_sysmon_config_to_worker(customer_code: str, session: AsyncSession) -> SysmonConfigDeploymentResult: |
| 155 | """ |
| 156 | Fetch and deploy a customer's sysmon config to the Wazuh master. |
| 157 | |
| 158 | Args: |
| 159 | customer_code: The customer code to deploy config for |
| 160 | session: Database session for connector lookup |
| 161 | |
| 162 | Returns: |
| 163 | SysmonConfigDeploymentResult: Results of the deployment operation |
| 164 | """ |
| 165 | temp_path = None |
| 166 | |
| 167 | try: |
| 168 | # Step 1: Fetch the config file |
| 169 | try: |
| 170 | config_data = await fetch_sysmon_config(customer_code) |
| 171 | except Exception as fetch_error: |
| 172 | return SysmonConfigDeploymentResult( |
| 173 | success=False, |
| 174 | message=f"Failed to fetch Sysmon config for {customer_code}", |
| 175 | customer_code=customer_code, |
| 176 | error_detail=str(fetch_error), |
| 177 | ) |
| 178 | |
| 179 | # Step 2: Save to temporary file |
| 180 | try: |
| 181 | temp_path = await save_to_temp_file(config_data) |
| 182 | except Exception as temp_error: |
| 183 | return SysmonConfigDeploymentResult( |
| 184 | success=False, |
| 185 | message="Failed to create temporary file", |
| 186 | customer_code=customer_code, |
| 187 | error_detail=str(temp_error), |
| 188 | ) |
| 189 | |
| 190 | # Step 3: Get Wazuh endpoint |
| 191 | try: |
| 192 | endpoint = await get_wazuh_endpoint(session) |
| 193 | except Exception as endpoint_error: |
| 194 | return SysmonConfigDeploymentResult( |
| 195 | success=False, |
| 196 | message="Failed to get Wazuh endpoint", |
| 197 | customer_code=customer_code, |
| 198 | error_detail=str(endpoint_error), |
| 199 | ) |
| 200 | |
| 201 | # Step 4: Upload to Wazuh via the agent group shared folder |
| 202 | success, response_data, error_text = await upload_to_wazuh(endpoint, customer_code, temp_path) |
| 203 | |
| 204 | if success: |
| 205 | logger.info(f"Successfully deployed Sysmon config to Wazuh for {customer_code}") |
| 206 | return SysmonConfigDeploymentResult( |
| 207 | success=True, |
| 208 | message=f"Sysmon config successfully deployed for {customer_code}", |
| 209 | customer_code=customer_code, |
| 210 | worker_success=response_data.get("success", True), |
| 211 | worker_message=response_data.get("message", "Deployed successfully"), |
| 212 | ) |
| 213 | else: |
| 214 | logger.error(f"Failed to deploy Sysmon config to Wazuh: {error_text}") |
| 215 | return SysmonConfigDeploymentResult( |
| 216 | success=False, |
| 217 | message="Error from Wazuh master when deploying Sysmon config", |
| 218 | customer_code=customer_code, |
| 219 | error_detail=error_text, |
| 220 | ) |
| 221 | |
| 222 | except Exception as e: |
| 223 | error_msg = f"Error deploying Sysmon config: {str(e)}" |
| 224 | logger.error(error_msg) |
| 225 | return SysmonConfigDeploymentResult( |
| 226 | success=False, |
| 227 | message="Exception occurred during Sysmon config deployment", |
| 228 | customer_code=customer_code, |
| 229 | error_detail=str(e), |
| 230 | ) |
| 231 | |
| 232 | finally: |
| 233 | # Clean up the temporary file |
| 234 | if temp_path and os.path.exists(temp_path): |
| 235 | try: |
| 236 | os.remove(temp_path) |
| 237 | logger.debug(f"Cleaned up temporary file: {temp_path}") |
| 238 | except Exception as cleanup_error: |
| 239 | logger.warning(f"Failed to clean up temporary file {temp_path}: {str(cleanup_error)}") |