Files
Odoo-Modules/fusion_login_audit/models/res_users.py
gsinghpal dced0c66a4 feat(fusion_login_audit): hook successful login via _update_last_login
Overrides res.users._update_last_login to create a fusion.login.audit
row with result=success after the parent runs. The write goes through
sudo() + mail_create_nolog=True. Any exception in the audit path is
caught and logged but never propagates — a broken audit table must
never block a real user from logging in.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-27 09:03:58 -04:00

137 lines
5.5 KiB
Python

# -*- coding: utf-8 -*-
import logging
from odoo import api, fields, models
# Top-level import (vs lazy inside the method): if the dep is missing — most
# likely because the dev container got recreated and dropped its pip install
# (see CLAUDE.md Workflow) — Odoo crashes at registry load with a clear
# `ModuleNotFoundError`, not deep in the auth path after the first login.
from user_agents import parse as ua_parse
_logger = logging.getLogger(__name__)
class ResUsers(models.Model):
_inherit = 'res.users'
@api.model
def _fc_build_event_vals(
self,
result,
attempted_login,
failure_reason=None,
user_id=None,
_override_ip=None,
_override_ua=None,
_credential=None,
):
"""Build the dict of values for a fusion.login.audit row.
Pulls IP / User-Agent from the live HTTP request when available.
Falls back to ('internal', '<no-request>') for XML-RPC / cron-initiated
auth, with geo_lookup_state='internal' so the geo cron skips them.
An empty IP from an otherwise-live request (rare; misconfigured
reverse proxy) also routes to the 'internal' fallback — an empty
string isn't useful audit data and is arguably suspicious.
The _override_* kwargs exist for tests so we don't have to fake a
full request. They are NOT a public API.
Password safety: `_credential` MAY contain a 'password' key from the
Odoo auth flow. We never read that key, never log it, never put it
in vals. The test `test_build_event_vals_strips_password` locks
this property in via `assertNotIn(secret, repr(vals))`.
"""
vals = {
'attempted_login': (attempted_login or '')[:255],
'result': result,
'failure_reason': failure_reason,
'event_time': fields.Datetime.now(),
'database': self.env.cr.dbname,
'user_id': user_id,
}
ip = _override_ip
ua_str = _override_ua
if ip is None or ua_str is None:
try:
from odoo.http import request
if request and getattr(request, 'httprequest', None):
if ip is None:
ip = request.httprequest.remote_addr
if ua_str is None:
ua_str = request.httprequest.user_agent.string or ''
except Exception:
_logger.debug("fusion_login_audit: no request context", exc_info=True)
if ip and ua_str is not None:
ua_text = ua_str or ''
vals['ip_address'] = ip[:45]
vals['user_agent_raw'] = ua_text[:512]
ua = ua_parse(ua_text)
vals['browser'] = (f"{ua.browser.family} {ua.browser.version_string}".strip())[:64]
vals['os'] = (f"{ua.os.family} {ua.os.version_string}".strip())[:64]
if ua.is_bot:
vals['device_type'] = 'bot'
elif ua.is_mobile:
vals['device_type'] = 'mobile'
elif ua.is_tablet:
vals['device_type'] = 'tablet'
elif ua.is_pc:
vals['device_type'] = 'desktop'
else:
vals['device_type'] = 'unknown'
vals['geo_lookup_state'] = 'pending'
else:
vals['ip_address'] = 'internal'
vals['user_agent_raw'] = '<no-request>'
vals['device_type'] = 'unknown'
vals['geo_lookup_state'] = 'internal'
# _credential is accepted in the signature so callers (T6 _check_credentials,
# T7 _login) can hand the dict in without filtering. The helper deliberately
# touches NO keys from it — see the password-safety note in the docstring.
# `_credential` is intentionally unread here; the parameter exists so future
# work can read `credential.get('type')` for `2fa_failed` discrimination
# only via the explicit failure_reason kwarg, never from the dict directly.
del _credential # explicit no-op — locks down the read surface
return vals
def _fc_record_login_event(self, result, failure_reason=None,
user_id=None, attempted_login=None,
_credential=None):
"""Build vals + create the audit row via sudo. Never raises.
Audit writes are wrapped so that a broken audit table can never
block a real user from logging in. The exception is logged and
swallowed; auth proceeds normally.
"""
try:
vals = self._fc_build_event_vals(
result=result,
attempted_login=attempted_login
or (self.login if self else None)
or 'unknown',
failure_reason=failure_reason,
user_id=user_id or (self.id if self else None),
_credential=_credential,
)
self.env['fusion.login.audit'].sudo().with_context(
mail_create_nolog=True
).create(vals)
except Exception:
_logger.exception(
"fusion_login_audit: failed to record %s row for %s",
result, attempted_login or (self.login if self else 'unknown'),
)
def _update_last_login(self):
result = super()._update_last_login()
# Self is the singleton recordset of the user that just logged in.
self._fc_record_login_event(result='success')
return result