import logging from ipaddress import ip_network logger = logging.getLogger(__name__) def migrate(migrator, database, fake=False, **kwargs): IgnoredByPort = migrator.orm["ignored_by_port_proto"] try: from im360.utils.net import pack_ip_network except ImportError: ips = [] # keep database structure in AV too else: q = IgnoredByPort.select(IgnoredByPort.ip).distinct().tuples() ips = [ip for ip, in q] for ip in ips: try: net, mask, version = pack_ip_network(ip_network(ip)) except ValueError: logger.warning("Invalid IP network %s", ip) IgnoredByPort.delete().where(IgnoredByPort.ip == ip).execute() else: IgnoredByPort.update( network_address=net, netmask=mask, version=version ).where(IgnoredByPort.ip == ip).execute() if ips: try: from im360.internals import geo with geo.reader() as geo_reader: for ip in ips: country = geo_reader.get_id(ip) IgnoredByPort.update( country=country, ).where(IgnoredByPort.ip == ip).execute() except (ImportError, OSError): logger.warning( "Failed to update countries data in ignored_by_port" ) migrator.add_not_null(IgnoredByPort, "network_address") migrator.add_not_null( IgnoredByPort, "netmask", ) migrator.add_not_null(IgnoredByPort, "version") def rollback(migrator, database, fake=False, **kwargs): IgnoredByPort = migrator.orm["ignored_by_port_proto"] migrator.drop_not_null( IgnoredByPort, "network_address", "netmask", "version" )