uffd/commands/cleanup.py

+import click
+from flask.cli import with_appcontext
+
+from uffd.tasks import cleanup_task
+from uffd.database import db
+
+@click.command('cleanup', help='Cleanup expired data')
+@with_appcontext
+def cleanup_command():
+	cleanup_task.run()
+	db.session.commit()

uffd/commands/gendevcert.py

+import os
+
+import click
+from werkzeug.serving import make_ssl_devcert
+
+@click.command("gendevcert", help='Generates a self-signed TLS certificate for development')
+def gendevcert_command(): #pylint: disable=unused-variable
+	if os.path.exists('devcert.crt') or os.path.exists('devcert.key'):
+		print('Refusing to overwrite existing "devcert.crt"/"devcert.key" file!')
+		return
+	make_ssl_devcert('devcert')
+	print('Certificate written to "devcert.crt", private key to "devcert.key".')
+	print('Run `flask run --cert devcert.crt --key devcert.key` to use it.')

uffd/commands/group.py

+from flask import current_app
+from flask.cli import AppGroup
+from sqlalchemy.exc import IntegrityError
+import click
+
+from uffd.database import db
+
+from uffd.models import Group
+
+group_command = AppGroup('group', help='Manage groups')
+
+@group_command.command(help='List names of all groups')
+def list():
+	with current_app.test_request_context():
+		for group in Group.query:
+			click.echo(group.name)
+
+@group_command.command(help='Show details of group')
+@click.argument('name')
+def show(name):
+	with current_app.test_request_context():
+		group = Group.query.filter_by(name=name).one_or_none()
+		if group is None:
+			raise click.ClickException(f'Group {name} not found')
+		click.echo(f'Name: {group.name}')
+		click.echo(f'Unix GID: {group.unix_gid}')
+		click.echo(f'Description: {group.description}')
+		click.echo(f'Members: {", ".join([user.loginname for user in group.members])}')
+
+@group_command.command(help='Create new group')
+@click.argument('name')
+@click.option('--description', default='', help='Set description text. Empty per default.')
+def create(name, description):
+	with current_app.test_request_context():
+		group = Group(description=description)
+		if not group.set_name(name):
+			raise click.ClickException('Invalid name')
+		try:
+			db.session.add(group)
+			db.session.commit()
+		except IntegrityError:
+			# pylint: disable=raise-missing-from
+			raise click.ClickException(f'A group with name "{name}" already exists')
+
+@group_command.command(help='Update group attributes')
+@click.argument('name')
+@click.option('--description', help='Set description text.')
+def update(name, description):
+	with current_app.test_request_context():
+		group = Group.query.filter_by(name=name).one_or_none()
+		if group is None:
+			raise click.ClickException(f'Group {name} not found')
+		if description is not None:
+			group.description = description
+		db.session.commit()
+
+@group_command.command(help='Delete group')
+@click.argument('name')
+def delete(name):
+	with current_app.test_request_context():
+		group = Group.query.filter_by(name=name).one_or_none()
+		if group is None:
+			raise click.ClickException(f'Group {name} not found')
+		db.session.delete(group)
+		db.session.commit()

uffd/commands/profile.py

+import os
+
+from flask import current_app
+from flask.cli import with_appcontext
+import click
+try:
+	from werkzeug.middleware.profiler import ProfilerMiddleware
+except ImportError:
+	from werkzeug.contrib.profiler import ProfilerMiddleware
+
+@click.command("profile", help='Runs app with profiler')
+@with_appcontext
+def profile_command(): #pylint: disable=unused-variable
+	# app.run() is silently ignored if executed from commands. We really want
+	# to do this, so we overwrite the check by overwriting the environment
+	# variable.
+	os.environ['FLASK_RUN_FROM_CLI'] = 'false'
+	current_app.wsgi_app = ProfilerMiddleware(current_app.wsgi_app, restrictions=[30])
+	current_app.run(debug=True)

uffd/commands/role.py

+from flask import current_app
+from flask.cli import AppGroup
+from sqlalchemy.exc import IntegrityError
+import click
+
+from uffd.database import db
+from uffd.models import Group, Role, RoleGroup
+
+role_command = AppGroup('role', help='Manage roles')
+
+# pylint: disable=too-many-arguments,too-many-locals
+
+def update_attrs(role, description=None, default=None,
+                 moderator_group=None, clear_moderator_group=False,
+                 clear_groups=False, add_group=tuple(), remove_group=tuple(),
+                 clear_roles=False, add_role=tuple(), remove_role=tuple()):
+	if description is not None:
+		role.description = description
+	if default is not None:
+		role.is_default = default
+	if clear_moderator_group:
+		role.moderator_group = None
+	elif moderator_group is not None:
+		group = Group.query.filter_by(name=moderator_group).one_or_none()
+		if group is None:
+			raise click.ClickException(f'Moderaor group {moderator_group} not found')
+		role.moderator_group = group
+	if clear_groups:
+		role.groups.clear()
+	for group_name in add_group:
+		group = Group.query.filter_by(name=group_name).one_or_none()
+		if group is None:
+			raise click.ClickException(f'Group {group_name} not found')
+		role.groups[group] = RoleGroup(group=group)
+	for group_name in remove_group:
+		group = Group.query.filter_by(name=group_name).one_or_none()
+		if group is None:
+			raise click.ClickException(f'Group {group_name} not found')
+		del role.groups[group]
+	if clear_roles:
+		role.included_roles.clear()
+	for role_name in add_role:
+		_role = Role.query.filter_by(name=role_name).one_or_none()
+		if _role is None:
+			raise click.ClickException(f'Role {role_name} not found')
+		role.included_roles.append(_role)
+	for role_name in remove_role:
+		_role = Role.query.filter_by(name=role_name).one_or_none()
+		if _role is None:
+			raise click.ClickException(f'Role {role_name} not found')
+		role.included_roles.remove(_role)
+
+@role_command.command(help='List names of all roles')
+def list():
+	with current_app.test_request_context():
+		for role in Role.query:
+			click.echo(role.name)
+
+@role_command.command(help='Show details of group')
+@click.argument('name')
+def show(name):
+	with current_app.test_request_context():
+		role = Role.query.filter_by(name=name).one_or_none()
+		if role is None:
+			raise click.ClickException(f'Role {name} not found')
+		click.echo(f'Name: {role.name}')
+		click.echo(f'Description: {role.description}')
+		click.echo(f'Default: {role.is_default}')
+		click.echo(f'Moderator group: {role.moderator_group.name if role.moderator_group else None}')
+		click.echo(f'Direct groups: {", ".join(sorted([group.name for group in role.groups]))}')
+		click.echo(f'Effective groups: {", ".join(sorted([group.name for group in role.groups_effective]))}')
+		click.echo(f'Included roles: {", ".join(sorted([irole.name for irole in role.included_roles]))}')
+		click.echo(f'Direct members: {", ".join(sorted([user.loginname for user in role.members]))}')
+		click.echo(f'Effective members: {", ".join(sorted([user.loginname for user in role.members_effective]))}')
+
+@role_command.command(help='Create new role')
+@click.argument('name')
+@click.option('--description', default='', help='Set description text.')
+@click.option('--default/--no-default', default=False, help='Mark role as default or not. Non-service users are auto-added to default roles.')
+@click.option('--moderator-group', metavar='GROUP_NAME', help='Set moderator group. No moderator group if unset.')
+@click.option('--add-group', multiple=True, metavar='GROUP_NAME', help='Add group granted to role members. Repeat to add multiple groups.')
+@click.option('--add-role', multiple=True, metavar='ROLE_NAME', help='Add role to inherit groups from. Repeat to add multiple roles.')
+def create(name, description, default, moderator_group, add_group, add_role):
+	with current_app.test_request_context():
+		try:
+			role = Role(name=name)
+			update_attrs(role, description, default, moderator_group,
+			             add_group=add_group, add_role=add_role)
+			db.session.add(role)
+			role.update_member_groups()
+			db.session.commit()
+		except IntegrityError:
+			# pylint: disable=raise-missing-from
+			raise click.ClickException(f'A role with name "{name}" already exists')
+
+@role_command.command(help='Update role attributes')
+@click.argument('name')
+@click.option('--description', help='Set description text.')
+@click.option('--default/--no-default', default=None, help='Mark role as default or not. Non-service users are auto-added to default roles.')
+@click.option('--moderator-group', metavar='GROUP_NAME', help='Set moderator group.')
+@click.option('--no-moderator-group', is_flag=True, flag_value=True, default=False, help='Clear moderator group setting.')
+@click.option('--clear-groups', is_flag=True, flag_value=True, default=False, help='Remove all groups granted to role members. Executed before --add-group.')
+@click.option('--add-group', multiple=True, metavar='GROUP_NAME', help='Add group granted to role members. Repeat to add multiple groups.')
+@click.option('--remove-group', multiple=True, metavar='GROUP_NAME', help='Remove group granted to role members. Repeat to remove multiple groups.')
+@click.option('--clear-roles', is_flag=True, flag_value=True, default=False, help='Remove all included roles. Executed before --add-role.')
+@click.option('--add-role', multiple=True, metavar='ROLE_NAME', help='Add role to inherit groups from. Repeat to add multiple roles.')
+@click.option('--remove-role', multiple=True, metavar='ROLE_NAME', help='Remove included role. Repeat to remove multiple roles.')
+def update(name, description, default, moderator_group, no_moderator_group,
+           clear_groups, add_group, remove_group, clear_roles, add_role, remove_role):
+	with current_app.test_request_context():
+		role = Role.query.filter_by(name=name).one_or_none()
+		if role is None:
+			raise click.ClickException(f'Role {name} not found')
+		old_members = set(role.members_effective)
+		update_attrs(role, description, default, moderator_group,
+		             no_moderator_group, clear_groups, add_group, remove_group,
+		             clear_roles, add_role, remove_role)
+		for user in old_members:
+			user.update_groups()
+		role.update_member_groups()
+		db.session.commit()
+
+@role_command.command(help='Delete role')
+@click.argument('name')
+def delete(name):
+	with current_app.test_request_context():
+		role = Role.query.filter_by(name=name).one_or_none()
+		if role is None:
+			raise click.ClickException(f'Role {name} not found')
+		old_members = set(role.members_effective)
+		db.session.delete(role)
+		for user in old_members:
+			user.update_groups()
+		db.session.commit()

uffd/commands/roles_update_all.py

+import sys
+
+from flask import current_app
+from flask.cli import with_appcontext
+import click
+
+from uffd.database import db
+from uffd.models import User
+
+@click.command('roles-update-all', help='Update group memberships for all users based on their roles')
+@click.option('--check-only', is_flag=True)
+@with_appcontext
+def roles_update_all_command(check_only): #pylint: disable=unused-variable
+	consistent = True
+	with current_app.test_request_context():
+		for user in User.query.all():
+			groups_added, groups_removed = user.update_groups()
+			if groups_added:
+				consistent = False
+				print('Adding groups [%s] to user %s'%(', '.join([group.name for group in groups_added]), user.loginname))
+			if groups_removed:
+				consistent = False
+				print('Removing groups [%s] from user %s'%(', '.join([group.name for group in groups_removed]), user.loginname))
+		if not check_only:
+			db.session.commit()
+		if check_only and not consistent:
+			print('No changes were made because --check-only is set')
+			print()
+			print('Error: Groups are not consistent with roles in database')
+			sys.exit(1)

uffd/commands/unique_email_addresses.py

+import click
+from flask.cli import with_appcontext
+from sqlalchemy.exc import IntegrityError
+
+from uffd.database import db
+from uffd.models import User, UserEmail, FeatureFlag
+
+# pylint completely fails to understand SQLAlchemy's query functions
+# pylint: disable=no-member
+
+@click.group('unique-email-addresses', help='Enable/disable e-mail address uniqueness checks')
+def unique_email_addresses_command():
+	pass
+
+@unique_email_addresses_command.command('enable')
+@with_appcontext
+def enable_unique_email_addresses_command():
+	if FeatureFlag.unique_email_addresses:
+		raise click.ClickException('Uniqueness checks for e-mail addresses are already enabled')
+	query = db.select([UserEmail.address_normalized, UserEmail.user_id])\
+		.group_by(UserEmail.address_normalized, UserEmail.user_id)\
+		.having(db.func.count(UserEmail.id.distinct()) > 1)
+	for address_normalized, user_id in db.session.execute(query).fetchall():
+		user = User.query.get(user_id)
+		user_emails = UserEmail.query.filter_by(address_normalized=address_normalized, user_id=user_id)
+		click.echo(f'User "{user.loginname}" has the same e-mail address multiple times:', err=True)
+		for user_email in user_emails:
+			if user_email.verified:
+				click.echo(f'- {user_email.address}', err=True)
+			else:
+				click.echo(f'- {user_email.address} (unverified)', err=True)
+		click.echo()
+	query = db.select([UserEmail.address_normalized, UserEmail.address])\
+		.where(UserEmail.verified)\
+		.group_by(UserEmail.address_normalized)\
+		.having(db.func.count(UserEmail.id.distinct()) > 1)
+	for address_normalized, address in db.session.execute(query).fetchall():
+		click.echo(f'E-mail address "{address}" is used by multiple users:', err=True)
+		user_emails = UserEmail.query.filter_by(address_normalized=address_normalized, verified=True)
+		for user_email in user_emails:
+			if user_email.address != address:
+				click.echo(f'- {user_email.user.loginname} ({user_email.address})', err=True)
+			else:
+				click.echo(f'- {user_email.user.loginname}', err=True)
+		click.echo()
+	try:
+		FeatureFlag.unique_email_addresses.enable()
+	except IntegrityError:
+		# pylint: disable=raise-missing-from
+		raise click.ClickException('''Some existing e-mail addresses violate uniqueness checks
+
+You need to fix this manually in the admin interface. Then run this command
+again to continue.''')
+	db.session.commit()
+	click.echo('Uniqueness checks for e-mail addresses enabled')
+
+@unique_email_addresses_command.command('disable')
+@with_appcontext
+def disable_unique_email_addresses_command():
+	if not FeatureFlag.unique_email_addresses:
+		raise click.ClickException('Uniqueness checks for e-mail addresses are already disabled')
+	click.echo('''Please note that the option to disable email address uniqueness checks will
+be remove in uffd v3.
+''', err=True)
+	FeatureFlag.unique_email_addresses.disable()
+	db.session.commit()
+	click.echo('Uniqueness checks for e-mail addresses disabled')

uffd/commands/user.py

+from flask import current_app
+from flask.cli import AppGroup
+from sqlalchemy.exc import IntegrityError
+import click
+
+from uffd.database import db
+from uffd.models import User, Role
+
+user_command = AppGroup('user', help='Manage users')
+
+# pylint: disable=too-many-arguments
+
+def update_attrs(user, mail=None, displayname=None, password=None,
+                 prompt_password=False, clear_roles=False,
+                 add_role=tuple(), remove_role=tuple(), deactivate=None):
+	if password is None and prompt_password:
+		password = click.prompt('Password', hide_input=True, confirmation_prompt='Confirm password')
+	if mail is not None and not user.set_primary_email_address(mail):
+		raise click.ClickException('Invalid mail address')
+	if displayname is not None and not user.set_displayname(displayname):
+		raise click.ClickException('Invalid displayname')
+	if password is not None and not user.set_password(password):
+		raise click.ClickException('Invalid password')
+	if deactivate is not None:
+		user.is_deactivated = deactivate
+	if clear_roles:
+		user.roles.clear()
+	for role_name in add_role:
+		role = Role.query.filter_by(name=role_name).one_or_none()
+		if role is None:
+			raise click.ClickException(f'Role {role_name} not found')
+		role.members.append(user)
+	for role_name in remove_role:
+		role = Role.query.filter_by(name=role_name).one_or_none()
+		if role is None:
+			raise click.ClickException(f'Role {role_name} not found')
+		role.members.remove(user)
+	user.update_groups()
+
+@user_command.command(help='List login names of all users')
+def list():
+	with current_app.test_request_context():
+		for user in User.query:
+			click.echo(user.loginname)
+
+@user_command.command(help='Show details of user')
+@click.argument('loginname')
+def show(loginname):
+	with current_app.test_request_context():
+		user = User.query.filter_by(loginname=loginname).one_or_none()
+		if user is None:
+			raise click.ClickException(f'User {loginname} not found')
+		click.echo(f'Loginname: {user.loginname}')
+		click.echo(f'Deactivated: {user.is_deactivated}')
+		click.echo(f'Displayname: {user.displayname}')
+		click.echo(f'Mail: {user.primary_email.address}')
+		click.echo(f'Service User: {user.is_service_user}')
+		click.echo(f'Roles: {", ".join([role.name for role in user.roles])}')
+		click.echo(f'Groups: {", ".join([group.name for group in user.groups])}')
+
+@user_command.command(help='Create new user')
+@click.argument('loginname')
+@click.option('--mail', required=True, metavar='EMAIL_ADDRESS', help='E-Mail address')
+@click.option('--displayname', help='Set display name. Defaults to login name.')
+@click.option('--service/--no-service', default=False, help='Create service or regular (default) user. '+\
+                                                            'Regular users automatically have roles marked as default. '+\
+                                                            'Service users do not.')
+@click.option('--password', help='Password for SSO login. Login disabled if unset.')
+@click.option('--prompt-password', is_flag=True, flag_value=True, default=False, help='Read password interactively from terminal.')
+@click.option('--add-role', multiple=True, help='Add role to user. Repeat to add multiple roles.', metavar='ROLE_NAME')
+@click.option('--deactivate', is_flag=True, flag_value=True, default=None, help='Deactivate account.')
+def create(loginname, mail, displayname, service, password, prompt_password, add_role, deactivate):
+	with current_app.test_request_context():
+		if displayname is None:
+			displayname = loginname
+		user = User(is_service_user=service)
+		if not user.set_loginname(loginname, ignore_blocklist=True):
+			raise click.ClickException('Invalid loginname')
+		try:
+			db.session.add(user)
+			update_attrs(user, mail, displayname, password, prompt_password, add_role=add_role, deactivate=deactivate)
+			db.session.commit()
+		except IntegrityError:
+			# pylint: disable=raise-missing-from
+			raise click.ClickException('Login name or e-mail address is already in use')
+
+@user_command.command(help='Update user attributes and roles')
+@click.argument('loginname')
+@click.option('--mail', metavar='EMAIL_ADDRESS', help='Set e-mail address.')
+@click.option('--displayname', help='Set display name.')
+@click.option('--password', help='Set password for SSO login.')
+@click.option('--prompt-password', is_flag=True, flag_value=True, default=False, help='Set password by reading it interactivly from terminal.')
+@click.option('--clear-roles', is_flag=True, flag_value=True, default=False, help='Remove all roles from user. Executed before --add-role.')
+@click.option('--add-role', multiple=True, help='Add role to user. Repeat to add multiple roles.')
+@click.option('--remove-role', multiple=True, help='Remove role from user. Repeat to remove multiple roles.')
+@click.option('--deactivate/--activate', default=None, help='Deactivate or reactivate account.')
+def update(loginname, mail, displayname, password, prompt_password, clear_roles, add_role, remove_role, deactivate):
+	with current_app.test_request_context():
+		user = User.query.filter_by(loginname=loginname).one_or_none()
+		if user is None:
+			raise click.ClickException(f'User {loginname} not found')
+		try:
+			update_attrs(user, mail, displayname, password, prompt_password, clear_roles, add_role, remove_role, deactivate)
+			db.session.commit()
+		except IntegrityError:
+			# pylint: disable=raise-missing-from
+			raise click.ClickException('E-mail address is already in use')
+
+@user_command.command(help='Delete user')
+@click.argument('loginname')
+def delete(loginname):
+	with current_app.test_request_context():
+		user = User.query.filter_by(loginname=loginname).one_or_none()
+		if user is None:
+			raise click.ClickException(f'User {loginname} not found')
+		db.session.delete(user)
+		db.session.commit()

uffd/csrf/csrf.py → uffd/csrf.py

@@ -4,9 +4,7 @@ from flask import Blueprint, request, session
 
 bp = Blueprint("csrf", __name__)
 
-# pylint: disable=invalid-name
-csrfEndpoints = []
-# pylint: enable=invalid-name
+csrf_endpoints = []
 
 def csrf_protect(blueprint=None, endpoint=None):
 	def wraper(func):
@@ -15,7 +13,7 @@ def csrf_protect(blueprint=None, endpoint=None):
 				urlendpoint = "{}.{}".format(blueprint.name, func.__name__)
 			else:
 				urlendpoint = func.__name__
-		csrfEndpoints.append(urlendpoint)
+		csrf_endpoints.append(urlendpoint)
 		@wraps(func)
 		def decorator(*args, **kwargs):
 			if '_csrf_token' in request.values:
@@ -32,6 +30,6 @@ def csrf_protect(blueprint=None, endpoint=None):
 
 @bp.app_url_defaults
 def csrf_inject(endpoint, values):
-	if endpoint not in csrfEndpoints or not session.get('_csrf_token'):
+	if endpoint not in csrf_endpoints or not session.get('_csrf_token'):
 		return
 	values['_csrf_token'] = session['_csrf_token']

uffd/csrf/__init__.py

-from .csrf import bp as csrf_bp, csrf_protect
-
-bp = [csrf_bp]

uffd/database.py

-from collections import OrderedDict
-
+from sqlalchemy import MetaData, event
+from sqlalchemy.types import TypeDecorator, Text
+from sqlalchemy.ext.mutable import MutableList
 from flask_sqlalchemy import SQLAlchemy
-from flask.json import JSONEncoder
-
-# pylint: disable=C0103
-db = SQLAlchemy()
-# pylint: enable=C0103
-
-class SQLAlchemyJSON(JSONEncoder):
-	def default(self, o):
-		if isinstance(o, db.Model):
-			result = OrderedDict()
-			for key in o.__mapper__.c.keys():
-				result[key] = getattr(o, key)
-			return result
-		return JSONEncoder.default(self, o)
+
+convention = {
+	'ix': 'ix_%(column_0_label)s',
+	'uq': 'uq_%(table_name)s_%(column_0_name)s',
+	'ck': 'ck_%(table_name)s_%(column_0_name)s',
+	'fk': 'fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s',
+	'pk': 'pk_%(table_name)s'
+}
+metadata = MetaData(naming_convention=convention)
+
+db = SQLAlchemy(metadata=metadata)
+
+def enable_sqlite_foreign_key_support(dbapi_connection, connection_record):
+	# pylint: disable=unused-argument
+	cursor = dbapi_connection.cursor()
+	cursor.execute('PRAGMA foreign_keys=ON')
+	cursor.close()
+
+# We want to enable SQLite foreign key support for app and test code, but not
+# for migrations.
+# The common way to add the handler to the Engine class (so it applies to all
+# instances) would also affect the migrations. With flask_sqlalchemy v2.4 and
+# newer we could overwrite SQLAlchemy.create_engine and add our handler there.
+# However Debian Buster and Bullseye ship v2.1, so we do this here and call
+# this function in create_app.
+def customize_db_engine(engine):
+	if engine.name == 'sqlite':
+		event.listen(engine, 'connect', enable_sqlite_foreign_key_support)
+	elif engine.name in ('mysql', 'mariadb'):
+		@event.listens_for(engine, 'connect')
+		def receive_connect(dbapi_connection, connection_record): # pylint: disable=unused-argument
+			cursor = dbapi_connection.cursor()
+			cursor.execute('SHOW VARIABLES LIKE "character_set_connection"')
+			character_set_connection = cursor.fetchone()[1]
+			if character_set_connection != 'utf8mb4':
+				raise Exception(f'Unsupported connection charset "{character_set_connection}". Make sure to add "?charset=utf8mb4" to SQLALCHEMY_DATABASE_URI!')
+			cursor.execute('SHOW VARIABLES LIKE "collation_database"')
+			collation_database  = cursor.fetchone()[1]
+			if collation_database != 'utf8mb4_nopad_bin':
+				raise Exception(f'Unsupported database collation "{collation_database}". Create the database with "CHARACTER SET utf8mb4 COLLATE utf8mb4_nopad_bin"!')
+			cursor.execute('SET NAMES utf8mb4 COLLATE utf8mb4_nopad_bin')
+			cursor.close()
+
+class CommaSeparatedList(TypeDecorator):
+	# For some reason TypeDecorator.process_literal_param and
+	# TypeEngine.python_type are abstract but not actually required
+	# pylint: disable=abstract-method
+
+	impl = Text
+	cache_ok = True
+
+	def process_bind_param(self, value, dialect):
+		if value is None:
+			return None
+		for item in value:
+			if ',' in item:
+				raise ValueError('Items of comma-separated list must not contain commas')
+		return ','.join(value)
+
+	def process_result_value(self, value, dialect):
+		if value is None:
+			return None
+		return MutableList(value.split(','))

uffd/default_config.cfg

-LDAP_USER_SEARCH_BASE="ou=users,dc=example,dc=com"
-LDAP_USER_SEARCH_FILTER=[("objectClass", "person")]
-LDAP_USER_OBJECTCLASSES=["top", "inetOrgPerson", "organizationalPerson", "person", "posixAccount"]
-LDAP_USER_MIN_UID=10000
-LDAP_USER_MAX_UID=18999
-LDAP_USER_GID=20001
-LDAP_USER_DN_ATTRIBUTE="uid"
-LDAP_USER_UID_ATTRIBUTE="uidNumber"
-LDAP_USER_UID_ALIASES=[]
-LDAP_USER_LOGINNAME_ATTRIBUTE="uid"
-LDAP_USER_LOGINNAME_ALIASES=[]
-LDAP_USER_DISPLAYNAME_ATTRIBUTE="cn"
-LDAP_USER_DISPLAYNAME_ALIASES=["givenName", "displayName"]
-LDAP_USER_MAIL_ATTRIBUTE="mail"
-LDAP_USER_MAIL_ALIASES=[]
-LDAP_USER_DEFAULT_ATTRIBUTES={
-	"sn": " ",
-	# All string values are subject to python str.format-style format expansion. To insert literal braces use "{{" and "}}".
-	# Variables: uid, loginname, displayname, mail and possibly other attributes of the User class
-	"homeDirectory": "/home/{loginname}",
-	"gidNumber": LDAP_USER_GID,
-	# "multiValueAttribute": ["value1", "value2"],
-}
+USER_GID=20001
+
+# Service and non-service users must either have the same UID range or must not overlap
+USER_MIN_UID=10000
+USER_MAX_UID=18999
+USER_SERVICE_MIN_UID=19000
+USER_SERVICE_MAX_UID=19999
 
-LDAP_GROUP_SEARCH_BASE="ou=groups,dc=example,dc=com"
-LDAP_GROUP_SEARCH_FILTER=[("objectClass","groupOfUniqueNames")]
-LDAP_GROUP_GID_ATTRIBUTE="gidNumber"
-LDAP_GROUP_NAME_ATTRIBUTE="cn"
-LDAP_GROUP_DESCRIPTION_ATTRIBUTE="description"
-LDAP_GROUP_MEMBER_ATTRIBUTE="uniqueMember"
-
-LDAP_MAIL_SEARCH_BASE="ou=postfix,dc=example,dc=com"
-LDAP_MAIL_SEARCH_FILTER=[("objectClass","postfixVirtual")]
-LDAP_MAIL_OBJECTCLASSES=["top", "postfixVirtual"]
-LDAP_MAIL_DN_ATTRIBUTE="uid"
-LDAP_MAIL_UID_ATTRIBUTE="uid"
-LDAP_MAIL_RECEIVERS_ATTRIBUTE="mailacceptinggeneralid"
-LDAP_MAIL_DESTINATIONS_ATTRIBUTE="maildrop"
-
-LDAP_SERVICE_BIND_DN=""
-LDAP_SERVICE_BIND_PASSWORD=""
-LDAP_SERVICE_URL="ldapi:///"
-LDAP_SERVICE_USE_STARTTLS=True
+GROUP_MIN_GID=20000
+GROUP_MAX_GID=49999
 
+# The period of time that a login lasts for.
 SESSION_LIFETIME_SECONDS=3600
+
+# The period of time that the session cookie lasts for. This is refreshed on each page load.
+PERMANENT_SESSION_LIFETIME=2678400
+
 # CSRF protection
 SESSION_COOKIE_SECURE=True
 SESSION_COOKIE_HTTPONLY=True
 SESSION_COOKIE_SAMESITE='Strict'
 
+LANGUAGES={
+	# Language identifier (see Accept-Language HTTP header) -> Display Name
+	"en": "EN",
+	"de": "DE",
+}
+
 ACL_ADMIN_GROUP="uffd_admin"
+# Group required to access selfservice functions (view selfservice, change profile/password/roles)
 ACL_SELFSERVICE_GROUP="uffd_access"
+# Group required to login
+ACL_ACCESS_GROUP="uffd_access"
+# Members can create invite links for signup
+ACL_SIGNUP_GROUP="uffd_signup"
 
 MAIL_SERVER='' # e.g. example.com
 MAIL_PORT=465
-MAIL_USERNAME='yourId@example.com'
+MAIL_USERNAME='yourId@example.com' # set to empty string to disable authentication
 MAIL_PASSWORD='*****'
 MAIL_USE_STARTTLS=True
 MAIL_FROM_ADDRESS='foo@bar.com'
 
+# Set to a domain name (e.g. "remailer.example.com") to enable remailer.
+# Requires special mail server configuration (see uffd-socketmapd). Can be
+# enabled/disabled per-service in the service settings. If enabled, services
+# no longer get real user mail addresses but instead special autogenerated
+# addresses that are replaced with the real mail addresses by the mail server.
+REMAILER_DOMAIN = ''
+REMAILER_OLD_DOMAINS = []
+# Secret used for construction and verification of remailer addresses.
+# If None, the value of SECRET_KEY is used.
+REMAILER_SECRET_KEY = None
+# Set to list of user loginnames to limit remailer to a small list of users.
+# Useful for debugging. If None remailer is active for all users (if
+# configured and enabled for a service). This option is deprecated. Use the
+# per-service setting in the web interface instead.
+REMAILER_LIMIT_TO_USERS = None
+
 # Do not enable this on a public service! There is no spam protection implemented at the moment.
 SELF_SIGNUP=False
 
+INVITE_MAX_VALID_DAYS=21
+
+LOGINNAME_BLOCKLIST=['^admin$', '^root$']
+
 #MFA_ICON_URL = 'https://example.com/logo.png'
 #MFA_RP_ID = 'example.com' # If unset, hostname from current request is used
 MFA_RP_NAME = 'Uffd Test Service' # Service name passed to U2F/FIDO2 authenticators
 
-ROLES_BASEROLES=['base']
-
 SQLALCHEMY_TRACK_MODIFICATIONS=False
 
 FOOTER_LINKS=[{"url": "https://example.com", "title": "example"}]
 
-OAUTH2_CLIENTS={
-	#'test_client_id' : {'client_secret': 'random_secret', 'redirect_uris': ['https://example.com/oauth']},
-	# You can optionally restrict access to users with a certain group. Set 'required_group' to the name of an LDAP group name or a list of groups.
-	# ... 'required_group': 'test_access_group' ... only allows users with group "test_access_group" access
-	# ... 'required_group': ['groupa', ['groupb', 'groupc']] ... allows users with group "groupa" as well as users with both "groupb" and "groupc" access
-}
+# The default page after login or clicking the top left home button is the self-service
+# page. If you would like it to be the services list instead, set this to True.
+DEFAULT_PAGE_SERVICES=False
 
 # Service overview page (disabled if empty)
 SERVICES=[
@@ -132,14 +130,24 @@ SERVICES_BANNER_PUBLIC=True
 # Enable the service overview page for users who are not logged in
 SERVICES_PUBLIC=True
 
+# An optional banner that will be displayed above the login form
+#LOGIN_BANNER='Always check the URL. Never enter your SSO password on any other site.'
+
 BRANDING_LOGO_URL='/static/empty.png'
+SITE_TITLE='uffd'
+
+# Name and contact mail address are displayed to users in a few places (plain text only!)
+ORGANISATION_NAME='Example Organisation'
+ORGANISATION_CONTACT='contact@example.com'
+
+# Optional text included in account registration mails (plain text only!)
+WELCOME_TEXT='See https://docs.example.com/ for further information.'
 
 # do NOT set in production
 
 #TEMPLATES_AUTO_RELOAD=True
 #SQLALCHEMY_ECHO=True
 #FLASK_ENV=development
-#LDAP_SERVICE_MOCK=True
 
 # DO set in production
 

uffd/fido2_compat.py

+# pylint: skip-file
+
+from flask_babel import gettext as _
+from warnings import warn
+from flask import request, current_app
+import urllib.parse
+
+
+# WebAuthn support is optional because fido2 has a pretty unstable
+# interface and might be difficult to install with the correct version
+
+try:
+	import fido2 as __fido2
+
+	if __fido2.__version__.startswith('0.5.'):
+		from fido2.client import ClientData
+		from fido2.server import Fido2Server, RelyingParty as __PublicKeyCredentialRpEntity
+		from fido2.ctap2 import AttestationObject, AuthenticatorData, AttestedCredentialData
+		from fido2 import cbor
+		cbor.encode = cbor.dumps
+		cbor.decode = lambda arg: cbor.loads(arg)[0]
+		class PublicKeyCredentialRpEntity(__PublicKeyCredentialRpEntity):
+			def __init__(self, name, id):
+				super().__init__(id, name)
+	elif __fido2.__version__.startswith('0.9.'):
+		from fido2.client import ClientData
+		from fido2.webauthn import PublicKeyCredentialRpEntity
+		from fido2.server import Fido2Server
+		from fido2.ctap2 import AttestationObject, AuthenticatorData, AttestedCredentialData
+		from fido2 import cbor
+	elif __fido2.__version__.startswith('1.'):
+		from fido2.webauthn import PublicKeyCredentialRpEntity, CollectedClientData as ClientData, AttestationObject, AuthenticatorData, AttestedCredentialData
+		from fido2.server import Fido2Server
+		from fido2 import cbor
+	else:
+		raise ImportError(f'Unsupported fido2 version: {__fido2.__version__}')
+
+	def get_webauthn_server():
+		hostname = urllib.parse.urlsplit(request.url).hostname
+		return Fido2Server(PublicKeyCredentialRpEntity(id=current_app.config.get('MFA_RP_ID', hostname),
+		                                               name=current_app.config['MFA_RP_NAME']))
+
+	WEBAUTHN_SUPPORTED = True
+except ImportError as err:
+	warn(_('2FA WebAuthn support disabled because import of the fido2 module failed (%s)')%err)
+	WEBAUTHN_SUPPORTED = False

uffd/invite/__init__.py

-from .views import bp as _bp
-
-bp = [_bp]

uffd/invite/templates/invite/list.html

-{% extends 'base.html' %}
-
-{% block body %}
-
-<div class="btn-toolbar mb-2">
-	<a class="btn btn-primary ml-auto" href="{{ url_for("invite.new") }}"><i class="fa fa-plus" aria-hidden="true"></i> New</a>
-</div>
-<div class="table-responsive">
-	<table class="table">
-		<thead>
-			<tr>
-				<th scope="col">Link</th>
-				<th scope="col">Status</th>
-				<th scope="col">Created on</th>
-				<th scope="col">Expires after</th>
-				<th scope="col">Permissions</th>
-				<th scope="col">Usages</th>
-				<th scope="col"></th>
-			</tr>
-		</thead>
-		<tbody>
-			{% for invite in invites|sort(attribute='created', reverse=True)|sort(attribute='active', reverse=True) %}
-			<tr>
-				<td><a style="width: 8em; display: inline-block;" class="text-truncate" href="{{ url_for('invite.use', token=invite.token) }}"><code>{{ invite.token }}</code></a></td>
-				<td>
-					{% if invite.disabled %}
-						Disabled
-					{% elif invite.voided %}
-						Voided
-					{% elif invite.expired %}
-						Expired
-					{% elif not invite.active %}
-						Invalid
-					{% elif invite.single_use %}
-						Valid once
-					{% else %}
-						Valid
-					{% endif %}
-				</td>
-				<td>{{ invite.created.strftime('%Y-%m-%d %H:%M') }}</td>
-				<td>{{ invite.valid_until.strftime('%Y-%m-%d %H:%M') }}</td>
-				<td>
-					{{ 'Signup' if invite.allow_signup }}{{ ', ' if invite.allow_signup and invite.roles }}
-					{% for role in invite.roles %}{{ ', ' if loop.index != 1 }}<a href="{{ url_for('role.show', roleid=role.id) }}" style="white-space: nowrap;"><i class="fas fa-key"></i>&thinsp;{{ role.name }}</a>{% endfor %}
-				</td>
-				<td>
-					<a href="#" data-toggle="modal" data-target="#modal-{{ invite.token }}">
-						<span style="white-space: nowrap;">{{ invite.signups|selectattr('completed')|list|length }} <i class="fas fa-users" title="user registrations"></i></span>,
-						<span style="white-space: nowrap;">{{ invite.grants|length }} <i class="fas fa-key" title="role grants"></i></span>
-					</a>
-				</td>
-				<td class="text-right">
-					{% if invite.active %}
-					<form action="{{ url_for('invite.disable', token=invite.token) }}" method="POST">
-					<button type="submit" class="btn btn-link btn-sm py-0" title="Disable"><i class="fas fa-ban" style="width: 1.5em;"></i></button>
-					</form>
-					{% else %}
-					<form action="{{ url_for('invite.reset', token=invite.token) }}" method="POST">
-					<button type="submit" class="btn btn-link btn-sm py-0" title="Reenable" {{ 'disabled' if invite.expired }}><i class="fas fa-redo" style="width: 1.5em;"></i></button>
-					</form>
-					{% endif %}
-				</td>
-			</tr>
-			{% endfor %}
-		</tbody>
-	</table>
-</div>
-
-{% for invite in invites %}
-<div class="modal" tabindex="-1" id="modal-{{ invite.token }}">
-	<div class="modal-dialog">
-		<div class="modal-content">
-			<div class="modal-header">
-				<h5 class="modal-title">Invite Usages</h5>
-				<button type="button" class="close" data-dismiss="modal" aria-label="Close">
-					<span aria-hidden="true">&times;</span>
-				</button>
-			</div>
-			<div class="modal-body">
-				<ul>
-					{% for signup in invite.signups if signup.completed %}
-					<li>Registration of user <a href="{{ url_for('user.show', uid=signup.user.uid) }}">{{ signup.user.loginname }}</a></li>
-					{% endfor %}
-					{% for grant in invite.grants if grant.user %}
-					<li>Roles granted to <a href="{{ url_for('user.show', uid=grant.user.uid) }}">{{ grant.user.loginname }}</a></li>
-					{% endfor %}
-				</ul>
-			</div>
-			<div class="modal-footer">
-				<button type="button" class="btn btn-primary" data-dismiss="modal">Close</button>
-			</div>
-		</div>
-	</div>
-</div>
-{% endfor %}
-
-{% endblock %}

uffd/invite/templates/invite/use.html

-{% extends 'base.html' %}
-
-{% block body %}
-
-<div class="row mt-2 justify-content-center">
-	<div class="col-lg-6 col-md-10" style="background: #f7f7f7; box-shadow: 0px 2px 2px rgba(0, 0, 0, 0.3); padding: 30px;">
-		<div class="text-center">
-			<img alt="branding logo" src="{{ config.get("BRANDING_LOGO_URL") }}" class="col-lg-8 col-md-12" >
-		</div>
-		<div class="col-12 mb-3">
-			<h2 class="text-center">Invite Link</h2>
-		</div>
-		{% if not is_valid_session() %}
-		<p>Welcome to the CCCV Single-Sign-On!</p>
-		{% endif %}
-
-		{% if invite.roles and invite.allow_signup %}
-		<p>With this link you can register a new user account with the following roles or add the roles to an existing account:</p>
-		{% elif invite.roles %}
-		<p>With this link you can add the following roles to an existing account:</p>
-		{% elif invite.allow_signup %}
-		<p>With this link you can register a new user account.</p>
-		{% endif %}
-		{% if invite.roles %}
-		<ul>
-			{% for role in invite.roles %}
-			<li>{{ role.name }}{% if role.description %}: {{ role.description }}{% endif %}</li>
-			{% endfor %}
-		</ul>
-		{% endif %}
-		{% if is_valid_session() %}
-			{% if invite.roles %}
-				<form method="POST" action="{{ url_for("invite.grant", token=invite.token) }}" class="mb-2">
-					<button type="submit" class="btn btn-primary btn-block">Add the roles to your account now</button>
-				</form>
-				<a href="{{ url_for("session.logout", ref=url_for("session.login", ref=request.url)) }}" class="btn btn-secondary btn-block">Logout and switch to a different account</a>
-			{% endif %}
-			{% if invite.allow_signup %}
-				<a href="{{ url_for("session.logout", ref=url_for("invite.signup_start", token=invite.token)) }}" class="btn btn-secondary btn-block">Logout to register a new account</a>
-			{% endif %}
-		{% else %}
-			{% if invite.allow_signup %}
-				<a href="{{ url_for("invite.signup_start", token=invite.token) }}" class="btn btn-primary btn-block">Register a new account</a>
-			{% endif %}
-			{% if invite.roles %}
-				<a href="{{ url_for("session.login", ref=request.url) }}" class="btn btn-primary btn-block">Login and add the roles to your account</a>
-			{% endif %}
-		{% endif %}
-	</div>
-</div>
-{% endblock %}

uffd/lazyconfig.py

-from collections import UserString, UserList
-
-from flask import current_app
-
-class LazyConfigString(UserString):
-	def __init__(self, seq=None, key=None, default=None, error=True):
-		# pylint: disable=super-init-not-called
-		self.__seq = seq
-		self.__key = key
-		self.__default = default
-		self.__error = error
-
-	@property
-	def data(self):
-		if self.__seq is not None:
-			obj = self.__seq
-		elif self.__error:
-			obj = current_app.config[self.__key]
-		else:
-			obj = current_app.config.get(self.__key, self.__default)
-		return str(obj)
-
-	def __bytes__(self):
-		return self.data.encode()
-
-	def __get__(self, obj, owner=None):
-		return self.data
-
-def lazyconfig_str(key, **kwargs):
-	return LazyConfigString(None, key, **kwargs)
-
-class LazyConfigList(UserList):
-	def __init__(self, seq=None, key=None, default=None, error=True):
-		# pylint: disable=super-init-not-called
-		self.__seq = seq
-		self.__key = key
-		self.__default = default
-		self.__error = error
-
-	@property
-	def data(self):
-		if self.__seq is not None:
-			obj = self.__seq
-		elif self.__error:
-			obj = current_app.config[self.__key]
-		else:
-			obj = current_app.config.get(self.__key, self.__default)
-		return obj
-
-	def __get__(self, obj, owner=None):
-		return self.data
-
-def lazyconfig_list(key, **kwargs):
-	return LazyConfigList(None, key, **kwargs)

uffd/ldap.py

-from flask import current_app, request, abort
-
-import ldap3
-
-from ldapalchemy import LDAPMapper, LDAPCommitError # pylint: disable=unused-import
-from ldapalchemy.model import Query
-
-class FlaskQuery(Query):
-	def get_or_404(self, dn):
-		res = self.get(dn)
-		if res is None:
-			abort(404)
-		return res
-
-	def first_or_404(self):
-		res = self.first()
-		if res is None:
-			abort(404)
-		return res
-
-class FlaskLDAPMapper(LDAPMapper):
-	def __init__(self):
-		super().__init__()
-		class Model(self.Model):
-			query_class = FlaskQuery
-
-		self.Model = Model # pylint: disable=invalid-name
-
-	@property
-	def session(self):
-		if not hasattr(request, 'ldap_session'):
-			request.ldap_session = self.Session(self.get_connection)
-		return request.ldap_session
-
-	def get_connection(self):
-		if hasattr(request, 'ldap_connection'):
-			return request.ldap_connection
-		if current_app.config.get('LDAP_SERVICE_MOCK', False):
-			if not current_app.debug:
-				raise Exception('LDAP_SERVICE_MOCK cannot be enabled on production instances')
-			# Entries are stored in-memory in the mocked `Connection` object. To make
-			# changes persistent across requests we reuse the same `Connection` object
-			# for all calls to `service_conn()` and `user_conn()`.
-			if not hasattr(current_app, 'ldap_mock'):
-				server = ldap3.Server.from_definition('ldap_mock', 'ldap_server_info.json', 'ldap_server_schema.json')
-				current_app.ldap_mock = ldap3.Connection(server, client_strategy=ldap3.MOCK_SYNC)
-				current_app.ldap_mock.strategy.entries_from_json('ldap_server_entries.json')
-				current_app.ldap_mock.bind()
-			return current_app.ldap_mock
-		server = ldap3.Server(current_app.config["LDAP_SERVICE_URL"], get_info=ldap3.ALL)
-		auto_bind = ldap3.AUTO_BIND_TLS_BEFORE_BIND if current_app.config["LDAP_SERVICE_USE_STARTTLS"] else True
-		request.ldap_connection = ldap3.Connection(server, current_app.config["LDAP_SERVICE_BIND_DN"],
-			current_app.config["LDAP_SERVICE_BIND_PASSWORD"], auto_bind=auto_bind)
-		return request.ldap_connection
-
-ldap = FlaskLDAPMapper()

uffd/mail/__init__.py

-from .views import bp as bp_ui
-
-bp = [bp_ui]

uffd/mail/models.py

-from uffd.ldap import ldap
-from uffd.lazyconfig import lazyconfig_str, lazyconfig_list
-
-class Mail(ldap.Model):
-	ldap_search_base = lazyconfig_str('LDAP_MAIL_SEARCH_BASE')
-	ldap_filter_params = lazyconfig_list('LDAP_MAIL_SEARCH_FILTER')
-	ldap_object_classes = lazyconfig_list('LDAP_MAIL_OBJECTCLASSES')
-	ldap_dn_attribute = lazyconfig_str('LDAP_MAIL_DN_ATTRIBUTE')
-	ldap_dn_base = lazyconfig_str('LDAP_MAIL_SEARCH_BASE')
-
-	uid = ldap.Attribute(lazyconfig_str('LDAP_MAIL_UID_ATTRIBUTE'))
-	receivers = ldap.Attribute(lazyconfig_str('LDAP_MAIL_RECEIVERS_ATTRIBUTE'), multi=True)
-	destinations = ldap.Attribute(lazyconfig_str('LDAP_MAIL_DESTINATIONS_ATTRIBUTE'), multi=True)