"""
Host definitions.
"""
import pvl.args
import pvl.dns.zone
import collections
import configobj
import ipaddr
import optparse
import os.path
import logging; log = logging.getLogger('pvl.hosts')
def optparser (parser) :
hosts = optparse.OptionGroup(parser, "Hosts input")
hosts.add_option('--hosts-charset', metavar='CHARSET', default='utf-8',
help="Encoding used for host files")
hosts.add_option('--hosts-domain', metavar='DOMAIN',
help="Default domain for hosts")
hosts.add_option('--hosts-include', metavar='PATH',
help="Optional path for hosts includes, beyond host config dir")
return hosts
class HostError (Exception) :
pass
class HostConfigError (HostError) :
pass
def parse_ethernet (value) :
"""
Normalize ethernet str.
"""
return ':'.join('%02x' % int(x, 16) for x in value.split(':'))
class Host (object) :
# the label used for alias4/6 hosts
ALIAS4_FMT = '{host}-ipv4'
ALIAS6_FMT = '{host}-ipv6'
@classmethod
def expand (cls, options, range, host, domain, ip, forward=None, reverse=None, **opts) :
host = pvl.dns.zone.parse_generate_field(host)
ip = pvl.dns.zone.parse_generate_field(ip)
if forward is not None :
_forward = pvl.dns.zone.parse_generate_field(forward)
else :
_forward = lambda i: forward
if reverse is not None :
_reverse = pvl.dns.zone.parse_generate_field(reverse)
else :
_reverse = lambda i: reverse
for i in range :
yield cls.build(options, host(i), domain,
ip = ip(i),
forward = _forward(i),
reverse = _reverse(i),
**opts
)
@classmethod
def config (cls, options, host, section=None, domain=None, ip=None, **extra) :
"""
Yield Hosts from a config section's scalars.
options - pvl.args Options
host - the name of the section (or file) containing this host item.
section - the parent section containing this host item, or None
used for domain
"""
if domain :
log.debug("%s: explicit domain: %s", host, domain)
elif '.' in host :
log.debug("%s: using as fqdn without domain", host)
domain = None
elif options.hosts_domain :
log.debug("%s: default domain to --hots-domain: %s", host, options.hosts_domain)
domain = options.hosts_domain
elif section :
log.debug("%s: default domain to section: %s", host, section)
domain = section
else :
raise ValueError("%s: no domain given" % (host, ))
if '{' in host :
pre, host = host.split('{', 1)
range, post = host.rsplit('}', 1)
range = pvl.dns.zone.parse_generate_range(range)
host = pre + "$" + post
for host in cls.expand(options, range, host, domain, ip, **extra) :
yield host
else :
yield cls.build(options, host, domain, ip=ip, **extra)
@classmethod
def build (cls, options, host, domain,
ip=None, ip6=None, owner=None, location=None,
alias=None, alias4=None, alias6=None,
forward=None, reverse=None,
down=None,
boot=None,
**extra) :
"""
Return a Host from a config section's scalars.
"""
if alias :
alias = alias.split()
else :
alias = ()
if alias4 :
alias4 = alias4.split()
else :
alias4 = ()
if alias6 :
alias6 = alias6.split()
else :
alias6 = ()
ethernet = { }
extensions = collections.defaultdict(dict)
for field, value in extra.iteritems() :
if ':' in field :
extension, field = field.split(':', 1)
extensions[extension][field] = value
continue
if '.' in field :
field, instance = field.split('.')
else :
instance = None
if field == 'ethernet' :
if instance :
ethernet[instance] = parse_ethernet(value)
else :
for eth in value.split() :
ethernet[len(ethernet)] = parse_ethernet(eth)
else :
raise ValueError("%s: Unknown host field: %s=%s" % (host, field, value))
if forward is None :
# normal zone
pass
elif forward :
# alias to external zone
pass
else :
# omit
forward = False
if reverse is None :
# normal zone
pass
elif reverse :
# alias to external zone
pass
else :
# omit
reverse = False
if down :
down = True
else :
down = None
if not location :
location = location_domain = None
elif '@' in location:
location, location_domain = location.split('@', 1)
else:
location_domain = None
return cls(host,
domain = domain,
ip = ipaddr.IPv4Address(ip) if ip else None,
ip6 = ipaddr.IPv6Address(ip6) if ip6 else None,
ethernet = ethernet,
alias = alias,
alias4 = alias4,
alias6 = alias6,
owner = owner,
location = location,
location_domain = location_domain,
boot = boot,
forward = forward,
reverse = reverse,
down = down,
**extensions
)
def __init__ (self, host,
domain=None,
ip=None, ip6=None,
ethernet={ },
alias=(),
owner=None,
location=None,
location_domain=None,
boot=None,
alias4=None, alias6=None,
forward=None, reverse=None,
down=None,
**extensions
) :
"""
host - str
domain - str
ip - ipaddr.IPv4Address
ip6 - ipaddr.IPv6Address
ethernet - { index: ethernet }
alias - list
owner - str: LDAP uid
location - location name@ part
location_domain - location @domain part
alias4 - list (CNAME -> A)
alias6 - list (CNAME -> AAAA)
forward - generate forward records, or CNAME into given zone
reverse - generate reverse records, or CNAME into given zone
down - not online
"""
self.host = host
self.domain = domain
self.ip = ip
self.ip6 = ip6
self.ethernet = ethernet
self.alias = alias
self.alias4 = alias4
self.alias6 = alias6
self.owner = owner
self.location = location
self.location_domain = location_domain
self.boot = boot
self.forward = forward
self.reverse = reverse
self.down = down
self.extensions = extensions
def fqdn (self) :
if '.' in self.host :
return self.host + '.'
elif self.domain :
return pvl.dns.zone.fqdn(self.host, self.domain)
else :
raise ValueError("%s: have no fqdn/domain" % (self, ))
def __str__ (self) :
return str(self.host)
def apply_host_includes (options, config_path, include) :
"""
Yield files from a given config's include=... value
"""
if options.hosts_include :
include_paths = (os.path.dirname(config_path), options.hosts_include)
else :
include_paths = (os.path.dirname(config_path), )
for include in include.split() :
for include_path in include_paths :
path = os.path.join(include_path, include)
if os.path.exists(path) :
break
else :
raise HostConfigError(config_path, "Unable to find include %s in include path: %s" % (include, ' '.join(include_paths)))
if include.endswith('/') :
for name in os.listdir(path) :
file_path = os.path.join(path, name)
if name.startswith('.') :
pass
elif not os.path.exists(file_path) :
log.warn("%s: skip nonexistant include: %s", config_path, file_path)
else :
log.info("%s: include: %s", config_path, file_path)
yield open(file_path)
else :
log.info("%s: include: %s", config_path, path)
yield open(path)
def apply_hosts_config (options, path, name, config, defaults={}, parent=None) :
"""
Load hosts from a ConfigObj section.
"""
scalars = dict((scalar, config[scalar]) for scalar in config.scalars)
if config.sections :
# includes?
for file in apply_host_includes(options, path, scalars.pop('include', '')) :
# within our domain context
for host in apply_hosts_file(options, file, parent=name) :
yield host
# recurse; this is a domain meta-section
params = dict(defaults)
params.update(**scalars) # override
log.debug("%s: %s -> %s", path, parent, name)
for section in config.sections :
for host in apply_hosts_config(options, path, section, config[section], params, name) :
yield host
elif name :
params = dict(defaults, **scalars)
log.debug("%s: %s: %s", path, parent, name)
# this is a host section
for host in Host.config(options, name, parent, **params) :
yield host
else :
raise ValueError("No sections in config")
def apply_hosts_file (options, file, parent=None) :
"""
Load Hosts from a file.
"""
# use file basename as default
path = file.name
name = os.path.basename(path)
try :
config = configobj.ConfigObj(file,
encoding = options.hosts_charset,
)
except configobj.ParseError as ex :
raise HostConfigError("%s:%d: %s" % (path, ex.line_number, ex))
return apply_hosts_config(options, path, name, config, parent=parent)
def apply_hosts (options, files) :
"""
Load Hosts from files.
"""
for file in files :
for host in apply_hosts_file(options, file) :
yield host
def sort_hosts (options, hosts) :
"""
Yields hosts with a sorting key.
"""
for host in hosts :
if host.ip :
sort = host.ip
else :
# sorts first
sort = ipaddr.IPAddress(0)
yield sort, host
def apply (options, args) :
"""
Load Hosts from arguments.
"""
# without unicode
files = pvl.args.apply_files(args, 'r')
# load configs
hosts = apply_hosts(options, files)
# sort
hosts = list(sort_hosts(options, hosts))
hosts.sort()
hosts = [host for sort, host in hosts]
return hosts