from __future__ import annotations __copyright__ = """ Copyright (C) 2014 Andreas Kloeckner Copyright (c) 2016 Polyconseil SAS. (the WSGI wrapping bits) Copyright (C) 2019 Isuru Fernando """ __license__ = """ Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import re from typing import ( # noqa TYPE_CHECKING, Any, Dict, List, Optional, Text, Tuple, Union, cast, ) import django.forms as forms import dulwich.client # noqa import paramiko import paramiko.client from crispy_forms.layout import Submit from django import http from django.contrib import messages from django.contrib.auth.decorators import login_required, permission_required from django.core.exceptions import PermissionDenied, SuspiciousOperation from django.db import transaction from django.shortcuts import get_object_or_404, redirect, render # noqa from django.urls import reverse from django.utils.translation import ( gettext, gettext_lazy as _, pgettext, pgettext_lazy, ) from django.views.decorators.csrf import csrf_exempt from django_select2.forms import Select2Widget from dulwich.repo import Repo from course.auth import with_course_api_auth from course.constants import participation_permission as pperm, participation_status from course.models import Course, Participation, ParticipationRole from course.utils import ( course_view, get_course_specific_language_choices, render_course_page, ) from relate.utils import HTML5DateInput, StyledForm, StyledModelForm, string_concat # {{{ for mypy if TYPE_CHECKING: import dulwich.web # noqa from dulwich.client import GitClient # noqa from dulwich.objects import Commit # noqa from course.auth import APIContext # noqa # }}} def _remove_prefix(prefix: bytes, s: bytes) -> bytes: assert s.startswith(prefix) return s[len(prefix):] def transfer_remote_refs( repo: Repo, fetch_pack_result: dulwich.client.FetchPackResult) -> None: valid_refs = [] for ref, sha in fetch_pack_result.refs.items(): if (ref.startswith(b"refs/heads/") and not ref.startswith(b"refs/heads/origin/")): new_ref = b"refs/remotes/origin/"+_remove_prefix(b"refs/heads/", ref) valid_refs.append(new_ref) repo[new_ref] = sha for ref in repo.get_refs().keys(): if ref.startswith(b"refs/remotes/origin/") and ref not in valid_refs: del repo[ref] def get_dulwich_client_and_remote_path_from_course( course: Course) -> Tuple[ Union[dulwich.client.GitClient, dulwich.client.SSHGitClient], bytes]: # noqa ssh_kwargs = {} if course.ssh_private_key: from io import StringIO key_file = StringIO(course.ssh_private_key) ssh_kwargs["pkey"] = paramiko.RSAKey.from_private_key(key_file) def get_dulwich_ssh_vendor(): from dulwich.contrib.paramiko_vendor import ParamikoSSHVendor vendor = ParamikoSSHVendor(**ssh_kwargs) return vendor # writing to another module's global variable: gross! dulwich.client.get_ssh_vendor = get_dulwich_ssh_vendor from dulwich.client import get_transport_and_path client, remote_path = get_transport_and_path( course.git_source) return client, remote_path # {{{ new course setup class CourseCreationForm(StyledModelForm): class Meta: model = Course fields = ( "identifier", "name", "number", "time_period", "start_date", "end_date", "hidden", "listed", "accepts_enrollment", "git_source", "ssh_private_key", "course_root_path", "course_file", "events_file", "enrollment_approval_required", "preapproval_require_verified_inst_id", "enrollment_required_email_suffix", "from_email", "notify_email", "force_lang", ) widgets = { "start_date": HTML5DateInput(), "end_date": HTML5DateInput(), "force_lang": forms.Select( choices=get_course_specific_language_choices()), } def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self.helper.add_input( Submit("submit", _("Validate and create"))) @permission_required("course.add_course") def set_up_new_course(request: http.HttpRequest) -> http.HttpResponse: if request.method == "POST": form = CourseCreationForm(request.POST) if form.is_valid(): new_course = form.save(commit=False) from course.content import get_course_repo_path repo_path = get_course_repo_path(new_course) try: import os os.makedirs(repo_path) repo = None try: with transaction.atomic(): repo = Repo.init(repo_path) client, remote_path = \ get_dulwich_client_and_remote_path_from_course( new_course) fetch_pack_result = client.fetch(remote_path, repo) if not fetch_pack_result.refs: raise RuntimeError(_("No refs found in remote repository" " (i.e. no master branch, no HEAD). " "This looks very much like a blank repository. " "Please create course.yml in the remote " "repository before creating your course.")) transfer_remote_refs(repo, fetch_pack_result) new_sha = repo[b"HEAD"] = fetch_pack_result.refs[b"HEAD"] vrepo = repo if new_course.course_root_path: from course.content import SubdirRepoWrapper vrepo = SubdirRepoWrapper( vrepo, new_course.course_root_path) from course.validation import validate_course_content validate_course_content( # type: ignore vrepo, new_course.course_file, new_course.events_file, new_sha) del vrepo new_course.active_git_commit_sha = new_sha.decode() new_course.save() # {{{ set up a participation for the course creator part = Participation() part.user = request.user part.course = new_course part.status = participation_status.active part.save() part.roles.set([ # created by signal handler for course creation ParticipationRole.objects.get( course=new_course, identifier="instructor") ]) # }}} messages.add_message(request, messages.INFO, _("Course content validated, creation " "succeeded.")) except Exception as e: # Don't coalesce this handler with the one below. We only want # to delete the directory if we created it. Trust me. # Make sure files opened for 'repo' above are actually closed. if repo is not None: # noqa repo.close() # noqa from relate.utils import force_remove_path try: force_remove_path(repo_path) except OSError: messages.add_message(request, messages.WARNING, gettext("Failed to delete unused " "repository directory '%s'.") % repo_path) # We don't raise the OSError thrown by force_remove_path # This is to ensure correct error msg for PY2. raise e else: assert repo is not None repo.close() except Exception as e: from traceback import print_exc print_exc() messages.add_message(request, messages.ERROR, string_concat( _("Course creation failed"), ": %(err_type)s: %(err_str)s") % {"err_type": type(e).__name__, "err_str": str(e)}) else: return redirect( "relate-course_page", new_course.identifier) else: form = CourseCreationForm() return render(request, "generic-form.html", { "form_description": _("Set up new course"), "form": form }) # }}} # {{{ update def is_ancestor_commit( repo: Repo, potential_ancestor: Commit, child: Commit, max_history_check_size: Optional[int] = None) -> bool: queue = [repo[parent] for parent in child.parents] while queue: entry = queue.pop() if entry == potential_ancestor: return True if max_history_check_size is not None: max_history_check_size -= 1 if max_history_check_size == 0: return False queue.extend(repo[parent] for parent in entry.parents) return False ALLOWED_COURSE_REVISIOIN_COMMANDS = [ "fetch", "fetch_update", "update", "fetch_preview", "preview", "end_preview"] def run_course_update_command( request, repo, content_repo, pctx, command, new_sha, may_update, prevent_discarding_revisions): if command not in ALLOWED_COURSE_REVISIOIN_COMMANDS: raise RuntimeError(_("invalid command")) if command.startswith("fetch"): if command != "fetch": command = command[6:] client, remote_path = \ get_dulwich_client_and_remote_path_from_course(pctx.course) fetch_pack_result = client.fetch(remote_path, repo) assert isinstance(fetch_pack_result, dulwich.client.FetchPackResult) transfer_remote_refs(repo, fetch_pack_result) remote_head = fetch_pack_result.refs[b"HEAD"] if prevent_discarding_revisions: # Guard against bad scenario: # Local is not ancestor of remote, i.e. the branches have diverged. if not is_ancestor_commit(repo, repo[b"HEAD"], repo[remote_head], max_history_check_size=20) and \ repo[b"HEAD"] != repo[remote_head]: raise RuntimeError(_("internal git repo has more commits. Fetch, " "merge and push.")) repo[b"HEAD"] = remote_head messages.add_message(request, messages.SUCCESS, _("Fetch successful.")) new_sha = remote_head if command == "fetch": return if command == "end_preview": pctx.participation.preview_git_commit_sha = None pctx.participation.save() messages.add_message(request, messages.INFO, _("Preview ended.")) return # {{{ validate from course.validation import ValidationError, validate_course_content try: warnings = validate_course_content( content_repo, pctx.course.course_file, pctx.course.events_file, new_sha, course=pctx.course) except ValidationError as e: messages.add_message(request, messages.ERROR, _("Course content did not validate successfully: '%s' " "Update not applied.") % str(e)) return else: if not warnings: messages.add_message(request, messages.SUCCESS, _("Course content validated successfully.")) else: messages.add_message(request, messages.WARNING, string_concat( _("Course content validated OK, with warnings: "), "