Compare commits
19 Commits
43b2edcbb5
...
9df3262d30
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9df3262d30 | ||
|
|
5d9609b5ee | ||
|
|
622f133f05 | ||
|
|
482f12256e | ||
|
|
86b8e59c95 | ||
|
|
1b8038d8e8 | ||
|
|
a2d13cf83b | ||
|
|
6f6aa6e90a | ||
|
|
0513ea23a4 | ||
|
|
72aa28e6c4 | ||
|
|
a7cf44249d | ||
|
|
0e6ebe7bc6 | ||
|
|
dced0c66a4 | ||
|
|
2ced576204 | ||
|
|
61a0cb244f | ||
|
|
aeea670064 | ||
|
|
b0836e1c93 | ||
|
|
a32946be44 | ||
|
|
01a85c475c |
17
.gitignore
vendored
Normal file
17
.gitignore
vendored
Normal file
@@ -0,0 +1,17 @@
|
|||||||
|
# Python bytecode
|
||||||
|
__pycache__/
|
||||||
|
*.py[cod]
|
||||||
|
*$py.class
|
||||||
|
|
||||||
|
# Editor / OS noise
|
||||||
|
.DS_Store
|
||||||
|
*.swp
|
||||||
|
*.swo
|
||||||
|
.vscode/
|
||||||
|
.idea/
|
||||||
|
|
||||||
|
# Odoo runtime
|
||||||
|
*.pyc-tmp
|
||||||
|
|
||||||
|
# Local-only diagnostic logs from test runs
|
||||||
|
_test_*.log
|
||||||
15
CLAUDE.md
15
CLAUDE.md
@@ -12,9 +12,24 @@
|
|||||||
3. **Backend OWL**: Use standalone `rpc()` from `@web/core/network/rpc`. NOT `useService("rpc")`. `static props = []` not `{}`.
|
3. **Backend OWL**: Use standalone `rpc()` from `@web/core/network/rpc`. NOT `useService("rpc")`. `static props = []` not `{}`.
|
||||||
4. **HTTP routes**: `type="jsonrpc"` — NOT `type="json"` (deprecated).
|
4. **HTTP routes**: `type="jsonrpc"` — NOT `type="json"` (deprecated).
|
||||||
5. **res.config.settings**: Only boolean/integer/float/char/selection/many2one/datetime. NO Date fields.
|
5. **res.config.settings**: Only boolean/integer/float/char/selection/many2one/datetime. NO Date fields.
|
||||||
|
**`config_parameter=` Boolean fields don't round-trip `False` as a string.** Odoo's `set_values()` calls `IrConfigParameter.set_param(key, value)`, and `set_param` deletes the row when `value` is falsy (False / None / empty). So writing `False` to a Boolean config field means the param no longer exists in `ir_config_parameter`; a subsequent `get_param(key)` returns the *default* (Python `False`), not `'False'`. Test like `self.assertFalse(ICP.get_param('...'))` — never `assertEqual(..., 'False')`. (Integer/Float/Char go through `repr(value)` / strip, so they DO persist as strings — `'90'`, `'0'`, etc.) Source: `odoo/addons/base/models/res_config.py::set_values` and `ir_config_parameter.py::set_param`.
|
||||||
6. **res.groups**: NO `users` field, NO `category_id` field.
|
6. **res.groups**: NO `users` field, NO `category_id` field.
|
||||||
|
**res.users**: field was renamed `groups_id` → `group_ids` (also `all_group_ids` for implied). The plural form is gone; using `groups_id` raises `ValueError: Invalid field 'groups_id' in 'res.users'`.
|
||||||
|
**`ir.ui.view`**: same rename — view-level visibility gating uses `group_ids`, not `groups_id`. A record like `<field name="groups_id" eval="[(4, ref('base.group_system'))]"/>` on an `ir.ui.view` raises `ValueError: Invalid field 'groups_id' in 'ir.ui.view'` at module install. (The XML *attribute* `groups="base.group_system"` on form elements like `<page>`, `<button>`, `<field>` is unrelated and still works.)
|
||||||
|
**`ir.rule` `groups` field is additive, not restrictive.** A rule with `groups=[some_group]` applies ONLY to users in that group — it does NOT restrict non-members. So `domain_force=[(1,'=',1)]` + `groups=[base.group_system]` does NOT mean "only admins see rows"; it means "admins see all rows (and the rule is silent on everyone else)". Non-admins are gated by the ACL (`ir.model.access.csv`), not the rule. To truly restrict by group at the rule layer, pair a global rule (`groups=[]`, `domain_force=[(0,'=',1)]` = block-all baseline) with a group-scoped allow rule. Default to letting the ACL do the gating; use rules for row-level filters that ACLs cannot express.
|
||||||
7. **Search views**: NO `group expand="0"` syntax.
|
7. **Search views**: NO `group expand="0"` syntax.
|
||||||
8. **SCSS imports**: `@import "./partial"` is FORBIDDEN in Odoo 19 custom SCSS. It prints a warning and silently falls back to the old cached bundle. Register every SCSS file (including `_partial.scss` tokens) as a separate entry in `web.assets_backend`. Put tokens first; Odoo concatenates bundle files so SCSS variables/mixins from the first file are visible to every later file.
|
8. **SCSS imports**: `@import "./partial"` is FORBIDDEN in Odoo 19 custom SCSS. It prints a warning and silently falls back to the old cached bundle. Register every SCSS file (including `_partial.scss` tokens) as a separate entry in `web.assets_backend`. Put tokens first; Odoo concatenates bundle files so SCSS variables/mixins from the first file are visible to every later file.
|
||||||
|
9. **SQL constraints & indexes**: Odoo 19 dropped `_sql_constraints = [(name, def, msg), ...]` and the `init()`/raw-SQL pattern. Both still parse but only emit a warning and are silently ignored. Use declarative class attributes instead:
|
||||||
|
```python
|
||||||
|
_check_qty_positive = models.Constraint('CHECK (qty > 0)', 'Quantity must be positive.')
|
||||||
|
_user_time_idx = models.Index('(user_id, event_time DESC)')
|
||||||
|
```
|
||||||
|
The attribute name after the leading underscore becomes the SQL object name suffix (`{table}_{suffix}`). `models.Index` accepts `DESC`, `WHERE` predicates, and `USING btree (...)`. Sources: `odoo/orm/model_classes.py` (warns at registry build), `odoo/orm/table_objects.py` (Constraint + Index classes).
|
||||||
|
10. **`res.users._login` is an instance method in Odoo 19**, not a classmethod as in earlier versions. Signature is `def _login(self, credential, user_agent_env)` — there is no `db` parameter. Override it like any normal instance method (`super()._login(credential, user_agent_env)`). When called via `authenticate()` on an empty recordset, `self` carries the right env. Older recipes that build a separate `api.Environment` from `odoo.modules.registry.Registry(db)` no longer apply. Source: `odoo/addons/base/models/res_users.py:760`.
|
||||||
|
11. **Inherited `ir.ui.view` records cannot have `groups`/`group_ids` on the record itself.** Odoo 19 raises `ParseError: Inherited view cannot have 'groups' defined on the record. Use 'groups' attributes inside the view definition` at install time. Move the gate to the inner XML nodes — every `<button>`, `<page>`, `<field>`, `<xpath>`, `<group>` etc. supports a `groups="base.group_system"` attribute. For an inherited form with a smart button + admin tab, put `groups=` on the button and the page individually; leave the `<record model="ir.ui.view">` clean.
|
||||||
|
12. **`mail.template` QWeb/inline_template `ctx` IS `self.env.context`** — not a nested dict you can pass. `MailRenderMixin._render_eval_context()` sets `ctx = self.env.context`, so `ctx.get('foo')` in subject/body resolves to `env.context.get('foo')`. To pass dynamic data to a template, spread keys directly into the context: `tmpl.with_context(**my_data).send_mail(res_id, ...)`. Calling `tmpl.with_context(ctx=my_data)` puts the dict at `env.context['ctx']`, and the template's `ctx.get('foo')` becomes `env.context.get('foo')` → `None` (looks like a silent rendering bug — subject ends up blank).
|
||||||
|
13. **`ir.cron` dropped `numbercall`** in Odoo 19. Old recipes set `<field name="numbercall">-1</field>` for "run forever"; that now raises `ValueError: Invalid field 'numbercall' in 'ir.cron'` at install time. Just omit the field — recurring crons keep running as long as `active=True`. Source: `odoo/addons/base/models/ir_cron.py` field list.
|
||||||
|
14. **`cr.commit()` / `cr.rollback()` raise AssertionError inside `TransactionCase`** — they are NOT silent no-ops in Odoo 19. The test cursor explicitly refuses both ("Cannot commit or rollback a cursor from inside a test, this will lead to a broken cursor when trying to rollback the test. Please rollback to a specific savepoint instead..."). For cron/worker code that needs per-row isolation so one bad row doesn't roll back the whole batch, use `with self.env.cr.savepoint(): ...` inside the loop instead of `cr.commit()`. Savepoints work in both prod (under the outer cron transaction) and tests (under the outer test transaction). The cron transaction commits the whole batch when the method returns; in tests everything rolls back cleanly. Source: `odoo/sql_db.py::TestCursor.commit` and `Cursor.savepoint()`.
|
||||||
|
|
||||||
15. **There is NO `sale.subscription` model in Odoo 19** (Enterprise `sale_subscription`). A subscription is a **`sale.order`** with `is_subscription=True`, `plan_id` → **`sale.subscription.plan`** (the recurrence), plus `subscription_state` / `next_invoice_date` / `recurring_monthly`. Any Many2one or relation that targets "a subscription" must point at `sale.order` (filter `domain=[('is_subscription','=',True)]`) — **not** `sale.subscription`, which does not exist and fails at install. The surviving `sale.subscription.*` records are only the plan + wizards/reports (`sale.subscription.plan`, `sale.subscription.report`, `sale.subscription.change.customer.wizard`, `sale.subscription.close.reason.wizard`). Verified on live `nexamain` (odoo-nexa, 19.0): `SELECT model FROM ir_model WHERE model LIKE 'sale.subscription%'`.
|
15. **There is NO `sale.subscription` model in Odoo 19** (Enterprise `sale_subscription`). A subscription is a **`sale.order`** with `is_subscription=True`, `plan_id` → **`sale.subscription.plan`** (the recurrence), plus `subscription_state` / `next_invoice_date` / `recurring_monthly`. Any Many2one or relation that targets "a subscription" must point at `sale.order` (filter `domain=[('is_subscription','=',True)]`) — **not** `sale.subscription`, which does not exist and fails at install. The surviving `sale.subscription.*` records are only the plan + wizards/reports (`sale.subscription.plan`, `sale.subscription.report`, `sale.subscription.change.customer.wizard`, `sale.subscription.close.reason.wizard`). Verified on live `nexamain` (odoo-nexa, 19.0): `SELECT model FROM ir_model WHERE model LIKE 'sale.subscription%'`.
|
||||||
|
|
||||||
|
|||||||
2703
docs/superpowers/plans/2026-05-26-fusion-login-audit.md
Normal file
2703
docs/superpowers/plans/2026-05-26-fusion-login-audit.md
Normal file
File diff suppressed because it is too large
Load Diff
444
docs/superpowers/specs/2026-05-26-fusion-login-audit-design.md
Normal file
444
docs/superpowers/specs/2026-05-26-fusion-login-audit-design.md
Normal file
@@ -0,0 +1,444 @@
|
|||||||
|
# Fusion Login Audit — Design Spec
|
||||||
|
|
||||||
|
**Status:** Approved, ready for implementation planning
|
||||||
|
**Date:** 2026-05-26
|
||||||
|
**Author:** Brainstormed with the user (Gurpreet) for the Westin Healthcare Odoo 19 deployment
|
||||||
|
**Target module path:** `K:\Github\Odoo-Modules\fusion_login_audit\`
|
||||||
|
**Production deploy target:** `/opt/odoo/custom-addons/fusion_login_audit/` on `odoo-westin` (VM 101, worker1, 192.168.1.40)
|
||||||
|
**Production DB:** `westin-v19` (Odoo 19, PostgreSQL)
|
||||||
|
|
||||||
|
## Background and motivation
|
||||||
|
|
||||||
|
A spot audit of user `info@gsafinancialconsulting.com` ("GSA Accounting", uid 63) revealed Odoo's built-in login tracking is effectively unusable for compliance:
|
||||||
|
|
||||||
|
- `res.users.log` rows are pruned by the daily `_gc_user_logs` cron — only the most recent login per user survives. For GSA Accounting the entire history collapsed to a single row at `2026-04-22 20:24 EDT`.
|
||||||
|
- `/var/log/odoo` on the production VM is empty because Odoo is configured at `log_level=warn` with stdout-only logging; INFO-level auth lines aren't captured anywhere.
|
||||||
|
- The container's json log is 444 KB and rotates frequently — nothing about the user remains.
|
||||||
|
- The existing `network_logger` module records outbound HTTP traffic from Odoo (uid=1 always), not user activity.
|
||||||
|
|
||||||
|
Result: today there is **no durable record** of who logged in, when, from where, or how often. A user with `base.group_system` + Technical Features and no 2FA — like GSA Accounting — could be active for months without any reconstructable trail.
|
||||||
|
|
||||||
|
This module closes that gap with a dedicated audit table that survives Odoo's GC, captures successful and failed authentications, surfaces results in the user form, and alerts admins on suspicious failure bursts.
|
||||||
|
|
||||||
|
## Goals
|
||||||
|
|
||||||
|
1. **Durable audit trail** of every password-authenticated login (success and failure) on `westin-v19`.
|
||||||
|
2. **Per-user visibility** for Settings admins via a tab + smart button on `res.users`.
|
||||||
|
3. **Failure-burst alerting** to admins on a configurable consecutive-failure threshold.
|
||||||
|
4. **Geo-enrichment** of IPs out-of-band so authentication latency is unaffected.
|
||||||
|
5. **Zero risk to the auth path** — an audit-write failure must never block a real login.
|
||||||
|
|
||||||
|
## Non-goals (v1)
|
||||||
|
|
||||||
|
- Logging every HTTP request / page view (explicitly de-scoped during brainstorming).
|
||||||
|
- Logging session resume events from auth cookies.
|
||||||
|
- API-key authentication (`credential['type'] == 'apikey'`) — bypasses `_check_credentials`. Documented as a known gap; addressable in a follow-up.
|
||||||
|
- OAuth / SSO logins — no OAuth provider configured on westin-v19.
|
||||||
|
- Self-service "view my own login activity" for end users — visibility is admin-only.
|
||||||
|
- Auto-disabling users on failed logins — flagged as a self-service DoS vector during brainstorming.
|
||||||
|
|
||||||
|
## Architecture overview
|
||||||
|
|
||||||
|
```
|
||||||
|
┌─────────────────────────────────────────────────────────────────────┐
|
||||||
|
│ Odoo authentication path │
|
||||||
|
│ │
|
||||||
|
│ /web/login → res.users._login() → res.users._check_credentials() │
|
||||||
|
│ ↓ │
|
||||||
|
│ (on success) │
|
||||||
|
│ ↓ │
|
||||||
|
│ res.users._update_last_login() │
|
||||||
|
│ ↓ │
|
||||||
|
│ ┌────────────────────┴────────────────────┐ │
|
||||||
|
│ │ │ │
|
||||||
|
│ ▼ ▼ │
|
||||||
|
│ fusion.login.audit (sudo create) Odoo's existing res_users_log │
|
||||||
|
│ result='success' + IP + UA │
|
||||||
|
│ │
|
||||||
|
│ (on AccessDenied) │
|
||||||
|
│ ↓ │
|
||||||
|
│ fusion.login.audit (sudo create) │
|
||||||
|
│ result='failure' + failure_reason + attempted_login │
|
||||||
|
│ ↓ │
|
||||||
|
│ _fc_recent_failure_count() >= threshold? │
|
||||||
|
│ ↓ yes │
|
||||||
|
│ _fc_send_failure_alert() → mail.mail to base.group_system │
|
||||||
|
└──────────────────────────────────┬──────────────────────────────────┘
|
||||||
|
│
|
||||||
|
┌─────────────────────┼─────────────────────┐
|
||||||
|
▼ ▼ ▼
|
||||||
|
cron: cron_geo_enrich cron: cron_retention_gc UI surfaces:
|
||||||
|
every 5 min daily 03:00 UTC - smart button on res.users
|
||||||
|
- reverse DNS - delete rows older than - "Login Activity" tab
|
||||||
|
- ip-api.com lookup x_fc_login_audit_ - Settings → Technical →
|
||||||
|
- 30-day local cache retention_days Login Audit menus
|
||||||
|
- Settings page section
|
||||||
|
```
|
||||||
|
|
||||||
|
The auth-path hooks are synchronous (must run inside the request). Geolocation, alerting, and retention are out-of-band so they cannot affect login latency.
|
||||||
|
|
||||||
|
## Module skeleton
|
||||||
|
|
||||||
|
```
|
||||||
|
fusion_login_audit/
|
||||||
|
├── __manifest__.py
|
||||||
|
├── __init__.py
|
||||||
|
├── models/
|
||||||
|
│ ├── __init__.py
|
||||||
|
│ ├── res_users.py # extends res.users with capture hooks + computed fields + smart-button action
|
||||||
|
│ ├── fusion_login_audit.py # the new audit record model
|
||||||
|
│ └── res_config_settings.py # alert threshold + window + retention settings
|
||||||
|
├── data/
|
||||||
|
│ ├── ir_cron_data.xml # cron_geo_enrich + cron_retention_gc
|
||||||
|
│ └── mail_template_data.xml # failed-login alert template
|
||||||
|
├── security/
|
||||||
|
│ ├── security.xml # record rule: read for base.group_system only
|
||||||
|
│ └── ir.model.access.csv
|
||||||
|
├── views/
|
||||||
|
│ ├── fusion_login_audit_views.xml # list / form / kanban / search
|
||||||
|
│ ├── res_users_views.xml # tab + smart button
|
||||||
|
│ ├── res_config_settings_views.xml # Settings section
|
||||||
|
│ └── menus.xml # Settings → Technical → Login Audit
|
||||||
|
├── tests/
|
||||||
|
│ ├── __init__.py
|
||||||
|
│ ├── test_login_audit.py
|
||||||
|
│ └── test_security.py
|
||||||
|
└── static/
|
||||||
|
└── description/
|
||||||
|
└── icon.png # copied from C:\Users\gsing\Downloads\fusion logs.png
|
||||||
|
```
|
||||||
|
|
||||||
|
**Manifest highlights**
|
||||||
|
|
||||||
|
- `version='19.0.1.0.0'` (project naming convention)
|
||||||
|
- `license='OPL-1'` (matches `fusion_accounts`)
|
||||||
|
- `depends=['base', 'mail']`
|
||||||
|
- `category='Tools'`
|
||||||
|
- `application=False` (it's a technical addon, not a top-level app)
|
||||||
|
|
||||||
|
**Dependencies (Python):** none new. Uses the `user_agents` library already shipped with Odoo. Geolocation calls `http://ip-api.com/json/<ip>` via the standard `requests` library (no API key required, 45 req/min free tier).
|
||||||
|
|
||||||
|
**Field naming:** new fields on existing models (`res.users`, `res.config.settings`) use the `x_fc_*` prefix per project CLAUDE.md. The new `fusion.login.audit` model uses unprefixed field names.
|
||||||
|
|
||||||
|
## Data model
|
||||||
|
|
||||||
|
### `fusion.login.audit` (new model, table `fusion_login_audit`)
|
||||||
|
|
||||||
|
| Field | Type | Notes |
|
||||||
|
|---|---|---|
|
||||||
|
| `user_id` | Many2one(`res.users`, `ondelete='set null'`) | Null if attempted login didn't match any user |
|
||||||
|
| `attempted_login` | Char(255), indexed | Always set — even on unknown-user failures |
|
||||||
|
| `result` | Selection(`success`, `failure`) | Indexed |
|
||||||
|
| `failure_reason` | Selection(`bad_password`, `unknown_user`, `disabled_user`, `2fa_failed`, `other`) | Null on success |
|
||||||
|
| `event_time` | Datetime, indexed, default `fields.Datetime.now()` | UTC; displayed in user TZ via standard widget |
|
||||||
|
| `ip_address` | Char(45) | IPv6-safe length |
|
||||||
|
| `ip_hostname` | Char(255) | Reverse DNS, populated by geo cron |
|
||||||
|
| `country_code` | Char(2), indexed | ISO-3166-1 alpha-2; null until cron runs |
|
||||||
|
| `country_name` | Char(64) | |
|
||||||
|
| `city` | Char(128) | |
|
||||||
|
| `geo_state` | Char(64) | Region/state name |
|
||||||
|
| `geo_lookup_state` | Selection(`pending`, `done`, `private_ip`, `internal`, `failed`) | Drives the geo cron worklist; `internal` = no HTTP request was attached |
|
||||||
|
| `user_agent_raw` | Char(512) | The full UA header |
|
||||||
|
| `browser` | Char(64) | e.g. "Chrome 140" — parsed |
|
||||||
|
| `os` | Char(64) | e.g. "Windows 11" — parsed |
|
||||||
|
| `device_type` | Selection(`desktop`, `mobile`, `tablet`, `bot`, `unknown`) | From `user_agents` |
|
||||||
|
| `database` | Char(64) | Multi-DB safety — which DB was logged into |
|
||||||
|
|
||||||
|
**Indexes (in addition to the column-level `indexed=True`):**
|
||||||
|
- `(user_id, event_time DESC)` — per-user history
|
||||||
|
- `(attempted_login, event_time DESC)` — failure-burst detection by login string
|
||||||
|
- `(geo_lookup_state, event_time)` — cron worklist
|
||||||
|
|
||||||
|
**No `_inherit = ['mail.thread']`** — audit rows are append-only and should not have chatter.
|
||||||
|
|
||||||
|
### `res.users` additions (per CLAUDE.md `x_fc_*` convention)
|
||||||
|
|
||||||
|
| Field | Type | Notes |
|
||||||
|
|---|---|---|
|
||||||
|
| `x_fc_login_audit_ids` | One2many(`fusion.login.audit`, `user_id`) | Backs the tab + smart-button count |
|
||||||
|
| `x_fc_login_audit_count` | Integer, compute, store=False | Smart-button label |
|
||||||
|
| `x_fc_last_successful_login` | Datetime, compute, store=True | Indexed; cheap "last seen" lookup |
|
||||||
|
| `x_fc_last_login_ip` | Char(45), compute, store=True | Surfaces last source IP in the form header |
|
||||||
|
|
||||||
|
The `store=True` computes are triggered by the create on `fusion.login.audit` (via `@api.depends('x_fc_login_audit_ids.event_time', 'x_fc_login_audit_ids.result')`).
|
||||||
|
|
||||||
|
### `res.config.settings` additions
|
||||||
|
|
||||||
|
Booleans / integers only (per CLAUDE.md — no Date fields on settings):
|
||||||
|
|
||||||
|
| Field | Default | Notes |
|
||||||
|
|---|---|---|
|
||||||
|
| `x_fc_login_audit_retention_days` | 365 | Retention GC cron honors this; 0 = keep forever |
|
||||||
|
| `x_fc_login_audit_alert_threshold` | 5 | Consecutive failures before alert |
|
||||||
|
| `x_fc_login_audit_alert_window_min` | 15 | Time window in minutes for "consecutive" |
|
||||||
|
| `x_fc_login_audit_alert_enabled` | True | Master kill-switch for alert emails |
|
||||||
|
|
||||||
|
Each is backed by an `ir.config_parameter` (`fusion_login_audit.retention_days`, etc.) so changes from the Settings page persist.
|
||||||
|
|
||||||
|
### Multi-company
|
||||||
|
|
||||||
|
`fusion.login.audit` is intentionally **company-agnostic**. Logins happen before any company context is established; synthesizing one would either break the unknown-user case or require a "system company" placeholder. Settings admins see all rows globally.
|
||||||
|
|
||||||
|
## Capture flow
|
||||||
|
|
||||||
|
### Successful login (`_update_last_login`)
|
||||||
|
|
||||||
|
```python
|
||||||
|
def _update_last_login(self):
|
||||||
|
result = super()._update_last_login()
|
||||||
|
try:
|
||||||
|
self._fc_record_login_event(result='success')
|
||||||
|
except Exception:
|
||||||
|
_logger.exception("fusion_login_audit: failed to record success row for %s", self.login)
|
||||||
|
return result
|
||||||
|
```
|
||||||
|
|
||||||
|
Called by Odoo only after the credential check has passed. Super() runs first so Odoo's own bookkeeping is unaffected.
|
||||||
|
|
||||||
|
### Failed login on known user (`_check_credentials`)
|
||||||
|
|
||||||
|
```python
|
||||||
|
def _check_credentials(self, credential, env):
|
||||||
|
try:
|
||||||
|
return super()._check_credentials(credential, env)
|
||||||
|
except AccessDenied:
|
||||||
|
try:
|
||||||
|
self._fc_record_login_failure(credential, reason='bad_password')
|
||||||
|
if self._fc_recent_failure_count(credential) >= self._fc_alert_threshold():
|
||||||
|
self._fc_send_failure_alert(credential)
|
||||||
|
except Exception:
|
||||||
|
_logger.exception("fusion_login_audit: failed to record/alert failure")
|
||||||
|
raise
|
||||||
|
```
|
||||||
|
|
||||||
|
TOTP failures (from `auth_totp`) also raise `AccessDenied` and are caught here. Distinguish via `credential.get('type') == 'totp'` to set `failure_reason='2fa_failed'`.
|
||||||
|
|
||||||
|
### Failed login on unknown user (`_login` classmethod)
|
||||||
|
|
||||||
|
```python
|
||||||
|
@classmethod
|
||||||
|
def _login(cls, db, credential, user_agent_env):
|
||||||
|
try:
|
||||||
|
return super()._login(db, credential, user_agent_env)
|
||||||
|
except AccessDenied:
|
||||||
|
try:
|
||||||
|
cls._fc_record_unknown_user_failure(db, credential, user_agent_env)
|
||||||
|
except Exception:
|
||||||
|
_logger.exception("fusion_login_audit: failed to record unknown-user failure")
|
||||||
|
raise
|
||||||
|
```
|
||||||
|
|
||||||
|
Without this override, unknown-user attempts never reach `_check_credentials` and would silently disappear from the audit. The classmethod sets `user_id=None` and stores the attempted login string.
|
||||||
|
|
||||||
|
### Context extraction (`_fc_build_event_vals`)
|
||||||
|
|
||||||
|
Single helper shared by all three paths:
|
||||||
|
|
||||||
|
```python
|
||||||
|
def _fc_build_event_vals(self, result, attempted_login, failure_reason=None):
|
||||||
|
from odoo.http import request
|
||||||
|
vals = {
|
||||||
|
'attempted_login': attempted_login,
|
||||||
|
'result': result,
|
||||||
|
'failure_reason': failure_reason,
|
||||||
|
'event_time': fields.Datetime.now(),
|
||||||
|
'database': self.env.cr.dbname,
|
||||||
|
'geo_lookup_state': 'pending',
|
||||||
|
}
|
||||||
|
if request and request.httprequest:
|
||||||
|
vals['ip_address'] = request.httprequest.remote_addr # respects proxy_mode
|
||||||
|
ua_str = request.httprequest.user_agent.string or ''
|
||||||
|
vals['user_agent_raw'] = ua_str[:512]
|
||||||
|
from user_agents import parse as ua_parse
|
||||||
|
ua = ua_parse(ua_str)
|
||||||
|
vals['browser'] = f"{ua.browser.family} {ua.browser.version_string}"[:64]
|
||||||
|
vals['os'] = f"{ua.os.family} {ua.os.version_string}"[:64]
|
||||||
|
vals['device_type'] = (
|
||||||
|
'mobile' if ua.is_mobile else
|
||||||
|
'tablet' if ua.is_tablet else
|
||||||
|
'bot' if ua.is_bot else
|
||||||
|
'desktop' if ua.is_pc else 'unknown'
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
vals['ip_address'] = 'internal'
|
||||||
|
vals['user_agent_raw'] = '<no-request>'
|
||||||
|
vals['geo_lookup_state'] = 'internal' # distinct from private_ip; cron skips both
|
||||||
|
return vals
|
||||||
|
```
|
||||||
|
|
||||||
|
### Write semantics
|
||||||
|
|
||||||
|
- All writes use `self.env['fusion.login.audit'].sudo().create(vals)` — low-privilege users can still generate their own audit rows despite the read-only record rule.
|
||||||
|
- `mail_create_nolog=True` context to avoid chatter noise.
|
||||||
|
- The password value is **never** present in `vals` and is hard-stripped from any `credential` dict before logging. A regression test asserts this.
|
||||||
|
|
||||||
|
## Async geolocation cron (`cron_geo_enrich`)
|
||||||
|
|
||||||
|
**Schedule:** every 5 minutes, `numbercall=-1`, `priority=10`.
|
||||||
|
|
||||||
|
**Worker logic:**
|
||||||
|
|
||||||
|
1. Select 100 oldest rows where `geo_lookup_state='pending'`.
|
||||||
|
2. For each row:
|
||||||
|
- **Private-IP shortcut:** if `ip_address` is in `10.0.0.0/8`, `172.16.0.0/12`, `192.168.0.0/16`, `127.0.0.0/8`, `::1`, or `fe80::/10` → set `geo_lookup_state='private_ip'`, `country_code='--'`, `city='Private network'`.
|
||||||
|
- **Cache check:** look for any prior row with the same `ip_address` and `country_code IS NOT NULL` and `event_time > now() - interval '30 days'`. If found, copy `country_code` / `country_name` / `city` / `geo_state` / `ip_hostname` locally; set state `done`. No external call.
|
||||||
|
- **Reverse DNS:** `socket.gethostbyaddr(ip)` with `socket.setdefaulttimeout(1.5)`.
|
||||||
|
- **HTTP lookup:** `requests.get('http://ip-api.com/json/' + ip, params={'fields': 'status,country,countryCode,regionName,city'}, timeout=3, headers={'User-Agent': 'Odoo-FusionLoginAudit/19.0'})`. The call passes through `network_logger` automatically.
|
||||||
|
- On `status='success'` → fill fields, set state `done`.
|
||||||
|
- On HTTP error, timeout, or `status='fail'` → set state `failed` (no retry).
|
||||||
|
3. `self.env.cr.commit()` after each row so one bad IP cannot roll back the batch.
|
||||||
|
4. **Rate limit defense:** if the response header `X-Rl` is `'0'`, break early and leave remaining rows as `pending` for the next run.
|
||||||
|
|
||||||
|
**Privacy:** the only outbound data is the IP itself. No user identifiers, no Odoo URL, no headers beyond `User-Agent: Odoo-FusionLoginAudit/19.0`. All outbound calls are auditable in `network_logger`.
|
||||||
|
|
||||||
|
## UI surfaces
|
||||||
|
|
||||||
|
### `res.users` form view
|
||||||
|
|
||||||
|
- **Smart button** in the button box, gated `groups="base.group_system"`:
|
||||||
|
```
|
||||||
|
┌──────────────┐
|
||||||
|
│ 🔑 N Logins │
|
||||||
|
└──────────────┘
|
||||||
|
```
|
||||||
|
Click → opens `fusion.login.audit` list view filtered to this user (`domain=[('user_id', '=', active_id)]`).
|
||||||
|
- **New tab "Login Activity"** appended after existing tabs, gated `groups="base.group_system"`:
|
||||||
|
- Header summary: `x_fc_last_successful_login`, `x_fc_last_login_ip` (readonly).
|
||||||
|
- Embedded one2many tree on `x_fc_login_audit_ids`, `limit="30"`, columns: `event_time`, `result` (colored badge), `ip_address`, `country_code` (with flag emoji display), `browser`, `os`, `failure_reason`.
|
||||||
|
- Tree is `create="false" edit="false" delete="false"`.
|
||||||
|
- "View full history →" button below the tree, same action as the smart button.
|
||||||
|
|
||||||
|
### Standalone views for `fusion.login.audit`
|
||||||
|
|
||||||
|
- **List view:** `event_time`, `user_id` (clickable), `attempted_login` (only when `user_id IS NULL`), `result` badge, `ip_address`, `country_code`, `city`, `browser`, `device_type`. Default sort `event_time DESC`.
|
||||||
|
- **Search view:** filters for "Successes", "Failures", "Last 24h", "Last 7d", "Last 30d", "Unknown users (no user_id)"; group-by IP / country / user.
|
||||||
|
- **Form view:** readonly; collapsible "Raw" section for `user_agent_raw`, `ip_hostname`, `database`, `geo_lookup_state`.
|
||||||
|
- **Kanban view:** grouped by `result`, color-coded green/red.
|
||||||
|
|
||||||
|
### Menus
|
||||||
|
|
||||||
|
Under **Settings → Technical → Login Audit**:
|
||||||
|
- "Login Events" → default list view
|
||||||
|
- "Failed Logins (24h)" → list view with default `[('result', '=', 'failure'), ('event_time', '>=', context_today() - 1)]`
|
||||||
|
|
||||||
|
### Settings page
|
||||||
|
|
||||||
|
New "Login Audit" section in **Settings → General Settings** (gated `groups="base.group_system"`):
|
||||||
|
- "Retention period (days)" — integer, help: "0 = keep forever"
|
||||||
|
- "Alert threshold" — integer
|
||||||
|
- "Alert window (minutes)" — integer
|
||||||
|
- "Send failed-login alerts" — boolean
|
||||||
|
|
||||||
|
## Security
|
||||||
|
|
||||||
|
### Group
|
||||||
|
|
||||||
|
No new group created. Read is bound to existing `base.group_system`. Rationale: brainstorming decision was "Settings admins only" — reusing the existing group avoids an extra checkbox to manage.
|
||||||
|
|
||||||
|
### Model access (`ir.model.access.csv`)
|
||||||
|
|
||||||
|
| Group | Read | Write | Create | Unlink |
|
||||||
|
|---|---|---|---|---|
|
||||||
|
| `base.group_system` | ✓ | ✗ | ✗ | ✗ |
|
||||||
|
|
||||||
|
**No write/create/unlink for any group via the UI.** Audit rows are only written via `sudo()` from inside the auth hooks. An audit log admins can mutate is not an audit log.
|
||||||
|
|
||||||
|
### Record rule
|
||||||
|
|
||||||
|
Single global rule on `fusion.login.audit`: read for `base.group_system` only. The user-form one2many is additionally gated at the view level via `groups="base.group_system"` (not via a more permissive record rule) so non-admins have no read path even if they craft a custom view.
|
||||||
|
|
||||||
|
### Field-level
|
||||||
|
|
||||||
|
- `failure_reason` stores a category, never the attempted password.
|
||||||
|
- `_fc_build_event_vals` strips `credential['password']` before any logging or row construction.
|
||||||
|
- The `credential` dict is never persisted.
|
||||||
|
- Regression test: no field on `fusion.login.audit` ever contains a known-test-password string.
|
||||||
|
|
||||||
|
## Retention
|
||||||
|
|
||||||
|
**Cron `cron_retention_gc`** — daily at 03:00 UTC, `numbercall=-1`:
|
||||||
|
|
||||||
|
```python
|
||||||
|
days = int(self.env['ir.config_parameter'].sudo().get_param(
|
||||||
|
'fusion_login_audit.retention_days', 365))
|
||||||
|
if days > 0:
|
||||||
|
cutoff = fields.Datetime.now() - timedelta(days=days)
|
||||||
|
self.env['fusion.login.audit'].sudo().search([
|
||||||
|
('event_time', '<', cutoff)
|
||||||
|
]).unlink()
|
||||||
|
```
|
||||||
|
|
||||||
|
Uses `unlink()` rather than raw `DELETE` so any ORM side effects fire. Expected DB load on `westin-v19`: 27 users × ~2 logins/day × 365 days ≈ 20k rows steady state — trivial for Postgres.
|
||||||
|
|
||||||
|
## Failed-login alert
|
||||||
|
|
||||||
|
**Mail template** in `data/mail_template_data.xml`:
|
||||||
|
|
||||||
|
- **Subject:** `[Login Audit] {threshold} failed login attempts for {attempted_login}`
|
||||||
|
- **Body:** simple HTML table of the last N failure rows for that `attempted_login` — timestamp, IP, country, user-agent summary.
|
||||||
|
- **Recipients:** all users in `base.group_system` with a non-empty `email`.
|
||||||
|
- **Send path:** `mail.mail` queue with `auto_delete=True` so the auth response isn't blocked.
|
||||||
|
|
||||||
|
**Cooldown:** 60 min per `attempted_login`, enforced via an `ir.config_parameter` keyed by `fusion_login_audit.last_alert:{attempted_login}` storing the last-send timestamp. Prevents a sustained attack from flooding admin inboxes.
|
||||||
|
|
||||||
|
**Kill-switch:** if `x_fc_login_audit_alert_enabled = False`, no alerts are sent regardless of threshold.
|
||||||
|
|
||||||
|
## Edge cases
|
||||||
|
|
||||||
|
| Case | Behavior |
|
||||||
|
|---|---|
|
||||||
|
| `request` is None (XML-RPC, internal auth from cron) | Row written with `ip_address='internal'`, `user_agent_raw='<no-request>'`, `geo_lookup_state='internal'` (cron skips) |
|
||||||
|
| Audit insert errors on a hot DB | Login still succeeds — every auth-path hook is wrapped in `try/except Exception: _logger.exception(...)` |
|
||||||
|
| User deleted while audit rows remain | `ondelete='set null'` preserves history; `attempted_login` keeps the readable identifier |
|
||||||
|
| Password reset / `auth_signup` | The reset itself generates no login event; the subsequent login does — matches expectation |
|
||||||
|
| API key authentication | **Out of scope v1** (bypasses `_check_credentials`); documented |
|
||||||
|
| OAuth / SSO | Out of scope v1; no provider configured on westin-v19 |
|
||||||
|
| Portal user (`share=True`) | Logged the same way; smart button remains admin-visible |
|
||||||
|
| Two requests racing on the same private IP | Each writes its own row; geo cache is best-effort, not transactional |
|
||||||
|
| `proxy_mode = False` in `odoo.conf` | `remote_addr` will be the reverse-proxy IP — known limitation, fixable by setting `proxy_mode = True` (out of scope) |
|
||||||
|
|
||||||
|
## Testing
|
||||||
|
|
||||||
|
### `tests/test_login_audit.py` (TransactionCase)
|
||||||
|
|
||||||
|
1. Successful login writes a row with `result='success'` and resolved `user_id`.
|
||||||
|
2. Bad password writes `result='failure'` with `failure_reason='bad_password'` and re-raises `AccessDenied`.
|
||||||
|
3. Unknown user writes `result='failure'` with `failure_reason='unknown_user'`, `user_id=None`, non-null `attempted_login`.
|
||||||
|
4. No field on the written row contains the attempted password (regression).
|
||||||
|
5. Geo cron: pending row gets enriched from local cache when same IP exists within 30 days (no HTTP call made).
|
||||||
|
6. Retention cron: rows older than `retention_days` are deleted; newer survive.
|
||||||
|
7. Alert email: 5 failures in 15 min queues exactly one `mail.mail`; a 6th failure within cooldown queues zero.
|
||||||
|
8. `database` field is populated from `self.env.cr.dbname`.
|
||||||
|
9. Audit-write exception inside `_update_last_login` does not block the login.
|
||||||
|
|
||||||
|
### `tests/test_security.py` (HttpCase)
|
||||||
|
|
||||||
|
1. Non-admin user gets `AccessError` on direct `search(fusion.login.audit)`.
|
||||||
|
2. Non-admin sees the user form view without the smart button or "Login Activity" tab (XML node hidden by `groups`).
|
||||||
|
3. Settings admin sees both.
|
||||||
|
|
||||||
|
## Deployment notes
|
||||||
|
|
||||||
|
- **Local install:** copy module to `K:\Github\Odoo-Modules\fusion_login_audit\` (bind-mounted into `odoo-modsdev-app` container). Update via:
|
||||||
|
```
|
||||||
|
docker exec odoo-modsdev-app odoo -d fusion-dev -i fusion_login_audit --stop-after-init
|
||||||
|
```
|
||||||
|
- **Production install:** sync to `/opt/odoo/custom-addons/fusion_login_audit/` on odoo-westin (via `auto_sync.sh` or git pull on the VM). Update via:
|
||||||
|
```
|
||||||
|
ssh odoo-westin "docker exec odoo-dev-app odoo -d westin-v19 -i fusion_login_audit --stop-after-init"
|
||||||
|
```
|
||||||
|
- **Icon:** copy `C:\Users\gsing\Downloads\fusion logs.png` to `K:\Github\Odoo-Modules\fusion_login_audit\static\description\icon.png`.
|
||||||
|
- **Verify `proxy_mode = True`** in `/opt/odoo/odoo.conf` on odoo-westin before relying on `ip_address` accuracy — otherwise `remote_addr` will be the reverse-proxy IP rather than the real client. Confirmed out of scope for this module, but flag for the operator.
|
||||||
|
- **Verify outbound to `ip-api.com:80`** is reachable from the odoo-westin VM (Tailscale/firewall) — if blocked, `geo_lookup_state` will simply be `failed` and the rest of the module is unaffected.
|
||||||
|
|
||||||
|
## Success criteria
|
||||||
|
|
||||||
|
- Logging in as any user creates exactly one `fusion.login.audit` row with `result='success'` and the correct IP/UA.
|
||||||
|
- Failed login attempts create exactly one row with `result='failure'` and the correct `failure_reason`.
|
||||||
|
- Unknown-user attempts create a row with `user_id=None` and the typed login string in `attempted_login`.
|
||||||
|
- The smart button on `res.users` shows the lifetime count and opens the filtered list.
|
||||||
|
- The "Login Activity" tab shows the last 30 events with correct color coding.
|
||||||
|
- After 5 failures from the same login string within 15 minutes, exactly one alert email arrives in the inbox of every Settings admin with an `email` set.
|
||||||
|
- The geo cron populates `country_code`, `city`, `ip_hostname` for public IPs within 10 minutes of the login.
|
||||||
|
- The retention cron, set to 1 day for a test, deletes rows older than 24 hours and leaves newer ones.
|
||||||
|
- All tests pass: `docker exec odoo-modsdev-app odoo -d fusion-dev --test-enable -i fusion_login_audit --stop-after-init`.
|
||||||
2
fusion_login_audit/__init__.py
Normal file
2
fusion_login_audit/__init__.py
Normal file
@@ -0,0 +1,2 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
from . import models
|
||||||
39
fusion_login_audit/__manifest__.py
Normal file
39
fusion_login_audit/__manifest__.py
Normal file
@@ -0,0 +1,39 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
# Copyright 2026 Nexa Systems Inc.
|
||||||
|
# License OPL-1 (Odoo Proprietary License v1.0)
|
||||||
|
{
|
||||||
|
'name': 'Fusion Login Audit',
|
||||||
|
'version': '19.0.1.0.0',
|
||||||
|
'category': 'Tools',
|
||||||
|
'summary': 'Durable login audit log with geo-enrichment, retention, and failure alerts.',
|
||||||
|
'description': """
|
||||||
|
Fusion Login Audit
|
||||||
|
==================
|
||||||
|
|
||||||
|
Captures every password authentication event (success + failure) in a
|
||||||
|
dedicated, append-only audit table. Surfaces history on the user form
|
||||||
|
as a smart button + tab (admins only). Async-enriches IPs with country,
|
||||||
|
city, and reverse DNS. Emails Settings admins on consecutive-failure
|
||||||
|
bursts. Daily retention cron honours a configurable horizon.
|
||||||
|
""",
|
||||||
|
'author': 'Nexa Systems Inc.',
|
||||||
|
'website': 'https://nexasystems.ca',
|
||||||
|
'license': 'OPL-1',
|
||||||
|
'depends': ['base', 'mail', 'base_setup'],
|
||||||
|
'external_dependencies': {
|
||||||
|
'python': ['user_agents'],
|
||||||
|
},
|
||||||
|
'data': [
|
||||||
|
'security/ir.model.access.csv',
|
||||||
|
'security/security.xml',
|
||||||
|
'data/mail_template_data.xml',
|
||||||
|
'data/ir_cron_data.xml',
|
||||||
|
'views/fusion_login_audit_views.xml',
|
||||||
|
'views/res_users_views.xml',
|
||||||
|
'views/res_config_settings_views.xml',
|
||||||
|
'views/menus.xml',
|
||||||
|
],
|
||||||
|
'installable': True,
|
||||||
|
'application': False,
|
||||||
|
'auto_install': False,
|
||||||
|
}
|
||||||
27
fusion_login_audit/data/ir_cron_data.xml
Normal file
27
fusion_login_audit/data/ir_cron_data.xml
Normal file
@@ -0,0 +1,27 @@
|
|||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<odoo>
|
||||||
|
<data noupdate="1">
|
||||||
|
|
||||||
|
<record id="cron_retention_gc" model="ir.cron">
|
||||||
|
<field name="name">Fusion Login Audit: Retention GC</field>
|
||||||
|
<field name="model_id" ref="model_fusion_login_audit"/>
|
||||||
|
<field name="state">code</field>
|
||||||
|
<field name="code">model._fc_retention_gc()</field>
|
||||||
|
<field name="interval_number">1</field>
|
||||||
|
<field name="interval_type">days</field>
|
||||||
|
<field name="active" eval="True"/>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
<record id="cron_geo_enrich" model="ir.cron">
|
||||||
|
<field name="name">Fusion Login Audit: Geo Enrichment</field>
|
||||||
|
<field name="model_id" ref="model_fusion_login_audit"/>
|
||||||
|
<field name="state">code</field>
|
||||||
|
<field name="code">model._fc_geo_enrich_pending(limit=100)</field>
|
||||||
|
<field name="interval_number">5</field>
|
||||||
|
<field name="interval_type">minutes</field>
|
||||||
|
<field name="active" eval="True"/>
|
||||||
|
<field name="priority">10</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
</data>
|
||||||
|
</odoo>
|
||||||
46
fusion_login_audit/data/mail_template_data.xml
Normal file
46
fusion_login_audit/data/mail_template_data.xml
Normal file
@@ -0,0 +1,46 @@
|
|||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<odoo>
|
||||||
|
<data noupdate="1">
|
||||||
|
|
||||||
|
<record id="mail_template_failure_burst" model="mail.template">
|
||||||
|
<field name="name">Fusion Login Audit — Failure Burst Alert</field>
|
||||||
|
<field name="model_id" ref="base.model_res_users"/>
|
||||||
|
<field name="subject">[Login Audit] Failed login attempts for {{ ctx.get('attempted_login') }}</field>
|
||||||
|
<field name="body_html" type="html">
|
||||||
|
<div>
|
||||||
|
<p>The login audit detected
|
||||||
|
<strong t-out="ctx.get('failure_count')"/> failed login attempt(s)
|
||||||
|
in the last <t t-out="ctx.get('window_min')"/> minute(s) for
|
||||||
|
<strong t-out="ctx.get('attempted_login')"/>.</p>
|
||||||
|
<p>Most recent attempts:</p>
|
||||||
|
<table border="1" cellpadding="4" cellspacing="0"
|
||||||
|
style="border-collapse: collapse; font-family: sans-serif; font-size: 12px;">
|
||||||
|
<thead style="background: #f3f4f6;">
|
||||||
|
<tr>
|
||||||
|
<th>Time</th>
|
||||||
|
<th>IP</th>
|
||||||
|
<th>Country</th>
|
||||||
|
<th>Browser</th>
|
||||||
|
<th>OS</th>
|
||||||
|
</tr>
|
||||||
|
</thead>
|
||||||
|
<tbody>
|
||||||
|
<tr t-foreach="ctx.get('rows', [])" t-as="row">
|
||||||
|
<td t-out="row['event_time']"/>
|
||||||
|
<td t-out="row['ip_address']"/>
|
||||||
|
<td t-out="row.get('country_code') or ''"/>
|
||||||
|
<td t-out="row.get('browser') or ''"/>
|
||||||
|
<td t-out="row.get('os') or ''"/>
|
||||||
|
</tr>
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
<p style="color: #6b7280; font-size: 11px;">
|
||||||
|
Sent by Fusion Login Audit. Tune the threshold and window in
|
||||||
|
Settings → General Settings → Login Audit.
|
||||||
|
</p>
|
||||||
|
</div>
|
||||||
|
</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
</data>
|
||||||
|
</odoo>
|
||||||
4
fusion_login_audit/models/__init__.py
Normal file
4
fusion_login_audit/models/__init__.py
Normal file
@@ -0,0 +1,4 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
from . import fusion_login_audit
|
||||||
|
from . import res_users
|
||||||
|
from . import res_config_settings
|
||||||
256
fusion_login_audit/models/fusion_login_audit.py
Normal file
256
fusion_login_audit/models/fusion_login_audit.py
Normal file
@@ -0,0 +1,256 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
import ipaddress
|
||||||
|
import logging
|
||||||
|
import socket
|
||||||
|
from datetime import timedelta
|
||||||
|
|
||||||
|
import requests
|
||||||
|
|
||||||
|
from odoo import api, fields, models
|
||||||
|
|
||||||
|
_logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class FusionLoginAudit(models.Model):
|
||||||
|
_name = 'fusion.login.audit'
|
||||||
|
_description = 'Login Audit Event'
|
||||||
|
_order = 'event_time desc, id desc'
|
||||||
|
_rec_name = 'attempted_login'
|
||||||
|
|
||||||
|
user_id = fields.Many2one(
|
||||||
|
'res.users', string='User', ondelete='set null', index=True,
|
||||||
|
help='Null when the attempted login did not match any user.',
|
||||||
|
)
|
||||||
|
attempted_login = fields.Char(
|
||||||
|
string='Attempted Login', size=255, required=True, index=True,
|
||||||
|
)
|
||||||
|
result = fields.Selection(
|
||||||
|
[('success', 'Success'), ('failure', 'Failure')],
|
||||||
|
string='Result', required=True, index=True,
|
||||||
|
)
|
||||||
|
failure_reason = fields.Selection(
|
||||||
|
[
|
||||||
|
('bad_password', 'Bad password'),
|
||||||
|
('unknown_user', 'Unknown user'),
|
||||||
|
('disabled_user', 'Disabled user'),
|
||||||
|
('2fa_failed', '2FA failed'),
|
||||||
|
('other', 'Other'),
|
||||||
|
],
|
||||||
|
string='Failure Reason',
|
||||||
|
)
|
||||||
|
event_time = fields.Datetime(
|
||||||
|
string='Event Time', required=True, index=True,
|
||||||
|
default=fields.Datetime.now,
|
||||||
|
)
|
||||||
|
ip_address = fields.Char(string='IP Address', size=45)
|
||||||
|
ip_hostname = fields.Char(string='Reverse DNS', size=255)
|
||||||
|
country_code = fields.Char(string='Country Code', size=2, index=True)
|
||||||
|
country_name = fields.Char(string='Country', size=64)
|
||||||
|
city = fields.Char(string='City', size=128)
|
||||||
|
geo_state = fields.Char(string='Region', size=64)
|
||||||
|
geo_lookup_state = fields.Selection(
|
||||||
|
[
|
||||||
|
('pending', 'Pending'),
|
||||||
|
('done', 'Done'),
|
||||||
|
('private_ip', 'Private IP'),
|
||||||
|
('internal', 'Internal (no request)'),
|
||||||
|
('failed', 'Lookup failed'),
|
||||||
|
],
|
||||||
|
string='Geo Lookup State', default='pending', index=True,
|
||||||
|
)
|
||||||
|
user_agent_raw = fields.Char(string='User Agent', size=512)
|
||||||
|
browser = fields.Char(string='Browser', size=64)
|
||||||
|
os = fields.Char(string='OS', size=64)
|
||||||
|
device_type = fields.Selection(
|
||||||
|
[
|
||||||
|
('desktop', 'Desktop'),
|
||||||
|
('mobile', 'Mobile'),
|
||||||
|
('tablet', 'Tablet'),
|
||||||
|
('bot', 'Bot'),
|
||||||
|
('unknown', 'Unknown'),
|
||||||
|
],
|
||||||
|
string='Device Type', default='unknown',
|
||||||
|
)
|
||||||
|
database = fields.Char(string='Database', size=64)
|
||||||
|
|
||||||
|
# Odoo 19 replaces the legacy `_sql_constraints = [...]` list with
|
||||||
|
# declarative `models.Constraint` attributes. The plan template used the
|
||||||
|
# legacy form, which now only emits a warning and is silently dropped.
|
||||||
|
_result_failure_reason_consistent = models.Constraint(
|
||||||
|
"CHECK ((result = 'success' AND failure_reason IS NULL) "
|
||||||
|
"OR (result = 'failure' AND failure_reason IS NOT NULL))",
|
||||||
|
'A failure row must have a failure_reason; a success row must not.',
|
||||||
|
)
|
||||||
|
|
||||||
|
# Composite indexes supporting the three hot queries:
|
||||||
|
# - per-user history (user_id, event_time DESC)
|
||||||
|
# - failure-burst by login (attempted_login, event_time DESC)
|
||||||
|
# - geo cron worklist (geo_lookup_state, event_time)
|
||||||
|
# Odoo 19 ships `models.Index` as the declarative replacement for the
|
||||||
|
# init()/raw-SQL pattern; the attribute name becomes the index suffix
|
||||||
|
# (e.g. `_user_time_idx` -> `fusion_login_audit_user_time_idx`).
|
||||||
|
_user_time_idx = models.Index('(user_id, event_time DESC)')
|
||||||
|
_login_time_idx = models.Index('(attempted_login, event_time DESC)')
|
||||||
|
_geo_state_idx = models.Index('(geo_lookup_state, event_time)')
|
||||||
|
|
||||||
|
@api.model
|
||||||
|
def _fc_retention_gc(self):
|
||||||
|
"""Delete audit rows older than `fusion_login_audit.retention_days`.
|
||||||
|
Called daily by ir.cron. retention_days=0 means keep forever."""
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
try:
|
||||||
|
days = int(ICP.get_param(
|
||||||
|
'fusion_login_audit.retention_days', 365))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
days = 365
|
||||||
|
if days <= 0:
|
||||||
|
return 0
|
||||||
|
cutoff = fields.Datetime.now() - timedelta(days=days)
|
||||||
|
old = self.sudo().search([('event_time', '<', cutoff)])
|
||||||
|
count = len(old)
|
||||||
|
if old:
|
||||||
|
old.unlink()
|
||||||
|
return count
|
||||||
|
|
||||||
|
_FC_PRIVATE_NETWORKS = (
|
||||||
|
ipaddress.ip_network('10.0.0.0/8'),
|
||||||
|
ipaddress.ip_network('172.16.0.0/12'),
|
||||||
|
ipaddress.ip_network('192.168.0.0/16'),
|
||||||
|
ipaddress.ip_network('127.0.0.0/8'),
|
||||||
|
ipaddress.ip_network('::1/128'),
|
||||||
|
ipaddress.ip_network('fe80::/10'),
|
||||||
|
)
|
||||||
|
|
||||||
|
@api.model
|
||||||
|
def _fc_is_private_ip(self, ip):
|
||||||
|
if not ip or ip == 'internal':
|
||||||
|
return False # 'internal' uses its own state
|
||||||
|
try:
|
||||||
|
addr = ipaddress.ip_address(ip)
|
||||||
|
except ValueError:
|
||||||
|
return False
|
||||||
|
return any(addr in net for net in self._FC_PRIVATE_NETWORKS)
|
||||||
|
|
||||||
|
@api.model
|
||||||
|
def _fc_geo_cache_hit(self, ip):
|
||||||
|
"""Return a dict of geo fields if we've resolved this IP in the last
|
||||||
|
30 days, else None."""
|
||||||
|
if not ip:
|
||||||
|
return None
|
||||||
|
cutoff = fields.Datetime.now() - timedelta(days=30)
|
||||||
|
cached = self.sudo().search([
|
||||||
|
('ip_address', '=', ip),
|
||||||
|
('geo_lookup_state', '=', 'done'),
|
||||||
|
('event_time', '>=', cutoff),
|
||||||
|
], limit=1, order='event_time desc')
|
||||||
|
if not cached:
|
||||||
|
return None
|
||||||
|
return {
|
||||||
|
'country_code': cached.country_code,
|
||||||
|
'country_name': cached.country_name,
|
||||||
|
'city': cached.city,
|
||||||
|
'geo_state': cached.geo_state,
|
||||||
|
'ip_hostname': cached.ip_hostname,
|
||||||
|
}
|
||||||
|
|
||||||
|
@api.model
|
||||||
|
def _fc_geo_reverse_dns(self, ip):
|
||||||
|
try:
|
||||||
|
socket.setdefaulttimeout(1.5)
|
||||||
|
host, _aliases, _ips = socket.gethostbyaddr(ip)
|
||||||
|
return (host or '')[:255]
|
||||||
|
except (socket.herror, socket.gaierror, OSError):
|
||||||
|
return ''
|
||||||
|
finally:
|
||||||
|
socket.setdefaulttimeout(None)
|
||||||
|
|
||||||
|
@api.model
|
||||||
|
def _fc_geo_http_lookup(self, ip):
|
||||||
|
"""Call ip-api.com. Returns (vals_dict, rate_limited_bool).
|
||||||
|
Falls back to ({}, False) on any error."""
|
||||||
|
try:
|
||||||
|
resp = requests.get(
|
||||||
|
'http://ip-api.com/json/' + ip,
|
||||||
|
params={'fields': 'status,country,countryCode,regionName,city'},
|
||||||
|
timeout=3,
|
||||||
|
headers={'User-Agent': 'Odoo-FusionLoginAudit/19.0'},
|
||||||
|
)
|
||||||
|
rate_limited = resp.headers.get('X-Rl', '') == '0'
|
||||||
|
if resp.status_code != 200:
|
||||||
|
return ({}, rate_limited)
|
||||||
|
data = resp.json() or {}
|
||||||
|
if data.get('status') != 'success':
|
||||||
|
return ({}, rate_limited)
|
||||||
|
return ({
|
||||||
|
'country_code': (data.get('countryCode') or '')[:2],
|
||||||
|
'country_name': (data.get('country') or '')[:64],
|
||||||
|
'geo_state': (data.get('regionName') or '')[:64],
|
||||||
|
'city': (data.get('city') or '')[:128],
|
||||||
|
}, rate_limited)
|
||||||
|
except (requests.RequestException, ValueError):
|
||||||
|
_logger.warning("fusion_login_audit: geo lookup failed for %s",
|
||||||
|
ip, exc_info=True)
|
||||||
|
return ({}, False)
|
||||||
|
|
||||||
|
@api.model
|
||||||
|
def _fc_geo_enrich_pending(self, limit=100):
|
||||||
|
"""Cron worker: process up to `limit` pending rows.
|
||||||
|
|
||||||
|
Per-row isolation is provided by `cr.savepoint()` rather than
|
||||||
|
`cr.commit()`/`cr.rollback()` — the latter raises an AssertionError
|
||||||
|
inside a TransactionCase (Odoo's test cursor refuses commit/rollback).
|
||||||
|
Savepoints work in both prod and tests; the outer cron transaction
|
||||||
|
commits the lot once the method returns. One bad IP rolls back only
|
||||||
|
its own savepoint, so the rest of the batch still lands.
|
||||||
|
"""
|
||||||
|
pending = self.sudo().search(
|
||||||
|
[('geo_lookup_state', '=', 'pending')],
|
||||||
|
order='event_time asc', limit=limit,
|
||||||
|
)
|
||||||
|
if not pending:
|
||||||
|
return 0
|
||||||
|
processed = 0
|
||||||
|
stop_after_this = False
|
||||||
|
for row in pending:
|
||||||
|
ip = row.ip_address
|
||||||
|
try:
|
||||||
|
with self.env.cr.savepoint():
|
||||||
|
if self._fc_is_private_ip(ip):
|
||||||
|
row.write({
|
||||||
|
'geo_lookup_state': 'private_ip',
|
||||||
|
'country_code': '--',
|
||||||
|
'country_name': 'Private network',
|
||||||
|
'city': 'Private network',
|
||||||
|
})
|
||||||
|
processed += 1
|
||||||
|
continue
|
||||||
|
|
||||||
|
cached = self._fc_geo_cache_hit(ip)
|
||||||
|
if cached:
|
||||||
|
cached['geo_lookup_state'] = 'done'
|
||||||
|
row.write(cached)
|
||||||
|
processed += 1
|
||||||
|
continue
|
||||||
|
|
||||||
|
hostname = self._fc_geo_reverse_dns(ip) if ip and ip != 'internal' else ''
|
||||||
|
vals, rate_limited = self._fc_geo_http_lookup(ip) if ip and ip != 'internal' else ({}, False)
|
||||||
|
if vals:
|
||||||
|
vals['ip_hostname'] = hostname
|
||||||
|
vals['geo_lookup_state'] = 'done'
|
||||||
|
row.write(vals)
|
||||||
|
else:
|
||||||
|
row.write({
|
||||||
|
'geo_lookup_state': 'failed',
|
||||||
|
'ip_hostname': hostname,
|
||||||
|
})
|
||||||
|
processed += 1
|
||||||
|
if rate_limited:
|
||||||
|
_logger.info("fusion_login_audit: ip-api rate limit "
|
||||||
|
"hit, stopping batch early")
|
||||||
|
stop_after_this = True
|
||||||
|
except Exception:
|
||||||
|
_logger.exception(
|
||||||
|
"fusion_login_audit: geo enrich failed for row %s", row.id)
|
||||||
|
if stop_after_this:
|
||||||
|
break
|
||||||
|
return processed
|
||||||
31
fusion_login_audit/models/res_config_settings.py
Normal file
31
fusion_login_audit/models/res_config_settings.py
Normal file
@@ -0,0 +1,31 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
from odoo import fields, models
|
||||||
|
|
||||||
|
|
||||||
|
class ResConfigSettings(models.TransientModel):
|
||||||
|
_inherit = 'res.config.settings'
|
||||||
|
|
||||||
|
x_fc_login_audit_retention_days = fields.Integer(
|
||||||
|
string='Login Audit Retention (days)',
|
||||||
|
default=365,
|
||||||
|
config_parameter='fusion_login_audit.retention_days',
|
||||||
|
help='Login audit rows older than this are deleted by the nightly '
|
||||||
|
'cron. Set to 0 to keep forever.',
|
||||||
|
)
|
||||||
|
x_fc_login_audit_alert_threshold = fields.Integer(
|
||||||
|
string='Alert After N Consecutive Failures',
|
||||||
|
default=5,
|
||||||
|
config_parameter='fusion_login_audit.alert_threshold',
|
||||||
|
help='When this many failures for the same attempted login occur '
|
||||||
|
'within the alert window, Settings admins receive one email.',
|
||||||
|
)
|
||||||
|
x_fc_login_audit_alert_window_min = fields.Integer(
|
||||||
|
string='Alert Window (minutes)',
|
||||||
|
default=15,
|
||||||
|
config_parameter='fusion_login_audit.alert_window_min',
|
||||||
|
)
|
||||||
|
x_fc_login_audit_alert_enabled = fields.Boolean(
|
||||||
|
string='Send Failed-Login Alerts',
|
||||||
|
default=True,
|
||||||
|
config_parameter='fusion_login_audit.alert_enabled',
|
||||||
|
)
|
||||||
392
fusion_login_audit/models/res_users.py
Normal file
392
fusion_login_audit/models/res_users.py
Normal file
@@ -0,0 +1,392 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from odoo import _, api, fields, models
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
|
||||||
|
# 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.
|
||||||
|
|
||||||
|
The row is written through an INDEPENDENT cursor
|
||||||
|
(``registry.cursor()``) so that:
|
||||||
|
|
||||||
|
* A failure-path call from ``_check_credentials`` survives the
|
||||||
|
outer transaction rollback that follows ``AccessDenied``
|
||||||
|
(the HTTP layer closes the cursor without committing, see
|
||||||
|
``odoo/service/model.py:retrying``).
|
||||||
|
* A broken audit table can never block a real user from logging
|
||||||
|
in: the cursor block is wrapped in try/except; exceptions are
|
||||||
|
logged and swallowed.
|
||||||
|
|
||||||
|
The independent cursor commits on context exit. Note that this
|
||||||
|
means the row is durable even if the caller's transaction later
|
||||||
|
rolls back — intentional for audit semantics: a recorded bad
|
||||||
|
password should NOT disappear because some unrelated downstream
|
||||||
|
op blew up.
|
||||||
|
"""
|
||||||
|
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,
|
||||||
|
)
|
||||||
|
with self.env.registry.cursor() as audit_cr:
|
||||||
|
audit_env = api.Environment(audit_cr, self.env.uid, self.env.context)
|
||||||
|
audit_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
|
||||||
|
|
||||||
|
def _fc_alert_threshold(self):
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
try:
|
||||||
|
return max(1, int(ICP.get_param(
|
||||||
|
'fusion_login_audit.alert_threshold', 5)))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return 5
|
||||||
|
|
||||||
|
def _fc_alert_window_min(self):
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
try:
|
||||||
|
return max(1, int(ICP.get_param(
|
||||||
|
'fusion_login_audit.alert_window_min', 15)))
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return 15
|
||||||
|
|
||||||
|
def _fc_alert_enabled(self):
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
# CLAUDE.md rule #5: Boolean config_parameter deletes on False.
|
||||||
|
# An absent key means True (the default). Explicit 'False' or 'false'
|
||||||
|
# means disabled.
|
||||||
|
raw = ICP.get_param('fusion_login_audit.alert_enabled', 'True')
|
||||||
|
return str(raw).strip().lower() != 'false'
|
||||||
|
|
||||||
|
def _fc_recent_failure_count(self, attempted_login):
|
||||||
|
"""Failures for this attempted_login within the alert window."""
|
||||||
|
from datetime import timedelta
|
||||||
|
if not attempted_login:
|
||||||
|
return 0
|
||||||
|
cutoff = fields.Datetime.now() - timedelta(
|
||||||
|
minutes=self._fc_alert_window_min())
|
||||||
|
return self.env['fusion.login.audit'].sudo().search_count([
|
||||||
|
('attempted_login', '=', attempted_login),
|
||||||
|
('result', '=', 'failure'),
|
||||||
|
('event_time', '>=', cutoff),
|
||||||
|
])
|
||||||
|
|
||||||
|
def _fc_send_failure_alert(self, attempted_login):
|
||||||
|
"""Queue one alert mail unless cooldown is active. Cooldown is
|
||||||
|
60 minutes, keyed by attempted_login, stored in ir.config_parameter."""
|
||||||
|
from datetime import timedelta
|
||||||
|
if not self._fc_alert_enabled():
|
||||||
|
return
|
||||||
|
if not attempted_login:
|
||||||
|
return
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
cd_key = f'fusion_login_audit.last_alert:{attempted_login}'
|
||||||
|
cd_raw = ICP.get_param(cd_key)
|
||||||
|
now = fields.Datetime.now()
|
||||||
|
if cd_raw:
|
||||||
|
try:
|
||||||
|
last = fields.Datetime.from_string(cd_raw)
|
||||||
|
if last and (now - last) < timedelta(minutes=60):
|
||||||
|
return # cooldown active
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
window = self._fc_alert_window_min()
|
||||||
|
cutoff = now - timedelta(minutes=window)
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
rows = Audit.search([
|
||||||
|
('attempted_login', '=', attempted_login),
|
||||||
|
('result', '=', 'failure'),
|
||||||
|
('event_time', '>=', cutoff),
|
||||||
|
], order='event_time desc', limit=20)
|
||||||
|
|
||||||
|
# Admin recipients: members of base.group_system (the Settings group)
|
||||||
|
# who have an email set and are not portal/share users. Note:
|
||||||
|
# CLAUDE.md rule #6 — res.groups has no `users` field in Odoo 19, so
|
||||||
|
# search res.users by group_ids directly. The __system__ superuser
|
||||||
|
# (uid=1) is excluded automatically by Odoo's default user filter.
|
||||||
|
admins = self.env['res.users'].sudo().search([
|
||||||
|
('group_ids', 'in', self.env.ref('base.group_system').id),
|
||||||
|
('email', '!=', False),
|
||||||
|
('share', '=', False),
|
||||||
|
])
|
||||||
|
if not admins:
|
||||||
|
return
|
||||||
|
|
||||||
|
tmpl = self.env.ref(
|
||||||
|
'fusion_login_audit.mail_template_failure_burst',
|
||||||
|
raise_if_not_found=False)
|
||||||
|
if not tmpl:
|
||||||
|
return
|
||||||
|
|
||||||
|
# CLAUDE.md rule #12: in mail.template QWeb, `ctx` IS env.context.
|
||||||
|
# So `ctx.get('foo')` resolves to env.context.get('foo'). Pass data
|
||||||
|
# by SPREADING keys into the context, not wrapping in a dict.
|
||||||
|
# `with_context(ctx=ctx_data)` would silently render an empty subject.
|
||||||
|
ctx_data = {
|
||||||
|
'attempted_login': attempted_login,
|
||||||
|
'failure_count': len(rows),
|
||||||
|
'window_min': window,
|
||||||
|
'rows': [{
|
||||||
|
'event_time': fields.Datetime.to_string(r.event_time),
|
||||||
|
'ip_address': r.ip_address or '',
|
||||||
|
'country_code': r.country_code or '',
|
||||||
|
'browser': r.browser or '',
|
||||||
|
'os': r.os or '',
|
||||||
|
} for r in rows],
|
||||||
|
}
|
||||||
|
for admin in admins:
|
||||||
|
tmpl.with_context(**ctx_data).send_mail(
|
||||||
|
admin.id,
|
||||||
|
email_values={'email_to': admin.email,
|
||||||
|
'auto_delete': True},
|
||||||
|
force_send=False,
|
||||||
|
)
|
||||||
|
ICP.set_param(cd_key, fields.Datetime.to_string(now))
|
||||||
|
|
||||||
|
def _check_credentials(self, credential, env):
|
||||||
|
try:
|
||||||
|
return super()._check_credentials(credential, env)
|
||||||
|
except AccessDenied:
|
||||||
|
cred_type = (credential or {}).get('type', 'password')
|
||||||
|
reason = '2fa_failed' if cred_type == 'totp' else 'bad_password'
|
||||||
|
attempted_login = (credential or {}).get('login') or self.login
|
||||||
|
self._fc_record_login_event(
|
||||||
|
result='failure',
|
||||||
|
failure_reason=reason,
|
||||||
|
user_id=self.id,
|
||||||
|
attempted_login=attempted_login,
|
||||||
|
_credential=credential,
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
if self._fc_recent_failure_count(attempted_login) \
|
||||||
|
>= self._fc_alert_threshold():
|
||||||
|
self._fc_send_failure_alert(attempted_login)
|
||||||
|
except Exception:
|
||||||
|
_logger.exception(
|
||||||
|
"fusion_login_audit: failed to send failure alert")
|
||||||
|
raise
|
||||||
|
|
||||||
|
def _login(self, credential, user_agent_env):
|
||||||
|
"""Catch the unknown-user branch of upstream _login.
|
||||||
|
|
||||||
|
In Odoo 19 ``_login`` is an *instance* method (not a classmethod as in
|
||||||
|
earlier versions). Upstream raises ``AccessDenied`` in three cases:
|
||||||
|
|
||||||
|
1. Unknown login string — ``_assert_can_auth`` or the user-lookup
|
||||||
|
``search()`` returns empty → ``_check_credentials`` never fires →
|
||||||
|
THIS override is the only chance to record the attempt.
|
||||||
|
2. Wrong password — user exists, ``_check_credentials`` raises →
|
||||||
|
our ``_check_credentials`` override already wrote a ``bad_password``
|
||||||
|
row → re-raise propagates up to here. We MUST NOT write a second
|
||||||
|
row.
|
||||||
|
3. 2FA failure — same as #2 but ``failure_reason='2fa_failed'``.
|
||||||
|
|
||||||
|
We distinguish #1 from #2/#3 by checking whether the login string
|
||||||
|
resolves to any user. If it does, ``_check_credentials`` ran (and
|
||||||
|
already logged); if it doesn't, the user lookup failed and we log
|
||||||
|
``unknown_user`` here.
|
||||||
|
|
||||||
|
``_fc_record_login_event`` writes through an INDEPENDENT cursor
|
||||||
|
(``self.env.registry.cursor()``), so the audit row survives the
|
||||||
|
outer transaction rollback that follows the re-raised
|
||||||
|
``AccessDenied``. Audit-side exceptions never block the re-raise.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
return super()._login(credential, user_agent_env)
|
||||||
|
except AccessDenied:
|
||||||
|
login = (credential or {}).get('login') or ''
|
||||||
|
try:
|
||||||
|
user_exists = bool(self.sudo().search(
|
||||||
|
[('login', '=', login)], limit=1))
|
||||||
|
except Exception:
|
||||||
|
user_exists = False # be permissive — log the row anyway
|
||||||
|
if not user_exists:
|
||||||
|
self._fc_record_login_event(
|
||||||
|
result='failure',
|
||||||
|
failure_reason='unknown_user',
|
||||||
|
user_id=False,
|
||||||
|
attempted_login=login or 'unknown',
|
||||||
|
_credential=credential,
|
||||||
|
)
|
||||||
|
raise
|
||||||
|
|
||||||
|
# ──────────────────────────────────────────────────────────────────
|
||||||
|
# Per-user surface — fields + action method backing the smart button
|
||||||
|
# and "Login Activity" tab on the user form view.
|
||||||
|
# ──────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
x_fc_login_audit_ids = fields.One2many(
|
||||||
|
'fusion.login.audit', 'user_id',
|
||||||
|
string='Login Activity',
|
||||||
|
)
|
||||||
|
x_fc_login_audit_count = fields.Integer(
|
||||||
|
string='Login Audit Count',
|
||||||
|
compute='_compute_x_fc_login_audit_count',
|
||||||
|
)
|
||||||
|
x_fc_last_successful_login = fields.Datetime(
|
||||||
|
string='Last Successful Login',
|
||||||
|
compute='_compute_x_fc_last_successful_login',
|
||||||
|
store=True,
|
||||||
|
)
|
||||||
|
x_fc_last_login_ip = fields.Char(
|
||||||
|
string='Last Login IP', size=45,
|
||||||
|
compute='_compute_x_fc_last_successful_login',
|
||||||
|
store=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
@api.depends('x_fc_login_audit_ids')
|
||||||
|
def _compute_x_fc_login_audit_count(self):
|
||||||
|
# Odoo 19: read_group → _read_group, returns list of tuples
|
||||||
|
# (group_key, aggregate_value) when given groupby + aggregates.
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
rows = Audit._read_group(
|
||||||
|
domain=[('user_id', 'in', self.ids)],
|
||||||
|
groupby=['user_id'],
|
||||||
|
aggregates=['__count'],
|
||||||
|
)
|
||||||
|
counts = {user.id: count for user, count in rows}
|
||||||
|
for user in self:
|
||||||
|
user.x_fc_login_audit_count = counts.get(user.id, 0)
|
||||||
|
|
||||||
|
@api.depends('x_fc_login_audit_ids.event_time',
|
||||||
|
'x_fc_login_audit_ids.result',
|
||||||
|
'x_fc_login_audit_ids.ip_address')
|
||||||
|
def _compute_x_fc_last_successful_login(self):
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
for user in self:
|
||||||
|
row = Audit.search(
|
||||||
|
[('user_id', '=', user.id), ('result', '=', 'success')],
|
||||||
|
order='event_time desc', limit=1,
|
||||||
|
)
|
||||||
|
user.x_fc_last_successful_login = row.event_time or False
|
||||||
|
user.x_fc_last_login_ip = row.ip_address or False
|
||||||
|
|
||||||
|
def action_fc_view_login_audit(self):
|
||||||
|
self.ensure_one()
|
||||||
|
return {
|
||||||
|
'name': _('Login Activity'),
|
||||||
|
'type': 'ir.actions.act_window',
|
||||||
|
'res_model': 'fusion.login.audit',
|
||||||
|
'view_mode': 'list,form',
|
||||||
|
'domain': [('user_id', '=', self.id)],
|
||||||
|
'context': {'create': False, 'edit': False, 'delete': False,
|
||||||
|
'default_user_id': self.id},
|
||||||
|
}
|
||||||
2
fusion_login_audit/security/ir.model.access.csv
Normal file
2
fusion_login_audit/security/ir.model.access.csv
Normal file
@@ -0,0 +1,2 @@
|
|||||||
|
id,name,model_id:id,group_id:id,perm_read,perm_write,perm_create,perm_unlink
|
||||||
|
access_fusion_login_audit_system,fusion.login.audit system,model_fusion_login_audit,base.group_system,1,0,0,0
|
||||||
|
17
fusion_login_audit/security/security.xml
Normal file
17
fusion_login_audit/security/security.xml
Normal file
@@ -0,0 +1,17 @@
|
|||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<odoo>
|
||||||
|
<data noupdate="1">
|
||||||
|
|
||||||
|
<record id="rule_fusion_login_audit_admin_read" model="ir.rule">
|
||||||
|
<field name="name">fusion.login.audit: admin read only</field>
|
||||||
|
<field name="model_id" ref="model_fusion_login_audit"/>
|
||||||
|
<field name="domain_force">[(1, '=', 1)]</field>
|
||||||
|
<field name="groups" eval="[(4, ref('base.group_system'))]"/>
|
||||||
|
<field name="perm_read" eval="True"/>
|
||||||
|
<field name="perm_write" eval="False"/>
|
||||||
|
<field name="perm_create" eval="False"/>
|
||||||
|
<field name="perm_unlink" eval="False"/>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
</data>
|
||||||
|
</odoo>
|
||||||
BIN
fusion_login_audit/static/description/icon.png
Normal file
BIN
fusion_login_audit/static/description/icon.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 41 KiB |
3
fusion_login_audit/tests/__init__.py
Normal file
3
fusion_login_audit/tests/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
from . import test_login_audit
|
||||||
|
from . import test_security
|
||||||
541
fusion_login_audit/tests/test_login_audit.py
Normal file
541
fusion_login_audit/tests/test_login_audit.py
Normal file
@@ -0,0 +1,541 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
from odoo import fields
|
||||||
|
from odoo.tests.common import TransactionCase, tagged
|
||||||
|
|
||||||
|
|
||||||
|
@tagged('post_install', '-at_install')
|
||||||
|
class TestFusionLoginAuditModel(TransactionCase):
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
# `_fc_record_login_event` uses `registry.cursor()` so that the audit
|
||||||
|
# row survives the outer rollback that follows AccessDenied (see
|
||||||
|
# res_users.py for the rationale). Inside a TransactionCase that
|
||||||
|
# rolls back per test, a fresh cursor on a new connection cannot
|
||||||
|
# see uncommitted records (the freshly-created test user FKs into
|
||||||
|
# the audit row), so we put the registry in test mode — that swaps
|
||||||
|
# `registry.cursor()` for a TestCursor that wraps the test cursor.
|
||||||
|
super().setUp()
|
||||||
|
self.registry_enter_test_mode()
|
||||||
|
# The alert tests below assume at least one admin has an email
|
||||||
|
# (otherwise the recipient filter empties and no mail is queued).
|
||||||
|
# In a fresh fusion-dev DB, base.user_admin's email is NULL; the
|
||||||
|
# superuser (__system__) has an email but is filtered out of normal
|
||||||
|
# res.users searches. Ensure admin has a usable email.
|
||||||
|
admin = self.env.ref('base.user_admin')
|
||||||
|
if not admin.email:
|
||||||
|
admin.sudo().write({'email': 'admin@test.example.com'})
|
||||||
|
|
||||||
|
def test_model_exists_and_creates(self):
|
||||||
|
"""Audit row can be created with all expected fields."""
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
rec = Audit.create({
|
||||||
|
'attempted_login': 'demo@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'ip_address': '203.0.113.5',
|
||||||
|
'user_agent_raw': 'Mozilla/5.0 Test',
|
||||||
|
'browser': 'Test 1.0',
|
||||||
|
'os': 'TestOS',
|
||||||
|
'device_type': 'desktop',
|
||||||
|
'database': self.env.cr.dbname,
|
||||||
|
'geo_lookup_state': 'pending',
|
||||||
|
})
|
||||||
|
self.assertTrue(rec.id)
|
||||||
|
self.assertEqual(rec.result, 'success')
|
||||||
|
self.assertEqual(rec.geo_lookup_state, 'pending')
|
||||||
|
self.assertEqual(rec.database, self.env.cr.dbname)
|
||||||
|
self.assertTrue(rec.event_time) # default fires
|
||||||
|
|
||||||
|
def test_failure_reason_optional(self):
|
||||||
|
"""failure_reason is null on success rows."""
|
||||||
|
rec = self.env['fusion.login.audit'].sudo().create({
|
||||||
|
'attempted_login': 'demo@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
})
|
||||||
|
self.assertFalse(rec.failure_reason)
|
||||||
|
|
||||||
|
def test_geo_state_internal_value(self):
|
||||||
|
"""`internal` is an accepted geo_lookup_state value (distinct from private_ip)."""
|
||||||
|
rec = self.env['fusion.login.audit'].sudo().create({
|
||||||
|
'attempted_login': 'demo@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'geo_lookup_state': 'internal',
|
||||||
|
})
|
||||||
|
self.assertEqual(rec.geo_lookup_state, 'internal')
|
||||||
|
|
||||||
|
def test_build_event_vals_with_no_request(self):
|
||||||
|
"""Without a live request, geo_lookup_state is 'internal'."""
|
||||||
|
ResUsers = self.env['res.users']
|
||||||
|
vals = ResUsers._fc_build_event_vals(
|
||||||
|
result='success',
|
||||||
|
attempted_login='cron@example.com',
|
||||||
|
)
|
||||||
|
self.assertEqual(vals['result'], 'success')
|
||||||
|
self.assertEqual(vals['attempted_login'], 'cron@example.com')
|
||||||
|
self.assertEqual(vals['ip_address'], 'internal')
|
||||||
|
self.assertEqual(vals['user_agent_raw'], '<no-request>')
|
||||||
|
self.assertEqual(vals['geo_lookup_state'], 'internal')
|
||||||
|
self.assertEqual(vals['database'], self.env.cr.dbname)
|
||||||
|
|
||||||
|
def test_build_event_vals_parses_user_agent(self):
|
||||||
|
"""Parser fills browser/os/device_type from a stub UA dict."""
|
||||||
|
ResUsers = self.env['res.users']
|
||||||
|
vals = ResUsers._fc_build_event_vals(
|
||||||
|
result='success',
|
||||||
|
attempted_login='ua@example.com',
|
||||||
|
_override_ip='203.0.113.5',
|
||||||
|
_override_ua='Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
|
||||||
|
'AppleWebKit/537.36 Chrome/140.0 Safari/537.36',
|
||||||
|
)
|
||||||
|
self.assertEqual(vals['ip_address'], '203.0.113.5')
|
||||||
|
self.assertIn('Chrome', vals['browser'])
|
||||||
|
self.assertIn('Windows', vals['os'])
|
||||||
|
self.assertEqual(vals['device_type'], 'desktop')
|
||||||
|
self.assertEqual(vals['geo_lookup_state'], 'pending')
|
||||||
|
|
||||||
|
def test_build_event_vals_strips_password(self):
|
||||||
|
"""If a credential dict sneaks in, no password leaks into vals."""
|
||||||
|
ResUsers = self.env['res.users']
|
||||||
|
vals = ResUsers._fc_build_event_vals(
|
||||||
|
result='failure',
|
||||||
|
attempted_login='leak@example.com',
|
||||||
|
failure_reason='bad_password',
|
||||||
|
_credential={'login': 'leak@example.com',
|
||||||
|
'password': 'super-secret-pw',
|
||||||
|
'type': 'password'},
|
||||||
|
)
|
||||||
|
serialized = repr(vals)
|
||||||
|
self.assertNotIn('super-secret-pw', serialized)
|
||||||
|
self.assertEqual(vals['failure_reason'], 'bad_password')
|
||||||
|
|
||||||
|
def test_update_last_login_writes_audit_row(self):
|
||||||
|
"""Calling _update_last_login on a user creates a success row."""
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Audit Tester',
|
||||||
|
'login': 'audit-tester@example.com',
|
||||||
|
'password': 'audit-tester-pw-1',
|
||||||
|
})
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
before = Audit.search_count([('user_id', '=', user.id)])
|
||||||
|
user._update_last_login()
|
||||||
|
after = Audit.search_count([('user_id', '=', user.id)])
|
||||||
|
self.assertEqual(after, before + 1)
|
||||||
|
row = Audit.search([('user_id', '=', user.id)],
|
||||||
|
order='event_time desc', limit=1)
|
||||||
|
self.assertEqual(row.result, 'success')
|
||||||
|
self.assertEqual(row.attempted_login, user.login)
|
||||||
|
self.assertFalse(row.failure_reason)
|
||||||
|
self.assertEqual(row.database, self.env.cr.dbname)
|
||||||
|
|
||||||
|
def test_audit_write_failure_does_not_block_login(self):
|
||||||
|
"""An exception inside the audit write must not propagate."""
|
||||||
|
from unittest.mock import patch
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Resilient Tester',
|
||||||
|
'login': 'resilient-tester@example.com',
|
||||||
|
'password': 'resilient-tester-pw-1',
|
||||||
|
})
|
||||||
|
|
||||||
|
def boom(self_, vals):
|
||||||
|
raise RuntimeError('simulated audit DB failure')
|
||||||
|
|
||||||
|
with patch.object(type(self.env['fusion.login.audit']),
|
||||||
|
'create', boom):
|
||||||
|
# Must not raise.
|
||||||
|
user._update_last_login()
|
||||||
|
|
||||||
|
def test_bad_password_writes_failure_row(self):
|
||||||
|
"""A wrong password creates a result=failure row with failure_reason='bad_password'."""
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Wrongpw Tester',
|
||||||
|
'login': 'wrongpw-tester@example.com',
|
||||||
|
'password': 'wrongpw-tester-pw-1',
|
||||||
|
})
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
before = Audit.search_count([('attempted_login', '=', user.login),
|
||||||
|
('result', '=', 'failure')])
|
||||||
|
# NB: cannot use `self.assertRaises(AccessDenied)` — it opens an extra
|
||||||
|
# savepoint (see odoo/tests/common.py::_assertRaises) that rolls back
|
||||||
|
# the audit row written from inside the override.
|
||||||
|
raised = False
|
||||||
|
try:
|
||||||
|
user._check_credentials(
|
||||||
|
{'login': user.login, 'password': 'definitely-wrong',
|
||||||
|
'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
raised = True
|
||||||
|
self.assertTrue(raised, "AccessDenied not raised on wrong password")
|
||||||
|
after = Audit.search_count([('attempted_login', '=', user.login),
|
||||||
|
('result', '=', 'failure')])
|
||||||
|
self.assertEqual(after, before + 1)
|
||||||
|
row = Audit.search([('attempted_login', '=', user.login),
|
||||||
|
('result', '=', 'failure')],
|
||||||
|
order='event_time desc', limit=1)
|
||||||
|
self.assertEqual(row.failure_reason, 'bad_password')
|
||||||
|
self.assertEqual(row.user_id, user)
|
||||||
|
|
||||||
|
def test_bad_password_never_appears_in_row(self):
|
||||||
|
"""The attempted password string never lands in any field."""
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
secret = 'NeverInTheRow-9f3a82'
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Leak Test',
|
||||||
|
'login': 'leak-test-2@example.com',
|
||||||
|
'password': 'leak-test-pw-1',
|
||||||
|
})
|
||||||
|
# NB: manual try/except instead of assertRaises — see note above.
|
||||||
|
raised = False
|
||||||
|
try:
|
||||||
|
user._check_credentials(
|
||||||
|
{'login': user.login, 'password': secret, 'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
raised = True
|
||||||
|
self.assertTrue(raised, "AccessDenied not raised on wrong password")
|
||||||
|
row = self.env['fusion.login.audit'].sudo().search(
|
||||||
|
[('attempted_login', '=', user.login),
|
||||||
|
('result', '=', 'failure')],
|
||||||
|
order='event_time desc', limit=1)
|
||||||
|
self.assertTrue(row, "Audit row not created for bad-password attempt")
|
||||||
|
for fname in ('attempted_login', 'failure_reason', 'user_agent_raw',
|
||||||
|
'browser', 'os', 'ip_address', 'ip_hostname',
|
||||||
|
'city', 'country_name', 'country_code', 'geo_state',
|
||||||
|
'database'):
|
||||||
|
self.assertNotIn(secret, (row[fname] or ''),
|
||||||
|
f"Password leaked into field {fname}")
|
||||||
|
|
||||||
|
def test_unknown_user_writes_failure_row(self):
|
||||||
|
"""A login attempt for a username that does not exist gets logged
|
||||||
|
with user_id=NULL and failure_reason='unknown_user'."""
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
bogus = 'this-user-does-not-exist@example.com'
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
before = Audit.search_count([('attempted_login', '=', bogus)])
|
||||||
|
# NB: manual try/except instead of assertRaises — see comment in
|
||||||
|
# test_bad_password_writes_failure_row. _login is an instance method
|
||||||
|
# in Odoo 19 (not a classmethod as in earlier versions); we call it
|
||||||
|
# on the empty recordset of res.users, which matches what
|
||||||
|
# authenticate() does internally.
|
||||||
|
raised = False
|
||||||
|
try:
|
||||||
|
self.env['res.users']._login(
|
||||||
|
{'login': bogus, 'password': 'whatever',
|
||||||
|
'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
raised = True
|
||||||
|
self.assertTrue(raised, "AccessDenied must propagate after the audit write")
|
||||||
|
after = Audit.search_count([('attempted_login', '=', bogus)])
|
||||||
|
self.assertEqual(after, before + 1)
|
||||||
|
row = Audit.search([('attempted_login', '=', bogus)],
|
||||||
|
order='event_time desc', limit=1)
|
||||||
|
self.assertFalse(row.user_id)
|
||||||
|
self.assertEqual(row.failure_reason, 'unknown_user')
|
||||||
|
self.assertEqual(row.result, 'failure')
|
||||||
|
|
||||||
|
def test_login_known_user_bad_password_single_row(self):
|
||||||
|
"""When _login is the entry point for an existing user with the
|
||||||
|
wrong password, only ONE failure row is written (bad_password from
|
||||||
|
_check_credentials) — NOT two (bad_password + unknown_user). The
|
||||||
|
unknown_user branch must only fire when the login string does not
|
||||||
|
resolve to any user.
|
||||||
|
|
||||||
|
Regression test for the duplicate-row bug discovered during the
|
||||||
|
production deploy smoke on westin-v19: a single failed login for
|
||||||
|
an existing user was creating two audit rows.
|
||||||
|
"""
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'NoDupTester',
|
||||||
|
'login': 'nodup-tester@example.com',
|
||||||
|
'password': 'nodup-tester-pw-1',
|
||||||
|
})
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
before = Audit.search_count([('attempted_login', '=', user.login)])
|
||||||
|
raised = False
|
||||||
|
try:
|
||||||
|
self.env['res.users']._login(
|
||||||
|
{'login': user.login, 'password': 'wrong-not-the-real-one',
|
||||||
|
'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
raised = True
|
||||||
|
self.assertTrue(raised)
|
||||||
|
after = Audit.search_count([('attempted_login', '=', user.login)])
|
||||||
|
self.assertEqual(after - before, 1,
|
||||||
|
"Exactly one row per failed login attempt — not two")
|
||||||
|
row = Audit.search([('attempted_login', '=', user.login)],
|
||||||
|
order='event_time desc', limit=1)
|
||||||
|
self.assertEqual(row.failure_reason, 'bad_password',
|
||||||
|
"Existing-user failure must record bad_password, "
|
||||||
|
"not unknown_user (the user IS in the system)")
|
||||||
|
|
||||||
|
def test_computed_last_successful_login(self):
|
||||||
|
"""x_fc_last_successful_login reflects the latest success row."""
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Compute Tester',
|
||||||
|
'login': 'compute-tester@example.com',
|
||||||
|
'password': 'compute-tester-pw-1',
|
||||||
|
})
|
||||||
|
# Use registry cursor so the audit row survives the transactional
|
||||||
|
# boundary the way the auth-time path does.
|
||||||
|
with self.env.registry.cursor() as audit_cr:
|
||||||
|
from odoo import api
|
||||||
|
audit_env = api.Environment(audit_cr, self.env.uid, self.env.context)
|
||||||
|
audit_env['fusion.login.audit'].sudo().create({
|
||||||
|
'user_id': user.id,
|
||||||
|
'attempted_login': user.login,
|
||||||
|
'result': 'success',
|
||||||
|
'database': self.env.cr.dbname,
|
||||||
|
'ip_address': '198.51.100.42',
|
||||||
|
})
|
||||||
|
user.invalidate_recordset(['x_fc_last_successful_login',
|
||||||
|
'x_fc_login_audit_count',
|
||||||
|
'x_fc_last_login_ip'])
|
||||||
|
self.assertTrue(user.x_fc_last_successful_login)
|
||||||
|
self.assertGreaterEqual(user.x_fc_login_audit_count, 1)
|
||||||
|
self.assertEqual(user.x_fc_last_login_ip, '198.51.100.42')
|
||||||
|
|
||||||
|
def test_action_view_login_audit_returns_window_action(self):
|
||||||
|
"""The smart-button action returns an act_window scoped to this user."""
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Action Tester',
|
||||||
|
'login': 'action-tester@example.com',
|
||||||
|
'password': 'action-tester-pw-1',
|
||||||
|
})
|
||||||
|
action = user.action_fc_view_login_audit()
|
||||||
|
self.assertEqual(action['res_model'], 'fusion.login.audit')
|
||||||
|
self.assertEqual(action['type'], 'ir.actions.act_window')
|
||||||
|
# Domain must filter to this user
|
||||||
|
self.assertIn(('user_id', '=', user.id), action['domain'])
|
||||||
|
|
||||||
|
def test_settings_round_trip(self):
|
||||||
|
"""Writing settings persists them via ir.config_parameter."""
|
||||||
|
Settings = self.env['res.config.settings'].sudo()
|
||||||
|
Settings.create({
|
||||||
|
'x_fc_login_audit_retention_days': 90,
|
||||||
|
'x_fc_login_audit_alert_threshold': 3,
|
||||||
|
'x_fc_login_audit_alert_window_min': 5,
|
||||||
|
'x_fc_login_audit_alert_enabled': False,
|
||||||
|
}).execute()
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
self.assertEqual(ICP.get_param('fusion_login_audit.retention_days'), '90')
|
||||||
|
self.assertEqual(ICP.get_param('fusion_login_audit.alert_threshold'), '3')
|
||||||
|
self.assertEqual(ICP.get_param('fusion_login_audit.alert_window_min'), '5')
|
||||||
|
# Odoo's set_param deletes the row when the value is falsy, so a
|
||||||
|
# Boolean field set to False yields get_param() == False (Python
|
||||||
|
# bool, the default), not the string 'False'.
|
||||||
|
self.assertFalse(ICP.get_param('fusion_login_audit.alert_enabled'))
|
||||||
|
|
||||||
|
def test_failure_burst_queues_one_email(self):
|
||||||
|
"""N consecutive failures (within window) queue exactly one mail.mail."""
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
ICP.set_param('fusion_login_audit.alert_threshold', '3')
|
||||||
|
ICP.set_param('fusion_login_audit.alert_window_min', '15')
|
||||||
|
ICP.set_param('fusion_login_audit.alert_enabled', 'True')
|
||||||
|
# Clear any cooldown leftover from earlier tests.
|
||||||
|
ICP.set_param('fusion_login_audit.last_alert:burst@example.com', '')
|
||||||
|
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Burst Tester',
|
||||||
|
'login': 'burst@example.com',
|
||||||
|
'password': 'burst-tester-pw-1',
|
||||||
|
})
|
||||||
|
Mail = self.env['mail.mail'].sudo()
|
||||||
|
before = Mail.search_count([('subject', 'ilike', 'burst@example.com')])
|
||||||
|
for _i in range(3):
|
||||||
|
raised = False
|
||||||
|
try:
|
||||||
|
user._check_credentials(
|
||||||
|
{'login': user.login, 'password': 'wrong',
|
||||||
|
'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
raised = True
|
||||||
|
self.assertTrue(raised)
|
||||||
|
after = Mail.search_count([('subject', 'ilike', 'burst@example.com')])
|
||||||
|
self.assertEqual(after, before + 1,
|
||||||
|
"Exactly one alert mail should be queued")
|
||||||
|
|
||||||
|
def test_cooldown_suppresses_second_alert(self):
|
||||||
|
"""Failures beyond the threshold within the cooldown queue zero more emails."""
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
ICP.set_param('fusion_login_audit.alert_threshold', '3')
|
||||||
|
ICP.set_param('fusion_login_audit.alert_window_min', '15')
|
||||||
|
ICP.set_param('fusion_login_audit.alert_enabled', 'True')
|
||||||
|
ICP.set_param('fusion_login_audit.last_alert:cool@example.com', '')
|
||||||
|
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Cooldown Tester',
|
||||||
|
'login': 'cool@example.com',
|
||||||
|
'password': 'cooldown-tester-pw-1',
|
||||||
|
})
|
||||||
|
Mail = self.env['mail.mail'].sudo()
|
||||||
|
for _i in range(3):
|
||||||
|
try:
|
||||||
|
user._check_credentials(
|
||||||
|
{'login': user.login, 'password': 'wrong',
|
||||||
|
'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
pass
|
||||||
|
count_after_3 = Mail.search_count([('subject', 'ilike', 'cool@example.com')])
|
||||||
|
for _i in range(2):
|
||||||
|
try:
|
||||||
|
user._check_credentials(
|
||||||
|
{'login': user.login, 'password': 'wrong',
|
||||||
|
'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
pass
|
||||||
|
count_after_5 = Mail.search_count([('subject', 'ilike', 'cool@example.com')])
|
||||||
|
self.assertEqual(count_after_5, count_after_3,
|
||||||
|
"Cooldown should suppress additional emails")
|
||||||
|
|
||||||
|
def test_alert_disabled_master_switch(self):
|
||||||
|
"""alert_enabled=False suppresses all alerts regardless of threshold."""
|
||||||
|
from odoo.exceptions import AccessDenied
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
ICP.set_param('fusion_login_audit.alert_threshold', '1')
|
||||||
|
ICP.set_param('fusion_login_audit.alert_window_min', '15')
|
||||||
|
# Use the actual boolean field's storage semantics — see CLAUDE.md rule #5.
|
||||||
|
# Writing False through the settings form deletes the param; here we
|
||||||
|
# set the string 'False' explicitly to simulate "disabled".
|
||||||
|
ICP.set_param('fusion_login_audit.alert_enabled', 'False')
|
||||||
|
ICP.set_param('fusion_login_audit.last_alert:disabled@example.com', '')
|
||||||
|
|
||||||
|
user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Disabled Tester',
|
||||||
|
'login': 'disabled@example.com',
|
||||||
|
'password': 'disabled-tester-pw-1',
|
||||||
|
})
|
||||||
|
Mail = self.env['mail.mail'].sudo()
|
||||||
|
before = Mail.search_count([('subject', 'ilike', 'disabled@example.com')])
|
||||||
|
try:
|
||||||
|
user._check_credentials(
|
||||||
|
{'login': user.login, 'password': 'wrong',
|
||||||
|
'type': 'password'},
|
||||||
|
{'interactive': False},
|
||||||
|
)
|
||||||
|
except AccessDenied:
|
||||||
|
pass
|
||||||
|
after = Mail.search_count([('subject', 'ilike', 'disabled@example.com')])
|
||||||
|
self.assertEqual(after, before, "Disabled alerts should queue nothing")
|
||||||
|
|
||||||
|
def test_retention_gc_deletes_old_rows(self):
|
||||||
|
"""The GC method deletes rows older than retention_days."""
|
||||||
|
from datetime import timedelta
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
ICP.set_param('fusion_login_audit.retention_days', '30')
|
||||||
|
|
||||||
|
now = fields.Datetime.now()
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
old = Audit.create({
|
||||||
|
'attempted_login': 'gc-old@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'event_time': now - timedelta(days=45),
|
||||||
|
})
|
||||||
|
recent = Audit.create({
|
||||||
|
'attempted_login': 'gc-recent@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'event_time': now - timedelta(days=5),
|
||||||
|
})
|
||||||
|
old_id, recent_id = old.id, recent.id
|
||||||
|
|
||||||
|
Audit._fc_retention_gc()
|
||||||
|
|
||||||
|
self.assertFalse(Audit.browse(old_id).exists(),
|
||||||
|
"Row older than retention_days should be gone")
|
||||||
|
self.assertTrue(Audit.browse(recent_id).exists(),
|
||||||
|
"Row inside retention_days should survive")
|
||||||
|
|
||||||
|
def test_retention_zero_keeps_forever(self):
|
||||||
|
"""retention_days=0 keeps all rows."""
|
||||||
|
from datetime import timedelta
|
||||||
|
ICP = self.env['ir.config_parameter'].sudo()
|
||||||
|
ICP.set_param('fusion_login_audit.retention_days', '0')
|
||||||
|
|
||||||
|
now = fields.Datetime.now()
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
ancient = Audit.create({
|
||||||
|
'attempted_login': 'forever@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'event_time': now - timedelta(days=3650),
|
||||||
|
})
|
||||||
|
ancient_id = ancient.id
|
||||||
|
|
||||||
|
Audit._fc_retention_gc()
|
||||||
|
|
||||||
|
self.assertTrue(Audit.browse(ancient_id).exists(),
|
||||||
|
"retention_days=0 must keep everything")
|
||||||
|
|
||||||
|
def test_geo_private_ip_shortcut(self):
|
||||||
|
"""Private IPs short-circuit to state='private_ip' without HTTP."""
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
rec = Audit.create({
|
||||||
|
'attempted_login': 'lan@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'ip_address': '192.168.1.40',
|
||||||
|
'geo_lookup_state': 'pending',
|
||||||
|
})
|
||||||
|
Audit._fc_geo_enrich_pending(limit=10)
|
||||||
|
rec.invalidate_recordset()
|
||||||
|
self.assertEqual(rec.geo_lookup_state, 'private_ip')
|
||||||
|
self.assertEqual(rec.country_code, '--')
|
||||||
|
|
||||||
|
def test_geo_cache_hit_avoids_http(self):
|
||||||
|
"""A second row with the same recent IP copies from cache."""
|
||||||
|
from unittest.mock import patch
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
# Seed a "done" row from the same IP.
|
||||||
|
Audit.create({
|
||||||
|
'attempted_login': 'seed@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'ip_address': '203.0.113.99',
|
||||||
|
'geo_lookup_state': 'done',
|
||||||
|
'country_code': 'CA',
|
||||||
|
'country_name': 'Canada',
|
||||||
|
'city': 'Toronto',
|
||||||
|
'geo_state': 'Ontario',
|
||||||
|
})
|
||||||
|
target = Audit.create({
|
||||||
|
'attempted_login': 'hit@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'ip_address': '203.0.113.99',
|
||||||
|
'geo_lookup_state': 'pending',
|
||||||
|
})
|
||||||
|
|
||||||
|
with patch(
|
||||||
|
'odoo.addons.fusion_login_audit.models.fusion_login_audit.requests.get'
|
||||||
|
) as mock_get:
|
||||||
|
Audit._fc_geo_enrich_pending(limit=10)
|
||||||
|
mock_get.assert_not_called()
|
||||||
|
|
||||||
|
target.invalidate_recordset()
|
||||||
|
self.assertEqual(target.geo_lookup_state, 'done')
|
||||||
|
self.assertEqual(target.country_code, 'CA')
|
||||||
|
self.assertEqual(target.city, 'Toronto')
|
||||||
|
|
||||||
|
def test_geo_internal_skipped(self):
|
||||||
|
"""Rows with geo_lookup_state='internal' are not picked up."""
|
||||||
|
Audit = self.env['fusion.login.audit'].sudo()
|
||||||
|
rec = Audit.create({
|
||||||
|
'attempted_login': 'cron@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'ip_address': 'internal',
|
||||||
|
'geo_lookup_state': 'internal',
|
||||||
|
})
|
||||||
|
# Should be a no-op for 'internal' state (cron only picks 'pending').
|
||||||
|
Audit._fc_geo_enrich_pending(limit=10)
|
||||||
|
rec.invalidate_recordset()
|
||||||
|
self.assertEqual(rec.geo_lookup_state, 'internal')
|
||||||
129
fusion_login_audit/tests/test_security.py
Normal file
129
fusion_login_audit/tests/test_security.py
Normal file
@@ -0,0 +1,129 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
from odoo.exceptions import AccessError
|
||||||
|
from odoo.tests.common import TransactionCase, tagged
|
||||||
|
|
||||||
|
|
||||||
|
@tagged('post_install', '-at_install')
|
||||||
|
class TestFusionLoginAuditSecurity(TransactionCase):
|
||||||
|
"""Tests for the layered protection on `fusion.login.audit`:
|
||||||
|
|
||||||
|
Layer 1 — ACL (security/ir.model.access.csv): grants read-only access to
|
||||||
|
`base.group_system` and nothing to any other group. Blocks write/create/
|
||||||
|
unlink for everyone via the ORM regardless of `sudo()`.
|
||||||
|
|
||||||
|
Layer 2 — Record rule (security/security.xml): group-specific rule that
|
||||||
|
grants admins an unrestricted domain (`[(1,'=',1)]`). The rule does NOT
|
||||||
|
actively restrict non-admins — Odoo's semantics for a group-scoped rule
|
||||||
|
is "the rule only applies to users in that group". Non-admins are gated
|
||||||
|
purely by the ACL, which denies them everything. The rule's value is
|
||||||
|
documentation + future-proofing (it keeps admin access explicit if the
|
||||||
|
ACL is ever loosened with a per-group read row; the admin path remains
|
||||||
|
explicit and self-documenting). It is NOT a security gate the ACL relies on.
|
||||||
|
|
||||||
|
Test naming reflects which layer actually does the work:
|
||||||
|
- test_acl_blocks_* — exercises Layer 1 (ACL alone is sufficient).
|
||||||
|
- test_admin_can_read_through_acl_and_rule — exercises both layers in
|
||||||
|
the positive path (admin must satisfy ACL grant
|
||||||
|
AND the admin-scoped rule's domain).
|
||||||
|
"""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
super().setUp()
|
||||||
|
self.audit_row = self.env['fusion.login.audit'].sudo().create({
|
||||||
|
'attempted_login': 'sec-test@example.com',
|
||||||
|
'result': 'success',
|
||||||
|
'database': self.env.cr.dbname,
|
||||||
|
})
|
||||||
|
# Internal non-admin user (active employee, not a Settings admin).
|
||||||
|
self.regular_user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Regular Tester',
|
||||||
|
'login': 'regular-tester@example.com',
|
||||||
|
'password': 'regular-tester-pw-1',
|
||||||
|
'group_ids': [(6, 0, [self.env.ref('base.group_user').id])],
|
||||||
|
})
|
||||||
|
# Portal user (share=True) — must not see audit data either.
|
||||||
|
self.portal_user = self.env['res.users'].sudo().create({
|
||||||
|
'name': 'Portal Tester',
|
||||||
|
'login': 'portal-tester@example.com',
|
||||||
|
'password': 'portal-tester-pw-1',
|
||||||
|
'group_ids': [(6, 0, [self.env.ref('base.group_portal').id])],
|
||||||
|
})
|
||||||
|
|
||||||
|
def test_admin_can_read_through_acl_and_rule(self):
|
||||||
|
"""A Settings admin satisfies both the ACL (grants read) and the
|
||||||
|
record rule (admin-only domain), so the read succeeds."""
|
||||||
|
admin = self.env.ref('base.user_admin')
|
||||||
|
rec = self.audit_row.with_user(admin).read(['attempted_login'])
|
||||||
|
self.assertEqual(rec[0]['attempted_login'], 'sec-test@example.com')
|
||||||
|
|
||||||
|
def test_acl_blocks_read_for_regular_user(self):
|
||||||
|
"""A `base.group_user` member has no ACL grant on the model. The
|
||||||
|
ACL alone denies the read; the record rule never gets consulted."""
|
||||||
|
with self.assertRaises(AccessError):
|
||||||
|
self.audit_row.with_user(self.regular_user).read(['attempted_login'])
|
||||||
|
|
||||||
|
def test_acl_blocks_read_for_portal_user(self):
|
||||||
|
"""A `base.group_portal` (share=True) user has no ACL grant either.
|
||||||
|
Audit data must never leak to a portal user — IP and attempted_login
|
||||||
|
are sensitive."""
|
||||||
|
with self.assertRaises(AccessError):
|
||||||
|
self.audit_row.with_user(self.portal_user).read(['attempted_login'])
|
||||||
|
|
||||||
|
def test_acl_blocks_write_for_admin(self):
|
||||||
|
"""Even Settings admins cannot write — the ACL grants no group any
|
||||||
|
write permission on this model (audit log is append-only). The rule's
|
||||||
|
`perm_write=False` means 'rule does not constrain this op', so this
|
||||||
|
denial is the ACL's work alone."""
|
||||||
|
admin = self.env.ref('base.user_admin')
|
||||||
|
with self.assertRaises(AccessError):
|
||||||
|
self.audit_row.with_user(admin).write({'attempted_login': 'tampered'})
|
||||||
|
|
||||||
|
def test_acl_blocks_unlink_for_admin(self):
|
||||||
|
"""Append-only also at the unlink boundary. ACL grants no group
|
||||||
|
delete permission; the record rule's `perm_unlink=False` exempts
|
||||||
|
it from gating this op."""
|
||||||
|
admin = self.env.ref('base.user_admin')
|
||||||
|
with self.assertRaises(AccessError):
|
||||||
|
self.audit_row.with_user(admin).unlink()
|
||||||
|
|
||||||
|
# Note: a "rule actively blocks non-admins" test was attempted but
|
||||||
|
# removed once the actual Odoo semantics were verified. A group-scoped
|
||||||
|
# rule (groups=[base.group_system]) only applies to users in that group.
|
||||||
|
# Granting a base.group_user member an ACL read row would let them read
|
||||||
|
# rows — the rule does not filter them. To make the rule truly restrictive
|
||||||
|
# we would need a global rule (groups=[]) with domain [(0,'=',1)] paired
|
||||||
|
# with the admin grant. That is a security-model redesign and out of
|
||||||
|
# scope for T3. The ACL already provides the actual gate.
|
||||||
|
|
||||||
|
# ─────────────────────────────────────────────────────────────────────
|
||||||
|
# T14: view-level visibility checks. The smart button and the "Login
|
||||||
|
# Activity" tab on res.users are gated by groups="base.group_system"
|
||||||
|
# on the inner XML nodes (the inherited view record itself cannot
|
||||||
|
# carry groups — CLAUDE.md rule #11). Verify the gate works by asking
|
||||||
|
# for the form view as a non-admin and confirming the x_fc_* fields
|
||||||
|
# are stripped from the arch.
|
||||||
|
# ─────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
def test_view_hides_button_and_tab_for_non_admin(self):
|
||||||
|
"""A regular user's get_view() does not contain the x_fc_login_audit_*
|
||||||
|
fields — they live behind groups="base.group_system" XML attributes."""
|
||||||
|
view = self.env['res.users'].with_user(self.regular_user).get_view(
|
||||||
|
view_id=self.env.ref('base.view_users_form').id,
|
||||||
|
view_type='form',
|
||||||
|
)
|
||||||
|
arch = view['arch']
|
||||||
|
self.assertNotIn('x_fc_login_audit_count', arch,
|
||||||
|
"Smart-button field must not leak into non-admin view")
|
||||||
|
self.assertNotIn('x_fc_login_audit_ids', arch,
|
||||||
|
"Login Activity tab must not leak into non-admin view")
|
||||||
|
|
||||||
|
def test_view_shows_button_and_tab_for_admin(self):
|
||||||
|
"""A Settings admin DOES see both nodes."""
|
||||||
|
admin = self.env.ref('base.user_admin')
|
||||||
|
view = self.env['res.users'].with_user(admin).get_view(
|
||||||
|
view_id=self.env.ref('base.view_users_form').id,
|
||||||
|
view_type='form',
|
||||||
|
)
|
||||||
|
arch = view['arch']
|
||||||
|
self.assertIn('x_fc_login_audit_count', arch)
|
||||||
|
self.assertIn('x_fc_login_audit_ids', arch)
|
||||||
118
fusion_login_audit/views/fusion_login_audit_views.xml
Normal file
118
fusion_login_audit/views/fusion_login_audit_views.xml
Normal file
@@ -0,0 +1,118 @@
|
|||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<odoo>
|
||||||
|
|
||||||
|
<!-- List -->
|
||||||
|
<record id="view_fusion_login_audit_list" model="ir.ui.view">
|
||||||
|
<field name="name">fusion.login.audit.list</field>
|
||||||
|
<field name="model">fusion.login.audit</field>
|
||||||
|
<field name="arch" type="xml">
|
||||||
|
<list create="false" edit="false" delete="false"
|
||||||
|
default_order="event_time desc"
|
||||||
|
decoration-success="result=='success'"
|
||||||
|
decoration-danger="result=='failure'">
|
||||||
|
<field name="event_time"/>
|
||||||
|
<field name="user_id"/>
|
||||||
|
<field name="attempted_login"/>
|
||||||
|
<field name="result" widget="badge"/>
|
||||||
|
<field name="failure_reason"/>
|
||||||
|
<field name="ip_address"/>
|
||||||
|
<field name="country_code"/>
|
||||||
|
<field name="city"/>
|
||||||
|
<field name="browser"/>
|
||||||
|
<field name="device_type"/>
|
||||||
|
<field name="database" optional="hide"/>
|
||||||
|
</list>
|
||||||
|
</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
<!-- Form (readonly) -->
|
||||||
|
<record id="view_fusion_login_audit_form" model="ir.ui.view">
|
||||||
|
<field name="name">fusion.login.audit.form</field>
|
||||||
|
<field name="model">fusion.login.audit</field>
|
||||||
|
<field name="arch" type="xml">
|
||||||
|
<form create="false" edit="false" delete="false">
|
||||||
|
<sheet>
|
||||||
|
<group>
|
||||||
|
<group string="Event">
|
||||||
|
<field name="event_time" readonly="1"/>
|
||||||
|
<field name="result" readonly="1" widget="badge"/>
|
||||||
|
<field name="failure_reason" readonly="1"/>
|
||||||
|
<field name="user_id" readonly="1"/>
|
||||||
|
<field name="attempted_login" readonly="1"/>
|
||||||
|
<field name="database" readonly="1"/>
|
||||||
|
</group>
|
||||||
|
<group string="Source">
|
||||||
|
<field name="ip_address" readonly="1"/>
|
||||||
|
<field name="ip_hostname" readonly="1"/>
|
||||||
|
<field name="country_code" readonly="1"/>
|
||||||
|
<field name="country_name" readonly="1"/>
|
||||||
|
<field name="geo_state" readonly="1"/>
|
||||||
|
<field name="city" readonly="1"/>
|
||||||
|
<field name="geo_lookup_state" readonly="1"/>
|
||||||
|
</group>
|
||||||
|
</group>
|
||||||
|
<group string="Client">
|
||||||
|
<field name="device_type" readonly="1"/>
|
||||||
|
<field name="browser" readonly="1"/>
|
||||||
|
<field name="os" readonly="1"/>
|
||||||
|
<field name="user_agent_raw" readonly="1"/>
|
||||||
|
</group>
|
||||||
|
</sheet>
|
||||||
|
</form>
|
||||||
|
</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
<!-- Search -->
|
||||||
|
<record id="view_fusion_login_audit_search" model="ir.ui.view">
|
||||||
|
<field name="name">fusion.login.audit.search</field>
|
||||||
|
<field name="model">fusion.login.audit</field>
|
||||||
|
<field name="arch" type="xml">
|
||||||
|
<search>
|
||||||
|
<field name="attempted_login"/>
|
||||||
|
<field name="user_id"/>
|
||||||
|
<field name="ip_address"/>
|
||||||
|
<field name="country_code"/>
|
||||||
|
<filter name="filter_success" string="Successes"
|
||||||
|
domain="[('result','=','success')]"/>
|
||||||
|
<filter name="filter_failure" string="Failures"
|
||||||
|
domain="[('result','=','failure')]"/>
|
||||||
|
<separator/>
|
||||||
|
<filter name="filter_24h" string="Last 24 hours"
|
||||||
|
domain="[('event_time','>=', (context_today() - relativedelta(days=1)).strftime('%Y-%m-%d 00:00:00'))]"/>
|
||||||
|
<filter name="filter_7d" string="Last 7 days"
|
||||||
|
domain="[('event_time','>=', (context_today() - relativedelta(days=7)).strftime('%Y-%m-%d 00:00:00'))]"/>
|
||||||
|
<filter name="filter_30d" string="Last 30 days"
|
||||||
|
domain="[('event_time','>=', (context_today() - relativedelta(days=30)).strftime('%Y-%m-%d 00:00:00'))]"/>
|
||||||
|
<separator/>
|
||||||
|
<filter name="filter_unknown_user" string="Unknown users"
|
||||||
|
domain="[('user_id','=',False)]"/>
|
||||||
|
<group>
|
||||||
|
<filter name="group_user" string="User"
|
||||||
|
context="{'group_by': 'user_id'}"/>
|
||||||
|
<filter name="group_country" string="Country"
|
||||||
|
context="{'group_by': 'country_code'}"/>
|
||||||
|
<filter name="group_ip" string="IP"
|
||||||
|
context="{'group_by': 'ip_address'}"/>
|
||||||
|
</group>
|
||||||
|
</search>
|
||||||
|
</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
<!-- Window actions -->
|
||||||
|
<record id="action_fusion_login_audit_all" model="ir.actions.act_window">
|
||||||
|
<field name="name">Login Events</field>
|
||||||
|
<field name="res_model">fusion.login.audit</field>
|
||||||
|
<field name="view_mode">list,form</field>
|
||||||
|
<field name="search_view_id" ref="view_fusion_login_audit_search"/>
|
||||||
|
<field name="context">{}</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
<record id="action_fusion_login_audit_failures_24h" model="ir.actions.act_window">
|
||||||
|
<field name="name">Failed Logins (24h)</field>
|
||||||
|
<field name="res_model">fusion.login.audit</field>
|
||||||
|
<field name="view_mode">list,form</field>
|
||||||
|
<field name="search_view_id" ref="view_fusion_login_audit_search"/>
|
||||||
|
<field name="context">{'search_default_filter_failure': 1, 'search_default_filter_24h': 1}</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
</odoo>
|
||||||
24
fusion_login_audit/views/menus.xml
Normal file
24
fusion_login_audit/views/menus.xml
Normal file
@@ -0,0 +1,24 @@
|
|||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<odoo>
|
||||||
|
|
||||||
|
<menuitem id="menu_fusion_login_audit_root"
|
||||||
|
name="Login Audit"
|
||||||
|
parent="base.menu_administration"
|
||||||
|
groups="base.group_system"
|
||||||
|
sequence="100"/>
|
||||||
|
|
||||||
|
<menuitem id="menu_fusion_login_audit_all"
|
||||||
|
name="Login Events"
|
||||||
|
parent="menu_fusion_login_audit_root"
|
||||||
|
action="action_fusion_login_audit_all"
|
||||||
|
groups="base.group_system"
|
||||||
|
sequence="10"/>
|
||||||
|
|
||||||
|
<menuitem id="menu_fusion_login_audit_failures"
|
||||||
|
name="Failed Logins (24h)"
|
||||||
|
parent="menu_fusion_login_audit_root"
|
||||||
|
action="action_fusion_login_audit_failures_24h"
|
||||||
|
groups="base.group_system"
|
||||||
|
sequence="20"/>
|
||||||
|
|
||||||
|
</odoo>
|
||||||
36
fusion_login_audit/views/res_config_settings_views.xml
Normal file
36
fusion_login_audit/views/res_config_settings_views.xml
Normal file
@@ -0,0 +1,36 @@
|
|||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<odoo>
|
||||||
|
|
||||||
|
<record id="view_res_config_settings_form_login_audit" model="ir.ui.view">
|
||||||
|
<field name="name">res.config.settings.form.login.audit</field>
|
||||||
|
<field name="model">res.config.settings</field>
|
||||||
|
<field name="inherit_id" ref="base_setup.res_config_settings_view_form"/>
|
||||||
|
<field name="arch" type="xml">
|
||||||
|
<xpath expr="//block[@id='user_default_rights']" position="after">
|
||||||
|
<block title="Login Audit"
|
||||||
|
name="login_audit_block"
|
||||||
|
groups="base.group_system">
|
||||||
|
<setting id="login_audit_retention"
|
||||||
|
string="Retention (days)"
|
||||||
|
help="0 = keep forever">
|
||||||
|
<field name="x_fc_login_audit_retention_days"/>
|
||||||
|
</setting>
|
||||||
|
<setting id="login_audit_alert_enabled"
|
||||||
|
string="Send failed-login alerts"
|
||||||
|
help="Email Settings admins when consecutive failures cross the threshold">
|
||||||
|
<field name="x_fc_login_audit_alert_enabled"/>
|
||||||
|
</setting>
|
||||||
|
<setting id="login_audit_alert_threshold"
|
||||||
|
string="Alert threshold (failures)">
|
||||||
|
<field name="x_fc_login_audit_alert_threshold"/>
|
||||||
|
</setting>
|
||||||
|
<setting id="login_audit_alert_window"
|
||||||
|
string="Alert window (minutes)">
|
||||||
|
<field name="x_fc_login_audit_alert_window_min"/>
|
||||||
|
</setting>
|
||||||
|
</block>
|
||||||
|
</xpath>
|
||||||
|
</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
</odoo>
|
||||||
56
fusion_login_audit/views/res_users_views.xml
Normal file
56
fusion_login_audit/views/res_users_views.xml
Normal file
@@ -0,0 +1,56 @@
|
|||||||
|
<?xml version="1.0" encoding="utf-8"?>
|
||||||
|
<odoo>
|
||||||
|
|
||||||
|
<record id="view_users_form_inherit_login_audit" model="ir.ui.view">
|
||||||
|
<field name="name">res.users.form.inherit.fusion_login_audit</field>
|
||||||
|
<field name="model">res.users</field>
|
||||||
|
<field name="inherit_id" ref="base.view_users_form"/>
|
||||||
|
<!-- Odoo 19: groups MUST be on the inherited XML nodes (button + page
|
||||||
|
below), NOT on the ir.ui.view record itself. Setting `group_ids`
|
||||||
|
on the record raises ParseError "Inherited view cannot have
|
||||||
|
'groups' defined on the record. Use 'groups' attributes inside
|
||||||
|
the view definition". -->
|
||||||
|
<field name="arch" type="xml">
|
||||||
|
<!-- Smart button -->
|
||||||
|
<xpath expr="//div[@name='button_box']" position="inside">
|
||||||
|
<button name="action_fc_view_login_audit"
|
||||||
|
type="object"
|
||||||
|
class="oe_stat_button"
|
||||||
|
icon="fa-key"
|
||||||
|
groups="base.group_system">
|
||||||
|
<field name="x_fc_login_audit_count" widget="statinfo"
|
||||||
|
string="Logins"/>
|
||||||
|
</button>
|
||||||
|
</xpath>
|
||||||
|
|
||||||
|
<!-- Login Activity tab appended at the end of the notebook -->
|
||||||
|
<xpath expr="//notebook" position="inside">
|
||||||
|
<page string="Login Activity"
|
||||||
|
name="fc_login_activity"
|
||||||
|
groups="base.group_system">
|
||||||
|
<group>
|
||||||
|
<field name="x_fc_last_successful_login" readonly="1"/>
|
||||||
|
<field name="x_fc_last_login_ip" readonly="1"/>
|
||||||
|
</group>
|
||||||
|
<field name="x_fc_login_audit_ids" readonly="1"
|
||||||
|
context="{'create': False, 'edit': False, 'delete': False}">
|
||||||
|
<list create="false" edit="false" delete="false"
|
||||||
|
limit="30" default_order="event_time desc">
|
||||||
|
<field name="event_time"/>
|
||||||
|
<field name="result" decoration-success="result=='success'"
|
||||||
|
decoration-danger="result=='failure'"
|
||||||
|
widget="badge"/>
|
||||||
|
<field name="failure_reason"/>
|
||||||
|
<field name="ip_address"/>
|
||||||
|
<field name="country_code"/>
|
||||||
|
<field name="city"/>
|
||||||
|
<field name="browser"/>
|
||||||
|
<field name="os"/>
|
||||||
|
</list>
|
||||||
|
</field>
|
||||||
|
</page>
|
||||||
|
</xpath>
|
||||||
|
</field>
|
||||||
|
</record>
|
||||||
|
|
||||||
|
</odoo>
|
||||||
Reference in New Issue
Block a user