| Server IP : 91.108.119.165 / Your IP : 216.73.217.1 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/alt/python311/lib/python3.11/site-packages/pyroute2/ndb/objects/ |
Upload File : |
'''
Run a network probe
===================
A successful network probe neither creates real network objects
like interfaces or addresses, nor database records. The only
important thing it does -- it raises no exception.
On the contrary, an unsuccessful network probe raises a
`NetlinkError` exception, cancelling the whole transaction.
A network probe is always run from the corresponding netlink
target: a local system, a remote system, a network namespace,
a container.
An example scenario:
* target alpha, set up eth0 10.0.0.2/24
* target beta, set up eth0 10.0.0.4/24
* ping 10.0.0.4 (beta) from target alpha
* ping 10.0.0.2 (alpha) from target beta
The code below sets up the addresses and checks ICMP responses. If
any step fails, the whole transaction will be rolled back automatically::
with NDB(log='debug') as ndb:
ndb.sources.add(kind='remote', hostname='alpha', username='root')
ndb.sources.add(kind='remote', hostname='beta', username='root')
with ndb.begin() as trx:
trx.push(
(ndb
.interfaces[{'target': 'alpha', 'ifname': 'eth0'}]
.set(state='up')
.add_ip(address='10.0.0.2', prefixlen=24)
),
(ndb
.interfaces[{'target': 'beta', 'ifname': 'eth0'}]
.set(state='up')
.add_ip(address='10.0.0.4', prefixlen=24)
),
(ndb
.probes
.create(target='alpha', kind='ping', dst='10.0.0.4')
),
(ndb
.probes
.create(target='beta', kind='ping', dst='10.0.0.2')
),
)
'''
from pyroute2.netlink.rtnl.probe_msg import probe_msg
from ..objects import RTNL_Object
def load_probe_msg(schema, target, event):
pass
schema = probe_msg.sql_schema().unique_index()
init = {
'specs': [['probes', schema]],
'classes': [['probes', probe_msg]],
'event_map': {probe_msg: [load_probe_msg]},
}
class Probe(RTNL_Object):
table = 'probes'
msg_class = probe_msg
api = 'probe'
def __init__(self, *argv, **kwarg):
kwarg['iclass'] = probe_msg
self.event_map = {probe_msg: 'load_probe_msg'}
super().__init__(*argv, **kwarg)
def check(self):
return True