cowork-local/core/accounts.py

"""Accounts + RBAC — Admin/Sub-admin/User identities shared across machines.

Stored one JSON file per account under ``<shared_dir>/accounts/`` (a plain
shared folder path — network share or a locally-synced OneDrive folder, see
``config.py``'s ``auth.shared_dir``). Deliberately NOT routed through the
Microsoft Graph API: Graph has no write access to an arbitrary share link,
only to the signed-in user's own drive, so a shared mutable store has to be
plain file I/O against a configured path instead.

Login validates a 12-character access code issued by an Admin (``code``),
or — for SSO — an already-verified company identity is matched to an
existing account by username (see ``ms365_auth.py``); SSO never creates an
account on its own, an Admin always provisions it first.
"""
from __future__ import annotations

import json
import re
import secrets
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Set, Tuple

from ..config import CONFIG_DIR

ROLES = ("admin", "subadmin", "user")

# A small LOCAL (never shared-folder) cache of the last successful login's
# identity — username + role only, never the access code — so a laptop that's
# off-VPN/off-network can still open the app as its last-known role. A
# revoked/edited account only takes effect once the shared folder is
# reachable again; see login_dialog.py.
_LAST_LOGIN_PATH = CONFIG_DIR / "last_login.json"


def save_last_login(username: str, role: str) -> None:
    """Nhớ tài khoản đăng nhập gần nhất để lần mở sau điền sẵn."""
    try:
        _LAST_LOGIN_PATH.parent.mkdir(parents=True, exist_ok=True)
        _LAST_LOGIN_PATH.write_text(
            json.dumps({"username": username, "role": role}), encoding="utf-8")
    except OSError:
        pass


def load_last_login() -> Optional[Tuple[str, str]]:
    """Cặp (tên đăng nhập, vai trò) của lần đăng nhập gần nhất; ``None`` nếu chưa có."""
    try:
        data = json.loads(_LAST_LOGIN_PATH.read_text(encoding="utf-8"))
        username, role = data.get("username", ""), data.get("role", "")
        if username and role in ROLES:
            return username, role
    except (OSError, json.JSONDecodeError, TypeError):
        pass
    return None

# Unambiguous alphanumeric alphabet for issued access codes — excludes
# characters easy to mis-type/mis-read (0/O, 1/I).
_CODE_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
CODE_LENGTH = 12


@dataclass
class Account:
    """Một tài khoản người dùng: tên đăng nhập, vai trò, tên hiển thị và nhóm."""
    username: str
    role: str
    display_name: str = ""
    department: str = ""
    email: str = ""
    group_id: str = ""
    code: str = ""
    created: str = ""
    created_by: str = ""


def accounts_dir(shared_dir: str) -> Path:
    """Thư mục chứa tài khoản, nằm trong thư mục chia sẻ của đội."""
    return Path(shared_dir).expanduser() / "accounts"


def _safe_username(username: str) -> str:
    """Normalize to lowercase alnum/./- only — matches the login screen's
    own auto-lowercase behavior, so a username is a stable, safe filename."""
    return re.sub(r"[^\w.\-]", "", (username or "").strip().lower())


def generate_code(existing_codes: Optional[Set[str]] = None) -> str:
    """A random, non-repeating 12-character access code."""
    existing = existing_codes or set()
    for _ in range(1000):
        code = "".join(secrets.choice(_CODE_ALPHABET) for _ in range(CODE_LENGTH))
        if code not in existing:
            return code
    raise RuntimeError("Could not generate a unique access code.")


def save_account(account: Account, directory: Path) -> Path:
    """Ghi một tài khoản ra ``<username>.json`` (tên file đã được làm sạch)."""
    directory.mkdir(parents=True, exist_ok=True)
    path = directory / f"{_safe_username(account.username)}.json"
    path.write_text(json.dumps(asdict(account), ensure_ascii=False, indent=2), encoding="utf-8")
    return path


def load_account(username: str, directory: Path) -> Optional[Account]:
    """Đọc một tài khoản theo tên đăng nhập; không có thì trả ``None``."""
    path = directory / f"{_safe_username(username)}.json"
    if not path.exists():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        known = {f for f in Account.__dataclass_fields__}
        return Account(**{k: v for k, v in data.items() if k in known})
    except (OSError, json.JSONDecodeError, TypeError):
        return None


def list_accounts(directory: Path) -> List[Account]:
    """Liệt kê mọi tài khoản trong thư mục; thư mục chưa có thì trả list rỗng."""
    if not directory.exists():
        return []
    out: List[Account] = []
    for path in sorted(directory.glob("*.json")):
        acc = load_account(path.stem, directory)
        if acc is not None:
            out.append(acc)
    out.sort(key=lambda a: a.username)
    return out


def delete_account(username: str, directory: Path) -> bool:
    """Xoá file tài khoản; trả về ``True`` nếu có file để xoá."""
    path = directory / f"{_safe_username(username)}.json"
    try:
        path.unlink()
        return True
    except OSError:
        return False


def find_by_username(username: str, directory: Path) -> Optional[Account]:
    """Bí danh của :func:`load_account`, giữ cho mã cũ gọi theo tên này vẫn chạy."""
    return load_account(username, directory)


def verify_login(username: str, code: str, directory: Path) -> Optional[Account]:
    """The matching Account when ``username``/``code`` are a valid pair."""
    account = load_account(username, directory)
    if account is None or not code or not account.code:
        return None
    return account if account.code == code else None


def new_account(username: str, role: str, display_name: str = "", department: str = "",
                email: str = "", group_id: str = "", created_by: str = "",
                existing_codes: Optional[Set[str]] = None) -> Account:
    """A fresh Account with a freshly generated, unique access code."""
    return Account(
        username=_safe_username(username),
        role=role if role in ROLES else "user",
        display_name=display_name,
        department=department,
        email=email,
        group_id=group_id,
        code=generate_code(existing_codes),
        created=datetime.now().isoformat(timespec="seconds"),
        created_by=created_by,
    )


# ---- single-admin invariant -------------------------------------------------
# The app allows exactly ONE account with role="admin" per shared folder. The
# helpers below are how every create/promote path checks and (for the
# first-run bootstrap) atomically claims that slot.

def admin_exists(directory: Path, exclude_username: str = "") -> bool:
    """True when some account other than ``exclude_username`` already holds
    the admin role."""
    return any(a.role == "admin" and a.username != _safe_username(exclude_username)
               for a in list_accounts(directory))


def claim_admin_slot(directory: Path) -> bool:
    """Atomically claim the right to create THE admin account.

    ``list-then-write`` alone leaves a race window: two never-configured
    machines pointed at the same share can both see an empty accounts folder
    and both create an "admin". An exclusive-create marker file closes it —
    ``open(..., "x")`` either succeeds for exactly one caller or raises for
    everyone else (also for later callers after a crash mid-bootstrap, which
    is fine: the marker plus admin_exists() are both checked by the caller).
    Returns True when this caller won the claim."""
    directory.mkdir(parents=True, exist_ok=True)
    marker = directory / ".admin_claimed"
    try:
        with open(marker, "x", encoding="utf-8") as fh:
            fh.write(datetime.now().isoformat(timespec="seconds"))
        return True
    except FileExistsError:
        return False
    except OSError:
        # Filesystems that can't do exclusive create (rare) — fall back to
        # the plain existence check so bootstrap isn't bricked entirely.
        return not admin_exists(directory)