diff --git a/README.md b/README.md index dcdfb08..f55ca97 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,193 @@ -# ChatSSH +# Terminal Chat -SSH Chat Server \ No newline at end of file +A feature-rich terminal-based chat application built with Python and Textual, supporting channels, direct messages, and real-time communication. + +## Features + +- **Global and custom channels** - Create public or private channels +- **Direct messaging** - Private conversations between users +- **Color-coded users** - Each user gets a unique color +- **Rich terminal UI** - Built with Textual framework +- **Command-based interface** - Easy-to-use slash commands +- **Live user list** - See who's online in real-time +- **Private channels** - Invite-only channels with access control + +## Installation + +1. Clone the repository: +```bash +git clone +cd ChatSSH +``` + +2. Install dependencies: +```bash +pip install -r requirements.txt +``` + +## Usage + +### Basic Usage + +Run the chat application: +```bash +python main.py +``` + +Or use the legacy entry point: +```bash +python server.py +``` + +You can optionally specify a username: +```bash +python main.py --username YourName +``` + +### Hosting as a Server + +To host the chat server that others can connect to via SSH: + +```bash +python main.py --ssh --host 0.0.0.0 --port 8022 +``` + +Options: +- `--ssh` - Enable SSH server mode +- `--host` - Host address to bind to (default: 0.0.0.0, listens on all interfaces) +- `--port` - Port number for SSH server (default: 8022) +- `--username` - Username for local client mode (not used in SSH mode) + +### Connecting to the SSH Server + +Once the server is running, users can connect using SSH: + +```bash +ssh -p 8022 username@server-address +``` + +**Authentication:** +- **Password:** Any password will be accepted (or just press Enter) +- For demo purposes, authentication is permissive +- In production, implement proper authentication in `ssh_server.py` + +**Example:** +```bash +# Connect to localhost +ssh -p 2222 alice@localhost +# When prompted for password, type anything or press Enter + +# Connect from another machine +ssh -p 8022 bob@192.168.1.100 +``` + +## Available Commands + +| Command | Shortcut | Description | +|---------|----------|-------------| +| `/dm [msg]` | `/d` | Send a direct message or switch to DM view | +| `/create [private]` | `/new` | Create a new channel (optionally private) | +| `/join ` | `/j` | Join an existing channel | +| `/leave ` | `/l` | Leave a channel | +| `/invite ` | `/inv` | Invite a user to a channel | +| `/delete ` | `/del` | Delete a channel you created | +| `/switch ` | `/s` | Switch to a different channel | +| `/channels` | `/ch` | List all your channels | +| `/users` | `/u` | List all online users | +| `/clear` | `/c` | Clear local chat history | +| `/quit` | `/q` | Exit the application | +| `/help` | `/h` | Show help message | + +## Project Structure + +``` +ChatSSH/ +├── constants.py # Color palettes and system constants +├── models.py # Data models (Channel, DirectMessage, UserConnection) +├── chat_server.py # Core chat server logic and state management +├── ui.py # Terminal UI components (Textual app) +├── main.py # Main entry point +├── server.py # Backwards-compatible wrapper +└── requirements.txt # Python dependencies +``` + +## Module Overview + +### `constants.py` +Defines color schemes for users and system messages. + +### `models.py` +Contains data models: +- `Channel` - Represents a chat channel with members and messages +- `DirectMessage` - Manages DM conversations between two users +- `UserConnection` - Handles user connection and message queuing + +### `chat_server.py` +Core server logic: +- `ChatServer` - Central server managing users, channels, and messages +- User management (add/remove users, assign colors) +- Channel operations (create, delete, join, leave, invite) +- Message routing (channel messages, DMs, broadcasts) + +### `ui.py` +Terminal user interface: +- `TerminalChatApp` - Main Textual application +- `ChatHistory` - Custom widget for displaying chat messages +- Command processing and view switching +- Real-time message updates + +### `main.py` +Application entry point with command-line argument parsing. + +## Requirements + +- Python 3.10+ +- textual +- asyncio (built-in) + +See `requirements.txt` for full dependency list. + +## Development + +The application uses a modular architecture for easy maintenance and extension: + +1. **Constants** - Centralized configuration +2. **Models** - Clean data structures +3. **Server Logic** - Business logic separate from UI +4. **UI Layer** - Textual-based interface + +To extend functionality: +- Add new commands in `ui.py` (`process_command` method) +- Implement server-side logic in `chat_server.py` +- Define data models in `models.py` + +## Keyboard Shortcuts + +- `Ctrl+C` or `Ctrl+D` - Quit application +- `Ctrl+H` - Show help +- `Enter` - Send message + +## Tips + +- Start in the `#global` channel by default +- Use `/dm ` to switch to direct message mode +- Private channels require an invitation to join +- Channel creators can delete their channels +- Maximum 500 messages stored per channel/DM + +## Future Enhancements + +- SSH server support for remote connections +- Message persistence (database storage) +- User authentication +- File sharing +- Message reactions and threads +- Search functionality + +## License + +[Add your license here] + +## Contributing + +[Add contribution guidelines here] diff --git a/chatapp/__init__.py b/chatapp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/chatapp/chat_server.py b/chatapp/chat_server.py new file mode 100644 index 0000000..fab29f3 --- /dev/null +++ b/chatapp/chat_server.py @@ -0,0 +1,251 @@ +"""Chat server logic and state management.""" + +import random +from datetime import datetime +from typing import Dict, Set, List + +from .constants import USER_COLORS +from .models import Channel, DirectMessage, UserConnection + + +class ChatServer: + """Central chat server managing users, channels, and messages.""" + + def __init__(self): + self.users: Dict[str, UserConnection] = {} + self.user_colors: Dict[str, str] = {} + self.available_colors = USER_COLORS.copy() + self.channels: Dict[str, Channel] = { + 'global': Channel('global', 'system', is_private=False) + } + self.dms: Dict[frozenset, DirectMessage] = {} + + def get_user_color(self, username: str) -> str: + if username not in self.user_colors: + if not self.available_colors: + self.available_colors = USER_COLORS.copy() + color = random.choice(self.available_colors) + self.available_colors.remove(color) + self.user_colors[username] = color + return self.user_colors[username] + + def release_user_color(self, username: str): + if username in self.user_colors: + color = self.user_colors[username] + if color not in self.available_colors: + self.available_colors.append(color) + del self.user_colors[username] + + def add_user(self, username: str, connection: UserConnection): + self.users[username] = connection + self.get_user_color(username) + self.channels['global'].add_member(username) + self.broadcast_to_channel('global', { + 'type': 'system', + 'message': f"{username} has joined the chat", + 'timestamp': datetime.now().strftime("%H:%M:%S") + }) + self.broadcast_user_list() + + def remove_user(self, username: str): + if username in self.users: + # Remove from all channels + for channel in self.channels.values(): + if channel.has_member(username): + channel.remove_member(username) + if channel.name != 'global' and len(channel.members) == 0: + # Delete empty non-global channels + pass + + del self.users[username] + self.release_user_color(username) + + self.broadcast_to_channel('global', { + 'type': 'system', + 'message': f"{username} has left the chat", + 'timestamp': datetime.now().strftime("%H:%M:%S") + }) + self.broadcast_user_list() + + def create_channel(self, name: str, creator: str, is_private: bool = False) -> bool: + if name in self.channels: + return False + self.channels[name] = Channel(name, creator, is_private) + return True + + def delete_channel(self, name: str, username: str) -> bool: + if name not in self.channels or name == 'global': + return False + channel = self.channels[name] + if channel.creator != username: + return False + + # Notify all members + self.broadcast_to_channel(name, { + 'type': 'system', + 'message': f"Channel #{name} has been deleted by {username}", + 'timestamp': datetime.now().strftime("%H:%M:%S") + }) + + del self.channels[name] + return True + + def invite_to_channel(self, channel_name: str, inviter: str, invitee: str) -> tuple[bool, str]: + if channel_name not in self.channels: + return False, f"Channel #{channel_name} does not exist" + + channel = self.channels[channel_name] + + if not channel.has_member(inviter): + return False, f"You are not a member of #{channel_name}" + + if invitee not in self.users: + return False, f"User '{invitee}' is not online" + + if channel.has_member(invitee): + return False, f"{invitee} is already in #{channel_name}" + + channel.invite(invitee) + + # Send invite notification to invitee + if invitee in self.users: + self.users[invitee].send_message({ + 'type': 'info', + 'message': f"{inviter} invited you to #{channel_name}. Use /join {channel_name} to accept", + 'timestamp': datetime.now().strftime("%H:%M:%S") + }) + + return True, f"Invited {invitee} to #{channel_name}" + + def join_channel(self, channel_name: str, username: str) -> tuple[bool, str]: + if channel_name not in self.channels: + return False, f"Channel #{channel_name} does not exist" + + channel = self.channels[channel_name] + + if channel.has_member(username): + return False, f"You are already in #{channel_name}" + + if channel.is_private and not channel.is_invited(username): + return False, f"#{channel_name} is private. You need an invitation to join" + + channel.add_member(username) + + self.broadcast_to_channel(channel_name, { + 'type': 'system', + 'message': f"{username} joined #{channel_name}", + 'timestamp': datetime.now().strftime("%H:%M:%S") + }) + + return True, f"Joined #{channel_name}" + + def leave_channel(self, channel_name: str, username: str) -> tuple[bool, str]: + if channel_name == 'global': + return False, "Cannot leave #global" + + if channel_name not in self.channels: + return False, f"Channel #{channel_name} does not exist" + + channel = self.channels[channel_name] + + if not channel.has_member(username): + return False, f"You are not in #{channel_name}" + + channel.remove_member(username) + + self.broadcast_to_channel(channel_name, { + 'type': 'system', + 'message': f"{username} left #{channel_name}", + 'timestamp': datetime.now().strftime("%H:%M:%S") + }) + + return True, f"Left #{channel_name}" + + def get_or_create_dm(self, user1: str, user2: str) -> DirectMessage: + dm_key = frozenset([user1, user2]) + if dm_key not in self.dms: + self.dms[dm_key] = DirectMessage(user1, user2) + return self.dms[dm_key] + + def send_dm(self, from_user: str, to_user: str, message: str): + timestamp = datetime.now().strftime("%H:%M:%S") + from_color = self.get_user_color(from_user) + + if to_user not in self.users: + self.users[from_user].send_message({ + 'type': 'error', + 'message': f"User '{to_user}' is not online", + 'timestamp': timestamp + }) + return + + dm = self.get_or_create_dm(from_user, to_user) + + msg_data = { + 'type': 'dm', + 'from': from_user, + 'to': to_user, + 'message': message, + 'timestamp': timestamp, + 'color': from_color + } + + dm.add_message(msg_data) + + # Send to both users + self.users[from_user].send_message(msg_data) + self.users[to_user].send_message(msg_data) + + def broadcast_to_channel(self, channel_name: str, msg_data: dict): + if channel_name not in self.channels: + return + + channel = self.channels[channel_name] + channel.add_message(msg_data) + + for member in channel.members: + if member in self.users: + self.users[member].send_message(msg_data) + + def send_channel_message(self, channel_name: str, username: str, message: str): + if channel_name not in self.channels: + return + + channel = self.channels[channel_name] + + if not channel.has_member(username): + self.users[username].send_message({ + 'type': 'error', + 'message': f"You are not a member of #{channel_name}", + 'timestamp': datetime.now().strftime("%H:%M:%S") + }) + return + + timestamp = datetime.now().strftime("%H:%M:%S") + color = self.get_user_color(username) + + msg_data = { + 'type': 'message', + 'channel': channel_name, + 'username': username, + 'message': message, + 'timestamp': timestamp, + 'color': color + } + + self.broadcast_to_channel(channel_name, msg_data) + + def broadcast_user_list(self): + users_with_colors = [ + {'username': u, 'color': self.get_user_color(u)} + for u in self.users.keys() + ] + msg_data = { + 'type': 'user_list', + 'users': users_with_colors + } + for user_conn in self.users.values(): + user_conn.send_message(msg_data) + + def get_user_channels(self, username: str) -> List[str]: + return [name for name, channel in self.channels.items() if channel.has_member(username)] diff --git a/chatapp/constants.py b/chatapp/constants.py new file mode 100644 index 0000000..4e65857 --- /dev/null +++ b/chatapp/constants.py @@ -0,0 +1,23 @@ +"""Color palettes and constants for the chat application.""" + +# Color palette for users +USER_COLORS = [ + "#FF6B6B", "#4ECDC4", "#45B7D1", "#FFA07A", "#98D8C8", "#F7DC6F", + "#BB8FCE", "#85C1E2", "#F8B739", "#52B788", "#E63946", "#06FFA5", + "#FB5607", "#8338EC", "#3A86FF", "#FF006E", "#FFBE0B", "#06D6A0", + "#EF476F", "#118AB2", +] + +# System colors +COLORS = { + 'system': '#FFA500', + 'error': '#FF4444', + 'success': '#00FF00', + 'info': '#4A9EFF', + 'whisper': '#FF00FF', + 'command': '#00FFFF', + 'warning': '#FFFF00', + 'channel': '#00FF00', + 'dm': '#FF00FF', + 'help': '#00FF00', +} diff --git a/chatapp/models.py b/chatapp/models.py new file mode 100644 index 0000000..caf8448 --- /dev/null +++ b/chatapp/models.py @@ -0,0 +1,70 @@ +"""Data models for chat application.""" + +import asyncio +from typing import Set + + +class Channel: + """Represents a chat channel or group.""" + def __init__(self, name: str, creator: str, is_private: bool = False): + self.name = name + self.creator = creator + self.is_private = is_private + self.members: Set[str] = {creator} + self.messages = [] + self.invites: Set[str] = set() # Pending invites + self.max_messages = 500 + + def add_member(self, username: str): + self.members.add(username) + if username in self.invites: + self.invites.remove(username) + + def remove_member(self, username: str): + if username in self.members: + self.members.remove(username) + + def add_message(self, msg_data: dict): + self.messages.append(msg_data) + if len(self.messages) > self.max_messages: + self.messages.pop(0) + + def has_member(self, username: str) -> bool: + return username in self.members + + def is_invited(self, username: str) -> bool: + return username in self.invites + + def invite(self, username: str): + if username not in self.members: + self.invites.add(username) + + +class DirectMessage: + """Represents a DM conversation between two users.""" + def __init__(self, user1: str, user2: str): + self.users = frozenset([user1, user2]) + self.messages = [] + self.max_messages = 500 + + def add_message(self, msg_data: dict): + self.messages.append(msg_data) + if len(self.messages) > self.max_messages: + self.messages.pop(0) + + def get_other_user(self, username: str) -> str: + users_list = list(self.users) + return users_list[0] if users_list[1] == username else users_list[1] + + +class UserConnection: + """Represents a connected user with a message queue.""" + def __init__(self, username: str, message_queue: asyncio.Queue): + self.username = username + self.message_queue = message_queue + + def send_message(self, msg_data: dict): + try: + self.message_queue.put_nowait(msg_data) + except: + pass diff --git a/chatapp/server.py b/chatapp/server.py new file mode 100644 index 0000000..fddec37 --- /dev/null +++ b/chatapp/server.py @@ -0,0 +1,40 @@ +""" +Backwards-compatible wrapper for the refactored Terminal Chat application. + +This file maintains the original server.py interface but imports from the new modular structure. +For new code, import directly from the specific modules (constants, models, chat_server, ui, ssh_server). + +To run the application, use: python main.py +""" + +# Re-export all components for backwards compatibility +from constants import USER_COLORS, COLORS +from models import Channel, DirectMessage, UserConnection +from chat_server import ChatServer +from ui import ChatHistory, TerminalChatApp +from ssh_server import start_ssh_server, SSHChatServer, handle_client +from main import start_terminal_chat, main + +# Create a global chat_server instance for backwards compatibility +chat_server = ChatServer() + +# Re-export for compatibility +__all__ = [ + 'USER_COLORS', + 'COLORS', + 'Channel', + 'DirectMessage', + 'UserConnection', + 'ChatServer', + 'chat_server', + 'ChatHistory', + 'TerminalChatApp', + 'start_terminal_chat', + 'start_ssh_server', + 'SSHChatServer', + 'handle_client', + 'main', +] + +if __name__ == "__main__": + main() diff --git a/chatapp/ssh_server.py b/chatapp/ssh_server.py new file mode 100644 index 0000000..7c71f9c --- /dev/null +++ b/chatapp/ssh_server.py @@ -0,0 +1,109 @@ +"""SSH server implementation for Terminal Chat.""" + +import asyncio +import asyncssh +import os +import sys +from typing import Optional + +from .models import UserConnection +from .chat_server import ChatServer +from simple_client import SimpleTextChatClient + + +class SSHChatServer(asyncssh.SSHServer): + """SSH server that handles authentication for the chat.""" + + def __init__(self, chat_server: ChatServer): + self.chat_server = chat_server + + def begin_auth(self, username: str) -> bool: + """Allow any username to connect.""" + return True + + def password_auth_supported(self) -> bool: + """Enable password authentication.""" + return True + + def validate_password(self, username: str, password: str) -> bool: + """Accept any password for simplicity (for demo purposes).""" + # In production, you'd want proper authentication + # For now, just return True to accept any password + return True + + def public_key_auth_supported(self) -> bool: + """Disable public key auth for simplicity.""" + return False + + +async def handle_client(process: asyncssh.SSHServerProcess, chat_server: ChatServer): + """Handle an SSH client connection and run the chat app.""" + username = process.channel.get_connection().get_extra_info('username') + + # Check if username is already taken + if username in chat_server.users: + base_username = username + counter = 1 + while username in chat_server.users: + username = f"{base_username}{counter}" + counter += 1 + + # Create message queue and user connection + message_queue = asyncio.Queue() + user_conn = UserConnection(username, message_queue) + chat_server.add_user(username, user_conn) + + try: + # Create and run the simple text chat client + client = SimpleTextChatClient(username, message_queue, chat_server, process) + await client.run() + except Exception as e: + process.stderr.write(f"Error: {e}\n") + finally: + # Clean up + if username in chat_server.users: + chat_server.remove_user(username) + process.exit(0) + + +async def start_ssh_server(host: str, port: int, chat_server: ChatServer): + """Start the SSH server for the chat application. + + Args: + host: The host address to bind to + port: The port number to listen on + chat_server: The ChatServer instance to use + """ + # Generate host key if it doesn't exist + host_key_path = 'ssh_host_key' + if not os.path.exists(host_key_path): + print(f"Generating SSH host key at {host_key_path}...") + key = asyncssh.generate_private_key('ssh-rsa') + key.write_private_key(host_key_path) + print("Host key generated successfully") + + async def process_factory(process: asyncssh.SSHServerProcess): + """Factory function to handle SSH processes.""" + await handle_client(process, chat_server) + + print(f"Starting SSH server on {host}:{port}...") + print(f"Users can connect with: ssh -p {port} @{host}") + print("Password: any password will work (press Enter for empty password)") + print("Press Ctrl+C to stop the server") + print("") + + try: + await asyncssh.listen( + host, + port, + server_factory=lambda: SSHChatServer(chat_server), + server_host_keys=[host_key_path], + process_factory=process_factory, + encoding='utf-8', + ) + + # Keep the server running + await asyncio.Event().wait() + except (OSError, asyncssh.Error) as e: + print(f"Error starting SSH server: {e}") + raise diff --git a/chatapp/ui.py b/chatapp/ui.py new file mode 100644 index 0000000..9c8cc5c --- /dev/null +++ b/chatapp/ui.py @@ -0,0 +1,489 @@ +"""Terminal UI components using Textual framework.""" + +import asyncio +from datetime import datetime +from typing import List + +from textual.app import App, ComposeResult +from textual.containers import Container, Vertical, Horizontal, ScrollableContainer +from textual.widgets import Header, Footer, Input, Static, RichLog +from textual.binding import Binding + +from .constants import COLORS + + +class ChatHistory(RichLog): + """Custom RichLog widget for chat history display.""" + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs, highlight=True, markup=True) + self.can_focus = False + + +class TerminalChatApp(App): + """Main terminal chat application UI.""" + + CSS = """ + Screen { + background: #1a1a1a; + } + + Header { + background: #2d2d2d; + color: #00ff00; + text-style: bold; + } + + Footer { + background: #2d2d2d; + } + + #main-container { + height: 1fr; + layout: horizontal; + background: #1a1a1a; + } + + #chat-panel { + width: 3fr; + height: 1fr; + layout: vertical; + } + + #chat-container { + height: 1fr; + border: heavy #00ff00; + background: #0a0a0a; + padding: 0; + } + + ChatHistory { + background: #0a0a0a; + color: #ffffff; + padding: 1 2; + border: none; + height: 1fr; + } + + #input-container { + height: auto; + padding: 1 2; + background: #1a1a1a; + border-top: heavy #00ff00; + } + + #user-input { + width: 1fr; + background: #2d2d2d; + border: solid #00ff00; + } + + #user-input:focus { + border: heavy #00ff00; + } + + #sidebar { + width: 22; + height: 1fr; + layout: vertical; + margin-left: 1; + } + + #channels-panel { + height: 1fr; + border: heavy #00ff00; + background: #0a0a0a; + padding: 1; + margin-bottom: 1; + } + + #users-panel { + height: 1fr; + border: heavy #00ff00; + background: #0a0a0a; + padding: 1; + } + + .panel-title { + color: #00ff00; + text-style: bold; + text-align: center; + margin-bottom: 1; + background: #1a1a1a; + padding: 1; + border-bottom: solid #00ff00; + } + + .panel-list { + height: 1fr; + padding: 1 0; + } + + .list-item { + padding: 0 1; + margin: 0 0 1 0; + } + + #status-bar { + dock: bottom; + height: 1; + background: #2d2d2d; + color: #00ff00; + padding: 0 2; + text-style: bold; + } + + #help-bar { + height: auto; + background: #1a1a1a; + color: #888888; + padding: 1 2; + border-top: solid #444444; + text-align: center; + } + """ + + BINDINGS = [ + Binding("ctrl+c", "quit", "Quit", show=False), + Binding("ctrl+d", "quit", "Quit", show=False), + Binding("ctrl+h", "show_help", "Help", show=True), + ] + + TITLE = "Terminal Chat" + + def __init__(self, username: str, message_queue: asyncio.Queue, chat_server): + super().__init__() + self.username = username + self.message_queue = message_queue + self.chat_server = chat_server + self.current_view = 'global' # Can be channel name or 'dm:username' + self.users_list = [] + + def compose(self) -> ComposeResult: + yield Header() + + with Container(id="main-container"): + with Vertical(id="chat-panel"): + with Container(id="chat-container"): + yield ChatHistory(id="chat-history") + + with Horizontal(id="input-container"): + yield Input( + placeholder="Type a message or /help for commands...", + id="user-input" + ) + + with Vertical(id="sidebar"): + with Vertical(id="channels-panel"): + yield Static("CHANNELS", classes="panel-title") + yield ScrollableContainer(id="channels-list", classes="panel-list") + + with Vertical(id="users-panel"): + yield Static("ONLINE USERS", classes="panel-title") + yield ScrollableContainer(id="users-list", classes="panel-list") + + yield Static( + "/dm - DM | /create - create | /join - join | /help - help", + id="help-bar" + ) + yield Static(f"Connected as: {self.username} | View: #global", id="status-bar") + yield Footer() + + def on_mount(self) -> None: + self.query_one("#user-input").focus() + self.display_info_message(f"Welcome to Terminal Chat, {self.username}!") + self.display_info_message("Type /help to see available commands") + self.set_interval(0.1, self.check_messages) + self.update_channels_list() + + def update_status_bar(self): + view_text = f"#{self.current_view}" if not self.current_view.startswith('dm:') else f"DM: {self.current_view[3:]}" + self.query_one("#status-bar", Static).update(f"Connected as: {self.username} | View: {view_text}") + + def update_channels_list(self): + channels_container = self.query_one("#channels-list", ScrollableContainer) + channels_container.remove_children() + + user_channels = self.chat_server.get_user_channels(self.username) + for channel_name in sorted(user_channels): + prefix = "> " if channel_name == self.current_view else " " + color = COLORS['channel'] if channel_name == self.current_view else "#888888" + channel_widget = Static(f"{prefix}[{color}]#{channel_name}[/{color}]", classes="list-item", markup=True) + channels_container.mount(channel_widget) + + def display_user_message(self, username: str, message: str, timestamp: str, color: str, channel: str = None): + chat_history = self.query_one(ChatHistory) + chat_history.write(f"[dim]{timestamp}[/dim] [{color}]{username}[/{color}]: {message}") + + def display_system_message(self, message: str, timestamp: str): + chat_history = self.query_one(ChatHistory) + chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['system']}][SYSTEM][/{COLORS['system']}] {message}") + + def display_error_message(self, message: str, timestamp: str = None): + if not timestamp: + timestamp = datetime.now().strftime("%H:%M:%S") + chat_history = self.query_one(ChatHistory) + chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['error']}][ERROR][/{COLORS['error']}] {message}") + + def display_info_message(self, message: str): + timestamp = datetime.now().strftime("%H:%M:%S") + chat_history = self.query_one(ChatHistory) + chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['info']}][INFO][/{COLORS['info']}] {message}") + + def display_success_message(self, message: str): + timestamp = datetime.now().strftime("%H:%M:%S") + chat_history = self.query_one(ChatHistory) + chat_history.write(f"[dim]{timestamp}[/dim] [{COLORS['success']}][SUCCESS][/{COLORS['success']}] {message}") + + def display_dm(self, from_user: str, message: str, timestamp: str, color: str): + chat_history = self.query_one(ChatHistory) + chat_history.write(f"[dim]{timestamp}[/dim] [{color}]{from_user}[/{color}]: {message}") + + def switch_view(self, view: str): + self.current_view = view + self.update_status_bar() + self.update_channels_list() + + chat_history = self.query_one(ChatHistory) + chat_history.clear() + + if view.startswith('dm:'): + other_user = view[3:] + dm = self.chat_server.get_or_create_dm(self.username, other_user) + for msg in dm.messages: + self.display_dm(msg['from'], msg['message'], msg['timestamp'], msg['color']) + else: + if view in self.chat_server.channels: + channel = self.chat_server.channels[view] + for msg in channel.messages: + if msg['type'] == 'system': + self.display_system_message(msg['message'], msg['timestamp']) + elif msg['type'] == 'message': + self.display_user_message(msg['username'], msg['message'], msg['timestamp'], msg['color']) + + def check_messages(self): + try: + while not self.message_queue.empty(): + msg_data = self.message_queue.get_nowait() + + # Only display if in the right view + if msg_data['type'] == 'message': + if msg_data.get('channel') == self.current_view: + self.display_user_message( + msg_data['username'], + msg_data['message'], + msg_data['timestamp'], + msg_data.get('color', '#FFFFFF') + ) + + elif msg_data['type'] == 'dm': + other_user = msg_data['from'] if msg_data['from'] != self.username else msg_data['to'] + if self.current_view == f"dm:{other_user}": + self.display_dm( + msg_data['from'], + msg_data['message'], + msg_data['timestamp'], + msg_data.get('color', '#FFFFFF') + ) + + elif msg_data['type'] == 'system': + if self.current_view == 'global': + self.display_system_message(msg_data['message'], msg_data['timestamp']) + + elif msg_data['type'] == 'error': + self.display_error_message(msg_data['message'], msg_data['timestamp']) + + elif msg_data['type'] == 'info': + self.display_info_message(msg_data['message']) + + elif msg_data['type'] == 'user_list': + self.update_user_list(msg_data['users']) + self.update_channels_list() + except: + pass + + def update_user_list(self, users: list): + self.users_list = users + users_container = self.query_one("#users-list", ScrollableContainer) + users_container.remove_children() + + for user_data in sorted(users, key=lambda x: x['username']): + username = user_data['username'] + color = user_data['color'] + prefix = "> " if username == self.username else " " + user_widget = Static(f"{prefix}[{color}]{username}[/{color}]", classes="list-item", markup=True) + users_container.mount(user_widget) + + def process_command(self, message: str) -> bool: + if not message.startswith('/'): + return False + + parts = message.split(' ', 2) + command = parts[0].lower() + + if command == '/help' or command == '/h': + self.show_help_message() + return True + + elif command == '/dm' or command == '/d': + if len(parts) < 2: + self.display_error_message("Usage: /dm [message]") + return True + + target_user = parts[1] + if target_user == self.username: + self.display_error_message("You cannot DM yourself") + return True + + if target_user not in self.chat_server.users: + self.display_error_message(f"User '{target_user}' is not online") + return True + + if len(parts) == 3: + self.chat_server.send_dm(self.username, target_user, parts[2]) + else: + self.switch_view(f"dm:{target_user}") + return True + + elif command == '/create' or command == '/new': + if len(parts) < 2: + self.display_error_message("Usage: /create [private]") + return True + + channel_name = parts[1].lower() + is_private = len(parts) > 2 and parts[2].lower() == 'private' + + if self.chat_server.create_channel(channel_name, self.username, is_private): + self.display_success_message(f"Created channel #{channel_name}") + self.switch_view(channel_name) + else: + self.display_error_message(f"Channel #{channel_name} already exists") + return True + + elif command == '/join' or command == '/j': + if len(parts) < 2: + self.display_error_message("Usage: /join ") + return True + + channel_name = parts[1].lower() + success, msg = self.chat_server.join_channel(channel_name, self.username) + + if success: + self.display_success_message(msg) + self.switch_view(channel_name) + else: + self.display_error_message(msg) + return True + + elif command == '/leave' or command == '/l': + if len(parts) < 2: + self.display_error_message("Usage: /leave ") + return True + + channel_name = parts[1].lower() + success, msg = self.chat_server.leave_channel(channel_name, self.username) + + if success: + self.display_success_message(msg) + self.switch_view('global') + else: + self.display_error_message(msg) + return True + + elif command == '/invite' or command == '/inv': + if len(parts) < 3: + self.display_error_message("Usage: /invite ") + return True + + channel_name = parts[1].lower() + invitee = parts[2] + + success, msg = self.chat_server.invite_to_channel(channel_name, self.username, invitee) + + if success: + self.display_success_message(msg) + else: + self.display_error_message(msg) + return True + + elif command == '/delete' or command == '/del': + if len(parts) < 2: + self.display_error_message("Usage: /delete ") + return True + + channel_name = parts[1].lower() + + if self.chat_server.delete_channel(channel_name, self.username): + self.display_success_message(f"Deleted channel #{channel_name}") + self.switch_view('global') + else: + self.display_error_message(f"Cannot delete #{channel_name}. You must be the creator") + return True + + elif command == '/switch' or command == '/s': + if len(parts) < 2: + self.display_error_message("Usage: /switch ") + return True + + channel_name = parts[1].lower() + + if channel_name in self.chat_server.channels and self.chat_server.channels[channel_name].has_member(self.username): + self.switch_view(channel_name) + else: + self.display_error_message(f"You are not a member of #{channel_name}") + return True + + elif command == '/channels' or command == '/ch': + channels = self.chat_server.get_user_channels(self.username) + self.display_info_message(f"Your channels: {', '.join(['#' + c for c in channels])}") + return True + + elif command == '/users' or command == '/u': + usernames = [u['username'] for u in self.users_list] + self.display_info_message(f"Online users ({len(usernames)}): {', '.join(sorted(usernames))}") + return True + + elif command == '/clear' or command == '/c': + chat_history = self.query_one(ChatHistory) + chat_history.clear() + self.display_info_message("Chat history cleared locally") + return True + + elif command == '/quit' or command == '/q': + self.exit() + return True + + else: + self.display_error_message(f"Unknown command: {command}. Type /help for available commands") + return True + + def show_help_message(self): + self.display_info_message("Available commands:") + chat_history = self.query_one(ChatHistory) + chat_history.write(f" [{COLORS['help']}]/dm [msg][/{COLORS['help']}] or [{COLORS['command']}]/d[/{COLORS['command']}] - DM a user") + chat_history.write(f" [{COLORS['help']}]/create [private][/{COLORS['help']}] or [{COLORS['command']}]/new[/{COLORS['command']}] - Create a channel") + chat_history.write(f" [{COLORS['help']}]/join [/{COLORS['help']}] or [{COLORS['command']}]/j[/{COLORS['command']}] - Join a channel") + chat_history.write(f" [{COLORS['help']}]/leave [/{COLORS['help']}] or [{COLORS['command']}/]/l[/{COLORS['command']}] - Leave a channel") + chat_history.write(f" [{COLORS['help']}]/invite [/{COLORS['help']}] or [{COLORS['command']}]/inv[/{COLORS['command']}] - Invite user to channel") + chat_history.write(f" [{COLORS['help']}]/delete [/{COLORS['help']}] or [{COLORS['command']}]/del[/{COLORS['command']}] - Delete a channel you created") + chat_history.write(f" [{COLORS['help']}]/switch [/{COLORS['help']}] or [{COLORS['command']}]/s[/{COLORS['command']}] - Switch to a channel") + chat_history.write(f" [{COLORS['help']}]/channels[/{COLORS['help']}] or [{COLORS['command']}]/ch[/{COLORS['command']}] - List your channels") + chat_history.write(f" [{COLORS['help']}]/users[/{COLORS['help']}] or [{COLORS['command']}]/u[/{COLORS['command']}] - List online users") + chat_history.write(f" [{COLORS['help']}]/clear[/{COLORS['help']}] or [{COLORS['command']}]/c[/{COLORS['command']}] - Clear chat history locally") + chat_history.write(f" [{COLORS['help']}]/quit[/{COLORS['help']}] or [{COLORS['command']}]/q[/{COLORS['command']}] - Quit the chat application") + + def on_input_submitted(self, event: Input.Submitted) -> None: + message = event.value.strip() + event.input.value = "" + + if not message: + return + + if self.process_command(message): + return + + # Regular message + if self.current_view.startswith('dm:'): + target_user = self.current_view[3:] + self.chat_server.send_dm(self.username, target_user, message) + else: + self.chat_server.send_channel_message(self.current_view, self.username, message) diff --git a/main.py b/main.py new file mode 100644 index 0000000..2aa6a3c --- /dev/null +++ b/main.py @@ -0,0 +1,75 @@ +"""Main entry point for the Terminal Chat application.""" + +import asyncio +import os +import argparse + +from chatapp.models import UserConnection +from chatapp.chat_server import ChatServer +from chatapp.ui import TerminalChatApp +from chatapp.ssh_server import start_ssh_server + + +# Global chat server instance +chat_server = ChatServer() + + +async def start_terminal_chat(username: str): + """Start a TerminalChatApp for a single local user. + + This creates a UserConnection attached to an asyncio.Queue, registers + the user on the chat_server, runs the textual app asynchronously, and + ensures the user is removed when the app exits. + """ + message_queue = asyncio.Queue() + user_conn = UserConnection(username, message_queue) + chat_server.add_user(username, user_conn) + + try: + app = TerminalChatApp(username, message_queue, chat_server) + await app.run_async() + finally: + chat_server.remove_user(username) + + +def main(): + """Main entry point with argument parsing.""" + parser = argparse.ArgumentParser(description="Run Terminal Chat locally or via SSH") + parser.add_argument("--ssh", action="store_true", help="Start SSH server mode") + parser.add_argument("--host", default="0.0.0.0", help="Host to bind SSH server to (default: 0.0.0.0)") + parser.add_argument("--port", type=int, default=8022, help="Port for SSH server (default: 8022)") + parser.add_argument("--username", help="Username to use for the local terminal chat") + args = parser.parse_args() + + if args.ssh: + # SSH server mode + try: + asyncio.run(start_ssh_server(args.host, args.port, chat_server)) + except KeyboardInterrupt: + print("\nSSH server stopped") + raise SystemExit(0) + except Exception as exc: + print(f"Error starting SSH server: {exc}") + raise SystemExit(1) + else: + # Local terminal mode + username = args.username or os.getenv("USER") or os.getenv("USERNAME") + if not username: + try: + username = input("Enter username: ").strip() + except KeyboardInterrupt: + raise SystemExit(1) + + if not username: + print("A username is required to start the chat.") + raise SystemExit(1) + + try: + asyncio.run(start_terminal_chat(username)) + except Exception as exc: + print(f"Error while running Terminal Chat: {exc}") + raise + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..33139cc --- /dev/null +++ b/requirements.txt @@ -0,0 +1,15 @@ +asyncssh==2.21.1 +cffi==2.0.0 +cryptography==46.0.3 +linkify-it-py==2.0.3 +markdown-it-py==4.0.0 +mdit-py-plugins==0.5.0 +mdurl==0.1.2 +platformdirs==4.5.0 +pycparser==2.23 +Pygments==2.19.2 +rich==14.2.0 +textual==6.6.0 +typing_extensions==4.15.0 +uc-micro-py==1.0.3 +pytest diff --git a/simple_client.py b/simple_client.py new file mode 100644 index 0000000..32e2943 --- /dev/null +++ b/simple_client.py @@ -0,0 +1,245 @@ +"""Simple text-based chat client for SSH connections.""" + +import asyncio +from datetime import datetime +from typing import Optional + +from constants import COLORS +from chat_server import ChatServer +from models import UserConnection + + +class SimpleTextChatClient: + """Simple text-based chat client that works over SSH.""" + + def __init__(self, username: str, message_queue: asyncio.Queue, chat_server: ChatServer, process): + self.username = username + self.message_queue = message_queue + self.chat_server = chat_server + self.process = process + self.current_view = 'global' + self.running = True + + async def run(self): + """Run the chat client.""" + # Send welcome message + self.write_line("=" * 60) + self.write_line(f"Welcome to Terminal Chat, {self.username}!") + self.write_line("=" * 60) + self.write_line("Type /help for commands or start chatting!") + self.write_line(f"Current channel: #{self.current_view}") + self.write_line("") + + # Start message processor + message_task = asyncio.create_task(self.process_messages()) + + # Start input handler + try: + while self.running: + self.process.stdout.write(f"[{self.current_view}] > ") + line = await self.process.stdin.readline() + + if not line: + break + + message = line.strip() + if message: + await self.handle_input(message) + except Exception as e: + self.write_line(f"Error: {e}") + finally: + self.running = False + message_task.cancel() + + def write_line(self, text: str): + """Write a line to the client.""" + self.process.stdout.write(f"{text}\n") + + async def process_messages(self): + """Process incoming messages from the queue.""" + while self.running: + try: + msg_data = await asyncio.wait_for(self.message_queue.get(), timeout=0.1) + + if msg_data['type'] == 'message': + if msg_data.get('channel') == self.current_view: + self.write_line( + f"[{msg_data['timestamp']}] {msg_data['username']}: {msg_data['message']}" + ) + + elif msg_data['type'] == 'dm': + other_user = msg_data['from'] if msg_data['from'] != self.username else msg_data['to'] + if self.current_view == f"dm:{other_user}": + self.write_line( + f"[{msg_data['timestamp']}] {msg_data['from']}: {msg_data['message']}" + ) + + elif msg_data['type'] == 'system': + if self.current_view == 'global': + self.write_line(f"[{msg_data['timestamp']}] [SYSTEM] {msg_data['message']}") + + elif msg_data['type'] == 'error': + self.write_line(f"[ERROR] {msg_data['message']}") + + elif msg_data['type'] == 'info': + self.write_line(f"[INFO] {msg_data['message']}") + + elif msg_data['type'] == 'user_list': + pass # Don't auto-display user list updates + + except asyncio.TimeoutError: + continue + except Exception as e: + continue + + async def handle_input(self, message: str): + """Handle user input.""" + if message.startswith('/'): + await self.handle_command(message) + else: + # Send regular message + if self.current_view.startswith('dm:'): + target_user = self.current_view[3:] + self.chat_server.send_dm(self.username, target_user, message) + else: + self.chat_server.send_channel_message(self.current_view, self.username, message) + + async def handle_command(self, message: str): + """Handle chat commands.""" + parts = message.split(' ', 2) + command = parts[0].lower() + + if command in ['/help', '/h']: + self.show_help() + + elif command in ['/dm', '/d']: + if len(parts) < 2: + self.write_line("[ERROR] Usage: /dm [message]") + return + + target_user = parts[1] + if target_user == self.username: + self.write_line("[ERROR] You cannot DM yourself") + return + + if target_user not in self.chat_server.users: + self.write_line(f"[ERROR] User '{target_user}' is not online") + return + + if len(parts) == 3: + self.chat_server.send_dm(self.username, target_user, parts[2]) + else: + self.switch_view(f"dm:{target_user}") + + elif command in ['/create', '/new']: + if len(parts) < 2: + self.write_line("[ERROR] Usage: /create [private]") + return + + channel_name = parts[1].lower() + is_private = len(parts) > 2 and parts[2].lower() == 'private' + + if self.chat_server.create_channel(channel_name, self.username, is_private): + self.write_line(f"[SUCCESS] Created channel #{channel_name}") + self.switch_view(channel_name) + else: + self.write_line(f"[ERROR] Channel #{channel_name} already exists") + + elif command in ['/join', '/j']: + if len(parts) < 2: + self.write_line("[ERROR] Usage: /join ") + return + + channel_name = parts[1].lower() + success, msg = self.chat_server.join_channel(channel_name, self.username) + + if success: + self.write_line(f"[SUCCESS] {msg}") + self.switch_view(channel_name) + else: + self.write_line(f"[ERROR] {msg}") + + elif command in ['/leave', '/l']: + if len(parts) < 2: + self.write_line("[ERROR] Usage: /leave ") + return + + channel_name = parts[1].lower() + success, msg = self.chat_server.leave_channel(channel_name, self.username) + + if success: + self.write_line(f"[SUCCESS] {msg}") + self.switch_view('global') + else: + self.write_line(f"[ERROR] {msg}") + + elif command in ['/invite', '/inv']: + if len(parts) < 3: + self.write_line("[ERROR] Usage: /invite ") + return + + channel_name = parts[1].lower() + invitee = parts[2] + + success, msg = self.chat_server.invite_to_channel(channel_name, self.username, invitee) + + if success: + self.write_line(f"[SUCCESS] {msg}") + else: + self.write_line(f"[ERROR] {msg}") + + elif command in ['/switch', '/s']: + if len(parts) < 2: + self.write_line("[ERROR] Usage: /switch ") + return + + channel_name = parts[1].lower() + + if channel_name in self.chat_server.channels and self.chat_server.channels[channel_name].has_member(self.username): + self.switch_view(channel_name) + else: + self.write_line(f"[ERROR] You are not a member of #{channel_name}") + + elif command in ['/channels', '/ch']: + channels = self.chat_server.get_user_channels(self.username) + self.write_line(f"Your channels: {', '.join(['#' + c for c in channels])}") + + elif command in ['/users', '/u']: + users = list(self.chat_server.users.keys()) + self.write_line(f"Online users ({len(users)}): {', '.join(sorted(users))}") + + elif command in ['/quit', '/q', '/exit']: + self.write_line("Goodbye!") + self.running = False + self.process.exit(0) + + else: + self.write_line(f"[ERROR] Unknown command: {command}. Type /help for available commands") + + def switch_view(self, view: str): + """Switch to a different channel or DM.""" + self.current_view = view + self.write_line("") + self.write_line("=" * 60) + if view.startswith('dm:'): + self.write_line(f"Switched to DM with {view[3:]}") + else: + self.write_line(f"Switched to channel #{view}") + self.write_line("=" * 60) + self.write_line("") + + def show_help(self): + """Show help message.""" + self.write_line("") + self.write_line("Available commands:") + self.write_line(" /dm [msg] or /d - DM a user") + self.write_line(" /create [private] /new - Create a channel") + self.write_line(" /join or /j - Join a channel") + self.write_line(" /leave or /l - Leave a channel") + self.write_line(" /invite or /inv - Invite user to channel") + self.write_line(" /switch or /s - Switch to a channel") + self.write_line(" /channels or /ch - List your channels") + self.write_line(" /users or /u - List online users") + self.write_line(" /quit or /q - Exit") + self.write_line(" /help or /h - Show this help") + self.write_line("") diff --git a/ssh_host_key b/ssh_host_key new file mode 100644 index 0000000..427dbf6 --- /dev/null +++ b/ssh_host_key @@ -0,0 +1,27 @@ +-----BEGIN OPENSSH PRIVATE KEY----- +b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABFwAAAAdzc2gtcn +NhAAAAAwEAAQAAAQEAvRktsoBMdrAC7HIwRwcteBu1gA8uIhFr4NZrFL8hdQ//y0C/0jdR +OJo7AO11Vb2p8NBzU8JI3fX1wBwh1sfVwVuRxieR/nbpektm9JI19DxZCAVfWNjzEA9nOl +8pumIIPUN3YuZy0OZ1/FCnnYJccHp00vzjQUXzqIhrYTP/gqPWUIBOssSrZhUtt3K9oKmq +JIqwlCPnpH0FV76PYEhk5t0Zwx3lsnZVKvD3HS88uJiZ5kbf9VfOt2d481WJ2iMgpk9Ehj +eEyFmFIlt/nh1w7ZLGjKiVmedezM79TG2i5NE4Ay1NHZpnmE6scMOAYa0NSV4pRuUSwLsa +3WElHrXqEQAAA7iNUpt4jVKbeAAAAAdzc2gtcnNhAAABAQC9GS2ygEx2sALscjBHBy14G7 +WADy4iEWvg1msUvyF1D//LQL/SN1E4mjsA7XVVvanw0HNTwkjd9fXAHCHWx9XBW5HGJ5H+ +dul6S2b0kjX0PFkIBV9Y2PMQD2c6Xym6Ygg9Q3di5nLQ5nX8UKedglxwenTS/ONBRfOoiG +thM/+Co9ZQgE6yxKtmFS23cr2gqaokirCUI+ekfQVXvo9gSGTm3RnDHeWydlUq8PcdLzy4 +mJnmRt/1V863Z3jzVYnaIyCmT0SGN4TIWYUiW3+eHXDtksaMqJWZ517Mzv1MbaLk0TgDLU +0dmmeYTqxww4BhrQ1JXilG5RLAuxrdYSUeteoRAAAAAwEAAQAAAQARa1j6bydsDQd96sBw +AJUjkfDgqrVSGdb9U58yXm95ckd63Jx3A9XBcDK0gYtcIkA+AOPI0NZHzR0t7OGAoLC8Es +B9V56TKXbVR04lBDWAadE9RApuG7EbVwHoPoUwaFCwPQ9pvscPn5U5kJ/6Klyz11H88CMK +m6QoD6YNop28Vcz+YULxOBZp0Ae+fa3dpfYFtvH/f1uNoc6Qexi6994Ba4EfjeWacLAgSL +Bg4DU0jSB96pCxJUvV2XZ940KdTkw2ECQXPTD5yDJj7ogj4aUZ81qLeAg6xkkVHHbh194G +ZnuWzMfNU0iuS6lrNCZWUx6VnZvNpbmUsET7yE8glOdpAAAAgGxxsMySXUKIkxOfzAmSUd +he3YI+2nns2DVm31/WJVZAq1txnVaqzjUhbNMNG0TksPLGluqGkGRQBpGQG3TOlhzkypcl +q+CLjUo/WX8FoB/GW0IUbz0bW1ucoBMu65QB1cfnD2O0sQi/VwCc2WMQEt0Zs/ivryeYF+ +kw8sUR6aURAAAAgQDqGmOfzmuIKT1bby8fadauXKtqcub/PXSwN5skArm+ZKKjj1Slnnl5 +kN9qcOFOF5M3Fe698R2M24waI5t8lajPH6kagyq2vvG+mKrEelVPd3UczlHinqkL/RPGUb +Lt4HJJuj1qTDlAdp/9InOpBeJ8Kpsqu1+Ow1u97GtzEZzIeQAAAIEAzskmET6HvrTltNea +v79TWvMhly1uEs1JK8OMDF8uK9InnHbhJMw+Q3z8r1zBcCBBxe6aRLRzH0mv8cqBcoKepc +oJJ9wUQOqYvWkBWASjvkYRg7F0gZlmdGD628wRmAGpc5TXSYWazR1PkBjErIaolsgh1GO8 +Ca774snGwKM3+FkAAAAAAQID +-----END OPENSSH PRIVATE KEY----- diff --git a/tests/test_chat_server.py b/tests/test_chat_server.py new file mode 100644 index 0000000..7cb5651 --- /dev/null +++ b/tests/test_chat_server.py @@ -0,0 +1,22 @@ +import pytest +from chatapp.chat_server import ChatServer +from chatapp.models import UserConnection +import asyncio + +@pytest.fixture +def chat_server(): + """Returns a ChatServer instance.""" + return ChatServer() + +@pytest.mark.asyncio +async def test_add_user(chat_server: ChatServer): + """Test adding a user to the chat server.""" + username = "testuser" + message_queue = asyncio.Queue() + user_conn = UserConnection(username, message_queue) + + chat_server.add_user(username, user_conn) + + assert username in chat_server.users + assert chat_server.users[username] == user_conn + assert username in chat_server.channels['global'].members