基础镜像与SDK

基础镜像

当我们使用chariot-plugin -g命令生成插件代码框架目录后，我们在根目录可以看到一个Dockerfile文件。它负责将我们编写的插件打包成docker镜像，从而启动并运行插件。

Dockerfile的基本内容如下:

FROM python:3.8-slim

WORKDIR /python/src
ADD . /python/src
RUN chmod +x /python/src/main.py

# replace apt source && config pip source
RUN sed -i "s@http://deb.debian.org@http://mirrors.aliyun.com@g" /etc/apt/sources.list && \
	sed -i "s@http://security.debian.org@http://mirrors.aliyun.com@g" /etc/apt/sources.list && \
	apt-get update && apt-get clean && \
	pip config set global.index-url https://mirrors.aliyun.com/pypi/simple && \
	/bin/echo -e [easy_install]\\nindex-url=https://mirrors.aliyun.com/pypi/simple > ~/.pydistutils.cfg

RUN apt-get update && apt-get install -y vim


RUN pip install loguru
RUN pip install fastapi
RUN pip install uvicorn
RUN pip install jsonpointer
RUN pip install requests


# Install pip dependencies
RUN if [ -f requirements.txt ]; then pip install -r requirements.txt; fi

ENTRYPOINT ["/python/src/main.py"]

可以看出，当运行chariot-plugin -r或chariot-plugin --http等命令时，开发工具将以python:3.8-slim作为基础镜像，打包成新的插件镜像，最后以插件镜像创建容器，通过ENTRYPOINT入口启动插件服务并执行。

插件开发工具默认使用python:3.8-slim作为基础镜像，请根据需要修改Dockerfile中基础镜像的引用。

注意: 由于我们的基础镜像放在私有仓库中，因此需要您在本地配置仓库的地址:

# 添加镜像仓库地址
$ vim /etc/docker/daemon.json
{
  "registry-mirrors": [
    "https://registry.cn-shenzhen.aliyuncs.com",
    "https://wf3hesj8.mirror.aliyuncs.com",
    "https://hub-mirror.c.163.com"
  ]
}

# 重启 docker 服务
$ systemctl restart docker

SDK

在基础镜像中，我们内置了插件SDK，名为SDK，我们开发的插件依赖于此。我们可以在生成的插件代码中看到如下引入:

from SDK.subassembly import Actions
from SDK.base import Core

因此只有通过我们提供的基础镜像打包的插件才能正常运行插件。

生成插件时，SDK就会自动生成，SDK提供基本的与千乘系统的交互功能，如遇特殊需求可随心更改SDK

我们将在下文介绍SDK提供的基础函数和使用方法。

参数

动作

plugin.spec.yaml文件中定义的输入变量可以在一个名为params的字典中可用。可以通过变量/键值常量访问输入的值，该常量是由代码生成到相应的模式文件中：

def run(self, params={}):
  api_key = params.get("api_key")  # Reflects a defined input called 'api_key'

由于这是一个字典，建议使用params.get()它来访问它里面的值，因为它：

如果缺少键值，则允许可选的默认值
如果键值不存在则返回None，但params['var']会抛出KeyError。

请注意，代码生成的常量仅适用于任何潜在变量嵌套的第一级。

触发器

触发器是长时间运行的进程，用于轮询/发出新事件，然后将事件发送到引擎以启动工作流。由于它们是工作流获取数据的方式，因此所有工作流都以触发器开始。激活工作流后，触发器代码将在Docker容器中执行。当触发器初始化时，连接对象被创建，然后触发器进入循环，等待消息或轮询新数据。

与动作不同，触发器不会根据输入返回任何内容 - 相反，它们会根据它执行某些动作，然后将其传送到引擎。

为了增加可用性，所有触发器都应支持一个叫frequency的输入，该输入以SDK语言传递给睡眠机制。

如果在规范配置文件中配置了triggers，它们将出现在<plugin_name>/triggers/<trigger_name>.py

代码

触发器代码应放在触发器运行循环的主体中。

您可以调整计时器以满足插件的需要。默认情况下，它每5秒发出一次事件。

该self.send()方法接受一个字典，并且是将字典传递给引擎以启动工作流的函数。这个字典将会变成JSON对象，然后可以通过UI被其他插件使用。

def run(self, params={}):
  """Run the trigger"""
  # send a test event
  while True:
    """TODO implement this"""
    self.send({})
    time.sleep(5)

连接

插件作者要负责编写代码来管理连接。生成的代码只是用来占位的，可以显示正确填充代码的位置。

您可以在connection.py使用一个叫data字典访问在plugin.spec.yaml被定义的连接变量。可以通过变量/键值常量访问输入的值，该常量由代码自动生成到相应的模式文件中：

def connection(self, params):
  hostname = params.get("hostname")  # Reflects a defined input called 'hostname'

您还需要在run()方法中访问连接变量以获取连接信息。

def run(self, params):
  value = self.var

请注意，代码生成的常量仅适用于任何潜在变量嵌套的第一级。在username_password凭证类型的情况下，代码生成的常量仅出现在顶级变量中。一个例子如下：

def connect(self, params):
  username_pass = params.get("credentials")  # A dictionary with username and password
  username, password = username_pass.get('username'), username_pass.get('password')

日志记录

日志记录包括警告和错误消息，它们将在“所有输出”的“日志”部分中显示给用户。

此外，您还可以通过抛出异常来进行日志记录，但这会导致插件运行失败。请参阅错误。

注意：您不能使用print()或其他stdout方法来显示文本。这是因为该产品要求在stdout上使用SDK所接收的的JSON载荷。

错误

请参阅新的错误处理指南。

测试

插件测试方法默认测试连接器连接情况，也可重写实现自己的测试方法，例如对API进行身份验证以查看是否输入了正确的凭据。在插件步骤保存到工作流程之前，测试方法将会被调用。它应该包含对插件各功能的一些测试，例如网络访问。

引发异常将导致测试方法失败。

@CatchErr.print_err_stack
def test(self, connect):

	try:
		self.connection(connect)
		output = {
			"type": "action_event",
			"version": "v1",
			"body": {
				"meta": {},
				"output": None,
				"status": "ok",
				"log": "succ",
			}
		}

		return output
	except Exception as e:
		assert False, f"连接器　测试失败 - {str(e)}"

在开发期间，测试可以从命令行执行，或者在配置插件后通过使用千乘系统测试面板执行。

有关测试插件的更多详细信息和示例，请参阅运行插件。

函数

函数是没被附加到任何东西的顶级对象。

我们建议你使用良好的程序设计，例如将程序分解为更小的函数。这使插件具有更高的可读性和可管理性。

动作|触发器文件之间共享的函数和方法应添加到util目录中。

用来压缩的插件有许多共享模块放置在util目录中。plugin_name/util

然后可以在行为/触发器文件中导入这些模块以供使用：

from util import utils

方法

方法是被附加到对象上面的函数，而不是在全局范围内可自由使用的函数。其中的主要区别是方法能够访问它们所附加对象的本地（可能是私有）状态。

与函数一样，请使用良好的程序设计实践，例如将程序分解为更小的部分。这使插件更具可读性和可管理性。

因此，您应该避免在可能的情况下在生成的对象上创建自定义方法，而是使用更容易恢复的utils。

下面给出一个例子，我们可以使用以下的方式，而不是通过大量的re.search和条件匹配来测试一个值的存在。get_value方法中的正则表达式从stdout里的字符串中匹配\nkey: value并提取值。

def get_value(self, key, stdout):
  '''Extracts value from key: value pair'''
  '''Example: regex = "\nDomain Name: (google.com)\n"'''
  regex = r"\n" + re.escape(key) + r": (.*)\n"
  r = re.search(regex, stdout)
  '''Only return the value in the group 1 if it exists'''
  if hasattr(r, 'group'):
    if r.lastindex == 1:
      return r.group(1)...
def run(self, params={}):
  '''Initialize list with keys for matching'''
  keys = [
    'Domain Name',
    'Registrar WHOIS Server',
    'Updated Date',
    'Creation Date',
    'Registrar',
    'Registrar Abuse Contact Email',
    'Registrar Abuse Contact Phone',
    'Registrant Country',
  ]

  for key in keys:
    '''Iterate over keys and store the extracted values into results'''
    results[key] = self.get_value(key, stdout)

  return results

一旦定义了函数，就可以在同一个python文件中调用它，例如 self.get_value(...)。