Slightly improve default RNG and make it configurable

The game (i.e. everything except the messages system) now uses a custom
Random subclass rather than the default Random implementation. This is
based on the ChaCha20 stream cipher; re-keying itself every 992 random
bytes processed and re-initializing from a (hopefully good) random
source (default os.urandom, i.e. /dev/urandom on linux) every time a
game is started. The RNG supports seed operations, allowing for
reproducible results in an eventual replay system.

requirements.txt

@@ -1,3 +1,4 @@
 antlr4-python3-runtime >= 4.8.0, < 4.9.0
 ruamel.yaml >= 0.16.0, < 1.0.0
 requests >= 2.24.0, < 3.0.0
+cryptography >= 38.0.4

src/__init__.py

@@ -3,7 +3,7 @@
 # This "bootstraps" the bot in preparation for importing the bulk of the code. Some imports
 # change behavior based on whether or not we're in debug mode, so that must be established before
 # we continue on to import other files
-from src import config, lineparse, locks, match
+from src import config, lineparse, locks, match, random
 
 # Initialize config.Main
 config.init()

src/gamemodes/maelstrom.py

@@ -1,14 +1,13 @@
-import random
 from collections import defaultdict, Counter
 
 from src.gamemodes import game_mode, GameMode
-from src.messages import messages
 from src.functions import get_players
 from src.gamestate import GameState
 from src.events import Event, EventListener
 from src.trans import chk_win_conditions
-from src import channels, users
+from src import users
 from src.cats import All, Team_Switcher, Win_Stealer, Wolf, Wolf_Objective, Vampire_Objective, Killer
+from src.random import random
 
 @game_mode("maelstrom", minp=8, maxp=24)
 class MaelstromMode(GameMode):

src/gamemodes/pactbreaker.py

@@ -3,7 +3,6 @@
 import functools
 import logging
 import math
-import random
 import re
 from collections import defaultdict
 from typing import Iterable, Optional
@@ -28,6 +27,7 @@
 from src.roles.vampire import on_player_protected as vampire_drained
 from src.roles.vampire import GameState as VampireGameState
 from src.roles.vigilante import vigilante_retract, vigilante_pass, vigilante_kill
+from src.random import random
 
 # dummy location for wolves/vigilantes/vampires that have elected to kill/bite instead of visit a location
 Limbo = Location("<<hunting>>")

src/gamemodes/random.py

@@ -1,11 +1,12 @@
-import random
 from collections import defaultdict
+
 from src.gamemodes import game_mode, GameMode
 from src.gamestate import GameState
 from src.events import EventListener, Event
 from src.trans import chk_win_conditions
 from src import users
 from src.cats import All, Wolf, Wolf_Objective, Vampire_Objective, Killer
+from src.random import random
 
 @game_mode("random", minp=8, maxp=24)
 class RandomMode(GameMode):

src/gamemodes/sleepy.py

@@ -1,4 +1,3 @@
-import random
 from collections import Counter, defaultdict
 
 from src.cats import Wolf
@@ -13,6 +12,7 @@
 from src.events import EventListener, Event
 from src import channels, config
 from src.users import User
+from src.random import random
 
 @game_mode("sleepy", minp=8, maxp=24)
 class SleepyMode(GameMode):

src/gamestate.py

@@ -10,7 +10,7 @@
 from src.cats import All
 from src import config
 from src.users import User
-from src import channels
+from src import channels, random
 
 if TYPE_CHECKING:
     from src.gamemodes import GameMode
@@ -77,6 +77,7 @@ def __init__(self, pregame_state: PregameState):
         self.next_phase: Optional[str] = None
         self.night_count: int = 0
         self.day_count: int = 0
+        self.rng_seed: bytes = random.get_seed()
 
     def __init_subclass__(cls, **kwargs):
         super().__init_subclass__(**kwargs)

src/locations.py

@@ -5,7 +5,7 @@
 from src.users import User
 from src.events import Event, event_listener
 
-__all__ = ["Location", "VillageSquare", "Graveyard", "Forest",
+__all__ = ["Location", "VillageSquare", "Graveyard", "Forest", "Streets",
            "get_players_in_location", "get_location", "get_home",
            "move_player", "move_player_home", "set_home"]
 

src/pregame.py

@@ -5,7 +5,6 @@
 
 import threading
 import itertools
-import random
 import time
 import math
 import re
@@ -24,6 +23,7 @@
 from src.dispatcher import MessageDispatcher
 from src.channels import Channel
 from src.locations import Location, set_home
+from src.random import random
 
 WAIT_TOKENS = 0
 WAIT_LAST = 0
@@ -240,6 +240,7 @@ def _isvalid(mode, allow_vote_only):
     # Initial checks passed, game mode has been fully initialized
     # We move from pregame state to in-game state
     channels.Main.game_state = ingame_state = GameState(pregame_state)
+    random.seed(ingame_state.rng_seed)
 
     event = Event("role_attribution", {"addroles": Counter()})
     if event.dispatch(ingame_state, villagers):

src/random.py

@@ -0,0 +1,114 @@
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms
+import os
+import random as py_random
+from typing import Optional
+
+__all__ = ["random", "seed_function", "get_seed"]
+# seed_function must be a function accepting an int and returning a bytestring containing that many random bytes
+seed_function = os.urandom
+
+# internal constants
+KEY_SIZE = 32
+NONCE_SIZE = 16
+BUFFER_SIZE = 1024
+
+def get_seed():
+    """Retrieve a seed suitable for passing to random.seed()."""
+    return seed_function(KEY_SIZE)
+
+class GameRNG(py_random.Random):
+    """A better RNG than default random while providing the same API.
+
+    This RNG makes use of chacha20 to provide a bitstream, with support
+    for deterministic keying for replays. An extensibility point is provided
+    via random.SEED_FUNCTION, which can be set to an arbitrary callback in
+    custom hooks.py to override default seeding behavior (os.urandom).
+
+    This is not secure for any sort of cryptographic use; it is purely to
+    provide a better RNG for gameplay purposes.
+    """
+    def __init__(self, seed=None):
+        self._cur: bytes = b""
+        self._next: bytes = b""
+        self._buffer: bytes = b""
+        self._offset: int = 0
+        super().__init__(seed)
+
+    def getstate(self):
+        return self._cur, self._offset
+
+    def setstate(self, state):
+        key, offset, reseed_count = state
+        if not isinstance(key, bytes) or len(key) != KEY_SIZE:
+            raise TypeError("Invalid key state")
+        if not isinstance(offset, int) or offset < KEY_SIZE or offset >= BUFFER_SIZE:
+            raise TypeError("Invalid offset state")
+
+        cipher = Cipher(algorithms.ChaCha20(key, b"\x00" * NONCE_SIZE), None)
+        enc = cipher.encryptor()
+        self._cur = key
+        self._buffer = enc.update(b"\x00" * BUFFER_SIZE)
+        self._next = self._buffer[0:KEY_SIZE]
+        self._offset = offset
+
+    def seed(self, a: Optional[bytes] = None, version: int = 2) -> None:
+        if a is None:
+            self._cur = self._next = seed_function(KEY_SIZE)
+        elif not isinstance(a, bytes):
+            raise TypeError(f"seed must be a bytes string, got {type(a)} instead")
+        elif len(a) != KEY_SIZE:
+            raise TypeError(f"seed must be {KEY_SIZE} bytes, got {len(a)} bytes instead")
+        else:
+            self._cur = self._next = a
+
+        self._reseed()
+
+    def getrandbits(self, k: int, /) -> int:
+        # implementation details largely lifted from SystemRandom.getrandbits
+        # and updated to use self.randbytes instead of urandom
+        if k < 0:
+            raise ValueError("Number of bits must be non-negative")
+
+        numbytes = (k + 7) // 8  # bits / 8 and rounded up
+        x = int.from_bytes(self.randbytes(numbytes))
+        return x >> (numbytes * 8 - k)  # trim excess bits
+
+    def random(self) -> float:
+        # implementation details largely lifted from SystemRandom.random
+        # and updated to use self.randbytes instead of urandom
+        # python floats are doubles, and doubles have 53 bits for the significand
+        # we don't want to populate exponent with any random data or else we massively bias results
+        return (int.from_bytes(self.randbytes(7)) >> 3) * (2 ** -53)
+
+    def randbytes(self, n) -> bytes:
+        if n < 0:
+            raise ValueError("Number of bytes must be non-negative")
+        elif n == 0:
+            return b""
+
+        buf = bytearray(n)
+        i = 0
+        while n > 0:
+            remaining = BUFFER_SIZE - self._offset
+            if remaining > n:
+                buf[i:i + n] = self._buffer[self._offset:self._offset + n]
+                self._offset += n
+                break
+            else:
+                buf[i:i + remaining] = self._buffer[self._offset:]
+                n -= remaining
+                i += remaining
+                self._reseed()
+
+        return bytes(buf)
+
+    def _reseed(self) -> None:
+        cipher = Cipher(algorithms.ChaCha20(self._next, b"\x00" * NONCE_SIZE), None)
+        enc = cipher.encryptor()
+        self._cur = self._next
+        self._buffer = enc.update(b"\x00" * BUFFER_SIZE)
+        self._next = self._buffer[0:KEY_SIZE]
+        self._offset = KEY_SIZE
+
+# named so things can do `from src.random import random` and use all of the `random.blah()` APIs without change
+random = GameRNG()

src/roles/_skel.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import re
-import random
 import itertools
 import math
 from collections import defaultdict
@@ -16,6 +15,7 @@
 from src.events import Event, event_listener
 from src.gamestate import GameState
 from src.users import User
+from src.random import random
 
 # Skeleton file for new roles. Not all events are represented, only the most common ones.
 

src/roles/amnesiac.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import random
 from typing import Optional
 
 from src import users, config
@@ -11,6 +10,7 @@
 from src.messages import messages
 from src.gamestate import GameState
 from src.users import User
+from src.random import random
 
 __all__ = ["get_blacklist", "get_stats_flag"]
 

src/roles/angel.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import random
 import re
 from typing import Optional, Union
 
@@ -16,6 +15,7 @@
 from src.messages import messages
 from src.status import try_misdirection, try_exchange, add_protection, add_dying
 from src.users import User
+from src.random import random
 
 GUARDED: UserDict[users.User, users.User] = UserDict()
 LASTGUARDED: UserDict[users.User, users.User] = UserDict()

src/roles/assassin.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import random
 import re
 from typing import Optional
 
@@ -14,6 +13,7 @@
 from src.messages import messages
 from src.status import try_misdirection, try_exchange, try_protection, add_dying, is_silent
 from src.users import User
+from src.random import random
 
 TARGETED: UserDict[users.User, users.User] = UserDict()
 PREV_ACTED = UserSet()

src/roles/bodyguard.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import random
 import re
 from typing import Optional, Union
 
@@ -15,6 +14,7 @@
 from src.users import User
 from src.dispatcher import MessageDispatcher
 from src.gamestate import GameState
+from src.random import random
 
 GUARDED: UserDict[User, User] = UserDict()
 PASSED = UserSet()

src/roles/clone.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import random
 import re
 from typing import Optional
 
@@ -14,6 +13,7 @@
 from src.messages import messages
 from src.trans import NIGHT_IDLE_EXEMPT
 from src.users import User
+from src.random import random
 
 CLONED: UserDict[users.User, users.User] = UserDict()
 CAN_ACT = UserSet()

src/roles/crazedshaman.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import itertools
-import random
 from typing import Optional
 
 from src.decorators import command
@@ -14,6 +13,7 @@
 from src.status import is_silent
 from src import users
 from src.users import User
+from src.random import random
 
 TOTEMS, LASTGIVEN, SHAMANS, RETARGET, ORIG_TARGET_MAP = setup_variables("crazed shaman", knows_totem=False)
 

src/roles/detective.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import random
 import re
 from typing import Optional
 
@@ -16,6 +15,7 @@
 from src.dispatcher import MessageDispatcher
 from src.gamestate import GameState
 from src.users import User
+from src.random import random
 
 INVESTIGATED = UserSet()
 

src/roles/doctor.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import math
-import random
 import re
 from typing import Optional
 
@@ -15,6 +14,7 @@
 from src.messages import messages
 from src.status import try_misdirection, try_exchange, remove_lycanthropy, remove_disease
 from src.users import User
+from src.random import random
 
 IMMUNIZED = UserSet()
 DOCTORS: UserDict[users.User, int] = UserDict()

src/roles/doomsayer.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import random
 import re
 from typing import Optional
 
@@ -16,7 +15,7 @@
 from src.users import User
 from src.dispatcher import MessageDispatcher
 from src.gamestate import GameState
-from src.locations import move_player_home
+from src.random import random
 
 register_wolf("doomsayer")