Python script, Unicode text, UTF-8 text executable
1
import os
2
import shutil
3
import random
4
import subprocess
5
import platform
6
import git
7
import mimetypes
8
import magic
9
import flask
10
import cairosvg
11
import celery
12
from functools import wraps
13
from datetime import datetime
14
from enum import Enum
15
from cairosvg import svg2png
16
from flask_sqlalchemy import SQLAlchemy
17
from flask_bcrypt import Bcrypt
18
from markupsafe import escape, Markup
19
from flask_migrate import Migrate
20
from PIL import Image
21
from flask_httpauth import HTTPBasicAuth
22
import config
23
from flask_babel import Babel, gettext, ngettext, force_locale
24
_ = gettext
26
n_ = gettext
27
app = flask.Flask(__name__)
29
app.config.from_mapping(
30
CELERY=dict(
31
broker_url=config.REDIS_URI,
32
result_backend=config.REDIS_URI,
33
task_ignore_result=True,
34
),
35
)
36
auth = HTTPBasicAuth()
38
app.config["SQLALCHEMY_DATABASE_URI"] = config.DB_URI
40
app.config["SECRET_KEY"] = config.DB_PASSWORD
41
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
42
app.config['BABEL_TRANSLATION_DIRECTORIES'] = "i18n"
43
db = SQLAlchemy(app)
45
bcrypt = Bcrypt(app)
46
migrate = Migrate(app, db)
47
from models import *
49
from misc_utils import *
50
import git_http
52
import jinja_utils
53
import celery_tasks
54
from celery import Celery, Task
55
import celery_integration
56
babel = Babel(app)
59
def get_locale():
62
if flask.request.cookies.get("language"):
63
return flask.request.cookies.get("language")
64
return flask.request.accept_languages.best_match(config.available_locales)
65
babel.init_app(app, locale_selector=get_locale)
68
with app.app_context():
71
locale_names = {}
72
for language in config.available_locales:
73
with force_locale(language):
74
# NOTE: Translate this to the language's name in that language, for example in French you would use français
75
locale_names[language] = gettext("English")
76
worker = celery_integration.init_celery_app(app)
79
repositories = flask.Blueprint("repository", __name__, template_folder="templates/repository/")
81
@app.context_processor
84
def default():
85
username = flask.session.get("username")
86
user_object = User.query.filter_by(username=username).first()
88
return {
90
"logged_in_user": username,
91
"user_object": user_object,
92
"Notification": Notification,
93
"unread": UserNotification.query.filter_by(user_username=username).filter(
94
UserNotification.attention_level > 0).count(),
95
"config": config,
96
"Markup": Markup,
97
"locale_names": locale_names,
98
}
99
@app.route("/")
102
def main():
103
if flask.session.get("username"):
104
return flask.render_template("home.html")
105
else:
106
return flask.render_template("no-home.html")
107
@app.route("/about/")
110
def about():
111
return flask.render_template("about.html", platform=platform)
112
@app.route("/language", methods=["POST"])
115
def set_locale():
116
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/", code=303)
117
if not flask.request.form.get("language"):
118
response.delete_cookie("language")
119
else:
120
response.set_cookie("language", flask.request.form.get("language"))
121
return response
123
@app.route("/cookie-dismiss")
126
def dismiss_banner():
127
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/", code=303)
128
response.set_cookie("cookie-banner", "1")
129
return response
130
@app.route("/help/")
133
def help_index():
134
return flask.render_template("help.html", faqs=config.faqs)
135
@app.route("/settings/", methods=["GET", "POST"])
138
def settings():
139
if not flask.session.get("username"):
140
flask.abort(401)
141
if flask.request.method == "GET":
142
user = User.query.filter_by(username=flask.session.get("username")).first()
143
return flask.render_template("user-settings.html", user=user)
145
else:
146
user = User.query.filter_by(username=flask.session.get("username")).first()
147
user.display_name = flask.request.form["displayname"]
149
user.URL = flask.request.form["url"]
150
user.company = flask.request.form["company"]
151
user.company_URL = flask.request.form["companyurl"]
152
user.email = flask.request.form.get("email") if flask.request.form.get("email") else None
153
user.location = flask.request.form["location"]
154
user.show_mail = True if flask.request.form.get("showmail") else False
155
user.bio = flask.request.form.get("bio")
156
db.session.commit()
158
flask.flash(Markup("<iconify-icon icon='mdi:check'></iconify-icon>" + _("Settings saved")), category="success")
160
return flask.redirect(f"/{flask.session.get('username')}", code=303)
161
@app.route("/favourites/", methods=["GET", "POST"])
164
def favourites():
165
if not flask.session.get("username"):
166
flask.abort(401)
167
if flask.request.method == "GET":
168
relationships = RepoFavourite.query.filter_by(user_username=flask.session.get("username"))
169
return flask.render_template("favourites.html", favourites=relationships)
171
@app.route("/notifications/", methods=["GET", "POST"])
174
def notifications():
175
if not flask.session.get("username"):
176
flask.abort(401)
177
if flask.request.method == "GET":
178
return flask.render_template("notifications.html", notifications=UserNotification.query.filter_by(
179
user_username=flask.session.get("username")))
180
@app.route("/accounts/", methods=["GET", "POST"])
183
def login():
184
if flask.request.method == "GET":
185
return flask.render_template("login.html")
186
else:
187
if "login" in flask.request.form:
188
username = flask.request.form["username"]
189
password = flask.request.form["password"]
190
user = User.query.filter_by(username=username).first()
192
if user and bcrypt.check_password_hash(user.password_hashed, password):
194
flask.session["username"] = user.username
195
flask.flash(
196
Markup("<iconify-icon icon='mdi:account'></iconify-icon>" + _("Successfully logged in as {username}").format(username=username)),
197
category="success")
198
return flask.redirect("/", code=303)
199
elif not user:
200
flask.flash(Markup("<iconify-icon icon='mdi:account-question'></iconify-icon>" + _("User not found")),
201
category="alert")
202
return flask.render_template("login.html")
203
else:
204
flask.flash(Markup("<iconify-icon icon='mdi:account-question'></iconify-icon>" + _("Invalid password")),
205
category="error")
206
return flask.render_template("login.html")
207
if "signup" in flask.request.form:
208
username = flask.request.form["username"]
209
password = flask.request.form["password"]
210
password2 = flask.request.form["password2"]
211
email = flask.request.form.get("email")
212
email2 = flask.request.form.get("email2") # repeat email is a honeypot
213
name = flask.request.form.get("name")
214
if not only_chars(username, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
216
flask.flash(Markup(
217
_("Usernames may only contain Latin alphabet, numbers, '-' and '_'")),
218
category="error")
219
return flask.render_template("login.html")
220
if username in config.RESERVED_NAMES:
222
flask.flash(
223
Markup(
224
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _("Sorry, {username} is a system path").format(username=username)),
225
category="error")
226
return flask.render_template("login.html")
227
user_check = User.query.filter_by(username=username).first()
229
if user_check:
230
flask.flash(
231
Markup(
232
"<iconify-icon icon='mdi:account-error'></iconify-icon>" + _("The username {username} is taken").format(username=username)),
233
category="error")
234
return flask.render_template("login.html")
235
if password2 != password:
237
flask.flash(Markup("<iconify-icon icon='mdi:key-alert'></iconify-icon>" + _("Make sure the passwords match")),
238
category="error")
239
return flask.render_template("login.html")
240
user = User(username, password, email, name)
242
db.session.add(user)
243
db.session.commit()
244
flask.session["username"] = user.username
245
flask.flash(Markup(
246
"<iconify-icon icon='mdi:account'></iconify-icon>" + _("Successfully created and logged in as {username}").format(username=username)),
247
category="success")
248
notification = Notification({"type": "welcome"})
250
db.session.add(notification)
251
db.session.commit()
252
result = celery_tasks.send_notification.delay(notification.id, [username], 1)
254
return flask.redirect("/", code=303)
256
@app.route("/newrepo/", methods=["GET", "POST"])
259
def new_repo():
260
if not flask.session.get("username"):
261
flask.abort(401)
262
if flask.request.method == "GET":
263
return flask.render_template("new-repo.html")
264
else:
265
name = flask.request.form["name"]
266
visibility = int(flask.request.form["visibility"])
267
if not only_chars(name, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
269
flask.flash(Markup(
270
"<iconify-icon icon='mdi:error'></iconify-icon>" + _("Repository names may only contain Latin alphabet, numbers, '-' and '_'")),
271
category="error")
272
return flask.render_template("new-repo.html")
273
user = User.query.filter_by(username=flask.session.get("username")).first()
275
repo = Repo(user, name, visibility)
277
db.session.add(repo)
278
db.session.commit()
279
if not os.path.exists(os.path.join(config.REPOS_PATH, repo.route)):
281
subprocess.run(["git", "init", repo.name],
282
cwd=os.path.join(config.REPOS_PATH, flask.session.get("username")))
283
flask.flash(Markup(_("Successfully created repository {name}").format(name=name)),
285
category="success")
286
return flask.redirect(repo.route, code=303)
287
@app.route("/logout")
290
def logout():
291
flask.session.clear()
292
flask.flash(Markup("<iconify-icon icon='mdi:account'></iconify-icon>" + _("Successfully logged out")), category="info")
293
return flask.redirect("/", code=303)
294
@app.route("/<username>/", methods=["GET", "POST"])
297
def user_profile(username):
298
old_relationship = UserFollow.query.filter_by(follower_username=flask.session.get("username"),
299
followed_username=username).first()
300
if flask.request.method == "GET":
301
user = User.query.filter_by(username=username).first()
302
match flask.request.args.get("action"):
303
case "repositories":
304
repos = Repo.query.filter_by(owner_name=username, visibility=2)
305
return flask.render_template("user-profile-repositories.html", user=user, repos=repos,
306
relationship=old_relationship)
307
case "followers":
308
return flask.render_template("user-profile-followers.html", user=user, relationship=old_relationship)
309
case "follows":
310
return flask.render_template("user-profile-follows.html", user=user, relationship=old_relationship)
311
case _:
312
return flask.render_template("user-profile-overview.html", user=user, relationship=old_relationship)
313
elif flask.request.method == "POST":
315
match flask.request.args.get("action"):
316
case "follow":
317
if username == flask.session.get("username"):
318
flask.abort(403)
319
if old_relationship:
320
db.session.delete(old_relationship)
321
else:
322
relationship = UserFollow(
323
flask.session.get("username"),
324
username
325
)
326
db.session.add(relationship)
327
db.session.commit()
328
user = db.session.get(User, username)
330
author = db.session.get(User, flask.session.get("username"))
331
notification = Notification({"type": "update", "version": "0.0.0"})
332
db.session.add(notification)
333
db.session.commit()
334
result = celery_tasks.send_notification.delay(notification.id, [username], 1)
336
db.session.commit()
338
return flask.redirect("?", code=303)
339
@app.route("/<username>/<repository>/")
342
def repository_index(username, repository):
343
return flask.redirect("./tree", code=302)
344
@app.route("/info/<username>/avatar")
347
def user_avatar(username):
348
serverUserdataLocation = os.path.join(config.USERDATA_PATH, username)
349
if not os.path.exists(serverUserdataLocation):
351
return flask.render_template("not-found.html"), 404
352
return flask.send_from_directory(serverUserdataLocation, "avatar.png")
354
@app.route("/<username>/<repository>/raw/<branch>/<path:subpath>")
357
def repository_raw(username, repository, branch, subpath):
358
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
359
repository) is not None):
360
flask.abort(403)
361
serverRepoLocation = os.path.join(config.REPOS_PATH, username, repository)
363
app.logger.info(f"Loading {serverRepoLocation}")
365
if not os.path.exists(serverRepoLocation):
367
app.logger.error(f"Cannot load {serverRepoLocation}")
368
return flask.render_template("not-found.html"), 404
369
repo = git.Repo(serverRepoLocation)
371
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
372
if not repo_data.default_branch:
373
if repo.heads:
374
repo_data.default_branch = repo.heads[0].name
375
else:
376
return flask.render_template("empty.html",
377
remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
378
if not branch:
379
branch = repo_data.default_branch
380
return flask.redirect(f"./{branch}", code=302)
381
if branch.startswith("tag:"):
383
ref = f"tags/{branch[4:]}"
384
elif branch.startswith("~"):
385
ref = branch[1:]
386
else:
387
ref = f"heads/{branch}"
388
ref = ref.replace("~", "/") # encode slashes for URL support
390
try:
392
repo.git.checkout("-f", ref)
393
except git.exc.GitCommandError:
394
return flask.render_template("not-found.html"), 404
395
return flask.send_from_directory(config.REPOS_PATH, os.path.join(username, repository, subpath))
397
@repositories.route("/<username>/<repository>/tree/", defaults={"branch": None, "subpath": ""})
400
@repositories.route("/<username>/<repository>/tree/<branch>/", defaults={"subpath": ""})
401
@repositories.route("/<username>/<repository>/tree/<branch>/<path:subpath>")
402
def repository_tree(username, repository, branch, subpath):
403
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
404
repository) is not None):
405
flask.abort(403)
406
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
408
app.logger.info(f"Loading {server_repo_location}")
410
if not os.path.exists(server_repo_location):
412
app.logger.error(f"Cannot load {server_repo_location}")
413
return flask.render_template("not-found.html"), 404
414
repo = git.Repo(server_repo_location)
416
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
417
if not repo_data.default_branch:
418
if repo.heads:
419
repo_data.default_branch = repo.heads[0].name
420
else:
421
return flask.render_template("empty.html",
422
remote=f"{config.www_protocol}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
423
if not branch:
424
branch = repo_data.default_branch
425
return flask.redirect(f"./{branch}", code=302)
426
if branch.startswith("tag:"):
428
ref = f"tags/{branch[4:]}"
429
elif branch.startswith("~"):
430
ref = branch[1:]
431
else:
432
ref = f"heads/{branch}"
433
ref = ref.replace("~", "/") # encode slashes for URL support
435
try:
437
repo.git.checkout("-f", ref)
438
except git.exc.GitCommandError:
439
return flask.render_template("not-found.html"), 404
440
branches = repo.heads
442
all_refs = []
444
for ref in repo.heads:
445
all_refs.append((ref, "head"))
446
for ref in repo.tags:
447
all_refs.append((ref, "tag"))
448
if os.path.isdir(os.path.join(server_repo_location, subpath)):
450
files = []
451
blobs = []
452
for entry in os.listdir(os.path.join(server_repo_location, subpath)):
454
if not os.path.basename(entry) == ".git":
455
files.append(os.path.join(subpath, entry))
456
infos = []
458
for file in files:
460
path = os.path.join(server_repo_location, file)
461
mimetype = guess_mime(path)
462
text = git_command(server_repo_location, None, "log", "--format='%H\n'", file).decode()
464
sha = text.split("\n")[0]
466
identifier = f"/{username}/{repository}/{sha}"
467
last_commit = Commit.query.filter_by(identifier=identifier).first()
468
info = {
470
"name": os.path.basename(file),
471
"serverPath": path,
472
"relativePath": file,
473
"link": os.path.join(f"/{username}/{repository}/tree/{branch}/", file),
474
"size": human_size(os.path.getsize(path)),
475
"mimetype": f"{mimetype}{f' ({mimetypes.guess_type(path)[1]})' if mimetypes.guess_type(path)[1] else ''}",
476
"commit": last_commit,
477
"shaSize": 7,
478
}
479
special_icon = config.match_icon(os.path.basename(file))
481
if special_icon:
482
info["icon"] = special_icon
483
elif os.path.isdir(path):
484
info["icon"] = config.folder_icon
485
elif mimetypes.guess_type(path)[0] in config.file_icons:
486
info["icon"] = config.file_icons[mimetypes.guess_type(path)[0]]
487
else:
488
info["icon"] = config.unknown_icon
489
if os.path.isdir(path):
491
infos.insert(0, info)
492
else:
493
infos.append(info)
494
return flask.render_template(
496
"repo-tree.html",
497
username=username,
498
repository=repository,
499
files=infos,
500
subpath=os.path.join("/", subpath),
501
branches=all_refs,
502
current=branch,
503
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
504
is_favourite=get_favourite(flask.session.get("username"), username, repository)
505
)
506
else:
507
path = os.path.join(server_repo_location, subpath)
508
if not os.path.exists(path):
510
return flask.render_template("not-found.html"), 404
511
mimetype = guess_mime(path)
513
mode = mimetype.split("/", 1)[0]
514
size = human_size(os.path.getsize(path))
515
special_icon = config.match_icon(os.path.basename(path))
517
if special_icon:
518
icon = special_icon
519
elif os.path.isdir(path):
520
icon = config.folder_icon
521
elif mimetypes.guess_type(path)[0] in config.file_icons:
522
icon = config.file_icons[mimetypes.guess_type(path)[0]]
523
else:
524
icon = config.unknown_icon
525
contents = None
527
if mode == "text":
528
contents = convert_to_html(path)
529
return flask.render_template(
531
"repo-file.html",
532
username=username,
533
repository=repository,
534
file=os.path.join(f"/{username}/{repository}/raw/{branch}/", subpath),
535
branches=all_refs,
536
current=branch,
537
mode=mode,
538
mimetype=mimetype,
539
detailedtype=magic.from_file(path),
540
size=size,
541
icon=icon,
542
subpath=os.path.join("/", subpath),
543
basename=os.path.basename(path),
544
contents=contents,
545
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
546
is_favourite=get_favourite(flask.session.get("username"), username, repository)
547
)
548
@repositories.route("/<username>/<repository>/forum/")
551
def repository_forum(username, repository):
552
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
553
repository) is not None):
554
flask.abort(403)
555
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
557
app.logger.info(f"Loading {server_repo_location}")
559
if not os.path.exists(server_repo_location):
561
app.logger.error(f"Cannot load {server_repo_location}")
562
return flask.render_template("not-found.html"), 404
563
repo = git.Repo(server_repo_location)
565
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
566
user = User.query.filter_by(username=flask.session.get("username")).first()
567
relationships = RepoAccess.query.filter_by(repo=repo_data)
568
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
569
return flask.render_template(
571
"repo-forum.html",
572
username=username,
573
repository=repository,
574
repo_data=repo_data,
575
relationships=relationships,
576
repo=repo,
577
user_relationship=user_relationship,
578
Post=Post,
579
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
580
is_favourite=get_favourite(flask.session.get("username"), username, repository),
581
default_branch=repo_data.default_branch
582
)
583
@repositories.route("/<username>/<repository>/forum/topic/<int:id>")
586
def repository_forum_topic(username, repository, id):
587
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
588
repository) is not None):
589
flask.abort(403)
590
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
592
app.logger.info(f"Loading {server_repo_location}")
594
if not os.path.exists(server_repo_location):
596
app.logger.error(f"Cannot load {server_repo_location}")
597
return flask.render_template("not-found.html"), 404
598
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
600
user = User.query.filter_by(username=flask.session.get("username")).first()
601
relationships = RepoAccess.query.filter_by(repo=repo_data)
602
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
603
post = Post.query.filter_by(id=id).first()
605
return flask.render_template(
607
"repo-topic.html",
608
username=username,
609
repository=repository,
610
repo_data=repo_data,
611
relationships=relationships,
612
user_relationship=user_relationship,
613
post=post,
614
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
615
is_favourite=get_favourite(flask.session.get("username"), username, repository),
616
default_branch=repo_data.default_branch
617
)
618
@repositories.route("/<username>/<repository>/forum/new", methods=["POST", "GET"])
621
def repository_forum_new(username, repository):
622
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
623
repository) is not None):
624
flask.abort(403)
625
serverRepoLocation = os.path.join(config.REPOS_PATH, username, repository)
627
app.logger.info(f"Loading {serverRepoLocation}")
629
if not os.path.exists(serverRepoLocation):
631
app.logger.error(f"Cannot load {serverRepoLocation}")
632
return flask.render_template("not-found.html"), 404
633
repo = git.Repo(serverRepoLocation)
635
repoData = Repo.query.filter_by(route=f"/{username}/{repository}").first()
636
user = User.query.filter_by(username=flask.session.get("username")).first()
637
relationships = RepoAccess.query.filter_by(repo=repoData)
638
userRelationship = RepoAccess.query.filter_by(repo=repoData, user=user).first()
639
post = Post(user, repoData, None, flask.request.form["subject"], flask.request.form["message"])
641
db.session.add(post)
643
db.session.commit()
644
return flask.redirect(
646
flask.url_for(".repository_forum_thread", username=username, repository=repository, post_id=post.number),
647
code=303)
648
@repositories.route("/<username>/<repository>/forum/<int:post_id>")
651
def repository_forum_thread(username, repository, post_id):
652
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
653
repository) is not None):
654
flask.abort(403)
655
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
657
app.logger.info(f"Loading {server_repo_location}")
659
if not os.path.exists(server_repo_location):
661
app.logger.error(f"Cannot load {server_repo_location}")
662
return flask.render_template("not-found.html"), 404
663
repo = git.Repo(server_repo_location)
665
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
666
user = User.query.filter_by(username=flask.session.get("username")).first()
667
relationships = RepoAccess.query.filter_by(repo=repo_data)
668
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
669
return flask.render_template(
671
"repo-forum-thread.html",
672
username=username,
673
repository=repository,
674
repo_data=repo_data,
675
relationships=relationships,
676
repo=repo,
677
Post=Post,
678
user_relationship=user_relationship,
679
post_id=post_id,
680
max_post_nesting=4,
681
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
682
is_favourite=get_favourite(flask.session.get("username"), username, repository),
683
parent=Post.query.filter_by(repo=repo_data, number=post_id).first(),
684
)
685
@repositories.route("/<username>/<repository>/forum/<int:post_id>/reply", methods=["POST"])
688
def repository_forum_reply(username, repository, post_id):
689
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
690
repository) is not None):
691
flask.abort(403)
692
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
694
app.logger.info(f"Loading {server_repo_location}")
696
if not os.path.exists(server_repo_location):
698
app.logger.error(f"Cannot load {server_repo_location}")
699
return flask.render_template("not-found.html"), 404
700
repo = git.Repo(server_repo_location)
702
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
703
user = User.query.filter_by(username=flask.session.get("username")).first()
704
relationships = RepoAccess.query.filter_by(repo=repo_data)
705
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
706
if not user:
707
flask.abort(401)
708
parent = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
710
post = Post(user, repo_data, parent, flask.request.form["subject"], flask.request.form["message"])
711
db.session.add(post)
713
post.update_date()
714
db.session.commit()
715
return flask.redirect(
717
flask.url_for(".repository_forum_thread", username=username, repository=repository, post_id=post_id),
718
code=303)
719
@repositories.route("/<username>/<repository>/forum/<int:post_id>/voteup", defaults={"score": 1})
722
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votedown", defaults={"score": -1})
723
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votes", defaults={"score": 0})
724
def repository_forum_vote(username, repository, post_id, score):
725
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
726
repository) is not None):
727
flask.abort(403)
728
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
730
app.logger.info(f"Loading {server_repo_location}")
732
if not os.path.exists(server_repo_location):
734
app.logger.error(f"Cannot load {server_repo_location}")
735
return flask.render_template("not-found.html"), 404
736
repo = git.Repo(server_repo_location)
738
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
739
user = User.query.filter_by(username=flask.session.get("username")).first()
740
relationships = RepoAccess.query.filter_by(repo=repo_data)
741
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
742
if not user:
743
flask.abort(401)
744
post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
746
if score:
748
old_relationship = PostVote.query.filter_by(user_username=user.username,
749
post_identifier=post.identifier).first()
750
if old_relationship:
751
if score == old_relationship.vote_score:
752
db.session.delete(old_relationship)
753
post.vote_sum -= old_relationship.vote_score
754
else:
755
post.vote_sum -= old_relationship.vote_score
756
post.vote_sum += score
757
old_relationship.vote_score = score
758
else:
759
relationship = PostVote(user, post, score)
760
post.vote_sum += score
761
db.session.add(relationship)
762
db.session.commit()
764
user_vote = PostVote.query.filter_by(user_username=user.username, post_identifier=post.identifier).first()
766
response = flask.make_response(str(post.vote_sum) + " " + str(user_vote.vote_score if user_vote else 0))
767
response.content_type = "text/plain"
768
return response
770
@repositories.route("/<username>/<repository>/favourite")
773
def repository_favourite(username, repository):
774
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
775
repository) is not None):
776
flask.abort(403)
777
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
779
app.logger.info(f"Loading {server_repo_location}")
781
if not os.path.exists(server_repo_location):
783
app.logger.error(f"Cannot load {server_repo_location}")
784
return flask.render_template("not-found.html"), 404
785
repo = git.Repo(server_repo_location)
787
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
788
user = User.query.filter_by(username=flask.session.get("username")).first()
789
relationships = RepoAccess.query.filter_by(repo=repo_data)
790
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
791
if not user:
792
flask.abort(401)
793
old_relationship = RepoFavourite.query.filter_by(user_username=user.username, repo_route=repo_data.route).first()
795
if old_relationship:
796
db.session.delete(old_relationship)
797
else:
798
relationship = RepoFavourite(user, repo_data)
799
db.session.add(relationship)
800
db.session.commit()
802
return flask.redirect(flask.url_for("favourites"), code=303)
804
@repositories.route("/<username>/<repository>/users/", methods=["GET", "POST"])
807
def repository_users(username, repository):
808
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
809
repository) is not None):
810
flask.abort(403)
811
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
813
app.logger.info(f"Loading {server_repo_location}")
815
if not os.path.exists(server_repo_location):
817
app.logger.error(f"Cannot load {server_repo_location}")
818
return flask.render_template("not-found.html"), 404
819
repo = git.Repo(server_repo_location)
821
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
822
user = User.query.filter_by(username=flask.session.get("username")).first()
823
relationships = RepoAccess.query.filter_by(repo=repo_data)
824
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
825
if flask.request.method == "GET":
827
return flask.render_template(
828
"repo-users.html",
829
username=username,
830
repository=repository,
831
repo_data=repo_data,
832
relationships=relationships,
833
repo=repo,
834
user_relationship=user_relationship,
835
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
836
is_favourite=get_favourite(flask.session.get("username"), username, repository)
837
)
838
else:
839
if get_permission_level(flask.session.get("username"), username, repository) != 2:
840
flask.abort(401)
841
if flask.request.form.get("new-username"):
843
# Create new relationship
844
new_user = User.query.filter_by(username=flask.request.form.get("new-username")).first()
845
relationship = RepoAccess(new_user, repo_data, flask.request.form.get("new-level"))
846
db.session.add(relationship)
847
db.session.commit()
848
if flask.request.form.get("update-username"):
849
# Create new relationship
850
updated_user = User.query.filter_by(username=flask.request.form.get("update-username")).first()
851
relationship = RepoAccess.query.filter_by(repo=repo_data, user=updated_user).first()
852
if flask.request.form.get("update-level") == -1:
853
relationship.delete()
854
else:
855
relationship.access_level = flask.request.form.get("update-level")
856
db.session.commit()
857
return flask.redirect(app.url_for(".repository_users", username=username, repository=repository))
859
@repositories.route("/<username>/<repository>/branches/")
862
def repository_branches(username, repository):
863
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
864
repository) is not None):
865
flask.abort(403)
866
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
868
app.logger.info(f"Loading {server_repo_location}")
870
if not os.path.exists(server_repo_location):
872
app.logger.error(f"Cannot load {server_repo_location}")
873
return flask.render_template("not-found.html"), 404
874
repo = git.Repo(server_repo_location)
876
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
877
return flask.render_template(
879
"repo-branches.html",
880
username=username,
881
repository=repository,
882
repo_data=repo_data,
883
repo=repo,
884
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
885
is_favourite=get_favourite(flask.session.get("username"), username, repository)
886
)
887
@repositories.route("/<username>/<repository>/log/", defaults={"branch": None})
890
@repositories.route("/<username>/<repository>/log/<branch>/")
891
def repository_log(username, repository, branch):
892
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
893
repository) is not None):
894
flask.abort(403)
895
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
897
app.logger.info(f"Loading {server_repo_location}")
899
if not os.path.exists(server_repo_location):
901
app.logger.error(f"Cannot load {server_repo_location}")
902
return flask.render_template("not-found.html"), 404
903
repo = git.Repo(server_repo_location)
905
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
906
if not repo_data.default_branch:
907
if repo.heads:
908
repo_data.default_branch = repo.heads[0].name
909
else:
910
return flask.render_template("empty.html",
911
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
912
if not branch:
913
branch = repo_data.default_branch
914
return flask.redirect(f"./{branch}", code=302)
915
if branch.startswith("tag:"):
917
ref = f"tags/{branch[4:]}"
918
elif branch.startswith("~"):
919
ref = branch[1:]
920
else:
921
ref = f"heads/{branch}"
922
ref = ref.replace("~", "/") # encode slashes for URL support
924
try:
926
repo.git.checkout("-f", ref)
927
except git.exc.GitCommandError:
928
return flask.render_template("not-found.html"), 404
929
branches = repo.heads
931
all_refs = []
933
for ref in repo.heads:
934
all_refs.append((ref, "head"))
935
for ref in repo.tags:
936
all_refs.append((ref, "tag"))
937
commit_list = [f"/{username}/{repository}/{sha}" for sha in
939
git_command(server_repo_location, None, "log", "--format='%H'").decode().split("\n")]
940
commits = Commit.query.filter(Commit.identifier.in_(commit_list))
942
return flask.render_template(
944
"repo-log.html",
945
username=username,
946
repository=repository,
947
branches=all_refs,
948
current=branch,
949
repo_data=repo_data,
950
repo=repo,
951
commits=commits,
952
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
953
is_favourite=get_favourite(flask.session.get("username"), username, repository)
954
)
955
@repositories.route("/<username>/<repository>/prs/", methods=["GET", "POST"])
958
def repository_prs(username, repository):
959
if not (get_visibility(username, repository) or get_permission_level(flask.session.get("username"), username,
960
repository) is not None):
961
flask.abort(403)
962
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
964
app.logger.info(f"Loading {server_repo_location}")
966
if not os.path.exists(server_repo_location):
968
app.logger.error(f"Cannot load {server_repo_location}")
969
return flask.render_template("not-found.html"), 404
970
if flask.request.method == "GET":
972
repo = git.Repo(server_repo_location)
973
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
974
user = User.query.filter_by(username=flask.session.get("username")).first()
975
return flask.render_template(
977
"repo-prs.html",
978
username=username,
979
repository=repository,
980
repo_data=repo_data,
981
repo=repo,
982
PullRequest=PullRequest,
983
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
984
is_favourite=get_favourite(flask.session.get("username"), username, repository),
985
default_branch=repo_data.default_branch,
986
branches=repo.branches
987
)
988
else:
990
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
991
head = flask.request.form.get("head")
992
head_route = flask.request.form.get("headroute")
993
base = flask.request.form.get("base")
994
if not head and base and head_route:
996
return flask.redirect(".", 400)
997
head_repo = git.Repo(os.path.join(config.REPOS_PATH, head_route.lstrip("/")))
999
base_repo = git.Repo(server_repo_location)
1000
print(head_repo)
1001
if head not in head_repo.branches or base not in base_repo.branches:
1003
flask.flash(Markup(
1004
"<iconify-icon icon='mdi:error'></iconify-icon>" + _("Bad branch name")),
1005
category="error")
1006
return flask.redirect(".", 303)
1007
head_data = db.session.get(Repo, head_route)
1009
if not head_data.visibility:
1010
flask.flash(Markup(
1011
"<iconify-icon icon='mdi:error'></iconify-icon>" + _("Head can't be restricted")),
1012
category="error")
1013
return flask.redirect(".", 303)
1014
pull_request = PullRequest(repo_data, head, head_data, base, db.session.get(User, flask.session["username"]))
1016
db.session.add(pull_request)
1018
db.session.commit()
1019
return flask.redirect(".", 303)
1021
@repositories.route("/<username>/<repository>/prs/merge", methods=["POST"])
1024
def repository_prs_merge(username, repository):
1025
if not get_permission_level(flask.session.get("username"), username, repository):
1026
flask.abort(401)
1027
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1029
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1030
repo = git.Repo(server_repo_location)
1031
id = flask.request.form.get("id")
1032
pull_request = db.session.get(PullRequest, id)
1034
if pull_request:
1036
result = celery_tasks.merge_heads.delay(
1037
pull_request.head_route,
1038
pull_request.head_branch,
1039
pull_request.base_route,
1040
pull_request.base_branch,
1041
simulate=True
1042
)
1043
task_result = worker.AsyncResult(result.id)
1044
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1046
# db.session.delete(pull_request)
1047
# db.session.commit()
1048
else:
1049
flask.abort(400)
1050
@repositories.route("/<username>/<repository>/prs/<int:id>/merge")
1053
def repository_prs_merge_stage_two(username, repository, id):
1054
if not get_permission_level(flask.session.get("username"), username, repository):
1055
flask.abort(401)
1056
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1058
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1059
repo = git.Repo(server_repo_location)
1060
pull_request = db.session.get(PullRequest, id)
1062
if pull_request:
1064
result = celery_tasks.merge_heads.delay(
1065
pull_request.head_route,
1066
pull_request.head_branch,
1067
pull_request.base_route,
1068
pull_request.base_branch,
1069
simulate=False
1070
)
1071
task_result = worker.AsyncResult(result.id)
1072
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1074
# db.session.delete(pull_request)
1075
# db.session.commit()
1076
else:
1077
flask.abort(400)
1078
@app.route("/task/<task_id>")
1081
def task_monitor(task_id):
1082
task_result = worker.AsyncResult(task_id)
1083
print(task_result.status)
1084
return flask.render_template("task-monitor.html", result=task_result)
1086
@repositories.route("/<username>/<repository>/prs/delete", methods=["POST"])
1089
def repository_prs_delete(username, repository):
1090
if not get_permission_level(flask.session.get("username"), username, repository):
1091
flask.abort(401)
1092
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1094
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1095
repo = git.Repo(server_repo_location)
1096
id = flask.request.form.get("id")
1097
pull_request = db.session.get(PullRequest, id)
1099
if pull_request:
1101
db.session.delete(pull_request)
1102
db.session.commit()
1103
return flask.redirect(".", 303)
1105
@repositories.route("/<username>/<repository>/settings/")
1108
def repository_settings(username, repository):
1109
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1110
flask.abort(401)
1111
return flask.render_template("repo-settings.html", username=username, repository=repository)
1113
@app.errorhandler(404)
1116
def e404(error):
1117
return flask.render_template("not-found.html"), 404
1118
@app.errorhandler(401)
1121
def e401(error):
1122
return flask.render_template("unauthorised.html"), 401
1123
@app.errorhandler(403)
1126
def e403(error):
1127
return flask.render_template("forbidden.html"), 403
1128
@app.errorhandler(418)
1131
def e418(error):
1132
return flask.render_template("teapot.html"), 418
1133
@app.errorhandler(405)
1136
def e405(error):
1137
return flask.render_template("method-not-allowed.html"), 405
1138
if __name__ == "__main__":
1141
app.run(debug=True, port=8080, host="0.0.0.0")
1142
app.register_blueprint(repositories)
1144