dnmgmt/whoisdb/helpers.py

168 lines
4.6 KiB
Python
Raw Normal View History

2017-03-22 04:06:24 +01:00
import whoisdb.models
2017-03-21 02:36:07 +01:00
import domains.models
2017-03-20 00:16:33 +01:00
2017-03-31 00:50:00 +02:00
from django.db.models import F, When, Max, Case, IntegerField
2017-03-28 01:43:24 +02:00
2017-03-27 12:18:02 +02:00
import ipaddress
2017-03-27 13:29:32 +02:00
import re
2017-03-22 04:06:24 +01:00
2017-03-20 00:16:33 +01:00
def _addFields(fields, obj, fieldNames):
for fieldName in fieldNames:
2017-03-21 02:36:07 +01:00
fields.append((fieldName.capitalize().replace("_", " "), getattr(obj, fieldName)))
2017-03-20 00:16:33 +01:00
def getWhoisObjectFields(obj, owner):
fields = []
2017-03-22 04:06:24 +01:00
if getattr(obj, "handle", None):
2017-03-20 00:16:33 +01:00
_addFields(fields, obj, ["handle"])
c = type(obj)
if c == whoisdb.models.Maintainer:
2017-03-27 02:22:51 +02:00
_addFields(fields, obj, ["description", "admin_c"])
2017-03-20 00:16:33 +01:00
if owner:
_addFields(fields, obj, ["auth"])
elif c == whoisdb.models.Contact:
_addFields(fields, obj, ["name", "mnt_by"])
elif c == whoisdb.models.ASBlock:
_addFields(fields, obj, ["name"])
fields.append(("AS Range", "%s - %s" % (obj.asBegin, obj.asEnd)))
2017-03-21 19:34:02 +01:00
_addFields(fields, obj, ["description", "parent_block", "mnt_by", "mnt_lower", "admin_c"])
2017-03-20 00:16:33 +01:00
elif c == whoisdb.models.ASNumber:
2017-03-27 12:18:02 +02:00
_addFields(fields, obj, ["name", "number", "description", "asblock", "volatile", "mnt_by", "mnt_lower", "admin_c"])
2017-03-20 00:16:33 +01:00
elif c == whoisdb.models.InetNum:
_addFields(fields, obj, ["name"])
fields.append(("Address CIDR", obj.prefix()))
_addFields(fields, obj, ["description", "parent_range", "origin_as", "mnt_by", "mnt_lower", "admin_c"])
2017-03-21 02:36:07 +01:00
elif c == domains.models.Domain:
_addFields(fields, obj, ["name", "nameservers", "mnt_by", "admin_c"])
elif c == domains.models.Nameserver:
_addFields(fields, obj, ["name", "glueIPv4", "glueIPv6", "mnt_by", "admin_c"])
elif c == domains.models.ReverseZone:
2017-03-22 04:06:24 +01:00
#_addFields(fields, obj, ["name"])
2017-03-21 02:36:07 +01:00
fields.append(("Address CIDR", obj.prefix()))
_addFields(fields, obj, ["parentNet", "nameservers"])
2017-03-20 00:16:33 +01:00
2017-03-31 03:20:23 +02:00
_addFields(fields, obj, ["created", "last_modified"])
2017-03-20 00:16:33 +01:00
return fields
2017-03-22 04:06:24 +01:00
2017-03-21 19:34:02 +01:00
def guessWhoisObject(self, handle):
# is it a normal handle?
pass
2017-03-27 12:18:02 +02:00
def findInDatabase(rawValue):
# is this an ip address?
2017-03-27 13:29:32 +02:00
rawValue = rawValue.strip().upper()
2017-03-27 12:18:02 +02:00
value = None
2017-03-27 13:29:32 +02:00
results = []
# try subnetwork
2017-03-27 12:18:02 +02:00
try:
value = ipaddress.ip_network(rawValue, strict=False)
except ValueError:
2017-03-27 13:29:32 +02:00
pass
if value:
# ssubnet
obj = whoisdb.models.InetNum.objects.filter(address=str(value.network_address), netmask=value.prefixlen)
results.extend(obj)
# single ip
value = None
try:
value = ipaddress.ip_address(rawValue)
except ValueError:
pass
if value:
2017-03-28 01:43:24 +02:00
# NOTE: this is only for "small subnets", we could increase this...
baseaddr = None
if value.version == 4:
basenet = ipaddress.ip_network("%s/24" % value, strict=False)
baseaddr = ".".join(str(basenet).split(".")[0:3]) + "."
else:
basenet = ipaddress.ip_network("%s/56" % value.exploded, strict=False)
baseaddr = ":".join(str(basenet).split(":")[0:4])[-2]
nets = whoisdb.models.InetNum.objects.filter(address__startswith=baseaddr).order_by("-netmask")
for net in nets:
if value in net.getNetwork():
results.append(net)
break
2017-03-27 13:29:32 +02:00
# asnumber?
m = re.match("^(?:AS)?(\d+)$", rawValue)
if m:
# asnumber!
2017-03-28 01:43:24 +02:00
num = int(m.group(1))
obj = whoisdb.models.ASNumber.objects.filter(number=num)
2017-03-27 13:29:32 +02:00
results.extend(obj)
2017-03-28 01:43:24 +02:00
# find a matching block
blocks = whoisdb.models.ASBlock.objects.filter(asBegin__lte=num, asEnd__gte=num).annotate(size=F('asEnd')-F('asBegin')).order_by('size')
if blocks.count() > 0:
results.append(blocks[0])
2017-03-27 13:29:32 +02:00
# asblocks? smallest asblock containing the range
# WHEN anotation foo... could also be done in asnumber match
# also look for number - number queries?
# domain?
if rawValue.endswith("DN") or rawValue.endswith("DN."):
2017-03-28 13:27:21 +02:00
value = rawValue.lower()
2017-03-27 13:29:32 +02:00
if not value.endswith("."):
value += "."
obj = domains.models.Domain.objects.filter(name=value)
2017-03-27 13:29:32 +02:00
results.extend(obj)
# contact by name?
# handlenames for Maintainer, Contact, InetNum, ASNumber, ASBlock
handleObjs = [
whoisdb.models.Contact,
whoisdb.models.Maintainer,
whoisdb.models.InetNum,
whoisdb.models.ASBlock,
whoisdb.models.ASNumber,
]
for handleObj in handleObjs:
2017-03-27 13:36:03 +02:00
obj = handleObj.objects.filter(handle=rawValue)
if not obj and len(rawValue) >= 3:
obj = handleObj.objects.filter(handle__startswith=rawValue)
results.extend(obj)
2017-03-27 13:29:32 +02:00
return results
2017-03-28 01:56:10 +02:00
def findHandleFromStr(rawValue):
handleObjs = [
whoisdb.models.Contact,
whoisdb.models.Maintainer,
whoisdb.models.InetNum,
whoisdb.models.ASBlock,
whoisdb.models.ASNumber,
]
for handleObj in handleObjs:
try:
return handleObj.objects.get(handle=rawValue)
except handleObj.DoesNotExist:
pass
return None
2017-03-31 00:50:00 +02:00
def orderQueryset(qs, userOwned, objOwned):
# when for
whens = [When(userOwned, then=2)]
if objOwned:
# add existing
whens.append(When(objOwned, then=1))
qs = qs.annotate(card=Max(Case(*whens, default=0, output_field=IntegerField()))).order_by("-card")
return qs