178 lines
7.2 KiB
Python
178 lines
7.2 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
|
|
|
|
from passerelle.base import signature
|
|
from passerelle.base.models import ApiUser, AccessRight
|
|
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'
|
|
|
|
# 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'
|