Commit fbc1f98b authored by BAIRE Anthony's avatar BAIRE Anthony
Browse files

reimplement job access control using UserPassesTestMixin

- use the same mixin for the api and the UI (so that the authorization
  code is not duplicated)
- renamed JobAuthorizationMixin as JobAuthMixin (because it is easier
  to type ;-)
- LoginRequiredMixin is no longer required because authentication
  (and redirection to the login page) is handled by UserPassesTestMixin
parent 58e759b9
Pipeline #41291 failed with stage
in 58 seconds
......@@ -6,20 +6,3 @@ from main.helpers import get_request_user
from main.models import Job
class JobAuthorizationMixin(object):
def dispatch(self, request, *args, **kwargs):
"""Authorize a user to see its jobs
Checks if a user has the right to see a specific job. If not, a 403
HTTP error is returned.
Only super users can see all jobs.
user = get_user(request)
job = Job.objects.get(pk=args[0])
if == str(user):
return super().dispatch(request, *args, **kwargs)
return HttpResponse(status=403)
......@@ -6,6 +6,6 @@ app_name = 'api'
urlpatterns = [
url(r'^jobs$',, name='jobs'),
url(r'^jobs/(\d+)', views.APIJobView.as_view(), name='job'),
url(r'^datastore/(\d+)/(.*)/(.*)', views.APIDownloadView, name='download'),
url(r'^jobs/(?P<pk>\d+)', views.APIJobView.as_view(), name='job'),
url(r'^datastore/(?P<pk>\d+)/(.*)/(.*)', views.APIDownloadView, name='download'),
......@@ -12,7 +12,7 @@ from django.views.generic import View
from main.models import Job, Webapp, JobQueue
from main.helpers import upload_data, get_base_url, lookup_job_file, get_request_user
from .mixins import JobAuthorizationMixin
from main.mixins import JobAuthMixin
log = logging.getLogger('allgo')
......@@ -41,23 +41,11 @@ def job_response(job, request):
class APIJobView(JobAuthMixin, View):
class APIJobView(JobAuthorizationMixin, View):
def get(self, request, *args, **kwargs):
user = get_user(request)
if not user:"API request without http authorisation %s %s %s", request.META['HTTP_USER_AGENT'],
request.META['REMOTE_ADDR'], request.META['QUERY_STRING'])
return HttpResponse(status=401)
job = Job.objects.get(id=int(args[0]), user=user.user)
if not job:
log.error("Job %s does not exist", jobid)
return HttpResponse(status=404)
return JsonResponse(job_response(job, request))
def get(self, request, pk):
job = Job.objects.get(id=pk)
return JsonResponse(job_response(job, request))
......@@ -105,7 +93,7 @@ def jobs(request):
return JsonResponse(job_response(job, request))
class APIDownloadView(JobAuthorizationMixin, View):
class APIDownloadView(JobAuthMixin, View):
def get(self, request, *args, **kwargs):
jobid = args[0]
from django.conf import settings
from django.contrib.auth.mixins import UserPassesTestMixin
from django.core.exceptions import PermissionDenied
from django.http import HttpResponse
from .models import Job
from .helpers import get_request_user
class IsProviderMixin(object):
......@@ -17,19 +19,18 @@ class IsProviderMixin(object):
raise PermissionDenied
class JobAuthorizationMixin(object):
"""Check authorization to access a given job
def dispatch(self, request, *args, **kwargs):
"""Authorize a user to see its jobs
class JobAuthMixin(UserPassesTestMixin):
"""Check authorization to access a given job"""
Checks if a user has the right to see a specific job. If not, a 403
HTTP error is returned.
Only super users can see all jobs.
def test_func(self):
"""Check if user has access to a job
- redirects to the login page if unauthenticated
- allow access if user is the job owner or if user is a superuser
job = Job.objects.get(pk=kwargs['pk'])
if job.user == request.user or request.user.is_superuser:
return super().dispatch(request, *args, **kwargs)
return HttpResponse(status=403)
user = get_request_user(self.request)
if user is None:
return False
self.raise_exception = True # to return a 403
job = Job.objects.filter(pk=self.kwargs['pk']).first()
return user.is_superuser or user == getattr(job, "user", ())
......@@ -63,7 +63,7 @@ from .forms import (
# Local imports
import config
from .helpers import get_base_url, get_ssh_data, upload_data, notify_controller, lookup_job_file, get_request_user
from .mixins import IsProviderMixin, JobAuthorizationMixin
from .mixins import IsProviderMixin, JobAuthMixin
from .models import (
......@@ -667,7 +667,7 @@ class JobList(LoginRequiredMixin, ListView):
return queryset
class JobDetail(LoginRequiredMixin, JobAuthorizationMixin, DetailView):
class JobDetail(JobAuthMixin, DetailView):
"""Get a job detail for a specific user
......@@ -844,7 +844,7 @@ class JobCreate(SuccessMessageMixin, CreateView):
kwargs['webapp'] = queryset
return kwargs
class JobAbort(LoginRequiredMixin, View):
class JobAbort(JobAuthMixin, View):
def post(self, request, *, pk):
job_id = int(pk)
# switch state to ABORTING if the job is running (this is done
......@@ -860,7 +860,7 @@ class JobAbort(LoginRequiredMixin, View):
class JobDelete(LoginRequiredMixin, JobAuthorizationMixin, DeleteView):
class JobDelete(JobAuthMixin, DeleteView):
"""Delete a job from the database
......@@ -920,7 +920,7 @@ class JobDelete(LoginRequiredMixin, JobAuthorizationMixin, DeleteView):
class JobFileDownload(LoginRequiredMixin, JobAuthorizationMixin, View):
class JobFileDownload(JobAuthMixin, View):
"""Download a given file"""
def get(self, request, *args, **kwargs):
......@@ -932,7 +932,7 @@ class JobFileDownload(LoginRequiredMixin, JobAuthorizationMixin, View):
return redirect("/datastore/%s/%s" % (job_id, filename))
class JobFileDownloadAll(LoginRequiredMixin, JobAuthorizationMixin, View):
class JobFileDownloadAll(JobAuthMixin, View):
"""Archive and download all files of a given job
model = Job
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