majuna/app/portal/list.py

import json
from datetime import datetime
from typing import Optional, Any

from flask import render_template, url_for, flash, redirect, Blueprint, Response
from flask.typing import ResponseReturnValue
from flask_wtf import FlaskForm
from sqlalchemy import exc
from wtforms import SelectField, StringField, SubmitField
from wtforms.validators import DataRequired

from app.extensions import db
from app.lists.bc2 import mirror_sites
from app.lists.bridgelines import bridgelines
from app.lists.mirror_mapping import mirror_mapping
from app.lists.redirector import redirector_data
from app.models.base import MirrorList, Pool
from app.portal.util import response_404, view_lifecycle

bp = Blueprint("list", __name__)


_SECTION_TEMPLATE_VARS = {
    "section": "list",
    "help_url": "https://bypass.censorship.guide/user/lists.html"
}


@bp.app_template_filter("provider_name")
def list_provider_name(key: str) -> str:
    return MirrorList.providers_supported.get(key, "Unknown")


@bp.app_template_filter("format_name")
def list_format_name(key: str) -> str:
    return MirrorList.formats_supported.get(key, "Unknown")


@bp.app_template_filter("list_encoding_name")
def list_encoding_name(key: str) -> str:
    return MirrorList.encodings_supported.get(key, "Unknown")


@bp.route('/list')
def list_list() -> ResponseReturnValue:
    lists = MirrorList.query.filter(MirrorList.destroyed.is_(None)).all()
    return render_template("list.html.j2",
                           title="Distribution Lists",
                           item="distribution list",
                           new_link=url_for("portal.list.list_new"),
                           items=lists,
                           **_SECTION_TEMPLATE_VARS
                           )


@bp.route('/preview/<format_>/<pool_id>')
def list_preview(format_: str, pool_id: int) -> ResponseReturnValue:
    pool = Pool.query.filter(Pool.id == pool_id).first()
    if not pool:
        return response_404(message="Pool not found")
    if format_ == "bca":
        return Response(json.dumps(mirror_mapping(pool)), content_type="application/json")
    if format_ == "bc2":
        return Response(json.dumps(mirror_sites(pool)), content_type="application/json")
    if format_ == "bridgelines":
        return Response(json.dumps(bridgelines(pool)), content_type="application/json")
    if format_ == "rdr":
        return Response(json.dumps(redirector_data(pool)), content_type="application/json")
    return response_404(message="Format not found")


@bp.route("/destroy/<list_id>", methods=['GET', 'POST'])
def list_destroy(list_id: int) -> ResponseReturnValue:
    list_ = MirrorList.query.filter(MirrorList.id == list_id, MirrorList.destroyed.is_(None)).first()
    if list_ is None:
        return response_404("The requested bridge configuration could not be found.")
    return view_lifecycle(
        header="Destroy mirror list?",
        message=list_.description,
        success_view="portal.list.list_list",
        success_message="This list will no longer be updated and may be deleted depending on the provider.",
        section="list",
        resource=list_,
        action="destroy"
    )


@bp.route("/new", methods=['GET', 'POST'])
@bp.route("/new/<group_id>", methods=['GET', 'POST'])
def list_new(group_id: Optional[int] = None) -> ResponseReturnValue:
    form = NewMirrorListForm()
    form.provider.choices = list(MirrorList.providers_supported.items())
    form.format.choices = list(MirrorList.formats_supported.items())
    form.encoding.choices = list(MirrorList.encodings_supported.items())
    if form.validate_on_submit():
        list_ = MirrorList()
        list_.pool_id = form.pool.data
        list_.provider = form.provider.data
        list_.format = form.format.data
        list_.encoding = form.encoding.data
        list_.description = form.description.data
        list_.container = form.container.data
        list_.branch = form.branch.data
        list_.role = form.role.data
        list_.filename = form.filename.data
        list_.created = datetime.utcnow()
        list_.updated = datetime.utcnow()
        try:
            db.session.add(list_)
            db.session.commit()
            flash("Created new mirror list.", "success")
            return redirect(url_for("portal.list.list_list"))
        except exc.SQLAlchemyError:
            flash("Failed to create new mirror list.", "danger")
            return redirect(url_for("portal.list.list_list"))
    if group_id:
        form.group.data = group_id
    return render_template("new.html.j2",
                           form=form,
                           **_SECTION_TEMPLATE_VARS)


class NewMirrorListForm(FlaskForm):  # type: ignore
    pool = SelectField('Resource Pool', validators=[DataRequired()])
    provider = SelectField('Provider', validators=[DataRequired()])
    format = SelectField('Distribution Method', validators=[DataRequired()])
    encoding = SelectField('Encoding', validators=[DataRequired()])
    description = StringField('Description', validators=[DataRequired()])
    container = StringField('Container', validators=[DataRequired()],
                            description="GitHub Project, GitLab Project or AWS S3 bucket name.")
    branch = StringField('Git Branch/AWS Region', validators=[DataRequired()],
                         description="For GitHub/GitLab, set this to the desired branch name, e.g. main. For AWS S3, "
                                     "set this field to the desired region, e.g. us-east-1.")
    role = StringField('Role ARN',
                       description="(Optional) ARN for IAM role to assume for interaction with the S3 bucket.")
    filename = StringField('Filename', validators=[DataRequired()])
    submit = SubmitField('Save Changes')

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.pool.choices = [
            (pool.id, pool.pool_name) for pool in Pool.query.all()
        ]


@bp.route('/edit/<list_id>', methods=['GET', 'POST'])
def list_edit(list_id: int) -> ResponseReturnValue:
    list_: Optional[MirrorList] = MirrorList.query.filter(MirrorList.id == list_id).first()
    if list_ is None:
        return Response(render_template("error.html.j2",
                                        header="404 Distribution List Not Found",
                                        message="The requested distribution list could not be found.",
                                        **_SECTION_TEMPLATE_VARS),
                        status=404)
    form = NewMirrorListForm(
        pool=list_.pool_id,
        provider=list_.provider,
        format=list_.format,
        encoding=list_.encoding,
        description=list_.description,
        container=list_.container,
        branch=list_.branch,
        role=list_.role,
        filename=list_.filename
    )
    form.provider.choices = list(MirrorList.providers_supported.items())
    form.format.choices = list(MirrorList.formats_supported.items())
    form.encoding.choices = list(MirrorList.encodings_supported.items())
    if form.validate_on_submit():
        list_.pool_id = form.pool.data
        list_.provider = form.provider.data
        list_.format = form.format.data
        list_.encoding = form.encoding.data
        list_.description = form.description.data
        list_.container = form.container.data
        list_.branch = form.branch.data
        list_.role = form.role.data
        list_.filename = form.filename.data
        list_.updated = datetime.utcnow()
        try:
            db.session.commit()
            flash("Saved changes to group.", "success")
        except exc.SQLAlchemyError:
            flash("An error occurred saving the changes to the distribution list.", "danger")
    return render_template("distlist.html.j2",
                           list=list_, form=form,
                           **_SECTION_TEMPLATE_VARS)