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_BASE_USER="ou=users,dc=example,dc=com"
-LDAP_BASE_GROUPS="ou=groups,dc=example,dc=com"
-LDAP_BASE_MAIL="ou=postfix,dc=example,dc=com"
+USER_GID=20001
 
-LDAP_SERVICE_BIND_DN=""
-LDAP_SERVICE_BIND_PASSWORD=""
-LDAP_SERVICE_URL="ldapi:///"
+# 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_USER_OBJECTCLASSES=["top", "inetOrgPerson", "organizationalPerson", "person", "posixAccount"]
-LDAP_USER_GID=20001
-LDAP_USER_MIN_UID=10000
-LDAP_USER_MAX_UID=18999
+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'
-MAIL_LDAP_OBJECTCLASSES=["top", "postfixVirtual"]
+
+# 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=[
+#	# Title is mandatory, all other fields are optional.
+#	# For permission_levels/groups/infos/links all fields are mandatory aside from required_group.
+#	{
+#		'title': 'Service Title',
+#		'subtitle': 'Service Subtitle',
+#		'description': 'Short description of the service as plain text',
+#		'url': 'https://example.com/',
+#		'logo_url': 'https://example.com/logo.png',
+#		# Basic access group name, service is accessible to everyone if empty
+#		'required_group': 'users',
+#		# Non-basic permission levels, the last matching entry is selected.
+#		# Users with a matching permission level are considered to have
+#		# access to the service (as if they have the basic access group).
+#		'permission_levels': [
+#			{'name': 'Moderator', 'required_group': 'moderators'},
+#			{'name': 'Admin', 'required_group': 'uffd_admin'},
+#		],
+#		# Per default all services are listed publicly (but grayed out for
+#		# guests/users without access). Confidential services are only visible
+#		# to users with access rights to the service.
+#		'confidential': True,
+#		# In-service groups, all matching items are visible
+#		'groups': [
+#			{'name': 'Group "crew_crew"', 'required_group': 'users'},
+#			{'name': 'Group "crew_logistik"', 'required_group': 'uffd_admin'},
+#		],
+#		# Infos are small/medium amounts of information displayed in a modal
+#		# dialog. All matching items are visible.
+#		'infos': [
+#			{
+#				'title': 'uffd',
+#				'button_text': 'Documentation', # Defaults to the title if not set
+#				'html': '<p>Some information about the service as html</p>',
+#				'required_group': 'users',
+#			},
+#		],
+#		# Links to external sites, all matching items are visible
+#		'links': [
+#			{'title': 'Link to an external site', 'url': '#', 'required_group': 'users'},
+#		]
+#	},
+]
+
+# A banner text that will be displayed above the services list
+SERVICES_BANNER=''
+
+# If the banner should be shown to users who are not logged in
+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/ldap/__init__.py

-from .ldap import bp as ldap_bp
-from .ldap import get_conn, user_conn, escape_filter_chars, uid_to_dn
-from .ldap import loginname_to_dn, mail_to_dn, get_next_uid, loginname_is_safe, mailname_is_safe
-from .ldap import get_ldap_array_attribute_safe, get_ldap_attribute_safe
-
-bp = [ldap_bp]

uffd/ldap/ldap.py

-import string
-
-from flask import Blueprint, current_app
-from ldap3.utils.conv import escape_filter_chars
-from ldap3.core.exceptions import LDAPBindError, LDAPCursorError, LDAPPasswordIsMandatoryError
-
-from ldap3 import Server, Connection, ALL, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, MOCK_SYNC
-
-bp = Blueprint("ldap", __name__)
-
-def fix_connection(conn):
-	old_search = conn.search
-	def search(*args, **kwargs):
-		kwargs.update({'attributes': [ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES]})
-		return old_search(*args, **kwargs)
-	conn.search = search
-	return conn
-
-def get_mock_conn():
-	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 = Server.from_definition('ldap_mock', 'ldap_server_info.json', 'ldap_server_schema.json')
-		current_app.ldap_mock = fix_connection(Connection(server, client_strategy=MOCK_SYNC))
-		current_app.ldap_mock.strategy.entries_from_json('ldap_server_entries.json')
-		current_app.ldap_mock.bind()
-	return current_app.ldap_mock
-
-def service_conn():
-	if current_app.config.get('LDAP_SERVICE_MOCK', False):
-		return get_mock_conn()
-	server = Server(current_app.config["LDAP_SERVICE_URL"], get_info=ALL)
-	return fix_connection(Connection(server, current_app.config["LDAP_SERVICE_BIND_DN"], current_app.config["LDAP_SERVICE_BIND_PASSWORD"], auto_bind=True))
-
-def user_conn(loginname, password):
-	if not loginname_is_safe(loginname):
-		return False
-	if current_app.config.get('LDAP_SERVICE_MOCK', False):
-		conn = get_mock_conn()
-		# Since we reuse the same conn for all calls to `user_conn()` we
-		# simulate the password check by rebinding. Note that ldap3's mocking
-		# implementation just compares the string in the objects's userPassword
-		# field with the password, no support for hashing or OpenLDAP-style
-		# password-prefixes ("{PLAIN}..." or "{ssha512}...").
-		try:
-			if not conn.rebind(loginname_to_dn(loginname), password):
-				return False
-		except (LDAPBindError, LDAPPasswordIsMandatoryError):
-			return False
-		return get_mock_conn()
-	server = Server(current_app.config["LDAP_SERVICE_URL"], get_info=ALL)
-	try:
-		return fix_connection(Connection(server, loginname_to_dn(loginname), password, auto_bind=True))
-	except (LDAPBindError, LDAPPasswordIsMandatoryError):
-		return False
-
-def get_conn():
-	return service_conn()
-
-def uid_to_dn(uid):
-	conn = get_conn()
-	conn.search(current_app.config["LDAP_BASE_USER"], '(&(objectclass=person)(uidNumber={}))'.format(escape_filter_chars(uid)))
-	if not len(conn.entries) == 1:
-		return None
-	return conn.entries[0].entry_dn
-
-def loginname_to_dn(loginname):
-	if loginname_is_safe(loginname):
-		return 'uid={},{}'.format(loginname, current_app.config["LDAP_BASE_USER"])
-	raise ValueError('unsafe login name')
-
-def mail_to_dn(uid):
-	if mailname_is_safe(uid):
-		return 'uid={},{}'.format(uid, current_app.config["LDAP_BASE_MAIL"])
-	raise Exception('unsafe mail name')
-
-def loginname_is_safe(value):
-	if len(value) > 32 or len(value) < 1:
-		return False
-	for char in value:
-		if not char in string.ascii_lowercase + string.digits + '_-':
-			return False
-	return True
-
-def mailname_is_safe(value):
-	return loginname_is_safe(value)
-
-def get_next_uid():
-	conn = get_conn()
-	conn.search(current_app.config["LDAP_BASE_USER"], '(objectclass=person)')
-	max_uid = current_app.config["LDAP_USER_MIN_UID"]
-	for i in conn.entries:
-		# skip out of range entries
-		if i['uidNumber'].value > current_app.config["LDAP_USER_MAX_UID"]:
-			continue
-		if i['uidNumber'].value < current_app.config["LDAP_USER_MIN_UID"]:
-			continue
-		max_uid = max(i['uidNumber'].value, max_uid)
-	next_uid = max_uid + 1
-	if uid_to_dn(next_uid):
-		raise Exception('No free uid found')
-	return next_uid
-
-def get_ldap_attribute_safe(ldapobject, attribute):
-	try:
-		result = ldapobject[attribute].value if attribute in ldapobject  else None
-	# we have to catch LDAPCursorError here, because ldap3 in older versions has a broken __contains__ function
-	# see https://github.com/cannatag/ldap3/issues/493
-	# fixed in version 2.5
-	# debian buster ships 2.4.1
-	except LDAPCursorError:
-		result = None
-	return result
-
-def get_ldap_array_attribute_safe(ldapobject, attribute):
-	# if the aray is empty, the attribute does not exist.
-	# if there is only one elemtent, ldap returns a string and not an array with one element
-	# we sanitize this to always be an array
-	result = get_ldap_attribute_safe(ldapobject, attribute)
-	if not result:
-		result = []
-	if isinstance(result, str):
-		result = [result]
-	return result

uffd/mail/__init__.py

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

uffd/mail/models.py

-from ldap3 import MODIFY_REPLACE
-from flask import current_app
-
-from uffd import ldap
-
-class Mail():
-	def __init__(self, uid=None, destinations=None, receivers=None, dn=None):
-		self.uid = uid
-		self.receivers = receivers if receivers else []
-		self.destinations = destinations if destinations else []
-		self.dn = dn
-
-	@classmethod
-	def from_ldap(cls, ldapobject):
-		return Mail(
-				uid=ldapobject['uid'].value,
-				receivers=ldap.get_ldap_array_attribute_safe(ldapobject, 'mailacceptinggeneralid'),
-				destinations=ldap.get_ldap_array_attribute_safe(ldapobject, 'maildrop'),
-				dn=ldapobject.entry_dn,
-			)
-
-	@classmethod
-	def from_ldap_dn(cls, dn):
-		conn = ldap.get_conn()
-		conn.search(dn, '(objectClass=postfixVirtual)')
-		if not len(conn.entries) == 1:
-			return None
-		return Mail.from_ldap(conn.entries[0])
-
-	def to_ldap(self, new=False):
-		conn = ldap.get_conn()
-		if new:
-			attributes = {
-				'uid': self.uid,
-				# same as for update
-				'mailacceptinggeneralid': self.receivers,
-				'maildrop': self.destinations,
-			}
-			self.dn = ldap.mail_to_dn(self.uid)
-			result = conn.add(self.dn, current_app.config['MAIL_LDAP_OBJECTCLASSES'], attributes)
-		else:
-			attributes = {
-				'mailacceptinggeneralid': [(MODIFY_REPLACE, self.receivers)],
-				'maildrop': [(MODIFY_REPLACE, self.destinations)],
-				}
-			result = conn.modify(self.dn, attributes)
-		return result

uffd/mail/views.py

-from flask import Blueprint, render_template, request, url_for, redirect, flash, current_app
-
-from uffd.navbar import register_navbar
-from uffd.csrf import csrf_protect
-from uffd.ldap import get_conn, escape_filter_chars
-from uffd.session import login_required, is_valid_session, get_current_user
-
-from uffd.mail.models import Mail
-
-bp = Blueprint("mail", __name__, template_folder='templates', url_prefix='/mail/')
-@bp.before_request
-@login_required()
-def mail_acl(): #pylint: disable=inconsistent-return-statements
-	if not mail_acl_check():
-		flash('Access denied')
-		return redirect(url_for('index'))
-
-def mail_acl_check():
-	return is_valid_session() and get_current_user().is_in_group(current_app.config['ACL_ADMIN_GROUP'])
-
-@bp.route("/")
-@register_navbar('Mail', icon='envelope', blueprint=bp, visible=mail_acl_check)
-def index():
-	conn = get_conn()
-	conn.search(current_app.config["LDAP_BASE_MAIL"], '(objectclass=postfixVirtual)')
-	mails = []
-	for i in conn.entries:
-		mails.append(Mail.from_ldap(i))
-	return render_template('mail_list.html', mails=mails)
-
-@bp.route("/<uid>")
-@bp.route("/new")
-def show(uid=None):
-	if not uid:
-		mail = Mail()
-	else:
-		conn = get_conn()
-		conn.search(current_app.config["LDAP_BASE_MAIL"], '(&(objectclass=postfixVirtual)(uid={}))'.format((escape_filter_chars(uid))))
-		assert len(conn.entries) == 1
-		mail = Mail.from_ldap(conn.entries[0])
-	return render_template('mail.html', mail=mail)
-
-@bp.route("/<uid>/update", methods=['POST'])
-@bp.route("/new", methods=['POST'])
-@csrf_protect(blueprint=bp)
-def update(uid=False):
-	conn = get_conn()
-	is_newmail = bool(not uid)
-	if is_newmail:
-		mail = Mail()
-	else:
-		conn = get_conn()
-		conn.search(current_app.config["LDAP_BASE_MAIL"], '(&(objectclass=postfixVirtual)(uid={}))'.format((escape_filter_chars(uid))))
-		assert len(conn.entries) == 1
-		mail = Mail.from_ldap(conn.entries[0])
-
-	if is_newmail:
-		mail.uid = request.form.get('mail-uid')
-	mail.receivers = request.form.get('mail-receivers', '').splitlines()
-	mail.destinations = request.form.get('mail-destinations', '').splitlines()
-
-	if mail.to_ldap(new=is_newmail):
-		flash('Mail mapping updated.')
-	else:
-		flash('Error updating mail mapping: {}'.format(conn.result['message']))
-		if is_newmail:
-			return redirect(url_for('mail.index'))
-	return redirect(url_for('mail.show', uid=mail.uid))
-
-@bp.route("/<uid>/del")
-@csrf_protect(blueprint=bp)
-def delete(uid):
-	conn = get_conn()
-	conn.search(current_app.config["LDAP_BASE_MAIL"], '(&(objectclass=postfixVirtual)(uid={}))'.format((escape_filter_chars(uid))))
-	assert len(conn.entries) == 1
-	mail = conn.entries[0]
-
-	if conn.delete(mail.entry_dn):
-		flash('Deleted mail mapping.')
-	else:
-		flash('Could not delete mail mapping: {}'.format(conn.result['message']))
-	return redirect(url_for('mail.index'))

uffd/mfa/__init__.py

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

uffd/mfa/templates/disable.html

-{% extends 'base.html' %}
-
-{% block body %}
-
-<p>When you proceed, all recovery codes, registered authenticator applications and devices will be invalidated. 
-You can later generate new recovery codes and setup your applications and devices again.</p>
-
-<form class="form" action="{{ url_for('mfa.disable_confirm') }}" method="POST">
-	<button type="submit" class="btn btn-danger btn-block">Disable two-factor authentication</button>
-</form>
-
-{% endblock %}

uffd/mfa/views.py

-from warnings import warn
-import urllib.parse
-
-from flask import Blueprint, render_template, session, request, redirect, url_for, flash, current_app, abort
-
-from uffd.database import db
-from uffd.mfa.models import MFAMethod, TOTPMethod, WebauthnMethod, RecoveryCodeMethod
-from uffd.session.views import get_current_user, login_required, pre_mfa_login_required
-from uffd.ldap import uid_to_dn
-from uffd.user.models import User
-from uffd.csrf import csrf_protect
-from uffd.ratelimit import Ratelimit, format_delay
-
-bp = Blueprint('mfa', __name__, template_folder='templates', url_prefix='/mfa/')
-
-mfa_ratelimit = Ratelimit('mfa', 1*60, 3)
-
-@bp.route('/', methods=['GET'])
-@login_required()
-def setup():
-	user = get_current_user()
-	recovery_methods = RecoveryCodeMethod.query.filter_by(dn=user.dn).all()
-	totp_methods = TOTPMethod.query.filter_by(dn=user.dn).all()
-	webauthn_methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
-	return render_template('setup.html', totp_methods=totp_methods, webauthn_methods=webauthn_methods, recovery_methods=recovery_methods)
-
-@bp.route('/setup/disable', methods=['GET'])
-@login_required()
-def disable():
-	return render_template('disable.html')
-
-@bp.route('/setup/disable', methods=['POST'])
-@login_required()
-@csrf_protect(blueprint=bp)
-def disable_confirm():
-	user = get_current_user()
-	MFAMethod.query.filter_by(dn=user.dn).delete()
-	db.session.commit()
-	return redirect(url_for('mfa.setup'))
-
-@bp.route('/admin/<int:uid>/disable')
-@login_required()
-@csrf_protect(blueprint=bp)
-def admin_disable(uid):
-	# Group cannot be checked with login_required kwarg, because the config
-	# variable is not available when the decorator is processed
-	if not get_current_user().is_in_group(current_app.config['ACL_ADMIN_GROUP']):
-		flash('Access denied')
-		return redirect(url_for('index'))
-	user = User.from_ldap_dn(uid_to_dn(uid))
-	MFAMethod.query.filter_by(dn=user.dn).delete()
-	db.session.commit()
-	flash('Two-factor authentication was reset')
-	return redirect(url_for('user.show', uid=uid))
-
-@bp.route('/setup/recovery', methods=['POST'])
-@login_required()
-@csrf_protect(blueprint=bp)
-def setup_recovery():
-	user = get_current_user()
-	for method in RecoveryCodeMethod.query.filter_by(dn=user.dn).all():
-		db.session.delete(method)
-	methods = []
-	for _ in range(10):
-		method = RecoveryCodeMethod(user)
-		methods.append(method)
-		db.session.add(method)
-	db.session.commit()
-	return render_template('setup_recovery.html', methods=methods)
-
-@bp.route('/setup/totp', methods=['GET'])
-@login_required()
-def setup_totp():
-	user = get_current_user()
-	method = TOTPMethod(user)
-	session['mfa_totp_key'] = method.key
-	return render_template('setup_totp.html', method=method, name=request.values['name'])
-
-@bp.route('/setup/totp', methods=['POST'])
-@login_required()
-@csrf_protect(blueprint=bp)
-def setup_totp_finish():
-	user = get_current_user()
-	if not RecoveryCodeMethod.query.filter_by(dn=user.dn).all():
-		flash('Generate recovery codes first!')
-		return redirect(url_for('mfa.setup'))
-	method = TOTPMethod(user, name=request.values['name'], key=session.pop('mfa_totp_key'))
-	if method.verify(request.form['code']):
-		db.session.add(method)
-		db.session.commit()
-		return redirect(url_for('mfa.setup'))
-	flash('Code is invalid')
-	return redirect(url_for('mfa.setup_totp', name=request.values['name']))
-
-@bp.route('/setup/totp/<int:id>/delete')
-@login_required()
-@csrf_protect(blueprint=bp)
-def delete_totp(id): #pylint: disable=redefined-builtin
-	user = get_current_user()
-	method = TOTPMethod.query.filter_by(dn=user.dn, id=id).first_or_404()
-	db.session.delete(method)
-	db.session.commit()
-	return redirect(url_for('mfa.setup'))
-
-# WebAuthn support is optional because fido2 has a pretty unstable
-# interface (v0.5.0 on buster and current version are completely
-# incompatible) and might be difficult to install with the correct version
-try:
-	from fido2.client import ClientData
-	from fido2.server import Fido2Server, RelyingParty
-	from fido2.ctap2 import AttestationObject, AuthenticatorData
-	from fido2 import cbor
-	WEBAUTHN_SUPPORTED = True
-except ImportError as err:
-	warn('2FA WebAuthn support disabled because import of the fido2 module failed (%s)'%err)
-	WEBAUTHN_SUPPORTED = False
-
-bp.add_app_template_global(WEBAUTHN_SUPPORTED, name='webauthn_supported')
-
-if WEBAUTHN_SUPPORTED:
-	def get_webauthn_server():
-		return Fido2Server(RelyingParty(current_app.config.get('MFA_RP_ID', urllib.parse.urlsplit(request.url).hostname), current_app.config['MFA_RP_NAME']))
-
-	@bp.route('/setup/webauthn/begin', methods=['POST'])
-	@login_required()
-	@csrf_protect(blueprint=bp)
-	def setup_webauthn_begin():
-		user = get_current_user()
-		if not RecoveryCodeMethod.query.filter_by(dn=user.dn).all():
-			abort(403)
-		methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
-		creds = [method.cred for method in methods]
-		server = get_webauthn_server()
-		registration_data, state = server.register_begin(
-			{
-				"id": user.dn.encode(),
-				"name": user.loginname,
-				"displayName": user.displayname,
-			},
-			creds,
-			user_verification='discouraged',
-		)
-		session["webauthn-state"] = state
-		return cbor.dumps(registration_data)
-
-	@bp.route('/setup/webauthn/complete', methods=['POST'])
-	@login_required()
-	@csrf_protect(blueprint=bp)
-	def setup_webauthn_complete():
-		user = get_current_user()
-		server = get_webauthn_server()
-		data = cbor.loads(request.get_data())[0]
-		client_data = ClientData(data["clientDataJSON"])
-		att_obj = AttestationObject(data["attestationObject"])
-		auth_data = server.register_complete(session["webauthn-state"], client_data, att_obj)
-		method = WebauthnMethod(user, auth_data.credential_data, name=data['name'])
-		db.session.add(method)
-		db.session.commit()
-		return cbor.dumps({"status": "OK"})
-
-	@bp.route("/auth/webauthn/begin", methods=["POST"])
-	@pre_mfa_login_required(no_redirect=True)
-	def auth_webauthn_begin():
-		user = get_current_user()
-		server = get_webauthn_server()
-		methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
-		creds = [method.cred for method in methods]
-		if not creds:
-			abort(404)
-		auth_data, state = server.authenticate_begin(creds, user_verification='discouraged')
-		session["webauthn-state"] = state
-		return cbor.dumps(auth_data)
-
-	@bp.route("/auth/webauthn/complete", methods=["POST"])
-	@pre_mfa_login_required(no_redirect=True)
-	def auth_webauthn_complete():
-		user = get_current_user()
-		server = get_webauthn_server()
-		methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
-		creds = [method.cred for method in methods]
-		if not creds:
-			abort(404)
-		data = cbor.loads(request.get_data())[0]
-		credential_id = data["credentialId"]
-		client_data = ClientData(data["clientDataJSON"])
-		auth_data = AuthenticatorData(data["authenticatorData"])
-		signature = data["signature"]
-		# authenticate_complete() (as of python-fido2 v0.5.0, the version in Debian Buster)
-		# does not check signCount, although the spec recommends it
-		server.authenticate_complete(
-			session.pop("webauthn-state"),
-			creds,
-			credential_id,
-			client_data,
-			auth_data,
-			signature,
-		)
-		session['user_mfa'] = True
-		return cbor.dumps({"status": "OK"})
-
-@bp.route('/setup/webauthn/<int:id>/delete')
-@login_required()
-@csrf_protect(blueprint=bp)
-def delete_webauthn(id): #pylint: disable=redefined-builtin
-	user = get_current_user()
-	method = WebauthnMethod.query.filter_by(dn=user.dn, id=id).first_or_404()
-	db.session.delete(method)
-	db.session.commit()
-	return redirect(url_for('mfa.setup'))
-
-@bp.route('/auth', methods=['GET'])
-@pre_mfa_login_required()
-def auth():
-	user = get_current_user()
-	recovery_methods = RecoveryCodeMethod.query.filter_by(dn=user.dn).all()
-	totp_methods = TOTPMethod.query.filter_by(dn=user.dn).all()
-	webauthn_methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
-	if not totp_methods and not webauthn_methods:
-		session['user_mfa'] = True
-	if session.get('user_mfa'):
-		return redirect(request.values.get('ref', url_for('index')))
-	return render_template('auth.html', ref=request.values.get('ref'), totp_methods=totp_methods,
-			webauthn_methods=webauthn_methods, recovery_methods=recovery_methods)
-
-@bp.route('/auth', methods=['POST'])
-@pre_mfa_login_required()
-def auth_finish():
-	user = get_current_user()
-	delay = mfa_ratelimit.get_delay(user.dn)
-	if delay:
-		flash('We received too many invalid attempts! Please wait at least %s.'%format_delay(delay))
-		return redirect(url_for('mfa.auth', ref=request.values.get('ref')))
-	recovery_methods = RecoveryCodeMethod.query.filter_by(dn=user.dn).all()
-	totp_methods = TOTPMethod.query.filter_by(dn=user.dn).all()
-	for method in totp_methods:
-		if method.verify(request.form['code']):
-			session['user_mfa'] = True
-			return redirect(request.values.get('ref', url_for('index')))
-	for method in recovery_methods:
-		if method.verify(request.form['code']):
-			db.session.delete(method)
-			db.session.commit()
-			session['user_mfa'] = True
-			if len(recovery_methods) <= 1:
-				flash('You have exhausted your recovery codes. Please generate new ones now!')
-				return redirect(url_for('mfa.setup'))
-			if len(recovery_methods) <= 5:
-				flash('You only have a few recovery codes remaining. Make sure to generate new ones before they run out.')
-				return redirect(url_for('mfa.setup'))
-			return redirect(request.values.get('ref', url_for('index')))
-	mfa_ratelimit.log(user.dn)
-	flash('Two-factor authentication failed')
-	return redirect(url_for('mfa.auth', ref=request.values.get('ref')))

uffd/migrations/README.md

+Database Migrations
+===================
+
+While we use Alembic in a single-database configuration, the migration scripts
+are compatible with both SQLite and MySQL/MariaDB.
+
+Compatability with SQLite almost always requires `batch_alter_table` operations
+to modify existing tables. These recreate the tables, copy the data and finally
+replace the old with the newly creaed ones. Alembic is configured to
+auto-generate those operations, but in most cases the generated code fails to
+fully reflect all details of the original schema. This way some contraints
+(i.e. `CHECK` contstrains on Enums) are lost. Define the full table and pass it
+with `copy_from` to `batch_alter_table` to prevent this.
+
+Compatability with MySQL requires special care when changing primary keys and
+when dealing with foreign keys. It often helps to temporarily remove foreign
+key constraints concerning the table that is subject to change. When adding an
+autoincrement id column as the new primary key of a table, recreate the table
+with `batch_alter_table`.
+
+The `check_migrations.py` script verifies that upgrading and downgrading works
+with both databases. While it is far from perfect, it catches many common
+errors. It runs automatically as part of the CI pipeline. Make sure to update
+the script when adding new tables and when making significant changes to
+existing tables.

uffd/migrations/alembic.ini

+# A generic, single database configuration.
+
+[alembic]
+# template used to generate migration files
+# file_template = %%(rev)s_%%(slug)s
+
+# set to 'true' to run the environment during
+# the 'revision' command, regardless of autogenerate
+# revision_environment = false
+
+
+# Logging configuration
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S