SSO and turn gate
This commit is contained in:
+120
-3
@@ -5,6 +5,8 @@ import importlib.util
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
@@ -23,6 +25,11 @@ class PersistenceTests(unittest.TestCase):
|
||||
"BUTTON_DATA_FILE": str(Path(directory) / "data.json"),
|
||||
"BUTTON_DATABASE_FILE": str(Path(directory) / "data.sqlite3"),
|
||||
"BUTTON_MAILING_LIST_FILE": str(Path(directory) / "mailing.csv"),
|
||||
"BUTTON_OIDC_ISSUER": "https://dex.example.test",
|
||||
"BUTTON_OIDC_CLIENT_ID": "button-test-client",
|
||||
"BUTTON_OIDC_CLIENT_SECRET": "test-client-secret",
|
||||
"BUTTON_OIDC_REDIRECT_URI": "https://button.example.test/auth/oidc/callback",
|
||||
"BUTTON_OIDC_REQUIRED_GROUP": "the-button-users",
|
||||
}):
|
||||
spec = importlib.util.spec_from_file_location("button_test_server", Path(__file__).with_name("server.py"))
|
||||
cls.game = importlib.util.module_from_spec(spec)
|
||||
@@ -38,6 +45,9 @@ class PersistenceTests(unittest.TestCase):
|
||||
self.game.failed_logins.clear()
|
||||
self.game.subscribed_emails.clear()
|
||||
self.game.presence.clear()
|
||||
# Exercise legacy data/API behavior in isolated unit tests only. A real
|
||||
# server cannot start without complete Dex configuration.
|
||||
self.game.OIDC_ENABLED = False
|
||||
|
||||
def sample_state(self):
|
||||
state = self.game.new_game()
|
||||
@@ -1219,6 +1229,9 @@ class PersistenceTests(unittest.TestCase):
|
||||
self.game.state["deadline"] = t + 1234
|
||||
with patch.object(self.game, "now_ms", return_value=t):
|
||||
self.assertEqual(request("/api/press", {}, token)[0], 200)
|
||||
status, denied = request("/api/press", {}, token)
|
||||
self.assertEqual(status, 409)
|
||||
self.assertIn("2 more distinct players", denied["error"])
|
||||
announcements = self.game.chat_messages()
|
||||
self.assertEqual(len(announcements), 2)
|
||||
self.assertEqual(announcements[1]["kind"], "milestone")
|
||||
@@ -1231,17 +1244,63 @@ class PersistenceTests(unittest.TestCase):
|
||||
self.game.state = self.game.load_state()
|
||||
self.assertEqual(self.game.chat_messages()[0], announcements[0])
|
||||
self.game.token_index = {th: pid for pid, p in self.game.state["players"].items() for th in p["token_hashes"]}
|
||||
_, bob = request("/api/register", {"name": "Bob", "password": "test-password"})
|
||||
self.assertEqual(request("/api/press", {}, bob["token"])[0], 200)
|
||||
status, denied = request("/api/press", {}, token)
|
||||
self.assertEqual(status, 409)
|
||||
self.assertIn("1 more distinct player", denied["error"])
|
||||
_, carol = request("/api/register", {"name": "Carol", "password": "test-password"})
|
||||
self.assertEqual(request("/api/press", {}, carol["token"])[0], 200)
|
||||
self.game.state = self.game.load_state()
|
||||
self.game.token_index = {th: pid for pid, p in self.game.state["players"].items() for th in p["token_hashes"]}
|
||||
self.assertEqual(request("/api/press", {}, token)[0], 200)
|
||||
status, logged_in = request("/api/login", {"name": "Alice", "password": "test-password"})
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(request("/api/logout", {}, token)[0], 200)
|
||||
self.assertEqual(request("/api/press", {}, token)[0], 401)
|
||||
self.assertEqual(len([m for m in self.game.chat_messages() if m["kind"] == "press"]), 2)
|
||||
self.assertEqual(len([m for m in self.game.chat_messages() if m["kind"] == "press"]), 4)
|
||||
loaded = self.game.load_state()
|
||||
player = next(iter(loaded["players"].values()))
|
||||
player = loaded["players"][self.game.find_player_by_name("Alice")]
|
||||
self.assertTrue(player["subscribed"])
|
||||
self.assertEqual(player["token_hashes"], [self.game.hash_token(logged_in["token"])])
|
||||
self.assertEqual(len(loaded["presses"]), 2)
|
||||
self.assertEqual(len(loaded["presses"]), 4)
|
||||
|
||||
def test_press_eligibility_counts_two_distinct_other_players(self):
|
||||
self.game.state = self.game.new_game()
|
||||
self.assertTrue(self.game.press_eligibility("alice")["eligible"])
|
||||
self.game.state["presses"] = [
|
||||
{"player_id": "alice"},
|
||||
{"player_id": "bob"},
|
||||
{"player_id": "bob"},
|
||||
]
|
||||
gate = self.game.press_eligibility("alice")
|
||||
self.assertFalse(gate["eligible"])
|
||||
self.assertEqual(gate["other_players_pressed"], 1)
|
||||
self.assertEqual(gate["other_players_remaining"], 1)
|
||||
self.game.state["presses"].append({"player_id": "alice"})
|
||||
self.game.state["presses"].extend([
|
||||
{"player_id": "carol"},
|
||||
{"player_id": "dave"},
|
||||
])
|
||||
gate = self.game.press_eligibility("alice")
|
||||
self.assertTrue(gate["eligible"])
|
||||
self.assertEqual(gate["other_players_pressed"], 2)
|
||||
self.assertEqual(gate["other_players_remaining"], 0)
|
||||
self.game.state["presses"].append({"player_id": "alice"})
|
||||
gate = self.game.press_eligibility("alice")
|
||||
self.assertFalse(gate["eligible"])
|
||||
self.assertEqual(gate["other_players_pressed"], 0)
|
||||
self.assertEqual(gate["other_players_remaining"], 2)
|
||||
|
||||
def test_press_eligibility_in_state_is_private_to_authenticated_player(self):
|
||||
request = self.presence_client()
|
||||
status, public_state = request(path="/api/state")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertIsNone(public_state["press_eligibility"])
|
||||
status, own_state = request(path="/api/state", token="alice-token")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(own_state["press_eligibility"]["other_players_pressed"], 0)
|
||||
self.assertEqual(own_state["press_eligibility"]["other_players_remaining"], 2)
|
||||
|
||||
def test_oidc_requires_verified_email_and_required_service_group(self):
|
||||
from joserfc import jwt
|
||||
@@ -1278,6 +1337,26 @@ class PersistenceTests(unittest.TestCase):
|
||||
with self.assertRaises(ValueError):
|
||||
self.game.verify_oidc_id_token(denied_token, "nonce-value")
|
||||
|
||||
def test_server_refuses_to_start_without_dex_configuration(self):
|
||||
env = os.environ.copy()
|
||||
for key in (
|
||||
"BUTTON_OIDC_ISSUER", "BUTTON_OIDC_CLIENT_ID", "BUTTON_OIDC_CLIENT_SECRET",
|
||||
"BUTTON_OIDC_REDIRECT_URI", "BUTTON_OIDC_REQUIRED_GROUP",
|
||||
):
|
||||
env.pop(key, None)
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-c", "import runpy; runpy.run_path('server.py')"],
|
||||
cwd=Path(__file__).parent,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
timeout=15,
|
||||
)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("Dex OIDC is required", result.stderr)
|
||||
|
||||
def test_oidc_subject_mapping_preserves_existing_player_and_scores(self):
|
||||
state = self.sample_state()
|
||||
state["presses"] = [dict(state["presses"][0])]
|
||||
@@ -1314,6 +1393,44 @@ class PersistenceTests(unittest.TestCase):
|
||||
with self.assertRaises(PermissionError):
|
||||
self.game.issue_oidc_player({**claims, "sub": "outside-user", "preferred_username": "Outside"}, ["other-service"])
|
||||
|
||||
def test_oidc_player_name_prefers_display_name_and_falls_back_to_uid(self):
|
||||
self.game.state = self.game.new_game()
|
||||
self.game.token_index = {}
|
||||
with patch.object(self.game, "OIDC_SUBJECT_LINKS", {}), \
|
||||
patch.object(self.game, "OIDC_ADMIN_GROUP", ""), \
|
||||
patch.object(self.game, "OIDC_REQUIRED_GROUP", "the-button-users"), \
|
||||
patch.object(self.game, "OIDC_ADMIN_SUBJECTS", set()), \
|
||||
patch.object(self.game, "check_ended"), \
|
||||
patch.object(self.game, "save_state"):
|
||||
display_name, _ = self.game.issue_oidc_player({
|
||||
"sub": "display-name-sub", "displayName": "Display Name", "name": "OIDC Name", "uid": "displayuid",
|
||||
}, ["the-button-users"])
|
||||
oidc_name, _ = self.game.issue_oidc_player({
|
||||
"sub": "oidc-name-sub", "name": "Dex Name", "uid": "dexuid",
|
||||
}, ["the-button-users"])
|
||||
uid_name, _ = self.game.issue_oidc_player({
|
||||
"sub": "uid-sub", "uid": "regularuid", "preferred_username": "otheruid",
|
||||
}, ["the-button-users"])
|
||||
self.assertEqual(display_name, "Display Name")
|
||||
self.assertEqual(oidc_name, "Dex Name")
|
||||
self.assertEqual(uid_name, "regularuid")
|
||||
|
||||
def test_oidc_login_updates_linked_player_name_without_losing_history(self):
|
||||
self.game.state = self.sample_state()
|
||||
self.game.state["presses"] = [dict(self.game.state["presses"][0])]
|
||||
self.game.token_index = {}
|
||||
claims = {"sub": "dex-sub-bob", "name": "Robert", "uid": "bobuid", "email": "bob@example.test"}
|
||||
with patch.object(self.game, "OIDC_SUBJECT_LINKS", {"dex-sub-bob": "bob"}), \
|
||||
patch.object(self.game, "OIDC_ADMIN_GROUP", ""), \
|
||||
patch.object(self.game, "OIDC_REQUIRED_GROUP", "the-button-users"), \
|
||||
patch.object(self.game, "OIDC_ADMIN_SUBJECTS", set()), \
|
||||
patch.object(self.game, "check_ended"), \
|
||||
patch.object(self.game, "save_state"):
|
||||
name, _ = self.game.issue_oidc_player(claims, ["the-button-users"])
|
||||
self.assertEqual(name, "Robert")
|
||||
self.assertEqual(self.game.state["players"]["bob"]["name"], "Robert")
|
||||
self.assertEqual(len(self.game.state["presses"]), 1)
|
||||
|
||||
def test_oidc_id_token_signature_issuer_audience_nonce_and_email_claims(self):
|
||||
from joserfc import jwt
|
||||
from joserfc.jwk import RSAKey
|
||||
|
||||
Reference in New Issue
Block a user