22

✏️ Editing: geo.py

from contextlib import contextmanager
from ipaddress import IPv4Address, IPv4Network, IPv6Address, IPv6Network
from typing import Union

import geoip2.database
from geoip2.errors import AddressNotFoundError

from defence360agent.contracts.config import CountryInfo
from defence360agent.utils.validate import IP

class Reader:
    def __init__(self, geoip2_reader):
        self._geoip2_reader = geoip2_reader

def get(
        self,
        address: Union[
            str, IPv4Address, IPv4Network, IPv6Address, IPv6Network
        ],
    ):
        """
        Returns geo country information from max mind's db request
        :param address: ip or network address
        e.g. '4.4.4.4, 1.2.0.0/16, 2001:678:4c::/48'
        :return: maxmind's geo info
        """
        try:
            ip = IP.adopt_to_ipvX_network(address)
        except ValueError:
            return None

try:
            obj = self._geoip2_reader.country(str(ip.network_address))
        except AddressNotFoundError:
            return None
        # According to documentation:
        #     https://geoip2.readthedocs.io/en/latest/#what-data-is-returned
        # (...) MaxMind does not always have every piece of data for any given
        # IP address. Because of these factors, it is possible for any request
        # to return a record where some or all of the attributes are
        # unpopulated. (...)
        return obj.country if obj else None

def get_id(
        self,
        address: Union[
            str, IPv4Address, IPv4Network, IPv6Address, IPv6Network
        ],
    ):
        """
        :param address: valid ipv4 address
        :return: maxmind's id of the country
        """
        country_info = self.get(address)
        if country_info:
            return country_info.geoname_id
        return None

def get_code(
        self,
        address: Union[
            str, IPv4Address, IPv4Network, IPv6Address, IPv6Network
        ],
    ):
        """
        :param address: valid ipv4 address
        :return: country code in ISO-3166 format
        """
        country_info = self.get(address)
        if country_info:
            return country_info.iso_code
        return None

@contextmanager
def reader():
    """
    :return Reader obj: instance to be reused to it's method calls
    """
    with geoip2.database.Reader(CountryInfo.DB) as geoip2_reader:
        yield Reader(geoip2_reader)

MR SKB