public
Last active — forked from /autovpn.py

AutoVPN for NetworkManager

  • Download Gist
autovpn.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111
#!/usr/bin/python
import sys
 
from dbus.mainloop.glib import DBusGMainLoop
import dbus
import gobject
 
 
class AutoVPN(object):
"""This should keep alive a VPN connection."""
 
def __init__(self, vpn_name, max_attempts=5, delay=5000):
self.vpn_name = vpn_name
self.max_attempts = max_attempts
self.delay = delay
self.failed_attempts = 0
self.bus = dbus.SystemBus()
self.get_network_manager().connect_to_signal("StateChanged", self.onNetworkStateChanged)
print "Maintaining connection for %s, reattempting up to %d times with %d ms between retries" % (vpn_name, max_attempts, delay)
 
def onNetworkStateChanged(self, state):
"""Handles network status changes and activates the VPN on established connection."""
print "Network state changed: %d" % state
if state == 3:
self.activate_vpn()
 
def onVpnStateChanged(self, state, reason):
"""Handles different VPN status changes and eventually reconnects the VPN."""
# vpn connected or user disconnected manually?
if state == 5 or (state == 7 and reason == 2):
print "VPN connected, or user disconnected manually"
self.failed_attempts = 0
return
# connection failed or unknown?
elif state in [6, 7]:
# reconnect if we haven't reached max_attempts
if not self.max_attempts or self.failed_attempts < self.max_attempts:
print "Connection failed, attempting to reconnect"
self.failed_attempts += 1
gobject.timeout_add(self.delay, self.activate_vpn)
else:
print "Connection failed, exceeded %d max attempts." % self.max_attempts
self.failed_attempts = 0
 
def get_network_manager(self):
"""Gets the network manager dbus interface."""
print "Getting NetworkManager DBUS interface"
proxy = self.bus.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager')
return dbus.Interface(proxy, 'org.freedesktop.NetworkManager')
 
def get_vpn_interface(self, name):
'Gets the VPN connection interface with the specified name.'
print "Getting %s VPN connection DBUS interface" % name
proxy = self.bus.get_object('org.freedesktop.NetworkManagerUserSettings', '/org/freedesktop/NetworkManagerSettings')
iface = dbus.Interface(proxy, 'org.freedesktop.NetworkManagerSettings')
connections = iface.ListConnections()
for connection in connections:
proxy = self.bus.get_object('org.freedesktop.NetworkManagerUserSettings', connection)
iface = dbus.Interface(proxy, 'org.freedesktop.NetworkManagerSettings.Connection')
con_settings = iface.GetSettings()['connection']
if con_settings['type'] == 'vpn' and con_settings['id'] == name:
print "Got %s interface" % name
return connection
print "Unable to acquire %s VPN interface. Does it exist?" % name
return None
 
def get_active_connection(self):
"""Gets the dbus interface of the first active
network connection or returns None.
"""
print "Getting active network connection"
proxy = self.bus.get_object('org.freedesktop.NetworkManager', '/org/freedesktop/NetworkManager')
iface = dbus.Interface(proxy, 'org.freedesktop.DBus.Properties')
active_connections = iface.Get('org.freedesktop.NetworkManager', 'ActiveConnections')
if len(active_connections) == 0:
print "No active connections found"
return None
print "Found %d active connection(s)" % active_connections.len
return active_connections[0]
 
def activate_vpn(self):
'Activates the vpn connection.'
print "Activating %s VPN connection" % self.vpn_name
vpn_con = self.get_vpn_interface(self.vpn_name)
active_con = self.get_active_connection()
if vpn_con and active_con:
new_con = self.get_network_manager().ActivateConnection(
'org.freedesktop.NetworkManagerUserSettings',
vpn_con,
dbus.ObjectPath("/"),
active_con,
)
proxy = self.bus.get_object('org.freedesktop.NetworkManager', new_con)
iface = dbus.Interface(proxy, 'org.freedesktop.NetworkManager.VPN.Connection')
iface.connect_to_signal('VpnStateChanged', self.onVpnStateChanged)
print "VPN Connection %s should be active" % self.vpn_name
 
 
if __name__ == '__main__':
if len(sys.argv) != 2:
print 'usage: autovpn <VPN_CONNECTION_NAME>'
sys.exit(0)
 
# set up the main loop
DBusGMainLoop(set_as_default=True)
loop = gobject.MainLoop()
 
AutoVPN(sys.argv[1])
# execute the main loop
loop.run()

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.