255 lines
8.3 KiB
Python
255 lines
8.3 KiB
Python
"""Users plugin
|
|
|
|
Provides routes used to manage users
|
|
"""
|
|
from http.client import NO_CONTENT, CREATED
|
|
from flask import Blueprint, request, jsonify, make_response
|
|
from werkzeug.exceptions import BadRequest, Forbidden, MethodNotAllowed, NotFound
|
|
|
|
from . import permissions
|
|
from flaschengeist import logger
|
|
from flaschengeist.config import config
|
|
from flaschengeist.plugins import Plugin
|
|
from flaschengeist.models.user import User
|
|
from flaschengeist.utils.decorators import login_required, extract_session, headers
|
|
from flaschengeist.controller import userController
|
|
from flaschengeist.utils.HTTP import created, no_content
|
|
from flaschengeist.utils.datetime import from_iso_format
|
|
|
|
|
|
class UsersPlugin(Plugin):
|
|
blueprint = Blueprint("users", __name__)
|
|
permissions = permissions.permissions
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users", methods=["POST"])
|
|
def register():
|
|
"""Register a new user
|
|
The password will be set to a random string of at lease 16byte entropy.
|
|
The user will receive a mail containing a link to set their own password.
|
|
|
|
Route: ``/users`` | Method: ``POST``
|
|
|
|
POST-data: Same as `flaschengeist.models.user.User`
|
|
|
|
Returns:
|
|
JSON encoded `flaschengeist.models.user.User` or HTTP error
|
|
"""
|
|
registration = config["users"].get("registration", False)
|
|
if not registration or registration not in ["managed", "public"]:
|
|
logger.debug("Config for Registration is set to >{}<".format(registration))
|
|
raise MethodNotAllowed
|
|
if registration == "managed":
|
|
extract_session(permissions.REGISTER)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
raise BadRequest
|
|
for required in ["firstname", "lastname", "mail"]:
|
|
if required not in data:
|
|
raise BadRequest("Missing required parameters")
|
|
|
|
logger.debug("Register new User...")
|
|
return make_response(jsonify(userController.register(data)), CREATED)
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users", methods=["GET"])
|
|
@login_required()
|
|
@headers({"Cache-Control": "private, must-revalidate, max-age=3600"})
|
|
def list_users(current_session):
|
|
"""List all existing users
|
|
|
|
Route: ``/users`` | Method: ``GET``
|
|
|
|
Args:
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON encoded array of `flaschengeist.models.user.User` or HTTP error
|
|
"""
|
|
logger.debug("Retrieve list of all users")
|
|
users = userController.get_users()
|
|
return jsonify(users)
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>", methods=["GET"])
|
|
@login_required()
|
|
@headers({"Cache-Control": "private, must-revalidate, max-age=300"})
|
|
def get_user(userid, current_session):
|
|
"""Retrieve user by userid
|
|
|
|
Route: ``/users/<userid>`` | Method: ``GET``
|
|
|
|
Args:
|
|
userid: UserID of user to retrieve
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON encoded `flaschengeist.models.user.User` or if userid is current user also containing permissions or HTTP error
|
|
"""
|
|
logger.debug("Get information of user {{ {} }}".format(userid))
|
|
user: User = userController.get_user(
|
|
userid, True
|
|
) # This is the only API point that should return data for deleted users
|
|
serial = user.serialize()
|
|
if userid == current_session.user_.userid:
|
|
serial["permissions"] = user.get_permissions()
|
|
return jsonify(serial)
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>/frontend", methods=["POST", "GET"])
|
|
@login_required()
|
|
def frontend(userid, current_session):
|
|
if current_session.user_.userid != userid:
|
|
raise Forbidden
|
|
|
|
if request.method == "POST":
|
|
if request.content_length > 1024**2:
|
|
raise BadRequest
|
|
current_session.user_.set_attribute("frontend", request.get_json())
|
|
return no_content()
|
|
else:
|
|
content = current_session.user_.get_attribute("frontend", None)
|
|
if content is None:
|
|
return no_content()
|
|
return jsonify(content)
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>/avatar", methods=["GET"])
|
|
@headers({"Cache-Control": "public, max-age=604800"})
|
|
def get_avatar(userid):
|
|
user = userController.get_user(userid)
|
|
return userController.load_avatar(user)
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>/avatar", methods=["POST"])
|
|
@login_required()
|
|
def set_avatar(userid, current_session):
|
|
user = userController.get_user(userid)
|
|
if userid != current_session.user_.userid and not current_session.user_.has_permission(permissions.EDIT):
|
|
raise Forbidden
|
|
|
|
file = request.files.get("file")
|
|
if file:
|
|
userController.save_avatar(user, file)
|
|
return created()
|
|
else:
|
|
raise BadRequest
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>/avatar", methods=["DELETE"])
|
|
@login_required()
|
|
def delete_avatar(userid, current_session):
|
|
user = userController.get_user(userid)
|
|
if userid != current_session.user_.userid and not current_session.user_.has_permission(permissions.EDIT):
|
|
raise Forbidden
|
|
userController.delete_avatar(user)
|
|
return no_content()
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>", methods=["DELETE"])
|
|
@login_required(permission=permissions.DELETE)
|
|
def delete_user(userid, current_session):
|
|
"""Delete user by userid
|
|
|
|
Route: ``/users/<userid>`` | Method: ``DELETE``
|
|
|
|
Args:
|
|
userid: UserID of user to retrieve
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
HTTP-204 or HTTP error
|
|
"""
|
|
logger.debug("Delete user {{ {} }}".format(userid))
|
|
user = userController.get_user(userid)
|
|
userController.delete_user(user)
|
|
return no_content()
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>", methods=["PUT"])
|
|
@login_required()
|
|
def edit_user(userid, current_session):
|
|
"""Modify user by userid
|
|
|
|
Route: ``/users/<userid>`` | Method: ``PUT``
|
|
|
|
POST-data: ```{firstname?: string, lastname?: string, display_name?: string, mail?: string,
|
|
password?: string, roles?: string[]}```
|
|
|
|
Args:
|
|
userid: UserID of user to retrieve
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
HTTP-204 or HTTP error
|
|
"""
|
|
logger.debug("Modify information of user {{ {} }}".format(userid))
|
|
user = userController.get_user(userid)
|
|
data = request.get_json()
|
|
|
|
password = None
|
|
new_password = data["new_password"] if "new_password" in data else None
|
|
|
|
author = user
|
|
if userid != current_session.user_.userid:
|
|
author = current_session.user_
|
|
if not author.has_permission(permissions.EDIT):
|
|
raise Forbidden
|
|
else:
|
|
if "password" not in data:
|
|
raise BadRequest("Password is missing")
|
|
password = data["password"]
|
|
|
|
for key in ["firstname", "lastname", "display_name", "mail"]:
|
|
if key in data:
|
|
setattr(user, key, data[key])
|
|
if "birthday" in data:
|
|
user.birthday = from_iso_format(data["birthday"])
|
|
|
|
if "roles" in data:
|
|
roles = set(data["roles"])
|
|
if not author.has_permission(permissions.SET_ROLES):
|
|
if len(roles) != len(user.roles) or set(user.roles) != roles:
|
|
raise Forbidden
|
|
else:
|
|
userController.set_roles(user, roles)
|
|
|
|
userController.modify_user(user, password, new_password)
|
|
userController.update_user(user)
|
|
return no_content()
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/notifications", methods=["GET"])
|
|
@login_required()
|
|
def notifications(current_session):
|
|
f = request.args.get("from", None)
|
|
if f is not None:
|
|
f = from_iso_format(f)
|
|
return jsonify(userController.get_notifications(current_session.user_, f))
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/notifications/<nid>", methods=["DELETE"])
|
|
@login_required()
|
|
def remove_notification(nid, current_session):
|
|
userController.delete_notification(nid, current_session.user_)
|
|
return no_content()
|
|
|
|
|
|
@UsersPlugin.blueprint.route("/users/<userid>/shortcuts", methods=["GET", "PUT"])
|
|
@login_required()
|
|
def shortcuts(userid, current_session):
|
|
if userid != current_session.user_.userid:
|
|
raise Forbidden
|
|
|
|
user = userController.get_user(userid)
|
|
if request.method == "GET":
|
|
return jsonify(user.get_attribute("users_link_shortcuts", []))
|
|
else:
|
|
data = request.get_json()
|
|
if not isinstance(data, list) or not all(isinstance(n, dict) for n in data):
|
|
raise BadRequest
|
|
user.set_attribute("users_link_shortcuts", data)
|
|
userController.persist()
|
|
return no_content()
|