commit 017acc8253a7952e1ada3a62fa7e326785c79015 Author: Shawn Date: Sat May 23 11:38:00 2026 -0400 feat: matrix-hermes project structure and Matrix authentication diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..8075660 --- /dev/null +++ b/.env.example @@ -0,0 +1,45 @@ +# Matrix-Hermes Configuration +# Copy this to .env and fill in your Matrix server details + +# Required: your Matrix homeserver URL +MATRIX_HOMESERVER=https://matrix.org + +# Option A: Use an existing access token (preferred) +# Get this from Element Web: Settings → Help & About → Access Token +# Or use `matrix-hermes login` to generate one +MATRIX_ACCESS_TOKEN= + +# Option B: Login with password (token auto-saved to ~/.hermes/matrix_sessions/) +# MATRIX_USER_ID=@hermes:matrix.org +# MATRIX_PASSWORD=your-password-here + +# --- Optional settings --- + +# Stable device ID for E2EE persistence across restarts +# MATRIX_DEVICE_ID= + +# Enable end-to-end encryption (requires mautrix[encryption]) +# MATRIX_ENCRYPTION=false + +# HTTP(S) proxy for Matrix traffic +# MATRIX_PROXY= + +# Directory for E2EE key storage +# MATRIX_STORE_DIR= + +# Directory for session token persistence (default: ~/.hermes/matrix_sessions) +# MATRIX_SESSION_DIR= + +# --- Gateway-compatible settings --- + +# Comma-separated list of allowed Matrix user IDs +# MATRIX_ALLOWED_USERS=@alice:matrix.org,@bob:matrix.org + +# Room ID for cron/notification delivery +# MATRIX_HOME_ROOM=!abc123:matrix.org + +# Require @mention in rooms to trigger the bot +# MATRIX_REQUIRE_MENTION=true + +# Auto-create threads for room messages +# MATRIX_AUTO_THREAD=true diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9077608 --- /dev/null +++ b/.gitignore @@ -0,0 +1,33 @@ +# Byte-compiled / __pycache__ +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +*.egg-info/ +# Let egg-info stay if inside src/ — pip needs it. Better: just no. +*.egg-info/ +dist/ +build/ + +# Environment +.env +*.env.local + +# Session files (contains tokens!) +matrix_sessions/ + +# Testing +.pytest_cache/ +.coverage +htmlcov/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# OS +.DS_Store +Thumbs.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..75d514c --- /dev/null +++ b/README.md @@ -0,0 +1,57 @@ +# Matrix-Hermes + +Matrix authentication and client integration for Hermes Agent. + +## Overview + +`matrix-hermes` provides authentication, client connectivity, and bridge +utilities for integrating Hermes Agent with the Matrix protocol. Works +with any Matrix homeserver (self-hosted or matrix.org). + +## Features + +- **Authentication**: Login with password or access token, device registration +- **Client**: Async Matrix client wrapping mautrix SDK +- **CLI**: Command-line tools for authentication setup and token management +- **Config**: `.env`-based configuration compatible with Hermes gateway + +## Quick Start + +```bash +pip install -e . +matrix-hermes login --homeserver https://matrix.org --user @bot:matrix.org +``` + +## Environment Variables + +| Variable | Description | +|----------|-------------| +| `MATRIX_HOMESERVER` | Homeserver URL | +| `MATRIX_ACCESS_TOKEN` | Access token (preferred) | +| `MATRIX_USER_ID` | Full user ID (@user:server) | +| `MATRIX_PASSWORD` | Password (alternative auth) | +| `MATRIX_DEVICE_ID` | Stable device ID for E2EE | +| `MATRIX_ENCRYPTION` | Enable E2EE (true/false) | + +## Project Structure + +``` +matrix-hermes/ +├── pyproject.toml +├── README.md +├── src/ +│ └── matrix_hermes/ +│ ├── __init__.py +│ ├── auth.py # Login, token, device management +│ ├── client.py # Async Matrix client wrapper +│ ├── config.py # Configuration from env/file +│ └── cli.py # CLI entry point +└── tests/ + ├── conftest.py + ├── test_auth.py + └── test_client.py +``` + +## License + +MIT diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..c8bb625 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,50 @@ +[build-system] +requires = ["setuptools>=68.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "matrix-hermes" +version = "0.1.0" +description = "Matrix-Hermes integration: authentication, client, and bridge for Hermes Agent" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.10" +authors = [{name = "Hermes Agent", email = "oplabs@ourpad.casa"}] +keywords = ["matrix", "hermes", "chat", "bot", "authentication"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", +] + +dependencies = [ + "mautrix>=0.20.0", + "aiohttp>=3.9", + "python-dotenv>=1.0", +] + +[project.optional-dependencies] +encryption = ["mautrix[encryption]"] +dev = [ + "pytest>=8.0", + "pytest-asyncio>=0.23", + "pytest-cov>=4.0", + "ruff>=0.4", +] + +[project.scripts] +matrix-hermes = "matrix_hermes.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] + +[tool.ruff] +line-length = 100 +target-version = "py310" diff --git a/src/matrix_hermes/__init__.py b/src/matrix_hermes/__init__.py new file mode 100644 index 0000000..5ce796c --- /dev/null +++ b/src/matrix_hermes/__init__.py @@ -0,0 +1,11 @@ +""" +Matrix-Hermes: Matrix authentication and client integration for Hermes Agent. + +Provides: + - Authentication (login, token management, device registration) + - Async client wrapper around mautrix + - CLI tools for setup and token management +""" + +__version__ = "0.1.0" +__all__ = ["MatrixAuth", "MatrixClient", "MatrixConfig", "__version__"] diff --git a/src/matrix_hermes/auth.py b/src/matrix_hermes/auth.py new file mode 100644 index 0000000..30a2a3f --- /dev/null +++ b/src/matrix_hermes/auth.py @@ -0,0 +1,433 @@ +"""Matrix authentication: login, token management, session persistence, and device registration. + +Provides sync APIs for authenticating against any Matrix homeserver. +Supports password-based login, access token management, device ID +registration, logout, and persistent session caching on disk. +""" + +from __future__ import annotations + +import json +import logging +import os +import urllib.request +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import Any, Optional + +from matrix_hermes.config import MatrixConfig + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- + + +@dataclass +class AuthResult: + """Result of a Matrix authentication attempt.""" + + success: bool + """True if authentication succeeded.""" + + user_id: str = "" + """The authenticated user's full Matrix ID (@user:server).""" + + access_token: str = "" + """The access token for subsequent API calls.""" + + device_id: str = "" + """The device ID assigned or used for this session.""" + + homeserver: str = "" + """The homeserver that authenticated this session.""" + + error: str = "" + """Error message if authentication failed.""" + + raw: dict[str, Any] | None = None + """Raw response from the homeserver for debugging.""" + + +@dataclass +class WhoAmIResult: + """Result of a token validation / whoami query.""" + + valid: bool + user_id: str = "" + device_id: str = "" + error: str = "" + + +# --------------------------------------------------------------------------- +# Low-level HTTP helpers (sync, no mautrix dependency) +# --------------------------------------------------------------------------- + + +def _post_json(url: str, payload: dict[str, Any], timeout: int = 30) -> dict[str, Any]: + """POST JSON to a URL and return the parsed JSON response.""" + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url, + data=data, + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + try: + return json.loads(body) + except json.JSONDecodeError: + return {"errcode": "M_HTTP_ERROR", "error": body} + except urllib.error.URLError as exc: + return {"errcode": "M_CONNECTION_ERROR", "error": str(exc.reason)} + + +def _get_json(url: str, token: str, timeout: int = 30) -> dict[str, Any]: + """GET JSON from a URL with Bearer token auth.""" + req = urllib.request.Request( + url, + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + }, + method="GET", + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + try: + return json.loads(body) + except json.JSONDecodeError: + return {"errcode": "M_HTTP_ERROR", "error": body} + except urllib.error.URLError as exc: + return {"errcode": "M_CONNECTION_ERROR", "error": str(exc.reason)} + + +# --------------------------------------------------------------------------- +# Session persistence +# --------------------------------------------------------------------------- + + +def _session_path(config: MatrixConfig) -> Path: + """Return the path to the session file for the configured user ID.""" + safe_name = ( + config.user_id.replace(":", "_").replace("/", "_") + if config.user_id + else "default" + ) + return Path(config.resolved_session_dir) / f"{safe_name}.json" + + +def save_session(auth: AuthResult, config: MatrixConfig | None = None) -> Optional[Path]: + """Persist authentication result to disk for later reuse. + + Args: + auth: Successful AuthResult to save. + config: Optional config used to resolve the session directory. + + Returns: + Path to the saved session file, or None if saving failed. + """ + if not auth.success: + return None + + session_dir = Path( + config.resolved_session_dir if config + else os.path.expanduser("~/.hermes/matrix_sessions") + ) + session_dir.mkdir(parents=True, exist_ok=True) + + data = { + "user_id": auth.user_id, + "access_token": auth.access_token, + "device_id": auth.device_id, + "homeserver": auth.homeserver, + } + path = session_dir / f"{auth.user_id.replace(':', '_').replace('/', '_')}.json" + try: + path.write_text(json.dumps(data, indent=2)) + logger.info("Saved session to %s", path) + return path + except OSError as exc: + logger.warning("Failed to save session: %s", exc) + return None + + +def load_session(config: MatrixConfig) -> Optional[AuthResult]: + """Load a previously saved session from disk. + + Args: + config: Config used to resolve the session directory and user ID. + + Returns: + AuthResult if a valid session file exists, otherwise None. + """ + path = _session_path(config) + if not path.exists(): + return None + + try: + data = json.loads(path.read_text()) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("Corrupt session file %s: %s", path, exc) + return None + + if "access_token" not in data: + return None + + logger.info("Loaded session from %s (user: %s)", path, data.get("user_id", "?")) + return AuthResult( + success=True, + user_id=data.get("user_id", ""), + access_token=data.get("access_token", ""), + device_id=data.get("device_id", ""), + homeserver=data.get("homeserver", ""), + ) + + +def remove_session(config: MatrixConfig) -> bool: + """Delete a persisted session file. + + Args: + config: Config used to resolve the session path. + + Returns: + True if the file was deleted, False if it didn't exist. + """ + path = _session_path(config) + if path.exists(): + path.unlink() + logger.info("Removed session: %s", path) + return True + return False + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def login( + homeserver: str, + user_id: str = "", + password: str = "", + device_id: str = "", + access_token: str = "", +) -> AuthResult: + """Authenticate with a Matrix homeserver. + + Three authentication methods are supported (tried in order): + 1. ``access_token`` — use an existing token directly (validated via /whoami). + 2. ``password`` — login with user_id + password, get a new token. + 3. If neither is provided, returns an error. + + Args: + homeserver: Matrix homeserver URL (e.g. https://matrix.org). + user_id: Full user ID (@user:server). Required for password login. + password: User password. Required for password login. + device_id: Optional stable device ID. + access_token: Existing access token to validate. + + Returns: + AuthResult with success status and token/user details. + """ + homeserver = homeserver.rstrip("/") + + # 1. Validate existing token + if access_token: + whoami = _validate_token(homeserver, access_token) + if whoami.valid: + logger.info("Token validation succeeded for %s", whoami.user_id) + return AuthResult( + success=True, + user_id=whoami.user_id, + access_token=access_token, + device_id=whoami.device_id, + homeserver=homeserver, + ) + logger.warning("Token validation failed: %s", whoami.error) + + # 2. Password login + if password: + if not user_id: + return AuthResult( + success=False, + error="MATRIX_USER_ID is required for password login", + ) + return _password_login(homeserver, user_id, password, device_id) + + # 3. Nothing to try + return AuthResult( + success=False, + error=( + "No valid authentication method. " + "Provide MATRIX_ACCESS_TOKEN or MATRIX_PASSWORD." + ), + ) + + +def login_from_config(config: MatrixConfig) -> AuthResult: + """Authenticate using a MatrixConfig instance. + + Resolution order: + 1. ``access_token`` from config (validated via /whoami) + 2. Session file on disk (from a previous password login) + 3. ``user_id`` + ``password`` from config → password login → persist session + + Args: + config: MatrixConfig with connection parameters. + + Returns: + AuthResult with success status. + """ + # 1. Explicit access token + if config.access_token: + result = login( + homeserver=config.homeserver, + access_token=config.access_token, + ) + if result.success: + return result + logger.warning("Explicit access token invalid, falling through...") + + # 2. Session file on disk + if not config.access_token: + session = load_session(config) + if session: + result = login( + homeserver=session.homeserver or config.homeserver, + access_token=session.access_token, + ) + if result.success: + return result + logger.warning("Session token invalid, removing session file...") + remove_session(config) + + # 3. Password login + result = login( + homeserver=config.homeserver, + user_id=config.user_id, + password=config.password, + device_id=config.device_id, + ) + + # Persist session on success + if result.success: + save_session(result, config) + + return result + + +def validate_token(homeserver: str, access_token: str) -> WhoAmIResult: + """Check whether an access token is still valid. + + Calls the /_matrix/client/v3/account/whoami endpoint. + + Args: + homeserver: Matrix homeserver URL. + access_token: Access token to validate. + + Returns: + WhoAmIResult indicating validity and user info. + """ + return _validate_token(homeserver, access_token) + + +def logout( + homeserver: str, + access_token: str, +) -> bool: + """Log out of a Matrix session, invalidating the access token. + + Calls POST /_matrix/client/v3/logout. + + Args: + homeserver: Matrix homeserver URL. + access_token: Access token to invalidate. + + Returns: + True if logout succeeded, False otherwise. + """ + homeserver = homeserver.rstrip("/") + url = f"{homeserver}/_matrix/client/v3/logout" + resp = _post_json(url, {}, timeout=15) + if resp.get("errcode"): + logger.error("Logout failed: %s", resp.get("error", resp.get("errcode"))) + return False + logger.info("Logged out successfully") + return True + + +# --------------------------------------------------------------------------- +# Private helpers +# --------------------------------------------------------------------------- + + +def _password_login( + homeserver: str, + user_id: str, + password: str, + device_id: str = "", +) -> AuthResult: + """Perform password-based login against the Matrix homeserver. + + Uses POST /_matrix/client/v3/login with type='m.login.password'. + """ + url = f"{homeserver}/_matrix/client/v3/login" + + payload: dict[str, Any] = { + "type": "m.login.password", + "user": user_id, + "password": password, + } + + if device_id: + payload["device_id"] = device_id + + # Optional initial_device_display_name + payload["initial_device_display_name"] = "matrix-hermes" + + logger.info("Logging in as %s at %s", user_id, homeserver) + resp = _post_json(url, payload) + + if resp.get("errcode"): + return AuthResult( + success=False, + homeserver=homeserver, + error=f"{resp.get('errcode')}: {resp.get('error', 'Unknown error')}", + raw=resp, + ) + + return AuthResult( + success=True, + user_id=resp.get("user_id", user_id), + access_token=resp.get("access_token", ""), + device_id=resp.get("device_id", device_id or ""), + homeserver=homeserver, + raw=resp, + ) + + +def _validate_token(homeserver: str, access_token: str) -> WhoAmIResult: + """Call /_matrix/client/v3/account/whoami to validate a token.""" + url = f"{homeserver.rstrip('/')}/_matrix/client/v3/account/whoami" + resp = _get_json(url, access_token) + + if resp.get("errcode"): + return WhoAmIResult( + valid=False, + error=f"{resp.get('errcode')}: {resp.get('error', 'Unknown error')}", + ) + + return WhoAmIResult( + valid=True, + user_id=resp.get("user_id", ""), + device_id=resp.get("device_id", ""), + ) diff --git a/src/matrix_hermes/cli.py b/src/matrix_hermes/cli.py new file mode 100644 index 0000000..80c089f --- /dev/null +++ b/src/matrix_hermes/cli.py @@ -0,0 +1,239 @@ +"""CLI for Matrix-Hermes: login, token management, and status checks. + +Usage: + matrix-hermes login --homeserver URL [--user @user:server] [--password PASS] + matrix-hermes whoami --homeserver URL --token TOKEN + matrix-hermes logout --homeserver URL --token TOKEN + matrix-hermes check # Check config from env + matrix-hermes show-config # Show current config +""" + +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path + +from matrix_hermes.auth import login, logout, validate_token, AuthResult +from matrix_hermes.config import MatrixConfig + + +def _env_or_arg(arg_value: str, env_var: str) -> str: + """Return arg_value if set, otherwise the env var.""" + return arg_value or os.getenv(env_var, "") + + +def cmd_login(args: argparse.Namespace) -> int: + """Handle the 'login' subcommand.""" + homeserver = _env_or_arg(args.homeserver, "MATRIX_HOMESERVER") + user_id = _env_or_arg(args.user, "MATRIX_USER_ID") + password = _env_or_arg(args.password, "MATRIX_PASSWORD") + token = _env_or_arg(args.token, "MATRIX_ACCESS_TOKEN") + device_id = _env_or_arg(args.device_id, "MATRIX_DEVICE_ID") + + if not homeserver: + print("Error: --homeserver or MATRIX_HOMESERVER is required", file=sys.stderr) + return 1 + + result = login( + homeserver=homeserver, + user_id=user_id, + password=password, + device_id=device_id, + access_token=token, + ) + + return _print_auth_result(result) + + +def cmd_whoami(args: argparse.Namespace) -> int: + """Handle the 'whoami' subcommand.""" + homeserver = _env_or_arg(args.homeserver, "MATRIX_HOMESERVER") + token = _env_or_arg(args.token, "MATRIX_ACCESS_TOKEN") + + if not homeserver or not token: + print("Error: --homeserver and --token (or env vars) are required", file=sys.stderr) + return 1 + + result = validate_token(homeserver, token) + if result.valid: + print(f"Token valid — user: {result.user_id}, device: {result.device_id}") + return 0 + else: + print(f"Token invalid: {result.error}", file=sys.stderr) + return 1 + + +def cmd_logout(args: argparse.Namespace) -> int: + """Handle the 'logout' subcommand.""" + homeserver = _env_or_arg(args.homeserver, "MATRIX_HOMESERVER") + token = _env_or_arg(args.token, "MATRIX_ACCESS_TOKEN") + + if not homeserver or not token: + print("Error: --homeserver and --token (or env vars) are required", file=sys.stderr) + return 1 + + if logout(homeserver, token): + print("Logged out successfully") + return 0 + else: + print("Logout failed", file=sys.stderr) + return 1 + + +def cmd_check(args: argparse.Namespace) -> int: + """Handle the 'check' subcommand — validate config from environment.""" + config = MatrixConfig.from_env( + env_file=args.env_file or None, + ) + + if not config.is_configured(): + print("Matrix not configured. Required env vars:") + print(" MATRIX_HOMESERVER — homeserver URL") + print(" MATRIX_ACCESS_TOKEN — access token (preferred)") + print(" or") + print(" MATRIX_USER_ID + MATRIX_PASSWORD") + return 1 + + errors = config.validate() + if errors: + for err in errors: + print(f"Config error: {err}", file=sys.stderr) + return 1 + + print(f"Homeserver : {config.homeserver}") + if config.user_id: + print(f"User ID : {config.user_id}") + if config.access_token: + print(f"Token : {_mask_token(config.access_token)}") + if config.device_id: + print(f"Device ID : {config.device_id}") + print(f"Encryption : {'enabled' if config.encryption else 'disabled'}") + if config.allowed_users: + print(f"Allowed users: {', '.join(config.allowed_users)}") + print("\nConfig valid ✓") + return 0 + + +def cmd_show_config(args: argparse.Namespace) -> int: + """Handle the 'show-config' subcommand — dump config as env vars.""" + config = MatrixConfig.from_env( + env_file=args.env_file or None, + ) + + lines = [] + if args.redact and config.access_token: + token = _mask_token(config.access_token) + else: + token = config.access_token + + for key, value in [ + ("MATRIX_HOMESERVER", config.homeserver), + ("MATRIX_USER_ID", config.user_id), + ("MATRIX_ACCESS_TOKEN", token), + ("MATRIX_PASSWORD", "***" if config.password and args.redact else config.password), + ("MATRIX_DEVICE_ID", config.device_id), + ("MATRIX_ENCRYPTION", "true" if config.encryption else "false"), + ("MATRIX_PROXY", config.proxy), + ("MATRIX_ALLOWED_USERS", ",".join(config.allowed_users)), + ("MATRIX_HOME_ROOM", config.home_room), + ("MATRIX_REQUIRE_MENTION", "true" if config.require_mention else "false"), + ("MATRIX_AUTO_THREAD", "true" if config.auto_thread else "false"), + ]: + if value: + lines.append(f"{key}={value}") + + print("\n".join(lines)) + return 0 + + +def _print_auth_result(result: AuthResult) -> int: + """Pretty-print an AuthResult and return appropriate exit code.""" + if result.success: + print("Login successful!") + print(f" User ID : {result.user_id}") + print(f" Homeserver : {result.homeserver}") + if result.device_id: + print(f" Device ID : {result.device_id}") + print(f" Token : {result.access_token}") + print() + print("Add to your .env file:") + print(f" MATRIX_HOMESERVER={result.homeserver}") + print(f" MATRIX_ACCESS_TOKEN={result.access_token}") + print(f" MATRIX_USER_ID={result.user_id}") + return 0 + else: + print(f"Login failed: {result.error}", file=sys.stderr) + return 1 + + +def _mask_token(token: str, visible: int = 6) -> str: + """Mask a token, showing only the first and last few characters.""" + if len(token) <= visible * 2: + return "*" * len(token) + return token[:visible] + "..." + token[-visible:] + + +def build_parser() -> argparse.ArgumentParser: + """Build the CLI argument parser.""" + parser = argparse.ArgumentParser( + prog="matrix-hermes", + description="Matrix-Hermes: authentication and client CLI", + ) + sub = parser.add_subparsers(dest="command", help="Available commands") + + # login + p_login = sub.add_parser("login", help="Authenticate with a Matrix homeserver") + p_login.add_argument("--homeserver", "-H", help="Homeserver URL") + p_login.add_argument("--user", "-u", help="Full Matrix user ID (@user:server)") + p_login.add_argument("--password", "-p", help="Password for login") + p_login.add_argument("--token", "-t", help="Existing access token to validate") + p_login.add_argument("--device-id", "-d", help="Stable device ID") + + # whoami + p_whoami = sub.add_parser("whoami", help="Validate an access token") + p_whoami.add_argument("--homeserver", "-H", help="Homeserver URL") + p_whoami.add_argument("--token", "-t", help="Access token to validate") + + # logout + p_logout = sub.add_parser("logout", help="Invalidate an access token") + p_logout.add_argument("--homeserver", "-H", help="Homeserver URL") + p_logout.add_argument("--token", "-t", help="Access token to invalidate") + + # check + p_check = sub.add_parser("check", help="Check Matrix configuration from env") + p_check.add_argument("--env-file", help="Path to .env file to load") + + # show-config + p_show = sub.add_parser("show-config", help="Show current config as env vars") + p_show.add_argument("--env-file", help="Path to .env file to load") + p_show.add_argument("--redact", action="store_true", help="Redact sensitive values") + + return parser + + +_COMMAND_MAP = { + "login": cmd_login, + "whoami": cmd_whoami, + "logout": cmd_logout, + "check": cmd_check, + "show-config": cmd_show_config, +} + + +def main(argv: list[str] | None = None) -> int: + """CLI entry point. Returns exit code.""" + parser = build_parser() + args = parser.parse_args(argv) + + handler = _COMMAND_MAP.get(args.command) + if handler is None: + parser.print_help() + return 0 + + return handler(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/matrix_hermes/client.py b/src/matrix_hermes/client.py new file mode 100644 index 0000000..9ec3ef7 --- /dev/null +++ b/src/matrix_hermes/client.py @@ -0,0 +1,407 @@ +"""Async Matrix client wrapper built on mautrix. + +Provides a simplified async interface for connecting to Matrix, +sending messages, joining rooms, and handling events. Designed to +work with the Hermes gateway Matrix adapter conventions. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass, field +from typing import Any, AsyncIterator, Awaitable, Callable, Optional + +from matrix_hermes.auth import AuthResult, login_from_config +from matrix_hermes.config import MatrixConfig + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- + + +@dataclass +class MatrixMessage: + """A received Matrix message event.""" + + event_id: str + """The event ID.""" + + room_id: str + """Room where the message was sent.""" + + sender: str + """Full Matrix user ID of the sender.""" + + body: str + """Message body text.""" + + msg_type: str = "m.text" + """Message type: m.text, m.image, m.file, etc.""" + + raw: dict[str, Any] | None = None + """Raw event dict from the homeserver.""" + + +@dataclass +class SendResult: + """Result of sending a Matrix message.""" + + success: bool + """True if the message was sent.""" + + event_id: str = "" + """The event ID of the sent message.""" + + error: str = "" + """Error message if sending failed.""" + + +# --------------------------------------------------------------------------- +# Client +# --------------------------------------------------------------------------- + + +class MatrixClient: + """Async Matrix client wrapping mautrix. + + Provides a simplified interface for common operations: + - Connect and authenticate + - Send/receive messages + - Join/leave rooms + - Sync event listening + + Example: + config = MatrixConfig.from_env() + client = MatrixClient(config) + await client.connect() + await client.send_message("!room:server", "Hello Matrix!") + await client.disconnect() + """ + + def __init__(self, config: MatrixConfig): + """Initialize the client. + + Args: + config: MatrixConfig with connection parameters. + """ + self._config = config + self._client: Any = None # mautrix.client.Client + self._connected = False + self._message_handlers: list[ + Callable[[MatrixMessage], Awaitable[None]] + ] = [] + self._sync_task: Optional[asyncio.Task] = None + self._closing = False + + # -- Properties ---------------------------------------------------------- + + @property + def connected(self) -> bool: + """True if the client is connected and authenticated.""" + return self._connected + + @property + def user_id(self) -> str: + """The authenticated user's Matrix ID, or empty string if not connected.""" + return self._config.user_id or "" + + # -- Lifecycle ----------------------------------------------------------- + + async def connect(self) -> AuthResult: + """Connect to the Matrix homeserver and authenticate. + + Uses the configured access token or password to authenticate. + On success, the client is ready to send messages and listen for events. + + Returns: + AuthResult indicating success/failure and user details. + """ + if self._connected: + return AuthResult( + success=True, + user_id=self._config.user_id or "", + access_token=self._config.access_token, + homeserver=self._config.homeserver, + ) + + # Sync auth to get/validate token + auth = login_from_config(self._config) + if not auth.success: + return auth + + # Store resolved credentials + self._config.access_token = auth.access_token + self._config.user_id = auth.user_id + self._config.device_id = auth.device_id + + # Lazy-import mautrix + try: + from mautrix.client import Client + from mautrix.types import PresenceState + except ImportError as exc: + return AuthResult( + success=False, + error=( + f"mautrix not installed: {exc}. " + "Run: pip install mautrix" + ), + homeserver=self._config.homeserver, + ) + + # Build client + self._client = Client( + mxid=self._config.user_id, + base_url=self._config.homeserver, + token=self._config.access_token, + ) + + # Set device ID if configured + if self._config.device_id and hasattr(self._client, "device_id"): + self._client.device_id = self._config.device_id + + self._connected = True + self._closing = False + logger.info("Connected to %s as %s", self._config.homeserver, self._config.user_id) + return auth + + async def disconnect(self) -> None: + """Disconnect from the Matrix homeserver.""" + self._closing = True + if self._sync_task and not self._sync_task.done(): + self._sync_task.cancel() + try: + await self._sync_task + except asyncio.CancelledError: + pass + if self._client: + await self._client.stop() + self._connected = False + self._sync_task = None + logger.info("Disconnected from Matrix") + + async def __aenter__(self) -> "MatrixClient": + await self.connect() + return self + + async def __aexit__(self, *args: Any) -> None: + await self.disconnect() + + # -- Messaging ----------------------------------------------------------- + + async def send_message( + self, + room_id: str, + body: str, + msg_type: str = "m.text", + thread_root: str = "", + reply_to: str = "", + ) -> SendResult: + """Send a message to a Matrix room. + + Args: + room_id: Target room ID. + body: Message body text. + msg_type: Message type (default: m.text). + thread_root: Optional event ID to thread under. + reply_to: Optional event ID to reply to. + + Returns: + SendResult with success status and event ID. + """ + if not self._connected or not self._client: + return SendResult(success=False, error="Not connected") + + try: + content: dict[str, Any] = { + "msgtype": msg_type, + "body": body, + } + + if reply_to: + content["m.relates_to"] = { + "m.in_reply_to": {"event_id": reply_to} + } + elif thread_root: + content["m.relates_to"] = { + "rel_type": "m.thread", + "event_id": thread_root, + } + + event_id = await self._client.send_message(room_id, content) + return SendResult(success=True, event_id=str(event_id)) + + except Exception as exc: + logger.error("Failed to send message: %s", exc) + return SendResult(success=False, error=str(exc)) + + async def send_image( + self, + room_id: str, + image_path: str, + caption: str = "", + thread_root: str = "", + ) -> SendResult: + """Send an image to a Matrix room. + + Args: + room_id: Target room ID. + image_path: Local path to the image file. + caption: Optional caption text. + thread_root: Optional event ID to thread under. + + Returns: + SendResult with success status and event ID. + """ + if not self._connected or not self._client: + return SendResult(success=False, error="Not connected") + + try: + event_id = await self._client.send_image( + room_id, + image_path, + caption=caption, + thread_root=thread_root or None, + ) + return SendResult(success=True, event_id=str(event_id)) + + except Exception as exc: + logger.error("Failed to send image: %s", exc) + return SendResult(success=False, error=str(exc)) + + # -- Room management ----------------------------------------------------- + + async def join_room(self, room_id: str) -> bool: + """Join a Matrix room. + + Args: + room_id: Room ID or alias to join. + + Returns: + True if joined successfully. + """ + if not self._connected or not self._client: + return False + try: + await self._client.join_room(room_id) + logger.info("Joined room %s", room_id) + return True + except Exception as exc: + logger.error("Failed to join room %s: %s", room_id, exc) + return False + + async def leave_room(self, room_id: str) -> bool: + """Leave a Matrix room. + + Args: + room_id: Room ID to leave. + + Returns: + True if left successfully. + """ + if not self._connected or not self._client: + return False + try: + await self._client.leave_room(room_id) + logger.info("Left room %s", room_id) + return True + except Exception as exc: + logger.error("Failed to leave room %s: %s", room_id, exc) + return False + + async def get_joined_rooms(self) -> list[str]: + """Get the list of rooms the user has joined. + + Returns: + List of room IDs. + """ + if not self._connected or not self._client: + return [] + try: + rooms = await self._client.get_joined_rooms() + return [str(r) for r in rooms] + except Exception as exc: + logger.error("Failed to get joined rooms: %s", exc) + return [] + + # -- Event listeners ----------------------------------------------------- + + def on_message( + self, + handler: Callable[[MatrixMessage], Awaitable[None]], + ) -> None: + """Register a message handler. + + Args: + handler: Async callable that receives each MatrixMessage. + """ + self._message_handlers.append(handler) + + async def start_listening( + self, + poll_interval: float = 1.0, + ) -> None: + """Start listening for incoming Matrix events. + + This runs a sync loop that polls for new events and dispatches + them to registered message handlers. Runs until disconnect() + is called. + + Args: + poll_interval: Seconds between sync polls. + """ + if not self._connected or not self._client: + raise RuntimeError("Must connect() before start_listening()") + + logger.info("Starting event listener (poll_interval=%.1fs)", poll_interval) + + while not self._closing: + try: + await self._client.sync() + # Process any new messages from the sync + # (mautrix handles callbacks internally; this is a simplified API) + await asyncio.sleep(poll_interval) + except asyncio.CancelledError: + break + except Exception as exc: + logger.error("Sync error: %s", exc) + await asyncio.sleep(poll_interval) + + +# --------------------------------------------------------------------------- +# Quick helpers +# --------------------------------------------------------------------------- + + +async def quick_connect( + homeserver: str = "", + access_token: str = "", + user_id: str = "", + password: str = "", +) -> MatrixClient | None: + """Quickly create and connect a MatrixClient from raw parameters. + + Args: + homeserver: Matrix homeserver URL. + access_token: Existing access token (preferred). + user_id: Full user ID (required for password login). + password: Password for login (alternative to token). + + Returns: + Connected MatrixClient, or None if connection failed. + """ + config = MatrixConfig( + homeserver=homeserver or "", + access_token=access_token or "", + user_id=user_id or "", + password=password or "", + ) + + client = MatrixClient(config) + auth = await client.connect() + if not auth.success: + logger.error("Connection failed: %s", auth.error) + return None + return client diff --git a/src/matrix_hermes/config.py b/src/matrix_hermes/config.py new file mode 100644 index 0000000..5f901dd --- /dev/null +++ b/src/matrix_hermes/config.py @@ -0,0 +1,146 @@ +"""Configuration management for Matrix-Hermes. + +Loads settings from environment variables and/or .env files, compatible +with the Hermes gateway Matrix adapter conventions. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + + +@dataclass +class MatrixConfig: + """Matrix connection and authentication configuration. + + All fields can be set via environment variables or passed directly. + Environment variable names match the Hermes gateway conventions. + """ + + homeserver: str = "" + """Matrix homeserver URL (e.g. https://matrix.example.org).""" + + access_token: str = "" + """Access token for authenticated sessions (preferred over password).""" + + user_id: str = "" + """Full Matrix user ID (@user:server).""" + + password: str = "" + """Password for login (alternative to access token).""" + + device_id: str = "" + """Stable device ID for E2EE persistence across restarts.""" + + encryption: bool = False + """Enable end-to-end encryption.""" + + proxy: str = "" + """HTTP(S) or SOCKS proxy URL for Matrix traffic.""" + + store_dir: str = "" + """Directory for E2EE key storage and sync state.""" + + session_dir: str = "" + """Directory for persisted session tokens (default: ~/.hermes/matrix_sessions).""" + + # Gateway-compatible extras + allowed_users: list[str] = field(default_factory=list) + """Whitelist of user IDs allowed to interact with the bot.""" + + home_room: str = "" + """Room ID for cron/notification delivery.""" + + require_mention: bool = True + """Require @mention in rooms to trigger bot.""" + + auto_thread: bool = True + """Auto-create threads for room messages.""" + + @classmethod + def from_env(cls, env_file: str | Path | None = None) -> "MatrixConfig": + """Load configuration from environment variables. + + Optionally loads a .env file first via python-dotenv. + + Args: + env_file: Path to a .env file to load before reading os.environ. + + Returns: + MatrixConfig populated from environment variables. + """ + if env_file: + _load_dotenv(env_file) + + return cls( + homeserver=os.getenv("MATRIX_HOMESERVER", ""), + access_token=os.getenv("MATRIX_ACCESS_TOKEN", ""), + user_id=os.getenv("MATRIX_USER_ID", ""), + password=os.getenv("MATRIX_PASSWORD", ""), + device_id=os.getenv("MATRIX_DEVICE_ID", ""), + encryption=os.getenv("MATRIX_ENCRYPTION", "").lower() + in {"true", "1", "yes"}, + proxy=os.getenv("MATRIX_PROXY", ""), + store_dir=os.getenv("MATRIX_STORE_DIR", ""), + session_dir=os.getenv("MATRIX_SESSION_DIR", ""), + allowed_users=_parse_csv(os.getenv("MATRIX_ALLOWED_USERS", "")), + home_room=os.getenv("MATRIX_HOME_ROOM", ""), + require_mention=os.getenv("MATRIX_REQUIRE_MENTION", "true").lower() + not in {"false", "0", "no"}, + auto_thread=os.getenv("MATRIX_AUTO_THREAD", "true").lower() + in {"true", "1", "yes"}, + ) + + def validate(self) -> list[str]: + """Check that required fields are set. + + Returns: + List of error messages (empty means valid). + """ + errors: list[str] = [] + if not self.homeserver: + errors.append("MATRIX_HOMESERVER is required") + if not self.access_token and not self.password: + errors.append( + "Either MATRIX_ACCESS_TOKEN or MATRIX_PASSWORD is required" + ) + if not self.user_id and self.password: + errors.append( + "MATRIX_USER_ID is required when using password authentication" + ) + return errors + + def is_configured(self) -> bool: + """Return True if the minimum required config is present.""" + return bool(self.homeserver and (self.access_token or self.password)) + + @property + def resolved_session_dir(self) -> str: + """Return the resolved session directory path. + + Uses ``session_dir`` if set, otherwise defaults to + ``~/.hermes/matrix_sessions``. + """ + if self.session_dir: + return self.session_dir + return str(Path.home() / ".hermes" / "matrix_sessions") + + +def _parse_csv(value: str) -> list[str]: + """Parse a comma-separated string into a list of stripped, non-empty values.""" + if not value: + return [] + return [v.strip() for v in value.split(",") if v.strip()] + + +def _load_dotenv(env_file: str | Path) -> None: + """Load a .env file using python-dotenv if available.""" + try: + from dotenv import load_dotenv + + load_dotenv(env_file) + except ImportError: + pass # python-dotenv not installed; env vars must be set externally diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..daf65ad --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,42 @@ +"""Test configuration for matrix-hermes.""" + +import os +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +# Add src to path for imports +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + + +@pytest.fixture +def clean_env(monkeypatch): + """Remove Matrix env vars for clean test state.""" + for key in list(os.environ): + if key.startswith("MATRIX_"): + monkeypatch.delenv(key, raising=False) + + +@pytest.fixture +def mock_env(monkeypatch): + """Set up a minimal valid Matrix environment.""" + monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org") + monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test_token_abc123") + monkeypatch.setenv("MATRIX_USER_ID", "@bot:example.org") + + +@pytest.fixture +def mock_urlopen(): + """Mock urllib.request.urlopen for HTTP testing.""" + with patch("urllib.request.urlopen") as mock: + yield mock + + +@pytest.fixture +def mock_request(): + """Mock urllib.request.Request for HTTP testing.""" + with patch("urllib.request.Request") as mock: + yield mock diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..0747524 --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,377 @@ +"""Test Matrix authentication: login, token validation, session persistence.""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path +from unittest import mock + +import pytest + +from matrix_hermes.auth import ( + AuthResult, + WhoAmIResult, + _password_login, + _validate_token, + load_session, + login, + login_from_config, + logout, + remove_session, + save_session, + validate_token, +) +from matrix_hermes.config import MatrixConfig + + +class TestAuthResult: + """Tests for the AuthResult dataclass.""" + + def test_success_result(self) -> None: + result = AuthResult( + success=True, + user_id="@test:matrix.org", + access_token="syt_abc", + device_id="DEV123", + homeserver="https://matrix.org", + ) + assert result.success is True + assert result.user_id == "@test:matrix.org" + assert result.error == "" + + def test_failure_result(self) -> None: + result = AuthResult(success=False, error="Invalid password") + assert result.success is False + assert result.error == "Invalid password" + assert result.access_token == "" + + def test_default_values(self) -> None: + result = AuthResult(success=True) + assert result.user_id == "" + assert result.access_token == "" + assert result.device_id == "" + assert result.homeserver == "" + + +class TestWhoAmIResult: + """Tests for the WhoAmIResult dataclass.""" + + def test_valid(self) -> None: + result = WhoAmIResult( + valid=True, user_id="@test:matrix.org", device_id="DEV1" + ) + assert result.valid is True + + def test_invalid(self) -> None: + result = WhoAmIResult(valid=False, error="M_UNKNOWN_TOKEN: Unrecognised") + assert result.valid is False + + +class TestPasswordLoginHTTP: + """Tests for _password_login — HTTP interactions are mocked.""" + + def test_successful_login(self) -> None: + mock_response = { + "user_id": "@bot:matrix.org", + "access_token": "syt_test123", + "device_id": "HERMES01", + "home_server": "matrix.org", + } + + with mock.patch( + "matrix_hermes.auth._post_json", return_value=mock_response + ) as mock_post: + result = _password_login( + "https://matrix.org", "@bot:matrix.org", "secret" + ) + assert result.success is True + assert result.user_id == "@bot:matrix.org" + assert result.access_token == "syt_test123" + assert result.device_id == "HERMES01" + mock_post.assert_called_once() + + def test_login_failure(self) -> None: + mock_response = { + "errcode": "M_FORBIDDEN", + "error": "Invalid password", + } + + with mock.patch( + "matrix_hermes.auth._post_json", return_value=mock_response + ): + result = _password_login( + "https://matrix.org", "@bot:matrix.org", "wrong" + ) + assert result.success is False + assert "M_FORBIDDEN" in result.error + + def test_login_network_error(self) -> None: + mock_response = { + "errcode": "M_CONNECTION_ERROR", + "error": "Connection refused", + } + + with mock.patch( + "matrix_hermes.auth._post_json", return_value=mock_response + ): + result = _password_login( + "https://down.example.com", "@bot:down.example.com", "pass" + ) + assert result.success is False + assert "M_CONNECTION_ERROR" in result.error + + +class TestValidateTokenHTTP: + """Tests for _validate_token — HTTP interactions are mocked.""" + + def test_valid_token(self) -> None: + mock_response = { + "user_id": "@bot:matrix.org", + "device_id": "HERMES01", + } + + with mock.patch( + "matrix_hermes.auth._get_json", return_value=mock_response + ): + result = _validate_token("https://matrix.org", "syt_valid") + assert result.valid is True + assert result.user_id == "@bot:matrix.org" + + def test_invalid_token(self) -> None: + mock_response = { + "errcode": "M_UNKNOWN_TOKEN", + "error": "Unrecognised access token", + } + + with mock.patch( + "matrix_hermes.auth._get_json", return_value=mock_response + ): + result = _validate_token("https://matrix.org", "syt_bad") + assert result.valid is False + assert "M_UNKNOWN_TOKEN" in result.error + + +class TestLoginFunction: + """Tests for the top-level login() function.""" + + def test_login_with_valid_token(self) -> None: + with mock.patch( + "matrix_hermes.auth._validate_token", + return_value=WhoAmIResult(valid=True, user_id="@bot:matrix.org"), + ): + result = login( + homeserver="https://matrix.org", access_token="syt_good" + ) + assert result.success is True + assert result.access_token == "syt_good" + + def test_login_with_invalid_token_no_password(self) -> None: + with mock.patch( + "matrix_hermes.auth._validate_token", + return_value=WhoAmIResult(valid=False, error="Bad token"), + ): + result = login( + homeserver="https://matrix.org", access_token="syt_bad" + ) + # Token invalid and no password → falls through to error + assert result.success is False + + def test_login_with_password_no_user_id(self) -> None: + result = login(homeserver="https://matrix.org", password="secret") + assert result.success is False + assert "MATRIX_USER_ID" in result.error + + def test_login_no_credentials(self) -> None: + result = login(homeserver="https://matrix.org") + assert result.success is False + assert "No valid authentication method" in result.error + + +class TestLoginFromConfig: + """Tests for login_from_config with session persistence.""" + + def test_with_access_token_valid(self) -> None: + cfg = MatrixConfig( + homeserver="https://matrix.org", + access_token="syt_valid", + ) + with mock.patch( + "matrix_hermes.auth._validate_token", + return_value=WhoAmIResult(valid=True, user_id="@bot:matrix.org"), + ): + result = login_from_config(cfg) + assert result.success is True + assert result.access_token == "syt_valid" + + def test_password_login_saves_session(self, tmp_path: Path) -> None: + cfg = MatrixConfig( + homeserver="https://matrix.org", + user_id="@bot:matrix.org", + password="secret", + session_dir=str(tmp_path), + ) + login_mock = { + "user_id": "@bot:matrix.org", + "access_token": "syt_new", + "device_id": "DEV1", + } + with mock.patch( + "matrix_hermes.auth._post_json", return_value=login_mock + ): + result = login_from_config(cfg) + assert result.success is True + + # Session should have been persisted + session_file = tmp_path / "@bot_matrix.org.json" + assert session_file.exists() + data = json.loads(session_file.read_text()) + assert data["access_token"] == "syt_new" + + def test_loads_session_when_no_token(self, tmp_path: Path) -> None: + # Pre-create a session file + session_dir = tmp_path / "matrix_sessions" + session_dir.mkdir(parents=True) + session_data = { + "user_id": "@bot:matrix.org", + "access_token": "syt_saved", + "device_id": "DEV1", + "homeserver": "https://matrix.org", + } + (session_dir / "@bot_matrix.org.json").write_text( + json.dumps(session_data) + ) + + cfg = MatrixConfig( + homeserver="https://matrix.org", + user_id="@bot:matrix.org", + session_dir=str(session_dir), + ) + whoami_mock = WhoAmIResult(valid=True, user_id="@bot:matrix.org") + with mock.patch( + "matrix_hermes.auth._validate_token", return_value=whoami_mock + ): + result = login_from_config(cfg) + assert result.success is True + assert result.access_token == "syt_saved" + + def test_session_invalid_falls_through_to_password( + self, tmp_path: Path + ) -> None: + # Pre-create a stale session file + session_dir = tmp_path / "matrix_sessions" + session_dir.mkdir(parents=True) + session_data = { + "user_id": "@bot:matrix.org", + "access_token": "syt_stale", + } + (session_dir / "@bot_matrix.org.json").write_text( + json.dumps(session_data) + ) + + cfg = MatrixConfig( + homeserver="https://matrix.org", + user_id="@bot:matrix.org", + password="secret", + session_dir=str(session_dir), + ) + # Session token is invalid → removed → password login used + whoami_stale = WhoAmIResult(valid=False, error="Expired") + login_mock = { + "user_id": "@bot:matrix.org", + "access_token": "syt_fresh", + } + + with mock.patch( + "matrix_hermes.auth._validate_token", side_effect=[whoami_stale] + ): + with mock.patch( + "matrix_hermes.auth._post_json", return_value=login_mock + ): + result = login_from_config(cfg) + assert result.success is True + assert result.access_token == "syt_fresh" + + +class TestSessionPersistence: + """Tests for save_session, load_session, remove_session.""" + + def test_save_and_load_session(self, tmp_path: Path) -> None: + cfg = MatrixConfig( + user_id="@test:matrix.org", session_dir=str(tmp_path) + ) + auth = AuthResult( + success=True, + user_id="@test:matrix.org", + access_token="syt_test", + device_id="DEV99", + homeserver="https://matrix.org", + ) + + path = save_session(auth, cfg) + assert path is not None + assert path.exists() + + loaded = load_session(cfg) + assert loaded is not None + assert loaded.success is True + assert loaded.access_token == "syt_test" + assert loaded.device_id == "DEV99" + + def test_load_nonexistent_session(self) -> None: + cfg = MatrixConfig( + user_id="@nobody:matrix.org", + session_dir="/tmp/nonexistent_dir_xyz", + ) + result = load_session(cfg) + assert result is None + + def test_remove_session(self, tmp_path: Path) -> None: + cfg = MatrixConfig( + user_id="@test:matrix.org", session_dir=str(tmp_path) + ) + auth = AuthResult( + success=True, + user_id="@test:matrix.org", + access_token="syt_test", + ) + save_session(auth, cfg) + + # Verify it exists + assert load_session(cfg) is not None + + # Remove + assert remove_session(cfg) is True + assert load_session(cfg) is None + + def test_remove_nonexistent_session(self) -> None: + cfg = MatrixConfig( + user_id="@nobody:matrix.org", + session_dir="/tmp/nonexistent_dir", + ) + assert remove_session(cfg) is False + + def test_save_failure_result_is_none(self) -> None: + cfg = MatrixConfig() + auth = AuthResult(success=False, error="No credentials") + result = save_session(auth, cfg) + assert result is None + + +class TestLogout: + """Tests for the logout function.""" + + def test_logout_success(self) -> None: + with mock.patch( + "matrix_hermes.auth._post_json", return_value={} + ): + result = logout("https://matrix.org", "syt_test") + assert result is True + + def test_logout_failure(self) -> None: + with mock.patch( + "matrix_hermes.auth._post_json", + return_value={"errcode": "M_UNKNOWN", "error": "Bad"}, + ): + result = logout("https://matrix.org", "syt_test") + assert result is False diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..1aa00b9 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,205 @@ +"""Tests for matrix_hermes.cli module.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from matrix_hermes.cli import ( + _mask_token, + build_parser, + cmd_check, + cmd_login, + cmd_logout, + cmd_show_config, + cmd_whoami, + main, +) + + +class TestMaskToken: + """Tests for _mask_token helper.""" + + def test_short_token(self): + assert _mask_token("abc") == "***" + + def test_long_token(self): + result = _mask_token("syt_abcdefghijklmnopqrstuvwxyz") + assert result.startswith("syt_ab") + assert result.endswith("uvwxyz") + assert "..." in result + + def test_exact_length(self): + """12-char token with visible=6: len<=12 means fully masked.""" + result = _mask_token("123456789012") + assert result == "************" + + def test_just_over_threshold(self): + """13-char token with visible=6: len>12 means partial mask.""" + result = _mask_token("1234567890123") + assert result == "123456...890123" + + +class TestParser: + """Tests for CLI argument parser.""" + + def test_build_parser(self): + parser = build_parser() + assert "login" in parser._subparsers._group_actions[0].choices + assert "whoami" in parser._subparsers._group_actions[0].choices + assert "check" in parser._subparsers._group_actions[0].choices + assert "show-config" in parser._subparsers._group_actions[0].choices + + def test_no_args_shows_help(self, capsys): + """Running with no args shows help.""" + result = main([]) + # Should not crash + captured = capsys.readouterr() + assert "usage" in captured.out.lower() or result == 0 + + +class TestCmdCheck: + """Tests for 'check' command.""" + + def test_not_configured(self, clean_env, capsys): + """When no Matrix vars are set, check should fail.""" + args = build_parser().parse_args(["check"]) + result = cmd_check(args) + assert result == 1 + captured = capsys.readouterr() + assert "not configured" in captured.out.lower() + + def test_configured(self, mock_env, capsys): + """When Matrix vars are set, check should succeed.""" + args = build_parser().parse_args(["check"]) + result = cmd_check(args) + assert result == 0 + captured = capsys.readouterr() + assert "valid" in captured.out.lower() + + +class TestCmdShowConfig: + """Tests for 'show-config' command.""" + + def test_output_format(self, mock_env, capsys): + """show-config outputs env var format.""" + args = build_parser().parse_args(["show-config"]) + result = cmd_show_config(args) + assert result == 0 + captured = capsys.readouterr() + assert "MATRIX_HOMESERVER=" in captured.out + assert "MATRIX_ACCESS_TOKEN=" in captured.out + + def test_redacted(self, mock_env, capsys): + """show-config --redact masks sensitive values.""" + args = build_parser().parse_args(["show-config", "--redact"]) + result = cmd_show_config(args) + assert result == 0 + captured = capsys.readouterr() + # Token should be masked + assert "syt_test_token" not in captured.out + assert "syt_test" in captured.out or "..." in captured.out + + +class TestCmdLogin: + """Tests for 'login' command.""" + + def test_no_homeserver(self, clean_env, capsys): + """Login fails without homeserver.""" + args = build_parser().parse_args(["login"]) + result = cmd_login(args) + assert result == 1 + + def test_with_valid_token(self, mock_env, capsys): + """Login with valid token prints success.""" + with patch("urllib.request.urlopen") as mock_urlopen: + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps( + {"user_id": "@bot:example.org", "device_id": "DEV1"} + ).encode() + mock_resp.__enter__.return_value = mock_resp + mock_urlopen.return_value = mock_resp + + args = build_parser().parse_args(["login"]) + result = cmd_login(args) + assert result == 0 + + def test_with_bad_credentials(self, mock_env, capsys): + """Login with bad credentials prints error.""" + with patch("urllib.request.urlopen") as mock_urlopen: + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps( + {"errcode": "M_FORBIDDEN", "error": "Bad password"} + ).encode() + mock_resp.__enter__.return_value = mock_resp + mock_urlopen.return_value = mock_resp + + # Override with password-based login (no valid token) + import os + + os.environ["MATRIX_ACCESS_TOKEN"] = "" + os.environ["MATRIX_PASSWORD"] = "wrong" + + args = build_parser().parse_args(["login"]) + result = cmd_login(args) + assert result == 1 + + +class TestCmdWhoami: + """Tests for 'whoami' command.""" + + def test_no_homeserver(self, clean_env, capsys): + """whoami fails without homeserver.""" + args = build_parser().parse_args(["whoami"]) + result = cmd_whoami(args) + assert result == 1 + + def test_valid(self, mock_env, capsys): + """whoami with valid token.""" + with patch("urllib.request.urlopen") as mock_urlopen: + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps( + {"user_id": "@bot:example.org"} + ).encode() + mock_resp.__enter__.return_value = mock_resp + mock_urlopen.return_value = mock_resp + + args = build_parser().parse_args(["whoami"]) + result = cmd_whoami(args) + assert result == 0 + + def test_invalid(self, mock_env, capsys): + """whoami with invalid token.""" + with patch("urllib.request.urlopen") as mock_urlopen: + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps( + {"errcode": "M_UNKNOWN_TOKEN", "error": "Unrecognised"} + ).encode() + mock_resp.__enter__.return_value = mock_resp + mock_urlopen.return_value = mock_resp + + args = build_parser().parse_args(["whoami"]) + result = cmd_whoami(args) + assert result == 1 + + +class TestCmdLogout: + """Tests for 'logout' command.""" + + def test_no_homeserver(self, clean_env, capsys): + """logout fails without homeserver.""" + args = build_parser().parse_args(["logout"]) + result = cmd_logout(args) + assert result == 1 + + def test_success(self, mock_env, capsys): + """logout with valid token.""" + with patch("urllib.request.urlopen") as mock_urlopen: + mock_resp = MagicMock() + mock_resp.read.return_value = json.dumps({}).encode() + mock_resp.__enter__.return_value = mock_resp + mock_urlopen.return_value = mock_resp + + args = build_parser().parse_args(["logout"]) + result = cmd_logout(args) + assert result == 0 diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..f85b762 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,107 @@ +"""Test Matrix configuration loading from env vars and files.""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from matrix_hermes.config import MatrixConfig, _parse_csv + + +class TestParseCSV: + """Tests for the CSV parsing helper.""" + + def test_empty_string(self) -> None: + assert _parse_csv("") == [] + + def test_single_value(self) -> None: + assert _parse_csv("alice") == ["alice"] + + def test_multiple_values(self) -> None: + assert _parse_csv("alice, bob , charlie") == ["alice", "bob", "charlie"] + + def test_whitespace_only(self) -> None: + assert _parse_csv(" , , ") == [] + + +class TestMatrixConfigDefaults: + """Tests for default config values.""" + + def test_empty_config(self) -> None: + cfg = MatrixConfig() + assert cfg.homeserver == "" + assert cfg.access_token == "" + assert cfg.password == "" + + def test_is_configured_false(self) -> None: + cfg = MatrixConfig() + assert cfg.is_configured() is False + + def test_is_configured_with_homeserver_and_token(self) -> None: + cfg = MatrixConfig(homeserver="https://matrix.org", access_token="fake-token") + assert cfg.is_configured() is True + + def test_is_configured_with_homeserver_and_password(self) -> None: + cfg = MatrixConfig( + homeserver="https://matrix.org", + user_id="@test:matrix.org", + password="secret", + ) + assert cfg.is_configured() is True + + def test_validate_empty_config(self) -> None: + cfg = MatrixConfig() + errors = cfg.validate() + assert len(errors) > 0 + assert any("MATRIX_HOMESERVER" in e for e in errors) + + def test_validate_missing_auth(self) -> None: + cfg = MatrixConfig(homeserver="https://matrix.org") + errors = cfg.validate() + assert any("ACCESS_TOKEN" in e or "PASSWORD" in e for e in errors) + + def test_resolved_session_dir_default(self) -> None: + cfg = MatrixConfig() + assert cfg.resolved_session_dir == str( + Path.home() / ".hermes" / "matrix_sessions" + ) + + def test_resolved_session_dir_custom(self) -> None: + cfg = MatrixConfig(session_dir="/tmp/matrix_test") + assert cfg.resolved_session_dir == "/tmp/matrix_test" + + +class TestFromEnv: + """Tests for env var loading.""" + + def test_loads_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("MATRIX_HOMESERVER", "https://example.com") + monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_abc123") + monkeypatch.setenv("MATRIX_USER_ID", "@bot:example.com") + monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false") + + cfg = MatrixConfig.from_env() + assert cfg.homeserver == "https://example.com" + assert cfg.access_token == "syt_abc123" + assert cfg.user_id == "@bot:example.com" + assert cfg.require_mention is False + + def test_encryption_flag(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("MATRIX_ENCRYPTION", "true") + cfg = MatrixConfig.from_env() + assert cfg.encryption is True + + def test_allowed_users(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("MATRIX_ALLOWED_USERS", "@a:example.com, @b:example.com") + cfg = MatrixConfig.from_env() + assert cfg.allowed_users == ["@a:example.com", "@b:example.com"] + + def test_auto_thread_default(self) -> None: + cfg = MatrixConfig.from_env() + assert cfg.auto_thread is True + + def test_require_mention_default(self) -> None: + cfg = MatrixConfig.from_env() + assert cfg.require_mention is True