yumoqing/hermes-web-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(hermes-web-cli): refactor user context, settings, services and sessions management

Browse Source 
- Remove deprecated UNKNOWN.egg-info and user_context.py
- Refactor crud_ops, db_tables, and init modules
- Update settings UI and save handlers (appearance, general, security)
- Update services list, remove, and test DSPY files
- Update sessions list DSPY file
- Add multi-user test script
- Update pyproject.toml dependencies
This commit is contained in:
yumoqing 2026-04-25 21:43:43 +08:00
parent 58dfbb0108
commit d423a03a6d
18 changed files with 832 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
UNKNOWN.egg-info/PKG-INFO
View File

@ -1,10 +0,0 @@
Metadata-Version: 2.1
Name: UNKNOWN
Version: 0.0.0
Summary: UNKNOWN
Home-page: UNKNOWN
License: UNKNOWN
Platform: UNKNOWN
UNKNOWN

5
UNKNOWN.egg-info/SOURCES.txt
View File

@ -1,5 +0,0 @@
pyproject.toml
UNKNOWN.egg-info/PKG-INFO
UNKNOWN.egg-info/SOURCES.txt
UNKNOWN.egg-info/dependency_links.txt
UNKNOWN.egg-info/top_level.txt

1
UNKNOWN.egg-info/dependency_links.txt
View File

@ -1 +0,0 @@

1
UNKNOWN.egg-info/top_level.txt
View File

@ -1 +0,0 @@

17
hermes_web_cli/crud_ops.py
View File

@ -14,10 +14,10 @@ SERVICES_CRUD = {
"create": {
"name": "create_service_record",
"description": "Create a new service record for the current user",
"parameters": ["user_id", "name", "service_url", "description"],
"parameters": ["user_id", "name", "service_url", "description", "apikey"],
"sql_template": """
INSERT INTO services (id, user_id, name, service_url, description, status, created_at, updated_at)
VALUES (${id}$, ${user_id}$, ${name}$, ${service_url}$, ${description}$, ${status}$, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
INSERT INTO services (id, user_id, name, service_url, description, apikey, status, created_at, updated_at)
VALUES (${id}$, ${user_id}$, ${name}$, ${service_url}$, ${description}$, ${apikey}$, ${status}$, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
""",
"return_fields": ["id"]
},
@ -26,35 +26,36 @@ SERVICES_CRUD = {
"description": "Get all services for the current user",
"parameters": ["user_id"],
"sql_template": """
SELECT id, user_id, name, service_url, description, status,
SELECT id, user_id, name, service_url, description, apikey, status,
created_at, updated_at
FROM services
WHERE user_id = ${user_id}$
ORDER BY created_at DESC
""",
"return_fields": ["id", "user_id", "name", "service_url", "description", "status", "created_at", "updated_at"]
"return_fields": ["id", "user_id", "name", "service_url", "description", "apikey", "status", "created_at", "updated_at"]
},
"read_by_id": {
"name": "get_service_by_id_and_user",
"description": "Get a specific service by ID for the current user",
"parameters": ["service_id", "user_id"],
"sql_template": """
SELECT id, user_id, name, service_url, description, status,
SELECT id, user_id, name, service_url, description, apikey, status,
created_at, updated_at
FROM services
WHERE id = ${service_id}$ AND user_id = ${user_id}$
""",
"return_fields": ["id", "user_id", "name", "service_url", "description", "status", "created_at", "updated_at"]
"return_fields": ["id", "user_id", "name", "service_url", "description", "apikey", "status", "created_at", "updated_at"]
},
"update": {
"name": "update_service_record",
"description": "Update an existing service record",
"parameters": ["service_id", "user_id", "name", "service_url", "description", "status"],
"parameters": ["service_id", "user_id", "name", "service_url", "description", "apikey", "status"],
"sql_template": """
UPDATE services
SET name = ${name}$,
service_url = ${service_url}$,
description = ${description}$,
apikey = ${apikey}$,
status = ${status}$,
updated_at = CURRENT_TIMESTAMP
WHERE id = ${service_id}$ AND user_id = ${user_id}$

7
hermes_web_cli/db_tables.py
View File

@ -42,6 +42,13 @@ SERVICES_TABLE = {
"nullable": True,
"description": "Optional service description"
},
{
"name": "apikey",
"type": "varchar(512)",
"nullable": True,
"default": "''",
"description": "API key for service authentication (X-API-Key header)"
},
{
"name": "status",
"type": "varchar(20)",

702
hermes_web_cli/init.py
View File

@ -18,14 +18,13 @@ from datetime import datetime
# Import sqlor database module
from sqlor.dbpools import get_sor_context
# Import user context helper
from appPublic.uniqueID import getID
>>>>>>> f741c58 (feat(hermes-web-cli): refactor user context, settings, services and sessions management)
# Import database table definitions and CRUD operations
from .db_tables import TABLE_DEFINITIONS
from .crud_ops import SERVICES_CRUD, SESSIONS_CRUD, SETTINGS_CRUD
def load_hermes_web_cli():
<<<<<<< HEAD
"""Initialize and load the hermes-web-cli module.
This function is called by Sage system during module loading.
@ -465,8 +464,523 @@ async def get_session_by_id(userid, session_id: str) -> Optional[Dict]:
except Exception as e:
print(f"Error getting session by ID: {str(e)}")
return None
=======
"""Initialize and load the hermes-web-cli module.
# Utility functions for validation
This function is called by Sage system during module loading.
It registers all module functions with the ServerEnv instance
so they can be called directly from .ui and .dspy files.
"""
from ahserver.serverenv import ServerEnv
# Initialize database tables if needed
try:
from .init_db import init_database
import asyncio
# Run database initialization in a new event loop if needed
try:
asyncio.get_running_loop()
# If we're already in an async context, create a task
asyncio.create_task(init_database())
except RuntimeError:
# No running loop, run synchronously
asyncio.run(init_database())
except Exception as e:
print(f"Warning: Database initialization failed: {str(e)}")
# Continue loading even if DB init fails - functions will handle errors gracefully
# Get the ServerEnv instance
env = ServerEnv()
# Register all module functions with ServerEnv
env.get_setting = get_setting
env.save_setting = save_setting
env.get_all_services = get_all_services
env.create_service = create_service
env.delete_service = delete_service
env.get_service_by_id = get_service_by_id
env.test_service_connection = test_service_connection
env.create_session = create_session
env.send_message_to_service = send_message_to_service
env.get_session_messages = get_session_messages
env.get_active_sessions = get_active_sessions
env.get_recent_sessions = get_recent_sessions
env.get_session_by_id = get_session_by_id
env.validate_service_url = validate_service_url
env.generate_session_id = generate_session_id
return True
# Database operations using sqlor-database-module
async def get_all_services(user_id: str) -> List[Dict]:
"""Get all registered Hermes services for the specified user from database.
Args:
user_id: The ID of the user whose services to retrieve
Returns:
List of service dictionaries belonging to the specified user
"""
try:
# Query services table with user_id filter using sqlor-database-module
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['read_all']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id})
# Convert datetime objects to ISO format strings for JSON serialization
result = []
for rec in recs:
service_dict = dict(rec)
if 'created_at' in service_dict and service_dict['created_at']:
service_dict['created_at'] = service_dict['created_at'].isoformat()
if 'updated_at' in service_dict and service_dict['updated_at']:
service_dict['updated_at'] = service_dict['updated_at'].isoformat()
result.append(service_dict)
return result
except Exception as e:
print(f"Error getting services: {str(e)}")
return []
async def create_service(name: str, url: str, user_id: str, description: str = "", apikey: str = "") -> str:
"""Create a new Hermes service registration for the specified user.
Args:
name: Service name
url: Service URL
user_id: The ID of the user creating the service
description: Service description (optional)
apikey: API key for the service (optional)
Returns:
The created service ID
"""
try:
# Validate service URL
if not await validate_service_url(url):
raise ValueError("Invalid service URL")
service_id = str(uuid.uuid4())
# Save to database using sqlor-database-module
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['create']['sql_template']
await sor.sqlExe(sql_template, {
'id': service_id,
'user_id': user_id,
'name': name,
'service_url': url,
'description': description,
'apikey': apikey,
'status': 'active'
})
return service_id
except Exception as e:
print(f"Error creating service: {str(e)}")
raise
async def delete_service(service_id: str, user_id: str) -> bool:
"""Delete a Hermes service registration (only if owned by specified user).
Args:
service_id: The ID of the service to delete
user_id: The ID of the user attempting deletion
Returns:
True if deleted successfully, False otherwise
"""
try:
# Verify service belongs to current user before deletion
service = await get_service_by_id(service_id, user_id)
if not service:
return False
if service.get("user_id") != user_id:
print(f"Permission denied: Service {service_id} does not belong to user {user_id}")
return False
# Delete from database using sqlor-database-module
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['delete']['sql_template']
await sor.sqlExe(sql_template, {
'service_id': service_id,
'user_id': user_id
})
# Also delete associated sessions
async with db.sqlorContext('hermes-web-cli') as sor:
await sor.sqlExe("""
DELETE FROM sessions
WHERE service_id = ${service_id}$ AND user_id = ${user_id}$
""", {
'service_id': service_id,
'user_id': user_id
})
return True
except Exception as e:
print(f"Error deleting service: {str(e)}")
return False
async def get_service_by_id(service_id: str, user_id: str) -> Optional[Dict]:
"""Get service configuration by ID (only if owned by specified user).
Args:
service_id: The ID of the service to retrieve
user_id: The ID of the user requesting the service
Returns:
Service dictionary if found and owned by user, None otherwise
"""
try:
# Query database directly with user_id filter for security
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['read_by_id']['sql_template']
recs = await sor.sqlExe(sql_template, {
'service_id': service_id,
'user_id': user_id
})
if len(recs) > 0:
service_dict = dict(recs[0])
if 'created_at' in service_dict and service_dict['created_at']:
service_dict['created_at'] = service_dict['created_at'].isoformat()
if 'updated_at' in service_dict and service_dict['updated_at']:
service_dict['updated_at'] = service_dict['updated_at'].isoformat()
return service_dict
return None
except Exception as e:
print(f"Error getting service: {str(e)}")
return None
# Service connection testing
async def test_service_connection(service_id: str) -> Tuple[bool, str]:
"""Test connection to a Hermes service endpoint.
Args:
service_id: The ID of the service to test
Returns:
Tuple[bool, str]: (is_connected, status_message)
"""
try:
# Get service configuration from database
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SERVICES_CRUD['operations']['read_by_id']['sql_template']
recs = await sor.sqlExe(sql_template, {
'service_id': service_id,
'user_id': ''
})
if not recs:
return False, "Service not found"
service = dict(recs[0])
url = service["service_url"]
apikey = service.get("apikey", "")
# Prepare headers
headers = {}
if apikey:
headers["Authorization"] = f"Bearer {apikey}"
# Test the /health endpoint or similar
timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(f"{url.rstrip('/')}/health", headers=headers) as response:
if response.status == 200:
return True, "Connected"
else:
return False, f"HTTP {response.status}"
except asyncio.TimeoutError:
return False, "Connection timeout"
except aiohttp.ClientConnectorError:
return False, "Connection refused"
except Exception as e:
return False, f"Error: {str(e)}"
# Session management
async def create_session(service_id: str, user_id: str, user_message: str = "") -> str:
"""Create a new session with a Hermes service."""
try:
# Get service configuration (verify it belongs to current user)
service = await get_service_by_id(service_id, user_id)
if not service:
raise ValueError(f"Service {service_id} not found or access denied")
service_url = service["service_url"]
apikey = service.get("apikey", "")
# Prepare headers
headers = {
"Content-Type": "application/json"
}
# Add Authorization header if API key is provided
if apikey:
headers["Authorization"] = f"Bearer {apikey}"
# Call remote service API to create session
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(
f"{service_url.rstrip('/')}/api/v1/sessions",
json={
"user_id": user_id,
"initial_message": user_message if user_message else None
},
headers=headers
) as response:
response.raise_for_status()
result = await response.json()
# Get the session ID from the remote service
remote_session_id = result.get("session_id", "")
if not remote_session_id:
raise ValueError("Remote service did not return a session ID")
# Create local session record in database
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['create']['sql_template']
await sor.sqlExe(sql_template, {
'session_id': remote_session_id,
'user_id': user_id,
'service_id': service_id,
'session_name': None,
'service_name': service.get("name", "Unknown Service")
})
# Return the session ID from the remote service
return remote_session_id
except Exception as e:
print(f"Error creating session: {str(e)}")
raise
async def send_message_to_service(service_id: str, session_id: str, message: str, user_id: str) -> Dict:
"""Send a message to a Hermes service and get response (only if session owned by specified user).
Args:
service_id: The service ID
session_id: The session ID
message: The message to send
user_id: The ID of the user sending the message
Returns:
Response from the service
"""
try:
# Verify session belongs to current user before sending message
session = await get_session_by_id(session_id, user_id)
if not session:
raise ValueError(f"Session {session_id} not found or access denied for user {user_id}")
service = await get_service_by_id(session['service_id'], user_id)
if not service:
raise ValueError(f"Service for session {session_id} not found or access denied for user {user_id}")
service_url = service["service_url"]
apikey = service.get("apikey", "")
# Prepare headers
headers = {
"Content-Type": "application/json"
}
# Add Authorization header if API key is provided
if apikey:
headers["Authorization"] = f"Bearer {apikey}"
# Call remote service API
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(
f"{service_url.rstrip('/')}/api/v1/sessions/{session_id}/messages",
json={
"message": message
},
headers=headers
) as response:
response.raise_for_status()
return await response.json()
except Exception as e:
print(f"Error sending message: {e}")
raise
async def get_session_messages(session_id: str, user_id: str) -> List[Dict]:
"""Get all messages for a session (only if session owned by specified user).
Args:
session_id: The session ID
user_id: The ID of the user requesting messages
Returns:
List of message dictionaries
"""
try:
# Verify session belongs to current user before getting messages
session = await get_session_by_id(session_id)
if not session:
print(f"Session {session_id} not found or access denied for user {user_id}")
return []
# Get the associated service
service = await get_service_by_id(session['service_id'], user_id)
if not service:
print(f"Service for session {session_id} not found or access denied")
return []
service_url = service["service_url"]
apikey = service.get("apikey", "")
# Prepare headers
headers = {
"Content-Type": "application/json"
}
# Add Authorization header if API key is provided
if apikey:
headers["Authorization"] = f"Bearer {apikey}"
# Call remote service API to get messages
timeout = aiohttp.ClientTimeout(total=30)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.get(
f"{service_url.rstrip('/')}/api/v1/sessions/{session_id}/messages",
headers=headers
) as response:
response.raise_for_status()
messages = await response.json()
# Update session last_active timestamp and message count in local database
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
await sor.sqlExe("""
UPDATE sessions
SET last_active = CURRENT_TIMESTAMP,
message_count = ${message_count}$
WHERE session_id = ${session_id}$ AND user_id = ${user_id}$
""", {
'session_id': session_id,
'user_id': user_id,
'message_count': len(messages)
})
return messages
except Exception as e:
print(f"Error getting session messages: {str(e)}")
return []
async def get_active_sessions(user_id: str) -> List[Dict]:
"""Get all active sessions for the specified user from database.
Args:
user_id: The ID of the user whose active sessions to retrieve
Returns:
List of active session dictionaries
"""
try:
# Query the sessions table for active sessions belonging to current user using sqlor-database-module
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['read_active']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id})
# Convert datetime objects to ISO format strings for JSON serialization
result = []
for rec in recs:
session_dict = dict(rec)
if 'created_at' in session_dict and session_dict['created_at']:
session_dict['created_at'] = session_dict['created_at'].isoformat()
if 'last_active' in session_dict and session_dict['last_active']:
session_dict['last_active'] = session_dict['last_active'].isoformat()
result.append(session_dict)
return result
except Exception as e:
print(f"Error getting active sessions: {str(e)}")
return []
async def get_recent_sessions(user_id: str, limit: int = 5) -> List[Dict]:
"""Get recent sessions for the specified user from database, ordered by creation time (most recent first).
Args:
user_id: The ID of the user whose recent sessions to retrieve
limit: Maximum number of sessions to return (default: 5)
Returns:
List of recent session dictionaries
"""
try:
# Query the sessions table for recent sessions belonging to current user using sqlor-database-module
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['read_recent']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id, 'limit': limit})
# Convert datetime objects to ISO format strings for JSON serialization
result = []
for rec in recs:
session_dict = dict(rec)
if 'created_at' in session_dict and session_dict['created_at']:
session_dict['created_at'] = session_dict['created_at'].isoformat()
if 'last_active' in session_dict and session_dict['last_active']:
session_dict['last_active'] = session_dict['last_active'].isoformat()
result.append(session_dict)
return result
except Exception as e:
print(f"Error getting recent sessions: {str(e)}")
return []
async def get_session_by_id(session_id: str, user_id: str) -> Optional[Dict]:
"""Get session details by session ID (only if owned by specified user).
Args:
session_id: The session ID to retrieve
user_id: The ID of the user requesting the session
Returns:
Session dictionary if found and owned by user, None otherwise
"""
try:
# Query database directly with user_id filter for security
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SESSIONS_CRUD['operations']['read_by_id']['sql_template']
recs = await sor.sqlExe(sql_template, {
'session_id': session_id,
'user_id': user_id
})
if len(recs) > 0:
session_dict = dict(recs[0])
if 'created_at' in session_dict and session_dict['created_at']:
session_dict['created_at'] = session_dict['created_at'].isoformat()
if 'last_active' in session_dict and session_dict['last_active']:
session_dict['last_active'] = session_dict['last_active'].isoformat()
return session_dict
return None
except Exception as e:
print(f"Error getting session by ID: {str(e)}")
return None
# Database operations using sqlor-database-module
def validate_service_url(url: str) -> bool:
"""Validate if a URL is a valid Hermes service endpoint."""
if not url.startswith(('http://', 'https://')):
@ -480,107 +994,115 @@ def generate_session_id() -> str:
return getID()
# Settings management
async def get_setting() -> Dict:
"""Get current user settings from database or return defaults."""
import json
async def get_setting(user_id: str) -> Dict:
"""Get user settings from database or return defaults.
# Get current user ID
user_id = await get_current_user_id()
Args:
user_id: The ID of the user whose settings to retrieve
default_settings = {
"security": {
"require_auth": False,
"encrypt_storage": False
},
"general": {
"default_model": "",
"session_timeout": 30,
"auto_save": True
},
"appearance": {
"theme": "dark"
}
}
Returns:
User settings dictionary
"""
import json
try:
# Query user settings from database
env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SETTINGS_CRUD['operations']['read']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id})
default_settings = {
"security": {
"require_auth": False,
"encrypt_storage": False
},
"general": {
"default_model": "",
"session_timeout": 30,
"auto_save": True
},
"appearance": {
"theme": "dark"
}
}
if len(recs) > 0:
settings_json = recs[0]['settings_json']
if settings_json:
saved_settings = json.loads(settings_json)
# Merge with defaults to ensure all keys exist
for section, defaults in default_settings.items():
if section not in saved_settings:
saved_settings[section] = defaults
else:
for key, value in defaults.items():
if key not in saved_settings[section]:
saved_settings[section][key] = value
return saved_settings
try:
# Query user settings from database
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SETTINGS_CRUD['operations']['read']['sql_template']
recs = await sor.sqlExe(sql_template, {'user_id': user_id})
except Exception as e:
print(f"Error getting settings: {str(e)}")
# Fall back to defaults on error
if len(recs) > 0:
settings_json = recs[0]['settings_json']
if settings_json:
saved_settings = json.loads(settings_json)
# Merge with defaults to ensure all keys exist
for section, defaults in default_settings.items():
if section not in saved_settings:
saved_settings[section] = defaults
else:
for key, value in defaults.items():
if key not in saved_settings[section]:
saved_settings[section][key] = value
return saved_settings
return default_settings
except Exception as e:
print(f"Error getting settings: {str(e)}")
# Fall back to defaults on error
async def save_setting(section: str, key: str, value) -> bool:
"""Save a specific setting value to current user's database record."""
import json
return default_settings
# Get current user ID
user_id = await get_current_user_id()
async def save_setting(section: str, key: str, value, user_id: str) -> bool:
"""Save a specific setting value to user\'s database record.
# Load existing settings or start with defaults
settings = await get_setting()
Args:
section: Settings section name
key: Setting key name
value: Setting value
user_id: The ID of the user whose settings to update
# Update the specific setting
if section not in settings:
settings[section] = {}
settings[section][key] = value
Returns:
True if saved successfully, False otherwise
"""
import json
try:
# Save to database using sqlor-database-module
env = ServerEnv()
async with get_sor_context(env, 'hermes-web-cli') as sor:
sql_template = SETTINGS_CRUD['operations']['create_or_update']['sql_template']
await sor.sqlExe(sql_template, {
'user_id': user_id,
'settings_json': json.dumps(settings)
})
return True
except Exception as e:
print(f"Error saving settings: {str(e)}")
return False
# Load existing settings or start with defaults
settings = await get_setting(user_id)
# Update the specific setting
if section not in settings:
settings[section] = {}
settings[section][key] = value
try:
# Save to database using sqlor-database-module
db = DBPools()
async with db.sqlorContext('hermes-web-cli') as sor:
sql_template = SETTINGS_CRUD['operations']['create_or_update']['sql_template']
await sor.sqlExe(sql_template, {
'user_id': user_id,
'settings_json': json.dumps(settings)
})
return True
except Exception as e:
print(f"Error saving settings: {str(e)}")
return False
# Module metadata
MODULE_NAME = "hermes-web-cli"
MODULE_VERSION = "0.1.0"
MODULE_VERSION = "0.2.0"
# Export all public functions
__all__ = [
'load_hermes_web_cli',
'get_all_services',
'create_service',
'delete_service',
'get_service_by_id',
'test_service_connection',
'create_session',
'send_message_to_service',
'get_session_messages',
'get_active_sessions',
'get_recent_sessions',
'get_session_by_id',
'validate_service_url',
'generate_session_id',
'get_setting',
'save_setting',
'get_current_user_id',
'MODULE_NAME',
'MODULE_VERSION'
'load_hermes_web_cli',
'get_all_services',
'create_service',
'delete_service',
'get_service_by_id',
'test_service_connection',
'create_session',
'send_message_to_service',
'get_session_messages',
'get_active_sessions',
'get_recent_sessions',
'get_session_by_id',
'validate_service_url',
'generate_session_id',
'get_setting',
'save_setting',
'MODULE_NAME',
'MODULE_VERSION'
]

2
pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hermes-web-cli"
version = "0.1.0"
version = "0.2.0"
description = "Hermes Web CLI module for multi-service management"
authors = [{name = "Your Name", email = "your.email@example.com"}]
license = {text = "MIT"}

153
test_multiuser.py Normal file
View File

@ -0,0 +1,153 @@
#!/usr/bin/env python3
"""
Test script to verify multi-user data isolation in hermes-web-cli module.
This script simulates multiple users and verifies that they can only access their own data.
"""
import asyncio
import sys
import os
# Add the module path to Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from hermes_web_cli.init import (
get_all_services, create_service, delete_service, get_service_by_id,
create_session, get_active_sessions, get_recent_sessions, get_session_by_id,
get_setting, save_setting
)
from hermes_web_cli.user_context import get_current_user_id
# Mock ahserver's get_user function for testing
async def mock_get_user_1():
return "user1"
async def mock_get_user_2():
return "user2"
async def test_multiuser_isolation():
"""Test that users can only access their own data."""
print("Testing multi-user data isolation...")
# Test user 1
print("\n--- Testing User 1 ---")
# Mock user context for user 1
import hermes_web_cli.user_context
hermes_web_cli.user_context.get_user = mock_get_user_1
# Create service for user 1
service1_id = await create_service(
name="User1 Service",
url="http://localhost:8080",
description="Service for user 1",
apikey="user1-key"
)
print(f"Created service for user 1: {service1_id}")
# Get services for user 1
services1 = await get_all_services()
print(f"User 1 sees {len(services1)} services")
# Verify user 1 can access their service
service1 = await get_service_by_id(service1_id)
assert service1 is not None, "User 1 should be able to access their own service"
print("User 1 can access their own service ✓")
# Test user 2
print("\n--- Testing User 2 ---")
# Mock user context for user 2
hermes_web_cli.user_context.get_user = mock_get_user_2
# Create service for user 2
service2_id = await create_service(
name="User2 Service",
url="http://localhost:8081",
description="Service for user 2",
apikey="user2-key"
)
print(f"Created service for user 2: {service2_id}")
# Get services for user 2
services2 = await get_all_services()
print(f"User 2 sees {len(services2)} services")
# Verify user 2 can access their service
service2 = await get_service_by_id(service2_id)
assert service2 is not None, "User 2 should be able to access their own service"
print("User 2 can access their own service ✓")
# Verify user 2 cannot access user 1's service
service1_from_user2 = await get_service_by_id(service1_id)
assert service1_from_user2 is None, "User 2 should NOT be able to access user 1's service"
print("User 2 cannot access user 1's service ✓")
# Verify user 1 still only sees their own service
hermes_web_cli.user_context.get_user = mock_get_user_1
services1_after = await get_all_services()
assert len(services1_after) == 1, "User 1 should still only see 1 service"
print("User 1 still only sees their own service ✓")
# Test sessions
print("\n--- Testing Session Isolation ---")
# Create session for user 1
hermes_web_cli.user_context.get_user = mock_get_user_1
session1_id = await create_session(service1_id, "user1", "Hello from user 1")
print(f"Created session for user 1: {session1_id}")
# Create session for user 2
hermes_web_cli.user_context.get_user = mock_get_user_2
session2_id = await create_session(service2_id, "user2", "Hello from user 2")
print(f"Created session for user 2: {session2_id}")
# Verify user 1 only sees their session
hermes_web_cli.user_context.get_user = mock_get_user_1
sessions1 = await get_active_sessions()
assert len(sessions1) == 1, "User 1 should only see 1 session"
assert sessions1[0]['session_id'] == session1_id, "User 1 should see their own session"
print("User 1 session isolation ✓")
# Verify user 2 only sees their session
hermes_web_cli.user_context.get_user = mock_get_user_2
sessions2 = await get_active_sessions()
assert len(sessions2) == 1, "User 2 should only see 1 session"
assert sessions2[0]['session_id'] == session2_id, "User 2 should see their own session"
print("User 2 session isolation ✓")
# Test settings
print("\n--- Testing Settings Isolation ---")
# Save setting for user 1
hermes_web_cli.user_context.get_user = mock_get_user_1
await save_setting("appearance", "theme", "dark")
# Save setting for user 2
hermes_web_cli.user_context.get_user = mock_get_user_2
await save_setting("appearance", "theme", "light")
# Verify user 1 gets their setting
hermes_web_cli.user_context.get_user = mock_get_user_1
settings1 = await get_setting()
assert settings1.get('appearance', {}).get('theme') == 'dark', "User 1 should have dark theme"
print("User 1 settings isolation ✓")
# Verify user 2 gets their setting
hermes_web_cli.user_context.get_user = mock_get_user_2
settings2 = await get_setting()
assert settings2.get('appearance', {}).get('theme') == 'light', "User 2 should have light theme"
print("User 2 settings isolation ✓")
# Clean up
print("\n--- Cleaning Up ---")
hermes_web_cli.user_context.get_user = mock_get_user_1
await delete_service(service1_id)
hermes_web_cli.user_context.get_user = mock_get_user_2
await delete_service(service2_id)
print("\n✅ All multi-user isolation tests passed!")
if __name__ == "__main__":
asyncio.run(test_multiuser_isolation())

7
wwwroot/hermes_services/list/index.dspy
View File

@ -2,10 +2,11 @@
# This .dspy file uses functions released by load_hermes_web_cli()
try:
# Use the function provided by hermes-web-cli module
userid = await get_user()
services = await get_all_services(userid)
# Get current user ID using ahserver's built-in get_user() function
user_id = await get_user()
# Use the function provided by hermes-web-cli module
services = await get_all_services(user_id)
# Format for code component (value, text pairs)
result = []
for service in services:

5
wwwroot/services/list/index.dspy
View File

@ -2,8 +2,11 @@
# This .dspy file uses functions provided by load_hermes_web_cli()
try:
# Get current user ID using ahserver's built-in get_user() function
user_id = await get_user()
# Use the function provided by the hermes-web-cli module
services = await get_all_services()
services = await get_all_services(user_id)
# Format services for UI display
result = []

8
wwwroot/services/remove/index.dspy
View File

@ -5,10 +5,12 @@ try:
service_id = params_kw.get('id')
if not service_id:
return {"error": "Service ID is required"}
userid = await get_user()
# Call the function provided by the hermes-web-cli module
success = await delete_service(userid, service_id)
# Get current user ID using ahserver's built-in get_user() function
user_id = await get_user()
# Call the function provided by the hermes-web-cli module
success = await delete_service(service_id, user_id)
if success:
return {"success": True, "message": "Service removed successfully"}
else:

2
wwwroot/services/test/index.dspy
View File

@ -7,7 +7,7 @@ try:
return {"error": "Service ID is required"}
# Call the function provided by the hermes-web-cli module
is_connected, status_msg = test_service_connection(service_id)
is_connected, status_msg = await test_service_connection(service_id)
return {"status": status_msg}

5
wwwroot/sessions/list/index.dspy
View File

@ -2,8 +2,11 @@
# This .dspy file uses functions provided by load_hermes_web_cli()
try:
# Get current user ID using ahserver's built-in get_user() function
user_id = await get_user()
# Use the function provided by the hermes-web-cli module
sessions = await get_active_sessions()
sessions = await get_active_sessions(user_id)
# Format sessions for UI display
result = []

3
wwwroot/settings.ui
View File

@ -16,7 +16,8 @@
"marginBottom": "20px"
}
},
{% set settings_data = get_setting() %}
{% set current_user = get_user() %}
{% set settings_data = get_setting(current_user) %}
{
"widgettype": "TabPanel",
"id": "settings-tabs",

6
wwwroot/settings/save/appearance/index.dspy
View File

@ -4,9 +4,11 @@
try:
theme = request.form.get('theme', 'dark')
# Save settings using the module function
await save_setting('appearance', 'theme', theme)
# Get current user ID
user_id = await get_user()
# Save settings using the module function
await save_setting('appearance', 'theme', theme, user_id)
return {"success": True, "message": "Appearance settings saved successfully"}
except Exception as e:

10
wwwroot/settings/save/general/index.dspy
View File

@ -11,11 +11,13 @@ try:
except:
session_timeout = 30
# Save settings using the module function
await save_setting('general', 'default_model', default_model)
await save_setting('general', 'session_timeout', session_timeout)
await save_setting('general', 'auto_save', auto_save)
# Get current user ID
user_id = await get_user()
# Save settings using the module function
await save_setting('general', 'default_model', default_model, user_id)
await save_setting('general', 'session_timeout', session_timeout, user_id)
await save_setting('general', 'auto_save', auto_save, user_id)
return {"success": True, "message": "General settings saved successfully"}
except Exception as e:

8
wwwroot/settings/save/security/index.dspy
View File

@ -5,10 +5,12 @@ try:
require_auth = request.form.get('require-auth', 'false') == 'true'
encrypt_storage = request.form.get('encrypt-storage', 'false') == 'true'
# Save settings using the module function
await save_setting('security', 'require_auth', require_auth)
await save_setting('security', 'encrypt_storage', encrypt_storage)
# Get current user ID
user_id = await get_user()
# Save settings using the module function
await save_setting('security', 'require_auth', require_auth, user_id)
await save_setting('security', 'encrypt_storage', encrypt_storage, user_id)
return {"success": True, "message": "Security settings saved successfully"}
except Exception as e: