Merge branch 'develop' into develop-2.9

This commit is contained in:
Jeremy Stretch 2020-06-17 16:57:17 -04:00
commit 328d639886
5 changed files with 113 additions and 186 deletions

View File

@ -1,5 +1,14 @@
# NetBox v2.8
## v2.8.7 (FUTURE)
### Bug Fixes
* [#4766](https://github.com/netbox-community/netbox/issues/4766) - Fix redirect after login when `next` is not specified
* [#4772](https://github.com/netbox-community/netbox/issues/4772) - Fix "brief" format for the secrets REST API endpoint
---
## v2.8.6 (2020-06-15)
### Enhancements

View File

@ -1,13 +1,22 @@
from rest_framework import serializers
from secrets.models import SecretRole
from secrets.models import Secret, SecretRole
from utilities.api import WritableNestedSerializer
__all__ = [
'NestedSecretRoleSerializer'
'NestedSecretRoleSerializer',
'NestedSecretSerializer',
]
class NestedSecretSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='secrets-api:secret-detail')
class Meta:
model = Secret
fields = ['id', 'url', 'name']
class NestedSecretRoleSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='secrets-api:secretrole-detail')
secret_count = serializers.IntegerField(read_only=True)

View File

@ -48,181 +48,71 @@ class SecretRoleTest(APIViewTestCases.APIViewTestCase):
SecretRole.objects.bulk_create(secret_roles)
# TODO: Standardize SecretTest
class SecretTest(APITestCase):
class SecretTest(APIViewTestCases.APIViewTestCase):
model = Secret
brief_fields = ['id', 'name', 'url']
def setUp(self):
super().setUp()
self.user.is_superuser = False
self.user.save()
self.add_permissions(
'secrets.add_secret',
'secrets.change_secret',
'secrets.delete_secret',
'secrets.view_secret',
)
# Create a UserKey for the test user
userkey = UserKey(user=self.user, public_key=PUBLIC_KEY)
userkey.save()
# Create a SessionKey for the user
self.master_key = userkey.get_master_key(PRIVATE_KEY)
session_key = SessionKey(userkey=userkey)
session_key.save(self.master_key)
self.header = {
'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key),
'HTTP_X_SESSION_KEY': base64.b64encode(session_key.key),
}
# Append the session key to the test client's request header
self.header['HTTP_X_SESSION_KEY'] = base64.b64encode(session_key.key)
self.plaintexts = (
'Secret #1 Plaintext',
'Secret #2 Plaintext',
'Secret #3 Plaintext',
site = Site.objects.create(name='Site 1', slug='site-1')
manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1')
devicerole = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
device = Device.objects.create(name='Device 1', site=site, device_type=devicetype, device_role=devicerole)
secret_roles = (
SecretRole(name='Secret Role 1', slug='secret-role-1'),
SecretRole(name='Secret Role 2', slug='secret-role-2'),
)
SecretRole.objects.bulk_create(secret_roles)
site = Site.objects.create(name='Test Site 1', slug='test-site-1')
manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Test Device Type 1')
devicerole = DeviceRole.objects.create(name='Test Device Role 1', slug='test-device-role-1')
self.device = Device.objects.create(
name='Test Device 1', site=site, device_type=devicetype, device_role=devicerole
secrets = (
Secret(device=device, role=secret_roles[0], name='Secret 1', plaintext='ABC'),
Secret(device=device, role=secret_roles[0], name='Secret 2', plaintext='DEF'),
Secret(device=device, role=secret_roles[0], name='Secret 3', plaintext='GHI'),
)
self.secretrole1 = SecretRole.objects.create(name='Test Secret Role 1', slug='test-secret-role-1')
self.secretrole2 = SecretRole.objects.create(name='Test Secret Role 2', slug='test-secret-role-2')
self.secret1 = Secret(
device=self.device, role=self.secretrole1, name='Test Secret 1', plaintext=self.plaintexts[0]
)
self.secret1.encrypt(self.master_key)
self.secret1.save()
self.secret2 = Secret(
device=self.device, role=self.secretrole1, name='Test Secret 2', plaintext=self.plaintexts[1]
)
self.secret2.encrypt(self.master_key)
self.secret2.save()
self.secret3 = Secret(
device=self.device, role=self.secretrole1, name='Test Secret 3', plaintext=self.plaintexts[2]
)
self.secret3.encrypt(self.master_key)
self.secret3.save()
for secret in secrets:
secret.encrypt(self.master_key)
secret.save()
def test_get_secret(self):
url = reverse('secrets-api:secret-detail', kwargs={'pk': self.secret1.pk})
# Secret plaintext should not be decrypted as the user has not been assigned to the role
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertIsNone(response.data['plaintext'])
# The plaintext should be present once the user has been assigned to the role
self.secretrole1.users.add(self.user)
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(response.data['plaintext'], self.plaintexts[0])
def test_list_secrets(self):
url = reverse('secrets-api:secret-list')
# Secret plaintext should not be decrypted as the user has not been assigned to the role
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 3)
for secret in response.data['results']:
self.assertIsNone(secret['plaintext'])
# The plaintext should be present once the user has been assigned to the role
self.secretrole1.users.add(self.user)
response = self.client.get(url, **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(response.data['count'], 3)
for i, secret in enumerate(response.data['results']):
self.assertEqual(secret['plaintext'], self.plaintexts[i])
def test_create_secret(self):
data = {
'device': self.device.pk,
'role': self.secretrole1.pk,
'name': 'Test Secret 4',
'plaintext': 'Secret #4 Plaintext',
}
# Assign test user to secret role
self.secretrole1.users.add(self.user)
url = reverse('secrets-api:secret-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(response.data['plaintext'], data['plaintext'])
self.assertEqual(Secret.objects.count(), 4)
secret4 = Secret.objects.get(pk=response.data['id'])
secret4.decrypt(self.master_key)
self.assertEqual(secret4.role_id, data['role'])
self.assertEqual(secret4.plaintext, data['plaintext'])
def test_create_secret_bulk(self):
data = [
self.create_data = [
{
'device': self.device.pk,
'role': self.secretrole1.pk,
'name': 'Test Secret 4',
'plaintext': 'Secret #4 Plaintext',
'device': device.pk,
'role': secret_roles[1].pk,
'name': 'Secret 4',
'plaintext': 'JKL',
},
{
'device': self.device.pk,
'role': self.secretrole1.pk,
'name': 'Test Secret 5',
'plaintext': 'Secret #5 Plaintext',
'device': device.pk,
'role': secret_roles[1].pk,
'name': 'Secret 5',
'plaintext': 'MNO',
},
{
'device': self.device.pk,
'role': self.secretrole1.pk,
'name': 'Test Secret 6',
'plaintext': 'Secret #6 Plaintext',
'device': device.pk,
'role': secret_roles[1].pk,
'name': 'Secret 6',
'plaintext': 'PQR',
},
]
# Assign test user to secret role
self.secretrole1.users.add(self.user)
url = reverse('secrets-api:secret-list')
response = self.client.post(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(Secret.objects.count(), 6)
self.assertEqual(response.data[0]['plaintext'], data[0]['plaintext'])
self.assertEqual(response.data[1]['plaintext'], data[1]['plaintext'])
self.assertEqual(response.data[2]['plaintext'], data[2]['plaintext'])
def test_update_secret(self):
data = {
'device': self.device.pk,
'role': self.secretrole2.pk,
'plaintext': 'NewPlaintext',
}
# Assign test user to secret role
self.secretrole1.users.add(self.user)
url = reverse('secrets-api:secret-detail', kwargs={'pk': self.secret1.pk})
response = self.client.put(url, data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(response.data['plaintext'], data['plaintext'])
self.assertEqual(Secret.objects.count(), 3)
secret1 = Secret.objects.get(pk=response.data['id'])
secret1.decrypt(self.master_key)
self.assertEqual(secret1.role_id, data['role'])
self.assertEqual(secret1.plaintext, data['plaintext'])
def test_delete_secret(self):
# Assign test user to secret role
self.secretrole1.users.add(self.user)
url = reverse('secrets-api:secret-detail', kwargs={'pk': self.secret1.pk})
response = self.client.delete(url, **self.header)
self.assertHttpStatus(response, status.HTTP_204_NO_CONTENT)
self.assertEqual(Secret.objects.count(), 2)
def prepare_instance(self, instance):
# Unlock the plaintext prior to evaluation of the instance
instance.decrypt(self.master_key)
return instance
class GetSessionKeyTest(APITestCase):

View File

@ -50,7 +50,7 @@ class LoginView(View):
logger.debug("Login form validation was successful")
# Determine where to direct user after successful login
redirect_to = request.POST.get('next')
redirect_to = request.POST.get('next', reverse('home'))
if redirect_to and not is_safe_url(url=redirect_to, allowed_hosts=request.get_host()):
logger.warning(f"Ignoring unsafe 'next' URL passed to login form: {redirect_to}")
redirect_to = reverse('home')

View File

@ -35,6 +35,54 @@ class TestCase(_TestCase):
self.client = Client()
self.client.force_login(self.user)
def prepare_instance(self, instance):
"""
Test cases can override this method to perform any necessary manipulation of an instance prior to its evaluation
against test data. For example, it can be used to decrypt a Secret's plaintext attribute.
"""
return instance
def model_to_dict(self, instance, fields, api=False):
"""
Return a dictionary representation of an instance.
"""
# Prepare the instance and call Django's model_to_dict() to extract all fields
model_dict = model_to_dict(self.prepare_instance(instance), fields=fields)
# Map any additional (non-field) instance attributes that were specified
for attr in fields:
if hasattr(instance, attr) and attr not in model_dict:
model_dict[attr] = getattr(instance, attr)
for key, value in list(model_dict.items()):
# TODO: Differentiate between tags assigned to the instance and a M2M field for tags (ex: ConfigContext)
if key == 'tags':
model_dict[key] = sorted(value, key=lambda t: t.name)
# Convert ManyToManyField to list of instance PKs
elif model_dict[key] and type(value) in (list, tuple) and hasattr(value[0], 'pk'):
model_dict[key] = [obj.pk for obj in value]
if api:
# Replace ContentType numeric IDs with <app_label>.<model>
if type(getattr(instance, key)) is ContentType:
ct = ContentType.objects.get(pk=value)
model_dict[key] = f'{ct.app_label}.{ct.model}'
# Convert IPNetwork instances to strings
if type(value) is IPNetwork:
model_dict[key] = str(value)
else:
# Convert ArrayFields to CSV strings
if type(instance._meta.get_field(key)) is ArrayField:
model_dict[key] = ','.join([str(v) for v in value])
return model_dict
#
# Permissions management
#
@ -67,41 +115,12 @@ class TestCase(_TestCase):
"""
Compare a model instance to a dictionary, checking that its attribute values match those specified
in the dictionary.
:instance: Python object instance
:data: Dictionary of test data used to define the instance
:api: Set to True is the data is a JSON representation of the instance
"""
model_dict = model_to_dict(instance, fields=data.keys())
for key, value in list(model_dict.items()):
# TODO: Differentiate between tags assigned to the instance and a M2M field for tags (ex: ConfigContext)
if key == 'tags':
model_dict[key] = sorted(value)
# Convert ManyToManyField to list of instance PKs
elif model_dict[key] and type(value) in (list, tuple) and hasattr(value[0], 'pk'):
model_dict[key] = [obj.pk for obj in value]
if api:
# Replace ContentType numeric IDs with <app_label>.<model>
field = instance._meta.get_field(key)
if type(field) is ForeignKey and field.related_model is ContentType:
ct = ContentType.objects.get(pk=value)
model_dict[key] = f'{ct.app_label}.{ct.model}'
elif type(field) is ManyToManyField and field.related_model is ContentType:
model_dict[key] = [f'{ct.app_label}.{ct.model}' for ct in value]
# Convert IPNetwork instances to strings
elif type(value) is IPNetwork:
model_dict[key] = str(value)
else:
# Convert ArrayFields to CSV strings
if type(instance._meta.get_field(key)) is ArrayField:
model_dict[key] = ','.join([str(v) for v in value])
model_dict = self.model_to_dict(instance, fields=data.keys(), api=api)
# Omit any dictionary keys which are not instance attributes
relevant_data = {