#!/usr/bin/python # -*- coding: utf-8 -*- import dbus WPAS_DBUS_OPATH = "/fi/epitest/hostap/WPASupplicant" WPAS_DBUS_INTERFACES_OPATH = "/fi/epitest/hostap/WPASupplicant/Interfaces" WPAS_DBUS_SERVICE = "fi.epitest.hostap.WPASupplicant" WPAS_DBUS_INTERFACE = "fi.epitest.hostap.WPASupplicant" WPAS_DBUS_INTERFACES_INTERFACE = "fi.epitest.hostap.WPASupplicant.Interface" WPAS_DBUS_NETWORK_INTERFACE = "fi.epitest.hostap.WPASupplicant.Network" WPAS_DBUS_BSSID_INTERFACE = "fi.epitest.hostap.WPASupplicant.BSSID" TIMEOUT = 60 class PasswordLengthError(Exception): pass class WPA_Supplicant_Network: def __init__(self, bus, path): self.bus = bus self.path = path self.net_obj = self.bus.get_object(WPAS_DBUS_SERVICE, path) self.net = dbus.Interface(self.net_obj, WPAS_DBUS_NETWORK_INTERFACE) # dict keys: ssid, bssid, key_mgmt, psk, scan_ssid, pairwise, group, eap, identity, # anonymous_identity, ca_cert, ca_cert2, client_cert, client_cert2, private_key, private_key2, # private_key_passwd, private_key2_passwd, phase1, phase2, eapol_flags # # Example: setNetwork({"ssid":dbus.String("MySSID", variant_level=1)), # "psk":dbus.String("MyPassword", variant_level=1))}) def setNetwork(self, options): self.net.set(options) def enableNetwork(self): self.net.enable() def disableNetwork(self): self.net.disable() class WPA_Supplicant_Interface: def __init__(self, bus, ifname, path): self.bus = bus self.ifname = ifname self.path = path self.if_obj = self.bus.get_object(WPAS_DBUS_SERVICE, path) self.iface = dbus.Interface(self. if_obj, WPAS_DBUS_INTERFACES_INTERFACE) def scan(self): self.iface.scan() def scanResults(self): return self.iface.scanResults() def addNetwork(self): path = self.iface.addNetwork() return self.getNetwork(path) def removeNetwork(self, network): self.iface.removeNetwork(network) def selectNetwork(self, network): self.iface.selectNetwork(network) def getNetworkPath(self, network_id): return "%s/Networks/%d" % (self.path, network_id) def getNetworkById(self, network_id): return WPA_Supplicant_Network(self.bus, self.getNetworkPath(network_id)) def getNetwork(self, network): return WPA_Supplicant_Network(self.bus, network) # mode should be between 0 and 2 def setAPScan(self, mode): self.iface.setAPScan(dbus.UInt32(mode)) def disconnect(self): self.iface.disconnect() def getState(self): return self.iface.state() def getCapabilities(self): return self.iface.capabilities() class WPA_Supplicant: def __init__(self): self.bus = dbus.SystemBus() self.wpas_obj = self.bus.get_object(WPAS_DBUS_SERVICE, WPAS_DBUS_OPATH) self.wpas = dbus.Interface(self.wpas_obj, WPAS_DBUS_INTERFACE) def getInterface(self, ifname): path = self.wpas.getInterface(ifname) return WPA_Supplicant_Interface(self.bus, ifname, path) # driver=wext, hostap, prism54, madwifi, atmel, ndiswrapper, ipw, wired def addInterface(self, ifname, driver): self.wpas.addInterface(ifname, {'driver': dbus.String(driver, variant_level=1)}) return self.getInterface(ifname) def removeInterface(self, ifname): try: iface_path = self.wpas.getInterface(ifname) except dbus.DBusException: iface_path = None if iface_path: self.wpas.removeInterface(dbus.ObjectPath(iface_path)) def detectWpaDriver(ifname): import os import sys import pardus.netutils # if device is an ethernet device, driver is "wired" device = pardus.netutils.IF(ifname) if (device.isEthernet()) and (not device.isWireless()): return "wired" path = os.path.join("/sys/class/net/", ifname, "device/driver/module") modname = None if os.path.exists(path): modname = os.readlink(path).split("/")[-1] if "hostap" in modname: return "hostap" if "prism54" in modname: return "prism54" if "atmel" in modname: return "atmel" # we fallback to wext # wext is the generic driver for ipw2100, ipw2200, ndiswrapper > 1.12, madwifi etc... return "wext" def getWpaInterface(ifname): wpa = WPA_Supplicant() try: iface = wpa.getInterface(ifname) except dbus.DBusException: driver = detectWpaDriver(ifname) iface = wpa.addInterface(ifname, driver) return iface def waitForAuthenticationComplete(iface, timeout, wait = 0.1): import time while timeout > 0: if iface.getState() == "COMPLETED": return True else: timeout -= wait time.sleep(wait) return False def checkServiceState(serviceName): bus = dbus.SystemBus() obj = bus.get_object("tr.org.pardus.comar", "/package/%s" % serviceName) state = obj.info(dbus_interface="tr.org.pardus.comar.System.Service")[2] return state in ("on", "started") def isDBusServiceActive(): return True def isWpaServiceActive(): return checkServiceState("wpa_supplicant") def isWpaServiceUsable(): return isDBusServiceActive() and isWpaServiceActive() def startWpaService(): if not isWpaServiceActive(): bus = dbus.SystemBus() obj = bus.get_object("tr.org.pardus.comar", "/package/wpa_supplicant") obj.start(dbus_interface="tr.org.pardus.comar.System.Service") import time timeout = 10 while timeout > 0: try: bus = dbus.SystemBus() net_obj = bus.get_object(WPAS_DBUS_SERVICE, WPAS_DBUS_OPATH) net = dbus.Interface(net_obj, WPAS_DBUS_NETWORK_INTERFACE) except dbus.DBusException: time.sleep(0.1) timeout -= 0.2 continue return True return False def setWpaAuthentication(ifname, ssid, password, timeout = TIMEOUT): password_length = len(password) if (password_length < 8) or (password_length > 63): raise PasswordLengthError("Password length should be between 8 and 63") iface = getWpaInterface(ifname) network = iface.addNetwork() network.setNetwork({"ssid": dbus.String(ssid, variant_level=1), "psk": dbus.String(password, variant_level=1)}) iface.selectNetwork(network.path) authentication = waitForAuthenticationComplete(iface, timeout) if not authentication: disableAuthentication(ifname) return authentication def disableAuthentication(ifname): wpa = WPA_Supplicant() wpa.removeInterface(ifname) bus = dbus.SystemBus() obj = bus.get_object("tr.org.pardus.comar", "/package/wpa_supplicant") obj.stop(dbus_interface="tr.org.pardus.comar.System.Service") class Wpa_EAP: ssid = "" phase1 = "" phase2 = "" key_mgmt = "IEEE8021X" eap = "PEAP" anonymous_identity = "" ca_cert = "" client_cert = "" private_key = "" private_key_passwd = "" def __init__(self, ifname): self.ifname = ifname self.iface = getWpaInterface(ifname) self.network = self.iface.addNetwork() def authenticate(self, username, password, timeout = TIMEOUT): basic = {"ssid": dbus.String(self.ssid, variant_level=1), "key_mgmt": dbus.String(self.key_mgmt, variant_level=1), "eap": dbus.String(self.eap, variant_level=1), "identity": dbus.String(username, variant_level=1)} if self.client_cert: basic["client_cert"] = dbus.String(self.client_cert, variant_level=1) if self.ca_cert: basic["ca_cert"] = dbus.String(self.ca_cert, variant_level=1) if self.private_key: basic["private_key"] = dbus.String(self.private_key, variant_level=1) if self.private_key_passwd: basic["private_key_passwd"] = dbus.String(self.private_key_passwd, variant_level=1) if self.phase2: basic["phase2"] = dbus.String("auth=%s"%self.phase2, variant_level=1) if password: basic["password"] = dbus.String(password, variant_level=1) if self.anonymous_identity: basic["anonymous_identity"] = dbus.String(self.anonymous_identity, variant_level=1) self.network.setNetwork(basic) self.iface.selectNetwork(self.network.path) authentication = waitForAuthenticationComplete(self.iface, timeout) if not authentication: disableAuthentication(self.ifname) return authentication