22

✏️ Editing: 111_ignore_list_ip_as_int.py

import ipaddress

def migrate(migrator, database, fake=False, **kwargs):
    if fake:
        return

IgnoreListNew = migrator.orm["ignore_list_new"]
    IgnoreList = migrator.orm["ignore_list"]
    try:
        from defence360agent.utils.validate import IP
        from im360.utils.net import pack_ip_network
    except ImportError:
        pass
    else:
        with database.atomic():
            # FIXME: after migrating to peewee 3 add .iterator()
            ip_strings = [item["ip"] for item in IgnoreList.select().dicts()]
            ips = set()
            for item in ip_strings:
                try:
                    ip = ipaddress.ip_network(item)
                except ValueError:
                    # malformed ip
                    continue
                ips.add(ip)
            for ip in ips:
                net, mask, version = pack_ip_network(ip)
                IgnoreListNew.create(
                    ip=IP.ip_net_to_string(ip),
                    network_address=net,
                    netmask=mask,
                    version=version,
                )

migrator.sql("DROP TABLE ignore_list")
    migrator.sql("ALTER TABLE ignore_list_new RENAME TO ignore_list")

def rollback(migrator, database, fake=False, **kwargs):
    """Write your rollback migrations here."""
    pass

MR SKB