libknot的Python绑定
项目描述
Libknot API in Python
管理Knot DNS守护进程的Python接口。
目录
简介
如果共享库libknot.so不在库搜索路径中，则需要先加载库，例如。
import libknot
libknot.Knot("/usr/lib/libknot.so")
控制模块
使用此模块可以创建脚本来执行效率高的任务，这些任务原本需要复杂的多次调用 knotc 的 shell 脚本。与守护进程通信时，它使用与 knotc 工具相同的机制，即通过 Unix 套接字进行通信。
模块 API 存储在 libknot.control 中。
使用控制模块
模块使用步骤包括几个步骤
- 初始化和连接到守护进程控制套接字。
- 一个或多个控制操作。通过向守护进程发送带有可选数据的命令来调用操作。之后必须接收操作结果。
- 关闭连接和反初始化。
控制模块示例
import json
import libknot.control
# Initialization
ctl = libknot.control.KnotCtl()
ctl.connect("/var/run/knot/knot.sock")
ctl.set_timeout(60)
try:
# Operation without parameters
ctl.send_block(cmd="conf-begin")
resp = ctl.receive_block()
# Operation with parameters
ctl.send_block(cmd="conf-set", section="zone", item="domain", data="test")
resp = ctl.receive_block()
ctl.send_block(cmd="conf-commit")
resp = ctl.receive_block()
# Operation with a result displayed in JSON format
ctl.send_block(cmd="conf-read", section="zone", item="domain")
resp = ctl.receive_block()
print(json.dumps(resp, indent=4))
except libknot.control.KnotCtlError as exc:
# Print libknot error
print(exc)
finally:
# Deinitialization
ctl.send(libknot.control.KnotCtlType.END)
ctl.close()
# List configured zones (including catalog member ones)
ctl.send_block(cmd="conf-list", flags="z")
resp = ctl.receive_block()
for zone in resp['zone']:
print(zone)
# Print expirations as unixtime for all secondary zones
ctl.send_block(cmd="zone-status", flags="u")
resp = ctl.receive_block()
for zone in resp:
if resp[zone]["role"] == "master":
continue
expiration = resp[zone]["expiration"]
if expiration == "-":
print("Zone %s not loaded" % zone)
else:
print("Zone %s expires at %s" % (zone, resp[zone]["expiration"]))
探测模块
使用此模块可以接收运行中的守护进程带有活动探针模块的流量数据。
模块 API 存储在 libknot.probe 中。
使用探针模块
模块使用步骤包括几个步骤
- 初始化一个或多个探针通道
- 定期从通道接收数据单元并进行数据处理
探针模块示例
import libknot.probe
# Initialization of the first probe channel stored in `/run/knot`
probe = libknot.probe.KnotProbe("/run/knot", 1)
# Array for storing up to 8 data units
data = libknot.probe.KnotProbeDataArray(8)
while (True):
# Receiving data units with timeout of 1000 ms
if probe.consume(data, 1000) > 0:
# Printing received data units in the default format
for item in data:
print(item)
Dname模块
此模块提供了一些与 dname 相关的操作。
使用 Dname 模块
从包含文本 dname 的字符串初始化 dname 对象。然后可以将 dname 转换为线格式或转换回文本格式。
Dname 模块示例
import libknot.dname
dname1 = libknot.dname.KnotDname("knot-dns.cz")
print("%s: wire: %s size: %u" % (dname1.str(), dname1.wire(), dname1.size()))
dname2 = libknot.dname.KnotDname("e\\120ample.c\om.")
print("%s: wire: %s size: %u" % (dname2.str(), dname2.wire(), dname2.size()))
dname3 = libknot.dname.KnotDname(dname_wire=b'\x02cz\x00')
print("%s: wire: %s size: %u" % (dname3.str(), dname3.wire(), dname3.size()))
knot-dns.cz.: wire: b'\x08knot-dns\x02cz\x00' size: 13
example.com.: wire: b'\x07example\x03com\x00' size: 13
cz.: wire: b'\x02cz\x00' size: 4
