@@ -19,6 +19,7 @@ from werkzeug.exceptions import abort
|
||||
from trytond import backend, config
|
||||
from trytond.exceptions import LoginException, RateLimitException
|
||||
from trytond.pool import Pool
|
||||
from trytond.tools import sqlite_apply_types
|
||||
from trytond.transaction import Transaction
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -184,11 +185,15 @@ def check_session(dbname, user, session, remote_addr=None):
|
||||
try:
|
||||
ir_session = Table('ir_session')
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(*ir_session.select(
|
||||
Coalesce(ir_session.write_date, ir_session.create_date),
|
||||
ir_session.key,
|
||||
where=((ir_session.create_uid == user)
|
||||
& (ir_session.ip_address == ip_addr))))
|
||||
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):
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
# This file is part of Tryton. The COPYRIGHT file at the top level of this
|
||||
# repository contains the full copyright notices and license terms.
|
||||
|
||||
from trytond import security
|
||||
from trytond.pool import Pool
|
||||
from trytond.tests.test_tryton import RouteTestCase, with_transaction
|
||||
from trytond.transaction import Transaction
|
||||
|
||||
|
||||
class SecurityTestCase(RouteTestCase):
|
||||
"Test security"
|
||||
module = 'res'
|
||||
|
||||
@classmethod
|
||||
def setUpDatabase(cls):
|
||||
pool = Pool()
|
||||
User = pool.get('res.user')
|
||||
User.create([{
|
||||
'name': 'user',
|
||||
'login': 'user',
|
||||
'password': '12345678',
|
||||
}])
|
||||
|
||||
@with_transaction()
|
||||
def tearDown(self):
|
||||
pool = Pool()
|
||||
Session = pool.get('ir.session')
|
||||
Session.delete(Session.search([]))
|
||||
Transaction().commit()
|
||||
|
||||
@with_transaction()
|
||||
def test_check_session(self):
|
||||
"Testing check_session"
|
||||
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()
|
||||
|
||||
user_id = security.check_session(db_name, user.id, key)
|
||||
self.assertEqual(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')
|
||||
|
||||
db_name = Transaction().database.name
|
||||
user, = User.search([('login', '=', 'user')])
|
||||
|
||||
user_id = security.check_session(db_name, user.id, "invalid key")
|
||||
self.assertIsNone(user_id)
|
||||
Reference in New Issue
Block a user