diff --git a/bot/bot_app.py b/bot/bot_app.py index ed2772b..54b70c8 100644 --- a/bot/bot_app.py +++ b/bot/bot_app.py @@ -3,7 +3,7 @@ import misskey as misskey from client import client_connection import db_utils as db -from config import NOTIFICATION_POLL_INTERVAL +from config import NOTIFICATION_POLL_INTERVAL, USE_WHITELIST from notification import process_notifications if __name__ == '__main__': @@ -15,6 +15,10 @@ if __name__ == '__main__': # Setup default administrators db.setup_administrators() + # Show whitelist status + whitelist_status = "enabled" if USE_WHITELIST else "disabled" + print(f'Instance whitelisting: {whitelist_status}') + print('Listening for notifications...') while True: if not process_notifications(client): diff --git a/bot/config.py b/bot/config.py index 9737608..57cbfa9 100644 --- a/bot/config.py +++ b/bot/config.py @@ -1,6 +1,7 @@ '''Essentials for the bot to function''' import configparser import json +import re from os import environ, path @@ -13,8 +14,10 @@ def get_config_file() -> str: env: str | None = environ.get('KEMOVERSE_ENV') if not env: raise ConfigError('Error: KEMOVERSE_ENV is unset') - if not (env in ['prod', 'dev']): - raise ConfigError(f'Error: Invalid environment: {env}') + + # Validate environment name contains only alphanumeric, dash, and underscore + if not re.match(r'^[a-zA-Z0-9_-]+$', env): + raise ValueError(f'KEMOVERSE_ENV "{env}" contains invalid characters. Only alphanumeric, dash (-), and underscore (_) are allowed.') config_path: str = f'config_{env}.ini' @@ -23,6 +26,50 @@ def get_config_file() -> str: return config_path +def normalize_user(user_string: str) -> str: + """ + Normalizes a user string to the format @user@domain.tld where domain is lowercase and user is case-sensitive + + Args: + user_string: User string in various formats + + Returns: + Normalized user string + + Raises: + ValueError: If the user string is invalid or domain is malformed + """ + if not user_string or not user_string.strip(): + raise ValueError("User string cannot be empty") + + user_string = user_string.strip() + + # Add leading @ if missing + if not user_string.startswith('@'): + user_string = '@' + user_string + + # Split into user and domain parts + parts = user_string[1:].split('@', 1) # Remove leading @ and split + if len(parts) != 2: + raise ValueError(f"Invalid user format: {user_string}. Expected @user@domain.tld") + + username, domain = parts + + if not username: + raise ValueError("Username cannot be empty") + + if not domain: + raise ValueError("Domain cannot be empty") + + # Validate domain format (basic check for valid domain structure) + domain_pattern = r'^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$' + if not re.match(domain_pattern, domain): + raise ValueError(f"Invalid domain format: {domain}") + + # Return normalized format: @user@domain.tld (domain lowercase, user case-sensitive) + return f"@{username}@{domain.lower()}" + + def get_rarity_to_weight( config_section: configparser.SectionProxy) -> dict[int, float]: """Parses Rarity_X keys from config and returns a {rarity: weight} dict.""" @@ -38,19 +85,26 @@ config = configparser.ConfigParser() config.read(get_config_file()) # Username for the bot -USER = config['credentials']['User'].lower() +if 'User' not in config['credentials'] or not config['credentials']['User'].strip(): + raise ConfigError("User must be specified in config.ini under [credentials]") + +USER = normalize_user(config['credentials']['User']) # API key for the bot KEY = config['credentials']['Token'] # Bot's Misskey instance URL INSTANCE = config['credentials']['Instance'].lower() +# Web server port +WEB_PORT = config['application'].getint('WebPort', 5000) +BIND_ADDRESS = config['application'].get('BindAddress', '127.0.0.1') + # Fedi handles in the traditional 'user@domain.tld' style, allows these users # to use extra admin exclusive commands with the bot ADMINS = json.loads(config['application']['DefaultAdmins']) # SQLite Database location -DB_PATH = config['application']['DatabaseLocation'] +DB_PATH = config['application'].get('DatabaseLocation', './gacha_game.db') # Whether to enable the instance whitelist -USE_WHITELIST = config['application']['UseWhitelist'] +USE_WHITELIST = config['application'].getboolean('UseWhitelist', True) NOTIFICATION_POLL_INTERVAL = int(config['notification']['PollInterval']) NOTIFICATION_BATCH_SIZE = int(config['notification']['BatchSize']) diff --git a/example_config.ini b/example_config.ini index 0ea2422..8c18c28 100644 --- a/example_config.ini +++ b/example_config.ini @@ -5,6 +5,11 @@ DefaultAdmins = ["@localadmin", "@remoteadmin@example.tld"] ; SQLite Database location DatabaseLocation = ./gacha_game.db +; Web server port (default: 5000) +WebPort = 5000 +; Web server bind address (default: 127.0.0.1, set to 0.0.0.0 to listen on all interfaces) +BindAddress = 127.0.0.1 + ; Whether to lmit access to the bot via an instance whitelist ; The whitelist can be adjusted via the application UseWhitelist = False diff --git a/setup_db.py b/setup_db.py index 241bb4e..f1e9085 100644 --- a/setup_db.py +++ b/setup_db.py @@ -57,16 +57,14 @@ def perform_migration(cursor: sqlite3.Cursor, migration: tuple[int, str]) -> Non def get_db_path() -> str | DBNotFoundError: '''Gets the DB path from config.ini''' env = os.environ.get('KEMOVERSE_ENV') - if not (env and env in ['prod', 'dev']): - raise KemoverseEnvUnset - - print(f'Running in "{env}" mode') config_path = f'config_{env}.ini' if not os.path.isfile(config_path): raise ConfigError(f'Could not find {config_path}') + print(f'Running in "{env}" mode') + config = ConfigParser() config.read(config_path) db_path = config['application']['DatabaseLocation'] @@ -96,7 +94,6 @@ def main(): return except KemoverseEnvUnset: print('Error: KEMOVERSE_ENV is either not set or has an invalid value.') - print('Please set KEMOVERSE_ENV to either "dev" or "prod" before running.') print(traceback.format_exc()) return diff --git a/web/app.py b/web/app.py index 61ed38f..fcd67a1 100644 --- a/web/app.py +++ b/web/app.py @@ -1,10 +1,15 @@ import sqlite3 +import sys +from pathlib import Path +# Add parent directory to Python path so we can import from bot/ +sys.path.append(str(Path(__file__).parent.parent)) + +from bot.config import WEB_PORT, BIND_ADDRESS, DB_PATH from flask import Flask, render_template, abort from werkzeug.exceptions import HTTPException app = Flask(__name__) -DB_PATH = "./gacha_game.db" # Adjust path if needed def get_db_connection(): conn = sqlite3.connect(DB_PATH) @@ -68,4 +73,4 @@ def submit_character(): if __name__ == '__main__': - app.run(host='0.0.0.0', port=5000, debug=True) + app.run(host=BIND_ADDRESS, port=WEB_PORT, debug=True)