scapy.automaton

Automata with states, transitions and actions.

class scapy.automaton.ATMT

Bases: object

ACTION = 'Action'

CONDITION = 'Condition'

IOEVENT = 'I/O event'

exception NewStateRequested(state_func: Any, automaton: ATMT, *args: Any, **kargs: Any)

Bases: Exception

action_parameters(*args: Any, **kargs: Any) → NewStateRequested

run() → Any

RECV = 'Receive condition'

STATE = 'State'

TIMEOUT = 'Timeout condition'

static action(cond: Any, prio: int = 0) → Callable[[_StateWrapper, _StateWrapper], _StateWrapper]

static condition(state: Any, prio: int = 0) → Callable[[_StateWrapper, _StateWrapper], _StateWrapper]

static ioevent(state: _StateWrapper, name: str, prio: int = 0, as_supersocket: str | None = None) → Callable[[_StateWrapper, _StateWrapper], _StateWrapper]

static receive_condition(state: _StateWrapper, prio: int = 0) → Callable[[_StateWrapper, _StateWrapper], _StateWrapper]

static state(initial: int = 0, final: int = 0, stop: int = 0, error: int = 0) → Callable[[DecoratorCallable], DecoratorCallable]

static timeout(state: _StateWrapper, timeout: int | float) → Callable[[_StateWrapper, _StateWrapper, Timer], _StateWrapper]

static timer(state: _StateWrapper, timeout: float | int, prio: int = 0) → Callable[[_StateWrapper, _StateWrapper, Timer], _StateWrapper]

class scapy.automaton.Automaton(self, debug=0, store=1, **kargs)

Bases: object

exception AutomatonError(msg: str, state: Message | None = None, result: str | None = None)

Bases: AutomatonException

exception AutomatonException(msg: str, state: Message | None = None, result: str | None = None)

Bases: Exception

exception AutomatonStopped(msg: str, state: Message | None = None, result: str | None = None)

Bases: AutomatonException

exception Breakpoint(msg: str, state: Message | None = None, result: str | None = None)

Bases: AutomatonStopped

exception CommandMessage(msg: str, state: Message | None = None, result: str | None = None)

Bases: AutomatonException

exception ErrorState(msg: str, state: Message | None = None, result: str | None = None)

Bases: AutomatonException

exception InterceptionPoint(msg: str, state: Message | None = None, result: str | None = None, packet: str | None = None)

Bases: AutomatonStopped

exception Singlestep(msg: str, state: Message | None = None, result: str | None = None)

Bases: AutomatonStopped

exception Stuck(msg: str, state: Message | None = None, result: str | None = None)

Bases: AutomatonException

accept_packet(pkt: Packet | None = None, wait: bool | None = False) → Any

actions: Dict[str, List[_StateWrapper]] = {}

add_breakpoints(*bps: Any) → None

add_interception_points(*ipts: Any) → None

conditions: Dict[str, List[_StateWrapper]] = {}

debug(lvl: int, msg: str) → None

destroy() → None

Destroys a stopped Automaton: this cleanups all opened file descriptors. Required on PyPy for instance where the garbage collector behaves differently.

forcestop(wait: bool = True) → None

initial_states: List[_StateWrapper] = []

ioevents: Dict[str, List[_StateWrapper]] = {}

ionames: List[str] = []

iosupersockets: List[SuperSocket] = []

master_filter(pkt: Packet) → bool

my_send(pkt: Packet) → None

parse_args(debug: int = 0, store: int = 1, **kargs: Any) → None

recv_conditions: Dict[str, List[_StateWrapper]] = {}

reject_packet(wait: bool | None = False) → Any

remove_breakpoints(*bps: Any) → None

remove_interception_points(*ipts: Any) → None

restart(*args: Any, **kargs: Any) → None

run(resume: Message | None = None, wait: bool | None = True) → Any

runbg(resume: Message | None = None, wait: bool | None = False) → None

send(pkt: Packet) → None

start(*args: Any, **kargs: Any) → None

state: NewStateRequested = None

states: Dict[str, _StateWrapper] = {}

stop(wait: bool = True) → None

stop_states: List[_StateWrapper] = []

timeout: Dict[str, _TimerList] = {}

timer_by_name(name: str) → Timer | None

class scapy.automaton.Automaton_metaclass(name: str, bases: Tuple[Any], dct: Dict[str, Any])

Bases: type

build_graph() → str

graph(**kargs: Any) → str | None

class scapy.automaton.Message(**args: Any)

Bases: object

exc_info: Tuple[None, None, None] | Tuple[BaseException, Exception, TracebackType] = None

pkt: Packet = None

result: str = None

state: Message = None

type: str = None

class scapy.automaton.ObjectPipe(name: str | None = None)

Bases: Generic[_T]

clear() → None

close() → None

empty() → bool

fileno() → int

flush() → None

read(n: int | None = 0) → _T | None

recv(n: int | None = 0) → _T | None

static select(sockets: List[SuperSocket], remain: float | None = 0.05) → List[SuperSocket]

send(obj: _T) → int

write(obj: _T) → None

class scapy.automaton.Timer(time: int | float, prio: int = 0, autoreload: bool = False)

Bases: object

get() → float

set(val: float) → None

scapy.automaton.select_objects(inputs: Iterable[Any], remain: float | int | None) → List[Any]

Select objects. Same than: select.select(inputs, [], [], remain)

But also works on Windows, only on objects whose fileno() returns a Windows event. For simplicity, just use ObjectPipe() as a queue that you can select on whatever the platform is.

If you want an object to be always included in the output of select_objects (i.e. it’s not selectable), just make fileno() return a strictly negative value.

Example

>>> a, b = ObjectPipe("a"), ObjectPipe("b")
>>> b.send("test")
>>> select_objects([a, b], 1)
[b]

Parameters:

• inputs – objects to process

• remain – timeout. If 0, return [].