import ipaddress
import logging
import os
+import re
import shutil
import socket
import sys
# Load our location module
import location
import location.downloader
+import location.export
+
from location.i18n import _
# Setup logging
# Output formatters
-class OutputFormatter(object):
- def __init__(self, ns):
- self.ns = ns
-
- def __enter__(self):
- # Open the output
- self.open()
-
- return self
-
- def __exit__(self, type, value, tb):
- if tb is None:
- self.close()
-
- @property
- def name(self):
- if "country_code" in self.ns:
- return "networks_country_%s" % self.ns.country_code[0]
-
- elif "asn" in self.ns:
- return "networks_AS%s" % self.ns.asn[0]
-
- def open(self):
- pass
-
- def close(self):
- pass
-
- def network(self, network):
- print(network)
-
-
-class IpsetOutputFormatter(OutputFormatter):
- """
- For nftables
- """
- def open(self):
- print("create %s hash:net family inet hashsize 1024 maxelem 65536" % self.name)
-
- def network(self, network):
- print("add %s %s" % (self.name, network))
-
-
-class NftablesOutputFormatter(OutputFormatter):
- """
- For nftables
- """
- def open(self):
- print("define %s = {" % self.name)
-
- def close(self):
- print("}")
-
- def network(self, network):
- print(" %s," % network)
-
-
-class XTGeoIPOutputFormatter(OutputFormatter):
- """
- Formats the output in that way, that it can be loaded by
- the xt_geoip kernel module from xtables-addons.
- """
- def network(self, network):
- n = ipaddress.ip_network("%s" % network)
-
- for address in (n.network_address, n.broadcast_address):
- bytes = socket.inet_pton(
- socket.AF_INET6 if address.version == 6 else socket.AF_INET,
- "%s" % address,
- )
-
- os.write(1, bytes)
-
-
class CLI(object):
- output_formats = {
- "ipset" : IpsetOutputFormatter,
- "list" : OutputFormatter,
- "nftables" : NftablesOutputFormatter,
- "xt_geoip" : XTGeoIPOutputFormatter,
- }
-
def parse_cli(self):
parser = argparse.ArgumentParser(
description=_("Location Database Command Line Interface"),
)
list_networks_by_as.add_argument("asn", nargs=1, type=int)
list_networks_by_as.add_argument("--family", choices=("ipv6", "ipv4"))
- list_networks_by_as.add_argument("--output-format",
- choices=self.output_formats.keys(), default="list")
+ list_networks_by_as.add_argument("--format",
+ choices=location.export.formats.keys(), default="list")
list_networks_by_as.set_defaults(func=self.handle_list_networks_by_as)
# List all networks in a country
)
list_networks_by_cc.add_argument("country_code", nargs=1)
list_networks_by_cc.add_argument("--family", choices=("ipv6", "ipv4"))
- list_networks_by_cc.add_argument("--output-format",
- choices=self.output_formats.keys(), default="list")
+ list_networks_by_cc.add_argument("--format",
+ choices=location.export.formats.keys(), default="list")
list_networks_by_cc.set_defaults(func=self.handle_list_networks_by_cc)
# List all networks with flags
action="store_true", help=_("Anycasts"),
)
list_networks_by_flags.add_argument("--family", choices=("ipv6", "ipv4"))
- list_networks_by_flags.add_argument("--output-format",
- choices=self.output_formats.keys(), default="list")
+ list_networks_by_flags.add_argument("--format",
+ choices=location.export.formats.keys(), default="list")
list_networks_by_flags.set_defaults(func=self.handle_list_networks_by_flags)
+ # Export
+ export = subparsers.add_parser("export",
+ help=_("Exports data in many formats to load it into packet filters"),
+ )
+ export.add_argument("--format", help=_("Output format"),
+ choices=location.export.formats.keys(), default="list")
+ export.add_argument("--directory", help=_("Output directory"), required=True)
+ export.add_argument("--family",
+ help=_("Specify address family"), choices=("ipv6", "ipv4"),
+ )
+ export.add_argument("objects", nargs="+", help=_("List country codes or ASNs to export"))
+ export.set_defaults(func=self.handle_export)
+
args = parser.parse_args()
# Configure logging
def __get_output_formatter(self, ns):
try:
- cls = self.output_formats[ns.output_format]
+ cls = location.export.formats[ns.format]
except KeyError:
- cls = OutputFormatter
+ cls = location.export.OutputFormatter
- return cls(ns)
+ return cls
def handle_list_networks_by_as(self, db, ns):
- with self.__get_output_formatter(ns) as f:
- for asn in ns.asn:
- # Print all matching networks
- for n in db.search_networks(asn=asn, family=ns.family):
- f.network(n)
+ writer = self.__get_output_formatter(ns)
+
+ for asn in ns.asn:
+ f = writer(sys.stdout, prefix="AS%s" % asn)
+
+ # Print all matching networks
+ for n in db.search_networks(asn=asn, family=ns.family):
+ f.write(n)
+
+ f.finish()
def handle_list_networks_by_cc(self, db, ns):
- with self.__get_output_formatter(ns) as f:
- for country_code in ns.country_code:
- # Print all matching networks
- for n in db.search_networks(country_code=country_code, family=ns.family):
- f.network(n)
+ writer = self.__get_output_formatter(ns)
+
+ for country_code in ns.country_code:
+ # Open standard output
+ f = writer(sys.stdout, prefix=country_code)
+
+ # Print all matching networks
+ for n in db.search_networks(country_code=country_code, family=ns.family):
+ f.write(n)
+
+ f.finish()
def handle_list_networks_by_flags(self, db, ns):
flags = 0
if not flags:
raise ValueError(_("You must at least pass one flag"))
- with self.__get_output_formatter(ns) as f:
- for n in db.search_networks(flags=flags, family=ns.family):
- f.network(n)
+ writer = self.__get_output_formatter(ns)
+ f = writer(sys.stdout, prefix="custom")
+
+ for n in db.search_networks(flags=flags, family=ns.family):
+ f.write(n)
+
+ f.finish()
+
+ def handle_export(self, db, ns):
+ countries, asns = [], []
+
+ # Translate family
+ if ns.family == "ipv6":
+ families = [ socket.AF_INET6 ]
+ elif ns.family == "ipv4":
+ families = [ socket.AF_INET ]
+ else:
+ families = [ socket.AF_INET6, socket.AF_INET ]
+
+ for object in ns.objects:
+ m = re.match("^AS(\d+)$", object)
+ if m:
+ object = int(m.group(1))
+
+ asns.append(object)
+
+ elif location.country_code_is_valid(object) \
+ or object in ("A1", "A2", "A3"):
+ countries.append(object)
+
+ else:
+ log.warning("Invalid argument: %s" % object)
+ continue
+
+ if not countries and not asns:
+ log.error("Nothing to export")
+ return 2
+
+ # Select the output format
+ writer = self.__get_output_formatter(ns)
+
+ e = location.export.Exporter(db, writer)
+ e.export(ns.directory, countries=countries, asns=asns, families=families)
def main():