parity: security_auth_bans — reopen BAN_PERMIT parity tasks

mud/account/account_service.py

@@ -6,6 +6,7 @@
 from mud.db.models import PlayerAccount, Character
 from mud.security.hash_utils import hash_password, verify_password
 from mud.security import bans
+from mud.security.bans import BanFlag
 
 
 def create_account(username: str, raw_password: str) -> bool:
@@ -49,7 +50,18 @@ def login_with_host(
     This wrapper checks both account and host bans and only then delegates to
     the standard login function.
     """
-    if bans.is_host_banned(host):
+    permit_host = bool(host and bans.is_host_banned(host, BanFlag.PERMIT))
+    if host and not permit_host and bans.is_host_banned(host, BanFlag.ALL):
+        return None
+    session = SessionLocal()
+    try:
+        exists = (
+            session.query(PlayerAccount).filter_by(username=username).first()
+            is not None
+        )
+    finally:
+        session.close()
+    if host and not permit_host and not exists and bans.is_host_banned(host, BanFlag.NEWBIES):
         return None
     return login(username, raw_password)
 

mud/commands/admin_commands.py

@@ -40,7 +40,7 @@ def cmd_ban(char: Character, args: str) -> str:
     host = args.strip()
     if not host:
         return "Usage: ban <host>"
-    bans.add_banned_host(host)
+    bans.add_banned_host(host, permanent=False)
     try:
         bans.save_bans_file()
     except Exception:
@@ -49,6 +49,18 @@ def cmd_ban(char: Character, args: str) -> str:
     return f"Banned {host}."
 
 
+def cmd_permban(char: Character, args: str) -> str:
+    host = args.strip()
+    if not host:
+        return "Usage: permban <host>"
+    bans.add_banned_host(host, permanent=True)
+    try:
+        bans.save_bans_file()
+    except Exception:
+        pass
+    return f"Banned {host}."
+
+
 def cmd_unban(char: Character, args: str) -> str:
     host = args.strip()
     if not host:
@@ -64,14 +76,12 @@ def cmd_unban(char: Character, args: str) -> str:
 
 
 def cmd_banlist(char: Character, args: str) -> str:
-    banned = sorted(list({h for h in list_hosts() for h in [h]}))
+    banned = list_hosts()
     if not banned:
         return "No sites banned."
     lines = ["Banned sites:"] + [f" - {h}" for h in banned]
     return "\n".join(lines)
 
 
 def list_hosts() -> list[str]:
-    # internal helper to read via saving/loading outward if needed later
-    # currently directly exposes in-memory set
-    return sorted({*bans._banned_hosts})  # type: ignore[attr-defined]
+    return sorted(entry.to_pattern() for entry in bans.get_ban_entries())

mud/commands/dispatcher.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 from dataclasses import dataclass
 import shlex
-from typing import Callable, Dict, List, Optional
+from typing import Callable, Dict, List, Optional, Tuple
 
 from mud.models.character import Character
 from .movement import do_north, do_south, do_east, do_west, do_up, do_down, do_enter
@@ -14,6 +14,7 @@
     cmd_teleport,
     cmd_spawn,
     cmd_ban,
+    cmd_permban,
     cmd_unban,
     cmd_banlist,
 )
@@ -62,7 +63,7 @@ class Command:
     Command("equipment", do_equipment, aliases=("eq",), min_position=Position.DEAD),
 
     # Communication
-    Command("say", do_say, min_position=Position.RESTING),
+    Command("say", do_say, aliases=("'",), min_position=Position.RESTING),
     Command("tell", do_tell, min_position=Position.RESTING),
     Command("shout", do_shout, min_position=Position.RESTING),
 
@@ -97,6 +98,7 @@ class Command:
     Command("@teleport", cmd_teleport, admin_only=True),
     Command("@spawn", cmd_spawn, admin_only=True),
     Command("ban", cmd_ban, admin_only=True),
+    Command("permban", cmd_permban, admin_only=True),
     Command("unban", cmd_unban, admin_only=True),
     Command("banlist", cmd_banlist, admin_only=True),
     Command("@redit", cmd_redit, admin_only=True),
@@ -121,40 +123,60 @@ def resolve_command(name: str) -> Optional[Command]:
     return matches[0] if matches else None
 
 
+def _split_command_and_args(input_str: str) -> Tuple[str, str]:
+    """Extract the leading command token and its remaining arguments."""
+
+    stripped = input_str.lstrip()
+    if not stripped:
+        return "", ""
+
+    first = stripped[0]
+    if not first.isalnum():
+        return first, stripped[1:].lstrip()
+
+    try:
+        parts = shlex.split(stripped)
+        if not parts:
+            return "", ""
+        head = parts[0]
+        tail = " ".join(parts[1:]) if len(parts) > 1 else ""
+        return head, tail
+    except ValueError:
+        fallback = stripped.split(None, 1)
+        if not fallback:
+            return "", ""
+        head = fallback[0]
+        tail = fallback[1] if len(fallback) > 1 else ""
+        return head, tail
+
+
 def _expand_aliases(char: Character, input_str: str, *, max_depth: int = 5) -> str:
     """Expand the first token using per-character aliases, up to max_depth."""
+
     s = input_str
     for _ in range(max_depth):
-        try:
-            parts = shlex.split(s)
-        except ValueError:
-            return s
-        if not parts:
+        head, tail = _split_command_and_args(s)
+        if not head:
             return s
-        head, tail = parts[0], parts[1:]
         expansion = char.aliases.get(head)
         if not expansion:
             return s
-        s = (expansion + (" " + " ".join(tail) if tail else "")).strip()
+        s = (expansion + (" " + tail if tail else "")).strip()
     return s
 
 
 def process_command(char: Character, input_str: str) -> str:
     if not input_str.strip():
         return "What?"
     expanded = _expand_aliases(char, input_str)
-    try:
-        parts = shlex.split(expanded)
-    except ValueError:
-        return "Huh?"
-    if not parts:
+    cmd_name, arg_str = _split_command_and_args(expanded)
+    if not cmd_name:
         return "What?"
-    cmd_name, *args = parts
     command = resolve_command(cmd_name)
     if not command:
         social = social_registry.get(cmd_name.lower())
         if social:
-            return perform_social(char, cmd_name, " ".join(args))
+            return perform_social(char, cmd_name, arg_str)
         return "Huh?"
     if command.admin_only and not getattr(char, "is_admin", False):
         return "You do not have permission to use this command."
@@ -177,7 +199,6 @@ def process_command(char: Character, input_str: str) -> str:
             return "No way!  You are still fighting!"
         # Fallback (should not happen)
         return "You can't do that right now."
-    arg_str = " ".join(args)
     # Log admin commands (accepted) to admin log for auditability.
     if command.admin_only and getattr(char, "is_admin", False):
         try:

mud/models/character.py

@@ -106,6 +106,8 @@ class Character:
     # Combat skill levels (0-100) for multi-attack mechanics
     second_attack_skill: int = 0
     third_attack_skill: int = 0
+    # Charm/follow hierarchy pointer (ROM ch->master)
+    master: Optional["Character"] = None
     # Combat state - currently fighting target
     fighting: Optional["Character"] = None
     # Enhanced damage skill level (0-100)

mud/net/connection.py

@@ -12,6 +12,8 @@
 from mud.commands import process_command
 from mud.net.session import Session, SESSIONS
 from mud.net.protocol import send_to_char
+from mud.security import bans
+from mud.security.bans import BanFlag
 
 
 async def handle_connection(
@@ -47,6 +49,18 @@ async def handle_connection(
             # Enforce site/account bans at login time
             account = login_with_host(username, password, host_for_ban)
             if not account:
+                permit_host = bool(
+                    host_for_ban and bans.is_host_banned(host_for_ban, BanFlag.PERMIT)
+                )
+                if host_for_ban and not permit_host:
+                    if bans.is_host_banned(host_for_ban, BanFlag.ALL):
+                        writer.write(b"Your site has been banned from this mud.\r\n")
+                        await writer.drain()
+                        return
+                    if bans.is_host_banned(host_for_ban, BanFlag.NEWBIES):
+                        writer.write(b"New players are not allowed from your site.\r\n")
+                        await writer.drain()
+                        return
                 if create_account(username, password):
                     account = login_with_host(username, password, host_for_ban)
                 else: