hermes-agent/plugins/platforms/google_chat/oauth.py

"""User OAuth helper for the Google Chat gateway adapter.

Google Chat's ``media.upload`` REST endpoint hard-rejects service-account
authentication:

    "This method doesn't support app authentication with a service
     account. Authenticate with a user account."

(See https://developers.google.com/workspace/chat/api/reference/rest/v1/media/upload
and https://developers.google.com/chat/api/guides/auth/users.)

For the bot to deliver native file attachments — the same drag-and-drop
file widget the user gets when they upload manually — each user must
grant the bot the ``chat.messages.create`` scope ONCE in their own DM.
The bot stores per-user refresh tokens and calls ``media.upload`` plus
the subsequent ``messages.create`` *as the requesting user* whenever a
file needs sending.

This module is BOTH a CLI tool (driven by the agent via slash commands or
terminal commands) AND a library imported by ``google_chat.py``:

    Library functions (called from the adapter at runtime):
        load_user_credentials(email=None) -> Credentials | None
        refresh_or_none(creds, email=None) -> Credentials | None
        build_user_chat_service(creds) -> chat_v1.Resource
        list_authorized_emails() -> List[str]

    CLI commands (driven by the agent through the /setup-files slash
    command, modeled on skills/productivity/google-workspace/scripts/setup.py):
        --check                          Exit 0 if auth is valid, else 1
        --client-secret /path/to.json    Persist OAuth client credentials
        --auth-url                       Print the OAuth URL for the user
        --auth-code CODE                 Exchange auth code for token
        --revoke                         Revoke and delete stored token
        --install-deps                   Install Python dependencies
        --email EMAIL                    Scope CLI ops to a specific user
                                         (defaults to legacy single-user
                                         mode when omitted)

The flow mirrors the existing google-workspace skill exactly so anyone
familiar with that flow can read this without surprises.

Token storage layout
--------------------
- Per-user tokens (keyed by sender email):
    ``${HERMES_HOME}/google_chat_user_tokens/<sanitized_email>.json``
- Legacy single-user token (fallback, untouched for backward compat):
    ``${HERMES_HOME}/google_chat_user_token.json``
- Per-user pending OAuth state during /setup-files start → exchange:
    ``${HERMES_HOME}/google_chat_user_oauth_pending/<sanitized_email>.json``
- Legacy pending state:
    ``${HERMES_HOME}/google_chat_user_oauth_pending.json``
- OAuth client secret (profile-scoped — each profile registers its own):
    ``${HERMES_HOME}/google_chat_user_client_secret.json``
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import re
import secrets
import stat
import subprocess
import sys
from pathlib import Path
from typing import Any, List, Optional, Tuple

# Pin the legacy logger name so operator-side log filters keep matching
# after the in-tree → plugin migration. See adapter.py for context.
logger = logging.getLogger("gateway.platforms.google_chat_user_oauth")

# Use the project's HERMES_HOME helper so the token follows the user's
# profile (e.g. tests can override via HERMES_HOME=/tmp/...).
try:
    from hermes_constants import display_hermes_home, get_hermes_home
except (ModuleNotFoundError, ImportError):
    # Fallback for environments where hermes_constants isn't importable
    # (mirrors the same fallback used by the google-workspace skill's
    # _hermes_home.py shim).
    def get_hermes_home() -> Path:
        val = os.environ.get("HERMES_HOME", "").strip()
        return Path(val) if val else Path.home() / ".hermes"

    def display_hermes_home() -> str:
        home = get_hermes_home()
        try:
            return "~/" + str(home.relative_to(Path.home()))
        except ValueError:
            return str(home)

from utils import atomic_replace


def _hermes_home() -> Path:
    """Resolve HERMES_HOME at call time (NOT module import).

    Tests and ``HERMES_HOME=...`` env overrides need this to be late-
    binding. If we cached the path at import time, switching profiles
    or tweaking env vars in tests would silently keep using the old
    path."""
    return get_hermes_home()


# Filesystem-safe key: lowercase, allow ``[a-z0-9._-@]``, replace anything
# else with ``_``. ``ramon.fernandez@nttdata.com`` stays human-readable
# (``ramon.fernandez@nttdata.com.json``) which makes admin debugging by
# ``ls ~/.hermes/google_chat_user_tokens/`` trivial.
_EMAIL_FS_RE = re.compile(r"[^a-z0-9._@-]+")


def _sanitize_email(email: str) -> str:
    cleaned = _EMAIL_FS_RE.sub("_", (email or "").strip().lower())
    return cleaned or "_unknown_"


def _legacy_token_path() -> Path:
    return _hermes_home() / "google_chat_user_token.json"


def _user_tokens_dir() -> Path:
    return _hermes_home() / "google_chat_user_tokens"


def _legacy_pending_path() -> Path:
    return _hermes_home() / "google_chat_user_oauth_pending.json"


def _user_pending_dir() -> Path:
    return _hermes_home() / "google_chat_user_oauth_pending"


def _token_path(email: Optional[str] = None) -> Path:
    """Return the on-disk token path for ``email`` or the legacy path."""
    if email:
        return _user_tokens_dir() / f"{_sanitize_email(email)}.json"
    return _legacy_token_path()


def _client_secret_path() -> Path:
    return _hermes_home() / "google_chat_user_client_secret.json"


def _pending_auth_path(email: Optional[str] = None) -> Path:
    if email:
        return _user_pending_dir() / f"{_sanitize_email(email)}.json"
    return _legacy_pending_path()


# Minimum scope for native Chat attachment delivery.
# `chat.messages.create` covers BOTH `media.upload` and the subsequent
# `messages.create` that references the attachmentDataRef. We deliberately
# do NOT request drive.file or other scopes — least privilege.
SCOPES: List[str] = [
    "https://www.googleapis.com/auth/chat.messages.create",
]

# Pip packages required for the OAuth flow.
_REQUIRED_PACKAGES = [
    "google-api-python-client",
    "google-auth-oauthlib",
    "google-auth-httplib2",
]

# Out-of-band redirect: Google deprecated the ``urn:ietf:wg:oauth:2.0:oob``
# flow, so we use a localhost redirect that's expected to FAIL. The user
# copies the auth code from the failed browser URL bar back into chat.
# Same trick used by skills/productivity/google-workspace/scripts/setup.py.
_REDIRECT_URI = "http://localhost:1"


# =============================================================================
# Library API — called from the adapter at runtime
# =============================================================================


def load_user_credentials(email: Optional[str] = None) -> Optional[Any]:
    """Load + validate persisted user OAuth credentials.

    ``email`` selects the per-user token file; ``None`` falls back to the
    legacy single-user path (left in place for installs that ran the
    pre-multi-user flow). Returns a ``google.oauth2.credentials.Credentials``
    instance ready for use, or ``None`` if no token is stored, the token
    is corrupt, or refresh fails. Adapter callers should treat ``None``
    as "user has not run /setup-files yet" and surface the setup-instructions
    fallback to the user.

    Does NOT raise on the no-token case — that's expected.
    """
    token_path = _token_path(email)
    if not token_path.exists():
        return None

    # Same class as slack_tokens.json: hand-provisioned or legacy-written
    # token files commonly end up 0o644. Warn so the owner tightens them.
    from utils import warn_if_credential_file_broadly_readable

    warn_if_credential_file_broadly_readable(
        token_path, label="[google_chat_user_oauth]", log=logger
    )

    try:
        from google.oauth2.credentials import Credentials
        from google.auth.transport.requests import Request
    except ImportError:
        logger.warning(
            "[google_chat_user_oauth] google-auth not installed; user-OAuth "
            "attachment delivery is disabled. Run `hermes setup` to install Google Chat support."
        )
        return None

    try:
        # Don't pass scopes — user may have authorized only a subset, and
        # passing scopes makes refresh validate them strictly. Same logic
        # as the google-workspace skill.
        creds = Credentials.from_authorized_user_file(str(token_path))
    except Exception as exc:
        logger.warning(
            "[google_chat_user_oauth] token at %s is corrupt: %s",
            token_path, exc,
        )
        return None

    if creds.valid:
        return creds

    if creds.expired and creds.refresh_token:
        try:
            creds.refresh(Request())
        except Exception as exc:
            logger.warning(
                "[google_chat_user_oauth] token refresh failed (user "
                "should re-run /setup-files): %s", exc,
            )
            return None
        # Persist refreshed token so next start picks up the new access
        # token without an unnecessary refresh round-trip.
        _persist_credentials(creds, token_path)
        return creds

    # Token exists but is unusable (e.g. revoked, no refresh token).
    return None


def refresh_or_none(creds: Any, email: Optional[str] = None) -> Optional[Any]:
    """Refresh ``creds`` if expired. Returns the credentials or ``None``.

    Used by the adapter just before calling media.upload to ensure the
    token is current. Returns ``None`` if refresh fails — caller falls
    back to the text-notice path. ``email`` controls where the refreshed
    token is written back; ``None`` keeps the legacy single-file path.
    """
    if creds is None:
        return None

    if creds.valid:
        return creds

    try:
        from google.auth.transport.requests import Request
    except ImportError:
        return None

    if creds.expired and creds.refresh_token:
        try:
            creds.refresh(Request())
            _persist_credentials(creds, _token_path(email))
            return creds
        except Exception as exc:
            logger.warning(
                "[google_chat_user_oauth] refresh failed: %s", exc,
            )
            return None

    return None


def build_user_chat_service(creds: Any) -> Any:
    """Build a Google Chat API client authenticated as the user.

    Used for media.upload + the subsequent messages.create that
    references the attachmentDataRef. The bot's separate SA-authed
    client (``self._chat_api`` in the adapter) is for everything else.
    """
    from googleapiclient.discovery import build as build_service
    return build_service("chat", "v1", credentials=creds, cache_discovery=False)


def list_authorized_emails() -> List[str]:
    """Return the set of user emails that have stored per-user tokens.

    Lists files in the per-user tokens dir; does NOT include the legacy
    single-user token (its owner is unknown). Sanitized filenames lose
    the ``+suffix`` part of plus-addressed emails — accept that and use
    this list only for admin display, not for trust decisions.
    """
    d = _user_tokens_dir()
    if not d.exists():
        return []
    out: List[str] = []
    for f in d.iterdir():
        if f.is_file() and f.suffix == ".json":
            out.append(f.stem)
    out.sort()
    return out


def _persist_credentials(creds: Any, token_path: Path) -> None:
    """Persist refreshed credentials atomically with private permissions."""
    try:
        _write_private_json(
            token_path,
            _normalize_authorized_user_payload(json.loads(creds.to_json())),
        )
    except Exception:
        logger.debug(
            "[google_chat_user_oauth] failed to persist credentials at %s",
            token_path, exc_info=True,
        )


# =============================================================================
# CLI commands — driven by the agent via /setup-files
# =============================================================================


def _normalize_authorized_user_payload(payload: dict) -> dict:
    """Ensure the persisted token JSON has the type field google-auth expects."""
    normalized = dict(payload)
    if not normalized.get("type"):
        normalized["type"] = "authorized_user"
    return normalized


def _write_private_json(path: Path, data: Any) -> None:
    """Atomically write JSON with 0o600 permissions where supported."""
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        os.chmod(path.parent, 0o700)
    except OSError:
        pass

    tmp_path = path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}")
    try:
        fd = os.open(
            str(tmp_path),
            os.O_WRONLY | os.O_CREAT | os.O_EXCL,
            stat.S_IRUSR | stat.S_IWUSR,
        )
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            json.dump(data, fh, indent=2, ensure_ascii=False)
            fh.flush()
            os.fsync(fh.fileno())
        atomic_replace(tmp_path, path)
        try:
            os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
        except OSError:
            pass
    finally:
        try:
            if tmp_path.exists():
                tmp_path.unlink()
        except OSError:
            pass


def _ensure_deps() -> None:
    """Check deps available; install if not; exit on failure."""
    try:
        import googleapiclient  # noqa: F401
        import google_auth_oauthlib  # noqa: F401
    except ImportError:
        if not install_deps():
            sys.exit(1)


def install_deps() -> bool:
    try:
        import googleapiclient  # noqa: F401
        import google_auth_oauthlib  # noqa: F401
        print("Dependencies already installed.")
        return True
    except ImportError:
        pass

    print("Installing Google Chat OAuth dependencies...")
    try:
        from hermes_cli.tools_config import _pip_install

        result = _pip_install(["--quiet"] + _REQUIRED_PACKAGES)
        if result.returncode != 0:
            raise RuntimeError((result.stderr or "install failed").strip()[:300])
        print("Dependencies installed.")
        return True
    except Exception as exc:
        print(f"ERROR: Failed to install dependencies: {exc}")
        print("Run `hermes setup` to repair the managed installation, then retry.")
        return False


def check_auth(email: Optional[str] = None) -> bool:
    """Print status; return True if creds are usable.

    Per-user when ``email`` given, legacy single-user when omitted.
    """
    token_path = _token_path(email)
    if not token_path.exists():
        print(f"NOT_AUTHENTICATED: No token at {token_path}")
        return False

    creds = load_user_credentials(email)
    if creds is None:
        print(f"TOKEN_INVALID: Re-run /setup-files (path: {token_path})")
        return False

    print(f"AUTHENTICATED: Token valid at {token_path}")
    return True


def store_client_secret(path: str) -> None:
    """Validate and copy the user's OAuth client_secret.json into HERMES_HOME."""
    src = Path(path).expanduser().resolve()
    if not src.exists():
        print(f"ERROR: File not found: {src}")
        sys.exit(1)

    try:
        data = json.loads(src.read_text())
    except json.JSONDecodeError:
        print("ERROR: File is not valid JSON.")
        sys.exit(1)

    if "installed" not in data and "web" not in data:
        print(
            "ERROR: Not a Google OAuth client secret file (missing "
            "'installed' or 'web' key)."
        )
        print(
            "Download from: https://console.cloud.google.com/apis/credentials"
        )
        sys.exit(1)

    target = _client_secret_path()
    _write_private_json(target, data)
    print(f"OK: Client secret saved to {target}")


def _save_pending_auth(*, state: str, code_verifier: str,
                      email: Optional[str] = None) -> None:
    pending = _pending_auth_path(email)
    _write_private_json(
        pending,
        {
            "state": state,
            "code_verifier": code_verifier,
            "redirect_uri": _REDIRECT_URI,
            "email": email or "",
        },
    )


def _load_pending_auth(email: Optional[str] = None) -> dict:
    pending = _pending_auth_path(email)
    if not pending.exists():
        print("ERROR: No pending OAuth session found. Run --auth-url first.")
        sys.exit(1)
    try:
        data = json.loads(pending.read_text())
    except Exception as exc:
        print(f"ERROR: Could not read pending OAuth session: {exc}")
        print("Run --auth-url again to start a fresh session.")
        sys.exit(1)
    if not data.get("state") or not data.get("code_verifier"):
        print("ERROR: Pending OAuth session is missing PKCE data.")
        print("Run --auth-url again.")
        sys.exit(1)
    return data


def _extract_code_and_state(code_or_url: str) -> Tuple[str, Optional[str]]:
    """Accept a raw auth code OR the full failed-redirect URL the user pastes."""
    if not code_or_url.startswith("http"):
        return code_or_url, None

    from urllib.parse import parse_qs, urlparse

    parsed = urlparse(code_or_url)
    params = parse_qs(parsed.query)
    if "code" not in params:
        print("ERROR: No 'code' parameter found in URL.")
        sys.exit(1)
    state = params.get("state", [None])[0]
    return params["code"][0], state


def get_auth_url(email: Optional[str] = None) -> None:
    """Print the OAuth URL for the user to visit. Persists PKCE state.

    ``email`` namespaces the pending state so two users can be mid-flow
    in parallel without trampling each other's PKCE verifier.
    """
    if not _client_secret_path().exists():
        print("ERROR: No client secret stored. Run --client-secret first.")
        sys.exit(1)

    _ensure_deps()
    from google_auth_oauthlib.flow import Flow

    flow = Flow.from_client_secrets_file(
        str(_client_secret_path()),
        scopes=SCOPES,
        redirect_uri=_REDIRECT_URI,
        autogenerate_code_verifier=True,
    )
    auth_url, state = flow.authorization_url(
        access_type="offline",
        prompt="consent",
    )
    _save_pending_auth(state=state, code_verifier=flow.code_verifier, email=email)
    print(auth_url)


def exchange_auth_code(code: str, email: Optional[str] = None) -> None:
    """Exchange an auth code (or pasted redirect URL) for a refresh token.

    ``email`` selects the destination token path. ``None`` writes to the
    legacy single-user path (kept for the existing CLI entrypoint and for
    pre-multi-user installs).
    """
    if not _client_secret_path().exists():
        print("ERROR: No client secret stored. Run --client-secret first.")
        sys.exit(1)

    pending_auth = _load_pending_auth(email)
    raw_callback = code
    code, returned_state = _extract_code_and_state(code)
    if returned_state and returned_state != pending_auth["state"]:
        print(
            "ERROR: OAuth state mismatch. Run --auth-url again to start a "
            "fresh session."
        )
        sys.exit(1)

    _ensure_deps()
    from google_auth_oauthlib.flow import Flow
    from urllib.parse import parse_qs, urlparse

    granted_scopes = list(SCOPES)
    if isinstance(raw_callback, str) and raw_callback.startswith("http"):
        params = parse_qs(urlparse(raw_callback).query)
        scope_val = (params.get("scope") or [""])[0].strip()
        if scope_val:
            granted_scopes = scope_val.split()

    flow = Flow.from_client_secrets_file(
        str(_client_secret_path()),
        scopes=granted_scopes,
        redirect_uri=pending_auth.get("redirect_uri", _REDIRECT_URI),
        state=pending_auth["state"],
        code_verifier=pending_auth["code_verifier"],
    )

    try:
        # Accept partial scopes — user may deselect items in the consent screen.
        os.environ["OAUTHLIB_RELAX_TOKEN_SCOPE"] = "1"
        flow.fetch_token(code=code)
    except Exception as exc:
        print(f"ERROR: Token exchange failed: {exc}")
        print("The code may have expired. Run --auth-url to get a fresh URL.")
        sys.exit(1)

    creds = flow.credentials
    token_payload = _normalize_authorized_user_payload(json.loads(creds.to_json()))

    actually_granted = (
        list(creds.granted_scopes or [])
        if hasattr(creds, "granted_scopes") and creds.granted_scopes
        else []
    )
    if actually_granted:
        token_payload["scopes"] = actually_granted
    elif granted_scopes != SCOPES:
        token_payload["scopes"] = granted_scopes

    token_path = _token_path(email)
    _write_private_json(token_path, token_payload)
    _pending_auth_path(email).unlink(missing_ok=True)

    print(f"OK: Authenticated. Token saved to {token_path}")
    rel_label = (
        f"{display_hermes_home()}/google_chat_user_tokens/{_sanitize_email(email)}.json"
        if email
        else f"{display_hermes_home()}/google_chat_user_token.json"
    )
    print(f"Profile path: {rel_label}")


def revoke(email: Optional[str] = None) -> None:
    """Revoke the stored token with Google and delete it locally.

    Per-user when ``email`` given, legacy single-user when omitted.
    """
    token_path = _token_path(email)
    if not token_path.exists():
        print("No token to revoke.")
        return

    _ensure_deps()
    from google.oauth2.credentials import Credentials
    from google.auth.transport.requests import Request

    try:
        creds = Credentials.from_authorized_user_file(str(token_path), SCOPES)
        if creds.expired and creds.refresh_token:
            creds.refresh(Request())

        import urllib.request
        urllib.request.urlopen(
            urllib.request.Request(
                f"https://oauth2.googleapis.com/revoke?token={creds.token}",
                method="POST",
                headers={"Content-Type": "application/x-www-form-urlencoded"},
            ),
            timeout=15,
        )
        print("Token revoked with Google.")
    except Exception as exc:
        print(f"Remote revocation failed (token may already be invalid): {exc}")

    token_path.unlink(missing_ok=True)
    _pending_auth_path(email).unlink(missing_ok=True)
    print(f"Deleted {token_path}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Google Chat user-OAuth setup for Hermes (native attachment delivery)"
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--check", action="store_true",
                       help="Check if auth is valid (exit 0=yes, 1=no)")
    group.add_argument("--client-secret", metavar="PATH",
                       help="Store OAuth client_secret.json")
    group.add_argument("--auth-url", action="store_true",
                       help="Print OAuth URL for user to visit")
    group.add_argument("--auth-code", metavar="CODE",
                       help="Exchange auth code for token")
    group.add_argument("--revoke", action="store_true",
                       help="Revoke and delete stored token")
    group.add_argument("--install-deps", action="store_true",
                       help="Install Python dependencies")
    parser.add_argument("--email", metavar="EMAIL", default=None,
                       help="Scope operation to a specific user's token "
                            "(default: legacy single-user path)")
    args = parser.parse_args()

    email = args.email or None
    if args.check:
        sys.exit(0 if check_auth(email) else 1)
    elif args.client_secret:
        store_client_secret(args.client_secret)
    elif args.auth_url:
        get_auth_url(email)
    elif args.auth_code:
        exchange_auth_code(args.auth_code, email)
    elif args.revoke:
        revoke(email)
    elif args.install_deps:
        sys.exit(0 if install_deps() else 1)


if __name__ == "__main__":
    main()