diff --git a/.env.example b/.env.example index dfd70908..e486141d 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,11 @@ TELEGRAM_BOT_USERNAME=your_bot_username # Base directory for project access (absolute path) APPROVED_DIRECTORY=/path/to/your/projects +# Multiple approved directories (optional, comma-separated absolute paths) +# When set, projects can be located in any of these directories +# Example: APPROVED_DIRECTORIES=/path/to/projects1,/path/to/projects2,/home/user/workspace +APPROVED_DIRECTORIES= + # === SECURITY SETTINGS === # Comma-separated list of allowed Telegram user IDs (optional) # Leave empty to allow all users (not recommended for production) @@ -63,6 +68,19 @@ USE_SDK=true # Get your API key from: https://console.anthropic.com/ ANTHROPIC_API_KEY= +# Custom base URL for Anthropic API (optional, for proxy/enterprise endpoints) +# Example: https://your-proxy.example.com/v1 +ANTHROPIC_BASE_URL= + +# Available Claude models for user selection (comma-separated) +# Users can switch models in-session with /model command +# Supports both full names and short aliases: +# Full names: claude-opus-4-6, claude-sonnet-4-6, claude-haiku-4-5 +# Short aliases: opus, sonnet, haiku, cc-opus, cc-sonnet, cc-haiku +# Useful for enterprise/proxy endpoints with custom naming schemes +# Example: ANTHROPIC_MODELS=cc-opus,cc-sonnet,cc-haiku +ANTHROPIC_MODELS= + # Path to Claude CLI executable (optional - will auto-detect if not specified) # Example: /usr/local/bin/claude or ~/.nvm/versions/node/v20.19.2/bin/claude CLAUDE_CLI_PATH= diff --git a/.gitignore b/.gitignore index 6e3390e3..7092352f 100644 --- a/.gitignore +++ b/.gitignore @@ -146,4 +146,5 @@ data/ sessions/ backups/ uploads/ -config/mcp.json \ No newline at end of file +config/mcp.json +config/projects.yaml diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index 0917d335..00000000 --- a/CLAUDE.md +++ /dev/null @@ -1,137 +0,0 @@ -# CLAUDE.md - -This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. - -## Project Overview - -Telegram bot providing remote access to Claude Code. Python 3.10+, built with Poetry, using `python-telegram-bot` for Telegram and `claude-agent-sdk` for Claude Code integration. - -## Commands - -```bash -make dev # Install all deps (including dev) -make install # Production deps only -make run # Run the bot -make run-debug # Run with debug logging -make test # Run tests with coverage -make lint # Black + isort + flake8 + mypy -make format # Auto-format with black + isort - -# Run a single test -poetry run pytest tests/unit/test_config.py -k test_name -v - -# Type checking only -poetry run mypy src -``` - -## Architecture - -### Claude SDK Integration - -`ClaudeIntegration` (facade in `src/claude/facade.py`) wraps `ClaudeSDKManager` (`src/claude/sdk_integration.py`), which uses `claude-agent-sdk` with `ClaudeSDKClient` for async streaming. Session IDs come from Claude's `ResultMessage`, not generated locally. - -Sessions auto-resume: per user+directory, persisted in SQLite. - -### Request Flow - -**Agentic mode** (default, `AGENTIC_MODE=true`): - -``` -Telegram message -> Security middleware (group -3) -> Auth middleware (group -2) --> Rate limit (group -1) -> MessageOrchestrator.agentic_text() (group 10) --> ClaudeIntegration.run_command() -> SDK --> Response parsed -> Stored in SQLite -> Sent back to Telegram -``` - -**External triggers** (webhooks, scheduler): - -``` -Webhook POST /webhooks/{provider} -> Signature verification -> Deduplication --> Publish WebhookEvent to EventBus -> AgentHandler.handle_webhook() --> ClaudeIntegration.run_command() -> Publish AgentResponseEvent --> NotificationService -> Rate-limited Telegram delivery -``` - -**Classic mode** (`AGENTIC_MODE=false`): Same middleware chain, but routes through full command/message handlers in `src/bot/handlers/` with 13 commands and inline keyboards. - -### Dependency Injection - -Bot handlers access dependencies via `context.bot_data`: -```python -context.bot_data["auth_manager"] -context.bot_data["claude_integration"] -context.bot_data["storage"] -context.bot_data["security_validator"] -``` - -### Key Directories - -- `src/config/` -- Pydantic Settings v2 config with env detection, feature flags (`features.py`), YAML project loader (`loader.py`) -- `src/bot/handlers/` -- Telegram command, message, and callback handlers (classic mode + project thread commands) -- `src/bot/middleware/` -- Auth, rate limit, security input validation -- `src/bot/features/` -- Git integration, file handling, quick actions, session export -- `src/bot/orchestrator.py` -- MessageOrchestrator: routes to agentic or classic handlers, project-topic routing -- `src/claude/` -- Claude integration facade, SDK/CLI managers, session management, tool monitoring -- `src/projects/` -- Multi-project support: `registry.py` (YAML project config), `thread_manager.py` (Telegram topic sync/routing) -- `src/storage/` -- SQLite via aiosqlite, repository pattern (users, sessions, messages, tool_usage, audit_log, cost_tracking, project_threads) -- `src/security/` -- Multi-provider auth (whitelist + token), input validators (with optional `disable_security_patterns`), rate limiter, audit logging -- `src/events/` -- EventBus (async pub/sub), event types, AgentHandler, EventSecurityMiddleware -- `src/api/` -- FastAPI webhook server, GitHub HMAC-SHA256 + Bearer token auth -- `src/scheduler/` -- APScheduler cron jobs, persistent storage in SQLite -- `src/notifications/` -- NotificationService, rate-limited Telegram delivery - -### Security Model - -5-layer defense: authentication (whitelist/token) -> directory isolation (APPROVED_DIRECTORY + path traversal prevention) -> input validation (blocks `..`, `;`, `&&`, `$()`, etc.) -> rate limiting (token bucket) -> audit logging. - -`SecurityValidator` blocks access to secrets (`.env`, `.ssh`, `id_rsa`, `.pem`) and dangerous shell patterns. Can be relaxed with `DISABLE_SECURITY_PATTERNS=true` (trusted environments only). - -`ToolMonitor` validates Claude's tool calls against allowlist/disallowlist, file path boundaries, and dangerous bash patterns. Tool name validation can be bypassed with `DISABLE_TOOL_VALIDATION=true`. - -Webhook authentication: GitHub HMAC-SHA256 signature verification, generic Bearer token for other providers, atomic deduplication via `webhook_events` table. - -### Configuration - -Settings loaded from environment variables via Pydantic Settings. Required: `TELEGRAM_BOT_TOKEN`, `TELEGRAM_BOT_USERNAME`, `APPROVED_DIRECTORY`. Key optional: `ALLOWED_USERS` (comma-separated Telegram IDs), `ANTHROPIC_API_KEY`, `ENABLE_MCP`, `MCP_CONFIG_PATH`. - -Agentic platform settings: `AGENTIC_MODE` (default true), `ENABLE_API_SERVER`, `API_SERVER_PORT` (default 8080), `GITHUB_WEBHOOK_SECRET`, `WEBHOOK_API_SECRET`, `ENABLE_SCHEDULER`, `NOTIFICATION_CHAT_IDS`. - -Security relaxation (trusted environments only): `DISABLE_SECURITY_PATTERNS` (default false), `DISABLE_TOOL_VALIDATION` (default false). - -Multi-project topics: `ENABLE_PROJECT_THREADS` (default false), `PROJECT_THREADS_MODE` (`private`|`group`), `PROJECT_THREADS_CHAT_ID` (required for group mode), `PROJECTS_CONFIG_PATH` (path to YAML project registry), `PROJECT_THREADS_SYNC_ACTION_INTERVAL_SECONDS` (default `1.1`, set `0` to disable pacing). See `config/projects.example.yaml`. - -Output verbosity: `VERBOSE_LEVEL` (default 1, range 0-2). Controls how much of Claude's background activity is shown to the user in real-time. 0 = quiet (only final response, typing indicator still active), 1 = normal (tool names + reasoning snippets shown during execution), 2 = detailed (tool names with input summaries + longer reasoning text). Users can override per-session via `/verbose 0|1|2`. A persistent typing indicator is refreshed every ~2 seconds at all levels. - -Voice transcription: `ENABLE_VOICE_MESSAGES` (default true), `VOICE_PROVIDER` (`mistral`|`openai`, default `mistral`), `MISTRAL_API_KEY`, `OPENAI_API_KEY`, `VOICE_TRANSCRIPTION_MODEL`. Provider implementation is in `src/bot/features/voice_handler.py`. - -Feature flags in `src/config/features.py` control: MCP, git integration, file uploads, quick actions, session export, image uploads, voice messages, conversation mode, agentic mode, API server, scheduler. - -### DateTime Convention - -All datetimes use timezone-aware UTC: `datetime.now(UTC)` (not `datetime.utcnow()`). SQLite adapters auto-convert TIMESTAMP/DATETIME columns to `datetime` objects via `detect_types=PARSE_DECLTYPES`. Model `from_row()` methods must guard `fromisoformat()` calls with `isinstance(val, str)` checks. - -## Code Style - -- Black (88 char line length), isort (black profile), flake8, mypy strict, autoflake for unused imports -- pytest-asyncio with `asyncio_mode = "auto"` -- structlog for all logging (JSON in prod, console in dev) -- Type hints required on all functions (`disallow_untyped_defs = true`) -- Use `datetime.now(UTC)` not `datetime.utcnow()` (deprecated) - -## Adding a New Bot Command - -### Agentic mode - -Agentic mode commands: `/start`, `/new`, `/status`, `/verbose`, `/repo`. If `ENABLE_PROJECT_THREADS=true`: `/sync_threads`. To add a new command: - -1. Add handler function in `src/bot/orchestrator.py` -2. Register in `MessageOrchestrator._register_agentic_handlers()` -3. Add to `MessageOrchestrator.get_bot_commands()` for Telegram's command menu -4. Add audit logging for the command - -### Classic mode - -1. Add handler function in `src/bot/handlers/command.py` -2. Register in `MessageOrchestrator._register_classic_handlers()` -3. Add to `MessageOrchestrator.get_bot_commands()` for Telegram's command menu -4. Add audit logging for the command diff --git a/README.md b/README.md index 34ca52d3..33255c37 100644 --- a/README.md +++ b/README.md @@ -230,6 +230,7 @@ ALLOWED_USERS=123456789 # Comma-separated Telegram user IDs ```bash # Claude ANTHROPIC_API_KEY=sk-ant-... # API key (optional if using CLI auth) +ANTHROPIC_BASE_URL=... # Custom API endpoint (optional, for proxy/enterprise endpoints) CLAUDE_MAX_COST_PER_USER=10.0 # Spending limit per user (USD) CLAUDE_TIMEOUT_SECONDS=300 # Operation timeout diff --git a/config/projects.example.yaml b/config/projects.example.yaml index 79eb896a..88638f28 100644 --- a/config/projects.example.yaml +++ b/config/projects.example.yaml @@ -1,10 +1,19 @@ projects: + # Relative path (resolved against APPROVED_DIRECTORY) - slug: claude-code-telegram name: Claude Code Telegram path: claude-code-telegram-main enabled: true + # Relative path - slug: infra name: Infrastructure path: infrastructure enabled: true + + # Absolute path example (must be within one of the approved directories) + # Uncomment and adjust if using APPROVED_DIRECTORIES + # - slug: external-project + # name: External Project + # path: /path/to/other/approved/directory/project + # enabled: true diff --git a/docs/configuration.md b/docs/configuration.md index 2098f8be..5607ea18 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -64,6 +64,9 @@ DISABLE_TOOL_VALIDATION=false # Authentication ANTHROPIC_API_KEY=sk-ant-api03-... # Optional: API key for SDK (uses CLI auth if omitted) +# Custom API endpoint (optional, for proxy/enterprise endpoints) +ANTHROPIC_BASE_URL=https://your-proxy.example.com/v1 # Optional: custom base URL for Anthropic API + # Maximum conversation turns before requiring new session CLAUDE_MAX_TURNS=10 diff --git a/src/bot/orchestrator.py b/src/bot/orchestrator.py index ac1d5304..90395f23 100644 --- a/src/bot/orchestrator.py +++ b/src/bot/orchestrator.py @@ -28,6 +28,7 @@ filters, ) +from ..claude.model_mapper import get_display_name, resolve_model_name from ..claude.sdk_integration import StreamUpdate from ..config.settings import Settings from ..projects import PrivateTopicsUnavailableError @@ -306,6 +307,7 @@ def _register_agentic_handlers(self, app: Application) -> None: ("new", self.agentic_new), ("status", self.agentic_status), ("verbose", self.agentic_verbose), + ("model", self.agentic_model), ("repo", self.agentic_repo), ("restart", command.restart_command), ] @@ -344,11 +346,11 @@ def _register_agentic_handlers(self, app: Application) -> None: group=10, ) - # Only cd: callbacks (for project selection), scoped by pattern + # Callback handlers for cd: (project selection) and model: (model selection) app.add_handler( CallbackQueryHandler( self._inject_deps(self._agentic_callback), - pattern=r"^cd:", + pattern=r"^(cd|model):", ) ) @@ -415,6 +417,7 @@ async def get_bot_commands(self) -> list: # type: ignore[type-arg] BotCommand("new", "Start a fresh session"), BotCommand("status", "Show session status"), BotCommand("verbose", "Set output verbosity (0/1/2)"), + BotCommand("model", "Switch Claude model"), BotCommand("repo", "List repos / switch workspace"), BotCommand("restart", "Restart the bot"), ] @@ -578,6 +581,152 @@ async def agentic_verbose( parse_mode="HTML", ) + def _get_selected_model(self, context: ContextTypes.DEFAULT_TYPE) -> Optional[str]: + """Return effective model: per-user override or config default. + + Note: Returns the user's selection as-is (could be alias or full name). + Resolution to full Anthropic model name happens in SDK integration layer. + """ + user_override = context.user_data.get("selected_model") + if user_override is not None: + return str(user_override) + return self.settings.claude_model + + async def agentic_model( + self, update: Update, context: ContextTypes.DEFAULT_TYPE + ) -> None: + """Switch Claude model: /model [model-name].""" + args = update.message.text.split()[1:] if update.message.text else [] + + # Get available models from config + available_models = self.settings.anthropic_models or [] + + if not args: + # Show current model and available options with inline keyboard + current = self._get_selected_model(context) + + # Get friendly display name (resolves aliases) + current_display_name = get_display_name(current) + current_display = ( + f"{escape_html(current_display_name)}" + if current + else "default" + ) + + if not available_models: + await update.message.reply_text( + f"Current model: {current_display}\n\n" + f"ℹ️ No models configured in ANTHROPIC_MODELS.\n" + f"Add models to your .env file:\n" + f"ANTHROPIC_MODELS=claude-opus-4-6,claude-sonnet-4-6,claude-haiku-4-5\n\n" + f"Usage: /model <model-name>", + parse_mode="HTML", + ) + return + + # Build model list display with friendly names + lines: List[str] = [] + for m in available_models: + # Show both display name and actual model string if different + display = get_display_name(m) + resolved = resolve_model_name(m) + + # Check if current selection matches (compare resolved names) + current_resolved = resolve_model_name(current) if current else None + marker = " ◀" if resolved == current_resolved else "" + + # Show display name with actual model in parentheses if different + if display != m: + lines.append( + f" • {escape_html(display)} " + f"({escape_html(m)}){marker}" + ) + else: + lines.append(f" • {escape_html(m)}{marker}") + + # Build inline keyboard (1 per row for clarity) + keyboard_rows: List[List[InlineKeyboardButton]] = [] + for model_name in available_models: + keyboard_rows.append( + [ + InlineKeyboardButton( + model_name, callback_data=f"model:{model_name}" + ) + ] + ) + + # Add "Reset to Default" button if user has override + if context.user_data.get("selected_model"): + keyboard_rows.append( + [ + InlineKeyboardButton( + "🔄 Reset to Default", callback_data="model:default" + ) + ] + ) + + reply_markup = InlineKeyboardMarkup(keyboard_rows) + + await update.message.reply_text( + f"Select Model\n\n" + f"Current: {current_display}\n\n" + "\n".join(lines), + parse_mode="HTML", + reply_markup=reply_markup, + ) + return + + # Handle command with argument: /model + model_name = " ".join(args).strip() + + # Handle reset to default + if model_name.lower() in ("default", "reset", "clear"): + context.user_data.pop("selected_model", None) + default = self.settings.claude_model or "CLI default" + default_display = get_display_name(default) + await update.message.reply_text( + f"✅ Model reset to default: {escape_html(default_display)}", + parse_mode="HTML", + ) + return + + # Validate model is in available list (if list is configured) + # Compare resolved names to support aliases + if available_models: + model_resolved = resolve_model_name(model_name) + available_resolved = [resolve_model_name(m) for m in available_models] + + if model_resolved not in available_resolved: + models_list = ", ".join( + f"{escape_html(m)}" for m in available_models + ) + await update.message.reply_text( + f"❌ Model {escape_html(model_name)} not in available list.\n\n" + f"Available models: {models_list}", + parse_mode="HTML", + ) + return + + # Set the model (store user's input, resolution happens in SDK layer) + context.user_data["selected_model"] = model_name + + # Clear session to avoid tool state conflicts + # (switching mid-conversation with pending tools causes API errors) + old_session_id = context.user_data.get("claude_session_id") + if old_session_id: + context.user_data.pop("claude_session_id", None) + logger.info( + "Cleared session due to model switch", + old_session_id=old_session_id, + new_model=model_name, + ) + + model_display = get_display_name(model_name) + await update.message.reply_text( + f"✅ Model switched to: {escape_html(model_display)}\n\n" + f"Starting fresh session with new model.", + parse_mode="HTML", + ) + def _format_verbose_progress( self, activity_log: List[Dict[str, Any]], @@ -932,6 +1081,9 @@ async def agentic_text( # Independent typing heartbeat — stays alive even with no stream events heartbeat = self._start_typing_heartbeat(chat) + # Get user's selected model (if any) + selected_model = self._get_selected_model(context) + success = True try: claude_response = await claude_integration.run_command( @@ -941,6 +1093,7 @@ async def agentic_text( session_id=session_id, on_stream=on_stream, force_new=force_new, + model=selected_model, ) # New session created successfully — clear the one-shot flag @@ -1558,52 +1711,115 @@ async def agentic_repo( async def _agentic_callback( self, update: Update, context: ContextTypes.DEFAULT_TYPE ) -> None: - """Handle cd: callbacks — switch directory and resume session if available.""" + """Handle cd: and model: callbacks.""" query = update.callback_query await query.answer() data = query.data - _, project_name = data.split(":", 1) + prefix, value = data.split(":", 1) - base = self.settings.approved_directory - new_path = base / project_name + if prefix == "cd": + # Handle directory change + project_name = value + base = self.settings.approved_directory + new_path = base / project_name + + if not new_path.is_dir(): + await query.edit_message_text( + f"Directory not found: {escape_html(project_name)}", + parse_mode="HTML", + ) + return + + context.user_data["current_directory"] = new_path + + # Look for a resumable session instead of always clearing + claude_integration = context.bot_data.get("claude_integration") + session_id = None + if claude_integration: + existing = await claude_integration._find_resumable_session( + query.from_user.id, new_path + ) + if existing: + session_id = existing.session_id + context.user_data["claude_session_id"] = session_id + + is_git = (new_path / ".git").is_dir() + git_badge = " (git)" if is_git else "" + session_badge = " · session resumed" if session_id else "" - if not new_path.is_dir(): await query.edit_message_text( - f"Directory not found: {escape_html(project_name)}", + f"Switched to {escape_html(project_name)}/" + f"{git_badge}{session_badge}", parse_mode="HTML", ) - return - context.user_data["current_directory"] = new_path + elif prefix == "model": + # Handle model selection + model_name = value + + if model_name == "default": + # Reset to config default + context.user_data.pop("selected_model", None) + default = self.settings.claude_model or "CLI default" + default_display = get_display_name(default) + await query.edit_message_text( + f"✅ Model reset to default: {escape_html(default_display)}", + parse_mode="HTML", + ) + return - # Look for a resumable session instead of always clearing - claude_integration = context.bot_data.get("claude_integration") - session_id = None - if claude_integration: - existing = await claude_integration._find_resumable_session( - query.from_user.id, new_path - ) - if existing: - session_id = existing.session_id - context.user_data["claude_session_id"] = session_id + # Validate model is in available list (if configured) + # Compare resolved names to support aliases + available_models = self.settings.anthropic_models or [] + if available_models: + model_resolved = resolve_model_name(model_name) + available_resolved = [resolve_model_name(m) for m in available_models] - is_git = (new_path / ".git").is_dir() - git_badge = " (git)" if is_git else "" - session_badge = " · session resumed" if session_id else "" + if model_resolved not in available_resolved: + await query.edit_message_text( + f"❌ Model {escape_html(model_name)} not in available list.", + parse_mode="HTML", + ) + return - await query.edit_message_text( - f"Switched to {escape_html(project_name)}/" - f"{git_badge}{session_badge}", - parse_mode="HTML", - ) + # Set the model (store user's input, resolution happens in SDK layer) + context.user_data["selected_model"] = model_name + + # Clear session to avoid tool state conflicts + old_session_id = context.user_data.get("claude_session_id") + if old_session_id: + context.user_data.pop("claude_session_id", None) + logger.info( + "Cleared session due to model switch", + old_session_id=old_session_id, + new_model=model_name, + ) - # Audit log - audit_logger = context.bot_data.get("audit_logger") - if audit_logger: - await audit_logger.log_command( - user_id=query.from_user.id, - command="cd", - args=[project_name], - success=True, + model_display = get_display_name(model_name) + await query.edit_message_text( + f"✅ Model switched to: {escape_html(model_display)}\n\n" + f"Starting fresh session with new model.", + parse_mode="HTML", ) + + # Audit log - only for cd commands + if prefix == "cd": + audit_logger = context.bot_data.get("audit_logger") + if audit_logger: + await audit_logger.log_command( + user_id=query.from_user.id, + command="cd", + args=[project_name], + success=True, + ) + elif prefix == "model": + # Audit log for model commands + audit_logger = context.bot_data.get("audit_logger") + if audit_logger: + await audit_logger.log_command( + user_id=query.from_user.id, + command="model", + args=[model_name], + success=True, + ) diff --git a/src/claude/facade.py b/src/claude/facade.py index fcb2ada6..9ec925ef 100644 --- a/src/claude/facade.py +++ b/src/claude/facade.py @@ -37,6 +37,7 @@ async def run_command( session_id: Optional[str] = None, on_stream: Optional[Callable[[StreamUpdate], None]] = None, force_new: bool = False, + model: Optional[str] = None, ) -> ClaudeResponse: """Run Claude Code command with full integration.""" logger.info( @@ -85,6 +86,7 @@ async def run_command( session_id=claude_session_id, continue_session=should_continue, stream_callback=on_stream, + model=model, ) except Exception as resume_error: # If resume failed (e.g., session expired/missing on Claude's side), @@ -109,6 +111,7 @@ async def run_command( session_id=None, continue_session=False, stream_callback=on_stream, + model=model, ) else: raise @@ -152,6 +155,7 @@ async def _execute( session_id: Optional[str] = None, continue_session: bool = False, stream_callback: Optional[Callable] = None, + model: Optional[str] = None, ) -> ClaudeResponse: """Execute command via SDK.""" return await self.sdk_manager.execute_command( @@ -160,6 +164,7 @@ async def _execute( session_id=session_id, continue_session=continue_session, stream_callback=stream_callback, + model=model, ) async def _find_resumable_session( diff --git a/src/claude/model_mapper.py b/src/claude/model_mapper.py new file mode 100644 index 00000000..35641454 --- /dev/null +++ b/src/claude/model_mapper.py @@ -0,0 +1,193 @@ +"""Model name mapper for proxy/enterprise endpoints. + +Maps short aliases (cc-opus, sonnet, etc.) to full Anthropic model names. +Useful when using ANTHROPIC_BASE_URL with custom model naming schemes. + +References: +- Claude 4.6 models: https://docs.anthropic.com/en/docs/about-claude/models +- Claude 4.5 models: https://docs.anthropic.com/en/docs/about-claude/models +""" + +from typing import Optional + +import structlog + +logger = structlog.get_logger() + +# Official Anthropic model names (as of 2026-03-10) +# Source: https://docs.anthropic.com/en/docs/about-claude/models +MODEL_ALIASES = { + # Claude 4.6 series (latest) + "opus": "claude-opus-4-6", + "cc-opus": "claude-opus-4-6", + "opus-4.6": "claude-opus-4-6", + "opus-4-6": "claude-opus-4-6", + "claude-opus": "claude-opus-4-6", + "sonnet": "claude-sonnet-4-6", + "cc-sonnet": "claude-sonnet-4-6", + "sonnet-4.6": "claude-sonnet-4-6", + "sonnet-4-6": "claude-sonnet-4-6", + "claude-sonnet": "claude-sonnet-4-6", + # Claude 4.5 series (with date versions) + "haiku": "claude-haiku-4-5-20251001", + "cc-haiku": "claude-haiku-4-5-20251001", + "haiku-4.5": "claude-haiku-4-5-20251001", + "haiku-4-5": "claude-haiku-4-5-20251001", + "claude-haiku": "claude-haiku-4-5-20251001", + # Legacy short names + "claude-haiku-4-5": "claude-haiku-4-5-20251001", + # Claude 3.7 series (if still supported) + "opus-3.7": "claude-opus-3-7-20250219", + "opus-3-7": "claude-opus-3-7-20250219", + # Claude 3.5 series (legacy) + "sonnet-3.5": "claude-3-5-sonnet-20241022", + "sonnet-3-5": "claude-3-5-sonnet-20241022", + "sonnet-20241022": "claude-3-5-sonnet-20241022", + "haiku-3.5": "claude-3-5-haiku-20241022", + "haiku-3-5": "claude-3-5-haiku-20241022", + "haiku-20241022": "claude-3-5-haiku-20241022", + # Legacy Claude 3 series + "opus-3": "claude-3-opus-20240229", + "sonnet-3": "claude-3-sonnet-20240229", + "haiku-3": "claude-3-haiku-20240307", +} + +# Reverse mapping: full name → shortest alias +MODEL_DISPLAY_NAMES = { + "claude-opus-4-6": "Opus 4.6", + "claude-sonnet-4-6": "Sonnet 4.6", + "claude-haiku-4-5-20251001": "Haiku 4.5", + "claude-haiku-4-5": "Haiku 4.5", # Legacy alias + "claude-opus-3-7-20250219": "Opus 3.7", + "claude-3-5-sonnet-20241022": "Sonnet 3.5", + "claude-3-5-haiku-20241022": "Haiku 3.5", + "claude-3-opus-20240229": "Opus 3", + "claude-3-sonnet-20240229": "Sonnet 3", + "claude-3-haiku-20240307": "Haiku 3", +} + + +def resolve_model_name(model_input: Optional[str]) -> Optional[str]: + """Resolve model alias to full Anthropic model name. + + Args: + model_input: Short alias (e.g., "cc-opus", "sonnet") or full name + + Returns: + Full Anthropic model name (e.g., "claude-opus-4-6") or None if input is None + + Examples: + >>> resolve_model_name("cc-opus") + 'claude-opus-4-6' + >>> resolve_model_name("sonnet") + 'claude-sonnet-4-6' + >>> resolve_model_name("claude-opus-4-6") # Already full name + 'claude-opus-4-6' + >>> resolve_model_name(None) + None + """ + if model_input is None: + return None + + # Normalize input + normalized = model_input.strip().lower() + + # Check if it's a known alias + if normalized in MODEL_ALIASES: + resolved = MODEL_ALIASES[normalized] + logger.debug( + "Resolved model alias", + input=model_input, + resolved=resolved, + ) + return resolved + + # Already a full name or custom model name, return as-is + logger.debug( + "Model name passed through (not an alias)", + input=model_input, + ) + return model_input + + +def get_display_name(model_name: Optional[str]) -> str: + """Get user-friendly display name for a model. + + Args: + model_name: Full model name or alias + + Returns: + Short display name (e.g., "Opus 4.6") or original name if unknown + + Examples: + >>> get_display_name("claude-opus-4-6") + 'Opus 4.6' + >>> get_display_name("cc-sonnet") + 'Sonnet 4.6' + >>> get_display_name("custom-model-xyz") + 'custom-model-xyz' + """ + if not model_name: + return "default" + + # Resolve alias first + resolved = resolve_model_name(model_name) + if not resolved: + return "default" + + # Look up display name + display = MODEL_DISPLAY_NAMES.get(resolved) + if display: + return display + + # Unknown model, return as-is + return resolved + + +def is_valid_model_alias(model_input: str) -> bool: + """Check if input is a known model alias. + + Args: + model_input: Model name or alias to check + + Returns: + True if it's a known alias, False otherwise + + Examples: + >>> is_valid_model_alias("cc-opus") + True + >>> is_valid_model_alias("claude-opus-4-6") + False # Full name, not an alias + >>> is_valid_model_alias("unknown-model") + False + """ + return model_input.strip().lower() in MODEL_ALIASES + + +def get_all_aliases() -> list[str]: + """Get list of all known model aliases. + + Returns: + List of alias strings + + Examples: + >>> aliases = get_all_aliases() + >>> "cc-opus" in aliases + True + """ + return sorted(MODEL_ALIASES.keys()) + + +def get_all_full_names() -> list[str]: + """Get list of all known full model names. + + Returns: + List of full Anthropic model names, deduplicated and sorted + + Examples: + >>> full_names = get_all_full_names() + >>> "claude-opus-4-6" in full_names + True + """ + unique_names = sorted(set(MODEL_ALIASES.values())) + return unique_names diff --git a/src/claude/sdk_integration.py b/src/claude/sdk_integration.py index adf553f4..4523ed58 100644 --- a/src/claude/sdk_integration.py +++ b/src/claude/sdk_integration.py @@ -36,6 +36,7 @@ ClaudeProcessError, ClaudeTimeoutError, ) +from .model_mapper import resolve_model_name from .monitor import _is_claude_internal_path, check_bash_directory_boundary logger = structlog.get_logger() @@ -146,6 +147,14 @@ def __init__( else: logger.info("No API key provided, using existing Claude CLI authentication") + # Set up custom base URL if provided + if config.anthropic_base_url_str: + os.environ["ANTHROPIC_BASE_URL"] = config.anthropic_base_url_str + logger.info( + "Using custom base URL for Claude SDK", + base_url=config.anthropic_base_url_str, + ) + async def execute_command( self, prompt: str, @@ -153,6 +162,7 @@ async def execute_command( session_id: Optional[str] = None, continue_session: bool = False, stream_callback: Optional[Callable[[StreamUpdate], None]] = None, + model: Optional[str] = None, ) -> ClaudeResponse: """Execute Claude Code command via SDK.""" start_time = asyncio.get_event_loop().time() @@ -195,9 +205,23 @@ def _stderr_callback(line: str) -> None: sdk_disallowed_tools = self.config.claude_disallowed_tools # Build Claude Agent options + # Use per-request model override if provided, otherwise fall back to config + user_model = model or self.config.claude_model or None + + # Resolve model aliases (e.g., "cc-opus" → "claude-opus-4-6") + # This supports enterprise/proxy endpoints with custom model names + effective_model = resolve_model_name(user_model) + + if effective_model != user_model: + logger.info( + "Resolved model alias", + input=user_model, + resolved=effective_model, + ) + options = ClaudeAgentOptions( max_turns=self.config.claude_max_turns, - model=self.config.claude_model or None, + model=effective_model, max_budget_usd=self.config.claude_max_cost_per_request, cwd=str(working_directory), allowed_tools=sdk_allowed_tools, diff --git a/src/config/settings.py b/src/config/settings.py index 77c34ea4..d7b3defe 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -41,6 +41,9 @@ class Settings(BaseSettings): # Security approved_directory: Path = Field(..., description="Base directory for projects") + approved_directories: Optional[List[Path]] = Field( + None, description="List of approved directories for projects (alternative to single approved_directory)" + ) allowed_users: Optional[List[int]] = Field( None, description="Allowed Telegram user IDs" ) @@ -74,9 +77,17 @@ class Settings(BaseSettings): None, description="Anthropic API key for SDK (optional if CLI logged in)", ) + anthropic_base_url: Optional[str] = Field( + None, + description="Base URL for Anthropic API (optional, defaults to standard Anthropic API)", + ) claude_model: Optional[str] = Field( None, description="Claude model to use (defaults to CLI default if unset)" ) + anthropic_models: Optional[List[str]] = Field( + None, + description="Available Claude models for user selection (comma-separated)", + ) claude_max_turns: int = Field( DEFAULT_CLAUDE_MAX_TURNS, description="Max conversation turns" ) @@ -301,10 +312,10 @@ def parse_int_list(cls, v: Any) -> Optional[List[int]]: return [int(uid) for uid in v] return v # type: ignore[no-any-return] - @field_validator("claude_allowed_tools", mode="before") + @field_validator("claude_allowed_tools", "anthropic_models", mode="before") @classmethod def parse_claude_allowed_tools(cls, v: Any) -> Optional[List[str]]: - """Parse comma-separated tool names.""" + """Parse comma-separated tool names and model lists.""" if v is None: return None if isinstance(v, str): @@ -327,6 +338,30 @@ def validate_approved_directory(cls, v: Any) -> Path: raise ValueError(f"Approved directory is not a directory: {path}") return path # type: ignore[no-any-return] + @field_validator("approved_directories", mode="before") + @classmethod + def validate_approved_directories(cls, v: Any) -> Optional[List[Path]]: + """Ensure all approved directories exist and are absolute.""" + if v is None: + return None + if isinstance(v, str): + # Parse comma-separated paths + paths = [p.strip() for p in v.split(",") if p.strip()] + v = [Path(p) for p in paths] + if isinstance(v, list): + validated_paths = [] + for path_item in v: + if isinstance(path_item, str): + path_item = Path(path_item) + path = path_item.resolve() + if not path.exists(): + raise ValueError(f"Approved directory does not exist: {path}") + if not path.is_dir(): + raise ValueError(f"Approved directory is not a directory: {path}") + validated_paths.append(path) + return validated_paths + return v # type: ignore[no-any-return] + @field_validator("mcp_config_path", mode="before") @classmethod def validate_mcp_config(cls, v: Any, info: Any) -> Optional[Path]: @@ -452,6 +487,13 @@ def validate_cross_field_dependencies(self) -> "Settings": return self + @property + def effective_approved_directories(self) -> List[Path]: + """Get the list of approved directories to use for validation.""" + if self.approved_directories: + return self.approved_directories + return [self.approved_directory] + @property def is_production(self) -> bool: """Check if running in production mode.""" @@ -486,6 +528,11 @@ def anthropic_api_key_str(self) -> Optional[str]: else None ) + @property + def anthropic_base_url_str(self) -> Optional[str]: + """Get Anthropic base URL as string.""" + return self.anthropic_base_url + @property def mistral_api_key_str(self) -> Optional[str]: """Get Mistral API key as string.""" diff --git a/src/main.py b/src/main.py index 02660733..090f11e3 100644 --- a/src/main.py +++ b/src/main.py @@ -128,6 +128,7 @@ async def create_application(config: Settings) -> Dict[str, Any]: auth_manager = AuthenticationManager(providers) security_validator = SecurityValidator( config.approved_directory, + approved_directories=config.approved_directories, disable_security_patterns=config.disable_security_patterns, ) rate_limiter = RateLimiter(config) @@ -242,6 +243,7 @@ def signal_handler(signum: int, frame: Any) -> None: registry = load_project_registry( config_path=config.projects_config_path, approved_directory=config.approved_directory, + approved_directories=config.approved_directories, ) project_threads_manager = ProjectThreadManager( registry=registry, diff --git a/src/projects/registry.py b/src/projects/registry.py index 5609eac3..7d74848a 100644 --- a/src/projects/registry.py +++ b/src/projects/registry.py @@ -40,7 +40,7 @@ def get_by_slug(self, slug: str) -> Optional[ProjectDefinition]: def load_project_registry( - config_path: Path, approved_directory: Path + config_path: Path, approved_directory: Path, approved_directories: Optional[List[Path]] = None ) -> ProjectRegistry: """Load and validate project definitions from YAML.""" if not config_path.exists(): @@ -56,10 +56,14 @@ def load_project_registry( if not isinstance(raw_projects, list) or not raw_projects: raise ValueError("Projects config must contain a non-empty 'projects' list") - approved_root = approved_directory.resolve() + # Build list of all approved directories + all_approved_dirs = [approved_directory.resolve()] + if approved_directories: + all_approved_dirs.extend([d.resolve() for d in approved_directories]) + seen_slugs = set() seen_names = set() - seen_rel_paths = set() + seen_abs_paths = set() projects: List[ProjectDefinition] = [] for idx, raw in enumerate(raw_projects): @@ -68,28 +72,59 @@ def load_project_registry( slug = str(raw.get("slug", "")).strip() name = str(raw.get("name", "")).strip() - rel_path_raw = str(raw.get("path", "")).strip() + path_raw = str(raw.get("path", "")).strip() enabled = bool(raw.get("enabled", True)) if not slug: raise ValueError(f"Project entry at index {idx} is missing 'slug'") if not name: raise ValueError(f"Project '{slug}' is missing 'name'") - if not rel_path_raw: + if not path_raw: raise ValueError(f"Project '{slug}' is missing 'path'") - rel_path = Path(rel_path_raw) - if rel_path.is_absolute(): - raise ValueError(f"Project '{slug}' path must be relative: {rel_path_raw}") - - absolute_path = (approved_root / rel_path).resolve() - - try: - absolute_path.relative_to(approved_root) - except ValueError as e: - raise ValueError( - f"Project '{slug}' path outside approved " f"directory: {rel_path_raw}" - ) from e + path_obj = Path(path_raw) + + # Handle both absolute and relative paths + if path_obj.is_absolute(): + # Absolute path - validate it's within one of the approved directories + absolute_path = path_obj.resolve() + + # Check if path is within any approved directory + is_within_any = False + matched_base = None + for approved_dir in all_approved_dirs: + try: + rel = absolute_path.relative_to(approved_dir) + is_within_any = True + matched_base = approved_dir + relative_path = rel + break + except ValueError: + continue + + if not is_within_any: + raise ValueError( + f"Project '{slug}' absolute path is outside all approved directories: {path_raw}" + ) + else: + # Relative path - resolve against primary approved_directory + relative_path = path_obj + absolute_path = (approved_directory / relative_path).resolve() + + # Validate it's within one of the approved directories + is_within_any = False + for approved_dir in all_approved_dirs: + try: + absolute_path.relative_to(approved_dir) + is_within_any = True + break + except ValueError: + continue + + if not is_within_any: + raise ValueError( + f"Project '{slug}' path outside all approved directories: {path_raw}" + ) if not absolute_path.exists() or not absolute_path.is_dir(): raise ValueError( @@ -97,23 +132,23 @@ def load_project_registry( f"is not a directory: {absolute_path}" ) - rel_path_norm = str(rel_path) + abs_path_str = str(absolute_path) if slug in seen_slugs: raise ValueError(f"Duplicate project slug: {slug}") if name in seen_names: raise ValueError(f"Duplicate project name: {name}") - if rel_path_norm in seen_rel_paths: - raise ValueError(f"Duplicate project path: {rel_path_norm}") + if abs_path_str in seen_abs_paths: + raise ValueError(f"Duplicate project path: {abs_path_str}") seen_slugs.add(slug) seen_names.add(name) - seen_rel_paths.add(rel_path_norm) + seen_abs_paths.add(abs_path_str) projects.append( ProjectDefinition( slug=slug, name=name, - relative_path=rel_path, + relative_path=relative_path, absolute_path=absolute_path, enabled=enabled, ) diff --git a/src/security/validators.py b/src/security/validators.py index 381ba321..044ed1a0 100644 --- a/src/security/validators.py +++ b/src/security/validators.py @@ -132,14 +132,17 @@ class SecurityValidator: ] def __init__( - self, approved_directory: Path, disable_security_patterns: bool = False + self, approved_directory: Path, approved_directories: Optional[List[Path]] = None, disable_security_patterns: bool = False ): - """Initialize validator with approved directory.""" + """Initialize validator with approved directory/directories.""" self.approved_directory = approved_directory.resolve() + self.approved_directories = [d.resolve() for d in approved_directories] if approved_directories else [] + self.all_approved_directories = [self.approved_directory] + self.approved_directories self.disable_security_patterns = disable_security_patterns logger.info( "Security validator initialized", approved_directory=str(self.approved_directory), + approved_directories=[str(d) for d in self.approved_directories], disable_security_patterns=self.disable_security_patterns, ) @@ -186,15 +189,20 @@ def validate_path( # Resolve path and check boundaries target = target.resolve() - # Ensure target is within approved directory - if not self._is_within_directory(target, self.approved_directory): + # Ensure target is within any of the approved directories + is_within_any = any( + self._is_within_directory(target, approved_dir) + for approved_dir in self.all_approved_directories + ) + + if not is_within_any: logger.warning( "Path traversal attempt detected", requested_path=user_path, resolved_path=str(target), - approved_directory=str(self.approved_directory), + approved_directories=[str(d) for d in self.all_approved_directories], ) - return False, None, "Access denied: path outside approved directory" + return False, None, "Access denied: path outside approved directories" logger.debug( "Path validation successful", diff --git a/tests/unit/test_projects/test_registry.py b/tests/unit/test_projects/test_registry.py index cf421d0a..1bb571c1 100644 --- a/tests/unit/test_projects/test_registry.py +++ b/tests/unit/test_projects/test_registry.py @@ -71,4 +71,4 @@ def test_load_project_registry_rejects_outside_approved_dir(tmp_path: Path) -> N with pytest.raises(ValueError) as exc_info: load_project_registry(config_file, approved) - assert "outside approved directory" in str(exc_info.value) + assert "outside all approved directories" in str(exc_info.value)