af9f735abc
Adds an MCP server exposing read tools (list_inbox, search, read_email) across three personal Gmail accounts using existing app passwords in /etc/openclaw/. Wired into claude-channels via --mcp-config. Slated for replacement by an OAuth2 Gmail+Calendar server in path 2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
"""Account config loader.
|
|
|
|
Reads a JSON file (default: /etc/hara/gmail-accounts.json) listing the Gmail
|
|
accounts Hara can act on, and the path to each account's IMAP/SMTP app
|
|
password. Passwords are loaded once via `sudo -n cat` because the password
|
|
files are root:991 0640 and the MCP server process runs as `danny`. The
|
|
result is cached in memory for the process lifetime.
|
|
|
|
Schema:
|
|
{
|
|
"accounts": [
|
|
{
|
|
"email": "user@example.com",
|
|
"password_file": "/etc/openclaw/gmail-user-app-password",
|
|
"imap_host": "imap.gmail.com",
|
|
"imap_port": 993,
|
|
"smtp_host": "smtp.gmail.com",
|
|
"smtp_port": 465
|
|
}
|
|
]
|
|
}
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
DEFAULT_CONFIG_PATH = "/etc/hara/gmail-accounts.json"
|
|
# NixOS keeps the setuid sudo wrapper at /run/wrappers/bin; non-NixOS distros
|
|
# put it in /usr/bin or /bin. We try $PATH first, then fall back to these.
|
|
_SUDO_FALLBACKS = ["/run/wrappers/bin/sudo", "/usr/bin/sudo", "/bin/sudo"]
|
|
|
|
|
|
def _find_sudo() -> str:
|
|
found = shutil.which("sudo")
|
|
if found:
|
|
return found
|
|
for candidate in _SUDO_FALLBACKS:
|
|
if Path(candidate).exists():
|
|
return candidate
|
|
raise RuntimeError(
|
|
"sudo not found on PATH or in known locations; "
|
|
"cannot read group-restricted password files"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Account:
|
|
email: str
|
|
password_file: str
|
|
imap_host: str
|
|
imap_port: int
|
|
smtp_host: str
|
|
smtp_port: int
|
|
|
|
|
|
class AccountStore:
|
|
"""Holds account metadata and lazily resolves passwords on first use."""
|
|
|
|
def __init__(self, accounts: list[Account]) -> None:
|
|
self._accounts = {a.email: a for a in accounts}
|
|
self._password_cache: dict[str, str] = {}
|
|
|
|
@classmethod
|
|
def from_config_file(cls, path: str | os.PathLike[str] | None = None) -> "AccountStore":
|
|
config_path = Path(path or os.environ.get("HARA_GMAIL_CONFIG", DEFAULT_CONFIG_PATH))
|
|
with config_path.open() as f:
|
|
data = json.load(f)
|
|
accounts = [
|
|
Account(
|
|
email=a["email"],
|
|
password_file=a["password_file"],
|
|
imap_host=a.get("imap_host", "imap.gmail.com"),
|
|
imap_port=int(a.get("imap_port", 993)),
|
|
smtp_host=a.get("smtp_host", "smtp.gmail.com"),
|
|
smtp_port=int(a.get("smtp_port", 465)),
|
|
)
|
|
for a in data.get("accounts", [])
|
|
]
|
|
return cls(accounts)
|
|
|
|
def emails(self) -> list[str]:
|
|
return list(self._accounts.keys())
|
|
|
|
def get(self, email: str) -> Account:
|
|
try:
|
|
return self._accounts[email]
|
|
except KeyError:
|
|
raise ValueError(f"Unknown account: {email!r}. Configured: {self.emails()}")
|
|
|
|
def password_for(self, email: str) -> str:
|
|
if email in self._password_cache:
|
|
return self._password_cache[email]
|
|
account = self.get(email)
|
|
# Prefer direct read if the file is reachable (e.g. after path 2
|
|
# migration where the daemon owns its own creds), fall back to
|
|
# `sudo -n cat` for the current /etc/openclaw/ layout.
|
|
try:
|
|
value = Path(account.password_file).read_text().strip()
|
|
except PermissionError:
|
|
value = subprocess.check_output(
|
|
[_find_sudo(), "-n", "cat", account.password_file],
|
|
text=True,
|
|
).strip()
|
|
if not value:
|
|
raise RuntimeError(f"Empty password file for {email}: {account.password_file}")
|
|
self._password_cache[email] = value
|
|
return value
|