SirBlob/ChatSSH
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Inital Code

Browse Source
This commit is contained in:
Sir Blob
2025-11-24 15:39:05 +00:00
parent b7c419758a
commit 30b7308b6a
13 changed files with 1558 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
README.md
View File

@@ -1,3 +1,193 @@
# ChatSSH
# Terminal Chat
SSH Chat Server
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 <repository-url>
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 <user> [msg]` | `/d` | Send a direct message or switch to DM view |
| `/create <name> [private]` | `/new` | Create a new channel (optionally private) |
| `/join <name>` | `/j` | Join an existing channel |
| `/leave <name>` | `/l` | Leave a channel |
| `/invite <channel> <user>` | `/inv` | Invite a user to a channel |
| `/delete <name>` | `/del` | Delete a channel you created |
| `/switch <name>` | `/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 <user>` 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]

0
chatapp/__init__.py Normal file
View File

251
chatapp/chat_server.py Normal file
View File

@@ -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)]

23
chatapp/constants.py Normal file
View File

@@ -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',
}

70
chatapp/models.py Normal file
View File

@@ -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

40
chatapp/server.py Normal file
View File

@@ -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()

109
chatapp/ssh_server.py Normal file
View File

@@ -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} <username>@{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

489
chatapp/ui.py Normal file
View File

@@ -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 <user> - DM | /create <channel> - create | /join <channel> - 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 <username> [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 <channel_name> [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 <channel_name>")
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 <channel_name>")
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 <channel_name> <username>")
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 <channel_name>")
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 <channel_name>")
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 <user> [msg][/{COLORS['help']}] or [{COLORS['command']}]/d[/{COLORS['command']}] - DM a user")
chat_history.write(f" [{COLORS['help']}]/create <name> [private][/{COLORS['help']}] or [{COLORS['command']}]/new[/{COLORS['command']}] - Create a channel")
chat_history.write(f" [{COLORS['help']}]/join <name>[/{COLORS['help']}] or [{COLORS['command']}]/j[/{COLORS['command']}] - Join a channel")
chat_history.write(f" [{COLORS['help']}]/leave <name>[/{COLORS['help']}] or [{COLORS['command']}/]/l[/{COLORS['command']}] - Leave a channel")
chat_history.write(f" [{COLORS['help']}]/invite <channel> <user>[/{COLORS['help']}] or [{COLORS['command']}]/inv[/{COLORS['command']}] - Invite user to channel")
chat_history.write(f" [{COLORS['help']}]/delete <name>[/{COLORS['help']}] or [{COLORS['command']}]/del[/{COLORS['command']}] - Delete a channel you created")
chat_history.write(f" [{COLORS['help']}]/switch <name>[/{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)

75
main.py Normal file
View File

@@ -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()

15
requirements.txt Normal file
View File

@@ -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

245
simple_client.py Normal file
View File

@@ -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 <username> [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 <channel_name> [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 <channel_name>")
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 <channel_name>")
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 <channel_name> <username>")
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 <channel_name>")
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 <user> [msg] or /d - DM a user")
self.write_line(" /create <name> [private] /new - Create a channel")
self.write_line(" /join <name> or /j - Join a channel")
self.write_line(" /leave <name> or /l - Leave a channel")
self.write_line(" /invite <ch> <user> or /inv - Invite user to channel")
self.write_line(" /switch <name> 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("")

27
ssh_host_key Normal file
View File

@@ -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-----

22
tests/test_chat_server.py Normal file
View File

@@ -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