paskia/paskia/db/operations.py

"""
Database for WebAuthn passkey authentication.

Read operations: Access _db directly, use build_* helpers to get public structs.
Context lookup: _db.session_ctx() returns full SessionContext with effective permissions.
Write operations: Functions that validate and commit, or raise ValueError.
"""

import hashlib
import logging
import os
from datetime import UTC, datetime, timedelta
from uuid import UUID

import uuid7

from paskia.config import SESSION_LIFETIME
from paskia.db.jsonl import (
    JsonlStore,
)
from paskia.db.structs import (
    DB,
    Config,
    Credential,
    Org,
    Permission,
    ResetToken,
    Role,
    Session,
    SessionContext,
    User,
)
from paskia.util.passphrase import is_well_formed as _is_passphrase

_logger = logging.getLogger(__name__)

# Global database instance (empty until init() loads data)
_db = DB()
_store = JsonlStore(_db)
_db._store = _store
_initialized = False


async def init(rp_id: str = "localhost", *args, **kwargs):
    """Load database from JSONL file."""
    global _db, _initialized
    if _initialized:
        _logger.debug("Database already initialized, skipping reload")
        return
    default_path = f"{rp_id}.paskiadb"
    db_path = os.environ.get("PASKIA_DB", default_path)
    await _store.load(db_path, rp_id=rp_id)
    _db = _store.db
    _initialized = True


# -------------------------------------------------------------------------
# Read/lookup functions
# -------------------------------------------------------------------------


def get_user_organization(user_uuid: UUID) -> tuple[Org, str]:
    """Get the organization a user belongs to and their role name.

    Raises ValueError if user not found.

    Call sites:
    - admin_create_user_registration_link: org only
    - admin_get_user_detail: org and role
    - admin_update_user_display_name: org only
    - admin_delete_user_credential: org only
    - admin_delete_user_session: org only
    - admin_update_user_role: org only
    """
    if user_uuid not in _db.users:
        raise ValueError(f"User {user_uuid} not found")
    user = _db.users[user_uuid]
    role = user.role
    return role.org, role.display_name


def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]:
    """Get all users in an organization with their role names.

    Returns list of (User, role_display_name) tuples.
    """
    org = _db.orgs[org_uuid]
    return [(u, u.role.display_name) for role in org.roles for u in role.users]


def get_user_credential_ids(user_uuid: UUID) -> list[bytes]:
    """Get credential IDs for a user (for WebAuthn exclude lists).

    Returns empty list if user has no credentials.
    """
    assert user_uuid
    return [c.credential_id for c in _db.users[user_uuid].credentials]


def _reset_key(passphrase: str) -> bytes:
    """Hash a passphrase to bytes for reset token storage."""
    if not _is_passphrase(passphrase):
        raise ValueError(
            "Trying to reset with a session token in place of a passphrase"
            if len(passphrase) == 16
            else "Invalid passphrase format"
        )
    return hashlib.sha512(passphrase.encode()).digest()[:9]


def get_reset_token(passphrase: str) -> ResetToken | None:
    """Get reset token by passphrase.

    Call sites:
    - Get reset token to validate it (authsession.py:34)
    """
    key = _reset_key(passphrase)
    return _db.reset_tokens.get(key)


# -------------------------------------------------------------------------
# Write operations (validate, modify, commit or raise ValueError)
# -------------------------------------------------------------------------


def create_permission(perm: Permission, *, ctx: SessionContext | None = None) -> None:
    """Create a new permission."""
    if perm.uuid in _db.permissions:
        raise ValueError(f"Permission {perm.uuid} already exists")
    with _db.transaction("admin:create_permission", ctx):
        perm.store()


def update_permission(
    uuid: UUID,
    scope: str,
    display_name: str,
    domain: str | None = None,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update a permission's scope, display_name, and domain.

    Only these fields can be modified; created_at and other metadata remain immutable.
    """
    if uuid not in _db.permissions:
        raise ValueError(f"Permission {uuid} not found")
    with _db.transaction("admin:update_permission", ctx):
        _db.permissions[uuid].scope = scope
        _db.permissions[uuid].display_name = display_name
        _db.permissions[uuid].domain = domain


def delete_permission(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
    """Delete a permission and remove it from all roles."""
    if uuid not in _db.permissions:
        raise ValueError(f"Permission {uuid} not found")
    with _db.transaction("admin:delete_permission", ctx):
        _db.permissions[uuid].delete()


def create_org(org: Org, *, ctx: SessionContext | None = None) -> None:
    """Create a new organization with an Administration role.

    Automatically creates an 'Administration' role with auth:org:admin permission.
    """
    if org.uuid in _db.orgs:
        raise ValueError(f"Organization {org.uuid} already exists")
    now = datetime.now(UTC)
    with _db.transaction("admin:create_org", ctx):
        new_org = Org.create(display_name=org.display_name, created_at=now)
        new_org.uuid = org.uuid
        new_org.store()
        # Create Administration role with org admin permission

        admin_role_uuid = uuid7.create(now)
        # Find the auth:org:admin permission UUID
        org_admin_perm_uuid = None
        for pid, p in _db.permissions.items():
            if p.scope == "auth:org:admin":
                org_admin_perm_uuid = pid
                break
        role_permissions = {org_admin_perm_uuid: True} if org_admin_perm_uuid else {}
        admin_role = Role(
            org_uuid=org.uuid,
            display_name="Administration",
            permissions=role_permissions,
        )
        admin_role.uuid = admin_role_uuid
        admin_role.store()


def update_org_name(
    uuid: UUID,
    display_name: str,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update organization display name."""
    if uuid not in _db.orgs:
        raise ValueError(f"Organization {uuid} not found")
    with _db.transaction("admin:update_org_name", ctx):
        _db.orgs[uuid].display_name = display_name


def delete_org(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
    """Delete organization and all its roles/users."""
    if uuid not in _db.orgs:
        raise ValueError(f"Organization {uuid} not found")
    with _db.transaction("admin:delete_org", ctx):
        _db.orgs[uuid].delete()


def add_permission_to_org(
    org_uuid: UUID,
    permission_uuid: UUID,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Grant a permission to an organization by UUID."""
    if org_uuid not in _db.orgs:
        raise ValueError(f"Organization {org_uuid} not found")

    if permission_uuid not in _db.permissions:
        raise ValueError(f"Permission {permission_uuid} not found")

    with _db.transaction("admin:add_permission_to_org", ctx):
        _db.permissions[permission_uuid].orgs[org_uuid] = True


def remove_permission_from_org(
    org_uuid: UUID,
    permission_uuid: UUID,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Remove a permission from an organization by UUID."""
    if org_uuid not in _db.orgs:
        raise ValueError(f"Organization {org_uuid} not found")

    if permission_uuid not in _db.permissions:
        return  # Permission not found, silently return

    with _db.transaction("admin:remove_permission_from_org", ctx):
        _db.permissions[permission_uuid].orgs.pop(org_uuid, None)


def create_role(role: Role, *, ctx: SessionContext | None = None) -> None:
    """Create a new role."""
    if role.uuid in _db.roles:
        raise ValueError(f"Role {role.uuid} already exists")
    if role.org_uuid not in _db.orgs:
        raise ValueError(f"Organization {role.org_uuid} not found")
    with _db.transaction("admin:create_role", ctx):
        role.store()


def update_role_name(
    uuid: UUID,
    display_name: str,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update role display name."""
    if uuid not in _db.roles:
        raise ValueError(f"Role {uuid} not found")
    with _db.transaction("admin:update_role_name", ctx):
        _db.roles[uuid].display_name = display_name


def add_permission_to_role(
    role_uuid: UUID,
    permission_uuid: UUID,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Add permission to role by UUID."""
    if role_uuid not in _db.roles:
        raise ValueError(f"Role {role_uuid} not found")
    if permission_uuid not in _db.permissions:
        raise ValueError(f"Permission {permission_uuid} not found")
    with _db.transaction("admin:add_permission_to_role", ctx):
        _db.roles[role_uuid].permissions[permission_uuid] = True


def remove_permission_from_role(
    role_uuid: UUID,
    permission_uuid: UUID,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Remove permission from role by UUID."""
    if role_uuid not in _db.roles:
        raise ValueError(f"Role {role_uuid} not found")
    with _db.transaction("admin:remove_permission_from_role", ctx):
        _db.roles[role_uuid].permissions.pop(permission_uuid, None)


def delete_role(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
    """Delete a role."""
    if uuid not in _db.roles:
        raise ValueError(f"Role {uuid} not found")
    # Check no users have this role
    role = _db.roles[uuid]
    if role.users:
        raise ValueError(f"Cannot delete role {uuid}: users still assigned")
    with _db.transaction("admin:delete_role", ctx):
        _db.roles[uuid].delete()


def create_user(new_user: User, *, ctx: SessionContext | None = None) -> None:
    """Create a new user."""
    if new_user.uuid in _db.users:
        raise ValueError(f"User {new_user.uuid} already exists")
    if new_user.role_uuid not in _db.roles:
        raise ValueError(f"Role {new_user.role_uuid} not found")
    with _db.transaction("admin:create_user", ctx):
        new_user.store()


def update_user_display_name(
    uuid: UUID,
    display_name: str,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update user display name.

    The acting user should be logged via ctx.
    For self-service (user updating own name), pass user's ctx.
    For admin operations, pass admin's ctx.
    """
    if isinstance(uuid, str):
        uuid = UUID(uuid)
    if uuid not in _db.users:
        raise ValueError(f"User {uuid} not found")
    with _db.transaction("update_user_display_name", ctx):
        _db.users[uuid].display_name = display_name


def update_user_theme(
    uuid: UUID,
    theme: str,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update user theme preference ('' for auto, 'light', 'dark')."""
    if isinstance(uuid, str):
        uuid = UUID(uuid)
    if uuid not in _db.users:
        raise ValueError(f"User {uuid} not found")
    if theme not in ("", "light", "dark"):
        raise ValueError(f"Invalid theme: {theme}")
    with _db.transaction("update_user_theme", ctx):
        _db.users[uuid].theme = theme


def update_user_role(
    uuid: UUID,
    role_uuid: UUID,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update user's role."""
    if uuid not in _db.users:
        raise ValueError(f"User {uuid} not found")
    if role_uuid not in _db.roles:
        raise ValueError(f"Role {role_uuid} not found")
    with _db.transaction("admin:update_user_role", ctx):
        _db.users[uuid].role_uuid = role_uuid


def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
    """Delete user and their credentials/sessions."""
    if uuid not in _db.users:
        raise ValueError(f"User {uuid} not found")
    with _db.transaction("admin:delete_user", ctx):
        _db.users[uuid].delete()


def create_credential(cred: Credential, *, ctx: SessionContext | None = None) -> None:
    """Create a new credential."""
    if cred.uuid in _db.credentials:
        raise ValueError(f"Credential {cred.uuid} already exists")
    if cred.user_uuid not in _db.users:
        raise ValueError(f"User {cred.user_uuid} not found")
    with _db.transaction("create_credential", ctx):
        cred.store()


def update_credential_sign_count(
    uuid: UUID,
    sign_count: int,
    last_used: datetime | None = None,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update credential sign count and last_used."""
    if uuid not in _db.credentials:
        raise ValueError(f"Credential {uuid} not found")
    with _db.transaction("update_credential_sign_count", ctx):
        _db.credentials[uuid].sign_count = sign_count
        if last_used:
            _db.credentials[uuid].last_used = last_used


def delete_credential(
    uuid: UUID,
    user_uuid: UUID | None = None,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Delete a credential and all sessions using it.

    If user_uuid is provided, validates that the credential belongs to that user.
    """
    if uuid not in _db.credentials:
        raise ValueError(f"Credential {uuid} not found")
    cred = _db.credentials[uuid]
    if user_uuid is not None:
        if cred.user_uuid != user_uuid:
            raise ValueError(f"Credential {uuid} does not belong to user {user_uuid}")
    with _db.transaction("delete_credential", ctx):
        cred.delete()


def create_session(
    user_uuid: UUID,
    credential_uuid: UUID,
    host: str,
    ip: str,
    user_agent: str,
    duration: timedelta = SESSION_LIFETIME,
    *,
    ctx: SessionContext | None = None,
) -> str:
    """Create a new session. Returns the session key."""
    if user_uuid not in _db.users:
        raise ValueError(f"User {user_uuid} not found")
    if credential_uuid not in _db.credentials:
        raise ValueError(f"Credential {credential_uuid} not found")
    now = datetime.now(UTC)
    session = Session.create(
        user=user_uuid,
        credential=credential_uuid,
        host=host,
        ip=ip,
        user_agent=user_agent,
        expiry=now + duration,
    )
    if session.key in _db.sessions:
        raise ValueError("Session already exists")
    with _db.transaction("create_session", ctx):
        session.store(now)
    return session.key


def update_session(
    key: str,
    host: str | None = None,
    ip: str | None = None,
    user_agent: str | None = None,
    expiry: datetime | None = None,
    *,
    ctx: SessionContext | None = None,
) -> None:
    """Update session metadata."""
    if key not in _db.sessions:
        raise ValueError("Session not found")
    with _db.transaction("update_session", ctx):
        s = _db.sessions[key]
        if host is not None:
            s.host = host
        if ip is not None:
            s.ip = ip
        if user_agent is not None:
            s.user_agent = user_agent
        if expiry is not None:
            s.expiry = expiry


def set_session_host(key: str, host: str, *, ctx: SessionContext | None = None) -> None:
    """Set the host for a session (first-time binding)."""
    update_session(key, host=host, ctx=ctx)


def delete_session(
    key: str, *, ctx: SessionContext | None = None, action: str = "delete_session"
) -> None:
    """Delete a session.

    The acting user should be logged via ctx.
    For user logout, pass ctx of the user's session and action="logout".
    For admin terminating a session, pass admin's ctx.
    """
    if key not in _db.sessions:
        raise ValueError("Session not found")
    with _db.transaction(action, ctx):
        _db.sessions[key].delete()


def delete_sessions_for_user(
    user_uuid: UUID, *, ctx: SessionContext | None = None
) -> None:
    """Delete all sessions for a user.

    The acting user should be logged via ctx.
    For user logout-all, pass ctx of the user's session.
    For admin bulk termination, pass admin's ctx.
    """
    user = _db.users.get(user_uuid)
    if not user:
        return
    with _db.transaction("admin:delete_sessions_for_user", ctx):
        for sess in user.sessions:
            sess.delete()


def create_reset_token(
    user_uuid: UUID,
    expiry: datetime,
    token_type: str,
    *,
    ctx: SessionContext | None = None,
    user: str | None = None,
) -> str:
    """Create a reset token and return the passphrase.

    The acting user should be logged via ctx.
    For self-service (user creating own recovery link), pass user's ctx.
    For admin operations, pass admin's ctx.
    For system operations (bootstrap), pass neither to log no user.
    For API operations where ctx is not available but user is known, pass user.

    Returns:
        The passphrase to give to the user.
    """
    if user_uuid not in _db.users:
        raise ValueError(f"User {user_uuid} not found")
    token, passphrase = ResetToken.create(
        user=user_uuid, expiry=expiry, token_type=token_type
    )
    if token.key in _db.reset_tokens:
        raise ValueError("Reset token already exists")
    with _db.transaction("create_reset_token", ctx, user=user):
        token.store()
    return passphrase


def delete_reset_token(key: bytes, *, ctx: SessionContext | None = None) -> None:
    """Delete a reset token."""
    if key not in _db.reset_tokens:
        raise ValueError("Reset token not found")
    with _db.transaction("delete_reset_token", ctx):
        _db.reset_tokens[key].delete()


# -------------------------------------------------------------------------
# Cleanup (called by background task)
# -------------------------------------------------------------------------


def cleanup_expired() -> int:
    """Remove expired sessions and reset tokens. Returns count removed."""
    now = datetime.now(UTC)
    count = 0
    with _db.transaction("expiry"):
        expired_sessions = [k for k, s in _db.sessions.items() if s.expiry < now]
        for k in expired_sessions:
            del _db.sessions[k]
            count += 1
        expired_tokens = [k for k, t in _db.reset_tokens.items() if t.expiry < now]
        for k in expired_tokens:
            del _db.reset_tokens[k]
            count += 1
    return count


# -------------------------------------------------------------------------
# Composite operations (used by app code)
# -------------------------------------------------------------------------


def login(
    user_uuid: UUID,
    credential_uuid: UUID,
    sign_count: int,
    host: str,
    ip: str,
    user_agent: str,
    duration: timedelta = SESSION_LIFETIME,
) -> str:
    """Update user/credential on login and create session in a single transaction.

    Updates:
    - user.last_seen, user.visits
    - credential.sign_count, credential.last_used
    Creates:
    - new session

    Returns the generated session token.
    """
    if isinstance(user_uuid, str):
        user_uuid = UUID(user_uuid)
    now = datetime.now(UTC)
    if user_uuid not in _db.users:
        raise ValueError(f"User {user_uuid} not found")
    if credential_uuid not in _db.credentials:
        raise ValueError(f"Credential {credential_uuid} not found")

    session = Session.create(
        user=user_uuid,
        credential=credential_uuid,
        host=host,
        ip=ip,
        user_agent=user_agent,
        expiry=now + duration,
    )
    user_str = str(user_uuid)
    with _db.transaction("login", user=user_str):
        session.store(now)
        # Update credential
        _db.credentials[credential_uuid].sign_count = sign_count
        _db.credentials[credential_uuid].last_used = now
    return session.key


def create_credential_session(
    user_uuid: UUID,
    credential: Credential,
    host: str,
    ip: str,
    user_agent: str,
    display_name: str | None = None,
    reset_key: bytes | None = None,
) -> str:
    """Create a credential and session together, optionally consuming a reset token.

    Used during registration to atomically:
    1. Update user display_name if provided
    2. Create the credential
    3. Create the session
    4. Delete the reset token if provided

    Returns the generated session token.
    """

    now = datetime.now(UTC)

    if user_uuid not in _db.users:
        raise ValueError(f"User {user_uuid} not found")

    session = Session.create(
        user=user_uuid,
        credential=credential.uuid,
        host=host,
        ip=ip,
        user_agent=user_agent,
        expiry=now + SESSION_LIFETIME,
    )
    user_str = str(user_uuid)
    with _db.transaction("create_credential_session", user=user_str):
        # Update display name if provided
        if display_name:
            _db.users[user_uuid].display_name = display_name

        # Align credential timestamps with transaction time
        credential.created_at = now
        credential.last_used = now
        credential.last_verified = now

        # Create credential
        credential.store()

        # Store session and record visit
        session.store(now)

        # Delete reset token if provided
        if reset_key:
            token = _db.reset_tokens.get(reset_key)
            if token:
                token.delete()
    return session.key


# -------------------------------------------------------------------------
# Config operations
# -------------------------------------------------------------------------


def get_config() -> Config:
    """Get the stored configuration."""
    return _db.config


async def set_config(config: Config) -> None:
    """Update the stored configuration."""
    with _db.transaction("update_config"):
        _db.config = config