@cryptotaxi247 / CoPilot
Code Commits Issues Pulls Sessions CI/CD
main
main
py 175 lines 6.14 KB
Raw
1 from enum import Enum
2 from typing import Any
3 from typing import List
4 from typing import Optional
5
6 from fastapi import HTTPException
7 from pydantic import BaseModel
8 from pydantic import Field
9
10
11 class AvailbleContentPacksOverview(str, Enum):
12 SOCFORTRESS_WAZUH_CONTENT_PACK = (
13 "The Wazuh Content Pack which includes Input, Stream, Pipeline Rules,"
14 " Pipelines, and Lookup Tables for Wazuh logs and the SOCFortress SIEM stack."
15 )
16
17
18 class AvailableContentPacks(str, Enum):
19 SOCFORTRESS_WAZUH_CONTENT_PACK = (
20 "The Wazuh Content Pack which includes Input, Stream, Pipeline Rules,"
21 " Pipelines, and Lookup Tables for Wazuh logs and the SOCFortress SIEM stack."
22 )
23 SOCFORTRESS_FORTINET_INPUT_SYSLOG_TCP = "The Fortinet Input Syslog TCP content pack"
24 SOCFORTRESS_FORTINET_INPUT_SYSLOG_UDP = "The Fortinet Input Syslog UDP content pack"
25 SOCFORTRESS_FORTINET_PROCESSING_PIPELINE = "The Fortinet Processing Pipeline content pack"
26 SOCFORTRESS_FORTINET_STREAM = "The Fortinet Stream content pack"
27 SOCFORTRESS_CROWDSTRIKE_INPUT_TCP = "The Crowdstrike Input TCP content pack"
28 SOCFORTRESS_CROWDSTRIKE_STREAM = "The Crowdstrike Stream content pack"
29 SOCFORTRESS_CROWDSTRIKE_PROCESSING_PIPELINE = "The Crowdstrike Processing Pipeline content pack"
30 SOCFORTRESS_BITDEFENDER_INPUT_TCP = "The Bitdefender Input TCP content pack"
31 SOCFORTRESS_BITDEFENDER_STREAM = "The Bitdefender Stream content pack"
32 SOCFORTRESS_DEFENDER_FOR_ENDPOINT_INPUT_TCP = "The Defender for Endpoint Input TCP content pack"
33 SOCFORTRESS_DEFENDER_FOR_ENDPOINT_STREAM = "The Defender for Endpoint Stream content pack"
34 SOCFORTRESS_SONICWALL_INPUT_SYSLOG_TLS = "The Sonicwall Input Syslog TLS content pack"
35 SOCFORTRESS_SONICWALL_STREAM = "The Sonicwall Stream content pack"
36 SOCFORTRESS_SONICWALL_PROCESSING_PIPELINE = "The Sonicwall Processing Pipeline content pack"
37
38
39 class ContentPackKeywords(BaseModel):
40 customer_name: Optional[str] = Field(None, description="Name of the customer")
41 customer_code: Optional[str] = Field(None, description="Code of the customer")
42 protocol_type: Optional[str] = Field(
43 None,
44 examples=["TCP"],
45 description="The protocol type of the content pack",
46 )
47 syslog_port: Optional[int] = Field(
48 None,
49 examples=[514],
50 description="The syslog port of the content pack",
51 )
52 tls_cert_file: Optional[str] = Field(
53 None,
54 examples=["/etc/graylog/sonicwall/cert.pem"],
55 description="The TLS certificate file path of the content pack",
56 )
57 tls_key_file: Optional[str] = Field(
58 None,
59 examples=["/etc/graylog/sonicwall/key.pem"],
60 description="The TLS key file path of the content pack",
61 )
62
63
64 class ContentPack(BaseModel):
65 name: str
66 description: str
67
68
69 class AvailableContentPacksResponse(BaseModel):
70 available_content_packs: List[ContentPack] = Field(
71 ...,
72 examples=[
73 {
74 "name": AvailableContentPacks.SOCFORTRESS_WAZUH_CONTENT_PACK.name,
75 "description": AvailableContentPacks.SOCFORTRESS_WAZUH_CONTENT_PACK.value,
76 },
77 ],
78 description="The available content packs for provisioning in Graylog",
79 )
80 success: bool = Field(
81 ...,
82 examples=[True],
83 description="Success of the request to get available content packs",
84 )
85 message: str = Field(
86 ...,
87 examples=["Available content packs retrieved successfully"],
88 description="Message from the request to get available content packs",
89 )
90
91
92 class ProvisionNetworkContentPackRequest(BaseModel):
93 content_pack_name: str = Field(
94 ...,
95 examples=["FORTINET"],
96 description="The name of the content pack to provision in Graylog",
97 )
98 keywords: Optional[ContentPackKeywords] = Field(
99 None,
100 description="The keywords of the content pack to provision in Graylog",
101 )
102
103
104 class ProvisionContentPackRequest(BaseModel):
105 content_pack_name: AvailableContentPacks = Field(
106 ...,
107 examples=[AvailableContentPacks.SOCFORTRESS_WAZUH_CONTENT_PACK],
108 description="The name of the content pack to provision in Graylog",
109 )
110 keywords: Optional[ContentPackKeywords] = Field(
111 None,
112 description="The keywords of the content pack to provision in Graylog",
113 )
114
115 def __init__(self, **data: Any):
116 content_pack_name = data.get("content_pack_name")
117 try:
118 data["content_pack_name"] = AvailableContentPacks[content_pack_name]
119 except KeyError:
120 raise HTTPException(
121 status_code=400,
122 detail=f"Content pack {content_pack_name} is not available. Please choose from the available content packs.",
123 )
124 super().__init__(**data)
125
126
127 class ProvisionGraylogResponse(BaseModel):
128 success: bool = Field(
129 ...,
130 examples=[True],
131 description="Success of the Graylog provisioning",
132 )
133 message: str = Field(
134 ...,
135 examples=["Graylog provisioned successfully"],
136 description="Message from the Graylog provisioning",
137 )
138
139
140 class ReplaceContentPackKeywords(BaseModel):
141 REPLACE_UUID_GLOBAL: str = Field(
142 ...,
143 examples=["12345678-1234-1234-1234-123456789012"],
144 description="The UUID of the content pack",
145 )
146 REPLACE_UUID_SPECIFIC: str = Field(
147 ...,
148 examples=["12345678-1234-1234-1234-123456789012"],
149 description="The UUID of the input",
150 )
151 customer_name: str = Field(
152 ...,
153 examples=["SOCFortress"],
154 description="The name of the customer",
155 )
156 customer_code: str = Field(
157 ...,
158 examples=["00001"],
159 description="The code of the customer",
160 )
161 SYSLOG_PORT: int = Field(
162 ...,
163 examples=[514],
164 description="The syslog port",
165 )
166 TLS_CERT_FILE: Optional[str] = Field(
167 None,
168 examples=["/etc/graylog/sonicwall/cert.pem"],
169 description="The TLS certificate file path",
170 )
171 TLS_KEY_FILE: Optional[str] = Field(
172 None,
173 examples=["/etc/graylog/sonicwall/key.pem"],
174 description="The TLS key file path",
175 )