# ContentDB # Copyright (C) 2023 rubenwardy # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import urllib.parse as urlparse from urllib.parse import urlencode import typing from flask import Blueprint, render_template, redirect, url_for, request, jsonify, abort, make_response, flash from flask_babel import lazy_gettext, gettext from flask_login import current_user, login_required from flask_wtf import FlaskForm from wtforms import StringField, SubmitField, URLField, SelectField from wtforms.validators import InputRequired, Length, Optional from app import csrf from app.blueprints.users.settings import get_setting_tabs from app.models import db, OAuthClient, User, Permission, APIToken, AuditSeverity, UserRank from app.utils import random_string, add_audit_log bp = Blueprint("oauth", __name__) def build_redirect_url(url: str, code: str, state: typing.Optional[str]): params = {"code": code} if state is not None: params["state"] = state url_parts = list(urlparse.urlparse(url)) query = dict(urlparse.parse_qsl(url_parts[4])) query.update(params) url_parts[4] = urlencode(query) return urlparse.urlunparse(url_parts) @bp.route("/oauth/authorize/", methods=["GET", "POST"]) @login_required def oauth_start(): response_type = request.args.get("response_type", "code") if response_type != "code": return "Unsupported response_type, only code is supported", 400 client_id = request.args.get("client_id", "") if client_id == "": return "Missing client_id", 400 redirect_uri = request.args.get("redirect_uri", "") if redirect_uri == "": return "Missing redirect_uri", 400 client = OAuthClient.query.get_or_404(client_id) if client.redirect_url != redirect_uri: return "redirect_uri does not match client", 400 if not client.approved and client.owner != current_user: abort(404) scope = request.args.get("scope", "public") if scope != "public": return "Unsupported scope, only public is supported", 400 state = request.args.get("state") token = APIToken.query.filter(APIToken.client == client, APIToken.owner == current_user).first() if token: token.access_token = random_string(32) token.auth_code = random_string(32) db.session.commit() return redirect(build_redirect_url(client.redirect_url, token.auth_code, state)) if request.method == "POST": action = request.form["action"] if action == "cancel": return redirect(client.redirect_url) elif action == "authorize": token = APIToken() token.access_token = random_string(32) token.name = f"Token for {client.title} by {client.owner.username}" token.owner = current_user token.client = client assert client is not None token.auth_code = random_string(32) db.session.add(token) add_audit_log(AuditSeverity.USER, current_user, f"Granted \"{scope}\" to OAuth2 application \"{client.title}\" by {client.owner.username} [{client_id}] ", url_for("users.profile", username=current_user.username)) db.session.commit() return redirect(build_redirect_url(client.redirect_url, token.auth_code, state)) return render_template("oauth/authorize.html", client=client) def error(code: int, msg: str): abort(make_response(jsonify({"success": False, "error": msg}), code)) @bp.route("/oauth/token/", methods=["POST"]) @csrf.exempt def oauth_grant(): form = request.form grant_type = request.args.get("grant_type", "authorization_code") if grant_type != "authorization_code": error(400, "Unsupported grant_type, only authorization_code is supported") client_id = form.get("client_id", "") if client_id == "": error(400, "Missing client_id") client_secret = form.get("client_secret", "") if client_secret == "": error(400, "Missing client_secret") code = form.get("code", "") if code == "": error(400, "Missing code") client = OAuthClient.query.filter_by(id=client_id, secret=client_secret).first() if client is None: error(400, "client_id and/or client_secret is incorrect") token = APIToken.query.filter_by(auth_code=code).first() if token is None or token.client != client: error(400, "Incorrect code. It may have already been redeemed") token.auth_code = None db.session.commit() return jsonify({ "success": True, "access_token": token.access_token, "token_type": "Bearer", }) @bp.route("/user/apps/") @login_required def list_clients_redirect(): return redirect(url_for("oauth.list_clients", username=current_user.username)) @bp.route("/users//apps/") @login_required def list_clients(username): user = User.query.filter_by(username=username).first_or_404() if not user.check_perm(current_user, Permission.CREATE_OAUTH_CLIENT): abort(403) return render_template("oauth/list_clients.html", user=user, tabs=get_setting_tabs(user), current_tab="oauth_clients") class OAuthClientForm(FlaskForm): title = StringField(lazy_gettext("Title"), [InputRequired(), Length(5, 30)]) description = StringField(lazy_gettext("Description"), [Optional()]) redirect_url = URLField(lazy_gettext("Redirect URL"), [InputRequired(), Length(5, 123)]) app_type = SelectField(lazy_gettext("App Type"), [InputRequired()], choices=[ ("server", "Server-side (client_secret is kept safe)"), ("client", "Client-side (client_secret is visible to all users)"), ], coerce=lambda x: x) submit = SubmitField(lazy_gettext("Save")) @bp.route("/users//apps/new/", methods=["GET", "POST"]) @bp.route("/users//apps//edit/", methods=["GET", "POST"]) @login_required def create_edit_client(username, id_=None): user = User.query.filter_by(username=username).first_or_404() if not user.check_perm(current_user, Permission.CREATE_OAUTH_CLIENT): abort(403) is_new = id_ is None client = None if id_ is not None: client = OAuthClient.query.get_or_404(id_) if client.owner != user: abort(404) form = OAuthClientForm(formdata=request.form, obj=client) if form.validate_on_submit(): if is_new: if OAuthClient.query.filter(OAuthClient.title.ilike(form.title.data.strip())).count() > 0: flash(gettext("An OAuth client with that title already exists. Please choose a new title."), "danger") return render_template("oauth/create_edit.html", user=user, form=form, client=client) client = OAuthClient() db.session.add(client) client.owner = user client.id = random_string(24) client.secret = random_string(32) client.approved = current_user.rank.at_least(UserRank.EDITOR) form.populate_obj(client) verb = "Created" if is_new else "Edited" add_audit_log(AuditSeverity.NORMAL, current_user, f"{verb} OAuth2 application {client.title} by {client.owner.username} [{client.id}]", url_for("oauth.create_edit_client", username=client.owner.username, id_=client.id)) db.session.commit() return redirect(url_for("oauth.create_edit_client", username=username, id_=client.id)) return render_template("oauth/create_edit.html", user=user, form=form, client=client) @bp.route("/users//apps//delete/", methods=["POST"]) @login_required def delete_client(username, id_): user = User.query.filter_by(username=username).first_or_404() if not user.check_perm(current_user, Permission.CREATE_OAUTH_CLIENT): abort(403) client = OAuthClient.query.get(id_) if client is None or client.owner != user: abort(404) add_audit_log(AuditSeverity.NORMAL, current_user, f"Deleted OAuth2 application {client.title} by {client.owner.username} [{client.id}]", url_for("users.profile", username=current_user.username)) db.session.delete(client) db.session.commit() return redirect(url_for("oauth.list_clients", username=username)) @bp.route("/users//apps//revoke-all/", methods=["POST"]) @login_required def revoke_all(username, id_): user = User.query.filter_by(username=username).first_or_404() if not user.check_perm(current_user, Permission.CREATE_OAUTH_CLIENT): abort(403) client = OAuthClient.query.get(id_) if client is None or client.owner != user: abort(404) add_audit_log(AuditSeverity.NORMAL, current_user, f"Revoked all user tokens for OAuth2 application {client.title} by {client.owner.username} [{client.id}]", url_for("oauth.create_edit_client", username=client.owner.username, id_=client.id)) client.tokens = [] db.session.commit() flash(gettext("Revoked all user tokens"), "success") return redirect(url_for("oauth.create_edit_client", username=client.owner.username, id_=client.id))