fix(auth): garder l identite Discord du cote verifie

This commit is contained in:
GGThed
2026-08-16 23:36:30 -04:00
parent 437b229c82
commit 9647003c3f
16 changed files with 833 additions and 340 deletions
+222
View File
@@ -0,0 +1,222 @@
"""The Discord snowflake is a verified identity, not profile text.
SEC-AUTH-005. Discord OAuth used to put its result in two hidden inputs;
registration then trusted those client-controlled values, and edit_profile
let the account owner replace the snowflake later. The bot uses that value to
route private messages and authorize reaction-driven writes.
"""
import time
from urllib.parse import parse_qs, urlparse
from app.routes import auth as auth_module
from app.routes.auth import (
MIN_REGISTRATION_SECONDS,
REGISTRATION_ISSUED_KEY,
)
FORM = {
'username': 'brandnew',
'email': '[email protected]',
'password': 'Password123',
'confirm_password': 'Password123',
'full_name': 'Brand New',
}
def _allow_registration(client):
with client.session_transaction() as session:
session[REGISTRATION_ISSUED_KEY] = time.time() - MIN_REGISTRATION_SECONDS - 1
def _user(app, username='brandnew'):
from app.models import User
with app.app_context():
return User.query.filter_by(username=username).first()
class _DiscordResponse:
def __init__(self, payload):
self.payload = payload
def raise_for_status(self):
return None
def json(self):
return self.payload
def _complete_profile_oauth(client, monkeypatch, discord_user_id):
monkeypatch.setattr(auth_module, 'DISCORD_CLIENT_ID', 'client-id')
monkeypatch.setattr(auth_module, 'DISCORD_CLIENT_SECRET', 'client-secret')
monkeypatch.setattr(
auth_module,
'DISCORD_REDIRECT_URI',
'https://example.test/auth/discord/callback',
)
start = client.get('/auth/discord/login')
state = parse_qs(urlparse(start.headers['Location']).query)['state'][0]
monkeypatch.setattr(
auth_module.requests,
'post',
lambda *args, **kwargs: _DiscordResponse({'access_token': 'token'}),
)
monkeypatch.setattr(
auth_module.requests,
'get',
lambda *args, **kwargs: _DiscordResponse(
{'id': discord_user_id, 'username': 'verified-user'}
),
)
return client.get(f'/auth/discord/callback?code=code&state={state}')
class TestRegistrationIdentity:
def test_a_posted_snowflake_without_oauth_is_ignored(self, app, client):
_allow_registration(client)
client.post(
'/auth/register',
data=dict(FORM, discord_user_id='111111111111111111'),
)
assert _user(app).discord_user_id is None
def test_the_server_side_oauth_identity_wins_over_the_form(self, app, client):
verified = '222222222222222222'
_allow_registration(client)
with client.session_transaction() as session:
session['discord_oauth'] = {
'id': verified,
'username': 'verified-user',
}
client.post(
'/auth/register',
data=dict(
FORM,
discord_user_id='111111111111111111',
discord_username='forged-user',
),
)
created = _user(app)
assert created.discord_user_id == verified
assert created.discord_username == 'verified-user'
def test_an_oauth_identity_already_in_use_is_refused(self, app, client, make_user):
verified = '222222222222222222'
make_user('player', discord_user_id=verified)
_allow_registration(client)
with client.session_transaction() as session:
session['discord_oauth'] = {
'id': verified,
'username': 'verified-user',
}
client.post('/auth/register', data=FORM)
assert _user(app) is None
def test_the_registration_page_has_no_client_identity_field(self, client):
with client.session_transaction() as session:
session['discord_oauth'] = {
'id': '222222222222222222',
'username': 'verified-user',
}
page = client.get('/auth/register').get_data(as_text=True)
assert 'name="discord_user_id"' not in page
class TestProfileIdentity:
def test_a_profile_post_cannot_replace_the_verified_snowflake(self, app, client, as_role):
original = '222222222222222222'
user_id = as_role('player', discord_user_id=original)
client.post(
'/users/profile/edit',
data={
'username': 'player1',
'full_name': 'Player 1',
'email': '[email protected]',
'discord_user_id': '111111111111111111',
},
)
from app.models import User
with app.app_context():
assert (
app.extensions['sqlalchemy'].session.get(User, user_id).discord_user_id == original
)
def test_the_profile_form_does_not_offer_the_snowflake(self, client, as_role):
as_role('player', discord_user_id='222222222222222222')
page = client.get('/users/profile/edit').get_data(as_text=True)
assert 'name="discord_user_id"' not in page
assert '/auth/discord/login' in page
def test_a_signed_in_user_can_relink_only_through_oauth(
self, app, client, as_role, monkeypatch
):
user_id = as_role('player', discord_user_id='111111111111111111')
response = _complete_profile_oauth(
client,
monkeypatch,
discord_user_id='222222222222222222',
)
from app.models import User
assert '/users/profile/edit' in response.headers['Location']
with app.app_context():
user = app.extensions['sqlalchemy'].session.get(User, user_id)
assert user.discord_user_id == '222222222222222222'
assert user.discord_username == 'verified-user'
def test_profile_oauth_refuses_an_identity_owned_by_someone_else(
self, app, client, as_role, make_user, monkeypatch
):
taken = '222222222222222222'
make_user('player', discord_user_id=taken)
user_id = as_role('player', discord_user_id='111111111111111111')
_complete_profile_oauth(client, monkeypatch, discord_user_id=taken)
from app.models import User
with app.app_context():
assert (
app.extensions['sqlalchemy'].session.get(User, user_id).discord_user_id
== '111111111111111111'
)
class TestAdministrativeFallback:
def test_an_admin_cannot_assign_a_snowflake_twice(self, app, client, as_role, make_user):
taken = '222222222222222222'
make_user('player', discord_user_id=taken)
target_id = make_user('player')
as_role('admin')
client.post(
f'/users/{target_id}/edit',
data={
'full_name': 'Target Player',
'email': '[email protected]',
'role': 'player',
'discord_user_id': taken,
},
)
from app.models import User
with app.app_context():
assert app.extensions['sqlalchemy'].session.get(User, target_id).discord_user_id is None
+61
View File
@@ -202,6 +202,67 @@ class TestCatalogueIntegrity:
f'{len(untranslated)} untranslated string(s) in {locale}: {untranslated[:5]}'
)
# Babel keeps its guessed translation when it marks an entry fuzzy,
# but gettext deliberately ignores that guess at runtime. Merely
# checking m.string therefore let five English fallbacks through after
# the branding merge, including a dangerously wrong French label.
fuzzy = [m.id for m in catalog if m.id and 'fuzzy' in m.flags]
assert not fuzzy, f'{len(fuzzy)} fuzzy string(s) in {locale}: {fuzzy[:5]}'
def test_the_catalogue_contains_every_message_in_the_source(self, tmp_path):
"""A translated PO can still be stale.
The previous guard only inspected entries already in the catalogue.
A new `_()` in Python or Jinja therefore stayed English without any
failure until somebody happened to run extraction by hand.
"""
import os
import subprocess
import sys
from babel.messages.pofile import read_po
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
extracted_path = tmp_path / 'messages.pot'
subprocess.run(
[
sys.executable,
'-m',
'babel.messages.frontend',
'extract',
'-F',
'babel.cfg',
'-k',
'_l',
'-o',
str(extracted_path),
'.',
],
cwd=root,
check=True,
capture_output=True,
text=True,
)
with open(extracted_path, encoding='utf-8') as handle:
source_ids = {message.id for message in read_po(handle) if message.id}
for locale in SUPPORTED_LOCALES:
path = os.path.join(
root,
'app',
'translations',
locale,
'LC_MESSAGES',
'messages.po',
)
with open(path, encoding='utf-8') as handle:
catalogue_ids = {message.id for message in read_po(handle) if message.id}
missing = source_ids - catalogue_ids
assert not missing, (
f'{len(missing)} source string(s) absent from {locale}: '
f'{sorted(missing, key=str)[:5]}'
)
class TestSelectorUnit:
def test_select_locale_returns_a_supported_code(self, app):
+33
View File
@@ -257,3 +257,36 @@ class TestSeedAccountCheck:
output = capsys.readouterr().out
assert 'password has been changed' in output
assert 'PASSWORD IS STILL' not in output
class TestDiscordIdentityCheck:
def test_it_finds_identity_collisions_before_the_unique_migration(self, live_db, app, capsys):
engine, _tamper = live_db
path = str(engine.url).replace('sqlite:///', '')
connection = sqlite3.connect(path)
for username in ('alice', 'bob'):
connection.execute(
'INSERT INTO users '
'(username, password_hash, role, full_name, email, '
'is_active_account, discord_user_id) '
"VALUES (?, 'hash', 'player', ?, ?, 1, '222222222222222222')",
(username, username.title(), f'{username}@example.test'),
)
connection.commit()
connection.close()
with app.app_context():
result = main(['--url', str(engine.url), '--check-discord-identities'])
output = capsys.readouterr().out
assert result == 1
assert '222222222222222222: 2 accounts (alice, bob)' in output
def test_a_clean_identity_set_does_not_change_the_exit_code(self, live_db, app, capsys):
engine, _tamper = live_db
with app.app_context():
result = main(['--url', str(engine.url), '--check-discord-identities'])
assert result == 0
assert 'No Discord identity is shared' in capsys.readouterr().out