feat: kithara-admin CLI tool and .env.example
- kithara-admin: full user/invite management CLI - list-users, create-user, delete-user, change-role, reset-password - create-invite, list-invites, revoke-invite - Uses bcrypt directly (not passlib) for host compatibility - Reads data/kithara.db directly for admin operations - .env.example: documented all config options with defaults - KITHARA_MODE, admin credentials, JWT secret, Lyria config - Fixed docker-compose env vars: KITHARA_ADMIN_USER/PASS (matching auth.py)
This commit is contained in:
rin.kagamine committed
1 parent
c571e28725
commit
52a9e368a7
2 files changed
+241
No files matched your search
@@ -0,0 +1,28 @@
|
||||
# Kithara Studio Configuration
|
||||
# Copy this file to .env and customize
|
||||
|
||||
# Google Lyria API key (required)
|
||||
LYRIA_API_KEY=your-api-key-here
|
||||
|
||||
# Default Lyria model
|
||||
LYRIA_DEFAULT_MODEL=lyria-3-clip-preview
|
||||
|
||||
# Default ID3 tag values (used when track metadata is empty)
|
||||
DEFAULT_ID3_ARTIST=Kagamine Rin
|
||||
DEFAULT_ID3_ALBUM=Kithara Studio
|
||||
DEFAULT_ID3_GENRE=Hardcore Techno
|
||||
|
||||
# Auth mode: single (no login, one user) or multi (full auth with JWT)
|
||||
KITHARA_MODE=single
|
||||
|
||||
# Bootstrap admin credentials (only used in multi mode on first startup)
|
||||
# Creates this admin user automatically if it doesn't exist
|
||||
# KITHARA_ADMIN_USER=admin
|
||||
# KITHARA_ADMIN_PASS=changeme
|
||||
|
||||
# JWT secret (auto-generated if not set, persists in DB)
|
||||
# Set this if you want tokens to survive DB resets
|
||||
# KITHARA_JWT_SECRET=your-secret-here
|
||||
|
||||
# Timezone
|
||||
TZ=America/New_York
|
||||
Executable
+213
@@ -0,0 +1,213 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Kithara Studio admin CLI — manage users, invite codes, and system settings."""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import os
|
||||
import asyncio
|
||||
import uuid
|
||||
import secrets
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'backend', 'src'))
|
||||
|
||||
import aiosqlite
|
||||
import bcrypt
|
||||
|
||||
DEFAULT_DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data', 'kithara.db')
|
||||
db_path = DEFAULT_DB_PATH
|
||||
|
||||
|
||||
async def get_db():
|
||||
db = await aiosqlite.connect(db_path)
|
||||
db.row_factory = aiosqlite.Row
|
||||
return db
|
||||
|
||||
|
||||
async def cmd_list_users(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
async with db.execute("SELECT id, username, display_name, role, api_key IS NOT NULL as has_api_key, created_at FROM users ORDER BY created_at") as cursor:
|
||||
rows = await cursor.fetchall()
|
||||
if not rows:
|
||||
print("No users found.")
|
||||
return
|
||||
print(f"{'ID':<36} {'Username':<20} {'Display Name':<20} {'Role':<10} {'API Key':<8} {'Created'}")
|
||||
print("-" * 120)
|
||||
for row in rows:
|
||||
print(f"{row['id']:<36} {row['username']:<20} {row['display_name'] or '':<20} {row['role']:<10} {'Yes' if row['has_api_key'] else 'No':<8} {row['created_at']}")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def cmd_create_user(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
async with db.execute("SELECT id FROM users WHERE username = ?", (args.username,)) as cursor:
|
||||
if await cursor.fetchone():
|
||||
print(f"Error: Username '{args.username}' already exists.")
|
||||
return
|
||||
password_hash = bcrypt.hashpw(args.password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
||||
user_id = str(uuid.uuid4())
|
||||
await db.execute(
|
||||
"INSERT INTO users (id, username, password_hash, display_name, role) VALUES (?, ?, ?, ?, ?)",
|
||||
(user_id, args.username, password_hash, args.display_name or args.username, args.role)
|
||||
)
|
||||
await db.commit()
|
||||
print(f"Created user '{args.username}' (id: {user_id}, role: {args.role})")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def cmd_delete_user(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
async with db.execute("SELECT id, username FROM users WHERE id = ? OR username = ?", (args.identifier, args.identifier)) as cursor:
|
||||
row = await cursor.fetchone()
|
||||
if not row:
|
||||
print(f"Error: User '{args.identifier}' not found.")
|
||||
return
|
||||
if not args.confirm:
|
||||
print(f"Are you sure you want to delete user '{row['username']}' (id: {row['id']})? Use --confirm to proceed.")
|
||||
return
|
||||
await db.execute("DELETE FROM users WHERE id = ?", (row['id'],))
|
||||
await db.commit()
|
||||
print(f"Deleted user '{row['username']}' (id: {row['id']})")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def cmd_change_role(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
result = await db.execute("UPDATE users SET role = ? WHERE id = ? OR username = ?", (args.role, args.identifier, args.identifier))
|
||||
await db.commit()
|
||||
if result.rowcount == 0:
|
||||
print(f"Error: User '{args.identifier}' not found.")
|
||||
else:
|
||||
print(f"Updated role to '{args.role}' for user '{args.identifier}'")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def cmd_reset_password(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
password_hash = bcrypt.hashpw(args.new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
||||
result = await db.execute(
|
||||
"UPDATE users SET password_hash = ? WHERE id = ? OR username = ?",
|
||||
(password_hash, args.identifier, args.identifier)
|
||||
)
|
||||
await db.commit()
|
||||
if result.rowcount == 0:
|
||||
print(f"Error: User '{args.identifier}' not found.")
|
||||
else:
|
||||
print(f"Reset password for user '{args.identifier}'")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def cmd_create_invite(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
code = args.code or secrets.token_urlsafe(16)
|
||||
await db.execute(
|
||||
"INSERT INTO invite_codes (code, created_by, max_uses, current_uses) VALUES (?, ?, ?, ?)",
|
||||
(code, 'admin', args.max_uses, 0)
|
||||
)
|
||||
await db.commit()
|
||||
print(f"Created invite code: {code}")
|
||||
if args.max_uses > 1:
|
||||
print(f" Max uses: {args.max_uses}")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def cmd_list_invites(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
async with db.execute("SELECT code, created_by, max_uses, current_uses, created_at FROM invite_codes ORDER BY created_at DESC") as cursor:
|
||||
rows = await cursor.fetchall()
|
||||
if not rows:
|
||||
print("No invite codes found.")
|
||||
return
|
||||
print(f"{'Code':<25} {'Created By':<15} {'Uses':<10} {'Max Uses':<10} {'Created'}")
|
||||
print("-" * 100)
|
||||
for row in rows:
|
||||
print(f"{row['code']:<25} {row['created_by']:<15} {row['current_uses']:<10} {row['max_uses']:<10} {row['created_at']}")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
async def cmd_revoke_invite(args):
|
||||
db = await get_db()
|
||||
try:
|
||||
result = await db.execute("DELETE FROM invite_codes WHERE code = ?", (args.code,))
|
||||
await db.commit()
|
||||
if result.rowcount == 0:
|
||||
print(f"Error: Invite code '{args.code}' not found.")
|
||||
else:
|
||||
print(f"Revoked invite code: {args.code}")
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
def main():
|
||||
global db_path
|
||||
|
||||
parser = argparse.ArgumentParser(description='Kithara Studio Admin CLI')
|
||||
parser.add_argument('--db', default=DEFAULT_DB_PATH, help='Path to kithara.db')
|
||||
|
||||
subparsers = parser.add_subparsers(dest='command', help='Available commands')
|
||||
|
||||
subparsers.add_parser('list-users', help='List all users')
|
||||
|
||||
create_user = subparsers.add_parser('create-user', help='Create a new user')
|
||||
create_user.add_argument('username', help='Username')
|
||||
create_user.add_argument('--password', required=True, help='Password')
|
||||
create_user.add_argument('--display-name', help='Display name (defaults to username)')
|
||||
create_user.add_argument('--role', default='user', choices=['user', 'admin'], help='User role')
|
||||
|
||||
delete_user = subparsers.add_parser('delete-user', help='Delete a user')
|
||||
delete_user.add_argument('identifier', help='User ID or username')
|
||||
delete_user.add_argument('--confirm', action='store_true', help='Confirm deletion')
|
||||
|
||||
change_role = subparsers.add_parser('change-role', help='Change user role')
|
||||
change_role.add_argument('identifier', help='User ID or username')
|
||||
change_role.add_argument('role', choices=['user', 'admin'], help='New role')
|
||||
|
||||
reset_pw = subparsers.add_parser('reset-password', help='Reset a user password')
|
||||
reset_pw.add_argument('identifier', help='User ID or username')
|
||||
reset_pw.add_argument('new_password', help='New password')
|
||||
|
||||
create_invite = subparsers.add_parser('create-invite', help='Create an invite code')
|
||||
create_invite.add_argument('--code', help='Custom invite code (auto-generated if not specified)')
|
||||
create_invite.add_argument('--max-uses', type=int, default=1, help='Max number of uses')
|
||||
|
||||
subparsers.add_parser('list-invites', help='List all invite codes')
|
||||
|
||||
revoke_invite = subparsers.add_parser('revoke-invite', help='Revoke an invite code')
|
||||
revoke_invite.add_argument('code', help='Invite code to revoke')
|
||||
|
||||
args = parser.parse_args()
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
db_path = args.db
|
||||
|
||||
cmd_map = {
|
||||
'list-users': cmd_list_users,
|
||||
'create-user': cmd_create_user,
|
||||
'delete-user': cmd_delete_user,
|
||||
'change-role': cmd_change_role,
|
||||
'reset-password': cmd_reset_password,
|
||||
'create-invite': cmd_create_invite,
|
||||
'list-invites': cmd_list_invites,
|
||||
'revoke-invite': cmd_revoke_invite,
|
||||
}
|
||||
|
||||
asyncio.run(cmd_map[args.command](args))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in new issue
Block a user