passerelle/tests/test_api_access.py

import re
import sys

from django.contrib.auth.models import User
from django.core.wsgi import get_wsgi_application
from django.contrib.contenttypes.models import ContentType
from django.core.urlresolvers import reverse

import pytest
from webtest import TestApp

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

pytestmark = pytest.mark.django_db

@pytest.fixture
def setup():
    app = TestApp(get_wsgi_application())
    oxyd = OxydSMSGateway.objects.create(title='eservices',
                slug='eservices',
                username='user',
                description='oxyd',
                password='secret')
    return app, oxyd

def test_anonymous_access(setup):
    app, oxyd = setup
    endpoint_url = reverse('generic-endpoint',
            kwargs={'connector': 'oxyd', 'slug': oxyd.slug, 'endpoint': 'send'})
    resp = app.post_json(endpoint_url, {}, status=403)
    assert resp.json['err'] == 1
    assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'

def test_access_with_good_signature(setup):
    app, oxyd = setup
    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 500 with
    # {"err_desc": "missing \"message\" in JSON payload"}
    resp = app.post_json(url, {}, status=500)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == 'missing "message" in JSON payload'

def test_access_http_auth(setup):
    app, oxyd = setup
    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, {}, status=500)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == 'missing "message" in JSON payload'

def test_access_apikey(setup):
    app, oxyd = setup
    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, status=500)
    resp.json['err'] == 1
    assert resp.json['err_desc'] == 'missing "from" in JSON payload'
    resp = app.post_json(endpoint_url + '?apikey=' + password[:3] , params, status=403)
    resp.json['err'] == 1
    assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'

def test_access_apiuser_with_no_key(setup):
    app, oxyd = setup
    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, status=500)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == 'missing "to" in JSON payload'

def test_access_apiuser_with_ip_restriction(setup):
    app, oxyd = setup
    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, {}, 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, {},
            extra_environ=[('REMOTE_ADDR', authorized_ip)], status=500)
    assert resp.json['err'] == 1
    assert resp.json['err_desc'] == 'missing "message" in JSON payload'