245 lines
10 KiB
Python
245 lines
10 KiB
Python
import re
|
|
import sys
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.urlresolvers import reverse
|
|
|
|
import pytest
|
|
import utils
|
|
|
|
from django.test import override_settings
|
|
|
|
from passerelle.base import signature
|
|
from passerelle.base.models import ApiUser, AccessRight, ResourceLog
|
|
from passerelle.apps.oxyd.models import OxydSMSGateway
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
@pytest.fixture
|
|
def oxyd(db):
|
|
return OxydSMSGateway.objects.create(title='eservices',
|
|
slug='eservices',
|
|
username='user',
|
|
description='oxyd',
|
|
password='secret')
|
|
|
|
def test_anonymous_access(app, oxyd):
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
resp = app.post_json(endpoint_url, params={}, status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
|
|
api = ApiUser.objects.create(username='public',
|
|
fullname='public',
|
|
description='access for all',
|
|
keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(OxydSMSGateway)
|
|
AccessRight.objects.create(codename='can_send_messages',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=oxyd.pk,
|
|
)
|
|
resp = app.post_json(endpoint_url, params={})
|
|
# for empty payload the connector returns an APIError with
|
|
# {"err_desc": "missing \"message\" in JSON payload"}
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'
|
|
|
|
def test_access_with_signature(app, oxyd):
|
|
api = ApiUser.objects.create(username='eservices',
|
|
fullname='Eservices User',
|
|
description='eservices',
|
|
keytype='SIGN',
|
|
key='12345')
|
|
obj_type = ContentType.objects.get_for_model(OxydSMSGateway)
|
|
|
|
AccessRight.objects.create(codename='can_send_messages',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=oxyd.pk,
|
|
)
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
url = signature.sign_url(endpoint_url + '?orig=eservices', '12345')
|
|
# for empty payload the connector returns an APIError with
|
|
# {"err_desc": "missing \"message\" in JSON payload"}
|
|
resp = app.post_json(url, params={})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'
|
|
# bad key
|
|
url = signature.sign_url(endpoint_url + '?orig=eservices', 'notmykey')
|
|
resp = app.post_json(url, params={}, status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
# add garbage after signature
|
|
url = signature.sign_url(endpoint_url + '?orig=eservices', '12345')
|
|
url = '%s&foo=bar' % url
|
|
resp = app.post_json(url, params={}, status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
|
|
# trusted user (from settings.KNOWN_SERVICES)
|
|
url = signature.sign_url(endpoint_url + '?orig=wcs1', 'abcde')
|
|
resp = app.post_json(url, params={})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'
|
|
# bad key
|
|
url = signature.sign_url(endpoint_url + '?orig=wcs1', 'notmykey')
|
|
resp = app.post_json(url, params={}, status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
|
|
|
|
def test_access_http_auth(app, oxyd):
|
|
username = 'apiuser'
|
|
password = '12345'
|
|
api = ApiUser.objects.create(username=username,
|
|
fullname='Api User',
|
|
description='api',
|
|
keytype='SIGN',
|
|
key=password)
|
|
obj_type = ContentType.objects.get_for_model(OxydSMSGateway)
|
|
|
|
AccessRight.objects.create(codename='can_send_messages',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=oxyd.pk,
|
|
)
|
|
app.authorization = ('Basic', (username, password))
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
resp = app.post_json(endpoint_url, params={})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'
|
|
|
|
def test_access_apikey(app, oxyd):
|
|
password = 'apiuser_12345'
|
|
api = ApiUser.objects.create(username='apiuser',
|
|
fullname='Api User',
|
|
description='api',
|
|
keytype='API',
|
|
key=password)
|
|
obj_type = ContentType.objects.get_for_model(OxydSMSGateway)
|
|
|
|
AccessRight.objects.create(codename='can_send_messages',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=oxyd.pk,
|
|
)
|
|
params = {'message': 'test'}
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
resp = app.post_json(endpoint_url + '?apikey=' + password , params=params)
|
|
resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Payload error: missing "from" in JSON payload'
|
|
resp = app.post_json(endpoint_url + '?apikey=' + password[:3] , params=params, status=403)
|
|
resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
|
|
def test_access_apiuser_with_no_key(app, oxyd):
|
|
api = ApiUser.objects.create(username='apiuser',
|
|
fullname='Api User',
|
|
description='api')
|
|
obj_type = ContentType.objects.get_for_model(OxydSMSGateway)
|
|
|
|
AccessRight.objects.create(codename='can_send_messages',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=oxyd.pk,
|
|
)
|
|
params = {'message': 'test', 'from': 'test api'}
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
resp = app.post_json(endpoint_url, params=params)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Payload error: missing "to" in JSON payload'
|
|
|
|
def test_access_apiuser_with_ip_restriction(app, oxyd):
|
|
authorized_ip = '176.31.123.109'
|
|
api = ApiUser.objects.create(username='apiuser',
|
|
fullname='Api User',
|
|
description='api',
|
|
ipsource=authorized_ip
|
|
)
|
|
obj_type = ContentType.objects.get_for_model(OxydSMSGateway)
|
|
|
|
AccessRight.objects.create(codename='can_send_messages',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=oxyd.pk,
|
|
)
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
resp = app.post_json(endpoint_url, params={}, extra_environ={'REMOTE_ADDR': '127.0.0.1'},
|
|
status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
resp = app.post_json(endpoint_url, params={},
|
|
extra_environ={'REMOTE_ADDR': authorized_ip})
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'Payload error: missing "message" in JSON payload'
|
|
|
|
def test_logged_requests_and_responses_max_size(app, oxyd, settings):
|
|
endpoint_url = reverse('generic-endpoint',
|
|
kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
|
|
api = ApiUser.objects.create(username='public',
|
|
fullname='public',
|
|
description='access for all',
|
|
keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(OxydSMSGateway)
|
|
AccessRight.objects.create(codename='can_send_messages',
|
|
apiuser=api,
|
|
resource_type=obj_type,
|
|
resource_pk=oxyd.pk,
|
|
)
|
|
payload = {
|
|
'message': 'hello',
|
|
'from': '+33699999999',
|
|
'to': ['+33688888888', '+33677777777'],
|
|
}
|
|
headers = {'Content-Type': 'foo/bar'}
|
|
response = oxyd.TEST_DEFAULTS['test_vectors'][0]
|
|
oxyd.set_log_level('DEBUG') # log request payload and response headers/content
|
|
settings.LOGGED_CONTENT_TYPES_MESSAGES = 'foo/bar' # response content to log
|
|
|
|
assert oxyd.logging_parameters.requests_max_size == 4999
|
|
assert oxyd.logging_parameters.responses_max_size == 5000
|
|
with utils.mock_url(oxyd.URL, response, headers=headers):
|
|
result = app.post_json(endpoint_url, params=payload)
|
|
assert len(ResourceLog.objects.all()) == 4
|
|
|
|
# initial POST query
|
|
assert len(ResourceLog.objects.all()[0].extra['connector_payload']) == 84
|
|
|
|
# connector POST queries
|
|
assert len(ResourceLog.objects.all()[1].extra['request_payload']) == 57
|
|
assert len(ResourceLog.objects.all()[2].extra['request_payload']) == 57
|
|
assert len(ResourceLog.objects.all()[1].extra['response_content']) in (210, 211)
|
|
assert len(ResourceLog.objects.all()[2].extra['response_content']) in (210, 211)
|
|
|
|
# connector reply
|
|
assert len(ResourceLog.objects.all()[3].extra['body']) in (86, 87)
|
|
|
|
# troncate logs
|
|
parameters = oxyd.logging_parameters
|
|
parameters.requests_max_size = 10
|
|
parameters.save()
|
|
parameters = oxyd.logging_parameters
|
|
parameters.responses_max_size = 20
|
|
parameters.save()
|
|
with utils.mock_url(oxyd.URL, response, headers=headers):
|
|
result = app.post_json(endpoint_url, params=payload)
|
|
assert len(ResourceLog.objects.all()) == 8
|
|
assert len(ResourceLog.objects.all()[4].extra['connector_payload']) == 10
|
|
assert len(ResourceLog.objects.all()[5].extra['request_payload']) == 12
|
|
assert len(ResourceLog.objects.all()[6].extra['request_payload']) == 12
|
|
assert len(ResourceLog.objects.all()[5].extra['response_content']) in (22, 23)
|
|
assert len(ResourceLog.objects.all()[6].extra['response_content']) in (22, 23)
|
|
assert len(ResourceLog.objects.all()[7].extra['body']) in (12, 13)
|