Viewing file:  087_ignored_by_port_fields.py (1.73 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
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"
)
|