loading...

基础镜像与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 -rchariot-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')

日志记录

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

Sdk 1

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

注意:您不能使用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)}"

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

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

Sdk 2

函数

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

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

动作|触发器文件之间共享的函数和方法应添加到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(...)。

image-missing