| 1 | from enum import Enum |
| 2 | from typing import Any |
| 3 | |
| 4 | from fastapi import HTTPException |
| 5 | from loguru import logger |
| 6 | from pydantic import BaseModel |
| 7 | from pydantic import Field |
| 8 | |
| 9 | |
| 10 | class AvailableNetworkConnectors(str, Enum): |
| 11 | FORTINET = ( |
| 12 | "The Fortinet Network Connector which includes Input, Stream, Pipeline Rules," |
| 13 | " Pipelines, and Lookup Tables for Fortinet logs and the SOCFortress SIEM stack." |
| 14 | ) |
| 15 | CROWDSTRIKE = ( |
| 16 | "The Crowdstrike Network Connector which includes Input, Stream, Pipeline Rules," |
| 17 | " Pipelines, and Lookup Tables for Crowdstrike logs and the SOCFortress SIEM stack." |
| 18 | ) |
| 19 | SONICWALL = ( |
| 20 | "The Sonicwall Network Connector which includes Input, Stream, Pipeline Rules," |
| 21 | " Pipelines, and Lookup Tables for Sonicwall logs and the SOCFortress SIEM stack." |
| 22 | ) |
| 23 | |
| 24 | |
| 25 | class DecommissionNetworkContentPackRequest(BaseModel): |
| 26 | network_connector: AvailableNetworkConnectors = Field( |
| 27 | ..., |
| 28 | examples=[AvailableNetworkConnectors.FORTINET.name], |
| 29 | description="The name of the content pack to provision in Graylog", |
| 30 | ) |
| 31 | customer_code: str = Field( |
| 32 | ..., |
| 33 | description="The customer code for the content pack to provision in Graylog", |
| 34 | examples=["00001"], |
| 35 | ) |
| 36 | |
| 37 | def __init__(self, **data: Any): |
| 38 | network_connector = data.get("network_connector") |
| 39 | if network_connector: |
| 40 | network_connector = network_connector.upper() |
| 41 | logger.info(f"Network Connector: {network_connector}") |
| 42 | try: |
| 43 | data["network_connector"] = AvailableNetworkConnectors[network_connector] |
| 44 | except KeyError: |
| 45 | raise HTTPException( |
| 46 | status_code=400, |
| 47 | detail=f"{network_connector} is not available. Please choose from the available network connectors.", |
| 48 | ) |
| 49 | super().__init__(**data) |
| 50 | |
| 51 | |
| 52 | class DecommissionNetworkContentPackResponse(BaseModel): |
| 53 | message: str = Field( |
| 54 | ..., |
| 55 | examples=["FORTINET Content Pack decommissioned successfully"], |
| 56 | description="Message from the request to decommission a content pack", |
| 57 | ) |
| 58 | success: bool = Field( |
| 59 | ..., |
| 60 | examples=[True], |
| 61 | description="Success of the request to decommission a content pack", |
| 62 | ) |