diff --git a/trytond/trytond/security.py b/trytond/trytond/security.py index 4dbe3cef83..b89f774bfa 100644 --- a/trytond/trytond/security.py +++ b/trytond/trytond/security.py @@ -125,30 +125,69 @@ def reset_password(dbname, user, context=None): def check(dbname, user, session, context=None): - for count in range(config.getint('database', 'retry'), -1, -1): - with Transaction().start(dbname, user, context=context) as transaction: - pool = _get_pool(dbname) - Session = pool.get('ir.session') - try: - find = Session.check(user, session) - break - except backend.DatabaseOperationalError: - if count: - continue - raise - finally: - transaction.commit() + remote_addr = _get_remote_addr(context) + + database_list = Pool.database_list() + if dbname in database_list: + for count in range(config.getint('database', 'retry'), -1, -1): + with Transaction().start(dbname, user, context=context) \ + as transaction: + pool = Pool(dbname) + Session = pool.get('ir.session') + try: + find = Session.check(user, session) + break + except backend.DatabaseOperationalError: + if count: + continue + raise + finally: + transaction.commit() + else: + if remote_addr: + ip_addr = str(ipaddress.ip_address(remote_addr)) + else: + ip_addr = None + now = dt.datetime.now() + timeout = dt.timedelta(config.getint('session', 'max_age')) + database = backend.Database(dbname) + conn = database.get_connection(readonly=True) + try: + ir_session = Table('ir_session') + cursor = conn.cursor() + session_query = ir_session.select( + Coalesce( + ir_session.write_date, ir_session.create_date).as_('date'), + ir_session.key, + where=((ir_session.create_uid == user) + & (ir_session.ip_address == ip_addr))) + if backend.name == 'sqlite': + sqlite_apply_types(session_query, ['DATETIME', None]) + cursor.execute(*session_query) + bad_session = False + for session_date, session_key in cursor: + if abs(session_date - now) < timeout: + if compare_digest(session_key, session): + find = session + break + else: + bad_session = True + else: + find = None if bad_session else '' + finally: + database.put_connection(conn) + if find is None: logger.error("session failed for '%s' from '%s' on database '%s'", - user, _get_remote_addr(context), dbname) + user, remote_addr, dbname) return elif not find: logger.info("session expired for '%s' from '%s' on database '%s'", - user, _get_remote_addr(context), dbname) + user, remote_addr, dbname) return else: logger.debug("session valid for '%s' from '%s' on database '%s'", - user, _get_remote_addr(context), dbname) + user, remote_addr, dbname) return user @@ -172,37 +211,6 @@ def check_timeout(dbname, user, session, context=None): return valid -def check_session(dbname, user, session, remote_addr=None): - "Check the session without using the pool" - database = backend.Database(dbname) - now = dt.datetime.now() - timeout = dt.timedelta(config.getint('session', 'max_age')) - conn = database.get_connection(readonly=True) - if remote_addr: - ip_addr = str(ipaddress.ip_address(remote_addr)) - else: - ip_addr = None - try: - ir_session = Table('ir_session') - cursor = conn.cursor() - session_query = ir_session.select( - Coalesce( - ir_session.write_date, ir_session.create_date).as_('date'), - ir_session.key, - where=((ir_session.create_uid == user) - & (ir_session.ip_address == ip_addr))) - if backend.name == 'sqlite': - sqlite_apply_types(session_query, ['DATETIME', None]) - cursor.execute(*session_query) - for session_date, session_key in cursor: - if abs(session_date - now) < timeout: - if compare_digest(session_key, session): - return True - return False - finally: - database.put_connection(conn) - - def reset(dbname, session, context): try: with Transaction().start(dbname, 0, context=context, autocommit=True): diff --git a/trytond/trytond/tests/test_security.py b/trytond/trytond/tests/test_security.py index e702893f47..27c4faea57 100644 --- a/trytond/trytond/tests/test_security.py +++ b/trytond/trytond/tests/test_security.py @@ -29,30 +29,44 @@ class SecurityTestCase(RouteTestCase): Transaction().commit() @with_transaction() - def test_check_session(self): - "Testing check_session" + def _get_auth(self): pool = Pool() User = pool.get('res.user') Session = pool.get('ir.session') - db_name = Transaction().database.name user, = User.search([('login', '=', 'user')]) with Transaction().set_user(user.id): key = Session.new() Transaction().commit() + return user.id, key - user_id = security.check_session(db_name, user.id, key) - self.assertEqual(user_id, user.id) + def test_security_check(self): + "Test security.check" + user_id, key = self._get_auth() + authenticated_user_id = security.check(self.db_name, user_id, key) + self.assertEqual(authenticated_user_id, user_id) - @with_transaction() - def test_check_session_invalid(self): - "Testing check_session with an invalid session" - pool = Pool() - User = pool.get('res.user') + def test_security_check_invalid(self): + "Test security.check with an invalid session" + user_id, _ = self._get_auth() + user_id = security.check(self.db_name, user_id, "invalid key") + self.assertIsNone(user_id) - db_name = Transaction().database.name - user, = User.search([('login', '=', 'user')]) + def test_security_check_no_pool(self): + "Test security.check without the pool" + user_id, key = self._get_auth() + Pool.stop(self.db_name) - user_id = security.check_session(db_name, user.id, "invalid key") + authenticated_user_id = security.check(self.db_name, user_id, key) + self.assertNotIn(self.db_name, Pool._pools) + self.assertEqual(authenticated_user_id, user_id) + + def test_security_check_no_pool_invalid(self): + "Test security.check without the pool on an invalid session" + user_id, _ = self._get_auth() + Pool.stop(self.db_name) + + user_id = security.check(self.db_name, user_id, "invalid key") + self.assertNotIn(self.db_name, Pool._pools) self.assertIsNone(user_id) diff --git a/trytond/trytond/tests/test_wsgi.py b/trytond/trytond/tests/test_wsgi.py index b1c6fb2fcc..55ea12aa61 100644 --- a/trytond/trytond/tests/test_wsgi.py +++ b/trytond/trytond/tests/test_wsgi.py @@ -1,13 +1,17 @@ # This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. +import base64 +from http import HTTPStatus from unittest.mock import Mock, sentinel from werkzeug.routing import Map, Rule +from trytond import security from trytond.exceptions import TrytonException +from trytond.pool import Pool from trytond.protocols.wrappers import Response -from trytond.tests.test_tryton import Client, TestCase +from trytond.tests.test_tryton import Client, RouteTestCase, TestCase from trytond.wsgi import Base64Converter, TrytondWSGI @@ -133,3 +137,87 @@ class WSGIAppTestCase(TestCase): self.assertEqual(next(response.response), b'baz') self.assertEqual(response.status, "418 I'M A TEAPOT") + + +class TrytonWSGITestCase(RouteTestCase): + module = 'res' + + @classmethod + def setUpDatabase(cls): + pool = Pool() + User = pool.get('res.user') + User.create([{ + 'name': 'user', + 'login': 'user', + 'password': '12345678', + }]) + + def test_session_valid_good_auth(self): + "Test that session_valid correctly authenticates" + app = TrytondWSGI() + + @app.route('//session_required') + @app.session_valid + def _route(request, database_name): + return Response(b'') + + user_id, key = security.login( + self.db_name, 'user', {'password': '12345678'}) + client = Client(app, Response) + session_hdr = 'Session ' + base64.b64encode( + f'user:{user_id}:{key}'.encode('utf8')).decode('utf8') + response = client.get( + f'/{self.db_name}/session_required', + headers=[('Authorization', session_hdr)]) + self.assertEqual(response.status_code, HTTPStatus.OK) + + def test_session_valid_no_pool(self): + "Test that session_valid does not use the pool" + app = TrytondWSGI() + + @app.route('//session_required') + @app.session_valid + def _route(request): + return Response(b'') + + user_id, key = security.login( + self.db_name, 'user', {'password': '12345678'}) + Pool.stop(self.db_name) + + client = Client(app, Response) + session_hdr = 'Session ' + base64.b64encode( + f'user:{user_id}:{key}'.encode('utf8')).decode('utf8') + client.get( + f'/{self.db_name}/session_required', + headers=[('Authorization', session_hdr)]) + self.assertNotIn(self.db_name, Pool._pools) + + def test_session_valid_bad_auth(self): + "Test that session_valid refuse wrong Authentication headers" + + app = TrytondWSGI() + + @app.route('//session_required') + @app.session_valid + def _route(request): + return Response(b'') + + client = Client(app, Response) + response = client.get( + f'/{self.db_name}/session_required', + headers=[('Authorization', 'Session bad token')]) + self.assertEqual(response.status_code, HTTPStatus.UNAUTHORIZED) + + def test_session_valid_no_auth(self): + "Test that session_valid refuse unauthenticated requests" + + app = TrytondWSGI() + + @app.route('//session_required') + @app.session_valid + def _route(request): + return Response(b'') + + client = Client(app, Response) + response = client.get(f'/{self.db_name}/session_required') + self.assertEqual(response.status_code, HTTPStatus.UNAUTHORIZED) diff --git a/trytond/trytond/wsgi.py b/trytond/trytond/wsgi.py index b0013d3602..e32c443375 100644 --- a/trytond/trytond/wsgi.py +++ b/trytond/trytond/wsgi.py @@ -102,8 +102,13 @@ class TrytondWSGI(object): session = request.authorization.get('session') dbname = request.view_args.get('database_name') - if not security.check_session( - dbname, userid, session, request.remote_addr): + session_check = security.check( + dbname, userid, session, { + '_request': { + 'remote_addr': request.remote_addr, + }, + }) + if session_check is None: _do_basic_auth(request) return func(request, *args, **kwargs)