基础镜像与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(...)。