diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index c301af2dd0cbd39beda03144d01dc0bf31cd5144..e7476c4f92d622a9380e449382874205a7db9b4e 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -1,4 +1,4 @@
-image: registry.git.cccv.de/infra/uffd/docker-images/buster
+image: registry.git.cccv.de/uffd/docker-images/buster
variables:
DEBIAN_FRONTEND: noninteractive
diff --git a/README.md b/README.md
index e6661c0c606f970aa6ea31c139883c09e05de637..e7f8c23c15f2aedbe077991d596f07f9f47bc05c 100644
--- a/README.md
+++ b/README.md
@@ -19,7 +19,11 @@ You can also use virtualenv with the supplied `requirements.txt`.
## development
-Before running uffd, you need to create the database with `flask db upgrade`.
+Clone this repository with the `--recurse-submodules` flag to retrieve submodule dependencies.
+
+Before running uffd, you need to create the database with `flask db upgrade`. The database is placed in
+`instance/uffd.sqlit3`.
+
Then use `flask run` to start the application:
```
@@ -61,6 +65,13 @@ hook-pre-app = exec:FLASK_APP=uffd flask db upgrade
tabs.
+## Config
+
+Uffd reads its default config from `uffd/default_config.cfg`.
+You can overwrite config variables by creating a config file in the `instance` folder.
+The file must be named `conifg.cfg` (Python syntax), `config.json` or `config.yml`/`config.yaml`.
+You can also set a custom file name with the environment variable `CONFIG_FILENAME`.
+
## Bind with service account or as user?
Uffd can use a dedicated service account for LDAP operations by setting `LDAP_SERVICE_BIND_DN`.
@@ -70,7 +81,6 @@ Or set `LDAP_SERVICE_USER_BIND` to use the credentials of the currently logged i
If you choose to run with user credentials, some features are not available, like password resets
or self signup, since in both cases, no user credentials can exist.
-
## OAuth2 Single-Sign-On Provider
Other services can use uffd as an OAuth2.0-based authentication provider.
diff --git a/migrations/versions/a594d3b3e05b_added_role_locked.py b/migrations/versions/a594d3b3e05b_added_role_locked.py
new file mode 100644
index 0000000000000000000000000000000000000000..5ca1b89a97010bdfb11c51f35410d9281187e3a8
--- /dev/null
+++ b/migrations/versions/a594d3b3e05b_added_role_locked.py
@@ -0,0 +1,23 @@
+"""added role.locked
+
+Revision ID: a594d3b3e05b
+Revises: 5cab70e95bf8
+Create Date: 2021-06-14 00:32:47.792794
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = 'a594d3b3e05b'
+down_revision = '5cab70e95bf8'
+branch_labels = None
+depends_on = None
+
+def upgrade():
+ with op.batch_alter_table('role', schema=None) as batch_op:
+ batch_op.add_column(sa.Column('locked', sa.Boolean(name=op.f('ck_role_locked')), nullable=False, default=False))
+
+def downgrade():
+ with op.batch_alter_table('role', schema=None) as batch_op:
+ batch_op.drop_column('locked')
diff --git a/tests/test_invite.py b/tests/test_invite.py
index d3691d05c65cbdb3ce68c93d94a668f79fb8b548..c21a15c16d370febac1744212ae77216fb7ac108 100644
--- a/tests/test_invite.py
+++ b/tests/test_invite.py
@@ -12,7 +12,7 @@ from uffd import create_app, db
from uffd.invite.models import Invite, InviteGrant, InviteSignup
from uffd.user.models import User, Group
from uffd.role.models import Role
-from uffd.session.views import get_current_user, is_valid_session, login_get_user
+from uffd.session.views import login_get_user
from utils import dump, UffdTestCase, db_flush
diff --git a/tests/test_mfa.py b/tests/test_mfa.py
index 4ea98102e0d48f6980dc40c8140e0cd5f8f46fd3..dc0c657cb3c7138b409474e6f3c20cd3a53d9a48 100644
--- a/tests/test_mfa.py
+++ b/tests/test_mfa.py
@@ -2,13 +2,12 @@ import unittest
import datetime
import time
-from flask import url_for, session
+from flask import url_for, session, request
# These imports are required, because otherwise we get circular imports?!
from uffd import ldap, user
from uffd.user.models import User
-from uffd.session.views import get_current_user, is_valid_session
from uffd.mfa.models import MFAMethod, MFAType, RecoveryCodeMethod, TOTPMethod, WebauthnMethod, _hotp
from uffd import create_app, db
@@ -176,21 +175,21 @@ class TestMfaViews(UffdTestCase):
r = self.client.post(path=url_for('mfa.disable_confirm'), follow_redirects=True)
dump('mfa_disable_submit', r)
self.assertEqual(r.status_code, 200)
- self.assertEqual(len(MFAMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(MFAMethod.query.filter_by(dn=request.user.dn).all()), 0)
self.assertEqual(len(MFAMethod.query.filter_by(dn=get_admin().dn).all()), admin_methods)
def test_disable_recovery_only(self):
self.login()
self.add_recovery_codes()
admin_methods = len(MFAMethod.query.filter_by(dn=get_admin().dn).all())
- self.assertNotEqual(len(MFAMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertNotEqual(len(MFAMethod.query.filter_by(dn=request.user.dn).all()), 0)
r = self.client.get(path=url_for('mfa.disable'), follow_redirects=True)
dump('mfa_disable_recovery_only', r)
self.assertEqual(r.status_code, 200)
r = self.client.post(path=url_for('mfa.disable_confirm'), follow_redirects=True)
dump('mfa_disable_recovery_only_submit', r)
self.assertEqual(r.status_code, 200)
- self.assertEqual(len(MFAMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(MFAMethod.query.filter_by(dn=request.user.dn).all()), 0)
self.assertEqual(len(MFAMethod.query.filter_by(dn=get_admin().dn).all()), admin_methods)
def test_admin_disable(self):
@@ -202,7 +201,7 @@ class TestMfaViews(UffdTestCase):
self.add_totp()
self.client.post(path=url_for('session.login'),
data={'loginname': 'testadmin', 'password': 'adminpassword'}, follow_redirects=True)
- self.assertTrue(is_valid_session())
+ self.assertIsNotNone(request.user)
admin_methods = len(MFAMethod.query.filter_by(dn=get_admin().dn).all())
r = self.client.get(path=url_for('mfa.admin_disable', uid=get_user().uid), follow_redirects=True)
dump('mfa_admin_disable', r)
@@ -212,11 +211,11 @@ class TestMfaViews(UffdTestCase):
def test_setup_recovery(self):
self.login()
- self.assertEqual(len(RecoveryCodeMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(RecoveryCodeMethod.query.filter_by(dn=request.user.dn).all()), 0)
r = self.client.post(path=url_for('mfa.setup_recovery'), follow_redirects=True)
dump('mfa_setup_recovery', r)
self.assertEqual(r.status_code, 200)
- methods = RecoveryCodeMethod.query.filter_by(dn=get_current_user().dn).all()
+ methods = RecoveryCodeMethod.query.filter_by(dn=request.user.dn).all()
self.assertNotEqual(len(methods), 0)
r = self.client.post(path=url_for('mfa.setup_recovery'), follow_redirects=True)
dump('mfa_setup_recovery_reset', r)
@@ -241,62 +240,62 @@ class TestMfaViews(UffdTestCase):
def test_setup_totp_finish(self):
self.login()
self.add_recovery_codes()
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 0)
r = self.client.get(path=url_for('mfa.setup_totp', name='My TOTP Authenticator'), follow_redirects=True)
- method = TOTPMethod(get_current_user(), key=session.get('mfa_totp_key', ''))
+ method = TOTPMethod(request.user, key=session.get('mfa_totp_key', ''))
code = _hotp(int(time.time()/30), method.raw_key)
r = self.client.post(path=url_for('mfa.setup_totp_finish', name='My TOTP Authenticator'), data={'code': code}, follow_redirects=True)
dump('mfa_setup_totp_finish', r)
self.assertEqual(r.status_code, 200)
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 1)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 1)
def test_setup_totp_finish_without_recovery(self):
self.login()
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 0)
r = self.client.get(path=url_for('mfa.setup_totp', name='My TOTP Authenticator'), follow_redirects=True)
- method = TOTPMethod(get_current_user(), key=session.get('mfa_totp_key', ''))
+ method = TOTPMethod(request.user, key=session.get('mfa_totp_key', ''))
code = _hotp(int(time.time()/30), method.raw_key)
r = self.client.post(path=url_for('mfa.setup_totp_finish', name='My TOTP Authenticator'), data={'code': code}, follow_redirects=True)
dump('mfa_setup_totp_finish_without_recovery', r)
self.assertEqual(r.status_code, 200)
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 0)
def test_setup_totp_finish_wrong_code(self):
self.login()
self.add_recovery_codes()
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 0)
r = self.client.get(path=url_for('mfa.setup_totp', name='My TOTP Authenticator'), follow_redirects=True)
- method = TOTPMethod(get_current_user(), key=session.get('mfa_totp_key', ''))
+ method = TOTPMethod(request.user, key=session.get('mfa_totp_key', ''))
code = _hotp(int(time.time()/30), method.raw_key)
code = str(int(code[0])+1)[-1] + code[1:]
r = self.client.post(path=url_for('mfa.setup_totp_finish', name='My TOTP Authenticator'), data={'code': code}, follow_redirects=True)
dump('mfa_setup_totp_finish_wrong_code', r)
self.assertEqual(r.status_code, 200)
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 0)
def test_setup_totp_finish_empty_code(self):
self.login()
self.add_recovery_codes()
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 0)
r = self.client.get(path=url_for('mfa.setup_totp', name='My TOTP Authenticator'), follow_redirects=True)
r = self.client.post(path=url_for('mfa.setup_totp_finish', name='My TOTP Authenticator'), data={'code': ''}, follow_redirects=True)
dump('mfa_setup_totp_finish_empty_code', r)
self.assertEqual(r.status_code, 200)
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 0)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 0)
def test_delete_totp(self):
self.login()
self.add_recovery_codes()
self.add_totp()
- method = TOTPMethod(get_current_user(), name='test')
+ method = TOTPMethod(request.user, name='test')
db.session.add(method)
db.session.commit()
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 2)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 2)
r = self.client.get(path=url_for('mfa.delete_totp', id=method.id), follow_redirects=True)
dump('mfa_delete_totp', r)
self.assertEqual(r.status_code, 200)
self.assertEqual(len(TOTPMethod.query.filter_by(id=method.id).all()), 0)
- self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_user().dn).all()), 1)
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=request.user.dn).all()), 1)
# TODO: webauthn setup tests
@@ -304,36 +303,36 @@ class TestMfaViews(UffdTestCase):
self.add_recovery_codes()
self.add_totp()
db.session.commit()
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
r = self.client.post(path=url_for('session.login'),
data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
dump('mfa_auth_redirected', r)
self.assertEqual(r.status_code, 200)
self.assertIn(b'/mfa/auth', r.data)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
r = self.client.get(path=url_for('mfa.auth'), follow_redirects=False)
dump('mfa_auth', r)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
def test_auth_disabled(self):
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
r = self.client.post(path=url_for('session.login'),
data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=False)
r = self.client.get(path=url_for('mfa.auth', ref='/redirecttarget'), follow_redirects=False)
self.assertEqual(r.status_code, 302)
self.assertTrue(r.location.endswith('/redirecttarget'))
- self.assertTrue(is_valid_session())
+ self.assertIsNotNone(request.user)
def test_auth_recovery_only(self):
self.add_recovery_codes()
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
r = self.client.post(path=url_for('session.login'),
data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=False)
r = self.client.get(path=url_for('mfa.auth', ref='/redirecttarget'), follow_redirects=False)
self.assertEqual(r.status_code, 302)
self.assertTrue(r.location.endswith('/redirecttarget'))
- self.assertTrue(is_valid_session())
+ self.assertIsNotNone(request.user)
def test_auth_recovery_code(self):
self.add_recovery_codes()
@@ -346,11 +345,11 @@ class TestMfaViews(UffdTestCase):
r = self.client.get(path=url_for('mfa.auth'), follow_redirects=False)
dump('mfa_auth_recovery_code', r)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
r = self.client.post(path=url_for('mfa.auth_finish', ref='/redirecttarget'), data={'code': method.code})
self.assertEqual(r.status_code, 302)
self.assertTrue(r.location.endswith('/redirecttarget'))
- self.assertTrue(is_valid_session())
+ self.assertIsNotNone(request.user)
self.assertEqual(len(RecoveryCodeMethod.query.filter_by(id=method_id).all()), 0)
def test_auth_totp_code(self):
@@ -364,12 +363,12 @@ class TestMfaViews(UffdTestCase):
r = self.client.get(path=url_for('mfa.auth'), follow_redirects=False)
dump('mfa_auth_totp_code', r)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
code = _hotp(int(time.time()/30), raw_key)
r = self.client.post(path=url_for('mfa.auth_finish'), data={'code': code}, follow_redirects=True)
dump('mfa_auth_totp_code_submit', r)
self.assertEqual(r.status_code, 200)
- self.assertTrue(is_valid_session())
+ self.assertIsNotNone(request.user)
def test_auth_empty_code(self):
self.add_recovery_codes()
@@ -377,11 +376,11 @@ class TestMfaViews(UffdTestCase):
self.login()
r = self.client.get(path=url_for('mfa.auth'), follow_redirects=False)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
r = self.client.post(path=url_for('mfa.auth_finish'), data={'code': ''}, follow_redirects=True)
dump('mfa_auth_empty_code', r)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
def test_auth_invalid_code(self):
self.add_recovery_codes()
@@ -393,13 +392,13 @@ class TestMfaViews(UffdTestCase):
self.login()
r = self.client.get(path=url_for('mfa.auth'), follow_redirects=False)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
code = _hotp(int(time.time()/30), raw_key)
code = str(int(code[0])+1)[-1] + code[1:]
r = self.client.post(path=url_for('mfa.auth_finish'), data={'code': code}, follow_redirects=True)
dump('mfa_auth_invalid_code', r)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
def test_auth_ratelimit(self):
self.add_recovery_codes()
@@ -409,17 +408,17 @@ class TestMfaViews(UffdTestCase):
db.session.add(method)
db.session.commit()
self.login()
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
code = _hotp(int(time.time()/30), raw_key)
inv_code = str(int(code[0])+1)[-1] + code[1:]
for i in range(20):
r = self.client.post(path=url_for('mfa.auth_finish'), data={'code': inv_code}, follow_redirects=True)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
r = self.client.post(path=url_for('mfa.auth_finish'), data={'code': code}, follow_redirects=True)
dump('mfa_auth_ratelimit', r)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
# TODO: webauthn auth tests
diff --git a/tests/test_oauth2.py b/tests/test_oauth2.py
index 1dc7e1c5ace260f37550e118d5bbb20c69dbffaf..b77a09925ed7702325c91c4b3e0a6ed694cb14e5 100644
--- a/tests/test_oauth2.py
+++ b/tests/test_oauth2.py
@@ -6,7 +6,6 @@ from flask import url_for
# These imports are required, because otherwise we get circular imports?!
from uffd import ldap, user
-from uffd.session.views import get_current_user
from uffd.user.models import User
from uffd.oauth2.models import OAuth2Client
from uffd import create_app, db, ldap
diff --git a/tests/test_rolemod.py b/tests/test_rolemod.py
index f8598ab6d73de7a84e19823fa5e226471251784a..f1baba1a694f9044d324cbd9089d370afd602f6a 100644
--- a/tests/test_rolemod.py
+++ b/tests/test_rolemod.py
@@ -1,7 +1,6 @@
from flask import url_for
from uffd.user.models import User, Group
-from uffd.session import get_current_user
from uffd.role.models import Role
from uffd.database import db
from uffd.ldap import ldap
diff --git a/tests/test_selfservice.py b/tests/test_selfservice.py
index 1d6ab94be099dc98da7a2927370fe38b3b3b1b39..6d5d96980406df9fd75fda7cf9aa41fe31ee36ec 100644
--- a/tests/test_selfservice.py
+++ b/tests/test_selfservice.py
@@ -1,12 +1,11 @@
import datetime
import unittest
-from flask import url_for
+from flask import url_for, request
# These imports are required, because otherwise we get circular imports?!
from uffd import ldap, user
-from uffd.session.views import get_current_user
from uffd.selfservice.models import MailToken, PasswordToken
from uffd.user.models import User
from uffd import create_app, db
@@ -28,88 +27,88 @@ class TestSelfservice(UffdTestCase):
r = self.client.get(path=url_for('selfservice.index'))
dump('selfservice_index', r)
self.assertEqual(r.status_code, 200)
- user = get_current_user()
+ user = request.user
self.assertIn(user.displayname.encode(), r.data)
self.assertIn(user.loginname.encode(), r.data)
self.assertIn(user.mail.encode(), r.data)
def test_update_displayname(self):
self.login()
- user = get_current_user()
+ user = request.user
r = self.client.post(path=url_for('selfservice.update'),
data={'displayname': 'New Display Name', 'mail': user.mail, 'password': '', 'password1': ''},
follow_redirects=True)
dump('update_displayname', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertEqual(_user.displayname, 'New Display Name')
def test_update_displayname_invalid(self):
self.login()
- user = get_current_user()
+ user = request.user
r = self.client.post(path=url_for('selfservice.update'),
data={'displayname': '', 'mail': user.mail, 'password': '', 'password1': ''},
follow_redirects=True)
dump('update_displayname_invalid', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertNotEqual(_user.displayname, '')
def test_update_mail(self):
self.login()
- user = get_current_user()
+ user = request.user
r = self.client.post(path=url_for('selfservice.update'),
data={'displayname': user.displayname, 'mail': 'newemail@example.com', 'password': '', 'password1': ''},
follow_redirects=True)
dump('update_mail', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertNotEqual(_user.mail, 'newemail@example.com')
token = MailToken.query.filter(MailToken.loginname == user.loginname).first()
self.assertEqual(token.newmail, 'newemail@example.com')
self.assertIn(token.token, str(self.app.last_mail.get_content()))
r = self.client.get(path=url_for('selfservice.token_mail', token=token.token), follow_redirects=True)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertEqual(_user.mail, 'newemail@example.com')
def test_update_mail_sendfailure(self):
self.app.config['MAIL_SKIP_SEND'] = 'fail'
self.login()
- user = get_current_user()
+ user = request.user
r = self.client.post(path=url_for('selfservice.update'),
data={'displayname': user.displayname, 'mail': 'newemail@example.com', 'password': '', 'password1': ''},
follow_redirects=True)
dump('update_mail_sendfailure', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertNotEqual(_user.mail, 'newemail@example.com')
# Maybe also check that there is no new token in the db
def test_token_mail_emptydb(self):
self.login()
- user = get_current_user()
+ user = request.user
r = self.client.get(path=url_for('selfservice.token_mail', token='A'*128), follow_redirects=True)
dump('token_mail_emptydb', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertEqual(_user.mail, user.mail)
def test_token_mail_invalid(self):
self.login()
- user = get_current_user()
+ user = request.user
db.session.add(MailToken(loginname=user.loginname, newmail='newusermail@example.com'))
db.session.commit()
r = self.client.get(path=url_for('selfservice.token_mail', token='A'*128), follow_redirects=True)
dump('token_mail_invalid', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertEqual(_user.mail, user.mail)
@unittest.skip('See #26')
def test_token_mail_wrong_user(self):
self.login()
- user = get_current_user()
+ user = request.user
admin_user = User.query.get('uid=testadmin,ou=users,dc=example,dc=com')
db.session.add(MailToken(loginname=user.loginname, newmail='newusermail@example.com'))
admin_token = MailToken(loginname='testadmin', newmail='newadminmail@example.com')
@@ -118,14 +117,14 @@ class TestSelfservice(UffdTestCase):
r = self.client.get(path=url_for('selfservice.token_mail', token=admin_token.token), follow_redirects=True)
dump('token_mail_wrong_user', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
_admin_user = User.query.get('uid=testadmin,ou=users,dc=example,dc=com')
self.assertEqual(_user.mail, user.mail)
self.assertEqual(_admin_user.mail, admin_user.mail)
def test_token_mail_expired(self):
self.login()
- user = get_current_user()
+ user = request.user
token = MailToken(loginname=user.loginname, newmail='newusermail@example.com',
created=(datetime.datetime.now() - datetime.timedelta(days=10)))
db.session.add(token)
@@ -133,7 +132,7 @@ class TestSelfservice(UffdTestCase):
r = self.client.get(path=url_for('selfservice.token_mail', token=token.token), follow_redirects=True)
dump('token_mail_expired', r)
self.assertEqual(r.status_code, 200)
- _user = get_current_user()
+ _user = request.user
self.assertEqual(_user.mail, user.mail)
tokens = MailToken.query.filter(MailToken.loginname == user.loginname).all()
self.assertEqual(len(tokens), 0)
diff --git a/tests/test_session.py b/tests/test_session.py
index dae41ab5a06e4e37d61b4f322911b4e62dd8fd26..0882b08e16d895cc474e951488a312ba88c74e14 100644
--- a/tests/test_session.py
+++ b/tests/test_session.py
@@ -1,12 +1,12 @@
import time
import unittest
-from flask import url_for
+from flask import url_for, request
# These imports are required, because otherwise we get circular imports?!
from uffd import ldap, user
-from uffd.session.views import get_current_user, login_required, is_valid_session
+from uffd.session.views import login_required
from uffd import create_app, db
from utils import dump, UffdTestCase
@@ -32,24 +32,24 @@ class TestSession(UffdTestCase):
def setUp(self):
super().setUp()
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
def login(self):
self.client.post(path=url_for('session.login'),
data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
- self.assertTrue(is_valid_session())
+ self.assertIsNotNone(request.user)
def assertLogin(self):
- self.assertTrue(is_valid_session())
+ self.assertIsNotNone(request.user)
self.assertEqual(self.client.get(path=url_for('test_login_required'),
follow_redirects=True).data, b'SUCCESS')
- self.assertEqual(get_current_user().loginname, 'testuser')
+ self.assertEqual(request.user.loginname, 'testuser')
def assertLogout(self):
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
self.assertNotEqual(self.client.get(path=url_for('test_login_required'),
follow_redirects=True).data, b'SUCCESS')
- self.assertEqual(get_current_user(), None)
+ self.assertEqual(request.user, None)
def test_login(self):
self.assertLogout()
@@ -131,7 +131,7 @@ class TestSession(UffdTestCase):
data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
dump('login_ratelimit', r)
self.assertEqual(r.status_code, 200)
- self.assertFalse(is_valid_session())
+ self.assertIsNone(request.user)
class TestSessionOL(TestSession):
use_openldap = True
diff --git a/tests/test_signup.py b/tests/test_signup.py
index 501e8896db3f63cc05c80bae0be5270f1f5e6378..65007789c8eb149bb5688b85ca5f964804c9e3db 100644
--- a/tests/test_signup.py
+++ b/tests/test_signup.py
@@ -2,7 +2,7 @@ import unittest
import datetime
import time
-from flask import url_for, session
+from flask import url_for, session, request
# These imports are required, because otherwise we get circular imports?!
from uffd import user
@@ -11,7 +11,7 @@ from uffd.ldap import ldap
from uffd import create_app, db
from uffd.signup.models import Signup
from uffd.user.models import User
-from uffd.session.views import get_current_user, is_valid_session, login_get_user
+from uffd.session.views import login_get_user
from utils import dump, UffdTestCase, db_flush
@@ -345,8 +345,8 @@ class TestSignupViews(UffdTestCase):
self.assertEqual(signup.user.mail, 'test@example.com')
if self.use_openldap:
self.assertIsNotNone(login_get_user('newuser', 'notsecret'))
- self.assertTrue(is_valid_session())
- self.assertEqual(get_current_user().loginname, 'newuser')
+ self.assertIsNotNone(request.user)
+ self.assertEqual(request.user.loginname, 'newuser')
def test_confirm_loggedin(self):
signup = Signup(loginname='newuser', displayname='New User', mail='test@example.com', password='notsecret')
@@ -354,16 +354,16 @@ class TestSignupViews(UffdTestCase):
self.client.post(path=url_for('session.login'),
data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
self.assertFalse(signup.completed)
- self.assertTrue(is_valid_session())
- self.assertEqual(get_current_user().loginname, 'testuser')
+ self.assertIsNotNone(request.user)
+ self.assertEqual(request.user.loginname, 'testuser')
r = self.client.get(path=url_for('signup.signup_confirm', token=signup.token), follow_redirects=True)
self.assertEqual(r.status_code, 200)
r = self.client.post(path=url_for('signup.signup_confirm_submit', token=signup.token), follow_redirects=True, data={'password': 'notsecret'})
self.assertEqual(r.status_code, 200)
signup = refetch_signup(signup)
self.assertTrue(signup.completed)
- self.assertTrue(is_valid_session())
- self.assertEqual(get_current_user().loginname, 'newuser')
+ self.assertIsNotNone(request.user)
+ self.assertEqual(request.user.loginname, 'newuser')
def test_confirm_notfound(self):
r = self.client.get(path=url_for('signup.signup_confirm', token='notasignuptoken'), follow_redirects=True)
diff --git a/tests/test_user.py b/tests/test_user.py
index 8fae44db05a0e630086d97f0cb88aaa96b635e01..f9fb574b7f5d1f66308d59878d7d24104307278a 100644
--- a/tests/test_user.py
+++ b/tests/test_user.py
@@ -6,7 +6,6 @@ from flask import url_for, session
# These imports are required, because otherwise we get circular imports?!
from uffd import ldap, user
-from uffd.session.views import get_current_user
from uffd.user.models import User
from uffd.role.models import Role
from uffd import create_app, db
@@ -87,16 +86,46 @@ class TestUserViews(UffdTestCase):
user = User.query.get('uid=newuser,ou=users,dc=example,dc=com')
roles = sorted([r.name for r in user.roles])
self.assertIsNotNone(user)
+ self.assertFalse(user.is_service_user)
self.assertEqual(user.loginname, 'newuser')
self.assertEqual(user.displayname, 'New User')
self.assertEqual(user.mail, 'newuser@example.com')
self.assertTrue(user.uid)
role1 = Role(name='role1')
- print('test_new', role1.db_members, role1.members, user.roles)
self.assertEqual(roles, ['base', 'role1'])
# TODO: confirm Mail is send, login not yet possible
#self.assertTrue(ldap.test_user_bind(user.dn, 'newpassword'))
+ def test_new_service(self):
+ db.session.add(Role(name='base'))
+ role1 = Role(name='role1')
+ db.session.add(role1)
+ role2 = Role(name='role2')
+ db.session.add(role2)
+ db.session.commit()
+ role1_id = role1.id
+ r = self.client.get(path=url_for('user.show'), follow_redirects=True)
+ dump('user_new_service', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(User.query.get('uid=newuser,ou=users,dc=example,dc=com'))
+ r = self.client.post(path=url_for('user.update'),
+ data={'loginname': 'newuser', 'mail': 'newuser@example.com', 'displayname': 'New User',
+ f'role-{role1_id}': '1', 'password': 'newpassword', 'serviceaccount': '1'}, follow_redirects=True)
+ dump('user_new_submit', r)
+ self.assertEqual(r.status_code, 200)
+ user = User.query.get('uid=newuser,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in user.roles])
+ self.assertIsNotNone(user)
+ self.assertTrue(user.is_service_user)
+ self.assertEqual(user.loginname, 'newuser')
+ self.assertEqual(user.displayname, 'New User')
+ self.assertEqual(user.mail, 'newuser@example.com')
+ self.assertTrue(user.uid)
+ role1 = Role(name='role1')
+ self.assertEqual(roles, ['role1'])
+ # TODO: confirm Mail is send, login not yet possible
+ #self.assertTrue(ldap.test_user_bind(user.dn, 'newpassword'))
+
def test_new_invalid_loginname(self):
r = self.client.post(path=url_for('user.update'),
data={'loginname': '!newuser', 'mail': 'newuser@example.com', 'displayname': 'New User',
diff --git a/uffd/__init__.py b/uffd/__init__.py
index 0ea45a235ca02783d804563e7add1dc6306f71c1..3a7472c212574fcd463c29647c2c531ee7a07fce 100644
--- a/uffd/__init__.py
+++ b/uffd/__init__.py
@@ -18,6 +18,24 @@ from uffd.template_helper import register_template_helper
from uffd.navbar import setup_navbar
# pylint: enable=wrong-import-position
+def load_config_file(app, cfg_name, silent=False):
+ cfg_path = os.path.join(app.instance_path, cfg_name)
+ if not os.path.exists(cfg_path):
+ if not silent:
+ raise Exception(f"Config file {cfg_path} not found")
+ return False
+
+ if cfg_path.endswith(".json"):
+ app.config.from_json(cfg_path)
+ elif cfg_path.endswith(".yaml") or cfg_path.endswith(".yml"):
+ import yaml # pylint: disable=import-outside-toplevel disable=import-error
+ with open(cfg_path, encoding='utf-8') as ymlfile:
+ data = yaml.safe_load(ymlfile)
+ app.config.from_mapping(data)
+ else:
+ app.config.from_pyfile(cfg_path, silent=True)
+ return True
+
def create_app(test_config=None): # pylint: disable=too-many-locals
# create and configure the app
app = Flask(__name__, instance_relative_config=False)
@@ -30,21 +48,20 @@ def create_app(test_config=None): # pylint: disable=too-many-locals
)
app.config.from_pyfile('default_config.cfg')
+ # load config
+ if test_config is not None:
+ app.config.from_mapping(test_config)
+ elif os.environ.get("CONFIG_FILENAME"):
+ load_config_file(app, os.environ["CONFIG_FILENAME"], silent=False)
+ else:
+ for cfg_name in ["config.cfg", "config.json", "config.yml", "config.yaml"]:
+ if load_config_file(app, cfg_name, silent=True):
+ break
+
register_template_helper(app)
setup_navbar(app)
- if not test_config:
- # load the instance config, if it exists, when not testing
- app.config.from_pyfile(os.path.join(app.instance_path, 'config.cfg'), silent=True)
- else:
- # load the test config if passed in
- app.config.from_mapping(test_config)
-
- # ensure the instance folder exists
- try:
- os.makedirs(app.instance_path)
- except OSError:
- pass
+ os.makedirs(app.instance_path, exist_ok=True)
db.init_app(app)
Migrate(app, db, render_as_batch=True)
diff --git a/uffd/default_config.cfg b/uffd/default_config.cfg
index 559626029e02042fdaa36db1b6d22f937477363d..022657fe3cfac6311f35881939c54b95f2697863 100644
--- a/uffd/default_config.cfg
+++ b/uffd/default_config.cfg
@@ -3,6 +3,8 @@ 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_SERVICE_MIN_UID=19000
+LDAP_USER_SERVICE_MAX_UID=19999
LDAP_USER_GID=20001
LDAP_USER_DN_ATTRIBUTE="uid"
LDAP_USER_UID_ATTRIBUTE="uidNumber"
diff --git a/uffd/invite/templates/invite/list.html b/uffd/invite/templates/invite/list.html
index a371f9e47c0dc44b3fc36f517bda5cffdf8b9978..00fd0ee41c1d352e1134a68dbeb78be6e3be8183 100644
--- a/uffd/invite/templates/invite/list.html
+++ b/uffd/invite/templates/invite/list.html
@@ -21,7 +21,7 @@
{% for invite in invites|sort(attribute='created', reverse=True)|sort(attribute='active', reverse=True) %}
<tr>
<td>
- {% if invite.creator == get_current_user() and invite.active %}
+ {% if invite.creator == request.user and invite.active %}
<a href="{{ url_for('invite.use', token=invite.token) }}"><code>{{ invite.short_token }}</code></a>
<button type="button" class="btn btn-link btn-sm p-0 copy-clipboard" data-copy="{{ url_for('invite.use', token=invite.token, _external=True) }}" title="Copy link to clipboard"><i class="fas fa-clipboard"></i></button>
<button type="button" class="btn btn-link btn-sm p-0" data-toggle="modal" data-target="#modal-{{ invite.id }}-qrcode" title="Show link as QR code"><i class="fas fa-qrcode"></i></button>
@@ -121,7 +121,7 @@
<form action="{{ url_for('invite.disable', invite_id=invite.id) }}" method="POST">
<button type="submit" class="btn btn-primary">Disable Link</button>
</form>
- {% elif invite.creator == get_current_user() and not invite.expired and invite.permitted %}
+ {% elif invite.creator == request.user and not invite.expired and invite.permitted %}
<form action="{{ url_for('invite.reset', invite_id=invite.id) }}" method="POST">
<button type="submit" class="btn btn-primary">Reenable Link</button>
</form>
@@ -132,7 +132,7 @@
</div>
{% endfor %}
-{% for invite in invites if invite.creator == get_current_user() %}
+{% for invite in invites if invite.creator == request.user %}
<div class="modal" tabindex="-1" id="modal-{{ invite.id }}-qrcode">
<div class="modal-dialog">
<div class="modal-content">
diff --git a/uffd/invite/templates/invite/use.html b/uffd/invite/templates/invite/use.html
index a51db8858e2538f797cb015fe76e05a1a321fdc4..4d5b53adb56d9afda80647ddbd25dc13c2c00f4e 100644
--- a/uffd/invite/templates/invite/use.html
+++ b/uffd/invite/templates/invite/use.html
@@ -10,7 +10,7 @@
<div class="col-12 mb-3">
<h2 class="text-center">Invite Link</h2>
</div>
- {% if not is_valid_session() %}
+ {% if not request.user %}
<p>Welcome to the CCCV Single-Sign-On!</p>
{% endif %}
@@ -28,7 +28,7 @@
{% endfor %}
</ul>
{% endif %}
- {% if is_valid_session() %}
+ {% if request.user %}
{% 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>
diff --git a/uffd/invite/views.py b/uffd/invite/views.py
index 518a64cb02373b4894330be7c93be607d35664f6..ece81c6465715c7f377f055a972a0d9b09ad2ea7 100644
--- a/uffd/invite/views.py
+++ b/uffd/invite/views.py
@@ -7,7 +7,7 @@ import sqlalchemy
from uffd.csrf import csrf_protect
from uffd.database import db
from uffd.ldap import ldap
-from uffd.session import get_current_user, login_required, is_valid_session
+from uffd.session import login_required
from uffd.role.models import Role
from uffd.invite.models import Invite, InviteSignup, InviteGrant
from uffd.user.models import User
@@ -16,18 +16,16 @@ from uffd.navbar import register_navbar
from uffd.ratelimit import host_ratelimit, format_delay
from uffd.signup.views import signup_ratelimit
-
bp = Blueprint('invite', __name__, template_folder='templates', url_prefix='/invite/')
def invite_acl():
- if not is_valid_session():
+ if not request.user:
return False
- user = get_current_user()
- if user.is_in_group(current_app.config['ACL_ADMIN_GROUP']):
+ if request.user.is_in_group(current_app.config['ACL_ADMIN_GROUP']):
return True
- if user.is_in_group(current_app.config['ACL_SIGNUP_GROUP']):
+ if request.user.is_in_group(current_app.config['ACL_SIGNUP_GROUP']):
return True
- if Role.query.filter(Role.moderator_group_dn.in_(user.group_dns)).count():
+ if Role.query.filter(Role.moderator_group_dn.in_(request.user.group_dns)).count():
return True
return False
@@ -57,27 +55,25 @@ def reset_acl_filter(user):
@register_navbar('Invites', icon='link', blueprint=bp, visible=invite_acl)
@invite_acl_required
def index():
- invites = Invite.query.filter(view_acl_filter(get_current_user())).all()
+ invites = Invite.query.filter(view_acl_filter(request.user)).all()
return render_template('invite/list.html', invites=invites)
@bp.route('/new')
@invite_acl_required
def new():
- user = get_current_user()
- if user.is_in_group(current_app.config['ACL_ADMIN_GROUP']):
+ if request.user.is_in_group(current_app.config['ACL_ADMIN_GROUP']):
allow_signup = True
roles = Role.query.all()
else:
- allow_signup = user.is_in_group(current_app.config['ACL_SIGNUP_GROUP'])
- roles = Role.query.filter(Role.moderator_group_dn.in_(user.group_dns)).all()
+ allow_signup = request.user.is_in_group(current_app.config['ACL_SIGNUP_GROUP'])
+ roles = Role.query.filter(Role.moderator_group_dn.in_(request.user.group_dns)).all()
return render_template('invite/new.html', roles=roles, allow_signup=allow_signup)
@bp.route('/new', methods=['POST'])
@invite_acl_required
@csrf_protect(blueprint=bp)
def new_submit():
- user = get_current_user()
- invite = Invite(creator=user,
+ invite = Invite(creator=request.user,
single_use=(request.values['single-use'] == '1'),
valid_until=datetime.datetime.fromisoformat(request.values['valid-until']),
allow_signup=(request.values.get('allow-signup', '0') == '1'))
@@ -101,7 +97,7 @@ def new_submit():
@invite_acl_required
@csrf_protect(blueprint=bp)
def disable(invite_id):
- invite = Invite.query.filter(view_acl_filter(get_current_user())).filter_by(id=invite_id).first_or_404()
+ invite = Invite.query.filter(view_acl_filter(request.user)).filter_by(id=invite_id).first_or_404()
invite.disable()
db.session.commit()
return redirect(url_for('.index'))
@@ -110,7 +106,7 @@ def disable(invite_id):
@invite_acl_required
@csrf_protect(blueprint=bp)
def reset(invite_id):
- invite = Invite.query.filter(reset_acl_filter(get_current_user())).filter_by(id=invite_id).first_or_404()
+ invite = Invite.query.filter(reset_acl_filter(request.user)).filter_by(id=invite_id).first_or_404()
invite.reset()
db.session.commit()
return redirect(url_for('.index'))
@@ -128,7 +124,7 @@ def use(token):
@csrf_protect(blueprint=bp)
def grant(token):
invite = Invite.query.filter_by(token=token).first_or_404()
- invite_grant = InviteGrant(invite=invite, user=get_current_user())
+ invite_grant = InviteGrant(invite=invite, user=request.user)
db.session.add(invite_grant)
success, msg = invite_grant.apply()
if not success:
diff --git a/uffd/mail/templates/mail_list.html b/uffd/mail/templates/mail/list.html
similarity index 100%
rename from uffd/mail/templates/mail_list.html
rename to uffd/mail/templates/mail/list.html
diff --git a/uffd/mail/templates/mail.html b/uffd/mail/templates/mail/show.html
similarity index 100%
rename from uffd/mail/templates/mail.html
rename to uffd/mail/templates/mail/show.html
diff --git a/uffd/mail/views.py b/uffd/mail/views.py
index 2aaebf559f44465c17ac6bde67796047b48d86e9..6ba83d10a6ffddb2e58f0755254597875991a3bd 100644
--- a/uffd/mail/views.py
+++ b/uffd/mail/views.py
@@ -3,7 +3,7 @@ from flask import Blueprint, render_template, request, url_for, redirect, flash,
from uffd.navbar import register_navbar
from uffd.csrf import csrf_protect
from uffd.ldap import ldap
-from uffd.session import login_required, is_valid_session, get_current_user
+from uffd.session import login_required
from uffd.mail.models import Mail
@@ -16,12 +16,12 @@ def mail_acl(): #pylint: disable=inconsistent-return-statements
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'])
+ return request.user and request.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():
- return render_template('mail_list.html', mails=Mail.query.all())
+ return render_template('mail/list.html', mails=Mail.query.all())
@bp.route("/<uid>")
@bp.route("/new")
@@ -29,7 +29,7 @@ def show(uid=None):
mail = Mail()
if uid is not None:
mail = Mail.query.filter_by(uid=uid).first_or_404()
- return render_template('mail.html', mail=mail)
+ return render_template('mail/show.html', mail=mail)
@bp.route("/<uid>/update", methods=['POST'])
@bp.route("/new", methods=['POST'])
diff --git a/uffd/mfa/templates/auth.html b/uffd/mfa/templates/mfa/auth.html
similarity index 100%
rename from uffd/mfa/templates/auth.html
rename to uffd/mfa/templates/mfa/auth.html
diff --git a/uffd/mfa/templates/disable.html b/uffd/mfa/templates/mfa/disable.html
similarity index 100%
rename from uffd/mfa/templates/disable.html
rename to uffd/mfa/templates/mfa/disable.html
diff --git a/uffd/mfa/templates/setup.html b/uffd/mfa/templates/mfa/setup.html
similarity index 100%
rename from uffd/mfa/templates/setup.html
rename to uffd/mfa/templates/mfa/setup.html
diff --git a/uffd/mfa/templates/setup_recovery.html b/uffd/mfa/templates/mfa/setup_recovery.html
similarity index 100%
rename from uffd/mfa/templates/setup_recovery.html
rename to uffd/mfa/templates/mfa/setup_recovery.html
diff --git a/uffd/mfa/templates/setup_totp.html b/uffd/mfa/templates/mfa/setup_totp.html
similarity index 100%
rename from uffd/mfa/templates/setup_totp.html
rename to uffd/mfa/templates/mfa/setup_totp.html
diff --git a/uffd/mfa/views.py b/uffd/mfa/views.py
index f7a3e239c6394471543baf940dbbaa9c01de45f4..7b95479f45e86fe37decbcb3b98fb17c2d5a6ece 100644
--- a/uffd/mfa/views.py
+++ b/uffd/mfa/views.py
@@ -5,7 +5,7 @@ from flask import Blueprint, render_template, session, request, redirect, url_fo
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.session.views import login_required, login_required_pre_mfa, set_request_user
from uffd.user.models import User
from uffd.csrf import csrf_protect
from uffd.ratelimit import Ratelimit, format_delay
@@ -17,23 +17,21 @@ 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)
+ recovery_methods = RecoveryCodeMethod.query.filter_by(dn=request.user.dn).all()
+ totp_methods = TOTPMethod.query.filter_by(dn=request.user.dn).all()
+ webauthn_methods = WebauthnMethod.query.filter_by(dn=request.user.dn).all()
+ return render_template('mfa/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')
+ return render_template('mfa/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()
+ MFAMethod.query.filter_by(dn=request.user.dn).delete()
db.session.commit()
return redirect(url_for('mfa.setup'))
@@ -43,7 +41,7 @@ def disable_confirm():
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']):
+ if not request.user.is_in_group(current_app.config['ACL_ADMIN_GROUP']):
flash('Access denied')
return redirect(url_for('index'))
user = User.query.filter_by(uid=uid).one()
@@ -56,34 +54,31 @@ def admin_disable(uid):
@login_required()
@csrf_protect(blueprint=bp)
def setup_recovery():
- user = get_current_user()
- for method in RecoveryCodeMethod.query.filter_by(dn=user.dn).all():
+ for method in RecoveryCodeMethod.query.filter_by(dn=request.user.dn).all():
db.session.delete(method)
methods = []
for _ in range(10):
- method = RecoveryCodeMethod(user)
+ method = RecoveryCodeMethod(request.user)
methods.append(method)
db.session.add(method)
db.session.commit()
- return render_template('setup_recovery.html', methods=methods)
+ return render_template('mfa/setup_recovery.html', methods=methods)
@bp.route('/setup/totp', methods=['GET'])
@login_required()
def setup_totp():
- user = get_current_user()
- method = TOTPMethod(user)
+ method = TOTPMethod(request.user)
session['mfa_totp_key'] = method.key
- return render_template('setup_totp.html', method=method, name=request.values['name'])
+ return render_template('mfa/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():
+ if not RecoveryCodeMethod.query.filter_by(dn=request.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'))
+ method = TOTPMethod(request.user, name=request.values['name'], key=session.pop('mfa_totp_key'))
if method.verify(request.form['code']):
db.session.add(method)
db.session.commit()
@@ -95,8 +90,7 @@ def setup_totp_finish():
@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()
+ method = TOTPMethod.query.filter_by(dn=request.user.dn, id=id).first_or_404()
db.session.delete(method)
db.session.commit()
return redirect(url_for('mfa.setup'))
@@ -124,17 +118,16 @@ if WEBAUTHN_SUPPORTED:
@login_required()
@csrf_protect(blueprint=bp)
def setup_webauthn_begin():
- user = get_current_user()
- if not RecoveryCodeMethod.query.filter_by(dn=user.dn).all():
+ if not RecoveryCodeMethod.query.filter_by(dn=request.user.dn).all():
abort(403)
- methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
+ methods = WebauthnMethod.query.filter_by(dn=request.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,
+ "id": request.user.dn.encode(),
+ "name": request.user.loginname,
+ "displayName": request.user.displayname,
},
creds,
user_verification='discouraged',
@@ -146,23 +139,21 @@ if WEBAUTHN_SUPPORTED:
@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'])
+ method = WebauthnMethod(request.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)
+ @login_required_pre_mfa(no_redirect=True)
def auth_webauthn_begin():
- user = get_current_user()
server = get_webauthn_server()
- methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
+ methods = WebauthnMethod.query.filter_by(dn=request.user_pre_mfa.dn).all()
creds = [method.cred for method in methods]
if not creds:
abort(404)
@@ -171,11 +162,10 @@ if WEBAUTHN_SUPPORTED:
return cbor.dumps(auth_data)
@bp.route("/auth/webauthn/complete", methods=["POST"])
- @pre_mfa_login_required(no_redirect=True)
+ @login_required_pre_mfa(no_redirect=True)
def auth_webauthn_complete():
- user = get_current_user()
server = get_webauthn_server()
- methods = WebauthnMethod.query.filter_by(dn=user.dn).all()
+ methods = WebauthnMethod.query.filter_by(dn=request.user_pre_mfa.dn).all()
creds = [method.cred for method in methods]
if not creds:
abort(404)
@@ -195,51 +185,53 @@ if WEBAUTHN_SUPPORTED:
signature,
)
session['user_mfa'] = True
+ set_request_user()
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()
+ method = WebauthnMethod.query.filter_by(dn=request.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()
+@login_required_pre_mfa()
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()
+ recovery_methods = RecoveryCodeMethod.query.filter_by(dn=request.user_pre_mfa.dn).all()
+ totp_methods = TOTPMethod.query.filter_by(dn=request.user_pre_mfa.dn).all()
+ webauthn_methods = WebauthnMethod.query.filter_by(dn=request.user_pre_mfa.dn).all()
if not totp_methods and not webauthn_methods:
session['user_mfa'] = True
+ set_request_user()
+
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,
+ return render_template('mfa/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()
+@login_required_pre_mfa()
def auth_finish():
- user = get_current_user()
- delay = mfa_ratelimit.get_delay(user.dn)
+ delay = mfa_ratelimit.get_delay(request.user_pre_mfa.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()
+ recovery_methods = RecoveryCodeMethod.query.filter_by(dn=request.user_pre_mfa.dn).all()
+ totp_methods = TOTPMethod.query.filter_by(dn=request.user_pre_mfa.dn).all()
for method in totp_methods:
if method.verify(request.form['code']):
session['user_mfa'] = True
+ set_request_user()
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
+ set_request_user()
if len(recovery_methods) <= 1:
flash('You have exhausted your recovery codes. Please generate new ones now!')
return redirect(url_for('mfa.setup'))
@@ -247,6 +239,6 @@ def auth_finish():
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)
+ mfa_ratelimit.log(request.user_pre_mfa.dn)
flash('Two-factor authentication failed')
return redirect(url_for('mfa.auth', ref=request.values.get('ref')))
diff --git a/uffd/oauth2/templates/error.html b/uffd/oauth2/templates/oauth2/error.html
similarity index 100%
rename from uffd/oauth2/templates/error.html
rename to uffd/oauth2/templates/oauth2/error.html
diff --git a/uffd/oauth2/templates/logout.html b/uffd/oauth2/templates/oauth2/logout.html
similarity index 100%
rename from uffd/oauth2/templates/logout.html
rename to uffd/oauth2/templates/oauth2/logout.html
diff --git a/uffd/oauth2/views.py b/uffd/oauth2/views.py
index 3ca1b9d96bde823d65141781e1302d0e7b3c2108..12e62febde73d9f22b56a0b2620b0222535fffee 100644
--- a/uffd/oauth2/views.py
+++ b/uffd/oauth2/views.py
@@ -7,7 +7,7 @@ from flask import Blueprint, request, jsonify, render_template, session, redirec
from flask_oauthlib.provider import OAuth2Provider
from uffd.database import db
-from uffd.session.views import get_current_user, login_required
+from uffd.session.views import login_required
from .models import OAuth2Client, OAuth2Grant, OAuth2Token
oauth = OAuth2Provider()
@@ -23,7 +23,7 @@ def load_grant(client_id, code):
@oauth.grantsetter
def save_grant(client_id, code, oauthreq, *args, **kwargs): # pylint: disable=unused-argument
expires = datetime.datetime.utcnow() + datetime.timedelta(seconds=100)
- grant = OAuth2Grant(user_dn=get_current_user().dn, client_id=client_id,
+ grant = OAuth2Grant(user_dn=request.user.dn, client_id=client_id,
code=code['code'], redirect_uri=oauthreq.redirect_uri, expires=expires, _scopes=' '.join(oauthreq.scopes))
db.session.add(grant)
db.session.commit()
@@ -89,7 +89,7 @@ def authorize(*args, **kwargs): # pylint: disable=unused-argument
session['oauth2-clients'] = session.get('oauth2-clients', [])
if client.client_id not in session['oauth2-clients']:
session['oauth2-clients'].append(client.client_id)
- return client.access_allowed(get_current_user())
+ return client.access_allowed(request.user)
@bp.route('/token', methods=['GET', 'POST'])
@oauth.token_handler
@@ -117,7 +117,7 @@ def error():
args = dict(request.values)
err = args.pop('error', 'unknown')
error_description = args.pop('error_description', '')
- return render_template('error.html', error=err, error_description=error_description, args=args)
+ return render_template('oauth2/error.html', error=err, error_description=error_description, args=args)
@bp.app_url_defaults
def inject_logout_params(endpoint, values):
@@ -131,4 +131,4 @@ def logout():
return redirect(request.values.get('ref', '/'))
client_ids = request.values['client_ids'].split(',')
clients = [OAuth2Client.from_id(client_id) for client_id in client_ids]
- return render_template('logout.html', clients=clients)
+ return render_template('oauth2/logout.html', clients=clients)
diff --git a/uffd/role/models.py b/uffd/role/models.py
index 4839b4ffe924dbe94d21bc42e3ece710f352c2de..dad00f53800a3115812f8d57e4d718c5dcb06240 100644
--- a/uffd/role/models.py
+++ b/uffd/role/models.py
@@ -1,4 +1,4 @@
-from sqlalchemy import Column, String, Integer, Text, ForeignKey
+from sqlalchemy import Column, String, Integer, Text, ForeignKey, Boolean
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declared_attr
@@ -91,6 +91,11 @@ class Role(db.Model):
db_groups = relationship("RoleGroup", backref="role", cascade="all, delete-orphan")
groups = DBRelationship('db_groups', Group, RoleGroup, backattr='role', backref='roles')
+ # Roles that are managed externally (e.g. by Ansible) can be locked to
+ # prevent accidental editing of name, moderator group, included roles
+ # and groups as well as deletion in the web interface.
+ locked = Column(Boolean(), default=False, nullable=False)
+
@property
def indirect_members(self):
users = set()
diff --git a/uffd/role/templates/role_list.html b/uffd/role/templates/role/list.html
similarity index 100%
rename from uffd/role/templates/role_list.html
rename to uffd/role/templates/role/list.html
diff --git a/uffd/role/templates/role.html b/uffd/role/templates/role/show.html
similarity index 88%
rename from uffd/role/templates/role.html
rename to uffd/role/templates/role/show.html
index 8cf25a5d9c4cf0a310373e6781e536934a2740fe..81d96395bdf6b8f5c2580acd42fb0ea7948d4b52 100644
--- a/uffd/role/templates/role.html
+++ b/uffd/role/templates/role/show.html
@@ -1,13 +1,19 @@
{% extends 'base.html' %}
{% block body %}
+{% if role.locked %}
+<div class="alert alert-warning" role="alert">
+Name, moderator group, included roles and groups of this role are managed externally. <a href="{{ url_for("role.unlock", roleid=role.id) }}" class="alert-link">Unlock this role</a> to edit them at the risk of having your changes overwritten.
+</div>
+{% endif %}
+
<form action="{{ url_for("role.update", roleid=role.id) }}" method="POST">
<div class="align-self-center">
<div class="float-sm-right pb-2">
<button type="submit" class="btn btn-primary"><i class="fa fa-save" aria-hidden="true"></i> Save</button>
<a href="{{ url_for("role.index") }}" class="btn btn-secondary">Cancel</a>
{% if role.id %}
- <a href="{{ url_for("role.delete", roleid=role.id) }}" onClick="return confirm('Are you sure?');" class="btn btn-danger"><i class="fa fa-trash" aria-hidden="true"></i> Delete</a>
+ <a href="{{ url_for("role.delete", roleid=role.id) }}" onClick="return confirm('Are you sure?');" class="btn btn-danger {{ 'disabled' if role.locked }}"><i class="fa fa-trash" aria-hidden="true"></i> Delete</a>
{% else %}
<a href="#" class="btn btn-danger disabled"><i class="fa fa-trash" aria-hidden="true"></i> Delete</a>
{% endif %}
@@ -28,7 +34,7 @@
<div class="tab-pane fade show active" id="settings" role="tabpanel" aria-labelledby="settings-tab">
<div class="form-group col">
<label for="role-name">Role Name</label>
- <input type="text" class="form-control" id="role-name" name="name" value="{{ role.name or '' }}">
+ <input type="text" class="form-control" id="role-name" name="name" value="{{ role.name or '' }}" {{ 'disabled' if role.locked }}>
<small class="form-text text-muted">
</small>
</div>
@@ -40,7 +46,7 @@
</div>
<div class="form-group col">
<label for="moderator-group">Moderator Group</label>
- <select class="form-control" id="moderator-group" name="moderator-group">
+ <select class="form-control" id="moderator-group" name="moderator-group" {{ 'disabled' if role.locked }}>
<option value="" class="text-muted">No Moderator Group</option>
{% for group in groups %}
<option value="{{ group.dn }}" {{ 'selected' if group == role.moderator_group }}>{{ group.name }}</option>
@@ -82,7 +88,7 @@
<td>
<div class="form-check">
<input class="form-check-input" type="checkbox" id="include-role-{{ r.id }}-checkbox" name="include-role-{{ r.id }}" value="1" aria-label="enabled"
- {% if r == role %}disabled{% endif %}
+ {% if r == role or role.locked %}disabled{% endif %}
{% if r in role.included_roles %}checked{% endif %}>
</div>
</td>
@@ -121,7 +127,7 @@
<tr id="group-{{ group.gid }}">
<td>
<div class="form-check">
- <input class="form-check-input" type="checkbox" id="group-{{ group.gid }}-checkbox" name="group-{{ group.gid }}" value="1" aria-label="enabled" {% if group in role.groups %}checked{% endif %}>
+ <input class="form-check-input" type="checkbox" id="group-{{ group.gid }}-checkbox" name="group-{{ group.gid }}" value="1" aria-label="enabled" {% if group in role.groups %}checked{% endif %} {{ 'disabled' if role.locked }}>
</div>
</td>
<td>
diff --git a/uffd/role/views.py b/uffd/role/views.py
index be6ecff5897fe2279f547da1ee9aeb00087ba2d4..a613a191522abe99499b860c20d98954cd8a68c8 100644
--- a/uffd/role/views.py
+++ b/uffd/role/views.py
@@ -7,7 +7,7 @@ from uffd.navbar import register_navbar
from uffd.csrf import csrf_protect
from uffd.role.models import Role
from uffd.user.models import User, Group
-from uffd.session import get_current_user, login_required, is_valid_session
+from uffd.session import login_required
from uffd.database import db
from uffd.ldap import ldap
@@ -44,23 +44,23 @@ def role_acl(): #pylint: disable=inconsistent-return-statements
return redirect(url_for('index'))
def role_acl_check():
- return is_valid_session() and get_current_user().is_in_group(current_app.config['ACL_ADMIN_GROUP'])
+ return request.user and request.user.is_in_group(current_app.config['ACL_ADMIN_GROUP'])
@bp.route("/")
@register_navbar('Roles', icon='key', blueprint=bp, visible=role_acl_check)
def index():
- return render_template('role_list.html', roles=Role.query.all())
+ return render_template('role/list.html', roles=Role.query.all())
@bp.route("/new")
def new():
- return render_template('role.html', role=Role(), groups=Group.query.all(), roles=Role.query.all())
+ return render_template('role/show.html', role=Role(), groups=Group.query.all(), roles=Role.query.all())
@bp.route("/<int:roleid>")
def show(roleid=None):
# prefetch all users so the ldap orm can cache them and doesn't run one ldap query per user
User.query.all()
role = Role.query.filter_by(id=roleid).one()
- return render_template('role.html', role=role, groups=Group.query.all(), roles=Role.query.all())
+ return render_template('role/show.html', role=role, groups=Group.query.all(), roles=Role.query.all())
@bp.route("/<int:roleid>/update", methods=['POST'])
@bp.route("/new", methods=['POST'])
@@ -71,22 +71,23 @@ def update(roleid=None):
db.session.add(role)
else:
role = Role.query.filter_by(id=roleid).one()
- role.name = request.values['name']
role.description = request.values['description']
- if not request.values['moderator-group']:
- role.moderator_group_dn = None
- else:
- role.moderator_group = Group.query.get(request.values['moderator-group'])
- for included_role in Role.query.all():
- if included_role != role and request.values.get('include-role-{}'.format(included_role.id)):
- role.included_roles.append(included_role)
- elif included_role in role.included_roles:
- role.included_roles.remove(included_role)
- for group in Group.query.all():
- if request.values.get('group-{}'.format(group.gid), False):
- role.groups.add(group)
+ if not role.locked:
+ role.name = request.values['name']
+ if not request.values['moderator-group']:
+ role.moderator_group_dn = None
else:
- role.groups.discard(group)
+ role.moderator_group = Group.query.get(request.values['moderator-group'])
+ for included_role in Role.query.all():
+ if included_role != role and request.values.get('include-role-{}'.format(included_role.id)):
+ role.included_roles.append(included_role)
+ elif included_role in role.included_roles:
+ role.included_roles.remove(included_role)
+ for group in Group.query.all():
+ if request.values.get('group-{}'.format(group.gid), False):
+ role.groups.add(group)
+ else:
+ role.groups.discard(group)
role.update_member_groups()
db.session.commit()
ldap.session.commit()
@@ -96,6 +97,9 @@ def update(roleid=None):
@csrf_protect(blueprint=bp)
def delete(roleid):
role = Role.query.filter_by(id=roleid).one()
+ if role.locked:
+ flash('Locked roles cannot be deleted')
+ return redirect(url_for('role.show', roleid=role.id))
oldmembers = set(role.members).union(role.indirect_members)
role.members.clear()
db.session.delete(role)
@@ -104,3 +108,11 @@ def delete(roleid):
db.session.commit()
ldap.session.commit()
return redirect(url_for('role.index'))
+
+@bp.route("/<int:roleid>/unlock")
+@csrf_protect(blueprint=bp)
+def unlock(roleid):
+ role = Role.query.filter_by(id=roleid).one()
+ role.locked = False
+ db.session.commit()
+ return redirect(url_for('role.show', roleid=role.id))
diff --git a/uffd/rolemod/views.py b/uffd/rolemod/views.py
index 174d923ba15be0db5375c67b478c343a9234084f..d9c5f83482d6764a16b8f25cc2283d831893d29a 100644
--- a/uffd/rolemod/views.py
+++ b/uffd/rolemod/views.py
@@ -4,14 +4,14 @@ from uffd.navbar import register_navbar
from uffd.csrf import csrf_protect
from uffd.role.models import Role
from uffd.user.models import User
-from uffd.session import get_current_user, login_required, is_valid_session
+from uffd.session import login_required
from uffd.database import db
from uffd.ldap import ldap
bp = Blueprint('rolemod', __name__, template_folder='templates', url_prefix='/rolemod/')
def user_is_rolemod():
- return is_valid_session() and Role.query.filter(Role.moderator_group_dn.in_(get_current_user().group_dns)).count()
+ return request.user and Role.query.filter(Role.moderator_group_dn.in_(request.user.group_dns)).count()
@bp.before_request
@login_required()
@@ -23,7 +23,7 @@ def acl_check(): #pylint: disable=inconsistent-return-statements
@bp.route("/")
@register_navbar('Moderation', icon='user-lock', blueprint=bp, visible=user_is_rolemod)
def index():
- roles = Role.query.filter(Role.moderator_group_dn.in_(get_current_user().group_dns)).all()
+ roles = Role.query.filter(Role.moderator_group_dn.in_(request.user.group_dns)).all()
return render_template('rolemod/list.html', roles=roles)
@bp.route("/<int:role_id>")
@@ -31,7 +31,7 @@ def show(role_id):
# prefetch all users so the ldap orm can cache them and doesn't run one ldap query per user
User.query.all()
role = Role.query.get_or_404(role_id)
- if role.moderator_group not in get_current_user().groups:
+ if role.moderator_group not in request.user.groups:
flash('Access denied')
return redirect(url_for('index'))
return render_template('rolemod/show.html', role=role)
@@ -40,7 +40,7 @@ def show(role_id):
@csrf_protect(blueprint=bp)
def update(role_id):
role = Role.query.get_or_404(role_id)
- if role.moderator_group not in get_current_user().groups:
+ if role.moderator_group not in request.user.groups:
flash('Access denied')
return redirect(url_for('index'))
if request.form['description'] != role.description:
@@ -55,7 +55,7 @@ def update(role_id):
@csrf_protect(blueprint=bp)
def delete_member(role_id, member_dn):
role = Role.query.get_or_404(role_id)
- if role.moderator_group not in get_current_user().groups:
+ if role.moderator_group not in request.user.groups:
flash('Access denied')
return redirect(url_for('index'))
member = User.query.get_or_404(member_dn)
diff --git a/uffd/selfservice/templates/forgot_password.html b/uffd/selfservice/templates/selfservice/forgot_password.html
similarity index 100%
rename from uffd/selfservice/templates/forgot_password.html
rename to uffd/selfservice/templates/selfservice/forgot_password.html
diff --git a/uffd/selfservice/templates/mailverification.mail.txt b/uffd/selfservice/templates/selfservice/mailverification.mail.txt
similarity index 100%
rename from uffd/selfservice/templates/mailverification.mail.txt
rename to uffd/selfservice/templates/selfservice/mailverification.mail.txt
diff --git a/uffd/selfservice/templates/newuser.mail.txt b/uffd/selfservice/templates/selfservice/newuser.mail.txt
similarity index 100%
rename from uffd/selfservice/templates/newuser.mail.txt
rename to uffd/selfservice/templates/selfservice/newuser.mail.txt
diff --git a/uffd/selfservice/templates/passwordreset.mail.txt b/uffd/selfservice/templates/selfservice/passwordreset.mail.txt
similarity index 100%
rename from uffd/selfservice/templates/passwordreset.mail.txt
rename to uffd/selfservice/templates/selfservice/passwordreset.mail.txt
diff --git a/uffd/selfservice/templates/self.html b/uffd/selfservice/templates/selfservice/self.html
similarity index 84%
rename from uffd/selfservice/templates/self.html
rename to uffd/selfservice/templates/selfservice/self.html
index 7cb7bf5652eaea2a3904d28fefbd02609498fc99..e3ce9467ef211d38109cea4a0839edf4698f595f 100644
--- a/uffd/selfservice/templates/self.html
+++ b/uffd/selfservice/templates/selfservice/self.html
@@ -40,6 +40,22 @@
<label for="user-password2">Password Repeat</label>
<input type="password" class="form-control" id="user-password2" name="password2" placeholder="●●●●●●●●">
</div>
+ <div class="form-group">
+ {% if user.roles_recursive|length %}
+ {% if user.roles_recursive|length == 1 %}
+ You have this role:
+ {% else %}
+ You currently have these roles:
+ {% endif %}
+ <ul>
+ {% for role in user.roles_recursive|sort(attribute="name") %}
+ <li>{{ role.name }}</li>
+ {% endfor %}
+ </ul>
+ {% else %}
+ You currently don't have any roles.
+ {% endif %}
+ </div>
<div class="form-group col-md-12">
<button type="submit" class="btn btn-primary float-right"><i class="fa fa-save" aria-hidden="true"></i> Save</button>
</div>
diff --git a/uffd/selfservice/templates/set_password.html b/uffd/selfservice/templates/selfservice/set_password.html
similarity index 100%
rename from uffd/selfservice/templates/set_password.html
rename to uffd/selfservice/templates/selfservice/set_password.html
diff --git a/uffd/selfservice/views.py b/uffd/selfservice/views.py
index 44556f363fb5273fe2decd47296e6d6e0cec588a..d469ef69310e27f1f7c60a47293fb923a96effb5 100644
--- a/uffd/selfservice/views.py
+++ b/uffd/selfservice/views.py
@@ -9,7 +9,7 @@ from flask import Blueprint, render_template, request, url_for, redirect, flash,
from uffd.navbar import register_navbar
from uffd.csrf import csrf_protect
from uffd.user.models import User
-from uffd.session import get_current_user, login_required, is_valid_session
+from uffd.session import login_required
from uffd.selfservice.models import PasswordToken, MailToken
from uffd.database import db
from uffd.ldap import ldap
@@ -20,17 +20,17 @@ bp = Blueprint("selfservice", __name__, template_folder='templates', url_prefix=
reset_ratelimit = Ratelimit('passwordreset', 1*60*60, 3)
@bp.route("/")
-@register_navbar('Selfservice', icon='portrait', blueprint=bp, visible=is_valid_session)
+@register_navbar('Selfservice', icon='portrait', blueprint=bp, visible=lambda: bool(request.user))
@login_required()
def index():
- return render_template('self.html', user=get_current_user())
+ return render_template('selfservice/self.html', user=request.user)
@bp.route("/update", methods=(['POST']))
@csrf_protect(blueprint=bp)
@login_required()
def update():
password_changed = False
- user = get_current_user()
+ user = request.user
if request.values['displayname'] != user.displayname:
if user.set_displayname(request.values['displayname']):
flash('Display name changed.')
@@ -57,7 +57,7 @@ def update():
@bp.route("/passwordreset", methods=(['GET', 'POST']))
def forgot_password():
if request.method == 'GET':
- return render_template('forgot_password.html')
+ return render_template('selfservice/forgot_password.html')
loginname = request.values['loginname']
mail = request.values['mail']
@@ -87,17 +87,17 @@ def token_password(token):
db.session.commit()
return redirect(url_for('session.login'))
if request.method == 'GET':
- return render_template('set_password.html', token=token)
+ return render_template('selfservice/set_password.html', token=token)
if not request.values['password1']:
flash('You need to set a password, please try again.')
- return render_template('set_password.html', token=token)
+ return render_template('selfservice/set_password.html', token=token)
if not request.values['password1'] == request.values['password2']:
flash('Passwords do not match, please try again.')
- return render_template('set_password.html', token=token)
+ return render_template('selfservice/set_password.html', token=token)
user = User.query.filter_by(loginname=dbtoken.loginname).one()
if not user.set_password(request.values['password1']):
flash('Password ist not valid, please try again.')
- return render_template('set_password.html', token=token)
+ return render_template('selfservice/set_password.html', token=token)
db.session.delete(dbtoken)
flash('New password set')
ldap.session.commit()
@@ -137,7 +137,7 @@ def send_mail_verification(loginname, newmail):
user = User.query.filter_by(loginname=loginname).one()
msg = EmailMessage()
- msg.set_content(render_template('mailverification.mail.txt', user=user, token=token.token))
+ msg.set_content(render_template('selfservice/mailverification.mail.txt', user=user, token=token.token))
msg['Subject'] = 'Mail verification'
send_mail(newmail, msg)
@@ -153,10 +153,10 @@ def send_passwordreset(user, new=False):
msg = EmailMessage()
if new:
- msg.set_content(render_template('newuser.mail.txt', user=user, token=token.token))
+ msg.set_content(render_template('selfservice/newuser.mail.txt', user=user, token=token.token))
msg['Subject'] = 'Welcome to the CCCV infrastructure'
else:
- msg.set_content(render_template('passwordreset.mail.txt', user=user, token=token.token))
+ msg.set_content(render_template('selfservice/passwordreset.mail.txt', user=user, token=token.token))
msg['Subject'] = 'Password reset'
send_mail(user.mail, msg)
diff --git a/uffd/services/templates/overview.html b/uffd/services/templates/services/overview.html
similarity index 100%
rename from uffd/services/templates/overview.html
rename to uffd/services/templates/services/overview.html
diff --git a/uffd/services/views.py b/uffd/services/views.py
index f4f9a805156966300b07748ac68f3dbd17a83d36..585de42c7955661d64ea74715d836ea2c6e2eabf 100644
--- a/uffd/services/views.py
+++ b/uffd/services/views.py
@@ -1,7 +1,6 @@
-from flask import Blueprint, render_template, current_app, abort
+from flask import Blueprint, render_template, current_app, abort, request
from uffd.navbar import register_navbar
-from uffd.session import is_valid_session, get_current_user
bp = Blueprint("services", __name__, template_folder='templates', url_prefix='/services')
@@ -69,25 +68,19 @@ def get_services(user=None):
return services
def services_visible():
- user = None
- if is_valid_session():
- user = get_current_user()
- return len(get_services(user)) > 0
+ return len(get_services(request.user)) > 0
@bp.route("/")
@register_navbar('Services', icon='sitemap', blueprint=bp, visible=services_visible)
def index():
- user = None
- if is_valid_session():
- user = get_current_user()
- services = get_services(user)
+ services = get_services(request.user)
if not current_app.config['SERVICES']:
abort(404)
banner = current_app.config.get('SERVICES_BANNER')
# Set the banner to None if it is not public and no user is logged in
- if not (current_app.config["SERVICES_BANNER_PUBLIC"] or user):
+ if not (current_app.config["SERVICES_BANNER_PUBLIC"] or request.user):
banner = None
- return render_template('overview.html', user=user, services=services, banner=banner)
+ return render_template('services/overview.html', user=request.user, services=services, banner=banner)
diff --git a/uffd/session/__init__.py b/uffd/session/__init__.py
index 5cddcdc6892a9c2e0a85e1047733432499b6d171..0e571f3a6aec00dcb8dd2d7eff8788441b735f3d 100644
--- a/uffd/session/__init__.py
+++ b/uffd/session/__init__.py
@@ -1,3 +1,3 @@
-from .views import bp as bp_ui, get_current_user, login_required, is_valid_session, set_session
+from .views import bp as bp_ui, login_required, set_session
bp = [bp_ui]
diff --git a/uffd/session/templates/login.html b/uffd/session/templates/session/login.html
similarity index 100%
rename from uffd/session/templates/login.html
rename to uffd/session/templates/session/login.html
diff --git a/uffd/session/views.py b/uffd/session/views.py
index 02a70d4e5edcbb42dfd84d24b2d99ee31b586538..2c57bc6df93c5ffae966ee219759c5a6c3cdd932 100644
--- a/uffd/session/views.py
+++ b/uffd/session/views.py
@@ -12,6 +12,21 @@ bp = Blueprint("session", __name__, template_folder='templates', url_prefix='/')
login_ratelimit = Ratelimit('login', 1*60, 3)
+@bp.before_app_request
+def set_request_user():
+ request.user = None
+ request.user_pre_mfa = None
+ if 'user_dn' not in session:
+ return
+ if 'logintime' not in session:
+ return
+ if datetime.datetime.now().timestamp() > session['logintime'] + current_app.config['SESSION_LIFETIME_SECONDS']:
+ return
+ user = User.query.get(session['user_dn'])
+ request.user_pre_mfa = user
+ if session.get('user_mfa'):
+ request.user = user
+
def login_get_user(loginname, password):
dn = User(loginname=loginname).dn
@@ -60,7 +75,7 @@ def set_session(user, password='', skip_mfa=False):
@bp.route("/login", methods=('GET', 'POST'))
def login():
if request.method == 'GET':
- return render_template('login.html', ref=request.values.get('ref'))
+ return render_template('session/login.html', ref=request.values.get('ref'))
username = request.form['loginname']
password = request.form['password']
@@ -71,47 +86,24 @@ def login():
flash('We received too many invalid login attempts for this user! Please wait at least %s.'%format_delay(login_delay))
else:
flash('We received too many requests from your ip address/network! Please wait at least %s.'%format_delay(host_delay))
- return render_template('login.html', ref=request.values.get('ref'))
+ return render_template('session/login.html', ref=request.values.get('ref'))
user = login_get_user(username, password)
if user is None:
login_ratelimit.log(username)
host_ratelimit.log()
flash('Login name or password is wrong')
- return render_template('login.html', ref=request.values.get('ref'))
+ return render_template('session/login.html', ref=request.values.get('ref'))
if not user.is_in_group(current_app.config['ACL_SELFSERVICE_GROUP']):
flash('You do not have access to this service')
- return render_template('login.html', ref=request.values.get('ref'))
+ return render_template('session/login.html', ref=request.values.get('ref'))
set_session(user, password=password)
return redirect(url_for('mfa.auth', ref=request.values.get('ref', url_for('index'))))
-def get_current_user():
- if 'user_dn' not in session:
- return None
- return User.query.get(session['user_dn'])
-bp.add_app_template_global(get_current_user)
-
-def login_valid():
- user = get_current_user()
- if user is None:
- return False
- if datetime.datetime.now().timestamp() > session['logintime'] + current_app.config['SESSION_LIFETIME_SECONDS']:
- return False
- return True
-
-def is_valid_session():
- if not login_valid():
- return False
- if not session.get('user_mfa'):
- return False
- return True
-bp.add_app_template_global(is_valid_session)
-
-def pre_mfa_login_required(no_redirect=False):
+def login_required_pre_mfa(no_redirect=False):
def wrapper(func):
@functools.wraps(func)
def decorator(*args, **kwargs):
- if not login_valid() or datetime.datetime.now().timestamp() > session['logintime'] + 10*60:
- session.clear()
+ if not request.user_pre_mfa:
if no_redirect:
abort(403)
flash('You need to login first')
@@ -124,12 +116,12 @@ def login_required(group=None):
def wrapper(func):
@functools.wraps(func)
def decorator(*args, **kwargs):
- if not login_valid():
+ if not request.user_pre_mfa:
flash('You need to login first')
return redirect(url_for('session.login', ref=request.url))
- if not session.get('user_mfa'):
+ if not request.user:
return redirect(url_for('mfa.auth', ref=request.url))
- if not get_current_user().is_in_group(group):
+ if not request.user.is_in_group(group):
flash('Access denied')
return redirect(url_for('index'))
return func(*args, **kwargs)
diff --git a/uffd/templates/base.html b/uffd/templates/base.html
index 023c3328aa64a18978935b04335aea2b4e92bab3..807af010be4f73c6d3ad2a8e38449c1cbbb4b6a8 100644
--- a/uffd/templates/base.html
+++ b/uffd/templates/base.html
@@ -67,7 +67,7 @@
</li>
{% endfor %}
</ul>
- {% if is_valid_session() %}
+ {% if request.user %}
<ul class="navbar-nav ml-auto">
<li class="nav-item">
<a class="nav-link" href="{{ url_for("session.logout") }}">
diff --git a/uffd/user/models.py b/uffd/user/models.py
index c0ce603e3d39fdf4dd49b4926ff9f957fefdcd93..2749884f85f9e77ed2f5a71cde3657f5e412dd46 100644
--- a/uffd/user/models.py
+++ b/uffd/user/models.py
@@ -8,13 +8,18 @@ from ldap3.utils.hashed import hashed, HASHED_SALTED_SHA512
from uffd.ldap import ldap
from uffd.lazyconfig import lazyconfig_str, lazyconfig_list
-def get_next_uid():
- max_uid = current_app.config['LDAP_USER_MIN_UID']
+def get_next_uid(service=False):
+ if service:
+ new_uid_min = current_app.config['LDAP_USER_SERVICE_MIN_UID']
+ new_uid_max = current_app.config['LDAP_USER_SERVICE_MAX_UID']
+ else:
+ new_uid_min = current_app.config['LDAP_USER_MIN_UID']
+ new_uid_max = current_app.config['LDAP_USER_MAX_UID']
+ next_uid = new_uid_min
for user in User.query.all():
- if user.uid <= current_app.config['LDAP_USER_MAX_UID']:
- max_uid = max(user.uid, max_uid)
- next_uid = max_uid + 1
- if next_uid > current_app.config['LDAP_USER_MAX_UID']:
+ if user.uid <= new_uid_max:
+ next_uid = max(next_uid, user.uid + 1)
+ if next_uid > new_uid_max:
raise Exception('No free uid found')
return next_uid
@@ -50,6 +55,18 @@ class BaseUser(ldap.Model):
def group_dns(self):
return [group.dn for group in self.groups]
+ @property
+ def is_service_user(self):
+ if self.uid is None:
+ return None
+ return self.uid >= current_app.config['LDAP_USER_SERVICE_MIN_UID'] and self.uid <= current_app.config['LDAP_USER_SERVICE_MAX_UID']
+
+ @is_service_user.setter
+ def is_service_user(self, value):
+ assert self.uid is None
+ if value:
+ self.uid = get_next_uid(service=True)
+
def add_default_attributes(self):
for name, values in current_app.config['LDAP_USER_DEFAULT_ATTRIBUTES'].items():
if self.ldap_object.getattr(name):
diff --git a/uffd/user/templates/group_list.html b/uffd/user/templates/group/list.html
similarity index 100%
rename from uffd/user/templates/group_list.html
rename to uffd/user/templates/group/list.html
diff --git a/uffd/user/templates/group.html b/uffd/user/templates/group/show.html
similarity index 100%
rename from uffd/user/templates/group.html
rename to uffd/user/templates/group/show.html
diff --git a/uffd/user/templates/user_list.html b/uffd/user/templates/user/list.html
similarity index 94%
rename from uffd/user/templates/user_list.html
rename to uffd/user/templates/user/list.html
index ddcaab3e89023d8fce96db1bf59a7b45bf5eeccd..93d7da5264b7e5e1625ed6f724bd9f2e7cff23d3 100644
--- a/uffd/user/templates/user_list.html
+++ b/uffd/user/templates/user/list.html
@@ -30,12 +30,15 @@
<a href="{{ url_for("user.show", uid=user.uid) }}">
{{ user.loginname }}
</a>
+ {% if user.is_service_user %}
+ <span class="badge badge-secondary">service</span>
+ {% endif %}
</td>
<td>
{{ user.displayname }}
</td>
<td>
- {% for role in user.roles|sort(attribute="name") if not role.name in config["ROLES_BASEROLES"] %}
+ {% for role in user.roles|sort(attribute="name") if not role.name in config["ROLES_BASEROLES"] or user.is_service_user %}
<a href="{{ url_for("role.show", roleid=role.id) }}">{{ role.name }}</a>{% if not loop.last %}, {% endif %}
{% endfor %}
</td>
diff --git a/uffd/user/templates/user.html b/uffd/user/templates/user/show.html
similarity index 89%
rename from uffd/user/templates/user.html
rename to uffd/user/templates/user/show.html
index b2feedf9a0417e59636d3c19bc8303342a211ea2..d638f7fc287820a287189010c48f7e6577b96849 100644
--- a/uffd/user/templates/user.html
+++ b/uffd/user/templates/user/show.html
@@ -27,13 +27,26 @@
<div class="tab-content border mb-2 pt-2" id="tabcontent">
<div class="tab-pane fade show active" id="profile" role="tabpanel" aria-labelledby="roles-tab">
<div class="form-group col">
- <label for="user-uid">uid</label>
+ <label for="user-uid">
+ uid
+ {% if user.is_service_user %}
+ <span class="badge badge-secondary">service</span>
+ {% endif %}
+ </label>
{% if user.uid %}
<input type="number" class="form-control" id="user-uid" name="uid" value="{{ user.uid or '' }}" readonly>
{% else %}
<input type="text" class="form-control" id="user-uid" name="uid" placeholder="will be choosen" readonly>
{% endif %}
</div>
+ {% if not user.uid %}
+ <div class="form-group col">
+ <div class="form-check">
+ <input class="form-check-input" type="checkbox" id="user-serviceaccount" name="serviceaccount" value="1" aria-label="enabled">
+ <label class="form-check-label" for="user-serviceaccount">Service User</label>
+ </div>
+ </div>
+ {% endif %}
<div class="form-group col">
<label for="user-loginname">Login Name</label>
<input type="text" class="form-control" id="user-loginname" name="loginname" value="{{ user.loginname or '' }}" {% if user.uid %}readonly{% endif %}>
@@ -92,8 +105,8 @@
<td>
<div class="form-check">
<input class="form-check-input" type="checkbox" id="role-{{ role.id }}-checkbox" name="role-{{ role.id }}" value="1" aria-label="enabled"
- {% if user in role.members or role.name in config["ROLES_BASEROLES"] %}checked {% endif %}
- {% if role.name in config["ROLES_BASEROLES"] %}disabled {% endif %}>
+ {% if user in role.members %}checked {% endif %}
+ {% if role.name in config["ROLES_BASEROLES"] and not user.is_service_user %}disabled {% endif %}>
</div>
</td>
<td>
diff --git a/uffd/user/views_group.py b/uffd/user/views_group.py
index 22fdd9e9d9def67a36fb5c18e2351878e61ebc0f..d4318b3cd86df28be912e3f7bcd0351bc4bed16a 100644
--- a/uffd/user/views_group.py
+++ b/uffd/user/views_group.py
@@ -1,7 +1,7 @@
-from flask import Blueprint, render_template, url_for, redirect, flash, current_app
+from flask import Blueprint, render_template, url_for, redirect, flash, current_app, request
from uffd.navbar import register_navbar
-from uffd.session import login_required, is_valid_session, get_current_user
+from uffd.session import login_required
from .models import Group
@@ -14,13 +14,13 @@ def group_acl(): #pylint: disable=inconsistent-return-statements
return redirect(url_for('index'))
def group_acl_check():
- return is_valid_session() and get_current_user().is_in_group(current_app.config['ACL_ADMIN_GROUP'])
+ return request.user and request.user.is_in_group(current_app.config['ACL_ADMIN_GROUP'])
@bp.route("/")
@register_navbar('Groups', icon='layer-group', blueprint=bp, visible=group_acl_check)
def index():
- return render_template('group_list.html', groups=Group.query.all())
+ return render_template('group/list.html', groups=Group.query.all())
@bp.route("/<int:gid>")
def show(gid):
- return render_template('group.html', group=Group.query.filter_by(gid=gid).first_or_404())
+ return render_template('group/show.html', group=Group.query.filter_by(gid=gid).first_or_404())
diff --git a/uffd/user/views_user.py b/uffd/user/views_user.py
index e25f89e7df1a197701ca17c8a52dc33c37b7bf77..a76beceb53ff8defc75b62b74db756ec9dbc981a 100644
--- a/uffd/user/views_user.py
+++ b/uffd/user/views_user.py
@@ -6,7 +6,7 @@ from flask import Blueprint, render_template, request, url_for, redirect, flash,
from uffd.navbar import register_navbar
from uffd.csrf import csrf_protect
from uffd.selfservice import send_passwordreset
-from uffd.session import login_required, is_valid_session, get_current_user
+from uffd.session import login_required
from uffd.role.models import Role
from uffd.database import db
from uffd.ldap import ldap, LDAPCommitError
@@ -22,18 +22,18 @@ def user_acl(): #pylint: disable=inconsistent-return-statements
return redirect(url_for('index'))
def user_acl_check():
- return is_valid_session() and get_current_user().is_in_group(current_app.config['ACL_ADMIN_GROUP'])
+ return request.user and request.user.is_in_group(current_app.config['ACL_ADMIN_GROUP'])
@bp.route("/")
@register_navbar('Users', icon='users', blueprint=bp, visible=user_acl_check)
def index():
- return render_template('user_list.html', users=User.query.all())
+ return render_template('user/list.html', users=User.query.all())
@bp.route("/<int:uid>")
@bp.route("/new")
def show(uid=None):
user = User() if uid is None else User.query.filter_by(uid=uid).first_or_404()
- return render_template('user.html', user=user, roles=Role.query.all())
+ return render_template('user/show.html', user=user, roles=Role.query.all())
@bp.route("/<int:uid>/update", methods=['POST'])
@bp.route("/new", methods=['POST'])
@@ -42,6 +42,8 @@ def update(uid=None):
if uid is None:
user = User()
ignore_blacklist = request.form.get('ignore-loginname-blacklist', False)
+ if request.form.get('serviceaccount'):
+ user.is_service_user = True
if not user.set_loginname(request.form['loginname'], ignore_blacklist=ignore_blacklist):
flash('Login name does not meet requirements')
return redirect(url_for('user.show'))
@@ -60,14 +62,19 @@ def update(uid=None):
ldap.session.add(user)
user.roles.clear()
for role in Role.query.all():
- if request.values.get('role-{}'.format(role.id), False) or role.name in current_app.config["ROLES_BASEROLES"]:
+ if request.values.get('role-{}'.format(role.id), False):
+ user.roles.add(role)
+ elif not user.is_service_user and role.name in current_app.config["ROLES_BASEROLES"]:
user.roles.add(role)
user.update_groups()
ldap.session.commit()
db.session.commit()
if uid is None:
- send_passwordreset(user, new=True)
- flash('User created. We sent the user a password reset link by mail')
+ if user.is_service_user:
+ flash('Service user created')
+ else:
+ send_passwordreset(user, new=True)
+ flash('User created. We sent the user a password reset link by mail')
else:
flash('User updated')
return redirect(url_for('user.show', uid=user.uid))