[location/libloc.git] / src / python / location-query.in

#!/usr/bin/python3
###############################################################################
#                                                                             #
# libloc - A library to determine the location of someone on the Internet     #
#                                                                             #
# Copyright (C) 2017 IPFire Development Team <info@ipfire.org>                #
#                                                                             #
# This library is free software; you can redistribute it and/or               #
# modify it under the terms of the GNU Lesser General Public                  #
# License as published by the Free Software Foundation; either                #
# version 2.1 of the License, or (at your option) any later version.          #
#                                                                             #
# This library 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           #
# Lesser General Public License for more details.                             #
#                                                                             #
###############################################################################

import argparse
import gettext
import ipaddress
import os
import socket
import sys
import syslog

# Load our location module
import location

# i18n
def _(singular, plural=None, n=None):
	if plural:
		return gettext.dngettext("libloc", singular, plural, n)

	return gettext.dgettext("libloc", singular)

# 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"),
		)
		subparsers = parser.add_subparsers()

		# Global configuration flags
		parser.add_argument("--debug", action="store_true",
			help=_("Enable debug output"))

		# version
		parser.add_argument("--version", action="version",
			version="%%(prog)s %s" % location.__version__)

		# database
		parser.add_argument("--database", "-d",
			default="@databasedir@/database.db", help=_("Path to database"),
		)

		# public key
		parser.add_argument("--public-key", "-k",
			default="@databasedir@/signing-key.pem", help=_("Public Signing Key"),
		)

		# lookup an IP address
		lookup = subparsers.add_parser("lookup",
			help=_("Lookup one or multiple IP addresses"),
		)
		lookup.add_argument("address", nargs="+")
		lookup.set_defaults(func=self.handle_lookup)

		# Get AS
		get_as = subparsers.add_parser("get-as",
			help=_("Get information about one or multiple Autonomous Systems"),
		)
		get_as.add_argument("asn", nargs="+")
		get_as.set_defaults(func=self.handle_get_as)

		# Search for AS
		search_as = subparsers.add_parser("search-as",
			help=_("Search for Autonomous Systems that match the string"),
		)
		search_as.add_argument("query", nargs=1)
		search_as.set_defaults(func=self.handle_search_as)

		# List all networks in an AS
		list_networks_by_as = subparsers.add_parser("list-networks-by-as",
			help=_("Lists all networks in an AS"),
		)
		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.set_defaults(func=self.handle_list_networks_by_as)

		# List all networks in a country
		list_networks_by_cc = subparsers.add_parser("list-networks-by-cc",
			help=_("Lists 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.set_defaults(func=self.handle_list_networks_by_cc)

		# List all networks with flags
		list_networks_by_flags = subparsers.add_parser("list-networks-by-flags",
			help=_("Lists all networks with flags"),
		)
		list_networks_by_flags.add_argument("--anonymous-proxy",
			action="store_true", help=_("Anonymous Proxies"),
		)
		list_networks_by_flags.add_argument("--satellite-provider",
			action="store_true", help=_("Satellite Providers"),
		)
		list_networks_by_flags.add_argument("--anycast",
			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.set_defaults(func=self.handle_list_networks_by_flags)

		args = parser.parse_args()

		# Print usage if no action was given
		if not "func" in args:
			parser.print_usage()
			sys.exit(2)

		return args

	def run(self):
		# Parse command line arguments
		args = self.parse_cli()

		# Open database
		try:
			db = location.Database(args.database)
		except FileNotFoundError as e:
			sys.stderr.write("location-query: Could not open database %s: %s\n" \
				% (args.database, e))
			sys.exit(1)

		# Translate family
		if args.family == "ipv6":
			args.family = socket.AF_INET6
		elif args.family == "ipv4":
			args.family = socket.AF_INET
		else:
			args.family = 0

		# Call function
		ret = args.func(db, args)

		# Return with exit code
		if ret:
			sys.exit(ret)

		# Otherwise just exit
		sys.exit(0)

	def handle_lookup(self, db, ns):
		ret = 0

		for address in ns.address:
			try:
				n = db.lookup(address)
			except ValueError:
				print(_("Invalid IP address: %s") % address, file=sys.stderr)

			args = {
				"address" : address,
				"network" : n,
			}

			# Nothing found?
			if not n:
				print(_("Nothing found for %(address)s") % args, file=sys.stderr)
				ret = 1
				continue

			# Try to retrieve the AS if we have an AS number
			if n.asn:
				a = db.get_as(n.asn)

				# If we have found an AS we will print it in the message
				if a:
					args.update({
						"as" : a,
					})

					print(_("%(address)s belongs to %(network)s which is a part of %(as)s") % args)
					continue

			print(_("%(address)s belongs to %(network)s") % args)

		return ret

	def handle_get_as(self, db, ns):
		"""
			Gets information about Autonomous Systems
		"""
		ret = 0

		for asn in ns.asn:
			try:
				asn = int(asn)
			except ValueError:
				print(_("Invalid ASN: %s") % asn, file=sys.stderr)
				ret = 1
				continue

			# Fetch AS from database
			a = db.get_as(asn)

			# Nothing found
			if not a:
				print(_("Could not find AS%s") % asn, file=sys.stderr)
				ret = 1
				continue

			print(_("AS%(asn)s belongs to %(name)s") % { "asn" : a.number, "name" : a.name })

		return ret

	def handle_search_as(self, db, ns):
		for query in ns.query:
			# Print all matches ASes
			for a in db.search_as(query):
				print(a)

	def __get_output_formatter(self, ns):
		try:
			cls = self.output_formats[ns.output_format]
		except KeyError:
			cls = OutputFormatter

		return cls(ns)

	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)

	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)

	def handle_list_networks_by_flags(self, db, ns):
		flags = 0

		if ns.anonymous_proxy:
			flags |= location.NETWORK_FLAG_ANONYMOUS_PROXY

		if ns.satellite_provider:
			flags |= location.NETWORK_FLAG_SATELLITE_PROVIDER

		if ns.anycast:
			flags |= location.NETWORK_FLAG_ANYCAST

		with self.__get_output_formatter(ns) as f:
			for n in db.search_networks(flags=flags, family=ns.family):
				f.network(n)


def main():
	# Run the command line interface
	c = CLI()
	c.run()

main()
Commit	Line	Data
	1	#!/usr/bin/python3
	2	###############################################################################
	3	# #
	4	# libloc - A library to determine the location of someone on the Internet #
	5	# #
	6	# Copyright (C) 2017 IPFire Development Team <info@ipfire.org> #
	7	# #
	8	# This library is free software; you can redistribute it and/or #
	9	# modify it under the terms of the GNU Lesser General Public #
	10	# License as published by the Free Software Foundation; either #
	11	# version 2.1 of the License, or (at your option) any later version. #
	12	# #
	13	# This library is distributed in the hope that it will be useful, #
	14	# but WITHOUT ANY WARRANTY; without even the implied warranty of #
	15	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU #
	16	# Lesser General Public License for more details. #
	17	# #
	18	###############################################################################
	19
	20	import argparse
	21	import gettext
	22	import ipaddress
	23	import os
	24	import socket
	25	import sys
	26	import syslog
	27
	28	# Load our location module
	29	import location
	30
	31	# i18n
	32	def _(singular, plural=None, n=None):
	33	if plural:
	34	return gettext.dngettext("libloc", singular, plural, n)
	35
	36	return gettext.dgettext("libloc", singular)
	37
	38	# Output formatters
	39
	40	class OutputFormatter(object):
	41	def __init__(self, ns):
	42	self.ns = ns
	43
	44	def __enter__(self):
	45	# Open the output
	46	self.open()
	47
	48	return self
	49
	50	def __exit__(self, type, value, tb):
	51	if tb is None:
	52	self.close()
	53
	54	@property
	55	def name(self):
	56	if "country_code" in self.ns:
	57	return "networks_country_%s" % self.ns.country_code[0]
	58
	59	elif "asn" in self.ns:
	60	return "networks_AS%s" % self.ns.asn[0]
	61
	62	def open(self):
	63	pass
	64
	65	def close(self):
	66	pass
	67
	68	def network(self, network):
	69	print(network)
	70
	71
	72	class IpsetOutputFormatter(OutputFormatter):
	73	"""
	74	For nftables
	75	"""
	76	def open(self):
	77	print("create %s hash:net family inet hashsize 1024 maxelem 65536" % self.name)
	78
	79	def network(self, network):
	80	print("add %s %s" % (self.name, network))
	81
	82
	83	class NftablesOutputFormatter(OutputFormatter):
	84	"""
	85	For nftables
	86	"""
	87	def open(self):
	88	print("define %s = {" % self.name)
	89
	90	def close(self):
	91	print("}")
	92
	93	def network(self, network):
	94	print(" %s," % network)
	95
	96
	97	class XTGeoIPOutputFormatter(OutputFormatter):
	98	"""
	99	Formats the output in that way, that it can be loaded by
	100	the xt_geoip kernel module from xtables-addons.
	101	"""
	102	def network(self, network):
	103	n = ipaddress.ip_network("%s" % network)
	104
	105	for address in (n.network_address, n.broadcast_address):
	106	bytes = socket.inet_pton(
	107	socket.AF_INET6 if address.version == 6 else socket.AF_INET,
	108	"%s" % address,
	109	)
	110
	111	os.write(1, bytes)
	112
	113
	114	class CLI(object):
	115	output_formats = {
	116	"ipset" : IpsetOutputFormatter,
	117	"list" : OutputFormatter,
	118	"nftables" : NftablesOutputFormatter,
	119	"xt_geoip" : XTGeoIPOutputFormatter,
	120	}
	121
	122	def parse_cli(self):
	123	parser = argparse.ArgumentParser(
	124	description=_("Location Database Command Line Interface"),
	125	)
	126	subparsers = parser.add_subparsers()
	127
	128	# Global configuration flags
	129	parser.add_argument("--debug", action="store_true",
	130	help=_("Enable debug output"))
	131
	132	# version
	133	parser.add_argument("--version", action="version",
	134	version="%%(prog)s %s" % location.__version__)
	135
	136	# database
	137	parser.add_argument("--database", "-d",
	138	default="@databasedir@/database.db", help=_("Path to database"),
	139	)
	140
	141	# public key
	142	parser.add_argument("--public-key", "-k",
	143	default="@databasedir@/signing-key.pem", help=_("Public Signing Key"),
	144	)
	145
	146	# lookup an IP address
	147	lookup = subparsers.add_parser("lookup",
	148	help=_("Lookup one or multiple IP addresses"),
	149	)
	150	lookup.add_argument("address", nargs="+")
	151	lookup.set_defaults(func=self.handle_lookup)
	152
	153	# Get AS
	154	get_as = subparsers.add_parser("get-as",
	155	help=_("Get information about one or multiple Autonomous Systems"),
	156	)
	157	get_as.add_argument("asn", nargs="+")
	158	get_as.set_defaults(func=self.handle_get_as)
	159
	160	# Search for AS
	161	search_as = subparsers.add_parser("search-as",
	162	help=_("Search for Autonomous Systems that match the string"),
	163	)
	164	search_as.add_argument("query", nargs=1)
	165	search_as.set_defaults(func=self.handle_search_as)
	166
	167	# List all networks in an AS
	168	list_networks_by_as = subparsers.add_parser("list-networks-by-as",
	169	help=_("Lists all networks in an AS"),
	170	)
	171	list_networks_by_as.add_argument("asn", nargs=1, type=int)
	172	list_networks_by_as.add_argument("--family", choices=("ipv6", "ipv4"))
	173	list_networks_by_as.add_argument("--output-format",
	174	choices=self.output_formats.keys(), default="list")
	175	list_networks_by_as.set_defaults(func=self.handle_list_networks_by_as)
	176
	177	# List all networks in a country
	178	list_networks_by_cc = subparsers.add_parser("list-networks-by-cc",
	179	help=_("Lists all networks in a country"),
	180	)
	181	list_networks_by_cc.add_argument("country_code", nargs=1)
	182	list_networks_by_cc.add_argument("--family", choices=("ipv6", "ipv4"))
	183	list_networks_by_cc.add_argument("--output-format",
	184	choices=self.output_formats.keys(), default="list")
	185	list_networks_by_cc.set_defaults(func=self.handle_list_networks_by_cc)
	186
	187	# List all networks with flags
	188	list_networks_by_flags = subparsers.add_parser("list-networks-by-flags",
	189	help=_("Lists all networks with flags"),
	190	)
	191	list_networks_by_flags.add_argument("--anonymous-proxy",
	192	action="store_true", help=_("Anonymous Proxies"),
	193	)
	194	list_networks_by_flags.add_argument("--satellite-provider",
	195	action="store_true", help=_("Satellite Providers"),
	196	)
	197	list_networks_by_flags.add_argument("--anycast",
	198	action="store_true", help=_("Anycasts"),
	199	)
	200	list_networks_by_flags.add_argument("--family", choices=("ipv6", "ipv4"))
	201	list_networks_by_flags.add_argument("--output-format",
	202	choices=self.output_formats.keys(), default="list")
	203	list_networks_by_flags.set_defaults(func=self.handle_list_networks_by_flags)
	204
	205	args = parser.parse_args()
	206
	207	# Print usage if no action was given
	208	if not "func" in args:
	209	parser.print_usage()
	210	sys.exit(2)
	211
	212	return args
	213
	214	def run(self):
	215	# Parse command line arguments
	216	args = self.parse_cli()
	217
	218	# Open database
	219	try:
	220	db = location.Database(args.database)
	221	except FileNotFoundError as e:
	222	sys.stderr.write("location-query: Could not open database %s: %s\n" \
	223	% (args.database, e))
	224	sys.exit(1)
	225
	226	# Translate family
	227	if args.family == "ipv6":
	228	args.family = socket.AF_INET6
	229	elif args.family == "ipv4":
	230	args.family = socket.AF_INET
	231	else:
	232	args.family = 0
	233
	234	# Call function
	235	ret = args.func(db, args)
	236
	237	# Return with exit code
	238	if ret:
	239	sys.exit(ret)
	240
	241	# Otherwise just exit
	242	sys.exit(0)
	243
	244	def handle_lookup(self, db, ns):
	245	ret = 0
	246
	247	for address in ns.address:
	248	try:
	249	n = db.lookup(address)
	250	except ValueError:
	251	print(_("Invalid IP address: %s") % address, file=sys.stderr)
	252
	253	args = {
	254	"address" : address,
	255	"network" : n,
	256	}
	257
	258	# Nothing found?
	259	if not n:
	260	print(_("Nothing found for %(address)s") % args, file=sys.stderr)
	261	ret = 1
	262	continue
	263
	264	# Try to retrieve the AS if we have an AS number
	265	if n.asn:
	266	a = db.get_as(n.asn)
	267
	268	# If we have found an AS we will print it in the message
	269	if a:
	270	args.update({
	271	"as" : a,
	272	})
	273
	274	print(_("%(address)s belongs to %(network)s which is a part of %(as)s") % args)
	275	continue
	276
	277	print(_("%(address)s belongs to %(network)s") % args)
	278
	279	return ret
	280
	281	def handle_get_as(self, db, ns):
	282	"""
	283	Gets information about Autonomous Systems
	284	"""
	285	ret = 0
	286
	287	for asn in ns.asn:
	288	try:
	289	asn = int(asn)
	290	except ValueError:
	291	print(_("Invalid ASN: %s") % asn, file=sys.stderr)
	292	ret = 1
	293	continue
	294
	295	# Fetch AS from database
	296	a = db.get_as(asn)
	297
	298	# Nothing found
	299	if not a:
	300	print(_("Could not find AS%s") % asn, file=sys.stderr)
	301	ret = 1
	302	continue
	303
	304	print(_("AS%(asn)s belongs to %(name)s") % { "asn" : a.number, "name" : a.name })
	305
	306	return ret
	307
	308	def handle_search_as(self, db, ns):
	309	for query in ns.query:
	310	# Print all matches ASes
	311	for a in db.search_as(query):
	312	print(a)
	313
	314	def __get_output_formatter(self, ns):
	315	try:
	316	cls = self.output_formats[ns.output_format]
	317	except KeyError:
	318	cls = OutputFormatter
	319
	320	return cls(ns)
	321
	322	def handle_list_networks_by_as(self, db, ns):
	323	with self.__get_output_formatter(ns) as f:
	324	for asn in ns.asn:
	325	# Print all matching networks
	326	for n in db.search_networks(asn=asn, family=ns.family):
	327	f.network(n)
	328
	329	def handle_list_networks_by_cc(self, db, ns):
	330	with self.__get_output_formatter(ns) as f:
	331	for country_code in ns.country_code:
	332	# Print all matching networks
	333	for n in db.search_networks(country_code=country_code, family=ns.family):
	334	f.network(n)
	335
	336	def handle_list_networks_by_flags(self, db, ns):
	337	flags = 0
	338
	339	if ns.anonymous_proxy:
	340	flags \|= location.NETWORK_FLAG_ANONYMOUS_PROXY
	341
	342	if ns.satellite_provider:
	343	flags \|= location.NETWORK_FLAG_SATELLITE_PROVIDER
	344
	345	if ns.anycast:
	346	flags \|= location.NETWORK_FLAG_ANYCAST
	347
	348	with self.__get_output_formatter(ns) as f:
	349	for n in db.search_networks(flags=flags, family=ns.family):
	350	f.network(n)
	351
	352
	353	def main():
	354	# Run the command line interface
	355	c = CLI()
	356	c.run()
	357
	358	main()