git_http.py
Python script, ASCII text executable
1import uuid 2 3from models import * 4from app import app, git_command, get_permission_level, get_visibility, db, bcrypt 5import os 6import shutil 7import config 8import flask 9import git 10import subprocess 11from flask_httpauth import HTTPBasicAuth 12import zlib 13import re 14import datetime 15 16auth = HTTPBasicAuth(realm=config.AUTH_REALM) 17 18auth_required = flask.Response("Unauthorized Access", 401, 19{"WWW-Authenticate": 'Basic realm="Login Required"'}) 20 21 22@auth.verify_password 23def verify_password(username, password): 24user = User.query.filter_by(username=username).first() 25 26if user and bcrypt.check_password_hash(user.password_hashed, password): 27flask.g.user = username 28return True 29 30return False 31 32 33@app.route("/<username>/<repository>/git-upload-pack", methods=["POST"]) 34@app.route("/git/<username>/<repository>/git-upload-pack", methods=["POST"]) 35@auth.login_required(optional=True) 36def git_upload_pack(username, repository): 37if auth.current_user() is None and not get_visibility(username, repository): 38return auth_required 39if not (get_visibility(username, repository) or get_permission_level(flask.g.user, username, 40repository) is not None): 41flask.abort(403) 42 43server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") 44text = git_command(server_repo_location, flask.request.data, "upload-pack", 45"--stateless-rpc", ".") 46 47return flask.Response(text, content_type="application/x-git-upload-pack-result") 48 49 50@app.route("/<username>/<repository>/git-receive-pack", methods=["POST"]) 51@app.route("/git/<username>/<repository>/git-receive-pack", methods=["POST"]) 52@auth.login_required 53def git_receive_pack(username, repository): 54if not get_permission_level(flask.g.user, username, repository): 55flask.abort(403) 56 57server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") 58text = git_command(server_repo_location, flask.request.data, "receive-pack", 59"--stateless-rpc", ".") 60 61if flask.request.data == b"0000": 62return flask.Response("", content_type="application/x-git-receive-pack-result") 63 64push_info = flask.request.data.split(b"\x00")[1].split(b" ")[0].decode() 65old_sha, new_sha, _ = push_info.split() 66 67commits_list = subprocess.check_output(["git", "rev-list", f"{old_sha}..{new_sha}"], cwd=server_repo_location).decode().strip().split("\n") 68 69sha = flask.request.data.split(b" ", 2)[1].decode() 70 71for sha in commits_list: 72info = git_command(server_repo_location, None, "show", "-s", 73"--format='%H%n%at%n%cn <%ce>%n%B'", sha).decode() 74 75print(info.split("\n", 3)) 76sha, time, identity, body = info.split("\n", 3) 77login = flask.g.user 78 79if not Commit.query.filter_by(identifier=f"/{username}/{repository}/{sha}").first(): 80user = User.query.filter_by(username=login).first() 81repo = Repo.query.filter_by(route=f"/{username}/{repository}").first() 82 83commit = Commit(sha, user, repo, time, body, identity) 84 85db.session.add(commit) 86db.session.commit() 87 88return flask.Response(text, content_type="application/x-git-receive-pack-result") 89 90 91@app.route("/<username>/<repository>/info/refs", methods=["GET", "POST"]) 92@app.route("/git/<username>/<repository>/info/refs", methods=["GET", "POST"]) 93@auth.login_required(optional=True) 94def git_info_refs(username, repository): 95server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") 96 97repo = git.Repo(server_repo_location) 98repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() 99if not repo_data.default_branch: 100if repo.heads: 101repo_data.default_branch = repo.heads[0].name 102repo.git.checkout("-f", repo_data.default_branch) 103 104if auth.current_user() is None and ( 105not get_visibility(username, repository) or flask.request.args.get( 106"service") == "git-receive-pack"): 107return auth_required 108try: 109if not (get_visibility(username, repository) or get_permission_level(flask.g.user, 110username, 111repository) is not None): 112flask.abort(403) 113except AttributeError: 114return auth_required 115 116service = flask.request.args.get("service") 117 118if service.startswith("git"): 119service = service[4:] 120else: 121flask.abort(403) 122 123if service == "receive-pack": 124try: 125if not get_permission_level(flask.g.user, username, repository): 126flask.abort(403) 127except AttributeError: 128return auth_required 129 130service_line = f"# service=git-{service}\n" 131service_line = (f"{len(service_line) + 4:04x}" + service_line).encode() 132 133if service == "upload-pack": 134text = service_line + b"0000" + git_command(server_repo_location, None, "upload-pack", 135"--stateless-rpc", 136"--advertise-refs", 137"--http-backend-info-refs", ".") 138elif service == "receive-pack": 139refs = git_command(server_repo_location, None, "receive-pack", 140"--http-backend-info-refs", ".") 141text = service_line + b"0000" + refs 142else: 143flask.abort(403) 144 145response = flask.Response(text, content_type=f"application/x-git-{service}-advertisement") 146response.headers["Cache-Control"] = "no-cache" 147 148return response