[kaspersky-enrichment] Add URL enrichment

internal-enrichment/kaspersky-enrichment/docker-compose.yml

@@ -9,7 +9,7 @@ services:
       #  Common parameters for connectors of type INTERNAL_ENRICHMENT
       - CONNECTOR_ID=ChangeMe
       # - CONNECTOR_NAME=Kaspersky Enrichment
-      # - CONNECTOR_SCOPE=StixFile,IPv4-Addr,Domain-Name,Hostname  # Support for additional observable types (URLs) will be added in future releases.
+      # - CONNECTOR_SCOPE=StixFile,IPv4-Addr,Domain-Name,Hostname,Url
       # - CONNECTOR_LOG_LEVEL=error
       # - CONNECTOR_AUTO=true
 
@@ -21,5 +21,6 @@ services:
       # - KASPERSKY_FILE_SECTIONS=LicenseInfo,Zone,FileGeneralInfo
       # - KASPERSKY_IPV4_SECTIONS=LicenseInfo,Zone,IpGeneralInfo
       # - KASPERSKY_DOMAIN_SECTIONS=LicenseInfo,Zone,DomainGeneralInfo
+      # - KASPERSKY_URL_SECTIONS=LicenseInfo,Zone,UrlGeneralInfo
 
     restart: always

internal-enrichment/kaspersky-enrichment/src/config.yml.sample

@@ -6,7 +6,7 @@ connector:
 #  type: 'INTERNAL_ENRICHMENT'
 #  id: 'ChangeMe'
 #  name: 'Kaspersky Enrichment'
-#  scope: 'StixFile,IPv4-Addr,Domain-Name,Hostname'  # Support for additional observable types (URLs) will be added in future releases.
+#  scope: 'StixFile,IPv4-Addr,Domain-Name,Hostname,Url'
 #  log_level: 'error'
 #  auto: true # Enable/disable auto-enrichment of observables
 
@@ -18,3 +18,4 @@ kaspersky:
 #  file_sections: 'LicenseInfo,Zone,FileGeneralInfo'
 #  ipv4_sections: 'LicenseInfo,Zone,IpGeneralInfo'
 #  domain_sections: 'LicenseInfo,Zone,DomainGeneralInfo'
+#  url_sections: 'LicenseInfo,Zone,UrlGeneralInfo'

internal-enrichment/kaspersky-enrichment/src/connector/connector.py

@@ -3,6 +3,7 @@
 from connector.use_cases.enrich_domain import DomainEnricher
 from connector.use_cases.enrich_file import FileEnricher
 from connector.use_cases.enrich_ipv4 import Ipv4Enricher
+from connector.use_cases.enrich_url import UrlEnricher
 from connector.utils import entity_in_scope
 from kaspersky_client import KasperskyClient
 from pycti import OpenCTIConnectorHelper
@@ -54,6 +55,7 @@ def __init__(
         file_sections = self.config.kaspersky.file_sections
         ipv4_sections = self.config.kaspersky.ipv4_sections
         domain_sections = self.config.kaspersky.domain_sections
+        url_sections = self.config.kaspersky.url_sections
         zone_octi_score_mapping = self.config.kaspersky.zone_octi_score_mapping
         api_key = self.config.kaspersky.api_key.get_secret_value()
 
@@ -70,26 +72,33 @@ def __init__(
         converter_to_stix = ConverterToStix(self.helper)
 
         self.file_enricher = FileEnricher(
-            helper=self.helper,
+            connector_logger=self.helper.connector_logger,
             client=client,
             sections=file_sections,
             zone_octi_score_mapping=zone_octi_score_mapping,
             converter_to_stix=converter_to_stix,
         )
         self.ipv4_enricher = Ipv4Enricher(
-            helper=self.helper,
+            connector_logger=self.helper.connector_logger,
             client=client,
             sections=ipv4_sections,
             zone_octi_score_mapping=zone_octi_score_mapping,
             converter_to_stix=converter_to_stix,
         )
         self.domain_enricher = DomainEnricher(
-            helper=self.helper,
+            connector_logger=self.helper.connector_logger,
             client=client,
             sections=domain_sections,
             zone_octi_score_mapping=zone_octi_score_mapping,
             converter_to_stix=converter_to_stix,
         )
+        self.url_enricher = UrlEnricher(
+            connector_logger=self.helper.connector_logger,
+            client=client,
+            sections=url_sections,
+            zone_octi_score_mapping=zone_octi_score_mapping,
+            converter_to_stix=converter_to_stix,
+        )
 
         # Define variables
         self.stix_objects = []
@@ -159,10 +168,10 @@ def process_message(self, data: dict) -> str:
                         octi_objects = self.domain_enricher.process_domain_enrichment(
                             observable
                         )
-                    # case "Url":
-                    #     octi_objects = self.url_enricher.process_url_enrichment(
-                    #         observable
-                    #     )
+                    case "Url":
+                        octi_objects = self.url_enricher.process_url_enrichment(
+                            observable
+                        )
                     case _:
                         raise ValueError(
                             "Entity type is not supported",

internal-enrichment/kaspersky-enrichment/src/connector/constants.py

@@ -36,7 +36,17 @@
             "Industries",
         ],
     },
-    "url_sections": {},
+    "url_sections": {
+        "mandatories_sections": ["LicenseInfo", "Zone", "UrlGeneralInfo"],
+        "supported_sections": [
+            "LicenseInfo",
+            "Zone",
+            "UrlGeneralInfo",
+            "FilesDownloaded",
+            "FilesAccessed",
+            "Industries",
+        ],
+    },
 }
 
 DATETIME_FORMAT = "%Y-%m-%dT%H:%MZ"

internal-enrichment/kaspersky-enrichment/src/connector/converter_to_stix.py

@@ -9,6 +9,7 @@
     DomainName,
     File,
     IPV4Address,
+    IPV6Address,
     Note,
     OrganizationAuthor,
     Reference,
@@ -65,7 +66,7 @@ def create_country(self, country_name: str) -> Country:
         """
         return Country(name=country_name, author=self.author, markings=[self.tlp_clear])
 
-    def create_domain(self, name: str, score: int) -> DomainName:
+    def create_domain(self, name: str, score: int = None) -> DomainName:
         """
         Create a Domain object
         """
@@ -88,6 +89,12 @@ def create_ipv4(self, ip: str) -> IPV4Address:
         """
         return IPV4Address(value=ip, author=self.author, markings=[self.tlp_amber])
 
+    def create_ipv6(self, ip: str) -> IPV4Address:
+        """
+        Create an IPv6 object
+        """
+        return IPV6Address(value=ip, author=self.author, markings=[self.tlp_amber])
+
     def create_note(self, observable: Reference, content: str) -> Note:
         """
         Create a note associated to the file observable

internal-enrichment/kaspersky-enrichment/src/connector/settings.py

@@ -60,7 +60,7 @@ class InternalEnrichmentConnectorConfig(BaseInternalEnrichmentConnectorConfig):
         description="Name of the connector.",
     )
     scope: ListFromString = Field(
-        default=["StixFile", "IPv4-Addr", "Domain-Name", "Hostname"],
+        default=["StixFile", "IPv4-Addr", "Domain-Name", "Hostname", "Url"],
         description="The scope or type of data the connector is importing, either a MIME type or Stix Object (for information only).",
     )
     auto: bool = Field(
@@ -120,11 +120,18 @@ class KasperskyConfig(BaseConfigModel):
         "LicenseInfo, Zone and DomainGeneralInfo are always set, can't be disabled. "
         "Only DomainDnsResolutions, FilesDownloaded, FilesAccessed and Industries are currently supported",
     )
+    url_sections: str = Field(
+        default="LicenseInfo,Zone,UrlGeneralInfo",
+        description="Sections wanted to investigate for the requested URL. "
+        "LicenseInfo, Zone and UrlGeneralInfo are always set, can't be disabled. "
+        "Only FilesDownloaded, FilesAccessed and Industries are currently supported",
+    )
 
     @field_validator(
         "file_sections",
         "ipv4_sections",
         "domain_sections",
+        "url_sections",
         mode="before",
     )
     @classmethod

internal-enrichment/kaspersky-enrichment/src/connector/use_cases/common.py

@@ -1,23 +1,26 @@
+import logging
+
 from connector.converter_to_stix import ConverterToStix
-from connector.utils import is_quota_exceeded
-from pycti import STIX_EXT_OCTI_SCO, OpenCTIConnectorHelper, OpenCTIStix2
+from connector.utils import get_first_and_last_seen_datetime, is_quota_exceeded
+from connectors_sdk.models import Reference
+from pycti import STIX_EXT_OCTI_SCO, OpenCTIStix2
 
 
 class BaseUseCases:
     def __init__(
         self,
-        helper: OpenCTIConnectorHelper,
+        connector_logger: logging.Logger,
         converter_to_stix: ConverterToStix,
     ):
-        self.helper = helper
+        self.connector_logger = connector_logger
         self.converter_to_stix = converter_to_stix
 
     def check_quota(self, license_info: dict) -> None:
         """
         Send a log warning if quota is exceeded
         """
         if is_quota_exceeded(license_info):
-            self.helper.connector_logger.warning(
+            self.connector_logger.warning(
                 "[CONNECTOR] The daily quota has been exceeded",
                 {
                     "day_requests": license_info["DayRequests"],
@@ -42,11 +45,62 @@ def generate_author_and_tlp_markings(self):
 
         return common_objects
 
-    def update_observable_score(self, zone: str, observable: dict):
+    def update_observable_score(self, zone: str, observable: dict) -> dict:
         """
         Update score in observable
         """
         score = self.zone_octi_score_mapping[zone.lower()]
         return OpenCTIStix2.put_attribute_in_extension(
             observable, STIX_EXT_OCTI_SCO, "score", score
         )
+
+    def manage_industries(self, observable_to_ref: Reference, industries: list) -> list:
+        """
+        Create sector and relation for each item in industries
+        """
+        self.connector_logger.info(
+            "[CONNECTOR] Process enrichment from Industries data..."
+        )
+
+        industry_objects = []
+        for industry in industries:
+            industry_object = self.converter_to_stix.create_sector(industry)
+
+            if industry_object:
+                industry_objects.append(industry_object.to_stix2_object())
+                industry_relation = self.converter_to_stix.create_relationship(
+                    relationship_type="related-to",
+                    source_obj=observable_to_ref,
+                    target_obj=industry_object,
+                )
+                industry_objects.append(industry_relation.to_stix2_object())
+        return industry_objects
+
+    def manage_files(self, files: list, observable_to_ref: Reference) -> list:
+        """
+        Create file and relation for each item
+        """
+        file_objects = []
+        for file in files:
+            obs_file = self.converter_to_stix.create_file(
+                hashes={"MD5": file["Md5"]},
+                score=self.zone_octi_score_mapping[file["Zone"].lower()],
+            )
+
+            if obs_file:
+                file_objects.append(obs_file.to_stix2_object())
+                file_first_seen_datetime, file_last_seen_datetime = (
+                    get_first_and_last_seen_datetime(
+                        file["FirstSeen"],
+                        file["LastSeen"],
+                    )
+                )
+                file_relation = self.converter_to_stix.create_relationship(
+                    source_obj=observable_to_ref,
+                    relationship_type="related-to",
+                    target_obj=obs_file,
+                    start_time=file_first_seen_datetime,
+                    stop_time=file_last_seen_datetime,
+                )
+                file_objects.append(file_relation.to_stix2_object())
+        return file_objects