feat: add snapshot and restore functionality to Wazuh Indexer
taylorwalton committed
Jan 28, 2026 at 11:53 UTC
18c3dbb83f9779eeb63f4974ad5737f42b83c76e
4 files changed
+123
backend/app/connectors/wazuh_indexer/routes/snapshot_and_restore.py
new
+34
@@ -0,0 +1,34 @@
1
+from fastapi import APIRouter
2
+from fastapi import HTTPException
3
+from loguru import logger
4
+
5
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepositoryListResponse
6
+from app.connectors.wazuh_indexer.services.snapshot_and_restore import list_snapshot_repositories
7
+
8
+wazuh_indexer_snapshots_router = APIRouter()
9
+
10
+
11
+@wazuh_indexer_snapshots_router.get(
12
+ "/repositories",
13
+ response_model=SnapshotRepositoryListResponse,
14
+ summary="List Snapshot Repositories",
15
+ description="Retrieve a list of all configured snapshot repositories in the Wazuh Indexer.",
16
+)
17
+async def get_snapshot_repositories() -> SnapshotRepositoryListResponse:
18
+ """
19
+ List all snapshot repositories configured in the Wazuh Indexer.
20
+
21
+ Returns:
22
+ SnapshotRepositoryListResponse: List of snapshot repositories with their settings.
23
+ """
24
+ logger.info("Received request to list snapshot repositories")
25
+
26
+ response = await list_snapshot_repositories()
27
+
28
+ if not response.success:
29
+ raise HTTPException(
30
+ status_code=500,
31
+ detail=response.message,
32
+ )
33
+
34
+ return response
backend/app/connectors/wazuh_indexer/schema/snapshot_and_restore.py
new
+34
@@ -0,0 +1,34 @@
1
+from typing import Any
2
+from typing import Dict
3
+from typing import List
4
+from typing import Optional
5
+
6
+from pydantic import BaseModel
7
+from pydantic import Field
8
+
9
+
10
+class SnapshotRepositorySettings(BaseModel):
11
+ """Settings for a snapshot repository."""
12
+ location: Optional[str] = Field(None, description="Repository location/path")
13
+ compress: Optional[bool] = Field(None, description="Whether snapshots are compressed")
14
+ chunk_size: Optional[str] = Field(None, description="Chunk size for snapshot files")
15
+ max_restore_bytes_per_sec: Optional[str] = Field(None, description="Max restore rate")
16
+ max_snapshot_bytes_per_sec: Optional[str] = Field(None, description="Max snapshot rate")
17
+ readonly: Optional[bool] = Field(None, description="Whether repository is read-only")
18
+
19
+
20
+class SnapshotRepository(BaseModel):
21
+ """Model for a single snapshot repository."""
22
+ name: str = Field(..., description="Name of the repository")
23
+ type: str = Field(..., description="Type of the repository (fs, s3, etc.)")
24
+ settings: Dict[str, Any] = Field(default_factory=dict, description="Repository settings")
25
+
26
+
27
+class SnapshotRepositoryListResponse(BaseModel):
28
+ """Response model for listing snapshot repositories."""
29
+ repositories: List[SnapshotRepository] = Field(
30
+ default_factory=list,
31
+ description="List of snapshot repositories",
32
+ )
33
+ success: bool = Field(..., description="Whether the operation was successful")
34
+ message: str = Field(..., description="Status message")
backend/app/connectors/wazuh_indexer/services/snapshot_and_restore.py
new
+49
@@ -0,0 +1,49 @@
1
+from typing import List
2
+
3
+from loguru import logger
4
+
5
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepository
6
+from app.connectors.wazuh_indexer.schema.snapshot_and_restore import SnapshotRepositoryListResponse
7
+from app.connectors.wazuh_indexer.utils.universal import create_wazuh_indexer_client
8
+
9
+
10
+async def list_snapshot_repositories() -> SnapshotRepositoryListResponse:
11
+ """
12
+ List all snapshot repositories configured in the Wazuh Indexer (OpenSearch).
13
+
14
+ Returns:
15
+ SnapshotRepositoryListResponse: Response containing list of repositories.
16
+ """
17
+ logger.info("Fetching snapshot repositories from Wazuh Indexer")
18
+
19
+ try:
20
+ es_client = await create_wazuh_indexer_client("Wazuh-Indexer")
21
+
22
+ # Get all snapshot repositories using the _snapshot API
23
+ response = es_client.snapshot.get_repository()
24
+
25
+ repositories: List[SnapshotRepository] = []
26
+
27
+ for repo_name, repo_data in response.items():
28
+ repository = SnapshotRepository(
29
+ name=repo_name,
30
+ type=repo_data.get("type", "unknown"),
31
+ settings=repo_data.get("settings", {}),
32
+ )
33
+ repositories.append(repository)
34
+
35
+ logger.info(f"Successfully retrieved {len(repositories)} snapshot repositories")
36
+
37
+ return SnapshotRepositoryListResponse(
38
+ repositories=repositories,
39
+ success=True,
40
+ message=f"Successfully retrieved {len(repositories)} snapshot repositories",
41
+ )
42
+
43
+ except Exception as e:
44
+ logger.error(f"Failed to list snapshot repositories: {e}")
45
+ return SnapshotRepositoryListResponse(
46
+ repositories=[],
47
+ success=False,
48
+ message=f"Failed to list snapshot repositories: {str(e)}",
49
+ )
backend/app/routers/wazuh_indexer.py
+6
@@ -3,6 +3,7 @@ from fastapi import APIRouter
3
from app.connectors.wazuh_indexer.routes.alerts import wazuh_indexer_alerts_router
4
from app.connectors.wazuh_indexer.routes.monitoring import wazuh_indexer_router
5
from app.connectors.wazuh_indexer.routes.sigma import wazuh_indexer_sigma_router
6
+from app.connectors.wazuh_indexer.routes.snapshot_and_restore import wazuh_indexer_snapshots_router
7
8
# Instantiate the APIRouter
9
router = APIRouter()
@@ -23,3 +24,8 @@ router.include_router(
24
prefix="/sigma",
25
tags=["wazuh-indexer-sigma"],
26
)
27
+router.include_router(
28
+ wazuh_indexer_snapshots_router,
29
+ prefix="/snapshots",
30
+ tags=["wazuh-indexer-snapshots"],
31
+)