Python validators 模块，ipv4() 实例源码

def route_micro(micro):
    '''
        Micro to real URL redirection handler.
    '''
    try:
        temp = lookup_micro(micro)

        if urlcheck(temp):
            return redirect(temp)
        elif domaincheck(temp):
            return redirect("http://" + temp)
        elif ipcheck(temp.split(':')[0]) and urlcheck('http://' + temp):
            # checks for plain ip or an ip with something after it
            return redirect("http://" + temp)
        else:
            abort(404)
    except Exception as e:
        # If micro is not registered, handle the exception from trying to look
        # it up and raise a 404 HTTP error.
        sys.stderr.write(str(e))
        abort(404)

def ip(self, interface_name, newip):
        """
            This function automatically will change netmask address.

            Checking interface first, if interface name found in Get().interfaces()
            validating Ipv4. After that applied ip address to interface

            interface_name: Applied Interface
            newip: New Ip Address

        """
        interface_check = Get().interfaces
        valid_ipv4 = validators.ipv4(newip)

        if not interface_name in interface_check:
            raise WrongInterfaceName("Wrong Interface Name %s" % interface_name)
        elif not valid_ipv4 is True:
            raise NotValidIPv4Address("Not Valid IPv4 Address %s" % newip)
        else:
            ifname = interface_name.encode(encoding='UTF-8')
            ipbytes = socket.inet_aton(newip)
            ifreq = struct.pack('16sH2s4s8s', ifname, AF_INET, b'\x00'*2, ipbytes, b'\x00'*8)
            fcntl.ioctl(self.sock, SIOCSIFADDR, ifreq)

def netmask(self, interface_name, netmask):
        """
            Checking interface first, if interface name found in Get().interfaces()
            validating Ipv4. After that applied ip address to interace

            interface_name = Applied Interface
            netmask = New netmask ip address
        """
        interface_check = Get().interfaces
        valid_ipv4 = validators.ipv4(netmask)

        if not interface_name in interface_check:
            raise WrongInterfaceName("Wrong Interface Name %s" % interface_name)
        elif not valid_ipv4 is True:
            raise NotValidIPv4Address("Not Valid IPv4 Address %s" % netmask)
        else:
            prefix_len = self.get_net_size(netmask.split('.'))
            ifname = interface_name.encode(encoding='UTF-8')
            netmask = ctypes.c_uint32(~((2 ** (32 - prefix_len)) - 1)).value
            nmbytes = socket.htonl(netmask)
            ifreq = struct.pack('16sH2sI8s', ifname, AF_INET, b'\x00'*2, nmbytes, b'\x00'*8) 
            fcntl.ioctl(self.sock, SIOCSIFNETMASK, ifreq)

def search(self):
        mod.display(self.module_name, "", "INFO", "Searching...")
        url = "https://openphish.com/"
        paths = [
            "feed.txt"
        ]
        for path in paths:
            content = Cache(self.module_name, url, path, self.search_method).content
            for line in content.split("\n"):
                try:
                    midle = line.split("//")[-1].split("/")[0]
                except:
                    midle = None
                if self.type == "URL":
                    if self.ioc in line:
                        mod.display(self.module_name, self.ioc, "FOUND", "%s%s"%(url, path))
                        return
                elif self.type == "IPv4" and validators.ipv4(midle):
                    if self.ioc == midle:
                        mod.display(self.module_name, self.ioc, "FOUND", "%s%s"%(url, path))
                        return
                elif self.type == "domain" and validators.domain(midle):
                    if midle == self.ioc:
                        mod.display(self.module_name, self.ioc, "FOUND", "%s%s"%(url, path))
                        return

def checkType(self, argument):
        """
            Identify observable type
        """
        if validators.url(argument):
            return "URL"
        elif validators.md5(argument):
            return "MD5"
        elif validators.sha1(argument):
            return "SHA1"
        elif validators.sha256(argument):
            return "SHA256"
        elif validators.sha512(argument):
            return "SHA512"
        elif validators.ipv4(argument):
            return "IPv4"
        elif validators.ipv6(argument):
            return "IPv6"
        elif validators.domain(argument):
            return "domain"
        else:
            mod.display("MAIN", argument, "ERROR", "Unable to retrieve observable type")
            return None

def search(self):
        mod.display(self.module_name, "", "INFO", "Searching...")
        url = "https://zeustracker.abuse.ch/"
        paths = [
            "blocklist.php?download=baddomains",
            "blocklist.php?download=ipblocklist",
            "blocklist.php?download=compromised"
        ]
        for path in paths:
            if self.type == "URL":
                try:
                    self.ioc = self.ioc.split("://")[1]
                except:
                    pass
            content = Cache(self.module_name, url, path, self.search_method).content
            for line in content.split("\n"):
                if path.split("=")[1] == "compromised":
                    if self.type == "URL":
                        if self.ioc == line:
                            mod.display(self.module_name, self.ioc, "FOUND", "%s%s"%(url, path))
                            return
                    else:
                        line = line.split("/")[0]
                        try:
                            line = line.split(":")[0]
                        except:
                            pass
                if self.type == "domain" and validators.domain(line.strip()):
                    if line.strip() == self.ioc:
                        mod.display(self.module_name, self.ioc, "FOUND", "%s%s"%(url, path))
                        return
                elif self.type == "IPv4" and validators.ipv4(line.strip()):
                    if line.strip() == self.ioc:
                        mod.display(self.module_name, self.ioc, "FOUND", "%s%s"%(url, path))
                        return

def setup_platform(hass, config, add_devices,
                   discovery_info=None):
    """Setup"""
    _LOGGER.debug("Starting up")

    import validators

    if DATA_ARPSPOOF not in hass.data:
        _LOGGER.error("ARP component not initialised!")
        return False

    devices = config.get(CONF_DEVICES, {})
    switches = []
    data = hass.data[DATA_ARPSPOOF]

    for object_id, device_config in devices.items():
        address = device_config.get(CONF_ADDRESS)
        friendly_name = device_config.get(CONF_FRIENDLY_NAME)
        type = -1

        if validators.ipv4(address):
            type = 0
        elif validators.mac_address(address):
            type = 1

        if type > -1:
            _LOGGER.debug("Adding '%s' as switch '%s'",
                          address, friendly_name)

            switches.append(
                ArpSpoofSwitch(
                    data,
                    object_id,
                    friendly_name,
                    address,
                    type,
                    device_config.get(CONF_ICON)
                )
            )
        else:
            _LOGGER.debug("Address '%s' is not valid IP or MAC", address)

    if not switches:
        _LOGGER.error("No devices added")
        return False

    add_devices(switches)