CVE-2022-3362 (Python, published 2022-11-14T21:15Z)
Description: value: Insufficient Session Expiration in GitHub repository ikus060/rdiffweb prior to 2.5.0.
Severity: nan (CVSS v3 base score: 9.8)
Weakness: CWE-613 \u2014 Insufficient Session Expiration
According to WASC, "Insufficient Session Expiration is when a web site permits an attacker to reuse old session credentials or session IDs for authorization."
Repository: https://github.com/ikus060/rdiffweb
Files changed:
README.md
rdiffweb/controller/api.py
rdiffweb/controller/page_admin_session.py
rdiffweb/controller/page_admin_users.py
rdiffweb/controller/page_delete.py
rdiffweb/controller/page_locations.py
rdiffweb/controller/page_pref_general.py
rdiffweb/controller/page_pref_mfa.py
rdiffweb/controller/page_pref_notification.py
rdiffweb/controller/page_pref_session.py
rdiffweb/controller/page_pref_sshkeys.py
rdiffweb/controller/page_pref_tokens.py
rdiffweb/controller/tests/test_api.py
rdiffweb/controller/tests/test_controller.py
rdiffweb/controller/tests/test_page_admin.py
rdiffweb/controller/tests/test_page_admin_users.py
rdiffweb/controller/tests/test_page_browse.py
rdiffweb/controller/tests/test_page_delete.py
rdiffweb/controller/tests/test_page_graphs.py
rdiffweb/controller/tests/test_page_history.py
rdiffweb/controller/tests/test_page_login.py
rdiffweb/controller/tests/test_page_logs.py
rdiffweb/controller/tests/test_page_mfa.py
rdiffweb/controller/tests/test_page_prefs_general.py
rdiffweb/controller/tests/test_page_prefs_mfa.py
rdiffweb/controller/tests/test_page_prefs_ssh.py
rdiffweb/controller/tests/test_page_prefs_tokens.py
rdiffweb/controller/tests/test_page_restore.py
rdiffweb/controller/tests/test_page_settings.py
rdiffweb/controller/tests/test_page_settings_remove_older.py
rdiffweb/controller/tests/test_page_settings_set_encoding.py
rdiffweb/controller/tests/test_page_status.py
rdiffweb/core/login.py
rdiffweb/core/model/__init__.py
rdiffweb/core/model/_repo.py
rdiffweb/core/model/_session.py
rdiffweb/core/model/_token.py
rdiffweb/core/model/_user.py
rdiffweb/core/model/tests/test_repo.py
rdiffweb/core/model/tests/test_token.py
rdiffweb/core/model/tests/test_user.py
rdiffweb/core/tests/test_login.py
rdiffweb/core/tests/test_notification.py
rdiffweb/core/tests/test_quota.py
rdiffweb/core/tests/test_rdw_templating.py
rdiffweb/rdw_app.py
rdiffweb/templates/access_token_added.html
rdiffweb/test.py
rdiffweb/tools/db.py
Commit message: Delete user's session on password change

* Revisit add, delete commit function
* Clean-up Access Token using a scheduled job


Vulnerable code:
        if userobj.validate_access_token(password):
        u.refresh_repos()
                cherrypy.engine.publish('schedule_task', repo.delete, path)
        self.app.currentuser.refresh_repos()
        user.add()
            user.refresh_repos(delete=True)
        userobj.add()
        data.add()
        UserObject.add_user('test', 'test123')
        UserObject.add_user('newuser')
        UserObject.add_user('test1')
                user.delete()
        user.add()
        user.add()
        RepoObject(userid=userobj.userid, repopath='testcases/broker-repo').add()
        RepoObject(userid=userobj.userid, repopath='testcases/testcases').add()
        user.add()
        user_obj.add()
        admin.add()
        admin.add()
        admin.add()
        admin.add()
        admin.add()
        admin.add()
        UserObject.add_user('otheruser', password='password')
        admin.add()
        userobj.add()
        userobj.add()
        userobj.add()
        userobj.add()
from rdiffweb.core.model import RepoObject, UserObject
        self.listener.user_password_changed.reset_mock()
        RepoObject(userid=userobj.userid, repopath='invalid').add()
        userobj.add()
        userobj.add()
        userobj.add()
        admin.add()
        admin.add()
        admin.add()
        admin.add()
        admin.add()
        userobj.add()
                ).add()
            userobj.add()
            row.add()
            row.add()
            as_user.refresh_repos()
        super().delete()
        session.add()
import datetime
                SshKey(userid=self.userid, fingerprint=key.fingerprint, key=key.getvalue()).add()
                SshKey.session.rollback()
            obj = Token(userid=self.userid, name=name, hash_token=hash_password(token), expiration_time=expiration_time)
            obj.add()
            Token.session.rollback()
        Base.delete(self)
        logger.info("updating user password [%s]", self.username)
            # If token expired. Let delete it.
                access_token.delete()
                # When it matches, let update the record.
                access_token.access_time = datetime.datetime.utcnow
                return True
        RepoObject(userid=userobj.userid, repopath='/testcases').add()
        RepoObject(userid=userobj.userid, repopath='/testcases/home/admin/data').add()
        RepoObject(userid=userobj.userid, repopath='/testcases').add()

        repo_obj.add()
        repo_obj.add()
        repo_obj.add()
        repo_obj = RepoObject(user=userobj, repopath='repopath').add()
        repo_obj = RepoObject(user=userobj, repopath='repopath').add()
        repo_obj = RepoObject(user=userobj, repopath='repopath').add()
import datetime
from rdiffweb.core.model import DuplicateSSHKeyError, RepoObject, Token, UserObject
        self.assertIsNotNone(userobj)
        self.assertIsNotNone(userobj)
        UserObject.add_user('denise')
        UserObject.add_user('annik')
        user.add()
        user.add()
            user.add()
        user.add()
        user.add()
        RepoObject(userid=userobj.userid, repopath='invalid').add()
        RepoObject(userid=userobj.userid, repopath='invalid').add()
    def test_add_access_token(self):
        # Given a user with an email
        userobj = UserObject.get_user(self.USERNAME)
        userobj.email = 'test@examples.com'
        userobj.add()
        # When adding a new token
        token = userobj.add_access_token('test')
        # Then a new token get created
        self.assertTrue(token)
        tokenobj = Token.query.filter(Token.userid == userobj.userid).first()
        self.assertTrue(tokenobj)
        self.assertEqual(None, tokenobj.expiration_time)
        self.assertEqual(None, tokenobj.access_time)
        # Then an email is sent to the user.
        self.listener.access_token_added.assert_called_once_with(userobj, 'test')
        self.listener.queue_mail.assert_called_once()

    def test_add_access_token_duplicate_name(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test')
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When adding a new token with the same name
        with self.assertRaises(ValueError):
            userobj.add_access_token('test')
        # Then token is not created
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # Then an email is not sent.
        self.listener.access_token_added.assert_called_once_with(userobj, 'test')

    def test_delete_access_token(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test')
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When deleting an access token
        userobj.delete_access_token('test')
        # Then Token get deleted
        self.assertEqual(0, Token.query.filter(Token.userid == userobj.userid).count())

    def test_delete_access_token_invalid(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test')
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When deleting an invalid access token
        with self.assertRaises(ValueError):
            userobj.delete_access_token('invalid')
        # Then Token not deleted
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())

    def test_delete_user_remove_access_tokens(self):
        # Given a user with an existing token
        userobj = UserObject.add_user('testuser', 'password')
        userobj.add_access_token('test')
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When deleting the user
        userobj.delete()
        # Then Token get deleted
        self.assertEqual(0, Token.query.filter(Token.userid == userobj.userid).count())

    def test_verify_access_token(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        token = userobj.add_access_token('test')
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When validating the token
        # Then token is valid
        self.assertTrue(userobj.validate_access_token(token))

    def test_verify_access_token_with_expired(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        token = userobj.add_access_token(
            'test', expiration_time=datetime.datetime.now() - datetime.timedelta(seconds=1)
        )
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When validating the token
        # Then token is invalid
        self.assertFalse(userobj.validate_access_token(token))
        # Then token get removed
        self.assertEqual(0, Token.query.filter(Token.userid == userobj.userid).count())

    def test_verify_access_token_with_invalid(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test', expiration_time=datetime.datetime.now())
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When validating the token
        # Then token is invalid
        self.assertFalse(userobj.validate_access_token('invalid'))

    def test_set_password(self):
        UserObject.add_user('jeff', 'password')
        user.add()
        repo.add()
        user.add()
        repo.add()
        user.add()
        repo.add()
        user.add()
        user.add()
from rdiffweb.test import AppTestCase, WebCase
class ListParentsTest(AppTestCase):
        UserObject.create_admin_user(cfg.admin_user, cfg.admin_password)
    <p>{% trans %}A new access token, named "{{ test }}", has been created.{% endtrans %}</p>
class AppTestCase(unittest.TestCase):

    REPO = 'testcases'

    USERNAME = 'admin'

    PASSWORD = 'admin123'

    default_config = {}

    app_class = RdiffwebApp

    @classmethod
    def setup_class(cls):
        if cls is AppTestCase:
            raise unittest.SkipTest("%s is an abstract base class" % cls.__name__)

    @classmethod
    def teardown_class(cls):
        pass

    def setUp(self):
        # Allow defining a custom database uri for testing.
        self.database_dir = tempfile.mkdtemp(prefix='rdiffweb_tests_db_')
        uri = os.path.join(self.database_dir, 'rdiffweb.tmp.db')
        uri = os.environ.get('RDIFFWEB_TEST_DATABASE_URI', uri)
        self.default_config['database-uri'] = uri
        cfg = self.app_class.parse_args(
            args=[], config_file_contents='\n'.join('%s=%s' % (k, v) for k, v in self.default_config.items())
        )
        # Create Application
        self.app = self.app_class(cfg)
        # Create repositories
        self.testcases = create_testcases_repo(self.app)
        # Register repository
        admin_user = UserObject.get_user(self.USERNAME)
        if admin_user:
            admin_user.user_root = self.testcases
            admin_user.refresh_repos()

    def tearDown(self):
        if hasattr(self, 'database_dir'):
            shutil.rmtree(self.database_dir)
            delattr(self, 'database_dir')
        if hasattr(self, 'testcases'):
            shutil.rmtree(self.testcases)
            delattr(self, 'testcases')


        UserObject.create_admin_user(self.USERNAME, self.PASSWORD)
        admin_user = UserObject.get_user(self.USERNAME)
from sqlalchemy import create_engine, event
    def add(self, commit=True):
        if commit:
            self.__class__.session.commit()
    def delete(self, commit=True):
        """
        Delete current object to session.
        """
        if commit:
            self.__class__.session.commit()
    def merge(self, commit=True):
        """
        Merge current object to session.
        """
        self.__class__.session.merge(self)
        if commit:
            self.__class__.session.commit()
            self._session.flush()
            self._session.commit()
        except Exception:
            logger.exception('error trying to flush and commit session')
        finally:

Fixed code:
* Revoke previous user's sessions on password change [CVE-2022-3362](https://nvd.nist.gov/vuln/detail/CVE-2022-3362)
        access_token = userobj.validate_access_token(password)
        if access_token:
            access_token.accessed()
            access_token.commit()
        if u.refresh_repos():
            u.commit()
                    session.commit()
        userobj.commit()
                    user.commit()
def delete_repo(repoobj, path):
    repoobj.delete(path)
    repoobj.commit()


                cherrypy.engine.publish('schedule_task', delete_repo, repo, path)
        if self.app.currentuser.refresh_repos():
            self.app.currentuser.commit()
        user.commit()
            user.commit()
            if user.refresh_repos(delete=True):
                user.commit()
            userobj.commit()
            userobj.commit()
        userobj.commit()
                    session.commit()
            userobj.commit()
            userobj.rollback()
            userobj.commit()
            userobj.commit()
        userobj.commit()
        userobj.commit()
        data.commit()
        userobj = UserObject.add_user('test', 'test123')
        userobj.commit()
        user = UserObject.add_user('newuser')
        user.commit()
        userobj = UserObject.add_user('test1')
        userobj.commit()
                user.delete().commit()
        user.commit()
        user.commit()
        RepoObject(userid=userobj.userid, repopath='testcases/broker-repo').add().commit()
        RepoObject(userid=userobj.userid, repopath='testcases/testcases').add().commit()
        user.commit()
        user_obj.commit()
        admin.commit()
        admin.commit()
        user_obj.commit()
        user_obj.commit()
        user_obj.commit()
        user_obj.commit()
        admin.commit()
        admin.commit()
        user_obj.commit()
        admin.commit()
        admin.commit()
        userobj = UserObject.add_user('otheruser', password='password')
        userobj.commit()
        admin.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
from rdiffweb.core.model import RepoObject, SessionObject, UserObject
        # Given a user with 3 active sessions
        self.cookies = None
        self._login(self.USERNAME, self.PASSWORD)
        self.cookies = None
        self._login(self.USERNAME, self.PASSWORD)
        self.assertEqual(3, SessionObject.query.count())
        self.listener.user_password_changed.reset_mock()
        # Then all users session get deleted except our own session.
        self.assertEqual(1, SessionObject.query.count())
        self.assertEqual(self.session_id, SessionObject.query.first().id)
        RepoObject(userid=userobj.userid, repopath='invalid').add().commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        user.commit()
        user.commit()
        user.commit()
        user.commit()
        user.commit()
        userobj.commit()
        user_obj.commit()
        admin.commit()
        user_obj.commit()
        admin.commit()
        admin.commit()
        user_obj.commit()
        admin.commit()
        user_obj.commit()
        admin.commit()
        userobj.commit()
                ).commit()
            userobj.commit()
            row.commit()
            row.commit()
            if as_user.refresh_repos():
                as_user.commit()
        return super().delete()
        session.add().commit()
        SessionObject.session.commit()
            SessionObject.session.commit()
from cherrypy.process.plugins import SimplePlugin

    def accessed(self):
        self.access_time = datetime.datetime.utcnow()


class TokenCleanup(SimplePlugin):

    execution_time = '23:00'

    def start(self):
        self.bus.log('Start Token Clean Up plugin')
        self.bus.publish('schedule_job', self.execution_time, self.clean_up)

    start.priority = 55

    def stop(self):
        self.bus.log('Stop Token Clean Up plugin')
        self.bus.publish('unschedule_job', self.clean_up)

    stop.priority = 45

    def clean_up(self):
        Token.query.filter(Token.expiration_time <= datetime.datetime.now()).delete()
        Token.session.commit()


cherrypy.token_cleanup = TokenCleanup(cherrypy.engine)
cherrypy.token_cleanup.subscribe()

cherrypy.config.namespaces['token_cleanup'] = lambda key, value: setattr(cherrypy.token_cleanup, key, value)
from ._session import SessionObject
        return userobj
                SshKey(userid=self.userid, fingerprint=key.fingerprint, key=key.getvalue()).add().flush()
            Token(
                userid=self.userid, name=name, hash_token=hash_password(token), expiration_time=expiration_time
            ).add().flush()
        return Base.delete(self)
        # Store password
        logger.info("updating user password [%s] and revoke sessions", self.username)
        # Revoke other session to force re-login
        session_id = cherrypy.serving.session.id if getattr(cherrypy.serving, 'session', None) else None
        SessionObject.query.filter(
            SessionObject.username == self.username,
            SessionObject.id != session_id,
        ).delete()

                # When it matches, return the record.
                return access_token
        RepoObject(userid=userobj.userid, repopath='/testcases').add().commit()
        RepoObject(userid=userobj.userid, repopath='/testcases/home/admin/data').add().commit()
        RepoObject(userid=userobj.userid, repopath='/testcases').add().commit()
        user.commit()
        user.commit()
        other.commit()
        user.commit()
        other.commit()
        repo_obj.commit()
        repo_obj.commit()
        repo_obj.commit()
        repo_obj = RepoObject(user=userobj, repopath='repopath').add().commit()
        repo_obj = RepoObject(user=userobj, repopath='repopath').add().commit()
        repo_obj = RepoObject(user=userobj, repopath='repopath').add().commit()
# -*- coding: utf-8 -*-
# rdiffweb, A web interface to rdiff-backup repositories
# Copyright (C) 2012-2021 rdiffweb contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""
Created on June 30, 2022

Module to test `user` model.

@author: Patrik Dufresne <patrik@ikus-soft.com>
"""
import datetime
from io import open
from unittest.mock import MagicMock

import cherrypy
import pkg_resources

import rdiffweb.test
from rdiffweb.core.model import Token, UserObject


class TokenTest(rdiffweb.test.WebCase):
    def _read_ssh_key(self):
        """Readthe pub key from test packages"""
        filename = pkg_resources.resource_filename('rdiffweb.core.tests', 'test_publickey_ssh_rsa.pub')
        with open(filename, 'r', encoding='utf8') as f:
            return f.readline()

    def _read_authorized_keys(self):
        """Read the content of test_authorized_keys"""
        filename = pkg_resources.resource_filename('rdiffweb.core.tests', 'test_authorized_keys')
        with open(filename, 'r', encoding='utf8') as f:
            return f.read()

    def setUp(self):
        super().setUp()
        self.listener = MagicMock()
        cherrypy.engine.subscribe('access_token_added', self.listener.access_token_added, priority=50)
        cherrypy.engine.subscribe('queue_mail', self.listener.queue_mail, priority=50)

    def tearDown(self):
        cherrypy.engine.unsubscribe('access_token_added', self.listener.access_token_added)
        cherrypy.engine.unsubscribe('queue_mail', self.listener.queue_mail)
        return super().tearDown()

    def test_check_schedule(self):
        # Given the application is started
        # Then remove_older job should be schedule
        self.assertEqual(1, len([job for job in cherrypy.scheduler.list_jobs() if job.name == 'clean_up']))

    def test_clean_up_without_expired(self):
        # Given a user with 3 Token
        user = UserObject.get_user(self.USERNAME)
        user.add_access_token('test1')
        user.add_access_token('test2')
        user.add_access_token('test3')
        user.commit()
        self.assertEqual(3, Token.query.count())
        # When running notification_job
        cherrypy.token_cleanup.clean_up()
        # Then token are not removed
        self.assertEqual(3, Token.query.count())

    def test_clean_up_with_expired(self):
        # Given a user with 3 Token
        user = UserObject.get_user(self.USERNAME)
        user.add_access_token('test1')
        user.add_access_token('test2')
        user.add_access_token('test3')
        for t in Token.query.all():
            t.expiration_time = datetime.datetime.now()
        user.commit()
        self.assertEqual(3, Token.query.count())
        # When running notification_job
        cherrypy.token_cleanup.clean_up()
        # Then token are not removed
        self.assertEqual(0, Token.query.count())

    def test_add_access_token(self):
        # Given a user with an email
        userobj = UserObject.get_user(self.USERNAME)
        userobj.email = 'test@examples.com'
        userobj.commit()
        # When adding a new token
        token = userobj.add_access_token('test')
        userobj.commit()
        # Then a new token get created
        self.assertTrue(token)
        tokenobj = Token.query.filter(Token.userid == userobj.userid).first()
        self.assertTrue(tokenobj)
        self.assertEqual(None, tokenobj.expiration_time)
        self.assertEqual(None, tokenobj.access_time)
        # Then an email is sent to the user.
        self.listener.access_token_added.assert_called_once_with(userobj, 'test')
        self.listener.queue_mail.assert_called_once()

    def test_add_access_token_duplicate_name(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test')
        userobj.commit()
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When adding a new token with the same name
        with self.assertRaises(ValueError):
            userobj.add_access_token('test')
            userobj.commit()
        userobj.rollback()
        # Then token is not created
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # Then an email is not sent.
        self.listener.access_token_added.assert_called_once_with(userobj, 'test')

    def test_delete_access_token(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test')
        userobj.commit()
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When deleting an access token
        userobj.delete_access_token('test')
        userobj.commit()
        # Then Token get deleted
        self.assertEqual(0, Token.query.filter(Token.userid == userobj.userid).count())

    def test_delete_access_token_invalid(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test')
        userobj.commit()
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When deleting an invalid access token
        with self.assertRaises(ValueError):
            userobj.delete_access_token('invalid')
        # Then Token not deleted
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())

    def test_delete_user_remove_access_tokens(self):
        # Given a user with an existing token
        userobj = UserObject.add_user('testuser', 'password')
        userobj.commit()
        userobj.add_access_token('test')
        userobj.commit()
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When deleting the user
        userobj.delete()
        userobj.commit()
        # Then Token get deleted
        self.assertEqual(0, Token.query.filter(Token.userid == userobj.userid).count())

    def test_verify_access_token(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        token = userobj.add_access_token('test')
        userobj.commit()
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When validating the token
        # Then token is valid
        self.assertTrue(userobj.validate_access_token(token))

    def test_verify_access_token_with_expired(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        token = userobj.add_access_token(
            'test', expiration_time=datetime.datetime.now() - datetime.timedelta(seconds=1)
        )
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When validating the token
        # Then token is invalid
        self.assertFalse(userobj.validate_access_token(token))
        # Then token is not removed
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())

    def test_verify_access_token_with_invalid(self):
        # Given a user with an existing token
        userobj = UserObject.get_user(self.USERNAME)
        userobj.add_access_token('test', expiration_time=datetime.datetime.now())
        userobj.commit()
        self.assertEqual(1, Token.query.filter(Token.userid == userobj.userid).count())
        # When validating the token
        # Then token is invalid
        self.assertFalse(userobj.validate_access_token('invalid'))
from rdiffweb.core.model import DuplicateSSHKeyError, RepoObject, UserObject
        userobj.commit()
        userobj.commit()
        user = UserObject.add_user('denise')
        user.commit()
        userobj.commit()
        user = UserObject.add_user('annik')
        user.commit()
        user.commit()
        user.commit()
        user.add().commit()
        user.commit()
        user.commit()
        user.commit()
        user.add().commit()
            user.add().commit()
        user.commit()
        user.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
            userobj.commit()
        userobj.commit()
        userobj.commit()
        repos[1].commit()
        RepoObject(userid=userobj.userid, repopath='invalid').add().commit()
        userobj.commit()
        RepoObject(userid=userobj.userid, repopath='invalid').add().commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj = UserObject.add_user('jeff', 'password')
        userobj.commit()
        user.commit()
        repo.commit()
        user.add().commit()
        repo.add().commit()
        user.add().commit()
        repo.add().commit()
        user.add().commit()
        user.add().commit()
        user.add().commit()
        user.add().commit()
        user.add().commit()
        user.add().commit()
        user.add().commit()
        user.add().commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
        userobj.commit()
from rdiffweb.test import WebCase
class ListParentsTest(WebCase):
        user = UserObject.create_admin_user(cfg.admin_user, cfg.admin_password)
        user.commit()
    <p>{% trans %}A new access token, named "{{ name }}", has been created.{% endtrans %}</p>
        admin_user = UserObject.create_admin_user(self.USERNAME, self.PASSWORD)
        admin_user.commit()
            admin_user.commit()
        cherrypy.tools.db.drop_all()
from sqlalchemy import create_engine, event, inspect
from sqlalchemy.exc import SQLAlchemyError
def _get_model_changes(model):
    """
    Return a dictionary containing changes made to the model since it was
    fetched from the database.

    The dictionary is of the form {'property_name': [old_value, new_value]}
    """
    state = inspect(model)
    changes = {}
    for attr in state.attrs:
        hist = attr.history
        if not hist.has_changes():
            continue
        if isinstance(attr.value, (list, tuple)) or len(hist.deleted) > 1 or len(hist.added) > 1:
            # If array, store array
            changes[attr.key] = [hist.deleted, hist.added]
        else:
            # If primitive, store primitive
            changes[attr.key] = [
                hist.deleted[0] if len(hist.deleted) >= 1 else None,
                hist.added[0] if len(hist.added) >= 1 else None,
            ]
    return changes


    def add(self):
    def delete(self):
    def commit(self):
        self.__class__.session.commit()
        return self

    def flush(self):
        self.__class__.session.flush()
        return self

    def rollback(self):
        self.__class__.session.rollback()
        return self
        self.get_session().commit()
        self.get_session().commit()
            # When terminating, raise an error if objects are not commit.
            if self._session.dirty or self._session.new or self._session.deleted:
                changes = ', '.join([str(_get_model_changes(obj)) for obj in self._session.dirty])
                logger.exception(
                    'session is dirty, some database object(s) are not commited, this indicate a bug in the application '
                    'dirty %s new %s deleted %s' % (changes, self._session.new, self._session.deleted)
                )
                raise SQLAlchemyError('session is dirty')
        finally: