网络网络安全【免费下载链接】scapyScapy: the Python-based interactive packet manipulation program library.项目地址https://gitcode.com/gh_mirrors/sc/scapy点击查看免费下载Scapy 内置的Automaton框架允许你用纯 Python 方法加装饰器的方式快速搭建确定性的网络协议状态机状态、转移、动作和超时都以装饰器元信息的形式附着在普通方法上非常适合实现 TFTP 客户端、TLS 握手、自动重传等需要监听收包 超时 状态切换的网络逻辑。读完本文你将掌握ATMT.state / condition / receive_condition / timeout / timer / action等装饰器的完整用法能独立写出可运行、可调试、可自动绘图的协议自动机。本文以仓库文档 automaton.rst 为主体结合 automaton.py 源码展开。什么是 Scapy AutomatonScapy 的 Automaton 并不拘泥于 Moore 或 Mealy 等特定自动机模型而是提供一套灵活的框架让你自己选择组织方式。它的核心特征是确定性一个自动机包含若干状态其中有起始状态start state以及若干结束状态end和错误状态error。状态之间存在转移transition转移可以由三种事件触发特定条件成立ATMT.condition收到特定数据包ATMT.receive_condition另有eof/ioevent等超时到达ATMT.timeout/ATMT.timer。当转移被触发时会执行一个或多个动作action。一个动作可以绑定到多个转移上参数既可以从状态传给转移也可以从转移传给状态与动作。从程序员视角看状态、转移、动作本质上都是Automaton子类的方法通过装饰器附加自动机运行所需的元信息。这一点在 automaton.py 的Automaton_metaclass元类中体现得最清楚元类在类创建时遍历所有被atmt_type标记的成员把它们分类收集进states、conditions、recv_conditions、timeout、actions、eofs等类级字典并按优先级排序从而形成一张可执行的状态转移表。第一个例子HelloWorld按照惯例状态方法名用大写但这并非强制先看一个最简自动机class HelloWorld(Automaton): ATMT.state(initial1) def BEGIN(self): print(StateBEGIN) ATMT.condition(BEGIN) def wait_for_nothing(self): print(Wait for nothing...) raise self.END() ATMT.action(wait_for_nothing) def on_nothing(self): print(Action on nothing condition) ATMT.state(final1) def END(self): print(StateEND)这里出现了 3 个装饰器ATMT.state把方法标记为状态可选参数initial、final、stop、error置为真值后成为对应特殊状态ATMT.condition标记转移方法参数是它所属状态的方法名即状态方法本身当自动机进入该状态后会被调用ATMT.action把方法绑定到某个转移上当该转移被触发即条件方法raise出新状态时执行。运行效果如下 aHelloWorld() a.run() StateBEGIN Wait for nothing... Action on nothing condition StateEND a.destroy()这个自动机的状态图如下它描述了BEGIN → END的流转这张图不是手绘的而是可以直接从代码自动生成 HelloWorld.graph()graph()通过build_graph()把状态、转移、超时、动作汇总成 DOT 描述再调用 utils.py 中的do_graph()交给 Graphviz 渲染见 automaton.py因此你的机器上需要安装dot命令。两个必须记住的使用要点Automaton可以用restart()重置之后可以再次运行。使用完毕后务必调用destroy()清理资源尤其在 PyPy 上垃圾回收行为不同不清理可能泄漏文件描述符。状态转移机制raise 一个状态异常理解 Automaton 的关键在于ATMT.state装饰器会把一个普通方法变换成一个返回异常对象的函数。当你raise这个异常时自动机的状态就切换了。如果状态切换发生在某个转移方法里绑定到该转移的所有动作会被依次调用。raise时传入的实参会被保留最终原样交付给目标状态方法。举例说明定义如下的状态ATMT.state() def MY_STATE(self, param1, param2): print(stateMY_STATE. param1%r param2%r % (param1, param2))它可以通过下面这个接收条件转移进入ATMT.receive_condition(ANOTHER_STATE) def received_ICMP(self, pkt): if ICMP in pkt: raise self.MY_STATE(got icmp, pkt[ICMP].type)这里raise self.MY_STATE(got icmp, pkt[ICMP].type)中的两个参数会被原样传给MY_STATE方法。如果转移上还绑定了动作而动作也需要参数那么就要在 raise 之前调用异常的action_parameters()方法暂存参数ATMT.action(received_ICMP) def on_ICMP(self, icmp_type, icmp_code): self.retaliate(icmp_type, icmp_code) ATMT.receive_condition(ANOTHER_STATE) def received_ICMP(self, pkt): if ICMP in pkt: raise self.MY_STATE(got icmp, pkt[ICMP].type).action_parameters( pkt[ICMP].type, pkt[ICMP].code)这样on_ICMP就能拿到icmp_type和icmp_code两个参数了。底层实现对应 automaton.py 中的ATMT.NewStateRequested异常类它保存目标状态、自动机实例、args/kargs并在构造时初始化一份空的action_args/action_kargsaction_parameters()就是向这两个属性写入参数并返回自身供raise表达式使用。转移被触发后_run_condition会捕获该异常把state_req.action_args/action_kargs透传给绑定动作再重新raise交给状态循环切换状态。真实示例一个可运行的 TFTP 读取客户端文档给出的完整实例是一个支持发起读请求的 TFTP 客户端这也是 Scapy 自带实现对应 tftp.py 中的 TFTP 自动机一族class TFTP_read(Automaton): def parse_args(self, filename, server, sportNone, port69, **kargs): Automaton.parse_args(self, **kargs) self.filename filename self.server server self.port port self.sport sport def master_filter(self, pkt): return ( IP in pkt and pkt[IP].src self.server and UDP in pkt and pkt[UDP].dport self.my_tid and (self.server_tid is None or pkt[UDP].sport self.server_tid) ) # BEGIN ATMT.state(initial1) def BEGIN(self): self.blocksize512 self.my_tid self.sport or RandShort()._fix() bind_bottom_up(UDP, TFTP, dportself.my_tid) self.server_tid None self.res b self.l3 IP(dstself.server)/UDP(sportself.my_tid, dportself.port)/TFTP() self.last_packet self.l3/TFTP_RRQ(filenameself.filename, modeoctet) self.send(self.last_packet) self.awaiting1 raise self.WAITING() # WAITING ATMT.state() def WAITING(self): pass ATMT.receive_condition(WAITING) def receive_data(self, pkt): if TFTP_DATA in pkt and pkt[TFTP_DATA].block self.awaiting: if self.server_tid is None: self.server_tid pkt[UDP].sport self.l3[UDP].dport self.server_tid raise self.RECEIVING(pkt) ATMT.action(receive_data) def send_ack(self): self.last_packet self.l3 / TFTP_ACK(block self.awaiting) self.send(self.last_packet) ATMT.receive_condition(WAITING, prio1) def receive_error(self, pkt): if TFTP_ERROR in pkt: raise self.ERROR(pkt) ATMT.timeout(WAITING, 3) def timeout_waiting(self): raise self.WAITING() ATMT.action(timeout_waiting) def retransmit_last_packet(self): self.send(self.last_packet) # RECEIVED ATMT.state() def RECEIVING(self, pkt): recvd pkt[Raw].load self.res recvd self.awaiting 1 if len(recvd) self.blocksize: raise self.WAITING() raise self.END() # ERROR ATMT.state(error1) def ERROR(self, pkt): split_bottom_up(UDP, TFTP, dportself.my_tid) return pkt[TFTP_ERROR].summary() # END ATMT.state(final1) def END(self): split_bottom_up(UDP, TFTP, dportself.my_tid) return self.res运行方式 atmt TFTP_read(my_file, 192.168.1.128) atmt.run() atmt.destroy()其状态图如下由TFTP_read.graph()自动生成这个例子浓缩了自动机的大部分关键技巧值得逐点拆解parse_args()传参构造TFTP_read(my_file, 192.168.1.128)时filename与server会进入parse_args同时必须调用父类的Automaton.parse_args(self, **kargs)完成框架自身的参数处理master_filter()集中过滤只接收来自目标服务器、目的端口等于本端 TID临时端口、且源端口匹配或尚未确定的 UDP 包这样各个receive_condition里就不用反复做这些检查了初始化状态BEGIN设置块大小 512、用RandShort()._fix()生成随机 TID、通过bind_bottom_up(UDP, TFTP, ...)绑定底层解析、发送 RRQ 请求然后raise self.WAITING()进入等待receive_conditionaction协作收到正确块号的数据后切到RECEIVING同时动作send_ack自动发送 ACK——注意动作send_ack没有参数因为 ACK 内容都来自实例属性prio控制判定顺序receive_error声明prio1保证它在默认优先级 0 的receive_data之后判定避免 ERROR 包被数据分支抢先处理timeout实现重传WAITING状态 3 秒无响应则raise self.WAITING()原地重入并通过绑定的动作retransmit_last_packet重发最后一个包错误与结束状态ERROR标记error1返回错误摘要END标记final1返回累计的self.res作为自动机运行结果。装饰器详解状态装饰器ATMT.stateATMT.state接受 4 个可选参数initial、final、stop、error置为非零即把状态标记为对应角色class Example(Automaton): ATMT.state(initial1) def BEGIN(self): pass ATMT.state() def SOME_STATE(self): pass ATMT.state(final1) def END(self): return Result of the automaton: 42 ATMT.state(stop1) def STOP(self): print(SHUTTING DOWN...) # e.g. close sockets... ATMT.condition(STOP) def is_stopping(self): raise self.END() ATMT.state(error1) def ERROR(self): return Partial result, or explanation # [...]各角色的语义参数语义initial1自动机启动时进入的状态final1自动机运行结束的标志该状态方法的返回值会成为run()的结果stop1调用atmt.stop()后无论当前处于哪个状态都会迁移到的停止状态error1标记自动机出错的状态进入后自动机以错误告终文档特别提示如果没有定义stop状态那么调用stop()与forcestop()的效果等价源码中确实如此_do_control收到STOP命令时若存在stop_state就切换到它否则直接跳出循环见 automaton.py。元类还强制约束整个类只能有一个 stop 状态定义第二个会直接抛ValueError见 automaton.py。文档以 TCP 客户端为例给出了一张典型状态图START事件是initial1STOP事件是stop1CLOSED事件是final1转移装饰器condition / receive_condition / eof / timeout / timer转移方法由ATMT.condition、ATMT.receive_condition、ATMT.eof、ATMT.timeout、ATMT.timer之一装饰它们都以所属状态的方法名为参数。timeout与timer还额外要求一个必填的timeout参数单位是秒。二者的区别是ATMT.timeout只触发一次ATMT.timer触发后自动重载适合周期性发送 keep-alive 包。condition与receive_condition有可选的prio参数用于强制判定顺序默认优先级为 0同优先级的转移调用顺序是不确定的。源码中Timer类的autoreload标志就是timeout与timer的唯一区别ATMT.timer构造Timer(timeout, prioprio, autoreloadTrue)见 automaton.py超时归零后_decrement会把溢出量加回计时器实现自动续期见 automaton.py。自动机进入某个状态后的执行顺序如下先执行状态方法本身状态方法返回后立即按优先级从小到大运行该状态绑定的condition方法直到某个条件raise出新状态之后每当收到一个通过master_filter过滤的数据包就按优先级从小到大调用receive_condition方法自进入当前状态起计时超时到达时调用对应的timeout/timer方法若 socket 在状态期间抛出EOFError连接关闭则调用该状态的ATMT.eof转移若没有定义 eof 转移自动机会以异常退出。class Example(Automaton): ATMT.state() def WAITING(self): pass ATMT.condition(WAITING) def it_is_raining(self): if not self.have_umbrella: raise self.ERROR_WET() ATMT.receive_condition(WAITING, prio1) def it_is_ICMP(self, pkt): if ICMP in pkt: raise self.RECEIVED_ICMP(pkt) ATMT.receive_condition(WAITING, prio2) def it_is_IP(self, pkt): if IP in pkt: raise self.RECEIVED_IP(pkt) ATMT.timeout(WAITING, 10.0) def waiting_timeout(self): raise self.ERROR_TIMEOUT()上述时序正是_do_iter()的循环逻辑先执行state.run()若error状态则抛ErrorState若final状态则记录final_state_output并返回随后按序运行conditions若既没有接收条件、ioevent 也没有定时器就会抛出Stuck异常stuck in [状态名]最后通过select同时等待命令管道、监听 socket 与 io 事件并递减/检查定时器见 automaton.py。动作装饰器ATMT.action动作方法用ATMT.action装饰第一个参数是它绑定的转移方法第二个可选参数是优先级prio默认 0。一个动作方法可以通过多次叠加装饰器绑定到多个转移from random import random class Example(Automaton): ATMT.state(initial1) def BEGIN(self): pass ATMT.state(final1) def END(self): pass ATMT.condition(BEGIN, prio1) def maybe_go_to_end(self): if random() 0.5: raise self.END() ATMT.condition(BEGIN, prio2) def certainly_go_to_end(self): raise self.END() ATMT.action(maybe_go_to_end) def maybe_action(self): print(We are lucky...) ATMT.action(certainly_go_to_end) def certainly_action(self): print(We are not lucky...) ATMT.action(maybe_go_to_end, prio1) ATMT.action(certainly_go_to_end, prio1) def always_action(self): print(This wasnt luck!...)两种可能的输出 aExample() a.run() We are not lucky... This wasnt luck!... a.run() We are lucky... This wasnt luck!... a.destroy()注意always_action分别以prio1绑定到两个转移上。对于certainly_go_to_end这一转移certainly_action默认 prio0先于always_actionprio1执行两个动作的调用顺序由元类收集后按prio排序决定见 automaton.py。动作可以接收参数在 raise 下一个状态时调用action_parameters()。下面的例子中send_copy动作接收由is_fin传过来的数据包class Example(Automaton): ATMT.state() def WAITING(self): pass ATMT.state() def FIN_RECEIVED(self): pass ATMT.receive_condition(WAITING) def is_fin(self, pkt): if pkt[TCP].flags.F: raise self.FIN_RECEIVED().action_parameters(pkt) ATMT.action(is_fin) def send_copy(self, pkt): send(pkt)两个需要覆写的钩子方法parse_args()与master_filter()是 Automaton 预留的两个覆写点parse_args()在__init__()和run()时用传入的实参调用用来参数化自动机的行为。框架默认实现见 automaton.py接受debug非零时把conf.logLevel调为 DEBUG、store是否把收发数据包存入self.packets和session会话处理三个内置参数其余参数留在self.socket_kargs里供创建底层 socket 使用覆写时必须记得调用父类版本。master_filter()每个被嗅探到的包都会先经过它返回True才被认为是本自动机感兴趣的包。默认实现直接返回True见 automaton.py。当你在实现具体协议时应当在这里过滤出属于本连接的数据包如 TFTP 示例中按源 IP、源/目的端口过滤这样各个转移里就无需重复做全套合法性检查。定时器的运行时配置有些协议允许在运行时调整定时器。框架提供timer_by_name()方法通过转移方法名拿到对应的Timer对象再调用set()修改超时值。Timer提供get()/set()读取与修改超时长度见 automaton.pyclass Example(Automaton): def __init__(self, *args, **kwargs): super(Example, self).__init__(*args, **kwargs) timer self.timer_by_name(waiting_timeout) timer.set(1) ATMT.state(initial1) def WAITING(self): pass ATMT.state(final1) def END(self): pass ATMT.timeout(WAITING, 10.0) def waiting_timeout(self): raise self.END()timer_by_name的实现会遍历类中所有状态的_TimerList按atmt_condname匹配返回对应的Timer实例见 automaton.py。这样可以在不修改类定义的前提下把默认 10 秒的超时改为 1 秒——对需要按协议参数动态调整 RTO 的场景非常实用。生命周期 API 与底层运行机制Automaton的公开生命周期 API 如下见 automaton.py方法作用start(*args, **kargs)启动控制线程自动机开始运行run(resumeNone, waitTrue)发送 RUN 命令并默认阻塞等待结束返回 final 状态的输出可被KeyboardInterrupt打断发送 FREEZErunbg(resumeNone, waitFalse)后台运行版本不阻塞当前线程stop(waitTrue)请求停止若有 stop 状态则迁移过去否则等价于强停forcestop(waitTrue)无条件立即停止restart(*args, **kargs)先stop()再start()用于重置后重新运行destroy()销毁停止后的自动机关闭命令管道与所有 io 管道PyPy 上必须显式调用isrunning()查询是否仍在运行自动机内部是一个双管道驱动的事件循环主线程通过cmdin管道下发RUN/NEXT/FREEZE/STOP/FORCESTOP/BREAKPOINT/INTERCEPT等命令见_ATMT_Commandautomaton.py控制线程_do_control执行状态循环并通过cmdout返回END/EXCEPTION/SINGLESTEP/BREAKPOINT/INTERCEPT消息。跨平台的多路复用由select_objects()完成它在 Windows 上用WaitForMultipleObjects模拟select语义见 automaton.py命令与 IO 数据通过ObjectPipe传递它内部用os.pipe()加队列实现可 select 的对象管道见 automaton.py。框架还内置了调试与交互能力debug参数与self.debug(lvl, msg)分级输出运行轨迹add_breakpoints/add_interception_points可以在指定状态挂起自动机配合next()单步推进发送数据包时可被拦截用accept_packet(pkt)/reject_packet()决定放行、替换或拒绝见 automaton.py类方法spawn(port, ...)可直接把一个自动机包装成 TCP 服务器为每个连接启动一个实例支持bgTrue后台模式见 automaton.py。进阶ioevent 与 SuperSocket 封装除了收包与超时自动机还支持自定义 IO 事件。ATMT.ioevent(state, name, prio0, as_supersocketNone)可以把任意 fd如本地管道接入状态机事件到达时触发对应转移as_supersocket参数可以把整个自动机暴露为一个 ScapySuperSocket_ATMT_supersocket见 automaton.py实现自动机即 socket的用法。仓库里最典型的实践是 TLS 自动机TLSClientAutomaton在初始化时逐个状态地完成 SSLv2 到 TLS 1.3 的握手见 automaton_cli.py。其文档字符串展示的 SuperSocket 用法就是ioeventas_supersocket的产物from scapy.layers.tls import * a TLSClientAutomaton.tlslink(Raw, serverscapy.net, dport443) a.send(HTTP()/HTTPRequest()) while True: a.recv()也可以配合TCPSession直接做 HTTPS 请求from scapy.all import * from scapy.layers.http import * from scapy.layers.tls.automaton_cli import * a TLSClientAutomaton.tlslink(HTTP, serverwww.google.com, dport443) pkt a.sr1(HTTP()/HTTPRequest(), sessionTCPSession(appTrue), timeout2)这类以状态机实现的协议栈在仓库中并不少见除 TLS 外smbclient.py、smbserver.py、kerberos.py、http.py 以及 krack/automaton.py 都大量使用ATMT.state等装饰器。你可以在 test/scapy/automaton.uts 中看到框架自身的回归测试用例作为编写自定义自动机的参考。总结Scapy Automaton 把状态机抽象为一组带装饰器的普通方法用异常控制流实现状态切换用元类自动完成状态表的收集与排序配合跨平台的select事件循环让你能用极少样板代码实现带超时、重传、错误处理的完整网络协议逻辑。核心要点可以概括为状态ATMT.state(initial/final/stop/error1)用raise self.STATE_NAME(...)切换转移condition进入状态即检查、receive_condition收包检查、eof连接关闭、timeout一次性超时、timer自动重载超时用prio控制判定顺序动作ATMT.action(转移方法, prio0)绑定可多绑、可传参action_parameters()框架钩子覆写parse_args()传参、master_filter()集中过滤生命周期run()返回 final 状态结果restart()复位destroy()必须收尾graph()免费获得状态图timer_by_name()动态调整超时。掌握这套机制后无论是实现协议客户端、服务端还是构造中间人、模糊测试等攻击面工具你都能用声明式的方式快速落地一个健壮、可调试、可复用的状态机。赞分享网络网络安全【免费下载链接】scapyScapy: the Python-based interactive packet manipulation program library.项目地址https://gitcode.com/gh_mirrors/sc/scapy点击查看免费下载相关推荐Scapy Automaton 状态机实战如何自动化任意网络协议交互Scapy Automaton 状态机实战如何自动化任意网络协议交互 Scapy 是 Python 生态中最强大的交互式网络报文构造与分析库而它的 Auto网络网络安全NetExec协议状态机实现构建可靠网络工具NetExec协议状态机实现构建可靠网络工具 在网络自动化和渗透测试领域处理多种协议如SMB、SSH、RDP等的连接状态是确保工具稳定性和可靠性的核心挑网络安全渗透测试CLI应用安全RIOT 无状态 DHCPv6 客户端测试实战基于 scapy Automaton 的端到端协议验证RIOT 无状态 DHCPv6 客户端测试实战基于 scapy Automaton 的端到端协议验证 本篇文章深入剖析 RIOT 仓库中的 gnrc_dhcp物联网嵌入式操作系统实时系统创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
