File geoiplookup-mb of Package mirrorbrain

#!/usr/bin/python3

import maxminddb
import os
import sys
import socket

class MirrorBrainHost:
    """represent an IP address, or rather some data associated with it"""

    def __init__(self, address, maxmind_asn_db='/var/lib/GeoIP/GeoLite2-ASN.mmdb', maxmind_city_db='/var/lib/GeoIP/GeoLite2-City.mmdb'):
        self.address = address
        self.ip = address
        self.ip6 = None
        self.asn = None
        self.asn6 = None
        self.prefix = None
        self.prefix6 = None
        self.city_info = None

        self.maxmind_asn_db = maxmind_asn_db
        self.maxmind_city_db = maxmind_city_db

        self._resolv_address(address)
        self._find_city_info()
        self._find_asn()


    def country_code(self):
        return self.city_info['country']['iso_code']


    def region_code(self):
        return self.city_info['continent']['code']


    def coordinates(self):
        lat = round(self.city_info['location']['latitude'],  3)
        lng = round(self.city_info['location']['longitude'], 3)
        return lat, lng


    def ipv6Only(self):
        if self.ip6 and not self.ip:
            return True
        else:
            return False


    def __str__(self):
        r = []
        if self.ip:
            r.append('%s (%s AS%s)' % (self.ip, self.prefix, self.asn))
        if self.ip6:
            r.append('%s (%s AS%s)' % (self.ip6, self.prefix6, self.asn6))
        return ' '.join(r)


    def _find_city_info(self):
        self.city_db = maxminddb.open_database(self.maxmind_city_db)

        if self.ip:
            try:
                self.city_info = self.city_db.get(self.ip)
            except Exception:
                # we get this error if mod_asn isn't installed as well
                pass

        if self.ip6:
            try:
                self.city_info = self.city_db.get(self.ip6)
            except Exception:
                # we get this error if mod_asn isn't installed as well
                pass


    def _find_asn(self):
        self.asn_db = maxminddb.open_database(self.maxmind_asn_db)

        if self.ip:
            try:
                res = self.asn_db.get(self.ip)
                self.prefix = "TODO"
                self.asn = self.asn_db.get(self.ip)['autonomous_system_number']
            except Exception:
                pass

        if self.ip6:
            try:
                res = self.asn_db.get(self.ip6)
                self.prefix6 =  "TODO"
                self.asn6 = self.asn_db.get(self.ip)['autonomous_system_number']
            except Exception:
                pass


    def _resolv_address(self, s):
        ips = []
        ip6s = []
        try:
            for res in socket.getaddrinfo(s, None):
                af, socktype, proto, canonname, sa = res
                if af == socket.AF_INET6:
                    if sa[0] not in ip6s:
                        ip6s.append(sa[0])
                else:
                    if sa[0] not in ips:
                        ips.append(sa[0])
        except socket.error as e:
            if e[0] == socket.EAI_NONAME:
                raise mb.mberr.NameOrServiceNotKnown(s)
            else:
                print('socket error msg:', str(e))
                return None

        # print (ips)
        # print (ip6s)
        if len(ips) > 1 or len(ip6s) > 1:
            print('>>> warning: %r resolves to multiple IP addresses: ' % s, file=sys.stderr)
            if len(ips) > 1:
                print(', '.join(ips), file=sys.stderr)
            if len(ip6s) > 1:
                print(', '.join(ip6s), file=sys.stderr)
            print('\n>>> see http://mirrorbrain.org/archive/mirrorbrain/0042.html why this could\n'
                  '>>> could be a problem, and what to do about it. But note that this is not\n'
                  '>>> necessarily a problem and could actually be intended depending on the\n'
                  '>>> mirror\'s configuration (see http://mirrorbrain.org/issues/issue152).\n'
                  '>>> It\'s best to talk to the mirror\'s admins.\n', file=sys.stderr)
        if ips:
            self.ip = ips[0]
        if ip6s:
            self.ip6 = ip6s[0]


script_name = os.path.basename(__file__)
address = sys.argv[3]

mb_host = MirrorBrainHost(address)
lat, lng = mb_host.coordinates()

if script_name == 'geoiplookup_contintent':
  print(mb_host.country_code())

if script_name == 'geoiplookup_city':
  print('Continent: ', mb_host.region_code() )
  print('Country:   ', mb_host.country_code() )
  print('Region id:  00')
  print('Region:     (null)')
  print('City:       (null)')
  print('Latitude:  ', lat )
  print('Longitude: ', lng )
openSUSE Build Service is sponsored by