__version__ = "0.0.1" import os import shutil import random import subprocess import platform import git import mimetypes import magic import flask import cairosvg import celery import shlex from functools import wraps from datetime import datetime from enum import Enum from cairosvg import svg2png from flask_sqlalchemy import SQLAlchemy from flask_bcrypt import Bcrypt from markupsafe import escape, Markup from flask_migrate import Migrate from PIL import Image from flask_httpauth import HTTPBasicAuth import config from flask_babel import Babel, gettext, ngettext, force_locale _ = gettext n_ = gettext app = flask.Flask(__name__) app.config.from_mapping( CELERY=dict( broker_url=config.REDIS_URI, result_backend=config.REDIS_URI, task_ignore_result=True, ), ) auth = HTTPBasicAuth() app.config["SQLALCHEMY_DATABASE_URI"] = config.DB_URI app.config["SECRET_KEY"] = config.DB_PASSWORD app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False app.config["BABEL_TRANSLATION_DIRECTORIES"] = "i18n" app.config["MAX_CONTENT_LENGTH"] = config.MAX_PAYLOAD_SIZE db = SQLAlchemy(app) bcrypt = Bcrypt(app) migrate = Migrate(app, db) from models import * from misc_utils import * import git_http import jinja_utils import celery_tasks from celery import Celery, Task import celery_integration babel = Babel(app) def get_locale(): if flask.request.cookies.get("language"): return flask.request.cookies.get("language") return flask.request.accept_languages.best_match(config.available_locales) babel.init_app(app, locale_selector=get_locale) with app.app_context(): locale_names = {} for language in config.available_locales: with force_locale(language): # NOTE: Translate this to the language's name in that language, for example in French you would use français locale_names[language] = gettext("English") worker = celery_integration.init_celery_app(app) repositories = flask.Blueprint("repository", __name__, template_folder="templates/repository/") @app.context_processor def default(): username = flask.session.get("username") user_object = User.query.filter_by(username=username).first() return { "logged_in_user": username, "user_object": user_object, "Notification": Notification, "unread": UserNotification.query.filter_by(user_username=username).filter( UserNotification.attention_level > 0).count(), "config": config, "Markup": Markup, "locale_names": locale_names, } @app.route("/") def main(): if flask.session.get("username"): return flask.render_template("home.html") else: return flask.render_template("no-home.html") @app.route("/about/") def about(): return flask.render_template("about.html", platform=platform, version=__version__) @app.route("/language", methods=["POST"]) def set_locale(): response = flask.redirect(flask.request.referrer if flask.request.referrer else "/", code=303) if not flask.request.form.get("language"): response.delete_cookie("language") else: response.set_cookie("language", flask.request.form.get("language")) return response @app.route("/cookie-dismiss") def dismiss_banner(): response = flask.redirect(flask.request.referrer if flask.request.referrer else "/", code=303) response.set_cookie("cookie-banner", "1") return response @app.route("/help/") def help_index(): return flask.render_template("help.html", faqs=config.faqs) @app.route("/settings/", methods=["GET", "POST"]) def settings(): if not flask.session.get("username"): flask.abort(401) if flask.request.method == "GET": user = User.query.filter_by(username=flask.session.get("username")).first() return flask.render_template("user-settings.html", user=user) else: user = User.query.filter_by(username=flask.session.get("username")).first() user.display_name = flask.request.form["displayname"] user.URL = flask.request.form["url"] user.company = flask.request.form["company"] user.company_URL = flask.request.form["companyurl"] user.email = flask.request.form.get("email") if flask.request.form.get( "email") else None user.location = flask.request.form["location"] user.show_mail = True if flask.request.form.get("showmail") else False user.bio = flask.request.form.get("bio") db.session.commit() flask.flash( Markup("" + _("Settings saved")), category="success") return flask.redirect(f"/{flask.session.get('username')}", code=303) @app.route("/favourites/", methods=["GET", "POST"]) def favourites(): if not flask.session.get("username"): flask.abort(401) if flask.request.method == "GET": relationships = RepoFavourite.query.filter_by( user_username=flask.session.get("username")) return flask.render_template("favourites.html", favourites=relationships) @app.route("/notifications/", methods=["GET", "POST"]) def notifications(): if not flask.session.get("username"): flask.abort(401) if flask.request.method == "GET": return flask.render_template("notifications.html", notifications=UserNotification.query.filter_by( user_username=flask.session.get("username"))) @app.route("/accounts/", methods=["GET", "POST"]) def login(): if flask.request.method == "GET": return flask.render_template("login.html") else: if "login" in flask.request.form: username = flask.request.form["username"] password = flask.request.form["password"] user = User.query.filter_by(username=username).first() if user and bcrypt.check_password_hash(user.password_hashed, password): flask.session["username"] = user.username flask.flash( Markup("" + _( "Successfully logged in as {username}").format(username=username)), category="success") return flask.redirect("/", code=303) elif not user: flask.flash(Markup( "" + _( "User not found")), category="alert") return flask.render_template("login.html") else: flask.flash(Markup( "" + _( "Invalid password")), category="error") return flask.render_template("login.html") if "signup" in flask.request.form: username = flask.request.form["username"] password = flask.request.form["password"] password2 = flask.request.form["password2"] email = flask.request.form.get("email") email2 = flask.request.form.get("email2") # repeat email is a honeypot name = flask.request.form.get("name") if not only_chars(username, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"): flask.flash(Markup( _("Usernames may only contain Latin alphabet, numbers, '-' and '_'")), category="error") return flask.render_template("login.html") if username in config.RESERVED_NAMES: flask.flash( Markup( "" + _( "Sorry, {username} is a system path").format( username=username)), category="error") return flask.render_template("login.html") user_check = User.query.filter_by(username=username).first() if user_check: flask.flash( Markup( "" + _( "The username {username} is taken").format( username=username)), category="error") return flask.render_template("login.html") if password2 != password: flask.flash(Markup("" + _( "Make sure the passwords match")), category="error") return flask.render_template("login.html") user = User(username, password, email, name) db.session.add(user) db.session.commit() flask.session["username"] = user.username flask.flash(Markup( "" + _( "Successfully created and logged in as {username}").format( username=username)), category="success") notification = Notification({"type": "welcome"}) db.session.add(notification) db.session.commit() result = celery_tasks.send_notification.delay(notification.id, [username], 1) return flask.redirect("/", code=303) @app.route("/newrepo/", methods=["GET", "POST"]) def new_repo(): if not flask.session.get("username"): flask.abort(401) if flask.request.method == "GET": return flask.render_template("new-repo.html") else: name = flask.request.form["name"] visibility = int(flask.request.form["visibility"]) if not only_chars(name, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"): flask.flash(Markup( "" + _( "Repository names may only contain Latin alphabet, numbers, '-' and '_'")), category="error") return flask.render_template("new-repo.html") user = User.query.filter_by(username=flask.session.get("username")).first() repo = Repo(user, name, visibility) db.session.add(repo) db.session.commit() if not os.path.exists(os.path.join(config.REPOS_PATH, repo.route)): subprocess.run(["git", "init", repo.name], cwd=os.path.join(config.REPOS_PATH, flask.session.get("username"))) flask.flash(Markup(_("Successfully created repository {name}").format(name=name)), category="success") return flask.redirect(repo.route, code=303) @app.route("/logout") def logout(): flask.session.clear() flask.flash(Markup( "" + _("Successfully logged out")), category="info") return flask.redirect("/", code=303) @app.route("//", methods=["GET", "POST"]) def user_profile(username): old_relationship = UserFollow.query.filter_by( follower_username=flask.session.get("username"), followed_username=username).first() if flask.request.method == "GET": user = User.query.filter_by(username=username).first() match flask.request.args.get("action"): case "repositories": repos = Repo.query.filter_by(owner_name=username, visibility=2) return flask.render_template("user-profile-repositories.html", user=user, repos=repos, relationship=old_relationship) case "followers": return flask.render_template("user-profile-followers.html", user=user, relationship=old_relationship) case "follows": return flask.render_template("user-profile-follows.html", user=user, relationship=old_relationship) case _: return flask.render_template("user-profile-overview.html", user=user, relationship=old_relationship) elif flask.request.method == "POST": match flask.request.args.get("action"): case "follow": if username == flask.session.get("username"): flask.abort(403) if old_relationship: db.session.delete(old_relationship) else: relationship = UserFollow( flask.session.get("username"), username ) db.session.add(relationship) db.session.commit() user = db.session.get(User, username) author = db.session.get(User, flask.session.get("username")) notification = Notification({"type": "update", "version": "0.0.0"}) db.session.add(notification) db.session.commit() result = celery_tasks.send_notification.delay(notification.id, [username], 1) db.session.commit() return flask.redirect("?", code=303) @app.route("///") def repository_index(username, repository): return flask.redirect("./tree", code=302) @app.route("/info//avatar") def user_avatar(username): serverUserdataLocation = os.path.join(config.USERDATA_PATH, username) if not os.path.exists(serverUserdataLocation): return flask.render_template("not-found.html"), 404 return flask.send_from_directory(serverUserdataLocation, "avatar.png") @app.route("///raw//") def repository_raw(username, repository, branch, subpath): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) serverRepoLocation = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {serverRepoLocation}") if not os.path.exists(serverRepoLocation): app.logger.error(f"Cannot load {serverRepoLocation}") return flask.render_template("not-found.html"), 404 repo = git.Repo(serverRepoLocation) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() if not repo_data.default_branch: if repo.heads: repo_data.default_branch = repo.heads[0].name else: return flask.render_template("empty.html", remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200 if not branch: branch = repo_data.default_branch return flask.redirect(f"./{branch}", code=302) if branch.startswith("tag:"): ref = f"tags/{branch[4:]}" elif branch.startswith("~"): ref = branch[1:] else: ref = f"heads/{branch}" ref = ref.replace("~", "/") # encode slashes for URL support try: repo.git.checkout("-f", ref) except git.exc.GitCommandError: return flask.render_template("not-found.html"), 404 return flask.send_from_directory(config.REPOS_PATH, os.path.join(username, repository, subpath)) @repositories.route("///tree/", defaults={"branch": None, "subpath": ""}) @repositories.route("///tree//", defaults={"subpath": ""}) @repositories.route("///tree//") def repository_tree(username, repository, branch, subpath): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() if not repo_data.default_branch: if repo.heads: repo_data.default_branch = repo.heads[0].name else: return flask.render_template("empty.html", remote=f"{config.www_protocol}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200 if not branch: branch = repo_data.default_branch return flask.redirect(f"./{branch}", code=302) if branch.startswith("tag:"): ref = f"tags/{branch[4:]}" elif branch.startswith("~"): ref = branch[1:] else: ref = f"heads/{branch}" ref = ref.replace("~", "/") # encode slashes for URL support try: repo.git.checkout("-f", ref) except git.exc.GitCommandError: return flask.render_template("not-found.html"), 404 branches = repo.heads all_refs = [] for ref in repo.heads: all_refs.append((ref, "head")) for ref in repo.tags: all_refs.append((ref, "tag")) if os.path.isdir(os.path.join(server_repo_location, subpath)): files = [] blobs = [] for entry in os.listdir(os.path.join(server_repo_location, subpath)): if not os.path.basename(entry) == ".git": files.append(os.path.join(subpath, entry)) infos = [] for file in files: path = os.path.join(server_repo_location, file) mimetype = guess_mime(path) text = git_command(server_repo_location, None, "log", "--format='%H\n'", shlex.quote(file)).decode() sha = text.split("\n")[0] identifier = f"/{username}/{repository}/{sha}" last_commit = db.session.get(Commit, identifier) info = { "name": os.path.basename(file), "serverPath": path, "relativePath": file, "link": os.path.join(f"/{username}/{repository}/tree/{branch}/", file), "size": human_size(os.path.getsize(path)), "mimetype": f"{mimetype}{f' ({mimetypes.guess_type(path)[1]})' if mimetypes.guess_type(path)[1] else ''}", "commit": last_commit, "shaSize": 7, } special_icon = config.match_icon(os.path.basename(file)) if special_icon: info["icon"] = special_icon elif os.path.isdir(path): info["icon"] = config.folder_icon elif mimetypes.guess_type(path)[0] in config.file_icons: info["icon"] = config.file_icons[mimetypes.guess_type(path)[0]] else: info["icon"] = config.unknown_icon if os.path.isdir(path): infos.insert(0, info) else: infos.append(info) return flask.render_template( "repo-tree.html", username=username, repository=repository, files=infos, subpath=os.path.join("/", subpath), branches=all_refs, current=branch, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) else: path = os.path.join(server_repo_location, subpath) if not os.path.exists(path): return flask.render_template("not-found.html"), 404 mimetype = guess_mime(path) mode = mimetype.split("/", 1)[0] size = human_size(os.path.getsize(path)) special_icon = config.match_icon(os.path.basename(path)) if special_icon: icon = special_icon elif os.path.isdir(path): icon = config.folder_icon elif mimetypes.guess_type(path)[0] in config.file_icons: icon = config.file_icons[mimetypes.guess_type(path)[0]] else: icon = config.unknown_icon contents = None if mode == "text": contents = convert_to_html(path) return flask.render_template( "repo-file.html", username=username, repository=repository, file=os.path.join(f"/{username}/{repository}/raw/{branch}/", subpath), branches=all_refs, current=branch, mode=mode, mimetype=mimetype, detailedtype=magic.from_file(path), size=size, icon=icon, subpath=os.path.join("/", subpath), basename=os.path.basename(path), contents=contents, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) @repositories.route("///commit/") def repository_commit(username, repository, sha): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() files = git_command(os.path.join(server_repo_location, ".git"), None, "diff-tree", "-r", "--name-only", "--no-commit-id", sha).decode().split("\n") return flask.render_template( "repo-commit.html", username=username, repository=repository, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), diff={file: git_command(os.path.join(server_repo_location, ".git"), None, "diff", str(sha), str(sha) + "^", file).decode().split("\n") for file in files} ) @repositories.route("///forum/") def repository_forum(username, repository): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repo_data) user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first() return flask.render_template( "repo-forum.html", username=username, repository=repository, repo_data=repo_data, relationships=relationships, repo=repo, user_relationship=user_relationship, Post=Post, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), default_branch=repo_data.default_branch ) @repositories.route("///forum/topic/") def repository_forum_topic(username, repository, id): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repo_data) user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first() post = Post.query.filter_by(id=id).first() return flask.render_template( "repo-topic.html", username=username, repository=repository, repo_data=repo_data, relationships=relationships, user_relationship=user_relationship, post=post, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), default_branch=repo_data.default_branch ) @repositories.route("///forum/new", methods=["POST", "GET"]) def repository_forum_new(username, repository): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) serverRepoLocation = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {serverRepoLocation}") if not os.path.exists(serverRepoLocation): app.logger.error(f"Cannot load {serverRepoLocation}") return flask.render_template("not-found.html"), 404 repo = git.Repo(serverRepoLocation) repoData = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repoData) userRelationship = RepoAccess.query.filter_by(repo=repoData, user=user).first() post = Post(user, repoData, None, flask.request.form["subject"], flask.request.form["message"]) db.session.add(post) db.session.commit() return flask.redirect( flask.url_for(".repository_forum_thread", username=username, repository=repository, post_id=post.number), code=303) @repositories.route("///forum/") def repository_forum_thread(username, repository, post_id): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repo_data) user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first() return flask.render_template( "repo-forum-thread.html", username=username, repository=repository, repo_data=repo_data, relationships=relationships, repo=repo, Post=Post, user_relationship=user_relationship, post_id=post_id, max_post_nesting=4, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), parent=Post.query.filter_by(repo=repo_data, number=post_id).first(), ) @repositories.route("///forum//reply", methods=["POST"]) def repository_forum_reply(username, repository, post_id): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repo_data) user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first() if not user: flask.abort(401) parent = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first() post = Post(user, repo_data, parent, flask.request.form["subject"], flask.request.form["message"]) db.session.add(post) post.update_date() db.session.commit() return flask.redirect( flask.url_for(".repository_forum_thread", username=username, repository=repository, post_id=post_id), code=303) @repositories.route("///forum//voteup", defaults={"score": 1}) @repositories.route("///forum//votedown", defaults={"score": -1}) @repositories.route("///forum//votes", defaults={"score": 0}) def repository_forum_vote(username, repository, post_id, score): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repo_data) user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first() if not user: flask.abort(401) post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first() if score: old_relationship = PostVote.query.filter_by(user_username=user.username, post_identifier=post.identifier).first() if old_relationship: if score == old_relationship.vote_score: db.session.delete(old_relationship) post.vote_sum -= old_relationship.vote_score else: post.vote_sum -= old_relationship.vote_score post.vote_sum += score old_relationship.vote_score = score else: relationship = PostVote(user, post, score) post.vote_sum += score db.session.add(relationship) db.session.commit() user_vote = PostVote.query.filter_by(user_username=user.username, post_identifier=post.identifier).first() response = flask.make_response( str(post.vote_sum) + " " + str(user_vote.vote_score if user_vote else 0)) response.content_type = "text/plain" return response @repositories.route("///favourite") def repository_favourite(username, repository): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repo_data) user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first() if not user: flask.abort(401) old_relationship = RepoFavourite.query.filter_by(user_username=user.username, repo_route=repo_data.route).first() if old_relationship: db.session.delete(old_relationship) else: relationship = RepoFavourite(user, repo_data) db.session.add(relationship) db.session.commit() return flask.redirect(flask.url_for("favourites"), code=303) @repositories.route("///users/", methods=["GET", "POST"]) def repository_users(username, repository): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() relationships = RepoAccess.query.filter_by(repo=repo_data) user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first() if flask.request.method == "GET": return flask.render_template( "repo-users.html", username=username, repository=repository, repo_data=repo_data, relationships=relationships, repo=repo, user_relationship=user_relationship, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) else: if get_permission_level(flask.session.get("username"), username, repository) != 2: flask.abort(401) if flask.request.form.get("new-username"): # Create new relationship new_user = User.query.filter_by( username=flask.request.form.get("new-username")).first() relationship = RepoAccess(new_user, repo_data, flask.request.form.get("new-level")) db.session.add(relationship) db.session.commit() if flask.request.form.get("update-username"): # Create new relationship updated_user = User.query.filter_by( username=flask.request.form.get("update-username")).first() relationship = RepoAccess.query.filter_by(repo=repo_data, user=updated_user).first() if flask.request.form.get("update-level") == -1: relationship.delete() else: relationship.access_level = flask.request.form.get("update-level") db.session.commit() return flask.redirect( app.url_for(".repository_users", username=username, repository=repository)) @repositories.route("///branches/") def repository_branches(username, repository): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() return flask.render_template( "repo-branches.html", username=username, repository=repository, repo_data=repo_data, repo=repo, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) @repositories.route("///log/", defaults={"branch": None}) @repositories.route("///log//") def repository_log(username, repository, branch): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() if not repo_data.default_branch: if repo.heads: repo_data.default_branch = repo.heads[0].name else: return flask.render_template("empty.html", remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200 if not branch: branch = repo_data.default_branch return flask.redirect(f"./{branch}", code=302) if branch.startswith("tag:"): ref = f"tags/{branch[4:]}" elif branch.startswith("~"): ref = branch[1:] else: ref = f"heads/{branch}" ref = ref.replace("~", "/") # encode slashes for URL support try: repo.git.checkout("-f", ref) except git.exc.GitCommandError: return flask.render_template("not-found.html"), 404 branches = repo.heads all_refs = [] for ref in repo.heads: all_refs.append((ref, "head")) for ref in repo.tags: all_refs.append((ref, "tag")) commit_list = [f"/{username}/{repository}/{sha}" for sha in git_command(server_repo_location, None, "log", "--format='%H'").decode().split("\n")] commits = Commit.query.filter(Commit.identifier.in_(commit_list)) return flask.render_template( "repo-log.html", username=username, repository=repository, branches=all_refs, current=branch, repo_data=repo_data, repo=repo, commits=commits, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) @repositories.route("///prs/", methods=["GET", "POST"]) def repository_prs(username, repository): if not (get_visibility(username, repository) or get_permission_level( flask.session.get("username"), username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) app.logger.info(f"Loading {server_repo_location}") if not os.path.exists(server_repo_location): app.logger.error(f"Cannot load {server_repo_location}") return flask.render_template("not-found.html"), 404 if flask.request.method == "GET": repo = git.Repo(server_repo_location) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() user = User.query.filter_by(username=flask.session.get("username")).first() return flask.render_template( "repo-prs.html", username=username, repository=repository, repo_data=repo_data, repo=repo, PullRequest=PullRequest, remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), default_branch=repo_data.default_branch, branches=repo.branches ) else: repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() head = flask.request.form.get("head") head_route = flask.request.form.get("headroute") base = flask.request.form.get("base") if not head and base and head_route: return flask.redirect(".", 400) head_repo = git.Repo(os.path.join(config.REPOS_PATH, head_route.lstrip("/"))) base_repo = git.Repo(server_repo_location) print(head_repo) if head not in head_repo.branches or base not in base_repo.branches: flask.flash(Markup( "" + _("Bad branch name")), category="error") return flask.redirect(".", 303) head_data = db.session.get(Repo, head_route) if not head_data.visibility: flask.flash(Markup( "" + _( "Head can't be restricted")), category="error") return flask.redirect(".", 303) pull_request = PullRequest(repo_data, head, head_data, base, db.session.get(User, flask.session["username"])) db.session.add(pull_request) db.session.commit() return flask.redirect(".", 303) @repositories.route("///prs/merge", methods=["POST"]) def repository_prs_merge(username, repository): if not get_permission_level(flask.session.get("username"), username, repository): flask.abort(401) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() repo = git.Repo(server_repo_location) id = flask.request.form.get("id") pull_request = db.session.get(PullRequest, id) if pull_request: result = celery_tasks.merge_heads.delay( pull_request.head_route, pull_request.head_branch, pull_request.base_route, pull_request.base_branch, simulate=True ) task_result = worker.AsyncResult(result.id) return flask.redirect(f"/task/{result.id}?pr-id={id}", 303) # db.session.delete(pull_request) # db.session.commit() else: flask.abort(400) @repositories.route("///prs//merge") def repository_prs_merge_stage_two(username, repository, id): if not get_permission_level(flask.session.get("username"), username, repository): flask.abort(401) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() repo = git.Repo(server_repo_location) pull_request = db.session.get(PullRequest, id) if pull_request: result = celery_tasks.merge_heads.delay( pull_request.head_route, pull_request.head_branch, pull_request.base_route, pull_request.base_branch, simulate=False ) task_result = worker.AsyncResult(result.id) return flask.redirect(f"/task/{result.id}?pr-id={id}", 303) # db.session.delete(pull_request) # db.session.commit() else: flask.abort(400) @app.route("/task/") def task_monitor(task_id): task_result = worker.AsyncResult(task_id) print(task_result.status) return flask.render_template("task-monitor.html", result=task_result) @repositories.route("///prs/delete", methods=["POST"]) def repository_prs_delete(username, repository): if not get_permission_level(flask.session.get("username"), username, repository): flask.abort(401) server_repo_location = os.path.join(config.REPOS_PATH, username, repository) repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() repo = git.Repo(server_repo_location) id = flask.request.form.get("id") pull_request = db.session.get(PullRequest, id) if pull_request: db.session.delete(pull_request) db.session.commit() return flask.redirect(".", 303) @repositories.route("///settings/") def repository_settings(username, repository): if get_permission_level(flask.session.get("username"), username, repository) != 2: flask.abort(401) return flask.render_template("repo-settings.html", username=username, repository=repository) @app.errorhandler(404) def e404(error): return flask.render_template("not-found.html"), 404 @app.errorhandler(401) def e401(error): return flask.render_template("unauthorised.html"), 401 @app.errorhandler(403) def e403(error): return flask.render_template("forbidden.html"), 403 @app.errorhandler(418) def e418(error): return flask.render_template("teapot.html"), 418 @app.errorhandler(405) def e405(error): return flask.render_template("method-not-allowed.html"), 405 if __name__ == "__main__": app.run(debug=True, port=8080, host="0.0.0.0") app.register_blueprint(repositories)