OCI MCP Server

""" Tools for managing OCI Security resources. """ import logging from typing import Dict, List, Any, Optional import oci logger = logging.getLogger(__name__) def list_security_lists(network_client: oci.core.VirtualNetworkClient, compartment_id: str, vcn_id: Optional[str] = None) -> List[Dict[str, Any]]: """ List all security lists in a compartment, optionally filtered by VCN. Args: network_client: OCI VirtualNetwork client compartment_id: OCID of the compartment vcn_id: Optional OCID of the VCN to filter by Returns: List of security lists with their details """ try: security_lists_response = oci.pagination.list_call_get_all_results( network_client.list_security_lists, compartment_id, vcn_id=vcn_id ) security_lists = [] for security_list in security_lists_response.data: security_lists.append({ "id": security_list.id, "display_name": security_list.display_name, "compartment_id": security_list.compartment_id, "vcn_id": security_list.vcn_id, "lifecycle_state": security_list.lifecycle_state, "time_created": str(security_list.time_created), "ingress_security_rules_count": len(security_list.ingress_security_rules) if security_list.ingress_security_rules else 0, "egress_security_rules_count": len(security_list.egress_security_rules) if security_list.egress_security_rules else 0, }) logger.info(f"Found {len(security_lists)} security lists in compartment {compartment_id}") return security_lists except Exception as e: logger.exception(f"Error listing security lists: {e}") raise def get_security_list(network_client: oci.core.VirtualNetworkClient, security_list_id: str) -> Dict[str, Any]: """ Get details of a specific security list. Args: network_client: OCI VirtualNetwork client security_list_id: OCID of the security list Returns: Details of the security list """ try: security_list = network_client.get_security_list(security_list_id).data # Format ingress rules ingress_rules = [] if security_list.ingress_security_rules: for rule in security_list.ingress_security_rules: ingress_rules.append({ "protocol": rule.protocol, "source": rule.source, "source_type": rule.source_type, "is_stateless": rule.is_stateless, "description": rule.description, "tcp_options": { "destination_port_range": { "min": rule.tcp_options.destination_port_range.min if rule.tcp_options and rule.tcp_options.destination_port_range else None, "max": rule.tcp_options.destination_port_range.max if rule.tcp_options and rule.tcp_options.destination_port_range else None, } if rule.tcp_options and rule.tcp_options.destination_port_range else None, "source_port_range": { "min": rule.tcp_options.source_port_range.min if rule.tcp_options and rule.tcp_options.source_port_range else None, "max": rule.tcp_options.source_port_range.max if rule.tcp_options and rule.tcp_options.source_port_range else None, } if rule.tcp_options and rule.tcp_options.source_port_range else None, } if rule.tcp_options else None, "udp_options": { "destination_port_range": { "min": rule.udp_options.destination_port_range.min if rule.udp_options and rule.udp_options.destination_port_range else None, "max": rule.udp_options.destination_port_range.max if rule.udp_options and rule.udp_options.destination_port_range else None, } if rule.udp_options and rule.udp_options.destination_port_range else None, "source_port_range": { "min": rule.udp_options.source_port_range.min if rule.udp_options and rule.udp_options.source_port_range else None, "max": rule.udp_options.source_port_range.max if rule.udp_options and rule.udp_options.source_port_range else None, } if rule.udp_options and rule.udp_options.source_port_range else None, } if rule.udp_options else None, "icmp_options": { "type": rule.icmp_options.type if rule.icmp_options else None, "code": rule.icmp_options.code if rule.icmp_options else None, } if rule.icmp_options else None, }) # Format egress rules egress_rules = [] if security_list.egress_security_rules: for rule in security_list.egress_security_rules: egress_rules.append({ "protocol": rule.protocol, "destination": rule.destination, "destination_type": rule.destination_type, "is_stateless": rule.is_stateless, "description": rule.description, "tcp_options": { "destination_port_range": { "min": rule.tcp_options.destination_port_range.min if rule.tcp_options and rule.tcp_options.destination_port_range else None, "max": rule.tcp_options.destination_port_range.max if rule.tcp_options and rule.tcp_options.destination_port_range else None, } if rule.tcp_options and rule.tcp_options.destination_port_range else None, "source_port_range": { "min": rule.tcp_options.source_port_range.min if rule.tcp_options and rule.tcp_options.source_port_range else None, "max": rule.tcp_options.source_port_range.max if rule.tcp_options and rule.tcp_options.source_port_range else None, } if rule.tcp_options and rule.tcp_options.source_port_range else None, } if rule.tcp_options else None, "udp_options": { "destination_port_range": { "min": rule.udp_options.destination_port_range.min if rule.udp_options and rule.udp_options.destination_port_range else None, "max": rule.udp_options.destination_port_range.max if rule.udp_options and rule.udp_options.destination_port_range else None, } if rule.udp_options and rule.udp_options.destination_port_range else None, "source_port_range": { "min": rule.udp_options.source_port_range.min if rule.udp_options and rule.udp_options.source_port_range else None, "max": rule.udp_options.source_port_range.max if rule.udp_options and rule.udp_options.source_port_range else None, } if rule.udp_options and rule.udp_options.source_port_range else None, } if rule.udp_options else None, "icmp_options": { "type": rule.icmp_options.type if rule.icmp_options else None, "code": rule.icmp_options.code if rule.icmp_options else None, } if rule.icmp_options else None, }) security_list_details = { "id": security_list.id, "display_name": security_list.display_name, "compartment_id": security_list.compartment_id, "vcn_id": security_list.vcn_id, "lifecycle_state": security_list.lifecycle_state, "time_created": str(security_list.time_created), "ingress_security_rules": ingress_rules, "egress_security_rules": egress_rules, } logger.info(f"Retrieved details for security list {security_list_id}") return security_list_details except Exception as e: logger.exception(f"Error getting security list details: {e}") raise def list_network_security_groups(network_client: oci.core.VirtualNetworkClient, compartment_id: str, vcn_id: Optional[str] = None) -> List[Dict[str, Any]]: """ List all network security groups in a compartment, optionally filtered by VCN. Args: network_client: OCI VirtualNetwork client compartment_id: OCID of the compartment vcn_id: Optional OCID of the VCN to filter by Returns: List of network security groups with their details """ try: nsgs_response = oci.pagination.list_call_get_all_results( network_client.list_network_security_groups, compartment_id, vcn_id=vcn_id ) nsgs = [] for nsg in nsgs_response.data: nsgs.append({ "id": nsg.id, "display_name": nsg.display_name, "compartment_id": nsg.compartment_id, "vcn_id": nsg.vcn_id, "lifecycle_state": nsg.lifecycle_state, "time_created": str(nsg.time_created), }) logger.info(f"Found {len(nsgs)} network security groups in compartment {compartment_id}") return nsgs except Exception as e: logger.exception(f"Error listing network security groups: {e}") raise def get_network_security_group(network_client: oci.core.VirtualNetworkClient, nsg_id: str) -> Dict[str, Any]: """ Get details of a specific network security group. Args: network_client: OCI VirtualNetwork client nsg_id: OCID of the network security group Returns: Details of the network security group """ try: nsg = network_client.get_network_security_group(nsg_id).data nsg_details = { "id": nsg.id, "display_name": nsg.display_name, "compartment_id": nsg.compartment_id, "vcn_id": nsg.vcn_id, "lifecycle_state": nsg.lifecycle_state, "time_created": str(nsg.time_created), } logger.info(f"Retrieved details for network security group {nsg_id}") return nsg_details except Exception as e: logger.exception(f"Error getting network security group details: {e}") raise def list_vaults(kms_vault_client: oci.key_management.KmsVaultClient, compartment_id: str) -> List[Dict[str, Any]]: """ List all vaults in a compartment. Args: kms_vault_client: OCI KmsVault client compartment_id: OCID of the compartment Returns: List of vaults with their details """ try: vaults_response = oci.pagination.list_call_get_all_results( kms_vault_client.list_vaults, compartment_id ) vaults = [] for vault in vaults_response.data: vaults.append({ "id": vault.id, "display_name": vault.display_name, "compartment_id": vault.compartment_id, "lifecycle_state": vault.lifecycle_state, "time_created": str(vault.time_created), "vault_type": vault.vault_type, "crypto_endpoint": vault.crypto_endpoint, "management_endpoint": vault.management_endpoint, "is_primary": vault.is_primary, }) logger.info(f"Found {len(vaults)} vaults in compartment {compartment_id}") return vaults except Exception as e: logger.exception(f"Error listing vaults: {e}") raise def get_vault(kms_vault_client: oci.key_management.KmsVaultClient, vault_id: str) -> Dict[str, Any]: """ Get details of a specific vault. Args: kms_vault_client: OCI KmsVault client vault_id: OCID of the vault Returns: Details of the vault """ try: vault = kms_vault_client.get_vault(vault_id).data vault_details = { "id": vault.id, "display_name": vault.display_name, "compartment_id": vault.compartment_id, "lifecycle_state": vault.lifecycle_state, "time_created": str(vault.time_created), "vault_type": vault.vault_type, "crypto_endpoint": vault.crypto_endpoint, "management_endpoint": vault.management_endpoint, "is_primary": vault.is_primary, "replica_details": { "replication_id": vault.replica_details.replication_id if vault.replica_details else None, } if vault.replica_details else None, } logger.info(f"Retrieved details for vault {vault_id}") return vault_details except Exception as e: logger.exception(f"Error getting vault details: {e}") raise def list_keys(config: dict, management_endpoint: str, compartment_id: str) -> List[Dict[str, Any]]: """ List all keys in a vault's compartment. Args: config: OCI config dict management_endpoint: Management endpoint from the vault compartment_id: OCID of the compartment Returns: List of keys with their details """ try: # Create KMS Management client with the vault's management endpoint kms_management_client = oci.key_management.KmsManagementClient(config, service_endpoint=management_endpoint) keys_response = oci.pagination.list_call_get_all_results( kms_management_client.list_keys, compartment_id ) keys = [] for key in keys_response.data: keys.append({ "id": key.id, "display_name": key.display_name, "compartment_id": key.compartment_id, "lifecycle_state": key.lifecycle_state, "time_created": str(key.time_created), "vault_id": key.vault_id, "protection_mode": key.protection_mode, "algorithm": key.algorithm, }) logger.info(f"Found {len(keys)} keys in compartment {compartment_id}") return keys except Exception as e: logger.exception(f"Error listing keys: {e}") raise def get_key(config: dict, management_endpoint: str, key_id: str) -> Dict[str, Any]: """ Get details of a specific key. Args: config: OCI config dict management_endpoint: Management endpoint from the vault key_id: OCID of the key Returns: Details of the key """ try: # Create KMS Management client with the vault's management endpoint kms_management_client = oci.key_management.KmsManagementClient(config, service_endpoint=management_endpoint) key = kms_management_client.get_key(key_id).data key_details = { "id": key.id, "display_name": key.display_name, "compartment_id": key.compartment_id, "lifecycle_state": key.lifecycle_state, "time_created": str(key.time_created), "vault_id": key.vault_id, "protection_mode": key.protection_mode, "algorithm": key.algorithm, "current_key_version": key.current_key_version, "key_shape": { "algorithm": key.key_shape.algorithm if key.key_shape else None, "length": key.key_shape.length if key.key_shape else None, "curve_id": key.key_shape.curve_id if key.key_shape else None, } if key.key_shape else None, "is_primary": key.is_primary, "replica_details": { "replication_id": key.replica_details.replication_id if key.replica_details else None, } if key.replica_details else None, } logger.info(f"Retrieved details for key {key_id}") return key_details except Exception as e: logger.exception(f"Error getting key details: {e}") raise