| 1 | from typing import List |
| 2 | from typing import Optional |
| 3 | |
| 4 | from fastapi import APIRouter |
| 5 | from fastapi import Depends |
| 6 | from fastapi import HTTPException |
| 7 | from fastapi import Security |
| 8 | from loguru import logger |
| 9 | from sqlalchemy import delete |
| 10 | from sqlalchemy import update |
| 11 | from sqlalchemy.exc import NoResultFound |
| 12 | from sqlalchemy.ext.asyncio import AsyncSession |
| 13 | from sqlalchemy.future import select |
| 14 | from sqlalchemy.orm import joinedload |
| 15 | |
| 16 | from app.auth.utils import AuthHandler |
| 17 | from app.connectors.grafana.services.folders import delete_folder |
| 18 | from app.connectors.graylog.services.management import delete_index_by_id |
| 19 | from app.connectors.graylog.services.streams import delete_stream |
| 20 | from app.customer_provisioning.services.grafana import delete_grafana_datasource |
| 21 | from app.db.db_session import get_db |
| 22 | from app.db.universal_models import Customers |
| 23 | from app.db.universal_models import CustomersMeta |
| 24 | from app.integrations.alert_creation_settings.models.alert_creation_settings import ( |
| 25 | AlertCreationSettings, |
| 26 | ) |
| 27 | from app.integrations.models.customer_integration_settings import AvailableIntegrations |
| 28 | from app.integrations.models.customer_integration_settings import CustomerIntegrations |
| 29 | from app.integrations.models.customer_integration_settings import ( |
| 30 | CustomerIntegrationsMeta, |
| 31 | ) |
| 32 | from app.integrations.models.customer_integration_settings import IntegrationAuthKeys |
| 33 | from app.integrations.models.customer_integration_settings import IntegrationConfig |
| 34 | from app.integrations.models.customer_integration_settings import IntegrationService |
| 35 | from app.integrations.models.customer_integration_settings import ( |
| 36 | IntegrationSubscription, |
| 37 | ) |
| 38 | from app.integrations.schema import AuthKey |
| 39 | from app.integrations.schema import AvailableIntegrationsResponse |
| 40 | from app.integrations.schema import CreateIntegrationAuthKeys |
| 41 | from app.integrations.schema import CreateIntegrationService |
| 42 | from app.integrations.schema import CustomerByAuthKeyResponse |
| 43 | from app.integrations.schema import CustomerIntegrationCreate |
| 44 | from app.integrations.schema import CustomerIntegrationCreateResponse |
| 45 | from app.integrations.schema import CustomerIntegrationDeleteResponse |
| 46 | from app.integrations.schema import CustomerIntegrationsMetaResponse |
| 47 | from app.integrations.schema import CustomerIntegrationsMetaSchema |
| 48 | from app.integrations.schema import CustomerIntegrationsResponse |
| 49 | from app.integrations.schema import DeleteCustomerIntegration |
| 50 | from app.integrations.schema import IntegrationWithAuthKeys |
| 51 | from app.integrations.schema import UpdateCustomerIntegration |
| 52 | from app.integrations.schema import UpdateMetaAutoRequest |
| 53 | from app.integrations.schema import UpdateMetaResponse |
| 54 | from app.network_connectors.models.network_connectors import ( |
| 55 | CustomerNetworkConnectorsMeta, |
| 56 | ) |
| 57 | |
| 58 | integration_settings_router = APIRouter() |
| 59 | |
| 60 | NETWORK_INTEGRATIONS = [ |
| 61 | "DefenderForEndpoint", |
| 62 | "BITDEFENDER", |
| 63 | "CROWDSTRIKE", |
| 64 | "FORTINET", |
| 65 | "Fortinet", |
| 66 | "Sonicwall", |
| 67 | # Add other network integrations as needed |
| 68 | ] |
| 69 | |
| 70 | |
| 71 | async def fetch_available_integrations(session: AsyncSession): |
| 72 | """ |
| 73 | Fetches available integrations and their auth keys from the database. |
| 74 | |
| 75 | Args: |
| 76 | session (AsyncSession): The database session. |
| 77 | |
| 78 | Returns: |
| 79 | List[IntegrationWithAuthKeys]: A list of available integrations with their auth keys. |
| 80 | """ |
| 81 | stmt = select(AvailableIntegrations).options( |
| 82 | joinedload(AvailableIntegrations.auth_keys), |
| 83 | ) |
| 84 | result = await session.execute(stmt) |
| 85 | |
| 86 | # Use unique() to avoid duplicates caused by joined eager loading |
| 87 | unique_integrations = result.unique().scalars().all() |
| 88 | |
| 89 | integrations_with_auth_keys = [] |
| 90 | for integration in unique_integrations: |
| 91 | auth_keys = [AuthKey(auth_key_name=key.auth_key_name) for key in integration.auth_keys] |
| 92 | integration_data = IntegrationWithAuthKeys( |
| 93 | id=integration.id, |
| 94 | integration_name=integration.integration_name, |
| 95 | description=integration.description, |
| 96 | integration_details=integration.integration_details, |
| 97 | auth_keys=auth_keys, |
| 98 | ) |
| 99 | integrations_with_auth_keys.append(integration_data) |
| 100 | |
| 101 | return integrations_with_auth_keys |
| 102 | |
| 103 | |
| 104 | async def validate_integration_name(integration_name: str, session: AsyncSession): |
| 105 | """ |
| 106 | Validate if the integration name exists in available integrations. |
| 107 | """ |
| 108 | available_integrations = await fetch_available_integrations(session) |
| 109 | if integration_name not in [ai.integration_name for ai in available_integrations]: |
| 110 | raise HTTPException( |
| 111 | status_code=400, |
| 112 | detail=f"Integration {integration_name} is not a valid integration.", |
| 113 | ) |
| 114 | |
| 115 | |
| 116 | async def validate_integration_auth_keys( |
| 117 | integration_name: str, |
| 118 | integration_auth_keys: List[AuthKey], |
| 119 | session: AsyncSession, |
| 120 | ): |
| 121 | """ |
| 122 | Validate if the integration auth keys are valid. |
| 123 | """ |
| 124 | available_integrations = await fetch_available_integrations(session) |
| 125 | integration = [ai for ai in available_integrations if ai.integration_name == integration_name][0] |
| 126 | available_auth_keys = [ak.auth_key_name for ak in integration.auth_keys] |
| 127 | # loop through the `available_auth_keys` and check if the `integration_auth_keys` contains the `auth_key_name` |
| 128 | for auth_key in available_auth_keys: |
| 129 | if auth_key not in [iak.auth_key_name for iak in integration_auth_keys]: |
| 130 | raise HTTPException( |
| 131 | status_code=400, |
| 132 | detail=f"Integration auth key {auth_key} does not exist.", |
| 133 | ) |
| 134 | |
| 135 | |
| 136 | async def validate_integration_auth_key_update( |
| 137 | integration_name: str, |
| 138 | integration_auth_key: List[AuthKey], |
| 139 | session: AsyncSession, |
| 140 | ): |
| 141 | """ |
| 142 | Validate if the integration auth key is valid. |
| 143 | """ |
| 144 | logger.info(f"integration_auth_key: {integration_auth_key}") |
| 145 | available_integrations = await fetch_available_integrations(session) |
| 146 | integration = [ai for ai in available_integrations if ai.integration_name == integration_name][0] |
| 147 | available_auth_keys = [ak.auth_key_name for ak in integration.auth_keys] |
| 148 | for auth_key in integration_auth_key: |
| 149 | if auth_key.auth_key_name not in available_auth_keys: |
| 150 | raise HTTPException( |
| 151 | status_code=400, |
| 152 | detail=f"Integration auth key {auth_key.auth_key_name} does not exist.", |
| 153 | ) |
| 154 | |
| 155 | |
| 156 | async def validate_customer_code(customer_code: str, session: AsyncSession): |
| 157 | """ |
| 158 | Validate if the customer code exists in the customers table. |
| 159 | """ |
| 160 | stmt = select(Customers).where(Customers.customer_code == customer_code) |
| 161 | result = await session.execute(stmt) |
| 162 | if result.scalars().first() is None: |
| 163 | raise HTTPException( |
| 164 | status_code=400, |
| 165 | detail=f"Customer {customer_code} does not exist.", |
| 166 | ) |
| 167 | |
| 168 | |
| 169 | async def validate_customer_meta(customer_code: str, session: AsyncSession): |
| 170 | """ |
| 171 | Validate if the customer code exists in the customers_meta table. |
| 172 | """ |
| 173 | stmt = select(CustomersMeta).where(CustomersMeta.customer_code == customer_code) |
| 174 | result = await session.execute(stmt) |
| 175 | if result.scalars().first() is None: |
| 176 | raise HTTPException( |
| 177 | status_code=400, |
| 178 | detail=f"Customer {customer_code} meta does not exist. Please provision the customer before creating an integration.", |
| 179 | ) |
| 180 | |
| 181 | |
| 182 | async def check_existing_customer_integration( |
| 183 | customer_code: str, |
| 184 | integration_name: str, |
| 185 | session: AsyncSession, |
| 186 | ): |
| 187 | """ |
| 188 | Check if the customer integration already exists. |
| 189 | """ |
| 190 | # Assuming IntegrationService has an 'integration_name' field or similar |
| 191 | stmt = ( |
| 192 | select(CustomerIntegrations) |
| 193 | .join(CustomerIntegrations.integration_subscriptions) |
| 194 | .join(IntegrationSubscription.integration_service) |
| 195 | .where( |
| 196 | CustomerIntegrations.customer_code == customer_code, |
| 197 | IntegrationService.service_name == integration_name, |
| 198 | ) |
| 199 | ) |
| 200 | result = await session.execute(stmt) |
| 201 | if result.scalars().first() is not None: |
| 202 | raise HTTPException( |
| 203 | status_code=400, |
| 204 | detail=f"Customer integration {customer_code} {integration_name} already exists.", |
| 205 | ) |
| 206 | |
| 207 | |
| 208 | async def check_existing_customer_integration_meta( |
| 209 | customer_code: str, |
| 210 | integration_name: str, |
| 211 | session: AsyncSession, |
| 212 | ): |
| 213 | """ |
| 214 | Check if the customer integration meta already exists for the customer code and integration name. |
| 215 | """ |
| 216 | stmt = select(CustomerIntegrationsMeta).where( |
| 217 | CustomerIntegrationsMeta.customer_code == customer_code, |
| 218 | CustomerIntegrationsMeta.integration_name == integration_name, |
| 219 | ) |
| 220 | result = await session.execute(stmt) |
| 221 | if result.scalars().first() is not None: |
| 222 | raise HTTPException( |
| 223 | status_code=400, |
| 224 | detail=f"Customer integration meta {customer_code} {integration_name} already exists.", |
| 225 | ) |
| 226 | |
| 227 | |
| 228 | async def create_integration_service( |
| 229 | integration_name: str, |
| 230 | settings: CreateIntegrationService, |
| 231 | session: AsyncSession, |
| 232 | ) -> IntegrationService: |
| 233 | """ |
| 234 | Create or fetch IntegrationService instance with custom configuration. |
| 235 | """ |
| 236 | integration_service = IntegrationService( |
| 237 | service_name=integration_name, |
| 238 | auth_type=settings.auth_type, |
| 239 | configs=[ |
| 240 | IntegrationConfig( |
| 241 | config_key=settings.config_key, |
| 242 | config_value=settings.config_value, |
| 243 | ), |
| 244 | ], |
| 245 | ) |
| 246 | session.add(integration_service) |
| 247 | await session.flush() |
| 248 | return integration_service |
| 249 | |
| 250 | |
| 251 | async def create_customer_integrations( |
| 252 | customer_code: str, |
| 253 | customer_name: str, |
| 254 | integration_service_id: int, |
| 255 | integration_service_name: str, |
| 256 | session: AsyncSession, |
| 257 | ) -> CustomerIntegrations: |
| 258 | """ |
| 259 | Create CustomerIntegrations instance. |
| 260 | """ |
| 261 | customer_integrations = CustomerIntegrations( |
| 262 | customer_code=customer_code, |
| 263 | customer_name=customer_name, |
| 264 | integration_service_id=integration_service_id, |
| 265 | integration_service_name=integration_service_name, |
| 266 | deployed=False, |
| 267 | ) |
| 268 | session.add(customer_integrations) |
| 269 | await session.flush() |
| 270 | return customer_integrations |
| 271 | |
| 272 | |
| 273 | async def create_integration_subscription( |
| 274 | customer_integrations: CustomerIntegrations, |
| 275 | integration_service: IntegrationService, |
| 276 | integration_auth_keys: List[CreateIntegrationAuthKeys], |
| 277 | session: AsyncSession, |
| 278 | ): |
| 279 | """ |
| 280 | Create IntegrationSubscription instance. |
| 281 | """ |
| 282 | logger.info(f"integration_auth_keys: {integration_auth_keys}") |
| 283 | for auth_key in integration_auth_keys: |
| 284 | new_integration_subscription = IntegrationSubscription( |
| 285 | customer_integrations=customer_integrations, |
| 286 | integration_service=integration_service, |
| 287 | integration_auth_keys=[ |
| 288 | IntegrationAuthKeys( |
| 289 | auth_key_name=auth_key.auth_key_name, |
| 290 | auth_value=auth_key.auth_value, |
| 291 | ), |
| 292 | ], |
| 293 | ) |
| 294 | session.add(new_integration_subscription) |
| 295 | await session.commit() |
| 296 | |
| 297 | |
| 298 | async def get_customer_and_service_ids(session, customer_code, integration_name): |
| 299 | try: |
| 300 | result = await session.execute( |
| 301 | select(CustomerIntegrations.id, IntegrationService.id) |
| 302 | .join( |
| 303 | IntegrationSubscription, |
| 304 | CustomerIntegrations.id == IntegrationSubscription.customer_id, |
| 305 | ) |
| 306 | .join( |
| 307 | IntegrationService, |
| 308 | IntegrationSubscription.integration_service_id == IntegrationService.id, |
| 309 | ) |
| 310 | .where( |
| 311 | CustomerIntegrations.customer_code == customer_code, |
| 312 | IntegrationService.service_name == integration_name, |
| 313 | ), |
| 314 | ) |
| 315 | return result.all() |
| 316 | except NoResultFound: |
| 317 | raise HTTPException(status_code=404, detail="Customer integration not found") |
| 318 | |
| 319 | |
| 320 | async def get_subscription_ids(session, customer_id, integration_service_id): |
| 321 | result = await session.execute( |
| 322 | select(IntegrationSubscription.id).where( |
| 323 | IntegrationSubscription.customer_id == customer_id, |
| 324 | IntegrationSubscription.integration_service_id == integration_service_id, |
| 325 | ), |
| 326 | ) |
| 327 | # Fetch all results |
| 328 | subscription_ids_raw = result.scalars().all() |
| 329 | |
| 330 | # Process the results |
| 331 | # If the result is a list of tuples (even with one element), extract the first element |
| 332 | if subscription_ids_raw and isinstance(subscription_ids_raw[0], tuple): |
| 333 | return [id_tuple[0] for id_tuple in subscription_ids_raw] |
| 334 | # If the result is a list of integers |
| 335 | elif subscription_ids_raw and isinstance(subscription_ids_raw[0], int): |
| 336 | return subscription_ids_raw |
| 337 | # If there are no results |
| 338 | else: |
| 339 | return [] |
| 340 | |
| 341 | |
| 342 | async def delete_metadata(session, subscription_ids): |
| 343 | await session.execute( |
| 344 | delete(IntegrationAuthKeys).where( |
| 345 | IntegrationAuthKeys.subscription_id.in_(subscription_ids), |
| 346 | ), |
| 347 | ) |
| 348 | |
| 349 | |
| 350 | async def delete_subscriptions(session, subscription_ids): |
| 351 | await session.execute( |
| 352 | delete(IntegrationSubscription).where( |
| 353 | IntegrationSubscription.id.in_(subscription_ids), |
| 354 | ), |
| 355 | ) |
| 356 | |
| 357 | |
| 358 | async def delete_configs(session, integration_service_id): |
| 359 | await session.execute( |
| 360 | delete(IntegrationConfig).where( |
| 361 | IntegrationConfig.integration_service_id == integration_service_id, |
| 362 | ), |
| 363 | ) |
| 364 | |
| 365 | |
| 366 | async def delete_integration_service(session, integration_service_id): |
| 367 | await session.execute( |
| 368 | delete(IntegrationService).where( |
| 369 | IntegrationService.id == integration_service_id, |
| 370 | ), |
| 371 | ) |
| 372 | |
| 373 | |
| 374 | async def delete_customer_integration_record(session, customer_id): |
| 375 | await session.execute( |
| 376 | delete(CustomerIntegrations).where(CustomerIntegrations.id == customer_id), |
| 377 | ) |
| 378 | |
| 379 | |
| 380 | async def find_customer_integration( |
| 381 | customer_code: str, |
| 382 | integration_name: str, |
| 383 | customer_integration_response, |
| 384 | ) -> Optional[CustomerIntegrations]: |
| 385 | for ci in customer_integration_response.available_integrations: |
| 386 | for subscription in ci.integration_subscriptions: |
| 387 | if subscription.integration_service.service_name == integration_name: |
| 388 | return ci |
| 389 | return None |
| 390 | |
| 391 | |
| 392 | def get_subscription_id( |
| 393 | customer_integration, |
| 394 | integration_name: str, |
| 395 | auth_key_name: str, |
| 396 | ) -> Optional[int]: |
| 397 | for subscription in customer_integration.integration_subscriptions: |
| 398 | if subscription.integration_service.service_name == integration_name: |
| 399 | for auth_key in subscription.integration_auth_keys: |
| 400 | if auth_key.auth_key_name == auth_key_name: |
| 401 | return subscription.id |
| 402 | return None |
| 403 | |
| 404 | |
| 405 | async def get_tenant_id( |
| 406 | customer_integration: CustomerIntegrationCreate, |
| 407 | session: AsyncSession, |
| 408 | ) -> str: |
| 409 | """ |
| 410 | Retrieves the Tenant ID for a given customer integration. This is the Office365 organization ID and |
| 411 | is used to create alerts for the customer in DFIR-IRIS. |
| 412 | """ |
| 413 | stmt = ( |
| 414 | select(IntegrationAuthKeys) |
| 415 | .join( |
| 416 | IntegrationSubscription, |
| 417 | IntegrationAuthKeys.subscription_id == IntegrationSubscription.id, |
| 418 | ) |
| 419 | .join( |
| 420 | CustomerIntegrations, |
| 421 | IntegrationSubscription.customer_id == CustomerIntegrations.id, |
| 422 | ) |
| 423 | .join( |
| 424 | IntegrationService, |
| 425 | IntegrationSubscription.integration_service_id == IntegrationService.id, |
| 426 | ) |
| 427 | .where( |
| 428 | CustomerIntegrations.customer_code == customer_integration.customer_code, |
| 429 | IntegrationService.service_name == customer_integration.integration_name, |
| 430 | IntegrationAuthKeys.auth_key_name == "TENANT_ID", |
| 431 | ) |
| 432 | ) |
| 433 | |
| 434 | result = await session.execute(stmt) |
| 435 | tenant_id = result.scalars().first() |
| 436 | if tenant_id is None: |
| 437 | raise HTTPException( |
| 438 | status_code=404, |
| 439 | detail=f"Tenant ID for customer {customer_integration.customer_code} not found.", |
| 440 | ) |
| 441 | logger.info(f"tenant_id: {tenant_id.auth_value}") |
| 442 | return tenant_id.auth_value |
| 443 | |
| 444 | |
| 445 | async def update_office365_organization_id( |
| 446 | customer_code: str, |
| 447 | tenant_id: str, |
| 448 | session: AsyncSession, |
| 449 | ): |
| 450 | """ |
| 451 | Updates the Office365 organization ID in the alert_creation_settings table. |
| 452 | """ |
| 453 | stmt = ( |
| 454 | update(AlertCreationSettings) |
| 455 | .where(AlertCreationSettings.customer_code == customer_code) |
| 456 | .values(office365_organization_id=tenant_id) |
| 457 | ) |
| 458 | await session.execute(stmt) |
| 459 | await session.commit() |
| 460 | |
| 461 | |
| 462 | async def get_integration_service_id( |
| 463 | integration_name: str, |
| 464 | session: AsyncSession, |
| 465 | ) -> int: |
| 466 | """ |
| 467 | Retrieves the AvailableIntegrations ID for a given integration name. |
| 468 | """ |
| 469 | stmt = select(AvailableIntegrations).where( |
| 470 | AvailableIntegrations.integration_name == integration_name, |
| 471 | ) |
| 472 | result = await session.execute(stmt) |
| 473 | integration_service = result.scalars().first() |
| 474 | if integration_service is None: |
| 475 | raise HTTPException( |
| 476 | status_code=404, |
| 477 | detail=f"Integration service {integration_name} not found.", |
| 478 | ) |
| 479 | return integration_service.id |
| 480 | |
| 481 | |
| 482 | async def get_integration_service_name( |
| 483 | integration_name: str, |
| 484 | session: AsyncSession, |
| 485 | ) -> str: |
| 486 | """ |
| 487 | Retrieves the AvailableIntegrations ID for a given integration name. |
| 488 | """ |
| 489 | stmt = select(AvailableIntegrations).where( |
| 490 | AvailableIntegrations.integration_name == integration_name, |
| 491 | ) |
| 492 | result = await session.execute(stmt) |
| 493 | integration_service = result.scalars().first() |
| 494 | if integration_service is None: |
| 495 | raise HTTPException( |
| 496 | status_code=404, |
| 497 | detail=f"Integration service {integration_name} not found.", |
| 498 | ) |
| 499 | return integration_service.integration_name |
| 500 | |
| 501 | |
| 502 | async def fetch_customer_integrations_data(session: AsyncSession): |
| 503 | """ |
| 504 | Fetches customer integrations data from the database. |
| 505 | """ |
| 506 | stmt = select(CustomerIntegrations).options( |
| 507 | joinedload(CustomerIntegrations.integration_subscriptions).joinedload( |
| 508 | IntegrationSubscription.integration_service, |
| 509 | ), |
| 510 | joinedload(CustomerIntegrations.integration_subscriptions).subqueryload( |
| 511 | IntegrationSubscription.integration_auth_keys, |
| 512 | ), |
| 513 | ) |
| 514 | result = await session.execute(stmt) |
| 515 | return result.scalars().unique().all() |
| 516 | |
| 517 | |
| 518 | def process_customer_integrations(customer_integrations_data): |
| 519 | """ |
| 520 | Processes customer integrations data and returns a list of CustomerIntegrations objects. |
| 521 | """ |
| 522 | processed_customer_integrations = [] |
| 523 | for ci in customer_integrations_data: |
| 524 | first_service_id = ci.integration_subscriptions[0].integration_service_id if ci.integration_subscriptions else None |
| 525 | customer_integration_obj = CustomerIntegrations( |
| 526 | id=ci.id, |
| 527 | customer_code=ci.customer_code, |
| 528 | customer_name=ci.customer_name, |
| 529 | integration_subscriptions=ci.integration_subscriptions, |
| 530 | integration_service_id=first_service_id, |
| 531 | integration_service_name=ci.integration_subscriptions[0].integration_service.service_name |
| 532 | if ci.integration_subscriptions |
| 533 | else None, |
| 534 | deployed=ci.deployed, |
| 535 | ) |
| 536 | processed_customer_integrations.append(customer_integration_obj) |
| 537 | return processed_customer_integrations |
| 538 | |
| 539 | |
| 540 | def generate_integration_response(customer_code: str, integration_name: str) -> CustomerIntegrationCreateResponse: |
| 541 | additional_info_map = { |
| 542 | "Office365": ( |
| 543 | "Make sure to update the Office365 integration block in the Wazuh Manager ossec.conf file and restart the Wazuh Manager service. " |
| 544 | "Also make sure to update the Office365 Graylog stream rule for this customer with the new organization ID if this has changed. " |
| 545 | "YouTube video: https://youtu.be/ihj2F2rA6BQ?si=p4c8Xnk6PX8r29IB" |
| 546 | ), |
| 547 | "Crowdstrike": ( |
| 548 | "Make sure to update the Crowdstrike docker application with the new connection details and restart the docker container. " |
| 549 | "YouTube video: https://youtu.be/YOVUOpZDEzM?si=jzpHw8vcnqnfVPzt" |
| 550 | ), |
| 551 | "BitDefender": ( |
| 552 | "Make sure to update the BitDefender docker application with the new connection details and restart the docker container." |
| 553 | ), |
| 554 | } |
| 555 | |
| 556 | additional_info = additional_info_map.get(integration_name, "") |
| 557 | if additional_info == "": |
| 558 | additional_info = None |
| 559 | |
| 560 | return CustomerIntegrationCreateResponse( |
| 561 | message=f"Customer integration {customer_code} {integration_name} successfully updated.", |
| 562 | success=True, |
| 563 | additional_info=additional_info, |
| 564 | ) |
| 565 | |
| 566 | |
| 567 | def generate_decommission_response(customer_code: str, integration_name: str) -> CustomerIntegrationDeleteResponse: |
| 568 | additional_info_map = { |
| 569 | "Office365": ( |
| 570 | "Make sure to remove the Office365 integration block from the Wazuh Manager ossec.conf file and restart the Wazuh Manager service. " |
| 571 | ), |
| 572 | "Crowdstrike": ("Make sure to remove the Crowdstrike docker application."), |
| 573 | "BitDefender": ("Make sure to remove the BitDefender docker application."), |
| 574 | } |
| 575 | |
| 576 | additional_info = additional_info_map.get(integration_name, "") |
| 577 | if additional_info == "": |
| 578 | additional_info = None |
| 579 | |
| 580 | return CustomerIntegrationDeleteResponse( |
| 581 | message=f"Customer integration {customer_code} {integration_name} successfully deleted.", |
| 582 | success=True, |
| 583 | additional_info=additional_info, |
| 584 | ) |
| 585 | |
| 586 | |
| 587 | @integration_settings_router.get( |
| 588 | "/available_integrations", |
| 589 | response_model=AvailableIntegrationsResponse, |
| 590 | description="Get a list of available integrations.", |
| 591 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 592 | ) |
| 593 | async def get_available_integrations( |
| 594 | session: AsyncSession = Depends(get_db), |
| 595 | ): |
| 596 | """ |
| 597 | Endpoint to get a list of available integrations. |
| 598 | """ |
| 599 | available_integrations = await fetch_available_integrations(session) |
| 600 | return AvailableIntegrationsResponse( |
| 601 | available_integrations=available_integrations, |
| 602 | message="Available integrations successfully retrieved.", |
| 603 | success=True, |
| 604 | ) |
| 605 | |
| 606 | |
| 607 | @integration_settings_router.get( |
| 608 | "/customer_integrations", |
| 609 | response_model=CustomerIntegrationsResponse, |
| 610 | description="Get a list of customer integrations.", |
| 611 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 612 | ) |
| 613 | async def get_customer_integrations(session: AsyncSession = Depends(get_db)): |
| 614 | """ |
| 615 | Endpoint to get a list of customer integrations. |
| 616 | """ |
| 617 | customer_integrations_data = await fetch_customer_integrations_data(session) |
| 618 | processed_customer_integrations = process_customer_integrations( |
| 619 | customer_integrations_data, |
| 620 | ) |
| 621 | |
| 622 | logger.info(f"Processed customer_integrations: {processed_customer_integrations}") |
| 623 | return CustomerIntegrationsResponse( |
| 624 | available_integrations=processed_customer_integrations, |
| 625 | message="Customer integrations successfully retrieved.", |
| 626 | success=True, |
| 627 | ) |
| 628 | |
| 629 | |
| 630 | @integration_settings_router.get( |
| 631 | "/customer_integrations_meta", |
| 632 | response_model=CustomerIntegrationsMetaResponse, |
| 633 | description="Get a list of customer integrations metadata.", |
| 634 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 635 | ) |
| 636 | async def get_customer_integrations_meta(session: AsyncSession = Depends(get_db)): |
| 637 | """ |
| 638 | Endpoint to get a list of customer integrations metadata. |
| 639 | """ |
| 640 | try: |
| 641 | stmt = select(CustomerIntegrationsMeta) |
| 642 | result = await session.execute(stmt) |
| 643 | customer_integrations_meta = result.scalars().all() |
| 644 | except Exception as e: |
| 645 | logger.error(f"Error while fetching customer integrations metadata: {e}") |
| 646 | customer_integrations_meta = [] |
| 647 | |
| 648 | logger.info(f"customer_integrations_meta: {customer_integrations_meta}") |
| 649 | return CustomerIntegrationsMetaResponse( |
| 650 | customer_integrations_meta=customer_integrations_meta, |
| 651 | message="Customer integrations metadata successfully retrieved.", |
| 652 | success=True, |
| 653 | ) |
| 654 | |
| 655 | |
| 656 | @integration_settings_router.get( |
| 657 | "/customer_integrations/{customer_code}", |
| 658 | response_model=CustomerIntegrationsResponse, |
| 659 | description="Get a list of customer integrations for a specific customer.", |
| 660 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 661 | ) |
| 662 | async def get_customer_integrations_by_customer_code( |
| 663 | customer_code: str, |
| 664 | session: AsyncSession = Depends(get_db), |
| 665 | ): |
| 666 | """ |
| 667 | Endpoint to get a list of customer integrations for a specific customer. |
| 668 | """ |
| 669 | stmt = ( |
| 670 | select(CustomerIntegrations) |
| 671 | .options( |
| 672 | joinedload(CustomerIntegrations.integration_subscriptions).joinedload( |
| 673 | IntegrationSubscription.integration_service, |
| 674 | ), |
| 675 | joinedload(CustomerIntegrations.integration_subscriptions).subqueryload( |
| 676 | IntegrationSubscription.integration_auth_keys, |
| 677 | ), # Load IntegrationAuthKeys |
| 678 | ) |
| 679 | .where(CustomerIntegrations.customer_code == customer_code) |
| 680 | ) |
| 681 | result = await session.execute(stmt) |
| 682 | customer_integrations = result.scalars().unique().all() |
| 683 | return CustomerIntegrationsResponse( |
| 684 | available_integrations=customer_integrations, |
| 685 | message="Customer integrations successfully retrieved.", |
| 686 | success=True, |
| 687 | ) |
| 688 | |
| 689 | |
| 690 | @integration_settings_router.get( |
| 691 | "/customer_integrations_meta/{customer_code}", |
| 692 | response_model=CustomerIntegrationsMetaResponse, |
| 693 | description="Get a list of customer integrations metadata for a specific customer.", |
| 694 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 695 | ) |
| 696 | async def get_customer_integrations_meta_by_customer_code( |
| 697 | customer_code: str, |
| 698 | session: AsyncSession = Depends(get_db), |
| 699 | ): |
| 700 | """ |
| 701 | Endpoint to get a list of customer integrations metadata for a specific customer. |
| 702 | """ |
| 703 | stmt = select(CustomerIntegrationsMeta).where( |
| 704 | CustomerIntegrationsMeta.customer_code == customer_code, |
| 705 | ) |
| 706 | result = await session.execute(stmt) |
| 707 | customer_integrations_meta = result.scalars().all() |
| 708 | logger.info(f"customer_integrations_meta: {customer_integrations_meta}") |
| 709 | return CustomerIntegrationsMetaResponse( |
| 710 | customer_integrations_meta=customer_integrations_meta, |
| 711 | message="Customer integrations metadata successfully retrieved.", |
| 712 | success=True, |
| 713 | ) |
| 714 | |
| 715 | |
| 716 | @integration_settings_router.post( |
| 717 | "/create_integration", |
| 718 | response_model=CustomerIntegrationCreateResponse, |
| 719 | description="Create a new customer integration.", |
| 720 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 721 | ) |
| 722 | async def create_integration( |
| 723 | customer_integration_create: CustomerIntegrationCreate, |
| 724 | session: AsyncSession = Depends(get_db), |
| 725 | ): |
| 726 | """ |
| 727 | Endpoint to create a new customer integration. |
| 728 | """ |
| 729 | await validate_integration_name( |
| 730 | customer_integration_create.integration_name, |
| 731 | session, |
| 732 | ) |
| 733 | await validate_integration_auth_keys( |
| 734 | customer_integration_create.integration_name, |
| 735 | customer_integration_create.integration_auth_keys, |
| 736 | session, |
| 737 | ) |
| 738 | await validate_customer_code(customer_integration_create.customer_code, session) |
| 739 | await validate_customer_meta(customer_integration_create.customer_code, session) |
| 740 | await check_existing_customer_integration( |
| 741 | customer_integration_create.customer_code, |
| 742 | customer_integration_create.integration_name, |
| 743 | session, |
| 744 | ) |
| 745 | integration_service_id = await get_integration_service_id( |
| 746 | customer_integration_create.integration_name, |
| 747 | session, |
| 748 | ) |
| 749 | integration_service_name = await get_integration_service_name( |
| 750 | customer_integration_create.integration_name, |
| 751 | session, |
| 752 | ) |
| 753 | |
| 754 | integration_service = await create_integration_service( |
| 755 | customer_integration_create.integration_name, |
| 756 | settings=customer_integration_create.integration_config, |
| 757 | session=session, |
| 758 | ) |
| 759 | customer_integrations = await create_customer_integrations( |
| 760 | customer_integration_create.customer_code, |
| 761 | customer_integration_create.customer_name, |
| 762 | integration_service_id=integration_service_id, |
| 763 | integration_service_name=integration_service_name, |
| 764 | session=session, |
| 765 | ) |
| 766 | await create_integration_subscription( |
| 767 | customer_integrations, |
| 768 | integration_service, |
| 769 | integration_auth_keys=customer_integration_create.integration_auth_keys, |
| 770 | session=session, |
| 771 | ) |
| 772 | |
| 773 | # Office365 specific integration handling |
| 774 | if customer_integration_create.integration_name == "Office365": |
| 775 | tenant_id = await get_tenant_id(customer_integration_create, session) |
| 776 | await update_office365_organization_id( |
| 777 | customer_integration_create.customer_code, |
| 778 | tenant_id, |
| 779 | session, |
| 780 | ) |
| 781 | |
| 782 | return CustomerIntegrationCreateResponse( |
| 783 | message=f"Customer integration {customer_integration_create.customer_code} {customer_integration_create.integration_name} successfully created.", |
| 784 | success=True, |
| 785 | ) |
| 786 | |
| 787 | |
| 788 | @integration_settings_router.post( |
| 789 | "/create_integration_meta", |
| 790 | response_model=CustomerIntegrationsMetaResponse, |
| 791 | description="Create a new customer integration metadata.", |
| 792 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 793 | ) |
| 794 | async def create_integration_meta( |
| 795 | customer_integration_meta: CustomerIntegrationsMetaSchema, |
| 796 | session: AsyncSession = Depends(get_db), |
| 797 | ): |
| 798 | """ |
| 799 | Endpoint to create a new customer integration metadata. |
| 800 | """ |
| 801 | await validate_customer_code(customer_integration_meta.customer_code, session) |
| 802 | await validate_customer_meta(customer_integration_meta.customer_code, session) |
| 803 | await check_existing_customer_integration_meta( |
| 804 | customer_integration_meta.customer_code, |
| 805 | customer_integration_meta.integration_name, |
| 806 | session, |
| 807 | ) |
| 808 | try: |
| 809 | new_customer_integration_meta = CustomerIntegrationsMeta( |
| 810 | **customer_integration_meta.model_dump(), |
| 811 | ) |
| 812 | session.add(new_customer_integration_meta) |
| 813 | await session.commit() |
| 814 | return CustomerIntegrationsMetaResponse( |
| 815 | message="Customer integration metadata successfully created.", |
| 816 | success=True, |
| 817 | ) |
| 818 | except Exception as e: |
| 819 | logger.error(f"Error while creating customer integration metadata: {e}") |
| 820 | return CustomerIntegrationsMetaResponse( |
| 821 | customer_integrations_meta=None, |
| 822 | message="Error while creating customer integration metadata.", |
| 823 | success=False, |
| 824 | ) |
| 825 | |
| 826 | |
| 827 | @integration_settings_router.put( |
| 828 | "/update_integration/{customer_code}", |
| 829 | response_model=CustomerIntegrationCreateResponse, |
| 830 | description="Update a customer integration.", |
| 831 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 832 | ) |
| 833 | async def update_integration( |
| 834 | customer_code: str, |
| 835 | customer_integration_update: UpdateCustomerIntegration, |
| 836 | session: AsyncSession = Depends(get_db), |
| 837 | ): |
| 838 | await validate_integration_name( |
| 839 | customer_integration_update.integration_name, |
| 840 | session, |
| 841 | ) |
| 842 | customer_integration_response = await get_customer_integrations_by_customer_code( |
| 843 | customer_code, |
| 844 | session, |
| 845 | ) |
| 846 | |
| 847 | if not customer_integration_response: |
| 848 | raise HTTPException(status_code=404, detail="Customer integrations not found") |
| 849 | |
| 850 | customer_integration = await find_customer_integration( |
| 851 | customer_code, |
| 852 | customer_integration_update.integration_name, |
| 853 | customer_integration_response, |
| 854 | ) |
| 855 | |
| 856 | if not customer_integration: |
| 857 | raise HTTPException( |
| 858 | status_code=404, |
| 859 | detail="Customer integration with specified service name not found.", |
| 860 | ) |
| 861 | |
| 862 | await validate_integration_auth_key_update( |
| 863 | customer_integration_update.integration_name, |
| 864 | customer_integration_update.integration_auth_keys, |
| 865 | session, |
| 866 | ) |
| 867 | |
| 868 | for auth_key in customer_integration_update.integration_auth_keys: |
| 869 | subscription_id = get_subscription_id( |
| 870 | customer_integration, |
| 871 | customer_integration_update.integration_name, |
| 872 | auth_key.auth_key_name, |
| 873 | ) |
| 874 | |
| 875 | if not subscription_id: |
| 876 | raise HTTPException( |
| 877 | status_code=404, |
| 878 | detail=f"Integration auth key {auth_key.auth_key_name} not found.", |
| 879 | ) |
| 880 | |
| 881 | await session.execute( |
| 882 | update(IntegrationAuthKeys) |
| 883 | .where(IntegrationAuthKeys.subscription_id == subscription_id) |
| 884 | .values( |
| 885 | auth_value=auth_key.auth_value, |
| 886 | ), |
| 887 | ) |
| 888 | |
| 889 | await session.commit() |
| 890 | |
| 891 | return generate_integration_response(customer_code, customer_integration_update.integration_name) |
| 892 | |
| 893 | |
| 894 | @integration_settings_router.put( |
| 895 | "/available_integrations", |
| 896 | response_model=AvailableIntegrationsResponse, |
| 897 | description="Update an available integration.", |
| 898 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 899 | ) |
| 900 | async def update_available_integrations( |
| 901 | available_integrations: List[AvailableIntegrations], |
| 902 | session: AsyncSession = Depends(get_db), |
| 903 | ): |
| 904 | """ |
| 905 | Endpoint to update an available integration. |
| 906 | """ |
| 907 | for integration in available_integrations: |
| 908 | stmt = select(AvailableIntegrations).where( |
| 909 | AvailableIntegrations.integration_name == integration.integration_name, |
| 910 | ) |
| 911 | result = await session.execute(stmt) |
| 912 | existing_integration = result.scalars().first() |
| 913 | |
| 914 | if existing_integration is None: |
| 915 | raise HTTPException( |
| 916 | status_code=404, |
| 917 | detail=f"Integration {integration.integration_name} not found.", |
| 918 | ) |
| 919 | |
| 920 | existing_integration.description = integration.description |
| 921 | existing_integration.integration_details = integration.integration_details |
| 922 | |
| 923 | await session.commit() |
| 924 | |
| 925 | return AvailableIntegrationsResponse( |
| 926 | available_integrations=available_integrations, |
| 927 | message="Available integrations successfully updated.", |
| 928 | success=True, |
| 929 | ) |
| 930 | |
| 931 | |
| 932 | async def fetch_customer_integration_meta(session: AsyncSession, customer_code: str, integration_name: str): |
| 933 | """ |
| 934 | Fetches customer integrations metadata from the database. |
| 935 | """ |
| 936 | stmt = select(CustomerIntegrationsMeta).where( |
| 937 | CustomerIntegrationsMeta.customer_code == customer_code, |
| 938 | CustomerIntegrationsMeta.integration_name == integration_name, |
| 939 | ) |
| 940 | result = await session.execute(stmt) |
| 941 | return result.scalars().first() |
| 942 | |
| 943 | |
| 944 | async def delete_customer_integration_meta(session: AsyncSession, customer_code: str, integration_name: str): |
| 945 | """ |
| 946 | Deletes customer integrations metadata from the database. |
| 947 | """ |
| 948 | await session.execute( |
| 949 | delete(CustomerIntegrationsMeta).where( |
| 950 | CustomerIntegrationsMeta.customer_code == customer_code, |
| 951 | CustomerIntegrationsMeta.integration_name == integration_name, |
| 952 | ), |
| 953 | ) |
| 954 | |
| 955 | |
| 956 | async def fetch_customer_network_connectors_meta(session: AsyncSession, customer_code: str, network_connector_name: str): |
| 957 | """ |
| 958 | Fetches customer network connectors metadata from the database. |
| 959 | """ |
| 960 | stmt = select(CustomerNetworkConnectorsMeta).where( |
| 961 | CustomerNetworkConnectorsMeta.customer_code == customer_code, |
| 962 | CustomerNetworkConnectorsMeta.network_connector_name == network_connector_name, |
| 963 | ) |
| 964 | result = await session.execute(stmt) |
| 965 | return result.scalars().first() |
| 966 | |
| 967 | |
| 968 | async def delete_customer_network_connectors_meta(session: AsyncSession, customer_code: str, network_connector_name: str): |
| 969 | """ |
| 970 | Deletes customer network connectors metadata from the database. |
| 971 | """ |
| 972 | await session.execute( |
| 973 | delete(CustomerNetworkConnectorsMeta).where( |
| 974 | CustomerNetworkConnectorsMeta.customer_code == customer_code, |
| 975 | CustomerNetworkConnectorsMeta.network_connector_name == network_connector_name, |
| 976 | ), |
| 977 | ) |
| 978 | |
| 979 | |
| 980 | @integration_settings_router.delete( |
| 981 | "/delete_integration", |
| 982 | response_model=CustomerIntegrationDeleteResponse, |
| 983 | description="Delete a customer integration.", |
| 984 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 985 | ) |
| 986 | async def delete_integration( |
| 987 | delete_customer_integration: DeleteCustomerIntegration, |
| 988 | session: AsyncSession = Depends(get_db), |
| 989 | ): |
| 990 | customer_code = delete_customer_integration.customer_code |
| 991 | integration_name = delete_customer_integration.integration_name |
| 992 | |
| 993 | # Check if this is a network integration |
| 994 | is_network_integration = integration_name in NETWORK_INTEGRATIONS |
| 995 | |
| 996 | results = await get_customer_and_service_ids( |
| 997 | session, |
| 998 | customer_code, |
| 999 | integration_name, |
| 1000 | ) |
| 1001 | # Check if results is not empty |
| 1002 | if results: |
| 1003 | # Unpack the first tuple in results |
| 1004 | customer_id, integration_service_id = results[0] |
| 1005 | else: |
| 1006 | # Handle the case where results is empty |
| 1007 | raise HTTPException(status_code=404, detail="Customer integration not found") |
| 1008 | |
| 1009 | # Check if the integration is deployed before proceeding with metadata cleanup |
| 1010 | stmt = select(CustomerIntegrations.deployed).where(CustomerIntegrations.id == customer_id) |
| 1011 | result = await session.execute(stmt) |
| 1012 | is_deployed = result.scalar() |
| 1013 | |
| 1014 | logger.info(f"Integration {integration_name} for customer {customer_code} deployment status: {is_deployed}") |
| 1015 | |
| 1016 | subscription_ids = await get_subscription_ids( |
| 1017 | session, |
| 1018 | customer_id, |
| 1019 | integration_service_id, |
| 1020 | ) |
| 1021 | if not subscription_ids: |
| 1022 | raise HTTPException( |
| 1023 | status_code=404, |
| 1024 | detail="No subscriptions found for customer integration", |
| 1025 | ) |
| 1026 | |
| 1027 | # Only proceed with infrastructure cleanup if the integration is deployed |
| 1028 | if is_deployed: |
| 1029 | logger.info("Integration is deployed, proceeding with full cleanup including infrastructure components") |
| 1030 | |
| 1031 | # Fetch metadata from appropriate table based on integration type |
| 1032 | if is_network_integration: |
| 1033 | meta_data = await fetch_customer_network_connectors_meta(session, customer_code, integration_name) |
| 1034 | else: |
| 1035 | meta_data = await fetch_customer_integration_meta(session, customer_code, integration_name) |
| 1036 | |
| 1037 | if not meta_data: |
| 1038 | raise HTTPException(status_code=404, detail=f"Metadata not found for {integration_name} integration") |
| 1039 | |
| 1040 | # Delete stream and index using metadata |
| 1041 | stream_id = meta_data.graylog_stream_id |
| 1042 | logger.info(f"stream_id: {stream_id}") |
| 1043 | await delete_stream(stream_id=stream_id) |
| 1044 | |
| 1045 | index_id = meta_data.graylog_index_id |
| 1046 | logger.info(f"index_id: {index_id}") |
| 1047 | await delete_index_by_id(index_id=index_id) |
| 1048 | |
| 1049 | # Delete the folder in Grafana |
| 1050 | grafana_org_id = meta_data.grafana_org_id |
| 1051 | grafana_dashboard_folder_id = meta_data.grafana_dashboard_folder_id |
| 1052 | |
| 1053 | await delete_folder(grafana_org_id, int(grafana_dashboard_folder_id)) |
| 1054 | |
| 1055 | # Delete the grafana datasource |
| 1056 | if meta_data.grafana_datasource_uid is not None: |
| 1057 | logger.info(f"Deleting Grafana datasource with UID: {meta_data.grafana_datasource_uid}") |
| 1058 | await delete_grafana_datasource( |
| 1059 | organization_id=grafana_org_id, |
| 1060 | datasource_uid=meta_data.grafana_datasource_uid, |
| 1061 | ) |
| 1062 | else: |
| 1063 | logger.info("No Grafana datasource UID found, skipping deletion.") |
| 1064 | |
| 1065 | # Delete metadata from appropriate table |
| 1066 | if is_network_integration: |
| 1067 | await delete_customer_network_connectors_meta(session, customer_code, integration_name) |
| 1068 | else: |
| 1069 | await delete_customer_integration_meta(session, customer_code, integration_name) |
| 1070 | |
| 1071 | else: |
| 1072 | logger.info( |
| 1073 | "Integration is not deployed, skipping infrastructure cleanup (Graylog streams/indexes, Grafana folders/datasources, metadata)", |
| 1074 | ) |
| 1075 | |
| 1076 | # Always delete the integration settings (auth keys, configs, subscriptions, etc.) |
| 1077 | logger.info(f"Deleting integration settings for {integration_name}") |
| 1078 | await delete_metadata(session, subscription_ids) # This deletes auth keys |
| 1079 | await delete_subscriptions(session, subscription_ids) |
| 1080 | await delete_configs(session, integration_service_id) |
| 1081 | await delete_integration_service(session, integration_service_id) |
| 1082 | await delete_customer_integration_record(session, customer_id) |
| 1083 | |
| 1084 | await session.commit() |
| 1085 | |
| 1086 | return generate_decommission_response(customer_code, integration_name) |
| 1087 | |
| 1088 | |
| 1089 | @integration_settings_router.get( |
| 1090 | "/integration_customer/{integration_name}/{auth_key_name}/{auth_key_value}", |
| 1091 | description="Get customer code by integration details and auth key value", |
| 1092 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 1093 | ) |
| 1094 | async def get_customer_by_auth_key( |
| 1095 | integration_name: str, |
| 1096 | auth_key_name: str, |
| 1097 | auth_key_value: str, |
| 1098 | session: AsyncSession = Depends(get_db), |
| 1099 | ) -> CustomerByAuthKeyResponse: |
| 1100 | """ |
| 1101 | Retrieve a customer code based on integration name, auth key name, and auth key value. |
| 1102 | |
| 1103 | This is useful for identifying which customer an integration belongs to when you only |
| 1104 | have specific integration details, like a tenant ID. |
| 1105 | |
| 1106 | Args: |
| 1107 | integration_name: The name of the integration (e.g., "Office365") |
| 1108 | auth_key_name: The name of the auth key (e.g., "TENANT_ID") |
| 1109 | auth_key_value: The value of the auth key to look up |
| 1110 | |
| 1111 | Returns: |
| 1112 | Customer code and name associated with the provided integration details |
| 1113 | """ |
| 1114 | try: |
| 1115 | # Build query to find customer by auth key value |
| 1116 | query = ( |
| 1117 | select(CustomerIntegrations.customer_code, CustomerIntegrations.customer_name) |
| 1118 | .join(IntegrationSubscription, CustomerIntegrations.id == IntegrationSubscription.customer_id) |
| 1119 | .join(IntegrationService, IntegrationSubscription.integration_service_id == IntegrationService.id) |
| 1120 | .join(IntegrationAuthKeys, IntegrationSubscription.id == IntegrationAuthKeys.subscription_id) |
| 1121 | .where( |
| 1122 | IntegrationService.service_name == integration_name, |
| 1123 | IntegrationAuthKeys.auth_key_name == auth_key_name, |
| 1124 | IntegrationAuthKeys.auth_value == auth_key_value, |
| 1125 | ) |
| 1126 | ) |
| 1127 | |
| 1128 | # Execute query |
| 1129 | result = await session.execute(query) |
| 1130 | customer_info = result.first() |
| 1131 | |
| 1132 | if customer_info is None: |
| 1133 | raise HTTPException( |
| 1134 | status_code=404, |
| 1135 | detail=f"No customer found with {integration_name} integration having {auth_key_name}={auth_key_value}", |
| 1136 | ) |
| 1137 | |
| 1138 | customer_code, customer_name = customer_info |
| 1139 | |
| 1140 | return CustomerByAuthKeyResponse( |
| 1141 | customer_code=customer_code, |
| 1142 | customer_name=customer_name, |
| 1143 | integration_name=integration_name, |
| 1144 | auth_key_name=auth_key_name, |
| 1145 | ) |
| 1146 | |
| 1147 | except HTTPException: |
| 1148 | raise |
| 1149 | except Exception as e: |
| 1150 | logger.error(f"Error looking up customer by integration auth key: {str(e)}") |
| 1151 | raise HTTPException(status_code=500, detail=f"Failed to look up customer: {str(e)}") |
| 1152 | |
| 1153 | |
| 1154 | @integration_settings_router.get( |
| 1155 | "/meta_auto/{customer_code}/{integration_name}", |
| 1156 | description="Get integration or network connector metadata automatically based on integration name", |
| 1157 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 1158 | ) |
| 1159 | async def get_meta_auto( |
| 1160 | customer_code: str, |
| 1161 | integration_name: str, |
| 1162 | session: AsyncSession = Depends(get_db), |
| 1163 | ): |
| 1164 | """ |
| 1165 | Automatically retrieve metadata from the appropriate table based on integration name. |
| 1166 | |
| 1167 | This route checks if the integration is in the NETWORK_INTEGRATIONS list and |
| 1168 | fetches from the appropriate table accordingly. |
| 1169 | |
| 1170 | Args: |
| 1171 | customer_code (str): The customer code to filter by |
| 1172 | integration_name (str): The integration/connector name to filter by |
| 1173 | session (AsyncSession): Database session |
| 1174 | |
| 1175 | Returns: |
| 1176 | dict: The metadata record from the appropriate table |
| 1177 | """ |
| 1178 | logger.info(f"Fetching metadata for customer {customer_code} and integration {integration_name}") |
| 1179 | |
| 1180 | is_network_integration = integration_name in NETWORK_INTEGRATIONS |
| 1181 | |
| 1182 | try: |
| 1183 | if is_network_integration: |
| 1184 | # Fetch from network connectors table |
| 1185 | stmt = select(CustomerNetworkConnectorsMeta).where( |
| 1186 | CustomerNetworkConnectorsMeta.customer_code == customer_code, |
| 1187 | CustomerNetworkConnectorsMeta.network_connector_name == integration_name, |
| 1188 | ) |
| 1189 | result = await session.execute(stmt) |
| 1190 | meta_record = result.scalars().first() |
| 1191 | |
| 1192 | if not meta_record: |
| 1193 | raise HTTPException( |
| 1194 | status_code=404, |
| 1195 | detail=f"Network connector metadata not found for customer {customer_code} and connector {integration_name}", |
| 1196 | ) |
| 1197 | |
| 1198 | logger.info(f"Successfully retrieved network connector metadata for {customer_code}/{integration_name}") |
| 1199 | |
| 1200 | else: |
| 1201 | # Fetch from regular integrations table |
| 1202 | stmt = select(CustomerIntegrationsMeta).where( |
| 1203 | CustomerIntegrationsMeta.customer_code == customer_code, |
| 1204 | CustomerIntegrationsMeta.integration_name == integration_name, |
| 1205 | ) |
| 1206 | result = await session.execute(stmt) |
| 1207 | meta_record = result.scalars().first() |
| 1208 | |
| 1209 | if not meta_record: |
| 1210 | raise HTTPException( |
| 1211 | status_code=404, |
| 1212 | detail=f"Integration metadata not found for customer {customer_code} and integration {integration_name}", |
| 1213 | ) |
| 1214 | |
| 1215 | logger.info(f"Successfully retrieved integration metadata for {customer_code}/{integration_name}") |
| 1216 | |
| 1217 | # Convert SQLModel to dict for response |
| 1218 | return { |
| 1219 | "success": True, |
| 1220 | "message": f"Successfully retrieved metadata for {customer_code}/{integration_name}", |
| 1221 | "data": meta_record.model_dump() if hasattr(meta_record, "dict") else meta_record.__dict__, |
| 1222 | "table_type": "network_connector" if is_network_integration else "integration", |
| 1223 | } |
| 1224 | |
| 1225 | except HTTPException: |
| 1226 | raise |
| 1227 | except Exception as e: |
| 1228 | logger.error(f"Error retrieving metadata: {str(e)}") |
| 1229 | raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") |
| 1230 | |
| 1231 | |
| 1232 | @integration_settings_router.put( |
| 1233 | "/update_meta_auto", |
| 1234 | response_model=UpdateMetaResponse, |
| 1235 | description="Automatically update integration or network connector metadata based on integration name", |
| 1236 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 1237 | ) |
| 1238 | async def update_meta_auto( |
| 1239 | update_request: UpdateMetaAutoRequest, |
| 1240 | session: AsyncSession = Depends(get_db), |
| 1241 | ): |
| 1242 | """ |
| 1243 | Automatically determine which table to update based on integration name. |
| 1244 | |
| 1245 | This route checks if the integration is in the NETWORK_INTEGRATIONS list and |
| 1246 | updates the appropriate table accordingly. |
| 1247 | """ |
| 1248 | is_network_integration = update_request.integration_name in NETWORK_INTEGRATIONS |
| 1249 | |
| 1250 | try: |
| 1251 | if is_network_integration: |
| 1252 | # Check if record exists in network connectors table |
| 1253 | stmt = select(CustomerNetworkConnectorsMeta).where( |
| 1254 | CustomerNetworkConnectorsMeta.customer_code == update_request.customer_code, |
| 1255 | CustomerNetworkConnectorsMeta.network_connector_name == update_request.integration_name, |
| 1256 | ) |
| 1257 | result = await session.execute(stmt) |
| 1258 | existing_record = result.scalars().first() |
| 1259 | |
| 1260 | if not existing_record: |
| 1261 | raise HTTPException( |
| 1262 | status_code=404, |
| 1263 | detail=f"Network connector metadata not found for customer {update_request.customer_code} and connector {update_request.integration_name}", |
| 1264 | ) |
| 1265 | |
| 1266 | # Update network connector metadata |
| 1267 | update_data = {} |
| 1268 | if update_request.graylog_input_id is not None: |
| 1269 | update_data["graylog_input_id"] = update_request.graylog_input_id |
| 1270 | if update_request.graylog_index_id is not None: |
| 1271 | update_data["graylog_index_id"] = update_request.graylog_index_id |
| 1272 | if update_request.graylog_stream_id is not None: |
| 1273 | update_data["graylog_stream_id"] = update_request.graylog_stream_id |
| 1274 | if update_request.graylog_pipeline_id is not None: |
| 1275 | update_data["graylog_pipeline_id"] = update_request.graylog_pipeline_id |
| 1276 | if update_request.graylog_content_pack_input_id is not None: |
| 1277 | update_data["graylog_content_pack_input_id"] = update_request.graylog_content_pack_input_id |
| 1278 | if update_request.graylog_content_pack_stream_id is not None: |
| 1279 | update_data["graylog_content_pack_stream_id"] = update_request.graylog_content_pack_stream_id |
| 1280 | if update_request.grafana_org_id is not None: |
| 1281 | update_data["grafana_org_id"] = update_request.grafana_org_id |
| 1282 | if update_request.grafana_dashboard_folder_id is not None: |
| 1283 | update_data["grafana_dashboard_folder_id"] = update_request.grafana_dashboard_folder_id |
| 1284 | if update_request.grafana_datasource_uid is not None: |
| 1285 | update_data["grafana_datasource_uid"] = update_request.grafana_datasource_uid |
| 1286 | |
| 1287 | if update_data: |
| 1288 | update_stmt = ( |
| 1289 | update(CustomerNetworkConnectorsMeta) |
| 1290 | .where( |
| 1291 | CustomerNetworkConnectorsMeta.customer_code == update_request.customer_code, |
| 1292 | CustomerNetworkConnectorsMeta.network_connector_name == update_request.integration_name, |
| 1293 | ) |
| 1294 | .values(**update_data) |
| 1295 | ) |
| 1296 | await session.execute(update_stmt) |
| 1297 | |
| 1298 | else: |
| 1299 | # Check if record exists in integrations table |
| 1300 | stmt = select(CustomerIntegrationsMeta).where( |
| 1301 | CustomerIntegrationsMeta.customer_code == update_request.customer_code, |
| 1302 | CustomerIntegrationsMeta.integration_name == update_request.integration_name, |
| 1303 | ) |
| 1304 | result = await session.execute(stmt) |
| 1305 | existing_record = result.scalars().first() |
| 1306 | |
| 1307 | if not existing_record: |
| 1308 | raise HTTPException( |
| 1309 | status_code=404, |
| 1310 | detail=f"Integration metadata not found for customer {update_request.customer_code} and integration {update_request.integration_name}", |
| 1311 | ) |
| 1312 | |
| 1313 | # Update regular integration metadata |
| 1314 | update_data = {} |
| 1315 | if update_request.graylog_input_id is not None: |
| 1316 | update_data["graylog_input_id"] = update_request.graylog_input_id |
| 1317 | if update_request.graylog_index_id is not None: |
| 1318 | update_data["graylog_index_id"] = update_request.graylog_index_id |
| 1319 | if update_request.graylog_stream_id is not None: |
| 1320 | update_data["graylog_stream_id"] = update_request.graylog_stream_id |
| 1321 | if update_request.grafana_org_id is not None: |
| 1322 | update_data["grafana_org_id"] = update_request.grafana_org_id |
| 1323 | if update_request.grafana_dashboard_folder_id is not None: |
| 1324 | update_data["grafana_dashboard_folder_id"] = update_request.grafana_dashboard_folder_id |
| 1325 | if update_request.grafana_datasource_uid is not None: |
| 1326 | update_data["grafana_datasource_uid"] = update_request.grafana_datasource_uid |
| 1327 | |
| 1328 | if update_data: |
| 1329 | update_stmt = ( |
| 1330 | update(CustomerIntegrationsMeta) |
| 1331 | .where( |
| 1332 | CustomerIntegrationsMeta.customer_code == update_request.customer_code, |
| 1333 | CustomerIntegrationsMeta.integration_name == update_request.integration_name, |
| 1334 | ) |
| 1335 | .values(**update_data) |
| 1336 | ) |
| 1337 | await session.execute(update_stmt) |
| 1338 | |
| 1339 | if not update_data: |
| 1340 | return UpdateMetaResponse(success=False, message="No fields provided for update") |
| 1341 | |
| 1342 | await session.commit() |
| 1343 | |
| 1344 | table_type = "network connector" if is_network_integration else "integration" |
| 1345 | logger.info( |
| 1346 | f"Updated {table_type} metadata for customer {update_request.customer_code}, {table_type} {update_request.integration_name}", |
| 1347 | ) |
| 1348 | |
| 1349 | return UpdateMetaResponse( |
| 1350 | success=True, |
| 1351 | message=f"Successfully updated {table_type} metadata for {update_request.customer_code}/{update_request.integration_name}", |
| 1352 | ) |
| 1353 | |
| 1354 | except HTTPException: |
| 1355 | raise |
| 1356 | except Exception as e: |
| 1357 | await session.rollback() |
| 1358 | logger.error(f"Error updating metadata: {str(e)}") |
| 1359 | raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}") |