179 lines
7.5 KiB
Python
179 lines
7.5 KiB
Python
import json
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
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.models.base import MirrorList
|
|
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,
|
|
extra_buttons=[
|
|
{
|
|
"link": url_for("portal.list.list_preview", format_=k),
|
|
"text": f"Preview {v}",
|
|
"style": "secondary"
|
|
}
|
|
for k, v in MirrorList.formats_supported.items()
|
|
],
|
|
**_SECTION_TEMPLATE_VARS
|
|
)
|
|
|
|
|
|
@bp.route('/preview/<format_>')
|
|
def list_preview(format_: str) -> ResponseReturnValue:
|
|
if format_ == "bca":
|
|
return Response(json.dumps(mirror_mapping()), content_type="application/json")
|
|
if format_ == "bc2":
|
|
return Response(json.dumps(mirror_sites()), content_type="application/json")
|
|
if format_ == "bridgelines":
|
|
return Response(json.dumps(bridgelines()), 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_.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
|
|
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')
|
|
|
|
|
|
@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(
|
|
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_.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)
|