Gaming/the-button
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Admin interface

Browse Source
This commit is contained in:
eric
2026-10-22 11:26:42 -04:00
parent 2a2d976013
commit 3b81cae024
6 changed files with 583 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test_server.py
+215
View File
@@ -48,6 +48,7 @@ class PersistenceTests(unittest.TestCase):
"token_hashes": ["first-session", "second-session"],
"created_at": 123,
"subscribed": True,
"is_admin": False, "is_banned": False, "is_deleted": False,
},
"bob": {
"name": "Bøb",
@@ -55,6 +56,7 @@ class PersistenceTests(unittest.TestCase):
"token_hashes": [],
"created_at": 456,
"subscribed": False,
"is_admin": False, "is_banned": False, "is_deleted": False,
},
}
state["presses"] = [
@@ -67,6 +69,219 @@ class PersistenceTests(unittest.TestCase):
def write_legacy(self, state):
self.game.DATA_FILE.write_text(json.dumps(state), encoding="utf-8")
def admin_client(self):
request = self.presence_client()
self.game.state["players"]["alice"]["is_admin"] = True
self.game.save_state(self.game.state)
return request
def test_admin_endpoints_deny_guests_and_regular_players(self):
request = self.admin_client()
for path in ("/api/admin",):
self.assertEqual(request(path=path)[0], 401)
self.assertEqual(request(token="bob-token", path=path)[0], 403)
for action in ("set_admin", "set_banned", "delete_player", "delete_message"):
payload = {"action": action, "target_id": "alice", "value": True, "at": 1}
self.assertEqual(request("POST", payload, path="/api/admin/action")[0], 401)
self.assertEqual(request("POST", payload, "bob-token", "/api/admin/action")[0], 403)
self.assertTrue(self.game.state["players"]["alice"]["is_admin"])
def test_admin_overview_does_not_expose_credentials_and_role_is_private(self):
request = self.admin_client()
status, dashboard = request(token="alice-token", path="/api/admin")
self.assertEqual(status, 200)
self.assertEqual(dashboard["actor_id"], "alice")
self.assertEqual(len(dashboard["players"]), 2)
raw = json.dumps(dashboard)
for secret in ("password_hash", "token_hash", "existing-password-hash", "first-session"):
self.assertNotIn(secret, raw)
self.assertTrue(request(token="alice-token", path="/api/me")[1]["is_admin"])
for row in self.game.public_leaderboard() + self.game.global_stats()["leaderboard"]:
self.assertNotIn("is_admin", row)
def test_admin_role_changes_revoke_sessions_and_survive_save_reload(self):
request = self.admin_client()
payload = {"action": "set_admin", "target_id": "bob", "value": True}
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
self.assertTrue(self.game.state["players"]["bob"]["is_admin"])
self.assertEqual(request(token="bob-token", path="/api/admin")[0], 401)
self.assertEqual(self.game.state["players"]["bob"]["token_hashes"], [])
self.game.save_state(self.game.state)
self.game.state = self.game.load_state()
self.assertTrue(self.game.state["players"]["bob"]["is_admin"])
# Simulate a new authenticated session, then demote it from a different admin.
self.game.token_index[self.game.hash_token("bob-new")] = "bob"
self.game.state["players"]["bob"]["token_hashes"] = [self.game.hash_token("bob-new")]
self.assertEqual(request(token="bob-new", path="/api/admin")[0], 200)
self.assertEqual(request("POST", {**payload, "value": False}, "alice-token", "/api/admin/action")[0], 200)
self.assertEqual(request(token="bob-new", path="/api/admin")[0], 401)
def test_ban_revokes_all_access_and_unban_requires_new_login(self):
request = self.admin_client()
self.game.state["players"]["bob"]["password_hash"] = self.game.hash_password("test-password")
self.game.save_state(self.game.state)
update = {"tab_id": "a" * 32, "active": True, "in_chat": True}
self.assertEqual(request("POST", update, "bob-token")[0], 200)
payload = {"action": "set_banned", "target_id": "bob", "value": True}
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
for path, method, data in (("/api/me", "GET", None), ("/api/press", "POST", {}),
("/api/chat", "POST", {"message": "Banned"}), ("/api/presence", "POST", update),
("/api/subscribe", "POST", {})):
self.assertEqual(request(method, data, "bob-token", path)[0], 401)
self.assertEqual(request()[1]["online_count"], 0)
self.assertEqual(request("POST", {"name": "Bøb", "password": "test-password"}, path="/api/login")[0], 401)
self.assertEqual(request("POST", {"action": "set_admin", "target_id": "bob", "value": True}, "alice-token", "/api/admin/action")[0], 409)
self.assertEqual(request("POST", {**payload, "value": False}, "alice-token", "/api/admin/action")[0], 200)
self.assertFalse(self.game.state["players"]["bob"]["is_banned"])
self.assertEqual(request(token="bob-token", path="/api/me")[0], 401)
self.assertEqual(request("POST", {"name": "Bøb", "password": "test-password"}, path="/api/login")[0], 200)
def test_admin_self_protection_and_last_active_admin(self):
request = self.admin_client()
for action, value in (("set_admin", False), ("set_banned", True), ("delete_player", None)):
payload = {"action": action, "target_id": "alice", "value": value}
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 409)
self.assertEqual(sum(self.game.active_admin(p) for p in self.game.state["players"].values()), 1)
self.assertEqual(request("POST", {"action": "set_admin", "target_id": "bob", "value": "true"}, "alice-token", "/api/admin/action")[0], 400)
self.assertEqual(request("POST", {"action": "delete_player", "target_id": "missing"}, "alice-token", "/api/admin/action")[0], 404)
self.assertEqual(request("POST", {"action": "execute_sql", "target_id": "bob"}, "alice-token", "/api/admin/action")[0], 400)
def test_admin_deletes_chat_without_id_reuse_or_cooldown_bypass(self):
request = self.admin_client()
with self.game.database() as connection:
at = self.game.now_ms()
cursor = connection.execute("INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
("bob", "Bøb", at, "Remove this"))
mid = cursor.lastrowid
payload = {"action": "delete_message", "target_id": str(mid), "at": at}
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
self.assertFalse(any(m["id"] == mid for m in self.game.chat_messages()))
self.assertEqual(request("POST", {"message": "Too soon"}, "bob-token", "/api/chat")[0], 429)
with self.game.database() as connection:
self.assertEqual(connection.execute("SELECT message FROM chat_messages WHERE id = ?", (mid,)).fetchone()[0], "[Message removed]")
self.game.load_state()
self.assertFalse(any(m["id"] == mid for m in self.game.chat_messages()))
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 404)
def test_admin_can_remove_system_milestone_without_reannouncing_it(self):
request = self.admin_client()
at = self.game.now_ms()
with self.game.database() as connection:
connection.execute("UPDATE community_milestones SET reached_at = ?, announced_at = ?, message = ? WHERE id = 'ten_players'", (at, at, "Milestone to remove"))
payload = {"action": "delete_message", "target_id": "milestone:ten_players", "at": at}
self.assertEqual(request("POST", payload, "alice-token", "/api/admin/action")[0], 200)
self.game.check_ended()
self.assertFalse(any(m["message"] == "Milestone to remove" for m in self.game.chat_messages()))
with self.game.database() as connection:
self.assertIsNotNone(connection.execute("SELECT reached_at FROM community_milestones WHERE id = 'ten_players'").fetchone()[0])
boundary = self.game.current_season(connection)["ends_at"]
self.game.check_ended(boundary)
self.assertFalse(any(m["at"] == at and m["kind"] == "milestone" for m in self.game.chat_messages()))
def test_admin_account_deletion_anonymizes_without_losing_game_history(self):
request = self.admin_client()
before_presses = copy.deepcopy(self.game.state["presses"])
with self.game.database() as connection:
season = self.game.current_season(connection)
self.game.archive_season(connection, season, self.game.state, self.game.state["deadline"], "countdown")
connection.execute("INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)", ("bob", "Bøb", 123, "My message"))
self.assertEqual(request("POST", {"action": "delete_player", "target_id": "bob"}, "alice-token", "/api/admin/action")[0], 200)
player = self.game.state["players"]["bob"]
self.assertTrue(player["is_deleted"])
self.assertEqual(player["name"], "Deleted player")
self.assertIsNone(player["password_hash"])
self.assertEqual(player["token_hashes"], [])
self.assertIsNone(self.game.find_player_by_name("Bøb"))
self.assertEqual(self.game.state["presses"], before_presses)
self.assertEqual(request(token="bob-token", path="/api/me")[0], 401)
self.assertNotIn("bob", [p["id"] for p in request(token="alice-token", path="/api/admin")[1]["players"]])
self.assertNotIn("Bøb", json.dumps(self.game.seasonal_leaderboard(), ensure_ascii=False))
self.assertTrue(any(m["message"] == "[Account deleted]" for m in self.game.chat_messages()))
self.game.state = self.game.load_state()
self.assertTrue(self.game.state["players"]["bob"]["is_deleted"])
with self.game.database() as connection:
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
self.assertEqual(connection.execute("SELECT action, actor_id FROM admin_audit").fetchone()["action"], "delete_player")
def test_admin_action_rollback_does_not_revoke_or_mutate(self):
request = self.admin_client()
before = copy.deepcopy(self.game.state)
with self.game.database() as connection:
connection.execute("""CREATE TRIGGER fail_admin_audit BEFORE INSERT ON admin_audit
BEGIN SELECT RAISE(ABORT, 'test rollback'); END""")
with self.assertRaises(sqlite3.IntegrityError):
self.game.admin_action("alice", {"action": "set_banned", "target_id": "bob", "value": True})
self.assertEqual(self.game.state, before)
self.assertIn(self.game.hash_token("bob-token"), self.game.token_index)
self.assertEqual(request(token="bob-token", path="/api/me")[0], 200)
with self.game.database() as connection:
connection.execute("DROP TRIGGER fail_admin_audit")
self.assertEqual(self.game.load_state(), before)
def test_public_registration_cannot_grant_admin_or_bypass_ban_flags(self):
request = self.admin_client()
data = {"name": "Newcomer", "password": "test-password", "is_admin": True, "is_banned": False}
status, result = request("POST", data, path="/api/register")
self.assertEqual(status, 201)
pid = self.game.find_player_by_name("Newcomer")
self.assertFalse(self.game.state["players"][pid].get("is_admin", False))
self.assertEqual(request(token=result["token"], path="/api/admin")[0], 403)
def test_admin_schema_upgrade_and_database_provisioning(self):
self.game.load_state()
self.game.state = self.sample_state()
self.game.save_state(self.game.state)
with self.game.database() as connection:
connection.execute("PRAGMA foreign_keys = OFF")
connection.execute("""CREATE TABLE old_players (
id TEXT PRIMARY KEY, name TEXT NOT NULL, password_hash TEXT,
created_at INTEGER NOT NULL, subscribed INTEGER NOT NULL
)""")
connection.execute("INSERT INTO old_players SELECT id, name, password_hash, created_at, subscribed FROM players")
connection.execute("DROP TABLE players")
connection.execute("ALTER TABLE old_players RENAME TO players")
loaded = self.game.load_state()
self.assertFalse(any(p["is_admin"] for p in loaded["players"].values()))
with self.game.database() as connection:
connection.execute("UPDATE players SET is_admin = 1 WHERE id = ?", ("alice",))
self.game.state = self.game.load_state()
self.assertTrue(self.game.state["players"]["alice"]["is_admin"])
self.game.save_state(self.game.state)
self.assertTrue(self.game.load_state()["players"]["alice"]["is_admin"])
with self.game.database() as connection:
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
def test_legacy_json_cannot_provision_administrators(self):
legacy = self.sample_state()
legacy["players"]["alice"].update(is_admin=True, is_banned=True, is_deleted=True)
self.write_legacy(legacy)
self.game.state = self.game.load_state()
for field in ("is_admin", "is_banned", "is_deleted"):
self.assertFalse(self.game.state["players"]["alice"][field])
self.assertFalse(self.game.load_state()["players"]["alice"][field])
def test_admin_malformed_request_and_cross_site_mutation(self):
request = self.admin_client()
self.assertEqual(request("POST", {}, "alice-token", "/api/admin/unknown")[0], 404)
httpd = ThreadingHTTPServer(("127.0.0.1", 0), self.game.Handler)
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
thread.start()
self.addCleanup(httpd.server_close)
self.addCleanup(thread.join)
self.addCleanup(httpd.shutdown)
headers = {"Authorization": "Bearer alice-token", "Content-Type": "application/json"}
for body, extra, expected in (("{broken", {}, 400), ("[]", {}, 400),
(json.dumps({"action": "set_banned", "target_id": "bob", "value": True}), {"Sec-Fetch-Site": "cross-site"}, 403)):
connection = HTTPConnection(*httpd.server_address, timeout=5)
try:
connection.request("POST", "/api/admin/action", body, {**headers, **extra})
response = connection.getresponse()
self.assertEqual(response.status, expected)
response.read()
finally:
connection.close()
self.assertFalse(self.game.state["players"]["bob"]["is_banned"])
def season_fixture(self):
at = int(datetime(2026, 10, 21, tzinfo=timezone.utc).timestamp() * 1000)
with patch.object(self.game, "now_ms", return_value=at):