~homeworkprod/byceps

e5e97995d86b2f8f8897dd15ebb007dd4208ab1d — Jochen Kupperschmidt 9 months ago c0a7a20
Introduce user login event, signal, and announcement
M byceps/announce/connections.py => byceps/announce/connections.py +5 -0
@@ 10,6 10,7 @@ Connect event signals to announcement handlers.

from typing import Optional

from ..events.auth import UserLoggedIn
from ..events.base import _BaseEvent
from ..events.board import (
    BoardPostingCreated,


@@ 53,6 54,7 @@ from ..events.user import (
    UserScreenNameChanged,
)
from ..events.user_badge import UserBadgeAwarded
from ..signals import auth as auth_signals
from ..signals import board as board_signals
from ..signals import news as news_signals
from ..signals import shop as shop_signals


@@ 64,6 66,7 @@ from ..signals import user_badge as user_badge_signals
from ..util.jobqueue import enqueue

from .handlers import (
    auth,
    board,
    news,
    shop_order,


@@ 77,6 80,7 @@ from .helpers import get_webhooks


EVENT_TYPES_TO_HANDLERS = {
    UserLoggedIn: auth.announce_user_logged_in,
    BoardTopicCreated: board.announce_board_topic_created,
    BoardTopicHidden: board.announce_board_topic_hidden,
    BoardTopicUnhidden: board.announce_board_topic_unhidden,


@@ 134,6 138,7 @@ def receive_signal(sender, *, event: Optional[_BaseEvent] = None) -> None:


SIGNALS = [
    auth_signals.user_logged_in,
    board_signals.topic_created,
    board_signals.topic_hidden,
    board_signals.topic_unhidden,

M byceps/announce/events.py => byceps/announce/events.py +2 -0
@@ 8,6 8,7 @@ Mapping between event types and names.
:License: Revised BSD (see `LICENSE` file for details)
"""

from ..events.auth import UserLoggedIn
from ..events.base import _BaseEvent
from ..events.board import (
    BoardPostingCreated,


@@ 54,6 55,7 @@ from ..events.user_badge import UserBadgeAwarded


EVENT_TYPES_TO_NAMES = {
    UserLoggedIn:                   'user-logged-in',
    BoardTopicCreated:              'board-topic-created',
    BoardTopicHidden:               'board-topic-hidden',
    BoardTopicLocked:               'board-topic-locked',

A byceps/announce/handlers/auth.py => byceps/announce/handlers/auth.py +24 -0
@@ 0,0 1,24 @@
"""
byceps.announce.handlers.auth
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Announce auth events.

:Copyright: 2006-2021 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from ...events.auth import UserLoggedIn
from ...services.webhooks.transfer.models import OutgoingWebhook

from ..helpers import call_webhook
from ..text_assembly import auth


def announce_user_logged_in(
    event: UserLoggedIn, webhook: OutgoingWebhook
) -> None:
    """Announce that a user has logged in."""
    text = auth.assemble_text_for_user_logged_in(event)

    call_webhook(webhook, text)

A byceps/announce/text_assembly/auth.py => byceps/announce/text_assembly/auth.py +27 -0
@@ 0,0 1,27 @@
"""
byceps.announce.text_assembly.auth
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Announce auth events.

:Copyright: 2006-2021 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from ...events.auth import UserLoggedIn
from ...services.site import service as site_service

from ._helpers import get_screen_name_or_fallback


def assemble_text_for_user_logged_in(event: UserLoggedIn) -> str:
    screen_name = get_screen_name_or_fallback(event.initiator_screen_name)

    site = None
    if event.site_id:
        site = site_service.find_site(event.site_id)

    if site:
        return f'{screen_name} hat sich auf Site "{site.title}" eingeloggt.'
    else:
        return f'{screen_name} hat sich eingeloggt.'

M byceps/blueprints/admin/authentication/login/views.py => byceps/blueprints/admin/authentication/login/views.py +7 -1
@@ 12,6 12,7 @@ from flask_babel import gettext
from .....services.authentication.exceptions import AuthenticationFailed
from .....services.authentication import service as authentication_service
from .....services.authentication.session import service as session_service
from .....signals import auth as auth_signals
from .....typing import UserID
from .....util.authorization import get_permissions_for_user
from .....util.framework.blueprint import create_blueprint


@@ 70,8 71,11 @@ def login():

    # Authorization succeeded.

    auth_token = session_service.log_in_user(user.id, request.remote_addr)
    auth_token, event = session_service.log_in_user(
        user.id, request.remote_addr
    )
    user_session.start(user.id, auth_token, permanent=permanent)

    flash_success(
        gettext(
            'Successfully logged in as %(screen_name)s.',


@@ 79,6 83,8 @@ def login():
        )
    )

    auth_signals.user_logged_in.send(None, event=event)


def _require_admin_access_permission(user_id: UserID) -> None:
    permissions = get_permissions_for_user(user_id)

M byceps/blueprints/site/authentication/login/views.py => byceps/blueprints/site/authentication/login/views.py +5 -1
@@ 21,6 21,7 @@ from .....services.site.transfer.models import Site
from .....services.verification_token import (
    service as verification_token_service,
)
from .....signals import auth as auth_signals
from .....typing import UserID
from .....util.framework.blueprint import create_blueprint
from .....util.framework.flash import flash_notice, flash_success


@@ 103,10 104,11 @@ def login():

    # Authorization succeeded.

    auth_token = session_service.log_in_user(
    auth_token, event = session_service.log_in_user(
        user.id, request.remote_addr, site_id=g.site_id
    )
    user_session.start(user.id, auth_token, permanent=permanent)

    flash_success(
        gettext(
            'Successfully logged in as %(screen_name)s.',


@@ 114,6 116,8 @@ def login():
        )
    )

    auth_signals.user_logged_in.send(None, event=event)

    return [('Location', url_for('dashboard.index'))]



A byceps/events/auth.py => byceps/events/auth.py +19 -0
@@ 0,0 1,19 @@
"""
byceps.events.auth
~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2021 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from dataclasses import dataclass
from typing import Optional

from ..services.site.transfer.models import SiteID

from .base import _BaseEvent


@dataclass(frozen=True)
class UserLoggedIn(_BaseEvent):
    site_id: Optional[SiteID]

M byceps/services/authentication/session/service.py => byceps/services/authentication/session/service.py +13 -4
@@ 8,14 8,15 @@ byceps.services.authentication.session.service

from datetime import datetime
from enum import Enum
from typing import Optional, Set
from typing import Optional, Set, Tuple
from uuid import UUID, uuid4

from ....database import db, insert_ignore_on_conflict, upsert
from ....events.auth import UserLoggedIn
from ....typing import UserID

from ...site.transfer.models import SiteID
from ...user import event_service as user_event_service
from ...user import event_service as user_event_service, service as user_service
from ...user.transfer.models import User

from ..exceptions import AuthenticationFailed


@@ 106,16 107,24 @@ def _is_token_valid_for_user(token: str, user_id: UserID) -> bool:

def log_in_user(
    user_id: UserID, ip_address: str, *, site_id: Optional[SiteID] = None
) -> str:
) -> Tuple[str, UserLoggedIn]:
    """Create a session token and record the log in."""
    session_token = get_session_token(user_id)

    occurred_at = datetime.utcnow()
    user = user_service.get_user(user_id)

    _create_login_event(user_id, occurred_at, ip_address, site_id=site_id)
    _record_recent_login(user_id, occurred_at)

    return session_token.token
    event = UserLoggedIn(
        occurred_at=occurred_at,
        initiator_id=user.id,
        initiator_screen_name=user.screen_name,
        site_id=site_id,
    )

    return session_token.token, event


def _create_login_event(

A byceps/signals/auth.py => byceps/signals/auth.py +15 -0
@@ 0,0 1,15 @@
"""
byceps.signals.auth
~~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2021 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from blinker import Namespace


auth_signals = Namespace()


user_logged_in = auth_signals.signal('user-logged-in')

M tests/integration/announce/irc/conftest.py => tests/integration/announce/irc/conftest.py +2 -1
@@ 40,10 40,11 @@ def webhook_settings():
                'user-account-deleted',
                'user-account-suspended',
                'user-account-unsuspended',
                'user-badge-awarded',
                'user-details-updated',
                'user-email-address-invalidated',
                'user-screen-name-changed',
                'user-badge-awarded',
                'user-logged-in',
            ]
        ),
        # public

A tests/integration/announce/irc/test_auth.py => tests/integration/announce/irc/test_auth.py +49 -0
@@ 0,0 1,49 @@
"""
:Copyright: 2006-2021 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

import pytest

import byceps.announce.connections  # Connect signal handlers.
from byceps.services.authentication.session import service as session_service
from byceps.signals import auth as auth_signals

from .helpers import assert_submitted_data, CHANNEL_ORGA_LOG, mocked_irc_bot


EXPECTED_CHANNEL = CHANNEL_ORGA_LOG


def test_user_logged_in_into_admin_app_announced(app, user):
    expected_text = 'Logvogel hat sich eingeloggt.'

    _, event = session_service.log_in_user(user.id, '10.10.23.42')

    with mocked_irc_bot() as mock:
        auth_signals.user_logged_in.send(None, event=event)

    assert_submitted_data(mock, EXPECTED_CHANNEL, expected_text)


def test_user_logged_in_into_site_app_announced(app, site, user):
    expected_text = (
        'Logvogel hat sich auf Site "ACMECon 2014 website" eingeloggt.'
    )

    _, event = session_service.log_in_user(
        user.id, '10.10.23.42', site_id=site.id
    )

    with mocked_irc_bot() as mock:
        auth_signals.user_logged_in.send(None, event=event)

    assert_submitted_data(mock, EXPECTED_CHANNEL, expected_text)


# helpers


@pytest.fixture(scope='module')
def user(make_user):
    return make_user('Logvogel')