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, {"WWW-Authenticate": 'Basic realm="Login Required"'}) 19 20 21@auth.verify_password 22def verify_password(username, password): 23user = User.query.filter_by(username=username).first() 24 25if user and bcrypt.check_password_hash(user.password_hashed, password): 26flask.g.user = username 27return True 28 29return False 30 31 32@app.route("/<username>/<repository>/git-upload-pack", methods=["POST"]) 33@app.route("/git/<username>/<repository>/git-upload-pack", methods=["POST"]) 34@auth.login_required(optional=True) 35def git_upload_pack(username, repository): 36if auth.current_user() is None and not get_visibility(username, repository): 37return auth_required 38if not (get_visibility(username, repository) or get_permission_level(flask.g.user, username, 39repository) is not None): 40flask.abort(403) 41 42server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") 43text = git_command(server_repo_location, flask.request.data, "upload-pack", "--stateless-rpc", ".") 44 45return flask.Response(text, content_type="application/x-git-upload-pack-result") 46 47 48@app.route("/<username>/<repository>/git-receive-pack", methods=["POST"]) 49@app.route("/git/<username>/<repository>/git-receive-pack", methods=["POST"]) 50@auth.login_required 51def git_receive_pack(username, repository): 52if not get_permission_level(flask.g.user, username, repository): 53flask.abort(403) 54 55server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") 56text = git_command(server_repo_location, flask.request.data, "receive-pack", "--stateless-rpc", ".") 57 58sha = flask.request.data.split(b" ", 2)[1].decode() 59info = git_command(server_repo_location, None, "show", "-s", "--format='%H%n%at%n%cn <%ce>%n%B'", sha).decode() 60 61if re.match("^[0-9a-fA-F]{40}$", info[:40]): 62print(info.split("\n", 3)) 63sha, time, identity, body = info.split("\n", 3) 64login = flask.g.user 65 66if not Commit.query.filter_by(identifier=f"/{username}/{repository}/{sha}").first(): 67user = User.query.filter_by(username=login).first() 68repo = Repo.query.filter_by(route=f"/{username}/{repository}").first() 69 70commit = Commit(sha, user, repo, time, body, identity) 71 72db.session.add(commit) 73db.session.commit() 74 75return flask.Response(text, content_type="application/x-git-receive-pack-result") 76 77 78@app.route("/<username>/<repository>/info/refs", methods=["GET", "POST"]) 79@app.route("/git/<username>/<repository>/info/refs", methods=["GET", "POST"]) 80@auth.login_required(optional=True) 81def git_info_refs(username, repository): 82server_repo_location = os.path.join(config.REPOS_PATH, username, repository, ".git") 83 84repo = git.Repo(server_repo_location) 85repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first() 86if not repo_data.default_branch: 87if repo.heads: 88repo_data.default_branch = repo.heads[0].name 89repo.git.checkout("-f", repo_data.default_branch) 90 91if auth.current_user() is None and ( 92not get_visibility(username, repository) or flask.request.args.get("service") == "git-receive-pack"): 93return auth_required 94try: 95if not (get_visibility(username, repository) or get_permission_level(flask.g.user, username, 96repository) is not None): 97flask.abort(403) 98except AttributeError: 99return auth_required 100 101service = flask.request.args.get("service") 102 103if service.startswith("git"): 104service = service[4:] 105else: 106flask.abort(403) 107 108if service == "receive-pack": 109try: 110if not get_permission_level(flask.g.user, username, repository): 111flask.abort(403) 112except AttributeError: 113return auth_required 114 115service_line = f"# service=git-{service}\n" 116service_line = (f"{len(service_line) + 4:04x}" + service_line).encode() 117 118if service == "upload-pack": 119text = service_line + b"0000" + git_command(server_repo_location, None, "upload-pack", "--stateless-rpc", 120"--advertise-refs", "--http-backend-info-refs", ".") 121elif service == "receive-pack": 122refs = git_command(server_repo_location, None, "receive-pack", "--http-backend-info-refs", ".") 123text = service_line + b"0000" + refs 124else: 125flask.abort(403) 126 127response = flask.Response(text, content_type=f"application/x-git-{service}-advertisement") 128response.headers["Cache-Control"] = "no-cache" 129 130return response 131