majuna/app/portal/list.py

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

from flask import Blueprint, Response, flash, redirect, render_template, url_for
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_.added = datetime.now(tz=timezone.utc)
        list_.updated = datetime.now(tz=timezone.utc)
        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.now(tz=timezone.utc)
        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
    )