Python script, Unicode text, UTF-8 text executable
1
__version__ = "0.1.2"
2
import os
4
import shutil
5
import random
6
import subprocess
7
import platform
8
import git
9
import mimetypes
10
import magic
11
import flask
12
import cairosvg
13
import celery
14
import shlex
15
from functools import wraps
16
from datetime import datetime
17
from enum import Enum
18
from cairosvg import svg2png
19
from flask_sqlalchemy import SQLAlchemy
20
from flask_bcrypt import Bcrypt
21
from markupsafe import escape, Markup
22
from flask_migrate import Migrate
23
from PIL import Image
24
from flask_httpauth import HTTPBasicAuth
25
import config
26
from flask_babel import Babel, gettext, ngettext, force_locale
27
_ = gettext
29
n_ = gettext
30
app = flask.Flask(__name__)
32
app.config.from_mapping(
33
CELERY=dict(
34
broker_url=config.REDIS_URI,
35
result_backend=config.REDIS_URI,
36
task_ignore_result=True,
37
),
38
)
39
auth = HTTPBasicAuth()
41
app.config["SQLALCHEMY_DATABASE_URI"] = config.DB_URI
43
app.config["SECRET_KEY"] = config.DB_PASSWORD
44
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
45
app.config["BABEL_TRANSLATION_DIRECTORIES"] = "i18n"
46
app.config["MAX_CONTENT_LENGTH"] = config.MAX_PAYLOAD_SIZE
47
db = SQLAlchemy(app)
49
bcrypt = Bcrypt(app)
50
migrate = Migrate(app, db)
51
from models import *
53
from misc_utils import *
54
import git_http
56
import jinja_utils
57
import celery_tasks
58
from celery import Celery, Task
59
import celery_integration
60
import pathlib
61
babel = Babel(app)
63
def get_locale():
66
if flask.request.cookies.get("language"):
67
return flask.request.cookies.get("language")
68
return flask.request.accept_languages.best_match(config.available_locales)
69
babel.init_app(app, locale_selector=get_locale)
72
with app.app_context():
74
locale_names = {}
75
for language in config.available_locales:
76
with force_locale(language):
77
# NOTE: Translate this to the language's name in that language, for example in French you would use français
78
locale_names[language] = gettext("English")
79
worker = celery_integration.init_celery_app(app)
81
repositories = flask.Blueprint("repository", __name__, template_folder="templates/repository/")
83
app.jinja_env.add_extension("jinja2.ext.do")
85
app.jinja_env.add_extension("jinja2.ext.loopcontrols")
86
app.jinja_env.add_extension("jinja2.ext.debug")
87
@app.context_processor
90
def default():
91
username = flask.session.get("username")
92
user_object = User.query.filter_by(username=username).first()
94
return {
96
"logged_in_user": username,
97
"user_object": user_object,
98
"Notification": Notification,
99
"unread": UserNotification.query.filter_by(user_username=username).filter(
100
UserNotification.attention_level > 0).count(),
101
"config": config,
102
"Markup": Markup,
103
"locale_names": locale_names,
104
}
105
@app.route("/")
108
def main():
109
if flask.session.get("username"):
110
return flask.render_template("home.html")
111
else:
112
return flask.render_template("no-home.html")
113
@app.route("/userstyle")
116
def userstyle():
117
if flask.session.get("username") and os.path.exists(os.path.join(config.REPOS_PATH, flask.session.get("username"), ".config", "theme.css")):
118
return flask.send_from_directory(os.path.join(config.REPOS_PATH, flask.session.get("username"), ".config"), "theme.css")
119
else:
120
return flask.Response("", mimetype="text/css")
121
@app.route("/about/")
124
def about():
125
return flask.render_template("about.html", platform=platform, version=__version__)
126
@app.route("/language", methods=["POST"])
129
def set_locale():
130
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/",
131
code=303)
132
if not flask.request.form.get("language"):
133
response.delete_cookie("language")
134
else:
135
response.set_cookie("language", flask.request.form.get("language"))
136
return response
138
@app.route("/cookie-dismiss")
141
def dismiss_banner():
142
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/",
143
code=303)
144
response.set_cookie("cookie-banner", "1")
145
return response
146
@app.route("/help/")
149
def help_index():
150
return flask.render_template("help.html", faqs=config.faqs)
151
@app.route("/settings/", methods=["GET", "POST"])
154
def settings():
155
if not flask.session.get("username"):
156
flask.abort(401)
157
if flask.request.method == "GET":
158
user = User.query.filter_by(username=flask.session.get("username")).first()
159
return flask.render_template("user-settings.html", user=user)
161
else:
162
user = User.query.filter_by(username=flask.session.get("username")).first()
163
user.display_name = flask.request.form["displayname"]
165
user.URL = flask.request.form["url"]
166
user.company = flask.request.form["company"]
167
user.company_URL = flask.request.form["companyurl"]
168
user.email = flask.request.form.get("email") if flask.request.form.get(
169
"email") else None
170
user.location = flask.request.form["location"]
171
user.show_mail = True if flask.request.form.get("showmail") else False
172
user.bio = flask.request.form.get("bio")
173
db.session.commit()
175
flask.flash(
177
Markup("<iconify-icon icon='mdi:check'></iconify-icon>" + _("Settings saved")),
178
category="success")
179
return flask.redirect(f"/{flask.session.get('username')}", code=303)
180
@app.route("/favourites/", methods=["GET", "POST"])
183
def favourites():
184
if not flask.session.get("username"):
185
flask.abort(401)
186
if flask.request.method == "GET":
187
relationships = RepoFavourite.query.filter_by(
188
user_username=flask.session.get("username"))
189
return flask.render_template("favourites.html", favourites=relationships)
191
@app.route("/notifications/", methods=["GET", "POST"])
194
def notifications():
195
if not flask.session.get("username"):
196
flask.abort(401)
197
if flask.request.method == "GET":
198
return flask.render_template("notifications.html",
199
notifications=UserNotification.query.filter_by(
200
user_username=flask.session.get("username")))
201
@app.route("/accounts/", methods=["GET", "POST"])
204
def login():
205
if flask.request.method == "GET":
206
return flask.render_template("login.html")
207
else:
208
if "login" in flask.request.form:
209
username = flask.request.form["username"]
210
password = flask.request.form["password"]
211
user = User.query.filter_by(username=username).first()
213
if user and bcrypt.check_password_hash(user.password_hashed, password):
215
flask.session["username"] = user.username
216
flask.flash(
217
Markup("<iconify-icon icon='mdi:account'></iconify-icon>" + _(
218
"Successfully logged in as {username}").format(username=username)),
219
category="success")
220
return flask.redirect("/", code=303)
221
elif not user:
222
flask.flash(Markup(
223
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
224
"User not found")),
225
category="alert")
226
return flask.render_template("login.html")
227
else:
228
flask.flash(Markup(
229
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
230
"Invalid password")),
231
category="error")
232
return flask.render_template("login.html")
233
if "signup" in flask.request.form:
234
username = flask.request.form["username"]
235
password = flask.request.form["password"]
236
password2 = flask.request.form["password2"]
237
email = flask.request.form.get("email")
238
email2 = flask.request.form.get("email2") # repeat email is a honeypot
239
name = flask.request.form.get("name")
240
if not only_chars(username,
242
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
243
flask.flash(Markup(
244
_("Usernames may only contain Latin alphabet, numbers, '-' and '_'")),
245
category="error")
246
return flask.render_template("login.html")
247
if username in config.RESERVED_NAMES:
249
flask.flash(
250
Markup(
251
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _(
252
"Sorry, {username} is a system path").format(
253
username=username)),
254
category="error")
255
return flask.render_template("login.html")
256
user_check = User.query.filter_by(username=username).first()
258
if user_check:
259
flask.flash(
260
Markup(
261
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _(
262
"The username {username} is taken").format(
263
username=username)),
264
category="error")
265
return flask.render_template("login.html")
266
if password2 != password:
268
flask.flash(Markup("<iconify-icon icon='mdi:key-alert'></iconify-icon>" + _(
269
"Make sure the passwords match")),
270
category="error")
271
return flask.render_template("login.html")
272
user = User(username, password, email, name)
274
db.session.add(user)
275
db.session.commit()
276
flask.session["username"] = user.username
277
flask.flash(Markup(
278
"<iconify-icon icon='mdi:account'></iconify-icon>" + _(
279
"Successfully created and logged in as {username}").format(
280
username=username)),
281
category="success")
282
notification = Notification({"type": "welcome"})
284
db.session.add(notification)
285
db.session.commit()
286
result = celery_tasks.send_notification.delay(notification.id, [username], 1)
288
return flask.redirect("/", code=303)
290
@app.route("/newrepo/", methods=["GET", "POST"])
293
def new_repo():
294
if not flask.session.get("username"):
295
flask.abort(401)
296
if flask.request.method == "GET":
297
return flask.render_template("new-repo.html")
298
else:
299
name = flask.request.form["name"]
300
visibility = int(flask.request.form["visibility"])
301
if not only_chars(name,
303
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
304
flask.flash(Markup(
305
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
306
"Repository names may only contain Latin alphabet, numbers, '-' and '_'")),
307
category="error")
308
return flask.render_template("new-repo.html")
309
user = User.query.filter_by(username=flask.session.get("username")).first()
311
repo = Repo(user, name, visibility)
313
db.session.add(repo)
314
db.session.commit()
315
flask.flash(Markup(_("Successfully created repository {name}").format(name=name)),
317
category="success")
318
return flask.redirect(repo.route, code=303)
319
@app.route("/logout")
322
def logout():
323
flask.session.clear()
324
flask.flash(Markup(
325
"<iconify-icon icon='mdi:account'></iconify-icon>" + _("Successfully logged out")),
326
category="info")
327
return flask.redirect("/", code=303)
328
@app.route("/<username>/", methods=["GET", "POST"])
331
def user_profile(username):
332
old_relationship = UserFollow.query.filter_by(
333
follower_username=flask.session.get("username"),
334
followed_username=username).first()
335
if flask.request.method == "GET":
336
user = User.query.filter_by(username=username).first()
337
match flask.request.args.get("action"):
338
case "repositories":
339
repos = Repo.query.filter_by(owner_name=username, visibility=2)
340
return flask.render_template("user-profile-repositories.html", user=user,
341
repos=repos,
342
relationship=old_relationship)
343
case "followers":
344
return flask.render_template("user-profile-followers.html", user=user,
345
relationship=old_relationship)
346
case "follows":
347
return flask.render_template("user-profile-follows.html", user=user,
348
relationship=old_relationship)
349
case _:
350
return flask.render_template("user-profile-overview.html", user=user,
351
relationship=old_relationship)
352
elif flask.request.method == "POST":
354
match flask.request.args.get("action"):
355
case "follow":
356
if username == flask.session.get("username"):
357
flask.abort(403)
358
if old_relationship:
359
db.session.delete(old_relationship)
360
else:
361
relationship = UserFollow(
362
flask.session.get("username"),
363
username
364
)
365
db.session.add(relationship)
366
db.session.commit()
367
user = db.session.get(User, username)
369
author = db.session.get(User, flask.session.get("username"))
370
notification = Notification({"type": "update", "version": "0.0.0"})
371
db.session.add(notification)
372
db.session.commit()
373
result = celery_tasks.send_notification.delay(notification.id, [username],
375
1)
376
db.session.commit()
378
return flask.redirect("?", code=303)
379
@app.route("/<username>/<repository>/")
382
def repository_index(username, repository):
383
return flask.redirect("./tree", code=302)
384
@app.route("/info/<username>/avatar")
387
def user_avatar(username):
388
serverUserdataLocation = os.path.join(config.USERDATA_PATH, username)
389
if not os.path.exists(serverUserdataLocation):
391
return flask.render_template("not-found.html"), 404
392
return flask.send_from_directory(serverUserdataLocation, "avatar.png")
394
@app.route("/<username>/<repository>/raw/<branch>/<path:subpath>")
397
def repository_raw(username, repository, branch, subpath):
398
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
399
if not os.path.exists(server_repo_location):
400
app.logger.error(f"Cannot load {server_repo_location}")
401
flask.abort(404)
402
if not (get_visibility(username, repository) or get_permission_level(
403
flask.session.get("username"), username,
404
repository) is not None):
405
flask.abort(403)
406
app.logger.info(f"Loading {server_repo_location}")
408
if not os.path.exists(server_repo_location):
410
app.logger.error(f"Cannot load {server_repo_location}")
411
return flask.render_template("not-found.html"), 404
412
repo = git.Repo(server_repo_location)
414
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
415
if not repo_data.default_branch:
416
if repo.heads:
417
repo_data.default_branch = repo.heads[0].name
418
else:
419
return flask.render_template("empty.html",
420
remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
421
if not branch:
422
branch = repo_data.default_branch
423
return flask.redirect(f"./{branch}", code=302)
424
if branch.startswith("tag:"):
426
ref = f"tags/{branch[4:]}"
427
elif branch.startswith("~"):
428
ref = branch[1:]
429
else:
430
ref = f"heads/{branch}"
431
ref = ref.replace("~", "/") # encode slashes for URL support
433
try:
435
repo.git.checkout("-f", ref)
436
except git.exc.GitCommandError:
437
return flask.render_template("not-found.html"), 404
438
return flask.send_from_directory(config.REPOS_PATH,
440
os.path.join(username, repository, subpath))
441
@repositories.route("/<username>/<repository>/tree/", defaults={"branch": None, "subpath": ""})
444
@repositories.route("/<username>/<repository>/tree/<branch>/", defaults={"subpath": ""})
445
@repositories.route("/<username>/<repository>/tree/<branch>/<path:subpath>")
446
def repository_tree(username, repository, branch, subpath):
447
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
448
if not os.path.exists(server_repo_location):
449
app.logger.error(f"Cannot load {server_repo_location}")
450
flask.abort(404)
451
if not (get_visibility(username, repository) or get_permission_level(
452
flask.session.get("username"), username,
453
repository) is not None):
454
flask.abort(403)
455
app.logger.info(f"Loading {server_repo_location}")
457
repo = git.Repo(server_repo_location)
459
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
460
if not repo_data.default_branch:
461
if repo.heads:
462
repo_data.default_branch = repo.heads[0].name
463
else:
464
return flask.render_template("empty.html",
465
remote=f"{config.www_protocol}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
466
if not branch:
467
branch = repo_data.default_branch
468
return flask.redirect(f"./{branch}", code=302)
469
if branch.startswith("tag:"):
471
ref = f"tags/{branch[4:]}"
472
elif branch.startswith("~"):
473
ref = branch[1:]
474
else:
475
ref = f"heads/{branch}"
476
ref = ref.replace("~", "/") # encode slashes for URL support
478
try:
480
repo.git.checkout("-f", ref)
481
except git.exc.GitCommandError:
482
return flask.render_template("not-found.html"), 404
483
branches = repo.heads
485
all_refs = []
487
for ref in repo.heads:
488
all_refs.append((ref, "head"))
489
for ref in repo.tags:
490
all_refs.append((ref, "tag"))
491
if os.path.isdir(os.path.join(server_repo_location, subpath)):
493
files = []
494
blobs = []
495
for entry in os.listdir(os.path.join(server_repo_location, subpath)):
497
if not os.path.basename(entry) == ".git":
498
files.append(os.path.join(subpath, entry))
499
infos = []
501
for file in files:
503
path = os.path.join(server_repo_location, file)
504
mimetype = guess_mime(path)
505
text = git_command(server_repo_location, None, "log", "--format='%H\n'",
507
shlex.quote(file)).decode()
508
sha = text.split("\n")[0]
510
identifier = f"/{username}/{repository}/{sha}"
511
last_commit = db.session.get(Commit, identifier)
513
info = {
515
"name": os.path.basename(file),
516
"serverPath": path,
517
"relativePath": file,
518
"link": os.path.join(f"/{username}/{repository}/tree/{branch}/", file),
519
"size": human_size(os.path.getsize(path)),
520
"mimetype": f"{mimetype}{f' ({mimetypes.guess_type(path)[1]})' if mimetypes.guess_type(path)[1] else ''}",
521
"commit": last_commit,
522
"shaSize": 7,
523
}
524
special_icon = config.match_icon(os.path.basename(file))
526
if special_icon:
527
info["icon"] = special_icon
528
elif os.path.isdir(path):
529
info["icon"] = config.folder_icon
530
elif mimetypes.guess_type(path)[0] in config.file_icons:
531
info["icon"] = config.file_icons[mimetypes.guess_type(path)[0]]
532
else:
533
info["icon"] = config.unknown_icon
534
if os.path.isdir(path):
536
infos.insert(0, info)
537
else:
538
infos.append(info)
539
return flask.render_template(
541
"repo-tree.html",
542
username=username,
543
repository=repository,
544
files=infos,
545
subpath=os.path.join("/", subpath),
546
branches=all_refs,
547
current=branch,
548
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
549
is_favourite=get_favourite(flask.session.get("username"), username, repository)
550
)
551
else:
552
path = os.path.join(server_repo_location, subpath)
553
if not os.path.exists(path):
555
return flask.render_template("not-found.html"), 404
556
mimetype = guess_mime(path)
558
mode = mimetype.split("/", 1)[0]
559
size = human_size(os.path.getsize(path))
560
special_icon = config.match_icon(os.path.basename(path))
562
if special_icon:
563
icon = special_icon
564
elif os.path.isdir(path):
565
icon = config.folder_icon
566
elif mimetypes.guess_type(path)[0] in config.file_icons:
567
icon = config.file_icons[mimetypes.guess_type(path)[0]]
568
else:
569
icon = config.unknown_icon
570
contents = None
572
if mode == "text":
573
contents = convert_to_html(path)
574
return flask.render_template(
576
"repo-file.html",
577
username=username,
578
repository=repository,
579
file=os.path.join(f"/{username}/{repository}/raw/{branch}/", subpath),
580
branches=all_refs,
581
current=branch,
582
mode=mode,
583
mimetype=mimetype,
584
detailedtype=magic.from_file(path),
585
size=size,
586
icon=icon,
587
subpath=os.path.join("/", subpath),
588
extension=pathlib.Path(path).suffix,
589
basename=os.path.basename(path),
590
contents=contents,
591
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
592
is_favourite=get_favourite(flask.session.get("username"), username, repository)
593
)
594
@repositories.route("/<username>/<repository>/commit/<sha>")
597
def repository_commit(username, repository, sha):
598
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
599
if not os.path.exists(server_repo_location):
600
app.logger.error(f"Cannot load {server_repo_location}")
601
flask.abort(404)
602
if not (get_visibility(username, repository) or get_permission_level(
603
flask.session.get("username"), username,
604
repository) is not None):
605
flask.abort(403)
606
app.logger.info(f"Loading {server_repo_location}")
608
if not os.path.exists(server_repo_location):
610
app.logger.error(f"Cannot load {server_repo_location}")
611
return flask.render_template("not-found.html"), 404
612
repo = git.Repo(server_repo_location)
614
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
615
files = git_command(os.path.join(server_repo_location, ".git"), None, "diff-tree", "-r",
617
"--name-only", "--no-commit-id", sha).decode().split("\n")[:-1]
618
print(files)
620
return flask.render_template(
622
"repo-commit.html",
623
username=username,
624
repository=repository,
625
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
626
is_favourite=get_favourite(flask.session.get("username"), username, repository),
627
diff={file: git_command(os.path.join(server_repo_location, ".git"), None, "diff",
628
str(sha) + "^!", "--", file).decode().split("\n") for
629
file in files}
630
)
631
@repositories.route("/<username>/<repository>/forum/")
634
def repository_forum(username, repository):
635
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
636
if not os.path.exists(server_repo_location):
637
app.logger.error(f"Cannot load {server_repo_location}")
638
flask.abort(404)
639
if not (get_visibility(username, repository) or get_permission_level(
640
flask.session.get("username"), username,
641
repository) is not None):
642
flask.abort(403)
643
app.logger.info(f"Loading {server_repo_location}")
645
if not os.path.exists(server_repo_location):
647
app.logger.error(f"Cannot load {server_repo_location}")
648
return flask.render_template("not-found.html"), 404
649
repo = git.Repo(server_repo_location)
651
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
652
user = User.query.filter_by(username=flask.session.get("username")).first()
653
relationships = RepoAccess.query.filter_by(repo=repo_data)
654
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
655
return flask.render_template(
657
"repo-forum.html",
658
username=username,
659
repository=repository,
660
repo_data=repo_data,
661
relationships=relationships,
662
repo=repo,
663
user_relationship=user_relationship,
664
Post=Post,
665
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
666
is_favourite=get_favourite(flask.session.get("username"), username, repository),
667
default_branch=repo_data.default_branch
668
)
669
@repositories.route("/<username>/<repository>/forum/topic/<int:id>")
672
def repository_forum_topic(username, repository, id):
673
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
674
if not os.path.exists(server_repo_location):
675
app.logger.error(f"Cannot load {server_repo_location}")
676
flask.abort(404)
677
if not (get_visibility(username, repository) or get_permission_level(
678
flask.session.get("username"), username,
679
repository) is not None):
680
flask.abort(403)
681
app.logger.info(f"Loading {server_repo_location}")
683
if not os.path.exists(server_repo_location):
685
app.logger.error(f"Cannot load {server_repo_location}")
686
return flask.render_template("not-found.html"), 404
687
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
689
user = User.query.filter_by(username=flask.session.get("username")).first()
690
relationships = RepoAccess.query.filter_by(repo=repo_data)
691
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
692
post = Post.query.filter_by(id=id).first()
694
return flask.render_template(
696
"repo-topic.html",
697
username=username,
698
repository=repository,
699
repo_data=repo_data,
700
relationships=relationships,
701
user_relationship=user_relationship,
702
post=post,
703
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
704
is_favourite=get_favourite(flask.session.get("username"), username, repository),
705
default_branch=repo_data.default_branch
706
)
707
@repositories.route("/<username>/<repository>/forum/new", methods=["POST", "GET"])
710
def repository_forum_new(username, repository):
711
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
712
if not os.path.exists(server_repo_location):
713
app.logger.error(f"Cannot load {server_repo_location}")
714
flask.abort(404)
715
if not (get_visibility(username, repository) or get_permission_level(
716
flask.session.get("username"), username,
717
repository) is not None):
718
flask.abort(403)
719
app.logger.info(f"Loading {server_repo_location}")
721
if not os.path.exists(server_repo_location):
723
app.logger.error(f"Cannot load {server_repo_location}")
724
return flask.render_template("not-found.html"), 404
725
repo = git.Repo(server_repo_location)
727
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
728
user = User.query.filter_by(username=flask.session.get("username")).first()
729
relationships = RepoAccess.query.filter_by(repo=repo_data)
730
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
731
post = Post(user, repo_data, None, flask.request.form["subject"],
733
flask.request.form["message"])
734
db.session.add(post)
736
db.session.commit()
737
return flask.redirect(
739
flask.url_for(".repository_forum_thread", username=username, repository=repository,
740
post_id=post.number),
741
code=303)
742
@repositories.route("/<username>/<repository>/forum/<int:post_id>")
745
def repository_forum_thread(username, repository, post_id):
746
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
747
if not os.path.exists(server_repo_location):
748
app.logger.error(f"Cannot load {server_repo_location}")
749
flask.abort(404)
750
if not (get_visibility(username, repository) or get_permission_level(
751
flask.session.get("username"), username,
752
repository) is not None):
753
flask.abort(403)
754
app.logger.info(f"Loading {server_repo_location}")
756
if not os.path.exists(server_repo_location):
758
app.logger.error(f"Cannot load {server_repo_location}")
759
return flask.render_template("not-found.html"), 404
760
repo = git.Repo(server_repo_location)
762
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
763
user = User.query.filter_by(username=flask.session.get("username")).first()
764
relationships = RepoAccess.query.filter_by(repo=repo_data)
765
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
766
return flask.render_template(
768
"repo-forum-thread.html",
769
username=username,
770
repository=repository,
771
repo_data=repo_data,
772
relationships=relationships,
773
repo=repo,
774
Post=Post,
775
user_relationship=user_relationship,
776
post_id=post_id,
777
max_post_nesting=4,
778
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
779
is_favourite=get_favourite(flask.session.get("username"), username, repository),
780
parent=Post.query.filter_by(repo=repo_data, number=post_id).first(),
781
)
782
@repositories.route("/<username>/<repository>/forum/<int:post_id>/reply", methods=["POST"])
785
def repository_forum_reply(username, repository, post_id):
786
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
787
if not os.path.exists(server_repo_location):
788
app.logger.error(f"Cannot load {server_repo_location}")
789
flask.abort(404)
790
if not (get_visibility(username, repository) or get_permission_level(
791
flask.session.get("username"), username,
792
repository) is not None):
793
flask.abort(403)
794
app.logger.info(f"Loading {server_repo_location}")
796
if not os.path.exists(server_repo_location):
798
app.logger.error(f"Cannot load {server_repo_location}")
799
return flask.render_template("not-found.html"), 404
800
repo = git.Repo(server_repo_location)
802
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
803
user = User.query.filter_by(username=flask.session.get("username")).first()
804
relationships = RepoAccess.query.filter_by(repo=repo_data)
805
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
806
if not user:
807
flask.abort(401)
808
parent = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
810
post = Post(user, repo_data, parent, flask.request.form["subject"],
811
flask.request.form["message"])
812
db.session.add(post)
814
post.update_date()
815
db.session.commit()
816
return flask.redirect(
818
flask.url_for(".repository_forum_thread", username=username, repository=repository,
819
post_id=post_id),
820
code=303)
821
@repositories.route("/<username>/<repository>/forum/<int:post_id>/voteup",
824
defaults={"score": 1})
825
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votedown",
826
defaults={"score": -1})
827
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votes", defaults={"score": 0})
828
def repository_forum_vote(username, repository, post_id, score):
829
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
830
if not os.path.exists(server_repo_location):
831
app.logger.error(f"Cannot load {server_repo_location}")
832
flask.abort(404)
833
if not (get_visibility(username, repository) or get_permission_level(
834
flask.session.get("username"), username,
835
repository) is not None):
836
flask.abort(403)
837
app.logger.info(f"Loading {server_repo_location}")
839
if not os.path.exists(server_repo_location):
841
app.logger.error(f"Cannot load {server_repo_location}")
842
return flask.render_template("not-found.html"), 404
843
repo = git.Repo(server_repo_location)
845
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
846
user = User.query.filter_by(username=flask.session.get("username")).first()
847
relationships = RepoAccess.query.filter_by(repo=repo_data)
848
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
849
if not user:
850
flask.abort(401)
851
post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
853
if score:
855
old_relationship = PostVote.query.filter_by(user_username=user.username,
856
post_identifier=post.identifier).first()
857
if old_relationship:
858
if score == old_relationship.vote_score:
859
db.session.delete(old_relationship)
860
post.vote_sum -= old_relationship.vote_score
861
else:
862
post.vote_sum -= old_relationship.vote_score
863
post.vote_sum += score
864
old_relationship.vote_score = score
865
else:
866
relationship = PostVote(user, post, score)
867
post.vote_sum += score
868
db.session.add(relationship)
869
db.session.commit()
871
user_vote = PostVote.query.filter_by(user_username=user.username,
873
post_identifier=post.identifier).first()
874
response = flask.make_response(
875
str(post.vote_sum) + " " + str(user_vote.vote_score if user_vote else 0))
876
response.content_type = "text/plain"
877
return response
879
@repositories.route("/<username>/<repository>/favourite")
882
def repository_favourite(username, repository):
883
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
884
if not os.path.exists(server_repo_location):
885
app.logger.error(f"Cannot load {server_repo_location}")
886
flask.abort(404)
887
if not (get_visibility(username, repository) or get_permission_level(
888
flask.session.get("username"), username,
889
repository) is not None):
890
flask.abort(403)
891
app.logger.info(f"Loading {server_repo_location}")
893
if not os.path.exists(server_repo_location):
895
app.logger.error(f"Cannot load {server_repo_location}")
896
return flask.render_template("not-found.html"), 404
897
repo = git.Repo(server_repo_location)
899
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
900
user = User.query.filter_by(username=flask.session.get("username")).first()
901
relationships = RepoAccess.query.filter_by(repo=repo_data)
902
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
903
if not user:
904
flask.abort(401)
905
old_relationship = RepoFavourite.query.filter_by(user_username=user.username,
907
repo_route=repo_data.route).first()
908
if old_relationship:
909
db.session.delete(old_relationship)
910
else:
911
relationship = RepoFavourite(user, repo_data)
912
db.session.add(relationship)
913
db.session.commit()
915
return flask.redirect(flask.url_for("favourites"), code=303)
917
@repositories.route("/<username>/<repository>/users/", methods=["GET", "POST"])
920
def repository_users(username, repository):
921
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
922
if not os.path.exists(server_repo_location):
923
app.logger.error(f"Cannot load {server_repo_location}")
924
flask.abort(404)
925
if not (get_visibility(username, repository) or get_permission_level(
926
flask.session.get("username"), username,
927
repository) is not None):
928
flask.abort(403)
929
app.logger.info(f"Loading {server_repo_location}")
931
if not os.path.exists(server_repo_location):
933
app.logger.error(f"Cannot load {server_repo_location}")
934
return flask.render_template("not-found.html"), 404
935
repo = git.Repo(server_repo_location)
937
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
938
user = User.query.filter_by(username=flask.session.get("username")).first()
939
relationships = RepoAccess.query.filter_by(repo=repo_data)
940
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
941
if flask.request.method == "GET":
943
return flask.render_template(
944
"repo-users.html",
945
username=username,
946
repository=repository,
947
repo_data=repo_data,
948
relationships=relationships,
949
repo=repo,
950
user_relationship=user_relationship,
951
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
952
is_favourite=get_favourite(flask.session.get("username"), username, repository)
953
)
954
else:
955
if get_permission_level(flask.session.get("username"), username, repository) != 2:
956
flask.abort(401)
957
if flask.request.form.get("new-username"):
959
# Create new relationship
960
new_user = User.query.filter_by(
961
username=flask.request.form.get("new-username")).first()
962
relationship = RepoAccess(new_user, repo_data, flask.request.form.get("new-level"))
963
db.session.add(relationship)
964
db.session.commit()
965
if flask.request.form.get("update-username"):
966
# Create new relationship
967
updated_user = User.query.filter_by(
968
username=flask.request.form.get("update-username")).first()
969
relationship = RepoAccess.query.filter_by(repo=repo_data, user=updated_user).first()
970
if flask.request.form.get("update-level") == -1:
971
relationship.delete()
972
else:
973
relationship.access_level = flask.request.form.get("update-level")
974
db.session.commit()
975
return flask.redirect(
977
app.url_for(".repository_users", username=username, repository=repository))
978
@repositories.route("/<username>/<repository>/branches/")
981
def repository_branches(username, repository):
982
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
983
if not os.path.exists(server_repo_location):
984
app.logger.error(f"Cannot load {server_repo_location}")
985
flask.abort(404)
986
if not (get_visibility(username, repository) or get_permission_level(
987
flask.session.get("username"), username,
988
repository) is not None):
989
flask.abort(403)
990
app.logger.info(f"Loading {server_repo_location}")
992
if not os.path.exists(server_repo_location):
994
app.logger.error(f"Cannot load {server_repo_location}")
995
return flask.render_template("not-found.html"), 404
996
repo = git.Repo(server_repo_location)
998
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
999
return flask.render_template(
1001
"repo-branches.html",
1002
username=username,
1003
repository=repository,
1004
repo_data=repo_data,
1005
repo=repo,
1006
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1007
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1008
)
1009
@repositories.route("/<username>/<repository>/log/", defaults={"branch": None})
1012
@repositories.route("/<username>/<repository>/log/<branch>/")
1013
def repository_log(username, repository, branch):
1014
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1015
if not os.path.exists(server_repo_location):
1016
app.logger.error(f"Cannot load {server_repo_location}")
1017
flask.abort(404)
1018
if not (get_visibility(username, repository) or get_permission_level(
1019
flask.session.get("username"), username,
1020
repository) is not None):
1021
flask.abort(403)
1022
app.logger.info(f"Loading {server_repo_location}")
1024
if not os.path.exists(server_repo_location):
1026
app.logger.error(f"Cannot load {server_repo_location}")
1027
return flask.render_template("not-found.html"), 404
1028
repo = git.Repo(server_repo_location)
1030
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1031
if not repo_data.default_branch:
1032
if repo.heads:
1033
repo_data.default_branch = repo.heads[0].name
1034
else:
1035
return flask.render_template("empty.html",
1036
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
1037
if not branch:
1038
branch = repo_data.default_branch
1039
return flask.redirect(f"./{branch}", code=302)
1040
if branch.startswith("tag:"):
1042
ref = f"tags/{branch[4:]}"
1043
elif branch.startswith("~"):
1044
ref = branch[1:]
1045
else:
1046
ref = f"heads/{branch}"
1047
ref = ref.replace("~", "/") # encode slashes for URL support
1049
try:
1051
repo.git.checkout("-f", ref)
1052
except git.exc.GitCommandError:
1053
return flask.render_template("not-found.html"), 404
1054
branches = repo.heads
1056
all_refs = []
1058
for ref in repo.heads:
1059
all_refs.append((ref, "head"))
1060
for ref in repo.tags:
1061
all_refs.append((ref, "tag"))
1062
commit_list = [f"/{username}/{repository}/{sha}" for sha in
1064
git_command(server_repo_location, None, "log",
1065
"--format='%H'").decode().split("\n")]
1066
commits = Commit.query.filter(Commit.identifier.in_(commit_list))
1068
return flask.render_template(
1070
"repo-log.html",
1071
username=username,
1072
repository=repository,
1073
branches=all_refs,
1074
current=branch,
1075
repo_data=repo_data,
1076
repo=repo,
1077
commits=commits,
1078
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1079
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1080
)
1081
@repositories.route("/<username>/<repository>/prs/", methods=["GET", "POST"])
1084
def repository_prs(username, repository):
1085
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1086
if not os.path.exists(server_repo_location):
1087
app.logger.error(f"Cannot load {server_repo_location}")
1088
flask.abort(404)
1089
if not (get_visibility(username, repository) or get_permission_level(
1090
flask.session.get("username"), username,
1091
repository) is not None):
1092
flask.abort(403)
1093
app.logger.info(f"Loading {server_repo_location}")
1095
if not os.path.exists(server_repo_location):
1097
app.logger.error(f"Cannot load {server_repo_location}")
1098
return flask.render_template("not-found.html"), 404
1099
if flask.request.method == "GET":
1101
repo = git.Repo(server_repo_location)
1102
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1103
user = User.query.filter_by(username=flask.session.get("username")).first()
1104
return flask.render_template(
1106
"repo-prs.html",
1107
username=username,
1108
repository=repository,
1109
repo_data=repo_data,
1110
repo=repo,
1111
PullRequest=PullRequest,
1112
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1113
is_favourite=get_favourite(flask.session.get("username"), username, repository),
1114
default_branch=repo_data.default_branch,
1115
branches=repo.branches
1116
)
1117
else:
1119
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1120
head = flask.request.form.get("head")
1121
head_route = flask.request.form.get("headroute")
1122
base = flask.request.form.get("base")
1123
if not head and base and head_route:
1125
return flask.redirect(".", 400)
1126
head_repo = git.Repo(os.path.join(config.REPOS_PATH, head_route.lstrip("/")))
1128
base_repo = git.Repo(server_repo_location)
1129
print(head_repo)
1130
if head not in head_repo.branches or base not in base_repo.branches:
1132
flask.flash(Markup(
1133
"<iconify-icon icon='mdi:error'></iconify-icon>" + _("Bad branch name")),
1134
category="error")
1135
return flask.redirect(".", 303)
1136
head_data = db.session.get(Repo, head_route)
1138
if not head_data.visibility:
1139
flask.flash(Markup(
1140
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
1141
"Head can't be restricted")),
1142
category="error")
1143
return flask.redirect(".", 303)
1144
pull_request = PullRequest(repo_data, head, head_data, base,
1146
db.session.get(User, flask.session["username"]))
1147
db.session.add(pull_request)
1149
db.session.commit()
1150
return flask.redirect(".", 303)
1152
@repositories.route("/<username>/<repository>/prs/merge", methods=["POST"])
1155
def repository_prs_merge(username, repository):
1156
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1157
if not os.path.exists(server_repo_location):
1158
app.logger.error(f"Cannot load {server_repo_location}")
1159
flask.abort(404)
1160
if not (get_visibility(username, repository) or get_permission_level(
1161
flask.session.get("username"), username,
1162
repository) is not None):
1163
flask.abort(403)
1164
if not get_permission_level(flask.session.get("username"), username, repository):
1166
flask.abort(401)
1167
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1169
repo = git.Repo(server_repo_location)
1170
id = flask.request.form.get("id")
1171
pull_request = db.session.get(PullRequest, id)
1173
if pull_request:
1175
result = celery_tasks.merge_heads.delay(
1176
pull_request.head_route,
1177
pull_request.head_branch,
1178
pull_request.base_route,
1179
pull_request.base_branch,
1180
simulate=True
1181
)
1182
task_result = worker.AsyncResult(result.id)
1183
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1185
# db.session.delete(pull_request)
1186
# db.session.commit()
1187
else:
1188
flask.abort(400)
1189
@repositories.route("/<username>/<repository>/prs/<int:id>/merge")
1192
def repository_prs_merge_stage_two(username, repository, id):
1193
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1194
if not os.path.exists(server_repo_location):
1195
app.logger.error(f"Cannot load {server_repo_location}")
1196
flask.abort(404)
1197
if not (get_visibility(username, repository) or get_permission_level(
1198
flask.session.get("username"), username,
1199
repository) is not None):
1200
flask.abort(403)
1201
if not get_permission_level(flask.session.get("username"), username, repository):
1203
flask.abort(401)
1204
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1206
repo = git.Repo(server_repo_location)
1207
pull_request = db.session.get(PullRequest, id)
1209
if pull_request:
1211
result = celery_tasks.merge_heads.delay(
1212
pull_request.head_route,
1213
pull_request.head_branch,
1214
pull_request.base_route,
1215
pull_request.base_branch,
1216
simulate=False
1217
)
1218
task_result = worker.AsyncResult(result.id)
1219
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1221
# db.session.delete(pull_request)
1222
# db.session.commit()
1223
else:
1224
flask.abort(400)
1225
@app.route("/task/<task_id>")
1228
def task_monitor(task_id):
1229
task_result = worker.AsyncResult(task_id)
1230
print(task_result.status)
1231
return flask.render_template("task-monitor.html", result=task_result)
1233
@repositories.route("/<username>/<repository>/prs/delete", methods=["POST"])
1236
def repository_prs_delete(username, repository):
1237
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1238
if not os.path.exists(server_repo_location):
1239
app.logger.error(f"Cannot load {server_repo_location}")
1240
flask.abort(404)
1241
if not (get_visibility(username, repository) or get_permission_level(
1242
flask.session.get("username"), username,
1243
repository) is not None):
1244
flask.abort(403)
1245
if not get_permission_level(flask.session.get("username"), username, repository):
1247
flask.abort(401)
1248
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1250
repo = git.Repo(server_repo_location)
1251
id = flask.request.form.get("id")
1252
pull_request = db.session.get(PullRequest, id)
1254
if pull_request:
1256
db.session.delete(pull_request)
1257
db.session.commit()
1258
return flask.redirect(".", 303)
1260
@repositories.route("/<username>/<repository>/settings/")
1263
def repository_settings(username, repository):
1264
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1265
flask.abort(401)
1266
return flask.render_template("repo-settings.html", username=username, repository=repository)
1268
@app.errorhandler(404)
1271
def e404(error):
1272
return flask.render_template("not-found.html"), 404
1273
@app.errorhandler(401)
1276
def e401(error):
1277
return flask.render_template("unauthorised.html"), 401
1278
@app.errorhandler(403)
1281
def e403(error):
1282
return flask.render_template("forbidden.html"), 403
1283
@app.errorhandler(418)
1286
def e418(error):
1287
return flask.render_template("teapot.html"), 418
1288
@app.errorhandler(405)
1291
def e405(error):
1292
return flask.render_template("method-not-allowed.html"), 405
1293
if __name__ == "__main__":
1296
app.run(debug=True, port=8080, host="0.0.0.0")
1297
app.register_blueprint(repositories)