Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • uffd/uffd
  • rixx/uffd
  • thies/uffd
  • leona/uffd
  • enbewe/uffd
  • strifel/uffd
  • thies/uffd-2
7 results
Show changes
Showing
with 1157 additions and 1927 deletions
import itertools
from uffd.remailer import remailer
from uffd.tasks import cleanup_task
from uffd.database import db
from uffd.models import Service, ServiceUser, User, UserEmail, RemailerMode
from tests.utils import UffdTestCase
class TestServiceUser(UffdTestCase):
def setUp(self):
super().setUp()
db.session.add_all([Service(name='service1', limit_access=False), Service(name='service2', remailer_mode=RemailerMode.ENABLED_V1, limit_access=False)])
db.session.commit()
def test_auto_create(self):
service_count = Service.query.count()
user_count = User.query.count()
self.assertEqual(ServiceUser.query.count(), service_count * user_count)
db.session.add(User(loginname='newuser1', displayname='New User', primary_email_address='new1@example.com'))
db.session.commit()
self.assertEqual(ServiceUser.query.count(), service_count * (user_count + 1))
db.session.add(Service(name='service3'))
db.session.commit()
self.assertEqual(ServiceUser.query.count(), (service_count + 1) * (user_count + 1))
db.session.add(User(loginname='newuser2', displayname='New User', primary_email_address='new2@example.com'))
db.session.add(User(loginname='newuser3', displayname='New User', primary_email_address='new3@example.com'))
db.session.add(Service(name='service4'))
db.session.add(Service(name='service5'))
db.session.commit()
self.assertEqual(ServiceUser.query.count(), (service_count + 3) * (user_count + 3))
def test_create_missing(self):
service_count = Service.query.count()
user_count = User.query.count()
self.assertEqual(ServiceUser.query.count(), service_count * user_count)
db.session.delete(ServiceUser.query.first())
db.session.commit()
self.assertEqual(ServiceUser.query.count(), service_count * user_count - 1)
cleanup_task.run()
db.session.commit()
self.assertEqual(ServiceUser.query.count(), service_count * user_count)
def test_effective_remailer_mode(self):
self.app.config['REMAILER_DOMAIN'] = 'remailer.example.com'
user = self.get_user()
service = Service.query.filter_by(name='service1').first()
service.remailer_mode = RemailerMode.ENABLED_V2
service_user = ServiceUser.query.get((service.id, user.id))
self.assertEqual(service_user.effective_remailer_mode, RemailerMode.ENABLED_V2)
self.app.config['REMAILER_LIMIT_TO_USERS'] = ['testadmin']
self.assertEqual(service_user.effective_remailer_mode, RemailerMode.DISABLED)
self.app.config['REMAILER_LIMIT_TO_USERS'] = ['testuser']
self.assertEqual(service_user.effective_remailer_mode, RemailerMode.ENABLED_V2)
self.app.config['REMAILER_LIMIT_TO_USERS'] = None
service_user.remailer_overwrite_mode = RemailerMode.ENABLED_V1
service.remailer_mode = RemailerMode.DISABLED
self.assertEqual(service_user.effective_remailer_mode, RemailerMode.ENABLED_V1)
self.app.config['REMAILER_DOMAIN'] = ''
self.assertEqual(service_user.effective_remailer_mode, RemailerMode.DISABLED)
def test_service_email(self):
user = self.get_user()
service = Service.query.filter_by(name='service1').first()
service_user = ServiceUser.query.get((service.id, user.id))
self.assertEqual(service_user.service_email, None)
service_user.service_email = UserEmail(user=user, address='foo@bar', verified=True)
with self.assertRaises(Exception):
service_user.service_email = UserEmail(user=user, address='foo2@bar', verified=False)
with self.assertRaises(Exception):
service_user.service_email = UserEmail(user=self.get_admin(), address='foo3@bar', verified=True)
def test_real_email(self):
user = self.get_user()
service = Service.query.filter_by(name='service1').first()
service_user = ServiceUser.query.get((service.id, user.id))
self.assertEqual(service_user.real_email, user.primary_email.address)
service_user.service_email = UserEmail(user=user, address='foo@bar', verified=True)
self.assertEqual(service_user.real_email, user.primary_email.address)
service.enable_email_preferences = True
self.assertEqual(service_user.real_email, service_user.service_email.address)
service.limit_access = True
self.assertEqual(service_user.real_email, user.primary_email.address)
service.access_group = self.get_admin_group()
self.assertEqual(service_user.real_email, user.primary_email.address)
service.access_group = self.get_users_group()
self.assertEqual(service_user.real_email, service_user.service_email.address)
def test_get_by_remailer_email(self):
user = self.get_user()
service = Service.query.filter_by(name='service1').first()
service_user = ServiceUser.query.get((service.id, user.id))
self.app.config['REMAILER_DOMAIN'] = 'remailer.example.com'
remailer_email = remailer.build_v1_address(service.id, user.id)
# 1. remailer not setup
self.app.config['REMAILER_DOMAIN'] = ''
self.assertIsNone(ServiceUser.get_by_remailer_email(user.primary_email.address))
self.assertIsNone(ServiceUser.get_by_remailer_email(remailer_email))
self.assertIsNone(ServiceUser.get_by_remailer_email('invalid'))
# 2. remailer setup
self.app.config['REMAILER_DOMAIN'] = 'remailer.example.com'
self.assertIsNone(ServiceUser.get_by_remailer_email(user.primary_email.address))
self.assertEqual(ServiceUser.get_by_remailer_email(remailer_email), service_user)
self.assertIsNone(ServiceUser.get_by_remailer_email('invalid'))
def test_email(self):
user = self.get_user()
service = Service.query.filter_by(name='service1').first()
service_user = ServiceUser.query.get((service.id, user.id))
self.app.config['REMAILER_DOMAIN'] = 'remailer.example.com'
remailer_email = remailer.build_v1_address(service.id, user.id)
# 1. remailer not setup
self.app.config['REMAILER_DOMAIN'] = ''
self.assertEqual(service_user.email, user.primary_email.address)
# 2. remailer setup + remailer disabled
self.app.config['REMAILER_DOMAIN'] = 'remailer.example.com'
self.assertEqual(service_user.email, user.primary_email.address)
# 3. remailer setup + remailer enabled + REMAILER_LIMIT_TO_USERS unset
service.remailer_mode = RemailerMode.ENABLED_V1
db.session.commit()
self.assertEqual(service_user.email, remailer_email)
# 4. remailer setup + remailer enabled + REMAILER_LIMIT_TO_USERS does not include user
self.app.config['REMAILER_LIMIT_TO_USERS'] = ['testadmin']
self.assertEqual(service_user.email, user.primary_email.address)
# 5. remailer setup + remailer enabled + REMAILER_LIMIT_TO_USERS includes user
self.app.config['REMAILER_LIMIT_TO_USERS'] = ['testuser']
self.assertEqual(service_user.email, remailer_email)
# 6. remailer setup + remailer disabled + user overwrite
self.app.config['REMAILER_LIMIT_TO_USERS'] = None
service.remailer_mode = RemailerMode.DISABLED
service_user.remailer_overwrite_mode = RemailerMode.ENABLED_V1
self.assertEqual(service_user.email, remailer_email)
# 7. remailer setup + remailer enabled + user overwrite
self.app.config['REMAILER_LIMIT_TO_USERS'] = None
service.remailer_mode = RemailerMode.ENABLED_V1
service_user.remailer_overwrite_mode = RemailerMode.DISABLED
self.assertEqual(service_user.email, user.primary_email.address)
def test_filter_query_by_email(self):
service = Service.query.filter_by(name='service1').first()
user = self.get_user()
self.app.config['REMAILER_DOMAIN'] = 'remailer.example.com'
remailer_email_v1 = remailer.build_v1_address(service.id, user.id)
remailer_email_v2 = remailer.build_v2_address(service.id, user.id)
email1 = user.primary_email
email2 = UserEmail(user=user, address='test2@example.com', verified=True)
db.session.add(email2)
service_user = ServiceUser.query.get((service.id, user.id))
all_service_users = ServiceUser.query.all()
cases = itertools.product(
# Input values
[
'test@example.com',
'test2@example.com',
'other@example.com',
remailer_email_v1,
remailer_email_v2,
],
# REMAILER_DOMAIN config
[None, 'remailer.example.com'],
# REMAILER_LIMIT config
[None, ['testuser', 'otheruser'], ['testadmin', 'otheruser']],
# service.remailer_mode
[RemailerMode.DISABLED, RemailerMode.ENABLED_V1, RemailerMode.ENABLED_V2],
# service.enable_email_preferences
[True, False],
# service.limit_access, service.access_group
[(False, None), (True, None), (True, self.get_admin_group()), (True, self.get_users_group())],
# service_user.service_email
[None, email1, email2],
# service_user.remailer_overwrite_mode
[None, RemailerMode.DISABLED, RemailerMode.ENABLED_V1, RemailerMode.ENABLED_V2],
)
for options in cases:
value = options[0]
self.app.config['REMAILER_DOMAIN'] = options[1]
self.app.config['REMAILER_LIMIT_TO_USERS'] = options[2]
service.remailer_mode = options[3]
service.enable_email_preferences = options[4]
service.limit_access, service.access_group = options[5]
service_user.service_email = options[6]
service_user.remailer_overwrite_mode = options[7]
a = {result for result in all_service_users if result.email == value}
b = set(ServiceUser.filter_query_by_email(ServiceUser.query, value).all())
if a != b:
self.fail(f'{a} != {b} with ' + repr(options))
import unittest
import datetime
from uffd.database import db
from uffd.models.session import Session, USER_AGENT_PARSER_SUPPORTED
from tests.utils import UffdTestCase
class TestSession(UffdTestCase):
def test_expire(self):
self.app.config['SESSION_LIFETIME_SECONDS'] = 100
self.app.config['PERMANENT_SESSION_LIFETIME'] = 10
user = self.get_user()
def make_session(created_age, last_used_age):
return Session(
user=user,
created=datetime.datetime.utcnow() - datetime.timedelta(seconds=created_age),
last_used=datetime.datetime.utcnow() - datetime.timedelta(seconds=last_used_age),
)
session1 = Session(user=user)
self.assertFalse(session1.expired)
session2 = make_session(0, 0)
self.assertFalse(session2.expired)
session3 = make_session(50, 5)
self.assertFalse(session3.expired)
session4 = make_session(50, 15)
self.assertTrue(session4.expired)
session5 = make_session(105, 5)
self.assertTrue(session5.expired)
session6 = make_session(105, 15)
self.assertTrue(session6.expired)
db.session.add_all([session1, session2, session3, session4, session5, session6])
db.session.commit()
self.assertEqual(set(Session.query.filter_by(expired=False).all()), {session1, session2, session3})
self.assertEqual(set(Session.query.filter_by(expired=True).all()), {session4, session5, session6})
def test_useragent_ua_parser(self):
if not USER_AGENT_PARSER_SUPPORTED:
self.skipTest('ua_parser not available')
session = Session(user_agent='Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:47.0) Gecko/20100101 Firefox/47.0')
self.assertEqual(session.user_agent_browser, 'Firefox')
self.assertEqual(session.user_agent_platform, 'Windows')
def test_useragent_no_ua_parser(self):
session = Session(user_agent='Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:47.0) Gecko/20100101 Firefox/47.0')
session.DISABLE_USER_AGENT_PARSER = True
self.assertEqual(session.user_agent_browser, 'Firefox')
self.assertEqual(session.user_agent_platform, 'Windows')
import datetime
from uffd.database import db
from uffd.models import Signup, User, FeatureFlag
from tests.utils import UffdTestCase, db_flush
def refetch_signup(signup):
db.session.add(signup)
db.session.commit()
id = signup.id
db.session.expunge(signup)
return Signup.query.get(id)
# We assume in all tests that Signup.validate and Signup.password.verify do
# not alter any state
class TestSignupModel(UffdTestCase):
def assert_validate_valid(self, signup):
valid, msg = signup.validate()
self.assertTrue(valid)
self.assertIsInstance(msg, str)
def assert_validate_invalid(self, signup):
valid, msg = signup.validate()
self.assertFalse(valid)
self.assertIsInstance(msg, str)
self.assertNotEqual(msg, '')
def assert_finish_success(self, signup, password):
self.assertIsNone(signup.user)
user, msg = signup.finish(password)
db.session.commit()
self.assertIsNotNone(user)
self.assertIsInstance(msg, str)
self.assertIsNotNone(signup.user)
def assert_finish_failure(self, signup, password):
prev_id = signup.user_id
user, msg = signup.finish(password)
self.assertIsNone(user)
self.assertIsInstance(msg, str)
self.assertNotEqual(msg, '')
self.assertEqual(signup.user_id, prev_id)
def test_password(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com')
self.assertFalse(signup.password.verify('notsecret'))
self.assertFalse(signup.password.verify(''))
self.assertFalse(signup.password.verify('wrongpassword'))
self.assertTrue(signup.set_password('notsecret'))
self.assertTrue(signup.password.verify('notsecret'))
self.assertFalse(signup.password.verify('wrongpassword'))
def test_expired(self):
# TODO: Find a better way to test this!
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assertFalse(signup.expired)
signup.created = created=datetime.datetime.utcnow() - datetime.timedelta(hours=49)
self.assertTrue(signup.expired)
def test_completed(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assertFalse(signup.completed)
signup.finish('notsecret')
db.session.commit()
self.assertTrue(signup.completed)
signup = refetch_signup(signup)
self.assertTrue(signup.completed)
def test_validate(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_validate_valid(signup)
self.assert_validate_valid(refetch_signup(signup))
def test_validate_completed(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_finish_success(signup, 'notsecret')
self.assert_validate_invalid(signup)
self.assert_validate_invalid(refetch_signup(signup))
def test_validate_expired(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com',
password='notsecret', created=datetime.datetime.utcnow()-datetime.timedelta(hours=49))
self.assert_validate_invalid(signup)
self.assert_validate_invalid(refetch_signup(signup))
def test_validate_loginname(self):
signup = Signup(loginname='', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_validate_invalid(signup)
self.assert_validate_invalid(refetch_signup(signup))
def test_validate_displayname(self):
signup = Signup(loginname='newuser', displayname='', mail='new@example.com', password='notsecret')
self.assert_validate_invalid(signup)
self.assert_validate_invalid(refetch_signup(signup))
def test_validate_mail(self):
signup = Signup(loginname='newuser', displayname='New User', mail='', password='notsecret')
self.assert_validate_invalid(signup)
self.assert_validate_invalid(refetch_signup(signup))
def test_validate_password(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com')
self.assertFalse(signup.set_password(''))
self.assert_validate_invalid(signup)
self.assert_validate_invalid(refetch_signup(signup))
def test_validate_exists(self):
signup = Signup(loginname='testuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_validate_invalid(signup)
self.assert_validate_invalid(refetch_signup(signup))
def test_finish(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_finish_success(signup, 'notsecret')
user = User.query.filter_by(loginname='newuser').one_or_none()
self.assertEqual(user.loginname, 'newuser')
self.assertEqual(user.displayname, 'New User')
self.assertEqual(user.primary_email.address, 'new@example.com')
def test_finish_completed(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_finish_success(signup, 'notsecret')
self.assert_finish_failure(refetch_signup(signup), 'notsecret')
def test_finish_expired(self):
# TODO: Find a better way to test this!
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com',
password='notsecret', created=datetime.datetime.utcnow()-datetime.timedelta(hours=49))
self.assert_finish_failure(signup, 'notsecret')
self.assert_finish_failure(refetch_signup(signup), 'notsecret')
def test_finish_wrongpassword(self):
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com')
self.assert_finish_failure(signup, '')
self.assert_finish_failure(signup, 'wrongpassword')
signup = refetch_signup(signup)
self.assert_finish_failure(signup, '')
self.assert_finish_failure(signup, 'wrongpassword')
signup = Signup(loginname='newuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_finish_failure(signup, 'wrongpassword')
self.assert_finish_failure(refetch_signup(signup), 'wrongpassword')
def test_finish_duplicate(self):
signup = Signup(loginname='testuser', displayname='New User', mail='new@example.com', password='notsecret')
self.assert_finish_failure(signup, 'notsecret')
self.assert_finish_failure(refetch_signup(signup), 'notsecret')
def test_finish_duplicate_email_strict_uniqueness(self):
FeatureFlag.unique_email_addresses.enable()
db.session.commit()
signup = Signup(loginname='newuser', displayname='New User', mail='test@example.com', password='notsecret')
self.assert_finish_failure(signup, 'notsecret')
def test_duplicate(self):
signup = Signup(loginname='newuser', displayname='New User', mail='test1@example.com', password='notsecret')
self.assert_validate_valid(signup)
db.session.add(signup)
db.session.commit()
signup1_id = signup.id
signup = Signup(loginname='newuser', displayname='New User', mail='test2@example.com', password='notsecret')
self.assert_validate_valid(signup)
db.session.add(signup)
db.session.commit()
signup2_id = signup.id
db_flush()
signup = Signup.query.get(signup2_id)
self.assert_finish_success(signup, 'notsecret')
db.session.commit()
db_flush()
signup = Signup.query.get(signup1_id)
self.assert_finish_failure(signup, 'notsecret')
user = User.query.filter_by(loginname='newuser').one_or_none()
self.assertEqual(user.primary_email.address, 'test2@example.com')
import datetime
import sqlalchemy
from uffd.database import db
from uffd.models import User, UserEmail, Group, FeatureFlag, IDAlreadyAllocatedError, IDRangeExhaustedError
from tests.utils import UffdTestCase, ModelTestCase
class TestUserModel(UffdTestCase):
def test_has_permission(self):
user_ = self.get_user() # has 'users' and 'uffd_access' group
admin = self.get_admin() # has 'users', 'uffd_access' and 'uffd_admin' group
self.assertTrue(user_.has_permission(None))
self.assertTrue(admin.has_permission(None))
self.assertTrue(user_.has_permission('users'))
self.assertTrue(admin.has_permission('users'))
self.assertFalse(user_.has_permission('notagroup'))
self.assertFalse(admin.has_permission('notagroup'))
self.assertFalse(user_.has_permission('uffd_admin'))
self.assertTrue(admin.has_permission('uffd_admin'))
self.assertFalse(user_.has_permission(['uffd_admin']))
self.assertTrue(admin.has_permission(['uffd_admin']))
self.assertFalse(user_.has_permission(['uffd_admin', 'notagroup']))
self.assertTrue(admin.has_permission(['uffd_admin', 'notagroup']))
self.assertFalse(user_.has_permission(['notagroup', 'uffd_admin']))
self.assertTrue(admin.has_permission(['notagroup', 'uffd_admin']))
self.assertTrue(user_.has_permission(['uffd_admin', 'users']))
self.assertTrue(admin.has_permission(['uffd_admin', 'users']))
self.assertTrue(user_.has_permission([['uffd_admin', 'users'], ['users', 'uffd_access']]))
self.assertTrue(admin.has_permission([['uffd_admin', 'users'], ['users', 'uffd_access']]))
self.assertFalse(user_.has_permission(['uffd_admin', ['users', 'notagroup']]))
self.assertTrue(admin.has_permission(['uffd_admin', ['users', 'notagroup']]))
def test_unix_uid_generation(self):
self.app.config['USER_MIN_UID'] = 10000
self.app.config['USER_MAX_UID'] = 18999
self.app.config['USER_SERVICE_MIN_UID'] = 19000
self.app.config['USER_SERVICE_MAX_UID'] = 19999
db.drop_all()
db.create_all()
user0 = User(loginname='user0', displayname='user0', primary_email_address='user0@example.com')
user1 = User(loginname='user1', displayname='user1', primary_email_address='user1@example.com')
user2 = User(loginname='user2', displayname='user2', primary_email_address='user2@example.com')
db.session.add_all([user0, user1, user2])
db.session.commit()
self.assertEqual(user0.unix_uid, 10000)
self.assertEqual(user1.unix_uid, 10001)
self.assertEqual(user2.unix_uid, 10002)
db.session.delete(user1)
db.session.commit()
user3 = User(loginname='user3', displayname='user3', primary_email_address='user3@example.com')
db.session.add(user3)
db.session.commit()
self.assertEqual(user3.unix_uid, 10003)
db.session.delete(user2)
db.session.commit()
user4 = User(loginname='user4', displayname='user4', primary_email_address='user4@example.com')
db.session.add(user4)
db.session.commit()
self.assertEqual(user4.unix_uid, 10004)
service0 = User(loginname='service0', displayname='service0', primary_email_address='service0@example.com', is_service_user=True)
service1 = User(loginname='service1', displayname='service1', primary_email_address='service1@example.com', is_service_user=True)
db.session.add_all([service0, service1])
db.session.commit()
self.assertEqual(service0.unix_uid, 19000)
self.assertEqual(service1.unix_uid, 19001)
def test_unix_uid_generation_overlapping(self):
self.app.config['USER_MIN_UID'] = 10000
self.app.config['USER_MAX_UID'] = 19999
self.app.config['USER_SERVICE_MIN_UID'] = 10000
self.app.config['USER_SERVICE_MAX_UID'] = 19999
db.drop_all()
db.create_all()
user0 = User(loginname='user0', displayname='user0', primary_email_address='user0@example.com')
service0 = User(loginname='service0', displayname='service0', primary_email_address='service0@example.com', is_service_user=True)
user1 = User(loginname='user1', displayname='user1', primary_email_address='user1@example.com')
db.session.add_all([user0, service0, user1])
db.session.commit()
self.assertEqual(user0.unix_uid, 10000)
self.assertEqual(service0.unix_uid, 10001)
self.assertEqual(user1.unix_uid, 10002)
def test_unix_uid_generation_overflow(self):
self.app.config['USER_MIN_UID'] = 10000
self.app.config['USER_MAX_UID'] = 10001
db.drop_all()
db.create_all()
user0 = User(loginname='user0', displayname='user0', primary_email_address='user0@example.com')
user1 = User(loginname='user1', displayname='user1', primary_email_address='user1@example.com')
db.session.add_all([user0, user1])
db.session.commit()
self.assertEqual(user0.unix_uid, 10000)
self.assertEqual(user1.unix_uid, 10001)
with self.assertRaises(sqlalchemy.exc.StatementError):
user2 = User(loginname='user2', displayname='user2', primary_email_address='user2@example.com')
db.session.add(user2)
db.session.commit()
def test_init_primary_email_address(self):
user = User(primary_email_address='foobar@example.com')
self.assertEqual(user.primary_email.address, 'foobar@example.com')
self.assertEqual(user.primary_email.verified, True)
self.assertEqual(user.primary_email.user, user)
user = User(primary_email_address='invalid')
self.assertEqual(user.primary_email.address, 'invalid')
self.assertEqual(user.primary_email.verified, True)
self.assertEqual(user.primary_email.user, user)
def test_set_primary_email_address(self):
user = User()
self.assertFalse(user.set_primary_email_address('invalid'))
self.assertIsNone(user.primary_email)
self.assertEqual(len(user.all_emails), 0)
self.assertTrue(user.set_primary_email_address('foobar@example.com'))
self.assertEqual(user.primary_email.address, 'foobar@example.com')
self.assertEqual(len(user.all_emails), 1)
self.assertFalse(user.set_primary_email_address('invalid'))
self.assertEqual(user.primary_email.address, 'foobar@example.com')
self.assertEqual(len(user.all_emails), 1)
self.assertTrue(user.set_primary_email_address('other@example.com'))
self.assertEqual(user.primary_email.address, 'other@example.com')
self.assertEqual(len(user.all_emails), 2)
self.assertEqual({user.all_emails[0].address, user.all_emails[1].address}, {'foobar@example.com', 'other@example.com'})
class TestUserEmailModel(UffdTestCase):
def test_normalize_address(self):
ref = UserEmail.normalize_address('foo@example.com')
self.assertEqual(ref, UserEmail.normalize_address('foo@example.com'))
self.assertEqual(ref, UserEmail.normalize_address('Foo@Example.Com'))
self.assertEqual(ref, UserEmail.normalize_address(' foo@example.com '))
self.assertNotEqual(ref, UserEmail.normalize_address('bar@example.com'))
self.assertNotEqual(ref, UserEmail.normalize_address('foo @example.com'))
# "No-Break Space" instead of SPACE (Unicode normalization + stripping)
self.assertEqual(ref, UserEmail.normalize_address('\u00A0foo@example.com '))
# Pre-composed "Angstrom Sign" vs. "A" + "Combining Ring Above" (Unicode normalization)
self.assertEqual(UserEmail.normalize_address('\u212B@example.com'), UserEmail.normalize_address('A\u030A@example.com'))
def test_address(self):
email = UserEmail()
self.assertIsNone(email.address)
self.assertIsNone(email.address_normalized)
email.address = 'Foo@example.com'
self.assertEqual(email.address, 'Foo@example.com')
self.assertEqual(email.address_normalized, UserEmail.normalize_address('Foo@example.com'))
with self.assertRaises(ValueError):
email.address = 'bar@example.com'
with self.assertRaises(ValueError):
email.address = None
def test_set_address(self):
email = UserEmail()
self.assertFalse(email.set_address('invalid'))
self.assertIsNone(email.address)
self.assertFalse(email.set_address(''))
self.assertFalse(email.set_address('@'))
self.app.config['REMAILER_DOMAIN'] = 'remailer.example.com'
self.assertFalse(email.set_address('foobar@remailer.example.com'))
self.assertFalse(email.set_address('v1-1-testuser@remailer.example.com'))
self.assertFalse(email.set_address('v1-1-testuser @ remailer.example.com'))
self.assertFalse(email.set_address('v1-1-testuser@REMAILER.example.com'))
self.assertFalse(email.set_address('v1-1-testuser@foobar@remailer.example.com'))
self.assertTrue(email.set_address('foobar@example.com'))
self.assertEqual(email.address, 'foobar@example.com')
def test_verified(self):
email = UserEmail(user=self.get_user(), address='foo@example.com')
db.session.add(email)
self.assertEqual(email.verified, False)
self.assertEqual(UserEmail.query.filter_by(address='foo@example.com', verified=True).count(), 0)
self.assertEqual(UserEmail.query.filter_by(address='foo@example.com', verified=False).count(), 1)
email.verified = True
self.assertEqual(email.verified, True)
self.assertEqual(UserEmail.query.filter_by(address='foo@example.com', verified=True).count(), 1)
self.assertEqual(UserEmail.query.filter_by(address='foo@example.com', verified=False).count(), 0)
with self.assertRaises(ValueError):
email.verified = False
self.assertEqual(email.verified, True)
with self.assertRaises(ValueError):
email.verified = None
self.assertEqual(email.verified, True)
def test_verification(self):
email = UserEmail(address='foo@example.com')
self.assertFalse(email.finish_verification('test'))
secret = email.start_verification()
self.assertTrue(email.verification_secret)
self.assertTrue(email.verification_secret.verify(secret))
self.assertFalse(email.verification_expired)
self.assertFalse(email.finish_verification('test'))
orig_expires = email.verification_expires
email.verification_expires = datetime.datetime.utcnow() - datetime.timedelta(days=1)
self.assertFalse(email.finish_verification(secret))
email.verification_expires = orig_expires
self.assertTrue(email.finish_verification(secret))
self.assertFalse(email.verification_secret)
self.assertTrue(email.verification_expired)
def test_enable_strict_constraints(self):
email = UserEmail(address='foo@example.com', user=self.get_user())
db.session.add(email)
db.session.commit()
self.assertIsNone(email.enable_strict_constraints)
FeatureFlag.unique_email_addresses.enable()
self.assertTrue(email.enable_strict_constraints)
FeatureFlag.unique_email_addresses.disable()
self.assertIsNone(email.enable_strict_constraints)
def assert_can_add_address(self, **kwargs):
user_email = UserEmail(**kwargs)
db.session.add(user_email)
db.session.commit()
db.session.delete(user_email)
db.session.commit()
def assert_cannot_add_address(self, **kwargs):
with self.assertRaises(sqlalchemy.exc.IntegrityError):
db.session.add(UserEmail(**kwargs))
db.session.commit()
db.session.rollback()
def test_unique_constraints_old(self):
# The same user cannot add the same exact address multiple times, but
# different users can have the same address
user = self.get_user()
admin = self.get_admin()
db.session.add(UserEmail(user=user, address='foo@example.com'))
db.session.add(UserEmail(user=user, address='bar@example.com', verified=True))
db.session.commit()
self.assert_can_add_address(user=user, address='foobar@example.com')
self.assert_can_add_address(user=user, address='foobar@example.com', verified=True)
self.assert_cannot_add_address(user=user, address='foo@example.com')
self.assert_can_add_address(user=user, address='FOO@example.com')
self.assert_cannot_add_address(user=user, address='bar@example.com')
self.assert_can_add_address(user=user, address='BAR@example.com')
self.assert_cannot_add_address(user=user, address='foo@example.com', verified=True)
self.assert_can_add_address(user=user, address='FOO@example.com', verified=True)
self.assert_cannot_add_address(user=user, address='bar@example.com', verified=True)
self.assert_can_add_address(user=user, address='BAR@example.com', verified=True)
self.assert_can_add_address(user=admin, address='foobar@example.com')
self.assert_can_add_address(user=admin, address='foobar@example.com', verified=True)
self.assert_can_add_address(user=admin, address='foo@example.com')
self.assert_can_add_address(user=admin, address='FOO@example.com')
self.assert_can_add_address(user=admin, address='bar@example.com')
self.assert_can_add_address(user=admin, address='BAR@example.com')
self.assert_can_add_address(user=admin, address='foo@example.com', verified=True)
self.assert_can_add_address(user=admin, address='FOO@example.com', verified=True)
self.assert_can_add_address(user=admin, address='bar@example.com', verified=True)
self.assert_can_add_address(user=admin, address='BAR@example.com', verified=True)
def test_unique_constraints_strict(self):
FeatureFlag.unique_email_addresses.enable()
# The same user cannot add the same (normalized) address multiple times,
# and different users cannot have the same verified (normalized) address
user = self.get_user()
admin = self.get_admin()
db.session.add(UserEmail(user=user, address='foo@example.com'))
db.session.add(UserEmail(user=user, address='bar@example.com', verified=True))
db.session.commit()
self.assert_can_add_address(user=user, address='foobar@example.com')
self.assert_can_add_address(user=user, address='foobar@example.com', verified=True)
self.assert_cannot_add_address(user=user, address='foo@example.com')
self.assert_cannot_add_address(user=user, address='FOO@example.com')
self.assert_cannot_add_address(user=user, address='bar@example.com')
self.assert_cannot_add_address(user=user, address='BAR@example.com')
self.assert_cannot_add_address(user=user, address='foo@example.com', verified=True)
self.assert_cannot_add_address(user=user, address='FOO@example.com', verified=True)
self.assert_cannot_add_address(user=user, address='bar@example.com', verified=True)
self.assert_cannot_add_address(user=user, address='BAR@example.com', verified=True)
self.assert_can_add_address(user=admin, address='foobar@example.com')
self.assert_can_add_address(user=admin, address='foobar@example.com', verified=True)
self.assert_can_add_address(user=admin, address='foo@example.com')
self.assert_can_add_address(user=admin, address='FOO@example.com')
self.assert_can_add_address(user=admin, address='bar@example.com')
self.assert_can_add_address(user=admin, address='BAR@example.com')
self.assert_can_add_address(user=admin, address='foo@example.com', verified=True)
self.assert_can_add_address(user=admin, address='FOO@example.com', verified=True)
self.assert_cannot_add_address(user=admin, address='bar@example.com', verified=True)
self.assert_cannot_add_address(user=admin, address='BAR@example.com', verified=True)
class TestIDAllocator(ModelTestCase):
def allocate_gids(self, *gids):
for gid in gids:
Group.unix_gid_allocator.allocate(gid)
def fetch_gid_allocations(self):
return [row[0] for row in db.session.execute(
db.select([Group.unix_gid_allocator.allocation_table])
.order_by(Group.unix_gid_allocator.allocation_table.c.id)
).fetchall()]
def test_empty(self):
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20000)
self.assertEqual(self.fetch_gid_allocations(), [20000])
def test_first(self):
self.allocate_gids(20000)
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20001)
self.assertEqual(self.fetch_gid_allocations(), [20000, 20001])
def test_out_of_range_before(self):
self.allocate_gids(19998)
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20000)
self.assertEqual(self.fetch_gid_allocations(), [19998, 20000])
def test_out_of_range_right_before(self):
self.allocate_gids(19999)
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20000)
self.assertEqual(self.fetch_gid_allocations(), [19999, 20000])
def test_out_of_range_after(self):
self.allocate_gids(20006)
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20000)
self.assertEqual(self.fetch_gid_allocations(), [20000, 20006])
def test_gap_at_beginning(self):
self.allocate_gids(20001)
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20000)
self.assertEqual(self.fetch_gid_allocations(), [20000, 20001])
def test_multiple_gaps(self):
self.allocate_gids(20000, 20001, 20003, 20005)
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20002)
self.assertEqual(self.fetch_gid_allocations(), [20000, 20001, 20002, 20003, 20005])
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20004)
self.assertEqual(self.fetch_gid_allocations(), [20000, 20001, 20002, 20003, 20004, 20005])
def test_last(self):
self.allocate_gids(20000, 20001, 20002, 20003, 20004)
self.assertEqual(Group.unix_gid_allocator.auto(20000, 20005), 20005)
self.assertEqual(self.fetch_gid_allocations(), [20000, 20001, 20002, 20003, 20004, 20005])
def test_overflow(self):
self.allocate_gids(20000, 20001, 20002, 20003, 20004, 20005)
with self.assertRaises(IDRangeExhaustedError):
Group.unix_gid_allocator.auto(20000, 20005)
self.assertEqual(self.fetch_gid_allocations(), [20000, 20001, 20002, 20003, 20004, 20005])
def test_conflict(self):
self.allocate_gids(20000)
with self.assertRaises(IDAlreadyAllocatedError):
self.allocate_gids(20000)
self.assertEqual(self.fetch_gid_allocations(), [20000])
class TestGroup(ModelTestCase):
def test_unix_gid_generation(self):
self.app.config['GROUP_MIN_GID'] = 20000
self.app.config['GROUP_MAX_GID'] = 49999
group0 = Group(name='group0', description='group0')
group1 = Group(name='group1', description='group1')
group2 = Group(name='group2', description='group2')
group3 = Group(name='group3', description='group3', unix_gid=20004)
db.session.add_all([group0, group1, group2, group3])
db.session.commit()
self.assertEqual(group0.unix_gid, 20000)
self.assertEqual(group1.unix_gid, 20001)
self.assertEqual(group2.unix_gid, 20002)
self.assertEqual(group3.unix_gid, 20004)
db.session.delete(group2)
db.session.commit()
group4 = Group(name='group4', description='group4')
group5 = Group(name='group5', description='group5')
db.session.add_all([group4, group5])
db.session.commit()
self.assertEqual(group4.unix_gid, 20003)
self.assertEqual(group5.unix_gid, 20005)
def test_unix_gid_generation_conflict(self):
self.app.config['GROUP_MIN_GID'] = 20000
self.app.config['GROUP_MAX_GID'] = 49999
group0 = Group(name='group0', description='group0', unix_gid=20023)
db.session.add(group0)
db.session.commit()
with self.assertRaises(IDAlreadyAllocatedError):
Group(name='group1', description='group1', unix_gid=20023)
def test_unix_gid_generation_overflow(self):
self.app.config['GROUP_MIN_GID'] = 20000
self.app.config['GROUP_MAX_GID'] = 20001
group0 = Group(name='group0', description='group0')
group1 = Group(name='group1', description='group1')
db.session.add_all([group0, group1])
db.session.commit()
self.assertEqual(group0.unix_gid, 20000)
self.assertEqual(group1.unix_gid, 20001)
db.session.commit()
with self.assertRaises(sqlalchemy.exc.StatementError):
group2 = Group(name='group2', description='group2')
db.session.add(group2)
db.session.commit()
version: 1
dn: uid=testuser,ou=users,dc=example,dc=com
objectClass: top
objectClass: organizationalPerson
objectClass: inetOrgPerson
objectClass: person
objectClass: posixAccount
cn: Test User
displayName: Test User
gidNumber: 20001
givenName: Test User
homeDirectory: /home/testuser
mail: testuser@example.com
sn:: IA==
uid: testuser
uidNumber: 10000
userPassword: {ssha512}P6mPgcE974bMZkYHnowsXheE74lqtR0HemVUjZxZT7cgPlEhE7fSU1DYEhOx1ZYhOTuE7Ei3EaMFSSoi9Jqf5MHHcjG9oVWL
dn: uid=testadmin,ou=users,dc=example,dc=com
objectClass: top
objectClass: organizationalPerson
objectClass: inetOrgPerson
objectClass: person
objectClass: posixAccount
cn: Test Admin
displayName: Test Admin
gidNumber: 20001
givenName: Test Admin
homeDirectory: /home/testadmin
mail: testadmin@example.com
sn:: IA==
uid: testadmin
uidNumber: 10001
userPassword: {ssha512}SGARsM9lNP9PQ4S+M/pmA7MIDvdyF9WZ8Ki2JvjvxIlMLene5+s+M+Qfi0lfJHOSqucd6CR0F7vDl32rEJNd1ZPCLbCO20pB
dn: uid=test,ou=postfix,dc=example,dc=com
objectClass: top
objectClass: postfixVirtual
uid: test
mailacceptinggeneralid: test1@example.com
mailacceptinggeneralid: test2@example.com
maildrop: testuser@mail.example.com
uid=testuser,ou=users,dc=example,dc=com
uid=testadmin,ou=users,dc=example,dc=com
uid=newuser,ou=users,dc=example,dc=com
uid=newuser1,ou=users,dc=example,dc=com
uid=newuser2,ou=users,dc=example,dc=com
uid=newuser3,ou=users,dc=example,dc=com
uid=newuser4,ou=users,dc=example,dc=com
uid=newuser5,ou=users,dc=example,dc=com
uid=newuser6,ou=users,dc=example,dc=com
uid=newuser7,ou=users,dc=example,dc=com
uid=newuser8,ou=users,dc=example,dc=com
uid=newuser9,ou=users,dc=example,dc=com
uid=newuser10,ou=users,dc=example,dc=com
uid=newuser11,ou=users,dc=example,dc=com
uid=newuser12,ou=users,dc=example,dc=com
uid=test,ou=postfix,dc=example,dc=com
uid=test1,ou=postfix,dc=example,dc=com
version: 1
dn: cn=users,ou=groups,dc=example,dc=com
changetype: modify
add: uniqueMember
uniqueMember: uid=testuser,ou=users,dc=example,dc=com
uniqueMember: uid=testadmin,ou=users,dc=example,dc=com
dn: cn=uffd_access,ou=groups,dc=example,dc=com
changetype: modify
add: uniqueMember
uniqueMember: uid=testuser,ou=users,dc=example,dc=com
uniqueMember: uid=testadmin,ou=users,dc=example,dc=com
dn: cn=uffd_admin,ou=groups,dc=example,dc=com
changetype: modify
add: uniqueMember
uniqueMember: uid=testadmin,ou=users,dc=example,dc=com
{
"entries": [
{
"dn": "uid=testuser,ou=users,dc=example,dc=com",
"raw": {
"cn": [
"Test User"
],
"createTimestamp": [
"20200101000000Z"
],
"creatorsName": [
"cn=admin,dc=example,dc=com"
],
"displayName": [
"Test User"
],
"entryDN": [
"uid=testuser,ou=users,dc=example,dc=com"
],
"entryUUID": [
"75e62c6a-03c2-11eb-adc1-0242ac120002"
],
"gidNumber": [
"20001"
],
"givenName": [
"Test User"
],
"hasSubordinates": [
"FALSE"
],
"homeDirectory": [
"/home/testuser"
],
"mail": [
"testuser@example.com"
],
"memberOf": [
"cn=uffd_access,ou=groups,dc=example,dc=com",
"cn=users,ou=groups,dc=example,dc=com"
],
"modifiersName": [
"cn=admin,dc=example,dc=com"
],
"modifyTimestamp": [
"20200101000000Z"
],
"objectClass": [
"top",
"inetOrgPerson",
"organizationalPerson",
"person",
"posixAccount"
],
"sn": [
" "
],
"structuralObjectClass": [
"inetOrgPerson"
],
"subschemaSubentry": [
"cn=Subschema"
],
"uid": [
"testuser"
],
"uidNumber": [
"10000"
],
"userPassword": [
"userpassword"
]
}
},
{
"dn": "uid=testadmin,ou=users,dc=example,dc=com",
"raw": {
"cn": [
"Test Admin"
],
"createTimestamp": [
"20200101000000Z"
],
"creatorsName": [
"cn=admin,dc=example,dc=com"
],
"displayName": [
"Test Admin"
],
"entryDN": [
"uid=testadmin,ou=users,dc=example,dc=com"
],
"entryUUID": [
"678c8470-03c2-11eb-adc1-0242ac120002"
],
"gidNumber": [
"20001"
],
"givenName": [
"Test Admin"
],
"hasSubordinates": [
"FALSE"
],
"homeDirectory": [
"/home/testadmin"
],
"mail": [
"testadmin@example.com"
],
"memberOf": [
"cn=users,ou=groups,dc=example,dc=com",
"cn=uffd_access,ou=groups,dc=example,dc=com",
"cn=uffd_admin,ou=groups,dc=example,dc=com"
],
"modifiersName": [
"cn=admin,dc=example,dc=com"
],
"modifyTimestamp": [
"20200101000000Z"
],
"objectClass": [
"top",
"inetOrgPerson",
"organizationalPerson",
"person",
"posixAccount"
],
"sn": [
" "
],
"structuralObjectClass": [
"inetOrgPerson"
],
"subschemaSubentry": [
"cn=Subschema"
],
"uid": [
"testadmin"
],
"uidNumber": [
"10001"
],
"userPassword": [
"adminpassword"
]
}
},
{
"dn": "cn=users,ou=groups,dc=example,dc=com",
"raw": {
"cn": [
"users"
],
"createTimestamp": [
"20200101000000Z"
],
"creatorsName": [
"cn=admin,dc=example,dc=com"
],
"description": [
"Base group for all users"
],
"entryDN": [
"cn=users,ou=groups,dc=example,dc=com"
],
"entryUUID": [
"1aec0e8c-03c3-11eb-adc1-0242ac120002"
],
"gidNumber": [
"20001"
],
"hasSubordinates": [
"FALSE"
],
"modifiersName": [
"cn=admin,dc=example,dc=com"
],
"modifyTimestamp": [
"20200101000000Z"
],
"objectClass": [
"posixGroup",
"groupOfUniqueNames",
"top"
],
"structuralObjectClass": [
"groupOfUniqueNames"
],
"subschemaSubentry": [
"cn=Subschema"
],
"uniqueMember": [
"cn=dummy,ou=system,dc=example,dc=com",
"uid=testuser,ou=users,dc=example,dc=com",
"uid=testadmin,ou=users,dc=example,dc=com"
]
}
},
{
"dn": "cn=uffd_access,ou=groups,dc=example,dc=com",
"raw": {
"cn": [
"uffd_access"
],
"createTimestamp": [
"20200101000000Z"
],
"creatorsName": [
"cn=admin,dc=example,dc=com"
],
"description": [
"User access to uffd selfservice"
],
"entryDN": [
"cn=uffd_access,ou=groups,dc=example,dc=com"
],
"entryUUID": [
"4fc8dd60-03c3-11eb-adc1-0242ac120002"
],
"gidNumber": [
"20002"
],
"hasSubordinates": [
"FALSE"
],
"modifiersName": [
"cn=admin,dc=example,dc=com"
],
"modifyTimestamp": [
"20200101000000Z"
],
"objectClass": [
"posixGroup",
"groupOfUniqueNames",
"top"
],
"structuralObjectClass": [
"groupOfUniqueNames"
],
"subschemaSubentry": [
"cn=Subschema"
],
"uniqueMember": [
"cn=dummy,ou=system,dc=example,dc=com",
"uid=testuser,ou=users,dc=example,dc=com",
"uid=testadmin,ou=users,dc=example,dc=com"
]
}
},
{
"dn": "cn=uffd_admin,ou=groups,dc=example,dc=com",
"raw": {
"cn": [
"uffd_admin"
],
"createTimestamp": [
"20200101000000Z"
],
"creatorsName": [
"cn=admin,dc=example,dc=com"
],
"description": [
"Admin access to uffd selfservice"
],
"entryDN": [
"cn=uffd_admin,ou=groups,dc=example,dc=com"
],
"entryUUID": [
"b5d869d6-03c3-11eb-adc1-0242ac120002"
],
"gidNumber": [
"20003"
],
"hasSubordinates": [
"FALSE"
],
"modifiersName": [
"cn=admin,dc=example,dc=com"
],
"modifyTimestamp": [
"20200101000000Z"
],
"objectClass": [
"posixGroup",
"groupOfUniqueNames",
"top"
],
"structuralObjectClass": [
"groupOfUniqueNames"
],
"subschemaSubentry": [
"cn=Subschema"
],
"uniqueMember": [
"cn=dummy,ou=system,dc=example,dc=com",
"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"
]
}
}
]
}
{
"raw": {
"altServer": [],
"configContext": [
"cn=config"
],
"entryDN": [
""
],
"namingContexts": [
"dc=example,dc=com"
],
"objectClass": [
"top",
"OpenLDAProotDSE"
],
"structuralObjectClass": [
"OpenLDAProotDSE"
],
"subschemaSubentry": [
"cn=Subschema"
],
"supportedCapabilities": [],
"supportedControl": [
"2.16.840.1.113730.3.4.18",
"2.16.840.1.113730.3.4.2",
"1.3.6.1.4.1.4203.1.10.1",
"1.3.6.1.1.22",
"1.2.840.113556.1.4.319",
"1.2.826.0.1.3344810.2.3",
"1.3.6.1.1.13.2",
"1.3.6.1.1.13.1",
"1.3.6.1.1.12"
],
"supportedExtension": [
"1.3.6.1.4.1.1466.20037",
"1.3.6.1.4.1.4203.1.11.1",
"1.3.6.1.4.1.4203.1.11.3",
"1.3.6.1.1.8"
],
"supportedFeatures": [
"1.3.6.1.1.14",
"1.3.6.1.4.1.4203.1.5.1",
"1.3.6.1.4.1.4203.1.5.2",
"1.3.6.1.4.1.4203.1.5.3",
"1.3.6.1.4.1.4203.1.5.4",
"1.3.6.1.4.1.4203.1.5.5"
],
"supportedLDAPVersion": [
"3"
],
"supportedSASLMechanisms": [
"DIGEST-MD5",
"CRAM-MD5",
"NTLM"
],
"vendorName": [],
"vendorVersion": []
},
"type": "DsaInfo"
}
This diff is collapsed.
......@@ -2,7 +2,7 @@ import unittest
from flask import Flask, Blueprint, session, url_for
from uffd.csrf import csrf_bp, csrf_protect
from uffd.csrf import bp as csrf_bp, csrf_protect
uid_counter = 0
......
This diff is collapsed.
import datetime
from urllib.parse import urlparse, parse_qs
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.models import DeviceLoginConfirmation
from uffd.oauth2.models import OAuth2Client, OAuth2DeviceLoginInitiation
from uffd import create_app, db, ldap
from utils import dump, UffdTestCase
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 = self.get_user() # has 'users' and 'uffd_access' group
admin = self.get_admin() # has 'users', 'uffd_access' and 'uffd_admin' group
client = OAuth2Client('test', '', [''], ['uffd_admin', ['users', 'notagroup']])
self.assertFalse(client.access_allowed(user))
self.assertTrue(client.access_allowed(admin))
# More required_group values are tested by TestUserModel.test_has_permission
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 assert_authorization(self, r):
while True:
if r.status_code != 302 or r.location.startswith('http://localhost:5009/callback'):
break
r = self.client.get(r.location, 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'], ['teststate'])
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 = self.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'))
def test_authorization(self):
self.login_as('user')
r = self.client.get(path=url_for('oauth2.authorize', response_type='code', client_id='test', state='teststate', redirect_uri='http://localhost:5009/callback'), follow_redirects=False)
self.assert_authorization(r)
def test_authorization_devicelogin_start(self):
ref = url_for('oauth2.authorize', response_type='code', client_id='test', state='teststate', redirect_uri='http://localhost:5009/callback')
r = self.client.get(path=url_for('session.devicelogin_start', ref=ref), follow_redirects=True)
# check response
initiation = OAuth2DeviceLoginInitiation.query.filter_by(id=session['devicelogin_id'], secret=session['devicelogin_secret']).one()
self.assertEqual(r.status_code, 200)
self.assertFalse(initiation.expired)
self.assertEqual(initiation.oauth2_client_id, 'test')
self.assertIsNotNone(initiation.description)
def test_authorization_devicelogin_auth(self):
with self.client.session_transaction() as _session:
initiation = OAuth2DeviceLoginInitiation(oauth2_client_id='test')
db.session.add(initiation)
confirmation = DeviceLoginConfirmation(initiation=initiation, user=self.get_user())
db.session.add(confirmation)
db.session.commit()
_session['devicelogin_id'] = initiation.id
_session['devicelogin_secret'] = initiation.secret
code = confirmation.code
self.client.get(path='/')
ref = url_for('oauth2.authorize', response_type='code', client_id='test', state='teststate', redirect_uri='http://localhost:5009/callback')
r = self.client.post(path=url_for('session.devicelogin_submit', ref=ref), data={'confirmation-code': code}, follow_redirects=False)
self.assert_authorization(r)
import unittest
from uffd.password_hash import *
class TestPasswordHashRegistry(unittest.TestCase):
def test(self):
registry = PasswordHashRegistry()
@registry.register
class TestPasswordHash:
METHOD_NAME = 'test'
def __init__(self, value, **kwargs):
self.value = value
self.kwargs = kwargs
@registry.register
class Test2PasswordHash:
METHOD_NAME = 'test2'
result = registry.parse('{test}data', key='value')
self.assertIsInstance(result, TestPasswordHash)
self.assertEqual(result.value, '{test}data')
self.assertEqual(result.kwargs, {'key': 'value'})
with self.assertRaises(ValueError):
registry.parse('{invalid}data')
with self.assertRaises(ValueError):
registry.parse('invalid')
with self.assertRaises(ValueError):
registry.parse('{invalid')
class TestPasswordHash(unittest.TestCase):
def setUp(self):
class TestPasswordHash(PasswordHash):
@classmethod
def from_password(cls, password):
cls(build_value(cls.METHOD_NAME, password))
def verify(self, password):
return self.data == password
class TestPasswordHash1(TestPasswordHash):
METHOD_NAME = 'test1'
class TestPasswordHash2(TestPasswordHash):
METHOD_NAME = 'test2'
self.TestPasswordHash1 = TestPasswordHash1
self.TestPasswordHash2 = TestPasswordHash2
def test(self):
obj = self.TestPasswordHash1('{test1}data')
self.assertEqual(obj.value, '{test1}data')
self.assertEqual(obj.data, 'data')
self.assertIs(obj.target_cls, self.TestPasswordHash1)
self.assertFalse(obj.needs_rehash)
def test_invalid(self):
with self.assertRaises(ValueError):
self.TestPasswordHash1('invalid')
with self.assertRaises(ValueError):
self.TestPasswordHash1('{invalid}data')
with self.assertRaises(ValueError):
self.TestPasswordHash1('{test2}data')
def test_target_cls(self):
obj = self.TestPasswordHash1('{test1}data', target_cls=self.TestPasswordHash1)
self.assertEqual(obj.value, '{test1}data')
self.assertEqual(obj.data, 'data')
self.assertIs(obj.target_cls, self.TestPasswordHash1)
self.assertFalse(obj.needs_rehash)
obj = self.TestPasswordHash1('{test1}data', target_cls=self.TestPasswordHash2)
self.assertEqual(obj.value, '{test1}data')
self.assertEqual(obj.data, 'data')
self.assertIs(obj.target_cls, self.TestPasswordHash2)
self.assertTrue(obj.needs_rehash)
obj = self.TestPasswordHash1('{test1}data', target_cls=PasswordHash)
self.assertEqual(obj.value, '{test1}data')
self.assertEqual(obj.data, 'data')
self.assertIs(obj.target_cls, PasswordHash)
self.assertFalse(obj.needs_rehash)
class TestPlaintextPasswordHash(unittest.TestCase):
def test_verify(self):
obj = PlaintextPasswordHash('{plain}password')
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
def test_from_password(self):
obj = PlaintextPasswordHash.from_password('password')
self.assertEqual(obj.value, '{plain}password')
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
class TestHashlibPasswordHash(unittest.TestCase):
def test_verify(self):
obj = SHA512PasswordHash('{sha512}sQnzu7wkTrgkQZF+0G1hi5AI3Qmzvv0bXgc5THBqi7mAsdd4Xll27ASbRt9fEyavWi6m0QP9B8lThf+rDKy8hg==')
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
def test_from_password(self):
obj = SHA512PasswordHash.from_password('password')
self.assertIsNotNone(obj.value)
self.assertTrue(obj.value.startswith('{sha512}'))
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
class TestSaltedHashlibPasswordHash(unittest.TestCase):
def test_verify(self):
obj = SaltedSHA512PasswordHash('{ssha512}dOeDLmVpHJThhHeag10Hm2g4T7s3SBE6rGHcXUolXJHVufY4qT782rwZ/0XE6cuLcBZ0KpnwmUzRpAEtZBdv+JYEEtZQs/uC')
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
def test_from_password(self):
obj = SaltedSHA512PasswordHash.from_password('password')
self.assertIsNotNone(obj.value)
self.assertTrue(obj.value.startswith('{ssha512}'))
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
class TestCryptPasswordHash(unittest.TestCase):
def test_verify(self):
obj = CryptPasswordHash('{crypt}$5$UbTTMBH9NRurlQcX$bUiUTyedvmArlVt.62ZLRV80e2v3DjcBp/tSDkP2imD')
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
def test_from_password(self):
obj = CryptPasswordHash.from_password('password')
self.assertIsNotNone(obj.value)
self.assertTrue(obj.value.startswith('{crypt}'))
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
class TestArgon2PasswordHash(unittest.TestCase):
def test_verify(self):
obj = Argon2PasswordHash('{argon2}$argon2id$v=19$m=102400,t=2,p=8$Jc8LpCgPLjwlN/7efHLvwQ$ZqSg3CFb2/hBb3X8hOq4aw')
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
obj = Argon2PasswordHash('{argon2}$invalid$')
self.assertFalse(obj.verify('password'))
def test_from_password(self):
obj = Argon2PasswordHash.from_password('password')
self.assertIsNotNone(obj.value)
self.assertTrue(obj.value.startswith('{argon2}'))
self.assertTrue(obj.verify('password'))
self.assertFalse(obj.verify('notpassword'))
def test_needs_rehash(self):
obj = Argon2PasswordHash('{argon2}$argon2id$v=19$m=102400,t=2,p=8$Jc8LpCgPLjwlN/7efHLvwQ$ZqSg3CFb2/hBb3X8hOq4aw')
self.assertFalse(obj.needs_rehash)
obj = Argon2PasswordHash('{argon2}$argon2id$v=19$m=102400,t=2,p=8$Jc8LpCgPLjwlN/7efHLvwQ$ZqSg3CFb2/hBb3X8hOq4aw', target_cls=PlaintextPasswordHash)
self.assertTrue(obj.needs_rehash)
obj = Argon2PasswordHash('{argon2}$argon2d$v=19$m=102400,t=2,p=8$kshPgLU1+h72l/Z8QWh8Ig$tYerKCe/5I2BCPKu8hCl2w')
self.assertTrue(obj.needs_rehash)
obj = Argon2PasswordHash('{argon2}$argon2id$v=19$m=102400,t=1,p=8$aa6i4vg/szKX5xHVGFaAeQ$v6j0ltuVqQaZlmuepaVJ1A')
self.assertTrue(obj.needs_rehash)
class TestInvalidPasswordHash(unittest.TestCase):
def test(self):
obj = InvalidPasswordHash('test')
self.assertEqual(obj.value, 'test')
self.assertFalse(obj.verify('test'))
self.assertTrue(obj.needs_rehash)
self.assertFalse(obj)
obj = InvalidPasswordHash(None)
self.assertIsNone(obj.value)
self.assertFalse(obj.verify('test'))
self.assertTrue(obj.needs_rehash)
self.assertFalse(obj)
class TestPasswordWrapper(unittest.TestCase):
def setUp(self):
class Test:
password_hash = None
password = PasswordHashAttribute('password_hash', PlaintextPasswordHash)
self.test = Test()
def test_get_none(self):
self.test.password_hash = None
obj = self.test.password
self.assertIsInstance(obj, InvalidPasswordHash)
self.assertEqual(obj.value, None)
self.assertTrue(obj.needs_rehash)
def test_get_valid(self):
self.test.password_hash = '{plain}password'
obj = self.test.password
self.assertIsInstance(obj, PlaintextPasswordHash)
self.assertEqual(obj.value, '{plain}password')
self.assertFalse(obj.needs_rehash)
def test_get_needs_rehash(self):
self.test.password_hash = '{ssha512}dOeDLmVpHJThhHeag10Hm2g4T7s3SBE6rGHcXUolXJHVufY4qT782rwZ/0XE6cuLcBZ0KpnwmUzRpAEtZBdv+JYEEtZQs/uC'
obj = self.test.password
self.assertIsInstance(obj, SaltedSHA512PasswordHash)
self.assertEqual(obj.value, '{ssha512}dOeDLmVpHJThhHeag10Hm2g4T7s3SBE6rGHcXUolXJHVufY4qT782rwZ/0XE6cuLcBZ0KpnwmUzRpAEtZBdv+JYEEtZQs/uC')
self.assertTrue(obj.needs_rehash)
def test_set(self):
self.test.password = 'password'
self.assertEqual(self.test.password_hash, '{plain}password')
def test_set_none(self):
self.test.password = None
self.assertIsNone(self.test.password_hash)
import time
from uffd.models.ratelimit import get_addrkey, format_delay, Ratelimit
from flask import Flask, Blueprint, session, url_for
from uffd.ratelimit import get_addrkey, format_delay, Ratelimit, RatelimitEvent
from utils import UffdTestCase
from tests.utils import UffdTestCase
class TestRatelimit(UffdTestCase):
def test_limiting(self):
......@@ -48,19 +44,3 @@ class TestRatelimit(UffdTestCase):
self.assertIsInstance(format_delay(120), str)
self.assertIsInstance(format_delay(3600), str)
self.assertIsInstance(format_delay(4000), str)
def test_cleanup(self):
ratelimit = Ratelimit('test', 1, 1)
ratelimit.log('')
ratelimit.log('1')
ratelimit.log('2')
ratelimit.log('3')
ratelimit.log('4')
time.sleep(1)
ratelimit.log('5')
self.assertEqual(RatelimitEvent.query.filter(RatelimitEvent.name == 'test').count(), 6)
ratelimit.cleanup()
self.assertEqual(RatelimitEvent.query.filter(RatelimitEvent.name == 'test').count(), 1)
time.sleep(1)
ratelimit.cleanup()
self.assertEqual(RatelimitEvent.query.filter(RatelimitEvent.name == 'test').count(), 0)
This diff is collapsed.
This diff is collapsed.
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 utils import dump, UffdTestCase
class TestServices(UffdTestCase):
def setUpApp(self):
self.app.config['SERVICES'] = [
{
'title': 'Service Title',
'subtitle': 'Service Subtitle',
'description': 'Short description of the service as plain text',
'url': 'https://example.com/',
'logo_url': '/static/fairy-dust-color.png',
'required_group': 'users',
'permission_levels': [
{'name': 'Moderator', 'required_group': 'moderators'},
{'name': 'Admin', 'required_group': 'uffd_admin'},
],
'confidential': True,
'groups': [
{'name': 'Group "crew_crew"', 'required_group': 'users'},
{'name': 'Group "crew_logistik"', 'required_group': 'uffd_admin'},
],
'infos': [
{'title': 'Documentation', 'html': '<p>Some information about the service as html</p>', 'required_group': 'users'},
],
'links': [
{'title': 'Link to an external site', 'url': '#', 'required_group': 'users'},
],
},
{
'title': 'Minimal Service Title',
}
]
self.app.config['SERVICES_PUBLIC'] = True
def test_index(self):
r = self.client.get(path=url_for('services.index'))
dump('services_index_public', r)
self.assertEqual(r.status_code, 200)
self.assertNotIn(b'https://example.com/', r.data)
self.login_as('user')
r = self.client.get(path=url_for('services.index'))
dump('services_index', r)
self.assertEqual(r.status_code, 200)
self.assertIn(b'https://example.com/', r.data)
This diff is collapsed.
This diff is collapsed.