main
py 646 lines 24.3 KB
Raw
1 import base64
2 import json
3 import os
4 from datetime import datetime
5
6 import aiofiles
7 import httpx
8 from fastapi import HTTPException
9 from loguru import logger
10 from sqlalchemy import and_
11 from sqlalchemy import update
12 from sqlalchemy.ext.asyncio import AsyncSession
13
14 from app.connectors.grafana.schema.dashboards import BitdefenderDashboard
15 from app.connectors.grafana.schema.dashboards import DashboardProvisionRequest
16 from app.connectors.grafana.services.dashboards import provision_dashboards
17 from app.connectors.grafana.utils.universal import create_grafana_client
18 from app.connectors.graylog.services.collector import (
19 get_content_pack_id_by_content_pack_name,
20 )
21 from app.connectors.graylog.services.collector import get_input_id_by_input_name
22 from app.connectors.graylog.services.collector import get_stream_id_by_stream_name
23 from app.connectors.graylog.services.streams import assign_stream_to_index
24 from app.connectors.graylog.utils.universal import send_post_request
25 from app.connectors.wazuh_indexer.services.monitoring import (
26 output_shard_number_to_be_set_based_on_nodes,
27 )
28 from app.customer_provisioning.schema.grafana import GrafanaDatasource
29 from app.customer_provisioning.schema.grafana import GrafanaDataSourceCreationResponse
30 from app.customer_provisioning.schema.graylog import GraylogIndexSetCreationResponse
31 from app.customer_provisioning.schema.graylog import TimeBasedIndexSet
32 from app.customer_provisioning.schema.provision import ProvisionNewCustomer
33 from app.customer_provisioning.services.grafana import create_grafana_folder
34 from app.customer_provisioning.services.grafana import get_opensearch_version
35 from app.customers.routes.customers import get_customer_meta
36 from app.integrations.bitdefender.schema.provision import BitdefenderCustomerDetails
37 from app.integrations.bitdefender.schema.provision import ProvisionBitdefenderAuthKeys
38 from app.integrations.bitdefender.schema.provision import ProvisionBitdefenderResponse
39 from app.integrations.models.customer_integration_settings import CustomerIntegrations
40 from app.network_connectors.models.network_connectors import (
41 CustomerNetworkConnectorsMeta,
42 )
43 from app.stack_provisioning.graylog.schema.provision import ContentPackKeywords
44 from app.stack_provisioning.graylog.schema.provision import (
45 ProvisionNetworkContentPackRequest,
46 )
47 from app.stack_provisioning.graylog.services.provision import (
48 provision_content_pack_network_connector,
49 )
50 from app.utils import get_connector_attribute
51 from app.utils import get_customer_meta_attribute
52
53
54 async def base64_encode(api_key: str):
55 """
56 Base64 encode the given API key.
57
58 Args:
59 api_key (str): The API key to encode.
60
61 Returns:
62 str: The base64 encoded API key.
63 """
64 # Add a : at the end then encode (Required for Basic Auth)
65 return base64.b64encode(f"{api_key}:".encode()).decode()
66
67
68 async def send_configuration_to_bitdefender(keys: ProvisionBitdefenderAuthKeys):
69 url = "https://cloud.gravityzone.bitdefender.com/api/v1.0/jsonrpc/push"
70 headers = {
71 "authorization": f"Basic {await base64_encode(keys.API_KEY)}",
72 "cache-control": "no-cache",
73 "content-type": "application/json",
74 }
75 data = {
76 "id": "1",
77 "jsonrpc": "2.0",
78 "method": "setPushEventSettings",
79 "params": {
80 "serviceSettings": {
81 "requireValidSslCertificate": False,
82 "authorization": await base64_auth_header_generator(keys.BASIC_AUTH_USERNAME, keys.BASIC_AUTH_PASSWORD),
83 "url": f"https://{keys.WEBSERVER_HOSTNAME}:{keys.WEBSERVER_PORT}/api",
84 },
85 "serviceType": "cef",
86 "status": 1,
87 "subscribeToEventTypes": {
88 "adcloudgz": True,
89 "antiexploit": True,
90 "aph": True,
91 "av": True,
92 "avc": True,
93 "dp": True,
94 "endpoint-moved-in": True,
95 "endpoint-moved-out": True,
96 "exchange-malware": True,
97 "exchange-user-credentials": True,
98 "fw": True,
99 "hd": True,
100 "hwid-change": True,
101 "install": True,
102 "modules": True,
103 "network-monitor": True,
104 "network-sandboxing": True,
105 "new-incident": True,
106 "ransomware-mitigation": True,
107 "registration": True,
108 "supa-update-status": True,
109 "sva": True,
110 "sva-load": True,
111 "task-status": True,
112 "troubleshooting-activity": True,
113 "uc": True,
114 "uninstall": True,
115 },
116 },
117 }
118
119 async with httpx.AsyncClient(verify=False) as client:
120 response = await client.post(url, headers=headers, json=data)
121 return response.json()
122
123
124 #### ! GRAYLOG ! ####
125 async def build_index_set_config(request: BitdefenderCustomerDetails) -> TimeBasedIndexSet:
126 """
127 Build the configuration for a time-based index set.
128
129 Args:
130 request (BitdefenderCustomerDetails): The request object containing customer information.
131
132 Returns:
133 TimeBasedIndexSet: The configured time-based index set.
134 """
135 return TimeBasedIndexSet(
136 title=f"{request.customer_name} - BITDEFENDER EVENTS",
137 description=f"{request.customer_name} - BITDEFENDER EVENTS",
138 index_prefix=f"bitdefender-{request.customer_code}",
139 rotation_strategy_class="org.graylog2.indexer.rotation.strategies.TimeBasedRotationStrategy",
140 rotation_strategy={
141 "type": "org.graylog2.indexer.rotation.strategies.TimeBasedRotationStrategyConfig",
142 "rotation_period": "P1D",
143 "rotate_empty_index_set": False,
144 "max_rotation_period": None,
145 },
146 retention_strategy_class="org.graylog2.indexer.retention.strategies.DeletionRetentionStrategy",
147 retention_strategy={
148 "type": "org.graylog2.indexer.retention.strategies.DeletionRetentionStrategyConfig",
149 "max_number_of_indices": request.hot_data_retention,
150 },
151 creation_date=datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
152 index_analyzer="standard",
153 shards=await output_shard_number_to_be_set_based_on_nodes(),
154 replicas=request.index_replicas,
155 index_optimization_max_num_segments=1,
156 index_optimization_disabled=False,
157 writable=True,
158 field_type_refresh_interval=5000,
159 )
160
161
162 # Function to send the POST request and handle the response
163 async def send_index_set_creation_request(
164 index_set: TimeBasedIndexSet,
165 ) -> GraylogIndexSetCreationResponse:
166 """
167 Sends a request to create an index set in Graylog.
168
169 Args:
170 index_set (TimeBasedIndexSet): The index set to be created.
171
172 Returns:
173 GraylogIndexSetCreationResponse: The response from Graylog after creating the index set.
174 """
175 json_index_set = json.dumps(index_set.model_dump())
176 logger.info(f"json_index_set set: {json_index_set}")
177 response_json = await send_post_request(
178 endpoint="/api/system/indices/index_sets",
179 data=index_set.model_dump(),
180 )
181 return GraylogIndexSetCreationResponse(**response_json)
182
183
184 # Refactored create_index_set function
185 async def create_index_set(
186 request: ProvisionNewCustomer,
187 ) -> GraylogIndexSetCreationResponse:
188 """
189 Creates an index set for a new customer.
190
191 Args:
192 request (ProvisionNewCustomer): The request object containing the customer information.
193
194 Returns:
195 GraylogIndexSetCreationResponse: The response object containing the result of the index set creation.
196 """
197 logger.info(f"Creating index set for customer {request.customer_name}")
198 index_set_config = await build_index_set_config(request)
199 return await send_index_set_creation_request(index_set_config)
200
201
202 async def provision_content_pack(customer_details):
203 """
204 Provisions a content pack for a customer.
205
206 Args:
207 customer_details (CustomerDetails): The details of the customer.
208
209 Returns:
210 ContentPack: The provisioned content pack.
211 """
212 return await provision_content_pack_network_connector(
213 content_pack_request=ProvisionNetworkContentPackRequest(
214 content_pack_name="BITDEFENDER",
215 keywords=ContentPackKeywords(
216 customer_name=customer_details.customer_name,
217 customer_code=customer_details.customer_code,
218 protocol_type=customer_details.protocal_type,
219 syslog_port=customer_details.syslog_port,
220 ),
221 ),
222 )
223
224
225 async def get_stream_and_index_ids(customer_details):
226 """
227 Retrieves the stream ID and index ID for a given customer.
228
229 Args:
230 customer_details (CustomerDetails): The details of the customer.
231
232 Returns:
233 tuple: A tuple containing the stream ID and index ID.
234 """
235 stream_id = await get_stream_id_by_stream_name(stream_name=f"{customer_details.customer_name} - BITDEFENDER LOGS AND EVENTS")
236 index_id = (await create_index_set(request=customer_details)).data.id
237 content_pack_stream_id = await get_content_pack_id_by_content_pack_name(
238 content_pack_name=f"{customer_details.customer_name}_BITDEFENDER_STREAM",
239 )
240 if customer_details.protocal_type == "Tcp":
241 content_pack_input_id = await get_content_pack_id_by_content_pack_name(
242 content_pack_name=f"{customer_details.customer_name}_BITDEFENDER_INPUT_TCP",
243 )
244 elif customer_details.protocal_type == "UDP":
245 content_pack_input_id = await get_content_pack_id_by_content_pack_name(
246 content_pack_name=f"{customer_details.customer_name}_BITDEFENDER_INPUT_SYSLOG_UDP",
247 )
248 return stream_id, index_id, content_pack_stream_id, content_pack_input_id
249
250
251 #### ! GRAFANA ! ####
252 async def create_grafana_datasource(
253 customer_code: str,
254 session: AsyncSession,
255 ) -> GrafanaDataSourceCreationResponse:
256 """
257 Creates a Grafana datasource for the specified customer.
258
259 Args:
260 customer_code (str): The customer code.
261 session (AsyncSession): The async session.
262
263 Returns:
264 GrafanaDataSourceCreationResponse: The response containing the created datasource details.
265 """
266 logger.info("Creating Grafana datasource")
267 grafana_client = await create_grafana_client("Grafana")
268 # Switch to the newly created organization
269 grafana_client.user.switch_actual_user_organisation(
270 (await get_customer_meta(customer_code, session)).customer_meta.customer_meta_grafana_org_id,
271 )
272 datasource_payload = GrafanaDatasource(
273 name="BITDEFENDER",
274 type="grafana-opensearch-datasource",
275 typeName="OpenSearch",
276 access="proxy",
277 url=await get_connector_attribute(
278 connector_id=1,
279 column_name="connector_url",
280 session=session,
281 ),
282 database=f"bitdefender-{customer_code}*",
283 basicAuth=True,
284 basicAuthUser=await get_connector_attribute(
285 connector_id=1,
286 column_name="connector_username",
287 session=session,
288 ),
289 secureJsonData={
290 "basicAuthPassword": await get_connector_attribute(
291 connector_id=1,
292 column_name="connector_password",
293 session=session,
294 ),
295 },
296 isDefault=False,
297 jsonData={
298 "database": f"bitdefender-{customer_code}*",
299 "flavor": "opensearch",
300 "includeFrozen": False,
301 "logLevelField": "severity",
302 "logMessageField": "summary",
303 "maxConcurrentShardRequests": 5,
304 "pplEnabled": True,
305 "timeField": "timestamp",
306 "tlsSkipVerify": True,
307 "version": await get_opensearch_version(),
308 },
309 readOnly=True,
310 )
311 results = grafana_client.datasource.create_datasource(
312 datasource=datasource_payload.model_dump(),
313 )
314 return GrafanaDataSourceCreationResponse(**results)
315
316
317 async def create_customer_network_connector_meta(
318 customer_details,
319 stream_id,
320 index_id,
321 content_pack_stream_id,
322 content_pack_input_id,
323 session,
324 ):
325 """
326 Create a CustomerNetworkConnectorsMeta object with the provided details.
327
328 Args:
329 customer_details (CustomerDetails): Details of the customer.
330 stream_id (int): ID of the Graylog stream.
331 index_id (int): ID of the Graylog index.
332 session (Session): Database session.
333
334 Returns:
335 CustomerNetworkConnectorsMeta: The created CustomerNetworkConnectorsMeta object.
336 """
337 return CustomerNetworkConnectorsMeta(
338 customer_code=customer_details.customer_code,
339 network_connector_name="BITDEFENDER",
340 graylog_stream_id=stream_id,
341 graylog_input_id=(await get_input_id_by_input_name(input_name=f"{customer_details.customer_name} - BITDEFENDER LOGS AND EVENTS")),
342 graylog_pipeline_id="NONE",
343 graylog_content_pack_input_id=content_pack_input_id,
344 graylog_content_pack_stream_id=content_pack_stream_id,
345 grafana_org_id=(
346 await get_customer_meta_attribute(
347 session=session,
348 customer_code=customer_details.customer_code,
349 column_name="customer_meta_grafana_org_id",
350 )
351 ),
352 graylog_index_id=index_id,
353 grafana_dashboard_folder_id=None,
354 grafana_datasource_uid=None,
355 )
356
357
358 async def validate_grafana_organization_id(customer_code, session):
359 """
360 Validate the Grafana organization ID for the customer.
361
362 Args:
363 customer_code (str): The customer code.
364 session (Session): Database session.
365
366 Returns:
367 int: The Grafana organization ID.
368 """
369 return await get_customer_meta_attribute(session=session, customer_code=customer_code, column_name="customer_meta_grafana_org_id")
370
371
372 async def provision_bitdefender(
373 customer_details: BitdefenderCustomerDetails,
374 keys: ProvisionBitdefenderAuthKeys,
375 session: AsyncSession,
376 ) -> ProvisionBitdefenderResponse:
377 """
378 Provisions a Bitdefender customer by performing the following steps:
379 1. Provisions the content pack for the customer.
380 2. Retrieves the stream and index IDs for the customer.
381 3. Creates customer network connector metadata.
382 4. Assigns the stream to the index.
383 5. Retrieves the pipeline ID for the "BITDEFENDER" subscription.
384 6. Connects the stream to the pipeline.
385 7. Inserts the customer network connector metadata into the database.
386 8. Creates a directory for the customer to store the docker compose and falconhose cfg.
387
388 Args:
389 customer_details (BitdefenderCustomerDetails): The details of the Bitdefender customer.
390 keys (ProvisionBitdefenderKeys): The keys required for provisioning.
391 session (AsyncSession): The database session.
392
393 Returns:
394 None
395 """
396 # If customer name contains a space, replace it with a _
397 if " " in customer_details.customer_name:
398 customer_details.customer_name = customer_details.customer_name.replace(" ", "_")
399 if await validate_grafana_organization_id(customer_details.customer_code, session) is None:
400 raise HTTPException(status_code=404, detail="Grafana organization ID not found. Please provision Grafana for the customer first.")
401 await provision_content_pack(customer_details)
402 stream_id, index_id, content_pack_stream_id, content_pack_input_id = await get_stream_and_index_ids(customer_details)
403 customer_network_connector_meta = await create_customer_network_connector_meta(
404 customer_details,
405 stream_id,
406 index_id,
407 content_pack_stream_id,
408 content_pack_input_id,
409 session,
410 )
411 await assign_stream_to_index(stream_id=stream_id, index_id=index_id)
412 # ! Commenting out the pipeline ID retrieval for now since I don't have the pipeline template ! #
413 # pipeline_id = await get_pipeline_id(subscription="BITDEFENDER")
414 # await connect_stream_to_pipeline(stream_and_pipeline=StreamConnectionToPipelineRequest(stream_id=stream_id, pipeline_ids=pipeline_id))
415 # Grafana Deployment
416 customer_network_connector_meta.grafana_datasource_uid = (
417 await create_grafana_datasource(
418 customer_code=customer_details.customer_code,
419 session=session,
420 )
421 ).datasource.uid
422 grafana_folder = await create_grafana_folder(
423 organization_id=(
424 await get_customer_meta(
425 customer_details.customer_code,
426 session,
427 )
428 ).customer_meta.customer_meta_grafana_org_id,
429 folder_title="BITDEFENDER",
430 )
431 await provision_dashboards(
432 DashboardProvisionRequest(
433 dashboards=[dashboard.name for dashboard in BitdefenderDashboard],
434 organizationId=(
435 await get_customer_meta(
436 customer_details.customer_code,
437 session,
438 )
439 ).customer_meta.customer_meta_grafana_org_id,
440 folderId=grafana_folder.id,
441 datasourceUid=customer_network_connector_meta.grafana_datasource_uid,
442 ),
443 )
444 customer_network_connector_meta.grafana_dashboard_folder_id = grafana_folder.uid
445 await insert_into_customer_network_connectors_meta_table(
446 customer_network_connectors_meta=customer_network_connector_meta,
447 session=session,
448 )
449 await create_customer_directory_if_needed(customer_name=customer_details.customer_name)
450 file = await load_and_replace_docker_compose(customer_name=customer_details.customer_name, port=keys.WEBSERVER_PORT)
451 await save_uploaded_file(
452 file=file,
453 filename=f"{customer_details.customer_name}_bitdefender_docker-compose.yml",
454 customer_name=customer_details.customer_name,
455 )
456 await load_and_replace_config_json(customer_details=customer_details, keys=keys, session=session)
457 await update_customer_integration_table(
458 customer_code=customer_details.customer_code,
459 session=session,
460 )
461
462 await send_configuration_to_bitdefender(keys)
463
464 return ProvisionBitdefenderResponse(
465 message="Bitdefender customer provisioned successfully",
466 success=True,
467 )
468
469
470 async def insert_into_customer_network_connectors_meta_table(
471 customer_network_connectors_meta: CustomerNetworkConnectorsMeta,
472 session: AsyncSession,
473 ) -> None:
474 """
475 Insert the customer network connectors meta into the database.
476
477 Args:
478 customer_network_connectors_meta (CustomerNetworkConnectorsMeta): The customer network connectors meta to insert.
479 session (AsyncSession): The async session object for database operations.
480
481 Returns:
482 None
483 """
484 logger.info("Inserting customer network connectors meta into the database")
485 session.add(customer_network_connectors_meta)
486 await session.commit()
487 logger.info("Customer network connectors meta inserted successfully")
488 return None
489
490
491 # ! Add the docker-compose.yml file to the `data` folder
492 project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
493 UPLOAD_FOLDER = os.path.join(project_root, "data")
494
495
496 async def create_customer_directory_if_needed(customer_name: str):
497 """
498 Create a directory for the customer in the UPLOAD_FOLDER if it doesn't exist.
499
500 Args:
501 customer_name (str): The name of the customer.
502 """
503 # Create the path to the customer's directory
504 # If customer name contains a space, replace it with a _
505 if " " in customer_name:
506 customer_name = customer_name.replace(" ", "_")
507 customer_directory = os.path.join(UPLOAD_FOLDER, customer_name)
508 # Check if the directory exists
509 if not os.path.exists(customer_directory):
510 # If it doesn't exist, create it
511 os.makedirs(customer_directory)
512
513
514 async def load_and_replace_docker_compose(customer_name: str, port: str):
515 """
516 Load the docker-compose.yml file and replace the placeholder with the customer name.
517
518 Args:
519 customer_name (str): The name of the customer.
520
521 Returns:
522 str: The content of the docker-compose.yml file with the placeholder replaced.
523 """
524 # Get the current directory:
525 current_directory = os.path.dirname(os.path.abspath(__file__))
526 # Go up one level
527 parent_directory = os.path.dirname(current_directory)
528 # If customer name contains a space, replace it with a _
529 if " " in customer_name:
530 customer_name = customer_name.replace(" ", "_")
531 # Open the docker-compose.yml file and read the content
532 with open(os.path.join(parent_directory, "templates", "docker-compose.yml"), "r") as file:
533 data = file.read()
534 data = data.replace("CUSTOMER_NAME", customer_name)
535 data = data.replace("PORT", port)
536 return data
537
538
539 async def save_uploaded_file(file, filename, customer_name):
540 """
541 Save the uploaded file to the server.
542
543 Args:
544 file: The file to save.
545 filename: The name of the file.
546
547 Returns:
548 str: The path to the saved file.
549 """
550 # If customer name contains a space, replace it with a _
551 if " " in customer_name:
552 customer_name = customer_name.replace(" ", "_")
553 customer_upload_folder = os.path.join(UPLOAD_FOLDER, customer_name)
554 async with aiofiles.open(os.path.join(customer_upload_folder, filename), "wb") as f:
555 await f.write(file.encode())
556 return os.path.join(customer_upload_folder, filename)
557
558
559 async def base64_auth_header_generator(username: str, password: str):
560 """
561 Generate the basic authentication header for the given username and password.
562
563 Args:
564 username (str): The username.
565 password (str): The password.
566
567 Returns:
568 str: The basic authentication header.
569 """
570 return f"Basic {base64.b64encode(f'{username}:{password}'.encode()).decode()}"
571
572
573 async def load_and_replace_config_json(
574 customer_details: BitdefenderCustomerDetails,
575 keys: ProvisionBitdefenderAuthKeys,
576 session: AsyncSession,
577 ):
578 """
579 Load the config.json file and replace the placeholders with the customer details.
580
581 Args:
582 customer_details (BitdefenderCustomerDetails): The details of the customer.
583 keys (ProvisionBitdefenderAuthKeys): The authentication keys for Bitdefender.
584
585 Returns:
586 str: The path to the modified config.json file.
587 """
588 connector_url = str(await get_connector_attribute(connector_id=3, column_name="connector_url", session=session))
589 connector_url = connector_url.replace("https://", "").replace("http://", "").replace(":9000", "")
590 encoded_string = await base64_auth_header_generator(keys.BASIC_AUTH_USERNAME, keys.BASIC_AUTH_PASSWORD)
591
592 # Define the JSON structure
593 data = {
594 "port": int(keys.WEBSERVER_PORT),
595 "syslog_port": int(keys.GRAYLOG_PORT),
596 "transport": customer_details.protocal_type,
597 "target": connector_url,
598 "authentication_string": encoded_string,
599 "secure": {
600 "enabled": True,
601 "key": "api/config/server.key",
602 "cert": "api/config/server.crt",
603 },
604 }
605
606 # If customer name contains a space, replace it with a _
607 if " " in customer_details.customer_name:
608 customer_details.customer_name = customer_details.customer_name.replace(" ", "_")
609
610 customer_upload_folder = os.path.join(UPLOAD_FOLDER, customer_details.customer_name)
611 os.makedirs(customer_upload_folder, exist_ok=True)
612
613 # Save the modified content back to the config.json file
614 config_path = os.path.join(customer_upload_folder, "config.json")
615 async with aiofiles.open(config_path, "w") as f:
616 await f.write(json.dumps(data, indent=4))
617
618 return config_path
619
620
621 async def update_customer_integration_table(
622 customer_code: str,
623 session: AsyncSession,
624 ) -> None:
625 """
626 Updates the `customer_integrations` table to set the `deployed` column to True where the `customer_code`
627 matches the given customer code and the `integration_service_name` is "Bitdefender".
628
629 Args:
630 customer_code (str): The customer code.
631 session (AsyncSession): The async session object for making HTTP requests.
632 """
633 logger.info(f"Updating customer integrations table for customer {customer_code}")
634 await session.execute(
635 update(CustomerIntegrations)
636 .where(
637 and_(
638 CustomerIntegrations.customer_code == customer_code,
639 CustomerIntegrations.integration_service_name == "BitDefender",
640 ),
641 )
642 .values(deployed=True),
643 )
644 await session.commit()
645
646 return None