MOHAN
b56435e258
API keys can now be locked to a single Odoo connection at creation time. A scoped key's tool calls silently default to that connection when none is specified, and are rejected outright if the caller explicitly requests a different one. Unscoped keys (including the legacy key) keep working across all connections as before. New instance_name column on api_keys, added via a runtime migration so it applies safely to already-deployed databases, not just fresh installs. Also fixes a significant pre-existing bug found while testing this: get_user_by_api_key (used for both REST and MCP auth) only ever checked the legacy single-key column — it never checked the api_keys table at all, meaning every named key created via POST /api/keys was completely unusable for actual authentication. You could create/list/revoke them, but never log in with one. Named keys now authenticate correctly everywhere. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
385 lines
15 KiB
Python
385 lines
15 KiB
Python
"""Auth business logic — signup, login, API key management, credential CRUD."""
|
|
from __future__ import annotations
|
|
from typing import Optional
|
|
from .models import User, OdooCredential, ApiKeyInfo
|
|
from .crypto import (
|
|
hash_password, verify_password,
|
|
generate_api_key, verify_api_key,
|
|
encrypt_credential, decrypt_credential,
|
|
generate_reset_token, hash_reset_token,
|
|
)
|
|
from ..database import db_execute, db_commit, db_fetchone, db_fetchall
|
|
|
|
RESET_TOKEN_TTL_MINUTES = 30
|
|
|
|
|
|
class AuthError(Exception):
|
|
pass
|
|
|
|
|
|
# ─── OdooInstanceConfig re-export for connection layer ────────────────────────
|
|
|
|
class OdooInstanceConfig:
|
|
def __init__(self, name: str, url: str, db: str, username: str, credential: str):
|
|
self.name = name
|
|
self.url = url
|
|
self.db = db
|
|
self.username = username
|
|
self._credential = credential
|
|
|
|
@property
|
|
def credential(self) -> str:
|
|
return self._credential
|
|
|
|
|
|
# ─── Users ────────────────────────────────────────────────────────────────────
|
|
|
|
def signup(email: str, password: str) -> tuple[User, str]:
|
|
"""Create a new user. Returns (user, full_api_key). API key shown only once."""
|
|
email = email.strip().lower()
|
|
if not email or not password:
|
|
raise AuthError("Email and password are required.")
|
|
if len(password) < 8:
|
|
raise AuthError("Password must be at least 8 characters.")
|
|
|
|
existing = db_fetchone("SELECT id FROM users WHERE email = ?", (email,))
|
|
if existing:
|
|
raise AuthError("An account with this email already exists.")
|
|
|
|
pw_hash = hash_password(password)
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
|
|
db_execute(
|
|
"INSERT INTO users (email, password_hash, api_key_hash, api_key_prefix) VALUES (?, ?, ?, ?)",
|
|
(email, pw_hash, key_hash, key_prefix),
|
|
)
|
|
db_commit()
|
|
|
|
row = db_fetchone("SELECT * FROM users WHERE email = ?", (email,))
|
|
user = _row_to_user(row)
|
|
return user, full_key
|
|
|
|
|
|
def login(email: str, password: str) -> User:
|
|
"""Verify email + password. Returns the user or raises AuthError."""
|
|
email = email.strip().lower()
|
|
row = db_fetchone("SELECT * FROM users WHERE email = ?", (email,))
|
|
if not row or not verify_password(password, row["password_hash"]):
|
|
raise AuthError("Invalid email or password.")
|
|
if not row["is_active"]:
|
|
raise AuthError("Account is disabled.")
|
|
return _row_to_user(row)
|
|
|
|
|
|
def get_user_by_api_key(api_key: str) -> Optional[User]:
|
|
"""Look up a user by their API key. Returns None if not found/invalid.
|
|
|
|
Checks two places: the legacy users.api_key_hash column (kept for
|
|
backward compatibility), and the api_keys table for named/scoped keys
|
|
created via POST /api/keys — a named key is a real credential and must
|
|
authenticate here too, not just show up in the Tokens list.
|
|
|
|
If the legacy key matches but has since been backfilled into api_keys
|
|
(via list_api_keys) and explicitly revoked there, it's rejected —
|
|
otherwise "revoking" the legacy key in the UI would silently do nothing.
|
|
"""
|
|
if not api_key or not api_key.startswith("mtom_"):
|
|
return None
|
|
prefix = api_key[:12]
|
|
|
|
rows = db_fetchall("SELECT * FROM users WHERE api_key_prefix = ? AND is_active = 1", (prefix,))
|
|
for row in rows:
|
|
if verify_api_key(api_key, row["api_key_hash"]):
|
|
revoked = db_fetchone(
|
|
"SELECT 1 FROM api_keys WHERE user_id = ? AND key_hash = ? AND is_active = 0",
|
|
(row["id"], row["api_key_hash"]),
|
|
)
|
|
if revoked:
|
|
return None
|
|
return _row_to_user(row)
|
|
|
|
key_rows = db_fetchall(
|
|
"SELECT * FROM api_keys WHERE key_prefix = ? AND is_active = 1 AND revoked_at IS NULL",
|
|
(prefix,),
|
|
)
|
|
for key_row in key_rows:
|
|
if verify_api_key(api_key, key_row["key_hash"]):
|
|
user_row = db_fetchone("SELECT * FROM users WHERE id = ? AND is_active = 1", (key_row["user_id"],))
|
|
if user_row:
|
|
return _row_to_user(user_row)
|
|
return None
|
|
|
|
|
|
def regenerate_api_key(user_id: int) -> str:
|
|
"""Generate a new API key for the user. Returns the full key (shown once).
|
|
|
|
Also syncs the backfilled 'Default (legacy)' row in api_keys (see
|
|
list_api_keys) if one exists, so the Tokens page doesn't show a stale
|
|
prefix/hash for a key that no longer works after rotation."""
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
db_execute(
|
|
"UPDATE users SET api_key_hash = ?, api_key_prefix = ? WHERE id = ?",
|
|
(key_hash, key_prefix, user_id),
|
|
)
|
|
db_execute(
|
|
"UPDATE api_keys SET key_hash = ?, key_prefix = ? WHERE user_id = ? AND name = 'Default (legacy)'",
|
|
(key_hash, key_prefix, user_id),
|
|
)
|
|
db_commit()
|
|
return full_key
|
|
|
|
|
|
def get_user_by_id(user_id: int) -> Optional[User]:
|
|
row = db_fetchone("SELECT * FROM users WHERE id = ?", (user_id,))
|
|
return _row_to_user(row) if row else None
|
|
|
|
|
|
# ─── Admin-only user management ────────────────────────────────────────────
|
|
|
|
def list_all_users() -> list[User]:
|
|
"""Admin-only: list every user, no ownership scoping."""
|
|
rows = db_fetchall("SELECT * FROM users ORDER BY created_at DESC")
|
|
return [_row_to_user(r) for r in rows]
|
|
|
|
|
|
def set_user_active(user_id: int, is_active: bool) -> bool:
|
|
"""Admin-only enable/disable. Takes effect immediately for new
|
|
get_user_by_api_key() lookups and new /api/login attempts (both already
|
|
check is_active) — no changes needed to that existing logic."""
|
|
cur = db_execute("UPDATE users SET is_active = ? WHERE id = ?", (int(is_active), user_id))
|
|
db_commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def delete_user(user_id: int) -> bool:
|
|
"""Admin-only permanent delete. Cascading cleanup of odoo_credentials,
|
|
api_keys, password_reset_tokens, tool_call_events is already handled by
|
|
the ON DELETE CASCADE / SET NULL foreign keys declared in database.py."""
|
|
cur = db_execute("DELETE FROM users WHERE id = ?", (user_id,))
|
|
db_commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
# ─── Multiple API Keys ──────────────────────────────────────────────────────
|
|
|
|
def create_api_key(user_id: int, name: str, instance_name: Optional[str] = None) -> tuple[int, str]:
|
|
"""Create a new named API key. If instance_name is set, this key is
|
|
scoped to that one Odoo connection only — see server.py's call_tool
|
|
wrapper for enforcement. Returns (key_id, full_key) — full_key shown once."""
|
|
full_key, key_hash, key_prefix = generate_api_key()
|
|
cur = db_execute(
|
|
"INSERT INTO api_keys (user_id, name, key_hash, key_prefix, instance_name) VALUES (?, ?, ?, ?, ?)",
|
|
(user_id, name, key_hash, key_prefix, instance_name),
|
|
)
|
|
db_commit()
|
|
return cur.lastrowid, full_key
|
|
|
|
|
|
def list_api_keys(user_id: int) -> list[ApiKeyInfo]:
|
|
"""List all named API keys for a user. Lazily backfills a 'Default (legacy)'
|
|
row mirroring users.api_key_hash the first time this is called, so the
|
|
legacy signup key shows up in the UI like any other key."""
|
|
rows = db_fetchall(
|
|
"SELECT * FROM api_keys WHERE user_id = ? ORDER BY created_at", (user_id,)
|
|
)
|
|
if not rows:
|
|
user_row = db_fetchone("SELECT api_key_hash, api_key_prefix FROM users WHERE id = ?", (user_id,))
|
|
if user_row and user_row["api_key_hash"]:
|
|
db_execute(
|
|
"INSERT INTO api_keys (user_id, name, key_hash, key_prefix) VALUES (?, ?, ?, ?)",
|
|
(user_id, "Default (legacy)", user_row["api_key_hash"], user_row["api_key_prefix"]),
|
|
)
|
|
db_commit()
|
|
rows = db_fetchall(
|
|
"SELECT * FROM api_keys WHERE user_id = ? ORDER BY created_at", (user_id,)
|
|
)
|
|
return [_row_to_api_key(r) for r in rows]
|
|
|
|
|
|
def revoke_api_key(user_id: int, key_id: int) -> bool:
|
|
"""Revoke a named API key. Returns False if not found/not owned by user."""
|
|
cur = db_execute(
|
|
"""UPDATE api_keys SET revoked_at = datetime('now'), is_active = 0
|
|
WHERE id = ? AND user_id = ? AND is_active = 1""",
|
|
(key_id, user_id),
|
|
)
|
|
db_commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def get_api_key_scope_for_mcp(api_key: str) -> tuple[Optional[int], Optional[str]]:
|
|
"""Resolve which api_keys row (if any) this raw key belongs to, plus its
|
|
instance scope (None = unscoped, can access any of the user's Odoo
|
|
connections). Used at MCP-session-start for analytics attribution
|
|
(api_key_id) and per-call instance enforcement (instance_name) — see
|
|
server.py's call_tool wrapper. Returns (None, None) only if the key has
|
|
never been backfilled into api_keys at all (harmless — analytics rows
|
|
for that session carry api_key_id=NULL until the user's first /api/keys
|
|
view triggers the backfill). The caller already validated the key via
|
|
get_user_by_api_key before calling this."""
|
|
if not api_key or not api_key.startswith("mtom_"):
|
|
return None, None
|
|
prefix = api_key[:12]
|
|
rows = db_fetchall(
|
|
"SELECT id, key_hash, instance_name FROM api_keys WHERE key_prefix = ? AND is_active = 1 AND revoked_at IS NULL",
|
|
(prefix,),
|
|
)
|
|
for row in rows:
|
|
if verify_api_key(api_key, row["key_hash"]):
|
|
db_execute("UPDATE api_keys SET last_used_at = datetime('now') WHERE id = ?", (row["id"],))
|
|
db_commit()
|
|
return row["id"], row["instance_name"]
|
|
return None, None
|
|
|
|
|
|
def _row_to_api_key(row) -> ApiKeyInfo:
|
|
return ApiKeyInfo(
|
|
id=row["id"],
|
|
name=row["name"],
|
|
key_prefix=row["key_prefix"],
|
|
created_at=row["created_at"],
|
|
last_used_at=row["last_used_at"],
|
|
revoked_at=row["revoked_at"],
|
|
is_active=bool(row["is_active"]),
|
|
instance_name=row["instance_name"],
|
|
)
|
|
|
|
|
|
# ─── Password Reset ─────────────────────────────────────────────────────────
|
|
|
|
def request_password_reset(email: str) -> Optional[str]:
|
|
"""Look up user by email. If found, create a reset token (expires in
|
|
RESET_TOKEN_TTL_MINUTES) and return the raw token for the caller to email
|
|
or log. Returns None if the email isn't found — callers must respond
|
|
identically either way to avoid leaking which emails are registered."""
|
|
email = email.strip().lower()
|
|
row = db_fetchone("SELECT id FROM users WHERE email = ?", (email,))
|
|
if not row:
|
|
return None
|
|
raw_token, token_hash = generate_reset_token()
|
|
db_execute(
|
|
"""INSERT INTO password_reset_tokens (user_id, token_hash, expires_at)
|
|
VALUES (?, ?, datetime('now', ?))""",
|
|
(row["id"], token_hash, f"+{RESET_TOKEN_TTL_MINUTES} minutes"),
|
|
)
|
|
db_commit()
|
|
return raw_token
|
|
|
|
|
|
def reset_password(raw_token: str, new_password: str) -> None:
|
|
"""Verify the reset token and update the user's password. Raises AuthError
|
|
if the token is missing/expired/already used. On success, also revokes
|
|
all active API keys for that user as a post-reset security measure."""
|
|
if len(new_password) < 8:
|
|
raise AuthError("Password must be at least 8 characters.")
|
|
token_hash = hash_reset_token(raw_token)
|
|
row = db_fetchone(
|
|
"""SELECT * FROM password_reset_tokens
|
|
WHERE token_hash = ? AND used_at IS NULL AND expires_at > datetime('now')""",
|
|
(token_hash,),
|
|
)
|
|
if not row:
|
|
raise AuthError("Invalid or expired reset link.")
|
|
|
|
user_id = row["user_id"]
|
|
pw_hash = hash_password(new_password)
|
|
db_execute("UPDATE users SET password_hash = ? WHERE id = ?", (pw_hash, user_id))
|
|
db_execute("UPDATE password_reset_tokens SET used_at = datetime('now') WHERE id = ?", (row["id"],))
|
|
db_execute(
|
|
"UPDATE api_keys SET revoked_at = datetime('now'), is_active = 0 WHERE user_id = ? AND is_active = 1",
|
|
(user_id,),
|
|
)
|
|
db_commit()
|
|
|
|
|
|
def _row_to_user(row) -> User:
|
|
return User(
|
|
id=row["id"],
|
|
email=row["email"],
|
|
api_key_prefix=row["api_key_prefix"],
|
|
is_active=bool(row["is_active"]),
|
|
created_at=row["created_at"],
|
|
)
|
|
|
|
|
|
# ─── Odoo Credentials ─────────────────────────────────────────────────────────
|
|
|
|
def upsert_odoo_credential(
|
|
user_id: int,
|
|
instance_name: str,
|
|
url: str,
|
|
database_name: str,
|
|
odoo_username: str,
|
|
credential: str,
|
|
is_api_key: bool = True,
|
|
) -> OdooCredential:
|
|
"""Add or replace an Odoo instance credential for a user."""
|
|
enc = encrypt_credential(credential)
|
|
existing = db_fetchone(
|
|
"SELECT id FROM odoo_credentials WHERE user_id = ? AND instance_name = ?",
|
|
(user_id, instance_name),
|
|
)
|
|
if existing:
|
|
db_execute(
|
|
"""UPDATE odoo_credentials
|
|
SET url=?, database_name=?, odoo_username=?, credential_enc=?, is_api_key=?
|
|
WHERE user_id=? AND instance_name=?""",
|
|
(url, database_name, odoo_username, enc, int(is_api_key), user_id, instance_name),
|
|
)
|
|
else:
|
|
db_execute(
|
|
"""INSERT INTO odoo_credentials
|
|
(user_id, instance_name, url, database_name, odoo_username, credential_enc, is_api_key)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)""",
|
|
(user_id, instance_name, url, database_name, odoo_username, enc, int(is_api_key)),
|
|
)
|
|
db_commit()
|
|
row = db_fetchone(
|
|
"SELECT * FROM odoo_credentials WHERE user_id=? AND instance_name=?",
|
|
(user_id, instance_name),
|
|
)
|
|
return _row_to_cred(row)
|
|
|
|
|
|
def list_odoo_credentials(user_id: int) -> list[OdooCredential]:
|
|
rows = db_fetchall("SELECT * FROM odoo_credentials WHERE user_id=? ORDER BY instance_name", (user_id,))
|
|
return [_row_to_cred(r) for r in rows]
|
|
|
|
|
|
def delete_odoo_credential(user_id: int, instance_name: str) -> bool:
|
|
cur = db_execute(
|
|
"DELETE FROM odoo_credentials WHERE user_id=? AND instance_name=?",
|
|
(user_id, instance_name),
|
|
)
|
|
db_commit()
|
|
return cur.rowcount > 0
|
|
|
|
|
|
def get_odoo_instance_configs(user_id: int) -> list[OdooInstanceConfig]:
|
|
"""Return decrypted Odoo configs for this user, ready for ConnectionManager."""
|
|
creds = list_odoo_credentials(user_id)
|
|
return [
|
|
OdooInstanceConfig(
|
|
name=c.instance_name,
|
|
url=c.url,
|
|
db=c.database_name,
|
|
username=c.odoo_username,
|
|
credential=decrypt_credential(c.credential_enc),
|
|
)
|
|
for c in creds
|
|
]
|
|
|
|
|
|
def _row_to_cred(row) -> OdooCredential:
|
|
return OdooCredential(
|
|
id=row["id"],
|
|
user_id=row["user_id"],
|
|
instance_name=row["instance_name"],
|
|
url=row["url"],
|
|
database_name=row["database_name"],
|
|
odoo_username=row["odoo_username"],
|
|
credential_enc=row["credential_enc"],
|
|
is_api_key=bool(row["is_api_key"]),
|
|
created_at=row["created_at"],
|
|
)
|