calculate-utils-2.2-lib/pym/utils/ip.py

#-*- coding: utf-8 -*-

# Copyright 2008-2010 Calculate Ltd. http://www.calculate-linux.org
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.

import math
from cl_utils import process, checkUtils, readFile, listDirectory, \
                     readLinesFile, getRunCommands
import sys
import re
from os import path

import cl_lang
tr = cl_lang.lang()
tr.setLocalDomain('cl_lib')
tr.setLanguage(sys.modules[__name__])

# ip digit from 0|1-255|254 (template)
IP_DIG = "[%s-9]|(?:1[0-9]|[1-9])[0-9]|2[0-4][0-9]|25[0-%s]"
# ip net 0-32
IP_NET_SUFFIX = "[0-9]|[12][0-9]|3[012]"
# ip digs 1-254,0-254,0-255
IP_DIGS = { 'dig1_254' : IP_DIG % (1,4), 'dig0_254' : IP_DIG % (0,4),
           'dig0_255' : IP_DIG % (0,5), }
# ip addr 10.0.0.12
IP_ADDR = "(%(dig1_254)s)\.(%(dig0_254)s)\.(%(dig0_254)s)\.(%(dig1_254)s)"%\
            IP_DIGS
IP_MASK = "(%(dig0_255)s)\.(%(dig0_255)s)\.(%(dig0_255)s)\.(%(dig0_255)s)"%\
            IP_DIGS
# ip addr for net 10.0.0.0
IP_NET = "(%(dig1_254)s)\.(%(dig0_254)s)\.(%(dig0_254)s)\.(%(dig0_254)s)"%\
            IP_DIGS
# ip and net 192.168.0.0/16
IP_ADDR_NET = "(%(ipaddr)s)/((%(ipnet)s))"%{'ipaddr':IP_NET,
                                            'ipnet':IP_NET_SUFFIX}

reIp = re.compile("^{0}$".format(IP_ADDR))
reNetSuffix = re.compile("^{0}$".format(IP_NET_SUFFIX))
reNet = re.compile("^{0}$".format(IP_ADDR_NET))
reMask = re.compile("^{0}$".format(IP_MASK))

def checkIp(ip):
    """Check ip"""
    return reIp.match(ip)

def checkNetSuffix(netSuffix):
    """Check net suffix"""
    return reNetSuffix.match(netSuffix)

def checkNet(net):
    """Check net"""
    if not reNet.match(net):
        return False
    ip,op,cidr = net.partition('/')
    mask = strIpToIntIp(cidrToMask(int(cidr)))
    return (strIpToIntIp(ip)&mask) == (strIpToIntIp(ip))

maskDigs = map(lambda x:str(x),(0b10000000,0b11000000,0b11100000,0b11110000,
                                0b11111000,0b11111100,0b11111110,0b11111111))

def checkMask(mask):
    """Check net"""
    if mask.count('.') != 3:
        return False
    zero = False
    for dig in mask.split('.'):
        if zero or not dig in maskDigs:
            if dig == "0":
                zero = True
            else:
                return False
    return True

def getIpAndMask(interface="eth0"):
    """Get ip and mask from interface"""
    ifconfig = process('/sbin/ifconfig',interface)
    res = re.search(r"inet addr:(\S+)\s.*Mask:(\S+)",ifconfig.read(),re.S)
    if res:
        return res.groups()
    else:
        return ("","")

def strIpToIntIp(addr):
    """Convert ip specified by string to integer"""
    addr = addr.split('.')
    return ((int(addr[0])<<24)|
           (int(addr[1])<<16)|
           (int(addr[2])<<8)|
           (int(addr[3])))
    return reduce(lambda x,y:x+(int(y[1])<<(y[0]*8)),
           enumerate(reversed(addr.split("."))),0)

def intIpToStrIp(addr):
    """Convert ip specified by integer to string"""
    return "{0}.{1}.{2}.{3}".format(
            addr>>24,(addr>>16)&0xff,(addr>>8)&0xff,addr&0xff)

def maskToCidr(mask):
    """Convert mask specified by str to net"""
    mask = strIpToIntIp(mask)
    return 32-int(math.log(((~mask) & 0xffffffff)+1,2))

def cidrToMask(cidr):
    """Convert net to mask specified by str"""
    return intIpToStrIp((2**cidr-1)<<(32-cidr))

def getIpNet(ip,mask=None,cidr=None):
    """Get net (xx.xx.xx.xx/xx) by ip address and mask"""
    ip = strIpToIntIp(ip)
    if not mask is None:
        net = maskToCidr(mask)
    else:
        net = int(cidr)
        mask = cidrToMask(net)
    mask = strIpToIntIp(mask)
    return "{ip}/{net}".format(ip=intIpToStrIp(ip&mask),
                               net=net)

def isIpInNet(checkip,*ipnets):
    """Check is ip in specified nets"""
    return map(lambda x:x[0],
          filter(lambda x:strIpToIntIp(checkip)&x[2] == strIpToIntIp(x[1])&x[2],
          map(lambda x:(x[0],x[1][0],strIpToIntIp(cidrToMask(int(x[1][1])))),
          map(lambda x:(x,x.partition('/')[0::2]),
          ipnets))))

def receiveMac(interface="eth0"):
    """Get MAC from interface"""
    ipconfigProg = checkUtils('/sbin/ifconfig')
    ifconfig = process(ipconfigProg,interface)
    res = re.search(r"HWaddr\s(\S+)",ifconfig.read(),re.S)
    if res:
        return res.group(1)
    else:
        return "00:00:00:00:00:00"

def receiveIpAndMask(interface="eth0"):
    """Get ip and mask from interface"""
    ipconfigProg = checkUtils('/sbin/ifconfig')
    ifconfig = process(ipconfigProg,interface)
    res = re.search(r"inet addr:(\S+)\s.*Mask:(\S+)",ifconfig.read(),re.S)
    if res:
        return res.groups()
    else:
        return ("","")

def isDhcpIp(interface="eth0"):
    """Get ip by dhcp or static"""
    # dhclient
    if filter(lambda x:interface in x and ("dhcpcd" in x or "dhclient" in x),
       getRunCommands()):
        return True
    else:
        return False

def getRouteTable(onlyIface=[]):
    """Get route table, exclude specifed iface"""
    ipProg = checkUtils('/sbin/ip')
    routes = process(ipProg,"route")
    if onlyIface:
        filterRe = re.compile("|".join(map(lambda x:r"dev %s"%x,onlyIface)))
        routes = filter(filterRe.search,routes)
    for line in routes:
        network,op,line = line.partition(" ")
        routeParams = map(lambda x:x.strip(),line.split())
        # (network,{'via':value,'dev':value})
        if network:
            yield (network,dict(zip(routeParams[0::2],routeParams[1::2])))

def getInterfaces():
    """Get available interfaces (discard which hasn't device)"""
    sysNet = "/sys/class/net"
    return filter(lambda x:path.exists(path.join(sysNet,x,"device")),
           listDirectory(sysNet))