diff --git a/app/logging_config.py b/app/logging_config.py index 163d798..e460bed 100644 --- a/app/logging_config.py +++ b/app/logging_config.py @@ -172,8 +172,40 @@ def configure_logging(app): # Module-level auth logger factory def get_auth_logger(): """Get the authentication event logger. - + Returns: logging.Logger: Logger for authentication events. """ - return logging.getLogger('team_tryouts.auth') \ No newline at end of file + return logging.getLogger('team_tryouts.auth') + + +def log_auth_event(event, **fields): + """Record a security-relevant event to auth.log. + + The handler, its rotation and its redaction filter were configured from + the start, but get_auth_logger was never imported anywhere: auth.log was + created and stayed empty. No login, failure, lockout, role change or + account deletion left any trace. + + Fields are emitted as `key=value` pairs, ordered, so the file stays + greppable without pulling in a JSON logging dependency. + + Note on `ip`: it is taken from request.remote_addr, which reflects + X-Forwarded-For. As long as Waitress runs with trusted_proxy='*' + (SEC-WEB-002), that value is attacker-controlled and must be read as an + indication rather than as evidence. + + Args: + event: Dotted event name, e.g. 'login.success'. + **fields: Additional context. Never pass a secret: values are + recorded verbatim apart from the redaction filter's patterns. + """ + from flask import has_request_context, request + + parts = ['event=%s' % event] + if has_request_context(): + parts.append('ip=%s' % request.remote_addr) + parts.append('path=%s' % request.path) + parts.extend('%s=%s' % (key, value) for key, value in fields.items()) + + get_auth_logger().info(' '.join(parts)) \ No newline at end of file diff --git a/app/routes/users.py b/app/routes/users.py index 15c4e78..147d8b3 100644 --- a/app/routes/users.py +++ b/app/routes/users.py @@ -24,6 +24,7 @@ from app.validators import ( CreateUserSchema, EditUserSchema, EditProfileSchema, UploadContractSchema, ) +from app.logging_config import log_auth_event import requests ALLOWED_CONTRACT_EXTENSIONS = {'pdf'} @@ -128,6 +129,11 @@ def edit_user(user_id): user = User.query.get_or_404(user_id) if request.method == 'POST': + # Captured up front, on purpose. A role change below calls + # db.session.remove(), which detaches current_user from the session: + # any later attribute access on it raises DetachedInstanceError. + actor_name, actor_id = current_user.username, current_user.id + def _rerender(): return render_template('pages/edit_user.html', user=user, roles=USER_TYPES, esport_games=ESPORT_GAMES, game_platforms=GAME_PLATFORMS, @@ -165,6 +171,10 @@ def edit_user(user_id): # change invalidates the identity map for this instance and anything # that references it via relationships. if user.role != role: + log_auth_event('account.role_changed', + actor=actor_name, actor_id=actor_id, + target=user.username, target_id=user.id, + previous_role=user.role, new_role=role) user_id_local = user.id db.session.execute( db.text("UPDATE users SET role = :role WHERE id = :id"), @@ -190,8 +200,14 @@ def edit_user(user_id): password = validated.get('password') if password: user.password_hash = hash_password(password) + log_auth_event('account.password_reset_by_admin', + actor=actor_name, actor_id=actor_id, + target=user.username, target_id=user.id) db.session.commit() + log_auth_event('account.updated', + actor=actor_name, actor_id=actor_id, + target=user.username, target_id=user.id, active=is_active) flash(f'User {user.username} updated successfully!', 'success') return redirect(url_for('users.list_users')) @@ -242,9 +258,13 @@ def delete_user(user_id): OrgTeam.query.filter_by(created_by=user_id).update({'created_by': current_user.id}) Contract.query.filter_by(uploaded_by_id=user_id).update({'uploaded_by_id': current_user.id}) + deleted_username, deleted_role = user.username, user.role db.session.delete(user) db.session.commit() - flash(f'User {user.username} has been removed.', 'success') + log_auth_event('account.deleted', + actor=current_user.username, actor_id=current_user.id, + target=deleted_username, target_id=user_id, role=deleted_role) + flash(f'User {deleted_username} has been removed.', 'success') return redirect(url_for('users.list_users')) @@ -289,6 +309,9 @@ def create_user(): ) db.session.add(user) db.session.commit() + log_auth_event('account.created_by_admin', + actor=current_user.username, actor_id=current_user.id, + target=user.username, target_id=user.id, role=role) flash(f'User {full_name} created as {role}!', 'success') return redirect(url_for('users.list_users')) @@ -373,6 +396,8 @@ def edit_profile(): password = validated.get('password') if password: current_user.password_hash = hash_password(password) + log_auth_event('account.password_changed', + username=current_user.username, user_id=current_user.id) db.session.commit() flash('Profile updated successfully!', 'success') diff --git a/tests/test_audit_logging.py b/tests/test_audit_logging.py new file mode 100644 index 0000000..a403f95 --- /dev/null +++ b/tests/test_audit_logging.py @@ -0,0 +1,200 @@ +"""Security event logging. + +OBS-001. logging_config.py configured a rotating auth.log handler, a named +'team_tryouts.auth' logger and a redaction filter — and get_auth_logger was +never imported anywhere. The file was created and stayed empty: no login, +no failure, no lockout, no role change, no deletion left a trace. After a +suspected compromise there was nothing to look at. +""" + +import logging + +import pytest + +from app.extensions import db +from app.models import User + + +@pytest.fixture +def auth_log(caplog): + """Capture records emitted on the auth logger. + + The logger sets propagate = False so its records never reach the root + handler caplog installs; the handler is attached directly instead. + """ + logger = logging.getLogger('team_tryouts.auth') + previous = logger.level + logger.setLevel(logging.INFO) + logger.addHandler(caplog.handler) + try: + yield caplog + finally: + logger.removeHandler(caplog.handler) + logger.setLevel(previous) + + +def _events(caplog): + return [r.getMessage() for r in caplog.records] + + +def _has_event(caplog, name): + return any(f'event={name}' in message for message in _events(caplog)) + + +class TestLoginEvents: + def test_a_successful_login_is_recorded(self, client, as_role, auth_log): + as_role('player') + assert _has_event(auth_log, 'login.success') + + def test_the_record_carries_who_and_from_where(self, app, client, as_role, auth_log): + user_id = as_role('coach') + with app.app_context(): + username = db.session.get(User, user_id).username + + message = next(m for m in _events(auth_log) if 'event=login.success' in m) + assert f'username={username}' in message + assert f'user_id={user_id}' in message + assert 'role=coach' in message + assert 'ip=' in message + + def test_a_wrong_password_is_recorded(self, app, client, make_user, login, auth_log): + user_id = make_user('player') + with app.app_context(): + username = db.session.get(User, user_id).username + + login(username, password='WrongPassword1') + + assert _has_event(auth_log, 'login.failure') + + def test_an_unknown_username_is_recorded_separately(self, client, login, auth_log): + login('no-such-account', password='WrongPassword1') + assert _has_event(auth_log, 'login.failure.unknown_user') + + def test_a_lockout_is_recorded(self, app, client, make_user, login, auth_log): + user_id = make_user('player') + with app.app_context(): + username = db.session.get(User, user_id).username + + for _ in range(5): + login(username, password='WrongPassword1') + + assert _has_event(auth_log, 'account.locked') + + def test_a_deactivated_account_attempt_is_recorded( + self, app, client, make_user, login, auth_log + ): + user_id = make_user('player') + with app.app_context(): + user = db.session.get(User, user_id) + user.is_active_account = False + username = user.username + db.session.commit() + + login(username) + + assert _has_event(auth_log, 'login.rejected.deactivated') + + def test_logout_is_recorded(self, client, as_role, auth_log): + as_role('player') + client.get('/auth/logout') + assert _has_event(auth_log, 'logout') + + +class TestAdministrativeEvents: + def test_account_creation_is_recorded(self, client, as_role, auth_log): + as_role('admin') + + client.post('/users/create', data={ + 'username': 'newcoach', 'email': 'newcoach@example.test', + 'password': 'Password123', 'full_name': 'New Coach', 'role': 'coach', + }, follow_redirects=True) + + assert _has_event(auth_log, 'account.created_by_admin') + + def test_a_role_change_records_both_roles(self, app, client, as_role, make_user, auth_log): + target_id = make_user('player') + as_role('admin') + + client.post(f'/users/{target_id}/edit', data={ + 'full_name': 'Promoted', 'email': 'promoted@example.test', + 'role': 'coach', 'is_active_account': 'on', + }, follow_redirects=True) + + message = next(m for m in _events(auth_log) if 'event=account.role_changed' in m) + assert 'previous_role=player' in message + assert 'new_role=coach' in message + assert f'target_id={target_id}' in message + + def test_account_deletion_is_recorded(self, client, as_role, make_user, auth_log): + victim_id = make_user('player') + as_role('admin') + + client.post(f'/users/{victim_id}/delete', follow_redirects=True) + + assert _has_event(auth_log, 'account.deleted') + + def test_a_self_service_password_change_is_recorded( + self, app, client, as_role, auth_log + ): + user_id = as_role('player') + with app.app_context(): + username = db.session.get(User, user_id).username + + client.post('/users/profile/edit', data={ + 'username': username, 'full_name': 'Same Name', + 'email': 'same@example.test', 'password': 'BrandNew123', + }, follow_redirects=True) + + assert _has_event(auth_log, 'account.password_changed') + + +class TestRedaction: + """The filter is a security control; it needs its own test. + + It used to inspect record.msg only, while every call site logs with %s + placeholders — so the sensitive value sat in record.args and was never + examined. + """ + + def test_a_secret_passed_through_args_is_redacted(self): + from app.logging_config import SensitiveDataFilter + + record = logging.LogRecord( + 'test', logging.ERROR, 'x.py', 1, + 'Database failure: %s', ('password=hunter2 host=db.internal',), None, + ) + SensitiveDataFilter().filter(record) + + rendered = record.getMessage() + assert 'hunter2' not in rendered + assert '[REDACTED]' in rendered + + def test_a_bearer_token_is_redacted(self): + from app.logging_config import SensitiveDataFilter + + record = logging.LogRecord( + 'test', logging.INFO, 'x.py', 1, + 'Calling Discord with %s', ('Bearer abcdef123456',), None, + ) + SensitiveDataFilter().filter(record) + + assert 'abcdef123456' not in record.getMessage() + + def test_an_ordinary_message_is_left_alone(self): + from app.logging_config import SensitiveDataFilter + + record = logging.LogRecord( + 'test', logging.INFO, 'x.py', 1, + 'event=login.success username=%s', ('alice',), None, + ) + SensitiveDataFilter().filter(record) + + assert record.getMessage() == 'event=login.success username=alice' + + def test_a_malformed_format_string_does_not_drop_the_record(self): + from app.logging_config import SensitiveDataFilter + + record = logging.LogRecord( + 'test', logging.INFO, 'x.py', 1, 'broken %d', ('not-a-number',), None, + ) + assert SensitiveDataFilter().filter(record) is True