passerelle/tests/test_api_access.py

import pytest
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse

from passerelle.apps.oxyd.models import OxydSMSGateway
from passerelle.base import signature
from passerelle.base.models import AccessRight, ApiUser

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={}, status=400)
    # for empty payload the connector returns an APIError with
    # {"err_desc": "'message' is a required property"}
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == "'message' is a required property"


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": "'message' is a required property"}
    resp = app.post_json(url, params={}, status=400)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == "'message' is a required property"

    # 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={}, status=400)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == "'message' is a required property"

    # 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={}, status=400)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == "'message' is a required property"


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, status=400)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == "'from' is a required property"
    resp = app.post_json(endpoint_url + '?apikey=' + password[:3], params=params, status=403)
    assert 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, status=400)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == "'to' is a required property"


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}, status=400)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == "'message' is a required property"