495 lines
22 KiB
Python
495 lines
22 KiB
Python
|
|
from freezegun import freeze_time
|
|
from itertools import product
|
|
|
|
from odoo import Command, exceptions, fields
|
|
from odoo.addons.mail.tests.common import mail_new_test_user
|
|
from odoo.addons.whatsapp.tests.common import WhatsAppCommon, MockIncomingWhatsApp
|
|
from odoo.tests import tagged, users
|
|
from odoo.tools import mute_logger
|
|
from odoo.tools.misc import limited_field_access_token
|
|
|
|
|
|
class WhatsAppSecurityCase(WhatsAppCommon):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
cls.user_employee2 = mail_new_test_user(
|
|
cls.env,
|
|
company_id=cls.company_admin.id,
|
|
email='user.employee.2@test.mycompany.com',
|
|
groups='base.group_user',
|
|
login='company_1_test_employee_2',
|
|
)
|
|
|
|
cls.template_protected_fields = cls.env['whatsapp.template'].create({
|
|
'body': 'Signup link: {{1}}',
|
|
'model_id': cls.env['ir.model']._get_id('res.partner'),
|
|
'name': 'Test Template with Protected Fields',
|
|
'status': 'approved',
|
|
'variable_ids': [
|
|
(0, 0, {
|
|
'demo_value': 'Customer',
|
|
'field_name': 'signup_type',
|
|
'field_type': 'field',
|
|
'line_type': 'body',
|
|
'name': '{{1}}',
|
|
}),
|
|
],
|
|
'wa_account_id': cls.whatsapp_account.id,
|
|
})
|
|
|
|
|
|
@tagged('wa_account', 'security')
|
|
class WhatsAppAccountSecurity(WhatsAppSecurityCase):
|
|
|
|
@mute_logger('odoo.addons.base.models.ir_rule')
|
|
def test_account_access(self):
|
|
|
|
self.assertTrue(self.whatsapp_account.with_user(self.user_admin).name)
|
|
self.assertTrue(self.whatsapp_account.with_user(self.user_employee).name)
|
|
with self.assertRaises(exceptions.AccessError):
|
|
self.assertTrue(self.whatsapp_account.with_user(self.user_employee_c2).name)
|
|
|
|
account_admin = self.whatsapp_account.with_user(self.user_admin)
|
|
account_admin.write({
|
|
'allowed_company_ids': [(4, self.company_2.id)],
|
|
})
|
|
self.assertTrue(self.whatsapp_account.with_user(self.user_employee_c2).name)
|
|
|
|
@users('admin')
|
|
def test_account_defaults(self):
|
|
|
|
account = self.env['whatsapp.account'].create({
|
|
'account_uid': 'azerty',
|
|
'app_secret': 'azerty',
|
|
'app_uid': 'contact',
|
|
'name': 'Test Account',
|
|
'phone_uid': '987987',
|
|
'token': 'TestToken',
|
|
})
|
|
self.assertEqual(account.allowed_company_ids, self.env.user.company_id)
|
|
self.assertEqual(account.notify_user_ids, self.env.user)
|
|
|
|
|
|
@tagged('wa_account', 'security', 'post_install', '-at_install')
|
|
class WhatsAppControllerSecurity(MockIncomingWhatsApp, WhatsAppSecurityCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
cls.whatsapp_account.app_secret = '1234567890abcdef'
|
|
|
|
@mute_logger('odoo.addons.whatsapp.controller.main')
|
|
def test_signature_verification(self):
|
|
|
|
signature = '0a354a1c094d43355c4b478408ba4344564de72fc8ff9699a64ea9095ecb5415'
|
|
response = self._make_webhook_request(
|
|
self.whatsapp_account,
|
|
headers={'X-Hub-Signature-256': f'sha256={signature}'})
|
|
self.assertFalse(response.get('result'))
|
|
|
|
for signature in [
|
|
False,
|
|
'sha256=',
|
|
signature,
|
|
f'sha256=a{signature[1:]}',
|
|
]:
|
|
with self.subTest(signature=signature):
|
|
headers = {'X-Hub-Signature-256': signature} if signature else None
|
|
response = self._make_webhook_request(self.whatsapp_account, headers=headers)
|
|
self.assertIn("403 Forbidden", response.get('error', {}).get('data', {}).get('message'))
|
|
|
|
|
|
@tagged('wa_message', 'security')
|
|
class WhatsAppDiscussSecurity(WhatsAppSecurityCase):
|
|
|
|
@users('admin')
|
|
@mute_logger('odoo.addons.base.models.ir_rule')
|
|
def test_member_creation(self):
|
|
channel_channel, channel_wa = self.env['discuss.channel'].create([
|
|
{
|
|
'channel_type': 'channel',
|
|
'name': 'Test',
|
|
'whatsapp_number': '+32456001122',
|
|
}, {
|
|
'channel_type': 'whatsapp',
|
|
'name': 'Test',
|
|
'whatsapp_number': '+32456001122',
|
|
}
|
|
])
|
|
with self.assertRaises(exceptions.ValidationError):
|
|
channel_channel.with_user(self.user_employee).with_context(
|
|
default_rtc_session_ids=[(0, 0, {'is_screen_sharing_on': True})]
|
|
).whatsapp_channel_join_and_pin()
|
|
|
|
with self.assertRaises(exceptions.AccessError):
|
|
channel_wa.with_user(self.user_employee).with_context(
|
|
default_rtc_session_ids=[(0, 0, {'is_screen_sharing_on': True})]
|
|
).whatsapp_channel_join_and_pin()
|
|
|
|
employee_channel = self.env['discuss.channel'].with_user(self.user_employee).create({
|
|
'channel_type': 'whatsapp',
|
|
'name': 'employee channel',
|
|
'whatsapp_number': '+32456001122',
|
|
})
|
|
|
|
def get_join_bus():
|
|
message = self.env["mail.message"].search([], order="id desc", limit=1)
|
|
member = self.env["discuss.channel.member"].search([], order="id desc", limit=1)
|
|
admin_write_date = fields.Datetime.to_string(self.user_admin.partner_id.write_date)
|
|
member_create_date = fields.Datetime.to_string(member.create_date)
|
|
return (
|
|
[
|
|
(self.env.cr.dbname, "discuss.channel", employee_channel.id),
|
|
(self.env.cr.dbname, "res.partner", self.user_admin.partner_id.id),
|
|
(self.env.cr.dbname, "discuss.channel", employee_channel.id, "members"),
|
|
(self.env.cr.dbname, "discuss.channel", employee_channel.id),
|
|
(self.env.cr.dbname, "discuss.channel", employee_channel.id),
|
|
],
|
|
[
|
|
{
|
|
"type": "mail.record/insert",
|
|
"payload": {
|
|
"discuss.channel": [
|
|
{
|
|
"id": employee_channel.id,
|
|
"last_interest_dt": "2020-03-22 10:31:06",
|
|
},
|
|
],
|
|
},
|
|
},
|
|
{
|
|
"type": "mail.record/insert",
|
|
"payload": {
|
|
"discuss.channel.member": [
|
|
{
|
|
"id": member.id,
|
|
"message_unread_counter": 0,
|
|
"message_unread_counter_bus_id": 0,
|
|
"new_message_separator": message.id + 1,
|
|
"persona": {
|
|
"id": self.user_admin.partner_id.id,
|
|
"type": "partner",
|
|
},
|
|
"syncUnread": True,
|
|
"thread": {
|
|
"id": employee_channel.id,
|
|
"model": "discuss.channel",
|
|
},
|
|
},
|
|
],
|
|
"res.partner": self._filter_partners_fields(
|
|
{"id": self.user_admin.partner_id.id, "name": "Mitchell Admin"},
|
|
),
|
|
},
|
|
},
|
|
{
|
|
"type": "mail.record/insert",
|
|
"payload": {
|
|
"discuss.channel": [{"id": employee_channel.id, "is_pinned": True}]
|
|
},
|
|
},
|
|
{
|
|
"type": "discuss.channel/new_message",
|
|
"payload": {
|
|
"data": {
|
|
"mail.message": self._filter_messages_fields(
|
|
{
|
|
"attachment_ids": [],
|
|
"author": {
|
|
"id": self.user_admin.partner_id.id,
|
|
"type": "partner",
|
|
},
|
|
"body": '<div class="o_mail_notification">joined the channel</div>',
|
|
"create_date": fields.Datetime.to_string(
|
|
message.create_date
|
|
),
|
|
"date": "2020-03-22 10:31:06",
|
|
"default_subject": "employee channel",
|
|
"email_from": '"Mitchell Admin" <test.admin@test.example.com>',
|
|
"id": message.id,
|
|
"is_discussion": True,
|
|
"is_note": False,
|
|
"linkPreviews": [],
|
|
"message_type": "notification",
|
|
"model": "discuss.channel",
|
|
"notifications": [],
|
|
"parentMessage": False,
|
|
"pinned_at": False,
|
|
"rating_id": False,
|
|
"reactions": [],
|
|
"recipients": [],
|
|
"record_name": "employee channel",
|
|
"res_id": employee_channel.id,
|
|
"scheduledDatetime": False,
|
|
"subject": False,
|
|
"subtype_description": False,
|
|
"thread": {
|
|
"id": employee_channel.id,
|
|
"model": "discuss.channel",
|
|
},
|
|
"write_date": fields.Datetime.to_string(message.write_date),
|
|
},
|
|
),
|
|
"mail.thread": [
|
|
{
|
|
"id": employee_channel.id,
|
|
"model": "discuss.channel",
|
|
"module_icon": "/mail/static/description/icon.png",
|
|
},
|
|
],
|
|
"res.partner": self._filter_partners_fields(
|
|
{
|
|
"avatar_128_access_token": limited_field_access_token(
|
|
self.partner_admin, "avatar_128"
|
|
),
|
|
"id": self.user_admin.partner_id.id,
|
|
"isInternalUser": True,
|
|
"is_company": False,
|
|
"name": "Mitchell Admin",
|
|
"userId": self.user_admin.id,
|
|
"write_date": admin_write_date,
|
|
},
|
|
),
|
|
},
|
|
"id": employee_channel.id,
|
|
},
|
|
},
|
|
{
|
|
"type": "mail.record/insert",
|
|
"payload": {
|
|
"discuss.channel": [{"id": employee_channel.id, "memberCount": 2}],
|
|
"discuss.channel.member": [
|
|
{
|
|
"create_date": member_create_date,
|
|
"fetched_message_id": message.id,
|
|
"id": member.id,
|
|
"last_seen_dt": "2020-03-22 10:31:06",
|
|
"persona": {
|
|
"id": self.user_admin.partner_id.id,
|
|
"type": "partner",
|
|
},
|
|
"seen_message_id": message.id,
|
|
"thread": {
|
|
"id": employee_channel.id,
|
|
"model": "discuss.channel",
|
|
},
|
|
},
|
|
],
|
|
"res.partner": self._filter_partners_fields(
|
|
{
|
|
"active": True,
|
|
"avatar_128_access_token": limited_field_access_token(
|
|
self.partner_admin, "avatar_128"
|
|
),
|
|
"email": "test.admin@test.example.com",
|
|
"id": self.user_admin.partner_id.id,
|
|
"im_status": "offline",
|
|
"isInternalUser": True,
|
|
"is_company": False,
|
|
"name": "Mitchell Admin",
|
|
"userId": self.user_admin.id,
|
|
"write_date": admin_write_date,
|
|
},
|
|
),
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
self._reset_bus()
|
|
with freeze_time("2020-03-22 10:31:06"), self.assertBus(get_params=get_join_bus):
|
|
employee_channel.with_user(self.user_admin).with_context(
|
|
default_rtc_session_ids=[Command.create({"is_screen_sharing_on": True})]
|
|
).whatsapp_channel_join_and_pin()
|
|
|
|
|
|
@tagged('wa_message', 'security')
|
|
class WhatsAppMessageSecurity(WhatsAppSecurityCase):
|
|
|
|
@mute_logger('odoo.addons.auth_signup.models.res_users',
|
|
'odoo.addons.base.models.ir_cron',
|
|
'odoo.addons.base.models.ir_model')
|
|
def test_message_signup_token(self):
|
|
|
|
env = self.env(user=self.user_admin)
|
|
whatsapp_template_signup = env['whatsapp.template'].create({
|
|
'body': 'Signup link: {{1}}',
|
|
'model_id': self.env['ir.model']._get_id('res.partner'),
|
|
'name': 'Template with Signup Url',
|
|
'status': 'approved',
|
|
'variable_ids': [
|
|
(0, 0, {
|
|
'demo_value': 'Customer',
|
|
'field_type': 'field',
|
|
'field_name': 'signup_type',
|
|
'line_type': 'body',
|
|
'name': '{{1}}',
|
|
}),
|
|
],
|
|
'wa_account_id': self.whatsapp_account.id,
|
|
})
|
|
|
|
env['res.users'].sudo().reset_password(self.user_admin.login)
|
|
env = self.env(user=self.user_wa_admin)
|
|
with self.assertRaises(exceptions.AccessError):
|
|
env.ref('base.user_admin').partner_id._get_signup_url()
|
|
|
|
mail_message = self.user_admin.partner_id.message_post(body='foo')
|
|
whatsapp_message = env['whatsapp.message'].create({
|
|
'mail_message_id': mail_message.id,
|
|
'mobile_number': '+32478000000',
|
|
'wa_account_id': whatsapp_template_signup.wa_account_id.id,
|
|
'wa_template_id': whatsapp_template_signup.id,
|
|
})
|
|
env.flush_all()
|
|
|
|
self.registry.enter_test_mode(self.cr)
|
|
self.addCleanup(self.registry.leave_test_mode)
|
|
cron_cr = self.registry.cursor()
|
|
self.addCleanup(cron_cr.close)
|
|
|
|
default_progress = {'done': 0, 'remaining': 0, 'timed_out_counter': 0}
|
|
with self.mockWhatsappGateway():
|
|
self.registry['ir.cron']._process_job(
|
|
self.registry.db_name,
|
|
cron_cr,
|
|
{**self.env.ref('whatsapp.ir_cron_send_whatsapp_queue').read(load=None)[0], **default_progress}
|
|
)
|
|
|
|
whatsapp_message.invalidate_recordset()
|
|
self.assertEqual(whatsapp_message.failure_reason, "We were not able to fetch value of field 'signup_type'")
|
|
|
|
|
|
@tagged('wa_template', 'security')
|
|
class WhatsAppTemplateSecurity(WhatsAppSecurityCase):
|
|
|
|
@mute_logger('odoo.addons.base.models.ir_model')
|
|
def test_tpl_create(self):
|
|
template = self.env['whatsapp.template'].with_user(self.user_wa_admin).create({
|
|
'body': 'Hello',
|
|
'name': 'Test',
|
|
})
|
|
self.assertEqual(template.body, 'Hello')
|
|
|
|
with self.assertRaises(exceptions.AccessError):
|
|
template = self.env['whatsapp.template'].with_user(self.user_employee).create({
|
|
'body': 'Hello',
|
|
'name': 'Test 2',
|
|
})
|
|
|
|
@mute_logger('odoo.addons.base.models.ir_rule')
|
|
def test_tpl_read_allowed_users(self):
|
|
template = self.env['whatsapp.template'].with_user(self.user_wa_admin).create({
|
|
'body': 'Hello',
|
|
'name': 'Test'})
|
|
self.assertEqual(template.with_user(self.user_employee).name, 'Test')
|
|
self.assertEqual(template.with_user(self.user_employee2).name, 'Test')
|
|
|
|
template.write({'allowed_user_ids': [(4, self.user_wa_admin.id), (4, self.user_employee.id)]})
|
|
self.assertEqual(template.with_user(self.user_employee).name, 'Test')
|
|
with self.assertRaises(exceptions.AccessError):
|
|
self.assertEqual(template.with_user(self.user_employee2).name, 'Test')
|
|
|
|
@mute_logger('odoo.addons.base.models.ir_model')
|
|
def test_tpl_phone_field_update(self):
|
|
|
|
template = self.env['whatsapp.template'].create({
|
|
'body': 'Hello Phone Field Chain',
|
|
'model_id': self.env['ir.model']._get_id('res.partner'),
|
|
'name': 'WhatsApp Template',
|
|
'template_name': 'Phone Field Chain',
|
|
'status': 'approved',
|
|
'wa_account_id': self.whatsapp_account.id,
|
|
})
|
|
test_partner = self.env['res.partner'].create({
|
|
'country_id': self.env.ref('base.be').id,
|
|
'mobile': '0455001122',
|
|
'name': 'Test Partner',
|
|
'phone': '0455334455',
|
|
})
|
|
|
|
field_paths_allowed = ['mobile', 'phone', 'phone_sanitized']
|
|
field_paths_allowed_ko = ['x_studio_phone']
|
|
field_paths_disallowed = ['name']
|
|
field_paths_disallowed_ko = ['my_custom_phone_field']
|
|
for field_paths, invalid, admin_only in [
|
|
(field_paths_allowed, False, False),
|
|
(field_paths_allowed_ko, True, False),
|
|
(field_paths_disallowed, False, True),
|
|
(field_paths_disallowed_ko, True, True),
|
|
]:
|
|
for field_path, test_user in product(field_paths, (self.user_employee, self.user_wa_admin, self.user_admin)):
|
|
with self.subTest(field_path=field_path, test_user_name=test_user.name):
|
|
template.sudo().write({'phone_field': 'mobile'})
|
|
template = template.with_user(test_user)
|
|
if test_user == self.user_employee or (admin_only and test_user == self.user_wa_admin):
|
|
with self.assertRaises(exceptions.AccessError):
|
|
template.write({'phone_field': field_path})
|
|
continue
|
|
if invalid:
|
|
with self.assertRaises(exceptions.ValidationError):
|
|
template.write({'phone_field': field_path})
|
|
continue
|
|
template.write({'phone_field': field_path})
|
|
test_partner = test_partner.with_user(test_user)
|
|
composer = self._instanciate_wa_composer_from_records(template, test_partner, with_user=test_user)
|
|
with self.mockWhatsappGateway():
|
|
if field_path == 'name':
|
|
with self.assertRaises(exceptions.UserError):
|
|
composer.action_send_whatsapp_template()
|
|
else:
|
|
composer.action_send_whatsapp_template()
|
|
|
|
def test_tpl_safe_field_access(self):
|
|
template = self.env['whatsapp.template'].create({
|
|
'body': "hello, I am from '{{1}}'.",
|
|
'model_id': self.env['ir.model']._get_id('res.users'),
|
|
'name': 'Test Template',
|
|
'status': 'approved',
|
|
})
|
|
|
|
template.with_user(self.user_admin).variable_ids = [
|
|
(5, 0, 0),
|
|
(0, 0, {
|
|
'demo_value': "pwned",
|
|
'field_name': 'password',
|
|
'field_type': "field",
|
|
'line_type': "body",
|
|
'name': "{{1}}",
|
|
}),
|
|
]
|
|
|
|
with self.assertRaises(exceptions.ValidationError):
|
|
template.with_user(self.user_wa_admin).variable_ids = [
|
|
(5, 0, 0),
|
|
(0, 0, {
|
|
'demo_value': "pwned",
|
|
'field_name': 'password',
|
|
'field_type': "field",
|
|
'line_type': "body",
|
|
'name': "{{1}}",
|
|
}),
|
|
]
|
|
|
|
with self.assertRaises(exceptions.ValidationError):
|
|
template.with_user(self.user_wa_admin).model_id = self.env['ir.model']._get_id('res.partner')
|
|
|
|
with self.assertRaises(exceptions.ValidationError):
|
|
self.env['whatsapp.template'].with_user(self.user_wa_admin).create({
|
|
'body': "hello, I am from '{{1}}'.",
|
|
'model_id': self.env['ir.model']._get_id('res.partner'),
|
|
'name': 'Test Template',
|
|
'status': 'approved',
|
|
'variable_ids': [(4, template.variable_ids.id)],
|
|
})
|
|
|
|
@users('user_wa_admin')
|
|
def test_tpl_update_wa_admin(self):
|
|
template = self.template_protected_fields.with_env(self.env)
|
|
|
|
template.write({'name': 'Can Update'})
|
|
self.assertEqual(template.name, 'Can Update')
|