Initial commit: Security packages

odoo-bringout-oca-ocb-auth_totp/auth_totp/models/__init__.py

@@ -0,0 +1,7 @@
+# -*- coding: utf-8 -*-
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+from . import auth_totp
+from . import ir_http
+from . import res_users
+from . import totp

odoo-bringout-oca-ocb-auth_totp/auth_totp/models/auth_totp.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+from odoo import api, models
+from odoo.addons.auth_totp.controllers.home import TRUSTED_DEVICE_AGE
+
+import logging
+_logger = logging.getLogger(__name__)
+
+
+class AuthTotpDevice(models.Model):
+
+    # init is overriden in res.users.apikeys to create a secret column 'key'
+    # use a different model to benefit from the secured methods while not mixing
+    # two different concepts
+
+    _name = "auth_totp.device"
+    _inherit = "res.users.apikeys"
+    _description = "Authentication Device"
+    _auto = False
+
+    def _check_credentials_for_uid(self, *, scope, key, uid):
+        """Return True if device key matches given `scope` for user ID `uid`"""
+        assert uid, "uid is required"
+        return self._check_credentials(scope=scope, key=key) == uid
+
+    @api.autovacuum
+    def _gc_device(self):
+        self._cr.execute("""
+            DELETE FROM auth_totp_device
+            WHERE create_date < (NOW() AT TIME ZONE 'UTC' - INTERVAL '%s SECONDS')
+        """, [TRUSTED_DEVICE_AGE])
+        _logger.info("GC'd %d totp devices entries", self._cr.rowcount)

odoo-bringout-oca-ocb-auth_totp/auth_totp/models/ir_http.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+from odoo import models
+from odoo.http import request
+
+class IrHttp(models.AbstractModel):
+    _inherit = 'ir.http'
+
+    def session_info(self):
+        info = super().session_info()
+        # because frontend session_info uses this key and is embedded in
+        # the view source
+        info["user_id"] = request.session.uid,
+        return info

odoo-bringout-oca-ocb-auth_totp/auth_totp/models/res_users.py

@@ -0,0 +1,177 @@
+# -*- coding: utf-8 -*-
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+import base64
+import functools
+import logging
+import os
+import re
+
+from odoo import _, api, fields, models
+from odoo.addons.base.models.res_users import check_identity
+from odoo.exceptions import AccessDenied, UserError
+from odoo.http import request
+from odoo.tools import sql
+
+from odoo.addons.auth_totp.models.totp import TOTP, TOTP_SECRET_SIZE
+
+_logger = logging.getLogger(__name__)
+
+compress = functools.partial(re.sub, r'\s', '')
+class Users(models.Model):
+    _inherit = 'res.users'
+
+    totp_secret = fields.Char(copy=False, groups=fields.NO_ACCESS, compute='_compute_totp_secret', inverse='_inverse_totp_secret', search='_search_totp_enable')
+    totp_enabled = fields.Boolean(string="Two-factor authentication", compute='_compute_totp_enabled', search='_search_totp_enable')
+    totp_trusted_device_ids = fields.One2many('auth_totp.device', 'user_id', string="Trusted Devices")
+
+    @property
+    def SELF_READABLE_FIELDS(self):
+        return super().SELF_READABLE_FIELDS + ['totp_enabled', 'totp_trusted_device_ids']
+
+    def init(self):
+        init_res = super().init()
+        if not sql.column_exists(self.env.cr, self._table, "totp_secret"):
+            self.env.cr.execute("ALTER TABLE res_users ADD COLUMN totp_secret varchar")
+        return init_res
+
+    def _mfa_type(self):
+        r = super()._mfa_type()
+        if r is not None:
+            return r
+        if self.totp_enabled:
+            return 'totp'
+
+    def _mfa_url(self):
+        r = super()._mfa_url()
+        if r is not None:
+            return r
+        if self._mfa_type() == 'totp':
+            return '/web/login/totp'
+
+    @api.depends('totp_secret')
+    def _compute_totp_enabled(self):
+        for r, v in zip(self, self.sudo()):
+            r.totp_enabled = bool(v.totp_secret)
+
+    def _rpc_api_keys_only(self):
+        # 2FA enabled means we can't allow password-based RPC
+        self.ensure_one()
+        return self.totp_enabled or super()._rpc_api_keys_only()
+
+    def _get_session_token_fields(self):
+        return super()._get_session_token_fields() | {'totp_secret'}
+
+    def _totp_check(self, code):
+        sudo = self.sudo()
+        key = base64.b32decode(sudo.totp_secret)
+        match = TOTP(key).match(code)
+        if match is None:
+            _logger.info("2FA check: FAIL for %s %r", self, sudo.login)
+            raise AccessDenied(_("Verification failed, please double-check the 6-digit code"))
+        _logger.info("2FA check: SUCCESS for %s %r", self, sudo.login)
+
+    def _totp_try_setting(self, secret, code):
+        if self.totp_enabled or self != self.env.user:
+            _logger.info("2FA enable: REJECT for %s %r", self, self.login)
+            return False
+
+        secret = compress(secret).upper()
+        match = TOTP(base64.b32decode(secret)).match(code)
+        if match is None:
+            _logger.info("2FA enable: REJECT CODE for %s %r", self, self.login)
+            return False
+
+        self.sudo().totp_secret = secret
+        if request:
+            self.env.flush_all()
+            # update session token so the user does not get logged out (cache cleared by change)
+            new_token = self.env.user._compute_session_token(request.session.sid)
+            request.session.session_token = new_token
+
+        _logger.info("2FA enable: SUCCESS for %s %r", self, self.login)
+        return True
+
+    @check_identity
+    def action_totp_disable(self):
+        logins = ', '.join(map(repr, self.mapped('login')))
+        if not (self == self.env.user or self.env.user._is_admin() or self.env.su):
+            _logger.info("2FA disable: REJECT for %s (%s) by uid #%s", self, logins, self.env.user.id)
+            return False
+
+        self.revoke_all_devices()
+        self.sudo().write({'totp_secret': False})
+
+        if request and self == self.env.user:
+            self.env.flush_all()
+            # update session token so the user does not get logged out (cache cleared by change)
+            new_token = self.env.user._compute_session_token(request.session.sid)
+            request.session.session_token = new_token
+
+        _logger.info("2FA disable: SUCCESS for %s (%s) by uid #%s", self, logins, self.env.user.id)
+        return {
+            'type': 'ir.actions.client',
+            'tag': 'display_notification',
+            'params': {
+                'type': 'warning',
+                'message': _("Two-factor authentication disabled for the following user(s): %s", ', '.join(self.mapped('name'))),
+                'next': {'type': 'ir.actions.act_window_close'},
+            }
+        }
+
+    @check_identity
+    def action_totp_enable_wizard(self):
+        if self.env.user != self:
+            raise UserError(_("Two-factor authentication can only be enabled for yourself"))
+
+        if self.totp_enabled:
+            raise UserError(_("Two-factor authentication already enabled"))
+
+        secret_bytes_count = TOTP_SECRET_SIZE // 8
+        secret = base64.b32encode(os.urandom(secret_bytes_count)).decode()
+        # format secret in groups of 4 characters for readability
+        secret = ' '.join(map(''.join, zip(*[iter(secret)]*4)))
+        w = self.env['auth_totp.wizard'].create({
+            'user_id': self.id,
+            'secret': secret,
+        })
+        return {
+            'type': 'ir.actions.act_window',
+            'target': 'new',
+            'res_model': 'auth_totp.wizard',
+            'name': _("Two-Factor Authentication Activation"),
+            'res_id': w.id,
+            'views': [(False, 'form')],
+            'context': self.env.context,
+        }
+
+    @check_identity
+    def revoke_all_devices(self):
+        self._revoke_all_devices()
+
+    def _revoke_all_devices(self):
+        self.totp_trusted_device_ids._remove()
+
+    @api.model
+    def change_password(self, old_passwd, new_passwd):
+        self.env.user._revoke_all_devices()
+        return super().change_password(old_passwd, new_passwd)
+
+    def _compute_totp_secret(self):
+        for user in self.filtered('id'):
+            self.env.cr.execute('SELECT totp_secret FROM res_users WHERE id=%s', (user.id,))
+            user.totp_secret = self.env.cr.fetchone()[0]
+
+    def _inverse_totp_secret(self):
+        for user in self.filtered('id'):
+            secret = user.totp_secret if user.totp_secret else None
+            self.env.cr.execute('UPDATE res_users SET totp_secret = %s WHERE id=%s', (secret, user.id))
+
+    def _search_totp_enable(self, operator, value):
+        value = not value if operator == '!=' else value
+        if value:
+            self.env.cr.execute("SELECT id FROM res_users WHERE totp_secret IS NOT NULL")
+        else:
+            self.env.cr.execute("SELECT id FROM res_users WHERE totp_secret IS NULL OR totp_secret='false'")
+        result = self.env.cr.fetchall()
+        return [('id', 'in', [x[0] for x in result])]

odoo-bringout-oca-ocb-auth_totp/auth_totp/models/totp.py

@@ -0,0 +1,56 @@
+# -*- coding: utf-8 -*-
+# Part of Odoo. See LICENSE file for full copyright and licensing details.
+
+import hmac
+import struct
+import time
+
+# 160 bits, as recommended by HOTP RFC 4226, section 4, R6.
+# Google Auth uses 80 bits by default but supports 160.
+TOTP_SECRET_SIZE = 160
+
+# The algorithm (and key URI format) allows customising these parameters but
+# google authenticator doesn't support it
+# https://github.com/google/google-authenticator/wiki/Key-Uri-Format
+ALGORITHM = 'sha1'
+DIGITS = 6
+TIMESTEP = 30
+
+class TOTP:
+    def __init__(self, key):
+        self._key = key
+
+    def match(self, code, t=None, window=TIMESTEP, timestep=TIMESTEP):
+        """
+        :param code: authenticator code to check against this key
+        :param int t: current timestamp (seconds)
+        :param int window: fuzz window to account for slow fingers, network
+                           latency, desynchronised clocks, ..., every code
+                           valid between t-window an t+window is considered
+                           valid
+        """
+        if t is None:
+            t = time.time()
+
+        low = int((t - window) / timestep)
+        high = int((t + window) / timestep) + 1
+
+        return next((
+            counter for counter in range(low, high)
+            if hotp(self._key, counter) == code
+        ), None)
+
+def hotp(secret, counter):
+    # C is the 64b counter encoded in big-endian
+    C = struct.pack(">Q", counter)
+    mac = hmac.new(secret, msg=C, digestmod=ALGORITHM).digest()
+    # the data offset is the last nibble of the hash
+    offset = mac[-1] & 0xF
+    # code is the 4 bytes at the offset interpreted as a 31b big-endian uint
+    # (31b to avoid sign concerns). This effectively limits digits to 9 and
+    # hard-limits it to 10: each digit is normally worth 3.32 bits but the
+    # 10th is only worth 1.1 (9 digits encode 29.9 bits).
+    code = struct.unpack_from('>I', mac, offset)[0] & 0x7FFFFFFF
+    r = code % (10 ** DIGITS)
+    # NOTE: use text / bytes instead of int?
+    return r