User session tracking #93587
@ -259,35 +259,10 @@ def check_verification_payload(
|
|||||||
return VerificationResult.OK, payload
|
return VerificationResult.OK, payload
|
||||||
|
|
||||||
|
|
||||||
def send_new_user_session(session):
|
def construct_new_user_session(user, session_data):
|
||||||
if not hasattr(session, 'user'):
|
|
||||||
log.error('programming error: called for a session without a user')
|
|
||||||
return False
|
|
||||||
user = session.user
|
|
||||||
|
|
||||||
# sending only a text/plain email to reduce the room for look-alike phishing emails
|
|
||||||
email_body_txt, subject = construct_new_user_session(session)
|
|
||||||
|
|
||||||
email = user.email
|
|
||||||
try:
|
|
||||||
send_mail(
|
|
||||||
subject,
|
|
||||||
message=email_body_txt,
|
|
||||||
from_email=None, # just use the configured default From-address.
|
|
||||||
recipient_list=[email],
|
|
||||||
fail_silently=False,
|
|
||||||
)
|
|
||||||
except (smtplib.SMTPException, OSError):
|
|
||||||
log.exception("failed to send a new user session email for account %s", user.pk)
|
|
||||||
return False
|
|
||||||
log.info("sent a new user session email for account %s", user.pk)
|
|
||||||
return True
|
|
||||||
|
|
||||||
|
|
||||||
def construct_new_user_session(session):
|
|
||||||
context = {
|
context = {
|
||||||
"session": session,
|
"session_data": session_data,
|
||||||
"user": session.user,
|
"user": user,
|
||||||
"subject": "Blender ID new sign-in",
|
"subject": "Blender ID new sign-in",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -6,9 +6,10 @@ from django.db.models import F
|
|||||||
from django.db.models.signals import m2m_changed, post_delete
|
from django.db.models.signals import m2m_changed, post_delete
|
||||||
from django.dispatch import receiver
|
from django.dispatch import receiver
|
||||||
|
|
||||||
from . import email, models
|
from . import models
|
||||||
import bid_main.utils as utils
|
import bid_main.utils as utils
|
||||||
import bid_main.file_utils
|
import bid_main.file_utils
|
||||||
|
import bid_main.tasks
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
@ -35,11 +36,15 @@ def process_new_login(sender, request, user, **kwargs):
|
|||||||
user.last_login_ip = F("current_login_ip")
|
user.last_login_ip = F("current_login_ip")
|
||||||
user.current_login_ip = request_ip
|
user.current_login_ip = request_ip
|
||||||
fields.update({"last_login_ip", "current_login_ip"})
|
fields.update({"last_login_ip", "current_login_ip"})
|
||||||
|
|
||||||
if user.has_confirmed_email:
|
if user.has_confirmed_email:
|
||||||
Oleg-Komarov marked this conversation as resolved
|
|||||||
try:
|
bid_main.tasks.send_new_user_session(
|
||||||
email.send_new_user_session(user_session)
|
user_pk=user.pk,
|
||||||
except Exception:
|
session_data={
|
||||||
log.exception('failed to send a new user session email')
|
'device': str(user_session.device or 'Unknown'),
|
||||||
|
'ip': user_session.ip,
|
||||||
|
},
|
||||||
|
)
|
||||||
user.save(update_fields=fields)
|
user.save(update_fields=fields)
|
||||||
|
|
||||||
|
|
||||||
|
28
bid_main/tasks.py
Normal file
28
bid_main/tasks.py
Normal file
@ -0,0 +1,28 @@
|
|||||||
|
import logging
|
||||||
|
|
||||||
|
from background_task import background
|
||||||
|
from background_task.tasks import TaskSchedule
|
||||||
|
from django.core.mail import send_mail
|
||||||
|
|
||||||
|
from bid_main.models import User
|
||||||
|
import bid_main.email
|
||||||
|
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@background(schedule={'action': TaskSchedule.RESCHEDULE_EXISTING})
|
||||||
|
def send_new_user_session(user_pk, session_data):
|
||||||
|
user = User.objects.get(pk=user_pk)
|
||||||
|
log.info("sending a new user session email for account %s", user.pk)
|
||||||
|
|
||||||
|
# sending only a text/plain email to reduce the room for look-alike phishing emails
|
||||||
|
email_body_txt, subject = bid_main.email.construct_new_user_session(user, session_data)
|
||||||
|
|
||||||
|
email = user.email
|
||||||
|
send_mail(
|
||||||
|
subject=subject,
|
||||||
|
message=email_body_txt,
|
||||||
|
from_email=None, # just use the configured default From-address.
|
||||||
|
recipient_list=[email],
|
||||||
|
)
|
@ -3,8 +3,8 @@ Dear {{ user.full_name|default:user.email }}!
|
|||||||
|
|
||||||
A new sign-in for your Blender ID account {{ user.email }}
|
A new sign-in for your Blender ID account {{ user.email }}
|
||||||
|
|
||||||
IP address: {{ session.ip }}
|
IP address: {{ session_data.ip }}
|
||||||
Device: {{ session.device }}
|
Device: {{ session_data.device }}
|
||||||
|
|
||||||
If this was you, you can ignore this message.
|
If this was you, you can ignore this message.
|
||||||
If this wasn't you, please change or reset your password.
|
If this wasn't you, please change or reset your password.
|
||||||
|
@ -1,8 +1,13 @@
|
|||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from django.core import mail
|
||||||
from django.test.client import Client
|
from django.test.client import Client
|
||||||
from django.test import TestCase
|
from django.test import TestCase
|
||||||
from django.urls import reverse
|
from django.urls import reverse
|
||||||
|
from django.utils import timezone
|
||||||
|
|
||||||
from bid_main.tests.factories import UserFactory
|
from bid_main.tests.factories import UserFactory
|
||||||
|
import bid_main.tasks
|
||||||
|
|
||||||
|
|
||||||
class TestUserSessions(TestCase):
|
class TestUserSessions(TestCase):
|
||||||
@ -41,3 +46,32 @@ class TestActiveSessions(TestCase):
|
|||||||
response = client1.get(reverse('bid_main:active_sessions'))
|
response = client1.get(reverse('bid_main:active_sessions'))
|
||||||
self.assertEqual(response.status_code, 302)
|
self.assertEqual(response.status_code, 302)
|
||||||
self.assertEqual(response['Location'], '/login?next=/active-sessions/')
|
self.assertEqual(response['Location'], '/login?next=/active-sessions/')
|
||||||
|
|
||||||
|
|
||||||
|
class TestNewUserSessionEmail(TestCase):
|
||||||
|
@patch(
|
||||||
|
'bid_main.tasks.send_new_user_session',
|
||||||
|
new=bid_main.tasks.send_new_user_session.task_function,
|
||||||
|
)
|
||||||
|
@patch(
|
||||||
|
'django.contrib.auth.base_user.AbstractBaseUser.check_password',
|
||||||
|
new=lambda _, pwd: pwd == 'hunter2',
|
||||||
|
)
|
||||||
|
def test_new_user_session_email(self):
|
||||||
|
user = UserFactory(confirmed_email_at=timezone.now())
|
||||||
|
client1 = Client()
|
||||||
|
client1.force_login(user)
|
||||||
|
|
||||||
|
# force_login doesn't set custom env for HttpRequest, so sending a real POST instead
|
||||||
|
client2 = Client(REMOTE_ADDR='127.0.1.1')
|
||||||
|
ua = 'Mozilla/5.0 (X11; Linux x86_64; rv:128.0) Gecko/20100101 Firefox/128.0'
|
||||||
|
response = client2.post(
|
||||||
|
'/login',
|
||||||
|
{'username': user.email, 'password': 'hunter2'},
|
||||||
|
headers={'user-agent': ua},
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, 302)
|
||||||
|
sent_email = mail.outbox.pop()
|
||||||
|
self.assertEqual(sent_email.to[0], user.email)
|
||||||
|
self.assertIn('IP address: 127.0.1.1', sent_email.body)
|
||||||
|
self.assertIn('Device: PC / Linux / Firefox 128.0', sent_email.body)
|
||||||
|
Loading…
Reference in New Issue
Block a user
should be a call of a task