LDAP
This commit is contained in:
+205
@@ -45,6 +45,7 @@ class PersistenceTests(unittest.TestCase):
|
||||
"alice": {
|
||||
"name": "Alice '); DROP TABLE players; --",
|
||||
"password_hash": "existing-password-hash",
|
||||
"oidc_sub": None,
|
||||
"token_hashes": ["first-session", "second-session"],
|
||||
"created_at": 123,
|
||||
"subscribed": True,
|
||||
@@ -53,6 +54,7 @@ class PersistenceTests(unittest.TestCase):
|
||||
"bob": {
|
||||
"name": "Bøb",
|
||||
"password_hash": None,
|
||||
"oidc_sub": None,
|
||||
"token_hashes": [],
|
||||
"created_at": 456,
|
||||
"subscribed": False,
|
||||
@@ -977,6 +979,15 @@ class PersistenceTests(unittest.TestCase):
|
||||
self.assertEqual(connection.execute("PRAGMA integrity_check").fetchone()[0], "ok")
|
||||
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
|
||||
|
||||
def test_legacy_players_receive_empty_oidc_subject(self):
|
||||
legacy = self.sample_state()
|
||||
for player in legacy["players"].values():
|
||||
player.pop("oidc_sub")
|
||||
self.write_legacy(legacy)
|
||||
loaded = self.game.load_state()
|
||||
self.assertEqual({player_id: player["oidc_sub"] for player_id, player in loaded["players"].items()}, {"alice": None, "bob": None})
|
||||
self.assertNotIn("oidc_sub", json.loads(self.game.DATA_FILE.read_text(encoding="utf-8"))["players"]["alice"])
|
||||
|
||||
def test_existing_database_ignores_legacy_even_if_invalid(self):
|
||||
self.write_legacy(self.sample_state())
|
||||
current = self.game.load_state()
|
||||
@@ -1232,6 +1243,200 @@ class PersistenceTests(unittest.TestCase):
|
||||
self.assertEqual(player["token_hashes"], [self.game.hash_token(logged_in["token"])])
|
||||
self.assertEqual(len(loaded["presses"]), 2)
|
||||
|
||||
def test_oidc_requires_verified_email_and_required_service_group(self):
|
||||
from joserfc import jwt
|
||||
from joserfc.jwk import RSAKey
|
||||
|
||||
signing_key = RSAKey.generate_key(2048, auto_kid=True)
|
||||
public_jwks = {"keys": [signing_key.as_dict(private=False)]}
|
||||
now = int(self.game.time.time())
|
||||
claims = {
|
||||
"iss": "https://dex.example.test",
|
||||
"sub": "stable-subject-123",
|
||||
"aud": "button-client",
|
||||
"iat": now,
|
||||
"exp": now + 300,
|
||||
"nonce": "nonce-value",
|
||||
"email": "alice@example.test",
|
||||
"email_verified": True,
|
||||
"preferred_username": "Alice",
|
||||
"groups": ["the-button-users"],
|
||||
}
|
||||
token = jwt.encode({"alg": "RS256", "kid": signing_key.kid}, claims, signing_key)
|
||||
metadata = {"_jwks": public_jwks}
|
||||
with patch.object(self.game, "OIDC_ISSUER", "https://dex.example.test"), \
|
||||
patch.object(self.game, "OIDC_CLIENT_ID", "button-client"), \
|
||||
patch.object(self.game, "OIDC_REQUIRED_GROUP", "the-button-users"), \
|
||||
patch.object(self.game, "get_oidc_metadata", return_value=metadata):
|
||||
verified, groups = self.game.verify_oidc_id_token(token, "nonce-value")
|
||||
self.assertEqual(verified["sub"], claims["sub"])
|
||||
self.assertIn("the-button-users", groups)
|
||||
with self.assertRaises(ValueError):
|
||||
self.game.verify_oidc_id_token(token, "wrong-nonce")
|
||||
denied = dict(claims, email_verified=False)
|
||||
denied_token = jwt.encode({"alg": "RS256", "kid": signing_key.kid}, denied, signing_key)
|
||||
with self.assertRaises(ValueError):
|
||||
self.game.verify_oidc_id_token(denied_token, "nonce-value")
|
||||
|
||||
def test_oidc_subject_mapping_preserves_existing_player_and_scores(self):
|
||||
state = self.sample_state()
|
||||
state["presses"] = [dict(state["presses"][0])]
|
||||
state["players"]["bob"]["oidc_sub"] = None
|
||||
self.game.state = state
|
||||
self.game.token_index = {}
|
||||
claims = {"sub": "dex-sub-bob", "preferred_username": "Bøb", "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_ADMIN_SUBJECTS", set()), \
|
||||
patch.object(self.game, "OIDC_REQUIRED_GROUP", "the-button-users"), \
|
||||
patch.object(self.game, "check_ended"), \
|
||||
patch.object(self.game, "save_state"):
|
||||
name, token = self.game.issue_oidc_player(claims, ["the-button-users"])
|
||||
self.assertEqual(name, "Bøb")
|
||||
self.assertEqual(self.game.state["players"]["bob"]["oidc_sub"], "dex-sub-bob")
|
||||
self.assertEqual(len(self.game.state["presses"]), 1)
|
||||
self.assertIn(self.game.hash_token(token), self.game.token_index)
|
||||
|
||||
def test_oidc_auto_registers_only_unique_player_names_and_authorized_groups(self):
|
||||
self.game.state = self.game.new_game()
|
||||
self.game.token_index = {}
|
||||
claims = {"sub": "dex-sub-new", "preferred_username": "Newcomer", "email": "new@example.test"}
|
||||
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"):
|
||||
name, token = self.game.issue_oidc_player(claims, ["the-button-users"])
|
||||
self.assertEqual(name, "Newcomer")
|
||||
with self.assertRaises(ValueError):
|
||||
self.game.issue_oidc_player({**claims, "sub": "different-sub"}, ["the-button-users"])
|
||||
with self.assertRaises(PermissionError):
|
||||
self.game.issue_oidc_player({**claims, "sub": "outside-user", "preferred_username": "Outside"}, ["other-service"])
|
||||
|
||||
def test_oidc_id_token_signature_issuer_audience_nonce_and_email_claims(self):
|
||||
from joserfc import jwt
|
||||
from joserfc.jwk import RSAKey
|
||||
|
||||
signing_key = RSAKey.generate_key(2048, auto_kid=True)
|
||||
metadata = {"_jwks": {"keys": [signing_key.as_dict(private=False)]}}
|
||||
now = int(self.game.time.time())
|
||||
claims = {
|
||||
"iss": "https://dex.example.test",
|
||||
"sub": "stable-oidc-id",
|
||||
"aud": "button-client",
|
||||
"iat": now,
|
||||
"exp": now + 300,
|
||||
"nonce": "expected-nonce",
|
||||
"email": "player@example.test",
|
||||
"email_verified": True,
|
||||
"preferred_username": "Player",
|
||||
"groups": ["the-button-users"],
|
||||
}
|
||||
signed = jwt.encode({"alg": "RS256", "kid": signing_key.kid}, claims, signing_key)
|
||||
config = (
|
||||
patch.object(self.game, "OIDC_ISSUER", "https://dex.example.test"),
|
||||
patch.object(self.game, "OIDC_CLIENT_ID", "button-client"),
|
||||
patch.object(self.game, "OIDC_REQUIRED_GROUP", "the-button-users"),
|
||||
patch.object(self.game, "get_oidc_metadata", return_value=metadata),
|
||||
)
|
||||
with config[0], config[1], config[2], config[3]:
|
||||
parsed, groups = self.game.verify_oidc_id_token(signed, "expected-nonce")
|
||||
self.assertEqual(parsed["sub"], claims["sub"])
|
||||
self.assertEqual(groups, ["the-button-users"])
|
||||
with self.assertRaises(ValueError):
|
||||
self.game.verify_oidc_id_token(signed, "wrong-nonce")
|
||||
invalid_email = dict(claims, email_verified=False)
|
||||
bad = jwt.encode({"alg": "RS256", "kid": signing_key.kid}, invalid_email, signing_key)
|
||||
with self.assertRaises(ValueError):
|
||||
self.game.verify_oidc_id_token(bad, "expected-nonce")
|
||||
unauthorized = dict(claims, groups=["other-service"])
|
||||
denied = jwt.encode({"alg": "RS256", "kid": signing_key.kid}, unauthorized, signing_key)
|
||||
with self.assertRaises(PermissionError):
|
||||
self.game.verify_oidc_id_token(denied, "expected-nonce")
|
||||
|
||||
def test_oidc_mode_rejects_local_password_registration_and_login(self):
|
||||
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)
|
||||
|
||||
def request(path, payload):
|
||||
connection = HTTPConnection(*httpd.server_address, timeout=5)
|
||||
try:
|
||||
headers = {"Content-Type": "application/json", "Origin": f"http://{httpd.server_address[0]}:{httpd.server_address[1]}"}
|
||||
connection.request("POST", path, json.dumps(payload), headers)
|
||||
response = connection.getresponse()
|
||||
return response.status
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
with patch.object(self.game, "OIDC_ENABLED", True):
|
||||
self.assertEqual(request("/api/register", {"name": "local", "password": "test-password"}), 403)
|
||||
self.assertEqual(request("/api/login", {"name": "local", "password": "test-password"}), 403)
|
||||
|
||||
def test_oidc_cookie_session_is_http_only_and_same_origin_protected(self):
|
||||
state = self.sample_state()
|
||||
state["players"]["bob"]["oidc_sub"] = "dex-sub-bob"
|
||||
self.game.state = state
|
||||
self.game.token_index = {}
|
||||
with self.game.lock, patch.object(self.game, "save_state"):
|
||||
bearer = self.game.issue_token("bob")
|
||||
|
||||
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)
|
||||
host, port = httpd.server_address
|
||||
origin = f"http://{host}:{port}"
|
||||
|
||||
def request(method, path, headers=None):
|
||||
connection = HTTPConnection(host, port, timeout=5)
|
||||
try:
|
||||
connection.request(method, path, "{}" if method == "POST" else None, headers or {})
|
||||
response = connection.getresponse()
|
||||
result = response.status, response.getheaders(), response.read()
|
||||
return result
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
with (
|
||||
patch.object(self.game, "OIDC_ENABLED", True),
|
||||
patch.object(self.game, "OIDC_COOKIE_SECURE", False),
|
||||
patch.object(self.game, "check_ended"),
|
||||
patch.object(self.game, "save_state"),
|
||||
):
|
||||
cookie = {"Cookie": f"button_session={bearer}"}
|
||||
self.assertEqual(request("GET", "/api/me", cookie)[0], 200)
|
||||
self.assertEqual(request("POST", "/api/logout", {**cookie, "Origin": "https://attacker.example"})[0], 403)
|
||||
status, headers, _ = request("POST", "/api/logout", {**cookie, "Origin": origin})
|
||||
self.assertEqual(status, 200)
|
||||
self.assertIn("Max-Age=0", dict(headers)["Set-Cookie"])
|
||||
self.assertEqual(request("GET", "/api/me", cookie)[0], 401)
|
||||
|
||||
def test_oidc_stable_subject_link_preserves_existing_game_history(self):
|
||||
state = self.sample_state()
|
||||
state["presses"] = [dict(state["presses"][0])]
|
||||
state["players"]["bob"]["oidc_sub"] = None
|
||||
self.game.state = state
|
||||
self.game.token_index = {}
|
||||
claims = {"sub": "dex-sub-bob", "preferred_username": "Bøb", "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, token = self.game.issue_oidc_player(claims, ["the-button-users"])
|
||||
self.assertEqual(name, "Bøb")
|
||||
self.assertEqual(self.game.state["players"]["bob"]["oidc_sub"], "dex-sub-bob")
|
||||
self.assertEqual(len(self.game.state["presses"]), 1)
|
||||
self.assertIn(self.game.hash_token(token), self.game.token_index)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user