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