177 lines
8.2 KiB
Python
177 lines
8.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
|
|
|
from datetime import datetime, timedelta
|
|
import random
|
|
import werkzeug
|
|
|
|
from openerp.addons.base.ir.ir_mail_server import MailDeliveryException
|
|
from openerp.osv import osv, fields
|
|
from openerp.tools.misc import DEFAULT_SERVER_DATETIME_FORMAT, ustr
|
|
from ast import literal_eval
|
|
from openerp.tools.translate import _
|
|
from openerp.exceptions import UserError
|
|
|
|
class SignupError(Exception):
|
|
pass
|
|
|
|
def random_token():
|
|
# the token has an entropy of about 120 bits (6 bits/char * 20 chars)
|
|
chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789'
|
|
return ''.join(random.SystemRandom().choice(chars) for i in xrange(20))
|
|
|
|
def now(**kwargs):
|
|
dt = datetime.now() + timedelta(**kwargs)
|
|
return dt.strftime(DEFAULT_SERVER_DATETIME_FORMAT)
|
|
|
|
class res_users(osv.Model):
|
|
_inherit = 'res.users'
|
|
|
|
def _get_state(self, cr, uid, ids, name, arg, context=None):
|
|
res = {}
|
|
for user in self.browse(cr, uid, ids, context):
|
|
res[user.id] = ('active' if user.login_date else 'new')
|
|
return res
|
|
|
|
_columns = {
|
|
'state': fields.function(_get_state, string='Status', type='selection',
|
|
selection=[('new', 'Never Connected'), ('active', 'Connected')]),
|
|
}
|
|
|
|
def signup(self, cr, uid, values, token=None, context=None):
|
|
""" signup a user, to either:
|
|
- create a new user (no token), or
|
|
- create a user for a partner (with token, but no user for partner), or
|
|
- change the password of a user (with token, and existing user).
|
|
:param values: a dictionary with field values that are written on user
|
|
:param token: signup token (optional)
|
|
:return: (dbname, login, password) for the signed up user
|
|
"""
|
|
if token:
|
|
# signup with a token: find the corresponding partner id
|
|
res_partner = self.pool.get('res.partner')
|
|
partner = res_partner._signup_retrieve_partner(
|
|
cr, uid, token, check_validity=True, raise_exception=True, context=None)
|
|
# invalidate signup token
|
|
partner.write({'signup_token': False, 'signup_type': False, 'signup_expiration': False})
|
|
|
|
partner_user = partner.user_ids and partner.user_ids[0] or False
|
|
|
|
# avoid overwriting existing (presumably correct) values with geolocation data
|
|
if partner.country_id or partner.zip or partner.city:
|
|
values.pop('city', None)
|
|
values.pop('country_id', None)
|
|
if partner.lang:
|
|
values.pop('lang', None)
|
|
|
|
if partner_user:
|
|
# user exists, modify it according to values
|
|
values.pop('login', None)
|
|
values.pop('name', None)
|
|
partner_user.write(values)
|
|
return (cr.dbname, partner_user.login, values.get('password'))
|
|
else:
|
|
# user does not exist: sign up invited user
|
|
values.update({
|
|
'name': partner.name,
|
|
'partner_id': partner.id,
|
|
'email': values.get('email') or values.get('login'),
|
|
})
|
|
if partner.company_id:
|
|
values['company_id'] = partner.company_id.id
|
|
values['company_ids'] = [(6, 0, [partner.company_id.id])]
|
|
self._signup_create_user(cr, uid, values, context=context)
|
|
else:
|
|
# no token, sign up an external user
|
|
values['email'] = values.get('email') or values.get('login')
|
|
self._signup_create_user(cr, uid, values, context=context)
|
|
|
|
return (cr.dbname, values.get('login'), values.get('password'))
|
|
|
|
def _signup_create_user(self, cr, uid, values, context=None):
|
|
""" create a new user from the template user """
|
|
ir_config_parameter = self.pool.get('ir.config_parameter')
|
|
template_user_id = literal_eval(ir_config_parameter.get_param(cr, uid, 'auth_signup.template_user_id', 'False'))
|
|
assert template_user_id and self.exists(cr, uid, template_user_id, context=context), 'Signup: invalid template user'
|
|
|
|
# check that uninvited users may sign up
|
|
if 'partner_id' not in values:
|
|
if not literal_eval(ir_config_parameter.get_param(cr, uid, 'auth_signup.allow_uninvited', 'False')):
|
|
raise SignupError('Signup is not allowed for uninvited users')
|
|
|
|
assert values.get('login'), "Signup: no login given for new user"
|
|
assert values.get('partner_id') or values.get('name'), "Signup: no name or partner given for new user"
|
|
|
|
# create a copy of the template user (attached to a specific partner_id if given)
|
|
values['active'] = True
|
|
context = dict(context or {}, no_reset_password=True)
|
|
try:
|
|
with cr.savepoint():
|
|
return self.copy(cr, uid, template_user_id, values, context=context)
|
|
except Exception, e:
|
|
# copy may failed if asked login is not available.
|
|
raise SignupError(ustr(e))
|
|
|
|
def reset_password(self, cr, uid, login, context=None):
|
|
""" retrieve the user corresponding to login (login or email),
|
|
and reset their password
|
|
"""
|
|
user_ids = self.search(cr, uid, [('login', '=', login)], context=context)
|
|
if not user_ids:
|
|
user_ids = self.search(cr, uid, [('email', '=', login)], context=context)
|
|
if len(user_ids) != 1:
|
|
raise Exception(_('Reset password: invalid username or email'))
|
|
return self.action_reset_password(cr, uid, user_ids, context=context)
|
|
|
|
def action_reset_password(self, cr, uid, ids, context=None):
|
|
""" create signup token for each user, and send their signup url by email """
|
|
# prepare reset password signup
|
|
create_mode = bool(context.get('create_user'))
|
|
res_partner = self.pool.get('res.partner')
|
|
partner_ids = [user.partner_id.id for user in self.browse(cr, uid, ids, context)]
|
|
|
|
# no time limit for initial invitation, only for reset password
|
|
expiration = False if create_mode else now(days=+1)
|
|
|
|
res_partner.signup_prepare(cr, uid, partner_ids, signup_type="reset", expiration=expiration, context=context)
|
|
|
|
context = dict(context or {})
|
|
|
|
# send email to users with their signup url
|
|
template = False
|
|
if create_mode:
|
|
try:
|
|
# get_object() raises ValueError if record does not exist
|
|
template = self.pool.get('ir.model.data').get_object(cr, uid, 'auth_signup', 'set_password_email')
|
|
except ValueError:
|
|
pass
|
|
if not bool(template):
|
|
template = self.pool.get('ir.model.data').get_object(cr, uid, 'auth_signup', 'reset_password_email')
|
|
assert template._name == 'mail.template'
|
|
|
|
for user in self.browse(cr, uid, ids, context):
|
|
if not user.email:
|
|
raise UserError(_("Cannot send email: user %s has no email address.") % user.name)
|
|
context['lang'] = user.lang
|
|
self.pool.get('mail.template').send_mail(cr, uid, template.id, user.id, force_send=True, raise_exception=True, context=context)
|
|
|
|
def create(self, cr, uid, values, context=None):
|
|
if context is None:
|
|
context = {}
|
|
# overridden to automatically invite user to sign up
|
|
user_id = super(res_users, self).create(cr, uid, values, context=context)
|
|
user = self.browse(cr, uid, user_id, context=context)
|
|
if user.email and not context.get('no_reset_password'):
|
|
context = dict(context, create_user=True)
|
|
try:
|
|
self.action_reset_password(cr, uid, [user.id], context=context)
|
|
except MailDeliveryException:
|
|
self.pool.get('res.partner').signup_cancel(cr, uid, [user.partner_id.id], context=context)
|
|
return user_id
|
|
|
|
def copy(self, cr, uid, id, default=None, context=None):
|
|
if not default or not default.get('email'):
|
|
# avoid sending email to the user we are duplicating
|
|
context = dict(context or {}, reset_password=False)
|
|
return super(res_users, self).copy(cr, uid, id, default=default, context=context)
|