Parse netmasks in IP addresses passed to server install.

ticket 1212

freeipa.spec.in

@@ -188,6 +188,7 @@ Requires: python-kerberos >= 1.1-3
 %endif
 Requires: authconfig
 Requires: gnupg
+Requires: iproute
 Requires: pyOpenSSL
 Requires: python-nss >= 0.11
 Requires: python-lxml

install/tools/ipa-dns-install

@@ -37,9 +37,10 @@ def parse_options():
                       sensitive=True, help="admin password")
     parser.add_option("-d", "--debug", dest="debug", action="store_true",
                       default=False, help="print debugging information")
-    parser.add_option("--ip-address", dest="ip_address", help="Master Server IP Address")
+    parser.add_option("--ip-address", dest="ip_address",
+                      type="ipnet", help="Master Server IP Address")
     parser.add_option("--forwarder", dest="forwarders", action="append",
-                      help="Add a DNS forwarder")
+                      type="ipaddr", help="Add a DNS forwarder")
     parser.add_option("--no-forwarders", dest="no_forwarders", action="store_true",
                       default=False, help="Do not add any DNS forwarders, use root servers instead")
     parser.add_option("--no-reverse", dest="no_reverse",
@@ -105,12 +106,14 @@ def main():
     if options.ip_address:
         ip_address = options.ip_address
     else:
-        ip_address = resolve_host(api.env.host)
+        hostaddr = resolve_host(api.env.host)
+        ip_address = hostaddr and ipautil.CheckedIPAddress(hostaddr)
     if not ip_address or not verify_ip_address(ip_address):
         if options.unattended:
             sys.exit("Unable to resolve IP address for host name")
         else:
             ip_address = read_ip_address(api.env.host, fstore)
+    ip_address = str(ip_address)
     logging.debug("will use ip_address: %s\n", ip_address)
 
     if options.no_forwarders:

install/tools/ipa-replica-install

@@ -61,7 +61,7 @@ def parse_options():
     parser.add_option("--setup-dns", dest="setup_dns", action="store_true",
                       default=False, help="configure bind with our zone")
     parser.add_option("--forwarder", dest="forwarders", action="append",
-                      help="Add a DNS forwarder")
+                      type="ipaddr", help="Add a DNS forwarder")
     parser.add_option("--no-forwarders", dest="no_forwarders", action="store_true",
                       default=False, help="Do not add any DNS forwarders, use root servers instead")
     parser.add_option("--no-reverse", dest="no_reverse", action="store_true",
@@ -270,6 +270,8 @@ def install_bind(config, options):
     ip_address = resolve_host(config.host_name)
     if not ip_address:
         sys.exit("Unable to resolve IP address for host name")
+    ip = installutils.parse_ip_address(ip_address)
+    ip_address = str(ip)
 
     create_reverse = True
     if options.unattended:
@@ -305,6 +307,8 @@ def install_dns_records(config, options):
     ip_address = resolve_host(config.host_name)
     if not ip_address:
         sys.exit("Unable to resolve IP address for host name")
+    ip = installutils.parse_ip_address(ip_address)
+    ip_address = str(ip)
 
     bind.add_master_dns_records(config.host_name, ip_address,
                                 config.realm_name, config.domain_name,

install/tools/ipa-replica-prepare

@@ -24,7 +24,6 @@ import logging, tempfile, shutil, os, pwd
 import traceback
 from ConfigParser import SafeConfigParser
 import krbV
-from optparse import OptionParser
 
 from ipapython import ipautil
 from ipaserver.install import bindinstance, dsinstance, installutils, certs
@@ -33,11 +32,12 @@ from ipaserver.install.replication import check_replication_plugin, enable_repli
 from ipaserver.install.installutils import resolve_host
 from ipaserver.plugins.ldap2 import ldap2
 from ipapython import version
+from ipapython.config import IPAOptionParser
 from ipalib import api, errors, util
 
 def parse_options():
     usage = "%prog [options] FQDN (e.g. replica.example.com)"
-    parser = OptionParser(usage=usage, version=version.VERSION)
+    parser = IPAOptionParser(usage=usage, version=version.VERSION)
 
     parser.add_option("--dirsrv_pkcs12", dest="dirsrv_pkcs12",
                       help="install certificate for the directory server")
@@ -54,7 +54,7 @@ def parse_options():
     parser.add_option("-p", "--password", dest="password", 
                       help="Directory Manager (existing master) password")
     parser.add_option("--ip-address", dest="ip_address",
-                      help="Add A and PTR records of the future replica")
+                      type="ipnet", help="Add A and PTR records of the future replica")
     parser.add_option("--ca", dest="ca_file", default="/root/cacert.p12",
                       help="Location of CA PKCS#12 file, default /root/cacert.p12")
     parser.add_option("--no-pkinit", dest="setup_pkinit", action="store_false",
@@ -79,7 +79,7 @@ def parse_options():
         parser.error("All PKCS#12 options are required if any are used.")
 
     if options.ip_address:
-        if not installutils.verify_ip_address(options.ip_address):
+        if not installutils.verify_ip_address(options.ip_address, match_local=False):
             parser.error("Bad IP address")
             sys.exit(1)
 
@@ -426,11 +426,12 @@ def main():
         name = domain.pop(0)
         domain = ".".join(domain)
 
-        zone = add_zone(domain, nsaddr=options.ip_address)
+        ip_address = str(options.ip_address)
-        add_rr(zone, name, "A", options.ip_address)
+        zone = add_zone(domain, nsaddr=ip_address)
+        add_rr(zone, name, "A", ip_address)
         ns_ip_address = resolve_host(api.env.host)
-        add_reverse_zone(options.ip_address, ns_ip_address)
+        add_reverse_zone(ip_address, ns_ip_address)
-        add_ptr_rr(options.ip_address, replica_fqdn)
+        add_ptr_rr(ip_address, replica_fqdn)
 
 try:
     if not os.geteuid()==0:

install/tools/ipa-server-install

@@ -99,11 +99,12 @@ def parse_options():
     parser.add_option("", "--external_ca_file", dest="external_ca_file",
                       help="File containing PKCS#10 of the external CA chain")
     parser.add_option("--hostname", dest="host_name", help="fully qualified name of server")
-    parser.add_option("--ip-address", dest="ip_address", help="Master Server IP Address")
+    parser.add_option("--ip-address", dest="ip_address",
+                      type="ipnet", help="Master Server IP Address")
     parser.add_option("--setup-dns", dest="setup_dns", action="store_true",
                       default=False, help="configure bind with our zone")
     parser.add_option("--forwarder", dest="forwarders", action="append",
-                      help="Add a DNS forwarder")
+                      type="ipaddr", help="Add a DNS forwarder")
     parser.add_option("--no-forwarders", dest="no_forwarders", action="store_true",
                       default=False, help="Do not add any DNS forwarders, use root servers instead")
     parser.add_option("--no-reverse", dest="no_reverse", action="store_true",
@@ -593,37 +594,34 @@ def main():
     domain_name = domain_name.lower()
 
     # Check we have a public IP that is associated with the hostname
-    ip = resolve_host(host_name)
+    hostaddr = resolve_host(host_name)
-    if ip is None:
+    if hostaddr is not None:
-        if options.ip_address:
+        ip = CheckedIPAddress(hostaddr)
-            ip = options.ip_address
+    else:
+        ip = options.ip_address
     if ip is None and options.unattended:
         sys.exit("Unable to resolve IP address for host name")
 
     if not verify_ip_address(ip):
-        ip = ""
+        ip = None
         if options.unattended:
             sys.exit(1)
 
-    if options.ip_address and options.ip_address != ip:
+    if options.ip_address:
-        if options.setup_dns:
+        if options.ip_address != ip and not options.setup_dns:
-            if not verify_ip_address(options.ip_address):
-                return 1
-            ip = options.ip_address
-        else:
             print >>sys.stderr, "Error: the hostname resolves to an IP address that is different"
             print >>sys.stderr, "from the one provided on the command line.  Please fix your DNS"
             print >>sys.stderr, "or /etc/hosts file and restart the installation."
             return 1
 
-    if options.unattended:
+        ip = options.ip_address
-        if not ip:
+        if not verify_ip_address(ip):
-            sys.exit("Unable to resolve IP address")
+            return 1
 
-    if not ip:
+    if ip is None:
         ip = read_ip_address(host_name, fstore)
-        logging.debug("read ip_address: %s\n" % ip)
+        logging.debug("read ip_address: %s\n" % str(ip))
-    ip_address = ip
+    ip_address = str(ip)
 
     print "The IPA Master Server will be configured with"
     print "Hostname:    " + host_name

ipapython/config.py

@@ -18,7 +18,8 @@
 #
 
 import ConfigParser
-from optparse import Option, Values, OptionParser, IndentedHelpFormatter
+from optparse import Option, Values, OptionParser, IndentedHelpFormatter, OptionValueError
+from copy import copy
 
 import socket
 import ipapython.dnsclient
@@ -46,12 +47,22 @@ class IPAFormatter(IndentedHelpFormatter):
             ret += "%s %s\n" % (spacing, line)
         return ret
 
+def check_ip_option(option, opt, value):
+    from ipapython.ipautil import CheckedIPAddress
+    try:
+        return CheckedIPAddress(value, parse_netmask=(option.type == "ipnet"))
+    except Exception as e:
+        raise OptionValueError("option %s: invalid IP address %s: %s" % (opt, value, e))
+
 class IPAOption(Option):
     """
     optparse.Option subclass with support of options labeled as
     security-sensitive such as passwords.
     """
     ATTRS = Option.ATTRS + ["sensitive"]
+    TYPES = Option.TYPES + ("ipaddr", "ipnet")
+    TYPE_CHECKER = copy(Option.TYPE_CHECKER)
+    TYPE_CHECKER["ipaddr"] = TYPE_CHECKER["ipnet"] = check_ip_option
 
 class IPAOptionParser(OptionParser):
     """

ipapython/ipautil.py

@@ -39,6 +39,7 @@ from types import *
 import re
 import xmlrpclib
 import datetime
+import netaddr
 from ipapython import config
 try:
     from subprocess import CalledProcessError
@@ -63,6 +64,72 @@ def get_domain_name():
 
     return domain_name
 
+class CheckedIPAddress(netaddr.IPAddress):
+    def __init__(self, addr, match_local=True, parse_netmask=True):
+        if isinstance(addr, CheckedIPAddress):
+            super(CheckedIPAddress, self).__init__(addr)
+            self.prefixlen = addr.prefixlen
+            self.defaultnet = addr.defaultnet
+            self.interface = addr.interface
+            return
+
+        net = None
+        iface = None
+        defnet = False
+
+        if isinstance(addr, netaddr.IPNetwork):
+            net = addr
+            addr = net.ip
+        elif isinstance(addr, netaddr.IPAddress):
+            pass
+        else:
+            try:
+                addr = netaddr.IPAddress(addr)
+            except ValueError:
+                net = netaddr.IPNetwork(addr)
+                if not parse_netmask:
+                    raise ValueError("netmask and prefix length not allowed here")
+                addr = net.ip
+
+        if addr.version not in (4, 6):
+            raise ValueError("unsupported IP version")
+        if addr.is_loopback():
+            raise ValueError("cannot use loopback IP address")
+
+        if match_local:
+            if addr.version == 4:
+                family = 'inet'
+            elif addr.version == 6:
+                family = 'inet6'
+
+            ipresult = run(['/sbin/ip', '-family', family, '-oneline', 'address', 'show'])
+            lines = ipresult[0].split('\n')
+            for line in lines:
+                fields = line.split()
+                if len(fields) < 4:
+                    continue
+
+                ifnet = netaddr.IPNetwork(fields[3])
+                if ifnet == net or ifnet.ip == addr:
+                    net = ifnet
+                    iface = fields[1]
+                    break
+
+        if net is None:
+            defnet = True
+            if addr.version == 4:
+                net = netaddr.IPNetwork(netaddr.cidr_abbrev_to_verbose(str(addr)))
+            elif addr.version == 6:
+                net = netaddr.IPNetwork(str(addr) + '/64')
+
+        super(CheckedIPAddress, self).__init__(addr)
+        self.prefixlen = net.prefixlen
+        self.defaultnet = defnet
+        self.interface = iface
+
+    def is_local(self):
+        return self.interface is not None
+
 def realm_to_suffix(realm_name):
     s = realm_name.split(".")
     terms = ["dc=" + x.lower() for x in s]

ipaserver/install/installutils.py

@@ -151,17 +151,18 @@ def verify_fqdn(host_name,no_host_dns=False):
     else:
         print "Warning: Hostname (%s) not found in DNS" % host_name
 
-def verify_ip_address(ip):
+def parse_ip_address(addr, match_local=True, parse_netmask=True):
-    is_ok = True
     try:
-        socket.inet_pton(socket.AF_INET, ip)
+        ip = ipautil.CheckedIPAddress(addr, match_local=match_local, parse_netmask=parse_netmask)
-    except:
+        if match_local and not ip.is_local():
-        try:
+            print "Warning: No network interface matches IP address %s" % addr
-            socket.inet_pton(socket.AF_INET6, ip)
+        return ip
-        except:
+    except Exception as e:
-            print "Unable to verify IP address"
+        print "Error: Invalid IP Address %s: %s" % (addr, e)
-            is_ok = False
+        return None
-    return is_ok
+
+def verify_ip_address(addr, match_local=True, parse_netmask=True):
+    return parse_ip_address(addr, match_local, parse_netmask) is not None
 
 def record_in_hosts(ip, host_name, file="/etc/hosts"):
     hosts = open(file, 'r').readlines()
@@ -194,19 +195,17 @@ def add_record_to_hosts(ip, host_name, file="/etc/hosts"):
 def read_ip_address(host_name, fstore):
     while True:
         ip = ipautil.user_input("Please provide the IP address to be used for this host name", allow_empty = False)
+        ip_parsed = parse_ip_address(ip)
 
-        if ip == "127.0.0.1" or ip == "::1":
+        if ip_parsed is not None:
-            print "The IPA Server can't use localhost as a valid IP"
-            continue
-
-        if verify_ip_address(ip):
             break
 
+    ip = str(ip_parsed)
     print "Adding ["+ip+" "+host_name+"] to your /etc/hosts file"
     fstore.backup_file("/etc/hosts")
     add_record_to_hosts(ip, host_name)
 
-    return ip
+    return ip_parsed
 
 def read_dns_forwarders():
     addrs = []
@@ -218,15 +217,13 @@ def read_dns_forwarders():
                                     allow_empty=True)
             if not ip:
                 break
-            if ip == "127.0.0.1" or ip == "::1":
+            ip_parsed = parse_ip_address(ip, match_local=False, parse_netmask=False)
-                print "You cannot use localhost as a DNS forwarder"
+            if ip_parsed is None:
-                continue
-            if not verify_ip_address(ip):
                 print "DNS forwarder %s not added" % ip
                 continue
 
             print "DNS forwarder %s added" % ip
-            addrs.append(ip)
+            addrs.append(str(ip_parsed))
 
     if not addrs:
         print "No DNS forwarders configured"

tests/test_ipapython/__init__.py

@@ -0,0 +1,22 @@
+# Authors:
+#   Jan Cholasta <jcholast@redhat.com>
+#
+# Copyright (C) 2011  Red Hat
+# see file 'COPYING' for use and warranty information
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Sub-package containing unit tests for `ipapython` package.
+"""

tests/test_ipapython/test_ipautil.py

@@ -0,0 +1,56 @@
+# Authors:
+#   Jan Cholasta <jcholast@redhat.com>
+#
+# Copyright (C) 2011  Red Hat
+# see file 'COPYING' for use and warranty information
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+Test the `ipapython/ipautil.py` module.
+"""
+
+import nose
+
+from ipapython import ipautil
+
+class CheckIPAddress:
+    def __init__(self, addr):
+        self.description = "Test IP address parsing and verification (%s)" % addr
+
+    def __call__(self, addr, words=None, prefixlen=None):
+        try:
+            ip = ipautil.CheckedIPAddress(addr, match_local=False)
+            assert ip.words == words and ip.prefixlen == prefixlen
+        except:
+            assert words is None and prefixlen is None
+
+def test_ip_address():
+    addrs = [
+        ('10.11.12.13',     (10, 11, 12, 13),   8),
+        ('10.11.12.13/14',  (10, 11, 12, 13),   14),
+        ('10.11.12.1337',),
+        ('10.11.12.13/33',),
+        ('127.0.0.1',),
+
+        ('2001::1',         (0x2001, 0, 0, 0, 0, 0, 0, 1), 64),
+        ('2001::1/72',      (0x2001, 0, 0, 0, 0, 0, 0, 1), 72),
+        ('2001::1beef',),
+        ('2001::1/129',),
+        ('::1',),
+
+        ('junk',)
+    ]
+
+    for addr in addrs:
+        yield (CheckIPAddress(addr[0]),) + addr