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