[socradar] Add SOCRadar external import connector

[Radargoger]

Proposed changes

• Add SOCRadar external import connector with support for:
  • IP addresses (IPv4/IPv6)
  • Domains
  • URLs
  • File hashes
• Implement STIX2 mapping with metadata:
  • Source attribution
  • First/last seen dates
  • Confidence levels
  • TLP marking
• Add rate limiting and pagination handling
• Include error handling and logging

Related issues

• None

Checklist

• I consider the submitted work as finished
• I tested the code for its functionality using different use cases
• I added/update the relevant documentation (either on github or on notion)
• Where necessary I refactored code to improve the overall quality

Further comments

This connector integrates SOCRadar's threat intelligence feeds into OpenCTI. Testing has been completed with:

• OpenCTI 6.4.2
• Python 3.11
• Various indicator types and formats
• STIX2 bundle creation and import verification

The implementation follows OpenCTI connector best practices and provides a robust solution for importing SOCRadar threat intelligence into the platform.

[flavienSindou]

Thank you for your contribution!

Could you please rework the code and documentation a bit? Once that’s done, I’ll be happy to approve your pull request.

[flavienSindou]

This would create stochastic id and will prevent de-duplication performed by the platform.

You should rather use pycti.Indicator.generate_id method.

[flavienSindou]

This would create stochastic id and will prevent de-duplication performed by the platform.

You should rather use pycti.StixCoreRelationship.generate_id method.

[flavienSindou]

stix2.Observable IDs are deterministic if not explicitly modified.
This allows the platform to de-duplicate these entities.

[flavienSindou]

It would be easier to use Opencti Scheduler

Here is an example of usage :

connectors/templates/external-import/src/external_import_connector/connector.py

Line 189 in 5a346b3

self.helper.schedule_iso(

[flavienSindou]

Could you update this configuration description to match your code please ?

[flavienSindou]

Could you please rework this example and remove unused variables please ?

[Radargoger]

Hello Team, We made an improvement based on your feedback. Let us know if anything needs to be changed. Best regards. Burak GOGER Senior Product Manager <https://socradar.io/> ***@***.*** +90 (552) 398 30 04 <+90+505+500+00+00> socradar.io *HQ Office:* 254 Chapman Rd, Ste 208 Newark, Delaware 19702 USA *EMEA Office:* İçerenköy Mah. Umut Sok. Quick Tower No:10-12 Ataşehir/İstanbul/Türkiye [image: linkedin] <https://www.linkedin.com/company/socradar> [image: twitter] <https://twitter.com/socradar> [image: linkedin] <https://www.youtube.com/channel/UClUyizkV30njCwQUcPhJ_qg> <https://www.gartner.com/reviews/market/security-threat-intelligence-services/vendor/socradar/product/socradar-digital-risk-protection-platform>

…

On Tue, Dec 24, 2024 at 5:31 PM flavienSindou ***@***.***> wrote: ***@***.**** commented on this pull request. Thank you for your contribution! Could you please rework the code and documentation a bit? Once that’s done, I’ll be happy to approve your pull request. ------------------------------ In external-import/socradar/src/lib/radar.py <#3072 (comment)> : > + + pattern = self._create_stix_pattern(value, feed_type) + if not pattern: + self.helper.log_error( + f"Could not create pattern for: {value} ({feed_type})" + ) + return + + # Create kill chain phase + kill_chain_phase = KillChainPhase( + kill_chain_name="lockheed-martin-cyber-kill-chain", + phase_name="reconnaissance", + ) + + indicator = Indicator( + id=f"indicator--{str(uuid.uuid4())}", This would create stochastic id and will prevent de-duplication performed by the platform. You should rather use pycti.Indicator.generate_id method. ------------------------------ In external-import/socradar/src/lib/radar.py <#3072 (comment)> : > + pattern=pattern, + valid_from=first_seen, + valid_until=last_seen, + labels=[feed_type, "malicious-activity"], + confidence=75, + indicator_types=self._get_indicator_type(feed_type), + kill_chain_phases=[kill_chain_phase], + created=first_seen, + modified=first_seen, + created_by_ref=maintainer_identity.id, # Use maintainer's identity + object_marking_refs=[TLP_WHITE], + ) + + # Create relationship between indicator and maintainer identity + relationship = Relationship( + id=f"relationship--{str(uuid.uuid4())}", This would create stochastic id and will prevent de-duplication performed by the platform. You should rather use pycti.StixCoreRelationship.generate_id method. ------------------------------ In external-import/socradar/src/lib/radar.py <#3072 (comment)> : > + # Send to OpenCTI + self.helper.send_stix2_bundle(bundle.serialize(), work_id=work_id) + + self.helper.log_info( + f"Created {feed_type} indicator for: {value} from {maintainer}" + ) + + except Exception as e: + self.helper.log_error(f"Error processing item {str(item)}: {str(e)}") + + def _create_observable(self, value, feed_type): + """Create appropriate observable based on value type with proper STIX ID""" + try: + if feed_type == "url" or self._matches_pattern(value, "url"): + return URL( + id=f"url--{str(uuid.uuid4())}", stix2.Observable IDs are deterministic if not explicitly modified. This allows the platform to de-duplicate this entities. ------------------------------ In external-import/socradar/src/lib/radar.py <#3072 (comment)> : > + if re.match(regex, value): + if pattern_type == "url": + return f"[url:value = '{value}']" + elif pattern_type == "domain": + return f"[domain-name:value = '{value}']" + elif pattern_type == "ipv4": + return f"[ipv4-addr:value = '{value}']" + elif pattern_type == "ipv6": + return f"[ipv6-addr:value = '{value}']" + + self.helper.log_error( + f"Could not determine pattern for value: {value} (type: {feed_type})" + ) + return None + + def run(self): It would be easier to use Opencti Scheduler Here is an example of usage : https://github.com/OpenCTI-Platform/connectors/blob/5a346b3902308712efa2ef4c9f7069fa3f6389be/templates/external-import/src/external_import_connector/connector.py#L189 ------------------------------ In external-import/socradar/README.md <#3072 (comment)> : > +| Parameter | Docker envvar | Mandatory | Description | +| --- | --- | --- | --- | +| `opencti_url` | `OPENCTI_URL` | Yes | The URL of the OpenCTI platform | +| `opencti_token` | `OPENCTI_TOKEN` | Yes | The default admin token configured in the OpenCTI platform | +| `connector_id` | `CONNECTOR_ID` | Yes | A valid arbitrary UUIDv4 for this connector | +| `connector_type` | `CONNECTOR_TYPE` | Yes | Must be 'EXTERNAL_IMPORT' | +| `connector_name` | `CONNECTOR_NAME` | Yes | Name of the connector | +| `connector_scope` | `CONNECTOR_SCOPE` | Yes | Scope of the connector (socradar) | +| `connector_confidence_level` | `CONNECTOR_CONFIDENCE_LEVEL` | Yes | Default confidence level for created data | +| `connector_log_level` | `CONNECTOR_LOG_LEVEL` | Yes | Logging level (debug, info, warn, error) | +| `radar_base_feed_url` | `RADAR_BASE_FEED_URL` | Yes | SocRadar API base URL | +| `radar_format_type` | `RADAR_FORMAT_TYPE` | Yes | Response format (.json) | +| `radar_socradar_key` | `RADAR_SOCRADAR_KEY` | Yes | Your SocRadar API key | +| `radar_interval` | `RADAR_INTERVAL` | Yes | Interval between runs in seconds | Could you update this configuration description to match your code please ? ------------------------------ In external-import/socradar/src/config.yml.sample <#3072 (comment)> : > +radar: + base_feed_url: "https://platform.socradar.com/api/threat/intelligence/feed_list/" + format_type: ".json?key=" + socradar_key: "SOCRADAR_KEY" + run_interval: 600 + collections_uuid: + collection_1: + id: ["COLLECTION_UUID"] + name: ["COLLECTION_NAME"] + default_marking: 'TLP:WHITE' + create_observables: true + create_indicators: true Could you please rework this example and remove unused variables please ? — Reply to this email directly, view it on GitHub <#3072 (review)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/BL7TPYX2IRZV2VSAV274U432HFV3PAVCNFSM6AAAAABS73TNPSVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDKMRSGAYDCMBRGI> . You are receiving this because you authored the thread.Message ID: ***@***.***>

--

-------------------------------------- Disclaimer: The content of this email is confidential and intended for the recipient specified in message only. It is strictly forbidden to share any part of this message with any third party, without a written consent of the sender. If you received this message by mistake, please reply to this message and follow with its deletion, so that we can ensure such a mistake does not occur in the future.

[flavienSindou]

Thank you for the improvements you’ve made.

I have a few additional remarks to ensure that your connector performs optimally.

Additionally, I still encourage you to use the OpenCTI dedicated Scheduler instead of implementing your own.

Let me know if you need any support with this—I’ll be happy to assist. I look forward to reviewing your next changes!

https://filigran.io/auto-backpressue-control-octi-connectors/

[flavienSindou]

radar.format_type seems missing from the documentation

[flavienSindou]

The expected parameter seems to be run_interval

[flavienSindou]

This would create stochastic id and will prevent de-duplication performed by the platform.

You should rather use pycti.Identity.generate_id method.

[flavienSindou]

This does not seems possible to use self.collections as expected in

    def _process_feed(self, work_id: str) -> None:
        """Step 5.0: Process feed collection"""
        try:
            self.helper.log_info("Starting feed collection...")

            # Step 5.1: Process each collection
            for collection_name, collection_data in self.collections.items():
                try:
                    collection_id = collection_data["id"][0]

A Mapping of Sequence is used in this code whereas a string would be loaded by

        self.collections = get_config_variable(
            "RADAR_COLLECTIONS_UUID", ["radar", "collections_uuid"], config
        )

when using RADAR_COLLECTIONS_UUID env var.

To resolve this, you could try casting the variable as follows:

if isinstance(my_raw_var, str):
    try: 
        my_var = json.loads(my_raw_var)
    except json.JSONDecodeError: 
        raise ValueError("Expecting a valid json string Mapping for my_raw_var")
        

[Radargoger]

Thank you, @flavienSindou, We’ve addressed the remarks and made the necessary improvements to ensure optimal performance. We’re excited to see this connector in action within the system and look forward to your thoughts on the final implementation. Wishing you a happy (belated) Christmas and all the best for the New Year! 🎄✨ Best regards Burak GOGER Senior Product Manager <https://socradar.io/> ***@***.*** +90 (552) 398 30 04 <+90+505+500+00+00> socradar.io *HQ Office:* 254 Chapman Rd, Ste 208 Newark, Delaware 19702 USA *EMEA Office:* İçerenköy Mah. Umut Sok. Quick Tower No:10-12 Ataşehir/İstanbul/Türkiye [image: linkedin] <https://www.linkedin.com/company/socradar> [image: twitter] <https://twitter.com/socradar> [image: linkedin] <https://www.youtube.com/channel/UClUyizkV30njCwQUcPhJ_qg> <https://www.gartner.com/reviews/market/security-threat-intelligence-services/vendor/socradar/product/socradar-digital-risk-protection-platform>

…

On Thu, Dec 26, 2024 at 4:08 PM flavienSindou ***@***.***> wrote: ***@***.**** approved this pull request. Thank you for the improvements you’ve made. I have a few additional remarks to ensure that your connector performs optimally. Additionally, I still encourage you to use the OpenCTI dedicated Scheduler instead of implementing your own. Let me know if you need any support with this—I’ll be happy to assist. I look forward to reviewing your next changes! https://filigran.io/auto-backpressue-control-octi-connectors/ ------------------------------ In socradar/README.md <#3072 (comment)> : > @@ -0,0 +1,40 @@ +# OpenCTI SOCRadar Connector + +The connector imports threat intelligence feeds from SOCRadar into OpenCTI. It processes various types of indicators including IP addresses, domains, URLs, and file hashes. + +SOCRadar provides comprehensive threat intelligence feeds that can be used to detect and prevent various types of cyber threats. The connector fetches these feeds and converts them into standardized STIX2 format for use in OpenCTI, enabling organizations to enhance their threat detection and response capabilities. + +## Installation + +### Requirements + +- OpenCTI Platform >= 6.4.5 +- SOCRadar API key +- Python 3.11+ + +### Configuration radar.format_type seems missing ------------------------------ In socradar/README.md <#3072 (comment)> : > + +### Requirements + +- OpenCTI Platform >= 6.4.5 +- SOCRadar API key +- Python 3.11+ + +### Configuration + +| Parameter | Docker envvar | Mandatory | Description | +| --- | --- | --- | --- | +| `opencti.url` | `OPENCTI_URL` | Yes | The URL of your OpenCTI platform | +| `opencti.token` | `OPENCTI_TOKEN` | Yes | Your OpenCTI admin token | +| `radar.base_feed_url` | `RADAR_BASE_FEED_URL` | Yes | SOCRadar API base URL | +| `radar.socradar_key` | `RADAR_SOCRADAR_KEY` | Yes | Your SOCRadar API key | +| `radar.interval` | `RADAR_INTERVAL` | Yes | Time between runs (in seconds, default: 600) | expected parameter seems to be run_interval ------------------------------ In socradar/src/lib/radar.py <#3072 (comment)> : > + "domain": ["domain-watchlist"], + "ip": ["ip-watchlist"], + "hash": ["file-hash-watchlist"], + } + return type_mapping.get(feed_type, ["malicious-activity"]) + + def _get_or_create_identity(self, maintainer_name): + """Step 2.1: Get existing identity or create new one for maintainer""" + try: + if maintainer_name in self.identity_mapping: + return self.identity_mapping[maintainer_name] + + current_time = datetime.utcnow() + + identity = Identity( + id=f"identity--{str(uuid.uuid4())}", This would create stochastic id and will prevent de-duplication performed by the platform. You should rather use pycti.Identity.generate_id method. ------------------------------ In socradar/src/lib/radar.py <#3072 (comment)> : > + # Step 1.0: Initialize connector from config + config_path = os.path.dirname(os.path.abspath(__file__)) + "/../config.yml" + config = yaml.load(open(config_path), Loader=yaml.SafeLoader) + self.helper = OpenCTIConnectorHelper(config) + + # Step 1.1: Get radar-specific configurations + self.base_url = get_config_variable( + "RADAR_BASE_FEED_URL", ["radar", "base_feed_url"], config + ) + self.format_type = get_config_variable( + "RADAR_FORMAT_TYPE", ["radar", "format_type"], config + ) + self.socradar_key = get_config_variable( + "RADAR_SOCRADAR_KEY", ["radar", "socradar_key"], config + ) + self.collections = get_config_variable( This does not seems possible to use self.collections as expected in def _process_feed(self, work_id: str) -> None: """Step 5.0: Process feed collection""" try: self.helper.log_info("Starting feed collection...") # Step 5.1: Process each collection for collection_name, collection_data in self.collections.items(): try: collection_id = collection_data["id"][0] A Mapping of Sequence is used in this code whereas a string would be loaded by self.collections = get_config_variable( "RADAR_COLLECTIONS_UUID", ["radar", "collections_uuid"], config ) when using RADAR_COLLECTIONS_UUID env var. To resolve this, you could try casting the variable as follows: if isinstance(my_raw_var, str): try: my_var = json.loads(my_raw_var) except json.JSONDecodeError: raise ValueError("Expecting a valid json string Mapping for my_raw_var") — Reply to this email directly, view it on GitHub <#3072 (review)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/BL7TPYTX4QDK7BZDSF44NDD2HP5U5AVCNFSM6AAAAABS73TNPSVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDKMRTGE2DOMZXGI> . You are receiving this because you authored the thread.Message ID: ***@***.***>

--

-------------------------------------- Disclaimer: The content of this email is confidential and intended for the recipient specified in message only. It is strictly forbidden to share any part of this message with any third party, without a written consent of the sender. If you received this message by mistake, please reply to this message and follow with its deletion, so that we can ensure such a mistake does not occur in the future.

[Radargoger]

Dear Team, I hope this message finds you well. I wanted to kindly follow up regarding the connector implementation we discussed earlier. As mentioned in my previous email, we’ve addressed the remarks and made the necessary improvements. We’re excited to see the connector in action. Please let us know if there’s anything else needed from our side to move forward or if you require further adjustments. Best regards, Burak GOGER Senior Product Manager <https://socradar.io/> ***@***.*** +90 (552) 398 30 04 <+90+505+500+00+00> socradar.io *HQ Office:* 254 Chapman Rd, Ste 208 Newark, Delaware 19702 USA *EMEA Office:* İçerenköy Mah. Umut Sok. Quick Tower No:10-12 Ataşehir/İstanbul/Türkiye [image: linkedin] <https://www.linkedin.com/company/socradar> [image: twitter] <https://twitter.com/socradar> [image: linkedin] <https://www.youtube.com/channel/UClUyizkV30njCwQUcPhJ_qg> <https://www.gartner.com/reviews/market/security-threat-intelligence-services/vendor/socradar/product/socradar-digital-risk-protection-platform>

…

On Fri, Dec 27, 2024 at 7:47 PM Burak Goger ***@***.***> wrote: Thank you, @flavienSindou, We’ve addressed the remarks and made the necessary improvements to ensure optimal performance. We’re excited to see this connector in action within the system and look forward to your thoughts on the final implementation. Wishing you a happy (belated) Christmas and all the best for the New Year! 🎄✨ Best regards Burak GOGER Senior Product Manager <https://socradar.io/> ***@***.*** +90 (552) 398 30 04 <+90+505+500+00+00> socradar.io *HQ Office:* 254 Chapman Rd, Ste 208 Newark, Delaware 19702 USA *EMEA Office:* İçerenköy Mah. Umut Sok. Quick Tower No:10-12 Ataşehir/İstanbul/Türkiye [image: linkedin] <https://www.linkedin.com/company/socradar> [image: twitter] <https://twitter.com/socradar> [image: linkedin] <https://www.youtube.com/channel/UClUyizkV30njCwQUcPhJ_qg> <https://www.gartner.com/reviews/market/security-threat-intelligence-services/vendor/socradar/product/socradar-digital-risk-protection-platform> On Thu, Dec 26, 2024 at 4:08 PM flavienSindou ***@***.***> wrote: > ***@***.**** approved this pull request. > > Thank you for the improvements you’ve made. > > I have a few additional remarks to ensure that your connector performs > optimally. > > Additionally, I still encourage you to use the OpenCTI dedicated > Scheduler instead of implementing your own. > > Let me know if you need any support with this—I’ll be happy to assist. I > look forward to reviewing your next changes! > > https://filigran.io/auto-backpressue-control-octi-connectors/ > ------------------------------ > > In socradar/README.md > <#3072 (comment)> > : > > > @@ -0,0 +1,40 @@ > +# OpenCTI SOCRadar Connector > + > +The connector imports threat intelligence feeds from SOCRadar into OpenCTI. It processes various types of indicators including IP addresses, domains, URLs, and file hashes. > + > +SOCRadar provides comprehensive threat intelligence feeds that can be used to detect and prevent various types of cyber threats. The connector fetches these feeds and converts them into standardized STIX2 format for use in OpenCTI, enabling organizations to enhance their threat detection and response capabilities. > + > +## Installation > + > +### Requirements > + > +- OpenCTI Platform >= 6.4.5 > +- SOCRadar API key > +- Python 3.11+ > + > +### Configuration > > radar.format_type seems missing > ------------------------------ > > In socradar/README.md > <#3072 (comment)> > : > > > + > +### Requirements > + > +- OpenCTI Platform >= 6.4.5 > +- SOCRadar API key > +- Python 3.11+ > + > +### Configuration > + > +| Parameter | Docker envvar | Mandatory | Description | > +| --- | --- | --- | --- | > +| `opencti.url` | `OPENCTI_URL` | Yes | The URL of your OpenCTI platform | > +| `opencti.token` | `OPENCTI_TOKEN` | Yes | Your OpenCTI admin token | > +| `radar.base_feed_url` | `RADAR_BASE_FEED_URL` | Yes | SOCRadar API base URL | > +| `radar.socradar_key` | `RADAR_SOCRADAR_KEY` | Yes | Your SOCRadar API key | > +| `radar.interval` | `RADAR_INTERVAL` | Yes | Time between runs (in seconds, default: 600) | > > expected parameter seems to be run_interval > ------------------------------ > > In socradar/src/lib/radar.py > <#3072 (comment)> > : > > > + "domain": ["domain-watchlist"], > + "ip": ["ip-watchlist"], > + "hash": ["file-hash-watchlist"], > + } > + return type_mapping.get(feed_type, ["malicious-activity"]) > + > + def _get_or_create_identity(self, maintainer_name): > + """Step 2.1: Get existing identity or create new one for maintainer""" > + try: > + if maintainer_name in self.identity_mapping: > + return self.identity_mapping[maintainer_name] > + > + current_time = datetime.utcnow() > + > + identity = Identity( > + id=f"identity--{str(uuid.uuid4())}", > > This would create stochastic id and will prevent de-duplication performed > by the platform. > > You should rather use pycti.Identity.generate_id method. > ------------------------------ > > In socradar/src/lib/radar.py > <#3072 (comment)> > : > > > + # Step 1.0: Initialize connector from config > + config_path = os.path.dirname(os.path.abspath(__file__)) + "/../config.yml" > + config = yaml.load(open(config_path), Loader=yaml.SafeLoader) > + self.helper = OpenCTIConnectorHelper(config) > + > + # Step 1.1: Get radar-specific configurations > + self.base_url = get_config_variable( > + "RADAR_BASE_FEED_URL", ["radar", "base_feed_url"], config > + ) > + self.format_type = get_config_variable( > + "RADAR_FORMAT_TYPE", ["radar", "format_type"], config > + ) > + self.socradar_key = get_config_variable( > + "RADAR_SOCRADAR_KEY", ["radar", "socradar_key"], config > + ) > + self.collections = get_config_variable( > > This does not seems possible to use self.collections as expected in > > def _process_feed(self, work_id: str) -> None: > """Step 5.0: Process feed collection""" > try: > self.helper.log_info("Starting feed collection...") > > # Step 5.1: Process each collection > for collection_name, collection_data in self.collections.items(): > try: > collection_id = collection_data["id"][0] > > A Mapping of Sequence is used in this code whereas a string would be > loaded by > > self.collections = get_config_variable( > "RADAR_COLLECTIONS_UUID", ["radar", "collections_uuid"], config > ) > > when using RADAR_COLLECTIONS_UUID env var. > > To resolve this, you could try casting the variable as follows: > > if isinstance(my_raw_var, str): > try: > my_var = json.loads(my_raw_var) > except json.JSONDecodeError: > raise ValueError("Expecting a valid json string Mapping for my_raw_var") > > > — > Reply to this email directly, view it on GitHub > <#3072 (review)>, > or unsubscribe > <https://github.com/notifications/unsubscribe-auth/BL7TPYTX4QDK7BZDSF44NDD2HP5U5AVCNFSM6AAAAABS73TNPSVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDKMRTGE2DOMZXGI> > . > You are receiving this because you authored the thread.Message ID: > ***@***.***> >

--

-------------------------------------- Disclaimer: The content of this email is confidential and intended for the recipient specified in message only. It is strictly forbidden to share any part of this message with any third party, without a written consent of the sender. If you received this message by mistake, please reply to this message and follow with its deletion, so that we can ensure such a mistake does not occur in the future.

[helene-nguyen]

@Radargoger Thank you and sorry for the delay of the answer.
To allow us to move forward, could you please sign all of your commits and resolve the conflicts ?