from datetime import datetime from typing import Optional, List from flask import render_template, url_for, flash, redirect, Response, Blueprint from flask.typing import ResponseReturnValue from flask_wtf import FlaskForm from sqlalchemy import exc from wtforms import SelectField, StringField, IntegerField, SubmitField from wtforms.validators import DataRequired, NumberRange from app.extensions import db from app.models.base import Group from app.models.bridges import BridgeConf from app.portal.util import response_404, view_lifecycle bp = Blueprint("bridgeconf", __name__) _SECTION_TEMPLATE_VARS = { "section": "bridgeconf", "help_url": "https://bypass.censorship.guide/user/bridges.html" } class NewBridgeConfForm(FlaskForm): # type: ignore provider = SelectField('Provider', validators=[DataRequired()]) method = SelectField('Distribution Method', validators=[DataRequired()]) description = StringField('Description') group = SelectField('Group', validators=[DataRequired()]) number = IntegerField('Number', validators=[NumberRange(1, message="One or more bridges must be created")]) submit = SubmitField('Save Changes') class EditBridgeConfForm(FlaskForm): # type: ignore description = StringField('Description') number = IntegerField('Number', validators=[NumberRange(1, message="One or more bridges must be created")]) submit = SubmitField('Save Changes') @bp.route("/list") def bridgeconf_list() -> ResponseReturnValue: bridgeconfs: List[BridgeConf] = BridgeConf.query.filter(BridgeConf.destroyed.is_(None)).all() return render_template("list.html.j2", title="Tor Bridge Configurations", item="bridge configuration", items=bridgeconfs, new_link=url_for("portal.bridgeconf.bridgeconf_new"), **_SECTION_TEMPLATE_VARS) @bp.route("/new", methods=['GET', 'POST']) @bp.route("/new/", methods=['GET', 'POST']) def bridgeconf_new(group_id: Optional[int] = None) -> ResponseReturnValue: form = NewBridgeConfForm() form.group.choices = [(x.id, x.group_name) for x in Group.query.all()] form.provider.choices = [ ("aws", "AWS Lightsail"), ("hcloud", "Hetzner Cloud"), ("ovh", "OVH Public Cloud"), ("gandi", "GandiCloud VPS") ] form.method.choices = [ ("any", "Any (BridgeDB)"), ("email", "E-Mail (BridgeDB)"), ("moat", "Moat (BridgeDB)"), ("https", "HTTPS (BridgeDB)"), ("none", "None (Private)") ] if form.validate_on_submit(): bridge_conf = BridgeConf() bridge_conf.group_id = form.group.data bridge_conf.provider = form.provider.data bridge_conf.method = form.method.data bridge_conf.description = form.description.data bridge_conf.number = form.number.data bridge_conf.created = datetime.utcnow() bridge_conf.updated = datetime.utcnow() try: db.session.add(bridge_conf) db.session.commit() flash(f"Created new bridge configuration {bridge_conf.id}.", "success") return redirect(url_for("portal.bridgeconf.bridgeconf_list")) except exc.SQLAlchemyError: flash("Failed to create new bridge configuration.", "danger") return redirect(url_for("portal.bridgeconf.bridgeconf_list")) if group_id: form.group.data = group_id return render_template("new.html.j2", form=form, **_SECTION_TEMPLATE_VARS) @bp.route('/edit/', methods=['GET', 'POST']) def bridgeconf_edit(bridgeconf_id: int) -> ResponseReturnValue: bridgeconf = BridgeConf.query.filter(BridgeConf.id == bridgeconf_id).first() if bridgeconf is None: return Response(render_template("error.html.j2", header="404 Bridge Configuration Not Found", message="The requested bridge configuration could not be found.", **_SECTION_TEMPLATE_VARS), status=404) form = EditBridgeConfForm(description=bridgeconf.description, number=bridgeconf.number) if form.validate_on_submit(): bridgeconf.description = form.description.data bridgeconf.number = form.number.data bridgeconf.updated = datetime.utcnow() try: db.session.commit() flash("Saved changes to bridge configuration.", "success") except exc.SQLAlchemyError: flash("An error occurred saving the changes to the bridge configuration.", "danger") return render_template("bridgeconf.html.j2", bridgeconf=bridgeconf, form=form, **_SECTION_TEMPLATE_VARS) @bp.route("/destroy/", methods=['GET', 'POST']) def bridgeconf_destroy(bridgeconf_id: int) -> ResponseReturnValue: bridgeconf = BridgeConf.query.filter(BridgeConf.id == bridgeconf_id, BridgeConf.destroyed.is_(None)).first() if bridgeconf is None: return response_404("The requested bridge configuration could not be found.") return view_lifecycle( header="Destroy bridge configuration?", message=bridgeconf.description, success_view="portal.bridgeconf.bridgeconf_list", success_message="All bridges from the destroyed configuration will shortly be destroyed at their providers.", resource=bridgeconf, action="destroy", **_SECTION_TEMPLATE_VARS )