Commit deb3c320 authored by LETORT Sebastien's avatar LETORT Sebastien

* Cors corrections

add the token as hidden input in the html form.
add the cors requirement in the fetch API js file.
Set the token value in the context.
Copy-paste the helpers.get_request_user function from test_api_job branch to be cors compliant.
Remove useless api mixin (cors effect), add message if no authentication could be done.
Update test to take cors into account and use the same test construction as in test_job_api. -> We need to build a mother class (TestApi ?) when branches merged.
Test are included in the integration script.

* little fix
DATE_MIN is timezone aware now. (I encounter a tz pb, don't understand how it appeared/disappeared.)
parent ef559183
......@@ -28,12 +28,12 @@ django_pylint:
- docker exec -i dev-django pylint3 --rcfile=.pylintrc allgo
allow_failure: true
#~ django_test:
#~ stage: test
#~ # only:
#~ # - /django/
#~ script:
#~ - docker exec -i dev-django python3 test
stage: test
# only:
# - /django/
- docker exec -i dev-django python3 test allgo.api.tests.test_ApiMetrics
stage: cleanning
......@@ -3,61 +3,98 @@
## AIMS : test the metric API.
## USAGE :
import logging
from unittest import skip
from datetime import datetime
import json
import logging
import random
from datetime import datetime
from unittest import skip
from django.contrib.auth.models import User, AnonymousUser
from django.core.exceptions import PermissionDenied
from django.test import RequestFactory, TestCase, tag
from django.contrib.auth.models import User
from django.http import JsonResponse
from django.test import RequestFactory, TestCase, tag
from django.urls import reverse
from main.models import AllgoUser
from main.tests import populate_db
from api.v1.views import Metrics
from main.tests import populate_db
log = logging.getLogger('test')
LOG = logging.getLogger('test')
# ======================================
class ApiMetricsTestCase(TestCase):
#FIXME: timzone pb with tags format & result
# Database returned an invalid datetime value. Are time zone definitions for your database installed?
APP_NAME = 'Fake'
API_URL = '/api/v1/metrics'
ROUTE_NAME = 'api:metrics'
def setUpTestData(cls):":> setUpTestData.\nCreate 4 users, one own the app, and various jobs are created.")":> setUpTestData.\nCreate 4 users, one own the app, and various jobs are created.")
populate_db.with_users([ 'Bob', 'Liz', 'Zaza', 'Lol' ])
populate_db.with_webapps({ 'Bob': [cls.APP_NAME] })
populate_db.with_users([ cls.USERNAME, 'Liz', 'Zaza', 'Lol' ])
populate_db.with_webapps({ cls.USERNAME: [cls.APP_NAME] })
def setUp(self):":> setUp.")
# Every test needs access to the request factory.
self.request = RequestFactory().get(self.API_URL)":> setUp.")
self._factory = RequestFactory()
self.req_factory = RequestFactory()
def route_name(self):
return self.ROUTE_NAME
def _build_url(cls, params=None):
def _build_url(self, **params):
"""params is a dictionnary"""
if params is None:
return cls.API_URL
url = reverse(self.route_name(), kwargs=params)
return url
def _define_api_user_with_token(self, username):
"""Identify the user by the token only.
CORS policy force the user to be authenticated by its token only.
We though define the user as AnonymousUser because AllgoValidAccountMixin require a user to be set in the request.
-> All this should be set in the request factory."""
self._request.user = AnonymousUser()
user_token = AllgoUser.objects.get(user__username=username).token
self._request.META['HTTP_AUTHORIZATION'] = "Token token={}".format(user_token)
def create_request(self,
"""Create the _request attribute, an HttpRequest object through the RequestFactory object.
If a user is defined, it is associated to the request.""" "create_request" )
if None is app_id:
app_id = self.SLEEP_ID
args = []
for k,v in params.items():
args.append( "{}={}".format(k,v) )
url = self._build_url(what=what, app_id=app_id)
self._request = self._factory.get(url, interval)
api_url = cls.API_URL + "=?" + "&".join(args)
return api_url
# I have to set no user
# ~ if not user:
# ~ user = self.USERNAME
if user:
def create_view(self, what):
SLEEP_ID = '1'
return Metrics.as_view()(self.request, SLEEP_ID, what)
def create_view(self, what='per_user', app_id=None): "create_view with *{}*".format(what) )
if None is app_id:
app_id = self.SLEEP_ID
if not hasattr(self, '_request'):
self.create_request(what=what, app_id=app_id) "request is {}".format(str(self._request)) )
return Metrics.as_view()(self._request, app_id, what)
def assert_json_struct(self, json_):
"""json_ is a dictionnary of json data"""
......@@ -75,14 +112,15 @@ class ApiMetricsTestCase(TestCase):
self.assertTrue(isinstance(record[k], type_))
def check_api(self, what, record_fmt):
url = self._build_url()
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Bob')
"""assert that the api return the correct json structure,
and check the format of a random record."""
self.create_request(user=self.USERNAME, what=what)
o_response = self.create_view(what)
self.assertIsInstance(o_response, JsonResponse)
str_response = o_response.content.decode('utf-8')
LOG.debug("json response = {}".format(str_response))
json_ = json.loads( str_response )
......@@ -94,42 +132,38 @@ class ApiMetricsTestCase(TestCase):
# TODO : tester l'erreur sur faute de format de date dans url
@tag( 'auth' )
def test_access_no_user(self ):
url = self._build_url()
self.request = self.req_factory.get(url)
def test_access_no_user(self):
self.request = self.create_request(user=None)
o_response = self.create_view()
o_response = self.create_view('per_user')
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 401, "without user = not authorized.")
@tag( 'permission' )
def test_access_bad_user(self):
url = self._build_url()
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Lol')
self.request = self.create_request(user='Lol')
LOG.debug("User has been set.")
o_response = self.create_view('per_user')
o_response = self.create_view()
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 403, "user without app = permission denied.")
@tag( 'params' )
def test_access_bad_webapp(self):
url = self._build_url()
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Lol')
self.request = self.create_request(user=self.USERNAME)
o_response = self.create_view(app_id=NON_EXISTING_APP_ID)
o_response = Metrics.as_view()(self.request, NON_EXISTING_APP_ID, 'per_user')
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 404, "Unknown app = not found.")
# test with assertJSONEqual ?
@tag( 'params' )
def test_access_bad_metrics(self):
o_user = User.objects.get(username='Bob')
self.request.user = o_user
self.request = self.create_request(user=self.USERNAME, what=what)
o_response = self.create_view(what=what)
o_response = self.create_view('peureuse')
self.assertIsInstance(o_response, JsonResponse)
self.assertEqual(o_response.status_code, 404, "Unknown metrics type = not found.")
......@@ -158,12 +192,12 @@ class ApiMetricsTestCase(TestCase):
def check_results(self, what, expected):
"""expected is a list of expected dictionnaries"""
url = self._build_url({ 'step': 'month', 'from': '2019-01-01', 'to':'2019-05-01' })
self.request = self.req_factory.get(url)
self.request.user = User.objects.get(username='Bob')
interval = {'step': 'month', 'from': '2019-01-01', 'to':'2019-05-01'}
self.create_request(user=self.USERNAME, what=what, interval=interval)
o_response = self.create_view(what)
str_response = o_response.content.decode('utf-8')
json_ = json.loads(str_response)
data = json_[self.APP_NAME]['data']
......@@ -188,7 +222,7 @@ class ApiMetricsTestCase(TestCase):
self.check_results('per_state', expected)
@tag( 'result' )
@tag( 'results' )
def test_per_user_results(self):
expected = [
{'uname': 'Bob', 'user': 1, 'n': 3, 'time_period': '2019-01-01'},
......@@ -29,7 +29,7 @@ DATE_IN_FMT = "%Y-%m-%d"
def str2date(date):
return datetime.strptime( date, DATE_IN_FMT )
DATE_MIN = str2date('2013-01-01')
DATE_MIN = make_aware(str2date('2013-01-01'))
# SLetort : I used this function to set the default
# because metrics function have all their params, possibly set to None
Mixins used in API (only).
In Django mixins are used to managed permissions.
import logging
from django.http import JsonResponse
from django.contrib.auth.mixins import UserPassesTestMixin
from main.helpers import get_request_user
log = logging.getLogger('allgo')
class ApiAuthMixin(UserPassesTestMixin):
"""API should be accessible uniquely to AllgoUser, ie logged in or with a token.
The user status is checked through get_request_user.
Return a json string with 401 status if unauthorized access.
def test_func(self):
user = get_request_user(self.request)
log.debug( "ApiAuthMixin.test_func user = " + str(user) )
if user is None:
return False
return True
def handle_no_permission(self):
log.debug( "ApiAuthMixin.handle_no_permission" )
# as it should be used only in api, is this test necessary ?
if self.request.path_info.startswith("/api/"):
return JsonResponse({"error": "401 Unauthorized"}, status=401)
return super().handle_no_permission()
......@@ -10,7 +10,6 @@ from django.urls import reverse
from django.views.decorators.csrf import csrf_exempt
from django.views.generic import View
from api.v1.mixins import ApiAuthMixin
import api.v1.metrics as metrics
from main.helpers import upload_data, get_base_url, lookup_job_file, get_request_user
......@@ -47,7 +46,7 @@ class APIJobView(JobAuthMixin, View):
return JsonResponse({'error': 'Job not found'}, status=404)
class Metrics(ApiAuthMixin, View):
class Metrics(View):
"""Metrics view provides only a get access."""
def get(self, request, app_id, what):
......@@ -61,7 +60,16 @@ class Metrics(ApiAuthMixin, View):
# WARN: here request.user is AnonymousUser, but as been recognised before, in ApiAuthMixin
o_user = get_request_user(request) # IMO doublon avec ApiAuthM
# faire une version API de get_request_user ? , ces 5 lignes de codes sont dupliqués dans API-jobs (une autre branche)
o_user = get_request_user(request)
if not o_user:
msg = "API request without http authorisation\n"
msg += "\tagent : {}\n".format(request.META.get('HTTP_USER_AGENT', 'agent_unknown'))
msg += "\tadress : {}\n".format(request.META.get('REMOTE_ADDR', 'remote_adress_unknown'))
msg += "\tquery : {}\n".format(request.META.get('QUERY_STRING', 'query_unknown'))
return JsonResponse({'error': 'API request without http authorisation'}, status=401)
app = Webapp.objects.get(id=app_id) # This is just for debug purpose
msg = "{} is asking for '{}' with app '{}'"\
.format(o_user, what,
......@@ -189,21 +189,25 @@ def get_request_user(request):
a User or None
if False == hasattr(request, 'user'):
return None
if request.user.is_authenticated:
return request.user
mo = re.match("Token token=(\S+)",
request.META.get('HTTP_AUTHORIZATION', ''))
if mo:
return getattr(
# FIXME: user token should have a unicity constraint
"user", None)
# else ?
path = request.path
if path == "/auth":
path = request.META['HTTP_X_ORIGINAL_URI']
if path.startswith("/api/"):
# authenticated by token for API requests
# NOTE: we must NOT authenticate by cookie because the CORS
# configuration in the nginx.conf allows all origins
mo = re.match("Token token=(\S+)",
request.META.get('HTTP_AUTHORIZATION', ''))
if mo:
return getattr(
# FIXME: user token should have a unicity constraint
"user", None)
# authenticated by cookie for other requests
if request.user.is_authenticated:
return request.user
def query_webapps_for_user(user):
......@@ -1502,7 +1502,11 @@ class Metrics(UserAccessMixin, TemplateView):
context = super().get_context_data(**kwargs)
context['apps'] = apps.order_by('name')
context['show_form'] = 0 != len(apps) "Site metrics. User = {}".format(str(self.request.user)) )
# 'get' returns a AllgoUser object
context['token'] = AllgoUser.objects \
.get( \
return context
......@@ -374,7 +374,21 @@ function draw_plots()
for( let which_chart in D_CHARTS )
let API_URL = build_API_url( which_chart );
let http_headers = {
'Authorization': 'Token token='+$('#token').val(),
'Accept': 'application/json'
let init = {
method: "GET",
headers: new Headers(http_headers),
mode: 'cors'
let req = new Request( API_URL, init );
//~ let options = { 'Authorization': "Token token=ss3ksGJWDIyKndXJU7o2LfIVyqnIsFNb"};
//~ fetch(API_URL, options).then(function(response)
if( !response.ok )
......@@ -11,6 +11,7 @@
<p class="container">You do not have access to any Webapp, so nothing to ask for.</p>
{% else %}
<form id="metrics_form" onsubmit="return draw_plots()">
<input type="hidden" id="token" name="token" value="{{token}}">
<label for="app">App</label>
<select id="app">
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment