[FEATURE] Allow arbitrary RCon commands in chat commands (#678)

* Allow to run arbitrary rcon commands from chat commands

* add example for rcon command chat command

* Allow autosettings to enable and disable single chat commands

* Update api_commands.py

Signed-off-by: Florian <florian.schmidt.welzow@t-online.de>

* refactor chat commands

* Add conditions to chat rcon commands

These conditions work in a very same way as conditions in Auto Settings
(they are shared between them both).
All conditions are available in chat rcon commands, conditions that rely
on the player_id metric are only available in chat commands. Using these
conditions in auto settings will implicitly default to "condition is
met".

Two new conditions are available for chat commands (requiring the
player_id metric source):
- player_flags: A condition that is met, when the player has at least
  one of the provided flags. A list of flags can be provided, which are
  combined with OR, meaning that the player requires only one of these
  flags to  meet the condition.
- player_id: A condition thgat is met, when the player's ID is in the
  provided list of static IDs. This ID can either be the player's Steam
  ID or Windows ID, depending on whatever the player's platform is

Allowing the use of the auto settings conditions is a bit redundant, as
a single chat command can still be enabled and disabled using the set_chat_command_enabled
api, which is also available in auto settings. However, this gives the
user freedom to enable/disable the command using auto settings, or by a
condition in the command itself, whatever they prefer.

* WIP: Arguments in chat rcon commands

* Remove api endpoint to enable/disable rcon chat commands

With conditions on chat commands, the main use case to allow
autosettings enabling/disabling commands, this is not required anymore.

* WIP: Move rcon chat commands to a separate config

* WIP: Seed default (disabled) rcon chat commands

* add missing permissions, endpoints and default

* correctly apply default

* add redeploy example command

* black/isort

* command_words need to be on the base class as well

---------

Signed-off-by: Florian <florian.schmidt.welzow@t-online.de>
Co-authored-by: C. Eric Mathey <emathey@protonmail.com>

rcon/api_commands.py

@@ -61,6 +61,7 @@
 from rcon.user_config.log_line_webhooks import LogLineWebhookUserConfig
 from rcon.user_config.log_stream import LogStreamUserConfig
 from rcon.user_config.name_kicks import NameKickUserConfig
+from rcon.user_config.rcon_chat_commands import RConChatCommandsUserConfig
 from rcon.user_config.rcon_connection_settings import RconConnectionSettingsUserConfig
 from rcon.user_config.rcon_server_settings import RconServerSettingsUserConfig
 from rcon.user_config.real_vip import RealVipUserConfig
@@ -1768,6 +1769,41 @@ def validate_chat_commands_config(
             reset_to_default=reset_to_default,
         )
 
+    def get_rcon_chat_commands_config(self):
+        return RConChatCommandsUserConfig.load_from_db()
+
+    def set_rcon_chat_commands_config(
+        self,
+        by: str,
+        config: dict[str, Any] | BaseUserConfig | None = None,
+        reset_to_default: bool = False,
+        **kwargs,
+    ) -> bool:
+        return self._validate_user_config(
+            command_name=inspect.currentframe().f_code.co_name,  # type: ignore
+            by=by,
+            model=RConChatCommandsUserConfig,
+            data=config or kwargs,
+            dry_run=False,
+            reset_to_default=reset_to_default,
+        )
+
+    def validate_rcon_chat_commands_config(
+        self,
+        by: str,
+        config: dict[str, Any] | BaseUserConfig | None = None,
+        reset_to_default: bool = False,
+        **kwargs,
+    ) -> bool:
+        return self._validate_user_config(
+            command_name=inspect.currentframe().f_code.co_name,  # type: ignore
+            by=by,
+            model=RConChatCommandsUserConfig,
+            data=config or kwargs,
+            dry_run=True,
+            reset_to_default=reset_to_default,
+        )
+
     def get_log_stream_config(self):
         return LogStreamUserConfig.load_from_db()
 

rcon/audit.py

@@ -53,7 +53,7 @@ def set_registered_mods(moderators_name_steamids: List[tuple]):
         red.hset("moderators", k, v)
 
 
-def online_mods(rcon=None) -> list[AdminUserType]:
+def online_mods() -> list[AdminUserType]:
     red = _red()
     return [
         json.loads(red.get(u)) for u in red.scan_iter(f"{HEARTBEAT_KEY_PREFIX}*", 1)

rcon/auto_kick.py

@@ -3,12 +3,12 @@
 
 from pydantic import HttpUrl
 
+from rcon.api_commands import RconAPI
 from rcon.discord import send_to_discord_audit
 from rcon.game_logs import on_connected
 from rcon.hooks import inject_player_ids
 from rcon.player_history import get_player_profile, player_has_flag
 from rcon.user_config.name_kicks import NameKickUserConfig
-from rcon.api_commands import RconAPI
 
 logger = logging.getLogger(__name__)
 

rcon/auto_settings.py

@@ -1,231 +1,15 @@
 import logging
-import re
 import time
-from datetime import datetime
 
 import pytz
 
 from rcon.api_commands import get_rcon_api
-from rcon.audit import ingame_mods, online_mods
-from rcon.commands import BrokenHllConnection, CommandFailedError
+from rcon.conditions import Condition, create_condition
+from rcon.rcon import do_run_commands
 from rcon.user_config.auto_settings import AutoSettingsConfig
-from rcon.user_config.utils import BaseUserConfig
-
 
 logger = logging.getLogger(__name__)
 
-USER_CONFIG_NAME_PATTERN = re.compile(r"set_.*_config")
-
-
-def _get_current_map_metric(rcon):
-    try:
-        rcon.current_map = str(rcon.get_map())
-    except (CommandFailedError, BrokenHllConnection):
-        logger.exception("Failed to get current map")
-    return str(rcon.current_map)
-
-
-METRICS = {
-    "player_count": lambda rcon: int(rcon.get_slots()["current_players"]),
-    "online_mods": lambda: len(online_mods()),
-    "ingame_mods": lambda: len(ingame_mods()),
-    "current_map": _get_current_map_metric,
-    "time_of_day": lambda tz: datetime.now(tz=tz),
-}
-
-
-def is_user_config_func(name: str) -> bool:
-    return re.match(USER_CONFIG_NAME_PATTERN, name) is not None
-
-
-class BaseCondition:
-    def __init__(self, min=0, max=100, inverse=False, *args, **kwargs):
-        self.min = int(min)
-        self.max = int(max)
-        self.inverse = bool(inverse)
-
-        self.metric_name = ""
-        self.metric_source = "rcon"
-
-    @property
-    def metric_getter(self):
-        try:
-            return METRICS[self.metric_name]
-        except:
-            return None
-
-    def is_valid(self, **metric_sources):
-        metric_source = metric_sources[self.metric_source]
-        comparand = self.metric_getter(metric_source)
-        res = self.min <= comparand <= self.max
-        logger.info(
-            "Applying condition %s: %s <= %s <= %s = %s. Inverse: %s",
-            self.metric_name,
-            self.min,
-            comparand,
-            self.max,
-            res,
-            self.inverse,
-        )
-        if self.inverse:
-            return not res
-        else:
-            return res
-
-
-class PlayerCountCondition(BaseCondition):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.metric_name = "player_count"
-
-
-class OnlineModsCondition(BaseCondition):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.metric_name = "online_mods"
-        self.metric_source = None
-
-    def is_valid(self, **metric_sources):
-        comparand = self.metric_getter()
-        res = self.min <= comparand <= self.max
-        logger.info(
-            "Applying condition %s: %s <= %s <= %s = %s. Inverse: %s",
-            self.metric_name,
-            self.min,
-            comparand,
-            self.max,
-            res,
-            self.inverse,
-        )
-        if self.inverse:
-            return not res
-        else:
-            return res
-
-
-class IngameModsCondition(OnlineModsCondition):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.metric_name = "ingame_mods"
-        self.metric_source = None
-
-
-class CurrentMapCondition(BaseCondition):
-    def __init__(
-        self, map_names=[], inverse=False, *args, **kwargs
-    ):  # Avoid unexpected arguments
-        self.map_names = map_names
-        self.inverse = inverse
-        self.metric_name = "current_map"
-        self.metric_source = "rcon"
-
-    def is_valid(self, **metric_sources):
-        metric_source = metric_sources[self.metric_source]
-        comparand = self.metric_getter(metric_source)
-        res = comparand in self.map_names
-        logger.info(
-            "Applying condition %s: %s in %s = %s. Inverse: %s",
-            self.metric_name,
-            comparand,
-            self.map_names,
-            res,
-            self.inverse,
-        )
-        if self.inverse:
-            return not res
-        else:
-            return res
-
-
-class TimeOfDayCondition(BaseCondition):
-    def __init__(
-        self, min="00:00", max="23:59", timezone="utc", inverse=False, *args, **kwargs
-    ):  # Avoid unexpected arguments
-        self.min = str(min)
-        self.max = str(max)
-        if self.max in ["24:00", "0:00"]:
-            self.max = "23:59"
-        self.inverse = bool(inverse)
-        if timezone.lower() == "utc":
-            self.tz = pytz.UTC
-        else:
-            self.tz = pytz.timezone(timezone)
-        self.metric_name = "time_of_day"
-        self.metric_source = None
-
-    def is_valid(self, **metric_sources):
-        try:
-            min_h, min_m = [int(i) for i in self.min.split(":")[:2]]
-            max_h, max_m = [int(i) for i in self.max.split(":")[:2]]
-            min = datetime.now(tz=self.tz).replace(hour=min_h, minute=min_m)
-            max = datetime.now(tz=self.tz).replace(hour=max_h, minute=max_m)
-        except:
-            logger.exception("Time Of Day condition is invalid and is ignored")
-            return False  # The condition should fail
-        comparand = datetime.now(tz=self.tz)
-        res = min <= comparand <= max
-        logger.info(
-            "Applying condition %s: %s <= %s:%s <= %s = %s. Inverse: %s",
-            self.metric_name,
-            self.min,
-            comparand.hour,
-            comparand.minute,
-            self.max,
-            res,
-            self.inverse,
-        )
-        if self.inverse:
-            return not res
-        else:
-            return res
-
-
-def create_condition(name, **kwargs):
-    kwargs["inverse"] = kwargs.get("not", False)  # Using "not" would cause issues later
-    if name == "player_count":
-        return PlayerCountCondition(**kwargs)
-    elif name == "online_mods":
-        return OnlineModsCondition(**kwargs)
-    elif name == "ingame_mods":
-        return IngameModsCondition(**kwargs)
-    elif name == "current_map":
-        return CurrentMapCondition(**kwargs)
-    elif name == "time_of_day":
-        return TimeOfDayCondition(**kwargs)
-    else:
-        raise ValueError("Invalid condition type: %s" % name)
-
-
-def do_run_commands(rcon, commands):
-    for command, params in commands.items():
-        try:
-            logger.info("Applying %s %s", command, params)
-
-            # Allow people to apply partial changes to a user config to make
-            # auto settings less gigantic
-            if is_user_config_func(command):
-                # super dirty we should probably make an actual look up table
-                # but all the names are consistent
-                get_config_command = f"g{command[1:]}"
-                config: BaseUserConfig = rcon.__getattribute__(get_config_command)()
-                # get the existing config, override anything set in params
-                merged_params = config.model_dump() | params
-
-                if "by" not in merged_params:
-                    merged_params["by"] = "AutoSettings"
-
-                rcon.__getattribute__(command)(**merged_params)
-            else:
-                # Non user config settings
-                rcon.__getattribute__(command)(**params)
-        except AttributeError as e:
-            logger.exception(
-                "%s is not a valid command, double check the name!", command
-            )
-        except Exception as e:
-            logger.exception("Unable to apply %s %s: %s", command, params, e)
-        time.sleep(5)  # go easy on the server
-
 
 def run():
     rcon = get_rcon_api()
@@ -254,7 +38,7 @@ def run():
             )
 
         for rule in config["rules"]:
-            conditions: list[BaseCondition] = []
+            conditions: list[Condition] = []
             commands = rule.get("commands", {})
             for c_name, c_params in rule.get("conditions", {}).items():
                 try:

rcon/barricade.py

@@ -1,9 +1,10 @@
-from asgiref.sync import async_to_sync
-from channels.layers import get_channel_layer
 from datetime import datetime
 from enum import Enum
+
 import os
 import pydantic
+from asgiref.sync import async_to_sync
+from channels.layers import get_channel_layer
 
 from django.conf import settings as django_settings
 django_settings.configure(CHANNEL_LAYERS = {

rcon/commands.py

@@ -739,19 +739,20 @@ def get_gamestate(self) -> list[str]:
 
     def get_objective_row(self, row: int):
         if not (0 <= row <= 4):
-            raise ValueError('Row must be between 0 and 4')
-        
+            raise ValueError("Row must be between 0 and 4")
+
         return self._get_list(
-            f'get objectiverow_{row}',
-            fail_msgs="Cannot execute command for this gamemode."
+            f"get objectiverow_{row}",
+            fail_msgs="Cannot execute command for this gamemode.",
         )
 
     def set_game_layout(self, objectives: Sequence[str]):
         if len(objectives) != 5:
             raise ValueError("5 objectives must be provided")
         self._str_request(
-            f'gamelayout "{objectives[0]}" "{objectives[1]}" "{objectives[2]}" "{objectives[3]}" "{objectives[4]}"', log_info=True,
-            can_fail=False
+            f'gamelayout "{objectives[0]}" "{objectives[1]}" "{objectives[2]}" "{objectives[3]}" "{objectives[4]}"',
+            log_info=True,
+            can_fail=False,
         )
         return list(objectives)
 
@@ -761,6 +762,7 @@ def get_game_mode(self):
         """
         return self._str_request("get gamemode", can_fail=False)
 
+
 if __name__ == "__main__":
     from rcon.settings import SERVER_INFO