""" """ # 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"])