import os import shutil import random import subprocess import platform import git import mimetypes import magic import flask import cairosvg 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 from celery import Celery, Task import config import celery_integration app = flask.Flask(__name__) app.config.from_mapping( CELERY=dict( broker_url=config.REDIS_URI, result_backend=config.REDIS_URI, task_ignore_result=True, ), ) worker = celery_integration.init_celery_app(app) auth = HTTPBasicAuth() app.config["SQLALCHEMY_DATABASE_URI"] = config.DB_URI app.config["SECRET_KEY"] = config.DB_PASSWORD app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False db = SQLAlchemy(app) bcrypt = Bcrypt(app) migrate = Migrate(app, db) from models import * import celery_tasks def git_command(repo, data, *args): if not os.path.isdir(repo): raise FileNotFoundError("Repo not found") env = os.environ.copy() command = ["git", *args] proc = subprocess.Popen(" ".join(command), cwd=repo, env=env, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE) print(command) if data: proc.stdin.write(data) out, err = proc.communicate() return out def only_chars(string, chars): for i in string: if i not in chars: return False return True def get_permission_level(logged_in, username, repository): user = User.query.filter_by(username=logged_in).first() repo = Repo.query.filter_by(route=f"/{username}/{repository}").first() if user and repo: permission = RepoAccess.query.filter_by(user=user, repo=repo).first() if permission: return permission.access_level return None def get_visibility(username, repository): repo = Repo.query.filter_by(route=f"/{username}/{repository}").first() if repo: return repo.visibility return None def get_favourite(logged_in, username, repository): print(logged_in, username, repository) relationship = RepoFavourite.query.filter_by(user_username=logged_in, repo_route=f"/{username}/{repository}").first() return relationship import git_http import jinja_utils def human_size(value, decimals=2, scale=1024, units=("B", "kiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB", "RiB", "QiB")): for unit in units: if value < scale: break value /= scale if int(value) == value: # do not return decimals if the value is already round return int(value), unit return round(value * 10 ** decimals) / 10 ** decimals, unit def guess_mime(path): if os.path.isdir(path): mimetype = "inode/directory" elif magic.from_file(path, mime=True): mimetype = magic.from_file(path, mime=True) else: mimetype = "application/octet-stream" return mimetype def convert_to_html(path): with open(path, "r") as f: contents = f.read() return contents 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() } @app.route("/") def main(): return flask.render_template("home.html") @app.route("/about/") def about(): return flask.render_template("about.html", platform=platform) @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.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(f"Successfully logged in as {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( f"Sorry, {username} is a system path"), category="error") return flask.render_template("login.html") user_check = User.query.filter_by(username=username).first() if user_check: flask.flash( Markup( f"The username {username} is taken"), 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( f"Successfully created and logged in as {username}"), category="success") 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(f"Successfully created repository {name}"), category="success") return flask.redirect(repo.route, code=303) @app.route("/logout") def logout(): flask.session.clear() flask.flash(Markup(f"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(followerUsername=flask.session.get("username"), followedUsername=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(ownerName=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) flask.flash(f"Sending notification in task {result.id}", "success") 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) repoData = Repo.query.filter_by(route=f"/{username}/{repository}").first() if not repoData.defaultBranch: if repo.heads: repoData.defaultBranch = 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 = repoData.defaultBranch 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.defaultBranch: if repo.heads: repo_data.defaultBranch = repo.heads[0].name else: return flask.render_template("empty.html", remote=f"http{'s' if config.suggestHTTPS else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200 if not branch: branch = repo_data.defaultBranch 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'", file).decode() sha = text.split("\n")[0] identifier = f"/{username}/{repository}/{sha}" last_commit = Commit.query.filter_by(identifier=identifier).first() 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.matchIcon(os.path.basename(file)) if special_icon: info["icon"] = special_icon elif os.path.isdir(path): info["icon"] = config.folderIcon elif mimetypes.guess_type(path)[0] in config.fileIcons: info["icon"] = config.fileIcons[mimetypes.guess_type(path)[0]] else: info["icon"] = config.unknownIcon 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.suggestHTTPS 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.matchIcon(os.path.basename(path)) if special_icon: icon = special_icon elif os.path.isdir(path): icon = config.folderIcon elif mimetypes.guess_type(path)[0] in config.fileIcons: icon = config.fileIcons[mimetypes.guess_type(path)[0]] else: icon = config.unknownIcon 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.suggestHTTPS else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) @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.suggestHTTPS else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), default_branch=repo_data.defaultBranch ) @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.suggestHTTPS else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), default_branch=repo_data.defaultBranch ) @repositories.route("///forum/create-topic", methods=["POST", "GET"]) def repository_forum_create_topic(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_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 == "POST": # Check if the user has permission to create a topic if not user_relationship or not user_relationship.permissionCreateTopic: flask.abort(403) # Get form data title = flask.request.form.get("title") content = flask.request.form.get("content") # Validate form data if not title or not content: flask.flash("Title and content are required", "error") return flask.redirect( flask.url_for("repository_forum_create_topic", username=username, repository=repository)) # Create a new topic new_topic = Post( title=title, content=content, author=user, repo=repo_data ) db.session.add(new_topic) db.session.commit() flask.flash("Topic created successfully!", "success") return flask.redirect( flask.url_for("repository_forum_topic", username=username, repository=repository, id=new_topic.id)) return flask.render_template( "repo-create-topic.html", username=username, repository=repository, repo_data=repo_data, relationships=relationships, user_relationship=user_relationship, remote=f"http{'s' if config.suggestHTTPS else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository), default_branch=repo_data.defaultBranch ) @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, user_relationship=user_relationship, post_id=post_id, max_post_nesting=4, remote=f"http{'s' if config.suggestHTTPS else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) @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.suggestHTTPS 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.suggestHTTPS 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.suggestHTTPS 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.suggestHTTPS else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}", is_favourite=get_favourite(flask.session.get("username"), username, repository) ) @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)