| 1 | # App specific imports |
| 2 | from typing import List |
| 3 | from typing import Optional |
| 4 | |
| 5 | from fastapi import APIRouter |
| 6 | from fastapi import Depends |
| 7 | from fastapi import File |
| 8 | from fastapi import HTTPException |
| 9 | from fastapi import Path |
| 10 | from fastapi import Query |
| 11 | from fastapi import Security |
| 12 | from fastapi import UploadFile |
| 13 | from loguru import logger |
| 14 | from sqlalchemy.ext.asyncio import AsyncSession |
| 15 | from sqlalchemy.future import select |
| 16 | |
| 17 | from app.auth.routes.auth import AuthHandler |
| 18 | from app.connectors.wazuh_manager.models.rules import DisabledRule |
| 19 | |
| 20 | # from app.connectors.wazuh_manager.schema.rules import RuleExclude |
| 21 | from app.connectors.wazuh_manager.schema.rules import AllDisabledRuleResponse |
| 22 | from app.connectors.wazuh_manager.schema.rules import RuleDisable |
| 23 | from app.connectors.wazuh_manager.schema.rules import RuleDisableResponse |
| 24 | from app.connectors.wazuh_manager.schema.rules import RuleEnable |
| 25 | from app.connectors.wazuh_manager.schema.rules import RuleEnableResponse |
| 26 | from app.connectors.wazuh_manager.schema.rules import RuleExcludeRequest |
| 27 | from app.connectors.wazuh_manager.schema.rules import RuleExcludeResponse |
| 28 | from app.connectors.wazuh_manager.schema.rules import WazuhRuleFileContentResponse |
| 29 | from app.connectors.wazuh_manager.schema.rules import WazuhRuleFilesResponse |
| 30 | from app.connectors.wazuh_manager.schema.rules import WazuhRuleFileUploadResponse |
| 31 | from app.connectors.wazuh_manager.schema.rules import WazuhRulesResponse |
| 32 | |
| 33 | # from app.connectors.wazuh_manager.schema.rules import RuleExclude |
| 34 | # from app.connectors.wazuh_manager.schema.rules import RuleExcludeResponse |
| 35 | from app.connectors.wazuh_manager.services.rules import disable_rule |
| 36 | from app.connectors.wazuh_manager.services.rules import enable_rule |
| 37 | from app.connectors.wazuh_manager.services.rules import get_wazuh_rule_file_content |
| 38 | from app.connectors.wazuh_manager.services.rules import get_wazuh_rule_files |
| 39 | from app.connectors.wazuh_manager.services.rules import get_wazuh_rules |
| 40 | from app.connectors.wazuh_manager.services.rules import post_to_copilot_ai_module |
| 41 | from app.connectors.wazuh_manager.services.rules import update_wazuh_rule_file |
| 42 | |
| 43 | # from app.connectors.wazuh_manager.services.rules import exclude_rule |
| 44 | from app.db.db_session import get_db |
| 45 | |
| 46 | # from app.connectors.wazuh_manager.services.rules import exclude_rule |
| 47 | |
| 48 | |
| 49 | NEW_LEVEL = "1" |
| 50 | wazuh_manager_rules_router = APIRouter() |
| 51 | auth_handler = AuthHandler() |
| 52 | |
| 53 | |
| 54 | @wazuh_manager_rules_router.get( |
| 55 | "/rule/disabled", |
| 56 | response_model=AllDisabledRuleResponse, |
| 57 | description="Get all disabled rules", |
| 58 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 59 | ) |
| 60 | async def get_disabled_rules( |
| 61 | session: AsyncSession = Depends(get_db), |
| 62 | ) -> AllDisabledRuleResponse: |
| 63 | """ |
| 64 | Retrieve all disabled rules from the database. |
| 65 | |
| 66 | Parameters: |
| 67 | - session: The database session to use (AsyncSession). |
| 68 | |
| 69 | Returns: |
| 70 | - AllDisabledRuleResponse: The response containing the list of disabled rules (AllDisabledRuleResponse). |
| 71 | """ |
| 72 | result = await session.execute(select(DisabledRule)) |
| 73 | disabled_rules = result.scalars().all() |
| 74 | return AllDisabledRuleResponse( |
| 75 | disabled_rules=disabled_rules, |
| 76 | success=True, |
| 77 | message="Successfully fetched all disabled rules", |
| 78 | ) |
| 79 | |
| 80 | |
| 81 | @wazuh_manager_rules_router.post( |
| 82 | "/rule/disable", |
| 83 | response_model=RuleDisableResponse, |
| 84 | description="Disable a Wazuh Rule", |
| 85 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 86 | ) |
| 87 | async def disable_wazuh_rule( |
| 88 | rule: RuleDisable, |
| 89 | session: AsyncSession = Depends(get_db), |
| 90 | username: str = Depends(AuthHandler().get_current_user), |
| 91 | ) -> RuleDisableResponse: |
| 92 | """ |
| 93 | Disable a Wazuh Rule. |
| 94 | |
| 95 | Args: |
| 96 | rule (RuleDisable): The rule to be disabled. |
| 97 | session (AsyncSession, optional): The database session. Defaults to Depends(get_db). |
| 98 | username (str, optional): The username of the current user. Defaults to Depends(AuthHandler().get_current_user). |
| 99 | |
| 100 | Returns: |
| 101 | RuleDisableResponse: The response containing the disabled rule information. |
| 102 | |
| 103 | Raises: |
| 104 | HTTPException: If the rule is already disabled or if the rule cannot be disabled. |
| 105 | """ |
| 106 | # Asynchronously check if the rule is already disabled |
| 107 | result = await session.execute( |
| 108 | select(DisabledRule).where(DisabledRule.rule_id == rule.rule_id), |
| 109 | ) |
| 110 | if result.scalars().first(): |
| 111 | raise HTTPException(status_code=500, detail="Rule is already disabled") |
| 112 | |
| 113 | # This should be converted to an async operation if it's not already |
| 114 | rule_disabled = await disable_rule(rule) |
| 115 | if rule_disabled: |
| 116 | new_disabled_rule = DisabledRule( |
| 117 | rule_id=rule.rule_id, |
| 118 | previous_level=rule_disabled.previous_level, |
| 119 | new_level=NEW_LEVEL, |
| 120 | reason_for_disabling=rule.reason_for_disabling, |
| 121 | length_of_time=rule.length_of_time, |
| 122 | disabled_by=username.username, |
| 123 | ) |
| 124 | session.add(new_disabled_rule) |
| 125 | await session.commit() |
| 126 | return rule_disabled |
| 127 | else: |
| 128 | raise HTTPException(status_code=404, detail="Was not able to disable rule") |
| 129 | |
| 130 | |
| 131 | @wazuh_manager_rules_router.post( |
| 132 | "/rule/enable", |
| 133 | response_model=RuleEnableResponse, |
| 134 | description="Enable a Wazuh Rule", |
| 135 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 136 | ) |
| 137 | async def enable_wazuh_rule( |
| 138 | rule: RuleEnable, |
| 139 | session: AsyncSession = Depends(get_db), |
| 140 | ) -> RuleEnableResponse: |
| 141 | """ |
| 142 | Enable a Wazuh rule. |
| 143 | |
| 144 | Args: |
| 145 | rule (RuleEnable): The rule to enable. |
| 146 | session (AsyncSession, optional): The database session. Defaults to Depends(get_db). |
| 147 | |
| 148 | Returns: |
| 149 | RuleEnableResponse: The response indicating whether the rule was enabled successfully. |
| 150 | |
| 151 | Raises: |
| 152 | HTTPException: If the rule is already enabled or if the rule could not be enabled. |
| 153 | """ |
| 154 | # Asynchronously fetch the disabled rule |
| 155 | logger.info(f"rule: {rule}") |
| 156 | result = await session.execute( |
| 157 | select(DisabledRule).where(DisabledRule.rule_id == rule.rule_id), |
| 158 | ) |
| 159 | disabled_rule = result.scalars().first() |
| 160 | |
| 161 | if not disabled_rule: |
| 162 | raise HTTPException(status_code=404, detail="Rule is already enabled") |
| 163 | |
| 164 | # This should be converted to an async operation if it's not already |
| 165 | rule_enabled = await enable_rule(rule, disabled_rule.previous_level) |
| 166 | |
| 167 | if rule_enabled: |
| 168 | await session.delete(disabled_rule) |
| 169 | await session.commit() |
| 170 | return rule_enabled |
| 171 | else: |
| 172 | raise HTTPException(status_code=404, detail="Was not able to enable rule") |
| 173 | |
| 174 | |
| 175 | @wazuh_manager_rules_router.post( |
| 176 | "/rule/exclude", |
| 177 | response_model=RuleExcludeResponse, |
| 178 | description="Retrieve recommended exclusion for a Wazuh Rule", |
| 179 | dependencies=[Security(AuthHandler().require_any_scope("admin"))], |
| 180 | ) |
| 181 | async def exclude_wazuh_rule(request: RuleExcludeRequest) -> RuleExcludeResponse: |
| 182 | return await post_to_copilot_ai_module(data=request) |
| 183 | |
| 184 | |
| 185 | @wazuh_manager_rules_router.get( |
| 186 | "/rules", |
| 187 | response_model=WazuhRulesResponse, |
| 188 | description="List Wazuh rules", |
| 189 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 190 | ) |
| 191 | async def list_wazuh_rules( |
| 192 | rule_ids: Optional[List[int]] = Query(None, description="List of rule IDs"), |
| 193 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 194 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 195 | offset: Optional[int] = Query(0, ge=0, description="First element to return"), |
| 196 | limit: Optional[int] = Query(500, ge=1, le=100000, description="Maximum number of elements"), |
| 197 | select: Optional[List[str]] = Query(None, description="Fields to return"), |
| 198 | sort: Optional[str] = Query(None, description="Sort fields"), |
| 199 | search: Optional[str] = Query(None, description="Search text"), |
| 200 | q: Optional[str] = Query(None, description="Query filter"), |
| 201 | status: Optional[str] = Query(None, description="Rule status filter"), |
| 202 | group: Optional[str] = Query(None, description="Rule group filter"), |
| 203 | level: Optional[str] = Query(None, description="Rule level filter"), |
| 204 | filename: Optional[List[str]] = Query(None, description="Filename filter"), |
| 205 | relative_dirname: Optional[str] = Query(None, description="Directory filter"), |
| 206 | pci_dss: Optional[str] = Query(None, description="PCI DSS filter"), |
| 207 | gdpr: Optional[str] = Query(None, description="GDPR filter"), |
| 208 | gpg13: Optional[str] = Query(None, description="GPG13 filter"), |
| 209 | hipaa: Optional[str] = Query(None, description="HIPAA filter"), |
| 210 | nist_800_53: Optional[str] = Query(None, description="NIST 800-53 filter"), |
| 211 | tsc: Optional[str] = Query(None, description="TSC filter"), |
| 212 | mitre: Optional[str] = Query(None, description="MITRE filter"), |
| 213 | distinct: Optional[bool] = Query(False, description="Distinct values only"), |
| 214 | ) -> WazuhRulesResponse: |
| 215 | """ |
| 216 | List Wazuh rules with comprehensive filtering options. |
| 217 | |
| 218 | Returns a list of Wazuh rules from the Wazuh Manager API with support for |
| 219 | filtering by various criteria including compliance frameworks and MITRE ATT&CK. |
| 220 | """ |
| 221 | # Use **locals() to pass all parameters efficiently |
| 222 | params = {k: v for k, v in locals().items() if k not in ["auth_handler"]} |
| 223 | return await get_wazuh_rules(**params) |
| 224 | |
| 225 | |
| 226 | @wazuh_manager_rules_router.get( |
| 227 | "/rules/files", |
| 228 | response_model=WazuhRuleFilesResponse, |
| 229 | description="List Wazuh rule files", |
| 230 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 231 | ) |
| 232 | async def list_wazuh_rule_files( |
| 233 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 234 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 235 | offset: Optional[int] = Query(0, ge=0, description="First element to return in the collection"), |
| 236 | limit: Optional[int] = Query(500, ge=1, le=100000, description="Maximum number of elements to return"), |
| 237 | sort: Optional[str] = Query(None, description="Sort the collection by a field or fields"), |
| 238 | search: Optional[str] = Query(None, description="Look for elements containing the specified string"), |
| 239 | relative_dirname: Optional[str] = Query(None, description="Filter by relative directory name"), |
| 240 | filename: Optional[List[str]] = Query(None, description="Filter by filename of rule files"), |
| 241 | status: Optional[str] = Query(None, description="Filter by list status (enabled, disabled, all)"), |
| 242 | q: Optional[str] = Query(None, description="Query to filter results by"), |
| 243 | select: Optional[List[str]] = Query(None, description="Select which fields to return"), |
| 244 | distinct: Optional[bool] = Query(False, description="Look for distinct values"), |
| 245 | ) -> WazuhRuleFilesResponse: |
| 246 | """ |
| 247 | Retrieve a list of Wazuh rule files from the Wazuh Manager. |
| 248 | |
| 249 | This endpoint provides access to all rule files used to define Wazuh rules, |
| 250 | including their status and location within the ruleset directory structure. |
| 251 | |
| 252 | Parameters: |
| 253 | - pretty: Format results for human readability |
| 254 | - wait_for_complete: Disable request timeout |
| 255 | - offset: Pagination offset (default: 0) |
| 256 | - limit: Maximum results per page (default: 500, max: 100000) |
| 257 | - sort: Fields to sort by (use +/- prefix for ascending/descending) |
| 258 | - search: Text search across file properties |
| 259 | - relative_dirname: Filter by relative directory path |
| 260 | - filename: Filter by specific rule filenames |
| 261 | - status: Filter by file status (enabled/disabled/all) |
| 262 | - q: Advanced query filter |
| 263 | - select: Comma-separated list of fields to return |
| 264 | - distinct: Return only distinct values |
| 265 | |
| 266 | Returns: |
| 267 | - WazuhRuleFilesResponse: List of rule files with their status and location metadata. |
| 268 | """ |
| 269 | # Use locals() to capture all parameters, excluding non-parameter variables |
| 270 | params = {k: v for k, v in locals().items()} |
| 271 | return await get_wazuh_rule_files(**params) |
| 272 | |
| 273 | |
| 274 | @wazuh_manager_rules_router.get( |
| 275 | "/rules/files/{filename}", |
| 276 | response_model=WazuhRuleFileContentResponse, |
| 277 | description="Get Wazuh rule file content", |
| 278 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 279 | ) |
| 280 | async def get_wazuh_rule_file_content_endpoint( |
| 281 | filename: str = Path(..., description="Filename (rule or decoder) to get content for"), |
| 282 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 283 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 284 | raw: Optional[bool] = Query(True, description="Format response in plain text"), |
| 285 | relative_dirname: Optional[str] = Query(None, description="Filter by relative directory name"), |
| 286 | ) -> WazuhRuleFileContentResponse: |
| 287 | """ |
| 288 | Get the content of a specified rule file in the ruleset. |
| 289 | |
| 290 | This endpoint retrieves the full content of a Wazuh rule file, which can contain |
| 291 | multiple rule groups and individual rules with their configurations. |
| 292 | |
| 293 | Parameters: |
| 294 | - filename: The name of the rule file to retrieve (required) |
| 295 | - pretty: Format results for human readability |
| 296 | - wait_for_complete: Disable request timeout |
| 297 | - raw: Return content as plain text instead of structured data |
| 298 | - relative_dirname: Filter by relative directory name |
| 299 | |
| 300 | Returns: |
| 301 | - WazuhRuleFileContentResponse: The content of the rule file, either as structured |
| 302 | data (default) or as raw text (when raw=true). |
| 303 | |
| 304 | Raises: |
| 305 | - 404: If the specified rule file is not found |
| 306 | - 500: If there's an error retrieving the file content |
| 307 | """ |
| 308 | # Use locals() to capture all parameters, excluding the filename path parameter |
| 309 | params = {k: v for k, v in locals().items() if k != "filename"} |
| 310 | return await get_wazuh_rule_file_content(filename, **params) |
| 311 | |
| 312 | |
| 313 | @wazuh_manager_rules_router.put( |
| 314 | "/rules/files/{filename}", |
| 315 | response_model=WazuhRuleFileUploadResponse, |
| 316 | description="Upload or update a Wazuh rule file", |
| 317 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 318 | ) |
| 319 | async def update_wazuh_rule_file_endpoint( |
| 320 | filename: str = Path(..., description="Name of the rule file to upload/update"), |
| 321 | file: UploadFile = File(..., description="Rule file content (XML format)"), |
| 322 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 323 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 324 | overwrite: Optional[bool] = Query(True, description="Whether to overwrite the file if it exists"), |
| 325 | relative_dirname: Optional[str] = Query(None, description="Relative directory name"), |
| 326 | ) -> WazuhRuleFileUploadResponse: |
| 327 | """ |
| 328 | Upload or update a Wazuh rule file in the ruleset. |
| 329 | |
| 330 | This endpoint allows you to upload a new rule file or update an existing one |
| 331 | in the Wazuh Manager ruleset. The file should be in XML format containing |
| 332 | valid Wazuh rule definitions. |
| 333 | |
| 334 | Parameters: |
| 335 | - filename: The name of the rule file to upload/update (required) |
| 336 | - file: The rule file content as a binary upload (required, should be XML format) |
| 337 | - pretty: Format results for human readability |
| 338 | - wait_for_complete: Disable request timeout |
| 339 | - overwrite: Whether to overwrite the file if it already exists |
| 340 | - relative_dirname: Relative directory name where the file should be placed |
| 341 | |
| 342 | Returns: |
| 343 | - WazuhRuleFileUploadResponse: Confirmation of successful upload/update with file details |
| 344 | |
| 345 | Raises: |
| 346 | - 400: If the file format is invalid or parameters are incorrect |
| 347 | - 409: If the file already exists and overwrite is False |
| 348 | - 500: If there's an error uploading the file |
| 349 | """ |
| 350 | # Validate file content type (should be XML or octet-stream) |
| 351 | if file.content_type and not any(ct in file.content_type.lower() for ct in ["xml", "text", "octet-stream", "application/xml"]): |
| 352 | logger.warning(f"Unexpected content type: {file.content_type}") |
| 353 | |
| 354 | try: |
| 355 | # Read file content |
| 356 | file_content = await file.read() |
| 357 | |
| 358 | # Validate that we have content |
| 359 | if not file_content: |
| 360 | raise HTTPException(status_code=400, detail="File content is empty") |
| 361 | |
| 362 | # Basic XML validation (check for XML tags) |
| 363 | file_content_str = file_content.decode("utf-8", errors="ignore") |
| 364 | if not file_content_str.strip().startswith("<"): |
| 365 | logger.warning("File does not appear to be XML format") |
| 366 | |
| 367 | logger.info(f"Received file upload: {filename}, size: {len(file_content)} bytes") |
| 368 | |
| 369 | # Call service function |
| 370 | return await update_wazuh_rule_file( |
| 371 | filename=filename, |
| 372 | file_content=file_content, |
| 373 | pretty=pretty, |
| 374 | wait_for_complete=wait_for_complete, |
| 375 | overwrite=overwrite, |
| 376 | relative_dirname=relative_dirname, |
| 377 | ) |
| 378 | |
| 379 | except UnicodeDecodeError: |
| 380 | raise HTTPException(status_code=400, detail="Invalid file encoding. File must be UTF-8 encoded XML.") |
| 381 | except Exception as e: |
| 382 | logger.error(f"Error processing file upload for {filename}: {e}") |
| 383 | raise HTTPException(status_code=500, detail=f"Error processing file upload: {str(e)}") |