Browse Source

[REF] auth_brute_force: Cover all auth entrypoints (#1219)

To fix https://github.com/OCA/server-tools/issues/1125 I needed to refactor the addon. To whitelist IPs now you use a config parameter, which renders res.banned.remote model unneeded.

The fix is affected by https://github.com/odoo/odoo/issues/24183 and will not work until it gets fixed upstream due to the technical limitations implied.
pull/1255/head
Jairo Llopis 7 years ago
committed by Jairo Llopis
parent
commit
95ff8c8594
  1. 53
      auth_brute_force/README.rst
  2. 3
      auth_brute_force/__init__.py
  3. 15
      auth_brute_force/__manifest__.py
  4. 3
      auth_brute_force/controllers/__init__.py
  5. 76
      auth_brute_force/controllers/main.py
  6. 15
      auth_brute_force/data/ir_config_parameter.xml
  7. 50
      auth_brute_force/migrations/10.0.2.0.0/pre-migrate.py
  8. 2
      auth_brute_force/models/__init__.py
  9. 182
      auth_brute_force/models/res_authentication_attempt.py
  10. 45
      auth_brute_force/models/res_banned_remote.py
  11. 155
      auth_brute_force/models/res_users.py
  12. 14
      auth_brute_force/security/ir_model_access.yml
  13. BIN
      auth_brute_force/static/description/screenshot_custom_ban.png
  14. 3
      auth_brute_force/tests/__init__.py
  15. 361
      auth_brute_force/tests/test_brute_force.py
  16. 7
      auth_brute_force/views/action.xml
  17. 4
      auth_brute_force/views/menu.xml
  18. 136
      auth_brute_force/views/view.xml

53
auth_brute_force/README.rst

@ -22,9 +22,18 @@ extra information about remote IP.
Configuration
=============
Once installed, you can change the ir.config_parameter value for the key
'auth_brute_force.max_attempt_qty' (10 by default) that define the max number
of attempts allowed before the user was banned.
You can use these configuration parameters that control this addon behavior:
* ``auth_brute_force.whitelist_remotes`` is a comma-separated list of
whitelisted IPs. Failures from these remotes are ignored.
* ``auth_brute_force.max_by_ip`` defaults to 50, and indicates the maximum
successive failures allowed for an IP. After hitting the limit, the IP gets
banned.
* ``auth_brute_force.max_by_ip_user`` defaults to 10, and indicates the
maximum successive failures allowed for any IP and user combination.
After hitting the limit, that user and IP combination is banned.
Usage
=====
@ -34,17 +43,14 @@ Admin user have the possibility to unblock a banned IP.
Logging
-------
This module generates some WARNING logs, in the three following cases:
This module generates some WARNING logs, in the following cases:
* Authentication failed from remote '127.0.0.1'. Login tried : 'admin'.
Attempt 1 / 10.
* When the IP limit is reached: *Authentication failed from remote 'x.x.x.x'.
The remote has been banned. Login tried: xxxx.*
* Authentication failed from remote '127.0.0.1'. The remote has been banned.
Login tried : 'admin'.
* Authentication tried from remote '127.0.0.1'. The request has been ignored
because the remote has been banned after 10 attempts without success. Login
tried : 'admin'.
* When the IP+user combination limit is reached:
*Authentication failed from remote 'x.x.x.x'.
The remote and login combination has been banned. Login tried: xxxx.*
Screenshot
----------
@ -53,13 +59,9 @@ Screenshot
.. image:: /auth_brute_force/static/description/screenshot_attempts_list.png
**Detail of a banned IP**
.. image:: /auth_brute_force/static/description/screenshot_custom_ban.png
.. image:: https://odoo-community.org/website/image/ir.attachment/5784_f2813bd/datas
:alt: Try me on Runbot
:alt: Try me on Runbot
:target: https://runbot.odoo-community.org/runbot/149/10.0
For further information, please visit:
@ -69,14 +71,18 @@ For further information, please visit:
Known issues / Roadmap
======================
* The ID used to identify a remote request is the IP provided in the request
(key 'REMOTE_ADDR').
* Remove 🐒 patch for https://github.com/odoo/odoo/issues/24183 in v12.
* Depending of server and / or user network configuration, the idenfication
of the user can be wrong, and mainly in the following cases:
* If the Odoo server is behind an Apache / NGinx proxy without redirection,
all the request will be have the value '127.0.0.1' for the REMOTE_ADDR key;
* If some users are behind the same Internet Service Provider, if a user is
banned, all the other users will be banned too;
* If the Odoo server is behind an Apache / NGinx proxy and it is not properly
configured, all requests will use the same IP address. Blocking such IP
could render Odoo unusable for all users! **Make sure your logs output the
correct IP for werkzeug traffic before installing this addon.**
* The IP metadata retrieval should use a better system. `See details here
<https://github.com/OCA/server-tools/pull/1219/files#r187014504>`_.
Bug Tracker
===========
@ -94,6 +100,7 @@ Contributors
* Sylvain LE GAL (https://twitter.com/legalsylvain)
* David Vidal <david.vidal@tecnativa.com>
* Jairo Llopis <jairo.llopis@tecnativa.com>
Maintainer
----------

3
auth_brute_force/__init__.py

@ -1,4 +1,3 @@
# -*- encoding: utf-8 -*-
# -*- coding: utf-8 -*-
from . import models
from . import controllers

15
auth_brute_force/__manifest__.py

@ -3,22 +3,21 @@
# Copyright 2017 Tecnativa - David Vidal
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
{
'name': 'Authentification - Brute-force Attack',
'version': '10.0.1.0.0',
'name': 'Authentification - Brute-Force Filter',
'version': '10.0.2.0.0',
'category': 'Tools',
'summary': "Tracks Authentication Attempts and Prevents Brute-force"
" Attacks module",
'summary': "Track Authentication Attempts and Prevent Brute-force Attacks",
'author': "GRAP, "
"Tecnativa, "
"Odoo Community Association (OCA)",
'website': 'http://www.grap.coop',
'website': 'https://github.com/OCA/server-tools',
'license': 'AGPL-3',
'depends': [
'web',
],
# If we don't depend on it, it would inhibit this addon
"auth_crypt",
],
'data': [
'security/ir_model_access.yml',
'data/ir_config_parameter.xml',
'views/view.xml',
'views/action.xml',
'views/menu.xml',

3
auth_brute_force/controllers/__init__.py

@ -1,3 +0,0 @@
# -*- coding: utf-8 -*-
from . import main

76
auth_brute_force/controllers/main.py

@ -1,76 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright 2015 GRAP - Sylvain LE GAL
# Copyright 2017 Tecnativa - David Vidal
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import logging
from odoo import fields, http, registry, SUPERUSER_ID
from odoo.api import Environment
from odoo.http import request
from odoo.addons.web.controllers.main import Home, ensure_db
_logger = logging.getLogger(__name__)
class LoginController(Home):
@http.route()
def web_login(self, redirect=None, **kw):
if request.httprequest.method == 'POST':
ensure_db()
remote = request.httprequest.remote_addr
# Get registry and cursor
with registry(request.session.db).cursor() as cursor:
env = Environment(cursor, SUPERUSER_ID, {})
config_obj = env['ir.config_parameter']
attempt_obj = env['res.authentication.attempt']
banned_remote_obj = env['res.banned.remote']
# Get Settings
max_attempts_qty = int(config_obj.get_param(
'auth_brute_force.max_attempt_qty'))
# Test if remote user is banned
banned = banned_remote_obj.search([('remote', '=', remote)])
if banned:
request.params['password'] = ''
_logger.warning(
"Authentication tried from remote '%s'. The request "
"has been ignored because the remote has been banned "
"after %d attempts without success. Login tried : '%s'"
"." % (remote, max_attempts_qty,
request.params['login']))
else:
# Try to authenticate
result = request.session.authenticate(
request.session.db, request.params['login'],
request.params['password'])
# Log attempt
attempt_obj.create({
'attempt_date': fields.Datetime.now(),
'login': request.params['login'],
'remote': remote,
'result': banned and 'banned' or (
result and 'successfull' or 'failed'),
})
cursor.commit()
if not banned and not result:
# Get last bad attempts quantity
attempts_qty = len(attempt_obj.search_last_failed(remote))
if max_attempts_qty <= attempts_qty:
# We ban the remote
_logger.warning(
"Authentication failed from remote '%s'. "
"The remote has been banned. Login tried : '%s'"
"." % (remote, request.params['login']))
banned_remote_obj.sudo().create({
'remote': remote,
'ban_date': fields.Datetime.now(),
})
cursor.commit()
else:
_logger.warning(
"Authentication failed from remote '%s'."
" Login tried : '%s'. Attempt %d / %d." % (
remote, request.params['login'], attempts_qty,
max_attempts_qty))
return super(LoginController, self).web_login(redirect=redirect, **kw)

15
auth_brute_force/data/ir_config_parameter.xml

@ -1,15 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!-- Copyright 2015 GRAP -Sylvain LE GAL
License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). -->
<odoo noupdate="1">
<data>
<record id="max_attempt_qty" model="ir.config_parameter">
<field name="key">auth_brute_force.max_attempt_qty</field>
<field name="value">10</field>
</record>
</data>
</odoo>

50
auth_brute_force/migrations/10.0.2.0.0/pre-migrate.py

@ -0,0 +1,50 @@
# -*- coding: utf-8 -*-
# Copyright 2018 Tecnativa - Jairo Llopis
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from psycopg2 import IntegrityError
def migrate(cr, version):
# Fix typo across DB
cr.execute(
""" UPDATE res_authentication_attempt
SET result = 'successful'
WHERE result = 'successfull'""",
)
# Store whitelist IPs in new format
cr.execute(
""" SELECT remote
FROM res_banned_remote
WHERE active IS FALSE""",
)
remotes = {record[0] for record in cr.fetchall()}
try:
with cr.savepoint():
cr.execute(
"INSERT INTO ir_config_parameter (key, value) VALUES (%s, %s)",
(
"auth_brute_force.whitelist_remotes",
",".join(remotes),
),
)
except IntegrityError:
# Parameter already exists
cr.execute(
"SELECT value FROM ir_config_parameter WHERE key = %s",
("auth_brute_force.whitelist_remotes",)
)
current = set(cr.fetchall()[0][0].split(","))
cr.execute(
"UPDATE ir_config_parameter SET value = %s WHERE key = %s",
(",".join(current | remotes),
"auth_brute_force.whitelist_remotes"),
)
# Update the configured IP limit parameter
cr.execute(
"UPDATE ir_config_parameter SET key = %s WHERE key = %s",
(
"auth_brute_force.whitelist_remotes",
"auth_brute_force.max_by_ip",
)
)

2
auth_brute_force/models/__init__.py

@ -1,4 +1,4 @@
# -*- encoding: utf-8 -*-
from . import res_banned_remote
from . import res_authentication_attempt
from . import res_users

182
auth_brute_force/models/res_authentication_attempt.py

@ -2,35 +2,171 @@
# Copyright 2015 GRAP - Sylvain LE GAL
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
from odoo import _, api, fields, models
import json
import logging
from urllib2 import urlopen
from odoo import api, fields, models
GEOLOCALISATION_URL = u"http://ip-api.com/json/{}"
_logger = logging.getLogger(__name__)
class ResAuthenticationAttempt(models.Model):
_name = 'res.authentication.attempt'
_order = 'attempt_date desc'
_ATTEMPT_RESULT = [
('successfull', _('Successfull')),
('failed', _('Failed')),
('banned', _('Banned')),
]
# Column Section
attempt_date = fields.Datetime(string='Attempt Date')
login = fields.Char(string='Tried Login')
remote = fields.Char(string='Remote ID')
_order = 'create_date desc'
login = fields.Char(string='Tried Login', index=True)
remote = fields.Char(string='Remote IP', index=True)
result = fields.Selection(
selection=_ATTEMPT_RESULT, string='Authentication Result')
string='Authentication Result',
selection=[
('successful', 'Successful'),
('failed', 'Failed'),
('banned', 'Banned'),
],
index=True,
)
remote_metadata = fields.Text(
string="Remote IP metadata",
compute='_compute_metadata',
store=True,
help="Metadata publicly available for remote IP",
)
whitelisted = fields.Boolean(
compute="_compute_whitelisted",
)
@api.multi
@api.depends('remote')
def _compute_metadata(self):
for item in self:
url = GEOLOCALISATION_URL.format(item.remote)
try:
res = json.loads(urlopen(url, timeout=5).read())
except Exception:
_logger.warning(
"Couldn't fetch details from %s",
url,
exc_info=True,
)
else:
item.remote_metadata = "\n".join(
'%s: %s' % pair for pair in res.items())
@api.multi
def _compute_whitelisted(self):
whitelist = self._whitelist_remotes()
for one in self:
one.whitelisted = one.remote in whitelist
# Custom Section
@api.model
def search_last_failed(self, remote):
def _hits_limit(self, limit, remote, login=None):
"""Know if a given remote hits a given limit.
:param int limit:
The maximum amount of failures allowed.
:param str remote:
The remote IP to search for.
:param str login:
If you want to check the IP+login combination limit, supply the
login.
"""
domain = [
("remote", "=", remote),
]
if login is not None:
domain += [("login", "=", login)]
# Find last successful login
last_ok = self.search(
[('result', '=', 'successfull'), ('remote', '=', remote)],
order='attempt_date desc', limit=1)
domain + [("result", "=", "successful")],
order='create_date desc',
limit=1,
)
if last_ok:
return self.search([
('remote', '=', remote),
('attempt_date', '>', last_ok.attempt_date)])
else:
return self.search([('remote', '=', remote)])
domain += [("create_date", ">", last_ok.create_date)]
# Count failures since last success, if any
recent_failures = self.search_count(
domain + [("result", "!=", "successful")],
)
# Did we hit the limit?
return recent_failures >= limit
@api.model
def _trusted(self, remote, login):
"""Checks if any the remote or remote+login are trusted.
:param str remote:
Remote IP from which the login attempt is taking place.
:param str login:
User login that is being tried.
:return bool:
``True`` means it is trusted. ``False`` means that it is banned.
"""
# Cannot ban without remote
if not remote:
return True
get_param = self.env["ir.config_parameter"].sudo().get_param
# Whitelisted remotes always pass
if remote in self._whitelist_remotes():
return True
# Check if remote is banned
limit = int(get_param("auth_brute_force.max_by_ip", 50))
if self._hits_limit(limit, remote):
_logger.warning(
"Authentication failed from remote '%s'. "
"The remote has been banned. "
"Login tried: %r.",
remote,
login,
)
return False
# Check if remote + login combination is banned
limit = int(get_param("auth_brute_force.max_by_ip_user", 10))
if self._hits_limit(limit, remote, login):
_logger.warning(
"Authentication failed from remote '%s'. "
"The remote and login combination has been banned. "
"Login tried: %r.",
remote,
login,
)
return False
# If you get here, you are a good boy (for now)
return True
def _whitelist_remotes(self):
"""Get whitelisted remotes.
:return set:
Remote IPs that are whitelisted currently.
"""
whitelist = self.env["ir.config_parameter"].sudo().get_param(
"auth_brute_force.whitelist_remotes",
"",
)
return set(whitelist.split(","))
@api.multi
def action_whitelist_add(self):
"""Add current remotes to whitelist."""
whitelist = self._whitelist_remotes()
whitelist |= set(self.mapped("remote"))
self.env["ir.config_parameter"].set_param(
"auth_brute_force.whitelist_remotes",
",".join(whitelist),
)
@api.multi
def action_whitelist_remove(self):
"""Remove current remotes from whitelist."""
whitelist = self._whitelist_remotes()
whitelist -= set(self.mapped("remote"))
self.env["ir.config_parameter"].set_param(
"auth_brute_force.whitelist_remotes",
",".join(whitelist),
)

45
auth_brute_force/models/res_banned_remote.py

@ -1,45 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright 2015 GRAP - Sylvain LE GAL
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl.html).
import urllib
import json
from odoo import api, fields, models
class ResBannedRemote(models.Model):
_name = 'res.banned.remote'
_rec_name = 'remote'
_GEOLOCALISATION_URL = "http://ip-api.com/json/{}"
# Column Section
description = fields.Text(
string='Description', compute='_compute_description', store=True)
ban_date = fields.Datetime(
string='Ban Date', required=True, default=fields.Datetime.now)
remote = fields.Char(string='Remote ID', required=True)
active = fields.Boolean(
string='Active', help="Uncheck this box to unban the remote",
default=True)
attempt_ids = fields.Many2many(
comodel_name='res.authentication.attempt', string='Attempts',
compute='_compute_attempt_ids')
# Compute Section
@api.multi
@api.depends('remote')
def _compute_description(self):
for item in self:
url = self._GEOLOCALISATION_URL.format(item.remote)
res = json.loads(urllib.urlopen(url).read())
item.description = ''
for k, v in res.iteritems():
item.description += '%s : %s\n' % (k, v)
@api.multi
def _compute_attempt_ids(self):
for item in self:
attempt_obj = self.env['res.authentication.attempt']
item.attempt_ids = attempt_obj.search_last_failed(item.remote)

155
auth_brute_force/models/res_users.py

@ -0,0 +1,155 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Tecnativa - Jairo Llopis
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
import logging
from contextlib import contextmanager
from threading import current_thread
from odoo import api, models, SUPERUSER_ID
from odoo.exceptions import AccessDenied
from odoo.service import wsgi_server
_logger = logging.getLogger(__name__)
class ResUsers(models.Model):
_inherit = "res.users"
# HACK https://github.com/odoo/odoo/issues/24183
# TODO Remove in v12, and use normal odoo.http.request to get details
@api.model_cr
def _register_hook(self):
"""🐒-patch XML-RPC controller to know remote address."""
original_fn = wsgi_server.application_unproxied
def _patch(environ, start_response):
current_thread().environ = environ
return original_fn(environ, start_response)
wsgi_server.application_unproxied = _patch
# Helpers to track authentication attempts
@classmethod
@contextmanager
def _auth_attempt(cls, login):
"""Start an authentication attempt and track its state."""
try:
# Check if this call is nested
attempt_id = current_thread().auth_attempt_id
except AttributeError:
# Not nested; create a new attempt
attempt_id = cls._auth_attempt_new(login)
if not attempt_id:
# No attempt was created, so there's nothing to do here
yield
return
try:
current_thread().auth_attempt_id = attempt_id
result = "successful"
try:
yield
except AccessDenied as error:
result = getattr(error, "reason", "failed")
raise
finally:
cls._auth_attempt_update({"result": result})
finally:
try:
del current_thread().auth_attempt_id
except AttributeError:
pass # It was deleted already
@classmethod
def _auth_attempt_force_raise(cls, login, method):
"""Force a method to raise an AccessDenied on falsey return."""
try:
with cls._auth_attempt(login):
result = method()
if not result:
# Force exception to record auth failure
raise AccessDenied()
except AccessDenied:
pass # `_auth_attempt()` did the hard part already
return result
@classmethod
def _auth_attempt_new(cls, login):
"""Store one authentication attempt, not knowing the result."""
# Get the right remote address
try:
remote_addr = current_thread().environ["REMOTE_ADDR"]
except (KeyError, AttributeError):
remote_addr = False
# Exit if it doesn't make sense to store this attempt
if not remote_addr:
return False
# Use a separate cursor to keep changes always
with cls.pool.cursor() as cr:
env = api.Environment(cr, SUPERUSER_ID, {})
attempt = env["res.authentication.attempt"].create({
"login": login,
"remote": remote_addr,
})
return attempt.id
@classmethod
def _auth_attempt_update(cls, values):
"""Update a given auth attempt if we still ignore its result."""
auth_id = getattr(current_thread(), "auth_attempt_id", False)
if not auth_id:
return {} # No running auth attempt; nothing to do
# Use a separate cursor to keep changes always
with cls.pool.cursor() as cr:
env = api.Environment(cr, SUPERUSER_ID, {})
attempt = env["res.authentication.attempt"].browse(auth_id)
# Update only on 1st call
if not attempt.result:
attempt.write(values)
return attempt.copy_data()[0] if attempt else {}
# Override all auth-related core methods
@classmethod
def _login(cls, db, login, password):
return cls._auth_attempt_force_raise(
login,
lambda: super(ResUsers, cls)._login(db, login, password),
)
@classmethod
def authenticate(cls, db, login, password, user_agent_env):
return cls._auth_attempt_force_raise(
login,
lambda: super(ResUsers, cls).authenticate(
db, login, password, user_agent_env),
)
@classmethod
def check(cls, db, uid, passwd):
with cls._auth_attempt(uid):
return super(ResUsers, cls).check(db, uid, passwd)
@api.model
def check_credentials(self, password):
"""This is the most important and specific auth check method.
When we get here, it means that Odoo already checked the user exists
in this database.
Other auth methods usually plug here.
"""
login = self.env.user.login
with self._auth_attempt(login):
# Update login, just in case we stored the UID before
attempt = self._auth_attempt_update({"login": login})
remote = attempt.get("remote")
# Fail if the remote is banned
trusted = self.env["res.authentication.attempt"]._trusted(
remote,
login,
)
if not trusted:
error = AccessDenied()
error.reason = "banned"
raise error
# Continue with other auth systems
return super(ResUsers, self).check_credentials(password)

14
auth_brute_force/security/ir_model_access.yml

@ -4,11 +4,6 @@
name: Authentication Attempt All Users
perm_read: true
- !record {model: ir.model.access, id: access_res_banned_remote_all}:
model_id: model_res_banned_remote
name: Banned Remote All Users
perm_read: true
- !record {model: ir.model.access, id: access_res_authentication_attempt_manager}:
group_id: base.group_system
model_id: model_res_authentication_attempt
@ -17,12 +12,3 @@
perm_read: true
perm_write: true
perm_unlink: true
- !record {model: ir.model.access, id: access_res_banned_remote_manager}:
group_id: base.group_system
model_id: model_res_banned_remote
name: Banned Remote Manager
perm_create: true
perm_read: true
perm_write: true
perm_unlink: true

BIN
auth_brute_force/static/description/screenshot_custom_ban.png

Before

Width: 601  |  Height: 331  |  Size: 31 KiB

3
auth_brute_force/tests/__init__.py

@ -0,0 +1,3 @@
# -*- coding: utf-8 -*-
from . import test_brute_force

361
auth_brute_force/tests/test_brute_force.py

@ -0,0 +1,361 @@
# -*- coding: utf-8 -*-
# Copyright 2017 Tecnativa - Jairo Llopis
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from threading import current_thread
from unittest import skipUnless
from urllib import urlencode
from mock import patch
from werkzeug.utils import redirect
from odoo import http
from odoo.tests.common import at_install, can_import, HttpCase, post_install
from odoo.tools import mute_logger
from ..models import res_authentication_attempt, res_users
GARBAGE_LOGGERS = (
"werkzeug",
res_authentication_attempt.__name__,
res_users.__name__,
)
@at_install(False)
@post_install(True)
# Skip CSRF validation on tests
@patch(http.__name__ + ".WebRequest.validate_csrf", return_value=True)
# Skip specific browser forgery on redirections
@patch(http.__name__ + ".redirect_with_hash", side_effect=redirect)
# Faster tests without calls to geolocation API
@patch(res_authentication_attempt.__name__ + ".urlopen", return_value="")
class BruteForceCase(HttpCase):
def setUp(self):
super(BruteForceCase, self).setUp()
# Some tests could retain environ from last test and produce fake
# results without this patch
# HACK https://github.com/odoo/odoo/issues/24183
# TODO Remove in v12
try:
del current_thread().environ
except AttributeError:
pass
# Complex password to avoid conflicts with `password_security`
self.good_password = "Admin$%02584"
self.data_demo = {
"login": "demo",
"password": "Demo%&/(908409**",
}
with self.cursor() as cr:
env = self.env(cr)
env["ir.config_parameter"].set_param(
"auth_brute_force.max_by_ip_user", 3)
env["ir.config_parameter"].set_param(
"auth_brute_force.max_by_ip", 4)
# Clean attempts to be able to count in tests
env["res.authentication.attempt"].search([]).unlink()
# Make sure involved users have good passwords
env.user.password = self.good_password
env["res.users"].search([
("login", "=", self.data_demo["login"]),
]).password = self.data_demo["password"]
@skipUnless(can_import("odoo.addons.web"), "Needs web addon")
@mute_logger(*GARBAGE_LOGGERS)
def test_web_login_existing(self, *args):
"""Remote is banned with real user on web login form."""
data1 = {
"login": "admin",
"password": "1234", # Wrong
}
# Make sure user is logged out
self.url_open("/web/session/logout", timeout=30)
# Fail 3 times
for n in range(3):
response = self.url_open("/web/login", bytes(urlencode(data1)), 30)
# If you fail, you get /web/login again
self.assertTrue(
response.geturl().endswith("/web/login"),
"Unexpected URL %s" % response.geturl(),
)
# Admin banned, demo not
with self.cursor() as cr:
env = self.env(cr)
self.assertFalse(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
data1["login"],
),
)
self.assertTrue(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
"demo",
),
)
# Now I know the password, but login is rejected too
data1["password"] = self.good_password
response = self.url_open("/web/login", bytes(urlencode(data1)), 30)
self.assertTrue(
response.geturl().endswith("/web/login"),
"Unexpected URL %s" % response.geturl(),
)
# IP has been banned, demo user cannot login
with self.cursor() as cr:
env = self.env(cr)
self.assertFalse(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
"demo",
),
)
# Attempts recorded
with self.cursor() as cr:
env = self.env(cr)
failed = env["res.authentication.attempt"].search([
("result", "=", "failed"),
("login", "=", data1["login"]),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(failed), 3)
banned = env["res.authentication.attempt"].search([
("result", "=", "banned"),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(banned), 1)
# Unban
banned.action_whitelist_add()
# Try good login, it should work now
response = self.url_open("/web/login", bytes(urlencode(data1)), 30)
self.assertTrue(response.geturl().endswith("/web"))
@skipUnless(can_import("odoo.addons.web"), "Needs web addon")
@mute_logger(*GARBAGE_LOGGERS)
def test_web_login_unexisting(self, *args):
"""Remote is banned with fake user on web login form."""
data1 = {
"login": "administrator", # Wrong
"password": self.good_password,
}
# Make sure user is logged out
self.url_open("/web/session/logout", timeout=30)
# Fail 3 times
for n in range(3):
response = self.url_open("/web/login", bytes(urlencode(data1)), 30)
# If you fail, you get /web/login again
self.assertTrue(
response.geturl().endswith("/web/login"),
"Unexpected URL %s" % response.geturl(),
)
# Admin banned, demo not
with self.cursor() as cr:
env = self.env(cr)
self.assertFalse(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
data1["login"],
),
)
self.assertTrue(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
self.data_demo["login"],
),
)
# Demo user can login
response = self.url_open(
"/web/login",
bytes(urlencode(self.data_demo)),
30,
)
# If you pass, you get /web
self.assertTrue(
response.geturl().endswith("/web"),
"Unexpected URL %s" % response.geturl(),
)
self.url_open("/web/session/logout", timeout=30)
# Attempts recorded
with self.cursor() as cr:
env = self.env(cr)
failed = env["res.authentication.attempt"].search([
("result", "=", "failed"),
("login", "=", data1["login"]),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(failed), 3)
banned = env["res.authentication.attempt"].search([
("result", "=", "banned"),
("login", "=", data1["login"]),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(banned), 0)
@mute_logger(*GARBAGE_LOGGERS)
def test_xmlrpc_login_existing(self, *args):
"""Remote is banned with real user on XML-RPC login."""
data1 = {
"login": "admin",
"password": "1234", # Wrong
}
# Fail 3 times
for n in range(3):
self.assertFalse(self.xmlrpc_common.authenticate(
self.env.cr.dbname, data1["login"], data1["password"], {}))
# Admin banned, demo not
with self.cursor() as cr:
env = self.env(cr)
self.assertFalse(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
data1["login"],
),
)
self.assertTrue(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
"demo",
),
)
# Now I know the password, but login is rejected too
data1["password"] = self.good_password
self.assertFalse(self.xmlrpc_common.authenticate(
self.env.cr.dbname, data1["login"], data1["password"], {}))
# IP has been banned, demo user cannot login
with self.cursor() as cr:
env = self.env(cr)
self.assertFalse(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
"demo",
),
)
# Attempts recorded
with self.cursor() as cr:
env = self.env(cr)
failed = env["res.authentication.attempt"].search([
("result", "=", "failed"),
("login", "=", data1["login"]),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(failed), 3)
banned = env["res.authentication.attempt"].search([
("result", "=", "banned"),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(banned), 1)
# Unban
banned.action_whitelist_add()
# Try good login, it should work now
self.assertTrue(self.xmlrpc_common.authenticate(
self.env.cr.dbname, data1["login"], data1["password"], {}))
@mute_logger(*GARBAGE_LOGGERS)
def test_xmlrpc_login_unexisting(self, *args):
"""Remote is banned with fake user on XML-RPC login."""
data1 = {
"login": "administrator", # Wrong
"password": self.good_password,
}
# Fail 3 times
for n in range(3):
self.assertFalse(self.xmlrpc_common.authenticate(
self.env.cr.dbname, data1["login"], data1["password"], {}))
# Admin banned, demo not
with self.cursor() as cr:
env = self.env(cr)
self.assertFalse(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
data1["login"],
),
)
self.assertTrue(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
self.data_demo["login"],
),
)
# Demo user can login
self.assertTrue(self.xmlrpc_common.authenticate(
self.env.cr.dbname,
self.data_demo["login"],
self.data_demo["password"],
{},
))
# Attempts recorded
with self.cursor() as cr:
env = self.env(cr)
failed = env["res.authentication.attempt"].search([
("result", "=", "failed"),
("login", "=", data1["login"]),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(failed), 3)
banned = env["res.authentication.attempt"].search([
("result", "=", "banned"),
("login", "=", data1["login"]),
("remote", "=", "127.0.0.1"),
])
self.assertEqual(len(banned), 0)
@mute_logger(*GARBAGE_LOGGERS)
def test_orm_login_existing(self, *args):
"""No bans on ORM login with an existing user."""
data1 = {
"login": "admin",
"password": "1234", # Wrong
}
with self.cursor() as cr:
env = self.env(cr)
# Fail 3 times
for n in range(3):
self.assertFalse(
env["res.users"].authenticate(
cr.dbname, data1["login"], data1["password"], {}))
self.assertEqual(
env["res.authentication.attempt"].search(count=True, args=[]),
0,
)
self.assertTrue(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
data1["login"],
),
)
# Now I know the password, and login works
data1["password"] = self.good_password
self.assertTrue(
env["res.users"].authenticate(
cr.dbname, data1["login"], data1["password"], {}))
@mute_logger(*GARBAGE_LOGGERS)
def test_orm_login_unexisting(self, *args):
"""No bans on ORM login with an unexisting user."""
data1 = {
"login": "administrator", # Wrong
"password": self.good_password,
}
with self.cursor() as cr:
env = self.env(cr)
# Fail 3 times
for n in range(3):
self.assertFalse(
env["res.users"].authenticate(
cr.dbname, data1["login"], data1["password"], {}))
self.assertEqual(
env["res.authentication.attempt"].search(count=True, args=[]),
0,
)
self.assertTrue(
env["res.authentication.attempt"]._trusted(
"127.0.0.1",
data1["login"],
),
)
# Now I know the user, and login works
data1["login"] = "admin"
self.assertTrue(
env["res.users"].authenticate(
cr.dbname, data1["login"], data1["password"], {}))

7
auth_brute_force/views/action.xml

@ -11,11 +11,4 @@
<field name="context">{"search_default_filter_no_success":1}</field>
</record>
<record id="action_res_banned_remote" model="ir.actions.act_window">
<field name="name">Banned Remotes</field>
<field name="res_model">res.banned.remote</field>
<field name="view_type">form</field>
<field name="view_mode">tree,form</field>
</record>
</odoo>

4
auth_brute_force/views/menu.xml

@ -7,8 +7,4 @@
parent="base.menu_users"
action="action_res_authentication_attempt"/>
<menuitem id="menu_res_banned_remote"
parent="base.menu_users"
action="action_res_banned_remote"/>
</odoo>

136
auth_brute_force/views/view.xml

@ -3,78 +3,74 @@
License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). -->
<odoo>
<!-- Model: res.authentication.attempt -->
<record id="view_res_authentication_attempt_tree" model="ir.ui.view">
<field name="model">res.authentication.attempt</field>
<field name="arch" type="xml">
<tree colors="orange: result == 'failed';red: result == 'banned'">
<field name="attempt_date" />
<field name="remote" />
<field name="login" />
<field name="result" />
</tree>
</field>
</record>
<!-- Model: res.authentication.attempt -->
<record id="view_res_authentication_attempt_tree" model="ir.ui.view">
<field name="model">res.authentication.attempt</field>
<field name="arch" type="xml">
<tree
decoration-warning="result == 'failed'"
decoration-error="result == 'banned'"
>
<field name="create_date"/>
<field name="remote"/>
<field name="login"/>
<field name="result"/>
</tree>
</field>
</record>
<record id="view_res_authentication_attempt_graph" model="ir.ui.view">
<field name="model">res.authentication.attempt</field>
<field name="arch" type="xml">
<graph>
<field name="attempt_date" />
<field name="result" />
</graph>
</field>
</record>
<record id="view_res_authentication_attempt_form" model="ir.ui.view">
<field name="model">res.authentication.attempt</field>
<field name="arch" type="xml">
<form>
<header>
<button
name="action_whitelist_add"
type="object"
string="Add remote to whitelist"
attrs="{'invisible': [('whitelisted', '=', True)]}"
/>
<button
name="action_whitelist_remove"
type="object"
string="Remove remote from whitelist"
attrs="{'invisible': [('whitelisted', '=', False)]}"
/>
</header>
<sheet>
<group>
<field name="create_date"/>
<field name="result"/>
<field name="login"/>
<field name="remote"/>
<field name="remote_metadata"/>
</group>
</sheet>
</form>
</field>
</record>
<record id="view_res_authentication_attempt_search" model="ir.ui.view">
<field name="model">res.authentication.attempt</field>
<field name="arch" type="xml">
<search>
<field name="login"/>
<filter name="filter_no_success" string="Without Success" domain="[('result','!=', 'successfull')]"/>
<filter name="filter_banned" string="Banned" domain="[('result','=', 'banned')]"/>
<filter name="filter_failed" string="Failed" domain="[('result','=', 'failed')]"/>
<filter name="filter_successful" string="Successful" domain="[('result','=', 'successfull')]"/>
</search>
</field>
</record>
<record id="view_res_authentication_attempt_graph" model="ir.ui.view">
<field name="model">res.authentication.attempt</field>
<field name="arch" type="xml">
<graph>
<field name="create_date"/>
<field name="result"/>
</graph>
</field>
</record>
<!-- Model: res.banned.remote -->
<record id="view_res_banned_remote_tree" model="ir.ui.view">
<field name="model">res.banned.remote</field>
<field name="arch" type="xml">
<tree colors="gray: active==False">
<field name="remote" />
<field name="ban_date" />
<field name="active" invisible="1" />
</tree>
</field>
</record>
<record id="view_res_banned_remote_form" model="ir.ui.view">
<field name="model">res.banned.remote</field>
<field name="arch" type="xml">
<form>
<sheet>
<group>
<field name="remote" />
<field name="ban_date" />
<field name="active" />
<field name="description" />
<field name="attempt_ids" />
</group>
</sheet>
</form>
</field>
</record>
<record id="view_res_banned_remote_search" model="ir.ui.view">
<field name="model">res.banned.remote</field>
<field name="arch" type="xml">
<search>
<field name="remote"/>
</search>
</field>
</record>
<record id="view_res_authentication_attempt_search" model="ir.ui.view">
<field name="model">res.authentication.attempt</field>
<field name="arch" type="xml">
<search>
<field name="login"/>
<filter name="filter_no_success" string="Without Success" domain="[('result','!=', 'successful')]"/>
<filter name="filter_banned" string="Banned" domain="[('result','=', 'banned')]"/>
<filter name="filter_failed" string="Failed" domain="[('result','=', 'failed')]"/>
<filter name="filter_successful" string="Successful" domain="[('result','=', 'successful')]"/>
</search>
</field>
</record>
</odoo>
Loading…
Cancel
Save