Agent Zero
437 lines
16 KiB
Python
437 lines
16 KiB
Python
"""Authentication service — login, logout, password reset, session management."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
import redis.asyncio as aioredis
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import get_settings
|
|
from app.core.audit import log_audit
|
|
from app.core.auth import (
|
|
create_session,
|
|
get_redis,
|
|
get_session_data,
|
|
hash_password,
|
|
hash_token,
|
|
invalidate_session,
|
|
update_session_tenant,
|
|
verify_password,
|
|
)
|
|
from app.core.hooks import do_action, apply_filters
|
|
from app.models.auth import PasswordResetToken
|
|
from app.models.tenant import Tenant
|
|
from app.models.user import User, UserTenant
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AuthService:
|
|
"""Handles authentication operations."""
|
|
|
|
async def login(
|
|
self,
|
|
db: AsyncSession,
|
|
redis: aioredis.Redis,
|
|
email: str,
|
|
password: str,
|
|
tenant_slug: str | None = None,
|
|
) -> tuple[str, str, User, Tenant, str] | None:
|
|
"""Authenticate user and create session.
|
|
Returns (session_id, csrf_token, user, tenant, role) or None.
|
|
|
|
Email is globally unique so we can safely use scalar_one_or_none().
|
|
The tenant is resolved from UserTenant via tenant_slug or the
|
|
user's default tenant membership.
|
|
"""
|
|
# Hook: auth.before_login — filter can modify email
|
|
email = await apply_filters("auth.before_login", email, db=db, password=password, tenant_slug=tenant_slug)
|
|
|
|
# Find user by email (globally unique now)
|
|
q = select(User).where(User.email == email, User.is_active == True) # noqa: E712
|
|
result = await db.execute(q)
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
return None
|
|
|
|
if not verify_password(password, user.password_hash):
|
|
return None
|
|
|
|
# Get user's default tenant or the one matching slug
|
|
ut_q = select(UserTenant).where(
|
|
UserTenant.user_id == user.id,
|
|
UserTenant.status == "active",
|
|
)
|
|
if tenant_slug:
|
|
ut_q = ut_q.join(Tenant, UserTenant.tenant_id == Tenant.id).where(
|
|
Tenant.slug == tenant_slug
|
|
)
|
|
else:
|
|
ut_q = ut_q.where(UserTenant.is_default == True) # noqa: E712
|
|
ut_result = await db.execute(ut_q)
|
|
user_tenant = ut_result.scalar_one_or_none()
|
|
|
|
# No fallback — if tenant_slug was provided, the membership must exist
|
|
# and be active in exactly that tenant. If no slug, the default membership
|
|
# must exist and be active.
|
|
if user_tenant is None:
|
|
if tenant_slug:
|
|
# Specific tenant requested but no active membership — fail
|
|
return None
|
|
# No default membership — check if there are multiple active memberships
|
|
ut_q2 = select(UserTenant).where(
|
|
UserTenant.user_id == user.id,
|
|
UserTenant.status == "active",
|
|
)
|
|
ut_result2 = await db.execute(ut_q2)
|
|
active_memberships = ut_result2.scalars().all()
|
|
if len(active_memberships) == 1:
|
|
# Exactly one active membership — use it
|
|
user_tenant = active_memberships[0]
|
|
elif len(active_memberships) == 0:
|
|
return None
|
|
else:
|
|
# Multiple active memberships without a default — must specify tenant_slug
|
|
return None
|
|
|
|
tenant_q = select(Tenant).where(Tenant.id == user_tenant.tenant_id)
|
|
tenant_result = await db.execute(tenant_q)
|
|
tenant = tenant_result.scalar_one_or_none()
|
|
if tenant is None:
|
|
return None
|
|
|
|
session_id, csrf_token = await create_session(
|
|
db, redis, user, tenant.id, role=user_tenant.role
|
|
)
|
|
|
|
# Log the login in audit trail via separate API session (crm_api with tenant context)
|
|
# crm_auth must not write to tenant tables — audit_log is a tenant table
|
|
try:
|
|
from app.core.db import get_session_factory, set_tenant_context
|
|
api_factory = get_session_factory()
|
|
async with api_factory() as audit_db:
|
|
await set_tenant_context(audit_db, tenant.id)
|
|
await log_audit(
|
|
audit_db,
|
|
tenant.id,
|
|
user.id,
|
|
"login",
|
|
"user",
|
|
user.id,
|
|
changes={"email": email},
|
|
)
|
|
await audit_db.commit()
|
|
except Exception:
|
|
logger.warning("Failed to write login audit log via API session", exc_info=True)
|
|
|
|
# Hook: auth.after_login
|
|
await do_action("auth.after_login", db=db, user=user, tenant=tenant, role=user_tenant.role, session_id=session_id)
|
|
|
|
return session_id, csrf_token, user, tenant, user_tenant.role
|
|
|
|
async def logout(self, redis: aioredis.Redis, session_id: str) -> bool:
|
|
"""Invalidate a session."""
|
|
await invalidate_session(redis, session_id)
|
|
return True
|
|
|
|
async def get_current_user_info(
|
|
self,
|
|
db: AsyncSession,
|
|
redis: aioredis.Redis,
|
|
session_id: str,
|
|
) -> dict[str, Any] | None:
|
|
"""Get current user info from session."""
|
|
session_data = await get_session_data(redis, session_id)
|
|
if session_data is None:
|
|
return None
|
|
|
|
# Fetch tenant name
|
|
tenant_q = select(Tenant).where(Tenant.id == uuid.UUID(session_data["tenant_id"]))
|
|
tenant_result = await db.execute(tenant_q)
|
|
tenant = tenant_result.scalar_one_or_none()
|
|
|
|
return {
|
|
"user_id": session_data["user_id"],
|
|
"email": session_data["email"],
|
|
"name": session_data["name"],
|
|
"role": session_data["role"],
|
|
"is_system_admin": session_data.get("is_system_admin", False),
|
|
"tenant_id": session_data["tenant_id"],
|
|
"tenant_name": tenant.name if tenant else None,
|
|
}
|
|
|
|
async def switch_tenant(
|
|
self,
|
|
db: AsyncSession,
|
|
redis: aioredis.Redis,
|
|
session_id: str,
|
|
new_tenant_id: uuid.UUID,
|
|
) -> dict[str, Any] | None:
|
|
"""Switch the active tenant for the current session."""
|
|
session_data = await get_session_data(redis, session_id)
|
|
if session_data is None:
|
|
return None
|
|
|
|
user_id = uuid.UUID(session_data["user_id"])
|
|
|
|
# Verify user has an ACTIVE membership in target tenant
|
|
ut_q = select(UserTenant).where(
|
|
UserTenant.user_id == user_id,
|
|
UserTenant.tenant_id == new_tenant_id,
|
|
UserTenant.status == "active",
|
|
)
|
|
ut_result = await db.execute(ut_q)
|
|
user_tenant = ut_result.scalar_one_or_none()
|
|
if user_tenant is None:
|
|
return None
|
|
|
|
updated = await update_session_tenant(redis, session_id, new_tenant_id, role=user_tenant.role)
|
|
if updated is None:
|
|
return None
|
|
|
|
# Fetch tenant name
|
|
tenant_q = select(Tenant).where(Tenant.id == new_tenant_id)
|
|
tenant_result = await db.execute(tenant_q)
|
|
tenant = tenant_result.scalar_one_or_none()
|
|
|
|
updated["tenant_name"] = tenant.name if tenant else None
|
|
return updated
|
|
|
|
async def request_password_reset(
|
|
self,
|
|
db: AsyncSession,
|
|
email: str,
|
|
tenant_id: uuid.UUID | None = None,
|
|
) -> bool:
|
|
"""Create a password reset token. Always returns True (no user enumeration)."""
|
|
q = select(User).where(User.email == email, User.is_active == True) # noqa: E712
|
|
result = await db.execute(q)
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
return True # Don't reveal whether email exists
|
|
|
|
# Resolve tenant_id from UserTenant (default or specified)
|
|
ut_q = select(UserTenant).where(UserTenant.user_id == user.id)
|
|
if tenant_id is not None:
|
|
ut_q = ut_q.where(UserTenant.tenant_id == tenant_id)
|
|
else:
|
|
ut_q = ut_q.where(UserTenant.is_default == True) # noqa: E712
|
|
ut_result = await db.execute(ut_q)
|
|
user_tenant = ut_result.scalar_one_or_none()
|
|
if user_tenant is None:
|
|
# Fallback: get first tenant membership
|
|
ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
|
|
ut_result2 = await db.execute(ut_q2)
|
|
user_tenant = ut_result2.scalar_one_or_none()
|
|
if user_tenant is None:
|
|
return True
|
|
|
|
# Invalidate previous unused tokens
|
|
prev_q = select(PasswordResetToken).where(
|
|
PasswordResetToken.user_id == user.id,
|
|
PasswordResetToken.used_at.is_(None),
|
|
)
|
|
prev_result = await db.execute(prev_q)
|
|
for prev_token in prev_result.scalars().all():
|
|
prev_token.used_at = datetime.now(UTC)
|
|
|
|
# Set tenant context for RLS (auth session uses crm_auth role)
|
|
from app.core.db import set_tenant_context
|
|
await set_tenant_context(db, user_tenant.tenant_id)
|
|
|
|
# Create new token
|
|
import secrets
|
|
|
|
raw_token = secrets.token_urlsafe(32)
|
|
token_hash = hash_token(raw_token)
|
|
settings = get_settings()
|
|
expires_at = datetime.now(UTC) + timedelta(hours=settings.password_reset_expiry_hours)
|
|
|
|
reset_token = PasswordResetToken(
|
|
tenant_id=user_tenant.tenant_id,
|
|
user_id=user.id,
|
|
token_hash=token_hash,
|
|
expires_at=expires_at,
|
|
)
|
|
db.add(reset_token)
|
|
await db.flush()
|
|
|
|
# Enqueue ARQ job to send the password reset email
|
|
try:
|
|
from app.core.jobs import enqueue_job
|
|
|
|
await enqueue_job(
|
|
"send_password_reset_email",
|
|
user_id=str(user.id),
|
|
email=user.email,
|
|
raw_token=raw_token,
|
|
expires_at=expires_at.isoformat(),
|
|
)
|
|
logger.info("Enqueued password reset email job for user %s", user.id)
|
|
except Exception:
|
|
logger.warning(
|
|
"ARQ enqueue failed for password reset email — "
|
|
"email will not be sent. Check Redis/ARQ connectivity.",
|
|
exc_info=True,
|
|
)
|
|
|
|
return True
|
|
|
|
async def confirm_password_reset(
|
|
self,
|
|
db: AsyncSession,
|
|
token: str,
|
|
new_password: str,
|
|
) -> bool:
|
|
"""Reset password using a valid token. Returns True on success."""
|
|
token_hash = hash_token(token)
|
|
q = select(PasswordResetToken).where(
|
|
PasswordResetToken.token_hash == token_hash,
|
|
PasswordResetToken.used_at.is_(None),
|
|
)
|
|
result = await db.execute(q)
|
|
reset_token = result.scalar_one_or_none()
|
|
|
|
if reset_token is None:
|
|
return False
|
|
|
|
if reset_token.expires_at < datetime.now(UTC):
|
|
return False # Token expired
|
|
|
|
# Get user
|
|
user_q = select(User).where(User.id == reset_token.user_id)
|
|
user_result = await db.execute(user_q)
|
|
user = user_result.scalar_one_or_none()
|
|
if user is None:
|
|
return False
|
|
|
|
# Update password
|
|
user.password_hash = hash_password(new_password)
|
|
reset_token.used_at = datetime.now(UTC)
|
|
await db.flush()
|
|
|
|
# Invalidate all active Redis sessions for this user
|
|
try:
|
|
redis = get_redis()
|
|
# Scan for session keys and check which belong to this user
|
|
import json
|
|
|
|
async for key in redis.scan_iter(match="session:*", count=100):
|
|
raw = await redis.get(key)
|
|
if raw is None:
|
|
continue
|
|
try:
|
|
session_data = json.loads(raw)
|
|
except (json.JSONDecodeError, TypeError):
|
|
continue
|
|
if session_data.get("user_id") == str(user.id):
|
|
await redis.delete(key)
|
|
logger.info("Deleted session %s for user %s after password reset", key, user.id)
|
|
except Exception:
|
|
logger.warning("Failed to invalidate Redis sessions for user %s", user.id, exc_info=True)
|
|
|
|
# Audit log entry for password reset
|
|
try:
|
|
from app.core.db import set_tenant_context
|
|
await set_tenant_context(db, reset_token.tenant_id)
|
|
await log_audit(
|
|
db,
|
|
reset_token.tenant_id,
|
|
user.id,
|
|
"password_reset",
|
|
"user",
|
|
user.id,
|
|
changes={"action": "password_changed"},
|
|
)
|
|
except Exception:
|
|
logger.warning("Failed to create audit log for password reset of user %s", user.id, exc_info=True)
|
|
|
|
return True
|
|
|
|
async def get_password_reset_token_raw(self, db: AsyncSession, email: str) -> str | None:
|
|
"""Get the raw (unhashed) reset token for testing purposes.
|
|
This simulates what would be sent via email.
|
|
"""
|
|
import secrets
|
|
|
|
q = select(User).where(User.email == email)
|
|
result = await db.execute(q)
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
return None
|
|
|
|
# Get tenant_id from UserTenant
|
|
ut_q = select(UserTenant).where(
|
|
UserTenant.user_id == user.id,
|
|
UserTenant.is_default == True, # noqa: E712
|
|
)
|
|
ut_result = await db.execute(ut_q)
|
|
user_tenant = ut_result.scalar_one_or_none()
|
|
if user_tenant is None:
|
|
ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
|
|
ut_result2 = await db.execute(ut_q2)
|
|
user_tenant = ut_result2.scalar_one_or_none()
|
|
if user_tenant is None:
|
|
return None
|
|
|
|
raw_token = secrets.token_urlsafe(32)
|
|
token_hash = hash_token(raw_token)
|
|
settings = get_settings()
|
|
expires_at = datetime.now(UTC) + timedelta(hours=settings.password_reset_expiry_hours)
|
|
|
|
reset_token = PasswordResetToken(
|
|
tenant_id=user_tenant.tenant_id,
|
|
user_id=user.id,
|
|
token_hash=token_hash,
|
|
expires_at=expires_at,
|
|
)
|
|
db.add(reset_token)
|
|
await db.flush()
|
|
return raw_token
|
|
|
|
async def create_expired_reset_token(self, db: AsyncSession, email: str) -> str | None:
|
|
"""Create an already-expired reset token for testing."""
|
|
import secrets
|
|
|
|
q = select(User).where(User.email == email)
|
|
result = await db.execute(q)
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
return None
|
|
|
|
# Get tenant_id from UserTenant
|
|
ut_q = select(UserTenant).where(
|
|
UserTenant.user_id == user.id,
|
|
UserTenant.is_default == True, # noqa: E712
|
|
)
|
|
ut_result = await db.execute(ut_q)
|
|
user_tenant = ut_result.scalar_one_or_none()
|
|
if user_tenant is None:
|
|
ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
|
|
ut_result2 = await db.execute(ut_q2)
|
|
user_tenant = ut_result2.scalar_one_or_none()
|
|
if user_tenant is None:
|
|
return None
|
|
|
|
raw_token = secrets.token_urlsafe(32)
|
|
token_hash = hash_token(raw_token)
|
|
expires_at = datetime.now(UTC) - timedelta(hours=1) # Already expired
|
|
|
|
reset_token = PasswordResetToken(
|
|
tenant_id=user_tenant.tenant_id,
|
|
user_id=user.id,
|
|
token_hash=token_hash,
|
|
expires_at=expires_at,
|
|
)
|
|
db.add(reset_token)
|
|
await db.flush()
|
|
return raw_token
|
|
|
|
|
|
auth_service = AuthService()
|