all: reformat python files with python black

Part of !537.

https://gitlab.freedesktop.org/NetworkManager/NetworkManager/-/merge_requests/537
This commit is contained in:
Sayed Shah
2020-06-09 16:28:32 -04:00
committed by Thomas Haller
parent b6febb0fd0
commit be822b52e6
44 changed files with 3179 additions and 1961 deletions

View File

@@ -14,15 +14,18 @@
#
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
import sys, uuid
main_loop = None
def print_values(setting, key, value, flags, data):
print(" %s.%s: %s" % (setting.get_name(), key, value))
# create an Ethernet connection and return it
def create_profile(name):
profile = NM.SimpleConnection.new()
@@ -49,6 +52,7 @@ def create_profile(name):
return profile
# callback function
def added_cb(client, result, data):
try:
@@ -58,16 +62,17 @@ def added_cb(client, result, data):
sys.stderr.write("Error: %s\n" % e)
main_loop.quit()
if __name__ == "__main__":
# parse arguments
persistent = False
if len(sys.argv) != 2 and len(sys.argv) != 3:
sys.exit('Usage: %s <connection name> [persistent]' % sys.argv[0])
sys.exit("Usage: %s <connection name> [persistent]" % sys.argv[0])
if len(sys.argv) == 3:
if sys.argv[2] in "persistent" and sys.argv[2][:1] == "p":
persistent = True
else:
sys.exit('Usage: %s <connection name> [persistent]' % sys.argv[0])
sys.exit("Usage: %s <connection name> [persistent]" % sys.argv[0])
profile_name = sys.argv[1]
main_loop = GLib.MainLoop()
@@ -82,4 +87,3 @@ if __name__ == "__main__":
client.add_connection_async(con, persistent, None, added_cb, None)
main_loop.run()

View File

@@ -7,32 +7,52 @@
import sys
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
###############################################################################
def usage():
print("Usage: %s [COMMAND [ARG]...]" % sys.argv[0])
print("")
print(" COMMANDS: [show]")
print(" create TIMEOUT [--destroy-all|--delete-new-connections|--disconnect-new-devices|--allow-overlapping|DEV]...")
print(
" create TIMEOUT [--destroy-all|--delete-new-connections|--disconnect-new-devices|--allow-overlapping|DEV]..."
)
print(" destroy PATH|NUMBER")
print(" rollback PATH|NUMBER")
print(" adjust-rollback-timeout PATH|NUMBER TIMEOUT")
print("")
sys.exit(1)
def show(c, ts = None):
def show(c, ts=None):
cr = c.get_created()
rt = c.get_rollback_timeout()
print("%s:" % c.get_path())
print(" created: %u%s" % (cr, "" if ts is None else (" (%s sec ago)" % ((ts - cr) / 1000.0))))
print(
" created: %u%s"
% (cr, "" if ts is None else (" (%s sec ago)" % ((ts - cr) / 1000.0)))
)
if rt == 0:
print(" timeout: infinity")
else:
print(" timeout: %u seconds%s" % (rt, "" if ts is None else (" (circa %s sec left)" % ((cr + (rt * 1000) - ts) / 1000.0))))
print(" devices: %s" % (' '.join(sorted(map(lambda x: x.get_iface(), c.get_devices())))))
print(
" timeout: %u seconds%s"
% (
rt,
""
if ts is None
else (" (circa %s sec left)" % ((cr + (rt * 1000) - ts) / 1000.0)),
)
)
print(
" devices: %s"
% (" ".join(sorted(map(lambda x: x.get_iface(), c.get_devices()))))
)
def find_checkpoint(client, path):
for c in client.get_checkpoints():
@@ -40,6 +60,7 @@ def find_checkpoint(client, path):
return c
return None
def validate_path(path, client):
try:
num = int(path)
@@ -47,8 +68,8 @@ def validate_path(path, client):
except Exception as e:
pass
if not path or path[0] != '/':
sys.exit('Invalid checkpoint path \"%s\"' % (path))
if not path or path[0] != "/":
sys.exit('Invalid checkpoint path "%s"' % (path))
if client is not None:
checkpoint = find_checkpoint(client, path)
@@ -57,6 +78,7 @@ def validate_path(path, client):
return path
def do_create(client):
flags = NM.CheckpointCreateFlags.NONE
if len(sys.argv) < 3:
@@ -65,13 +87,13 @@ def do_create(client):
timeout = int(sys.argv[2])
devices = []
for arg in sys.argv[3:]:
if arg == '--destroy-all':
if arg == "--destroy-all":
flags |= NM.CheckpointCreateFlags.DESTROY_ALL
elif arg == '--delete-new-connections':
elif arg == "--delete-new-connections":
flags |= NM.CheckpointCreateFlags.DELETE_NEW_CONNECTIONS
elif arg == '--disconnect-new-devices':
elif arg == "--disconnect-new-devices":
flags |= NM.CheckpointCreateFlags.DISCONNECT_NEW_DEVICES
elif arg == '--allow-overlapping':
elif arg == "--allow-overlapping":
flags |= NM.CheckpointCreateFlags.ALLOW_OVERLAPPING
else:
d = client.get_device_by_iface(arg)
@@ -89,6 +111,7 @@ def do_create(client):
client.checkpoint_create(devices, timeout, flags, None, create_cb, None)
def do_destroy(client):
if len(sys.argv) < 3:
sys.exit("Missing checkpoint path")
@@ -105,6 +128,7 @@ def do_destroy(client):
client.checkpoint_destroy(path, None, destroy_cb, None)
def do_rollback(client):
if len(sys.argv) < 3:
sys.exit("Missing checkpoint path")
@@ -127,6 +151,7 @@ def do_rollback(client):
client.checkpoint_rollback(path, None, rollback_cb, None)
def do_adjust_rollback_timeout(client):
if len(sys.argv) < 3:
sys.exit("Missing checkpoint path")
@@ -147,27 +172,31 @@ def do_adjust_rollback_timeout(client):
sys.stderr.write("Failed: %s\n" % e.message)
main_loop.quit()
client.checkpoint_adjust_rollback_timeout(path, add_timeout, None, adjust_rollback_timeout_cb, None)
client.checkpoint_adjust_rollback_timeout(
path, add_timeout, None, adjust_rollback_timeout_cb, None
)
def do_show(client):
ts = NM.utils_get_timestamp_msec()
for c in client.get_checkpoints():
show(c, ts)
if __name__ == '__main__':
if __name__ == "__main__":
nm_client = NM.Client.new(None)
main_loop = GLib.MainLoop()
if len(sys.argv) < 2 or sys.argv[1] == 'show':
if len(sys.argv) < 2 or sys.argv[1] == "show":
do_show(nm_client)
sys.exit(0)
elif sys.argv[1] == 'create':
elif sys.argv[1] == "create":
do_create(nm_client)
elif sys.argv[1] == 'destroy':
elif sys.argv[1] == "destroy":
do_destroy(nm_client)
elif sys.argv[1] == 'rollback':
elif sys.argv[1] == "rollback":
do_rollback(nm_client)
elif sys.argv[1] == 'adjust-rollback-timeout':
elif sys.argv[1] == "adjust-rollback-timeout":
do_adjust_rollback_timeout(nm_client)
else:
usage()

View File

@@ -16,24 +16,25 @@
import sys
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
# supported connection types
connection_types = {
NM.SETTING_VPN_SETTING_NAME,
NM.SETTING_WIRELESS_SETTING_NAME,
NM.SETTING_WIRED_SETTING_NAME,
NM.SETTING_BOND_SETTING_NAME,
NM.SETTING_BRIDGE_SETTING_NAME,
NM.SETTING_TEAM_SETTING_NAME,
NM.SETTING_INFINIBAND_SETTING_NAME,
NM.SETTING_PPPOE_SETTING_NAME,
NM.SETTING_ADSL_SETTING_NAME,
NM.SETTING_BLUETOOTH_SETTING_NAME,
NM.SETTING_WIMAX_SETTING_NAME,
NM.SETTING_OLPC_MESH_SETTING_NAME,
NM.SETTING_GENERIC_SETTING_NAME,
NM.SETTING_VPN_SETTING_NAME,
NM.SETTING_WIRELESS_SETTING_NAME,
NM.SETTING_WIRED_SETTING_NAME,
NM.SETTING_BOND_SETTING_NAME,
NM.SETTING_BRIDGE_SETTING_NAME,
NM.SETTING_TEAM_SETTING_NAME,
NM.SETTING_INFINIBAND_SETTING_NAME,
NM.SETTING_PPPOE_SETTING_NAME,
NM.SETTING_ADSL_SETTING_NAME,
NM.SETTING_BLUETOOTH_SETTING_NAME,
NM.SETTING_WIMAX_SETTING_NAME,
NM.SETTING_OLPC_MESH_SETTING_NAME,
NM.SETTING_GENERIC_SETTING_NAME,
}
@@ -45,9 +46,14 @@ if __name__ == "__main__":
if len(sys.argv) == 2:
ctype = sys.argv[1]
if ctype not in connection_types:
sys.exit('Usage: %s [<type>]\nAllowed types: %s' % (sys.argv[0], allowed_types))
sys.exit(
"Usage: %s [<type>]\nAllowed types: %s"
% (sys.argv[0], allowed_types)
)
else:
sys.exit('Usage: %s [<type>]\nAllowed types: %s' % (sys.argv[0], allowed_types))
sys.exit(
"Usage: %s [<type>]\nAllowed types: %s" % (sys.argv[0], allowed_types)
)
# create Client object
client = NM.Client.new(None)
@@ -63,5 +69,4 @@ if __name__ == "__main__":
client.deactivate_connection(ac, None)
sys.stdout.write("\033[32m -> succeeded\033[0m\n")
except Exception as e:
sys.stderr.write("\033[31m -> failed\033[0m (%s)\n" % e.message)
sys.stderr.write("\033[31m -> failed\033[0m (%s)\n" % e.message)

View File

@@ -6,7 +6,8 @@
import sys
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
#
@@ -18,6 +19,7 @@ from gi.repository import GLib, NM
main_loop = None
def do_notify(self, property):
print("notify: %s" % property)
ip4cfg = self.get_ip4_config()
@@ -25,26 +27,26 @@ def do_notify(self, property):
print("ip4-config: %s" % ip4cfg.get_path())
main_loop.quit()
def state_changed(obj, arg1, arg2, arg3):
print("State changed: New: %d, Old: %d, Reason: %d" % (arg1, arg2, arg3))
# Device is connected
if arg1 == 100:
obj.connect('notify::ip4-config', do_notify)
obj.connect("notify::ip4-config", do_notify)
if __name__ == "__main__":
if len(sys.argv) != 2:
sys.exit('Usage: %s <interface>' % sys.argv[0])
sys.exit("Usage: %s <interface>" % sys.argv[0])
dev_iface = sys.argv[1]
c = NM.Client.new(None)
dev = c.get_device_by_iface(dev_iface)
if dev is None:
sys.exit('Device \'%s\' not found' % dev_iface)
sys.exit("Device '%s' not found" % dev_iface)
print("Device: %s - %s" % (dev_iface, dev.get_device_type().value_name))
print("---------------------------------------")
dev.connect('state-changed', state_changed)
dev.connect("state-changed", state_changed)
main_loop = GLib.MainLoop()
main_loop.run()

View File

@@ -5,24 +5,26 @@
#
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
# This example shows how to monitor the DNS configuration
main_loop = None
def handle_config(config):
print " ---- new configuration ----"
for entry in config:
print " * servers: %s" % ', '.join(map(str, entry.get_nameservers()))
print " * servers: %s" % ", ".join(map(str, entry.get_nameservers()))
domains = entry.get_domains()
if domains and domains[0]:
print " domains: %s" % ', '.join(map(str, domains))
print " domains: %s" % ", ".join(map(str, domains))
if entry.get_interface():
print " interface: %s" % entry.get_interface()
print " interface: %s" % entry.get_interface()
print " priority: %d" % entry.get_priority()
@@ -31,9 +33,11 @@ def handle_config(config):
print ""
def dns_config_changed(self, property):
handle_config(self.get_dns_configuration())
main_loop = None
if __name__ == "__main__":

View File

@@ -6,7 +6,8 @@
import sys
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
#
@@ -26,13 +27,15 @@ from gi.repository import GLib, NM
main_loop = None
def connection_saved(connection, error, data):
print ("Connection '%s' saved.") % (connection.get_id())
print("Connection '%s' saved.") % (connection.get_id())
main_loop.quit()
if __name__ == "__main__":
if len(sys.argv) != 2 and len(sys.argv) != 3:
sys.exit('Usage: %s <connection name or UUID> [new zone]' % sys.argv[0])
sys.exit("Usage: %s <connection name or UUID> [new zone]" % sys.argv[0])
main_loop = GLib.MainLoop()
client = NM.Client.new(None)

View File

@@ -7,7 +7,8 @@
# This example lists currently active connections
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
if __name__ == "__main__":
@@ -16,6 +17,4 @@ if __name__ == "__main__":
for ac in acons:
print("%s (%s) - %s" % (ac.get_id(), ac.get_uuid(), ac.get_connection_type()))
if len(acons) == 0:
print("No active connections")
print("No active connections")

View File

@@ -7,7 +7,8 @@
# This example lists all devices, both real and placeholder ones
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
if __name__ == "__main__":
@@ -18,11 +19,18 @@ if __name__ == "__main__":
print "------------"
for d in devices:
if d.is_real():
print "%s (%s): %s" % (d.get_iface(), d.get_type_description(), d.get_state())
print "%s (%s): %s" % (
d.get_iface(),
d.get_type_description(),
d.get_state(),
)
print "\nUnrealized/placeholder devices"
print "------------------------------"
for d in devices:
if not d.is_real():
print "%s (%s): %s" % (d.get_iface(), d.get_type_description(), d.get_state())
print "%s (%s): %s" % (
d.get_iface(),
d.get_type_description(),
d.get_state(),
)

View File

@@ -5,7 +5,8 @@
#
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
if __name__ == "__main__":
@@ -13,7 +14,10 @@ if __name__ == "__main__":
devices = client.get_devices()
for d in devices:
print("{:<16} {:<16} {}".format(d.get_iface(),
"(" + d.get_type_description() + ")",
NM.utils_enum_to_str(NM.DeviceInterfaceFlags,
d.get_interface_flags())))
print(
"{:<16} {:<16} {}".format(
d.get_iface(),
"(" + d.get_type_description() + ")",
NM.utils_enum_to_str(NM.DeviceInterfaceFlags, d.get_interface_flags()),
)
)

View File

@@ -6,7 +6,8 @@
import sys
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
# This example shows how to get a list of LLDP neighbors for a given interface.
@@ -15,24 +16,24 @@ main_loop = None
if __name__ == "__main__":
if len(sys.argv) != 2:
sys.exit('Usage: %s <interface>' % sys.argv[0])
sys.exit("Usage: %s <interface>" % sys.argv[0])
dev_iface = sys.argv[1]
c = NM.Client.new(None)
dev = c.get_device_by_iface(dev_iface)
if dev is None:
sys.exit('Device \'%s\' not found' % dev_iface)
sys.exit("Device '%s' not found" % dev_iface)
neighbors = dev.get_lldp_neighbors()
for neighbor in neighbors:
ret, chassis = neighbor.get_attr_string_value('chassis-id')
ret, port = neighbor.get_attr_string_value('port-id')
ret, chassis = neighbor.get_attr_string_value("chassis-id")
ret, port = neighbor.get_attr_string_value("port-id")
print "Neighbor: %s - %s" % (chassis, port)
for attr in neighbor.get_attr_names():
attr_type = neighbor.get_attr_type(attr)
if attr_type.equal(GLib.VariantType.new('s')):
if attr_type.equal(GLib.VariantType.new("s")):
ret, value = neighbor.get_attr_string_value(attr)
print " %-32s: %s" % (attr, value)
elif attr_type.equal(GLib.VariantType.new('u')):
elif attr_type.equal(GLib.VariantType.new("u")):
ret, value = neighbor.get_attr_uint_value(attr)
print " %-32s: %u" % (attr, value)

View File

@@ -6,7 +6,8 @@
import sys, socket
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
#
@@ -14,11 +15,12 @@ from gi.repository import NM
# from NMIP4Config and NMIP6Config (got out of NMDevice)
#
def show_addresses(dev, family):
if (family == socket.AF_INET):
ip_cfg = dev.get_ip4_config()
if family == socket.AF_INET:
ip_cfg = dev.get_ip4_config()
else:
ip_cfg = dev.get_ip6_config()
ip_cfg = dev.get_ip6_config()
if ip_cfg is None:
print("None")
@@ -35,8 +37,9 @@ def show_addresses(dev, family):
print("%s/%d") % (addr, prefix)
def show_gateway(dev, family):
if (family == socket.AF_INET):
if family == socket.AF_INET:
ip_cfg = dev.get_ip4_config()
else:
ip_cfg = dev.get_ip6_config()
@@ -45,16 +48,17 @@ def show_gateway(dev, family):
gw = "None"
else:
gw = ip_cfg.get_gateway()
if gw == '':
if gw == "":
gw = "None"
print(gw)
def show_routes(dev, family):
if (family == socket.AF_INET):
ip_cfg = dev.get_ip4_config()
if family == socket.AF_INET:
ip_cfg = dev.get_ip4_config()
else:
ip_cfg = dev.get_ip6_config()
ip_cfg = dev.get_ip6_config()
if ip_cfg is None:
print("None")
@@ -75,31 +79,31 @@ def show_routes(dev, family):
def show_dns(dev, family):
if (family == socket.AF_INET):
ip_cfg = dev.get_ip4_config()
if family == socket.AF_INET:
ip_cfg = dev.get_ip4_config()
else:
ip_cfg = dev.get_ip6_config()
ip_cfg = dev.get_ip6_config()
if ip_cfg is None:
print("None")
return
print ("Nameservers: %s") % (ip_cfg.get_nameservers())
print ("Domains: %s") % (ip_cfg.get_domains())
print ("Searches: %s") % (ip_cfg.get_searches())
if (family == socket.AF_INET):
print ("WINS: %s") % (ip_cfg.get_wins_servers())
print("Nameservers: %s") % (ip_cfg.get_nameservers())
print("Domains: %s") % (ip_cfg.get_domains())
print("Searches: %s") % (ip_cfg.get_searches())
if family == socket.AF_INET:
print("WINS: %s") % (ip_cfg.get_wins_servers())
if __name__ == "__main__":
if len(sys.argv) != 2:
sys.exit('Usage: %s <interface>' % sys.argv[0])
sys.exit("Usage: %s <interface>" % sys.argv[0])
dev_iface = sys.argv[1]
c = NM.Client.new(None)
dev = c.get_device_by_iface(dev_iface)
if dev is None:
sys.exit('Device \'%s\' not found' % dev_iface)
sys.exit("Device '%s' not found" % dev_iface)
print("Device: %s - %s" % (dev_iface, dev.get_device_type().value_name))
print("---------------------------------------")
@@ -142,4 +146,3 @@ if __name__ == "__main__":
print("------------")
show_dns(dev, socket.AF_INET6)
print

View File

@@ -5,14 +5,17 @@
#
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
# This example asks settings service for all configured connections.
def print_values(setting, key, value, flags, data):
print(" %s.%s: %s" % (setting.get_name(), key, value))
if __name__ == "__main__":
# create Client object
client = NM.Client.new(None)
@@ -25,4 +28,3 @@ if __name__ == "__main__":
print("=== %s : %s ===" % (c.get_id(), c.get_path()))
c.for_each_setting_value(print_values, None)
print("\n")

View File

@@ -7,40 +7,47 @@
import sys
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
def find_connections(nm_client, arg_type, arg_id):
for c in nm_client.get_connections():
if arg_type in [None, 'id'] and c.get_id() == arg_id:
if arg_type in [None, "id"] and c.get_id() == arg_id:
yield c
if arg_type in [None, 'uuid'] and c.get_uuid() == arg_id:
if arg_type in [None, "uuid"] and c.get_uuid() == arg_id:
yield c
def find_connection_first(nm_client, arg_type, arg_id):
for f in find_connections(nm_client, arg_type, arg_id):
return f
def con_to_str(con):
s_con = con.get_setting_connection()
return '"%s" (%s)' % (s_con.get_id(), s_con.get_uuid())
def usage():
arg0 = sys.argv[0]
arg0_spaced = ' ' * len(arg0)
print('Usage: %s [ --clone ( [id] <id> | [uuid] <uuid> ) ] \\' % (arg0))
print(' %s [ --to-disk | --in-memory ] \\' % (arg0_spaced))
print(' %s [ --block-autoconnect ] \\' % (arg0_spaced))
print(' %s [ --id <new-id> ] \\' % (arg0_spaced))
print(' %s [ --uuid <new-uuid> ] \\' % (arg0_spaced))
arg0_spaced = " " * len(arg0)
print("Usage: %s [ --clone ( [id] <id> | [uuid] <uuid> ) ] \\" % (arg0))
print(" %s [ --to-disk | --in-memory ] \\" % (arg0_spaced))
print(" %s [ --block-autoconnect ] \\" % (arg0_spaced))
print(" %s [ --id <new-id> ] \\" % (arg0_spaced))
print(" %s [ --uuid <new-uuid> ] \\" % (arg0_spaced))
return 1
def die(msg, print_usage=False):
print(msg)
if print_usage:
usage()
sys.exit(1)
def main():
main_loop = GLib.MainLoop()
@@ -56,56 +63,62 @@ def main():
argv = list(sys.argv[1:])
while argv:
if argv[0] == '--clone':
if argv[0] == "--clone":
match_type = None
if len(argv) < 2:
die('missing argument for --clone option')
if argv[0] in ['id', 'uuid']:
die("missing argument for --clone option")
if argv[0] in ["id", "uuid"]:
match_type = argv[0]
if len(argv) < 3:
die('missing argument for "--clone %s" option' % (match_type))
argv = argv[1:]
if cons:
die('cannot specify --clone argument more than once')
die("cannot specify --clone argument more than once")
cons.extend(find_connections(nm_client, match_type, argv[1]))
if len(cons) == 0:
die('could not find connection for "--clone %s%s"' % ((match_type or ''), argv[1]))
die(
'could not find connection for "--clone %s%s"'
% ((match_type or ""), argv[1])
)
if len(cons) != 1:
die('could not find unique connection for "--clone %s%s"' % ((match_type or ''), argv[1]))
die(
'could not find unique connection for "--clone %s%s"'
% ((match_type or ""), argv[1])
)
argv = argv[2:]
continue
if argv[0] in ['--block-autoconnect']:
if argv[0] in ["--block-autoconnect"]:
arg_block_autoconnect = NM.SettingsAddConnection2Flags.BLOCK_AUTOCONNECT
argv = argv[1:]
continue
if argv[0] in ['--to-disk', '--in-memory']:
if argv[0] == '--to-disk':
if argv[0] in ["--to-disk", "--in-memory"]:
if argv[0] == "--to-disk":
v = NM.SettingsAddConnection2Flags.TO_DISK
elif argv[0] == '--in-memory':
elif argv[0] == "--in-memory":
v = NM.SettingsAddConnection2Flags.IN_MEMORY
else:
assert(False)
assert False
if arg_mode is not None:
die('duplicate storage modes ("%s")' % (argv[0]))
arg_mode = v
argv = argv[1:]
continue
if argv[0] in ['--id']:
if argv[0] in ["--id"]:
if len(argv) < 2:
die('missing argument for --id option')
die("missing argument for --id option")
arg_id = argv[1]
argv = argv[2:]
continue
if argv[0] in ['--uuid']:
if argv[0] in ["--uuid"]:
if len(argv) < 2:
die('missing argument for --uuid option')
die("missing argument for --uuid option")
arg_uuid = argv[1]
argv = argv[2:]
continue
die('unknown argument "%s"' % (argv[0]))
if len(cons) != 1:
die('missing --clone argument', True)
die("missing --clone argument", True)
con = cons[0]
@@ -117,31 +130,46 @@ def main():
s_con.set_property(NM.SETTING_CONNECTION_UUID, arg_uuid or NM.utils_uuid_generate())
result = {}
def _add_connection2_cb(cl, async_result, user_data):
try:
c, r = nm_client.add_connection2_finish(async_result)
except Exception as e:
result['error'] = e
result["error"] = e
else:
result['result'] = r
result['connection'] = c
result["result"] = r
result["connection"] = c
main_loop.quit()
nm_client.add_connection2(con2.to_dbus(NM.ConnectionSerializationFlags.ALL),
(arg_mode if arg_mode is not None else NM.SettingsAddConnection2Flags.TO_DISK)
| arg_block_autoconnect,
None,
False,
None,
_add_connection2_cb,
None)
nm_client.add_connection2(
con2.to_dbus(NM.ConnectionSerializationFlags.ALL),
(arg_mode if arg_mode is not None else NM.SettingsAddConnection2Flags.TO_DISK)
| arg_block_autoconnect,
None,
False,
None,
_add_connection2_cb,
None,
)
main_loop.run()
if 'error' in result:
die('update connection %s failed [%s]: %s' % (con_to_str(con2), ' '.join(sys.argv), result['error']))
if "error" in result:
die(
"update connection %s failed [%s]: %s"
% (con_to_str(con2), " ".join(sys.argv), result["error"])
)
print('update connection %s succeeded [%s]: %s, %s' % (con_to_str(con2), ' '.join(sys.argv), result['connection'].get_path(), result['result']))
print(
"update connection %s succeeded [%s]: %s, %s"
% (
con_to_str(con2),
" ".join(sys.argv),
result["connection"].get_path(),
result["result"],
)
)
if __name__ == '__main__':
if __name__ == "__main__":
main()

View File

@@ -11,33 +11,37 @@ import sys
import re
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
def usage():
print('Usage: %s [[id] <id>]' % (sys.argv[0]))
print(' %s [[uuid] <uuid>]' % (sys.argv[0]))
print("Usage: %s [[id] <id>]" % (sys.argv[0]))
print(" %s [[uuid] <uuid>]" % (sys.argv[0]))
return 1
def find_connection(nm_client, arg_type, arg_id):
for c in nm_client.get_connections():
if arg_type in [None, 'id'] and c.get_id() == arg_id:
if arg_type in [None, "id"] and c.get_id() == arg_id:
return c
if arg_type in [None, 'uuid'] and c.get_uuid() == arg_id:
if arg_type in [None, "uuid"] and c.get_uuid() == arg_id:
return c
def main():
if len(sys.argv) < 2 or len(sys.argv) > 3:
return usage()
if len(sys.argv) == 3:
arg_type = sys.argv[1]
arg_id = sys.argv[2]
if arg_type not in ['id', 'uuid']:
if arg_type not in ["id", "uuid"]:
return usage()
else:
arg_type = None
arg_id = sys.argv[1]
arg_log = '%s"%s"' % ((' with %s ' % (arg_type)) if arg_type else '', arg_id)
arg_log = '%s"%s"' % ((" with %s " % (arg_type)) if arg_type else "", arg_id)
main_loop = GLib.MainLoop()
@@ -45,57 +49,61 @@ def main():
con = find_connection(nm_client, arg_type, arg_id)
if con is None:
print('could not find a connection %s' % (arg_log))
print("could not find a connection %s" % (arg_log))
return 1
s_con = con.get_setting_connection()
if s_con is None:
print('connection %s has no [connection] setting' % (arg_log))
print("connection %s has no [connection] setting" % (arg_log))
return 1
arg_log = '"%s" (%s)' % (s_con.get_id(), s_con.get_uuid())
stable_id = s_con.get_stable_id()
if not stable_id:
print('connection %s has no stable-id set' % (arg_log))
print("connection %s has no stable-id set" % (arg_log))
return 1
re_match = re.search('\A(.*)-([0-9]+)\Z', stable_id)
re_match = re.search("\A(.*)-([0-9]+)\Z", stable_id)
if not re_match:
stable_id = stable_id + '-1'
stable_id = stable_id + "-1"
else:
stable_id = re_match.group(1) + '-' + str(int(re_match.group(2)) + 1)
stable_id = re_match.group(1) + "-" + str(int(re_match.group(2)) + 1)
con2 = NM.SimpleConnection.new_clone(con)
s_con = con2.get_setting_connection()
s_con.set_property(NM.SETTING_CONNECTION_STABLE_ID, stable_id)
result = {}
def _update2_cb(con, async_result, user_data):
try:
r = con.update2_finish(async_result)
except Exception as e:
result['error'] = e
result["error"] = e
else:
result['result'] = r
result["result"] = r
main_loop.quit()
con.update2(con2.to_dbus(NM.ConnectionSerializationFlags.ALL),
NM.SettingsUpdate2Flags.BLOCK_AUTOCONNECT,
None,
None,
_update2_cb,
None)
con.update2(
con2.to_dbus(NM.ConnectionSerializationFlags.ALL),
NM.SettingsUpdate2Flags.BLOCK_AUTOCONNECT,
None,
None,
_update2_cb,
None,
)
main_loop.run()
if 'error' in result:
print('update connection %s failed: %s' % (arg_log, result['error']))
if "error" in result:
print("update connection %s failed: %s" % (arg_log, result["error"]))
return 1
print('update connection %s succeeded: %s' % (arg_log, result['result']))
print("update connection %s succeeded: %s" % (arg_log, result["result"]))
print('set stable-id to "%s"' % (stable_id))
return 0
if __name__ == '__main__':
if __name__ == "__main__":
sys.exit(main())

View File

@@ -7,40 +7,50 @@
import sys
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
def find_connections(nm_client, arg_type, arg_id):
for c in nm_client.get_connections():
if arg_type in [None, 'id'] and c.get_id() == arg_id:
if arg_type in [None, "id"] and c.get_id() == arg_id:
yield c
if arg_type in [None, 'uuid'] and c.get_uuid() == arg_id:
if arg_type in [None, "uuid"] and c.get_uuid() == arg_id:
yield c
def find_connection_first(nm_client, arg_type, arg_id):
for f in find_connections(nm_client, arg_type, arg_id):
return f
def con_to_str(con):
s_con = con.get_setting_connection()
return '"%s" (%s)' % (s_con.get_id(), s_con.get_uuid())
def usage():
arg0 = sys.argv[0]
arg0_spaced = ' ' * len(arg0)
print('Usage: %s [ [id] <id> | [uuid] <uuid> ] \\' % (arg0))
print(' %s [ --to-disk | --in-memory | --in-memory-detached | --in-memory-only ] \\' % (arg0_spaced))
print(' %s [ --block-autoconnect ] \\' % (arg0_spaced))
print(' %s [ --volatile ] \\' % (arg0_spaced))
print(' %s [ --no-reapply ] \\' % (arg0_spaced))
arg0_spaced = " " * len(arg0)
print("Usage: %s [ [id] <id> | [uuid] <uuid> ] \\" % (arg0))
print(
" %s [ --to-disk | --in-memory | --in-memory-detached | --in-memory-only ] \\"
% (arg0_spaced)
)
print(" %s [ --block-autoconnect ] \\" % (arg0_spaced))
print(" %s [ --volatile ] \\" % (arg0_spaced))
print(" %s [ --no-reapply ] \\" % (arg0_spaced))
return 1
def die(msg, print_usage=False):
print(msg)
if print_usage:
usage()
sys.exit(1)
def main():
main_loop = GLib.MainLoop()
@@ -56,9 +66,9 @@ def main():
argv = list(sys.argv[1:])
while argv:
if argv[0] in ['id', 'uuid']:
if argv[0] in ["id", "uuid"]:
if cons:
die('cannot specify multiple connections')
die("cannot specify multiple connections")
if len(argv) < 2:
die('missing argument for "%s" specifier' % (argv[0]))
cons.extend(find_connections(nm_client, argv[0], argv[1]))
@@ -68,31 +78,36 @@ def main():
die('could not find unique connection for "%s %s"' % (argv[0], argv[1]))
argv = argv[2:]
continue
if argv[0] in ['--block-autoconnect']:
if argv[0] in ["--block-autoconnect"]:
arg_block_autoconnect = NM.SettingsUpdate2Flags.BLOCK_AUTOCONNECT
argv = argv[1:]
continue
if argv[0] in ['--volatile']:
if argv[0] in ["--volatile"]:
arg_volatile = NM.SettingsUpdate2Flags.VOLATILE
argv = argv[1:]
continue
if argv[0] in ['--no-reapply']:
if argv[0] in ["--no-reapply"]:
arg_no_reapply = NM.SettingsUpdate2Flags.NO_REAPPLY
argv = argv[1:]
continue
if argv[0] in ['--to-disk', '--in-memory', '--in-memory-detached', '--in-memory-only']:
if argv[0] == '--to-disk':
if argv[0] in [
"--to-disk",
"--in-memory",
"--in-memory-detached",
"--in-memory-only",
]:
if argv[0] == "--to-disk":
v = NM.SettingsUpdate2Flags.TO_DISK
elif argv[0] == '--in-memory':
elif argv[0] == "--in-memory":
v = NM.SettingsUpdate2Flags.IN_MEMORY
elif argv[0] == '--in-memory-detached':
elif argv[0] == "--in-memory-detached":
v = NM.SettingsUpdate2Flags.IN_MEMORY_DETACHED
elif argv[0] == '--in-memory-only':
elif argv[0] == "--in-memory-only":
v = NM.SettingsUpdate2Flags.IN_MEMORY_ONLY
elif argv[0] == '--keep':
elif argv[0] == "--keep":
v = NM.SettingsUpdate2Flags.NONE
else:
assert(False)
assert False
if arg_mode is not None:
die('duplicate storage modes ("%s")' % (argv[0]))
arg_mode = v
@@ -102,45 +117,55 @@ def main():
die('unknown argument "%s"' % (argv[0]))
cons.extend(find_connections(nm_client, None, argv[0]))
if len(cons) == 0:
die('could not find connection for "%s"' % (argv[0]))
die('could not find connection for "%s"' % (argv[0]))
if len(cons) != 1:
die('could not find unique connection for "%s"' % (argv[0]))
die('could not find unique connection for "%s"' % (argv[0]))
argv = argv[1:]
continue
if len(cons) != 1:
die('missing connection argument', True)
die("missing connection argument", True)
con = cons[0]
con2 = NM.SimpleConnection.new_clone(con)
result = {}
def _update2_cb(con, async_result, user_data):
try:
r = con.update2_finish(async_result)
except Exception as e:
result['error'] = e
result["error"] = e
else:
result['result'] = r
result["result"] = r
main_loop.quit()
con.update2(con2.to_dbus(NM.ConnectionSerializationFlags.ALL),
(arg_mode if arg_mode is not None else NM.SettingsUpdate2Flags.NONE)
| arg_block_autoconnect
| arg_volatile
| arg_no_reapply,
None,
None,
_update2_cb,
None)
con.update2(
con2.to_dbus(NM.ConnectionSerializationFlags.ALL),
(arg_mode if arg_mode is not None else NM.SettingsUpdate2Flags.NONE)
| arg_block_autoconnect
| arg_volatile
| arg_no_reapply,
None,
None,
_update2_cb,
None,
)
main_loop.run()
if 'error' in result:
die('update connection %s failed [%s]: %s' % (con_to_str(con2), ' '.join(sys.argv), result['error']))
if "error" in result:
die(
"update connection %s failed [%s]: %s"
% (con_to_str(con2), " ".join(sys.argv), result["error"])
)
print('update connection %s succeeded [%s]: %s' % (con_to_str(con2), ' '.join(sys.argv), result['result']))
print(
"update connection %s succeeded [%s]: %s"
% (con_to_str(con2), " ".join(sys.argv), result["result"])
)
if __name__ == '__main__':
if __name__ == "__main__":
main()

View File

@@ -39,59 +39,60 @@ import sys
import re
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
def pr(v):
import pprint
pprint.pprint(v, indent=4, depth=5, width=60)
def parse_args():
args = {
'set': False,
'set-gobject': False,
'filter': [],
'data': []
}
args = {"set": False, "set-gobject": False, "filter": [], "data": []}
i = 1
while i < len(sys.argv):
a = sys.argv[i]
if i == 1:
if a in ['s', 'set']:
args['set'] = True
if a in ["s", "set"]:
args["set"] = True
i += 1
continue
elif a in ['g', 'get']:
args['set'] = False
elif a in ["g", "get"]:
args["set"] = False
i += 1
continue
if a in ['id', 'uuid']:
args['filter'].append((a, sys.argv[i+1]))
if a in ["id", "uuid"]:
args["filter"].append((a, sys.argv[i + 1]))
i += 2
continue
if a in ['--set-gobject']:
args['set-gobject'] = True
if a in ["--set-gobject"]:
args["set-gobject"] = True
i += 1
continue
if a == 'data':
if a == "data":
i += 1
a = sys.argv[i]
if args['set']:
if a == '-d':
args['data'].append((sys.argv[i+1], None))
if args["set"]:
if a == "-d":
args["data"].append((sys.argv[i + 1], None))
else:
args['data'].append((a, sys.argv[i+1]))
args["data"].append((a, sys.argv[i + 1]))
i += 2
else:
args['data'].append(a)
args["data"].append(a)
i += 1
return args
def connection_to_str(connection):
return '%s (%s)' % (connection.get_id(), connection.get_uuid())
return "%s (%s)" % (connection.get_id(), connection.get_uuid())
def connections_filter(connections, filter_data):
connections = list(sorted(connections, key=connection_to_str))
@@ -102,20 +103,21 @@ def connections_filter(connections, filter_data):
# them multiple times.
l = []
for f in filter_data:
if f[0] == 'id':
if f[0] == "id":
for c in connections:
if f[1] == c.get_id():
l.append(c)
else:
assert(f[0] == 'uuid')
assert f[0] == "uuid"
for c in connections:
if f[1] == c.get_uuid():
l.append(c)
return l
def print_user_data(connection, data_allow_regex, data, prefix=''):
def print_user_data(connection, data_allow_regex, data, prefix=""):
s_u = connection.get_setting(NM.SettingUser)
n = 'none'
n = "none"
keys_len = 0
keys = []
if s_u is not None:
@@ -123,17 +125,17 @@ def print_user_data(connection, data_allow_regex, data, prefix=''):
keys_len = len(all_keys)
if data:
for d in data:
if data_allow_regex and len(d) > 0 and d[0] == '~':
if data_allow_regex and len(d) > 0 and d[0] == "~":
r = re.compile(d[1:])
keys.extend([k for k in all_keys if r.match(k)])
else:
keys.append (d)
keys.append(d)
else:
keys.extend(all_keys)
n = '%s' % (keys_len)
n = "%s" % (keys_len)
print('%s%s [%s]' % (prefix, connection_to_str(connection), n))
dd = { }
print("%s%s [%s]" % (prefix, connection_to_str(connection), n))
dd = {}
if s_u is not None:
dd = s_u.get_property(NM.SETTING_USER_DATA)
for k in keys:
@@ -145,7 +147,7 @@ def print_user_data(connection, data_allow_regex, data, prefix=''):
else:
print('%s MISSING: "%s"' % (prefix, k))
else:
assert(v == dd.get(k, None))
assert v == dd.get(k, None)
print('%s SET: "%s" = "%s"' % (prefix, k, v))
else:
print('%s MISSING: "%s"' % (prefix, k))
@@ -155,20 +157,19 @@ def do_get(connections, data):
first_line = True
connections = list(connections)
if not connections:
print('no matching connections (use id|uuid argument)')
print("no matching connections (use id|uuid argument)")
sys.exit(1)
for c in connections:
if first_line:
first_line = False
else:
print('')
print("")
print_user_data(c, True, data)
def do_set(connection, data, set_gobject):
print_user_data(connection, False,
[d[0] for d in data],
prefix = 'BEFORE: ')
print('')
print_user_data(connection, False, [d[0] for d in data], prefix="BEFORE: ")
print("")
s_u = connection.get_setting(NM.SettingUser)
if s_u is None:
connection.add_setting(NM.SettingUser())
@@ -197,35 +198,37 @@ def do_set(connection, data, set_gobject):
print('error setting key "%s" = "%s": %s' % (key, val, e))
sys.exit(1)
try:
connection.commit_changes(True, None)
except Exception as e:
print('failure to commit connection: %s' % (e))
print("failure to commit connection: %s" % (e))
sys.exit(1)
print('')
print_user_data(connection, False,
[d[0] for d in data],
prefix = 'AFTER: ')
print("")
print_user_data(connection, False, [d[0] for d in data], prefix="AFTER: ")
###############################################################################
if __name__ == '__main__':
if __name__ == "__main__":
args = parse_args()
nm_client = NM.Client.new(None)
connections = connections_filter(nm_client.get_connections(), args['filter'])
connections = connections_filter(nm_client.get_connections(), args["filter"])
if args['set']:
if not args['data']:
print('Requires one or more arguments to set or delete')
if args["set"]:
if not args["data"]:
print("Requires one or more arguments to set or delete")
sys.exit(1)
if len(connections) != 1:
print('To set the user-data of a connection, exactly one connection must be selected via id|uuid. Instead, %s connection matched ([%s])' %
(len(connections), ', '.join([connection_to_str(c) for c in connections])))
print(
"To set the user-data of a connection, exactly one connection must be selected via id|uuid. Instead, %s connection matched ([%s])"
% (
len(connections),
", ".join([connection_to_str(c) for c in connections]),
)
)
sys.exit(1)
do_set(connections[0], args['data'], args['set-gobject'])
do_set(connections[0], args["data"], args["set-gobject"])
else:
do_get(connections, args['data'])
do_get(connections, args["data"])

View File

@@ -7,7 +7,8 @@
import locale
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import NM
#
@@ -19,40 +20,50 @@ from gi.repository import NM
# an error without it: http://www.python.org/dev/peps/pep-0263/
#
def clamp(value, minvalue, maxvalue):
return max(minvalue, min(value, maxvalue))
def ssid_to_utf8(ap):
ssid = ap.get_ssid()
if not ssid:
return ""
return NM.utils_ssid_to_utf8(ap.get_ssid().get_data())
def print_device_info(device):
active_ap = dev.get_active_access_point()
ssid = None
if active_ap is not None:
ssid = ssid_to_utf8(active_ap)
info = "Device: %s | Driver: %s | Active AP: %s" % (dev.get_iface(), dev.get_driver(), ssid)
info = "Device: %s | Driver: %s | Active AP: %s" % (
dev.get_iface(),
dev.get_driver(),
ssid,
)
print(info)
print('=' * len(info))
print("=" * len(info))
def mode_to_string(mode):
if mode == getattr(NM, '80211Mode').INFRA:
if mode == getattr(NM, "80211Mode").INFRA:
return "INFRA"
if mode == getattr(NM, '80211Mode').ADHOC:
if mode == getattr(NM, "80211Mode").ADHOC:
return "ADHOC"
if mode == getattr(NM, '80211Mode').AP:
if mode == getattr(NM, "80211Mode").AP:
return "AP"
return "UNKNOWN"
def flags_to_string(flags):
if flags & getattr(NM, '80211ApFlags').PRIVACY:
if flags & getattr(NM, "80211ApFlags").PRIVACY:
return "PRIVACY"
return "NONE"
def security_flags_to_string(flags):
NM_AP_FLAGS = getattr(NM, '80211ApSecurityFlags')
NM_AP_FLAGS = getattr(NM, "80211ApSecurityFlags")
str = ""
if flags & NM_AP_FLAGS.PAIR_WEP40:
str = str + " PAIR_WEP40"
@@ -79,43 +90,50 @@ def security_flags_to_string(flags):
else:
return "NONE"
def flags_to_security(flags, wpa_flags, rsn_flags):
str = ""
if ((flags & getattr(NM, '80211ApFlags').PRIVACY) and
(wpa_flags == 0) and (rsn_flags == 0)):
str = str + " WEP"
if (
(flags & getattr(NM, "80211ApFlags").PRIVACY)
and (wpa_flags == 0)
and (rsn_flags == 0)
):
str = str + " WEP"
if wpa_flags != 0:
str = str + " WPA1"
if rsn_flags != 0:
str = str + " WPA2"
if ((wpa_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_802_1X) or
(rsn_flags & getattr(NM, '80211ApSecurityFlags').KEY_MGMT_802_1X)):
if (wpa_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X) or (
rsn_flags & getattr(NM, "80211ApSecurityFlags").KEY_MGMT_802_1X
):
str = str + " 802.1X"
return str.lstrip()
def print_ap_info(ap):
strength = ap.get_strength()
frequency = ap.get_frequency()
flags = ap.get_flags()
wpa_flags = ap.get_wpa_flags()
rsn_flags = ap.get_rsn_flags()
print("SSID: %s" % (ssid_to_utf8(ap)))
print("BSSID: %s" % (ap.get_bssid()))
print("Frequency: %s" % (frequency))
print("Channel: %s" % (NM.utils_wifi_freq_to_channel(frequency)))
print("Mode: %s" % (mode_to_string(ap.get_mode())))
print("Flags: %s" % (flags_to_string(flags)))
print("WPA flags: %s" % (security_flags_to_string(wpa_flags)))
print("RSN flags: %s" % (security_flags_to_string(rsn_flags)))
print("Security: %s" % (flags_to_security(flags, wpa_flags, rsn_flags)))
print("SSID: %s" % (ssid_to_utf8(ap)))
print("BSSID: %s" % (ap.get_bssid()))
print("Frequency: %s" % (frequency))
print("Channel: %s" % (NM.utils_wifi_freq_to_channel(frequency)))
print("Mode: %s" % (mode_to_string(ap.get_mode())))
print("Flags: %s" % (flags_to_string(flags)))
print("WPA flags: %s" % (security_flags_to_string(wpa_flags)))
print("RSN flags: %s" % (security_flags_to_string(rsn_flags)))
print("Security: %s" % (flags_to_security(flags, wpa_flags, rsn_flags)))
print("Strength: %s %s%%" % (NM.utils_wifi_strength_bars(strength), strength))
print
if __name__ == "__main__":
# Python apparently doesn't call setlocale() on its own? We have to call this or else
# NM.utils_wifi_strength_bars() will think the locale is ASCII-only, and return the
# fallback characters rather than the unicode bars
locale.setlocale(locale.LC_ALL, '')
locale.setlocale(locale.LC_ALL, "")
nmc = NM.Client.new(None)
devs = nmc.get_devices()
@@ -125,4 +143,3 @@ if __name__ == "__main__":
print_device_info(dev)
for ap in dev.get_access_points():
print_ap_info(ap)

View File

@@ -13,7 +13,8 @@
#
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
import sys, socket
@@ -25,7 +26,9 @@ if __name__ == "__main__":
method = sys.argv[2]
if (method == "static" or method == "manual") and len(sys.argv) < 5:
print("Usage: %s %s static address prefix [gateway]" % (sys.argv[0], sys.argv[1]))
print(
"Usage: %s %s static address prefix [gateway]" % (sys.argv[0], sys.argv[1])
)
sys.exit(1)
uuid = sys.argv[1]
@@ -69,4 +72,3 @@ if __name__ == "__main__":
except Exception as e:
sys.stderr.write("Error: %s\n" % e)
break

View File

@@ -9,7 +9,8 @@
# VPN plugin.
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
import sys
@@ -35,17 +36,21 @@ for vpn_info in NM.VpnPluginInfo.list_load():
break
if connection is None:
print("None of the VPN plugins was able to import \"%s\"" % (filename))
print('None of the VPN plugins was able to import "%s"' % (filename))
sys.exit(1)
connection.normalize()
print("connection imported from \"%s\" using plugin \"%s\" (\"%s\", %s)" % (filename, vpn_info.get_filename(), connection.get_id(), connection.get_uuid()))
print(
'connection imported from "%s" using plugin "%s" ("%s", %s)'
% (filename, vpn_info.get_filename(), connection.get_id(), connection.get_uuid())
)
client = NM.Client.new(None)
main_loop = GLib.MainLoop()
def added_cb(client, result, data):
try:
client.add_connection_finish(result)
@@ -54,6 +59,7 @@ def added_cb(client, result, data):
print("ERROR: failed to add connection: %s\n" % e)
main_loop.quit()
client.add_connection_async(connection, True, None, added_cb, None)
main_loop.run()

View File

@@ -9,12 +9,14 @@
import sys
import uuid
import gi
gi.require_version('NM', '1.0')
gi.require_version("NM", "1.0")
from gi.repository import GLib, NM
main_loop = None
client = None
def create_profile(name, peer_mac):
profile = NM.SimpleConnection.new()
@@ -32,8 +34,10 @@ def create_profile(name, peer_mac):
s_wifi_p2p = NM.SettingWifiP2P.new()
s_wifi_p2p.set_property(NM.SETTING_WIFI_P2P_PEER, peer_mac)
s_wifi_p2p.set_property(NM.SETTING_WIFI_P2P_WFD_IES,
GLib.Bytes.new(b'\x00\x00\x06\x00\x90\x1c\x44\x00\xc8'))
s_wifi_p2p.set_property(
NM.SETTING_WIFI_P2P_WFD_IES,
GLib.Bytes.new(b"\x00\x00\x06\x00\x90\x1c\x44\x00\xc8"),
)
profile.add_setting(s_con)
profile.add_setting(s_ip4)
@@ -42,6 +46,7 @@ def create_profile(name, peer_mac):
return profile
def activated_cb(client, result, data):
try:
client.add_and_activate_connection2_finish(result)
@@ -50,35 +55,37 @@ def activated_cb(client, result, data):
sys.stderr.write("Error: %s\n" % e)
main_loop.quit()
def scan_timeout_cb(device):
peers = device.get_peers()
if len(peers) == 0:
main_loop.quit()
sys.exit("No peer found")
print("\n {:20} {:30} {:3} {:30}".format("MAC", "Name", "Sig", "Wfd-IEs"));
print("\n {:20} {:30} {:3} {:30}".format("MAC", "Name", "Sig", "Wfd-IEs"))
for p in peers:
if p.get_wfd_ies() is not None:
ies = p.get_wfd_ies().get_data().hex()
else:
ies = ""
print(" {:20} {:30} {:3} {:30}".format(p.get_hw_address(),
p.get_name(),
p.get_strength(),
ies))
print(
" {:20} {:30} {:3} {:30}".format(
p.get_hw_address(), p.get_name(), p.get_strength(), ies
)
)
print("")
# Connect to first peer
profile = create_profile('P2P-connection', peers[0].get_hw_address())
client.add_and_activate_connection2(profile,
device,
"/",
GLib.Variant('a{sv}', {}),
None,
activated_cb,
None)
print(" * Connecting to peer {} using profile '{}'".format(peers[0].get_hw_address(),
profile.get_id()))
profile = create_profile("P2P-connection", peers[0].get_hw_address())
client.add_and_activate_connection2(
profile, device, "/", GLib.Variant("a{sv}", {}), None, activated_cb, None
)
print(
" * Connecting to peer {} using profile '{}'".format(
peers[0].get_hw_address(), profile.get_id()
)
)
def start_find_cb(device, async_result, user_data):
try:
@@ -87,9 +94,10 @@ def start_find_cb(device, async_result, user_data):
sys.stderr.write("Error: %s\n" % e)
main_loop.quit()
print(" * Scanning on device {}...".format(device.get_iface()));
print(" * Scanning on device {}...".format(device.get_iface()))
GLib.timeout_add(10000, scan_timeout_cb, device)
if __name__ == "__main__":
client = NM.Client.new(None)
device = None
@@ -103,7 +111,7 @@ if __name__ == "__main__":
if device is None:
sys.exit("No Wi-Fi P2P device found")
device.start_find(GLib.Variant('a{sv}', {}), None, start_find_cb, None)
device.start_find(GLib.Variant("a{sv}", {}), None, start_find_cb, None)
main_loop = GLib.MainLoop()
main_loop.run()