mirror of
https://github.com/peeringdb/peeringdb.git
synced 2024-05-11 05:55:09 +00:00
1076 lines
30 KiB
Python
1076 lines
30 KiB
Python
import ipaddress
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import requests
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.cache import cache
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.management import call_command
|
|
from django.test import RequestFactory, override_settings
|
|
from rest_framework.exceptions import ValidationError as RestValidationError
|
|
|
|
import peeringdb_server.geo as geo
|
|
from peeringdb_server.context import current_request
|
|
from peeringdb_server.models import (
|
|
Facility,
|
|
InternetExchange,
|
|
IXLan,
|
|
IXLanPrefix,
|
|
Network,
|
|
NetworkContact,
|
|
NetworkIXLan,
|
|
Organization,
|
|
ProtectedAction,
|
|
)
|
|
from peeringdb_server.validators import (
|
|
validate_address_space,
|
|
validate_asn_prefix,
|
|
validate_distance_geocode,
|
|
validate_info_prefixes4,
|
|
validate_info_prefixes6,
|
|
validate_irr_as_set,
|
|
validate_latitude,
|
|
validate_longitude,
|
|
validate_phonenumber,
|
|
validate_prefix_overlap,
|
|
validate_social_media,
|
|
validate_website_override,
|
|
)
|
|
from tests.test_ixf_member_import_protocol import setup_test_data
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
INVALID_ADDRESS_SPACES = [
|
|
"0.0.0.0/1",
|
|
"0.0.0.0/8",
|
|
"10.0.0.0/8",
|
|
"127.0.0.0/8",
|
|
"169.254.0.0/16",
|
|
# FIXME: this fails still
|
|
#'172.0.0.0/11',
|
|
"172.16.0.0/12",
|
|
"192.0.2.0/24",
|
|
"192.168.0.0/16",
|
|
"198.18.0.0/15",
|
|
"198.51.100.0/24",
|
|
"203.0.113.0/24",
|
|
# FIXME: this fails still
|
|
#'224.0.0.0/3',
|
|
"224.0.0.0/4",
|
|
"240.0.0.0/4",
|
|
"100.64.0.0/10",
|
|
"0000::/8",
|
|
"0064:ff9b::/96",
|
|
"0100::/8",
|
|
"0200::/7",
|
|
"0400::/6",
|
|
"0800::/5",
|
|
"1000::/4",
|
|
"2001::/33",
|
|
"2001:0:8000::/33",
|
|
"2001:0002::/48",
|
|
"2001:0003::/32",
|
|
"2001:10::/28",
|
|
"2001:20::/28",
|
|
"2001:db8::/32",
|
|
"2002::/16",
|
|
"3ffe::/16",
|
|
"4000::/2",
|
|
"4000::/3",
|
|
"5f00::/8",
|
|
"6000::/3",
|
|
"8000::/3",
|
|
"a000::/3",
|
|
"c000::/3",
|
|
"e000::/4",
|
|
"f000::/5",
|
|
"f800::/6",
|
|
"fc00::/7",
|
|
"fe80::/10",
|
|
"fec0::/10",
|
|
"ff00::/8",
|
|
]
|
|
|
|
|
|
@pytest.fixture(params=INVALID_ADDRESS_SPACES)
|
|
def prefix(request):
|
|
return request.param
|
|
|
|
|
|
# @pytest.mark.django_db
|
|
def test_validate_address_space(prefix):
|
|
"""
|
|
Tests peeringdb_server.validators.validate_address_space
|
|
"""
|
|
with pytest.raises(ValidationError) as exc:
|
|
validate_address_space(ipaddress.ip_network(str(prefix)))
|
|
|
|
|
|
@override_settings(DATA_QUALITY_MAX_PREFIX_V4_LIMIT=500000)
|
|
def test_validate_info_prefixes4():
|
|
"""
|
|
Tests peeringdb_server.validators.validate_info_prefixes4
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
validate_info_prefixes4(500001)
|
|
with pytest.raises(ValidationError):
|
|
validate_info_prefixes4(-1)
|
|
validate_info_prefixes4(500000)
|
|
assert validate_info_prefixes4(None) == 0
|
|
assert validate_info_prefixes4("") == 0
|
|
|
|
|
|
@override_settings(DATA_QUALITY_MAX_PREFIX_V6_LIMIT=500000)
|
|
def test_validate_info_prefixes6():
|
|
"""
|
|
Tests peeringdb_server.validators.validate_info_prefixes6
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
validate_info_prefixes6(500001)
|
|
with pytest.raises(ValidationError):
|
|
validate_info_prefixes6(-1)
|
|
validate_info_prefixes6(500000)
|
|
assert validate_info_prefixes6(None) == 0
|
|
assert validate_info_prefixes6("") == 0
|
|
|
|
|
|
@override_settings(
|
|
DATA_QUALITY_MIN_PREFIXLEN_V4=24,
|
|
DATA_QUALITY_MAX_PREFIXLEN_V4=24,
|
|
DATA_QUALITY_MIN_PREFIXLEN_V6=48,
|
|
DATA_QUALITY_MAX_PREFIXLEN_V6=48,
|
|
)
|
|
def test_validate_prefixlen():
|
|
"""
|
|
Tests prefix length limits
|
|
"""
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("37.77.32.0/20")
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("131.72.77.240/28")
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("2403:c240::/32")
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("2001:504:0:2::/64")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_validate_prefix_overlap():
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(name="Text exchange", status="ok", org=org)
|
|
ixlan = ix.ixlan
|
|
|
|
pfx1 = IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
protocol="IPv4",
|
|
prefix=ipaddress.ip_network("198.32.125.0/24"),
|
|
status="ok",
|
|
)
|
|
|
|
with pytest.raises(ValidationError) as exc:
|
|
validate_prefix_overlap("198.32.124.0/23")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value,validated",
|
|
[
|
|
# success validation
|
|
("RIPE::AS-FOO", "RIPE::AS-FOO"),
|
|
("AS-FOO@RIPE", "AS-FOO@RIPE"),
|
|
("AS-FOO-BAR@RIPE", "AS-FOO-BAR@RIPE"),
|
|
("ripe::as-foo", "RIPE::AS-FOO"),
|
|
("as-foo@ripe", "AS-FOO@RIPE"),
|
|
("as-foo@ripe as-bar@ripe", "AS-FOO@RIPE AS-BAR@RIPE"),
|
|
("as-foo@ripe,as-bar@ripe", "AS-FOO@RIPE AS-BAR@RIPE"),
|
|
("as-foo@ripe, as-bar@ripe", "AS-FOO@RIPE AS-BAR@RIPE"),
|
|
(
|
|
"RIPE::AS12345:AS-FOO RIPE::AS12345:AS-FOO:AS9876",
|
|
"RIPE::AS12345:AS-FOO RIPE::AS12345:AS-FOO:AS9876",
|
|
),
|
|
("ripe::as-foo:as123:as345", "RIPE::AS-FOO:AS123:AS345"),
|
|
("RIPE::AS12345", "RIPE::AS12345"),
|
|
("AS12345@RIPE", "AS12345@RIPE"),
|
|
("RIPE::AS123456:RS-FOO", "RIPE::AS123456:RS-FOO"),
|
|
("as-foo", "AS-FOO"),
|
|
("rs-foo", "RS-FOO"),
|
|
("as-foo as-bar", "AS-FOO AS-BAR"),
|
|
("rs-foo as-bar", "RS-FOO AS-BAR"),
|
|
("rs-foo rs-bar", "RS-FOO RS-BAR"),
|
|
("AS15562", "AS15562"),
|
|
("AS-15562", "AS-15562"),
|
|
("AS15562 AS33333", "AS15562 AS33333"),
|
|
# hyphenated source validation
|
|
# we currently do not have valid hyphentated sources in the IRR_SOURCE
|
|
# so this test is commented out
|
|
# ("AS-20C@ARIN-NONAUTH", "AS-20C@ARIN-NONAUTH"),
|
|
# ("ARIN-NONAUTH::AS-20C", "ARIN-NONAUTH::AS-20C"),
|
|
# fail validation
|
|
("UNKNOWN::AS-FOO", False),
|
|
("AS-FOO@UNKNOWN", False),
|
|
("ASFOO@UNKNOWN", False),
|
|
("UNKNOWN::ASFOO", False),
|
|
("RIPE:AS-FOO", False),
|
|
("RIPE::RS15562:RS-FOO", False),
|
|
("RIPE::AS123456:RS-FOO:AS-FOO", False),
|
|
('!"*([])?.=+/\\', False),
|
|
('RIPE::!"*([])?.=+/\\', False),
|
|
('!"*([])?.=+/\\@RIPE', False),
|
|
# > DATA_QUALITY_MAX_IRR_DEPTH
|
|
("ripe::as-foo:as123:as345:as678", False),
|
|
],
|
|
)
|
|
def test_validate_irr_as_set(value, validated):
|
|
if not validated:
|
|
with pytest.raises(ValidationError):
|
|
validate_irr_as_set(value)
|
|
else:
|
|
assert validate_irr_as_set(value) == validated
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_validate_phonenumber():
|
|
# test standalone validator
|
|
|
|
validate_phonenumber("+1 206 555 0199")
|
|
validate_phonenumber("012065550199", "US")
|
|
|
|
with pytest.raises(ValidationError):
|
|
validate_phonenumber("invalid number")
|
|
|
|
with pytest.raises(ValidationError):
|
|
validate_phonenumber("012065550199")
|
|
|
|
# test model field validation
|
|
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(
|
|
name="Text exchange",
|
|
status="ok",
|
|
org=org,
|
|
country="US",
|
|
city="Some city",
|
|
region_continent="North America",
|
|
media="Ethernet",
|
|
)
|
|
net = Network.objects.create(name="Text network", asn=12345, status="ok", org=org)
|
|
poc = NetworkContact.objects.create(network=net, status="ok", role="Abuse")
|
|
|
|
# test poc phone validation
|
|
|
|
with pytest.raises(ValidationError):
|
|
poc.phone = "invalid"
|
|
poc.full_clean()
|
|
|
|
poc.phone = "+1 206 555 0199"
|
|
poc.full_clean()
|
|
|
|
# test ix phone validation
|
|
|
|
with pytest.raises(ValidationError):
|
|
ix.tech_phone = "invalid"
|
|
ix.full_clean()
|
|
|
|
ix.tech_phone = "+1 206 555 0199"
|
|
ix.full_clean()
|
|
|
|
with pytest.raises(ValidationError):
|
|
ix.policy_phone = "invalid"
|
|
ix.full_clean()
|
|
|
|
ix.policy_phone = "+1 206 555 0199"
|
|
ix.full_clean()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_validate_ixpfx_ixlan_status_match():
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(
|
|
name="Text exchange", status="pending", org=org
|
|
)
|
|
ixlan = ix.ixlan
|
|
|
|
pfx = IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
protocol="IPv4",
|
|
prefix=ipaddress.ip_network("198.32.125.0/24"),
|
|
status="ok",
|
|
)
|
|
|
|
with pytest.raises(ValidationError) as exc1:
|
|
pfx.clean()
|
|
|
|
assert (
|
|
exc1.value.args[0]
|
|
== "IXLanPrefix with status 'ok' cannot be linked to a IXLan with status 'pending'."
|
|
)
|
|
|
|
ixlan.status = "deleted"
|
|
ixlan.save()
|
|
pfx.status = "pending"
|
|
pfx.save()
|
|
|
|
with pytest.raises(ValidationError) as exc2:
|
|
pfx.clean()
|
|
|
|
assert (
|
|
exc2.value.args[0]
|
|
== "IXLanPrefix with status 'pending' cannot be linked to a IXLan with status 'deleted'."
|
|
)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@override_settings(
|
|
DATA_QUALITY_MAX_PREFIX_V4_LIMIT=500000,
|
|
DATA_QUALITY_MAX_PREFIX_V6_LIMIT=500000,
|
|
DATA_QUALITY_MIN_PREFIXLEN_V4=24,
|
|
DATA_QUALITY_MAX_PREFIXLEN_V4=24,
|
|
DATA_QUALITY_MIN_PREFIXLEN_V6=48,
|
|
DATA_QUALITY_MAX_PREFIXLEN_V6=48,
|
|
DATA_QUALITY_MAX_IRR_DEPTH=3,
|
|
DATA_QUALITY_MIN_SPEED=10,
|
|
DATA_QUALITY_MAX_SPEED=100,
|
|
)
|
|
def test_bypass_validation():
|
|
User = get_user_model()
|
|
|
|
superuser = User.objects.create_user(
|
|
username="superuser",
|
|
password="superuser",
|
|
email="su@localhost",
|
|
is_superuser=True,
|
|
)
|
|
user = User.objects.create_user(
|
|
username="user", password="user", email="user@localhost"
|
|
)
|
|
|
|
factory = RequestFactory()
|
|
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(
|
|
name="Text exchange",
|
|
status="ok",
|
|
org=org,
|
|
country="US",
|
|
city="Some city",
|
|
region_continent="North America",
|
|
media="Ethernet",
|
|
)
|
|
net = Network.objects.create(name="Text network", asn=12345, status="ok", org=org)
|
|
|
|
# super user should bypass validation
|
|
|
|
request = factory.get("/")
|
|
request.user = superuser
|
|
with current_request(request):
|
|
validate_address_space("37.77.32.0/20")
|
|
validate_address_space("131.72.77.240/28")
|
|
validate_address_space("2403:c240::/32")
|
|
validate_address_space("2001:504:0:2::/64")
|
|
validate_info_prefixes4(500001)
|
|
validate_info_prefixes6(500001)
|
|
NetworkIXLan(speed=1, network=net, ixlan=ix.ixlan).clean()
|
|
NetworkIXLan(speed=1000, network=net, ixlan=ix.ixlan).clean()
|
|
validate_irr_as_set("ripe::as-foo:as123:as345:as678")
|
|
|
|
# user should NOT bypass validation
|
|
|
|
request = factory.get("/")
|
|
request.user = user
|
|
with current_request(request):
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("37.77.32.0/20")
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("131.72.77.240/28")
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("2403:c240::/32")
|
|
with pytest.raises(ValidationError):
|
|
validate_address_space("2001:504:0:2::/64")
|
|
with pytest.raises(ValidationError):
|
|
validate_info_prefixes4(500001)
|
|
with pytest.raises(ValidationError):
|
|
validate_info_prefixes6(500001)
|
|
with pytest.raises(ValidationError):
|
|
NetworkIXLan(speed=1, network=net, ixlan=ix.ixlan).clean()
|
|
with pytest.raises(ValidationError):
|
|
NetworkIXLan(speed=1000, network=net, ixlan=ix.ixlan).clean()
|
|
with pytest.raises(ValidationError):
|
|
validate_irr_as_set("ripe::as-foo:as123:as345:as678")
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ghost_peer_vs_real_peer_one_netixlan():
|
|
"""
|
|
Tests that a real peer can claim the ip addresses of a gohst peer. #983
|
|
|
|
In this test both ipv4 and ipv6 exist on the same netixlan.
|
|
"""
|
|
|
|
# set up entities
|
|
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
|
|
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
|
|
network_other = Network.objects.create(
|
|
asn=1010, name="AS1010", status="ok", org=org
|
|
)
|
|
ixlan = ix.ixlan
|
|
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
|
|
ixlan.save()
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="195.69.144.0/22",
|
|
protocol="IPv4",
|
|
)
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="2001:7f8:1::/64",
|
|
protocol="IPv6",
|
|
)
|
|
|
|
IP4 = "195.69.147.250"
|
|
IP6 = "2001:7f8:1::a500:2906:1"
|
|
|
|
ghost_peer = NetworkIXLan.objects.create(
|
|
network=network_other,
|
|
ixlan=ixlan,
|
|
asn=network_other.asn,
|
|
speed=20000,
|
|
ipaddr4=IP4,
|
|
ipaddr6=IP6,
|
|
status="ok",
|
|
is_rs_peer=False,
|
|
operational=False,
|
|
)
|
|
|
|
# setup IX-F cache
|
|
|
|
data = setup_test_data("ixf.member.1")
|
|
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", data)
|
|
|
|
ix = ixlan.ix
|
|
|
|
# real peer should exist in IX-F data
|
|
|
|
real4, real6 = ix.peer_exists_in_ixf_data(1001, IP4, IP6)
|
|
assert real4
|
|
assert real6
|
|
|
|
# ghost peer should NOT exist in IX-F data
|
|
|
|
ghost4, ghost6 = ix.peer_exists_in_ixf_data(1010, IP4, IP6)
|
|
assert not ghost4
|
|
assert not ghost6
|
|
|
|
# create and save a real peer that has the same ip addresses
|
|
# as the ghost peer
|
|
|
|
real_peer = NetworkIXLan(
|
|
network=network,
|
|
status="ok",
|
|
ipaddr4=IP4,
|
|
ipaddr6=IP6,
|
|
ixlan=ixlan,
|
|
speed=1000,
|
|
asn=network.asn,
|
|
)
|
|
|
|
# run full validation (this will run `validate_real_vs_ghost_peer`)
|
|
|
|
real_peer.full_clean()
|
|
real_peer.save()
|
|
|
|
# real peer has been saved and since it claimed both ip4 and ip6, the ghost
|
|
# peer is now deleted
|
|
|
|
ghost_peer.refresh_from_db()
|
|
|
|
assert ghost_peer.status == "deleted"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ghost_peer_vs_real_peer_two_netixlan():
|
|
"""
|
|
Tests that a real peer can claim the ip addresses of a gohst peer. #983
|
|
|
|
In this test both ipv4 and ipv6 exist on separate netixlans.
|
|
|
|
In this test both conflicting netixlans will have neither ipv4 nor ipv6 set in the end
|
|
and will be deleted
|
|
"""
|
|
|
|
# set up entities
|
|
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
|
|
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
|
|
network_other = Network.objects.create(
|
|
asn=1010, name="AS1010", status="ok", org=org
|
|
)
|
|
ixlan = ix.ixlan
|
|
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
|
|
ixlan.save()
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="195.69.144.0/22",
|
|
protocol="IPv4",
|
|
)
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="2001:7f8:1::/64",
|
|
protocol="IPv6",
|
|
)
|
|
|
|
IP4 = "195.69.147.250"
|
|
IP6 = "2001:7f8:1::a500:2906:1"
|
|
|
|
ghost_peer_a = NetworkIXLan.objects.create(
|
|
network=network_other,
|
|
ixlan=ixlan,
|
|
asn=network_other.asn,
|
|
speed=20000,
|
|
ipaddr4=IP4,
|
|
ipaddr6=None,
|
|
status="ok",
|
|
is_rs_peer=False,
|
|
operational=False,
|
|
)
|
|
|
|
ghost_peer_b = NetworkIXLan.objects.create(
|
|
network=network_other,
|
|
ixlan=ixlan,
|
|
asn=network_other.asn,
|
|
speed=20000,
|
|
ipaddr4=None,
|
|
ipaddr6=IP6,
|
|
status="ok",
|
|
is_rs_peer=False,
|
|
operational=False,
|
|
)
|
|
|
|
# setup IX-F data
|
|
|
|
data = setup_test_data("ixf.member.1")
|
|
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", data)
|
|
|
|
ix = ixlan.ix
|
|
|
|
# real peer should exist in IX-F data
|
|
|
|
real4, real6 = ix.peer_exists_in_ixf_data(1001, IP4, IP6)
|
|
assert real4
|
|
assert real6
|
|
|
|
# ghost peer should NOT exist in IX-F data
|
|
|
|
ghost4, ghost6 = ix.peer_exists_in_ixf_data(1010, IP4, IP6)
|
|
assert not ghost4
|
|
assert not ghost6
|
|
|
|
# create and save a real peer that has the same ip addresses
|
|
# as the ghost peer
|
|
|
|
real_peer = NetworkIXLan(
|
|
network=network,
|
|
status="ok",
|
|
ipaddr4=IP4,
|
|
ipaddr6=IP6,
|
|
ixlan=ixlan,
|
|
speed=1000,
|
|
asn=network.asn,
|
|
)
|
|
|
|
# run full validation (this will run `validate_real_vs_ghost_peer`)
|
|
|
|
real_peer.full_clean()
|
|
real_peer.save()
|
|
|
|
# real peer has been saved and since it claimed both ip4 and ip6, the ghost
|
|
# peer is now deleted
|
|
|
|
ghost_peer_a.refresh_from_db()
|
|
ghost_peer_b.refresh_from_db()
|
|
|
|
assert ghost_peer_a.status == "deleted"
|
|
assert ghost_peer_b.status == "deleted"
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ghost_peer_vs_real_peer_two_netixlan_partial():
|
|
"""
|
|
Tests that a real peer can claim the ip addresses of a gohst peer. #983
|
|
|
|
In this test both ipv4 and ipv6 exist on separate netixlans.
|
|
|
|
In this test the conflicting netixlans will have the other ip address still set and will not be deleted.
|
|
"""
|
|
|
|
# set up entities
|
|
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
|
|
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
|
|
network_other = Network.objects.create(
|
|
asn=1010, name="AS1010", status="ok", org=org
|
|
)
|
|
ixlan = ix.ixlan
|
|
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
|
|
ixlan.save()
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="195.69.144.0/22",
|
|
protocol="IPv4",
|
|
)
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="2001:7f8:1::/64",
|
|
protocol="IPv6",
|
|
)
|
|
|
|
IP4 = "195.69.147.250"
|
|
IP6 = "2001:7f8:1::a500:2906:1"
|
|
|
|
ghost_peer_a = NetworkIXLan.objects.create(
|
|
network=network_other,
|
|
ixlan=ixlan,
|
|
asn=network_other.asn,
|
|
speed=20000,
|
|
ipaddr4=IP4,
|
|
ipaddr6="2001:7f8:1::a500:2906:2",
|
|
status="ok",
|
|
is_rs_peer=False,
|
|
operational=False,
|
|
)
|
|
|
|
ghost_peer_b = NetworkIXLan.objects.create(
|
|
network=network_other,
|
|
ixlan=ixlan,
|
|
asn=network_other.asn,
|
|
speed=20000,
|
|
ipaddr4="195.69.147.251",
|
|
ipaddr6=IP6,
|
|
status="ok",
|
|
is_rs_peer=False,
|
|
operational=False,
|
|
)
|
|
|
|
# setup IX-F data
|
|
|
|
data = setup_test_data("ixf.member.1")
|
|
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", data)
|
|
|
|
ix = ixlan.ix
|
|
|
|
# real peer should exist in IX-F data
|
|
|
|
real4, real6 = ix.peer_exists_in_ixf_data(1001, IP4, IP6)
|
|
assert real4
|
|
assert real6
|
|
|
|
# ghost peer should NOT exist in IX-F data
|
|
|
|
ghost4, ghost6 = ix.peer_exists_in_ixf_data(1010, IP4, IP6)
|
|
assert not ghost4
|
|
assert not ghost6
|
|
|
|
# create and save a real peer that has the same ip addresses
|
|
# as the ghost peer
|
|
|
|
real_peer = NetworkIXLan(
|
|
network=network,
|
|
status="ok",
|
|
ipaddr4=IP4,
|
|
ipaddr6=IP6,
|
|
ixlan=ixlan,
|
|
speed=1000,
|
|
asn=network.asn,
|
|
)
|
|
|
|
# run full validation (this will run `validate_real_vs_ghost_peer`)
|
|
|
|
real_peer.full_clean()
|
|
real_peer.save()
|
|
|
|
# real peer has been saved and since it only claimed one ip address
|
|
# from either ghost peer, both ghost peers remain
|
|
|
|
ghost_peer_a.refresh_from_db()
|
|
ghost_peer_b.refresh_from_db()
|
|
|
|
assert ghost_peer_a.status == "ok"
|
|
assert ghost_peer_a.ipaddr4 is None
|
|
assert ghost_peer_a.ipaddr6 is not None
|
|
|
|
assert ghost_peer_b.status == "ok"
|
|
assert ghost_peer_b.ipaddr4 is not None
|
|
assert ghost_peer_b.ipaddr6 is None
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_ghost_peer_vs_real_peer_invalid_ixf_data():
|
|
"""
|
|
Tests that a real peer can claim the ip addresses of a gohst peer. #983
|
|
|
|
Test the handling of invalid IX-F data, in which case the ghost peer vs real peer
|
|
logic should be skipped.
|
|
"""
|
|
|
|
# set up entities
|
|
|
|
org = Organization.objects.create(name="Test org", status="ok")
|
|
ix = InternetExchange.objects.create(name="Test ix", status="ok", org=org)
|
|
network = Network.objects.create(asn=1001, name="AS1001", status="ok", org=org)
|
|
network_other = Network.objects.create(
|
|
asn=1010, name="AS1010", status="ok", org=org
|
|
)
|
|
ixlan = ix.ixlan
|
|
ixlan.ixf_ixp_member_list_url = "https://localhost/IX-F"
|
|
ixlan.save()
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="195.69.144.0/22",
|
|
protocol="IPv4",
|
|
)
|
|
|
|
IXLanPrefix.objects.create(
|
|
ixlan=ixlan,
|
|
status="ok",
|
|
prefix="2001:7f8:1::/64",
|
|
protocol="IPv6",
|
|
)
|
|
|
|
IP4 = "195.69.147.250"
|
|
IP6 = "2001:7f8:1::a500:2906:1"
|
|
|
|
ghost_peer = NetworkIXLan.objects.create(
|
|
network=network_other,
|
|
ixlan=ixlan,
|
|
asn=network_other.asn,
|
|
speed=20000,
|
|
ipaddr4=IP4,
|
|
ipaddr6=IP6,
|
|
status="ok",
|
|
is_rs_peer=False,
|
|
operational=False,
|
|
)
|
|
# setup IX-F data
|
|
|
|
cache.set(f"IXF-CACHE-{ix.ixlan.ixf_ixp_member_list_url}", {"invalid": "data"})
|
|
|
|
ix = ixlan.ix
|
|
|
|
real_peer = NetworkIXLan(
|
|
network=network,
|
|
status="ok",
|
|
ipaddr4=IP4,
|
|
ipaddr6=IP6,
|
|
ixlan=ixlan,
|
|
speed=1000,
|
|
asn=network.asn,
|
|
)
|
|
|
|
# run full validation (this will run `validate_real_vs_ghost_peer`)
|
|
|
|
with pytest.raises(Exception) as excinfo:
|
|
real_peer.full_clean()
|
|
assert "IP already exists" in str(excinfo.value)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value,validated",
|
|
[
|
|
# success validation
|
|
(
|
|
[
|
|
{"service": "website", "identifier": "https://www.example.com"},
|
|
{"service": "x", "identifier": "unknown12"},
|
|
],
|
|
[
|
|
{"service": "website", "identifier": "https://www.example.com"},
|
|
{"service": "x", "identifier": "unknown12"},
|
|
],
|
|
),
|
|
(
|
|
[
|
|
{"service": "instagram", "identifier": "john__doe_"},
|
|
{"service": "tiktok", "identifier": "unknown_12"},
|
|
],
|
|
[
|
|
{"service": "instagram", "identifier": "john__doe_"},
|
|
{"service": "tiktok", "identifier": "unknown_12"},
|
|
],
|
|
),
|
|
(
|
|
[
|
|
{"service": "instagram", "identifier": "john-doe-"},
|
|
{"service": "tiktok", "identifier": "-unknown-12"},
|
|
],
|
|
[
|
|
{"service": "instagram", "identifier": "john-doe-"},
|
|
{"service": "tiktok", "identifier": "-unknown-12"},
|
|
],
|
|
),
|
|
# fail validation
|
|
(
|
|
[
|
|
{"service": "website", "identifier": "https://www.example.com"},
|
|
{"service": "x", "identifier": "unknown11."},
|
|
],
|
|
False,
|
|
),
|
|
([{"service": "instagram", "identifier": "john__doe_*"}], False),
|
|
(
|
|
[
|
|
{"service": "website", "identifier": ""},
|
|
{"service": "x", "identifier": "unknown"},
|
|
],
|
|
False,
|
|
),
|
|
(
|
|
[
|
|
{"service": "website", "identifier": "https://www.example.com"},
|
|
{"service": "", "identifier": "unknown"},
|
|
],
|
|
False,
|
|
),
|
|
(
|
|
[
|
|
{"service": "website", "identifier": None},
|
|
{"service": "x", "identifier": "unknown"},
|
|
],
|
|
False,
|
|
),
|
|
(
|
|
[
|
|
{"service": "website", "identifier": "https://www.example.com"},
|
|
{"service": None, "identifier": "unknown"},
|
|
],
|
|
False,
|
|
),
|
|
(
|
|
[
|
|
{"service": "website", "identifier": "https://www.example.com"},
|
|
{"service": "website", "identifier": "https://www.unknown.com"},
|
|
],
|
|
False,
|
|
),
|
|
(
|
|
[
|
|
{"service": "website", "identifier": "https://www.example.com"},
|
|
{"foo": "bar"},
|
|
],
|
|
False,
|
|
),
|
|
(
|
|
{
|
|
"service": {
|
|
"service": "website",
|
|
"identifier": "https://www.example.com",
|
|
},
|
|
},
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.django_db
|
|
def test_validate_social_media(value, validated):
|
|
if not validated:
|
|
with pytest.raises(ValidationError):
|
|
validate_social_media(value)
|
|
else:
|
|
assert validate_social_media(value) == validated
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"website,org_website,validated",
|
|
[
|
|
# success validation website not null
|
|
(
|
|
"https://www.example.com",
|
|
"https://www.example1.com",
|
|
"https://www.example.com",
|
|
),
|
|
# success validation website null and overrided by organization website
|
|
(None, "https://www.example1.com", "https://www.example1.com"),
|
|
# fail validation
|
|
(
|
|
None,
|
|
None,
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.django_db
|
|
def test_validate_website_override(website, org_website, validated):
|
|
if not validated:
|
|
with pytest.raises(ValidationError):
|
|
validate_website_override(website, org_website)
|
|
else:
|
|
assert validate_website_override(website, org_website) == validated
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_org_create_with_none_social_media():
|
|
org = Organization.objects.create(name="Test org", status="ok", social_media=None)
|
|
assert org.social_media == {}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value,is_valid,validated",
|
|
[
|
|
# success validation
|
|
("63311", True, "63311"),
|
|
("as63311", True, "63311"),
|
|
("asn63311", True, "63311"),
|
|
("AS63311", True, "63311"),
|
|
("ASN63311", True, "63311"),
|
|
# fail validation
|
|
("AN63311", False, None),
|
|
("6as3311", False, None),
|
|
("63311asn", False, None),
|
|
],
|
|
)
|
|
@pytest.mark.django_db
|
|
def test_validate_asn_prefix(value, is_valid, validated):
|
|
print(is_valid)
|
|
if not is_valid:
|
|
with pytest.raises(RestValidationError):
|
|
validate_asn_prefix(value)
|
|
else:
|
|
assert validate_asn_prefix(value) == validated
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value,is_valid,validated",
|
|
[
|
|
# success validation
|
|
(37.7749, True, 37.7749),
|
|
(-23.5505, True, -23.5505),
|
|
(51.5074, True, 51.5074),
|
|
(40.7128, True, 40.7128),
|
|
("-33.8688", True, -33.8688),
|
|
# fail validation
|
|
(95.1234, False, None),
|
|
(-120.5678, False, None),
|
|
("-122.5678", False, None),
|
|
("abcdef", False, None),
|
|
],
|
|
)
|
|
@pytest.mark.django_db
|
|
def test_validate_latitude(value, is_valid, validated):
|
|
if not is_valid:
|
|
with pytest.raises(ValidationError):
|
|
validate_latitude(value)
|
|
else:
|
|
assert validate_latitude(value) == validated
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"value,is_valid,validated",
|
|
[
|
|
# success validation
|
|
(-122.4194, True, -122.4194),
|
|
(-46.6333, True, -46.6333),
|
|
(-0.1270, True, -0.1270),
|
|
(-74.0060, True, -74.0060),
|
|
("151.2093", True, 151.2093),
|
|
# fail validation
|
|
(190.1234, False, None),
|
|
(-250.5678, False, None),
|
|
("360.9876", False, None),
|
|
("abcdef", False, None),
|
|
],
|
|
)
|
|
@pytest.mark.django_db
|
|
def test_validate_longitude(value, is_valid, validated):
|
|
if not is_valid:
|
|
with pytest.raises(ValidationError):
|
|
validate_longitude(value)
|
|
else:
|
|
assert validate_longitude(value) == validated
|
|
|
|
|
|
def geo_mock_init(self, key, timeout):
|
|
pass
|
|
|
|
|
|
def geo_gmaps_mock_geocode_freeform(location):
|
|
return {"lat": 40.712776, "lng": -74.005974}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"current_geocode,new_geocode,city,is_valid,validated",
|
|
[
|
|
# test success with exists geocode (max 1km from previous geocode)
|
|
(
|
|
(50.951533, 1.852570),
|
|
(50.951533, 1.851440),
|
|
"london",
|
|
True,
|
|
(50.951533, 1.851440),
|
|
),
|
|
(
|
|
(40.712776, -74.005974),
|
|
(40.712790, -74.003974),
|
|
"new york",
|
|
True,
|
|
(40.712790, -74.003974),
|
|
),
|
|
# # test fail with exists geocode (max 1km from previous geocode)
|
|
((40.712776, -74.005974), (40.712790, -73.003974), "new york", False, None),
|
|
((50.951533, 1.852570), (51.951533, 0.851440), "london", False, None),
|
|
# test success with not exists geocode (max 50km from city)
|
|
(
|
|
(None, None),
|
|
(40.712771, -74.005970),
|
|
"new york",
|
|
True,
|
|
(40.712771, -74.005970),
|
|
),
|
|
(
|
|
(None, None),
|
|
(40.716822, -73.991032),
|
|
"new york",
|
|
True,
|
|
(40.716822, -73.991032),
|
|
),
|
|
# test fail with not exists geocode (max 50km from city)
|
|
((None, None), (36.169941, -115.139832), "new york", False, None),
|
|
((None, None), (36.201902, -115.328808), "new york", False, None),
|
|
],
|
|
)
|
|
@patch.object(geo.GoogleMaps, "__init__", geo_mock_init)
|
|
@patch.object(geo.Melissa, "__init__", geo_mock_init)
|
|
@pytest.mark.django_db
|
|
def test_validate_distance_geocode(
|
|
current_geocode, new_geocode, city, is_valid, validated, settings
|
|
):
|
|
settings.MELISSA_KEY = ""
|
|
settings.GOOGLE_GEOLOC_API_KEY = ""
|
|
with patch.object(
|
|
geo.GoogleMaps, "geocode_freeform", side_effect=geo_gmaps_mock_geocode_freeform
|
|
):
|
|
if not is_valid:
|
|
with pytest.raises(ValidationError):
|
|
validate_distance_geocode(current_geocode, new_geocode, city)
|
|
else:
|
|
assert (
|
|
validate_distance_geocode(current_geocode, new_geocode, city)
|
|
== validated
|
|
)
|