majuna/app/portal/__init__.py

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.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(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)