调试OpenCTI Connector时遇"An URL must be set"错误求助

我正在开发一个简易OpenCTI Connector以了解其工作机制，本地运行时提示"An URL must be set"错误。该连接器的功能是通过指定URL获取黑名单IPv4地址，转换为STIX Bundle后发送至OpenCTI，属于基础类型的连接器。我已在本地修改了配置文件中的token和id，但问题依然存在。

Connector代码 #

# coding: utf-8

import os
import sys
import yaml
import time
import requests

from datetime import datetime
from pycti import OpenCTIConnectorHelper, get_config_variable
from stix2 import Indicator,Bundle

class blacklistIPConnector:
    def __init__(self):
        # Instantiate the connector helper from config
        config_file_path = os.path.dirname(os.path.abspath(__file__)) + "/config.yml"
        config = (
            yaml.load(open(config_file_path), Loader=yaml.FullLoader)
            if os.path.isfile(config_file_path)
            else {}
        )
        self.helper = OpenCTIConnectorHelper(config)

        #Extra config
        self.update_existing_data = get_config_variable(
            "CONNECTOR_UPDATE_EXISTING_DATA",
            ["connector", "update_existing_data"],
            config,
        )
        self.confidence_level = get_config_variable(
            "CONNECTOR_CONFIDENCE_LEVEL",
            ["connector", "confidence_level"],
            config,
        )
        self.blacklistIP_interval = get_config_variable(
            "BLACKLISTIP_INTERVAL", ["blacklistIP", "interval"], config, True
        )
        self.blacklistIP_url = get_config_variable(
            "BLACKLISTIP_URL", ["blacklistIP", "url"], config, False
        )

    def get_interval(self) -> int:
        return int(self.blacklistIP_interval) * 60 * 60 * 24
    
    def _collect_intelligence(self) -> list:
        time_now = datetime.now()
        current_time = time_now.strftime("%H:%M")

        print("The current date and time is :", current_time)
        url = 'http://api.blocklist.de/getlast.php?time='+current_time
        response = requests.get(url)
        self.helper.log_debug("websites response is : " + response)
        if response.status_code == 200:
            #data = response.text.splitlines()
            message = (
                f"{self.helper.connect_name} connector successfully retrieved data and converting it to STIX2 Format "
                + str(time_now)
            )
            #self.helper.log_info(message)
            ##Calling the stix transformer
            data_to_bundle = self.create_stix_objects(response)
            print("HELOOOOOOOOOOOOOOOOO THESE ARE THE DATA TRANSFORMED TO STIX :"+data_to_bundle)
            message = (
                f"{self.helper.connect_name} Formated to STIX2 bundle "
                + str(time_now)
            )
            self.helper.log_info(message)
            return data_to_bundle
        else:
            message = (
                f"{self.helper.connect_name} Failed to retrieve data on "
                + str(time_now)
            )
            self.helper.log_warning(message)

    def create_stix_objects(self, data):
        stix_objects = []
        for line in data.splitlines():
            timestamp = int(time.time())
            # Assuming each line contains an IP address and other data
            ip = line.split(',')[0]  # Adjust this based on the actual data format
            indicator = Indicator(
                pattern="[ipv4-addr:value = '{}']".format(ip),
                pattern_type="stix",
                valid_from="'{}'".format(timestamp)  # Adjust the timestamp as needed
            )
            stix_objects.append(indicator)

        return Bundle(objects=stix_objects).serialize()

    def process_data(self):
        try:
            # Get the current timestamp and check
            timestamp = int(time.time())
            current_state = self.helper.get_state()
            if current_state is not None and "last_run" in current_state:
                last_run = current_state["last_run"]
                self.helper.log_info(
                    "Connector last run: "
                    + datetime.utcfromtimestamp(last_run).strftime("%Y-%m-%d %H:%M:%S")
                )
            else:
                last_run = None
                self.helper.log_info("Connector has never run")
            # If the last_run is more than interval-1 day
            if last_run is None or (
                (timestamp - last_run) > ((int(self.blacklistIP_interval) - 1) * 60 * 60 * 24)
            ):
                self.helper.log_info("Connector will run!")

                now = datetime.utcfromtimestamp(timestamp)
                friendly_name = "blacklistIP run @ " + now.strftime("%Y-%m-%d %H:%M:%S")
                work_id = self.helper.api.work.initiate_work(
                    self.helper.connect_id, friendly_name
                )
                # Retrieve blacklistIP stix file
                if (
                    self.blacklistIP_url is not None
                    and len(self.blacklistIP_url) > 0
                ):
                    blacklist_data = self._collect_intelligence(self.blacklistIP_url)
                    self.helper.log_debug(blacklist_data)
                    self.send_bundle(work_id, blacklist_data)

                # Store the current timestamp as a last run
                message = "Connector successfully run, storing last_run as " + str(
                    timestamp
                )
                self.helper.log_info(message)
                self.helper.set_state({"last_run": timestamp})
                self.helper.api.work.to_processed(work_id, message)
                self.helper.log_info(
                    "Last_run stored, next run in: "
                    + str(round(self.get_interval() / 60 / 60 / 24, 2))
                    + " days"
                )
            else:
                new_interval = self.get_interval() - (timestamp - last_run)
                self.helper.log_info(
                    "Connector will not run, next run in: "
                    + str(round(new_interval / 60 / 60 / 24, 2))
                    + " days"
                )
        except (KeyboardInterrupt, SystemExit):
            self.helper.log_info("Connector stop")
            sys.exit(0)
        except Exception as e:
            self.helper.log_error(str(e))

    def send_bundle(self, work_id: str, serialized_bundle: str) -> None:
        try:
            self.helper.send_stix2_bundle(
                serialized_bundle,
                entities_types=self.helper.connect_scope,
                update=self.update_existing_data,
                work_id=work_id,
            )
        except Exception as e:
            self.helper.log_error(f"Error while sending bundle: {e}")

    def run(self) -> None:
        self.helper.log_info("Fetching BlacklistIP framework...")

        get_run_and_terminate = getattr(self.helper, "get_run_and_terminate", None)
        if callable(get_run_and_terminate) and self.helper.get_run_and_terminate():
            self.process_data()
            self.helper.force_ping()
        else:
            while True:
                self.process_data()
                time.sleep(60)

if __name__ == "__main__":
    try:
        connector = blacklistIPConnector()
        connector.run()
    except Exception as e:
        print(e)
        time.sleep(10)
        exit(0)

config.yml.sample配置文件 #

opencti:
  url: 'http://localhost:8080'
  token: 'changeme'

connector:
  id: 'changeme'
  type: 'EXTERNAL_IMPORT'
  name: 'blacklistIP'
  scope: 'attack-pattern' # MIME type or SCO
  confidence_level: 100 # From 0 (Unknown) to 100 (Fully trusted)
  log_level: 'info'
  update_existing_data: false
  run_and_terminate: false

blacklistIP:
  interval: 1 # in days
  url: 'http://api.blocklist.de/getlast.php?time='

错误原因与解决方案 #

1. 配置文件未正确加载 #

代码中加载的是config.yml，如果仅修改了config.yml.sample但未将其复制并重命名为config.yml，会导致配置为空，OpenCTI ConnectorHelper初始化时会检查opencti.url等必填项，从而抛出"An URL must be set"错误。

解决：
将config.yml.sample复制一份，命名为config.yml，并在其中正确配置opencti.url、opencti.token、connector.id等参数。

2. 方法调用参数不匹配 #

在process_data方法中调用self._collect_intelligence(self.blacklistIP_url)，但_collect_intelligence方法定义未接收参数，这会触发错误，同时方法内部硬编码了URL，未使用配置中的self.blacklistIP_url。

解决：
修改_collect_intelligence方法定义，支持传入URL参数，并使用配置值作为默认：

def _collect_intelligence(self, url=None) -> list:
    target_url = url if url is not None else self.blacklistIP_url
    time_now = datetime.now()
    current_time = time_now.strftime("%H:%M")
    full_url = f"{target_url}{current_time}"
    
    print("The current date and time is :", current_time)
    response = requests.get(full_url)
    # 修复日志拼接错误
    self.helper.log_debug("websites response is : " + response.text)
    if response.status_code == 200:
        message = (
            f"{self.helper.connect_name} connector successfully retrieved data and converting it to STIX2 Format "
            + str(time_now)
        )
        self.helper.log_info(message)
        # 传入response.text而非response对象
        data_to_bundle = self.create_stix_objects(response.text)
        print("HELOOOOOOOOOOOOOOOOO THESE ARE THE DATA TRANSFORMED TO STIX :"+data_to_bundle)
        message = (
            f"{self.helper.connect_name} Formated to STIX2 bundle "
            + str(time_now)
        )
        self.helper.log_info(message)
        return data_to_bundle
    else:
        message = (
            f"{self.helper.connect_name} Failed to retrieve data on "
            + str(time_now)
        )
        self.helper.log_warning(message)

3. STIX对象格式错误 #

create_stix_objects方法中valid_from使用了带单引号的时间戳，不符合STIX规范，会导致后续发送Bundle失败。

解决：
将valid_from改为ISO格式的UTC时间字符串：

valid_from=datetime.utcfromtimestamp(timestamp).isoformat()

4. 连接器Scope配置错误 #

当前connector.scope设为attack-pattern，但连接器实际处理的是ipv4-addr类型的指标，会导致OpenCTI无法正确识别导入的实体。

解决：
修改config.yml中的connector.scope为ipv4-addr。

内容的提问来源于stack exchange，提问作者mohamed AFASSI