451 lines
14 KiB
Python
451 lines
14 KiB
Python
""" """
|
|
|
|
# pyright: basic
|
|
|
|
from django.shortcuts import get_object_or_404, redirect, resolve_url
|
|
from django.views.generic import RedirectView
|
|
from django.views.generic.detail import DetailView, SingleObjectMixin
|
|
from django.views.generic.list import ListView
|
|
from django.views.generic.edit import CreateView, UpdateView
|
|
from django.core.exceptions import SuspiciousOperation
|
|
from django.http import Http404, HttpResponseRedirect
|
|
from django.http.request import HttpRequest
|
|
|
|
from markdown2 import markdown
|
|
|
|
from . import models, forms
|
|
from interface.utils import check_signed_url
|
|
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AuthorizedResourceMixin(object):
|
|
"""
|
|
Handles these parts of the permission system:
|
|
* signed urls
|
|
* persistent authenticated resources
|
|
* Admin enforcing
|
|
"""
|
|
|
|
request: HttpRequest
|
|
kwargs: dict
|
|
_authorized: dict
|
|
|
|
SESSION_KEY = "authorized"
|
|
admin_required = False
|
|
|
|
def is_authorized(self):
|
|
"By default check if superuser or a signed request"
|
|
if self.request.is_admin: # type: ignore
|
|
# logger.debug("is_authorized: superuser")
|
|
return True
|
|
|
|
if "sig" in self.request.GET:
|
|
check_signed_url(self.request.get_full_path())
|
|
self.on_signed_request()
|
|
return True
|
|
|
|
return False
|
|
|
|
def on_signed_request(self):
|
|
pass
|
|
|
|
def is_authorized_key(self, resource, key, auth):
|
|
current = self.get_authorized_keys(resource).get(str(key), None)
|
|
if current is None:
|
|
# logger.debug("is_authorized_key: %s %s not in session", resource, key)
|
|
return False
|
|
if auth == current:
|
|
return True
|
|
|
|
# logger.info("Authorisation revoked")
|
|
self.del_authorized_key(resource, key)
|
|
return False
|
|
|
|
def get_authorized_keys(self, resource):
|
|
"Returns a set of authorized keys for this resource"
|
|
return self._authorized.get(resource, {})
|
|
|
|
def add_authorized_key(self, resource, key, auth):
|
|
"Adds a key to the authorized list for this resource"
|
|
current = self.get_authorized_keys(resource)
|
|
current[str(key)] = auth
|
|
self._authorized[resource] = current
|
|
self.request.session[self.SESSION_KEY] = self._authorized # type: ignore
|
|
|
|
def del_authorized_key(self, resource, key):
|
|
logger.info("Revoking authorization for %s %s", resource, key)
|
|
current = self.get_authorized_keys(resource)
|
|
if current.pop(key, None) is None:
|
|
return False
|
|
if current:
|
|
self._authorized[resource] = current
|
|
else:
|
|
self._authorized.pop(resource)
|
|
self.request.session[self.SESSION_KEY] = self._authorized # type: ignore
|
|
return True
|
|
|
|
def request_denied(self):
|
|
raise Http404(
|
|
"Either the given resource doesn't exist or you dont have access to it."
|
|
)
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
self._authorized = request.session.get("authorized", {})
|
|
request.is_admin = request.user.is_superuser
|
|
|
|
if not self.is_authorized():
|
|
return self.request_denied()
|
|
|
|
if self.admin_required and not request.is_admin:
|
|
return self.request_denied()
|
|
|
|
return super().dispatch(request, *args, **kwargs) # type: ignore
|
|
|
|
|
|
# TODO: RevokeResourceView - increment nonce
|
|
|
|
|
|
class ForgetResourceView(AuthorizedResourceMixin, RedirectView):
|
|
def is_authorized(self):
|
|
return True
|
|
|
|
def get_redirect_url(self, resource, key):
|
|
self.del_authorized_key(resource, key)
|
|
return "/"
|
|
|
|
|
|
class EnsembleMixin(AuthorizedResourceMixin):
|
|
ensemble_slug_kwarg = "ensemble"
|
|
limited_project_access = False
|
|
|
|
def is_authorized(self):
|
|
ensemble_slug = self.kwargs[self.ensemble_slug_kwarg]
|
|
self.ensemble = get_object_or_404(models.Ensemble, slug=ensemble_slug)
|
|
|
|
if super().is_authorized():
|
|
return True
|
|
|
|
if self.ensemble.has_admin(self.request.user): # type: ignore
|
|
self.request.is_admin = True # type: ignore
|
|
return True
|
|
|
|
if self.is_authorized_key("ensemble", ensemble_slug, self.ensemble.nonce):
|
|
return True
|
|
|
|
authorized = set([int(x) for x in self.get_authorized_keys("project").keys()])
|
|
projects = set(self.ensemble.projects.values_list("pk", flat=True))
|
|
logger.debug("is_authorized: %r & %r", authorized, projects)
|
|
if authorized & projects:
|
|
self.limited_project_access = True
|
|
logger.debug("is_authorized: allowing due to project link")
|
|
return True
|
|
return False
|
|
|
|
def get_object(self, queryset=None):
|
|
return self.ensemble
|
|
|
|
|
|
class ProjectMixin(AuthorizedResourceMixin):
|
|
project_kwarg = "project"
|
|
|
|
def is_authorized(self):
|
|
project_id = self.kwargs[self.project_kwarg]
|
|
self.project = get_object_or_404(
|
|
models.Project.objects.select_related("ensemble"), pk=project_id
|
|
)
|
|
|
|
if super().is_authorized():
|
|
return True
|
|
|
|
# check if the current user is an admin on the ensemble
|
|
if self.project.ensemble.has_admin(self.request.user): # type: ignore
|
|
logger.debug("is_authorized: ensemble admin for project")
|
|
self.request.is_admin = True # type: ignore
|
|
return True
|
|
|
|
if self.is_authorized_key(
|
|
"ensemble", self.project.ensemble.pk, self.project.ensemble.nonce
|
|
):
|
|
logger.debug("is_authorized: has ensemble link for project")
|
|
return True
|
|
|
|
if self.is_authorized_key("project", project_id, self.project.nonce):
|
|
logger.debug("is_authorized: has project link")
|
|
return True
|
|
|
|
return False
|
|
|
|
# filter any generated querysets
|
|
def get_queryset(self):
|
|
return super().get_queryset().filter(project=self.project) # type: ignore
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs) # type: ignore
|
|
if "project" in self.kwargs:
|
|
context["project"] = self.project
|
|
context["modules"] = self.project.active_modules
|
|
return context
|
|
|
|
|
|
class CrispyFormMixin(object):
|
|
cancel_url = None
|
|
|
|
def get_cancel_url(self):
|
|
return self.cancel_url
|
|
|
|
|
|
""" ENSEMBLE VIEWS """
|
|
|
|
|
|
class EnsembleListView(AuthorizedResourceMixin, ListView):
|
|
def is_authorized(self):
|
|
return True
|
|
|
|
def get_queryset(self):
|
|
return models.Ensemble.objects.for_user(
|
|
self.request.user, # type: ignore
|
|
self.get_authorized_keys("ensemble").keys(),
|
|
self.get_authorized_keys("project").keys(),
|
|
)
|
|
|
|
|
|
class EnsembleDetailView(EnsembleMixin, DetailView):
|
|
def request_denied(self):
|
|
if "auth" in self.request.GET:
|
|
if self.request.GET["auth"] != self.ensemble.auth():
|
|
raise SuspiciousOperation("Bad ensemble link")
|
|
self.add_authorized_key("ensemble", self.ensemble.slug, self.ensemble.nonce)
|
|
return HttpResponseRedirect(
|
|
resolve_url("ensemble_detail", self.ensemble.slug)
|
|
)
|
|
|
|
return super().request_denied()
|
|
|
|
def get_context_data(self, **kwargs):
|
|
data = super().get_context_data(**kwargs)
|
|
inactive = "inactive" in self.request.GET and self.request.is_admin # type: ignore
|
|
|
|
projects = self.ensemble.projects.all()
|
|
if self.limited_project_access:
|
|
projects = projects.filter(
|
|
pk__in=self.get_authorized_keys("project").keys()
|
|
)
|
|
|
|
if inactive:
|
|
projects = projects.order_by("-pk")
|
|
else:
|
|
projects = projects.active().current()
|
|
|
|
data["inactive"] = inactive
|
|
data["object_list"] = projects
|
|
if self.request.is_admin: # type: ignore
|
|
data["ensemble_link"] = self.request.path + "?auth=" + self.ensemble.auth()
|
|
return data
|
|
|
|
|
|
class EnsembleRevokeView(SingleObjectMixin, RedirectView):
|
|
def get_redirect_url(self):
|
|
return
|
|
|
|
|
|
""" PROJECT VIEWS """
|
|
|
|
|
|
class ProjectListView(ProjectMixin, ListView):
|
|
def is_authorized(self):
|
|
return True
|
|
|
|
def get_project_queryset(self):
|
|
return models.Project.objects.for_user(
|
|
self.request.user, # type: ignore
|
|
self.get_authorized_keys("project"),
|
|
self.get_authorized_keys("ensemble"),
|
|
)
|
|
|
|
def get_queryset(self):
|
|
return self.get_project_queryset().current().active()
|
|
|
|
|
|
class ProjectCreateView(EnsembleMixin, CreateView):
|
|
admin_required = True
|
|
model = models.Project
|
|
template_name = "interface/default_form.html"
|
|
title = "Add a new project"
|
|
form_class = forms.ProjectForm
|
|
|
|
def form_valid(self, form):
|
|
self.object = form.save(commit=False)
|
|
self.object.ensemble = self.ensemble
|
|
self.object.owner = self.request.user # type: ignore
|
|
self.object.save()
|
|
|
|
for module in form.cleaned_data["modules"]:
|
|
self.object.modules.create(name=module)
|
|
|
|
return redirect("project_detail", project=self.object.pk)
|
|
|
|
|
|
class ProjectDetailView(ProjectMixin, DetailView):
|
|
def request_denied(self):
|
|
if "auth" in self.request.GET:
|
|
if self.request.GET["auth"] != self.project.auth():
|
|
raise SuspiciousOperation("Bad project link")
|
|
self.add_authorized_key("project", self.project.pk, self.project.nonce)
|
|
return HttpResponseRedirect(resolve_url("project_detail", self.project.pk))
|
|
return super().request_denied()
|
|
|
|
def get_object(self, queryset=None):
|
|
return self.project
|
|
|
|
def get_context_data(self, **kwargs):
|
|
data = super().get_context_data(**kwargs)
|
|
if self.request.is_admin: # type: ignore
|
|
# data['project_link'] = signed_url(f'{self.request.path}?nonce={self.project.nonce}')
|
|
data["project_link"] = self.request.path + "?auth=" + self.project.auth()
|
|
return data
|
|
|
|
|
|
class ProjectUpdateView(ProjectMixin, UpdateView):
|
|
admin_required = True
|
|
template_name = "interface/default_form.html"
|
|
pk_url_kwarg = "project"
|
|
form_class = forms.ProjectForm
|
|
|
|
def get_object(self):
|
|
return self.project
|
|
|
|
def get_initial(self):
|
|
data = super().get_initial()
|
|
data["modules"] = self.object.active_modules
|
|
print(data)
|
|
return data
|
|
|
|
def form_valid(self, form):
|
|
self.object = form.save()
|
|
current = set(self.object.active_modules)
|
|
desired = set(form.cleaned_data["modules"])
|
|
self.object.modules.exclude(name__in=desired).delete()
|
|
for module in desired - current:
|
|
self.object.modules.create(name=module)
|
|
return redirect("project_detail", self.kwargs["project"])
|
|
|
|
@property
|
|
def cancel_url(self):
|
|
return resolve_url("project_detail", self.kwargs["project"])
|
|
|
|
|
|
# Old Makefile from submission module
|
|
# class ProjectMakefileView(EnsembleMixin, DetailView):
|
|
# template_name = 'interface/project_submissions.mk'
|
|
# content_type = 'text/plain'
|
|
#
|
|
# def get_queryset(self):
|
|
# if self.request.is_admin:
|
|
# return models.Project.objects.all()
|
|
#
|
|
# return models.Project.objects.filter(ensemble=self.request.ensemble_id)
|
|
#
|
|
# def get_context_data(self, **kwargs):
|
|
# data = super().get_context_data(**kwargs)
|
|
#
|
|
# data['submissions'] = []
|
|
# data['targets'] = []
|
|
# for s in self.object.submissions:
|
|
# name = s.short_name
|
|
# data['targets'].append(name)
|
|
# data['submissions'].append({
|
|
# 'url': self.request.build_absolute_uri(signed_url('submission_download', project=self.kwargs['pk'], pk=s.pk)),
|
|
# 'name': name,
|
|
# })
|
|
#
|
|
# return data
|
|
|
|
""" WIKI VIEWS """
|
|
|
|
|
|
class WikiView(ProjectMixin, DetailView):
|
|
template_name = "interface/wiki.html"
|
|
model = models.WikiPage
|
|
|
|
def get_context_data(self, **kwargs):
|
|
data = super().get_context_data(**kwargs)
|
|
data["wiki_html"] = markdown(self.object.markdown)
|
|
return data
|
|
|
|
|
|
class WikiCreateView(ProjectMixin, CreateView):
|
|
admin_required = True
|
|
model = models.WikiPage
|
|
template_name = "interface/default_form.html"
|
|
form_class = forms.WikiForm
|
|
|
|
def cancel_url(self):
|
|
return resolve_url("project_detail", self.kwargs["project"])
|
|
|
|
def form_valid(self, form):
|
|
self.object = form.save(commit=False)
|
|
self.object.project = self.project
|
|
self.object.save()
|
|
return redirect("wiki", project=self.object.project_id, pk=self.object.pk)
|
|
|
|
|
|
class WikiEditView(ProjectMixin, UpdateView):
|
|
admin_required = True
|
|
model = models.WikiPage
|
|
form_class = forms.WikiForm
|
|
|
|
def cancel_url(self):
|
|
return resolve_url("wiki", self.kwargs["project"], self.kwargs["pk"])
|
|
|
|
|
|
""" RESOURCE VIEWS """
|
|
|
|
|
|
class ResourceCreateView(ProjectMixin, CreateView):
|
|
admin_required = True
|
|
model = models.Resource
|
|
form_class = forms.ResourceForm
|
|
template_name = "interface/default_form.html"
|
|
title = "Add a new resource"
|
|
|
|
def form_valid(self, form):
|
|
self.object = form.save(commit=False)
|
|
self.object.project = self.project
|
|
self.object.save()
|
|
return redirect(
|
|
"resource_upload", project=self.object.project_id, pk=self.object.pk
|
|
)
|
|
|
|
|
|
class ResourceUploadView(ProjectMixin, UpdateView):
|
|
admin_required = True
|
|
model = models.Resource
|
|
fields = ["file"]
|
|
template_name = "interface/default_form.html"
|
|
|
|
def get_success_url(self):
|
|
return resolve_url("resource_list", project=self.kwargs["project"])
|
|
|
|
|
|
class ResourceListView(ProjectMixin, ListView):
|
|
model = models.Resource
|
|
|
|
def get_queryset(self):
|
|
qs = super().get_queryset()
|
|
if not self.request.is_admin: # type: ignore
|
|
qs = qs.filter(visible=True)
|
|
return qs
|
|
|
|
|
|
class ResourceEditView(ProjectMixin, UpdateView):
|
|
admin_required = True
|
|
model = models.Resource
|
|
fields = ["name", "description", "visible"]
|
|
template_name = "interface/default_form.html"
|
|
|
|
def get_success_url(self):
|
|
return resolve_url("resource_list", project=self.kwargs["project"])
|