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.automation import bp as automation 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(automation, url_prefix="/automation") 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/") def destroy_mirror_list(list_id): return "not implemented" @portal.route("/list/new", methods=['GET', 'POST']) @portal.route("/list/new/", 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)