By using this site, you agree to have cookies stored on your device, strictly for functional purposes, such as storing your session and preferences.

Dismiss

 git_http.py

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