# Copyright 2022 Karlsruhe Institute of Technology
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from flask_babel import gettext as _
from flask_babel import lazy_gettext as _l
from flask_babel import ngettext
from flask_login import current_user
from marshmallow import ValidationError
from wtforms.validators import DataRequired
from wtforms.validators import Length
from wtforms.validators import StopValidation
import kadi.lib.constants as const
from kadi.lib.conversion import empty_str
from kadi.lib.conversion import lower
from kadi.lib.conversion import normalize
from kadi.lib.conversion import strip
from kadi.lib.forms import BaseForm
from kadi.lib.forms import DynamicMultiSelectField
from kadi.lib.forms import JSONField
from kadi.lib.forms import LFTextAreaField
from kadi.lib.forms import SelectField
from kadi.lib.forms import StringField
from kadi.lib.forms import validate_identifier
from kadi.lib.permissions.core import has_permission
from kadi.lib.permissions.utils import get_group_roles
from kadi.lib.permissions.utils import get_user_roles
from kadi.modules.accounts.models import User
from kadi.modules.groups.models import Group
from .schemas import ResourceRoleDataSchema
[docs]
class RolesField(JSONField):
"""Custom field to process and validate user and group roles of resources.
Uses :class:`.ResourceRoleDataSchema` for its validation.
:param roles: A list of roles, each item consisting of a tuple containing the actual
role value and title to be displayed, similar to the choices in select fields.
:param allow_none: (optional) Whether to allow ``None`` as a valid role value.
"""
def __init__(self, *args, roles, allow_none=False, **kwargs):
kwargs["default"] = []
super().__init__(*args, **kwargs)
self.roles = roles
self.allow_none = allow_none
self.initial = []
def _value(self):
return self.initial
[docs]
def to_dict(self):
data = super().to_dict()
data["roles"] = [(val, str(title)) for val, title in self.roles]
data["allow_none"] = self.allow_none
return data
[docs]
def set_initial_data(
self, data=None, resource=None, user=None, keep_user_roles=False
):
"""Set the initial data of this field.
:param data: (optional) The form data to use for prefilling. Defaults to the
submitted data of the current field instance.
:param resource: (optional) An existing resource, which can be used to set the
initial data instead of the given form data. One of :class:`.Record`,
:class:`.Collection` or :class:`.Template`.
:param user: (optional) A user that will be used for checking various access
permissions when setting the data. Defaults to the current user.
:param keep_user_roles: (optional) Flag indicating whether to keep any roles of
the given user.
"""
data = data if data is not None else getattr(self, "data", [])
user = user if user is not None else current_user
initial_data = []
if resource is not None:
for _user, role in get_user_roles(
resource.__tablename__, object_id=resource.id
):
if keep_user_roles or _user != user:
initial_data.append(
{
"subject_type": "user",
"subject": [_user.id, f"@{_user.identity.username}"],
"role": role.name,
}
)
for group, role in get_group_roles(
resource.__tablename__, object_id=resource.id
):
# Exclude any group that is not readable by the given user.
if group is not None and has_permission(
user, "read", "group", group.id
):
initial_data.append(
{
"subject_type": "group",
"subject": [group.id, f"@{group.identifier}"],
"role": role.name,
}
)
else:
for role_meta in data:
subject = None
if role_meta["subject_type"] == "user":
_user = User.query.get_active(role_meta["subject_id"])
if _user is not None and (keep_user_roles or _user != user):
subject = [_user.id, f"@{_user.identity.username}"]
else:
group = Group.query.get_active(role_meta["subject_id"])
# Exclude any group that is not readable by the given user.
if group is not None and has_permission(
user, "read", "group", group.id
):
subject = [group.id, f"@{group.identifier}"]
if subject is not None:
initial_data.append(
{
"subject_type": role_meta["subject_type"],
"subject": subject,
"role": role_meta["role"],
}
)
self.initial = initial_data
[docs]
def check_duplicate_identifier(model, identifier, exclude=None):
"""Check for a duplicate identifier in a form.
:param model: The model class to check the identifier of. One of :class:`.Record`,
:class:`.Collection`, :class:`.Template` or :class:`.Group`.
:param identifier: The identifier to check.
:param exclude: (optional) An instance of the model that should be excluded in the
check.
"""
obj_to_check = model.query.filter_by(identifier=identifier).first()
if obj_to_check is not None and (exclude is None or exclude != obj_to_check):
raise StopValidation(_("Identifier is already in use."))