1 # Odoo14 防暴力破解登录密码

  2 # 主要工具：redis

  3 # 实现思路：限制每个用户24小时内登录失败次数。连续超过5次失败后，需要等待一定时间后才能再次尝试登录

  4 # 配置：在你的配置文件中增加redis配置信息(如，我的是myodoo.cfg：./odoo-bin -c myodoo.cfg)

  5 [REDIS]

  6 ODOO_SESSION_REDIS=1

  7 ODOO_SESSION_REDIS_HOST=192.168.31.20

  8 ODOO_SESSION_REDIS_PORT=6379

  9

 10 # 代码

 11 # -*- coding: utf-8 -*-

 12 from odoo import models

 13 import contextlib

 14 from odoo.http import request

 15 import logging

 16 from odoo.exceptions import AccessDenied

 17 import datetime

 18 from odoo.tools import collections

 19 from odoo import _

 20 from configparser import ConfigParser

 21 from odoo.tools import config as odoocfg

 22 from odoo.tools.func import lazy_property

 23

 24 _logger = logging.getLogger(__name__)

 25

 26 try:

 27     import redis

 28     from redis.sentinel import Sentinel

 29 except ImportError:

 30     redis = None  # noqa

 31     _logger.debug("Cannot 'import redis'.")

 32

 33 import threading

 34

 35 class SingletonRedis(object):

 36     _instance_lock = threading.Lock()

 37

 38     def __init__(self):

 39         sentinel_host = self.get_redis_config('ODOO_SESSION_REDIS_SENTINEL_HOST')

 40         sentinel_port = int(self.get_redis_config('ODOO_SESSION_REDIS_SENTINEL_PORT', 26379))

 41         password = self.get_redis_config('ODOO_SESSION_REDIS_PASSWORD')

 42         sentinel_master_name = self.get_redis_config('ODOO_SESSION_REDIS_SENTINEL_MASTER_NAME')

 43         url = self.get_redis_config('ODOO_SESSION_REDIS_URL')

 44         host = self.get_redis_config('ODOO_SESSION_REDIS_HOST', 'localhost')

 45         port = int(self.get_redis_config('ODOO_SESSION_REDIS_PORT', 6379))

 46         if sentinel_host:

 47             sentinel = Sentinel([(sentinel_host, sentinel_port)],

 48                                 password=password)

 49             self.rd = sentinel.master_for(sentinel_master_name)

 50         elif url:

 51             self.rd = redis.from_url(url)

 52         else:

 53             self.rd = redis.Redis(host=host, port=port, password=password, db=1)

 54

 55     def get_redis_config(self, pcname, defv=None):

 56         # print("============================:%s:%s" % (pcname, str(odoocfg.rcfile)))

 57         config_cfg = odoocfg.rcfile

 58         conf = ConfigParser()

 59         conf.read(config_cfg)

 60         if conf.has_option('REDIS', pcname):

 61             return conf.get('REDIS', pcname)

 62         else:

 63             if defv:

 64                 return defv

 65             else:

 66                 return None

 67

 68     @classmethod

 69     def instance(cls, *args, **kwargs):

 70         if not hasattr(SingletonRedis, "_instance"):

 71             SingletonRedis._instance = SingletonRedis(*args, **kwargs)

 72         return SingletonRedis._instance

 73         # with SingletonRedis._instance_lock:   不用考虑多线程问题，redis会自己解决。所以单例模式到这里就可以了

 74         #     if not hasattr(SingletonRedis, "_instance"):

 75         #         SingletonRedis._instance = SingletonRedis(*args, **kwargs)

 76         # return SingletonRedis._instance

 77

 78

 79 class LoginLimit(models.Model):

 80     _inherit = 'res.users'

 81

 82     def _bys_login_success(self, uname):

 83         rd = SingletonRedis()

 84         rd.rd.delete(uname)

 85         rd.rd.delete('bys666888_'+uname)

 86

 87     def _bys_on_login_cooldown(self, uname):

 88         # 一天刷新一次，

 89         # 一天有10次错误的机会，

 90         # 前5次没有限制，

 91         # 后5次每次必须隔5分钟后尝试。

 92         # 10次机会耗尽,只能等24小时后再次尝试

 93         days = 1

 94         max_err = 10

 95         min_err = 5

 96         interval_sec = 5 * 60

 97

 98         # 查看redis中有没有数据

 99         rd = SingletonRedis()

100         rd = rd.rd

101

102         # 有的话就拿出数据，并对请求次数判断是否过期

103         failures = rd.get(uname)

104         print(uname, failures)

105         if failures:

106             failures = int(failures) + 1

107             if failures > max_err:

108                 raise AccessDenied(_("please wait %s hours before trying again.") % (str(days*24)))

109             if failures >= min_err:

110                 a = rd.ttl('bys666888_'+uname)

111                 if a > 0:

112                     raise AccessDenied(_("please wait %s seconds before trying again.") % (a))

113                     # return True

114                 else:

115                     rd.set(name='bys666888_'+uname,value=1,ex=interval_sec)

116             rd.set(name=uname,value=failures,ex=86400*days)

117         else:

118             rd.set(name=uname,value=1,ex=86400*days)

119

120

121     @contextlib.contextmanager

122     def _assert_can_auth(self):

123         """ Checks that the current environment even allows the current auth

124         request to happen.

125

126         The baseline implementation is a simple linear login cooldown: after

127         a number of failures trying to log-in, the user (by login) is put on

128         cooldown. During the cooldown period, login *attempts* are ignored

129         and logged.

130

131         .. warning::

132

133             The login counter is not shared between workers and not

134             specifically thread-safe, the feature exists mostly for

135             rate-limiting on large number of login attempts (brute-forcing

136             passwords) so that should not be much of an issue.

137

138             For a more complex strategy (e.g. database or distribute storage)

139             override this method. To simply change the cooldown criteria

140             (configuration, ...) override _on_login_cooldown instead.

141

142         .. note::

143

144             This is a *context manager* so it can be called around the login

145             procedure without having to call it itself.

146         """

147         # needs request for remote address

148         if not request:

149             yield

150             return

151

152         user_name = request.httprequest.form['login']

153         if self._bys_on_login_cooldown(user_name):

154             raise AccessDenied(_("Too many login failures,1 \r\nplease wait a bit before trying again."))

155         try:

156             yield

157         except AccessDenied:

158             # self._bys_login_failures(user_name)

159             raise

160         else:

161             self._bys_login_success(user_name)