mirror of
https://github.com/peeringdb/peeringdb.git
synced 2024-05-11 05:55:09 +00:00
146 lines
4.6 KiB
Python
146 lines
4.6 KiB
Python
import json
|
|
import base64
|
|
|
|
from django.http import JsonResponse, HttpResponse
|
|
from django.conf import settings
|
|
from django.utils.translation import ugettext_lazy as _
|
|
from django.contrib.auth import authenticate
|
|
|
|
from django_namespace_perms.util import has_perms
|
|
from ratelimit.decorators import ratelimit, is_ratelimited
|
|
|
|
from peeringdb_server import ixf
|
|
from peeringdb_server.models import (
|
|
IXLan, Network, NetworkIXLan,
|
|
)
|
|
|
|
RATELIMITS = settings.RATELIMITS
|
|
|
|
def enable_basic_auth(fn):
|
|
"""
|
|
a simple decorator to enable basic auth for a specific view
|
|
"""
|
|
def wrapped(request, *args, **kwargs):
|
|
if 'HTTP_AUTHORIZATION' in request.META:
|
|
auth = request.META['HTTP_AUTHORIZATION'].split()
|
|
if len(auth) == 2:
|
|
if auth[0].lower() == "basic":
|
|
username, password = base64.b64decode(auth[1]).split(':', 1)
|
|
request.user = authenticate(username=username, password=password)
|
|
if not request.user:
|
|
return JsonResponse({"non_field_errors":["Invalid credentials"]}, status=401)
|
|
return fn(request, *args, **kwargs)
|
|
return wrapped
|
|
|
|
def pretty_response(data):
|
|
return HttpResponse(
|
|
json.dumps(data, indent=2), content_type="application/json")
|
|
|
|
def error_response(msg, status=400):
|
|
return JsonResponse({"non_field_errors":[msg]},status=status)
|
|
|
|
|
|
|
|
@ratelimit(key="ip", rate=RATELIMITS["view_import_ixlan_ixf_preview"], group="ixf_preview")
|
|
@enable_basic_auth
|
|
def view_import_ixlan_ixf_preview(request, ixlan_id):
|
|
|
|
# check if request was blocked by rate limiting
|
|
was_limited = getattr(request, "limited", False)
|
|
if was_limited:
|
|
return error_response(_("Please wait a bit before requesting " \
|
|
"another ixf import preview."), status=400)
|
|
|
|
try:
|
|
ixlan = IXLan.objects.get(id=ixlan_id)
|
|
except IXLan.DoesNotExist:
|
|
return error_response(_("Ixlan not found"), status=404)
|
|
|
|
if not has_perms(request.user, ixlan, "update"):
|
|
return error_response(_("Permission denied"), status=403)
|
|
|
|
importer = ixf.Importer()
|
|
importer.update(ixlan, save=False)
|
|
|
|
return pretty_response(importer.log)
|
|
|
|
|
|
@ratelimit(key="ip", rate=RATELIMITS["view_import_net_ixf_postmortem"], group="ixf_postmortem")
|
|
@enable_basic_auth
|
|
def view_import_net_ixf_postmortem(request, net_id):
|
|
|
|
# check if request was blocked by rate limiting
|
|
|
|
was_limited = getattr(request, "limited", False)
|
|
if was_limited:
|
|
return error_response(_("Please wait a bit before requesting " \
|
|
"another IX-F import postmortem."), status=400)
|
|
|
|
# load net
|
|
|
|
try:
|
|
net = Network.objects.get(id=net_id, status="ok")
|
|
except Network.DoesNotExist:
|
|
return error_response(_("Network not found"), status=404)
|
|
|
|
if not has_perms(request.user, net, "update"):
|
|
return error_response(_("Permission denied"), status=403)
|
|
|
|
# make sure limit is within bounds and a valid number
|
|
|
|
try:
|
|
limit = int(request.GET.get("limit", 25))
|
|
except:
|
|
limit = 25
|
|
|
|
|
|
errors = []
|
|
|
|
if limit < 1:
|
|
limit = 1
|
|
|
|
elif limit > settings.IXF_POSTMORTEM_LIMIT:
|
|
errors.append(_("Postmortem length cannot exceed {} entries").format(
|
|
settings.IXF_POSTMORTEM_LIMIT))
|
|
|
|
post_mortem = ixf.PostMortem()
|
|
log = post_mortem.generate(net.asn, limit=limit)
|
|
|
|
return pretty_response({"data":log, "non_field_errors":errors})
|
|
|
|
|
|
@ratelimit(key="ip", rate=RATELIMITS["view_import_ixlan_ixf_preview"], group="ixf_preview")
|
|
@enable_basic_auth
|
|
def view_import_net_ixf_preview(request, net_id):
|
|
|
|
# check if request was blocked by rate limiting
|
|
was_limited = getattr(request, "limited", False)
|
|
if was_limited:
|
|
return error_response(_("Please wait a bit before requesting " \
|
|
"another ixf import preview."), status=400)
|
|
|
|
try:
|
|
net = Network.objects.get(id=net_id, status="ok")
|
|
except Network.DoesNotExist:
|
|
return error_response(_("Network not found"), status=404)
|
|
|
|
if not has_perms(request.user, net, "update"):
|
|
return error_response(_("Permission denied"), status=403)
|
|
|
|
|
|
total_log = {"data":[], "errors":[]}
|
|
|
|
for ixlan in net.ixlan_set_ixf_enabled:
|
|
importer = ixf.Importer()
|
|
importer.cache_only=True
|
|
success, netixlans, deleted, log = importer.update(ixlan, asn=net.asn, save=False)
|
|
total_log["data"].extend(log["data"])
|
|
total_log["errors"].extend(["{}({}): {}".format(ixlan.ix.name, ixlan.id, err)
|
|
for err in log["errors"]])
|
|
|
|
|
|
return pretty_response(total_log)
|
|
|
|
|
|
|