127 lines
5 KiB
Python
127 lines
5 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
from flask import Blueprint, render_template, flash, redirect, url_for, request
|
|
from sqlalchemy import exc, desc, or_
|
|
|
|
from app.extensions import db
|
|
from app.models.alarms import Alarm
|
|
from app import Origin, Proxy
|
|
from app.models.base import Group, MirrorList
|
|
from app.portal.forms import LifecycleForm, NewMirrorListForm
|
|
from app.portal.bridgeconf import bp as bridgeconf
|
|
from app.portal.bridge import bp as bridge
|
|
from app.portal.group import bp as group
|
|
from app.portal.origin import bp as origin
|
|
from app.portal.onion import bp as onion
|
|
from app.portal.proxy import bp as proxy
|
|
from app.portal.util import response_404, view_lifecycle
|
|
|
|
portal = Blueprint("portal", __name__, template_folder="templates", static_folder="static")
|
|
|
|
portal.register_blueprint(bridgeconf, url_prefix="/bridgeconf")
|
|
portal.register_blueprint(bridge, url_prefix="/bridge")
|
|
portal.register_blueprint(group, url_prefix="/group")
|
|
portal.register_blueprint(origin, url_prefix="/origin")
|
|
portal.register_blueprint(onion, url_prefix="/onion")
|
|
portal.register_blueprint(proxy, url_prefix="/proxy")
|
|
|
|
|
|
@portal.app_template_filter("mirror_expiry")
|
|
def calculate_mirror_expiry(s: datetime) -> str:
|
|
expiry = s + timedelta(days=3)
|
|
countdown = expiry - datetime.utcnow()
|
|
if countdown.days == 0:
|
|
return f"{countdown.seconds // 3600} hours"
|
|
return f"{countdown.days} days"
|
|
|
|
|
|
@portal.app_template_filter("format_datetime")
|
|
def format_datetime(s: datetime) -> str:
|
|
if s is None:
|
|
return "Unknown"
|
|
return s.strftime("%a, %d %b %Y %H:%M:%S")
|
|
|
|
|
|
@portal.route("/")
|
|
def portal_home():
|
|
groups = Group.query.order_by(Group.group_name).all()
|
|
now = datetime.now(timezone.utc)
|
|
proxies = Proxy.query.filter(Proxy.destroyed == None).all()
|
|
last24 = len(Proxy.query.filter(Proxy.deprecated > (now - timedelta(days=1))).all())
|
|
last72 = len(Proxy.query.filter(Proxy.deprecated > (now - timedelta(days=3))).all())
|
|
lastweek = len(Proxy.query.filter(Proxy.deprecated > (now - timedelta(days=7))).all())
|
|
return render_template("home.html.j2", section="home", groups=groups, last24=last24, last72=last72,
|
|
lastweek=lastweek, proxies=proxies)
|
|
|
|
|
|
@portal.route("/search")
|
|
def search():
|
|
query = request.args.get("query")
|
|
proxies = Proxy.query.filter(or_(Proxy.url.contains(query)), Proxy.destroyed == None).all()
|
|
origins = Origin.query.filter(or_(Origin.description.contains(query), Origin.domain_name.contains(query))).all()
|
|
return render_template("search.html.j2", section="home", proxies=proxies, origins=origins)
|
|
|
|
|
|
@portal.route('/alarms')
|
|
def view_alarms():
|
|
three_days_ago = datetime.now(timezone.utc) - timedelta(days=3)
|
|
alarms = Alarm.query.filter(Alarm.last_updated >= three_days_ago).order_by(
|
|
Alarm.alarm_state, desc(Alarm.state_changed)).all()
|
|
return render_template("alarms.html.j2", section="alarm", alarms=alarms)
|
|
|
|
|
|
@portal.route('/lists')
|
|
def view_mirror_lists():
|
|
mirrorlists = MirrorList.query.filter(MirrorList.destroyed == None).all()
|
|
return render_template("list.html.j2",
|
|
section="list",
|
|
title="Mirror Lists",
|
|
item="mirror list",
|
|
new_link=url_for("portal.new_mirror_list"),
|
|
items=mirrorlists)
|
|
|
|
|
|
@portal.route("/list/destroy/<list_id>")
|
|
def destroy_mirror_list(list_id):
|
|
return "not implemented"
|
|
|
|
|
|
@portal.route("/list/new", methods=['GET', 'POST'])
|
|
@portal.route("/list/new/<group_id>", methods=['GET', 'POST'])
|
|
def new_mirror_list(group_id=None):
|
|
form = NewMirrorListForm()
|
|
form.provider.choices = [
|
|
("github", "GitHub"),
|
|
("gitlab", "GitLab"),
|
|
("s3", "AWS S3"),
|
|
]
|
|
form.format.choices = [
|
|
("bc2", "Bypass Censorship v2"),
|
|
("bc3", "Bypass Censorship v3"),
|
|
("bca", "Bypass Censorship Analytics"),
|
|
("bridgelines", "Tor Bridge Lines")
|
|
]
|
|
form.container.description = "GitHub Project, GitLab Project or AWS S3 bucket name."
|
|
form.branch.description = "Ignored for AWS S3."
|
|
if form.validate_on_submit():
|
|
mirror_list = MirrorList()
|
|
mirror_list.provider = form.provider.data
|
|
mirror_list.format = form.format.data
|
|
mirror_list.description = form.description.data
|
|
mirror_list.container = form.container.data
|
|
mirror_list.branch = form.branch.data
|
|
mirror_list.filename = form.filename.data
|
|
mirror_list.created = datetime.utcnow()
|
|
mirror_list.updated = datetime.utcnow()
|
|
try:
|
|
db.session.add(mirror_list)
|
|
db.session.commit()
|
|
flash(f"Created new mirror list.", "success")
|
|
return redirect(url_for("portal.view_mirror_lists"))
|
|
except exc.SQLAlchemyError as e:
|
|
print(e)
|
|
flash("Failed to create new mirror list.", "danger")
|
|
return redirect(url_for("portal.view_mirror_lists"))
|
|
if group_id:
|
|
form.group.data = group_id
|
|
return render_template("new.html.j2", section="list", form=form)
|