0002-rest_authentication-add-api-client-authentication-67.patch

Emmanuel Cazenave, 05 septembre 2022 14:32

Voir les différences: en ligne côte à côte

Subject: [PATCH 2/2] rest_authentication: add api client authentication
 (#67085)

 debian/debian_config_common.py    |   5 +-
 hobo/rest_authentication.py       |  66 +++++++++++++++++
 hobo/test_urls.py                 |  12 ++++
 tests/settings.py                 |   8 +++
 tests/test_rest_authentication.py | 115 ++++++++++++++++++++++++++++++
 tox.ini                           |   1 +
 6 files changed, 206 insertions(+), 1 deletion(-)
 create mode 100644 tests/test_rest_authentication.py

     if 'rest_framework' in INSTALLED_APPS:
         if 'REST_FRAMEWORK' not in globals():
             REST_FRAMEWORK = {}
         REST_FRAMEWORK['DEFAULT_AUTHENTICATION_CLASSES'] = ('hobo.rest_authentication.PublikAuthentication',)
         REST_FRAMEWORK['DEFAULT_AUTHENTICATION_CLASSES'] = (
             'hobo.rest_authentication.PublikAuthentication',
             'hobo.rest_authentication.APIClientAuthentication',
+        )
         REST_FRAMEWORK['DEFAULT_PERMISSION_CLASSES'] = ('rest_framework.permissions.IsAuthenticated',)
         REST_FRAMEWORK['DEFAULT_RENDERER_CLASSES'] = ('rest_framework.renderers.JSONRenderer',)

     import logging
     import requests
     from django.conf import settings
     from django.contrib.auth import get_user_model
     from django.contrib.auth.models import AnonymousUser
-...
     from rest_framework import authentication, exceptions, status
     from hobo import signature
     from hobo.requests_wrapper import Requests
     try:
         from mellon.models import UserSAMLIdentifier
-...
             return 'Publik Service Admin'
     class APIClientUser:
         is_active = True
         is_anonymous = False
         is_authenticated = True
         is_superuser = False
         roles = []
         def __init__(self, is_active, is_anonymous, is_authenticated, is_superuser, roles):
             self.is_active = is_active
             self.is_anonymous = is_anonymous
             self.is_authenticated = is_authenticated
             self.is_superuser = is_superuser
             self.roles = roles
         @classmethod
         def from_dict(cls, data):
             return cls(
                 is_active=data['is_active'],
                 is_anonymous=data['is_anonymous'],
                 is_authenticated=data['is_authenticated'],
                 is_superuser=data['is_superuser'],
                 roles=data['roles'],
+            )
     class PublikAuthenticationFailed(exceptions.APIException):
         status_code = status.HTTP_401_UNAUTHORIZED
         default_code = 'invalid-signature'
-...
             user = self.resolve_user(request)
             self.logger.info('user authenticated with signature %s', user)
             return (user, None)
     class APIClientAuthenticationUnavailable(exceptions.APIException):
         status_code = status.HTTP_503_SERVICE_UNAVAILABLE
         default_code = 'IDP temporarily unavailable, try again later.'
         def __init__(self):
             self.detail = {'err': 1, 'err_desc': self.default_code}
     class APIClientAuthentication(authentication.BasicAuthentication):
         def authenticate_credentials(self, identifier, password, request=None):
             idp_services = list(getattr(settings, 'KNOWN_SERVICES', {}).get('authentic', {}).values())
             if not idp_services:
                 return None
             authentic = idp_services[0]
             url = authentic['url'] + 'api/check-api-client/'
             try:
                 response = Requests().post(url, json={'identifier': identifier, 'password': password})
             except requests.Timeout:
                 raise APIClientAuthenticationUnavailable()
             except requests.RequestException as err:
                 raise APIClientAuthenticationUnavailable()
             try:
                 response.raise_for_status()
             except requests.exceptions.RequestException:
                 return None
             result = response.json()
             if 'err' not in result or 'data' not in result or result['err'] == 1:
                 return None
             try:
                 api_client = APIClientUser.from_dict(result['data'])
             except Exception:
                 return None
             return api_client, None

     from django.conf.urls import url
     from django.core.exceptions import PermissionDenied
     from django.http import HttpResponse
     from rest_framework import permissions
     from rest_framework.response import Response
     from rest_framework.views import APIView
     def helloworld(request):
-...
         return HttpResponse('Hello world %s' % request.META['REMOTE_ADDR'])
     class AuthenticatedTestView(APIView):
         permission_classes = [permissions.IsAuthenticated]
         def get(self, request):
             return Response({'some': 'data'})
     urlpatterns = [
         url(r'^$', helloworld),
         url(r'^authenticated-testview/', AuthenticatedTestView.as_view()),
+    ]

+    }
     TEMPLATES[0]['OPTIONS'].setdefault('builtins', []).append('hobo.templatetags.hobo')
     REST_FRAMEWORK = {
         'DEFAULT_AUTHENTICATION_CLASSES': (
             'hobo.rest_authentication.PublikAuthentication',
             'hobo.rest_authentication.APIClientAuthentication',
         ),
         'DEFAULT_RENDERER_CLASSES': ('rest_framework.renderers.JSONRenderer',),
+    }

     import base64
     import pytest
     import requests
     import responses
     from django.test import RequestFactory
     @pytest.fixture
     def settings_with_idp(settings):
         settings.ROOT_URLCONF = 'hobo.test_urls'
         settings.KNOWN_SERVICES = {
             'authentic': {
                 'idp': {
                     'title': 'Foobar',
                     'url': 'https://idp.example.invalid/',
                     'orig': 'example.org',
                     'secret': 'xxx',
+                }
+            }
+        }
         return settings
     @pytest.fixture
     def app_with_auth(app):
         app.authorization = ('Basic', ('foo', 'bar'))
         return app
     def test_no_known_services(app_with_auth, db, settings):
         settings.ROOT_URLCONF = 'hobo.test_urls'
         resp = app_with_auth.get('/authenticated-testview/', status=403)
     def test_no_idp_in_known_services(app_with_auth, db, settings):
         settings.ROOT_URLCONF = 'hobo.test_urls'
         settings.KNOWN_SERVICES = {}
         resp = app_with_auth.get('/authenticated-testview/', status=403)
     def test_idp_connection_error(app_with_auth, db, settings_with_idp):
         with responses.RequestsMock() as rsps:
             rsps.post('https://idp.example.invalid/api/check-api-client/', status=403)
             resp = app_with_auth.get('/authenticated-testview/', status=403)
     def test_idp_timeout(app_with_auth, db, settings_with_idp):
         with responses.RequestsMock() as rsps:
             rsps.post('https://idp.example.invalid/api/check-api-client/', body=requests.Timeout('...'))
             resp = app_with_auth.get('/authenticated-testview/', status=503)
             assert resp.json == {'err': 1, 'err_desc': 'IDP temporarily unavailable, try again later.'}
     def test_idp_connection_error(app_with_auth, db, settings_with_idp):
         with responses.RequestsMock() as rsps:
             rsps.post('https://idp.example.invalid/api/check-api-client/', body=requests.RequestException('...'))
             resp = app_with_auth.get('/authenticated-testview/', status=503)
             assert resp.json == {'err': 1, 'err_desc': 'IDP temporarily unavailable, try again later.'}
     def test_idp_no_err_key(app_with_auth, db, settings_with_idp):
         with responses.RequestsMock() as rsps:
             rsps.post(
                 'https://idp.example.invalid/api/check-api-client/',
                 json={'foo': 'bar'},
                 status=200,
+            )
             resp = app_with_auth.get('/authenticated-testview/', status=403)
     def test_idp_app_error(app_with_auth, db, settings_with_idp):
         with responses.RequestsMock() as rsps:
             rsps.post(
                 'https://idp.example.invalid/api/check-api-client/',
                 json={'err': 1},
                 status=200,
+            )
             resp = app_with_auth.get('/authenticated-testview/', status=403)
     def test_idp_wrong_serialization(app_with_auth, db, settings_with_idp):
         with responses.RequestsMock() as rsps:
             rsps.post(
                 'https://idp.example.invalid/api/check-api-client/',
                 json={'err': 0, 'data': {'foo': 'bar'}},
                 status=200,
+            )
             resp = app_with_auth.get('/authenticated-testview/', status=403)
     def test_no_credentials(app, db, settings_with_idp):
         # test that the '/authenticated-testview/' really requires authentication,
         # otherwise all the others tests are meaningless.
         resp = app.get('/authenticated-testview/', status=403)
     def test_access_granted(app_with_auth, db, settings_with_idp):
         with responses.RequestsMock() as rsps:
             rsps.post(
                 'https://idp.example.invalid/api/check-api-client/',
                 json={
                     'err': 0,
                     'data': {
                         'is_active': True,
                         'is_anonymous': False,
                         'is_authenticated': True,
                         'is_superuser': False,
                         'restrict_to_anonymised_data': False,
                         'roles': [],
                     },
                 },
                 status=200,
+            )
             resp = app_with_auth.get('/authenticated-testview/')

     	mock<4
     	httmock
     	requests
             responses
     	pytest-freezegun
     	xmlschema<1.1
     	enum34<=1.1.6
+    -

Projet

Général

Profil

Produits Entr'ouvert » Hobo

0002-rest_authentication-add-api-client-authentication-67.patch