import uuid from models import * from app import app, db, bcrypt from misc_utils import * from common import git_command import os import shutil import config import flask import git import subprocess from flask_httpauth import HTTPBasicAuth import zlib import re import datetime auth = HTTPBasicAuth(realm=config.AUTH_REALM) auth_required = flask.Response("Unauthorized Access", 401, {"WWW-Authenticate": 'Basic realm="Login Required"'}) @auth.verify_password def verify_password(username, password): user = User.query.filter_by(username=username).first() if user and bcrypt.check_password_hash(user.password_hashed, password): flask.g.user = username return True return False def get_commit_identity(identity, pusher, repo): email = identity.rpartition("<")[2].rpartition(">")[0].strip() # If the email is not valid, attribute the commit to the pusher. if not email: return pusher email_users = User.query.filter_by(email=email).all() # If no user has the email, attribute the commit to the pusher. if not email_users: return pusher # If only one user has the email, attribute the commit to them. if len(email_users) == 1: return email_users[0] # If it's ambiguous, attribute the commit to an user with a higher permission level. for user in email_users: if repo.owner == user: return user for user in email_users: relationship = RepoAccess.query.filter_by(user=user, repo=repo).first() if relationship.permission_level == 2: return user for user in email_users: relationship = RepoAccess.query.filter_by(user=user, repo=repo).first() if relationship.permission_level == 1: return user # If no user has a higher permission level, attribute the commit to the pusher :( return pusher @app.route("///git-upload-pack", methods=["POST"]) @app.route("/git///git-upload-pack", methods=["POST"]) @auth.login_required(optional=True) def git_upload_pack(username, repository): if auth.current_user() is None and not get_visibility(username, repository): return auth_required if not (get_visibility(username, repository) or get_permission_level(flask.g.user, username, repository) is not None): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") text = git_command(server_repo_location, flask.request.data, "upload-pack", "--stateless-rpc", ".") return flask.Response(text, content_type="application/x-git-upload-pack-result") @app.route("///git-receive-pack", methods=["POST"]) @app.route("/git///git-receive-pack", methods=["POST"]) @auth.login_required def git_receive_pack(username, repository): if not get_permission_level(flask.g.user, username, repository): flask.abort(403) server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") text = git_command(server_repo_location, flask.request.data, "receive-pack", "--stateless-rpc", ".") if flask.request.data == b"0000": return flask.Response("", content_type="application/x-git-receive-pack-result") push_info = flask.request.data.split(b"\x00")[0].decode() if not push_info: return flask.Response(text, content_type="application/x-git-receive-pack-result") old_sha, new_sha, ref = push_info[4:].split() # discard first 4 characters, used for line length which we don't need if old_sha == "0" * 40: commits_list = subprocess.check_output(["git", "rev-list", new_sha], cwd=server_repo_location).decode().strip().split("\n") else: commits_list = subprocess.check_output(["git", "rev-list", f"{old_sha}..{new_sha}"], cwd=server_repo_location).decode().strip().split("\n") for sha in reversed(commits_list): info = git_command(server_repo_location, None, "show", "-s", "--format='%H%n%at%n%cn <%ce>%n%B'", sha).decode() sha, time, identity, body = info.split("\n", 3) login = flask.g.user if not Commit.query.filter_by(identifier=f"/{username}/{repository}/{sha}").first(): logged_in_user = User.query.filter_by(username=login).first() user = get_commit_identity(identity, logged_in_user, db.session.get(Repo, f"/{username}/{repository}")) repo = Repo.query.filter_by(route=f"/{username}/{repository}").first() commit = Commit(sha, user, repo, time, body, identity) db.session.add(commit) db.session.commit() if ref.startswith("refs/heads/"): # if the push is to a branch ref = ref.rpartition("/")[2] # get the branch name only repo_data = db.session.get(Repo, f"/{username}/{repository}") if ref == repo_data.site_branch: # Update the site from celery_tasks import copy_site copy_site.delay(repo_data.route) return flask.Response(text, content_type="application/x-git-receive-pack-result") @app.route("///info/refs", methods=["GET", "POST"]) @app.route("/git///info/refs", methods=["GET", "POST"]) @auth.login_required(optional=True) def git_info_refs(username, repository): server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") 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 repo.git.checkout("-f", repo_data.default_branch) if auth.current_user() is None and ( not get_visibility(username, repository) or flask.request.args.get( "service") == "git-receive-pack"): return auth_required try: if not (get_visibility(username, repository) or get_permission_level(flask.g.user, username, repository) is not None): flask.abort(403) except AttributeError: return auth_required service = flask.request.args.get("service") if service.startswith("git"): service = service[4:] else: flask.abort(403) if service == "receive-pack": try: if not get_permission_level(flask.g.user, username, repository): flask.abort(403) except AttributeError: return auth_required service_line = f"# service=git-{service}\n" service_line = (f"{len(service_line) + 4:04x}" + service_line).encode() if service == "upload-pack": text = service_line + b"0000" + git_command(server_repo_location, None, "upload-pack", "--stateless-rpc", "--advertise-refs", "--http-backend-info-refs", ".") elif service == "receive-pack": refs = git_command(server_repo_location, None, "receive-pack", "--http-backend-info-refs", ".") text = service_line + b"0000" + refs else: flask.abort(403) response = flask.Response(text, content_type=f"application/x-git-{service}-advertisement") response.headers["Cache-Control"] = "no-cache" return response