diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 4663281930afa35a855b5fe1bc8ed4f8f3f1b7ed..ed8b364207d594f6a9a0baee846272a86b77e132 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -13,20 +13,46 @@ before_script:
linter:
stage: test
script:
- - python3 -m pylint --rcfile .pylintrc --output-format=text uffd | tee pylint.txt
+ - pip3 install pylint-gitlab # this force-updates jinja2 and some other packages!
+ - python3 -m pylint --exit-zero --rcfile .pylintrc --output-format=pylint_gitlab.GitlabCodeClimateReporter uffd > codeclimate.json
+ - python3 -m pylint --exit-zero --rcfile .pylintrc --output-format=pylint_gitlab.GitlabPagesHtmlReporter uffd > pylint.html
+ - python3 -m pylint --rcfile .pylintrc --output-format=text uffd
artifacts:
- paths:
- - pylint.txt
+ when: always
+ paths:
+ - pylint.html
+ reports:
+ codequality: codeclimate.json
-#unittest:
-# stage: test
-# script:
-# - python3 -m coverage run runTests.py
-# - python3 -m coverage report --include "./*"
-# - python3 -m coverage report -m --include "./*" > report.txt
-# - python3 -m coverage html --include "./*"
-# artifacts:
-# paths:
-# - htmlcov/*
-# - .coverage
-# - report.txt
+unittests:
+ stage: test
+ script:
+ - python3-coverage run --include './*.py' --omit 'tests/*.py' -m pytest --junitxml=report.xml
+ - python3-coverage report -m
+ - python3-coverage html
+ - python3-coverage xml
+ artifacts:
+ when: always
+ paths:
+ - htmlcov/index.html
+ - htmlcov
+ expose_as: 'Coverage Report'
+ reports:
+ cobertura: coverage.xml
+ junit: report.xml
+ coverage: '/^TOTAL.*\s+(\d+\%)$/'
+
+html5validator:
+ stage: test
+ script:
+ - rm -rf pages
+ - mkdir -p pages
+ - cp -r uffd/static pages/static
+ - DUMP_PAGES=pages python3 -m unittest discover tests
+ - sed -i -e 's/href="\/static\//href=".\/static\//g' -e 's/src="\/static\//src=".\/static\//g' pages/*.html
+ - html5validator --root pages 2>&1 | tee html5validator.log
+ artifacts:
+ when: on_failure
+ paths:
+ - pages
+ - html5validator.log
diff --git a/ldap_server_entries.json b/ldap_server_entries.json
index d52627fd93dcd22df248fe95137eb4c4b29c03b8..a6b5d8c58dd4ce37133a814b7aec70263877f613 100644
--- a/ldap_server_entries.json
+++ b/ldap_server_entries.json
@@ -4,7 +4,7 @@
"dn": "uid=testuser,ou=users,dc=example,dc=com",
"raw": {
"cn": [
- "testuser"
+ "Test User"
],
"createTimestamp": [
"20200101000000Z"
@@ -25,7 +25,7 @@
"20001"
],
"givenName": [
- "testuser"
+ "Test User"
],
"hasSubordinates": [
"FALSE"
@@ -77,7 +77,7 @@
"dn": "uid=testadmin,ou=users,dc=example,dc=com",
"raw": {
"cn": [
- "testadmin"
+ "Test Admin"
],
"createTimestamp": [
"20200101000000Z"
@@ -98,7 +98,7 @@
"20001"
],
"givenName": [
- "Testadmin"
+ "Test Admin"
],
"hasSubordinates": [
"FALSE"
@@ -296,6 +296,52 @@
"uid=testadmin,ou=users,dc=example,dc=com"
]
}
+ },
+ {
+ "dn": "uid=test,ou=postfix,dc=example,dc=com",
+ "raw": {
+ "createTimestamp": [
+ "20200101000000Z"
+ ],
+ "creatorsName": [
+ "cn=admin,dc=example,dc=com"
+ ],
+ "entryDN": [
+ "uid=test,ou=postfix,dc=example,dc=com"
+ ],
+ "entryUUID": [
+ "926e5273-a545-4dfe-8f20-d1eeaf41d796"
+ ],
+ "hasSubordinates": [
+ "FALSE"
+ ],
+ "mailacceptinggeneralid": [
+ "test1@example.com",
+ "test2@example.com"
+ ],
+ "maildrop": [
+ "testuser@mail.example.com"
+ ],
+ "modifiersName": [
+ "cn=admin,dc=example,dc=com"
+ ],
+ "modifyTimestamp": [
+ "20200101000000Z"
+ ],
+ "objectClass": [
+ "top",
+ "postfixVirtual"
+ ],
+ "structuralObjectClass": [
+ "postfixVirtual"
+ ],
+ "subschemaSubentry": [
+ "cn=Subschema"
+ ],
+ "uid": [
+ "test"
+ ]
+ }
}
]
}
diff --git a/tests/test_csrf.py b/tests/test_csrf.py
new file mode 100644
index 0000000000000000000000000000000000000000..1486beb74b7a5ce827cdae2a1bba168e4a990454
--- /dev/null
+++ b/tests/test_csrf.py
@@ -0,0 +1,102 @@
+import unittest
+
+from flask import Flask, Blueprint, session, url_for
+
+from uffd.csrf import csrf_bp, csrf_protect
+
+uid_counter = 0
+
+class TestCSRF(unittest.TestCase):
+ unprotected_ep = 'foo'
+ protected_ep = 'bar'
+
+ def setUp(self):
+ self.app = Flask(__name__)
+ self.app.testing = True
+ self.app.config['SECRET_KEY'] = 'DEBUGKEY'
+ self.app.register_blueprint(csrf_bp)
+
+ @self.app.route('/', methods=['GET', 'POST'])
+ def index():
+ return 'SUCCESS', 200
+
+ @self.app.route('/login', methods=['GET', 'POST'])
+ def login():
+ global uid_counter
+ session['_csrf_token'] = 'secret_csrf_token%d'%uid_counter
+ uid_counter += 1
+ return 'Ok', 200
+
+ @self.app.route('/logout', methods=['GET', 'POST'])
+ def logout():
+ session.clear()
+ return 'Ok', 200
+
+ @self.app.route('/foo', methods=['GET', 'POST'])
+ def foo():
+ return 'SUCCESS', 200
+
+ @self.app.route('/bar', methods=['GET', 'POST'])
+ @csrf_protect()
+ def bar():
+ return 'SUCCESS', 200
+
+ self.bp = Blueprint('bp', __name__)
+
+ @self.bp.route('/foo', methods=['GET', 'POST'])
+ @csrf_protect(blueprint=self.bp) # This time on .foo and not on .bar!
+ def foo():
+ return 'SUCCESS', 200
+
+ @self.bp.route('/bar', methods=['GET', 'POST'])
+ def bar():
+ return 'SUCCESS', 200
+
+ self.app.register_blueprint(self.bp, url_prefix='/bp/')
+ self.client = self.app.test_client()
+ self.client.__enter__()
+ # Just do some request so that we can use url_for
+ self.client.get(path='/')
+
+ def tearDown(self):
+ self.client.__exit__(None, None, None)
+
+ def set_token(self):
+ self.client.get(path='/login')
+
+ def clear_token(self):
+ self.client.get(path='/logout')
+
+ def test_notoken_unprotected(self):
+ url = url_for(self.unprotected_ep)
+ self.assertTrue('csrf' not in url)
+ self.assertEqual(self.client.get(path=url).data, b'SUCCESS')
+
+ def test_token_unprotected(self):
+ self.set_token()
+ self.test_notoken_unprotected()
+
+ def test_notoken_protected(self):
+ url = url_for(self.protected_ep)
+ self.assertNotEqual(self.client.get(path=url).data, b'SUCCESS')
+
+ def test_token_protected(self):
+ self.set_token()
+ url = url_for(self.protected_ep)
+ self.assertEqual(self.client.get(path=url).data, b'SUCCESS')
+
+ def test_wrong_token_protected(self):
+ self.set_token()
+ url = url_for(self.protected_ep)
+ self.set_token()
+ self.assertNotEqual(self.client.get(path=url).data, b'SUCCESS')
+
+ def test_deleted_token_protected(self):
+ self.set_token()
+ url = url_for(self.protected_ep)
+ self.clear_token()
+ self.assertNotEqual(self.client.get(path=url).data, b'SUCCESS')
+
+class TestBlueprintCSRF(TestCSRF):
+ unprotected_ep = 'bp.bar'
+ protected_ep = 'bp.foo'
diff --git a/tests/test_mail.py b/tests/test_mail.py
new file mode 100644
index 0000000000000000000000000000000000000000..cbe3c4abe3386fb2efcb5a7ddea922e87bf49f39
--- /dev/null
+++ b/tests/test_mail.py
@@ -0,0 +1,92 @@
+import datetime
+import time
+
+from flask import url_for, session
+
+# These imports are required, because otherwise we get circular imports?!
+from uffd import ldap, user
+
+from uffd.ldap import get_conn
+from uffd.mail.models import Mail
+from uffd import create_app, db
+
+from utils import dump, UffdTestCase
+
+def get_mail():
+ return Mail.from_ldap_dn('uid=test,ou=postfix,dc=example,dc=com')
+
+class TestMailViews(UffdTestCase):
+ def setUp(self):
+ super().setUp()
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testadmin', 'password': 'adminpassword'}, follow_redirects=True)
+
+ def test_index(self):
+ r = self.client.get(path=url_for('mail.index'), follow_redirects=True)
+ dump('mail_index', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_index_empty(self):
+ conn = get_conn()
+ conn.delete(get_mail().dn)
+ self.assertIsNone(get_mail())
+ r = self.client.get(path=url_for('mail.index'), follow_redirects=True)
+ dump('mail_index_empty', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_show(self):
+ r = self.client.get(path=url_for('mail.show', uid=get_mail().uid), follow_redirects=True)
+ dump('mail_show', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_new(self):
+ r = self.client.get(path=url_for('mail.show'), follow_redirects=True)
+ dump('mail_new', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_update(self):
+ m = get_mail()
+ self.assertIsNotNone(m)
+ self.assertEqual(m.uid, 'test')
+ self.assertEqual(sorted(m.receivers), ['test1@example.com', 'test2@example.com'])
+ self.assertEqual(sorted(m.destinations), ['testuser@mail.example.com'])
+ r = self.client.post(path=url_for('mail.update', uid=m.uid),
+ data={'mail-uid': 'test1', 'mail-receivers': 'foo@bar.com\ntest@bar.com',
+ 'mail-destinations': 'testuser@mail.example.com\ntestadmin@mail.example.com'}, follow_redirects=True)
+ dump('mail_update', r)
+ self.assertEqual(r.status_code, 200)
+ m = get_mail()
+ self.assertIsNotNone(m)
+ self.assertEqual(m.uid, 'test')
+ self.assertEqual(sorted(m.receivers), ['foo@bar.com', 'test@bar.com'])
+ self.assertEqual(sorted(m.destinations), ['testadmin@mail.example.com', 'testuser@mail.example.com'])
+
+ def test_create(self):
+ r = self.client.post(path=url_for('mail.update'),
+ data={'mail-uid': 'test1', 'mail-receivers': 'foo@bar.com\ntest@bar.com',
+ 'mail-destinations': 'testuser@mail.example.com\ntestadmin@mail.example.com'}, follow_redirects=True)
+ dump('mail_create', r)
+ self.assertEqual(r.status_code, 200)
+ m = Mail.from_ldap_dn('uid=test1,ou=postfix,dc=example,dc=com')
+ self.assertEqual(m.uid, 'test1')
+ self.assertEqual(sorted(m.receivers), ['foo@bar.com', 'test@bar.com'])
+ self.assertEqual(sorted(m.destinations), ['testadmin@mail.example.com', 'testuser@mail.example.com'])
+
+ def test_create_error(self):
+ r = self.client.post(path=url_for('mail.update'),
+ data={'mail-uid': 'test', 'mail-receivers': 'foo@bar.com\ntest@bar.com',
+ 'mail-destinations': 'testuser@mail.example.com\ntestadmin@mail.example.com'}, follow_redirects=True)
+ dump('mail_create_error', r)
+ self.assertEqual(r.status_code, 200)
+ m = get_mail()
+ self.assertIsNotNone(m)
+ self.assertEqual(m.uid, 'test')
+ self.assertEqual(sorted(m.receivers), ['test1@example.com', 'test2@example.com'])
+ self.assertEqual(sorted(m.destinations), ['testuser@mail.example.com'])
+
+ def test_delete(self):
+ self.assertIsNotNone(get_mail())
+ r = self.client.get(path=url_for('mail.delete', uid=get_mail().uid), follow_redirects=True)
+ dump('mail_delete', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(get_mail())
diff --git a/tests/test_mfa.py b/tests/test_mfa.py
new file mode 100644
index 0000000000000000000000000000000000000000..b42a20ae11ed9eea103b013f5e4b40ab4fe057b5
--- /dev/null
+++ b/tests/test_mfa.py
@@ -0,0 +1,425 @@
+import unittest
+import datetime
+import time
+
+from flask import url_for, session
+
+# 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
+
+from utils import dump, UffdTestCase
+
+class TestMfaPrimitives(unittest.TestCase):
+ def test_hotp(self):
+ self.assertEqual(_hotp(5555555, b'\xae\xa3T\x05\x89\xd6\xb76\xf61r\x92\xcc\xb5WZ\xe6)\x05q'), '458290')
+ self.assertEqual(_hotp(5555555, b'\xae\xa3T\x05\x89\xd6\xb76\xf61r\x92\xcc\xb5WZ\xe6)\x05q', digits=8), '20458290')
+ for digits in range(1, 10):
+ self.assertEqual(len(_hotp(1, b'abcd', digits=digits)), digits)
+ self.assertEqual(_hotp(1234, b''), '161024')
+ self.assertEqual(_hotp(0, b'\x04\x8fM\xcc\x7f\x82\x9c$a\x1b\xb3'), '279354')
+ self.assertEqual(_hotp(2**64-1, b'abcde'), '899292')
+
+def get_user():
+ return User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+
+def get_admin():
+ return User.from_ldap_dn('uid=testadmin,ou=users,dc=example,dc=com')
+
+def get_fido2_test_cred():
+ try:
+ from fido2.ctap2 import AttestedCredentialData
+ except ImportError:
+ self.skipTest('fido2 could not be imported')
+ # Example public key from webauthn spec 6.5.1.1
+ return AttestedCredentialData(bytes.fromhex('00000000000000000000000000000000'+'0040'+'053cbcc9d37a61d3bac87cdcc77ee326256def08ab15775d3a720332e4101d14fae95aeee3bc9698781812e143c0597dc6e180595683d501891e9dd030454c0a'+'A501020326200121582065eda5a12577c2bae829437fe338701a10aaa375e1bb5b5de108de439c08551d2258201e52ed75701163f7f9e40ddf9f341b3dc9ba860af7e0ca7ca7e9eecd0084d19c'))
+
+class TestMfaMethodModels(UffdTestCase):
+ def test_common_attributes(self):
+ method = MFAMethod(user=get_user(), name='testname')
+ self.assertTrue(method.created <= datetime.datetime.now())
+ self.assertEqual(method.name, 'testname')
+ self.assertEqual(method.user.loginname, 'testuser')
+ method.user = get_admin()
+ self.assertEqual(method.user.loginname, 'testadmin')
+
+ def test_recovery_code_method(self):
+ method = RecoveryCodeMethod(user=get_user())
+ db.session.add(method)
+ db.session.commit()
+ db.session = db.create_scoped_session() # Ensure the next query does not return the cached method object
+ _method = RecoveryCodeMethod.query.get(method.id)
+ self.assertFalse(hasattr(_method, 'code'))
+ self.assertFalse(_method.verify(''))
+ self.assertFalse(_method.verify('A'*8))
+ self.assertTrue(_method.verify(method.code))
+
+ def test_totp_method_attributes(self):
+ method = TOTPMethod(user=get_user(), name='testname')
+ self.assertEqual(method.name, 'testname')
+ # Restore method with key parameter
+ _method = TOTPMethod(user=get_user(), key=method.key, name='testname')
+ self.assertEqual(_method.name, 'testname')
+ self.assertEqual(method.raw_key, _method.raw_key)
+ self.assertEqual(method.issuer, _method.issuer)
+ self.assertEqual(method.accountname, _method.accountname)
+ self.assertEqual(method.key_uri, _method.key_uri)
+ db.session.add(method)
+ db.session.commit()
+ db.session = db.create_scoped_session() # Ensure the next query does not return the cached method object
+ # Restore method from db
+ _method = TOTPMethod.query.get(method.id)
+ self.assertEqual(_method.name, 'testname')
+ self.assertEqual(method.raw_key, _method.raw_key)
+ self.assertEqual(method.issuer, _method.issuer)
+ self.assertEqual(method.accountname, _method.accountname)
+ self.assertEqual(method.key_uri, _method.key_uri)
+
+ def test_totp_method_verify(self):
+ method = TOTPMethod(user=get_user())
+ counter = int(time.time()/30)
+ self.assertFalse(method.verify(''))
+ self.assertFalse(method.verify(_hotp(counter-2, method.raw_key)))
+ self.assertTrue(method.verify(_hotp(counter, method.raw_key)))
+ self.assertFalse(method.verify(_hotp(counter+2, method.raw_key)))
+
+ def test_webauthn_method(self):
+ data = get_fido2_test_cred()
+ method = WebauthnMethod(user=get_user(), cred=data, name='testname')
+ self.assertEqual(method.name, 'testname')
+ db.session.add(method)
+ db.session.commit()
+ db.session = db.create_scoped_session() # Ensure the next query does not return the cached method object
+ _method = WebauthnMethod.query.get(method.id)
+ self.assertEqual(_method.name, 'testname')
+ self.assertEqual(bytes(method.cred), bytes(_method.cred))
+ self.assertEqual(data.credential_id, _method.cred.credential_id)
+ self.assertEqual(data.public_key, _method.cred.public_key)
+ # We only test (de-)serialization here, as everything else is currently implemented in the views
+
+class TestMfaViews(UffdTestCase):
+ def setUp(self):
+ super().setUp()
+ db.session.add(RecoveryCodeMethod(user=get_admin()))
+ db.session.add(TOTPMethod(user=get_admin(), name='Admin Phone'))
+ # We don't want to skip all tests only because fido2 is not installed!
+ #db.session.add(WebauthnMethod(user=get_admin(), cred=get_fido2_test_cred(), name='Admin FIDO2 dongle'))
+ db.session.commit()
+
+ def login(self):
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
+
+ def add_recovery_codes(self, count=10):
+ user = get_user()
+ for _ in range(count):
+ db.session.add(RecoveryCodeMethod(user=user))
+ db.session.commit()
+
+ def add_totp(self):
+ db.session.add(TOTPMethod(user=get_user(), name='My phone'))
+ db.session.commit()
+
+ def add_webauthn(self):
+ db.session.add(WebauthnMethod(user=get_user(), cred=get_fido2_test_cred(), name='My FIDO2 dongle'))
+ db.session.commit()
+
+ def test_setup_disabled(self):
+ self.login()
+ r = self.client.get(path=url_for('mfa.setup'), follow_redirects=True)
+ dump('mfa_setup_disabled', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_setup_recovery_codes(self):
+ self.login()
+ self.add_recovery_codes()
+ r = self.client.get(path=url_for('mfa.setup'), follow_redirects=True)
+ dump('mfa_setup_only_recovery_codes', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_setup_enabled(self):
+ self.login()
+ self.add_recovery_codes()
+ self.add_totp()
+ self.add_webauthn()
+ r = self.client.get(path=url_for('mfa.setup'), follow_redirects=True)
+ dump('mfa_setup_enabled', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_setup_few_recovery_codes(self):
+ self.login()
+ self.add_totp()
+ self.add_recovery_codes(1)
+ r = self.client.get(path=url_for('mfa.setup'), follow_redirects=True)
+ dump('mfa_setup_few_recovery_codes', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_setup_no_recovery_codes(self):
+ self.login()
+ self.add_totp()
+ r = self.client.get(path=url_for('mfa.setup'), follow_redirects=True)
+ dump('mfa_setup_no_recovery_codes', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_disable(self):
+ self.login()
+ self.add_recovery_codes()
+ self.add_totp()
+ admin_methods = len(MFAMethod.query.filter_by(dn=get_admin().dn).all())
+ r = self.client.get(path=url_for('mfa.disable'), follow_redirects=True)
+ dump('mfa_disable', r)
+ self.assertEqual(r.status_code, 200)
+ 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=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)
+ 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=get_admin().dn).all()), admin_methods)
+
+ def test_admin_disable(self):
+ for method in MFAMethod.query.filter_by(dn=get_admin().dn).all():
+ if not isinstance(method, RecoveryCodeMethod):
+ db.session.delete(method)
+ db.session.commit()
+ self.add_recovery_codes()
+ self.add_totp()
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testadmin', 'password': 'adminpassword'}, follow_redirects=True)
+ self.assertTrue(is_valid_session())
+ 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)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(len(MFAMethod.query.filter_by(dn=get_user().dn).all()), 0)
+ self.assertEqual(len(MFAMethod.query.filter_by(dn=get_admin().dn).all()), admin_methods)
+
+ def test_setup_recovery(self):
+ self.login()
+ self.assertEqual(len(RecoveryCodeMethod.query.filter_by(dn=get_current_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()
+ self.assertNotEqual(len(methods), 0)
+ r = self.client.post(path=url_for('mfa.setup_recovery'), follow_redirects=True)
+ dump('mfa_setup_recovery_reset', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(len(RecoveryCodeMethod.query.filter_by(id=methods[0].id).all()), 0)
+ self.assertNotEqual(len(methods), 0)
+
+ def test_setup_totp(self):
+ self.login()
+ self.add_recovery_codes()
+ r = self.client.get(path=url_for('mfa.setup_totp', name='My TOTP Authenticator'), follow_redirects=True)
+ dump('mfa_setup_totp', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertNotEqual(len(session.get('mfa_totp_key', '')), 0)
+
+ def test_setup_totp_without_recovery(self):
+ self.login()
+ r = self.client.get(path=url_for('mfa.setup_totp', name='My TOTP Authenticator'), follow_redirects=True)
+ dump('mfa_setup_totp_without_recovery', r)
+ self.assertEqual(r.status_code, 200)
+
+ 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)
+ 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', ''))
+ 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)
+
+ def test_setup_totp_finish_without_recovery(self):
+ self.login()
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_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', ''))
+ 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)
+
+ 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)
+ 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', ''))
+ 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)
+
+ 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)
+ 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)
+
+ def test_delete_totp(self):
+ self.login()
+ self.add_recovery_codes()
+ self.add_totp()
+ method = TOTPMethod(get_current_user(), name='test')
+ db.session.add(method)
+ db.session.commit()
+ self.assertEqual(len(TOTPMethod.query.filter_by(dn=get_current_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)
+
+ # TODO: webauthn setup tests
+
+ def test_auth_integration(self):
+ self.add_recovery_codes()
+ self.add_totp()
+ db.session.commit()
+ self.assertFalse(is_valid_session())
+ 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())
+ 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())
+
+ def test_auth_disabled(self):
+ self.assertFalse(is_valid_session())
+ 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())
+
+ def test_auth_recovery_only(self):
+ self.add_recovery_codes()
+ self.assertFalse(is_valid_session())
+ 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())
+
+ def test_auth_recovery_code(self):
+ self.add_recovery_codes()
+ self.add_totp()
+ method = RecoveryCodeMethod(user=get_user())
+ db.session.add(method)
+ db.session.commit()
+ method_id = method.id
+ self.login()
+ 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())
+ 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.assertEqual(len(RecoveryCodeMethod.query.filter_by(id=method_id).all()), 0)
+
+ def test_auth_totp_code(self):
+ self.add_recovery_codes()
+ self.add_totp()
+ method = TOTPMethod(user=get_user(), name='testname')
+ raw_key = method.raw_key
+ db.session.add(method)
+ db.session.commit()
+ self.login()
+ 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())
+ 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())
+
+ def test_auth_empty_code(self):
+ self.add_recovery_codes()
+ self.add_totp()
+ 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())
+ 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())
+
+ def test_auth_invalid_code(self):
+ self.add_recovery_codes()
+ self.add_totp()
+ method = TOTPMethod(user=get_user(), name='testname')
+ raw_key = method.raw_key
+ db.session.add(method)
+ db.session.commit()
+ 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())
+ 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())
+
+ @unittest.skip('Not implemented, see #10')
+ def test_auth_ratelimit(self):
+ self.add_recovery_codes()
+ self.add_totp()
+ method = TOTPMethod(user=get_user(), name='testname')
+ raw_key = method.raw_key
+ db.session.add(method)
+ db.session.commit()
+ self.login()
+ self.assertFalse(is_valid_session())
+ 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())
+ 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())
+
+ # TODO: webauthn auth tests
diff --git a/tests/test_oauth2.py b/tests/test_oauth2.py
new file mode 100644
index 0000000000000000000000000000000000000000..c5d024eca7f63b9fb58538836d99bba8efcf66f9
--- /dev/null
+++ b/tests/test_oauth2.py
@@ -0,0 +1,110 @@
+import datetime
+from urllib.parse import urlparse, parse_qs
+
+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
+
+from utils import dump, UffdTestCase
+
+def get_user():
+ return User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+
+def get_admin():
+ return User.from_ldap_dn('uid=testadmin,ou=users,dc=example,dc=com')
+
+class TestOAuth2Client(UffdTestCase):
+ def setUpApp(self):
+ self.app.config['OAUTH2_CLIENTS'] = {
+ 'test': {'client_secret': 'testsecret', 'redirect_uris': ['http://localhost:5009/callback', 'http://localhost:5009/callback2']},
+ 'test1': {'client_secret': 'testsecret1', 'redirect_uris': ['http://localhost:5008/callback'], 'required_group': 'users'},
+ }
+
+ def test_from_id(self):
+ client = OAuth2Client.from_id('test')
+ self.assertEqual(client.client_id, 'test')
+ self.assertEqual(client.client_secret, 'testsecret')
+ self.assertEqual(client.redirect_uris, ['http://localhost:5009/callback', 'http://localhost:5009/callback2'])
+ self.assertEqual(client.default_redirect_uri, 'http://localhost:5009/callback')
+ self.assertEqual(client.default_scopes, ['profile'])
+ self.assertEqual(client.client_type, 'confidential')
+ client = OAuth2Client.from_id('test1')
+ self.assertEqual(client.client_id, 'test1')
+ self.assertEqual(client.required_group, 'users')
+
+ def test_access_allowed(self):
+ user = get_user() # has 'users' and 'uffd_access' group
+ admin = get_admin() # has 'users', 'uffd_access' and 'uffd_admin' group
+ client = OAuth2Client('test', '', [''], None)
+ self.assertTrue(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], 'users')
+ self.assertTrue(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], 'notagroup')
+ self.assertFalse(client.access_allowed(user))
+ self.assertFalse(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], 'uffd_admin')
+ self.assertFalse(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], ['uffd_admin'])
+ self.assertFalse(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], ['uffd_admin', 'notagroup'])
+ self.assertFalse(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], ['notagroup', 'uffd_admin' ])
+ self.assertFalse(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], ['uffd_admin', 'users'])
+ self.assertTrue(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], ['uffd_admin', 'users'])
+ self.assertTrue(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], [['uffd_admin', 'users'], ['users', 'uffd_access']])
+ self.assertTrue(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+ client = OAuth2Client('test', '', [''], ['uffd_admin', ['users', 'notagroup']])
+ self.assertFalse(client.access_allowed(user))
+ self.assertTrue(client.access_allowed(admin))
+
+class TestViews(UffdTestCase):
+ def setUpApp(self):
+ self.app.config['OAUTH2_CLIENTS'] = {
+ 'test': {'client_secret': 'testsecret', 'redirect_uris': ['http://localhost:5009/callback', 'http://localhost:5009/callback2']},
+ 'test1': {'client_secret': 'testsecret1', 'redirect_uris': ['http://localhost:5008/callback'], 'required_group': 'uffd_admin'},
+ }
+
+ def test_authorization(self):
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
+ state = 'teststate'
+ r = self.client.get(path=url_for('oauth2.authorize', response_type='code', client_id='test', state=state, redirect_uri='http://localhost:5009/callback'), follow_redirects=False)
+ self.assertEqual(r.status_code, 302)
+ self.assertTrue(r.location.startswith('http://localhost:5009/callback'))
+ args = parse_qs(urlparse(r.location).query)
+ self.assertEqual(args['state'], [state])
+ code = args['code'][0]
+ r = self.client.post(path=url_for('oauth2.token'),
+ data={'grant_type': 'authorization_code', 'code': code, 'redirect_uri': 'http://localhost:5009/callback', 'client_id': 'test', 'client_secret': 'testsecret'}, follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.content_type, 'application/json')
+ self.assertEqual(r.json['token_type'], 'Bearer')
+ self.assertEqual(r.json['scope'], 'profile')
+ token = r.json['access_token']
+ r = self.client.get(path=url_for('oauth2.userinfo'), headers=[('Authorization', 'Bearer %s'%token)], follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.content_type, 'application/json')
+ user = get_user()
+ self.assertEqual(r.json['id'], user.uid)
+ self.assertEqual(r.json['name'], user.displayname)
+ self.assertEqual(r.json['nickname'], user.loginname)
+ self.assertEqual(r.json['email'], user.mail)
+ self.assertTrue(r.json.get('groups'))
diff --git a/tests/test_role.py b/tests/test_role.py
new file mode 100644
index 0000000000000000000000000000000000000000..097a54bf3d60de0380d612a63cf130118f9b83ce
--- /dev/null
+++ b/tests/test_role.py
@@ -0,0 +1,93 @@
+import datetime
+import time
+
+from flask import url_for, session
+
+# These imports are required, because otherwise we get circular imports?!
+from uffd import ldap, user
+
+from uffd.user.models import Group
+from uffd.role.models import Role
+from uffd import create_app, db
+
+from utils import dump, UffdTestCase
+
+class TestRoleViews(UffdTestCase):
+ def setUp(self):
+ super().setUp()
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testadmin', 'password': 'adminpassword'}, follow_redirects=True)
+
+ def test_index(self):
+ db.session.add(Role('base', 'Base role description'))
+ db.session.add(Role('test1', 'Test1 role description'))
+ db.session.commit()
+ r = self.client.get(path=url_for('role.index'), follow_redirects=True)
+ dump('role_index', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_index_empty(self):
+ r = self.client.get(path=url_for('role.index'), follow_redirects=True)
+ dump('role_index_empty', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_show(self):
+ role = Role('base', 'Base role description')
+ db.session.add(role)
+ db.session.commit()
+ r = self.client.get(path=url_for('role.show', roleid=role.id), follow_redirects=True)
+ dump('role_show', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_new(self):
+ r = self.client.get(path=url_for('role.show'), follow_redirects=True)
+ dump('role_new', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_update(self):
+ role = Role('base', 'Base role description')
+ db.session.add(role)
+ db.session.commit()
+ role.add_group(Group.from_ldap_dn('cn=uffd_admin,ou=groups,dc=example,dc=com'))
+ db.session.commit()
+ self.assertEqual(role.name, 'base')
+ self.assertEqual(role.description, 'Base role description')
+ self.assertEqual(role.group_dns(), ['cn=uffd_admin,ou=groups,dc=example,dc=com'])
+ r = self.client.post(path=url_for('role.update', roleid=role.id),
+ data={'name': 'base1', 'description': 'Base role description1', 'group-20001': '1', 'group-20002': '1'},
+ follow_redirects=True)
+ dump('role_update', r)
+ self.assertEqual(r.status_code, 200)
+ role = Role.query.get(role.id)
+ self.assertEqual(role.name, 'base1')
+ self.assertEqual(role.description, 'Base role description1')
+ self.assertEqual(sorted(role.group_dns()), ['cn=uffd_access,ou=groups,dc=example,dc=com',
+ 'cn=users,ou=groups,dc=example,dc=com'])
+ # TODO: verify that group memberships are updated (currently not possible with ldap mock!)
+
+ def test_create(self):
+ self.assertIsNone(Role.query.filter_by(name='base').first())
+ r = self.client.post(path=url_for('role.update'),
+ data={'name': 'base', 'description': 'Base role description', 'group-20001': '1', 'group-20002': '1'},
+ follow_redirects=True)
+ dump('role_create', r)
+ self.assertEqual(r.status_code, 200)
+ role = Role.query.filter_by(name='base').first()
+ self.assertIsNotNone(role)
+ self.assertEqual(role.name, 'base')
+ self.assertEqual(role.description, 'Base role description')
+ self.assertEqual(sorted(role.group_dns()), ['cn=uffd_access,ou=groups,dc=example,dc=com',
+ 'cn=users,ou=groups,dc=example,dc=com'])
+ # TODO: verify that group memberships are updated (currently not possible with ldap mock!)
+
+ def test_delete(self):
+ role = Role('base', 'Base role description')
+ db.session.add(role)
+ db.session.commit()
+ role_id = role.id
+ self.assertIsNotNone(Role.query.get(role_id))
+ r = self.client.get(path=url_for('role.delete', roleid=role.id), follow_redirects=True)
+ dump('role_delete', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(Role.query.get(role_id))
+ # TODO: verify that group memberships are updated (currently not possible with ldap mock!)
diff --git a/tests/test_selfservice.py b/tests/test_selfservice.py
new file mode 100644
index 0000000000000000000000000000000000000000..3c5800f9615e2938ed404947d5a28a965bd4783c
--- /dev/null
+++ b/tests/test_selfservice.py
@@ -0,0 +1,258 @@
+import datetime
+import unittest
+
+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.selfservice.models import MailToken, PasswordToken
+from uffd.user.models import User
+from uffd import create_app, db, ldap
+
+from utils import dump, UffdTestCase
+
+def get_ldap_password():
+ conn = ldap.get_conn()
+ conn.search('uid=testuser,ou=users,dc=example,dc=com', '(objectClass=person)')
+ return conn.entries[0]['userPassword']
+
+class TestSelfservice(UffdTestCase):
+ def setUpApp(self):
+ self.app.config['MAIL_SKIP_SEND'] = True
+
+ def login(self):
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
+
+ def test_index(self):
+ self.login()
+ r = self.client.get(path=url_for('selfservice.index'))
+ dump('selfservice_index', r)
+ self.assertEqual(r.status_code, 200)
+ user = get_current_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()
+ 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()
+ self.assertEqual(_user.displayname, 'New Display Name')
+
+ def test_update_displayname_invalid(self):
+ self.login()
+ user = get_current_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()
+ self.assertNotEqual(_user.displayname, '')
+
+ def test_update_mail(self):
+ self.login()
+ user = get_current_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()
+ 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()
+ 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()
+ 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()
+ 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()
+ 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()
+ self.assertEqual(_user.mail, user.mail)
+
+ def test_token_mail_invalid(self):
+ self.login()
+ user = get_current_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()
+ self.assertEqual(_user.mail, user.mail)
+
+ @unittest.skip('See #26')
+ def test_token_mail_wrong_user(self):
+ self.login()
+ user = get_current_user()
+ admin_user = User.from_ldap_dn('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')
+ db.session.add(admin_token)
+ db.session.commit()
+ 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()
+ _admin_user = User.from_ldap_dn('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()
+ token = MailToken(loginname=user.loginname, newmail='newusermail@example.com',
+ created=(datetime.datetime.now() - datetime.timedelta(days=10)))
+ db.session.add(token)
+ db.session.commit()
+ 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()
+ self.assertEqual(_user.mail, user.mail)
+ tokens = MailToken.query.filter(MailToken.loginname == user.loginname).all()
+ self.assertEqual(len(tokens), 0)
+
+ def test_forgot_password(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ r = self.client.get(path=url_for('selfservice.forgot_password'))
+ dump('forgot_password', r)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('selfservice.forgot_password'),
+ data={'loginname': user.loginname, 'mail': user.mail}, follow_redirects=True)
+ dump('forgot_password_submit', r)
+ self.assertEqual(r.status_code, 200)
+ token = PasswordToken.query.filter(PasswordToken.loginname == user.loginname).first()
+ self.assertIsNotNone(token)
+ self.assertIn(token.token, str(self.app.last_mail.get_content()))
+
+ def test_forgot_password_wrong_user(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ r = self.client.get(path=url_for('selfservice.forgot_password'))
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('selfservice.forgot_password'),
+ data={'loginname': 'not_a_user', 'mail': user.mail}, follow_redirects=True)
+ dump('forgot_password_submit_wrong_user', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertFalse(hasattr(self.app, 'last_mail'))
+ self.assertEqual(len(PasswordToken.query.all()), 0)
+
+ def test_forgot_password_wrong_email(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ r = self.client.get(path=url_for('selfservice.forgot_password'), follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('selfservice.forgot_password'),
+ data={'loginname': user.loginname, 'mail': 'not_an_email@example.com'}, follow_redirects=True)
+ dump('forgot_password_submit_wrong_email', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertFalse(hasattr(self.app, 'last_mail'))
+ self.assertEqual(len(PasswordToken.query.all()), 0)
+
+ def test_token_password(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ oldpw = get_ldap_password()
+ token = PasswordToken(loginname=user.loginname)
+ db.session.add(token)
+ db.session.commit()
+ r = self.client.get(path=url_for('selfservice.token_password', token=token.token), follow_redirects=True)
+ dump('token_password', r)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('selfservice.token_password', token=token.token),
+ data={'password1': 'newpassword', 'password2': 'newpassword'}, follow_redirects=True)
+ dump('token_password_submit', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertNotEqual(oldpw, get_ldap_password())
+ # TODO: Verify that the new password is actually correct
+ self.assertEqual(len(PasswordToken.query.all()), 0)
+
+ def test_token_password_emptydb(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ oldpw = get_ldap_password()
+ r = self.client.get(path=url_for('selfservice.token_password', token='A'*128), follow_redirects=True)
+ dump('token_password_emptydb', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIn(b'Token expired, please try again', r.data)
+ r = self.client.post(path=url_for('selfservice.token_password', token='A'*128),
+ data={'password1': 'newpassword', 'password2': 'newpassword'}, follow_redirects=True)
+ dump('token_password_emptydb_submit', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIn(b'Token expired, please try again', r.data)
+ self.assertEqual(oldpw, get_ldap_password())
+
+ def test_token_password_invalid(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ oldpw = get_ldap_password()
+ token = PasswordToken(loginname=user.loginname)
+ db.session.add(token)
+ db.session.commit()
+ r = self.client.get(path=url_for('selfservice.token_password', token='A'*128), follow_redirects=True)
+ dump('token_password_invalid', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIn(b'Token expired, please try again', r.data)
+ r = self.client.post(path=url_for('selfservice.token_password', token='A'*128),
+ data={'password1': 'newpassword', 'password2': 'newpassword'}, follow_redirects=True)
+ dump('token_password_invalid_submit', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIn(b'Token expired, please try again', r.data)
+ self.assertEqual(oldpw, get_ldap_password())
+
+ def test_token_password_expired(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ oldpw = get_ldap_password()
+ token = PasswordToken(loginname=user.loginname,
+ created=(datetime.datetime.now() - datetime.timedelta(days=10)))
+ db.session.add(token)
+ db.session.commit()
+ r = self.client.get(path=url_for('selfservice.token_password', token=token.token), follow_redirects=True)
+ dump('token_password_invalid_expired', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIn(b'Token expired, please try again', r.data)
+ r = self.client.post(path=url_for('selfservice.token_password', token=token.token),
+ data={'password1': 'newpassword', 'password2': 'newpassword'}, follow_redirects=True)
+ dump('token_password_invalid_expired_submit', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIn(b'Token expired, please try again', r.data)
+ self.assertEqual(oldpw, get_ldap_password())
+
+ def test_token_password_different_passwords(self):
+ user = User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+ oldpw = get_ldap_password()
+ token = PasswordToken(loginname=user.loginname)
+ db.session.add(token)
+ db.session.commit()
+ r = self.client.get(path=url_for('selfservice.token_password', token=token.token), follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('selfservice.token_password', token=token.token),
+ data={'password1': 'newpassword', 'password2': 'differentpassword'}, follow_redirects=True)
+ dump('token_password_different_passwords_submit', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(oldpw, get_ldap_password())
+
diff --git a/tests/test_session.py b/tests/test_session.py
new file mode 100644
index 0000000000000000000000000000000000000000..fb66d2f21ea7fe7565dc9b151b6a5c07711a5de5
--- /dev/null
+++ b/tests/test_session.py
@@ -0,0 +1,136 @@
+import time
+import unittest
+
+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, login_required, is_valid_session
+from uffd import create_app, db
+
+from utils import dump, UffdTestCase
+
+class TestSession(UffdTestCase):
+ def setUpApp(self):
+ self.app.config['SESSION_LIFETIME_SECONDS'] = 2
+
+ @self.app.route('/test_login_required')
+ @login_required()
+ def test_login_required():
+ return 'SUCCESS', 200
+
+ @self.app.route('/test_group_required1')
+ @login_required(group='users')
+ def test_group_required1():
+ return 'SUCCESS', 200
+
+ @self.app.route('/test_group_required2')
+ @login_required(group='notagroup')
+ def test_group_required2():
+ return 'SUCCESS', 200
+
+ def setUp(self):
+ super().setUp()
+ self.assertFalse(is_valid_session())
+
+ def login(self):
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
+ self.assertTrue(is_valid_session())
+
+ def assertLogin(self):
+ self.assertTrue(is_valid_session())
+ self.assertEqual(self.client.get(path=url_for('test_login_required'),
+ follow_redirects=True).data, b'SUCCESS')
+ self.assertEqual(get_current_user().loginname, 'testuser')
+
+ def assertLogout(self):
+ self.assertFalse(is_valid_session())
+ self.assertNotEqual(self.client.get(path=url_for('test_login_required'),
+ follow_redirects=True).data, b'SUCCESS')
+ self.assertEqual(get_current_user(), None)
+
+ def test_login(self):
+ self.assertLogout()
+ r = self.client.get(path=url_for('session.login'), follow_redirects=True)
+ dump('login', r)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
+ dump('login_post', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertLogin()
+
+ def test_redirect(self):
+ r = self.client.post(path=url_for('session.login', ref=url_for('test_login_required')),
+ data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.data, b'SUCCESS')
+
+ def test_wrong_password(self):
+ r = self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'wrongpassword'}, follow_redirects=True)
+ dump('login_wrong_password', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertLogout()
+
+ @unittest.skip('See #27')
+ def test_empty_password(self):
+ r = self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': ''}, follow_redirects=True)
+ dump('login_empty_password', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertLogout()
+
+ def test_wrong_user(self):
+ r = self.client.post(path=url_for('session.login'),
+ data={'loginname': 'nouser', 'password': 'userpassword'}, follow_redirects=True)
+ dump('login_wrong_user', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertLogout()
+
+ def test_empty_user(self):
+ r = self.client.post(path=url_for('session.login'),
+ data={'loginname': '', 'password': 'userpassword'}, follow_redirects=True)
+ dump('login_empty_user', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertLogout()
+
+ def test_no_access(self):
+ r = self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testservice', 'password': 'servicepassword'}, follow_redirects=True)
+ dump('login_no_access', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertLogout()
+
+ def test_group_required(self):
+ self.login()
+ self.assertEqual(self.client.get(path=url_for('test_group_required1'),
+ follow_redirects=True).data, b'SUCCESS')
+ self.assertNotEqual(self.client.get(path=url_for('test_group_required2'),
+ follow_redirects=True).data, b'SUCCESS')
+
+ def test_logout(self):
+ self.login()
+ r = self.client.get(path=url_for('session.logout'), follow_redirects=True)
+ dump('logout', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertLogout()
+
+ @unittest.skip('See #29')
+ def test_timeout(self):
+ self.login()
+ time.sleep(3)
+ self.assertLogout()
+
+ @unittest.skip('Not implemented, see #10')
+ def test_ratelimit(self):
+ for i in range(20):
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'wrongpassword_%i'%i}, follow_redirects=True)
+ r = self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testuser', 'password': 'userpassword'}, follow_redirects=True)
+ dump('login_ratelimit', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertFalse(is_valid_session())
diff --git a/tests/test_user.py b/tests/test_user.py
new file mode 100644
index 0000000000000000000000000000000000000000..3b095fd6ad9f1bb53db0110190cb62bfcb4da31b
--- /dev/null
+++ b/tests/test_user.py
@@ -0,0 +1,316 @@
+import datetime
+import time
+import unittest
+
+from flask import url_for, session
+
+# These imports are required, because otherwise we get circular imports?!
+from uffd import ldap, user
+
+from uffd.user.models import User
+from uffd.role.models import Role
+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
+
+from utils import dump, UffdTestCase
+
+def get_user():
+ return User.from_ldap_dn('uid=testuser,ou=users,dc=example,dc=com')
+
+def get_user_password():
+ conn = ldap.get_conn()
+ conn.search('uid=testuser,ou=users,dc=example,dc=com', '(objectClass=person)')
+ return conn.entries[0]['userPassword']
+
+def get_admin():
+ return User.from_ldap_dn('uid=testadmin,ou=users,dc=example,dc=com')
+
+class TestUserViews(UffdTestCase):
+ def setUp(self):
+ super().setUp()
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testadmin', 'password': 'adminpassword'}, follow_redirects=True)
+
+ def test_index(self):
+ r = self.client.get(path=url_for('user.index'), follow_redirects=True)
+ dump('user_index', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_new(self):
+ db.session.add(Role('base'))
+ role1 = Role('role1')
+ db.session.add(role1)
+ role2 = Role('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', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(User.from_ldap_dn('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'}, follow_redirects=True)
+ dump('user_new_submit', r)
+ self.assertEqual(r.status_code, 200)
+ user = User.from_ldap_dn('uid=newuser,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser')
+ self.assertEqual(user.displayname, 'New User')
+ self.assertEqual(user.mail, 'newuser@example.com')
+ self.assertTrue(user.uid)
+ self.assertEqual(roles, ['base', 'role1'])
+ # TODO: check password hash
+
+ 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',
+ 'password': 'newpassword'}, follow_redirects=True)
+ dump('user_new_invalid_loginname', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(User.from_ldap_dn('uid=newuser,ou=users,dc=example,dc=com'))
+
+ def test_new_empty_loginname(self):
+ r = self.client.post(path=url_for('user.update'),
+ data={'loginname': '', 'mail': 'newuser@example.com', 'displayname': 'New User',
+ 'password': 'newpassword'}, follow_redirects=True)
+ dump('user_new_empty_loginname', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(User.from_ldap_dn('uid=newuser,ou=users,dc=example,dc=com'))
+
+ def test_new_empty_email(self):
+ r = self.client.post(path=url_for('user.update'),
+ data={'loginname': 'newuser', 'mail': '', 'displayname': 'New User',
+ 'password': 'newpassword'}, follow_redirects=True)
+ dump('user_new_empty_email', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(User.from_ldap_dn('uid=newuser,ou=users,dc=example,dc=com'))
+
+ def test_new_invalid_display_name(self):
+ r = self.client.post(path=url_for('user.update'),
+ data={'loginname': 'newuser', 'mail': 'newuser@example.com', 'displayname': 'A'*200,
+ 'password': 'newpassword'}, follow_redirects=True)
+ dump('user_new_invalid_display_name', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(User.from_ldap_dn('uid=newuser,ou=users,dc=example,dc=com'))
+
+ def test_update(self):
+ user = get_user()
+ db.session.add(Role('base'))
+ role1 = Role('role1')
+ db.session.add(role1)
+ role2 = Role('role2')
+ db.session.add(role2)
+ role2.add_member(user)
+ db.session.commit()
+ role1_id = role1.id
+ oldpw = get_user_password()
+ r = self.client.get(path=url_for('user.show', uid=user.uid), follow_redirects=True)
+ dump('user_update', r)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('user.update', uid=user.uid),
+ data={'loginname': 'testuser', 'mail': 'newuser@example.com', 'displayname': 'New User',
+ f'role-{role1_id}': '1', 'password': ''}, follow_redirects=True)
+ dump('user_update_submit', r)
+ self.assertEqual(r.status_code, 200)
+ _user = get_user()
+ roles = sorted([r.name for r in Role.get_for_user(_user)])
+ self.assertEqual(_user.displayname, 'New User')
+ self.assertEqual(_user.mail, 'newuser@example.com')
+ self.assertEqual(_user.uid, user.uid)
+ self.assertEqual(_user.loginname, user.loginname)
+ self.assertEqual(get_user_password(), oldpw)
+ self.assertEqual(roles, ['base', 'role1'])
+
+ def test_update_password(self):
+ user = get_user()
+ oldpw = get_user_password()
+ r = self.client.get(path=url_for('user.show', uid=user.uid), follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('user.update', uid=user.uid),
+ data={'loginname': 'testuser', 'mail': 'newuser@example.com', 'displayname': 'New User',
+ 'password': 'newpassword'}, follow_redirects=True)
+ dump('user_update_password', r)
+ self.assertEqual(r.status_code, 200)
+ _user = get_user()
+ self.assertEqual(_user.displayname, 'New User')
+ self.assertEqual(_user.mail, 'newuser@example.com')
+ self.assertEqual(_user.uid, user.uid)
+ self.assertEqual(_user.loginname, user.loginname)
+ self.assertNotEqual(get_user_password(), oldpw)
+
+ @unittest.skip('See #28')
+ def test_update_invalid_password(self):
+ user = get_user()
+ oldpw = get_user_password()
+ r = self.client.get(path=url_for('user.show', uid=user.uid), follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('user.update', uid=user.uid),
+ data={'loginname': 'testuser', 'mail': 'newuser@example.com', 'displayname': 'New User',
+ 'password': 'A'}, follow_redirects=True)
+ dump('user_update_password', r)
+ self.assertEqual(r.status_code, 200)
+ _user = get_user()
+ self.assertEqual(get_user_password(), oldpw)
+ self.assertEqual(_user.displayname, user.displayname)
+ self.assertEqual(_user.mail, user.mail)
+ self.assertEqual(_user.loginname, user.loginname)
+
+ def test_update_empty_email(self):
+ user = get_user()
+ oldpw = get_user_password()
+ r = self.client.get(path=url_for('user.show', uid=user.uid), follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('user.update', uid=user.uid),
+ data={'loginname': 'testuser', 'mail': '', 'displayname': 'New User',
+ 'password': 'newpassword'}, follow_redirects=True)
+ dump('user_update_empty_mail', r)
+ self.assertEqual(r.status_code, 200)
+ _user = get_user()
+ self.assertEqual(_user.displayname, user.displayname)
+ self.assertEqual(_user.mail, user.mail)
+ self.assertEqual(_user.loginname, user.loginname)
+ self.assertEqual(get_user_password(), oldpw)
+
+ def test_update_invalid_display_name(self):
+ user = get_user()
+ oldpw = get_user_password()
+ r = self.client.get(path=url_for('user.show', uid=user.uid), follow_redirects=True)
+ self.assertEqual(r.status_code, 200)
+ r = self.client.post(path=url_for('user.update', uid=user.uid),
+ data={'loginname': 'testuser', 'mail': 'newuser@example.com', 'displayname': 'A'*200,
+ 'password': 'newpassword'}, follow_redirects=True)
+ dump('user_update_invalid_display_name', r)
+ self.assertEqual(r.status_code, 200)
+ _user = get_user()
+ self.assertEqual(_user.displayname, user.displayname)
+ self.assertEqual(_user.mail, user.mail)
+ self.assertEqual(_user.loginname, user.loginname)
+ self.assertEqual(get_user_password(), oldpw)
+
+ def test_show(self):
+ r = self.client.get(path=url_for('user.show', uid=get_user().uid), follow_redirects=True)
+ dump('user_show', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_delete(self):
+ user = get_user()
+ r = self.client.get(path=url_for('user.delete', uid=user.uid), follow_redirects=True)
+ dump('user_delete', r)
+ self.assertEqual(r.status_code, 200)
+ self.assertIsNone(get_user())
+
+ def test_csvimport(self):
+ db.session.add(Role('base'))
+ role1 = Role('role1')
+ db.session.add(role1)
+ role2 = Role('role2')
+ db.session.add(role2)
+ db.session.commit()
+ data = f'''\
+newuser1,newuser1@example.com,
+newuser2,newuser2@example.com,{role1.id}
+newuser3,newuser3@example.com,{role1.id};{role2.id}
+newuser4,newuser4@example.com,9999
+newuser5,newuser5@example.com,notanumber
+newuser6,newuser6@example.com,{role1.id};{role2.id};
+newuser7,invalidmail,
+newuser8,,
+,newuser9@example.com,
+,,
+
+,,,
+newuser10,newuser10@example.com,
+newuser11,newuser11@example.com, {role1.id};{role2.id}
+newuser12,newuser12@example.com,{role1.id};{role1.id}
+<invalid tag-like thingy>'''
+ r = self.client.post(path=url_for('user.csvimport'), data={'csv': data}, follow_redirects=True)
+ dump('user_csvimport', r)
+ self.assertEqual(r.status_code, 200)
+ user = User.from_ldap_dn('uid=newuser1,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser1')
+ self.assertEqual(user.displayname, 'newuser1')
+ self.assertEqual(user.mail, 'newuser1@example.com')
+ self.assertEqual(roles, ['base'])
+ user = User.from_ldap_dn('uid=newuser2,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser2')
+ self.assertEqual(user.displayname, 'newuser2')
+ self.assertEqual(user.mail, 'newuser2@example.com')
+ self.assertEqual(roles, ['base', 'role1'])
+ user = User.from_ldap_dn('uid=newuser3,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser3')
+ self.assertEqual(user.displayname, 'newuser3')
+ self.assertEqual(user.mail, 'newuser3@example.com')
+ self.assertEqual(roles, ['base', 'role1', 'role2'])
+ user = User.from_ldap_dn('uid=newuser4,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser4')
+ self.assertEqual(user.displayname, 'newuser4')
+ self.assertEqual(user.mail, 'newuser4@example.com')
+ self.assertEqual(roles, ['base'])
+ user = User.from_ldap_dn('uid=newuser5,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser5')
+ self.assertEqual(user.displayname, 'newuser5')
+ self.assertEqual(user.mail, 'newuser5@example.com')
+ self.assertEqual(roles, ['base'])
+ user = User.from_ldap_dn('uid=newuser6,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser6')
+ self.assertEqual(user.displayname, 'newuser6')
+ self.assertEqual(user.mail, 'newuser6@example.com')
+ self.assertEqual(roles, ['base', 'role1', 'role2'])
+ self.assertIsNone(User.from_ldap_dn('uid=newuser7,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.from_ldap_dn('uid=newuser8,ou=users,dc=example,dc=com'))
+ self.assertIsNone(User.from_ldap_dn('uid=newuser9,ou=users,dc=example,dc=com'))
+ user = User.from_ldap_dn('uid=newuser10,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser10')
+ self.assertEqual(user.displayname, 'newuser10')
+ self.assertEqual(user.mail, 'newuser10@example.com')
+ self.assertEqual(roles, ['base'])
+ user = User.from_ldap_dn('uid=newuser11,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser11')
+ self.assertEqual(user.displayname, 'newuser11')
+ self.assertEqual(user.mail, 'newuser11@example.com')
+ # Currently the csv import is not very robust, imho newuser11 should have role1 and role2!
+ #self.assertEqual(roles, ['base', 'role1', 'role2'])
+ self.assertEqual(roles, ['base', 'role2'])
+ user = User.from_ldap_dn('uid=newuser12,ou=users,dc=example,dc=com')
+ roles = sorted([r.name for r in Role.get_for_user(user)])
+ self.assertIsNotNone(user)
+ self.assertEqual(user.loginname, 'newuser12')
+ self.assertEqual(user.displayname, 'newuser12')
+ self.assertEqual(user.mail, 'newuser12@example.com')
+ self.assertEqual(roles, ['base', 'role1'])
+
+class TestGroupViews(UffdTestCase):
+ def setUp(self):
+ super().setUp()
+ self.client.post(path=url_for('session.login'),
+ data={'loginname': 'testadmin', 'password': 'adminpassword'}, follow_redirects=True)
+
+ def test_index(self):
+ r = self.client.get(path=url_for('group.index'), follow_redirects=True)
+ dump('group_index', r)
+ self.assertEqual(r.status_code, 200)
+
+ def test_show(self):
+ r = self.client.get(path=url_for('group.show', gid=20001), follow_redirects=True)
+ dump('group_show', r)
+ self.assertEqual(r.status_code, 200)
+
diff --git a/tests/utils.py b/tests/utils.py
new file mode 100644
index 0000000000000000000000000000000000000000..8cdd4772c5287fecd3876918a7ac31acdd3ce674
--- /dev/null
+++ b/tests/utils.py
@@ -0,0 +1,43 @@
+import os
+import tempfile
+import shutil
+import unittest
+
+from uffd import create_app, db
+
+def dump(basename, resp):
+ basename = basename.replace('.', '_').replace('/', '_')
+ suffix = '.html'
+ root = os.environ.get('DUMP_PAGES')
+ if not root:
+ return
+ os.makedirs(root, exist_ok=True)
+ path = os.path.join(root, basename+suffix)
+ with open(path, 'xb') as f:
+ f.write(resp.data)
+
+class UffdTestCase(unittest.TestCase):
+ def setUp(self):
+ self.dir = tempfile.mkdtemp()
+ # It would be far better to create a minimal app here, but since the
+ # session module depends on almost everything else, that is not really feasable
+ self.app = create_app({
+ 'TESTING': True,
+ 'DEBUG': True,
+ 'SQLALCHEMY_DATABASE_URI': 'sqlite:///%s/db.sqlite'%self.dir,
+ 'SECRET_KEY': 'DEBUGKEY',
+ 'LDAP_SERVICE_MOCK': True,
+ })
+ self.setUpApp()
+ self.client = self.app.test_client()
+ self.client.__enter__()
+ # Just do some request so that we can use url_for
+ self.client.get(path='/')
+ db.create_all()
+
+ def setUpApp(self):
+ pass
+
+ def tearDown(self):
+ self.client.__exit__(None, None, None)
+ shutil.rmtree(self.dir)
diff --git a/uffd/mail/templates/mail.html b/uffd/mail/templates/mail.html
index 36021697550b986fe5902ae9e13a0084b74f3cff..44f74f47264aec0bba60bea128e507d2ba9a892b 100644
--- a/uffd/mail/templates/mail.html
+++ b/uffd/mail/templates/mail.html
@@ -11,14 +11,14 @@
</div>
<div class="form-group col">
<label for="mail-receivers">Receiving addresses</label>
- <textarea rows="10" class="form-control" name="mail-receivers">{{ mail.receivers|join('\n') }}</textarea>
+ <textarea rows="10" class="form-control" id="mail-receivers" name="mail-receivers">{{ mail.receivers|join('\n') }}</textarea>
<small class="form-text text-muted">
One address per line
</small>
</div>
<div class="form-group col">
<label for="mail-destinations">Destinations</label>
- <textarea rows="10" class="form-control" name="mail-destinations">{{ mail.destinations|join('\n') }}</textarea>
+ <textarea rows="10" class="form-control" id="mail-destinations" name="mail-destinations">{{ mail.destinations|join('\n') }}</textarea>
<small class="form-text text-muted">
One address per line
</small>
diff --git a/uffd/mail/templates/mail_list.html b/uffd/mail/templates/mail_list.html
index 32fa3b6448a407221ae4c9fa245a3ecf959e795c..844689516dc0d78d21a2126b4d886319db1ad6ff 100644
--- a/uffd/mail/templates/mail_list.html
+++ b/uffd/mail/templates/mail_list.html
@@ -11,7 +11,7 @@
<th scope="col">destinations</th>
<th scope="col">
<p class="text-right">
- <a type="button" class="btn btn-primary" href="{{ url_for("mail.show") }}">
+ <a class="btn btn-primary" href="{{ url_for("mail.show") }}">
<i class="fa fa-plus" aria-hidden="true"></i> New
</a>
</p>
@@ -52,5 +52,5 @@
</tbody>
</table>
</div>
-</dev>
+</div>
{% endblock %}
diff --git a/uffd/mfa/templates/auth.html b/uffd/mfa/templates/auth.html
index 831bc87a74ed93418ee6ec6139765cf0ac656c7b..13c1f0c60add51f9ed25eddc3a8c794c57823af9 100644
--- a/uffd/mfa/templates/auth.html
+++ b/uffd/mfa/templates/auth.html
@@ -6,7 +6,7 @@
<div class="row mt-2 justify-content-center">
<div class="col-lg-6 col-md-10" style="background: #f7f7f7; box-shadow: 0px 2px 2px rgba(0, 0, 0, 0.3); padding: 30px;">
<div class="text-center">
- <img src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
+ <img alt="CCC logo" src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
</div>
<div class="col-12 mb-3">
<h2 class="text-center">Two-Factor Authentication</h2>
diff --git a/uffd/role/templates/role_list.html b/uffd/role/templates/role_list.html
index 007ddf387db50f98572201800448e147d790ebd9..1e2f31fa6e17394a3397c533b747f37cd7a0277e 100644
--- a/uffd/role/templates/role_list.html
+++ b/uffd/role/templates/role_list.html
@@ -11,7 +11,7 @@
<th scope="col">description</th>
<th scope="col">
<p class="text-right">
- <a type="button" class="btn btn-primary" href="{{ url_for("role.show") }}">
+ <a class="btn btn-primary" href="{{ url_for("role.show") }}">
<i class="fa fa-plus" aria-hidden="true"></i> New
</a>
</p>
@@ -42,5 +42,5 @@
</tbody>
</table>
</div>
-</dev>
+</div>
{% endblock %}
diff --git a/uffd/selfservice/templates/forgot_password.html b/uffd/selfservice/templates/forgot_password.html
index 641f7dcb982189ffefcf07add68283239dee2a25..9cc647e0bb36f7fd694dd946c0acd549dd3f2b75 100644
--- a/uffd/selfservice/templates/forgot_password.html
+++ b/uffd/selfservice/templates/forgot_password.html
@@ -5,7 +5,7 @@
<div class="row mt-2 justify-content-center">
<div class="col-lg-6 col-md-10" style="background: #f7f7f7; box-shadow: 0px 2px 2px rgba(0, 0, 0, 0.3); padding: 30px;">
<div class="text-center">
- <img src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
+ <img alt="CCC logo" src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
</div>
<div class="col-12">
<h2 class="text-center">Forgot password</h2>
diff --git a/uffd/selfservice/templates/set_password.html b/uffd/selfservice/templates/set_password.html
index 67f3a5a2ca4dc67c3400fbd8977f1a520537ac96..50569bc372289554f24ed622bc715f9888612d7f 100644
--- a/uffd/selfservice/templates/set_password.html
+++ b/uffd/selfservice/templates/set_password.html
@@ -5,7 +5,7 @@
<div class="row mt-2 justify-content-center">
<div class="col-lg-6 col-md-10" style="background: #f7f7f7; box-shadow: 0px 2px 2px rgba(0, 0, 0, 0.3); padding: 30px;">
<div class="text-center">
- <img src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
+ <img alt="CCC logo" src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
</div>
<div class="col-12">
<h2 class="text-center">Reset password</h2>
diff --git a/uffd/selfservice/views.py b/uffd/selfservice/views.py
index bcefe55fadb6a10ebecf8e38ed6941985dbacc61..4770c950f0d8aaca587be67ea127e6d56d854230 100644
--- a/uffd/selfservice/views.py
+++ b/uffd/selfservice/views.py
@@ -145,17 +145,27 @@ def send_passwordreset(loginname):
send_mail(user.mail, msg)
def send_mail(to_address, msg):
+ msg['From'] = current_app.config['MAIL_FROM_ADDRESS']
+ msg['To'] = to_address
+ msg['Date'] = email.utils.formatdate(localtime=1)
+ msg['Message-ID'] = email.utils.make_msgid()
try:
+ if current_app.debug:
+ current_app.last_mail = None
+ current_app.logger.debug('Trying to send email to %s:\n'%(to_address)+str(msg))
+ if current_app.debug and current_app.config.get('MAIL_SKIP_SEND', False):
+ if current_app.config['MAIL_SKIP_SEND'] == 'fail':
+ raise smtplib.SMTPException()
+ current_app.last_mail = msg
+ return True
server = smtplib.SMTP(host=current_app.config['MAIL_SERVER'], port=current_app.config['MAIL_PORT'])
if current_app.config['MAIL_USE_STARTTLS']:
server.starttls()
server.login(current_app.config['MAIL_USERNAME'], current_app.config['MAIL_PASSWORD'])
- msg['From'] = current_app.config['MAIL_FROM_ADDRESS']
- msg['To'] = to_address
- msg['Date'] = email.utils.formatdate(localtime=1)
- msg['Message-ID'] = email.utils.make_msgid()
server.send_message(msg)
server.quit()
+ if current_app.debug:
+ current_app.last_mail = msg
return True
except smtplib.SMTPException:
flash('Mail to "{}" could not be sent!'.format(to_address))
diff --git a/uffd/session/templates/login.html b/uffd/session/templates/login.html
index 0332394891e194433e77b0b57500461164511c82..63189ed099a7157359dee4ddd50770451560d37e 100644
--- a/uffd/session/templates/login.html
+++ b/uffd/session/templates/login.html
@@ -5,7 +5,7 @@
<div class="row mt-2 justify-content-center">
<div class="col-lg-6 col-md-10" style="background: #f7f7f7; box-shadow: 0px 2px 2px rgba(0, 0, 0, 0.3); padding: 30px;">
<div class="text-center">
- <img src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
+ <img alt="CCC logo" src="{{ url_for("static", filename="chaosknoten.png") }}" class="col-lg-8 col-md-12" >
</div>
<div class="col-12">
<h2 class="text-center">Login</h2>
diff --git a/uffd/template_helper.py b/uffd/template_helper.py
index 998bafaa4ecf3a7e745e226c7ddb73c7de25226f..cc02d55621fb506141e3ef5fbd9ba01149af018c 100644
--- a/uffd/template_helper.py
+++ b/uffd/template_helper.py
@@ -22,7 +22,7 @@ def register_template_helper(app):
svg.set(key, value)
buf = io.BytesIO()
img.save(buf)
- return Markup(buf.getvalue().decode())
+ return Markup(buf.getvalue().decode().replace('<?xml version=\'1.0\' encoding=\'UTF-8\'?>\n', ''))
@app.template_filter()
def datauri(data, mimetype='text/plain'): #pylint: disable=unused-variable
diff --git a/uffd/user/templates/group.html b/uffd/user/templates/group.html
index 6ca116ea5005eaa8c3fc0c328a25e83a49343948..2bfcebd418c1d57c6feba66d6fbe2e5565bdc7fc 100644
--- a/uffd/user/templates/group.html
+++ b/uffd/user/templates/group.html
@@ -8,7 +8,7 @@
<input type="number" class="form-control" id="group-gid" name="gid" value="{{ group.gid }}" readonly>
</div>
<div class="form-group col">
- <label for="group-name">name</label>
+ <label for="group-loginname">name</label>
<input type="text" class="form-control" id="group-loginname" name="loginname" value="{{ group.name }}" readonly>
</div>
<div class="col">
diff --git a/uffd/user/templates/group_list.html b/uffd/user/templates/group_list.html
index eca01e423bff881454af86019d201d339a8e854f..3b3a41dfbbedf42f7cce67609409e71a8d5c0f77 100644
--- a/uffd/user/templates/group_list.html
+++ b/uffd/user/templates/group_list.html
@@ -30,5 +30,5 @@
</tbody>
</table>
</div>
-</dev>
+</div>
{% endblock %}
diff --git a/uffd/user/templates/user.html b/uffd/user/templates/user.html
index aa10e9ea7723b37c48d5c495e31bead1350de0b0..1411c9f12269143f62b7c80bfe05704ca5337651 100644
--- a/uffd/user/templates/user.html
+++ b/uffd/user/templates/user.html
@@ -31,7 +31,7 @@
{% if user.uid %}
<input type="number" class="form-control" id="user-uid" name="uid" value="{{ user.uid }}" readonly>
{% else %}
- <input type="string" class="form-control" id="user-uid" name="uid" placeholder="will be choosen" readonly>
+ <input type="text" class="form-control" id="user-uid" name="uid" placeholder="will be choosen" readonly>
{% endif %}
</div>
<div class="form-group col">
diff --git a/uffd/user/templates/user_list.html b/uffd/user/templates/user_list.html
index 21d437c38749670ebb7063f9996acb1637d32d12..34102af796966bbbcd35c90c2ed2557b4bda423d 100644
--- a/uffd/user/templates/user_list.html
+++ b/uffd/user/templates/user_list.html
@@ -11,7 +11,7 @@
<th scope="col">display name</th>
<th scope="col">
<p class="text-right">
- <a type="button" class="btn btn-primary" href="{{ url_for("user.show") }}">
+ <a class="btn btn-primary" href="{{ url_for("user.show") }}">
<i class="fa fa-plus" aria-hidden="true"></i> New
</a>
<button type="button" class="btn btn-primary" data-toggle="modal" data-target="#csvimport">
@@ -61,8 +61,8 @@
</div>
<div class="modal-body">
<p>
- The format should be "loginname,mailaddres,groupid1;groupid2;".
- Neither setting the display name, nor setting roles or passwords is supported (yet).
+ The format should be "loginname,mailaddres,roleid1;roleid2".
+ Neither setting the display name nor setting passwords is supported (yet).
Example:
</p>
<pre>