| 1 | from typing import List |
| 2 | from typing import Optional |
| 3 | |
| 4 | from fastapi import APIRouter |
| 5 | from fastapi import Path |
| 6 | from fastapi import Query |
| 7 | from fastapi import Request |
| 8 | from fastapi import Security |
| 9 | |
| 10 | from app.auth.routes.auth import AuthHandler |
| 11 | from app.connectors.wazuh_manager.schema.groups import ( |
| 12 | WazuhGroupConfigurationUpdateResponse, |
| 13 | ) |
| 14 | from app.connectors.wazuh_manager.schema.groups import WazuhGroupFileResponse |
| 15 | from app.connectors.wazuh_manager.schema.groups import WazuhGroupFilesResponse |
| 16 | from app.connectors.wazuh_manager.schema.groups import WazuhGroupsResponse |
| 17 | from app.connectors.wazuh_manager.services.groups import get_wazuh_group_file |
| 18 | from app.connectors.wazuh_manager.services.groups import get_wazuh_group_files |
| 19 | from app.connectors.wazuh_manager.services.groups import get_wazuh_groups |
| 20 | from app.connectors.wazuh_manager.services.groups import ( |
| 21 | update_wazuh_group_configuration, |
| 22 | ) |
| 23 | |
| 24 | wazuh_manager_groups_router = APIRouter() |
| 25 | auth_handler = AuthHandler() |
| 26 | |
| 27 | |
| 28 | @wazuh_manager_groups_router.get( |
| 29 | "/groups", |
| 30 | response_model=WazuhGroupsResponse, |
| 31 | description="Get Wazuh manager groups", |
| 32 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 33 | ) |
| 34 | async def list_wazuh_groups( |
| 35 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 36 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 37 | groups_list: Optional[List[str]] = Query(None, description="List of group IDs (separated by comma)"), |
| 38 | offset: Optional[int] = Query(0, ge=0, description="First element to return in the collection"), |
| 39 | limit: Optional[int] = Query(500, ge=1, le=100000, description="Maximum number of elements to return"), |
| 40 | sort: Optional[str] = Query(None, description="Sort the collection by a field or fields"), |
| 41 | search: Optional[str] = Query(None, description="Look for elements containing the specified string"), |
| 42 | hash: Optional[str] = Query( |
| 43 | None, |
| 44 | description="Select algorithm to generate the returned checksums", |
| 45 | pattern="^(md5|sha1|sha224|sha256|sha384|sha512|blake2b|blake2s|sha3_224|sha3_256|sha3_384|sha3_512)$", |
| 46 | ), |
| 47 | q: Optional[str] = Query(None, description="Query to filter results by"), |
| 48 | select: Optional[List[str]] = Query(None, description="Select which fields to return"), |
| 49 | distinct: Optional[bool] = Query(False, description="Look for distinct values"), |
| 50 | ) -> WazuhGroupsResponse: |
| 51 | """ |
| 52 | Get information about all groups or a list of them. |
| 53 | |
| 54 | Returns a list containing basic information about each group such as number of agents |
| 55 | belonging to the group and the checksums of the configuration and shared files. |
| 56 | |
| 57 | Parameters: |
| 58 | - pretty: Format results for human readability |
| 59 | - wait_for_complete: Disable request timeout |
| 60 | - groups_list: List of group IDs to filter by |
| 61 | - offset: Pagination offset (default: 0) |
| 62 | - limit: Maximum results per page (default: 500, max: 100000) |
| 63 | - sort: Fields to sort by (use +/- prefix for ascending/descending) |
| 64 | - search: Text search across group properties |
| 65 | - hash: Algorithm to generate checksums (md5, sha1, etc.) |
| 66 | - q: Advanced query filter |
| 67 | - select: Comma-separated list of fields to return |
| 68 | - distinct: Return only distinct values |
| 69 | |
| 70 | Returns: |
| 71 | - WazuhGroupsResponse: List of groups with their information and checksums. |
| 72 | """ |
| 73 | # Use **locals() to pass all parameters efficiently |
| 74 | params = {k: v for k, v in locals().items() if k not in ["auth_handler"]} |
| 75 | return await get_wazuh_groups(**params) |
| 76 | |
| 77 | |
| 78 | @wazuh_manager_groups_router.get( |
| 79 | "/groups/{group_id}/files", |
| 80 | response_model=WazuhGroupFilesResponse, |
| 81 | description="Get files in a Wazuh group", |
| 82 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 83 | ) |
| 84 | async def get_wazuh_group_files_endpoint( |
| 85 | group_id: str = Path(..., description="Group ID (name of the group)"), |
| 86 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 87 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 88 | offset: Optional[int] = Query(0, ge=0, description="First element to return in the collection"), |
| 89 | limit: Optional[int] = Query(500, ge=1, le=100000, description="Maximum number of elements to return"), |
| 90 | sort: Optional[str] = Query(None, description="Sort the collection by a field or fields"), |
| 91 | search: Optional[str] = Query(None, description="Look for elements containing the specified string"), |
| 92 | hash: Optional[str] = Query( |
| 93 | None, |
| 94 | description="Select algorithm to generate the returned checksums", |
| 95 | pattern="^(md5|sha1|sha224|sha256|sha384|sha512|blake2b|blake2s|sha3_224|sha3_256|sha3_384|sha3_512)$", |
| 96 | ), |
| 97 | q: Optional[str] = Query(None, description="Query to filter results by"), |
| 98 | select: Optional[List[str]] = Query(None, description="Select which fields to return"), |
| 99 | distinct: Optional[bool] = Query(False, description="Look for distinct values"), |
| 100 | ) -> WazuhGroupFilesResponse: |
| 101 | """ |
| 102 | Return the files placed under the group directory. |
| 103 | |
| 104 | This endpoint retrieves a list of all files in a specific Wazuh group directory, |
| 105 | including their filenames and hash checksums. |
| 106 | |
| 107 | Parameters: |
| 108 | - group_id: The ID (name) of the group (required) |
| 109 | - pretty: Format results for human readability |
| 110 | - wait_for_complete: Disable request timeout |
| 111 | - offset: Pagination offset (default: 0) |
| 112 | - limit: Maximum results per page (default: 500, max: 100000) |
| 113 | - sort: Fields to sort by (use +/- prefix for ascending/descending) |
| 114 | - search: Text search across file properties |
| 115 | - hash: Algorithm to generate checksums (md5, sha1, etc.) |
| 116 | - q: Advanced query filter |
| 117 | - select: Comma-separated list of fields to return |
| 118 | - distinct: Return only distinct values |
| 119 | |
| 120 | Returns: |
| 121 | - WazuhGroupFilesResponse: List of files in the group with their checksums. |
| 122 | |
| 123 | Raises: |
| 124 | - 404: If the specified group is not found |
| 125 | - 500: If there's an error retrieving the group files |
| 126 | """ |
| 127 | # Use locals() to capture all parameters, excluding path parameters |
| 128 | params = {k: v for k, v in locals().items() if k not in ["group_id"]} |
| 129 | return await get_wazuh_group_files(group_id, **params) |
| 130 | |
| 131 | |
| 132 | @wazuh_manager_groups_router.get( |
| 133 | "/groups/{group_id}/files/{filename}", |
| 134 | response_model=WazuhGroupFileResponse, |
| 135 | description="Get a file in a Wazuh group", |
| 136 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 137 | ) |
| 138 | async def get_wazuh_group_file_endpoint( |
| 139 | group_id: str = Path(..., description="Group ID (name of the group)"), |
| 140 | filename: str = Path(..., description="Filename (e.g., agent.conf)"), |
| 141 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 142 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 143 | type: Optional[List[str]] = Query(None, description="Type of file", pattern="^(conf|rootkit_files|rootkit_trojans|rcl)$"), |
| 144 | raw: Optional[bool] = Query(True, description="Format response in plain text"), |
| 145 | ) -> WazuhGroupFileResponse: |
| 146 | """ |
| 147 | Return the content of the specified group file. |
| 148 | |
| 149 | This endpoint retrieves the content of a specific file from a Wazuh group. |
| 150 | The filename will typically be 'agent.conf' for group configuration. |
| 151 | By default, content is returned as raw text (raw=True). |
| 152 | |
| 153 | Parameters: |
| 154 | - group_id: The ID (name) of the group (required) |
| 155 | - filename: The name of the file to retrieve (required, e.g., "agent.conf") |
| 156 | - pretty: Format results for human readability |
| 157 | - wait_for_complete: Disable request timeout |
| 158 | - type: Type of file (conf, rootkit_files, rootkit_trojans, rcl) |
| 159 | - raw: Return content as plain text instead of structured data (default: True) |
| 160 | |
| 161 | Returns: |
| 162 | - WazuhGroupFileResponse: The content of the group file, either as raw text |
| 163 | (default) or as structured data (when raw=false). |
| 164 | |
| 165 | Raises: |
| 166 | - 404: If the specified group or file is not found |
| 167 | - 500: If there's an error retrieving the file content |
| 168 | """ |
| 169 | # Use locals() to capture all parameters, excluding path parameters |
| 170 | params = {k: v for k, v in locals().items() if k not in ["group_id", "filename"]} |
| 171 | return await get_wazuh_group_file(group_id, filename, **params) |
| 172 | |
| 173 | |
| 174 | @wazuh_manager_groups_router.put( |
| 175 | "/groups/{group_id}/configuration", |
| 176 | response_model=WazuhGroupConfigurationUpdateResponse, |
| 177 | description="Update group configuration", |
| 178 | dependencies=[Security(AuthHandler().get_current_user, scopes=["admin"])], |
| 179 | ) |
| 180 | async def update_wazuh_group_configuration_endpoint( |
| 181 | request: Request, |
| 182 | group_id: str = Path(..., description="Group ID (name of the group)"), |
| 183 | pretty: Optional[bool] = Query(False, description="Show results in human-readable format"), |
| 184 | wait_for_complete: Optional[bool] = Query(False, description="Disable timeout response"), |
| 185 | ) -> WazuhGroupConfigurationUpdateResponse: |
| 186 | """ |
| 187 | Update an specified group's configuration. |
| 188 | |
| 189 | This API call expects a full valid XML file with the shared configuration tags/syntax. |
| 190 | The configuration will be applied to all agents belonging to the specified group. |
| 191 | |
| 192 | Parameters: |
| 193 | - group_id: The ID (name) of the group to update (required) |
| 194 | - request: Raw XML configuration content in the request body |
| 195 | - pretty: Format results for human readability |
| 196 | - wait_for_complete: Disable request timeout |
| 197 | |
| 198 | Request Body: |
| 199 | - Raw XML configuration content (Content-Type: application/xml) |
| 200 | |
| 201 | Returns: |
| 202 | - WazuhGroupConfigurationUpdateResponse: Confirmation of successful update |
| 203 | |
| 204 | Raises: |
| 205 | - 400: If the configuration content is invalid or malformed XML |
| 206 | - 404: If the specified group is not found |
| 207 | - 500: If there's an error updating the configuration |
| 208 | """ |
| 209 | # Read the raw XML content from the request body |
| 210 | configuration_content = await request.body() |
| 211 | configuration_xml = configuration_content.decode("utf-8") |
| 212 | |
| 213 | # Use locals() to capture all parameters, excluding path and body parameters |
| 214 | params = {k: v for k, v in locals().items() if k not in ["group_id", "request", "configuration_content", "configuration_xml"]} |
| 215 | return await update_wazuh_group_configuration(group_id, configuration_xml, **params) |