[IMP] core: add env to various auth methods

Allows accessing various keys, especially whether this is an
interactive login or not.

Also have the xml-rpc `login` delegate to `authenticate` instead of
having its own half-assed implementation.

And remove some dead code: as far as I can tell, Session.authenticate
is never called with a uid.
This commit is contained in:
Xavier Morel
2020-08-14 21:20:47 +00:00
committed by Olivier Dony
parent d21420d87a
commit 950d962d95
8 changed files with 42 additions and 55 deletions
+4 -4
View File
@@ -10,9 +10,9 @@ class Users(models.Model):
_inherit = "res.users"
@classmethod
def _login(cls, db, login, password):
def _login(cls, db, login, password, user_agent_env):
try:
return super(Users, cls)._login(db, login, password)
return super(Users, cls)._login(db, login, password, user_agent_env=user_agent_env)
except AccessDenied as e:
with registry(db).cursor() as cr:
cr.execute("SELECT id FROM res_users WHERE lower(login)=%s", (login,))
@@ -28,9 +28,9 @@ class Users(models.Model):
return Ldap._get_or_create_user(conf, login, entry)
raise e
def _check_credentials(self, password):
def _check_credentials(self, password, env):
try:
super(Users, self)._check_credentials(password)
return super(Users, self)._check_credentials(password, env)
except AccessDenied:
if self.env.user.active:
Ldap = self.env['res.company.ldap']
+2 -2
View File
@@ -109,9 +109,9 @@ class ResUsers(models.Model):
# return user credentials
return (self.env.cr.dbname, login, access_token)
def _check_credentials(self, password):
def _check_credentials(self, password, env):
try:
return super(ResUsers, self)._check_credentials(password)
return super(ResUsers, self)._check_credentials(password, env)
except AccessDenied:
res = self.sudo().search([('id', '=', self.env.uid), ('oauth_access_token', '=', password)])
if not res:
@@ -5,9 +5,9 @@ from odoo.http import request
class ResUsers(models.Model):
_inherit = "res.users"
def _check_credentials(self, password):
def _check_credentials(self, password, env):
"""Make all wishlists from session belong to its owner user."""
result = super(ResUsers, self)._check_credentials(password)
result = super(ResUsers, self)._check_credentials(password, env)
if request and request.session.get('wishlist_ids'):
self.env["product.wishlist"]._check_wishlist_from_session()
return result
+17 -14
View File
@@ -318,7 +318,7 @@ class Users(models.Model):
)
self.invalidate_cache(['password'], [uid])
def _check_credentials(self, password):
def _check_credentials(self, password, env):
""" Validates the current user's password.
Override this method to plug additional authentication methods.
@@ -630,7 +630,7 @@ class Users(models.Model):
return self._order
@classmethod
def _login(cls, db, login, password):
def _login(cls, db, login, password, user_agent_env):
if not password:
raise AccessDenied()
ip = request.httprequest.environ['REMOTE_ADDR'] if request else 'n/a'
@@ -642,7 +642,7 @@ class Users(models.Model):
if not user:
raise AccessDenied()
user = user.with_user(user)
user._check_credentials(password)
user._check_credentials(password, user_agent_env)
tz = request.httprequest.cookies.get('tz') if request else None
if tz in pytz.all_timezones and (not user.tz or not user.login_date):
# first login or missing tz -> set tz to browser tz
@@ -667,7 +667,7 @@ class Users(models.Model):
:param dict user_agent_env: environment dictionary describing any
relevant environment attributes
"""
uid = cls._login(db, login, password)
uid = cls._login(db, login, password, user_agent_env=user_agent_env)
if user_agent_env and user_agent_env.get('base_location'):
with cls.pool.cursor() as cr:
env = api.Environment(cr, uid, {})
@@ -693,14 +693,13 @@ class Users(models.Model):
db = cls.pool.db_name
if cls.__uid_cache[db].get(uid) == passwd:
return
cr = cls.pool.cursor()
try:
with contextlib.closing(cls.pool.cursor()) as cr:
self = api.Environment(cr, uid, {})[cls._name]
with self._assert_can_auth():
self._check_credentials(passwd)
if not self.env.user.active:
raise AccessDenied()
self._check_credentials(passwd, {'interactive': False})
cls.__uid_cache[db][uid] = passwd
finally:
cr.close()
def _get_session_token_fields(self):
return {'id', 'login', 'password', 'active'}
@@ -739,11 +738,15 @@ class Users(models.Model):
:raise: odoo.exceptions.AccessDenied when old password is wrong
:raise: odoo.exceptions.UserError when new password is not set or empty
"""
self.check(self._cr.dbname, self._uid, old_passwd)
if new_passwd:
# use self.env.user here, because it has uid=SUPERUSER_ID
return self.env.user.write({'password': new_passwd})
raise UserError(_("Setting empty passwords is not allowed for security reasons!"))
if not old_passwd:
raise AccessDenied()
if not new_passwd:
raise UserError(_("Setting empty passwords is not allowed for security reasons!"))
# alternatively: use identitycheck wizard?
self._check_credentials(old_passwd, {'interactive': True})
# use self.env.user here, because it has uid=SUPERUSER_ID
return self.env.user.write({'password': new_passwd})
def preference_save(self):
return {
+12 -13
View File
@@ -974,7 +974,7 @@ class OpenERPSession(sessions.Session):
return self.__setitem__(k, v)
object.__setattr__(self, k, v)
def authenticate(self, db, login=None, password=None, uid=None):
def authenticate(self, db, login=None, password=None):
"""
Authenticate the current user with the given db, login and
password. If successful, store the authentication parameters in the
@@ -984,25 +984,24 @@ class OpenERPSession(sessions.Session):
to authenticate the user.
"""
if uid is None:
wsgienv = request.httprequest.environ
env = dict(
base_location=request.httprequest.url_root.rstrip('/'),
HTTP_HOST=wsgienv['HTTP_HOST'],
REMOTE_ADDR=wsgienv['REMOTE_ADDR'],
)
uid = odoo.registry(db)['res.users'].authenticate(db, login, password, env)
else:
security.check(db, uid, password)
wsgienv = request.httprequest.environ
env = dict(
interactive=True,
base_location=request.httprequest.url_root.rstrip('/'),
HTTP_HOST=wsgienv['HTTP_HOST'],
REMOTE_ADDR=wsgienv['REMOTE_ADDR'],
)
uid = odoo.registry(db)['res.users'].authenticate(db, login, password, env)
self.rotate = True
self.db = db
self.uid = uid
self.login = login
self.session_token = uid and security.compute_session_token(self, request.env)
self.session_token = security.compute_session_token(self, request.env)
request.uid = uid
request.disable_db = False
if uid: self.get_context()
self.get_context()
return uid
def check_security(self):
+4 -9
View File
@@ -7,8 +7,6 @@ import odoo.tools
from odoo.exceptions import AccessDenied
from odoo.tools.translate import _
from . import security
_logger = logging.getLogger(__name__)
RPC_VERSION_1 = {
@@ -19,17 +17,14 @@ RPC_VERSION_1 = {
}
def exp_login(db, login, password):
# TODO: legacy indirection through 'security', should use directly
# the res.users model
res = security.login(db, login, password)
msg = res and 'successful login' or 'bad login or password'
_logger.info("%s from '%s' using database '%s'", msg, login, db.lower())
return res or False
return exp_authenticate(db, login, password, None)
def exp_authenticate(db, login, password, user_agent_env):
if user_agent_env is None:
user_agent_env = {}
res_users = odoo.registry(db)['res.users']
try:
return res_users.authenticate(db, login, password, user_agent_env)
return res_users.authenticate(db, login, password, {**user_agent_env, 'interactive': False})
except AccessDenied:
return False
-7
View File
@@ -4,13 +4,6 @@
import odoo
import odoo.exceptions
def login(db, login, password):
res_users = odoo.registry(db)['res.users']
try:
return res_users._login(db, login, password)
except odoo.exceptions.AccessDenied:
return False
def check(db, uid, passwd):
res_users = odoo.registry(db)['res.users']
return res_users.check(db, uid, passwd)
+1 -4
View File
@@ -1319,12 +1319,9 @@ class HttpCaseCommon(BaseCase):
return
db = get_db_name()
uid = self.registry['res.users'].authenticate(db, user, password, None)
uid = self.registry['res.users'].authenticate(db, user, password, {'interactive': False})
env = api.Environment(self.cr, uid, {})
# self.session.authenticate(db, user, password, uid=uid)
# OpenERPSession.authenticate accesses the current request, which we
# don't have, so reimplement it manually...
session = self.session
session.db = db