byceps.blueprints.authentication.views._get_site() - Code Metrics - Inspection of "Promote login enabled site setting to site propert..." - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 127990...e34625 )

by Jochen

created 2019-12-09 03:16 UTC

byceps.blueprints.authentication.views._get_site() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	0
dl	0
loc	2
rs	10
c	0
b	0
f	0

"""
byceps.blueprints.authentication.views
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2019 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""

from flask import abort, g, request, url_for

from ...config import get_site_mode
from ...services.authentication.exceptions import AuthenticationFailed
from ...services.authentication import service as authentication_service
from ...services.authentication.password import service as password_service
from ...services.authentication.password import (
    reset_service as password_reset_service,
)
from ...services.authentication.session import service as session_service
from ...services.consent import consent_service
from ...services.email import service as email_service
from ...services.site import service as site_service
from ...services.terms import version_service as terms_version_service
from ...services.user import service as user_service
from ...services.verification_token import service as verification_token_service
from ...util.framework.blueprint import create_blueprint
from ...util.framework.flash import flash_error, flash_notice, flash_success
from ...util.framework.templating import templated
from ...util.views import redirect_to, respond_no_content

from ..admin.core.authorization import AdminPermission
from ..user.creation.views import _find_privacy_policy_consent_subject_id

from .forms import (
    LoginForm,
    RequestPasswordResetForm,
    ResetPasswordForm,
    UpdatePasswordForm,
)
from . import service, session as user_session


blueprint = create_blueprint('authentication', __name__)


# -------------------------------------------------------------------- #
# log in/out


@blueprint.route('/login')
@templated
def login_form():
    """Show login form."""
    logged_in = g.current_user.is_active
    if logged_in:
        flash_notice(
            f'Du bist bereits als Benutzer "{g.current_user.screen_name}" '
            'angemeldet.'
        )

    in_admin_mode = get_site_mode().is_admin()

    if not _is_login_enabled(in_admin_mode):
        return {
            'login_enabled': False,
        }

    form = LoginForm()
    user_account_creation_enabled = _is_user_account_creation_enabled(in_admin_mode)

    return {
        'login_enabled': True,
        'logged_in': logged_in,
        'form': form,
        'user_account_creation_enabled': user_account_creation_enabled,
    }


@blueprint.route('/login', methods=['POST'])
@respond_no_content
def login():
    """Allow the user to authenticate with e-mail address and password."""
    if g.current_user.is_active:
        return

    in_admin_mode = get_site_mode().is_admin()

    if not _is_login_enabled(in_admin_mode):
        abort(403, 'Log in to this site is generally disabled.')

    form = LoginForm(request.form)

    screen_name = form.screen_name.data.strip()
    password = form.password.data
    permanent = form.permanent.data
    if not all([screen_name, password]):
        abort(403)

    try:
        user = authentication_service.authenticate(screen_name, password)
    except AuthenticationFailed:
        abort(403)

    if in_admin_mode:
        permissions = service.get_permissions_for_user(user.id)
        if AdminPermission.access not in permissions:
            # The user lacks the admin access permission which is required
            # to enter the admin area.
            abort(403)

    if not in_admin_mode:
        required_consent_subject_ids = _get_required_consent_subject_ids()
        if _is_consent_required(user.id, required_consent_subject_ids):
            verification_token = verification_token_service.create_for_terms_consent(
                user.id
            )

            consent_form_url = url_for(
                'consent.consent_form', token=verification_token.token
            )

            return [('Location', consent_form_url)]

    # Authorization succeeded.

    session_token = session_service.get_session_token(user.id)

    service.create_login_event(user.id, request.remote_addr)

    user_session.start(user.id, session_token.token, permanent=permanent)
    flash_success(f'Erfolgreich eingeloggt als {user.screen_name}.')


def _get_required_consent_subject_ids():
    subject_ids = []

    terms_version = terms_version_service.find_current_version_for_brand(
        g.brand_id
    )
    if terms_version:
        subject_ids.append(terms_version.consent_subject_id)

    privacy_policy_consent_subject_id = (
        _find_privacy_policy_consent_subject_id()
    )

    if privacy_policy_consent_subject_id:
        subject_ids.append(privacy_policy_consent_subject_id)

    return subject_ids


def _is_consent_required(user_id, subject_ids):
    for subject_id in subject_ids:
        if not consent_service.has_user_consented_to_subject(
            user_id, subject_id
        ):
            return True

    return False


@blueprint.route('/logout', methods=['POST'])
@respond_no_content
def logout():
    """Log out user by deleting the corresponding cookie."""
    user_session.end()
    flash_success('Erfolgreich ausgeloggt.')


# -------------------------------------------------------------------- #
# password update


@blueprint.route('/password/update')
@templated
def password_update_form(erroneous_form=None):
    """Show a form to update the current user's password."""
    _get_current_user_or_404()

    form = erroneous_form if erroneous_form else UpdatePasswordForm()

    return {'form': form}


@blueprint.route('/password', methods=['POST'])
def password_update():
    """Update the current user's password."""
    user = _get_current_user_or_404()

    form = UpdatePasswordForm(request.form)

    if not form.validate():
        return password_update_form(form)

    password = form.new_password.data

    password_service.update_password_hash(user.id, password, user.id)

    flash_success('Dein Passwort wurde geändert. Bitte melde dich erneut an.')
    return redirect_to('.login_form')


# -------------------------------------------------------------------- #
# password reset


@blueprint.route('/password/reset/request')
@templated
def request_password_reset_form(erroneous_form=None):
    """Show a form to request a password reset."""
    form = erroneous_form if erroneous_form else RequestPasswordResetForm()

    return {'form': form}


@blueprint.route('/password/reset/request', methods=['POST'])
def request_password_reset():
    """Request a password reset."""
    form = RequestPasswordResetForm(request.form)
    if not form.validate():
        return request_password_reset_form(form)

    screen_name = form.screen_name.data.strip()
    user = user_service.find_user_by_screen_name(
        screen_name, case_insensitive=True
    )

    if user is None:
        flash_error(f'Der Benutzername "{screen_name}" ist unbekannt.')
        return request_password_reset_form(form)

    if not user.email_address_verified:
        flash_error(
            f'Die E-Mail-Adresse für das Benutzerkonto "{screen_name}" '
            'wurde noch nicht bestätigt.'
        )
        return redirect_to('user_email_address.request_confirmation_email')

    sender = None
    if get_site_mode().is_public():
        site = site_service.get_site(g.site_id)
        email_config = email_service.get_config(site.email_config_id)
        sender = email_config.sender

    password_reset_service.prepare_password_reset(
        user, request.url_root, sender=sender
    )

    flash_success(
        'Ein Link zum Setzen eines neuen Passworts '
        f'für den Benutzernamen "{user.screen_name}" '
        'wurde an die hinterlegte E-Mail-Adresse versendet.'
    )
    return request_password_reset_form()


@blueprint.route('/password/reset/token/<token>')
@templated
def password_reset_form(token, erroneous_form=None):
    """Show a form to reset the current user's password."""
    verification_token = verification_token_service.find_for_password_reset_by_token(
        token
    )

    _verify_password_reset_token(verification_token)

    form = erroneous_form if erroneous_form else ResetPasswordForm()

    return {
        'form': form,
        'token': token,
    }


@blueprint.route('/password/reset/token/<token>', methods=['POST'])
def password_reset(token):
    """Reset the current user's password."""
    verification_token = verification_token_service.find_for_password_reset_by_token(
        token
    )

    _verify_password_reset_token(verification_token)

    form = ResetPasswordForm(request.form)
    if not form.validate():
        return password_reset_form(token, form)

    password = form.new_password.data

    password_reset_service.reset_password(verification_token, password)

    flash_success('Das Passwort wurde geändert.')
    return redirect_to('.login_form')


def _verify_password_reset_token(verification_token):
    if verification_token is None or verification_token.is_expired:
        flash_error(
            'Es wurde kein gültiges Token angegeben. '
            'Ein Token ist nur 24 Stunden lang gültig.'
        )
        abort(404)


# -------------------------------------------------------------------- #
# helpers


def _is_user_account_creation_enabled(in_admin_mode):
    if in_admin_mode:
        return False

    site = _get_site()
    return site.user_account_creation_enabled


def _is_login_enabled(in_admin_mode):
    if in_admin_mode:
        return True

    site = _get_site()
    return site.login_enabled


def _get_site():
    return site_service.get_site(g.site_id)


def _get_current_user_or_404():
    user = g.current_user
    if not user.is_active:
        abort(404)

    return user


1			"""
2			byceps.blueprints.authentication.views
3			~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5			:Copyright: 2006-2019 Jochen Kupperschmidt
6			:License: Modified BSD, see LICENSE for details.
7			"""
8
9			from flask import abort, g, request, url_for
10
11			from ...config import get_site_mode
12			from ...services.authentication.exceptions import AuthenticationFailed
13			from ...services.authentication import service as authentication_service
14			from ...services.authentication.password import service as password_service
15			from ...services.authentication.password import (
16			reset_service as password_reset_service,
17			)
18			from ...services.authentication.session import service as session_service
19			from ...services.consent import consent_service
20			from ...services.email import service as email_service
21			from ...services.site import service as site_service
22			from ...services.terms import version_service as terms_version_service
23			from ...services.user import service as user_service
24			from ...services.verification_token import service as verification_token_service
25			from ...util.framework.blueprint import create_blueprint
26			from ...util.framework.flash import flash_error, flash_notice, flash_success
27			from ...util.framework.templating import templated
28			from ...util.views import redirect_to, respond_no_content
29
30			from ..admin.core.authorization import AdminPermission
31			from ..user.creation.views import _find_privacy_policy_consent_subject_id
32
33			from .forms import (
34			LoginForm,
35			RequestPasswordResetForm,
36			ResetPasswordForm,
37			UpdatePasswordForm,
38			)
39			from . import service, session as user_session
40
41
42			blueprint = create_blueprint('authentication', __name__)
43
44
45			# -------------------------------------------------------------------- #
46			# log in/out
47
48
49			@blueprint.route('/login')
50			@templated
51			def login_form():
52			"""Show login form."""
53			logged_in = g.current_user.is_active
54			if logged_in:
55			flash_notice(
56			f'Du bist bereits als Benutzer "{g.current_user.screen_name}" '
57			'angemeldet.'
58			)
59
60			in_admin_mode = get_site_mode().is_admin()
61
62			if not _is_login_enabled(in_admin_mode):
63			return {
64			'login_enabled': False,
65			}
66
67			form = LoginForm()
68			user_account_creation_enabled = _is_user_account_creation_enabled(in_admin_mode)
69
70			return {
71			'login_enabled': True,
72			'logged_in': logged_in,
73			'form': form,
74			'user_account_creation_enabled': user_account_creation_enabled,
75			}
76
77
78			@blueprint.route('/login', methods=['POST'])
79			@respond_no_content
80			def login():
81			"""Allow the user to authenticate with e-mail address and password."""
82			if g.current_user.is_active:
83			return
84
85			in_admin_mode = get_site_mode().is_admin()
86
87			if not _is_login_enabled(in_admin_mode):
88			abort(403, 'Log in to this site is generally disabled.')
89
90			form = LoginForm(request.form)
91
92			screen_name = form.screen_name.data.strip()
93			password = form.password.data
94			permanent = form.permanent.data
95			if not all([screen_name, password]):
96			abort(403)
97
98			try:
99			user = authentication_service.authenticate(screen_name, password)
100			except AuthenticationFailed:
101			abort(403)
102
103			if in_admin_mode:
104			permissions = service.get_permissions_for_user(user.id)
105			if AdminPermission.access not in permissions:
106			# The user lacks the admin access permission which is required
107			# to enter the admin area.
108			abort(403)
109
110			if not in_admin_mode:
111			required_consent_subject_ids = _get_required_consent_subject_ids()
112			if _is_consent_required(user.id, required_consent_subject_ids):
113			verification_token = verification_token_service.create_for_terms_consent(
114			user.id
115			)
116
117			consent_form_url = url_for(
118			'consent.consent_form', token=verification_token.token
119			)
120
121			return [('Location', consent_form_url)]
122
123			# Authorization succeeded.
124
125			session_token = session_service.get_session_token(user.id)
126
127			service.create_login_event(user.id, request.remote_addr)
128
129			user_session.start(user.id, session_token.token, permanent=permanent)
130			flash_success(f'Erfolgreich eingeloggt als {user.screen_name}.')
131
132
133			def _get_required_consent_subject_ids():
134			subject_ids = []
135
136			terms_version = terms_version_service.find_current_version_for_brand(
137			g.brand_id
138			)
139			if terms_version:
140			subject_ids.append(terms_version.consent_subject_id)
141
142			privacy_policy_consent_subject_id = (
143			_find_privacy_policy_consent_subject_id()
144			)
145
146			if privacy_policy_consent_subject_id:
147			subject_ids.append(privacy_policy_consent_subject_id)
148
149			return subject_ids
150
151
152			def _is_consent_required(user_id, subject_ids):
153			for subject_id in subject_ids:
154			if not consent_service.has_user_consented_to_subject(
155			user_id, subject_id
156			):
157			return True
158
159			return False
160
161
162			@blueprint.route('/logout', methods=['POST'])
163			@respond_no_content
164			def logout():
165			"""Log out user by deleting the corresponding cookie."""
166			user_session.end()
167			flash_success('Erfolgreich ausgeloggt.')
168
169
170			# -------------------------------------------------------------------- #
171			# password update
172
173
174			@blueprint.route('/password/update')
175			@templated
176			def password_update_form(erroneous_form=None):
177			"""Show a form to update the current user's password."""
178			_get_current_user_or_404()
179
180			form = erroneous_form if erroneous_form else UpdatePasswordForm()
181
182			return {'form': form}
183
184
185			@blueprint.route('/password', methods=['POST'])
186			def password_update():
187			"""Update the current user's password."""
188			user = _get_current_user_or_404()
189
190			form = UpdatePasswordForm(request.form)
191
192			if not form.validate():
193			return password_update_form(form)
194
195			password = form.new_password.data
196
197			password_service.update_password_hash(user.id, password, user.id)
198
199			flash_success('Dein Passwort wurde geändert. Bitte melde dich erneut an.')
200			return redirect_to('.login_form')
201
202
203			# -------------------------------------------------------------------- #
204			# password reset
205
206
207			@blueprint.route('/password/reset/request')
208			@templated
209			def request_password_reset_form(erroneous_form=None):
210			"""Show a form to request a password reset."""
211			form = erroneous_form if erroneous_form else RequestPasswordResetForm()
212
213			return {'form': form}
214
215
216			@blueprint.route('/password/reset/request', methods=['POST'])
217			def request_password_reset():
218			"""Request a password reset."""
219			form = RequestPasswordResetForm(request.form)
220			if not form.validate():
221			return request_password_reset_form(form)
222
223			screen_name = form.screen_name.data.strip()
224			user = user_service.find_user_by_screen_name(
225			screen_name, case_insensitive=True
226			)
227
228			if user is None:
229			flash_error(f'Der Benutzername "{screen_name}" ist unbekannt.')
230			return request_password_reset_form(form)
231
232			if not user.email_address_verified:
233			flash_error(
234			f'Die E-Mail-Adresse für das Benutzerkonto "{screen_name}" '
235			'wurde noch nicht bestätigt.'
236			)
237			return redirect_to('user_email_address.request_confirmation_email')
238
239			sender = None
240			if get_site_mode().is_public():
241			site = site_service.get_site(g.site_id)
242			email_config = email_service.get_config(site.email_config_id)
243			sender = email_config.sender
244
245			password_reset_service.prepare_password_reset(
246			user, request.url_root, sender=sender
247			)
248
249			flash_success(
250			'Ein Link zum Setzen eines neuen Passworts '
251			f'für den Benutzernamen "{user.screen_name}" '
252			'wurde an die hinterlegte E-Mail-Adresse versendet.'
253			)
254			return request_password_reset_form()
255
256
257			@blueprint.route('/password/reset/token/<token>')
258			@templated
259			def password_reset_form(token, erroneous_form=None):
260			"""Show a form to reset the current user's password."""
261			verification_token = verification_token_service.find_for_password_reset_by_token(
262			token
263			)
264
265			_verify_password_reset_token(verification_token)
266
267			form = erroneous_form if erroneous_form else ResetPasswordForm()
268
269			return {
270			'form': form,
271			'token': token,
272			}
273
274
275			@blueprint.route('/password/reset/token/<token>', methods=['POST'])
276			def password_reset(token):
277			"""Reset the current user's password."""
278			verification_token = verification_token_service.find_for_password_reset_by_token(
279			token
280			)
281
282			_verify_password_reset_token(verification_token)
283
284			form = ResetPasswordForm(request.form)
285			if not form.validate():
286			return password_reset_form(token, form)
287
288			password = form.new_password.data
289
290			password_reset_service.reset_password(verification_token, password)
291
292			flash_success('Das Passwort wurde geändert.')
293			return redirect_to('.login_form')
294
295
296			def _verify_password_reset_token(verification_token):
297			if verification_token is None or verification_token.is_expired:
298			flash_error(
299			'Es wurde kein gültiges Token angegeben. '
300			'Ein Token ist nur 24 Stunden lang gültig.'
301			)
302			abort(404)
303
304
305			# -------------------------------------------------------------------- #
306			# helpers
307
308
309			def _is_user_account_creation_enabled(in_admin_mode):
310			if in_admin_mode:
311			return False
312
313			site = _get_site()
314			return site.user_account_creation_enabled
315
316
317			def _is_login_enabled(in_admin_mode):
318			if in_admin_mode:
319			return True
320
321			site = _get_site()
322			return site.login_enabled
323
324
325			def _get_site():
326			return site_service.get_site(g.site_id)
327
328
329			def _get_current_user_or_404():
330			user = g.current_user
331			if not user.is_active:
332			abort(404)
333
334			return user
335

byceps / byceps

Push — master ( 127990...e34625 )

byceps.blueprints.authentication.views._get_site() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like